diff --git a/lib/crates/fabro-cli/tests/it/cmd/support.rs b/lib/crates/fabro-cli/tests/it/cmd/support.rs index 53e75af42..3db026f63 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/support.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/support.rs @@ -661,15 +661,7 @@ pub(crate) fn run_events(run_dir: &Path) -> Vec { run_dir, &format!("/api/v1/runs/{run_id}/events"), )); - let items = response["data"] - .as_array() - .cloned() - .expect("event list response should contain a data array"); - items - .into_iter() - .map(serde_json::from_value) - .collect::, _>>() - .expect("wire event envelope list should parse") + crate::support::parse_event_envelopes(&response) } pub(crate) fn wait_for_event_names(run_dir: &Path, expected: &[&str]) { diff --git a/lib/crates/fabro-cli/tests/it/support/mod.rs b/lib/crates/fabro-cli/tests/it/support/mod.rs index 5e965f312..5fcd359e0 100644 --- a/lib/crates/fabro-cli/tests/it/support/mod.rs +++ b/lib/crates/fabro-cli/tests/it/support/mod.rs @@ -1,6 +1,7 @@ use std::path::{Path, PathBuf}; use assert_cmd::Command; +use fabro_store::EventEnvelope; use fabro_test::TestContext; use fabro_types::RunId; macro_rules! fabro_json_snapshot { @@ -60,6 +61,17 @@ pub(crate) fn unique_run_id() -> String { RunId::new().to_string() } +pub(crate) fn parse_event_envelopes(response: &serde_json::Value) -> Vec { + response["data"] + .as_array() + .expect("event list response should contain a data array") + .iter() + .cloned() + .map(serde_json::from_value) + .collect::, _>>() + .expect("wire event envelope list should parse") +} + pub(crate) struct LightweightCli { home_dir: tempfile::TempDir, } diff --git a/lib/crates/fabro-cli/tests/it/workflow/mod.rs b/lib/crates/fabro-cli/tests/it/workflow/mod.rs index b07bed9e2..5ada38260 100644 --- a/lib/crates/fabro-cli/tests/it/workflow/mod.rs +++ b/lib/crates/fabro-cli/tests/it/workflow/mod.rs @@ -173,15 +173,7 @@ fn run_events(run_dir: &Path) -> Vec { storage_dir, &format!("/api/v1/runs/{run_id}/events"), )); - let items = response["data"] - .as_array() - .cloned() - .expect("event list response should contain a data array"); - items - .into_iter() - .map(serde_json::from_value) - .collect::, _>>() - .expect("wire event envelope list should parse") + crate::support::parse_event_envelopes(&response) } macro_rules! sandbox_tests { diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index 80d872997..63db2baad 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -34,7 +34,7 @@ use fabro_store::{ ArtifactStore, Database, EventEnvelope, EventPayload, PendingInterviewRecord, StageId, }; use fabro_types::{ - ActorKind, ActorRef, EventBody, InterviewQuestionRecord, InterviewQuestionType, RunBlobId, + ActorRef, EventBody, InterviewQuestionRecord, InterviewQuestionType, RunBlobId, RunClientProvenance, RunControlAction, RunEvent, RunId, RunProvenance, RunServerProvenance, RunSubjectProvenance, Settings, }; @@ -5155,12 +5155,7 @@ async fn append_control_request( } fn actor_from_subject(subject: &AuthenticatedSubject) -> Option { - let login = subject.login.clone()?; - Some(ActorRef { - kind: ActorKind::User, - id: Some(login.clone()), - display: Some(login), - }) + subject.login.clone().map(ActorRef::user) } fn schedule_worker_kill(state: Arc, run_id: RunId, worker_pid: u32) { diff --git a/lib/crates/fabro-types/src/run_event/mod.rs b/lib/crates/fabro-types/src/run_event/mod.rs index 80fe77849..b9949bbd6 100644 --- a/lib/crates/fabro-types/src/run_event/mod.rs +++ b/lib/crates/fabro-types/src/run_event/mod.rs @@ -44,6 +44,17 @@ pub struct ActorRef { pub display: Option, } +impl ActorRef { + #[must_use] + pub fn user(login: String) -> Self { + Self { + kind: ActorKind::User, + id: Some(login.clone()), + display: Some(login), + } + } +} + #[derive(Debug, Clone, PartialEq)] pub struct RunEvent { pub id: String, @@ -677,50 +688,46 @@ impl RunEvent { } pub fn to_value(&self) -> serde_json::Result { + fn insert_opt( + map: &mut Map, + key: &str, + value: Option<&T>, + ) -> serde_json::Result<()> { + if let Some(v) = value { + map.insert(key.to_string(), serde_json::to_value(v)?); + } + Ok(()) + } + let mut map = Map::new(); - map.insert("id".to_string(), serde_json::to_value(&self.id)?); + map.insert("id".to_string(), Value::String(self.id.clone())); map.insert("ts".to_string(), serde_json::to_value(self.ts)?); map.insert("run_id".to_string(), serde_json::to_value(self.run_id)?); map.insert( "event".to_string(), Value::String(self.body.event_name().to_string()), ); - if let Some(value) = &self.session_id { - map.insert("session_id".to_string(), Value::String(value.clone())); - } - if let Some(value) = &self.parent_session_id { - map.insert( - "parent_session_id".to_string(), - Value::String(value.clone()), - ); - } - if let Some(value) = &self.node_id { - map.insert("node_id".to_string(), Value::String(value.clone())); - } - if let Some(value) = &self.node_label { - map.insert("node_label".to_string(), Value::String(value.clone())); - } - if let Some(value) = &self.stage_id { - map.insert("stage_id".to_string(), serde_json::to_value(value)?); - } - if let Some(value) = &self.parallel_group_id { - map.insert( - "parallel_group_id".to_string(), - serde_json::to_value(value)?, - ); - } - if let Some(value) = &self.parallel_branch_id { - map.insert( - "parallel_branch_id".to_string(), - serde_json::to_value(value)?, - ); - } - if let Some(value) = &self.tool_call_id { - map.insert("tool_call_id".to_string(), Value::String(value.clone())); - } - if let Some(actor) = &self.actor { - map.insert("actor".to_string(), serde_json::to_value(actor)?); - } + insert_opt(&mut map, "session_id", self.session_id.as_ref())?; + insert_opt( + &mut map, + "parent_session_id", + self.parent_session_id.as_ref(), + )?; + insert_opt(&mut map, "node_id", self.node_id.as_ref())?; + insert_opt(&mut map, "node_label", self.node_label.as_ref())?; + insert_opt(&mut map, "stage_id", self.stage_id.as_ref())?; + insert_opt( + &mut map, + "parallel_group_id", + self.parallel_group_id.as_ref(), + )?; + insert_opt( + &mut map, + "parallel_branch_id", + self.parallel_branch_id.as_ref(), + )?; + insert_opt(&mut map, "tool_call_id", self.tool_call_id.as_ref())?; + insert_opt(&mut map, "actor", self.actor.as_ref())?; map.insert("properties".to_string(), self.body.properties_value()?); Ok(Value::Object(map)) } diff --git a/lib/crates/fabro-workflow/src/context.rs b/lib/crates/fabro-workflow/src/context.rs index 482089f38..00eb7a861 100644 --- a/lib/crates/fabro-workflow/src/context.rs +++ b/lib/crates/fabro-workflow/src/context.rs @@ -136,7 +136,6 @@ 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}; @@ -188,13 +187,7 @@ impl WorkflowContext for Context { 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(), - }) + Some(StageScope::from_context(self, node_id)) } } diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs index 2f299a7e3..c5eb56052 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/lib/crates/fabro-workflow/src/event.rs @@ -1435,12 +1435,12 @@ fn stored_event_fields_for_variant(event: &Event) -> StoredEventFields { } fn actor_from_provenance(provenance: &RunProvenance) -> Option { - let login = provenance.subject.as_ref()?.login.clone()?; - Some(ActorRef { - kind: ActorKind::User, - id: Some(login.clone()), - display: Some(login), - }) + provenance + .subject + .as_ref()? + .login + .clone() + .map(ActorRef::user) } fn agent_tool_call_id(event: &AgentEvent) -> Option<&str> { @@ -2475,18 +2475,25 @@ pub struct StageScope { } impl StageScope { - /// Build scope for a handler invocation. Prefers the current_stage_scope - /// set by the fidelity lifecycle before_attempt hook, but falls back to - /// a scope synthesized from the node id and the context's visit count - /// for tests and other direct-handler call sites that don't go through - /// the full lifecycle. - pub fn for_handler(context: &WfContext, node_id: impl Into) -> Self { - context.current_stage_scope().unwrap_or_else(|| Self { + /// Build a scope from the given node id, sourcing visit count and parallel + /// ids from the current context. + pub fn from_context(context: &WfContext, node_id: impl Into) -> Self { + 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(), - }) + } + } + + /// Build scope for a handler invocation. Prefers the `current_stage_scope` + /// seeded by the fidelity lifecycle before_attempt hook, and falls back to + /// synthesizing one from `node_id` for direct-handler call sites (tests, + /// etc.) that don't go through the full lifecycle. + pub fn for_handler(context: &WfContext, node_id: impl Into) -> Self { + context + .current_stage_scope() + .unwrap_or_else(|| Self::from_context(context, node_id)) } } diff --git a/lib/crates/fabro-workflow/src/handler/parallel.rs b/lib/crates/fabro-workflow/src/handler/parallel.rs index 0592595b9..acbb8fc7c 100644 --- a/lib/crates/fabro-workflow/src/handler/parallel.rs +++ b/lib/crates/fabro-workflow/src/handler/parallel.rs @@ -214,12 +214,11 @@ impl Handler for ParallelHandler { ); branch_context.set( keys::INTERNAL_PARALLEL_GROUP_ID, - serde_json::to_value(¶llel_group_id).expect("StageId serializes as string"), + serde_json::Value::String(parallel_group_id.to_string()), ); branch_context.set( keys::INTERNAL_PARALLEL_BRANCH_ID, - serde_json::to_value(¶llel_branch_id) - .expect("ParallelBranchId serializes as string"), + serde_json::Value::String(parallel_branch_id.to_string()), ); let (branch_sandbox, worktree_path): (Arc, Option) = if let ( @@ -280,8 +279,6 @@ 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 { @@ -304,7 +301,12 @@ 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 branch_scope = StageScope { + node_id: setup.target_id.clone(), + visit: 1, + parallel_group_id: Some(group_id.clone()), + parallel_branch_id: Some(setup.parallel_branch_id.clone()), + }; let handle = tokio::spawn(async move { let _permit = sem @@ -405,10 +407,13 @@ impl Handler for ParallelHandler { match sha_result { Ok(r) if r.exit_code == 0 => { let sha = r.stdout.trim().to_string(); - emitter.emit(&Event::GitCommit { - node_id: Some(setup.target_id.clone()), - sha: sha.clone(), - }); + emitter.emit_scoped( + &Event::GitCommit { + node_id: Some(setup.target_id.clone()), + sha: sha.clone(), + }, + &branch_scope, + ); Some(sha) } _ => None,