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

refactor(events): 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 d70ba41445
commit 49767a43fe
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(),
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);

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

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

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

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

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"
},
"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]"
}
]

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"
},
"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]"
},
{

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

@ -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<StageId>;
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 {
@ -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<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)]

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

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

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 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<StageId>,
#[serde(default, skip_serializing_if = "Option::is_none")]
parallel_branch_id: Option<ParallelBranchId>,
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<StageId>,
#[serde(default, skip_serializing_if = "Option::is_none")]
parallel_branch_id: Option<ParallelBranchId>,
duration_ms: u64,
status: String,
preferred_label: Option<String>,
@ -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<StageId>,
#[serde(default, skip_serializing_if = "Option::is_none")]
parallel_branch_id: Option<ParallelBranchId>,
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<StageId>,
#[serde(default, skip_serializing_if = "Option::is_none")]
parallel_branch_id: Option<ParallelBranchId>,
attempt: usize,
max_attempts: usize,
delay_ms: u64,
@ -377,10 +359,6 @@ pub enum Event {
session_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
parent_session_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
parallel_group_id: Option<StageId>,
#[serde(default, skip_serializing_if = "Option::is_none")]
parallel_branch_id: Option<ParallelBranchId>,
},
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<StageId>,
pub parallel_branch_id: Option<ParallelBranchId>,
}
pub fn to_run_event_at(run_id: &RunId, event: &Event, ts: chrono::DateTime<Utc>) -> 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<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);
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");

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

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

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::{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<dyn fabro_agent::Sandbox>,
_tool_hooks: Option<Arc<dyn fabro_agent::ToolHookCallback>>,
) -> Result<CodergenResult, FabroError> {
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,

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::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!(

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

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::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<Emitter>, event: &Event) {
fn emit(&self, default_emitter: &Arc<Emitter>, 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

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 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<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.
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<Emitter>,
file_tracking: Arc<Mutex<FileTracking>>,
) {
@ -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,

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

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::{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, FabroError>(BranchResult {
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::{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();

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::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<WorkflowGraph> 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

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

@ -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<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)
}
fn stage_parallel_ids(state: &WfRunState) -> (Option<StageId>, Option<ParallelBranchId>) {
(
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<WorkflowGraph> for EventLifecycle {
}
let gv = node.inner();
let stage_index = state.stage_index;
let visit = stage_visit(state, &gv.id);
let (parallel_group_id, parallel_branch_id) = stage_parallel_ids(state);
let 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<WorkflowGraph> for EventLifecycle {
state: &WfRunState,
) -> CoreResult<NodeDecision<Option<BilledModelUsage>>> {
let gv = ctx.node.inner();
let (parallel_group_id, parallel_branch_id) = stage_parallel_ids(state);
self.emitter.emit(&Event::StageStarted {
node_id: gv.id.clone(),
name: gv.label().to_string(),
index: state.stage_index,
visit: stage_visit(state, &gv.id),
parallel_group_id,
parallel_branch_id,
handler_type: gv.handler_type().unwrap_or_default().to_string(),
attempt: ctx.attempt as usize,
max_attempts: ctx.max_attempts as usize,
});
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<WorkflowGraph> 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<WorkflowGraph> for EventLifecycle {
}
let gv = node.inner();
let stage_index = state.stage_index;
let visit = stage_visit(state, &gv.id);
let (parallel_group_id, parallel_branch_id) = stage_parallel_ids(state);
let 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::<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<_, _>>())
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::<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,
);
} 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::<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(())
}
@ -367,37 +367,44 @@ impl RunLifecycle<WorkflowGraph> 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::<BTreeMap<_, _>>(),
context_values: context_values.into_iter().collect::<BTreeMap<_, _>>(),
node_outcomes: node_outcomes.into_iter().collect::<BTreeMap<_, _>>(),
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::<BTreeMap<_, _>>(),
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::<BTreeMap<_, _>>(),
context_values: context_values.into_iter().collect::<BTreeMap<_, _>>(),
node_outcomes: node_outcomes.into_iter().collect::<BTreeMap<_, _>>(),
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::<BTreeMap<_, _>>(),
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 {

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::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<WorkflowGraph> 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}"
)));

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

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

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

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

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(),
session_id: Some(event.session_id.clone()),
parent_session_id: event.parent_session_id.clone(),
parallel_group_id: None,
parallel_branch_id: None,
});
}
})