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

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,15 +607,63 @@ 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
 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()));
 }