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 1

Fix envelope event verification regressions

Browse source
This commit is contained in:
Bryan Helmkamp 2026-03-30 17:38:57 -04:00
parent f43fac10ce
commit 1c85dbb367
No known key found for this signature in database
16 changed files with 189 additions and 126 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

80
lib/crates/fabro-cli/src/commands/run/logs.rs
View file

@ -564,16 +564,21 @@ pub(crate) fn format_event_pretty(line: &str, styles: &Styles) -> Option<String>
))
}
"setup.completed" => {
let count = prop_field(&envelope, "command_count")
.and_then(serde_json::Value::as_u64)
.unwrap_or(0);
let count = prop_field(&envelope, "command_count").and_then(serde_json::Value::as_u64);
let duration = format_duration_ms(prop_field(&envelope, "duration_ms"));
Some(format!(
"{} Setup: {} commands {}",
styles.dim.apply_to(&ts),
count,
styles.dim.apply_to(&duration),
))
Some(match count {
Some(count) => format!(
"{} Setup: {} commands {}",
styles.dim.apply_to(&ts),
count,
styles.dim.apply_to(&duration),
),
None => format!(
"{} Setup: {}",
styles.dim.apply_to(&ts),
styles.dim.apply_to(&duration),
),
})
}
"agent.compaction.completed" => {
let original = prop_field(&envelope, "original_turn_count")
@ -839,9 +844,9 @@ mod tests {
fn since_filters_by_timestamp() {
let cutoff = "2026-01-01T12:00:00Z".parse::<DateTime<Utc>>().unwrap();
let lines = vec![
r#"{"ts":"2026-01-01T11:00:00Z","event":"StageStarted"}"#.to_string(),
r#"{"ts":"2026-01-01T12:30:00Z","event":"StageCompleted"}"#.to_string(),
r#"{"ts":"2026-01-01T13:00:00Z","event":"WorkflowRunCompleted"}"#.to_string(),
r#"{"ts":"2026-01-01T11:00:00Z","event":"stage.started"}"#.to_string(),
r#"{"ts":"2026-01-01T12:30:00Z","event":"stage.completed"}"#.to_string(),
r#"{"ts":"2026-01-01T13:00:00Z","event":"run.completed"}"#.to_string(),
];
let result = apply_filters(&lines, Some(&cutoff), None);
assert_eq!(result.len(), 2);
@ -850,7 +855,7 @@ mod tests {
#[test]
fn raw_lines_pass_through_verbatim() {
let lines = vec![
r#"{"ts":"2026-01-01T12:00:00Z","event":"StageStarted","node_label":"plan"}"#
r#"{"ts":"2026-01-01T12:00:00Z","event":"stage.started","node_label":"plan"}"#
.to_string(),
];
let result = apply_filters(&lines, None, None);
@ -860,7 +865,7 @@ mod tests {
#[test]
fn pretty_stage_started() {
let styles = no_color_styles();
let line = r#"{"ts":"2026-01-01T14:23:09Z","event":"StageStarted","node_label":"plan","node_id":"plan","stage_index":0}"#;
let line = r#"{"ts":"2026-01-01T14:23:09Z","event":"stage.started","node_label":"plan","node_id":"plan","properties":{"index":0}}"#;
let result = format_event_pretty(line, &styles).unwrap();
assert!(result.contains("plan"), "got: {result}");
assert!(result.contains("\u{25b6}"), "got: {result}");
@ -869,18 +874,18 @@ mod tests {
#[test]
fn pretty_stage_completed() {
let styles = no_color_styles();
let line = r#"{"ts":"2026-01-01T14:23:15Z","event":"StageCompleted","node_label":"plan","cost":0.12,"duration_ms":8000,"turns":3,"tool_calls":2,"total_tokens":15200}"#;
let line = r#"{"ts":"2026-01-01T14:23:15Z","event":"stage.completed","node_label":"plan","properties":{"duration_ms":8000,"status":"success","usage":{"cost":0.12,"input_tokens":10000,"output_tokens":5200}}}"#;
let result = format_event_pretty(line, &styles).unwrap();
assert!(result.contains("plan"), "got: {result}");
assert!(result.contains("$0.12"), "got: {result}");
assert!(result.contains("8s"), "got: {result}");
assert!(result.contains("3 turns"), "got: {result}");
assert!(result.contains("15.2k toks"), "got: {result}");
}
#[test]
fn pretty_assistant_message() {
let styles = no_color_styles();
let line = r#"{"ts":"2026-01-01T14:23:12Z","event":"Agent.AssistantMessage","node_id":"plan","model":"claude-opus-4-6","text":"I'll start by reading the code.","usage":{"input_tokens":100,"output_tokens":50},"tool_call_count":0}"#;
let line = r#"{"ts":"2026-01-01T14:23:12Z","event":"agent.message","node_id":"plan","properties":{"model":"claude-opus-4-6","text":"I'll start by reading the code.","usage":{"input_tokens":100,"output_tokens":50},"tool_call_count":0}}"#;
let result = format_event_pretty(line, &styles).unwrap();
assert!(result.contains("plan"), "got: {result}");
assert!(result.contains("claude-opus-4-6"), "got: {result}");
@ -890,7 +895,7 @@ mod tests {
#[test]
fn pretty_tool_call_started() {
let styles = no_color_styles();
let line = r#"{"ts":"2026-01-01T14:23:12Z","event":"Agent.ToolCallStarted","tool_name":"read_file","tool_call_id":"tc_1","arguments":{"path":"src/main.rs"}}"#;
let line = r#"{"ts":"2026-01-01T14:23:12Z","event":"agent.tool.started","properties":{"tool_name":"read_file","tool_call_id":"tc_1","arguments":{"path":"src/main.rs"}}}"#;
let result = format_event_pretty(line, &styles).unwrap();
assert!(result.contains("read_file"), "got: {result}");
assert!(result.contains("src/main.rs"), "got: {result}");
@ -899,15 +904,14 @@ mod tests {
#[test]
fn pretty_skips_noise_events() {
let styles = no_color_styles();
let line = r#"{"ts":"2026-01-01T14:23:12Z","event":"Agent.TextDelta","delta":"hello"}"#;
let line = r#"{"ts":"2026-01-01T14:23:12Z","event":"agent.text.delta","properties":{"delta":"hello"}}"#;
assert!(format_event_pretty(line, &styles).is_none());
}
#[test]
fn pretty_skips_assistant_output_replace_noise_event() {
let styles = no_color_styles();
let line =
r#"{"ts":"2026-01-01T14:23:12Z","event":"Agent.AssistantOutputReplace","text":""}"#;
let line = r#"{"ts":"2026-01-01T14:23:12Z","event":"agent.output.replace","properties":{"text":""}}"#;
assert!(format_event_pretty(line, &styles).is_none());
}
@ -921,7 +925,7 @@ mod tests {
#[test]
fn pretty_workflow_run_started() {
let styles = no_color_styles();
let line = r#"{"ts":"2026-01-01T14:23:01Z","run_id":"abc123","event":"WorkflowRunStarted","workflow_name":"smoke"}"#;
let line = r#"{"ts":"2026-01-01T14:23:01Z","run_id":"abc123","event":"run.started","properties":{"name":"smoke"}}"#;
let result = format_event_pretty(line, &styles).unwrap();
assert!(result.contains("smoke"), "got: {result}");
assert!(result.contains("abc123"), "got: {result}");
@ -930,7 +934,7 @@ mod tests {
#[test]
fn pretty_workflow_run_started_with_goal() {
let styles = no_color_styles();
let line = r#"{"ts":"2026-01-01T14:23:01Z","run_id":"abc123","event":"WorkflowRunStarted","workflow_name":"smoke","goal":"Fix the bug"}"#;
let line = r#"{"ts":"2026-01-01T14:23:01Z","run_id":"abc123","event":"run.started","properties":{"name":"smoke","goal":"Fix the bug"}}"#;
let result = format_event_pretty(line, &styles).unwrap();
assert!(result.contains("smoke"), "got: {result}");
assert!(result.contains("abc123"), "got: {result}");
@ -941,7 +945,7 @@ mod tests {
#[test]
fn pretty_workflow_run_started_without_goal_no_extra_lines() {
let styles = no_color_styles();
let line = r#"{"ts":"2026-01-01T14:23:01Z","run_id":"abc123","event":"WorkflowRunStarted","workflow_name":"smoke"}"#;
let line = r#"{"ts":"2026-01-01T14:23:01Z","run_id":"abc123","event":"run.started","properties":{"name":"smoke"}}"#;
let result = format_event_pretty(line, &styles).unwrap();
assert!(!result.contains('\n'), "got: {result}");
}
@ -949,7 +953,7 @@ mod tests {
#[test]
fn pretty_workflow_run_completed() {
let styles = no_color_styles();
let line = r#"{"ts":"2026-01-01T14:23:32Z","run_id":"abc123","event":"WorkflowRunCompleted","duration_ms":25000,"status":"success","total_cost":0.57,"usage":{"input_tokens":5000,"output_tokens":2000,"total_tokens":7000,"cache_read_tokens":3000,"cache_write_tokens":500,"reasoning_tokens":800}}"#;
let line = r#"{"ts":"2026-01-01T14:23:32Z","run_id":"abc123","event":"run.completed","properties":{"duration_ms":25000,"status":"success","total_cost":0.57,"usage":{"input_tokens":5000,"output_tokens":2000,"total_tokens":7000,"cache_read_tokens":3000,"cache_write_tokens":500,"reasoning_tokens":800}}}"#;
let result = format_event_pretty(line, &styles).unwrap();
assert!(result.contains("SUCCESS"), "got: {result}");
assert!(result.contains("25s"), "got: {result}");
@ -963,7 +967,7 @@ mod tests {
#[test]
fn pretty_workflow_run_completed_backward_compat() {
let styles = no_color_styles();
let line = r#"{"ts":"2026-01-01T14:23:32Z","run_id":"abc123","event":"WorkflowRunCompleted","duration_ms":25000,"total_cost":0.57}"#;
let line = r#"{"ts":"2026-01-01T14:23:32Z","run_id":"abc123","event":"run.completed","properties":{"duration_ms":25000,"total_cost":0.57}}"#;
let result = format_event_pretty(line, &styles).unwrap();
assert!(result.contains("SUCCESS"), "got: {result}");
assert!(result.contains("25s"), "got: {result}");
@ -974,7 +978,7 @@ mod tests {
#[test]
fn pretty_workflow_run_completed_fail_status() {
let styles = no_color_styles();
let line = r#"{"ts":"2026-01-01T14:23:32Z","event":"WorkflowRunCompleted","duration_ms":25000,"status":"fail"}"#;
let line = r#"{"ts":"2026-01-01T14:23:32Z","event":"run.completed","properties":{"duration_ms":25000,"status":"fail"}}"#;
let result = format_event_pretty(line, &styles).unwrap();
assert!(result.contains("FAIL"), "got: {result}");
}
@ -982,7 +986,7 @@ mod tests {
#[test]
fn pretty_pull_request_created() {
let styles = no_color_styles();
let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"PullRequestCreated","pr_url":"https://github.com/owner/repo/pull/42","pr_number":42,"draft":false}"#;
let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"pull_request.created","properties":{"pr_url":"https://github.com/owner/repo/pull/42","pr_number":42,"draft":false}}"#;
let result = format_event_pretty(line, &styles).unwrap();
assert!(result.contains("PR:"), "got: {result}");
assert!(
@ -994,7 +998,7 @@ mod tests {
#[test]
fn pretty_pull_request_created_draft() {
let styles = no_color_styles();
let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"PullRequestCreated","pr_url":"https://github.com/owner/repo/pull/42","pr_number":42,"draft":true}"#;
let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"pull_request.created","properties":{"pr_url":"https://github.com/owner/repo/pull/42","pr_number":42,"draft":true}}"#;
let result = format_event_pretty(line, &styles).unwrap();
assert!(result.contains("Draft PR:"), "got: {result}");
}
@ -1002,7 +1006,7 @@ mod tests {
#[test]
fn pretty_pull_request_failed() {
let styles = no_color_styles();
let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"PullRequestFailed","error":"auth token expired"}"#;
let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"pull_request.failed","properties":{"error":"auth token expired"}}"#;
let result = format_event_pretty(line, &styles).unwrap();
assert!(result.contains("PR failed:"), "got: {result}");
assert!(result.contains("auth token expired"), "got: {result}");
@ -1011,7 +1015,7 @@ mod tests {
#[test]
fn pretty_run_notice_warn() {
let styles = no_color_styles();
let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"RunNotice","level":"warn","code":"sandbox_cleanup_failed","message":"sandbox cleanup failed: boom"}"#;
let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"run.notice","properties":{"level":"warn","code":"sandbox_cleanup_failed","message":"sandbox cleanup failed: boom"}}"#;
let result = format_event_pretty(line, &styles).unwrap();
assert!(result.contains("Warning:"), "got: {result}");
assert!(
@ -1024,7 +1028,7 @@ mod tests {
#[test]
fn pretty_run_notice_error() {
let styles = no_color_styles();
let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"RunNotice","level":"error","code":"launch_failed","message":"failed to start engine"}"#;
let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"run.notice","properties":{"level":"error","code":"launch_failed","message":"failed to start engine"}}"#;
let result = format_event_pretty(line, &styles).unwrap();
assert!(result.contains("Error:"), "got: {result}");
assert!(result.contains("failed to start engine"), "got: {result}");
@ -1034,12 +1038,22 @@ mod tests {
#[test]
fn pretty_workflow_run_failed() {
let styles = no_color_styles();
let line = r#"{"ts":"2026-01-01T14:23:32Z","run_id":"abc123","event":"WorkflowRunFailed","error":"sandbox timeout"}"#;
let line = r#"{"ts":"2026-01-01T14:23:32Z","run_id":"abc123","event":"run.failed","properties":{"error":"sandbox timeout"}}"#;
let result = format_event_pretty(line, &styles).unwrap();
assert!(result.contains("Failed"), "got: {result}");
assert!(result.contains("sandbox timeout"), "got: {result}");
}
#[test]
fn pretty_setup_completed_without_command_count() {
let styles = no_color_styles();
let line = r#"{"ts":"2026-01-01T14:23:32Z","event":"setup.completed","properties":{"duration_ms":800}}"#;
let result = format_event_pretty(line, &styles).unwrap();
assert!(result.contains("Setup:"), "got: {result}");
assert!(result.contains("800ms"), "got: {result}");
assert!(!result.contains("0 commands"), "got: {result}");
}
#[test]
fn format_duration_ms_subsecond() {
assert_eq!(format_duration_ms(Some(&serde_json::json!(500))), "500ms");

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

@ -570,9 +570,11 @@ mod tests {
#[test]
fn parse_edge_selected() {
let fields = json_map(serde_json::json!({
"from_node_id": "a",
"to_node_id": "b",
"label": "yes"
"properties": {
"from_node": "a",
"to_node": "b",
"label": "yes"
}
}));
let event = from_envelope_fields("edge.selected", &fields).unwrap();

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

@ -486,6 +486,7 @@ mod tests {
fn event_payload(run_id: RunId, ts: &str, event: &str) -> EventPayload {
EventPayload::new(
serde_json::json!({
"id": format!("evt-{run_id}-{event}"),
"ts": ts,
"run_id": run_id.to_string(),
"event": event
@ -540,14 +541,14 @@ mod tests {
run.append_event(&event_payload(
run_id,
"2026-03-27T12:00:00.000Z",
"WorkflowRunStarted",
"run.started",
))
.await
.unwrap();
run.append_event(&event_payload(
run_id,
"2026-03-27T12:00:01.000Z",
"StageCompleted",
"stage.completed",
))
.await
.unwrap();

84
lib/crates/fabro-cli/tests/it/cmd/run.rs
View file

@ -104,125 +104,125 @@ fn dry_run_writes_jsonl_and_live_json() {
fabro_json_snapshot!(context, &progress_summary, @r#"
[
{
"event": "Sandbox.Initializing",
"sandbox_provider": "local"
"event": "sandbox.initializing",
"provider": "local"
},
{
"event": "Sandbox.Ready",
"sandbox_provider": "local"
"event": "sandbox.ready",
"provider": "local"
},
{
"event": "SandboxInitialized"
"event": "sandbox.initialized"
},
{
"event": "WorkflowRunStarted",
"workflow_name": "Simple",
"event": "run.started",
"name": "Simple",
"goal": "Run tests and report results"
},
{
"event": "StageStarted",
"event": "stage.started",
"node_id": "start",
"node_label": "Start",
"handler_type": "start",
"stage_index": 0
"index": 0
},
{
"event": "StageCompleted",
"event": "stage.completed",
"node_id": "start",
"node_label": "Start",
"stage_index": 0,
"index": 0,
"status": "success"
},
{
"event": "EdgeSelected",
"from_node_id": "start",
"to_node_id": "run_tests",
"event": "edge.selected",
"from_node": "start",
"to_node": "run_tests",
"reason": "unconditional"
},
{
"event": "CheckpointCompleted",
"event": "checkpoint.completed",
"node_id": "start",
"node_label": "start",
"status": "success"
},
{
"event": "StageStarted",
"event": "stage.started",
"node_id": "run_tests",
"node_label": "Run Tests",
"handler_type": "agent",
"stage_index": 1
"index": 1
},
{
"event": "StageCompleted",
"event": "stage.completed",
"node_id": "run_tests",
"node_label": "Run Tests",
"stage_index": 1,
"index": 1,
"status": "success"
},
{
"event": "EdgeSelected",
"from_node_id": "run_tests",
"to_node_id": "report",
"event": "edge.selected",
"from_node": "run_tests",
"to_node": "report",
"reason": "unconditional"
},
{
"event": "CheckpointCompleted",
"event": "checkpoint.completed",
"node_id": "run_tests",
"node_label": "run_tests",
"status": "success"
},
{
"event": "StageStarted",
"event": "stage.started",
"node_id": "report",
"node_label": "Report",
"handler_type": "agent",
"stage_index": 2
"index": 2
},
{
"event": "StageCompleted",
"event": "stage.completed",
"node_id": "report",
"node_label": "Report",
"stage_index": 2,
"index": 2,
"status": "success"
},
{
"event": "EdgeSelected",
"from_node_id": "report",
"to_node_id": "exit",
"event": "edge.selected",
"from_node": "report",
"to_node": "exit",
"reason": "unconditional"
},
{
"event": "CheckpointCompleted",
"event": "checkpoint.completed",
"node_id": "report",
"node_label": "report",
"status": "success"
},
{
"event": "StageStarted",
"event": "stage.started",
"node_id": "exit",
"node_label": "Exit",
"handler_type": "exit",
"stage_index": 3
"index": 3
},
{
"event": "StageCompleted",
"event": "stage.completed",
"node_id": "exit",
"node_label": "Exit",
"stage_index": 3,
"index": 3,
"status": "success"
},
{
"event": "WorkflowRunCompleted",
"event": "run.completed",
"status": "success",
"artifact_count": 0
},
{
"event": "Sandbox.CleanupStarted",
"sandbox_provider": "local"
"event": "sandbox.cleanup.started",
"provider": "local"
},
{
"event": "Sandbox.CleanupCompleted",
"sandbox_provider": "local"
"event": "sandbox.cleanup.completed",
"provider": "local"
}
]
"#);
@ -232,8 +232,8 @@ fn dry_run_writes_jsonl_and_live_json() {
let live_summary = compact_progress_event(&live_content);
fabro_json_snapshot!(context, &live_summary, @r#"
{
"event": "Sandbox.CleanupCompleted",
"sandbox_provider": "local"
"event": "sandbox.cleanup.completed",
"provider": "local"
}
"#);

9
lib/crates/fabro-cli/tests/it/scenario/recovery.rs
View file

@ -59,14 +59,19 @@ fn latest_metadata_checkpoint(repo_dir: &Path, run_id: &str) -> Checkpoint {
fn run_commit_shas_by_node(run_dir: &Path) -> serde_json::Map<String, serde_json::Value> {
let mut shas_by_node = serde_json::Map::new();
for event in read_jsonl(run_dir.join("progress.jsonl")) {
if event["event"].as_str() != Some("GitCommit") {
if !matches!(event["event"].as_str(), Some("git.commit" | "GitCommit")) {
continue;
}
let Some(node_id) = event["node_id"].as_str() else {
continue;
};
let Some(sha) = event["sha"].as_str() else {
let Some(sha) = event
.get("properties")
.and_then(|properties| properties.get("sha"))
.and_then(serde_json::Value::as_str)
.or_else(|| event["sha"].as_str())
else {
continue;
};

23
lib/crates/fabro-cli/tests/it/support/mod.rs
View file

@ -41,23 +41,34 @@ pub(crate) fn read_jsonl(path: impl AsRef<Path>) -> Vec<Value> {
}
pub(crate) fn compact_progress_event(event: &Value) -> Value {
fn event_value<'a>(event: &'a Value, key: &str) -> Option<&'a Value> {
event
.get(key)
.or_else(|| {
event
.get("properties")
.and_then(|properties| properties.get(key))
})
.filter(|value| !value.is_null())
}
let mut compact = serde_json::Map::new();
for key in [
"event",
"sandbox_provider",
"workflow_name",
"provider",
"name",
"goal",
"node_id",
"node_label",
"handler_type",
"stage_index",
"index",
"status",
"from_node_id",
"to_node_id",
"from_node",
"to_node",
"reason",
"artifact_count",
] {
if let Some(value) = event.get(key).filter(|value| !value.is_null()) {
if let Some(value) = event_value(event, key) {
compact.insert(key.to_string(), value.clone());
}
}

8
lib/crates/fabro-cli/tests/it/workflow/full_stack.rs
View file

@ -51,12 +51,12 @@ fn scenario_full_stack(sandbox: &str) {
// Progress events
assert!(
has_event(&run_dir, "WorkflowRunStarted"),
"progress should contain WorkflowRunStarted"
has_event(&run_dir, "run.started"),
"progress should contain run.started"
);
assert!(
has_event(&run_dir, "WorkflowRunCompleted"),
"progress should contain WorkflowRunCompleted"
has_event(&run_dir, "run.completed"),
"progress should contain run.completed"
);
// All expected nodes completed

1
lib/crates/fabro-store/src/disk_projecting.rs
View file

@ -512,6 +512,7 @@ mod tests {
fn event_payload(run_id: &str, ts: &str, event: &str) -> EventPayload {
EventPayload::new(
serde_json::json!({
"id": format!("evt-{run_id}-{event}"),
"ts": ts,
"run_id": test_run_id(run_id).to_string(),
"event": event,

3
lib/crates/fabro-store/src/memory.rs
View file

@ -986,6 +986,7 @@ mod tests {
assert!(matches!(err, StoreError::InvalidEvent(_)));
let invalid_run_id: EventPayload = serde_json::from_value(serde_json::json!({
"id": "evt-invalid-run",
"ts": "2026-03-27T12:00:00Z",
"run_id": "other-run",
"event": "StageStarted"
@ -1022,6 +1023,7 @@ mod tests {
.unwrap();
let first = EventPayload::new(
serde_json::json!({
"id": "evt-1",
"ts": "2026-03-27T12:00:00.000Z",
"run_id": test_run_id("run-1").to_string(),
"event": "WorkflowRunStarted"
@ -1031,6 +1033,7 @@ mod tests {
.unwrap();
let second = EventPayload::new(
serde_json::json!({
"id": "evt-2",
"ts": "2026-03-27T12:00:01.000Z",
"run_id": test_run_id("run-1").to_string(),
"event": "StageCompleted"

1
lib/crates/fabro-store/src/slate/mod.rs
View file

@ -496,6 +496,7 @@ mod tests {
fn event_payload(run_id: &str, ts: &str, event: &str) -> EventPayload {
EventPayload::new(
serde_json::json!({
"id": format!("evt-{run_id}-{event}"),
"ts": ts,
"run_id": test_run_id(run_id).to_string(),
"event": event

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

@ -1326,6 +1326,11 @@ impl EventEmitter {
}
}
#[must_use]
pub fn run_id(&self) -> RunId {
self.run_id
}
pub fn on_event(&self, listener: impl Fn(&RunEventEnvelope) + Send + Sync + 'static) {
self.listeners
.lock()
@ -1343,6 +1348,11 @@ impl EventEmitter {
);
}
let envelope = canonicalize_event(&self.run_id, event);
self.dispatch_envelope(&envelope);
}
pub(crate) fn dispatch_envelope(&self, envelope: &RunEventEnvelope) {
self.last_event_at.store(epoch_millis(), Ordering::Relaxed);
// Clone the listener list so we don't hold the lock during dispatch.
// This prevents deadlocks if a listener calls emit() reentrantly.
// Note: listeners added during this emit() won't receive the current event.
@ -1538,7 +1548,7 @@ mod tests {
);
assert_eq!(envelope.event, "run.failed");
assert_eq!(envelope.properties["error"], "boom");
assert_eq!(envelope.properties["error"], "Handler error: boom");
assert_eq!(envelope.properties["duration_ms"], 900);
}

13
lib/crates/fabro-workflow/src/handler/manager_loop.rs
View file

@ -4,12 +4,6 @@ use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use async_trait::async_trait;
use chrono::Utc;
use fabro_config::FabroSettings;
use fabro_store::{InMemoryStore, Store};
use fabro_types::RunId;
use crate::condition::evaluate_condition;
use crate::context::keys;
use crate::context::{Context, WorkflowContext};
@ -20,7 +14,11 @@ use crate::pipeline;
use crate::pipeline::types::Initialized;
use crate::run_dir::visit_from_context;
use crate::run_options::RunOptions;
use async_trait::async_trait;
use chrono::Utc;
use fabro_config::FabroSettings;
use fabro_graphviz::graph::{AttrValue, Graph, Node};
use fabro_store::{InMemoryStore, Store};
use tokio::time::{sleep, timeout};
use super::{EngineServices, Handler};
@ -166,7 +164,8 @@ impl Handler for SubWorkflowHandler {
settings: fabro_config::FabroSettings::default(),
run_dir: child_logs,
cancel_token: Some(cancel_token),
run_id: RunId::new(),
// Child workflows are part of the parent run's event stream.
run_id: services.emitter.run_id(),
labels: HashMap::new(),
workflow_slug: None,
github_app: None,

12
lib/crates/fabro-workflow/src/operations/start.rs
View file

@ -974,7 +974,7 @@ mod tests {
async fn start_captures_checkpoint_git_sha_in_conclusion() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
let emitter = Arc::new(EventEmitter::default());
let emitter = Arc::new(EventEmitter::new(fixtures::RUN_1));
let registry = Arc::new(test_registry());
let injected = Arc::new(AtomicBool::new(false));
@ -1016,7 +1016,7 @@ mod tests {
async fn start_loads_persisted_from_run_dir() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
let emitter = Arc::new(EventEmitter::default());
let emitter = Arc::new(EventEmitter::new(fixtures::RUN_1));
let registry = Arc::new(test_registry());
persisted_workflow(MINIMAL_DOT, &run_dir);
@ -1036,7 +1036,7 @@ mod tests {
async fn start_invokes_on_node_callback_before_execution() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
let emitter = Arc::new(EventEmitter::default());
let emitter = Arc::new(EventEmitter::new(fixtures::RUN_1));
let registry = Arc::new(test_registry());
let visited = Arc::new(Mutex::new(Vec::new()));
@ -1065,7 +1065,7 @@ mod tests {
async fn start_errors_when_checkpoint_exists() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
let emitter = Arc::new(EventEmitter::default());
let emitter = Arc::new(EventEmitter::new(fixtures::RUN_1));
let registry = Arc::new(test_registry());
persisted_workflow(MINIMAL_DOT, &run_dir);
@ -1102,7 +1102,7 @@ mod tests {
async fn resume_errors_when_checkpoint_missing() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
let emitter = Arc::new(EventEmitter::default());
let emitter = Arc::new(EventEmitter::new(fixtures::RUN_1));
let registry = Arc::new(test_registry());
persisted_workflow(MINIMAL_DOT, &run_dir);
@ -1124,7 +1124,7 @@ mod tests {
async fn resume_errors_when_run_already_finished_successfully() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
let emitter = Arc::new(EventEmitter::default());
let emitter = Arc::new(EventEmitter::new(fixtures::RUN_1));
let registry = Arc::new(test_registry());
persisted_workflow(MINIMAL_DOT, &run_dir);

42
lib/crates/fabro-workflow/src/pipeline/execute/tests.rs
View file

@ -77,6 +77,14 @@ fn test_run_id(label: &str) -> RunId {
}
}
fn test_emitter(label: &str) -> EventEmitter {
EventEmitter::new(test_run_id(label))
}
fn test_emitter_arc(label: &str) -> Arc<EventEmitter> {
Arc::new(test_emitter(label))
}
fn test_run_options(run_dir: &Path, run_id: &str) -> RunOptions {
RunOptions {
run_dir: run_dir.to_path_buf(),
@ -169,7 +177,7 @@ async fn execute_runs_start_to_exit_and_returns_final_context() {
run_id: test_run_id("run-test"),
run_store: test_run_store(&run_dir, &test_run_id("run-test")).await,
dry_run: false,
emitter: Arc::new(crate::event::EventEmitter::default()),
emitter: test_emitter_arc("run-test"),
sandbox: SandboxSpec::Local {
working_directory: std::env::current_dir().unwrap(),
},
@ -426,7 +434,7 @@ async fn execute_runs_simple_workflow() {
let dir = tempfile::tempdir().unwrap();
let outcome = run_graph(
make_registry(),
Arc::new(EventEmitter::default()),
test_emitter_arc("test-run"),
local_env(),
&simple_graph(),
&test_run_options(dir.path(), "test-run"),
@ -441,7 +449,7 @@ async fn execute_saves_checkpoint() {
let dir = tempfile::tempdir().unwrap();
run_graph(
make_registry(),
Arc::new(EventEmitter::default()),
test_emitter_arc("test-run"),
local_env(),
&simple_graph(),
&test_run_options(dir.path(), "test-run"),
@ -456,7 +464,7 @@ async fn execute_emits_events() {
let dir = tempfile::tempdir().unwrap();
let events = Arc::new(std::sync::Mutex::new(Vec::new()));
let events_clone = Arc::clone(&events);
let emitter = EventEmitter::default();
let emitter = test_emitter("test-run");
emitter.on_event(move |event| {
events_clone.lock().unwrap().push(format!("{event:?}"));
});
@ -479,7 +487,7 @@ async fn execute_error_when_no_start_node() {
let dir = tempfile::tempdir().unwrap();
let result = run_graph(
make_registry(),
Arc::new(EventEmitter::default()),
test_emitter_arc("test-run"),
local_env(),
&Graph::new("empty"),
&test_run_options(dir.path(), "test-run"),
@ -493,7 +501,7 @@ async fn execute_mirrors_graph_goal_to_context() {
let dir = tempfile::tempdir().unwrap();
run_graph(
make_registry(),
Arc::new(EventEmitter::default()),
test_emitter_arc("test-run"),
local_env(),
&simple_graph(),
&test_run_options(dir.path(), "test-run"),
@ -542,7 +550,7 @@ async fn execute_conditional_routing_uses_unconditional_success_path() {
run_graph(
make_registry(),
Arc::new(EventEmitter::default()),
test_emitter_arc("test-run"),
local_env(),
&g,
&test_run_options(dir.path(), "test-run"),
@ -567,7 +575,7 @@ async fn execute_writes_start_json_and_node_status() {
run_graph(
make_registry(),
Arc::new(EventEmitter::default()),
test_emitter_arc("test-run"),
local_env(),
&simple_graph(),
&run_options,
@ -629,7 +637,7 @@ async fn timeout_causes_fail_status_json() {
registry.register("slow", Box::new(SlowHandler { sleep_ms: 500 }));
run_graph(
registry,
Arc::new(EventEmitter::default()),
test_emitter_arc("test-run"),
local_env(),
&g,
&test_run_options(dir.path(), "test-run"),
@ -671,7 +679,7 @@ async fn execute_cancelled_mid_run() {
let result = run_graph(
registry,
Arc::new(EventEmitter::default()),
test_emitter_arc("test-run"),
local_env(),
&g,
&run_options,
@ -689,7 +697,7 @@ async fn max_node_visits_errors_on_cycle() {
let result = run_graph(
make_registry(),
Arc::new(EventEmitter::default()),
test_emitter_arc("test-run"),
local_env(),
&g,
&test_run_options(dir.path(), "test-run"),
@ -724,7 +732,7 @@ async fn panic_handler_writes_panic_txt() {
registry.register("panicker", Box::new(PanickingHandler));
let _ = run_graph(
registry,
Arc::new(EventEmitter::default()),
test_emitter_arc("test-run"),
local_env(),
&g,
&test_run_options(dir.path(), "test-run"),
@ -745,7 +753,7 @@ async fn loop_circuit_breaker_aborts_on_repeated_failure() {
let result = run_graph(
registry,
Arc::new(EventEmitter::default()),
test_emitter_arc("test-run"),
local_env(),
&looping_fail_graph(),
&test_run_options(dir.path(), "test-run"),
@ -794,7 +802,7 @@ async fn stall_watchdog_triggers_on_hung_handler() {
registry.register("slow", Box::new(SlowHandler { sleep_ms: 60_000 }));
let result = run_graph(
registry,
Arc::new(EventEmitter::default()),
test_emitter_arc("test-run"),
local_env(),
&g,
&test_run_options(dir.path(), "test-run"),
@ -843,7 +851,7 @@ async fn retry_emits_stage_started_per_attempt() {
let events = Arc::new(std::sync::Mutex::new(Vec::<RunEventEnvelope>::new()));
let events_clone = Arc::clone(&events);
let emitter = EventEmitter::default();
let emitter = test_emitter("retry-events-test");
emitter.on_event(move |event| {
events_clone.lock().unwrap().push(event.clone());
});
@ -883,7 +891,7 @@ async fn run_with_lifecycle_emits_initialize_and_setup_events() {
let dir = tempfile::tempdir().unwrap();
let events = Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
let events_clone = Arc::clone(&events);
let emitter = EventEmitter::default();
let emitter = test_emitter("order-test");
emitter.on_event(move |event| {
let name = match event.event.as_str() {
"sandbox.initialized" => "SandboxInitialized",
@ -965,7 +973,7 @@ async fn git_checkpoint_skips_start_node() {
let events = Arc::new(std::sync::Mutex::new(Vec::<RunEventEnvelope>::new()));
let events_clone = Arc::clone(&events);
let emitter = EventEmitter::default();
let emitter = test_emitter("git-cp-test");
emitter.on_event(move |event| {
events_clone.lock().unwrap().push(event.clone());
});

8
lib/crates/fabro-workflow/src/test_support.rs
View file

@ -21,6 +21,13 @@ struct InitializedOptions {
checkpoint: Option<Checkpoint>,
}
fn bound_emitter(run_id: fabro_types::RunId, observer: Arc<EventEmitter>) -> Arc<EventEmitter> {
let emitter = Arc::new(EventEmitter::new(run_id));
let observer_clone = Arc::clone(&observer);
emitter.on_event(move |event| observer_clone.dispatch_envelope(event));
emitter
}
async fn initialized(
registry: HandlerRegistry,
emitter: Arc<EventEmitter>,
@ -42,6 +49,7 @@ async fn initialized(
inner_store,
run_options.run_dir.clone(),
));
let emitter = bound_emitter(run_options.run_id, emitter);
Initialized {
graph: graph.clone(),
source: String::new(),

6
lib/crates/fabro-workflow/tests/it/integration.rs
View file

@ -12448,11 +12448,11 @@ async fn e2e_stall_watchdog_triggers_from_dot_parsed_pipeline() {
"expected error to contain 'stall watchdog', got: {err}"
);
// Verify StallWatchdogTimeout event was emitted
// Verify the canonical watchdog timeout envelope was emitted.
let collected = events.lock().unwrap();
assert!(
collected.iter().any(|e| e.contains("StallWatchdogTimeout")),
"expected StallWatchdogTimeout event in: {collected:?}"
collected.iter().any(|e| e.contains("watchdog.timeout")),
"expected watchdog.timeout event in: {collected:?}"
);
}