From 588515dbf6afd6e0c3830d89357dde029de187a2 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Mon, 23 Mar 2026 14:59:35 -0400 Subject: [PATCH] Attractor spec hunks 14 & 16: remove error_policy and k_of_n/quorum from parallel handler Remove ErrorPolicy enum (continue/fail_fast/ignore) and the k_of_n/quorum join policies from the parallel handler, leaving only wait_all and first_success. This deletes ~180 lines of conditional logic including FailFast early termination, the ParallelEarlyTermination event, and all related tests and documentation. Co-Authored-By: Claude Opus 4.6 (1M context) --- docs-internal/demo/06-parallel.fabro | 2 +- docs-internal/demo/11-ensemble.fabro | 2 +- docs-internal/events-strategy.md | 3 +- docs/execution/observability.mdx | 2 +- docs/execution/outcomes.mdx | 2 +- docs/reference/dot-language.mdx | 3 +- docs/tutorials/ensemble.mdx | 9 +- docs/tutorials/parallel-review.mdx | 19 +- docs/workflows/stages-and-nodes.mdx | 13 +- .../fabro-cli/src/commands/run_progress.rs | 6 +- lib/crates/fabro-workflows/README.md | 2 +- lib/crates/fabro-workflows/src/event.rs | 48 +---- .../fabro-workflows/src/handler/parallel.rs | 179 +----------------- .../references/dot-language.md | 2 +- .../references/example-workflows.md | 4 +- test/docs/tutorials/ensemble/ensemble.fabro | 2 +- .../tutorials/parallel-review/parallel.fabro | 2 +- .../stages-and-nodes/all-node-types.fabro | 2 +- test/parallel.fabro | 2 +- 19 files changed, 30 insertions(+), 274 deletions(-) diff --git a/docs-internal/demo/06-parallel.fabro b/docs-internal/demo/06-parallel.fabro index 6ef15e16f..7f7191a3c 100644 --- a/docs-internal/demo/06-parallel.fabro +++ b/docs-internal/demo/06-parallel.fabro @@ -5,7 +5,7 @@ digraph Parallel { start [shape=Mdiamond, label="Start"] exit [shape=Msquare, label="Exit"] - fork [label="Fork Analysis", shape=component, join_policy="wait_all", error_policy="continue"] + fork [label="Fork Analysis", shape=component, join_policy="wait_all"] security [label="Security Audit", prompt="Examine the codebase for security concerns: hardcoded secrets, injection risks, unsafe dependencies. List findings as bullet points.", shape=tab, reasoning_effort="low"] architecture [label="Architecture Review", prompt="Assess the codebase architecture: separation of concerns, dependency structure, modularity. List findings as bullet points.", shape=tab, reasoning_effort="low"] diff --git a/docs-internal/demo/11-ensemble.fabro b/docs-internal/demo/11-ensemble.fabro index 8af88cb36..3a8655cde 100644 --- a/docs-internal/demo/11-ensemble.fabro +++ b/docs-internal/demo/11-ensemble.fabro @@ -14,7 +14,7 @@ digraph Ensemble { start [shape=Mdiamond, label="Start"] exit [shape=Msquare, label="Exit"] - fork [label="Fan Out", shape=component, join_policy="wait_all", error_policy="continue"] + fork [label="Fan Out", shape=component, join_policy="wait_all"] opus [label="Opus", prompt="Analyze the goal. Provide your independent assessment, recommendations, and any code or prose needed. Be thorough.", shape=tab] gemini [label="Gemini", prompt="Analyze the goal. Provide your independent assessment, recommendations, and any code or prose needed. Be thorough.", shape=tab] diff --git a/docs-internal/events-strategy.md b/docs-internal/events-strategy.md index 564692cac..37c46ca09 100644 --- a/docs-internal/events-strategy.md +++ b/docs-internal/events-strategy.md @@ -172,11 +172,10 @@ WorkflowRunEvent::MyNewEvent { node_id, duration_ms, .. } => { | Event | JSONL fields | |---|---| -| `ParallelStarted` | `branch_count`, `join_policy`, `error_policy` | +| `ParallelStarted` | `branch_count`, `join_policy` | | `ParallelBranchStarted` | `node_id`, `node_label`, `branch_index` | | `ParallelBranchCompleted` | `node_id`, `node_label`, `branch_index`, `duration_ms`, `status` | | `ParallelCompleted` | `duration_ms`, `success_count`, `failure_count` | -| `ParallelEarlyTermination` | `reason`, `completed_count`, `pending_count` | ### Graph navigation diff --git a/docs/execution/observability.mdx b/docs/execution/observability.mdx index 0956fea7c..482baf897 100644 --- a/docs/execution/observability.mdx +++ b/docs/execution/observability.mdx @@ -75,7 +75,7 @@ Events fall into several categories: | Event | Key fields | Description | |---|---|---| -| `ParallelStarted` | `branch_count`, `join_policy`, `error_policy` | Fan-out begins | +| `ParallelStarted` | `branch_count`, `join_policy` | Fan-out begins | | `ParallelBranchStarted` | `branch`, `index` | Individual branch begins | | `ParallelBranchCompleted` | `branch`, `duration_ms`, `status` | Branch finishes | | `ParallelCompleted` | `duration_ms`, `success_count`, `failure_count` | All branches done | diff --git a/docs/execution/outcomes.mdx b/docs/execution/outcomes.mdx index 741c7ba0a..1e5ad4613 100644 --- a/docs/execution/outcomes.mdx +++ b/docs/execution/outcomes.mdx @@ -27,7 +27,7 @@ Each node type has its own rules for which statuses it can return: |---|---|---| | **Command** | `success`, `fail` | `success` when exit code is 0; `fail` otherwise | | **Agent / Prompt** | `success`, `fail`, `partial_success`, `retry`, `skipped` | Defaults to `success`. The LLM can set any status via a [routing directive](/agents/outputs#routing-directives) JSON object in its response. Backend errors produce `retry` (if retryable) or `fail`. | -| **Parallel** | `success`, `partial_success`, `fail` | Depends on the `join_policy`. `wait_all`: `success` if no failures, `partial_success` if some branches failed. `first_success` / `k_of_n` / `quorum`: `success` if threshold met, else `fail`. | +| **Parallel** | `success`, `partial_success`, `fail` | Depends on the `join_policy`. `wait_all`: `success` if no failures, `partial_success` if some branches failed. `first_success`: `success` if threshold met, else `fail`. | | **Human** | `success` | Always succeeds — the user's selection becomes a routing signal via `preferred_label` | | **Conditional** | `success` | Always succeeds — routing is handled by the engine's edge selection | | **Start / Exit / Wait** | `success` | Always succeed | diff --git a/docs/reference/dot-language.mdx b/docs/reference/dot-language.mdx index 6962bcb90..617d4313c 100644 --- a/docs/reference/dot-language.mdx +++ b/docs/reference/dot-language.mdx @@ -219,8 +219,7 @@ Start nodes can also be identified by ID (`start` or `Start`). Exit nodes can be | Attribute | Type | Description | |---|---|---| -| `join_policy` | String | When the merge can proceed: `wait_all` (default), `first_success`, `k_of_n(N)`, `quorum(F)` | -| `error_policy` | String | How branch failures are handled: `continue` (default), `fail_fast`, `ignore` | +| `join_policy` | String | When the merge can proceed: `wait_all` (default), `first_success` | | `max_parallel` | Integer | Maximum concurrent branches (default: 4) | ### Wait nodes diff --git a/docs/tutorials/ensemble.mdx b/docs/tutorials/ensemble.mdx index 24f80a2e7..6e4241d38 100644 --- a/docs/tutorials/ensemble.mdx +++ b/docs/tutorials/ensemble.mdx @@ -28,7 +28,7 @@ digraph Ensemble { start [shape=Mdiamond, label="Start"] exit [shape=Msquare, label="Exit"] - fork [label="Fan Out", shape=component, join_policy="wait_all", error_policy="continue"] + fork [label="Fan Out", shape=component, join_policy="wait_all"] opus [label="Opus", prompt="Analyze the goal. Provide your independent assessment, recommendations, and any code or prose needed. Be thorough.", shape=tab] gemini [label="Gemini", prompt="Analyze the goal. Provide your independent assessment, recommendations, and any code or prose needed. Be thorough.", shape=tab] @@ -56,7 +56,7 @@ fabro run files-internal/demo/11-ensemble.fabro ``` -This workflow requires API keys for all four providers (`ANTHROPIC_API_KEY`, `GEMINI_API_KEY`, `OPENAI_API_KEY`, `INCEPTION_API_KEY`). If a provider key is missing, that branch will fail — but `error_policy="continue"` ensures the other branches still complete. +This workflow requires API keys for all four providers (`ANTHROPIC_API_KEY`, `GEMINI_API_KEY`, `OPENAI_API_KEY`, `INCEPTION_API_KEY`). If a provider key is missing, that branch will fail — but the remaining branches still complete. ## How it works @@ -78,7 +78,7 @@ Each branch receives the same prompt but runs on a completely different model. T ### 2. Merge results -The `merge` node collects all four responses. With `error_policy="continue"`, it waits for every branch — even if some fail. A missing API key or provider outage doesn't cancel the entire workflow. +The `merge` node collects all four responses. It waits for every branch — even if some fail. A missing API key or provider outage doesn't cancel the entire workflow. ### 3. Synthesize @@ -88,7 +88,7 @@ The `synth` node receives all four perspectives in its preamble and produces a u This workflow combines two patterns from earlier tutorials: -- **Parallel execution** from [Parallel Review](/tutorials/parallel-review) — fan-out/fan-in with join and error policies +- **Parallel execution** from [Parallel Review](/tutorials/parallel-review) — fan-out/fan-in with join policies - **Model routing** from [Multi-Model Routing](/tutorials/multi-model) — stylesheet selectors assigning different providers to each node The key difference from the parallel review tutorial is that here each branch uses a _different provider_, not just a different prompt. This gives you genuinely independent perspectives — each model has different training data, different reasoning patterns, and different blind spots. @@ -106,7 +106,6 @@ The tradeoff is cost and latency — you're making 4x the LLM calls. Use single- ## What you've learned - **Ensemble workflows** fan out the same task to multiple providers -- **`error_policy="continue"`** keeps the workflow running even when some branches fail - **ID selectors** (`#opus`, `#gemini`) assign each branch to a specific model - A **synthesis node** compares perspectives and produces a unified result - Combine parallel execution and model routing for diverse, independent analysis diff --git a/docs/tutorials/parallel-review.mdx b/docs/tutorials/parallel-review.mdx index afda3c47e..f7f2f66db 100644 --- a/docs/tutorials/parallel-review.mdx +++ b/docs/tutorials/parallel-review.mdx @@ -19,7 +19,7 @@ digraph Parallel { start [shape=Mdiamond, label="Start"] exit [shape=Msquare, label="Exit"] - fork [label="Fork Analysis", shape=component, join_policy="wait_all", error_policy="continue"] + fork [label="Fork Analysis", shape=component, join_policy="wait_all"] security [label="Security Audit", prompt="Examine the codebase for security concerns: hardcoded secrets, injection risks, unsafe dependencies. List findings as bullet points.", shape=tab, reasoning_effort="low"] architecture [label="Architecture Review", prompt="Assess the codebase architecture: separation of concerns, dependency structure, modularity. List findings as bullet points.", shape=tab, reasoning_effort="low"] @@ -48,7 +48,7 @@ fabro run files-internal/demo/06-parallel.fabro The `fork` node has `shape=component`, making it a **parallel fan-out node**. Every outgoing edge becomes a concurrent branch: ```dot -fork [label="Fork Analysis", shape=component, join_policy="wait_all", error_policy="continue"] +fork [label="Fork Analysis", shape=component, join_policy="wait_all"] fork -> security fork -> architecture @@ -65,20 +65,6 @@ The `join_policy` controls when execution can proceed past the merge: |---|---| | `wait_all` | Wait for every branch to finish (default) | | `first_success` | Proceed as soon as one branch succeeds | -| `k_of_n(N)` | Proceed after N branches succeed | -| `quorum(0.5)` | Proceed after a fraction of branches succeed | - -### Error policies - -The `error_policy` controls what happens when a branch fails: - -| Policy | Behavior | -|---|---| -| `continue` | Run all branches even if some fail (default) | -| `fail_fast` | Cancel remaining branches as soon as one fails | -| `ignore` | Treat all branch failures as successes | - -This workflow uses `error_policy="continue"` so that a failure in one review perspective doesn't cancel the others. ## Fan-in with the merge node @@ -109,7 +95,6 @@ This is useful when branches are resource-intensive (e.g., each running a full a - **Fan-out nodes** (`shape=component`) spawn concurrent branches - **Merge nodes** (`shape=tripleoctagon`) collect branch results - **Join policies** control when execution can proceed past the merge -- **Error policies** control how branch failures are handled - Each branch gets an isolated copy of the context ## Next diff --git a/docs/workflows/stages-and-nodes.mdx b/docs/workflows/stages-and-nodes.mdx index dcf74345c..261b5744b 100644 --- a/docs/workflows/stages-and-nodes.mdx +++ b/docs/workflows/stages-and-nodes.mdx @@ -158,7 +158,7 @@ Conditions support `=`, `!=`, `&&`, and context variable lookups (e.g. `context. Fans out to execute multiple branches concurrently. Each branch gets its own isolated context. ```dot -fork [label="Fan Out", shape=component, join_policy="wait_all", error_policy="continue"] +fork [label="Fan Out", shape=component, join_policy="wait_all"] fork -> security fork -> architecture @@ -168,7 +168,6 @@ fork -> quality | Attribute | Description | |---|---| | `join_policy` | When the merge can proceed (see table below) | -| `error_policy` | How branch failures are handled (see table below) | | `max_parallel` | Maximum concurrent branches (default: 4) | **Join policies:** @@ -177,16 +176,6 @@ fork -> quality |---|---| | `wait_all` | Wait for every branch to finish (default) | | `first_success` | Proceed as soon as one branch succeeds | -| `k_of_n(N)` | Proceed after N branches succeed (e.g. `k_of_n(2)`) | -| `quorum(F)` | Proceed after a fraction of branches succeed (e.g. `quorum(0.5)`) | - -**Error policies:** - -| Policy | Behavior | -|---|---| -| `continue` | Run all branches even if some fail, then evaluate the join policy (default) | -| `fail_fast` | Cancel remaining branches as soon as one fails | -| `ignore` | Treat all branch failures as successes | ### Merge (fan-in) diff --git a/lib/crates/fabro-cli/src/commands/run_progress.rs b/lib/crates/fabro-cli/src/commands/run_progress.rs index 0e0ecde20..334aae433 100644 --- a/lib/crates/fabro-cli/src/commands/run_progress.rs +++ b/lib/crates/fabro-cli/src/commands/run_progress.rs @@ -1767,7 +1767,6 @@ mod tests { ui.handle_event(&WorkflowRunEvent::ParallelStarted { branch_count: 2, join_policy: "wait_all".into(), - error_policy: "continue".into(), }); assert_eq!(ui.parallel_parent.as_deref(), Some("fork1")); @@ -1828,7 +1827,6 @@ mod tests { ui.handle_event(&WorkflowRunEvent::ParallelStarted { branch_count: 1, join_policy: "wait_all".into(), - error_policy: "continue".into(), }); ui.handle_event(&WorkflowRunEvent::ParallelBranchStarted { branch: "security".into(), @@ -1852,7 +1850,6 @@ mod tests { ui.handle_event(&WorkflowRunEvent::ParallelStarted { branch_count: 1, join_policy: "wait_all".into(), - error_policy: "continue".into(), }); ui.handle_event(&WorkflowRunEvent::ParallelBranchStarted { branch: "risky".into(), @@ -1947,7 +1944,6 @@ mod tests { ui.handle_event(&WorkflowRunEvent::ParallelStarted { branch_count: 2, join_policy: "wait_all".into(), - error_policy: "continue".into(), }); // In Plain mode, active_stages is empty so parallel_parent is a sentinel assert!(ui.parallel_parent.is_some()); @@ -2081,7 +2077,7 @@ mod tests { // Set up a parent stage and start parallel let parent = r#"{"ts":"2026-01-01T12:00:00Z","event":"StageStarted","node_id":"fork","node_label":"Fork","stage_index":0,"attempt":1,"max_attempts":1}"#; ui.handle_json_line(parent); - let par = r#"{"ts":"2026-01-01T12:00:01Z","event":"ParallelStarted","branch_count":2,"join_policy":"wait_all","error_policy":"continue"}"#; + let par = r#"{"ts":"2026-01-01T12:00:01Z","event":"ParallelStarted","branch_count":2,"join_policy":"wait_all"}"#; ui.handle_json_line(par); assert!(ui.parallel_parent.is_some()); diff --git a/lib/crates/fabro-workflows/README.md b/lib/crates/fabro-workflows/README.md index 0f7fcc70f..12e04d42a 100644 --- a/lib/crates/fabro-workflows/README.md +++ b/lib/crates/fabro-workflows/README.md @@ -154,7 +154,7 @@ Nodes with `shape=hexagon` or `type="human"` pause execution for human input. Ou ### Parallel Execution -Nodes with `shape=component` fan out to branches concurrently. Configurable join policies: `wait_all` (default), `first_success`, `k_of_n(N)`, `quorum(0.5)`. Error policies: `continue`, `fail_fast`, `ignore`. +Nodes with `shape=component` fan out to branches concurrently. Configurable join policies: `wait_all` (default), `first_success`. ### Checkpoints and Resume diff --git a/lib/crates/fabro-workflows/src/event.rs b/lib/crates/fabro-workflows/src/event.rs index 3114aa0dc..29460ee55 100644 --- a/lib/crates/fabro-workflows/src/event.rs +++ b/lib/crates/fabro-workflows/src/event.rs @@ -96,7 +96,6 @@ pub enum WorkflowRunEvent { ParallelStarted { branch_count: usize, join_policy: String, - error_policy: String, }, ParallelBranchStarted { branch: String, @@ -196,11 +195,6 @@ pub enum WorkflowRunEvent { stage: String, event: AgentEvent, }, - ParallelEarlyTermination { - reason: String, - completed_count: usize, - pending_count: usize, - }, SubgraphStarted { node_id: String, start_node: String, @@ -457,12 +451,8 @@ impl WorkflowRunEvent { Self::ParallelStarted { branch_count, join_policy, - error_policy, } => { - debug!( - branch_count, - join_policy, error_policy, "Parallel execution started" - ); + debug!(branch_count, join_policy, "Parallel execution started"); } Self::ParallelBranchStarted { branch, index } => { debug!(branch, index, "Parallel branch started"); @@ -571,16 +561,6 @@ impl WorkflowRunEvent { } => { info!(working_directory, "Sandbox initialized"); } - Self::ParallelEarlyTermination { - reason, - completed_count, - pending_count, - } => { - warn!( - reason, - completed_count, pending_count, "Parallel early termination" - ); - } Self::SubgraphStarted { node_id, start_node, @@ -1421,15 +1401,13 @@ mod tests { let event = WorkflowRunEvent::ParallelStarted { branch_count: 3, join_policy: "wait_all".to_string(), - error_policy: "continue".to_string(), }; let json = serde_json::to_string(&event).unwrap(); assert!(json.contains("\"join_policy\":\"wait_all\"")); - assert!(json.contains("\"error_policy\":\"continue\"")); let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap(); assert!( - matches!(deserialized, WorkflowRunEvent::ParallelStarted { join_policy, error_policy, .. } if join_policy == "wait_all" && error_policy == "continue") + matches!(deserialized, WorkflowRunEvent::ParallelStarted { join_policy, .. } if join_policy == "wait_all") ); } @@ -1588,28 +1566,6 @@ mod tests { assert!(matches!(deserialized, WorkflowRunEvent::Agent { stage, .. } if stage == "code")); } - #[test] - fn parallel_early_termination_event_serialization() { - let event = WorkflowRunEvent::ParallelEarlyTermination { - reason: "fail_fast_branch_failed".to_string(), - completed_count: 2, - pending_count: 3, - }; - let json = serde_json::to_string(&event).unwrap(); - assert!(json.contains("ParallelEarlyTermination")); - assert!(json.contains("\"completed_count\":2")); - assert!(json.contains("\"pending_count\":3")); - - let deserialized: WorkflowRunEvent = serde_json::from_str(&json).unwrap(); - assert!(matches!( - deserialized, - WorkflowRunEvent::ParallelEarlyTermination { - completed_count: 2, - .. - } - )); - } - #[test] fn subgraph_started_event_serialization() { let event = WorkflowRunEvent::SubgraphStarted { diff --git a/lib/crates/fabro-workflows/src/handler/parallel.rs b/lib/crates/fabro-workflows/src/handler/parallel.rs index b8545f646..1eb3f87d7 100644 --- a/lib/crates/fabro-workflows/src/handler/parallel.rs +++ b/lib/crates/fabro-workflows/src/handler/parallel.rs @@ -27,8 +27,6 @@ pub struct ParallelHandler; enum JoinPolicy { WaitAll, FirstSuccess, - KOfN(usize), - Quorum(f64), } impl std::fmt::Display for JoinPolicy { @@ -36,8 +34,6 @@ impl std::fmt::Display for JoinPolicy { match self { Self::WaitAll => write!(f, "wait_all"), Self::FirstSuccess => write!(f, "first_success"), - Self::KOfN(k) => write!(f, "k_of_n({k})"), - Self::Quorum(frac) => write!(f, "quorum({frac})"), } } } @@ -46,51 +42,9 @@ fn parse_join_policy(raw: &str) -> JoinPolicy { if raw == "first_success" { return JoinPolicy::FirstSuccess; } - if let Some(inner) = raw - .strip_prefix("k_of_n(") - .and_then(|s| s.strip_suffix(')')) - { - if let Ok(k) = inner.trim().parse::() { - return JoinPolicy::KOfN(k); - } - } - if let Some(inner) = raw - .strip_prefix("quorum(") - .and_then(|s| s.strip_suffix(')')) - { - if let Ok(frac) = inner.trim().parse::() { - return JoinPolicy::Quorum(frac); - } - } JoinPolicy::WaitAll } -/// Parse error policy from node attributes. -#[derive(Debug, Clone, PartialEq, Eq)] -enum ErrorPolicy { - Continue, - FailFast, - Ignore, -} - -impl std::fmt::Display for ErrorPolicy { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - Self::Continue => write!(f, "continue"), - Self::FailFast => write!(f, "fail_fast"), - Self::Ignore => write!(f, "ignore"), - } - } -} - -fn parse_error_policy(raw: &str) -> ErrorPolicy { - match raw { - "fail_fast" => ErrorPolicy::FailFast, - "ignore" => ErrorPolicy::Ignore, - _ => ErrorPolicy::Continue, - } -} - struct BranchResult { id: String, outcome: Outcome, @@ -182,17 +136,10 @@ impl Handler for ParallelHandler { .and_then(|v| v.as_str()) .unwrap_or("wait_all"), ); - let error_policy = parse_error_policy( - node.attrs - .get("error_policy") - .and_then(|v| v.as_str()) - .unwrap_or("continue"), - ); services.emitter.emit(&WorkflowRunEvent::ParallelStarted { branch_count: branches.len(), join_policy: join_policy.to_string(), - error_policy: error_policy.to_string(), }); { let mut hook_ctx = HookContext::new( @@ -436,65 +383,27 @@ impl Handler for ParallelHandler { } // Collect results - let total_branches = handles.len(); let mut results: Vec = Vec::new(); - for (handle_index, handle) in handles.into_iter().enumerate() { + for handle in handles { match handle.await { Ok(Ok(result)) => { - if error_policy == ErrorPolicy::FailFast - && result.outcome.status == StageStatus::Fail - { - results.push(result); - services - .emitter - .emit(&WorkflowRunEvent::ParallelEarlyTermination { - reason: "fail_fast_branch_failed".to_string(), - completed_count: results.len(), - pending_count: total_branches - handle_index - 1, - }); - break; - } results.push(result); } Ok(Err(e)) => { - let result = BranchResult { + results.push(BranchResult { id: String::new(), outcome: e.to_fail_outcome(), head_sha: None, worktree_path: None, - }; - if error_policy == ErrorPolicy::FailFast { - results.push(result); - services - .emitter - .emit(&WorkflowRunEvent::ParallelEarlyTermination { - reason: "fail_fast_handler_error".to_string(), - completed_count: results.len(), - pending_count: total_branches - handle_index - 1, - }); - break; - } - results.push(result); + }); } Err(join_err) => { - let result = BranchResult { + results.push(BranchResult { id: String::new(), outcome: Outcome::fail_classify(format!("task join error: {join_err}")), head_sha: None, worktree_path: None, - }; - if error_policy == ErrorPolicy::FailFast { - results.push(result); - services - .emitter - .emit(&WorkflowRunEvent::ParallelEarlyTermination { - reason: "fail_fast_join_error".to_string(), - completed_count: results.len(), - pending_count: total_branches - handle_index - 1, - }); - break; - } - results.push(result); + }); } } } @@ -579,7 +488,7 @@ impl Handler for ParallelHandler { // Evaluate join policy let status = match join_policy { JoinPolicy::WaitAll => { - if fail_count == 0 || error_policy == ErrorPolicy::Ignore { + if fail_count == 0 { StageStatus::Success } else { StageStatus::PartialSuccess @@ -592,23 +501,6 @@ impl Handler for ParallelHandler { StageStatus::Fail } } - JoinPolicy::KOfN(k) => { - if success_count >= k { - StageStatus::Success - } else { - StageStatus::Fail - } - } - JoinPolicy::Quorum(fraction) => { - let total_f64 = total as f64; - let threshold_f64 = (fraction * total_f64).ceil(); - let threshold = threshold_f64 as usize; - if success_count >= threshold { - StageStatus::Success - } else { - StageStatus::Fail - } - } }; // Find the join/convergence node: follow each branch's outgoing edges @@ -773,53 +665,10 @@ mod tests { assert_eq!(outcome.status, StageStatus::Success); } - #[tokio::test] - async fn parallel_handler_k_of_n_policy() { - let services = make_services(); - let mut node = Node::new("par"); - node.attrs.insert( - "join_policy".to_string(), - AttrValue::String("k_of_n(2)".to_string()), - ); - let context = Context::new(); - let mut graph = Graph::new("test"); - graph.nodes.insert("par".to_string(), node.clone()); - graph - .nodes - .insert("branch_a".to_string(), Node::new("branch_a")); - graph - .nodes - .insert("branch_b".to_string(), Node::new("branch_b")); - graph - .nodes - .insert("branch_c".to_string(), Node::new("branch_c")); - graph.edges.push(Edge::new("par", "branch_a")); - graph.edges.push(Edge::new("par", "branch_b")); - graph.edges.push(Edge::new("par", "branch_c")); - - let run_dir = Path::new("/tmp/test"); - let outcome = ParallelHandler - .execute(&node, &context, &graph, run_dir, &services) - .await - .unwrap(); - - // All 3 succeed (default StartHandler returns success), need 2 - assert_eq!(outcome.status, StageStatus::Success); - } - #[test] fn join_policy_display() { assert_eq!(JoinPolicy::WaitAll.to_string(), "wait_all"); assert_eq!(JoinPolicy::FirstSuccess.to_string(), "first_success"); - assert_eq!(JoinPolicy::KOfN(3).to_string(), "k_of_n(3)"); - assert_eq!(JoinPolicy::Quorum(0.5).to_string(), "quorum(0.5)"); - } - - #[test] - fn error_policy_display() { - assert_eq!(ErrorPolicy::Continue.to_string(), "continue"); - assert_eq!(ErrorPolicy::FailFast.to_string(), "fail_fast"); - assert_eq!(ErrorPolicy::Ignore.to_string(), "ignore"); } #[test] @@ -829,26 +678,10 @@ mod tests { parse_join_policy("first_success"), JoinPolicy::FirstSuccess )); - assert!(matches!( - parse_join_policy("k_of_n(3)"), - JoinPolicy::KOfN(3) - )); - assert!(matches!( - parse_join_policy("quorum(0.5)"), - JoinPolicy::Quorum(_) - )); // Invalid falls back to WaitAll assert!(matches!(parse_join_policy("invalid"), JoinPolicy::WaitAll)); } - #[test] - fn parse_error_policy_variants() { - assert_eq!(parse_error_policy("continue"), ErrorPolicy::Continue); - assert_eq!(parse_error_policy("fail_fast"), ErrorPolicy::FailFast); - assert_eq!(parse_error_policy("ignore"), ErrorPolicy::Ignore); - assert_eq!(parse_error_policy("unknown"), ErrorPolicy::Continue); - } - #[tokio::test] async fn parallel_handler_simulate() { let services = make_services(); diff --git a/skills/fabro-create-workflow/references/dot-language.md b/skills/fabro-create-workflow/references/dot-language.md index 3830e1ec0..1eaed0019 100644 --- a/skills/fabro-create-workflow/references/dot-language.md +++ b/skills/fabro-create-workflow/references/dot-language.md @@ -96,7 +96,7 @@ Route execution based on conditions. Must have multiple outgoing edges with `con ### Parallel Fan-Out (component) -Attributes: `join_policy` (wait_all, first_success, k_of_n(N), quorum(F)), `error_policy` (continue, fail_fast, ignore), `max_parallel` (default: 4). +Attributes: `join_policy` (wait_all, first_success), `max_parallel` (default: 4). ### Fan-In / Merge (tripleoctagon) diff --git a/skills/fabro-create-workflow/references/example-workflows.md b/skills/fabro-create-workflow/references/example-workflows.md index 7ec1761f1..1411a577f 100644 --- a/skills/fabro-create-workflow/references/example-workflows.md +++ b/skills/fabro-create-workflow/references/example-workflows.md @@ -122,7 +122,7 @@ digraph Parallel { start [shape=Mdiamond, label="Start"] exit [shape=Msquare, label="Exit"] - fork [label="Fork Analysis", shape=component, join_policy="wait_all", error_policy="continue"] + fork [label="Fork Analysis", shape=component, join_policy="wait_all"] security [label="Security Audit", prompt="Examine the codebase for security concerns: hardcoded secrets, injection risks, unsafe dependencies. List findings as bullet points.", shape=tab, reasoning_effort="low"] architecture [label="Architecture Review", prompt="Assess the codebase architecture: separation of concerns, dependency structure, modularity. List findings as bullet points.", shape=tab, reasoning_effort="low"] @@ -190,7 +190,7 @@ digraph Ensemble { start [shape=Mdiamond, label="Start"] exit [shape=Msquare, label="Exit"] - fork [label="Fan Out", shape=component, join_policy="wait_all", error_policy="continue"] + fork [label="Fan Out", shape=component, join_policy="wait_all"] opus [label="Opus", prompt="Analyze the goal. Provide your independent assessment and recommendations. Be thorough.", shape=tab] gemini [label="Gemini", prompt="Analyze the goal. Provide your independent assessment and recommendations. Be thorough.", shape=tab] diff --git a/test/docs/tutorials/ensemble/ensemble.fabro b/test/docs/tutorials/ensemble/ensemble.fabro index 8af88cb36..3a8655cde 100644 --- a/test/docs/tutorials/ensemble/ensemble.fabro +++ b/test/docs/tutorials/ensemble/ensemble.fabro @@ -14,7 +14,7 @@ digraph Ensemble { start [shape=Mdiamond, label="Start"] exit [shape=Msquare, label="Exit"] - fork [label="Fan Out", shape=component, join_policy="wait_all", error_policy="continue"] + fork [label="Fan Out", shape=component, join_policy="wait_all"] opus [label="Opus", prompt="Analyze the goal. Provide your independent assessment, recommendations, and any code or prose needed. Be thorough.", shape=tab] gemini [label="Gemini", prompt="Analyze the goal. Provide your independent assessment, recommendations, and any code or prose needed. Be thorough.", shape=tab] diff --git a/test/docs/tutorials/parallel-review/parallel.fabro b/test/docs/tutorials/parallel-review/parallel.fabro index 9a79c2940..a4cab30e9 100644 --- a/test/docs/tutorials/parallel-review/parallel.fabro +++ b/test/docs/tutorials/parallel-review/parallel.fabro @@ -5,7 +5,7 @@ digraph Parallel { start [shape=Mdiamond, label="Start"] exit [shape=Msquare, label="Exit"] - fork [label="Fork Analysis", shape=component, join_policy="wait_all", error_policy="continue"] + fork [label="Fork Analysis", shape=component, join_policy="wait_all"] security [label="Security Audit", prompt="Examine the codebase for security concerns: hardcoded secrets, injection risks, unsafe dependencies. List findings as bullet points.", shape=tab, reasoning_effort="low"] architecture [label="Architecture Review", prompt="Assess the codebase architecture: separation of concerns, dependency structure, modularity. List findings as bullet points.", shape=tab, reasoning_effort="low"] diff --git a/test/docs/workflows/stages-and-nodes/all-node-types.fabro b/test/docs/workflows/stages-and-nodes/all-node-types.fabro index 51b8a8e63..bbc9fb0be 100644 --- a/test/docs/workflows/stages-and-nodes/all-node-types.fabro +++ b/test/docs/workflows/stages-and-nodes/all-node-types.fabro @@ -7,7 +7,7 @@ digraph AllNodeTypes { test [label="Run Tests", shape=parallelogram, script="cargo test 2>&1 || true"] gate [shape=diamond, label="Tests passing?"] cooldown [label="Wait 30s", shape=insulator, duration="30s"] - fork [label="Fan Out", shape=component, join_policy="wait_all", error_policy="continue"] + fork [label="Fan Out", shape=component, join_policy="wait_all"] security [label="Security Review"] architecture [label="Architecture Review"] quality [label="Quality Review"] diff --git a/test/parallel.fabro b/test/parallel.fabro index 30da4a3ff..96cd323fa 100644 --- a/test/parallel.fabro +++ b/test/parallel.fabro @@ -4,7 +4,7 @@ digraph Parallel { start [shape=Mdiamond] exit [shape=Msquare] - fork [label="Fork Work", shape=component, join_policy="wait_all", error_policy="continue"] + fork [label="Fork Work", shape=component, join_policy="wait_all"] branch1 [label="Branch 1"] branch2 [label="Branch 2"] merge [label="Merge Results", shape=tripleoctagon]