Ci/fmt probe - #5415
Conversation
Moves the `mod tests` block out of `flows/bus.rs` into a sibling
`bus_tests.rs`, using the `#[cfg(test)] #[path = "..."] mod tests;`
convention already used throughout this crate (`subconscious/factory.rs`,
`skills/ops.rs`, `config/schema/channels.rs`, `platform/about_app/catalog.rs`,
and ~170 other sites).
Pure move: the test bodies are unchanged apart from being dedented one
level now that they are no longer nested inside an inline `mod tests { .. }`.
Also adds the tests for the tinyhumansai#5268 item 2 fix (settle dedup state from the
graph the run actually executed).
Closes tinyhumansai#5268 (item 2). `DedupCommitSubscriber` settled a finished run's `dedup` nodes by reading the flow's CURRENT saved graph. A flow edited while a run was still in flight was therefore mis-settled: a `dedup` node the run had written `tentative` keys under, but which was deleted or renamed before `FlowRunFinished` fired, was no longer found — so those keys were neither committed nor released, and the items silently reprocessed on the flow's next run. The converse also held: a node added after the run started was settled by a run that never executed it. `DedupCommitSubscriber` now also handles `DomainEvent::FlowRunStarted` (already published by `flows::ops::flows_run{,_detached}` right after the `flow_runs` row insert, and already tagged "cron" so the existing domain filter admits it) and pins that run's dedup node ids into the existing per-flow `flow_state` KV table under `dedup::run_nodes:<run_id>`. Settlement prefers that snapshot and clears it once the run is settled. - No schema migration: reuses the `flow_state` table the dedup keys already live in. The `dedup::` double-colon prefix cannot collide with the engine's `dedup:<node_id>:<committed|tentative>` keys. - Zero added I/O for the common case: a graph with no `dedup` node writes no snapshot at all, so an absent key stays an unambiguous "no snapshot". - Falls back to the saved graph when no snapshot exists (runs in flight across the upgrade, resumed runs, or a failed snapshot write), so the degraded path is exactly the historical behaviour. Also moves the test module to `bus_tests.rs` per the crate-wide `#[cfg(test)] #[path = "..."] mod tests;` convention.
CI's `cargo fmt --all -- --check` gate rejected the previous revision. Three formatting-sensitive constructs are reworked, each matched against a formatter-clean precedent already in the tree rather than guessed at: * `dedup_node_ids_in` broke a method chain as `flow.graph` / `.nodes` / …, merging the chain parent with its first field access. The same expression elsewhere in this file is written with the parent alone on its own line, so the receiver is now bound to a `nodes` local and the chain starts from it — the shape `flows/ops.rs` already uses at the same indent level. * `clear_run_snapshot` had an `if let Err(e) = store::kv_delete(...)` line sitting exactly on the 100-column limit. The key is hoisted into a local, which shortens the line and matches how `snapshotted_dedup_node_ids` immediately above already builds the same key. * `handle` had been rewritten from the original `if let` into a `match` with a nested `match` inside a block arm, which changed formatting that was previously known-good for no behavioural benefit. The `FlowRunFinished` block is restored verbatim and the new `FlowRunStarted` handling is a short `if let` above it, delegating to a new `snapshot_run_nodes` method that holds the graph load. No behaviour change: the same snapshot is written at run start, read at settlement, and cleared afterwards. Pure formatting and extraction. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
📝 WalkthroughWalkthroughThe change snapshots deduplication nodes when a flow run starts, uses the snapshot during completion settlement, falls back to the saved graph when needed, and removes the snapshot afterward. It also moves bus tests into ChangesRun-scoped deduplication settlement
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant FlowRunStarted
participant FlowStateStore
participant FlowRunFinished
participant DedupNodes
FlowRunStarted->>FlowStateStore: save dedup-node IDs by run ID
FlowRunFinished->>FlowStateStore: load snapshot or saved-graph fallback
FlowRunFinished->>DedupNodes: commit or release deduplication keys
FlowRunFinished->>FlowStateStore: delete run snapshot
Possibly related PRs
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches 💡 1🛠️ Fix failing CI checks 💡
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment Warning |
|
| Filename | Overview |
|---|---|
| src/openhuman/flows/bus.rs | Adds start-time dedup snapshots and finish-time cleanup, but the two previously reported graph-consistency paths remain reachable. |
| src/openhuman/flows/bus_tests.rs | Extracts the former inline test module and adds snapshot-focused coverage without changing production behavior itself. |
Sequence Diagram
sequenceDiagram
participant Runner as Flow runner
participant Bus as Event bus
participant Subscriber as Dedup subscriber
participant Store as Flow/KV store
Runner->>Store: Load flow selected for execution
Runner->>Bus: Publish FlowRunStarted
Runner->>Runner: Execute loaded flow
Bus-->>Subscriber: Deliver start event asynchronously
Subscriber->>Store: Reload current flow
Subscriber->>Store: Write run-node snapshot
Runner->>Bus: Publish FlowRunFinished
Bus-->>Subscriber: Deliver finish event
Subscriber->>Store: Read snapshot or current-flow fallback
Subscriber->>Store: Settle dedup state and clear snapshot
Reviews (2): Last reviewed commit: "ci: remove the temporary rustfmt probe w..." | Re-trigger Greptile
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
![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: aa3bfab37c
ℹ️ 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".
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
🧹 Nitpick comments (1)
src/openhuman/flows/bus.rs (1)
730-739: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winLog the malformed or empty snapshot branch.
snapshotted_dedup_node_idsreturnsNonesilently when the stored value is not an array, or when it holds no usable strings. The read-error branch above warns, so an operator can see it. The malformed branch produces no log line at all, so a corrupt snapshot is indistinguishable from "no snapshot" in the logs while settlement silently degrades to the saved graph.Add a branch diagnostic that keeps the same fallback behavior.
♻️ Proposed diagnostic for the malformed-snapshot branch
- let ids: Vec<String> = value - .as_array()? - .iter() - .filter_map(Value::as_str) - .map(str::to_string) - .collect(); - if ids.is_empty() { - return None; - } - Some(ids) + let ids: Vec<String> = match value.as_array() { + Some(items) => items + .iter() + .filter_map(Value::as_str) + .map(str::to_string) + .collect(), + None => Vec::new(), + }; + if ids.is_empty() { + tracing::warn!( + target: "flows", %namespace, %run_id, + "[dedup-commit] this run's dedup snapshot is not a non-empty string array — \ + falling back to the flow's saved graph" + ); + return None; + } + Some(ids)🤖 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 730 - 739, Update snapshotted_dedup_node_ids to log a warning when the stored value is not an array or when filtering yields no usable string IDs, while preserving its existing None fallback behavior. Keep valid non-empty snapshots returning Some(ids), and include enough context in the diagnostic to identify the malformed or empty snapshot.Source: Coding guidelines
🤖 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.
Nitpick comments:
In `@src/openhuman/flows/bus.rs`:
- Around line 730-739: Update snapshotted_dedup_node_ids to log a warning when
the stored value is not an array or when filtering yields no usable string IDs,
while preserving its existing None fallback behavior. Keep valid non-empty
snapshots returning Some(ids), and include enough context in the diagnostic to
identify the malformed or empty snapshot.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 3d89899e-250e-43c8-8670-cd4b8e6c22e7
📒 Files selected for processing (2)
src/openhuman/flows/bus.rssrc/openhuman/flows/bus_tests.rs
1 participant
Summary
Problem
Solution
Submission Checklist
diff-cover) meet the gate enforced by.github/workflows/ci-lite.yml. Runpnpm test:coverageandpnpm test:rustlocally; PRs below 80% on changed lines will not merge.docs/TEST-COVERAGE-MATRIX.mdreflect this change (orN/A: behaviour-only change)## Relateddocs/RELEASE-MANUAL-SMOKE.md)Closes #NNNin the## RelatedsectionImpact
Related
AI Authored PR Metadata (required for Codex/Linear PRs)
Linear Issue
Commit & Branch
Validation Run
pnpm --filter openhuman-app format:checkpnpm typecheckValidation Blocked
command:error:impact:Behavior Changes
Parity Contract
Duplicate / Superseded PR Handling
Summary by CodeRabbit
Bug Fixes
Tests