From a2650503e8517d0bd41c3fd6d09e79d41e709514 Mon Sep 17 00:00:00 2001 From: Fabro Date: Thu, 23 Jul 2026 16:47:50 +0000 Subject: [PATCH 1/9] 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 2/9] 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 3/9] 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 4/9] 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 5/9] 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 9b85653a2dc62007a5ec2ce8cc8b8f1f4c817ef0 Mon Sep 17 00:00:00 2001 From: Fabro Date: Thu, 23 Jul 2026 20:29:19 +0000 Subject: [PATCH 6/9] 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 7/9] 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 a58d65fd3c2a2a972939fb46489ca2d475881cc0 Mon Sep 17 00:00:00 2001 From: Fabro Date: Thu, 23 Jul 2026 20:51:43 +0000 Subject: [PATCH 8/9] 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 9/9] 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)