diff --git a/src/openhuman/flows/ops.rs b/src/openhuman/flows/ops.rs index 3f5f7bae3f..3704acdad9 100644 --- a/src/openhuman/flows/ops.rs +++ b/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" ); diff --git a/src/openhuman/flows/ops_tests.rs b/src/openhuman/flows/ops_tests.rs index b4c1663c1b..93409a4793 100644 --- a/src/openhuman/flows/ops_tests.rs +++ b/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(); diff --git a/src/openhuman/flows/store_tests.rs b/src/openhuman/flows/store_tests.rs index 778712a0e9..cd679fae42 100644 --- a/src/openhuman/flows/store_tests.rs +++ b/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();