feat(flows): host DedupCommitSubscriber for the exactly-once dedup node (#5265)

Co-authored-by: Claude <noreply@anthropic.com>

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<Mutex<HashSet<DomainGroup>>> = 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");

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",

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

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."
             ))),
         }

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()));
 }
 

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:<node_id>:*` keys exist for a flow is to derive `<node_id>` 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:<flow_id>"`, 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<Config>,
+    /// Test-only instrumentation — see [`CommitTestHooks`]. Always `None` in
+    /// production (`DedupCommitSubscriber::new`).
+    #[cfg(test)]
+    test_hooks: Option<Arc<CommitTestHooks>>,
+}
+
+/// 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<Mutex<HashMap<K, Arc<tokio::sync::Mutex<()>>>>>` 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<Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>> =
+    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<tokio::sync::Mutex<()>> {
+    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<Config>) -> 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<Config>, hooks: Arc<CommitTestHooks>) -> 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<String> {
+        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<String> {
+    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<String>,
+) -> anyhow::Result<()> {
+    let mut keys: Vec<String> = 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"
+        );
+    }
 }

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<NodeKindContract> {
     tinyflows::catalog::all_contracts()
@@ -96,7 +113,7 @@ pub fn all_node_kind_contracts() -> Vec<NodeKindContract> {
 }
 
 /// 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<NodeKindContract> {
     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]

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, \

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<String> {
             };
             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,
     }
 }

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();