diff --git a/lib/crates/fabro-cli/src/commands/run/runner.rs b/lib/crates/fabro-cli/src/commands/run/runner.rs index 6fb88adb9..d39ed2652 100644 --- a/lib/crates/fabro-cli/src/commands/run/runner.rs +++ b/lib/crates/fabro-cli/src/commands/run/runner.rs @@ -13,7 +13,7 @@ use std::time::Duration; use anyhow::{Context, Result, anyhow}; use async_trait::async_trait; -use fabro_api::types::{RunManifest, WorkerBootstrapResponse}; +use fabro_api::types::{RunManifest, WorkerBootstrapGithubIntegration, WorkerBootstrapResponse}; use fabro_client::ServerTarget; use fabro_config::user::active_settings_path; use fabro_config::{ServerSettingsBuilder, Storage, load_llm_catalog_settings}; @@ -28,8 +28,10 @@ use fabro_server::run_tool_manifest; use fabro_store::{EventEnvelope, RunProjection, RunProjectionReducer}; use fabro_tool::fabro_client::ClientBackend; use fabro_types::settings::InterpString; -use fabro_types::settings::run::{RunMode, RunNamespace}; use fabro_types::settings::server::GithubIntegrationStrategy; +use fabro_types::worker_bootstrap::{ + WORKER_BOOTSTRAP_CONFIG_PATH, WORKER_BOOTSTRAP_RUN_DIR, WORKER_BOOTSTRAP_STORAGE_DIR, +}; use fabro_types::{ ArtifactUpload, EventBody, FailureReason, Principal, RunBlobId, RunEvent, RunId, WorkflowSettings, @@ -64,10 +66,6 @@ use crate::args::{RunWorkerBootstrap, RunWorkerMode}; use crate::server_client; use crate::shared::github::{GitHubCredentialLookup, build_github_credentials}; -const API_BOOTSTRAP_STORAGE_DIR: &str = "/tmp/fabro-worker/storage"; -const API_BOOTSTRAP_RUN_DIR: &str = "/tmp/fabro-worker/run"; -const API_BOOTSTRAP_CONFIG_PATH: &str = "/tmp/fabro-worker/settings.toml"; - #[derive(Debug, Clone)] struct WorkerBootstrapFiles { storage_dir: Option, @@ -78,9 +76,9 @@ struct WorkerBootstrapFiles { impl WorkerBootstrapFiles { fn api() -> Self { Self { - storage_dir: Some(PathBuf::from(API_BOOTSTRAP_STORAGE_DIR)), - config_path: Some(PathBuf::from(API_BOOTSTRAP_CONFIG_PATH)), - run_dir: PathBuf::from(API_BOOTSTRAP_RUN_DIR), + storage_dir: Some(PathBuf::from(WORKER_BOOTSTRAP_STORAGE_DIR)), + config_path: Some(PathBuf::from(WORKER_BOOTSTRAP_CONFIG_PATH)), + run_dir: PathBuf::from(WORKER_BOOTSTRAP_RUN_DIR), } } } @@ -118,27 +116,33 @@ pub(crate) async fn execute( let target = server.parse::()?; let client = server_client::connect_server_target_with_bearer(&target, worker_token).await?; - let run_store = HttpRunStore::connect(run_id, client.clone_for_reuse()).await?; + let (run_store, bootstrap_payload) = tokio::try_join!( + HttpRunStore::connect(run_id, client.clone_for_reuse()), + fetch_worker_bootstrap_payload(client.clone_for_reuse(), run_id, bootstrap), + )?; let run_state = run_store .state() .await .with_context(|| format!("failed to load run state for {run_id}"))?; let run_spec = &run_state.spec; - let bootstrap_files = match bootstrap { - RunWorkerBootstrap::Local => WorkerBootstrapFiles { - storage_dir, - config_path: None, - run_dir, - }, + let (bootstrap_files, github_settings) = match bootstrap { + RunWorkerBootstrap::Local => ( + WorkerBootstrapFiles { + storage_dir, + config_path: None, + run_dir, + }, + GitHubCredentialSettings::from_default_config(), + ), RunWorkerBootstrap::Api => { - let bootstrap_payload = client - .get_run_worker_bootstrap(&run_id) - .await - .context("failed to retrieve worker bootstrap payload")?; + let bootstrap_payload = + bootstrap_payload.context("API bootstrap payload missing for API bootstrap")?; let files = WorkerBootstrapFiles::api(); + let github_settings = + GitHubCredentialSettings::from_bootstrap(&bootstrap_payload.github); write_api_bootstrap_files(&bootstrap_payload, &files) .context("failed to write worker bootstrap files")?; - files + (files, github_settings) } }; let llm_catalog_settings = load_llm_catalog_settings(bootstrap_files.config_path.as_deref()) @@ -203,7 +207,7 @@ pub(crate) async fn execute( maybe_build_github_credentials( &run_spec.settings, vault_guard.as_deref(), - bootstrap_files.config_path.as_deref(), + &github_settings, lookup, )? }; @@ -350,6 +354,21 @@ fn load_worker_vault(storage_dir: Option<&Path>) -> Result Result> { + match bootstrap { + RunWorkerBootstrap::Local => Ok(None), + RunWorkerBootstrap::Api => client + .get_run_worker_bootstrap(&run_id) + .await + .map(Some) + .context("failed to retrieve worker bootstrap payload"), + } +} + fn write_api_bootstrap_files( payload: &WorkerBootstrapResponse, files: &WorkerBootstrapFiles, @@ -1274,15 +1293,44 @@ fn stamp_system_worker(mut event: RunEvent) -> RunEvent { fn maybe_build_github_credentials( settings: &WorkflowSettings, vault: Option<&fabro_vault::Vault>, - config_path: Option<&Path>, + github_settings: &GitHubCredentialSettings, lookup: GitHubCredentialLookup, ) -> Result> { let resolved_run = &settings.run; - let github_settings = match config_path { - Some(path) => github_credential_settings_from_bootstrap_config(path)?, - None => ServerSettingsBuilder::load_default() + let required = resolved_run.requires_github_credentials(); + if !required && !resolved_run.github_credentials_useful_for_pull_request() { + return Ok(None); + } + + let credentials = build_github_credentials( + github_settings.strategy, + github_settings.app_id.as_deref(), + github_settings.app_slug.as_deref(), + vault, + lookup, + ); + + // A hard requirement propagates errors; the pull-request soft fallback + // tolerates missing credentials. + if required { + credentials + } else { + Ok(credentials.ok().flatten()) + } +} + +#[derive(Default)] +struct GitHubCredentialSettings { + strategy: GithubIntegrationStrategy, + app_id: Option, + app_slug: Option, +} + +impl GitHubCredentialSettings { + fn from_default_config() -> Self { + ServerSettingsBuilder::load_default() .ok() - .map(|settings| GitHubCredentialSettings { + .map(|settings| Self { strategy: settings.server.integrations.github.strategy, app_id: settings .server @@ -1299,90 +1347,16 @@ fn maybe_build_github_credentials( .as_ref() .map(InterpString::as_source), }) - .unwrap_or_default(), - }; - - if requires_github_credentials(resolved_run) { - return build_github_credentials( - github_settings.strategy, - github_settings.app_id.as_deref(), - github_settings.app_slug.as_deref(), - vault, - lookup, - ); + .unwrap_or_default() } - let pull_request_enabled = - resolved_run.execution.mode != RunMode::DryRun && resolved_run.pull_request.is_some(); - if pull_request_enabled { - return Ok(build_github_credentials( - github_settings.strategy, - github_settings.app_id.as_deref(), - github_settings.app_slug.as_deref(), - vault, - lookup, - ) - .ok() - .flatten()); + fn from_bootstrap(github: &WorkerBootstrapGithubIntegration) -> Self { + Self { + strategy: github.strategy, + app_id: github.app_id.clone(), + app_slug: github.slug.clone(), + } } - - Ok(None) -} - -#[derive(Default)] -struct GitHubCredentialSettings { - strategy: GithubIntegrationStrategy, - app_id: Option, - app_slug: Option, -} - -#[derive(serde::Deserialize)] -struct WorkerBootstrapSettingsFile { - server: Option, -} - -#[derive(serde::Deserialize)] -struct WorkerBootstrapServerSettings { - integrations: Option, -} - -#[derive(serde::Deserialize)] -struct WorkerBootstrapIntegrationsSettings { - github: Option, -} - -#[derive(serde::Deserialize)] -struct WorkerBootstrapGithubSettings { - #[serde(default)] - strategy: GithubIntegrationStrategy, - app_id: Option, - slug: Option, -} - -#[expect( - clippy::disallowed_methods, - reason = "Worker bootstrap reads one small generated settings file during startup." -)] -fn github_credential_settings_from_bootstrap_config( - path: &Path, -) -> Result { - let source = fs::read_to_string(path) - .with_context(|| format!("failed to read worker settings from {}", path.display()))?; - let settings = toml::from_str::(&source) - .with_context(|| format!("failed to parse worker settings from {}", path.display()))?; - let Some(github) = settings - .server - .and_then(|server| server.integrations) - .and_then(|integrations| integrations.github) - else { - return Ok(GitHubCredentialSettings::default()); - }; - - Ok(GitHubCredentialSettings { - strategy: github.strategy, - app_id: github.app_id.as_ref().map(InterpString::as_source), - app_slug: github.slug.as_ref().map(InterpString::as_source), - }) } #[expect( @@ -1393,17 +1367,6 @@ fn process_env_var(name: &str) -> Option { std::env::var(name).ok() } -/// Hard-gate for the CLI worker path: a run-level token is requested, or -/// a clone-based sandbox in non-dry-run mode will need credentials to -/// pull the repository. Pull-request-driven credential acquisition is -/// handled separately by the caller as a soft fallback. -fn requires_github_credentials(run: &RunNamespace) -> bool { - if run.integrations.github.is_token_requested() { - return true; - } - run.execution.mode != RunMode::DryRun && run.environment.provider.is_clone_based() -} - fn install_signal_handlers( run_control: Arc, cancel_token: CancellationToken, @@ -2073,31 +2036,17 @@ mod tests { settings } - #[expect( - clippy::disallowed_methods, - reason = "Test helper writes a tiny bootstrap settings fixture." - )] - fn write_worker_github_config(path: &std::path::Path) { - std::fs::write( - path, - r#" -_version = 1 - -[server.integrations.github] -enabled = true -strategy = "app" -app_id = "12345" -slug = "fabro-dev" -"#, - ) - .unwrap(); + fn worker_github_app_settings() -> super::GitHubCredentialSettings { + super::GitHubCredentialSettings { + strategy: GithubIntegrationStrategy::App, + app_id: Some("12345".to_string()), + app_slug: Some("fabro-dev".to_string()), + } } #[test] fn api_bootstrap_github_app_credentials_load_from_worker_vault() { let temp = tempfile::tempdir().unwrap(); - let config_path = temp.path().join("settings.toml"); - write_worker_github_config(&config_path); let vault_path = temp.path().join("secrets.json"); let mut vault = Vault::load(vault_path).unwrap(); vault @@ -2113,7 +2062,7 @@ slug = "fabro-dev" let credentials = maybe_build_github_credentials( &settings, Some(&vault), - Some(&config_path), + &worker_github_app_settings(), GitHubCredentialLookup::ApiBootstrapVault, ) .unwrap() @@ -2135,15 +2084,13 @@ slug = "fabro-dev" #[test] fn api_bootstrap_github_app_credentials_fail_when_vault_secret_is_missing() { let temp = tempfile::tempdir().unwrap(); - let config_path = temp.path().join("settings.toml"); - write_worker_github_config(&config_path); let vault = Vault::load(temp.path().join("secrets.json")).unwrap(); let settings = workflow_settings_requesting_github_credentials(); let err = maybe_build_github_credentials( &settings, Some(&vault), - Some(&config_path), + &worker_github_app_settings(), GitHubCredentialLookup::ApiBootstrapVault, ) .unwrap_err(); @@ -2155,8 +2102,6 @@ slug = "fabro-dev" #[test] fn api_bootstrap_github_app_credentials_fail_for_invalid_pem() { let temp = tempfile::tempdir().unwrap(); - let config_path = temp.path().join("settings.toml"); - write_worker_github_config(&config_path); let vault_path = temp.path().join("secrets.json"); let mut vault = Vault::load(vault_path).unwrap(); vault @@ -2172,7 +2117,7 @@ slug = "fabro-dev" let err = maybe_build_github_credentials( &settings, Some(&vault), - Some(&config_path), + &worker_github_app_settings(), GitHubCredentialLookup::ApiBootstrapVault, ) .unwrap_err(); @@ -2193,8 +2138,6 @@ slug = "fabro-dev" RunNamespace, }; - use super::super::requires_github_credentials; - fn run_with( permissions: HashMap, provider: &str, @@ -2217,28 +2160,28 @@ slug = "fabro-dev" // Even with local sandbox + dry-run, non-empty permissions // force credential acquisition. let run = run_with(permissions, "local", RunMode::DryRun); - assert!(requires_github_credentials(&run)); + assert!(run.requires_github_credentials()); } #[test] fn requires_github_credentials_for_clone_based_provider() { let run = run_with(HashMap::new(), "docker", RunMode::Normal); - assert!(requires_github_credentials(&run)); + assert!(run.requires_github_credentials()); let daytona = run_with(HashMap::new(), "daytona", RunMode::Normal); - assert!(requires_github_credentials(&daytona)); + assert!(daytona.requires_github_credentials()); } #[test] fn does_not_require_github_credentials_for_local_clean_run() { let run = run_with(HashMap::new(), "local", RunMode::Normal); - assert!(!requires_github_credentials(&run)); + assert!(!run.requires_github_credentials()); } #[test] fn does_not_require_github_credentials_for_clone_provider_in_dry_run() { let run = run_with(HashMap::new(), "docker", RunMode::DryRun); - assert!(!requires_github_credentials(&run)); + assert!(!run.requires_github_credentials()); } } } diff --git a/lib/crates/fabro-cli/src/shared/github.rs b/lib/crates/fabro-cli/src/shared/github.rs index ec8058dbf..bf92ab8fc 100644 --- a/lib/crates/fabro-cli/src/shared/github.rs +++ b/lib/crates/fabro-cli/src/shared/github.rs @@ -1,6 +1,4 @@ use anyhow::anyhow; -use base64::Engine as _; -use base64::engine::general_purpose::STANDARD as BASE64_STANDARD; use fabro_github::{GitHubAppCredentials, GitHubCredentials}; use fabro_static::EnvVars; use fabro_types::settings::server::GithubIntegrationStrategy; @@ -60,26 +58,11 @@ fn build_github_app_credentials_from_vault( anyhow!("GITHUB_APP_PRIVATE_KEY is missing from the worker bootstrap vault") })?; let private_key_pem = - decode_pem_value(EnvVars::GITHUB_APP_PRIVATE_KEY, &raw).map_err(anyhow::Error::msg)?; - Ok(Some(GitHubCredentials::App(GitHubAppCredentials { - app_id: app_id.to_string(), - private_key_pem, - slug: app_slug - .map(str::trim) - .filter(|slug| !slug.is_empty()) - .map(str::to_string), - }))) -} - -fn decode_pem_value(name: &str, raw: &str) -> Result { - if raw.starts_with("-----") { - return Ok(raw.to_string()); - } - let pem_bytes = BASE64_STANDARD - .decode(raw) - .map_err(|err| format!("{name} is not valid PEM or base64: {err}"))?; - String::from_utf8(pem_bytes) - .map_err(|err| format!("{name} base64 decoded to invalid UTF-8: {err}")) + fabro_github::decode_private_key_pem(EnvVars::GITHUB_APP_PRIVATE_KEY, &raw) + .map_err(anyhow::Error::msg)?; + Ok(Some(GitHubCredentials::App( + GitHubAppCredentials::from_pem_with_slug(app_id, app_slug, private_key_pem), + ))) } /// Look up GitHub token: GITHUB_TOKEN env -> vault GITHUB_TOKEN -> GH_TOKEN env diff --git a/lib/crates/fabro-github/src/lib.rs b/lib/crates/fabro-github/src/lib.rs index bac1f5fbc..ee40e2cc6 100644 --- a/lib/crates/fabro-github/src/lib.rs +++ b/lib/crates/fabro-github/src/lib.rs @@ -105,7 +105,7 @@ impl GitHubAppCredentials { let Ok(raw) = std::env::var(EnvVars::GITHUB_APP_PRIVATE_KEY) else { return Ok(None); }; - decode_pem_env(EnvVars::GITHUB_APP_PRIVATE_KEY, &raw).map(Some) + decode_private_key_pem(EnvVars::GITHUB_APP_PRIVATE_KEY, &raw).map(Some) } pub fn from_env(app_id: Option<&str>) -> Result, String> { @@ -122,14 +122,24 @@ impl GitHubAppCredentials { let Some(private_key_pem) = Self::private_key_from_env()? else { return Ok(None); }; - Ok(Some(Self { + Ok(Some(Self::from_pem_with_slug( + app_id, + slug, + private_key_pem, + ))) + } + + /// Build credentials from an already-resolved private key PEM, normalizing + /// the optional slug (trim, drop if empty). + pub fn from_pem_with_slug(app_id: &str, slug: Option<&str>, private_key_pem: String) -> Self { + Self { app_id: app_id.to_string(), private_key_pem, slug: slug .map(str::trim) .filter(|slug| !slug.is_empty()) .map(str::to_string), - })) + } } pub fn installation_url(&self, owner: &str) -> Option { @@ -269,7 +279,7 @@ pub async fn gh_auth_token() -> anyhow::Result { Ok(token) } -fn decode_pem_env(name: &str, raw: &str) -> Result { +pub fn decode_private_key_pem(name: &str, raw: &str) -> Result { if raw.starts_with("-----") { return Ok(raw.to_string()); } @@ -1319,9 +1329,12 @@ mod tests { use super::*; #[test] - fn decode_pem_env_accepts_raw_pem() { + fn decode_private_key_pem_accepts_raw_pem() { let pem = "-----BEGIN TEST KEY-----\nabc\n-----END TEST KEY-----"; - assert_eq!(decode_pem_env("GITHUB_APP_PRIVATE_KEY", pem).unwrap(), pem); + assert_eq!( + decode_private_key_pem("GITHUB_APP_PRIVATE_KEY", pem).unwrap(), + pem + ); } #[test] @@ -1348,18 +1361,18 @@ mod tests { } #[test] - fn decode_pem_env_accepts_base64_pem() { + fn decode_private_key_pem_accepts_base64_pem() { let pem = "-----BEGIN TEST KEY-----\nabc\n-----END TEST KEY-----"; let encoded = STANDARD.encode(pem); assert_eq!( - decode_pem_env("GITHUB_APP_PRIVATE_KEY", &encoded).unwrap(), + decode_private_key_pem("GITHUB_APP_PRIVATE_KEY", &encoded).unwrap(), pem ); } #[test] - fn decode_pem_env_rejects_invalid_base64() { - let err = decode_pem_env("GITHUB_APP_PRIVATE_KEY", "%%%not-base64%%%").unwrap_err(); + fn decode_private_key_pem_rejects_invalid_base64() { + let err = decode_private_key_pem("GITHUB_APP_PRIVATE_KEY", "%%%not-base64%%%").unwrap_err(); assert!(err.contains("GITHUB_APP_PRIVATE_KEY is not valid PEM or base64")); } diff --git a/lib/crates/fabro-server/src/diagnostics.rs b/lib/crates/fabro-server/src/diagnostics.rs index 6b8d1f1ab..c15396213 100644 --- a/lib/crates/fabro-server/src/diagnostics.rs +++ b/lib/crates/fabro-server/src/diagnostics.rs @@ -1,8 +1,6 @@ use std::sync::Arc; use std::time::Duration; -use base64::Engine as _; -use base64::engine::general_purpose::STANDARD as BASE64_STANDARD; use fabro_auth::auth_issue_message; use fabro_llm::client::Client as LlmClient; use fabro_llm::model_test::{ModelTestStatus, run_basic_model_probe}; @@ -73,16 +71,6 @@ pub(crate) enum ProviderProbeStatus { Error, } -fn decode_pem_value(name: &str, value: &str) -> Result { - if value.starts_with("-----") { - return Ok(value.to_string()); - } - let bytes = BASE64_STANDARD - .decode(value) - .map_err(|e| format!("{name} is not valid PEM or base64: {e}"))?; - String::from_utf8(bytes).map_err(|e| format!("{name} base64 decoded to invalid UTF-8: {e}")) -} - fn validate_session_secret(value: &str) -> Result<(), String> { session_secret::validate_session_secret(value) } @@ -500,7 +488,10 @@ async fn check_github_app(state: &AppState) -> CheckResult { }; }; - let private_key = match decode_pem_value(EnvVars::GITHUB_APP_PRIVATE_KEY, &private_key_raw) { + let private_key = match fabro_github::decode_private_key_pem( + EnvVars::GITHUB_APP_PRIVATE_KEY, + &private_key_raw, + ) { Ok(value) => value, Err(err) => { return CheckResult { diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index a68351121..0bf936065 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -18,8 +18,6 @@ use axum::response::{IntoResponse, Response}; use axum::routing::{get, post}; use axum::{Json, Router}; use axum_extra::extract::cookie::Key; -use base64::Engine as _; -use base64::engine::general_purpose::STANDARD as BASE64_STANDARD; use bytes::Bytes; use chrono::{DateTime, Utc}; pub use fabro_api::types::{ @@ -89,7 +87,7 @@ use fabro_store::{ }; #[cfg(test)] use fabro_types::BlockedReason; -use fabro_types::settings::run::{NotificationRouteSettings, RunMode}; +use fabro_types::settings::run::NotificationRouteSettings; use fabro_types::settings::server::{ GithubIntegrationSettings, GithubIntegrationStrategy, LogDestination, ServerWorkerRuntime, }; @@ -1497,7 +1495,8 @@ impl AppState { let Some(raw) = raw else { return Ok(None); }; - let private_key_pem = decode_secret_pem(EnvVars::GITHUB_APP_PRIVATE_KEY, &raw)?; + let private_key_pem = + fabro_github::decode_private_key_pem(EnvVars::GITHUB_APP_PRIVATE_KEY, &raw)?; Ok(Some(fabro_github::GitHubCredentials::App( fabro_github::GitHubAppCredentials { app_id, @@ -1552,12 +1551,13 @@ impl AppState { } = resolved_settings; let server_settings = Arc::new(server_settings); let manifest_run_defaults = Arc::new(manifest_run_defaults); + let llm_catalog_settings = Arc::new(llm_catalog_settings); let manifest_run_settings = resolve_manifest_run_settings_with_catalog( manifest_run_defaults.as_ref(), &self.environment_store, ); let catalog = Arc::new( - Catalog::from_builtin_with_overrides(&llm_catalog_settings) + Catalog::from_builtin_with_overrides(llm_catalog_settings.as_ref()) .context("building LLM model catalog")?, ); resolve_canonical_origin(&server_settings.server, &self.env_lookup) @@ -1575,6 +1575,10 @@ impl AppState { .server_settings .write() .expect("server settings lock poisoned") = server_settings; + *self + .llm_catalog_settings + .write() + .expect("LLM catalog settings lock poisoned") = llm_catalog_settings; *self.catalog.write().expect("catalog lock poisoned") = catalog; Ok(()) } @@ -1597,17 +1601,6 @@ async fn resolve_llm_client_from_source( }) } -fn decode_secret_pem(name: &str, raw: &str) -> Result { - if raw.starts_with("-----") { - return Ok(raw.to_string()); - } - let pem_bytes = BASE64_STANDARD - .decode(raw) - .map_err(|err| format!("{name} is not valid PEM or base64: {err}"))?; - String::from_utf8(pem_bytes) - .map_err(|err| format!("{name} base64 decoded to invalid UTF-8: {err}")) -} - fn resolve_interp_string(value: &InterpString) -> anyhow::Result { value .resolve(process_env_var) @@ -3575,15 +3568,18 @@ fn worker_launch_spec( Ok(WorkerLaunchSpec::Docker(DockerWorkerLaunchSpec { common, image: resolve_required_worker_docker_setting( + state, docker.image.as_ref(), "server.worker.docker.image", )?, server_url: resolve_required_worker_docker_setting( + state, docker.server_url.as_ref(), "server.worker.docker.server_url", )?, - network: resolve_optional_worker_docker_setting(docker.network.as_ref())?, + network: resolve_optional_worker_docker_setting(state, docker.network.as_ref())?, docker_socket: resolve_optional_worker_docker_setting( + state, docker.docker_socket.as_ref(), )? .map(PathBuf::from), @@ -3594,13 +3590,16 @@ fn worker_launch_spec( } fn resolve_required_worker_docker_setting( + state: &AppState, value: Option<&InterpString>, path: &str, ) -> anyhow::Result { - let resolved = resolve_interp_string( - value.with_context(|| format!("{path} is required when server.worker.runtime = docker"))?, - ) - .with_context(|| format!("resolve {path}"))?; + let resolved = + state + .resolve_interp(value.with_context(|| { + format!("{path} is required when server.worker.runtime = docker") + })?) + .with_context(|| format!("resolve {path}"))?; if resolved.trim().is_empty() { anyhow::bail!("{path} must not be empty when server.worker.runtime = docker"); } @@ -3608,10 +3607,11 @@ fn resolve_required_worker_docker_setting( } fn resolve_optional_worker_docker_setting( + state: &AppState, value: Option<&InterpString>, ) -> anyhow::Result> { value - .map(resolve_interp_string) + .map(|value| state.resolve_interp(value)) .transpose() .map(|value| value.filter(|resolved| !resolved.trim().is_empty())) } @@ -3972,16 +3972,11 @@ async fn execute_run_in_process(state: Arc, run_id: RunId) { let github_app_result = { let run_spec = persisted.run_spec(); let settings = &run_spec.settings.run; - let clone_can_use_github_credentials = settings.execution.mode != RunMode::DryRun - && settings.environment.provider.is_clone_based() - && run_spec - .repo_origin_url() - .is_some_and(|origin| !origin.trim().is_empty()); - let pull_request_can_use_github_credentials = - settings.execution.mode != RunMode::DryRun && settings.pull_request.is_some(); if settings.integrations.github.is_token_requested() { state.github_credentials(github_settings) - } else if clone_can_use_github_credentials || pull_request_can_use_github_credentials { + } else if settings.github_credentials_useful_for_clone(run_spec.repo_origin_url()) + || settings.github_credentials_useful_for_pull_request() + { match state.github_credentials(github_settings) { Ok(github_app) => Ok(github_app), Err(err) => { diff --git a/lib/crates/fabro-server/src/server/handler/worker.rs b/lib/crates/fabro-server/src/server/handler/worker.rs index 93e96b9a9..b3a147e03 100644 --- a/lib/crates/fabro-server/src/server/handler/worker.rs +++ b/lib/crates/fabro-server/src/server/handler/worker.rs @@ -5,14 +5,13 @@ use fabro_model::catalog::{CredentialRef, HeaderValueRef, LlmCatalogSettings}; use fabro_model::{Catalog, ProviderId}; use fabro_static::EnvVars; use fabro_types::settings::InterpString; -use fabro_types::settings::run::{EnvironmentProvider, McpTransport, RunMode, RunNamespace}; +use fabro_types::settings::run::{EnvironmentProvider, McpTransport, RunNamespace}; use fabro_types::settings::server::{GithubIntegrationSettings, GithubIntegrationStrategy}; use fabro_types::{ - Graph, ServerSettings, WorkerBootstrapGithubIntegration, WorkerBootstrapResponse, - WorkerBootstrapSecret, is_llm_handler_type, + ServerSettings, WorkerBootstrapGithubIntegration, WorkerBootstrapResponse, + WorkerBootstrapSecret, }; use fabro_vault::Vault; -use fabro_workflow::handler::llm::routing; use fabro_workflow::operations; use serde::Serialize; use toml::ser; @@ -44,31 +43,33 @@ async fn retrieve_worker_bootstrap( let server_settings = state.server_settings(); let github = github_bootstrap_metadata(&server_settings.server.integrations.github); - let config_toml = - match worker_bootstrap_config_toml(state.llm_catalog_settings().as_ref(), &github) { - Ok(config_toml) => config_toml, - Err(err) => { - return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) - .into_response(); - } - }; + let config_toml = match worker_bootstrap_config_toml(state.llm_catalog_settings().as_ref()) { + Ok(config_toml) => config_toml, + Err(err) => { + return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) + .into_response(); + } + }; let run_spec = &cached.projection.spec; let catalog = state.catalog(); - let configured_providers = - operations::configured_providers_for_start(Some(&state.vault), Arc::clone(&catalog)) - .await - .into_iter() - .collect::>(); + // Workers only receive vault-delivered secrets, so scope provider discovery + // to the vault rather than the server's process environment. + let configured = state.configured_llm_provider_ids().await; + let reachable_providers = operations::reachable_provider_ids( + catalog.as_ref(), + &configured, + &run_spec.settings.run, + run_spec.graph(), + ); let vault = state.vault.read().await; let selector = WorkerBootstrapSecretSelector { - repo_origin_url: run_spec.repo_origin_url(), - run_settings: &run_spec.settings.run, - accepted_graph: run_spec.graph(), - catalog: catalog.as_ref(), - configured_providers: &configured_providers, - server_settings: server_settings.as_ref(), - server_vault: &vault, + repo_origin_url: run_spec.repo_origin_url(), + run_settings: &run_spec.settings.run, + catalog: catalog.as_ref(), + reachable_providers: &reachable_providers, + server_settings: server_settings.as_ref(), + server_vault: &vault, }; let secrets = selector .required_secret_names() @@ -92,32 +93,10 @@ struct WorkerBootstrapConfig<'a> { #[serde(rename = "_version")] version: u8, llm: &'a LlmCatalogSettings, - server: WorkerBootstrapServerConfig, } -#[derive(Serialize)] -struct WorkerBootstrapServerConfig { - integrations: WorkerBootstrapIntegrationsConfig, -} - -#[derive(Serialize)] -struct WorkerBootstrapIntegrationsConfig { - github: WorkerBootstrapGithubIntegration, -} - -fn worker_bootstrap_config_toml( - llm: &LlmCatalogSettings, - github: &WorkerBootstrapGithubIntegration, -) -> Result { - toml::to_string(&WorkerBootstrapConfig { - version: 1, - llm, - server: WorkerBootstrapServerConfig { - integrations: WorkerBootstrapIntegrationsConfig { - github: github.clone(), - }, - }, - }) +fn worker_bootstrap_config_toml(llm: &LlmCatalogSettings) -> Result { + toml::to_string(&WorkerBootstrapConfig { version: 1, llm }) } fn github_bootstrap_metadata( @@ -142,13 +121,12 @@ fn secret_response_for_name(vault: &Vault, name: &str) -> Option { - repo_origin_url: Option<&'a str>, - run_settings: &'a RunNamespace, - accepted_graph: &'a Graph, - catalog: &'a Catalog, - configured_providers: &'a BTreeSet, - server_settings: &'a ServerSettings, - server_vault: &'a Vault, + repo_origin_url: Option<&'a str>, + run_settings: &'a RunNamespace, + catalog: &'a Catalog, + reachable_providers: &'a BTreeSet, + server_settings: &'a ServerSettings, + server_vault: &'a Vault, } impl WorkerBootstrapSecretSelector<'_> { @@ -163,8 +141,8 @@ impl WorkerBootstrapSecretSelector<'_> { } fn collect_llm_provider_secrets(&self, names: &mut BTreeSet) { - for provider_id in self.reachable_llm_provider_ids() { - let Some(provider) = self.catalog.provider(&provider_id) else { + for provider_id in self.reachable_providers { + let Some(provider) = self.catalog.provider(provider_id) else { continue; }; if let Some(auth) = &provider.auth { @@ -182,38 +160,6 @@ impl WorkerBootstrapSecretSelector<'_> { } } - fn reachable_llm_provider_ids(&self) -> BTreeSet { - let configured = self - .configured_providers - .iter() - .cloned() - .collect::>(); - let Ok(start) = operations::resolve_start_llm(self.catalog, &configured, self.run_settings) - else { - return BTreeSet::new(); - }; - - let mut provider_ids = BTreeSet::new(); - provider_ids.insert(start.provider_id.clone()); - for fallback in &start.fallback_chain { - provider_ids.insert(ProviderId::from(fallback.provider.as_str())); - } - for node in self.accepted_graph.nodes.values() { - if !is_llm_handler_type(node.handler_type()) { - continue; - } - if let Ok(context) = routing::resolve_node_provider_context( - self.catalog, - &start.provider_id, - &start.model, - node, - ) { - provider_ids.insert(context.provider_id); - } - } - provider_ids - } - fn collect_mcp_header_secrets(&self, names: &mut BTreeSet) { for mcp in self.run_settings.agent.mcps.values() { let McpTransport::Http { headers, .. } = &mcp.transport else { @@ -249,17 +195,11 @@ impl WorkerBootstrapSecretSelector<'_> { return true; } - if self.run_settings.execution.mode == RunMode::DryRun { - return false; - } - - let clone_can_use_github_credentials = - self.run_settings.environment.provider.is_clone_based() - && self - .repo_origin_url - .is_some_and(|origin| !origin.trim().is_empty()); - let pull_request_can_use_github_credentials = self.run_settings.pull_request.is_some(); - clone_can_use_github_credentials || pull_request_can_use_github_credentials + self.run_settings + .github_credentials_useful_for_clone(self.repo_origin_url) + || self + .run_settings + .github_credentials_useful_for_pull_request() } fn collect_sandbox_provider_secrets(&self, names: &mut BTreeSet) { @@ -291,8 +231,7 @@ mod tests { ) .expect("test vault entry should persist"); let catalog = Catalog::from_builtin().expect("test catalog should build"); - let configured_providers = BTreeSet::new(); - let graph = Graph::new("test"); + let reachable_providers = BTreeSet::new(); let server_settings = fabro_config::ServerSettingsBuilder::from_toml( r#" _version = 1 @@ -305,13 +244,12 @@ methods = ["dev-token"] let local_settings = RunNamespace::default(); let local_selector = WorkerBootstrapSecretSelector { - repo_origin_url: None, - run_settings: &local_settings, - accepted_graph: &graph, - catalog: &catalog, - configured_providers: &configured_providers, - server_settings: &server_settings, - server_vault: &vault, + repo_origin_url: None, + run_settings: &local_settings, + catalog: &catalog, + reachable_providers: &reachable_providers, + server_settings: &server_settings, + server_vault: &vault, }; assert!( !local_selector @@ -322,13 +260,12 @@ methods = ["dev-token"] let mut daytona_settings = RunNamespace::default(); daytona_settings.environment.provider = EnvironmentProvider::Daytona; let daytona_selector = WorkerBootstrapSecretSelector { - repo_origin_url: None, - run_settings: &daytona_settings, - accepted_graph: &graph, - catalog: &catalog, - configured_providers: &configured_providers, - server_settings: &server_settings, - server_vault: &vault, + repo_origin_url: None, + run_settings: &daytona_settings, + catalog: &catalog, + reachable_providers: &reachable_providers, + server_settings: &server_settings, + server_vault: &vault, }; assert!( daytona_selector diff --git a/lib/crates/fabro-server/src/server/tests.rs b/lib/crates/fabro-server/src/server/tests.rs index 39ee52afd..70ba097cf 100644 --- a/lib/crates/fabro-server/src/server/tests.rs +++ b/lib/crates/fabro-server/src/server/tests.rs @@ -9,6 +9,8 @@ use std::sync::{Arc as StdArc, Mutex as StdMutex}; use axum::body::Body; use axum::http::{Method, Request, header}; +use base64::Engine as _; +use base64::engine::general_purpose::STANDARD as BASE64_STANDARD; use chrono::{Duration as ChronoDuration, Utc}; use fabro_automation::{AutomationId, AutomationTarget}; use fabro_config::bind::Bind; @@ -23,6 +25,7 @@ use fabro_llm::types::{Message as LlmMessage, Request as LlmRequest, TokenCounts use fabro_model::catalog::LlmCatalogSettings; use fabro_model::{Catalog, ModelRef, ProviderId, ReasoningEffort, Speed}; use fabro_types::settings::ServerAuthMethod; +use fabro_types::settings::run::RunMode; use fabro_types::{ AgentBackend, AttrValue, AuthMethod, CommandTermination, FailureCategory, FailureDetail, Graph, InterviewQuestionRecord, Node, Outcome, QuestionType, RunBlobId, RunId, RunSpec, @@ -1113,9 +1116,10 @@ async fn worker_bootstrap_includes_default_provider_secret_and_excludes_unrelate ); } let config_toml = body["config_toml"].as_str().unwrap(); - assert!(config_toml.contains("[server.integrations.github]")); + assert!(!config_toml.contains("[server.integrations.github]")); assert!(!config_toml.contains("[server.storage]")); assert!(!config_toml.contains(EnvVars::SESSION_SECRET)); + assert_eq!(body["github"]["strategy"], "token"); } #[tokio::test] @@ -1424,9 +1428,40 @@ mode = "dry_run" [server.storage] root = "/srv/new" "#; + let updated_llm_catalog_settings: LlmCatalogSettings = toml::from_str( + r#" +[providers.acme] +display_name = "Acme" +adapter = "openai_compatible" +agent_profile = "openai" +base_url = "https://api.acme.test/v1" + +[providers.acme.auth] +credentials = ["env:ACME_API_KEY"] + +[models."acme-large"] +provider = "acme" +display_name = "Acme Large" +family = "acme" +default = true + +[models."acme-large".limits] +context_window = 128000 + +[models."acme-large".features] +tools = true +vision = false +reasoning = false +"#, + ) + .expect("catalog fixture should parse"); state - .replace_runtime_settings(resolved_runtime_settings_from_toml(updated)) + .replace_runtime_settings(resolved_runtime_settings_for_tests( + server_settings_from_toml(updated), + manifest_run_defaults_from_toml(updated), + updated_llm_catalog_settings, + )) .expect("valid settings should replace current state"); assert_eq!(state.canonical_origin().unwrap(), "http://new.example.com"); @@ -1450,6 +1485,13 @@ root = "/srv/new" .and_then(|execution| execution.mode), Some(RunMode::DryRun) ); + assert!( + state + .llm_catalog_settings() + .models + .contains_key("acme-large") + ); + assert!(state.catalog().get("acme-large").is_some()); } #[test] @@ -2675,7 +2717,7 @@ destination = "file" #[test] fn worker_launch_spec_uses_docker_worker_settings() { let storage_dir = tempfile::tempdir().unwrap(); - let state = worker_command_test_state_with_extra_config( + let state = worker_command_test_state_with_extra_config_and_env_lookup( storage_dir.path(), &["dev-token"], Some(TEST_DEV_TOKEN), @@ -2684,12 +2726,20 @@ fn worker_launch_spec_uses_docker_worker_settings() { runtime = "docker" [server.worker.docker] -image = "ghcr.io/fabro-sh/fabro-worker:test" -server_url = "http://fabro-server:3333" -network = "fabro-net" -docker_socket = "/var/run/docker.sock" +image = "{{ env.WORKER_IMAGE }}" +server_url = "{{ env.WORKER_SERVER_URL }}" +network = "{{ env.WORKER_NETWORK }}" +docker_socket = "{{ env.WORKER_SOCKET }}" remove_on_exit = false "#, + &[], + |name| match name { + "WORKER_IMAGE" => Some("ghcr.io/fabro-sh/fabro-worker:test".to_string()), + "WORKER_SERVER_URL" => Some("http://fabro-server:3333".to_string()), + "WORKER_NETWORK" => Some("fabro-net".to_string()), + "WORKER_SOCKET" => Some("/var/run/docker.sock".to_string()), + _ => None, + }, ); state .vault diff --git a/lib/crates/fabro-server/src/worker_runtime.rs b/lib/crates/fabro-server/src/worker_runtime.rs index 0da2b3797..4f07e3ed5 100644 --- a/lib/crates/fabro-server/src/worker_runtime.rs +++ b/lib/crates/fabro-server/src/worker_runtime.rs @@ -1,7 +1,6 @@ use std::collections::HashMap; use std::path::PathBuf; use std::process::Stdio; -use std::sync::Arc; use anyhow::{Context, Result}; use async_trait::async_trait; @@ -15,20 +14,24 @@ use bollard::models::HostConfig; use fabro_static::EnvVars; use fabro_types::RunId; use fabro_types::settings::server::LogDestination; +use fabro_types::worker_bootstrap::{WORKER_BOOTSTRAP_RUN_DIR, WORKER_BOOTSTRAP_STORAGE_DIR}; use futures_util::StreamExt; use futures_util::future::BoxFuture; use futures_util::stream::BoxStream; use tokio::io::{AsyncBufReadExt, AsyncRead, BufReader}; use tokio::process::Command; -use tokio::sync::{Mutex, mpsc}; -use tokio_stream::wrappers::UnboundedReceiverStream; +use tokio::sync::mpsc; +use tokio_stream::wrappers::ReceiverStream; + +const DOCKER_SOCKET_CONTAINER_PATH: &str = "/var/run/docker.sock"; +const WORKER_OUTPUT_BUFFER_LINES: usize = 1024; + +/// Docker Engine API status codes the worker runtime branches on. +const DOCKER_STATUS_NAME_CONFLICT: u16 = 409; +const DOCKER_STATUS_NOT_FOUND: u16 = 404; use crate::spawn_env::apply_worker_env; -const DOCKER_WORKER_STORAGE_DIR: &str = "/tmp/fabro-worker/storage"; -const DOCKER_WORKER_RUN_DIR: &str = "/tmp/fabro-worker/run"; -const DOCKER_SOCKET_CONTAINER_PATH: &str = "/var/run/docker.sock"; - const DOCKER_WORKER_ENV_ALLOWLIST: &[&str] = &[ EnvVars::RUST_LOG, EnvVars::RUST_BACKTRACE, @@ -65,11 +68,10 @@ pub(crate) enum WorkerRef { /// A worker running as a local subprocess. `pre_exec_setpgid` ensures the /// child is the leader of its own process group with `pgid == pid`, so a /// single PID identifies both the process and its group. - Local { - pid: u32, - }, + Local { pid: u32 }, Docker { - container_id: String, + container_id: String, + remove_on_exit: bool, }, } @@ -253,8 +255,7 @@ impl WorkerRuntime for LocalWorkerRuntime { } pub(crate) struct DockerWorkerRuntime { - docker: Docker, - remove_on_exit: Arc>>, + docker: Docker, } impl DockerWorkerRuntime { @@ -265,10 +266,7 @@ impl DockerWorkerRuntime { } fn from_docker(docker: Docker) -> Self { - Self { - docker, - remove_on_exit: Arc::new(Mutex::new(HashMap::new())), - } + Self { docker } } } @@ -284,69 +282,65 @@ impl WorkerRuntime for DockerWorkerRuntime { anyhow::bail!("Docker worker runtime received local launch spec"); }; - let (container_id, _container_name) = - create_docker_worker_container(&self.docker, &spec).await?; - self.remove_on_exit - .lock() - .await - .insert(container_id.clone(), spec.remove_on_exit); - self.docker + let container_id = create_docker_worker_container(&self.docker, &spec).await?; + if let Err(err) = self + .docker .start_container(&container_id, None::>) .await - .with_context(|| format!("failed to start Docker worker container {container_id}"))?; + { + if spec.remove_on_exit { + remove_docker_worker_container(&self.docker, &container_id).await; + } + return Err(anyhow::Error::new(err).context(format!( + "failed to start Docker worker container {container_id}" + ))); + } let output = docker_worker_output_stream(self.docker.clone(), container_id.clone()); let docker = self.docker.clone(); - let remove_on_exit = Arc::clone(&self.remove_on_exit); + let remove_on_exit = spec.remove_on_exit; let wait_container_id = container_id.clone(); let wait: BoxFuture<'static, Result> = Box::pin(async move { - let policy = remove_on_exit - .lock() - .await - .get(&wait_container_id) - .copied() - .unwrap_or(spec.remove_on_exit); let exit = wait_for_docker_worker(&docker, &wait_container_id).await; - remove_on_exit.lock().await.remove(&wait_container_id); - if policy { + if remove_on_exit { remove_docker_worker_container(&docker, &wait_container_id).await; } exit }); Ok(StartedWorker { - worker_ref: WorkerRef::Docker { container_id }, + worker_ref: WorkerRef::Docker { + container_id, + remove_on_exit: spec.remove_on_exit, + }, output, wait, }) } async fn request_stop(&self, worker_ref: &WorkerRef) { - let WorkerRef::Docker { container_id } = worker_ref else { + let WorkerRef::Docker { container_id, .. } = worker_ref else { return; }; kill_docker_worker(&self.docker, container_id, "SIGTERM").await; } async fn force_stop(&self, worker_ref: &WorkerRef) { - let WorkerRef::Docker { container_id } = worker_ref else { + let WorkerRef::Docker { + container_id, + remove_on_exit, + } = worker_ref + else { return; }; kill_docker_worker(&self.docker, container_id, "SIGKILL").await; - let remove = self - .remove_on_exit - .lock() - .await - .get(container_id) - .copied() - .unwrap_or(false); - if remove { + if *remove_on_exit { remove_docker_worker_container(&self.docker, container_id).await; } } async fn is_alive(&self, worker_ref: &WorkerRef) -> bool { - let WorkerRef::Docker { container_id } = worker_ref else { + let WorkerRef::Docker { container_id, .. } = worker_ref else { return false; }; let Ok(details) = self @@ -367,7 +361,7 @@ fn local_worker_output_stream(stderr: R) -> BoxStream<'static, Result break, Err(err) => { - let _ = tx.send(Err( - anyhow::Error::new(err).context("failed to read worker stderr") - )); + let _ = tx + .send(Err( + anyhow::Error::new(err).context("failed to read worker stderr") + )) + .await; break; } } } }); - UnboundedReceiverStream::new(rx).boxed() + ReceiverStream::new(rx).boxed() } async fn create_docker_worker_container( docker: &Docker, spec: &DockerWorkerLaunchSpec, -) -> Result<(String, String)> { - let mut last_error = None; - for attempt in 0..2 { - let name = docker_worker_container_name(&spec.common.run_id); +) -> Result { + // A name collision (the ULID suffix is unlikely but not impossible) is + // retried once with a freshly generated name. + let mut attempt = 0; + loop { let options = Some(CreateContainerOptions { - name: name.clone(), + name: docker_worker_container_name(&spec.common.run_id), platform: None, }); match docker .create_container(options, docker_worker_container_config(spec)) .await { - Ok(container) => return Ok((container.id, name)), - Err(err) if attempt == 0 && docker_status_code(&err) == Some(409) => { - last_error = Some(err); + Ok(container) => return Ok(container.id), + Err(err) + if attempt == 0 + && docker_status_code(&err) == Some(DOCKER_STATUS_NAME_CONFLICT) => + { + attempt += 1; } Err(err) => { return Err( @@ -422,9 +423,6 @@ async fn create_docker_worker_container( } } } - - let err = last_error.expect("name-conflict retry should retain the last Docker error"); - Err(anyhow::Error::new(err).context("failed to create Docker worker container")) } fn docker_worker_container_config(spec: &DockerWorkerLaunchSpec) -> Config { @@ -456,9 +454,9 @@ fn docker_worker_command(spec: &DockerWorkerLaunchSpec) -> Vec { "--server".to_string(), spec.server_url.clone(), "--storage-dir".to_string(), - DOCKER_WORKER_STORAGE_DIR.to_string(), + WORKER_BOOTSTRAP_STORAGE_DIR.to_string(), "--run-dir".to_string(), - DOCKER_WORKER_RUN_DIR.to_string(), + WORKER_BOOTSTRAP_RUN_DIR.to_string(), "--run-id".to_string(), spec.common.run_id.to_string(), "--mode".to_string(), @@ -524,7 +522,7 @@ fn docker_worker_output_stream( docker: Docker, container_id: String, ) -> BoxStream<'static, Result> { - let (tx, rx) = mpsc::unbounded_channel(); + let (tx, rx) = mpsc::channel(WORKER_OUTPUT_BUFFER_LINES); tokio::spawn(async move { let mut logs = docker.logs::( &container_id, @@ -540,21 +538,23 @@ fn docker_worker_output_stream( match item { Ok(output) => { for line in worker_lines_from_log_output(output) { - if tx.send(Ok(line)).is_err() { + if tx.send(Ok(line)).await.is_err() { return; } } } Err(err) => { - let _ = tx.send(Err( - anyhow::Error::new(err).context("failed to read Docker worker logs") - )); + let _ = tx + .send(Err( + anyhow::Error::new(err).context("failed to read Docker worker logs") + )) + .await; return; } } } }); - UnboundedReceiverStream::new(rx).boxed() + ReceiverStream::new(rx).boxed() } fn worker_lines_from_log_output(output: LogOutput) -> Vec { @@ -629,7 +629,7 @@ async fn remove_docker_worker_container(docker: &Docker, container_id: &str) { ) .await { - if docker_status_code(&err) != Some(404) { + if docker_status_code(&err) != Some(DOCKER_STATUS_NOT_FOUND) { tracing::warn!( container_id, error = %err, @@ -759,9 +759,9 @@ mod tests { "--server".to_string(), "http://fabro-server:3333".to_string(), "--storage-dir".to_string(), - DOCKER_WORKER_STORAGE_DIR.to_string(), + WORKER_BOOTSTRAP_STORAGE_DIR.to_string(), "--run-dir".to_string(), - DOCKER_WORKER_RUN_DIR.to_string(), + WORKER_BOOTSTRAP_RUN_DIR.to_string(), "--run-id".to_string(), spec.common.run_id.to_string(), "--mode".to_string(), diff --git a/lib/crates/fabro-types/src/settings/run.rs b/lib/crates/fabro-types/src/settings/run.rs index f138157a0..3caebf5e3 100644 --- a/lib/crates/fabro-types/src/settings/run.rs +++ b/lib/crates/fabro-types/src/settings/run.rs @@ -77,6 +77,27 @@ impl Default for RunNamespace { } impl RunNamespace { + /// True when the run is guaranteed to need GitHub credentials: a run-level + /// token is requested, or a clone-based sandbox in non-dry-run mode must + /// pull the repository. Pull-request-driven acquisition is handled + /// separately by callers as a soft fallback (see + /// [`Self::github_credentials_useful_for_pull_request`]). + pub fn requires_github_credentials(&self) -> bool { + self.integrations.github.is_token_requested() + || (self.execution.mode != RunMode::DryRun + && self.environment.provider.is_clone_based()) + } + + pub fn github_credentials_useful_for_clone(&self, repo_origin_url: Option<&str>) -> bool { + self.execution.mode != RunMode::DryRun + && self.environment.provider.is_clone_based() + && repo_origin_url.is_some_and(|origin| !origin.trim().is_empty()) + } + + pub fn github_credentials_useful_for_pull_request(&self) -> bool { + self.execution.mode != RunMode::DryRun && self.pull_request.is_some() + } + pub fn substitute_variables(&mut self, mut lookup: F) -> Result<(), ResolveEnvError> where F: FnMut(&str) -> Option, diff --git a/lib/crates/fabro-types/src/worker_bootstrap.rs b/lib/crates/fabro-types/src/worker_bootstrap.rs index 1e2e37038..460928ce4 100644 --- a/lib/crates/fabro-types/src/worker_bootstrap.rs +++ b/lib/crates/fabro-types/src/worker_bootstrap.rs @@ -5,6 +5,10 @@ use serde::{Deserialize, Serialize}; use crate::SecretType; use crate::settings::server::GithubIntegrationStrategy; +pub const WORKER_BOOTSTRAP_STORAGE_DIR: &str = "/tmp/fabro-worker/storage"; +pub const WORKER_BOOTSTRAP_RUN_DIR: &str = "/tmp/fabro-worker/run"; +pub const WORKER_BOOTSTRAP_CONFIG_PATH: &str = "/tmp/fabro-worker/settings.toml"; + #[derive(Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct WorkerBootstrapResponse { pub config_toml: String, diff --git a/lib/crates/fabro-workflow/src/handler/llm/routing.rs b/lib/crates/fabro-workflow/src/handler/llm/routing.rs index a3c30fdac..952c31939 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/routing.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/routing.rs @@ -38,12 +38,12 @@ pub(crate) fn node_needs_api_backend(node: &Node) -> bool { } #[derive(Clone)] -pub struct ProviderContext { - pub provider_id: ProviderId, - pub profile_kind: AgentProfileKind, +pub(crate) struct ProviderContext { + pub(crate) provider_id: ProviderId, + pub(crate) profile_kind: AgentProfileKind, } -pub fn resolve_provider_context( +pub(crate) fn resolve_provider_context( catalog: &Catalog, default_provider_id: &ProviderId, model: &str, @@ -76,7 +76,7 @@ pub fn resolve_provider_context( }) } -pub fn resolve_node_provider_context( +pub(crate) fn resolve_node_provider_context( catalog: &Catalog, default_provider_id: &ProviderId, default_model: &str, diff --git a/lib/crates/fabro-workflow/src/operations/mod.rs b/lib/crates/fabro-workflow/src/operations/mod.rs index e36cf99f5..59aa7e461 100644 --- a/lib/crates/fabro-workflow/src/operations/mod.rs +++ b/lib/crates/fabro-workflow/src/operations/mod.rs @@ -20,10 +20,7 @@ pub use resume::resume; pub use retry::{RetryOutcome, RetryRunInput, retry_run}; pub use rewind::{RewindInput, RewindOutcome, rewind}; pub use source::WorkflowInput; -pub use start::{ - StartLlmResolution, StartServices, Started, configured_providers_for_start, resolve_start_llm, - start, -}; +pub use start::{StartServices, Started, reachable_provider_ids, start}; pub use timeline::{ForkTarget, RunTimeline, TimelineEntry, build_timeline, timeline}; pub use validate::{ValidateInput, validate}; diff --git a/lib/crates/fabro-workflow/src/operations/start.rs b/lib/crates/fabro-workflow/src/operations/start.rs index 1141979af..ef1e3bfd8 100644 --- a/lib/crates/fabro-workflow/src/operations/start.rs +++ b/lib/crates/fabro-workflow/src/operations/start.rs @@ -1,4 +1,4 @@ -use std::collections::HashMap; +use std::collections::{BTreeSet, HashMap}; use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; @@ -22,7 +22,9 @@ use fabro_types::settings::run::{ TlsMode as ResolvedTlsMode, }; use fabro_types::settings::{InterpString, ModelRegistry, ResolvedModelRef}; -use fabro_types::{ManifestPath, RunId, RunRunnableSource, SandboxProviderKind}; +use fabro_types::{ + Graph, ManifestPath, RunId, RunRunnableSource, SandboxProviderKind, is_llm_handler_type, +}; use fabro_vault::Vault; use tokio::runtime::Handle; use tokio::sync::RwLock as AsyncRwLock; @@ -85,10 +87,10 @@ struct RunSession { } #[derive(Debug, Clone, PartialEq, Eq)] -pub struct StartLlmResolution { - pub model: String, - pub provider_id: ProviderId, - pub fallback_chain: Vec, +pub(crate) struct StartLlmResolution { + pub(crate) model: String, + pub(crate) provider_id: ProviderId, + pub(crate) fallback_chain: Vec, } pub struct StartServices { @@ -482,7 +484,7 @@ impl RunSession { } } -pub async fn configured_providers_for_start( +pub(crate) async fn configured_providers_for_start( vault: Option<&Arc>>, catalog: Arc, ) -> Vec { @@ -561,7 +563,7 @@ fn resolve_docker_config(settings: &ResolvedRunSettings) -> DockerSandboxOptions docker_config_from_environment(&settings.environment, !settings.clone.enabled) } -pub fn resolve_start_llm( +pub(crate) fn resolve_start_llm( catalog: &Catalog, configured: &[ProviderId], settings: &ResolvedRunSettings, @@ -597,6 +599,38 @@ pub fn resolve_start_llm( }) } +/// Provider ids whose secrets a worker could need for this run: the resolved +/// start provider, its fallback chain, and any per-node provider overrides in +/// the accepted graph. Returns an empty set when start resolution fails (e.g. +/// no configured providers), matching the run's own start behavior. +pub fn reachable_provider_ids( + catalog: &Catalog, + configured: &[ProviderId], + settings: &ResolvedRunSettings, + graph: &Graph, +) -> BTreeSet { + let Ok(start) = resolve_start_llm(catalog, configured, settings) else { + return BTreeSet::new(); + }; + + let mut provider_ids = BTreeSet::new(); + provider_ids.insert(start.provider_id.clone()); + for fallback in &start.fallback_chain { + provider_ids.insert(ProviderId::from(fallback.provider.as_str())); + } + for node in graph.nodes.values() { + if !is_llm_handler_type(node.handler_type()) { + continue; + } + if let Ok(context) = + routing::resolve_node_provider_context(catalog, &start.provider_id, &start.model, node) + { + provider_ids.insert(context.provider_id); + } + } + provider_ids +} + fn resolve_fallback_chain( catalog: &Catalog, _provider: &ProviderId,