diff --git a/Cargo.lock b/Cargo.lock
index 7e7332504d0..202bd2df72f 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4606,6 +4606,7 @@ dependencies = [
  "tinyflows",
  "tinyflows-catalog",
  "tinyflows-copilot",
+ "tinyflows-drivers",
  "tinyflows-schedule",
  "tinyflows-sqlite",
  "tinyhosts",
@@ -7118,9 +7119,11 @@ dependencies = [
  "rusqlite",
  "serde",
  "serde_json",
+ "sha2 0.11.0",
  "tinyagents-harness",
  "tinyagents-tasks",
  "tinyinference-llm",
+ "tinystoragedrivers-core",
  "tinytools",
  "tokio",
  "tracing",
@@ -7502,6 +7505,7 @@ dependencies = [
 name = "tinyflows-catalog"
 version = "0.1.0"
 dependencies = [
+ "anyhow",
  "hex",
  "serde",
  "serde_json",
@@ -7519,6 +7523,25 @@ dependencies = [
  "serde_json",
 ]
 
+[[package]]
+name = "tinyflows-drivers"
+version = "0.1.0"
+dependencies = [
+ "anyhow",
+ "async-trait",
+ "chrono",
+ "serde",
+ "serde_json",
+ "sha2 0.11.0",
+ "tinyflows",
+ "tinyflows-catalog",
+ "tinyflows-schedule",
+ "tinystoragedrivers-core",
+ "tokio",
+ "tracing",
+ "uuid",
+]
+
 [[package]]
 name = "tinyflows-schedule"
 version = "0.1.0"
diff --git a/crates/openhuman-app/Cargo.lock b/crates/openhuman-app/Cargo.lock
index 5e25e810900..4eaaa616ba8 100644
--- a/crates/openhuman-app/Cargo.lock
+++ b/crates/openhuman-app/Cargo.lock
@@ -4526,6 +4526,7 @@ dependencies = [
  "tinyflows",
  "tinyflows-catalog",
  "tinyflows-copilot",
+ "tinyflows-drivers",
  "tinyflows-schedule",
  "tinyflows-sqlite",
  "tinyhosts",
@@ -7375,9 +7376,11 @@ dependencies = [
  "rusqlite",
  "serde",
  "serde_json",
+ "sha2 0.11.0",
  "tinyagents-harness",
  "tinyagents-tasks",
  "tinyinference-llm",
+ "tinystoragedrivers-core",
  "tinytools",
  "tokio",
  "tracing",
@@ -7773,6 +7776,7 @@ dependencies = [
 name = "tinyflows-catalog"
 version = "0.1.0"
 dependencies = [
+ "anyhow",
  "hex",
  "serde",
  "serde_json",
@@ -7790,6 +7794,25 @@ dependencies = [
  "serde_json",
 ]
 
+[[package]]
+name = "tinyflows-drivers"
+version = "0.1.0"
+dependencies = [
+ "anyhow",
+ "async-trait",
+ "chrono",
+ "serde",
+ "serde_json",
+ "sha2 0.11.0",
+ "tinyflows",
+ "tinyflows-catalog",
+ "tinyflows-schedule",
+ "tinystoragedrivers-core",
+ "tokio",
+ "tracing",
+ "uuid",
+]
+
 [[package]]
 name = "tinyflows-schedule"
 version = "0.1.0"
diff --git a/crates/openhuman-cli/Cargo.toml b/crates/openhuman-cli/Cargo.toml
index 0d0cc8cfe70..e15948e9fc5 100644
--- a/crates/openhuman-cli/Cargo.toml
+++ b/crates/openhuman-cli/Cargo.toml
@@ -93,6 +93,11 @@ path = "../../tests/storage_approvals_e2e.rs"
 name = "storage_domains_e2e"
 path = "../../tests/storage_domains_e2e.rs"
 
+[[test]]
+# Its own binary: it installs a storage backend into the process-wide slot.
+name = "storage_flows_e2e"
+path = "../../tests/storage_flows_e2e.rs"
+
 [[test]]
 # Its own binary: it installs a storage backend into the process-wide slot and
 # sets the keyring master-key environment.
diff --git a/crates/openhuman-core/Cargo.toml b/crates/openhuman-core/Cargo.toml
index 4988aec33a6..9eb5b35ca5c 100644
--- a/crates/openhuman-core/Cargo.toml
+++ b/crates/openhuman-core/Cargo.toml
@@ -81,6 +81,11 @@ tinyflows-catalog = { path = "../../vendor/tinyflows/crates/tinyflows-catalog",
 # The cron job/run store half (`schedule`) is unconditional, so `tinyflows-sqlite`
 # is too; only its `catalog` half (which needs the engine) rides the `flows` gate.
 tinyflows-sqlite = { path = "../../vendor/tinyflows/crates/tinyflows-sqlite", default-features = false, features = ["schedule"] }
+# tinyflows' stores on the storage ports, used instead of the SQLite files when
+# a storage backend is configured (`storage`). The cron half (`schedule`) is
+# unconditional like cron itself; the catalog, flow state and checkpointer
+# (`catalog`, which needs the engine) ride the `flows` gate.
+tinyflows-drivers = { path = "../../vendor/tinyflows/crates/tinyflows-drivers", default-features = false, features = ["schedule"] }
 # The authoring copilot's words: the `workflow_builder` / `flow_discovery`
 # standing archetypes and the turn brief that opens a builder turn. Harness-free
 # by construction — it names no tool trait and no model client — so this host
@@ -147,6 +152,7 @@ tinyagents-definition = { path = "../../vendor/tinyagents/crates/tinyagents-defi
 tinyagents-live = { path = "../../vendor/tinyagents/crates/tinyagents-live", optional = true }
 tinyagents-graph = { path = "../../vendor/tinyagents/crates/tinyagents-graph", features = [
     "sqlite",
+    "storage-drivers",
 ] }
 tinyagents-registry = { path = "../../vendor/tinyagents/crates/tinyagents-registry" }
 tinyagents-session = { path = "../../vendor/tinyagents/crates/tinyagents-session", features = ["storage-drivers"] }
@@ -632,6 +638,9 @@ tinymemory-api = { path = "../../vendor/tinymemory/crates/tinymemory-api", featu
 # write path can produce. Declared here rather than in `[dependencies]` so the
 # shipped binary never carries those two writers.
 tinyflows-sqlite = { path = "../../vendor/tinyflows/crates/tinyflows-sqlite", features = ["test-fixtures"] }
+# The document catalog's `*_for_test` fixtures, which the flows store forwards
+# in test builds.
+tinyflows-drivers = { path = "../../vendor/tinyflows/crates/tinyflows-drivers", default-features = false, features = ["schedule", "test-fixtures"] }
 # Test fixtures derive a known account from the BIP-39 vector phrase. Production
 # does not derive at all — see the note on the main `tinywallet` entry. Cargo
 # does not link dev-dependency features into the shipped binary, so enabling
@@ -914,7 +923,7 @@ media = ["tinyagents-harness/media"]
 # runtime `DomainSet::flows` flag (#4796): the feature narrows the compile-time
 # surface, `DomainSet` gates it at runtime.
 # NOTE: this gate does NOT drop `tinyagents` itself — 26+ domains consume it.
-flows = ["dep:tinyflows", "dep:tinyflows-catalog", "dep:tinyflows-copilot", "tinyflows-sqlite/catalog"]
+flows = ["dep:tinyflows", "dep:tinyflows-catalog", "dep:tinyflows-copilot", "tinyflows-drivers/catalog", "tinyflows-sqlite/catalog"]
 # Skills domains: `openhuman::skills` (metadata/discovery/install),
 # `openhuman::skill_runtime` (SKILL.md execution + the skill_executor agent),
 # and `openhuman::skill_registry` (remote catalogs + the skill_setup agent).
diff --git a/crates/openhuman-core/src/agent/orchestration/delegation.rs b/crates/openhuman-core/src/agent/orchestration/delegation.rs
index 93ed724f52b..453415c22a7 100644
--- a/crates/openhuman-core/src/agent/orchestration/delegation.rs
+++ b/crates/openhuman-core/src/agent/orchestration/delegation.rs
@@ -81,11 +81,26 @@ pub(crate) async fn run_subagent_delegation_with_parent_context(
     // durable graphs simply expire (orphaned tasks are reconciled at boot per
     // 07.2). Checkpoint metadata (thread/checkpoint/parent/run ids) stays
     // inspectable through the crate `Checkpointer` API.
-    let checkpoint_db = config.workspace_dir.join("graph_checkpoints.db");
-    let checkpointer: Arc<dyn Checkpointer<DelegationState>> = Arc::new(
-        SqliteCheckpointer::<DelegationState>::open(&checkpoint_db)
-            .map_err(|e| format!("open durable graph checkpoint store: {e}"))?,
-    );
+    //
+    // With a storage backend configured (`crate::storage`) the checkpoints
+    // live there instead, in the acting agent's scope.
+    let checkpointer: Arc<dyn Checkpointer<DelegationState>> =
+        match crate::storage::current_scoped()
+            .map_err(|e| format!("resolve the storage scope for graph checkpoints: {e}"))?
+        {
+            Some(scoped) => Arc::new(tinyagents_graph::checkpoint::DriverCheckpointer::<
+                DelegationState,
+            >::with_prefix(
+                Arc::clone(scoped.documents()), "delegation_graph"
+            )),
+            None => {
+                let checkpoint_db = config.workspace_dir.join("graph_checkpoints.db");
+                Arc::new(
+                    SqliteCheckpointer::<DelegationState>::open(&checkpoint_db)
+                        .map_err(|e| format!("open durable graph checkpoint store: {e}"))?,
+                )
+            }
+        };
 
     tracing::info!(
         target: LOG_TARGET,
diff --git a/crates/openhuman-core/src/cron/README.md b/crates/openhuman-core/src/cron/README.md
index 1720daed6de..2ebd49c5e5f 100644
--- a/crates/openhuman-core/src/cron/README.md
+++ b/crates/openhuman-core/src/cron/README.md
@@ -231,6 +231,16 @@ missing row, reschedules a drifted one, and removes the retired
 `memory_context_refresh` row. It runs from [`core/runtime/services.rs`](../core/runtime/services.rs) and from
 [`security/credentials/ops/user_scope.rs`](../security/credentials/ops/user_scope.rs).
 
+### On a storage backend
+
+With a storage backend configured (`OPENHUMAN_STORAGE_URL` / `[storage] url`,
+see `crate::storage`), every `store` function is served by
+`tinyflows_drivers::schedule::CronDocuments` instead of `cron/jobs.db`: the
+same jobs and runs on the document port, in the acting agent's storage scope,
+with the same history cap and due-job batch size. `reschedule_after_run`
+advances a job only if its stored schedule still matches the job that fired,
+so two processes sharing one database don't double-advance it.
+
 ## Layout
 
 | Path | What it does |
diff --git a/crates/openhuman-core/src/cron/store.rs b/crates/openhuman-core/src/cron/store.rs
index 374d6bcb3f6..8956f165f3b 100644
--- a/crates/openhuman-core/src/cron/store.rs
+++ b/crates/openhuman-core/src/cron/store.rs
@@ -1,5 +1,9 @@
 //! Cron persistence: a thin host wrapper over `tinyflows_sqlite::schedule`.
 //!
+//! With a storage backend configured ([`crate::storage`]) every function is
+//! served by `tinyflows_drivers::schedule::CronDocuments` instead, in the
+//! acting agent's storage scope, with the same limits.
+//!
 //! The SQLite job store and run history live upstream (schema, CRUD, output
 //! truncation, pruning). This module only turns the host [`Config`] into the
 //! store's [`CronStoreOptions`] — database path under the workspace, run-history
@@ -8,6 +12,7 @@
 use crate::config::Config;
 use anyhow::Result;
 use chrono::{DateTime, Utc};
+use tinyflows_drivers::schedule::CronDocuments;
 use tinyflows_schedule::DeliveryStatus;
 use tinyflows_schedule::{CronJob, CronJobPatch, CronRun, DeliveryConfig, Schedule, SessionTarget};
 use tinyflows_sqlite::schedule::{self as upstream, AgentJobSpec, CronStoreOptions};
@@ -22,7 +27,26 @@ fn opts(config: &Config) -> CronStoreOptions {
     }
 }
 
+/// The document store for this call when the host configured a backend.
+fn documents(config: &Config) -> Result<Option<CronDocuments>> {
+    Ok(crate::storage::current_scoped()?.map(|scoped| {
+        CronDocuments::new(std::sync::Arc::clone(scoped.documents()))
+            .with_limits(config.cron.max_run_history, config.scheduler.max_tasks)
+    }))
+}
+
+/// Runs a document-store call from this synchronous API.
+fn run<T: Send + 'static>(
+    future: impl std::future::Future<Output = Result<T>> + Send + 'static,
+) -> Result<T> {
+    crate::storage::block_on_anyhow(future)
+}
+
 pub fn add_job(config: &Config, expression: &str, command: &str) -> Result<CronJob> {
+    if let Some(docs) = documents(config)? {
+        let (expression, command) = (expression.to_string(), command.to_string());
+        return run(async move { docs.add_job(&expression, &command).await });
+    }
     upstream::add_job(&opts(config), expression, command)
 }
 
@@ -32,6 +56,10 @@ pub fn add_shell_job(
     schedule: Schedule,
     command: &str,
 ) -> Result<CronJob> {
+    if let Some(docs) = documents(config)? {
+        let command = command.to_string();
+        return run(async move { docs.add_shell_job(name, schedule, &command).await });
+    }
     upstream::add_shell_job(&opts(config), name, schedule, command)
 }
 
@@ -46,6 +74,21 @@ pub fn add_agent_job(
     delivery: Option<DeliveryConfig>,
     delete_after_run: bool,
 ) -> Result<CronJob> {
+    if let Some(docs) = documents(config)? {
+        let prompt = prompt.to_string();
+        return run(async move {
+            docs.add_agent_job(
+                name,
+                schedule,
+                &prompt,
+                session_target,
+                model,
+                delivery,
+                delete_after_run,
+            )
+            .await
+        });
+    }
     upstream::add_agent_job(
         &opts(config),
         name,
@@ -73,6 +116,23 @@ pub fn add_agent_job_with_definition(
     agent_id: Option<String>,
     enabled: bool,
 ) -> Result<CronJob> {
+    if let Some(docs) = documents(config)? {
+        let prompt = prompt.to_string();
+        return run(async move {
+            docs.add_agent_job_with_definition(
+                name,
+                schedule,
+                &prompt,
+                session_target,
+                model,
+                delivery,
+                delete_after_run,
+                agent_id,
+                enabled,
+            )
+            .await
+        });
+    }
     upstream::add_agent_job_with_definition(
         &opts(config),
         name,
@@ -89,6 +149,9 @@ pub fn add_agent_job_with_definition(
 
 /// Adds an agent job described by `spec`, including its origin conversation.
 pub fn add_agent_job_from_spec(config: &Config, spec: AgentJobSpec) -> Result<CronJob> {
+    if let Some(docs) = documents(config)? {
+        return run(async move { docs.add_agent_job_from_spec(spec).await });
+    }
     upstream::add_agent_job_from_spec(&opts(config), spec)
 }
 
@@ -99,22 +162,43 @@ pub fn add_flow_schedule_job(
     flow_id: &str,
     schedule: Schedule,
 ) -> Result<CronJob> {
+    if let Some(docs) = documents(config)? {
+        let flow_id = flow_id.to_string();
+        return run(async move { docs.add_flow_schedule_job(&flow_id, schedule).await });
+    }
     upstream::add_flow_schedule_job(&opts(config), flow_id, schedule)
 }
 
 pub fn find_flow_schedule_job(config: &Config, flow_id: &str) -> Result<Option<CronJob>> {
+    if let Some(docs) = documents(config)? {
+        let flow_id = flow_id.to_string();
+        return run(async move { docs.find_flow_schedule_job(&flow_id).await });
+    }
     upstream::find_flow_schedule_job(&opts(config), flow_id)
 }
 
 pub fn list_jobs(config: &Config) -> Result<Vec<CronJob>> {
+    if let Some(docs) = documents(config)? {
+        return run(async move { docs.list_jobs().await });
+    }
     upstream::list_jobs(&opts(config))
 }
 
 pub fn get_job(config: &Config, job_id: &str) -> Result<CronJob> {
+    if let Some(docs) = documents(config)? {
+        let job_id = job_id.to_string();
+        return run(async move { docs.get_job(&job_id).await });
+    }
     upstream::get_job(&opts(config), job_id)
 }
 
 pub fn remove_job(config: &Config, id: &str) -> Result<()> {
+    if let Some(docs) = documents(config)? {
+        let owned = id.to_string();
+        run(async move { docs.remove_job(&owned).await })?;
+        println!("✅ Removed cron job {id}");
+        return Ok(());
+    }
     upstream::remove_job(&opts(config), id)?;
     println!("✅ Removed cron job {id}");
     Ok(())
@@ -122,19 +206,32 @@ pub fn remove_job(config: &Config, id: &str) -> Result<()> {
 
 /// Deletes every cron job in the workspace (E2E `openhuman.test_reset`).
 pub fn clear_all_jobs(config: &Config) -> Result<usize> {
+    if let Some(docs) = documents(config)? {
+        return run(async move { docs.clear_all_jobs().await });
+    }
     upstream::clear_all_jobs(&opts(config))
 }
 
 /// Removes duplicate jobs sharing a `name`, keeping the one with most history.
 pub fn dedup_named_jobs(config: &Config) -> Result<usize> {
+    if let Some(docs) = documents(config)? {
+        return run(async move { docs.dedup_named_jobs().await });
+    }
     upstream::dedup_named_jobs(&opts(config))
 }
 
 pub fn due_jobs(config: &Config, now: DateTime<Utc>) -> Result<Vec<CronJob>> {
+    if let Some(docs) = documents(config)? {
+        return run(async move { docs.due_jobs(now).await });
+    }
     upstream::due_jobs(&opts(config), now)
 }
 
 pub fn update_job(config: &Config, job_id: &str, patch: CronJobPatch) -> Result<CronJob> {
+    if let Some(docs) = documents(config)? {
+        let job_id = job_id.to_string();
+        return run(async move { docs.update_job(&job_id, patch).await });
+    }
     upstream::update_job(&opts(config), job_id, patch)
 }
 
@@ -145,6 +242,13 @@ pub fn record_last_run(
     success: bool,
     output: &str,
 ) -> Result<()> {
+    if let Some(docs) = documents(config)? {
+        let (job_id, output) = (job_id.to_string(), output.to_string());
+        return run(async move {
+            docs.record_last_run(&job_id, finished_at, success, &output)
+                .await
+        });
+    }
     upstream::record_last_run(&opts(config), job_id, finished_at, success, output)
 }
 
@@ -154,6 +258,10 @@ pub fn reschedule_after_run(
     success: bool,
     output: &str,
 ) -> Result<()> {
+    if let Some(docs) = documents(config)? {
+        let (job, output) = (job.clone(), output.to_string());
+        return run(async move { docs.reschedule_after_run(&job, success, &output).await });
+    }
     upstream::reschedule_after_run(&opts(config), job, success, output)
 }
 
@@ -166,6 +274,24 @@ pub fn record_run(
     output: Option<&str>,
     duration_ms: i64,
 ) -> Result<()> {
+    if let Some(docs) = documents(config)? {
+        let (job_id, status, output) = (
+            job_id.to_string(),
+            status.to_string(),
+            output.map(str::to_string),
+        );
+        return run(async move {
+            docs.record_run(
+                &job_id,
+                started_at,
+                finished_at,
+                &status,
+                output.as_deref(),
+                duration_ms,
+            )
+            .await
+        });
+    }
     upstream::record_run(
         &opts(config),
         job_id,
@@ -189,6 +315,25 @@ pub fn record_run_with_delivery(
     duration_ms: i64,
     delivery_status: Option<DeliveryStatus>,
 ) -> Result<()> {
+    if let Some(docs) = documents(config)? {
+        let (job_id, status, output) = (
+            job_id.to_string(),
+            status.to_string(),
+            output.map(str::to_string),
+        );
+        return run(async move {
+            docs.record_run_with_delivery(
+                &job_id,
+                started_at,
+                finished_at,
+                &status,
+                output.as_deref(),
+                duration_ms,
+                delivery_status,
+            )
+            .await
+        });
+    }
     upstream::record_run_with_delivery(
         &opts(config),
         job_id,
@@ -203,10 +348,18 @@ pub fn record_run_with_delivery(
 
 /// Removes "queued" placeholder rows so only the real result row remains.
 pub fn delete_queued_runs(config: &Config, job_id: &str) -> Result<usize> {
+    if let Some(docs) = documents(config)? {
+        let job_id = job_id.to_string();
+        return run(async move { docs.delete_queued_runs(&job_id).await });
+    }
     upstream::delete_queued_runs(&opts(config), job_id)
 }
 
 pub fn list_runs(config: &Config, job_id: &str, limit: usize) -> Result<Vec<CronRun>> {
+    if let Some(docs) = documents(config)? {
+        let job_id = job_id.to_string();
+        return run(async move { docs.list_runs(&job_id, limit).await });
+    }
     upstream::list_runs(&opts(config), job_id, limit)
 }
 
diff --git a/crates/openhuman-core/src/flows/README.md b/crates/openhuman-core/src/flows/README.md
index a4ed70dc040..8b43c155ced 100644
--- a/crates/openhuman-core/src/flows/README.md
+++ b/crates/openhuman-core/src/flows/README.md
@@ -233,6 +233,19 @@ so the digest subscriber, `flows_delete`, the `memory` node adapter
 (`tinyflows::memory_adapter::OpenHumanMemory`) and the agent tools all tag the
 same way.
 
+### On a storage backend
+
+With a storage backend configured (`OPENHUMAN_STORAGE_URL` / `[storage] url`,
+see `crate::storage`), the catalog (`store.rs`), drafts (`draft_store.rs`),
+per-flow state and the run checkpointer move off `flows/flows.db`,
+`flows/drafts/` and `flows/checkpoints.db` onto the document port, in the
+acting agent's scope: `tinyflows_drivers::catalog::FlowCatalogDocuments`,
+`FlowStateDocuments` (through `tinyflows/state.rs`'s `FlowState`, which the
+engine's `StateStore` and the dedup settlement share) and
+`DriverCheckpointer`. When the scope cannot be resolved (SaaS mode with no
+acting agent) flow state fails every call rather than falling back to the
+local file.
+
 ## Layout
 
 Top level:
diff --git a/crates/openhuman-core/src/flows/bus/dedup_commit.rs b/crates/openhuman-core/src/flows/bus/dedup_commit.rs
index 93a8aa75e5a..8116d8a9049 100644
--- a/crates/openhuman-core/src/flows/bus/dedup_commit.rs
+++ b/crates/openhuman-core/src/flows/bus/dedup_commit.rs
@@ -5,13 +5,13 @@
 use crate::config::Config;
 use crate::core::events::DomainEvent;
 use crate::flows::store;
+use crate::flows::tinyflows::state::FlowState;
 use async_trait::async_trait;
 use std::collections::HashMap;
 use std::sync::{Arc, LazyLock, Mutex};
 use tinybus::EventHandler;
 use tinyflows::model::NodeKind;
 use tinyflows::nodes::control_flow::dedup_settle::{self, CommitOutcome, Settlement};
-use tinyflows_sqlite::flows::SqliteStateStore;
 
 /// Listens for `DomainEvent::FlowRunFinished` and settles every `dedup` node
 /// in the finished flow's graph — the host half of the commit-on-success
@@ -246,7 +246,7 @@ impl DedupCommitSubscriber {
         tracing::trace!(target: "flows", %flow_id, %run_id, "[dedup-commit] acquired per-flow commit lock");
         self.maybe_test_delay().await;
 
-        let state = SqliteStateStore::new(store::dir(&self.config), format!("flow:{flow_id}"));
+        let state = FlowState::open(&self.config, format!("flow:{flow_id}"));
         for node_id in node_ids {
             self.settle_node(&state, &node_id, success, flow_id, run_id);
         }
@@ -259,7 +259,7 @@ impl DedupCommitSubscriber {
     /// `tinyflows`' pure commit/release logic, logging each outcome.
     fn settle_node(
         &self,
-        store: &SqliteStateStore,
+        store: &FlowState,
         node_id: &str,
         success: bool,
         flow_id: &str,
diff --git a/crates/openhuman-core/src/flows/draft_store.rs b/crates/openhuman-core/src/flows/draft_store.rs
index 15a05345ed0..652cf834f1a 100644
--- a/crates/openhuman-core/src/flows/draft_store.rs
+++ b/crates/openhuman-core/src/flows/draft_store.rs
@@ -2,9 +2,10 @@
 //!
 //! Same shape and same reason as [`super::store`]: the storage lives in
 //! `tinyflows_sqlite::drafts` and takes a directory; this supplies the
-//! directory. Drafts land in `<workspace_dir>/flows/drafts/<id>.json`.
+//! directory. Drafts land in `<workspace_dir>/flows/drafts/<id>.json`, or in
+//! the storage backend's catalog when one is configured.
 
-use super::store::dir;
+use super::store::{dir, documents, run};
 use crate::config::Config;
 use anyhow::Result;
 use serde_json::Value;
@@ -19,12 +20,19 @@ pub fn create_draft(
     graph: Value,
     origin: DraftOrigin,
 ) -> Result<FlowDraft> {
+    if let Some(docs) = documents(config)? {
+        return run(async move { docs.create_draft(flow_id, name, graph, origin).await });
+    }
     tinyflows_sqlite::drafts::create_draft(&dir(config), flow_id, name, graph, origin)
 }
 
 /// Binds [`tinyflows_sqlite::drafts::get_draft`] to this host's catalog directory.
 #[inline]
 pub fn get_draft(config: &Config, id: &str) -> Result<Option<FlowDraft>> {
+    if let Some(docs) = documents(config)? {
+        let id = id.to_string();
+        return run(async move { docs.get_draft(&id).await });
+    }
     tinyflows_sqlite::drafts::get_draft(&dir(config), id)
 }
 
@@ -37,17 +45,28 @@ pub fn update_draft(
     graph: Option<Value>,
     flow_id: Option<Option<String>>,
 ) -> Result<FlowDraft> {
+    if let Some(docs) = documents(config)? {
+        let id = id.to_string();
+        return run(async move { docs.update_draft(&id, name, graph, flow_id).await });
+    }
     tinyflows_sqlite::drafts::update_draft(&dir(config), id, name, graph, flow_id)
 }
 
 /// Binds [`tinyflows_sqlite::drafts::list_drafts`] to this host's catalog directory.
 #[inline]
 pub fn list_drafts(config: &Config) -> Result<Vec<FlowDraft>> {
+    if let Some(docs) = documents(config)? {
+        return run(async move { docs.list_drafts().await });
+    }
     tinyflows_sqlite::drafts::list_drafts(&dir(config))
 }
 
 /// Binds [`tinyflows_sqlite::drafts::delete_draft`] to this host's catalog directory.
 #[inline]
 pub fn delete_draft(config: &Config, id: &str) -> Result<bool> {
+    if let Some(docs) = documents(config)? {
+        let id = id.to_string();
+        return run(async move { docs.delete_draft(&id).await });
+    }
     tinyflows_sqlite::drafts::delete_draft(&dir(config), id)
 }
diff --git a/crates/openhuman-core/src/flows/store.rs b/crates/openhuman-core/src/flows/store.rs
index 7d3bc72d621..d5e39bc9c40 100644
--- a/crates/openhuman-core/src/flows/store.rs
+++ b/crates/openhuman-core/src/flows/store.rs
@@ -5,6 +5,10 @@
 //! OpenHuman. What is left here is the one fact the crate cannot know: *which*
 //! directory this host keeps its catalog in.
 //!
+//! With a storage backend configured ([`crate::storage`]) each function is
+//! served by `tinyflows_drivers::catalog::FlowCatalogDocuments` instead, in
+//! the acting agent's scope — the same catalog on the document port.
+//!
 //! Every function below is that one substitution and nothing else. They are
 //! spelled out rather than replaced by a `pub use` so the existing
 //! `store::*(config, …)` call sites keep resolving unchanged, and so the seam
@@ -18,8 +22,23 @@ use tinyflows_catalog::{
     Flow, FlowRevision, FlowRun, FlowRunStep, FlowSuggestion, SuggestionStatus,
 };
 
+use tinyflows_drivers::catalog::FlowCatalogDocuments;
 pub use tinyflows_sqlite::flows::{FlowUpdateError, MAX_FLOW_RUNS_PER_FLOW};
 
+/// The document catalog for this call when the host configured a storage
+/// backend ([`crate::storage`]), in the acting agent's scope.
+pub(crate) fn documents(_config: &Config) -> Result<Option<FlowCatalogDocuments>> {
+    Ok(crate::storage::current_scoped()?
+        .map(|scoped| FlowCatalogDocuments::new(std::sync::Arc::clone(scoped.documents()))))
+}
+
+/// Runs a document-catalog call from this synchronous API.
+pub(crate) fn run<T: Send + 'static>(
+    future: impl std::future::Future<Output = Result<T>> + Send + 'static,
+) -> Result<T> {
+    crate::storage::block_on_anyhow(future)
+}
+
 /// Where this host keeps the flow catalog: `<workspace_dir>/flows`.
 ///
 /// `flows.db`, `checkpoints.db` and the `drafts/` directory are all created
@@ -31,12 +50,20 @@ pub fn dir(config: &Config) -> PathBuf {
 /// Binds [`tinyflows_sqlite::flows::upsert_flow`] to this host's catalog directory.
 #[inline]
 pub fn upsert_flow(config: &Config, flow: &Flow) -> Result<()> {
+    if let Some(docs) = documents(config)? {
+        let flow = flow.clone();
+        return run(async move { docs.upsert_flow(&flow).await });
+    }
     tinyflows_sqlite::flows::upsert_flow(&dir(config), flow)
 }
 
 /// Binds [`tinyflows_sqlite::flows::insert_duplicate_flow`] to this host's catalog directory.
 #[inline]
 pub fn insert_duplicate_flow(config: &Config, source: &Flow, new_name: String) -> Result<Flow> {
+    if let Some(docs) = documents(config)? {
+        let source = source.clone();
+        return run(async move { docs.insert_duplicate_flow(&source, new_name).await });
+    }
     tinyflows_sqlite::flows::insert_duplicate_flow(&dir(config), source, new_name)
 }
 
@@ -49,36 +76,60 @@ pub fn create_flow(
     require_approval: bool,
     enabled: bool,
 ) -> Result<Flow> {
+    if let Some(docs) = documents(config)? {
+        return run(async move {
+            docs.create_flow(name, graph, require_approval, enabled)
+                .await
+        });
+    }
     tinyflows_sqlite::flows::create_flow(&dir(config), name, graph, require_approval, enabled)
 }
 
 /// Binds [`tinyflows_sqlite::flows::get_flow`] to this host's catalog directory.
 #[inline]
 pub fn get_flow(config: &Config, id: &str) -> Result<Option<Flow>> {
+    if let Some(docs) = documents(config)? {
+        let id = id.to_string();
+        return run(async move { docs.get_flow(&id).await });
+    }
     tinyflows_sqlite::flows::get_flow(&dir(config), id)
 }
 
 /// Binds [`tinyflows_sqlite::flows::list_flows`] to this host's catalog directory.
 #[inline]
 pub fn list_flows(config: &Config) -> Result<(Vec<Flow>, usize)> {
+    if let Some(docs) = documents(config)? {
+        return run(async move { docs.list_flows().await });
+    }
     tinyflows_sqlite::flows::list_flows(&dir(config))
 }
 
 /// Binds [`tinyflows_sqlite::flows::list_enabled_flows`] to this host's catalog directory.
 #[inline]
 pub fn list_enabled_flows(config: &Config) -> Result<(Vec<Flow>, usize)> {
+    if let Some(docs) = documents(config)? {
+        return run(async move { docs.list_enabled_flows().await });
+    }
     tinyflows_sqlite::flows::list_enabled_flows(&dir(config))
 }
 
 /// Binds [`tinyflows_sqlite::flows::remove_flow`] to this host's catalog directory.
 #[inline]
 pub fn remove_flow(config: &Config, id: &str) -> Result<()> {
+    if let Some(docs) = documents(config)? {
+        let id = id.to_string();
+        return run(async move { docs.remove_flow(&id).await });
+    }
     tinyflows_sqlite::flows::remove_flow(&dir(config), id)
 }
 
 /// Binds [`tinyflows_sqlite::flows::set_enabled`] to this host's catalog directory.
 #[inline]
 pub fn set_enabled(config: &Config, id: &str, enabled: bool) -> Result<Flow> {
+    if let Some(docs) = documents(config)? {
+        let id = id.to_string();
+        return run(async move { docs.set_enabled(&id, enabled).await });
+    }
     tinyflows_sqlite::flows::set_enabled(&dir(config), id, enabled)
 }
 
@@ -94,6 +145,24 @@ pub fn update_flow_graph(
     force_disarm_if_automatic: bool,
     expected_updated_at: Option<&str>,
 ) -> std::result::Result<Flow, FlowUpdateError> {
+    if let Some(docs) = documents(config).map_err(FlowUpdateError::Store)? {
+        let id = id.to_string();
+        let expected_updated_at = expected_updated_at.map(str::to_string);
+        return crate::storage::block_on(async move {
+            Ok(docs
+                .update_flow_graph(
+                    &id,
+                    name,
+                    graph,
+                    require_approval,
+                    enabled_override,
+                    force_disarm_if_automatic,
+                    expected_updated_at.as_deref(),
+                )
+                .await)
+        })
+        .map_err(|error| FlowUpdateError::Store(error.into()))?;
+    }
     tinyflows_sqlite::flows::update_flow_graph(
         &dir(config),
         id,
@@ -109,6 +178,10 @@ pub fn update_flow_graph(
 /// Binds [`tinyflows_sqlite::flows::list_revisions`] to this host's catalog directory.
 #[inline]
 pub fn list_revisions(config: &Config, flow_id: &str, limit: usize) -> Result<Vec<FlowRevision>> {
+    if let Some(docs) = documents(config)? {
+        let flow_id = flow_id.to_string();
+        return run(async move { docs.list_revisions(&flow_id, limit).await });
+    }
     tinyflows_sqlite::flows::list_revisions(&dir(config), flow_id, limit)
 }
 
@@ -119,18 +192,33 @@ pub fn revision_by_id(
     flow_id: &str,
     revision_id: &str,
 ) -> Result<Option<FlowRevision>> {
+    if let Some(docs) = documents(config)? {
+        let flow_id = flow_id.to_string();
+        let revision_id = revision_id.to_string();
+        return run(async move { docs.revision_by_id(&flow_id, &revision_id).await });
+    }
     tinyflows_sqlite::flows::revision_by_id(&dir(config), flow_id, revision_id)
 }
 
 /// Binds [`tinyflows_sqlite::flows::record_run`] to this host's catalog directory.
 #[inline]
 pub fn record_run(config: &Config, id: &str, status: &str) -> Result<()> {
+    if let Some(docs) = documents(config)? {
+        let id = id.to_string();
+        let status = status.to_string();
+        return run(async move { docs.record_run(&id, &status).await });
+    }
     tinyflows_sqlite::flows::record_run(&dir(config), id, status)
 }
 
 /// Binds [`tinyflows_sqlite::flows::kv_get`] to this host's catalog directory.
 #[inline]
 pub fn kv_get(config: &Config, namespace: &str, key: &str) -> Result<Option<serde_json::Value>> {
+    if let Some(docs) = documents(config)? {
+        let namespace = namespace.to_string();
+        let key = key.to_string();
+        return run(async move { docs.kv_get(&namespace, &key).await });
+    }
     tinyflows_sqlite::flows::kv_get(&dir(config), namespace, key)
 }
 
@@ -142,6 +230,12 @@ pub fn kv_set(
     key: &str,
     value: &serde_json::Value,
 ) -> Result<()> {
+    if let Some(docs) = documents(config)? {
+        let namespace = namespace.to_string();
+        let key = key.to_string();
+        let value = value.clone();
+        return run(async move { docs.kv_set(&namespace, &key, &value).await });
+    }
     tinyflows_sqlite::flows::kv_set(&dir(config), namespace, key, value)
 }
 
@@ -154,12 +248,26 @@ pub fn insert_flow_run(
     thread_id: &str,
     started_at: &str,
 ) -> Result<()> {
+    if let Some(docs) = documents(config)? {
+        let id = id.to_string();
+        let flow_id = flow_id.to_string();
+        let thread_id = thread_id.to_string();
+        let started_at = started_at.to_string();
+        return run(async move {
+            docs.insert_flow_run(&id, &flow_id, &thread_id, &started_at)
+                .await
+        });
+    }
     tinyflows_sqlite::flows::insert_flow_run(&dir(config), id, flow_id, thread_id, started_at)
 }
 
 /// Binds [`tinyflows_sqlite::flows::prune_flow_runs`] to this host's catalog directory.
 #[inline]
 pub fn prune_flow_runs(config: &Config, flow_id: &str, keep: usize) -> Result<usize> {
+    if let Some(docs) = documents(config)? {
+        let flow_id = flow_id.to_string();
+        return run(async move { docs.prune_flow_runs(&flow_id, keep).await });
+    }
     tinyflows_sqlite::flows::prune_flow_runs(&dir(config), flow_id, keep)
 }
 
@@ -175,6 +283,27 @@ pub fn finish_flow_run(
     error: Option<&str>,
     graph_hash: Option<&str>,
 ) -> Result<bool> {
+    if let Some(docs) = documents(config)? {
+        let id = id.to_string();
+        let status = status.to_string();
+        let finished_at = finished_at.to_string();
+        let steps = steps.to_vec();
+        let pending_approvals = pending_approvals.to_vec();
+        let error = error.map(str::to_string);
+        let graph_hash = graph_hash.map(str::to_string);
+        return run(async move {
+            docs.finish_flow_run(
+                &id,
+                &status,
+                &finished_at,
+                &steps,
+                &pending_approvals,
+                error.as_deref(),
+                graph_hash.as_deref(),
+            )
+            .await
+        });
+    }
     tinyflows_sqlite::flows::finish_flow_run(
         &dir(config),
         id,
@@ -190,6 +319,11 @@ pub fn finish_flow_run(
 /// Binds [`tinyflows_sqlite::flows::upsert_flow_run_step`] to this host's catalog directory.
 #[inline]
 pub fn upsert_flow_run_step(config: &Config, run_id: &str, step: &FlowRunStep) -> Result<()> {
+    if let Some(docs) = documents(config)? {
+        let run_id = run_id.to_string();
+        let step = step.clone();
+        return run(async move { docs.upsert_flow_run_step(&run_id, &step).await });
+    }
     tinyflows_sqlite::flows::upsert_flow_run_step(&dir(config), run_id, step)
 }
 
@@ -201,6 +335,12 @@ pub fn expire_parked_runs(
     now: &str,
     error_msg: &str,
 ) -> Result<Vec<(String, String)>> {
+    if let Some(docs) = documents(config)? {
+        let cutoff = cutoff.to_string();
+        let now = now.to_string();
+        let error_msg = error_msg.to_string();
+        return run(async move { docs.expire_parked_runs(&cutoff, &now, &error_msg).await });
+    }
     tinyflows_sqlite::flows::expire_parked_runs(&dir(config), cutoff, now, error_msg)
 }
 
@@ -210,6 +350,10 @@ pub fn list_running_run_ids(
     config: &Config,
     started_before: &str,
 ) -> Result<Vec<(String, String)>> {
+    if let Some(docs) = documents(config)? {
+        let started_before = started_before.to_string();
+        return run(async move { docs.list_running_run_ids(&started_before).await });
+    }
     tinyflows_sqlite::flows::list_running_run_ids(&dir(config), started_before)
 }
 
@@ -225,6 +369,15 @@ pub fn force_run_status_for_test(
     status: &str,
     error: Option<&str>,
 ) -> Result<()> {
+    if let Some(docs) = documents(config)? {
+        let id = id.to_string();
+        let status = status.to_string();
+        let error = error.map(str::to_string);
+        return run(async move {
+            docs.force_run_status_for_test(&id, &status, error.as_deref())
+                .await
+        });
+    }
     tinyflows_sqlite::flows::force_run_status_for_test(&dir(config), id, status, error)
 }
 
@@ -239,6 +392,14 @@ pub fn force_corrupt_graph_json_for_test(
     flow_id: &str,
     raw_graph_json: &str,
 ) -> Result<()> {
+    if let Some(docs) = documents(config)? {
+        let flow_id = flow_id.to_string();
+        let raw_graph_json = raw_graph_json.to_string();
+        return run(async move {
+            docs.force_corrupt_graph_json_for_test(&flow_id, &raw_graph_json)
+                .await
+        });
+    }
     tinyflows_sqlite::flows::force_corrupt_graph_json_for_test(
         &dir(config),
         flow_id,
@@ -249,36 +410,61 @@ pub fn force_corrupt_graph_json_for_test(
 /// Binds [`tinyflows_sqlite::flows::mark_run_resuming`] to this host's catalog directory.
 #[inline]
 pub fn mark_run_resuming(config: &Config, id: &str) -> Result<bool> {
+    if let Some(docs) = documents(config)? {
+        let id = id.to_string();
+        return run(async move { docs.mark_run_resuming(&id).await });
+    }
     tinyflows_sqlite::flows::mark_run_resuming(&dir(config), id)
 }
 
 /// Binds [`tinyflows_sqlite::flows::mark_run_interrupted`] to this host's catalog directory.
 #[inline]
 pub fn mark_run_interrupted(config: &Config, id: &str, now: &str, reason: &str) -> Result<bool> {
+    if let Some(docs) = documents(config)? {
+        let id = id.to_string();
+        let now = now.to_string();
+        let reason = reason.to_string();
+        return run(async move { docs.mark_run_interrupted(&id, &now, &reason).await });
+    }
     tinyflows_sqlite::flows::mark_run_interrupted(&dir(config), id, now, reason)
 }
 
 /// Binds [`tinyflows_sqlite::flows::get_flow_run`] to this host's catalog directory.
 #[inline]
 pub fn get_flow_run(config: &Config, id: &str) -> Result<Option<FlowRun>> {
+    if let Some(docs) = documents(config)? {
+        let id = id.to_string();
+        return run(async move { docs.get_flow_run(&id).await });
+    }
     tinyflows_sqlite::flows::get_flow_run(&dir(config), id)
 }
 
 /// Binds [`tinyflows_sqlite::flows::list_flow_runs`] to this host's catalog directory.
 #[inline]
 pub fn list_flow_runs(config: &Config, flow_id: &str, limit: usize) -> Result<Vec<FlowRun>> {
+    if let Some(docs) = documents(config)? {
+        let flow_id = flow_id.to_string();
+        return run(async move { docs.list_flow_runs(&flow_id, limit).await });
+    }
     tinyflows_sqlite::flows::list_flow_runs(&dir(config), flow_id, limit)
 }
 
 /// Binds [`tinyflows_sqlite::flows::list_all_flow_runs`] to this host's catalog directory.
 #[inline]
 pub fn list_all_flow_runs(config: &Config, limit: usize) -> Result<Vec<FlowRun>> {
+    if let Some(docs) = documents(config)? {
+        return run(async move { docs.list_all_flow_runs(limit).await });
+    }
     tinyflows_sqlite::flows::list_all_flow_runs(&dir(config), limit)
 }
 
 /// Binds [`tinyflows_sqlite::flows::upsert_suggestions`] to this host's catalog directory.
 #[inline]
 pub fn upsert_suggestions(config: &Config, suggestions: &[FlowSuggestion]) -> Result<usize> {
+    if let Some(docs) = documents(config)? {
+        let suggestions = suggestions.to_vec();
+        return run(async move { docs.upsert_suggestions(&suggestions).await });
+    }
     tinyflows_sqlite::flows::upsert_suggestions(&dir(config), suggestions)
 }
 
@@ -289,11 +475,18 @@ pub fn list_suggestions(
     status: Option<SuggestionStatus>,
     limit: usize,
 ) -> Result<Vec<FlowSuggestion>> {
+    if let Some(docs) = documents(config)? {
+        return run(async move { docs.list_suggestions(status, limit).await });
+    }
     tinyflows_sqlite::flows::list_suggestions(&dir(config), status, limit)
 }
 
 /// Binds [`tinyflows_sqlite::flows::set_suggestion_status`] to this host's catalog directory.
 #[inline]
 pub fn set_suggestion_status(config: &Config, id: &str, status: SuggestionStatus) -> Result<bool> {
+    if let Some(docs) = documents(config)? {
+        let id = id.to_string();
+        return run(async move { docs.set_suggestion_status(&id, status).await });
+    }
     tinyflows_sqlite::flows::set_suggestion_status(&dir(config), id, status)
 }
diff --git a/crates/openhuman-core/src/flows/tinyflows/caps/ops.rs b/crates/openhuman-core/src/flows/tinyflows/caps/ops.rs
index 0244e907c06..176bb393e37 100644
--- a/crates/openhuman-core/src/flows/tinyflows/caps/ops.rs
+++ b/crates/openhuman-core/src/flows/tinyflows/caps/ops.rs
@@ -590,7 +590,8 @@ impl ToolInvoker for OpenHumanTools {
 /// and [`super::memory_adapter::OpenHumanMemory`] for `memory`, for its
 /// contract).
 ///
-/// `state_namespace` scopes the [`tinyflows_sqlite::flows::SqliteStateStore`] KV so two saved flows that
+/// `state_namespace` scopes the flow state KV ([`crate::flows::tinyflows::state::FlowState`]:
+/// `flows.db`, or the storage backend when one is configured) so two saved flows that
 /// use the same state key never read or overwrite each other — callers pass a
 /// per-flow namespace (e.g. `"flow:<id>"`). This KV namespace is unrelated to
 /// flow-scoped memory: `OpenHumanMemory` tags memory items with
@@ -621,10 +622,8 @@ pub fn build_capabilities(config: Arc<Config>, state_namespace: impl Into<String
             config: config.clone(),
             security: security.clone(),
         }),
-        state: Arc::new(tinyflows_sqlite::flows::SqliteStateStore::new(
-            crate::flows::store::dir(&config),
-            state_namespace,
-        )),
+        state: crate::flows::tinyflows::state::FlowState::open(&config, state_namespace)
+            .into_state_store(),
         agent: Some(Arc::new(OpenHumanAgentRunner {
             config: config.clone(),
         })),
@@ -659,7 +658,9 @@ pub fn build_capabilities(config: Arc<Config>, state_namespace: impl Into<String
 
 /// Opens the durable, cross-process checkpointer a `flows_run` uses via
 /// `tinyflows::engine::run_with_checkpointer` — this host's
-/// [`SqliteCheckpointer`], stored under `<workspace_dir>/flows/checkpoints.db`.
+/// [`SqliteCheckpointer`], stored under `<workspace_dir>/flows/checkpoints.db`,
+/// or, when the host configured a storage backend ([`crate::storage`]),
+/// `tinyflows_drivers::DriverCheckpointer` over it in the acting agent's scope.
 ///
 /// It became host-owned when tinyflows vendored its state-graph runtime and
 /// dropped the SQLite backend with it (tinyflows PR #43). The port keeps the
@@ -669,6 +670,12 @@ pub fn build_capabilities(config: Arc<Config>, state_namespace: impl Into<String
 pub fn open_flow_checkpointer(
     config: &Config,
 ) -> anyhow::Result<Arc<dyn tinyflows::engine::Checkpointer<serde_json::Value>>> {
+    if let Some(scoped) = crate::storage::current_scoped()? {
+        tracing::debug!(target: "flows", "[flows] opening checkpointer on the storage backend");
+        return Ok(Arc::new(tinyflows_drivers::DriverCheckpointer::<
+            serde_json::Value,
+        >::new(Arc::clone(scoped.documents()))));
+    }
     let db_path = config.workspace_dir.join("flows").join("checkpoints.db");
     if let Some(parent) = db_path.parent() {
         std::fs::create_dir_all(parent)
diff --git a/crates/openhuman-core/src/flows/tinyflows/mod.rs b/crates/openhuman-core/src/flows/tinyflows/mod.rs
index 95ce70fc2bd..457927b27d8 100644
--- a/crates/openhuman-core/src/flows/tinyflows/mod.rs
+++ b/crates/openhuman-core/src/flows/tinyflows/mod.rs
@@ -19,6 +19,7 @@ pub mod caps;
 pub mod langfuse_export;
 pub mod memory_adapter;
 pub mod observability;
+pub mod state;
 #[cfg(test)]
 #[path = "tinyflows_tests.rs"]
 mod tests;
diff --git a/crates/openhuman-core/src/flows/tinyflows/state.rs b/crates/openhuman-core/src/flows/tinyflows/state.rs
new file mode 100644
index 00000000000..482257958d0
--- /dev/null
+++ b/crates/openhuman-core/src/flows/tinyflows/state.rs
@@ -0,0 +1,113 @@
+//! A flow's namespaced state (`flow:<id>`), on whichever store this host uses.
+//!
+//! The engine reads and writes it through `StateStore` during a run, and the
+//! dedup settlement (`flows::bus::dedup_commit`) reads it back through
+//! `DedupKv` after the run, so both must name the same records. Without a
+//! storage backend that is `tinyflows_sqlite::flows::SqliteStateStore` over
+//! `flows/flows.db`; with one it is
+//! `tinyflows_drivers::catalog::FlowStateDocuments` in the acting agent's
+//! scope. When the scope cannot be resolved (SaaS mode with no acting agent)
+//! every call fails instead of falling back to the local file.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use serde_json::Value;
+use tinyflows::caps::StateStore;
+use tinyflows::error::{EngineError, Result as EngineResult};
+use tinyflows::nodes::control_flow::dedup_settle::DedupKv;
+use tinyflows_drivers::catalog::FlowStateDocuments;
+use tinyflows_sqlite::flows::SqliteStateStore;
+
+use crate::config::Config;
+
+/// One flow's state namespace on the store this call resolves to.
+#[derive(Debug, Clone)]
+pub enum FlowState {
+    /// `flows/flows.db` (no storage backend).
+    Sqlite(SqliteStateStore),
+    /// The storage backend's catalog.
+    Documents(FlowStateDocuments),
+    /// The storage scope could not be resolved; every call fails with this.
+    Unavailable(String),
+}
+
+impl FlowState {
+    /// The state of `namespace` for this call.
+    pub fn open(config: &Config, namespace: impl Into<String>) -> Self {
+        let namespace = namespace.into();
+        match crate::flows::store::documents(config) {
+            Ok(Some(catalog)) => Self::Documents(FlowStateDocuments::new(catalog, namespace)),
+            Ok(None) => Self::Sqlite(SqliteStateStore::new(
+                crate::flows::store::dir(config),
+                namespace,
+            )),
+            Err(error) => {
+                tracing::warn!(
+                    target: "flows",
+                    %error,
+                    "[flows] flow state unavailable: the storage scope did not resolve"
+                );
+                Self::Unavailable(error.to_string())
+            }
+        }
+    }
+
+    /// As the engine's `StateStore` capability.
+    pub fn into_state_store(self) -> Arc<dyn StateStore> {
+        Arc::new(self)
+    }
+
+    fn unavailable(reason: &str) -> String {
+        format!("flow state unavailable: {reason}")
+    }
+}
+
+#[async_trait]
+impl StateStore for FlowState {
+    async fn load(&self, key: &str) -> EngineResult<Option<Value>> {
+        match self {
+            Self::Sqlite(store) => store.load(key).await,
+            Self::Documents(store) => store.load(key).await,
+            Self::Unavailable(reason) => Err(EngineError::Capability(Self::unavailable(reason))),
+        }
+    }
+
+    async fn store(&self, key: &str, value: Value) -> EngineResult<()> {
+        match self {
+            Self::Sqlite(store) => store.store(key, value).await,
+            Self::Documents(store) => store.store(key, value).await,
+            Self::Unavailable(reason) => Err(EngineError::Capability(Self::unavailable(reason))),
+        }
+    }
+}
+
+impl DedupKv for FlowState {
+    fn kv_get(&self, key: &str) -> std::result::Result<Option<Value>, String> {
+        match self {
+            Self::Sqlite(store) => store.kv_get(key),
+            Self::Documents(store) => store.kv_get(key),
+            Self::Unavailable(reason) => Err(Self::unavailable(reason)),
+        }
+    }
+
+    fn kv_set(&self, key: &str, value: &Value) -> std::result::Result<(), String> {
+        match self {
+            Self::Sqlite(store) => store.kv_set(key, value),
+            Self::Documents(store) => store.kv_set(key, value),
+            Self::Unavailable(reason) => Err(Self::unavailable(reason)),
+        }
+    }
+
+    fn kv_delete(&self, key: &str) -> std::result::Result<(), String> {
+        match self {
+            Self::Sqlite(store) => store.kv_delete(key),
+            Self::Documents(store) => store.kv_delete(key),
+            Self::Unavailable(reason) => Err(Self::unavailable(reason)),
+        }
+    }
+}
+
+#[cfg(test)]
+#[path = "state_tests.rs"]
+mod tests;
diff --git a/crates/openhuman-core/src/flows/tinyflows/state_tests.rs b/crates/openhuman-core/src/flows/tinyflows/state_tests.rs
new file mode 100644
index 00000000000..a37ea0ea2c5
--- /dev/null
+++ b/crates/openhuman-core/src/flows/tinyflows/state_tests.rs
@@ -0,0 +1,74 @@
+use super::*;
+use crate::storage::{MemoryStorage, Scope, StorageBackend};
+use serde_json::json;
+use tinyflows_drivers::catalog::FlowCatalogDocuments;
+
+fn documents(namespace: &str) -> FlowState {
+    let scoped = MemoryStorage::new()
+        .for_scope(&Scope::new("local").unwrap())
+        .unwrap();
+    FlowState::Documents(FlowStateDocuments::new(
+        FlowCatalogDocuments::new(Arc::clone(scoped.documents())),
+        namespace,
+    ))
+}
+
+fn sqlite(dir: &std::path::Path, namespace: &str) -> FlowState {
+    FlowState::Sqlite(SqliteStateStore::new(dir, namespace))
+}
+
+/// What a run stores through `StateStore` is what dedup settlement reads
+/// through `DedupKv`, on both stores.
+async fn engine_and_dedup_share_records(state: FlowState) {
+    state.store("seen", json!(["a"])).await.unwrap();
+    assert_eq!(state.kv_get("seen").unwrap(), Some(json!(["a"])));
+    state.kv_set("seen", &json!(["a", "b"])).unwrap();
+    assert_eq!(state.load("seen").await.unwrap(), Some(json!(["a", "b"])));
+    state.kv_delete("seen").unwrap();
+    assert_eq!(state.load("seen").await.unwrap(), None);
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn the_document_store_backs_both_views() {
+    engine_and_dedup_share_records(documents("flow:a")).await;
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn the_sqlite_store_backs_both_views() {
+    let dir = tempfile::tempdir().unwrap();
+    engine_and_dedup_share_records(sqlite(dir.path(), "flow:a")).await;
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn namespaces_keep_flows_apart() {
+    let dir = tempfile::tempdir().unwrap();
+    let (a, b) = (sqlite(dir.path(), "flow:a"), sqlite(dir.path(), "flow:b"));
+    a.store("k", json!(1)).await.unwrap();
+    assert_eq!(b.load("k").await.unwrap(), None);
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn an_unresolved_scope_fails_every_call() {
+    let state = FlowState::Unavailable("no acting agent".to_string());
+    assert!(state.load("k").await.is_err());
+    assert!(state.store("k", json!(1)).await.is_err());
+    let error = state.kv_get("k").unwrap_err();
+    assert!(error.contains("no acting agent"), "{error}");
+    assert!(state.kv_set("k", &json!(1)).is_err());
+    assert!(state.kv_delete("k").is_err());
+}
+
+#[test]
+fn without_a_backend_the_flow_state_is_sqlite() {
+    let dir = tempfile::tempdir().unwrap();
+    let config = Config {
+        workspace_dir: dir.path().to_path_buf(),
+        ..Config::default()
+    };
+    if crate::storage::installed().is_none() {
+        assert!(matches!(
+            FlowState::open(&config, "flow:a"),
+            FlowState::Sqlite(_)
+        ));
+    }
+}
diff --git a/crates/openhuman-core/src/storage/README.md b/crates/openhuman-core/src/storage/README.md
index 7f17ebd280e..14e34a43d48 100644
--- a/crates/openhuman-core/src/storage/README.md
+++ b/crates/openhuman-core/src/storage/README.md
@@ -52,6 +52,15 @@ on `storage-mongodb`.
   approvals (`security::approval`), paired devices (`security::devices`),
   notifications (`desktop::notifications`) and task sources
   (`integrations::task_sources`).
+- tinyflows' own stores on the ports (`tinyflows-drivers`), picked per call
+  the same way: cron jobs and runs (`cron::store`, `CronDocuments`), the flow
+  catalog and drafts (`flows::store`, `flows::draft_store`,
+  `FlowCatalogDocuments`), per-flow engine state and dedup settlement
+  (`flows::tinyflows::state::FlowState`), and the flow-run checkpointer
+  (`DriverCheckpointer`). The delegation graph's checkpointer uses
+  tinyagents-graph's `DriverCheckpointer`.
+- `block_on_anyhow(future)`: `block_on` for those stores, whose errors are
+  `anyhow::Error` (a typed `FlowUpdateError` passes through unchanged).
 - Approvals (`security::approval`), through `store_documents.rs`.
 - Secrets (`storage::secrets`): the keyring's user secrets
   (`security::keyring::get` / `set` / `delete`) and the credential stores'
diff --git a/crates/openhuman-core/src/storage/mod.rs b/crates/openhuman-core/src/storage/mod.rs
index d6b1ab5ba2d..1ff9cf94e89 100644
--- a/crates/openhuman-core/src/storage/mod.rs
+++ b/crates/openhuman-core/src/storage/mod.rs
@@ -182,6 +182,21 @@ where
     bridge.run(future)?
 }
 
+/// [`block_on`] for a future whose error is an `anyhow::Error` — the stores
+/// tinyflows puts on the ports return those, with typed errors (such as
+/// `FlowUpdateError`) a caller may downcast, so they pass through unchanged.
+///
+/// # Errors
+///
+/// The future's own error, or the bridge's when it cannot start.
+pub fn block_on_anyhow<T, F>(future: F) -> anyhow::Result<T>
+where
+    F: Future<Output = anyhow::Result<T>> + Send + 'static,
+    T: Send + 'static,
+{
+    block_on(async move { Ok(future.await) })?
+}
+
 /// Whether a backend with this driver name may be shared by several
 /// processes at once. A MongoDB database can be; SQLite files, the memory
 /// driver and plain files belong to the one process that opened them.
diff --git a/tests/storage_flows_e2e.rs b/tests/storage_flows_e2e.rs
new file mode 100644
index 00000000000..b2f0c032e13
--- /dev/null
+++ b/tests/storage_flows_e2e.rs
@@ -0,0 +1,92 @@
+//! Cron jobs, the flow catalog and flow state on a configured storage
+//! backend, end to end through the core's public functions.
+//!
+//! Its own test binary because it installs a backend into the process-wide
+//! storage slot, which would reroute every other suite's stores in a shared
+//! process. One test, so nothing in this binary races the slot either.
+
+use std::sync::Arc;
+
+use openhuman_core::config::Config;
+use openhuman_core::cron::{self, Schedule};
+use openhuman_core::flows;
+use openhuman_core::flows::tinyflows::state::FlowState;
+use serde_json::json;
+
+#[tokio::test(flavor = "multi_thread")]
+async fn a_configured_backend_holds_cron_flows_and_flow_state() {
+    let workspace = tempfile::tempdir().unwrap();
+    let config = Config {
+        workspace_dir: workspace.path().to_path_buf(),
+        ..Config::default()
+    };
+    openhuman_core::storage::install(Arc::new(openhuman_core::storage::MemoryStorage::new()));
+
+    // Cron: the store functions run on blocking threads, as the scheduler does.
+    let cron_config = config.clone();
+    tokio::task::spawn_blocking(move || {
+        let job = cron::add_shell_job(
+            &cron_config,
+            Some("tick".to_string()),
+            Schedule::Every { every_ms: 60_000 },
+            "echo hi",
+        )
+        .unwrap();
+        assert_eq!(cron::list_jobs(&cron_config).unwrap().len(), 1);
+        let now = chrono::Utc::now();
+        cron::record_run(&cron_config, &job.id, now, now, "ok", Some("hi"), 3).unwrap();
+        assert_eq!(cron::list_runs(&cron_config, &job.id, 10).unwrap().len(), 1);
+        cron::remove_job(&cron_config, &job.id).unwrap();
+        assert!(cron::list_jobs(&cron_config).unwrap().is_empty());
+    })
+    .await
+    .unwrap();
+
+    // The flow catalog, through the RPC operations.
+    let graph = json!({
+        "nodes": [
+            { "id": "t", "kind": "trigger", "name": "Manual" },
+            { "id": "a", "kind": "agent", "name": "Summarize", "config": { "prompt": "hi" } }
+        ],
+        "edges": [ { "from_node": "t", "to_node": "a" } ]
+    });
+    let flow = flows::ops::flows_create(&config, "Digest".to_string(), graph, false)
+        .await
+        .unwrap()
+        .value;
+    let listed = flows::ops::flows_list(&config).await.unwrap().value;
+    assert_eq!(listed.len(), 1);
+    assert_eq!(listed[0].id, flow.id);
+
+    // Flow state resolves to the backend, and its records read back through
+    // the catalog's KV (the engine/dedup sharing is unit-tested in core).
+    let state_config = config.clone();
+    tokio::task::spawn_blocking(move || {
+        assert!(matches!(
+            FlowState::open(&state_config, "flow:digest"),
+            FlowState::Documents(_)
+        ));
+        flows::kv_set(&state_config, "flow:digest", "seen", &json!(["a"])).unwrap();
+        assert_eq!(
+            flows::kv_get(&state_config, "flow:digest", "seen").unwrap(),
+            Some(json!(["a"]))
+        );
+    })
+    .await
+    .unwrap();
+
+    for path in ["cron/jobs.db", "flows/flows.db"] {
+        assert!(
+            !workspace.path().join(path).exists(),
+            "{path} was not written"
+        );
+    }
+
+    // Without a backend the classic databases are back in use.
+    assert!(openhuman_core::storage::clear());
+    assert!(flows::ops::flows_list(&config)
+        .await
+        .unwrap()
+        .value
+        .is_empty());
+}
diff --git a/vendor/tinyflows b/vendor/tinyflows
index f9df351f2cd..fe508caddc6 160000
--- a/vendor/tinyflows
+++ b/vendor/tinyflows
@@ -1 +1 @@
-Subproject commit f9df351f2cd73db99dd9ea19560001a4c35a12bc
+Subproject commit fe508caddc62402ea6ac5eda3e2db8943eb39d10