scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-08-28 05:27:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Attractor spec hunks 14 & 16: remove error_policy and k_of_n/quorum from parallel handler

Browse source 
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) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-03-23 14:59:35 -04:00
parent b9fe1282d3
commit 588515dbf6
19 changed files with 30 additions and 274 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
docs-internal/demo/06-parallel.fabro
View file

@ -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"]

2
docs-internal/demo/11-ensemble.fabro
View file

@ -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]

3
docs-internal/events-strategy.md
View file

@ -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

2
docs/execution/observability.mdx
View file

@ -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 |

2
docs/execution/outcomes.mdx
View file

@ -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 |

3
docs/reference/dot-language.mdx
View file

@ -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

9
docs/tutorials/ensemble.mdx
View file

@ -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
```
<Note>
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.
</Note>
## 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

19
docs/tutorials/parallel-review.mdx
View file

@ -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

13
docs/workflows/stages-and-nodes.mdx
View file

@ -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)

6
lib/crates/fabro-cli/src/commands/run_progress.rs
View file

@ -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());

2
lib/crates/fabro-workflows/README.md
View file

@ -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

48
lib/crates/fabro-workflows/src/event.rs
View file

@ -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 {

179
lib/crates/fabro-workflows/src/handler/parallel.rs
View file

@ -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::<usize>() {
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::<f64>() {
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<BranchResult> = 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();

2
skills/fabro-create-workflow/references/dot-language.md
View file

@ -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)

4
skills/fabro-create-workflow/references/example-workflows.md
View file

@ -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]

2
test/docs/tutorials/ensemble/ensemble.fabro
View file

@ -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]

2
test/docs/tutorials/parallel-review/parallel.fabro
View file

@ -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"]

2
test/docs/workflows/stages-and-nodes/all-node-types.fabro
View file

@ -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"]

2
test/parallel.fabro
View file

@ -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]