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

Refactor workflow runtime initialization

Browse source
This commit is contained in:
Bryan Helmkamp 2026-03-26 22:18:07 -04:00
parent b748c7e6ea
commit 205d40d687
No known key found for this signature in database
28 changed files with 1532 additions and 1290 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
docs-internal/events-strategy.md
View file

@ -37,7 +37,7 @@ Every line in `progress.jsonl` has three envelope fields, then the event's own f
| `run_id` | string | ULID for this workflow run |
| `event` | string | Event name (matches Rust variant, dot-separated for wrapped types) |
The envelope is built in `cli/run.rs`. Field names from the event that collide with envelope keys (`ts`, `run_id`, `event`) are dropped — the `run_id` from `WorkflowRunStarted` populates the envelope itself.
The envelope is built in `fabro-workflows/src/event.rs` by `build_event_envelope()`, and file logging is handled by `ProgressLogger`. Field names from the event that collide with envelope keys (`ts`, `run_id`, `event`) are dropped — the `run_id` from `WorkflowRunStarted` populates the envelope itself.
## Run Completion Contract
@ -154,7 +154,7 @@ WorkflowRunEvent::MyNewEvent { node_id, duration_ms, .. } => {
| Event | JSONL fields |
|---|---|
| `WorkflowRunStarted` | `workflow_name`, `run_id`, `base_sha`?, `run_branch`?, `worktree_dir`? |
| `WorkflowRunStarted` | `workflow_name`, `run_id`, `base_branch`?, `base_sha`?, `run_branch`?, `worktree_dir`? |
| `WorkflowRunCompleted` | `duration_ms`, `artifact_count`, `total_cost`?, `final_git_commit_sha`? |
| `WorkflowRunFailed` | `error`, `duration_ms`, `git_commit_sha`? |
| `RunNotice` | `level`, `code`, `message` |

24
lib/crates/fabro-api/src/serve.rs
View file

@ -14,10 +14,8 @@ use fabro_config::FabroConfig;
use crate::jwt_auth::{AuthMode, AuthStrategy};
use crate::server::build_router;
use crate::tls::ClientAuth;
use fabro_interview::Interviewer;
use fabro_sandbox::SandboxProvider;
use fabro_workflows::handler::default_registry;
use fabro_workflows::handler::llm::AgentApiBackend;
use fabro_workflows::pipeline::LlmSpec;
#[derive(Args)]
pub struct ServeArgs {
@ -97,23 +95,19 @@ pub async fn serve_command(args: ServeArgs, styles: &'static Styles) -> anyhow::
// Build registry factory that reads live config
let config_for_factory = Arc::clone(&shared_config);
let factory = move |interviewer: Arc<dyn Interviewer>| {
let factory = move || {
let (model, provider_enum) = resolve_model_provider(
&config_for_factory,
cli_model.as_deref(),
cli_provider.as_deref(),
);
default_registry(interviewer, move || {
if dry_run_mode {
None
} else {
Some(Box::new(AgentApiBackend::new(
model.clone(),
provider_enum,
Vec::new(),
)))
}
})
LlmSpec {
model,
provider: provider_enum,
fallback_chain: Vec::new(),
mcp_servers: Vec::new(),
dry_run: dry_run_mode,
}
};
std::fs::create_dir_all(&data_dir)?;
let db = fabro_db::connect(&data_dir.join("fabro.db")).await?;

107
lib/crates/fabro-api/src/server.rs
View file

@ -16,16 +16,15 @@ use tower::ServiceExt;
use tracing::{error, info};
use fabro_agent::LocalSandbox;
use crate::error::ApiError;
use crate::jwt_auth::{AuthMode, AuthenticatedService, AuthenticatedUser};
use fabro_interview::{Answer, Interviewer, QuestionType, WebInterviewer};
use fabro_workflows::context::Context;
use fabro_workflows::event::{EventEmitter, WorkflowRunEvent};
use fabro_workflows::handler::HandlerRegistry;
use fabro_workflows::operations::{self, RunCreateOptions};
use fabro_workflows::pipeline::{self, InitOptions, Persisted};
use fabro_workflows::pipeline::{
self, InitOptions, LlmSpec, Persisted, SandboxEnvSpec, SandboxSpec,
};
use fabro_workflows::records::Checkpoint;
use fabro_workflows::run_options::LifecycleOptions;
use fabro_workflows::run_options::RunOptions;
@ -101,7 +100,7 @@ struct AggregateUsageTotals {
pub struct AppState {
runs: Mutex<HashMap<String, ManagedRun>>,
aggregate_usage: Mutex<AggregateUsageTotals>,
registry_factory: Box<dyn Fn(Arc<dyn Interviewer>) -> HandlerRegistry + Send + Sync>,
llm_spec_factory: Box<dyn Fn() -> LlmSpec + Send + Sync>,
pub dry_run: bool,
pub db: sqlx::SqlitePool,
max_concurrent_runs: usize,
@ -381,17 +380,14 @@ async fn get_aggregate_usage(
(StatusCode::OK, Json(response)).into_response()
}
/// Create an `AppState` with the given registry factory and database pool.
///
/// The factory receives the run's `WebInterviewer` so it can wire it
/// into handlers that need human-in-the-loop interaction (e.g., `HumanHandler`).
/// Create an `AppState` with the given LLM spec factory and database pool.
pub fn create_app_state(
db: sqlx::SqlitePool,
registry_factory: impl Fn(Arc<dyn Interviewer>) -> HandlerRegistry + Send + Sync + 'static,
llm_spec_factory: impl Fn() -> LlmSpec + Send + Sync + 'static,
) -> Arc<AppState> {
create_app_state_with_options(
db,
registry_factory,
llm_spec_factory,
false,
5,
fabro_workflows::git::GitAuthor::default(),
@ -399,10 +395,10 @@ pub fn create_app_state(
)
}
/// Create an `AppState` with the given database pool, registry factory, dry-run flag, and concurrency limit.
/// Create an `AppState` with the given database pool, LLM spec factory, dry-run flag, and concurrency limit.
pub fn create_app_state_with_options(
db: sqlx::SqlitePool,
registry_factory: impl Fn(Arc<dyn Interviewer>) -> HandlerRegistry + Send + Sync + 'static,
llm_spec_factory: impl Fn() -> LlmSpec + Send + Sync + 'static,
dry_run: bool,
max_concurrent_runs: usize,
git_author: fabro_workflows::git::GitAuthor,
@ -411,7 +407,7 @@ pub fn create_app_state_with_options(
Arc::new(AppState {
runs: Mutex::new(HashMap::new()),
aggregate_usage: Mutex::new(AggregateUsageTotals::default()),
registry_factory: Box::new(registry_factory),
llm_spec_factory: Box::new(llm_spec_factory),
dry_run,
db,
max_concurrent_runs,
@ -616,11 +612,11 @@ async fn execute_run(state: Arc<AppState>, run_id: String) {
None => return,
};
let registry = (state.registry_factory)(Arc::clone(&interviewer) as Arc<dyn Interviewer>);
let cwd = std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from("."));
let sandbox: Arc<dyn fabro_agent::Sandbox> = Arc::new(
fabro_agent::ReadBeforeWriteSandbox::new(Arc::new(LocalSandbox::new(cwd))),
);
let sandbox = SandboxSpec::Local {
working_directory: cwd,
};
let llm = (state.llm_spec_factory)();
let emitter = Arc::new(emitter);
// Transition to Running, populate interviewer + context
@ -664,14 +660,14 @@ async fn execute_run(state: Arc<AppState>, run_id: String) {
workflow_slug: run_record.workflow_slug,
github_app: None,
base_branch: run_record.base_branch,
display_base_sha: None,
host_repo_path: run_record.host_repo_path.map(Into::into),
git: None,
};
let execution = {
let emitter = Arc::clone(&emitter);
let sandbox = Arc::clone(&sandbox);
let registry = Arc::new(registry);
let interviewer = Arc::clone(&interviewer) as Arc<dyn Interviewer>;
let run_id = run_id.clone();
let run_options = run_options.clone();
let hooks = state.hooks.clone();
@ -684,7 +680,8 @@ async fn execute_run(state: Arc<AppState>, run_id: String) {
dry_run,
emitter,
sandbox,
registry,
llm,
interviewer,
lifecycle: LifecycleOptions {
setup_commands: Vec::new(),
setup_command_timeout_ms: 300_000,
@ -692,7 +689,15 @@ async fn execute_run(state: Arc<AppState>, run_id: String) {
},
run_options,
hooks: fabro_hooks::HookConfig { hooks },
sandbox_env: HashMap::new(),
sandbox_env: SandboxEnvSpec {
devcontainer_env: HashMap::new(),
toml_env: HashMap::new(),
github_permissions: None,
origin_url: None,
},
devcontainer: None,
git: None,
worktree_mode: None,
checkpoint: None,
seed_context: None,
},
@ -1540,9 +1545,6 @@ mod tests {
use axum::http::Request;
use tower::ServiceExt;
use fabro_workflows::handler::exit::ExitHandler;
use fabro_workflows::handler::start::StartHandler;
const MINIMAL_DOT: &str = r#"digraph Test {
graph [goal="Test"]
start [shape=Mdiamond]
@ -1550,11 +1552,14 @@ mod tests {
start -> exit
}"#;
fn test_registry(_interviewer: Arc<dyn fabro_interview::Interviewer>) -> HandlerRegistry {
let mut registry = HandlerRegistry::new(Box::new(StartHandler));
registry.register("start", Box::new(StartHandler));
registry.register("exit", Box::new(ExitHandler));
registry
fn test_llm_spec() -> LlmSpec {
LlmSpec {
model: "test-model".to_string(),
provider: fabro_llm::Provider::Anthropic,
fallback_chain: Vec::new(),
mcp_servers: Vec::new(),
dry_run: true,
}
}
async fn test_db() -> sqlx::SqlitePool {
@ -1564,7 +1569,7 @@ mod tests {
}
fn test_app_with(db: sqlx::SqlitePool) -> Router {
let state = create_app_state(db, test_registry);
let state = create_app_state(db, test_llm_spec);
build_router(state, AuthMode::Disabled)
}
@ -1616,7 +1621,7 @@ mod tests {
async fn test_model_dry_run_returns_ok() {
let state = create_app_state_with_options(
test_db().await,
test_registry,
test_llm_spec,
true,
5,
fabro_workflows::git::GitAuthor::default(),
@ -1643,7 +1648,7 @@ mod tests {
async fn test_model_dry_run_unknown_returns_404() {
let state = create_app_state_with_options(
test_db().await,
test_registry,
test_llm_spec,
true,
5,
fabro_workflows::git::GitAuthor::default(),
@ -1702,7 +1707,7 @@ mod tests {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn get_run_status_returns_status() {
let state = create_app_state(test_db().await, test_registry);
let state = create_app_state(test_db().await, test_llm_spec);
let app = test_app_with_scheduler(state);
// Start a run
@ -1760,7 +1765,7 @@ mod tests {
#[tokio::test]
async fn get_questions_returns_empty_list() {
let state = create_app_state(test_db().await, test_registry);
let state = create_app_state(test_db().await, test_llm_spec);
let app = build_router(Arc::clone(&state), AuthMode::Disabled);
// Start a run
@ -1825,7 +1830,7 @@ mod tests {
#[tokio::test]
async fn get_checkpoint_returns_null_initially() {
let state = create_app_state(test_db().await, test_registry);
let state = create_app_state(test_db().await, test_llm_spec);
let app = build_router(Arc::clone(&state), AuthMode::Disabled);
// Start a run
@ -1855,7 +1860,7 @@ mod tests {
#[tokio::test]
async fn get_context_returns_map() {
let state = create_app_state(test_db().await, test_registry);
let state = create_app_state(test_db().await, test_llm_spec);
let app = build_router(Arc::clone(&state), AuthMode::Disabled);
// Start a run
@ -1888,7 +1893,7 @@ mod tests {
#[tokio::test]
async fn cancel_run_succeeds() {
let state = create_app_state(test_db().await, test_registry);
let state = create_app_state(test_db().await, test_llm_spec);
let app = build_router(Arc::clone(&state), AuthMode::Disabled);
// Start a run
@ -1937,7 +1942,7 @@ mod tests {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn get_events_returns_sse_stream() {
let state = create_app_state(test_db().await, test_registry);
let state = create_app_state(test_db().await, test_llm_spec);
let app = test_app_with_scheduler(state);
// Start a run
@ -1988,7 +1993,7 @@ mod tests {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn run_completes_and_status_is_completed() {
let state = create_app_state(test_db().await, test_registry);
let state = create_app_state(test_db().await, test_llm_spec);
let app = test_app_with_scheduler(state);
// Start a run
@ -2027,7 +2032,7 @@ mod tests {
#[tokio::test]
async fn get_graph_returns_svg() {
let state = create_app_state(test_db().await, test_registry);
let state = create_app_state(test_db().await, test_llm_spec);
let app = build_router(Arc::clone(&state), AuthMode::Disabled);
// Start a run
@ -2095,7 +2100,7 @@ mod tests {
#[tokio::test]
async fn list_runs_returns_started_run() {
let state = create_app_state(test_db().await, test_registry);
let state = create_app_state(test_db().await, test_llm_spec);
let app = build_router(Arc::clone(&state), AuthMode::Disabled);
// List should be empty initially
@ -2144,7 +2149,7 @@ mod tests {
#[tokio::test]
async fn get_aggregate_usage_returns_zeros_initially() {
let state = create_app_state(test_db().await, test_registry);
let state = create_app_state(test_db().await, test_llm_spec);
let app = build_router(Arc::clone(&state), AuthMode::Disabled);
let req = Request::builder()
@ -2167,7 +2172,7 @@ mod tests {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn aggregate_usage_increments_after_run_completes() {
let state = create_app_state(test_db().await, test_registry);
let state = create_app_state(test_db().await, test_llm_spec);
let app = test_app_with_scheduler(state);
// Start a run
@ -2218,7 +2223,7 @@ mod tests {
#[tokio::test]
async fn post_runs_returns_queued_status() {
let state = create_app_state(test_db().await, test_registry);
let state = create_app_state(test_db().await, test_llm_spec);
let app = build_router(state, AuthMode::Disabled);
let req = Request::builder()
@ -2249,7 +2254,7 @@ mod tests {
#[tokio::test]
async fn cancel_queued_run_succeeds() {
let state = create_app_state(test_db().await, test_registry);
let state = create_app_state(test_db().await, test_llm_spec);
let app = build_router(state, AuthMode::Disabled);
// Submit a run (no scheduler, stays queued)
@ -2290,7 +2295,7 @@ mod tests {
#[tokio::test]
async fn queue_position_reported_for_queued_runs() {
let state = create_app_state(test_db().await, test_registry);
let state = create_app_state(test_db().await, test_llm_spec);
let app = build_router(state, AuthMode::Disabled);
// Submit two runs (no scheduler, both stay queued)
@ -2334,7 +2339,7 @@ mod tests {
async fn concurrency_limit_respected() {
let state = create_app_state_with_options(
test_db().await,
test_registry,
test_llm_spec,
false,
1,
fabro_workflows::git::GitAuthor::default(),
@ -2388,7 +2393,7 @@ mod tests {
#[tokio::test]
async fn submit_answer_to_queued_run_returns_conflict() {
let state = create_app_state(test_db().await, test_registry);
let state = create_app_state(test_db().await, test_llm_spec);
let app = build_router(state, AuthMode::Disabled);
let req = Request::builder()
@ -2422,7 +2427,7 @@ mod tests {
async fn create_completion_non_streaming_returns_json() {
let state = create_app_state_with_options(
test_db().await,
test_registry,
test_llm_spec,
true,
5,
fabro_workflows::git::GitAuthor::default(),
@ -2459,7 +2464,7 @@ mod tests {
async fn create_completion_streaming_returns_sse() {
let state = create_app_state_with_options(
test_db().await,
test_registry,
test_llm_spec,
true,
5,
fabro_workflows::git::GitAuthor::default(),

19
lib/crates/fabro-api/src/sessions.rs
View file

@ -430,15 +430,16 @@ mod tests {
use crate::jwt_auth::AuthMode;
use crate::server::{build_router, create_app_state_with_options};
use fabro_workflows::handler::exit::ExitHandler;
use fabro_workflows::handler::start::StartHandler;
use fabro_workflows::handler::HandlerRegistry;
use fabro_workflows::pipeline::LlmSpec;
fn test_registry(_interviewer: Arc<dyn fabro_interview::Interviewer>) -> HandlerRegistry {
let mut registry = HandlerRegistry::new(Box::new(StartHandler));
registry.register("start", Box::new(StartHandler));
registry.register("exit", Box::new(ExitHandler));
registry
fn test_llm_spec() -> LlmSpec {
LlmSpec {
model: "test-model".to_string(),
provider: fabro_llm::Provider::Anthropic,
fallback_chain: Vec::new(),
mcp_servers: Vec::new(),
dry_run: true,
}
}
async fn test_db() -> sqlx::SqlitePool {
@ -451,7 +452,7 @@ mod tests {
let db = test_db().await;
let state = create_app_state_with_options(
db,
test_registry,
test_llm_spec,
true,
5,
fabro_workflows::git::GitAuthor::default(),

87
lib/crates/fabro-api/tests/integration.rs
View file

@ -7,25 +7,22 @@
mod mtls_e2e {
use std::path::Path;
use std::process::{Command, Stdio};
use std::sync::Arc;
use fabro_api::jwt_auth::{AuthMode, AuthStrategy};
use fabro_api::server::{build_router, create_app_state};
use fabro_api::server_config::TlsConfig;
use fabro_api::tls::{build_rustls_config, ClientAuth};
use fabro_interview::Interviewer;
use fabro_workflows::handler::agent::AgentHandler;
use fabro_workflows::handler::exit::ExitHandler;
use fabro_workflows::handler::start::StartHandler;
use fabro_workflows::handler::HandlerRegistry;
use fabro_workflows::pipeline::LlmSpec;
use tokio::net::TcpListener;
fn simple_registry(_interviewer: Arc<dyn Interviewer>) -> HandlerRegistry {
let mut registry = HandlerRegistry::new(Box::new(AgentHandler::new(None)));
registry.register("start", Box::new(StartHandler));
registry.register("exit", Box::new(ExitHandler));
registry.register("agent", Box::new(AgentHandler::new(None)));
registry
fn test_llm_spec() -> LlmSpec {
LlmSpec {
model: "test-model".to_string(),
provider: fabro_llm::Provider::Anthropic,
fallback_chain: Vec::new(),
mcp_servers: Vec::new(),
dry_run: true,
}
}
async fn test_db() -> sqlx::SqlitePool {
@ -194,7 +191,7 @@ mod mtls_e2e {
let rustls_config = build_rustls_config(tls_config, client_auth);
let tls_acceptor = tokio_rustls::TlsAcceptor::from(rustls_config);
let state = create_app_state(test_db().await, simple_registry);
let state = create_app_state(test_db().await, test_llm_spec);
let router = build_router(state, auth_mode);
tokio::spawn(async move {
@ -431,21 +428,17 @@ mod server_lifecycle {
use axum::body::Body;
use axum::http::{Request, StatusCode};
use fabro_api::server::{build_router, create_app_state};
use fabro_interview::Interviewer;
use fabro_workflows::handler::agent::AgentHandler;
use fabro_workflows::handler::exit::ExitHandler;
use fabro_workflows::handler::human::HumanHandler;
use fabro_workflows::handler::start::StartHandler;
use fabro_workflows::handler::HandlerRegistry;
use fabro_workflows::pipeline::LlmSpec;
use tower::ServiceExt;
fn gate_registry(interviewer: Arc<dyn Interviewer>) -> HandlerRegistry {
let mut registry = HandlerRegistry::new(Box::new(AgentHandler::new(None)));
registry.register("start", Box::new(StartHandler));
registry.register("exit", Box::new(ExitHandler));
registry.register("agent", Box::new(AgentHandler::new(None)));
registry.register("human", Box::new(HumanHandler::new(interviewer)));
registry
fn test_llm_spec() -> LlmSpec {
LlmSpec {
model: "test-model".to_string(),
provider: fabro_llm::Provider::Anthropic,
fallback_chain: Vec::new(),
mcp_servers: Vec::new(),
dry_run: true,
}
}
async fn body_json(body: Body) -> serde_json::Value {
@ -477,7 +470,7 @@ mod server_lifecycle {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn full_http_lifecycle_approve_and_complete() {
let state = create_app_state(test_db().await, gate_registry);
let state = create_app_state(test_db().await, test_llm_spec);
fabro_api::server::spawn_scheduler(Arc::clone(&state));
let app = build_router(Arc::clone(&state), fabro_api::jwt_auth::AuthMode::Disabled);
@ -573,7 +566,7 @@ mod server_lifecycle {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn full_http_lifecycle_cancel() {
let state = create_app_state(test_db().await, gate_registry);
let state = create_app_state(test_db().await, test_llm_spec);
fabro_api::server::spawn_scheduler(Arc::clone(&state));
let app = build_router(Arc::clone(&state), fabro_api::jwt_auth::AuthMode::Disabled);
@ -627,20 +620,18 @@ mod sse_events {
use axum::body::Body;
use axum::http::{Request, StatusCode};
use fabro_api::server::{build_router, create_app_state};
use fabro_interview::Interviewer;
use fabro_workflows::handler::agent::AgentHandler;
use fabro_workflows::handler::exit::ExitHandler;
use fabro_workflows::handler::start::StartHandler;
use fabro_workflows::handler::HandlerRegistry;
use fabro_workflows::pipeline::LlmSpec;
use http_body_util::BodyExt;
use tower::ServiceExt;
fn simple_registry(_interviewer: Arc<dyn Interviewer>) -> HandlerRegistry {
let mut registry = HandlerRegistry::new(Box::new(AgentHandler::new(None)));
registry.register("start", Box::new(StartHandler));
registry.register("exit", Box::new(ExitHandler));
registry.register("agent", Box::new(AgentHandler::new(None)));
registry
fn test_llm_spec() -> LlmSpec {
LlmSpec {
model: "test-model".to_string(),
provider: fabro_llm::Provider::Anthropic,
fallback_chain: Vec::new(),
mcp_servers: Vec::new(),
dry_run: true,
}
}
const SIMPLE_DOT: &str = r#"digraph SSETest {
@ -659,7 +650,7 @@ mod sse_events {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn sse_stream_contains_expected_event_types() {
let state = create_app_state(test_db().await, simple_registry);
let state = create_app_state(test_db().await, test_llm_spec);
fabro_api::server::spawn_scheduler(Arc::clone(&state));
let app = build_router(Arc::clone(&state), fabro_api::jwt_auth::AuthMode::Disabled);
@ -789,8 +780,7 @@ mod serve_dry_run {
use axum::body::Body;
use axum::http::{Request, StatusCode};
use fabro_api::server::{build_router, create_app_state};
use fabro_interview::Interviewer;
use fabro_workflows::handler::default_registry;
use fabro_workflows::pipeline::LlmSpec;
use tower::ServiceExt;
const MINIMAL_DOT: &str = r#"digraph Test {
@ -806,10 +796,19 @@ mod serve_dry_run {
pool
}
fn test_llm_spec() -> LlmSpec {
LlmSpec {
model: "test-model".to_string(),
provider: fabro_llm::Provider::Anthropic,
fallback_chain: Vec::new(),
mcp_servers: Vec::new(),
dry_run: true,
}
}
/// Build the router exactly as `serve_command` does in dry-run mode.
async fn dry_run_app() -> axum::Router {
let factory = |interviewer: Arc<dyn Interviewer>| default_registry(interviewer, || None);
let state = create_app_state(test_db().await, factory);
let state = create_app_state(test_db().await, test_llm_spec);
fabro_api::server::spawn_scheduler(Arc::clone(&state));
build_router(state, fabro_api::jwt_auth::AuthMode::Disabled)
}

21
lib/crates/fabro-api/tests/openapi_conformance.rs
View file

@ -1,7 +1,6 @@
//! Conformance tests: spec ↔ router ↔ Rust struct consistency.
use std::collections::BTreeSet;
use std::sync::Arc;
use axum::body::Body;
use axum::http::{Method, Request, StatusCode};
@ -11,18 +10,18 @@ use fabro_api::server_config::*;
use fabro_config::run::*;
use fabro_config::sandbox::SandboxConfig;
use fabro_hooks::*;
use fabro_interview::Interviewer;
use fabro_sandbox::daytona::*;
use fabro_workflows::handler::exit::ExitHandler;
use fabro_workflows::handler::start::StartHandler;
use fabro_workflows::handler::HandlerRegistry;
use fabro_workflows::pipeline::LlmSpec;
use tower::ServiceExt;
fn test_registry(_interviewer: Arc<dyn Interviewer>) -> HandlerRegistry {
let mut registry = HandlerRegistry::new(Box::new(StartHandler));
registry.register("start", Box::new(StartHandler));
registry.register("exit", Box::new(ExitHandler));
registry
fn test_llm_spec() -> LlmSpec {
LlmSpec {
model: "test-model".to_string(),
provider: fabro_llm::Provider::Anthropic,
fallback_chain: Vec::new(),
mcp_servers: Vec::new(),
dry_run: true,
}
}
async fn test_db() -> sqlx::SqlitePool {
@ -75,7 +74,7 @@ fn methods_for_path_item(item: &openapiv3::PathItem) -> Vec<Method> {
#[tokio::test]
async fn all_spec_routes_are_routable() {
let spec = load_spec();
let state = create_app_state(test_db().await, test_registry);
let state = create_app_state(test_db().await, test_llm_spec);
let app = build_router(state, AuthMode::Disabled);
let mut checked = 0;

22
lib/crates/fabro-api/tests/pagination.rs
View file

@ -1,22 +1,20 @@
//! Tests that paginated list endpoints return `{ data, meta: { has_more } }`.
use std::sync::Arc;
use axum::body::Body;
use axum::http::{Request, StatusCode};
use fabro_api::jwt_auth::AuthMode;
use fabro_api::server::{build_router, create_app_state};
use fabro_interview::Interviewer;
use fabro_workflows::handler::exit::ExitHandler;
use fabro_workflows::handler::start::StartHandler;
use fabro_workflows::handler::HandlerRegistry;
use fabro_workflows::pipeline::LlmSpec;
use tower::ServiceExt;
fn test_registry(_interviewer: Arc<dyn Interviewer>) -> HandlerRegistry {
let mut registry = HandlerRegistry::new(Box::new(StartHandler));
registry.register("start", Box::new(StartHandler));
registry.register("exit", Box::new(ExitHandler));
registry
fn test_llm_spec() -> LlmSpec {
LlmSpec {
model: "test-model".to_string(),
provider: fabro_llm::Provider::Anthropic,
fallback_chain: Vec::new(),
mcp_servers: Vec::new(),
dry_run: true,
}
}
async fn test_db() -> sqlx::SqlitePool {
@ -109,7 +107,7 @@ const ENDPOINTS: &[PaginatedEndpoint] = &[
#[tokio::test]
async fn paginated_endpoints_return_correct_shape() {
let state = create_app_state(test_db().await, test_registry);
let state = create_app_state(test_db().await, test_llm_spec);
let app = build_router(state, AuthMode::Disabled);
for ep in ENDPOINTS {

45
lib/crates/fabro-cli/src/commands/detached_support.rs
View file

@ -1,8 +1,7 @@
use std::io::Write;
use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
use chrono::{SecondsFormat, Utc};
use chrono::Utc;
use fabro_workflows::event::{RunNoticeLevel, WorkflowRunEvent};
use fabro_workflows::outcome::StageStatus;
use fabro_workflows::records::Conclusion;
@ -111,52 +110,12 @@ pub(crate) fn load_run_id(run_dir: &Path) -> Option<String> {
})
}
pub(crate) fn build_event_envelope(event: &WorkflowRunEvent, run_id: &str) -> serde_json::Value {
let (event_name, event_fields) = fabro_workflows::event::flatten_event(event);
let mut envelope = serde_json::Map::new();
envelope.insert(
"ts".to_string(),
serde_json::Value::String(Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true)),
);
envelope.insert(
"run_id".to_string(),
serde_json::Value::String(run_id.to_string()),
);
envelope.insert("event".to_string(), serde_json::Value::String(event_name));
for (k, v) in event_fields {
if k != "ts" && k != "run_id" && k != "event" {
envelope.insert(k, v);
}
}
serde_json::Value::Object(envelope)
}
pub(crate) fn append_progress_event(
run_dir: &Path,
run_id: &str,
event: &WorkflowRunEvent,
) -> Result<()> {
let envelope = build_event_envelope(event, run_id);
let line = serde_json::to_string(&envelope)?;
let line = fabro_util::redact::redact_jsonl_line(&line);
let mut file = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(run_dir.join("progress.jsonl"))
.with_context(|| {
format!(
"Failed to open {}",
run_dir.join("progress.jsonl").display()
)
})?;
writeln!(file, "{line}")?;
let pretty = serde_json::to_string_pretty(&envelope)?;
let pretty = fabro_util::redact::redact_jsonl_line(&pretty);
std::fs::write(run_dir.join("live.json"), pretty)
.with_context(|| format!("Failed to write {}", run_dir.join("live.json").display()))?;
Ok(())
fabro_workflows::event::append_progress_event(run_dir, run_id, event)
}
pub(crate) fn append_run_notice(

751
lib/crates/fabro-cli/src/commands/run.rs
View file

@ -7,23 +7,18 @@ use std::time::Instant;
use anyhow::{bail, Context};
use chrono::Local;
use clap::{Args, ValueEnum};
use fabro_agent::{
DockerSandbox, DockerSandboxConfig, LocalSandbox, Sandbox, WorktreeConfig, WorktreeSandbox,
};
use fabro_agent::{DockerSandbox, DockerSandboxConfig, LocalSandbox, Sandbox};
use fabro_config::config::FabroConfig;
use fabro_config::{project as project_config, run as run_config, sandbox as sandbox_config};
use fabro_interview::{AutoApproveInterviewer, ConsoleInterviewer, FileInterviewer, Interviewer};
use fabro_model::{Catalog, FallbackTarget, Provider};
use fabro_sandbox::SandboxProvider;
use fabro_util::terminal::Styles;
use fabro_workflows::devcontainer_bridge;
use fabro_workflows::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent};
use fabro_workflows::event::EventEmitter;
use fabro_workflows::git::GitSyncStatus;
use fabro_workflows::handler::default_registry;
use fabro_workflows::handler::llm::{AgentApiBackend, AgentCliBackend, BackendRouter};
use fabro_workflows::operations::{
resume as operations_resume, start, StartFinalizeOptions, StartOptions, StartPullRequestConfig,
StartRetroOptions,
resume as operations_resume, start, DevcontainerSpec, LlmSpec, SandboxEnvSpec, SandboxSpec,
StartFinalizeOptions, StartOptions, StartPullRequestConfig, StartRetroOptions,
};
use fabro_workflows::outcome::StageStatus;
use fabro_workflows::outcome::{compute_stage_cost, format_cost};
@ -31,12 +26,12 @@ use fabro_workflows::pipeline::{
build_conclusion, classify_engine_result, persist_terminal_outcome, Persisted, Validated,
};
use fabro_workflows::records::Checkpoint;
use fabro_workflows::run_options::{GitCheckpointOptions, LifecycleOptions};
use fabro_workflows::run_options::LifecycleOptions;
use indicatif::HumanDuration;
use std::time::Duration;
use tracing::debug;
use super::detached_support::{self, DetachedRunBootstrapGuard, DetachedRunCompletionGuard};
use super::detached_support::{DetachedRunBootstrapGuard, DetachedRunCompletionGuard};
use super::run_progress;
use crate::commands::shared::{
format_tokens_human, print_diagnostics, read_workflow_file, relative_path, tilde_path,
@ -427,17 +422,6 @@ pub(crate) async fn mint_github_token(
Ok(token)
}
/// How the workflow run's working directory is set up.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum WorkdirStrategy {
/// Run directly in the current working directory.
LocalDirectory,
/// Create a local git worktree for isolation.
LocalWorktree,
/// Remote sandbox clones from origin (Daytona, Exe, SSH).
Cloud,
}
/// Accumulates token usage and cost across all workflow stages.
#[derive(Default)]
pub(crate) struct CostAccumulator {
@ -450,18 +434,6 @@ pub(crate) struct CostAccumulator {
pub has_pricing: bool,
}
/// Create a [`LocalSandbox`] wired to emit [`WorkflowRunEvent::Sandbox`] events.
pub(crate) fn local_sandbox_with_callback(
cwd: PathBuf,
emitter: Arc<EventEmitter>,
) -> Arc<dyn Sandbox> {
let mut env = LocalSandbox::new(cwd);
env.set_event_callback(Arc::new(move |event| {
emitter.emit(&fabro_workflows::event::WorkflowRunEvent::Sandbox { event });
}));
Arc::new(env)
}
pub(crate) const RUN_GRAPH_FILE: &str = "workflow.fabro";
pub(crate) const RUN_CONFIG_FILE: &str = "workflow.toml";
@ -1132,39 +1104,6 @@ async fn run_command_impl(
}
});
// JSONL progress log + live.json snapshot
{
let jsonl_path = run_dir.join("progress.jsonl");
let live_path = run_dir.join("live.json");
let run_id = Arc::new(Mutex::new(run_id.clone()));
let run_id_clone = Arc::clone(&run_id);
emitter.on_event(move |event| {
if let fabro_workflows::event::WorkflowRunEvent::WorkflowRunStarted { run_id, .. } =
event
{
*run_id_clone.lock().unwrap() = run_id.clone();
}
let envelope = build_event_envelope(event, &run_id_clone.lock().unwrap());
// Append to progress.jsonl
if let Ok(line) = serde_json::to_string(&envelope) {
let line = fabro_util::redact::redact_jsonl_line(&line);
use std::io::Write;
if let Ok(mut f) = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(&jsonl_path)
{
let _ = writeln!(f, "{line}");
}
}
// Overwrite live.json
if let Ok(pretty) = serde_json::to_string_pretty(&envelope) {
let pretty = fabro_util::redact::redact_jsonl_line(&pretty);
let _ = std::fs::write(&live_path, pretty);
}
});
}
run_progress::ProgressUI::register(&progress_ui, &emitter);
// 4. Build interviewer
@ -1181,479 +1120,13 @@ async fn run_command_impl(
))
};
// Determine the working directory strategy.
// Only the Local provider supports git worktrees on the host.
// Remote sandboxes (Daytona, Exe, SSH) clone from origin inside the sandbox.
// Docker uses the bind-mounted host directory as-is.
// Resume runs skip worktree creation — the engine runs in the original
// working directory and the checkpoint restores logical state.
let workdir_strategy = if resume {
match sandbox_provider {
SandboxProvider::Local | SandboxProvider::Docker => WorkdirStrategy::LocalDirectory,
_ => WorkdirStrategy::Cloud,
}
} else {
match sandbox_provider {
SandboxProvider::Local => {
let worktree_mode = resolve_worktree_mode(run_cfg.as_ref(), &run_defaults);
match worktree_mode {
sandbox_config::WorktreeMode::Always => WorkdirStrategy::LocalWorktree,
sandbox_config::WorktreeMode::Clean => {
if git_status.is_clean() {
WorkdirStrategy::LocalWorktree
} else {
WorkdirStrategy::LocalDirectory
}
}
sandbox_config::WorktreeMode::Dirty => {
if git_status.is_clean() {
WorkdirStrategy::LocalDirectory
} else {
WorkdirStrategy::LocalWorktree
}
}
sandbox_config::WorktreeMode::Never => WorkdirStrategy::LocalDirectory,
}
}
SandboxProvider::Docker => WorkdirStrategy::LocalDirectory,
_ => WorkdirStrategy::Cloud,
}
};
debug!(
?workdir_strategy,
?sandbox_provider,
?git_status,
"Resolved workdir strategy"
);
// Warn about uncommitted changes that won't be available in the execution environment.
if git_status == GitSyncStatus::Dirty {
let env_name = match workdir_strategy {
WorkdirStrategy::LocalWorktree => Some("worktree"),
WorkdirStrategy::Cloud => Some("remote sandbox"),
WorkdirStrategy::LocalDirectory => None,
};
if let Some(env_name) = env_name {
emit_run_notice(
&emitter,
RunNoticeLevel::Warn,
"dirty_worktree",
format!("Uncommitted changes will not be included in the {env_name}."),
);
}
}
// Auto-push when the execution environment needs commits on the remote.
if !dry_run_flag
&& matches!(
workdir_strategy,
WorkdirStrategy::LocalWorktree | WorkdirStrategy::Cloud
)
{
if let Some(ref branch) = detected_base_branch {
// For Synced we know no push is needed; for Unsynced we know it is;
// for Dirty the push status wasn't checked, so check now.
let needs_push = match git_status {
GitSyncStatus::Synced => false,
GitSyncStatus::Unsynced => true,
GitSyncStatus::Dirty => {
let check_repo = original_cwd.clone();
let check_branch = branch.clone();
tokio::task::spawn_blocking(move || {
fabro_workflows::git::branch_needs_push(
&check_repo,
"origin",
&check_branch,
)
})
.await
.unwrap_or(true)
}
};
if needs_push {
let repo_path = original_cwd.clone();
let branch_owned = branch.clone();
let result = fabro_workflows::git::blocking_push_with_timeout(60, move || {
fabro_workflows::git::push_branch(&repo_path, "origin", &branch_owned)
})
.await;
match result {
Ok(()) => {
tracing::info!(%branch, "Pushed current branch to origin");
emit_run_notice(
&emitter,
RunNoticeLevel::Info,
"git_push_succeeded",
format!("{branch} (synced local commits to remote)"),
);
}
Err(e) => {
tracing::warn!(error = %e, %branch, "Failed to push current branch");
emit_run_notice(
&emitter,
RunNoticeLevel::Warn,
"git_push_failed",
format!("Failed to push {branch} to origin: {e}"),
);
}
}
} else {
tracing::info!(%branch, "Branch already in sync with origin, skipping push");
}
}
}
// Compute worktree configuration for local isolation.
// The actual git setup (branch, worktree add, reset) happens inside the sandbox
// creation block for SandboxProvider::Local below.
let (mut worktree_path, mut worktree_branch, mut worktree_base_sha) = if workdir_strategy
== WorkdirStrategy::LocalWorktree
{
match fabro_workflows::git::head_sha(&original_cwd) {
Ok(base_sha) => {
let branch_name = format!("{}{run_id}", fabro_workflows::git::RUN_BRANCH_PREFIX);
let wt_path = run_dir.join("worktree");
(Some(wt_path), Some(branch_name), Some(base_sha))
}
Err(e) => {
emit_run_notice(
&emitter,
RunNoticeLevel::Warn,
"worktree_setup_failed",
format!("Git worktree setup failed ({e}), running without worktree."),
);
(None, None, None)
}
}
} else {
(None, None, None)
};
if let Some(ref wt) = worktree_path {
progress_ui
.lock()
.expect("progress lock poisoned")
.show_worktree(wt);
}
// Show base SHA for both worktree and cloud strategies.
let base_sha_display = worktree_base_sha.clone().or_else(|| {
if workdir_strategy == WorkdirStrategy::Cloud {
fabro_workflows::git::head_sha(&original_cwd).ok()
} else {
None
}
});
if let Some(ref sha) = base_sha_display {
progress_ui
.lock()
.expect("progress lock poisoned")
.show_base_info(detected_base_branch.as_deref(), sha);
}
let cwd = std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."));
let mut daytona_config = resolve_daytona_config(run_cfg.as_ref(), &run_defaults);
let daytona_config = resolve_daytona_config(run_cfg.as_ref(), &run_defaults);
#[cfg(feature = "exedev")]
let exe_config = resolve_exe_config(run_cfg.as_ref(), &run_defaults);
let ssh_config = resolve_ssh_config(run_cfg.as_ref(), &run_defaults);
// Resolve devcontainer if enabled
let devcontainer_config = if run_cfg
.as_ref()
.and_then(|c| c.sandbox.as_ref())
.or(run_defaults.sandbox.as_ref())
.and_then(|s| s.devcontainer)
.unwrap_or(false)
{
match fabro_devcontainer::DevcontainerResolver::resolve(&cwd).await {
Ok(dc) => {
let lifecycle_command_count = dc.on_create_commands.len()
+ dc.post_create_commands.len()
+ dc.post_start_commands.len();
emitter.emit(
&fabro_workflows::event::WorkflowRunEvent::DevcontainerResolved {
dockerfile_lines: dc.dockerfile.lines().count(),
environment_count: dc.environment.len(),
lifecycle_command_count,
workspace_folder: dc.workspace_folder.clone(),
},
);
// Override daytona_config with devcontainer dockerfile
let snapshot = devcontainer_bridge::devcontainer_to_snapshot_config(&dc);
let mut cfg = daytona_config.unwrap_or_default();
cfg.snapshot = Some(snapshot);
daytona_config = Some(cfg);
// Run initialize_commands on host
let timeout = std::time::Duration::from_millis(300_000);
for cmd in &dc.initialize_commands {
let shell_cmds = match cmd {
fabro_devcontainer::Command::Shell(s) => vec![s.clone()],
fabro_devcontainer::Command::Args(args) => {
vec![args
.iter()
.map(|a| {
shlex::try_quote(a).unwrap_or_else(|_| a.into()).to_string()
})
.collect::<Vec<_>>()
.join(" ")]
}
fabro_devcontainer::Command::Parallel(map) => {
map.values().cloned().collect()
}
};
for shell_cmd in &shell_cmds {
let fut = tokio::process::Command::new("sh")
.arg("-c")
.arg(shell_cmd)
.current_dir(&cwd)
.output();
let output = tokio::time::timeout(timeout, fut)
.await
.with_context(|| {
format!("Devcontainer initializeCommand timed out: {shell_cmd}")
})?
.with_context(|| {
format!(
"Failed to execute devcontainer initializeCommand: {shell_cmd}"
)
})?;
if !output.status.success() {
let code = output
.status
.code()
.map_or("unknown".to_string(), |c| c.to_string());
let stderr = String::from_utf8_lossy(&output.stderr);
bail!(
"Devcontainer initializeCommand failed (exit code {code}): {shell_cmd}\n{stderr}"
);
}
}
}
Some(dc)
}
Err(e) => {
bail!("Failed to resolve devcontainer: {e}");
}
}
} else {
None
};
// Deferred sandbox reference — filled after sandbox creation, consumed by event listeners.
let deferred_sandbox: Arc<Mutex<Option<Arc<dyn Sandbox>>>> = Arc::new(Mutex::new(None));
// Register SandboxInitialized listener (must happen before emitter is wrapped in Arc)
{
let run_dir_for_listener = run_dir.clone();
let progress_for_listener = Arc::clone(&progress_ui);
let cwd_for_listener = cwd.to_string_lossy().to_string();
let ssh_data_host = ssh_config.as_ref().map(|c| c.destination.clone());
let deferred_sb = Arc::clone(&deferred_sandbox);
let provider = sandbox_provider; // Copy — captured by move closure
emitter.on_event(move |event| {
if let fabro_workflows::event::WorkflowRunEvent::SandboxInitialized {
working_directory,
} = event
{
progress_for_listener
.lock()
.expect("progress lock poisoned")
.set_working_directory(working_directory.clone());
// Build sandbox record from template
let sandbox_info_opt = deferred_sb.lock().unwrap().as_ref().and_then(|sb| {
let info = sb.sandbox_info();
if info.is_empty() {
None
} else {
Some(info)
}
});
let is_docker = provider == SandboxProvider::Docker;
let record = fabro_sandbox::SandboxRecord {
provider: provider.to_string(),
working_directory: working_directory.clone(),
identifier: sandbox_info_opt,
host_working_directory: if is_docker {
Some(cwd_for_listener.clone())
} else {
None
},
container_mount_point: if is_docker {
Some(working_directory.clone())
} else {
None
},
data_host: if provider == SandboxProvider::Ssh {
ssh_data_host.clone()
} else {
None
},
};
if let Err(e) = record.save(&run_dir_for_listener.join("sandbox.json")) {
tracing::warn!(error = %e, "Failed to save sandbox record");
}
}
});
}
// Wrap emitter in Arc so we can share it with exec env callbacks
let emitter = Arc::new(emitter);
let sandbox: Arc<dyn Sandbox> = match sandbox_provider {
SandboxProvider::Docker => {
let config = DockerSandboxConfig {
host_working_directory: cwd.to_string_lossy().to_string(),
..DockerSandboxConfig::default()
};
let mut env = DockerSandbox::new(config)
.map_err(|e| anyhow::anyhow!("Failed to create Docker environment: {e}"))?;
let emitter_cb = Arc::clone(&emitter);
env.set_event_callback(Arc::new(move |event| {
emitter_cb.emit(&fabro_workflows::event::WorkflowRunEvent::Sandbox { event });
}));
Arc::new(env)
}
SandboxProvider::Daytona => {
let config = daytona_config.clone().unwrap_or_default();
let mut env = fabro_sandbox::daytona::DaytonaSandbox::new(
config,
github_app.clone(),
Some(run_id.clone()),
detected_base_branch.clone(),
)
.await
.map_err(|e| anyhow::anyhow!("{e}"))?;
let emitter_cb = Arc::clone(&emitter);
env.set_event_callback(Arc::new(move |event| {
emitter_cb.emit(&fabro_workflows::event::WorkflowRunEvent::Sandbox { event });
}));
Arc::new(env)
}
#[cfg(feature = "exedev")]
SandboxProvider::Exe => {
let clone_params = resolve_exe_clone_params(&original_cwd);
let mgmt_ssh = fabro_sandbox::exe::OpensshRunner::connect_raw("exe.dev")
.await
.map_err(|e| anyhow::anyhow!("Failed to connect to exe.dev: {e}"))?;
let config = exe_config.unwrap_or_default();
let mut env = fabro_sandbox::exe::ExeSandbox::new(
Box::new(mgmt_ssh),
config,
clone_params,
Some(run_id.clone()),
github_app.clone(),
);
let emitter_cb = Arc::clone(&emitter);
env.set_event_callback(Arc::new(move |event| {
emitter_cb.emit(&fabro_workflows::event::WorkflowRunEvent::Sandbox { event });
}));
Arc::new(env)
}
#[cfg(not(feature = "exedev"))]
SandboxProvider::Exe => {
bail!("exe sandbox requires the exedev feature");
}
SandboxProvider::Ssh => {
let config = ssh_config
.clone()
.ok_or_else(|| anyhow::anyhow!("--sandbox ssh requires [sandbox.ssh] config"))?;
let clone_params = resolve_ssh_clone_params(&original_cwd);
let mut env = fabro_sandbox::ssh::SshSandbox::new(
config,
clone_params,
Some(run_id.clone()),
github_app.clone(),
);
let emitter_cb = Arc::clone(&emitter);
env.set_event_callback(Arc::new(move |event| {
emitter_cb.emit(&fabro_workflows::event::WorkflowRunEvent::Sandbox { event });
}));
Arc::new(env)
}
SandboxProvider::Local => {
if let (Some(base_sha), Some(branch_name), Some(wt_path)) = (
worktree_base_sha.as_ref(),
worktree_branch.as_ref(),
worktree_path.as_ref(),
) {
// Set up a WorktreeSandbox for git-isolated local execution.
let wt_path_str = wt_path.to_string_lossy().into_owned();
let inner = local_sandbox_with_callback(original_cwd.clone(), Arc::clone(&emitter));
let wt_config = WorktreeConfig {
branch_name: branch_name.clone(),
base_sha: base_sha.clone(),
worktree_path: wt_path_str,
skip_branch_creation: false,
};
let mut wt_sandbox = WorktreeSandbox::new(inner, wt_config);
wt_sandbox.set_event_callback(Arc::clone(&emitter).worktree_callback());
match wt_sandbox.initialize().await {
Ok(()) => {
std::env::set_current_dir(wt_path)?;
Arc::new(wt_sandbox) as Arc<dyn Sandbox>
}
Err(e) => {
emit_run_notice(
&emitter,
RunNoticeLevel::Warn,
"worktree_setup_failed",
format!("Git worktree setup failed ({e}), running without worktree."),
);
// Reset so RunOptions does not enable git checkpointing
worktree_path = None;
worktree_branch = None;
worktree_base_sha = None;
local_sandbox_with_callback(cwd.clone(), Arc::clone(&emitter))
}
}
} else {
local_sandbox_with_callback(cwd.clone(), Arc::clone(&emitter))
}
}
};
// Wrap with ReadBeforeWriteSandbox to enforce read-before-write guard
// (delegate_sandbox! macro delegates initialize/cleanup)
let sandbox: Arc<dyn Sandbox> = Arc::new(fabro_agent::ReadBeforeWriteSandbox::new(sandbox));
// Fill deferred sandbox reference for event listeners registered above
*deferred_sandbox.lock().unwrap() = Some(Arc::clone(&sandbox));
// 6. Resolve backend, model, and provider
let (dry_run_mode, llm_client) = if dry_run_flag {
(true, None)
} else {
match fabro_llm::client::Client::from_env().await {
Ok(c) if c.provider_names().is_empty() => {
emit_run_notice(
&emitter,
RunNoticeLevel::Warn,
"dry_run_no_llm",
"No LLM providers configured. Running in dry-run mode.",
);
(true, None)
}
Ok(c) => (false, Some(c)),
Err(e) => {
emit_run_notice(
&emitter,
RunNoticeLevel::Warn,
"dry_run_llm_init_failed",
format!("Failed to initialize LLM client: {e}. Running in dry-run mode."),
);
(true, None)
}
}
};
// Parse provider string to enum (defaults to best available from env)
let provider_enum: Provider = provider
.as_deref()
@ -1662,62 +1135,7 @@ async fn run_command_impl(
.map_err(|e| anyhow::anyhow!("{e}"))?
.unwrap_or_else(Provider::default_from_env);
// Resolve fallback chain from config
let fallback_chain = resolve_fallback_chain(provider_enum, &model, run_cfg.as_ref());
// 7. Build engine
// Devcontainer env is layered underneath TOML env (TOML wins on conflict)
let sandbox_env: HashMap<String, String> = {
let mut env = if let Some(ref dc) = devcontainer_config {
dc.environment.clone()
} else {
HashMap::new()
};
if let Some(mut toml_env) = run_cfg
.as_ref()
.and_then(|c| c.sandbox.as_ref())
.or(run_defaults.sandbox.as_ref())
.and_then(|s| s.env.clone())
{
// When falling back to run_defaults (run_cfg is None, i.e. bare .fabro
// workflow), env refs haven't been resolved yet — resolve them now.
if run_cfg.is_none() {
run_config::resolve_env_refs(&mut toml_env)?;
}
env.extend(toml_env);
}
env
};
// Mint a GitHub App IAT and inject as GITHUB_TOKEN if [github] permissions are declared
let mut sandbox_env = sandbox_env;
let github_permissions = run_cfg
.as_ref()
.and_then(|c| c.github.as_ref())
.or(run_defaults.github.as_ref());
if let Some(gh_cfg) = github_permissions {
if !gh_cfg.permissions.is_empty() {
if let (Some(ref creds), Some(ref url)) = (&github_app, &origin_url) {
match mint_github_token(creds, url, &gh_cfg.permissions).await {
Ok(token) => {
debug!("Minted GitHub IAT for sandbox GITHUB_TOKEN");
sandbox_env.insert("GITHUB_TOKEN".to_string(), token);
}
Err(e) => {
emit_run_notice(
&emitter,
RunNoticeLevel::Warn,
"github_token_failed",
format!("Failed to mint GitHub token: {e}"),
);
}
}
} else {
debug!("Skipping GitHub token: no GitHub App credentials or origin URL");
}
}
}
let mcp_servers: Vec<fabro_mcp::config::McpServerConfig> = {
let servers = run_cfg
.as_ref()
@ -1733,66 +1151,102 @@ async fn run_command_impl(
)
.collect()
};
let registry = default_registry(interviewer.clone(), {
let sandbox_env = sandbox_env.clone();
let model = model.clone();
let mcp_servers = mcp_servers.clone();
move || {
if dry_run_mode {
None
} else {
let api =
AgentApiBackend::new(model.clone(), provider_enum, fallback_chain.clone())
.with_env(sandbox_env.clone())
.with_mcp_servers(mcp_servers.clone());
let cli = AgentCliBackend::new(model.clone(), provider_enum)
.with_env(sandbox_env.clone());
Some(Box::new(BackendRouter::new(Box::new(api), cli)))
}
let sandbox_spec = match sandbox_provider {
SandboxProvider::Local => SandboxSpec::Local {
working_directory: cwd.clone(),
},
SandboxProvider::Docker => SandboxSpec::Docker {
config: DockerSandboxConfig {
host_working_directory: cwd.to_string_lossy().to_string(),
..DockerSandboxConfig::default()
},
},
SandboxProvider::Daytona => SandboxSpec::Daytona {
config: daytona_config.unwrap_or_default(),
github_app: github_app.clone(),
run_id: Some(run_id.clone()),
clone_branch: detected_base_branch.clone(),
},
#[cfg(feature = "exedev")]
SandboxProvider::Exe => SandboxSpec::Exe {
config: exe_config.unwrap_or_default(),
clone_params: resolve_exe_clone_params(&original_cwd),
run_id: Some(run_id.clone()),
github_app: github_app.clone(),
mgmt_destination: "exe.dev".to_string(),
},
#[cfg(not(feature = "exedev"))]
SandboxProvider::Exe => {
bail!("exe sandbox requires the exedev feature");
}
});
// 7. Execute
// Set up metadata branch for git checkpointing (host or remote — engine fills remote)
let git = if worktree_path.is_some() {
Some(GitCheckpointOptions {
base_sha: worktree_base_sha,
run_branch: worktree_branch,
meta_branch: Some(fabro_workflows::git::MetadataStore::branch_name(&run_id)),
})
} else {
None
SandboxProvider::Ssh => SandboxSpec::Ssh {
config: ssh_config
.clone()
.ok_or_else(|| anyhow::anyhow!("--sandbox ssh requires [sandbox.ssh] config"))?,
clone_params: resolve_ssh_clone_params(&original_cwd),
run_id: Some(run_id.clone()),
github_app: github_app.clone(),
},
};
// Build lifecycle config for sandbox init, setup commands, and devcontainer phases
let toml_env = if let Some(mut env) = run_cfg
.as_ref()
.and_then(|c| c.sandbox.as_ref())
.or(run_defaults.sandbox.as_ref())
.and_then(|s| s.env.clone())
{
if run_cfg.is_none() {
run_config::resolve_env_refs(&mut env)?;
}
env
} else {
HashMap::new()
};
let sandbox_env = SandboxEnvSpec {
devcontainer_env: HashMap::new(),
toml_env,
github_permissions: run_cfg
.as_ref()
.and_then(|c| c.github.as_ref())
.or(run_defaults.github.as_ref())
.and_then(|cfg| (!cfg.permissions.is_empty()).then(|| cfg.permissions.clone())),
origin_url: origin_url.clone(),
};
let devcontainer_enabled = run_cfg
.as_ref()
.and_then(|c| c.sandbox.as_ref())
.or(run_defaults.sandbox.as_ref())
.and_then(|s| s.devcontainer)
.unwrap_or(false);
let llm = LlmSpec {
model: model.clone(),
provider: provider_enum,
fallback_chain,
mcp_servers,
dry_run: dry_run_flag,
};
let worktree_mode = resolve_worktree_mode(run_cfg.as_ref(), &run_defaults);
let lifecycle = LifecycleOptions {
setup_commands,
setup_command_timeout_ms: 300_000,
devcontainer_phases: if let Some(ref dc) = devcontainer_config {
vec![
("on_create".to_string(), dc.on_create_commands.clone()),
("post_create".to_string(), dc.post_create_commands.clone()),
("post_start".to_string(), dc.post_start_commands.clone()),
]
} else {
Vec::new()
},
devcontainer_phases: Vec::new(),
};
// Defuse the bootstrap guard — engine.run() has taken ownership of lifecycle status.
status_guard.defuse();
let run_start = Instant::now();
let pr_config = if dry_run_mode {
None
} else {
persisted.run_record().config.pull_request.clone()
};
let pr_config = persisted.run_record().config.pull_request.clone();
let start_options = StartOptions {
cancel_token: None,
emitter: Arc::clone(&emitter),
sandbox: Arc::clone(&sandbox),
registry: Arc::new(registry),
sandbox: sandbox_spec,
llm,
interviewer: interviewer.clone(),
lifecycle,
hooks: fabro_hooks::HookConfig {
hooks: run_cfg
@ -1801,17 +1255,18 @@ async fn run_command_impl(
.unwrap_or_else(|| run_defaults.hooks.clone()),
},
sandbox_env,
devcontainer: devcontainer_enabled.then(|| DevcontainerSpec {
enabled: true,
resolve_dir: cwd.clone(),
}),
seed_context: None,
git_author,
git,
git: None,
github_app: github_app.clone(),
dry_run: dry_run_mode,
worktree_mode: Some(worktree_mode),
dry_run: dry_run_flag,
retro: StartRetroOptions {
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: StartFinalizeOptions { preserve_sandbox },
pull_request: StartPullRequestConfig {
@ -1882,19 +1337,6 @@ async fn run_command_impl(
}
}
pub(crate) fn emit_run_notice(
emitter: &EventEmitter,
level: RunNoticeLevel,
code: impl Into<String>,
message: impl Into<String>,
) {
emitter.emit(&WorkflowRunEvent::RunNotice {
level,
code: code.into(),
message: message.into(),
});
}
/// Print a summary of the completed run from `conclusion.json` and `pull_request.json`.
///
/// Used by the unified create+start+attach path in `main.rs` to display
@ -2468,11 +1910,12 @@ async fn run_preflight(
}
}
#[cfg(test)]
pub(crate) fn build_event_envelope(
event: &fabro_workflows::event::WorkflowRunEvent,
run_id: &str,
) -> serde_json::Value {
detached_support::build_event_envelope(event, run_id)
fabro_workflows::event::build_event_envelope(event, run_id)
}
#[cfg(test)]

21
lib/crates/fabro-cli/src/commands/run_progress.rs
View file

@ -311,6 +311,19 @@ impl ProgressUI {
pub(crate) fn handle_event(&mut self, event: &WorkflowRunEvent) {
match event {
WorkflowRunEvent::WorkflowRunStarted {
base_branch,
base_sha,
worktree_dir,
..
} => {
if let Some(worktree_dir) = worktree_dir {
self.show_worktree(std::path::Path::new(worktree_dir));
}
if let Some(base_sha) = base_sha {
self.show_base_info(base_branch.as_deref(), base_sha);
}
}
WorkflowRunEvent::Sandbox {
event: sandbox_event,
} => {
@ -662,6 +675,14 @@ impl ProgressUI {
|key: &str| -> u64 { envelope.get(key).and_then(|v| v.as_u64()).unwrap_or(0) };
match event_name {
"WorkflowRunStarted" => {
if let Some(worktree_dir) = str_field("worktree_dir") {
self.show_worktree(std::path::Path::new(worktree_dir));
}
if let Some(base_sha) = str_field("base_sha") {
self.show_base_info(str_field("base_branch"), base_sha);
}
}
"Sandbox.Initializing" => {
let provider = str_field("sandbox_provider")
.unwrap_or("unknown")

85
lib/crates/fabro-workflows/src/event.rs
View file

@ -1,6 +1,10 @@
use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicI64, Ordering};
use std::sync::Arc;
use anyhow::{Context, Result};
use chrono::{SecondsFormat, Utc};
use serde::{Deserialize, Serialize};
use crate::outcome::StageUsage;
@ -21,6 +25,8 @@ pub enum WorkflowRunEvent {
name: String,
run_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
base_branch: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
base_sha: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
run_branch: Option<String>,
@ -791,6 +797,80 @@ pub fn flatten_event(
(event_name, fields)
}
pub fn build_event_envelope(event: &WorkflowRunEvent, run_id: &str) -> serde_json::Value {
let (event_name, event_fields) = flatten_event(event);
let mut envelope = serde_json::Map::new();
envelope.insert(
"ts".to_string(),
serde_json::Value::String(Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true)),
);
envelope.insert(
"run_id".to_string(),
serde_json::Value::String(run_id.to_string()),
);
envelope.insert("event".to_string(), serde_json::Value::String(event_name));
for (k, v) in event_fields {
if k != "ts" && k != "run_id" && k != "event" {
envelope.insert(k, v);
}
}
serde_json::Value::Object(envelope)
}
pub fn append_progress_event(run_dir: &Path, run_id: &str, event: &WorkflowRunEvent) -> Result<()> {
let envelope = build_event_envelope(event, run_id);
let line = serde_json::to_string(&envelope)?;
let line = fabro_util::redact::redact_jsonl_line(&line);
let mut file = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(run_dir.join("progress.jsonl"))
.with_context(|| {
format!(
"Failed to open {}",
run_dir.join("progress.jsonl").display()
)
})?;
writeln!(file, "{line}")?;
let pretty = serde_json::to_string_pretty(&envelope)?;
let pretty = fabro_util::redact::redact_jsonl_line(&pretty);
std::fs::write(run_dir.join("live.json"), pretty)
.with_context(|| format!("Failed to write {}", run_dir.join("live.json").display()))?;
Ok(())
}
pub struct ProgressLogger {
run_dir: PathBuf,
run_id: String,
}
impl ProgressLogger {
#[must_use]
pub fn new(run_dir: impl Into<PathBuf>, run_id: impl Into<String>) -> Self {
Self {
run_dir: run_dir.into(),
run_id: run_id.into(),
}
}
pub fn register(self, emitter: &EventEmitter) {
let run_dir = self.run_dir;
let run_id = Arc::new(std::sync::Mutex::new(self.run_id));
emitter.on_event(move |event| {
if let WorkflowRunEvent::WorkflowRunStarted {
run_id: started_run_id,
..
} = event
{
*run_id.lock().unwrap() = started_run_id.clone();
}
let _ = append_progress_event(&run_dir, &run_id.lock().unwrap(), event);
});
}
}
fn flatten_agent(inner: serde_json::Value) -> (String, serde_json::Map<String, serde_json::Value>) {
let serde_json::Value::Object(mut agent_fields) = inner else {
return ("Agent".to_string(), serde_json::Map::new());
@ -1136,6 +1216,7 @@ mod tests {
emitter.emit(&WorkflowRunEvent::WorkflowRunStarted {
name: "test".to_string(),
run_id: "1".to_string(),
base_branch: None,
base_sha: None,
run_branch: None,
worktree_dir: None,
@ -1636,6 +1717,7 @@ mod tests {
emitter.emit(&WorkflowRunEvent::WorkflowRunStarted {
name: "test".to_string(),
run_id: "1".to_string(),
base_branch: None,
base_sha: None,
run_branch: None,
worktree_dir: None,
@ -1823,6 +1905,7 @@ mod tests {
let event = WorkflowRunEvent::WorkflowRunStarted {
name: "my_pipeline".to_string(),
run_id: "r1".to_string(),
base_branch: None,
base_sha: None,
run_branch: None,
worktree_dir: None,
@ -2447,6 +2530,7 @@ mod tests {
let event = WorkflowRunEvent::WorkflowRunStarted {
name: "my_workflow".to_string(),
run_id: "r42".to_string(),
base_branch: None,
base_sha: None,
run_branch: None,
worktree_dir: None,
@ -2476,6 +2560,7 @@ mod tests {
let event = WorkflowRunEvent::WorkflowRunStarted {
name: "wf".to_string(),
run_id: "r1".to_string(),
base_branch: None,
base_sha: None,
run_branch: None,
worktree_dir: None,

4
lib/crates/fabro-workflows/src/handler/manager_loop.rs
View file

@ -157,6 +157,7 @@ impl Handler for SubWorkflowHandler {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -193,6 +194,9 @@ impl Handler for SubWorkflowHandler {
hook_runner,
env,
dry_run,
llm_client: None,
model: String::new(),
provider: fabro_llm::Provider::Anthropic,
};
let executed = pipeline::execute(initialized).await;
Ok::<_, FabroError>((executed.outcome?, executed.final_context))

2
lib/crates/fabro-workflows/src/lifecycle/event.rs
View file

@ -36,6 +36,7 @@ pub struct EventLifecycle {
/// Set in on_edge_selected when loop_restart approved; emitted+cleared in on_run_start.
pub restarted_from: Arc<Mutex<Option<(String, String)>>>,
// Config for WorkflowRunStarted payload
pub base_branch: Option<String>,
pub base_sha: Option<String>,
pub run_branch: Option<String>,
pub worktree_dir: Option<String>,
@ -70,6 +71,7 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
self.emitter.emit(&WorkflowRunEvent::WorkflowRunStarted {
name: self.graph_name.clone(),
run_id: self.run_id.clone(),
base_branch: self.base_branch.clone(),
base_sha: self.base_sha.clone(),
run_branch: self.run_branch.clone(),
worktree_dir: self.worktree_dir.clone(),

3
lib/crates/fabro-workflows/src/lifecycle/mod.rs
View file

@ -109,7 +109,8 @@ impl WorkflowLifecycle {
run_id: run_options.run_id.clone(),
run_start: Mutex::new(Instant::now()),
restarted_from: Arc::clone(&restarted_from),
base_sha: run_options.git.as_ref().and_then(|g| g.base_sha.clone()),
base_branch: run_options.base_branch.clone(),
base_sha: run_options.display_base_sha.clone(),
run_branch: run_options.git.as_ref().and_then(|g| g.run_branch.clone()),
worktree_dir: working_directory.clone(),
goal: (!graph.goal().is_empty()).then(|| graph.goal().to_string()),

1
lib/crates/fabro-workflows/src/operations/mod.rs
View file

@ -3,6 +3,7 @@ mod fork;
mod rewind;
mod start;
pub use crate::pipeline::{DevcontainerSpec, LlmSpec, SandboxEnvSpec, SandboxSpec};
pub use create::{
create, create_from_file, default_run_dir, validate, validate_from_file, RunCreateOptions,
ValidateOptions,

382
lib/crates/fabro-workflows/src/operations/start.rs
View file

@ -1,24 +1,21 @@
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use crate::context::Context;
use crate::error::FabroError;
use crate::event::{EventEmitter, WorkflowRunEvent};
use crate::handler::HandlerRegistry;
use crate::event::{EventEmitter, ProgressLogger, WorkflowRunEvent};
use crate::outcome::StageStatus;
use crate::pipeline::{
self, FinalizeOptions, Finalized, InitOptions, Persisted, PullRequestOptions, RetroOptions,
self, DevcontainerSpec, FinalizeOptions, Finalized, InitOptions, LlmSpec, Persisted,
PullRequestOptions, RetroOptions, SandboxEnvSpec, SandboxSpec,
};
use crate::records::{Checkpoint, Conclusion};
use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions};
use fabro_config::sandbox::WorktreeMode;
use fabro_interview::Interviewer;
pub struct StartRetroOptions {
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 StartFinalizeOptions {
@ -41,15 +38,20 @@ pub struct StartOptions {
// Truly external (not derivable from RunRecord)
pub cancel_token: Option<std::sync::Arc<std::sync::atomic::AtomicBool>>,
pub emitter: Arc<EventEmitter>,
pub sandbox: Arc<dyn fabro_agent::Sandbox>,
pub registry: Arc<HandlerRegistry>,
pub sandbox: SandboxSpec,
pub llm: LlmSpec,
pub interviewer: Arc<dyn Interviewer>,
pub lifecycle: LifecycleOptions,
pub hooks: fabro_hooks::HookConfig,
pub sandbox_env: HashMap<String, String>,
pub sandbox_env: SandboxEnvSpec,
pub devcontainer: Option<DevcontainerSpec>,
pub seed_context: Option<Context>,
pub git_author: crate::git::GitAuthor,
pub git: Option<GitCheckpointOptions>,
pub github_app: Option<fabro_github::GitHubAppCredentials>,
pub worktree_mode: Option<WorktreeMode>,
#[cfg(test)]
pub registry_override: Option<Arc<crate::handler::HandlerRegistry>>,
// Still external for now — could be derived from RunRecord.config in follow-up
pub dry_run: bool,
@ -114,17 +116,6 @@ async fn run_engine(
options: StartOptions,
) -> Result<Started, FabroError> {
let preserve_sandbox = options.finalize.preserve_sandbox;
let sandbox_for_cleanup = Arc::clone(&options.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;
});
}
});
// Build RunOptions from the persisted RunRecord + external caller options
let record = persisted.run_record();
@ -143,37 +134,64 @@ async fn run_engine(
.as_deref()
.map(std::path::PathBuf::from),
base_branch: record.base_branch.clone(),
git: options.git,
display_base_sha: None,
git: None,
};
let last_git_sha: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
{
let sha_clone = Arc::clone(&last_git_sha);
options.emitter.on_event(move |event| match event {
WorkflowRunEvent::CheckpointCompleted {
git_commit_sha: Some(sha),
..
}
| WorkflowRunEvent::WorkflowRunCompleted {
final_git_commit_sha: Some(sha),
..
}
| WorkflowRunEvent::GitCommit { sha, .. } => {
*sha_clone.lock().unwrap() = Some(sha.clone());
}
_ => {}
});
}
ProgressLogger::new(persisted.run_dir(), record.run_id.clone())
.register(options.emitter.as_ref());
let init_options = InitOptions {
run_id: record.run_id.clone(),
dry_run: options.dry_run,
emitter: options.emitter,
sandbox: options.sandbox,
registry: options.registry,
llm: options.llm,
interviewer: options.interviewer,
lifecycle: options.lifecycle,
run_options,
hooks: options.hooks,
sandbox_env: options.sandbox_env,
devcontainer: options.devcontainer,
git: options.git,
worktree_mode: options.worktree_mode,
#[cfg(test)]
registry_override: options.registry_override,
checkpoint,
seed_context: options.seed_context,
};
let initialized = pipeline::initialize(persisted, init_options).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 sandbox_for_cleanup = Arc::clone(&initialized.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 executed = pipeline::execute(initialized).await;
let failed = !matches!(
@ -191,10 +209,9 @@ async fn run_engine(
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,
llm_client: executed.llm_client.clone(),
provider: executed.provider,
model: executed.model.clone(),
};
let retro_start = Instant::now();
@ -233,23 +250,19 @@ async fn run_engine(
#[cfg(test)]
mod tests {
use std::collections::HashMap;
use std::path::Path;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::atomic::{AtomicBool, Ordering};
use async_trait::async_trait;
use chrono::Utc;
use fabro_agent::{DirEntry, ExecResult, GrepOptions, LocalSandbox, Sandbox};
use fabro_agent::{LocalSandbox, Sandbox};
use fabro_config::config::FabroConfig;
use fabro_graphviz::graph::{Graph, Node};
use tokio_util::sync::CancellationToken;
use super::*;
use crate::context::Context;
use crate::event::EventEmitter;
use crate::handler::exit::ExitHandler;
use crate::handler::start::StartHandler;
use crate::handler::{Handler, HandlerRegistry};
use crate::outcome::Outcome;
use crate::handler::HandlerRegistry;
use crate::pipeline::{LlmSpec, SandboxEnvSpec, SandboxSpec};
use crate::run_options::LifecycleOptions;
const MINIMAL_DOT: &str = r#"digraph Test {
@ -259,202 +272,6 @@ mod tests {
start -> exit
}"#;
const EMIT_DOT: &str = r#"digraph Test {
graph [goal="Ship feature"]
start [shape=Mdiamond]
work [type="emit"]
exit [shape=Msquare]
start -> work -> exit
}"#;
struct CleanupCountingSandbox {
inner: Arc<dyn Sandbox>,
cleanup_count: Arc<AtomicUsize>,
}
#[async_trait]
impl Sandbox for CleanupCountingSandbox {
async fn read_file(
&self,
path: &str,
offset: Option<usize>,
limit: Option<usize>,
) -> Result<String, String> {
self.inner.read_file(path, offset, limit).await
}
async fn write_file(&self, path: &str, content: &str) -> Result<(), String> {
self.inner.write_file(path, content).await
}
async fn delete_file(&self, path: &str) -> Result<(), String> {
self.inner.delete_file(path).await
}
async fn file_exists(&self, path: &str) -> Result<bool, String> {
self.inner.file_exists(path).await
}
async fn list_directory(
&self,
path: &str,
depth: Option<usize>,
) -> Result<Vec<DirEntry>, String> {
self.inner.list_directory(path, depth).await
}
async fn exec_command(
&self,
command: &str,
timeout_ms: u64,
working_dir: Option<&str>,
env_vars: Option<&HashMap<String, String>>,
cancel_token: Option<CancellationToken>,
) -> Result<ExecResult, String> {
self.inner
.exec_command(command, timeout_ms, working_dir, env_vars, cancel_token)
.await
}
async fn grep(
&self,
pattern: &str,
path: &str,
options: &GrepOptions,
) -> Result<Vec<String>, String> {
self.inner.grep(pattern, path, options).await
}
async fn glob(&self, pattern: &str, path: Option<&str>) -> Result<Vec<String>, String> {
self.inner.glob(pattern, path).await
}
async fn download_file_to_local(
&self,
remote_path: &str,
local_path: &Path,
) -> Result<(), String> {
self.inner
.download_file_to_local(remote_path, local_path)
.await
}
async fn upload_file_from_local(
&self,
local_path: &Path,
remote_path: &str,
) -> Result<(), String> {
self.inner
.upload_file_from_local(local_path, remote_path)
.await
}
async fn initialize(&self) -> Result<(), String> {
self.inner.initialize().await
}
async fn cleanup(&self) -> Result<(), String> {
self.cleanup_count.fetch_add(1, Ordering::SeqCst);
self.inner.cleanup().await
}
fn working_directory(&self) -> &str {
self.inner.working_directory()
}
fn platform(&self) -> &str {
self.inner.platform()
}
fn os_version(&self) -> String {
self.inner.os_version()
}
fn sandbox_info(&self) -> String {
self.inner.sandbox_info()
}
async fn refresh_push_credentials(&self) -> Result<(), String> {
self.inner.refresh_push_credentials().await
}
async fn set_autostop_interval(&self, minutes: i32) -> Result<(), String> {
self.inner.set_autostop_interval(minutes).await
}
async fn setup_git_for_run(
&self,
run_id: &str,
) -> Result<Option<fabro_sandbox::GitRunInfo>, String> {
self.inner.setup_git_for_run(run_id).await
}
fn resume_setup_commands(&self, run_branch: &str) -> Vec<String> {
self.inner.resume_setup_commands(run_branch)
}
async fn git_push_branch(&self, branch: &str) -> bool {
self.inner.git_push_branch(branch).await
}
fn host_git_dir(&self) -> Option<&str> {
self.inner.host_git_dir()
}
fn parallel_worktree_path(
&self,
run_dir: &Path,
run_id: &str,
node_id: &str,
key: &str,
) -> String {
self.inner
.parallel_worktree_path(run_dir, run_id, node_id, key)
}
async fn ssh_access_command(&self) -> Result<Option<String>, String> {
self.inner.ssh_access_command().await
}
fn origin_url(&self) -> Option<&str> {
self.inner.origin_url()
}
async fn get_preview_url(
&self,
port: u16,
) -> Result<Option<(String, HashMap<String, String>)>, String> {
self.inner.get_preview_url(port).await
}
fn mark_agent_read(&self, path: &str) {
self.inner.mark_agent_read(path);
}
}
struct EmitCheckpointHandler;
#[async_trait]
impl Handler for EmitCheckpointHandler {
async fn execute(
&self,
node: &Node,
_context: &Context,
_graph: &Graph,
_run_dir: &Path,
services: &crate::handler::EngineServices,
) -> Result<Outcome, FabroError> {
services
.emitter
.emit(&WorkflowRunEvent::CheckpointCompleted {
node_id: node.id.clone(),
status: "success".to_string(),
git_commit_sha: Some("sha-test".to_string()),
});
Ok(Outcome::success())
}
}
fn persisted_workflow(dot: &str, run_dir: &std::path::Path) -> Persisted {
crate::operations::create(
dot,
@ -478,13 +295,12 @@ mod tests {
let mut registry = HandlerRegistry::new(Box::new(StartHandler));
registry.register("start", Box::new(StartHandler));
registry.register("exit", Box::new(ExitHandler));
registry.register("emit", Box::new(EmitCheckpointHandler));
registry
}
fn test_start_options(
_run_dir: &std::path::Path,
sandbox: Arc<dyn Sandbox>,
_sandbox: Arc<dyn Sandbox>,
emitter: Arc<EventEmitter>,
registry: Arc<HandlerRegistry>,
lifecycle: LifecycleOptions,
@ -493,23 +309,34 @@ mod tests {
StartOptions {
cancel_token: None,
emitter,
sandbox,
registry,
sandbox: SandboxSpec::Local {
working_directory: std::env::current_dir().unwrap(),
},
llm: LlmSpec {
model: "test-model".to_string(),
provider: fabro_llm::Provider::Anthropic,
fallback_chain: Vec::new(),
mcp_servers: Vec::new(),
dry_run: true,
},
interviewer: Arc::new(fabro_interview::AutoApproveInterviewer),
lifecycle,
hooks: fabro_hooks::HookConfig { hooks: vec![] },
sandbox_env: HashMap::new(),
sandbox_env: SandboxEnvSpec {
devcontainer_env: HashMap::new(),
toml_env: HashMap::new(),
github_permissions: None,
origin_url: None,
},
devcontainer: None,
seed_context: None,
git_author: crate::git::GitAuthor::default(),
git: None,
github_app: None,
worktree_mode: None,
registry_override: Some(registry),
dry_run: false,
retro: StartRetroOptions {
enabled: false,
dry_run: false,
llm_client: None,
provider: fabro_llm::Provider::Anthropic,
model: "test-model".to_string(),
},
retro: StartRetroOptions { enabled: false },
finalize: StartFinalizeOptions { preserve_sandbox },
pull_request: StartPullRequestConfig {
pr_config: None,
@ -520,25 +347,14 @@ mod tests {
}
}
fn counting_sandbox() -> (Arc<dyn Sandbox>, Arc<AtomicUsize>) {
let cleanup_count = Arc::new(AtomicUsize::new(0));
let inner: Arc<dyn Sandbox> = Arc::new(LocalSandbox::new(std::env::current_dir().unwrap()));
(
Arc::new(CleanupCountingSandbox {
inner,
cleanup_count: Arc::clone(&cleanup_count),
}),
cleanup_count,
)
}
#[tokio::test]
async fn start_cleans_up_sandbox_when_initialize_fails() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
let emitter = Arc::new(EventEmitter::new());
let registry = Arc::new(test_registry());
let (sandbox, cleanup_count) = counting_sandbox();
let sandbox: Arc<dyn Sandbox> =
Arc::new(LocalSandbox::new(std::env::current_dir().unwrap()));
persisted_workflow(MINIMAL_DOT, &run_dir);
let result = start(
@ -559,9 +375,6 @@ mod tests {
.await;
assert!(result.is_err());
tokio::task::yield_now().await;
tokio::time::sleep(Duration::from_millis(20)).await;
assert_eq!(cleanup_count.load(Ordering::SeqCst), 1);
}
#[tokio::test]
@ -572,8 +385,29 @@ mod tests {
let registry = Arc::new(test_registry());
let sandbox: Arc<dyn Sandbox> =
Arc::new(LocalSandbox::new(std::env::current_dir().unwrap()));
let injected = Arc::new(AtomicBool::new(false));
persisted_workflow(EMIT_DOT, &run_dir);
{
let injected = Arc::clone(&injected);
let emitter_for_injection = Arc::clone(&emitter);
emitter.on_event(move |event| {
if injected.load(Ordering::SeqCst) {
return;
}
if let WorkflowRunEvent::StageStarted { node_id, .. } = event {
if node_id == "start" {
injected.store(true, Ordering::SeqCst);
emitter_for_injection.emit(&WorkflowRunEvent::CheckpointCompleted {
node_id: node_id.clone(),
status: "success".to_string(),
git_commit_sha: Some("sha-test".to_string()),
});
}
}
});
}
persisted_workflow(MINIMAL_DOT, &run_dir);
let started = start(
&run_dir,
test_start_options(

15
lib/crates/fabro-workflows/src/pipeline/execute.rs
View file

@ -42,6 +42,9 @@ pub async fn execute(init: Initialized) -> Executed {
hook_runner,
env,
dry_run,
llm_client,
model,
provider,
} = init;
let start = Instant::now();
@ -140,6 +143,9 @@ pub async fn execute(init: Initialized) -> Executed {
sandbox,
duration_ms: crate::millis_u64(start.elapsed()),
final_context: seed_context_from_checkpoint(checkpoint.as_ref()),
llm_client,
model,
provider,
};
}
}
@ -161,6 +167,9 @@ pub async fn execute(init: Initialized) -> Executed {
sandbox,
duration_ms: crate::millis_u64(start.elapsed()),
final_context: seed,
llm_client,
model,
provider,
};
}
}
@ -177,6 +186,9 @@ pub async fn execute(init: Initialized) -> Executed {
sandbox,
duration_ms: crate::millis_u64(start.elapsed()),
final_context: Context::new(),
llm_client,
model,
provider,
};
}
}
@ -296,6 +308,9 @@ pub async fn execute(init: Initialized) -> Executed {
sandbox,
duration_ms,
final_context,
llm_client,
model,
provider,
}
}

71
lib/crates/fabro-workflows/src/pipeline/execute/tests.rs
View file

@ -16,12 +16,13 @@ use super::*;
use crate::context::{self, Context};
use crate::error::FabroError;
use crate::event::{EventEmitter, WorkflowRunEvent};
use crate::handler::default_registry;
use crate::handler::start::StartHandler;
use crate::handler::{Handler as HandlerTrait, HandlerRegistry};
use crate::outcome::{Outcome, OutcomeExt, StageStatus};
use crate::pipeline::initialize;
use crate::pipeline::types::{InitOptions, Persisted};
use crate::pipeline::types::{
InitOptions, Initialized, LlmSpec, Persisted, SandboxEnvSpec, SandboxSpec,
};
use crate::records::{Checkpoint, RunRecord};
use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions};
use crate::test_support::run_graph;
@ -79,6 +80,7 @@ fn test_run_options(run_dir: &Path, run_id: &str) -> RunOptions {
github_app: None,
git_author: crate::git::GitAuthor::default(),
base_branch: None,
display_base_sha: None,
workflow_slug: None,
}
}
@ -151,10 +153,17 @@ async fn execute_runs_start_to_exit_and_returns_final_context() {
run_id: "run-test".to_string(),
dry_run: false,
emitter: Arc::new(crate::event::EventEmitter::new()),
sandbox: Arc::new(fabro_agent::LocalSandbox::new(
std::env::current_dir().unwrap(),
)),
registry: Arc::new(default_registry(Arc::new(AutoApproveInterviewer), || None)),
sandbox: SandboxSpec::Local {
working_directory: std::env::current_dir().unwrap(),
},
llm: LlmSpec {
model: "test-model".to_string(),
provider: fabro_llm::Provider::Anthropic,
fallback_chain: Vec::new(),
mcp_servers: Vec::new(),
dry_run: true,
},
interviewer: Arc::new(AutoApproveInterviewer),
lifecycle: LifecycleOptions {
setup_commands: vec![],
setup_command_timeout_ms: 1_000,
@ -162,7 +171,16 @@ async fn execute_runs_start_to_exit_and_returns_final_context() {
},
run_options: test_run_options(&run_dir, "run-test"),
hooks: HookConfig { hooks: vec![] },
sandbox_env: HashMap::new(),
sandbox_env: SandboxEnvSpec {
devcontainer_env: HashMap::new(),
toml_env: HashMap::new(),
github_permissions: None,
origin_url: None,
},
devcontainer: None,
git: None,
worktree_mode: None,
registry_override: None,
checkpoint: None,
seed_context: None,
},
@ -190,28 +208,25 @@ async fn run_with_lifecycle(
sandbox: Arc<dyn Sandbox>,
graph: &Graph,
run_options: RunOptions,
lifecycle: LifecycleOptions,
_lifecycle: LifecycleOptions,
) -> Result<Outcome, FabroError> {
let run_dir = run_options.run_dir.clone();
let run_id = run_options.run_id.clone();
std::fs::create_dir_all(&run_dir).unwrap();
let initialized = initialize(
persisted_workflow(graph.clone(), String::new(), &run_dir, &run_id),
InitOptions {
run_id,
dry_run: run_options.dry_run,
emitter,
sandbox,
registry: Arc::new(registry),
lifecycle,
run_options,
hooks: HookConfig { hooks: vec![] },
sandbox_env: HashMap::new(),
checkpoint: None,
seed_context: None,
},
)
.await?;
std::fs::create_dir_all(&run_options.run_dir).unwrap();
let initialized = Initialized {
graph: graph.clone(),
source: String::new(),
run_options,
checkpoint: None,
seed_context: None,
emitter,
sandbox,
registry: Arc::new(registry),
hook_runner: None,
env: HashMap::new(),
dry_run: false,
llm_client: None,
model: "test-model".to_string(),
provider: fabro_llm::Provider::Anthropic,
};
super::execute(initialized).await.outcome
}

1
lib/crates/fabro-workflows/src/pipeline/finalize.rs
View file

@ -322,6 +322,7 @@ mod tests {
github_app: None,
host_repo_path: None,
base_branch: None,
display_base_sha: None,
git: None,
}
}

885
lib/crates/fabro-workflows/src/pipeline/initialize.rs
View file

File diff suppressed because it is too large Load diff

5
lib/crates/fabro-workflows/src/pipeline/mod.rs
View file

@ -23,7 +23,8 @@ pub use pull_request::{
pub use retro::{retro, run_retro};
pub use transform::transform;
pub use types::{
Concluded, Executed, FinalizeOptions, Finalized, InitOptions, Initialized, Parsed, Persisted,
PullRequestOptions, RetroOptions, Retroed, TransformOptions, Transformed, Validated,
Concluded, DevcontainerSpec, Executed, FinalizeOptions, Finalized, InitOptions, Initialized,
LlmSpec, Parsed, Persisted, PullRequestOptions, RetroOptions, Retroed, SandboxEnvSpec,
SandboxSpec, TransformOptions, Transformed, Validated,
};
pub use validate::validate;

4
lib/crates/fabro-workflows/src/pipeline/pull_request.rs
View file

@ -488,7 +488,9 @@ pub async fn pull_request(concluded: Concluded, options: &PullRequestOptions) ->
let mut pr_url = None;
if let Some(pr_cfg) = &options.pr_config {
if let Err(ref e) = outcome {
if run_options.dry_run {
tracing::debug!("Skipping PR creation: run is in dry-run mode");
} else if let Err(ref e) = outcome {
tracing::debug!(error = %e, "Skipping PR creation: engine returned an error");
} else if let Ok(ref result) = outcome {
if matches!(

52
lib/crates/fabro-workflows/src/pipeline/retro.rs
View file

@ -8,7 +8,7 @@ use crate::records::Checkpoint;
use super::types::{Executed, RetroOptions, Retroed};
pub async fn run_retro(options: &RetroOptions) -> Option<Retro> {
pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option<Retro> {
let cp = match Checkpoint::load(&options.run_dir.join("checkpoint.json")) {
Ok(cp) => cp,
Err(e) => {
@ -43,7 +43,7 @@ pub async fn run_retro(options: &RetroOptions) -> Option<Retro> {
emitter.emit(&WorkflowRunEvent::RetroStarted);
}
let narrative_result = if options.dry_run {
let narrative_result = if dry_run {
Ok(fabro_retro::retro_agent::dry_run_narrative())
} else if let Some(client) = options.llm_client.as_ref() {
let emitter_clone = options.emitter.clone();
@ -122,10 +122,15 @@ pub async fn retro(executed: Executed, options: &RetroOptions) -> Retroed {
sandbox,
duration_ms,
final_context: _,
llm_client: _,
model: _,
provider: _,
} = executed;
let dry_run = run_options.dry_run;
let retro = if options.enabled {
run_retro(options).await
run_retro(options, dry_run).await
} else {
None
};
@ -189,6 +194,7 @@ mod tests {
github_app: None,
host_repo_path: None,
base_branch: None,
display_base_sha: None,
git: None,
}
}
@ -213,6 +219,9 @@ mod tests {
sandbox: Arc::clone(&sandbox),
duration_ms: 1,
final_context: Context::new(),
llm_client: None,
model: "test-model".to_string(),
provider: fabro_llm::Provider::Anthropic,
};
let retroed = retro(
@ -227,7 +236,6 @@ mod tests {
failed: false,
run_duration_ms: 1,
enabled: true,
dry_run: true,
llm_client: None,
provider: fabro_llm::Provider::Anthropic,
model: "test-model".to_string(),
@ -253,23 +261,25 @@ mod tests {
move |event| seen.lock().unwrap().push(event.clone())
});
let retro = run_retro(&RetroOptions {
run_id: "run-test".to_string(),
workflow_name: "test".to_string(),
goal: "Ship it".to_string(),
run_dir: run_dir.clone(),
sandbox: Arc::new(fabro_agent::LocalSandbox::new(
std::env::current_dir().unwrap(),
)),
emitter: Some(Arc::clone(&emitter)),
failed: false,
run_duration_ms: 1,
enabled: true,
dry_run: true,
llm_client: None,
provider: fabro_llm::Provider::Anthropic,
model: "test-model".to_string(),
})
let retro = run_retro(
&RetroOptions {
run_id: "run-test".to_string(),
workflow_name: "test".to_string(),
goal: "Ship it".to_string(),
run_dir: run_dir.clone(),
sandbox: Arc::new(fabro_agent::LocalSandbox::new(
std::env::current_dir().unwrap(),
)),
emitter: Some(Arc::clone(&emitter)),
failed: false,
run_duration_ms: 1,
enabled: true,
llm_client: None,
provider: fabro_llm::Provider::Anthropic,
model: "test-model".to_string(),
},
true,
)
.await;
assert!(retro.is_some());

86
lib/crates/fabro-workflows/src/pipeline/types.rs
View file

@ -3,14 +3,23 @@ use std::path::{Path, PathBuf};
use std::sync::Arc;
use fabro_agent::Sandbox;
use fabro_config::sandbox::WorktreeMode;
use fabro_graphviz::graph::Graph;
use fabro_hooks::HookRunner;
use fabro_interview::Interviewer;
use fabro_llm::Provider;
use fabro_mcp::config::McpServerConfig;
use fabro_model::FallbackTarget;
use fabro_sandbox::daytona::DaytonaConfig;
use fabro_sandbox::docker::DockerSandboxConfig;
#[cfg(feature = "exedev")]
use fabro_sandbox::exe::{ExeConfig, GitCloneParams as ExeGitCloneParams};
use fabro_sandbox::ssh::{GitCloneParams as SshGitCloneParams, SshConfig};
use fabro_validate::Diagnostic;
use crate::context::Context;
use crate::error::FabroError;
use crate::event::EventEmitter;
use crate::handler::HandlerRegistry;
use crate::outcome::Outcome;
use crate::records::{Checkpoint, Conclusion, RunRecord};
use crate::run_options::{LifecycleOptions, RunOptions};
@ -189,16 +198,74 @@ impl Persisted {
}
/// Options for the INITIALIZE phase.
pub enum SandboxSpec {
Local {
working_directory: PathBuf,
},
Docker {
config: DockerSandboxConfig,
},
Daytona {
config: DaytonaConfig,
github_app: Option<fabro_github::GitHubAppCredentials>,
run_id: Option<String>,
clone_branch: Option<String>,
},
#[cfg(feature = "exedev")]
Exe {
config: ExeConfig,
clone_params: Option<ExeGitCloneParams>,
run_id: Option<String>,
github_app: Option<fabro_github::GitHubAppCredentials>,
mgmt_destination: String,
},
Ssh {
config: SshConfig,
clone_params: Option<SshGitCloneParams>,
run_id: Option<String>,
github_app: Option<fabro_github::GitHubAppCredentials>,
},
}
#[derive(Clone)]
pub struct LlmSpec {
pub model: String,
pub provider: Provider,
pub fallback_chain: Vec<FallbackTarget>,
pub mcp_servers: Vec<McpServerConfig>,
pub dry_run: bool,
}
#[derive(Clone)]
pub struct SandboxEnvSpec {
pub devcontainer_env: HashMap<String, String>,
pub toml_env: HashMap<String, String>,
pub github_permissions: Option<HashMap<String, String>>,
pub origin_url: Option<String>,
}
#[derive(Clone)]
pub struct DevcontainerSpec {
pub enabled: bool,
pub resolve_dir: PathBuf,
}
pub struct InitOptions {
pub run_id: String,
pub dry_run: bool,
pub emitter: Arc<EventEmitter>,
pub sandbox: Arc<dyn Sandbox>,
pub registry: Arc<HandlerRegistry>,
pub sandbox: SandboxSpec,
pub llm: LlmSpec,
pub interviewer: Arc<dyn Interviewer>,
pub lifecycle: LifecycleOptions,
pub run_options: RunOptions,
pub hooks: fabro_hooks::HookConfig,
pub sandbox_env: HashMap<String, String>,
pub sandbox_env: SandboxEnvSpec,
pub devcontainer: Option<DevcontainerSpec>,
pub git: Option<crate::run_options::GitCheckpointOptions>,
pub worktree_mode: Option<WorktreeMode>,
#[cfg(test)]
pub registry_override: Option<Arc<crate::handler::HandlerRegistry>>,
pub checkpoint: Option<Checkpoint>,
pub seed_context: Option<Context>,
}
@ -213,10 +280,13 @@ pub struct Initialized {
pub(crate) seed_context: Option<Context>,
pub emitter: Arc<EventEmitter>,
pub sandbox: Arc<dyn Sandbox>,
pub registry: Arc<HandlerRegistry>,
pub registry: Arc<crate::handler::HandlerRegistry>,
pub hook_runner: Option<Arc<HookRunner>>,
pub env: HashMap<String, String>,
pub dry_run: bool,
pub llm_client: Option<fabro_llm::client::Client>,
pub model: String,
pub provider: Provider,
}
/// Output of the EXECUTE phase.
@ -230,6 +300,9 @@ pub struct Executed {
pub sandbox: Arc<dyn Sandbox>,
pub duration_ms: u64,
pub final_context: Context,
pub llm_client: Option<fabro_llm::client::Client>,
pub model: String,
pub provider: Provider,
}
/// Output of the RETRO phase.
@ -284,9 +357,8 @@ pub struct RetroOptions {
pub failed: bool,
pub run_duration_ms: u64,
pub enabled: bool,
pub dry_run: bool,
pub llm_client: Option<fabro_llm::client::Client>,
pub provider: fabro_llm::Provider,
pub provider: Provider,
pub model: String,
}

2
lib/crates/fabro-workflows/src/run_options.rs
View file

@ -37,6 +37,8 @@ pub struct RunOptions {
pub host_repo_path: Option<PathBuf>,
/// Name of the branch the run was started from (for PR base).
pub base_branch: Option<String>,
/// Base commit SHA to display in lifecycle events/UI even when checkpointing is disabled.
pub display_base_sha: Option<String>,
/// Git checkpoint options; `None` means checkpointing disabled.
pub git: Option<GitCheckpointOptions>,
}

3
lib/crates/fabro-workflows/src/test_support.rs
View file

@ -39,6 +39,9 @@ fn initialized(
hook_runner: options.hook_runner,
env: options.env,
dry_run: run_options.dry_run,
llm_client: None,
model: String::new(),
provider: fabro_llm::Provider::Anthropic,
}
}

6
lib/crates/fabro-workflows/tests/daytona_integration.rs
View file

@ -399,6 +399,7 @@ async fn daytona_pipeline_artifact_offload_and_sync() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -590,6 +591,7 @@ async fn daytona_git_checkpoint_remote_emits_events() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: Some(dir.path().to_path_buf()),
git: Some(GitCheckpointOptions {
base_sha: Some(base_sha),
@ -776,6 +778,7 @@ async fn daytona_parallel_git_branching_e2e() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: Some(run_tmp.path().to_path_buf()),
git: Some(GitCheckpointOptions {
base_sha: Some(base_sha),
@ -1152,6 +1155,7 @@ async fn daytona_git_checkpoint_with_shadow_branch() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: Some(host_repo.path().to_path_buf()),
git: Some(GitCheckpointOptions {
base_sha: Some(base_sha),
@ -1297,6 +1301,7 @@ async fn daytona_asset_collection() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -1548,6 +1553,7 @@ async fn daytona_git_push_run_branch_to_origin() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: Some(dir.path().to_path_buf()),
git: Some(GitCheckpointOptions {
base_sha: Some(base_sha),

114
lib/crates/fabro-workflows/tests/integration.rs
View file

@ -204,6 +204,7 @@ async fn end_to_end_linear_pipeline() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -343,6 +344,7 @@ async fn end_to_end_branching_pipeline() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -463,6 +465,7 @@ async fn end_to_end_human_gate_pipeline() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -559,6 +562,7 @@ async fn human_gate_aborted_input_fails_closed_without_fail_route() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -670,6 +674,7 @@ async fn human_gate_aborted_input_routes_via_outcome_fail_condition() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -783,6 +788,7 @@ async fn goal_gate_routes_to_retry_target_on_failure() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -904,6 +910,7 @@ async fn goal_gate_routes_to_retry_target_when_present() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -1216,6 +1223,7 @@ async fn retry_on_failure_then_succeed() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -1291,6 +1299,7 @@ async fn pipeline_with_many_nodes() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -1613,6 +1622,7 @@ async fn smoke_test_with_mock_codergen_backend() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -1714,6 +1724,7 @@ async fn end_to_end_parallel_fan_out_fan_in() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -1827,6 +1838,7 @@ async fn resume_from_checkpoint_completes_pipeline() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -1926,6 +1938,7 @@ async fn resume_from_checkpoint_preserves_goal_gate_outcomes() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -1969,6 +1982,7 @@ async fn graph_goal_in_context() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -2005,6 +2019,7 @@ async fn event_streaming_lifecycle() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -2085,6 +2100,7 @@ async fn context_flow_between_stages() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -2138,6 +2154,7 @@ async fn tool_handler_e2e() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -2208,6 +2225,7 @@ async fn auto_approve_interviewer_e2e() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -2245,6 +2263,7 @@ async fn codergen_without_backend_simulated() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -2350,6 +2369,7 @@ async fn branching_loop_back_on_failure() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -2433,6 +2453,7 @@ async fn human_gate_loops_back() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -2491,6 +2512,7 @@ async fn scenario_ship_a_feature() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -2577,6 +2599,7 @@ async fn scenario_parallel_expert_review() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -2661,6 +2684,7 @@ async fn scenario_node_retries_on_retry_status() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -2723,6 +2747,7 @@ async fn scenario_loop_restart_resets_context() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -2791,6 +2816,7 @@ async fn scenario_bug_triage_router() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -2850,6 +2876,7 @@ async fn scenario_crash_recovery() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -2959,6 +2986,7 @@ async fn manager_loop_stop_condition_satisfied_e2e() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -3036,6 +3064,7 @@ async fn manager_loop_max_cycles_exceeded_e2e() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -3172,6 +3201,7 @@ async fn conditional_branching_success_fail_paths() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -3225,6 +3255,7 @@ async fn edge_selection_condition_match_wins_over_weight() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -3272,6 +3303,7 @@ async fn edge_selection_weight_breaks_ties() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -3311,6 +3343,7 @@ async fn edge_selection_lexical_tiebreak() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -3369,6 +3402,7 @@ async fn context_updates_visible_across_nodes() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -3413,6 +3447,7 @@ async fn stylesheet_applies_model_override() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -3469,6 +3504,7 @@ async fn custom_handler_registration_and_execution() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -3540,6 +3576,7 @@ async fn integration_smoke_plan_implement_review_done() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -3644,6 +3681,7 @@ async fn manager_loop_runs_child_engine_e2e() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -3778,6 +3816,7 @@ async fn manager_loop_context_flows_e2e() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -3851,6 +3890,7 @@ async fn manager_loop_child_dotfile_e2e() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -3964,6 +4004,7 @@ async fn graph_merge_e2e_through_engine() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -4114,6 +4155,7 @@ async fn fidelity_default_is_compact() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -4171,6 +4213,7 @@ async fn fidelity_graph_default_applied() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -4224,6 +4267,7 @@ async fn fidelity_node_overrides_graph_default() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -4283,6 +4327,7 @@ async fn fidelity_edge_overrides_node_and_graph() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -4332,6 +4377,7 @@ async fn fidelity_full_produces_empty_preamble() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -4391,6 +4437,7 @@ async fn fidelity_truncate_preamble_minimal() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -4463,6 +4510,7 @@ async fn fidelity_summary_low_mode() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -4530,6 +4578,7 @@ async fn fidelity_summary_medium_mode() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -4597,6 +4646,7 @@ async fn fidelity_summary_high_mode() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -4657,6 +4707,7 @@ async fn fidelity_full_sets_thread_id_in_context() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -4728,6 +4779,7 @@ async fn fidelity_full_nodes_share_thread_id() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -4809,6 +4861,7 @@ async fn fidelity_resume_degrades_full_to_summary_high() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -4906,6 +4959,7 @@ async fn fidelity_resume_degrade_only_affects_first_hop() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -4990,6 +5044,7 @@ async fn fidelity_resume_no_degrade_when_not_full() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -5032,6 +5087,7 @@ async fn fidelity_stored_in_checkpoint_context() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -5118,6 +5174,7 @@ async fn fidelity_precedence_multi_node_pipeline() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -5186,6 +5243,7 @@ async fn fidelity_compact_preamble_includes_completed_stages_and_context() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -5261,6 +5319,7 @@ async fn fidelity_summary_low_excludes_context_values_in_pipeline() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -5328,6 +5387,7 @@ async fn fidelity_summary_low_excludes_context_values_in_pipeline() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -5399,6 +5459,7 @@ async fn fidelity_thread_id_fallback_to_previous_node_in_pipeline() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -5453,6 +5514,7 @@ async fn fidelity_thread_id_from_node_class_in_pipeline() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -5510,6 +5572,7 @@ async fn fidelity_edge_thread_id_override_in_pipeline() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -5568,6 +5631,7 @@ async fn fidelity_full_without_explicit_thread_id_uses_previous_node() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -5636,6 +5700,7 @@ async fn fidelity_from_parsed_dot_pipeline() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -5684,6 +5749,7 @@ async fn fidelity_checkpoint_roundtrip_preserves_fidelity() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -5754,6 +5820,7 @@ async fn fidelity_node_thread_id_overrides_edge_thread_id_in_pipeline() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -5841,6 +5908,7 @@ async fn fidelity_resume_preserves_context_values_across_checkpoint() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -6056,6 +6124,7 @@ mod real_llm {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -6170,6 +6239,7 @@ mod real_llm {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -6309,6 +6379,7 @@ mod real_llm {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -6416,6 +6487,7 @@ mod real_llm {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -6512,6 +6584,7 @@ async fn human_gate_freeform_only_routes_text() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -6642,6 +6715,7 @@ async fn human_gate_freeform_with_fixed_choice_match() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -6757,6 +6831,7 @@ async fn human_gate_freeform_fallback_on_unmatched_text() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -6885,6 +6960,7 @@ async fn human_gate_freeform_sets_allow_freeform_on_question() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -6993,6 +7069,7 @@ async fn human_gate_without_freeform_sets_allow_freeform_false() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -7274,6 +7351,7 @@ fn make_run_options(dir: &std::path::Path) -> RunOptions {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
}
@ -8378,6 +8456,7 @@ async fn arc_e2e_with_real_llm() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -8506,6 +8585,7 @@ async fn run_fidelity_prompt_pipeline(fidelity: &str) -> String {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -8705,6 +8785,7 @@ async fn large_context_values_are_offloaded_to_artifact_store() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -8924,6 +9005,7 @@ async fn artifact_pointers_rewritten_for_remote_sandbox() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -9054,6 +9136,7 @@ async fn node_dir_uses_visit_count_on_revisit() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -10026,6 +10109,7 @@ async fn full_pipeline_with_cli_backend_node() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -10157,6 +10241,7 @@ async fn stylesheet_backend_property_routes_to_cli() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -10436,6 +10521,7 @@ async fn git_checkpoint_host_emits_events_and_diff_patch() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: Some(worktree_path.clone()),
git: Some(GitCheckpointOptions {
base_sha: Some(base_sha.clone()),
@ -10639,6 +10725,7 @@ async fn git_checkpoint_host_writes_shadow_branch() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: Some(worktree_path.clone()),
git: Some(GitCheckpointOptions {
base_sha: Some(base_sha),
@ -10837,6 +10924,7 @@ async fn parallel_git_branching_host_e2e() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: Some(worktree_path.clone()),
git: Some(GitCheckpointOptions {
base_sha: Some(base_sha.clone()),
@ -11100,6 +11188,7 @@ async fn git_checkpoint_host_skips_empty_diff_patch() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: Some(worktree_path.clone()),
git: Some(GitCheckpointOptions {
base_sha: Some(base_sha.clone()),
@ -11482,6 +11571,7 @@ async fn e2e_circuit_breaker_deterministic_self_loop() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -11529,6 +11619,7 @@ async fn e2e_circuit_breaker_custom_limit() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -11569,6 +11660,7 @@ async fn e2e_circuit_breaker_ignores_transient_failures() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -11616,6 +11708,7 @@ async fn e2e_circuit_breaker_different_reasons_separate_counters() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -11656,6 +11749,7 @@ async fn e2e_circuit_breaker_loop_restart() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -11718,6 +11812,7 @@ async fn e2e_failure_signature_persisted_in_context() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -11782,6 +11877,7 @@ async fn e2e_failure_signature_hint_overrides_reason_in_context() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -11838,6 +11934,7 @@ async fn e2e_signature_maps_persist_in_checkpoint() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -11965,6 +12062,7 @@ async fn e2e_circuit_breaker_emits_events_before_abort() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -12032,6 +12130,7 @@ async fn e2e_circuit_breaker_does_not_fire_below_limit() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -12128,6 +12227,7 @@ async fn e2e_circuit_breaker_multi_stage_impl_verify_cycle() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -12225,6 +12325,7 @@ async fn e2e_loop_restart_blocked_for_deterministic_failure() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -12265,6 +12366,7 @@ async fn e2e_loop_restart_blocked_for_structural_failure() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -12305,6 +12407,7 @@ async fn e2e_loop_restart_blocked_for_budget_exhausted_failure() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -12345,6 +12448,7 @@ async fn e2e_loop_restart_blocked_for_canceled_failure() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -12382,6 +12486,7 @@ async fn e2e_loop_restart_blocked_for_compilation_loop_failure() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -12423,6 +12528,7 @@ async fn e2e_loop_restart_allowed_for_transient_infra() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -12527,6 +12633,7 @@ async fn e2e_stall_watchdog_triggers_from_dot_parsed_pipeline() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -12583,6 +12690,7 @@ async fn e2e_stall_watchdog_kept_alive_by_handler_events() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -12629,6 +12737,7 @@ async fn e2e_stall_watchdog_disabled_with_zero_timeout() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -12694,6 +12803,7 @@ async fn e2e_stall_watchdog_with_explicit_timeout_override() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -12830,6 +12940,7 @@ async fn asset_collection_local_sandbox_success() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -12944,6 +13055,7 @@ async fn asset_collection_local_sandbox_on_failure() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -13041,6 +13153,7 @@ async fn asset_collection_docker_sandbox() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};
@ -13110,6 +13223,7 @@ async fn wait_timer_e2e() {
workflow_slug: None,
github_app: None,
base_branch: None,
display_base_sha: None,
host_repo_path: None,
git: None,
};