diff --git a/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs b/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs index 880338894..f950f2216 100644 --- a/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs +++ b/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs @@ -510,9 +510,6 @@ mod tests { node_id: "plan".into(), name: "Plan".into(), index: 0, - visit: 1, - parallel_group_id: None, - parallel_branch_id: None, duration_ms: 5000, status: "success".into(), preferred_label: None, @@ -557,8 +554,6 @@ mod tests { }, session_id: None, parent_session_id: None, - parallel_group_id: None, - parallel_branch_id: None, }; let stored = to_run_event(&fixtures::RUN_1, &event); diff --git a/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs b/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs index 5e15deb73..cb30ea84f 100644 --- a/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs +++ b/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs @@ -479,8 +479,6 @@ mod tests { event, session_id: None, parent_session_id: None, - parallel_group_id: None, - parallel_branch_id: None, } } @@ -489,9 +487,6 @@ mod tests { node_id: node_id.into(), name: name.into(), index: 0, - visit: 1, - parallel_group_id: None, - parallel_branch_id: None, handler_type: String::new(), attempt: 1, max_attempts: 1, @@ -515,9 +510,6 @@ mod tests { node_id: node_id.into(), name: name.into(), index: 0, - visit: 1, - parallel_group_id: None, - parallel_branch_id: None, duration_ms: 5000, status: "success".into(), preferred_label: None, @@ -714,9 +706,6 @@ mod tests { node_id: "code".into(), name: "Code".into(), index: 0, - visit: 1, - parallel_group_id: None, - parallel_branch_id: None, attempt: 2, max_attempts: 3, delay_ms: 1500, @@ -961,9 +950,6 @@ mod tests { node_id: "code".into(), name: "Code".into(), index: 0, - visit: 1, - parallel_group_id: None, - parallel_branch_id: None, attempt: 2, max_attempts: 3, delay_ms: 1500, @@ -1168,14 +1154,12 @@ mod tests { node_id: "code".into(), name: "Code".into(), index: 0, - visit: 1, - parallel_group_id: None, - parallel_branch_id: None, handler_type: "agent".into(), attempt: 1, max_attempts: 1, }, started_ts, + None, )) .unwrap(); let tool_started = serde_json::to_string(&to_run_event_at( @@ -1189,6 +1173,7 @@ mod tests { }, ), started_ts, + None, )) .unwrap(); let tool_completed = serde_json::to_string(&to_run_event_at( @@ -1203,6 +1188,7 @@ mod tests { }, ), completed_ts, + None, )) .unwrap(); diff --git a/lib/crates/fabro-cli/src/commands/store/dump.rs b/lib/crates/fabro-cli/src/commands/store/dump.rs index 929deb4c1..611f14c04 100644 --- a/lib/crates/fabro-cli/src/commands/store/dump.rs +++ b/lib/crates/fabro-cli/src/commands/store/dump.rs @@ -599,9 +599,6 @@ mod tests { node_id: "code".to_string(), name: "Code".to_string(), index: 1, - visit: 2, - parallel_group_id: None, - parallel_branch_id: None, duration_ms: 250, status: "partial_success".to_string(), preferred_label: None, diff --git a/lib/crates/fabro-cli/tests/it/cmd/attach.rs b/lib/crates/fabro-cli/tests/it/cmd/attach.rs index cedc4882d..9ba4c9d74 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/attach.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/attach.rs @@ -726,6 +726,7 @@ fn attach_json_errors_without_prompting_for_human_input() { "status": "success" }, "run_id": "[ULID]", + "stage_id": "start@1", "ts": "[TIMESTAMP]" }, { @@ -766,6 +767,7 @@ fn attach_json_errors_without_prompting_for_human_input() { "stage": "approve" }, "run_id": "[ULID]", + "stage_id": "approve@1", "ts": "[TIMESTAMP]" } ] diff --git a/lib/crates/fabro-cli/tests/it/cmd/run.rs b/lib/crates/fabro-cli/tests/it/cmd/run.rs index 606e6ccfa..977859943 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/run.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/run.rs @@ -1031,6 +1031,7 @@ fn json_run_implies_auto_approve_for_human_gates() { "status": "success" }, "run_id": "[ULID]", + "stage_id": "start@1", "ts": "[TIMESTAMP]" }, { @@ -1071,11 +1072,14 @@ fn json_run_implies_auto_approve_for_human_gates() { "stage": "approve" }, "run_id": "[ULID]", + "stage_id": "approve@1", "ts": "[TIMESTAMP]" }, { "event": "interview.completed", "id": "[EVENT_ID]", + "node_id": "approve", + "node_label": "approve", "properties": { "answer": "A", "duration_ms": "[DURATION_MS]", @@ -1083,6 +1087,7 @@ fn json_run_implies_auto_approve_for_human_gates() { "question_id": "[ULID]" }, "run_id": "[ULID]", + "stage_id": "approve@1", "ts": "[TIMESTAMP]" }, { @@ -1199,6 +1204,7 @@ fn json_run_implies_auto_approve_for_human_gates() { "status": "success" }, "run_id": "[ULID]", + "stage_id": "approve@1", "ts": "[TIMESTAMP]" }, { @@ -1227,6 +1233,7 @@ fn json_run_implies_auto_approve_for_human_gates() { "script": "echo shipped" }, "run_id": "[ULID]", + "stage_id": "ship@1", "ts": "[TIMESTAMP]" }, { @@ -1242,6 +1249,7 @@ fn json_run_implies_auto_approve_for_human_gates() { "timed_out": false }, "run_id": "[ULID]", + "stage_id": "ship@1", "ts": "[TIMESTAMP]" }, { @@ -1371,6 +1379,7 @@ fn json_run_implies_auto_approve_for_human_gates() { "status": "success" }, "run_id": "[ULID]", + "stage_id": "ship@1", "ts": "[TIMESTAMP]" }, { diff --git a/lib/crates/fabro-workflow/src/context.rs b/lib/crates/fabro-workflow/src/context.rs index 153d8911c..482089f38 100644 --- a/lib/crates/fabro-workflow/src/context.rs +++ b/lib/crates/fabro-workflow/src/context.rs @@ -135,6 +135,8 @@ pub mod keys { pub use fabro_core::Context; +use crate::event::StageScope; +use crate::run_dir::visit_from_context; use fabro_graphviz::Fidelity; use fabro_types::{ParallelBranchId, StageId}; @@ -146,6 +148,10 @@ pub trait WorkflowContext { fn run_id(&self) -> String; fn parallel_group_id(&self) -> Option; fn parallel_branch_id(&self) -> Option; + /// Build the stage-level emit scope from the currently-executing node and its + /// accumulated visit count. Returns `None` for run-level emissions where no + /// stage is active (i.e., `CURRENT_NODE` is unset). + fn current_stage_scope(&self) -> Option; } impl WorkflowContext for Context { @@ -177,6 +183,19 @@ impl WorkflowContext for Context { self.get(keys::INTERNAL_PARALLEL_BRANCH_ID) .and_then(|value| serde_json::from_value(value).ok()) } + + fn current_stage_scope(&self) -> Option { + let node_id = self + .get(keys::CURRENT_NODE) + .and_then(|value| value.as_str().map(String::from))?; + let visit = u32::try_from(visit_from_context(self)).unwrap_or(u32::MAX); + Some(StageScope { + node_id, + visit, + parallel_group_id: self.parallel_group_id(), + parallel_branch_id: self.parallel_branch_id(), + }) + } } #[cfg(test)] diff --git a/lib/crates/fabro-workflow/src/error.rs b/lib/crates/fabro-workflow/src/error.rs index a6ea8e01a..942b442f1 100644 --- a/lib/crates/fabro-workflow/src/error.rs +++ b/lib/crates/fabro-workflow/src/error.rs @@ -1692,9 +1692,6 @@ mod tests { node_id: "code".into(), 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 b2e0f96ef..060f7d116 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/lib/crates/fabro-workflow/src/event.rs @@ -19,8 +19,10 @@ use tokio::io::{AsyncWrite, AsyncWriteExt}; use tokio::sync::{Mutex as AsyncMutex, mpsc, oneshot}; use uuid::Uuid; +use crate::context::{Context as WfContext, WorkflowContext}; use crate::error::FabroError; use crate::outcome::{BilledModelUsage, FailureDetail, Outcome}; +use crate::run_dir::visit_from_context; use fabro_agent::{AgentEvent, SandboxEvent, WorktreeEvent, WorktreeEventCallback}; use fabro_llm::types::TokenCounts as LlmTokenCounts; use fabro_util::redact::redact_json_value; @@ -138,11 +140,6 @@ pub enum Event { node_id: String, 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, @@ -151,11 +148,6 @@ pub enum Event { node_id: String, 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, @@ -186,11 +178,6 @@ pub enum Event { node_id: String, 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, }, @@ -198,11 +185,6 @@ pub enum Event { node_id: String, 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, @@ -377,10 +359,6 @@ 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, @@ -1321,53 +1299,45 @@ fn stage_status_from_string(status: &str) -> StageStatus { serde_json::from_value(Value::String(status.to_string())).expect("valid stage status") } -fn stored_event_fields(event: &Event) -> StoredEventFields { +fn stored_event_fields(event: &Event, scope: Option<&StageScope>) -> StoredEventFields { + let mut fields = stored_event_fields_for_variant(event); + if let Some(scope) = scope { + if fields.node_id.is_none() { + fields.node_id = Some(scope.node_id.clone()); + fields.node_label = default_node_label(Some(&scope.node_id), fields.node_label); + } + if fields.stage_id.is_none() { + fields.stage_id = Some(StageId::new(scope.node_id.clone(), scope.visit)); + } + if fields.parallel_group_id.is_none() { + fields + .parallel_group_id + .clone_from(&scope.parallel_group_id); + } + if fields.parallel_branch_id.is_none() { + fields + .parallel_branch_id + .clone_from(&scope.parallel_branch_id); + } + } + fields +} + +fn stored_event_fields_for_variant(event: &Event) -> StoredEventFields { match event { Event::RunCreated { provenance, .. } => StoredEventFields { actor: provenance.as_ref().and_then(actor_from_provenance), ..StoredEventFields::default() }, - Event::StageCompleted { - 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, - .. - } => { + Event::StageCompleted { node_id, name, .. } + | Event::StageFailed { node_id, name, .. } + | Event::StageStarted { node_id, name, .. } + | Event::StageRetrying { node_id, name, .. } => { let node_id_str = node_id.clone(); let node_label = default_node_label(Some(&node_id_str), Some(name.clone())); - let stage_id = Some(StageId::new(node_id_str.clone(), *visit)); 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() } } @@ -1399,8 +1369,6 @@ 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); @@ -1413,10 +1381,9 @@ 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() } } Event::GitCommit { node_id, .. } => node_stored_fields(node_id.clone()), @@ -2481,12 +2448,44 @@ fn event_body_from_event(event: &Event) -> EventBody { } } -pub fn to_run_event(run_id: &RunId, event: &Event) -> RunEvent { - to_run_event_at(run_id, event, Utc::now()) +/// Stage-level scope threaded through event emission to populate +/// `stage_id` / `parallel_group_id` / `parallel_branch_id` on events +/// that happen inside a concrete stage execution. +#[derive(Clone, Debug)] +pub struct StageScope { + pub node_id: String, + pub visit: u32, + pub parallel_group_id: Option, + pub parallel_branch_id: Option, } -pub fn to_run_event_at(run_id: &RunId, event: &Event, ts: chrono::DateTime) -> RunEvent { - let fields = stored_event_fields(event); +impl StageScope { + /// Build scope for a handler invocation. Prefers the current_stage_scope + /// set by the fidelity lifecycle before_attempt hook, but falls back to + /// a scope synthesized from the node id and the context's visit count + /// for tests and other direct-handler call sites that don't go through + /// the full lifecycle. + pub fn for_handler(context: &WfContext, node_id: impl Into) -> Self { + context.current_stage_scope().unwrap_or_else(|| Self { + node_id: node_id.into(), + visit: u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX), + parallel_group_id: context.parallel_group_id(), + parallel_branch_id: context.parallel_branch_id(), + }) + } +} + +pub fn to_run_event(run_id: &RunId, event: &Event) -> RunEvent { + to_run_event_at(run_id, event, Utc::now(), None) +} + +pub fn to_run_event_at( + run_id: &RunId, + event: &Event, + ts: chrono::DateTime, + scope: Option<&StageScope>, +) -> RunEvent { + let fields = stored_event_fields(event, scope); let body = event_body_from_event(event); RunEvent { id: Uuid::now_v7().to_string(), @@ -2759,6 +2758,14 @@ impl Emitter { } pub fn emit(&self, event: &Event) { + self.emit_with_scope(event, None); + } + + pub fn emit_scoped(&self, event: &Event, scope: &StageScope) { + self.emit_with_scope(event, Some(scope)); + } + + fn emit_with_scope(&self, event: &Event, scope: Option<&StageScope>) { self.last_event_at.store(epoch_millis(), Ordering::Relaxed); event.trace(); if let Event::WorkflowRunStarted { run_id, .. } = event { @@ -2767,7 +2774,7 @@ impl Emitter { "workflow run started event must match emitter run_id" ); } - let stored = to_run_event(&self.run_id, event); + let stored = to_run_event_at(&self.run_id, event, Utc::now(), scope); self.dispatch_run_event(&stored); } @@ -2858,15 +2865,12 @@ mod tests { #[test] fn run_event_stage_completed_places_node_fields_in_header() { - let stored = to_run_event( + let stored = to_run_event_at( &fixtures::RUN_2, &Event::StageCompleted { node_id: "plan".to_string(), 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, @@ -2885,6 +2889,13 @@ mod tests { attempt: 1, max_attempts: 1, }, + Utc::now(), + Some(&StageScope { + node_id: "plan".to_string(), + visit: 1, + parallel_group_id: None, + parallel_branch_id: None, + }), ); assert_eq!(stored.event_name(), "stage.completed"); @@ -2906,9 +2917,6 @@ mod tests { node_id: "plan".to_string(), 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, @@ -2943,9 +2951,6 @@ mod tests { node_id: "code".to_string(), 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, @@ -2975,8 +2980,6 @@ 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, }, ); @@ -3151,8 +3154,6 @@ mod tests { }, session_id: None, parent_session_id: None, - parallel_group_id: None, - parallel_branch_id: None, }), "agent.sub.spawned" ); @@ -3160,19 +3161,23 @@ mod tests { #[test] fn stage_started_populates_parallel_ids_when_present() { - let stored = to_run_event( + let stored = to_run_event_at( &fixtures::RUN_1, &Event::StageStarted { node_id: "review".to_string(), name: "review".to_string(), index: 1, - visit: 1, - parallel_group_id: Some(StageId::new("fanout", 2)), - parallel_branch_id: Some(ParallelBranchId::new(StageId::new("fanout", 2), 1)), handler_type: "agent".to_string(), attempt: 1, max_attempts: 1, }, + Utc::now(), + Some(&StageScope { + node_id: "review".to_string(), + visit: 1, + parallel_group_id: Some(StageId::new("fanout", 2)), + parallel_branch_id: Some(ParallelBranchId::new(StageId::new("fanout", 2), 1)), + }), ); assert_eq!(stored.parallel_group_id, Some(StageId::new("fanout", 2))); assert_eq!( @@ -3216,7 +3221,7 @@ mod tests { #[test] fn agent_tool_started_populates_tool_call_id_and_stage_id() { - let stored = to_run_event( + let stored = to_run_event_at( &fixtures::RUN_1, &Event::Agent { stage: "code".to_string(), @@ -3228,9 +3233,14 @@ mod tests { }, session_id: Some("ses_1".to_string()), parent_session_id: None, + }, + Utc::now(), + Some(&StageScope { + node_id: "code".to_string(), + visit: 3, parallel_group_id: Some(StageId::new("fanout", 2)), parallel_branch_id: Some(ParallelBranchId::new(StageId::new("fanout", 2), 0)), - }, + }), ); assert_eq!(stored.stage_id, Some(StageId::new("code", 3))); assert_eq!(stored.tool_call_id.as_deref(), Some("call_abc")); @@ -3241,6 +3251,70 @@ mod tests { ); } + #[test] + fn stage_scope_populates_stage_id_on_non_stage_events() { + // Events tied to a concrete stage execution but lacking scope in their + // own variant fields (CheckpointCompleted, CommandStarted, PromptCompleted, + // Prompt, InterviewStarted, Failover, GitCommit) should pick up stage_id + // / parallel_group_id / parallel_branch_id from the scope argument. + let scope = StageScope { + node_id: "build".to_string(), + visit: 2, + parallel_group_id: Some(StageId::new("fanout", 1)), + parallel_branch_id: Some(ParallelBranchId::new(StageId::new("fanout", 1), 0)), + }; + + let command_started = to_run_event_at( + &fixtures::RUN_1, + &Event::CommandStarted { + node_id: "build".to_string(), + script: "echo".to_string(), + command: "echo".to_string(), + language: "shell".to_string(), + timeout_ms: None, + }, + Utc::now(), + Some(&scope), + ); + assert_eq!(command_started.stage_id, Some(StageId::new("build", 2))); + assert_eq!(command_started.parallel_group_id, scope.parallel_group_id); + assert_eq!(command_started.parallel_branch_id, scope.parallel_branch_id); + + let prompt = to_run_event_at( + &fixtures::RUN_1, + &Event::Prompt { + stage: "build".to_string(), + visit: 2, + text: "do it".to_string(), + mode: None, + provider: None, + model: None, + }, + Utc::now(), + Some(&scope), + ); + assert_eq!(prompt.stage_id, Some(StageId::new("build", 2))); + + let git_commit = to_run_event_at( + &fixtures::RUN_1, + &Event::GitCommit { + node_id: Some("build".to_string()), + sha: "deadbeef".to_string(), + }, + Utc::now(), + Some(&scope), + ); + assert_eq!(git_commit.stage_id, Some(StageId::new("build", 2))); + } + + #[test] + fn run_level_events_without_scope_leave_stage_id_absent() { + let stored = to_run_event(&fixtures::RUN_1, &Event::RunRunning { reason: None }); + assert!(stored.stage_id.is_none()); + assert!(stored.parallel_group_id.is_none()); + assert!(stored.parallel_branch_id.is_none()); + } + #[test] fn agent_assistant_message_populates_agent_actor() { let stored = to_run_event( @@ -3256,8 +3330,6 @@ 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 6f1a84551..24ccf9fac 100644 --- a/lib/crates/fabro-workflow/src/git.rs +++ b/lib/crates/fabro-workflow/src/git.rs @@ -449,9 +449,6 @@ mod tests { node_id: "work".into(), 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 85ea0e26d..541c2d226 100644 --- a/lib/crates/fabro-workflow/src/handler/agent.rs +++ b/lib/crates/fabro-workflow/src/handler/agent.rs @@ -10,7 +10,7 @@ use fabro_types::RunId; use crate::context::keys; use crate::context::{Context, WorkflowContext}; use crate::error::FabroError; -use crate::event::{Emitter, Event}; +use crate::event::{Emitter, Event, StageScope}; use crate::outcome::{ BilledModelUsage, FailureCategory, FailureDetail, Outcome, OutcomeExt, StageStatus, }; @@ -256,14 +256,18 @@ impl Handler for AgentHandler { .map(String::from) .or_else(|| Some(Provider::default_from_env().as_str().to_string())); let prompt_model = node.model().map(String::from); - services.emitter.emit(&Event::Prompt { - stage: node.id.clone(), - visit, - text: prompt.clone(), - mode: Some("agent".to_string()), - provider: prompt_provider, - model: prompt_model, - }); + let stage_scope = StageScope::for_handler(context, &node.id); + services.emitter.emit_scoped( + &Event::Prompt { + stage: node.id.clone(), + visit, + text: prompt.clone(), + mode: Some("agent".to_string()), + provider: prompt_provider, + model: prompt_model, + }, + &stage_scope, + ); // 3. Call LLM backend (agent loop) let thread_id = context.thread_id(); @@ -329,13 +333,16 @@ impl Handler for AgentHandler { .map(String::from) .or_else(|| Some(Provider::default_from_env().as_str().to_string())) .unwrap_or_default(); - services.emitter.emit(&Event::PromptCompleted { - node_id: node.id.clone(), - response: response_text.clone(), - model: response_model, - provider: response_provider, - billing: stage_usage.clone(), - }); + services.emitter.emit_scoped( + &Event::PromptCompleted { + node_id: node.id.clone(), + response: response_text.clone(), + model: response_model, + provider: response_provider, + billing: stage_usage.clone(), + }, + &stage_scope, + ); // Build and write status let mut outcome = Outcome::success(); @@ -709,19 +716,21 @@ mod tests { _sandbox: &Arc, _tool_hooks: Option>, ) -> Result { - emitter.emit(&crate::event::Event::Agent { - stage: node.id.clone(), - visit: u32::try_from(crate::run_dir::visit_from_context(context)) - .unwrap_or(u32::MAX), - event: fabro_agent::AgentEvent::SessionStarted { - provider: Some("openai".to_string()), - model: Some("gpt-5.4".to_string()), + let scope = StageScope::for_handler(context, &node.id); + emitter.emit_scoped( + &crate::event::Event::Agent { + stage: node.id.clone(), + visit: u32::try_from(crate::run_dir::visit_from_context(context)) + .unwrap_or(u32::MAX), + event: fabro_agent::AgentEvent::SessionStarted { + provider: Some("openai".to_string()), + model: Some("gpt-5.4".to_string()), + }, + session_id: Some("session_123".to_string()), + parent_session_id: None, }, - 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(), - }); + &scope, + ); Ok(CodergenResult::Text { text: "done".to_string(), usage: None, diff --git a/lib/crates/fabro-workflow/src/handler/command.rs b/lib/crates/fabro-workflow/src/handler/command.rs index b25a717ac..09d1e520b 100644 --- a/lib/crates/fabro-workflow/src/handler/command.rs +++ b/lib/crates/fabro-workflow/src/handler/command.rs @@ -4,6 +4,7 @@ use crate::context::Context; use crate::context::keys; use crate::error::FabroError; use crate::event::Event; +use crate::event::StageScope; use crate::outcome::{Outcome, OutcomeExt}; use async_trait::async_trait; use fabro_graphviz::graph::{Graph, Node}; @@ -57,7 +58,7 @@ impl Handler for CommandHandler { async fn execute( &self, node: &Node, - _context: &Context, + context: &Context, _graph: &Graph, _run_dir: &Path, services: &EngineServices, @@ -90,13 +91,17 @@ impl Handler for CommandHandler { } else { script.to_string() }; - services.emitter.emit(&Event::CommandStarted { - node_id: node.id.clone(), - script: script.to_string(), - command: command.clone(), - language: language.to_string(), - timeout_ms: timeout_ms(node), - }); + let stage_scope = StageScope::for_handler(context, &node.id); + services.emitter.emit_scoped( + &Event::CommandStarted { + node_id: node.id.clone(), + script: script.to_string(), + command: command.clone(), + language: language.to_string(), + timeout_ms: timeout_ms(node), + }, + &stage_scope, + ); let timeout_ms = node .timeout() @@ -118,14 +123,17 @@ impl Handler for CommandHandler { let result = result.map_err(|e| FabroError::handler(format!("Failed to spawn script: {e}")))?; - services.emitter.emit(&Event::CommandCompleted { - node_id: node.id.clone(), - stdout: result.stdout.clone(), - stderr: result.stderr.clone(), - exit_code: (!result.timed_out).then_some(result.exit_code), - duration_ms: result.duration_ms, - timed_out: result.timed_out, - }); + services.emitter.emit_scoped( + &Event::CommandCompleted { + node_id: node.id.clone(), + stdout: result.stdout.clone(), + stderr: result.stderr.clone(), + exit_code: (!result.timed_out).then_some(result.exit_code), + duration_ms: result.duration_ms, + timed_out: result.timed_out, + }, + &stage_scope, + ); if result.timed_out { return Err(FabroError::handler(format!( diff --git a/lib/crates/fabro-workflow/src/handler/fan_in.rs b/lib/crates/fabro-workflow/src/handler/fan_in.rs index 69fae9878..458ff4886 100644 --- a/lib/crates/fabro-workflow/src/handler/fan_in.rs +++ b/lib/crates/fabro-workflow/src/handler/fan_in.rs @@ -4,7 +4,7 @@ use std::sync::Arc; use crate::context::Context; use crate::context::keys; use crate::error::FabroError; -use crate::event::{Emitter, Event}; +use crate::event::{Emitter, Event, StageScope}; use crate::outcome::{Outcome, OutcomeExt}; use crate::run_dir::visit_from_context; use crate::sandbox_git::git_merge_ff_only; @@ -232,15 +232,19 @@ async fn llm_evaluate( ); let visit_u32 = u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX); + let stage_scope = StageScope::for_handler(context, node_id); - emitter.emit(&Event::Prompt { - stage: node_id.to_string(), - visit: visit_u32, - text: full_prompt.clone(), - mode: Some("fan_in".to_string()), - provider: None, - model: None, - }); + emitter.emit_scoped( + &Event::Prompt { + stage: node_id.to_string(), + visit: visit_u32, + text: full_prompt.clone(), + mode: Some("fan_in".to_string()), + provider: None, + model: None, + }, + &stage_scope, + ); // Build a synthetic node for the backend call let eval_node = Node::new("fan_in_eval"); @@ -269,13 +273,16 @@ async fn llm_evaluate( .unwrap_or_else(|| "unknown".to_string()); let response_text = serde_json::to_string_pretty(&outcome).unwrap_or_else(|_| "{}".to_string()); - emitter.emit(&Event::PromptCompleted { - node_id: node_id.to_string(), - response: response_text.clone(), - model: String::new(), - provider: String::new(), - billing: None, - }); + emitter.emit_scoped( + &Event::PromptCompleted { + node_id: node_id.to_string(), + response: response_text.clone(), + model: String::new(), + provider: String::new(), + billing: None, + }, + &stage_scope, + ); Ok(Candidate { id: best_id, status: outcome.status.to_string(), @@ -283,13 +290,16 @@ async fn llm_evaluate( }) } Ok(CodergenResult::Text { text, .. }) => { - emitter.emit(&Event::PromptCompleted { - node_id: node_id.to_string(), - response: text.clone(), - model: String::new(), - provider: String::new(), - billing: None, - }); + emitter.emit_scoped( + &Event::PromptCompleted { + node_id: node_id.to_string(), + response: text.clone(), + model: String::new(), + provider: String::new(), + billing: None, + }, + &stage_scope, + ); // The LLM responded with text; try to find a matching candidate ID let text = text.trim().to_string(); diff --git a/lib/crates/fabro-workflow/src/handler/human.rs b/lib/crates/fabro-workflow/src/handler/human.rs index ba3842be4..7475378cf 100644 --- a/lib/crates/fabro-workflow/src/handler/human.rs +++ b/lib/crates/fabro-workflow/src/handler/human.rs @@ -8,7 +8,7 @@ use async_trait::async_trait; use crate::context::Context; use crate::context::keys; use crate::error::FabroError; -use crate::event::{Emitter, Event}; +use crate::event::{Emitter, Event, StageScope}; use crate::millis_u64; use crate::outcome::{Outcome, OutcomeExt}; use fabro_graphviz::graph::{Graph, Node}; @@ -88,10 +88,10 @@ impl HumanHandler { self } - fn emit(&self, default_emitter: &Arc, event: &Event) { + fn emit(&self, default_emitter: &Arc, event: &Event, scope: &StageScope) { match &self.emitter { - Some(emitter) => emitter.emit(event), - None => default_emitter.emit(event), + Some(emitter) => emitter.emit_scoped(event, scope), + None => default_emitter.emit_scoped(event, scope), } } } @@ -209,6 +209,7 @@ impl Handler for HumanHandler { // 3. Present to interviewer let question_text = node.label().to_string(); let question_id = question.id.clone(); + let stage_scope = StageScope::for_handler(context, &node.id); self.emit( &services.emitter, &Event::InterviewStarted { @@ -228,6 +229,7 @@ impl Handler for HumanHandler { timeout_seconds: question.timeout_seconds, context_display: question.context_display.clone(), }, + &stage_scope, ); let interview_start = Instant::now(); let answer = self.interviewer.ask(question).await; @@ -242,6 +244,7 @@ impl Handler for HumanHandler { stage: node.id.clone(), duration_ms: millis_u64(interview_start.elapsed()), }, + &stage_scope, ); let default_choice = node .attrs @@ -279,6 +282,7 @@ impl Handler for HumanHandler { reason: "interrupted".to_string(), duration_ms: millis_u64(interview_start.elapsed()), }, + &stage_scope, ); return Ok(unanswered_human_gate( "human interaction interrupted before an answer was provided", @@ -293,6 +297,7 @@ impl Handler for HumanHandler { answer: answer_text(&answer), duration_ms: millis_u64(interview_start.elapsed()), }, + &stage_scope, ); return Ok(unanswered_human_gate("human skipped interaction")); } @@ -306,6 +311,7 @@ impl Handler for HumanHandler { answer: answer_text(&answer), duration_ms: millis_u64(interview_start.elapsed()), }, + &stage_scope, ); // 6. Try fixed-choice match diff --git a/lib/crates/fabro-workflow/src/handler/llm/api.rs b/lib/crates/fabro-workflow/src/handler/llm/api.rs index 73390f201..3e03af58d 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/api.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/api.rs @@ -3,6 +3,7 @@ use std::sync::{Arc, Mutex}; use async_trait::async_trait; +use crate::event::StageScope; use fabro_agent::{ AgentEvent, AgentProfile, AnthropicProfile, GeminiProfile, OpenAiProfile, Sandbox, Session, SessionOptions, Turn, @@ -13,7 +14,6 @@ use fabro_llm::types::{Message, Request, TokenCounts}; use fabro_mcp::config::McpServerSettings; use fabro_model::FallbackTarget; use fabro_model::Provider; -use fabro_types::{ParallelBranchId, StageId}; use tokio::sync::Mutex as TokioMutex; use super::super::agent::{CodergenBackend, CodergenResult}; @@ -22,7 +22,6 @@ use crate::context::{Context, WorkflowContext}; use crate::error::FabroError; use crate::event::{Emitter, Event}; use crate::outcome::billed_model_usage_from_llm; -use crate::run_dir::visit_from_context; use fabro_graphviz::graph::Node; fn build_profile(model: &str, provider: Provider) -> Box { @@ -38,21 +37,6 @@ fn build_profile(model: &str, provider: Provider) -> Box { } } -#[derive(Clone)] -struct StageEventScope { - visit: u32, - parallel_group_id: Option, - parallel_branch_id: Option, -} - -fn current_stage_event_scope(context: &Context) -> StageEventScope { - StageEventScope { - visit: u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX), - 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. @@ -98,7 +82,7 @@ fn track_file_event(event: &AgentEvent, state: &mut FileTracking) { fn spawn_event_forwarder( session: &Session, node_id: String, - scope: StageEventScope, + scope: StageScope, emitter: Arc, file_tracking: Arc>, ) { @@ -115,15 +99,16 @@ fn spawn_event_forwarder( if !event.event.is_streaming_noise() && !matches!(&event.event, AgentEvent::ProcessingEnd) { - emitter.emit(&Event::Agent { - stage: node_id.clone(), - 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(), - }); + emitter.emit_scoped( + &Event::Agent { + stage: node_id.clone(), + visit: scope.visit, + event: event.event.clone(), + session_id: Some(event.session_id.clone()), + parent_session_id: event.parent_session_id.clone(), + }, + &scope, + ); } } }); @@ -469,7 +454,7 @@ impl CodergenBackend for AgentApiBackend { touched: HashSet::new(), last: None, })); - let event_scope = current_stage_event_scope(context); + let event_scope = StageScope::for_handler(context, &node.id); // Subscribe to session events: forward to pipeline emitter + track files. spawn_event_forwarder( @@ -503,14 +488,17 @@ impl CodergenBackend for AgentApiBackend { let mut succeeded = false; for target in &self.fallback_chain { - emitter.emit(&Event::Failover { - stage: node.id.clone(), - from_provider: from_provider.clone(), - from_model: from_model.clone(), - to_provider: target.provider.clone(), - to_model: target.model.clone(), - error: error_msg.clone(), - }); + emitter.emit_scoped( + &Event::Failover { + stage: node.id.clone(), + from_provider: from_provider.clone(), + from_model: from_model.clone(), + to_provider: target.provider.clone(), + to_model: target.model.clone(), + error: error_msg.clone(), + }, + &event_scope, + ); let target_provider: Provider = match target.provider.parse() { Ok(p) => p, diff --git a/lib/crates/fabro-workflow/src/handler/llm/cli.rs b/lib/crates/fabro-workflow/src/handler/llm/cli.rs index 6f1bb7401..9e52d8c63 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/cli.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/cli.rs @@ -10,7 +10,7 @@ use tokio::time::sleep; use super::super::agent::{CodergenBackend, CodergenResult}; use crate::context::Context; use crate::error::FabroError; -use crate::event::{Emitter, Event}; +use crate::event::{Emitter, Event, StageScope}; use crate::outcome::billed_model_usage_from_llm; use crate::run_dir::visit_from_context; use fabro_graphviz::graph::Node; @@ -496,14 +496,18 @@ impl CodergenBackend for AgentCliBackend { ensure_cli(cli, provider, sandbox, emitter).await?; let command = cli_command_for_provider(provider, model, &prompt_path); - emitter.emit(&Event::AgentCliStarted { - node_id: node.id.clone(), - visit: current_visit(_context), - mode: "cli".to_string(), - provider: provider.as_str().to_string(), - model: model.to_string(), - command: command.clone(), - }); + let stage_scope = StageScope::for_handler(_context, &node.id); + emitter.emit_scoped( + &Event::AgentCliStarted { + node_id: node.id.clone(), + visit: current_visit(_context), + mode: "cli".to_string(), + provider: provider.as_str().to_string(), + model: model.to_string(), + command: command.clone(), + }, + &stage_scope, + ); // Forward provider API key and custom env vars so the CLI tool can authenticate. // Build a HashMap to pass via exec_command's env_vars parameter — this @@ -620,13 +624,16 @@ impl CodergenBackend for AgentCliBackend { timed_out: false, duration_ms, }; - emitter.emit(&Event::AgentCliCompleted { - node_id: node.id.clone(), - stdout: result.stdout.clone(), - stderr: result.stderr.clone(), - exit_code: result.exit_code, - duration_ms: result.duration_ms, - }); + emitter.emit_scoped( + &Event::AgentCliCompleted { + node_id: node.id.clone(), + stdout: result.stdout.clone(), + stderr: result.stderr.clone(), + exit_code: result.exit_code, + duration_ms: result.duration_ms, + }, + &stage_scope, + ); // 3e. Cleanup temp files let _ = sandbox diff --git a/lib/crates/fabro-workflow/src/handler/parallel.rs b/lib/crates/fabro-workflow/src/handler/parallel.rs index 968e9044b..0592595b9 100644 --- a/lib/crates/fabro-workflow/src/handler/parallel.rs +++ b/lib/crates/fabro-workflow/src/handler/parallel.rs @@ -10,7 +10,7 @@ use tokio::sync::Semaphore; use crate::context::keys; use crate::context::{Context, WorkflowContext}; use crate::error::FabroError; -use crate::event::Event; +use crate::event::{Event, StageScope}; use crate::git::sanitize_ref_component; use crate::hook_context::set_hook_node; use crate::millis_u64; @@ -280,6 +280,8 @@ impl Handler for ParallelHandler { }); } + let parent_scope = StageScope::for_handler(context, &node.id); + // --- Fan out: concurrent execution --- let mut handles = Vec::new(); for setup in branch_setups { @@ -302,6 +304,7 @@ impl Handler for ParallelHandler { .map(|gs| gs.git_author.clone()) .unwrap_or_default(); let group_id = parallel_group_id.clone(); + let branch_scope = parent_scope.clone(); let handle = tokio::spawn(async move { let _permit = sem @@ -309,12 +312,15 @@ impl Handler for ParallelHandler { .await .map_err(|e| FabroError::handler(format!("semaphore error: {e}")))?; - emitter.emit(&Event::ParallelBranchStarted { - parallel_group_id: group_id.clone(), - parallel_branch_id: setup.parallel_branch_id.clone(), - branch: setup.target_id.clone(), - index: setup.branch_index, - }); + emitter.emit_scoped( + &Event::ParallelBranchStarted { + parallel_group_id: group_id.clone(), + parallel_branch_id: setup.parallel_branch_id.clone(), + branch: setup.target_id.clone(), + index: setup.branch_index, + }, + &branch_scope, + ); let branch_start = Instant::now(); let Some(target_node) = graph.nodes.get(&setup.target_id) else { @@ -322,15 +328,18 @@ impl Handler for ParallelHandler { "branch target node not found: {}", setup.target_id )); - emitter.emit(&Event::ParallelBranchCompleted { - parallel_group_id: group_id.clone(), - parallel_branch_id: setup.parallel_branch_id.clone(), - branch: setup.target_id.clone(), - index: setup.branch_index, - duration_ms: millis_u64(branch_start.elapsed()), - status: "fail".to_string(), - head_sha: None, - }); + emitter.emit_scoped( + &Event::ParallelBranchCompleted { + parallel_group_id: group_id.clone(), + parallel_branch_id: setup.parallel_branch_id.clone(), + branch: setup.target_id.clone(), + index: setup.branch_index, + duration_ms: millis_u64(branch_start.elapsed()), + status: "fail".to_string(), + head_sha: None, + }, + &branch_scope, + ); return Ok(BranchResult { id: setup.target_id.clone(), outcome, @@ -408,15 +417,18 @@ impl Handler for ParallelHandler { None }; - emitter.emit(&Event::ParallelBranchCompleted { - parallel_group_id: group_id.clone(), - parallel_branch_id: setup.parallel_branch_id.clone(), - branch: setup.target_id.clone(), - index: setup.branch_index, - duration_ms: millis_u64(branch_start.elapsed()), - status: outcome.status.to_string(), - head_sha: head_sha.clone(), - }); + emitter.emit_scoped( + &Event::ParallelBranchCompleted { + parallel_group_id: group_id.clone(), + parallel_branch_id: setup.parallel_branch_id.clone(), + branch: setup.target_id.clone(), + index: setup.branch_index, + duration_ms: millis_u64(branch_start.elapsed()), + status: outcome.status.to_string(), + head_sha: head_sha.clone(), + }, + &branch_scope, + ); Ok::(BranchResult { id: setup.target_id, diff --git a/lib/crates/fabro-workflow/src/handler/prompt.rs b/lib/crates/fabro-workflow/src/handler/prompt.rs index fe0134d10..9dc5e25c2 100644 --- a/lib/crates/fabro-workflow/src/handler/prompt.rs +++ b/lib/crates/fabro-workflow/src/handler/prompt.rs @@ -3,7 +3,7 @@ use std::path::Path; use crate::context::keys; use crate::context::{Context, WorkflowContext}; use crate::error::FabroError; -use crate::event::Event; +use crate::event::{Event, StageScope}; use crate::outcome::Outcome; use crate::run_dir::visit_from_context; use async_trait::async_trait; @@ -91,14 +91,18 @@ impl Handler for PromptHandler { .map(String::from) .or_else(|| Some(Provider::default_from_env().as_str().to_string())); let prompt_model = node.model().map(String::from); - services.emitter.emit(&Event::Prompt { - stage: node.id.clone(), - visit: u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX), - text: prompt.clone(), - mode: Some("prompt".to_string()), - provider: prompt_provider.clone(), - model: prompt_model.clone(), - }); + let stage_scope = StageScope::for_handler(context, &node.id); + services.emitter.emit_scoped( + &Event::Prompt { + stage: node.id.clone(), + visit: u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX), + text: prompt.clone(), + mode: Some("prompt".to_string()), + provider: prompt_provider.clone(), + model: prompt_model.clone(), + }, + &stage_scope, + ); // 3. Call LLM backend (one_shot) let (response_text, stage_usage, backend_files_touched) = @@ -140,13 +144,16 @@ impl Handler for PromptHandler { .or_else(|| Some(Provider::default_from_env().as_str().to_string())) .unwrap_or_default(); - services.emitter.emit(&Event::PromptCompleted { - node_id: node.id.clone(), - response: response_text.clone(), - model: response_model, - provider: response_provider, - billing: stage_usage.clone(), - }); + services.emitter.emit_scoped( + &Event::PromptCompleted { + node_id: node.id.clone(), + response: response_text.clone(), + model: response_model, + provider: response_provider, + billing: stage_usage.clone(), + }, + &stage_scope, + ); // 4. Build and write status let mut outcome = Outcome::success(); diff --git a/lib/crates/fabro-workflow/src/lifecycle/artifact.rs b/lib/crates/fabro-workflow/src/lifecycle/artifact.rs index 78e980dda..a101857b4 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/artifact.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/artifact.rs @@ -19,6 +19,7 @@ use crate::artifact_upload::ArtifactSink; use crate::event::{Emitter, Event, RunNoticeLevel}; use crate::graph::WorkflowGraph; use crate::graph::WorkflowNode; +use crate::lifecycle::event::stage_scope_for; use crate::outcome::BilledModelUsage; use crate::runtime_store::RunStoreHandle; use fabro_core::lifecycle::NodeDecision; @@ -136,18 +137,22 @@ impl RunLifecycle for ArtifactLifecycle { }); return Ok(()); } + let scope = stage_scope_for(state, node_id); for asset in &summary.captured_assets { self.captured_artifact_count.fetch_add(1, Ordering::Relaxed); - self.emitter.emit(&Event::ArtifactCaptured { - node_id: node_id.to_string(), - attempt: ctx.attempt, - node_slug: node_slug.clone(), - path: asset.path.clone(), - mime: asset.mime.clone(), - content_md5: asset.content_md5.clone(), - content_sha256: asset.content_sha256.clone(), - bytes: asset.bytes, - }); + self.emitter.emit_scoped( + &Event::ArtifactCaptured { + node_id: node_id.to_string(), + attempt: ctx.attempt, + node_slug: node_slug.clone(), + path: asset.path.clone(), + mime: asset.mime.clone(), + content_md5: asset.content_md5.clone(), + content_sha256: asset.content_sha256.clone(), + bytes: asset.bytes, + }, + &scope, + ); } } Ok(_) => {} // no files collected diff --git a/lib/crates/fabro-workflow/src/lifecycle/event.rs b/lib/crates/fabro-workflow/src/lifecycle/event.rs index 00ce55b28..2c9691f2b 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/event.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/event.rs @@ -19,11 +19,11 @@ use crate::artifact; use crate::context; use crate::context::WorkflowContext; use crate::error::FabroError; -use crate::event::{Emitter, Event}; +use crate::event::{Emitter, Event, StageScope}; use crate::graph::WorkflowGraph; use crate::graph::WorkflowNode; use crate::outcome::{BilledModelUsage, FailureCategory, FailureDetail, Outcome, StageStatus}; -use fabro_types::{BilledTokenCounts, ParallelBranchId, RunId, StageId, StatusReason}; +use fabro_types::{BilledTokenCounts, RunId, StatusReason}; type WfRunState = ExecutionState>; type WfNodeResult = NodeResult>; @@ -85,11 +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(), - ) +pub(crate) fn stage_scope_for(state: &WfRunState, node_id: &str) -> StageScope { + StageScope { + node_id: node_id.to_string(), + visit: stage_visit(state, node_id), + parallel_group_id: state.context.parallel_group_id(), + parallel_branch_id: state.context.parallel_branch_id(), + } } #[async_trait] @@ -133,49 +135,48 @@ 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 scope = stage_scope_for(state, &gv.id); let (loop_failure_signatures, restart_failure_signatures) = snapshot_failure_signatures(&self.circuit_breaker); - self.emitter.emit(&Event::StageStarted { - 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(), - handler_type: gv.handler_type().unwrap_or_default().to_string(), - attempt: 1, - max_attempts: 1, - }); - self.emitter.emit(&Event::StageCompleted { - node_id: gv.id.clone(), - 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, - suggested_next_ids: Vec::new(), - billing: None, - failure: None, - notes: None, - files_touched: Vec::new(), - context_updates: None, - jump_to_node: None, - context_values: None, - node_visits: None, - loop_failure_signatures, - restart_failure_signatures, - response: state - .context - .get(&context::keys::response_key(&gv.id)) - .and_then(|value| value.as_str().map(ToOwned::to_owned)), - attempt: 1, - max_attempts: 1, - }); + self.emitter.emit_scoped( + &Event::StageStarted { + node_id: gv.id.clone(), + name: gv.label().to_string(), + index: stage_index, + handler_type: gv.handler_type().unwrap_or_default().to_string(), + attempt: 1, + max_attempts: 1, + }, + &scope, + ); + self.emitter.emit_scoped( + &Event::StageCompleted { + node_id: gv.id.clone(), + name: gv.label().to_string(), + index: stage_index, + duration_ms: 0, + status: StageStatus::Success.to_string(), + preferred_label: None, + suggested_next_ids: Vec::new(), + billing: None, + failure: None, + notes: None, + files_touched: Vec::new(), + context_updates: None, + jump_to_node: None, + context_values: None, + node_visits: None, + loop_failure_signatures, + restart_failure_signatures, + response: state + .context + .get(&context::keys::response_key(&gv.id)) + .and_then(|value| value.as_str().map(ToOwned::to_owned)), + attempt: 1, + max_attempts: 1, + }, + &scope, + ); } async fn before_attempt( @@ -184,18 +185,18 @@ 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, - }); + let scope = stage_scope_for(state, &gv.id); + self.emitter.emit_scoped( + &Event::StageStarted { + node_id: gv.id.clone(), + name: gv.label().to_string(), + index: state.stage_index, + handler_type: gv.handler_type().unwrap_or_default().to_string(), + attempt: ctx.attempt as usize, + max_attempts: ctx.max_attempts as usize, + }, + &scope, + ); Ok(NodeDecision::Continue) } @@ -208,35 +209,34 @@ impl RunLifecycle for EventLifecycle { let gv = ctx.node.inner(); 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); + let scope = stage_scope_for(state, &gv.id); - 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) - }), - will_retry: true, - }); + self.emitter.emit_scoped( + &Event::StageFailed { + node_id: gv.id.clone(), + name: gv.label().to_string(), + index: stage_index, + failure: outcome.failure.clone().unwrap_or_else(|| { + FailureDetail::new("handler failed", FailureCategory::TransientInfra) + }), + will_retry: true, + }, + &scope, + ); - self.emitter.emit(&Event::StageRetrying { - node_id: gv.id.clone(), - 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 - .backoff_delay - .map_or(0, |d| u64::try_from(d.as_millis()).unwrap()), - }); + self.emitter.emit_scoped( + &Event::StageRetrying { + node_id: gv.id.clone(), + name: gv.label().to_string(), + index: stage_index, + attempt: ctx.attempt as usize, + max_attempts: ctx.result.max_attempts as usize, + delay_ms: ctx + .backoff_delay + .map_or(0, |d| u64::try_from(d.as_millis()).unwrap()), + }, + &scope, + ); } Ok(()) } @@ -254,66 +254,66 @@ 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 scope = stage_scope_for(state, &gv.id); let duration_ms = u64::try_from(result.duration.as_millis()).unwrap(); let (loop_failure_signatures, restart_failure_signatures) = snapshot_failure_signatures(&self.circuit_breaker); if outcome.status == StageStatus::Fail { - self.emitter.emit(&Event::StageFailed { - node_id: gv.id.clone(), - 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) - }), - will_retry: false, - }); - } else { - self.emitter.emit(&Event::StageCompleted { - node_id: gv.id.clone(), - 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(), - suggested_next_ids: outcome.suggested_next_ids.clone(), - billing: outcome.usage.clone(), - failure: outcome.failure.clone(), - notes: outcome.notes.clone(), - files_touched: outcome.files_touched.clone(), - context_updates: (!outcome.context_updates.is_empty()).then(|| { - outcome - .context_updates - .clone() - .into_iter() - .collect::>() - }), - jump_to_node: outcome.jump_to_node.clone(), - context_values: { - let snapshot = state.context.snapshot(); - (!snapshot.is_empty()).then(|| snapshot.into_iter().collect::>()) + self.emitter.emit_scoped( + &Event::StageFailed { + node_id: gv.id.clone(), + name: gv.label().to_string(), + index: stage_index, + failure: outcome.failure.clone().unwrap_or_else(|| { + FailureDetail::new("handler failed", FailureCategory::Deterministic) + }), + will_retry: false, }, - node_visits: (!state.node_visits.is_empty()).then(|| { - state - .node_visits - .clone() - .into_iter() - .collect::>() - }), - loop_failure_signatures, - restart_failure_signatures, - response: response_from_outcome(&gv.id, outcome), - attempt: result.attempts as usize, - max_attempts: result.max_attempts as usize, - }); + &scope, + ); + } else { + self.emitter.emit_scoped( + &Event::StageCompleted { + node_id: gv.id.clone(), + name: gv.label().to_string(), + index: stage_index, + duration_ms, + status: outcome.status.to_string(), + preferred_label: outcome.preferred_label.clone(), + suggested_next_ids: outcome.suggested_next_ids.clone(), + billing: outcome.usage.clone(), + failure: outcome.failure.clone(), + notes: outcome.notes.clone(), + files_touched: outcome.files_touched.clone(), + context_updates: (!outcome.context_updates.is_empty()).then(|| { + outcome + .context_updates + .clone() + .into_iter() + .collect::>() + }), + jump_to_node: outcome.jump_to_node.clone(), + context_values: { + let snapshot = state.context.snapshot(); + (!snapshot.is_empty()) + .then(|| snapshot.into_iter().collect::>()) + }, + node_visits: (!state.node_visits.is_empty()).then(|| { + state + .node_visits + .clone() + .into_iter() + .collect::>() + }), + loop_failure_signatures, + restart_failure_signatures, + response: response_from_outcome(&gv.id, outcome), + attempt: result.attempts as usize, + max_attempts: result.max_attempts as usize, + }, + &scope, + ); } Ok(()) } @@ -367,37 +367,44 @@ impl RunLifecycle for EventLifecycle { node_outcomes.insert(node.id().to_string(), result.outcome.clone()); artifact::normalize_durable_outcomes(&mut node_outcomes); - self.emitter.emit(&Event::CheckpointCompleted { - node_id: node.id().to_string(), - status, - current_node: node.id().to_string(), - completed_nodes: state.completed_nodes.clone(), - node_retries: state - .node_retries - .clone() - .into_iter() - .collect::>(), - context_values: context_values.into_iter().collect::>(), - node_outcomes: node_outcomes.into_iter().collect::>(), - next_node_id: next_node_id.map(ToOwned::to_owned), - git_commit_sha: git_sha.clone(), - loop_failure_signatures: loop_failure_signatures.unwrap_or_default(), - restart_failure_signatures: restart_failure_signatures.unwrap_or_default(), - node_visits: state - .node_visits - .clone() - .into_iter() - .collect::>(), - diff, - }); + let scope = stage_scope_for(state, node.id()); + self.emitter.emit_scoped( + &Event::CheckpointCompleted { + node_id: node.id().to_string(), + status, + current_node: node.id().to_string(), + completed_nodes: state.completed_nodes.clone(), + node_retries: state + .node_retries + .clone() + .into_iter() + .collect::>(), + context_values: context_values.into_iter().collect::>(), + node_outcomes: node_outcomes.into_iter().collect::>(), + next_node_id: next_node_id.map(ToOwned::to_owned), + git_commit_sha: git_sha.clone(), + loop_failure_signatures: loop_failure_signatures.unwrap_or_default(), + restart_failure_signatures: restart_failure_signatures.unwrap_or_default(), + node_visits: state + .node_visits + .clone() + .into_iter() + .collect::>(), + diff, + }, + &scope, + ); // Emit GitCommit + GitPush events if git produced results if let Some(ref result) = git_result { if let Some(ref sha) = result.commit_sha { - self.emitter.emit(&Event::GitCommit { - node_id: Some(node.id().to_string()), - sha: sha.clone(), - }); + self.emitter.emit_scoped( + &Event::GitCommit { + node_id: Some(node.id().to_string()), + sha: sha.clone(), + }, + &scope, + ); } for (branch, success) in &result.push_results { self.emitter.emit(&Event::GitPush { diff --git a/lib/crates/fabro-workflow/src/lifecycle/git.rs b/lib/crates/fabro-workflow/src/lifecycle/git.rs index 1b5ba6b4c..f2d38ee6d 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/git.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/git.rs @@ -15,6 +15,7 @@ use crate::event::{Emitter, Event, RunNoticeLevel}; use crate::git::MetadataStore; use crate::graph::WorkflowGraph; use crate::graph::WorkflowNode; +use crate::lifecycle::event::stage_scope_for; use crate::outcome::{BilledModelUsage, Outcome, StageStatus}; use crate::run_dump::RunDump; use crate::run_options::RunOptions; @@ -283,10 +284,14 @@ impl RunLifecycle for GitLifecycle { } Err(e) => { // Emit CheckpointFailed and return error - self.emitter.emit(&Event::CheckpointFailed { - node_id: node_id.to_string(), - error: e.clone(), - }); + let scope = stage_scope_for(state, node_id); + self.emitter.emit_scoped( + &Event::CheckpointFailed { + node_id: node_id.to_string(), + error: e.clone(), + }, + &scope, + ); return Err(CoreError::Other(format!( "git checkpoint commit failed for node '{node_id}': {e}" ))); diff --git a/lib/crates/fabro-workflow/src/operations/create.rs b/lib/crates/fabro-workflow/src/operations/create.rs index f0886f405..fe22570be 100644 --- a/lib/crates/fabro-workflow/src/operations/create.rs +++ b/lib/crates/fabro-workflow/src/operations/create.rs @@ -219,6 +219,7 @@ async fn persist_created_run( manifest_blob, }, record.run_id.created_at(), + None, ); let payload = fabro_store::EventPayload::new( serde_json::to_value(&stored).map_err(|err| FabroError::engine(err.to_string()))?, diff --git a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs index 33c5bd1e8..914a9daf5 100644 --- a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs +++ b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs @@ -1197,9 +1197,6 @@ mod tests { node_id: "plan".to_string(), 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 7073e4cf7..1511c0ced 100644 --- a/lib/crates/fabro-workflow/src/pipeline/retro.rs +++ b/lib/crates/fabro-workflow/src/pipeline/retro.rs @@ -76,8 +76,6 @@ 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, }); } })