feat(flows): host DedupCommitSubscriber for the exactly-once dedup node - #5265
Conversation
Implements the host side of the tinyflows `dedup` node (PR2, issue tinyhumansai#5263): `DedupCommitSubscriber` in `flows::bus`, listening for `FlowRunFinished` and settling every `dedup` node in the finished flow's graph via the same per-flow `StateStore` namespace (`"flow:<id>"`) the engine's `dedup` node itself reads/writes through. - On success (`completed`/`completed_with_warnings`): unions each dedup node's `tentative` key set into `committed`, then clears `tentative`. - On failure/cancel/interrupt (or any unrecognized status — fail closed toward "not committed"): clears `tentative` only, leaving `committed` untouched so the released items retry next run. - Node ids are derived from the flow's saved graph (`kind == "dedup"`) since `StateStore` has no prefix-scan. Registered alongside `FlowTriggerSubscriber` / `FlowRunDigestSubscriber` in the same `DomainGroup::Flows` startup block. - `flows::store::kv_delete` added next to `kv_get`/`kv_set` to clear `tentative` (rather than writing `[]`). - `node_contracts.rs` gains the `dedup` kind's host overlay (run-level commit semantics, canonical split_out → dedup → action placement); the memory overlay's dedup deferral note now redirects to the dedup node instead. The 13->14 kind-count drift tests (node_contracts, propose_workflow description, list_node_kinds) are updated in lockstep. - workflow_builder prompt.md teaches the `dedup` node as THE way to do "process each item once", replacing the tinyhumansai#5227 "deferred to a dedicated primitive" notes; the guard test asserting that deferral is inverted to assert the dedup node is taught instead. - `about_app` catalog gains a Dedup Node (Flows) capability entry. - vendor/tinyflows gitlink bumped to c81a4cd (dedup node kind + filter half, not modified here — see PR1). 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
📝 WalkthroughWalkthroughAdds a ChangesDedup workflow node
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant FlowRun as Flow run
participant Subscriber as DedupCommitSubscriber
participant Graph as Saved flow graph
participant State as flow_state
FlowRun->>Subscriber: FlowRunFinished
Subscriber->>Graph: Find dedup nodes
Subscriber->>State: Load tentative and committed keys
Subscriber->>State: Commit union on success or clear tentative keys
Possibly related issues
Possibly related PRs
Suggested labels: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
Comment |
…ows gitlink - Per-flow-id async keyed lock (FLOW_COMMIT_LOCKS) around the commit read-modify-write, so overlapping FlowRunFinished events for the same flow can't clobber each other's committed set (closes the durable half of the StateStore race; residual node-side tentative race documented + deferred). - Bump vendor/tinyflows gitlink to 0ce4fbc (adds the concurrency docs). - Regenerate Cargo.lock + app/src-tauri/Cargo.lock: adds the futures/tokio entries tinyhumans-sdk needs — a pre-existing --locked drift from main's e213bc5 (gitlink bumped without regenerating the lock), which blocks CI for any PR branched off current main.
…dup commit Cargo.lock / app/src-tauri/Cargo.lock picked up the pre-existing tinyhumans-sdk futures/tokio lock drift (from main's e213bc5) the last time `cargo check` ran locally without --locked. That drift is unrelated to the dedup per-flow commit lock and the tinyflows gitlink bump (docs-only, same tinyflows version) — revert both lockfiles back to their prior committed state so this branch's diff stays scoped to the dedup fix.
…d CI) The prior revert (24d3c3f) removed the lock regen assuming it was avoidable drift, but `cargo metadata --locked` fails without it — main's e213bc5 bumped the tinyhumans-sdk gitlink without regenerating either lockfile, so every PR off current main must carry the futures/tokio lock entries to build --locked.
coderabbitai
Bot
added
feature
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: f2532976b5
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| 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); |
There was a problem hiding this comment.
When two runs of the same flow overlap, this reads the node's shared tentative set and commits everything currently there for whichever run happens to finish successfully. A scheduled/app-event run can overlap with an explicit RPC/detached run, and the second run may have already written keys to the same dedup:<node>:tentative row before its downstream action finishes; if the first run then succeeds, those in-flight keys are marked committed even if the second run later fails or is cancelled, causing future runs to drop work that never completed. The host needs either per-run tentative buckets or a same-flow run exclusion around dedup flows before commit-on-success can be safe.
Useful? React with 👍 / 👎.
| match store::get_flow(&self.config, flow_id) { | ||
| Ok(Some(flow)) => flow | ||
| .graph | ||
| .nodes | ||
| .iter() |
There was a problem hiding this comment.
This loads the current saved flow definition at finish time, not the graph that the run actually executed. If a long-running flow is edited while it is in flight, deleting or renaming a dedup node before FlowRunFinished means the subscriber won't settle the tentative keys that the running graph wrote, so successful work is retried on the next run; conversely newly saved node ids can be settled for a run that never executed them. Persist or derive the dedup node ids from the prepared/run snapshot instead of the mutable flow definition.
Useful? React with 👍 / 👎.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
🧹 Nitpick comments (4)
src/openhuman/flows/bus.rs (4)
516-532: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick winNote the lock's process-local scope.
FLOW_COMMIT_LOCKSis a process-global static, so it serializes only within one core process; two processes sharing the sameflow_stateSQLite file (e.g. CLI plus sidecar, or an overlapping restart) can still interleave the read-modify-write. Worth stating that explicitly alongside the node-side limitation already documented here.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/openhuman/flows/bus.rs` around lines 516 - 532, Update the commit atomicity documentation around handle_finished and FLOW_COMMIT_LOCKS to explicitly state that the process-global lock serializes settlements only within a single core process. Document that separate processes sharing the same flow_state SQLite file can still interleave the read-modify-write, alongside the existing node-side limitation.
475-480: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚖️ Poor tradeoff
bus.rsis now well past the module-size guideline.With
DedupCommitSubscriberand its tests this file is ~1.7k lines hosting three unrelated subscribers. Consider splitting intobus/dedup_commit.rs(plus its own tests module) to keep single-responsibility modules.As per coding guidelines: "Prefer Rust modules of approximately 500 lines or fewer and maintain small, single-responsibility Unix-style modules."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/openhuman/flows/bus.rs` around lines 475 - 480, Split the DedupCommitSubscriber implementation and its associated tests out of bus.rs into a dedicated bus/dedup_commit.rs module. Update module declarations, imports, and visibility as needed so the subscriber’s behavior and tests remain unchanged, while bus.rs retains only the other subscriber responsibilities.Source: Coding guidelines
677-691: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valueBlocking SQLite work runs on the async executor while the per-flow lock is held.
commit/releasecall synchronous rusqlite throughstore::kv_*, so the whole critical section blocks the runtime thread. With many dedup nodes this can stall other tasks on the same worker. Consider wrapping the settlement loop intokio::task::spawn_blocking(orblock_in_place), keeping the lock acquisition async.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/openhuman/flows/bus.rs` around lines 677 - 691, The settlement loop under the per-flow lock must not execute synchronous SQLite work on the async executor. Keep lock acquisition and test-delay handling in the async flow, then move the `commit`/`release` loop for `namespace` and `node_ids` into `tokio::task::spawn_blocking` (or `block_in_place`), preserving the lock until settlement completes and handling any blocking-task failure appropriately.
697-734: 🩺 Stability & Availability | 🔵 Trivial
committedset grows without bound.Every successful run unions more keys into a single JSON array row that is fully read, deserialized, re-sorted, and rewritten on each settlement — O(n) per run with no eviction. For long-lived, high-volume flows this row will keep growing. Worth planning a bound (max size / TTL / per-key rows) before this ships widely.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/openhuman/flows/bus.rs` around lines 697 - 734, Update the deduplication flow around commit to bound retention of committed keys, using an explicit maximum size, TTL, or per-key storage strategy instead of continually unioning into one unbounded JSON array. Ensure load, persistence, and settlement behavior consistently evicts or expires old entries while preserving deduplication for retained keys; anchor the change in commit and the existing load_key_set/store_key_set flow.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/core/jsonrpc.rs`:
- Around line 2165-2184: Update the Flows registration logic surrounding
DedupCommitSubscriber::new and subscribe_global so a failed subscription does
not permanently mark DomainGroup::Flows as initialized. Record the group only
after subscribe_global returns Some and the handle is retained, or otherwise
ensure a None result remains retryable on later invocations while preserving the
existing warning.
In `@src/openhuman/flows/agents/workflow_builder/prompt.md`:
- Line 633: Update the fenced workflow example in prompt.md to include a
language identifier immediately after the opening fence, using text or the
repository’s supported DSL label so markdownlint MD040 passes.
- Around line 637-642: Update config.key guidance in
src/openhuman/flows/agents/workflow_builder/prompt.md lines 637-642 to state
that resolved keys may contain selected item-derived data, and recommend opaque
stable IDs rather than claiming they are never content. Update the corresponding
local-storage wording in src/openhuman/about_app/catalog_data.rs lines 1536-1544
to remove “never the item's content” and accurately describe that only the
resolved key is stored.
In `@src/openhuman/flows/node_contracts.rs`:
- Around line 86-92: Complete the terminal-status contract by documenting
unknown as a release status alongside failed, cancelled, and interrupted. Update
src/openhuman/flows/node_contracts.rs lines 86-92 and lines 195-202,
src/openhuman/flows/agents/workflow_builder/prompt.md lines 649-653, and
src/openhuman/about_app/catalog_data.rs lines 1539-1542; ensure the overlay
assertion explicitly verifies unknown is documented.
In `@src/openhuman/flows/tools.rs`:
- Around line 490-493: Update the NodeKind::Dedup branch to pass the formatted
“key: {k}” hint through truncate_hint, preserving the existing optional-value
behavior and MAX_CONFIG_HINT_CHARS bound. Add a regression test covering an
oversized config.key and asserting the returned dedup hint is truncated.
---
Nitpick comments:
In `@src/openhuman/flows/bus.rs`:
- Around line 516-532: Update the commit atomicity documentation around
handle_finished and FLOW_COMMIT_LOCKS to explicitly state that the
process-global lock serializes settlements only within a single core process.
Document that separate processes sharing the same flow_state SQLite file can
still interleave the read-modify-write, alongside the existing node-side
limitation.
- Around line 475-480: Split the DedupCommitSubscriber implementation and its
associated tests out of bus.rs into a dedicated bus/dedup_commit.rs module.
Update module declarations, imports, and visibility as needed so the
subscriber’s behavior and tests remain unchanged, while bus.rs retains only the
other subscriber responsibilities.
- Around line 677-691: The settlement loop under the per-flow lock must not
execute synchronous SQLite work on the async executor. Keep lock acquisition and
test-delay handling in the async flow, then move the `commit`/`release` loop for
`namespace` and `node_ids` into `tokio::task::spawn_blocking` (or
`block_in_place`), preserving the lock until settlement completes and handling
any blocking-task failure appropriately.
- Around line 697-734: Update the deduplication flow around commit to bound
retention of committed keys, using an explicit maximum size, TTL, or per-key
storage strategy instead of continually unioning into one unbounded JSON array.
Ensure load, persistence, and settlement behavior consistently evicts or expires
old entries while preserving deduplication for retained keys; anchor the change
in commit and the existing load_key_set/store_key_set flow.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: dca0db01-c015-42ed-9075-8158b8cf6a36
⛔ Files ignored due to path filters (2)
Cargo.lockis excluded by!**/*.lockapp/src-tauri/Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (10)
src/core/jsonrpc.rssrc/openhuman/about_app/catalog_data.rssrc/openhuman/flows/agents/workflow_builder/prompt.mdsrc/openhuman/flows/builder_tools.rssrc/openhuman/flows/builder_tools_tests.rssrc/openhuman/flows/bus.rssrc/openhuman/flows/node_contracts.rssrc/openhuman/flows/store.rssrc/openhuman/flows/tools.rsvendor/tinyflows
| // 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" | ||
| ); | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Make failed subscriber registration retryable.
group_first_time(DomainGroup::Flows) records the group before this registration. If subscribe_global returns None, this only logs a warning; later invocations skip the whole Flows block, leaving DedupCommitSubscriber permanently absent and tentative keys unsettled. Mark the group only after successful registration or keep failed registrations retryable.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/core/jsonrpc.rs` around lines 2165 - 2184, Update the Flows registration
logic surrounding DedupCommitSubscriber::new and subscribe_global so a failed
subscription does not permanently mark DomainGroup::Flows as initialized. Record
the group only after subscribe_global returns Some and the handle is retained,
or otherwise ensure a None result remains retryable on later invocations while
preserving the existing warning.
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| /// Failure path: clear `tentative` only, leaving `committed` untouched so | ||
| /// the released keys retry on the flow's next run. | ||
| fn release(&self, namespace: &str, node_id: &str, flow_id: &str, run_id: &str) { | ||
| let released = | ||
| load_key_set(&self.config, namespace, &dedup_node::tentative_key(node_id)).len(); | ||
| if released == 0 { | ||
| tracing::trace!(target: "flows", %flow_id, %run_id, node_id, "[dedup-commit] no tentative keys — nothing to release"); | ||
| return; | ||
| } | ||
| match store::kv_delete(&self.config, namespace, &dedup_node::tentative_key(node_id)) { | ||
| Ok(()) => tracing::debug!( | ||
| target: "flows", %flow_id, %run_id, node_id, released, | ||
| "[dedup-commit] released tentative keys — will retry next run" | ||
| ), | ||
| Err(e) => tracing::warn!( | ||
| target: "flows", %flow_id, %run_id, node_id, error = %e, | ||
| "[dedup-commit] failed to release tentative — those keys remain tentative until \ | ||
| a future successful commit reconciles them (harmless: committed stays untouched \ | ||
| either way, so no item is ever wrongly marked done)" | ||
| ), | ||
| } | ||
| } |
There was a problem hiding this comment.
release() purely for log count
load_key_set does a full kv_get + JSON deserialization + HashSet construction just to obtain released for the trace/debug messages and the == 0 early-return guard. kv_delete already silently no-ops on a missing key, so the early-return is only an optimization. Because kv_delete returns Result<()> (the affected-row count is discarded), the count can't be recovered from the delete itself — but the simplest fix is to just skip the early-return and drop the count. If exact counts in the success log are valuable, the alternative is to change kv_delete to return Result<usize> and use the affected-row count directly.
| /// Failure path: clear `tentative` only, leaving `committed` untouched so | |
| /// the released keys retry on the flow's next run. | |
| fn release(&self, namespace: &str, node_id: &str, flow_id: &str, run_id: &str) { | |
| let released = | |
| load_key_set(&self.config, namespace, &dedup_node::tentative_key(node_id)).len(); | |
| if released == 0 { | |
| tracing::trace!(target: "flows", %flow_id, %run_id, node_id, "[dedup-commit] no tentative keys — nothing to release"); | |
| return; | |
| } | |
| match store::kv_delete(&self.config, namespace, &dedup_node::tentative_key(node_id)) { | |
| Ok(()) => tracing::debug!( | |
| target: "flows", %flow_id, %run_id, node_id, released, | |
| "[dedup-commit] released tentative keys — will retry next run" | |
| ), | |
| Err(e) => tracing::warn!( | |
| target: "flows", %flow_id, %run_id, node_id, error = %e, | |
| "[dedup-commit] failed to release tentative — those keys remain tentative until \ | |
| a future successful commit reconciles them (harmless: committed stays untouched \ | |
| either way, so no item is ever wrongly marked done)" | |
| ), | |
| } | |
| } | |
| /// Failure path: clear `tentative` only, leaving `committed` untouched so | |
| /// the released keys retry on the flow's next run. | |
| fn release(&self, namespace: &str, node_id: &str, flow_id: &str, run_id: &str) { | |
| 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 — 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)" | |
| ), | |
| } | |
| } |
Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
|
| Filename | Overview |
|---|---|
| src/openhuman/flows/bus.rs | Core addition: DedupCommitSubscriber (~350 lines) + FLOW_COMMIT_LOCKS per-flow serialization. Logic, locking model, and error handling are sound; one minor double-load in release() for logging count. |
| src/openhuman/flows/store.rs | Adds kv_delete — correct SQL DELETE with no-op on missing key, proper anyhow context wrapping. |
| src/core/jsonrpc.rs | Registers DedupCommitSubscriber in the DomainGroup::Flows startup block, following the same pattern as the existing digest subscriber. |
| src/openhuman/flows/node_contracts.rs | Adds dedup overlay with run-level commit semantics and canonical placement notes; memory overlay redirected to dedup; 13→14 kind bump with updated tests. |
| src/openhuman/flows/tools.rs | Adds dedup to the ProposeWorkflowTool enum list and config_hint; count bump to 14 is consistent. |
| src/openhuman/flows/builder_tools.rs | Mechanical 13→14 count bumps in description strings for ListNodeKindsTool and GetNodeKindContractTool. |
| src/openhuman/about_app/catalog_data.rs | New automation.flow_dedup_node capability entry with accurate description and Beta status. |
| src/openhuman/flows/agents/workflow_builder/prompt.md | Teaches the builder agent the dedup node, its placement pattern, and removes the #5227-era deferral note. Instruction is clear and accurate. |
Sequence Diagram
sequenceDiagram
participant Engine as DedupNode (engine)
participant Store as flow_state KV
participant Sub as DedupCommitSubscriber
participant Bus as Event Bus
Note over Engine,Store: During flow run
Engine->>Store: "kv_get(dedup:id:committed)"
Store-->>Engine: committed set (filter items)
Engine->>Store: "kv_set(dedup:id:tentative, new keys)"
Note over Bus,Sub: Flow run finishes
Bus->>Sub: "FlowRunFinished {flow_id, status}"
Sub->>Sub: acquire FLOW_COMMIT_LOCKS[flow_id]
alt success (completed / completed_with_warnings)
Sub->>Store: "kv_get(dedup:id:tentative)"
Store-->>Sub: tentative set
Sub->>Store: "kv_get(dedup:id:committed)"
Store-->>Sub: committed set
Sub->>Store: "kv_set(dedup:id:committed, union)"
Sub->>Store: "kv_delete(dedup:id:tentative)"
else failure (failed / cancelled / interrupted)
Sub->>Store: "kv_delete(dedup:id:tentative)"
end
Sub->>Sub: release FLOW_COMMIT_LOCKS[flow_id]
Reviews (1): Last reviewed commit: "chore(flows): re-point tinyflows gitlink..." | Re-trigger Greptile
- privacy wording (prompt.md + about_app): a key expr can contain item PII, so only the resolved key is stored but authors should key off a non-sensitive id. - node_contracts: terminal-status note now covers 'unknown'/all non-success statuses releasing tentative. - tools.rs config_hint: bound the dedup key hint with truncate_hint (+ test). - bus.rs release(): drop the extra load_key_set read used only for a log count. - prompt.md: add language to the fenced workflow example (markdownlint MD040). - jsonrpc.rs + docs: document the deferred items (per-run tentative scoping, group-registration retryability, run-graph snapshot) as follow-ups, not reworked in this PR.
There was a problem hiding this comment.
graycyrus has reached the 50-credit limit for trial accounts. To continue receiving code reviews, upgrade your plan.
Review comments addressed — pushed in
|
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/openhuman/about_app/catalog_data.rs (1)
1536-1549: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winDo not promise exactly-once processing here.
Because non-success runs leave keys uncommitted, downstream effects can be replayed on retry. This is at-least-once execution with exactly-once commitment after successful completion; update the description to make that limitation explicit.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/openhuman/about_app/catalog_data.rs` around lines 1536 - 1549, The dedup node description currently overstates its guarantee by promising exactly-once processing. Update the description string in the catalog entry to describe at-least-once execution with exactly-once commitment after successful completion, explicitly noting that downstream effects may replay when non-successful runs retry; preserve the existing key storage and privacy details.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@src/openhuman/about_app/catalog_data.rs`:
- Around line 1536-1549: The dedup node description currently overstates its
guarantee by promising exactly-once processing. Update the description string in
the catalog entry to describe at-least-once execution with exactly-once
commitment after successful completion, explicitly noting that downstream
effects may replay when non-successful runs retry; preserve the existing key
storage and privacy details.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 392b7689-6e11-4236-9140-25008ed3bcbd
📒 Files selected for processing (7)
src/core/jsonrpc.rssrc/openhuman/about_app/catalog_data.rssrc/openhuman/flows/agents/workflow_builder/prompt.mdsrc/openhuman/flows/bus.rssrc/openhuman/flows/node_contracts.rssrc/openhuman/flows/tools.rssrc/openhuman/flows/tools_tests.rs
🚧 Files skipped from review as they are similar to previous changes (5)
- src/core/jsonrpc.rs
- src/openhuman/flows/agents/workflow_builder/prompt.md
- src/openhuman/flows/node_contracts.rs
- src/openhuman/flows/tools.rs
- src/openhuman/flows/bus.rs
There was a problem hiding this comment.
graycyrus has reached the 50-credit limit for trial accounts. To continue receiving code reviews, upgrade your plan.
|
Re-review after the
Deferred (kept the PR scoped, tracked):
The earlier red |
1 participant
Host half of the exactly-once
dedupnode (issue #5263). Depends on tinyhumansai/tinyflows#25 (engine — thededupfilter + StateStore contract); the submodule gitlink is bumped to that branch and must re-point to the squash-merge SHA before this goes green. Pairs with #5264 (frontend).What
DedupCommitSubscriber(flows/bus.rs) onFlowRunFinished: enumerates dedup nodes from the flow graph, then per node — success (completed/completed_with_warnings): uniontentative→committed, deletetentative; anything else (failed/cancelled/interrupted/unknown — fail-closed): deletetentativeonly. So a failed run never marks items done → they retry. Reuses the engine's exactflow:<id>namespace + importedcommitted_key/tentative_key(no drift). Registered in theDomainGroup::Flowsstartup block;flowsleaf-gate.flows::store::kv_delete— real DELETE onflow_state.deduphost overlay (run-level commit semantics + canonicalsplit_out → dedup → actionplacement). Thememoryoverlay's "deferred to a dedicated primitive" note now redirects to the dedup node.memorynode — in-graph memory read + flow-scoped write #5227-era deferral.automation.flow_dedup_nodeentry.Commit granularity
Run-level (the engine has no per-item success isolation). True exactly-once for single-action flows; at-least-once (safe: retry) for per-item side-effects. Documented.
Testing
dedup_commit_*inbus.rs(union-on-success,completed_with_warningssuccess, release-on-failure/cancel/interrupt, two independent nodes) + node_contracts (overlay_preserves_all_14_kinds, dedup overlay, memory→dedup redirect) +list_node_kinds_tool_returns_all_fourteen.cargo check/722 tests/fmt/clippy all clean (GGML_NATIVE=OFF).Summary by CodeRabbit
dedupworkflow node for exactly-once per-item processing viaconfig.key, committing on successful runs (including warnings) and leaving uncommitted for retries after failures/cancellation.dedup(14 node types total), including the newautomation.flow_dedup_nodecapability.