From 212cffeb1dff01e330ad56361150e29044fd6852 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Mon, 11 May 2026 11:25:44 -0400 Subject: [PATCH] feat: project ACP backend events --- lib/crates/fabro-server/src/server.rs | 69 ++++--- .../fabro-server/src/server/handler/steer.rs | 10 +- lib/crates/fabro-server/src/server/tests.rs | 144 ++++++++++++++ lib/crates/fabro-store/src/run_state.rs | 186 +++++++++++++++++- lib/crates/fabro-types/src/run_event/misc.rs | 31 +++ lib/crates/fabro-types/src/run_event/mod.rs | 18 ++ .../fabro-workflow/src/event/convert.rs | 150 ++++++++++++++ lib/crates/fabro-workflow/src/event/events.rs | 57 ++++++ lib/crates/fabro-workflow/src/event/names.rs | 4 + .../fabro-workflow/src/event/stored_fields.rs | 15 +- .../fabro-workflow/src/handler/llm/acp.rs | 104 +++++++++- .../fabro-workflow/src/operations/fork.rs | 38 ++++ 12 files changed, 775 insertions(+), 51 deletions(-) diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index e099f8d75..386f5021c 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -189,30 +189,30 @@ impl ListResponse { /// Snapshot of a managed run. struct ManagedRun { - dot_source: String, - status: RunStatus, - error: Option, - created_at: chrono::DateTime, - enqueued_at: Instant, + dot_source: String, + status: RunStatus, + error: Option, + created_at: chrono::DateTime, + enqueued_at: Instant, // Populated when running: - answer_transport: Option, + answer_transport: Option, accepted_questions: HashSet, /// Stage IDs of currently steerable API-mode (SDK) agent sessions, /// keyed to the session id that owns the active lease. Used by the /// steerability predicate. - active_api_stages: HashMap, - /// Stage IDs of currently running CLI-mode agent sessions, observed - /// from `agent.cli.started/completed` plus `stage.completed`/ + active_api_stages: HashMap, + /// Stage IDs of currently running non-steerable agent sessions, observed + /// from CLI/ACP start/completion events plus `stage.completed`/ /// `stage.failed` backstops. - active_cli_stages: HashSet, - event_tx: Option>, - checkpoint: Option, - cancel_tx: Option>, - cancel_token: Option, - worker_pid: Option, - worker_pgid: Option, - run_dir: Option, - execution_mode: RunExecutionMode, + active_non_steerable_agent_stages: HashSet, + event_tx: Option>, + checkpoint: Option, + cancel_tx: Option>, + cancel_token: Option, + worker_pid: Option, + worker_pgid: Option, + run_dir: Option, + execution_mode: RunExecutionMode, } #[derive(Clone, Copy)] @@ -1953,7 +1953,7 @@ fn clear_live_run_state(run: &mut ManagedRun) { run.answer_transport = None; run.accepted_questions.clear(); run.active_api_stages.clear(); - run.active_cli_stages.clear(); + run.active_non_steerable_agent_stages.clear(); run.event_tx = None; run.cancel_tx = None; run.cancel_token = None; @@ -2291,7 +2291,7 @@ fn managed_run( answer_transport: None, accepted_questions: HashSet::new(), active_api_stages: HashMap::new(), - active_cli_stages: HashSet::new(), + active_non_steerable_agent_stages: HashSet::new(), event_tx: None, checkpoint: None, cancel_tx: None, @@ -2388,7 +2388,7 @@ fn update_live_run_from_event(state: &AppState, run_id: RunId, event: &RunEvent) }; managed_run.error = None; managed_run.active_api_stages.clear(); - managed_run.active_cli_stages.clear(); + managed_run.active_non_steerable_agent_stages.clear(); } EventBody::RunFailed(props) => { managed_run.status = RunStatus::Failed { @@ -2396,7 +2396,7 @@ fn update_live_run_from_event(state: &AppState, run_id: RunId, event: &RunEvent) }; managed_run.error = Some(props.error.clone()); managed_run.active_api_stages.clear(); - managed_run.active_cli_stages.clear(); + managed_run.active_non_steerable_agent_stages.clear(); } // Track API-mode steerable sessions. Activated/deactivated are // leased by session id so stale deactivations cannot clear a newer @@ -2425,17 +2425,24 @@ fn update_live_run_from_event(state: &AppState, run_id: RunId, event: &RunEvent) } } } - // Track CLI-mode agent stages. CLI started/completed are coarser - // and sometimes fail to emit `completed` on error paths — the - // stage.completed/stage.failed handler below is the backstop. - EventBody::AgentCliStarted(_) => { + // Track non-steerable agent stages. CLI/ACP started/completed are + // coarser and sometimes fail to emit terminal events on error paths; + // stage.completed/stage.failed below are the backstops. + EventBody::AgentCliStarted(_) | EventBody::AgentAcpStarted(_) => { if let Some(stage_id) = event.stage_id.as_ref() { - managed_run.active_cli_stages.insert(stage_id.clone()); + managed_run + .active_non_steerable_agent_stages + .insert(stage_id.clone()); } } - EventBody::AgentCliCompleted(_) => { + EventBody::AgentCliCompleted(_) + | EventBody::AgentAcpCompleted(_) + | EventBody::AgentAcpCancelled(_) + | EventBody::AgentAcpTimedOut(_) => { if let Some(stage_id) = &event.stage_id { - managed_run.active_cli_stages.remove(stage_id); + managed_run + .active_non_steerable_agent_stages + .remove(stage_id); } } // Stage lifecycle backstop: cover both completion and failure @@ -2443,7 +2450,9 @@ fn update_live_run_from_event(state: &AppState, run_id: RunId, event: &RunEvent) EventBody::StageCompleted(_) | EventBody::StageFailed(_) => { if let Some(stage_id) = &event.stage_id { managed_run.active_api_stages.remove(stage_id); - managed_run.active_cli_stages.remove(stage_id); + managed_run + .active_non_steerable_agent_stages + .remove(stage_id); } } _ => {} diff --git a/lib/crates/fabro-server/src/server/handler/steer.rs b/lib/crates/fabro-server/src/server/handler/steer.rs index 6ee20497c..98a4b319a 100644 --- a/lib/crates/fabro-server/src/server/handler/steer.rs +++ b/lib/crates/fabro-server/src/server/handler/steer.rs @@ -124,12 +124,14 @@ async fn control_run( // - If at least one API-mode session is active → forward. // - Else if no agent stages are active at all → forward (worker hub buffers // for the next session). - // - Else (active agents exist but all are CLI-mode) → 409. - if managed_run.active_api_stages.is_empty() && !managed_run.active_cli_stages.is_empty() { + // - Else (active agents exist but all are non-steerable) → 409. + if managed_run.active_api_stages.is_empty() + && !managed_run.active_non_steerable_agent_stages.is_empty() + { return ApiError::with_code( StatusCode::CONFLICT, - "All currently running agent stages are CLI-mode and cannot be steered.", - "cli_agent_not_steerable", + "All currently running agent stages use a non-steerable backend.", + "agent_not_steerable", ) .into_response(); } diff --git a/lib/crates/fabro-server/src/server/tests.rs b/lib/crates/fabro-server/src/server/tests.rs index 0e0d7aa0a..e67707ee9 100644 --- a/lib/crates/fabro-server/src/server/tests.rs +++ b/lib/crates/fabro-server/src/server/tests.rs @@ -7164,6 +7164,150 @@ fn active_api_stage_projection_ignores_stale_deactivation() { ); } +fn acp_event_for_stage(run_id: &RunId, event: &workflow_event::Event) -> fabro_types::RunEvent { + workflow_event::to_run_event_at( + run_id, + event, + Utc::now(), + Some(&workflow_event::StageScope { + node_id: "agent".to_string(), + visit: 1, + parallel_group_id: None, + parallel_branch_id: None, + }), + ) +} + +#[tokio::test] +async fn steer_with_active_acp_stage_returns_non_steerable_conflict() { + let state = test_app_state(); + let app = crate::test_support::build_test_router(Arc::clone(&state)); + let run_id = fixtures::RUN_1; + let (control_tx, _control_rx) = tokio::sync::mpsc::channel(1); + let _temp_dir = insert_running_control_run( + &state, + run_id, + Some(RunAnswerTransport::Subprocess { control_tx }), + ); + + let started = acp_event_for_stage(&run_id, &workflow_event::Event::AgentAcpStarted { + node_id: "agent".to_string(), + visit: 1, + mode: "acp".to_string(), + provider: "openai".to_string(), + model: "fake-acp".to_string(), + command: "python fake_agent.py".to_string(), + }); + update_live_run_from_event(&state, run_id, &started); + + let req = Request::builder() + .method("POST") + .uri(api(&format!("/runs/{run_id}/steer"))) + .header("content-type", "application/json") + .body(Body::from(r#"{"text":"try again"}"#)) + .unwrap(); + + let response = app.oneshot(req).await.unwrap(); + assert_eq!(response.status(), StatusCode::CONFLICT); + let body = body_json(response.into_body()).await; + assert_eq!(body["errors"][0]["code"], "agent_not_steerable"); +} + +#[tokio::test] +async fn active_acp_stage_marker_clears_on_terminal_paths() { + let terminal_events: Vec = vec![ + workflow_event::Event::AgentAcpCompleted { + node_id: "agent".to_string(), + stdout: "done".to_string(), + stderr: String::new(), + stop_reason: "end_turn".to_string(), + duration_ms: 42, + }, + workflow_event::Event::AgentAcpCancelled { + node_id: "agent".to_string(), + stdout: "partial".to_string(), + stderr: "cancelled".to_string(), + duration_ms: 7, + }, + workflow_event::Event::AgentAcpTimedOut { + node_id: "agent".to_string(), + stdout: "partial".to_string(), + stderr: "timeout".to_string(), + duration_ms: 99, + }, + workflow_event::Event::StageCompleted { + node_id: "agent".to_string(), + name: "agent".to_string(), + index: 0, + duration_ms: 1, + status: "success".to_string(), + preferred_label: None, + suggested_next_ids: Vec::new(), + billing: None, + 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, + response: None, + attempt: 1, + max_attempts: 1, + }, + workflow_event::Event::StageFailed { + node_id: "agent".to_string(), + name: "agent".to_string(), + index: 0, + failure: FailureDetail::new("failed", FailureCategory::Deterministic), + will_retry: false, + duration_ms: 1, + billing: None, + actor: None, + }, + ]; + + for terminal_event in terminal_events { + let state = test_app_state(); + let app = crate::test_support::build_test_router(Arc::clone(&state)); + let run_id = fixtures::RUN_1; + let (control_tx, mut control_rx) = tokio::sync::mpsc::channel(1); + let _temp_dir = insert_running_control_run( + &state, + run_id, + Some(RunAnswerTransport::Subprocess { control_tx }), + ); + let started = acp_event_for_stage(&run_id, &workflow_event::Event::AgentAcpStarted { + node_id: "agent".to_string(), + visit: 1, + mode: "acp".to_string(), + provider: "openai".to_string(), + model: "fake-acp".to_string(), + command: "python fake_agent.py".to_string(), + }); + update_live_run_from_event(&state, run_id, &started); + let terminal = acp_event_for_stage(&run_id, &terminal_event); + update_live_run_from_event(&state, run_id, &terminal); + + let req = Request::builder() + .method("POST") + .uri(api(&format!("/runs/{run_id}/steer"))) + .header("content-type", "application/json") + .body(Body::from(r#"{"text":"try again"}"#)) + .unwrap(); + + let response = app.oneshot(req).await.unwrap(); + assert_status!(response, StatusCode::ACCEPTED).await; + let envelope = control_rx.recv().await.unwrap(); + assert!(matches!( + envelope.message, + WorkerControlMessage::Steer { ref text, .. } if text == "try again" + )); + } +} + #[tokio::test] async fn get_graph_returns_svg() { let state = test_app_state(); diff --git a/lib/crates/fabro-store/src/run_state.rs b/lib/crates/fabro-store/src/run_state.rs index d1bdf7b2a..5f79dd4c2 100644 --- a/lib/crates/fabro-store/src/run_state.rs +++ b/lib/crates/fabro-store/src/run_state.rs @@ -3,8 +3,9 @@ use std::str::FromStr; use chrono::{DateTime, Utc}; use fabro_types::run_event::{ - AgentCliStartedProps, AgentSessionActivatedProps, CheckpointCompletedProps, RunCompletedProps, - RunFailedProps, StageCompletedProps, StagePromptProps, + AgentAcpStartedProps, AgentCliStartedProps, AgentSessionActivatedProps, + CheckpointCompletedProps, RunCompletedProps, RunFailedProps, StageCompletedProps, + StagePromptProps, }; use fabro_types::settings::run::RunSandboxSettings; use fabro_types::{ @@ -371,6 +372,13 @@ impl RunProjectionReducer for RunProjection { }; stage.provider_used = Some(provider_used_from_agent_cli_started(props)); } + EventBody::AgentAcpStarted(props) => { + let Some(stage) = stage_at_stored_or_visit(self, stored, props.visit, event.seq) + else { + return Ok(()); + }; + stage.provider_used = Some(provider_used_from_agent_acp_started(props)); + } EventBody::CommandStarted(props) => { let script_invocation = serde_json::to_value(props).map_err(|err| { Error::InvalidEvent(format!("invalid command.started payload: {err}")) @@ -426,6 +434,39 @@ impl RunProjectionReducer for RunProjection { CommandTermination::TimedOut, )?; } + EventBody::AgentAcpCompleted(props) => { + let Some(stage) = stage_at_current_visit(self, stored, event.seq) else { + return Ok(()); + }; + apply_agent_acp_terminal( + stage, + props, + merge_agent_cli_output(&props.stdout, &props.stderr), + CommandTermination::Exited, + )?; + } + EventBody::AgentAcpCancelled(props) => { + let Some(stage) = stage_at_current_visit(self, stored, event.seq) else { + return Ok(()); + }; + apply_agent_acp_terminal( + stage, + props, + merge_agent_cli_output(&props.stdout, &props.stderr), + CommandTermination::Cancelled, + )?; + } + EventBody::AgentAcpTimedOut(props) => { + let Some(stage) = stage_at_current_visit(self, stored, event.seq) else { + return Ok(()); + }; + apply_agent_acp_terminal( + stage, + props, + merge_agent_cli_output(&props.stdout, &props.stderr), + CommandTermination::TimedOut, + )?; + } EventBody::ParallelCompleted(props) => { let parallel_results = serde_json::to_value(&props.results).map_err(|err| { Error::InvalidEvent(format!("invalid parallel.completed payload: {err}")) @@ -848,6 +889,18 @@ fn provider_used_from_agent_cli_started(props: &AgentCliStartedProps) -> Value { Value::Object(provider_used) } +fn provider_used_from_agent_acp_started(props: &AgentAcpStartedProps) -> Value { + let mut provider_used = serde_json::Map::new(); + provider_used.insert("mode".to_string(), Value::String("acp".to_string())); + provider_used.insert( + "provider".to_string(), + Value::String(props.provider.clone()), + ); + provider_used.insert("model".to_string(), Value::String(props.model.clone())); + provider_used.insert("command".to_string(), Value::String(props.command.clone())); + Value::Object(provider_used) +} + fn apply_agent_cli_terminal( stage: &mut StageProjection, props: &impl serde::Serialize, @@ -862,6 +915,20 @@ fn apply_agent_cli_terminal( Ok(()) } +fn apply_agent_acp_terminal( + stage: &mut StageProjection, + props: &impl serde::Serialize, + output: String, + termination: CommandTermination, +) -> Result<()> { + let script_timing = serde_json::to_value(props) + .map_err(|err| Error::InvalidEvent(format!("invalid agent.acp terminal payload: {err}")))?; + stage.output = Some(output); + stage.termination = Some(termination); + stage.script_timing = Some(script_timing); + Ok(()) +} + fn merge_agent_cli_output(stdout: &str, stderr: &str) -> String { match (stdout.is_empty(), stderr.is_empty()) { (true, true) => String::new(), @@ -878,11 +945,13 @@ mod tests { use chrono::Utc; use fabro_types::run_event::run::RunFailedProps; use fabro_types::run_event::{ - AgentCliCancelledProps, AgentCliCompletedProps, AgentCliTimedOutProps, AgentMessageProps, - AgentSessionActivatedProps, AgentSessionEndedProps, AgentSessionStartedProps, - CheckpointCompletedProps, InterviewCompletedProps, InterviewOption, InterviewStartedProps, - RunControlEffectProps, StageCompletedProps, StageFailedProps, StagePromptProps, - StageRetryingProps, StageStartedProps, + AgentAcpCancelledProps, AgentAcpCompletedProps, AgentAcpStartedProps, + AgentAcpTimedOutProps, AgentCliCancelledProps, AgentCliCompletedProps, + AgentCliTimedOutProps, AgentMessageProps, AgentSessionActivatedProps, + AgentSessionEndedProps, AgentSessionStartedProps, CheckpointCompletedProps, + InterviewCompletedProps, InterviewOption, InterviewStartedProps, RunControlEffectProps, + StageCompletedProps, StageFailedProps, StagePromptProps, StageRetryingProps, + StageStartedProps, }; use fabro_types::{ BilledModelUsage, BilledTokenCounts, BlockedReason, Checkpoint, CheckpointRecord, @@ -1287,6 +1356,109 @@ mod tests { assert!(stage.provider_used.is_none()); } + #[test] + fn agent_acp_started_updates_stage_provider_used() { + let mut state = initialized_projection(); + let stage_id = StageId::new("code", 1); + start_stage(&mut state, &stage_id); + + state + .apply_event(&test_stage_event( + 4, + EventBody::AgentAcpStarted(AgentAcpStartedProps { + visit: 1, + mode: "acp".to_string(), + provider: "openai".to_string(), + model: "fake-acp".to_string(), + command: "python fake_agent.py".to_string(), + }), + stage_id.clone(), + )) + .unwrap(); + + let stage = state.stage(&stage_id).unwrap(); + assert_eq!( + stage.provider_used.as_ref().unwrap(), + &json!({ + "mode": "acp", + "provider": "openai", + "model": "fake-acp", + "command": "python fake_agent.py" + }) + ); + } + + #[test] + fn agent_acp_completed_updates_stage_output_projection() { + let mut state = initialized_projection(); + let stage_id = StageId::new("code", 1); + start_stage(&mut state, &stage_id); + + state + .apply_event(&test_stage_event( + 4, + EventBody::AgentAcpCompleted(AgentAcpCompletedProps { + stdout: "done".to_string(), + stderr: "warn".to_string(), + stop_reason: "end_turn".to_string(), + duration_ms: 42, + }), + stage_id.clone(), + )) + .unwrap(); + + let stage = state.stage(&stage_id).unwrap(); + assert_eq!(stage.output.as_deref(), Some("done\nwarn")); + assert_eq!(stage.termination, Some(CommandTermination::Exited)); + assert_eq!( + stage.script_timing.as_ref().unwrap()["stop_reason"], + serde_json::json!("end_turn") + ); + } + + #[test] + fn agent_acp_cancelled_and_timed_out_update_terminal_projection() { + let mut cancelled = initialized_projection(); + let cancelled_stage_id = StageId::new("cancelled", 1); + start_stage(&mut cancelled, &cancelled_stage_id); + + cancelled + .apply_event(&test_stage_event( + 4, + EventBody::AgentAcpCancelled(AgentAcpCancelledProps { + stdout: "partial".to_string(), + stderr: "cancelled".to_string(), + duration_ms: 7, + }), + cancelled_stage_id.clone(), + )) + .unwrap(); + + let stage = cancelled.stage(&cancelled_stage_id).unwrap(); + assert_eq!(stage.output.as_deref(), Some("partial\ncancelled")); + assert_eq!(stage.termination, Some(CommandTermination::Cancelled)); + + let mut timed_out = initialized_projection(); + let timed_out_stage_id = StageId::new("timed_out", 1); + start_stage(&mut timed_out, &timed_out_stage_id); + + timed_out + .apply_event(&test_stage_event( + 4, + EventBody::AgentAcpTimedOut(AgentAcpTimedOutProps { + stdout: "partial".to_string(), + stderr: "timeout".to_string(), + duration_ms: 99, + }), + timed_out_stage_id.clone(), + )) + .unwrap(); + + let stage = timed_out.stage(&timed_out_stage_id).unwrap(); + assert_eq!(stage.output.as_deref(), Some("partial\ntimeout")); + assert_eq!(stage.termination, Some(CommandTermination::TimedOut)); + } + #[test] fn agent_cli_completed_updates_stage_output_projection() { let mut state = initialized_projection(); diff --git a/lib/crates/fabro-types/src/run_event/misc.rs b/lib/crates/fabro-types/src/run_event/misc.rs index 86dba4f64..22c1f340e 100644 --- a/lib/crates/fabro-types/src/run_event/misc.rs +++ b/lib/crates/fabro-types/src/run_event/misc.rs @@ -335,6 +335,37 @@ pub struct AgentCliTimedOutProps { pub duration_ms: u64, } +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct AgentAcpStartedProps { + pub visit: u32, + pub mode: String, + pub provider: String, + pub model: String, + pub command: String, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct AgentAcpCompletedProps { + pub stdout: String, + pub stderr: String, + pub stop_reason: String, + pub duration_ms: u64, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct AgentAcpCancelledProps { + pub stdout: String, + pub stderr: String, + pub duration_ms: u64, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct AgentAcpTimedOutProps { + pub stdout: String, + pub stderr: String, + pub duration_ms: u64, +} + #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct PullRequestCreatedProps { pub pr_url: String, diff --git a/lib/crates/fabro-types/src/run_event/mod.rs b/lib/crates/fabro-types/src/run_event/mod.rs index d9887e016..b473ad6b4 100644 --- a/lib/crates/fabro-types/src/run_event/mod.rs +++ b/lib/crates/fabro-types/src/run_event/mod.rs @@ -290,6 +290,14 @@ pub enum EventBody { AgentCliCancelled(AgentCliCancelledProps), #[serde(rename = "agent.cli.timed_out")] AgentCliTimedOut(AgentCliTimedOutProps), + #[serde(rename = "agent.acp.started")] + AgentAcpStarted(AgentAcpStartedProps), + #[serde(rename = "agent.acp.completed")] + AgentAcpCompleted(AgentAcpCompletedProps), + #[serde(rename = "agent.acp.cancelled")] + AgentAcpCancelled(AgentAcpCancelledProps), + #[serde(rename = "agent.acp.timed_out")] + AgentAcpTimedOut(AgentAcpTimedOutProps), #[serde(rename = "pull_request.created")] PullRequestCreated(PullRequestCreatedProps), #[serde(rename = "pull_request.failed")] @@ -484,6 +492,10 @@ impl EventBody { Self::AgentCliCompleted(_) => "agent.cli.completed", Self::AgentCliCancelled(_) => "agent.cli.cancelled", Self::AgentCliTimedOut(_) => "agent.cli.timed_out", + Self::AgentAcpStarted(_) => "agent.acp.started", + Self::AgentAcpCompleted(_) => "agent.acp.completed", + Self::AgentAcpCancelled(_) => "agent.acp.cancelled", + Self::AgentAcpTimedOut(_) => "agent.acp.timed_out", Self::PullRequestCreated(_) => "pull_request.created", Self::PullRequestFailed(_) => "pull_request.failed", Self::DevcontainerResolved(_) => "devcontainer.resolved", @@ -629,6 +641,12 @@ fn is_known_event_name(event: &str) -> bool { | "command.completed" | "agent.cli.started" | "agent.cli.completed" + | "agent.cli.cancelled" + | "agent.cli.timed_out" + | "agent.acp.started" + | "agent.acp.completed" + | "agent.acp.cancelled" + | "agent.acp.timed_out" | "pull_request.created" | "pull_request.failed" | "devcontainer.resolved" diff --git a/lib/crates/fabro-workflow/src/event/convert.rs b/lib/crates/fabro-workflow/src/event/convert.rs index 6495b24e6..58998f26e 100644 --- a/lib/crates/fabro-workflow/src/event/convert.rs +++ b/lib/crates/fabro-workflow/src/event/convert.rs @@ -1134,6 +1134,52 @@ fn event_body_from_event(event: &Event) -> EventBody { stderr: stderr.clone(), duration_ms: *duration_ms, }), + Event::AgentAcpStarted { + visit, + mode, + provider, + model, + command, + .. + } => EventBody::AgentAcpStarted(fabro_types::AgentAcpStartedProps { + visit: *visit, + mode: mode.clone(), + provider: provider.clone(), + model: model.clone(), + command: command.clone(), + }), + Event::AgentAcpCompleted { + stdout, + stderr, + stop_reason, + duration_ms, + .. + } => EventBody::AgentAcpCompleted(fabro_types::AgentAcpCompletedProps { + stdout: stdout.clone(), + stderr: stderr.clone(), + stop_reason: stop_reason.clone(), + duration_ms: *duration_ms, + }), + Event::AgentAcpCancelled { + stdout, + stderr, + duration_ms, + .. + } => EventBody::AgentAcpCancelled(fabro_types::AgentAcpCancelledProps { + stdout: stdout.clone(), + stderr: stderr.clone(), + duration_ms: *duration_ms, + }), + Event::AgentAcpTimedOut { + stdout, + stderr, + duration_ms, + .. + } => EventBody::AgentAcpTimedOut(fabro_types::AgentAcpTimedOutProps { + stdout: stdout.clone(), + stderr: stderr.clone(), + duration_ms: *duration_ms, + }), Event::PullRequestCreated { pr_url, pr_number, @@ -2009,6 +2055,110 @@ mod tests { } } + #[test] + fn agent_acp_events_map_to_event_bodies_with_stage_scope() { + let scope = StageScope { + node_id: "code".to_string(), + visit: 2, + parallel_group_id: Some(StageId::new("fanout", 1)), + parallel_branch_id: Some(ParallelBranchId::new(StageId::new("fanout", 1), 0)), + }; + + let started = to_run_event_at( + &fixtures::RUN_1, + &Event::AgentAcpStarted { + node_id: "code".to_string(), + visit: 2, + mode: "acp".to_string(), + provider: "openai".to_string(), + model: "fake-acp".to_string(), + command: "python fake_agent.py".to_string(), + }, + Utc::now(), + Some(&scope), + ); + assert_eq!(started.event_name(), "agent.acp.started"); + assert_eq!(started.node_id.as_deref(), Some("code")); + assert_eq!(started.stage_id, Some(StageId::new("code", 2))); + assert_eq!(started.parallel_group_id, scope.parallel_group_id); + assert_eq!(started.parallel_branch_id, scope.parallel_branch_id); + match &started.body { + EventBody::AgentAcpStarted(props) => { + assert_eq!(props.visit, 2); + assert_eq!(props.mode, "acp"); + assert_eq!(props.provider, "openai"); + assert_eq!(props.model, "fake-acp"); + assert_eq!(props.command, "python fake_agent.py"); + } + other => panic!("expected AgentAcpStarted, got {other:?}"), + } + + let completed = to_run_event_at( + &fixtures::RUN_1, + &Event::AgentAcpCompleted { + node_id: "code".to_string(), + stdout: "done".to_string(), + stderr: "warn".to_string(), + stop_reason: "end_turn".to_string(), + duration_ms: 42, + }, + Utc::now(), + Some(&scope), + ); + assert_eq!(completed.event_name(), "agent.acp.completed"); + match &completed.body { + EventBody::AgentAcpCompleted(props) => { + assert_eq!(props.stdout, "done"); + assert_eq!(props.stderr, "warn"); + assert_eq!(props.stop_reason, "end_turn"); + assert_eq!(props.duration_ms, 42); + } + other => panic!("expected AgentAcpCompleted, got {other:?}"), + } + + let cancelled = to_run_event_at( + &fixtures::RUN_1, + &Event::AgentAcpCancelled { + node_id: "code".to_string(), + stdout: "partial".to_string(), + stderr: "cancelled".to_string(), + duration_ms: 7, + }, + Utc::now(), + Some(&scope), + ); + assert_eq!(cancelled.event_name(), "agent.acp.cancelled"); + assert_eq!(cancelled.stage_id, Some(StageId::new("code", 2))); + assert!(matches!( + cancelled.body, + EventBody::AgentAcpCancelled(fabro_types::AgentAcpCancelledProps { + duration_ms: 7, + .. + }) + )); + + let timed_out = to_run_event_at( + &fixtures::RUN_1, + &Event::AgentAcpTimedOut { + node_id: "code".to_string(), + stdout: "partial".to_string(), + stderr: "timeout".to_string(), + duration_ms: 99, + }, + Utc::now(), + Some(&scope), + ); + assert_eq!(timed_out.event_name(), "agent.acp.timed_out"); + assert_eq!(timed_out.stage_id, Some(StageId::new("code", 2))); + assert!(matches!( + timed_out.body, + EventBody::AgentAcpTimedOut(fabro_types::AgentAcpTimedOutProps { + duration_ms: 99, + .. + }) + )); + } + #[test] fn stall_watchdog_timeout_populates_watchdog_actor() { let stored = to_run_event(&fixtures::RUN_1, &Event::StallWatchdogTimeout { diff --git a/lib/crates/fabro-workflow/src/event/events.rs b/lib/crates/fabro-workflow/src/event/events.rs index 9d3a76cad..eac03f337 100644 --- a/lib/crates/fabro-workflow/src/event/events.rs +++ b/lib/crates/fabro-workflow/src/event/events.rs @@ -621,6 +621,33 @@ pub enum Event { stderr: String, duration_ms: u64, }, + AgentAcpStarted { + node_id: String, + visit: u32, + mode: String, + provider: String, + model: String, + command: String, + }, + AgentAcpCompleted { + node_id: String, + stdout: String, + stderr: String, + stop_reason: String, + duration_ms: u64, + }, + AgentAcpCancelled { + node_id: String, + stdout: String, + stderr: String, + duration_ms: u64, + }, + AgentAcpTimedOut { + node_id: String, + stdout: String, + stderr: String, + duration_ms: u64, + }, PullRequestCreated { pr_url: String, pr_number: u64, @@ -1378,6 +1405,36 @@ impl Event { } => { debug!(node_id, duration_ms, "Agent CLI timed out"); } + Self::AgentAcpStarted { + node_id, + provider, + model, + .. + } => { + debug!(node_id, provider, model, "Agent ACP started"); + } + Self::AgentAcpCompleted { + node_id, + stop_reason, + duration_ms, + .. + } => { + debug!(node_id, stop_reason, duration_ms, "Agent ACP completed"); + } + Self::AgentAcpCancelled { + node_id, + duration_ms, + .. + } => { + debug!(node_id, duration_ms, "Agent ACP cancelled"); + } + Self::AgentAcpTimedOut { + node_id, + duration_ms, + .. + } => { + debug!(node_id, duration_ms, "Agent ACP timed out"); + } Self::PullRequestCreated { pr_url, pr_number, diff --git a/lib/crates/fabro-workflow/src/event/names.rs b/lib/crates/fabro-workflow/src/event/names.rs index a4a630b8f..45b8aa0db 100644 --- a/lib/crates/fabro-workflow/src/event/names.rs +++ b/lib/crates/fabro-workflow/src/event/names.rs @@ -137,6 +137,10 @@ pub fn event_name(event: &Event) -> &'static str { Event::AgentSteerDropped { .. } => "agent.steer.dropped", Event::AgentCliCancelled { .. } => "agent.cli.cancelled", Event::AgentCliTimedOut { .. } => "agent.cli.timed_out", + Event::AgentAcpStarted { .. } => "agent.acp.started", + Event::AgentAcpCompleted { .. } => "agent.acp.completed", + Event::AgentAcpCancelled { .. } => "agent.acp.cancelled", + Event::AgentAcpTimedOut { .. } => "agent.acp.timed_out", Event::PullRequestCreated { .. } => "pull_request.created", Event::PullRequestFailed { .. } => "pull_request.failed", Event::DevcontainerResolved { .. } => "devcontainer.resolved", diff --git a/lib/crates/fabro-workflow/src/event/stored_fields.rs b/lib/crates/fabro-workflow/src/event/stored_fields.rs index afa202c1f..20abd7a96 100644 --- a/lib/crates/fabro-workflow/src/event/stored_fields.rs +++ b/lib/crates/fabro-workflow/src/event/stored_fields.rs @@ -122,7 +122,20 @@ fn stored_event_fields_for_variant(event: &Event) -> StoredEventFields { | Event::AgentCliStarted { node_id, .. } | Event::AgentCliCompleted { node_id, .. } | Event::AgentCliCancelled { node_id, .. } - | Event::AgentCliTimedOut { node_id, .. } => node_stored_fields(Some(node_id.clone())), + | Event::AgentCliTimedOut { node_id, .. } + | Event::AgentAcpCompleted { node_id, .. } + | Event::AgentAcpCancelled { node_id, .. } + | Event::AgentAcpTimedOut { node_id, .. } => node_stored_fields(Some(node_id.clone())), + Event::AgentAcpStarted { node_id, visit, .. } => { + let node_id_str = node_id.clone(); + let node_label = default_node_label(Some(&node_id_str), None); + StoredEventFields { + node_id: Some(node_id_str.clone()), + node_label, + stage_id: Some(StageId::new(node_id_str, *visit)), + ..StoredEventFields::default() + } + } Event::AgentSessionStarted { session_id, parent_session_id, diff --git a/lib/crates/fabro-workflow/src/handler/llm/acp.rs b/lib/crates/fabro-workflow/src/handler/llm/acp.rs index 48e2b09bc..72935b864 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/acp.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/acp.rs @@ -9,6 +9,7 @@ use fabro_agent::{Sandbox, StaticEnvProvider, ToolEnvProvider}; use fabro_auth::{CliAgentKind, CredentialResolver, CredentialUsage, ResolvedCredential}; use fabro_graphviz::graph::Node; use fabro_model::Provider; +use fabro_util::time::elapsed_ms; use tokio_util::sync::CancellationToken; use super::super::agent::{CodergenBackend, CodergenResult}; @@ -16,7 +17,7 @@ use super::cli::{AgentCli, process_env_var}; use super::{changed_files, node_runtime}; use crate::context::Context; use crate::error::Error; -use crate::event::{Emitter, RunNoticeCode, RunNoticeLevel, StageScope}; +use crate::event::{Emitter, Event, RunNoticeCode, RunNoticeLevel, StageScope}; pub struct AgentAcpBackend { model: String, @@ -71,11 +72,12 @@ impl AgentAcpBackend { node: &Node, prompt: String, emitter: &Arc, + stage_scope: &StageScope, sandbox: &Arc, cancel_token: CancellationToken, ) -> Result { let files_before = changed_files::detect_changed_files(sandbox).await; - let _model = node.model().unwrap_or(&self.model); + let model = node.model().unwrap_or(&self.model); let provider = node .provider() .and_then(|value| value.parse::().ok()) @@ -98,7 +100,21 @@ impl AgentAcpBackend { Arc::new(move || emitter.touch()) as Arc }; - let result = fabro_acp::run_acp_turn(AcpRunRequest { + let command_display = command.to_string(); + emitter.emit_scoped( + &Event::AgentAcpStarted { + node_id: node.id.clone(), + visit: stage_scope.visit, + mode: "acp".to_string(), + provider: provider.to_string(), + model: model.to_string(), + command: command_display, + }, + stage_scope, + ); + + let launch_start = std::time::Instant::now(); + let result = match fabro_acp::run_acp_turn(AcpRunRequest { command, prompt, cwd: sandbox.working_directory().to_string(), @@ -109,7 +125,62 @@ impl AgentAcpBackend { on_activity: Some(on_activity), }) .await - .map_err(acp_error_to_workflow)?; + { + Ok(result) => { + emitter.emit_scoped( + &Event::AgentAcpCompleted { + node_id: node.id.clone(), + stdout: result.text.clone(), + stderr: result.stderr.clone(), + stop_reason: stop_reason_to_string(&result.stop_reason), + duration_ms: result.duration_ms, + }, + stage_scope, + ); + result + } + Err(AcpError::Cancelled) => { + emitter.emit_scoped( + &Event::AgentAcpCancelled { + node_id: node.id.clone(), + stdout: String::new(), + stderr: String::new(), + duration_ms: elapsed_ms(launch_start), + }, + stage_scope, + ); + return Err(Error::Cancelled); + } + Err(AcpError::TimedOut { stderr }) => { + emitter.emit_scoped( + &Event::AgentAcpTimedOut { + node_id: node.id.clone(), + stdout: String::new(), + stderr: stderr.clone(), + duration_ms: elapsed_ms(launch_start), + }, + stage_scope, + ); + return Err(acp_error_to_workflow(AcpError::TimedOut { stderr })); + } + Err(AcpError::StopReason { stop_reason, text }) => { + emitter.emit_scoped( + &Event::AgentAcpCompleted { + node_id: node.id.clone(), + stdout: text.clone(), + stderr: String::new(), + stop_reason: stop_reason.clone(), + duration_ms: elapsed_ms(launch_start), + }, + stage_scope, + ); + return Err(acp_error_to_workflow(AcpError::StopReason { + stop_reason, + text, + })); + } + Err(error) => return Err(acp_error_to_workflow(error)), + }; let (files_touched, last_file_touched) = changed_files::files_touched_since(sandbox, &files_before).await; @@ -201,15 +272,23 @@ impl CodergenBackend for AgentAcpBackend { &self, node: &Node, prompt: &str, - _context: &Context, + context: &Context, _thread_id: Option<&str>, emitter: &Arc, sandbox: &Arc, _tool_hooks: Option>, cancel_token: CancellationToken, ) -> Result { - self.run_turn(node, prompt.to_string(), emitter, sandbox, cancel_token) - .await + let stage_scope = StageScope::for_handler(context, &node.id); + self.run_turn( + node, + prompt.to_string(), + emitter, + &stage_scope, + sandbox, + cancel_token, + ) + .await } async fn one_shot( @@ -218,7 +297,7 @@ impl CodergenBackend for AgentAcpBackend { prompt: &str, system_prompt: Option<&str>, emitter: &Arc, - _stage_scope: &StageScope, + stage_scope: &StageScope, sandbox: &Arc, cancel_token: CancellationToken, ) -> Result { @@ -226,11 +305,18 @@ impl CodergenBackend for AgentAcpBackend { Some(system_prompt) => format!("System:\n{system_prompt}\n\nUser:\n{prompt}"), None => prompt.to_string(), }; - self.run_turn(node, prompt, emitter, sandbox, cancel_token) + self.run_turn(node, prompt, emitter, stage_scope, sandbox, cancel_token) .await } } +fn stop_reason_to_string(stop_reason: &(impl serde::Serialize + std::fmt::Debug)) -> String { + serde_json::to_value(stop_reason) + .ok() + .and_then(|value| value.as_str().map(str::to_string)) + .unwrap_or_else(|| format!("{stop_reason:?}")) +} + fn acp_error_to_workflow(error: AcpError) -> Error { match error { AcpError::Cancelled => Error::Cancelled, diff --git a/lib/crates/fabro-workflow/src/operations/fork.rs b/lib/crates/fabro-workflow/src/operations/fork.rs index ebe8eaed5..2341ddd5b 100644 --- a/lib/crates/fabro-workflow/src/operations/fork.rs +++ b/lib/crates/fabro-workflow/src/operations/fork.rs @@ -231,6 +231,9 @@ fn replay_event_for_fork_projection(body: &EventBody) -> bool { | EventBody::AgentCliStarted(_) | EventBody::AgentCliCancelled(_) | EventBody::AgentCliTimedOut(_) + | EventBody::AgentAcpStarted(_) + | EventBody::AgentAcpCancelled(_) + | EventBody::AgentAcpTimedOut(_) | EventBody::CommandStarted(_) | EventBody::CommandCompleted(_) | EventBody::ParallelCompleted(_) @@ -316,6 +319,41 @@ mod tests { )); } + #[test] + fn fork_replay_preserves_agent_acp_projection_events() { + assert!(replay_event_for_fork_projection( + &EventBody::AgentAcpStarted(fabro_types::run_event::AgentAcpStartedProps { + visit: 1, + mode: "acp".to_string(), + provider: "openai".to_string(), + model: "fake-acp".to_string(), + command: "python fake_agent.py".to_string(), + }) + )); + assert!(replay_event_for_fork_projection( + &EventBody::AgentAcpCancelled(fabro_types::run_event::AgentAcpCancelledProps { + stdout: "partial".to_string(), + stderr: "cancelled".to_string(), + duration_ms: 7, + }) + )); + assert!(replay_event_for_fork_projection( + &EventBody::AgentAcpTimedOut(fabro_types::run_event::AgentAcpTimedOutProps { + stdout: "partial".to_string(), + stderr: "timeout".to_string(), + duration_ms: 99, + }) + )); + assert!(!replay_event_for_fork_projection( + &EventBody::AgentAcpCompleted(fabro_types::run_event::AgentAcpCompletedProps { + stdout: "done".to_string(), + stderr: String::new(), + stop_reason: "end_turn".to_string(), + duration_ms: 42, + }) + )); + } + #[tokio::test] async fn fork_persists_historical_node_projection_through_target_checkpoint() { let store = test_store();