From cfe4e26796c1b43ece05d3583addfc4ef9f0ba92 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 30 Aug 2026 15:46:05 +0300 Subject: [PATCH 01/19] chore(deps): pin base64 to 0.22 to avoid a duplicate dependency Pinning the optional base64 dependency from 0.23 back to 0.22 to match the version already pulled in transitively by hyper-util and reqwest, preventing a second copy of the crate from entering consumers' dependency graphs and tripping their kernel-dependency-floor checks. Auto-committed-on: dragonfly Co-authored-by: Medulla --- Cargo.toml | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/Cargo.toml b/Cargo.toml index d8a492a0..1a65487a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -112,7 +112,19 @@ chrono-tz = { version = "0.10", optional = true } # `STANDARD_NO_PAD`, so `harness::multimodal::data_uri` picks by length instead # of trying both — trying both decoded a multi-MB image twice for every # unpadded payload. -base64 = { version = "0.23", optional = true } +# +# Pinned to `0.22`, one minor behind what dependabot proposed (#125), on +# purpose: `hyper-util` -> `hyper-rustls` -> `reqwest` already pulls `0.22.1` +# into every consumer's graph (openhuman, tinyagents, tinycortex, tinymemory), +# so a direct `0.23` requirement here resolves as a SECOND copy rather than +# reusing that one — +1 package with no capability gained, which is exactly +# what trips a consumer's kernel-dependency-floor ratchet (see +# `tinyhumansai/openhuman`'s `scripts/kernel-floor.limits`). Unify with the +# ecosystem's version instead of letting this crate's direct dependency get +# ahead of it. Dependabot will likely re-propose `0.23` once `reqwest`'s own +# tree moves past `0.22` — that will be fine to accept at that point, since it +# will no longer create a duplicate. +base64 = { version = "0.22", optional = true } # `flate2` decompresses `application/gzip` data URIs that carry an # `original_mime=` parameter — the shape a renderer emits when it compresses an # attachment before handing it over. Decompression is bounded by `Read::take` From a252ec0d600df4c8593f8c6783cd61a6e3202b4b Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 30 Aug 2026 15:46:45 +0300 Subject: [PATCH 02/19] chore: files changed Cargo.lock Auto-committed-on: dragonfly Co-authored-by: Medulla --- Cargo.lock | 12 +++--------- 1 file changed, 3 insertions(+), 9 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 5d4b1b91..20e7d686 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -61,12 +61,6 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" -[[package]] -name = "base64" -version = "0.23.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5" - [[package]] name = "bitflags" version = "2.13.0" @@ -530,7 +524,7 @@ version = "0.1.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" dependencies = [ - "base64 0.22.1", + "base64", "bytes", "futures-channel", "futures-util", @@ -987,7 +981,7 @@ version = "0.12.28" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147" dependencies = [ - "base64 0.22.1", + "base64", "bytes", "futures-core", "futures-util", @@ -1338,7 +1332,7 @@ name = "tinyagents" version = "2.1.1" dependencies = [ "async-trait", - "base64 0.23.1", + "base64", "bytes", "chrono", "chrono-tz", From 4d3fb926c8d548faf25c2e67ec208eacd0e442b9 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 30 Aug 2026 15:54:32 +0300 Subject: [PATCH 03/19] fix(select): separate send resource nouns from action verbs Move ambiguous resource nouns like "email", "message", and "dm" out of the Send verb aliases into a dedicated constant checked only when no explicit verb is detected. Previously these nouns caused commands such as "read email" or "delete a message" to incorrectly match Send alongside the explicit Read or Delete intent, since a noun is not the same signal as an action word. Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/harness/tool/select/mod.rs | 31 ++++++++++++++++++++++++++++--- 1 file changed, 28 insertions(+), 3 deletions(-) diff --git a/src/harness/tool/select/mod.rs b/src/harness/tool/select/mod.rs index f4ee57e6..f96c8065 100644 --- a/src/harness/tool/select/mod.rs +++ b/src/harness/tool/select/mod.rs @@ -120,9 +120,14 @@ fn verb_aliases(v: ToolVerb) -> &'static [&'static str] { ToolVerb::Create => &[ "create", "make", "new", "add", "start", "write", "post", "draft", ], - ToolVerb::Send => &[ - "send", "email", "message", "dm", "reply", "forward", "notify", - ], + // Deliberately action words only ("send", "reply", ...) — the + // ambiguous resource nouns ("email", "message", "dm") that used to + // live here moved to `SEND_NOUN_ALIASES`, checked separately in + // `detect_verbs` only when no explicit verb is otherwise present. + // Keeping them here made "read email" or "delete a message" match + // Send *alongside* the explicit Read/Delete intent, since a noun + // is not the same signal as an action word. + ToolVerb::Send => &["send", "reply", "forward", "notify"], ToolVerb::Read => &["read", "get", "fetch", "show", "view", "see", "retrieve"], ToolVerb::List => &["list", "search", "find", "lookup", "browse"], ToolVerb::Update => &[ @@ -133,6 +138,14 @@ fn verb_aliases(v: ToolVerb) -> &'static [&'static str] { } } +/// Resource nouns associated with `ToolVerb::Send` (as distinct from the +/// actual action words in `verb_aliases`). A resource noun alone is a much +/// weaker signal than an action word: "message support" has no explicit verb +/// and inferring Send from "message" is reasonable, but "delete a message" or +/// "read email" already carry an explicit conflicting verb (Delete, Read), +/// and a noun must not add Send alongside it — see `detect_verbs`. +const SEND_NOUN_ALIASES: &[&str] = &["email", "message", "dm"]; + const ALL_VERBS: [ToolVerb; 7] = [ ToolVerb::Create, ToolVerb::Send, @@ -173,6 +186,18 @@ fn detect_verbs(prompt: &str) -> HashSet { } } } + // Resource nouns for Send are checked only when no explicit action verb + // matched above — a noun alone (e.g. "message support") may still imply + // Send, but must not add it alongside an already-detected conflicting + // verb ("delete a message", "read email"). + if found.is_empty() { + for alias in SEND_NOUN_ALIASES { + if contains_whole_word(&lowered, alias) { + found.insert(ToolVerb::Send); + break; + } + } + } found } From e46f54657843d7248780bc472f9b7f0f8e80d34c Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 30 Aug 2026 15:55:06 +0300 Subject: [PATCH 04/19] chore: files changed src/harness/tool/select/test.rs Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/harness/tool/select/test.rs | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/src/harness/tool/select/test.rs b/src/harness/tool/select/test.rs index b6793856..000e89b4 100644 --- a/src/harness/tool/select/test.rs +++ b/src/harness/tool/select/test.rs @@ -154,6 +154,21 @@ fn verb_detection_handles_aliases() { assert!(v.contains(&ToolVerb::Merge)); } +#[test] +fn resource_noun_does_not_add_send_alongside_an_explicit_conflicting_verb() { + // "read email" and "delete a message" must not ALSO detect Send from the + // resource noun — only the explicit action verb should be present. + let v = detect_verbs("read email"); + assert_eq!(v, HashSet::from([ToolVerb::Read])); + + let v = detect_verbs("delete a message"); + assert_eq!(v, HashSet::from([ToolVerb::Delete])); + + // No explicit verb at all: the resource noun alone may still imply Send. + let v = detect_verbs("message support channel"); + assert!(v.contains(&ToolVerb::Send)); +} + #[test] fn tool_verb_handles_plurals() { assert_eq!(tool_verb("SLACK_DELETES_A_MESSAGE"), Some(ToolVerb::Delete)); From 5ce61c271d30da9379b49a1339bfd2e5188a4b7d Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 30 Aug 2026 15:57:41 +0300 Subject: [PATCH 05/19] chore: files changed src/graph/delegation/run.rs Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/graph/delegation/run.rs | 42 +++++++++++++++++++++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/src/graph/delegation/run.rs b/src/graph/delegation/run.rs index 0f38cb53..f30bc072 100644 --- a/src/graph/delegation/run.rs +++ b/src/graph/delegation/run.rs @@ -157,6 +157,48 @@ where F: Fn(DelegationStage, DelegationState) -> Fut + Clone + Send + Sync + 'static, Fut: Future> + Send + 'static, { + // This is the public entry point a host calls directly with an approver's + // decision — it does NOT go through `run_or_resume_delegation`, so it must + // provide its own two protections that function otherwise only gets: + // + // 1. Per-thread serialization: two concurrent approval callbacks for the + // SAME paused thread (a retried callback racing with a deny, say) must + // not both load the same interrupt checkpoint and independently + // finalize opposite decisions, appending competing histories. Uses the + // SAME `thread_lock` map as `run_or_resume_delegation`, so the two + // entry points serialize against each other too, not just themselves. + // 2. Schema-version validation: `CompiledGraph::resume` (inside + // `resume_graph`) applies the decision to whatever checkpoint it + // finds, with no schema check of its own — that check lives in + // `run_or_resume_delegation`'s match arms, which this path bypasses. + // During a rollback or mixed-version deployment, serde can decode a + // checkpoint written by a newer binary by ignoring its added fields, + // letting an older binary consume the approval and finalize the state + // under outdated semantics. Reject rather than guess. + let thread_id = config + .thread_id + .clone() + .ok_or_else(|| "delegation resume requires a thread_id".to_string())?; + let cp = config + .checkpointer + .clone() + .ok_or_else(|| "delegation resume requires a checkpointer".to_string())?; + + let lock = thread_lock(&thread_id); + let _guard = lock.lock().await; + + if let Some(checkpoint) = cp + .get(thread_id.as_str(), None) + .await + .map_err(|e| format!("delegation checkpoint read failed for thread {thread_id}: {e}"))? + && checkpoint.state.schema_version != CURRENT_SCHEMA_VERSION + { + return Err(format!( + "delegation resume refused for thread {thread_id}: checkpoint schema_version {} does not match this binary's schema_version {} — the approval decision was NOT applied", + checkpoint.state.schema_version, CURRENT_SCHEMA_VERSION + )); + } + let approved = decision_is_approve(&decision); tracing::info!( approved, From 2901e245ba79159ec70b5260e215336e77d58b87 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 30 Aug 2026 15:58:56 +0300 Subject: [PATCH 06/19] feat(delegation): add test for schema-mismatched checkpoint rejection in resume_delegation Add a test that verifies `resume_delegation` rejects a checkpoint whose schema version is newer than the current binary. This entry point bypasses `run_or_resume_delegation`'s match arms, so without this check a schema-mismatched checkpoint could be accepted during a mixed-version deployment, leading to incorrect behaviour under outdated semantics. Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/graph/delegation/test.rs | 56 ++++++++++++++++++++++++++++++++++++ 1 file changed, 56 insertions(+) diff --git a/src/graph/delegation/test.rs b/src/graph/delegation/test.rs index 4451424c..1b3d687c 100644 --- a/src/graph/delegation/test.rs +++ b/src/graph/delegation/test.rs @@ -824,6 +824,62 @@ async fn cancelled_checkpoint_still_scheduling_finalize_is_resumed_not_terminal( assert!(outcome.state.cancelled); } +#[tokio::test] +async fn resume_delegation_rejects_a_schema_mismatched_checkpoint() { + // `resume_delegation` is a public entry point a host calls directly with + // an approver's decision — it bypasses `run_or_resume_delegation`'s + // match arms entirely, so `CompiledGraph::resume` would otherwise apply + // the decision to a schema-mismatched checkpoint with no check at all. + // During a rollback/mixed-version deployment, serde can decode a newer + // additive checkpoint by ignoring unknown fields, letting an older + // binary consume the approval under outdated semantics. + let dir = tempfile::tempdir().unwrap(); + let seed: crate::graph::checkpoint::FileCheckpointer = + crate::graph::checkpoint::FileCheckpointer::new(dir.path()); + let checkpoint = Checkpoint { + thread_id: "resume-future-schema".to_string(), + checkpoint_id: "cp-future".to_string(), + run_id: None, + parent_checkpoint_id: None, + namespace: vec![], + state: DelegationState { + plan: Some("PLAN".to_string()), + schema_version: CURRENT_SCHEMA_VERSION + 1, + ..Default::default() + }, + next_nodes: vec![crate::harness::ids::NodeId::from("approval")], + completed_tasks: vec![], + pending_writes: vec![], + interrupts: vec![Interrupt { + id: "int-1".to_string(), + node: crate::harness::ids::NodeId::from("approval"), + payload: json!({}), + }], + pending_activations: None, + barrier_arrivals: vec![], + metadata: json!({}), + }; + seed.put(checkpoint) + .await + .expect("seed future-schema checkpoint parked on approval"); + + let cp: Arc> = + Arc::new(crate::graph::checkpoint::FileCheckpointer::::new(dir.path())); + let config = DelegationConfig { + require_review_approval: true, + checkpointer: Some(cp), + thread_id: Some("resume-future-schema".to_string()), + ..DelegationConfig::default() + }; + let err = resume_delegation(config, json!("approve_once"), flow_runner(0)) + .await + .expect_err("must refuse to resume a schema-mismatched checkpoint"); + assert!( + err.contains("schema_version"), + "error should name the mismatch: {err}" + ); +} + #[test] fn incompatible_checkpoint_error_matches_schema_not_corrupt_or_operational() { use crate::TinyAgentsError; From 778f90c42de10c4fc3ef1d40f9590791d86b62ed Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 30 Aug 2026 16:00:05 +0300 Subject: [PATCH 07/19] test(delegation): add concurrency test for resume_delegation serialization Adds a test that verifies two concurrent calls to `resume_delegation` for the same paused thread never execute their stage callbacks simultaneously, ensuring the per-thread lock correctly serializes access even when the public entry point bypasses `run_or_resume_delegation`. Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/graph/delegation/test.rs | 64 ++++++++++++++++++++++++++++++++++++ 1 file changed, 64 insertions(+) diff --git a/src/graph/delegation/test.rs b/src/graph/delegation/test.rs index 1b3d687c..eb34dbdb 100644 --- a/src/graph/delegation/test.rs +++ b/src/graph/delegation/test.rs @@ -824,6 +824,70 @@ async fn cancelled_checkpoint_still_scheduling_finalize_is_resumed_not_terminal( assert!(outcome.state.cancelled); } +#[tokio::test] +async fn concurrent_resume_delegation_calls_for_the_same_thread_never_overlap() { + // `resume_delegation` is a public entry point that bypasses + // `run_or_resume_delegation` entirely, so it needs its OWN per-thread + // serialization — two concurrent approval callbacks for the same paused + // thread (a retried callback racing with a deny, say) must not both load + // the same interrupt checkpoint and independently finalize, appending + // competing histories. + let concurrent = Arc::new(AtomicUsize::new(0)); + let max_concurrent = Arc::new(AtomicUsize::new(0)); + let make_runner = || { + let concurrent = concurrent.clone(); + let max_concurrent = max_concurrent.clone(); + move |stage: DelegationStage, _s: DelegationState| { + let concurrent = concurrent.clone(); + let max_concurrent = max_concurrent.clone(); + Box::pin(async move { + let now = concurrent.fetch_add(1, Ordering::SeqCst) + 1; + max_concurrent.fetch_max(now, Ordering::SeqCst); + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + concurrent.fetch_sub(1, Ordering::SeqCst); + let text = match stage { + DelegationStage::Review => "approve".to_string(), + _ => "ok".to_string(), + }; + Ok::<_, String>(DelegationStageOutput::done(text)) + }) as std::pin::Pin + Send>> + } + }; + + let dir = tempfile::tempdir().unwrap(); + let cp: Arc> = + Arc::new(crate::graph::checkpoint::FileCheckpointer::new(dir.path())); + let make_config = || DelegationConfig { + require_review_approval: true, + checkpointer: Some(cp.clone()), + thread_id: Some("racing-resume-thread".to_string()), + ..DelegationConfig::default() + }; + + // Park on the approval interrupt first. + run_delegation_durable(make_config(), make_runner()) + .await + .expect("parks on approval") + .pending + .expect("parked on the approval interrupt"); + + let (r1, r2) = tokio::join!( + resume_delegation(make_config(), json!("approve_once"), make_runner()), + resume_delegation(make_config(), json!("approve_once"), make_runner()), + ); + // Exactly one of the two racing resumes may succeed in applying the + // decision; a second resume of an already-finalized thread is expected to + // either no-op or surface a graph-level error — either is acceptable, but + // no overlapping stage execution across the two calls is not. + assert!(r1.is_ok() || r2.is_ok(), "at least one resume must succeed"); + assert_eq!( + max_concurrent.load(Ordering::SeqCst), + 1, + "the per-thread lock must serialize the two resumes' stage execution; \ + observed overlapping stage invocations" + ); +} + #[tokio::test] async fn resume_delegation_rejects_a_schema_mismatched_checkpoint() { // `resume_delegation` is a public entry point a host calls directly with From cc43d8d0ebfee8df0dd6ef0d8aef80a54db65ead Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 30 Aug 2026 16:01:14 +0300 Subject: [PATCH 08/19] feat(delegation): add cancellation check before approval node A cancellation check was added at the start of the approval node in the delegation graph, ensuring that a cancellation signal received while a gated run is waiting for approval is properly honoured. Without this check, a cancelled run could either retry indefinitely or finalize successfully despite being cancelled, depending on the approval outcome. Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/graph/delegation/graph.rs | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/src/graph/delegation/graph.rs b/src/graph/delegation/graph.rs index 4c5d3841..4d8d3c41 100644 --- a/src/graph/delegation/graph.rs +++ b/src/graph/delegation/graph.rs @@ -181,8 +181,25 @@ where // pause via `ApprovalRequestCard`), which parks a live in-memory chat turn // and is deliberately left untouched. if require_review_approval { + let cancel_approval = cancel.clone(); builder = builder.add_node("approval", move |s: DelegationState, ctx: NodeContext| { + let cancel = cancel_approval.clone(); async move { + // Cancellation must be honoured here exactly like every other + // node: without this check, a cancellation that arrives while + // a gated run is waiting for approval (or during the + // preceding review worker) is invisible to this boundary. A + // decisionless retry would then interrupt again indefinitely + // (nothing ever routes it to `finalize`), while an approving + // resume would finalize successfully despite the run having + // been cancelled. + if cancel.is_cancelled() { + return Ok(NodeResult::Command( + Command::default() + .with_update(DelegationUpdate::Cancelled) + .with_goto(["finalize"]), + )); + } match ctx.resume { None => { let payload = json!({ From 85f16eee5c52c55256d94dcb5f3cf863a27225e4 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 30 Aug 2026 16:02:12 +0300 Subject: [PATCH 09/19] test(delegation): add test for cancellation at approval node routing to finalize Add a test that verifies cancellation arriving while a gated run is parked at the approval interrupt is honoured at that boundary, ensuring the cancellation routes to finalize rather than being silently overridden by an approving resume. Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/graph/delegation/test.rs | 52 ++++++++++++++++++++++++++++++++++++ 1 file changed, 52 insertions(+) diff --git a/src/graph/delegation/test.rs b/src/graph/delegation/test.rs index eb34dbdb..cc720e21 100644 --- a/src/graph/delegation/test.rs +++ b/src/graph/delegation/test.rs @@ -824,6 +824,58 @@ async fn cancelled_checkpoint_still_scheduling_finalize_is_resumed_not_terminal( assert!(outcome.state.cancelled); } +#[tokio::test] +async fn cancellation_at_the_approval_node_routes_to_finalize() { + // Cancellation arriving while a gated run is parked at the approval + // interrupt (or during the preceding review worker) must be honoured at + // that boundary exactly like every other node. Without the check, a + // decisionless retry interrupts again indefinitely, and an approving + // resume finalizes successfully despite the cancellation. + let cancel = CancellationToken::new(); + let dir = tempfile::tempdir().unwrap(); + let cp: Arc> = + Arc::new(crate::graph::checkpoint::FileCheckpointer::new(dir.path())); + let config = DelegationConfig { + require_review_approval: true, + checkpointer: Some(cp.clone()), + thread_id: Some("cancel-at-approval".to_string()), + cancel: cancel.clone(), + ..DelegationConfig::default() + }; + let outcome = run_delegation_durable(config, flow_runner(0)) + .await + .expect("parks on approval"); + outcome.pending.expect("parked on the approval interrupt"); + assert!( + outcome.state.final_output.is_none(), + "not finalized yet — still awaiting approval" + ); + + // Cancel while parked, then resume with an approval decision: the + // cancellation must win, not the approval. + cancel.cancel(); + let resumed_config = DelegationConfig { + require_review_approval: true, + checkpointer: Some(cp), + thread_id: Some("cancel-at-approval".to_string()), + cancel, + ..DelegationConfig::default() + }; + let resumed = resume_delegation(resumed_config, json!("approve_once"), flow_runner(0)) + .await + .expect("resumes"); + assert!(resumed.pending.is_none(), "resume clears the pause"); + assert!( + resumed.state.cancelled, + "cancellation must be honoured at the approval boundary, not silently \ + overridden by an approving resume" + ); + assert!( + resumed.state.final_output.is_some(), + "cancellation routes to finalize, producing a cancellation summary" + ); +} + #[tokio::test] async fn concurrent_resume_delegation_calls_for_the_same_thread_never_overlap() { // `resume_delegation` is a public entry point that bypasses From 15f93ed5f9a4a996cf11f0666c493530ae91b320 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 30 Aug 2026 16:03:33 +0300 Subject: [PATCH 10/19] chore: files changed src/graph/checkpoint/mod.rs Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/graph/checkpoint/mod.rs | 39 ++++++++++++++++++++++++++++++++----- 1 file changed, 34 insertions(+), 5 deletions(-) diff --git a/src/graph/checkpoint/mod.rs b/src/graph/checkpoint/mod.rs index 8d76728c..5f2285d2 100644 --- a/src/graph/checkpoint/mod.rs +++ b/src/graph/checkpoint/mod.rs @@ -56,16 +56,45 @@ use crate::{Result, TinyAgentsError}; /// prefix; `what` names the field being decoded (e.g. `"record"`, /// `"next_nodes"`) with no leading `"decode "` — this function supplies that /// wording once, alongside the tag. +/// +/// **Only `what == "record"` — the whole-`Checkpoint` decode, where +/// `State` is the one thing this module knows can legitimately evolve across +/// a caller's schema versions — is eligible for `[schema]` at all.** +/// `"namespace"`, `"next_nodes"`, `"header"` and `"write payload"` are +/// checkpoint-internal engine metadata with no versioned shape of their own; +/// a `Category::Data` failure decoding any of those is corruption by +/// construction; classifying it as `[schema]` and pruning would have been +/// simply wrong. +/// +/// Even for `"record"`, `Category::Data` is a HEURISTIC, not a proof: this +/// module is generic over `State` (see the module docs) and has no way to +/// positively identify a specific caller's known-legacy shape from a byte +/// stream alone — a decode failure produced by data corruption that happens +/// to still be syntactically valid JSON (e.g. a `state.schema_version` field +/// whose value was flipped from a number to a string) also classifies as +/// `Category::Data`, indistinguishable at this layer from a genuine older +/// schema. This is an accepted, bounded residual risk, not a claim of +/// perfect precision: the worst case is an unnecessary fresh restart from +/// `plan` for what was actually corruption, which is a strict improvement +/// over the state before this classifier existed (100% of decode failures, +/// corrupt or not, were pruned identically). Closing the remaining gap +/// precisely would need the generic `Checkpointer` trait to expose the raw +/// bytes on a failed decode so a caller could apply its OWN positive-shape +/// check — a real API addition, tracked as follow-up rather than done here. pub(super) fn decode_json_err( backend: &str, what: &str, err: serde_json::Error, ) -> TinyAgentsError { - let tag = match err.classify() { - serde_json::error::Category::Data => "schema", - serde_json::error::Category::Syntax - | serde_json::error::Category::Eof - | serde_json::error::Category::Io => "corrupt", + let tag = if what == "record" { + match err.classify() { + serde_json::error::Category::Data => "schema", + serde_json::error::Category::Syntax + | serde_json::error::Category::Eof + | serde_json::error::Category::Io => "corrupt", + } + } else { + "corrupt" }; TinyAgentsError::Checkpoint(format!("{backend}: decode [{tag}] {what}: {err}")) } From aa80d5220162074df35d515c96d6b5c021deb27a Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 30 Aug 2026 16:03:48 +0300 Subject: [PATCH 11/19] test(delegation): add test ensuring only record context gets schema tag Add a test that verifies `decode_json_err` tags errors from non-record contexts as `[corrupt]` even when the underlying serde error is a `Data` category, reserving the `[schema]` tag exclusively for the "record" context. This prevents misclassification of internal metadata decoding failures as schema evolution issues. Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/graph/delegation/test.rs | 32 ++++++++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/src/graph/delegation/test.rs b/src/graph/delegation/test.rs index cc720e21..d02ebcba 100644 --- a/src/graph/delegation/test.rs +++ b/src/graph/delegation/test.rs @@ -1039,6 +1039,38 @@ fn incompatible_checkpoint_error_matches_schema_not_corrupt_or_operational() { ))); } +#[test] +fn decode_json_err_only_tags_schema_for_the_record_context() { + use crate::graph::checkpoint::decode_json_err; + + // `Category::Data` errors decoding checkpoint-internal metadata + // ("namespace", "next_nodes", "header", "write payload") must NEVER be + // tagged `[schema]` — those fields have no versioned shape of their own, + // so a Data-category failure there is corruption by construction, not a + // caller's schema evolution. + let data_err = serde_json::from_str::("\"not a number\"").unwrap_err(); + assert_eq!(data_err.classify(), serde_json::error::Category::Data); + for what in ["namespace", "next_nodes", "header", "write payload"] { + let wrapped = decode_json_err("sqlite checkpointer", what, { + serde_json::from_str::("\"not a number\"").unwrap_err() + }); + assert!( + format!("{wrapped}").contains(&format!("decode [corrupt] {what}")), + "non-record contexts must always tag [corrupt], even for a \ + Data-category error: {wrapped}" + ); + } + let _ = data_err; + + // The "record" context is the only one eligible for [schema]. + let wrapped = decode_json_err( + "sqlite checkpointer", + "record", + serde_json::from_str::("\"not a number\"").unwrap_err(), + ); + assert!(format!("{wrapped}").contains("decode [schema] record")); +} + #[test] fn decode_json_err_classifies_data_errors_as_schema_and_others_as_corrupt() { use crate::graph::checkpoint::decode_json_err; From 6bade4bc8ffa9896a49ec39a0f80474812a793ea Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 30 Aug 2026 16:12:59 +0300 Subject: [PATCH 12/19] feat(dag): document silent edge drop for duplicate node ids Add a prose block to the cycle-detection function explaining that when a caller supplies two DagNode entries with the same id, only the first one's edges are kept and the second's are silently discarded, which can mask cycles that the second declaration would have introduced. This behaviour is a known footgun for callers that reuse ids, so the documentation makes the contract explicit. Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/graph/dag/mod.rs | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/src/graph/dag/mod.rs b/src/graph/dag/mod.rs index fd447735..74240939 100644 --- a/src/graph/dag/mod.rs +++ b/src/graph/dag/mod.rs @@ -75,6 +75,16 @@ pub fn has_cycle(nodes: &[DagNode<'_>]) -> bool { // two conflicting declarations of the same id (e.g. `a[]` and `a[b]`) get // merged into edges that no single declaration actually names, which can // fabricate a cycle that would not exist under either declaration alone. + // + // Consequence for callers: a caller that wants to add dependency edges to + // a node that already exists in `nodes` must merge those edges into that + // node's single `DagNode` declaration, not append a second `DagNode` with + // the same id and the new edges — the second declaration's edges are + // silently dropped here (first declaration wins), so a real cycle they + // would have introduced goes undetected. This is unreachable in a caller + // that only ever mints fresh, never-reused ids (e.g. a fresh UUID per + // node), but is a live footgun for any caller that revisits an existing + // id. let mut processed: HashSet<&str> = HashSet::with_capacity(ids.len()); for node in nodes { if !processed.insert(node.id) { From b726a5860c86433096c70ea7b5ef53a854b1abc3 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 30 Aug 2026 16:13:19 +0300 Subject: [PATCH 13/19] fix(dag): clarify duplicate-id behaviour in has_cycle docs The doc comment for `has_cycle` now explains that when duplicate ids appear, only the first `DagNode` declaration is kept and any later node with the same id has its edges silently dropped, rather than contributing to the cycle check. This makes the deduplication semantics explicit for callers who may need to merge edges into the original declaration. Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/graph/dag/mod.rs | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/graph/dag/mod.rs b/src/graph/dag/mod.rs index 74240939..d7a5d109 100644 --- a/src/graph/dag/mod.rs +++ b/src/graph/dag/mod.rs @@ -63,8 +63,12 @@ pub use types::{DagIssue, DagNode}; /// Reports whether the dependency edges contain a cycle. /// /// Edges naming an id that is not a declared node are ignored (see the module -/// docs). Duplicate ids are collapsed to one node, so they cannot produce a -/// false positive. +/// docs). Duplicate ids are collapsed to one node (first declaration wins), +/// so they cannot produce a false positive — but a caller that wants to add +/// edges to a node id that already exists must merge them into that one +/// declaration; a second `DagNode` with the same id has its edges silently +/// dropped instead of contributing to the cycle check (see the comment at +/// the dedupe below). pub fn has_cycle(nodes: &[DagNode<'_>]) -> bool { let ids: HashSet<&str> = nodes.iter().map(|n| n.id).collect(); let mut indegree: HashMap<&str, usize> = ids.iter().map(|&id| (id, 0)).collect(); From 7575dbe654a87d8bf3f7dac0c33762dbf0854231 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 30 Aug 2026 16:22:41 +0300 Subject: [PATCH 14/19] fix(select): restore Send verb detection for noun-only matches MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The condition for checking resource nouns that imply the Send verb was changed from checking whether no verbs were found to checking whether Send itself was already detected. The original logic added Send when no action verb matched at all, but this caused a ranking regression: a phrase like "Post a message to #general" matches the Create verb via "post", so the noun check was skipped entirely and Send was never added, dropping SLACK_SEND_MESSAGE out of the top 15 results. The fix restores the pre-extraction behaviour where Send is added whenever any of its aliases—action words or nouns—match, keeping the verb table honest without changing which verbs are found. Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/harness/tool/select/mod.rs | 19 ++++++++++++++----- 1 file changed, 14 insertions(+), 5 deletions(-) diff --git a/src/harness/tool/select/mod.rs b/src/harness/tool/select/mod.rs index f96c8065..c63553eb 100644 --- a/src/harness/tool/select/mod.rs +++ b/src/harness/tool/select/mod.rs @@ -186,11 +186,20 @@ fn detect_verbs(prompt: &str) -> HashSet { } } } - // Resource nouns for Send are checked only when no explicit action verb - // matched above — a noun alone (e.g. "message support") may still imply - // Send, but must not add it alongside an already-detected conflicting - // verb ("delete a message", "read email"). - if found.is_empty() { + // Resource nouns for Send are checked whenever `Send` was not already + // matched by one of its action aliases above. This is exactly equivalent + // to the pre-extraction behaviour, where these nouns lived in `Send`'s + // own alias list: `Send` was added iff ANY of its aliases matched, action + // word or noun. Splitting the nouns out keeps the verb table honest + // (a noun is not an action word) without changing which verbs are found. + // + // Gating this on `found.is_empty()` instead is NOT equivalent and is a + // real ranking regression: "Post a message to #general" matches "post" + // (a `Create` alias), so `found` is non-empty and `Send` is never added + // — which ranks SLACK_CREATE_CHANNEL above SLACK_SEND_MESSAGE and drops + // SLACK_SEND_MESSAGE out of the top 15 entirely. Pinned by the host's + // pre-extraction ranking snapshot. + if !found.contains(&ToolVerb::Send) { for alias in SEND_NOUN_ALIASES { if contains_whole_word(&lowered, alias) { found.insert(ToolVerb::Send); From f5af08c9bcde9c48b42aa071e4cf2071ee3daf62 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 30 Aug 2026 16:25:03 +0300 Subject: [PATCH 15/19] chore: files changed src/harness/tool/select/mod.rs Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/harness/tool/select/mod.rs | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/src/harness/tool/select/mod.rs b/src/harness/tool/select/mod.rs index c63553eb..13beb228 100644 --- a/src/harness/tool/select/mod.rs +++ b/src/harness/tool/select/mod.rs @@ -199,7 +199,19 @@ fn detect_verbs(prompt: &str) -> HashSet { // — which ranks SLACK_CREATE_CHANNEL above SLACK_SEND_MESSAGE and drops // SLACK_SEND_MESSAGE out of the top 15 entirely. Pinned by the host's // pre-extraction ranking snapshot. - if !found.contains(&ToolVerb::Send) { + // A noun does not override a verb that CONFLICTS with sending ("read + // email", "delete a message" are Read/Delete, not Send). It does apply + // alongside `Create`, because "post/write/draft a message" is a send + // intent expressed with a creation verb — and suppressing it there is + // what dropped SLACK_SEND_MESSAGE out of the top 15 for "Post a message + // to #general". Pinned by the host's pre-extraction ranking snapshot. + let conflicts_with_send = found.iter().any(|v| { + matches!( + v, + ToolVerb::Read | ToolVerb::List | ToolVerb::Update | ToolVerb::Delete | ToolVerb::Merge + ) + }); + if !found.contains(&ToolVerb::Send) && !conflicts_with_send { for alias in SEND_NOUN_ALIASES { if contains_whole_word(&lowered, alias) { found.insert(ToolVerb::Send); From a578da83692f6340bb0ce8c4013784bdba66fe00 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 30 Aug 2026 16:50:40 +0300 Subject: [PATCH 16/19] chore: files changed src/graph/delegation/test.rs Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/graph/delegation/test.rs | 123 ++++++++++++++++++++++++----------- 1 file changed, 85 insertions(+), 38 deletions(-) diff --git a/src/graph/delegation/test.rs b/src/graph/delegation/test.rs index d02ebcba..6a6fa97a 100644 --- a/src/graph/delegation/test.rs +++ b/src/graph/delegation/test.rs @@ -878,68 +878,115 @@ async fn cancellation_at_the_approval_node_routes_to_finalize() { #[tokio::test] async fn concurrent_resume_delegation_calls_for_the_same_thread_never_overlap() { - // `resume_delegation` is a public entry point that bypasses - // `run_or_resume_delegation` entirely, so it needs its OWN per-thread - // serialization — two concurrent approval callbacks for the same paused - // thread (a retried callback racing with a deny, say) must not both load - // the same interrupt checkpoint and independently finalize, appending - // competing histories. - let concurrent = Arc::new(AtomicUsize::new(0)); - let max_concurrent = Arc::new(AtomicUsize::new(0)); - let make_runner = || { - let concurrent = concurrent.clone(); - let max_concurrent = max_concurrent.clone(); - move |stage: DelegationStage, _s: DelegationState| { - let concurrent = concurrent.clone(); - let max_concurrent = max_concurrent.clone(); - Box::pin(async move { - let now = concurrent.fetch_add(1, Ordering::SeqCst) + 1; - max_concurrent.fetch_max(now, Ordering::SeqCst); - tokio::time::sleep(std::time::Duration::from_millis(20)).await; - concurrent.fetch_sub(1, Ordering::SeqCst); - let text = match stage { - DelegationStage::Review => "approve".to_string(), - _ => "ok".to_string(), - }; - Ok::<_, String>(DelegationStageOutput::done(text)) - }) as std::pin::Pin + Send>> + // The approval interrupt's resume path routes straight from `approval` to + // `finalize` WITHOUT calling `run_stage` at all (`finalize` only reads + // accumulated state), so a `run_stage`-based concurrency probe cannot + // observe the lock during `resume_delegation` — it only ever sees the + // SETUP `run_delegation_durable` call's stage execution. Instrument the + // checkpointer's `get` instead, since that IS on `resume_delegation`'s + // critical path (the schema-mismatch check reads the checkpoint before + // dispatching), and is where two racing resumes would actually overlap + // without the per-thread lock. + struct BlockingGetCheckpointer { + inner: Arc>, + concurrent: Arc, + max_concurrent: Arc, + } + + #[async_trait::async_trait] + impl Checkpointer for BlockingGetCheckpointer { + async fn put( + &self, + checkpoint: Checkpoint, + ) -> crate::Result { + self.inner.put(checkpoint).await } - }; + + async fn get( + &self, + thread_id: &str, + checkpoint_id: Option<&str>, + ) -> crate::Result>> { + let now = self.concurrent.fetch_add(1, Ordering::SeqCst) + 1; + self.max_concurrent.fetch_max(now, Ordering::SeqCst); + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + let result = self.inner.get(thread_id, checkpoint_id).await; + self.concurrent.fetch_sub(1, Ordering::SeqCst); + result + } + + async fn list( + &self, + thread_id: &str, + ) -> crate::Result> { + self.inner.list(thread_id).await + } + + async fn list_threads(&self) -> crate::Result> { + self.inner.list_threads().await + } + + async fn delete_thread(&self, thread_id: &str) -> crate::Result<()> { + self.inner.delete_thread(thread_id).await + } + + async fn delete_checkpoints( + &self, + thread_id: &str, + ids: &[String], + ) -> crate::Result { + self.inner.delete_checkpoints(thread_id, ids).await + } + } let dir = tempfile::tempdir().unwrap(); - let cp: Arc> = + let real_cp: Arc> = Arc::new(crate::graph::checkpoint::FileCheckpointer::new(dir.path())); - let make_config = || DelegationConfig { + + // Park on the approval interrupt first, through the real checkpointer. + let setup_config = DelegationConfig { require_review_approval: true, - checkpointer: Some(cp.clone()), + checkpointer: Some(real_cp.clone()), thread_id: Some("racing-resume-thread".to_string()), ..DelegationConfig::default() }; - - // Park on the approval interrupt first. - run_delegation_durable(make_config(), make_runner()) + run_delegation_durable(setup_config, flow_runner(0)) .await .expect("parks on approval") .pending .expect("parked on the approval interrupt"); + let concurrent = Arc::new(AtomicUsize::new(0)); + let max_concurrent = Arc::new(AtomicUsize::new(0)); + let wrapped: Arc> = Arc::new(BlockingGetCheckpointer { + inner: real_cp, + concurrent: concurrent.clone(), + max_concurrent: max_concurrent.clone(), + }); + let make_config = || DelegationConfig { + require_review_approval: true, + checkpointer: Some(wrapped.clone()), + thread_id: Some("racing-resume-thread".to_string()), + ..DelegationConfig::default() + }; + let (r1, r2) = tokio::join!( - resume_delegation(make_config(), json!("approve_once"), make_runner()), - resume_delegation(make_config(), json!("approve_once"), make_runner()), + resume_delegation(make_config(), json!("approve_once"), flow_runner(0)), + resume_delegation(make_config(), json!("approve_once"), flow_runner(0)), ); // Exactly one of the two racing resumes may succeed in applying the // decision; a second resume of an already-finalized thread is expected to // either no-op or surface a graph-level error — either is acceptable, but - // no overlapping stage execution across the two calls is not. + // overlapping `get` calls across the two resumes is not: that is exactly + // the race the per-thread lock exists to prevent. assert!(r1.is_ok() || r2.is_ok(), "at least one resume must succeed"); assert_eq!( max_concurrent.load(Ordering::SeqCst), 1, - "the per-thread lock must serialize the two resumes' stage execution; \ - observed overlapping stage invocations" + "the per-thread lock must serialize the two resumes' checkpoint reads; \ + observed overlapping `get` calls" ); } - #[tokio::test] async fn resume_delegation_rejects_a_schema_mismatched_checkpoint() { // `resume_delegation` is a public entry point a host calls directly with From 0a0368b8dea816b0c917f40b5c2d8d1b428fce57 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 30 Aug 2026 16:52:28 +0300 Subject: [PATCH 17/19] chore: files changed src/graph/delegation/run.rs Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/graph/delegation/run.rs | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/src/graph/delegation/run.rs b/src/graph/delegation/run.rs index f30bc072..d02c9dda 100644 --- a/src/graph/delegation/run.rs +++ b/src/graph/delegation/run.rs @@ -184,9 +184,7 @@ where .clone() .ok_or_else(|| "delegation resume requires a checkpointer".to_string())?; - let lock = thread_lock(&thread_id); - let _guard = lock.lock().await; - + // TEMP: lock removed for verification if let Some(checkpoint) = cp .get(thread_id.as_str(), None) .await From a10c9ce7d6792d56d78f99c443e893d42558b6b0 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 30 Aug 2026 16:54:15 +0300 Subject: [PATCH 18/19] chore: files changed src/graph/delegation/run.rs Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/graph/delegation/run.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/graph/delegation/run.rs b/src/graph/delegation/run.rs index d02c9dda..f30bc072 100644 --- a/src/graph/delegation/run.rs +++ b/src/graph/delegation/run.rs @@ -184,7 +184,9 @@ where .clone() .ok_or_else(|| "delegation resume requires a checkpointer".to_string())?; - // TEMP: lock removed for verification + let lock = thread_lock(&thread_id); + let _guard = lock.lock().await; + if let Some(checkpoint) = cp .get(thread_id.as_str(), None) .await From a7a5619fa8f77c39d7982f406d3b9d90c3d5b24e Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 30 Aug 2026 16:56:42 +0300 Subject: [PATCH 19/19] chore: files changed src/harness/tool/select/test.rs Auto-committed-on: dragonfly Co-authored-by: Medulla --- src/harness/tool/select/test.rs | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/src/harness/tool/select/test.rs b/src/harness/tool/select/test.rs index 000e89b4..92795324 100644 --- a/src/harness/tool/select/test.rs +++ b/src/harness/tool/select/test.rs @@ -144,8 +144,14 @@ fn stopwords_removed() { #[test] fn verb_detection_handles_aliases() { + // Exact assertion, not `contains(Send) || contains(Create)`: the + // regression this pins is specifically that `Send` must be retained + // ALONGSIDE `Create` here, not merely that one of the two survives — an + // implementation that suppresses `Send` whenever ANY verb is found would + // still pass a permissive `||` assertion while reintroducing the exact + // ranking regression (`SLACK_SEND_MESSAGE` falling out of the top-k). let v = detect_verbs("post a message to general channel"); - assert!(v.contains(&ToolVerb::Send) || v.contains(&ToolVerb::Create)); + assert_eq!(v, HashSet::from([ToolVerb::Create, ToolVerb::Send])); let v = detect_verbs("delete all promotional emails"); assert!(v.contains(&ToolVerb::Delete));