From 9cedd9fc0a423bf18c8ea0e1015bdc423ad1364e Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Thu, 26 Mar 2026 22:48:47 -0400 Subject: [PATCH] Fix verification regressions after runtime init refactor --- lib/crates/fabro-api/src/server.rs | 55 ++++++++++++++++- lib/crates/fabro-api/tests/integration.rs | 23 +++++++- lib/crates/fabro-cli/src/commands/run.rs | 1 + .../fabro-workflows/src/operations/start.rs | 2 - .../src/pipeline/execute/tests.rs | 59 ++++++++++++------- .../src/pipeline/initialize.rs | 12 +--- .../fabro-workflows/src/pipeline/types.rs | 1 - 7 files changed, 115 insertions(+), 38 deletions(-) diff --git a/lib/crates/fabro-api/src/server.rs b/lib/crates/fabro-api/src/server.rs index b4c7a42fd..111f0e546 100644 --- a/lib/crates/fabro-api/src/server.rs +++ b/lib/crates/fabro-api/src/server.rs @@ -96,11 +96,16 @@ struct AggregateUsageTotals { by_model: HashMap, } +type LlmSpecFactory = dyn Fn() -> LlmSpec + Send + Sync; +type RegistryFactoryOverride = + dyn Fn(Arc) -> fabro_workflows::handler::HandlerRegistry + Send + Sync; + /// Shared application state for the server. pub struct AppState { runs: Mutex>, aggregate_usage: Mutex, - llm_spec_factory: Box LlmSpec + Send + Sync>, + llm_spec_factory: Box, + registry_factory_override: Option>, pub dry_run: bool, pub db: sqlx::SqlitePool, max_concurrent_runs: usize, @@ -395,6 +400,26 @@ pub fn create_app_state( ) } +#[doc(hidden)] +pub fn create_app_state_with_registry_factory( + db: sqlx::SqlitePool, + llm_spec_factory: impl Fn() -> LlmSpec + Send + Sync + 'static, + registry_factory_override: impl Fn(Arc) -> fabro_workflows::handler::HandlerRegistry + + Send + + Sync + + 'static, +) -> Arc { + build_app_state( + db, + Box::new(llm_spec_factory), + Some(Box::new(registry_factory_override)), + false, + 5, + fabro_workflows::git::GitAuthor::default(), + Vec::new(), + ) +} + /// 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, @@ -403,11 +428,32 @@ pub fn create_app_state_with_options( max_concurrent_runs: usize, git_author: fabro_workflows::git::GitAuthor, hooks: Vec, +) -> Arc { + build_app_state( + db, + Box::new(llm_spec_factory), + None, + dry_run, + max_concurrent_runs, + git_author, + hooks, + ) +} + +fn build_app_state( + db: sqlx::SqlitePool, + llm_spec_factory: Box, + registry_factory_override: Option>, + dry_run: bool, + max_concurrent_runs: usize, + git_author: fabro_workflows::git::GitAuthor, + hooks: Vec, ) -> Arc { Arc::new(AppState { runs: Mutex::new(HashMap::new()), aggregate_usage: Mutex::new(AggregateUsageTotals::default()), - llm_spec_factory: Box::new(llm_spec_factory), + llm_spec_factory, + registry_factory_override, dry_run, db, max_concurrent_runs, @@ -617,6 +663,10 @@ async fn execute_run(state: Arc, run_id: String) { working_directory: cwd, }; let llm = (state.llm_spec_factory)(); + let registry_override = state + .registry_factory_override + .as_ref() + .map(|factory| Arc::new(factory(Arc::clone(&interviewer) as Arc))); let emitter = Arc::new(emitter); // Transition to Running, populate interviewer + context @@ -698,6 +748,7 @@ async fn execute_run(state: Arc, run_id: String) { devcontainer: None, git: None, worktree_mode: None, + registry_override, checkpoint: None, seed_context: None, }, diff --git a/lib/crates/fabro-api/tests/integration.rs b/lib/crates/fabro-api/tests/integration.rs index 5d7170a0b..8e8e59835 100644 --- a/lib/crates/fabro-api/tests/integration.rs +++ b/lib/crates/fabro-api/tests/integration.rs @@ -427,7 +427,13 @@ mod server_lifecycle { use axum::body::Body; use axum::http::{Request, StatusCode}; - use fabro_api::server::{build_router, create_app_state}; + use fabro_api::server::{build_router, create_app_state_with_registry_factory}; + 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; @@ -441,6 +447,15 @@ mod server_lifecycle { } } + fn gate_registry(interviewer: Arc) -> 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 + } + async fn body_json(body: Body) -> serde_json::Value { let bytes = axum::body::to_bytes(body, usize::MAX).await.unwrap(); serde_json::from_slice(&bytes).unwrap() @@ -470,7 +485,8 @@ 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, test_llm_spec); + let state = + create_app_state_with_registry_factory(test_db().await, test_llm_spec, gate_registry); fabro_api::server::spawn_scheduler(Arc::clone(&state)); let app = build_router(Arc::clone(&state), fabro_api::jwt_auth::AuthMode::Disabled); @@ -566,7 +582,8 @@ 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, test_llm_spec); + let state = + create_app_state_with_registry_factory(test_db().await, test_llm_spec, gate_registry); fabro_api::server::spawn_scheduler(Arc::clone(&state)); let app = build_router(Arc::clone(&state), fabro_api::jwt_auth::AuthMode::Disabled); diff --git a/lib/crates/fabro-cli/src/commands/run.rs b/lib/crates/fabro-cli/src/commands/run.rs index 9d607380f..bb822a2e9 100644 --- a/lib/crates/fabro-cli/src/commands/run.rs +++ b/lib/crates/fabro-cli/src/commands/run.rs @@ -1264,6 +1264,7 @@ async fn run_command_impl( git: None, github_app: github_app.clone(), worktree_mode: Some(worktree_mode), + registry_override: None, dry_run: dry_run_flag, retro: StartRetroOptions { enabled: !no_retro_flag && project_config::is_retro_enabled(), diff --git a/lib/crates/fabro-workflows/src/operations/start.rs b/lib/crates/fabro-workflows/src/operations/start.rs index 90d36fdac..2082c1ec0 100644 --- a/lib/crates/fabro-workflows/src/operations/start.rs +++ b/lib/crates/fabro-workflows/src/operations/start.rs @@ -50,7 +50,6 @@ pub struct StartOptions { pub git: Option, pub github_app: Option, pub worktree_mode: Option, - #[cfg(test)] pub registry_override: Option>, // Still external for now — could be derived from RunRecord.config in follow-up @@ -174,7 +173,6 @@ async fn run_engine( devcontainer: options.devcontainer, git: options.git, worktree_mode: options.worktree_mode, - #[cfg(test)] registry_override: options.registry_override, checkpoint, seed_context: options.seed_context, diff --git a/lib/crates/fabro-workflows/src/pipeline/execute/tests.rs b/lib/crates/fabro-workflows/src/pipeline/execute/tests.rs index e43947ef1..2047aeca5 100644 --- a/lib/crates/fabro-workflows/src/pipeline/execute/tests.rs +++ b/lib/crates/fabro-workflows/src/pipeline/execute/tests.rs @@ -20,9 +20,7 @@ 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, Initialized, LlmSpec, Persisted, SandboxEnvSpec, SandboxSpec, -}; +use crate::pipeline::types::{InitOptions, LlmSpec, Persisted, SandboxEnvSpec, SandboxSpec}; use crate::records::{Checkpoint, RunRecord}; use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions}; use crate::test_support::run_graph; @@ -208,25 +206,46 @@ async fn run_with_lifecycle( sandbox: Arc, graph: &Graph, run_options: RunOptions, - _lifecycle: LifecycleOptions, + lifecycle: LifecycleOptions, ) -> Result { 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, - }; + let run_dir = run_options.run_dir.clone(); + let run_id = run_options.run_id.clone(); + let initialized = initialize( + persisted_workflow(graph.clone(), String::new(), &run_dir, &run_id), + InitOptions { + run_id, + dry_run: false, + emitter, + sandbox: SandboxSpec::Local { + working_directory: PathBuf::from(sandbox.working_directory()), + }, + 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, + run_options, + hooks: HookConfig { hooks: vec![] }, + 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: Some(Arc::new(registry)), + checkpoint: None, + seed_context: None, + }, + ) + .await?; super::execute(initialized).await.outcome } diff --git a/lib/crates/fabro-workflows/src/pipeline/initialize.rs b/lib/crates/fabro-workflows/src/pipeline/initialize.rs index b89a1cf12..e60928af7 100644 --- a/lib/crates/fabro-workflows/src/pipeline/initialize.rs +++ b/lib/crates/fabro-workflows/src/pipeline/initialize.rs @@ -722,10 +722,10 @@ pub async fn initialize( &options.emitter, ) .await?; - #[cfg(test)] let (registry, llm_client, effective_dry_run) = if let Some(registry) = options.registry_override.clone() { - (registry, None, options.dry_run || options.llm.dry_run) + // A caller-supplied registry owns execution behavior for its handlers. + (registry, None, options.dry_run) } else { build_registry( &options.llm, @@ -735,14 +735,6 @@ pub async fn initialize( ) .await? }; - #[cfg(not(test))] - let (registry, llm_client, effective_dry_run) = build_registry( - &options.llm, - Arc::clone(&options.interviewer), - &env, - &options.emitter, - ) - .await?; if effective_dry_run { options.dry_run = true; options.run_options.dry_run = true; diff --git a/lib/crates/fabro-workflows/src/pipeline/types.rs b/lib/crates/fabro-workflows/src/pipeline/types.rs index 87993947c..95ad7ffc7 100644 --- a/lib/crates/fabro-workflows/src/pipeline/types.rs +++ b/lib/crates/fabro-workflows/src/pipeline/types.rs @@ -264,7 +264,6 @@ pub struct InitOptions { pub devcontainer: Option, pub git: Option, pub worktree_mode: Option, - #[cfg(test)] pub registry_override: Option>, pub checkpoint: Option, pub seed_context: Option,