diff --git a/lib/crates/fabro-cli/src/server_client.rs b/lib/crates/fabro-cli/src/server_client.rs index b5fc56bc0..a4aa44822 100644 --- a/lib/crates/fabro-cli/src/server_client.rs +++ b/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 { - 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 { 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; diff --git a/lib/crates/fabro-cli/tests/it/cmd/support.rs b/lib/crates/fabro-cli/tests/it/cmd/support.rs index 5823a60de..62e7cf38e 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/support.rs +++ b/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 { .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::, _>>() .expect("wire event envelope list should parse") } -fn wire_event_envelope_value_into_store(value: serde_json::Value) -> Result { - 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; diff --git a/lib/crates/fabro-cli/tests/it/workflow/mod.rs b/lib/crates/fabro-cli/tests/it/workflow/mod.rs index 0a5d25e37..6fe7383f1 100644 --- a/lib/crates/fabro-cli/tests/it/workflow/mod.rs +++ b/lib/crates/fabro-cli/tests/it/workflow/mod.rs @@ -173,7 +173,15 @@ fn run_events(run_dir: &Path) -> Vec { 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::, _>>() + .expect("wire event envelope list should parse") } macro_rules! sandbox_tests { diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index 76e8b5e13..1672da3d2 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/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 { - // Wire EventEnvelope is flattened: seq sits alongside the RunEvent - // payload fields at the top level. The progenitor-generated type - // reflects that shape, so we merge seq into the payload value and - // deserialize directly. - let mut value = event.payload.as_value().clone(); - match value.as_object_mut() { - 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()); - } - } + let value = event.to_wire_value().map_err(|err| { + ApiError::new( + StatusCode::INTERNAL_SERVER_ERROR, + format!("Failed to serialize stored event: {err}"), + ) + .into_response() + })?; serde_json::from_value(value).map_err(|err| { ApiError::new( StatusCode::INTERNAL_SERVER_ERROR, diff --git a/lib/crates/fabro-store/src/types.rs b/lib/crates/fabro-store/src/types.rs index 80073b6f3..1c63553b5 100644 --- a/lib/crates/fabro-store/src/types.rs +++ b/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 { + 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 { + 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); + } +} diff --git a/lib/crates/fabro-workflow/src/context.rs b/lib/crates/fabro-workflow/src/context.rs index fc3cc9463..2b33ceedb 100644 --- a/lib/crates/fabro-workflow/src/context.rs +++ b/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; fn preamble(&self) -> String; fn run_id(&self) -> String; + fn parallel_group_id(&self) -> Option; + fn parallel_branch_id(&self) -> Option; } 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 { + self.get(keys::INTERNAL_PARALLEL_GROUP_ID) + .and_then(|value| value.as_str().map(String::from)) + } + + fn parallel_branch_id(&self) -> Option { + 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(); diff --git a/lib/crates/fabro-workflow/src/error.rs b/lib/crates/fabro-workflow/src/error.rs index b81f690a6..a6ea8e01a 100644 --- a/lib/crates/fabro-workflow/src/error.rs +++ b/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, }; diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs index 80887a138..eddf0f2d7 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/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, + #[serde(default, skip_serializing_if = "Option::is_none")] + parallel_branch_id: Option, 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, + #[serde(default, skip_serializing_if = "Option::is_none")] + parallel_branch_id: Option, duration_ms: u64, status: String, preferred_label: Option, @@ -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, + #[serde(default, skip_serializing_if = "Option::is_none")] + parallel_branch_id: Option, 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, + #[serde(default, skip_serializing_if = "Option::is_none")] + parallel_branch_id: Option, attempt: usize, max_attempts: usize, delay_ms: u64, @@ -360,6 +377,10 @@ pub enum Event { session_id: Option, #[serde(default, skip_serializing_if = "Option::is_none")] parent_session_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + parallel_group_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + parallel_branch_id: Option, }, 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"); diff --git a/lib/crates/fabro-workflow/src/git.rs b/lib/crates/fabro-workflow/src/git.rs index 504bd03e5..6f1a84551 100644 --- a/lib/crates/fabro-workflow/src/git.rs +++ b/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, diff --git a/lib/crates/fabro-workflow/src/handler/agent.rs b/lib/crates/fabro-workflow/src/handler/agent.rs index 6c1beb854..85ea0e26d 100644 --- a/lib/crates/fabro-workflow/src/handler/agent.rs +++ b/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(), diff --git a/lib/crates/fabro-workflow/src/handler/llm/api.rs b/lib/crates/fabro-workflow/src/handler/llm/api.rs index 7c59044ef..d8f7a934f 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/api.rs +++ b/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, + parallel_branch_id: Option, +} + +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, file_tracking: Arc>, ) { @@ -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), ); diff --git a/lib/crates/fabro-workflow/src/handler/parallel.rs b/lib/crates/fabro-workflow/src/handler/parallel.rs index bec2bb5d8..ccd07125d 100644 --- a/lib/crates/fabro-workflow/src/handler/parallel.rs +++ b/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!(¶llel_group_id), + ); + branch_context.set( + keys::INTERNAL_PARALLEL_BRANCH_ID, + serde_json::json!(¶llel_branch_id), + ); let (branch_sandbox, worktree_path): (Arc, Option) = 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, diff --git a/lib/crates/fabro-workflow/src/lifecycle/event.rs b/lib/crates/fabro-workflow/src/lifecycle/event.rs index f6cbae9ed..6bca3b85e 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/event.rs +++ b/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, Option) { + ( + state.context.parallel_group_id(), + state.context.parallel_branch_id(), + ) +} + #[async_trait] impl RunLifecycle for EventLifecycle { async fn on_run_start(&self, _graph: &WorkflowGraph, _state: &WfRunState) -> CoreResult<()> { @@ -126,6 +134,7 @@ impl RunLifecycle 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 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 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 for EventLifecycle { state: &WfRunState, ) -> CoreResult>> { 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 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 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 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 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 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(), diff --git a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs index 0f8631b79..33c5bd1e8 100644 --- a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs +++ b/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, diff --git a/lib/crates/fabro-workflow/src/pipeline/retro.rs b/lib/crates/fabro-workflow/src/pipeline/retro.rs index 1511c0ced..7073e4cf7 100644 --- a/lib/crates/fabro-workflow/src/pipeline/retro.rs +++ b/lib/crates/fabro-workflow/src/pipeline/retro.rs @@ -76,6 +76,8 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option { 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, }); } })