scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-08-28 05:27:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

refactor(events): simplify envelope metadata plumbing

Browse source 
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.
This commit is contained in:
Bryan Helmkamp 2026-04-09 11:08:38 -04:00
parent 58f400c70c
commit 44def7866e
15 changed files with 280 additions and 85 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

31
lib/crates/fabro-cli/src/server_client.rs
View file

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

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

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

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

@ -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 {

27
lib/crates/fabro-server/src/server.rs
View file

@ -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
// 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,

81
lib/crates/fabro-store/src/types.rs
View file

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

36
lib/crates/fabro-workflow/src/context.rs
View file

@ -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();

2
lib/crates/fabro-workflow/src/error.rs
View file

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

79
lib/crates/fabro-workflow/src/event.rs
View file

@ -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");

2
lib/crates/fabro-workflow/src/git.rs
View file

@ -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,

2
lib/crates/fabro-workflow/src/handler/agent.rs
View file

@ -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(),

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

@ -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),
);

14
lib/crates/fabro-workflow/src/handler/parallel.rs
View file

@ -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,

26
lib/crates/fabro-workflow/src/lifecycle/event.rs
View file

@ -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(),

2
lib/crates/fabro-workflow/src/pipeline/pull_request.rs
View file

@ -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,

2
lib/crates/fabro-workflow/src/pipeline/retro.rs
View file

@ -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,
});
}
})