diff --git a/src/core/jsonrpc.rs b/src/core/jsonrpc.rs index 8aaf349f2..cec370099 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())) @@ -2162,6 +2178,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 427cd77bb..411c5e7ff 100644 --- a/src/openhuman/about_app/catalog_data.rs +++ b/src/openhuman/about_app/catalog_data.rs @@ -1528,6 +1528,30 @@ 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/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, + 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 50983b878..1dd102df6 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,54 @@ 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: + +```text +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). + **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 + 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 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 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 7f36fa3df..6a1c80142 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 f72ea1618..34b717ab0 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 ffb4b402d..72d15da28 100644 --- a/src/openhuman/flows/bus.rs +++ b/src/openhuman/flows/bus.rs @@ -17,9 +17,10 @@ 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 tinyflows::model::TriggerKind; +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; /// 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,381 @@ 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. +/// +/// **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, + #[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 + /// 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. + /// + /// **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 + .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" + ); + + // 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 { + self.commit(&namespace, &node_id, flow_id, run_id); + } else { + 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 + /// 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. + /// + /// 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) { + match store::kv_delete(&self.config, namespace, &dedup_node::tentative_key(node_id)) { + Ok(()) => tracing::debug!( + 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, + "[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 +908,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 +1343,410 @@ 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"]) + ); + } + + // ── 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/src/openhuman/flows/node_contracts.rs b/src/openhuman/flows/node_contracts.rs index 520bc35ae..d74e9e0cb 100644 --- a/src/openhuman/flows/node_contracts.rs +++ b/src/openhuman/flows/node_contracts.rs @@ -77,16 +77,33 @@ 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. Only \ + completed/completed_with_warnings unions this run's tentative keys into \ + 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 \ + 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 +113,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 +163,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 +172,38 @@ 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}"); + // 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}"); } #[test] diff --git a/src/openhuman/flows/store.rs b/src/openhuman/flows/store.rs index 0927a17b6..0712aaac2 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 685eb9dce..9fe80ca90 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| 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 d2133a079..98bfb289d 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(); diff --git a/vendor/tinyflows b/vendor/tinyflows index ff9950186..ead2615f6 160000 --- a/vendor/tinyflows +++ b/vendor/tinyflows @@ -1 +1 @@ -Subproject commit ff9950186ac66775a554c80aa420920b4f39e4de +Subproject commit ead2615f655a619d3dbe5c3d9d4f91f7574d2a09