fix(flows): store resilience — skip corrupt rows, transactional step upserts, once-per-process schema init

[graycyrus]

Stacked on #5286 (fix/flows-resume-run-lifecycle). This branch contains that PR's commit as its base. Review only the second commit, and merge after #5286.

Summary

• One corrupt or newer-schema row could brick the entire flows surface. list_flows, all app_event trigger dispatch, and the boot schedule reconcile each hard-failed on the first bad row.
• Corrupt rows are now skipped and loudly counted, not silently omitted.
• Makes the per-node step upsert transactional, so parallel branches can no longer lose a step.
• Stops re-running the full DDL batch on every single store call.

Problem

R-M4 (major). map_flow_row propagates migrate/deserialize errors as row errors, and list_flows / list_enabled_flows did flows.push(row?) — so the first bad row failed the whole query.

The realistic trigger is a downgrade: a user runs a newer build that persists a graph at a newer schema_version, then goes back. tinyflows::migrate::migrate cannot downgrade, so that single row breaks:

• every flows_list (the Workflows page),
• every list_enabled_flows — which drives all app_event trigger dispatch in bus.rs,
• and reconcile_schedule_triggers_on_boot.

The whole flows surface goes down because of one row. The sibling draft_store::list_drafts already skips-and-logs corrupt entries; the SQLite store had no such tolerance.

R-m1. upsert_flow_run_step was an untransacted read-modify-write (SELECT steps_json → mutate in memory → UPDATE) on a fresh connection. Two observer callbacks for parallel branch nodes could interleave — both read [A], one writes [A,B], the other [A,C] — and B vanished from the live view. Its status/duration_ms were lost permanently, since the post-hoc settle_steps reconstruction refills a node only with status: None.

R-m8. with_connection opened a new connection and re-ran the full DDL batch (5 CREATE TABLE + 6 CREATE INDEX + PRAGMA journal_mode=WAL + a PRAGMA table_info migration probe) on every call — including the per-step upsert fired for every node of every live run. Idempotent, but real churn on hot runs.

Solution

• Skip, but loudly. A new list_flow_rows helper decodes rows one at a time and skips/logs any that fail to parse or migrate (logging the id and error, never graph_json). list_flows/list_enabled_flows return (Vec<Flow>, usize), and the skip count is surfaced rather than swallowed: flows_list puts an "N workflows could not be loaded" line in RpcOutcome.logs, and the boot reconcile + bus.rs app-event path each warn!. A silently short flow list would be a worse failure mode than a hard error, so the count is the point.
• Transactional upsert via a new with_immediate_transaction helper (BEGIN IMMEDIATE / COMMIT / ROLLBACK through execute_batch, since Connection::transaction() needs &mut Connection and does not fit with_connection's &Connection closure). The existing busy_timeout = 5000 pragma covers the lock wait.
• Schema init once per process, per database path, gated by OnceLock<Mutex<HashSet<PathBuf>>>. Keyed by path rather than a single global flag, so each test's distinct TempDir workspace still initializes correctly. Per-connection pragmas (busy_timeout, foreign_keys) still reapply on every open, since those are not persisted in the db file.

A #[cfg(test)] force_corrupt_graph_json_for_test fixture door (mirroring the existing force_run_status_for_test) stages corrupt/newer-schema rows for the tests.

Submission Checklist

• Tests added or updated (happy path + at least one failure / edge case) per Testing Strategy
• Diff coverage ≥ 80% — 9 new tests: corrupt-row skip + count on both list functions, corrupt-disabled-row not counted, two concurrent-upsert races (different and identical node ids), fresh-db schema init, independent init across two db paths, and boot reconcile surviving a corrupt row. cargo test --lib openhuman::flows = 563 passed, 0 failed
• Coverage matrix updated — N/A: resilience fix to existing paths, no feature rows added/removed/renamed
• All affected feature IDs from the matrix are listed under ## Related — N/A: no matrix feature rows affected
• No new external network dependencies introduced
• Manual smoke checklist updated if this touches release-cut surfaces — N/A: no release-cut surface behaviour change on the healthy path
• Linked issue closed via Closes #NNN — N/A: found by code review, no tracking issue filed yet

Impact

• Runtime/platform: Rust core only.
• Behaviour: a corrupt/newer-schema row no longer takes down listing, trigger dispatch, or boot reconcile. It is skipped, and the count is reported so the UI can say "N workflows could not be loaded" rather than quietly showing a short list.
• API shape: list_flows / list_enabled_flows now return (Vec<Flow>, usize). All in-repo callers are updated.
• Performance: removes a full DDL batch + migration probe from every store call, including the per-node step write on every live run.
• Concurrency: BEGIN IMMEDIATE introduces a short write lock on step upserts; the transaction body is kept minimal and the existing busy timeout covers it.

Related

• Closes: N/A
• Follow-up PR(s)/TODOs: surface the skipped-row count in the Workflows UI (the backend now provides it); consider an explicit "this workflow was saved by a newer version" state rather than a generic skip.
• Merge order: stacked on fix(flows): give flows_resume the run-lifecycle safety flows_run already had #5286 — merge that first. Also touches store.rs, which docs(flows): fix contract-drift comments and repair the workflow-builder prompt structure #5290 edits; rebase if that lands first.

---

AI Authored PR Metadata (required for Codex/Linear PRs)

Linear Issue

• Key: N/A
• URL: N/A

Commit & Branch

• Branch: fix/flows-store-resilience
• Commit SHA: 4618fdf96 (plus fix(flows): give flows_resume the run-lifecycle safety flows_run already had #5286's 0b7105fa7 as its base)

Validation Run

• pnpm --filter openhuman-app format:check — N/A, no frontend files changed
• pnpm typecheck — N/A, no TypeScript changed
• Focused tests: GGML_NATIVE=OFF cargo test --lib openhuman::flows → 563 passed, 0 failed
• Rust fmt/check (if changed): GGML_NATIVE=OFF cargo check --manifest-path Cargo.toml clean
• Tauri fmt/check (if changed): N/A, app/src-tauri untouched

Validation Blocked

• command: N/A
• error: N/A
• impact: N/A

Behavior Changes

• Intended behavior change: an undecodable flow row is skipped and counted instead of failing the whole query; step persistence is atomic; schema DDL runs once per process per database.
• User-visible effect: the Workflows page, app-event triggers, and boot reconcile keep working when one row cannot be decoded, and the user is told how many were skipped.

Parity Contract

• Legacy behavior preserved: on a healthy database every list returns exactly the same flows in the same order, with a skip count of 0; the retention prune, WAL/busy-timeout pragmas, and add_column_if_missing migrations are unchanged.
• Guard/fallback/dispatch parity checks: list_enabled_flows still gates app-event dispatch identically for decodable rows; only the undecodable-row path differs.

Duplicate / Superseded PR Handling

• Duplicate PR(s): none
• Canonical PR: this one
• Resolution: N/A

[coderabbitai]

Warning

Review limit reached

You’ve reached a temporary PR review limit under our Fair Usage Limits Policy.

Your recent review volume is higher than typical usage, so adaptive limits are currently applied.

Next review available in: 49 minutes

Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available.
You're only billed for reviews past your plan's rate limits ($0.25/file).

How can I continue?

After more reviews become available, a review can be triggered using the @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews.

How do review limits work?

CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability.

For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window.

Please refer docs for additional details.

Review details

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 4aed0724-3c84-4171-959b-8a1e9744a67c

📥 Commits

Reviewing files that changed from the base of the PR and between 6a21f3f and 4394e0b.

📒 Files selected for processing (5)

• src/openhuman/flows/bus.rs
• src/openhuman/flows/ops.rs
• src/openhuman/flows/ops_tests.rs
• src/openhuman/flows/store.rs
• src/openhuman/flows/store_tests.rs

---

_{Comment @coderabbitai help to get the list of available commands.}

[graycyrus]

Review follow-up: fixed a regression this PR introduced.

Gating the schema DDL behind a per-path "already initialized" set cost the store its self-healing. Previously the DDL ran on every with_connection call, so a database deleted or replaced at runtime (workspace reset, manual deletion, a restore) recovered on the very next call — Connection::open creates a fresh empty file and CREATE TABLE IF NOT EXISTS repopulated it. With the cache, the set still reported "initialized" while the file behind it was empty, and every subsequent query failed no such table: flow_definitions until the process restarted.

Reproduced directly: after deleting flows.db mid-process, list_flows returned Err(no such table: flow_definitions).

Fix: a cache hit is now confirmed against the file on disk via one indexed sqlite_master lookup before it is honoured, re-running init (with a warning) when the tables are gone. That keeps the R-m8 win — the ~11-statement DDL batch no longer runs per call — while restoring self-healing at the cost of a single cheap query. Pinned by schema_reinitializes_when_the_database_file_is_deleted_at_runtime, which fails without the fix.

564 flows tests pass, cargo fmt --check clean.

[greptile-apps]

graycyrus has reached the 50-credit limit for trial accounts. To continue receiving code reviews, upgrade your plan.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 4394e0b5a7

ℹ️ 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".

[chatgpt-codex-connector]

Re-run migrations before trusting cached schemas

When this process has already cached a workspace path and the database file is later restored/replaced with an older or partial flows.db, this early return only verifies that flow_definitions exists. It skips init_schema, so missing migrated columns like require_approval (or other tables such as flow_runs/flow_state) are not recreated; the next store call can then fail with no such column: require_approval or no such table until the core restarts. The cache-hit probe should verify the required schema/migrations, or re-run the idempotent migration batch when anything is missing, not just when the primary table is absent.

Useful? React with 👍 / 👎.