scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-11 22:53:00 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

refactor(events): thread stage scope through emitter

Browse source 
Populate stage_id / parallel_group_id / parallel_branch_id on every
event tied to a concrete stage execution, per the spec at
docs-internal/fabro-event-schema-v2-concrete-shape.md:223-279.

Before this commit, stored_event_fields() only set stage_id for the
four Event::Stage* variants and Event::Agent -- the only variants
that carried visit/parallel_group_id/parallel_branch_id in their
payload. Every other stage-scoped event (Checkpoint*, PromptCompleted,
Command*, AgentCli*, Prompt, Interview*, Failover, StallWatchdog,
GitCommit, ArtifactCaptured) fell through to node_stored_fields()
and left stage_id as None.

New approach: scope is carried alongside the event, not on the
variant.

- fabro-workflow/src/event.rs: new StageScope type
  { node_id, visit, parallel_group_id, parallel_branch_id }. New
  Emitter::emit_scoped(&event, &scope) for stage-level emission.
  to_run_event_at and stored_event_fields take an
  Option<&StageScope> that merges into the returned envelope
  fields. StageScope::for_handler(context, node_id) is the
  canonical handler-side constructor -- prefers
  context.current_stage_scope() set by the fidelity lifecycle,
  falls back to a scope synthesized from the node_id + context
  visit count for tests that don't go through the full lifecycle.
- fabro-workflow/src/context.rs: new
  WorkflowContext::current_stage_scope() method reads CURRENT_NODE,
  internal.node_visit_count, internal.parallel_group_id,
  internal.parallel_branch_id from the context.
- Remove the now-redundant visit/parallel_group_id/parallel_branch_id
  fields from Event::Stage{Started,Completed,Failed,Retrying} and
  the parallel_* fields from Event::Agent. These existed only to
  feed stored_event_fields() and are obsolete once scope is
  threaded through the emitter.

Emission site migration (all stage-scoped handlers now use
emit_scoped):

- lifecycle/event.rs: StageStarted, StageCompleted, StageFailed,
  StageRetrying, CheckpointCompleted, GitCommit (from on_checkpoint)
- lifecycle/git.rs: CheckpointFailed
- lifecycle/artifact.rs: ArtifactCaptured
- handler/command.rs: CommandStarted, CommandCompleted
- handler/prompt.rs: Prompt, PromptCompleted
- handler/agent.rs: Prompt, PromptCompleted
- handler/fan_in.rs: Prompt, PromptCompleted
- handler/human.rs: InterviewStarted, InterviewTimeout,
  InterviewInterrupted, InterviewCompleted
- handler/llm/api.rs: Failover, Agent (via spawn_event_forwarder
  which now carries a StageScope across the tokio::spawn boundary)
- handler/llm/cli.rs: AgentCliStarted, AgentCliCompleted
- handler/parallel.rs: ParallelBranchStarted, ParallelBranchCompleted

StallWatchdogTimeout stays on plain emit() because the watchdog
fires from an error path without a live stage context.

Deleted the local StageEventScope struct + current_stage_event_scope
helper from handler/llm/api.rs; it's generalized into StageScope.

Tests: two new unit tests in event.rs --
stage_scope_populates_stage_id_on_non_stage_events verifies
CommandStarted / Prompt / GitCommit all pick up stage_id from scope,
run_level_events_without_scope_leave_stage_id_absent confirms
run.* events still get no stage scope. Updated all test fixtures
across fabro-workflow, fabro-cli to drop the removed Event variant
fields. Accepted two insta snapshot updates in
fabro-cli/tests/it/cmd/{attach,run}.rs that now include the
formerly-missing stage_id fields on checkpoint and interview events.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-04-09 16:34:59 -04:00
parent 9b0b8d94fc
commit c6a78a4286
No known key found for this signature in database
23 changed files with 608 additions and 474 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
lib/crates/fabro-cli/src/commands/run/run_progress/event.rs
View file

@ -510,9 +510,6 @@ mod tests {
node_id: "plan".into(), node_id: "plan".into(),
name: "Plan".into(), name: "Plan".into(),
index: 0, index: 0,
visit: 1,
parallel_group_id: None,
parallel_branch_id: None,
duration_ms: 5000, duration_ms: 5000,
status: "success".into(), status: "success".into(),
preferred_label: None, preferred_label: None,
@ -557,8 +554,6 @@ mod tests {
}, },
session_id: None, session_id: None,
parent_session_id: None, parent_session_id: None,
parallel_group_id: None,
parallel_branch_id: None,
}; };
let stored = to_run_event(&fixtures::RUN_1, &event); let stored = to_run_event(&fixtures::RUN_1, &event);

20
lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs
View file

@ -479,8 +479,6 @@ mod tests {
event, event,
session_id: None, session_id: None,
parent_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(), node_id: node_id.into(),
name: name.into(), name: name.into(),
index: 0, index: 0,
visit: 1,
parallel_group_id: None,
parallel_branch_id: None,
handler_type: String::new(), handler_type: String::new(),
attempt: 1, attempt: 1,
max_attempts: 1, max_attempts: 1,
@ -515,9 +510,6 @@ mod tests {
node_id: node_id.into(), node_id: node_id.into(),
name: name.into(), name: name.into(),
index: 0, index: 0,
visit: 1,
parallel_group_id: None,
parallel_branch_id: None,
duration_ms: 5000, duration_ms: 5000,
status: "success".into(), status: "success".into(),
preferred_label: None, preferred_label: None,
@ -714,9 +706,6 @@ mod tests {
node_id: "code".into(), node_id: "code".into(),
name: "Code".into(), name: "Code".into(),
index: 0, index: 0,
visit: 1,
parallel_group_id: None,
parallel_branch_id: None,
attempt: 2, attempt: 2,
max_attempts: 3, max_attempts: 3,
delay_ms: 1500, delay_ms: 1500,
@ -961,9 +950,6 @@ mod tests {
node_id: "code".into(), node_id: "code".into(),
name: "Code".into(), name: "Code".into(),
index: 0, index: 0,
visit: 1,
parallel_group_id: None,
parallel_branch_id: None,
attempt: 2, attempt: 2,
max_attempts: 3, max_attempts: 3,
delay_ms: 1500, delay_ms: 1500,
@ -1168,14 +1154,12 @@ mod tests {
node_id: "code".into(), node_id: "code".into(),
name: "Code".into(), name: "Code".into(),
index: 0, index: 0,
visit: 1,
parallel_group_id: None,
parallel_branch_id: None,
handler_type: "agent".into(), handler_type: "agent".into(),
attempt: 1, attempt: 1,
max_attempts: 1, max_attempts: 1,
}, },
started_ts, started_ts,
None,
)) ))
.unwrap(); .unwrap();
let tool_started = serde_json::to_string(&to_run_event_at( let tool_started = serde_json::to_string(&to_run_event_at(
@ -1189,6 +1173,7 @@ mod tests {
}, },
), ),
started_ts, started_ts,
None,
)) ))
.unwrap(); .unwrap();
let tool_completed = serde_json::to_string(&to_run_event_at( let tool_completed = serde_json::to_string(&to_run_event_at(
@ -1203,6 +1188,7 @@ mod tests {
}, },
), ),
completed_ts, completed_ts,
None,
)) ))
.unwrap(); .unwrap();

3
lib/crates/fabro-cli/src/commands/store/dump.rs
View file

@ -599,9 +599,6 @@ mod tests {
node_id: "code".to_string(), node_id: "code".to_string(),
name: "Code".to_string(), name: "Code".to_string(),
index: 1, index: 1,
visit: 2,
parallel_group_id: None,
parallel_branch_id: None,
duration_ms: 250, duration_ms: 250,
status: "partial_success".to_string(), status: "partial_success".to_string(),
preferred_label: None, preferred_label: None,

2
lib/crates/fabro-cli/tests/it/cmd/attach.rs
View file

@ -726,6 +726,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
"status": "success" "status": "success"
}, },
"run_id": "[ULID]", "run_id": "[ULID]",
"stage_id": "start@1",
"ts": "[TIMESTAMP]" "ts": "[TIMESTAMP]"
}, },
{ {
@ -766,6 +767,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
"stage": "approve" "stage": "approve"
}, },
"run_id": "[ULID]", "run_id": "[ULID]",
"stage_id": "approve@1",
"ts": "[TIMESTAMP]" "ts": "[TIMESTAMP]"
} }
] ]

9
lib/crates/fabro-cli/tests/it/cmd/run.rs
View file

@ -1031,6 +1031,7 @@ fn json_run_implies_auto_approve_for_human_gates() {
"status": "success" "status": "success"
}, },
"run_id": "[ULID]", "run_id": "[ULID]",
"stage_id": "start@1",
"ts": "[TIMESTAMP]" "ts": "[TIMESTAMP]"
}, },
{ {
@ -1071,11 +1072,14 @@ fn json_run_implies_auto_approve_for_human_gates() {
"stage": "approve" "stage": "approve"
}, },
"run_id": "[ULID]", "run_id": "[ULID]",
"stage_id": "approve@1",
"ts": "[TIMESTAMP]" "ts": "[TIMESTAMP]"
}, },
{ {
"event": "interview.completed", "event": "interview.completed",
"id": "[EVENT_ID]", "id": "[EVENT_ID]",
"node_id": "approve",
"node_label": "approve",
"properties": { "properties": {
"answer": "A", "answer": "A",
"duration_ms": "[DURATION_MS]", "duration_ms": "[DURATION_MS]",
@ -1083,6 +1087,7 @@ fn json_run_implies_auto_approve_for_human_gates() {
"question_id": "[ULID]" "question_id": "[ULID]"
}, },
"run_id": "[ULID]", "run_id": "[ULID]",
"stage_id": "approve@1",
"ts": "[TIMESTAMP]" "ts": "[TIMESTAMP]"
}, },
{ {
@ -1199,6 +1204,7 @@ fn json_run_implies_auto_approve_for_human_gates() {
"status": "success" "status": "success"
}, },
"run_id": "[ULID]", "run_id": "[ULID]",
"stage_id": "approve@1",
"ts": "[TIMESTAMP]" "ts": "[TIMESTAMP]"
}, },
{ {
@ -1227,6 +1233,7 @@ fn json_run_implies_auto_approve_for_human_gates() {
"script": "echo shipped" "script": "echo shipped"
}, },
"run_id": "[ULID]", "run_id": "[ULID]",
"stage_id": "ship@1",
"ts": "[TIMESTAMP]" "ts": "[TIMESTAMP]"
}, },
{ {
@ -1242,6 +1249,7 @@ fn json_run_implies_auto_approve_for_human_gates() {
"timed_out": false "timed_out": false
}, },
"run_id": "[ULID]", "run_id": "[ULID]",
"stage_id": "ship@1",
"ts": "[TIMESTAMP]" "ts": "[TIMESTAMP]"
}, },
{ {
@ -1371,6 +1379,7 @@ fn json_run_implies_auto_approve_for_human_gates() {
"status": "success" "status": "success"
}, },
"run_id": "[ULID]", "run_id": "[ULID]",
"stage_id": "ship@1",
"ts": "[TIMESTAMP]" "ts": "[TIMESTAMP]"
}, },
{ {

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

@ -135,6 +135,8 @@ pub mod keys {
pub use fabro_core::Context; pub use fabro_core::Context;
use crate::event::StageScope;
use crate::run_dir::visit_from_context;
use fabro_graphviz::Fidelity; use fabro_graphviz::Fidelity;
use fabro_types::{ParallelBranchId, StageId}; use fabro_types::{ParallelBranchId, StageId};
@ -146,6 +148,10 @@ pub trait WorkflowContext {
fn run_id(&self) -> String; fn run_id(&self) -> String;
fn parallel_group_id(&self) -> Option<StageId>; fn parallel_group_id(&self) -> Option<StageId>;
fn parallel_branch_id(&self) -> Option<ParallelBranchId>; fn parallel_branch_id(&self) -> Option<ParallelBranchId>;
/// 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<StageScope>;
} }
impl WorkflowContext for Context { impl WorkflowContext for Context {
@ -177,6 +183,19 @@ impl WorkflowContext for Context {
self.get(keys::INTERNAL_PARALLEL_BRANCH_ID) self.get(keys::INTERNAL_PARALLEL_BRANCH_ID)
.and_then(|value| serde_json::from_value(value).ok()) .and_then(|value| serde_json::from_value(value).ok())
} }
fn current_stage_scope(&self) -> Option<StageScope> {
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)] #[cfg(test)]

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

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

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

@ -19,8 +19,10 @@ use tokio::io::{AsyncWrite, AsyncWriteExt};
use tokio::sync::{Mutex as AsyncMutex, mpsc, oneshot}; use tokio::sync::{Mutex as AsyncMutex, mpsc, oneshot};
use uuid::Uuid; use uuid::Uuid;
use crate::context::{Context as WfContext, WorkflowContext};
use crate::error::FabroError; use crate::error::FabroError;
use crate::outcome::{BilledModelUsage, FailureDetail, Outcome}; use crate::outcome::{BilledModelUsage, FailureDetail, Outcome};
use crate::run_dir::visit_from_context;
use fabro_agent::{AgentEvent, SandboxEvent, WorktreeEvent, WorktreeEventCallback}; use fabro_agent::{AgentEvent, SandboxEvent, WorktreeEvent, WorktreeEventCallback};
use fabro_llm::types::TokenCounts as LlmTokenCounts; use fabro_llm::types::TokenCounts as LlmTokenCounts;
use fabro_util::redact::redact_json_value; use fabro_util::redact::redact_json_value;
@ -138,11 +140,6 @@ pub enum Event {
node_id: String, node_id: String,
name: String, name: String,
index: usize, index: usize,
visit: u32,
#[serde(default, skip_serializing_if = "Option::is_none")]
parallel_group_id: Option<StageId>,
#[serde(default, skip_serializing_if = "Option::is_none")]
parallel_branch_id: Option<ParallelBranchId>,
handler_type: String, handler_type: String,
attempt: usize, attempt: usize,
max_attempts: usize, max_attempts: usize,
@ -151,11 +148,6 @@ pub enum Event {
node_id: String, node_id: String,
name: String, name: String,
index: usize, index: usize,
visit: u32,
#[serde(default, skip_serializing_if = "Option::is_none")]
parallel_group_id: Option<StageId>,
#[serde(default, skip_serializing_if = "Option::is_none")]
parallel_branch_id: Option<ParallelBranchId>,
duration_ms: u64, duration_ms: u64,
status: String, status: String,
preferred_label: Option<String>, preferred_label: Option<String>,
@ -186,11 +178,6 @@ pub enum Event {
node_id: String, node_id: String,
name: String, name: String,
index: usize, index: usize,
visit: u32,
#[serde(default, skip_serializing_if = "Option::is_none")]
parallel_group_id: Option<StageId>,
#[serde(default, skip_serializing_if = "Option::is_none")]
parallel_branch_id: Option<ParallelBranchId>,
failure: FailureDetail, failure: FailureDetail,
will_retry: bool, will_retry: bool,
}, },
@ -198,11 +185,6 @@ pub enum Event {
node_id: String, node_id: String,
name: String, name: String,
index: usize, index: usize,
visit: u32,
#[serde(default, skip_serializing_if = "Option::is_none")]
parallel_group_id: Option<StageId>,
#[serde(default, skip_serializing_if = "Option::is_none")]
parallel_branch_id: Option<ParallelBranchId>,
attempt: usize, attempt: usize,
max_attempts: usize, max_attempts: usize,
delay_ms: u64, delay_ms: u64,
@ -377,10 +359,6 @@ pub enum Event {
session_id: Option<String>, session_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
parent_session_id: Option<String>, parent_session_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
parallel_group_id: Option<StageId>,
#[serde(default, skip_serializing_if = "Option::is_none")]
parallel_branch_id: Option<ParallelBranchId>,
}, },
SubgraphStarted { SubgraphStarted {
node_id: String, 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") 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 { match event {
Event::RunCreated { provenance, .. } => StoredEventFields { Event::RunCreated { provenance, .. } => StoredEventFields {
actor: provenance.as_ref().and_then(actor_from_provenance), actor: provenance.as_ref().and_then(actor_from_provenance),
..StoredEventFields::default() ..StoredEventFields::default()
}, },
Event::StageCompleted { Event::StageCompleted { node_id, name, .. }
node_id, | Event::StageFailed { node_id, name, .. }
name, | Event::StageStarted { node_id, name, .. }
visit, | Event::StageRetrying { node_id, name, .. } => {
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_id_str = node_id.clone();
let node_label = default_node_label(Some(&node_id_str), Some(name.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 { StoredEventFields {
node_id: Some(node_id_str), node_id: Some(node_id_str),
node_label, node_label,
stage_id,
parallel_group_id: parallel_group_id.clone(),
parallel_branch_id: parallel_branch_id.clone(),
..StoredEventFields::default() ..StoredEventFields::default()
} }
} }
@ -1399,8 +1369,6 @@ fn stored_event_fields(event: &Event) -> StoredEventFields {
event: agent_event, event: agent_event,
session_id, session_id,
parent_session_id, parent_session_id,
parallel_group_id,
parallel_branch_id,
} => { } => {
let node_id = Some(stage.clone()); let node_id = Some(stage.clone());
let node_label = default_node_label(node_id.as_ref(), None); 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_id,
node_label, node_label,
stage_id, stage_id,
parallel_group_id: parallel_group_id.clone(),
parallel_branch_id: parallel_branch_id.clone(),
tool_call_id, tool_call_id,
actor, actor,
..StoredEventFields::default()
} }
} }
Event::GitCommit { node_id, .. } => node_stored_fields(node_id.clone()), 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 { /// Stage-level scope threaded through event emission to populate
to_run_event_at(run_id, event, Utc::now()) /// `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<StageId>,
pub parallel_branch_id: Option<ParallelBranchId>,
} }
pub fn to_run_event_at(run_id: &RunId, event: &Event, ts: chrono::DateTime<Utc>) -> RunEvent { impl StageScope {
let fields = stored_event_fields(event); /// 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<String>) -> 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<Utc>,
scope: Option<&StageScope>,
) -> RunEvent {
let fields = stored_event_fields(event, scope);
let body = event_body_from_event(event); let body = event_body_from_event(event);
RunEvent { RunEvent {
id: Uuid::now_v7().to_string(), id: Uuid::now_v7().to_string(),
@ -2759,6 +2758,14 @@ impl Emitter {
} }
pub fn emit(&self, event: &Event) { 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); self.last_event_at.store(epoch_millis(), Ordering::Relaxed);
event.trace(); event.trace();
if let Event::WorkflowRunStarted { run_id, .. } = event { if let Event::WorkflowRunStarted { run_id, .. } = event {
@ -2767,7 +2774,7 @@ impl Emitter {
"workflow run started event must match emitter run_id" "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); self.dispatch_run_event(&stored);
} }
@ -2858,15 +2865,12 @@ mod tests {
#[test] #[test]
fn run_event_stage_completed_places_node_fields_in_header() { fn run_event_stage_completed_places_node_fields_in_header() {
let stored = to_run_event( let stored = to_run_event_at(
&fixtures::RUN_2, &fixtures::RUN_2,
&Event::StageCompleted { &Event::StageCompleted {
node_id: "plan".to_string(), node_id: "plan".to_string(),
name: "Plan".to_string(), name: "Plan".to_string(),
index: 0, index: 0,
visit: 1,
parallel_group_id: None,
parallel_branch_id: None,
duration_ms: 5000, duration_ms: 5000,
status: "success".to_string(), status: "success".to_string(),
preferred_label: None, preferred_label: None,
@ -2885,6 +2889,13 @@ mod tests {
attempt: 1, attempt: 1,
max_attempts: 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"); assert_eq!(stored.event_name(), "stage.completed");
@ -2906,9 +2917,6 @@ mod tests {
node_id: "plan".to_string(), node_id: "plan".to_string(),
name: "Plan".to_string(), name: "Plan".to_string(),
index: 0, index: 0,
visit: 1,
parallel_group_id: None,
parallel_branch_id: None,
duration_ms: 5000, duration_ms: 5000,
status: "success".to_string(), status: "success".to_string(),
preferred_label: None, preferred_label: None,
@ -2943,9 +2951,6 @@ mod tests {
node_id: "code".to_string(), node_id: "code".to_string(),
name: "Code".to_string(), name: "Code".to_string(),
index: 1, index: 1,
visit: 1,
parallel_group_id: None,
parallel_branch_id: None,
failure: FailureDetail::new( failure: FailureDetail::new(
"lint failed", "lint failed",
crate::outcome::FailureCategory::Deterministic, crate::outcome::FailureCategory::Deterministic,
@ -2975,8 +2980,6 @@ mod tests {
}, },
session_id: Some("ses_child".to_string()), session_id: Some("ses_child".to_string()),
parent_session_id: Some("ses_parent".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, session_id: None,
parent_session_id: None, parent_session_id: None,
parallel_group_id: None,
parallel_branch_id: None,
}), }),
"agent.sub.spawned" "agent.sub.spawned"
); );
@ -3160,19 +3161,23 @@ mod tests {
#[test] #[test]
fn stage_started_populates_parallel_ids_when_present() { fn stage_started_populates_parallel_ids_when_present() {
let stored = to_run_event( let stored = to_run_event_at(
&fixtures::RUN_1, &fixtures::RUN_1,
&Event::StageStarted { &Event::StageStarted {
node_id: "review".to_string(), node_id: "review".to_string(),
name: "review".to_string(), name: "review".to_string(),
index: 1, 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(), handler_type: "agent".to_string(),
attempt: 1, attempt: 1,
max_attempts: 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!(stored.parallel_group_id, Some(StageId::new("fanout", 2)));
assert_eq!( assert_eq!(
@ -3216,7 +3221,7 @@ mod tests {
#[test] #[test]
fn agent_tool_started_populates_tool_call_id_and_stage_id() { 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, &fixtures::RUN_1,
&Event::Agent { &Event::Agent {
stage: "code".to_string(), stage: "code".to_string(),
@ -3228,9 +3233,14 @@ mod tests {
}, },
session_id: Some("ses_1".to_string()), session_id: Some("ses_1".to_string()),
parent_session_id: None, parent_session_id: None,
},
Utc::now(),
Some(&StageScope {
node_id: "code".to_string(),
visit: 3,
parallel_group_id: Some(StageId::new("fanout", 2)), parallel_group_id: Some(StageId::new("fanout", 2)),
parallel_branch_id: Some(ParallelBranchId::new(StageId::new("fanout", 2), 0)), parallel_branch_id: Some(ParallelBranchId::new(StageId::new("fanout", 2), 0)),
}, }),
); );
assert_eq!(stored.stage_id, Some(StageId::new("code", 3))); assert_eq!(stored.stage_id, Some(StageId::new("code", 3)));
assert_eq!(stored.tool_call_id.as_deref(), Some("call_abc")); 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] #[test]
fn agent_assistant_message_populates_agent_actor() { fn agent_assistant_message_populates_agent_actor() {
let stored = to_run_event( let stored = to_run_event(
@ -3256,8 +3330,6 @@ mod tests {
}, },
session_id: Some("ses_agent".to_string()), session_id: Some("ses_agent".to_string()),
parent_session_id: None, parent_session_id: None,
parallel_group_id: None,
parallel_branch_id: None,
}, },
); );
let actor = stored.actor.as_ref().expect("actor set"); let actor = stored.actor.as_ref().expect("actor set");

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

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

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

@ -10,7 +10,7 @@ use fabro_types::RunId;
use crate::context::keys; use crate::context::keys;
use crate::context::{Context, WorkflowContext}; use crate::context::{Context, WorkflowContext};
use crate::error::FabroError; use crate::error::FabroError;
use crate::event::{Emitter, Event}; use crate::event::{Emitter, Event, StageScope};
use crate::outcome::{ use crate::outcome::{
BilledModelUsage, FailureCategory, FailureDetail, Outcome, OutcomeExt, StageStatus, BilledModelUsage, FailureCategory, FailureDetail, Outcome, OutcomeExt, StageStatus,
}; };
@ -256,14 +256,18 @@ impl Handler for AgentHandler {
.map(String::from) .map(String::from)
.or_else(|| Some(Provider::default_from_env().as_str().to_string())); .or_else(|| Some(Provider::default_from_env().as_str().to_string()));
let prompt_model = node.model().map(String::from); let prompt_model = node.model().map(String::from);
services.emitter.emit(&Event::Prompt { let stage_scope = StageScope::for_handler(context, &node.id);
stage: node.id.clone(), services.emitter.emit_scoped(
visit, &Event::Prompt {
text: prompt.clone(), stage: node.id.clone(),
mode: Some("agent".to_string()), visit,
provider: prompt_provider, text: prompt.clone(),
model: prompt_model, mode: Some("agent".to_string()),
}); provider: prompt_provider,
model: prompt_model,
},
&stage_scope,
);
// 3. Call LLM backend (agent loop) // 3. Call LLM backend (agent loop)
let thread_id = context.thread_id(); let thread_id = context.thread_id();
@ -329,13 +333,16 @@ impl Handler for AgentHandler {
.map(String::from) .map(String::from)
.or_else(|| Some(Provider::default_from_env().as_str().to_string())) .or_else(|| Some(Provider::default_from_env().as_str().to_string()))
.unwrap_or_default(); .unwrap_or_default();
services.emitter.emit(&Event::PromptCompleted { services.emitter.emit_scoped(
node_id: node.id.clone(), &Event::PromptCompleted {
response: response_text.clone(), node_id: node.id.clone(),
model: response_model, response: response_text.clone(),
provider: response_provider, model: response_model,
billing: stage_usage.clone(), provider: response_provider,
}); billing: stage_usage.clone(),
},
&stage_scope,
);
// Build and write status // Build and write status
let mut outcome = Outcome::success(); let mut outcome = Outcome::success();
@ -709,19 +716,21 @@ mod tests {
_sandbox: &Arc<dyn fabro_agent::Sandbox>, _sandbox: &Arc<dyn fabro_agent::Sandbox>,
_tool_hooks: Option<Arc<dyn fabro_agent::ToolHookCallback>>, _tool_hooks: Option<Arc<dyn fabro_agent::ToolHookCallback>>,
) -> Result<CodergenResult, FabroError> { ) -> Result<CodergenResult, FabroError> {
emitter.emit(&crate::event::Event::Agent { let scope = StageScope::for_handler(context, &node.id);
stage: node.id.clone(), emitter.emit_scoped(
visit: u32::try_from(crate::run_dir::visit_from_context(context)) &crate::event::Event::Agent {
.unwrap_or(u32::MAX), stage: node.id.clone(),
event: fabro_agent::AgentEvent::SessionStarted { visit: u32::try_from(crate::run_dir::visit_from_context(context))
provider: Some("openai".to_string()), .unwrap_or(u32::MAX),
model: Some("gpt-5.4".to_string()), 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()), &scope,
parent_session_id: None, );
parallel_group_id: context.parallel_group_id(),
parallel_branch_id: context.parallel_branch_id(),
});
Ok(CodergenResult::Text { Ok(CodergenResult::Text {
text: "done".to_string(), text: "done".to_string(),
usage: None, usage: None,

40
lib/crates/fabro-workflow/src/handler/command.rs
View file

@ -4,6 +4,7 @@ use crate::context::Context;
use crate::context::keys; use crate::context::keys;
use crate::error::FabroError; use crate::error::FabroError;
use crate::event::Event; use crate::event::Event;
use crate::event::StageScope;
use crate::outcome::{Outcome, OutcomeExt}; use crate::outcome::{Outcome, OutcomeExt};
use async_trait::async_trait; use async_trait::async_trait;
use fabro_graphviz::graph::{Graph, Node}; use fabro_graphviz::graph::{Graph, Node};
@ -57,7 +58,7 @@ impl Handler for CommandHandler {
async fn execute( async fn execute(
&self, &self,
node: &Node, node: &Node,
_context: &Context, context: &Context,
_graph: &Graph, _graph: &Graph,
_run_dir: &Path, _run_dir: &Path,
services: &EngineServices, services: &EngineServices,
@ -90,13 +91,17 @@ impl Handler for CommandHandler {
} else { } else {
script.to_string() script.to_string()
}; };
services.emitter.emit(&Event::CommandStarted { let stage_scope = StageScope::for_handler(context, &node.id);
node_id: node.id.clone(), services.emitter.emit_scoped(
script: script.to_string(), &Event::CommandStarted {
command: command.clone(), node_id: node.id.clone(),
language: language.to_string(), script: script.to_string(),
timeout_ms: timeout_ms(node), command: command.clone(),
}); language: language.to_string(),
timeout_ms: timeout_ms(node),
},
&stage_scope,
);
let timeout_ms = node let timeout_ms = node
.timeout() .timeout()
@ -118,14 +123,17 @@ impl Handler for CommandHandler {
let result = let result =
result.map_err(|e| FabroError::handler(format!("Failed to spawn script: {e}")))?; result.map_err(|e| FabroError::handler(format!("Failed to spawn script: {e}")))?;
services.emitter.emit(&Event::CommandCompleted { services.emitter.emit_scoped(
node_id: node.id.clone(), &Event::CommandCompleted {
stdout: result.stdout.clone(), node_id: node.id.clone(),
stderr: result.stderr.clone(), stdout: result.stdout.clone(),
exit_code: (!result.timed_out).then_some(result.exit_code), stderr: result.stderr.clone(),
duration_ms: result.duration_ms, exit_code: (!result.timed_out).then_some(result.exit_code),
timed_out: result.timed_out, duration_ms: result.duration_ms,
}); timed_out: result.timed_out,
},
&stage_scope,
);
if result.timed_out { if result.timed_out {
return Err(FabroError::handler(format!( return Err(FabroError::handler(format!(

56
lib/crates/fabro-workflow/src/handler/fan_in.rs
View file

@ -4,7 +4,7 @@ use std::sync::Arc;
use crate::context::Context; use crate::context::Context;
use crate::context::keys; use crate::context::keys;
use crate::error::FabroError; use crate::error::FabroError;
use crate::event::{Emitter, Event}; use crate::event::{Emitter, Event, StageScope};
use crate::outcome::{Outcome, OutcomeExt}; use crate::outcome::{Outcome, OutcomeExt};
use crate::run_dir::visit_from_context; use crate::run_dir::visit_from_context;
use crate::sandbox_git::git_merge_ff_only; 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 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 { emitter.emit_scoped(
stage: node_id.to_string(), &Event::Prompt {
visit: visit_u32, stage: node_id.to_string(),
text: full_prompt.clone(), visit: visit_u32,
mode: Some("fan_in".to_string()), text: full_prompt.clone(),
provider: None, mode: Some("fan_in".to_string()),
model: None, provider: None,
}); model: None,
},
&stage_scope,
);
// Build a synthetic node for the backend call // Build a synthetic node for the backend call
let eval_node = Node::new("fan_in_eval"); let eval_node = Node::new("fan_in_eval");
@ -269,13 +273,16 @@ async fn llm_evaluate(
.unwrap_or_else(|| "unknown".to_string()); .unwrap_or_else(|| "unknown".to_string());
let response_text = let response_text =
serde_json::to_string_pretty(&outcome).unwrap_or_else(|_| "{}".to_string()); serde_json::to_string_pretty(&outcome).unwrap_or_else(|_| "{}".to_string());
emitter.emit(&Event::PromptCompleted { emitter.emit_scoped(
node_id: node_id.to_string(), &Event::PromptCompleted {
response: response_text.clone(), node_id: node_id.to_string(),
model: String::new(), response: response_text.clone(),
provider: String::new(), model: String::new(),
billing: None, provider: String::new(),
}); billing: None,
},
&stage_scope,
);
Ok(Candidate { Ok(Candidate {
id: best_id, id: best_id,
status: outcome.status.to_string(), status: outcome.status.to_string(),
@ -283,13 +290,16 @@ async fn llm_evaluate(
}) })
} }
Ok(CodergenResult::Text { text, .. }) => { Ok(CodergenResult::Text { text, .. }) => {
emitter.emit(&Event::PromptCompleted { emitter.emit_scoped(
node_id: node_id.to_string(), &Event::PromptCompleted {
response: text.clone(), node_id: node_id.to_string(),
model: String::new(), response: text.clone(),
provider: String::new(), model: String::new(),
billing: None, provider: String::new(),
}); billing: None,
},
&stage_scope,
);
// The LLM responded with text; try to find a matching candidate ID // The LLM responded with text; try to find a matching candidate ID
let text = text.trim().to_string(); let text = text.trim().to_string();

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

@ -8,7 +8,7 @@ use async_trait::async_trait;
use crate::context::Context; use crate::context::Context;
use crate::context::keys; use crate::context::keys;
use crate::error::FabroError; use crate::error::FabroError;
use crate::event::{Emitter, Event}; use crate::event::{Emitter, Event, StageScope};
use crate::millis_u64; use crate::millis_u64;
use crate::outcome::{Outcome, OutcomeExt}; use crate::outcome::{Outcome, OutcomeExt};
use fabro_graphviz::graph::{Graph, Node}; use fabro_graphviz::graph::{Graph, Node};
@ -88,10 +88,10 @@ impl HumanHandler {
self self
} }
fn emit(&self, default_emitter: &Arc<Emitter>, event: &Event) { fn emit(&self, default_emitter: &Arc<Emitter>, event: &Event, scope: &StageScope) {
match &self.emitter { match &self.emitter {
Some(emitter) => emitter.emit(event), Some(emitter) => emitter.emit_scoped(event, scope),
None => default_emitter.emit(event), None => default_emitter.emit_scoped(event, scope),
} }
} }
} }
@ -209,6 +209,7 @@ impl Handler for HumanHandler {
// 3. Present to interviewer // 3. Present to interviewer
let question_text = node.label().to_string(); let question_text = node.label().to_string();
let question_id = question.id.clone(); let question_id = question.id.clone();
let stage_scope = StageScope::for_handler(context, &node.id);
self.emit( self.emit(
&services.emitter, &services.emitter,
&Event::InterviewStarted { &Event::InterviewStarted {
@ -228,6 +229,7 @@ impl Handler for HumanHandler {
timeout_seconds: question.timeout_seconds, timeout_seconds: question.timeout_seconds,
context_display: question.context_display.clone(), context_display: question.context_display.clone(),
}, },
&stage_scope,
); );
let interview_start = Instant::now(); let interview_start = Instant::now();
let answer = self.interviewer.ask(question).await; let answer = self.interviewer.ask(question).await;
@ -242,6 +244,7 @@ impl Handler for HumanHandler {
stage: node.id.clone(), stage: node.id.clone(),
duration_ms: millis_u64(interview_start.elapsed()), duration_ms: millis_u64(interview_start.elapsed()),
}, },
&stage_scope,
); );
let default_choice = node let default_choice = node
.attrs .attrs
@ -279,6 +282,7 @@ impl Handler for HumanHandler {
reason: "interrupted".to_string(), reason: "interrupted".to_string(),
duration_ms: millis_u64(interview_start.elapsed()), duration_ms: millis_u64(interview_start.elapsed()),
}, },
&stage_scope,
); );
return Ok(unanswered_human_gate( return Ok(unanswered_human_gate(
"human interaction interrupted before an answer was provided", "human interaction interrupted before an answer was provided",
@ -293,6 +297,7 @@ impl Handler for HumanHandler {
answer: answer_text(&answer), answer: answer_text(&answer),
duration_ms: millis_u64(interview_start.elapsed()), duration_ms: millis_u64(interview_start.elapsed()),
}, },
&stage_scope,
); );
return Ok(unanswered_human_gate("human skipped interaction")); return Ok(unanswered_human_gate("human skipped interaction"));
} }
@ -306,6 +311,7 @@ impl Handler for HumanHandler {
answer: answer_text(&answer), answer: answer_text(&answer),
duration_ms: millis_u64(interview_start.elapsed()), duration_ms: millis_u64(interview_start.elapsed()),
}, },
&stage_scope,
); );
// 6. Try fixed-choice match // 6. Try fixed-choice match

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

@ -3,6 +3,7 @@ use std::sync::{Arc, Mutex};
use async_trait::async_trait; use async_trait::async_trait;
use crate::event::StageScope;
use fabro_agent::{ use fabro_agent::{
AgentEvent, AgentProfile, AnthropicProfile, GeminiProfile, OpenAiProfile, Sandbox, Session, AgentEvent, AgentProfile, AnthropicProfile, GeminiProfile, OpenAiProfile, Sandbox, Session,
SessionOptions, Turn, SessionOptions, Turn,
@ -13,7 +14,6 @@ use fabro_llm::types::{Message, Request, TokenCounts};
use fabro_mcp::config::McpServerSettings; use fabro_mcp::config::McpServerSettings;
use fabro_model::FallbackTarget; use fabro_model::FallbackTarget;
use fabro_model::Provider; use fabro_model::Provider;
use fabro_types::{ParallelBranchId, StageId};
use tokio::sync::Mutex as TokioMutex; use tokio::sync::Mutex as TokioMutex;
use super::super::agent::{CodergenBackend, CodergenResult}; use super::super::agent::{CodergenBackend, CodergenResult};
@ -22,7 +22,6 @@ use crate::context::{Context, WorkflowContext};
use crate::error::FabroError; use crate::error::FabroError;
use crate::event::{Emitter, Event}; use crate::event::{Emitter, Event};
use crate::outcome::billed_model_usage_from_llm; use crate::outcome::billed_model_usage_from_llm;
use crate::run_dir::visit_from_context;
use fabro_graphviz::graph::Node; use fabro_graphviz::graph::Node;
fn build_profile(model: &str, provider: Provider) -> Box<dyn AgentProfile> { fn build_profile(model: &str, provider: Provider) -> Box<dyn AgentProfile> {
@ -38,21 +37,6 @@ fn build_profile(model: &str, provider: Provider) -> Box<dyn AgentProfile> {
} }
} }
#[derive(Clone)]
struct StageEventScope {
visit: u32,
parallel_group_id: Option<StageId>,
parallel_branch_id: Option<ParallelBranchId>,
}
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. /// Shared state for tracking file modifications from agent tool calls.
struct FileTracking { struct FileTracking {
/// Maps tool_call_id → file_path for in-flight write/edit calls. /// 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( fn spawn_event_forwarder(
session: &Session, session: &Session,
node_id: String, node_id: String,
scope: StageEventScope, scope: StageScope,
emitter: Arc<Emitter>, emitter: Arc<Emitter>,
file_tracking: Arc<Mutex<FileTracking>>, file_tracking: Arc<Mutex<FileTracking>>,
) { ) {
@ -115,15 +99,16 @@ fn spawn_event_forwarder(
if !event.event.is_streaming_noise() if !event.event.is_streaming_noise()
&& !matches!(&event.event, AgentEvent::ProcessingEnd) && !matches!(&event.event, AgentEvent::ProcessingEnd)
{ {
emitter.emit(&Event::Agent { emitter.emit_scoped(
stage: node_id.clone(), &Event::Agent {
visit: scope.visit, stage: node_id.clone(),
event: event.event.clone(), visit: scope.visit,
session_id: Some(event.session_id.clone()), event: event.event.clone(),
parent_session_id: event.parent_session_id.clone(), session_id: Some(event.session_id.clone()),
parallel_group_id: scope.parallel_group_id.clone(), parent_session_id: event.parent_session_id.clone(),
parallel_branch_id: scope.parallel_branch_id.clone(), },
}); &scope,
);
} }
} }
}); });
@ -469,7 +454,7 @@ impl CodergenBackend for AgentApiBackend {
touched: HashSet::new(), touched: HashSet::new(),
last: None, 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. // Subscribe to session events: forward to pipeline emitter + track files.
spawn_event_forwarder( spawn_event_forwarder(
@ -503,14 +488,17 @@ impl CodergenBackend for AgentApiBackend {
let mut succeeded = false; let mut succeeded = false;
for target in &self.fallback_chain { for target in &self.fallback_chain {
emitter.emit(&Event::Failover { emitter.emit_scoped(
stage: node.id.clone(), &Event::Failover {
from_provider: from_provider.clone(), stage: node.id.clone(),
from_model: from_model.clone(), from_provider: from_provider.clone(),
to_provider: target.provider.clone(), from_model: from_model.clone(),
to_model: target.model.clone(), to_provider: target.provider.clone(),
error: error_msg.clone(), to_model: target.model.clone(),
}); error: error_msg.clone(),
},
&event_scope,
);
let target_provider: Provider = match target.provider.parse() { let target_provider: Provider = match target.provider.parse() {
Ok(p) => p, Ok(p) => p,

39
lib/crates/fabro-workflow/src/handler/llm/cli.rs
View file

@ -10,7 +10,7 @@ use tokio::time::sleep;
use super::super::agent::{CodergenBackend, CodergenResult}; use super::super::agent::{CodergenBackend, CodergenResult};
use crate::context::Context; use crate::context::Context;
use crate::error::FabroError; 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::outcome::billed_model_usage_from_llm;
use crate::run_dir::visit_from_context; use crate::run_dir::visit_from_context;
use fabro_graphviz::graph::Node; use fabro_graphviz::graph::Node;
@ -496,14 +496,18 @@ impl CodergenBackend for AgentCliBackend {
ensure_cli(cli, provider, sandbox, emitter).await?; ensure_cli(cli, provider, sandbox, emitter).await?;
let command = cli_command_for_provider(provider, model, &prompt_path); let command = cli_command_for_provider(provider, model, &prompt_path);
emitter.emit(&Event::AgentCliStarted { let stage_scope = StageScope::for_handler(_context, &node.id);
node_id: node.id.clone(), emitter.emit_scoped(
visit: current_visit(_context), &Event::AgentCliStarted {
mode: "cli".to_string(), node_id: node.id.clone(),
provider: provider.as_str().to_string(), visit: current_visit(_context),
model: model.to_string(), mode: "cli".to_string(),
command: command.clone(), 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. // 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 // Build a HashMap to pass via exec_command's env_vars parameter — this
@ -620,13 +624,16 @@ impl CodergenBackend for AgentCliBackend {
timed_out: false, timed_out: false,
duration_ms, duration_ms,
}; };
emitter.emit(&Event::AgentCliCompleted { emitter.emit_scoped(
node_id: node.id.clone(), &Event::AgentCliCompleted {
stdout: result.stdout.clone(), node_id: node.id.clone(),
stderr: result.stderr.clone(), stdout: result.stdout.clone(),
exit_code: result.exit_code, stderr: result.stderr.clone(),
duration_ms: result.duration_ms, exit_code: result.exit_code,
}); duration_ms: result.duration_ms,
},
&stage_scope,
);
// 3e. Cleanup temp files // 3e. Cleanup temp files
let _ = sandbox let _ = sandbox

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

@ -10,7 +10,7 @@ use tokio::sync::Semaphore;
use crate::context::keys; use crate::context::keys;
use crate::context::{Context, WorkflowContext}; use crate::context::{Context, WorkflowContext};
use crate::error::FabroError; use crate::error::FabroError;
use crate::event::Event; use crate::event::{Event, StageScope};
use crate::git::sanitize_ref_component; use crate::git::sanitize_ref_component;
use crate::hook_context::set_hook_node; use crate::hook_context::set_hook_node;
use crate::millis_u64; 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 --- // --- Fan out: concurrent execution ---
let mut handles = Vec::new(); let mut handles = Vec::new();
for setup in branch_setups { for setup in branch_setups {
@ -302,6 +304,7 @@ impl Handler for ParallelHandler {
.map(|gs| gs.git_author.clone()) .map(|gs| gs.git_author.clone())
.unwrap_or_default(); .unwrap_or_default();
let group_id = parallel_group_id.clone(); let group_id = parallel_group_id.clone();
let branch_scope = parent_scope.clone();
let handle = tokio::spawn(async move { let handle = tokio::spawn(async move {
let _permit = sem let _permit = sem
@ -309,12 +312,15 @@ impl Handler for ParallelHandler {
.await .await
.map_err(|e| FabroError::handler(format!("semaphore error: {e}")))?; .map_err(|e| FabroError::handler(format!("semaphore error: {e}")))?;
emitter.emit(&Event::ParallelBranchStarted { emitter.emit_scoped(
parallel_group_id: group_id.clone(), &Event::ParallelBranchStarted {
parallel_branch_id: setup.parallel_branch_id.clone(), parallel_group_id: group_id.clone(),
branch: setup.target_id.clone(), parallel_branch_id: setup.parallel_branch_id.clone(),
index: setup.branch_index, branch: setup.target_id.clone(),
}); index: setup.branch_index,
},
&branch_scope,
);
let branch_start = Instant::now(); let branch_start = Instant::now();
let Some(target_node) = graph.nodes.get(&setup.target_id) else { let Some(target_node) = graph.nodes.get(&setup.target_id) else {
@ -322,15 +328,18 @@ impl Handler for ParallelHandler {
"branch target node not found: {}", "branch target node not found: {}",
setup.target_id setup.target_id
)); ));
emitter.emit(&Event::ParallelBranchCompleted { emitter.emit_scoped(
parallel_group_id: group_id.clone(), &Event::ParallelBranchCompleted {
parallel_branch_id: setup.parallel_branch_id.clone(), parallel_group_id: group_id.clone(),
branch: setup.target_id.clone(), parallel_branch_id: setup.parallel_branch_id.clone(),
index: setup.branch_index, branch: setup.target_id.clone(),
duration_ms: millis_u64(branch_start.elapsed()), index: setup.branch_index,
status: "fail".to_string(), duration_ms: millis_u64(branch_start.elapsed()),
head_sha: None, status: "fail".to_string(),
}); head_sha: None,
},
&branch_scope,
);
return Ok(BranchResult { return Ok(BranchResult {
id: setup.target_id.clone(), id: setup.target_id.clone(),
outcome, outcome,
@ -408,15 +417,18 @@ impl Handler for ParallelHandler {
None None
}; };
emitter.emit(&Event::ParallelBranchCompleted { emitter.emit_scoped(
parallel_group_id: group_id.clone(), &Event::ParallelBranchCompleted {
parallel_branch_id: setup.parallel_branch_id.clone(), parallel_group_id: group_id.clone(),
branch: setup.target_id.clone(), parallel_branch_id: setup.parallel_branch_id.clone(),
index: setup.branch_index, branch: setup.target_id.clone(),
duration_ms: millis_u64(branch_start.elapsed()), index: setup.branch_index,
status: outcome.status.to_string(), duration_ms: millis_u64(branch_start.elapsed()),
head_sha: head_sha.clone(), status: outcome.status.to_string(),
}); head_sha: head_sha.clone(),
},
&branch_scope,
);
Ok::<BranchResult, FabroError>(BranchResult { Ok::<BranchResult, FabroError>(BranchResult {
id: setup.target_id, id: setup.target_id,

39
lib/crates/fabro-workflow/src/handler/prompt.rs
View file

@ -3,7 +3,7 @@ use std::path::Path;
use crate::context::keys; use crate::context::keys;
use crate::context::{Context, WorkflowContext}; use crate::context::{Context, WorkflowContext};
use crate::error::FabroError; use crate::error::FabroError;
use crate::event::Event; use crate::event::{Event, StageScope};
use crate::outcome::Outcome; use crate::outcome::Outcome;
use crate::run_dir::visit_from_context; use crate::run_dir::visit_from_context;
use async_trait::async_trait; use async_trait::async_trait;
@ -91,14 +91,18 @@ impl Handler for PromptHandler {
.map(String::from) .map(String::from)
.or_else(|| Some(Provider::default_from_env().as_str().to_string())); .or_else(|| Some(Provider::default_from_env().as_str().to_string()));
let prompt_model = node.model().map(String::from); let prompt_model = node.model().map(String::from);
services.emitter.emit(&Event::Prompt { let stage_scope = StageScope::for_handler(context, &node.id);
stage: node.id.clone(), services.emitter.emit_scoped(
visit: u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX), &Event::Prompt {
text: prompt.clone(), stage: node.id.clone(),
mode: Some("prompt".to_string()), visit: u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX),
provider: prompt_provider.clone(), text: prompt.clone(),
model: prompt_model.clone(), mode: Some("prompt".to_string()),
}); provider: prompt_provider.clone(),
model: prompt_model.clone(),
},
&stage_scope,
);
// 3. Call LLM backend (one_shot) // 3. Call LLM backend (one_shot)
let (response_text, stage_usage, backend_files_touched) = 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())) .or_else(|| Some(Provider::default_from_env().as_str().to_string()))
.unwrap_or_default(); .unwrap_or_default();
services.emitter.emit(&Event::PromptCompleted { services.emitter.emit_scoped(
node_id: node.id.clone(), &Event::PromptCompleted {
response: response_text.clone(), node_id: node.id.clone(),
model: response_model, response: response_text.clone(),
provider: response_provider, model: response_model,
billing: stage_usage.clone(), provider: response_provider,
}); billing: stage_usage.clone(),
},
&stage_scope,
);
// 4. Build and write status // 4. Build and write status
let mut outcome = Outcome::success(); let mut outcome = Outcome::success();

25
lib/crates/fabro-workflow/src/lifecycle/artifact.rs
View file

@ -19,6 +19,7 @@ use crate::artifact_upload::ArtifactSink;
use crate::event::{Emitter, Event, RunNoticeLevel}; use crate::event::{Emitter, Event, RunNoticeLevel};
use crate::graph::WorkflowGraph; use crate::graph::WorkflowGraph;
use crate::graph::WorkflowNode; use crate::graph::WorkflowNode;
use crate::lifecycle::event::stage_scope_for;
use crate::outcome::BilledModelUsage; use crate::outcome::BilledModelUsage;
use crate::runtime_store::RunStoreHandle; use crate::runtime_store::RunStoreHandle;
use fabro_core::lifecycle::NodeDecision; use fabro_core::lifecycle::NodeDecision;
@ -136,18 +137,22 @@ impl RunLifecycle<WorkflowGraph> for ArtifactLifecycle {
}); });
return Ok(()); return Ok(());
} }
let scope = stage_scope_for(state, node_id);
for asset in &summary.captured_assets { for asset in &summary.captured_assets {
self.captured_artifact_count.fetch_add(1, Ordering::Relaxed); self.captured_artifact_count.fetch_add(1, Ordering::Relaxed);
self.emitter.emit(&Event::ArtifactCaptured { self.emitter.emit_scoped(
node_id: node_id.to_string(), &Event::ArtifactCaptured {
attempt: ctx.attempt, node_id: node_id.to_string(),
node_slug: node_slug.clone(), attempt: ctx.attempt,
path: asset.path.clone(), node_slug: node_slug.clone(),
mime: asset.mime.clone(), path: asset.path.clone(),
content_md5: asset.content_md5.clone(), mime: asset.mime.clone(),
content_sha256: asset.content_sha256.clone(), content_md5: asset.content_md5.clone(),
bytes: asset.bytes, content_sha256: asset.content_sha256.clone(),
}); bytes: asset.bytes,
},
&scope,
);
} }
} }
Ok(_) => {} // no files collected Ok(_) => {} // no files collected

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

@ -19,11 +19,11 @@ use crate::artifact;
use crate::context; use crate::context;
use crate::context::WorkflowContext; use crate::context::WorkflowContext;
use crate::error::FabroError; use crate::error::FabroError;
use crate::event::{Emitter, Event}; use crate::event::{Emitter, Event, StageScope};
use crate::graph::WorkflowGraph; use crate::graph::WorkflowGraph;
use crate::graph::WorkflowNode; use crate::graph::WorkflowNode;
use crate::outcome::{BilledModelUsage, FailureCategory, FailureDetail, Outcome, StageStatus}; 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<Option<BilledModelUsage>>; type WfRunState = ExecutionState<Option<BilledModelUsage>>;
type WfNodeResult = NodeResult<Option<BilledModelUsage>>; type WfNodeResult = NodeResult<Option<BilledModelUsage>>;
@ -85,11 +85,13 @@ fn stage_visit(state: &WfRunState, node_id: &str) -> u32 {
u32::try_from(visits.max(1)).unwrap_or(u32::MAX) u32::try_from(visits.max(1)).unwrap_or(u32::MAX)
} }
fn stage_parallel_ids(state: &WfRunState) -> (Option<StageId>, Option<ParallelBranchId>) { pub(crate) fn stage_scope_for(state: &WfRunState, node_id: &str) -> StageScope {
( StageScope {
state.context.parallel_group_id(), node_id: node_id.to_string(),
state.context.parallel_branch_id(), visit: stage_visit(state, node_id),
) parallel_group_id: state.context.parallel_group_id(),
parallel_branch_id: state.context.parallel_branch_id(),
}
} }
#[async_trait] #[async_trait]
@ -133,49 +135,48 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
} }
let gv = node.inner(); let gv = node.inner();
let stage_index = state.stage_index; let stage_index = state.stage_index;
let visit = stage_visit(state, &gv.id); let scope = stage_scope_for(state, &gv.id);
let (parallel_group_id, parallel_branch_id) = stage_parallel_ids(state);
let (loop_failure_signatures, restart_failure_signatures) = let (loop_failure_signatures, restart_failure_signatures) =
snapshot_failure_signatures(&self.circuit_breaker); snapshot_failure_signatures(&self.circuit_breaker);
self.emitter.emit(&Event::StageStarted { self.emitter.emit_scoped(
node_id: gv.id.clone(), &Event::StageStarted {
name: gv.label().to_string(), node_id: gv.id.clone(),
index: stage_index, name: gv.label().to_string(),
visit, index: stage_index,
parallel_group_id: parallel_group_id.clone(), handler_type: gv.handler_type().unwrap_or_default().to_string(),
parallel_branch_id: parallel_branch_id.clone(), attempt: 1,
handler_type: gv.handler_type().unwrap_or_default().to_string(), max_attempts: 1,
attempt: 1, },
max_attempts: 1, &scope,
}); );
self.emitter.emit(&Event::StageCompleted { self.emitter.emit_scoped(
node_id: gv.id.clone(), &Event::StageCompleted {
name: gv.label().to_string(), node_id: gv.id.clone(),
index: stage_index, name: gv.label().to_string(),
visit, index: stage_index,
parallel_group_id, duration_ms: 0,
parallel_branch_id, status: StageStatus::Success.to_string(),
duration_ms: 0, preferred_label: None,
status: StageStatus::Success.to_string(), suggested_next_ids: Vec::new(),
preferred_label: None, billing: None,
suggested_next_ids: Vec::new(), failure: None,
billing: None, notes: None,
failure: None, files_touched: Vec::new(),
notes: None, context_updates: None,
files_touched: Vec::new(), jump_to_node: None,
context_updates: None, context_values: None,
jump_to_node: None, node_visits: None,
context_values: None, loop_failure_signatures,
node_visits: None, restart_failure_signatures,
loop_failure_signatures, response: state
restart_failure_signatures, .context
response: state .get(&context::keys::response_key(&gv.id))
.context .and_then(|value| value.as_str().map(ToOwned::to_owned)),
.get(&context::keys::response_key(&gv.id)) attempt: 1,
.and_then(|value| value.as_str().map(ToOwned::to_owned)), max_attempts: 1,
attempt: 1, },
max_attempts: 1, &scope,
}); );
} }
async fn before_attempt( async fn before_attempt(
@ -184,18 +185,18 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
state: &WfRunState, state: &WfRunState,
) -> CoreResult<NodeDecision<Option<BilledModelUsage>>> { ) -> CoreResult<NodeDecision<Option<BilledModelUsage>>> {
let gv = ctx.node.inner(); let gv = ctx.node.inner();
let (parallel_group_id, parallel_branch_id) = stage_parallel_ids(state); let scope = stage_scope_for(state, &gv.id);
self.emitter.emit(&Event::StageStarted { self.emitter.emit_scoped(
node_id: gv.id.clone(), &Event::StageStarted {
name: gv.label().to_string(), node_id: gv.id.clone(),
index: state.stage_index, name: gv.label().to_string(),
visit: stage_visit(state, &gv.id), index: state.stage_index,
parallel_group_id, handler_type: gv.handler_type().unwrap_or_default().to_string(),
parallel_branch_id, attempt: ctx.attempt as usize,
handler_type: gv.handler_type().unwrap_or_default().to_string(), max_attempts: ctx.max_attempts as usize,
attempt: ctx.attempt as usize, },
max_attempts: ctx.max_attempts as usize, &scope,
}); );
Ok(NodeDecision::Continue) Ok(NodeDecision::Continue)
} }
@ -208,35 +209,34 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
let gv = ctx.node.inner(); let gv = ctx.node.inner();
let outcome = &ctx.result.outcome; let outcome = &ctx.result.outcome;
let stage_index = state.stage_index; let stage_index = state.stage_index;
let visit = stage_visit(state, &gv.id); let scope = stage_scope_for(state, &gv.id);
let (parallel_group_id, parallel_branch_id) = stage_parallel_ids(state);
self.emitter.emit(&Event::StageFailed { self.emitter.emit_scoped(
node_id: gv.id.clone(), &Event::StageFailed {
name: gv.label().to_string(), node_id: gv.id.clone(),
index: stage_index, name: gv.label().to_string(),
visit, index: stage_index,
parallel_group_id: parallel_group_id.clone(), failure: outcome.failure.clone().unwrap_or_else(|| {
parallel_branch_id: parallel_branch_id.clone(), FailureDetail::new("handler failed", FailureCategory::TransientInfra)
failure: outcome.failure.clone().unwrap_or_else(|| { }),
FailureDetail::new("handler failed", FailureCategory::TransientInfra) will_retry: true,
}), },
will_retry: true, &scope,
}); );
self.emitter.emit(&Event::StageRetrying { self.emitter.emit_scoped(
node_id: gv.id.clone(), &Event::StageRetrying {
name: gv.label().to_string(), node_id: gv.id.clone(),
index: stage_index, name: gv.label().to_string(),
visit, index: stage_index,
parallel_group_id, attempt: ctx.attempt as usize,
parallel_branch_id, max_attempts: ctx.result.max_attempts as usize,
attempt: ctx.attempt as usize, delay_ms: ctx
max_attempts: ctx.result.max_attempts as usize, .backoff_delay
delay_ms: ctx .map_or(0, |d| u64::try_from(d.as_millis()).unwrap()),
.backoff_delay },
.map_or(0, |d| u64::try_from(d.as_millis()).unwrap()), &scope,
}); );
} }
Ok(()) Ok(())
} }
@ -254,66 +254,66 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
} }
let gv = node.inner(); let gv = node.inner();
let stage_index = state.stage_index; let stage_index = state.stage_index;
let visit = stage_visit(state, &gv.id); let scope = stage_scope_for(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 duration_ms = u64::try_from(result.duration.as_millis()).unwrap();
let (loop_failure_signatures, restart_failure_signatures) = let (loop_failure_signatures, restart_failure_signatures) =
snapshot_failure_signatures(&self.circuit_breaker); snapshot_failure_signatures(&self.circuit_breaker);
if outcome.status == StageStatus::Fail { if outcome.status == StageStatus::Fail {
self.emitter.emit(&Event::StageFailed { self.emitter.emit_scoped(
node_id: gv.id.clone(), &Event::StageFailed {
name: gv.label().to_string(), node_id: gv.id.clone(),
index: stage_index, name: gv.label().to_string(),
visit, index: stage_index,
parallel_group_id, failure: outcome.failure.clone().unwrap_or_else(|| {
parallel_branch_id, FailureDetail::new("handler failed", FailureCategory::Deterministic)
failure: outcome.failure.clone().unwrap_or_else(|| { }),
FailureDetail::new("handler failed", FailureCategory::Deterministic) will_retry: false,
}),
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::<BTreeMap<_, _>>()
}),
jump_to_node: outcome.jump_to_node.clone(),
context_values: {
let snapshot = state.context.snapshot();
(!snapshot.is_empty()).then(|| snapshot.into_iter().collect::<BTreeMap<_, _>>())
}, },
node_visits: (!state.node_visits.is_empty()).then(|| { &scope,
state );
.node_visits } else {
.clone() self.emitter.emit_scoped(
.into_iter() &Event::StageCompleted {
.collect::<BTreeMap<_, _>>() node_id: gv.id.clone(),
}), name: gv.label().to_string(),
loop_failure_signatures, index: stage_index,
restart_failure_signatures, duration_ms,
response: response_from_outcome(&gv.id, outcome), status: outcome.status.to_string(),
attempt: result.attempts as usize, preferred_label: outcome.preferred_label.clone(),
max_attempts: result.max_attempts as usize, 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::<BTreeMap<_, _>>()
}),
jump_to_node: outcome.jump_to_node.clone(),
context_values: {
let snapshot = state.context.snapshot();
(!snapshot.is_empty())
.then(|| snapshot.into_iter().collect::<BTreeMap<_, _>>())
},
node_visits: (!state.node_visits.is_empty()).then(|| {
state
.node_visits
.clone()
.into_iter()
.collect::<BTreeMap<_, _>>()
}),
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(()) Ok(())
} }
@ -367,37 +367,44 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
node_outcomes.insert(node.id().to_string(), result.outcome.clone()); node_outcomes.insert(node.id().to_string(), result.outcome.clone());
artifact::normalize_durable_outcomes(&mut node_outcomes); artifact::normalize_durable_outcomes(&mut node_outcomes);
self.emitter.emit(&Event::CheckpointCompleted { let scope = stage_scope_for(state, node.id());
node_id: node.id().to_string(), self.emitter.emit_scoped(
status, &Event::CheckpointCompleted {
current_node: node.id().to_string(), node_id: node.id().to_string(),
completed_nodes: state.completed_nodes.clone(), status,
node_retries: state current_node: node.id().to_string(),
.node_retries completed_nodes: state.completed_nodes.clone(),
.clone() node_retries: state
.into_iter() .node_retries
.collect::<BTreeMap<_, _>>(), .clone()
context_values: context_values.into_iter().collect::<BTreeMap<_, _>>(), .into_iter()
node_outcomes: node_outcomes.into_iter().collect::<BTreeMap<_, _>>(), .collect::<BTreeMap<_, _>>(),
next_node_id: next_node_id.map(ToOwned::to_owned), context_values: context_values.into_iter().collect::<BTreeMap<_, _>>(),
git_commit_sha: git_sha.clone(), node_outcomes: node_outcomes.into_iter().collect::<BTreeMap<_, _>>(),
loop_failure_signatures: loop_failure_signatures.unwrap_or_default(), next_node_id: next_node_id.map(ToOwned::to_owned),
restart_failure_signatures: restart_failure_signatures.unwrap_or_default(), git_commit_sha: git_sha.clone(),
node_visits: state loop_failure_signatures: loop_failure_signatures.unwrap_or_default(),
.node_visits restart_failure_signatures: restart_failure_signatures.unwrap_or_default(),
.clone() node_visits: state
.into_iter() .node_visits
.collect::<BTreeMap<_, _>>(), .clone()
diff, .into_iter()
}); .collect::<BTreeMap<_, _>>(),
diff,
},
&scope,
);
// Emit GitCommit + GitPush events if git produced results // Emit GitCommit + GitPush events if git produced results
if let Some(ref result) = git_result { if let Some(ref result) = git_result {
if let Some(ref sha) = result.commit_sha { if let Some(ref sha) = result.commit_sha {
self.emitter.emit(&Event::GitCommit { self.emitter.emit_scoped(
node_id: Some(node.id().to_string()), &Event::GitCommit {
sha: sha.clone(), node_id: Some(node.id().to_string()),
}); sha: sha.clone(),
},
&scope,
);
} }
for (branch, success) in &result.push_results { for (branch, success) in &result.push_results {
self.emitter.emit(&Event::GitPush { self.emitter.emit(&Event::GitPush {

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

@ -15,6 +15,7 @@ use crate::event::{Emitter, Event, RunNoticeLevel};
use crate::git::MetadataStore; use crate::git::MetadataStore;
use crate::graph::WorkflowGraph; use crate::graph::WorkflowGraph;
use crate::graph::WorkflowNode; use crate::graph::WorkflowNode;
use crate::lifecycle::event::stage_scope_for;
use crate::outcome::{BilledModelUsage, Outcome, StageStatus}; use crate::outcome::{BilledModelUsage, Outcome, StageStatus};
use crate::run_dump::RunDump; use crate::run_dump::RunDump;
use crate::run_options::RunOptions; use crate::run_options::RunOptions;
@ -283,10 +284,14 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
} }
Err(e) => { Err(e) => {
// Emit CheckpointFailed and return error // Emit CheckpointFailed and return error
self.emitter.emit(&Event::CheckpointFailed { let scope = stage_scope_for(state, node_id);
node_id: node_id.to_string(), self.emitter.emit_scoped(
error: e.clone(), &Event::CheckpointFailed {
}); node_id: node_id.to_string(),
error: e.clone(),
},
&scope,
);
return Err(CoreError::Other(format!( return Err(CoreError::Other(format!(
"git checkpoint commit failed for node '{node_id}': {e}" "git checkpoint commit failed for node '{node_id}': {e}"
))); )));

1
lib/crates/fabro-workflow/src/operations/create.rs
View file

@ -219,6 +219,7 @@ async fn persist_created_run(
manifest_blob, manifest_blob,
}, },
record.run_id.created_at(), record.run_id.created_at(),
None,
); );
let payload = fabro_store::EventPayload::new( let payload = fabro_store::EventPayload::new(
serde_json::to_value(&stored).map_err(|err| FabroError::engine(err.to_string()))?, serde_json::to_value(&stored).map_err(|err| FabroError::engine(err.to_string()))?,

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

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

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

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