fix(flows): close stale-read race in flows_update disarm + actionable-node reachability

src/openhuman/flows/ops.rs

@@ -367,21 +367,46 @@ pub(crate) fn graph_has_outbound_side_effect(graph: &WorkflowGraph) -> bool {
 }
 
 /// Whether `graph` has anything for [`flows_run`] to actually *do* — i.e. at
-/// least one non-`trigger` node, wired up by at least one edge. A graph made
-/// of nothing but a bare `trigger` node (or a `trigger` plus unreachable/
-/// disconnected nodes with no edges at all) can compile and "run" cleanly
-/// while producing no work whatsoever — the exact live finding this guards:
-/// a trigger-only flow reported `status="completed" pending_approvals=0`
-/// having done nothing, which reads as a successful automation to anyone not
-/// staring at the node count. Used by `flows_run` to attach a
-/// human-readable note to an otherwise-silent "success".
+/// least one non-`trigger` node **reachable from the trigger** by following
+/// directed edges. A graph made of nothing but a bare `trigger` node (or a
+/// `trigger` plus unreachable/disconnected nodes — even ones wired to each
+/// other by their own edges, just not to the trigger) can compile and "run"
+/// cleanly while producing no work whatsoever — the exact live finding this
+/// guards: a trigger-only flow reported `status="completed"
+/// pending_approvals=0` having done nothing, which reads as a successful
+/// automation to anyone not staring at the node count. Used by `flows_run`
+/// to attach a human-readable note to an otherwise-silent "success".
+///
+/// Deliberately a reachability walk rather than "any edge at all exists":
+/// `nodes.len() > 1 && !edges.is_empty()` would count a disconnected
+/// component's internal edges as actionable even though nothing downstream
+/// of the trigger ever runs.
 pub(crate) fn graph_has_actionable_nodes(graph: &WorkflowGraph) -> bool {
-    let non_trigger_nodes = graph
-        .nodes
-        .iter()
-        .filter(|n| n.kind != NodeKind::Trigger)
-        .count();
-    non_trigger_nodes > 0 && !graph.edges.is_empty()
+    let Some(trigger) = graph.trigger() else {
+        // No single resolvable trigger to walk from — fall back to the
+        // coarse "any non-trigger node wired up by an edge" check so a
+        // malformed/ambiguous-trigger graph doesn't spuriously suppress the
+        // empty-flow note.
+        return graph.nodes.iter().any(|n| n.kind != NodeKind::Trigger) && !graph.edges.is_empty();
+    };
+
+    let mut visited: std::collections::HashSet<&str> = std::collections::HashSet::new();
+    let mut stack = vec![trigger.id.as_str()];
+    while let Some(current) = stack.pop() {
+        if !visited.insert(current) {
+            continue;
+        }
+        for next in graph.successors(current) {
+            if !visited.contains(next) {
+                stack.push(next);
+            }
+        }
+    }
+
+    visited
+        .into_iter()
+        .filter_map(|id| graph.node(id))
+        .any(|n| n.kind != NodeKind::Trigger)
 }
 
 /// Produces host-side, **non-fatal** validation warnings for a graph — today
@@ -2491,15 +2516,29 @@ fn map_flow_update_error(e: store::FlowUpdateError) -> String {
 /// tool, the canvas Save button, a proposal apply, or any other
 /// `flows_update` caller — and go LIVE immediately with no user review
 /// (confirmed live: a flow started firing on an unreviewed 8am schedule).
-/// So: when the *new* graph's trigger is automatic, the flow is *currently*
-/// enabled, and the *previous* graph's trigger was NOT automatic (a
-/// manual/none → automatic transition), this forces the persisted `enabled`
-/// back to `false` in the same store write — the user must explicitly
-/// re-arm via `flows_set_enabled` after reviewing the new trigger.
-/// Deliberately narrower than Rule 1's at-create version: a flow that was
-/// already an enabled *automatic*-trigger flow being legitimately re-edited
-/// (e.g. tweaking a cron expression) is left alone — the user already opted
-/// in once, and re-disarming on every edit would just be friction.
+/// So: when the *new* graph's trigger is automatic and the *previous*
+/// graph's trigger was NOT automatic (a manual/none → automatic
+/// transition), this forces the persisted `enabled` back to `false` in the
+/// same store write — the user must explicitly re-arm via
+/// `flows_set_enabled` after reviewing the new trigger. An automatic →
+/// automatic re-edit (e.g. tweaking a cron expression) is left alone — the
+/// user already opted in once, and re-disarming on every edit would just be
+/// friction.
+///
+/// The override is applied **unconditionally** on a manual/none → automatic
+/// transition — it does *not* gate on whether the flow *looked* enabled in
+/// the `existing` read above. That read is a snapshot taken before
+/// `store::update_flow_graph`'s own guarded UPDATE re-reads the row; a
+/// concurrent `flows_set_enabled(id, true)` landing in the gap would leave
+/// this snapshot stale while the row is actually enabled by the time the
+/// guarded UPDATE runs — and since `set_enabled` bumps `updated_at` too,
+/// such a race wouldn't even trip the optimistic-concurrency conflict, it
+/// would just silently persist the automatic graph as enabled (the exact
+/// bug this rule exists to close). Gating on the stale `existing.enabled`
+/// re-opens that race; forcing the override on every transition, enabled-or-
+/// not, is exactly as safe as Rule 1's at-create version — a transition on
+/// an already-disabled flow is just a no-op write of `enabled=false` over
+/// `enabled=false`.
 pub async fn flows_update(
     config: &Config,
     id: &str,
@@ -2523,19 +2562,26 @@ pub async fn flows_update(
         }
     };
 
-    // B29 Rule 1 analogue: only disarm a manual/none → automatic transition
-    // on an already-enabled flow. An automatic → automatic re-edit, or a
-    // flow that isn't enabled to begin with, is untouched.
+    // B29 Rule 1 analogue: disarm every manual/none → automatic trigger
+    // transition, unconditionally — see the doc comment above for why this
+    // must NOT gate on the (possibly stale) `existing.enabled` read.
     let was_auto = trigger_is_automatic(&existing.graph);
     let now_auto = trigger_is_automatic(&graph);
-    let should_disarm = now_auto && existing.enabled && !was_auto;
-    let enabled_override = should_disarm.then_some(false);
+    let is_manual_to_auto_transition = now_auto && !was_auto;
+    let enabled_override = is_manual_to_auto_transition.then_some(false);
+    // Best-effort flag for the info log / result message below: whether the
+    // flow *appeared* live going into this update. Not used for the
+    // override decision itself (that's unconditional, see above) — only to
+    // avoid telling the user "flow was auto-disabled" when it was already
+    // disabled going in.
+    let should_disarm = is_manual_to_auto_transition && existing.enabled;
     tracing::debug!(
         target: "flows",
         flow_id = %id,
         was_auto,
         now_auto,
         currently_enabled = existing.enabled,
+        is_manual_to_auto_transition,
         should_disarm,
         "[flows] flows_update: auto-trigger disarm decision inputs"
     );

src/openhuman/flows/ops_tests.rs

@@ -301,6 +301,46 @@ async fn flows_run_on_graph_with_actionable_nodes_has_no_empty_flow_note() {
     );
 }
 
+/// `graph_has_actionable_nodes` must walk from the trigger, not merely check
+/// "any non-trigger node plus any edge". A component with edges of its own,
+/// but no path back to the trigger, is unreachable and must still surface
+/// the "nothing to run" note — a naive count-based check would have missed
+/// this and wrongly suppressed the note.
+#[tokio::test]
+async fn flows_run_on_graph_with_disconnected_component_still_surfaces_empty_flow_note() {
+    let tmp = TempDir::new().unwrap();
+    let config = test_config(&tmp);
+
+    let graph = json!({
+        "name": "disconnected",
+        "nodes": [
+            { "id": "t", "kind": "trigger", "name": "Trigger" },
+            { "id": "a", "kind": "output_parser", "name": "Orphan A" },
+            { "id": "b", "kind": "output_parser", "name": "Orphan B" }
+        ],
+        "edges": [
+            // "a" -> "b" is wired up, but neither is reachable from "t" — the
+            // trigger has no outgoing edges at all.
+            { "from_node": "a", "to_node": "b" }
+        ]
+    });
+    let created = flows_create(&config, "disconnected".to_string(), graph, false)
+        .await
+        .unwrap();
+
+    let outcome = flows_run(&config, &created.value.id, json!({}), FlowRunTrigger::Rpc)
+        .await
+        .unwrap();
+
+    let note = outcome.value["note"]
+        .as_str()
+        .expect("a component disconnected from the trigger must still surface the empty-flow note");
+    assert!(
+        note.contains("no actionable nodes") || note.to_lowercase().contains("nothing"),
+        "note should explain that nothing ran, got: {note}"
+    );
+}
+
 #[tokio::test]
 async fn flows_run_reports_pending_approval_and_blocks_downstream() {
     let tmp = TempDir::new().unwrap();
@@ -697,6 +737,56 @@ async fn flows_update_disables_on_manual_to_automatic_trigger_transition_when_en
     );
 }
 
+/// Regression: the manual→automatic disarm must apply unconditionally, not
+/// only when `flows_update`'s own `existing` read observes `enabled: true`.
+/// A live race (Codex, this PR) could leave that read stale — a concurrent
+/// `flows_set_enabled(id, true)` landing between the read and the guarded
+/// write would previously compute `should_disarm = false` from the stale
+/// snapshot and let the automatic graph persist enabled. This test pins the
+/// non-racy half of that contract directly at the `flows_update` level: even
+/// starting from an *observed* `enabled: false`, a manual→automatic
+/// transition still writes the override (a no-op here since the flow was
+/// already disabled) rather than skipping it — see
+/// `store::update_flow_graph_override_wins_over_concurrently_enabled_row`
+/// (store_tests.rs) for the deterministic proof that this override also wins
+/// a genuine concurrent-enable race.
+#[tokio::test]
+async fn flows_update_disarms_manual_to_automatic_transition_even_when_already_disabled() {
+    let tmp = TempDir::new().unwrap();
+    let config = test_config(&tmp);
+
+    let created = flows_create(
+        &config,
+        "manual-then-scheduled".to_string(),
+        manual_trigger_graph(),
+        false,
+    )
+    .await
+    .unwrap();
+    flows_set_enabled(&config, &created.value.id, false)
+        .await
+        .unwrap();
+
+    let updated = flows_update(
+        &config,
+        &created.value.id,
+        None,
+        Some(schedule_trigger_graph("0 8 * * *")),
+        None,
+        None,
+    )
+    .await
+    .unwrap();
+
+    assert!(
+        !updated.value.enabled,
+        "a manual→automatic transition must never leave the flow enabled, regardless of \
+         whether it looked enabled going in"
+    );
+    let reloaded = flows_get(&config, &created.value.id).await.unwrap();
+    assert!(!reloaded.value.enabled);
+}
+
 #[tokio::test]
 async fn flows_update_preserves_enabled_when_already_automatic() {
     let tmp = TempDir::new().unwrap();

src/openhuman/flows/store_tests.rs

@@ -107,6 +107,110 @@ fn update_flow_graph_bumps_updated_at_and_preserves_created_at() {
     assert_eq!(updated.graph.name, "renamed-graph");
 }
 
+/// `enabled_override: None` must leave the persisted `enabled` column
+/// exactly as it was — `update_flow_graph` re-reads the current row and
+/// falls back to `current.enabled`, not to whatever the caller might have
+/// observed earlier.
+#[test]
+fn update_flow_graph_with_none_override_preserves_current_enabled_column() {
+    let tmp = TempDir::new().unwrap();
+    let config = test_config(&tmp);
+    let flow = create_flow(&config, "demo".to_string(), trigger_graph(), false, true).unwrap();
+    assert!(flow.enabled, "flow created enabled");
+
+    let updated = update_flow_graph(
+        &config,
+        &flow.id,
+        flow.name.clone(),
+        trigger_graph(),
+        false,
+        None, // enabled_override
+        None,
+    )
+    .unwrap();
+
+    assert!(
+        updated.enabled,
+        "a None override must preserve the row's current enabled state"
+    );
+    let reloaded = get_flow(&config, &flow.id).unwrap().unwrap();
+    assert!(reloaded.enabled);
+}
+
+/// `enabled_override: Some(false)` must force-persist `enabled=false`
+/// regardless of what the row's `enabled` column currently holds — this is
+/// the mechanism `flows_update`'s B29 Rule 1 analogue relies on to disarm a
+/// manual→automatic trigger transition in the same guarded write.
+#[test]
+fn update_flow_graph_with_some_false_override_forces_disabled() {
+    let tmp = TempDir::new().unwrap();
+    let config = test_config(&tmp);
+    let flow = create_flow(&config, "demo".to_string(), trigger_graph(), false, true).unwrap();
+    assert!(flow.enabled, "flow created enabled");
+
+    let updated = update_flow_graph(
+        &config,
+        &flow.id,
+        flow.name.clone(),
+        trigger_graph(),
+        false,
+        Some(false), // enabled_override
+        None,
+    )
+    .unwrap();
+
+    assert!(
+        !updated.enabled,
+        "a Some(false) override must force enabled=false even though the row was enabled"
+    );
+    let reloaded = get_flow(&config, &flow.id).unwrap().unwrap();
+    assert!(!reloaded.enabled);
+}
+
+/// Regression for the silent live-arming race Codex flagged on this PR:
+/// `flows_update` (ops.rs) makes its manual→automatic disarm decision from
+/// an *outer* `existing` read taken before `update_flow_graph`'s own guarded
+/// UPDATE re-reads the row. If a concurrent `flows_set_enabled(id, true)`
+/// landed in that gap — which bumps `updated_at`, so it would NOT trip the
+/// optimistic-concurrency conflict — the outer read would be stale while the
+/// row is actually enabled by write time. This proves the mechanism the fix
+/// relies on to close that race: an `enabled_override` of `Some(false)`
+/// (what `flows_update` now passes unconditionally on a manual→automatic
+/// transition, never gated on the stale outer read) always wins over
+/// whatever the row's `enabled` column was concurrently flipped to,
+/// simulated here by flipping it with `set_enabled` between the two calls.
+#[test]
+fn update_flow_graph_override_wins_over_concurrently_enabled_row() {
+    let tmp = TempDir::new().unwrap();
+    let config = test_config(&tmp);
+    let flow = create_flow(&config, "demo".to_string(), trigger_graph(), false, false).unwrap();
+    assert!(!flow.enabled, "flow created disabled");
+
+    // Simulates a concurrent `flows_set_enabled(id, true)` racing in after
+    // `flows_update`'s outer `existing` read observed `enabled: false`, but
+    // before its guarded `update_flow_graph` write below.
+    let raced = set_enabled(&config, &flow.id, true).unwrap();
+    assert!(raced.enabled);
+
+    let updated = update_flow_graph(
+        &config,
+        &flow.id,
+        flow.name.clone(),
+        trigger_graph(),
+        false,
+        Some(false), // the unconditional disarm override
+        None,
+    )
+    .unwrap();
+
+    assert!(
+        !updated.enabled,
+        "the disarm override must win over a concurrently-enabled row, not the reverse"
+    );
+    let reloaded = get_flow(&config, &flow.id).unwrap().unwrap();
+    assert!(!reloaded.enabled);
+}
+
 #[test]
 fn record_run_sets_last_run_fields() {
     let tmp = TempDir::new().unwrap();