From af27e98e1cade1d2225a63eddfcdb3022be177bd Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Fri, 24 Jul 2026 08:28:58 -0400 Subject: [PATCH] refactor: simplify parallel handler and overview parsing - Extract emit_branch_completed() to replace three near-identical ParallelBranchCompleted constructions; status now reads consistently from outcome.status - Add context_diff_public() so parallel.rs and manager_loop.rs share the diff-minus-engine-internal-keys step; move context_diff tests next to the function in context.rs - Replace fan_in's dead BranchShape struct with the canonical Vec (from_value moves, so no payload cloning) - Narrow parseParallelOverview to ParallelBranchSummary {id, status}; its only consumer renders just those fields - Drop helpers.test.ts's duplicate envelope() fixture in favor of the shared makeEventEnvelope Co-Authored-By: Claude Fable 5 --- .../stage-renderers/helpers.test.ts | 66 +++++---------- .../app/components/stage-renderers/helpers.ts | 25 +++--- lib/components/fabro-workflow/src/context.rs | 77 +++++++++++++++++ .../fabro-workflow/src/handler/fan_in.rs | 15 +--- .../src/handler/manager_loop.rs | 79 +----------------- .../fabro-workflow/src/handler/parallel.rs | 83 +++++++++++-------- 6 files changed, 166 insertions(+), 179 deletions(-) diff --git a/apps/fabro-web/app/components/stage-renderers/helpers.test.ts b/apps/fabro-web/app/components/stage-renderers/helpers.test.ts index 4fcf47e95..0982cdc99 100644 --- a/apps/fabro-web/app/components/stage-renderers/helpers.test.ts +++ b/apps/fabro-web/app/components/stage-renderers/helpers.test.ts @@ -1,6 +1,7 @@ import { describe, expect, test } from "bun:test"; import type { EventEnvelope } from "@qltysh/fabro-api-client"; +import { makeEventEnvelope } from "../../lib/test-utils"; import { extractStageContext, parseHumanInterviewPairs, @@ -8,21 +9,10 @@ import { parseReducerTranscript, } from "./helpers"; -function envelope(seq: number, partial: Partial): EventEnvelope { - return { - seq, - id: `evt-${seq}`, - ts: `2026-04-09T12:00:0${seq}Z`, - run_id: "run-1", - event: "stage.prompt", - ...partial, - } as EventEnvelope; -} - describe("parseHumanInterviewPairs", () => { test("pairs interview.started with interview.completed by question_id", () => { const events: EventEnvelope[] = [ - envelope(1, { + makeEventEnvelope(1, { event: "interview.started", properties: { question_id: "q-1", @@ -35,7 +25,7 @@ describe("parseHumanInterviewPairs", () => { allow_freeform: false, }, }), - envelope(2, { + makeEventEnvelope(2, { event: "interview.completed", properties: { question_id: "q-1", @@ -65,7 +55,7 @@ describe("parseHumanInterviewPairs", () => { test("leaves resolution null for unanswered (still pending) questions", () => { const events: EventEnvelope[] = [ - envelope(1, { + makeEventEnvelope(1, { event: "interview.started", properties: { question_id: "q-1", @@ -80,7 +70,7 @@ describe("parseHumanInterviewPairs", () => { test("preserves option description and preview metadata from started events", () => { const events: EventEnvelope[] = [ - envelope(1, { + makeEventEnvelope(1, { event: "interview.started", properties: { question_id: "q-1", @@ -110,19 +100,19 @@ describe("parseHumanInterviewPairs", () => { test("captures timeout and interrupted resolutions", () => { const events: EventEnvelope[] = [ - envelope(1, { + makeEventEnvelope(1, { event: "interview.started", properties: { question_id: "q-1", question: "?", question_type: "freeform" }, }), - envelope(2, { + makeEventEnvelope(2, { event: "interview.timeout", properties: { question_id: "q-1", duration_ms: 30000 }, }), - envelope(3, { + makeEventEnvelope(3, { event: "interview.started", properties: { question_id: "q-2", question: "?", question_type: "freeform" }, }), - envelope(4, { + makeEventEnvelope(4, { event: "interview.interrupted", properties: { question_id: "q-2", @@ -145,11 +135,11 @@ describe("parseHumanInterviewPairs", () => { describe("parseParallelOverview", () => { test("rolls up branch_count and status-only results", () => { const events: EventEnvelope[] = [ - envelope(1, { + makeEventEnvelope(1, { event: "parallel.started", properties: { branch_count: 3 }, }), - envelope(2, { + makeEventEnvelope(2, { event: "parallel.completed", properties: { duration_ms: 12000, @@ -182,21 +172,9 @@ describe("parseParallelOverview", () => { failureCount: 1, durationMs: 12000, results: [ - { - id: "branch-a", - status: "succeeded", - context_updates: { "response.branch-a": "A" }, - }, - { - id: "branch-b", - status: "succeeded", - context_updates: { "command.output": { stdout: "B" } }, - }, - { - id: "branch-c", - status: "failed", - context_updates: { "response.branch-c": "C" }, - }, + { id: "branch-a", status: "succeeded" }, + { id: "branch-b", status: "succeeded" }, + { id: "branch-c", status: "failed" }, ], isComplete: true, }); @@ -204,7 +182,7 @@ describe("parseParallelOverview", () => { test("reports in-flight when only the started event is present", () => { const events: EventEnvelope[] = [ - envelope(1, { + makeEventEnvelope(1, { event: "parallel.started", properties: { branch_count: 4 }, }), @@ -223,7 +201,7 @@ describe("parseReducerTranscript", () => { test("parses the standard prompt transcript when a reducer ran", () => { const events: EventEnvelope[] = [ - envelope(1, { + makeEventEnvelope(1, { event: "stage.prompt", properties: { mode: "prompt", @@ -231,7 +209,7 @@ describe("parseReducerTranscript", () => { model: "claude-sonnet-4-6", }, }), - envelope(2, { + makeEventEnvelope(2, { event: "prompt.completed", properties: { response: "The branch results are joined.", @@ -251,11 +229,11 @@ describe("parseReducerTranscript", () => { test("uses normal prompt mode for the reducer transcript", () => { const events: EventEnvelope[] = [ - envelope(1, { + makeEventEnvelope(1, { event: "stage.prompt", properties: { mode: "prompt", text: "Standard reducer" }, }), - envelope(2, { + makeEventEnvelope(2, { event: "prompt.completed", properties: { response: "Standard response" }, }), @@ -268,7 +246,7 @@ describe("parseReducerTranscript", () => { describe("extractStageContext", () => { test("keeps author-set keys and drops engine bookkeeping keys", () => { const events: EventEnvelope[] = [ - envelope(1, { + makeEventEnvelope(1, { event: "stage.completed", properties: { context_updates: { @@ -296,7 +274,7 @@ describe("extractStageContext", () => { test("extracts routing hints from preferred_label and suggested_next_ids", () => { const events: EventEnvelope[] = [ - envelope(1, { + makeEventEnvelope(1, { event: "stage.completed", properties: { preferred_label: "approve", @@ -312,7 +290,7 @@ describe("extractStageContext", () => { test("returns null when the stage only wrote engine keys", () => { const events: EventEnvelope[] = [ - envelope(1, { + makeEventEnvelope(1, { event: "stage.completed", properties: { context_updates: { last_stage: "implement", "command.output": "blob:x" }, diff --git a/apps/fabro-web/app/components/stage-renderers/helpers.ts b/apps/fabro-web/app/components/stage-renderers/helpers.ts index 3c631a9e2..2ea42b1f4 100644 --- a/apps/fabro-web/app/components/stage-renderers/helpers.ts +++ b/apps/fabro-web/app/components/stage-renderers/helpers.ts @@ -1,7 +1,5 @@ import { StageOutcome } from "@qltysh/fabro-api-client"; -import type { EventEnvelope, ParallelBranchResult } from "@qltysh/fabro-api-client"; - -export type { ParallelBranchResult }; +import type { EventEnvelope } from "@qltysh/fabro-api-client"; import { getArray, getNumber, getObject, getString, type UnknownRecord } from "../../lib/unknown"; @@ -146,12 +144,18 @@ export function parseHumanInterviewPairs(events: EventEnvelope[]): HumanIntervie return Array.from(pairs.values()).sort((a, b) => a.question.ts.localeCompare(b.question.ts)); } +/** Identity and outcome of one branch, parsed from `parallel.completed`. */ +export interface ParallelBranchSummary { + id: string; + status: StageOutcome; +} + export interface ParallelOverview { branchCount: number | null; successCount: number | null; failureCount: number | null; durationMs: number | null; - results: ParallelBranchResult[]; + results: ParallelBranchSummary[]; isComplete: boolean; } @@ -165,7 +169,7 @@ export function parseParallelOverview(events: EventEnvelope[]): ParallelOverview let successCount: number | null = null; let failureCount: number | null = null; let durationMs: number | null = null; - let results: ParallelBranchResult[] = []; + let results: ParallelBranchSummary[] = []; let isComplete = false; for (const event of events) { @@ -184,15 +188,10 @@ export function parseParallelOverview(events: EventEnvelope[]): ParallelOverview if (!record) return null; const id = getString(record, "id"); const status = asStageOutcome(getString(record, "status")); - const contextUpdates = getObject(record, "context_updates"); - if (!id || !status || !contextUpdates) return null; - return { - id, - status, - context_updates: contextUpdates, - } satisfies ParallelBranchResult; + if (!id || !status) return null; + return { id, status } satisfies ParallelBranchSummary; }) - .filter((r): r is ParallelBranchResult => r != null); + .filter((r): r is ParallelBranchSummary => r != null); if (branchCount == null) branchCount = results.length; } } diff --git a/lib/components/fabro-workflow/src/context.rs b/lib/components/fabro-workflow/src/context.rs index 87c93e4c7..86c5ea79e 100644 --- a/lib/components/fabro-workflow/src/context.rs +++ b/lib/components/fabro-workflow/src/context.rs @@ -159,6 +159,19 @@ pub(crate) fn context_diff( .collect() } +/// [`context_diff`] restricted to user-visible keys: the diff that should +/// propagate outside the executing scope (to a parent workflow or across a +/// parallel fork), with engine-internal keys removed. +pub(crate) fn context_diff_public( + before: &HashMap, + after: HashMap, +) -> HashMap { + context_diff(before, after) + .into_iter() + .filter(|(key, _)| !keys::is_engine_internal_key(key)) + .collect() +} + /// One entry of the [`keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES`] stash. /// /// The stash is a JSON array indexed by the parallel node's outgoing-edge @@ -256,6 +269,70 @@ mod tests { assert_eq!(ctx.get("missing"), None); } + #[test] + fn context_diff_detects_additions() { + let before = HashMap::new(); + let mut after = HashMap::new(); + after.insert("key".to_string(), serde_json::json!("value")); + let diff = context_diff(&before, after); + assert_eq!(diff.len(), 1); + assert_eq!(diff.get("key"), Some(&serde_json::json!("value"))); + } + + #[test] + fn context_diff_detects_changes() { + let mut before = HashMap::new(); + before.insert("key".to_string(), serde_json::json!("old")); + let mut after = HashMap::new(); + after.insert("key".to_string(), serde_json::json!("new")); + let diff = context_diff(&before, after); + assert_eq!(diff.len(), 1); + assert_eq!(diff.get("key"), Some(&serde_json::json!("new"))); + } + + #[test] + fn context_diff_ignores_unchanged() { + let mut before = HashMap::new(); + before.insert("key".to_string(), serde_json::json!("same")); + let mut after = HashMap::new(); + after.insert("key".to_string(), serde_json::json!("same")); + let diff = context_diff(&before, after); + assert!(diff.is_empty()); + } + + #[test] + fn context_diff_ignores_deletions() { + let mut before = HashMap::new(); + before.insert("removed".to_string(), serde_json::json!("gone")); + let after = HashMap::new(); + let diff = context_diff(&before, after); + assert!(diff.is_empty()); + } + + #[test] + fn context_diff_public_excludes_engine_internal_keys() { + let before = HashMap::new(); + let mut after = HashMap::new(); + after.insert("graph.goal".to_string(), serde_json::json!("child goal")); + after.insert( + "internal.run_id".to_string(), + serde_json::json!("child-run"), + ); + after.insert( + "thread.main.current_node".to_string(), + serde_json::json!("exit"), + ); + after.insert("current_node".to_string(), serde_json::json!("exit")); + after.insert("response.plan".to_string(), serde_json::json!("the plan")); + after.insert("review.result".to_string(), serde_json::json!("approved")); + + let filtered = context_diff_public(&before, after); + + assert_eq!(filtered.len(), 2); + assert!(filtered.contains_key("response.plan")); + assert!(filtered.contains_key("review.result")); + } + #[test] fn get_string_with_value() { let ctx = Context::new(); diff --git a/lib/components/fabro-workflow/src/handler/fan_in.rs b/lib/components/fabro-workflow/src/handler/fan_in.rs index b6b73db05..2665744fc 100644 --- a/lib/components/fabro-workflow/src/handler/fan_in.rs +++ b/lib/components/fabro-workflow/src/handler/fan_in.rs @@ -3,6 +3,7 @@ use std::sync::Arc; use async_trait::async_trait; use fabro_graphviz::graph::{Graph, Node}; +use fabro_types::ParallelBranchResult; use super::agent::CodergenBackend; use super::prompt::PromptHandler; @@ -90,22 +91,12 @@ impl Handler for FanInHandler { } } -/// Validate that `parallel.results` exists and has the typed shape without -/// cloning the (potentially hydrated) branch payloads into a full -/// [`ParallelBranchResult`] vec that would go unused. +/// Validate that `parallel.results` exists and has the typed shape. fn validated_branch_count(context: &Context) -> Result { - #[derive(serde::Deserialize)] - struct BranchShape { - #[expect(dead_code, reason = "deserialized only to validate the shape")] - id: String, - #[expect(dead_code, reason = "deserialized only to validate the shape")] - status: fabro_types::StageOutcome, - } - let value = context .get(keys::PARALLEL_RESULTS) .ok_or_else(|| Error::handler("No parallel results to join"))?; - let results: Vec = serde_json::from_value(value) + let results: Vec = serde_json::from_value(value) .map_err(|err| Error::handler_with_source("Invalid parallel results", err))?; Ok(results.len()) } diff --git a/lib/components/fabro-workflow/src/handler/manager_loop.rs b/lib/components/fabro-workflow/src/handler/manager_loop.rs index 431ed5d76..13644e997 100644 --- a/lib/components/fabro-workflow/src/handler/manager_loop.rs +++ b/lib/components/fabro-workflow/src/handler/manager_loop.rs @@ -14,7 +14,7 @@ use tokio::time::{sleep, timeout}; use super::{EngineServices, Handler}; use crate::artifact_upload::ArtifactSink; use crate::condition::evaluate_condition; -use crate::context::{Context, WorkflowContext, context_diff, keys}; +use crate::context::{Context, WorkflowContext, context_diff_public, keys}; use crate::error::Error; use crate::operations::{ValidateInput, WorkflowInput, validate}; use crate::outcome::{Outcome, OutcomeExt, StageOutcome}; @@ -282,13 +282,8 @@ impl Handler for SubWorkflowHandler { Err(e) => return Ok(Outcome::fail_classify(format!("Child task panicked: {e}"))), }; - // Compute context diff, filtering engine-internal keys - let raw_diff = - context_diff(&before_snapshot, child_final_context.snapshot()); - let diff: HashMap = raw_diff - .into_iter() - .filter(|(key, _)| !keys::is_engine_internal_key(key)) - .collect(); + let diff = + context_diff_public(&before_snapshot, child_final_context.snapshot()); tracing::debug!( node = %node.id, @@ -803,74 +798,6 @@ mod tests { assert_eq!(parse_duration_str("bad"), Duration::from_secs(45)); } - #[test] - fn context_diff_detects_additions() { - let before = HashMap::new(); - let mut after = HashMap::new(); - after.insert("key".to_string(), serde_json::json!("value")); - let diff = context_diff(&before, after); - assert_eq!(diff.len(), 1); - assert_eq!(diff.get("key"), Some(&serde_json::json!("value"))); - } - - #[test] - fn context_diff_detects_changes() { - let mut before = HashMap::new(); - before.insert("key".to_string(), serde_json::json!("old")); - let mut after = HashMap::new(); - after.insert("key".to_string(), serde_json::json!("new")); - let diff = context_diff(&before, after); - assert_eq!(diff.len(), 1); - assert_eq!(diff.get("key"), Some(&serde_json::json!("new"))); - } - - #[test] - fn context_diff_ignores_unchanged() { - let mut before = HashMap::new(); - before.insert("key".to_string(), serde_json::json!("same")); - let mut after = HashMap::new(); - after.insert("key".to_string(), serde_json::json!("same")); - let diff = context_diff(&before, after); - assert!(diff.is_empty()); - } - - #[test] - fn context_diff_ignores_deletions() { - let mut before = HashMap::new(); - before.insert("removed".to_string(), serde_json::json!("gone")); - let after = HashMap::new(); - let diff = context_diff(&before, after); - assert!(diff.is_empty()); - } - - #[test] - fn context_diff_excludes_engine_internal_keys() { - let before = HashMap::new(); - let mut after = HashMap::new(); - after.insert("graph.goal".to_string(), serde_json::json!("child goal")); - after.insert( - "internal.run_id".to_string(), - serde_json::json!("child-run"), - ); - after.insert( - "thread.main.current_node".to_string(), - serde_json::json!("exit"), - ); - after.insert("current_node".to_string(), serde_json::json!("exit")); - after.insert("response.plan".to_string(), serde_json::json!("the plan")); - after.insert("review.result".to_string(), serde_json::json!("approved")); - - let raw_diff = context_diff(&before, after); - let filtered: HashMap = raw_diff - .into_iter() - .filter(|(key, _)| !keys::is_engine_internal_key(key)) - .collect(); - - assert_eq!(filtered.len(), 2); - assert!(filtered.contains_key("response.plan")); - assert!(filtered.contains_key("review.result")); - } - #[tokio::test] async fn context_flows_parent_to_child_and_back_excludes_internals() { struct ContextEchoHandler; diff --git a/lib/components/fabro-workflow/src/handler/parallel.rs b/lib/components/fabro-workflow/src/handler/parallel.rs index 6102f1ec4..3efd0e14d 100644 --- a/lib/components/fabro-workflow/src/handler/parallel.rs +++ b/lib/components/fabro-workflow/src/handler/parallel.rs @@ -12,9 +12,9 @@ use tokio::sync::Semaphore; use tokio::task::JoinHandle; use super::{EngineServices, Handler}; -use crate::context::{Context, ParallelBranchPreamble, WorkflowContext, context_diff, keys}; +use crate::context::{Context, ParallelBranchPreamble, WorkflowContext, context_diff_public, keys}; use crate::error::Error; -use crate::event::{Event, RunNoticeCode, RunNoticeLevel, StageScope}; +use crate::event::{Emitter, Event, RunNoticeCode, RunNoticeLevel, StageScope}; use crate::hook_context::set_hook_node; use crate::outcome::{FailureCategory, FailureDetail, Outcome, OutcomeExt}; use crate::{artifact, millis_u64}; @@ -238,16 +238,14 @@ async fn run_branches( status: outcome.status, context_updates, }; - branch_services.run.emitter.emit_scoped( - &Event::ParallelBranchCompleted { - parallel_group_id: group_id.clone(), - parallel_branch_id: parallel_branch_id.clone(), - branch: target_id.clone(), - index: branch_index, - duration_ms: millis_u64(branch_start.elapsed()), - status: result.status, - }, + emit_branch_completed( + &branch_services.run.emitter, &branch_scope, + group_id.clone(), + parallel_branch_id.clone(), + branch_index, + millis_u64(branch_start.elapsed()), + outcome.status, ); Ok::(BranchResult { result, outcome }) }; @@ -257,16 +255,14 @@ async fn run_branches( Err(payload) => { let result = failed_branch_result(&target_id, super::format_panic_message(&payload)); - branch_services.run.emitter.emit_scoped( - &Event::ParallelBranchCompleted { - parallel_group_id: group_id, - parallel_branch_id, - branch: target_id, - index: branch_index, - duration_ms: millis_u64(branch_start.elapsed()), - status: result.result.status, - }, + emit_branch_completed( + &branch_services.run.emitter, &branch_scope, + group_id, + parallel_branch_id, + branch_index, + millis_u64(branch_start.elapsed()), + result.outcome.status, ); Ok(result) } @@ -299,16 +295,14 @@ async fn run_branches( ), }; if emit_completion { - services.run.emitter.emit_scoped( - &Event::ParallelBranchCompleted { - parallel_group_id: parallel_group_id.clone(), - parallel_branch_id: dispatch.branch_id, - branch: dispatch.target_id, - index: dispatch.index, - duration_ms: 0, - status: result.result.status, - }, + emit_branch_completed( + &services.run.emitter, &dispatch.scope, + parallel_group_id.clone(), + dispatch.branch_id, + dispatch.index, + 0, + result.outcome.status, ); } if result.outcome.failure_category() == Some(FailureCategory::Canceled) { @@ -421,14 +415,35 @@ fn branch_context_updates( .iter() .map(|(key, value)| (key.clone(), value.clone())) .collect::>(); - updates.extend( - context_diff(before, after) - .into_iter() - .filter(|(key, _)| !keys::is_engine_internal_key(key)), - ); + updates.extend(context_diff_public(before, after)); updates } +/// Emit `ParallelBranchCompleted` for the branch that `scope` identifies; +/// `scope.node_id` is the branch target by construction +/// ([`StageScope::for_parallel_branch`]). +fn emit_branch_completed( + emitter: &Emitter, + scope: &StageScope, + parallel_group_id: StageId, + parallel_branch_id: ParallelBranchId, + index: usize, + duration_ms: u64, + status: StageOutcome, +) { + emitter.emit_scoped( + &Event::ParallelBranchCompleted { + parallel_group_id, + parallel_branch_id, + branch: scope.node_id.clone(), + index, + duration_ms, + status, + }, + scope, + ); +} + fn failed_branch_result(id: &str, reason: impl Into) -> BranchResult { let outcome = Outcome::fail_classify(reason); BranchResult {