refactor(events): simplify envelope metadata plumbing

Centralize flattened EventEnvelope conversion in fabro-store so the CLI,
server, and test helpers reuse one wire-shape path. Also thread parallel
group and branch ids through nested stage and agent events so the new
envelope fields stay populated inside parallel branches.

lib/crates/fabro-cli/src/server_client.rs

@@ -74,43 +74,16 @@ impl RunAttachEventStream {
         for payload in sse::drain_sse_payloads(&mut self.pending_bytes, finalize) {
             let value: serde_json::Value = serde_json::from_str(&payload)?;
             self.buffered_events
-                .push_back(wire_event_envelope_into_store(value)?);
+                .push_back(EventEnvelope::from_wire_value(value)?);
         }
         Ok(())
     }
 }
 
-/// Converts a flattened wire `EventEnvelope` JSON value (seq alongside the
-/// RunEvent payload fields at the top level) into the internal
-/// `fabro_store::EventEnvelope` which keeps seq and payload separate.
-fn wire_event_envelope_into_store(value: serde_json::Value) -> Result<EventEnvelope> {
-    let serde_json::Value::Object(mut obj) = value else {
-        bail!("expected wire EventEnvelope JSON object");
-    };
-    let seq_value = obj
-        .remove("seq")
-        .context("wire EventEnvelope missing seq field")?;
-    let seq: u32 = match seq_value {
-        serde_json::Value::Number(n) => n
-            .as_u64()
-            .and_then(|v| u32::try_from(v).ok())
-            .context("wire EventEnvelope seq is out of u32 range")?,
-        _ => bail!("wire EventEnvelope seq is not a number"),
-    };
-    let run_id_str = obj
-        .get("run_id")
-        .and_then(|v| v.as_str())
-        .context("wire EventEnvelope missing run_id")?;
-    let run_id: RunId = run_id_str.parse().context("invalid run_id in wire event")?;
-    let payload = fabro_store::EventPayload::new(serde_json::Value::Object(obj), &run_id)
-        .map_err(|err| anyhow!("wire EventEnvelope payload failed store validation: {err}"))?;
-    Ok(EventEnvelope { seq, payload })
-}
-
 fn wire_event_envelope_from_generated(value: types::EventEnvelope) -> Result<EventEnvelope> {
     let value =
         serde_json::to_value(value).context("failed to serialize generated EventEnvelope")?;
-    wire_event_envelope_into_store(value)
+    EventEnvelope::from_wire_value(value).map_err(Into::into)
 }
 
 pub(crate) use fabro_store::RunProjection;

lib/crates/fabro-cli/tests/it/cmd/support.rs

@@ -14,7 +14,6 @@ use fabro_config::Storage;
 use fabro_server::bind::Bind;
 use fabro_store::EventEnvelope;
 use fabro_test::TestContext;
-use fabro_types::RunId;
 use serde_json::Value;
 use shlex::try_quote;
 
@@ -667,33 +666,11 @@ pub(crate) fn run_events(run_dir: &Path) -> Vec<EventEnvelope> {
         .expect("event list response should contain a data array");
     items
         .into_iter()
-        .map(wire_event_envelope_value_into_store)
+        .map(EventEnvelope::from_wire_value)
         .collect::<Result<Vec<_>, _>>()
         .expect("wire event envelope list should parse")
 }
 
-fn wire_event_envelope_value_into_store(value: serde_json::Value) -> Result<EventEnvelope, String> {
-    let mut obj = match value {
-        serde_json::Value::Object(obj) => obj,
-        _ => return Err("wire envelope is not an object".to_string()),
-    };
-    let seq = obj
-        .remove("seq")
-        .and_then(|v| v.as_u64())
-        .and_then(|v| u32::try_from(v).ok())
-        .ok_or_else(|| "wire envelope missing valid seq".to_string())?;
-    let run_id_str = obj
-        .get("run_id")
-        .and_then(|v| v.as_str())
-        .ok_or_else(|| "wire envelope missing run_id".to_string())?;
-    let run_id: RunId = run_id_str
-        .parse()
-        .map_err(|err| format!("invalid run_id in wire envelope: {err}"))?;
-    let payload = fabro_store::EventPayload::new(serde_json::Value::Object(obj), &run_id)
-        .map_err(|err| format!("wire envelope payload failed store validation: {err}"))?;
-    Ok(EventEnvelope { seq, payload })
-}
-
 pub(crate) fn wait_for_event_names(run_dir: &Path, expected: &[&str]) {
     let deadline = std::time::Instant::now() + COMMAND_TIMEOUT;
 

lib/crates/fabro-cli/tests/it/workflow/mod.rs

@@ -173,7 +173,15 @@ fn run_events(run_dir: &Path) -> Vec<EventEnvelope> {
         storage_dir,
         &format!("/api/v1/runs/{run_id}/events"),
     ));
-    serde_json::from_value(response["data"].clone()).expect("event list should parse")
+    let items = response["data"]
+        .as_array()
+        .cloned()
+        .expect("event list response should contain a data array");
+    items
+        .into_iter()
+        .map(EventEnvelope::from_wire_value)
+        .collect::<Result<Vec<_>, _>>()
+        .expect("wire event envelope list should parse")
 }
 
 macro_rules! sandbox_tests {

lib/crates/fabro-server/src/server.rs

@@ -2381,26 +2381,13 @@ fn octet_stream_response(bytes: Bytes) -> Response {
 
 #[allow(clippy::result_large_err)]
 fn api_event_envelope_from_store(event: &EventEnvelope) -> Result<ApiEventEnvelope, Response> {
-    // Wire EventEnvelope is flattened: seq sits alongside the RunEvent
+    let value = event.to_wire_value().map_err(|err| {
-    // payload fields at the top level. The progenitor-generated type
+        ApiError::new(
-    // reflects that shape, so we merge seq into the payload value and
+            StatusCode::INTERNAL_SERVER_ERROR,
-    // deserialize directly.
+            format!("Failed to serialize stored event: {err}"),
-    let mut value = event.payload.as_value().clone();
+        )
-    match value.as_object_mut() {
+        .into_response()
-        Some(map) => {
+    })?;
-            map.insert(
-                "seq".to_string(),
-                serde_json::Value::Number(i64::from(event.seq).into()),
-            );
-        }
-        None => {
-            return Err(ApiError::new(
-                StatusCode::INTERNAL_SERVER_ERROR,
-                "stored event payload is not a JSON object".to_string(),
-            )
-            .into_response());
-        }
-    }
     serde_json::from_value(value).map_err(|err| {
         ApiError::new(
             StatusCode::INTERNAL_SERVER_ERROR,

lib/crates/fabro-store/src/types.rs

@@ -85,3 +85,84 @@ pub struct EventEnvelope {
     pub seq: u32,
     pub payload: EventPayload,
 }
+
+impl EventEnvelope {
+    pub fn from_wire_value(value: serde_json::Value) -> Result<Self> {
+        let serde_json::Value::Object(mut obj) = value else {
+            return Err(StoreError::InvalidEvent(
+                "wire EventEnvelope must be a JSON object".into(),
+            ));
+        };
+        let seq = obj
+            .remove("seq")
+            .and_then(|value| value.as_u64())
+            .and_then(|value| u32::try_from(value).ok())
+            .ok_or_else(|| {
+                StoreError::InvalidEvent("wire EventEnvelope missing valid seq".into())
+            })?;
+        let run_id = obj
+            .get("run_id")
+            .and_then(|value| value.as_str())
+            .ok_or_else(|| StoreError::InvalidEvent("wire EventEnvelope missing run_id".into()))?
+            .parse()
+            .map_err(|err| StoreError::InvalidEvent(format!("invalid wire run_id: {err}")))?;
+        let payload = EventPayload::new(serde_json::Value::Object(obj), &run_id)?;
+        Ok(Self { seq, payload })
+    }
+
+    pub fn to_wire_value(&self) -> Result<serde_json::Value> {
+        let mut value = self.payload.as_value().clone();
+        let map = value.as_object_mut().ok_or_else(|| {
+            StoreError::InvalidEvent("stored event payload must be a JSON object".into())
+        })?;
+        map.insert(
+            "seq".to_string(),
+            serde_json::Value::Number(u64::from(self.seq).into()),
+        );
+        Ok(value)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use chrono::{TimeZone, Utc};
+
+    use fabro_types::{EventBody, RunEvent, fixtures, run_event::RunCompletedProps};
+
+    use super::{EventEnvelope, EventPayload};
+
+    #[test]
+    fn wire_event_envelope_round_trips() {
+        let event = RunEvent {
+            id: "evt_1".to_string(),
+            ts: Utc.with_ymd_and_hms(2026, 4, 9, 12, 0, 0).unwrap(),
+            run_id: fixtures::RUN_1,
+            node_id: Some("code".to_string()),
+            node_label: Some("Code".to_string()),
+            stage_id: Some("code@1".to_string()),
+            parallel_group_id: None,
+            parallel_branch_id: None,
+            session_id: None,
+            parent_session_id: None,
+            tool_call_id: None,
+            actor: None,
+            body: EventBody::RunCompleted(RunCompletedProps {
+                duration_ms: 42,
+                artifact_count: 0,
+                status: "success".to_string(),
+                reason: None,
+                total_usd_micros: None,
+                final_git_commit_sha: None,
+                final_patch: None,
+                billing: None,
+            }),
+        };
+        let payload = EventPayload::new(event.to_value().unwrap(), &fixtures::RUN_1).unwrap();
+        let envelope = EventEnvelope { seq: 7, payload };
+
+        let wire = envelope.to_wire_value().unwrap();
+        let parsed = EventEnvelope::from_wire_value(wire).unwrap();
+
+        assert_eq!(parsed, envelope);
+    }
+}

lib/crates/fabro-workflow/src/context.rs

@@ -23,6 +23,8 @@ pub mod keys {
     pub const INTERNAL_THREAD_ID: &str = "internal.thread_id";
     pub const INTERNAL_NODE_VISIT_COUNT: &str = "internal.node_visit_count";
     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";
 
     // --- current.* keys ---
     pub const CURRENT_PREAMBLE: &str = "current.preamble";
@@ -141,6 +143,8 @@ pub trait WorkflowContext {
     fn thread_id(&self) -> Option<String>;
     fn preamble(&self) -> String;
     fn run_id(&self) -> String;
+    fn parallel_group_id(&self) -> Option<String>;
+    fn parallel_branch_id(&self) -> Option<String>;
 }
 
 impl WorkflowContext for Context {
@@ -162,6 +166,16 @@ impl WorkflowContext for Context {
     fn run_id(&self) -> String {
         self.get_string(keys::INTERNAL_RUN_ID, "unknown")
     }
+
+    fn parallel_group_id(&self) -> Option<String> {
+        self.get(keys::INTERNAL_PARALLEL_GROUP_ID)
+            .and_then(|value| value.as_str().map(String::from))
+    }
+
+    fn parallel_branch_id(&self) -> Option<String> {
+        self.get(keys::INTERNAL_PARALLEL_BRANCH_ID)
+            .and_then(|value| value.as_str().map(String::from))
+    }
 }
 
 #[cfg(test)]
@@ -309,6 +323,28 @@ mod tests {
         assert_eq!(ctx.thread_id(), Some("main".to_string()));
     }
 
+    #[test]
+    fn parallel_ids_default() {
+        let ctx = Context::new();
+        assert_eq!(ctx.parallel_group_id(), None);
+        assert_eq!(ctx.parallel_branch_id(), None);
+    }
+
+    #[test]
+    fn parallel_ids_set() {
+        let ctx = Context::new();
+        ctx.set(
+            keys::INTERNAL_PARALLEL_GROUP_ID,
+            serde_json::json!("fanout@2"),
+        );
+        ctx.set(
+            keys::INTERNAL_PARALLEL_BRANCH_ID,
+            serde_json::json!("fanout@2:1"),
+        );
+        assert_eq!(ctx.parallel_group_id(), Some("fanout@2".to_string()));
+        assert_eq!(ctx.parallel_branch_id(), Some("fanout@2:1".to_string()));
+    }
+
     #[test]
     fn node_visit_count_default() {
         let ctx = Context::new();

lib/crates/fabro-workflow/src/error.rs

@@ -1693,6 +1693,8 @@ mod tests {
             name: "code".into(),
             index: 0,
             visit: 1,
+            parallel_group_id: None,
+            parallel_branch_id: None,
             failure: failure.clone(),
             will_retry: false,
         };

lib/crates/fabro-workflow/src/event.rs

@@ -5,7 +5,8 @@ use std::sync::atomic::{AtomicI64, Ordering};
 
 use ::fabro_types::run_event as fabro_types;
 use ::fabro_types::{
-    BilledTokenCounts, RunBlobId, RunControlAction, RunEvent, RunId, StageStatus, StatusReason,
+    BilledTokenCounts, RunBlobId, RunControlAction, RunEvent, RunId, StageId, StageStatus,
+    StatusReason,
 };
 use anyhow::{Context, Result};
 use chrono::Utc;
@@ -138,6 +139,10 @@ pub enum Event {
         name: String,
         index: usize,
         visit: u32,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        parallel_group_id: Option<String>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        parallel_branch_id: Option<String>,
         handler_type: String,
         attempt: usize,
         max_attempts: usize,
@@ -147,6 +152,10 @@ pub enum Event {
         name: String,
         index: usize,
         visit: u32,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        parallel_group_id: Option<String>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        parallel_branch_id: Option<String>,
         duration_ms: u64,
         status: String,
         preferred_label: Option<String>,
@@ -178,6 +187,10 @@ pub enum Event {
         name: String,
         index: usize,
         visit: u32,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        parallel_group_id: Option<String>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        parallel_branch_id: Option<String>,
         failure: FailureDetail,
         will_retry: bool,
     },
@@ -186,6 +199,10 @@ pub enum Event {
         name: String,
         index: usize,
         visit: u32,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        parallel_group_id: Option<String>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        parallel_branch_id: Option<String>,
         attempt: usize,
         max_attempts: usize,
         delay_ms: u64,
@@ -360,6 +377,10 @@ pub enum Event {
         session_id: Option<String>,
         #[serde(default, skip_serializing_if = "Option::is_none")]
         parent_session_id: Option<String>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        parallel_group_id: Option<String>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        parallel_branch_id: Option<String>,
     },
     SubgraphStarted {
         node_id: String,
@@ -1301,33 +1322,43 @@ fn stored_event_fields(event: &Event) -> StoredEventFields {
             node_id,
             name,
             visit,
+            parallel_group_id,
+            parallel_branch_id,
             ..
         }
         | Event::StageFailed {
             node_id,
             name,
             visit,
+            parallel_group_id,
+            parallel_branch_id,
             ..
         }
         | Event::StageStarted {
             node_id,
             name,
             visit,
+            parallel_group_id,
+            parallel_branch_id,
             ..
         }
         | Event::StageRetrying {
             node_id,
             name,
             visit,
+            parallel_group_id,
+            parallel_branch_id,
             ..
         } => {
             let node_id_str = node_id.clone();
             let node_label = default_node_label(Some(&node_id_str), Some(name.clone()));
-            let stage_id = Some(format!("{node_id_str}@{visit}"));
+            let stage_id = Some(StageId::new(node_id_str.clone(), *visit).to_string());
             StoredEventFields {
                 node_id: Some(node_id_str),
                 node_label,
                 stage_id,
+                parallel_group_id: parallel_group_id.clone(),
+                parallel_branch_id: parallel_branch_id.clone(),
                 ..StoredEventFields::default()
             }
         }
@@ -1335,7 +1366,7 @@ fn stored_event_fields(event: &Event) -> StoredEventFields {
         | Event::ParallelCompleted { node_id, visit, .. } => {
             let node_id_str = node_id.clone();
             let node_label = default_node_label(Some(&node_id_str), None);
-            let parallel_group_id = Some(format!("{node_id_str}@{visit}"));
+            let parallel_group_id = Some(StageId::new(node_id_str.clone(), *visit).to_string());
             StoredEventFields {
                 node_id: Some(node_id_str),
                 node_label,
@@ -1367,10 +1398,12 @@ fn stored_event_fields(event: &Event) -> StoredEventFields {
             event: agent_event,
             session_id,
             parent_session_id,
+            parallel_group_id,
+            parallel_branch_id,
         } => {
             let node_id = Some(stage.clone());
             let node_label = default_node_label(node_id.as_ref(), None);
-            let stage_id = Some(format!("{stage}@{visit}"));
+            let stage_id = Some(StageId::new(stage.clone(), *visit).to_string());
             let tool_call_id = agent_tool_call_id(agent_event).map(str::to_string);
             let actor = agent_actor_for_event(agent_event, session_id.as_deref());
             StoredEventFields {
@@ -1379,6 +1412,8 @@ fn stored_event_fields(event: &Event) -> StoredEventFields {
                 node_id,
                 node_label,
                 stage_id,
+                parallel_group_id: parallel_group_id.clone(),
+                parallel_branch_id: parallel_branch_id.clone(),
                 tool_call_id,
                 actor,
                 ..StoredEventFields::default()
@@ -2859,6 +2894,8 @@ mod tests {
                 name: "Plan".to_string(),
                 index: 0,
                 visit: 1,
+                parallel_group_id: None,
+                parallel_branch_id: None,
                 duration_ms: 5000,
                 status: "success".to_string(),
                 preferred_label: None,
@@ -2899,6 +2936,8 @@ mod tests {
                 name: "Plan".to_string(),
                 index: 0,
                 visit: 1,
+                parallel_group_id: None,
+                parallel_branch_id: None,
                 duration_ms: 5000,
                 status: "success".to_string(),
                 preferred_label: None,
@@ -2934,6 +2973,8 @@ mod tests {
                 name: "Code".to_string(),
                 index: 1,
                 visit: 1,
+                parallel_group_id: None,
+                parallel_branch_id: None,
                 failure: FailureDetail::new(
                     "lint failed",
                     crate::outcome::FailureCategory::Deterministic,
@@ -2963,6 +3004,8 @@ mod tests {
                 },
                 session_id: Some("ses_child".to_string()),
                 parent_session_id: Some("ses_parent".to_string()),
+                parallel_group_id: None,
+                parallel_branch_id: None,
             },
         );
 
@@ -3137,11 +3180,33 @@ mod tests {
                 },
                 session_id: None,
                 parent_session_id: None,
+                parallel_group_id: None,
+                parallel_branch_id: None,
             }),
             "agent.sub.spawned"
         );
     }
 
+    #[test]
+    fn stage_started_populates_parallel_ids_when_present() {
+        let stored = to_run_event(
+            &fixtures::RUN_1,
+            &Event::StageStarted {
+                node_id: "review".to_string(),
+                name: "review".to_string(),
+                index: 1,
+                visit: 1,
+                parallel_group_id: Some("fanout@2".to_string()),
+                parallel_branch_id: Some("fanout@2:1".to_string()),
+                handler_type: "agent".to_string(),
+                attempt: 1,
+                max_attempts: 1,
+            },
+        );
+        assert_eq!(stored.parallel_group_id.as_deref(), Some("fanout@2"));
+        assert_eq!(stored.parallel_branch_id.as_deref(), Some("fanout@2:1"));
+    }
+
     #[test]
     fn parallel_started_populates_parallel_group_id() {
         let stored = to_run_event(
@@ -3186,10 +3251,14 @@ mod tests {
                 },
                 session_id: Some("ses_1".to_string()),
                 parent_session_id: None,
+                parallel_group_id: Some("fanout@2".to_string()),
+                parallel_branch_id: Some("fanout@2:0".to_string()),
             },
         );
         assert_eq!(stored.stage_id.as_deref(), Some("code@3"));
         assert_eq!(stored.tool_call_id.as_deref(), Some("call_abc"));
+        assert_eq!(stored.parallel_group_id.as_deref(), Some("fanout@2"));
+        assert_eq!(stored.parallel_branch_id.as_deref(), Some("fanout@2:0"));
     }
 
     #[test]
@@ -3207,6 +3276,8 @@ mod tests {
                 },
                 session_id: Some("ses_agent".to_string()),
                 parent_session_id: None,
+                parallel_group_id: None,
+                parallel_branch_id: None,
             },
         );
         let actor = stored.actor.as_ref().expect("actor set");

lib/crates/fabro-workflow/src/git.rs

@@ -450,6 +450,8 @@ mod tests {
                 name: "Work".into(),
                 index: 2,
                 visit: 2,
+                parallel_group_id: None,
+                parallel_branch_id: None,
                 duration_ms: 100,
                 status: "success".into(),
                 preferred_label: None,

lib/crates/fabro-workflow/src/handler/agent.rs

@@ -719,6 +719,8 @@ mod tests {
                     },
                     session_id: Some("session_123".to_string()),
                     parent_session_id: None,
+                    parallel_group_id: context.parallel_group_id(),
+                    parallel_branch_id: context.parallel_branch_id(),
                 });
                 Ok(CodergenResult::Text {
                     text: "done".to_string(),

lib/crates/fabro-workflow/src/handler/llm/api.rs

@@ -41,6 +41,21 @@ fn current_visit(context: &Context) -> u32 {
     u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX)
 }
 
+#[derive(Clone)]
+struct StageEventScope {
+    visit: u32,
+    parallel_group_id: Option<String>,
+    parallel_branch_id: Option<String>,
+}
+
+fn current_stage_event_scope(context: &Context) -> StageEventScope {
+    StageEventScope {
+        visit: current_visit(context),
+        parallel_group_id: context.parallel_group_id(),
+        parallel_branch_id: context.parallel_branch_id(),
+    }
+}
+
 /// Shared state for tracking file modifications from agent tool calls.
 struct FileTracking {
     /// Maps tool_call_id → file_path for in-flight write/edit calls.
@@ -86,7 +101,7 @@ fn track_file_event(event: &AgentEvent, state: &mut FileTracking) {
 fn spawn_event_forwarder(
     session: &Session,
     node_id: String,
-    visit: u32,
+    scope: StageEventScope,
     emitter: Arc<Emitter>,
     file_tracking: Arc<Mutex<FileTracking>>,
 ) {
@@ -105,10 +120,12 @@ fn spawn_event_forwarder(
             {
                 emitter.emit(&Event::Agent {
                     stage: node_id.clone(),
-                    visit,
+                    visit: scope.visit,
                     event: event.event.clone(),
                     session_id: Some(event.session_id.clone()),
                     parent_session_id: event.parent_session_id.clone(),
+                    parallel_group_id: scope.parallel_group_id.clone(),
+                    parallel_branch_id: scope.parallel_branch_id.clone(),
                 });
             }
         }
@@ -455,12 +472,13 @@ impl CodergenBackend for AgentApiBackend {
             touched: HashSet::new(),
             last: None,
         }));
+        let event_scope = current_stage_event_scope(context);
 
         // Subscribe to session events: forward to pipeline emitter + track files.
         spawn_event_forwarder(
             &session,
             node.id.clone(),
-            current_visit(context),
+            event_scope.clone(),
             Arc::clone(emitter),
             Arc::clone(&file_tracking),
         );
@@ -525,7 +543,7 @@ impl CodergenBackend for AgentApiBackend {
                     spawn_event_forwarder(
                         &session,
                         node.id.clone(),
-                        current_visit(context),
+                        event_scope.clone(),
                         Arc::clone(emitter),
                         Arc::clone(&file_tracking),
                     );

lib/crates/fabro-workflow/src/handler/parallel.rs

@@ -4,7 +4,7 @@ use std::time::Instant;
 
 use async_trait::async_trait;
 use fabro_agent::{Sandbox, WorktreeOptions, WorktreeSandbox};
-use fabro_types::RunId;
+use fabro_types::{RunId, StageId};
 use tokio::sync::Semaphore;
 
 use crate::context::keys;
@@ -152,7 +152,7 @@ impl Handler for ParallelHandler {
         );
 
         let parallel_visit = u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX);
-        let parallel_group_id = format!("{}@{}", node.id, parallel_visit);
+        let parallel_group_id = StageId::new(node.id.clone(), parallel_visit).to_string();
 
         services.emitter.emit(&Event::ParallelStarted {
             node_id: node.id.clone(),
@@ -208,6 +208,15 @@ impl Handler for ParallelHandler {
         for (branch_index, edge) in branches.iter().enumerate() {
             let target_id = edge.to.clone();
             let branch_context = context.fork();
+            let parallel_branch_id = format!("{parallel_group_id}:{branch_index}");
+            branch_context.set(
+                keys::INTERNAL_PARALLEL_GROUP_ID,
+                serde_json::json!(&parallel_group_id),
+            );
+            branch_context.set(
+                keys::INTERNAL_PARALLEL_BRANCH_ID,
+                serde_json::json!(&parallel_branch_id),
+            );
 
             let (branch_sandbox, worktree_path): (Arc<dyn Sandbox>, Option<PathBuf>) = if let (
                 Some(ref gs),
@@ -257,7 +266,6 @@ impl Handler for ParallelHandler {
                 (Arc::clone(&services.sandbox), None)
             };
 
-            let parallel_branch_id = format!("{parallel_group_id}:{branch_index}");
             branch_setups.push(BranchSetup {
                 target_id,
                 branch_index,

lib/crates/fabro-workflow/src/lifecycle/event.rs

@@ -17,6 +17,7 @@ use super::circuit_breaker::CircuitBreakerLifecycle;
 use super::git::GitCheckpointResult;
 use crate::artifact;
 use crate::context;
+use crate::context::WorkflowContext;
 use crate::error::FabroError;
 use crate::event::{Emitter, Event};
 use crate::graph::WorkflowGraph;
@@ -84,6 +85,13 @@ fn stage_visit(state: &WfRunState, node_id: &str) -> u32 {
     u32::try_from(visits.max(1)).unwrap_or(u32::MAX)
 }
 
+fn stage_parallel_ids(state: &WfRunState) -> (Option<String>, Option<String>) {
+    (
+        state.context.parallel_group_id(),
+        state.context.parallel_branch_id(),
+    )
+}
+
 #[async_trait]
 impl RunLifecycle<WorkflowGraph> for EventLifecycle {
     async fn on_run_start(&self, _graph: &WorkflowGraph, _state: &WfRunState) -> CoreResult<()> {
@@ -126,6 +134,7 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
         let gv = node.inner();
         let stage_index = state.stage_index;
         let visit = stage_visit(state, &gv.id);
+        let (parallel_group_id, parallel_branch_id) = stage_parallel_ids(state);
         let (loop_failure_signatures, restart_failure_signatures) =
             snapshot_failure_signatures(&self.circuit_breaker);
         self.emitter.emit(&Event::StageStarted {
@@ -133,6 +142,8 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
             name: gv.label().to_string(),
             index: stage_index,
             visit,
+            parallel_group_id: parallel_group_id.clone(),
+            parallel_branch_id: parallel_branch_id.clone(),
             handler_type: gv.handler_type().unwrap_or_default().to_string(),
             attempt: 1,
             max_attempts: 1,
@@ -142,6 +153,8 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
             name: gv.label().to_string(),
             index: stage_index,
             visit,
+            parallel_group_id,
+            parallel_branch_id,
             duration_ms: 0,
             status: StageStatus::Success.to_string(),
             preferred_label: None,
@@ -171,11 +184,14 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
         state: &WfRunState,
     ) -> CoreResult<NodeDecision<Option<BilledModelUsage>>> {
         let gv = ctx.node.inner();
+        let (parallel_group_id, parallel_branch_id) = stage_parallel_ids(state);
         self.emitter.emit(&Event::StageStarted {
             node_id: gv.id.clone(),
             name: gv.label().to_string(),
             index: state.stage_index,
             visit: stage_visit(state, &gv.id),
+            parallel_group_id,
+            parallel_branch_id,
             handler_type: gv.handler_type().unwrap_or_default().to_string(),
             attempt: ctx.attempt as usize,
             max_attempts: ctx.max_attempts as usize,
@@ -193,12 +209,15 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
             let outcome = &ctx.result.outcome;
             let stage_index = state.stage_index;
             let visit = stage_visit(state, &gv.id);
+            let (parallel_group_id, parallel_branch_id) = stage_parallel_ids(state);
 
             self.emitter.emit(&Event::StageFailed {
                 node_id: gv.id.clone(),
                 name: gv.label().to_string(),
                 index: stage_index,
                 visit,
+                parallel_group_id: parallel_group_id.clone(),
+                parallel_branch_id: parallel_branch_id.clone(),
                 failure: outcome.failure.clone().unwrap_or_else(|| {
                     FailureDetail::new("handler failed", FailureCategory::TransientInfra)
                 }),
@@ -210,6 +229,8 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
                 name: gv.label().to_string(),
                 index: stage_index,
                 visit,
+                parallel_group_id,
+                parallel_branch_id,
                 attempt: ctx.attempt as usize,
                 max_attempts: ctx.result.max_attempts as usize,
                 delay_ms: ctx
@@ -234,6 +255,7 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
         let gv = node.inner();
         let stage_index = state.stage_index;
         let visit = stage_visit(state, &gv.id);
+        let (parallel_group_id, parallel_branch_id) = stage_parallel_ids(state);
         let duration_ms = u64::try_from(result.duration.as_millis()).unwrap();
         let (loop_failure_signatures, restart_failure_signatures) =
             snapshot_failure_signatures(&self.circuit_breaker);
@@ -244,6 +266,8 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
                 name: gv.label().to_string(),
                 index: stage_index,
                 visit,
+                parallel_group_id,
+                parallel_branch_id,
                 failure: outcome.failure.clone().unwrap_or_else(|| {
                     FailureDetail::new("handler failed", FailureCategory::Deterministic)
                 }),
@@ -255,6 +279,8 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
                 name: gv.label().to_string(),
                 index: stage_index,
                 visit,
+                parallel_group_id,
+                parallel_branch_id,
                 duration_ms,
                 status: outcome.status.to_string(),
                 preferred_label: outcome.preferred_label.clone(),

lib/crates/fabro-workflow/src/pipeline/pull_request.rs

@@ -1198,6 +1198,8 @@ mod tests {
                 name: "plan".to_string(),
                 index: 0,
                 visit: 1,
+                parallel_group_id: None,
+                parallel_branch_id: None,
                 duration_ms: 1,
                 status: "success".to_string(),
                 preferred_label: None,

lib/crates/fabro-workflow/src/pipeline/retro.rs

@@ -76,6 +76,8 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option<Retro> {
                             event: event.event.clone(),
                             session_id: Some(event.session_id.clone()),
                             parent_session_id: event.parent_session_id.clone(),
+                            parallel_group_id: None,
+                            parallel_branch_id: None,
                         });
                     }
                 })