From f414a7d719f94b4c20d4ff736726b3fd0bf50948 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Thu, 9 Apr 2026 22:11:20 -0400 Subject: [PATCH] chore(simplify): events schema v2 cleanup from review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cleanup pass on the events schema v2 work merged from origin/main. Quality fixes: - prompt.rs: drop dead `_visit` local; use stage_scope.visit at the emit site (the value was being recomputed inline next to a scope that already had it). - llm/cli.rs: rename `_context` to `context` in CodergenBackend::run (it's actually used now); delete the lingering `current_visit` helper that was deleted from llm/api.rs in c6a78a428 but missed here; use stage_scope.visit at the emit site. - llm/api.rs: rename `event_scope` to `stage_scope` for consistency with every other handler. - agent.rs, fan_in.rs, parallel.rs: same `visit_from_context` → `stage_scope.visit` substitution at every event-emit site. - parallel.rs: switch ParallelStarted/ParallelCompleted from `emit` to `emit_scoped` so they carry stage_id in the envelope. - event.rs: fix the StageScope::for_handler docstring — the lifecycle hook is `before_node`, not `before_attempt`. Reuse fixes: - run_event/mod.rs: add `ActorRef::agent(session_id, display)` symmetric with the existing `ActorRef::user`; use it from agent_actor_for_event in workflow event.rs. Correctness fixes: - event.rs: introduce `StageScope::for_parallel_branch` to name the "branch starts at visit 1" invariant the parallel handler was hardcoding via a struct literal at parallel.rs:307. This makes the assumption auditable and gives a single place to fix when parallel nodes ever loop. Efficiency fixes: - stage_id.rs: switch StageId/ParallelBranchId Serialize impls from `serializer.serialize_str(&self.to_string())` to `collect_str(self)`, removing one transient String allocation per ID per emitted event. Hardening: - event.rs: add `#[must_use]` on `to_run_event`, `to_run_event_at`, and `event_name`. - store/types.rs: add a second wire-envelope round-trip test that populates stage_id, parallel_group_id, parallel_branch_id, session_id, parent_session_id, tool_call_id, and actor — the existing test only exercised stage_id, so a regression in any of the other envelope fields' #[serde(flatten)] interaction would have been silent. All 3810 workspace tests pass; clippy and fmt clean. Co-Authored-By: Claude Opus 4.6 (1M context) --- lib/crates/fabro-store/src/types.rs | 58 ++++++++++++++++++- lib/crates/fabro-types/src/run_event/mod.rs | 9 +++ lib/crates/fabro-types/src/stage_id.rs | 4 +- lib/crates/fabro-workflow/src/event.rs | 43 +++++++++++--- .../fabro-workflow/src/handler/agent.rs | 7 +-- .../fabro-workflow/src/handler/fan_in.rs | 4 +- .../fabro-workflow/src/handler/llm/api.rs | 8 +-- .../fabro-workflow/src/handler/llm/cli.rs | 11 +--- .../fabro-workflow/src/handler/parallel.rs | 50 +++++++++------- .../fabro-workflow/src/handler/prompt.rs | 4 +- 10 files changed, 142 insertions(+), 56 deletions(-) diff --git a/lib/crates/fabro-store/src/types.rs b/lib/crates/fabro-store/src/types.rs index 35b485f01..aceafbc85 100644 --- a/lib/crates/fabro-store/src/types.rs +++ b/lib/crates/fabro-store/src/types.rs @@ -91,7 +91,10 @@ pub struct EventEnvelope { mod tests { use chrono::{TimeZone, Utc}; - use fabro_types::{EventBody, RunEvent, StageId, fixtures, run_event::RunCompletedProps}; + use fabro_types::{ + ActorRef, EventBody, ParallelBranchId, RunEvent, StageId, fixtures, + run_event::RunCompletedProps, + }; use super::{EventEnvelope, EventPayload}; @@ -133,4 +136,57 @@ mod tests { let parsed: EventEnvelope = serde_json::from_value(wire).unwrap(); assert_eq!(parsed, envelope); } + + #[test] + fn wire_event_envelope_round_trips_with_all_envelope_fields() { + let group = StageId::new("review", 2); + let branch = ParallelBranchId::new(group.clone(), 3); + let event = RunEvent { + id: "evt_2".to_string(), + ts: Utc.with_ymd_and_hms(2026, 4, 9, 13, 0, 0).unwrap(), + run_id: fixtures::RUN_1, + node_id: Some("review".to_string()), + node_label: Some("Review".to_string()), + stage_id: Some(StageId::new("review", 2)), + parallel_group_id: Some(group), + parallel_branch_id: Some(branch), + session_id: Some("ses_42".to_string()), + parent_session_id: Some("ses_root".to_string()), + tool_call_id: Some("tool_call_xyz".to_string()), + actor: Some(ActorRef::agent( + Some("ses_42".to_string()), + Some("claude-sonnet".to_string()), + )), + body: EventBody::RunCompleted(RunCompletedProps { + duration_ms: 100, + artifact_count: 1, + status: "success".to_string(), + reason: None, + total_usd_micros: None, + final_git_commit_sha: None, + final_patch: None, + billing: None, + }), + }; + let payload = EventPayload::new(event.to_value().unwrap(), &fixtures::RUN_1).unwrap(); + let envelope = EventEnvelope { seq: 99, payload }; + + let wire = serde_json::to_value(&envelope).unwrap(); + assert_eq!(wire["seq"], 99); + assert_eq!(wire["id"], "evt_2"); + assert_eq!(wire["stage_id"], "review@2"); + assert_eq!(wire["parallel_group_id"], "review@2"); + assert_eq!(wire["parallel_branch_id"], "review@2:3"); + assert_eq!(wire["session_id"], "ses_42"); + assert_eq!(wire["parent_session_id"], "ses_root"); + assert_eq!(wire["tool_call_id"], "tool_call_xyz"); + assert_eq!(wire["actor"]["kind"], "agent"); + assert_eq!(wire["actor"]["id"], "ses_42"); + assert_eq!(wire["actor"]["display"], "claude-sonnet"); + assert_eq!(wire["event"], "run.completed"); + assert!(wire.get("payload").is_none(), "wire shape must be flat"); + + let parsed: EventEnvelope = serde_json::from_value(wire).unwrap(); + assert_eq!(parsed, envelope); + } } diff --git a/lib/crates/fabro-types/src/run_event/mod.rs b/lib/crates/fabro-types/src/run_event/mod.rs index ca6d59107..18f6389ab 100644 --- a/lib/crates/fabro-types/src/run_event/mod.rs +++ b/lib/crates/fabro-types/src/run_event/mod.rs @@ -53,6 +53,15 @@ impl ActorRef { display: Some(login), } } + + #[must_use] + pub fn agent(session_id: Option, display: Option) -> Self { + Self { + kind: ActorKind::Agent, + id: session_id, + display, + } + } } #[derive(Debug, Clone, PartialEq)] diff --git a/lib/crates/fabro-types/src/stage_id.rs b/lib/crates/fabro-types/src/stage_id.rs index 846d61cb8..eef135774 100644 --- a/lib/crates/fabro-types/src/stage_id.rs +++ b/lib/crates/fabro-types/src/stage_id.rs @@ -76,7 +76,7 @@ impl Serialize for StageId { where S: Serializer, { - serializer.serialize_str(&self.to_string()) + serializer.collect_str(self) } } @@ -157,7 +157,7 @@ impl Serialize for ParallelBranchId { where S: Serializer, { - serializer.serialize_str(&self.to_string()) + serializer.collect_str(self) } } diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs index 5b9bdc2f5..7e56abb92 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/lib/crates/fabro-workflow/src/event.rs @@ -5,8 +5,8 @@ use std::sync::atomic::{AtomicI64, Ordering}; use ::fabro_types::run_event as fabro_types; use ::fabro_types::{ - ActorKind, ActorRef, BilledTokenCounts, ParallelBranchId, RunBlobId, RunControlAction, - RunEvent, RunId, RunProvenance, StageId, StageStatus, StatusReason, + ActorRef, BilledTokenCounts, ParallelBranchId, RunBlobId, RunControlAction, RunEvent, RunId, + RunProvenance, StageId, StageStatus, StatusReason, }; use anyhow::{Context, Result}; use chrono::Utc; @@ -1141,6 +1141,7 @@ impl Event { } } +#[must_use] pub fn event_name(event: &Event) -> &'static str { match event { Event::RunCreated { .. } => "run.created", @@ -1453,11 +1454,10 @@ fn agent_tool_call_id(event: &AgentEvent) -> Option<&str> { fn agent_actor_for_event(event: &AgentEvent, session_id: Option<&str>) -> Option { match event { - AgentEvent::AssistantMessage { model, .. } => Some(ActorRef { - kind: ActorKind::Agent, - id: session_id.map(str::to_string), - display: Some(model.clone()), - }), + AgentEvent::AssistantMessage { model, .. } => Some(ActorRef::agent( + session_id.map(str::to_string), + Some(model.clone()), + )), _ => None, } } @@ -2487,7 +2487,7 @@ impl StageScope { } /// Build scope for a handler invocation. Prefers the `current_stage_scope` - /// seeded by the fidelity lifecycle before_attempt hook, and falls back to + /// seeded by the fidelity lifecycle `before_node` 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 { @@ -2495,12 +2495,38 @@ impl StageScope { .current_stage_scope() .unwrap_or_else(|| Self::from_context(context, node_id)) } + + /// Build scope for the branch-lifecycle events emitted by the parallel + /// handler (`ParallelBranchStarted`, `ParallelBranchCompleted`, and the + /// pre-dispatch `GitCommit` for the branch worktree). + /// + /// `target_visit` is the visit count of `target_node_id` for this + /// particular branch dispatch. The parallel handler currently passes + /// `1` because branches haven't been re-entered yet at the point of + /// scope construction; a future change that loops a parallel node + /// must pass the actual visit so envelope `stage_id`s stay accurate. + #[must_use] + pub fn for_parallel_branch( + target_node_id: impl Into, + target_visit: u32, + parallel_group_id: StageId, + parallel_branch_id: ParallelBranchId, + ) -> Self { + Self { + node_id: target_node_id.into(), + visit: target_visit, + parallel_group_id: Some(parallel_group_id), + parallel_branch_id: Some(parallel_branch_id), + } + } } +#[must_use] pub fn to_run_event(run_id: &RunId, event: &Event) -> RunEvent { to_run_event_at(run_id, event, Utc::now(), None) } +#[must_use] pub fn to_run_event_at( run_id: &RunId, event: &Event, @@ -2846,6 +2872,7 @@ impl Emitter { #[cfg(test)] mod tests { use super::*; + use ::fabro_types::ActorKind; use ::fabro_types::fixtures; use std::sync::{Arc, Mutex}; diff --git a/lib/crates/fabro-workflow/src/handler/agent.rs b/lib/crates/fabro-workflow/src/handler/agent.rs index 541c2d226..484f65b5e 100644 --- a/lib/crates/fabro-workflow/src/handler/agent.rs +++ b/lib/crates/fabro-workflow/src/handler/agent.rs @@ -14,7 +14,6 @@ use crate::event::{Emitter, Event, StageScope}; use crate::outcome::{ BilledModelUsage, FailureCategory, FailureDetail, Outcome, OutcomeExt, StageStatus, }; -use crate::run_dir::visit_from_context; use crate::transforms::variable_expansion::expand_vars; use fabro_graphviz::graph::{Graph, Node}; @@ -250,7 +249,6 @@ impl Handler for AgentHandler { format!("{preamble}\n\n{expanded}") }; - let visit = u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX); let prompt_provider = node .provider() .map(String::from) @@ -260,7 +258,7 @@ impl Handler for AgentHandler { services.emitter.emit_scoped( &Event::Prompt { stage: node.id.clone(), - visit, + visit: stage_scope.visit, text: prompt.clone(), mode: Some("agent".to_string()), provider: prompt_provider, @@ -720,8 +718,7 @@ mod tests { 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), + visit: scope.visit, event: fabro_agent::AgentEvent::SessionStarted { provider: Some("openai".to_string()), model: Some("gpt-5.4".to_string()), diff --git a/lib/crates/fabro-workflow/src/handler/fan_in.rs b/lib/crates/fabro-workflow/src/handler/fan_in.rs index 458ff4886..0c1cc67dc 100644 --- a/lib/crates/fabro-workflow/src/handler/fan_in.rs +++ b/lib/crates/fabro-workflow/src/handler/fan_in.rs @@ -6,7 +6,6 @@ use crate::context::keys; use crate::error::FabroError; 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; use async_trait::async_trait; use fabro_agent::Sandbox; @@ -231,13 +230,12 @@ async fn llm_evaluate( Respond with the ID of the best candidate." ); - 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_scoped( &Event::Prompt { stage: node_id.to_string(), - visit: visit_u32, + visit: stage_scope.visit, text: full_prompt.clone(), mode: Some("fan_in".to_string()), provider: None, diff --git a/lib/crates/fabro-workflow/src/handler/llm/api.rs b/lib/crates/fabro-workflow/src/handler/llm/api.rs index 3e03af58d..9b13ac08c 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/api.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/api.rs @@ -454,13 +454,13 @@ impl CodergenBackend for AgentApiBackend { touched: HashSet::new(), last: None, })); - let event_scope = StageScope::for_handler(context, &node.id); + let stage_scope = StageScope::for_handler(context, &node.id); // Subscribe to session events: forward to pipeline emitter + track files. spawn_event_forwarder( &session, node.id.clone(), - event_scope.clone(), + stage_scope.clone(), Arc::clone(emitter), Arc::clone(&file_tracking), ); @@ -497,7 +497,7 @@ impl CodergenBackend for AgentApiBackend { to_model: target.model.clone(), error: error_msg.clone(), }, - &event_scope, + &stage_scope, ); let target_provider: Provider = match target.provider.parse() { @@ -528,7 +528,7 @@ impl CodergenBackend for AgentApiBackend { spawn_event_forwarder( &session, node.id.clone(), - event_scope.clone(), + stage_scope.clone(), Arc::clone(emitter), Arc::clone(&file_tracking), ); diff --git a/lib/crates/fabro-workflow/src/handler/llm/cli.rs b/lib/crates/fabro-workflow/src/handler/llm/cli.rs index 9e52d8c63..08e2b04f8 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/cli.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/cli.rs @@ -12,7 +12,6 @@ use crate::context::Context; use crate::error::FabroError; 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; use fabro_llm::types::TokenCounts; @@ -55,10 +54,6 @@ impl AgentCli { } } -fn current_visit(context: &Context) -> u32 { - u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX) -} - /// Ensure the CLI tool for the given provider is installed in the sandbox. /// /// Checks if the CLI binary exists; if not, installs Node.js (if missing) and @@ -461,7 +456,7 @@ impl CodergenBackend for AgentCliBackend { &self, node: &Node, prompt: &str, - _context: &Context, + context: &Context, _thread_id: Option<&str>, emitter: &Arc, sandbox: &Arc, @@ -496,11 +491,11 @@ impl CodergenBackend for AgentCliBackend { ensure_cli(cli, provider, sandbox, emitter).await?; let command = cli_command_for_provider(provider, model, &prompt_path); - let stage_scope = StageScope::for_handler(_context, &node.id); + let stage_scope = StageScope::for_handler(context, &node.id); emitter.emit_scoped( &Event::AgentCliStarted { node_id: node.id.clone(), - visit: current_visit(_context), + visit: stage_scope.visit, mode: "cli".to_string(), provider: provider.as_str().to_string(), model: model.to_string(), diff --git a/lib/crates/fabro-workflow/src/handler/parallel.rs b/lib/crates/fabro-workflow/src/handler/parallel.rs index acbb8fc7c..de674fe9a 100644 --- a/lib/crates/fabro-workflow/src/handler/parallel.rs +++ b/lib/crates/fabro-workflow/src/handler/parallel.rs @@ -151,15 +151,18 @@ impl Handler for ParallelHandler { .unwrap_or("wait_all"), ); - let parallel_visit = u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX); - let parallel_group_id = StageId::new(node.id.clone(), parallel_visit); + let parallel_stage_scope = StageScope::for_handler(context, &node.id); + let parallel_group_id = StageId::new(node.id.clone(), parallel_stage_scope.visit); - services.emitter.emit(&Event::ParallelStarted { - node_id: node.id.clone(), - visit: parallel_visit, - branch_count: branches.len(), - join_policy: join_policy.to_string(), - }); + services.emitter.emit_scoped( + &Event::ParallelStarted { + node_id: node.id.clone(), + visit: parallel_stage_scope.visit, + branch_count: branches.len(), + join_policy: join_policy.to_string(), + }, + ¶llel_stage_scope, + ); { let run_id = context .run_id() @@ -301,12 +304,12 @@ impl Handler for ParallelHandler { .map(|gs| gs.git_author.clone()) .unwrap_or_default(); let group_id = parallel_group_id.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 branch_scope = StageScope::for_parallel_branch( + setup.target_id.clone(), + 1, + group_id.clone(), + setup.parallel_branch_id.clone(), + ); let handle = tokio::spawn(async move { let _permit = sem @@ -526,14 +529,17 @@ impl Handler for ParallelHandler { context.set(keys::PARALLEL_RESULTS, serde_json::json!(results_json)); context.set(keys::PARALLEL_BRANCH_COUNT, serde_json::json!(total)); - services.emitter.emit(&Event::ParallelCompleted { - node_id: node.id.clone(), - visit: u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX), - duration_ms: millis_u64(parallel_start.elapsed()), - success_count, - failure_count: fail_count, - results: results_json.clone(), - }); + services.emitter.emit_scoped( + &Event::ParallelCompleted { + node_id: node.id.clone(), + visit: parallel_stage_scope.visit, + duration_ms: millis_u64(parallel_start.elapsed()), + success_count, + failure_count: fail_count, + results: results_json.clone(), + }, + ¶llel_stage_scope, + ); { let run_id = context .run_id() diff --git a/lib/crates/fabro-workflow/src/handler/prompt.rs b/lib/crates/fabro-workflow/src/handler/prompt.rs index 9dc5e25c2..67fc8227c 100644 --- a/lib/crates/fabro-workflow/src/handler/prompt.rs +++ b/lib/crates/fabro-workflow/src/handler/prompt.rs @@ -5,7 +5,6 @@ use crate::context::{Context, WorkflowContext}; use crate::error::FabroError; use crate::event::{Event, StageScope}; use crate::outcome::Outcome; -use crate::run_dir::visit_from_context; use async_trait::async_trait; use fabro_graphviz::graph::{Graph, Node}; use fabro_model::Provider; @@ -60,7 +59,6 @@ impl Handler for PromptHandler { } else { format!("{preamble}\n\n{expanded}") }; - let _visit = u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX); // 1b. Discover project docs for system prompt when project_memory is enabled let system_prompt = if node.project_memory() { @@ -95,7 +93,7 @@ impl Handler for PromptHandler { services.emitter.emit_scoped( &Event::Prompt { stage: node.id.clone(), - visit: u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX), + visit: stage_scope.visit, text: prompt.clone(), mode: Some("prompt".to_string()), provider: prompt_provider.clone(),