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

Expand workflow event payloads and stabilize CLI logs

Browse source 
Add richer run, stage, prompt, command, retro, and agent session event
metadata so progress output and stored workflow events carry the context
needed by the new plan. Normalize event serialization and update CLI log
handling to prefer progress.jsonl with consistent redaction, and fix the
detached wait/log race covered by the updated integration and snapshot
tests.
This commit is contained in:
Bryan Helmkamp 2026-04-01 19:24:49 -04:00
parent 2178cf16dd
commit 81fd7aa8e6
No known key found for this signature in database
25 changed files with 1556 additions and 507 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

29
lib/crates/fabro-agent/src/event.rs
View file

@ -52,10 +52,22 @@ mod tests {
let emitter = EventEmitter::new();
let mut receiver = emitter.subscribe();
emitter.emit("sess-1".into(), AgentEvent::SessionStarted);
emitter.emit(
"sess-1".into(),
AgentEvent::SessionStarted {
provider: Some("anthropic".into()),
model: Some("claude-opus".into()),
},
);
let event = receiver.recv().await.unwrap();
assert!(matches!(event.event, AgentEvent::SessionStarted));
assert!(matches!(
event.event,
AgentEvent::SessionStarted {
provider: Some(_),
model: Some(_)
}
));
assert_eq!(event.session_id, "sess-1");
assert_eq!(event.parent_session_id, None);
}
@ -120,7 +132,10 @@ mod tests {
let mut receiver = emitter.subscribe();
emitter.forward(SessionEvent {
event: AgentEvent::SessionStarted,
event: AgentEvent::SessionStarted {
provider: Some("anthropic".into()),
model: Some("claude-opus".into()),
},
timestamp: SystemTime::now(),
session_id: "child".into(),
parent_session_id: Some("parent".into()),
@ -129,6 +144,12 @@ mod tests {
let event = receiver.recv().await.unwrap();
assert_eq!(event.session_id, "child");
assert_eq!(event.parent_session_id.as_deref(), Some("parent"));
assert!(matches!(event.event, AgentEvent::SessionStarted));
assert!(matches!(
event.event,
AgentEvent::SessionStarted {
provider: Some(_),
model: Some(_)
}
));
}
}

15
lib/crates/fabro-agent/src/session.rs
View file

@ -101,8 +101,13 @@ impl Session {
/// Initialize session by discovering project docs and capturing environment context.
/// Call before `process_input`.
pub async fn initialize(&mut self) {
self.event_emitter
.emit(self.id.clone(), AgentEvent::SessionStarted);
self.event_emitter.emit(
self.id.clone(),
AgentEvent::SessionStarted {
provider: Some(self.provider_profile.provider().to_string()),
model: Some(self.provider_profile.model().to_string()),
},
);
let doc_root = self
.config
@ -1280,7 +1285,7 @@ mod tests {
assert!(
events
.iter()
.any(|e| matches!(e.event, AgentEvent::SessionStarted))
.any(|e| matches!(e.event, AgentEvent::SessionStarted { .. }))
);
assert!(
events
@ -1574,7 +1579,7 @@ mod tests {
assert!(
!events
.iter()
.any(|e| matches!(e.event, AgentEvent::SessionStarted)),
.any(|e| matches!(e.event, AgentEvent::SessionStarted { .. })),
"SessionStarted should not be emitted for a closed session"
);
}
@ -1803,7 +1808,7 @@ mod tests {
let mut session_start_count = 0;
let mut session_end_count = 0;
while let Ok(event) = rx.try_recv() {
if matches!(event.event, AgentEvent::SessionStarted) {
if matches!(event.event, AgentEvent::SessionStarted { .. }) {
session_start_count += 1;
}
if matches!(event.event, AgentEvent::SessionEnded) {

12
lib/crates/fabro-agent/src/subagent.rs
View file

@ -99,7 +99,7 @@ impl SubAgentManager {
| AgentEvent::ReasoningDelta { .. }
| AgentEvent::ToolCallOutputDelta { .. }
| AgentEvent::AssistantTextStart
| AgentEvent::SessionStarted
| AgentEvent::SessionStarted { .. }
| AgentEvent::SessionEnded
| AgentEvent::ProcessingEnd
| AgentEvent::SkillExpanded { .. }
@ -722,13 +722,19 @@ mod tests {
let mut rx = parent.subscribe();
callback(SubAgentCallbackEvent::Forwarded(SessionEvent {
event: AgentEvent::SessionStarted,
event: AgentEvent::SessionStarted {
provider: Some("anthropic".into()),
model: Some("claude-opus".into()),
},
timestamp: std::time::SystemTime::now(),
session_id: "child".into(),
parent_session_id: None,
}));
callback(SubAgentCallbackEvent::Forwarded(SessionEvent {
event: AgentEvent::SessionStarted,
event: AgentEvent::SessionStarted {
provider: Some("anthropic".into()),
model: Some("claude-opus".into()),
},
timestamp: std::time::SystemTime::now(),
session_id: "grandchild".into(),
parent_session_id: Some("child".into()),

47
lib/crates/fabro-agent/src/types.rs
View file

@ -91,7 +91,12 @@ pub enum SessionState {
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum AgentEvent {
SessionStarted,
SessionStarted {
#[serde(default, skip_serializing_if = "Option::is_none")]
provider: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
model: Option<String>,
},
SessionEnded,
ProcessingEnd,
UserInput {
@ -199,8 +204,13 @@ impl AgentEvent {
pub fn trace(&self, session_id: &str) {
use tracing::{debug, error, info, warn};
match self {
Self::SessionStarted => {
info!(session_id, "Agent session started");
Self::SessionStarted { provider, model } => {
info!(
session_id,
provider = provider.as_deref().unwrap_or(""),
model = model.as_deref().unwrap_or(""),
"Agent session started"
);
}
Self::SessionEnded => {
info!(session_id, "Agent session ended");
@ -397,12 +407,21 @@ mod tests {
#[test]
fn session_event_construction() {
let event = SessionEvent {
event: AgentEvent::SessionStarted,
event: AgentEvent::SessionStarted {
provider: Some("anthropic".into()),
model: Some("claude-opus".into()),
},
timestamp: SystemTime::now(),
session_id: "sess_1".into(),
parent_session_id: None,
};
assert!(matches!(event.event, AgentEvent::SessionStarted));
assert!(matches!(
event.event,
AgentEvent::SessionStarted {
provider: Some(_),
model: Some(_)
}
));
assert_eq!(event.session_id, "sess_1");
assert_eq!(event.parent_session_id, None);
}
@ -528,7 +547,10 @@ mod tests {
#[test]
fn session_event_serde_round_trip_without_parent_session_id() {
let event = SessionEvent {
event: AgentEvent::SessionStarted,
event: AgentEvent::SessionStarted {
provider: Some("anthropic".into()),
model: Some("claude-opus".into()),
},
timestamp: SystemTime::now(),
session_id: "sess_42".into(),
parent_session_id: None,
@ -543,13 +565,22 @@ mod tests {
let deserialized: SessionEvent = serde_json::from_str(&json).unwrap();
assert_eq!(deserialized.session_id, "sess_42");
assert_eq!(deserialized.parent_session_id, None);
assert!(matches!(deserialized.event, AgentEvent::SessionStarted));
assert!(matches!(
deserialized.event,
AgentEvent::SessionStarted {
provider: Some(_),
model: Some(_)
}
));
}
#[test]
fn session_event_serde_round_trip_with_parent_session_id() {
let event = SessionEvent {
event: AgentEvent::SessionStarted,
event: AgentEvent::SessionStarted {
provider: Some("openai".into()),
model: Some("gpt-5.4".into()),
},
timestamp: SystemTime::now(),
session_id: "sess_child".into(),
parent_session_id: Some("sess_parent".into()),

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

@ -7,6 +7,7 @@ use anyhow::{Context, Result, bail};
use chrono::{DateTime, Utc};
use fabro_config::FabroSettingsExt;
use fabro_store::RunStore;
use fabro_util::redact::redact_jsonl_line;
use fabro_util::terminal::Styles;
use fabro_workflow::run_lookup::{resolve_run_combined, runs_base};
use futures::StreamExt;
@ -32,7 +33,9 @@ pub(crate) async fn run(args: &LogsArgs, styles: &Styles, globals: &GlobalArgs)
let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await?;
let progress_path = run.path.join("progress.jsonl");
let (all_lines, last_seq, use_store_follow) = if let Some(run_store) = run_store.as_ref() {
let (all_lines, last_seq, use_store_follow) = if progress_path.exists() {
(read_lines(&progress_path)?, 0, false)
} else if let Some(run_store) = run_store.as_ref() {
match run_store.list_events().await {
Ok(events) => {
let last_seq = events.last().map_or(0, |event| event.seq);
@ -43,22 +46,11 @@ pub(crate) async fn run(args: &LogsArgs, styles: &Styles, globals: &GlobalArgs)
(lines, last_seq, true)
}
Err(err) => {
if !progress_path.exists() {
return Err(err).context("Failed to list store-backed run events");
}
warn!(
run_id = %run.run_id,
error = %err,
"Failed to read events from store; falling back to progress.jsonl"
);
(read_lines(&progress_path)?, 0, false)
return Err(err).context("Failed to list store-backed run events");
}
}
} else {
if !progress_path.exists() {
bail!("No progress.jsonl found for run '{}'", run.run_id);
}
(read_lines(&progress_path)?, 0, false)
bail!("No progress.jsonl found for run '{}'", run.run_id);
};
let filtered = apply_filters(&all_lines, since_cutoff.as_ref(), args.tail);
@ -323,7 +315,8 @@ async fn flush_remaining_store_events(
}
fn event_payload_line(event: &fabro_store::EventEnvelope) -> Result<String> {
serde_json::to_string(event.payload.as_value()).map_err(Into::into)
let line = serde_json::to_string(event.payload.as_value())?;
Ok(redact_jsonl_line(&line))
}
fn render_indented_markdown(styles: &Styles, text: &str, indent: &str) -> String {

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

@ -603,6 +603,12 @@ mod tests {
failure: None,
notes: None,
files_touched: Vec::new(),
context_updates: None,
jump_to_node: None,
context_values: None,
node_visits: None,
loop_failure_signatures: None,
restart_failure_signatures: None,
attempt: 1,
max_attempts: 1,
};

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

@ -493,7 +493,7 @@ mod tests {
node_id: node_id.into(),
name: name.into(),
index: 0,
handler_type: None,
handler_type: String::new(),
script: None,
attempt: 1,
max_attempts: 1,
@ -534,6 +534,12 @@ mod tests {
failure: None,
notes: None,
files_touched: Vec::new(),
context_updates: None,
jump_to_node: None,
context_values: None,
node_visits: None,
loop_failure_signatures: None,
restart_failure_signatures: None,
attempt: 1,
max_attempts: 1,
}
@ -578,6 +584,7 @@ mod tests {
index: 0,
duration_ms: 2000,
status: "success".into(),
head_sha: None,
},
);
let stage = &ui.stage.active_stages["fork1"];
@ -664,6 +671,10 @@ mod tests {
stage_started("code", "Code"),
WorkflowRunEvent::SandboxInitialized {
working_directory: "/home/daytona/workspace".into(),
provider: "daytona".into(),
identifier: None,
host_working_directory: None,
container_mount_point: None,
},
agent_event(
"code",
@ -900,6 +911,10 @@ mod tests {
&mut ui,
WorkflowRunEvent::SandboxInitialized {
working_directory: "/home/daytona/workspace".into(),
provider: "daytona".into(),
identifier: None,
host_working_directory: None,
container_mount_point: None,
},
);
emit(
@ -1107,6 +1122,7 @@ mod tests {
index: 0,
duration_ms: 500,
status: "success".into(),
head_sha: None,
},
);

27
lib/crates/fabro-cli/src/commands/run/wait.rs
View file

@ -14,6 +14,11 @@ use crate::shared::format_duration_ms;
use crate::store;
use crate::user_config::load_user_settings_with_globals;
#[cfg(test)]
const WAIT_STARTUP_GRACE: std::time::Duration = std::time::Duration::from_millis(500);
#[cfg(not(test))]
const WAIT_STARTUP_GRACE: std::time::Duration = std::time::Duration::from_secs(3);
pub(crate) async fn run(args: &WaitArgs, styles: &Styles, globals: &GlobalArgs) -> Result<()> {
let cli_settings = load_user_settings_with_globals(globals)?;
let base = runs_base(&cli_settings.storage_dir());
@ -28,22 +33,24 @@ pub(crate) async fn run(args: &WaitArgs, styles: &Styles, globals: &GlobalArgs)
.timeout
.map(|secs| std::time::Instant::now() + std::time::Duration::from_secs(secs));
let interval = std::time::Duration::from_millis(args.interval);
let started_waiting_at = std::time::Instant::now();
let final_status = loop {
let status = match run_store.as_ref() {
Some(run_store) => match run_store.get_status().await {
Ok(Some(record)) => record.status,
Ok(None) => RunStatus::Dead,
Err(_) => match RunStatusRecord::load(&status_path) {
Ok(record) => record.status,
Err(_) => RunStatus::Dead,
},
},
None => match RunStatusRecord::load(&status_path) {
Ok(record) => record.status,
Err(_) => RunStatus::Dead,
Ok(Some(record)) => Some(record.status),
Ok(None) => RunStatusRecord::load(&status_path).ok().map(|record| record.status),
Err(_) => RunStatusRecord::load(&status_path).ok().map(|record| record.status),
},
None => RunStatusRecord::load(&status_path).ok().map(|record| record.status),
};
let status = status.unwrap_or_else(|| {
if started_waiting_at.elapsed() < WAIT_STARTUP_GRACE {
RunStatus::Submitted
} else {
RunStatus::Dead
}
});
if status.is_terminal() {
break status;

266
lib/crates/fabro-cli/tests/it/cmd/attach.rs
View file

@ -242,120 +242,266 @@ fn attach_json_errors_without_prompting_for_human_input() {
fabro_json_snapshot!(context, &progress, @r#"
[
{
"event": "run.created",
"id": "[EVENT_ID]",
"ts": "[TIMESTAMP]",
"properties": {
"graph": {
"attrs": {
"goal": {
"String": "Wait for approval"
}
},
"edges": [
{
"attrs": {},
"from": "start",
"to": "approve"
},
{
"attrs": {
"label": {
"String": "[A] Approve"
}
},
"from": "approve",
"to": "ship"
},
{
"attrs": {
"label": {
"String": "[R] Revise"
}
},
"from": "approve",
"to": "revise"
},
{
"attrs": {},
"from": "ship",
"to": "exit"
},
{
"attrs": {},
"from": "revise",
"to": "exit"
}
],
"name": "HumanGate",
"nodes": {
"approve": {
"attrs": {
"label": {
"String": "Approve?"
},
"shape": {
"String": "hexagon"
}
},
"id": "approve"
},
"exit": {
"attrs": {
"label": {
"String": "Exit"
},
"shape": {
"String": "Msquare"
}
},
"id": "exit"
},
"revise": {
"attrs": {
"script": {
"String": "echo revised"
},
"shape": {
"String": "parallelogram"
}
},
"id": "revise"
},
"ship": {
"attrs": {
"script": {
"String": "echo shipped"
},
"shape": {
"String": "parallelogram"
}
},
"id": "ship"
},
"start": {
"attrs": {
"label": {
"String": "Start"
},
"shape": {
"String": "Mdiamond"
}
},
"id": "start"
}
}
},
"host_repo_path": "[TEMP_DIR]",
"labels": {},
"run_dir": "[STORAGE_DIR]/runs/20260401-[ULID]",
"settings": {
"goal": "Wait for approval",
"llm": {
"fallbacks": null,
"model": "gpt-5.4",
"provider": "openai"
},
"mode": "standalone",
"no_retro": true,
"sandbox": {
"daytona": null,
"devcontainer": null,
"env": null,
"local": null,
"preserve": null,
"provider": "local"
},
"storage_dir": "[STORAGE_DIR]"
},
"workflow_slug": "human-gate",
"workflow_source": "digraph HumanGate {/n graph [goal=\"Wait for approval\"]/n start [shape=Mdiamond, label=\"Start\"]/n exit [shape=Msquare, label=\"Exit\"]/n approve [shape=hexagon, label=\"Approve?\"]/n ship [shape=parallelogram, script=\"echo shipped\"]/n revise [shape=parallelogram, script=\"echo revised\"]/n start -> approve/n approve -> ship [label=\"[A] Approve\"]/n approve -> revise [label=\"[R] Revise\"]/n ship -> exit/n revise -> exit/n}/n",
"working_directory": "[TEMP_DIR]"
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "sandbox.initializing",
"id": "[EVENT_ID]",
"properties": {
"provider": "local"
}
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"id": "[EVENT_ID]",
"ts": "[TIMESTAMP]",
"run_id": "[ULID]",
"event": "sandbox.ready",
"id": "[EVENT_ID]",
"properties": {
"cpu": null,
"duration_ms": "[DURATION_MS]",
"memory": null,
"name": null,
"provider": "local",
"url": null
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "sandbox.initialized",
"id": "[EVENT_ID]",
"properties": {
"provider": "local",
"duration_ms": "[DURATION_MS]",
"name": null,
"cpu": null,
"memory": null,
"url": null
}
},
{
"id": "[EVENT_ID]",
"ts": "[TIMESTAMP]",
"run_id": "[ULID]",
"event": "sandbox.initialized",
"properties": {
"working_directory": "[TEMP_DIR]"
}
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"id": "[EVENT_ID]",
"ts": "[TIMESTAMP]",
"run_id": "[ULID]",
"event": "run.started",
"id": "[EVENT_ID]",
"properties": {
"name": "HumanGate",
"goal": "Wait for approval"
}
"goal": "Wait for approval",
"name": "HumanGate"
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"id": "[EVENT_ID]",
"ts": "[TIMESTAMP]",
"run_id": "[ULID]",
"event": "stage.started",
"id": "[EVENT_ID]",
"node_id": "start",
"node_label": "Start",
"properties": {
"max_attempts": 1,
"attempt": 1,
"handler_type": "start",
"index": 0,
"handler_type": "start"
}
"max_attempts": 1
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"id": "[EVENT_ID]",
"ts": "[TIMESTAMP]",
"run_id": "[ULID]",
"event": "stage.completed",
"id": "[EVENT_ID]",
"node_id": "start",
"node_label": "Start",
"properties": {
"max_attempts": 1,
"attempt": 1,
"index": 0,
"context_values": {
"current.preamble": "Goal: Wait for approval/n",
"current_node": "start",
"graph.goal": "Wait for approval",
"internal.fidelity": "compact",
"internal.node_visit_count": 1,
"internal.run_id": "[ULID]",
"internal.thread_id": null
},
"duration_ms": "[DURATION_MS]",
"status": "success",
"preferred_label": null,
"suggested_next_ids": [],
"usage": null,
"files_touched": [],
"index": 0,
"max_attempts": 1,
"node_visits": {
"start": 1
},
"notes": null,
"files_touched": []
}
"preferred_label": null,
"status": "success",
"suggested_next_ids": [],
"usage": null
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"id": "[EVENT_ID]",
"ts": "[TIMESTAMP]",
"run_id": "[ULID]",
"event": "edge.selected",
"id": "[EVENT_ID]",
"properties": {
"from_node": "start",
"to_node": "approve",
"label": null,
"condition": null,
"from_node": "start",
"is_jump": false,
"label": null,
"reason": "unconditional",
"stage_status": "success",
"is_jump": false
}
"to_node": "approve"
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"id": "[EVENT_ID]",
"ts": "[TIMESTAMP]",
"run_id": "[ULID]",
"event": "checkpoint.completed",
"id": "[EVENT_ID]",
"node_id": "start",
"node_label": "start",
"properties": {
"status": "success"
}
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"id": "[EVENT_ID]",
"ts": "[TIMESTAMP]",
"run_id": "[ULID]",
"event": "stage.started",
"id": "[EVENT_ID]",
"node_id": "approve",
"node_label": "Approve?",
"properties": {
"max_attempts": 1,
"attempt": 1,
"handler_type": "human",
"index": 1,
"handler_type": "human"
}
"max_attempts": 1
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
}
]
"#);

114
lib/crates/fabro-cli/tests/it/cmd/logs.rs
View file

@ -51,36 +51,41 @@ fn logs_completed_run_outputs_raw_ndjson() {
r#""id":"[0-9a-f-]+""#.to_string(),
r#""id":"[EVENT_ID]""#.to_string(),
));
filters.push((
r#""run_dir":"(?:\[DRY_RUN_DIR\]|\[STORAGE_DIR\]/runs/REDACTED)""#.to_string(),
r#""run_dir":"[RUN_DIR]""#.to_string(),
));
let mut cmd = context.command();
cmd.args(["logs", &run.run_id]);
fabro_snapshot!(filters, cmd, @r###"
fabro_snapshot!(filters, cmd, @r#"
success: true
exit_code: 0
----- stdout -----
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"sandbox.initializing","properties":{"provider":"local"}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"sandbox.ready","properties":{"provider":"local","duration_ms": [DURATION_MS],"name":null,"cpu":null,"memory":null,"url":null}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"sandbox.initialized","properties":{"working_directory":"[TEMP_DIR]"}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"run.started","properties":{"name":"Simple","goal":"Run tests and report results"}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"stage.started","node_id":"start","node_label":"Start","properties":{"max_attempts":1,"attempt":1,"index":0,"handler_type":"start"}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"stage.completed","node_id":"start","node_label":"Start","properties":{"max_attempts":1,"attempt":1,"index":0,"duration_ms": [DURATION_MS],"status":"success","preferred_label":null,"suggested_next_ids":[],"usage":null,"notes":"[Simulated] start","files_touched":[]}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"edge.selected","properties":{"from_node":"start","to_node":"run_tests","label":null,"condition":null,"reason":"unconditional","stage_status":"success","is_jump":false}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"checkpoint.completed","node_id":"start","node_label":"start","properties":{"status":"success"}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"stage.started","node_id":"run_tests","node_label":"Run Tests","properties":{"max_attempts":1,"attempt":1,"index":1,"handler_type":"agent"}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"stage.completed","node_id":"run_tests","node_label":"Run Tests","properties":{"max_attempts":1,"attempt":1,"index":1,"duration_ms": [DURATION_MS],"status":"success","preferred_label":null,"suggested_next_ids":[],"usage":null,"notes":"[Simulated] run_tests","files_touched":[]}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"edge.selected","properties":{"from_node":"run_tests","to_node":"report","label":null,"condition":null,"reason":"unconditional","stage_status":"success","is_jump":false}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"checkpoint.completed","node_id":"run_tests","node_label":"run_tests","properties":{"status":"success"}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"stage.started","node_id":"report","node_label":"Report","properties":{"max_attempts":1,"attempt":1,"index":2,"handler_type":"agent"}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"stage.completed","node_id":"report","node_label":"Report","properties":{"max_attempts":1,"attempt":1,"index":2,"duration_ms": [DURATION_MS],"status":"success","preferred_label":null,"suggested_next_ids":[],"usage":null,"notes":"[Simulated] report","files_touched":[]}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"edge.selected","properties":{"from_node":"report","to_node":"exit","label":null,"condition":null,"reason":"unconditional","stage_status":"success","is_jump":false}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"checkpoint.completed","node_id":"report","node_label":"report","properties":{"status":"success"}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"stage.started","node_id":"exit","node_label":"Exit","properties":{"max_attempts":1,"attempt":1,"index":3,"handler_type":"exit"}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"stage.completed","node_id":"exit","node_label":"Exit","properties":{"max_attempts":1,"attempt":1,"index":3,"duration_ms": [DURATION_MS],"status":"success","preferred_label":null,"suggested_next_ids":[],"usage":null,"notes":null,"files_touched":[]}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"run.completed","properties":{"duration_ms": [DURATION_MS],"artifact_count":0,"status":"success"}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"sandbox.cleanup.started","properties":{"provider":"local"}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"sandbox.cleanup.completed","properties":{"provider":"local","duration_ms": [DURATION_MS]}}
{"event":"run.created","id":"[EVENT_ID]","properties":{"graph":{"attrs":{"goal":{"String":"Run tests and report results"},"rankdir":{"String":"LR"}},"edges":[{"attrs":{},"from":"start","to":"run_tests"},{"attrs":{},"from":"run_tests","to":"report"},{"attrs":{},"from":"report","to":"exit"}],"name":"Simple","nodes":{"exit":{"attrs":{"label":{"String":"Exit"},"shape":{"String":"Msquare"}},"id":"exit"},"report":{"attrs":{"label":{"String":"Report"},"prompt":{"String":"Summarize the test results"}},"id":"report"},"run_tests":{"attrs":{"label":{"String":"Run Tests"},"prompt":{"String":"Run the test suite and report results"}},"id":"run_tests"},"start":{"attrs":{"label":{"String":"Start"},"shape":{"String":"Mdiamond"}},"id":"start"}}},"host_repo_path":"[TEMP_DIR]","labels":{},"run_dir":"[RUN_DIR]","settings":{"auto_approve":true,"dry_run":true,"fabro":{"root":"fabro/"},"features":{"retros":false,"session_sandboxes":false},"goal":"Run tests and report results","hooks":[{"blocking":true,"command":"cargo fmt","event":"post_tool_use","matcher":"write_file|edit_file|apply_patch","name":"cargo-fmt","sandbox":null,"timeout_ms":null}],"llm":{"fallbacks":null,"model":"claude-sonnet-4-6","provider":"anthropic"},"mode":"standalone","no_retro":true,"pull_request":{"auto_merge":false,"draft":false,"enabled":true,"merge_strategy":"squash"},"sandbox":{"daytona":{"auto_stop_interval":30,"labels":{"repo":"fabro-sh/fabro"},"network":null,"skip_clone":false,"snapshot":{"cpu":4,"disk":20,"dockerfile":"FROM ubuntu:24.04/n/nRUN apt-get update && apt-get install -y --no-install-recommends curl git ca-certificates build-essential pkg-config libssl-dev unzip python3 && rm -rf /var/lib/apt/lists/*/n/n# GitHub CLI/nRUN curl -fsSL https://cli.github.com/packages/githubcli-archive-keyring.gpg | dd of=/usr/share/keyrings/githubcli-archive-keyring.gpg && echo \"deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/githubcli-archive-keyring.gpg] https://cli.github.com/packages stable main\" | tee /etc/apt/sources.list.d/github-cli.list > /dev/null && apt-get update && apt-get install -y --no-install-recommends gh && rm -rf /var/lib/apt/lists/*/n/n# Rust/nRUN curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y/nENV PATH=\"/root/.cargo/bin:${PATH}\"/nRUN cargo install cargo-nextest --locked/nENV CARGO_INCREMENTAL=0/n/n# Bun/nRUN curl -fsSL https://bun.sh/install | bash/nENV PATH=\"/root/.bun/bin:${PATH}\"/n/nWORKDIR /root/n","memory":8,"name":"fabro-v6"}},"devcontainer":null,"env":null,"local":null,"preserve":null,"provider":"local"},"storage_dir":"[STORAGE_DIR]","version":1},"workflow_slug":"simple","workflow_source":"digraph Simple {/n graph [goal=\"Run tests and report results\"]/n rankdir=LR/n/n start [shape=Mdiamond, label=\"Start\"]/n exit [shape=Msquare, label=\"Exit\"]/n/n run_tests [label=\"Run Tests\", prompt=\"Run the test suite and report results\"]/n report [label=\"Report\", prompt=\"Summarize the test results\"]/n/n start -> run_tests -> report -> exit/n}/n","working_directory":"[TEMP_DIR]"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"sandbox.initializing","id":"[EVENT_ID]","properties":{"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"sandbox.ready","id":"[EVENT_ID]","properties":{"cpu":null,"duration_ms": [DURATION_MS],"memory":null,"name":null,"provider":"local","url":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"sandbox.initialized","id":"[EVENT_ID]","properties":{"provider":"local","working_directory":"[TEMP_DIR]"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"run.started","id":"[EVENT_ID]","properties":{"goal":"Run tests and report results","name":"Simple"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.started","id":"[EVENT_ID]","node_id":"start","node_label":"Start","properties":{"attempt":1,"handler_type":"start","index":0,"max_attempts":1},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.completed","id":"[EVENT_ID]","node_id":"start","node_label":"Start","properties":{"attempt":1,"context_values":{"current.preamble":"Goal: Run tests and report results/n","current_node":"start","graph.goal":"Run tests and report results","graph.rankdir":"LR","internal.fidelity":"compact","internal.node_visit_count":1,"internal.run_id":"[ULID]","internal.thread_id":null},"duration_ms": [DURATION_MS],"files_touched":[],"index":0,"max_attempts":1,"node_visits":{"start":1},"notes":"[Simulated] start","preferred_label":null,"status":"success","suggested_next_ids":[],"usage":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"edge.selected","id":"[EVENT_ID]","properties":{"condition":null,"from_node":"start","is_jump":false,"label":null,"reason":"unconditional","stage_status":"success","to_node":"run_tests"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"checkpoint.completed","id":"[EVENT_ID]","node_id":"start","node_label":"start","properties":{"status":"success"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.started","id":"[EVENT_ID]","node_id":"run_tests","node_label":"Run Tests","properties":{"attempt":1,"handler_type":"agent","index":1,"max_attempts":1},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.completed","id":"[EVENT_ID]","node_id":"run_tests","node_label":"Run Tests","properties":{"attempt":1,"context_updates":{"last_response":"[Simulated] Response for stage: run_tests","last_stage":"run_tests","response.run_tests":"[Simulated] Response for stage: run_tests"},"context_values":{"current.preamble":"Goal: Run tests and report results/n","current_node":"run_tests","failure_class":"","failure_signature":"","graph.goal":"Run tests and report results","graph.rankdir":"LR","internal.fidelity":"compact","internal.node_visit_count":1,"internal.retry_count.start":0,"internal.run_id":"[ULID]","internal.thread_id":"start","outcome":"success","thread.start.current_node":"run_tests"},"duration_ms": [DURATION_MS],"files_touched":[],"index":1,"max_attempts":1,"node_visits":{"run_tests":1,"start":1},"notes":"[Simulated] run_tests","preferred_label":null,"status":"success","suggested_next_ids":[],"usage":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"edge.selected","id":"[EVENT_ID]","properties":{"condition":null,"from_node":"run_tests","is_jump":false,"label":null,"reason":"unconditional","stage_status":"success","to_node":"report"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"checkpoint.completed","id":"[EVENT_ID]","node_id":"run_tests","node_label":"run_tests","properties":{"status":"success"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.started","id":"[EVENT_ID]","node_id":"report","node_label":"Report","properties":{"attempt":1,"handler_type":"agent","index":2,"max_attempts":1},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.completed","id":"[EVENT_ID]","node_id":"report","node_label":"Report","properties":{"attempt":1,"context_updates":{"last_response":"[Simulated] Response for stage: report","last_stage":"report","response.report":"[Simulated] Response for stage: report"},"context_values":{"current.preamble":"Goal: Run tests and report results/n/n## Completed stages/n- **run_tests**: success/n","current_node":"report","failure_class":"","failure_signature":"","graph.goal":"Run tests and report results","graph.rankdir":"LR","internal.fidelity":"compact","internal.node_visit_count":1,"internal.retry_count.run_tests":0,"internal.retry_count.start":0,"internal.run_id":"[ULID]","internal.thread_id":"run_tests","last_response":"[Simulated] Response for stage: run_tests","last_stage":"run_tests","outcome":"success","response.run_tests":"[Simulated] Response for stage: run_tests","thread.run_tests.current_node":"report","thread.start.current_node":"run_tests"},"duration_ms": [DURATION_MS],"files_touched":[],"index":2,"max_attempts":1,"node_visits":{"report":1,"run_tests":1,"start":1},"notes":"[Simulated] report","preferred_label":null,"status":"success","suggested_next_ids":[],"usage":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"edge.selected","id":"[EVENT_ID]","properties":{"condition":null,"from_node":"report","is_jump":false,"label":null,"reason":"unconditional","stage_status":"success","to_node":"exit"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"checkpoint.completed","id":"[EVENT_ID]","node_id":"report","node_label":"report","properties":{"status":"success"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.started","id":"[EVENT_ID]","node_id":"exit","node_label":"Exit","properties":{"attempt":1,"handler_type":"exit","index":3,"max_attempts":1},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.completed","id":"[EVENT_ID]","node_id":"exit","node_label":"Exit","properties":{"attempt":1,"duration_ms": [DURATION_MS],"files_touched":[],"index":3,"max_attempts":1,"notes":null,"preferred_label":null,"status":"success","suggested_next_ids":[],"usage":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"run.completed","id":"[EVENT_ID]","properties":{"artifact_count":0,"duration_ms": [DURATION_MS],"status":"success"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"sandbox.cleanup.started","id":"[EVENT_ID]","properties":{"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"sandbox.cleanup.completed","id":"[EVENT_ID]","properties":{"duration_ms": [DURATION_MS],"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
----- stderr -----
"###);
"#);
}
#[test]
@ -100,17 +105,21 @@ fn logs_tail_limits_output() {
r#""id":"[0-9a-f-]+""#.to_string(),
r#""id":"[EVENT_ID]""#.to_string(),
));
filters.push((
r#""run_dir":"(?:\[DRY_RUN_DIR\]|\[STORAGE_DIR\]/runs/REDACTED)""#.to_string(),
r#""run_dir":"[RUN_DIR]""#.to_string(),
));
let mut cmd = context.command();
cmd.args(["logs", "--tail", "2", &run.run_id]);
fabro_snapshot!(filters, cmd, @r###"
fabro_snapshot!(filters, cmd, @r#"
success: true
exit_code: 0
----- stdout -----
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"sandbox.cleanup.started","properties":{"provider":"local"}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"sandbox.cleanup.completed","properties":{"provider":"local","duration_ms": [DURATION_MS]}}
{"event":"sandbox.cleanup.started","id":"[EVENT_ID]","properties":{"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"sandbox.cleanup.completed","id":"[EVENT_ID]","properties":{"duration_ms": [DURATION_MS],"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
----- stderr -----
"###);
"#);
}
#[test]
@ -167,34 +176,39 @@ fn logs_follow_detached_run_streams_until_completion() {
r#""id":"[0-9a-f-]+""#.to_string(),
r#""id":"[EVENT_ID]""#.to_string(),
));
filters.push((
r#""run_dir":"(?:\[DRY_RUN_DIR\]|\[STORAGE_DIR\]/runs/REDACTED)""#.to_string(),
r#""run_dir":"[RUN_DIR]""#.to_string(),
));
let mut cmd = context.command();
cmd.args(["logs", "--follow", &run.run_id]);
fabro_snapshot!(filters, cmd, @r###"
fabro_snapshot!(filters, cmd, @r#"
success: true
exit_code: 0
----- stdout -----
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"sandbox.initializing","properties":{"provider":"local"}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"sandbox.ready","properties":{"provider":"local","duration_ms": [DURATION_MS],"name":null,"cpu":null,"memory":null,"url":null}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"sandbox.initialized","properties":{"working_directory":"[TEMP_DIR]"}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"run.started","properties":{"name":"Simple","goal":"Run tests and report results"}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"stage.started","node_id":"start","node_label":"Start","properties":{"max_attempts":1,"attempt":1,"index":0,"handler_type":"start"}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"stage.completed","node_id":"start","node_label":"Start","properties":{"max_attempts":1,"attempt":1,"index":0,"duration_ms": [DURATION_MS],"status":"success","preferred_label":null,"suggested_next_ids":[],"usage":null,"notes":"[Simulated] start","files_touched":[]}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"edge.selected","properties":{"from_node":"start","to_node":"run_tests","label":null,"condition":null,"reason":"unconditional","stage_status":"success","is_jump":false}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"checkpoint.completed","node_id":"start","node_label":"start","properties":{"status":"success"}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"stage.started","node_id":"run_tests","node_label":"Run Tests","properties":{"max_attempts":1,"attempt":1,"index":1,"handler_type":"agent"}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"stage.completed","node_id":"run_tests","node_label":"Run Tests","properties":{"max_attempts":1,"attempt":1,"index":1,"duration_ms": [DURATION_MS],"status":"success","preferred_label":null,"suggested_next_ids":[],"usage":null,"notes":"[Simulated] run_tests","files_touched":[]}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"edge.selected","properties":{"from_node":"run_tests","to_node":"report","label":null,"condition":null,"reason":"unconditional","stage_status":"success","is_jump":false}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"checkpoint.completed","node_id":"run_tests","node_label":"run_tests","properties":{"status":"success"}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"stage.started","node_id":"report","node_label":"Report","properties":{"max_attempts":1,"attempt":1,"index":2,"handler_type":"agent"}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"stage.completed","node_id":"report","node_label":"Report","properties":{"max_attempts":1,"attempt":1,"index":2,"duration_ms": [DURATION_MS],"status":"success","preferred_label":null,"suggested_next_ids":[],"usage":null,"notes":"[Simulated] report","files_touched":[]}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"edge.selected","properties":{"from_node":"report","to_node":"exit","label":null,"condition":null,"reason":"unconditional","stage_status":"success","is_jump":false}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"checkpoint.completed","node_id":"report","node_label":"report","properties":{"status":"success"}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"stage.started","node_id":"exit","node_label":"Exit","properties":{"max_attempts":1,"attempt":1,"index":3,"handler_type":"exit"}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"stage.completed","node_id":"exit","node_label":"Exit","properties":{"max_attempts":1,"attempt":1,"index":3,"duration_ms": [DURATION_MS],"status":"success","preferred_label":null,"suggested_next_ids":[],"usage":null,"notes":null,"files_touched":[]}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"run.completed","properties":{"duration_ms": [DURATION_MS],"artifact_count":0,"status":"success"}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"sandbox.cleanup.started","properties":{"provider":"local"}}
{"id":"[EVENT_ID]","ts":"[TIMESTAMP]","run_id":"[ULID]","event":"sandbox.cleanup.completed","properties":{"provider":"local","duration_ms": [DURATION_MS]}}
{"event":"run.created","id":"[EVENT_ID]","properties":{"graph":{"attrs":{"goal":{"String":"Run tests and report results"},"rankdir":{"String":"LR"}},"edges":[{"attrs":{},"from":"start","to":"run_tests"},{"attrs":{},"from":"run_tests","to":"report"},{"attrs":{},"from":"report","to":"exit"}],"name":"Simple","nodes":{"exit":{"attrs":{"label":{"String":"Exit"},"shape":{"String":"Msquare"}},"id":"exit"},"report":{"attrs":{"label":{"String":"Report"},"prompt":{"String":"Summarize the test results"}},"id":"report"},"run_tests":{"attrs":{"label":{"String":"Run Tests"},"prompt":{"String":"Run the test suite and report results"}},"id":"run_tests"},"start":{"attrs":{"label":{"String":"Start"},"shape":{"String":"Mdiamond"}},"id":"start"}}},"host_repo_path":"[TEMP_DIR]","labels":{},"run_dir":"[RUN_DIR]","settings":{"auto_approve":true,"dry_run":true,"fabro":{"root":"fabro/"},"features":{"retros":false,"session_sandboxes":false},"goal":"Run tests and report results","hooks":[{"blocking":true,"command":"cargo fmt","event":"post_tool_use","matcher":"write_file|edit_file|apply_patch","name":"cargo-fmt","sandbox":null,"timeout_ms":null}],"llm":{"fallbacks":null,"model":"claude-sonnet-4-6","provider":"anthropic"},"mode":"standalone","no_retro":true,"pull_request":{"auto_merge":false,"draft":false,"enabled":true,"merge_strategy":"squash"},"sandbox":{"daytona":{"auto_stop_interval":30,"labels":{"repo":"fabro-sh/fabro"},"network":null,"skip_clone":false,"snapshot":{"cpu":4,"disk":20,"dockerfile":"FROM ubuntu:24.04/n/nRUN apt-get update && apt-get install -y --no-install-recommends curl git ca-certificates build-essential pkg-config libssl-dev unzip python3 && rm -rf /var/lib/apt/lists/*/n/n# GitHub CLI/nRUN curl -fsSL https://cli.github.com/packages/githubcli-archive-keyring.gpg | dd of=/usr/share/keyrings/githubcli-archive-keyring.gpg && echo \"deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/githubcli-archive-keyring.gpg] https://cli.github.com/packages stable main\" | tee /etc/apt/sources.list.d/github-cli.list > /dev/null && apt-get update && apt-get install -y --no-install-recommends gh && rm -rf /var/lib/apt/lists/*/n/n# Rust/nRUN curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y/nENV PATH=\"/root/.cargo/bin:${PATH}\"/nRUN cargo install cargo-nextest --locked/nENV CARGO_INCREMENTAL=0/n/n# Bun/nRUN curl -fsSL https://bun.sh/install | bash/nENV PATH=\"/root/.bun/bin:${PATH}\"/n/nWORKDIR /root/n","memory":8,"name":"fabro-v6"}},"devcontainer":null,"env":null,"local":null,"preserve":null,"provider":"local"},"storage_dir":"[STORAGE_DIR]","version":1},"workflow_slug":"simple","workflow_source":"digraph Simple {/n graph [goal=\"Run tests and report results\"]/n rankdir=LR/n/n start [shape=Mdiamond, label=\"Start\"]/n exit [shape=Msquare, label=\"Exit\"]/n/n run_tests [label=\"Run Tests\", prompt=\"Run the test suite and report results\"]/n report [label=\"Report\", prompt=\"Summarize the test results\"]/n/n start -> run_tests -> report -> exit/n}/n","working_directory":"[TEMP_DIR]"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"sandbox.initializing","id":"[EVENT_ID]","properties":{"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"sandbox.ready","id":"[EVENT_ID]","properties":{"cpu":null,"duration_ms": [DURATION_MS],"memory":null,"name":null,"provider":"local","url":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"sandbox.initialized","id":"[EVENT_ID]","properties":{"provider":"local","working_directory":"[TEMP_DIR]"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"run.started","id":"[EVENT_ID]","properties":{"goal":"Run tests and report results","name":"Simple"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.started","id":"[EVENT_ID]","node_id":"start","node_label":"Start","properties":{"attempt":1,"handler_type":"start","index":0,"max_attempts":1},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.completed","id":"[EVENT_ID]","node_id":"start","node_label":"Start","properties":{"attempt":1,"context_values":{"current.preamble":"Goal: Run tests and report results/n","current_node":"start","graph.goal":"Run tests and report results","graph.rankdir":"LR","internal.fidelity":"compact","internal.node_visit_count":1,"internal.run_id":"[ULID]","internal.thread_id":null},"duration_ms": [DURATION_MS],"files_touched":[],"index":0,"max_attempts":1,"node_visits":{"start":1},"notes":"[Simulated] start","preferred_label":null,"status":"success","suggested_next_ids":[],"usage":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"edge.selected","id":"[EVENT_ID]","properties":{"condition":null,"from_node":"start","is_jump":false,"label":null,"reason":"unconditional","stage_status":"success","to_node":"run_tests"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"checkpoint.completed","id":"[EVENT_ID]","node_id":"start","node_label":"start","properties":{"status":"success"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.started","id":"[EVENT_ID]","node_id":"run_tests","node_label":"Run Tests","properties":{"attempt":1,"handler_type":"agent","index":1,"max_attempts":1},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.completed","id":"[EVENT_ID]","node_id":"run_tests","node_label":"Run Tests","properties":{"attempt":1,"context_updates":{"last_response":"[Simulated] Response for stage: run_tests","last_stage":"run_tests","response.run_tests":"[Simulated] Response for stage: run_tests"},"context_values":{"current.preamble":"Goal: Run tests and report results/n","current_node":"run_tests","failure_class":"","failure_signature":"","graph.goal":"Run tests and report results","graph.rankdir":"LR","internal.fidelity":"compact","internal.node_visit_count":1,"internal.retry_count.start":0,"internal.run_id":"[ULID]","internal.thread_id":"start","outcome":"success","thread.start.current_node":"run_tests"},"duration_ms": [DURATION_MS],"files_touched":[],"index":1,"max_attempts":1,"node_visits":{"run_tests":1,"start":1},"notes":"[Simulated] run_tests","preferred_label":null,"status":"success","suggested_next_ids":[],"usage":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"edge.selected","id":"[EVENT_ID]","properties":{"condition":null,"from_node":"run_tests","is_jump":false,"label":null,"reason":"unconditional","stage_status":"success","to_node":"report"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"checkpoint.completed","id":"[EVENT_ID]","node_id":"run_tests","node_label":"run_tests","properties":{"status":"success"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.started","id":"[EVENT_ID]","node_id":"report","node_label":"Report","properties":{"attempt":1,"handler_type":"agent","index":2,"max_attempts":1},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.completed","id":"[EVENT_ID]","node_id":"report","node_label":"Report","properties":{"attempt":1,"context_updates":{"last_response":"[Simulated] Response for stage: report","last_stage":"report","response.report":"[Simulated] Response for stage: report"},"context_values":{"current.preamble":"Goal: Run tests and report results/n/n## Completed stages/n- **run_tests**: success/n","current_node":"report","failure_class":"","failure_signature":"","graph.goal":"Run tests and report results","graph.rankdir":"LR","internal.fidelity":"compact","internal.node_visit_count":1,"internal.retry_count.run_tests":0,"internal.retry_count.start":0,"internal.run_id":"[ULID]","internal.thread_id":"run_tests","last_response":"[Simulated] Response for stage: run_tests","last_stage":"run_tests","outcome":"success","response.run_tests":"[Simulated] Response for stage: run_tests","thread.run_tests.current_node":"report","thread.start.current_node":"run_tests"},"duration_ms": [DURATION_MS],"files_touched":[],"index":2,"max_attempts":1,"node_visits":{"report":1,"run_tests":1,"start":1},"notes":"[Simulated] report","preferred_label":null,"status":"success","suggested_next_ids":[],"usage":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"edge.selected","id":"[EVENT_ID]","properties":{"condition":null,"from_node":"report","is_jump":false,"label":null,"reason":"unconditional","stage_status":"success","to_node":"exit"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"checkpoint.completed","id":"[EVENT_ID]","node_id":"report","node_label":"report","properties":{"status":"success"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.started","id":"[EVENT_ID]","node_id":"exit","node_label":"Exit","properties":{"attempt":1,"handler_type":"exit","index":3,"max_attempts":1},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"stage.completed","id":"[EVENT_ID]","node_id":"exit","node_label":"Exit","properties":{"attempt":1,"duration_ms": [DURATION_MS],"files_touched":[],"index":3,"max_attempts":1,"notes":null,"preferred_label":null,"status":"success","suggested_next_ids":[],"usage":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"run.completed","id":"[EVENT_ID]","properties":{"artifact_count":0,"duration_ms": [DURATION_MS],"status":"success"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"sandbox.cleanup.started","id":"[EVENT_ID]","properties":{"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"sandbox.cleanup.completed","id":"[EVENT_ID]","properties":{"duration_ms": [DURATION_MS],"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
----- stderr -----
"###);
"#);
}

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

File diff suppressed because it is too large Load diff

5
lib/crates/fabro-retro/src/retro_agent.rs
View file

@ -646,7 +646,10 @@ mod tests {
let handle = spawn_retro_event_writer(rx, jsonl_path.clone());
tx.send(SessionEvent {
event: AgentEvent::SessionStarted,
event: AgentEvent::SessionStarted {
provider: Some("anthropic".into()),
model: Some("claude-opus".into()),
},
timestamp: SystemTime::now(),
session_id: "retro-test".into(),
parent_session_id: None,

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

@ -9,6 +9,7 @@ use fabro_store::{EventPayload, RunStore};
use fabro_types::RunId;
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value};
use std::collections::BTreeMap;
use tokio::sync::{mpsc, oneshot};
use uuid::Uuid;
@ -46,6 +47,26 @@ pub struct RunEventEnvelope {
/// Events emitted during workflow run execution for observability.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum WorkflowRunEvent {
RunCreated {
run_id: RunId,
settings: serde_json::Value,
graph: serde_json::Value,
#[serde(default, skip_serializing_if = "Option::is_none")]
workflow_source: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
workflow_config: Option<String>,
labels: BTreeMap<String, String>,
run_dir: String,
working_directory: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
host_repo_path: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
base_branch: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
workflow_slug: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
db_prefix: Option<String>,
},
WorkflowRunStarted {
name: String,
run_id: RunId,
@ -87,7 +108,7 @@ pub enum WorkflowRunEvent {
node_id: String,
name: String,
index: usize,
handler_type: Option<String>,
handler_type: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
script: Option<String>,
attempt: usize,
@ -106,6 +127,18 @@ pub enum WorkflowRunEvent {
failure: Option<FailureDetail>,
notes: Option<String>,
files_touched: Vec<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
context_updates: Option<BTreeMap<String, serde_json::Value>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
jump_to_node: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
context_values: Option<BTreeMap<String, serde_json::Value>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
node_visits: Option<BTreeMap<String, usize>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
loop_failure_signatures: Option<BTreeMap<String, usize>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
restart_failure_signatures: Option<BTreeMap<String, usize>>,
attempt: usize,
max_attempts: usize,
},
@ -137,6 +170,8 @@ pub enum WorkflowRunEvent {
index: usize,
duration_ms: u64,
status: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
head_sha: Option<String>,
},
ParallelCompleted {
duration_ms: u64,
@ -163,6 +198,8 @@ pub enum WorkflowRunEvent {
status: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
git_commit_sha: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
diff: Option<String>,
},
CheckpointFailed {
node_id: String,
@ -220,6 +257,20 @@ pub enum WorkflowRunEvent {
Prompt {
stage: String,
text: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
mode: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
provider: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
model: Option<String>,
},
PromptCompleted {
node_id: String,
response: String,
model: String,
provider: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
usage: Option<StageUsage>,
},
/// Forwarded from an agent session, tagged with the workflow stage.
Agent {
@ -247,6 +298,13 @@ pub enum WorkflowRunEvent {
/// Emitted after the sandbox has been initialized (by engine lifecycle).
SandboxInitialized {
working_directory: String,
provider: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
identifier: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
host_working_directory: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
container_mount_point: Option<String>,
},
SetupStarted {
command_count: usize,
@ -312,6 +370,36 @@ pub enum WorkflowRunEvent {
error: String,
duration_ms: u64,
},
CommandStarted {
node_id: String,
script: String,
language: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
timeout_ms: Option<u64>,
},
CommandCompleted {
node_id: String,
stdout: String,
stderr: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
exit_code: Option<i32>,
duration_ms: u64,
timed_out: bool,
},
AgentCliStarted {
node_id: String,
mode: String,
provider: String,
model: String,
command: String,
},
AgentCliCompleted {
node_id: String,
stdout: String,
stderr: String,
exit_code: i32,
duration_ms: u64,
},
PullRequestCreated {
pr_url: String,
pr_number: u64,
@ -353,9 +441,16 @@ pub enum WorkflowRunEvent {
exit_code: i32,
stderr: String,
},
RetroStarted,
RetroStarted {
#[serde(default, skip_serializing_if = "Option::is_none")]
provider: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
model: Option<String>,
},
RetroCompleted {
duration_ms: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
retro: Option<serde_json::Value>,
},
RetroFailed {
error: String,
@ -367,6 +462,9 @@ impl WorkflowRunEvent {
pub fn trace(&self) {
use tracing::{debug, error, info, warn};
match self {
Self::RunCreated { run_id, run_dir, .. } => {
info!(run_id = %run_id, run_dir, "Run created");
}
Self::WorkflowRunStarted { name, run_id, .. } => {
info!(workflow = name.as_str(), run_id = %run_id, "Workflow run started");
}
@ -414,7 +512,7 @@ impl WorkflowRunEvent {
node_id,
stage = name.as_str(),
index,
handler_type = handler_type.as_deref().unwrap_or(""),
handler_type,
attempt,
max_attempts,
"Stage started"
@ -501,6 +599,7 @@ impl WorkflowRunEvent {
index,
duration_ms,
status,
..
} => {
debug!(
branch,
@ -566,7 +665,7 @@ impl WorkflowRunEvent {
if *success {
debug!(branch, "Git fetch succeeded");
} else {
warn!(branch, "Git fetch failed");
warn!(branch, "Git fetch failed");
}
}
Self::GitReset { sha } => {
@ -590,14 +689,43 @@ impl WorkflowRunEvent {
Self::LoopRestart { from_node, to_node } => {
debug!(from_node, to_node, "Loop restart");
}
Self::Prompt { stage, text } => {
debug!(stage, text_len = text.len(), "Prompt sent");
Self::Prompt {
stage,
text,
mode,
provider,
model,
} => {
debug!(
stage,
text_len = text.len(),
mode = mode.as_deref().unwrap_or(""),
provider = provider.as_deref().unwrap_or(""),
model = model.as_deref().unwrap_or(""),
"Prompt sent"
);
}
Self::PromptCompleted {
node_id,
model,
provider,
..
} => {
debug!(node_id, model, provider, "Prompt completed");
}
Self::Agent { .. } | Self::Sandbox { .. } => {}
Self::SandboxInitialized {
working_directory, ..
working_directory,
provider,
identifier,
..
} => {
info!(working_directory, "Sandbox initialized");
info!(
working_directory,
provider,
identifier = identifier.as_deref().unwrap_or(""),
"Sandbox initialized"
);
}
Self::SubgraphStarted {
node_id,
@ -707,6 +835,45 @@ impl WorkflowRunEvent {
} => {
error!(cli_name, provider, error, duration_ms, "CLI ensure failed");
}
Self::CommandStarted {
node_id,
language,
timeout_ms,
..
} => {
debug!(node_id, language, timeout_ms, "Command started");
}
Self::CommandCompleted {
node_id,
exit_code,
duration_ms,
timed_out,
..
} => {
debug!(
node_id,
exit_code,
duration_ms,
timed_out,
"Command completed"
);
}
Self::AgentCliStarted {
node_id,
provider,
model,
..
} => {
debug!(node_id, provider, model, "Agent CLI started");
}
Self::AgentCliCompleted {
node_id,
exit_code,
duration_ms,
..
} => {
debug!(node_id, exit_code, duration_ms, "Agent CLI completed");
}
Self::PullRequestCreated {
pr_url,
pr_number,
@ -779,10 +946,14 @@ impl WorkflowRunEvent {
command, index, exit_code, "Devcontainer lifecycle command failed"
);
}
Self::RetroStarted => {
info!("Retro started");
Self::RetroStarted { provider, model } => {
info!(
provider = provider.as_deref().unwrap_or(""),
model = model.as_deref().unwrap_or(""),
"Retro started"
);
}
Self::RetroCompleted { duration_ms } => {
Self::RetroCompleted { duration_ms, .. } => {
info!(duration_ms, "Retro completed");
}
Self::RetroFailed { error, duration_ms } => {
@ -794,6 +965,7 @@ impl WorkflowRunEvent {
pub fn event_name(event: &WorkflowRunEvent) -> &'static str {
match event {
WorkflowRunEvent::RunCreated { .. } => "run.created",
WorkflowRunEvent::WorkflowRunStarted { .. } => "run.started",
WorkflowRunEvent::WorkflowRunCompleted { .. } => "run.completed",
WorkflowRunEvent::WorkflowRunFailed { .. } => "run.failed",
@ -821,8 +993,9 @@ pub fn event_name(event: &WorkflowRunEvent) -> &'static str {
WorkflowRunEvent::EdgeSelected { .. } => "edge.selected",
WorkflowRunEvent::LoopRestart { .. } => "loop.restart",
WorkflowRunEvent::Prompt { .. } => "stage.prompt",
WorkflowRunEvent::PromptCompleted { .. } => "prompt.completed",
WorkflowRunEvent::Agent { event, .. } => match event {
AgentEvent::SessionStarted => "agent.session.started",
AgentEvent::SessionStarted { .. } => "agent.session.started",
AgentEvent::SessionEnded => "agent.session.ended",
AgentEvent::ProcessingEnd => "agent.processing.end",
AgentEvent::UserInput { .. } => "agent.input",
@ -882,6 +1055,10 @@ pub fn event_name(event: &WorkflowRunEvent) -> &'static str {
WorkflowRunEvent::CliEnsureStarted { .. } => "cli.ensure.started",
WorkflowRunEvent::CliEnsureCompleted { .. } => "cli.ensure.completed",
WorkflowRunEvent::CliEnsureFailed { .. } => "cli.ensure.failed",
WorkflowRunEvent::CommandStarted { .. } => "command.started",
WorkflowRunEvent::CommandCompleted { .. } => "command.completed",
WorkflowRunEvent::AgentCliStarted { .. } => "agent.cli.started",
WorkflowRunEvent::AgentCliCompleted { .. } => "agent.cli.completed",
WorkflowRunEvent::PullRequestCreated { .. } => "pull_request.created",
WorkflowRunEvent::PullRequestFailed { .. } => "pull_request.failed",
WorkflowRunEvent::DevcontainerResolved { .. } => "devcontainer.resolved",
@ -896,7 +1073,7 @@ pub fn event_name(event: &WorkflowRunEvent) -> &'static str {
"devcontainer.lifecycle.completed"
}
WorkflowRunEvent::DevcontainerLifecycleFailed { .. } => "devcontainer.lifecycle.failed",
WorkflowRunEvent::RetroStarted => "retro.started",
WorkflowRunEvent::RetroStarted { .. } => "retro.started",
WorkflowRunEvent::RetroCompleted { .. } => "retro.completed",
WorkflowRunEvent::RetroFailed { .. } => "retro.failed",
}
@ -968,7 +1145,7 @@ fn default_node_label(node_id: Option<&String>, node_label: Option<String>) -> O
fn extract_envelope_fields(event: &WorkflowRunEvent) -> EnvelopeFields {
match event {
WorkflowRunEvent::WorkflowRunStarted { .. } => {
WorkflowRunEvent::RunCreated { .. } | WorkflowRunEvent::WorkflowRunStarted { .. } => {
let mut fields = tagged_variant_fields(event);
fields.remove("run_id");
EnvelopeFields {
@ -1010,7 +1187,12 @@ fn extract_envelope_fields(event: &WorkflowRunEvent) -> EnvelopeFields {
| WorkflowRunEvent::CheckpointFailed { .. }
| WorkflowRunEvent::SubgraphStarted { .. }
| WorkflowRunEvent::SubgraphCompleted { .. }
| WorkflowRunEvent::AssetCaptured { .. } => {
| WorkflowRunEvent::AssetCaptured { .. }
| WorkflowRunEvent::PromptCompleted { .. }
| WorkflowRunEvent::CommandStarted { .. }
| WorkflowRunEvent::CommandCompleted { .. }
| WorkflowRunEvent::AgentCliStarted { .. }
| WorkflowRunEvent::AgentCliCompleted { .. } => {
let mut fields = tagged_variant_fields(event);
let node_id = remove_string(&mut fields, "node_id");
let node_label =
@ -1122,10 +1304,18 @@ fn extract_envelope_fields(event: &WorkflowRunEvent) -> EnvelopeFields {
}
pub fn canonicalize_event(run_id: &RunId, event: &WorkflowRunEvent) -> RunEventEnvelope {
canonicalize_event_at(run_id, event, Utc::now())
}
pub fn canonicalize_event_at(
run_id: &RunId,
event: &WorkflowRunEvent,
ts: chrono::DateTime<Utc>,
) -> RunEventEnvelope {
let fields = extract_envelope_fields(event);
RunEventEnvelope {
id: Uuid::now_v7().to_string(),
ts: Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true),
ts: ts.to_rfc3339_opts(SecondsFormat::Millis, true),
run_id: run_id.to_string(),
event: event_name(event).to_string(),
session_id: fields.session_id,
@ -1166,7 +1356,7 @@ pub fn append_progress_event_with_line(
})?;
writeln!(file, "{line}")?;
let pretty = serde_json::to_string_pretty(envelope)?;
let pretty = serde_json::to_string_pretty(&normalized_envelope_value(envelope)?)?;
let pretty = redact_jsonl_line(&pretty);
std::fs::write(run_dir.join("live.json"), pretty)
.with_context(|| format!("Failed to write {}", run_dir.join("live.json").display()))?;
@ -1175,10 +1365,31 @@ pub fn append_progress_event_with_line(
}
pub fn redacted_event_json(envelope: &RunEventEnvelope) -> Result<String> {
let line = serde_json::to_string(envelope)?;
let line = serde_json::to_string(&normalized_envelope_value(envelope)?)?;
Ok(redact_jsonl_line(&line))
}
fn normalized_envelope_value(envelope: &RunEventEnvelope) -> Result<Value> {
let value = serde_json::to_value(envelope)?;
Ok(normalize_json_value(value))
}
fn normalize_json_value(value: Value) -> Value {
match value {
Value::Object(map) => Value::Object(
map.into_iter()
.map(|(key, value)| (key, normalize_json_value(value)))
.collect::<BTreeMap<_, _>>()
.into_iter()
.collect::<Map<_, _>>(),
),
Value::Array(values) => {
Value::Array(values.into_iter().map(normalize_json_value).collect())
}
other => other,
}
}
pub fn event_payload_from_redacted_json(line: &str, run_id: &RunId) -> Result<EventPayload> {
let value = serde_json::from_str(line).context("Failed to parse redacted event payload")?;
EventPayload::new(value, run_id).map_err(anyhow::Error::from)
@ -1444,6 +1655,12 @@ mod tests {
failure: None,
notes: None,
files_touched: Vec::new(),
context_updates: None,
jump_to_node: None,
context_values: None,
node_visits: None,
loop_failure_signatures: None,
restart_failure_signatures: None,
attempt: 1,
max_attempts: 1,
},
@ -1567,7 +1784,13 @@ mod tests {
#[test]
fn build_redacted_event_payload_requires_id() {
let envelope = canonicalize_event(&fixtures::RUN_8, &WorkflowRunEvent::RetroStarted);
let envelope = canonicalize_event(
&fixtures::RUN_8,
&WorkflowRunEvent::RetroStarted {
provider: None,
model: None,
},
);
let payload = build_redacted_event_payload(&envelope, &fixtures::RUN_8).unwrap();
assert_eq!(payload.as_value()["id"], envelope.id);
@ -1576,7 +1799,13 @@ mod tests {
#[test]
fn event_name_matches_new_dot_notation() {
assert_eq!(event_name(&WorkflowRunEvent::RetroStarted), "retro.started");
assert_eq!(
event_name(&WorkflowRunEvent::RetroStarted {
provider: None,
model: None,
}),
"retro.started"
);
assert_eq!(
event_name(&WorkflowRunEvent::ParallelBranchStarted {
branch: "fork".to_string(),

17
lib/crates/fabro-workflow/src/handler/command.rs
View file

@ -6,6 +6,7 @@ use fabro_store::NodeVisitRef;
use crate::context::Context;
use crate::context::keys;
use crate::error::FabroError;
use crate::event::WorkflowRunEvent;
use crate::outcome::{Outcome, OutcomeExt};
use crate::run_dir::{node_dir, visit_from_context};
use fabro_graphviz::graph::{Graph, Node};
@ -107,6 +108,13 @@ impl Handler for CommandHandler {
)
.await?;
services.emitter.emit(&WorkflowRunEvent::CommandStarted {
node_id: node.id.clone(),
script: script.to_string(),
language: language.to_string(),
timeout_ms: timeout_ms(node),
});
let command = if language == "python" {
format!("python3 -c {}", shell_quote(script))
} else {
@ -153,6 +161,15 @@ impl Handler for CommandHandler {
)
.await?;
services.emitter.emit(&WorkflowRunEvent::CommandCompleted {
node_id: node.id.clone(),
stdout: result.stdout.clone(),
stderr: result.stderr.clone(),
exit_code: (!result.timed_out).then_some(result.exit_code),
duration_ms: result.duration_ms,
timed_out: result.timed_out,
});
if result.timed_out {
return Err(FabroError::handler(format!(
"Script timed out after {timeout_ms}ms: {script}",

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

@ -98,9 +98,7 @@ fn spawn_event_forwarder(
// Forward non-streaming agent events to pipeline
if !matches!(
&event.event,
AgentEvent::SessionStarted
| AgentEvent::SessionEnded
| AgentEvent::ProcessingEnd
AgentEvent::ProcessingEnd
| AgentEvent::AssistantTextStart
| AgentEvent::AssistantOutputReplace { .. }
| AgentEvent::TextDelta { .. }
@ -496,6 +494,9 @@ impl CodergenBackend for AgentApiBackend {
emitter.emit(&WorkflowRunEvent::Prompt {
stage: node.id.clone(),
text: prompt.to_string(),
mode: Some("agent".to_string()),
provider: Some(actual_provider.as_str().to_string()),
model: Some(actual_model.clone()),
});
// Record turn count before processing so we only aggregate new usage.

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

@ -494,6 +494,13 @@ impl CodergenBackend for AgentCliBackend {
ensure_cli(cli, provider, sandbox, emitter).await?;
let command = cli_command_for_provider(provider, model, &prompt_path);
emitter.emit(&WorkflowRunEvent::AgentCliStarted {
node_id: node.id.clone(),
mode: "cli".to_string(),
provider: provider.as_str().to_string(),
model: model.to_string(),
command: command.clone(),
});
let _ = fs::create_dir_all(stage_dir).await;
let provider_used = serde_json::json!({
@ -636,6 +643,13 @@ impl CodergenBackend for AgentCliBackend {
timed_out: false,
duration_ms,
};
emitter.emit(&WorkflowRunEvent::AgentCliCompleted {
node_id: node.id.clone(),
stdout: result.stdout.clone(),
stderr: result.stderr.clone(),
exit_code: result.exit_code,
duration_ms: result.duration_ms,
});
// 3e. Cleanup temp files
let _ = sandbox

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

@ -302,6 +302,7 @@ impl Handler for ParallelHandler {
index: setup.branch_index,
duration_ms: millis_u64(branch_start.elapsed()),
status: "fail".to_string(),
head_sha: None,
});
return Ok(BranchResult {
id: setup.target_id.clone(),
@ -382,6 +383,7 @@ impl Handler for ParallelHandler {
index: setup.branch_index,
duration_ms: millis_u64(branch_start.elapsed()),
status: outcome.status.to_string(),
head_sha: head_sha.clone(),
});
Ok::<BranchResult, FabroError>(BranchResult {

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

@ -8,6 +8,7 @@ use fabro_store::NodeVisitRef;
use crate::context::keys;
use crate::context::{Context, WorkflowContext};
use crate::error::FabroError;
use crate::event::WorkflowRunEvent;
use crate::outcome::Outcome;
use crate::run_dir::{node_dir, visit_from_context};
use fabro_graphviz::graph::{Graph, Node};
@ -139,6 +140,25 @@ impl Handler for PromptHandler {
)
};
let response_model = stage_usage
.as_ref()
.map(|usage| usage.model.clone())
.or_else(|| node.model().map(String::from))
.unwrap_or_default();
let response_provider = node
.provider()
.map(String::from)
.or_else(|| Some(Provider::default_from_env().as_str().to_string()))
.unwrap_or_default();
services.emitter.emit(&WorkflowRunEvent::PromptCompleted {
node_id: node.id.clone(),
response: response_text.clone(),
model: response_model,
provider: response_provider,
usage: stage_usage.clone(),
});
// 4. Write response to logs
if let Some(ref store) = services.run_store {
store

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

@ -1,5 +1,6 @@
use std::sync::{Arc, Mutex};
use std::time::Instant;
use std::collections::BTreeMap;
use async_trait::async_trait;
@ -20,20 +21,11 @@ use crate::graph::WorkflowNode;
use crate::outcome::{
FailureCategory, FailureDetail, Outcome, StageStatus, StageUsage, stage_usage_to_llm,
};
use fabro_graphviz::graph::types::Node as GvNode;
use fabro_types::RunId;
type WfRunState = RunState<Option<StageUsage>>;
type WfNodeResult = NodeResult<Option<StageUsage>>;
fn node_script(node: &GvNode) -> Option<String> {
node.attrs
.get("script")
.or_else(|| node.attrs.get("tool_command"))
.and_then(|v| v.as_str())
.map(String::from)
}
/// Sub-lifecycle responsible for emitting workflow run events.
pub(crate) struct EventLifecycle {
pub emitter: Arc<EventEmitter>,
@ -99,8 +91,8 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
node_id: gv.id.clone(),
name: gv.label().to_string(),
index: stage_index,
handler_type: gv.handler_type().map(String::from),
script: node_script(gv),
handler_type: gv.handler_type().unwrap_or_default().to_string(),
script: None,
attempt: 1,
max_attempts: 1,
});
@ -116,6 +108,12 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
failure: None,
notes: None,
files_touched: Vec::new(),
context_updates: None,
jump_to_node: None,
context_values: None,
node_visits: None,
loop_failure_signatures: None,
restart_failure_signatures: None,
attempt: 1,
max_attempts: 1,
});
@ -131,8 +129,8 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
node_id: gv.id.clone(),
name: gv.label().to_string(),
index: state.stage_index,
handler_type: gv.handler_type().map(String::from),
script: node_script(gv),
handler_type: gv.handler_type().unwrap_or_default().to_string(),
script: None,
attempt: ctx.attempt as usize,
max_attempts: ctx.max_attempts as usize,
});
@ -211,6 +209,18 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
failure: outcome.failure.clone(),
notes: outcome.notes.clone(),
files_touched: outcome.files_touched.clone(),
context_updates: (!outcome.context_updates.is_empty())
.then(|| outcome.context_updates.clone().into_iter().collect::<BTreeMap<_, _>>()),
jump_to_node: outcome.jump_to_node.clone(),
context_values: {
let snapshot = state.context.snapshot();
(!snapshot.is_empty())
.then(|| snapshot.into_iter().collect::<BTreeMap<_, _>>())
},
node_visits: (!state.node_visits.is_empty())
.then(|| state.node_visits.clone().into_iter().collect::<BTreeMap<_, _>>()),
loop_failure_signatures: None,
restart_failure_signatures: None,
attempt: result.attempts as usize,
max_attempts: result.max_attempts as usize,
});
@ -259,11 +269,13 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
let git_result = self.checkpoint_git_result.lock().unwrap().clone();
let git_sha = git_result.as_ref().and_then(|r| r.commit_sha.clone());
let diff = git_result.as_ref().and_then(|r| r.diff.clone());
self.emitter.emit(&WorkflowRunEvent::CheckpointCompleted {
node_id: node.id().to_string(),
status,
git_commit_sha: git_sha.clone(),
diff,
});
// Emit GitCommit + GitPush events if git produced results

5
lib/crates/fabro-workflow/src/lifecycle/git.rs
View file

@ -30,6 +30,7 @@ type WfNodeResult = NodeResult<Option<StageUsage>>;
pub(crate) struct GitCheckpointResult {
pub commit_sha: Option<String>,
pub push_results: Vec<(String, bool)>,
pub diff: Option<String>,
}
/// Sub-lifecycle responsible for git operations (checkpoint commits, pushes, diffs).
@ -199,6 +200,7 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
let mut git_result = GitCheckpointResult {
commit_sha: Some(sha.clone()),
push_results: Vec::new(),
diff: None,
};
match self.run_store.get_checkpoint().await {
@ -290,7 +292,8 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
match git_diff(&*self.sandbox, &prev).await {
Ok(patch) if !patch.is_empty() => {
let _ = std::fs::write(&diff_dest, patch);
let _ = std::fs::write(&diff_dest, &patch);
git_result.diff = Some(patch);
}
Ok(_) => {}
Err(err) => {

53
lib/crates/fabro-workflow/src/operations/create.rs
View file

@ -1,4 +1,5 @@
use std::collections::HashMap;
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use chrono::{Local, Utc};
@ -17,6 +18,7 @@ use crate::run_status::{RunStatus, write_run_status};
use crate::transforms::{Transform, expand_vars};
use fabro_sandbox::daytona::detect_repo_info;
use crate::event::{WorkflowRunEvent, append_progress_event, canonicalize_event_at};
use super::source::{ResolveWorkflowInput, WorkflowInput, resolve_workflow};
const RUN_CONFIG_FILE: &str = "workflow.toml";
@ -116,6 +118,11 @@ pub fn create(request: CreateRunInput) -> Result<CreatedRun, FabroError> {
write_run_config_snapshot(&run_dir, resolved.workflow_toml_path.as_deref())?;
write_run_status(&run_dir, RunStatus::Submitted, None);
emit_run_created_event(
&persisted,
&resolved.raw_source,
resolved.workflow_toml_path.as_deref(),
)?;
Ok(CreatedRun {
persisted,
@ -125,6 +132,52 @@ pub fn create(request: CreateRunInput) -> Result<CreatedRun, FabroError> {
})
}
fn emit_run_created_event(
persisted: &Persisted,
workflow_source: &str,
workflow_toml_path: Option<&Path>,
) -> Result<(), FabroError> {
let record = persisted.run_record();
let workflow_config = workflow_toml_path.and_then(|path| std::fs::read_to_string(path).ok());
let settings = sort_json_value(
serde_json::to_value(&record.settings).map_err(|err| FabroError::engine(err.to_string()))?,
);
let graph = sort_json_value(
serde_json::to_value(&record.graph).map_err(|err| FabroError::engine(err.to_string()))?,
);
let event = WorkflowRunEvent::RunCreated {
run_id: record.run_id,
settings,
graph,
workflow_source: (!workflow_source.is_empty()).then(|| workflow_source.to_string()),
workflow_config,
labels: record.labels.clone().into_iter().collect::<BTreeMap<_, _>>(),
run_dir: persisted.run_dir().display().to_string(),
working_directory: record.working_directory.display().to_string(),
host_repo_path: record.host_repo_path.clone(),
base_branch: record.base_branch.clone(),
workflow_slug: record.workflow_slug.clone(),
db_prefix: None,
};
let envelope = canonicalize_event_at(&record.run_id, &event, record.created_at);
append_progress_event(persisted.run_dir(), &envelope)
.map_err(|err| FabroError::engine(err.to_string()))
}
fn sort_json_value(value: serde_json::Value) -> serde_json::Value {
match value {
serde_json::Value::Object(map) => serde_json::Value::Object(
map.into_iter()
.map(|(key, value)| (key, sort_json_value(value)))
.collect(),
),
serde_json::Value::Array(values) => {
serde_json::Value::Array(values.into_iter().map(sort_json_value).collect())
}
other => other,
}
}
fn validate_sandbox_provider(settings: &FabroSettings) -> Result<(), FabroError> {
if let Some(provider) = settings
.sandbox_settings()

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

@ -1001,6 +1001,7 @@ mod tests {
node_id: "start".to_string(),
status: "success".to_string(),
git_commit_sha: Some("sha-test".to_string()),
diff: None,
});
}
});

10
lib/crates/fabro-workflow/src/pipeline/initialize.rs
View file

@ -512,10 +512,14 @@ pub async fn initialize(
return Err(FabroError::engine(msg));
}
options.emitter.emit(&WorkflowRunEvent::SandboxInitialized {
working_directory: sandbox.working_directory().to_string(),
});
let sandbox_record = options.sandbox.to_sandbox_record(&*sandbox);
options.emitter.emit(&WorkflowRunEvent::SandboxInitialized {
working_directory: sandbox_record.working_directory.clone(),
provider: sandbox_record.provider.clone(),
identifier: sandbox_record.identifier.clone(),
host_working_directory: sandbox_record.host_working_directory.clone(),
container_mount_point: sandbox_record.container_mount_point.clone(),
});
if let Err(err) = options.run_store.put_sandbox(&sandbox_record).await {
tracing::warn!(error = %err, "Failed to save sandbox record to store");
}

14
lib/crates/fabro-workflow/src/pipeline/retro.rs
View file

@ -55,7 +55,10 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option<Retro> {
let retro_start = std::time::Instant::now();
if let Some(ref emitter) = options.emitter {
emitter.emit(&WorkflowRunEvent::RetroStarted);
emitter.emit(&WorkflowRunEvent::RetroStarted {
provider: Some(options.provider.as_str().to_string()),
model: Some(options.model.clone()),
});
}
let narrative_result = if dry_run {
@ -68,9 +71,7 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option<Retro> {
emitter.touch();
if !matches!(
&event.event,
fabro_agent::AgentEvent::SessionStarted
| fabro_agent::AgentEvent::SessionEnded
| fabro_agent::AgentEvent::AssistantTextStart
fabro_agent::AgentEvent::AssistantTextStart
| fabro_agent::AgentEvent::AssistantOutputReplace { .. }
| fabro_agent::AgentEvent::TextDelta { .. }
| fabro_agent::AgentEvent::ReasoningDelta { .. }
@ -103,7 +104,10 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option<Retro> {
let duration_ms = u64::try_from(retro_start.elapsed().as_millis()).unwrap();
if let Some(ref emitter) = options.emitter {
match &narrative_result {
Ok(_) => emitter.emit(&WorkflowRunEvent::RetroCompleted { duration_ms }),
Ok(_) => emitter.emit(&WorkflowRunEvent::RetroCompleted {
duration_ms,
retro: serde_json::to_value(&retro).ok(),
}),
Err(e) => emitter.emit(&WorkflowRunEvent::RetroFailed {
error: e.to_string(),
duration_ms,

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

@ -12081,6 +12081,9 @@ impl Handler for KeepaliveHandler {
services.emitter.emit(&WorkflowRunEvent::Prompt {
stage: node.id.clone(),
text: "keepalive".to_string(),
mode: None,
provider: None,
model: None,
});
}
Ok(Outcome::success())