mirror of
https://github.com/fabro-sh/fabro.git
synced 2026-08-28 05:27:41 +00:00
Refactor workflow lifecycle into operations
This commit is contained in:
Bryan Helmkamp
parent
032b3c33e1
commit
6ba28531ad
36 changed files with 1003 additions and 1005 deletions
|
|
@ -23,9 +23,10 @@ use crate::jwt_auth::{AuthMode, AuthenticatedService, AuthenticatedUser};
|
|||
use fabro_interview::{Answer, Interviewer, QuestionType, WebInterviewer};
|
||||
use fabro_workflows::checkpoint::Checkpoint;
|
||||
use fabro_workflows::context::Context;
|
||||
use fabro_workflows::engine::{RunSettings, WorkflowRunEngine};
|
||||
use fabro_workflows::engine::WorkflowRunEngine;
|
||||
use fabro_workflows::event::{EventEmitter, WorkflowRunEvent};
|
||||
use fabro_workflows::handler::HandlerRegistry;
|
||||
use fabro_workflows::run_settings::RunSettings;
|
||||
|
||||
pub use fabro_types::{
|
||||
ApiQuestion, ApiQuestionOption, PaginatedRunList, PaginationMeta,
|
||||
|
|
|
|||
|
|
@ -108,8 +108,8 @@ fn build_live_diff_cmd(base_sha: &str, stat: bool, shortstat: bool) -> String {
|
|||
);
|
||||
format!(
|
||||
"{} add -N . && {} diff{flags} {quoted_sha}",
|
||||
fabro_workflows::engine::GIT_REMOTE,
|
||||
fabro_workflows::engine::GIT_REMOTE
|
||||
fabro_workflows::sandbox_git::GIT_REMOTE,
|
||||
fabro_workflows::sandbox_git::GIT_REMOTE
|
||||
)
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -24,29 +24,28 @@ pub struct ForkArgs {
|
|||
|
||||
pub fn run(args: &ForkArgs, styles: &Styles) -> Result<()> {
|
||||
let repo = Repository::discover(".").context("not in a git repository")?;
|
||||
let run_id = fabro_workflows::run_rewind::find_run_id_by_prefix(&repo, &args.run_id)?;
|
||||
let run_id = fabro_workflows::operations::find_run_id_by_prefix(&repo, &args.run_id)?;
|
||||
let store = Store::new(repo);
|
||||
|
||||
let timeline = fabro_workflows::run_rewind::build_timeline(&store, &run_id)?;
|
||||
let timeline = fabro_workflows::operations::build_timeline(&store, &run_id)?;
|
||||
|
||||
if args.list {
|
||||
let parallel_map = fabro_workflows::run_rewind::load_parallel_map(&store, &run_id);
|
||||
let parallel_map = fabro_workflows::operations::load_parallel_map(&store, &run_id);
|
||||
super::rewind::print_timeline(&timeline, ¶llel_map, styles);
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let entry = if let Some(target_str) = &args.target {
|
||||
let target = fabro_workflows::run_rewind::parse_target(target_str)?;
|
||||
let parallel_map = fabro_workflows::run_rewind::load_parallel_map(&store, &run_id);
|
||||
fabro_workflows::run_rewind::resolve_target(&timeline, &target, ¶llel_map)?
|
||||
let target = fabro_workflows::operations::parse_target(target_str)?;
|
||||
let parallel_map = fabro_workflows::operations::load_parallel_map(&store, &run_id);
|
||||
fabro_workflows::operations::resolve_target(&timeline, &target, ¶llel_map)?
|
||||
} else {
|
||||
timeline
|
||||
.last()
|
||||
.ok_or_else(|| anyhow::anyhow!("no checkpoints found for run {run_id}"))?
|
||||
};
|
||||
|
||||
let new_run_id =
|
||||
fabro_workflows::run_fork::execute_fork(&store, &run_id, entry, !args.no_push)?;
|
||||
let new_run_id = fabro_workflows::operations::fork(&store, &run_id, entry, !args.no_push)?;
|
||||
|
||||
eprintln!(
|
||||
"\nForked run {} -> {}",
|
||||
|
|
|
|||
|
|
@ -57,7 +57,8 @@ static RANKDIR_RE: LazyLock<regex::Regex> =
|
|||
pub fn run(args: &GraphArgs, styles: &Styles) -> anyhow::Result<()> {
|
||||
let (dot_path, _cfg) = fabro_config::project::resolve_workflow(&args.workflow)?;
|
||||
|
||||
let (_graph, diagnostics) = fabro_workflows::workflow::prepare_from_file(&dot_path)?;
|
||||
let validated = fabro_workflows::operations::create_from_file(&dot_path)?;
|
||||
let diagnostics = validated.diagnostics();
|
||||
|
||||
print_diagnostics(&diagnostics, styles);
|
||||
|
||||
|
|
|
|||
|
|
@ -1,35 +1,38 @@
|
|||
use std::io::IsTerminal;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, Instant};
|
||||
use std::time::Instant;
|
||||
|
||||
use anyhow::{bail, Context};
|
||||
use clap::Args;
|
||||
use fabro_agent::{DockerSandbox, DockerSandboxConfig, Sandbox, WorktreeConfig, WorktreeSandbox};
|
||||
use fabro_config::config::FabroConfig;
|
||||
use fabro_graphviz::graph::Graph;
|
||||
use fabro_interview::{AutoApproveInterviewer, ConsoleInterviewer, Interviewer};
|
||||
use fabro_model::{Catalog, Provider};
|
||||
use fabro_util::terminal::Styles;
|
||||
use fabro_workflows::backend::{AgentApiBackend, AgentCliBackend, BackendRouter};
|
||||
use fabro_workflows::checkpoint::Checkpoint;
|
||||
use fabro_workflows::engine::{GitCheckpointSettings, RunSettings};
|
||||
use fabro_workflows::event::{EventEmitter, RunNoticeLevel};
|
||||
use fabro_workflows::operations::{
|
||||
create_from_graph, start, StartFinalizeConfig, StartOptions, StartRetroConfig,
|
||||
};
|
||||
use fabro_workflows::outcome::StageStatus;
|
||||
use fabro_workflows::pipeline::{
|
||||
build_conclusion, classify_engine_result, persist_terminal_outcome,
|
||||
};
|
||||
use fabro_workflows::run_record::RunRecord;
|
||||
use fabro_workflows::run_settings::{GitCheckpointSettings, LifecycleConfig, RunSettings};
|
||||
use fabro_workflows::sandbox_provider::SandboxProvider;
|
||||
use indicatif::HumanDuration;
|
||||
|
||||
use super::detached_support::{DetachedRunBootstrapGuard, DetachedRunCompletionGuard};
|
||||
use super::run::{
|
||||
build_conclusion, build_event_envelope, cached_graph_path, classify_engine_result,
|
||||
default_run_dir, emit_run_notice, generate_retro, local_sandbox_with_callback,
|
||||
mint_github_token, persist_terminal_outcome, prepare_workflow_with_project_config,
|
||||
print_assets, print_final_output, resolve_daytona_config, resolve_fallback_chain,
|
||||
resolve_model_provider, resolve_ssh_clone_params, resolve_ssh_config, write_finalize_commit,
|
||||
write_run_config_snapshot, CliSandboxProvider, RunArgs,
|
||||
build_event_envelope, cached_graph_path, default_run_dir, emit_run_notice,
|
||||
local_sandbox_with_callback, mint_github_token, prepare_workflow_with_project_config,
|
||||
print_assets, print_final_output, print_retro_result, print_run_conclusion,
|
||||
resolve_daytona_config, resolve_fallback_chain, resolve_model_provider,
|
||||
resolve_ssh_clone_params, resolve_ssh_config, write_run_config_snapshot, CliSandboxProvider,
|
||||
RunArgs,
|
||||
};
|
||||
use crate::commands::shared::tilde_path;
|
||||
use fabro_config::project as project_config;
|
||||
use fabro_config::run as run_config;
|
||||
use fabro_workflows::devcontainer_bridge;
|
||||
|
|
@ -94,7 +97,7 @@ pub struct ResumeArgs {
|
|||
/// Intermediate state produced by the two resolution paths (checkpoint-file vs. git-branch).
|
||||
struct ResumeContext {
|
||||
checkpoint: Checkpoint,
|
||||
graph: Graph,
|
||||
validated: fabro_workflows::pipeline::Validated,
|
||||
run_id: String,
|
||||
run_dir: PathBuf,
|
||||
run_cfg: Option<FabroConfig>,
|
||||
|
|
@ -201,7 +204,9 @@ async fn prepare_from_checkpoint(
|
|||
true,
|
||||
false,
|
||||
)?;
|
||||
let (graph, source, _diagnostics) = prepared.validated.into_parts();
|
||||
let source = prepared.raw_source.clone();
|
||||
let validated = prepared.validated;
|
||||
let graph = validated.graph().clone();
|
||||
let run_cfg = prepared.run_cfg;
|
||||
let sandbox_provider = prepared.sandbox_provider;
|
||||
let workflow_slug = prepared.workflow_slug;
|
||||
|
|
@ -473,7 +478,7 @@ async fn prepare_from_checkpoint(
|
|||
|
||||
Ok(ResumeContext {
|
||||
checkpoint,
|
||||
graph,
|
||||
validated,
|
||||
run_id,
|
||||
run_dir,
|
||||
run_cfg,
|
||||
|
|
@ -509,7 +514,7 @@ async fn prepare_from_branch(
|
|||
(stripped.to_string(), run_arg.to_string())
|
||||
} else {
|
||||
let repo = git2::Repository::discover(".").context("not in a git repository")?;
|
||||
let id = fabro_workflows::run_rewind::find_run_id_by_prefix(&repo, run_arg)?;
|
||||
let id = fabro_workflows::operations::find_run_id_by_prefix(&repo, run_arg)?;
|
||||
let branch = format!("{}{}", fabro_workflows::git::RUN_BRANCH_PREFIX, id);
|
||||
(id, branch)
|
||||
};
|
||||
|
|
@ -544,7 +549,7 @@ async fn prepare_from_branch(
|
|||
.or_else(|| repo_info.as_ref().and_then(|(_, branch)| branch.clone()));
|
||||
let base_sha = start_record.as_ref().and_then(|s| s.base_sha.clone());
|
||||
|
||||
let (graph, graph_source, run_cfg, mut sandbox_provider, workflow_slug) =
|
||||
let (validated, graph_source, run_cfg, mut sandbox_provider, workflow_slug) =
|
||||
if let Some(ref workflow_path) = args.workflow {
|
||||
let prepared = prepare_workflow_with_project_config(
|
||||
&resume_as_run_args(args, workflow_path.clone()),
|
||||
|
|
@ -553,19 +558,16 @@ async fn prepare_from_branch(
|
|||
true,
|
||||
false,
|
||||
)?;
|
||||
{
|
||||
let (graph, source, _diagnostics) = prepared.validated.into_parts();
|
||||
(
|
||||
graph,
|
||||
source,
|
||||
prepared.run_cfg,
|
||||
prepared.sandbox_provider,
|
||||
prepared.workflow_slug,
|
||||
)
|
||||
}
|
||||
(
|
||||
prepared.validated,
|
||||
prepared.raw_source,
|
||||
prepared.run_cfg,
|
||||
prepared.sandbox_provider,
|
||||
prepared.workflow_slug,
|
||||
)
|
||||
} else if let Some(ref rec) = record {
|
||||
// Use the fully transformed graph from the RunRecord
|
||||
let graph = rec.graph.clone();
|
||||
let validated = create_from_graph(rec.graph.clone(), String::new());
|
||||
let source = String::new(); // no DOT source needed — graph is from RunRecord
|
||||
let run_cfg = Some(rec.config.clone());
|
||||
let sandbox_provider = if args.dry_run {
|
||||
|
|
@ -581,7 +583,7 @@ async fn prepare_from_branch(
|
|||
args.sandbox.map(Into::into).unwrap_or(sp)
|
||||
};
|
||||
(
|
||||
graph,
|
||||
validated,
|
||||
source,
|
||||
run_cfg,
|
||||
sandbox_provider,
|
||||
|
|
@ -590,6 +592,7 @@ async fn prepare_from_branch(
|
|||
} else {
|
||||
bail!("no run.json found on metadata branch for run {run_id}");
|
||||
};
|
||||
let graph = validated.graph().clone();
|
||||
|
||||
eprintln!(
|
||||
"{} {} from branch {} ({})",
|
||||
|
|
@ -903,7 +906,7 @@ async fn prepare_from_branch(
|
|||
|
||||
Ok(ResumeContext {
|
||||
checkpoint,
|
||||
graph,
|
||||
validated,
|
||||
run_id,
|
||||
run_dir,
|
||||
run_cfg,
|
||||
|
|
@ -931,7 +934,7 @@ async fn run_resumed(
|
|||
) -> anyhow::Result<()> {
|
||||
let ResumeContext {
|
||||
checkpoint,
|
||||
graph,
|
||||
validated,
|
||||
run_id,
|
||||
run_dir,
|
||||
mut run_cfg,
|
||||
|
|
@ -948,21 +951,7 @@ async fn run_resumed(
|
|||
github_app,
|
||||
mut status_guard,
|
||||
} = ctx;
|
||||
|
||||
// Track the last git commit SHA from CheckpointCompleted events
|
||||
let last_git_sha: Arc<std::sync::Mutex<Option<String>>> = Arc::new(std::sync::Mutex::new(None));
|
||||
{
|
||||
let sha_clone = Arc::clone(&last_git_sha);
|
||||
emitter.on_event(move |event| {
|
||||
if let fabro_workflows::event::WorkflowRunEvent::CheckpointCompleted {
|
||||
git_commit_sha: Some(sha),
|
||||
..
|
||||
} = event
|
||||
{
|
||||
*sha_clone.lock().unwrap() = Some(sha.clone());
|
||||
}
|
||||
});
|
||||
}
|
||||
let graph = validated.graph().clone();
|
||||
|
||||
// Create progress UI (verbose mode shows detailed turn/tool counts and token usage)
|
||||
let is_tty = std::io::stderr().is_terminal();
|
||||
|
|
@ -1238,34 +1227,7 @@ async fn run_resumed(
|
|||
}
|
||||
}
|
||||
});
|
||||
let mut engine = fabro_workflows::engine::WorkflowRunEngine::with_interviewer(
|
||||
registry,
|
||||
Arc::clone(&emitter),
|
||||
interviewer,
|
||||
Arc::clone(&sandbox),
|
||||
);
|
||||
if !sandbox_env.is_empty() {
|
||||
engine.set_env(sandbox_env);
|
||||
}
|
||||
if dry_run_mode {
|
||||
engine.set_dry_run(true);
|
||||
}
|
||||
// Wire up hook runner from run defaults (mirrors run_command)
|
||||
{
|
||||
let hooks = run_cfg
|
||||
.as_ref()
|
||||
.map(|cfg| &cfg.hooks)
|
||||
.unwrap_or(&run_defaults.hooks);
|
||||
if !hooks.is_empty() {
|
||||
let hook_config = fabro_hooks::HookConfig {
|
||||
hooks: hooks.clone(),
|
||||
};
|
||||
let runner = fabro_hooks::HookRunner::new(hook_config);
|
||||
engine.set_hook_runner(Arc::new(runner));
|
||||
}
|
||||
}
|
||||
|
||||
let lifecycle = fabro_workflows::engine::LifecycleConfig {
|
||||
let lifecycle = LifecycleConfig {
|
||||
setup_commands,
|
||||
setup_command_timeout_ms: 300_000,
|
||||
devcontainer_phases,
|
||||
|
|
@ -1274,29 +1236,56 @@ async fn run_resumed(
|
|||
// Defuse the bootstrap guard — engine.run() has taken ownership of lifecycle status.
|
||||
status_guard.defuse();
|
||||
|
||||
// Safety net: if we panic or return early, best-effort cleanup via spawn (mirrors run_command).
|
||||
let preserve = super::run::resolve_preserve_sandbox(
|
||||
args.preserve_sandbox,
|
||||
run_cfg.as_ref(),
|
||||
&run_defaults,
|
||||
);
|
||||
let sandbox_for_cleanup = Arc::clone(&sandbox);
|
||||
let cleanup_guard = scopeguard::guard((), move |()| {
|
||||
if preserve {
|
||||
return;
|
||||
}
|
||||
let rt = tokio::runtime::Handle::try_current();
|
||||
if let Ok(handle) = rt {
|
||||
handle.spawn(async move {
|
||||
let _ = sandbox_for_cleanup.cleanup().await;
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
let run_start = Instant::now();
|
||||
let engine_result = engine
|
||||
.run_with_lifecycle(&graph, &mut settings, lifecycle, Some(&checkpoint))
|
||||
.await;
|
||||
let pr_config = settings.pull_request().cloned();
|
||||
let started = start(
|
||||
validated,
|
||||
StartOptions {
|
||||
init: fabro_workflows::pipeline::InitOptions {
|
||||
run_id: run_id.clone(),
|
||||
run_dir: run_dir.clone(),
|
||||
dry_run: dry_run_mode,
|
||||
emitter: Arc::clone(&emitter),
|
||||
sandbox: Arc::clone(&sandbox),
|
||||
registry: Arc::new(registry),
|
||||
lifecycle,
|
||||
run_settings: settings,
|
||||
hooks: fabro_hooks::HookConfig {
|
||||
hooks: run_cfg
|
||||
.as_ref()
|
||||
.map(|cfg| cfg.hooks.clone())
|
||||
.unwrap_or_else(|| run_defaults.hooks.clone()),
|
||||
},
|
||||
sandbox_env,
|
||||
checkpoint: Some(checkpoint),
|
||||
seed_context: None,
|
||||
},
|
||||
retro: StartRetroConfig {
|
||||
enabled: !args.no_retro && project_config::is_retro_enabled(),
|
||||
dry_run: dry_run_mode,
|
||||
llm_client: if dry_run_mode {
|
||||
None
|
||||
} else {
|
||||
fabro_llm::client::Client::from_env().await.ok()
|
||||
},
|
||||
provider: provider_enum,
|
||||
model: model.clone(),
|
||||
},
|
||||
finalize: StartFinalizeConfig {
|
||||
preserve_sandbox: preserve,
|
||||
pr_config,
|
||||
github_app: github_app.clone(),
|
||||
origin_url: origin_url.clone(),
|
||||
model: model.clone(),
|
||||
},
|
||||
},
|
||||
)
|
||||
.await;
|
||||
let run_duration_ms = run_start.elapsed().as_millis() as u64;
|
||||
let mut completion_guard = DetachedRunCompletionGuard::arm(&run_dir);
|
||||
|
||||
|
|
@ -1305,263 +1294,45 @@ async fn run_resumed(
|
|||
let _ = std::env::set_current_dir(cwd);
|
||||
}
|
||||
|
||||
let (final_status, failure_reason, run_status, status_reason) =
|
||||
classify_engine_result(&engine_result);
|
||||
let conclusion = build_conclusion(
|
||||
&run_dir,
|
||||
final_status.clone(),
|
||||
failure_reason,
|
||||
run_duration_ms,
|
||||
last_git_sha.lock().unwrap().clone(),
|
||||
);
|
||||
|
||||
// Auto-derive retro
|
||||
if !args.no_retro && project_config::is_retro_enabled() {
|
||||
let failed = match &engine_result {
|
||||
Ok(ref o) => o.status == StageStatus::Fail,
|
||||
Err(_) => true,
|
||||
};
|
||||
|
||||
let llm_client = if dry_run_mode {
|
||||
None
|
||||
} else {
|
||||
fabro_llm::client::Client::from_env().await.ok()
|
||||
};
|
||||
|
||||
generate_retro(
|
||||
&settings.run_id,
|
||||
&graph.name,
|
||||
graph.goal(),
|
||||
&run_dir,
|
||||
failed,
|
||||
run_duration_ms,
|
||||
dry_run_mode,
|
||||
llm_client.as_ref(),
|
||||
&sandbox,
|
||||
provider_enum,
|
||||
&model,
|
||||
styles,
|
||||
Some(Arc::clone(&emitter)),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
// Finish progress bars after retro (retro stage uses the same ProgressUI)
|
||||
progress_ui.lock().expect("progress lock poisoned").finish();
|
||||
|
||||
// Write finalize commit with retro.json + final node files (captures last diff.patch)
|
||||
write_finalize_commit(&settings, &run_dir).await;
|
||||
|
||||
// Auto-create PR on successful completion (mirrors run_command)
|
||||
let mut pushed_branch: Option<String> = None;
|
||||
let mut pr_url: Option<String> = None;
|
||||
if let Some(pr_cfg) = settings.pull_request() {
|
||||
if settings.dry_run {
|
||||
debug!("Skipping PR creation: dry-run mode");
|
||||
} else if let Err(ref e) = engine_result {
|
||||
debug!(error = %e, "Skipping PR creation: engine returned an error");
|
||||
} else if let Ok(ref outcome) = engine_result {
|
||||
if !matches!(
|
||||
outcome.status,
|
||||
StageStatus::Success | StageStatus::PartialSuccess
|
||||
) {
|
||||
debug!(status = ?outcome.status, "Skipping PR creation: run status is not success");
|
||||
} else {
|
||||
let diff = tokio::fs::read_to_string(run_dir.join("final.patch"))
|
||||
.await
|
||||
.unwrap_or_default();
|
||||
if let (
|
||||
Some(ref base_branch),
|
||||
Some(ref run_branch),
|
||||
Some(ref creds),
|
||||
Some(ref origin),
|
||||
) = (
|
||||
&settings.base_branch,
|
||||
settings.git.as_ref().and_then(|g| g.run_branch.as_ref()),
|
||||
&github_app,
|
||||
&origin_url,
|
||||
) {
|
||||
if settings.git.is_some() {
|
||||
pushed_branch = Some(run_branch.to_string());
|
||||
}
|
||||
|
||||
let auto_merge = if pr_cfg.auto_merge {
|
||||
Some(fabro_workflows::pull_request::AutoMergeConfig {
|
||||
merge_strategy: pr_cfg.merge_strategy,
|
||||
})
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
match fabro_workflows::pull_request::maybe_open_pull_request(
|
||||
creds,
|
||||
origin,
|
||||
base_branch,
|
||||
run_branch,
|
||||
graph.goal(),
|
||||
&diff,
|
||||
&model,
|
||||
pr_cfg.draft,
|
||||
auto_merge,
|
||||
&run_dir,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(Some(record)) => {
|
||||
emitter.emit(
|
||||
&fabro_workflows::event::WorkflowRunEvent::PullRequestCreated {
|
||||
pr_url: record.html_url.clone(),
|
||||
pr_number: record.number,
|
||||
draft: pr_cfg.draft,
|
||||
},
|
||||
);
|
||||
pr_url = Some(record.html_url.clone());
|
||||
if let Err(e) = record.save(&run_dir.join("pull_request.json")) {
|
||||
tracing::warn!(error = %e, "Failed to save pull_request.json");
|
||||
}
|
||||
}
|
||||
Ok(None) => {} // empty diff, logged at DEBUG
|
||||
Err(e) => {
|
||||
emitter.emit(
|
||||
&fabro_workflows::event::WorkflowRunEvent::PullRequestFailed {
|
||||
error: e.to_string(),
|
||||
},
|
||||
);
|
||||
emit_run_notice(
|
||||
&emitter,
|
||||
RunNoticeLevel::Warn,
|
||||
"pull_request_failed",
|
||||
format!("PR creation failed: {e}"),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
let final_status = match started {
|
||||
Ok(started) => {
|
||||
if let Some(ref retro) = started.retro {
|
||||
print_retro_result(retro, started.retro_duration, &run_dir, styles);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
debug!("Skipping PR creation: pull_request not enabled in config");
|
||||
}
|
||||
|
||||
// Defuse the cleanup guard — we are about to do explicit cleanup
|
||||
scopeguard::ScopeGuard::into_inner(cleanup_guard);
|
||||
|
||||
// Cleanup sandbox via engine (fires SandboxCleanup hook)
|
||||
// Before cleanup, print preserve banner (mirrors run_command)
|
||||
if preserve {
|
||||
let info = sandbox.sandbox_info();
|
||||
if !info.is_empty() {
|
||||
emit_run_notice(
|
||||
&emitter,
|
||||
RunNoticeLevel::Info,
|
||||
"sandbox_preserved",
|
||||
format!("sandbox preserved: {info}"),
|
||||
);
|
||||
} else {
|
||||
emit_run_notice(
|
||||
&emitter,
|
||||
RunNoticeLevel::Info,
|
||||
"sandbox_preserved",
|
||||
"sandbox preserved",
|
||||
let finalized = started.finalized;
|
||||
print_run_conclusion(
|
||||
&finalized.conclusion,
|
||||
&run_id,
|
||||
&run_dir,
|
||||
finalized.pushed_branch.as_deref(),
|
||||
finalized.pr_url.as_deref(),
|
||||
styles,
|
||||
);
|
||||
print_final_output(&run_dir, styles);
|
||||
print_assets(&run_dir, styles);
|
||||
finalized.conclusion.status.clone()
|
||||
}
|
||||
Err(err) => {
|
||||
let engine_result: Result<fabro_workflows::outcome::Outcome, _> = Err(err.clone());
|
||||
let (final_status, failure_reason, run_status, status_reason) =
|
||||
classify_engine_result(&engine_result);
|
||||
let conclusion = build_conclusion(
|
||||
&run_dir,
|
||||
final_status.clone(),
|
||||
failure_reason,
|
||||
run_duration_ms,
|
||||
None,
|
||||
);
|
||||
persist_terminal_outcome(&run_dir, &conclusion, run_status, status_reason);
|
||||
print_run_conclusion(&conclusion, &run_id, &run_dir, None, None, styles);
|
||||
print_final_output(&run_dir, styles);
|
||||
print_assets(&run_dir, styles);
|
||||
final_status
|
||||
}
|
||||
}
|
||||
if let Err(e) = engine
|
||||
.cleanup_sandbox(&settings.run_id, &graph.name, preserve)
|
||||
.await
|
||||
{
|
||||
tracing::warn!(error = %e, "Sandbox cleanup failed");
|
||||
emit_run_notice(
|
||||
&emitter,
|
||||
RunNoticeLevel::Warn,
|
||||
"sandbox_cleanup_failed",
|
||||
format!("sandbox cleanup failed: {e}"),
|
||||
);
|
||||
}
|
||||
|
||||
persist_terminal_outcome(&run_dir, &conclusion, run_status, status_reason);
|
||||
completion_guard.defuse();
|
||||
|
||||
eprintln!("\n{}", styles.bold.apply_to("=== Run Result ==="));
|
||||
eprintln!("{}", styles.dim.apply_to(format!("Run: {run_id}")));
|
||||
let status_str = final_status.to_string().to_uppercase();
|
||||
let status_color = match final_status {
|
||||
StageStatus::Success | StageStatus::PartialSuccess => &styles.bold_green,
|
||||
_ => &styles.bold_red,
|
||||
};
|
||||
eprintln!("Status: {}", status_color.apply_to(&status_str));
|
||||
eprintln!(
|
||||
"Duration: {}",
|
||||
HumanDuration(Duration::from_millis(run_duration_ms))
|
||||
);
|
||||
|
||||
{
|
||||
use crate::commands::shared::format_tokens_human;
|
||||
use fabro_workflows::cost::format_cost;
|
||||
let acc = accumulator.lock().unwrap();
|
||||
let total_tokens = acc.total_input_tokens + acc.total_output_tokens;
|
||||
if total_tokens > 0 {
|
||||
if acc.has_pricing {
|
||||
eprintln!(
|
||||
"{}",
|
||||
styles.dim.apply_to(format!(
|
||||
"Cost: {} ({} toks)",
|
||||
format_cost(acc.total_cost),
|
||||
format_tokens_human(total_tokens)
|
||||
))
|
||||
);
|
||||
} else {
|
||||
eprintln!(
|
||||
"{}",
|
||||
styles
|
||||
.dim
|
||||
.apply_to(format!("Toks: {}", format_tokens_human(total_tokens)))
|
||||
);
|
||||
}
|
||||
if acc.total_cache_read_tokens > 0 {
|
||||
eprintln!(
|
||||
"{}",
|
||||
styles.dim.apply_to(format!(
|
||||
"Cache: {} read, {} write",
|
||||
format_tokens_human(acc.total_cache_read_tokens),
|
||||
format_tokens_human(acc.total_cache_write_tokens),
|
||||
)),
|
||||
);
|
||||
}
|
||||
if acc.total_reasoning_tokens > 0 {
|
||||
eprintln!(
|
||||
"{}",
|
||||
styles.dim.apply_to(format!(
|
||||
"Reasoning: {} tokens",
|
||||
format_tokens_human(acc.total_reasoning_tokens),
|
||||
)),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
eprintln!(
|
||||
"{}",
|
||||
styles
|
||||
.dim
|
||||
.apply_to(format!("Run: {}", tilde_path(&run_dir)))
|
||||
);
|
||||
|
||||
if let Some(failure) = conclusion.failure_reason.as_deref() {
|
||||
eprintln!("Failure: {}", styles.red.apply_to(failure));
|
||||
}
|
||||
|
||||
if pushed_branch.is_some() || pr_url.is_some() {
|
||||
eprintln!();
|
||||
if let Some(ref branch) = pushed_branch {
|
||||
eprintln!("{} {branch}", styles.bold.apply_to("Pushed branch:"));
|
||||
}
|
||||
if let Some(ref url) = pr_url {
|
||||
eprintln!("{} {url}", styles.bold.apply_to("Pull request:"));
|
||||
}
|
||||
}
|
||||
|
||||
print_final_output(&run_dir, styles);
|
||||
print_assets(&run_dir, styles);
|
||||
completion_guard.defuse();
|
||||
|
||||
fabro_util::run_log::deactivate();
|
||||
match final_status {
|
||||
|
|
|
|||
|
|
@ -28,22 +28,22 @@ pub struct RewindArgs {
|
|||
|
||||
pub fn run(args: &RewindArgs, styles: &Styles) -> Result<()> {
|
||||
let repo = Repository::discover(".").context("not in a git repository")?;
|
||||
let run_id = fabro_workflows::run_rewind::find_run_id_by_prefix(&repo, &args.run_id)?;
|
||||
let run_id = fabro_workflows::operations::find_run_id_by_prefix(&repo, &args.run_id)?;
|
||||
let store = Store::new(repo);
|
||||
|
||||
let timeline = fabro_workflows::run_rewind::build_timeline(&store, &run_id)?;
|
||||
let timeline = fabro_workflows::operations::build_timeline(&store, &run_id)?;
|
||||
|
||||
if args.list || args.target.is_none() {
|
||||
let parallel_map = fabro_workflows::run_rewind::load_parallel_map(&store, &run_id);
|
||||
let parallel_map = fabro_workflows::operations::load_parallel_map(&store, &run_id);
|
||||
print_timeline(&timeline, ¶llel_map, styles);
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let target = fabro_workflows::run_rewind::parse_target(args.target.as_deref().unwrap())?;
|
||||
let parallel_map = fabro_workflows::run_rewind::load_parallel_map(&store, &run_id);
|
||||
let entry = fabro_workflows::run_rewind::resolve_target(&timeline, &target, ¶llel_map)?;
|
||||
let target = fabro_workflows::operations::parse_target(args.target.as_deref().unwrap())?;
|
||||
let parallel_map = fabro_workflows::operations::load_parallel_map(&store, &run_id);
|
||||
let entry = fabro_workflows::operations::resolve_target(&timeline, &target, ¶llel_map)?;
|
||||
|
||||
fabro_workflows::run_rewind::execute_rewind(&store, &run_id, entry, !args.no_push)?;
|
||||
fabro_workflows::operations::rewind(&store, &run_id, entry, !args.no_push)?;
|
||||
|
||||
eprintln!(
|
||||
"\nTo resume: fabro resume {}",
|
||||
|
|
@ -54,7 +54,7 @@ pub fn run(args: &RewindArgs, styles: &Styles) -> Result<()> {
|
|||
}
|
||||
|
||||
pub(crate) fn print_timeline(
|
||||
timeline: &[fabro_workflows::run_rewind::TimelineEntry],
|
||||
timeline: &[fabro_workflows::operations::TimelineEntry],
|
||||
parallel_map: &std::collections::HashMap<String, String>,
|
||||
styles: &Styles,
|
||||
) {
|
||||
|
|
|
|||
|
|
@ -19,11 +19,17 @@ use fabro_workflows::backend::{AgentApiBackend, AgentCliBackend, BackendRouter};
|
|||
use fabro_workflows::checkpoint::Checkpoint;
|
||||
use fabro_workflows::cost::{compute_stage_cost, format_cost};
|
||||
use fabro_workflows::devcontainer_bridge;
|
||||
use fabro_workflows::engine::{GitCheckpointSettings, RunSettings, WorkflowRunEngine};
|
||||
use fabro_workflows::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent};
|
||||
use fabro_workflows::git::GitSyncStatus;
|
||||
use fabro_workflows::handler::default_registry;
|
||||
use fabro_workflows::operations::{
|
||||
create_from_graph, start, StartFinalizeConfig, StartOptions, StartRetroConfig,
|
||||
};
|
||||
use fabro_workflows::outcome::StageStatus;
|
||||
use fabro_workflows::pipeline::{
|
||||
build_conclusion, classify_engine_result, persist_terminal_outcome,
|
||||
};
|
||||
use fabro_workflows::run_settings::{GitCheckpointSettings, LifecycleConfig, RunSettings};
|
||||
use fabro_workflows::sandbox_provider::SandboxProvider;
|
||||
use indicatif::HumanDuration;
|
||||
use std::time::Duration;
|
||||
|
|
@ -35,10 +41,6 @@ use crate::commands::shared::{
|
|||
format_tokens_human, print_diagnostics, read_workflow_file, relative_path, tilde_path,
|
||||
};
|
||||
|
||||
pub(crate) use fabro_workflows::pipeline::{
|
||||
build_conclusion, classify_engine_result, persist_terminal_outcome, write_finalize_commit,
|
||||
};
|
||||
|
||||
#[derive(Debug, Clone, Copy, ValueEnum)]
|
||||
pub enum CliSandboxProvider {
|
||||
Local,
|
||||
|
|
@ -743,7 +745,7 @@ pub(crate) fn prepare_workflow_with_project_config(
|
|||
|
||||
/// Pre-prepared run state, used to skip workflow preparation in `run_command_impl`.
|
||||
struct RecordBasedRun {
|
||||
graph: fabro_graphviz::graph::Graph,
|
||||
validated: fabro_workflows::pipeline::Validated,
|
||||
raw_source: String,
|
||||
run_cfg: Option<FabroConfig>,
|
||||
sandbox_provider: SandboxProvider,
|
||||
|
|
@ -789,7 +791,7 @@ pub async fn run_from_record(
|
|||
|
||||
let record_run = RecordBasedRun {
|
||||
raw_source: String::new(), // Raw DOT provenance is best-effort for record-based runs
|
||||
graph: record.graph.clone(),
|
||||
validated: create_from_graph(record.graph.clone(), String::new()),
|
||||
run_cfg: Some(record.config.clone()),
|
||||
sandbox_provider,
|
||||
model: model.clone(),
|
||||
|
|
@ -853,10 +855,9 @@ pub async fn run_command(
|
|||
run_defaults,
|
||||
workflow_toml_path,
|
||||
} = prepare_workflow(&args, run_defaults, styles, false)?;
|
||||
let (graph, _source, _diagnostics) = validated.into_parts();
|
||||
|
||||
let record_run = RecordBasedRun {
|
||||
graph,
|
||||
validated,
|
||||
raw_source,
|
||||
run_cfg,
|
||||
sandbox_provider,
|
||||
|
|
@ -878,7 +879,7 @@ async fn run_command_impl(
|
|||
record_run: Option<RecordBasedRun>,
|
||||
) -> anyhow::Result<()> {
|
||||
let (
|
||||
graph,
|
||||
validated,
|
||||
raw_source,
|
||||
mut run_cfg,
|
||||
sandbox_provider,
|
||||
|
|
@ -889,7 +890,7 @@ async fn run_command_impl(
|
|||
workflow_toml_path,
|
||||
) = match record_run {
|
||||
Some(rr) => (
|
||||
rr.graph,
|
||||
rr.validated,
|
||||
rr.raw_source,
|
||||
rr.run_cfg,
|
||||
rr.sandbox_provider,
|
||||
|
|
@ -901,6 +902,7 @@ async fn run_command_impl(
|
|||
),
|
||||
None => unreachable!("run_command_impl always receives a RecordBasedRun"),
|
||||
};
|
||||
let graph = validated.graph().clone();
|
||||
|
||||
// For record-based runs from run_from_record, workflow is None (preparation was skipped).
|
||||
let from_record = args.workflow.is_none();
|
||||
|
|
@ -1048,21 +1050,6 @@ async fn run_command_impl(
|
|||
// 3. Build event emitter
|
||||
let emitter = EventEmitter::new();
|
||||
|
||||
// Track the last git commit SHA from CheckpointCompleted events
|
||||
let last_git_sha: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
|
||||
{
|
||||
let sha_clone = Arc::clone(&last_git_sha);
|
||||
emitter.on_event(move |event| {
|
||||
if let fabro_workflows::event::WorkflowRunEvent::CheckpointCompleted {
|
||||
git_commit_sha: Some(sha),
|
||||
..
|
||||
} = event
|
||||
{
|
||||
*sha_clone.lock().unwrap() = Some(sha.clone());
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// Cost accumulator — shared across all verbosity levels
|
||||
let accumulator = Arc::new(Mutex::new(CostAccumulator::default()));
|
||||
let acc_clone = Arc::clone(&accumulator);
|
||||
|
|
@ -1689,32 +1676,6 @@ async fn run_command_impl(
|
|||
}
|
||||
}
|
||||
});
|
||||
let mut engine = WorkflowRunEngine::with_interviewer(
|
||||
registry,
|
||||
Arc::clone(&emitter),
|
||||
interviewer,
|
||||
Arc::clone(&sandbox),
|
||||
);
|
||||
if !sandbox_env.is_empty() {
|
||||
engine.set_env(sandbox_env);
|
||||
}
|
||||
if dry_run_mode {
|
||||
engine.set_dry_run(true);
|
||||
}
|
||||
// Wire up hook runner from run config or run defaults
|
||||
{
|
||||
let hooks = run_cfg
|
||||
.as_ref()
|
||||
.map(|c| &c.hooks)
|
||||
.unwrap_or(&run_defaults.hooks);
|
||||
if !hooks.is_empty() {
|
||||
let hook_config = fabro_hooks::HookConfig {
|
||||
hooks: hooks.clone(),
|
||||
};
|
||||
let runner = fabro_hooks::HookRunner::new(hook_config);
|
||||
engine.set_hook_runner(Arc::new(runner));
|
||||
}
|
||||
}
|
||||
|
||||
// 7. Execute
|
||||
// Set up metadata branch for git checkpointing (host or remote — engine fills remote)
|
||||
|
|
@ -1728,7 +1689,7 @@ async fn run_command_impl(
|
|||
None
|
||||
};
|
||||
|
||||
let mut config = RunSettings {
|
||||
let config = RunSettings {
|
||||
config: settings_config,
|
||||
run_dir: run_dir.clone(),
|
||||
cancel_token: None,
|
||||
|
|
@ -1754,7 +1715,7 @@ async fn run_command_impl(
|
|||
};
|
||||
|
||||
// Build lifecycle config for sandbox init, setup commands, and devcontainer phases
|
||||
let lifecycle = fabro_workflows::engine::LifecycleConfig {
|
||||
let lifecycle = LifecycleConfig {
|
||||
setup_commands,
|
||||
setup_command_timeout_ms: 300_000,
|
||||
devcontainer_phases: if let Some(ref dc) = devcontainer_config {
|
||||
|
|
@ -1771,287 +1732,97 @@ async fn run_command_impl(
|
|||
// Defuse the bootstrap guard — engine.run() has taken ownership of lifecycle status.
|
||||
status_guard.defuse();
|
||||
|
||||
// Safety net: if we panic or return early, best-effort cleanup via spawn.
|
||||
let sandbox_for_cleanup = Arc::clone(&sandbox);
|
||||
let cleanup_guard = scopeguard::guard((), move |()| {
|
||||
if preserve_sandbox {
|
||||
return;
|
||||
}
|
||||
let rt = tokio::runtime::Handle::try_current();
|
||||
if let Ok(handle) = rt {
|
||||
handle.spawn(async move {
|
||||
let _ = sandbox_for_cleanup.cleanup().await;
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
let run_start = Instant::now();
|
||||
let engine_result = engine
|
||||
.run_with_lifecycle(&graph, &mut config, lifecycle, None)
|
||||
.await;
|
||||
let pr_config = config.pull_request().cloned();
|
||||
let started = start(
|
||||
validated,
|
||||
StartOptions {
|
||||
init: fabro_workflows::pipeline::InitOptions {
|
||||
run_id: run_id.clone(),
|
||||
run_dir: run_dir.clone(),
|
||||
dry_run: dry_run_mode,
|
||||
emitter: Arc::clone(&emitter),
|
||||
sandbox: Arc::clone(&sandbox),
|
||||
registry: Arc::new(registry),
|
||||
lifecycle,
|
||||
run_settings: config,
|
||||
hooks: fabro_hooks::HookConfig {
|
||||
hooks: run_cfg
|
||||
.as_ref()
|
||||
.map(|c| c.hooks.clone())
|
||||
.unwrap_or_else(|| run_defaults.hooks.clone()),
|
||||
},
|
||||
sandbox_env,
|
||||
checkpoint: None,
|
||||
seed_context: None,
|
||||
},
|
||||
retro: StartRetroConfig {
|
||||
enabled: !no_retro_flag && project_config::is_retro_enabled(),
|
||||
dry_run: dry_run_mode,
|
||||
llm_client: llm_client.clone(),
|
||||
provider: provider_enum,
|
||||
model: model.clone(),
|
||||
},
|
||||
finalize: StartFinalizeConfig {
|
||||
preserve_sandbox,
|
||||
pr_config,
|
||||
github_app: github_app.clone(),
|
||||
origin_url: origin_url.clone(),
|
||||
model: model.clone(),
|
||||
},
|
||||
},
|
||||
)
|
||||
.await;
|
||||
let run_duration_ms = run_start.elapsed().as_millis() as u64;
|
||||
let mut completion_guard = DetachedRunCompletionGuard::arm(&run_dir);
|
||||
|
||||
// Restore cwd (worktree is kept for `fabro cp` access; pruned separately)
|
||||
let _ = std::env::set_current_dir(&original_cwd);
|
||||
|
||||
let (final_status, failure_reason, run_status, status_reason) =
|
||||
classify_engine_result(&engine_result);
|
||||
let conclusion = build_conclusion(
|
||||
&run_dir,
|
||||
final_status.clone(),
|
||||
failure_reason,
|
||||
run_duration_ms,
|
||||
last_git_sha.lock().unwrap().clone(),
|
||||
);
|
||||
|
||||
// Auto-derive retro (always, cheap) and optionally run retro agent
|
||||
if !no_retro_flag && project_config::is_retro_enabled() {
|
||||
let failed = match &engine_result {
|
||||
Ok(ref o) => o.status == StageStatus::Fail,
|
||||
Err(_) => true,
|
||||
};
|
||||
generate_retro(
|
||||
&config.run_id,
|
||||
&graph.name,
|
||||
graph.goal(),
|
||||
&run_dir,
|
||||
failed,
|
||||
run_duration_ms,
|
||||
dry_run_mode,
|
||||
llm_client.as_ref(),
|
||||
&sandbox,
|
||||
provider_enum,
|
||||
&model,
|
||||
styles,
|
||||
Some(Arc::clone(&emitter)),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
// Finish progress bars after retro (retro stage uses the same ProgressUI)
|
||||
progress_ui.lock().expect("progress lock poisoned").finish();
|
||||
|
||||
// Write finalize commit with retro.json + final node files (captures last diff.patch)
|
||||
write_finalize_commit(&config, &run_dir).await;
|
||||
|
||||
// Auto-create PR on successful completion (skip in dry-run mode)
|
||||
let mut pushed_branch: Option<String> = None;
|
||||
let mut pr_url: Option<String> = None;
|
||||
if let Some(pr_cfg) = config.pull_request() {
|
||||
if dry_run_mode {
|
||||
debug!("Skipping PR creation: dry-run mode");
|
||||
} else if let Err(ref e) = engine_result {
|
||||
debug!(error = %e, "Skipping PR creation: engine returned an error");
|
||||
} else if let Ok(ref outcome) = engine_result {
|
||||
if !matches!(
|
||||
outcome.status,
|
||||
StageStatus::Success | StageStatus::PartialSuccess
|
||||
) {
|
||||
debug!(status = ?outcome.status, "Skipping PR creation: run status is not success");
|
||||
} else {
|
||||
let diff = tokio::fs::read_to_string(run_dir.join("final.patch"))
|
||||
.await
|
||||
.unwrap_or_default();
|
||||
if let (
|
||||
Some(ref base_branch),
|
||||
Some(ref run_branch),
|
||||
Some(ref creds),
|
||||
Some(ref origin),
|
||||
) = (
|
||||
&config.base_branch,
|
||||
config.git.as_ref().and_then(|g| g.run_branch.as_ref()),
|
||||
&github_app,
|
||||
&origin_url,
|
||||
) {
|
||||
// Run branch was pushed during checkpoint commits;
|
||||
// just record it for the PR creation.
|
||||
if config.git.is_some() {
|
||||
pushed_branch = Some(run_branch.to_string());
|
||||
}
|
||||
|
||||
let auto_merge = if pr_cfg.auto_merge {
|
||||
Some(fabro_workflows::pull_request::AutoMergeConfig {
|
||||
merge_strategy: pr_cfg.merge_strategy,
|
||||
})
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
match fabro_workflows::pull_request::maybe_open_pull_request(
|
||||
creds,
|
||||
origin,
|
||||
base_branch,
|
||||
run_branch,
|
||||
graph.goal(),
|
||||
&diff,
|
||||
&model,
|
||||
pr_cfg.draft,
|
||||
auto_merge,
|
||||
&run_dir,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(Some(record)) => {
|
||||
emitter.emit(
|
||||
&fabro_workflows::event::WorkflowRunEvent::PullRequestCreated {
|
||||
pr_url: record.html_url.clone(),
|
||||
pr_number: record.number,
|
||||
draft: pr_cfg.draft,
|
||||
},
|
||||
);
|
||||
pr_url = Some(record.html_url.clone());
|
||||
if let Err(e) = record.save(&run_dir.join("pull_request.json")) {
|
||||
tracing::warn!(error = %e, "Failed to save pull_request.json");
|
||||
}
|
||||
}
|
||||
Ok(None) => {} // empty diff, logged at DEBUG
|
||||
Err(e) => {
|
||||
emitter.emit(
|
||||
&fabro_workflows::event::WorkflowRunEvent::PullRequestFailed {
|
||||
error: e.to_string(),
|
||||
},
|
||||
);
|
||||
emit_run_notice(
|
||||
&emitter,
|
||||
RunNoticeLevel::Warn,
|
||||
"pull_request_failed",
|
||||
format!("PR creation failed: {e}"),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
let final_status = match started {
|
||||
Ok(started) => {
|
||||
if let Some(ref retro) = started.retro {
|
||||
print_retro_result(retro, started.retro_duration, &run_dir, styles);
|
||||
}
|
||||
let finalized = started.finalized;
|
||||
print_run_conclusion(
|
||||
&finalized.conclusion,
|
||||
&run_id,
|
||||
&run_dir,
|
||||
finalized.pushed_branch.as_deref(),
|
||||
finalized.pr_url.as_deref(),
|
||||
styles,
|
||||
);
|
||||
print_final_output(&run_dir, styles);
|
||||
print_assets(&run_dir, styles);
|
||||
finalized.conclusion.status.clone()
|
||||
}
|
||||
Err(err) => {
|
||||
let engine_result = Err(err.clone());
|
||||
let (final_status, failure_reason, run_status, status_reason) =
|
||||
classify_engine_result(&engine_result);
|
||||
let conclusion = build_conclusion(
|
||||
&run_dir,
|
||||
final_status.clone(),
|
||||
failure_reason,
|
||||
run_duration_ms,
|
||||
None,
|
||||
);
|
||||
persist_terminal_outcome(&run_dir, &conclusion, run_status, status_reason);
|
||||
print_run_conclusion(&conclusion, &run_id, &run_dir, None, None, styles);
|
||||
print_final_output(&run_dir, styles);
|
||||
print_assets(&run_dir, styles);
|
||||
final_status
|
||||
}
|
||||
} else {
|
||||
debug!("Skipping PR creation: pull_request not enabled in config");
|
||||
}
|
||||
|
||||
// 8. Print result
|
||||
eprintln!("\n{}", styles.bold.apply_to("=== Run Result ==="),);
|
||||
|
||||
eprintln!("{}", styles.dim.apply_to(format!("Run: {run_id}")));
|
||||
let status_str = final_status.to_string().to_uppercase();
|
||||
let status_color = match final_status {
|
||||
StageStatus::Success | StageStatus::PartialSuccess => &styles.bold_green,
|
||||
_ => &styles.bold_red,
|
||||
};
|
||||
eprintln!("Status: {}", status_color.apply_to(&status_str),);
|
||||
eprintln!(
|
||||
"Duration: {}",
|
||||
HumanDuration(Duration::from_millis(run_duration_ms))
|
||||
);
|
||||
|
||||
{
|
||||
let acc = accumulator.lock().unwrap();
|
||||
let total_tokens = acc.total_input_tokens + acc.total_output_tokens;
|
||||
if total_tokens > 0 {
|
||||
if acc.has_pricing {
|
||||
eprintln!(
|
||||
"{}",
|
||||
styles.dim.apply_to(format!(
|
||||
"Cost: {} ({} toks)",
|
||||
format_cost(acc.total_cost),
|
||||
format_tokens_human(total_tokens)
|
||||
))
|
||||
);
|
||||
} else {
|
||||
eprintln!(
|
||||
"{}",
|
||||
styles
|
||||
.dim
|
||||
.apply_to(format!("Toks: {}", format_tokens_human(total_tokens)))
|
||||
);
|
||||
}
|
||||
if acc.total_cache_read_tokens > 0 {
|
||||
eprintln!(
|
||||
"{}",
|
||||
styles.dim.apply_to(format!(
|
||||
"Cache: {} read, {} write",
|
||||
format_tokens_human(acc.total_cache_read_tokens),
|
||||
format_tokens_human(acc.total_cache_write_tokens),
|
||||
)),
|
||||
);
|
||||
}
|
||||
if acc.total_reasoning_tokens > 0 {
|
||||
eprintln!(
|
||||
"{}",
|
||||
styles.dim.apply_to(format!(
|
||||
"Reasoning: {} tokens",
|
||||
format_tokens_human(acc.total_reasoning_tokens),
|
||||
)),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
eprintln!(
|
||||
"{}",
|
||||
styles
|
||||
.dim
|
||||
.apply_to(format!("Run: {}", tilde_path(&run_dir)))
|
||||
);
|
||||
|
||||
if let Some(failure) = conclusion.failure_reason.as_deref() {
|
||||
eprintln!("Failure: {}", styles.red.apply_to(failure));
|
||||
}
|
||||
|
||||
if pushed_branch.is_some() || pr_url.is_some() {
|
||||
eprintln!();
|
||||
if let Some(ref branch) = pushed_branch {
|
||||
eprintln!("{} {branch}", styles.bold.apply_to("Pushed branch:"));
|
||||
}
|
||||
if let Some(ref url) = pr_url {
|
||||
eprintln!("{} {url}", styles.bold.apply_to("Pull request:"));
|
||||
}
|
||||
}
|
||||
|
||||
print_final_output(&run_dir, styles);
|
||||
print_assets(&run_dir, styles);
|
||||
|
||||
// 9. Cleanup sandbox (defuse the scopeguard so we await properly)
|
||||
scopeguard::ScopeGuard::into_inner(cleanup_guard);
|
||||
if preserve_sandbox {
|
||||
let info = sandbox.sandbox_info();
|
||||
if !info.is_empty() {
|
||||
emit_run_notice(
|
||||
&emitter,
|
||||
RunNoticeLevel::Info,
|
||||
"sandbox_preserved",
|
||||
format!("sandbox preserved: {info}"),
|
||||
);
|
||||
} else {
|
||||
emit_run_notice(
|
||||
&emitter,
|
||||
RunNoticeLevel::Info,
|
||||
"sandbox_preserved",
|
||||
"sandbox preserved",
|
||||
);
|
||||
}
|
||||
}
|
||||
if let Err(e) = engine
|
||||
.cleanup_sandbox(&run_id, &graph.name, preserve_sandbox)
|
||||
.await
|
||||
{
|
||||
tracing::warn!(error = %e, "Sandbox cleanup failed");
|
||||
emit_run_notice(
|
||||
&emitter,
|
||||
RunNoticeLevel::Warn,
|
||||
"sandbox_cleanup_failed",
|
||||
format!("sandbox cleanup failed: {e}"),
|
||||
);
|
||||
}
|
||||
|
||||
persist_terminal_outcome(&run_dir, &conclusion, run_status, status_reason);
|
||||
completion_guard.defuse();
|
||||
|
||||
// 10. Exit code
|
||||
fabro_util::run_log::deactivate();
|
||||
match final_status {
|
||||
StageStatus::Success | StageStatus::PartialSuccess => Ok(()),
|
||||
_ => {
|
||||
std::process::exit(1);
|
||||
}
|
||||
_ => std::process::exit(1),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -2078,6 +1849,36 @@ pub fn print_run_summary(run_dir: &Path, run_id: &str, styles: &Styles) {
|
|||
return;
|
||||
};
|
||||
|
||||
// PR info from pull_request.json (saved by _run_engine)
|
||||
let pr_url = std::fs::read_to_string(run_dir.join("pull_request.json"))
|
||||
.ok()
|
||||
.and_then(|content| {
|
||||
serde_json::from_str::<fabro_workflows::pull_request::PullRequestRecord>(&content)
|
||||
.ok()
|
||||
.map(|record| record.html_url)
|
||||
});
|
||||
|
||||
print_run_conclusion(
|
||||
&conclusion,
|
||||
run_id,
|
||||
run_dir,
|
||||
None,
|
||||
pr_url.as_deref(),
|
||||
styles,
|
||||
);
|
||||
|
||||
print_final_output(run_dir, styles);
|
||||
print_assets(run_dir, styles);
|
||||
}
|
||||
|
||||
pub(crate) fn print_run_conclusion(
|
||||
conclusion: &fabro_workflows::conclusion::Conclusion,
|
||||
run_id: &str,
|
||||
run_dir: &Path,
|
||||
pushed_branch: Option<&str>,
|
||||
pr_url: Option<&str>,
|
||||
styles: &Styles,
|
||||
) {
|
||||
eprintln!("\n{}", styles.bold.apply_to("=== Run Result ==="));
|
||||
eprintln!("{}", styles.dim.apply_to(format!("Run: {run_id}")));
|
||||
|
||||
|
|
@ -2147,22 +1948,74 @@ pub fn print_run_summary(run_dir: &Path, run_id: &str, styles: &Styles) {
|
|||
eprintln!("Failure: {}", styles.red.apply_to(failure));
|
||||
}
|
||||
|
||||
// PR info from pull_request.json (saved by _run_engine)
|
||||
if let Ok(content) = std::fs::read_to_string(run_dir.join("pull_request.json")) {
|
||||
if let Ok(record) =
|
||||
serde_json::from_str::<fabro_workflows::pull_request::PullRequestRecord>(&content)
|
||||
{
|
||||
eprintln!();
|
||||
eprintln!(
|
||||
"{} {}",
|
||||
styles.bold.apply_to("Pull request:"),
|
||||
record.html_url
|
||||
);
|
||||
if pushed_branch.is_some() || pr_url.is_some() {
|
||||
eprintln!();
|
||||
if let Some(branch) = pushed_branch {
|
||||
eprintln!("{} {branch}", styles.bold.apply_to("Pushed branch:"));
|
||||
}
|
||||
if let Some(url) = pr_url {
|
||||
eprintln!("{} {url}", styles.bold.apply_to("Pull request:"));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
print_final_output(run_dir, styles);
|
||||
print_assets(run_dir, styles);
|
||||
pub(crate) fn print_retro_result(
|
||||
retro: &fabro_retro::retro::Retro,
|
||||
duration: Duration,
|
||||
run_dir: &Path,
|
||||
styles: &Styles,
|
||||
) {
|
||||
eprintln!("\n{}", styles.bold.apply_to("=== Retro ==="));
|
||||
|
||||
let retro_dur = run_progress::format_duration_short(duration);
|
||||
let smoothness_str = retro
|
||||
.smoothness
|
||||
.as_ref()
|
||||
.map(|s| s.to_string())
|
||||
.unwrap_or_else(|| "unknown".to_string());
|
||||
let outcome_str = retro.outcome.as_deref().unwrap_or("No outcome recorded");
|
||||
let line1_content = format!("Retro: {smoothness_str} - {outcome_str}");
|
||||
let term_width = console::Term::stderr().size().1 as usize;
|
||||
let pad1 = term_width.saturating_sub(line1_content.len() + retro_dur.len());
|
||||
eprintln!(
|
||||
"{} {}{:pad1$}{}",
|
||||
styles.bold.apply_to("Retro:"),
|
||||
styles
|
||||
.dim
|
||||
.apply_to(format!("{smoothness_str} - {outcome_str}")),
|
||||
"",
|
||||
styles.dim.apply_to(&retro_dur),
|
||||
);
|
||||
|
||||
let friction_count = retro.friction_points.as_ref().map(|v| v.len()).unwrap_or(0);
|
||||
let open_count = retro.open_items.as_ref().map(|v| v.len()).unwrap_or(0);
|
||||
if friction_count > 0 || open_count > 0 {
|
||||
let mut parts = Vec::new();
|
||||
if friction_count > 0 {
|
||||
let noun = if friction_count == 1 {
|
||||
"friction point"
|
||||
} else {
|
||||
"friction points"
|
||||
};
|
||||
parts.push(format!("{friction_count} {noun}"));
|
||||
}
|
||||
if open_count > 0 {
|
||||
let noun = if open_count == 1 {
|
||||
"open item"
|
||||
} else {
|
||||
"open items"
|
||||
};
|
||||
parts.push(format!("{open_count} {noun}"));
|
||||
}
|
||||
eprintln!(" {}", styles.dim.apply_to(parts.join(" · ")));
|
||||
}
|
||||
|
||||
let retro_path = format!("{}/retro.json", tilde_path(run_dir));
|
||||
eprintln!(
|
||||
" {} {}",
|
||||
styles.dim.apply_to("Retro saved to"),
|
||||
styles.underline.apply_to(&retro_path),
|
||||
);
|
||||
}
|
||||
|
||||
/// Print the final stage output from the checkpoint, if available.
|
||||
|
|
@ -2557,108 +2410,6 @@ async fn run_preflight(
|
|||
}
|
||||
}
|
||||
|
||||
/// Generate a retro report for a completed workflow run.
|
||||
///
|
||||
/// Derives a basic retro from the checkpoint, then optionally runs the retro agent
|
||||
/// for a richer narrative. Errors are logged as warnings rather than propagated.
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub(crate) async fn generate_retro(
|
||||
run_id: &str,
|
||||
workflow_name: &str,
|
||||
goal: &str,
|
||||
run_dir: &std::path::Path,
|
||||
failed: bool,
|
||||
run_duration_ms: u64,
|
||||
dry_run_mode: bool,
|
||||
llm_client: Option<&fabro_llm::client::Client>,
|
||||
sandbox: &Arc<dyn fabro_agent::Sandbox>,
|
||||
provider_enum: Provider,
|
||||
model: &str,
|
||||
styles: &'static Styles,
|
||||
emitter: Option<Arc<EventEmitter>>,
|
||||
) {
|
||||
eprintln!("\n{}", styles.bold.apply_to("=== Retro ==="));
|
||||
if emitter.is_none() {
|
||||
eprintln!(
|
||||
"{}",
|
||||
styles.dim.apply_to(format!("Running retro ({model})..."))
|
||||
);
|
||||
}
|
||||
|
||||
let retro_start = std::time::Instant::now();
|
||||
let retro = fabro_workflows::pipeline::run_retro(&fabro_workflows::pipeline::RetroOptions {
|
||||
run_id: run_id.to_string(),
|
||||
workflow_name: workflow_name.to_string(),
|
||||
goal: goal.to_string(),
|
||||
run_dir: run_dir.to_path_buf(),
|
||||
sandbox: Arc::clone(sandbox),
|
||||
emitter,
|
||||
failed,
|
||||
run_duration_ms,
|
||||
enabled: true,
|
||||
dry_run: dry_run_mode,
|
||||
llm_client: llm_client.cloned(),
|
||||
provider: provider_enum,
|
||||
model: model.to_string(),
|
||||
})
|
||||
.await;
|
||||
|
||||
let retro_dur = run_progress::format_duration_short(retro_start.elapsed());
|
||||
if let Some(retro) = retro {
|
||||
let smoothness_str = retro
|
||||
.smoothness
|
||||
.as_ref()
|
||||
.map(|s| s.to_string())
|
||||
.unwrap_or_else(|| "unknown".to_string());
|
||||
let outcome_str = retro.outcome.as_deref().unwrap_or("No outcome recorded");
|
||||
let line1_content = format!("Retro: {smoothness_str} \u{2014} {outcome_str}");
|
||||
let term_width = console::Term::stderr().size().1 as usize;
|
||||
let dur_len = retro_dur.len();
|
||||
let pad1 = term_width.saturating_sub(line1_content.len() + dur_len);
|
||||
eprintln!(
|
||||
"{} {}{:pad1$}{}",
|
||||
styles.bold.apply_to("Retro:"),
|
||||
styles
|
||||
.dim
|
||||
.apply_to(format!("{smoothness_str} \u{2014} {outcome_str}")),
|
||||
"",
|
||||
styles.dim.apply_to(&retro_dur),
|
||||
);
|
||||
|
||||
let friction_count = retro.friction_points.as_ref().map(|v| v.len()).unwrap_or(0);
|
||||
let open_count = retro.open_items.as_ref().map(|v| v.len()).unwrap_or(0);
|
||||
if friction_count > 0 || open_count > 0 {
|
||||
let mut parts = Vec::new();
|
||||
if friction_count > 0 {
|
||||
let noun = if friction_count == 1 {
|
||||
"friction point"
|
||||
} else {
|
||||
"friction points"
|
||||
};
|
||||
parts.push(format!("{friction_count} {noun}"));
|
||||
}
|
||||
if open_count > 0 {
|
||||
let noun = if open_count == 1 {
|
||||
"open item"
|
||||
} else {
|
||||
"open items"
|
||||
};
|
||||
parts.push(format!("{open_count} {noun}"));
|
||||
}
|
||||
eprintln!(" {}", styles.dim.apply_to(parts.join(" \u{00b7} ")));
|
||||
}
|
||||
|
||||
let retro_path = format!("{}/retro.json", tilde_path(run_dir));
|
||||
eprintln!(
|
||||
" {} {}",
|
||||
styles.dim.apply_to("Retro saved to"),
|
||||
styles.underline.apply_to(&retro_path),
|
||||
);
|
||||
} else {
|
||||
eprintln!("{}", styles.dim.apply_to("Retro unavailable"));
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn build_event_envelope(
|
||||
event: &fabro_workflows::event::WorkflowRunEvent,
|
||||
run_id: &str,
|
||||
|
|
|
|||
|
|
@ -16,7 +16,9 @@ pub struct ValidateArgs {
|
|||
pub fn run(args: &ValidateArgs, styles: &Styles) -> anyhow::Result<()> {
|
||||
let (dot_path, _cfg) = fabro_config::project::resolve_workflow(&args.workflow)?;
|
||||
|
||||
let (graph, diagnostics) = fabro_workflows::workflow::prepare_from_file(&dot_path)?;
|
||||
let validated = fabro_workflows::operations::create_from_file(&dot_path)?;
|
||||
let graph = validated.graph();
|
||||
let diagnostics = validated.diagnostics();
|
||||
|
||||
eprintln!(
|
||||
"{} ({} nodes, {} edges)",
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ use fabro_core::graph::{EdgeSelection, EdgeSpec, Graph, NodeSpec};
|
|||
use fabro_graphviz::graph::types::{Edge as GvEdge, Graph as GvGraph, Node as GvNode};
|
||||
|
||||
use crate::context::Context;
|
||||
use crate::engine;
|
||||
use crate::graph_ops;
|
||||
use crate::outcome::{Outcome, StageUsage};
|
||||
|
||||
// ---- WorkflowNode ----
|
||||
|
|
@ -26,7 +26,7 @@ impl NodeSpec for WorkflowNode {
|
|||
}
|
||||
|
||||
fn is_terminal(&self) -> bool {
|
||||
engine::is_terminal(&self.0)
|
||||
graph_ops::is_terminal(&self.0)
|
||||
}
|
||||
|
||||
fn max_visits(&self) -> Option<usize> {
|
||||
|
|
@ -103,7 +103,7 @@ impl Graph for WorkflowGraph {
|
|||
outcome: &Outcome,
|
||||
context: &Context,
|
||||
) -> Option<EdgeSelection<Self>> {
|
||||
let selection = engine::select_edge(
|
||||
let selection = graph_ops::select_edge(
|
||||
node.inner(),
|
||||
outcome,
|
||||
context,
|
||||
|
|
@ -120,10 +120,10 @@ impl Graph for WorkflowGraph {
|
|||
&self,
|
||||
outcomes: &HashMap<String, Outcome>,
|
||||
) -> std::result::Result<(), String> {
|
||||
engine::check_goal_gates(self.inner(), outcomes)
|
||||
graph_ops::check_goal_gates(self.inner(), outcomes)
|
||||
}
|
||||
|
||||
fn get_retry_target(&self, failed_node_id: &str) -> Option<String> {
|
||||
engine::get_retry_target(failed_node_id, self.inner())
|
||||
graph_ops::get_retry_target(failed_node_id, self.inner())
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -14,9 +14,9 @@ use crate::context::Context;
|
|||
|
||||
use super::graph::WorkflowGraph;
|
||||
use super::WorkflowNode;
|
||||
use crate::engine;
|
||||
use crate::handler::{format_panic_message, EngineServices};
|
||||
use crate::outcome::{Outcome, StageStatus};
|
||||
use crate::{graph_ops, run_dir};
|
||||
|
||||
/// Production node handler that bridges fabro-core's NodeHandler to the
|
||||
/// existing fabro-workflows Handler trait via EngineServices.
|
||||
|
|
@ -98,7 +98,7 @@ impl NodeHandler<WorkflowGraph> for WorkflowNodeHandler {
|
|||
Err(panic_payload) => {
|
||||
let msg = format_panic_message(panic_payload);
|
||||
let visit = context.node_visit_count().max(1);
|
||||
let panic_dir = crate::engine::node_dir(&self.run_dir, &gv_node.id, visit);
|
||||
let panic_dir = run_dir::node_dir(&self.run_dir, &gv_node.id, visit);
|
||||
let _ = std::fs::create_dir_all(&panic_dir);
|
||||
let _ = std::fs::write(panic_dir.join("panic.txt"), &msg);
|
||||
Err(CoreError::handler(HandlerErrorDetail {
|
||||
|
|
@ -113,7 +113,7 @@ impl NodeHandler<WorkflowGraph> for WorkflowNodeHandler {
|
|||
|
||||
fn retry_policy(&self, node: &WorkflowNode, _graph: &WorkflowGraph) -> CoreRetryPolicy {
|
||||
let gv_node = node.inner();
|
||||
let wf_policy = engine::build_retry_policy(gv_node, &self.graph);
|
||||
let wf_policy = graph_ops::build_retry_policy(gv_node, &self.graph);
|
||||
CoreRetryPolicy {
|
||||
max_attempts: wf_policy.max_attempts,
|
||||
backoff: wf_policy.backoff,
|
||||
|
|
|
|||
|
|
@ -10,8 +10,8 @@ use fabro_core::state::RunState;
|
|||
|
||||
use super::super::graph::WorkflowGraph;
|
||||
use super::super::WorkflowNode;
|
||||
use crate::engine;
|
||||
use crate::error::{FailureCategory, FailureSignature};
|
||||
use crate::graph_ops::classify_outcome;
|
||||
use crate::outcome::{OutcomeExt, StageStatus, StageUsage};
|
||||
|
||||
type WfRunState = RunState<Option<StageUsage>>;
|
||||
|
|
@ -69,7 +69,7 @@ impl RunLifecycle<WorkflowGraph> for CircuitBreakerLifecycle {
|
|||
let outcome = &result.outcome;
|
||||
|
||||
let outcome_failure_category = if outcome.status == StageStatus::Fail {
|
||||
engine::classify_outcome(outcome)
|
||||
classify_outcome(outcome)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
|
@ -117,7 +117,7 @@ impl RunLifecycle<WorkflowGraph> for CircuitBreakerLifecycle {
|
|||
let outcome = ctx.outcome;
|
||||
|
||||
// Guard: only TransientInfra failures may trigger loop_restart
|
||||
let failure_class = engine::classify_outcome(outcome);
|
||||
let failure_class = classify_outcome(outcome);
|
||||
if let Some(fc) = failure_class {
|
||||
if fc != FailureCategory::TransientInfra {
|
||||
return Ok(EdgeDecision::Block(format!(
|
||||
|
|
|
|||
|
|
@ -12,9 +12,10 @@ use super::super::graph::WorkflowGraph;
|
|||
use super::super::WorkflowNode;
|
||||
use super::circuit_breaker::CircuitBreakerLifecycle;
|
||||
use crate::checkpoint::Checkpoint;
|
||||
use crate::engine::{self, RunSettings};
|
||||
use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent};
|
||||
use crate::outcome::StageUsage;
|
||||
use crate::run_dir::{write_node_status, write_start_record};
|
||||
use crate::run_settings::RunSettings;
|
||||
|
||||
type WfRunState = RunState<Option<StageUsage>>;
|
||||
type WfNodeResult = NodeResult<Option<StageUsage>>;
|
||||
|
|
@ -38,7 +39,7 @@ impl RunLifecycle<WorkflowGraph> for DiskLifecycle {
|
|||
_state: &WfRunState,
|
||||
) -> fabro_core::error::Result<()> {
|
||||
// Write start.json
|
||||
engine::write_start_record(&self.run_dir, &self.config);
|
||||
write_start_record(&self.run_dir, &self.config);
|
||||
// Write run status as Running
|
||||
crate::run_status::write_run_status(
|
||||
&self.run_dir,
|
||||
|
|
@ -56,7 +57,7 @@ impl RunLifecycle<WorkflowGraph> for DiskLifecycle {
|
|||
) -> fabro_core::error::Result<()> {
|
||||
let gv = node.inner();
|
||||
let visit = state.node_visits.get(gv.id.as_str()).copied().unwrap_or(1);
|
||||
engine::write_node_status(&self.run_dir, &gv.id, visit, &result.outcome);
|
||||
write_node_status(&self.run_dir, &gv.id, visit, &result.outcome);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -12,8 +12,8 @@ use super::super::graph::WorkflowGraph;
|
|||
use super::super::WorkflowNode;
|
||||
use super::git::GitCheckpointResult;
|
||||
use crate::artifact::ArtifactStore;
|
||||
use crate::engine;
|
||||
use crate::event::{EventEmitter, WorkflowRunEvent};
|
||||
use crate::graph_ops::node_script;
|
||||
use crate::outcome::{FailureCategory, FailureDetail, Outcome, StageStatus, StageUsage};
|
||||
|
||||
type WfRunState = RunState<Option<StageUsage>>;
|
||||
|
|
@ -87,7 +87,7 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
|
|||
name: gv.label().to_string(),
|
||||
index: stage_index,
|
||||
handler_type: gv.handler_type().map(String::from),
|
||||
script: engine::node_script(gv),
|
||||
script: node_script(gv),
|
||||
attempt: 1,
|
||||
max_attempts: 1,
|
||||
});
|
||||
|
|
@ -119,7 +119,7 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
|
|||
name: gv.label().to_string(),
|
||||
index: state.stage_index,
|
||||
handler_type: gv.handler_type().map(String::from),
|
||||
script: engine::node_script(gv),
|
||||
script: node_script(gv),
|
||||
attempt: ctx.attempt as usize,
|
||||
max_attempts: ctx.max_attempts as usize,
|
||||
});
|
||||
|
|
|
|||
|
|
@ -9,7 +9,7 @@ use fabro_core::state::RunState;
|
|||
use super::super::graph::WorkflowGraph;
|
||||
use super::super::WorkflowNode;
|
||||
use crate::context::keys;
|
||||
use crate::engine;
|
||||
use crate::graph_ops::{resolve_fidelity, resolve_thread_id};
|
||||
use crate::outcome::StageUsage;
|
||||
use crate::preamble::build_preamble;
|
||||
|
||||
|
|
@ -67,7 +67,7 @@ impl RunLifecycle<WorkflowGraph> for FidelityLifecycle {
|
|||
|
||||
// 1. Fidelity resolution via resolve_fidelity: edge → node → graph default → Compact
|
||||
let incoming_edge_ref = incoming.as_ref().map(|d| d.edge.as_ref());
|
||||
let fidelity = engine::resolve_fidelity(incoming_edge_ref, gv_node, &self.graph);
|
||||
let fidelity = resolve_fidelity(incoming_edge_ref, gv_node, &self.graph);
|
||||
|
||||
// 2. Fidelity degradation on resume (full → summary:high)
|
||||
let fidelity = {
|
||||
|
|
@ -99,7 +99,7 @@ impl RunLifecycle<WorkflowGraph> for FidelityLifecycle {
|
|||
.set(keys::CURRENT_PREAMBLE, serde_json::json!(preamble));
|
||||
|
||||
// 5. Thread ID resolution via resolve_thread_id: edge → node → graph default → class → previous
|
||||
let thread_id = engine::resolve_thread_id(
|
||||
let thread_id = resolve_thread_id(
|
||||
incoming_edge_ref,
|
||||
gv_node,
|
||||
&self.graph,
|
||||
|
|
|
|||
|
|
@ -12,9 +12,11 @@ use fabro_core::state::RunState;
|
|||
use super::super::graph::WorkflowGraph;
|
||||
use super::super::WorkflowNode;
|
||||
use crate::artifact::ArtifactStore;
|
||||
use crate::engine::{self, RunSettings};
|
||||
use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent};
|
||||
use crate::outcome::{Outcome, StageStatus, StageUsage};
|
||||
use crate::run_dir::node_dir;
|
||||
use crate::run_settings::RunSettings;
|
||||
use crate::sandbox_git::{git_checkpoint, git_diff, git_push_host};
|
||||
|
||||
type WfRunState = RunState<Option<StageUsage>>;
|
||||
type WfNodeResult = NodeResult<Option<StageUsage>>;
|
||||
|
|
@ -149,7 +151,7 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
|
|||
|
||||
// Run branch commit via sandbox
|
||||
let completed_count = state.completed_nodes.len();
|
||||
let commit_result = engine::git_checkpoint(
|
||||
let commit_result = git_checkpoint(
|
||||
&*self.sandbox,
|
||||
&self.run_id,
|
||||
node_id,
|
||||
|
|
@ -192,7 +194,7 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
|
|||
true
|
||||
} else if let Some(repo_path) = self.config.host_repo_path.as_ref() {
|
||||
let refspec = format!("refs/heads/{branch}");
|
||||
engine::git_push_host(
|
||||
git_push_host(
|
||||
repo_path,
|
||||
&refspec,
|
||||
&self.config.github_app,
|
||||
|
|
@ -213,7 +215,7 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
|
|||
self.config.host_repo_path.as_ref(),
|
||||
) {
|
||||
let refspec = format!("refs/heads/{meta_branch}");
|
||||
let meta_push_ok = engine::git_push_host(
|
||||
let meta_push_ok = git_push_host(
|
||||
repo_path,
|
||||
&refspec,
|
||||
&self.config.github_app,
|
||||
|
|
@ -235,9 +237,9 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
|
|||
.clone()
|
||||
.or_else(|| self.config.git.as_ref().and_then(|g| g.base_sha.clone()))
|
||||
.unwrap_or_else(|| sha.clone());
|
||||
let diff_dest = engine::node_dir(&self.run_dir, node_id, visit).join("diff.patch");
|
||||
let diff_dest = node_dir(&self.run_dir, node_id, visit).join("diff.patch");
|
||||
|
||||
match engine::git_diff(&*self.sandbox, &prev).await {
|
||||
match git_diff(&*self.sandbox, &prev).await {
|
||||
Ok(patch) if !patch.is_empty() => {
|
||||
let _ = std::fs::write(&diff_dest, patch);
|
||||
}
|
||||
|
|
@ -277,7 +279,7 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
|
|||
{
|
||||
if let Some(base_sha) = self.config.git.as_ref().and_then(|g| g.base_sha.clone()) {
|
||||
let diff_dest = self.run_dir.join("final.patch");
|
||||
match engine::git_diff(&*self.sandbox, &base_sha).await {
|
||||
match git_diff(&*self.sandbox, &base_sha).await {
|
||||
Ok(patch) if !patch.is_empty() => {
|
||||
let _ = std::fs::write(&diff_dest, patch);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -12,7 +12,7 @@ use fabro_core::state::RunState;
|
|||
|
||||
use super::super::graph::WorkflowGraph;
|
||||
use super::super::WorkflowNode;
|
||||
use crate::engine::set_hook_node;
|
||||
use crate::graph_ops::set_hook_node;
|
||||
use crate::outcome::{Outcome, OutcomeExt, StageStatus, StageUsage};
|
||||
use fabro_hooks::{HookContext, HookDecision, HookEvent, HookRunner};
|
||||
use fabro_sandbox::Sandbox;
|
||||
|
|
|
|||
|
|
@ -27,9 +27,9 @@ use super::graph::WorkflowGraph;
|
|||
use super::WorkflowNode;
|
||||
use crate::artifact::ArtifactStore;
|
||||
use crate::context;
|
||||
use crate::engine::RunSettings;
|
||||
use crate::event::EventEmitter;
|
||||
use crate::outcome::{Outcome, StageUsage};
|
||||
use crate::run_settings::RunSettings;
|
||||
use fabro_hooks::HookRunner;
|
||||
use fabro_sandbox::Sandbox;
|
||||
|
||||
|
|
@ -307,7 +307,7 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
|
|||
) -> CoreResult<()> {
|
||||
let outcome = &result.outcome;
|
||||
let retry_count = state.node_retries.get(node.id()).copied().unwrap_or(0);
|
||||
let failure_class = crate::engine::classify_outcome(outcome);
|
||||
let failure_class = crate::graph_ops::classify_outcome(outcome);
|
||||
let failure_signature = failure_class
|
||||
.map(|category| {
|
||||
let signature_hint = outcome
|
||||
|
|
|
|||
|
|
@ -28,17 +28,11 @@ use fabro_graphviz::graph::{Edge, Node};
|
|||
use fabro_hooks::{HookContext, HookDecision, HookEvent, HookRunner};
|
||||
use fabro_interview::Interviewer;
|
||||
|
||||
pub(crate) use crate::graph_ops::{
|
||||
build_retry_policy, check_goal_gates, classify_outcome, get_retry_target, is_terminal,
|
||||
node_script, set_hook_node,
|
||||
};
|
||||
pub use crate::graph_ops::{
|
||||
resolve_fidelity, resolve_thread_id, select_edge, EdgeSelection, RetryPolicy,
|
||||
};
|
||||
pub use crate::run_dir::{node_dir, visit_from_context};
|
||||
pub(crate) use crate::run_dir::{write_node_status, write_start_record};
|
||||
pub use crate::run_settings::{GitCheckpointSettings, LifecycleConfig, RunSettings};
|
||||
pub(crate) use crate::sandbox_git::git_diff;
|
||||
pub use crate::sandbox_git::{
|
||||
git_add_worktree, git_checkpoint, git_create_branch_at, git_merge_ff_only, git_push_host,
|
||||
git_remove_worktree, git_replace_worktree, GitState, GIT_REMOTE,
|
||||
|
|
@ -613,6 +607,7 @@ mod tests {
|
|||
use super::*;
|
||||
use crate::handler::start::StartHandler;
|
||||
use crate::handler::Handler as HandlerTrait;
|
||||
use crate::outcome::OutcomeExt;
|
||||
use async_trait::async_trait;
|
||||
use fabro_graphviz::graph::AttrValue;
|
||||
use std::time::Duration;
|
||||
|
|
|
|||
|
|
@ -249,8 +249,8 @@ impl Handler for AgentHandler {
|
|||
};
|
||||
|
||||
// 2. Write prompt to logs
|
||||
let visit = crate::engine::visit_from_context(context);
|
||||
let stage_dir = crate::engine::node_dir(run_dir, &node.id, visit);
|
||||
let visit = crate::run_dir::visit_from_context(context);
|
||||
let stage_dir = crate::run_dir::node_dir(run_dir, &node.id, visit);
|
||||
tokio::fs::create_dir_all(&stage_dir).await?;
|
||||
tokio::fs::write(stage_dir.join("prompt.md"), &prompt).await?;
|
||||
|
||||
|
|
|
|||
|
|
@ -84,8 +84,8 @@ impl Handler for CommandHandler {
|
|||
)));
|
||||
}
|
||||
|
||||
let visit = crate::engine::visit_from_context(context);
|
||||
let stage_dir = crate::engine::node_dir(run_dir, &node.id, visit);
|
||||
let visit = crate::run_dir::visit_from_context(context);
|
||||
let stage_dir = crate::run_dir::node_dir(run_dir, &node.id, visit);
|
||||
tokio::fs::create_dir_all(&stage_dir).await?;
|
||||
|
||||
let invocation = serde_json::json!({
|
||||
|
|
|
|||
|
|
@ -116,7 +116,7 @@ impl Handler for FanInHandler {
|
|||
};
|
||||
|
||||
if let (Some(ref sha), Some(_)) = (&best_head_sha, services.git_state()) {
|
||||
crate::engine::git_merge_ff_only(&*services.sandbox, sha).await;
|
||||
crate::sandbox_git::git_merge_ff_only(&*services.sandbox, sha).await;
|
||||
}
|
||||
|
||||
let mut outcome = Outcome::success();
|
||||
|
|
@ -231,8 +231,8 @@ async fn llm_evaluate(
|
|||
);
|
||||
|
||||
// Write prompt to logs
|
||||
let visit = crate::engine::visit_from_context(context);
|
||||
let stage_dir = crate::engine::node_dir(run_dir, node_id, visit);
|
||||
let visit = crate::run_dir::visit_from_context(context);
|
||||
let stage_dir = crate::run_dir::node_dir(run_dir, node_id, visit);
|
||||
tokio::fs::create_dir_all(&stage_dir).await?;
|
||||
tokio::fs::write(stage_dir.join("prompt.md"), &full_prompt).await?;
|
||||
|
||||
|
|
|
|||
|
|
@ -9,10 +9,11 @@ use async_trait::async_trait;
|
|||
use crate::condition::evaluate_condition;
|
||||
use crate::context::keys;
|
||||
use crate::context::{Context, WorkflowContext};
|
||||
use crate::engine::{RunSettings, WorkflowRunEngine};
|
||||
use crate::engine::WorkflowRunEngine;
|
||||
use crate::error::FabroError;
|
||||
use crate::operations::{create, create_from_file, CreateOptions};
|
||||
use crate::outcome::{Outcome, OutcomeExt, StageStatus};
|
||||
use crate::workflow::{prepare_from_file, prepare_from_source};
|
||||
use crate::run_settings::RunSettings;
|
||||
use fabro_graphviz::graph::{Graph, Node};
|
||||
|
||||
use super::{EngineServices, Handler};
|
||||
|
|
@ -52,7 +53,10 @@ fn parse_child_graph(node: &Node) -> Result<Graph, FabroError> {
|
|||
.get("stack.child_dot_source")
|
||||
.and_then(|v| v.as_str())
|
||||
{
|
||||
return prepare_from_source(dot);
|
||||
let validated = create(dot, CreateOptions::default())?;
|
||||
validated.raise_on_errors()?;
|
||||
let (graph, _, _) = validated.into_parts();
|
||||
return Ok(graph);
|
||||
}
|
||||
if let Some(path) = node
|
||||
.attrs
|
||||
|
|
@ -60,8 +64,9 @@ fn parse_child_graph(node: &Node) -> Result<Graph, FabroError> {
|
|||
.or_else(|| node.attrs.get("stack.child_dotfile"))
|
||||
.and_then(|v| v.as_str())
|
||||
{
|
||||
let (graph, diagnostics) = prepare_from_file(std::path::Path::new(path))?;
|
||||
fabro_validate::raise_on_errors(&diagnostics)?;
|
||||
let validated = create_from_file(std::path::Path::new(path))?;
|
||||
validated.raise_on_errors()?;
|
||||
let (graph, _, _) = validated.into_parts();
|
||||
return Ok(graph);
|
||||
}
|
||||
Err(FabroError::handler("No child workflow source".to_string()))
|
||||
|
|
@ -128,7 +133,7 @@ impl Handler for SubWorkflowHandler {
|
|||
};
|
||||
|
||||
// Build child RunSettings
|
||||
let visit = crate::engine::visit_from_context(context) as u64;
|
||||
let visit = crate::run_dir::visit_from_context(context) as u64;
|
||||
let child_logs = run_dir.join(format!("nodes/{}_{visit}/child", node.id));
|
||||
let _ = std::fs::create_dir_all(&child_logs);
|
||||
|
||||
|
|
|
|||
|
|
@ -19,10 +19,10 @@ use async_trait::async_trait;
|
|||
use fabro_agent::Sandbox;
|
||||
|
||||
use crate::context::Context;
|
||||
use crate::engine::GitState;
|
||||
use crate::error::FabroError;
|
||||
use crate::event::EventEmitter;
|
||||
use crate::outcome::{Outcome, OutcomeExt};
|
||||
use crate::sandbox_git::GitState;
|
||||
use fabro_graphviz::graph::{shape_to_handler_type, Graph, Node};
|
||||
use fabro_hooks::{HookContext, HookDecision, HookRunner};
|
||||
use fabro_interview::Interviewer;
|
||||
|
|
|
|||
|
|
@ -8,9 +8,9 @@ use tokio::sync::Semaphore;
|
|||
|
||||
use crate::context::keys;
|
||||
use crate::context::{Context, WorkflowContext};
|
||||
use crate::engine::set_hook_node;
|
||||
use crate::error::FabroError;
|
||||
use crate::event::WorkflowRunEvent;
|
||||
use crate::graph_ops::set_hook_node;
|
||||
use crate::millis_u64;
|
||||
use crate::outcome::{Outcome, OutcomeExt, StageStatus};
|
||||
use fabro_graphviz::graph::{Graph, Node};
|
||||
|
|
@ -162,7 +162,7 @@ impl Handler for ParallelHandler {
|
|||
|
||||
// --- Git isolation: checkpoint "parallel base" before fan-out ---
|
||||
let base_sha: Option<String> = if let Some(ref gs) = git_state {
|
||||
let result = crate::engine::git_checkpoint(
|
||||
let result = crate::sandbox_git::git_checkpoint(
|
||||
&*services.sandbox,
|
||||
&gs.run_id,
|
||||
&node.id,
|
||||
|
|
@ -205,7 +205,7 @@ impl Handler for ParallelHandler {
|
|||
(&git_state, &base_sha)
|
||||
{
|
||||
let branch_key = &target_id;
|
||||
let visit = crate::engine::visit_from_context(&branch_context);
|
||||
let visit = crate::run_dir::visit_from_context(&branch_context);
|
||||
let branch_name = format!(
|
||||
"fabro/run/parallel/{}/{}/pass{}/{}",
|
||||
gs.run_id,
|
||||
|
|
@ -327,7 +327,7 @@ impl Handler for ParallelHandler {
|
|||
let nid = &setup.target_id;
|
||||
let status_str = outcome.status.to_string();
|
||||
// Use exec_command to commit and capture HEAD in the branch worktree
|
||||
let git_r = crate::engine::GIT_REMOTE;
|
||||
let git_r = crate::sandbox_git::GIT_REMOTE;
|
||||
let add_cmd = format!("{git_r} add -A");
|
||||
let add_result = setup
|
||||
.sandbox
|
||||
|
|
@ -414,7 +414,7 @@ impl Handler for ParallelHandler {
|
|||
for result in &results {
|
||||
if let Some(ref wt_path) = result.worktree_path {
|
||||
let wt_str = wt_path.to_string_lossy().into_owned();
|
||||
crate::engine::git_remove_worktree(&*services.sandbox, &wt_str).await;
|
||||
crate::sandbox_git::git_remove_worktree(&*services.sandbox, &wt_str).await;
|
||||
services
|
||||
.emitter
|
||||
.emit(&WorkflowRunEvent::GitWorktreeRemove { path: wt_str });
|
||||
|
|
@ -431,7 +431,7 @@ impl Handler for ParallelHandler {
|
|||
successful.sort_by(|a, b| a.id.cmp(&b.id));
|
||||
if let Some(winner) = successful.first() {
|
||||
let sha = winner.head_sha.as_ref().unwrap();
|
||||
crate::engine::git_merge_ff_only(&*services.sandbox, sha).await;
|
||||
crate::sandbox_git::git_merge_ff_only(&*services.sandbox, sha).await;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -463,8 +463,8 @@ impl Handler for ParallelHandler {
|
|||
context.set(keys::PARALLEL_RESULTS, serde_json::json!(results_json));
|
||||
context.set(keys::PARALLEL_BRANCH_COUNT, serde_json::json!(total));
|
||||
|
||||
let visit = crate::engine::visit_from_context(context);
|
||||
let node_dir = crate::engine::node_dir(run_dir, &node.id, visit);
|
||||
let visit = crate::run_dir::visit_from_context(context);
|
||||
let node_dir = crate::run_dir::node_dir(run_dir, &node.id, visit);
|
||||
let _ = tokio::fs::create_dir_all(&node_dir).await;
|
||||
if let Ok(json) = serde_json::to_string_pretty(&results_json) {
|
||||
let _ = tokio::fs::write(node_dir.join("parallel_results.json"), json).await;
|
||||
|
|
|
|||
|
|
@ -86,8 +86,8 @@ impl Handler for PromptHandler {
|
|||
};
|
||||
|
||||
// 2. Write prompt to logs
|
||||
let visit = crate::engine::visit_from_context(context);
|
||||
let stage_dir = crate::engine::node_dir(run_dir, &node.id, visit);
|
||||
let visit = crate::run_dir::visit_from_context(context);
|
||||
let stage_dir = crate::run_dir::node_dir(run_dir, &node.id, visit);
|
||||
tokio::fs::create_dir_all(&stage_dir).await?;
|
||||
tokio::fs::write(stage_dir.join("prompt.md"), &prompt).await?;
|
||||
|
||||
|
|
|
|||
|
|
@ -106,6 +106,7 @@ pub mod git;
|
|||
pub mod graph_ops;
|
||||
pub mod graph_render;
|
||||
pub mod handler;
|
||||
pub mod operations;
|
||||
pub mod outcome;
|
||||
pub mod pipeline;
|
||||
pub mod preamble;
|
||||
|
|
@ -123,6 +124,8 @@ pub mod sandbox_reconnect;
|
|||
pub mod sandbox_record;
|
||||
pub mod start_record;
|
||||
pub mod stylesheet;
|
||||
#[doc(hidden)]
|
||||
pub mod test_support;
|
||||
pub mod transform;
|
||||
pub mod vars;
|
||||
pub mod workflow;
|
||||
|
|
|
|||
191
lib/crates/fabro-workflows/src/operations/create.rs
Normal file
191
lib/crates/fabro-workflows/src/operations/create.rs
Normal file
|
|
@ -0,0 +1,191 @@
|
|||
use std::path::{Path, PathBuf};
|
||||
|
||||
use fabro_graphviz::graph::Graph;
|
||||
|
||||
use crate::error::FabroError;
|
||||
use crate::pipeline::{self, TransformOptions, Validated};
|
||||
use crate::transform::Transform;
|
||||
|
||||
pub struct CreateOptions {
|
||||
pub base_dir: Option<PathBuf>,
|
||||
pub custom_transforms: Vec<Box<dyn Transform>>,
|
||||
}
|
||||
|
||||
impl Default for CreateOptions {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
base_dir: None,
|
||||
custom_transforms: Vec::new(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Parse, transform, and validate a DOT source string.
|
||||
///
|
||||
/// Returns `Validated` even when validation produced errors. Call
|
||||
/// `validated.raise_on_errors()` if the caller wants to fail fast.
|
||||
pub fn create(dot_source: &str, options: CreateOptions) -> Result<Validated, FabroError> {
|
||||
let parsed = pipeline::parse(dot_source)?;
|
||||
let transformed = pipeline::transform(
|
||||
parsed,
|
||||
&TransformOptions {
|
||||
base_dir: options.base_dir,
|
||||
custom_transforms: options.custom_transforms,
|
||||
},
|
||||
);
|
||||
Ok(pipeline::validate(transformed, &[]))
|
||||
}
|
||||
|
||||
/// Read a DOT file, apply file inlining from its parent directory, then create.
|
||||
pub fn create_from_file(path: &Path) -> Result<Validated, FabroError> {
|
||||
let source = std::fs::read_to_string(path)
|
||||
.map_err(|e| FabroError::Parse(format!("Failed to read {}: {e}", path.display())))?;
|
||||
let base_dir = path.parent().unwrap_or(Path::new("."));
|
||||
create(
|
||||
&source,
|
||||
CreateOptions {
|
||||
base_dir: Some(base_dir.to_path_buf()),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
/// Build a validated workflow from an already-materialized graph.
|
||||
///
|
||||
/// This is used by detached/resume CLI paths that load a graph from `RunRecord`
|
||||
/// instead of re-parsing DOT source.
|
||||
#[doc(hidden)]
|
||||
pub fn create_from_graph(graph: Graph, source: impl Into<String>) -> Validated {
|
||||
Validated::new(graph, source.into(), vec![])
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use fabro_graphviz::graph::AttrValue;
|
||||
|
||||
const MINIMAL_DOT: &str = r#"digraph Test {
|
||||
graph [goal="Build feature"]
|
||||
start [shape=Mdiamond]
|
||||
exit [shape=Msquare]
|
||||
start -> exit
|
||||
}"#;
|
||||
|
||||
#[test]
|
||||
fn create_minimal() {
|
||||
let validated = create(MINIMAL_DOT, CreateOptions::default()).unwrap();
|
||||
validated.raise_on_errors().unwrap();
|
||||
|
||||
assert_eq!(validated.graph().name, "Test");
|
||||
assert!(validated.graph().find_start_node().is_some());
|
||||
assert!(validated.graph().find_exit_node().is_some());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn create_applies_variable_expansion() {
|
||||
let dot = r#"digraph Test {
|
||||
graph [goal="Fix bugs"]
|
||||
start [shape=Mdiamond]
|
||||
work [prompt="Goal: $goal"]
|
||||
exit [shape=Msquare]
|
||||
start -> work -> exit
|
||||
}"#;
|
||||
let validated = create(dot, CreateOptions::default()).unwrap();
|
||||
validated.raise_on_errors().unwrap();
|
||||
|
||||
let prompt = validated.graph().nodes["work"]
|
||||
.attrs
|
||||
.get("prompt")
|
||||
.and_then(AttrValue::as_str)
|
||||
.unwrap();
|
||||
assert_eq!(prompt, "Goal: Fix bugs");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn create_applies_stylesheet() {
|
||||
let dot = r#"digraph Test {
|
||||
graph [goal="Test", model_stylesheet="* { model: sonnet; }"]
|
||||
start [shape=Mdiamond]
|
||||
work [label="Work"]
|
||||
exit [shape=Msquare]
|
||||
start -> work -> exit
|
||||
}"#;
|
||||
let validated = create(dot, CreateOptions::default()).unwrap();
|
||||
validated.raise_on_errors().unwrap();
|
||||
|
||||
assert_eq!(
|
||||
validated.graph().nodes["work"].attrs.get("model"),
|
||||
Some(&AttrValue::String("claude-sonnet-4-6".into()))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn create_returns_error_on_invalid_dot() {
|
||||
let result = create("not a graph", CreateOptions::default());
|
||||
assert!(result.is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn create_returns_validation_diagnostics() {
|
||||
let dot = r#"digraph Test {
|
||||
graph [goal="Test"]
|
||||
work [label="Work"]
|
||||
}"#;
|
||||
let validated = create(dot, CreateOptions::default()).unwrap();
|
||||
|
||||
assert!(validated.has_errors());
|
||||
assert!(validated.raise_on_errors().is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn create_supports_custom_transforms() {
|
||||
struct TagTransform;
|
||||
|
||||
impl Transform for TagTransform {
|
||||
fn apply(&self, graph: &mut fabro_graphviz::graph::Graph) {
|
||||
for node in graph.nodes.values_mut() {
|
||||
node.attrs
|
||||
.insert("tagged".to_string(), AttrValue::Boolean(true));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let validated = create(
|
||||
MINIMAL_DOT,
|
||||
CreateOptions {
|
||||
custom_transforms: vec![Box::new(TagTransform)],
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.unwrap();
|
||||
validated.raise_on_errors().unwrap();
|
||||
|
||||
assert_eq!(
|
||||
validated.graph().nodes["start"].attrs.get("tagged"),
|
||||
Some(&AttrValue::Boolean(true))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn create_from_file_uses_parent_directory_for_inlining() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let data_path = dir.path().join("goal.txt");
|
||||
let dot_path = dir.path().join("workflow.fabro");
|
||||
|
||||
std::fs::write(&data_path, "ship it").unwrap();
|
||||
std::fs::write(
|
||||
&dot_path,
|
||||
r#"digraph Test {
|
||||
graph [goal="@goal.txt"]
|
||||
start [shape=Mdiamond]
|
||||
exit [shape=Msquare]
|
||||
start -> exit
|
||||
}"#,
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
let validated = create_from_file(&dot_path).unwrap();
|
||||
validated.raise_on_errors().unwrap();
|
||||
assert_eq!(validated.graph().goal(), "ship it");
|
||||
}
|
||||
}
|
||||
1
lib/crates/fabro-workflows/src/operations/fork.rs
Normal file
1
lib/crates/fabro-workflows/src/operations/fork.rs
Normal file
|
|
@ -0,0 +1 @@
|
|||
pub use crate::run_fork::execute_fork as fork;
|
||||
12
lib/crates/fabro-workflows/src/operations/mod.rs
Normal file
12
lib/crates/fabro-workflows/src/operations/mod.rs
Normal file
|
|
@ -0,0 +1,12 @@
|
|||
mod create;
|
||||
mod fork;
|
||||
mod rewind;
|
||||
mod start;
|
||||
|
||||
pub use create::{create, create_from_file, create_from_graph, CreateOptions};
|
||||
pub use fork::fork;
|
||||
pub use rewind::{
|
||||
build_timeline, find_run_id_by_prefix, load_parallel_map, parse_target, resolve_target,
|
||||
rewind, TimelineEntry,
|
||||
};
|
||||
pub use start::{start, StartFinalizeConfig, StartOptions, StartRetroConfig, Started};
|
||||
4
lib/crates/fabro-workflows/src/operations/rewind.rs
Normal file
4
lib/crates/fabro-workflows/src/operations/rewind.rs
Normal file
|
|
@ -0,0 +1,4 @@
|
|||
pub use crate::run_rewind::{
|
||||
build_timeline, execute_rewind as rewind, find_run_id_by_prefix, load_parallel_map,
|
||||
parse_target, resolve_target, TimelineEntry,
|
||||
};
|
||||
117
lib/crates/fabro-workflows/src/operations/start.rs
Normal file
117
lib/crates/fabro-workflows/src/operations/start.rs
Normal file
|
|
@ -0,0 +1,117 @@
|
|||
use std::sync::{Arc, Mutex};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use crate::error::FabroError;
|
||||
use crate::event::WorkflowRunEvent;
|
||||
use crate::outcome::StageStatus;
|
||||
use crate::pipeline::{self, FinalizeOptions, Finalized, InitOptions, RetroOptions, Validated};
|
||||
|
||||
pub struct StartRetroConfig {
|
||||
pub enabled: bool,
|
||||
pub dry_run: bool,
|
||||
pub llm_client: Option<fabro_llm::client::Client>,
|
||||
pub provider: fabro_llm::Provider,
|
||||
pub model: String,
|
||||
}
|
||||
|
||||
pub struct StartFinalizeConfig {
|
||||
pub preserve_sandbox: bool,
|
||||
pub pr_config: Option<fabro_config::run::PullRequestConfig>,
|
||||
pub github_app: Option<fabro_github::GitHubAppCredentials>,
|
||||
pub origin_url: Option<String>,
|
||||
pub model: String,
|
||||
}
|
||||
|
||||
pub struct StartOptions {
|
||||
pub init: InitOptions,
|
||||
pub retro: StartRetroConfig,
|
||||
pub finalize: StartFinalizeConfig,
|
||||
}
|
||||
|
||||
pub struct Started {
|
||||
pub finalized: Finalized,
|
||||
pub retro: Option<fabro_retro::retro::Retro>,
|
||||
pub retro_duration: Duration,
|
||||
}
|
||||
|
||||
/// Run a validated workflow through initialize, execute, retro, and finalize.
|
||||
pub async fn start(validated: Validated, options: StartOptions) -> Result<Started, FabroError> {
|
||||
let preserve_sandbox = options.finalize.preserve_sandbox;
|
||||
let sandbox_for_cleanup = Arc::clone(&options.init.sandbox);
|
||||
let cleanup_guard = scopeguard::guard((), move |()| {
|
||||
if preserve_sandbox {
|
||||
return;
|
||||
}
|
||||
if let Ok(handle) = tokio::runtime::Handle::try_current() {
|
||||
handle.spawn(async move {
|
||||
let _ = sandbox_for_cleanup.cleanup().await;
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
let initialized = pipeline::initialize(validated, options.init).await?;
|
||||
|
||||
let last_git_sha: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
|
||||
{
|
||||
let sha_clone = Arc::clone(&last_git_sha);
|
||||
initialized.emitter.on_event(move |event| {
|
||||
if let WorkflowRunEvent::CheckpointCompleted {
|
||||
git_commit_sha: Some(sha),
|
||||
..
|
||||
} = event
|
||||
{
|
||||
*sha_clone.lock().unwrap() = Some(sha.clone());
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
let executed = pipeline::execute(initialized).await;
|
||||
let failed = !matches!(
|
||||
executed.outcome.as_ref().map(|outcome| &outcome.status),
|
||||
Ok(StageStatus::Success) | Ok(StageStatus::PartialSuccess)
|
||||
);
|
||||
|
||||
let retro_opts = RetroOptions {
|
||||
run_id: executed.settings.run_id.clone(),
|
||||
workflow_name: executed.graph.name.clone(),
|
||||
goal: executed.graph.goal().to_string(),
|
||||
run_dir: executed.settings.run_dir.clone(),
|
||||
sandbox: Arc::clone(&executed.sandbox),
|
||||
emitter: Some(Arc::clone(&executed.emitter)),
|
||||
failed,
|
||||
run_duration_ms: executed.duration_ms,
|
||||
enabled: options.retro.enabled,
|
||||
dry_run: options.retro.dry_run,
|
||||
llm_client: options.retro.llm_client,
|
||||
provider: options.retro.provider,
|
||||
model: options.retro.model,
|
||||
};
|
||||
|
||||
let retro_start = Instant::now();
|
||||
let retroed = pipeline::retro(executed, &retro_opts).await;
|
||||
let retro_duration = retro_start.elapsed();
|
||||
|
||||
let finalize_opts = FinalizeOptions {
|
||||
run_dir: retroed.settings.run_dir.clone(),
|
||||
run_id: retroed.settings.run_id.clone(),
|
||||
workflow_name: retroed.graph.name.clone(),
|
||||
hook_runner: retroed.hook_runner.clone(),
|
||||
preserve_sandbox: options.finalize.preserve_sandbox,
|
||||
pr_config: options.finalize.pr_config,
|
||||
github_app: options.finalize.github_app,
|
||||
origin_url: options.finalize.origin_url,
|
||||
model: options.finalize.model,
|
||||
last_git_sha: last_git_sha.lock().unwrap().clone(),
|
||||
};
|
||||
|
||||
let retro = retroed.retro.clone();
|
||||
let finalized = pipeline::finalize(retroed, &finalize_opts).await?;
|
||||
|
||||
scopeguard::ScopeGuard::into_inner(cleanup_guard);
|
||||
|
||||
Ok(Started {
|
||||
finalized,
|
||||
retro,
|
||||
retro_duration,
|
||||
})
|
||||
}
|
||||
|
|
@ -357,6 +357,7 @@ pub async fn finalize(
|
|||
run_id: settings.run_id,
|
||||
outcome,
|
||||
conclusion,
|
||||
pushed_branch: settings.git.as_ref().and_then(|g| g.run_branch.clone()),
|
||||
pr_url,
|
||||
})
|
||||
}
|
||||
|
|
|
|||
|
|
@ -158,6 +158,7 @@ pub struct Finalized {
|
|||
pub run_id: String,
|
||||
pub outcome: Result<Outcome, FabroError>,
|
||||
pub conclusion: Conclusion,
|
||||
pub pushed_branch: Option<String>,
|
||||
pub pr_url: Option<String>,
|
||||
}
|
||||
|
||||
|
|
|
|||
119
lib/crates/fabro-workflows/src/test_support.rs
Normal file
119
lib/crates/fabro-workflows/src/test_support.rs
Normal file
|
|
@ -0,0 +1,119 @@
|
|||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
|
||||
use fabro_agent::Sandbox;
|
||||
|
||||
use crate::checkpoint::Checkpoint;
|
||||
use crate::engine::WorkflowRunEngine;
|
||||
use crate::error::Result;
|
||||
use crate::event::EventEmitter;
|
||||
use crate::handler::HandlerRegistry;
|
||||
use crate::outcome::Outcome;
|
||||
use crate::run_settings::RunSettings;
|
||||
|
||||
pub async fn run_graph(
|
||||
registry: HandlerRegistry,
|
||||
emitter: Arc<EventEmitter>,
|
||||
sandbox: Arc<dyn Sandbox>,
|
||||
graph: &fabro_graphviz::graph::Graph,
|
||||
settings: &RunSettings,
|
||||
) -> Result<Outcome> {
|
||||
let engine = WorkflowRunEngine::new(registry, emitter, sandbox);
|
||||
engine.run(graph, settings).await
|
||||
}
|
||||
|
||||
pub async fn run_graph_with_hooks(
|
||||
registry: HandlerRegistry,
|
||||
emitter: Arc<EventEmitter>,
|
||||
sandbox: Arc<dyn Sandbox>,
|
||||
graph: &fabro_graphviz::graph::Graph,
|
||||
settings: &RunSettings,
|
||||
hook_runner: Arc<fabro_hooks::HookRunner>,
|
||||
env: Option<HashMap<String, String>>,
|
||||
) -> Result<Outcome> {
|
||||
let mut engine = WorkflowRunEngine::new(registry, emitter, sandbox);
|
||||
engine.set_hook_runner(hook_runner);
|
||||
if let Some(env) = env {
|
||||
engine.set_env(env);
|
||||
}
|
||||
engine.run(graph, settings).await
|
||||
}
|
||||
|
||||
pub async fn run_graph_from_checkpoint(
|
||||
registry: HandlerRegistry,
|
||||
emitter: Arc<EventEmitter>,
|
||||
sandbox: Arc<dyn Sandbox>,
|
||||
graph: &fabro_graphviz::graph::Graph,
|
||||
settings: &RunSettings,
|
||||
checkpoint: &Checkpoint,
|
||||
) -> Result<Outcome> {
|
||||
let engine = WorkflowRunEngine::new(registry, emitter, sandbox);
|
||||
engine
|
||||
.run_from_checkpoint(graph, settings, checkpoint)
|
||||
.await
|
||||
}
|
||||
|
||||
pub struct WorkflowRunner {
|
||||
registry: std::sync::Mutex<Option<HandlerRegistry>>,
|
||||
emitter: Arc<EventEmitter>,
|
||||
sandbox: Arc<dyn Sandbox>,
|
||||
}
|
||||
|
||||
impl WorkflowRunner {
|
||||
#[must_use]
|
||||
pub fn new(
|
||||
registry: HandlerRegistry,
|
||||
emitter: Arc<EventEmitter>,
|
||||
sandbox: Arc<dyn Sandbox>,
|
||||
) -> Self {
|
||||
Self {
|
||||
registry: std::sync::Mutex::new(Some(registry)),
|
||||
emitter,
|
||||
sandbox,
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn run(
|
||||
&self,
|
||||
graph: &fabro_graphviz::graph::Graph,
|
||||
settings: &RunSettings,
|
||||
) -> Result<Outcome> {
|
||||
let registry = self
|
||||
.registry
|
||||
.lock()
|
||||
.unwrap()
|
||||
.take()
|
||||
.expect("WorkflowRunner may only be used once");
|
||||
run_graph(
|
||||
registry,
|
||||
Arc::clone(&self.emitter),
|
||||
Arc::clone(&self.sandbox),
|
||||
graph,
|
||||
settings,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn run_from_checkpoint(
|
||||
&self,
|
||||
graph: &fabro_graphviz::graph::Graph,
|
||||
settings: &RunSettings,
|
||||
checkpoint: &Checkpoint,
|
||||
) -> Result<Outcome> {
|
||||
let registry = self
|
||||
.registry
|
||||
.lock()
|
||||
.unwrap()
|
||||
.take()
|
||||
.expect("WorkflowRunner may only be used once");
|
||||
run_graph_from_checkpoint(
|
||||
registry,
|
||||
Arc::clone(&self.emitter),
|
||||
Arc::clone(&self.sandbox),
|
||||
graph,
|
||||
settings,
|
||||
checkpoint,
|
||||
)
|
||||
.await
|
||||
}
|
||||
}
|
||||
|
|
@ -15,13 +15,14 @@ use fabro_sandbox::daytona::{DaytonaConfig, DaytonaSandbox, DaytonaSnapshotConfi
|
|||
use fabro_workflows::artifact::sync_artifacts_to_env;
|
||||
use fabro_workflows::checkpoint::Checkpoint;
|
||||
use fabro_workflows::context::Context;
|
||||
use fabro_workflows::engine::{GitCheckpointSettings, RunSettings, WorkflowRunEngine};
|
||||
use fabro_workflows::error::FabroError;
|
||||
use fabro_workflows::event::EventEmitter;
|
||||
use fabro_workflows::handler::exit::ExitHandler;
|
||||
use fabro_workflows::handler::start::StartHandler;
|
||||
use fabro_workflows::handler::{Handler, HandlerRegistry};
|
||||
use fabro_workflows::outcome::{Outcome, OutcomeExt, StageStatus};
|
||||
use fabro_workflows::run_settings::{GitCheckpointSettings, RunSettings};
|
||||
use fabro_workflows::test_support::WorkflowRunner;
|
||||
|
||||
async fn create_env() -> DaytonaSandbox {
|
||||
let creds = load_github_app_credentials();
|
||||
|
|
@ -386,7 +387,7 @@ async fn daytona_pipeline_artifact_offload_and_sync() {
|
|||
registry.register("start", Box::new(StartHandler));
|
||||
registry.register("exit", Box::new(ExitHandler));
|
||||
|
||||
let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), env.clone());
|
||||
let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), env.clone());
|
||||
let config = RunSettings {
|
||||
config: FabroConfig::default(),
|
||||
run_dir: dir.path().to_path_buf(),
|
||||
|
|
@ -577,7 +578,7 @@ async fn daytona_git_checkpoint_remote_emits_events() {
|
|||
registry.register("start", Box::new(StartHandler));
|
||||
registry.register("exit", Box::new(ExitHandler));
|
||||
|
||||
let engine = WorkflowRunEngine::new(registry, Arc::new(emitter), env.clone());
|
||||
let engine = WorkflowRunner::new(registry, Arc::new(emitter), env.clone());
|
||||
let config = RunSettings {
|
||||
config: FabroConfig::default(),
|
||||
run_dir: dir.path().to_path_buf(),
|
||||
|
|
@ -762,7 +763,7 @@ async fn daytona_parallel_git_branching_e2e() {
|
|||
registry.register("parallel", Box::new(ParallelHandler));
|
||||
registry.register("parallel.fan_in", Box::new(FanInHandler::new(None)));
|
||||
|
||||
let engine = WorkflowRunEngine::new(registry, Arc::new(emitter), Arc::clone(&env));
|
||||
let engine = WorkflowRunner::new(registry, Arc::new(emitter), Arc::clone(&env));
|
||||
|
||||
let config = RunSettings {
|
||||
config: FabroConfig::default(),
|
||||
|
|
@ -1139,7 +1140,7 @@ async fn daytona_git_checkpoint_with_shadow_branch() {
|
|||
registry.register("exit", Box::new(ExitHandler));
|
||||
|
||||
let meta_branch = MetadataStore::branch_name(&run_id);
|
||||
let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), env.clone());
|
||||
let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), env.clone());
|
||||
let config = RunSettings {
|
||||
config: FabroConfig::default(),
|
||||
run_dir: dir.path().to_path_buf(),
|
||||
|
|
@ -1246,7 +1247,7 @@ async fn daytona_asset_collection() {
|
|||
registry.register("start", Box::new(StartHandler));
|
||||
registry.register("exit", Box::new(ExitHandler));
|
||||
|
||||
let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), env.clone());
|
||||
let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), env.clone());
|
||||
|
||||
let mut graph = Graph::new("DaytonaAssetTest");
|
||||
graph.attrs.insert(
|
||||
|
|
@ -1535,7 +1536,7 @@ async fn daytona_git_push_run_branch_to_origin() {
|
|||
registry.register("start", Box::new(StartHandler));
|
||||
registry.register("exit", Box::new(ExitHandler));
|
||||
|
||||
let engine = WorkflowRunEngine::new(registry, Arc::new(EventEmitter::new()), env.clone());
|
||||
let engine = WorkflowRunner::new(registry, Arc::new(EventEmitter::new()), env.clone());
|
||||
let config = RunSettings {
|
||||
config: FabroConfig::default(),
|
||||
run_dir: dir.path().to_path_buf(),
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load diff
Loading…
Add table
Reference in a new issue