From a2650503e8517d0bd41c3fd6d09e79d41e709514 Mon Sep 17 00:00:00 2001 From: Fabro Date: Thu, 23 Jul 2026 16:47:50 +0000 Subject: [PATCH 01/15] fabro(01KY7Y01REECZ24XXTMBZ3PPV9): toolchain (succeeded) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fabro-Run: 01KY7Y01REECZ24XXTMBZ3PPV9 Fabro-Completed: 2 Fabro-Checkpoint: 970f0d7912c042bbab4648a91d1f17accdac29b9 ⚒️ Generated with [Fabro](https://fabro.sh) From dfddfb747b5f438c1ad16bcfaae09b7d7d01f018 Mon Sep 17 00:00:00 2001 From: Fabro Date: Thu, 23 Jul 2026 16:50:14 +0000 Subject: [PATCH 02/15] fabro(01KY7Y01REECZ24XXTMBZ3PPV9): preflight_compile (succeeded) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fabro-Run: 01KY7Y01REECZ24XXTMBZ3PPV9 Fabro-Completed: 3 Fabro-Checkpoint: 4bc92b4b11014acd4f58de58881ba25509ade1ae ⚒️ Generated with [Fabro](https://fabro.sh) From 1879fe0906049f3bd4d113f5de2736c7f3ff5af7 Mon Sep 17 00:00:00 2001 From: Fabro Date: Thu, 23 Jul 2026 16:52:52 +0000 Subject: [PATCH 03/15] fabro(01KY7Y01REECZ24XXTMBZ3PPV9): preflight_lint (succeeded) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fabro-Run: 01KY7Y01REECZ24XXTMBZ3PPV9 Fabro-Completed: 4 Fabro-Checkpoint: bb25ac7d9f129de489aa019914494d2e5e49e5d9 ⚒️ Generated with [Fabro](https://fabro.sh) From 1a2bd7966dc15ac5228546964879e1b4b577ad85 Mon Sep 17 00:00:00 2001 From: Fabro Date: Thu, 23 Jul 2026 17:41:29 +0000 Subject: [PATCH 04/15] fabro(01KY7Y01REECZ24XXTMBZ3PPV9): implement (succeeded) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fabro-Run: 01KY7Y01REECZ24XXTMBZ3PPV9 Fabro-Completed: 5 Fabro-Checkpoint: 1f2ff54692f4e8627086993be6aafa876c57e3fa ⚒️ Generated with [Fabro](https://fabro.sh) --- docs/public/execution/context.mdx | 10 + docs/public/reference/dot-language.mdx | 10 +- docs/public/workflows/stages-and-nodes.mdx | 2 + lib/crates/fabro-validate/src/rules/mod.rs | 1 + .../src/rules/parallel_branch.rs | 56 ++++ .../rules/parallel_branch_inert_attribute.rs | 203 ++++++++---- .../rules/thread_id_requires_fidelity_full.rs | 73 ++++- lib/crates/fabro-workflow/src/artifact.rs | 57 ++++ lib/crates/fabro-workflow/src/context.rs | 1 + .../fabro-workflow/src/handler/parallel.rs | 238 +++++++++++++- .../fabro-workflow/src/lifecycle/event.rs | 37 ++- .../fabro-workflow/src/lifecycle/fidelity.rs | 291 ++++++++++++++++++ .../fabro-workflow/tests/it/integration.rs | 191 ++++++++++++ 13 files changed, 1093 insertions(+), 77 deletions(-) create mode 100644 lib/crates/fabro-validate/src/rules/parallel_branch.rs diff --git a/docs/public/execution/context.mdx b/docs/public/execution/context.mdx index c67012b3c..a08b2d960 100644 --- a/docs/public/execution/context.mdx +++ b/docs/public/execution/context.mdx @@ -139,6 +139,16 @@ Fidelity can be set at three levels. The first match wins: If none of these are set, fidelity defaults to `compact`. +### Parallel branch fidelity + +The first node in each parallel branch uses this precedence: + +1. `fidelity` on the fork-to-branch edge +2. `fidelity` on the branch node +3. Otherwise, inherit the fork's preamble unchanged + +Fabro renders any branch-specific preambles before fan-out from the fork's context snapshot, then places them into the isolated branch contexts. An explicit branch-level `full` degrades to `summary:high` because concurrent branches cannot share conversation sessions. `thread_id` on a branch node or fork-to-branch edge is inert. + ### Full fidelity and threads `full` fidelity is typically used with `thread_id` to create a shared conversation across multiple nodes. Nodes with the same `thread_id` share a single LLM session, preserving full context continuity: diff --git a/docs/public/reference/dot-language.mdx b/docs/public/reference/dot-language.mdx index f01fbf2ed..3e407242b 100644 --- a/docs/public/reference/dot-language.mdx +++ b/docs/public/reference/dot-language.mdx @@ -201,8 +201,8 @@ Start nodes can also be identified by ID (`start` or `Start`). Exit nodes can be | `prompt` | String | Task instructions for the LLM. Supports file references with `@path/to/file.md` | | `reasoning_effort` | String | `low`, `medium`, or `high` (default: `high`) | | `max_tokens` | Integer | Maximum output tokens | -| `fidelity` | String | How much prior context is passed: `compact`, `full`, `summary:high`, `summary:medium`, `summary:low`, `truncate` | -| `thread_id` | String | Groups nodes into a shared conversation thread | +| `fidelity` | String | How much prior context is passed: `compact`, `full`, `summary:high`, `summary:medium`, `summary:low`, `truncate`. On a node entered directly from a parallel fork, this is overridden by the fork-to-branch edge; explicit `full` degrades to `summary:high`. | +| `thread_id` | String | Groups nodes into a shared conversation thread. Inert when the node is entered directly from a parallel fork. | | `model` | String | Explicit model ID (overrides stylesheet) | | `provider` | String | Explicit provider name (overrides stylesheet). Auto-inferred from the model catalog when omitted. | | `project_memory` | Boolean | When `true` (default), prompt nodes discover and include project docs (`AGENTS.md`, `CLAUDE.md`, etc.) as a system prompt. Set to `false` to disable. | @@ -252,6 +252,8 @@ audit [ | `join_policy` | String | When the merge can proceed: `wait_all` (default), `first_success` | | `max_parallel` | Integer | Maximum concurrent branches (default: 4) | +For the first node in each branch, `fidelity` resolves from the fork-to-branch edge, then the branch node; without either, the fork preamble is inherited unchanged. Branch-specific preambles are rendered before fan-out from the fork's context snapshot. Concurrent branches cannot share sessions, so explicit branch `full` becomes `summary:high`, and branch-level `thread_id` is inert. + ### Wait nodes | Attribute | Type | Description | @@ -282,8 +284,8 @@ audit [ | `label` | String | Display text; also used for human gate option matching | | `condition` | String | Boolean expression for conditional routing (see below) | | `weight` | Integer | Priority for tiebreaking (higher wins, default: 0) | -| `fidelity` | String | Override fidelity level for this transition | -| `thread_id` | String | Override thread ID for this transition | +| `fidelity` | String | Override fidelity level for this transition. On a fork-to-branch edge, takes precedence over the branch node; explicit `full` degrades to `summary:high`. | +| `thread_id` | String | Override thread ID for this transition. Inert on fork-to-branch edges. | | `loop_restart` | Boolean | Restart the workflow from this edge's target when taken: stage history and retry counts clear and the context resets to empty (visit counts are kept). Failed outcomes may only take it for `transient_infra` failures — see [Failures](/execution/failures#loop-restart-edges) | | `freeform` | Boolean | When `true` on a human-gate edge, accept free-text input instead of fixed choices | diff --git a/docs/public/workflows/stages-and-nodes.mdx b/docs/public/workflows/stages-and-nodes.mdx index 1dd0f71ac..1b8c252c2 100644 --- a/docs/public/workflows/stages-and-nodes.mdx +++ b/docs/public/workflows/stages-and-nodes.mdx @@ -170,6 +170,8 @@ fork -> quality | `join_policy` | When the merge can proceed (see table below) | | `max_parallel` | Maximum concurrent branches (default: 4) | +For each branch's first node, fidelity resolves from the fork-to-branch edge, then the branch node; otherwise it inherits the fork preamble unchanged. Fabro renders branch-specific preambles before fan-out from the fork snapshot. Branch-level `full` degrades to `summary:high` because concurrent branches cannot share sessions, and `thread_id` on a branch node or fork-to-branch edge is inert. + **Join policies:** | Policy | Behavior | diff --git a/lib/crates/fabro-validate/src/rules/mod.rs b/lib/crates/fabro-validate/src/rules/mod.rs index 428004240..87075bc17 100644 --- a/lib/crates/fabro-validate/src/rules/mod.rs +++ b/lib/crates/fabro-validate/src/rules/mod.rs @@ -12,6 +12,7 @@ mod inert_attribute; mod model_support; mod node_model_known; mod orphan_custom_outcome; +mod parallel_branch; mod parallel_branch_inert_attribute; mod prompt_on_llm_nodes; mod random_selection_no_conditions; diff --git a/lib/crates/fabro-validate/src/rules/parallel_branch.rs b/lib/crates/fabro-validate/src/rules/parallel_branch.rs new file mode 100644 index 000000000..2f1cba2e7 --- /dev/null +++ b/lib/crates/fabro-validate/src/rules/parallel_branch.rs @@ -0,0 +1,56 @@ +use std::collections::BTreeSet; + +use fabro_graphviz::graph::{Edge, Graph}; + +pub(super) struct ParallelBranches<'a> { + graph: &'a Graph, + fork_ids: BTreeSet<&'a str>, +} + +impl<'a> ParallelBranches<'a> { + pub(super) fn new(graph: &'a Graph) -> Self { + let fork_ids = graph + .nodes + .values() + .filter(|node| node.handler_type() == Some("parallel")) + .map(|node| node.id.as_str()) + .collect(); + Self { graph, fork_ids } + } + + pub(super) fn is_empty(&self) -> bool { + self.fork_ids.is_empty() + } + + pub(super) fn is_fork_edge(&self, edge: &Edge) -> bool { + self.fork_ids.contains(edge.from.as_str()) + } + + pub(super) fn branch_targets(&self) -> BTreeSet<&str> { + self.graph + .edges + .iter() + .filter(|edge| self.is_fork_edge(edge)) + .map(|edge| edge.to.as_str()) + .collect() + } + + pub(super) fn is_branch_only_node(&self, node_id: &str) -> bool { + self.branch_only_parents(node_id).is_some() + } + + pub(super) fn branch_only_parents(&self, node_id: &str) -> Option> { + let mut incoming = self + .graph + .edges + .iter() + .filter(|edge| edge.to == node_id) + .peekable(); + incoming.peek()?; + + incoming + .map(|edge| self.is_fork_edge(edge).then(|| edge.from.clone())) + .collect::>>() + .map(|parents| parents.into_iter().collect()) + } +} diff --git a/lib/crates/fabro-validate/src/rules/parallel_branch_inert_attribute.rs b/lib/crates/fabro-validate/src/rules/parallel_branch_inert_attribute.rs index eb051281e..da36dbbea 100644 --- a/lib/crates/fabro-validate/src/rules/parallel_branch_inert_attribute.rs +++ b/lib/crates/fabro-validate/src/rules/parallel_branch_inert_attribute.rs @@ -1,18 +1,16 @@ -use std::collections::BTreeSet; - use fabro_graphviz::graph::Graph; +use super::parallel_branch::ParallelBranches; use crate::{Diagnostic, LintRule, Severity}; pub(super) fn rule() -> Box { Box::new(Rule) } -/// Attributes that parallel branch execution does not resolve. Branch nodes -/// are dispatched with a snapshot of the context taken when the parallel node -/// started, so per-branch `fidelity` never changes what a branch sees, and -/// per-branch `thread_id` never replaces the thread inherited in that snapshot. -const BRANCH_IGNORED_ATTRS: &[&str] = &["fidelity", "thread_id"]; +/// Attributes that parallel branch execution does not resolve. Per-branch +/// preambles now honor fidelity, while `thread_id` remains inert because +/// concurrent branches cannot share an LLM session. +const BRANCH_IGNORED_ATTRS: &[&str] = &["thread_id"]; struct Rule; @@ -24,21 +22,8 @@ fn quoted_list(ids: &[String]) -> String { .join(", ") } -fn fix_message(attr: &str, parallel_ids: &[String]) -> String { +fn fix_message(attr: &str) -> String { match attr { - "fidelity" => { - if parallel_ids.len() == 1 { - format!( - "Set fidelity on the parallel node {} (or its incoming edge) to control what every branch sees", - quoted_list(parallel_ids), - ) - } else { - format!( - "Set fidelity on the parallel nodes {} (or their incoming edges) to control what every branch sees", - quoted_list(parallel_ids), - ) - } - } "thread_id" => format!( "Remove '{attr}': parallel branches inherit the thread resolved when the parallel node started" ), @@ -46,30 +31,49 @@ fn fix_message(attr: &str, parallel_ids: &[String]) -> String { } } +fn full_fidelity_message() -> String { + "parallel branches run at most at summary:high; full is degraded at runtime because branches cannot share a session".to_string() +} + +fn full_fidelity_fix(parallel_ids: &[String]) -> String { + let parent = if parallel_ids.len() == 1 { + format!("parallel node {}", quoted_list(parallel_ids)) + } else { + format!("parallel nodes {}", quoted_list(parallel_ids)) + }; + format!( + "Use fidelity=\"summary:high\" or another lower mode on this branch; to reuse a full session before fan-out, set fidelity=\"full\" on {parent} or its incoming edge" + ) +} + impl LintRule for Rule { fn name(&self) -> &'static str { "parallel_branch_inert_attribute" } fn apply(&self, graph: &Graph) -> Vec { - let parallel_ids: BTreeSet<&str> = graph - .nodes - .values() - .filter(|n| n.handler_type() == Some("parallel")) - .map(|n| n.id.as_str()) - .collect(); - if parallel_ids.is_empty() { + let branches = ParallelBranches::new(graph); + if branches.is_empty() { return Vec::new(); } let mut diagnostics = Vec::new(); - // Branch edges (parallel node -> branch target) carrying an attribute - // that branch dispatch never reads. for edge in &graph.edges { - if !parallel_ids.contains(edge.from.as_str()) { + if !branches.is_fork_edge(edge) { continue; } + if edge.fidelity() == Some("full") { + diagnostics.push(Diagnostic { + rule: self.name().to_string(), + severity: Severity::Warning, + message: full_fidelity_message(), + node_id: None, + edge: Some((edge.from.clone(), edge.to.clone())), + fix: Some(full_fidelity_fix(std::slice::from_ref(&edge.from))), + ..Diagnostic::default() + }); + } for attr in BRANCH_IGNORED_ATTRS { if !edge.attrs.contains_key(*attr) { continue; @@ -83,42 +87,32 @@ impl LintRule for Rule { ), node_id: None, edge: Some((edge.from.clone(), edge.to.clone())), - fix: Some(fix_message(attr, std::slice::from_ref(&edge.from))), + fix: Some(fix_message(attr)), ..Diagnostic::default() }); } } - // Branch target nodes carrying such an attribute — but only when every - // incoming edge comes from a parallel node. A node that is also - // reachable through a normal edge resolves the attribute on that path, - // so it is not inert there. - let branch_targets: BTreeSet<&str> = graph - .edges - .iter() - .filter(|e| parallel_ids.contains(e.from.as_str())) - .map(|e| e.to.as_str()) - .collect(); - for target in branch_targets { - let only_branch_entries = graph - .edges - .iter() - .filter(|e| e.to == target) - .all(|e| parallel_ids.contains(e.from.as_str())); - if !only_branch_entries { + // A node with any normal incoming path still resolves its attributes on + // that path, so branch-only diagnostics do not apply to it. + for target in branches.branch_targets() { + let Some(parents) = branches.branch_only_parents(target) else { continue; - } + }; let Some(node) = graph.nodes.get(target) else { continue; }; - let parents: Vec = graph - .edges - .iter() - .filter(|e| e.to == target && parallel_ids.contains(e.from.as_str())) - .map(|e| e.from.clone()) - .collect::>() - .into_iter() - .collect(); + if node.fidelity() == Some("full") { + diagnostics.push(Diagnostic { + rule: self.name().to_string(), + severity: Severity::Warning, + message: full_fidelity_message(), + node_id: Some(node.id.clone()), + edge: None, + fix: Some(full_fidelity_fix(&parents)), + ..Diagnostic::default() + }); + } for attr in BRANCH_IGNORED_ATTRS { if !node.attrs.contains_key(*attr) { continue; @@ -133,7 +127,7 @@ impl LintRule for Rule { ), node_id: Some(node.id.clone()), edge: None, - fix: Some(fix_message(attr, &parents)), + fix: Some(fix_message(attr)), ..Diagnostic::default() }); } @@ -181,7 +175,7 @@ mod tests { } #[test] - fn warns_on_fidelity_on_branch_node() { + fn accepts_non_full_fidelity_on_branch_node() { let mut g = parallel_graph(); g.nodes .get_mut("branch_a") @@ -191,14 +185,73 @@ mod tests { "fidelity".to_string(), AttrValue::String("truncate".to_string()), ); + + assert!(Rule.apply(&g).is_empty()); + } + + #[test] + fn warns_when_full_fidelity_on_branch_node_degrades() { + let mut g = parallel_graph(); + g.nodes + .get_mut("branch_a") + .expect("graph has branch_a") + .attrs + .insert( + "fidelity".to_string(), + AttrValue::String("full".to_string()), + ); + let d = Rule.apply(&g); + assert_eq!(d.len(), 1); assert_eq!(d[0].severity, Severity::Warning); assert_eq!(d[0].node_id.as_deref(), Some("branch_a")); - assert!(d[0].message.contains("'fidelity'")); + assert!(d[0].message.contains("full")); + assert!(d[0].message.contains("summary:high")); assert!(d[0].fix.as_deref().is_some_and(|f| f.contains("'fork'"))); } + #[test] + fn accepts_every_non_full_fidelity_on_branch_edges() { + for fidelity in [ + "truncate", + "compact", + "summary:low", + "summary:medium", + "summary:high", + ] { + let mut g = parallel_graph(); + g.edges[1].attrs.insert( + "fidelity".to_string(), + AttrValue::String(fidelity.to_string()), + ); + + assert!( + Rule.apply(&g).is_empty(), + "{fidelity} should be accepted on a branch edge" + ); + } + } + + #[test] + fn warns_when_full_fidelity_on_branch_edge_degrades() { + let mut g = parallel_graph(); + g.edges[1].attrs.insert( + "fidelity".to_string(), + AttrValue::String("full".to_string()), + ); + + let d = Rule.apply(&g); + + assert_eq!(d.len(), 1); + assert_eq!( + d[0].edge, + Some(("fork".to_string(), "branch_a".to_string())) + ); + assert!(d[0].message.contains("full")); + assert!(d[0].message.contains("summary:high")); + } + #[test] fn warns_on_thread_id_on_branch_edge() { let mut g = parallel_graph(); @@ -221,6 +274,31 @@ mod tests { ); } + #[test] + fn warns_on_thread_id_on_branch_only_node() { + let mut g = parallel_graph(); + g.nodes + .get_mut("branch_a") + .expect("graph has branch_a") + .attrs + .insert( + "thread_id".to_string(), + AttrValue::String("impl".to_string()), + ); + + let d = Rule.apply(&g); + + assert_eq!(d.len(), 1); + assert_eq!(d[0].node_id.as_deref(), Some("branch_a")); + assert!(d[0].message.contains("'thread_id'")); + assert_eq!( + d[0].fix.as_deref(), + Some( + "Remove 'thread_id': parallel branches inherit the thread resolved when the parallel node started" + ) + ); + } + #[test] fn accepts_fidelity_on_the_parallel_node_itself() { let mut g = parallel_graph(); @@ -265,11 +343,10 @@ mod tests { .attrs .insert( "fidelity".to_string(), - AttrValue::String("truncate".to_string()), + AttrValue::String("full".to_string()), ); let d = Rule.apply(&g); assert_eq!(d.len(), 1); - assert!(d[0].message.contains("'fork', 'fork2'")); let fix = d[0].fix.as_deref().expect("diagnostic has a fix"); assert!(fix.contains("'fork', 'fork2'")); assert!(fix.contains("parallel nodes")); diff --git a/lib/crates/fabro-validate/src/rules/thread_id_requires_fidelity_full.rs b/lib/crates/fabro-validate/src/rules/thread_id_requires_fidelity_full.rs index 457553788..a73335da5 100644 --- a/lib/crates/fabro-validate/src/rules/thread_id_requires_fidelity_full.rs +++ b/lib/crates/fabro-validate/src/rules/thread_id_requires_fidelity_full.rs @@ -1,5 +1,6 @@ use fabro_graphviz::graph::Graph; +use super::parallel_branch::ParallelBranches; use crate::{Diagnostic, LintRule, Severity}; pub(super) fn rule() -> Box { @@ -20,9 +21,13 @@ impl LintRule for Rule { fn apply(&self, graph: &Graph) -> Vec { let mut diagnostics = Vec::new(); let graph_default_full = graph.default_fidelity() == Some("full"); + let branches = ParallelBranches::new(graph); for node in graph.nodes.values() { - if node.thread_id().is_some() && node.fidelity() != Some("full") && !graph_default_full + if node.thread_id().is_some() + && !branches.is_branch_only_node(&node.id) + && node.fidelity() != Some("full") + && !graph_default_full { diagnostics.push(Diagnostic { rule: self.name().to_string(), @@ -41,7 +46,7 @@ impl LintRule for Rule { } for edge in &graph.edges { - if edge.thread_id().is_some() { + if edge.thread_id().is_some() && !branches.is_fork_edge(edge) { let edge_full = edge.fidelity() == Some("full"); let target_full = graph.nodes.get(&edge.to).and_then(|n| n.fidelity()) == Some("full"); @@ -82,12 +87,29 @@ impl LintRule for Rule { #[cfg(test)] mod tests { - use fabro_graphviz::graph::{AttrValue, Edge, Node}; + use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node}; use super::Rule; use crate::rules::test_support::minimal_graph; use crate::{LintRule, Severity}; + fn parallel_graph() -> Graph { + let mut g = minimal_graph(); + let mut fork = Node::new("fork"); + fork.attrs.insert( + "shape".to_string(), + AttrValue::String("component".to_string()), + ); + g.nodes.insert("fork".to_string(), fork); + g.nodes.insert("branch".to_string(), Node::new("branch")); + g.edges = vec![ + Edge::new("start", "fork"), + Edge::new("fork", "branch"), + Edge::new("branch", "exit"), + ]; + g + } + #[test] fn thread_id_requires_fidelity_full_node_warns() { let mut g = minimal_graph(); @@ -206,6 +228,51 @@ mod tests { assert!(d.is_empty()); } + #[test] + fn skips_thread_id_on_parallel_branch_edge() { + let mut g = parallel_graph(); + g.edges[1].attrs.insert( + "thread_id".to_string(), + AttrValue::String("branch-thread".to_string()), + ); + + assert!(Rule.apply(&g).is_empty()); + } + + #[test] + fn skips_thread_id_on_branch_only_node() { + let mut g = parallel_graph(); + g.nodes + .get_mut("branch") + .expect("graph has branch") + .attrs + .insert( + "thread_id".to_string(), + AttrValue::String("branch-thread".to_string()), + ); + + assert!(Rule.apply(&g).is_empty()); + } + + #[test] + fn checks_thread_id_on_branch_node_with_normal_entry() { + let mut g = parallel_graph(); + g.edges.push(Edge::new("start", "branch")); + g.nodes + .get_mut("branch") + .expect("graph has branch") + .attrs + .insert( + "thread_id".to_string(), + AttrValue::String("shared-thread".to_string()), + ); + + let d = Rule.apply(&g); + + assert_eq!(d.len(), 1); + assert_eq!(d[0].node_id.as_deref(), Some("branch")); + } + #[test] fn thread_id_requires_fidelity_full_graph_warns() { let mut g = minimal_graph(); diff --git a/lib/crates/fabro-workflow/src/artifact.rs b/lib/crates/fabro-workflow/src/artifact.rs index 3f9f9eb7d..975fee26d 100644 --- a/lib/crates/fabro-workflow/src/artifact.rs +++ b/lib/crates/fabro-workflow/src/artifact.rs @@ -79,6 +79,7 @@ pub fn format_artifact_reference(path: &str) -> String { pub fn durable_context_snapshot(context: &Context) -> HashMap { let mut snapshot = context.snapshot(); snapshot.remove(context::keys::CURRENT_PREAMBLE); + snapshot.remove(context::keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES); normalize_durable_updates(&mut snapshot); snapshot } @@ -99,6 +100,9 @@ pub fn normalize_checkpoint_for_resume(checkpoint: &mut Checkpoint) { checkpoint .context_values .remove(context::keys::CURRENT_PREAMBLE); + checkpoint + .context_values + .remove(context::keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES); normalize_durable_updates(&mut checkpoint.context_values); normalize_durable_outcomes(&mut checkpoint.node_outcomes); } @@ -515,6 +519,59 @@ mod tests { ); } + #[test] + fn durable_context_snapshot_drops_parallel_branch_preambles() { + let context = Context::new(); + context.set( + context::keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES, + serde_json::json!({"branch-a": "runtime only"}), + ); + context.set("response.work", serde_json::json!("durable")); + + let snapshot = durable_context_snapshot(&context); + + assert!(!snapshot.contains_key(context::keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES)); + assert_eq!( + snapshot.get("response.work"), + Some(&serde_json::json!("durable")) + ); + } + + #[test] + fn normalize_checkpoint_for_resume_drops_parallel_branch_preambles() { + let mut checkpoint = crate::records::Checkpoint { + timestamp: chrono::Utc::now(), + current_node: "work".to_string(), + completed_nodes: vec!["work".to_string()], + node_retries: HashMap::new(), + context_values: HashMap::from([ + ( + context::keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES.to_string(), + serde_json::json!({"branch-a": "runtime only"}), + ), + ("response.work".to_string(), serde_json::json!("durable")), + ]), + node_outcomes: HashMap::new(), + next_node_id: Some("exit".to_string()), + git_commit_sha: None, + loop_failure_signatures: HashMap::new(), + restart_failure_signatures: HashMap::new(), + node_visits: HashMap::new(), + }; + + normalize_checkpoint_for_resume(&mut checkpoint); + + assert!( + !checkpoint + .context_values + .contains_key(context::keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES) + ); + assert_eq!( + checkpoint.context_values.get("response.work"), + Some(&serde_json::json!("durable")) + ); + } + #[test] fn normalize_checkpoint_for_resume_converts_managed_blob_file_refs_and_drops_preamble() { let blob_id = fabro_types::RunBlobId::new(b"managed"); diff --git a/lib/crates/fabro-workflow/src/context.rs b/lib/crates/fabro-workflow/src/context.rs index af1233586..b911e2210 100644 --- a/lib/crates/fabro-workflow/src/context.rs +++ b/lib/crates/fabro-workflow/src/context.rs @@ -25,6 +25,7 @@ pub mod keys { pub const INTERNAL_PARENT_PREAMBLE: &str = "internal.parent_preamble"; pub const INTERNAL_PARALLEL_GROUP_ID: &str = "internal.parallel_group_id"; pub const INTERNAL_PARALLEL_BRANCH_ID: &str = "internal.parallel_branch_id"; + pub const INTERNAL_PARALLEL_BRANCH_PREAMBLES: &str = "internal.parallel_branch_preambles"; // --- current.* keys --- pub const CURRENT_PREAMBLE: &str = "current.preamble"; diff --git a/lib/crates/fabro-workflow/src/handler/parallel.rs b/lib/crates/fabro-workflow/src/handler/parallel.rs index 22aa2aa9c..ac4e4f3e1 100644 --- a/lib/crates/fabro-workflow/src/handler/parallel.rs +++ b/lib/crates/fabro-workflow/src/handler/parallel.rs @@ -4,6 +4,7 @@ use std::time::Instant; use async_trait::async_trait; use fabro_agent::{Sandbox, WorktreeOptions, WorktreeSandbox}; +use fabro_graphviz::Fidelity; use fabro_graphviz::graph::{AttrValue, Graph, Node}; use fabro_hooks::{HookContext, HookEvent}; use fabro_types::{ParallelBranchId, RunId, StageId}; @@ -56,6 +57,39 @@ struct BranchResult { worktree_path: Option, } +struct BranchPreamble { + fidelity: Fidelity, + preamble: String, +} + +fn parse_branch_preambles( + value: Option, + branch_count: usize, +) -> Option>> { + let serde_json::Value::Array(entries) = value? else { + return None; + }; + if entries.len() != branch_count { + return None; + } + + entries + .into_iter() + .map(|entry| match entry { + serde_json::Value::Null => Some(None), + serde_json::Value::Object(entry) if entry.len() == 2 => { + let fidelity = entry.get("fidelity")?.as_str()?.parse().ok()?; + let preamble = entry.get("preamble")?.as_str()?; + Some(Some(BranchPreamble { + fidelity, + preamble: preamble.to_string(), + })) + } + _ => None, + }) + .collect() +} + #[async_trait] impl Handler for ParallelHandler { async fn simulate( @@ -220,6 +254,10 @@ impl Handler for ParallelHandler { None }; + let branch_preambles = parse_branch_preambles( + context.get(keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES), + branches.len(), + ); let mut branch_setups: Vec = Vec::new(); for (branch_index, edge) in branches.iter().enumerate() { let target_id = edge.to.clone(); @@ -236,6 +274,21 @@ impl Handler for ParallelHandler { keys::INTERNAL_PARALLEL_BRANCH_ID, serde_json::Value::String(parallel_branch_id.to_string()), ); + if let Some(entry) = branch_preambles + .as_ref() + .and_then(|entries| entries.get(branch_index)) + .and_then(Option::as_ref) + { + branch_context.set(keys::CURRENT_PREAMBLE, serde_json::json!(&entry.preamble)); + branch_context.set( + keys::INTERNAL_FIDELITY, + serde_json::json!(entry.fidelity.to_string()), + ); + } + branch_context.set( + keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES, + serde_json::Value::Null, + ); let (branch_sandbox, worktree_path): (Arc, Option) = if let ( Some(ref gs), @@ -297,6 +350,10 @@ impl Handler for ParallelHandler { worktree_path, }); } + context.set( + keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES, + serde_json::Value::Null, + ); // --- Fan out: concurrent execution --- let mut handles = Vec::new(); @@ -693,7 +750,7 @@ fn parallel_branch_commit_cmd( #[cfg(test)] mod tests { - use std::sync::Arc; + use std::sync::{Arc, Mutex}; use std::time::Duration; use fabro_graphviz::graph::{AttrValue, Edge}; @@ -756,6 +813,185 @@ mod tests { context } + #[derive(Clone, Debug, PartialEq)] + struct BranchContextCapture { + node_id: String, + preamble: String, + fidelity: String, + stash: Option, + } + + struct BranchContextRecordingHandler { + captures: Arc>>, + } + + #[async_trait] + impl Handler for BranchContextRecordingHandler { + async fn execute( + &self, + node: &Node, + context: &Context, + _graph: &Graph, + _run_dir: &Path, + _services: &EngineServices, + ) -> Result { + self.captures.lock().unwrap().push(BranchContextCapture { + node_id: node.id.clone(), + preamble: context.preamble(), + fidelity: context.fidelity().to_string(), + stash: context.get(keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES), + }); + Ok(Outcome::success()) + } + } + + async fn execute_with_branch_stash( + stash: Option, + duplicate_target: bool, + ) -> (Context, Vec) { + let captures = Arc::new(Mutex::new(Vec::new())); + let recorder = BranchContextRecordingHandler { + captures: Arc::clone(&captures), + }; + let mut registry = super::super::HandlerRegistry::new(Box::new(recorder)); + registry.register( + "record", + Box::new(BranchContextRecordingHandler { + captures: Arc::clone(&captures), + }), + ); + let mut services = EngineServices::test_default(); + services.registry = Arc::new(registry); + + let mut node = Node::new("par"); + node.attrs.insert( + "shape".to_string(), + AttrValue::String("component".to_string()), + ); + let mut branch_a = Node::new("branch_a"); + branch_a + .attrs + .insert("type".to_string(), AttrValue::String("record".to_string())); + let mut branch_b = Node::new("branch_b"); + branch_b + .attrs + .insert("type".to_string(), AttrValue::String("record".to_string())); + + let mut graph = Graph::new("test"); + graph.nodes.insert(node.id.clone(), node.clone()); + graph.nodes.insert(branch_a.id.clone(), branch_a); + graph.nodes.insert(branch_b.id.clone(), branch_b); + graph.edges.push(Edge::new("par", "branch_a")); + graph.edges.push(Edge::new( + "par", + if duplicate_target { + "branch_a" + } else { + "branch_b" + }, + )); + + let context = test_context(); + context.set(keys::CURRENT_PREAMBLE, serde_json::json!("fork preamble")); + context.set(keys::INTERNAL_FIDELITY, serde_json::json!("compact")); + if let Some(stash) = stash { + context.set(keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES, stash); + } + + let run_dir = tempfile::tempdir().unwrap(); + ParallelHandler + .execute(&node, &context, &graph, run_dir.path(), &services) + .await + .unwrap(); + + let captures = captures.lock().unwrap().clone(); + (context, captures) + } + + #[tokio::test] + async fn parallel_handler_applies_indexed_branch_preambles_and_clears_stash() { + let stash = serde_json::json!([ + {"fidelity": "truncate", "preamble": "branch zero"}, + {"fidelity": "summary:high", "preamble": "branch one"} + ]); + + let (context, mut captures) = execute_with_branch_stash(Some(stash), false).await; + captures.sort_by(|left, right| left.node_id.cmp(&right.node_id)); + + assert_eq!(captures.len(), 2); + assert_eq!(captures[0].node_id, "branch_a"); + assert_eq!(captures[0].preamble, "branch zero"); + assert_eq!(captures[0].fidelity, "truncate"); + assert_eq!(captures[0].stash, Some(serde_json::Value::Null)); + assert_eq!(captures[1].node_id, "branch_b"); + assert_eq!(captures[1].preamble, "branch one"); + assert_eq!(captures[1].fidelity, "summary:high"); + assert_eq!(captures[1].stash, Some(serde_json::Value::Null)); + assert_eq!( + context.get(keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES), + Some(serde_json::Value::Null) + ); + } + + #[tokio::test] + async fn parallel_handler_uses_edge_index_for_duplicate_targets() { + let stash = serde_json::json!([ + {"fidelity": "truncate", "preamble": "first edge"}, + {"fidelity": "summary:low", "preamble": "second edge"} + ]); + + let (_context, captures) = execute_with_branch_stash(Some(stash), true).await; + let observed = captures + .iter() + .map(|capture| (capture.preamble.as_str(), capture.fidelity.as_str())) + .collect::>(); + + assert_eq!(observed.len(), 2); + assert!(observed.contains(&("first edge", "truncate"))); + assert!(observed.contains(&("second edge", "summary:low"))); + assert!( + captures + .iter() + .all(|capture| capture.stash == Some(serde_json::Value::Null)) + ); + } + + #[tokio::test] + async fn parallel_handler_legacy_stashes_inherit_fork_context() { + for stash in [ + None, + Some(serde_json::Value::Null), + Some(serde_json::json!({ + "fidelity": "truncate", + "preamble": "not an array" + })), + Some(serde_json::json!([ + {"fidelity": "truncate", "preamble": "wrong length"} + ])), + Some(serde_json::json!([ + {"fidelity": "truncate"}, + null + ])), + Some(serde_json::json!([ + {"fidelity": "not-a-fidelity", "preamble": "malformed fidelity"}, + null + ])), + ] { + let (context, captures) = execute_with_branch_stash(stash, false).await; + + assert_eq!(captures.len(), 2); + assert!(captures.iter().all(|capture| { + capture.preamble == "fork preamble" + && capture.fidelity == "compact" + && capture.stash == Some(serde_json::Value::Null) + })); + assert_eq!( + context.get(keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES), + Some(serde_json::Value::Null) + ); + } + } + #[tokio::test] async fn parallel_handler_no_branches() { let services = make_services(); diff --git a/lib/crates/fabro-workflow/src/lifecycle/event.rs b/lib/crates/fabro-workflow/src/lifecycle/event.rs index 5828dc86b..3965d7261 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/event.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/event.rs @@ -14,7 +14,7 @@ use fabro_types::{Principal, RunId, StageTiming}; use super::circuit_breaker::CircuitBreakerLifecycle; use super::git::GitCheckpointResult; -use crate::context::WorkflowContext; +use crate::context::{Context, WorkflowContext}; use crate::event::{Emitter, Event, StageScope}; use crate::graph::{WorkflowGraph, WorkflowNode}; use crate::outcome::{BilledModelUsage, FailureCategory, FailureDetail, Outcome, StageOutcome}; @@ -92,6 +92,12 @@ fn response_from_outcome(node_id: &str, outcome: &Outcome) -> Option { .and_then(|value| value.as_str().map(ToOwned::to_owned)) } +fn stage_context_values(workflow_context: &Context) -> Option> { + let mut snapshot = workflow_context.snapshot(); + snapshot.remove(context::keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES); + (!snapshot.is_empty()).then(|| snapshot.into_iter().collect()) +} + pub(super) fn stage_visit(state: &WfRunState, node_id: &str) -> u32 { let visits = state.node_visits.get(node_id).copied().unwrap_or(1); u32::try_from(visits).unwrap_or(u32::MAX) @@ -318,11 +324,7 @@ impl RunLifecycle for EventLifecycle { .collect::>() }), jump_to_node: outcome.jump_to_node.clone(), - context_values: { - let snapshot = state.context.snapshot(); - (!snapshot.is_empty()) - .then(|| snapshot.into_iter().collect::>()) - }, + context_values: stage_context_values(&state.context), node_visits: (!state.node_visits.is_empty()).then(|| { state .node_visits @@ -446,3 +448,26 @@ impl RunLifecycle for EventLifecycle { Ok(()) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn stage_context_values_drops_parallel_branch_preambles() { + let workflow_context = Context::new(); + workflow_context.set( + context::keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES, + serde_json::json!([{"fidelity": "summary:high", "preamble": "runtime only"}]), + ); + workflow_context.set("response.work", serde_json::json!("durable")); + + let values = stage_context_values(&workflow_context).expect("snapshot should not be empty"); + + assert!(!values.contains_key(context::keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES)); + assert_eq!( + values.get("response.work"), + Some(&serde_json::json!("durable")) + ); + } +} diff --git a/lib/crates/fabro-workflow/src/lifecycle/fidelity.rs b/lib/crates/fabro-workflow/src/lifecycle/fidelity.rs index 3f45c8142..3f0b24228 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/fidelity.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/fidelity.rs @@ -78,6 +78,11 @@ impl RunLifecycle for FidelityLifecycle { node: &WorkflowNode, state: &WfRunState, ) -> CoreResult { + state.context.set( + keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES, + serde_json::Value::Null, + ); + let incoming = self .incoming_edge_data .lock() @@ -138,6 +143,46 @@ impl RunLifecycle for FidelityLifecycle { .context .set(keys::CURRENT_PREAMBLE, serde_json::json!(preamble)); + if gv_node.handler_type() == Some("parallel") { + let mut branch_preambles = Vec::new(); + for (branch_index, edge) in self.graph.outgoing_edges(node.id()).iter().enumerate() { + let Some(target_node) = self.graph.nodes.get(&edge.to) else { + branch_preambles.push(serde_json::Value::Null); + continue; + }; + let resolution = resolve_parallel_branch_fidelity(edge, target_node, fidelity); + if resolution.requested() == Some(keys::Fidelity::Full) { + tracing::warn!( + parallel_node = %node.id(), + branch = %edge.to, + branch_index, + fidelity = %keys::Fidelity::Full, + effective_fidelity = %keys::Fidelity::SummaryHigh, + "Parallel branch fidelity degraded" + ); + } + let Some(branch_fidelity) = resolution.entry_fidelity() else { + branch_preambles.push(serde_json::Value::Null); + continue; + }; + let branch_preamble = build_preamble( + branch_fidelity, + &resolved_context, + &self.graph, + &state.completed_nodes, + &resolved_outcomes, + ); + branch_preambles.push(serde_json::json!({ + "fidelity": branch_fidelity.to_string(), + "preamble": branch_preamble, + })); + } + state.context.set( + keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES, + serde_json::Value::Array(branch_preambles), + ); + } + // 5. Thread ID resolution via resolve_thread_id: edge → node → graph default → // class → previous let thread_id = resolve_thread_id( @@ -198,6 +243,47 @@ impl RunLifecycle for FidelityLifecycle { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct ParallelBranchFidelityResolution { + requested: Option, + effective: Option, +} + +impl ParallelBranchFidelityResolution { + fn requested(self) -> Option { + self.requested + } + + fn entry_fidelity(self) -> Option { + self.effective + } +} + +/// Resolve explicit branch fidelity with edge-over-node precedence. +/// +/// Branches with no explicit fidelity inherit the parallel node's preamble. +/// Explicit full fidelity is degraded because concurrent branches cannot share +/// an LLM session. An effective fidelity equal to the parallel node also +/// inherits, avoiding a redundant preamble render. +fn resolve_parallel_branch_fidelity( + edge: &GvEdge, + target_node: &GvNode, + parallel_fidelity: keys::Fidelity, +) -> ParallelBranchFidelityResolution { + let requested = edge + .fidelity() + .and_then(|value| value.parse().ok()) + .or_else(|| target_node.fidelity().and_then(|value| value.parse().ok())); + let effective = requested + .map(keys::Fidelity::degraded) + .filter(|fidelity| *fidelity != parallel_fidelity); + + ParallelBranchFidelityResolution { + requested, + effective, + } +} + /// Resolve the context fidelity for a node, following the precedence: /// 1. Incoming edge `fidelity` attribute /// 2. Target node `fidelity` attribute @@ -263,11 +349,216 @@ fn resolve_thread_id( #[cfg(test)] mod tests { + use std::path::Path; + use std::time::Duration; + + use fabro_core::graph::Graph as CoreGraph; use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node}; + use fabro_store::Database; + use fabro_types::fixtures; + use object_store::memory::InMemory; use super::*; + use crate::context::WorkflowContext; use crate::context::keys::Fidelity; + fn fidelity_attr(value: &str) -> AttrValue { + AttrValue::String(value.to_string()) + } + + fn parallel_workflow_graph( + fork_fidelity: Option<&str>, + branch_a_fidelity: Option<&str>, + ) -> WorkflowGraph { + let mut graph = Graph::new("parallel-fidelity"); + let mut start = Node::new("start"); + start + .attrs + .insert("shape".to_string(), fidelity_attr("Mdiamond")); + let mut fork = Node::new("fork"); + fork.attrs + .insert("shape".to_string(), fidelity_attr("component")); + if let Some(fidelity) = fork_fidelity { + fork.attrs + .insert("fidelity".to_string(), fidelity_attr(fidelity)); + } + let mut branch_a = Node::new("branch_a"); + if let Some(fidelity) = branch_a_fidelity { + branch_a + .attrs + .insert("fidelity".to_string(), fidelity_attr(fidelity)); + } + let branch_b = Node::new("branch_b"); + let mut work = Node::new("work"); + work.attrs.insert("shape".to_string(), fidelity_attr("box")); + + graph.nodes.insert(start.id.clone(), start); + graph.nodes.insert(fork.id.clone(), fork); + graph.nodes.insert(branch_a.id.clone(), branch_a); + graph.nodes.insert(branch_b.id.clone(), branch_b); + graph.nodes.insert(work.id.clone(), work); + graph.edges.push(Edge::new("start", "fork")); + graph.edges.push(Edge::new("fork", "branch_a")); + graph.edges.push(Edge::new("fork", "branch_b")); + + WorkflowGraph(Arc::new(graph)) + } + + async fn test_lifecycle(graph: &WorkflowGraph, run_dir: &Path) -> FidelityLifecycle { + let store = Arc::new(Database::new( + Arc::new(InMemory::new()), + "", + Duration::from_millis(1), + None, + )); + let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); + let sandbox: Arc = + Arc::new(fabro_agent::LocalSandbox::new(run_dir.to_path_buf())); + FidelityLifecycle::new( + graph.0.clone(), + sandbox, + RunStoreHandle::local(run_store), + run_dir.to_path_buf(), + ) + } + + #[test] + fn parallel_branch_fidelity_edge_overrides_node() { + let mut node = Node::new("branch"); + node.attrs + .insert("fidelity".to_string(), fidelity_attr("compact")); + let mut edge = Edge::new("fork", "branch"); + edge.attrs + .insert("fidelity".to_string(), fidelity_attr("truncate")); + + let resolved = resolve_parallel_branch_fidelity(&edge, &node, Fidelity::SummaryHigh); + + assert_eq!(resolved.requested(), Some(Fidelity::Truncate)); + assert_eq!(resolved.entry_fidelity(), Some(Fidelity::Truncate)); + } + + #[test] + fn parallel_branch_fidelity_without_attribute_inherits() { + let node = Node::new("branch"); + let edge = Edge::new("fork", "branch"); + + let resolution = resolve_parallel_branch_fidelity(&edge, &node, Fidelity::Compact); + + assert_eq!(resolution.requested(), None); + assert_eq!(resolution.entry_fidelity(), None); + } + + #[test] + fn parallel_branch_full_fidelity_degrades_to_summary_high() { + let mut node = Node::new("branch"); + node.attrs + .insert("fidelity".to_string(), fidelity_attr("full")); + let edge = Edge::new("fork", "branch"); + + let resolved = resolve_parallel_branch_fidelity(&edge, &node, Fidelity::Compact); + + assert_eq!(resolved.requested(), Some(Fidelity::Full)); + assert_eq!(resolved.entry_fidelity(), Some(Fidelity::SummaryHigh)); + } + + #[test] + fn parallel_branch_fidelity_equal_to_fork_inherits() { + let mut node = Node::new("branch"); + node.attrs + .insert("fidelity".to_string(), fidelity_attr("summary:high")); + let edge = Edge::new("fork", "branch"); + + let resolution = resolve_parallel_branch_fidelity(&edge, &node, Fidelity::SummaryHigh); + + assert_eq!(resolution.requested(), Some(Fidelity::SummaryHigh)); + assert_eq!(resolution.entry_fidelity(), None); + } + + #[test] + fn explicit_full_branch_equal_to_degraded_fork_inherits() { + let mut node = Node::new("branch"); + node.attrs + .insert("fidelity".to_string(), fidelity_attr("full")); + let edge = Edge::new("fork", "branch"); + + let resolution = resolve_parallel_branch_fidelity(&edge, &node, Fidelity::SummaryHigh); + + assert_eq!(resolution.requested(), Some(Fidelity::Full)); + assert_eq!(resolution.entry_fidelity(), None); + } + + #[test] + fn full_fork_without_branch_fidelity_does_not_create_entry() { + let node = Node::new("branch"); + let edge = Edge::new("fork", "branch"); + + let resolution = resolve_parallel_branch_fidelity(&edge, &node, Fidelity::Full); + + assert_eq!(resolution.requested(), None); + assert_eq!(resolution.entry_fidelity(), None); + } + + #[tokio::test] + async fn parallel_before_node_rebuilds_branch_preamble_stash() { + let graph = parallel_workflow_graph(None, Some("truncate")); + let run_dir = tempfile::tempdir().unwrap(); + let lifecycle = test_lifecycle(&graph, run_dir.path()).await; + let state: WfRunState = ExecutionState::new(&graph).unwrap(); + let fork = graph.get_node("fork").unwrap(); + + lifecycle.before_node(&fork, &state).await.unwrap(); + state.context.set( + keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES, + serde_json::json!(["stale", "entries", "must disappear"]), + ); + lifecycle.before_node(&fork, &state).await.unwrap(); + + let stash = state + .context + .get(keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES) + .expect("parallel stash should be set"); + let entries = stash.as_array().expect("parallel stash should be an array"); + assert_eq!(entries.len(), 2); + assert!(entries[0].is_object()); + assert!(entries[1].is_null()); + } + + #[tokio::test] + async fn non_parallel_before_node_overwrites_branch_preamble_stash_with_null() { + let graph = parallel_workflow_graph(None, Some("truncate")); + let run_dir = tempfile::tempdir().unwrap(); + let lifecycle = test_lifecycle(&graph, run_dir.path()).await; + let state: WfRunState = ExecutionState::new(&graph).unwrap(); + let fork = graph.get_node("fork").unwrap(); + let work = graph.get_node("work").unwrap(); + + lifecycle.before_node(&fork, &state).await.unwrap(); + lifecycle.before_node(&work, &state).await.unwrap(); + + assert_eq!( + state.context.get(keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES), + Some(serde_json::Value::Null) + ); + } + + #[tokio::test] + async fn resumed_full_fork_degrades_without_rendering_fallback_branches() { + let graph = parallel_workflow_graph(Some("full"), None); + let run_dir = tempfile::tempdir().unwrap(); + let lifecycle = test_lifecycle(&graph, run_dir.path()).await; + lifecycle.set_degrade_fidelity_on_resume(true); + let state: WfRunState = ExecutionState::new(&graph).unwrap(); + let fork = graph.get_node("fork").unwrap(); + + lifecycle.before_node(&fork, &state).await.unwrap(); + + assert_eq!(state.context.fidelity(), Fidelity::SummaryHigh); + assert_eq!( + state.context.get(keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES), + Some(serde_json::json!([null, null])) + ); + } + #[test] fn fidelity_defaults_to_compact() { let node = Node::new("work"); diff --git a/lib/crates/fabro-workflow/tests/it/integration.rs b/lib/crates/fabro-workflow/tests/it/integration.rs index 87ebeee4d..88e1f1aa9 100644 --- a/lib/crates/fabro-workflow/tests/it/integration.rs +++ b/lib/crates/fabro-workflow/tests/it/integration.rs @@ -4936,6 +4936,27 @@ struct FidelityCapturingHandler { captures: FidelityCaptures, } +struct ParallelFidelitySeedHandler; + +#[async_trait::async_trait] +impl Handler for ParallelFidelitySeedHandler { + async fn execute( + &self, + _node: &Node, + _context: &Context, + _graph: &Graph, + _run_dir: &Path, + _services: &fabro_workflow::handler::EngineServices, + ) -> Result { + let mut outcome = Outcome::success(); + outcome.context_updates.insert( + "parallel_fidelity_marker".to_string(), + serde_json::json!("marker visible to inherited preambles"), + ); + Ok(outcome) + } +} + #[async_trait::async_trait] impl Handler for FidelityCapturingHandler { async fn execute( @@ -9333,6 +9354,176 @@ async fn run_fidelity_prompt_pipeline(fidelity: &str) -> String { .expect("report prompt should exist") } +async fn run_parallel_fidelity_capture( + fork_fidelity: Option<&str>, + branch_node_fidelity: Option<&str>, + branch_edge_fidelity: Option<&str>, +) -> FidelityCaptures { + use fabro_workflow::handler::fan_in::FanInHandler; + use fabro_workflow::handler::parallel::ParallelHandler; + + let mut graph = make_graph_with_start_exit("ParallelFidelityTest"); + graph.attrs.insert( + "goal".to_string(), + AttrValue::String("Verify parallel branch context".to_string()), + ); + + let mut seed = Node::new("seed"); + seed.attrs.insert( + "type".to_string(), + AttrValue::String("parallel_fidelity_seed".to_string()), + ); + let mut fork = Node::new("fork"); + fork.attrs.insert( + "shape".to_string(), + AttrValue::String("component".to_string()), + ); + if let Some(fidelity) = fork_fidelity { + fork.attrs.insert( + "fidelity".to_string(), + AttrValue::String(fidelity.to_string()), + ); + } + let mut branch_a = Node::new("branch_a"); + branch_a.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + if let Some(fidelity) = branch_node_fidelity { + branch_a.attrs.insert( + "fidelity".to_string(), + AttrValue::String(fidelity.to_string()), + ); + } + let mut branch_b = Node::new("branch_b"); + branch_b.attrs.insert( + "type".to_string(), + AttrValue::String("fidelity_capture".to_string()), + ); + let mut fan_in = Node::new("fan_in"); + fan_in.attrs.insert( + "shape".to_string(), + AttrValue::String("tripleoctagon".to_string()), + ); + + graph.nodes.insert(seed.id.clone(), seed); + graph.nodes.insert(fork.id.clone(), fork); + graph.nodes.insert(branch_a.id.clone(), branch_a); + graph.nodes.insert(branch_b.id.clone(), branch_b); + graph.nodes.insert(fan_in.id.clone(), fan_in); + graph.edges.push(Edge::new("start", "seed")); + graph.edges.push(Edge::new("seed", "fork")); + let mut branch_a_edge = Edge::new("fork", "branch_a"); + if let Some(fidelity) = branch_edge_fidelity { + branch_a_edge.attrs.insert( + "fidelity".to_string(), + AttrValue::String(fidelity.to_string()), + ); + } + graph.edges.push(branch_a_edge); + graph.edges.push(Edge::new("fork", "branch_b")); + graph.edges.push(Edge::new("branch_a", "fan_in")); + graph.edges.push(Edge::new("branch_b", "fan_in")); + graph.edges.push(Edge::new("fan_in", "exit")); + + let captures = FidelityCaptures::new(); + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("parallel", Box::new(ParallelHandler)); + registry.register( + "parallel.fan_in", + Box::new(FanInHandler::new(Some(Box::new(MockCodergenBackend)))), + ); + registry.register( + "parallel_fidelity_seed", + Box::new(ParallelFidelitySeedHandler), + ); + registry.register( + "fidelity_capture", + Box::new(FidelityCapturingHandler { + captures: captures.clone(), + }), + ); + + let dir = tempfile::tempdir().expect("parallel fidelity run directory should be created"); + let engine = WorkflowRunner::new(registry, Arc::new(Emitter::default()), local_env()); + let run_options = RunOptions { + settings: WorkflowSettings::default(), + run_dir: dir.path().to_path_buf(), + cancel_token: CancellationToken::new(), + run_id: test_run_id("parallel-fidelity"), + labels: std::collections::HashMap::new(), + workflow_slug: None, + github_app: None, + base_branch: None, + display_base_sha: None, + pre_run_git: None, + fork_source_ref: None, + git: None, + }; + let (outcome, _state) = engine + .run_with_state(&graph, &run_options) + .await + .expect("parallel fidelity workflow should succeed"); + assert_eq!(outcome.status, StageOutcome::Succeeded); + captures +} + +fn captured_fidelity_preamble(captures: &FidelityCaptures, node_id: &str) -> (String, String) { + let fidelity = captures + .fidelities + .lock() + .unwrap() + .iter() + .find(|(captured_node_id, _)| captured_node_id == node_id) + .map(|(_, fidelity)| fidelity.clone()) + .expect("branch fidelity should be captured"); + let preamble = captures + .preambles + .lock() + .unwrap() + .iter() + .find(|(captured_node_id, _)| captured_node_id == node_id) + .map(|(_, preamble)| preamble.clone()) + .expect("branch preamble should be captured"); + (fidelity, preamble) +} + +#[tokio::test] +async fn parallel_branches_get_per_branch_preambles_by_fidelity() { + let captures = run_parallel_fidelity_capture(None, Some("truncate"), None).await; + + let (branch_a_fidelity, branch_a_preamble) = captured_fidelity_preamble(&captures, "branch_a"); + let (branch_b_fidelity, branch_b_preamble) = captured_fidelity_preamble(&captures, "branch_b"); + + assert_eq!(branch_a_fidelity, "truncate"); + assert!(!branch_a_preamble.contains("parallel_fidelity_marker")); + assert_eq!(branch_b_fidelity, "compact"); + assert!(branch_b_preamble.contains("parallel_fidelity_marker")); +} + +#[tokio::test] +async fn parallel_fork_fidelity_still_applies_to_all_branches() { + let captures = run_parallel_fidelity_capture(Some("truncate"), None, None).await; + + for branch_id in ["branch_a", "branch_b"] { + let (fidelity, preamble) = captured_fidelity_preamble(&captures, branch_id); + assert_eq!(fidelity, "truncate"); + assert!(!preamble.contains("parallel_fidelity_marker")); + } +} + +#[tokio::test] +async fn parallel_branch_edge_fidelity_overrides_node_fidelity() { + let captures = + run_parallel_fidelity_capture(None, Some("summary:high"), Some("truncate")).await; + + let (fidelity, preamble) = captured_fidelity_preamble(&captures, "branch_a"); + assert_eq!(fidelity, "truncate"); + assert!(!preamble.contains("parallel_fidelity_marker")); +} + #[tokio::test] async fn fidelity_prompt_compact() { let prompt = run_fidelity_prompt_pipeline("compact").await; From 56119990beb0cfbcbec078e331389ec8becc4e6c Mon Sep 17 00:00:00 2001 From: Fabro Date: Thu, 23 Jul 2026 19:10:39 +0000 Subject: [PATCH 05/15] fabro(01KY7Y01REECZ24XXTMBZ3PPV9): simplify_fable (succeeded) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fabro-Run: 01KY7Y01REECZ24XXTMBZ3PPV9 Fabro-Completed: 6 Fabro-Checkpoint: 497f92f4d91a807f2d66934a0a7e9ccc36cec5fd ⚒️ Generated with [Fabro](https://fabro.sh) --- Cargo.lock | 1 + lib/crates/fabro-graphviz/Cargo.toml | 3 + lib/crates/fabro-graphviz/src/fidelity.rs | 30 ++- .../src/rules/parallel_branch.rs | 27 +-- .../rules/parallel_branch_inert_attribute.rs | 74 +++---- .../rules/thread_id_requires_fidelity_full.rs | 3 + lib/crates/fabro-workflow/src/artifact.rs | 17 +- lib/crates/fabro-workflow/src/context.rs | 24 +++ .../fabro-workflow/src/handler/parallel.rs | 47 ++-- .../fabro-workflow/src/lifecycle/event.rs | 4 + .../fabro-workflow/src/lifecycle/fidelity.rs | 202 ++++++++++-------- 11 files changed, 262 insertions(+), 170 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 098d71cbf..6a4a8069d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2659,6 +2659,7 @@ dependencies = [ "nom", "regex", "serde", + "serde_json", "strum 0.28.0", "thiserror 2.0.18", ] diff --git a/lib/crates/fabro-graphviz/Cargo.toml b/lib/crates/fabro-graphviz/Cargo.toml index 98e017097..3b33151d8 100644 --- a/lib/crates/fabro-graphviz/Cargo.toml +++ b/lib/crates/fabro-graphviz/Cargo.toml @@ -21,3 +21,6 @@ regex = { workspace = true } serde = { workspace = true } strum.workspace = true thiserror = { workspace = true } + +[dev-dependencies] +serde_json = { workspace = true } diff --git a/lib/crates/fabro-graphviz/src/fidelity.rs b/lib/crates/fabro-graphviz/src/fidelity.rs index 8b33c98e8..98605c135 100644 --- a/lib/crates/fabro-graphviz/src/fidelity.rs +++ b/lib/crates/fabro-graphviz/src/fidelity.rs @@ -1,9 +1,24 @@ +use serde::{Deserialize, Serialize}; use strum::{Display, EnumString, VariantArray}; /// Fidelity mode controlling how much prior context is provided to LLM /// sessions. -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Display, EnumString, VariantArray)] +#[derive( + Debug, + Clone, + Copy, + Default, + PartialEq, + Eq, + Hash, + Display, + EnumString, + VariantArray, + Serialize, + Deserialize, +)] #[strum(serialize_all = "lowercase")] +#[serde(rename_all = "lowercase")] pub enum Fidelity { /// Complete context, no summarization — sessions share a thread. Full, @@ -14,12 +29,15 @@ pub enum Fidelity { Compact, /// Brief textual summary (~600 token target). #[strum(serialize = "summary:low")] + #[serde(rename = "summary:low")] SummaryLow, /// Moderate textual summary (~1500 token target). #[strum(serialize = "summary:medium")] + #[serde(rename = "summary:medium")] SummaryMedium, /// Detailed per-stage Markdown report. #[strum(serialize = "summary:high")] + #[serde(rename = "summary:high")] SummaryHigh, } @@ -73,4 +91,14 @@ mod tests { fn fidelity_unknown_mode_errors() { assert!("bogus".parse::().is_err()); } + + #[test] + fn fidelity_serde_matches_strum_display() { + for mode in Fidelity::variants() { + let json = serde_json::to_value(mode).unwrap(); + assert_eq!(json, serde_json::Value::String(mode.to_string())); + let parsed: Fidelity = serde_json::from_value(json).unwrap(); + assert_eq!(parsed, *mode); + } + } } diff --git a/lib/crates/fabro-validate/src/rules/parallel_branch.rs b/lib/crates/fabro-validate/src/rules/parallel_branch.rs index 2f1cba2e7..d8a7d2fbd 100644 --- a/lib/crates/fabro-validate/src/rules/parallel_branch.rs +++ b/lib/crates/fabro-validate/src/rules/parallel_branch.rs @@ -35,22 +35,25 @@ impl<'a> ParallelBranches<'a> { .collect() } + /// True when every incoming edge of `node_id` comes from a parallel fork + /// (and there is at least one). Such a node only ever runs as a branch. pub(super) fn is_branch_only_node(&self, node_id: &str) -> bool { - self.branch_only_parents(node_id).is_some() + let incoming = self.graph.incoming_edges(node_id); + !incoming.is_empty() && incoming.iter().all(|edge| self.is_fork_edge(edge)) } + /// The sorted, deduplicated fork parents of a branch-only node, or `None` + /// when the node has a non-fork entry path (or no entry at all). pub(super) fn branch_only_parents(&self, node_id: &str) -> Option> { - let mut incoming = self + if !self.is_branch_only_node(node_id) { + return None; + } + let parents: BTreeSet<&str> = self .graph - .edges - .iter() - .filter(|edge| edge.to == node_id) - .peekable(); - incoming.peek()?; - - incoming - .map(|edge| self.is_fork_edge(edge).then(|| edge.from.clone())) - .collect::>>() - .map(|parents| parents.into_iter().collect()) + .incoming_edges(node_id) + .into_iter() + .map(|edge| edge.from.as_str()) + .collect(); + Some(parents.into_iter().map(String::from).collect()) } } diff --git a/lib/crates/fabro-validate/src/rules/parallel_branch_inert_attribute.rs b/lib/crates/fabro-validate/src/rules/parallel_branch_inert_attribute.rs index da36dbbea..bac461103 100644 --- a/lib/crates/fabro-validate/src/rules/parallel_branch_inert_attribute.rs +++ b/lib/crates/fabro-validate/src/rules/parallel_branch_inert_attribute.rs @@ -7,11 +7,15 @@ pub(super) fn rule() -> Box { Box::new(Rule) } -/// Attributes that parallel branch execution does not resolve. Per-branch -/// preambles now honor fidelity, while `thread_id` remains inert because -/// concurrent branches cannot share an LLM session. +/// Attributes that parallel branch execution does not resolve. Only +/// `thread_id` is inert on branches (concurrent branches cannot share an LLM +/// session); per-branch `fidelity` is honored via pre-rendered preambles. const BRANCH_IGNORED_ATTRS: &[&str] = &["thread_id"]; +const FULL_FIDELITY_MESSAGE: &str = "Parallel branches run at most at summary:high; full is degraded at runtime because branches cannot share a session"; + +const THREAD_ID_FIX: &str = "Remove 'thread_id': parallel branches inherit the thread resolved when the parallel node started"; + struct Rule; /// Renders one or more parallel-node ids as `'a'` or `'a', 'b'`. @@ -22,19 +26,6 @@ fn quoted_list(ids: &[String]) -> String { .join(", ") } -fn fix_message(attr: &str) -> String { - match attr { - "thread_id" => format!( - "Remove '{attr}': parallel branches inherit the thread resolved when the parallel node started" - ), - _ => format!("Remove '{attr}'"), - } -} - -fn full_fidelity_message() -> String { - "parallel branches run at most at summary:high; full is degraded at runtime because branches cannot share a session".to_string() -} - fn full_fidelity_fix(parallel_ids: &[String]) -> String { let parent = if parallel_ids.len() == 1 { format!("parallel node {}", quoted_list(parallel_ids)) @@ -46,6 +37,23 @@ fn full_fidelity_fix(parallel_ids: &[String]) -> String { ) } +fn full_fidelity_diagnostic( + rule_name: &str, + node_id: Option, + edge: Option<(String, String)>, + parallel_ids: &[String], +) -> Diagnostic { + Diagnostic { + rule: rule_name.to_string(), + severity: Severity::Warning, + message: FULL_FIDELITY_MESSAGE.to_string(), + node_id, + edge, + fix: Some(full_fidelity_fix(parallel_ids)), + ..Diagnostic::default() + } +} + impl LintRule for Rule { fn name(&self) -> &'static str { "parallel_branch_inert_attribute" @@ -64,15 +72,12 @@ impl LintRule for Rule { continue; } if edge.fidelity() == Some("full") { - diagnostics.push(Diagnostic { - rule: self.name().to_string(), - severity: Severity::Warning, - message: full_fidelity_message(), - node_id: None, - edge: Some((edge.from.clone(), edge.to.clone())), - fix: Some(full_fidelity_fix(std::slice::from_ref(&edge.from))), - ..Diagnostic::default() - }); + diagnostics.push(full_fidelity_diagnostic( + self.name(), + None, + Some((edge.from.clone(), edge.to.clone())), + std::slice::from_ref(&edge.from), + )); } for attr in BRANCH_IGNORED_ATTRS { if !edge.attrs.contains_key(*attr) { @@ -87,7 +92,7 @@ impl LintRule for Rule { ), node_id: None, edge: Some((edge.from.clone(), edge.to.clone())), - fix: Some(fix_message(attr)), + fix: Some(THREAD_ID_FIX.to_string()), ..Diagnostic::default() }); } @@ -103,15 +108,12 @@ impl LintRule for Rule { continue; }; if node.fidelity() == Some("full") { - diagnostics.push(Diagnostic { - rule: self.name().to_string(), - severity: Severity::Warning, - message: full_fidelity_message(), - node_id: Some(node.id.clone()), - edge: None, - fix: Some(full_fidelity_fix(&parents)), - ..Diagnostic::default() - }); + diagnostics.push(full_fidelity_diagnostic( + self.name(), + Some(node.id.clone()), + None, + &parents, + )); } for attr in BRANCH_IGNORED_ATTRS { if !node.attrs.contains_key(*attr) { @@ -127,7 +129,7 @@ impl LintRule for Rule { ), node_id: Some(node.id.clone()), edge: None, - fix: Some(fix_message(attr)), + fix: Some(THREAD_ID_FIX.to_string()), ..Diagnostic::default() }); } diff --git a/lib/crates/fabro-validate/src/rules/thread_id_requires_fidelity_full.rs b/lib/crates/fabro-validate/src/rules/thread_id_requires_fidelity_full.rs index a73335da5..157b2a032 100644 --- a/lib/crates/fabro-validate/src/rules/thread_id_requires_fidelity_full.rs +++ b/lib/crates/fabro-validate/src/rules/thread_id_requires_fidelity_full.rs @@ -23,6 +23,9 @@ impl LintRule for Rule { let graph_default_full = graph.default_fidelity() == Some("full"); let branches = ParallelBranches::new(graph); + // thread_id is inert on parallel branches, where + // parallel_branch_inert_attribute already says "remove thread_id" — + // advising fidelity="full" there would contradict it. for node in graph.nodes.values() { if node.thread_id().is_some() && !branches.is_branch_only_node(&node.id) diff --git a/lib/crates/fabro-workflow/src/artifact.rs b/lib/crates/fabro-workflow/src/artifact.rs index 975fee26d..bbf7f7ed8 100644 --- a/lib/crates/fabro-workflow/src/artifact.rs +++ b/lib/crates/fabro-workflow/src/artifact.rs @@ -78,12 +78,18 @@ pub fn format_artifact_reference(path: &str) -> String { pub fn durable_context_snapshot(context: &Context) -> HashMap { let mut snapshot = context.snapshot(); - snapshot.remove(context::keys::CURRENT_PREAMBLE); - snapshot.remove(context::keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES); + strip_transient_keys(&mut snapshot); normalize_durable_updates(&mut snapshot); snapshot } +/// Remove runtime-only keys that must never reach durable storage or events. +pub fn strip_transient_keys(values: &mut HashMap) { + for key in context::keys::TRANSIENT_CONTEXT_KEYS { + values.remove(*key); + } +} + pub fn normalize_durable_updates(updates: &mut HashMap) { for value in updates.values_mut() { normalize_durable_value(value); @@ -97,12 +103,7 @@ pub fn normalize_durable_outcomes(node_outcomes: &mut HashMap) } pub fn normalize_checkpoint_for_resume(checkpoint: &mut Checkpoint) { - checkpoint - .context_values - .remove(context::keys::CURRENT_PREAMBLE); - checkpoint - .context_values - .remove(context::keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES); + strip_transient_keys(&mut checkpoint.context_values); normalize_durable_updates(&mut checkpoint.context_values); normalize_durable_outcomes(&mut checkpoint.node_outcomes); } diff --git a/lib/crates/fabro-workflow/src/context.rs b/lib/crates/fabro-workflow/src/context.rs index b911e2210..a890c0e8f 100644 --- a/lib/crates/fabro-workflow/src/context.rs +++ b/lib/crates/fabro-workflow/src/context.rs @@ -25,6 +25,9 @@ pub mod keys { pub const INTERNAL_PARENT_PREAMBLE: &str = "internal.parent_preamble"; pub const INTERNAL_PARALLEL_GROUP_ID: &str = "internal.parallel_group_id"; pub const INTERNAL_PARALLEL_BRANCH_ID: &str = "internal.parallel_branch_id"; + /// Stash of pre-rendered per-branch preambles for a parallel node; see + /// [`super::ParallelBranchPreamble`] for the entry shape and the + /// producer/consumer contract. pub const INTERNAL_PARALLEL_BRANCH_PREAMBLES: &str = "internal.parallel_branch_preambles"; // --- current.* keys --- @@ -45,6 +48,12 @@ pub mod keys { pub const PARALLEL_FAN_IN_BEST_OUTCOME: &str = "parallel.fan_in.best_outcome"; pub const PARALLEL_FAN_IN_BEST_HEAD_SHA: &str = "parallel.fan_in.best_head_sha"; + /// Runtime-only keys stripped from durable context projections + /// (checkpoint snapshots and resume normalization). Add new transient + /// keys here so both strip sites stay in sync. + pub const TRANSIENT_CONTEXT_KEYS: &[&str] = + &[CURRENT_PREAMBLE, INTERNAL_PARALLEL_BRANCH_PREAMBLES]; + // --- Prefix constants (for filtering and dynamic keys) --- pub const GRAPH_PREFIX: &str = "graph."; pub const INTERNAL_PREFIX: &str = "internal."; @@ -136,9 +145,24 @@ pub mod keys { pub use fabro_core::Context; use fabro_graphviz::Fidelity; use fabro_types::{ParallelBranchId, StageId}; +use serde::{Deserialize, Serialize}; use crate::event::StageScope; +/// One entry of the [`keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES`] stash. +/// +/// The stash is a JSON array indexed by the parallel node's outgoing-edge +/// order (`Graph::outgoing_edges` preserves declaration order, so producer and +/// consumer align even with duplicate targets). `null` entries mean the branch +/// inherits the fork's preamble. `FidelityLifecycle::before_node` produces the +/// stash; `ParallelHandler::execute` consumes and clears it. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ParallelBranchPreamble { + pub fidelity: Fidelity, + pub preamble: String, +} + /// Domain-specific typed accessors for workflow context values. pub trait WorkflowContext { fn fidelity(&self) -> Fidelity; diff --git a/lib/crates/fabro-workflow/src/handler/parallel.rs b/lib/crates/fabro-workflow/src/handler/parallel.rs index ac4e4f3e1..32e4a8b10 100644 --- a/lib/crates/fabro-workflow/src/handler/parallel.rs +++ b/lib/crates/fabro-workflow/src/handler/parallel.rs @@ -4,14 +4,13 @@ use std::time::Instant; use async_trait::async_trait; use fabro_agent::{Sandbox, WorktreeOptions, WorktreeSandbox}; -use fabro_graphviz::Fidelity; use fabro_graphviz::graph::{AttrValue, Graph, Node}; use fabro_hooks::{HookContext, HookEvent}; use fabro_types::{ParallelBranchId, RunId, StageId}; use tokio::sync::Semaphore; use super::{EngineServices, Handler}; -use crate::context::{Context, WorkflowContext, keys}; +use crate::context::{Context, ParallelBranchPreamble, WorkflowContext, keys}; use crate::error::Error; use crate::event::{Event, RunNoticeCode, RunNoticeLevel, StageScope}; use crate::git::sanitize_ref_component; @@ -57,15 +56,15 @@ struct BranchResult { worktree_path: Option, } -struct BranchPreamble { - fidelity: Fidelity, - preamble: String, -} - +/// Parse the per-branch preamble stash produced by `FidelityLifecycle`. +/// +/// Outer `None` means the stash is absent, malformed, or has the wrong branch +/// count — every branch then inherits the fork context (legacy behavior). +/// Inner `None` means that single branch inherits. fn parse_branch_preambles( value: Option, branch_count: usize, -) -> Option>> { +) -> Option>> { let serde_json::Value::Array(entries) = value? else { return None; }; @@ -77,15 +76,7 @@ fn parse_branch_preambles( .into_iter() .map(|entry| match entry { serde_json::Value::Null => Some(None), - serde_json::Value::Object(entry) if entry.len() == 2 => { - let fidelity = entry.get("fidelity")?.as_str()?.parse().ok()?; - let preamble = entry.get("preamble")?.as_str()?; - Some(Some(BranchPreamble { - fidelity, - preamble: preamble.to_string(), - })) - } - _ => None, + entry => serde_json::from_value(entry).ok().map(Some), }) .collect() } @@ -258,6 +249,13 @@ impl Handler for ParallelHandler { context.get(keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES), branches.len(), ); + // Clear the stash before forking so branch contexts never carry the + // outer array — a nested parallel branch target must not misread it as + // its own. The write-back diff also clears it on the run state. + context.set( + keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES, + serde_json::Value::Null, + ); let mut branch_setups: Vec = Vec::new(); for (branch_index, edge) in branches.iter().enumerate() { let target_id = edge.to.clone(); @@ -279,16 +277,15 @@ impl Handler for ParallelHandler { .and_then(|entries| entries.get(branch_index)) .and_then(Option::as_ref) { - branch_context.set(keys::CURRENT_PREAMBLE, serde_json::json!(&entry.preamble)); + branch_context.set( + keys::CURRENT_PREAMBLE, + serde_json::Value::String(entry.preamble.clone()), + ); branch_context.set( keys::INTERNAL_FIDELITY, - serde_json::json!(entry.fidelity.to_string()), + serde_json::Value::String(entry.fidelity.to_string()), ); } - branch_context.set( - keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES, - serde_json::Value::Null, - ); let (branch_sandbox, worktree_path): (Arc, Option) = if let ( Some(ref gs), @@ -350,10 +347,6 @@ impl Handler for ParallelHandler { worktree_path, }); } - context.set( - keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES, - serde_json::Value::Null, - ); // --- Fan out: concurrent execution --- let mut handles = Vec::new(); diff --git a/lib/crates/fabro-workflow/src/lifecycle/event.rs b/lib/crates/fabro-workflow/src/lifecycle/event.rs index 3965d7261..f547f5fe5 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/event.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/event.rs @@ -92,6 +92,10 @@ fn response_from_outcome(node_id: &str, outcome: &Outcome) -> Option { .and_then(|value| value.as_str().map(ToOwned::to_owned)) } +/// Context values for `StageCompleted` events. Unlike +/// `artifact::strip_transient_keys`, this keeps `CURRENT_PREAMBLE` — stage +/// events have always included the active preamble — and drops only the +/// parallel stash, which can embed every branch's rendered preamble. fn stage_context_values(workflow_context: &Context) -> Option> { let mut snapshot = workflow_context.snapshot(); snapshot.remove(context::keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES); diff --git a/lib/crates/fabro-workflow/src/lifecycle/fidelity.rs b/lib/crates/fabro-workflow/src/lifecycle/fidelity.rs index 3f0b24228..437550cc1 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/fidelity.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/fidelity.rs @@ -1,3 +1,4 @@ +use std::collections::HashMap; use std::path::PathBuf; use std::sync::{Arc, Mutex}; @@ -10,10 +11,10 @@ use fabro_core::state::ExecutionState; use fabro_graphviz::graph::types::{Edge as GvEdge, Graph as GvGraph, Node as GvNode}; use crate::artifact; -use crate::context::keys; +use crate::context::{Context, ParallelBranchPreamble, keys}; use crate::graph::{WorkflowGraph, WorkflowNode}; use crate::handler::llm::preamble::build_preamble; -use crate::outcome::BilledModelUsage; +use crate::outcome::{BilledModelUsage, Outcome}; use crate::runtime_store::RunStoreHandle; type WfRunState = ExecutionState>; @@ -61,6 +62,60 @@ impl FidelityLifecycle { "fidelity mutex should not be poisoned: no code panics while holding this lock", ) = flag; } + + /// Render the per-branch preamble stash for a parallel node, indexed by + /// outgoing-edge order (the same order `ParallelHandler` fans out in). + /// `Null` entries inherit the fork's preamble. + fn build_parallel_branch_preambles( + &self, + node_id: &str, + fork_fidelity: keys::Fidelity, + resolved_context: &Context, + resolved_outcomes: &HashMap, + completed_nodes: &[String], + ) -> Vec { + let mut rendered: HashMap = HashMap::new(); + self.graph + .outgoing_edges(node_id) + .into_iter() + .enumerate() + .map(|(branch_index, edge)| { + let Some(target_node) = self.graph.nodes.get(&edge.to) else { + return serde_json::Value::Null; + }; + let resolution = resolve_parallel_branch_fidelity(edge, target_node, fork_fidelity); + if resolution.requested == Some(keys::Fidelity::Full) { + tracing::warn!( + parallel_node = %node_id, + branch = %edge.to, + branch_index, + effective_fidelity = %keys::Fidelity::Full.degraded(), + "Parallel branch fidelity degraded from full" + ); + } + let Some(branch_fidelity) = resolution.effective else { + return serde_json::Value::Null; + }; + rendered + .entry(branch_fidelity) + .or_insert_with(|| { + let entry = ParallelBranchPreamble { + fidelity: branch_fidelity, + preamble: build_preamble( + branch_fidelity, + resolved_context, + &self.graph, + completed_nodes, + resolved_outcomes, + ), + }; + serde_json::to_value(entry) + .expect("ParallelBranchPreamble serialization cannot fail") + }) + .clone() + }) + .collect() + } } #[async_trait] @@ -143,47 +198,23 @@ impl RunLifecycle for FidelityLifecycle { .context .set(keys::CURRENT_PREAMBLE, serde_json::json!(preamble)); + // 5. Parallel nodes: pre-render per-branch preambles into the stash that + // ParallelHandler consumes at fan-out. if gv_node.handler_type() == Some("parallel") { - let mut branch_preambles = Vec::new(); - for (branch_index, edge) in self.graph.outgoing_edges(node.id()).iter().enumerate() { - let Some(target_node) = self.graph.nodes.get(&edge.to) else { - branch_preambles.push(serde_json::Value::Null); - continue; - }; - let resolution = resolve_parallel_branch_fidelity(edge, target_node, fidelity); - if resolution.requested() == Some(keys::Fidelity::Full) { - tracing::warn!( - parallel_node = %node.id(), - branch = %edge.to, - branch_index, - fidelity = %keys::Fidelity::Full, - effective_fidelity = %keys::Fidelity::SummaryHigh, - "Parallel branch fidelity degraded" - ); - } - let Some(branch_fidelity) = resolution.entry_fidelity() else { - branch_preambles.push(serde_json::Value::Null); - continue; - }; - let branch_preamble = build_preamble( - branch_fidelity, - &resolved_context, - &self.graph, - &state.completed_nodes, - &resolved_outcomes, - ); - branch_preambles.push(serde_json::json!({ - "fidelity": branch_fidelity.to_string(), - "preamble": branch_preamble, - })); - } + let branch_preambles = self.build_parallel_branch_preambles( + node.id(), + fidelity, + &resolved_context, + &resolved_outcomes, + &state.completed_nodes, + ); state.context.set( keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES, serde_json::Value::Array(branch_preambles), ); } - // 5. Thread ID resolution via resolve_thread_id: edge → node → graph default → + // 6. Thread ID resolution via resolve_thread_id: edge → node → graph default → // class → previous let thread_id = resolve_thread_id( incoming_edge_ref, @@ -192,13 +223,13 @@ impl RunLifecycle for FidelityLifecycle { state.previous_node_id.as_deref(), ); - // 6. Set thread.{tid}.current_node + // 7. Set thread.{tid}.current_node if let Some(ref tid) = thread_id { let key = keys::thread_current_node_key(tid); state.context.set(key, serde_json::json!(node.id())); } - // 7. Set INTERNAL_THREAD_ID (or null) + // 8. Set INTERNAL_THREAD_ID (or null) match thread_id { Some(tid) => { state @@ -212,7 +243,7 @@ impl RunLifecycle for FidelityLifecycle { } } - // 8. Set INTERNAL_NODE_VISIT_COUNT and CURRENT_NODE + // 9. Set INTERNAL_NODE_VISIT_COUNT and CURRENT_NODE let visits = state.node_visits.get(node.id()).copied().unwrap_or(1); state .context @@ -243,22 +274,14 @@ impl RunLifecycle for FidelityLifecycle { } } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy)] struct ParallelBranchFidelityResolution { + /// The explicit fidelity requested on the edge or node, pre-degradation. requested: Option, + /// The fidelity to render an entry for; `None` inherits the fork preamble. effective: Option, } -impl ParallelBranchFidelityResolution { - fn requested(self) -> Option { - self.requested - } - - fn entry_fidelity(self) -> Option { - self.effective - } -} - /// Resolve explicit branch fidelity with edge-over-node precedence. /// /// Branches with no explicit fidelity inherit the parallel node's preamble. @@ -270,10 +293,7 @@ fn resolve_parallel_branch_fidelity( target_node: &GvNode, parallel_fidelity: keys::Fidelity, ) -> ParallelBranchFidelityResolution { - let requested = edge - .fidelity() - .and_then(|value| value.parse().ok()) - .or_else(|| target_node.fidelity().and_then(|value| value.parse().ok())); + let requested = explicit_fidelity(Some(edge), target_node).map(|(fidelity, _)| fidelity); let effective = requested .map(keys::Fidelity::degraded) .filter(|fidelity| *fidelity != parallel_fidelity); @@ -284,6 +304,23 @@ fn resolve_parallel_branch_fidelity( } } +/// Explicit fidelity from the incoming edge attribute, else the node +/// attribute, with the winning source labeled for logging. +fn explicit_fidelity( + incoming_edge: Option<&GvEdge>, + node: &GvNode, +) -> Option<(keys::Fidelity, &'static str)> { + incoming_edge + .and_then(|e| e.fidelity()) + .and_then(|s| s.parse().ok()) + .map(|f| (f, "edge")) + .or_else(|| { + node.fidelity() + .and_then(|s| s.parse().ok()) + .map(|f| (f, "node")) + }) +} + /// Resolve the context fidelity for a node, following the precedence: /// 1. Incoming edge `fidelity` attribute /// 2. Target node `fidelity` attribute @@ -294,13 +331,8 @@ fn resolve_fidelity( node: &GvNode, graph: &GvGraph, ) -> keys::Fidelity { - let (resolved, source) = if let Some(f) = incoming_edge - .and_then(|e| e.fidelity()) - .and_then(|s| s.parse().ok()) - { - (f, "edge") - } else if let Some(f) = node.fidelity().and_then(|s| s.parse().ok()) { - (f, "node") + let (resolved, source) = if let Some((f, source)) = explicit_fidelity(incoming_edge, node) { + (f, source) } else if let Some(f) = graph.default_fidelity().and_then(|s| s.parse().ok()) { (f, "graph") } else { @@ -362,7 +394,7 @@ mod tests { use crate::context::WorkflowContext; use crate::context::keys::Fidelity; - fn fidelity_attr(value: &str) -> AttrValue { + fn str_attr(value: &str) -> AttrValue { AttrValue::String(value.to_string()) } @@ -374,23 +406,23 @@ mod tests { let mut start = Node::new("start"); start .attrs - .insert("shape".to_string(), fidelity_attr("Mdiamond")); + .insert("shape".to_string(), str_attr("Mdiamond")); let mut fork = Node::new("fork"); fork.attrs - .insert("shape".to_string(), fidelity_attr("component")); + .insert("shape".to_string(), str_attr("component")); if let Some(fidelity) = fork_fidelity { fork.attrs - .insert("fidelity".to_string(), fidelity_attr(fidelity)); + .insert("fidelity".to_string(), str_attr(fidelity)); } let mut branch_a = Node::new("branch_a"); if let Some(fidelity) = branch_a_fidelity { branch_a .attrs - .insert("fidelity".to_string(), fidelity_attr(fidelity)); + .insert("fidelity".to_string(), str_attr(fidelity)); } let branch_b = Node::new("branch_b"); let mut work = Node::new("work"); - work.attrs.insert("shape".to_string(), fidelity_attr("box")); + work.attrs.insert("shape".to_string(), str_attr("box")); graph.nodes.insert(start.id.clone(), start); graph.nodes.insert(fork.id.clone(), fork); @@ -426,15 +458,15 @@ mod tests { fn parallel_branch_fidelity_edge_overrides_node() { let mut node = Node::new("branch"); node.attrs - .insert("fidelity".to_string(), fidelity_attr("compact")); + .insert("fidelity".to_string(), str_attr("compact")); let mut edge = Edge::new("fork", "branch"); edge.attrs - .insert("fidelity".to_string(), fidelity_attr("truncate")); + .insert("fidelity".to_string(), str_attr("truncate")); let resolved = resolve_parallel_branch_fidelity(&edge, &node, Fidelity::SummaryHigh); - assert_eq!(resolved.requested(), Some(Fidelity::Truncate)); - assert_eq!(resolved.entry_fidelity(), Some(Fidelity::Truncate)); + assert_eq!(resolved.requested, Some(Fidelity::Truncate)); + assert_eq!(resolved.effective, Some(Fidelity::Truncate)); } #[test] @@ -444,47 +476,45 @@ mod tests { let resolution = resolve_parallel_branch_fidelity(&edge, &node, Fidelity::Compact); - assert_eq!(resolution.requested(), None); - assert_eq!(resolution.entry_fidelity(), None); + assert_eq!(resolution.requested, None); + assert_eq!(resolution.effective, None); } #[test] fn parallel_branch_full_fidelity_degrades_to_summary_high() { let mut node = Node::new("branch"); - node.attrs - .insert("fidelity".to_string(), fidelity_attr("full")); + node.attrs.insert("fidelity".to_string(), str_attr("full")); let edge = Edge::new("fork", "branch"); let resolved = resolve_parallel_branch_fidelity(&edge, &node, Fidelity::Compact); - assert_eq!(resolved.requested(), Some(Fidelity::Full)); - assert_eq!(resolved.entry_fidelity(), Some(Fidelity::SummaryHigh)); + assert_eq!(resolved.requested, Some(Fidelity::Full)); + assert_eq!(resolved.effective, Some(Fidelity::SummaryHigh)); } #[test] fn parallel_branch_fidelity_equal_to_fork_inherits() { let mut node = Node::new("branch"); node.attrs - .insert("fidelity".to_string(), fidelity_attr("summary:high")); + .insert("fidelity".to_string(), str_attr("summary:high")); let edge = Edge::new("fork", "branch"); let resolution = resolve_parallel_branch_fidelity(&edge, &node, Fidelity::SummaryHigh); - assert_eq!(resolution.requested(), Some(Fidelity::SummaryHigh)); - assert_eq!(resolution.entry_fidelity(), None); + assert_eq!(resolution.requested, Some(Fidelity::SummaryHigh)); + assert_eq!(resolution.effective, None); } #[test] fn explicit_full_branch_equal_to_degraded_fork_inherits() { let mut node = Node::new("branch"); - node.attrs - .insert("fidelity".to_string(), fidelity_attr("full")); + node.attrs.insert("fidelity".to_string(), str_attr("full")); let edge = Edge::new("fork", "branch"); let resolution = resolve_parallel_branch_fidelity(&edge, &node, Fidelity::SummaryHigh); - assert_eq!(resolution.requested(), Some(Fidelity::Full)); - assert_eq!(resolution.entry_fidelity(), None); + assert_eq!(resolution.requested, Some(Fidelity::Full)); + assert_eq!(resolution.effective, None); } #[test] @@ -494,8 +524,8 @@ mod tests { let resolution = resolve_parallel_branch_fidelity(&edge, &node, Fidelity::Full); - assert_eq!(resolution.requested(), None); - assert_eq!(resolution.entry_fidelity(), None); + assert_eq!(resolution.requested, None); + assert_eq!(resolution.effective, None); } #[tokio::test] From 7ff153d2229d37c004b513397d656cc3f617c59d Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Thu, 23 Jul 2026 15:21:56 -0400 Subject: [PATCH 06/15] chore: bump verify timeout to 20m --- .fabro/workflows/implement-plan/workflow.fabro | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.fabro/workflows/implement-plan/workflow.fabro b/.fabro/workflows/implement-plan/workflow.fabro index 8c7ad4256..e2a59dcae 100644 --- a/.fabro/workflows/implement-plan/workflow.fabro +++ b/.fabro/workflows/implement-plan/workflow.fabro @@ -12,7 +12,7 @@ digraph ImplementPlan { implement [label="Implement", prompt="Read the plan file referenced in the goal and implement every step. Make all the code changes described in the plan. Use red/green TDD.", model="openai/gpt-5.6-sol", provider="openrouter", reasoning_effort="xhigh"] simplify_fable [label="Simplify (Claude Fable 5)", prompt="@prompts/simplify.md", model="anthropic/claude-fable-5", provider="openrouter", reasoning_effort="xhigh"] simplify_sol [label="Simplify (GPT-5.6 Sol)", prompt="@prompts/simplify.md", model="openai/gpt-5.6-sol", provider="openrouter", reasoning_effort="max"] - verify [label="Verify", shape=parallelogram, script="git fetch origin main 2>&1 && git merge --no-edit --no-stat origin/main 2>&1 && cargo +nightly-2026-04-14 fmt --all 2>&1 && cargo dev docs refresh 2>&1 && cargo +nightly-2026-04-14 fmt --check --all 2>&1 && { command -v rg >/dev/null 2>&1 || { echo 'rg is required for verify'; exit 127; }; } && ! rg -n 'AuthMode::Disabled|RunAuthMethod|RunSubjectProvenance|\bActorRef\b|\bActorKind\b|AuthenticatedSubject|AuthenticatedService|AuthorizeRunScoped|AuthorizeRunBlob|AuthorizeStageArtifact|AuthorizeCommandLog|auth_method\s*==\s*\"disabled\"' lib/crates apps lib/packages docs/public/api-reference/fabro-api.yaml 2>&1 && cargo +nightly-2026-04-14 clippy --workspace --all-targets -- -D warnings 2>&1 && cargo nextest run --workspace --status-level slow --profile ci 2>&1 && cargo dev docs check 2>&1 && bun install --frozen-lockfile 2>&1 && (cd apps/fabro-web && bun run typecheck) 2>&1 && (cd apps/fabro-web && bun run test) 2>&1 && (cd lib/packages/fabro-api-client && bun run typecheck) 2>&1 && cargo dev build -- -p fabro-cli --release 2>&1", goal_gate=true, retry_target="fixup"] + verify [label="Verify", shape=parallelogram, script="git fetch origin main 2>&1 && git merge --no-edit --no-stat origin/main 2>&1 && cargo +nightly-2026-04-14 fmt --all 2>&1 && cargo dev docs refresh 2>&1 && cargo +nightly-2026-04-14 fmt --check --all 2>&1 && { command -v rg >/dev/null 2>&1 || { echo 'rg is required for verify'; exit 127; }; } && ! rg -n 'AuthMode::Disabled|RunAuthMethod|RunSubjectProvenance|\bActorRef\b|\bActorKind\b|AuthenticatedSubject|AuthenticatedService|AuthorizeRunScoped|AuthorizeRunBlob|AuthorizeStageArtifact|AuthorizeCommandLog|auth_method\s*==\s*\"disabled\"' lib/crates apps lib/packages docs/public/api-reference/fabro-api.yaml 2>&1 && cargo +nightly-2026-04-14 clippy --workspace --all-targets -- -D warnings 2>&1 && cargo nextest run --workspace --status-level slow --profile ci 2>&1 && cargo dev docs check 2>&1 && bun install --frozen-lockfile 2>&1 && (cd apps/fabro-web && bun run typecheck) 2>&1 && (cd apps/fabro-web && bun run test) 2>&1 && (cd lib/packages/fabro-api-client && bun run typecheck) 2>&1 && cargo dev build -- -p fabro-cli --release 2>&1", timeout="20m", goal_gate=true, retry_target="fixup"] fixup [label="Fixup", prompt="The verify step failed. Read the build output from context and fix all format, clippy, Rust test, docs, TypeScript typecheck/test, and build failures.", model="anthropic/claude-fable-5", provider="openrouter", reasoning_effort="xhigh", max_visits=3] start -> toolchain From 02437fb18a3ff3dc2d92f1084929fc7c50d19429 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Thu, 23 Jul 2026 15:28:19 -0400 Subject: [PATCH 07/15] fix(workflow): constrain last-file routing fallback --- docs/public/agents/outputs.mdx | 2 + docs/public/reference/dot-language.mdx | 2 +- .../fabro-workflow/src/handler/agent.rs | 202 ++++++++++++++---- .../src/handler/structured_output.rs | 34 +++ 4 files changed, 195 insertions(+), 45 deletions(-) diff --git a/docs/public/agents/outputs.mdx b/docs/public/agents/outputs.mdx index e03b4e08f..d62073dea 100644 --- a/docs/public/agents/outputs.mdx +++ b/docs/public/agents/outputs.mdx @@ -95,6 +95,8 @@ Agent nodes can provide routing directives through fallback files. Fabro checks This fallback chain applies to normal routing extraction and to `output_schema="routing"`. For validated routing, Fabro only advances to the next source when the current source has no JSON object or no object with recognized routing fields. If the current source contains malformed routing JSON or valid JSON with wrong routing field types, validation fails and Fabro starts the repair loop instead. +The last-file fallback only reads `.json` and `.md` files (case-insensitive), and the routing JSON must be the final standalone object in the file, followed only by whitespace. Fabro ignores other file types and routing JSON followed by any other content. These restrictions do not apply to the dedicated `status.json` fallback. + Prompt nodes do not use file fallbacks; they validate or extract routing directives from the response text only. If no source provides routing directives, the transition falls through to condition matching, unconditional edges, or weight-based tiebreaking as described in [Transitions](/workflows/transitions). diff --git a/docs/public/reference/dot-language.mdx b/docs/public/reference/dot-language.mdx index f01fbf2ed..2184ab1d4 100644 --- a/docs/public/reference/dot-language.mdx +++ b/docs/public/reference/dot-language.mdx @@ -235,7 +235,7 @@ audit [ - On validation failure, Fabro sends validation feedback to the same active context before failing: prompt nodes keep the prior assistant response in the message list, and API-backed agent nodes repair in the same live session. - `output_retries` defaults to `2` and controls only these corrective structured-output turns. Negative values are treated as `0`. It is not the same as `max_retries` and does not consume workflow retry attempts. - Custom schema output is stored in context at `output.{node_id}`. Routing schema output updates routing fields and any `context_updates`. -- Agent routing fallbacks still apply to `output_schema="routing"`: response text first, then `status.json`, then the last file touched by the agent. Custom schemas and prompt nodes validate response text only. +- Agent routing fallbacks still apply to `output_schema="routing"`: response text first, then `status.json`, then the last file touched by the agent. The last-file fallback only accepts `.json` and `.md` files (case-insensitive) whose final standalone JSON object contains the routing directive; only whitespace may follow it. Custom schemas and prompt nodes validate response text only. - `backend="acp"` with `output_schema` is unsupported in this release. ### Command nodes diff --git a/lib/crates/fabro-workflow/src/handler/agent.rs b/lib/crates/fabro-workflow/src/handler/agent.rs index 16dbe0230..9ba8c2fe1 100644 --- a/lib/crates/fabro-workflow/src/handler/agent.rs +++ b/lib/crates/fabro-workflow/src/handler/agent.rs @@ -19,6 +19,8 @@ use crate::event::{Emitter, Event, StageScope}; use crate::interview_runtime::WorkflowAgentQuestionRuntime; use crate::outcome::{BilledModelUsage, Outcome, OutcomeExt}; +const LAST_FILE_ROUTING_EXTENSIONS: &[&str] = &["json", "md"]; + /// Result from a `CodergenBackend` invocation. #[allow( clippy::large_enum_variant, @@ -169,8 +171,8 @@ pub(crate) async fn validate_agent_output_sources( } if let Some(path) = last_file_touched { - if let Some(contents) = read_sandbox_file(sandbox, path).await { - return structured_output::validate_response_text(schema, &contents); + if let Some(routing_json) = read_last_file_routing_json(sandbox, path).await { + return structured_output::validate_response_text(schema, &routing_json); } } @@ -181,6 +183,19 @@ async fn read_sandbox_file(sandbox: &Arc, path: &str) -> Option, path: &str) -> Option { + let extension = Path::new(path).extension()?.to_str()?; + if !LAST_FILE_ROUTING_EXTENSIONS + .iter() + .any(|allowed| extension.eq_ignore_ascii_case(allowed)) + { + return None; + } + + let contents = read_sandbox_file(sandbox, path).await?; + structured_output::terminal_json_object(&contents).map(str::to_owned) +} + /// Truncate a string to at most `max_chars` characters (char-boundary safe). pub(crate) fn truncate(s: &str, max_chars: usize) -> &str { if s.len() <= max_chars { @@ -382,7 +397,7 @@ impl Handler for AgentHandler { } else { // 7b. Parse routing directives from response text, falling back to // status.json written by the agent into the sandbox CWD, then to - // the last file the agent wrote. + // a terminal JSON object in an eligible last-written file. let found_in_response = extract_status_fields(&response_text, &mut outcome); if !found_in_response { let mut found_in_status_json = false; @@ -393,9 +408,10 @@ impl Handler for AgentHandler { } if !found_in_status_json { if let Some(ref path) = last_file_touched { - if let Some(contents) = read_sandbox_file(&services.run.sandbox, path).await + if let Some(routing_json) = + read_last_file_routing_json(&services.run.sandbox, path).await { - extract_status_fields(&contents, &mut outcome); + extract_status_fields(&routing_json, &mut outcome); } } } @@ -506,6 +522,68 @@ mod tests { context } + struct LastFileBackend { + path: String, + } + + #[async_trait] + impl CodergenBackend for LastFileBackend { + async fn run(&self, _request: CodergenRunRequest<'_>) -> Result { + Ok(CodergenResult::Text { + text: "Done writing results.".to_string(), + usage: None, + files_touched: vec![self.path.clone()], + last_file_touched: Some(self.path.clone()), + timing: StageTiming::default(), + }) + } + } + + async fn execute_with_last_file(path: &str, contents: &str) -> Outcome { + let sandbox_dir = TempDir::new().unwrap(); + std::fs::write(sandbox_dir.path().join(path), contents).unwrap(); + + let handler = AgentHandler::new(Some(Box::new(LastFileBackend { + path: path.to_string(), + }))); + let node = Node::new("step"); + let context = test_context(); + let graph = Graph::new("test"); + let tmp = TempDir::new().unwrap(); + + let mut services = EngineServices::test_default(); + services.run = + services + .run + .with_sandbox(std::sync::Arc::new(fabro_agent::LocalSandbox::new( + sandbox_dir.path().to_path_buf(), + ))); + + handler + .execute(&node, &context, &graph, tmp.path(), &services) + .await + .unwrap() + } + + async fn validate_routing_with_last_file( + path: &str, + contents: &str, + ) -> Result { + let sandbox_dir = TempDir::new().unwrap(); + std::fs::write(sandbox_dir.path().join(path), contents).unwrap(); + let sandbox: Arc = Arc::new(fabro_agent::LocalSandbox::new( + sandbox_dir.path().to_path_buf(), + )); + + validate_agent_output_sources( + &OutputSchemaKind::Routing, + "Done writing results.", + &sandbox, + Some(path), + ) + .await + } + #[tokio::test] async fn codergen_handler_simulate() { let handler = AgentHandler::new(None); @@ -731,49 +809,13 @@ mod tests { #[tokio::test] async fn codergen_handler_extracts_status_from_last_file_touched() { - struct LastFileBackend; - - #[async_trait] - impl CodergenBackend for LastFileBackend { - async fn run(&self, _request: CodergenRunRequest<'_>) -> Result { - Ok(CodergenResult::Text { - text: "Done writing results.".to_string(), - usage: None, - files_touched: vec!["results.md".to_string()], - last_file_touched: Some("results.md".to_string()), - timing: StageTiming::default(), - }) - } - } - - let sandbox_dir = TempDir::new().unwrap(); - // Write status fields into the file the agent "touched" — no status.json - std::fs::write( - sandbox_dir.path().join("results.md"), + let outcome = execute_with_last_file( + "results.md", r#"# Results {"context_updates": {"verified": "true"}} "#, ) - .unwrap(); - - let handler = AgentHandler::new(Some(Box::new(LastFileBackend))); - let node = Node::new("step"); - let context = test_context(); - let graph = Graph::new("test"); - let tmp = TempDir::new().unwrap(); - - let mut services = EngineServices::test_default(); - services.run = - services - .run - .with_sandbox(std::sync::Arc::new(fabro_agent::LocalSandbox::new( - sandbox_dir.path().to_path_buf(), - ))); - - let outcome = handler - .execute(&node, &context, &graph, tmp.path(), &services) - .await - .unwrap(); + .await; assert_eq!(outcome.status, crate::outcome::StageOutcome::Succeeded); assert_eq!( @@ -782,6 +824,33 @@ mod tests { ); } + #[tokio::test] + async fn codergen_handler_ignores_nonterminal_status_in_last_markdown_file() { + let outcome = execute_with_last_file( + "results.md", + r#"{"outcome":"failed","failure_reason":"tests failed"} + +All checks passed. +"#, + ) + .await; + + assert_eq!(outcome.status, crate::outcome::StageOutcome::Succeeded); + assert!(outcome.failure.is_none()); + } + + #[tokio::test] + async fn codergen_handler_ignores_terminal_status_in_disallowed_last_file() { + let outcome = execute_with_last_file( + "command.rs", + r#"{"outcome":"failed","failure_reason":"tests failed"}"#, + ) + .await; + + assert_eq!(outcome.status, crate::outcome::StageOutcome::Succeeded); + assert!(outcome.failure.is_none()); + } + #[tokio::test] async fn codergen_handler_output_schema_routing_uses_status_json_fallback_when_response_has_no_json() { @@ -819,6 +888,51 @@ mod tests { assert_eq!(outcome.preferred_label.as_deref(), Some("review")); } + #[tokio::test] + async fn validated_routing_accepts_terminal_status_in_json_file_case_insensitively() { + let validated = validate_routing_with_last_file( + "results.JSON", + "# Results\n\n{\"preferred_next_label\":\"review\"}\n", + ) + .await + .unwrap(); + + assert_eq!( + validated.value, + serde_json::json!({"preferred_next_label": "review"}), + ); + } + + #[tokio::test] + async fn validated_routing_ignores_nonterminal_status_in_last_markdown_file() { + let error = validate_routing_with_last_file( + "results.md", + "{\"outcome\":\"failed\",\"failure_reason\":\"tests failed\"}\nAll checks passed.", + ) + .await + .unwrap_err(); + + assert_eq!( + error.kind(), + structured_output::StructuredOutputErrorKind::NoJsonObject, + ); + } + + #[tokio::test] + async fn validated_routing_ignores_terminal_status_in_disallowed_last_file() { + let error = validate_routing_with_last_file( + "command.rs", + r#"{"outcome":"failed","failure_reason":"tests failed"}"#, + ) + .await + .unwrap_err(); + + assert_eq!( + error.kind(), + structured_output::StructuredOutputErrorKind::NoJsonObject, + ); + } + #[tokio::test] async fn codergen_handler_output_schema_routing_rejects_malformed_response_before_status_json_fallback() { diff --git a/lib/crates/fabro-workflow/src/handler/structured_output.rs b/lib/crates/fabro-workflow/src/handler/structured_output.rs index 546124b43..ea2c2aad1 100644 --- a/lib/crates/fabro-workflow/src/handler/structured_output.rs +++ b/lib/crates/fabro-workflow/src/handler/structured_output.rs @@ -264,6 +264,15 @@ fn find_json_objects(text: &str) -> Vec<&str> { results } +/// Return the outermost balanced JSON object that ends the text, ignoring +/// trailing whitespace. +pub(crate) fn terminal_json_object(text: &str) -> Option<&str> { + let trimmed = text.trim_end(); + find_json_objects(trimmed) + .into_iter() + .find(|candidate| trimmed.ends_with(candidate)) +} + pub(crate) fn extract_status_fields(text: &str, outcome: &mut Outcome) -> bool { let candidates = find_json_objects(text); @@ -493,6 +502,31 @@ mod tests { assert!(error.messages()[0].contains("recognized routing field")); } + #[test] + fn terminal_json_object_accepts_final_object_after_prose() { + let object = + terminal_json_object("# Results\n\n{\"context_updates\":{\"verified\":true}}\n\n"); + + assert_eq!(object, Some(r#"{"context_updates":{"verified":true}}"#),); + } + + #[test] + fn terminal_json_object_returns_outermost_nested_object() { + let object = terminal_json_object(r#"Results: {"context_updates":{"verified":true}}"#); + + assert_eq!(object, Some(r#"{"context_updates":{"verified":true}}"#),); + } + + #[test] + fn terminal_json_object_rejects_object_followed_by_content() { + assert_eq!( + terminal_json_object( + "{\"outcome\":\"failed\",\"failure_reason\":\"tests failed\"}\nMore details", + ), + None, + ); + } + #[test] fn routing_json_with_wrong_field_type_is_invalid() { let error = From 886371e4748ad00d0456e16f36365756a28620d9 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Thu, 23 Jul 2026 15:47:17 -0400 Subject: [PATCH 08/15] refactor(workflow): clean up last-file routing fallback review nits Document the contract of read_last_file_routing_json (terminal JSON extraction only; routing validation happens downstream), extract a shared sandbox_with_file test helper, and drop the misleading "standalone" wording from the fallback docs. Co-Authored-By: Claude Fable 5 --- docs/public/agents/outputs.mdx | 2 +- docs/public/reference/dot-language.mdx | 2 +- .../fabro-workflow/src/handler/agent.rs | 26 ++++++++++--------- 3 files changed, 16 insertions(+), 14 deletions(-) diff --git a/docs/public/agents/outputs.mdx b/docs/public/agents/outputs.mdx index d62073dea..28e3659a6 100644 --- a/docs/public/agents/outputs.mdx +++ b/docs/public/agents/outputs.mdx @@ -95,7 +95,7 @@ Agent nodes can provide routing directives through fallback files. Fabro checks This fallback chain applies to normal routing extraction and to `output_schema="routing"`. For validated routing, Fabro only advances to the next source when the current source has no JSON object or no object with recognized routing fields. If the current source contains malformed routing JSON or valid JSON with wrong routing field types, validation fails and Fabro starts the repair loop instead. -The last-file fallback only reads `.json` and `.md` files (case-insensitive), and the routing JSON must be the final standalone object in the file, followed only by whitespace. Fabro ignores other file types and routing JSON followed by any other content. These restrictions do not apply to the dedicated `status.json` fallback. +The last-file fallback only reads `.json` and `.md` files (case-insensitive), and the routing JSON must be the final JSON object in the file, with only whitespace after it. Fabro ignores other file types and routing JSON followed by any other content. These restrictions do not apply to the dedicated `status.json` fallback. Prompt nodes do not use file fallbacks; they validate or extract routing directives from the response text only. diff --git a/docs/public/reference/dot-language.mdx b/docs/public/reference/dot-language.mdx index 2184ab1d4..1543e5938 100644 --- a/docs/public/reference/dot-language.mdx +++ b/docs/public/reference/dot-language.mdx @@ -235,7 +235,7 @@ audit [ - On validation failure, Fabro sends validation feedback to the same active context before failing: prompt nodes keep the prior assistant response in the message list, and API-backed agent nodes repair in the same live session. - `output_retries` defaults to `2` and controls only these corrective structured-output turns. Negative values are treated as `0`. It is not the same as `max_retries` and does not consume workflow retry attempts. - Custom schema output is stored in context at `output.{node_id}`. Routing schema output updates routing fields and any `context_updates`. -- Agent routing fallbacks still apply to `output_schema="routing"`: response text first, then `status.json`, then the last file touched by the agent. The last-file fallback only accepts `.json` and `.md` files (case-insensitive) whose final standalone JSON object contains the routing directive; only whitespace may follow it. Custom schemas and prompt nodes validate response text only. +- Agent routing fallbacks still apply to `output_schema="routing"`: response text first, then `status.json`, then the last file touched by the agent. The last-file fallback only accepts `.json` and `.md` files (case-insensitive) whose final JSON object contains the routing directive; only whitespace may follow it. Custom schemas and prompt nodes validate response text only. - `backend="acp"` with `output_schema` is unsupported in this release. ### Command nodes diff --git a/lib/crates/fabro-workflow/src/handler/agent.rs b/lib/crates/fabro-workflow/src/handler/agent.rs index 9ba8c2fe1..850c9de6d 100644 --- a/lib/crates/fabro-workflow/src/handler/agent.rs +++ b/lib/crates/fabro-workflow/src/handler/agent.rs @@ -183,6 +183,9 @@ async fn read_sandbox_file(sandbox: &Arc, path: &str) -> Option, path: &str) -> Option { let extension = Path::new(path).extension()?.to_str()?; if !LAST_FILE_ROUTING_EXTENSIONS @@ -539,9 +542,17 @@ mod tests { } } - async fn execute_with_last_file(path: &str, contents: &str) -> Outcome { + fn sandbox_with_file(path: &str, contents: &str) -> (TempDir, Arc) { let sandbox_dir = TempDir::new().unwrap(); std::fs::write(sandbox_dir.path().join(path), contents).unwrap(); + let sandbox: Arc = Arc::new(fabro_agent::LocalSandbox::new( + sandbox_dir.path().to_path_buf(), + )); + (sandbox_dir, sandbox) + } + + async fn execute_with_last_file(path: &str, contents: &str) -> Outcome { + let (_sandbox_dir, sandbox) = sandbox_with_file(path, contents); let handler = AgentHandler::new(Some(Box::new(LastFileBackend { path: path.to_string(), @@ -552,12 +563,7 @@ mod tests { let tmp = TempDir::new().unwrap(); let mut services = EngineServices::test_default(); - services.run = - services - .run - .with_sandbox(std::sync::Arc::new(fabro_agent::LocalSandbox::new( - sandbox_dir.path().to_path_buf(), - ))); + services.run = services.run.with_sandbox(sandbox); handler .execute(&node, &context, &graph, tmp.path(), &services) @@ -569,11 +575,7 @@ mod tests { path: &str, contents: &str, ) -> Result { - let sandbox_dir = TempDir::new().unwrap(); - std::fs::write(sandbox_dir.path().join(path), contents).unwrap(); - let sandbox: Arc = Arc::new(fabro_agent::LocalSandbox::new( - sandbox_dir.path().to_path_buf(), - )); + let (_sandbox_dir, sandbox) = sandbox_with_file(path, contents); validate_agent_output_sources( &OutputSchemaKind::Routing, From c9b53031282fa360d27e5de2ac147bb955a64fca Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Thu, 23 Jul 2026 15:52:25 -0400 Subject: [PATCH 09/15] fix(workflow): flush agent events before stage completion --- .../fabro-workflow/src/handler/llm/api.rs | 58 +++++++++++++++++-- .../fabro-workflow/tests/it/integration.rs | 30 +++++++++- 2 files changed, 81 insertions(+), 7 deletions(-) diff --git a/lib/crates/fabro-workflow/src/handler/llm/api.rs b/lib/crates/fabro-workflow/src/handler/llm/api.rs index 2b0ffc1df..93ee9fed8 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/api.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/api.rs @@ -24,7 +24,7 @@ use fabro_model::{AgentProfileKind, Catalog, FallbackTarget, ModelRef, ProviderI use fabro_types::settings::run::RunModelControls; use fabro_types::{PermissionLevel, RunId, SessionCapability, StageId, StageTiming}; use serde::de::DeserializeOwned; -use tokio::sync::Mutex as TokioMutex; +use tokio::sync::{Mutex as TokioMutex, mpsc}; use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; @@ -532,16 +532,50 @@ fn emit_agent_tools_available( /// Spawn a task that subscribes to session events and: /// 1. Tracks file changes (write_file/edit_file tool calls) into shared state. /// 2. Forwards non-streaming agent events to the pipeline emitter. +/// +/// The returned handle exposes a per-input barrier. A successful +/// `process_input_with_runtime` emits `ProcessingEnd` after all events for +/// that input, so waiting for the barrier keeps terminal stage events from +/// overtaking queued agent events. +struct EventForwarder { + processing_end_rx: mpsc::UnboundedReceiver<()>, + task: JoinHandle<()>, +} + +impl EventForwarder { + async fn wait_for_processing_end(&mut self) { + if self.processing_end_rx.recv().await.is_none() { + tracing::warn!("Agent event forwarder stopped before processing input events"); + } + } + + fn abort(&self) { + self.task.abort(); + } +} + +impl Drop for EventForwarder { + fn drop(&mut self) { + self.task.abort(); + } +} + fn spawn_event_forwarder( session: &Session, node_id: String, scope: StageScope, emitter: Arc, file_tracking: Arc>, -) { +) -> EventForwarder { let mut rx = session.subscribe(); - tokio::spawn(async move { + let root_session_id = session.id().to_string(); + let (processing_end_tx, processing_end_rx) = mpsc::unbounded_channel(); + let task = tokio::spawn(async move { while let Ok(event) = rx.recv().await { + let is_root_processing_end = event.session_id == root_session_id + && event.parent_session_id.is_none() + && matches!(&event.event, AgentEvent::ProcessingEnd); + // Reset watchdog on every event, including streaming deltas emitter.touch(); @@ -573,8 +607,17 @@ fn spawn_event_forwarder( &scope, ); } + + if is_root_processing_end { + let _ = processing_end_tx.send(()); + } } }); + + EventForwarder { + processing_end_rx, + task, + } } /// LLM backend that delegates to an `agent` Session per invocation. @@ -1213,7 +1256,7 @@ impl CodergenBackend for AgentApiBackend { let stage_scope = StageScope::for_handler(context, &node.id); // Subscribe to session events: forward to pipeline emitter + track files. - spawn_event_forwarder( + let mut event_forwarder = spawn_event_forwarder( &session, node.id.clone(), stage_scope.clone(), @@ -1284,6 +1327,7 @@ impl CodergenBackend for AgentApiBackend { inference_duration = inference_duration.saturating_add(timing.inference); tool_duration = tool_duration.saturating_add(timing.tool); if process_result.is_ok() { + event_forwarder.wait_for_processing_end().await; total_usage += session.last_input_usage(); UsdMicros::accumulate(&mut total_cost, session.last_input_cost()); } @@ -1315,6 +1359,7 @@ impl CodergenBackend for AgentApiBackend { let mut succeeded = false; bridge.abort(); + event_forwarder.abort(); discard_session(&mut session, &mut lease, emitter); for (index, target) in self.fallback_chain.iter().enumerate() { @@ -1368,7 +1413,7 @@ impl CodergenBackend for AgentApiBackend { bridge.replace(cancel_token.clone(), &session); // Re-subscribe to forward events + track files from the new session - spawn_event_forwarder( + event_forwarder = spawn_event_forwarder( &session, node.id.clone(), stage_scope.clone(), @@ -1420,6 +1465,7 @@ impl CodergenBackend for AgentApiBackend { tool_duration = tool_duration.saturating_add(timing.tool); match process_result { Ok(()) => { + event_forwarder.wait_for_processing_end().await; total_usage += session.last_input_usage(); UsdMicros::accumulate(&mut total_cost, session.last_input_cost()); succeeded = true; @@ -1492,6 +1538,7 @@ impl CodergenBackend for AgentApiBackend { tool_duration = tool_duration.saturating_add(timing.tool); match repair_result { Ok(()) => { + event_forwarder.wait_for_processing_end().await; total_usage += session.last_input_usage(); UsdMicros::accumulate(&mut total_cost, session.last_input_cost()); repair_attempts += 1; @@ -1534,6 +1581,7 @@ impl CodergenBackend for AgentApiBackend { // Collect files_touched from the shared tracking state. let (files_touched, last_file_touched) = file_tracking_snapshot(&file_tracking); + drop(event_forwarder); if let Some(lease) = lease.take() { lease.release(); diff --git a/lib/crates/fabro-workflow/tests/it/integration.rs b/lib/crates/fabro-workflow/tests/it/integration.rs index 2b817434a..71a1e1103 100644 --- a/lib/crates/fabro-workflow/tests/it/integration.rs +++ b/lib/crates/fabro-workflow/tests/it/integration.rs @@ -2319,7 +2319,7 @@ reasoning = false ); } -#[tokio::test] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn workflow_persists_authoritative_openrouter_cost_for_agent_stage() { use fabro_auth::EnvCredentialSource; use fabro_workflow::steering_hub::SteeringHub; @@ -2398,8 +2398,18 @@ base_url = "{}" registry.register("start", Box::new(StartHandler)); registry.register("exit", Box::new(ExitHandler)); + let events = Arc::new(std::sync::Mutex::new(Vec::new())); + let events_for_listener = Arc::clone(&events); + let emitter = Arc::new(Emitter::default()); + emitter.on_event(move |event| { + if event.event_name() == "agent.message" { + std::thread::sleep(Duration::from_millis(50)); + } + events_for_listener.lock().unwrap().push(event.clone()); + }); + let dir = tempfile::tempdir().unwrap(); - let engine = WorkflowRunner::new(registry, Arc::new(Emitter::default()), local_env()); + let engine = WorkflowRunner::new(registry, emitter, local_env()); let run_options = RunOptions { settings: WorkflowSettings::default(), run_dir: dir.path().to_path_buf(), @@ -2432,6 +2442,22 @@ base_url = "{}" Some(AUTHORITATIVE_COST_USD_MICROS), "provider-reported usage.cost should override the catalog estimate" ); + + let events = events.lock().unwrap(); + let agent_message = events + .iter() + .position(|event| event.event_name() == "agent.message") + .expect("agent message should be emitted"); + let stage_completed = events + .iter() + .position(|event| { + event.event_name() == "stage.completed" && event.node_id.as_deref() == Some("work") + }) + .expect("work stage completion should be emitted"); + assert!( + agent_message < stage_completed, + "agent messages must be forwarded before terminal stage events" + ); } // --------------------------------------------------------------------------- From 8c37f5fae55624edb1f922f509a478f454edc7d5 Mon Sep 17 00:00:00 2001 From: "fabro-releases[bot]" Date: Thu, 23 Jul 2026 20:10:14 +0000 Subject: [PATCH 10/15] Bump version to 0.303.0-nightly.4 --- Cargo.lock | 102 ++++++++++++++++++++++++++--------------------------- Cargo.toml | 2 +- 2 files changed, 52 insertions(+), 52 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index d02be647e..014915d5c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2239,7 +2239,7 @@ dependencies = [ [[package]] name = "fabro-acp" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "agent-client-protocol", "agent-client-protocol-tokio", @@ -2258,7 +2258,7 @@ dependencies = [ [[package]] name = "fabro-agent" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "async-trait", @@ -2300,7 +2300,7 @@ dependencies = [ [[package]] name = "fabro-api" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "chrono", "fabro-automation", @@ -2323,7 +2323,7 @@ dependencies = [ [[package]] name = "fabro-auth" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "async-trait", @@ -2348,7 +2348,7 @@ dependencies = [ [[package]] name = "fabro-automation" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "chrono", @@ -2367,11 +2367,11 @@ dependencies = [ [[package]] name = "fabro-build-support" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" [[package]] name = "fabro-checkpoint" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "chrono", "fabro-config", @@ -2387,7 +2387,7 @@ dependencies = [ [[package]] name = "fabro-cli" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "assert_cmd", @@ -2489,7 +2489,7 @@ dependencies = [ [[package]] name = "fabro-client" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "bytes", @@ -2518,7 +2518,7 @@ dependencies = [ [[package]] name = "fabro-config" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "chrono", @@ -2547,7 +2547,7 @@ dependencies = [ [[package]] name = "fabro-core" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "async-trait", "fabro-types", @@ -2562,7 +2562,7 @@ dependencies = [ [[package]] name = "fabro-db" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "chrono", @@ -2574,7 +2574,7 @@ dependencies = [ [[package]] name = "fabro-dev" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "assert_cmd", @@ -2593,7 +2593,7 @@ dependencies = [ [[package]] name = "fabro-dump" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "bytes", @@ -2607,7 +2607,7 @@ dependencies = [ [[package]] name = "fabro-environment" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "chrono", @@ -2629,7 +2629,7 @@ dependencies = [ [[package]] name = "fabro-github" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "base64", @@ -2651,7 +2651,7 @@ dependencies = [ [[package]] name = "fabro-graphviz" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "fabro-types", @@ -2665,7 +2665,7 @@ dependencies = [ [[package]] name = "fabro-hooks" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "async-trait", "fabro-agent", @@ -2688,7 +2688,7 @@ dependencies = [ [[package]] name = "fabro-http" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "fabro-static", "http 1.4.0", @@ -2698,7 +2698,7 @@ dependencies = [ [[package]] name = "fabro-install" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "base64", @@ -2717,7 +2717,7 @@ dependencies = [ [[package]] name = "fabro-interview" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "async-trait", "dialoguer", @@ -2732,7 +2732,7 @@ dependencies = [ [[package]] name = "fabro-llm" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "async-trait", @@ -2773,7 +2773,7 @@ dependencies = [ [[package]] name = "fabro-macros" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "clap", "fabro-options-metadata", @@ -2784,7 +2784,7 @@ dependencies = [ [[package]] name = "fabro-manifest" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "fabro-api", @@ -2802,7 +2802,7 @@ dependencies = [ [[package]] name = "fabro-mcp" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "axum", @@ -2822,7 +2822,7 @@ dependencies = [ [[package]] name = "fabro-mcp-server" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "chrono", @@ -2849,7 +2849,7 @@ dependencies = [ [[package]] name = "fabro-mcp-store" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "chrono", "fabro-db", @@ -2867,7 +2867,7 @@ dependencies = [ [[package]] name = "fabro-model" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "fabro-static", "http 1.4.0", @@ -2883,7 +2883,7 @@ dependencies = [ [[package]] name = "fabro-oauth" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "axum", @@ -2905,7 +2905,7 @@ dependencies = [ [[package]] name = "fabro-options-metadata" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "serde", "serde_json", @@ -2913,7 +2913,7 @@ dependencies = [ [[package]] name = "fabro-proc" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "cc", "libc", @@ -2922,7 +2922,7 @@ dependencies = [ [[package]] name = "fabro-redact" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "aho-corasick", "ref-cast", @@ -2938,7 +2938,7 @@ dependencies = [ [[package]] name = "fabro-sandbox" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "async-trait", @@ -2983,7 +2983,7 @@ dependencies = [ [[package]] name = "fabro-server" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "async-trait", @@ -3075,7 +3075,7 @@ dependencies = [ [[package]] name = "fabro-slack" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "fabro-http", "fabro-interview", @@ -3097,18 +3097,18 @@ dependencies = [ [[package]] name = "fabro-spa" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "rust-embed", ] [[package]] name = "fabro-static" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" [[package]] name = "fabro-store" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "async-trait", "bytes", @@ -3138,7 +3138,7 @@ dependencies = [ [[package]] name = "fabro-telemetry" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "base64", @@ -3164,7 +3164,7 @@ dependencies = [ [[package]] name = "fabro-template" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "fabro-types", @@ -3178,7 +3178,7 @@ dependencies = [ [[package]] name = "fabro-test" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "assert_cmd", @@ -3203,7 +3203,7 @@ dependencies = [ [[package]] name = "fabro-tool" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "async-trait", @@ -3224,7 +3224,7 @@ dependencies = [ [[package]] name = "fabro-tracker" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "async-trait", @@ -3238,7 +3238,7 @@ dependencies = [ [[package]] name = "fabro-types" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "chrono", "clap", @@ -3260,7 +3260,7 @@ dependencies = [ [[package]] name = "fabro-util" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "console 0.15.11", @@ -3281,7 +3281,7 @@ dependencies = [ [[package]] name = "fabro-validate" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "fabro-acp", "fabro-graphviz", @@ -3294,7 +3294,7 @@ dependencies = [ [[package]] name = "fabro-variable" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "chrono", @@ -3311,7 +3311,7 @@ dependencies = [ [[package]] name = "fabro-vault" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "chrono", @@ -3330,7 +3330,7 @@ dependencies = [ [[package]] name = "fabro-workflow" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "assert_cmd", @@ -8494,7 +8494,7 @@ dependencies = [ [[package]] name = "twin-github" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "axum", "base64", @@ -8513,7 +8513,7 @@ dependencies = [ [[package]] name = "twin-openai" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" dependencies = [ "anyhow", "async-stream", diff --git a/Cargo.toml b/Cargo.toml index 32978ca47..6cdec975b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,7 +5,7 @@ resolver = "2" [workspace.package] edition = "2021" -version = "0.303.0-nightly.3" +version = "0.303.0-nightly.4" license = "MIT" [workspace.dependencies] From 9b85653a2dc62007a5ec2ce8cc8b8f1f4c817ef0 Mon Sep 17 00:00:00 2001 From: Fabro Date: Thu, 23 Jul 2026 20:29:19 +0000 Subject: [PATCH 11/15] fabro(01KY7Y01REECZ24XXTMBZ3PPV9): simplify_sol (succeeded) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fabro-Run: 01KY7Y01REECZ24XXTMBZ3PPV9 Fabro-Completed: 7 Fabro-Checkpoint: 334aea8fdc267cb8b5d7ba3e67bfb0d6d0341d61 ⚒️ Generated with [Fabro](https://fabro.sh) --- lib/crates/fabro-workflow/src/artifact.rs | 4 +- lib/crates/fabro-workflow/src/context.rs | 17 ++-- .../fabro-workflow/src/lifecycle/fidelity.rs | 87 ++++++++++--------- 3 files changed, 54 insertions(+), 54 deletions(-) diff --git a/lib/crates/fabro-workflow/src/artifact.rs b/lib/crates/fabro-workflow/src/artifact.rs index bbf7f7ed8..8ea8e38f0 100644 --- a/lib/crates/fabro-workflow/src/artifact.rs +++ b/lib/crates/fabro-workflow/src/artifact.rs @@ -83,8 +83,8 @@ pub fn durable_context_snapshot(context: &Context) -> HashMap { snapshot } -/// Remove runtime-only keys that must never reach durable storage or events. -pub fn strip_transient_keys(values: &mut HashMap) { +/// Remove runtime-only keys that must never reach durable storage. +pub(crate) fn strip_transient_keys(values: &mut HashMap) { for key in context::keys::TRANSIENT_CONTEXT_KEYS { values.remove(*key); } diff --git a/lib/crates/fabro-workflow/src/context.rs b/lib/crates/fabro-workflow/src/context.rs index a890c0e8f..b02a8dc51 100644 --- a/lib/crates/fabro-workflow/src/context.rs +++ b/lib/crates/fabro-workflow/src/context.rs @@ -48,10 +48,8 @@ pub mod keys { pub const PARALLEL_FAN_IN_BEST_OUTCOME: &str = "parallel.fan_in.best_outcome"; pub const PARALLEL_FAN_IN_BEST_HEAD_SHA: &str = "parallel.fan_in.best_head_sha"; - /// Runtime-only keys stripped from durable context projections - /// (checkpoint snapshots and resume normalization). Add new transient - /// keys here so both strip sites stay in sync. - pub const TRANSIENT_CONTEXT_KEYS: &[&str] = + /// Runtime-only keys stripped from durable context projections. + pub(crate) const TRANSIENT_CONTEXT_KEYS: &[&str] = &[CURRENT_PREAMBLE, INTERNAL_PARALLEL_BRANCH_PREAMBLES]; // --- Prefix constants (for filtering and dynamic keys) --- @@ -152,15 +150,12 @@ use crate::event::StageScope; /// One entry of the [`keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES`] stash. /// /// The stash is a JSON array indexed by the parallel node's outgoing-edge -/// order (`Graph::outgoing_edges` preserves declaration order, so producer and -/// consumer align even with duplicate targets). `null` entries mean the branch -/// inherits the fork's preamble. `FidelityLifecycle::before_node` produces the -/// stash; `ParallelHandler::execute` consumes and clears it. +/// order. `null` entries mean the branch inherits the fork's preamble. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] -pub struct ParallelBranchPreamble { - pub fidelity: Fidelity, - pub preamble: String, +pub(crate) struct ParallelBranchPreamble { + pub(crate) fidelity: Fidelity, + pub(crate) preamble: String, } /// Domain-specific typed accessors for workflow context values. diff --git a/lib/crates/fabro-workflow/src/lifecycle/fidelity.rs b/lib/crates/fabro-workflow/src/lifecycle/fidelity.rs index 437550cc1..37d2c7e33 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/fidelity.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/fidelity.rs @@ -74,47 +74,52 @@ impl FidelityLifecycle { resolved_outcomes: &HashMap, completed_nodes: &[String], ) -> Vec { - let mut rendered: HashMap = HashMap::new(); - self.graph - .outgoing_edges(node_id) - .into_iter() - .enumerate() - .map(|(branch_index, edge)| { - let Some(target_node) = self.graph.nodes.get(&edge.to) else { - return serde_json::Value::Null; - }; - let resolution = resolve_parallel_branch_fidelity(edge, target_node, fork_fidelity); - if resolution.requested == Some(keys::Fidelity::Full) { - tracing::warn!( - parallel_node = %node_id, - branch = %edge.to, - branch_index, - effective_fidelity = %keys::Fidelity::Full.degraded(), - "Parallel branch fidelity degraded from full" - ); - } - let Some(branch_fidelity) = resolution.effective else { - return serde_json::Value::Null; - }; - rendered - .entry(branch_fidelity) - .or_insert_with(|| { - let entry = ParallelBranchPreamble { - fidelity: branch_fidelity, - preamble: build_preamble( - branch_fidelity, - resolved_context, - &self.graph, - completed_nodes, - resolved_outcomes, - ), - }; - serde_json::to_value(entry) - .expect("ParallelBranchPreamble serialization cannot fail") - }) - .clone() - }) - .collect() + let edges = self.graph.outgoing_edges(node_id); + let mut preambles: Vec = Vec::with_capacity(edges.len()); + let mut rendered: HashMap = HashMap::new(); + + for (branch_index, edge) in edges.into_iter().enumerate() { + let Some(target_node) = self.graph.nodes.get(&edge.to) else { + preambles.push(serde_json::Value::Null); + continue; + }; + let resolution = resolve_parallel_branch_fidelity(edge, target_node, fork_fidelity); + if resolution.requested == Some(keys::Fidelity::Full) { + tracing::warn!( + parallel_node = %node_id, + branch = %edge.to, + branch_index, + effective_fidelity = %keys::Fidelity::Full.degraded(), + "Parallel branch fidelity degraded from full" + ); + } + let Some(branch_fidelity) = resolution.effective else { + preambles.push(serde_json::Value::Null); + continue; + }; + if let Some(&rendered_index) = rendered.get(&branch_fidelity) { + preambles.push(preambles[rendered_index].clone()); + continue; + } + + let entry = ParallelBranchPreamble { + fidelity: branch_fidelity, + preamble: build_preamble( + branch_fidelity, + resolved_context, + &self.graph, + completed_nodes, + resolved_outcomes, + ), + }; + rendered.insert(branch_fidelity, preambles.len()); + preambles.push( + serde_json::to_value(entry) + .expect("ParallelBranchPreamble serialization cannot fail"), + ); + } + + preambles } } From a89528caa625ff53e4c5b4b38d7215b98140bef3 Mon Sep 17 00:00:00 2001 From: Fabro Date: Thu, 23 Jul 2026 20:33:27 +0000 Subject: [PATCH 12/15] fabro(01KY7Y01REECZ24XXTMBZ3PPV9): verify (failed) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fabro-Run: 01KY7Y01REECZ24XXTMBZ3PPV9 Fabro-Completed: 8 Fabro-Checkpoint: 69abcb72b5ece4f5bcc3d4cf75cc67c48f9b1837 ⚒️ Generated with [Fabro](https://fabro.sh) From e736786bf22aaeb077241357fb3ad2616cb74b71 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Thu, 23 Jul 2026 16:46:40 -0400 Subject: [PATCH 13/15] fix(test): eliminate install and session-root races --- lib/crates/fabro-cli/tests/it/cmd/install.rs | 81 +++++++++++++++++--- lib/crates/fabro-test/src/lib.rs | 11 ++- 2 files changed, 78 insertions(+), 14 deletions(-) diff --git a/lib/crates/fabro-cli/tests/it/cmd/install.rs b/lib/crates/fabro-cli/tests/it/cmd/install.rs index fe2c9cf61..e0eb0b75a 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/install.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/install.rs @@ -338,10 +338,16 @@ fn keep_existing_settings_persists_secrets_without_rewriting_server_target() { let mut context = test_context!(); let storage_dir = context.temp_dir.join("install-storage"); context.manage_storage_dir(&storage_dir); - let existing_web_url = unused_loopback_web_url(); - let requested_web_url = unused_loopback_web_url(); - write_http_install_settings(&context, &storage_dir, &existing_web_url, "keep-me"); - login_with_storage_dev_token(&context, &storage_dir, &existing_web_url); + let socket_path = context.temp_dir.join("install-storage.sock"); + let existing_web_url = "https://existing.example.test".to_string(); + let requested_web_url = "https://requested.example.test".to_string(); + write_unix_install_settings( + &context, + &storage_dir, + &socket_path, + &existing_web_url, + "keep-me", + ); let path = fake_gh_path(&context, "ghp_keep_existing"); let output = context @@ -382,15 +388,19 @@ fn keep_existing_settings_persists_secrets_without_rewriting_server_target() { .and_then(toml::Value::as_str), Some(existing_web_url.as_str()) ); + let target = parsed + .get("cli") + .and_then(toml::Value::as_table) + .and_then(|cli| cli.get("target")) + .and_then(toml::Value::as_table) + .expect("cli.target should remain configured"); assert_eq!( - parsed - .get("cli") - .and_then(toml::Value::as_table) - .and_then(|cli| cli.get("target")) - .and_then(toml::Value::as_table) - .and_then(|target| target.get("url")) - .and_then(toml::Value::as_str), - Some(existing_web_url.as_str()) + target.get("type").and_then(toml::Value::as_str), + Some("unix") + ); + assert_eq!( + target.get("path").and_then(toml::Value::as_str), + socket_path.to_str() ); assert!( !settings.contains(&requested_web_url), @@ -874,6 +884,53 @@ mode = "{}" ); } +fn write_unix_install_settings( + context: &fabro_test::TestContext, + storage_dir: &std::path::Path, + socket_path: &std::path::Path, + web_url: &str, + metadata_mode: &str, +) { + write_raw_home_settings( + context, + &format!( + r#" +_version = 1 + +[server.storage] +root = "{}" + +[server.api] +url = "{}/api/v1" + +[server.web] +enabled = true +url = "{}" + +[server.auth] +methods = ["dev-token"] + +[server.listen] +type = "unix" +path = "{}" + +[cli.target] +type = "unix" +path = "{}" + +[project.metadata] +mode = "{}" +"#, + storage_dir.display(), + web_url, + web_url, + socket_path.display(), + socket_path.display(), + metadata_mode + ), + ); +} + fn login_with_storage_dev_token( context: &fabro_test::TestContext, storage_dir: &std::path::Path, diff --git a/lib/crates/fabro-test/src/lib.rs b/lib/crates/fabro-test/src/lib.rs index 9a3e9bc86..74a5f6727 100644 --- a/lib/crates/fabro-test/src/lib.rs +++ b/lib/crates/fabro-test/src/lib.rs @@ -497,8 +497,15 @@ fn with_session_lock(root: &Path, f: impl FnOnce() -> T) -> T { // cleanup_session_root can remove_dir_all between the two calls. let deadline = std::time::Instant::now() + SESSION_LOCK_TIMEOUT; let lock_file = loop { - std::fs::create_dir_all(root) - .unwrap_or_else(|err| panic!("failed to create {}: {err}", root.display())); + if let Err(err) = std::fs::create_dir_all(root) { + assert!( + std::time::Instant::now() < deadline, + "failed to create {} after retries: {err}", + root.display() + ); + std::thread::sleep(Duration::from_millis(10)); + continue; + } ensure_parent_dir(&lock_path); match File::create(&lock_path) { Ok(f) => break f, From a58d65fd3c2a2a972939fb46489ca2d475881cc0 Mon Sep 17 00:00:00 2001 From: Fabro Date: Thu, 23 Jul 2026 20:51:43 +0000 Subject: [PATCH 14/15] fabro(01KY7Y01REECZ24XXTMBZ3PPV9): fixup (succeeded) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fabro-Run: 01KY7Y01REECZ24XXTMBZ3PPV9 Fabro-Completed: 9 Fabro-Checkpoint: 3cefb504a7a71a4e26595ec35e6c867e5fe45ca0 ⚒️ Generated with [Fabro](https://fabro.sh) From fd4826742339911be63216b6c7fed34a3ec1dabd Mon Sep 17 00:00:00 2001 From: Fabro Date: Thu, 23 Jul 2026 20:57:21 +0000 Subject: [PATCH 15/15] fabro(01KY7Y01REECZ24XXTMBZ3PPV9): verify (succeeded) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fabro-Run: 01KY7Y01REECZ24XXTMBZ3PPV9 Fabro-Completed: 10 Fabro-Checkpoint: cecbfc763b42d83f12a72c9303c5ea2bac51c54f ⚒️ Generated with [Fabro](https://fabro.sh)