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<ParallelBranchResult> (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 <noreply@anthropic.com>

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>): 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" },

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;
     }
   }

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<String, serde_json::Value>,
+    after: HashMap<String, serde_json::Value>,
+) -> HashMap<String, serde_json::Value> {
+    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();

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<usize, Error> {
-    #[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<BranchShape> = serde_json::from_value(value)
+    let results: Vec<ParallelBranchResult> = serde_json::from_value(value)
         .map_err(|err| Error::handler_with_source("Invalid parallel results", err))?;
     Ok(results.len())
 }

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<String, serde_json::Value> = 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<String, serde_json::Value> = 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;

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, Error>(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::<BTreeMap<_, _>>();
-    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<String>) -> BranchResult {
     let outcome = Outcome::fail_classify(reason);
     BranchResult {