scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-15 23:32:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

refactor: remove legacy node file writes from workflow handlers

Browse source
This commit is contained in:
Bryan Helmkamp 2026-04-03 23:12:40 -07:00
parent 47ce67d147
commit 8250909282
No known key found for this signature in database
21 changed files with 480 additions and 684 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
lib/crates/fabro-cli/src/commands/store/dump.rs
View file

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

19
lib/crates/fabro-cli/tests/it/cmd/rm.rs
View file

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

1
lib/crates/fabro-cli/tests/it/cmd/run.rs
View file

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

18
lib/crates/fabro-cli/tests/it/workflow/agent_linear.rs
View file

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

8
lib/crates/fabro-cli/tests/it/workflow/command_agent_mixed.rs
View file

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

8
lib/crates/fabro-cli/tests/it/workflow/command_pipeline.rs
View file

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

5
lib/crates/fabro-cli/tests/it/workflow/full_stack.rs
View file

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

17
lib/crates/fabro-cli/tests/it/workflow/mod.rs
View file

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

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

@ -45,6 +45,20 @@ pub struct NodeState {
pub stderr: Option<String>,
}
#[derive(Debug, Clone, serde::Deserialize)]
struct RunUsage {
input_tokens: i64,
output_tokens: i64,
#[serde(default)]
reasoning_tokens: Option<i64>,
#[serde(default)]
cache_read_tokens: Option<i64>,
#[serde(default)]
cache_write_tokens: Option<i64>,
#[serde(default)]
cost: Option<f64>,
}
#[derive(Debug, Clone, Default)]
pub(crate) struct EventProjectionCache {
pub last_seq: u32,
@ -475,7 +489,7 @@ fn conclusion_from_completed(
properties: &serde_json::Map<String, Value>,
timestamp: DateTime<Utc>,
) -> Result<Conclusion> {
let usage = optional_json::<fabro_types::StageUsage>(properties, "usage")?;
let usage = optional_json::<RunUsage>(properties, "usage")?;
Ok(Conclusion {
timestamp,
status: StageStatus::from_str(&required_string(properties, "status")?).map_err(|err| {

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

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

97
lib/crates/fabro-workflow/src/git.rs
View file

@ -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<u8>)> {
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;

190
lib/crates/fabro-workflow/src/handler/agent.rs
View file

@ -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<serde_json::Value>,
) -> 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"

140
lib/crates/fabro-workflow/src/handler/command.rs
View file

@ -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<Outcome, FabroError> {
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);

12
lib/crates/fabro-workflow/src/handler/llm/api.rs
View file

@ -415,7 +415,7 @@ impl CodergenBackend for AgentApiBackend {
tool_hooks: Option<Arc<dyn fabro_agent::ToolHookCallback>>,
) -> Result<CodergenResult, FabroError> {
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::<Provider>().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();

43
lib/crates/fabro-workflow/src/handler/parallel.rs
View file

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

40
lib/crates/fabro-workflow/src/handler/prompt.rs
View file

@ -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<Outcome, FabroError> {
// 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)
}

43
lib/crates/fabro-workflow/src/lifecycle/disk.rs
View file

@ -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<Option<StageUsage>>;
type WfNodeResult = NodeResult<Option<StageUsage>>;
@ -59,45 +59,10 @@ pub(super) fn build_checkpoint(
impl RunLifecycle<WorkflowGraph> 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(())
}

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

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

1
lib/crates/fabro-workflow/src/operations/start.rs
View file

@ -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<Started, FabroError> {
std::fs::create_dir_all(run_dir).map_err(|err| FabroError::Io(err.to_string()))?;
let state = services
.run_store
.state()

143
lib/crates/fabro-workflow/src/test_support.rs
View file

@ -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<Checkpoint>,
}
struct InitializedState {
initialized: Initialized,
store_logger: StoreProgressLogger,
}
fn bound_emitter(run_id: fabro_types::RunId, observer: &Arc<EventEmitter>) -> Arc<EventEmitter> {
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<EventEmitter>,
sandbox: Arc<dyn Sandbox>,
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<EventEmitter>,
@ -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<EventEmitter>,
sandbox: Arc<dyn Sandbox>,
graph: &GvGraph,
run_options: &RunOptions,
hook_runner: Arc<fabro_hooks::HookRunner>,
env: Option<HashMap<String, String>>,
) -> 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<EventEmitter>,
@ -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,

329
lib/crates/fabro-workflow/tests/it/integration.rs
View file

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