diff --git a/crates/arc-cli/tests/cli.rs b/crates/arc-cli/tests/cli.rs index 2ed055546..43dc1f477 100644 --- a/crates/arc-cli/tests/cli.rs +++ b/crates/arc-cli/tests/cli.rs @@ -543,7 +543,7 @@ fn dry_run_writes_jsonl_and_live_json() { // Events should contain WorkflowRunStarted (may not be first due to exec env events) let has_run_started = lines.iter().any(|line| { let parsed: serde_json::Value = serde_json::from_str(line).unwrap(); - parsed["event"].get("WorkflowRunStarted").is_some() + parsed["event"].as_str() == Some("WorkflowRunStarted") }); assert!( has_run_started, diff --git a/crates/arc-workflows/src/cli/run.rs b/crates/arc-workflows/src/cli/run.rs index 0731e2875..65a889bb3 100644 --- a/crates/arc-workflows/src/cli/run.rs +++ b/crates/arc-workflows/src/cli/run.rs @@ -205,11 +205,25 @@ pub async fn run_command(args: RunArgs, styles: &'static Styles) -> anyhow::Resu if let crate::event::WorkflowRunEvent::WorkflowRunStarted { run_id, .. } = event { *run_id_clone.lock().unwrap() = run_id.clone(); } - let envelope = serde_json::json!({ - "timestamp": Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true), - "run_id": *run_id_clone.lock().unwrap(), - "event": event, - }); + let (event_name, event_fields) = crate::event::flatten_event(event); + let mut envelope = serde_json::Map::new(); + envelope.insert( + "timestamp".to_string(), + serde_json::Value::String( + Utc::now() + .to_rfc3339_opts(chrono::SecondsFormat::Millis, true), + ), + ); + envelope.insert( + "run_id".to_string(), + serde_json::Value::String(run_id_clone.lock().unwrap().clone()), + ); + envelope.insert( + "event".to_string(), + serde_json::Value::String(event_name), + ); + envelope.extend(event_fields); + let envelope = serde_json::Value::Object(envelope); // Append to progress.jsonl if let Ok(line) = serde_json::to_string(&envelope) { let line = arc_util::redact::redact_jsonl_line(&line); diff --git a/crates/arc-workflows/src/event.rs b/crates/arc-workflows/src/event.rs index 65db0a9a1..968b20942 100644 --- a/crates/arc-workflows/src/event.rs +++ b/crates/arc-workflows/src/event.rs @@ -438,6 +438,161 @@ impl WorkflowRunEvent { } } +/// Flatten a `WorkflowRunEvent` into its event name and a map of top-level fields. +/// +/// Simple variants like `StageStarted` return `("StageStarted", {fields})`. +/// Wrapper variants use dot notation: +/// - `Agent { stage, event: ToolCallStarted { .. } }` → `"Agent.ToolCallStarted"` +/// - `Sandbox { event: Initializing { .. } }` → `"Sandbox.Initializing"` +/// - `Agent { stage, event: SubAgentEvent { event: inner, .. } }` → `"Agent.SubAgentEvent.{Inner}"` +/// with one level of flattening; deeper nesting stays as JSON. +pub fn flatten_event( + event: &WorkflowRunEvent, +) -> (String, serde_json::Map) { + let value = serde_json::to_value(event).expect("WorkflowRunEvent must serialize"); + match value { + serde_json::Value::Object(map) => { + // Externally-tagged enum: { "VariantName": { fields } } + let (variant_name, inner) = map.into_iter().next().expect("enum must have one key"); + match variant_name.as_str() { + "Agent" => flatten_agent(inner), + "Sandbox" => flatten_sandbox(inner), + _ => { + let fields = match inner { + serde_json::Value::Object(m) => m, + _ => serde_json::Map::new(), + }; + (variant_name, fields) + } + } + } + // Unit variants serialize as strings + serde_json::Value::String(name) => (name, serde_json::Map::new()), + _ => ("Unknown".to_string(), serde_json::Map::new()), + } +} + +fn flatten_agent(inner: serde_json::Value) -> (String, serde_json::Map) { + let serde_json::Value::Object(mut agent_fields) = inner else { + return ("Agent".to_string(), serde_json::Map::new()); + }; + let stage = agent_fields.remove("stage"); + let agent_event = agent_fields.remove("event").unwrap_or(serde_json::Value::Null); + + match agent_event { + serde_json::Value::Object(event_map) => { + let (inner_name, inner_value) = event_map + .into_iter() + .next() + .expect("agent event must have one key"); + if inner_name == "SubAgentEvent" { + flatten_sub_agent_event(stage, inner_value) + } else { + let mut fields = match inner_value { + serde_json::Value::Object(m) => m, + _ => serde_json::Map::new(), + }; + if let Some(s) = stage { + fields.insert("stage".to_string(), s); + } + (format!("Agent.{inner_name}"), fields) + } + } + // Unit variant inside Agent (e.g. SessionStarted) + serde_json::Value::String(name) => { + let mut fields = serde_json::Map::new(); + if let Some(s) = stage { + fields.insert("stage".to_string(), s); + } + (format!("Agent.{name}"), fields) + } + _ => { + let mut fields = serde_json::Map::new(); + if let Some(s) = stage { + fields.insert("stage".to_string(), s); + } + ("Agent".to_string(), fields) + } + } +} + +fn flatten_sandbox( + inner: serde_json::Value, +) -> (String, serde_json::Map) { + let serde_json::Value::Object(mut sandbox_fields) = inner else { + return ("Sandbox".to_string(), serde_json::Map::new()); + }; + let sandbox_event = sandbox_fields + .remove("event") + .unwrap_or(serde_json::Value::Null); + + match sandbox_event { + serde_json::Value::Object(event_map) => { + let (inner_name, inner_value) = event_map + .into_iter() + .next() + .expect("sandbox event must have one key"); + let fields = match inner_value { + serde_json::Value::Object(m) => m, + _ => serde_json::Map::new(), + }; + (format!("Sandbox.{inner_name}"), fields) + } + serde_json::Value::String(name) => (format!("Sandbox.{name}"), serde_json::Map::new()), + _ => ("Sandbox".to_string(), serde_json::Map::new()), + } +} + +fn flatten_sub_agent_event( + stage: Option, + inner_value: serde_json::Value, +) -> (String, serde_json::Map) { + let serde_json::Value::Object(mut sub_fields) = inner_value else { + let mut fields = serde_json::Map::new(); + if let Some(s) = stage { + fields.insert("stage".to_string(), s); + } + return ("Agent.SubAgentEvent".to_string(), fields); + }; + let agent_id = sub_fields.remove("agent_id"); + let depth = sub_fields.remove("depth"); + let nested_event = sub_fields.remove("event").unwrap_or(serde_json::Value::Null); + + let (nested_name, nested_fields) = match &nested_event { + serde_json::Value::Object(event_map) => { + let (name, value) = event_map + .iter() + .next() + .expect("nested event must have one key"); + let fields = match value { + serde_json::Value::Object(m) => m.clone(), + _ => serde_json::Map::new(), + }; + (Some(name.clone()), fields) + } + serde_json::Value::String(name) => (Some(name.clone()), serde_json::Map::new()), + _ => (None, serde_json::Map::new()), + }; + + let event_name = match &nested_name { + Some(name) => format!("Agent.SubAgentEvent.{name}"), + None => "Agent.SubAgentEvent".to_string(), + }; + + let mut fields = nested_fields; + if let Some(s) = stage { + fields.insert("stage".to_string(), s); + } + if let Some(id) = agent_id { + fields.insert("agent_id".to_string(), id); + } + if let Some(d) = depth { + fields.insert("depth".to_string(), d); + } + + (event_name, fields) +} + /// Current time as epoch milliseconds. fn epoch_millis() -> i64 { std::time::SystemTime::now() @@ -1032,6 +1187,86 @@ mod tests { ); } + #[test] + fn flatten_event_simple_variant() { + let event = WorkflowRunEvent::StageStarted { + name: "plan".to_string(), + index: 0, + handler_type: Some("codergen".to_string()), + attempt: 1, + max_attempts: 3, + }; + let (name, fields) = flatten_event(&event); + assert_eq!(name, "StageStarted"); + assert_eq!(fields["name"], "plan"); + assert_eq!(fields["index"], 0); + assert_eq!(fields["handler_type"], "codergen"); + assert_eq!(fields["attempt"], 1); + assert_eq!(fields["max_attempts"], 3); + } + + #[test] + fn flatten_event_agent_tool_call_started() { + let event = WorkflowRunEvent::Agent { + stage: "code".to_string(), + event: AgentEvent::ToolCallStarted { + tool_name: "read_file".to_string(), + tool_call_id: "call_1".to_string(), + arguments: serde_json::json!({"path": "/tmp/test.txt"}), + }, + }; + let (name, fields) = flatten_event(&event); + assert_eq!(name, "Agent.ToolCallStarted"); + assert_eq!(fields["stage"], "code"); + assert_eq!(fields["tool_name"], "read_file"); + assert_eq!(fields["tool_call_id"], "call_1"); + } + + #[test] + fn flatten_event_sandbox_initializing() { + let event = WorkflowRunEvent::Sandbox { + event: SandboxEvent::Initializing { + provider: "docker".into(), + }, + }; + let (name, fields) = flatten_event(&event); + assert_eq!(name, "Sandbox.Initializing"); + assert_eq!(fields["provider"], "docker"); + } + + #[test] + fn flatten_event_agent_sub_agent_event() { + let event = WorkflowRunEvent::Agent { + stage: "code".to_string(), + event: AgentEvent::SubAgentEvent { + agent_id: "sub_1".to_string(), + depth: 1, + event: Box::new(AgentEvent::ToolCallStarted { + tool_name: "write_file".to_string(), + tool_call_id: "call_2".to_string(), + arguments: serde_json::json!({}), + }), + }, + }; + let (name, fields) = flatten_event(&event); + assert_eq!(name, "Agent.SubAgentEvent.ToolCallStarted"); + assert_eq!(fields["stage"], "code"); + assert_eq!(fields["agent_id"], "sub_1"); + assert_eq!(fields["depth"], 1); + assert_eq!(fields["tool_name"], "write_file"); + } + + #[test] + fn flatten_event_agent_session_started() { + let event = WorkflowRunEvent::Agent { + stage: "plan".to_string(), + event: AgentEvent::SessionStarted, + }; + let (name, fields) = flatten_event(&event); + assert_eq!(name, "Agent.SessionStarted"); + assert_eq!(fields["stage"], "plan"); + } + #[test] fn setup_events_serialization() { let events = vec![ diff --git a/crates/arc-workflows/src/retro.rs b/crates/arc-workflows/src/retro.rs index 458b786d9..bde8a50dd 100644 --- a/crates/arc-workflows/src/retro.rs +++ b/crates/arc-workflows/src/retro.rs @@ -6,7 +6,6 @@ use serde::{Deserialize, Serialize}; use crate::checkpoint::Checkpoint; use crate::error::{ArcError, Result}; -use crate::event::WorkflowRunEvent; use crate::outcome::StageStatus; #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -183,18 +182,16 @@ pub fn extract_stage_durations(logs_root: &Path) -> HashMap { let Ok(envelope) = serde_json::from_str::(line) else { continue; }; - let Some(event_value) = envelope.get("event") else { + if envelope.get("event").and_then(|v| v.as_str()) != Some("StageCompleted") { continue; - }; - let Ok(event) = serde_json::from_value::(event_value.clone()) else { - continue; - }; - if let WorkflowRunEvent::StageCompleted { - name, duration_ms, .. - } = event - { - durations.insert(name, duration_ms); } + let Some(name) = envelope.get("name").and_then(|v| v.as_str()) else { + continue; + }; + let Some(duration_ms) = envelope.get("duration_ms").and_then(|v| v.as_u64()) else { + continue; + }; + durations.insert(name.to_string(), duration_ms); } durations } @@ -524,44 +521,38 @@ mod tests { let event1 = serde_json::json!({ "timestamp": "2025-01-01T00:00:00.000Z", "run_id": "r1", - "event": { - "StageCompleted": { - "name": "plan", - "index": 0, - "duration_ms": 5000, - "status": "success", - "preferred_label": null, - "suggested_next_ids": [], - "usage": null, - "failure_reason": null, - "notes": null, - "files_touched": [], - "attempt": 1, - "max_attempts": 1, - "failure_class": null - } - } + "event": "StageCompleted", + "name": "plan", + "index": 0, + "duration_ms": 5000, + "status": "success", + "preferred_label": null, + "suggested_next_ids": [], + "usage": null, + "failure_reason": null, + "notes": null, + "files_touched": [], + "attempt": 1, + "max_attempts": 1, + "failure_class": null }); let event2 = serde_json::json!({ "timestamp": "2025-01-01T00:00:05.000Z", "run_id": "r1", - "event": { - "StageCompleted": { - "name": "code", - "index": 1, - "duration_ms": 15000, - "status": "success", - "preferred_label": null, - "suggested_next_ids": [], - "usage": null, - "failure_reason": null, - "notes": null, - "files_touched": [], - "attempt": 1, - "max_attempts": 1, - "failure_class": null - } - } + "event": "StageCompleted", + "name": "code", + "index": 1, + "duration_ms": 15000, + "status": "success", + "preferred_label": null, + "suggested_next_ids": [], + "usage": null, + "failure_reason": null, + "notes": null, + "files_touched": [], + "attempt": 1, + "max_attempts": 1, + "failure_class": null }); let content = format!( "{}\n{}\n",