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

refactor(server): refine worker bootstrap config and credential resolution

Browse source 
Iterate on the docker worker runtime bootstrap path and apply a
review-driven cleanup across the server, CLI worker, and workflow crates.

- cache resolved LlmCatalogSettings on AppState and serve worker bootstrap
  config + provider secrets scoped to the vault, via a single
  operations::reachable_provider_ids seam (workflow provider-resolution
  internals revert to pub(crate))
- consolidate PEM decoding into fabro_github::decode_private_key_pem
  (server, diagnostics, CLI) and share GitHubAppCredentials::from_pem_with_slug
  across the env and vault credential paths
- gate GitHub credentials through RunNamespace predicates instead of
  duplicated inline RunMode checks; collapse the duplicated credential call
- simplify Docker container creation (drop the production expect()/Option
  accumulator) and name the Docker Engine status codes

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-05-30 10:46:46 -04:00
parent 9d9c48ba25
commit db453a11b2
No known key found for this signature in database
13 changed files with 409 additions and 441 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -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<PathBuf>,
@ -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::<ServerTarget>()?;
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<Option<Arc<AsyncRwLoc
Ok(Some(Arc::new(AsyncRwLock::new(vault))))
}
async fn fetch_worker_bootstrap_payload(
client: server_client::Client,
run_id: RunId,
bootstrap: RunWorkerBootstrap,
) -> Result<Option<WorkerBootstrapResponse>> {
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<Option<fabro_github::GitHubCredentials>> {
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<String>,
app_slug: Option<String>,
}
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<String>,
app_slug: Option<String>,
}
#[derive(serde::Deserialize)]
struct WorkerBootstrapSettingsFile {
server: Option<WorkerBootstrapServerSettings>,
}
#[derive(serde::Deserialize)]
struct WorkerBootstrapServerSettings {
integrations: Option<WorkerBootstrapIntegrationsSettings>,
}
#[derive(serde::Deserialize)]
struct WorkerBootstrapIntegrationsSettings {
github: Option<WorkerBootstrapGithubSettings>,
}
#[derive(serde::Deserialize)]
struct WorkerBootstrapGithubSettings {
#[serde(default)]
strategy: GithubIntegrationStrategy,
app_id: Option<InterpString>,
slug: Option<InterpString>,
}
#[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<GitHubCredentialSettings> {
let source = fs::read_to_string(path)
.with_context(|| format!("failed to read worker settings from {}", path.display()))?;
let settings = toml::from_str::<WorkerBootstrapSettingsFile>(&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<String> {
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<RunControlState>,
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<String, InterpString>,
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());
}
}
}

27
lib/crates/fabro-cli/src/shared/github.rs
View file

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

33
lib/crates/fabro-github/src/lib.rs
View file

@ -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<Option<Self>, 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<String> {
@ -269,7 +279,7 @@ pub async fn gh_auth_token() -> anyhow::Result<String> {
Ok(token)
}
fn decode_pem_env(name: &str, raw: &str) -> Result<String, String> {
pub fn decode_private_key_pem(name: &str, raw: &str) -> Result<String, String> {
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"));
}

17
lib/crates/fabro-server/src/diagnostics.rs
View file

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

55
lib/crates/fabro-server/src/server.rs
View file

@ -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<String, String> {
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<String> {
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<String> {
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<Option<String>> {
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<AppState>, 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) => {

169
lib/crates/fabro-server/src/server/handler/worker.rs
View file

@ -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::<BTreeSet<_>>();
// 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<String, ser::Error> {
toml::to_string(&WorkerBootstrapConfig {
version: 1,
llm,
server: WorkerBootstrapServerConfig {
integrations: WorkerBootstrapIntegrationsConfig {
github: github.clone(),
},
},
})
fn worker_bootstrap_config_toml(llm: &LlmCatalogSettings) -> Result<String, ser::Error> {
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<WorkerBootstrap
}
struct WorkerBootstrapSecretSelector<'a> {
repo_origin_url: Option<&'a str>,
run_settings: &'a RunNamespace,
accepted_graph: &'a Graph,
catalog: &'a Catalog,
configured_providers: &'a BTreeSet<ProviderId>,
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<ProviderId>,
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<String>) {
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<ProviderId> {
let configured = self
.configured_providers
.iter()
.cloned()
.collect::<Vec<_>>();
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<String>) {
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<String>) {
@ -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

64
lib/crates/fabro-server/src/server/tests.rs
View file

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

146
lib/crates/fabro-server/src/worker_runtime.rs
View file

@ -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<Mutex<HashMap<String, bool>>>,
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::<StartContainerOptions<String>>)
.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<WorkerExit>> = 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<R>(stderr: R) -> BoxStream<'static, Result<WorkerO
where
R: AsyncRead + Send + Unpin + 'static,
{
let (tx, rx) = mpsc::unbounded_channel();
let (tx, rx) = mpsc::channel(WORKER_OUTPUT_BUFFER_LINES);
tokio::spawn(async move {
let mut lines = BufReader::new(stderr).lines();
loop {
@ -378,6 +372,7 @@ where
stream: WorkerOutputStreamKind::Stderr,
line,
}))
.await
.is_err()
{
break;
@ -385,35 +380,41 @@ where
}
Ok(None) => 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<String> {
// 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<String> {
@ -456,9 +454,9 @@ fn docker_worker_command(spec: &DockerWorkerLaunchSpec) -> Vec<String> {
"--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<WorkerOutputLine>> {
let (tx, rx) = mpsc::unbounded_channel();
let (tx, rx) = mpsc::channel(WORKER_OUTPUT_BUFFER_LINES);
tokio::spawn(async move {
let mut logs = docker.logs::<String>(
&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<WorkerOutputLine> {
@ -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(),

21
lib/crates/fabro-types/src/settings/run.rs
View file

@ -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<F>(&mut self, mut lookup: F) -> Result<(), ResolveEnvError>
where
F: FnMut(&str) -> Option<String>,

4
lib/crates/fabro-types/src/worker_bootstrap.rs
View file

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

10
lib/crates/fabro-workflow/src/handler/llm/routing.rs
View file

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

5
lib/crates/fabro-workflow/src/operations/mod.rs
View file

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

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

@ -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<FallbackTarget>,
pub(crate) struct StartLlmResolution {
pub(crate) model: String,
pub(crate) provider_id: ProviderId,
pub(crate) fallback_chain: Vec<FallbackTarget>,
}
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<AsyncRwLock<Vault>>>,
catalog: Arc<Catalog>,
) -> Vec<ProviderId> {
@ -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<ProviderId> {
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,