scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-10-01 02:04:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Reach run sandboxes from the server through the sandbox driver

Browse source 
`fabro-server/src/sandbox_access.rs` is the server's own path to a run's
sandbox: it connects the record's provider (the driver's Host, Docker and
Daytona providers in process, a plugin executable for any other kind),
keys ownership on the `petri.run` label Petri stamps on every sandbox it
creates, attaches by the recorded id, and for a host record designates
the recorded directory again when the id lives only in the worker's
registry. The Docker client resolves its endpoint from the same variables
Petri forwards to its plugin, so both meet on one daemon.

The doctor's Docker check and the Daytona credential probe move here with
`DaytonaCredentials`, and the `/sandboxes` inventory is rebuilt over the
driver's `list`, narrowed to sandboxes that carry Petri's run label.
Preflight asks the provider for its health instead of creating and
deleting a throwaway sandbox in Fabro's own shape, which no run uses; the
git retry policy behind the repository probe moves into run_manifest.

The callers still on fabro-sandbox's reconnect read their access through
a `legacy_provider_access` shim until they move.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-09-19 13:47:02 -04:00
parent c4ed995b44
commit 0cc645b2d1
No known key found for this signature in database
13 changed files with 1609 additions and 214 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
Cargo.lock generated
View file

@ -2709,6 +2709,11 @@ dependencies = [
"rand 0.9.4",
"reqwest 0.12.28",
"sandbox-driver",
"sandbox-driver-daytona",
"sandbox-driver-docker",
"sandbox-driver-host",
"sandbox-driver-protocol",
"sandbox-driver-testing",
"serde",
"serde_json",
"serde_yaml",

5
lib/apps/fabro-server/Cargo.toml
View file

@ -36,6 +36,10 @@ fabro-workflow-version = { path = "../../components/fabro-workflow-version" }
fabro-sandbox = { path = "../../components/fabro-sandbox" }
fabro-pebble-sandbox = { path = "../../components/fabro-pebble-sandbox" }
sandbox-driver.workspace = true
sandbox-driver-host.workspace = true
sandbox-driver-docker.workspace = true
sandbox-driver-daytona.workspace = true
sandbox-driver-protocol.workspace = true
fabro-github = { path = "../../components/fabro-github" }
pebble-agent.workspace = true
pebble-coding-agent.workspace = true
@ -124,6 +128,7 @@ tokio-util.workspace = true
tokio-tungstenite.workspace = true
fabro-macros = { path = "../../foundation/fabro-macros" }
fabro-sandbox = { path = "../../components/fabro-sandbox", features = ["test-support"] }
sandbox-driver-testing.workspace = true
fabro-store = { path = "../../components/fabro-store", features = ["test-support"] }
fabro-test = { workspace = true }
fabro-types = { path = "../../foundation/fabro-types", features = ["test-support"] }

12
lib/apps/fabro-server/src/diagnostics.rs
View file

@ -9,7 +9,6 @@ use fabro_llm::Client;
use fabro_llm::lithos_catalog::{Catalog, CatalogProvider};
use fabro_llm::probe::{self, ModelTestStatus};
use fabro_redact::redact_string;
use fabro_sandbox::daytona;
use fabro_static::EnvVars;
use fabro_types::SandboxProviderKind;
use fabro_types::settings::ServerAuthMethod;
@ -24,6 +23,7 @@ use serde::Serialize;
use tokio::time::error::Elapsed;
use tokio::time::timeout;
use crate::sandbox_access::{self, DaytonaCredentialProbeTimeout, DaytonaKeyCheck};
use crate::server::AppState;
const EXTERNAL_SERVICE_PROBE_TIMEOUT: Duration = Duration::from_secs(15);
@ -586,9 +586,9 @@ async fn check_docker_sandbox(state: &AppState) -> CheckResult {
.providers
.is_enabled(&SandboxProviderKind::DOCKER),
|| async {
fabro_sandbox::check_docker_daemon()
sandbox_access::check_docker_daemon()
.await
.map_err(|err| err.display_with_causes())
.map_err(|err| format!("{err:#}"))
},
DOCKER_PROBE_TIMEOUT,
)
@ -671,7 +671,7 @@ async fn check_cloud_sandbox(state: &AppState) -> CheckResult {
cloud_sandbox_probe_check(probe)
}
fn cloud_sandbox_probe_check(probe: anyhow::Result<daytona::DaytonaKeyCheck>) -> CheckResult {
fn cloud_sandbox_probe_check(probe: anyhow::Result<DaytonaKeyCheck>) -> CheckResult {
match probe {
Ok(check) if check.ok() => CheckResult {
name: "Cloud Sandbox".to_string(),
@ -695,7 +695,7 @@ fn cloud_sandbox_probe_check(probe: anyhow::Result<daytona::DaytonaKeyCheck>) ->
)),
},
Err(err) => {
if let Some(timeout) = err.downcast_ref::<daytona::DaytonaCredentialProbeTimeout>() {
if let Some(timeout) = err.downcast_ref::<DaytonaCredentialProbeTimeout>() {
return CheckResult {
name: "Cloud Sandbox".to_string(),
status: CheckStatus::Error,
@ -1240,7 +1240,7 @@ enabled = false
#[test]
fn check_cloud_sandbox_reports_timeout() {
let result = cloud_sandbox_probe_check(Err(anyhow::Error::new(
daytona::DaytonaCredentialProbeTimeout::new(Duration::from_millis(1)),
DaytonaCredentialProbeTimeout::new(Duration::from_millis(1)),
)));
assert_eq!(result.name, "Cloud Sandbox");

9
lib/apps/fabro-server/src/install.rs
View file

@ -26,8 +26,6 @@ use fabro_install::{
};
use fabro_llm::lithos_catalog::{Catalog, CatalogProvider};
use fabro_llm::probe::{self, ApiKeyProbeError, ModelTestStatus};
use fabro_sandbox::daytona;
use fabro_sandbox::driver::DaytonaCredentials;
use fabro_static::EnvVars;
use fabro_store::ArtifactStore;
use fabro_types::settings::server::ObjectStoreSettings;
@ -49,6 +47,9 @@ use tracing::{error, info, warn};
use zeroize::Zeroizing;
use crate::error::ApiError;
use crate::sandbox_access::{
DAYTONA_CREDENTIAL_PROBE_TIMEOUT, DaytonaCredentials, DaytonaKeyCheck, check_daytona_api_key,
};
use crate::serve::{self, DEFAULT_TCP_PORT};
use crate::server_secrets::{ServerSecrets, process_env_snapshot};
use crate::{security_headers, server, static_files};
@ -1004,14 +1005,14 @@ async fn post_install_sandbox_test(
async fn check_install_daytona_api_key(
state: &InstallAppState,
api_key: String,
) -> anyhow::Result<daytona::DaytonaKeyCheck> {
) -> anyhow::Result<DaytonaKeyCheck> {
let credentials = DaytonaCredentials::new(api_key)
.with_api_url(state.upstreams.daytona_api_base_url.clone())
.with_organization_id(state.upstreams.daytona_organization_id.clone())
.with_http_client(Some(
fabro_http::http_client().context("failed to build HTTP client")?,
));
daytona::check_daytona_api_key(&credentials, daytona::DAYTONA_CREDENTIAL_PROBE_TIMEOUT).await
check_daytona_api_key(&credentials, DAYTONA_CREDENTIAL_PROBE_TIMEOUT).await
}
async fn put_install_sandbox(

1
lib/apps/fabro-server/src/lib.rs
View file

@ -44,6 +44,7 @@ mod run_selector;
mod run_title_generation;
#[cfg(test)]
mod run_tool_create;
mod sandbox_access;
pub mod security_headers;
pub mod serve;
pub mod server;

2
lib/apps/fabro-server/src/run_files.rs
View file

@ -1177,7 +1177,7 @@ async fn reconnect_run_sandbox(
.cloned()
.ok_or_else(|| ApiError::new(StatusCode::NOT_FOUND, "Run sandbox was not created."))?;
let access = state
.provider_access()
.legacy_provider_access()
.await
.map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?;
let sandbox = reconnect_for_run(&record, &access, Some(*run_id), None)

472
lib/apps/fabro-server/src/run_manifest.rs
View file

@ -1,8 +1,8 @@
use std::collections::HashMap;
use std::future::Future;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;
use std::sync::{Arc, Mutex, PoisonError};
use std::time::{Duration, SystemTime};
use anyhow::{Context as _, Result, anyhow, bail};
use fabro_api::types;
@ -19,9 +19,6 @@ use fabro_petri::check::Launch;
use fabro_petri::run_graph;
use fabro_petri::runtime::RuntimeSpec;
use fabro_proc::ProcessError;
use fabro_sandbox::{
CloneRequest, ProviderAccess, RunSandbox, SandboxSpec, sandbox_spec_for_environment,
};
use fabro_static::EnvVars;
use fabro_types::diagnostic::{Diagnostic, Severity};
use fabro_types::settings::cli::OutputVerbosity;
@ -36,12 +33,17 @@ use fabro_workflow::Error as WorkflowError;
use fabro_workflow::workflow_bundle::{BundledWorkflow, ParsedWorkflowConfig, WorkflowBundle};
use futures_util::stream::{self, StreamExt};
use lithos_llm::catalog::ProviderId;
use sandbox_driver::{
GitBackoff, GitCredentials, GitFailure, GitFailureKind, GitRetryPolicy, HealthStatus,
ProviderHealth,
};
use tokio::process::Command;
use tokio::task;
#[cfg(test)]
use tokio::time;
use tokio_util::sync::CancellationToken;
use crate::sandbox_access::{self, ProviderAccess};
use crate::server::{AppState, petri_runs};
use crate::{petri_check, run_compiler};
@ -830,50 +832,11 @@ async fn run_ls_remote(mut command: Command) -> std::result::Result<(), String>
})
}
fn preflight_sandbox_spec(
sandbox_provider: &SandboxProviderKind,
prepared: &PreparedManifest,
resolved_run: &RunNamespace,
access: &ProviderAccess,
) -> std::result::Result<SandboxSpec, fabro_sandbox::Error> {
let clone_origin_url = prepared
.git
.as_ref()
.map(|git| fabro_github::normalize_repo_origin_url(&git.origin_url));
let clone_branch = prepared.git.as_ref().map(|git| git.branch.clone());
if sandbox_provider.bundled() == Some(BundledProvider::Local) {
let working_directory = resolved_run
.environment
.local_working_directory(Some(&prepared.source_directory))
.map_err(|err| {
fabro_sandbox::Error::context(
"Failed to resolve local environment working directory",
err,
)
})?;
return Ok(SandboxSpec::local(working_directory, access.clone()));
}
// No vault is available on this path, so a `{{ secrets.* }}` value keeps
// its source form. Preflight never clones.
let spec = sandbox_spec_for_environment(
&resolved_run.environment,
resolved_run.environment.unresolved_env(),
)?;
let clone = CloneRequest {
origin_url: clone_origin_url,
branch: clone_branch,
..CloneRequest::none()
};
Ok(SandboxSpec {
kind: sandbox_provider.clone(),
access: access.clone(),
spec,
clone,
run_id: None,
})
}
/// The sandbox check of preflight: the run's provider is reachable and its
/// credential accepted, as the provider's own health check reports. No
/// sandbox is created: Petri creates the run's, in the run's own shape,
/// when the run starts. A `local` environment must also resolve the
/// directory the run would work in.
async fn run_sandbox_check(
checks: &mut Vec<CheckResult>,
sandbox_provider: &SandboxProviderKind,
@ -881,86 +844,78 @@ async fn run_sandbox_check(
resolved_run: &RunNamespace,
access: &ProviderAccess,
) -> bool {
let spec = match preflight_sandbox_spec(sandbox_provider, prepared, resolved_run, access) {
Ok(spec) => spec,
Err(err) => {
if sandbox_provider.bundled() == Some(BundledProvider::Local) {
if let Err(err) = resolved_run
.environment
.local_working_directory(Some(&prepared.source_directory))
{
checks.push(CheckResult {
name: "Sandbox".into(),
status: CheckStatus::Error,
summary: "failed".into(),
details: vec![CheckDetail::new(format!("Provider: {sandbox_provider}"))],
remediation: Some(err.to_string()),
remediation: Some(format!(
"Failed to resolve local environment working directory: {err}"
)),
});
return false;
}
};
let sandbox_result: Result<Arc<RunSandbox>, String> = spec.build(None).await.map_err(|err| {
if *sandbox_provider == SandboxProviderKind::DAYTONA {
format!("Daytona sandbox creation failed: {err}")
} else {
err.to_string()
}
});
}
let health = sandbox_access::provider_health(sandbox_provider, access)
.await
.map_err(|err| format!("{err:#}"));
let check = sandbox_health_check(sandbox_provider, health);
let passed = check.status == CheckStatus::Pass;
checks.push(check);
passed
}
match sandbox_result {
Ok(sandbox) => match sandbox.initialize().await {
Ok(()) => {
let mut details = vec![CheckDetail::new(format!("Provider: {sandbox_provider}"))];
if sandbox_provider.clones_workspace()
&& prepared.git.is_none()
&& !clone_disabled_for_provider(sandbox_provider, resolved_run)
{
details.push(CheckDetail {
text: "No clone source present; sandbox workspace will be empty".into(),
warn: true,
});
}
if let Err(err) = sandbox.delete().await {
checks.push(CheckResult {
name: "Sandbox".into(),
status: CheckStatus::Error,
summary: "cleanup failed".into(),
details,
remediation: Some(format!("Sandbox cleanup failed: {err}")),
});
return false;
}
checks.push(CheckResult {
name: "Sandbox".into(),
status: CheckStatus::Pass,
summary: sandbox_provider.to_string(),
details,
remediation: None,
});
true
}
Err(err) => {
let cleanup_error = sandbox.delete().await.err();
checks.push(CheckResult {
name: "Sandbox".into(),
status: CheckStatus::Error,
summary: "failed".into(),
details: vec![CheckDetail::new(format!("Provider: {sandbox_provider}"))],
remediation: Some(cleanup_error.map_or_else(
|| format!("Sandbox init failed: {err}"),
|cleanup| {
format!("Sandbox init failed: {err}; cleanup also failed: {cleanup}")
},
)),
});
false
}
/// The preflight check for a provider's health report: a connection
/// failure and an unreachable or rejected backend fail with the reason; a
/// healthy backend, or one whose provider has no health check, passes.
fn sandbox_health_check(
sandbox_provider: &SandboxProviderKind,
health: std::result::Result<ProviderHealth, String>,
) -> CheckResult {
let details = vec![CheckDetail::new(format!("Provider: {sandbox_provider}"))];
let failure = |remediation: String| CheckResult {
name: "Sandbox".into(),
status: CheckStatus::Error,
summary: "failed".into(),
details: details.clone(),
remediation: Some(remediation),
};
let health = match health {
Ok(health) => health,
Err(err) => return failure(err),
};
match health.status {
HealthStatus::Ok | HealthStatus::Unknown => CheckResult {
name: "Sandbox".into(),
status: CheckStatus::Pass,
summary: sandbox_provider.to_string(),
details,
remediation: None,
},
Err(err) => {
checks.push(CheckResult {
name: "Sandbox".into(),
status: CheckStatus::Error,
summary: "failed".into(),
details: vec![CheckDetail::new(format!("Provider: {sandbox_provider}"))],
remediation: Some(err),
});
false
}
HealthStatus::Unreachable => failure(format!(
"{sandbox_provider} backend is unreachable: {}",
health
.message
.unwrap_or_else(|| "the backend did not answer".to_string())
)),
HealthStatus::Unauthorized if !health.missing_permissions.is_empty() => failure(format!(
"{sandbox_provider} credential is missing required permissions: {}",
health.missing_permissions.join(", ")
)),
HealthStatus::Unauthorized => failure(format!(
"{sandbox_provider} rejected the credential: {}",
health
.message
.unwrap_or_else(|| "the credential was rejected".to_string())
)),
_ => failure(format!(
"{sandbox_provider} reported an unknown health state"
)),
}
}
@ -1268,8 +1223,8 @@ where
F: FnMut() -> Fut,
Fut: Future<Output = std::result::Result<(), String>>,
{
fabro_sandbox::retry_git_messages(
&fabro_sandbox::repository_probe_policy(),
retry_git_messages(
&repository_probe_policy(),
Some(&snapshot),
"repository probe",
run,
@ -1277,6 +1232,105 @@ where
.await
}
// ── Fabro's retry budget for git operations against GitHub ─────────────────
//
// The driver owns the retry loop and the decision
// (`sandbox_driver::retry_git`): a remote that cannot be reached is retried,
// a rejected credential is retried only while the token is fresh enough to
// still be replicating to GitHub's git endpoints, a static credential fails
// fast, and a command whose outcome is unknown is never replayed. Fabro
// keeps what is policy: how many attempts the host-side repository probe
// gets, how it paces them, and when the credential it runs with was minted.
//
// Retries reuse the same token on purpose. Replication of a given token
// only makes progress, so each attempt strictly improves the odds, while
// re-minting would restart the replication clock.
/// The username GitHub expects with an installation token or PAT.
const GITHUB_TOKEN_USERNAME: &str = "x-access-token";
/// Backoff between attempts: 3s, then 9s.
///
/// GitHub's guidance for token replication is to wait a few seconds and
/// retry with the same token. Sub-second delays land inside the same
/// replication window and spend an attempt for nothing.
fn replication_backoff() -> GitBackoff {
GitBackoff::new(Duration::from_secs(3), 3.0, Duration::from_secs(10))
}
/// Host-side repository probes get 3 attempts at replication pacing, with
/// no deadline of their own.
fn repository_probe_policy() -> GitRetryPolicy {
GitRetryPolicy::new(3, replication_backoff())
}
/// Credentials carrying only the token's mint time, which is all the
/// driver's decision reads for git that ran outside a sandbox. The token
/// itself never leaves its snapshot.
fn credential_age(snapshot: Option<&TokenSnapshot>) -> Option<GitCredentials> {
let snapshot = snapshot?;
let credentials = GitCredentials::new(GITHUB_TOKEN_USERNAME, "");
Some(match snapshot.minted_at() {
Some(minted_at) => credentials.minted_at(SystemTime::from(minted_at)),
None => credentials,
})
}
/// The driver's failure for a rendered git message, so git that ran
/// outside a sandbox (the host-side repository probe) is classified the
/// same way as git the driver ran.
fn classified_git_failure(operation: &str, message: &str) -> sandbox_driver::Error {
sandbox_driver::Error::Git(GitFailure::classified(
operation,
GitFailureKind::from_message(message),
None,
))
}
/// Runs a host-side git operation that reports failures as rendered
/// messages under `policy`, retrying while the driver's decision says the
/// message is transient for the token behind `snapshot`. The final failure
/// comes back as the operation's own message.
async fn retry_git_messages<F, Fut>(
policy: &GitRetryPolicy,
snapshot: Option<&TokenSnapshot>,
operation: &str,
mut run: F,
) -> std::result::Result<(), String>
where
F: FnMut() -> Fut,
Fut: Future<Output = std::result::Result<(), String>>,
{
let credentials = credential_age(snapshot);
// The operation's own message is kept beside the classified failure the
// driver decides on, so the caller reads the message it knows.
let last_message = Mutex::new(None);
let result = sandbox_driver::retry_git(
policy,
credentials.as_ref(),
operation,
|_attempt, _timeout| {
let attempt = run();
let last_message = &last_message;
async move {
attempt.await.map_err(|message| {
let error = classified_git_failure(operation, &message);
*last_message.lock().unwrap_or_else(PoisonError::into_inner) = Some(message);
error
})
}
},
)
.await;
match result {
Ok(_) => Ok(()),
Err(failure) => Err(last_message
.into_inner()
.unwrap_or_else(PoisonError::into_inner)
.unwrap_or_else(|| failure.error.to_string())),
}
}
async fn run_probe_ls_remote(url: &str, token: &ResolvedToken) -> std::result::Result<(), String> {
let mut command = Command::new("git");
fabro_github::apply_probe_git_env(&mut command, token.token.expose());
@ -1933,29 +1987,177 @@ provider = "local"
);
}
fn health(status: HealthStatus, message: Option<&str>) -> ProviderHealth {
let mut health = ProviderHealth::new(status);
health.message = message.map(str::to_string);
health
}
#[test]
fn preflight_sandbox_spec_disables_docker_clone_but_preserves_clone_metadata() {
let (prepared, resolved) = prepared_and_resolved_for_sandbox(
fn a_healthy_or_uncheckable_provider_passes_the_sandbox_check() {
for status in [HealthStatus::Ok, HealthStatus::Unknown] {
let check =
sandbox_health_check(&SandboxProviderKind::DOCKER, Ok(health(status, None)));
assert_eq!(check.status, CheckStatus::Pass, "{status:?}");
assert_eq!(check.summary, "docker");
assert_eq!(check.details[0].text, "Provider: docker");
assert!(check.remediation.is_none());
}
}
#[test]
fn an_unreachable_backend_fails_the_sandbox_check_with_its_reason() {
let check = sandbox_health_check(
&SandboxProviderKind::DOCKER,
true,
Some(git_context("https://github.com/acme/widgets", "main")),
Ok(health(
HealthStatus::Unreachable,
Some("connection refused on /var/run/docker.sock"),
)),
);
assert_eq!(check.status, CheckStatus::Error);
assert_eq!(check.summary, "failed");
assert_eq!(
check.remediation.as_deref(),
Some("docker backend is unreachable: connection refused on /var/run/docker.sock")
);
}
#[test]
fn a_rejected_credential_fails_the_sandbox_check_naming_the_missing_scopes() {
let mut unauthorized = health(HealthStatus::Unauthorized, Some("forbidden"));
unauthorized.missing_permissions = vec!["write:sandboxes".to_string()];
let check = sandbox_health_check(&SandboxProviderKind::DAYTONA, Ok(unauthorized));
assert_eq!(check.status, CheckStatus::Error);
assert_eq!(
check.remediation.as_deref(),
Some("daytona credential is missing required permissions: write:sandboxes")
);
let spec = preflight_sandbox_spec(
&SandboxProviderKind::DOCKER,
let check = sandbox_health_check(
&SandboxProviderKind::DAYTONA,
Ok(health(HealthStatus::Unauthorized, Some("bad key"))),
);
assert_eq!(
check.remediation.as_deref(),
Some("daytona rejected the credential: bad key")
);
}
#[test]
fn a_provider_that_cannot_connect_fails_the_sandbox_check_with_the_connect_error() {
let check = sandbox_health_check(
&SandboxProviderKind::DAYTONA,
Err("Daytona sandboxes require DAYTONA_API_KEY in the vault".to_string()),
);
assert_eq!(check.status, CheckStatus::Error);
assert_eq!(
check.remediation.as_deref(),
Some("Daytona sandboxes require DAYTONA_API_KEY in the vault")
);
}
#[tokio::test]
async fn the_local_sandbox_check_passes_through_the_host_providers_health() {
let (prepared, resolved) =
prepared_and_resolved_for_sandbox(&SandboxProviderKind::LOCAL, false, None);
let mut checks = Vec::new();
let passed = run_sandbox_check(
&mut checks,
&SandboxProviderKind::LOCAL,
&prepared,
&resolved,
&ProviderAccess::default(),
);
)
.await;
assert!(passed, "{checks:?}");
assert_eq!(checks[0].name, "Sandbox");
assert_eq!(checks[0].status, CheckStatus::Pass);
}
let spec = spec.expect("Docker preflight sandbox spec");
assert_eq!(spec.kind, SandboxProviderKind::DOCKER);
assert!(spec.clone.skip);
assert_eq!(
spec.clone.origin_url.as_deref(),
Some("https://github.com/acme/widgets")
#[tokio::test]
async fn the_daytona_sandbox_check_fails_without_a_vault_key() {
let (prepared, resolved) =
prepared_and_resolved_for_sandbox(&SandboxProviderKind::DAYTONA, false, None);
let mut checks = Vec::new();
let passed = run_sandbox_check(
&mut checks,
&SandboxProviderKind::DAYTONA,
&prepared,
&resolved,
&ProviderAccess::default(),
)
.await;
assert!(!passed);
assert_eq!(checks[0].status, CheckStatus::Error);
assert!(
checks[0]
.remediation
.as_deref()
.unwrap_or_default()
.contains("DAYTONA_API_KEY"),
"{checks:?}"
);
assert_eq!(spec.clone.branch.as_deref(), Some("main"));
}
fn snapshot(age: Duration) -> TokenSnapshot {
let now = chrono::Utc::now();
TokenSnapshot {
generation: 1,
provenance: fabro_github::token_source::TokenProvenance::Minted {
minted_at: now - chrono::Duration::from_std(age).unwrap(),
expires_at: now + chrono::Duration::hours(1),
},
}
}
fn static_snapshot() -> TokenSnapshot {
TokenSnapshot {
generation: 0,
provenance: fabro_github::token_source::TokenProvenance::Static,
}
}
#[test]
fn probe_backoff_paces_at_replication_intervals() {
let backoff = repository_probe_policy().backoff;
assert_eq!(backoff.delay_after(1), Duration::from_secs(3));
assert_eq!(backoff.delay_after(2), Duration::from_secs(9));
}
#[tokio::test(start_paused = true)]
async fn host_side_retries_keep_the_operations_own_message() {
let calls = Mutex::new(0_u32);
let result = retry_git_messages(
&repository_probe_policy(),
Some(&snapshot(Duration::from_secs(1))),
"repository probe",
|| {
let attempt = {
let mut calls = calls.lock().unwrap();
*calls += 1;
*calls
};
async move {
if attempt < 3 {
Err(format!("remote: Repository not found. (attempt {attempt})"))
} else {
Ok(())
}
}
},
)
.await;
assert_eq!(result, Ok(()));
assert_eq!(*calls.lock().unwrap(), 3);
let permanent = retry_git_messages(
&repository_probe_policy(),
Some(&static_snapshot()),
"repository probe",
|| async { Err("remote: Repository not found.".to_owned()) },
)
.await;
assert_eq!(permanent, Err("remote: Repository not found.".to_owned()));
}
#[test]

1139
lib/apps/fabro-server/src/sandbox_access.rs Normal file
View file

File diff suppressed because it is too large Load diff

70
lib/apps/fabro-server/src/server.rs
View file

@ -65,9 +65,7 @@ use fabro_petri::controls::{RunControls, SteerError};
use fabro_petri::projector::Projector;
use fabro_redact::redact_jsonl_line;
use fabro_sandbox::details::sandbox_details;
use fabro_sandbox::driver::{DaytonaCredentials, ProviderAccess, ProviderConnectOptions};
use fabro_sandbox::reconnect::reconnect_for_run;
use fabro_sandbox::{SandboxInventory, daytona};
use fabro_slack::client::{PostedMessage as SlackPostedMessage, SlackClient};
use fabro_slack::config::{
SlackCredentialResolution,
@ -150,6 +148,10 @@ use crate::principal_middleware::{
};
use crate::request_id::{self, RequestId};
use crate::run_files::{FilesInFlight, new_files_in_flight};
use crate::sandbox_access::{
self, DAYTONA_CREDENTIAL_PROBE_TIMEOUT, DaytonaCredentials, DaytonaKeyCheck, ProviderAccess,
SandboxInventory,
};
use crate::server_secrets::ServerSecrets;
use crate::spawn_env::apply_render_graph_env;
use crate::worker_control::{
@ -1546,11 +1548,30 @@ impl AppState {
})
}
/// The same access in `fabro-sandbox`'s shape, for the callers still on
/// its reconnect path.
pub(crate) async fn legacy_provider_access(
&self,
) -> Result<fabro_sandbox::ProviderAccess, SecretStoreError> {
Ok(fabro_sandbox::ProviderAccess {
providers: self.server_settings().server.sandbox.providers.clone(),
daytona: self
.vault_secret(EnvVars::DAYTONA_API_KEY)
.await?
.map(|api_key| {
fabro_sandbox::DaytonaCredentials::from_api_key(api_key, |name| {
self.config_env_lookup(name)
})
.with_http_client(self.http_client().ok())
}),
})
}
pub(crate) async fn check_daytona_api_key(
&self,
api_key: String,
) -> anyhow::Result<daytona::DaytonaKeyCheck> {
self.check_daytona_api_key_with_timeout(api_key, daytona::DAYTONA_CREDENTIAL_PROBE_TIMEOUT)
) -> anyhow::Result<DaytonaKeyCheck> {
self.check_daytona_api_key_with_timeout(api_key, DAYTONA_CREDENTIAL_PROBE_TIMEOUT)
.await
}
@ -1558,8 +1579,9 @@ impl AppState {
&self,
api_key: String,
probe_timeout: Duration,
) -> anyhow::Result<daytona::DaytonaKeyCheck> {
daytona::check_daytona_api_key(&self.daytona_credentials(api_key), probe_timeout).await
) -> anyhow::Result<DaytonaKeyCheck> {
sandbox_access::check_daytona_api_key(&self.daytona_credentials(api_key), probe_timeout)
.await
}
/// Borrow the persistent store so sibling modules can open run readers
@ -2386,35 +2408,23 @@ fn build_sandbox_inventory(
http_client: Option<fabro_http::HttpClient>,
) -> SandboxInventory {
let provider_settings = &server_settings.server.sandbox.providers;
let access = ProviderAccess {
providers: provider_settings.clone(),
daytona: daytona_api_key.map(|api_key| {
DaytonaCredentials::from_api_key(api_key, |name| env_lookup(name))
.with_http_client(http_client)
}),
};
let mut inventory = SandboxInventory::empty();
if provider_settings.is_enabled(&SandboxProviderKind::LOCAL) {
inventory = inventory.with_host_directories(SandboxProviderKind::LOCAL);
}
if let Some(docker) = provider_settings.get(&SandboxProviderKind::DOCKER) {
if docker.enabled {
inventory = inventory.with_lazy(
SandboxProviderKind::DOCKER,
docker.clone(),
ProviderConnectOptions::default(),
);
}
if provider_settings.is_enabled(&SandboxProviderKind::DOCKER) {
inventory = inventory.with_lazy(SandboxProviderKind::DOCKER, access.clone());
}
if let Some(daytona) = provider_settings.get(&SandboxProviderKind::DAYTONA) {
if let Some(api_key) = daytona_api_key.filter(|_| daytona.enabled) {
let credentials = DaytonaCredentials::from_api_key(api_key, |name| env_lookup(name))
.with_http_client(http_client);
inventory = inventory.with_lazy(
SandboxProviderKind::DAYTONA,
daytona.clone(),
ProviderConnectOptions {
host_registry_root: None,
daytona: Some(credentials),
},
);
}
if provider_settings.is_enabled(&SandboxProviderKind::DAYTONA) && access.daytona.is_some() {
inventory = inventory.with_lazy(SandboxProviderKind::DAYTONA, access);
}
inventory
@ -2859,7 +2869,7 @@ async fn delete_run_sandbox_resource(
}
let access = state
.provider_access()
.legacy_provider_access()
.await
.map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?;
let sandbox = match reconnect_for_run(&record, &access, Some(id), None).await {

2
lib/apps/fabro-server/src/server/handler/sandbox.rs
View file

@ -748,7 +748,7 @@ async fn reconnect_run_sandbox_instance(
}
async fn load_provider_access(state: &AppState) -> Result<ProviderAccess, Response> {
state.provider_access().await.map_err(|err| {
state.legacy_provider_access().await.map_err(|err| {
tracing::error!(error = ?err, "Loading Daytona API key failed");
ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,

99
lib/apps/fabro-server/src/server/handler/sandboxes.rs
View file

@ -4,12 +4,12 @@ use axum::extract::{Path, State};
use axum::http::StatusCode;
use axum::routing::get;
use axum::{Json, Router};
use fabro_sandbox::SandboxLookupError;
use fabro_types::{SandboxInfo, SandboxListResponse, SandboxProviderKind};
use super::super::AppState;
use crate::error::ApiError;
use crate::principal_middleware::RequiredRunManagementActor;
use crate::sandbox_access::SandboxLookupError;
pub(super) fn routes() -> Router<Arc<AppState>> {
Router::new()
@ -77,16 +77,20 @@ fn provider_list(providers: &[SandboxProviderKind]) -> String {
#[cfg(test)]
mod tests {
use std::sync::Arc;
use axum::body::{Body, to_bytes};
use axum::http::{Request, StatusCode};
use fabro_sandbox::SandboxInventory;
use fabro_sandbox::driver::{ConnectedProvider, ProviderConnectOptions};
use fabro_sandbox::test_support::{managed_scripted_sandbox, scripted_inventory_provider};
use fabro_types::SandboxProviderKind;
use fabro_types::settings::server::{SandboxPluginSettings, ServerSandboxProviderSettings};
use fabro_types::settings::server::{
SandboxPluginSettings, ServerSandboxProviderSettings, ServerSandboxProvidersSettings,
};
use sandbox_driver::SandboxProvider;
use serde_json::{Value, json};
use tower::ServiceExt;
use crate::sandbox_access::test_support::{petri_scripted_sandbox, scripted_provider};
use crate::sandbox_access::{ProviderAccess, SandboxInventory};
use crate::test_support::{TestAppStateBuilder, build_test_router};
fn app_with_inventory(inventory: SandboxInventory) -> axum::Router {
@ -96,30 +100,36 @@ mod tests {
build_test_router(state)
}
/// A connected provider of `kind` holding fabro-managed sandboxes `ids`.
fn provider(kind: SandboxProviderKind, ids: &[&str]) -> ConnectedProvider {
scripted_inventory_provider(
kind,
ids.iter().map(|id| managed_scripted_sandbox(id)).collect(),
/// A provider of `kind` holding the sandboxes Petri created for a run,
/// `ids`.
fn provider(kind: &SandboxProviderKind, ids: &[&str]) -> Arc<dyn SandboxProvider> {
scripted_provider(
kind.as_str(),
ids.iter()
.map(|id| petri_scripted_sandbox(id, "01HY0000000000000000000000"))
.collect(),
)
}
/// A plugin kind whose executable does not exist, so every lookup fails
/// to connect.
fn with_unreachable_plugin(inventory: SandboxInventory, name: &str) -> SandboxInventory {
let settings = ServerSandboxProviderSettings {
enabled: true,
plugin: Some(SandboxPluginSettings {
path: Some(format!("/nonexistent/fabro-sandbox-{name}")),
dev: true,
..SandboxPluginSettings::default()
}),
};
inventory.with_lazy(
SandboxProviderKind::try_new(name).expect("valid kind"),
settings,
ProviderConnectOptions::default(),
)
let kind = SandboxProviderKind::try_new(name).expect("valid kind");
let mut providers = ServerSandboxProvidersSettings::default();
providers
.entries
.insert(kind.clone(), ServerSandboxProviderSettings {
enabled: true,
plugin: Some(SandboxPluginSettings {
path: Some(format!("/nonexistent/fabro-sandbox-{name}")),
dev: true,
..SandboxPluginSettings::default()
}),
});
inventory.with_lazy(kind, ProviderAccess {
providers,
daytona: None,
})
}
fn req_get(uri: &str) -> Request<Body> {
@ -139,10 +149,10 @@ mod tests {
#[tokio::test]
async fn list_returns_provider_backed_data_without_run_projection_state() {
let app = app_with_inventory(
SandboxInventory::empty()
.with_connected(provider(SandboxProviderKind::DOCKER, &["docker-native-id"])),
);
let app = app_with_inventory(SandboxInventory::empty().with_connected(
SandboxProviderKind::DOCKER,
provider(&SandboxProviderKind::DOCKER, &["docker-native-id"]),
));
let response = app.oneshot(req_get("/api/v1/sandboxes")).await.unwrap();
@ -158,8 +168,14 @@ mod tests {
async fn retrieve_searches_all_configured_providers() {
let app = app_with_inventory(
SandboxInventory::empty()
.with_connected(provider(SandboxProviderKind::DOCKER, &[]))
.with_connected(provider(SandboxProviderKind::DAYTONA, &["native-id"])),
.with_connected(
SandboxProviderKind::DOCKER,
provider(&SandboxProviderKind::DOCKER, &[]),
)
.with_connected(
SandboxProviderKind::DAYTONA,
provider(&SandboxProviderKind::DAYTONA, &["native-id"]),
),
);
let response = app
@ -177,8 +193,14 @@ mod tests {
async fn no_matching_sandbox_returns_404() {
let app = app_with_inventory(
SandboxInventory::empty()
.with_connected(provider(SandboxProviderKind::DOCKER, &[]))
.with_connected(provider(SandboxProviderKind::DAYTONA, &[])),
.with_connected(
SandboxProviderKind::DOCKER,
provider(&SandboxProviderKind::DOCKER, &[]),
)
.with_connected(
SandboxProviderKind::DAYTONA,
provider(&SandboxProviderKind::DAYTONA, &[]),
),
);
let response = app
@ -193,8 +215,14 @@ mod tests {
async fn duplicate_native_ids_return_409() {
let app = app_with_inventory(
SandboxInventory::empty()
.with_connected(provider(SandboxProviderKind::DOCKER, &["same-id"]))
.with_connected(provider(SandboxProviderKind::DAYTONA, &["same-id"])),
.with_connected(
SandboxProviderKind::DOCKER,
provider(&SandboxProviderKind::DOCKER, &["same-id"]),
)
.with_connected(
SandboxProviderKind::DAYTONA,
provider(&SandboxProviderKind::DAYTONA, &["same-id"]),
),
);
let response = app
@ -215,7 +243,10 @@ mod tests {
#[tokio::test]
async fn provider_lookup_uncertainty_returns_502() {
let app = app_with_inventory(with_unreachable_plugin(
SandboxInventory::empty().with_connected(provider(SandboxProviderKind::DOCKER, &[])),
SandboxInventory::empty().with_connected(
SandboxProviderKind::DOCKER,
provider(&SandboxProviderKind::DOCKER, &[]),
),
"e2b",
));

2
lib/apps/fabro-server/src/server/handler/sessions.rs
View file

@ -727,7 +727,7 @@ async fn build_agent(
AskFabroBuildError::SandboxUnavailable(anyhow::anyhow!("run sandbox was not created"))
})?;
let access = state
.provider_access()
.legacy_provider_access()
.await
.map_err(|err| AskFabroBuildError::Agent(anyhow::Error::new(err)))?;
let sandbox = reconnect_for_run(sandbox_instance, &access, Some(run_id), None)

5
lib/apps/fabro-server/src/test_support.rs
View file

@ -18,7 +18,6 @@ use fabro_config::user::default_storage_dir;
use fabro_config::{LlmLayer, RunLayer, ServerSettingsBuilder, Storage, envfile};
use fabro_db::DbPool;
use fabro_llm::lithos_catalog::Catalog;
use fabro_sandbox::SandboxInventory;
use fabro_static::EnvVars;
use fabro_store::{ArtifactStore, Database, test_support as store_test_support};
use fabro_types::settings::ServerAuthMethod;
@ -39,6 +38,7 @@ use crate::interp::process_env_var;
use crate::jwt_auth::{AuthMode, ConfiguredAuth};
#[cfg(test)]
use crate::principal_middleware::{AuthContextSlot, RequestAuthContext};
use crate::sandbox_access::SandboxInventory;
use crate::server::{
self, AppState, AppStateConfig, EnvLookup, ResolvedAppStateSettings, RouterOptions,
build_app_state,
@ -157,7 +157,8 @@ impl TestAppStateBuilder {
self
}
pub fn sandbox_inventory(mut self, sandbox_inventory: SandboxInventory) -> Self {
#[cfg(test)]
pub(crate) fn sandbox_inventory(mut self, sandbox_inventory: SandboxInventory) -> Self {
self.sandbox_inventory = Some(sandbox_inventory);
self
}