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

refactor(events): tidy schema v2 plumbing

Browse source 
Quality cleanup on top of the v2 envelope commits:

- fabro-workflow/src/event.rs: add ActorKind/ActorRef/RunProvenance
  to the existing ::fabro_types import block so call sites can use
  unqualified names (restores CLAUDE.md import style). Extract a
  node_stored_fields helper to collapse 4 near-identical match arms
  in stored_event_fields. Drop the no-op ..default() from the Agent
  arm where all 9 fields are set explicitly.
- fabro-types/src/run_event/mod.rs: collapse 9 copies of the
  obj.get/as_str/to_string chain in from_ref behind an opt_str
  closure.
- fabro-server/src/server.rs: dedupe the two identical error
  closures in api_event_envelope_from_store. Skip the typed
  ApiEventEnvelope roundtrip in sse_event_from_store so streamed
  events go straight from the wire Value to a JSON string.
- fabro-workflow/src/handler/llm/api.rs: inline current_visit into
  its sole caller current_stage_event_scope.

Also fixes pre-existing test compile breakage carried in by the
v2 commits: restore the fabro_types::RunId import in support.rs
(removed by b51403ae but still referenced by find_run_dir), and
thread parallel_group_id/parallel_branch_id: None through 9
Event::Stage*/Event::Agent constructors in run_progress and
store/dump tests that 91d61016 missed.

No behavior change aside from the SSE hot path avoiding one full
strong-type deserialize + reserialize per event.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-04-09 11:42:02 -04:00
parent b51403ae64
commit 747ccc0383
No known key found for this signature in database
8 changed files with 62 additions and 106 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -511,6 +511,8 @@ mod tests {
name: "Plan".into(),
index: 0,
visit: 1,
parallel_group_id: None,
parallel_branch_id: None,
duration_ms: 5000,
status: "success".into(),
preferred_label: None,
@ -555,6 +557,8 @@ 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);

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

@ -479,6 +479,8 @@ mod tests {
event,
session_id: None,
parent_session_id: None,
parallel_group_id: None,
parallel_branch_id: None,
}
}
@ -488,6 +490,8 @@ mod tests {
name: name.into(),
index: 0,
visit: 1,
parallel_group_id: None,
parallel_branch_id: None,
handler_type: String::new(),
attempt: 1,
max_attempts: 1,
@ -512,6 +516,8 @@ mod tests {
name: name.into(),
index: 0,
visit: 1,
parallel_group_id: None,
parallel_branch_id: None,
duration_ms: 5000,
status: "success".into(),
preferred_label: None,
@ -709,6 +715,8 @@ mod tests {
name: "Code".into(),
index: 0,
visit: 1,
parallel_group_id: None,
parallel_branch_id: None,
attempt: 2,
max_attempts: 3,
delay_ms: 1500,
@ -954,6 +962,8 @@ mod tests {
name: "Code".into(),
index: 0,
visit: 1,
parallel_group_id: None,
parallel_branch_id: None,
attempt: 2,
max_attempts: 3,
delay_ms: 1500,
@ -1159,6 +1169,8 @@ mod tests {
name: "Code".into(),
index: 0,
visit: 1,
parallel_group_id: None,
parallel_branch_id: None,
handler_type: "agent".into(),
attempt: 1,
max_attempts: 1,

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

@ -600,6 +600,8 @@ mod tests {
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,

1
lib/crates/fabro-cli/tests/it/cmd/support.rs
View file

@ -14,6 +14,7 @@ use fabro_config::Storage;
use fabro_server::bind::Bind;
use fabro_store::EventEnvelope;
use fabro_test::TestContext;
use fabro_types::RunId;
use serde_json::Value;
use shlex::try_quote;

17
lib/crates/fabro-server/src/server.rs
View file

@ -1480,8 +1480,8 @@ fn event_matches_run_filter(event: &EventEnvelope, run_filter: Option<&HashSet<R
}
fn sse_event_from_store(event: &EventEnvelope) -> Option<Event> {
let event = api_event_envelope_from_store(event).ok()?;
let data = serde_json::to_string(&event).ok()?;
let wire = event.to_wire_value().ok()?;
let data = serde_json::to_string(&wire).ok()?;
let data = redact_jsonl_line(&data);
Some(Event::default().data(data))
}
@ -2381,20 +2381,15 @@ fn octet_stream_response(bytes: Bytes) -> Response {
#[allow(clippy::result_large_err)]
fn api_event_envelope_from_store(event: &EventEnvelope) -> Result<ApiEventEnvelope, Response> {
let value = event.to_wire_value().map_err(|err| {
fn serialize_error(err: impl std::fmt::Display) -> Response {
ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
format!("Failed to serialize stored event: {err}"),
)
.into_response()
})?;
serde_json::from_value(value).map_err(|err| {
ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
format!("Failed to serialize stored event: {err}"),
)
.into_response()
})
}
let value = event.to_wire_value().map_err(serialize_error)?;
serde_json::from_value(value).map_err(serialize_error)
}
fn clear_live_run_state(run: &mut ManagedRun) {

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

@ -595,6 +595,7 @@ impl RunEvent {
let obj = value.as_object().ok_or_else(|| {
<serde_json::Error as DeError>::custom("run event must be a JSON object")
})?;
let opt_str = |key: &str| obj.get(key).and_then(Value::as_str).map(str::to_string);
let id = obj.get("id").and_then(Value::as_str).ok_or_else(|| {
<serde_json::Error as DeError>::custom("missing or non-string field: id")
})?;
@ -621,38 +622,14 @@ impl RunEvent {
id: id.to_string(),
ts,
run_id,
node_id: obj
.get("node_id")
.and_then(Value::as_str)
.map(str::to_string),
node_label: obj
.get("node_label")
.and_then(Value::as_str)
.map(str::to_string),
stage_id: obj
.get("stage_id")
.and_then(Value::as_str)
.map(str::to_string),
parallel_group_id: obj
.get("parallel_group_id")
.and_then(Value::as_str)
.map(str::to_string),
parallel_branch_id: obj
.get("parallel_branch_id")
.and_then(Value::as_str)
.map(str::to_string),
session_id: obj
.get("session_id")
.and_then(Value::as_str)
.map(str::to_string),
parent_session_id: obj
.get("parent_session_id")
.and_then(Value::as_str)
.map(str::to_string),
tool_call_id: obj
.get("tool_call_id")
.and_then(Value::as_str)
.map(str::to_string),
node_id: opt_str("node_id"),
node_label: opt_str("node_label"),
stage_id: opt_str("stage_id"),
parallel_group_id: opt_str("parallel_group_id"),
parallel_branch_id: opt_str("parallel_branch_id"),
session_id: opt_str("session_id"),
parent_session_id: opt_str("parent_session_id"),
tool_call_id: opt_str("tool_call_id"),
actor,
event,
properties: &properties,

85
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::{
BilledTokenCounts, RunBlobId, RunControlAction, RunEvent, RunId, StageId, StageStatus,
StatusReason,
ActorKind, ActorRef, BilledTokenCounts, RunBlobId, RunControlAction, RunEvent, RunId,
RunProvenance, StageId, StageStatus, StatusReason,
};
use anyhow::{Context, Result};
use chrono::Utc;
@ -55,7 +55,7 @@ pub enum Event {
#[serde(default, skip_serializing_if = "Option::is_none")]
db_prefix: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
provenance: Option<::fabro_types::RunProvenance>,
provenance: Option<RunProvenance>,
#[serde(default, skip_serializing_if = "Option::is_none")]
manifest_blob: Option<RunBlobId>,
},
@ -1289,13 +1289,22 @@ struct StoredEventFields {
parallel_group_id: Option<String>,
parallel_branch_id: Option<String>,
tool_call_id: Option<String>,
actor: Option<::fabro_types::ActorRef>,
actor: Option<ActorRef>,
}
fn default_node_label(node_id: Option<&String>, node_label: Option<String>) -> Option<String> {
node_label.or_else(|| node_id.cloned())
}
fn node_stored_fields(node_id: Option<String>) -> StoredEventFields {
let node_label = default_node_label(node_id.as_ref(), None);
StoredEventFields {
node_id,
node_label,
..StoredEventFields::default()
}
}
fn billed_token_counts_from_llm(usage: &LlmTokenCounts) -> BilledTokenCounts {
BilledTokenCounts {
input_tokens: usage.input_tokens,
@ -1383,15 +1392,7 @@ fn stored_event_fields(event: &Event) -> StoredEventFields {
| Event::CommandStarted { node_id, .. }
| Event::CommandCompleted { node_id, .. }
| Event::AgentCliStarted { node_id, .. }
| Event::AgentCliCompleted { node_id, .. } => {
let node_id = Some(node_id.clone());
let node_label = default_node_label(node_id.as_ref(), None);
StoredEventFields {
node_id,
node_label,
..StoredEventFields::default()
}
}
| Event::AgentCliCompleted { node_id, .. } => node_stored_fields(Some(node_id.clone())),
Event::Agent {
stage,
visit,
@ -1416,17 +1417,9 @@ fn stored_event_fields(event: &Event) -> StoredEventFields {
parallel_branch_id: parallel_branch_id.clone(),
tool_call_id,
actor,
..StoredEventFields::default()
}
}
Event::GitCommit { node_id, .. } => {
let node_label = default_node_label(node_id.as_ref(), None);
StoredEventFields {
node_id: node_id.clone(),
node_label,
..StoredEventFields::default()
}
}
Event::GitCommit { node_id, .. } => node_stored_fields(node_id.clone()),
Event::ParallelBranchStarted {
parallel_group_id,
parallel_branch_id,
@ -1453,35 +1446,16 @@ fn stored_event_fields(event: &Event) -> StoredEventFields {
| Event::InterviewStarted { stage, .. }
| Event::InterviewTimeout { stage, .. }
| Event::InterviewInterrupted { stage, .. }
| Event::Failover { stage, .. } => {
let node_id = Some(stage.clone());
let node_label = default_node_label(node_id.as_ref(), None);
StoredEventFields {
node_id,
node_label,
..StoredEventFields::default()
}
}
Event::StallWatchdogTimeout { node, .. } => {
let node_id = Some(node.clone());
let node_label = default_node_label(node_id.as_ref(), None);
StoredEventFields {
node_id,
node_label,
..StoredEventFields::default()
}
}
| Event::Failover { stage, .. } => node_stored_fields(Some(stage.clone())),
Event::StallWatchdogTimeout { node, .. } => node_stored_fields(Some(node.clone())),
_ => StoredEventFields::default(),
}
}
fn actor_from_provenance(
provenance: &::fabro_types::RunProvenance,
) -> Option<::fabro_types::ActorRef> {
let subject = provenance.subject.as_ref()?;
let login = subject.login.clone()?;
Some(::fabro_types::ActorRef {
kind: ::fabro_types::ActorKind::User,
fn actor_from_provenance(provenance: &RunProvenance) -> Option<ActorRef> {
let login = provenance.subject.as_ref()?.login.clone()?;
Some(ActorRef {
kind: ActorKind::User,
id: Some(login.clone()),
display: Some(login),
})
@ -1495,13 +1469,10 @@ fn agent_tool_call_id(event: &AgentEvent) -> Option<&str> {
}
}
fn agent_actor_for_event(
event: &AgentEvent,
session_id: Option<&str>,
) -> Option<::fabro_types::ActorRef> {
fn agent_actor_for_event(event: &AgentEvent, session_id: Option<&str>) -> Option<ActorRef> {
match event {
AgentEvent::AssistantMessage { model, .. } => Some(::fabro_types::ActorRef {
kind: ::fabro_types::ActorKind::Agent,
AgentEvent::AssistantMessage { model, .. } => Some(ActorRef {
kind: ActorKind::Agent,
id: session_id.map(str::to_string),
display: Some(model.clone()),
}),
@ -3281,16 +3252,14 @@ mod tests {
},
);
let actor = stored.actor.as_ref().expect("actor set");
assert_eq!(actor.kind, ::fabro_types::ActorKind::Agent);
assert_eq!(actor.kind, ActorKind::Agent);
assert_eq!(actor.id.as_deref(), Some("ses_agent"));
assert_eq!(actor.display.as_deref(), Some("claude-sonnet"));
}
#[test]
fn run_created_populates_user_actor_from_provenance() {
use ::fabro_types::{
Graph, RunAuthMethod, RunProvenance, RunSubjectProvenance, Settings, fixtures,
};
use ::fabro_types::{Graph, RunAuthMethod, RunSubjectProvenance, Settings, fixtures};
let provenance = RunProvenance {
server: None,
@ -3322,7 +3291,7 @@ mod tests {
},
);
let actor = stored.actor.as_ref().expect("actor set");
assert_eq!(actor.kind, ::fabro_types::ActorKind::User);
assert_eq!(actor.kind, ActorKind::User);
assert_eq!(actor.id.as_deref(), Some("alice"));
assert_eq!(actor.display.as_deref(), Some("alice"));
}

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

@ -37,10 +37,6 @@ fn build_profile(model: &str, provider: Provider) -> Box<dyn AgentProfile> {
}
}
fn current_visit(context: &Context) -> u32 {
u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX)
}
#[derive(Clone)]
struct StageEventScope {
visit: u32,
@ -50,7 +46,7 @@ struct StageEventScope {
fn current_stage_event_scope(context: &Context) -> StageEventScope {
StageEventScope {
visit: current_visit(context),
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(),
}