From 0d546f8ac0bc05c578d8fa69250a8c41e7480722 Mon Sep 17 00:00:00 2001 From: "cyrus@tinyhumans.ai" Date: Wed, 29 Jul 2026 16:01:14 +0530 Subject: [PATCH 1/6] feat(flows): host the dedup node's commit-on-success half MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements the host side of the tinyflows `dedup` node (PR2, issue #5263): `DedupCommitSubscriber` in `flows::bus`, listening for `FlowRunFinished` and settling every `dedup` node in the finished flow's graph via the same per-flow `StateStore` namespace (`"flow:"`) the engine's `dedup` node itself reads/writes through. - On success (`completed`/`completed_with_warnings`): unions each dedup node's `tentative` key set into `committed`, then clears `tentative`. - On failure/cancel/interrupt (or any unrecognized status β€” fail closed toward "not committed"): clears `tentative` only, leaving `committed` untouched so the released items retry next run. - Node ids are derived from the flow's saved graph (`kind == "dedup"`) since `StateStore` has no prefix-scan. Registered alongside `FlowTriggerSubscriber` / `FlowRunDigestSubscriber` in the same `DomainGroup::Flows` startup block. - `flows::store::kv_delete` added next to `kv_get`/`kv_set` to clear `tentative` (rather than writing `[]`). - `node_contracts.rs` gains the `dedup` kind's host overlay (run-level commit semantics, canonical split_out β†’ dedup β†’ action placement); the memory overlay's dedup deferral note now redirects to the dedup node instead. The 13->14 kind-count drift tests (node_contracts, propose_workflow description, list_node_kinds) are updated in lockstep. - workflow_builder prompt.md teaches the `dedup` node as THE way to do "process each item once", replacing the #5227 "deferred to a dedicated primitive" notes; the guard test asserting that deferral is inverted to assert the dedup node is taught instead. - `about_app` catalog gains a Dedup Node (Flows) capability entry. - vendor/tinyflows gitlink bumped to c81a4cd (dedup node kind + filter half, not modified here β€” see PR1). πŸ€– Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- src/core/jsonrpc.rs | 20 + src/openhuman/about_app/catalog_data.rs | 19 + .../flows/agents/workflow_builder/prompt.md | 57 +- src/openhuman/flows/builder_tools.rs | 8 +- src/openhuman/flows/builder_tools_tests.rs | 5 +- src/openhuman/flows/bus.rs | 501 +++++++++++++++++- src/openhuman/flows/node_contracts.rs | 57 +- src/openhuman/flows/store.rs | 19 + src/openhuman/flows/tools.rs | 18 +- vendor/tinyflows | 2 +- 10 files changed, 673 insertions(+), 33 deletions(-) diff --git a/src/core/jsonrpc.rs b/src/core/jsonrpc.rs index 8aaf349f2c..e2a273eb24 100644 --- a/src/core/jsonrpc.rs +++ b/src/core/jsonrpc.rs @@ -2162,6 +2162,26 @@ fn register_domain_subscribers( "[event_bus] failed to register flows run-digest subscriber β€” bus not initialized" ); } + // Dedup commit-on-success (issue #5263 PR2): on every terminal + // `FlowRunFinished`, settles each `dedup` node in the flow's graph + // β€” unions tentative keys into committed on success, releases + // (clears) tentative on failure/cancel/interrupt. This is the + // host half of the `dedup` node's exactly-once contract; the node + // itself (`tinyflows::nodes::control_flow::dedup`) only ever + // reads `committed` and writes `tentative`. Registered in the + // same `group_first_time(DomainGroup::Flows)` block as the other + // two flows subscribers above β€” see the digest subscriber's + // comment just above for why a second `group_first_time` guard + // here would be redundant. + if let Some(handle) = crate::core::event_bus::subscribe_global(Arc::new( + crate::openhuman::flows::bus::DedupCommitSubscriber::new(Arc::new(config.clone())), + )) { + std::mem::forget(handle); + } else { + log::warn!( + "[event_bus] failed to register flows dedup-commit subscriber β€” bus not initialized" + ); + } } } else { log::debug!("[event_bus] flows trigger subscriber SKIPPED β€” Flows domain disabled"); diff --git a/src/openhuman/about_app/catalog_data.rs b/src/openhuman/about_app/catalog_data.rs index 427cd77bb9..8df78def0a 100644 --- a/src/openhuman/about_app/catalog_data.rs +++ b/src/openhuman/about_app/catalog_data.rs @@ -1528,6 +1528,25 @@ pub(super) const CAPABILITIES: &[Capability] = &[ status: CapabilityStatus::Beta, privacy: LOCAL_RAW, }, + Capability { + id: "automation.flow_dedup_node", + name: "Dedup Node (Flows)", + domain: "flows", + category: CapabilityCategory::Automation, + description: "A `dedup` node inside a saved workflow graph, giving the flow durable \ + exactly-once processing per item with no agent turn or extra plumbing \ + involved. It drops an item whose per-item key was already committed by a \ + prior successful run, and otherwise passes it through. Committing happens \ + automatically: keys the node passes through are marked done only once the \ + whole run finishes successfully; a failed/cancelled/interrupted run leaves \ + them unmarked so the same items retry next time. Only the opaque per-item \ + key strings are stored in the flow's own private, flow-scoped state β€” never \ + the item's content, and never the user's personal memory.", + how_to: "Flows editor > add a `dedup` node right after the item source; set config.key \ + to a stable per-item id expression, e.g. \"=item.id\".", + status: CapabilityStatus::Beta, + privacy: LOCAL_RAW, + }, Capability { id: "automation.view_cron_jobs", name: "View Cron Jobs", diff --git a/src/openhuman/flows/agents/workflow_builder/prompt.md b/src/openhuman/flows/agents/workflow_builder/prompt.md index 50983b878d..1a1c1d8942 100644 --- a/src/openhuman/flows/agents/workflow_builder/prompt.md +++ b/src/openhuman/flows/agents/workflow_builder/prompt.md @@ -154,7 +154,7 @@ rather than a general context recall), use `memory_hybrid_search` in its You have a machine-readable belt; use it instead of relying on memory: -- **Introspect the DSL:** `list_node_kinds` β†’ the 13 kinds; `get_node_kind_contract +- **Introspect the DSL:** `list_node_kinds` β†’ the 14 kinds; `get_node_kind_contract { kind }` β†’ one kind's exact config fields, ports, an example, and its gotchas. Consult these instead of guessing config shapes (this is the source of truth; the summary below is just orientation). @@ -304,7 +304,7 @@ A `WorkflowGraph` is `{ name?, nodes: [...], edges: [...] }`. - **Exactly ONE `trigger` node is required.** Every other node should be reachable from it; a dry-run helps catch orphans. -### The 13 node kinds +### The 14 node kinds > The authoritative, always-current config shapes, ports, examples, and gotchas > for each kind live in the `list_node_kinds` / `get_node_kind_contract { kind }` @@ -419,10 +419,8 @@ A `WorkflowGraph` is `{ name?, nodes: [...], edges: [...] }`. way.** Semantic `recall` ranks results by similarity, not exact key membership, so there is no sound `recall β†’ condition` pattern that correctly answers "have I already handled this exact item" β€” don't - improvise one. A dedicated dedup primitive is deferred to a future - iteration; until it lands, tell the user the workflow can't guarantee - exactly-once processing rather than shipping a graph that looks like it - dedupes but doesn't. + improvise one. Use a **`dedup` node** instead; see "The `dedup` node" + below. Use memory reads sparingly β€” only when the workflow genuinely needs the user's context, rather than hardcoding what memory already holds. @@ -565,6 +563,10 @@ A `WorkflowGraph` is `{ name?, nodes: [...], edges: [...] }`. 12. **`sub_workflow`** β€” `config.workflow` = an embedded child `WorkflowGraph`. 13. **`memory`** β€” reads or writes host-managed memory directly, no agent turn involved. See "The `memory` node" just below for the full reference. +14. **`dedup`** β€” commit-on-success exactly-once filter: drops an item whose + per-item key was already committed by a prior successful run. See "The + `dedup` node" below β€” this is THE way to do "process each item once", + not a memory recall/condition graph. ### The `memory` node @@ -605,8 +607,8 @@ an agent turn itself. **Dedup is out of scope for this node.** Exact "have I already processed this item" membership checks are not reliably expressible via semantic `recall` β€” `results` is similarity-ranked, not an exact-match lookup, so a -`recall β†’ condition` graph cannot safely gate on it. Don't author one; a -dedicated dedup primitive is deferred to a future iteration. Put `remember` +`recall β†’ condition` graph cannot safely gate on it. Don't author one; use a +**`dedup` node** instead (below). Put `remember` **after** the real action it records, not before β€” a failed action must never be mistaken for a completed one on the next run. See the `agent` node kind's "Reading the user's memory at run time" section above for how this node @@ -614,6 +616,45 @@ relates to `tool_call oh:memory_recall` and `flow_memory_agent` β€” all three are valid; `memory` is the right choice specifically when a non-reasoning node needs to branch on the result. +### The `dedup` node + +**This is THE way to do "process each item once / never repeat" β€” always +reach for it over an improvised memory recall/condition graph.** A `dedup` +node is a commit-on-success exactly-once filter: it drops an item whose +per-item key was already durably committed by a PRIOR successful run, and +otherwise passes the item through. Whether this run's newly-seen keys get +committed (on success) or released to retry (on failure) is handled +internally by the host after the run finishes β€” you never wire that +decision yourself. + +The correct pattern is ONE `dedup` node placed right after the items are +produced and BEFORE the action that must run at most once per item: + +``` +trigger β†’ fetch β†’ split_out β†’ dedup [key="=item.id"] β†’ …action… +``` + +- **`config.key`** (required, `"=expr"`) β€” the per-item dedup key, e.g. + `"=item.id"`. Key off a stable id that already exists at that point in the + graph β€” an issue number, message id, url, or similar β€” never something + derived from the action's own output. A key that resolves to null, + missing, or an empty string fails OPEN: the item passes through and is not + recorded (never silently dropped just because a key couldn't be computed). +- **Place it BEFORE the work, not after.** Unlike the `memory` node's + `remember` (which you place AFTER the action), `dedup` goes first in the + chain β€” it already handles "mark seen only after success" internally, so + do NOT also wire a separate `memory[remember]`/`condition` dedupe graph + alongside it; that duplicates what `dedup` already does and can disagree + with it. +- **Commit is run-level.** A saved flow's dedup nodes are settled off the + run's single terminal status: every dedup node that ran in a + `completed`/`completed_with_warnings` run gets its newly-seen keys + committed; every dedup node in a `failed`/`cancelled`/`interrupted` run has + its newly-seen keys released so they retry next time. For a flow with one + action per run this is exactly what you want; a flow chaining several + independent actions after a single `dedup` should be aware that one + action's failure retries ALL of them next run, not just the failing one. + ### Expressions: the `=` / jq convention Any config **string** beginning with `=` is an **expression** evaluated against diff --git a/src/openhuman/flows/builder_tools.rs b/src/openhuman/flows/builder_tools.rs index c7a20f6bcc..3a870c78df 100644 --- a/src/openhuman/flows/builder_tools.rs +++ b/src/openhuman/flows/builder_tools.rs @@ -2343,7 +2343,7 @@ impl Tool for ListAgentProfilesTool { // list_node_kinds / get_node_kind_contract β€” queryable DSL schema (F2) // ───────────────────────────────────────────────────────────────────────────── -/// `list_node_kinds`: enumerate the 13 tinyflows node kinds with a one-line +/// `list_node_kinds`: enumerate the 14 tinyflows node kinds with a one-line /// summary each. The DSL counterpart of `search_tool_catalog` for Composio /// actions β€” a cheap first call to orient before fetching a full contract. pub struct ListNodeKindsTool; @@ -2369,7 +2369,7 @@ impl Tool for ListNodeKindsTool { } fn description(&self) -> &str { - "List the 13 tinyflows node kinds you can put in a WorkflowGraph, each with a one-line \ + "List the 14 tinyflows node kinds you can put in a WorkflowGraph, each with a one-line \ summary and its config field names. Read-only, no args. Returns a JSON array of { kind, \ summary, required_config, optional_config }. Call get_node_kind_contract { kind } for the \ full config-field shapes, ports, an example node, and authoring gotchas of any one kind β€” \ @@ -2460,7 +2460,7 @@ impl Tool for GetNodeKindContractTool { "properties": { "kind": { "type": "string", - "description": "One of the 13 node kinds, e.g. 'tool_call' (from list_node_kinds).", + "description": "One of the 14 node kinds, e.g. 'tool_call' (from list_node_kinds).", "enum": crate::openhuman::flows::NODE_KINDS, } }, @@ -2488,7 +2488,7 @@ impl Tool for GetNodeKindContractTool { &contract, )?)), None => Ok(ToolResult::error(format!( - "'{kind}' is not a tinyflows node kind β€” call list_node_kinds for the 13 valid \ + "'{kind}' is not a tinyflows node kind β€” call list_node_kinds for the 14 valid \ kinds." ))), } diff --git a/src/openhuman/flows/builder_tools_tests.rs b/src/openhuman/flows/builder_tools_tests.rs index f72ea16184..34b717ab0d 100644 --- a/src/openhuman/flows/builder_tools_tests.rs +++ b/src/openhuman/flows/builder_tools_tests.rs @@ -1945,16 +1945,17 @@ async fn save_workflow_accepts_correctly_schemad_graph() { } #[tokio::test] -async fn list_node_kinds_tool_returns_all_thirteen() { +async fn list_node_kinds_tool_returns_all_fourteen() { let tool = ListNodeKindsTool::new(); let result = tool.execute(json!({})).await.unwrap(); assert!(!result.is_error, "{}", result.output()); let parsed: Value = serde_json::from_str(&result.output()).unwrap(); let kinds = parsed["node_kinds"].as_array().unwrap(); - assert_eq!(kinds.len(), 13); + assert_eq!(kinds.len(), 14); // Each entry carries a kind + summary + the config-field name lists. assert!(kinds.iter().any(|k| k["kind"] == "tool_call")); assert!(kinds.iter().any(|k| k["kind"] == "memory")); + assert!(kinds.iter().any(|k| k["kind"] == "dedup")); assert!(kinds.iter().all(|k| k.get("summary").is_some())); } diff --git a/src/openhuman/flows/bus.rs b/src/openhuman/flows/bus.rs index ffb4b402d1..0a3f0734be 100644 --- a/src/openhuman/flows/bus.rs +++ b/src/openhuman/flows/bus.rs @@ -19,7 +19,8 @@ use async_trait::async_trait; use serde_json::Value; use std::collections::HashSet; use std::sync::{Arc, Mutex}; -use tinyflows::model::TriggerKind; +use tinyflows::model::{NodeKind, TriggerKind}; +use tinyflows::nodes::control_flow::dedup as dedup_node; /// Reads `trigger_kind` from a flow's trigger node config, deserializing into /// `tinyflows::model::TriggerKind`. Returns `None` when the flow doesn't have @@ -471,6 +472,233 @@ fn render_run_digest(flow_name: &str, run: &FlowRun) -> String { truncate_chars(&out, DIGEST_MAX_CHARS) } +/// Listens for `DomainEvent::FlowRunFinished` and settles every `dedup` node +/// in the finished flow's graph β€” the host half of the commit-on-success +/// exactly-once contract the tinyflows `dedup` node depends on (issue #5263 +/// PR2; the filter half β€” `DedupNode` β€” is PR1, already in `vendor/tinyflows`; +/// see `tinyflows::nodes::control_flow::dedup`'s module docs for the full +/// two-sided contract this subscriber implements). +/// +/// For every `dedup` node found in the flow's saved graph: +/// - **Success** (`"completed"` / `"completed_with_warnings"`): unions the +/// node's `tentative` key set into its `committed` set, then clears +/// `tentative`. `completed_with_warnings` counts as success β€” the run +/// reached a terminal, non-retried outcome, so the items it processed are +/// genuinely done even if some non-fatal step warned. +/// - **Anything else** (`"failed"` / `"cancelled"` / `"interrupted"`, or any +/// future/unrecognized status string): clears `tentative` only, leaving +/// `committed` untouched, so the released keys are exactly as unseen as +/// before this run and the flow's next run reprocesses them. An +/// unrecognized status is deliberately treated as failure, not success β€” +/// "retry an already-done item" is always safe, "silently mark an +/// uncertain outcome as done" is not. +/// +/// `StateStore` exposes no prefix-scan, so the only way to know which +/// `dedup::*` keys exist for a flow is to derive `` from +/// the flow's own saved graph β€” this subscriber loads `flow_id`'s graph on +/// every event rather than trying to infer node ids from the event itself. +/// +/// Reuses the exact same per-flow `StateStore` namespace +/// (`"flow:"`, see `tinyflows::caps::build_capabilities` in +/// `src/openhuman/tinyflows/caps.rs`) the engine's `FlowStateStore` hands the +/// `dedup` node during the run β€” that collision with the node's own keys is +/// the entire point. +/// +/// Best-effort throughout: every failure here is logged via `tracing::warn!` +/// and swallowed, never propagated β€” by the time this subscriber observes +/// `FlowRunFinished`, the run has already settled its own `flow_runs` row, so +/// a state-store hiccup here must never retroactively affect run status. A +/// failed commit degrades to "retry next run" (an item is reprocessed, never +/// lost); a failed release degrades to "stays tentative", which the `dedup` +/// node treats as unseen anyway since it only ever consults `committed` β€” +/// neither failure mode risks silently dropping an item. +pub struct DedupCommitSubscriber { + config: Arc, +} + +impl DedupCommitSubscriber { + pub fn new(config: Arc) -> Self { + Self { config } + } + + /// The node ids of every `dedup` node in `flow_id`'s saved graph, or an + /// empty vec (logged, not propagated) if the flow can't be loaded β€” a + /// flow deleted between run-finish and this handler firing, or a + /// transient store error, both degrade to "nothing to settle" rather than + /// panicking the event bus. + fn dedup_node_ids(&self, flow_id: &str) -> Vec { + match store::get_flow(&self.config, flow_id) { + Ok(Some(flow)) => flow + .graph + .nodes + .iter() + .filter(|n| n.kind == NodeKind::Dedup) + .map(|n| n.id.clone()) + .collect(), + Ok(None) => { + tracing::debug!(target: "flows", %flow_id, "[dedup-commit] flow no longer exists β€” skipping"); + Vec::new() + } + Err(e) => { + tracing::warn!(target: "flows", %flow_id, error = %e, "[dedup-commit] failed to load flow graph β€” skipping"); + Vec::new() + } + } + } + + async fn handle_finished(&self, flow_id: &str, run_id: &str, status: &str) { + let node_ids = self.dedup_node_ids(flow_id); + if node_ids.is_empty() { + tracing::trace!(target: "flows", %flow_id, %run_id, %status, "[dedup-commit] no dedup nodes in this flow β€” nothing to settle"); + return; + } + + let success = matches!(status, "completed" | "completed_with_warnings"); + tracing::debug!( + target: "flows", %flow_id, %run_id, %status, success, + dedup_node_count = node_ids.len(), + "[dedup-commit] settling dedup nodes for finished run" + ); + + let namespace = format!("flow:{flow_id}"); + for node_id in node_ids { + if success { + self.commit(&namespace, &node_id, flow_id, run_id); + } else { + self.release(&namespace, &node_id, flow_id, run_id); + } + } + } + + /// Success path: union this node's `tentative` set into `committed`, then + /// clear `tentative`. + fn commit(&self, namespace: &str, node_id: &str, flow_id: &str, run_id: &str) { + let tentative_key = dedup_node::tentative_key(node_id); + let committed_key = dedup_node::committed_key(node_id); + + let tentative = load_key_set(&self.config, namespace, &tentative_key); + if tentative.is_empty() { + tracing::trace!(target: "flows", %flow_id, %run_id, node_id, "[dedup-commit] no tentative keys β€” nothing to commit"); + return; + } + + let mut committed = load_key_set(&self.config, namespace, &committed_key); + let added = tentative + .iter() + .filter(|k| committed.insert((*k).clone())) + .count(); + + if let Err(e) = store_key_set(&self.config, namespace, &committed_key, &committed) { + tracing::warn!( + target: "flows", %flow_id, %run_id, node_id, error = %e, + "[dedup-commit] failed to write committed set β€” tentative left in place, will \ + retry the commit on this node's next successful run" + ); + return; + } + tracing::debug!( + target: "flows", %flow_id, %run_id, node_id, added, committed_len = committed.len(), + "[dedup-commit] committed tentative keys" + ); + + if let Err(e) = store::kv_delete(&self.config, namespace, &tentative_key) { + tracing::warn!( + target: "flows", %flow_id, %run_id, node_id, error = %e, + "[dedup-commit] committed but failed to clear tentative β€” harmless: the next \ + run's dedup load will re-union the same, now-already-committed keys (committed \ + is a set, so re-adding them is a no-op)" + ); + } + } + + /// Failure path: clear `tentative` only, leaving `committed` untouched so + /// the released keys retry on the flow's next run. + fn release(&self, namespace: &str, node_id: &str, flow_id: &str, run_id: &str) { + let released = + load_key_set(&self.config, namespace, &dedup_node::tentative_key(node_id)).len(); + if released == 0 { + tracing::trace!(target: "flows", %flow_id, %run_id, node_id, "[dedup-commit] no tentative keys β€” nothing to release"); + return; + } + match store::kv_delete(&self.config, namespace, &dedup_node::tentative_key(node_id)) { + Ok(()) => tracing::debug!( + target: "flows", %flow_id, %run_id, node_id, released, + "[dedup-commit] released tentative keys β€” will retry next run" + ), + Err(e) => tracing::warn!( + target: "flows", %flow_id, %run_id, node_id, error = %e, + "[dedup-commit] failed to release tentative β€” those keys remain tentative until \ + a future successful commit reconciles them (harmless: committed stays untouched \ + either way, so no item is ever wrongly marked done)" + ), + } + } +} + +#[async_trait] +impl EventHandler for DedupCommitSubscriber { + fn name(&self) -> &str { + "flows::dedup_commit" + } + + fn domains(&self) -> Option<&[&str]> { + // Same reasoning as `FlowRunDigestSubscriber::domains` just above: + // `FlowRunFinished` is tagged `"cron"` by `DomainEvent::domain()`. + Some(&["cron"]) + } + + async fn handle(&self, event: &DomainEvent) { + if let DomainEvent::FlowRunFinished { + flow_id, + run_id, + status, + } = event + { + self.handle_finished(flow_id, run_id, status).await; + } + } +} + +/// Loads a `dedup` node's key set (stored as a JSON array of strings) from +/// the flow-state KV table. Mirrors +/// `tinyflows::nodes::control_flow::dedup`'s own key-set loader: a missing +/// key, a non-array value, or an array with non-string elements all degrade +/// to an empty set rather than an error β€” a first run against a fresh store +/// has nothing recorded yet, which is not a fault. +fn load_key_set(config: &Config, namespace: &str, key: &str) -> HashSet { + match store::kv_get(config, namespace, key) { + Ok(Some(value)) => value + .as_array() + .map(|arr| { + arr.iter() + .filter_map(Value::as_str) + .map(str::to_string) + .collect() + }) + .unwrap_or_default(), + Ok(None) => HashSet::new(), + Err(e) => { + tracing::warn!(target: "flows", %namespace, key, error = %e, "[dedup-commit] failed to load key set β€” treating as empty"); + HashSet::new() + } + } +} + +/// Persists `set` under `key` as a JSON array of strings, sorted for a +/// stable, diffable on-disk representation (membership is exact-match either +/// way, so sort order carries no semantic meaning). +fn store_key_set( + config: &Config, + namespace: &str, + key: &str, + set: &HashSet, +) -> anyhow::Result<()> { + let mut keys: Vec = set.iter().cloned().collect(); + keys.sort_unstable(); + let value = Value::Array(keys.into_iter().map(Value::String).collect()); + store::kv_set(config, namespace, key, &value) +} + #[cfg(test)] mod tests { use super::*; @@ -532,6 +760,38 @@ mod tests { } } + fn dedup_node(id: &str) -> Node { + Node { + id: id.to_string(), + kind: NodeKind::Dedup, + type_version: 1, + name: id.to_string(), + config: json!({ "key": "=item.id" }), + ports: Vec::new(), + position: None, + } + } + + /// A saved flow with a `trigger` node plus one `dedup` node with id + /// `dedup_id` β€” the minimal graph [`DedupCommitSubscriber::dedup_node_ids`] + /// needs to find something to settle. + fn flow_with_dedup_node(id: &str, dedup_id: &str) -> Flow { + Flow { + id: id.to_string(), + name: id.to_string(), + enabled: true, + graph: WorkflowGraph { + nodes: vec![trigger_node(json!({})), dedup_node(dedup_id)], + ..Default::default() + }, + created_at: "2026-01-01T00:00:00Z".to_string(), + updated_at: "2026-01-01T00:00:00Z".to_string(), + last_run_at: None, + last_status: None, + require_approval: false, + } + } + #[test] fn name_and_domains_are_stable() { let tmp = tempfile::TempDir::new().unwrap(); @@ -935,4 +1195,243 @@ mod tests { assert!(digest.contains("n1")); assert!(digest.chars().count() <= DIGEST_MAX_CHARS); } + + // ── DedupCommitSubscriber ──────────────────────────────────────── + + fn dedup_state_namespace(flow_id: &str) -> String { + // MUST match `tinyflows::build_capabilities`'s `state_namespace` + // (`src/openhuman/tinyflows/caps.rs`) β€” this test asserts the + // subscriber collides with the SAME keys the engine's `dedup` node + // itself reads/writes, not just "some" namespace. + format!("flow:{flow_id}") + } + + #[test] + fn dedup_commit_name_and_domains_are_stable() { + let tmp = tempfile::TempDir::new().unwrap(); + let sub = DedupCommitSubscriber::new(test_config(&tmp)); + assert_eq!(sub.name(), "flows::dedup_commit"); + assert_eq!(sub.domains(), Some(&["cron"][..])); + } + + #[tokio::test] + async fn dedup_commit_ignores_unrelated_events() { + let tmp = tempfile::TempDir::new().unwrap(); + let sub = DedupCommitSubscriber::new(test_config(&tmp)); + // Must not panic for any event other than `FlowRunFinished`. + sub.handle(&DomainEvent::CronJobTriggered { + job_id: "j1".into(), + job_name: "test".into(), + job_type: "shell".into(), + }) + .await; + } + + #[tokio::test] + async fn dedup_commit_flow_with_no_dedup_nodes_is_a_noop() { + let tmp = tempfile::TempDir::new().unwrap(); + let config = test_config(&tmp); + let flow = flow_with_trigger_config("f-no-dedup", true, json!({})); + store::upsert_flow(&config, &flow).unwrap(); + + let sub = DedupCommitSubscriber::new(config); + // Must not panic when the flow has no `dedup` node at all. + sub.handle(&DomainEvent::FlowRunFinished { + flow_id: "f-no-dedup".into(), + run_id: "run-1".into(), + status: "completed".into(), + }) + .await; + } + + #[tokio::test] + async fn dedup_commit_unions_tentative_into_committed_and_clears_tentative_on_success() { + let tmp = tempfile::TempDir::new().unwrap(); + let config = test_config(&tmp); + let flow = flow_with_dedup_node("f-ok", "dd"); + store::upsert_flow(&config, &flow).unwrap(); + + let namespace = dedup_state_namespace("f-ok"); + store::kv_set(&config, &namespace, "dedup:dd:committed", &json!(["a"])).unwrap(); + store::kv_set( + &config, + &namespace, + "dedup:dd:tentative", + &json!(["b", "c"]), + ) + .unwrap(); + + let sub = DedupCommitSubscriber::new(config.clone()); + sub.handle(&DomainEvent::FlowRunFinished { + flow_id: "f-ok".into(), + run_id: "run-ok".into(), + status: "completed".into(), + }) + .await; + + let committed = store::kv_get(&config, &namespace, "dedup:dd:committed") + .unwrap() + .expect("committed key must still exist"); + let mut committed: Vec<&str> = committed + .as_array() + .unwrap() + .iter() + .map(|v| v.as_str().unwrap()) + .collect(); + committed.sort_unstable(); + assert_eq!(committed, vec!["a", "b", "c"], "committed = union"); + + assert!( + store::kv_get(&config, &namespace, "dedup:dd:tentative") + .unwrap() + .is_none(), + "tentative must be cleared after a successful commit" + ); + } + + #[tokio::test] + async fn dedup_commit_treats_completed_with_warnings_as_success() { + let tmp = tempfile::TempDir::new().unwrap(); + let config = test_config(&tmp); + let flow = flow_with_dedup_node("f-warn", "dd"); + store::upsert_flow(&config, &flow).unwrap(); + + let namespace = dedup_state_namespace("f-warn"); + store::kv_set(&config, &namespace, "dedup:dd:tentative", &json!(["x"])).unwrap(); + + let sub = DedupCommitSubscriber::new(config.clone()); + sub.handle(&DomainEvent::FlowRunFinished { + flow_id: "f-warn".into(), + run_id: "run-warn".into(), + status: "completed_with_warnings".into(), + }) + .await; + + let committed = store::kv_get(&config, &namespace, "dedup:dd:committed") + .unwrap() + .expect("completed_with_warnings must still commit"); + assert_eq!(committed, json!(["x"])); + assert!(store::kv_get(&config, &namespace, "dedup:dd:tentative") + .unwrap() + .is_none()); + } + + #[tokio::test] + async fn dedup_commit_releases_tentative_without_touching_committed_on_failure() { + let tmp = tempfile::TempDir::new().unwrap(); + let config = test_config(&tmp); + let flow = flow_with_dedup_node("f-failed", "dd"); + store::upsert_flow(&config, &flow).unwrap(); + + let namespace = dedup_state_namespace("f-failed"); + store::kv_set(&config, &namespace, "dedup:dd:committed", &json!(["a"])).unwrap(); + store::kv_set(&config, &namespace, "dedup:dd:tentative", &json!(["b"])).unwrap(); + + let sub = DedupCommitSubscriber::new(config.clone()); + sub.handle(&DomainEvent::FlowRunFinished { + flow_id: "f-failed".into(), + run_id: "run-failed".into(), + status: "failed".into(), + }) + .await; + + assert_eq!( + store::kv_get(&config, &namespace, "dedup:dd:committed") + .unwrap() + .unwrap(), + json!(["a"]), + "committed must be untouched by a failed run" + ); + assert!( + store::kv_get(&config, &namespace, "dedup:dd:tentative") + .unwrap() + .is_none(), + "tentative must be released (cleared) on failure so the item retries" + ); + } + + #[tokio::test] + async fn dedup_commit_releases_tentative_on_cancelled_and_interrupted() { + for status in ["cancelled", "interrupted"] { + let tmp = tempfile::TempDir::new().unwrap(); + let config = test_config(&tmp); + let flow_id = format!("f-{status}"); + let flow = flow_with_dedup_node(&flow_id, "dd"); + store::upsert_flow(&config, &flow).unwrap(); + + let namespace = dedup_state_namespace(&flow_id); + store::kv_set(&config, &namespace, "dedup:dd:tentative", &json!(["z"])).unwrap(); + + let sub = DedupCommitSubscriber::new(config.clone()); + sub.handle(&DomainEvent::FlowRunFinished { + flow_id: flow_id.clone(), + run_id: format!("run-{status}"), + status: status.to_string(), + }) + .await; + + assert!( + store::kv_get(&config, &namespace, "dedup:dd:committed") + .unwrap() + .is_none(), + "status {status} must never commit" + ); + assert!( + store::kv_get(&config, &namespace, "dedup:dd:tentative") + .unwrap() + .is_none(), + "status {status} must release tentative" + ); + } + } + + #[tokio::test] + async fn dedup_commit_two_dedup_nodes_settle_independently() { + let tmp = tempfile::TempDir::new().unwrap(); + let config = test_config(&tmp); + let flow = Flow { + id: "f-multi".to_string(), + name: "f-multi".to_string(), + enabled: true, + graph: WorkflowGraph { + nodes: vec![ + trigger_node(json!({})), + dedup_node("dd1"), + dedup_node("dd2"), + ], + ..Default::default() + }, + created_at: "2026-01-01T00:00:00Z".to_string(), + updated_at: "2026-01-01T00:00:00Z".to_string(), + last_run_at: None, + last_status: None, + require_approval: false, + }; + store::upsert_flow(&config, &flow).unwrap(); + + let namespace = dedup_state_namespace("f-multi"); + store::kv_set(&config, &namespace, "dedup:dd1:tentative", &json!(["a"])).unwrap(); + store::kv_set(&config, &namespace, "dedup:dd2:tentative", &json!(["b"])).unwrap(); + + let sub = DedupCommitSubscriber::new(config.clone()); + sub.handle(&DomainEvent::FlowRunFinished { + flow_id: "f-multi".into(), + run_id: "run-multi".into(), + status: "completed".into(), + }) + .await; + + assert_eq!( + store::kv_get(&config, &namespace, "dedup:dd1:committed") + .unwrap() + .unwrap(), + json!(["a"]) + ); + assert_eq!( + store::kv_get(&config, &namespace, "dedup:dd2:committed") + .unwrap() + .unwrap(), + json!(["b"]) + ); + } } diff --git a/src/openhuman/flows/node_contracts.rs b/src/openhuman/flows/node_contracts.rs index 520bc35ae3..8d72b42fe6 100644 --- a/src/openhuman/flows/node_contracts.rs +++ b/src/openhuman/flows/node_contracts.rs @@ -77,16 +77,32 @@ fn apply_host_overlay(contract: NodeKindContract) -> NodeKindContract { "scope: \"flow\" reads/writes the SAME per-flow memory namespace the \ flow_memory_recall / flow_memory_remember agent tools use β€” a memory[remember] \ node and a flow_memory_remember tool call inside an agent node on the same flow \ - see each other's writes. Exact \"process each item once\" dedup is NOT reliably \ - expressible via semantic recall (results are similarity-ranked, not an exact \ - membership check) β€” that is deferred to a dedicated primitive; don't improvise a \ - recallβ†’condition dedupe graph.", + see each other's writes. For exact \"process each item once\" dedup, use a dedup \ + node instead β€” semantic recall is similarity-ranked, not an exact membership check, \ + so a recallβ†’condition graph cannot safely express it.", ), + "dedup" => contract + .with_note( + "Commit is run-LEVEL, not node-level: the host settles every dedup node in the \ + flow off the run's single terminal FlowRunFinished status. \ + completed/completed_with_warnings unions this run's tentative keys into \ + committed for EVERY dedup node that ran; failed/cancelled/interrupted releases \ + tentative for ALL of them, untouched committed, so the whole run's items retry \ + next time β€” one node failing mid-run releases every dedup node's tentative in \ + that run, not just the failing one's.", + ) + .with_note( + "Canonical placement: split_out β†’ dedup [key=\"=item.id\"] β†’ …action…, i.e. one \ + dedup node right after the items are produced, keyed on a stable id that exists \ + at that point (issue number, message id, url), placed BEFORE the action. It \ + already marks a key seen only after the run succeeds, so don't also wire a \ + separate memory remember/condition dedupe graph alongside it.", + ), _ => contract, } } -/// All 13 node-kind contracts with this host's overlay applied, in +/// All 14 node-kind contracts with this host's overlay applied, in /// [`NODE_KINDS`] order. pub fn all_node_kind_contracts() -> Vec { tinyflows::catalog::all_contracts() @@ -96,7 +112,7 @@ pub fn all_node_kind_contracts() -> Vec { } /// The overlaid contract for one node kind, or `None` if `kind` is not one of -/// the 13. +/// the 14. pub fn node_kind_contract(kind: &str) -> Option { tinyflows::catalog::contract_for(kind).map(apply_host_overlay) } @@ -146,8 +162,8 @@ mod tests { use super::*; #[test] - fn overlay_preserves_all_13_kinds() { - assert_eq!(all_node_kind_contracts().len(), 13); + fn overlay_preserves_all_14_kinds() { + assert_eq!(all_node_kind_contracts().len(), 14); for kind in NODE_KINDS { assert!(node_kind_contract(kind).is_some(), "missing {kind}"); } @@ -155,17 +171,34 @@ mod tests { } #[test] - fn memory_overlay_adds_flow_memory_coherence_facts() { + fn memory_overlay_adds_flow_memory_coherence_facts_and_redirects_dedup_to_its_own_node() { let c = node_kind_contract("memory").unwrap(); let notes = c.notes.join("\n"); assert!(notes.contains("flow_memory_recall"), "{notes}"); assert!(notes.contains("flow_memory_remember"), "{notes}"); assert!(notes.contains("SAME per-flow memory namespace"), "{notes}"); - // The dedup recipe was removed (P1 review fix): semantic recall - // cannot express exact "have I seen this key" membership, so the - // overlay must not teach a recallβ†’condition dedupe pattern. + // The recallβ†’condition dedupe recipe stays gone (P1 review fix): + // semantic recall cannot express exact "have I seen this key" + // membership, so the overlay must not teach that pattern. assert!(!notes.contains("Canonical dedupe pattern"), "{notes}"); assert!(!notes.contains("item.json.found"), "{notes}"); + // The "deferred to a dedicated primitive" note is gone now that the + // dedup node exists β€” the memory overlay redirects to it instead. + assert!( + !notes.contains("deferred to a dedicated primitive"), + "{notes}" + ); + assert!(notes.contains("use a dedup node instead"), "{notes}"); + } + + #[test] + fn dedup_overlay_teaches_run_level_commit_semantics_and_placement() { + let c = node_kind_contract("dedup").unwrap(); + let notes = c.notes.join("\n"); + assert!(notes.contains("FlowRunFinished"), "{notes}"); + assert!(notes.contains("completed_with_warnings"), "{notes}"); + assert!(notes.contains("failed/cancelled/interrupted"), "{notes}"); + assert!(notes.contains("split_out β†’ dedup"), "{notes}"); } #[test] diff --git a/src/openhuman/flows/store.rs b/src/openhuman/flows/store.rs index 0927a17b68..0712aaac2c 100644 --- a/src/openhuman/flows/store.rs +++ b/src/openhuman/flows/store.rs @@ -613,6 +613,25 @@ pub fn kv_set( }) } +/// Deletes one key from the `flow_state` KV table, scoped to `namespace`. +/// A no-op (not an error) when the key doesn't exist. +/// +/// Used by `flows::bus::DedupCommitSubscriber` (issue #5263 PR2) to clear a +/// `dedup` node's `tentative` key set once a run's outcome has been settled β€” +/// preferred over `kv_set(.., json!([]))` because an absent key reads back as +/// `None` (an unambiguous "nothing pending"), matching what a fresh flow that +/// never ran a dedup node also reads back as. +pub fn kv_delete(config: &Config, namespace: &str, key: &str) -> Result<()> { + with_connection(config, |conn| { + conn.execute( + "DELETE FROM flow_state WHERE namespace = ?1 AND key = ?2", + params![namespace, key], + ) + .context("Failed to delete flow state value")?; + Ok(()) + }) +} + /// Shared column list for every `flow_runs` SELECT β€” keeps /// [`map_flow_run_row`]'s positional `row.get(N)` calls in sync. const FLOW_RUN_COLUMNS: &str = "id, flow_id, thread_id, status, started_at, finished_at, \ diff --git a/src/openhuman/flows/tools.rs b/src/openhuman/flows/tools.rs index 685eb9dce2..f49fed48e9 100644 --- a/src/openhuman/flows/tools.rs +++ b/src/openhuman/flows/tools.rs @@ -57,7 +57,7 @@ impl Tool for ProposeWorkflowTool { (to_port just stays \"main\"); routing is keyed exclusively on from_port, so a label \ on to_port instead silently turns the branch into an unconditional fan-out and is a \ hard reject. Exactly ONE \ - trigger node is required. The 13 node kinds: trigger (config.trigger_kind: manual | \ + trigger node is required. The 14 node kinds: trigger (config.trigger_kind: manual | \ schedule | webhook | app_event | form | chat_message | evaluation | system | \ execute_by_workflow; schedule needs config.schedule = {kind:\"cron\",expr,tz?} | \ {kind:\"at\",at} | {kind:\"every\",every_ms}; app_event needs config.toolkit + \ @@ -75,9 +75,12 @@ impl Tool for ProposeWorkflowTool { this flow's own memory and the ONLY scope remember/forget may target, \"flows\" is \ cross-flow READ-ONLY; config.query for recall/search; config.flavour for the flavour \ slug; config.key/config.value for remember/forget. Place remember AFTER the real \ - action it records, never before, so a failed action never marks an item as done. \ - Exact \"process each item once\" dedup is not reliably expressible via semantic \ - recall β€” don't improvise a recall/condition dedupe graph). If \ + action it records, never before, so a failed action never marks an item as done), \ + dedup (config.key REQUIRED: an \"=expr\" per-item key, e.g. \"=item.id\"; drops an item \ + whose key was already committed by a PRIOR successful run, else passes it through. Use \ + this β€” not a memory recall/condition graph β€” for exact \"process each item once\": \ + place it right after the item source and before the action, e.g. split_out β†’ dedup β†’ \ + …action…). If \ validation fails, fix the graph and call this tool again." } @@ -104,7 +107,8 @@ impl Tool for ProposeWorkflowTool { "enum": [ "trigger", "agent", "tool_call", "http_request", "code", "condition", "switch", "merge", "split_out", - "transform", "output_parser", "sub_workflow", "memory" + "transform", "output_parser", "sub_workflow", "memory", + "dedup" ] }, "name": { "type": "string", "description": "Human-readable node name." }, @@ -483,6 +487,10 @@ fn config_hint(node: &Node) -> Option { }; Some(truncate_hint(&hint)) } + NodeKind::Dedup => cfg + .get("key") + .and_then(Value::as_str) + .map(|k| format!("key: {k}")), NodeKind::Merge | NodeKind::OutputParser | NodeKind::Trigger => None, } } diff --git a/vendor/tinyflows b/vendor/tinyflows index ff9950186a..c81a4cd5a0 160000 --- a/vendor/tinyflows +++ b/vendor/tinyflows @@ -1 +1 @@ -Subproject commit ff9950186ac66775a554c80aa420920b4f39e4de +Subproject commit c81a4cd5a0d3fc8b0ebd98dc3991e6ea381ee0c3 From 0233350bbab4a8e73b62fe25cb925baecdbcc657 Mon Sep 17 00:00:00 2001 From: "cyrus@tinyhumans.ai" Date: Wed, 29 Jul 2026 16:27:25 +0530 Subject: [PATCH 2/6] fix(dedup): serialize DedupCommitSubscriber per flow id + bump tinyflows gitlink MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Per-flow-id async keyed lock (FLOW_COMMIT_LOCKS) around the commit read-modify-write, so overlapping FlowRunFinished events for the same flow can't clobber each other's committed set (closes the durable half of the StateStore race; residual node-side tentative race documented + deferred). - Bump vendor/tinyflows gitlink to 0ce4fbc (adds the concurrency docs). - Regenerate Cargo.lock + app/src-tauri/Cargo.lock: adds the futures/tokio entries tinyhumans-sdk needs β€” a pre-existing --locked drift from main's e213bc5fc (gitlink bumped without regenerating the lock), which blocks CI for any PR branched off current main. --- Cargo.lock | 4 +- app/src-tauri/Cargo.lock | 2 + src/openhuman/flows/bus.rs | 295 ++++++++++++++++++++++++++++++++++++- vendor/tinyflows | 2 +- 4 files changed, 298 insertions(+), 5 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 15b9c82afe..df4abb5378 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7308,7 +7308,7 @@ dependencies = [ "jaq-core", "jaq-json", "jaq-std", - "reqwest 0.12.28", + "reqwest", "serde", "serde_json", "thiserror 2.0.18", @@ -7322,11 +7322,13 @@ name = "tinyhumans-sdk" version = "0.1.0" dependencies = [ "base64 0.22.1", + "futures", "percent-encoding", "reqwest", "serde", "serde_json", "thiserror 2.0.18", + "tokio", "url", ] diff --git a/app/src-tauri/Cargo.lock b/app/src-tauri/Cargo.lock index 94ad8b9e79..593fe519e7 100644 --- a/app/src-tauri/Cargo.lock +++ b/app/src-tauri/Cargo.lock @@ -9038,11 +9038,13 @@ name = "tinyhumans-sdk" version = "0.1.0" dependencies = [ "base64 0.22.1", + "futures", "percent-encoding", "reqwest 0.12.28", "serde", "serde_json", "thiserror 2.0.18", + "tokio", "url", ] diff --git a/src/openhuman/flows/bus.rs b/src/openhuman/flows/bus.rs index 0a3f0734be..e739998fa4 100644 --- a/src/openhuman/flows/bus.rs +++ b/src/openhuman/flows/bus.rs @@ -17,8 +17,8 @@ use crate::openhuman::flows::{flow_namespace, Flow, FlowRun}; use crate::openhuman::memory::{Memory, MemoryCategory, MemoryTaint}; use async_trait::async_trait; use serde_json::Value; -use std::collections::HashSet; -use std::sync::{Arc, Mutex}; +use std::collections::{HashMap, HashSet}; +use std::sync::{Arc, LazyLock, Mutex}; use tinyflows::model::{NodeKind, TriggerKind}; use tinyflows::nodes::control_flow::dedup as dedup_node; @@ -512,13 +512,121 @@ fn render_run_digest(flow_name: &str, run: &FlowRun) -> String { /// lost); a failed release degrades to "stays tentative", which the `dedup` /// node treats as unseen anyway since it only ever consults `committed` β€” /// neither failure mode risks silently dropping an item. +/// +/// **Commit atomicity (issue #5265, CodeRabbit "Major" on the dedup engine +/// PR):** the per-node commit itself is a read-modify-write +/// (`load(committed) β†’ union(tentative) β†’ store(committed) β†’ delete +/// (tentative)`), not a compare-and-swap. Two overlapping `FlowRunFinished` +/// events for the SAME `flow_id` (e.g. a scheduled run and a manual re-run +/// racing each other) could otherwise interleave their read-modify-writes +/// and have the second writer's `store(committed)` clobber the first +/// writer's union, silently losing that run's committed keys +/// (last-writer-wins). [`handle_finished`](Self::handle_finished) closes +/// that DURABLE half of the race by serializing all of a given flow's +/// dedup-node settlement through a per-`flow_id` lock (see +/// [`FLOW_COMMIT_LOCKS`]) β€” different flows never contend. This does NOT +/// fix the node-side half: the `dedup` node's own in-run `StateStore` +/// read-modify-write (a single run unioning its own newly-seen items into +/// `tentative`) is a separate, still-open limitation documented on +/// `tinyflows::nodes::control_flow::dedup`'s side; a full CAS-based +/// `StateStore` is deferred. pub struct DedupCommitSubscriber { config: Arc, + /// Test-only instrumentation β€” see [`CommitTestHooks`]. Always `None` in + /// production (`DedupCommitSubscriber::new`). + #[cfg(test)] + test_hooks: Option>, +} + +/// Process-global registry of per-flow commit locks (issue #5265). Keyed by +/// `flow_id` so unrelated flows never contend with each other; the shared +/// `tokio::sync::Mutex<()>` per key lets [`DedupCommitSubscriber:: +/// handle_finished`] hold a guard across its whole (synchronous) +/// read-modify-write section for that flow. Mirrors the same +/// `LazyLock>>>>` keyed-lock +/// idiom `update_memory_md`'s `WORKSPACE_WRITE_LOCKS` uses for an analogous +/// read-modify-write race (#4458) β€” grepped for an existing pattern before +/// adding this one; that's the closest match in the crate. +/// +/// Deliberately unbounded, matching that precedent: flow ids are bounded in +/// practice (a user's saved flow set), so an evicting map would be +/// complexity this doesn't need yet. +static FLOW_COMMIT_LOCKS: LazyLock>>>> = + LazyLock::new(|| Mutex::new(HashMap::new())); + +/// Returns (creating if needed) the shared async commit lock for `flow_id`. +fn flow_commit_lock(flow_id: &str) -> Arc> { + let mut map = FLOW_COMMIT_LOCKS + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + Arc::clone( + map.entry(flow_id.to_string()) + .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(()))), + ) +} + +/// Test-only scheduling/witness hooks for proving [`FLOW_COMMIT_LOCKS`]' +/// mutual exclusion. Deliberately **instance-scoped** (owned by one +/// [`DedupCommitSubscriber`], via [`DedupCommitSubscriber::with_test_hooks`]) +/// rather than a process-global static: cargo's test harness runs different +/// `#[tokio::test]` functions concurrently on separate OS threads, and a +/// global counter would have unrelated tests' ordinary (unarmed, +/// effectively-instant) commits interleave with β€” and pollute β€” a +/// concurrency test's high-water-mark measurement purely by scheduling +/// chance. Scoping the hooks to one test's own `Arc` means only tasks that +/// share that specific subscriber instance can ever touch its counters. +#[cfg(test)] +#[derive(Default)] +struct CommitTestHooks { + delay_ms: std::sync::atomic::AtomicU64, + concurrent: std::sync::atomic::AtomicUsize, + max_concurrent: std::sync::atomic::AtomicUsize, } impl DedupCommitSubscriber { pub fn new(config: Arc) -> Self { - Self { config } + Self { + config, + #[cfg(test)] + test_hooks: None, + } + } + + /// Test constructor: attaches [`CommitTestHooks`] so a test can arm a + /// delay inside the commit critical section and observe how many + /// `handle_finished` calls were concurrently inside it. + #[cfg(test)] + fn with_test_hooks(config: Arc, hooks: Arc) -> Self { + Self { + config, + test_hooks: Some(hooks), + } + } + + /// No-op unless [`Self::with_test_hooks`] attached hooks β€” awaited right + /// after `handle_finished` acquires the per-flow commit lock, while + /// still holding it. This is what makes it possible to force two + /// spawned tasks to genuinely interleave on a single-threaded test + /// executor (there are no other `.await` points inside the + /// commit/release critical section to give the executor a chance to + /// poll a contending task) β€” a test can then prove the lock, not + /// accidental scheduling luck, is what serializes two overlapping + /// `FlowRunFinished` events for the same flow. Compiles to an empty + /// async fn body (zero-cost) in non-test builds. + async fn maybe_test_delay(&self) { + #[cfg(test)] + if let Some(hooks) = &self.test_hooks { + use std::sync::atomic::Ordering; + let now = hooks.concurrent.fetch_add(1, Ordering::SeqCst) + 1; + hooks.max_concurrent.fetch_max(now, Ordering::SeqCst); + + let ms = hooks.delay_ms.load(Ordering::SeqCst); + if ms > 0 { + tokio::time::sleep(std::time::Duration::from_millis(ms)).await; + } + + hooks.concurrent.fetch_sub(1, Ordering::SeqCst); + } } /// The node ids of every `dedup` node in `flow_id`'s saved graph, or an @@ -560,6 +668,17 @@ impl DedupCommitSubscriber { "[dedup-commit] settling dedup nodes for finished run" ); + // Serialize this flow's settlement against any other overlapping + // `FlowRunFinished` handling for the SAME flow_id β€” held across the + // whole read-modify-write loop below so two overlapping runs can + // never interleave their load(committed)+union(tentative)+ + // store(committed) and lose one run's keys. See `FLOW_COMMIT_LOCKS` + // docs for the full race this closes. + let lock = flow_commit_lock(flow_id); + let lock_guard = lock.lock().await; + tracing::trace!(target: "flows", %flow_id, %run_id, "[dedup-commit] acquired per-flow commit lock"); + self.maybe_test_delay().await; + let namespace = format!("flow:{flow_id}"); for node_id in node_ids { if success { @@ -568,6 +687,9 @@ impl DedupCommitSubscriber { self.release(&namespace, &node_id, flow_id, run_id); } } + + drop(lock_guard); + tracing::trace!(target: "flows", %flow_id, %run_id, "[dedup-commit] released per-flow commit lock"); } /// Success path: union this node's `tentative` set into `committed`, then @@ -1434,4 +1556,171 @@ mod tests { json!(["b"]) ); } + + // ── per-flow commit serialization (issue #5265) ─────────────────── + // + // CodeRabbit "Major" on the dedup engine PR: the commit's + // load(committed)+union(tentative)+store(committed) is a + // read-modify-write, not a CAS. Two overlapping `FlowRunFinished` + // events for the SAME flow could otherwise interleave and have the + // second writer's store clobber the first writer's union, silently + // losing that run's committed keys. `handle_finished` now serializes + // settlement per `flow_id` via `FLOW_COMMIT_LOCKS`. + // + // Two tests, deliberately split: + // + // - `..._never_runs_two_commits_for_the_same_flow_concurrently` spawns a + // burst of genuinely overlapping `FlowRunFinished` events for the SAME + // flow_id and proves the LOCK itself provides mutual exclusion (the + // high-water mark of concurrently-active critical sections never + // exceeds 1) β€” this is the "spawn two tasks contending on the same + // flow_id" case. + // - `..._serial_commits_for_the_same_flow_accumulate_via_union` proves + // the property that mutual exclusion protects: settling run after run + // for the same node never clobbers an earlier run's committed keys β€” + // each contributes to the union. + // + // These are split rather than combined into one "two runs with two + // different tentative sets, truly concurrently, assert union" test + // because `tentative` is a single shared KV row per node (not + // per-run) β€” forcing two *different* tentative contents to both survive + // a genuinely simultaneous read would require injecting a write from + // outside `handle_finished` in the middle of its critical section, which + // instead exercises the SEPARATE, still-open node-side race (the + // `dedup` node's own in-run `tentative` read-modify-write, documented on + // `DedupCommitSubscriber` above as explicitly NOT fixed by this lock). + // Together, the two tests below establish the same guarantee end to + // end: the lock enforces serialization (test 1), and serialization is + // sufficient for correctness (test 2). + + #[tokio::test] + async fn dedup_commit_never_runs_two_commits_for_the_same_flow_concurrently() { + let tmp = tempfile::TempDir::new().unwrap(); + let config = test_config(&tmp); + let flow = flow_with_dedup_node("f-race", "dd"); + store::upsert_flow(&config, &flow).unwrap(); + + let namespace = dedup_state_namespace("f-race"); + store::kv_set(&config, &namespace, "dedup:dd:tentative", &json!(["seed"])).unwrap(); + + // Arm the test-only scheduling hook (see `CommitTestHooks`): every + // `handle_finished` call sleeps briefly while holding the per-flow + // lock, and records how many calls are concurrently inside that + // window. Instance-scoped (not a global static) so this doesn't + // interfere with β€” or get polluted by β€” unrelated tests that cargo + // runs concurrently on other threads. Without a correctly-scoped + // lock, a burst of overlapping `FlowRunFinished` events for the SAME + // flow_id would pile up inside the critical section together + // instead of queuing. + let hooks = Arc::new(CommitTestHooks::default()); + hooks + .delay_ms + .store(20, std::sync::atomic::Ordering::SeqCst); + + let sub = Arc::new(DedupCommitSubscriber::with_test_hooks( + config.clone(), + hooks.clone(), + )); + let mut handles = Vec::new(); + for i in 0..5 { + let sub = sub.clone(); + handles.push(tokio::spawn(async move { + sub.handle(&DomainEvent::FlowRunFinished { + flow_id: "f-race".into(), + run_id: format!("run-{i}"), + status: "completed".into(), + }) + .await; + })); + } + for handle in handles { + handle.await.unwrap(); + } + + assert_eq!( + hooks.concurrent.load(std::sync::atomic::Ordering::SeqCst), + 0, + "every critical-section entry must have a matching exit" + ); + assert_eq!( + hooks + .max_concurrent + .load(std::sync::atomic::Ordering::SeqCst), + 1, + "the per-flow lock must serialize overlapping FlowRunFinished handling for the \ + same flow_id β€” at most one commit critical section may be active at a time" + ); + } + + #[tokio::test] + async fn dedup_commit_serial_commits_for_the_same_flow_accumulate_via_union() { + let tmp = tempfile::TempDir::new().unwrap(); + let config = test_config(&tmp); + let flow = flow_with_dedup_node("f-serial", "dd"); + store::upsert_flow(&config, &flow).unwrap(); + + let namespace = dedup_state_namespace("f-serial"); + let sub = DedupCommitSubscriber::new(config.clone()); + + // Run A finishes, having tentatively seen "a". + store::kv_set(&config, &namespace, "dedup:dd:tentative", &json!(["a"])).unwrap(); + sub.handle(&DomainEvent::FlowRunFinished { + flow_id: "f-serial".into(), + run_id: "run-a".into(), + status: "completed".into(), + }) + .await; + + // Run B finishes later, having independently tentatively seen "b". + // The per-flow lock (proven by the concurrency test above) is what + // guarantees two overlapping runs' `FlowRunFinished` handling + // reduces to exactly this serialized order in practice β€” so this is + // the correctness property that mutual exclusion is protecting. + store::kv_set(&config, &namespace, "dedup:dd:tentative", &json!(["b"])).unwrap(); + sub.handle(&DomainEvent::FlowRunFinished { + flow_id: "f-serial".into(), + run_id: "run-b".into(), + status: "completed".into(), + }) + .await; + + let committed = store::kv_get(&config, &namespace, "dedup:dd:committed") + .unwrap() + .expect("committed key must exist after both runs settle"); + let mut committed: Vec<&str> = committed + .as_array() + .unwrap() + .iter() + .map(|v| v.as_str().unwrap()) + .collect(); + committed.sort_unstable(); + assert_eq!( + committed, + vec!["a", "b"], + "settling run B must not clobber run A's already-committed keys β€” committed is a \ + running union across every run that has settled, never a last-writer-wins overwrite" + ); + assert!( + store::kv_get(&config, &namespace, "dedup:dd:tentative") + .unwrap() + .is_none(), + "tentative must be cleared after each successful commit" + ); + } + + #[test] + fn flow_commit_lock_returns_the_same_arc_for_the_same_flow_id_and_differs_across_flows() { + let a1 = flow_commit_lock("f-lock-a"); + let a2 = flow_commit_lock("f-lock-a"); + assert!( + Arc::ptr_eq(&a1, &a2), + "the same flow_id must share one lock instance" + ); + + let b = flow_commit_lock("f-lock-b"); + assert!( + !Arc::ptr_eq(&a1, &b), + "different flow_ids must not contend on the same lock" + ); + } } diff --git a/vendor/tinyflows b/vendor/tinyflows index c81a4cd5a0..0ce4fbc60b 160000 --- a/vendor/tinyflows +++ b/vendor/tinyflows @@ -1 +1 @@ -Subproject commit c81a4cd5a0d3fc8b0ebd98dc3991e6ea381ee0c3 +Subproject commit 0ce4fbc60b0ebef06bdd6cc48b3e588107979d8d From 24d3c3fa683fb5bb8e62ca596aa7f7ffd41429b4 Mon Sep 17 00:00:00 2001 From: "cyrus@tinyhumans.ai" Date: Wed, 29 Jul 2026 16:31:39 +0530 Subject: [PATCH 3/6] chore(vendor): revert unrelated tinyhumans-sdk lockfile drift from dedup commit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cargo.lock / app/src-tauri/Cargo.lock picked up the pre-existing tinyhumans-sdk futures/tokio lock drift (from main's e213bc5fc) the last time `cargo check` ran locally without --locked. That drift is unrelated to the dedup per-flow commit lock and the tinyflows gitlink bump (docs-only, same tinyflows version) β€” revert both lockfiles back to their prior committed state so this branch's diff stays scoped to the dedup fix. --- Cargo.lock | 4 +--- app/src-tauri/Cargo.lock | 2 -- 2 files changed, 1 insertion(+), 5 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index df4abb5378..15b9c82afe 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7308,7 +7308,7 @@ dependencies = [ "jaq-core", "jaq-json", "jaq-std", - "reqwest", + "reqwest 0.12.28", "serde", "serde_json", "thiserror 2.0.18", @@ -7322,13 +7322,11 @@ name = "tinyhumans-sdk" version = "0.1.0" dependencies = [ "base64 0.22.1", - "futures", "percent-encoding", "reqwest", "serde", "serde_json", "thiserror 2.0.18", - "tokio", "url", ] diff --git a/app/src-tauri/Cargo.lock b/app/src-tauri/Cargo.lock index 593fe519e7..94ad8b9e79 100644 --- a/app/src-tauri/Cargo.lock +++ b/app/src-tauri/Cargo.lock @@ -9038,13 +9038,11 @@ name = "tinyhumans-sdk" version = "0.1.0" dependencies = [ "base64 0.22.1", - "futures", "percent-encoding", "reqwest 0.12.28", "serde", "serde_json", "thiserror 2.0.18", - "tokio", "url", ] From f06bbf46e41e0f3fd9f3a0a92b13389f8f1310e5 Mon Sep 17 00:00:00 2001 From: "cyrus@tinyhumans.ai" Date: Wed, 29 Jul 2026 16:33:27 +0530 Subject: [PATCH 4/6] fix(deps): re-apply tinyhumans-sdk lockfile regen (needed for --locked CI) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The prior revert (24d3c3fa6) removed the lock regen assuming it was avoidable drift, but `cargo metadata --locked` fails without it β€” main's e213bc5fc bumped the tinyhumans-sdk gitlink without regenerating either lockfile, so every PR off current main must carry the futures/tokio lock entries to build --locked. --- Cargo.lock | 4 +++- app/src-tauri/Cargo.lock | 2 ++ 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/Cargo.lock b/Cargo.lock index 15b9c82afe..df4abb5378 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7308,7 +7308,7 @@ dependencies = [ "jaq-core", "jaq-json", "jaq-std", - "reqwest 0.12.28", + "reqwest", "serde", "serde_json", "thiserror 2.0.18", @@ -7322,11 +7322,13 @@ name = "tinyhumans-sdk" version = "0.1.0" dependencies = [ "base64 0.22.1", + "futures", "percent-encoding", "reqwest", "serde", "serde_json", "thiserror 2.0.18", + "tokio", "url", ] diff --git a/app/src-tauri/Cargo.lock b/app/src-tauri/Cargo.lock index 94ad8b9e79..593fe519e7 100644 --- a/app/src-tauri/Cargo.lock +++ b/app/src-tauri/Cargo.lock @@ -9038,11 +9038,13 @@ name = "tinyhumans-sdk" version = "0.1.0" dependencies = [ "base64 0.22.1", + "futures", "percent-encoding", "reqwest 0.12.28", "serde", "serde_json", "thiserror 2.0.18", + "tokio", "url", ] From f2532976b56f4aa6a710fe5a923926c3a3de85ae Mon Sep 17 00:00:00 2001 From: "cyrus@tinyhumans.ai" Date: Wed, 29 Jul 2026 19:35:20 +0530 Subject: [PATCH 5/6] chore(flows): re-point tinyflows gitlink to squash-merge ead2615 (#25 merged) --- vendor/tinyflows | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vendor/tinyflows b/vendor/tinyflows index 0ce4fbc60b..ead2615f65 160000 --- a/vendor/tinyflows +++ b/vendor/tinyflows @@ -1 +1 @@ -Subproject commit 0ce4fbc60b0ebef06bdd6cc48b3e588107979d8d +Subproject commit ead2615f655a619d3dbe5c3d9d4f91f7574d2a09 From 03a9471fd31186f16db51d603388514dd42b6981 Mon Sep 17 00:00:00 2001 From: "cyrus@tinyhumans.ai" Date: Wed, 29 Jul 2026 22:40:06 +0530 Subject: [PATCH 6/6] fix(dedup): address #5265 review comments - privacy wording (prompt.md + about_app): a key expr can contain item PII, so only the resolved key is stored but authors should key off a non-sensitive id. - node_contracts: terminal-status note now covers 'unknown'/all non-success statuses releasing tentative. - tools.rs config_hint: bound the dedup key hint with truncate_hint (+ test). - bus.rs release(): drop the extra load_key_set read used only for a log count. - prompt.md: add language to the fenced workflow example (markdownlint MD040). - jsonrpc.rs + docs: document the deferred items (per-run tentative scoping, group-registration retryability, run-graph snapshot) as follow-ups, not reworked in this PR. --- src/core/jsonrpc.rs | 16 +++++++ src/openhuman/about_app/catalog_data.rs | 13 ++++-- .../flows/agents/workflow_builder/prompt.md | 21 +++++++--- src/openhuman/flows/bus.rs | 42 +++++++++++++++---- src/openhuman/flows/node_contracts.rs | 15 ++++--- src/openhuman/flows/tools.rs | 2 +- src/openhuman/flows/tools_tests.rs | 34 +++++++++++++++ 7 files changed, 119 insertions(+), 24 deletions(-) diff --git a/src/core/jsonrpc.rs b/src/core/jsonrpc.rs index e2a273eb24..cec370099c 100644 --- a/src/core/jsonrpc.rs +++ b/src/core/jsonrpc.rs @@ -1934,6 +1934,22 @@ fn register_domain_subscribers( // set of already-registered groups lets a later, wider DomainSet install // exactly the newly-enabled groups (and no group twice). `insert` returns // `true` only the first time a group is seen. + // + // **Known limitation (issue #5265, CodeRabbit "Major" on the dedup engine + // PR):** this marks a group "done" the moment its `if group_first_time(…)` + // block is entered, not once every `subscribe_global` call inside it + // actually returns `Some`. A transient `subscribe_global` failure (the + // global bus not yet initialized) inside one of those blocks β€” e.g. the + // Flows block's `FlowTriggerSubscriber` / `FlowRunDigestSubscriber` / + // `DedupCommitSubscriber` registrations β€” only logs a warning; the group + // is still marked done, so no later call ever retries it, leaving that + // subscriber permanently absent for the process's lifetime. This is a + // pre-existing pattern shared by every `group_first_time(DomainGroup::…)` + // call site in this function, not something introduced by (or specific + // to) the dedup subscriber β€” reworking it (e.g. marking the group done + // only after every registration in its block succeeds, or making + // individual registrations retryable) is out of scope for the dedup PR + // and is reported as a separate follow-up issue instead of fixed here. fn group_first_time(group: DomainGroup) -> bool { static DONE: OnceLock>> = OnceLock::new(); DONE.get_or_init(|| Mutex::new(HashSet::new())) diff --git a/src/openhuman/about_app/catalog_data.rs b/src/openhuman/about_app/catalog_data.rs index 8df78def0a..411c5e7ff0 100644 --- a/src/openhuman/about_app/catalog_data.rs +++ b/src/openhuman/about_app/catalog_data.rs @@ -1538,10 +1538,15 @@ pub(super) const CAPABILITIES: &[Capability] = &[ involved. It drops an item whose per-item key was already committed by a \ prior successful run, and otherwise passes it through. Committing happens \ automatically: keys the node passes through are marked done only once the \ - whole run finishes successfully; a failed/cancelled/interrupted run leaves \ - them unmarked so the same items retry next time. Only the opaque per-item \ - key strings are stored in the flow's own private, flow-scoped state β€” never \ - the item's content, and never the user's personal memory.", + whole run finishes successfully; a failed/cancelled/interrupted/unknown (or \ + any other non-success) run leaves them unmarked so the same items retry next \ + time. Only the resolved per-item key value is stored, locally, in the flow's \ + own private, flow-scoped state β€” never the item's full content, and never \ + the user's personal memory. The key is whatever the workflow author's \ + `config.key` expression resolves to, so it can carry item-derived data if \ + keyed off a sensitive field β€” author flows to key off an opaque, \ + non-sensitive stable id (an issue number, message id, url) rather than \ + personal data.", how_to: "Flows editor > add a `dedup` node right after the item source; set config.key \ to a stable per-item id expression, e.g. \"=item.id\".", status: CapabilityStatus::Beta, diff --git a/src/openhuman/flows/agents/workflow_builder/prompt.md b/src/openhuman/flows/agents/workflow_builder/prompt.md index 1a1c1d8942..1dd102df63 100644 --- a/src/openhuman/flows/agents/workflow_builder/prompt.md +++ b/src/openhuman/flows/agents/workflow_builder/prompt.md @@ -630,7 +630,7 @@ decision yourself. The correct pattern is ONE `dedup` node placed right after the items are produced and BEFORE the action that must run at most once per item: -``` +```text trigger β†’ fetch β†’ split_out β†’ dedup [key="=item.id"] β†’ …action… ``` @@ -640,6 +640,13 @@ trigger β†’ fetch β†’ split_out β†’ dedup [key="=item.id"] β†’ …action… derived from the action's own output. A key that resolves to null, missing, or an empty string fails OPEN: the item passes through and is not recorded (never silently dropped just because a key couldn't be computed). + **Privacy:** `config.key` is an arbitrary `=`-expression, so whatever it + resolves to IS what gets durably stored in the flow's own private state β€” + the resolved key value can contain item-derived data if you key off one + (e.g. `"=item.email"`). Only that resolved key is stored, never the item's + full content, but the key itself is not guaranteed to be non-sensitive β€” + key off an opaque, stable, non-sensitive id (an issue number, message id, + url) rather than a field that itself carries PII. - **Place it BEFORE the work, not after.** Unlike the `memory` node's `remember` (which you place AFTER the action), `dedup` goes first in the chain β€” it already handles "mark seen only after success" internally, so @@ -649,11 +656,13 @@ trigger β†’ fetch β†’ split_out β†’ dedup [key="=item.id"] β†’ …action… - **Commit is run-level.** A saved flow's dedup nodes are settled off the run's single terminal status: every dedup node that ran in a `completed`/`completed_with_warnings` run gets its newly-seen keys - committed; every dedup node in a `failed`/`cancelled`/`interrupted` run has - its newly-seen keys released so they retry next time. For a flow with one - action per run this is exactly what you want; a flow chaining several - independent actions after a single `dedup` should be aware that one - action's failure retries ALL of them next run, not just the failing one. + committed; every dedup node in any OTHER terminal status β€” `failed`, + `cancelled`, `interrupted`, `unknown`, or any status this host doesn't + recognize yet β€” has its newly-seen keys released so they retry next time. + For a flow with one action per run this is exactly what you want; a flow + chaining several independent actions after a single `dedup` should be + aware that one action's failure retries ALL of them next run, not just the + failing one. ### Expressions: the `=` / jq convention diff --git a/src/openhuman/flows/bus.rs b/src/openhuman/flows/bus.rs index e739998fa4..72d15da281 100644 --- a/src/openhuman/flows/bus.rs +++ b/src/openhuman/flows/bus.rs @@ -634,6 +634,32 @@ impl DedupCommitSubscriber { /// flow deleted between run-finish and this handler firing, or a /// transient store error, both degrade to "nothing to settle" rather than /// panicking the event bus. + /// + /// **Known limitation (issue #5265, Codex "P2" on the dedup engine PR):** + /// this reads the flow's CURRENT saved definition at settlement time, not + /// a snapshot of the graph the finishing run actually executed. Nothing + /// today persists a per-run graph/node-id snapshot β€” `prepare_flow_run` + /// loads `Flow` fresh into the spawned run's own task, and that copy is + /// discarded once the run starts; the `FlowRun` row has no `graph` field. + /// If a long-running flow is edited (or deleted) while a run is still in + /// flight: + /// - a `dedup` node the run wrote `tentative` keys under, then deleted or + /// renamed before `FlowRunFinished` fires, is no longer found here β€” its + /// tentative keys are neither committed nor released, so those items + /// silently retry on the flow's next run (safe-direction: at worst a + /// duplicate, never a lost item, matching this subsystem's existing + /// safe-failure posture β€” see the module doc's "Best-effort throughout" + /// paragraph); + /// - conversely a `dedup` node id newly added to the saved graph after the + /// run started is settled here even though the run never executed it + /// (a harmless no-op: it has no `tentative` keys to commit/release, see + /// `commit`/`release`'s early returns). + /// + /// Closing this properly means persisting a per-run graph/dedup-node-id + /// snapshot at run-start (`start_flow_run_row` or a sibling write) and + /// having this method read that snapshot instead of `store::get_flow` β€” + /// a schema + call-site change bigger than this PR's scope; reported as a + /// follow-up rather than attempted here. fn dedup_node_ids(&self, flow_id: &str) -> Vec { match store::get_flow(&self.config, flow_id) { Ok(Some(flow)) => flow @@ -735,17 +761,17 @@ impl DedupCommitSubscriber { /// Failure path: clear `tentative` only, leaving `committed` untouched so /// the released keys retry on the flow's next run. + /// + /// Deliberately does NOT `load_key_set` first to report a count: that + /// would be a full `kv_get` + JSON deserialize + `HashSet` build purely + /// for a log line, and `kv_delete` already silently no-ops on a missing + /// key, so there is no early-return to save either (Greptile, issue + /// #5265). fn release(&self, namespace: &str, node_id: &str, flow_id: &str, run_id: &str) { - let released = - load_key_set(&self.config, namespace, &dedup_node::tentative_key(node_id)).len(); - if released == 0 { - tracing::trace!(target: "flows", %flow_id, %run_id, node_id, "[dedup-commit] no tentative keys β€” nothing to release"); - return; - } match store::kv_delete(&self.config, namespace, &dedup_node::tentative_key(node_id)) { Ok(()) => tracing::debug!( - target: "flows", %flow_id, %run_id, node_id, released, - "[dedup-commit] released tentative keys β€” will retry next run" + target: "flows", %flow_id, %run_id, node_id, + "[dedup-commit] released tentative keys (if any) β€” will retry next run" ), Err(e) => tracing::warn!( target: "flows", %flow_id, %run_id, node_id, error = %e, diff --git a/src/openhuman/flows/node_contracts.rs b/src/openhuman/flows/node_contracts.rs index 8d72b42fe6..d74e9e0cb7 100644 --- a/src/openhuman/flows/node_contracts.rs +++ b/src/openhuman/flows/node_contracts.rs @@ -84,12 +84,13 @@ fn apply_host_overlay(contract: NodeKindContract) -> NodeKindContract { "dedup" => contract .with_note( "Commit is run-LEVEL, not node-level: the host settles every dedup node in the \ - flow off the run's single terminal FlowRunFinished status. \ + flow off the run's single terminal FlowRunFinished status. Only \ completed/completed_with_warnings unions this run's tentative keys into \ - committed for EVERY dedup node that ran; failed/cancelled/interrupted releases \ - tentative for ALL of them, untouched committed, so the whole run's items retry \ - next time β€” one node failing mid-run releases every dedup node's tentative in \ - that run, not just the failing one's.", + committed for EVERY dedup node that ran; EVERY other status β€” \ + failed/cancelled/interrupted, unknown, or any status this host doesn't \ + recognize yet β€” releases tentative for ALL of them (untouched committed), so \ + the whole run's items retry next time β€” one node failing mid-run releases every \ + dedup node's tentative in that run, not just the failing one's.", ) .with_note( "Canonical placement: split_out β†’ dedup [key=\"=item.id\"] β†’ …action…, i.e. one \ @@ -198,6 +199,10 @@ mod tests { assert!(notes.contains("FlowRunFinished"), "{notes}"); assert!(notes.contains("completed_with_warnings"), "{notes}"); assert!(notes.contains("failed/cancelled/interrupted"), "{notes}"); + // CodeRabbit (PR #5265): the release path is really "every status + // other than the two success strings" β€” `unknown` and any future + // status must be documented alongside the known failure statuses. + assert!(notes.contains("unknown"), "{notes}"); assert!(notes.contains("split_out β†’ dedup"), "{notes}"); } diff --git a/src/openhuman/flows/tools.rs b/src/openhuman/flows/tools.rs index f49fed48e9..9fe80ca90e 100644 --- a/src/openhuman/flows/tools.rs +++ b/src/openhuman/flows/tools.rs @@ -490,7 +490,7 @@ fn config_hint(node: &Node) -> Option { NodeKind::Dedup => cfg .get("key") .and_then(Value::as_str) - .map(|k| format!("key: {k}")), + .map(|k| truncate_hint(&format!("key: {k}"))), NodeKind::Merge | NodeKind::OutputParser | NodeKind::Trigger => None, } } diff --git a/src/openhuman/flows/tools_tests.rs b/src/openhuman/flows/tools_tests.rs index d2133a079c..98bfb289d8 100644 --- a/src/openhuman/flows/tools_tests.rs +++ b/src/openhuman/flows/tools_tests.rs @@ -178,6 +178,40 @@ async fn summary_step_count_and_kinds_are_correct() { assert_eq!(steps[1]["config_hint"], "slack.post_message"); } +#[test] +fn dedup_config_hint_is_truncated_for_a_long_key_expression() { + // CodeRabbit (PR #5265): unlike the other config_hint branches, the + // dedup branch returned `format!("key: {k}")` unwrapped by + // `truncate_hint`, so an oversized `config.key` expression could make + // the proposal/summary payload unbounded. + let long_key = format!("=item.{}", "x".repeat(200)); + let graph = WorkflowGraph { + nodes: vec![Node { + id: "dd".to_string(), + kind: NodeKind::Dedup, + type_version: 1, + name: "Dedup".to_string(), + config: json!({ "key": long_key }), + ports: Vec::new(), + position: None, + }], + ..Default::default() + }; + + let summary = build_summary(&graph); + let hint = summary["steps"][0]["config_hint"].as_str().unwrap(); + assert!( + hint.chars().count() <= MAX_CONFIG_HINT_CHARS, + "hint not truncated: {} chars: {hint}", + hint.chars().count() + ); + assert!(hint.ends_with('…'), "expected an ellipsis marker: {hint}"); + assert!( + hint.starts_with("key: "), + "expected the key: prefix: {hint}" + ); +} + #[tokio::test] async fn summary_trigger_describes_schedule() { let tmp = TempDir::new().unwrap();