diff --git a/lib/crates/fabro-cli/src/commands/store/dump.rs b/lib/crates/fabro-cli/src/commands/store/dump.rs index ef96487fc..f9ce6e11d 100644 --- a/lib/crates/fabro-cli/src/commands/store/dump.rs +++ b/lib/crates/fabro-cli/src/commands/store/dump.rs @@ -448,6 +448,7 @@ mod tests { &WorkflowRunEvent::CommandStarted { node_id: "code".to_string(), script: "echo hi".to_string(), + command: "echo hi".to_string(), language: "sh".to_string(), timeout_ms: None, }, diff --git a/lib/crates/fabro-cli/tests/it/cmd/rm.rs b/lib/crates/fabro-cli/tests/it/cmd/rm.rs index 688538fe4..a84d5c5df 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/rm.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/rm.rs @@ -202,17 +202,15 @@ fn rm_partial_failure_json_includes_removed_and_errors() { } #[test] -fn rm_json_reports_removed_run_when_store_delete_fails() { +fn rm_json_removes_run_when_store_locator_is_corrupt() { let context = test_context!(); let run = setup_completed_dry_run(&context); let by_id_path = find_store_catalog_entry(&context.storage_dir.join("store"), &run.run_id); let original = std::fs::read(&by_id_path) .unwrap_or_else(|err| panic!("failed to read {}: {err}", by_id_path.display())); - // This intentionally mutates the backing store metadata to force a store-only - // delete failure. Reproducing that failure through public commands is not - // practical, and the command contract under test is still `fabro rm`'s JSON - // partial-success reporting. + // Corrupt the by-id locator. Deletion should still succeed via the by-start + // fallback path instead of surfacing a false partial failure. std::fs::write(&by_id_path, b"{not valid json") .unwrap_or_else(|err| panic!("failed to corrupt {}: {err}", by_id_path.display())); scopeguard::defer! { @@ -226,20 +224,15 @@ fn rm_json_reports_removed_run_when_store_delete_fails() { .expect("command should run"); assert!( - !output.status.success(), - "rm should report the store failure" + output.status.success(), + "rm should still succeed when the locator is corrupt" ); let value: Value = serde_json::from_slice(&output.stdout).expect("rm JSON should parse"); assert_eq!( value["removed"], Value::Array(vec![Value::String(run.run_id.clone())]) ); - assert_eq!(value["errors"][0]["identifier"], run.run_id); - assert!( - value["errors"][0]["error"] - .as_str() - .is_some_and(|error| error.contains("failed to delete store state")) - ); + assert_eq!(value["errors"], Value::Array(Vec::new())); assert!( !run.run_dir.exists(), "run directory should still be deleted" diff --git a/lib/crates/fabro-cli/tests/it/cmd/run.rs b/lib/crates/fabro-cli/tests/it/cmd/run.rs index 823f83723..d333ade5f 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/run.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/run.rs @@ -690,6 +690,7 @@ fn json_run_implies_auto_approve_for_human_gates() { "node_id": "ship", "node_label": "ship", "properties": { + "command": "echo shipped", "language": "shell", "script": "echo shipped" }, diff --git a/lib/crates/fabro-cli/tests/it/workflow/agent_linear.rs b/lib/crates/fabro-cli/tests/it/workflow/agent_linear.rs index b41d29595..4041bfe23 100644 --- a/lib/crates/fabro-cli/tests/it/workflow/agent_linear.rs +++ b/lib/crates/fabro-cli/tests/it/workflow/agent_linear.rs @@ -1,6 +1,8 @@ use fabro_test::test_context; -use super::{completed_nodes, find_run_dir, fixture, read_conclusion, sandbox_tests, timeout_for}; +use super::{ + completed_nodes, find_run_dir, fixture, has_event, read_conclusion, sandbox_tests, timeout_for, +}; sandbox_tests!(agent_linear, keys = ["ANTHROPIC_API_KEY"]); @@ -32,14 +34,12 @@ fn scenario_agent_linear(sandbox: &str) { "work should be completed" ); - let prompt_path = run_dir.join("nodes/work/prompt.md"); - assert!(prompt_path.exists(), "nodes/work/prompt.md should exist"); - - let response_path = run_dir.join("nodes/work/response.md"); assert!( - response_path.exists(), - "nodes/work/response.md should exist" + has_event(&run_dir, "stage.prompt"), + "progress should contain stage.prompt" + ); + assert!( + has_event(&run_dir, "stage.completed"), + "progress should contain stage.completed" ); - let response = std::fs::read_to_string(&response_path).unwrap(); - assert!(!response.is_empty(), "response.md should not be empty"); } diff --git a/lib/crates/fabro-cli/tests/it/workflow/command_agent_mixed.rs b/lib/crates/fabro-cli/tests/it/workflow/command_agent_mixed.rs index 78c834073..f385a154a 100644 --- a/lib/crates/fabro-cli/tests/it/workflow/command_agent_mixed.rs +++ b/lib/crates/fabro-cli/tests/it/workflow/command_agent_mixed.rs @@ -1,6 +1,9 @@ use fabro_test::test_context; -use super::{completed_nodes, find_run_dir, fixture, read_conclusion, sandbox_tests, timeout_for}; +use super::{ + completed_nodes, find_run_dir, fixture, read_conclusion, sandbox_tests, store_dump_export, + timeout_for, +}; sandbox_tests!(command_agent_mixed, keys = ["ANTHROPIC_API_KEY"]); @@ -40,7 +43,8 @@ fn scenario_command_agent_mixed(sandbox: &str) { "verify should be completed" ); - let stdout = std::fs::read_to_string(run_dir.join("nodes/verify/stdout.log")) + let export_dir = store_dump_export(&context, &run_dir.file_name().unwrap().to_string_lossy()); + let stdout = std::fs::read_to_string(export_dir.join("nodes/verify/visit-1/stdout.log")) .expect("verify stdout.log should exist"); assert!( stdout.contains("SCENARIO_FLAG_42"), diff --git a/lib/crates/fabro-cli/tests/it/workflow/command_pipeline.rs b/lib/crates/fabro-cli/tests/it/workflow/command_pipeline.rs index 56c85b034..7d09af4b9 100644 --- a/lib/crates/fabro-cli/tests/it/workflow/command_pipeline.rs +++ b/lib/crates/fabro-cli/tests/it/workflow/command_pipeline.rs @@ -1,6 +1,9 @@ use fabro_test::test_context; -use super::{completed_nodes, find_run_dir, fixture, read_conclusion, sandbox_tests, timeout_for}; +use super::{ + completed_nodes, find_run_dir, fixture, read_conclusion, sandbox_tests, store_dump_export, + timeout_for, +}; sandbox_tests!(command_pipeline); @@ -39,7 +42,8 @@ fn scenario_command_pipeline(sandbox: &str) { "step2 should be completed" ); - let stdout1 = std::fs::read_to_string(run_dir.join("nodes/step1/stdout.log")) + let export_dir = store_dump_export(&context, &run_dir.file_name().unwrap().to_string_lossy()); + let stdout1 = std::fs::read_to_string(export_dir.join("nodes/step1/visit-1/stdout.log")) .expect("step1 stdout.log should exist"); assert!( stdout1.contains("hello-from-step1"), diff --git a/lib/crates/fabro-cli/tests/it/workflow/full_stack.rs b/lib/crates/fabro-cli/tests/it/workflow/full_stack.rs index c55c5acf4..b1601d6f6 100644 --- a/lib/crates/fabro-cli/tests/it/workflow/full_stack.rs +++ b/lib/crates/fabro-cli/tests/it/workflow/full_stack.rs @@ -2,7 +2,7 @@ use fabro_test::test_context; use super::{ completed_nodes, find_run_dir, fixture, has_event, read_conclusion, read_json, sandbox_tests, - timeout_for, + store_dump_export, timeout_for, }; sandbox_tests!(full_stack, keys = ["ANTHROPIC_API_KEY"]); @@ -68,7 +68,8 @@ fn scenario_full_stack(sandbox: &str) { } // Verify node stdout should contain PASS - let stdout = std::fs::read_to_string(run_dir.join("nodes/verify/stdout.log")) + let export_dir = store_dump_export(&context, &run_dir.file_name().unwrap().to_string_lossy()); + let stdout = std::fs::read_to_string(export_dir.join("nodes/verify/visit-1/stdout.log")) .expect("verify stdout.log should exist"); assert!( stdout.contains("PASS"), diff --git a/lib/crates/fabro-cli/tests/it/workflow/mod.rs b/lib/crates/fabro-cli/tests/it/workflow/mod.rs index 5c6c413af..809a27bd7 100644 --- a/lib/crates/fabro-cli/tests/it/workflow/mod.rs +++ b/lib/crates/fabro-cli/tests/it/workflow/mod.rs @@ -11,6 +11,7 @@ mod real_cli; use std::path::{Path, PathBuf}; use std::time::Duration; +use fabro_test::TestContext; use serde_json::Value; pub(super) fn fixture(name: &str) -> PathBuf { @@ -53,6 +54,22 @@ pub(super) fn has_event(run_dir: &Path, event_name: &str) -> bool { }) } +pub(super) fn store_dump_export(context: &TestContext, run_id: &str) -> PathBuf { + let output_dir = context.temp_dir.join(format!("store-dump-{run_id}")); + context + .command() + .args([ + "store", + "dump", + "--output", + output_dir.to_str().unwrap(), + run_id, + ]) + .assert() + .success(); + output_dir +} + /// Find the single run directory under `storage_dir/runs/`. pub(super) fn find_run_dir(storage_dir: &Path) -> PathBuf { let runs_base = storage_dir.join("runs"); diff --git a/lib/crates/fabro-store/src/run_state.rs b/lib/crates/fabro-store/src/run_state.rs index 5c8d86f49..044ecb718 100644 --- a/lib/crates/fabro-store/src/run_state.rs +++ b/lib/crates/fabro-store/src/run_state.rs @@ -45,6 +45,20 @@ pub struct NodeState { pub stderr: Option, } +#[derive(Debug, Clone, serde::Deserialize)] +struct RunUsage { + input_tokens: i64, + output_tokens: i64, + #[serde(default)] + reasoning_tokens: Option, + #[serde(default)] + cache_read_tokens: Option, + #[serde(default)] + cache_write_tokens: Option, + #[serde(default)] + cost: Option, +} + #[derive(Debug, Clone, Default)] pub(crate) struct EventProjectionCache { pub last_seq: u32, @@ -475,7 +489,7 @@ fn conclusion_from_completed( properties: &serde_json::Map, timestamp: DateTime, ) -> Result { - let usage = optional_json::(properties, "usage")?; + let usage = optional_json::(properties, "usage")?; Ok(Conclusion { timestamp, status: StageStatus::from_str(&required_string(properties, "status")?).map_err(|err| { diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs index f2686afea..8ca733485 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/lib/crates/fabro-workflow/src/event.rs @@ -429,6 +429,7 @@ pub enum WorkflowRunEvent { CommandStarted { node_id: String, script: String, + command: String, language: String, #[serde(default, skip_serializing_if = "Option::is_none")] timeout_ms: Option, diff --git a/lib/crates/fabro-workflow/src/git.rs b/lib/crates/fabro-workflow/src/git.rs index deb90a884..a6b175292 100644 --- a/lib/crates/fabro-workflow/src/git.rs +++ b/lib/crates/fabro-workflow/src/git.rs @@ -306,52 +306,6 @@ pub fn sanitize_ref_component(s: &str) -> String { } /// Filenames allowed in per-node directories on the shadow branch. -const NODE_FILE_ALLOWLIST: &[&str] = &[ - "prompt.md", - "response.md", - "status.json", - "provider_used.json", - "diff.patch", - "script_invocation.json", - "script_timing.json", - "parallel_results.json", -]; - -/// Maximum size (bytes) for a single node file. Files larger than this are skipped. -const MAX_NODE_FILE_SIZE: u64 = 512 * 1024; - -/// Scan `{run_dir}/nodes/` for allowlisted files and return them as -/// `("nodes/{subdir}/{filename}", bytes)` entries suitable for the shadow tree. -pub fn scan_node_files(run_dir: &Path) -> Vec<(String, Vec)> { - let nodes_dir = run_dir.join("nodes"); - let Ok(entries) = std::fs::read_dir(&nodes_dir) else { - return Vec::new(); - }; - - let mut result = Vec::new(); - for entry in entries.flatten() { - let path = entry.path(); - if !path.is_dir() { - continue; - } - let subdir_name = match path.file_name().and_then(|n| n.to_str()) { - Some(n) => n.to_string(), - None => continue, - }; - for filename in NODE_FILE_ALLOWLIST { - let file_path = path.join(filename); - match std::fs::metadata(&file_path) { - Ok(meta) if meta.is_file() && meta.len() <= MAX_NODE_FILE_SIZE => {} - _ => continue, - } - if let Ok(data) = std::fs::read(&file_path) { - result.push((format!("nodes/{subdir_name}/{filename}"), data)); - } - } - } - result -} - #[cfg(test)] mod tests { use super::*; @@ -455,57 +409,6 @@ mod tests { assert!(!wt_path.exists()); } - #[test] - fn scan_node_files_picks_up_allowlisted() { - let dir = tempfile::tempdir().unwrap(); - let run_dir = dir.path(); - let node_dir = run_dir.join("nodes").join("work"); - fs::create_dir_all(&node_dir).unwrap(); - fs::write(node_dir.join("prompt.md"), "hello").unwrap(); - fs::write(node_dir.join("response.md"), "world").unwrap(); - fs::write(node_dir.join("not_allowed.txt"), "skip me").unwrap(); - - let files = scan_node_files(run_dir); - let paths: Vec<&str> = files.iter().map(|(p, _)| p.as_str()).collect(); - assert!(paths.contains(&"nodes/work/prompt.md")); - assert!(paths.contains(&"nodes/work/response.md")); - assert!(!paths.iter().any(|p| p.contains("not_allowed"))); - } - - #[test] - fn scan_node_files_skips_oversized() { - let dir = tempfile::tempdir().unwrap(); - let run_dir = dir.path(); - let node_dir = run_dir.join("nodes").join("big"); - fs::create_dir_all(&node_dir).unwrap(); - // Write a file just over the 512KB limit - let big_data = vec![0u8; 512 * 1024 + 1]; - fs::write(node_dir.join("prompt.md"), &big_data).unwrap(); - - let files = scan_node_files(run_dir); - assert!(files.is_empty()); - } - - #[test] - fn scan_node_files_handles_visit_suffixes() { - let dir = tempfile::tempdir().unwrap(); - let run_dir = dir.path(); - let node_dir = run_dir.join("nodes").join("work-visit_2"); - fs::create_dir_all(&node_dir).unwrap(); - fs::write(node_dir.join("status.json"), "{}").unwrap(); - - let files = scan_node_files(run_dir); - assert_eq!(files.len(), 1); - assert_eq!(files[0].0, "nodes/work-visit_2/status.json"); - } - - #[test] - fn scan_node_files_empty_when_no_nodes_dir() { - let dir = tempfile::tempdir().unwrap(); - let files = scan_node_files(dir.path()); - assert!(files.is_empty()); - } - #[tokio::test] async fn scan_node_files_from_state_reconstructs_allowlisted_entries() { use fabro_store::EventPayload; diff --git a/lib/crates/fabro-workflow/src/handler/agent.rs b/lib/crates/fabro-workflow/src/handler/agent.rs index b31ddfd81..7680d290a 100644 --- a/lib/crates/fabro-workflow/src/handler/agent.rs +++ b/lib/crates/fabro-workflow/src/handler/agent.rs @@ -5,13 +5,12 @@ use std::sync::Arc; use async_trait::async_trait; use fabro_agent::Sandbox; use fabro_model::Provider; -use fabro_types::{RunId, StageId}; -use tokio::fs; +use fabro_types::RunId; use crate::context::keys; use crate::context::{Context, WorkflowContext}; use crate::error::FabroError; -use crate::event::EventEmitter; +use crate::event::{EventEmitter, WorkflowRunEvent}; use crate::outcome::{ FailureCategory, FailureDetail, Outcome, OutcomeExt, StageStatus, StageUsage, }; @@ -198,91 +197,6 @@ pub(crate) fn truncate(s: &str, max_chars: usize) -> &str { } } -pub(crate) fn stage_dir(run_dir: &Path, node_id: &str, visit: u32) -> std::path::PathBuf { - let node_dir = if visit <= 1 { - node_id.to_string() - } else { - format!("{node_id}-visit_{visit}") - }; - run_dir.join("nodes").join(node_dir) -} - -pub(crate) fn status_json_value(outcome: &Outcome) -> serde_json::Value { - let mut status = serde_json::Map::new(); - status.insert( - "status".to_string(), - serde_json::Value::String(outcome.status.to_string()), - ); - status.insert( - "outcome".to_string(), - serde_json::Value::String(outcome.status.to_string()), - ); - if let Some(label) = outcome.preferred_label.as_ref() { - status.insert( - "preferred_next_label".to_string(), - serde_json::Value::String(label.clone()), - ); - } - if !outcome.suggested_next_ids.is_empty() { - status.insert( - "suggested_next_ids".to_string(), - serde_json::json!(outcome.suggested_next_ids), - ); - } - if let Some(failure) = outcome.failure.as_ref() { - status.insert( - "failure_reason".to_string(), - serde_json::Value::String(failure.message.clone()), - ); - } - if !outcome.context_updates.is_empty() { - status.insert( - "context_updates".to_string(), - serde_json::Value::Object( - outcome - .context_updates - .iter() - .map(|(key, value)| (key.clone(), value.clone())) - .collect(), - ), - ); - } - serde_json::Value::Object(status) -} - -pub(crate) async fn write_provider_used_file( - services: &EngineServices, - stage_dir: &Path, - node_id: &str, - visit: u32, - fallback: Option, -) -> Result<(), FabroError> { - let provider_used = services - .run_store - .state() - .await - .ok() - .and_then(|state| { - let node = StageId::new(node_id, visit); - state - .node(&node) - .and_then(|node_state| node_state.provider_used.clone()) - }) - .or(fallback); - let Some(provider_used) = provider_used else { - return Ok(()); - }; - fs::write( - stage_dir.join("provider_used.json"), - serde_json::to_vec_pretty(&provider_used).map_err(|err| { - FabroError::handler(format!("failed to serialize provider_used.json: {err}")) - })?, - ) - .await - .map_err(|err| FabroError::handler(format!("failed to write provider_used.json: {err}")))?; - Ok(()) -} - /// 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 { @@ -337,13 +251,19 @@ impl Handler for AgentHandler { }; let visit = u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX); - let stage_dir = stage_dir(_run_dir, &node.id, visit); - fs::create_dir_all(&stage_dir) - .await - .map_err(|err| FabroError::handler(format!("failed to create stage dir: {err}")))?; - fs::write(stage_dir.join("prompt.md"), &prompt) - .await - .map_err(|err| FabroError::handler(format!("failed to write prompt file: {err}")))?; + 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, + text: prompt.clone(), + mode: Some("agent".to_string()), + provider: prompt_provider, + model: prompt_model, + }); // 3. Call LLM backend (agent loop) let thread_id = context.thread_id(); @@ -399,6 +319,24 @@ impl Handler for AgentHandler { ) }; + let response_model = stage_usage + .as_ref() + .map(|usage| usage.model.clone()) + .or_else(|| node.model().map(String::from)) + .unwrap_or_default(); + let response_provider = node + .provider() + .map(String::from) + .or_else(|| Some(Provider::default_from_env().as_str().to_string())) + .unwrap_or_default(); + services.emitter.emit(&WorkflowRunEvent::PromptCompleted { + node_id: node.id.clone(), + response: response_text.clone(), + model: response_model, + provider: response_provider, + usage: stage_usage.clone(), + }); + // Build and write status let mut outcome = Outcome::success(); outcome.notes = Some(format!("Stage completed: {}", node.id)); @@ -447,33 +385,6 @@ impl Handler for AgentHandler { } outcome.usage = stage_usage; outcome.files_touched = backend_files_touched; - fs::write(stage_dir.join("response.md"), &response_text) - .await - .map_err(|err| FabroError::handler(format!("failed to write response file: {err}")))?; - fs::write( - stage_dir.join("status.json"), - serde_json::to_vec_pretty(&status_json_value(&outcome)) - .map_err(|err| FabroError::handler(format!("failed to serialize status: {err}")))?, - ) - .await - .map_err(|err| FabroError::handler(format!("failed to write status file: {err}")))?; - write_provider_used_file( - services, - &stage_dir, - &node.id, - visit, - Some(serde_json::json!({ - "mode": if node.backend() == Some("cli") { "cli" } else { "agent" }, - "provider": node - .provider() - .map_or_else( - || Provider::default_from_env().as_str().to_string(), - String::from, - ), - "model": node.model().map(String::from).unwrap_or_default(), - })), - ) - .await?; Ok(outcome) } @@ -569,16 +480,20 @@ mod tests { AttrValue::String("Build a feature".to_string()), ); let tmp = TempDir::new().unwrap(); + let (services, run_store, logger) = make_services_with_run_store().await; handler - .execute(&node, &context, &graph, tmp.path(), &make_services()) + .execute(&node, &context, &graph, tmp.path(), &services) .await .unwrap(); + logger.flush().await; - let prompt_content = - std::fs::read_to_string(tmp.path().join("nodes").join("plan").join("prompt.md")) - .unwrap(); - assert_eq!(prompt_content, "Achieve: Build a feature"); + let state = run_store.state().await.unwrap(); + let node_state = state.node(&StageId::new("plan", 1)).unwrap(); + assert_eq!( + node_state.prompt.as_deref(), + Some("Achieve: Build a feature") + ); } #[tokio::test] @@ -592,16 +507,17 @@ mod tests { let context = test_context(); let graph = Graph::new("test"); let tmp = TempDir::new().unwrap(); + let (services, run_store, logger) = make_services_with_run_store().await; handler - .execute(&node, &context, &graph, tmp.path(), &make_services()) + .execute(&node, &context, &graph, tmp.path(), &services) .await .unwrap(); + logger.flush().await; - let prompt_content = - std::fs::read_to_string(tmp.path().join("nodes").join("work").join("prompt.md")) - .unwrap(); - assert_eq!(prompt_content, "Do work"); + let state = run_store.state().await.unwrap(); + let node_state = state.node(&StageId::new("work", 1)).unwrap(); + assert_eq!(node_state.prompt.as_deref(), Some("Do work")); } #[tokio::test] @@ -1299,15 +1215,17 @@ Some text in between. ); let graph = Graph::new("test"); let tmp = TempDir::new().unwrap(); + let (services, run_store, logger) = make_services_with_run_store().await; handler - .execute(&node, &context, &graph, tmp.path(), &make_services()) + .execute(&node, &context, &graph, tmp.path(), &services) .await .unwrap(); + logger.flush().await; - let prompt_content = - std::fs::read_to_string(tmp.path().join("nodes").join("report").join("prompt.md")) - .unwrap(); + let state = run_store.state().await.unwrap(); + let node_state = state.node(&StageId::new("report", 1)).unwrap(); + let prompt_content = node_state.prompt.as_deref().unwrap(); assert!( prompt_content.contains("## Script Output\nAll tests passed"), "prompt.md should contain preamble" diff --git a/lib/crates/fabro-workflow/src/handler/command.rs b/lib/crates/fabro-workflow/src/handler/command.rs index a01534ea2..8eb437f24 100644 --- a/lib/crates/fabro-workflow/src/handler/command.rs +++ b/lib/crates/fabro-workflow/src/handler/command.rs @@ -7,7 +7,6 @@ use crate::event::WorkflowRunEvent; use crate::outcome::{Outcome, OutcomeExt}; use async_trait::async_trait; use fabro_graphviz::graph::{Graph, Node}; -use tokio::fs; use super::{EngineServices, Handler}; @@ -60,7 +59,7 @@ impl Handler for CommandHandler { node: &Node, _context: &Context, _graph: &Graph, - run_dir: &Path, + _run_dir: &Path, services: &EngineServices, ) -> Result { let script = node @@ -86,34 +85,18 @@ impl Handler for CommandHandler { ))); } - services.emitter.emit(&WorkflowRunEvent::CommandStarted { - node_id: node.id.clone(), - script: script.to_string(), - language: language.to_string(), - timeout_ms: timeout_ms(node), - }); - let command = if language == "python" { format!("python3 -c {}", shell_quote(script)) } else { script.to_string() }; - let stage_dir = run_dir.join("nodes").join(&node.id); - fs::create_dir_all(&stage_dir) - .await - .map_err(|err| FabroError::handler(format!("failed to create stage dir: {err}")))?; - fs::write( - stage_dir.join("script_invocation.json"), - serde_json::to_vec_pretty(&serde_json::json!({ - "script": script, - "command": command, - "language": language, - "timeout_ms": timeout_ms(node), - })) - .map_err(|err| FabroError::handler(format!("failed to serialize invocation: {err}")))?, - ) - .await - .map_err(|err| FabroError::handler(format!("failed to write invocation file: {err}")))?; + services.emitter.emit(&WorkflowRunEvent::CommandStarted { + node_id: node.id.clone(), + script: script.to_string(), + command: command.clone(), + language: language.to_string(), + timeout_ms: timeout_ms(node), + }); let timeout_ms = node .timeout() @@ -138,23 +121,6 @@ impl Handler for CommandHandler { duration_ms: result.duration_ms, timed_out: result.timed_out, }); - fs::write(stage_dir.join("stdout.log"), &result.stdout) - .await - .map_err(|err| FabroError::handler(format!("failed to write stdout log: {err}")))?; - fs::write(stage_dir.join("stderr.log"), &result.stderr) - .await - .map_err(|err| FabroError::handler(format!("failed to write stderr log: {err}")))?; - fs::write( - stage_dir.join("script_timing.json"), - serde_json::to_vec_pretty(&serde_json::json!({ - "duration_ms": result.duration_ms, - "exit_code": (!result.timed_out).then_some(result.exit_code), - "timed_out": result.timed_out, - })) - .map_err(|err| FabroError::handler(format!("failed to serialize timing: {err}")))?, - ) - .await - .map_err(|err| FabroError::handler(format!("failed to write timing file: {err}")))?; if result.timed_out { return Err(FabroError::handler(format!( @@ -391,19 +357,17 @@ mod tests { let context = Context::new(); let graph = Graph::new("test"); let run_dir = tempfile::tempdir().unwrap(); + let (services, run_store, logger) = make_services_with_run_store().await; handler - .execute(&node, &context, &graph, run_dir.path(), &make_services()) + .execute(&node, &context, &graph, run_dir.path(), &services) .await .unwrap(); + logger.flush().await; - let invocation_path = run_dir - .path() - .join("nodes") - .join("script_node") - .join("script_invocation.json"); - let content = std::fs::read_to_string(&invocation_path).unwrap(); - let json: serde_json::Value = serde_json::from_str(&content).unwrap(); + let snapshot = run_store.state().await.unwrap(); + let node_state = snapshot.node(&StageId::new("script_node", 1)).unwrap(); + let json = node_state.script_invocation.as_ref().unwrap(); assert_eq!(json["command"], "echo hello"); assert_eq!(json["language"], "shell"); assert_eq!(json["timeout_ms"], serde_json::Value::Null); @@ -424,19 +388,17 @@ mod tests { let context = Context::new(); let graph = Graph::new("test"); let run_dir = tempfile::tempdir().unwrap(); + let (services, run_store, logger) = make_services_with_run_store().await; handler - .execute(&node, &context, &graph, run_dir.path(), &make_services()) + .execute(&node, &context, &graph, run_dir.path(), &services) .await .unwrap(); + logger.flush().await; - let invocation_path = run_dir - .path() - .join("nodes") - .join("script_node") - .join("script_invocation.json"); - let content = std::fs::read_to_string(&invocation_path).unwrap(); - let json: serde_json::Value = serde_json::from_str(&content).unwrap(); + let snapshot = run_store.state().await.unwrap(); + let node_state = snapshot.node(&StageId::new("script_node", 1)).unwrap(); + let json = node_state.script_invocation.as_ref().unwrap(); assert_eq!(json["command"], "echo hello"); assert_eq!(json["language"], "shell"); assert_eq!(json["timeout_ms"], 5000); @@ -453,16 +415,19 @@ mod tests { let context = Context::new(); let graph = Graph::new("test"); let run_dir = tempfile::tempdir().unwrap(); + let (services, run_store, logger) = make_services_with_run_store().await; handler - .execute(&node, &context, &graph, run_dir.path(), &make_services()) + .execute(&node, &context, &graph, run_dir.path(), &services) .await .unwrap(); + logger.flush().await; - let stage_dir = run_dir.path().join("nodes").join("script_node"); - let stdout = std::fs::read_to_string(stage_dir.join("stdout.log")).unwrap(); + let snapshot = run_store.state().await.unwrap(); + let node_state = snapshot.node(&StageId::new("script_node", 1)).unwrap(); + let stdout = node_state.stdout.as_deref().unwrap(); assert_eq!(stdout.trim(), "hello"); - let stderr = std::fs::read_to_string(stage_dir.join("stderr.log")).unwrap(); + let stderr = node_state.stderr.as_deref().unwrap(); assert_eq!(stderr, ""); } @@ -477,14 +442,17 @@ mod tests { let context = Context::new(); let graph = Graph::new("test"); let run_dir = tempfile::tempdir().unwrap(); + let (services, run_store, logger) = make_services_with_run_store().await; handler - .execute(&node, &context, &graph, run_dir.path(), &make_services()) + .execute(&node, &context, &graph, run_dir.path(), &services) .await .unwrap(); + logger.flush().await; - let stage_dir = run_dir.path().join("nodes").join("script_node"); - let stderr = std::fs::read_to_string(stage_dir.join("stderr.log")).unwrap(); + let snapshot = run_store.state().await.unwrap(); + let node_state = snapshot.node(&StageId::new("script_node", 1)).unwrap(); + let stderr = node_state.stderr.as_deref().unwrap(); assert_eq!(stderr.trim(), "oops"); } @@ -499,19 +467,17 @@ mod tests { let context = Context::new(); let graph = Graph::new("test"); let run_dir = tempfile::tempdir().unwrap(); + let (services, run_store, logger) = make_services_with_run_store().await; handler - .execute(&node, &context, &graph, run_dir.path(), &make_services()) + .execute(&node, &context, &graph, run_dir.path(), &services) .await .unwrap(); + logger.flush().await; - let timing_path = run_dir - .path() - .join("nodes") - .join("script_node") - .join("script_timing.json"); - let content = std::fs::read_to_string(&timing_path).unwrap(); - let json: serde_json::Value = serde_json::from_str(&content).unwrap(); + let snapshot = run_store.state().await.unwrap(); + let node_state = snapshot.node(&StageId::new("script_node", 1)).unwrap(); + let json = node_state.script_timing.as_ref().unwrap(); assert!(json["duration_ms"].is_u64()); assert_eq!(json["exit_code"], 0); assert_eq!(json["timed_out"], false); @@ -526,19 +492,17 @@ mod tests { let context = Context::new(); let graph = Graph::new("test"); let run_dir = tempfile::tempdir().unwrap(); + let (services, run_store, logger) = make_services_with_run_store().await; handler - .execute(&node, &context, &graph, run_dir.path(), &make_services()) + .execute(&node, &context, &graph, run_dir.path(), &services) .await .unwrap(); + logger.flush().await; - let timing_path = run_dir - .path() - .join("nodes") - .join("script_node") - .join("script_timing.json"); - let content = std::fs::read_to_string(&timing_path).unwrap(); - let json: serde_json::Value = serde_json::from_str(&content).unwrap(); + let snapshot = run_store.state().await.unwrap(); + let node_state = snapshot.node(&StageId::new("script_node", 1)).unwrap(); + let json = node_state.script_timing.as_ref().unwrap(); assert_eq!(json["exit_code"], 1); assert_eq!(json["timed_out"], false); } @@ -558,19 +522,17 @@ mod tests { let context = Context::new(); let graph = Graph::new("test"); let run_dir = tempfile::tempdir().unwrap(); + let (services, run_store, logger) = make_services_with_run_store().await; let _err = handler - .execute(&node, &context, &graph, run_dir.path(), &make_services()) + .execute(&node, &context, &graph, run_dir.path(), &services) .await .unwrap_err(); + logger.flush().await; - let timing_path = run_dir - .path() - .join("nodes") - .join("script_node") - .join("script_timing.json"); - let content = std::fs::read_to_string(&timing_path).unwrap(); - let json: serde_json::Value = serde_json::from_str(&content).unwrap(); + let snapshot = run_store.state().await.unwrap(); + let node_state = snapshot.node(&StageId::new("script_node", 1)).unwrap(); + let json = node_state.script_timing.as_ref().unwrap(); assert!(json["duration_ms"].is_u64()); assert_eq!(json["exit_code"], serde_json::Value::Null); assert_eq!(json["timed_out"], true); diff --git a/lib/crates/fabro-workflow/src/handler/llm/api.rs b/lib/crates/fabro-workflow/src/handler/llm/api.rs index d1e93538c..70dd2318b 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/api.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/api.rs @@ -415,7 +415,7 @@ impl CodergenBackend for AgentApiBackend { tool_hooks: Option>, ) -> Result { let actual_model = node.model().unwrap_or(&self.model).to_string(); - let actual_provider = node + let _actual_provider = node .provider() .and_then(|p| p.parse::().ok()) .unwrap_or(self.provider); @@ -470,16 +470,6 @@ impl CodergenBackend for AgentApiBackend { Arc::clone(&file_tracking), ); - // 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()), - model: Some(actual_model.clone()), - }); - // Record turn count before processing so we only aggregate new usage. let turns_before = session.history().turns().len(); diff --git a/lib/crates/fabro-workflow/src/handler/parallel.rs b/lib/crates/fabro-workflow/src/handler/parallel.rs index 2a768356a..8994348dc 100644 --- a/lib/crates/fabro-workflow/src/handler/parallel.rs +++ b/lib/crates/fabro-workflow/src/handler/parallel.rs @@ -5,7 +5,6 @@ use std::time::Instant; use async_trait::async_trait; use fabro_agent::{Sandbox, WorktreeOptions, WorktreeSandbox}; use fabro_types::RunId; -use tokio::fs; use tokio::sync::Semaphore; use crate::context::keys; @@ -476,20 +475,6 @@ impl Handler for ParallelHandler { entry }) .collect(); - let stage_dir = run_dir.join("nodes").join(&node.id); - fs::create_dir_all(&stage_dir) - .await - .map_err(|err| FabroError::handler(format!("failed to create stage dir: {err}")))?; - fs::write( - stage_dir.join("parallel_results.json"), - serde_json::to_vec_pretty(&results_json).map_err(|err| { - FabroError::handler(format!("failed to serialize parallel results: {err}")) - })?, - ) - .await - .map_err(|err| { - FabroError::handler(format!("failed to write parallel results file: {err}")) - })?; context.set(keys::PARALLEL_RESULTS, serde_json::json!(results_json)); context.set(keys::PARALLEL_BRANCH_COUNT, serde_json::json!(total)); @@ -595,7 +580,7 @@ fn find_join_node(results: &[BranchResult], graph: &Graph) -> Option { mod tests { use super::*; use fabro_graphviz::graph::{AttrValue, Edge}; - use fabro_store::SlateStore; + use fabro_store::{SlateStore, StageId}; use fabro_types::fixtures; use object_store::memory::InMemory; use std::sync::Arc; @@ -639,7 +624,15 @@ mod tests { #[tokio::test] async fn parallel_handler_with_branches() { - let services = make_services(); + let store = test_store(); + let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); + let services = EngineServices { + emitter: Arc::new(crate::event::EventEmitter::new(fixtures::RUN_1)), + run_store: run_store.clone(), + ..EngineServices::test_default() + }; + let logger = crate::event::StoreProgressLogger::new(run_store.clone()); + logger.register(services.emitter.as_ref()); let mut node = Node::new("par"); node.attrs.insert( "shape".to_string(), @@ -662,6 +655,7 @@ mod tests { .execute(&node, &context, &graph, tmp.path(), &services) .await .unwrap(); + logger.flush().await; assert_eq!(outcome.status, StageStatus::Success); assert!(outcome.notes.as_deref().unwrap().contains("2 branches")); @@ -670,18 +664,9 @@ mod tests { let results = context.get(keys::PARALLEL_RESULTS); assert!(results.is_some()); - // Check parallel_results.json was written - let results_path = tmp - .path() - .join("nodes") - .join("par") - .join("parallel_results.json"); - assert!( - results_path.exists(), - "parallel_results.json should be written" - ); - let content = std::fs::read_to_string(&results_path).unwrap(); - let parsed: serde_json::Value = serde_json::from_str(&content).unwrap(); + let state = run_store.state().await.unwrap(); + let node_state = state.node(&StageId::new("par", 1)).unwrap(); + let parsed = node_state.parallel_results.as_ref().unwrap(); assert!( parsed.is_array(), "parallel_results.json should be a JSON array" diff --git a/lib/crates/fabro-workflow/src/handler/prompt.rs b/lib/crates/fabro-workflow/src/handler/prompt.rs index 5be3a1049..8aa8ab7e9 100644 --- a/lib/crates/fabro-workflow/src/handler/prompt.rs +++ b/lib/crates/fabro-workflow/src/handler/prompt.rs @@ -1,20 +1,17 @@ use std::path::Path; -use async_trait::async_trait; -use tokio::fs; - use crate::context::keys; use crate::context::{Context, WorkflowContext}; use crate::error::FabroError; use crate::event::WorkflowRunEvent; use crate::outcome::Outcome; use crate::run_dir::visit_from_context; +use async_trait::async_trait; use fabro_graphviz::graph::{Graph, Node}; use fabro_model::Provider; use super::agent::{ - CodergenBackend, CodergenResult, expand_variables, extract_status_fields, stage_dir, - status_json_value, truncate, write_provider_used_file, + CodergenBackend, CodergenResult, expand_variables, extract_status_fields, truncate, }; use super::{EngineServices, Handler}; @@ -48,7 +45,7 @@ impl Handler for PromptHandler { node: &Node, context: &Context, graph: &Graph, - run_dir: &Path, + _run_dir: &Path, services: &EngineServices, ) -> Result { // 1. Build prompt (prepend fidelity preamble if present) @@ -63,14 +60,7 @@ impl Handler for PromptHandler { } else { format!("{preamble}\n\n{expanded}") }; - let visit = u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX); - let stage_dir = stage_dir(run_dir, &node.id, visit); - fs::create_dir_all(&stage_dir) - .await - .map_err(|err| FabroError::handler(format!("failed to create stage dir: {err}")))?; - fs::write(stage_dir.join("prompt.md"), &prompt) - .await - .map_err(|err| FabroError::handler(format!("failed to write prompt file: {err}")))?; + let _visit = u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX); // 1b. Discover project docs for system prompt when project_memory is enabled let system_prompt = if node.project_memory() { @@ -176,28 +166,6 @@ impl Handler for PromptHandler { extract_status_fields(&response_text, &mut outcome); outcome.usage = stage_usage; outcome.files_touched = backend_files_touched; - fs::write(stage_dir.join("response.md"), &response_text) - .await - .map_err(|err| FabroError::handler(format!("failed to write response file: {err}")))?; - fs::write( - stage_dir.join("status.json"), - serde_json::to_vec_pretty(&status_json_value(&outcome)) - .map_err(|err| FabroError::handler(format!("failed to serialize status: {err}")))?, - ) - .await - .map_err(|err| FabroError::handler(format!("failed to write status file: {err}")))?; - write_provider_used_file( - services, - &stage_dir, - &node.id, - visit, - Some(serde_json::json!({ - "mode": "prompt", - "provider": prompt_provider.unwrap_or_default(), - "model": prompt_model.unwrap_or_default(), - })), - ) - .await?; Ok(outcome) } diff --git a/lib/crates/fabro-workflow/src/lifecycle/disk.rs b/lib/crates/fabro-workflow/src/lifecycle/disk.rs index 427247278..d142d7ca3 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/disk.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/disk.rs @@ -15,7 +15,7 @@ use super::circuit_breaker::CircuitBreakerLifecycle; use super::git::GitCheckpointResult; use crate::graph::WorkflowGraph; use crate::graph::WorkflowNode; -use crate::outcome::{OutcomeExt, StageUsage}; +use crate::outcome::StageUsage; type WfRunState = ExecutionState>; type WfNodeResult = NodeResult>; @@ -59,45 +59,10 @@ pub(super) fn build_checkpoint( impl RunLifecycle for DiskLifecycle { async fn after_node( &self, - node: &WorkflowNode, - result: &mut WfNodeResult, - state: &WfRunState, + _node: &WorkflowNode, + _result: &mut WfNodeResult, + _state: &WfRunState, ) -> CoreResult<()> { - let visit = - u32::try_from(*state.node_visits.get(node.id()).unwrap_or(&1)).unwrap_or(u32::MAX); - let node_dir = if visit <= 1 { - self.run_dir.join("nodes").join(node.id()) - } else { - self.run_dir - .join("nodes") - .join(format!("{}-visit_{visit}", node.id())) - }; - fs::create_dir_all(&node_dir) - .await - .map_err(|err| CoreError::Other(format!("failed to create node dir: {err}")))?; - let mut status = serde_json::Map::new(); - status.insert( - "status".to_string(), - serde_json::Value::String(result.outcome.status.to_string()), - ); - status.insert( - "outcome".to_string(), - serde_json::Value::String(result.outcome.status.to_string()), - ); - if let Some(failure_reason) = result.outcome.failure_reason() { - status.insert( - "failure_reason".to_string(), - serde_json::Value::String(failure_reason.to_string()), - ); - } - fs::write( - node_dir.join("status.json"), - serde_json::to_vec_pretty(&serde_json::Value::Object(status)).map_err(|err| { - CoreError::Other(format!("failed to serialize status.json: {err}")) - })?, - ) - .await - .map_err(|err| CoreError::Other(format!("failed to write status.json: {err}")))?; Ok(()) } diff --git a/lib/crates/fabro-workflow/src/operations/fork.rs b/lib/crates/fabro-workflow/src/operations/fork.rs index 36b173712..372a4a62d 100644 --- a/lib/crates/fabro-workflow/src/operations/fork.rs +++ b/lib/crates/fabro-workflow/src/operations/fork.rs @@ -72,12 +72,10 @@ fn fork_from_entry( .map_err(|e| anyhow::anyhow!("failed to read source metadata: {e}"))?; let mut run_record_bytes = None; - let mut start_record_bytes = None; let mut sandbox_bytes = None; for (path, data) in source_entries { match path { "run.json" => run_record_bytes = Some(data), - "start.json" => start_record_bytes = Some(data), "sandbox.json" => sandbox_bytes = Some(data), _ => {} } @@ -91,21 +89,15 @@ fn fork_from_entry( let new_run_record_bytes = serde_json::to_vec_pretty(&run_record).context("failed to serialize new run.json")?; - let new_start_record_bytes = if start_record_bytes.is_some() { - let now = new_run_id.created_at(); - let start_record = StartRecord { - run_id: new_run_id, - start_time: now, - run_branch: Some(new_run_branch.clone()), - base_sha: None, - }; - Some( - serde_json::to_vec_pretty(&start_record) - .context("failed to serialize new start.json")?, - ) - } else { - None + let now = new_run_id.created_at(); + let start_record = StartRecord { + run_id: new_run_id, + start_time: now, + run_branch: Some(new_run_branch.clone()), + base_sha: None, }; + let new_start_record_bytes = + serde_json::to_vec_pretty(&start_record).context("failed to serialize new start.json")?; let checkpoint_bytes = store .read_blob_at(entry.metadata_commit_oid, "checkpoint.json") @@ -123,9 +115,7 @@ fn fork_from_entry( serde_json::to_vec_pretty(&checkpoint).context("failed to serialize checkpoint.json")?; let mut init_entries: Vec<(&str, &[u8])> = vec![("run.json", &new_run_record_bytes)]; - if let Some(ref start_record) = new_start_record_bytes { - init_entries.push(("start.json", start_record)); - } + init_entries.push(("start.json", &new_start_record_bytes)); if let Some(ref sandbox) = sandbox_bytes { init_entries.push(("sandbox.json", sandbox)); } @@ -133,8 +123,10 @@ fn fork_from_entry( new_bs .write_entries(&init_entries, "init run") .map_err(|e| anyhow::anyhow!("failed to write init metadata entries: {e}"))?; + let mut checkpoint_entries: Vec<(&str, &[u8])> = vec![("checkpoint.json", &checkpoint_bytes)]; + checkpoint_entries.extend(init_entries.iter().copied()); new_bs - .write_entry("checkpoint.json", &checkpoint_bytes, "checkpoint") + .write_entries(&checkpoint_entries, "checkpoint") .map_err(|e| anyhow::anyhow!("failed to write metadata entries: {e}"))?; if push { diff --git a/lib/crates/fabro-workflow/src/operations/start.rs b/lib/crates/fabro-workflow/src/operations/start.rs index 62f31a634..ab94042cb 100644 --- a/lib/crates/fabro-workflow/src/operations/start.rs +++ b/lib/crates/fabro-workflow/src/operations/start.rs @@ -80,6 +80,7 @@ pub struct Started { /// Start a fresh workflow run. Errors if a checkpoint already exists (use `resume()` instead). pub async fn start(run_dir: &Path, services: StartServices) -> Result { + std::fs::create_dir_all(run_dir).map_err(|err| FabroError::Io(err.to_string()))?; let state = services .run_store .state() diff --git a/lib/crates/fabro-workflow/src/test_support.rs b/lib/crates/fabro-workflow/src/test_support.rs index 376e27cd7..304baf5e5 100644 --- a/lib/crates/fabro-workflow/src/test_support.rs +++ b/lib/crates/fabro-workflow/src/test_support.rs @@ -5,11 +5,11 @@ use std::time::Duration; use fabro_agent::Sandbox; use fabro_graphviz::graph::Graph as GvGraph; -use fabro_store::SlateStore; +use fabro_store::{RunProjection, SlateStore}; use object_store::memory::InMemory; -use crate::error::Result; -use crate::event::{EventEmitter, WorkflowRunEvent, append_workflow_event}; +use crate::error::{FabroError, Result}; +use crate::event::{EventEmitter, StoreProgressLogger, WorkflowRunEvent, append_workflow_event}; use crate::handler::HandlerRegistry; use crate::outcome::Outcome; use crate::pipeline; @@ -23,6 +23,11 @@ struct InitializedOptions { checkpoint: Option, } +struct InitializedState { + initialized: Initialized, + store_logger: StoreProgressLogger, +} + fn bound_emitter(run_id: fabro_types::RunId, observer: &Arc) -> Arc { let emitter = Arc::new(EventEmitter::new(run_id)); let observer_clone = Arc::clone(observer); @@ -37,7 +42,7 @@ async fn initialized( graph: &GvGraph, run_options: &RunOptions, options: InitializedOptions, -) -> Initialized { +) -> InitializedState { std::fs::create_dir_all(&run_options.run_dir).expect("failed to create run dir"); let store = Arc::new(SlateStore::new( Arc::new(InMemory::new()), @@ -80,23 +85,28 @@ async fn initialized( .await .expect("failed to seed run.created event in run store"); let emitter = bound_emitter(run_options.run_id, &emitter); - Initialized { - graph: graph.clone(), - source: String::new(), - run_options: run_options.clone(), - run_store, - checkpoint: options.checkpoint, - seed_context: None, - emitter, - sandbox, - registry: Arc::new(registry), - on_node: None, - hook_runner: options.hook_runner, - env: options.env, - dry_run: run_options.dry_run_enabled(), - llm_client: None, - model: String::new(), - provider: fabro_llm::Provider::Anthropic, + let store_logger = StoreProgressLogger::new(run_store.clone()); + store_logger.register(emitter.as_ref()); + InitializedState { + initialized: Initialized { + graph: graph.clone(), + source: String::new(), + run_options: run_options.clone(), + run_store, + checkpoint: options.checkpoint, + seed_context: None, + emitter, + sandbox, + registry: Arc::new(registry), + on_node: None, + hook_runner: options.hook_runner, + env: options.env, + dry_run: run_options.dry_run_enabled(), + llm_client: None, + model: String::new(), + provider: fabro_llm::Provider::Anthropic, + }, + store_logger, } } @@ -120,10 +130,41 @@ pub async fn run_graph( }, ) .await; - let executed = pipeline::execute(initialized).await; + let executed = pipeline::execute(initialized.initialized).await; executed.outcome } +pub async fn run_graph_with_state( + registry: HandlerRegistry, + emitter: Arc, + sandbox: Arc, + graph: &GvGraph, + run_options: &RunOptions, +) -> Result<(Outcome, RunProjection)> { + let initialized = initialized( + registry, + emitter, + sandbox, + graph, + run_options, + InitializedOptions { + hook_runner: None, + env: HashMap::new(), + checkpoint: None, + }, + ) + .await; + let executed = pipeline::execute(initialized.initialized).await; + let outcome = executed.outcome?; + initialized.store_logger.flush().await; + let state = executed + .run_store + .state() + .await + .map_err(|err| FabroError::engine(err.to_string()))?; + Ok((outcome, state)) +} + pub async fn run_graph_with_hooks( registry: HandlerRegistry, emitter: Arc, @@ -146,10 +187,43 @@ pub async fn run_graph_with_hooks( }, ) .await; - let executed = pipeline::execute(initialized).await; + let executed = pipeline::execute(initialized.initialized).await; executed.outcome } +pub async fn run_graph_with_hooks_and_state( + registry: HandlerRegistry, + emitter: Arc, + sandbox: Arc, + graph: &GvGraph, + run_options: &RunOptions, + hook_runner: Arc, + env: Option>, +) -> Result<(Outcome, RunProjection)> { + let initialized = initialized( + registry, + emitter, + sandbox, + graph, + run_options, + InitializedOptions { + hook_runner: Some(hook_runner), + env: env.unwrap_or_default(), + checkpoint: None, + }, + ) + .await; + let executed = pipeline::execute(initialized.initialized).await; + let outcome = executed.outcome?; + initialized.store_logger.flush().await; + let state = executed + .run_store + .state() + .await + .map_err(|err| FabroError::engine(err.to_string()))?; + Ok((outcome, state)) +} + pub async fn run_graph_from_checkpoint( registry: HandlerRegistry, emitter: Arc, @@ -171,7 +245,7 @@ pub async fn run_graph_from_checkpoint( }, ) .await; - let executed = pipeline::execute(initialized).await; + let executed = pipeline::execute(initialized.initialized).await; executed.outcome } @@ -212,6 +286,27 @@ impl WorkflowRunner { .await } + pub async fn run_with_state( + &self, + graph: &GvGraph, + run_options: &RunOptions, + ) -> Result<(Outcome, RunProjection)> { + let registry = self + .registry + .lock() + .unwrap() + .take() + .expect("WorkflowRunner may only be used once"); + Box::pin(run_graph_with_state( + registry, + Arc::clone(&self.emitter), + Arc::clone(&self.sandbox), + graph, + run_options, + )) + .await + } + pub async fn run_from_checkpoint( &self, graph: &GvGraph, diff --git a/lib/crates/fabro-workflow/tests/it/integration.rs b/lib/crates/fabro-workflow/tests/it/integration.rs index 510631f3c..cc42be568 100644 --- a/lib/crates/fabro-workflow/tests/it/integration.rs +++ b/lib/crates/fabro-workflow/tests/it/integration.rs @@ -235,8 +235,8 @@ async fn end_to_end_linear_pipeline() { host_repo_path: None, git: None, }; - let outcome = engine - .run(&graph, &run_options) + let (outcome, state) = engine + .run_with_state(&graph, &run_options) .await .expect("run should succeed"); assert_eq!(outcome.status, StageStatus::Success); @@ -253,22 +253,15 @@ async fn end_to_end_linear_pipeline() { .contains(&"codergen_step".to_string()) ); - // Codergen handler writes prompt.md, response.md, status.json - let stage_dir = dir.path().join("nodes").join("codergen_step"); + let node_state = state + .node(&fabro_types::StageId::new("codergen_step", 1)) + .unwrap(); assert!( - stage_dir.join("prompt.md").exists(), - "prompt.md should exist" + node_state.response.is_some(), + "response should be projected" ); - assert!( - stage_dir.join("response.md").exists(), - "response.md should exist" - ); - assert!( - stage_dir.join("status.json").exists(), - "status.json should exist" - ); - - let prompt_content = std::fs::read_to_string(stage_dir.join("prompt.md")).unwrap(); + assert!(node_state.status.is_some(), "status should be projected"); + let prompt_content = node_state.prompt.as_deref().unwrap(); assert!( prompt_content.ends_with("Implement the feature"), "prompt should end with original prompt, got: {prompt_content}" @@ -1636,8 +1629,8 @@ async fn smoke_test_with_mock_codergen_backend() { host_repo_path: None, git: None, }; - let outcome = engine - .run(&graph, &run_options) + let (outcome, state) = engine + .run_with_state(&graph, &run_options) .await .expect("smoke test should succeed"); assert_eq!(outcome.status, StageStatus::Success); @@ -1658,19 +1651,20 @@ async fn smoke_test_with_mock_codergen_backend() { "should NOT have traversed fix path" ); - // Verify response.md was written by the mock backend - let plan_response = - std::fs::read_to_string(dir.path().join("nodes").join("plan").join("response.md")) - .expect("plan response should exist"); + let plan_state = state.node(&fabro_types::StageId::new("plan", 1)).unwrap(); + let plan_response = plan_state + .response + .as_deref() + .expect("plan response should exist"); assert!( plan_response.contains("Response for plan"), "mock backend should have written response, got: {plan_response}" ); - // Verify prompt.md had $goal expanded by the AgentHandler - let plan_prompt = - std::fs::read_to_string(dir.path().join("nodes").join("plan").join("prompt.md")) - .expect("plan prompt should exist"); + let plan_prompt = plan_state + .prompt + .as_deref() + .expect("plan prompt should exist"); assert!( plan_prompt.ends_with("Plan to achieve: Build and validate"), "prompt should end with original prompt, got: {plan_prompt}" @@ -2150,7 +2144,10 @@ async fn tool_handler_e2e() { host_repo_path: None, git: None, }; - let outcome = engine.run(&graph, &run_options).await.expect("run"); + let (outcome, _state) = engine + .run_with_state(&graph, &run_options) + .await + .expect("run"); assert_eq!(outcome.status, StageStatus::Success); let cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); @@ -2221,7 +2218,10 @@ async fn auto_approve_interviewer_e2e() { host_repo_path: None, git: None, }; - let outcome = engine.run(&graph, &run_options).await.expect("run"); + let (outcome, _state) = engine + .run_with_state(&graph, &run_options) + .await + .expect("run"); assert_eq!(outcome.status, StageStatus::Success); let cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); @@ -2358,7 +2358,10 @@ async fn branching_loop_back_on_failure() { host_repo_path: None, git: None, }; - let outcome = engine.run(&graph, &run_options).await.expect("run"); + let (outcome, _state) = engine + .run_with_state(&graph, &run_options) + .await + .expect("run"); assert_eq!(outcome.status, StageStatus::Success); let cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); @@ -2440,7 +2443,10 @@ async fn human_gate_loops_back() { host_repo_path: None, git: None, }; - let outcome = engine.run(&graph, &run_options).await.expect("run"); + let (outcome, _state) = engine + .run_with_state(&graph, &run_options) + .await + .expect("run"); assert_eq!(outcome.status, StageStatus::Success); let cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); @@ -2497,7 +2503,10 @@ async fn scenario_ship_a_feature() { host_repo_path: None, git: None, }; - let outcome = engine.run(&graph, &run_options).await.expect("run"); + let (outcome, _state) = engine + .run_with_state(&graph, &run_options) + .await + .expect("run"); assert_eq!(outcome.status, StageStatus::Success); let cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); @@ -3531,30 +3540,20 @@ async fn integration_smoke_plan_implement_review_done() { host_repo_path: None, git: None, }; - let outcome = engine.run(&graph, &run_options).await.expect("run"); + let (outcome, state) = engine + .run_with_state(&graph, &run_options) + .await + .expect("run"); assert_eq!(outcome.status, StageStatus::Success); - // Verify all nodes completed let cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); assert!(cp.completed_nodes.contains(&"plan".to_string())); assert!(cp.completed_nodes.contains(&"implement".to_string())); assert!(cp.completed_nodes.contains(&"review".to_string())); - // Verify prompt.md and response.md exist - assert!( - dir.path() - .join("nodes") - .join("plan") - .join("prompt.md") - .exists() - ); - assert!( - dir.path() - .join("nodes") - .join("plan") - .join("response.md") - .exists() - ); + let plan_state = state.node(&fabro_types::StageId::new("plan", 1)).unwrap(); + assert!(plan_state.prompt.is_some()); + assert!(plan_state.response.is_some()); // Verify events let collected = events.lock().unwrap(); @@ -6031,9 +6030,9 @@ mod real_llm { host_repo_path: None, git: None, }; - let outcome = tokio::time::timeout( + let (outcome, state) = tokio::time::timeout( std::time::Duration::from_secs(120), - engine.run(&graph, &run_options), + engine.run_with_state(&graph, &run_options), ) .await .expect("should not timeout") @@ -6052,9 +6051,10 @@ mod real_llm { assert_eq!(last_stage, Some("review")); // Verify actual LLM responses were written - let plan_response = - std::fs::read_to_string(dir.path().join("nodes").join("plan").join("response.md")) - .unwrap(); + let plan_response = state + .node(&fabro_types::StageId::new("plan", 1)) + .and_then(|node| node.response.as_deref()) + .unwrap(); assert!( !plan_response.is_empty(), "LLM should have generated a response" @@ -6370,9 +6370,9 @@ mod real_llm { host_repo_path: None, git: None, }; - let outcome = tokio::time::timeout( + let (outcome, state) = tokio::time::timeout( std::time::Duration::from_secs(30), - engine.run(&graph, &run_options), + engine.run_with_state(&graph, &run_options), ) .await .expect("should not timeout") @@ -6380,12 +6380,10 @@ mod real_llm { assert_eq!(outcome.status, StageStatus::Success); - let response_path = dir - .path() - .join("nodes") - .join("classify") - .join("response.md"); - let response = std::fs::read_to_string(&response_path).unwrap(); + let response = state + .node(&fabro_types::StageId::new("classify", 1)) + .and_then(|node| node.response.as_deref()) + .unwrap(); assert!(!response.is_empty(), "response.md should be non-empty"); } } @@ -7181,6 +7179,23 @@ impl HookTestRunner { ) .await } + + async fn run_with_state( + &self, + graph: &Graph, + run_options: &RunOptions, + ) -> Result<(Outcome, fabro_store::RunProjection), FabroError> { + fabro_workflow::test_support::run_graph_with_hooks_and_state( + make_linear_registry(), + Arc::clone(&self.emitter), + local_env(), + graph, + run_options, + Arc::clone(&self.hook_runner), + None, + ) + .await + } } fn emitter_with_events() -> ( @@ -7353,17 +7368,15 @@ async fn hook_stage_start_proceed_allows_execution() { let dir = tempfile::tempdir().unwrap(); let run_options = make_run_options(dir.path()); - let outcome = engine.run(&graph, &run_options).await.unwrap(); + let (outcome, state) = engine.run_with_state(&graph, &run_options).await.unwrap(); assert_eq!(outcome.status, StageStatus::Success); - // Work node should have executed (response.md exists) assert!( - dir.path() - .join("nodes") - .join("work") - .join("response.md") - .exists(), - "response.md should exist when StageStart hook proceeds" + state + .node(&fabro_types::StageId::new("work", 1)) + .and_then(|node| node.response.as_ref()) + .is_some(), + "response should exist when StageStart hook proceeds" ); } @@ -7379,18 +7392,16 @@ async fn hook_stage_start_skip_bypasses_node() { let dir = tempfile::tempdir().unwrap(); let run_options = make_run_options(dir.path()); - let outcome = engine.run(&graph, &run_options).await.unwrap(); + let (outcome, state) = engine.run_with_state(&graph, &run_options).await.unwrap(); // Pipeline reached exit with goal gates satisfied — per spec, SUCCESS. assert_eq!(outcome.status, StageStatus::Success); - // response.md should NOT exist for the work node (it was skipped) assert!( - !dir.path() - .join("nodes") - .join("work") - .join("response.md") - .exists(), - "response.md should not exist when StageStart hook skips node" + state + .node(&fabro_types::StageId::new("work", 1)) + .and_then(|node| node.response.as_ref()) + .is_none(), + "response should not exist when StageStart hook skips node" ); // StageStarted should NOT be emitted for hook-skipped stages (the stage never started) @@ -7440,27 +7451,23 @@ async fn hook_stage_start_matcher_filters_by_node_id() { let dir = tempfile::tempdir().unwrap(); let run_options = make_run_options(dir.path()); - let outcome = engine.run(&graph, &run_options).await.unwrap(); + let (outcome, state) = engine.run_with_state(&graph, &run_options).await.unwrap(); // Pipeline reached exit with goal gates satisfied — per spec, SUCCESS. assert_eq!(outcome.status, StageStatus::Success); - // step1 should have executed (response.md exists) assert!( - dir.path() - .join("nodes") - .join("step1") - .join("response.md") - .exists(), + state + .node(&fabro_types::StageId::new("step1", 1)) + .and_then(|node| node.response.as_ref()) + .is_some(), "step1 should execute because matcher doesn't match it" ); - // step2 should have been skipped (no response.md) assert!( - !dir.path() - .join("nodes") - .join("step2") - .join("response.md") - .exists(), + state + .node(&fabro_types::StageId::new("step2", 1)) + .and_then(|node| node.response.as_ref()) + .is_none(), "step2 should be skipped because matcher matches it" ); } @@ -7971,25 +7978,22 @@ async fn hook_matcher_regex_pattern() { let dir = tempfile::tempdir().unwrap(); let run_options = make_run_options(dir.path()); - let outcome = engine.run(&graph, &run_options).await.unwrap(); + let (outcome, state) = engine.run_with_state(&graph, &run_options).await.unwrap(); // Pipeline reached exit with goal gates satisfied — per spec, SUCCESS. assert_eq!(outcome.status, StageStatus::Success); - // Both step1 and step2 should be skipped assert!( - !dir.path() - .join("nodes") - .join("step1") - .join("response.md") - .exists(), + state + .node(&fabro_types::StageId::new("step1", 1)) + .and_then(|node| node.response.as_ref()) + .is_none(), "step1 should be skipped by regex ^step" ); assert!( - !dir.path() - .join("nodes") - .join("step2") - .join("response.md") - .exists(), + state + .node(&fabro_types::StageId::new("step2", 1)) + .and_then(|node| node.response.as_ref()) + .is_none(), "step2 should be skipped by regex ^step" ); } @@ -8231,13 +8235,15 @@ async fn run_fidelity_prompt_pipeline(fidelity: &str) -> String { host_repo_path: None, git: None, }; - engine - .run(&graph, &run_options) + let (_outcome, state) = engine + .run_with_state(&graph, &run_options) .await .expect("pipeline should succeed"); - std::fs::read_to_string(dir.path().join("nodes").join("report").join("prompt.md")) - .expect("report/prompt.md should exist") + state + .node(&fabro_types::StageId::new("report", 1)) + .and_then(|node| node.prompt.clone()) + .expect("report prompt should exist") } #[tokio::test] @@ -8429,8 +8435,8 @@ async fn large_context_values_are_offloaded_to_artifact_store() { host_repo_path: None, git: None, }; - let outcome = engine - .run(&graph, &run_options) + let (outcome, _state) = engine + .run_with_state(&graph, &run_options) .await .expect("pipeline should succeed"); assert_eq!(outcome.status, StageStatus::Success); @@ -8647,8 +8653,8 @@ async fn artifact_pointers_rewritten_for_remote_sandbox() { host_repo_path: None, git: None, }; - let outcome = engine - .run(&graph, &run_options) + let (outcome, _state) = engine + .run_with_state(&graph, &run_options) .await .expect("pipeline should succeed"); assert_eq!(outcome.status, StageStatus::Success); @@ -8776,43 +8782,28 @@ async fn node_dir_uses_visit_count_on_revisit() { host_repo_path: None, git: None, }; - let outcome = engine - .run(&graph, &run_options) + let (outcome, state) = engine + .run_with_state(&graph, &run_options) .await .expect("pipeline should succeed"); assert_eq!(outcome.status, StageStatus::Success); - // First visit: nodes/gated_work/status.json - let first = dir - .path() - .join("nodes") - .join("gated_work") - .join("status.json"); - assert!( - first.exists(), - "first visit directory should exist at {}", - first.display() + let first = state + .node(&fabro_types::StageId::new("gated_work", 1)) + .unwrap(); + let second = state + .node(&fabro_types::StageId::new("gated_work", 2)) + .unwrap(); + assert_eq!( + first.status.as_ref().unwrap().status, + StageStatus::Fail, + "first visit should fail" ); - - // Second visit: nodes/gated_work-visit_2/status.json - let second = dir - .path() - .join("nodes") - .join("gated_work-visit_2") - .join("status.json"); - assert!( - second.exists(), - "second visit directory should exist at {}", - second.display() + assert_eq!( + second.status.as_ref().unwrap().status, + StageStatus::Success, + "second visit should succeed" ); - - // Verify distinct content (first = fail, second = success) - let first_json: serde_json::Value = - serde_json::from_str(&std::fs::read_to_string(&first).unwrap()).unwrap(); - let second_json: serde_json::Value = - serde_json::from_str(&std::fs::read_to_string(&second).unwrap()).unwrap(); - assert_eq!(first_json["status"], "fail"); - assert_eq!(second_json["status"], "success"); } // --------------------------------------------------------------------------- @@ -9660,49 +9651,37 @@ async fn full_pipeline_with_cli_backend_node() { host_repo_path: None, git: None, }; - let outcome = engine - .run(&graph, &run_options) + let (outcome, state) = engine + .run_with_state(&graph, &run_options) .await .expect("pipeline should succeed"); assert_eq!(outcome.status, StageStatus::Success); - // Verify api_work used mock (its response.md should contain "Response for") - let api_response = std::fs::read_to_string( - dir.path() - .join("nodes") - .join("api_work") - .join("response.md"), - ) - .unwrap(); + let api_response = state + .node(&fabro_types::StageId::new("api_work", 1)) + .and_then(|node| node.response.as_deref()) + .unwrap(); assert!( api_response.starts_with("Response for api_work"), "API node should use mock: {api_response}" ); - // Verify cli_work used CLI backend (its response.md should contain CLI response) - let cli_response = std::fs::read_to_string( - dir.path() - .join("nodes") - .join("cli_work") - .join("response.md"), - ) - .unwrap(); + let cli_response = state + .node(&fabro_types::StageId::new("cli_work", 1)) + .and_then(|node| node.response.as_deref()) + .unwrap(); assert_eq!( cli_response, "CLI completed the task.", "CLI node should use CLI backend: {cli_response}" ); - // Verify cli_work wrote provider_used.json with mode=cli - let provider_json: serde_json::Value = serde_json::from_str( - &std::fs::read_to_string( - dir.path() - .join("nodes") - .join("cli_work") - .join("provider_used.json"), - ) - .unwrap(), - ) - .unwrap(); + let provider_json = state + .node(&fabro_types::StageId::new("cli_work", 1)) + .unwrap() + .provider_used + .as_ref() + .unwrap() + .clone(); assert_eq!(provider_json["mode"], "cli"); } @@ -9790,14 +9769,16 @@ async fn stylesheet_backend_property_routes_to_cli() { host_repo_path: None, git: None, }; - let outcome = engine - .run(&graph, &run_options) + let (outcome, state) = engine + .run_with_state(&graph, &run_options) .await .expect("pipeline should succeed"); assert_eq!(outcome.status, StageStatus::Success); - let response = - std::fs::read_to_string(dir.path().join("nodes").join("work").join("response.md")).unwrap(); + let response = state + .node(&fabro_types::StageId::new("work", 1)) + .and_then(|node| node.response.as_deref()) + .unwrap(); assert_eq!( response, "Styled CLI response.", "stylesheet-driven node should use CLI backend"