diff --git a/Cargo.lock b/Cargo.lock index b5ca659e1..ae4b13ec1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2579,6 +2579,7 @@ dependencies = [ "fabro-util", "serde", "serde_json", + "strum 0.28.0", "thiserror 2.0.18", "tokio", "tokio-util", diff --git a/docs/public/changelog/2026-08-26.mdx b/docs/public/changelog/2026-08-26.mdx new file mode 100644 index 000000000..8580d1b6b --- /dev/null +++ b/docs/public/changelog/2026-08-26.mdx @@ -0,0 +1,34 @@ +--- +title: "Explicit succeed failure policy" +date: "2026-08-26" +--- + +`on_failure` now accepts a third policy, `succeed`, alongside `route` and +`exit`. A failed node with an effective `succeed` policy and no explicit +recovery route finishes as `succeeded` and follows normal success routing. +The original failure details stay on the `stage.completed` event and in the +checkpoint, and the outcome's notes record the promotion. + +Set it on a node to mark a best-effort step inside a strict graph, or on the +graph to apply it everywhere: + +```dot +digraph Review { + graph [on_failure="exit"] + start [shape=Mdiamond] + exit [shape=Msquare] + required_check [script="./required-check"] + optional_scan [script="./optional-scan" on_failure="succeed"] + + start -> required_check -> optional_scan -> exit +} +``` + +An explicit `condition="outcome=failed"` edge still takes priority over the +promotion. A promoted outcome satisfies goal gates, and a failed parallel +branch with a `succeed` policy counts as succeeded in its parent's result. + +`auto_status=true` is now a deprecated alias for `on_failure="succeed"`. +Existing workflows keep working, and validation reports a new +`auto_status_deprecated` warning with the replacement. The alias no longer +promotes `partially_succeeded` outcomes; only `failed` outcomes are affected. diff --git a/docs/public/docs.json b/docs/public/docs.json index 28938510c..4a02c15a2 100644 --- a/docs/public/docs.json +++ b/docs/public/docs.json @@ -309,6 +309,7 @@ "group": "August 2026", "icon": "clock-rotate-left", "pages": [ + "changelog/2026-08-26", "changelog/2026-08-25", "changelog/2026-08-23", "changelog/2026-08-21", diff --git a/docs/public/execution/failures.mdx b/docs/public/execution/failures.mdx index 73b81bddb..76fb6ee2c 100644 --- a/docs/public/execution/failures.mdx +++ b/docs/public/execution/failures.mdx @@ -66,6 +66,27 @@ Set `on_failure` on a node to control that node alone. The node-level attribute The policy applies only to `failed`. Other outcomes keep their normal routing behavior. For parallel nodes, the policy uses the completed parallel node's final outcome. It does not stop or cancel individual branches early. +## Treat a failed node as succeeded + +Set `on_failure="succeed"` on a best-effort node so its failure never blocks the workflow. This pairs well with a strict graph default: + +```dot title="best-effort-node.fabro" +digraph Review { + graph [on_failure="exit"] + + start [shape=Mdiamond] + exit [shape=Msquare] + required_check [script="./required-check"] + optional_scan [script="./optional-scan" on_failure="succeed"] + + start -> required_check -> optional_scan -> exit +} +``` + +When `optional_scan` fails, Fabro first checks explicit recovery routes with the `failed` outcome. If none match, it rewrites the outcome to `succeeded` and routes the node as a success. Retries still run first; only the final outcome changes. The original failure stays on the `stage.completed` event and in the checkpoint, and the outcome's notes record the promotion. A promoted outcome satisfies a goal gate. Setting `on_failure="succeed"` on the graph applies it to every node. + +`succeed` applies only to `failed`. It does not change a `partially_succeeded` outcome. `auto_status=true` is the deprecated spelling of this policy; validation warns and suggests `on_failure="succeed"`. + ## Retry layers Fabro retries failures at three levels: **LLM retries** handle transient API errors inside a single model call, **turn-level retries** recover from dropped streams mid-response, and **node retries** re-execute the entire node handler when the first two levels aren't enough. These layers are independent — a node retry re-runs the full handler, which gets its own fresh set of LLM and turn-level retries. @@ -332,8 +353,8 @@ A node failure does **not** automatically terminate the run. Fabro follows this 4. **Node retries** — re-execute the entire handler (per the retry policy) 5. **Direct jump** — use `jump_to_node` when the outcome supplies one 6. **Explicit edge routing** — look for a matching condition, preferred label, or suggested next node -7. **Failure policy** — with an effective `on_failure="exit"` (node-level `on_failure` first, then graph-level), skip the unconditional edge; with `route` or no attribute, keep normal fallback routing -8. **Unconditional edge** — in `route` mode, use an edge without a condition as the fallback +7. **Failure policy** — with no explicit route, apply the effective `on_failure` (node-level `on_failure` first, then graph-level): `exit` skips the unconditional edge, `succeed` promotes the outcome to `succeeded` and routes it as a success, and `route` (or no attribute) keeps normal fallback routing +8. **Unconditional edge** — in `route` mode, or after a `succeed` promotion, use an edge without a condition as the fallback 9. **Retry target** — if no edge was selected, check `retry_target` and `fallback_retry_target` on the node, then on the graph 10. **Run failure** — if none of the above produces a path forward, the run terminates diff --git a/docs/public/execution/outcomes.mdx b/docs/public/execution/outcomes.mdx index 58935f3af..3e2450835 100644 --- a/docs/public/execution/outcomes.mdx +++ b/docs/public/execution/outcomes.mdx @@ -83,24 +83,28 @@ In this example, if the agent returns a retryable failure and all 5 standard-pol See [Retry policies](/execution/failures#retry-policies) for the available presets and backoff settings. -## `auto_status` +## Succeed on failure -When `auto_status=true`, any non-`succeeded` and non-`skipped` outcome is silently overridden to `succeeded` after the handler completes. This is applied after the retry loop, so retries still happen normally — only the final outcome is overridden. +When a node's effective `on_failure` policy is `succeed`, a `failed` outcome with no explicit recovery route is promoted to `succeeded`. This is applied after the retry loop, so retries still happen normally — only the final outcome changes. The original failure details stay on the `stage.completed` event and in the checkpoint, and the outcome's notes record the promotion. | Attribute | Type | Default | |---|---|---| -| `auto_status` | Boolean | `false` | +| `on_failure` | String | inherits the graph-level `on_failure` (default `route`) | ```dot scan [ label="Scan", shape=parallelogram, - auto_status=true, + on_failure="succeed", script="find . -name '*.log' | head -20" ] ``` -Use `auto_status` for nodes whose failure should never block the workflow — optional scans, best-effort cleanup steps, or informational commands where the output matters more than the exit code. +Use `on_failure="succeed"` for nodes whose failure should never block the workflow — optional scans, best-effort cleanup steps, or informational commands where the output matters more than the exit code. An explicit `condition="outcome=failed"` edge still takes priority; the promotion applies only when no explicit route matches. The policy applies only to `failed` and leaves `partially_succeeded` unchanged. See [Failed-node routing policy](/workflows/transitions#failed-node-routing-policy) for the full set of policies. + + +`auto_status=true` is the deprecated spelling of `on_failure="succeed"`. Fabro still accepts it as an alias, and validation reports an `auto_status_deprecated` warning with the replacement. Unlike the old attribute, the alias no longer promotes `partially_succeeded` outcomes. + ## Goal gate interaction @@ -111,7 +115,7 @@ make the workflow fail. Nodes marked with `goal_gate=true` are checked when the workflow reaches the exit node. A goal gate is satisfied if its last outcome was `succeeded` **or** `partially_succeeded`. Any other outcome (`failed`, `skipped`) causes the workflow to fail, even though execution reached the exit. -This means `allow_partial=true` on a goal gate node lets the gate pass even if the node exhausted its retries — the promoted `partially_succeeded` outcome counts as passing. +This means `allow_partial=true` on a goal gate node lets the gate pass even if the node exhausted its retries — the promoted `partially_succeeded` outcome counts as passing. Likewise, a `succeeded` outcome promoted by `on_failure="succeed"` satisfies the gate. See [Goal gates](/execution/failures#goal-gates) for retry target resolution and failure behavior. diff --git a/docs/public/reference/dot-language.mdx b/docs/public/reference/dot-language.mdx index 3f42716c2..f5378b2c9 100644 --- a/docs/public/reference/dot-language.mdx +++ b/docs/public/reference/dot-language.mdx @@ -77,7 +77,7 @@ rankdir=LR | `rankdir` | Identifier | Layout direction: `LR` (left-to-right) or `TB` (top-to-bottom) | | `model_stylesheet` | String | CSS-like rules for model assignment. The root value supports a MiniJinja template with `inputs` and `vars` (see [Model Stylesheets](/workflows/stylesheets)) | | `default_max_retries` | Integer | Default retry count for all nodes (default: 0) | -| `on_failure` | String | Failed-node routing policy: `route` (default) or `exit` | +| `on_failure` | String | Failed-node policy when no explicit recovery route matches: `route` (default), `exit`, or `succeed` | | `retry_target` | String | Default node ID to jump to on retry | | `fallback_retry_target` | String | Fallback retry target if primary target fails | | `default_fidelity` | String | Default [fidelity level](/execution/context) for all nodes | @@ -199,13 +199,13 @@ Other node types still need their shape, because their attributes don't identify | `class` | String | Classes for [stylesheet](/workflows/stylesheets) targeting. Separate multiple classes with spaces. Commas are also accepted for compatibility. | | `timeout` | Duration | Execution timeout (e.g. `900s`). An agent's wait for human input does not consume this budget. On a human node, this is the response deadline. | | `max_visits` | Integer | Max times this node can execute in a run. Overrides the graph-level `max_node_visits` for this node. | -| `on_failure` | String | Failed-node routing policy for this node: `route` or `exit`. Overrides the graph-level `on_failure`. | +| `on_failure` | String | Failed-node policy for this node: `route`, `exit`, or `succeed`. Overrides the graph-level `on_failure`. See [Node Outcomes](/execution/outcomes#succeed-on-failure). | | `max_retries` | Integer | Override default retry count | | `retry_policy` | String | Named preset: `none`, `standard`, `aggressive`, `linear`, `patient` | | `retry_target` | String | Node ID to jump to on retry | | `fallback_retry_target` | String | Fallback node ID if primary `retry_target` is unreachable | | `goal_gate` | Boolean | When `true`, workflow fails if this node didn't finish with `succeeded` or `partially_succeeded`. See [Node Outcomes](/execution/outcomes#goal-gate-interaction). | -| `auto_status` | Boolean | When `true`, overrides any non-`succeeded`/non-`skipped` outcome to `succeeded` after the handler completes. See [Node Outcomes](/execution/outcomes#auto_status). | +| `auto_status` | Boolean | Deprecated alias for `on_failure="succeed"`. Validation warns when it is present. | | `allow_partial` | Boolean | When `true` and retries are exhausted on a retry-requesting failure, promotes the outcome to `partially_succeeded` instead of `failed`. Default `false`. See [Node Outcomes](/execution/outcomes#allow_partial). | | `selection` | String | Edge tiebreaking strategy: `deterministic` (default) or `random` (weighted-random). Cannot be combined with conditional edges. | diff --git a/docs/public/workflows/transitions.mdx b/docs/public/workflows/transitions.mdx index 325ae7ec8..5cab4a79e 100644 --- a/docs/public/workflows/transitions.mdx +++ b/docs/public/workflows/transitions.mdx @@ -13,7 +13,7 @@ When a node completes, it produces an **outcome** with a [stage outcome](/execut 2. **Condition match** — Edges with a `condition` attribute are evaluated first. If one or more conditions match, the edge with the highest `weight` wins (lexical tiebreak on target node ID). 3. **Preferred label** — If the node's outcome includes a preferred label (for example, from a human gate selection), the edge whose `label` matches is chosen. 4. **Suggested next** — If the node suggests a specific next node ID, the edge pointing to that node is chosen. -5. **Failure policy** — For a failed outcome with an effective `on_failure="exit"` policy (node-level `on_failure` first, then graph-level), Fabro skips the unconditional fallback. +5. **Failure policy** — For a failed outcome with no explicit route, the effective `on_failure` policy (node-level `on_failure` first, then graph-level) decides what happens next. `exit` skips the unconditional fallback. `succeed` promotes the outcome to `succeeded` and routes it as a success. `route` continues to the unconditional fallback. 6. **Unconditional fallback** — Edges without conditions are considered last, again using `weight` then lexical tiebreak. 7. **Retry target** — For a failed outcome with no selected edge, Fabro checks node-level and graph-level `retry_target` and `fallback_retry_target` values. @@ -21,10 +21,13 @@ If no edge or retry target supplies a next node, the workflow ends. A failed nod ## Failed-node routing policy -The `on_failure` attribute controls whether a failed node can take an unconditional edge: +The `on_failure` attribute controls what happens to a failed node when no explicit recovery route matches: -- `route` keeps the existing routing behavior. It is the default. -- `exit` stops normal fallback routing after explicit routes fail to match. +| Policy | Effective outcome | Fallback routing | +|---|---|---| +| `route` (default) | stays `failed` | takes the unconditional edge | +| `exit` | stays `failed` | skips the unconditional edge; the run ends unless a retry target applies | +| `succeed` | becomes `succeeded` | uses normal success routing | Set it at the graph level to apply the policy to every node, or on a node to control that node alone. A node-level `on_failure` overrides the graph level. A node without the attribute inherits the graph policy. @@ -60,17 +63,38 @@ digraph Build { } ``` -The `exit` policy only applies to the `failed` outcome. It does not change routing for `succeeded`, `partially_succeeded`, or `skipped` outcomes. +Use `succeed` for a best-effort node whose failure must not block the workflow. Its failure becomes a `succeeded` outcome, so the node's normal success routing applies: -Conditioned edges, matching preferred labels, and matching suggested node IDs are explicit recovery routes. They still take priority under `exit`. An unmatched preferred label or suggested node ID does not make an unconditional edge explicit. +```dot title="best-effort-node.fabro" +digraph Review { + graph [on_failure="exit"] -Retry targets also remain available. Fabro checks them after it skips the unconditional fallback. Use graph-level `max_node_visits` or node-level `max_visits` to bound workflows whose retry targets return to a failing path. + start [shape=Mdiamond] + exit [shape=Msquare] + required_check [script="./required-check"] + optional_scan [script="./optional-scan" on_failure="succeed"] -A failed human gate never falls through to an unconditional edge, regardless of policy. Node-level `on_failure="route"` does not change that; route an interrupted gate explicitly with `condition="outcome=failed"`. + start -> required_check -> optional_scan -> exit +} +``` + +Under `succeed`, Fabro first checks explicit routes against the original `failed` outcome. If a `condition="outcome=failed"` edge, a matching preferred label, a matching suggested next node, or a handler jump applies, the outcome stays `failed` and that route is taken. Otherwise Fabro rewrites the outcome to `succeeded` before it records the node, so goal gates, the run context, events, and routing all see the promoted outcome. Edge selection then runs again: `condition="outcome=succeeded"` edges and unconditional edges apply. The original failure details stay on the `stage.completed` event and in the checkpoint, and the outcome's notes record which scope promoted it. A promoted outcome is not `failed`, so retry targets do not apply to it. + +Both `exit` and `succeed` apply only to the `failed` outcome. They do not change routing for `succeeded`, `partially_succeeded`, or `skipped` outcomes. + +Conditioned edges, matching preferred labels, and matching suggested node IDs are explicit recovery routes. They take priority under every policy. An unmatched preferred label or suggested node ID does not make an unconditional edge explicit. + +Retry targets also remain available under `exit`. Fabro checks them after it skips the unconditional fallback. Use graph-level `max_node_visits` or node-level `max_visits` to bound workflows whose retry targets return to a failing path. + +A failed human gate never falls through to an unconditional edge as a failure, regardless of policy. Node-level `on_failure="route"` does not change that; route an interrupted gate explicitly with `condition="outcome=failed"`. Under `succeed`, an interrupted gate with no explicit route is promoted like any other node and then follows its success routing. When `exit` stops routing, Fabro checkpoints the failed node without a next node and ends the run as failed. It does not execute the graph's exit node or emit an edge selection for an edge it did not take. An explicit recovery route can still reach the exit node normally. -For a parallel node, `exit` sees the final outcome returned by the parallel handler. It can stop routing for a failed parallel outcome. It does not add branch-level fail-fast behavior, and a `partially_succeeded` parallel outcome continues normally. +For a parallel node, `exit` and `succeed` see the final outcome returned by the parallel handler. `exit` can stop routing for a failed parallel outcome; `succeed` promotes it. Neither adds branch-level fail-fast behavior, and a `partially_succeeded` parallel outcome continues normally. Inside the fan-out, a branch node whose effective policy is `succeed` counts as `succeeded` in the parent's aggregate when it fails. Branches have no edge routing, so there is no explicit route to prefer. + + +`auto_status=true` is the deprecated spelling of node-level `on_failure="succeed"`. Fabro still accepts it as an alias and validation warns with the replacement. See [Node Outcomes](/execution/outcomes#succeed-on-failure). + ## Edge attributes @@ -201,7 +225,7 @@ gate -> fast_path [condition="outcome=succeeded"] gate -> slow_path ``` -For a failed outcome, `on_failure="exit"` skips this fallback after explicit routes are checked. The default `on_failure="route"` keeps the behavior shown above. +For a failed outcome, `on_failure="exit"` skips this fallback after explicit routes are checked, and `on_failure="succeed"` promotes the outcome to `succeeded` before taking it. The default `on_failure="route"` keeps the behavior shown above. ## Weight tiebreaking diff --git a/lib/apps/fabro-cli/tests/it/cmd/validate.rs b/lib/apps/fabro-cli/tests/it/cmd/validate.rs index dad1ac0eb..0af6cf8fa 100644 --- a/lib/apps/fabro-cli/tests/it/cmd/validate.rs +++ b/lib/apps/fabro-cli/tests/it/cmd/validate.rs @@ -390,11 +390,29 @@ fn invalid_node_on_failure_is_a_validation_failure() { Workflow: InvalidNodeOnFailure (3 nodes, 2 edges) Graph: [FIXTURES]/on_failure_node_invalid.fabro error [node: work]: Node 'work' has invalid on_failure value 'stop' (on_failure_valid) - fix: Use one of: route, exit + fix: Use one of: route, exit, succeed × Validation failed "); } +#[test] +fn deprecated_auto_status_warns_with_succeed_policy_replacement() { + let context = test_context!(); + let mut cmd = context.validate(); + cmd.arg(fixture("auto_status_deprecated.fabro")); + fabro_snapshot!(context.filters(), cmd, @" + success: true + exit_code: 0 + ----- stdout ----- + ----- stderr ----- + Workflow: DeprecatedAutoStatus (3 nodes, 2 edges) + Graph: [FIXTURES]/auto_status_deprecated.fabro + warning [node: scan]: Node 'scan' sets deprecated 'auto_status=true' (auto_status_deprecated) + fix: Use on_failure=\"succeed\" instead + Validation: OK + "); +} + #[test] fn invalid_on_failure_is_a_validation_failure() { let context = test_context!(); @@ -408,7 +426,7 @@ fn invalid_on_failure_is_a_validation_failure() { Workflow: InvalidOnFailure (2 nodes, 1 edges) Graph: [FIXTURES]/on_failure_invalid.fabro error: Graph has invalid on_failure value 'stop' (on_failure_valid) - fix: Use one of: route, exit + fix: Use one of: route, exit, succeed × Validation failed "); } diff --git a/lib/components/fabro-validate/src/rules/auto_status_deprecated.rs b/lib/components/fabro-validate/src/rules/auto_status_deprecated.rs new file mode 100644 index 000000000..5f258a52b --- /dev/null +++ b/lib/components/fabro-validate/src/rules/auto_status_deprecated.rs @@ -0,0 +1,163 @@ +use fabro_graphviz::graph::Graph; + +use crate::{Diagnostic, LintRule, Severity}; + +pub(super) fn rule() -> Box { + Box::new(Rule) +} + +/// `auto_status=true` is the deprecated spelling of `on_failure="succeed"`. +/// The runtime still honors it as an alias; this rule points workflows at the +/// explicit policy. +struct Rule; + +impl LintRule for Rule { + fn name(&self) -> &'static str { + "auto_status_deprecated" + } + + fn apply(&self, graph: &Graph) -> Vec { + let mut nodes: Vec<_> = graph + .nodes + .values() + .filter(|node| node.attrs.contains_key("auto_status")) + .collect(); + nodes.sort_unstable_by(|a, b| a.id.cmp(&b.id)); + + nodes + .into_iter() + .map(|node| { + let (message, fix) = if node.attrs.contains_key("on_failure") { + ( + format!( + "Node '{}' sets deprecated 'auto_status', which is ignored because \ + 'on_failure' is set", + node.id + ), + "Remove 'auto_status'", + ) + } else if node.auto_status() { + ( + format!("Node '{}' sets deprecated 'auto_status=true'", node.id), + "Use on_failure=\"succeed\" instead", + ) + } else { + ( + format!( + "Node '{}' sets deprecated 'auto_status', which has no effect \ + unless it is true", + node.id + ), + "Remove 'auto_status'", + ) + }; + Diagnostic { + rule: self.name().to_string(), + severity: Severity::Warning, + message, + node_id: Some(node.id.clone()), + fix: Some(fix.to_string()), + ..Diagnostic::default() + } + }) + .collect() + } +} + +#[cfg(test)] +mod tests { + use fabro_graphviz::graph::{AttrValue, Graph}; + + use super::Rule; + use crate::rules::test_support::{minimal_graph, node_with_attrs}; + use crate::{LintRule, Severity}; + + fn graph_with_auto_status(value: AttrValue, on_failure: Option<&str>) -> Graph { + let mut graph = minimal_graph(); + let mut node = match on_failure { + Some(policy) => node_with_attrs("work", &[("on_failure", policy)]), + None => node_with_attrs("work", &[]), + }; + node.attrs.insert("auto_status".to_string(), value); + graph.nodes.insert("work".to_string(), node); + graph + } + + #[test] + fn accepts_graphs_without_auto_status() { + let mut graph = minimal_graph(); + graph.nodes.insert( + "work".to_string(), + node_with_attrs("work", &[("on_failure", "succeed")]), + ); + assert!(Rule.apply(&graph).is_empty()); + } + + #[test] + fn warns_for_auto_status_true_and_suggests_succeed_policy() { + let graph = graph_with_auto_status(AttrValue::Boolean(true), None); + + let diagnostics = Rule.apply(&graph); + + assert_eq!(diagnostics.len(), 1); + assert_eq!(diagnostics[0].severity, Severity::Warning); + assert_eq!( + diagnostics[0].message, + "Node 'work' sets deprecated 'auto_status=true'" + ); + assert_eq!(diagnostics[0].node_id.as_deref(), Some("work")); + assert_eq!( + diagnostics[0].fix.as_deref(), + Some("Use on_failure=\"succeed\" instead") + ); + } + + #[test] + fn warns_that_auto_status_is_ignored_when_on_failure_is_set() { + let graph = graph_with_auto_status(AttrValue::Boolean(true), Some("exit")); + + let diagnostics = Rule.apply(&graph); + + assert_eq!(diagnostics.len(), 1); + assert_eq!(diagnostics[0].severity, Severity::Warning); + assert_eq!( + diagnostics[0].message, + "Node 'work' sets deprecated 'auto_status', which is ignored because 'on_failure' is set" + ); + assert_eq!(diagnostics[0].fix.as_deref(), Some("Remove 'auto_status'")); + } + + #[test] + fn warns_for_auto_status_false() { + let graph = graph_with_auto_status(AttrValue::Boolean(false), None); + + let diagnostics = Rule.apply(&graph); + + assert_eq!(diagnostics.len(), 1); + assert_eq!(diagnostics[0].severity, Severity::Warning); + assert_eq!( + diagnostics[0].message, + "Node 'work' sets deprecated 'auto_status', which has no effect unless it is true" + ); + assert_eq!(diagnostics[0].fix.as_deref(), Some("Remove 'auto_status'")); + } + + #[test] + fn reports_nodes_in_id_order() { + let mut graph = minimal_graph(); + for id in ["zeta", "alpha"] { + let mut node = node_with_attrs(id, &[]); + node.attrs + .insert("auto_status".to_string(), AttrValue::Boolean(true)); + graph.nodes.insert(id.to_string(), node); + } + + let ids: Vec<_> = Rule + .apply(&graph) + .into_iter() + .map(|diagnostic| diagnostic.node_id.unwrap()) + .collect(); + + assert_eq!(ids, ["alpha", "zeta"]); + } +} diff --git a/lib/components/fabro-validate/src/rules/mod.rs b/lib/components/fabro-validate/src/rules/mod.rs index 2d0366ab7..c73657718 100644 --- a/lib/components/fabro-validate/src/rules/mod.rs +++ b/lib/components/fabro-validate/src/rules/mod.rs @@ -1,4 +1,5 @@ mod all_conditional_edges; +mod auto_status_deprecated; mod backend_valid; mod command_requires_script; mod condition_syntax; @@ -65,6 +66,7 @@ pub fn built_in_rules() -> Vec> { all_conditional_edges::rule(), orphan_custom_outcome::rule(), on_failure_valid::rule(), + auto_status_deprecated::rule(), script_absolute_cd::rule(), command_requires_script::rule(), import_error::rule(), diff --git a/lib/components/fabro-validate/src/rules/on_failure_valid.rs b/lib/components/fabro-validate/src/rules/on_failure_valid.rs index 8f0d87987..ee3b8dd38 100644 --- a/lib/components/fabro-validate/src/rules/on_failure_valid.rs +++ b/lib/components/fabro-validate/src/rules/on_failure_valid.rs @@ -97,7 +97,7 @@ mod tests { let mut graph = minimal_graph(); assert!(Rule.apply(&graph).is_empty()); - for value in ["route", "exit"] { + for value in ["route", "exit", "succeed"] { graph.attrs.insert( "on_failure".to_string(), AttrValue::String(value.to_string()), @@ -109,7 +109,7 @@ mod tests { #[test] fn accepts_supported_node_values() { let mut graph = minimal_graph(); - for value in ["route", "exit"] { + for value in ["route", "exit", "succeed"] { graph.nodes.insert( "work".to_string(), node_with_attrs("work", &[("on_failure", value)]), @@ -136,7 +136,7 @@ mod tests { ); assert_eq!( diagnostics[0].fix.as_deref(), - Some("Use one of: route, exit") + Some("Use one of: route, exit, succeed") ); } @@ -176,7 +176,7 @@ mod tests { assert_eq!(diagnostics[0].node_id.as_deref(), Some("work")); assert_eq!( diagnostics[0].fix.as_deref(), - Some("Use one of: route, exit") + Some("Use one of: route, exit, succeed") ); } diff --git a/lib/components/fabro-workflow/src/context.rs b/lib/components/fabro-workflow/src/context.rs index 4d4ebebe0..d2e055e12 100644 --- a/lib/components/fabro-workflow/src/context.rs +++ b/lib/components/fabro-workflow/src/context.rs @@ -169,8 +169,49 @@ use fabro_graphviz::Fidelity; use fabro_types::{ParallelBranchId, RunId, StageId}; use serde::{Deserialize, Serialize}; -use crate::error::Error; +use crate::error::{Error, FailureSignature, FailureSignatureExt}; use crate::event::StageScope; +use crate::outcome::{Outcome, OutcomeExt}; + +/// Applies the context values derived from a completed node result. +/// +/// Edge-policy projection and the durable `after_record` lifecycle use this +/// same function so conditional routes observe identical values. +pub(crate) fn apply_recorded_outcome_context( + context: &Context, + node_id: &str, + outcome: &Outcome, + retry_count: u32, +) { + let failure_class = outcome.classified_failure_category(); + let failure_signature = failure_class + .map(|category| { + let signature_hint = outcome + .failure + .as_ref() + .and_then(|failure| failure.signature.as_deref()); + FailureSignature::new(node_id, category, signature_hint, outcome.failure_reason()) + .to_string() + }) + .unwrap_or_default(); + + context.set( + keys::retry_count_key(node_id), + serde_json::json!(retry_count), + ); + context.set(keys::OUTCOME, serde_json::json!(outcome.status.to_string())); + context.set( + keys::FAILURE_CLASS, + serde_json::json!(failure_class.map_or(String::new(), |class| class.to_string())), + ); + context.set( + keys::FAILURE_SIGNATURE, + serde_json::json!(failure_signature), + ); + if let Some(preferred_label) = &outcome.preferred_label { + context.set(keys::PREFERRED_LABEL, serde_json::json!(preferred_label)); + } +} /// Keys whose values changed or were added in `after` relative to `before`. /// Takes `after` by value so changed entries move instead of clone. diff --git a/lib/components/fabro-workflow/src/graph.rs b/lib/components/fabro-workflow/src/graph.rs index 1381828aa..734cfc64c 100644 --- a/lib/components/fabro-workflow/src/graph.rs +++ b/lib/components/fabro-workflow/src/graph.rs @@ -5,10 +5,11 @@ use std::sync::Arc; use fabro_core::error::{Error as CoreError, Result as CoreResult}; use fabro_core::graph::{EdgeSelection as CoreEdgeSelection, EdgeSpec, Graph, NodeSpec}; +use fabro_core::outcome::NodeResult; use fabro_graphviz::graph::types::{Edge as GvEdge, Graph as GvGraph, Node as GvNode}; use fabro_types::ResolvedOnFailure; -use crate::context::Context; +use crate::context::{self, Context}; use crate::outcome::{BilledModelUsage, Outcome}; // ---- WorkflowNode ---- @@ -120,6 +121,20 @@ impl Graph for WorkflowGraph { }) } + fn project_result_context( + &self, + node: &Self::Node, + result: &NodeResult, + context: &Context, + ) { + context::apply_recorded_outcome_context( + context, + node.id(), + &result.outcome, + result.attempts.saturating_sub(1), + ); + } + fn check_goal_gates( &self, outcomes: &HashMap, diff --git a/lib/components/fabro-workflow/src/graph/routing.rs b/lib/components/fabro-workflow/src/graph/routing.rs index 0de75ab97..e34dffec3 100644 --- a/lib/components/fabro-workflow/src/graph/routing.rs +++ b/lib/components/fabro-workflow/src/graph/routing.rs @@ -1,7 +1,7 @@ use std::collections::HashMap; +use fabro_core::graph::EdgeSelectionReason; use fabro_graphviz::graph::types::{Edge as GvEdge, Graph as GvGraph, Node as GvNode}; -use fabro_types::OnFailure; use rand::Rng; use crate::condition::evaluate_condition; @@ -11,7 +11,7 @@ use crate::outcome::Outcome; /// Result of edge selection: the chosen edge and the reason it was selected. pub(crate) struct SelectedGraphEdge<'a> { pub(crate) edge: &'a GvEdge, - pub(crate) reason: &'static str, + pub(crate) reason: EdgeSelectionReason, } /// Check whether a node is a terminal (exit) node. @@ -44,7 +44,7 @@ pub(crate) fn select_edge<'a>( if !condition_matched.is_empty() { return pick_edge(&condition_matched, selection).map(|edge| SelectedGraphEdge { edge, - reason: "condition", + reason: EdgeSelectionReason::Condition, }); } @@ -56,7 +56,7 @@ pub(crate) fn select_edge<'a>( if normalize_label(label) == normalized_pref { return Some(SelectedGraphEdge { edge, - reason: "preferred_label", + reason: EdgeSelectionReason::PreferredLabel, }); } } @@ -69,16 +69,12 @@ pub(crate) fn select_edge<'a>( if edge.condition().is_none_or(str::is_empty) && edge.to == *suggested_id { return Some(SelectedGraphEdge { edge, - reason: "suggested_next", + reason: EdgeSelectionReason::SuggestedNext, }); } } } - if outcome.status.is_failure() && graph.resolve_on_failure(node).policy() == OnFailure::Exit { - return None; - } - if blocks_unconditional_failure_fallthrough(node, outcome) { return None; } @@ -91,7 +87,7 @@ pub(crate) fn select_edge<'a>( if !unconditional.is_empty() { return pick_edge(&unconditional, selection).map(|edge| SelectedGraphEdge { edge, - reason: "unconditional", + reason: EdgeSelectionReason::Unconditional, }); } @@ -238,6 +234,7 @@ mod tests { use std::collections::HashMap; use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node}; + use fabro_types::{OnFailure, ResolvedOnFailure}; use super::*; use crate::context::Context; @@ -257,20 +254,6 @@ mod tests { g } - fn set_on_failure(graph: &mut Graph, on_failure: OnFailure) { - graph.attrs.insert( - "on_failure".to_string(), - AttrValue::String(on_failure.to_string()), - ); - } - - fn set_node_on_failure(graph: &mut Graph, node_id: &str, value: &str) { - graph.nodes.get_mut(node_id).unwrap().attrs.insert( - "on_failure".to_string(), - AttrValue::String(value.to_string()), - ); - } - #[test] fn normalize_label_lowercase_and_trim() { assert_eq!(normalize_label(" Yes "), "yes"); @@ -394,41 +377,25 @@ mod tests { let context = Context::new(); let sel = select_edge(node, &outcome, &context, &g, "deterministic").unwrap(); assert_eq!(sel.edge.to, "b"); - assert_eq!(sel.reason, "unconditional"); + assert_eq!(sel.reason, EdgeSelectionReason::Unconditional); } #[test] - fn failed_outcome_takes_unconditional_edge_in_default_and_route_modes() { - for policy in [None, Some(OnFailure::Route)] { - let mut graph = make_graph_with_edges(vec![Edge::new("a", "b")]); - if let Some(policy) = policy { - set_on_failure(&mut graph, policy); - } - let node = graph.nodes.get("a").unwrap(); - let outcome = Outcome::fail_classify("boom"); - - let selected = - select_edge(node, &outcome, &Context::new(), &graph, "deterministic").unwrap(); - - assert_eq!(selected.edge.to, "b"); - assert_eq!(selected.reason, "unconditional"); - } - } - - #[test] - fn exit_policy_blocks_unconditional_edge_for_failed_outcome() { - let mut graph = make_graph_with_edges(vec![Edge::new("a", "b")]); - set_on_failure(&mut graph, OnFailure::Exit); + fn failed_outcome_selects_unconditional_edge() { + let graph = make_graph_with_edges(vec![Edge::new("a", "b")]); let node = graph.nodes.get("a").unwrap(); let outcome = Outcome::fail_classify("boom"); - assert!(select_edge(node, &outcome, &Context::new(), &graph, "deterministic").is_none()); + let selected = + select_edge(node, &outcome, &Context::new(), &graph, "deterministic").unwrap(); + + assert_eq!(selected.edge.to, "b"); + assert_eq!(selected.reason, EdgeSelectionReason::Unconditional); } #[test] - fn exit_policy_allows_unconditional_edge_for_non_failed_outcomes() { - let mut graph = make_graph_with_edges(vec![Edge::new("a", "b")]); - set_on_failure(&mut graph, OnFailure::Exit); + fn non_failed_outcomes_select_unconditional_edge() { + let graph = make_graph_with_edges(vec![Edge::new("a", "b")]); let node = graph.nodes.get("a").unwrap(); let mut partial = Outcome::success(); partial.status = StageOutcome::PartiallySucceeded; @@ -437,19 +404,18 @@ mod tests { let selected = select_edge(node, &outcome, &Context::new(), &graph, "deterministic").unwrap(); assert_eq!(selected.edge.to, "b"); - assert_eq!(selected.reason, "unconditional"); + assert_eq!(selected.reason, EdgeSelectionReason::Unconditional); } } #[test] - fn exit_policy_allows_matching_failure_condition() { + fn failure_condition_is_an_explicit_selection() { let mut recovery = Edge::new("a", "recover"); recovery.attrs.insert( "condition".to_string(), AttrValue::String("outcome=failed".to_string()), ); - let mut graph = make_graph_with_edges(vec![recovery, Edge::new("a", "fallback")]); - set_on_failure(&mut graph, OnFailure::Exit); + let graph = make_graph_with_edges(vec![recovery, Edge::new("a", "fallback")]); let node = graph.nodes.get("a").unwrap(); let outcome = Outcome::fail_classify("boom"); @@ -457,18 +423,17 @@ mod tests { select_edge(node, &outcome, &Context::new(), &graph, "deterministic").unwrap(); assert_eq!(selected.edge.to, "recover"); - assert_eq!(selected.reason, "condition"); + assert_eq!(selected.reason, EdgeSelectionReason::Condition); } #[test] - fn exit_policy_allows_matching_preferred_and_suggested_routes() { + fn preferred_and_suggested_routes_are_explicit_selections() { let mut preferred = Edge::new("a", "preferred"); preferred.attrs.insert( "label".to_string(), AttrValue::String("Recover".to_string()), ); - let mut graph = make_graph_with_edges(vec![preferred, Edge::new("a", "suggested")]); - set_on_failure(&mut graph, OnFailure::Exit); + let graph = make_graph_with_edges(vec![preferred, Edge::new("a", "suggested")]); let node = graph.nodes.get("a").unwrap(); let mut preferred_outcome = Outcome::fail_classify("boom"); @@ -482,7 +447,7 @@ mod tests { ) .unwrap(); assert_eq!(selected.edge.to, "preferred"); - assert_eq!(selected.reason, "preferred_label"); + assert_eq!(selected.reason, EdgeSelectionReason::PreferredLabel); let mut suggested_outcome = Outcome::fail_classify("boom"); suggested_outcome.suggested_next_ids = vec!["suggested".to_string()]; @@ -495,118 +460,26 @@ mod tests { ) .unwrap(); assert_eq!(selected.edge.to, "suggested"); - assert_eq!(selected.reason, "suggested_next"); + assert_eq!(selected.reason, EdgeSelectionReason::SuggestedNext); } #[test] - fn exit_policy_blocks_fallback_for_unmatched_routing_hints() { - let mut graph = make_graph_with_edges(vec![Edge::new("a", "fallback")]); - set_on_failure(&mut graph, OnFailure::Exit); - let node = graph.nodes.get("a").unwrap(); - - let mut preferred_outcome = Outcome::fail_classify("boom"); - preferred_outcome.preferred_label = Some("missing".to_string()); - assert!( - select_edge( - node, - &preferred_outcome, - &Context::new(), - &graph, - "deterministic" - ) - .is_none() + fn promoted_outcome_selects_succeeded_condition() { + let mut on_success = Edge::new("a", "next"); + on_success.attrs.insert( + "condition".to_string(), + AttrValue::String("outcome=succeeded".to_string()), ); - - let mut suggested_outcome = Outcome::fail_classify("boom"); - suggested_outcome.suggested_next_ids = vec!["missing".to_string()]; - assert!( - select_edge( - node, - &suggested_outcome, - &Context::new(), - &graph, - "deterministic" - ) - .is_none() - ); - } - - #[test] - fn node_exit_overrides_graph_route_and_blocks_unconditional_edge() { - for graph_policy in [None, Some(OnFailure::Route)] { - let mut graph = make_graph_with_edges(vec![Edge::new("a", "b")]); - if let Some(policy) = graph_policy { - set_on_failure(&mut graph, policy); - } - set_node_on_failure(&mut graph, "a", "exit"); - let node = graph.nodes.get("a").unwrap(); - let outcome = Outcome::fail_classify("boom"); - - assert!( - select_edge(node, &outcome, &Context::new(), &graph, "deterministic").is_none() - ); - } - } - - #[test] - fn node_route_overrides_graph_exit_and_takes_unconditional_edge() { - let mut graph = make_graph_with_edges(vec![Edge::new("a", "b")]); - set_on_failure(&mut graph, OnFailure::Exit); - set_node_on_failure(&mut graph, "a", "route"); + let graph = make_graph_with_edges(vec![on_success, Edge::new("a", "fallback")]); let node = graph.nodes.get("a").unwrap(); - let outcome = Outcome::fail_classify("boom"); + let mut outcome = Outcome::fail_classify("boom"); + outcome.apply_on_failure(ResolvedOnFailure::node(OnFailure::Succeed)); let selected = select_edge(node, &outcome, &Context::new(), &graph, "deterministic").unwrap(); - assert_eq!(selected.edge.to, "b"); - assert_eq!(selected.reason, "unconditional"); - } - - #[test] - fn node_policy_does_not_affect_routing_from_other_nodes() { - let mut graph = make_graph_with_edges(vec![Edge::new("a", "b")]); - set_node_on_failure(&mut graph, "b", "exit"); - let node = graph.nodes.get("a").unwrap(); - let outcome = Outcome::fail_classify("boom"); - - let selected = - select_edge(node, &outcome, &Context::new(), &graph, "deterministic").unwrap(); - - assert_eq!(selected.edge.to, "b"); - assert_eq!(selected.reason, "unconditional"); - } - - #[test] - fn invalid_node_on_failure_value_inherits_graph_policy() { - let mut graph = make_graph_with_edges(vec![Edge::new("a", "b")]); - set_on_failure(&mut graph, OnFailure::Exit); - set_node_on_failure(&mut graph, "a", "bogus"); - let node = graph.nodes.get("a").unwrap(); - let outcome = Outcome::fail_classify("boom"); - - assert!(select_edge(node, &outcome, &Context::new(), &graph, "deterministic").is_none()); - } - - #[test] - fn failed_human_gate_ignores_node_route_override() { - let mut graph = make_graph_with_edges(vec![ - Edge::new("gate", "approve"), - Edge::new("gate", "skip"), - ]); - set_on_failure(&mut graph, OnFailure::Exit); - set_node_on_failure(&mut graph, "gate", "route"); - let gate = graph.nodes.get_mut("gate").unwrap(); - gate.attrs.insert( - "shape".to_string(), - AttrValue::String("hexagon".to_string()), - ); - let node = graph.nodes.get("gate").unwrap().clone(); - let outcome = Outcome::fail_deterministic( - "human interaction interrupted before an answer was provided", - ); - - assert!(select_edge(&node, &outcome, &Context::new(), &graph, "deterministic").is_none()); + assert_eq!(selected.edge.to, "next"); + assert_eq!(selected.reason, EdgeSelectionReason::Condition); } #[test] @@ -627,7 +500,7 @@ mod tests { let context = Context::new(); let sel = select_edge(node, &outcome, &context, &g, "deterministic").unwrap(); assert_eq!(sel.edge.to, "success_path"); - assert_eq!(sel.reason, "condition"); + assert_eq!(sel.reason, EdgeSelectionReason::Condition); } #[test] @@ -649,7 +522,7 @@ mod tests { let context = Context::new(); let sel = select_edge(node, &outcome, &context, &g, "deterministic").unwrap(); assert_eq!(sel.edge.to, "fix"); - assert_eq!(sel.reason, "preferred_label"); + assert_eq!(sel.reason, EdgeSelectionReason::PreferredLabel); } #[test] @@ -663,7 +536,7 @@ mod tests { let context = Context::new(); let sel = select_edge(node, &outcome, &context, &g, "deterministic").unwrap(); assert_eq!(sel.edge.to, "path2"); - assert_eq!(sel.reason, "suggested_next"); + assert_eq!(sel.reason, EdgeSelectionReason::SuggestedNext); } #[test] @@ -679,7 +552,7 @@ mod tests { let context = Context::new(); let sel = select_edge(node, &outcome, &context, &g, "deterministic").unwrap(); assert_eq!(sel.edge.to, "high"); - assert_eq!(sel.reason, "unconditional"); + assert_eq!(sel.reason, EdgeSelectionReason::Unconditional); } #[test] @@ -692,7 +565,7 @@ mod tests { let context = Context::new(); let sel = select_edge(node, &outcome, &context, &g, "deterministic").unwrap(); assert_eq!(sel.edge.to, "alpha"); - assert_eq!(sel.reason, "unconditional"); + assert_eq!(sel.reason, EdgeSelectionReason::Unconditional); } #[test] @@ -709,7 +582,7 @@ mod tests { let context = Context::new(); let sel = select_edge(node, &outcome, &context, &g, "deterministic").unwrap(); assert_eq!(sel.edge.to, "cond_path"); - assert_eq!(sel.reason, "condition"); + assert_eq!(sel.reason, EdgeSelectionReason::Condition); } #[test] @@ -722,7 +595,7 @@ mod tests { let context = Context::new(); let sel = select_edge(node, &outcome, &context, &g, "random").unwrap(); assert!(sel.edge.to == "b" || sel.edge.to == "c"); - assert_eq!(sel.reason, "unconditional"); + assert_eq!(sel.reason, EdgeSelectionReason::Unconditional); } #[test] @@ -740,29 +613,26 @@ mod tests { let context = Context::new(); let sel = select_edge(node, &outcome, &context, &g, "random").unwrap(); assert_eq!(sel.edge.to, "approve"); - assert_eq!(sel.reason, "preferred_label"); + assert_eq!(sel.reason, EdgeSelectionReason::PreferredLabel); } #[test] fn select_edge_failed_human_gate_does_not_fall_through_to_unconditional() { - for policy in [OnFailure::Route, OnFailure::Exit] { - let mut graph = make_graph_with_edges(vec![ - Edge::new("gate", "approve"), - Edge::new("gate", "skip"), - ]); - set_on_failure(&mut graph, policy); - let mut node = graph.nodes.get("gate").unwrap().clone(); - node.attrs.insert( - "shape".to_string(), - AttrValue::String("hexagon".to_string()), - ); - let outcome = Outcome::fail_deterministic( - "human interaction interrupted before an answer was provided", - ); - let context = Context::new(); + let graph = make_graph_with_edges(vec![ + Edge::new("gate", "approve"), + Edge::new("gate", "skip"), + ]); + let mut node = graph.nodes.get("gate").unwrap().clone(); + node.attrs.insert( + "shape".to_string(), + AttrValue::String("hexagon".to_string()), + ); + let outcome = Outcome::fail_deterministic( + "human interaction interrupted before an answer was provided", + ); + let context = Context::new(); - assert!(select_edge(&node, &outcome, &context, &graph, "deterministic").is_none()); - } + assert!(select_edge(&node, &outcome, &context, &graph, "deterministic").is_none()); } #[test] @@ -773,23 +643,20 @@ mod tests { AttrValue::String("outcome=failed".to_string()), ); let approve = Edge::new("gate", "approve"); - for policy in [OnFailure::Route, OnFailure::Exit] { - let mut graph = make_graph_with_edges(vec![fail.clone(), approve.clone()]); - set_on_failure(&mut graph, policy); - let mut node = graph.nodes.get("gate").unwrap().clone(); - node.attrs.insert( - "shape".to_string(), - AttrValue::String("hexagon".to_string()), - ); - let outcome = Outcome::fail_deterministic( - "human interaction interrupted before an answer was provided", - ); - let context = Context::new(); + let graph = make_graph_with_edges(vec![fail, approve]); + let mut node = graph.nodes.get("gate").unwrap().clone(); + node.attrs.insert( + "shape".to_string(), + AttrValue::String("hexagon".to_string()), + ); + let outcome = Outcome::fail_deterministic( + "human interaction interrupted before an answer was provided", + ); + let context = Context::new(); - let sel = select_edge(&node, &outcome, &context, &graph, "deterministic").unwrap(); - assert_eq!(sel.edge.to, "retry"); - assert_eq!(sel.reason, "condition"); - } + let sel = select_edge(&node, &outcome, &context, &graph, "deterministic").unwrap(); + assert_eq!(sel.edge.to, "retry"); + assert_eq!(sel.reason, EdgeSelectionReason::Condition); } #[test] diff --git a/lib/components/fabro-workflow/src/handler/parallel.rs b/lib/components/fabro-workflow/src/handler/parallel.rs index edd09a56b..35796115c 100644 --- a/lib/components/fabro-workflow/src/handler/parallel.rs +++ b/lib/components/fabro-workflow/src/handler/parallel.rs @@ -528,7 +528,7 @@ async fn run_branches( ); let mut attempt = 0_u32; - let outcome = loop { + let mut outcome = loop { attempt = attempt.saturating_add(1); let attempt_result = node_handler::execute_single_attempt( &target, @@ -569,6 +569,12 @@ async fn run_branches( backoff_or_cancel(delay, &branch_services).await?; permit = acquire_branch_permit(&semaphore, &branch_services).await?; }; + // Branches have no edge routing, so `succeed` has no + // explicit recovery route to defer to: a failed branch + // under that policy always counts as succeeded in the + // parent's aggregate, with its failure kept on the + // outcome. + outcome.apply_on_failure(graph.resolve_on_failure(&target)); let context_updates = branch_context_updates( &parent_snapshot, @@ -1401,6 +1407,122 @@ mod tests { } } + /// Fails the named branch node and succeeds everywhere else. + struct FailNamedBranchHandler(&'static str); + + #[async_trait] + impl Handler for FailNamedBranchHandler { + async fn execute( + &self, + node: &Node, + _context: &Context, + _graph: &Graph, + _run_dir: &Path, + _services: &EngineServices, + ) -> Result { + if node.id == self.0 { + Ok(Outcome::fail_classify("branch boom")) + } else { + Ok(Outcome::success()) + } + } + } + + /// Mutates a test graph to opt a branch into the `succeed` policy. + type PolicyEdit = fn(&mut Graph); + + async fn run_parallel_with_failing_branch_a(graph: &Graph, node: &Node) -> Outcome { + let mut services = make_services(); + services.registry = Arc::new(super::super::HandlerRegistry::new(Box::new( + FailNamedBranchHandler("branch_a"), + ))); + ParallelHandler + .execute( + node, + &test_context(), + graph, + Path::new("/tmp/test"), + &services, + ) + .await + .unwrap() + } + + #[tokio::test] + async fn parallel_handler_failed_branch_without_policy_is_partial() { + let (node, graph) = parallel_graph(); + + let outcome = run_parallel_with_failing_branch_a(&graph, &node).await; + + assert_eq!(outcome.status, StageOutcome::PartiallySucceeded); + let results: Vec = + serde_json::from_value(outcome.context_updates[keys::PARALLEL_RESULTS].clone()) + .unwrap(); + assert!(results[0].status.is_failure()); + assert_eq!(results[1].status, StageOutcome::Succeeded); + } + + #[tokio::test] + async fn parallel_handler_succeed_policy_counts_failed_branch_as_succeeded() { + let cases: [(&str, PolicyEdit); 3] = [ + ("node", |graph| { + graph.nodes.get_mut("branch_a").unwrap().attrs.insert( + "on_failure".to_string(), + AttrValue::String("succeed".to_string()), + ); + }), + ("graph", |graph| { + graph.attrs.insert( + "on_failure".to_string(), + AttrValue::String("succeed".to_string()), + ); + }), + ("alias", |graph| { + graph + .nodes + .get_mut("branch_a") + .unwrap() + .attrs + .insert("auto_status".to_string(), AttrValue::Boolean(true)); + }), + ]; + for (scope, apply) in cases { + let (node, mut graph) = parallel_graph(); + apply(&mut graph); + + let outcome = run_parallel_with_failing_branch_a(&graph, &node).await; + + assert_eq!(outcome.status, StageOutcome::Succeeded, "scope {scope}"); + assert_eq!( + outcome.notes.as_deref(), + Some("Parallel node dispatched 2 branches (2 succeeded, 0 failed)"), + "scope {scope}" + ); + let results: Vec = + serde_json::from_value(outcome.context_updates[keys::PARALLEL_RESULTS].clone()) + .unwrap(); + assert!( + results + .iter() + .all(|result| result.status == StageOutcome::Succeeded), + "scope {scope}" + ); + } + } + + #[tokio::test] + async fn parallel_handler_exit_policy_does_not_change_branch_outcomes() { + let (node, mut graph) = parallel_graph(); + graph.attrs.insert( + "on_failure".to_string(), + AttrValue::String("exit".to_string()), + ); + + let outcome = run_parallel_with_failing_branch_a(&graph, &node).await; + + assert_eq!(outcome.status, StageOutcome::PartiallySucceeded); + } + #[tokio::test] async fn parallel_handler_no_branches() { let outcome = ParallelHandler diff --git a/lib/components/fabro-workflow/src/lifecycle/auto_status.rs b/lib/components/fabro-workflow/src/lifecycle/auto_status.rs deleted file mode 100644 index fb48bbde7..000000000 --- a/lib/components/fabro-workflow/src/lifecycle/auto_status.rs +++ /dev/null @@ -1,37 +0,0 @@ -use async_trait::async_trait; -use fabro_core::error::Result as CoreResult; -use fabro_core::lifecycle::RunLifecycle; -use fabro_core::outcome::NodeResult; -use fabro_core::state::ExecutionState; - -use crate::graph::{WorkflowGraph, WorkflowNode}; -use crate::outcome::{BilledModelUsage, StageOutcome}; - -type WfRunState = ExecutionState>; -type WfNodeResult = NodeResult>; - -/// Sub-lifecycle responsible for auto-status override on nodes with -/// `auto_status=true`. -pub(crate) struct AutoStatusLifecycle; - -#[async_trait] -impl RunLifecycle for AutoStatusLifecycle { - async fn after_node( - &self, - node: &WorkflowNode, - result: &mut WfNodeResult, - _state: &WfRunState, - ) -> CoreResult<()> { - let gv = node.inner(); - let outcome = &mut result.outcome; - if gv.auto_status() - && outcome.status != StageOutcome::Succeeded - && outcome.status != StageOutcome::Skipped - { - outcome.status = StageOutcome::Succeeded; - outcome.notes = - Some("auto-status: handler completed without writing status".to_string()); - } - Ok(()) - } -} diff --git a/lib/components/fabro-workflow/src/lifecycle/mod.rs b/lib/components/fabro-workflow/src/lifecycle/mod.rs index 64a6323f2..dd4fa7338 100644 --- a/lib/components/fabro-workflow/src/lifecycle/mod.rs +++ b/lib/components/fabro-workflow/src/lifecycle/mod.rs @@ -1,5 +1,4 @@ pub(crate) mod artifact; -pub(crate) mod auto_status; pub(crate) mod circuit_breaker; pub(crate) mod event; pub(crate) mod fidelity; @@ -26,7 +25,6 @@ use fabro_sandbox::Sandbox; use fabro_types::RunId; use self::artifact::ArtifactLifecycle; -use self::auto_status::AutoStatusLifecycle; use self::circuit_breaker::CircuitBreakerLifecycle; use self::event::EventLifecycle; use self::fidelity::FidelityLifecycle; @@ -34,10 +32,10 @@ use self::git::{GitCheckpointResult, GitLifecycle}; use self::hook::HookLifecycle; use crate::artifact_upload::ArtifactSink; use crate::context; -use crate::error::{FailureSignature, FailureSignatureExt}; +use crate::error::FailureSignature; use crate::event::Emitter; use crate::graph::{WorkflowGraph, WorkflowNode}; -use crate::outcome::{BilledModelUsage, Outcome, OutcomeExt}; +use crate::outcome::{BilledModelUsage, Outcome}; use crate::run_control::RunControlState; use crate::run_metadata::{RunMetadataRuntime, RunMetadataWriterHandle}; use crate::run_options::RunOptions; @@ -56,7 +54,6 @@ pub(crate) struct WorkflowLifecycle { event: EventLifecycle, hook: HookLifecycle, fidelity: FidelityLifecycle, - auto_status: AutoStatusLifecycle, circuit_breaker: Arc, git: GitLifecycle, artifact: ArtifactLifecycle, @@ -187,7 +184,6 @@ impl WorkflowLifecycle { event, hook, fidelity, - auto_status: AutoStatusLifecycle, circuit_breaker, git, artifact, @@ -361,7 +357,6 @@ impl RunLifecycle for WorkflowLifecycle { result: &mut WfNodeResult, state: &WfRunState, ) -> CoreResult<()> { - self.auto_status.after_node(node, result, state).await?; self.circuit_breaker.after_node(node, result, state).await?; self.artifact.after_node(node, result, state).await?; self.event.after_node(node, result, state).await?; @@ -375,47 +370,13 @@ impl RunLifecycle for WorkflowLifecycle { result: &WfNodeResult, state: &WfRunState, ) -> CoreResult<()> { - let outcome = &result.outcome; let retry_count = state.node_retries.get(node.id()).copied().unwrap_or(0); - let failure_class = outcome.classified_failure_category(); - let failure_signature = failure_class - .map(|category| { - let signature_hint = outcome - .failure - .as_ref() - .and_then(|f| f.signature.as_deref()); - FailureSignature::new( - node.id(), - category, - signature_hint, - outcome.failure_reason(), - ) - .to_string() - }) - .unwrap_or_default(); - - state.context.set( - context::keys::retry_count_key(node.id()), - serde_json::json!(retry_count), + context::apply_recorded_outcome_context( + &state.context, + node.id(), + &result.outcome, + retry_count, ); - state.context.set( - context::keys::OUTCOME, - serde_json::json!(outcome.status.to_string()), - ); - state.context.set( - context::keys::FAILURE_CLASS, - serde_json::json!(failure_class.map_or(String::new(), |fc| fc.to_string())), - ); - state.context.set( - context::keys::FAILURE_SIGNATURE, - serde_json::json!(failure_signature), - ); - if let Some(ref preferred_label) = outcome.preferred_label { - state.context.set( - context::keys::PREFERRED_LABEL, - serde_json::json!(preferred_label), - ); - } Ok(()) } diff --git a/lib/components/fabro-workflow/tests/it/integration.rs b/lib/components/fabro-workflow/tests/it/integration.rs index 4c6ac37b3..2f6602cf5 100644 --- a/lib/components/fabro-workflow/tests/it/integration.rs +++ b/lib/components/fabro-workflow/tests/it/integration.rs @@ -1143,7 +1143,11 @@ impl Handler for OnFailureRecordingHandler { ) -> Result { self.visits.lock().unwrap().push(node.id.clone()); if node.id == "work" { - Ok(Outcome::fail_classify("forced work failure")) + let mut outcome = Outcome::fail_classify("forced work failure"); + outcome + .context_updates + .insert("recovery_ready".to_string(), serde_json::json!(true)); + Ok(outcome) } else { Ok(Outcome::success()) } @@ -1319,6 +1323,127 @@ async fn node_on_failure_route_overrides_graph_exit_policy() { assert_eq!(*run.visits.lock().unwrap(), vec!["work", "downstream"]); } +#[tokio::test] +async fn node_on_failure_succeed_promotes_failed_node_and_keeps_failure_in_events() { + let graph = on_failure_graph( + r#"graph [on_failure="exit"] + work [on_failure="succeed"]"#, + ); + let emitter = Emitter::default(); + let events = collect_events(&emitter); + let run = run_on_failure(&graph, emitter).await; + + assert_eq!(run.outcome.status, StageOutcome::Succeeded); + assert_eq!(*run.visits.lock().unwrap(), vec!["work", "downstream"]); + + let checkpoint = run + .state + .current_checkpoint() + .expect("downstream should be checkpointed"); + let work = &checkpoint.node_outcomes["work"]; + assert_eq!(work.status, StageOutcome::Succeeded); + assert_eq!(work.failure_reason(), Some("forced work failure")); + assert_eq!( + work.notes.as_deref(), + Some("node on_failure=succeed promoted a failed outcome to succeeded") + ); + + let events = events.lock().unwrap(); + let completed = events + .iter() + .find_map(|event| match &event.body { + EventBody::StageCompleted(props) if event.node_id.as_deref() == Some("work") => { + Some(props.clone()) + } + _ => None, + }) + .expect("promoted work stage should emit stage.completed"); + assert_eq!(completed.status, StageOutcome::Succeeded); + assert_eq!( + completed + .failure + .as_ref() + .map(|failure| failure.message.as_str()), + Some("forced work failure") + ); + assert!(!events.iter().any(|event| { + matches!(&event.body, EventBody::StageFailed(_)) && event.node_id.as_deref() == Some("work") + })); + assert!( + events + .iter() + .any(|event| matches!(&event.body, EventBody::RunCompleted(_))) + ); +} + +#[tokio::test] +async fn auto_status_true_is_an_alias_for_on_failure_succeed() { + let graph = on_failure_graph( + r#"graph [on_failure="exit"] + work [auto_status=true]"#, + ); + let run = run_on_failure(&graph, Emitter::default()).await; + + assert_eq!(run.outcome.status, StageOutcome::Succeeded); + assert_eq!(*run.visits.lock().unwrap(), vec!["work", "downstream"]); + let checkpoint = run + .state + .current_checkpoint() + .expect("downstream should be checkpointed"); + assert_eq!( + checkpoint.node_outcomes["work"].status, + StageOutcome::Succeeded + ); +} + +#[tokio::test] +async fn node_on_failure_succeed_prefers_explicit_failure_edge() { + let graph = on_failure_graph( + r#"work [on_failure="succeed"] + recovery + work -> recovery [condition="outcome=failed"] + recovery -> exit"#, + ); + let run = run_on_failure(&graph, Emitter::default()).await; + + assert_eq!(run.outcome.status, StageOutcome::Succeeded); + assert_eq!(*run.visits.lock().unwrap(), vec!["work", "recovery"]); + let checkpoint = run + .state + .current_checkpoint() + .expect("recovery should be checkpointed"); + assert!(checkpoint.node_outcomes["work"].status.is_failure()); +} + +#[tokio::test] +async fn node_on_failure_succeed_tests_recovery_edge_with_pending_result_context() { + let graph = on_failure_graph( + r#"work [on_failure="succeed"] + recovery + work -> recovery [condition="outcome=failed && context.recovery_ready=true && context.failure_class=deterministic"] + recovery -> exit"#, + ); + let run = run_on_failure(&graph, Emitter::default()).await; + + assert_eq!(run.outcome.status, StageOutcome::Succeeded); + assert_eq!(*run.visits.lock().unwrap(), vec!["work", "recovery"]); + let checkpoint = run + .state + .current_checkpoint() + .expect("recovery should be checkpointed"); + assert!(checkpoint.node_outcomes["work"].status.is_failure()); +} + +#[tokio::test] +async fn node_on_failure_succeed_satisfies_goal_gate_without_retry_target() { + let graph = + on_failure_graph(r#"work [on_failure="succeed" goal_gate=true retry_target="start"]"#); + let run = run_on_failure(&graph, Emitter::default()).await; + + assert_eq!(run.outcome.status, StageOutcome::Succeeded); + assert_eq!(*run.visits.lock().unwrap(), vec!["work", "downstream"]); +} + #[tokio::test] async fn goal_gate_routes_to_retry_target_on_failure() { // Pipeline: diff --git a/lib/foundation/fabro-core/Cargo.toml b/lib/foundation/fabro-core/Cargo.toml index 0c287f927..0b6579b0c 100644 --- a/lib/foundation/fabro-core/Cargo.toml +++ b/lib/foundation/fabro-core/Cargo.toml @@ -18,6 +18,7 @@ fabro-types = { path = "../fabro-types" } fabro-util = { path = "../fabro-util" } serde.workspace = true serde_json.workspace = true +strum.workspace = true thiserror.workspace = true tokio.workspace = true tokio-util.workspace = true diff --git a/lib/foundation/fabro-core/src/executor.rs b/lib/foundation/fabro-core/src/executor.rs index 7f976584b..d087ccc0c 100644 --- a/lib/foundation/fabro-core/src/executor.rs +++ b/lib/foundation/fabro-core/src/executor.rs @@ -9,13 +9,15 @@ use tokio_util::sync::CancellationToken; use crate::context::Context; use crate::error::{Error, Result, VisitLimitSource}; -use crate::graph::{EdgeSpec, Graph, NodeSpec}; +use crate::graph::{EdgeSelection, EdgeSpec, Graph, NodeSpec}; use crate::handler::NodeHandler; use crate::lifecycle::{ AttemptContext, AttemptResultContext, EdgeContext, EdgeDecision, NodeDecision, NoopLifecycle, RunLifecycle, }; -use crate::outcome::{NodeResult, NodeResultExt, Outcome, OutcomeMeta}; +use crate::outcome::{ + FailureDetail, NodeResult, NodeResultExt, Outcome, OutcomeMeta, StageOutcome, +}; use crate::state::ExecutionState; /// Build a [`NodeResult`] from an attempt outcome, pulling the inference and @@ -66,6 +68,34 @@ enum NextStep { End, } +#[derive(PartialEq)] +struct RoutingFingerprint { + status: StageOutcome, + preferred_label: Option, + suggested_next_ids: Vec, + context_updates: std::collections::HashMap, + jump_to_node: Option, + failure: Option, +} + +impl From<&Outcome> for RoutingFingerprint { + fn from(outcome: &Outcome) -> Self { + Self { + status: outcome.status, + preferred_label: outcome.preferred_label.clone(), + suggested_next_ids: outcome.suggested_next_ids.clone(), + context_updates: outcome.context_updates.clone(), + jump_to_node: outcome.jump_to_node.clone(), + failure: outcome.failure.clone(), + } + } +} + +struct PreparedEdgeSelection { + fingerprint: RoutingFingerprint, + selection: Option>, +} + pub struct ExecutorBuilder { handler: Arc>, lifecycle: Option>>, @@ -208,47 +238,56 @@ impl Executor { state.increment_visits(node.id()); // before_node lifecycle - let node_result = match self.lifecycle.before_node(&node, &state).await? { - NodeDecision::Skip(outcome) => { - let mut result = NodeResult::from_skip(*outcome); - self.lifecycle - .after_node(&node, &mut result, &state) - .await?; - result - } - NodeDecision::Block(msg) => { - return Err(Error::blocked(msg)); - } - NodeDecision::Continue => { - // Execute with retry, racing against stall token - let execution_result = if let Some(ref stall) = self.options.stall_token { - tokio::select! { - r = self.execute_with_retry(&node, &state, graph) => r, - () = stall.cancelled() => { - return Err(Error::StallTimeout { - node_id: node.id().to_string(), - }); + let (node_result, prepared_selection) = + match self.lifecycle.before_node(&node, &state).await? { + NodeDecision::Skip(outcome) => { + let mut result = NodeResult::from_skip(*outcome); + self.lifecycle + .after_node(&node, &mut result, &state) + .await?; + (result, None) + } + NodeDecision::Block(msg) => { + return Err(Error::blocked(msg)); + } + NodeDecision::Continue => { + // Execute with retry, racing against stall token + let execution_result = if let Some(ref stall) = self.options.stall_token { + tokio::select! { + r = self.execute_with_retry(&node, &state, graph) => r, + () = stall.cancelled() => { + return Err(Error::StallTimeout { + node_id: node.id().to_string(), + }); + } } + } else { + self.execute_with_retry(&node, &state, graph).await + }; + let mut result = match execution_result { + Ok(result) => result, + Err(Error::Cancelled) => { + state.cancelled = true; + let outcome = Outcome::fail("run cancelled"); + self.lifecycle.on_run_end(&outcome, &state).await; + return Err(Error::Cancelled); + } + Err(err) => return Err(err), + }; + let mut prepared_selection = self + .apply_succeed_policy(&node, &mut result, &state, graph) + .await?; + self.lifecycle + .after_node(&node, &mut result, &state) + .await?; + if prepared_selection.as_ref().is_some_and(|prepared| { + prepared.fingerprint != RoutingFingerprint::from(&result.outcome) + }) { + prepared_selection = None; } - } else { - self.execute_with_retry(&node, &state, graph).await - }; - let mut result = match execution_result { - Ok(result) => result, - Err(Error::Cancelled) => { - state.cancelled = true; - let outcome = Outcome::fail("run cancelled"); - self.lifecycle.on_run_end(&outcome, &state).await; - return Err(Error::Cancelled); - } - Err(err) => return Err(err), - }; - self.lifecycle - .after_node(&node, &mut result, &state) - .await?; - result - } - }; + (result, prepared_selection) + } + }; state.record(node.id(), &node_result); self.lifecycle @@ -258,7 +297,7 @@ impl Executor { // Determine next step let last_outcome = &state.node_outcomes[node.id()]; let next = self - .resolve_next_step(&node, last_outcome, &state, graph) + .resolve_next_step(&node, last_outcome, &state, graph, prepared_selection) .await?; // Checkpoint AFTER edge selection so next_node_id is known @@ -285,7 +324,10 @@ impl Executor { if outcome.status.is_failure() { let resolved = graph.resolve_on_failure(&node); let message = match resolved.policy() { - OnFailure::Route => { + // A failed outcome under `succeed` only reaches + // the end when an explicit route matched but + // produced no next node, which mirrors `route`. + OnFailure::Route | OnFailure::Succeed => { format!("stage {} failed with no outgoing fail edge", node.id()) } OnFailure::Exit => format!( @@ -423,12 +465,65 @@ impl Executor { unreachable!("loop always returns or continues") } + /// Applies the `on_failure="succeed"` policy to a failed node result. + /// + /// This runs before the lifecycle observes the result, so the recorded + /// outcome, context keys, goal gates, events, and routing all see the + /// effective outcome. Explicit recovery routes take priority: a failed + /// outcome that carries a jump, or that an explicit edge would route, + /// stays `failed`. + async fn apply_succeed_policy( + &self, + node: &G::Node, + result: &mut NodeResult, + state: &ExecutionState, + graph: &G, + ) -> Result>> { + let outcome = &result.outcome; + if !outcome.status.is_failure() || outcome.jump_to_node.is_some() { + return Ok(None); + } + let resolved = graph.resolve_on_failure(node); + if resolved.policy() != OnFailure::Succeed { + return Ok(None); + } + let projected_context = state.context.fork(); + projected_context.apply_updates(&result.outcome.context_updates); + graph.project_result_context(node, result, &projected_context); + let routing_context = self + .handler + .context_for_edge_selection(&projected_context, graph) + .await?; + if let Some(selection) = graph + .select_edge(node, outcome, &routing_context) + .filter(|selection| selection.reason.is_explicit()) + { + return Ok(Some(PreparedEdgeSelection { + fingerprint: RoutingFingerprint::from(&result.outcome), + selection: Some(selection), + })); + } + if result.outcome.apply_on_failure(resolved) { + tracing::debug!( + node = %node.id(), + scope = %resolved.scope(), + "on_failure=succeed promoted failed outcome" + ); + } + graph.project_result_context(node, result, &routing_context); + Ok(Some(PreparedEdgeSelection { + fingerprint: RoutingFingerprint::from(&result.outcome), + selection: graph.select_edge(node, &result.outcome, &routing_context), + })) + } + async fn resolve_next_step( &self, node: &G::Node, outcome: &Outcome, state: &ExecutionState, graph: &G, + prepared_selection: Option>, ) -> Result { // Jump takes priority if let Some(ref target) = outcome.jump_to_node { @@ -447,14 +542,26 @@ impl Executor { } } - // Normal edge selection - let routing_context = self - .handler - .context_for_edge_selection(&state.context, graph) - .await?; - if let Some(selection) = graph.select_edge(node, outcome, &routing_context) { + // Normal edge selection. A failed `succeed` result prepared this + // decision while its original failure context was still available. + let selection = if let Some(prepared) = prepared_selection { + prepared.selection + } else { + let routing_context = self + .handler + .context_for_edge_selection(&state.context, graph) + .await?; + graph.select_edge(node, outcome, &routing_context) + } + .filter(|selection| { + !outcome.status.is_failure() + || selection.reason.is_explicit() + || graph.resolve_on_failure(node).policy() == OnFailure::Route + }); + if let Some(selection) = selection { let target = selection.edge.target().to_string(); let is_restart = selection.edge.is_loop_restart(); + let reason: &'static str = selection.reason.into(); let ctx = EdgeContext { from: node.id(), @@ -462,7 +569,7 @@ impl Executor { edge: Some(selection.edge.clone()), is_jump: false, outcome, - reason: selection.reason, + reason, }; match self.lifecycle.on_edge_selected(&ctx, state).await? { EdgeDecision::Continue => { @@ -2324,6 +2431,317 @@ mod tests { assert!(state.node_outcomes.contains_key("downstream")); } + /// Records the outcome status each lifecycle callback observed, so tests + /// can prove the `succeed` policy is applied before `after_node`. + struct StatusCaptureLifecycle(Arc>>); + + #[async_trait] + impl RunLifecycle for StatusCaptureLifecycle { + async fn after_node( + &self, + node: &TestNode, + result: &mut NodeResult, + _state: &ExecutionState, + ) -> Result<()> { + self.0 + .lock() + .unwrap() + .push((node.id().to_string(), result.outcome.status)); + Ok(()) + } + } + + fn succeed_policy_graph() -> TestGraph { + TestGraph::new( + vec![ + TestNode::new("work").with_on_failure(OnFailure::Succeed), + TestNode::new("downstream"), + TestNode::terminal("end"), + ], + vec![ + TestEdge::new("work", "downstream"), + TestEdge::new("downstream", "end"), + ], + "work", + ) + } + + fn fail_work_handler() -> Arc> { + Arc::new( + DispatchHandler::new(Arc::new(AlwaysSucceedHandler)) + .with_handler("work", Arc::new(AlwaysFailHandler::new("boom"))), + ) + } + + #[tokio::test] + async fn executor_succeed_policy_promotes_failed_node_before_lifecycle_and_continues() { + let graph = succeed_policy_graph(); + let state = ExecutionState::new(&graph).unwrap(); + let seen = Arc::new(Mutex::new(Vec::new())); + let executor = ExecutorBuilder::new(fail_work_handler()) + .lifecycle(Box::new(StatusCaptureLifecycle(Arc::clone(&seen)))) + .build(); + + let (outcome, state) = executor.run(&graph, state).await.unwrap(); + + assert_eq!(outcome.status, StageOutcome::Succeeded); + let work = &state.node_outcomes["work"]; + assert_eq!(work.status, StageOutcome::Succeeded); + assert_eq!( + work.failure + .as_ref() + .map(|failure| failure.message.as_str()), + Some("boom"), + "the original failure stays on the recorded outcome" + ); + assert_eq!( + work.notes.as_deref(), + Some("node on_failure=succeed promoted a failed outcome to succeeded") + ); + assert!(state.node_outcomes.contains_key("downstream")); + assert_eq!( + seen.lock().unwrap().clone(), + vec![ + ("work".to_string(), StageOutcome::Succeeded), + ("downstream".to_string(), StageOutcome::Succeeded), + ], + "after_node observes the effective outcome" + ); + } + + #[tokio::test] + async fn executor_succeed_policy_resolves_routing_context_once_per_node() { + struct CountingHandler(Arc); + + #[async_trait] + impl NodeHandler for CountingHandler { + async fn execute( + &self, + node: &TestNode, + _context: &Context, + _graph: &TestGraph, + ) -> Result { + if node.id() == "work" { + Ok(Outcome::fail("boom")) + } else { + Ok(Outcome::success()) + } + } + + async fn context_for_edge_selection( + &self, + context: &Context, + _graph: &TestGraph, + ) -> Result { + self.0.fetch_add(1, Ordering::SeqCst); + Ok(context.clone()) + } + } + + let graph = succeed_policy_graph(); + let state = ExecutionState::new(&graph).unwrap(); + let calls = Arc::new(AtomicU32::new(0)); + let executor = ExecutorBuilder::new(Arc::new(CountingHandler(Arc::clone(&calls)))).build(); + + let (outcome, _) = executor.run(&graph, state).await.unwrap(); + + assert_eq!(outcome.status, StageOutcome::Succeeded); + assert_eq!(calls.load(Ordering::SeqCst), 2); + } + + #[tokio::test] + async fn executor_succeed_policy_keeps_failed_outcome_when_explicit_route_matches() { + let graph = TestGraph::new( + vec![ + TestNode::new("work").with_on_failure(OnFailure::Succeed), + TestNode::new("recovery"), + TestNode::new("downstream"), + TestNode::terminal("end"), + ], + vec![ + TestEdge::new("work", "recovery").with_label("failed"), + TestEdge::new("work", "downstream"), + TestEdge::new("recovery", "end"), + TestEdge::new("downstream", "end"), + ], + "work", + ); + let state = ExecutionState::new(&graph).unwrap(); + let executor = ExecutorBuilder::new(fail_work_handler()).build(); + + let (outcome, state) = executor.run(&graph, state).await.unwrap(); + + assert_eq!(outcome.status, StageOutcome::Succeeded); + assert_eq!(state.node_outcomes["work"].status, StageOutcome::Failed { + retry_requested: false, + }); + assert!(state.node_outcomes.contains_key("recovery")); + assert!(!state.node_outcomes.contains_key("downstream")); + } + + #[tokio::test] + async fn executor_succeed_policy_keeps_failed_outcome_with_jump() { + struct FailWithJump; + + #[async_trait] + impl NodeHandler for FailWithJump { + async fn execute( + &self, + _node: &TestNode, + _context: &Context, + _graph: &TestGraph, + ) -> Result { + let mut outcome = Outcome::fail("boom"); + outcome.jump_to_node = Some("recovery".to_string()); + Ok(outcome) + } + } + + let graph = TestGraph::new( + vec![ + TestNode::new("work").with_on_failure(OnFailure::Succeed), + TestNode::new("recovery"), + TestNode::new("downstream"), + TestNode::terminal("end"), + ], + vec![ + TestEdge::new("work", "downstream"), + TestEdge::new("recovery", "end"), + TestEdge::new("downstream", "end"), + ], + "work", + ); + let state = ExecutionState::new(&graph).unwrap(); + let handler = DispatchHandler::new(Arc::new(AlwaysSucceedHandler)) + .with_handler("work", Arc::new(FailWithJump)); + let executor = + ExecutorBuilder::new(Arc::new(handler) as Arc>).build(); + + let (outcome, state) = executor.run(&graph, state).await.unwrap(); + + assert_eq!(outcome.status, StageOutcome::Succeeded); + assert!(state.node_outcomes["work"].status.is_failure()); + assert!(state.node_outcomes.contains_key("recovery")); + assert!(!state.node_outcomes.contains_key("downstream")); + } + + #[tokio::test] + async fn executor_succeed_policy_satisfies_goal_gate() { + let graph = TestGraph::new( + vec![ + TestNode::new("work").with_on_failure(OnFailure::Succeed), + TestNode::terminal("end").with_goal_gate("work", StageOutcome::Succeeded), + ], + vec![TestEdge::new("work", "end")], + "work", + ); + let state = ExecutionState::new(&graph).unwrap(); + let executor = ExecutorBuilder::new(fail_work_handler()).build(); + + let (outcome, _state) = executor.run(&graph, state).await.unwrap(); + + assert_eq!(outcome.status, StageOutcome::Succeeded); + } + + #[tokio::test] + async fn executor_succeed_policy_skips_retry_target() { + let graph = TestGraph::new( + vec![ + TestNode::new("work").with_on_failure(OnFailure::Succeed), + TestNode::new("downstream"), + TestNode::new("retry_only"), + TestNode::terminal("end"), + ], + vec![ + TestEdge::new("work", "downstream"), + TestEdge::new("downstream", "end"), + TestEdge::new("retry_only", "end"), + ], + "work", + ) + .with_retry_target("work", "retry_only"); + let state = ExecutionState::new(&graph).unwrap(); + let executor = ExecutorBuilder::new(fail_work_handler()).build(); + + let (outcome, state) = executor.run(&graph, state).await.unwrap(); + + assert_eq!(outcome.status, StageOutcome::Succeeded); + assert!(state.node_outcomes.contains_key("downstream")); + assert!( + !state.node_outcomes.contains_key("retry_only"), + "a promoted outcome is not failed, so retry targets do not apply" + ); + } + + #[tokio::test] + async fn executor_graph_succeed_policy_promotes_every_failed_node() { + let graph = TestGraph::new( + vec![ + TestNode::new("work"), + TestNode::new("downstream"), + TestNode::terminal("end"), + ], + vec![ + TestEdge::new("work", "downstream"), + TestEdge::new("downstream", "end"), + ], + "work", + ) + .with_on_failure(OnFailure::Succeed); + let state = ExecutionState::new(&graph).unwrap(); + let executor = ExecutorBuilder::new( + Arc::new(AlwaysFailHandler::new("boom")) as Arc> + ) + .build(); + + let (outcome, state) = executor.run(&graph, state).await.unwrap(); + + assert_eq!(outcome.status, StageOutcome::Succeeded); + for node_id in ["work", "downstream"] { + let recorded = &state.node_outcomes[node_id]; + assert_eq!(recorded.status, StageOutcome::Succeeded); + assert_eq!( + recorded.notes.as_deref(), + Some("graph on_failure=succeed promoted a failed outcome to succeeded") + ); + } + } + + #[tokio::test] + async fn executor_succeed_policy_leaves_partial_outcome_unchanged() { + struct PartialHandler; + + #[async_trait] + impl NodeHandler for PartialHandler { + async fn execute( + &self, + _node: &TestNode, + _context: &Context, + _graph: &TestGraph, + ) -> Result { + let mut outcome = Outcome::success(); + outcome.status = StageOutcome::PartiallySucceeded; + Ok(outcome) + } + } + + let graph = succeed_policy_graph(); + let state = ExecutionState::new(&graph).unwrap(); + let handler = DispatchHandler::new(Arc::new(AlwaysSucceedHandler)) + .with_handler("work", Arc::new(PartialHandler)); + let executor = + ExecutorBuilder::new(Arc::new(handler) as Arc>).build(); + + let (outcome, state) = executor.run(&graph, state).await.unwrap(); + + assert_eq!(outcome.status, StageOutcome::Succeeded); + assert_eq!( + state.node_outcomes["work"].status, + StageOutcome::PartiallySucceeded + ); + assert_eq!(state.node_outcomes["work"].notes, None); + } + #[tokio::test] async fn executor_goal_gate_retry_target_to_terminal_fails_without_looping() { let terminal_visits = Arc::new(AtomicU32::new(0)); diff --git a/lib/foundation/fabro-core/src/graph.rs b/lib/foundation/fabro-core/src/graph.rs index e39ccd593..9688e81e1 100644 --- a/lib/foundation/fabro-core/src/graph.rs +++ b/lib/foundation/fabro-core/src/graph.rs @@ -4,7 +4,26 @@ use fabro_types::ResolvedOnFailure; use crate::context::Context; use crate::error::Result; -use crate::outcome::{Outcome, OutcomeMeta}; +use crate::outcome::{NodeResult, Outcome, OutcomeMeta}; + +/// How edge selection chose an edge. +#[derive(Debug, Clone, Copy, PartialEq, Eq, strum::Display, strum::IntoStaticStr)] +#[strum(serialize_all = "snake_case")] +pub enum EdgeSelectionReason { + Condition, + PreferredLabel, + SuggestedNext, + Unconditional, +} + +impl EdgeSelectionReason { + /// Returns whether this selection is an explicit route supplied by the + /// node result or an edge condition. + #[must_use] + pub const fn is_explicit(self) -> bool { + !matches!(self, Self::Unconditional) + } +} pub trait NodeSpec: Send + Sync + Clone { fn id(&self) -> &str; @@ -20,7 +39,7 @@ pub trait EdgeSpec: Send + Sync + Clone { pub struct EdgeSelection { pub edge: G::Edge, - pub reason: &'static str, + pub reason: EdgeSelectionReason, } pub trait Graph: Send + Sync { @@ -37,6 +56,19 @@ pub trait Graph: Send + Sync { outcome: &Outcome, context: &Context, ) -> Option>; + /// Projects derived values from a pending node result into a context used + /// to test edge conditions before the result is durably recorded. + /// + /// Implementations can add the same derived values that their lifecycle + /// writes after recording. The executor applies `context_updates` before + /// this method runs. + fn project_result_context( + &self, + _node: &Self::Node, + _result: &NodeResult, + _context: &Context, + ) { + } fn check_goal_gates( &self, outcomes: &HashMap>, diff --git a/lib/foundation/fabro-core/src/lib.rs b/lib/foundation/fabro-core/src/lib.rs index 853d4eeaf..70f1fa6d7 100644 --- a/lib/foundation/fabro-core/src/lib.rs +++ b/lib/foundation/fabro-core/src/lib.rs @@ -15,7 +15,7 @@ pub mod test_fixtures; pub use context::Context; pub use error::{Error, HandlerErrorDetail, Result, VisitLimitSource}; pub use executor::{Executor, ExecutorBuilder, ExecutorOptions}; -pub use graph::{EdgeSelection, EdgeSpec, Graph, NodeSpec}; +pub use graph::{EdgeSelection, EdgeSelectionReason, EdgeSpec, Graph, NodeSpec}; pub use handler::NodeHandler; pub use lifecycle::{ AttemptContext, AttemptResultContext, CompositeLifecycle, EdgeContext, EdgeDecision, diff --git a/lib/foundation/fabro-core/src/test_fixtures.rs b/lib/foundation/fabro-core/src/test_fixtures.rs index 1a503deb3..1beb70cc3 100644 --- a/lib/foundation/fabro-core/src/test_fixtures.rs +++ b/lib/foundation/fabro-core/src/test_fixtures.rs @@ -7,7 +7,7 @@ use fabro_types::{OnFailure, ResolvedOnFailure}; use crate::context::Context; use crate::error::{Error, HandlerErrorDetail, Result}; -use crate::graph::{EdgeSelection, EdgeSpec, Graph, NodeSpec}; +use crate::graph::{EdgeSelection, EdgeSelectionReason, EdgeSpec, Graph, NodeSpec}; use crate::handler::NodeHandler; use crate::outcome::{FailureCategory, FailureDetail, Outcome, StageOutcome}; use crate::retry::RetryPolicy; @@ -199,7 +199,7 @@ impl Graph for TestGraph { { return Some(EdgeSelection { edge: e.clone(), - reason: "preferred_label", + reason: EdgeSelectionReason::PreferredLabel, }); } } @@ -212,7 +212,7 @@ impl Graph for TestGraph { { return Some(EdgeSelection { edge: e.clone(), - reason: "condition", + reason: EdgeSelectionReason::Condition, }); } @@ -221,21 +221,16 @@ impl Graph for TestGraph { if let Some(e) = edges.iter().find(|e| e.to == *suggested) { return Some(EdgeSelection { edge: e.clone(), - reason: "suggested_next", + reason: EdgeSelectionReason::SuggestedNext, }); } } - if outcome.status.is_failure() && self.resolve_on_failure(node).policy() == OnFailure::Exit - { - return None; - } - // Fourth: unconditional (no label) if let Some(e) = edges.iter().find(|e| e.label.is_none()) { return Some(EdgeSelection { edge: e.clone(), - reason: "unconditional", + reason: EdgeSelectionReason::Unconditional, }); } @@ -532,7 +527,7 @@ mod tests { let ctx = Context::new(); let sel = g.select_edge(&node, &outcome, &ctx).unwrap(); assert_eq!(sel.edge.target(), "b"); - assert_eq!(sel.reason, "condition"); + assert_eq!(sel.reason, EdgeSelectionReason::Condition); } #[test] @@ -543,7 +538,7 @@ mod tests { let ctx = Context::new(); let sel = g.select_edge(&node, &outcome, &ctx).unwrap(); assert_eq!(sel.edge.target(), "end"); - assert_eq!(sel.reason, "unconditional"); + assert_eq!(sel.reason, EdgeSelectionReason::Unconditional); } #[test] diff --git a/lib/foundation/fabro-types/src/graph.rs b/lib/foundation/fabro-types/src/graph.rs index 4b4f18233..3cbbd4906 100644 --- a/lib/foundation/fabro-types/src/graph.rs +++ b/lib/foundation/fabro-types/src/graph.rs @@ -6,7 +6,11 @@ use strum::VariantNames; use crate::AgentBackend; -/// Policy for routing a failed node when no explicit recovery edge matches. +/// Policy for a failed node when no explicit recovery route matches. +/// +/// Explicit routes (a jump, a matching edge condition, a matching preferred +/// label, or a matching suggested next node) take priority under every +/// policy. The policy decides what happens when none of them match. #[derive( Debug, Clone, @@ -24,9 +28,15 @@ use crate::AgentBackend; #[serde(rename_all = "snake_case")] #[strum(serialize_all = "snake_case")] pub enum OnFailure { + /// The outcome stays `failed` and may take an unconditional edge. #[default] Route, + /// The outcome stays `failed` and skips the unconditional edge, so the + /// run ends unless a retry target applies. Exit, + /// The outcome becomes `succeeded` and follows normal success routing. + /// The original failure details stay on the outcome for observability. + Succeed, } impl OnFailure { @@ -349,13 +359,18 @@ impl Node { self.str_attr("fallback_retry_target") } - /// Node-level failure routing policy override. `None` means the node - /// inherits the graph-level policy. Invalid values are rejected during - /// workflow validation, so runtime resolution treats them as absent. + /// Node-level failure policy override. `None` means the node inherits + /// the graph-level policy. Invalid values are rejected during workflow + /// validation, so runtime resolution treats them as absent. + /// + /// The deprecated `auto_status=true` attribute is a compatibility alias + /// for `on_failure="succeed"`. An explicit `on_failure` attribute wins. #[must_use] pub fn on_failure(&self) -> Option { - self.str_attr("on_failure") - .and_then(|value| value.parse().ok()) + match self.attrs.get("on_failure") { + Some(value) => value.as_str().and_then(|value| value.parse().ok()), + None => self.auto_status().then_some(OnFailure::Succeed), + } } #[must_use] @@ -392,6 +407,8 @@ impl Node { self.str_attr("speed") } + /// Deprecated spelling of `on_failure="succeed"`. Validation warns when + /// it is present; [`Node::on_failure`] resolves the alias at runtime. #[must_use] pub fn auto_status(&self) -> bool { self.bool_attr("auto_status").unwrap_or(false) @@ -614,7 +631,7 @@ impl Graph { .and_then(AttrValue::as_str) } - /// Graph-level failure routing policy. Invalid values are rejected during + /// Graph-level failure policy. Invalid values are rejected during /// workflow validation, so runtime resolution can use the compatibility /// default. #[must_use] @@ -626,7 +643,7 @@ impl Graph { .unwrap_or_default() } - /// Effective failure routing policy for a node. A node-level `on_failure` + /// Effective failure policy for a node. A node-level `on_failure` /// attribute overrides the graph level; an absent (or invalid, hence /// validation-rejected) node attribute inherits the graph policy. #[must_use] @@ -770,9 +787,11 @@ mod tests { fn on_failure_parses_and_displays_supported_values() { assert_eq!("route".parse::().unwrap(), OnFailure::Route); assert_eq!("exit".parse::().unwrap(), OnFailure::Exit); + assert_eq!("succeed".parse::().unwrap(), OnFailure::Succeed); assert_eq!(OnFailure::Route.to_string(), "route"); assert_eq!(OnFailure::Exit.to_string(), "exit"); - assert_eq!(OnFailure::expected_values(), "route, exit"); + assert_eq!(OnFailure::Succeed.to_string(), "succeed"); + assert_eq!(OnFailure::expected_values(), "route, exit, succeed"); } #[test] @@ -815,6 +834,43 @@ mod tests { assert_eq!(node.on_failure(), None); } + #[test] + fn node_auto_status_is_an_alias_for_on_failure_succeed() { + let mut node = Node::new("work"); + node.attrs + .insert("auto_status".to_string(), AttrValue::Boolean(true)); + assert!(node.auto_status()); + assert_eq!(node.on_failure(), Some(OnFailure::Succeed)); + + // An explicit on_failure attribute wins over the alias. + node.attrs.insert( + "on_failure".to_string(), + AttrValue::String("exit".to_string()), + ); + assert_eq!(node.on_failure(), Some(OnFailure::Exit)); + + // auto_status=false does not set a policy. + let mut node = Node::new("work"); + node.attrs + .insert("auto_status".to_string(), AttrValue::Boolean(false)); + assert_eq!(node.on_failure(), None); + } + + #[test] + fn explicit_invalid_on_failure_does_not_fall_back_to_auto_status() { + for value in [ + AttrValue::String("invalid".to_string()), + AttrValue::Boolean(true), + ] { + let mut node = Node::new("work"); + node.attrs + .insert("auto_status".to_string(), AttrValue::Boolean(true)); + node.attrs.insert("on_failure".to_string(), value); + + assert_eq!(node.on_failure(), None); + } + } + #[test] fn resolve_on_failure_prefers_node_policy_over_graph_policy() { let mut graph = Graph::new("test"); diff --git a/lib/foundation/fabro-types/src/outcome.rs b/lib/foundation/fabro-types/src/outcome.rs index f522f73e7..4aadc9aac 100644 --- a/lib/foundation/fabro-types/src/outcome.rs +++ b/lib/foundation/fabro-types/src/outcome.rs @@ -8,7 +8,9 @@ use serde::{Deserialize, Deserializer, Serialize, Serializer}; use serde_json::Value; use strum::{Display, EnumString, IntoStaticStr}; -use crate::{ExecOutputTail, FailureSignature, StageTiming, SystemActorKind}; +use crate::{ + ExecOutputTail, FailureSignature, OnFailure, ResolvedOnFailure, StageTiming, SystemActorKind, +}; pub trait OutcomeMeta: Default + Clone + Send + Sync + fmt::Debug + Serialize + DeserializeOwned + 'static @@ -329,13 +331,90 @@ impl Outcome { ..Self::default() } } + + /// Applies a resolved failure policy to this outcome. + /// + /// `succeed` promotes a `failed` outcome and records the policy scope. + /// The original `failure` stays available for durable diagnostics. Other + /// policies and statuses do not change the outcome. + pub fn apply_on_failure(&mut self, policy: ResolvedOnFailure) -> bool { + if policy.policy() != OnFailure::Succeed || !self.status.is_failure() { + return false; + } + self.status = StageOutcome::Succeeded; + let note = format!( + "{} on_failure=succeed promoted a failed outcome to succeeded", + policy.scope() + ); + self.notes = Some(match self.notes.take() { + Some(existing) => format!("{existing}\n{note}"), + None => note, + }); + true + } } #[cfg(test)] mod tests { use serde_json::json; - use super::{FailureCategory, FailureDetail, StageOutcome, StageState}; + use super::{FailureCategory, FailureDetail, Outcome, StageOutcome, StageState}; + use crate::{OnFailure, ResolvedOnFailure}; + + #[test] + fn apply_on_failure_keeps_failure_and_records_scope() { + let mut outcome: Outcome = Outcome::fail("boom"); + assert!(outcome.apply_on_failure(ResolvedOnFailure::node(OnFailure::Succeed))); + + assert_eq!(outcome.status, StageOutcome::Succeeded); + assert_eq!( + outcome + .failure + .as_ref() + .map(|failure| failure.message.as_str()), + Some("boom") + ); + assert_eq!( + outcome.notes.as_deref(), + Some("node on_failure=succeed promoted a failed outcome to succeeded") + ); + } + + #[test] + fn apply_on_failure_appends_to_existing_notes() { + let mut outcome: Outcome = Outcome::fail("boom"); + outcome.notes = Some("handler note".to_string()); + assert!(outcome.apply_on_failure(ResolvedOnFailure::graph(OnFailure::Succeed))); + + assert_eq!( + outcome.notes.as_deref(), + Some("handler note\ngraph on_failure=succeed promoted a failed outcome to succeeded") + ); + } + + #[test] + fn apply_on_failure_ignores_non_failed_outcomes() { + let mut partial: Outcome = Outcome::success(); + partial.status = StageOutcome::PartiallySucceeded; + let mut skipped: Outcome = Outcome::skipped("not needed"); + + for outcome in [&mut partial, &mut skipped] { + let before = outcome.clone(); + assert!(!outcome.apply_on_failure(ResolvedOnFailure::node(OnFailure::Succeed))); + assert_eq!(*outcome, before); + } + } + + #[test] + fn apply_on_failure_ignores_non_succeed_policies() { + for policy in [OnFailure::Route, OnFailure::Exit] { + let mut outcome: Outcome = Outcome::fail("boom"); + let before = outcome.clone(); + + assert!(!outcome.apply_on_failure(ResolvedOnFailure::node(policy))); + assert_eq!(outcome, before); + } + } #[test] fn stage_outcome_failed_serde_is_lossy_for_retry_intent() { diff --git a/test/auto_status_deprecated.fabro b/test/auto_status_deprecated.fabro new file mode 100644 index 000000000..b5a540a67 --- /dev/null +++ b/test/auto_status_deprecated.fabro @@ -0,0 +1,7 @@ +digraph DeprecatedAutoStatus { + start [shape=Mdiamond] + exit [shape=Msquare] + scan [prompt="Scan the tree" auto_status=true] + + start -> scan -> exit +}