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

chore(simplify): events schema v2 cleanup from review

Browse source 
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) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-04-09 22:11:20 -04:00
parent 09b616cb12
commit f414a7d719
No known key found for this signature in database
10 changed files with 142 additions and 56 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

58
lib/crates/fabro-store/src/types.rs
View file

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

9
lib/crates/fabro-types/src/run_event/mod.rs
View file

@ -53,6 +53,15 @@ impl ActorRef {
display: Some(login),
}
}
#[must_use]
pub fn agent(session_id: Option<String>, display: Option<String>) -> Self {
Self {
kind: ActorKind::Agent,
id: session_id,
display,
}
}
}
#[derive(Debug, Clone, PartialEq)]

4
lib/crates/fabro-types/src/stage_id.rs
View file

@ -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)
}
}

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

@ -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<ActorRef> {
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<String>) -> 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<String>,
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};

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

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

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

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

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

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

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

@ -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<Emitter>,
sandbox: &Arc<dyn Sandbox>,
@ -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(),

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

@ -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(),
},
&parallel_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(),
},
&parallel_stage_scope,
);
{
let run_id = context
.run_id()

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

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