From fc3a68c24491688871bf77e46c7e0411d918d0ba Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 08:09:52 +0300 Subject: [PATCH 01/19] chore(checkpoint): add drivers module Add the checkpoint drivers module to the graph checkpoint package. Auto-committed-on: dragonfly Co-authored-by: Medulla --- .../tinyflows/src/graph/checkpoint/drivers.rs | 407 ++++++++++++++++++ 1 file changed, 407 insertions(+) create mode 100644 crates/tinyflows/src/graph/checkpoint/drivers.rs diff --git a/crates/tinyflows/src/graph/checkpoint/drivers.rs b/crates/tinyflows/src/graph/checkpoint/drivers.rs new file mode 100644 index 00000000..1c019f38 --- /dev/null +++ b/crates/tinyflows/src/graph/checkpoint/drivers.rs @@ -0,0 +1,407 @@ +//! A [`Checkpointer`] over `tinystoragedrivers` document ports. +//! +//! [`DriverCheckpointer`] puts graph checkpoints, their pending writes and the +//! on whatever backend the host opened (SQLite on a +//! desktop, MongoDB in the cloud, memory in tests). The document handle is +//! already bound to a tenant scope, so two tenants' threads never meet even +//! when they share a thread id. +//! +//! # Layout +//! +//! Three collections, named from a prefix (default `graph`): +//! +//! - `_checkpoints`: one document per stored checkpoint, +//! `{thread, namespace, seq, checkpoint_id, record}` (`namespace` is the +//! encoded subgraph namespace, so scoped reads filter on it). `seq` is a per-thread insertion +//! counter, so listing is a query sorted on it and duplicate checkpoint ids +//! resolve to the latest write, exactly like the append-only backends. +//! - `_threads`: one counter document per thread, advanced with a +//! compare-and-swap so concurrent writers never share a `seq`. +//! - `_writes`: one document per `(thread, namespace, checkpoint)` +//! holding the merged pending writes (see [`merge_writes`]). + +use std::marker::PhantomData; +use std::sync::Arc; + +use async_trait::async_trait; +use serde::Serialize; +use serde::de::DeserializeOwned; +use serde_json::{Value, json}; +use crate::graph::error::{GraphError, Result}; +use crate::graph::ids::CheckpointId; +use tinystoragedrivers_core::{ + CollectionSpec, DocumentStore, DocumentStoreExt, ErrorKind, Filter, IndexSpec, Precondition, + Query, Sort, StorageError, Versioned, +}; +use tokio::sync::OnceCell; + +use super::in_memory::require_checkpoint_id; +use super::{Checkpoint, CheckpointConfig, CheckpointMetadata, Checkpointer, PendingWrite, merge_writes}; + +/// How many times a compare-and-swap loop retries before giving up. +const CAS_ATTEMPTS: usize = 64; + +/// Map a storage driver failure onto the graph's checkpoint error. +fn map_error(error: StorageError) -> GraphError { + GraphError::Checkpoint(format!("storage driver: {error}")) +} + +/// Longest id stored as is; longer ones are hashed. +const MAX_KEY_LEN: usize = 400; + +/// A document id for `parts`: length-prefixed so no two tuples collide, and +/// replaced by its SHA-256 when it would exceed the driver's id limit. +fn key(parts: &[&str]) -> String { + let joined: String = parts + .iter() + .map(|part| format!("{}:{part}", part.len())) + .collect::>() + .join("/"); + if joined.len() <= MAX_KEY_LEN { + joined + } else { + use sha2::{Digest, Sha256}; + let digest = Sha256::digest(joined.as_bytes()); + let hex: String = digest.iter().map(|byte| format!("{byte:02x}")).collect(); + format!("h:{hex}") + } +} + +/// `namespace` as one string, injectively: each component is +/// length-prefixed, so `["a", "b"]` and `["a/b"]` never meet. +fn namespace_key(namespace: &[String]) -> String { + namespace + .iter() + .map(|part| format!("{}:{part};", part.len())) + .collect() +} + +/// A [`Checkpointer`] that stores everything in a driver [`DocumentStore`]. +pub struct DriverCheckpointer { + docs: Arc, + checkpoints: String, + threads: String, + writes: String, + declared: Arc>, + _state: PhantomData State>, +} + +impl Clone for DriverCheckpointer { + fn clone(&self) -> Self { + Self { + docs: Arc::clone(&self.docs), + checkpoints: self.checkpoints.clone(), + threads: self.threads.clone(), + writes: self.writes.clone(), + declared: Arc::clone(&self.declared), + _state: PhantomData, + } + } +} + +impl std::fmt::Debug for DriverCheckpointer { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("DriverCheckpointer") + .field("checkpoints", &self.checkpoints) + .finish_non_exhaustive() + } +} + +impl DriverCheckpointer { + /// Store checkpoints in the `graph_*` collections of `docs`. + pub fn new(docs: Arc) -> Self { + Self::with_prefix(docs, "graph") + } + + /// Store checkpoints in the `_*` collections of `docs`, so + /// several independent checkpointers can share one backend. + pub fn with_prefix(docs: Arc, prefix: &str) -> Self { + Self { + docs, + checkpoints: format!("{prefix}_checkpoints"), + threads: format!("{prefix}_threads"), + writes: format!("{prefix}_writes"), + declared: Arc::new(OnceCell::new()), + _state: PhantomData, + } + } + + async fn declared(&self) -> Result<()> { + self.declared + .get_or_try_init(|| async { + let specs = [ + CollectionSpec::new(&self.checkpoints) + .index(IndexSpec::new("by_thread_seq", ["thread", "seq"])) + .index(IndexSpec::new("by_thread_id", ["thread", "checkpoint_id"])) + .index(IndexSpec::new( + "by_thread_namespace", + ["thread", "namespace", "seq"], + )), + CollectionSpec::new(&self.threads), + CollectionSpec::new(&self.writes) + .index(IndexSpec::new("by_thread", ["thread"])), + ]; + for spec in &specs { + self.docs.ensure_collection(spec).await.map_err(map_error)?; + } + Ok::<(), GraphError>(()) + }) + .await + .map(|_| ()) + } + + /// Reserve the next insertion sequence number for `thread`. + async fn next_seq(&self, thread: &str) -> Result { + let id = key(&[thread]); + for _ in 0..CAS_ATTEMPTS { + let current = self.docs.get(&self.threads, &id).await.map_err(map_error)?; + let (seq, precondition) = match ¤t { + Some(found) => ( + found.doc.get("next").and_then(Value::as_u64).unwrap_or(0), + found.unchanged(), + ), + None => (0, Precondition::Absent), + }; + let doc = json!({ "thread": thread, "next": seq + 1 }); + match self.docs.put(&self.threads, &id, doc, precondition).await { + Ok(_) => return Ok(seq), + Err(error) if error.kind() == ErrorKind::Conflict => continue, + Err(error) => return Err(map_error(error)), + } + } + Err(GraphError::Checkpoint(format!( + "could not reserve a checkpoint sequence for thread `{thread}`" + ))) + } + + /// Every stored checkpoint document of `thread`, in insertion order. + async fn thread_docs(&self, thread: &str) -> Result>> { + self.declared().await?; + let query = Query::filter(Filter::eq("thread", thread)).sort(Sort::asc("seq")); + self.docs + .query_all(&self.checkpoints, &query) + .await + .map_err(map_error) + } + + fn writes_id(thread: &str, namespace: &[String], checkpoint_id: &str) -> String { + key(&[thread, &namespace_key(namespace), checkpoint_id]) + } + + async fn drop_writes(&self, filter: Filter) -> Result<()> { + self.docs + .delete_where(&self.writes, &filter) + .await + .map(|_| ()) + .map_err(map_error) + } +} + +impl DriverCheckpointer +where + State: DeserializeOwned, +{ + fn decode(stored: Versioned) -> Result> { + let record = stored.doc.get("record").cloned().unwrap_or(Value::Null); + serde_json::from_value(record).map_err(|error| { + GraphError::Checkpoint(format!("storage driver checkpointer: decode record: {error}")) + }) + } +} + +#[async_trait] +impl Checkpointer for DriverCheckpointer +where + State: Serialize + DeserializeOwned + Send + Sync + 'static, +{ + async fn put(&self, checkpoint: Checkpoint) -> Result { + self.declared().await?; + let id = CheckpointId::new(checkpoint.checkpoint_id.clone()); + let seq = self.next_seq(&checkpoint.thread_id).await?; + let doc = json!({ + "thread": checkpoint.thread_id, + "namespace": namespace_key(&checkpoint.namespace), + "seq": seq, + "checkpoint_id": checkpoint.checkpoint_id, + "record": serde_json::to_value(&checkpoint)?, + }); + let doc_id = key(&[&checkpoint.thread_id, &format!("{seq:020}")]); + self.docs + .put(&self.checkpoints, &doc_id, doc, Precondition::Absent) + .await + .map_err(map_error)?; + Ok(id) + } + + async fn get( + &self, + thread_id: &str, + checkpoint_id: Option<&str>, + ) -> Result>> { + self.declared().await?; + let mut filter = Filter::eq("thread", thread_id); + if let Some(id) = checkpoint_id { + filter = filter.and(Filter::eq("checkpoint_id", id)); + } + let query = Query::filter(filter).sort(Sort::desc("seq")).limit(1); + let page = self + .docs + .query(&self.checkpoints, &query) + .await + .map_err(map_error)?; + page.items.into_iter().next().map(Self::decode).transpose() + } + + /// One indexed query on the stored namespace, so a parent run and a + /// subgraph sharing a thread never load each other's checkpoints, even + /// when they reuse a checkpoint id. + async fn get_scoped( + &self, + thread_id: &str, + checkpoint_id: Option<&str>, + namespace: &[String], + ) -> Result>> { + self.declared().await?; + let mut filter = + Filter::eq("thread", thread_id).and(Filter::eq("namespace", namespace_key(namespace))); + if let Some(id) = checkpoint_id { + filter = filter.and(Filter::eq("checkpoint_id", id)); + } + let query = Query::filter(filter).sort(Sort::desc("seq")).limit(1); + let page = self + .docs + .query(&self.checkpoints, &query) + .await + .map_err(map_error)?; + page.items.into_iter().next().map(Self::decode).transpose() + } + + async fn list(&self, thread_id: &str) -> Result> { + Ok(self + .get_thread(thread_id) + .await? + .iter() + .map(Checkpoint::to_metadata) + .collect()) + } + + async fn get_thread(&self, thread_id: &str) -> Result>> { + self.thread_docs(thread_id) + .await? + .into_iter() + .map(Self::decode) + .collect() + } + + async fn list_threads(&self) -> Result> { + self.declared().await?; + let counters = self + .docs + .query_all(&self.threads, &Query::all()) + .await + .map_err(map_error)?; + let mut threads = Vec::new(); + for counter in counters { + let Some(thread) = counter.doc.get("thread").and_then(Value::as_str) else { + continue; + }; + let live = self + .docs + .count(&self.checkpoints, &Filter::eq("thread", thread)) + .await + .map_err(map_error)?; + if live > 0 { + threads.push(thread.to_owned()); + } + } + Ok(threads) + } + + async fn delete_thread(&self, thread_id: &str) -> Result<()> { + self.declared().await?; + self.docs + .delete_where(&self.checkpoints, &Filter::eq("thread", thread_id)) + .await + .map_err(map_error)?; + // Keep the sequence counter: a later thread of the same name continues + // after it, so its records never interleave with stale cursors. + self.drop_writes(Filter::eq("thread", thread_id)).await + } + + async fn delete_checkpoints(&self, thread_id: &str, ids: &[String]) -> Result { + if ids.is_empty() { + return Ok(0); + } + self.declared().await?; + let filter = Filter::eq("thread", thread_id).and(Filter::one_of( + "checkpoint_id", + ids.iter().map(String::as_str), + )); + let removed = self + .docs + .delete_where(&self.checkpoints, &filter) + .await + .map_err(map_error)?; + self.drop_writes(Filter::eq("thread", thread_id).and(Filter::one_of( + "checkpoint_id", + ids.iter().map(String::as_str), + ))) + .await?; + Ok(usize::try_from(removed).unwrap_or(usize::MAX)) + } + + async fn put_writes(&self, config: &CheckpointConfig, writes: &[PendingWrite]) -> Result<()> { + let checkpoint_id = require_checkpoint_id(config)?; + if writes.is_empty() { + return Ok(()); + } + self.declared().await?; + let id = Self::writes_id(&config.thread_id, &config.namespace, &checkpoint_id); + for _ in 0..CAS_ATTEMPTS { + let current = self.docs.get(&self.writes, &id).await.map_err(map_error)?; + let (mut stored, precondition): (Vec, Precondition) = match ¤t { + Some(found) => ( + serde_json::from_value( + found.doc.get("writes").cloned().unwrap_or(Value::Null), + )?, + found.unchanged(), + ), + None => (Vec::new(), Precondition::Absent), + }; + merge_writes(&mut stored, writes); + let doc = json!({ + "thread": config.thread_id, + "namespace": config.namespace, + "checkpoint_id": checkpoint_id, + "writes": serde_json::to_value(&stored)?, + }); + match self.docs.put(&self.writes, &id, doc, precondition).await { + Ok(_) => return Ok(()), + Err(error) if error.kind() == ErrorKind::Conflict => continue, + Err(error) => return Err(map_error(error)), + } + } + Err(GraphError::Checkpoint(format!( + "could not record pending writes for thread `{}`", + config.thread_id + ))) + } + + async fn get_writes(&self, config: &CheckpointConfig) -> Result> { + let Some(checkpoint_id) = self.resolve_write_target(config).await? else { + return Ok(Vec::new()); + }; + self.declared().await?; + let id = Self::writes_id(&config.thread_id, &config.namespace, &checkpoint_id); + let Some(found) = self.docs.get(&self.writes, &id).await.map_err(map_error)? else { + return Ok(Vec::new()); + }; + Ok(serde_json::from_value( + found.doc.get("writes").cloned().unwrap_or(Value::Null), + )?) + } +} + +#[cfg(test)] +#[path = "drivers_tests.rs"] +mod tests; From 7a0fba4e34064575ff5a8ec9b454a9b12c64edeb Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 08:10:02 +0300 Subject: [PATCH 02/19] chore(checkpoint): add drivers module Add the checkpoint drivers module to the graph checkpoint package. Auto-committed-on: dragonfly Co-authored-by: Medulla --- .../tinyflows/src/graph/checkpoint/drivers.rs | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/crates/tinyflows/src/graph/checkpoint/drivers.rs b/crates/tinyflows/src/graph/checkpoint/drivers.rs index 1c019f38..41c310c8 100644 --- a/crates/tinyflows/src/graph/checkpoint/drivers.rs +++ b/crates/tinyflows/src/graph/checkpoint/drivers.rs @@ -1,14 +1,16 @@ //! A [`Checkpointer`] over `tinystoragedrivers` document ports. //! -//! [`DriverCheckpointer`] puts graph checkpoints, their pending writes and the -//! on whatever backend the host opened (SQLite on a -//! desktop, MongoDB in the cloud, memory in tests). The document handle is +//! [`DriverCheckpointer`] puts graph checkpoints and their pending writes on +//! whatever backend the host opened (SQLite on a desktop, MongoDB in the +//! cloud, memory in tests). The document handle is //! already bound to a tenant scope, so two tenants' threads never meet even //! when they share a thread id. //! //! # Layout //! -//! Three collections, named from a prefix (default `graph`): +//! Three collections, named from a prefix (default [`DEFAULT_PREFIX`], kept +//! apart from TinyAgents' own `graph_*` checkpointer collections so both can +//! share one backend): //! //! - `_checkpoints`: one document per stored checkpoint, //! `{thread, namespace, seq, checkpoint_id, record}` (`namespace` is the @@ -76,6 +78,9 @@ fn namespace_key(namespace: &[String]) -> String { .collect() } +/// The collection prefix [`DriverCheckpointer::new`] uses. +pub const DEFAULT_PREFIX: &str = "flows_graph"; + /// A [`Checkpointer`] that stores everything in a driver [`DocumentStore`]. pub struct DriverCheckpointer { docs: Arc, @@ -108,9 +113,9 @@ impl std::fmt::Debug for DriverCheckpointer { } impl DriverCheckpointer { - /// Store checkpoints in the `graph_*` collections of `docs`. + /// Store checkpoints in the `flows_graph_*` collections of `docs`. pub fn new(docs: Arc) -> Self { - Self::with_prefix(docs, "graph") + Self::with_prefix(docs, DEFAULT_PREFIX) } /// Store checkpoints in the `_*` collections of `docs`, so From bf8168048162673419035df6a50f8a652d42ad42 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 08:10:25 +0300 Subject: [PATCH 03/19] feat(caps): add capability drivers for checkpointing Introduce a capability layer that exposes checkpoint drivers through the caps module, letting graphs persist and restore state via pluggable backends. The checkpoint module now routes through these drivers instead of hardcoding driver behaviour. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinyflows/Cargo.toml | 10 ++ crates/tinyflows/src/caps/drivers.rs | 99 +++++++++++++++++++ crates/tinyflows/src/caps/mod.rs | 5 + .../tinyflows/src/graph/checkpoint/drivers.rs | 4 +- crates/tinyflows/src/graph/checkpoint/mod.rs | 4 + 5 files changed, 120 insertions(+), 2 deletions(-) create mode 100644 crates/tinyflows/src/caps/drivers.rs diff --git a/crates/tinyflows/Cargo.toml b/crates/tinyflows/Cargo.toml index bb4ce305..d84deb40 100644 --- a/crates/tinyflows/Cargo.toml +++ b/crates/tinyflows/Cargo.toml @@ -48,8 +48,18 @@ font8x8 = { version = "0.3", default-features = false, features = ["unicode"], o jaq-core = "3.1.0" jaq-std = { version = "3.0.1", default-features = false } jaq-json = { version = "2.0.1", features = ["serde"] } +# `storage-drivers` only: the storage ports (no database client). Named by git +# tag so a host that vendors tinystoragedrivers once (OpenHuman, through +# tinyagents) can `[patch]` this URL onto its single copy, as it does for +# tinytools. +tinystoragedrivers-core = { git = "https://github.com/tinyhumansai/tinystoragedrivers", tag = "v0.4.0", optional = true } [features] +# Off by default; `DriverStateStore` and `graph::checkpoint::DriverCheckpointer` +# over `tinystoragedrivers` document ports, so a host keeps run state and graph +# checkpoints in whichever backend it opened (SQLite, MongoDB, memory). Needs +# Rust 1.88, the storage crates' MSRV. +storage-drivers = ["dep:tinystoragedrivers-core"] # Off by default; enables in-memory mock capability impls for downstream tests # and examples. Mocks are always available inside this crate's own tests. default = [] diff --git a/crates/tinyflows/src/caps/drivers.rs b/crates/tinyflows/src/caps/drivers.rs new file mode 100644 index 00000000..65e57314 --- /dev/null +++ b/crates/tinyflows/src/caps/drivers.rs @@ -0,0 +1,99 @@ +//! [`StateStore`] over a `tinystoragedrivers` [`DocumentStore`]. +//! +//! A host that has opened a storage backend hands the engine a scoped +//! document handle and wraps it here; every key is one document +//! `{ "key": , "value": }` in one collection (default +//! [`DriverStateStore::DEFAULT_COLLECTION`]). The handle is already bound to a +//! tenant scope, so two tenants' runs never read each other's state. + +use std::sync::Arc; + +use async_trait::async_trait; +use serde_json::{Value, json}; +use tinystoragedrivers_core::{ + CollectionSpec, DocumentStore, ErrorKind, Precondition, StorageError, +}; +use tokio::sync::OnceCell; + +use super::StateStore; +use crate::error::{Result, TinyFlowsError}; + +/// A [`StateStore`] keeping each key in a driver [`DocumentStore`]. +#[derive(Debug, Clone)] +pub struct DriverStateStore { + docs: Arc, + collection: String, + declared: Arc>, +} + +impl DriverStateStore { + /// The collection [`Self::new`] uses. + pub const DEFAULT_COLLECTION: &'static str = "flows_state"; + + /// State in [`Self::DEFAULT_COLLECTION`] of `docs`. + pub fn new(docs: Arc) -> Self { + Self::with_collection(docs, Self::DEFAULT_COLLECTION) + } + + /// State in `collection` of `docs`, so independent stores can share one + /// backend. + pub fn with_collection(docs: Arc, collection: impl Into) -> Self { + Self { + docs, + collection: collection.into(), + declared: Arc::new(OnceCell::new()), + } + } + + async fn declared(&self) -> Result<()> { + self.declared + .get_or_try_init(|| async { + self.docs + .ensure_collection(&CollectionSpec::new(&self.collection)) + .await + .map_err(map_error) + }) + .await + .map(|_| ()) + } +} + +/// A state key as a document id: keys longer than the driver's id limit are +/// refused rather than truncated or hashed, so two keys never share a slot. +fn map_error(error: StorageError) -> TinyFlowsError { + match error.kind() { + ErrorKind::InvalidInput => TinyFlowsError::Validation(format!("state store: {error}")), + _ => TinyFlowsError::Capability(format!("state store: {error}")), + } +} + +#[async_trait] +impl StateStore for DriverStateStore { + async fn load(&self, key: &str) -> Result> { + self.declared().await?; + let found = self + .docs + .get(&self.collection, key) + .await + .map_err(map_error)?; + Ok(found.and_then(|mut stored| stored.doc.get_mut("value").map(Value::take))) + } + + async fn store(&self, key: &str, value: Value) -> Result<()> { + self.declared().await?; + self.docs + .put( + &self.collection, + key, + json!({ "key": key, "value": value }), + Precondition::None, + ) + .await + .map(|_| ()) + .map_err(map_error) + } +} + +#[cfg(test)] +#[path = "drivers_tests.rs"] +mod tests; diff --git a/crates/tinyflows/src/caps/mod.rs b/crates/tinyflows/src/caps/mod.rs index 1f20f42e..7da84644 100644 --- a/crates/tinyflows/src/caps/mod.rs +++ b/crates/tinyflows/src/caps/mod.rs @@ -6,6 +6,8 @@ //! curated Composio tools, `HttpRequestTool`, and sandboxed code runtimes. pub mod agent; +#[cfg(feature = "storage-drivers")] +mod drivers; pub mod approval; #[cfg(any(test, feature = "host-caps"))] pub mod host; @@ -205,6 +207,9 @@ pub trait StateStore: Send + Sync { async fn store(&self, key: &str, value: Value) -> Result<()>; } +#[cfg(feature = "storage-drivers")] +pub use drivers::DriverStateStore; + /// The bundle of capabilities handed to the engine for a run. /// /// Construct one per run from the host's concrete implementations. It carries diff --git a/crates/tinyflows/src/graph/checkpoint/drivers.rs b/crates/tinyflows/src/graph/checkpoint/drivers.rs index 41c310c8..ba954987 100644 --- a/crates/tinyflows/src/graph/checkpoint/drivers.rs +++ b/crates/tinyflows/src/graph/checkpoint/drivers.rs @@ -37,8 +37,8 @@ use tinystoragedrivers_core::{ }; use tokio::sync::OnceCell; -use super::in_memory::require_checkpoint_id; -use super::{Checkpoint, CheckpointConfig, CheckpointMetadata, Checkpointer, PendingWrite, merge_writes}; +use super::{ + require_checkpoint_id,Checkpoint, CheckpointConfig, CheckpointMetadata, Checkpointer, PendingWrite, merge_writes}; /// How many times a compare-and-swap loop retries before giving up. const CAS_ATTEMPTS: usize = 64; diff --git a/crates/tinyflows/src/graph/checkpoint/mod.rs b/crates/tinyflows/src/graph/checkpoint/mod.rs index 7d7d7401..11884bdd 100644 --- a/crates/tinyflows/src/graph/checkpoint/mod.rs +++ b/crates/tinyflows/src/graph/checkpoint/mod.rs @@ -13,9 +13,13 @@ //! at superstep boundaries only — never mid-node — so resuming always reruns a //! node from its start. +#[cfg(feature = "storage-drivers")] +mod drivers; mod file; mod types; +#[cfg(feature = "storage-drivers")] +pub use drivers::{DEFAULT_PREFIX as DRIVER_DEFAULT_PREFIX, DriverCheckpointer}; pub use file::FileCheckpointer; pub use types::{ BarrierArrivals, Checkpoint, CheckpointConfig, CheckpointMetadata, CheckpointSource, From a217c43d670a854673700e242f1d1db4e8d6c112 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 08:11:59 +0300 Subject: [PATCH 04/19] fix(tinyflows): map storage errors to capability errors The driver state store now reports every storage failure as a capability error instead of splitting invalid input into a validation error, since a key the driver cannot store as an id is a capability limit rather than caller input. The storage-drivers feature also enables sha2, which the driver crate needs for its id handling. Auto-committed-on: dragonfly Co-authored-by: Medulla --- Cargo.lock | 12 +++++ crates/tinyflows/Cargo.toml | 2 +- crates/tinyflows/src/caps/drivers.rs | 19 ++++---- crates/tinyflows/src/caps/drivers_tests.rs | 48 +++++++++++++++++++ .../src/graph/checkpoint/drivers_tests.rs | 0 5 files changed, 69 insertions(+), 12 deletions(-) create mode 100644 crates/tinyflows/src/caps/drivers_tests.rs create mode 100644 crates/tinyflows/src/graph/checkpoint/drivers_tests.rs diff --git a/Cargo.lock b/Cargo.lock index 00af684e..c117610a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2650,6 +2650,7 @@ dependencies = [ "sha2", "tempfile", "thiserror", + "tinystoragedrivers-core", "tokio", "tracing", ] @@ -2721,6 +2722,17 @@ dependencies = [ "uuid", ] +[[package]] +name = "tinystoragedrivers-core" +version = "0.4.0" +source = "git+https://github.com/tinyhumansai/tinystoragedrivers?tag=v0.4.0#f7d603878985966581f90a474eb36bfa1e1ba0f3" +dependencies = [ + "async-trait", + "serde", + "serde_json", + "thiserror", +] + [[package]] name = "tinystr" version = "0.8.4" diff --git a/crates/tinyflows/Cargo.toml b/crates/tinyflows/Cargo.toml index d84deb40..4cacce52 100644 --- a/crates/tinyflows/Cargo.toml +++ b/crates/tinyflows/Cargo.toml @@ -59,7 +59,7 @@ tinystoragedrivers-core = { git = "https://github.com/tinyhumansai/tinystoragedr # over `tinystoragedrivers` document ports, so a host keeps run state and graph # checkpoints in whichever backend it opened (SQLite, MongoDB, memory). Needs # Rust 1.88, the storage crates' MSRV. -storage-drivers = ["dep:tinystoragedrivers-core"] +storage-drivers = ["dep:tinystoragedrivers-core", "dep:sha2"] # Off by default; enables in-memory mock capability impls for downstream tests # and examples. Mocks are always available inside this crate's own tests. default = [] diff --git a/crates/tinyflows/src/caps/drivers.rs b/crates/tinyflows/src/caps/drivers.rs index 65e57314..e984920f 100644 --- a/crates/tinyflows/src/caps/drivers.rs +++ b/crates/tinyflows/src/caps/drivers.rs @@ -10,13 +10,11 @@ use std::sync::Arc; use async_trait::async_trait; use serde_json::{Value, json}; -use tinystoragedrivers_core::{ - CollectionSpec, DocumentStore, ErrorKind, Precondition, StorageError, -}; +use tinystoragedrivers_core::{CollectionSpec, DocumentStore, Precondition, StorageError}; use tokio::sync::OnceCell; use super::StateStore; -use crate::error::{Result, TinyFlowsError}; +use crate::error::{EngineError, Result}; /// A [`StateStore`] keeping each key in a driver [`DocumentStore`]. #[derive(Debug, Clone)] @@ -58,13 +56,12 @@ impl DriverStateStore { } } -/// A state key as a document id: keys longer than the driver's id limit are -/// refused rather than truncated or hashed, so two keys never share a slot. -fn map_error(error: StorageError) -> TinyFlowsError { - match error.kind() { - ErrorKind::InvalidInput => TinyFlowsError::Validation(format!("state store: {error}")), - _ => TinyFlowsError::Capability(format!("state store: {error}")), - } +/// A storage failure as the engine's capability error. A key the driver +/// cannot store as an id (empty, over its length limit, holding NUL) is +/// refused here rather than truncated or hashed, so two keys never share a +/// slot. +fn map_error(error: StorageError) -> EngineError { + EngineError::Capability(format!("state store: {error}")) } #[async_trait] diff --git a/crates/tinyflows/src/caps/drivers_tests.rs b/crates/tinyflows/src/caps/drivers_tests.rs new file mode 100644 index 00000000..5473940e --- /dev/null +++ b/crates/tinyflows/src/caps/drivers_tests.rs @@ -0,0 +1,48 @@ +use super::*; +use tinystoragedrivers_core::{MemoryStorage, Scope, StorageBackend}; + +fn docs(storage: &MemoryStorage, scope: &str) -> Arc { + Arc::clone( + storage + .for_scope(&Scope::new(scope).unwrap()) + .unwrap() + .documents(), + ) +} + +#[tokio::test] +async fn state_round_trips_and_overwrites() { + let store = DriverStateStore::new(docs(&MemoryStorage::new(), "local")); + assert_eq!(store.load("k").await.unwrap(), None); + store.store("k", json!({ "n": 1 })).await.unwrap(); + assert_eq!(store.load("k").await.unwrap(), Some(json!({ "n": 1 }))); + store.store("k", json!(null)).await.unwrap(); + assert_eq!( + store.load("k").await.unwrap(), + Some(Value::Null), + "a stored null is a value, not a missing key" + ); +} + +#[tokio::test] +async fn scopes_and_collections_keep_state_apart() { + let storage = MemoryStorage::new(); + let alice = DriverStateStore::new(docs(&storage, "alice")); + let bob = DriverStateStore::new(docs(&storage, "bob")); + let other = DriverStateStore::with_collection(docs(&storage, "alice"), "other_state"); + alice.store("run/1", json!("alice")).await.unwrap(); + assert_eq!(bob.load("run/1").await.unwrap(), None); + assert_eq!(other.load("run/1").await.unwrap(), None); +} + +#[tokio::test] +async fn an_unstorable_key_is_a_capability_error() { + let store = DriverStateStore::new(docs(&MemoryStorage::new(), "local")); + let error = store.store("", json!(1)).await.unwrap_err(); + assert!(matches!(error, EngineError::Capability(_)), "{error:?}"); + let error = DriverStateStore::with_collection(docs(&MemoryStorage::new(), "local"), "bad name") + .load("k") + .await + .unwrap_err(); + assert!(error.to_string().contains("state store"), "{error}"); +} diff --git a/crates/tinyflows/src/graph/checkpoint/drivers_tests.rs b/crates/tinyflows/src/graph/checkpoint/drivers_tests.rs new file mode 100644 index 00000000..e69de29b From 4b369d512e601ce7d35406636671724c2f446e9b Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 08:12:26 +0300 Subject: [PATCH 05/19] test(checkpoint): cover driver checkpoint round-trips Add tests exercising the checkpoint drivers to verify state is saved and restored correctly across round-trips. Auto-committed-on: dragonfly Co-authored-by: Medulla --- .../src/graph/checkpoint/drivers_tests.rs | 374 ++++++++++++++++++ 1 file changed, 374 insertions(+) diff --git a/crates/tinyflows/src/graph/checkpoint/drivers_tests.rs b/crates/tinyflows/src/graph/checkpoint/drivers_tests.rs index e69de29b..cb070598 100644 --- a/crates/tinyflows/src/graph/checkpoint/drivers_tests.rs +++ b/crates/tinyflows/src/graph/checkpoint/drivers_tests.rs @@ -0,0 +1,374 @@ +//! Behaviour of the storage-driver checkpointer: the same checks the +//! host-owned SQLite checkpointer (`tinyflows-sqlite`) is held to, run against +//! a driver document store, plus the isolation and key-encoding guarantees +//! the driver layout adds. + +use serde_json::json; +use super::*; +use crate::graph::ids::NodeId; +use tinystoragedrivers_core::{MemoryStorage, Scope, StorageBackend}; + +fn docs(storage: &MemoryStorage, scope: &str) -> Arc { + Arc::clone( + storage + .for_scope(&Scope::new(scope).unwrap()) + .unwrap() + .documents(), + ) +} + +fn store() -> DriverCheckpointer { + DriverCheckpointer::new(docs(&MemoryStorage::new(), "local")) +} + +/// One checkpoint in `thread`, at `step`, chained to `parent`. +fn checkpoint( + thread: &str, + id: &str, + parent: Option<&str>, + step: u64, + state: serde_json::Value, +) -> Checkpoint { + Checkpoint { + thread_id: thread.to_string(), + checkpoint_id: id.to_string(), + run_id: Some("run-1".to_string()), + parent_checkpoint_id: parent.map(str::to_string), + namespace: Vec::new(), + state, + next_nodes: vec![NodeId::new("next")], + completed_tasks: vec![NodeId::new("done")], + pending_writes: Vec::new(), + interrupts: Vec::new(), + pending_activations: None, + barrier_arrivals: Vec::new(), + metadata: json!({ "source": "loop", "step": step }), + } +} + +/// `get(None)` is what resume calls, and "latest" has to mean latest by +/// insertion, not by id — the ids are opaque and nothing orders them. +#[tokio::test] +async fn get_without_an_id_returns_the_most_recently_written_checkpoint() { + let store = store(); + store + .put(checkpoint("t1", "cp-1", None, 1, json!({ "n": 1 }))) + .await + .unwrap(); + store + .put(checkpoint("t1", "cp-2", Some("cp-1"), 2, json!({ "n": 2 }))) + .await + .unwrap(); + + let latest = store.get("t1", None).await.unwrap().expect("latest"); + assert_eq!(latest.checkpoint_id, "cp-2"); + assert_eq!(latest.state, json!({ "n": 2 })); + + let addressed = store.get("t1", Some("cp-1")).await.unwrap().expect("cp-1"); + assert_eq!(addressed.state, json!({ "n": 1 })); + + assert!(store.get("t1", Some("nope")).await.unwrap().is_none()); + assert!(store.get("other-thread", None).await.unwrap().is_none()); +} + +/// Threads must not see each other: a flow run addresses its own thread id, +/// and one flow's checkpoints leaking into another's resume would replay the +/// wrong graph. +#[tokio::test] +async fn threads_are_isolated_and_listed_in_insertion_order() { + let store = store(); + store + .put(checkpoint("t1", "a", None, 1, json!(1))) + .await + .unwrap(); + store + .put(checkpoint("t2", "b", None, 1, json!(2))) + .await + .unwrap(); + store + .put(checkpoint("t1", "c", Some("a"), 2, json!(3))) + .await + .unwrap(); + + let listed: Vec = store + .list("t1") + .await + .unwrap() + .into_iter() + .map(|m| m.checkpoint_id) + .collect(); + assert_eq!(listed, vec!["a".to_string(), "c".to_string()]); + + let mut threads = store.list_threads().await.unwrap(); + threads.sort(); + assert_eq!(threads, vec!["t1".to_string(), "t2".to_string()]); +} + +/// A namespaced read is how a parent run and the sub-workflows it embeds stay +/// out of each other's checkpoints — they share a thread id and differ only +/// here. +#[tokio::test] +async fn scoped_reads_see_only_their_own_namespace() { + let store = store(); + let mut root = checkpoint("t1", "root-1", None, 1, json!("root")); + root.namespace = Vec::new(); + let mut child = checkpoint("t1", "child-1", None, 1, json!("child")); + child.namespace = vec!["sub".to_string()]; + store.put(root).await.unwrap(); + store.put(child).await.unwrap(); + + let scoped = store + .get_scoped("t1", None, &["sub".to_string()]) + .await + .unwrap() + .expect("child checkpoint"); + assert_eq!(scoped.state, json!("child")); + + let at_root = store + .get_scoped("t1", None, &[]) + .await + .unwrap() + .expect("root checkpoint"); + assert_eq!(at_root.state, json!("root")); +} + +/// The partial-failure ledger, and its replace-vs-ignore rule — the thing in +/// this file most likely to be got subtly wrong, because both halves look like +/// "write it again". +/// +/// A **data** write (`idx >= 0`) is append-once: a superstep that is retried +/// after a partial failure re-emits what it already emitted, and taking the +/// second copy would fold the same emission twice on resume. A +/// **control-plane** write (`idx < 0`, e.g. a resume value) legitimately +/// changes on a retry and must upsert. Both are pushed into SQL as two +/// conflict clauses rather than a read-then-write, so they stay correct under +/// concurrent writers. +#[tokio::test] +async fn data_writes_are_append_once_and_control_plane_writes_upsert() { + let store = store(); + store + .put(checkpoint("t1", "cp-1", None, 1, json!({}))) + .await + .unwrap(); + let config = CheckpointConfig { + thread_id: "t1".to_string(), + checkpoint_id: Some("cp-1".to_string()), + namespace: Vec::new(), + }; + + // Data write, then the same (task_id, idx) again: the first value stands. + store + .put_writes( + &config, + &[PendingWrite::data("n1", "task-a", 0, "out", json!("first"))], + ) + .await + .unwrap(); + store + .put_writes( + &config, + &[PendingWrite::data( + "n1", + "task-a", + 0, + "out", + json!("second"), + )], + ) + .await + .unwrap(); + + let writes = store.get_writes(&config).await.unwrap(); + assert_eq!( + writes.len(), + 1, + "a re-run task must not duplicate its write" + ); + assert_eq!( + writes[0].payload, + json!("first"), + "a data write is append-once — a retry must not overwrite what already landed" + ); + + // Control-plane write at the reserved resume index: the newest value wins. + let resume = |payload| { + PendingWrite::data( + "n1", + "task-a", + tinyflows::graph::checkpoint::WRITES_IDX_RESUME, + "__resume__", + payload, + ) + }; + store + .put_writes(&config, &[resume(json!("old"))]) + .await + .unwrap(); + store + .put_writes(&config, &[resume(json!("new"))]) + .await + .unwrap(); + + let writes = store.get_writes(&config).await.unwrap(); + let control: Vec<_> = writes.iter().filter(|w| w.is_control_plane()).collect(); + assert_eq!( + control.len(), + 1, + "control-plane writes are keyed, not appended" + ); + assert_eq!( + control[0].payload, + json!("new"), + "a control-plane write must upsert — a resume value changes on a retry" + ); +} + +/// Deleting a thread is how a flow's history is dropped when the flow is +/// deleted; it must take that thread's writes with it and leave every other +/// thread alone. +#[tokio::test] +async fn deleting_a_thread_removes_its_checkpoints_and_leaves_others() { + let store = store(); + store + .put(checkpoint("doomed", "a", None, 1, json!(1))) + .await + .unwrap(); + store + .put(checkpoint("kept", "b", None, 1, json!(2))) + .await + .unwrap(); + + store.delete_thread("doomed").await.unwrap(); + + assert!(store.get("doomed", None).await.unwrap().is_none()); + assert!(store.list("doomed").await.unwrap().is_empty()); + assert!(store.get("kept", None).await.unwrap().is_some()); +} + +/// Pruning bounds a long-running flow's history. It keeps the newest N, which +/// is the end resume reads from. +#[tokio::test] +async fn prune_keeps_the_newest_checkpoints() { + let store = store(); + for i in 1..=5 { + store + .put(checkpoint("t1", &format!("cp-{i}"), None, i, json!(i))) + .await + .unwrap(); + } + + let removed = store.prune("t1", 2).await.unwrap(); + assert_eq!(removed, 3); + + let remaining: Vec = store + .list("t1") + .await + .unwrap() + .into_iter() + .map(|m| m.checkpoint_id) + .collect(); + assert_eq!(remaining, vec!["cp-4".to_string(), "cp-5".to_string()]); +} + +/// An in-memory database lives on its connection, so clones have to share one +/// — a clone that silently got its own empty database would be a resume that +/// finds nothing. + +#[tokio::test] +async fn scopes_and_prefixes_keep_threads_apart() { + let storage = MemoryStorage::new(); + let alice = DriverCheckpointer::::new(docs(&storage, "alice")); + let bob = DriverCheckpointer::::new(docs(&storage, "bob")); + let other = + DriverCheckpointer::::with_prefix(docs(&storage, "alice"), "other"); + alice + .put(checkpoint("t", "c1", None, 1, json!(1))) + .await + .unwrap(); + assert!(bob.get("t", None).await.unwrap().is_none()); + assert!(other.get("t", None).await.unwrap().is_none()); + assert_eq!(alice.list_threads().await.unwrap(), vec!["t".to_owned()]); + assert!(bob.list_threads().await.unwrap().is_empty()); +} + +#[tokio::test] +async fn scoped_reads_stay_in_their_namespace_when_ids_repeat() { + let saver = store(); + let child = vec!["sub".to_string()]; + saver + .put(checkpoint("t", "same", None, 1, json!(1))) + .await + .unwrap(); + let mut nested = checkpoint("t", "same", None, 2, json!(2)); + nested.namespace = child.clone(); + saver.put(nested).await.unwrap(); + let root = saver.get_scoped("t", Some("same"), &[]).await.unwrap().unwrap(); + assert_eq!(root.state, json!(1)); + let sub = saver.get_scoped("t", Some("same"), &child).await.unwrap().unwrap(); + assert_eq!(sub.state, json!(2)); + assert!( + saver + .get_scoped("t", None, &["other".to_string()]) + .await + .unwrap() + .is_none() + ); +} + +#[tokio::test] +async fn pending_writes_never_merge_across_lookalike_namespaces() { + let saver = store(); + saver + .put(checkpoint("t", "c", None, 1, json!(1))) + .await + .unwrap(); + let config = |namespace: Vec| CheckpointConfig { + thread_id: "t".to_string(), + checkpoint_id: Some("c".to_string()), + namespace, + }; + let split = config(vec!["a".to_string(), "b".to_string()]); + let joined = config(vec!["a\u{1f}b".to_string()]); + let write: PendingWrite = serde_json::from_value(json!({ + "task_id": "task", "idx": 0, "channel": "out", "value": "split" + })) + .unwrap_or_else(|_| PendingWrite::new("task", 0, "out", json!("split"))); + saver.put_writes(&split, &[write]).await.unwrap(); + assert!(saver.get_writes(&joined).await.unwrap().is_empty()); + assert_eq!(saver.get_writes(&split).await.unwrap().len(), 1); +} + +#[test] +fn keys_are_unambiguous_and_long_ones_hashed() { + assert_ne!(key(&["a/b", "c"]), key(&["a", "b/c"])); + let hashed = key(&[&"t".repeat(500)]); + assert!(hashed.starts_with("h:") && hashed.len() == 66, "{hashed}"); + assert_ne!(hashed, key(&[&"u".repeat(500)])); + let ns = |parts: &[&str]| parts.iter().map(|p| (*p).to_string()).collect::>(); + assert_ne!(namespace_key(&ns(&["a", "b"])), namespace_key(&ns(&["a/b"]))); + assert_ne!(namespace_key(&ns(&["ab"])), namespace_key(&ns(&["a", "b"]))); +} + +#[tokio::test] +async fn a_corrupt_record_is_a_checkpoint_error() { + let storage = MemoryStorage::new(); + let docs = docs(&storage, "local"); + let saver = DriverCheckpointer::::new(Arc::clone(&docs)); + saver + .put(checkpoint("t", "c", None, 1, json!(1))) + .await + .unwrap(); + let stored = docs + .query_all(&saver.checkpoints, &Query::all()) + .await + .unwrap() + .remove(0); + let mut doc = stored.doc; + doc["record"] = json!("not a checkpoint"); + docs.put(&saver.checkpoints, &stored.id, doc, Precondition::None) + .await + .unwrap(); + let error = saver.get("t", None).await.unwrap_err(); + assert!(matches!(error, GraphError::Checkpoint(_)), "{error:?}"); + assert!(map_error(StorageError::unavailable("busy")).to_string().contains("busy")); +} From e939d659204992f94bb25c2f67c793ae3fc7f779 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 08:12:58 +0300 Subject: [PATCH 06/19] refactor(checkpoint): tidy imports and test helpers Reordered imports and reformatted long expressions in the storage driver checkpointer and its tests. The pending-write test now builds its fixture through the PendingWrite constructor instead of a hand-rolled JSON fallback, which keeps the test aligned with the type's actual API. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinyflows/src/caps/mod.rs | 2 +- .../tinyflows/src/graph/checkpoint/drivers.rs | 12 +++++--- .../src/graph/checkpoint/drivers_tests.rs | 30 +++++++++++++------ 3 files changed, 30 insertions(+), 14 deletions(-) diff --git a/crates/tinyflows/src/caps/mod.rs b/crates/tinyflows/src/caps/mod.rs index 7da84644..6ae8f0b9 100644 --- a/crates/tinyflows/src/caps/mod.rs +++ b/crates/tinyflows/src/caps/mod.rs @@ -6,9 +6,9 @@ //! curated Composio tools, `HttpRequestTool`, and sandboxed code runtimes. pub mod agent; +pub mod approval; #[cfg(feature = "storage-drivers")] mod drivers; -pub mod approval; #[cfg(any(test, feature = "host-caps"))] pub mod host; #[cfg(any(test, feature = "mock"))] diff --git a/crates/tinyflows/src/graph/checkpoint/drivers.rs b/crates/tinyflows/src/graph/checkpoint/drivers.rs index ba954987..65a12748 100644 --- a/crates/tinyflows/src/graph/checkpoint/drivers.rs +++ b/crates/tinyflows/src/graph/checkpoint/drivers.rs @@ -25,12 +25,12 @@ use std::marker::PhantomData; use std::sync::Arc; +use crate::graph::error::{GraphError, Result}; +use crate::graph::ids::CheckpointId; use async_trait::async_trait; use serde::Serialize; use serde::de::DeserializeOwned; use serde_json::{Value, json}; -use crate::graph::error::{GraphError, Result}; -use crate::graph::ids::CheckpointId; use tinystoragedrivers_core::{ CollectionSpec, DocumentStore, DocumentStoreExt, ErrorKind, Filter, IndexSpec, Precondition, Query, Sort, StorageError, Versioned, @@ -38,7 +38,9 @@ use tinystoragedrivers_core::{ use tokio::sync::OnceCell; use super::{ - require_checkpoint_id,Checkpoint, CheckpointConfig, CheckpointMetadata, Checkpointer, PendingWrite, merge_writes}; + Checkpoint, CheckpointConfig, CheckpointMetadata, Checkpointer, PendingWrite, merge_writes, + require_checkpoint_id, +}; /// How many times a compare-and-swap loop retries before giving up. const CAS_ATTEMPTS: usize = 64; @@ -209,7 +211,9 @@ where fn decode(stored: Versioned) -> Result> { let record = stored.doc.get("record").cloned().unwrap_or(Value::Null); serde_json::from_value(record).map_err(|error| { - GraphError::Checkpoint(format!("storage driver checkpointer: decode record: {error}")) + GraphError::Checkpoint(format!( + "storage driver checkpointer: decode record: {error}" + )) }) } } diff --git a/crates/tinyflows/src/graph/checkpoint/drivers_tests.rs b/crates/tinyflows/src/graph/checkpoint/drivers_tests.rs index cb070598..a7245115 100644 --- a/crates/tinyflows/src/graph/checkpoint/drivers_tests.rs +++ b/crates/tinyflows/src/graph/checkpoint/drivers_tests.rs @@ -3,9 +3,9 @@ //! a driver document store, plus the isolation and key-encoding guarantees //! the driver layout adds. -use serde_json::json; use super::*; use crate::graph::ids::NodeId; +use serde_json::json; use tinystoragedrivers_core::{MemoryStorage, Scope, StorageBackend}; fn docs(storage: &MemoryStorage, scope: &str) -> Arc { @@ -302,9 +302,17 @@ async fn scoped_reads_stay_in_their_namespace_when_ids_repeat() { let mut nested = checkpoint("t", "same", None, 2, json!(2)); nested.namespace = child.clone(); saver.put(nested).await.unwrap(); - let root = saver.get_scoped("t", Some("same"), &[]).await.unwrap().unwrap(); + let root = saver + .get_scoped("t", Some("same"), &[]) + .await + .unwrap() + .unwrap(); assert_eq!(root.state, json!(1)); - let sub = saver.get_scoped("t", Some("same"), &child).await.unwrap().unwrap(); + let sub = saver + .get_scoped("t", Some("same"), &child) + .await + .unwrap() + .unwrap(); assert_eq!(sub.state, json!(2)); assert!( saver @@ -329,10 +337,7 @@ async fn pending_writes_never_merge_across_lookalike_namespaces() { }; let split = config(vec!["a".to_string(), "b".to_string()]); let joined = config(vec!["a\u{1f}b".to_string()]); - let write: PendingWrite = serde_json::from_value(json!({ - "task_id": "task", "idx": 0, "channel": "out", "value": "split" - })) - .unwrap_or_else(|_| PendingWrite::new("task", 0, "out", json!("split"))); + let write = PendingWrite::data("n", "task", 0, "out", json!("split")); saver.put_writes(&split, &[write]).await.unwrap(); assert!(saver.get_writes(&joined).await.unwrap().is_empty()); assert_eq!(saver.get_writes(&split).await.unwrap().len(), 1); @@ -345,7 +350,10 @@ fn keys_are_unambiguous_and_long_ones_hashed() { assert!(hashed.starts_with("h:") && hashed.len() == 66, "{hashed}"); assert_ne!(hashed, key(&[&"u".repeat(500)])); let ns = |parts: &[&str]| parts.iter().map(|p| (*p).to_string()).collect::>(); - assert_ne!(namespace_key(&ns(&["a", "b"])), namespace_key(&ns(&["a/b"]))); + assert_ne!( + namespace_key(&ns(&["a", "b"])), + namespace_key(&ns(&["a/b"])) + ); assert_ne!(namespace_key(&ns(&["ab"])), namespace_key(&ns(&["a", "b"]))); } @@ -370,5 +378,9 @@ async fn a_corrupt_record_is_a_checkpoint_error() { .unwrap(); let error = saver.get("t", None).await.unwrap_err(); assert!(matches!(error, GraphError::Checkpoint(_)), "{error:?}"); - assert!(map_error(StorageError::unavailable("busy")).to_string().contains("busy")); + assert!( + map_error(StorageError::unavailable("busy")) + .to_string() + .contains("busy") + ); } From 313a9d05ffbd5ff2921eced95ff2a31e7110a05f Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 08:13:14 +0300 Subject: [PATCH 07/19] test(checkpoint): use crate-relative path for resume index The driver test referenced the resume write index through the crate's public path, which fails to resolve when the test is compiled inside the crate. It now uses the crate-relative path so the test builds in both contexts. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinyflows/src/graph/checkpoint/drivers_tests.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/tinyflows/src/graph/checkpoint/drivers_tests.rs b/crates/tinyflows/src/graph/checkpoint/drivers_tests.rs index a7245115..c859d697 100644 --- a/crates/tinyflows/src/graph/checkpoint/drivers_tests.rs +++ b/crates/tinyflows/src/graph/checkpoint/drivers_tests.rs @@ -195,7 +195,7 @@ async fn data_writes_are_append_once_and_control_plane_writes_upsert() { PendingWrite::data( "n1", "task-a", - tinyflows::graph::checkpoint::WRITES_IDX_RESUME, + crate::graph::checkpoint::WRITES_IDX_RESUME, "__resume__", payload, ) From 3f3e341708712cac01406dc6fb4fe57ec38ebfd4 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 08:15:32 +0300 Subject: [PATCH 08/19] docs: document storage-driver backends feature Add CHANGELOG and README entries for the new `storage-drivers` cargo feature, covering `caps::DriverStateStore` and `graph::checkpoint::DriverCheckpointer` over a tinystoragedrivers document port. The feature is off by default and requires Rust 1.88. Auto-committed-on: dragonfly Co-authored-by: Medulla --- CHANGELOG.md | 11 +++++++++++ README.md | 8 ++++++++ 2 files changed, 19 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 00fd437b..818773df 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **Storage-driver backends (`storage-drivers` feature).** + `caps::DriverStateStore` implements `StateStore` and + `graph::checkpoint::DriverCheckpointer` implements the graph `Checkpointer` + over a `tinystoragedrivers` document port (git dependency at `v0.4.0`, so a + host that vendors the storage crates once can `[patch]` it). The + checkpointer stores one document per checkpoint with a per-thread + compare-and-swap sequence, merges pending writes under compare-and-swap, + pushes namespace-scoped reads down to one indexed query, length-prefixes + every key component and hashes over-long ids with SHA-256. Off by default; + the feature needs Rust 1.88. + - **Node postconditions (`tinyflows::postcondition`).** A node may declare `config.postcondition = { "require": ..., "field": ... }` — `non_empty`, `field_present` or `non_empty_list` — and the engine checks it against every diff --git a/README.md b/README.md index 9ae42b86..9d189779 100644 --- a/README.md +++ b/README.md @@ -90,6 +90,14 @@ Rust 2024 · MSRV 1.85 · `#![forbid(unsafe_code)]` · GPL-3.0-or-later. - Host-injected capability traits: `LlmProvider`, `ToolInvoker`, `HttpClient`, `CodeRunner`, `ShellRunner`, and `StateStore`. Deterministic in-memory mocks ship behind the `mock` cargo feature (`caps::mock::mock_capabilities()`). +- Storage-driver backends behind the `storage-drivers` cargo feature: + `caps::DriverStateStore` (a `StateStore`) and + `graph::checkpoint::DriverCheckpointer` (a graph `Checkpointer` with + checkpoints, pending writes and namespace-scoped reads), both over a + [tinystoragedrivers](https://github.com/tinyhumansai/tinystoragedrivers) + document port. The host opens SQLite, MongoDB or memory and hands over a + tenant-scoped handle; the default collections (`flows_state`, + `flows_graph_*`) stay apart from TinyAgents' own. Needs Rust 1.88. - Opaque `connection_ref` credential references — the host resolves them to real secrets; the crate never sees them. - Versioned wire format: graph `schema_version` and per-node `type_version`, with From d4799064ab6fb5fcc91885e9a915a72fe4d38ceb Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 08:25:41 +0300 Subject: [PATCH 09/19] feat(drivers): add checkpoint and state store driver capabilities Introduce checkpoint and state store driver traits so graphs can persist and restore execution state through pluggable backends. The capabilities are wired into the driver registry and covered by tests. Auto-committed-on: dragonfly Co-authored-by: Medulla --- .../drivers.rs => tinyflows-drivers/src/checkpoint.rs} | 0 .../src/checkpoint_tests.rs} | 0 .../caps/drivers.rs => tinyflows-drivers/src/state_store.rs} | 0 .../src/state_store_tests.rs} | 0 crates/tinyflows/src/graph/checkpoint/mod.rs | 4 ---- 5 files changed, 4 deletions(-) rename crates/{tinyflows/src/graph/checkpoint/drivers.rs => tinyflows-drivers/src/checkpoint.rs} (100%) rename crates/{tinyflows/src/graph/checkpoint/drivers_tests.rs => tinyflows-drivers/src/checkpoint_tests.rs} (100%) rename crates/{tinyflows/src/caps/drivers.rs => tinyflows-drivers/src/state_store.rs} (100%) rename crates/{tinyflows/src/caps/drivers_tests.rs => tinyflows-drivers/src/state_store_tests.rs} (100%) diff --git a/crates/tinyflows/src/graph/checkpoint/drivers.rs b/crates/tinyflows-drivers/src/checkpoint.rs similarity index 100% rename from crates/tinyflows/src/graph/checkpoint/drivers.rs rename to crates/tinyflows-drivers/src/checkpoint.rs diff --git a/crates/tinyflows/src/graph/checkpoint/drivers_tests.rs b/crates/tinyflows-drivers/src/checkpoint_tests.rs similarity index 100% rename from crates/tinyflows/src/graph/checkpoint/drivers_tests.rs rename to crates/tinyflows-drivers/src/checkpoint_tests.rs diff --git a/crates/tinyflows/src/caps/drivers.rs b/crates/tinyflows-drivers/src/state_store.rs similarity index 100% rename from crates/tinyflows/src/caps/drivers.rs rename to crates/tinyflows-drivers/src/state_store.rs diff --git a/crates/tinyflows/src/caps/drivers_tests.rs b/crates/tinyflows-drivers/src/state_store_tests.rs similarity index 100% rename from crates/tinyflows/src/caps/drivers_tests.rs rename to crates/tinyflows-drivers/src/state_store_tests.rs diff --git a/crates/tinyflows/src/graph/checkpoint/mod.rs b/crates/tinyflows/src/graph/checkpoint/mod.rs index 11884bdd..7d7d7401 100644 --- a/crates/tinyflows/src/graph/checkpoint/mod.rs +++ b/crates/tinyflows/src/graph/checkpoint/mod.rs @@ -13,13 +13,9 @@ //! at superstep boundaries only — never mid-node — so resuming always reruns a //! node from its start. -#[cfg(feature = "storage-drivers")] -mod drivers; mod file; mod types; -#[cfg(feature = "storage-drivers")] -pub use drivers::{DEFAULT_PREFIX as DRIVER_DEFAULT_PREFIX, DriverCheckpointer}; pub use file::FileCheckpointer; pub use types::{ BarrierArrivals, Checkpoint, CheckpointConfig, CheckpointMetadata, CheckpointSource, From 5fe9db41af0c9104560330ca42f3814c3d900545 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 08:25:51 +0300 Subject: [PATCH 10/19] refactor(caps): remove storage-drivers feature and driver state store The optional storage-drivers feature and its DriverStateStore re-export are gone, along with the tinystoragedrivers-core dependency they pulled in. The storage ports remain defined by the StateStore trait, so hosts can still supply their own backend without the extra crate. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinyflows/Cargo.toml | 10 ---------- crates/tinyflows/src/caps/mod.rs | 5 ----- 2 files changed, 15 deletions(-) diff --git a/crates/tinyflows/Cargo.toml b/crates/tinyflows/Cargo.toml index 4cacce52..bb4ce305 100644 --- a/crates/tinyflows/Cargo.toml +++ b/crates/tinyflows/Cargo.toml @@ -48,18 +48,8 @@ font8x8 = { version = "0.3", default-features = false, features = ["unicode"], o jaq-core = "3.1.0" jaq-std = { version = "3.0.1", default-features = false } jaq-json = { version = "2.0.1", features = ["serde"] } -# `storage-drivers` only: the storage ports (no database client). Named by git -# tag so a host that vendors tinystoragedrivers once (OpenHuman, through -# tinyagents) can `[patch]` this URL onto its single copy, as it does for -# tinytools. -tinystoragedrivers-core = { git = "https://github.com/tinyhumansai/tinystoragedrivers", tag = "v0.4.0", optional = true } [features] -# Off by default; `DriverStateStore` and `graph::checkpoint::DriverCheckpointer` -# over `tinystoragedrivers` document ports, so a host keeps run state and graph -# checkpoints in whichever backend it opened (SQLite, MongoDB, memory). Needs -# Rust 1.88, the storage crates' MSRV. -storage-drivers = ["dep:tinystoragedrivers-core", "dep:sha2"] # Off by default; enables in-memory mock capability impls for downstream tests # and examples. Mocks are always available inside this crate's own tests. default = [] diff --git a/crates/tinyflows/src/caps/mod.rs b/crates/tinyflows/src/caps/mod.rs index 6ae8f0b9..1f20f42e 100644 --- a/crates/tinyflows/src/caps/mod.rs +++ b/crates/tinyflows/src/caps/mod.rs @@ -7,8 +7,6 @@ pub mod agent; pub mod approval; -#[cfg(feature = "storage-drivers")] -mod drivers; #[cfg(any(test, feature = "host-caps"))] pub mod host; #[cfg(any(test, feature = "mock"))] @@ -207,9 +205,6 @@ pub trait StateStore: Send + Sync { async fn store(&self, key: &str, value: Value) -> Result<()>; } -#[cfg(feature = "storage-drivers")] -pub use drivers::DriverStateStore; - /// The bundle of capabilities handed to the engine for a run. /// /// Construct one per run from the host's concrete implementations. It carries From 90dd5b1382df0361f042ff62872189f246bc3d44 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 08:26:14 +0300 Subject: [PATCH 11/19] feat(checkpoint): add checkpoint and state store drivers Introduce a checkpoint driver and a state store abstraction so workflow execution state can be persisted and restored. The new modules ship with tests covering the checkpoint round-trip behaviour. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinyflows-drivers/Cargo.toml | 31 +++++++++++++++++++ crates/tinyflows-drivers/src/checkpoint.rs | 23 ++++++++++---- .../tinyflows-drivers/src/checkpoint_tests.rs | 4 +-- crates/tinyflows-drivers/src/lib.rs | 26 ++++++++++++++++ crates/tinyflows-drivers/src/state_store.rs | 6 ++-- 5 files changed, 79 insertions(+), 11 deletions(-) create mode 100644 crates/tinyflows-drivers/Cargo.toml create mode 100644 crates/tinyflows-drivers/src/lib.rs diff --git a/crates/tinyflows-drivers/Cargo.toml b/crates/tinyflows-drivers/Cargo.toml new file mode 100644 index 00000000..5881f14f --- /dev/null +++ b/crates/tinyflows-drivers/Cargo.toml @@ -0,0 +1,31 @@ +# `tinyflows` capabilities and checkpoints on `tinystoragedrivers` ports. +# +# A crate of its own rather than a feature of the engine: the storage crates +# are git dependencies (OpenHuman vendors them once and `[patch]`es the URL), +# and the published `tinyflows` package cannot carry a git dependency, even an +# optional one. It also keeps the engine's MSRV where it is; this crate needs +# the storage crates' 1.88. +[package] +name = "tinyflows-drivers" +version = "0.1.0" +description = "tinyflows StateStore and graph Checkpointer over tinystoragedrivers document ports" +edition.workspace = true +rust-version = "1.88" +license.workspace = true +repository.workspace = true +homepage.workspace = true +readme = "README.md" +publish = false + +[dependencies] +tinyflows = { workspace = true } +tinystoragedrivers-core = { git = "https://github.com/tinyhumansai/tinystoragedrivers", tag = "v0.4.0" } +async-trait = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +sha2 = { workspace = true } +tokio = { workspace = true } +tracing = { workspace = true } + +[dev-dependencies] +tokio = { workspace = true } diff --git a/crates/tinyflows-drivers/src/checkpoint.rs b/crates/tinyflows-drivers/src/checkpoint.rs index 65a12748..81b6361f 100644 --- a/crates/tinyflows-drivers/src/checkpoint.rs +++ b/crates/tinyflows-drivers/src/checkpoint.rs @@ -25,8 +25,8 @@ use std::marker::PhantomData; use std::sync::Arc; -use crate::graph::error::{GraphError, Result}; -use crate::graph::ids::CheckpointId; +use tinyflows::graph::error::{GraphError, Result}; +use tinyflows::graph::ids::CheckpointId; use async_trait::async_trait; use serde::Serialize; use serde::de::DeserializeOwned; @@ -37,9 +37,9 @@ use tinystoragedrivers_core::{ }; use tokio::sync::OnceCell; -use super::{ - Checkpoint, CheckpointConfig, CheckpointMetadata, Checkpointer, PendingWrite, merge_writes, - require_checkpoint_id, +use tinyflows::graph::{ + Checkpoint, CheckpointConfig, CheckpointMetadata, CheckpointTuple, Checkpointer, + PendingWrite, merge_writes, }; /// How many times a compare-and-swap loop retries before giving up. @@ -83,6 +83,17 @@ fn namespace_key(namespace: &[String]) -> String { /// The collection prefix [`DriverCheckpointer::new`] uses. pub const DEFAULT_PREFIX: &str = "flows_graph"; +/// The checkpoint id a write must name; a write against "the latest +/// checkpoint" has no meaning, so it is refused. +fn require_checkpoint_id(config: &CheckpointConfig) -> Result { + config.checkpoint_id.clone().ok_or_else(|| { + GraphError::Checkpoint(format!( + "put_writes requires an explicit checkpoint_id (thread `{}`)", + config.thread_id + )) + }) +} + /// A [`Checkpointer`] that stores everything in a driver [`DocumentStore`]. pub struct DriverCheckpointer { docs: Arc, @@ -412,5 +423,5 @@ where } #[cfg(test)] -#[path = "drivers_tests.rs"] +#[path = "checkpoint_tests.rs"] mod tests; diff --git a/crates/tinyflows-drivers/src/checkpoint_tests.rs b/crates/tinyflows-drivers/src/checkpoint_tests.rs index c859d697..cae13496 100644 --- a/crates/tinyflows-drivers/src/checkpoint_tests.rs +++ b/crates/tinyflows-drivers/src/checkpoint_tests.rs @@ -4,7 +4,7 @@ //! the driver layout adds. use super::*; -use crate::graph::ids::NodeId; +use tinyflows::graph::ids::NodeId; use serde_json::json; use tinystoragedrivers_core::{MemoryStorage, Scope, StorageBackend}; @@ -195,7 +195,7 @@ async fn data_writes_are_append_once_and_control_plane_writes_upsert() { PendingWrite::data( "n1", "task-a", - crate::graph::checkpoint::WRITES_IDX_RESUME, + tinyflows::graph::checkpoint::WRITES_IDX_RESUME, "__resume__", payload, ) diff --git a/crates/tinyflows-drivers/src/lib.rs b/crates/tinyflows-drivers/src/lib.rs new file mode 100644 index 00000000..d4db5e8c --- /dev/null +++ b/crates/tinyflows-drivers/src/lib.rs @@ -0,0 +1,26 @@ +//! `tinyflows` run state and graph checkpoints on +//! [tinystoragedrivers](https://github.com/tinyhumansai/tinystoragedrivers) +//! document ports. +//! +//! A host that has opened a storage backend (SQLite on a desktop, MongoDB in +//! the cloud, memory in tests) hands these a document handle — usually one +//! already bound to a tenant scope — and plugs them into the engine: +//! +//! - [`DriverStateStore`] is the engine's [`StateStore`] capability +//! (`Capabilities::state`), one document per key. +//! - [`DriverCheckpointer`] is a graph [`Checkpointer`]: checkpoints, pending +//! writes, namespace-scoped reads and a bulk `state_history`, for +//! `engine::run_with_checkpointer` / `resume_with_checkpointer`. +//! +//! The default collections (`flows_state`, `flows_graph_*`) stay apart from +//! TinyAgents' own checkpointer collections, so both can share one scoped +//! backend. +//! +//! [`StateStore`]: tinyflows::caps::StateStore +//! [`Checkpointer`]: tinyflows::graph::Checkpointer + +mod checkpoint; +mod state_store; + +pub use checkpoint::{DEFAULT_PREFIX, DriverCheckpointer}; +pub use state_store::DriverStateStore; diff --git a/crates/tinyflows-drivers/src/state_store.rs b/crates/tinyflows-drivers/src/state_store.rs index e984920f..23dfd6fa 100644 --- a/crates/tinyflows-drivers/src/state_store.rs +++ b/crates/tinyflows-drivers/src/state_store.rs @@ -13,8 +13,8 @@ use serde_json::{Value, json}; use tinystoragedrivers_core::{CollectionSpec, DocumentStore, Precondition, StorageError}; use tokio::sync::OnceCell; -use super::StateStore; -use crate::error::{EngineError, Result}; +use tinyflows::caps::StateStore; +use tinyflows::error::{EngineError, Result}; /// A [`StateStore`] keeping each key in a driver [`DocumentStore`]. #[derive(Debug, Clone)] @@ -92,5 +92,5 @@ impl StateStore for DriverStateStore { } #[cfg(test)] -#[path = "drivers_tests.rs"] +#[path = "state_store_tests.rs"] mod tests; From e349461ef292eda852f7809786cf18cef3e5a04b Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 08:26:38 +0300 Subject: [PATCH 12/19] feat(checkpoint): add indexed state history lookup The driver now overrides state_history to fetch a namespace's checkpoints and pending writes with one indexed query each, then walks the lineage in memory, replacing the trait default's two round trips per ancestor on remote backends. A by_namespace index and a stored ns_key field back the new query, and the write ledger is preferred over inline writes to match resolved_writes. Auto-committed-on: dragonfly Co-authored-by: Medulla --- Cargo.lock | 15 ++- crates/tinyflows-drivers/src/checkpoint.rs | 107 +++++++++++++++++- .../tinyflows-drivers/src/checkpoint_tests.rs | 2 +- 3 files changed, 117 insertions(+), 7 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c117610a..ca82726d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2650,7 +2650,6 @@ dependencies = [ "sha2", "tempfile", "thiserror", - "tinystoragedrivers-core", "tokio", "tracing", ] @@ -2690,6 +2689,20 @@ dependencies = [ "serde_json", ] +[[package]] +name = "tinyflows-drivers" +version = "0.1.0" +dependencies = [ + "async-trait", + "serde", + "serde_json", + "sha2", + "tinyflows", + "tinystoragedrivers-core", + "tokio", + "tracing", +] + [[package]] name = "tinyflows-schedule" version = "0.1.0" diff --git a/crates/tinyflows-drivers/src/checkpoint.rs b/crates/tinyflows-drivers/src/checkpoint.rs index 81b6361f..5f531d27 100644 --- a/crates/tinyflows-drivers/src/checkpoint.rs +++ b/crates/tinyflows-drivers/src/checkpoint.rs @@ -25,12 +25,12 @@ use std::marker::PhantomData; use std::sync::Arc; -use tinyflows::graph::error::{GraphError, Result}; -use tinyflows::graph::ids::CheckpointId; use async_trait::async_trait; use serde::Serialize; use serde::de::DeserializeOwned; use serde_json::{Value, json}; +use tinyflows::graph::error::{GraphError, Result}; +use tinyflows::graph::ids::CheckpointId; use tinystoragedrivers_core::{ CollectionSpec, DocumentStore, DocumentStoreExt, ErrorKind, Filter, IndexSpec, Precondition, Query, Sort, StorageError, Versioned, @@ -38,8 +38,8 @@ use tinystoragedrivers_core::{ use tokio::sync::OnceCell; use tinyflows::graph::{ - Checkpoint, CheckpointConfig, CheckpointMetadata, CheckpointTuple, Checkpointer, - PendingWrite, merge_writes, + Checkpoint, CheckpointConfig, CheckpointMetadata, CheckpointTuple, Checkpointer, PendingWrite, + merge_writes, }; /// How many times a compare-and-swap loop retries before giving up. @@ -157,7 +157,8 @@ impl DriverCheckpointer { )), CollectionSpec::new(&self.threads), CollectionSpec::new(&self.writes) - .index(IndexSpec::new("by_thread", ["thread"])), + .index(IndexSpec::new("by_thread", ["thread"])) + .index(IndexSpec::new("by_namespace", ["thread", "ns_key"])), ]; for spec in &specs { self.docs.ensure_collection(spec).await.map_err(map_error)?; @@ -296,6 +297,101 @@ where page.items.into_iter().next().map(Self::decode).transpose() } + /// One indexed read of the namespace's checkpoints and one of its + /// pending writes, then the lineage walk in memory — instead of the + /// trait default's `get_scoped` plus `get_writes` per hop, which on a + /// remote backend is two round trips per ancestor. + async fn state_history( + &self, + thread_id: &str, + namespace: &[String], + limit: Option, + ) -> Result>> { + self.declared().await?; + let ns_key = namespace_key(namespace); + let query = Query::filter( + Filter::eq("thread", thread_id).and(Filter::eq("namespace", ns_key.as_str())), + ) + .sort(Sort::asc("seq")); + let records = self + .docs + .query_all(&self.checkpoints, &query) + .await + .map_err(map_error)?; + if records.is_empty() { + return Ok(Vec::new()); + } + let mut writes: std::collections::HashMap> = + std::collections::HashMap::new(); + let stored_writes = self + .docs + .query_all( + &self.writes, + &Query::filter( + Filter::eq("thread", thread_id).and(Filter::eq("ns_key", ns_key.as_str())), + ), + ) + .await + .map_err(map_error)?; + for stored in stored_writes { + let Some(id) = stored.doc.get("checkpoint_id").and_then(Value::as_str) else { + continue; + }; + let batch: Vec = + serde_json::from_value(stored.doc.get("writes").cloned().unwrap_or(Value::Null))?; + writes.insert(id.to_string(), batch); + } + + // Last write wins for a re-used id, matching `get`. + let mut by_id: std::collections::HashMap> = + std::collections::HashMap::with_capacity(records.len()); + let mut cursor: Option = None; + for stored in records { + let checkpoint = Self::decode(stored)?; + cursor = Some(checkpoint.checkpoint_id.clone()); + by_id.insert(checkpoint.checkpoint_id.clone(), checkpoint); + } + + let mut out = Vec::new(); + while let Some(id) = cursor { + if limit.is_some_and(|limit| out.len() >= limit) { + break; + } + // `remove` doubles as the cycle guard: each id is visited once. + let Some(checkpoint) = by_id.remove(&id) else { + break; + }; + cursor = checkpoint.parent_checkpoint_id.clone(); + let config = CheckpointConfig { + thread_id: checkpoint.thread_id.clone(), + checkpoint_id: Some(checkpoint.checkpoint_id.clone()), + namespace: checkpoint.namespace.clone(), + }; + let parent_config = + checkpoint + .parent_checkpoint_id + .as_ref() + .map(|parent| CheckpointConfig { + thread_id: checkpoint.thread_id.clone(), + checkpoint_id: Some(parent.clone()), + namespace: checkpoint.namespace.clone(), + }); + // The persisted write ledger when there is one, the record's own + // inline writes otherwise — `resolved_writes`' rule. + let pending_writes = writes + .remove(&checkpoint.checkpoint_id) + .filter(|batch| !batch.is_empty()) + .unwrap_or_else(|| checkpoint.pending_writes.clone()); + out.push(CheckpointTuple { + config, + checkpoint, + parent_config, + pending_writes, + }); + } + Ok(out) + } + async fn list(&self, thread_id: &str) -> Result> { Ok(self .get_thread(thread_id) @@ -392,6 +488,7 @@ where let doc = json!({ "thread": config.thread_id, "namespace": config.namespace, + "ns_key": namespace_key(&config.namespace), "checkpoint_id": checkpoint_id, "writes": serde_json::to_value(&stored)?, }); diff --git a/crates/tinyflows-drivers/src/checkpoint_tests.rs b/crates/tinyflows-drivers/src/checkpoint_tests.rs index cae13496..725e99ae 100644 --- a/crates/tinyflows-drivers/src/checkpoint_tests.rs +++ b/crates/tinyflows-drivers/src/checkpoint_tests.rs @@ -4,8 +4,8 @@ //! the driver layout adds. use super::*; -use tinyflows::graph::ids::NodeId; use serde_json::json; +use tinyflows::graph::ids::NodeId; use tinystoragedrivers_core::{MemoryStorage, Scope, StorageBackend}; fn docs(storage: &MemoryStorage, scope: &str) -> Arc { From 4ac94a43f03801ce014ba8e881fcd9b4efc9b6f0 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 08:26:47 +0300 Subject: [PATCH 13/19] refactor(checkpoint): import merge_writes directly Move the merge_writes import out of the grouped tinyflows::graph use statement into its own line, matching the crate's import ordering. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinyflows-drivers/src/checkpoint.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/tinyflows-drivers/src/checkpoint.rs b/crates/tinyflows-drivers/src/checkpoint.rs index 5f531d27..7e5ca7dc 100644 --- a/crates/tinyflows-drivers/src/checkpoint.rs +++ b/crates/tinyflows-drivers/src/checkpoint.rs @@ -37,9 +37,9 @@ use tinystoragedrivers_core::{ }; use tokio::sync::OnceCell; +use tinyflows::graph::checkpoint::merge_writes; use tinyflows::graph::{ Checkpoint, CheckpointConfig, CheckpointMetadata, CheckpointTuple, Checkpointer, PendingWrite, - merge_writes, }; /// How many times a compare-and-swap loop retries before giving up. From 4a81db554eee3f50afee59defb3591f2fe6a101d Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 08:28:18 +0300 Subject: [PATCH 14/19] test(drivers): cover thread deletion, checkpoint pruning and state history Add tests for deleting a thread and its pending writes, deleting named checkpoints without touching siblings or other threads, and walking a lineage newest-first in a single read including cycle handling. The history test also checks the batched result matches the trait default's hop-by-hop walk. Auto-committed-on: dragonfly Co-authored-by: Medulla --- CHANGELOG.md | 20 +- README.md | 12 +- crates/tinyflows-drivers/README.md | 46 +++++ .../tinyflows-drivers/src/checkpoint_tests.rs | 194 ++++++++++++++++++ 4 files changed, 256 insertions(+), 16 deletions(-) create mode 100644 crates/tinyflows-drivers/README.md diff --git a/CHANGELOG.md b/CHANGELOG.md index 818773df..93d6f4ac 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,16 +9,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added -- **Storage-driver backends (`storage-drivers` feature).** - `caps::DriverStateStore` implements `StateStore` and - `graph::checkpoint::DriverCheckpointer` implements the graph `Checkpointer` - over a `tinystoragedrivers` document port (git dependency at `v0.4.0`, so a - host that vendors the storage crates once can `[patch]` it). The - checkpointer stores one document per checkpoint with a per-thread +- **`tinyflows-drivers` (new, unpublished crate).** `DriverStateStore` + implements the engine's `StateStore` and `DriverCheckpointer` the graph + `Checkpointer` over a `tinystoragedrivers` document port (git dependency at + `v0.4.0`, so a host that vendors the storage crates once can `[patch]` it). + The checkpointer stores one document per checkpoint with a per-thread compare-and-swap sequence, merges pending writes under compare-and-swap, - pushes namespace-scoped reads down to one indexed query, length-prefixes - every key component and hashes over-long ids with SHA-256. Off by default; - the feature needs Rust 1.88. + pushes namespace-scoped reads down to one indexed query, reads a whole + `state_history` in two queries, length-prefixes every key component and + hashes over-long ids with SHA-256. A separate crate rather than an engine + feature because the published `tinyflows` package cannot carry a git + dependency; the engine is unchanged and keeps its MSRV, this crate needs + Rust 1.88. - **Node postconditions (`tinyflows::postcondition`).** A node may declare `config.postcondition = { "require": ..., "field": ... }` — `non_empty`, diff --git a/README.md b/README.md index 9d189779..eb0d752b 100644 --- a/README.md +++ b/README.md @@ -90,14 +90,12 @@ Rust 2024 · MSRV 1.85 · `#![forbid(unsafe_code)]` · GPL-3.0-or-later. - Host-injected capability traits: `LlmProvider`, `ToolInvoker`, `HttpClient`, `CodeRunner`, `ShellRunner`, and `StateStore`. Deterministic in-memory mocks ship behind the `mock` cargo feature (`caps::mock::mock_capabilities()`). -- Storage-driver backends behind the `storage-drivers` cargo feature: - `caps::DriverStateStore` (a `StateStore`) and - `graph::checkpoint::DriverCheckpointer` (a graph `Checkpointer` with - checkpoints, pending writes and namespace-scoped reads), both over a +- Storage-driver backends in the companion `tinyflows-drivers` crate: + `DriverStateStore` (a `StateStore`) and `DriverCheckpointer` (a graph + `Checkpointer`) over a [tinystoragedrivers](https://github.com/tinyhumansai/tinystoragedrivers) - document port. The host opens SQLite, MongoDB or memory and hands over a - tenant-scoped handle; the default collections (`flows_state`, - `flows_graph_*`) stay apart from TinyAgents' own. Needs Rust 1.88. + document port, so a host keeps run state and checkpoints in the SQLite, + MongoDB or memory backend it already opened. - Opaque `connection_ref` credential references — the host resolves them to real secrets; the crate never sees them. - Versioned wire format: graph `schema_version` and per-node `type_version`, with diff --git a/crates/tinyflows-drivers/README.md b/crates/tinyflows-drivers/README.md new file mode 100644 index 00000000..74b07d4f --- /dev/null +++ b/crates/tinyflows-drivers/README.md @@ -0,0 +1,46 @@ +# tinyflows-drivers + +`tinyflows` run state and graph checkpoints on +[tinystoragedrivers](https://github.com/tinyhumansai/tinystoragedrivers) +document ports. A host that has opened a storage backend (SQLite on a desktop, +MongoDB in the cloud, memory in tests) gets the engine's durable pieces on it, +scoped per tenant by the document handle it passes in. + +| Type | Implements | Layout | +| --- | --- | --- | +| `DriverStateStore` | `tinyflows::caps::StateStore` | one document per key in `flows_state` | +| `DriverCheckpointer` | `tinyflows::graph::Checkpointer` | `flows_graph_checkpoints`, `_threads`, `_writes` | + +```rust +use std::sync::Arc; +use tinyflows_drivers::{DriverCheckpointer, DriverStateStore}; + +let docs = Arc::clone(scoped_storage.documents()); +let state = Arc::new(DriverStateStore::new(Arc::clone(&docs))); +let checkpointer = DriverCheckpointer::::new(docs); +``` + +## Checkpointer guarantees + +- One document per checkpoint with a per-thread `seq` advanced by + compare-and-swap, so listing is insertion order and a re-used checkpoint id + resolves to its latest write, as in the SQLite and file backends. +- `get_scoped` is one indexed `(thread, namespace, seq)` query, so a parent + run and its subgraphs never load each other's checkpoints, even when ids + repeat. +- `state_history` reads the namespace's checkpoints and pending writes in two + queries and walks the lineage in memory, rather than two round trips per + ancestor. +- Pending writes merge with `merge_writes` under compare-and-swap. +- Key components are length-prefixed and namespaces encoded injectively; ids + over 400 bytes become their SHA-256. + +## Why a separate crate + +`tinystoragedrivers` is a git dependency: OpenHuman vendors it once (through +TinyAgents) and points the URL at that copy with +`[patch."https://github.com/tinyhumansai/tinystoragedrivers"]`, so there is +exactly one `DocumentStore` trait in its graph. The published `tinyflows` +package cannot carry a git dependency, so these adapters live here +(`publish = false`). The engine stays dependency-free and keeps its MSRV; this +crate needs Rust 1.88, the storage crates' MSRV. diff --git a/crates/tinyflows-drivers/src/checkpoint_tests.rs b/crates/tinyflows-drivers/src/checkpoint_tests.rs index 725e99ae..14942131 100644 --- a/crates/tinyflows-drivers/src/checkpoint_tests.rs +++ b/crates/tinyflows-drivers/src/checkpoint_tests.rs @@ -384,3 +384,197 @@ async fn a_corrupt_record_is_a_checkpoint_error() { .contains("busy") ); } + +fn addressed(thread: &str, id: &str) -> CheckpointConfig { + CheckpointConfig { + thread_id: thread.to_string(), + checkpoint_id: Some(id.to_string()), + namespace: Vec::new(), + } +} + +#[tokio::test] +async fn deleting_a_thread_drops_its_pending_writes() { + let saver = store(); + saver + .put(checkpoint("doomed", "a", None, 1, json!(1))) + .await + .unwrap(); + let write = PendingWrite::data("n", "task", 0, "out", json!("x")); + saver + .put_writes(&addressed("doomed", "a"), &[write]) + .await + .unwrap(); + saver.delete_thread("doomed").await.unwrap(); + assert!( + saver + .get_writes(&addressed("doomed", "a")) + .await + .unwrap() + .is_empty() + ); + // A thread recreated under the same name starts clean. + saver + .put(checkpoint("doomed", "a", None, 1, json!(2))) + .await + .unwrap(); + assert!( + saver + .get_writes(&addressed("doomed", "a")) + .await + .unwrap() + .is_empty() + ); +} + +#[tokio::test] +async fn deleting_checkpoints_removes_only_the_named_ones_and_their_writes() { + let saver = store(); + assert_eq!(saver.delete_checkpoints("t", &[]).await.unwrap(), 0); + for (thread, id) in [("t", "a"), ("t", "b"), ("t", "c"), ("other", "a")] { + saver + .put(checkpoint(thread, id, None, 1, json!(id))) + .await + .unwrap(); + saver + .put_writes( + &addressed(thread, id), + &[PendingWrite::data("n", "task", 0, "out", json!(id))], + ) + .await + .unwrap(); + } + let removed = saver + .delete_checkpoints("t", &["a".to_string(), "c".to_string()]) + .await + .unwrap(); + assert_eq!(removed, 2); + let left: Vec = saver + .list("t") + .await + .unwrap() + .into_iter() + .map(|m| m.checkpoint_id) + .collect(); + assert_eq!(left, vec!["b".to_string()]); + assert!( + saver + .get_writes(&addressed("t", "a")) + .await + .unwrap() + .is_empty() + ); + assert_eq!( + saver.get_writes(&addressed("t", "b")).await.unwrap().len(), + 1 + ); + assert_eq!( + saver.list("other").await.unwrap().len(), + 1, + "other threads keep theirs" + ); + assert_eq!( + saver + .get_writes(&addressed("other", "a")) + .await + .unwrap() + .len(), + 1 + ); +} + +#[tokio::test] +async fn state_history_walks_the_lineage_newest_first_in_one_read() { + let saver = store(); + saver + .put(checkpoint("t", "a", None, 1, json!(1))) + .await + .unwrap(); + saver + .put(checkpoint("t", "b", Some("a"), 2, json!(2))) + .await + .unwrap(); + // An off-lineage sibling of `b` is not part of `c`'s history. + saver + .put(checkpoint("t", "x", Some("a"), 2, json!(9))) + .await + .unwrap(); + saver + .put(checkpoint("t", "c", Some("b"), 3, json!(3))) + .await + .unwrap(); + let ledger = PendingWrite::data("n", "task", 0, "out", json!("ledger")); + saver + .put_writes(&addressed("t", "b"), &[ledger.clone()]) + .await + .unwrap(); + // Another namespace never leaks into the root's history. + let mut nested = checkpoint("t", "z", Some("c"), 4, json!(0)); + nested.namespace = vec!["sub".to_string()]; + saver.put(nested).await.unwrap(); + + let history = saver.state_history("t", &[], None).await.unwrap(); + let ids: Vec<&str> = history + .iter() + .map(|tuple| tuple.checkpoint.checkpoint_id.as_str()) + .collect(); + assert_eq!(ids, vec!["c", "b", "a"]); + assert_eq!( + history[1].pending_writes, + vec![ledger], + "the write ledger wins" + ); + assert_eq!( + history[0] + .parent_config + .as_ref() + .and_then(|p| p.checkpoint_id.as_deref()), + Some("b") + ); + assert!(history[2].parent_config.is_none()); + + let capped = saver.state_history("t", &[], Some(2)).await.unwrap(); + assert_eq!(capped.len(), 2); + assert!( + saver + .state_history("none", &[], None) + .await + .unwrap() + .is_empty() + ); + + // Same answer as the trait default walks it hop by hop. + let mut default_walk = Vec::new(); + let mut cursor = None; + while let Some(tuple) = saver + .get_tuple(CheckpointConfig { + thread_id: "t".to_string(), + checkpoint_id: cursor.clone(), + namespace: Vec::new(), + }) + .await + .unwrap() + { + default_walk.push(tuple.checkpoint.checkpoint_id.clone()); + cursor = tuple.checkpoint.parent_checkpoint_id.clone(); + if cursor.is_none() { + break; + } + } + assert_eq!(default_walk, vec!["c", "b", "a"]); +} + +#[tokio::test] +async fn state_history_stops_at_a_lineage_cycle() { + let saver = store(); + saver + .put(checkpoint("t", "a", Some("b"), 1, json!(1))) + .await + .unwrap(); + saver + .put(checkpoint("t", "b", Some("a"), 2, json!(2))) + .await + .unwrap(); + let history = saver.state_history("t", &[], None).await.unwrap(); + assert_eq!(history.len(), 2, "each checkpoint is visited once"); +} From b4c89eb672c2e9541e2af2a3c72668629dbe640a Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 08:28:50 +0300 Subject: [PATCH 15/19] test(checkpoint): use from_ref for single-element write slice Replace the cloned single-element slice with std::slice::from_ref in the state history test, avoiding an unnecessary clone of the ledger write. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinyflows-drivers/src/checkpoint_tests.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/tinyflows-drivers/src/checkpoint_tests.rs b/crates/tinyflows-drivers/src/checkpoint_tests.rs index 14942131..fa1db534 100644 --- a/crates/tinyflows-drivers/src/checkpoint_tests.rs +++ b/crates/tinyflows-drivers/src/checkpoint_tests.rs @@ -505,7 +505,7 @@ async fn state_history_walks_the_lineage_newest_first_in_one_read() { .unwrap(); let ledger = PendingWrite::data("n", "task", 0, "out", json!("ledger")); saver - .put_writes(&addressed("t", "b"), &[ledger.clone()]) + .put_writes(&addressed("t", "b"), std::slice::from_ref(&ledger)) .await .unwrap(); // Another namespace never leaks into the root's history. From ac545241a9e9051b80f792f082e395fa02d30dd1 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 08:40:37 +0300 Subject: [PATCH 16/19] feat(checkpoint): add checkpoint history and key listing Adds a history view over stored checkpoints and a helper for enumerating checkpoint keys, so callers can inspect past states and discover existing checkpoints without knowing their identifiers in advance. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinyflows-drivers/src/checkpoint.rs | 51 +---- .../src/checkpoint_history_tests.rs | 199 ++++++++++++++++++ .../tinyflows-drivers/src/checkpoint_keys.rs | 52 +++++ .../tinyflows-drivers/src/checkpoint_tests.rs | 194 ----------------- crates/tinyflows-drivers/src/lib.rs | 1 + 5 files changed, 257 insertions(+), 240 deletions(-) create mode 100644 crates/tinyflows-drivers/src/checkpoint_history_tests.rs create mode 100644 crates/tinyflows-drivers/src/checkpoint_keys.rs diff --git a/crates/tinyflows-drivers/src/checkpoint.rs b/crates/tinyflows-drivers/src/checkpoint.rs index 7e5ca7dc..3c5ce98d 100644 --- a/crates/tinyflows-drivers/src/checkpoint.rs +++ b/crates/tinyflows-drivers/src/checkpoint.rs @@ -37,6 +37,8 @@ use tinystoragedrivers_core::{ }; use tokio::sync::OnceCell; +use crate::checkpoint_keys::{key, map_error, namespace_key, require_checkpoint_id}; + use tinyflows::graph::checkpoint::merge_writes; use tinyflows::graph::{ Checkpoint, CheckpointConfig, CheckpointMetadata, CheckpointTuple, Checkpointer, PendingWrite, @@ -45,55 +47,9 @@ use tinyflows::graph::{ /// How many times a compare-and-swap loop retries before giving up. const CAS_ATTEMPTS: usize = 64; -/// Map a storage driver failure onto the graph's checkpoint error. -fn map_error(error: StorageError) -> GraphError { - GraphError::Checkpoint(format!("storage driver: {error}")) -} - -/// Longest id stored as is; longer ones are hashed. -const MAX_KEY_LEN: usize = 400; - -/// A document id for `parts`: length-prefixed so no two tuples collide, and -/// replaced by its SHA-256 when it would exceed the driver's id limit. -fn key(parts: &[&str]) -> String { - let joined: String = parts - .iter() - .map(|part| format!("{}:{part}", part.len())) - .collect::>() - .join("/"); - if joined.len() <= MAX_KEY_LEN { - joined - } else { - use sha2::{Digest, Sha256}; - let digest = Sha256::digest(joined.as_bytes()); - let hex: String = digest.iter().map(|byte| format!("{byte:02x}")).collect(); - format!("h:{hex}") - } -} - -/// `namespace` as one string, injectively: each component is -/// length-prefixed, so `["a", "b"]` and `["a/b"]` never meet. -fn namespace_key(namespace: &[String]) -> String { - namespace - .iter() - .map(|part| format!("{}:{part};", part.len())) - .collect() -} - /// The collection prefix [`DriverCheckpointer::new`] uses. pub const DEFAULT_PREFIX: &str = "flows_graph"; -/// The checkpoint id a write must name; a write against "the latest -/// checkpoint" has no meaning, so it is refused. -fn require_checkpoint_id(config: &CheckpointConfig) -> Result { - config.checkpoint_id.clone().ok_or_else(|| { - GraphError::Checkpoint(format!( - "put_writes requires an explicit checkpoint_id (thread `{}`)", - config.thread_id - )) - }) -} - /// A [`Checkpointer`] that stores everything in a driver [`DocumentStore`]. pub struct DriverCheckpointer { docs: Arc, @@ -522,3 +478,6 @@ where #[cfg(test)] #[path = "checkpoint_tests.rs"] mod tests; +#[cfg(test)] +#[path = "checkpoint_history_tests.rs"] +mod history_tests; diff --git a/crates/tinyflows-drivers/src/checkpoint_history_tests.rs b/crates/tinyflows-drivers/src/checkpoint_history_tests.rs new file mode 100644 index 00000000..60c2214f --- /dev/null +++ b/crates/tinyflows-drivers/src/checkpoint_history_tests.rs @@ -0,0 +1,199 @@ +//! Deletion and history-walk behaviour of the storage-driver checkpointer. + +use super::tests::{checkpoint, store}; +use super::*; +use serde_json::json; + +fn addressed(thread: &str, id: &str) -> CheckpointConfig { + CheckpointConfig { + thread_id: thread.to_string(), + checkpoint_id: Some(id.to_string()), + namespace: Vec::new(), + } +} + +#[tokio::test] +async fn deleting_a_thread_drops_its_pending_writes() { + let saver = store(); + saver + .put(checkpoint("doomed", "a", None, 1, json!(1))) + .await + .unwrap(); + let write = PendingWrite::data("n", "task", 0, "out", json!("x")); + saver + .put_writes(&addressed("doomed", "a"), &[write]) + .await + .unwrap(); + saver.delete_thread("doomed").await.unwrap(); + assert!( + saver + .get_writes(&addressed("doomed", "a")) + .await + .unwrap() + .is_empty() + ); + // A thread recreated under the same name starts clean. + saver + .put(checkpoint("doomed", "a", None, 1, json!(2))) + .await + .unwrap(); + assert!( + saver + .get_writes(&addressed("doomed", "a")) + .await + .unwrap() + .is_empty() + ); +} + +#[tokio::test] +async fn deleting_checkpoints_removes_only_the_named_ones_and_their_writes() { + let saver = store(); + assert_eq!(saver.delete_checkpoints("t", &[]).await.unwrap(), 0); + for (thread, id) in [("t", "a"), ("t", "b"), ("t", "c"), ("other", "a")] { + saver + .put(checkpoint(thread, id, None, 1, json!(id))) + .await + .unwrap(); + saver + .put_writes( + &addressed(thread, id), + &[PendingWrite::data("n", "task", 0, "out", json!(id))], + ) + .await + .unwrap(); + } + let removed = saver + .delete_checkpoints("t", &["a".to_string(), "c".to_string()]) + .await + .unwrap(); + assert_eq!(removed, 2); + let left: Vec = saver + .list("t") + .await + .unwrap() + .into_iter() + .map(|m| m.checkpoint_id) + .collect(); + assert_eq!(left, vec!["b".to_string()]); + assert!( + saver + .get_writes(&addressed("t", "a")) + .await + .unwrap() + .is_empty() + ); + assert_eq!( + saver.get_writes(&addressed("t", "b")).await.unwrap().len(), + 1 + ); + assert_eq!( + saver.list("other").await.unwrap().len(), + 1, + "other threads keep theirs" + ); + assert_eq!( + saver + .get_writes(&addressed("other", "a")) + .await + .unwrap() + .len(), + 1 + ); +} + +#[tokio::test] +async fn state_history_walks_the_lineage_newest_first_in_one_read() { + let saver = store(); + saver + .put(checkpoint("t", "a", None, 1, json!(1))) + .await + .unwrap(); + saver + .put(checkpoint("t", "b", Some("a"), 2, json!(2))) + .await + .unwrap(); + // An off-lineage sibling of `b` is not part of `c`'s history. + saver + .put(checkpoint("t", "x", Some("a"), 2, json!(9))) + .await + .unwrap(); + saver + .put(checkpoint("t", "c", Some("b"), 3, json!(3))) + .await + .unwrap(); + let ledger = PendingWrite::data("n", "task", 0, "out", json!("ledger")); + saver + .put_writes(&addressed("t", "b"), std::slice::from_ref(&ledger)) + .await + .unwrap(); + // Another namespace never leaks into the root's history. + let mut nested = checkpoint("t", "z", Some("c"), 4, json!(0)); + nested.namespace = vec!["sub".to_string()]; + saver.put(nested).await.unwrap(); + + let history = saver.state_history("t", &[], None).await.unwrap(); + let ids: Vec<&str> = history + .iter() + .map(|tuple| tuple.checkpoint.checkpoint_id.as_str()) + .collect(); + assert_eq!(ids, vec!["c", "b", "a"]); + assert_eq!( + history[1].pending_writes, + vec![ledger], + "the write ledger wins" + ); + assert_eq!( + history[0] + .parent_config + .as_ref() + .and_then(|p| p.checkpoint_id.as_deref()), + Some("b") + ); + assert!(history[2].parent_config.is_none()); + + let capped = saver.state_history("t", &[], Some(2)).await.unwrap(); + assert_eq!(capped.len(), 2); + assert!( + saver + .state_history("none", &[], None) + .await + .unwrap() + .is_empty() + ); + + // Same answer as the trait default walks it hop by hop. + let mut default_walk = Vec::new(); + let mut cursor = None; + while let Some(tuple) = saver + .get_tuple(CheckpointConfig { + thread_id: "t".to_string(), + checkpoint_id: cursor.clone(), + namespace: Vec::new(), + }) + .await + .unwrap() + { + default_walk.push(tuple.checkpoint.checkpoint_id.clone()); + cursor = tuple.checkpoint.parent_checkpoint_id.clone(); + if cursor.is_none() { + break; + } + } + assert_eq!(default_walk, vec!["c", "b", "a"]); +} + +#[tokio::test] +async fn state_history_stops_at_a_lineage_cycle() { + let saver = store(); + saver + .put(checkpoint("t", "a", Some("b"), 1, json!(1))) + .await + .unwrap(); + saver + .put(checkpoint("t", "b", Some("a"), 2, json!(2))) + .await + .unwrap(); + let history = saver.state_history("t", &[], None).await.unwrap(); + assert_eq!(history.len(), 2, "each checkpoint is visited once"); +} diff --git a/crates/tinyflows-drivers/src/checkpoint_keys.rs b/crates/tinyflows-drivers/src/checkpoint_keys.rs new file mode 100644 index 00000000..cf9be118 --- /dev/null +++ b/crates/tinyflows-drivers/src/checkpoint_keys.rs @@ -0,0 +1,52 @@ +//! Document-id and error helpers shared by the checkpointer. + +use tinyflows::graph::CheckpointConfig; +use tinyflows::graph::error::{GraphError, Result}; +use tinystoragedrivers_core::StorageError; + +/// Map a storage driver failure onto the graph's checkpoint error. +pub(crate) fn map_error(error: StorageError) -> GraphError { + GraphError::Checkpoint(format!("storage driver: {error}")) +} + +/// Longest id stored as is; longer ones are hashed. +const MAX_KEY_LEN: usize = 400; + +/// A document id for `parts`: length-prefixed so no two tuples collide, and +/// replaced by its SHA-256 when it would exceed the driver's id limit. +pub(crate) fn key(parts: &[&str]) -> String { + let joined: String = parts + .iter() + .map(|part| format!("{}:{part}", part.len())) + .collect::>() + .join("/"); + if joined.len() <= MAX_KEY_LEN { + joined + } else { + use sha2::{Digest, Sha256}; + let digest = Sha256::digest(joined.as_bytes()); + let hex: String = digest.iter().map(|byte| format!("{byte:02x}")).collect(); + format!("h:{hex}") + } +} + +/// `namespace` as one string, injectively: each component is +/// length-prefixed, so `["a", "b"]` and `["a/b"]` never meet. +pub(crate) fn namespace_key(namespace: &[String]) -> String { + namespace + .iter() + .map(|part| format!("{}:{part};", part.len())) + .collect() +} + +/// The checkpoint id a write must name; a write against "the latest +/// checkpoint" has no meaning, so it is refused. +pub(crate) fn require_checkpoint_id(config: &CheckpointConfig) -> Result { + config.checkpoint_id.clone().ok_or_else(|| { + GraphError::Checkpoint(format!( + "put_writes requires an explicit checkpoint_id (thread `{}`)", + config.thread_id + )) + }) +} + diff --git a/crates/tinyflows-drivers/src/checkpoint_tests.rs b/crates/tinyflows-drivers/src/checkpoint_tests.rs index fa1db534..725e99ae 100644 --- a/crates/tinyflows-drivers/src/checkpoint_tests.rs +++ b/crates/tinyflows-drivers/src/checkpoint_tests.rs @@ -384,197 +384,3 @@ async fn a_corrupt_record_is_a_checkpoint_error() { .contains("busy") ); } - -fn addressed(thread: &str, id: &str) -> CheckpointConfig { - CheckpointConfig { - thread_id: thread.to_string(), - checkpoint_id: Some(id.to_string()), - namespace: Vec::new(), - } -} - -#[tokio::test] -async fn deleting_a_thread_drops_its_pending_writes() { - let saver = store(); - saver - .put(checkpoint("doomed", "a", None, 1, json!(1))) - .await - .unwrap(); - let write = PendingWrite::data("n", "task", 0, "out", json!("x")); - saver - .put_writes(&addressed("doomed", "a"), &[write]) - .await - .unwrap(); - saver.delete_thread("doomed").await.unwrap(); - assert!( - saver - .get_writes(&addressed("doomed", "a")) - .await - .unwrap() - .is_empty() - ); - // A thread recreated under the same name starts clean. - saver - .put(checkpoint("doomed", "a", None, 1, json!(2))) - .await - .unwrap(); - assert!( - saver - .get_writes(&addressed("doomed", "a")) - .await - .unwrap() - .is_empty() - ); -} - -#[tokio::test] -async fn deleting_checkpoints_removes_only_the_named_ones_and_their_writes() { - let saver = store(); - assert_eq!(saver.delete_checkpoints("t", &[]).await.unwrap(), 0); - for (thread, id) in [("t", "a"), ("t", "b"), ("t", "c"), ("other", "a")] { - saver - .put(checkpoint(thread, id, None, 1, json!(id))) - .await - .unwrap(); - saver - .put_writes( - &addressed(thread, id), - &[PendingWrite::data("n", "task", 0, "out", json!(id))], - ) - .await - .unwrap(); - } - let removed = saver - .delete_checkpoints("t", &["a".to_string(), "c".to_string()]) - .await - .unwrap(); - assert_eq!(removed, 2); - let left: Vec = saver - .list("t") - .await - .unwrap() - .into_iter() - .map(|m| m.checkpoint_id) - .collect(); - assert_eq!(left, vec!["b".to_string()]); - assert!( - saver - .get_writes(&addressed("t", "a")) - .await - .unwrap() - .is_empty() - ); - assert_eq!( - saver.get_writes(&addressed("t", "b")).await.unwrap().len(), - 1 - ); - assert_eq!( - saver.list("other").await.unwrap().len(), - 1, - "other threads keep theirs" - ); - assert_eq!( - saver - .get_writes(&addressed("other", "a")) - .await - .unwrap() - .len(), - 1 - ); -} - -#[tokio::test] -async fn state_history_walks_the_lineage_newest_first_in_one_read() { - let saver = store(); - saver - .put(checkpoint("t", "a", None, 1, json!(1))) - .await - .unwrap(); - saver - .put(checkpoint("t", "b", Some("a"), 2, json!(2))) - .await - .unwrap(); - // An off-lineage sibling of `b` is not part of `c`'s history. - saver - .put(checkpoint("t", "x", Some("a"), 2, json!(9))) - .await - .unwrap(); - saver - .put(checkpoint("t", "c", Some("b"), 3, json!(3))) - .await - .unwrap(); - let ledger = PendingWrite::data("n", "task", 0, "out", json!("ledger")); - saver - .put_writes(&addressed("t", "b"), std::slice::from_ref(&ledger)) - .await - .unwrap(); - // Another namespace never leaks into the root's history. - let mut nested = checkpoint("t", "z", Some("c"), 4, json!(0)); - nested.namespace = vec!["sub".to_string()]; - saver.put(nested).await.unwrap(); - - let history = saver.state_history("t", &[], None).await.unwrap(); - let ids: Vec<&str> = history - .iter() - .map(|tuple| tuple.checkpoint.checkpoint_id.as_str()) - .collect(); - assert_eq!(ids, vec!["c", "b", "a"]); - assert_eq!( - history[1].pending_writes, - vec![ledger], - "the write ledger wins" - ); - assert_eq!( - history[0] - .parent_config - .as_ref() - .and_then(|p| p.checkpoint_id.as_deref()), - Some("b") - ); - assert!(history[2].parent_config.is_none()); - - let capped = saver.state_history("t", &[], Some(2)).await.unwrap(); - assert_eq!(capped.len(), 2); - assert!( - saver - .state_history("none", &[], None) - .await - .unwrap() - .is_empty() - ); - - // Same answer as the trait default walks it hop by hop. - let mut default_walk = Vec::new(); - let mut cursor = None; - while let Some(tuple) = saver - .get_tuple(CheckpointConfig { - thread_id: "t".to_string(), - checkpoint_id: cursor.clone(), - namespace: Vec::new(), - }) - .await - .unwrap() - { - default_walk.push(tuple.checkpoint.checkpoint_id.clone()); - cursor = tuple.checkpoint.parent_checkpoint_id.clone(); - if cursor.is_none() { - break; - } - } - assert_eq!(default_walk, vec!["c", "b", "a"]); -} - -#[tokio::test] -async fn state_history_stops_at_a_lineage_cycle() { - let saver = store(); - saver - .put(checkpoint("t", "a", Some("b"), 1, json!(1))) - .await - .unwrap(); - saver - .put(checkpoint("t", "b", Some("a"), 2, json!(2))) - .await - .unwrap(); - let history = saver.state_history("t", &[], None).await.unwrap(); - assert_eq!(history.len(), 2, "each checkpoint is visited once"); -} diff --git a/crates/tinyflows-drivers/src/lib.rs b/crates/tinyflows-drivers/src/lib.rs index d4db5e8c..6c9bc9d0 100644 --- a/crates/tinyflows-drivers/src/lib.rs +++ b/crates/tinyflows-drivers/src/lib.rs @@ -20,6 +20,7 @@ //! [`Checkpointer`]: tinyflows::graph::Checkpointer mod checkpoint; +mod checkpoint_keys; mod state_store; pub use checkpoint::{DEFAULT_PREFIX, DriverCheckpointer}; From bf5d844def808f66952b9f9751d5fd33b35b2f1c Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 08:40:45 +0300 Subject: [PATCH 17/19] feat(state_store): add checkpoint persistence tests Add tests covering checkpoint save and restore behaviour in the state store, verifying that persisted state round-trips correctly. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinyflows-drivers/src/checkpoint_tests.rs | 12 ++++++------ crates/tinyflows-drivers/src/state_store.rs | 14 +++++++++++++- 2 files changed, 19 insertions(+), 7 deletions(-) diff --git a/crates/tinyflows-drivers/src/checkpoint_tests.rs b/crates/tinyflows-drivers/src/checkpoint_tests.rs index 725e99ae..fd21a05e 100644 --- a/crates/tinyflows-drivers/src/checkpoint_tests.rs +++ b/crates/tinyflows-drivers/src/checkpoint_tests.rs @@ -17,12 +17,12 @@ fn docs(storage: &MemoryStorage, scope: &str) -> Arc { ) } -fn store() -> DriverCheckpointer { +pub(super) fn store() -> DriverCheckpointer { DriverCheckpointer::new(docs(&MemoryStorage::new(), "local")) } /// One checkpoint in `thread`, at `step`, chained to `parent`. -fn checkpoint( +pub(super) fn checkpoint( thread: &str, id: &str, parent: Option<&str>, @@ -52,19 +52,19 @@ fn checkpoint( async fn get_without_an_id_returns_the_most_recently_written_checkpoint() { let store = store(); store - .put(checkpoint("t1", "cp-1", None, 1, json!({ "n": 1 }))) + .put(checkpoint("t1", "zz-first", None, 1, json!({ "n": 1 }))) .await .unwrap(); store - .put(checkpoint("t1", "cp-2", Some("cp-1"), 2, json!({ "n": 2 }))) + .put(checkpoint("t1", "aa-second", Some("zz-first"), 2, json!({ "n": 2 }))) .await .unwrap(); let latest = store.get("t1", None).await.unwrap().expect("latest"); - assert_eq!(latest.checkpoint_id, "cp-2"); + assert_eq!(latest.checkpoint_id, "aa-second"); assert_eq!(latest.state, json!({ "n": 2 })); - let addressed = store.get("t1", Some("cp-1")).await.unwrap().expect("cp-1"); + let addressed = store.get("t1", Some("zz-first")).await.unwrap().expect("zz-first"); assert_eq!(addressed.state, json!({ "n": 1 })); assert!(store.get("t1", Some("nope")).await.unwrap().is_none()); diff --git a/crates/tinyflows-drivers/src/state_store.rs b/crates/tinyflows-drivers/src/state_store.rs index 23dfd6fa..51576c11 100644 --- a/crates/tinyflows-drivers/src/state_store.rs +++ b/crates/tinyflows-drivers/src/state_store.rs @@ -73,7 +73,19 @@ impl StateStore for DriverStateStore { .get(&self.collection, key) .await .map_err(map_error)?; - Ok(found.and_then(|mut stored| stored.doc.get_mut("value").map(Value::take))) + match found { + None => Ok(None), + Some(mut stored) => stored + .doc + .get_mut("value") + .map(Value::take) + .map(Some) + .ok_or_else(|| { + EngineError::Capability(format!( + "state store: document `{key}` has no `value` field" + )) + }), + } } async fn store(&self, key: &str, value: Value) -> Result<()> { From 04fa4a950faa83bf1946fb0fdfef143b6cd746d6 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 08:53:13 +0300 Subject: [PATCH 18/19] test(checkpoint): cover documents missing a value field Add a state store test asserting that a document without a value field surfaces as a capability error rather than missing state, and tidy the checkpoint test module ordering and formatting. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinyflows-drivers/src/checkpoint.rs | 8 ++++---- .../tinyflows-drivers/src/checkpoint_keys.rs | 1 - .../tinyflows-drivers/src/checkpoint_tests.rs | 14 +++++++++++-- .../src/state_store_tests.rs | 20 +++++++++++++++++++ 4 files changed, 36 insertions(+), 7 deletions(-) diff --git a/crates/tinyflows-drivers/src/checkpoint.rs b/crates/tinyflows-drivers/src/checkpoint.rs index 3c5ce98d..0e558adf 100644 --- a/crates/tinyflows-drivers/src/checkpoint.rs +++ b/crates/tinyflows-drivers/src/checkpoint.rs @@ -33,7 +33,7 @@ use tinyflows::graph::error::{GraphError, Result}; use tinyflows::graph::ids::CheckpointId; use tinystoragedrivers_core::{ CollectionSpec, DocumentStore, DocumentStoreExt, ErrorKind, Filter, IndexSpec, Precondition, - Query, Sort, StorageError, Versioned, + Query, Sort, Versioned, }; use tokio::sync::OnceCell; @@ -475,9 +475,9 @@ where } } -#[cfg(test)] -#[path = "checkpoint_tests.rs"] -mod tests; #[cfg(test)] #[path = "checkpoint_history_tests.rs"] mod history_tests; +#[cfg(test)] +#[path = "checkpoint_tests.rs"] +mod tests; diff --git a/crates/tinyflows-drivers/src/checkpoint_keys.rs b/crates/tinyflows-drivers/src/checkpoint_keys.rs index cf9be118..8ea9f7a1 100644 --- a/crates/tinyflows-drivers/src/checkpoint_keys.rs +++ b/crates/tinyflows-drivers/src/checkpoint_keys.rs @@ -49,4 +49,3 @@ pub(crate) fn require_checkpoint_id(config: &CheckpointConfig) -> Result )) }) } - diff --git a/crates/tinyflows-drivers/src/checkpoint_tests.rs b/crates/tinyflows-drivers/src/checkpoint_tests.rs index fd21a05e..bd2fd627 100644 --- a/crates/tinyflows-drivers/src/checkpoint_tests.rs +++ b/crates/tinyflows-drivers/src/checkpoint_tests.rs @@ -56,7 +56,13 @@ async fn get_without_an_id_returns_the_most_recently_written_checkpoint() { .await .unwrap(); store - .put(checkpoint("t1", "aa-second", Some("zz-first"), 2, json!({ "n": 2 }))) + .put(checkpoint( + "t1", + "aa-second", + Some("zz-first"), + 2, + json!({ "n": 2 }), + )) .await .unwrap(); @@ -64,7 +70,11 @@ async fn get_without_an_id_returns_the_most_recently_written_checkpoint() { assert_eq!(latest.checkpoint_id, "aa-second"); assert_eq!(latest.state, json!({ "n": 2 })); - let addressed = store.get("t1", Some("zz-first")).await.unwrap().expect("zz-first"); + let addressed = store + .get("t1", Some("zz-first")) + .await + .unwrap() + .expect("zz-first"); assert_eq!(addressed.state, json!({ "n": 1 })); assert!(store.get("t1", Some("nope")).await.unwrap().is_none()); diff --git a/crates/tinyflows-drivers/src/state_store_tests.rs b/crates/tinyflows-drivers/src/state_store_tests.rs index 5473940e..a0762c99 100644 --- a/crates/tinyflows-drivers/src/state_store_tests.rs +++ b/crates/tinyflows-drivers/src/state_store_tests.rs @@ -46,3 +46,23 @@ async fn an_unstorable_key_is_a_capability_error() { .unwrap_err(); assert!(error.to_string().contains("state store"), "{error}"); } + +#[tokio::test] +async fn a_document_without_a_value_is_a_capability_error_not_missing_state() { + let backing = docs(&MemoryStorage::new(), "local"); + let store = DriverStateStore::new(Arc::clone(&backing)); + // Declares the collection, then plant a document some other writer shaped. + store.store("seed", json!(1)).await.unwrap(); + backing + .put( + &store.collection, + "bad", + json!({ "key": "bad" }), + Precondition::None, + ) + .await + .unwrap(); + let error = store.load("bad").await.unwrap_err(); + assert!(matches!(error, EngineError::Capability(_)), "{error:?}"); + assert!(error.to_string().contains("no `value` field"), "{error}"); +} From 7ce87400875fdcab45e39c48b30c5440fb6fa41c Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 9 Oct 2026 08:53:20 +0300 Subject: [PATCH 19/19] test(checkpoint): import StorageError in checkpoint tests Add the StorageError import to the checkpoint test module so the tests can reference the storage error type. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinyflows-drivers/src/checkpoint_tests.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/tinyflows-drivers/src/checkpoint_tests.rs b/crates/tinyflows-drivers/src/checkpoint_tests.rs index bd2fd627..f2cb40f0 100644 --- a/crates/tinyflows-drivers/src/checkpoint_tests.rs +++ b/crates/tinyflows-drivers/src/checkpoint_tests.rs @@ -6,7 +6,7 @@ use super::*; use serde_json::json; use tinyflows::graph::ids::NodeId; -use tinystoragedrivers_core::{MemoryStorage, Scope, StorageBackend}; +use tinystoragedrivers_core::{MemoryStorage, Scope, StorageBackend, StorageError}; fn docs(storage: &MemoryStorage, scope: &str) -> Arc { Arc::clone(