How Workflow Runs Use TinyFlows Durable Graphs with Checkpointing in OpenHuman

OpenHuman persists workflow executions as TinyFlows durable graphs by using a SQLite-backed checkpointer that records every node output to a journaled checkpoint file, enabling runs to resume from the exact state after crashes or interruptions.

Workflow runs in the OpenHuman automation engine leverage TinyFlows durable graphs to create fault-tolerant, resumable execution pipelines. This integration stores complete graph states—including node outputs, metadata, and execution context—to a local SQLite database, allowing interrupted processes to resume from precise checkpoints rather than restarting from scratch.

The Durable Graph Architecture

OpenHuman structures workflow runs around capabilities (caps) that define the graph's behavior. The high-level entry point open_flow_checkpointer, re-exported from src/openhuman/flows/tinyflows/mod.rs, orchestrates the creation of durable execution contexts.

The system compiles a set of capability adapters into a runnable graph:

• Memory nodes for state persistence between steps
• HTTP adapters for external API calls
• LLM adapters for language model invocations
• Native tool sets for built-in operations

These components are assembled via build_capabilities in src/openhuman/flows/tinyflows/caps/mod.rs, producing a Graph instance that understands how to serialize its state for durability.

Checkpoint Persistence Layer

When a run initiates, the engine instantiates a SqliteCheckpointer defined in src/openhuman/flows/tinyflows/checkpoint_sqlite.rs. This implementation of the TinyFlows Checkpointer trait writes a record for every graph node that produces output through checkpoint_writes.

All checkpoint data resides in a SQLite file at <workspace>/flows/checkpoints.db, maintaining the same on-disk format used prior to TinyFlows integration. Each row contains six critical fields:

• thread_id – The TinyFlows run/thread identifier exposed to the FlowRunObserver
• checkpoint_id – A user-supplied or auto-generated identifier for the specific checkpoint
• parent_checkpoint_id – References the preceding checkpoint, enabling parent-chain traversal
• run_id – The overall flow run identifier grouping related checkpoints
• namespace – A JSON-encoded isolation boundary per flow
• seq – Monotonically increasing sequence number for deterministic ordering

During execution, the engine calls put on the checkpointer to persist the current graph state, including metadata, next nodes, and the full Checkpoint<State> payload.

Runtime Integration and Resumption

The observability layer in src/openhuman/flows/tinyflows/observability.rs wraps runs with a journal recording start, checkpoint, and finish events. This layer forwards events to the FlowRunObserver, which receives the checkpointer's thread_id to correlate live runs with persisted storage.

If a process crashes or is manually stopped, the persisted state enables precise resumption:

1. The runtime reads the latest checkpoint from <workspace>/flows/checkpoints.db
2. resume_with_checkpointer_journaled_observed reconstructs the graph state
3. Execution continues from the last successfully persisted node

This mechanism ensures that long-running workflows—such as multi-step LLM chains or complex data pipelines—maintain atomic progress guarantees.

Lifecycle Management and Pruning

The checkpointer implementation includes automated maintenance to prevent unbounded storage growth. According to the source code test suite, the system supports:

• Retention policies – prune_keeps_the_newest_checkpoints removes older entries when configured limits are exceeded
• Thread deletion – deleting_a_thread_removes_its_checkpoints_and_leaves_others ensures that deleting a specific thread identifier removes all associated rows without affecting other runs

These features allow production deployments to maintain predictable database sizes while preserving the durability guarantees required for critical automation workflows.

Implementation Example

The following Rust code demonstrates constructing a durable workflow with checkpointing:

use openhuman::flows::tinyflows::{
    caps::{self, build_capabilities},
    open_flow_checkpointer,
    checkpoint_sqlite::SqliteCheckpointer,
};
use tinyflows::graph::{Graph, RunConfig};

// 1. Build the graph capabilities (IO adapters, LLM, HTTP, etc.)
let caps = build_capabilities(
    caps::memory_node::MemoryNode::default(),
    caps::http::HttpAdapter::default(),
    caps::llm::LlmAdapter::default(),
    caps::tools::NativeToolSet::default(),
);

// 2. Create a SQLite-backed checkpointer (workspace is the current OpenHuman workspace)
let cp = SqliteCheckpointer::open(":memory:").expect("in-memory checkpointer");

// 3. Assemble the run configuration (you can name the run, set a checkpoint id, …)
let cfg = RunConfig::default()
    .with_checkpoint_id(Some("run-1".to_string()))
    .with_thread_id("thread-42".to_string());

// 4. Start the flow with a checkpointer so every node write is persisted
let run = open_flow_checkpointer(caps, cp, cfg).expect("run created");

// 5. Execute the graph (example: a simple sequential flow)
run.run().await.expect("flow completed");

// 6. Later – resume from the last checkpoint
let resumed = open_flow_checkpointer(caps, cp, cfg)
    .and_then(|r| r.resume())
    .await
    .expect("resumed run");

For persistent storage between process restarts, replace ":memory:" with the workspace path: SqliteCheckpointer::open("<workspace>/flows/checkpoints.db").

Summary

• OpenHuman integrates TinyFlows durable graphs through the open_flow_checkpointer entry point in src/openhuman/flows/tinyflows/mod.rs.
• The SqliteCheckpointer persists every node output to <workspace>/flows/checkpoints.db with metadata including thread_id, checkpoint_id, and parent references.
• The Checkpointer trait implementation enables crash recovery via resume_with_checkpointer_journaled_observed in the observability layer.
• Checkpoint pruning and thread deletion keep storage bounded in production environments.
• Workflow runs combine capability adapters (memory, HTTP, LLM) with durable storage to create fault-tolerant automation pipelines.

Frequently Asked Questions

Where does OpenHuman store workflow checkpoint data?

Checkpoint data is stored in a SQLite database file located at <workspace>/flows/checkpoints.db within the current OpenHuman workspace. This path predates the TinyFlows integration, ensuring backward compatibility with existing deployment structures while storing the thread_id, checkpoint_id, and serialized graph state for every persisted node output.

How does resuming a workflow run work after a process crash?

The SqliteCheckpointer implementation allows the runtime to call resume_with_checkpointer_journaled_observed from src/openhuman/flows/tinyflows/observability.rs. This method reads the latest checkpoint row matching the run's thread_id and checkpoint_id, reconstructs the Checkpoint<State> payload including pending nodes and metadata, and resumes execution from the exact point of interruption without reprocessing completed steps.

What information is contained in each checkpoint row?

Each checkpoint row contains six columns: thread_id (the execution identifier), checkpoint_id (the specific save point), parent_checkpoint_id (for chain traversal), run_id (the overall flow identifier), namespace (JSON-encoded isolation data), and seq (monotonically increasing ordering sequence). Together, these fields enable precise state reconstruction and historical audit trails.

How does OpenHuman prevent checkpoint databases from growing indefinitely?

The checkpointer supports configurable pruning through retention policies validated by tests such as prune_keeps_the_newest_checkpoints, which removes older entries when limits are exceeded. Additionally, deleting_a_thread_removes_its_checkpoints_and_leaves_others ensures that cleaning up a specific workflow thread deletes only its associated rows, preserving other runs while reclaiming storage space.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}