fix(flows): validate full schema on cache hit + make store init atomic

[mysma-9403]

Summary

• Follow-up to perf(cron): run store schema DDL once per process, not per query #5708 (cron) / perf(task-sources): run store schema DDL once per process, not per query #5709 (task-sources). flows::store is the store those two were derived from (R-m8), and it still carries the same two gaps CodeRabbit + Codex flagged and I fixed there: the cache-hit verify trusted a single table, and initialization was not atomic per path. This closes both here.
• Complete-schema validation. Replace the single-table sqlite_master probe with a PRAGMA user_version check: init_schema stamps FLOWS_SCHEMA_VERSION after a full, successful migration, and a cache hit is honoured only when the on-disk version matches — so a database replaced at runtime with an older/partial schema (missing a migrated column such as require_approval/graph_hash, or one of the other tables) is re-migrated instead of trusted and then failed with no such column.
• Atomic per-path init. Hold the INITIALIZED_SCHEMAS guard across the verify + init_schema + insert so two first callers for the same path can't both run the DDL.

Problem

flows::store::with_connection funnels every store op — including upsert_flow_run_step, called once per node per live run. Its schema-init gating (R-m8) was the template for the cron/task-sources gating, and it has the two defects review found in those copies:

1. The cache-hit verify probed only flow_definitions via sqlite_master. A database deleted at runtime yields a fresh empty file (no table → re-init, fine), but one replaced with an older/partial schema — flow_definitions present but missing require_approval/graph_hash, or missing flow_runs/flow_suggestions/flow_revisions — was trusted, and later SELECTs failed with no such column/no such table. The pre-gating code (full idempotent DDL every call) self-healed that; the single-table probe does not.
2. init_schema ran with the guard released, so two first callers for the same path could both run the DDL.

Solution

The same fix that landed on cron/task-sources, applied to the original:

• init_schema stamps PRAGMA user_version = FLOWS_SCHEMA_VERSION after the DDL + add_column_if_missing migrations succeed (persistent db-file setting, like the existing WAL pragma).
• ensure_schema_initialized holds the guard across the whole verify + init, and on a cache hit reads PRAGMA user_version; a hit is honoured only when it matches, otherwise it falls through to the idempotent init_schema (re-migrating a drifted or fresh db). On a hit the critical section is a single PRAGMA user_version read — cheaper than, and equivalent in locking to, the sqlite_master query it replaces, so the hot upsert_flow_run_step path is not slowed.
• The per-connection pragmas (busy_timeout, foreign_keys = ON) and the persistent journal_mode = WAL are unchanged.

New older_on_disk_schema_under_a_cached_path_is_remigrated test drops require_approval and resets user_version under a cached path, then asserts the next store op re-migrates rather than failing no such column. The existing schema_reinitializes_when_the_database_file_is_deleted_at_runtime test still pins the deleted-file case.

Incidental prerequisite (first commit)

The vendored tinyflows::observability::ExecutionStep gained a transcript field, but four in-repo #[cfg(test)] constructors used exhaustive struct literals, so the flows test tree doesn't compile on main (E0063 ×4) — the library is fine, so pushes to main (Rust test lanes don't run there) didn't catch it. The first commit appends ..Default::default() to each literal — exactly the migration path the crate's own ExecutionStep doc prescribes (it derives Default for this). It's a prerequisite: the new flows::store test can't compile until the flows test tree does. Kept as a separate commit so it's easy to see / cherry-pick.

Impact

• Runtime: core / CLI (behind the default flows feature). No wire, schema, or config change; no migration.
• Correctness: restores the pre-gating self-heal for column/table drift (not just a deleted db) and makes init atomic. No hot-path slowdown.
• Security/compat: none.

Known tradeoff: FLOWS_SCHEMA_VERSION must be bumped whenever a table/migration is added to init_schema (documented on the const). A future migration added without bumping it would silently skip drift-detection for older DBs. The alternative — enumerating expected columns on each hit — carries the same discipline with more surface, so user_version is the right call.

Submission Checklist

• Tests added or updated (happy path + at least one failure / edge case) — new older_on_disk_schema_under_a_cached_path_is_remigrated pins the column-drift branch; the existing deleted-file self-heal test and the full flows::store suite still route through with_connection.
• Diff coverage ≥ 80% — the new gating lines are covered by the new test plus the existing flows::store suite. Ran cargo test --lib flows::store locally (now that the test tree compiles again).
• N/A: behaviour-preserving hardening — no feature rows added, removed, or renamed in docs/TEST-COVERAGE-MATRIX.md.
• N/A: no matrix feature IDs are affected by this change.
• No new external network dependencies introduced.
• N/A: does not touch a release-cut surface in docs/RELEASE-MANUAL-SMOKE.md.
• N/A: no linked tracking issue — follow-up to perf(cron): run store schema DDL once per process, not per query #5708 / perf(task-sources): run store schema DDL once per process, not per query #5709, whose reviews (CodeRabbit + Codex) identified these two gaps in the shared pattern.

Related

• Follow-up to: perf(cron): run store schema DDL once per process, not per query #5708 (cron), perf(task-sources): run store schema DDL once per process, not per query #5709 (task-sources) — same two fixes, applied to the flows::store precedent they were derived from.
• Closes: N/A (no tracking issue)

---

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

Linear Issue

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

Commit & Branch

• Branch: perf/flows-store-schema-validation
• Commit SHA: f953881e1 (hardening); 79f3f8497 (test-tree unblock)

Validation Run

• N/A: no app/ frontend changes — pnpm --filter openhuman-app format:check not applicable.
• N/A: no TypeScript changes — pnpm typecheck not applicable.
• Focused tests: GGML_NATIVE=OFF cargo test --lib flows::store (default flows feature) — passes, incl. both self-heal tests.
• Rust fmt/check: cargo fmt (clean). flows is a default feature, so — unlike the cron/task-sources PRs — this compiles and tests under default features directly (the first commit is what makes the flows test tree compile again).
• N/A: no app/src-tauri shell changes — Tauri fmt/check not applicable.

Validation Blocked

• command: N/A
• error: N/A — the only blocker (the ExecutionStep test-tree break) is fixed by this PR's first commit.
• impact: N/A

Behavior Changes

• Intended behavior change: none observable in the happy path — a hardening of the runtime-replaced-db edge case + atomic init. The deleted/drifted-db self-heal is preserved (now for column drift too).
• User-visible effect: none.

[tinysweeper]

tinysweeper found nothing blocking. Approving.

$0.0000 · 0 in / 0 out · 401 embedded · openrouter/openai/text-embedding-3-small

[tinysweeper]

How this change flows

2 changed behaviours across 13 relationships. 6 surrounding behaviours are shown (60 graph nodes walked). 44 further behaviours left out to keep the diagram readable.

flowchart LR
  n0["observer_persists_each_step_incrementally<br/>changed"]:::changed
  n1["...n_the_database_file_is_deleted_at_runtime<br/>changed"]:::changed
  n2["format"]:::impacted
  n3["concurrent_step_upserts_do_not_lose_a_step"]:::impacted
  n4["with_connection"]:::impacted
  n5["create_get_list_delete_roundtrip"]:::impacted
  n6["get_flow"]:::impacted
  n7["create_flow"]:::impacted
  n0 -->|calls| n2
  n0 -->|tests| n2
  n1 -->|calls| n7
  n1 -->|tests| n7
  n3 -->|calls| n7
  n3 -->|tests| n7
  n4 -->|calls| n2
  n5 -->|calls| n6
  n5 -->|tests| n6
  n5 -->|calls| n7
  n5 -->|tests| n7
  n6 -->|calls| n2
  n6 -->|calls| n4
  classDef changed fill:#0d4429,stroke:#238636,color:#e6edf3
  classDef impacted fill:#161b22,stroke:#6e7681,color:#c9d1d9
  classDef flagged fill:#5a1e02,stroke:#d93f0b,color:#ffffff
  classDef blocking fill:#67060c,stroke:#f85149,color:#ffffff

Loading

_{Green: changed behaviour. Grey: surrounding behaviour. Arrows name the call, use, implementation, or test relationship. Orange: has findings. Red: has a finding that blocks the merge.}

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 66f3ef6c-0362-4c14-b551-e6f99b50cc9d

📥 Commits

Reviewing files that changed from the base of the PR and between 1f332dd and f953881.

📒 Files selected for processing (4)

• src/openhuman/flows/ops_tests.rs
• src/openhuman/flows/store.rs
• src/openhuman/flows/store_tests.rs
• src/openhuman/flows/tinyflows/observability.rs

Included review availability: Your plan provides up to 10 included reviews per hour; 9 remain after this review.

---

📝 Walkthrough

Walkthrough

The flow store now tracks and validates SQLite schema versions on cached database access. It reruns migrations for older schemas and stamps successful initialization. Tests simulate schema drift, and ExecutionStep fixtures now use default field values.

Changes

Flow schema migration

Layer / File(s)	Summary
Schema version validation and migration src/openhuman/flows/store.rs, src/openhuman/flows/store_tests.rs	The store checks PRAGMA user_version for cached databases, reruns migrations when the version is outdated, and writes the version after successful initialization. The regression test verifies column restoration, row preservation, and subsequent writes.

ExecutionStep test fixtures

Layer / File(s)	Summary
Defaulted ExecutionStep fixtures src/openhuman/flows/ops_tests.rs, src/openhuman/flows/tinyflows/observability.rs	Test fixtures use ..Default::default() for unspecified ExecutionStep fields.

Estimated code review effort: 3 (Moderate) | ~20 minutes

Merge Risk: ⚪ Minimal · up to f9538

This PR hardens flow-store schema recovery and initialization without changing normal user-visible behavior; no actionable merge-blocking risk remains after normal checks and review.

Suggested reviewers: senamakel, graycyrus

Poem

A rabbit checks the schema bright,
Finds old fields hiding out of sight.
Migrations hop, defaults appear,
Tests grow tidy, crisp, and clear.
“Fresh rows work!” the rabbit cheers.

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Docstring Coverage	✅ Passed	Docstring check was indeterminate for this PR — some files could not be analyzed in time. Not blocking.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title clearly summarizes the main changes: full schema validation on cache hits and atomic store initialization.

---

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.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

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

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: f953881e12

ℹ️ 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]

Validate schema objects in addition to the version stamp

When a cached database is partially damaged or replaced while retaining user_version = 1—for example, if flow_definitions is dropped from an otherwise current database—this returns success without checking any tables or columns, and the subsequent operation fails with no such table until restart. The previous sqlite_master probe repaired at least the missing-flow_definitions case, so using the version as the sole integrity check regresses that recovery path; retain structural checks for the required schema objects in addition to the migration version.

Useful? React with 👍 / 👎.

[M3gA-Mind]

Thanks for this, @mysma-9403 — closing because the code it guards was extracted to another crate.

src/openhuman/flows/store.rs is no longer a SQLite store; it's a 305-line binding that supplies a directory, with schema, SQL, migrations and concurrency all moved into tinyflows_sqlite. On upstream/main:

• git grep -E "fn with_connection|INITIALIZED_SCHEMAS|fn init_schema" -- src/openhuman/flows/ → no matches. Every function this modifies is gone.
• src/openhuman/flows/store_tests.rs, the file this adds 55 lines to, does not exist.

So the premise — an INITIALIZED_SCHEMAS cache in flows::store needing a schema-validation guard — no longer has a subject here. If the concern is still real, it belongs upstream in tinyflows_sqlite and would be welcome there.

Worth saying plainly: this was killed by a refactor, not by quality. Your #5708 and #5709 are good and are staying open.

[mysma-9403]

Follow-up push (2978f0c35): proactively simplified ensure_schema_initialized to a lock-free fast path, keeping this PR in lockstep with the sibling store-init-gate change on #5708.

The on-disk PRAGMA user_version is already authoritative (init_schema stamps it only after a full, successful migration), so it's now read lock-free first and the common already-initialized call returns without acquiring the process-global mutex at all. The mutex is taken only on a version mismatch (init/re-init), with a second user_version read under the lock (double-check). This:

• removes the global lock from the hot path (no cross-path serialization on the common case);
• preserves atomic-per-path init — two first callers racing on the same fresh path run the DDL exactly once, the loser observing the winner's stamp on the recheck;
• preserves migrated-column validation — a stale/partial on-disk schema carries a non-matching user_version and falls through to the idempotent init_schema.

INITIALIZED_SCHEMAS no longer gates the DDL (the on-disk version does); it's kept purely as a diagnostic marker so the "deleted/replaced at runtime" warn! fires only for a path this process already initialized, not on every fresh-boot init (user_version 0). Both existing regression tests still pass.

This mirrors the exact change CodeRabbit verified on #5708 ("the original hot-path serialization finding is addressed; a path-specific lock registry is not required" — #5708 (comment)). Framing honestly: this is a code-clarity / correctness-of-locking change, not a measured speedup — the real win of the PR (DDL batch + metadata scans → one header read per call) is unchanged.

[mysma-9403]

You're completely right, @M3gA-Mind — thanks for the clear write-up, and apologies for the noisy follow-up push above: that came from a sync automation that pushed the lock-free variant to all three sibling branches without first re-checking this PR's state or that flows::store still existed on upstream/main. It doesn't — the store moved into tinyflows_sqlite, so this diff targets code that's gone and 2978f0c35 can be ignored. Happy to leave this closed. If the same cache-hit schema-validation / lock-free-init concern turns out to apply to tinyflows_sqlite's own store, I'll raise it upstream there instead. Appreciated the note on #5708/#5709.