diff --git a/lib/crates/fabro-cli/src/commands/run/rewind.rs b/lib/crates/fabro-cli/src/commands/run/rewind.rs index 928d6e36e..b116369b2 100644 --- a/lib/crates/fabro-cli/src/commands/run/rewind.rs +++ b/lib/crates/fabro-cli/src/commands/run/rewind.rs @@ -7,7 +7,7 @@ use fabro_util::terminal::Styles; use fabro_workflow::event::{WorkflowRunEvent, append_workflow_event}; use fabro_workflow::git::MetadataStore; use fabro_workflow::operations::{ - RewindInput, RewindTarget, RunTimeline, build_timeline_or_rebuild, + RewindInput, RewindTarget, RunTimeline, TimelineEntry, build_timeline_or_rebuild, find_run_id_by_prefix_or_store, rewind, }; use fabro_workflow::records::{RunRecord, RunRecordExt, StartRecord, StartRecordExt}; @@ -62,12 +62,14 @@ pub(crate) async fn run(args: &RewindArgs, styles: &Styles, globals: &GlobalArgs &store, &RewindInput { run_id, - target, + target: target.clone(), push: !args.no_push, }, )?; if let Some(run_info) = run_info.as_ref() { - reset_rewound_run_state(&store, durable_store.as_ref(), &run_id, &run_info.path).await?; + let entry = timeline.resolve(&target)?; + reset_rewound_run_state(&store, durable_store.as_ref(), &run_id, &run_info.path, entry) + .await?; } let run_id_string = run_id.to_string(); @@ -105,6 +107,7 @@ async fn reset_rewound_run_state( durable_store: &dyn fabro_store::Store, run_id: &fabro_types::RunId, run_dir: &std::path::Path, + entry: &TimelineEntry, ) -> Result<()> { let existing_run_store = durable_store .open_run_reader(run_id) @@ -131,18 +134,28 @@ async fn reset_rewound_run_state( .context("failed to restore run record after rewind: missing run metadata")?; let checkpoint = MetadataStore::read_checkpoint(git_store.repo_dir(), &run_id.to_string())? .context("rewound metadata branch is missing checkpoint.json")?; + let previous_status = if let Some(run_store) = existing_run_store.as_ref() { + run_store + .get_status() + .await + .ok() + .flatten() + .map(|status| status.status.to_string()) + } else { + None + }; let _ = std::fs::remove_file(run_dir.join("detached_failure.json")); - durable_store - .delete_run(run_id) - .await - .map_err(|err| anyhow::anyhow!("failed to reset durable store run: {err}"))?; - let run_dir_string = run_dir.to_string_lossy().to_string(); let run_store = durable_store - .create_run(run_id, run_record.created_at, Some(&run_dir_string)) + .open_run(run_id) .await - .map_err(|err| anyhow::anyhow!("failed to recreate durable store run: {err}"))?; + .map_err(|err| anyhow::anyhow!("failed to open durable store run for rewind reset: {err}"))? + .context("failed to reset durable store run after rewind: missing run store")?; + run_store + .reset_for_rewind() + .await + .map_err(|err| anyhow::anyhow!("failed to clear rewound run state: {err}"))?; run_store .put_run(&run_record) .await @@ -159,6 +172,26 @@ async fn reset_rewound_run_state( .await .map_err(|err| anyhow::anyhow!("failed to restore graph after rewind: {err}"))?; } + append_workflow_event( + run_store.as_ref(), + run_id, + &WorkflowRunEvent::RunRewound { + target_checkpoint_ordinal: entry.ordinal, + target_node_id: entry.node_name.clone(), + target_visit: entry.visit, + previous_status, + run_commit_sha: entry.run_commit_sha.clone(), + }, + ) + .await + .map_err(|err| anyhow::anyhow!("failed to append run rewound event: {err}"))?; + append_workflow_event( + run_store.as_ref(), + run_id, + &restored_checkpoint_event(&checkpoint), + ) + .await + .map_err(|err| anyhow::anyhow!("failed to append restored checkpoint event: {err}"))?; run_store .put_status(&fabro_types::RunStatusRecord::new( RunStatus::Submitted, @@ -180,6 +213,36 @@ async fn reset_rewound_run_state( Ok(()) } +fn restored_checkpoint_event(checkpoint: &fabro_types::Checkpoint) -> WorkflowRunEvent { + let current_status = checkpoint + .node_outcomes + .get(&checkpoint.current_node) + .map_or_else(|| "success".to_string(), |outcome| outcome.status.to_string()); + WorkflowRunEvent::CheckpointCompleted { + node_id: checkpoint.current_node.clone(), + status: current_status, + current_node: checkpoint.current_node.clone(), + completed_nodes: checkpoint.completed_nodes.clone(), + node_retries: checkpoint.node_retries.clone().into_iter().collect(), + context_values: checkpoint.context_values.clone().into_iter().collect(), + node_outcomes: checkpoint.node_outcomes.clone().into_iter().collect(), + next_node_id: checkpoint.next_node_id.clone(), + git_commit_sha: checkpoint.git_commit_sha.clone(), + loop_failure_signatures: checkpoint + .loop_failure_signatures + .iter() + .map(|(sig, count)| (sig.to_string(), *count)) + .collect(), + restart_failure_signatures: checkpoint + .restart_failure_signatures + .iter() + .map(|(sig, count)| (sig.to_string(), *count)) + .collect(), + node_visits: checkpoint.node_visits.clone().into_iter().collect(), + diff: None, + } +} + pub(crate) fn print_timeline(timeline: &RunTimeline, styles: &Styles) { if timeline.entries.is_empty() { eprintln!("No checkpoints found."); diff --git a/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs b/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs index 32dc11844..245bf72a3 100644 --- a/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs +++ b/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs @@ -631,6 +631,7 @@ mod tests { fn round_trip_agent_tool_call() { let event = WorkflowRunEvent::Agent { stage: "code".into(), + visit: 1, event: AgentEvent::ToolCallStarted { tool_name: "read_file".into(), tool_call_id: "tc1".into(), diff --git a/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs b/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs index a38b2da72..a65801357 100644 --- a/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs +++ b/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs @@ -482,6 +482,7 @@ mod tests { fn agent_event(stage: &str, event: AgentEvent) -> WorkflowRunEvent { WorkflowRunEvent::Agent { stage: stage.into(), + visit: 1, event, session_id: None, parent_session_id: None, diff --git a/lib/crates/fabro-cli/tests/it/cmd/rewind.rs b/lib/crates/fabro-cli/tests/it/cmd/rewind.rs index 0e79c3a3f..3543d0e0e 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/rewind.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/rewind.rs @@ -4,7 +4,7 @@ use fabro_test::{fabro_snapshot, run_and_format, test_context}; use super::support::{ git_filters, git_stdout, output_stderr as support_stderr, run_branch_commits_since_base, - setup_git_backed_changed_run, + run_events, run_snapshot, setup_git_backed_changed_run, }; #[test] @@ -122,3 +122,68 @@ fn rewind_target_updates_metadata_and_resume_hint() { "rewound timeline should drop @2: {list}" ); } + +#[test] +fn rewind_preserves_event_history_and_clears_terminal_snapshot_state() { + let context = test_context!(); + let setup = setup_git_backed_changed_run(&context); + let before_events = run_events(&setup.run.run_dir); + assert!( + before_events.iter().any(|event| event.payload.as_value()["event"] == "run.completed"), + "setup run should be completed before rewind" + ); + + let mut cmd = context.command(); + cmd.current_dir(&setup.repo_dir); + cmd.args(["rewind", &setup.run.run_id, "@1", "--no-push"]); + let output = cmd.output().expect("rewind should execute"); + assert!( + output.status.success(), + "rewind should succeed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + support_stderr(&output), + ); + + let after_events = run_events(&setup.run.run_dir); + assert_eq!( + after_events.len(), + before_events.len() + 3, + "rewind should append run.rewound, checkpoint.completed, and run.submitted" + ); + assert_eq!( + after_events[..before_events.len()] + .iter() + .map(|event| event.payload.as_value()["event"].as_str().unwrap()) + .collect::>(), + before_events + .iter() + .map(|event| event.payload.as_value()["event"].as_str().unwrap()) + .collect::>(), + "rewind should preserve the prior event prefix" + ); + assert_eq!( + after_events[before_events.len()].payload.as_value()["event"], + "run.rewound" + ); + assert_eq!( + after_events[before_events.len() + 1].payload.as_value()["event"], + "checkpoint.completed" + ); + assert_eq!( + after_events[before_events.len() + 2].payload.as_value()["event"], + "run.submitted" + ); + + let snapshot = run_snapshot(&setup.run.run_dir); + assert_eq!( + snapshot.status.as_ref().map(|status| &status.status), + Some(&fabro_types::RunStatus::Submitted) + ); + assert!(snapshot.conclusion.is_none(), "rewind should clear conclusion"); + assert!(snapshot.final_patch.is_none(), "rewind should clear final patch"); + assert!(snapshot.pull_request.is_none(), "rewind should clear pull request"); + assert!( + snapshot.nodes.is_empty(), + "rewind should clear node snapshots that belonged to the prior execution" + ); +} diff --git a/lib/crates/fabro-cli/tests/it/cmd/support.rs b/lib/crates/fabro-cli/tests/it/cmd/support.rs index e40fb330b..e0f6c2761 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/support.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/support.rs @@ -4,7 +4,7 @@ use std::process::Output; use std::sync::Arc; use std::time::{Duration, Instant}; -use fabro_store::{RunSnapshot, RunStore, SlateStore, Store}; +use fabro_store::{EventEnvelope, RunSnapshot, RunStore, SlateStore, Store}; use fabro_test::TestContext; use fabro_types::RunId; use object_store::local::LocalFileSystem; @@ -494,6 +494,12 @@ pub(crate) fn run_snapshot(run_dir: &Path) -> RunSnapshot { .expect("run store snapshot should exist") } +pub(crate) fn run_events(run_dir: &Path) -> Vec { + run_store(run_dir) + .and_then(|store| block_on(store.list_events()).ok()) + .expect("run store events should exist") +} + pub(crate) fn git_stdout(repo_dir: &Path, args: &[&str]) -> String { stdout(&git_success(repo_dir, args)) } diff --git a/lib/crates/fabro-store/src/lib.rs b/lib/crates/fabro-store/src/lib.rs index 8d48a6a39..07c3c3fc5 100644 --- a/lib/crates/fabro-store/src/lib.rs +++ b/lib/crates/fabro-store/src/lib.rs @@ -122,6 +122,8 @@ pub trait RunStore: Send + Sync { async fn put_pull_request(&self, record: &PullRequestRecord) -> Result<()>; async fn get_pull_request(&self) -> Result>; + async fn reset_for_rewind(&self) -> Result<()>; + async fn append_event(&self, payload: &EventPayload) -> Result; async fn list_events(&self) -> Result>; async fn list_events_from(&self, seq: u32) -> Result>; diff --git a/lib/crates/fabro-store/src/memory.rs b/lib/crates/fabro-store/src/memory.rs index be1065b55..febc85ffe 100644 --- a/lib/crates/fabro-store/src/memory.rs +++ b/lib/crates/fabro-store/src/memory.rs @@ -535,6 +535,18 @@ impl RunStore for InMemoryRunStore { self.get_json(keys::pull_request()).await } + async fn reset_for_rewind(&self) -> Result<()> { + let mut data = self.data.lock().await; + data.retain(|key, _| { + key == keys::init() + || key == keys::run() + || key == keys::start() + || key == keys::graph() + || key.starts_with(keys::EVENTS_PREFIX) + }); + Ok(()) + } + async fn append_event(&self, payload: &EventPayload) -> Result { payload.validate(&self.run_id)?; diff --git a/lib/crates/fabro-store/src/slate/run_store.rs b/lib/crates/fabro-store/src/slate/run_store.rs index 1b75cad95..5d704da64 100644 --- a/lib/crates/fabro-store/src/slate/run_store.rs +++ b/lib/crates/fabro-store/src/slate/run_store.rs @@ -468,6 +468,32 @@ impl RunStore for SlateRunStore { self.inner.db.get_json(keys::pull_request()).await } + async fn reset_for_rewind(&self) -> Result<()> { + let db = self.inner.db.writer()?; + for key in [ + keys::status(), + keys::checkpoint(), + keys::conclusion(), + keys::retro(), + keys::sandbox(), + keys::final_patch(), + keys::pull_request(), + keys::retro_prompt(), + keys::retro_response(), + ] { + db.delete(key).await?; + } + for prefix in [ + b"nodes/".as_slice(), + keys::CHECKPOINTS_PREFIX.as_bytes(), + keys::ARTIFACT_VALUES_PREFIX.as_bytes(), + keys::ARTIFACT_NODES_PREFIX.as_bytes(), + ] { + delete_prefix(db, prefix).await?; + } + Ok(()) + } + async fn append_event(&self, payload: &EventPayload) -> Result { payload.validate(&self.inner.run_id)?; let seq = self.inner.event_seq.fetch_add(1, Ordering::SeqCst); @@ -762,6 +788,18 @@ async fn put_bytes(db: &slatedb::Db, key: &str, value: &[u8]) -> Result<()> { Ok(()) } +async fn delete_prefix(db: &slatedb::Db, prefix: &[u8]) -> Result<()> { + let mut iter = db.scan_prefix(prefix).await?; + let mut keys = Vec::new(); + while let Some(entry) = iter.next().await? { + keys.push(key_to_string(&entry.key)?); + } + for key in keys { + db.delete(key).await?; + } + Ok(()) +} + async fn get_bytes(db: &slatedb::Db, key: &str) -> Result> { Ok(db.get(key).await?) } diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs index f2ffd2880..f36fa973a 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/lib/crates/fabro-workflow/src/event.rs @@ -5,7 +5,7 @@ use std::sync::atomic::{AtomicI64, Ordering}; use anyhow::{Context, Result}; use chrono::{SecondsFormat, Utc}; -use fabro_store::{EventPayload, RunStore}; +use fabro_store::{EventPayload, NodeVisitRef, RunStore}; use fabro_types::RunId; use serde::{Deserialize, Serialize}; use serde_json::{Map, Value}; @@ -95,17 +95,18 @@ pub enum WorkflowRunEvent { #[serde(default, skip_serializing_if = "Option::is_none")] reason: Option, }, - RunPaused { - #[serde(default, skip_serializing_if = "Option::is_none")] - reason: Option, - }, RunRemoving { #[serde(default, skip_serializing_if = "Option::is_none")] reason: Option, }, - RunDead { + RunRewound { + target_checkpoint_ordinal: usize, + target_node_id: String, + target_visit: usize, #[serde(default, skip_serializing_if = "Option::is_none")] - reason: Option, + previous_status: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + run_commit_sha: Option, }, WorkflowRunCompleted { duration_ms: u64, @@ -307,6 +308,7 @@ pub enum WorkflowRunEvent { }, Prompt { stage: String, + visit: u32, text: String, #[serde(default, skip_serializing_if = "Option::is_none")] mode: Option, @@ -326,6 +328,7 @@ pub enum WorkflowRunEvent { /// Forwarded from an agent session, tagged with the workflow stage. Agent { stage: String, + visit: u32, event: AgentEvent, #[serde(default, skip_serializing_if = "Option::is_none")] session_id: Option, @@ -439,6 +442,7 @@ pub enum WorkflowRunEvent { }, AgentCliStarted { node_id: String, + visit: u32, mode: String, provider: String, model: String, @@ -539,14 +543,24 @@ impl WorkflowRunEvent { Self::RunRunning { reason } => { info!(?reason, "Run running"); } - Self::RunPaused { reason } => { - info!(?reason, "Run paused"); - } Self::RunRemoving { reason } => { info!(?reason, "Run removing"); } - Self::RunDead { reason } => { - warn!(?reason, "Run dead"); + Self::RunRewound { + target_checkpoint_ordinal, + target_node_id, + target_visit, + previous_status, + run_commit_sha, + } => { + info!( + target_checkpoint_ordinal, + target_node_id, + target_visit, + previous_status = previous_status.as_deref().unwrap_or(""), + run_commit_sha = run_commit_sha.as_deref().unwrap_or(""), + "Run rewound" + ); } Self::WorkflowRunCompleted { duration_ms, @@ -787,6 +801,7 @@ impl WorkflowRunEvent { mode, provider, model, + .. } => { debug!( stage, @@ -1065,9 +1080,8 @@ pub fn event_name(event: &WorkflowRunEvent) -> &'static str { WorkflowRunEvent::RunSubmitted { .. } => "run.submitted", WorkflowRunEvent::RunStarting { .. } => "run.starting", WorkflowRunEvent::RunRunning { .. } => "run.running", - WorkflowRunEvent::RunPaused { .. } => "run.paused", WorkflowRunEvent::RunRemoving { .. } => "run.removing", - WorkflowRunEvent::RunDead { .. } => "run.dead", + WorkflowRunEvent::RunRewound { .. } => "run.rewound", WorkflowRunEvent::WorkflowRunCompleted { .. } => "run.completed", WorkflowRunEvent::WorkflowRunFailed { .. } => "run.failed", WorkflowRunEvent::RunNotice { .. } => "run.notice", @@ -1314,12 +1328,16 @@ fn extract_envelope_fields(event: &WorkflowRunEvent) -> EnvelopeFields { let mut fields = tagged_variant_fields(event); let node_id = remove_string(&mut fields, "stage"); let node_label = default_node_label(node_id.as_ref(), None); + let visit = fields.remove("visit"); fields.remove("session_id"); fields.remove("parent_session_id"); - let properties = fields.remove("event").map_or_else( + let mut properties = fields.remove("event").map_or_else( || Value::Object(Map::new()), |value| Value::Object(tagged_variant_fields_from_value(value)), ); + if let (Some(visit), Value::Object(map)) = (visit, &mut properties) { + map.insert("visit".to_string(), visit); + } EnvelopeFields { session_id: session_id.clone(), parent_session_id: parent_session_id.clone(), @@ -1552,6 +1570,15 @@ impl StoreProgressLogger { if let Err(err) = run_store.append_event(&payload).await { tracing::warn!(error = %err, "Failed to append event to run store"); } + if let Err(err) = + project_provider_used_from_event_payload(run_store.as_ref(), &payload) + .await + { + tracing::warn!( + error = %err, + "Failed to project provider metadata from event" + ); + } } StoreProgressCommand::Flush(tx) => { let _ = tx.send(()); @@ -1597,6 +1624,80 @@ impl StoreProgressLogger { } } +async fn project_provider_used_from_event_payload( + run_store: &dyn RunStore, + payload: &EventPayload, +) -> Result<()> { + let value = payload.as_value(); + let Some(event_name) = value.get("event").and_then(Value::as_str) else { + return Ok(()); + }; + let Some(node_id) = value.get("node_id").and_then(Value::as_str) else { + return Ok(()); + }; + let Some(properties) = value.get("properties").and_then(Value::as_object) else { + return Ok(()); + }; + let Some(visit) = properties + .get("visit") + .and_then(Value::as_u64) + .and_then(|visit| u32::try_from(visit).ok()) + else { + return Ok(()); + }; + + let provider_used = match event_name { + "stage.prompt" => { + let mut provider_used = Map::new(); + if let Some(mode) = properties.get("mode").and_then(Value::as_str) { + provider_used.insert("mode".to_string(), Value::String(mode.to_string())); + } + if let Some(provider) = properties.get("provider").and_then(Value::as_str) { + provider_used.insert("provider".to_string(), Value::String(provider.to_string())); + } + if let Some(model) = properties.get("model").and_then(Value::as_str) { + provider_used.insert("model".to_string(), Value::String(model.to_string())); + } + (!provider_used.is_empty()).then_some(Value::Object(provider_used)) + } + "agent.session.started" => { + let mut provider_used = Map::new(); + provider_used.insert("mode".to_string(), Value::String("agent".to_string())); + if let Some(provider) = properties.get("provider").and_then(Value::as_str) { + provider_used.insert("provider".to_string(), Value::String(provider.to_string())); + } + if let Some(model) = properties.get("model").and_then(Value::as_str) { + provider_used.insert("model".to_string(), Value::String(model.to_string())); + } + Some(Value::Object(provider_used)) + } + "agent.cli.started" => { + let mut provider_used = Map::new(); + provider_used.insert("mode".to_string(), Value::String("cli".to_string())); + if let Some(provider) = properties.get("provider").and_then(Value::as_str) { + provider_used.insert("provider".to_string(), Value::String(provider.to_string())); + } + if let Some(model) = properties.get("model").and_then(Value::as_str) { + provider_used.insert("model".to_string(), Value::String(model.to_string())); + } + if let Some(command) = properties.get("command").and_then(Value::as_str) { + provider_used.insert("command".to_string(), Value::String(command.to_string())); + } + Some(Value::Object(provider_used)) + } + _ => None, + }; + + let Some(provider_used) = provider_used else { + return Ok(()); + }; + + run_store + .put_node_provider_used(&NodeVisitRef { node_id, visit }, &provider_used) + .await + .map_err(anyhow::Error::from) +} + /// Current time as epoch milliseconds. fn epoch_millis() -> i64 { let millis = std::time::SystemTime::now() @@ -1856,6 +1957,7 @@ mod tests { &fixtures::RUN_4, &WorkflowRunEvent::Agent { stage: "code".to_string(), + visit: 2, event: AgentEvent::ToolCallStarted { tool_name: "read_file".to_string(), tool_call_id: "call_1".to_string(), @@ -1873,6 +1975,7 @@ mod tests { assert_eq!(envelope.parent_session_id.as_deref(), Some("ses_parent")); assert_eq!(envelope.properties["tool_name"], "read_file"); assert_eq!(envelope.properties["tool_call_id"], "call_1"); + assert_eq!(envelope.properties["visit"], 2); } #[test] @@ -1975,6 +2078,7 @@ mod tests { assert_eq!( event_name(&WorkflowRunEvent::Agent { stage: "code".to_string(), + visit: 1, event: AgentEvent::SubAgentSpawned { agent_id: "a1".to_string(), depth: 1, diff --git a/lib/crates/fabro-workflow/src/handler/agent.rs b/lib/crates/fabro-workflow/src/handler/agent.rs index bcb7b396b..43242ded9 100644 --- a/lib/crates/fabro-workflow/src/handler/agent.rs +++ b/lib/crates/fabro-workflow/src/handler/agent.rs @@ -200,33 +200,6 @@ pub(crate) fn truncate(s: &str, max_chars: usize) -> &str { } } -pub(crate) async fn sync_provider_used_to_store( - stage_dir: &Path, - node_ref: &NodeVisitRef<'_>, - services: &EngineServices, -) -> Result<(), FabroError> { - let Some(ref store) = services.run_store else { - return Ok(()); - }; - - let path = stage_dir.join("provider_used.json"); - let json = match fs::read_to_string(&path).await { - Ok(json) => json, - Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(()), - Err(err) => { - return Err(FabroError::handler(format!( - "Failed to read provider_used.json: {err}" - ))); - } - }; - let value: serde_json::Value = serde_json::from_str(&json) - .map_err(|err| FabroError::handler(format!("Failed to parse provider_used.json: {err}")))?; - store - .put_node_provider_used(node_ref, &value) - .await - .map_err(|err| FabroError::handler(err.to_string())) -} - /// Shared simulate implementation for LLM-backed handlers (agent & prompt). /// Produces a simulated outcome with standard context updates. pub(crate) fn simulate_llm_handler(node: &Node) -> Outcome { @@ -329,10 +302,7 @@ impl Handler for AgentHandler { ) .await; match result { - Ok(CodergenResult::Full(outcome)) => { - sync_provider_used_to_store(&stage_dir, &node_ref, services).await?; - return Ok(outcome); - } + Ok(CodergenResult::Full(outcome)) => return Ok(outcome), Ok(CodergenResult::Text { text, usage, @@ -364,8 +334,6 @@ impl Handler for AgentHandler { } else { fs::write(stage_dir.join("response.md"), &response_text).await?; } - sync_provider_used_to_store(&stage_dir, &node_ref, services).await?; - // 7. Build and write status let mut outcome = Outcome::success(); outcome.notes = Some(format!("Stage completed: {}", node.id)); @@ -433,7 +401,11 @@ mod tests { EngineServices::test_default() } - async fn make_services_with_run_store() -> (EngineServices, Arc) { + async fn make_services_with_run_store() -> ( + EngineServices, + Arc, + crate::event::StoreProgressLogger, + ) { let store = InMemoryStore::default(); let run_store = store .create_run(&fixtures::RUN_1, chrono::Utc::now(), None) @@ -443,7 +415,9 @@ mod tests { run_store: Some(Arc::clone(&run_store)), ..EngineServices::test_default() }; - (services, run_store) + let logger = crate::event::StoreProgressLogger::new(Arc::clone(&run_store)); + logger.register(services.emitter.as_ref()); + (services, run_store, logger) } fn test_context() -> Context { @@ -706,27 +680,32 @@ mod tests { } #[tokio::test] - async fn codergen_handler_persists_provider_used_in_run_store() { - struct ProviderUsedBackend; + async fn codergen_handler_projects_provider_used_from_agent_session_events() { + struct ProviderEventBackend; #[async_trait] - impl CodergenBackend for ProviderUsedBackend { + impl CodergenBackend for ProviderEventBackend { async fn run( &self, - _node: &Node, + node: &Node, _prompt: &str, - _context: &Context, + context: &Context, _thread_id: Option<&str>, - _emitter: &Arc, - stage_dir: &Path, + emitter: &Arc, + _stage_dir: &Path, _sandbox: &Arc, _tool_hooks: Option>, ) -> Result { - std::fs::write( - stage_dir.join("provider_used.json"), - r#"{"mode":"agent","provider":"openai","model":"gpt-5.4"}"#, - ) - .unwrap(); + emitter.emit(&crate::event::WorkflowRunEvent::Agent { + stage: node.id.clone(), + visit: crate::run_dir::visit_from_context(context) as u32, + event: fabro_agent::AgentEvent::SessionStarted { + provider: Some("openai".to_string()), + model: Some("gpt-5.4".to_string()), + }, + session_id: Some("session_123".to_string()), + parent_session_id: None, + }); Ok(CodergenResult::Text { text: "done".to_string(), usage: None, @@ -736,17 +715,18 @@ mod tests { } } - let handler = AgentHandler::new(Some(Box::new(ProviderUsedBackend))); + let handler = AgentHandler::new(Some(Box::new(ProviderEventBackend))); let node = Node::new("step"); let context = test_context(); let graph = Graph::new("test"); let tmp = TempDir::new().unwrap(); - let (services, run_store) = make_services_with_run_store().await; + let (services, run_store, logger) = make_services_with_run_store().await; handler .execute(&node, &context, &graph, tmp.path(), &services) .await .unwrap(); + logger.flush().await; let snapshot = run_store .get_node(&NodeVisitRef { diff --git a/lib/crates/fabro-workflow/src/handler/llm/api.rs b/lib/crates/fabro-workflow/src/handler/llm/api.rs index 066d06164..16f41e63d 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/api.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/api.rs @@ -23,6 +23,7 @@ use crate::error::FabroError; use crate::event::{EventEmitter, WorkflowRunEvent}; use crate::outcome::StageUsage; use crate::outcome::compute_stage_cost; +use crate::run_dir::visit_from_context; use fabro_graphviz::graph::Node; fn build_profile(model: &str, provider: Provider) -> Box { @@ -38,6 +39,10 @@ fn build_profile(model: &str, provider: Provider) -> Box { } } +fn current_visit(context: &Context) -> u32 { + u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX) +} + /// Shared state for tracking file modifications from agent tool calls. struct FileTracking { /// Maps tool_call_id → file_path for in-flight write/edit calls. @@ -83,6 +88,7 @@ fn track_file_event(event: &AgentEvent, state: &mut FileTracking) { fn spawn_event_forwarder( session: &Session, node_id: String, + visit: u32, emitter: Arc, file_tracking: Arc>, ) { @@ -101,6 +107,7 @@ fn spawn_event_forwarder( { emitter.emit(&WorkflowRunEvent::Agent { stage: node_id.clone(), + visit, event: event.event.clone(), session_id: Some(event.session_id.clone()), parent_session_id: event.parent_session_id.clone(), @@ -424,7 +431,7 @@ impl CodergenBackend for AgentApiBackend { context: &Context, thread_id: Option<&str>, emitter: &Arc, - stage_dir: &std::path::Path, + _stage_dir: &std::path::Path, sandbox: &Arc, tool_hooks: Option>, ) -> Result { @@ -479,6 +486,7 @@ impl CodergenBackend for AgentApiBackend { spawn_event_forwarder( &session, node.id.clone(), + current_visit(context), Arc::clone(emitter), Arc::clone(&file_tracking), ); @@ -486,6 +494,7 @@ impl CodergenBackend for AgentApiBackend { // Emit Prompt event before processing emitter.emit(&WorkflowRunEvent::Prompt { stage: node.id.clone(), + visit: current_visit(context), text: prompt.to_string(), mode: Some("agent".to_string()), provider: Some(actual_provider.as_str().to_string()), @@ -552,6 +561,7 @@ impl CodergenBackend for AgentApiBackend { spawn_event_forwarder( &session, node.id.clone(), + current_visit(context), Arc::clone(emitter), Arc::clone(&file_tracking), ); @@ -633,15 +643,6 @@ impl CodergenBackend for AgentApiBackend { (v, s.last.clone()) }; - let provider_used = serde_json::json!({ - "mode": "agent", - "provider": actual_provider.as_str(), - "model": &actual_model, - }); - if let Ok(json) = serde_json::to_string_pretty(&provider_used) { - let _ = std::fs::write(stage_dir.join("provider_used.json"), json); - } - // Cache session back for reuse on success. if let Some(key) = reuse_key { self.sessions.lock().unwrap().insert(key, session); diff --git a/lib/crates/fabro-workflow/src/handler/llm/cli.rs b/lib/crates/fabro-workflow/src/handler/llm/cli.rs index 5fc21503d..a72ddce4b 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/cli.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/cli.rs @@ -15,6 +15,7 @@ use crate::error::FabroError; use crate::event::{EventEmitter, WorkflowRunEvent}; use crate::outcome::StageUsage; use crate::outcome::compute_stage_cost; +use crate::run_dir::visit_from_context; use fabro_graphviz::graph::Node; /// Maps a provider to its corresponding CLI tool metadata. @@ -56,6 +57,10 @@ impl AgentCli { } } +fn current_visit(context: &Context) -> u32 { + u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX) +} + /// Ensure the CLI tool for the given provider is installed in the sandbox. /// /// Checks if the CLI binary exists; if not, installs Node.js (if missing) and @@ -496,6 +501,7 @@ impl CodergenBackend for AgentCliBackend { let command = cli_command_for_provider(provider, model, &prompt_path); emitter.emit(&WorkflowRunEvent::AgentCliStarted { node_id: node.id.clone(), + visit: current_visit(_context), mode: "cli".to_string(), provider: provider.as_str().to_string(), model: model.to_string(), diff --git a/lib/crates/fabro-workflow/src/handler/prompt.rs b/lib/crates/fabro-workflow/src/handler/prompt.rs index 0f1c71b32..fd7e7b797 100644 --- a/lib/crates/fabro-workflow/src/handler/prompt.rs +++ b/lib/crates/fabro-workflow/src/handler/prompt.rs @@ -15,8 +15,7 @@ use fabro_graphviz::graph::{Graph, Node}; use tokio::fs; use super::agent::{ - CodergenBackend, CodergenResult, expand_variables, extract_status_fields, - sync_provider_used_to_store, truncate, + CodergenBackend, CodergenResult, expand_variables, extract_status_fields, truncate, }; use super::{EngineServices, Handler}; @@ -107,6 +106,20 @@ impl Handler for PromptHandler { fs::write(stage_dir.join("prompt.md"), &prompt).await?; } + let prompt_provider = node + .provider() + .map(String::from) + .or_else(|| Some(Provider::default_from_env().as_str().to_string())); + let prompt_model = node.model().map(String::from); + services.emitter.emit(&WorkflowRunEvent::Prompt { + stage: node.id.clone(), + visit: u32::try_from(visit).unwrap_or(u32::MAX), + text: prompt.clone(), + mode: Some("prompt".to_string()), + provider: prompt_provider.clone(), + model: prompt_model.clone(), + }); + // 3. Call LLM backend (one_shot) let (response_text, stage_usage, backend_files_touched) = if let Some(backend) = &self.backend { @@ -114,10 +127,7 @@ impl Handler for PromptHandler { .one_shot(node, &prompt, system_prompt.as_deref(), &stage_dir) .await; match result { - Ok(CodergenResult::Full(outcome)) => { - sync_provider_used_to_store(&stage_dir, &node_ref, services).await?; - return Ok(outcome); - } + Ok(CodergenResult::Full(outcome)) => return Ok(outcome), Ok(CodergenResult::Text { text, usage, @@ -167,7 +177,6 @@ impl Handler for PromptHandler { } else { fs::write(stage_dir.join("response.md"), &response_text).await?; } - sync_provider_used_to_store(&stage_dir, &node_ref, services).await?; // 5. Build and write status let mut outcome = Outcome::success(); @@ -205,7 +214,11 @@ mod tests { EngineServices::test_default() } - async fn make_services_with_run_store() -> (EngineServices, Arc) { + async fn make_services_with_run_store() -> ( + EngineServices, + Arc, + crate::event::StoreProgressLogger, + ) { let store = InMemoryStore::default(); let run_store = store .create_run(&fixtures::RUN_1, chrono::Utc::now(), None) @@ -215,7 +228,9 @@ mod tests { run_store: Some(Arc::clone(&run_store)), ..EngineServices::test_default() }; - (services, run_store) + let logger = crate::event::StoreProgressLogger::new(Arc::clone(&run_store)); + logger.register(services.emitter.as_ref()); + (services, run_store, logger) } #[tokio::test] @@ -318,7 +333,7 @@ mod tests { } #[tokio::test] - async fn prompt_handler_persists_provider_used_in_run_store() { + async fn prompt_handler_projects_provider_used_from_prompt_events() { use fabro_agent::Sandbox; struct ProviderOneShotBackend; @@ -344,13 +359,8 @@ mod tests { _node: &Node, _prompt: &str, _system_prompt: Option<&str>, - stage_dir: &Path, + _stage_dir: &Path, ) -> Result { - std::fs::write( - stage_dir.join("provider_used.json"), - r#"{"mode":"prompt","provider":"openai","model":"gpt-5.4"}"#, - ) - .unwrap(); Ok(CodergenResult::Text { text: "one-shot response".to_string(), usage: None, @@ -369,12 +379,13 @@ mod tests { let context = Context::new(); let graph = Graph::new("test"); let tmp = TempDir::new().unwrap(); - let (services, run_store) = make_services_with_run_store().await; + let (services, run_store, logger) = make_services_with_run_store().await; handler .execute(&node, &context, &graph, tmp.path(), &services) .await .unwrap(); + logger.flush().await; let snapshot = run_store .get_node(&NodeVisitRef { diff --git a/lib/crates/fabro-workflow/src/pipeline/retro.rs b/lib/crates/fabro-workflow/src/pipeline/retro.rs index 58cf394b1..1c687c6ec 100644 --- a/lib/crates/fabro-workflow/src/pipeline/retro.rs +++ b/lib/crates/fabro-workflow/src/pipeline/retro.rs @@ -76,6 +76,7 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option { if !event.event.is_streaming_noise() { emitter.emit(&WorkflowRunEvent::Agent { stage: "retro".to_string(), + visit: 1, event: event.event.clone(), session_id: Some(event.session_id.clone()), parent_session_id: event.parent_session_id.clone(), diff --git a/lib/crates/fabro-workflow/tests/it/integration.rs b/lib/crates/fabro-workflow/tests/it/integration.rs index 33293c152..a896505ba 100644 --- a/lib/crates/fabro-workflow/tests/it/integration.rs +++ b/lib/crates/fabro-workflow/tests/it/integration.rs @@ -12063,6 +12063,7 @@ impl Handler for KeepaliveHandler { tokio::time::sleep(std::time::Duration::from_millis(self.interval_ms)).await; services.emitter.emit(&WorkflowRunEvent::Prompt { stage: node.id.clone(), + visit: 1, text: "keepalive".to_string(), mode: None, provider: None,