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

feat: project ACP backend events

Browse source
This commit is contained in:
Bryan Helmkamp 2026-05-11 11:25:44 -04:00
parent 1c3edfe5f7
commit 212cffeb1d
No known key found for this signature in database
12 changed files with 775 additions and 51 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -189,30 +189,30 @@ impl<T: serde::Serialize> ListResponse<T> {
/// Snapshot of a managed run.
struct ManagedRun {
dot_source: String,
status: RunStatus,
error: Option<String>,
created_at: chrono::DateTime<chrono::Utc>,
enqueued_at: Instant,
dot_source: String,
status: RunStatus,
error: Option<String>,
created_at: chrono::DateTime<chrono::Utc>,
enqueued_at: Instant,
// Populated when running:
answer_transport: Option<RunAnswerTransport>,
answer_transport: Option<RunAnswerTransport>,
accepted_questions: HashSet<String>,
/// 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<StageId, String>,
/// Stage IDs of currently running CLI-mode agent sessions, observed
/// from `agent.cli.started/completed` plus `stage.completed`/
active_api_stages: HashMap<StageId, String>,
/// 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<StageId>,
event_tx: Option<broadcast::Sender<RunEvent>>,
checkpoint: Option<Checkpoint>,
cancel_tx: Option<oneshot::Sender<()>>,
cancel_token: Option<CancellationToken>,
worker_pid: Option<u32>,
worker_pgid: Option<u32>,
run_dir: Option<std::path::PathBuf>,
execution_mode: RunExecutionMode,
active_non_steerable_agent_stages: HashSet<StageId>,
event_tx: Option<broadcast::Sender<RunEvent>>,
checkpoint: Option<Checkpoint>,
cancel_tx: Option<oneshot::Sender<()>>,
cancel_token: Option<CancellationToken>,
worker_pid: Option<u32>,
worker_pgid: Option<u32>,
run_dir: Option<std::path::PathBuf>,
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);
}
}
_ => {}

10
lib/crates/fabro-server/src/server/handler/steer.rs
View file

@ -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();
}

144
lib/crates/fabro-server/src/server/tests.rs
View file

@ -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<workflow_event::Event> = 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();

186
lib/crates/fabro-store/src/run_state.rs
View file

@ -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();

31
lib/crates/fabro-types/src/run_event/misc.rs
View file

@ -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,

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

@ -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"

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

@ -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 {

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

@ -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,

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

@ -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",

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

@ -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,

104
lib/crates/fabro-workflow/src/handler/llm/acp.rs
View file

@ -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<Emitter>,
stage_scope: &StageScope,
sandbox: &Arc<dyn Sandbox>,
cancel_token: CancellationToken,
) -> Result<CodergenResult, Error> {
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::<Provider>().ok())
@ -98,7 +100,21 @@ impl AgentAcpBackend {
Arc::new(move || emitter.touch()) as Arc<dyn Fn() + Send + Sync>
};
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<Emitter>,
sandbox: &Arc<dyn Sandbox>,
_tool_hooks: Option<Arc<dyn fabro_agent::ToolHookCallback>>,
cancel_token: CancellationToken,
) -> Result<CodergenResult, Error> {
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<Emitter>,
_stage_scope: &StageScope,
stage_scope: &StageScope,
sandbox: &Arc<dyn Sandbox>,
cancel_token: CancellationToken,
) -> Result<CodergenResult, Error> {
@ -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,

38
lib/crates/fabro-workflow/src/operations/fork.rs
View file

@ -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();