scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-30 01:53:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

fix: run built-in sandbox providers in process

Browse source 
Register lazy Host, Docker, and Daytona factories for Petri execution,
fork, and prune. Share server provider configuration, preserve lease
fingerprints, and source Daytona credentials from the vault.

Remove built-in plugin setup and skip gates; add a release-mode worker
and prune regression to catch the failure that blocked nightly builds.

Co-Authored-By: Codex <noreply@openai.com>
This commit is contained in:
Scott Werner 2026-09-24 17:14:47 -04:00
parent c874d81425
commit 6ac6d5495f
42 changed files with 751 additions and 913 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.config/nextest.toml
View file

@ -16,7 +16,7 @@ leak-timeout = "500ms"
slow-timeout = { period = "2s", terminate-after = 3 }
# fabro-petri's adapter tests run whole workflows on the host sandbox
# through the sandbox-driver plugin, and one of them calls the twin.
# through the in-process provider, and one of them calls the twin.
[[profile.default.overrides]]
filter = "package(fabro-petri)"
slow-timeout = { period = "5s", terminate-after = 4 }

83
.github/workflows/rust.yml vendored
View file

@ -144,31 +144,7 @@ jobs:
with:
cache-on-failure: true
- uses: taiki-e/install-action@773334c0e05d7e699e4d78234494308223f3a2cf # nextest
# Every Petri run takes its scope through a sandbox-driver plugin
# executable that Petri finds on PATH: `sandbox-driver-host` for the
# `local` provider, `sandbox-driver-docker` for `docker`. Installed
# at the commit Cargo.lock resolves sandbox-driver to, so the plugins
# and the in-process driver are one build; a from-source build, so the
# two executables are cached by OS and commit and only rebuilt when the
# lockfile moves the driver.
- name: Read the sandbox-driver commit Cargo.lock resolves
id: sandbox-driver
run: |
rev="$(cargo metadata --locked --format-version 1 | jq -r '.packages[] | select(.name == "sandbox-driver") | .source' | sed 's/.*#//' | sort -u)"
[[ "$rev" =~ ^[0-9a-f]{40}$ ]]
echo "rev=$rev" >> "$GITHUB_OUTPUT"
- name: Restore the sandbox-driver plugin executables
id: sandbox-driver-cache
uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4.3.0
with:
path: |
~/.cargo/bin/sandbox-driver-host
~/.cargo/bin/sandbox-driver-docker
key: sandbox-driver-plugins-${{ runner.os }}-${{ steps.sandbox-driver.outputs.rev }}
- name: Install the sandbox-driver plugin executables
if: steps.sandbox-driver-cache.outputs.cache-hit != 'true'
run: cargo install --locked --git https://github.com/lithoscomputer/sandbox-driver --rev "${{ steps.sandbox-driver.outputs.rev }}" sandbox-driver-host sandbox-driver-docker
# The images the suite's Docker tests run. The plugin pulls a missing
# The images the suite's Docker tests run. The provider pulls a missing
# image on first use, but a 1 GiB pull inside a run's wait is a flake,
# so pull them here, where a registry problem reads as one. Most tests
# leave the image to Petri, whose Docker scope runs on its default
@ -183,6 +159,8 @@ jobs:
docker pull "ghcr.io/lithoscomputer/ubuntu-24.04:slim-$pin"
docker pull ghcr.io/lithoscomputer/ubuntu-22.04:slim
- run: cargo nextest run --locked --workspace --status-level slow --profile ci
- name: Verify built-in Host execution and prune in a release build
run: cargo nextest run --locked --release -p fabro-cli --test it -E 'test(built_in_host_runs_and_prunes_without_plugins)' --profile ci
# The twin-mode ignored suites this job once ran belonged to fabro-agent,
# which pebble's coding agent replaced; the agent loop's workflow-level
# tests run in the suite above, and pebble's own suite covers the loop.
@ -195,9 +173,9 @@ jobs:
permissions:
contents: read
env:
# The Docker scenarios skip when the executable, the daemon or the
# The Docker scenarios skip when the daemon or the
# image is missing; in CI a skip is a failure.
FABRO_REQUIRE_SANDBOX_PLUGINS: "1"
FABRO_REQUIRE_SANDBOX_BACKENDS: "1"
steps:
- uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
with:
@ -218,36 +196,11 @@ jobs:
pin="$(sed -n 's/^const RUNNER_PIN: &str = "\([0-9a-f]*\)";$/\1/p' "$backend")"
test -n "$pin"
docker pull "ghcr.io/lithoscomputer/ubuntu-24.04:slim-$pin"
# Every Petri run takes its scope through a sandbox-driver plugin
# executable that Petri finds on PATH: `sandbox-driver-host` for the
# `local` provider, `sandbox-driver-docker` for `docker`. Installed
# at the commit Cargo.lock resolves sandbox-driver to, so the plugins
# and the in-process driver are one build; a from-source build, so the
# two executables are cached by OS and commit and only rebuilt when the
# lockfile moves the driver.
- name: Read the sandbox-driver commit Cargo.lock resolves
id: sandbox-driver
run: |
rev="$(cargo metadata --locked --format-version 1 | jq -r '.packages[] | select(.name == "sandbox-driver") | .source' | sed 's/.*#//' | sort -u)"
[[ "$rev" =~ ^[0-9a-f]{40}$ ]]
echo "rev=$rev" >> "$GITHUB_OUTPUT"
- name: Restore the sandbox-driver plugin executables
id: sandbox-driver-cache
uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4.3.0
with:
path: |
~/.cargo/bin/sandbox-driver-host
~/.cargo/bin/sandbox-driver-docker
key: sandbox-driver-plugins-${{ runner.os }}-${{ steps.sandbox-driver.outputs.rev }}
- name: Install the sandbox-driver plugin executables
if: steps.sandbox-driver-cache.outputs.cache-hit != 'true'
run: cargo install --locked --git https://github.com/lithoscomputer/sandbox-driver --rev "${{ steps.sandbox-driver.outputs.rev }}" sandbox-driver-host sandbox-driver-docker
# The workflow scenarios on the Docker provider. The scenarios are e2e
# tests (ignored by default); the key-free ones run here, the
# LLM-backed ones self-skip without credentials.
- run: cargo nextest run --locked --profile ci --status-level slow --run-ignored only -p fabro-cli --test it -E 'test(/::docker_/)'
# The Petri runs (not ignored: they skip without the host plugin, which
# the environment above forbids).
# Petri adapter runs use the built-in providers; Docker is required here.
- run: cargo nextest run --locked --profile ci --status-level slow -p fabro-petri
test-macos:
@ -267,29 +220,5 @@ jobs:
with:
cache-on-failure: true
- uses: taiki-e/install-action@773334c0e05d7e699e4d78234494308223f3a2cf # nextest
# Every Petri run takes its scope through a sandbox-driver plugin
# executable that Petri finds on PATH: `sandbox-driver-host` for the
# `local` provider, `sandbox-driver-docker` for `docker`. Installed
# at the commit Cargo.lock resolves sandbox-driver to, so the plugins
# and the in-process driver are one build; a from-source build, so the
# two executables are cached by OS and commit and only rebuilt when the
# lockfile moves the driver.
- name: Read the sandbox-driver commit Cargo.lock resolves
id: sandbox-driver
run: |
rev="$(cargo metadata --locked --format-version 1 | jq -r '.packages[] | select(.name == "sandbox-driver") | .source' | sed 's/.*#//' | sort -u)"
[[ "$rev" =~ ^[0-9a-f]{40}$ ]]
echo "rev=$rev" >> "$GITHUB_OUTPUT"
- name: Restore the sandbox-driver plugin executables
id: sandbox-driver-cache
uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4.3.0
with:
path: |
~/.cargo/bin/sandbox-driver-host
~/.cargo/bin/sandbox-driver-docker
key: sandbox-driver-plugins-${{ runner.os }}-${{ steps.sandbox-driver.outputs.rev }}
- name: Install the sandbox-driver plugin executables
if: steps.sandbox-driver-cache.outputs.cache-hit != 'true'
run: cargo install --locked --git https://github.com/lithoscomputer/sandbox-driver --rev "${{ steps.sandbox-driver.outputs.rev }}" sandbox-driver-host sandbox-driver-docker
# No Docker daemon on the macOS runner: the Docker tests skip there.
- run: cargo nextest run --locked --workspace --status-level slow --profile ci

38
Cargo.lock generated
View file

@ -2530,6 +2530,7 @@ dependencies = [
"fabro-interview",
"fabro-llm",
"fabro-petri",
"fabro-static",
"fabro-store",
"fabro-test",
"fabro-tool",
@ -2547,6 +2548,10 @@ dependencies = [
"petri-runtime",
"petri-store",
"petri-testkit",
"sandbox-driver",
"sandbox-driver-daytona",
"sandbox-driver-docker",
"sandbox-driver-host",
"serde",
"serde_json",
"sqlx",
@ -2654,8 +2659,6 @@ 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",
@ -5157,7 +5160,7 @@ checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220"
[[package]]
name = "petri-attractor-steps"
version = "0.1.0"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#cbab2c5f134dcbd625f554b6fd6d9704a31bcfd9"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#153d586871824b8023a5691e49a1e99421bae282"
dependencies = [
"async-trait",
"globset",
@ -5188,7 +5191,7 @@ dependencies = [
[[package]]
name = "petri-driver"
version = "0.1.0"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#cbab2c5f134dcbd625f554b6fd6d9704a31bcfd9"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#153d586871824b8023a5691e49a1e99421bae282"
dependencies = [
"async-trait",
"getrandom 0.3.4",
@ -5208,7 +5211,7 @@ dependencies = [
[[package]]
name = "petri-engine"
version = "0.1.0"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#cbab2c5f134dcbd625f554b6fd6d9704a31bcfd9"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#153d586871824b8023a5691e49a1e99421bae282"
dependencies = [
"petri-ir",
"serde",
@ -5220,7 +5223,7 @@ dependencies = [
[[package]]
name = "petri-execution"
version = "0.1.0"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#cbab2c5f134dcbd625f554b6fd6d9704a31bcfd9"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#153d586871824b8023a5691e49a1e99421bae282"
dependencies = [
"async-trait",
"petri-driver",
@ -5244,7 +5247,7 @@ dependencies = [
[[package]]
name = "petri-executor"
version = "0.1.0"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#cbab2c5f134dcbd625f554b6fd6d9704a31bcfd9"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#153d586871824b8023a5691e49a1e99421bae282"
dependencies = [
"async-trait",
"libc",
@ -5259,7 +5262,7 @@ dependencies = [
[[package]]
name = "petri-executor-sandbox"
version = "0.1.0"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#cbab2c5f134dcbd625f554b6fd6d9704a31bcfd9"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#153d586871824b8023a5691e49a1e99421bae282"
dependencies = [
"async-trait",
"petri-executor",
@ -5281,7 +5284,7 @@ dependencies = [
[[package]]
name = "petri-frontend"
version = "0.1.0"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#cbab2c5f134dcbd625f554b6fd6d9704a31bcfd9"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#153d586871824b8023a5691e49a1e99421bae282"
dependencies = [
"marked-yaml",
"petri-ir",
@ -5295,7 +5298,7 @@ dependencies = [
[[package]]
name = "petri-frontend-attractor"
version = "0.1.0"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#cbab2c5f134dcbd625f554b6fd6d9704a31bcfd9"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#153d586871824b8023a5691e49a1e99421bae282"
dependencies = [
"minijinja",
"petri-frontend",
@ -5312,7 +5315,7 @@ dependencies = [
[[package]]
name = "petri-frontend-fabro"
version = "0.1.0"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#cbab2c5f134dcbd625f554b6fd6d9704a31bcfd9"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#153d586871824b8023a5691e49a1e99421bae282"
dependencies = [
"petri-frontend",
"petri-frontend-attractor",
@ -5328,7 +5331,7 @@ dependencies = [
[[package]]
name = "petri-frontend-native"
version = "0.1.0"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#cbab2c5f134dcbd625f554b6fd6d9704a31bcfd9"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#153d586871824b8023a5691e49a1e99421bae282"
dependencies = [
"petri-frontend",
"petri-ir",
@ -5339,7 +5342,7 @@ dependencies = [
[[package]]
name = "petri-ir"
version = "0.1.0"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#cbab2c5f134dcbd625f554b6fd6d9704a31bcfd9"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#153d586871824b8023a5691e49a1e99421bae282"
dependencies = [
"regex",
"serde",
@ -5352,7 +5355,7 @@ dependencies = [
[[package]]
name = "petri-runtime"
version = "0.1.0"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#cbab2c5f134dcbd625f554b6fd6d9704a31bcfd9"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#153d586871824b8023a5691e49a1e99421bae282"
dependencies = [
"async-trait",
"petri-driver",
@ -5373,7 +5376,7 @@ dependencies = [
[[package]]
name = "petri-steps"
version = "0.1.0"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#cbab2c5f134dcbd625f554b6fd6d9704a31bcfd9"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#153d586871824b8023a5691e49a1e99421bae282"
dependencies = [
"async-trait",
"petri-executor",
@ -5389,7 +5392,7 @@ dependencies = [
[[package]]
name = "petri-store"
version = "0.1.0"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#cbab2c5f134dcbd625f554b6fd6d9704a31bcfd9"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#153d586871824b8023a5691e49a1e99421bae282"
dependencies = [
"async-trait",
"getrandom 0.3.4",
@ -5404,7 +5407,7 @@ dependencies = [
[[package]]
name = "petri-testkit"
version = "0.1.0"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#cbab2c5f134dcbd625f554b6fd6d9704a31bcfd9"
source = "git+https://github.com/lithoscomputer/petri.git?branch=main#153d586871824b8023a5691e49a1e99421bae282"
dependencies = [
"async-trait",
"petri-driver",
@ -5415,6 +5418,7 @@ dependencies = [
"petri-steps",
"petri-store",
"sandbox-driver",
"sandbox-driver-host",
"serde",
"serde_json",
"smol_str",

4
Cargo.toml
View file

@ -88,9 +88,7 @@ tokio-tungstenite = { version = "0.26", features = ["rustls-tls-webpki-roots"] }
futures-util = "0.3"
# sandbox-driver: the sandbox provider layer. Bundled Host, Docker, and
# Daytona providers link in-process; third-party providers run as stdio
# plugins through sandbox-driver-protocol. The CI plugin jobs install the
# driver executables at the commit `Cargo.lock` resolves `sandbox-driver` to,
# so the plugins and the in-process driver are one build.
# plugins through sandbox-driver-protocol.
sandbox-driver = { git = "https://github.com/lithoscomputer/sandbox-driver", branch = "main" }
sandbox-driver-protocol = { git = "https://github.com/lithoscomputer/sandbox-driver", branch = "main" }
sandbox-driver-host = { git = "https://github.com/lithoscomputer/sandbox-driver", branch = "main" }

2
docs/public/administration/sandboxing.mdx
View file

@ -7,6 +7,8 @@ Sandboxes isolate agent execution from the host machine. When an agent runs a sh
Fabro bundles three sandbox providers: `local` (no isolation), `docker` (container-level), and `daytona` (cloud VM). Additional providers run as [sandbox-driver](https://github.com/lithoscomputer/sandbox-driver) plugins configured under `[server.sandbox.providers.<kind>]`; an environment selects one by its kind name. See [Environments](/execution/environments) for full provider-specific configuration and [Server configuration](/administration/server-configuration#serversandboxproviders-section) for plugin settings.
The built-in providers run in process for execution, resume, and pruning. Fabro does not need `sandbox-driver-host`, `sandbox-driver-docker`, or `sandbox-driver-daytona` executables or checksum pins for these operations.
Operators can enable or disable which providers the server may launch with
`[server.sandbox.providers.<kind>]` in `settings.toml`. Missing bundled entries default to
`enabled = true`; setting `enabled = false` rejects new runs whose effective provider is disabled,

5
docs/public/administration/server-configuration.mdx
View file

@ -189,6 +189,11 @@ enabled = true
enabled = true
```
The built-in `local`, `docker`, and `daytona` providers run in process. `local` selects Petri's
Host provider. Built-in entries reject `path`, `sha256`, `dev`, `args`, `env`, and `inherit_env`;
no plugin executable or checksum is needed. The legacy built-in
`PETRI_SANDBOX_{HOST,DOCKER,DAYTONA}_{PLUGIN,SHA256}` variables are ignored.
Any other key names a [sandbox-driver](https://github.com/lithoscomputer/sandbox-driver) plugin:
an executable that speaks the sandbox-driver JSON-RPC protocol on stdin and stdout. The kind must
be lowercase ASCII letters, digits, and interior hyphens. The plugin starts with a scrubbed

16
lib/apps/fabro-cli/src/commands/run/petri_worker.rs
View file

@ -78,6 +78,7 @@ use fabro_petri::hooks::HooksSpec;
use fabro_petri::interview::{Approval, FabroInterviewer};
use fabro_petri::petri::OwnerId;
use fabro_petri::platform_records::{HttpPlatformRecords, PlatformRecords};
use fabro_petri::providers::{DaytonaCredentials, SandboxProviderConfig};
use fabro_petri::runtime::{self, RuntimeSpec};
use fabro_petri::secrets::VaultSecrets;
use fabro_petri::{HttpRunStore, admission};
@ -613,7 +614,13 @@ async fn runtime_spec(
None
}
};
let daytona = vault
.read()
.await
.get(EnvVars::DAYTONA_API_KEY)
.map(|key| DaytonaCredentials::from_api_key(key.to_owned(), provider_env));
Ok(RuntimeSpec {
sandbox: SandboxProviderConfig::from_lookup(daytona, provider_env),
settings_toml: None,
mcp_catalog_toml: None,
model_client,
@ -622,3 +629,12 @@ async fn runtime_spec(
run_tools,
})
}
/// Non-secret provider selection inherited from the server.
#[expect(
clippy::disallowed_methods,
reason = "worker boundary snapshots inherited provider selection"
)]
fn provider_env(name: &str) -> Option<String> {
std::env::var(name).ok()
}

5
lib/apps/fabro-cli/tests/it/scenario/artifacts.rs
View file

@ -7,7 +7,7 @@ use std::time::Duration;
use fabro_test::{fabro_snapshot, test_context};
use super::petri::{RunningServer, host_plugin, run_detached, wait_for_success};
use super::petri::{RunningServer, run_detached, wait_for_success};
use crate::cmd::support::{read_text, text_tree};
/// Three command stages that leave files under `assets/`. The second and
@ -45,9 +45,6 @@ fn artifact_workspace(context: &fabro_test::TestContext) -> PathBuf {
#[tokio::test(flavor = "multi_thread")]
async fn artifact_commands_read_the_artifacts_the_hooks_collected() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let server = RunningServer::start().await;
let workspace = artifact_workspace(&context);

146
lib/apps/fabro-cli/tests/it/scenario/petri.rs
View file

@ -5,18 +5,14 @@
//!
//! Each test starts its own foreground server on disk storage, because the
//! session's shared daemon keeps its object store in memory and the resume
//! scenario restarts the server. The runs take their host scope through the
//! sandbox-driver host plugin, so the tests skip, and say why, when the
//! executable is not found, unless `FABRO_REQUIRE_SANDBOX_PLUGINS` is set.
//! The plugin's path override crosses into the server and its workers the
//! way `PATH` does.
//! scenario restarts the server. Host scopes run in process.
//!
//! The harness here (the server, the detached run, the status and event
//! reads) is shared with the run-tools scenarios in `petri_tools.rs`.
#![expect(
clippy::disallowed_methods,
reason = "these scenarios start a real server subprocess, locate the plugin through the process environment, and poll processes"
reason = "these scenarios start a real server subprocess, poll processes"
)]
#![expect(
clippy::disallowed_types,
@ -47,36 +43,9 @@ use fabro_vault::{SecretType, Vault};
use crate::cmd::support::created_run_id;
use crate::support::{TEST_DEV_TOKEN, TEST_SESSION_SECRET, seed_dev_token_auth};
const HOST_PLUGIN: &str = "sandbox-driver-host";
pub(super) const REQUIRE_ENV: &str = "FABRO_REQUIRE_SANDBOX_PLUGINS";
pub(super) const RUN_TIMEOUT: Duration = Duration::from_mins(1);
pub(super) const POLL: Duration = Duration::from_millis(50);
/// The host plugin as Petri's lookup finds it: the override variable, else
/// the executable on `PATH`. `None`, after saying so, when the test should
/// skip; a panic when the environment forbids a skip.
pub(super) fn host_plugin() -> Option<PathBuf> {
let found = env::var_os(EnvVars::PETRI_SANDBOX_HOST_PLUGIN)
.map(PathBuf::from)
.or_else(|| {
env::split_paths(&env::var_os(EnvVars::PATH)?)
.map(|dir| dir.join(HOST_PLUGIN))
.find(|candidate| candidate.is_file())
});
if found.is_none() {
assert!(
env::var_os(REQUIRE_ENV).is_none(),
"{REQUIRE_ENV} is set, but {HOST_PLUGIN} is not on PATH and {} is unset",
EnvVars::PETRI_SANDBOX_HOST_PLUGIN
);
eprintln!(
"skipping: {HOST_PLUGIN} is not on PATH and {} is unset",
EnvVars::PETRI_SANDBOX_HOST_PLUGIN
);
}
found
}
/// A foreground server on its own disk storage, dev-token auth, started
/// from the compiled `fabro` binary. Dropping it kills the process.
pub(super) struct RunningServer {
@ -676,9 +645,6 @@ pub(super) fn wait_until_gate_is_polled(gate: &Path) {
/// records through the HTTP store, and its lease ended with it.
#[tokio::test(flavor = "multi_thread")]
async fn a_petri_run_executes_in_the_server_launched_worker() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let server = RunningServer::start().await;
let workspace = write_petri_workspace(&context, "echo hello from petri");
@ -730,9 +696,6 @@ async fn a_petri_run_executes_in_the_server_launched_worker() {
/// resume mode, which finishes the run with one terminal lifecycle record.
#[tokio::test(flavor = "multi_thread")]
async fn a_petri_run_resumes_in_a_new_worker_after_the_server_restarts() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let mut server = RunningServer::start().await;
let gate = context.temp_dir.join("resume.gate");
@ -748,7 +711,7 @@ async fn a_petri_run_resumes_in_a_new_worker_after_the_server_restarts() {
eprintln!("stage is waiting on the gate");
// The crash: the server first, so it never observes the worker exit,
// then the worker's whole process group, plugin and stage included.
// then the worker's whole process group, stage included.
server.kill();
fabro_proc::sigkill_process_group(worker);
let deadline = Instant::now() + Duration::from_secs(10);
@ -923,9 +886,6 @@ fn two_gates_dot(markers: &Path) -> String {
/// and the run's stream records the interview.
#[tokio::test(flavor = "multi_thread")]
async fn a_human_gate_in_the_worker_is_answered_through_the_api() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let server = RunningServer::start().await;
let markers = context.temp_dir.join("markers");
@ -980,9 +940,6 @@ async fn a_human_gate_in_the_worker_is_answered_through_the_api() {
/// the API in the other order, binds to its own branch.
#[tokio::test(flavor = "multi_thread")]
async fn two_parallel_gates_in_the_worker_each_bind_their_own_answer() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let server = RunningServer::start().await;
let markers = context.temp_dir.join("markers");
@ -1033,9 +990,6 @@ async fn two_parallel_gates_in_the_worker_each_bind_their_own_answer() {
/// pending.
#[tokio::test(flavor = "multi_thread")]
async fn an_unanswered_gate_in_the_worker_expires_with_its_default() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let server = RunningServer::start().await;
let markers = context.temp_dir.join("markers");
@ -1136,9 +1090,6 @@ fn pretty_filters(context: &fabro_test::TestContext) -> Vec<(String, String)> {
/// `wait` and `runs inspect` read the projection.
#[tokio::test(flavor = "multi_thread")]
async fn a_finished_petri_run_reads_back_through_the_cli() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let server = RunningServer::start().await;
let workspace = write_petri_workspace(&context, "echo hello from petri");
@ -1257,9 +1208,6 @@ async fn a_finished_petri_run_reads_back_through_the_cli() {
/// the gate and the attach exits with the run's status.
#[tokio::test(flavor = "multi_thread")]
async fn attach_asks_a_petri_gate_at_the_terminal_and_answers_it() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let server = RunningServer::start().await;
let markers = context.temp_dir.join("markers");
@ -1306,9 +1254,6 @@ async fn attach_asks_a_petri_gate_at_the_terminal_and_answers_it() {
/// the run goes on follow, and the terminal lifecycle record ends it.
#[tokio::test(flavor = "multi_thread")]
async fn events_follow_streams_a_petri_run_live_to_its_end() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let server = RunningServer::start().await;
let gate = context.temp_dir.join("go");
@ -1400,7 +1345,7 @@ fn three_stage_bundle(context: &fabro_test::TestContext, gate: &Path) -> PathBuf
/// Kill the server first, so it never observes the worker exit, then the
/// worker's whole process group, then the stage's own process group when a
/// stage was waiting on `gate`: a stage process runs in a group of its own
/// under the host plugin, and a machine crash takes it with everything
/// under the Host provider, and a machine crash takes it with everything
/// else, where a killed worker alone would leave it writing into the
/// workspace.
pub(super) fn crash(server: &mut RunningServer, worker: u32, gate: Option<&Path>) {
@ -1465,9 +1410,6 @@ fn three_stage_subjects(run_id: &str) -> Vec<String> {
/// snapshot (its partial output gone), and the next stage sees both.
#[tokio::test(flavor = "multi_thread")]
async fn a_crash_after_a_durable_finish_keeps_its_one_commit() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let mut server = RunningServer::start().await;
let gate = context.temp_dir.join("two.gate");
@ -1506,9 +1448,6 @@ async fn a_crash_after_a_durable_finish_keeps_its_one_commit() {
/// is not durable, the stage reruns once, and one commit exists for it.
#[tokio::test(flavor = "multi_thread")]
async fn a_crash_before_the_commit_lands_reruns_the_stage_once() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let mut server = RunningServer::start().await;
let gate = context.temp_dir.join("two.gate");
@ -1547,9 +1486,6 @@ async fn a_crash_before_the_commit_lands_reruns_the_stage_once() {
/// repository, the stage does not rerun, and one commit exists for it.
#[tokio::test(flavor = "multi_thread")]
async fn a_crash_before_the_record_reconciles_it_from_the_run_branch() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let mut server = RunningServer::start().await;
let gate = context.temp_dir.join("two.gate");
@ -1591,9 +1527,6 @@ async fn a_crash_before_the_record_reconciles_it_from_the_run_branch() {
/// repository, and the next stage sees the checkpoint's files.
#[tokio::test(flavor = "multi_thread")]
async fn a_deleted_workspace_is_restored_from_its_snapshot() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let mut server = RunningServer::start().await;
let gate = context.temp_dir.join("two.gate");
@ -1628,9 +1561,6 @@ async fn a_deleted_workspace_is_restored_from_its_snapshot() {
/// route reruns on the same files.
#[tokio::test(flavor = "multi_thread")]
async fn a_failure_route_sees_the_same_committed_files_after_a_crash() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let mut server = RunningServer::start().await;
let gate = context.temp_dir.join("fix.gate");
@ -1683,9 +1613,6 @@ async fn a_failure_route_sees_the_same_committed_files_after_a_crash() {
/// leaves it failed without launching a worker.
#[tokio::test(flavor = "multi_thread")]
async fn a_failed_checkpoint_fails_the_run_and_a_restart_leaves_it_failed() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let mut server = RunningServer::start().await;
let workspace = write_petri_workflow(
@ -1760,9 +1687,6 @@ async fn a_failed_checkpoint_fails_the_run_and_a_restart_leaves_it_failed() {
/// exit after the delete brings nothing back.
#[tokio::test(flavor = "multi_thread")]
async fn a_delete_right_after_the_run_reads_ended_is_accepted() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let server = RunningServer::start().await;
let workspace = write_petri_workspace(&context, "true");
@ -1805,3 +1729,65 @@ async fn a_delete_right_after_the_run_reads_ended_is_accepted() {
);
server.shutdown();
}
/// The release build must acquire and prune a real Host scope without any
/// plugin executable or checksum. This also runs in CI's release profile.
#[tokio::test(flavor = "multi_thread")]
async fn built_in_host_runs_and_prunes_without_plugins() {
let context = test_context!();
let path = "/usr/bin:/bin:/usr/sbin:/sbin";
let binary_dir = Path::new(env!("CARGO_BIN_EXE_fabro"))
.parent()
.expect("binary directory");
for kind in ["host", "docker", "daytona"] {
let executable = format!("sandbox-driver-{kind}");
for directory in env::split_paths(path).chain([binary_dir.to_path_buf()]) {
assert!(
!directory.join(&executable).exists(),
"test requires no {executable} in {}",
directory.display()
);
}
}
let server = RunningServer::start_with_env("", &[], &[
(EnvVars::PATH, path),
(
EnvVars::PETRI_SANDBOX_HOST_PLUGIN,
"/nonexistent/sandbox-driver-host",
),
(EnvVars::PETRI_SANDBOX_HOST_SHA256, "invalid-pin"),
(EnvVars::PETRI_SANDBOX_PLUGIN_DEV, "0"),
])
.await;
let workspace = write_petri_workspace(&context, "echo built-in > built-in.txt");
let run_id = run_detached(&context, &server, &workspace);
wait_for_success(&server, &run_id).await;
let run_dir = server.petri_run_dir(&run_id);
let scopes = run_dir.join("scopes");
let scope = std::fs::read_dir(&scopes)
.expect("the real Host scope exists")
.next()
.expect("one scope")
.expect("the scope reads")
.path();
assert_eq!(
std::fs::read_to_string(scope.join("work/built-in.txt"))
.expect("the worker wrote its file"),
"built-in\n"
);
let response = fabro_test::test_http_client()
.delete(format!("{}/api/v1/runs/{run_id}", server.api_base_url))
.bearer_auth(TEST_DEV_TOKEN)
.send()
.await
.expect("the delete sends");
let status = response.status();
let detail = response.text().await.unwrap_or_default();
assert_eq!(
status,
fabro_http::StatusCode::NO_CONTENT,
"prune failed: {detail}"
);
assert!(!scope.exists(), "prune removed the managed Host workspace");
server.shutdown();
}

35
lib/apps/fabro-cli/tests/it/scenario/petri_controls.rs
View file

@ -15,9 +15,8 @@
//! the worker never answers (a test hook mutes the worker's answers).
//!
//! The harness is `petri.rs`'s: a foreground server on disk storage, the
//! run started with `fabro run --detach`, and the host scope through the
//! sandbox-driver host plugin, so the tests skip, and say why, when the
//! plugin is not found.
//! run started with `fabro run --detach`, and the Host scope running in
//! process.
#![expect(
clippy::disallowed_methods,
@ -37,9 +36,9 @@ use fabro_test::{TwinScenario, TwinScenarios, TwinToolCall, test_context, twin_o
use serde_json::{Value, json};
use super::petri::{
RunningServer, answer, count_of, host_plugin, run_detached, run_detached_with, run_json,
run_status, run_stream, settled_stream, stream_names, wait_for_questions, wait_for_status,
wait_for_worker, wait_until_gate_is_polled, write_petri_workflow,
RunningServer, answer, count_of, run_detached, run_detached_with, run_json, run_status,
run_stream, settled_stream, stream_names, wait_for_questions, wait_for_status, wait_for_worker,
wait_until_gate_is_polled, write_petri_workflow,
};
use crate::support::TEST_DEV_TOKEN;
@ -331,9 +330,6 @@ async fn assert_petri_succeeded(server: &RunningServer, run_id: &str) {
/// both carry the pause and the unpause.
#[tokio::test(flavor = "multi_thread")]
async fn a_pause_holds_the_next_stage_until_the_unpause() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let server = RunningServer::start().await;
let gate = context.temp_dir.join("a.gate");
@ -405,9 +401,6 @@ async fn a_pause_holds_the_next_stage_until_the_unpause() {
/// no control request is recorded, since none went through the API.
#[tokio::test(flavor = "multi_thread")]
async fn the_user_signals_pause_and_unpause_the_worker() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let server = RunningServer::start().await;
let gate = context.temp_dir.join("a.gate");
@ -482,9 +475,6 @@ async fn the_user_signals_pause_and_unpause_the_worker() {
/// stream carries the `control.requested` record, and the run succeeds.
#[tokio::test(flavor = "multi_thread")]
async fn a_steer_reaches_the_agent_stage_on_the_twin() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let twin = twin_openai().await;
let namespace = format!("{}::{}", module_path!(), line!());
@ -614,9 +604,6 @@ async fn a_steer_reaches_the_agent_stage_on_the_twin() {
/// reaches that stage's session and no other.
#[tokio::test(flavor = "multi_thread")]
async fn two_live_agent_stages_are_steered_apart_by_their_labels() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let twin = twin_openai().await;
let namespace = format!("{}::{}", module_path!(), line!());
@ -776,9 +763,6 @@ async fn two_live_agent_stages_are_steered_apart_by_their_labels() {
/// `attractor.turn.interrupted` report, and the run succeeds.
#[tokio::test(flavor = "multi_thread")]
async fn an_interrupt_ends_the_turn_and_its_text_is_the_next_input() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let twin = twin_openai().await;
let namespace = format!("{}::{}", module_path!(), line!());
@ -912,9 +896,6 @@ const UNNAMED_INTERRUPT_REFUSAL: &str =
/// answer routes the run to its end.
#[tokio::test(flavor = "multi_thread")]
async fn an_interrupt_of_a_gate_stage_is_refused_with_no_live_turn() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let server = RunningServer::start().await;
let marker = context.temp_dir.join("yes.marker");
@ -993,9 +974,6 @@ async fn an_interrupt_of_a_gate_stage_is_refused_with_no_live_turn() {
/// muted through the server's test hook, forwarded to the worker by name.
#[tokio::test(flavor = "multi_thread")]
async fn a_control_the_worker_never_answers_is_pending() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let server =
RunningServer::start_with_env("", &[], &[(EnvVars::FABRO_TEST_CONTROL_ACKS_MUTED, "1")])
@ -1048,9 +1026,6 @@ async fn a_control_the_worker_never_answers_is_pending() {
/// without a new admission: a pause holds admission, never running work.)
#[tokio::test(flavor = "multi_thread")]
async fn a_run_paused_before_a_crash_resumes_paused() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let mut server = RunningServer::start().await;
let gate = context.temp_dir.join("a.gate");

56
lib/apps/fabro-cli/tests/it/scenario/petri_docker.rs
View file

@ -9,15 +9,13 @@
//! Petri created and reads a file the workflow wrote there, its model the
//! twin.
//!
//! The runs take their scope through the sandbox-driver Docker plugin on
//! this machine's daemon, so the tests skip, and say why, when the
//! executable is not found or no daemon answers, unless
//! `FABRO_REQUIRE_SANDBOX_PLUGINS` is set and the plugin is missing. The
//! server, the detached run and the crash come from `petri.rs`.
//! The runs use the built-in Docker provider on this machine's daemon.
//! Tests skip when no daemon answers, unless `FABRO_REQUIRE_SANDBOX_BACKENDS`
//! requires it. The server, detached run and crash come from `petri.rs`.
#![expect(
clippy::disallowed_methods,
reason = "these scenarios locate the plugin through the process environment and drive the Docker daemon with its CLI"
reason = "these scenarios inspect backend availability and drive the Docker daemon with its CLI"
)]
#![expect(clippy::print_stderr, reason = "a skipped test says why on its stderr")]
@ -33,40 +31,19 @@ use fabro_test::{
use serde_json::json;
use super::petri::{
REQUIRE_ENV, RunningServer, crash, run_detached_in, wait_for_status, wait_for_success,
wait_for_worker, write_petri_workflow,
RunningServer, crash, run_detached_in, wait_for_status, wait_for_success, wait_for_worker,
write_petri_workflow,
};
use crate::support::TEST_DEV_TOKEN;
const DOCKER_PLUGIN: &str = "sandbox-driver-docker";
/// The server-side environment the runs select.
const REQUIRE_ENV: &str = "FABRO_REQUIRE_SANDBOX_BACKENDS";
const ENVIRONMENT: &str = "docker";
/// The twin's model, for the Ask Fabro session.
const MODEL: &str = "gpt-5.4";
/// The Docker plugin as Petri's lookup finds it, with a daemon that
/// answers. `None`, after saying so, when the test should skip; a panic
/// when the environment forbids a skip and the plugin is missing.
fn docker_plugin() -> Option<PathBuf> {
let found = env::var_os(EnvVars::PETRI_SANDBOX_DOCKER_PLUGIN)
.map(PathBuf::from)
.or_else(|| {
env::split_paths(&env::var_os(EnvVars::PATH)?)
.map(|dir| dir.join(DOCKER_PLUGIN))
.find(|candidate| candidate.is_file())
});
let Some(found) = found else {
assert!(
env::var_os(REQUIRE_ENV).is_none(),
"{REQUIRE_ENV} is set, but {DOCKER_PLUGIN} is not on PATH and {} is unset",
EnvVars::PETRI_SANDBOX_DOCKER_PLUGIN
);
eprintln!(
"skipping: {DOCKER_PLUGIN} is not on PATH and {} is unset",
EnvVars::PETRI_SANDBOX_DOCKER_PLUGIN
);
return None;
};
/// A reachable Docker daemon. CI requires the backend instead of skipping.
fn docker_available() -> bool {
let daemon = Command::new("docker")
.args(["version", "--format", "{{.Server.Version}}"])
.stdout(Stdio::null())
@ -74,10 +51,13 @@ fn docker_plugin() -> Option<PathBuf> {
.status()
.is_ok_and(|status| status.success());
if !daemon {
assert!(
env::var_os(REQUIRE_ENV).is_none(),
"{REQUIRE_ENV} is set, but no Docker daemon answers"
);
eprintln!("skipping: no Docker daemon answers");
return None;
}
Some(found)
daemon
}
/// A server with a Docker environment beside the default local one.
@ -294,7 +274,7 @@ fn restore_actions(server: &RunningServer, run_id: &str) -> Vec<String> {
/// nothing of the workspace is on the host.
#[tokio::test(flavor = "multi_thread")]
async fn a_docker_run_publishes_every_stages_checkpoint_from_the_container() {
if docker_plugin().is_none() {
if !docker_available() {
return;
}
let context = test_context!();
@ -325,7 +305,7 @@ async fn a_docker_run_publishes_every_stages_checkpoint_from_the_container() {
/// the second stage sees the first stage's files and nothing else.
#[tokio::test(flavor = "multi_thread")]
async fn a_retained_container_whose_workspace_drifted_is_reset_on_restart() {
if docker_plugin().is_none() {
if !docker_available() {
return;
}
let context = test_context!();
@ -370,7 +350,7 @@ async fn a_retained_container_whose_workspace_drifted_is_reset_on_restart() {
/// repository, and the second stage sees the first stage's files.
#[tokio::test(flavor = "multi_thread")]
async fn a_lost_container_is_replaced_and_its_workspace_restored_from_the_snapshot() {
if docker_plugin().is_none() {
if !docker_available() {
return;
}
let context = test_context!();
@ -463,7 +443,7 @@ async fn question_inputs(twin: &fabro_test::TwinOpenAi, namespace: &str) -> Vec<
/// follow-up request carries the file's content back as the tool's answer.
#[tokio::test(flavor = "multi_thread")]
async fn an_ask_fabro_turn_reads_a_file_inside_the_runs_container() {
if docker_plugin().is_none() {
if !docker_available() {
return;
}
let context = test_context!();

18
lib/apps/fabro-cli/tests/it/scenario/petri_fork.rs
View file

@ -20,8 +20,7 @@ use std::process::{Command, Output};
use fabro_test::test_context;
use super::petri::{
RunningServer, host_plugin, run_detached, run_json, wait_for_status, wait_for_success,
write_petri_workflow,
RunningServer, run_detached, run_json, wait_for_status, wait_for_success, write_petri_workflow,
};
/// Three command stages that build on each other's files: `one` writes a
@ -178,9 +177,6 @@ fn read(workspace: &Path, name: &str) -> String {
/// new run says where it came from.
#[tokio::test(flavor = "multi_thread")]
async fn a_fork_at_the_first_stage_continues_with_the_rest_on_its_files() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let server = RunningServer::start().await;
let bundle = write_petri_workflow(&context, &three_stage_dot());
@ -260,9 +256,6 @@ async fn a_fork_at_the_first_stage_continues_with_the_rest_on_its_files() {
/// and finishes the run.
#[tokio::test(flavor = "multi_thread")]
async fn a_retry_reruns_the_failed_stage_and_succeeds_when_the_failure_was_transient() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let server = RunningServer::start().await;
let marker = context.temp_dir.join("flaky.marker");
@ -316,9 +309,6 @@ async fn a_retry_reruns_the_failed_stage_and_succeeds_when_the_failure_was_trans
/// names the run that replaced it.
#[tokio::test(flavor = "multi_thread")]
async fn a_rewind_supersedes_its_source() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let server = RunningServer::start().await;
let bundle = write_petri_workflow(&context, &three_stage_dot());
@ -381,9 +371,6 @@ async fn a_rewind_supersedes_its_source() {
/// its position and commit, as the CLI prints it and as the API serves it.
#[tokio::test(flavor = "multi_thread")]
async fn the_timeline_lists_every_checkpoint_with_its_commit() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let server = RunningServer::start().await;
let bundle = write_petri_workflow(&context, &three_stage_dot());
@ -453,9 +440,6 @@ async fn the_timeline_lists_every_checkpoint_with_its_commit() {
/// refuses it, and the refusal says why. The join, in the root, is.
#[tokio::test(flavor = "multi_thread")]
async fn a_fork_inside_a_parallel_branch_is_refused() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let server = RunningServer::start().await;
let bundle = write_petri_workflow(&context, &parallel_dot());

17
lib/apps/fabro-cli/tests/it/scenario/petri_tools.rs
View file

@ -1,4 +1,4 @@
//! Fabro's run tools inside a Petri run (integration plan item F3.4): a
//! Fabro's run tools inside a Petri run: a
//! workflow that enables `[run.agent] fabro_tools` runs on Petri in the
//! worker the server launched, and the agent stage's model, the twin, calls
//! the run tools the worker registered through Petri's host tool
@ -9,9 +9,7 @@
//!
//! The harness is `petri.rs`'s: a foreground server on disk storage with
//! the `openai` provider repointed at the twin, its key in the vault, and
//! the run started with `fabro run --detach`. The runs take their host
//! scope through the sandbox-driver host plugin, so the tests skip, and say
//! why, when it is not found.
//! the run started with `fabro run --detach`. Host scopes run in process.
#![expect(
clippy::disallowed_methods,
@ -33,7 +31,7 @@ use fabro_test::{TwinScenario, TwinScenarios, TwinToolCall, test_context, twin_o
use fabro_types::{WorkflowPath, WorkflowVersion};
use serde_json::{Value, json};
use super::petri::{RunningServer, host_plugin, run_detached_with, run_json, wait_for_status};
use super::petri::{RunningServer, run_detached_with, run_json, wait_for_status};
use crate::support::TEST_DEV_TOKEN;
const MODEL: &str = "gpt-5.4";
@ -227,9 +225,6 @@ async fn wait_for_children(server: &RunningServer, parent_id: &str) -> Vec<Value
/// Petri's record under the stage.
#[tokio::test(flavor = "multi_thread")]
async fn an_agent_starts_a_child_run_with_a_run_tool_inside_a_petri_run() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let twin = twin_openai().await;
let namespace = format!("{}::{}", module_path!(), line!());
@ -322,9 +317,6 @@ async fn an_agent_starts_a_child_run_with_a_run_tool_inside_a_petri_run() {
/// call.
#[tokio::test(flavor = "multi_thread")]
async fn a_run_hook_blocks_a_run_tool_inside_a_petri_run() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let twin = twin_openai().await;
let namespace = format!("{}::{}", module_path!(), line!());
@ -391,9 +383,6 @@ async fn a_run_hook_blocks_a_run_tool_inside_a_petri_run() {
/// parent session.
#[tokio::test(flavor = "multi_thread")]
async fn a_sub_agent_calls_an_inherited_run_tool_inside_a_petri_run() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let twin = twin_openai().await;
let namespace = format!("{}::{}", module_path!(), line!());

32
lib/apps/fabro-cli/tests/it/workflow/docker.rs
View file

@ -1,11 +1,7 @@
//! The Docker provider for the workflow scenarios: an environment on
//! [`DOCKER_IMAGE`], on an isolated server.
//!
//! Petri serves every provider through a sandbox-driver plugin executable it
//! finds on `PATH` (`sandbox-driver-docker` here); CI installs the
//! executables at the `sandbox-driver` commit in `Cargo.lock`, and a developer
//! installs them with
//! `cargo install --locked --git https://github.com/lithoscomputer/sandbox-driver --rev <rev> sandbox-driver-host sandbox-driver-docker`.
//! Petri uses the built-in Docker provider; no plugin executable is needed.
//! A scenario configured here runs against its own server so the environment
//! it creates never leaks into the shared session server.
@ -18,7 +14,7 @@
reason = "a skipped scenario says why on the test's stderr"
)]
use std::path::{Path, PathBuf};
use std::path::Path;
use std::process::{Command, Stdio};
use fabro_test::{TestContext, expect_reqwest_status};
@ -26,11 +22,10 @@ use serde_json::json;
use crate::cmd::support::server_endpoint;
/// Set in CI so a missing executable or daemon fails the test instead of
/// Set in CI so a missing daemon or image fails the test instead of
/// skipping it.
const REQUIRE_ENV: &str = "FABRO_REQUIRE_SANDBOX_PLUGINS";
const REQUIRE_ENV: &str = "FABRO_REQUIRE_SANDBOX_BACKENDS";
const DOCKER_IMAGE: &str = "buildpack-deps:noble";
const DOCKER_PLUGIN: &str = "sandbox-driver-docker";
/// The environment id the scenario selects with `--environment`.
pub(crate) const ENVIRONMENT: &str = "docker";
@ -39,17 +34,6 @@ pub(crate) const ENVIRONMENT: &str = "docker";
/// prerequisites are missing and the test should skip.
pub(crate) fn configure(context: &mut TestContext) -> Option<&'static str> {
let required = std::env::var_os(REQUIRE_ENV).is_some();
if plugin_executable().is_none() {
assert!(
!required,
"{REQUIRE_ENV} is set but {DOCKER_PLUGIN} is not on PATH"
);
eprintln!(
"skipping: {DOCKER_PLUGIN} is not on PATH; install the sandbox-driver executables at \
the rev Cargo.toml pins"
);
return None;
}
if !docker_image_available() {
assert!(
!required,
@ -75,14 +59,6 @@ methods = ["dev-token"]
Some(ENVIRONMENT)
}
/// The Docker plugin executable on `PATH`, when installed.
fn plugin_executable() -> Option<PathBuf> {
let path = std::env::var_os("PATH")?;
std::env::split_paths(&path)
.map(|dir| dir.join(DOCKER_PLUGIN))
.find(|candidate| candidate.is_file())
}
fn docker_image_available() -> bool {
Command::new("docker")
.args(["image", "inspect", DOCKER_IMAGE])

7
lib/apps/fabro-cli/tests/it/workflow/mod.rs
View file

@ -178,10 +178,9 @@ fn run_stream_items(run_dir: &Path) -> Vec<RunStreamItem> {
/// - `docker`: the Docker provider, an environment on `buildpack-deps:noble`
/// created on an isolated server.
///
/// Petri serves each provider through the matching sandbox-driver plugin
/// executable on `PATH`. The `docker` variant skips without
/// `sandbox-driver-docker` or without a Docker daemon that has the image,
/// unless `FABRO_REQUIRE_SANDBOX_PLUGINS` is set, as CI sets it.
/// The built-in providers run in process. The `docker` variant skips when
/// no Docker daemon has the required image, unless CI requires the backend
/// with `FABRO_REQUIRE_SANDBOX_BACKENDS`.
macro_rules! sandbox_tests {
($name:ident) => {
sandbox_tests!($name, keys = []);

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

@ -35,8 +35,6 @@ fabro-workflow = { path = "../../components/fabro-workflow" }
fabro-workflow-version = { path = "../../components/fabro-workflow-version" }
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

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

@ -5,6 +5,7 @@ use anyhow::{Result, anyhow};
use fabro_api::types;
use fabro_config::{RunLayer, SettingsLayer, WorkflowSettingsBuilder, project};
use fabro_manifest::CollectedWorkflowClosure;
use fabro_petri::providers::SandboxProviderConfig;
use fabro_petri::run_graph;
use fabro_petri::runtime::RuntimeSpec;
@ -55,6 +56,7 @@ fn offline_runtime(run: Option<&RunLayer>) -> RuntimeSpec {
..SettingsLayer::default()
};
RuntimeSpec {
sandbox: SandboxProviderConfig::default(),
settings_toml: toml::to_string(&layer).ok(),
mcp_catalog_toml: None,
model_client: None,

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

@ -21,7 +21,7 @@
//! Credentials arrive explicitly. Nothing here reads the process
//! environment for a secret: the Daytona key comes from the vault through
//! [`DaytonaCredentials`]. The Docker client resolves its endpoint from the
//! same variables Petri forwards to its Docker plugin (`DOCKER_HOST` and
//! same variables Fabro forwards to its worker (`DOCKER_HOST` and
//! its TLS companions), so the server and the run's containers meet on one
//! daemon.
@ -32,7 +32,8 @@ use std::time::Duration;
use anyhow::Context as _;
use fabro_config::Storage;
use fabro_static::EnvVars;
use fabro_petri::providers;
pub(crate) use fabro_petri::providers::DaytonaCredentials;
use fabro_types::settings::server::{
SandboxPluginSettings, ServerSandboxProviderSettings, ServerSandboxProvidersSettings,
};
@ -46,8 +47,6 @@ use sandbox_driver::{
Sandbox, SandboxFilter, SandboxId, SandboxProvider, SandboxSource, SandboxSpec, SandboxState,
WaitOptions,
};
use sandbox_driver_daytona::{DaytonaConfig, DaytonaProvider};
use sandbox_driver_docker::DockerProvider;
use sandbox_driver_host::HostProvider;
use sandbox_driver_protocol::{PluginConfig, PluginSupervisor};
use tokio::sync::OnceCell;
@ -63,78 +62,9 @@ pub(crate) const PETRI_RUN_LABEL: &str = "petri.run";
/// it up under the same name.
const PLUGIN_BINARY_PREFIX: &str = "sandbox-driver";
/// `User-Agent` Fabro presents to remote sandbox control planes.
const USER_AGENT: &str = concat!("fabro-server/", env!("CARGO_PKG_VERSION"));
/// Budget for the credential probe `fabro doctor` and the install flow run.
pub(crate) const DAYTONA_CREDENTIAL_PROBE_TIMEOUT: Duration = Duration::from_secs(20);
/// Explicit Daytona credentials: the SDK's configuration with the API key
/// always present and a `Debug` that never prints it. The process
/// environment is never consulted.
#[derive(Clone)]
pub(crate) struct DaytonaCredentials(DaytonaConfig);
impl DaytonaCredentials {
/// Credentials for `api_key` against Daytona's public control plane,
/// presenting Fabro's `User-Agent`.
#[must_use]
pub(crate) fn new(api_key: String) -> Self {
Self(DaytonaConfig {
api_key: Some(api_key),
user_agent: Some(USER_AGENT.to_string()),
..DaytonaConfig::default()
})
}
/// Credentials for a vault API key, with the control-plane URL and
/// organization taken from `lookup` (server configuration). Nothing is
/// read implicitly.
pub(crate) fn from_api_key(api_key: String, lookup: impl Fn(&str) -> Option<String>) -> Self {
Self::new(api_key)
.with_api_url(
lookup(EnvVars::DAYTONA_API_URL).or_else(|| lookup(EnvVars::DAYTONA_SERVER_URL)),
)
.with_organization_id(lookup(EnvVars::DAYTONA_ORGANIZATION_ID))
}
/// The control-plane URL; Daytona's public API when `None`.
#[must_use]
pub(crate) fn with_api_url(mut self, api_url: Option<String>) -> Self {
self.0.api_url = api_url;
self
}
#[must_use]
pub(crate) fn with_organization_id(mut self, organization_id: Option<String>) -> Self {
self.0.organization_id = organization_id;
self
}
/// A shared HTTP client; tests pass a no-proxy client here.
#[must_use]
pub(crate) fn with_http_client(mut self, http_client: Option<fabro_http::HttpClient>) -> Self {
self.0.http_client = http_client;
self
}
/// The SDK configuration the driver's Daytona provider connects with.
#[must_use]
fn config(&self) -> &DaytonaConfig {
&self.0
}
}
impl std::fmt::Debug for DaytonaCredentials {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("DaytonaCredentials")
.field("api_url", &self.0.api_url)
.field("organization_id", &self.0.organization_id)
.field("target", &self.0.target)
.finish_non_exhaustive()
}
}
/// What the server needs to reach every provider a run record can name:
/// its provider settings (which kinds are enabled, which run as plugins),
/// the Daytona credentials from the vault, and the storage root under
@ -236,19 +166,15 @@ pub(crate) async fn connect_provider(
};
Ok(match kind.bundled() {
Some(BundledProvider::Local) => Arc::new(HostProvider::new()),
Some(BundledProvider::Docker) => {
Arc::new(DockerProvider::connect_unverified().map_err(driver)?)
}
Some(BundledProvider::Docker) => providers::connect_docker().map_err(driver)?,
Some(BundledProvider::Daytona) => {
let credentials = access
.daytona
.as_ref()
.ok_or(ConnectError::MissingDaytonaCredentials)?;
Arc::new(
DaytonaProvider::connect_explicit(credentials.config().clone())
.await
.map_err(driver)?,
)
providers::connect_daytona(credentials)
.await
.map_err(driver)?
}
None => {
let plugin = settings
@ -1241,22 +1167,6 @@ mod tests {
assert_eq!(config.inherit_env, vec!["PATH"]);
}
#[test]
fn daytona_credentials_debug_never_prints_the_key() {
let credentials = DaytonaCredentials::from_api_key("dtn_secret_key".to_string(), |name| {
(name == EnvVars::DAYTONA_ORGANIZATION_ID).then(|| "org-1".to_string())
});
// The rendering stays out of the assertion messages: a failure must
// not print the key it is checking for.
let rendered = format!("{credentials:?}");
assert!(!rendered.contains("dtn_secret_key"));
assert!(rendered.contains("org-1"));
assert_eq!(
credentials.config().api_key.as_deref(),
Some("dtn_secret_key")
);
}
#[test]
fn missing_scopes_render_as_the_provider_reports_them() {
let check = DaytonaKeyCheck {

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

@ -63,6 +63,7 @@ use fabro_llm::{ClientOptions, FabroClient};
use fabro_mcp_store::McpServerStore;
use fabro_petri::controls::{RunControls, SteerError};
use fabro_petri::projector::Projector;
use fabro_petri::providers::SandboxProviderConfig;
use fabro_petri::prune::{self, PruneError, PruneRequest};
use fabro_redact::redact_jsonl_line;
use fabro_slack::client::{PostedMessage as SlackPostedMessage, SlackClient};
@ -1545,6 +1546,18 @@ impl AppState {
.with_http_client(self.http_client().ok())
}
/// The same provider selection for execution, fork and prune; credentials
/// arrive from the caller's vault read, never the process environment.
pub(crate) fn sandbox_provider_config(
&self,
daytona_api_key: Option<String>,
) -> SandboxProviderConfig {
SandboxProviderConfig::from_lookup(
daytona_api_key.map(|key| self.daytona_credentials(key)),
|name| self.config_env_lookup(name),
)
}
/// Everything a reconnect needs to reach a run's provider: the server's
/// provider settings and the Daytona credentials from the vault (`None`
/// when no key is stored).
@ -2875,7 +2888,18 @@ async fn delete_run_sandbox_resource(
.run_scratch(&id)
.root()
.join("petri");
let daytona_api_key = state
.vault_secret(EnvVars::DAYTONA_API_KEY)
.await
.map_err(|err| {
error!(error = ?err, "Loading sandbox credentials failed");
ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
"secret store operation failed",
)
})?;
let report = prune::prune(PruneRequest {
sandbox: state.sandbox_provider_config(daytona_api_key),
run_id: id.to_string(),
run_dir,
store: state.petri_runs.shared_store(),

14
lib/apps/fabro-server/src/server/handler/lineage.rs
View file

@ -1,5 +1,5 @@
//! A run's checkpoint timeline, and the runs made from it: fork, rewind and
//! retry (the integration plan's F5.1).
//! retry.
//!
//! The timeline is the run's `checkpoint` platform records, labelled with
//! the stages the projector folded them onto. A fork resolves a target on
@ -26,6 +26,7 @@ use fabro_petri::SqliteRunStore;
use fabro_petri::fork::{self as petri_fork, ForkError, ForkRequest};
use fabro_petri::petri::RunStore;
use fabro_petri::platform_records::SqlitePlatformRecords;
use fabro_static::EnvVars;
use fabro_store::{PlatformRecordKind, RunProjection};
use fabro_types::{FailureReason, Principal, RunId};
use fabro_util::error as error_util;
@ -281,6 +282,16 @@ async fn fork_at(
.await
.map_err(|err| fork_error(&err))?;
let daytona_api_key = state
.vault_secret(EnvVars::DAYTONA_API_KEY)
.await
.map_err(|err| {
error!(error = ?err, "Loading sandbox credentials failed");
ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
"secret store operation failed",
)
})?;
let new_run_id = RunId::new();
let storage = Storage::new(state.server_storage_dir());
let source_run_dir = storage.run_scratch(&id).root().to_path_buf();
@ -299,6 +310,7 @@ async fn fork_at(
.map_err(workflow_operation_error)?;
let seeded = petri_fork::fork(ForkRequest {
sandbox: state.sandbox_provider_config(daytona_api_key),
source: id,
fork: new_run_id,
source_run_dir: source_run_dir.join("petri"),

7
lib/apps/fabro-server/src/server/petri_runs.rs
View file

@ -50,6 +50,7 @@ use fabro_petri::recovery::{self, Recovery, RecoveryRequest};
use fabro_petri::runtime::{self, RuntimeSpec};
use fabro_petri::secrets::VaultSecrets;
use fabro_petri::{SqliteRunStore, admission, projection, run_graph};
use fabro_static::EnvVars;
use fabro_store::platform_records::{RunLifecycleKind, RunLifecycleRecord};
use fabro_types::settings::McpTransport;
use fabro_types::settings::run::{ApprovalMode, McpServerSettings, RunMode};
@ -94,6 +95,7 @@ pub(crate) fn runtime_spec(
}
};
RuntimeSpec {
sandbox: state.sandbox_provider_config(None),
settings_toml,
mcp_catalog_toml,
model_client,
@ -456,6 +458,9 @@ pub(crate) async fn execute(state: Arc<AppState>, run_id: RunId) {
))),
&run_state.spec.settings.run,
);
let mut runtime = runtime_spec(&state, &eligible, dry_run);
runtime.sandbox =
state.sandbox_provider_config(vault.get(EnvVars::DAYTONA_API_KEY).map(str::to_owned));
let request = RunRequest {
run_id: run_id.to_string(),
run_dir: run_dir.join("petri"),
@ -468,7 +473,7 @@ pub(crate) async fn execute(state: Arc<AppState>, run_id: RunId) {
.observe_store(Arc::new(SqliteRunStore::new(state.db_pool.clone()))),
state: Arc::clone(&state),
}),
runtime: runtime_spec(&state, &eligible, dry_run),
runtime,
provider: run_state.spec.settings.run.environment.provider.clone(),
cancel,
// The in-process test path drives no pause: the server's transport

89
lib/apps/fabro-server/src/spawn_env.rs
View file

@ -51,21 +51,12 @@ const WORKER_ENV_ALLOWLIST: &[&str] = &[
EnvVars::AWS_CONTAINER_CREDENTIALS_RELATIVE_URI,
EnvVars::AWS_CONTAINER_CREDENTIALS_FULL_URI,
EnvVars::AWS_CONTAINER_AUTHORIZATION_TOKEN_FILE,
// Petri's sandbox-driver plugins are resolved in the worker, where a
// Petri run executes: the plugin path, checksum and dev-mode overrides
// cross with `PATH`, so the worker finds the plugins the server would.
// A plugin the server's settings configure is set on top of these by
// `sandbox_plugin_env`, for every configured kind.
EnvVars::PETRI_SANDBOX_HOST_PLUGIN,
EnvVars::PETRI_SANDBOX_HOST_SHA256,
EnvVars::PETRI_SANDBOX_DOCKER_PLUGIN,
EnvVars::PETRI_SANDBOX_DOCKER_SHA256,
EnvVars::PETRI_SANDBOX_DAYTONA_PLUGIN,
EnvVars::PETRI_SANDBOX_DAYTONA_SHA256,
// Preserve generic plugin configuration. Built-in providers run in process
// and never receive executable paths or checksum overrides.
EnvVars::PETRI_SANDBOX_PLUGIN_DEV,
EnvVars::PETRI_SANDBOX_DOCKER_HOST_ADDRESS,
EnvVars::PETRI_SANDBOX_ACTION_HOST_IMAGE,
// The Docker daemon selection: the worker's Docker plugin reads these
// The Docker daemon selection: the worker's Docker provider reads these
// from its own process, so the worker's sandboxes go to the daemon the
// server uses (a remote or TLS daemon, a named context), not the
// default socket.
@ -75,10 +66,11 @@ const WORKER_ENV_ALLOWLIST: &[&str] = &[
EnvVars::DOCKER_API_VERSION,
EnvVars::DOCKER_CONFIG,
EnvVars::DOCKER_CONTEXT,
// Daytona's control-plane selection, the non-secret half: the plugin
// reads them from the worker. The API key comes from the vault, set on
// the command by the launch (`WorkerLaunchSpec::daytona_api_key`).
// Daytona's non-secret selection. The worker reads the API key from
// the vault and supplies it explicitly to the in-process provider.
EnvVars::DAYTONA_API_URL,
EnvVars::DAYTONA_SERVER_URL,
EnvVars::DAYTONA_TARGET,
EnvVars::DAYTONA_ORGANIZATION_ID,
// A test's checkpoint gates: the worker's hooks hold at a named point
// until the test releases them, so a crash can be placed there.
@ -110,8 +102,8 @@ fn apply_worker_env_with(
/// The plugin variables Petri reads in the worker, derived from the
/// server's `[server.sandbox.providers.<kind>]` settings: for every enabled
/// kind that carries plugin settings, `PETRI_SANDBOX_<KIND>_PLUGIN` from
/// its `path` and `PETRI_SANDBOX_<KIND>_SHA256` from its `sha256`, and
/// third-party kind that carries plugin settings, `PETRI_SANDBOX_<KIND>_PLUGIN`
/// from its `path` and `PETRI_SANDBOX_<KIND>_SHA256` from its `sha256`, and
/// `PETRI_SANDBOX_PLUGIN_DEV=1` when any of them sets `dev`. The kind is
/// uppercased with hyphens as underscores, as Petri names the variable. A
/// kind whose settings name no path is left to Petri's own lookup
@ -123,6 +115,9 @@ pub(crate) fn sandbox_plugin_env(
let mut env = Vec::new();
let mut dev = false;
for (kind, plugin) in providers.enabled_plugins() {
if kind.bundled().is_some() {
continue;
}
let upper = kind.as_str().to_ascii_uppercase().replace('-', "_");
if let Some(path) = &plugin.path {
env.push((format!("PETRI_SANDBOX_{upper}_PLUGIN"), path.clone()));
@ -260,6 +255,11 @@ mod tests {
"https://daytona.internal/api".to_string(),
),
("DAYTONA_ORGANIZATION_ID".to_string(), "org-1".to_string()),
(
"DAYTONA_SERVER_URL".to_string(),
"https://daytona-alias.internal/api".to_string(),
),
("DAYTONA_TARGET".to_string(), "us".to_string()),
("DAYTONA_API_KEY".to_string(), "leak".to_string()),
]);
let mut cmd = env_command();
@ -295,18 +295,18 @@ mod tests {
Some("xterm-256color")
);
assert_eq!(actual.get("NO_COLOR").map(String::as_str), Some("1"));
// Petri's plugin overrides cross so the worker resolves the same
// sandbox-driver plugins the server would.
assert_eq!(
actual.get("PETRI_SANDBOX_HOST_PLUGIN").map(String::as_str),
Some("/opt/petri/sandbox-driver-host")
);
// Built-in plugin paths and pins never reach the worker.
for kind in ["HOST", "DOCKER", "DAYTONA"] {
for suffix in ["PLUGIN", "SHA256"] {
assert!(!actual.contains_key(&format!("PETRI_SANDBOX_{kind}_{suffix}")));
}
}
assert_eq!(
actual.get("PETRI_SANDBOX_PLUGIN_DEV").map(String::as_str),
Some("1")
);
// The Docker daemon selection crosses whole, so the worker's Docker
// plugin drives the daemon the server uses.
// provider drives the daemon the server uses.
assert_eq!(
actual.get("DOCKER_HOST").map(String::as_str),
Some("tcp://build-daemon.internal:2376")
@ -341,6 +341,11 @@ mod tests {
actual.get("DAYTONA_ORGANIZATION_ID").map(String::as_str),
Some("org-1")
);
assert_eq!(
actual.get("DAYTONA_SERVER_URL").map(String::as_str),
Some("https://daytona-alias.internal/api")
);
assert_eq!(actual.get("DAYTONA_TARGET").map(String::as_str), Some("us"));
assert!(!actual.contains_key("DAYTONA_API_KEY"));
assert_eq!(actual.get("CLICOLOR").map(String::as_str), Some("0"));
assert_eq!(actual.get("CLICOLOR_FORCE").map(String::as_str), Some("1"));
@ -395,9 +400,8 @@ mod tests {
}
/// A configured plugin reaches the worker under the names Petri reads,
/// a configured path wins over the ambient variable of the same name,
/// a kind the settings leave to `PATH` keeps the ambient one, and a
/// disabled kind's plugin never crosses.
/// while built-in paths and pins and disabled third-party plugins never
/// cross.
#[tokio::test]
async fn configured_plugins_reach_the_worker_and_win_over_ambient_variables() {
let mut providers = ServerSandboxProvidersSettings::default();
@ -433,6 +437,13 @@ mod tests {
"/ambient/sandbox-driver-daytona".to_string(),
),
]);
let mut env = env;
for kind in ["HOST", "DOCKER", "DAYTONA"] {
env.insert(
format!("PETRI_SANDBOX_{kind}_SHA256"),
"invalid-pin".to_string(),
);
}
let mut cmd = env_command();
apply_worker_env_with(&mut cmd, &sandbox_plugin_env(&providers), &|name| {
env.get(name).map(OsString::from)
@ -453,25 +464,11 @@ mod tests {
Some("1"),
"one plugin in dev mode puts the worker's lookup in dev mode"
);
assert_eq!(
actual
.get("PETRI_SANDBOX_DOCKER_PLUGIN")
.map(String::as_str),
Some("/opt/fabro/plugins/sandbox-driver-docker"),
"the settings win over the ambient variable"
);
assert_eq!(
actual.get("PETRI_SANDBOX_HOST_PLUGIN").map(String::as_str),
Some("/ambient/sandbox-driver-host"),
"a kind without settings keeps the allowlisted ambient variable"
);
assert_eq!(
actual
.get("PETRI_SANDBOX_DAYTONA_PLUGIN")
.map(String::as_str),
Some("/ambient/sandbox-driver-daytona"),
"settings without a path leave the ambient variable in place"
);
for kind in ["HOST", "DOCKER", "DAYTONA"] {
for suffix in ["PLUGIN", "SHA256"] {
assert!(!actual.contains_key(&format!("PETRI_SANDBOX_{kind}_{suffix}")));
}
}
assert!(
!actual.contains_key("PETRI_SANDBOX_FLY_IO_PLUGIN"),
"a disabled kind's plugin does not cross"

102
lib/apps/fabro-server/tests/it/scenario/petri.rs
View file

@ -7,14 +7,11 @@
//! run, which the CLI's scenario tests cover with the real binary
//! (`lib/apps/fabro-cli/tests/it/scenario/petri.rs`).
//!
//! The runs that execute take their host scope through the sandbox-driver
//! host plugin, so those tests skip, and say why, when the executable is not
//! found, unless `FABRO_REQUIRE_SANDBOX_PLUGINS` is set. The create-time
//! refusals need no plugin and always run.
//! Built-in Host scopes run in process without a plugin executable.
#![expect(
clippy::disallowed_methods,
reason = "the tests locate the plugin executable through the process environment"
reason = "the tests inspect backend availability through the process environment"
)]
#![expect(clippy::print_stderr, reason = "a skipped test says why on its stderr")]
@ -46,11 +43,7 @@ use crate::helpers::{
test_settings, wait_for_run_status,
};
const HOST_PLUGIN: &str = "sandbox-driver-host";
const HOST_PLUGIN_OVERRIDE: &str = "PETRI_SANDBOX_HOST_PLUGIN";
const DOCKER_PLUGIN: &str = "sandbox-driver-docker";
const DOCKER_PLUGIN_OVERRIDE: &str = "PETRI_SANDBOX_DOCKER_PLUGIN";
const REQUIRE_ENV: &str = "FABRO_REQUIRE_SANDBOX_PLUGINS";
const REQUIRE_ENV: &str = "FABRO_REQUIRE_SANDBOX_BACKENDS";
const OPENAI_MODEL: &str = "gpt-5.4";
@ -110,46 +103,8 @@ const PARALLEL_DOT: &str = r#"digraph Parallel {
pub(super) const PLAIN_SETTINGS: &str = "_version = 1\n\n[workflow]\ngraph = \"workflow.fabro\"\n";
/// The host plugin as Petri's lookup finds it: the override variable, else
/// the executable on `PATH`. `None`, after saying so, when the test should
/// skip; a panic when the environment forbids a skip.
pub(super) fn host_plugin() -> Option<PathBuf> {
let found = env::var_os(HOST_PLUGIN_OVERRIDE)
.map(PathBuf::from)
.or_else(|| {
env::split_paths(&env::var_os("PATH")?)
.map(|dir| dir.join(HOST_PLUGIN))
.find(|candidate| candidate.is_file())
});
if found.is_none() {
assert!(
env::var_os(REQUIRE_ENV).is_none(),
"{REQUIRE_ENV} is set, but {HOST_PLUGIN} is not on PATH and {HOST_PLUGIN_OVERRIDE} is unset"
);
eprintln!("skipping: {HOST_PLUGIN} is not on PATH and {HOST_PLUGIN_OVERRIDE} is unset");
}
found
}
/// The Docker plugin as Petri's lookup finds it, with a daemon that
/// answers. `None`, after saying so, when the test should skip; a panic
/// when the environment forbids a skip and the plugin is missing.
fn docker_plugin() -> Option<PathBuf> {
let found = env::var_os(DOCKER_PLUGIN_OVERRIDE)
.map(PathBuf::from)
.or_else(|| {
env::split_paths(&env::var_os("PATH")?)
.map(|dir| dir.join(DOCKER_PLUGIN))
.find(|candidate| candidate.is_file())
});
let Some(found) = found else {
assert!(
env::var_os(REQUIRE_ENV).is_none(),
"{REQUIRE_ENV} is set, but {DOCKER_PLUGIN} is not on PATH and {DOCKER_PLUGIN_OVERRIDE} is unset"
);
eprintln!("skipping: {DOCKER_PLUGIN} is not on PATH and {DOCKER_PLUGIN_OVERRIDE} is unset");
return None;
};
/// A reachable Docker daemon. CI requires the backend instead of skipping.
fn docker_available() -> bool {
let daemon = Command::new("docker")
.args(["version", "--format", "{{.Server.Version}}"])
.stdout(Stdio::null())
@ -157,10 +112,13 @@ fn docker_plugin() -> Option<PathBuf> {
.status()
.is_ok_and(|status| status.success());
if !daemon {
assert!(
env::var_os(REQUIRE_ENV).is_none(),
"{REQUIRE_ENV} is set, but no Docker daemon answers"
);
eprintln!("skipping: no Docker daemon answers");
return None;
}
Some(found)
daemon
}
/// Register a version whose entrypoint is `workflow.fabro`, with the given
@ -290,9 +248,6 @@ async fn create_run_response(app: &axum::Router, intent: serde_json::Value) -> s
/// run succeeded, and Petri's record of the run says the same.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_hello_bundle_runs_on_petri() {
if host_plugin().is_none() {
return;
}
let workspace = tempfile::tempdir().expect("workspace tempdir");
let twin = twin_openai().await;
let namespace = format!("{}::{}", module_path!(), line!());
@ -367,9 +322,6 @@ async fn the_hello_bundle_runs_on_petri() {
/// A command-only bundle runs on Petri, and Petri's record agrees.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_command_bundle_runs_on_petri() {
if host_plugin().is_none() {
return;
}
let workspace = tempfile::tempdir().expect("workspace tempdir");
let settings = settings_from_toml("_version = 1\n\n[run.environment]\nid = \"local\"\n");
let state = test_app_state_with_options(settings, 5);
@ -413,9 +365,6 @@ async fn a_command_bundle_runs_on_petri() {
/// under the fork, and the fork carries the branch results.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_parallel_bundle_projects_its_branches_through_the_server() {
if host_plugin().is_none() {
return;
}
let workspace = tempfile::tempdir().expect("workspace tempdir");
let settings = settings_from_toml("_version = 1\n\n[run.environment]\nid = \"local\"\n");
let state = test_app_state_with_options(settings, 5);
@ -574,9 +523,6 @@ async fn wait_for_question(app: &axum::Router, run_id: &str) -> serde_json::Valu
/// interview as a legacy stage's would.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_human_gate_is_answered_through_the_questions_api() {
if host_plugin().is_none() {
return;
}
let workspace = tempfile::tempdir().expect("workspace tempdir");
let markers = tempfile::tempdir().expect("marker tempdir");
let settings = settings_from_toml("_version = 1\n\n[run.environment]\nid = \"local\"\n");
@ -696,9 +642,6 @@ async fn a_human_gate_is_answered_through_the_questions_api() {
/// `sandbox cp` reach the sandbox after the run.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_runs_projection_carries_its_host_sandbox_instance() {
if host_plugin().is_none() {
return;
}
let workspace = tempfile::tempdir().expect("workspace tempdir");
let settings = settings_from_toml("_version = 1\n\n[run.environment]\nid = \"local\"\n");
let state = test_app_state_with_options(settings, 5);
@ -764,9 +707,6 @@ async fn a_runs_projection_carries_its_host_sandbox_instance() {
/// host workspace Petri kept along with the run.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn deleting_a_run_prunes_its_host_workspace_through_petri() {
if host_plugin().is_none() {
return;
}
let workspace = tempfile::tempdir().expect("workspace tempdir");
let settings = settings_from_toml("_version = 1\n\n[run.environment]\nid = \"local\"\n");
let state = test_app_state_with_options(settings, 5);
@ -884,9 +824,6 @@ fn delete(run_id: &str) -> Request<Body> {
/// end after the delete brings nothing back.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_delete_right_after_the_run_reads_ended_is_accepted() {
if host_plugin().is_none() {
return;
}
let workspace = tempfile::tempdir().expect("workspace tempdir");
let settings = settings_from_toml("_version = 1\n\n[run.environment]\nid = \"local\"\n");
let state = test_app_state_with_options(settings, 5);
@ -955,7 +892,7 @@ async fn a_delete_right_after_the_run_reads_ended_is_accepted() {
/// attaches to it on the daemon.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_runs_projection_carries_its_docker_sandbox_instance() {
if docker_plugin().is_none() {
if !docker_available() {
return;
}
let settings = settings_from_toml("_version = 1\n\n[run.environment]\nid = \"docker\"\n");
@ -1072,7 +1009,7 @@ async fn a_runs_projection_carries_its_docker_sandbox_instance() {
/// run label.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_server_attaches_to_the_container_petri_created() {
if docker_plugin().is_none() {
if !docker_available() {
return;
}
let twin = twin_openai().await;
@ -1293,7 +1230,7 @@ fn get(path: &str) -> Request<Body> {
/// The image the server's `docker-small` environment names in the tests
/// below: a runner image with `git` for the checkpoint commit, and not the
/// plugin's default, so the container proves the catalog's image reached it.
/// provider's default, so the container proves the catalog's image reached it.
const CATALOG_IMAGE: &str = "ghcr.io/lithoscomputer/ubuntu-22.04:slim";
/// A Docker environment in the server's catalog, with the image it runs.
@ -1397,7 +1334,7 @@ async fn admitted_root_graph(app: &axum::Router, run_id: &str) -> serde_json::Va
/// A bundle that names a server environment it does not declare admits: the
/// catalog's `[environments.docker-small]` reaches Petri through the
/// settings layer, its image lands on the lowered environment, and, with
/// the Docker plugin and a daemon, the run's container runs that image.
/// a Docker daemon, the run's container runs that image.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_bundle_naming_a_catalog_environment_runs_on_docker_with_its_image() {
let settings = settings_from_toml("_version = 1\n\n[run.environment]\nid = \"local\"\n");
@ -1431,7 +1368,7 @@ async fn a_bundle_naming_a_catalog_environment_runs_on_docker_with_its_image() {
graph["params"]["fabro.launch"]
);
if docker_plugin().is_none() {
if !docker_available() {
return;
}
start_run(&app, &run_id).await;
@ -1583,9 +1520,6 @@ const AGENT_DOT: &str = r#"digraph Agent {
/// server's tool to the model.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_bundle_naming_a_catalog_mcp_server_lists_its_tools_to_the_model() {
if host_plugin().is_none() {
return;
}
let workspace = tempfile::tempdir().expect("workspace tempdir");
let twin = twin_openai().await;
let namespace = format!("{}::{}", module_path!(), line!());
@ -1770,9 +1704,6 @@ async fn assert_run_goal(app: &axum::Router, namespace: &str, run_id: &str, goal
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_goal_override_is_the_goal_the_stages_execute_with() {
const GOAL: &str = "Add a limerick to the README instead of a haiku";
if host_plugin().is_none() {
return;
}
let [(workflow_path, workflow), (settings_path, settings)] = hello_files();
let (_state, app, namespace, run_id) = run_hello_agent(
&[(workflow_path, &workflow), (settings_path, &settings)],
@ -1788,9 +1719,6 @@ async fn a_goal_override_is_the_goal_the_stages_execute_with() {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_run_goal_file_layer_is_the_goal_the_stages_execute_with() {
const GOAL: &str = "Write a limerick about workflow engines into the README";
if host_plugin().is_none() {
return;
}
let [(workflow_path, workflow), _] = hello_files();
let settings =
"_version = 1\n[workflow]\ngraph = \"workflow.fabro\"\n[run.goal]\nfile = \"goal.md\"\n";

11
lib/apps/fabro-server/tests/it/scenario/petri_stream.rs
View file

@ -6,9 +6,7 @@
//! concurrent child executions' events.
//!
//! The runs execute in the server process under the handler-registry test
//! override and take their host scope through the sandbox-driver host
//! plugin, so the tests skip, and say why, when the executable is not
//! found (see `petri.rs`).
//! override and take their Host scope through the built-in provider.
//!
//! With `FABRO_CAPTURE_PETRI_FIXTURES` set, a scenario also writes its
//! settled projection and full stream as JSON under the web app's test
@ -17,7 +15,7 @@
#![expect(
clippy::disallowed_methods,
reason = "the tests locate the plugin executable and the capture switch through the process environment"
reason = "the tests read the capture switch through the process environment"
)]
#![expect(clippy::print_stderr, reason = "a skipped test says why on its stderr")]
@ -35,7 +33,7 @@ use http_body_util::BodyExt;
use tokio::time::timeout;
use tower::ServiceExt;
use super::petri::{PLAIN_SETTINGS, host_plugin, intent, register_version, settled_state};
use super::petri::{PLAIN_SETTINGS, intent, register_version, settled_state};
use crate::helpers::{
api, create_and_start_run_from_intent, repo_root, response_json, run_json, settings_from_toml,
test_app_state_with_options, test_app_with_scheduler, wait_for_run_status,
@ -231,9 +229,6 @@ fn wait_for_marker(path: &std::path::Path) {
/// gap, no duplicate, with the notice between the branches' events.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_reconnecting_client_receives_every_stream_item_once_in_order() {
if host_plugin().is_none() {
return;
}
let workspace = tempfile::tempdir().expect("workspace tempdir");
let markers = tempfile::tempdir().expect("marker tempdir");
let settings = settings_from_toml("_version = 1\n\n[run.environment]\nid = \"local\"\n");

5
lib/components/fabro-petri/Cargo.toml
View file

@ -23,6 +23,11 @@ fabro-api = { path = "../../foundation/fabro-api" }
fabro-client = { path = "../../foundation/fabro-client" }
fabro-db = { path = "../../foundation/fabro-db" }
fabro-http.workspace = true
fabro-static = { path = "../../foundation/fabro-static" }
sandbox-driver.workspace = true
sandbox-driver-host.workspace = true
sandbox-driver-docker.workspace = true
sandbox-driver-daytona.workspace = true
fabro-interview = { path = "../fabro-interview" }
fabro-store = { path = "../fabro-store" }
fabro-types = { path = "../../foundation/fabro-types" }

12
lib/components/fabro-petri/README.md
View file

@ -172,16 +172,15 @@ Integration tests live under `tests/`:
- `runs.rs` runs the `hello` bundle in memory through `Runtime::standard()`
with the Fabro frontend and the model-free stub registry, then a
command-only workflow on the host sandbox through the real step registry.
Both skip, and say why, when the `sandbox-driver-host` plugin executable
is not on `PATH` (every run takes its scope's environment through it);
the sandbox-plugins CI job requires them.
Both acquire real Host scopes through the built-in in-process provider;
no plugin executable or checksum is required.
- `check.rs` admits the `hello` bundle and round-trips its graph through
the blob store, binds the launch, admits a version whose `workflow.toml`
names `engine = "petri"`, reads the project settings from the map, and
refuses an unknown attribute, an unknown `[workflow]` key
(`unsupported.workflow_toml.key`, named in `workflow.toml`) and, with a
model client over the test catalog, an unknown model
(`attractor.model.unknown`). No plugin is needed.
(`attractor.model.unknown`). No sandbox is acquired.
- `sqlite_store.rs` runs Petri's store conformance suite
(`petri_testkit::run_store::conformance`) against `SqliteRunStore`, plus the
operator release, lease exclusivity, a crash between appends, and blob
@ -206,8 +205,7 @@ Integration tests live under `tests/`:
client over a vault that holds the key, and checks the skills step
searched the configured Fabro home.
Those four need the host plugin like `runs.rs` does, and `model.rs` also
starts the twin.
Those four use the in-process Host provider; `model.rs` also starts the twin.
- `projection.rs` builds the view live (every append signals the
projector) for the `hello` bundle on the stub registry, a command-only
@ -217,7 +215,7 @@ starts the twin.
recovers a crash between the record commit and the view transaction by
applying only the missing suffix, with the positions and `stream_seq`
continuing; runs two projectors over one store with child executions; and
holds the view at a torn tail. All skip without the host plugin.
holds the view at a torn tail. These run without a Host plugin.
The conformance suite over `HttpRunStore` needs a server to talk to, so it
lives with the server's integration tests

4
lib/components/fabro-petri/src/fork.rs
View file

@ -59,9 +59,12 @@ use crate::checkpoint::{CheckpointKey, RunWorkspaces};
use crate::platform_records::{PlatformRecordError, PlatformRecords};
use crate::projection::FoldState;
use crate::projector::ProjectError;
use crate::providers::{self, SandboxProviderConfig};
/// One fork to seed.
pub struct ForkRequest {
/// The provider configuration used by this server-side operation.
pub sandbox: SandboxProviderConfig,
/// The run whose records are copied.
pub source: RunId,
/// The new run's id: its Petri run key and its own run scratch.
@ -178,6 +181,7 @@ pub async fn fork(request: ForkRequest) -> Result<Forked, ForkError> {
let mut options = RunOptions::new(&request.fork_run_dir);
options.run_key = Some(fork_key.clone());
let runtime = Runtime::standard()
.in_process_providers(providers::built_in_providers(&request.sandbox))
.options(options)
.store(Arc::clone(&request.store));
let forked = host::fork_from(&runtime, &*source_logs, request.position, ForkOptions {

3
lib/components/fabro-petri/src/lib.rs
View file

@ -12,6 +12,8 @@
//! run's records are its source of truth in Fabro's tables;
//! - [`runtime`]: the Petri runtime Fabro assembles, at create time and at
//! execution;
//! - [`providers`]: lazy in-process Host, Docker and Daytona factories, sharing
//! explicit configuration with the server's sandbox access;
//! - [`check`]: Petri compiles a workflow version's bundle at create time, and
//! its diagnostics come back in a shape Fabro maps onto its own;
//! - [`admission`]: the admitted graphs in Fabro's blob store, named on the run
@ -76,6 +78,7 @@ pub mod petri;
pub mod platform_records;
pub mod projection;
pub mod projector;
pub mod providers;
pub mod prune;
pub mod recovery;
pub mod run_graph;

336
lib/components/fabro-petri/src/providers.rs Normal file
View file

@ -0,0 +1,336 @@
//! Built-in sandbox providers shared by Petri execution and server access.
//!
//! Configuration is captured by the caller, with Daytona credentials from
//! the vault. Factories connect lazily per run; a Host-only run needs neither
//! Docker nor Daytona. Host registry ownership stays with Petri: server
//! attach uses an observer instead of these factories.
use std::sync::Arc;
use async_trait::async_trait;
use fabro_static::EnvVars;
use petri_runtime::{
InProcessProviders, ProviderContext, ProviderFactory, ProviderNetwork, fingerprint,
};
use sandbox_driver::{AuthError, Error, ProviderKind, SandboxProvider};
use sandbox_driver_daytona::{DaytonaConfig, DaytonaProvider};
use sandbox_driver_docker::DockerProvider;
use sandbox_driver_host::HostProvider;
const USER_AGENT: &str = concat!("fabro-server/", env!("CARGO_PKG_VERSION"));
/// Explicit Daytona credentials: the SDK's configuration with the API key
/// always present and a `Debug` that never prints it. The process
/// environment is never consulted.
#[derive(Clone)]
pub struct DaytonaCredentials(DaytonaConfig);
impl DaytonaCredentials {
/// Credentials for `api_key` against Daytona's public control plane,
/// presenting Fabro's `User-Agent`.
#[must_use]
pub fn new(api_key: String) -> Self {
Self(DaytonaConfig {
api_key: Some(api_key),
user_agent: Some(USER_AGENT.to_string()),
..DaytonaConfig::default()
})
}
/// Credentials for a vault API key, with the control-plane URL and
/// organization taken from `lookup` (server configuration). Nothing is
/// read implicitly.
pub fn from_api_key(api_key: String, lookup: impl Fn(&str) -> Option<String>) -> Self {
Self::new(api_key)
.with_api_url(
lookup(EnvVars::DAYTONA_API_URL).or_else(|| lookup(EnvVars::DAYTONA_SERVER_URL)),
)
.with_organization_id(lookup(EnvVars::DAYTONA_ORGANIZATION_ID))
.with_target(lookup(EnvVars::DAYTONA_TARGET))
}
/// The control-plane URL; Daytona's public API when `None`.
#[must_use]
pub fn with_api_url(mut self, api_url: Option<String>) -> Self {
self.0.api_url = api_url;
self
}
#[must_use]
pub fn with_organization_id(mut self, organization_id: Option<String>) -> Self {
self.0.organization_id = organization_id;
self
}
/// The configured Daytona placement target, kept unset when omitted.
#[must_use]
pub fn with_target(mut self, target: Option<String>) -> Self {
self.0.target = target;
self
}
/// A shared HTTP client; tests pass a no-proxy client here.
#[must_use]
pub fn with_http_client(mut self, http_client: Option<fabro_http::HttpClient>) -> Self {
self.0.http_client = http_client;
self
}
/// The SDK configuration the driver's Daytona provider connects with.
#[must_use]
fn config(&self) -> &DaytonaConfig {
&self.0
}
}
impl std::fmt::Debug for DaytonaCredentials {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("DaytonaCredentials")
.field("api_url", &self.0.api_url)
.field("organization_id", &self.0.organization_id)
.field("target", &self.0.target)
.finish_non_exhaustive()
}
}
/// The provider configuration supplied to a run or prune. Defaults are
/// local Docker selection and no Daytona credentials; constructing this
/// configuration never connects to a backend or requires a credential.
#[derive(Clone, Debug, Default)]
pub struct SandboxProviderConfig {
pub docker_host: Option<String>,
pub docker_host_address: Option<String>,
pub daytona: Option<DaytonaCredentials>,
}
impl SandboxProviderConfig {
/// Snapshot the Docker network/fingerprint selection from the same
/// environment the Docker client uses. Daytona credentials are explicit.
pub fn from_lookup(
daytona: Option<DaytonaCredentials>,
lookup: impl Fn(&str) -> Option<String>,
) -> Self {
Self {
docker_host: lookup(EnvVars::DOCKER_HOST),
docker_host_address: lookup(EnvVars::PETRI_SANDBOX_DOCKER_HOST_ADDRESS)
.filter(|value| !value.trim().is_empty()),
daytona,
}
}
}
/// The Docker connection used by both the server and Petri. The driver reads
/// the caller process's Docker endpoint/TLS environment; health is checked
/// by the caller so diagnostics can report an unavailable daemon.
#[expect(
clippy::result_large_err,
reason = "preserve sandbox-driver's typed error, as required by Petri's ProviderFactory contract"
)]
pub fn connect_docker() -> sandbox_driver::Result<Arc<dyn SandboxProvider>> {
Ok(Arc::new(DockerProvider::connect_unverified()?))
}
/// Connect using only the vault credentials and explicit control-plane
/// configuration; no ambient secret fallback is permitted.
pub async fn connect_daytona(
credentials: &DaytonaCredentials,
) -> sandbox_driver::Result<Arc<dyn SandboxProvider>> {
Ok(Arc::new(
DaytonaProvider::connect_explicit(credentials.config().clone()).await?,
))
}
/// One lazy factory per built-in kind. Missing Daytona credentials fail
/// only when a Daytona scope is acquired, never for admission or a Host run.
pub fn built_in_providers(config: &SandboxProviderConfig) -> InProcessProviders {
InProcessProviders::new()
.with(Arc::new(HostFactory))
.with(Arc::new(DockerFactory {
host: config.docker_host.clone(),
host_address: config.docker_host_address.clone(),
}))
.with(Arc::new(DaytonaFactory(config.daytona.clone())))
}
struct HostFactory;
#[async_trait]
impl ProviderFactory for HostFactory {
fn kind(&self) -> &'static str {
"host"
}
fn fingerprint_seed(&self, context: &ProviderContext) -> String {
fingerprint::host(
context
.host_registry()
.expect("Petri supplies the Host registry"),
)
}
fn network(&self) -> ProviderNetwork {
ProviderNetwork::host()
}
async fn connect(
&self,
context: &ProviderContext,
) -> sandbox_driver::Result<Arc<dyn SandboxProvider>> {
let registry = context
.host_registry()
.expect("Petri supplies the Host registry");
Ok(Arc::new(HostProvider::with_registry(registry).await?))
}
}
struct DockerFactory {
host: Option<String>,
host_address: Option<String>,
}
impl DockerFactory {
fn seed(&self) -> String {
fingerprint::docker(self.host.as_deref())
}
}
#[async_trait]
impl ProviderFactory for DockerFactory {
fn kind(&self) -> &'static str {
"docker"
}
fn fingerprint_seed(&self, _context: &ProviderContext) -> String {
self.seed()
}
fn network(&self) -> ProviderNetwork {
ProviderNetwork::docker(self.host.as_deref(), self.host_address.as_deref())
}
async fn connect(
&self,
_context: &ProviderContext,
) -> sandbox_driver::Result<Arc<dyn SandboxProvider>> {
connect_docker()
}
}
struct DaytonaFactory(Option<DaytonaCredentials>);
impl DaytonaFactory {
fn seed(&self) -> String {
let config = self.0.as_ref().map(DaytonaCredentials::config);
fingerprint::daytona(
config.and_then(|config| config.api_url.as_deref()),
config.and_then(|config| config.organization_id.as_deref()),
config.and_then(|config| config.target.as_deref()),
)
}
}
#[async_trait]
impl ProviderFactory for DaytonaFactory {
fn kind(&self) -> &'static str {
"daytona"
}
fn fingerprint_seed(&self, _context: &ProviderContext) -> String {
self.seed()
}
fn region(&self) -> Option<&str> {
self.0
.as_ref()
.and_then(|credentials| credentials.config().target.as_deref())
.filter(|region| !region.is_empty())
}
fn network(&self) -> ProviderNetwork {
ProviderNetwork::none()
}
async fn connect(
&self,
_context: &ProviderContext,
) -> sandbox_driver::Result<Arc<dyn SandboxProvider>> {
let credentials = self.0.as_ref().ok_or_else(|| Error::Auth(AuthError::new(
ProviderKind::try_new("daytona").expect("the built-in provider kind is valid"),
"Daytona requires DAYTONA_API_KEY in the vault; run `fabro secret set DAYTONA_API_KEY`",
)))?;
connect_daytona(credentials).await
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn docker_fingerprint_keeps_the_plugin_namespace_for_unset_and_configured_hosts() {
for (host, expected) in [
(None, "docker:default"),
(Some(" "), "docker:default"),
(Some(" tcp://daemon:2376 "), "docker:tcp://daemon:2376"),
(
Some("unix:///var/run/docker.sock"),
"docker:unix:///var/run/docker.sock",
),
] {
let config = SandboxProviderConfig::from_lookup(None, |name| {
(name == EnvVars::DOCKER_HOST)
.then(|| host.map(str::to_owned))
.flatten()
});
let factory = DockerFactory {
host: config.docker_host,
host_address: config.docker_host_address,
};
assert_eq!(factory.seed(), expected);
}
}
#[test]
fn daytona_fingerprint_keeps_unset_values_and_the_configured_target() {
let unset = DaytonaCredentials::from_api_key("test-key".to_string(), |_| None);
assert_eq!(DaytonaFactory(Some(unset)).seed(), "daytona:::");
let configured =
DaytonaCredentials::from_api_key("test-key".to_string(), |name| match name {
EnvVars::DAYTONA_API_URL => Some("https://daytona.example".to_string()),
EnvVars::DAYTONA_SERVER_URL => Some("https://ignored.example".to_string()),
EnvVars::DAYTONA_ORGANIZATION_ID => Some("org-1".to_string()),
EnvVars::DAYTONA_TARGET => Some("us".to_string()),
_ => None,
});
let factory = DaytonaFactory(Some(configured));
assert_eq!(factory.seed(), "daytona:https://daytona.example:org-1:us");
assert_eq!(factory.region(), Some("us"));
let blank =
DaytonaCredentials::new("test-key".to_string()).with_target(Some(String::new()));
assert_eq!(DaytonaFactory(Some(blank)).region(), None);
}
#[test]
fn daytona_url_alias_and_http_client_survive_the_shared_configuration() {
let credentials = DaytonaCredentials::from_api_key("test-key".to_string(), |name| {
(name == EnvVars::DAYTONA_SERVER_URL).then(|| "https://alias.example".to_string())
})
.with_http_client(Some(fabro_test::test_http_client()));
assert_eq!(
credentials.config().api_url.as_deref(),
Some("https://alias.example")
);
assert!(credentials.config().http_client.is_some());
}
#[test]
fn provider_configuration_debug_never_prints_the_key() {
let key = "dtn_test_sensitive_value";
let config = SandboxProviderConfig::from_lookup(
Some(DaytonaCredentials::from_api_key(key.to_string(), |_| None)),
|_| None,
);
let rendered = format!("{config:?}");
assert!(!rendered.contains(key));
}
}

12
lib/components/fabro-petri/src/prune.rs
View file

@ -6,8 +6,8 @@
//! sandbox prune` deletes them, over the store the run's records live in,
//! rather than through a provider call of Fabro's own: Petri opens the run
//! for writing, so a live worker that still holds the lease refuses the
//! delete; it checks each lease's fingerprint against the plugin it
//! launches, so a changed daemon or account is a problem to report, never
//! delete; it checks each lease's fingerprint against the provider it
//! connects, so a changed daemon or account is a problem to report, never
//! a delete on another backend; it writes the delete intent before the
//! provider call and the tombstone after, beside the run's other records;
//! and each provider removes its sandbox's managed workspace, a host
@ -29,9 +29,12 @@ use petri_execution::{RunKey, RunStore};
use petri_runtime::{RunOptions, Runtime};
use crate::engine;
use crate::providers::{self, SandboxProviderConfig};
/// One run whose sandboxes are to be deleted.
pub struct PruneRequest {
/// The provider configuration used by this server-side operation.
pub sandbox: SandboxProviderConfig,
/// The Fabro run id, which is Petri's run key.
pub run_id: String,
/// Where the run's worker ran Petri: its host registry and action-host
@ -69,7 +72,10 @@ pub async fn prune(request: PruneRequest) -> Result<PruneReport, PruneError> {
options.run_key = Some(RunKey::new(request.run_id.as_str()));
options.retention = engine::RETENTION;
options.sandbox.backend = backend;
let runtime = Runtime::bare().store(request.store).options(options);
let runtime = Runtime::bare()
.in_process_providers(providers::built_in_providers(&request.sandbox))
.store(request.store)
.options(options);
petri_prune::prune(&runtime)
.await
.map_err(|error| match error {

18
lib/components/fabro-petri/src/runtime.rs
View file

@ -1,6 +1,9 @@
//! The Petri runtime Fabro runs its workflows on, assembled the same way at
//! create time (for `Runtime::check`) and at execution.
//!
//! Built-in sandbox factories are installed for every runtime and connect
//! only when a scope is acquired.
//!
//! The pieces are Petri's own: [`Runtime::standard`] with the Fabro frontend
//! carrying the server's settings layer, the Attractor step kinds (the real
//! ones, or the simulated registry for a dry run), the model client as the
@ -26,10 +29,13 @@ use petri_runtime::Runtime;
use tracing::debug;
use crate::host_tools;
use crate::providers::{self, SandboxProviderConfig};
/// What every Petri runtime Fabro builds is configured with.
#[derive(Clone, Default)]
pub struct RuntimeSpec {
/// Explicit provider configuration. Factories connect only at acquire.
pub sandbox: SandboxProviderConfig,
/// The operator's settings layer, as `~/.fabro/settings.toml` text: the
/// lowest of the three layers the Fabro frontend reads (`[run.model]`
/// defaults, `[[run.hooks]]`, `[run.agent.mcps]`, `[run.environment]`
@ -64,11 +70,13 @@ impl RuntimeSpec {
/// registry: only execution swaps in the stubs.
#[must_use]
pub fn runtime(&self, for_execution: bool) -> Runtime {
let mut runtime = Runtime::standard().frontend(
Fabro::new()
.with_settings_toml(self.settings_toml.clone())
.with_mcp_catalog_toml(self.mcp_catalog_toml.clone()),
);
let mut runtime = Runtime::standard()
.in_process_providers(providers::built_in_providers(&self.sandbox))
.frontend(
Fabro::new()
.with_settings_toml(self.settings_toml.clone())
.with_mcp_catalog_toml(self.mcp_catalog_toml.clone()),
);
if let Some(client) = &self.model_client {
runtime = runtime.capability(PebbleClient(client.clone()));
}

9
lib/components/fabro-petri/tests/blobs.rs
View file

@ -1,9 +1,7 @@
//! A large stage value leaves the run's records for Fabro's blob table
//! under `blob://sha256/<hex>`, and comes back from the same table.
//!
//! The run takes its host scope through the sandbox-driver host plugin, so
//! the test skips, and says why, when the executable is not found, unless
//! `FABRO_REQUIRE_SANDBOX_PLUGINS` is set.
//! Built-in Host scopes run in process without a plugin executable.
mod support;
@ -17,7 +15,7 @@ use fabro_petri::runtime::RuntimeSpec;
use fabro_store::{BlobStore, test_support};
use fabro_types::BlobHash;
use petri_attractor_steps::blobs::{BLOB_REF_PREFIX, OFFLOAD_THRESHOLD, parse_blob_ref};
use support::{SETTINGS, Silent, admit, all_records, host_plugin, no_questions, run_request};
use support::{SETTINGS, Silent, admit, all_records, no_questions, run_request};
/// One line of the command's output.
const LINE: &str = "xxxxxxxx";
@ -38,9 +36,6 @@ fn workflow(lines: usize) -> String {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_large_output_round_trips_through_the_blob_table() {
if host_plugin().is_none() {
return;
}
let root = tempfile::tempdir().expect("a temp dir");
let pool = test_support::in_memory_pool_with(&[
fabro_db::BLOBS_MIGRATION_SQL,

107
lib/components/fabro-petri/tests/hooks.rs
View file

@ -4,14 +4,12 @@
//! failure route, the fatal checkpoint, and the run-end hooks are checked
//! against the workspace's Git history and the run's records.
//!
//! Every run acquires its scope through the sandbox-driver host plugin, so
//! the tests skip when that executable is not found, unless
//! `FABRO_REQUIRE_SANDBOX_PLUGINS` is set. The crash cases of the recovery
//! protocol need a worker to kill and live in the CLI's scenario suite.
//! Built-in Host scopes run in process without a plugin executable.
//! The crash cases need a worker to kill and live in the CLI's scenario suite.
#![expect(
clippy::disallowed_methods,
reason = "the tests locate the plugin executable through the process environment and read the workspace's history with git"
reason = "the tests inspect backend availability and read the workspace's history with git"
)]
#![expect(clippy::print_stderr, reason = "a skipped test says why on its stderr")]
@ -31,6 +29,7 @@ use fabro_petri::controls::RunControls;
use fabro_petri::engine::{self, Execution, RunRequest, RunStatus};
use fabro_petri::hooks::HooksSpec;
use fabro_petri::platform_records::PlatformRecords;
use fabro_petri::providers::{DaytonaCredentials, SandboxProviderConfig};
use fabro_petri::recovery::{self, Recovery, RecoveryRequest};
use fabro_petri::runtime::RuntimeSpec;
use fabro_petri::test_support::{MemoryBlobs, MemoryPlatformRecords};
@ -45,32 +44,7 @@ use tokio_util::sync::CancellationToken;
mod support;
const HOST_PLUGIN: &str = "sandbox-driver-host";
const HOST_PLUGIN_OVERRIDE: &str = "PETRI_SANDBOX_HOST_PLUGIN";
const DOCKER_PLUGIN: &str = "sandbox-driver-docker";
const DOCKER_PLUGIN_OVERRIDE: &str = "PETRI_SANDBOX_DOCKER_PLUGIN";
const REQUIRE_ENV: &str = "FABRO_REQUIRE_SANDBOX_PLUGINS";
/// The host plugin as Petri's lookup finds it: the override variable, else
/// the executable on `PATH`. `None`, after saying so, when the test should
/// skip; a panic when the environment forbids a skip.
fn host_plugin() -> Option<PathBuf> {
let found = env::var_os(HOST_PLUGIN_OVERRIDE)
.map(PathBuf::from)
.or_else(|| {
env::split_paths(&env::var_os("PATH")?)
.map(|dir| dir.join(HOST_PLUGIN))
.find(|candidate| candidate.is_file())
});
if found.is_none() {
assert!(
env::var_os(REQUIRE_ENV).is_none(),
"{REQUIRE_ENV} is set, but {HOST_PLUGIN} is not on PATH and {HOST_PLUGIN_OVERRIDE} is unset"
);
eprintln!("skipping: {HOST_PLUGIN} is not on PATH and {HOST_PLUGIN_OVERRIDE} is unset");
}
found
}
const REQUIRE_ENV: &str = "FABRO_REQUIRE_SANDBOX_BACKENDS";
/// A command-only bundle: the stage lines go between `start` and `exit`,
/// the edge lines after them.
@ -107,26 +81,8 @@ fn admit(workflow: &str, settings: &str) -> AdmittedGraphs {
}
}
/// The Docker plugin as Petri's lookup finds it, with a daemon that
/// answers. `None`, after saying so, when the test should skip; a panic
/// when the environment forbids a skip and the plugin is missing.
fn docker_plugin() -> Option<PathBuf> {
let found = env::var_os(DOCKER_PLUGIN_OVERRIDE)
.map(PathBuf::from)
.or_else(|| {
env::split_paths(&env::var_os("PATH")?)
.map(|dir| dir.join(DOCKER_PLUGIN))
.find(|candidate| candidate.is_file())
});
let Some(found) = found else {
assert!(
env::var_os(REQUIRE_ENV).is_none(),
"{REQUIRE_ENV} is set, but {DOCKER_PLUGIN} is not on PATH and {DOCKER_PLUGIN_OVERRIDE} \
is unset"
);
eprintln!("skipping: {DOCKER_PLUGIN} is not on PATH and {DOCKER_PLUGIN_OVERRIDE} is unset");
return None;
};
/// A reachable Docker daemon. CI requires the backend instead of skipping.
fn docker_available() -> bool {
let daemon = std::process::Command::new("docker")
.args(["version", "--format", "{{.Server.Version}}"])
.stdout(std::process::Stdio::null())
@ -134,10 +90,13 @@ fn docker_plugin() -> Option<PathBuf> {
.status()
.is_ok_and(|status| status.success());
if !daemon {
assert!(
env::var_os(REQUIRE_ENV).is_none(),
"{REQUIRE_ENV} is set, but no Docker daemon answers"
);
eprintln!("skipping: no Docker daemon answers");
return None;
}
Some(found)
daemon
}
/// One run's pieces: the store, its platform records, where it ran.
@ -194,12 +153,22 @@ impl Harness {
) -> engine::RunOutcome {
let (interviewer, observers) = no_questions();
let hooks = self.hooks(&provider);
let daytona = (provider == SandboxProviderKind::DAYTONA).then(|| {
DaytonaCredentials::from_api_key(
env::var("DAYTONA_API_KEY").expect("live Daytona credentials"),
|name| env::var(name).ok(),
)
});
let sandbox = SandboxProviderConfig::from_lookup(daytona, |name| env::var(name).ok());
let request = RunRequest {
run_id: self.run_id.to_string(),
run_dir: self.run_dir.clone(),
execution: Execution::Start(admit(workflow, settings)),
store: Arc::clone(&self.store) as Arc<dyn petri_store::RunStore>,
runtime: RuntimeSpec::default(),
runtime: RuntimeSpec {
sandbox,
..RuntimeSpec::default()
},
provider,
cancel: CancellationToken::new(),
controls: RunControls::new(),
@ -359,9 +328,6 @@ fn stages(inspection: &RunInspection) -> Vec<(String, String)> {
/// reached Petri's local service through Fabro's wrapper.
#[tokio::test]
async fn every_finish_is_committed_and_recorded() {
if host_plugin().is_none() {
return;
}
let harness = Harness::new();
let workflow = workflow(
" write [shape=parallelogram, script=\"echo one > out.txt\"]\n check \
@ -436,9 +402,6 @@ async fn every_finish_is_committed_and_recorded() {
/// end.
#[tokio::test]
async fn artifacts_the_branch_and_the_diffs_are_recorded() {
if host_plugin().is_none() {
return;
}
let mut harness = Harness::new();
harness.artifacts = vec!["assets/**".to_string()];
let workflow = workflow(
@ -607,9 +570,6 @@ async fn checkpoint_nodes(harness: &Harness) -> Vec<(String, u64)> {
/// and its failure route runs on the committed files.
#[tokio::test]
async fn a_failed_stage_is_committed_and_its_route_sees_the_files() {
if host_plugin().is_none() {
return;
}
let harness = Harness::new();
let workflow = workflow(
" work [shape=parallelogram, script=\"echo partial > out.txt; exit 1\"]\n fix \
@ -650,9 +610,6 @@ async fn a_failed_stage_is_committed_and_its_route_sees_the_files() {
/// checkpoint's error, and a restart reports it failed without resuming.
#[tokio::test]
async fn a_failed_checkpoint_ends_the_run_with_no_route() {
if host_plugin().is_none() {
return;
}
let harness = Harness::new();
let workflow = workflow(
" wreck [shape=parallelogram, script=\"rm -rf .git && echo garbage > .git && echo wrecked \
@ -716,9 +673,6 @@ async fn a_failed_checkpoint_ends_the_run_with_no_route() {
/// starts over, and a run that finished has nothing to bring back.
#[tokio::test]
async fn recovery_starts_an_unknown_run_and_resumes_a_finished_one() {
if host_plugin().is_none() {
return;
}
let harness = Harness::new();
assert_eq!(harness.recover().await, Recovery::Start);
@ -756,9 +710,6 @@ async fn a_run_hook_blocks_a_tool_effect_through_the_forwarded_service() {
use serde_json::json;
const MODEL: &str = "gpt-5.6-sol";
if host_plugin().is_none() {
return;
}
let twin = fabro_test::twin_openai().await;
let namespace = format!("{}::{}", module_path!(), line!());
TwinScenarios::new(namespace.clone())
@ -865,9 +816,6 @@ async fn the_records_name_the_root_invocations_workspace() {
use fabro_petri::workspace::WorkspaceLookup;
use petri_execution::InvocationId;
if host_plugin().is_none() {
return;
}
let harness = Harness::new();
let workflow = workflow(
" write [shape=parallelogram, script=\"echo one > out.txt\"]",
@ -899,9 +847,6 @@ async fn the_records_name_the_root_invocations_workspace() {
/// problem.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn parallel_branches_checkpoint_the_shared_workspace_in_turn() {
if host_plugin().is_none() {
return;
}
let harness = Harness::new();
let workflow = workflow(
" fork [shape=component]\n a [shape=parallelogram, script=\"echo a > a.txt\"]\n b \
@ -955,7 +900,7 @@ async fn parallel_branches_checkpoint_the_shared_workspace_in_turn() {
/// its ref, with the platform records naming the same commits.
#[tokio::test]
async fn a_docker_run_commits_inside_the_container_and_publishes_every_checkpoint() {
if docker_plugin().is_none() {
if !docker_available() {
return;
}
assert_sandbox_run_publishes_every_checkpoint(SandboxProviderKind::DOCKER).await;
@ -963,8 +908,8 @@ async fn a_docker_run_commits_inside_the_container_and_publishes_every_checkpoin
/// The same protocol on Daytona: the sandbox-driver facets are provider
/// neutral, so the commit, the bundle and the restore take one path. Live:
/// it needs `DAYTONA_API_KEY` and the Daytona plugin, and provisions a
/// sandbox.
/// it needs `DAYTONA_API_KEY`, passed explicitly to the provider, and
/// provisions a sandbox.
#[tokio::test]
#[ignore = "requires live Daytona credentials and provisions a sandbox"]
async fn a_daytona_run_commits_inside_the_sandbox_and_publishes_every_checkpoint() {

61
lib/components/fabro-petri/tests/host_tools.rs
View file

@ -1,23 +1,14 @@
//! Fabro's run tools on a Petri run from this crate (integration plan item
//! F3.4): `RuntimeSpec::run_tools` installs the adapter as Petri's host
//! tool capability, a workflow with one agent stage runs on the real step
//! registry against a scripted model, and the stage's session gets the
//! tools the legacy worker registers, bound to the run: the model is
//! advertised every run tool, its `fabro_run_create` call reaches Fabro's
//! API with the Petri run as the child's parent, the API's answer comes
//! back to the model, and the call is in the run's record under the stage.
//! Fabro's run tools on a Petri run from this crate: `RuntimeSpec::run_tools`
//! installs the adapter as Petri's host tool capability, a workflow with one
//! agent stage runs on the real step registry against a scripted model, and the
//! stage's session gets the tools the legacy worker registers, bound to the
//! run: the model is advertised every run tool, its `fabro_run_create` call
//! reaches Fabro's API with the Petri run as the child's parent, the API's
//! answer comes back to the model, and the call is in the run's record under
//! the stage.
//!
//! Every run takes its scope through the sandbox-driver host plugin, so
//! the tests skip when that executable is not found, unless
//! `FABRO_REQUIRE_SANDBOX_PLUGINS` is set.
//! Built-in Host scopes run in process without a plugin executable.
#![expect(
clippy::disallowed_methods,
reason = "the tests locate the plugin executable through the process environment"
)]
#![expect(clippy::print_stderr, reason = "a skipped test says why on its stderr")]
use std::env;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;
@ -48,10 +39,6 @@ use petri_store::{Access, MemoryRunStore, RunKey, RunStore};
use serde_json::json;
use tokio::fs;
const HOST_PLUGIN: &str = "sandbox-driver-host";
const HOST_PLUGIN_OVERRIDE: &str = "PETRI_SANDBOX_HOST_PLUGIN";
const REQUIRE_ENV: &str = "FABRO_REQUIRE_SANDBOX_PLUGINS";
/// One agent stage on the native backend, pinned to the scripted model.
const AGENT_WORKFLOW: &str = r#"digraph Agent {
graph [goal="Start a child run", backend="api", default_max_retries=0]
@ -63,27 +50,6 @@ const AGENT_WORKFLOW: &str = r#"digraph Agent {
const AGENT_SETTINGS: &str = "_version = 1\n\n[workflow]\ngraph = \"workflow.fabro\"\n";
/// The host plugin as Petri's lookup finds it: the override variable, else
/// the executable on `PATH`. `None`, after saying so, when the test should
/// skip; a panic when the environment forbids a skip.
fn host_plugin() -> Option<PathBuf> {
let found = env::var_os(HOST_PLUGIN_OVERRIDE)
.map(PathBuf::from)
.or_else(|| {
env::split_paths(&env::var_os("PATH")?)
.map(|dir| dir.join(HOST_PLUGIN))
.find(|candidate| candidate.is_file())
});
if found.is_none() {
assert!(
env::var_os(REQUIRE_ENV).is_none(),
"{REQUIRE_ENV} is set, but {HOST_PLUGIN} is not on PATH and {HOST_PLUGIN_OVERRIDE} is unset"
);
eprintln!("skipping: {HOST_PLUGIN} is not on PATH and {HOST_PLUGIN_OVERRIDE} is unset");
}
found
}
/// Write the agent bundle into `<root>/.fabro/workflows/agent`; the
/// workflow file.
async fn install_bundle(root: &Path) -> PathBuf {
@ -187,9 +153,6 @@ fn advertised(request: &Request) -> Vec<(String, String)> {
/// answer reaches the model; the call is in the record under the stage.
#[tokio::test]
async fn a_petri_stage_calls_a_run_tool_bound_to_the_run() {
if host_plugin().is_none() {
return;
}
let root = tempfile::tempdir().expect("a temp dir");
let workflow = install_bundle(root.path()).await;
let run_id = RunId::new();
@ -271,9 +234,6 @@ async fn a_petri_stage_calls_a_run_tool_bound_to_the_run() {
/// one tool the model called marked invoked.
#[tokio::test]
async fn the_projection_lists_the_stages_tools_and_marks_the_one_called() {
if host_plugin().is_none() {
return;
}
let root = tempfile::tempdir().expect("a temp dir");
let workflow = install_bundle(root.path()).await;
let run_id = RunId::new();
@ -375,9 +335,6 @@ async fn the_projection_lists_the_stages_tools_and_marks_the_one_called() {
/// rather than parenting a child run to the wrong run.
#[tokio::test]
async fn services_for_another_run_give_the_stage_no_run_tools() {
if host_plugin().is_none() {
return;
}
let root = tempfile::tempdir().expect("a temp dir");
let workflow = install_bundle(root.path()).await;
let run_id = RunId::new();

21
lib/components/fabro-petri/tests/interview.rs
View file

@ -6,9 +6,7 @@
//! timeout with the gate's default, an auto-approved run answers itself,
//! and a cancelled run interrupts its question.
//!
//! Every run takes its host scope through the sandbox-driver host plugin,
//! so the tests skip, and say why, when the executable is not found,
//! unless `FABRO_REQUIRE_SANDBOX_PLUGINS` is set.
//! Built-in Host scopes run in process without a plugin executable.
mod support;
@ -25,7 +23,7 @@ use fabro_petri::runtime::RuntimeSpec;
use fabro_types::{Principal, QuestionType, SystemActorKind};
use petri_execution::{Delivery, InterviewReceipt, RECEIPT_FILE, ReplyRecord};
use petri_store::MemoryRunStore;
use support::{SETTINGS, admit, all_records, host_plugin, run_request, wait_until};
use support::{SETTINGS, admit, all_records, run_request, wait_until};
use tokio::fs;
/// A board of every notice the adapter posted.
@ -175,9 +173,6 @@ impl Gate {
/// submitted under the posted id, as the API delivers it, routes the gate.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_gate_answered_under_the_posted_id_routes_on_the_answer() {
if host_plugin().is_none() {
return;
}
let gate = Gate::new();
let workflow = one_gate(&gate.markers, "");
let runtime = RuntimeSpec::default();
@ -263,9 +258,6 @@ async fn a_gate_answered_under_the_posted_id_routes_on_the_answer() {
/// the other order, lands on its own branch.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn two_parallel_gates_each_bind_their_own_answer() {
if host_plugin().is_none() {
return;
}
let gate = Gate::new();
let workflow = two_gates(&gate.markers);
let runtime = RuntimeSpec::default();
@ -334,9 +326,6 @@ async fn two_parallel_gates_each_bind_their_own_answer() {
/// the default's branch runs.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn an_unanswered_question_expires_with_the_gates_default() {
if host_plugin().is_none() {
return;
}
let gate = Gate::new();
let workflow = one_gate(
&gate.markers,
@ -384,9 +373,6 @@ async fn an_unanswered_question_expires_with_the_gates_default() {
/// engine, and still posts the question and its answer.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn an_auto_approved_run_answers_yes_at_once() {
if host_plugin().is_none() {
return;
}
let gate = Gate::new();
let workflow = one_gate(&gate.markers, "");
let runtime = RuntimeSpec::default();
@ -427,9 +413,6 @@ async fn an_auto_approved_run_answers_yes_at_once() {
/// question is interrupted, the gate fails closed and the run is cancelled.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_cancelled_run_interrupts_its_pending_question() {
if host_plugin().is_none() {
return;
}
let gate = Gate::new();
let workflow = one_gate(&gate.markers, "");
let runtime = RuntimeSpec::default();

11
lib/components/fabro-petri/tests/model.rs
View file

@ -4,10 +4,8 @@
//! The `hello` bundle's agent stage calls the OpenAI twin through a model
//! client built over a vault that holds the key; the twin requires a
//! bearer token and logs requests under it, so a request logged under the
//! vault's key proves the key came from the vault. The run takes its host
//! scope through the sandbox-driver host plugin, so the test skips, and
//! says why, when the executable is not found, unless
//! `FABRO_REQUIRE_SANDBOX_PLUGINS` is set.
//! vault's key proves the key came from the vault. The Host scope runs in
//! process.
mod support;
@ -24,7 +22,7 @@ use fabro_types::SecretType;
use fabro_vault::Vault;
use lithos_llm::catalog::ProviderId;
use petri_store::MemoryRunStore;
use support::{Silent, all_records, hello_bundle, host_plugin, no_questions, run_request};
use support::{Silent, all_records, hello_bundle, no_questions, run_request};
use tokio::fs;
use tokio::sync::RwLock as AsyncRwLock;
@ -32,9 +30,6 @@ const OPENAI_MODEL: &str = "gpt-5.4";
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_model_call_authenticates_through_the_vault_and_skills_read_the_home() {
if host_plugin().is_none() {
return;
}
let twin = twin_openai().await;
let namespace = format!("{}::{}", module_path!(), line!());
TwinScenarios::new(&namespace)

87
lib/components/fabro-petri/tests/projection.rs
View file

@ -7,20 +7,11 @@
//! a live run costs its new records, with the cache that makes it so
//! dropped at a restart, after the idle period and at the run's finish.
//!
//! Every run here takes its scope's environment through the sandbox-driver
//! host plugin, so the tests skip, and say why, when the executable is not
//! found, unless `FABRO_REQUIRE_SANDBOX_PLUGINS` is set.
#![expect(
clippy::disallowed_methods,
reason = "the tests locate the plugin executable through the process environment"
)]
#![expect(clippy::print_stderr, reason = "a skipped test says why on its stderr")]
//! Built-in Host scopes run in process without a plugin executable.
mod support;
use std::collections::BTreeSet;
use std::env;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::{Duration, Instant};
@ -32,8 +23,9 @@ use fabro_petri::check::Launch;
use fabro_petri::engine::{self, RunStatus as EngineRunStatus};
use fabro_petri::interview::{Approval, FabroInterviewer};
use fabro_petri::projector::{self, Projector};
use fabro_petri::providers::SandboxProviderConfig;
use fabro_petri::runtime::RuntimeSpec;
use fabro_petri::{SqliteRunStore, test_support as petri_support};
use fabro_petri::{SqliteRunStore, providers, test_support as petri_support};
use fabro_store::platform_records::{
PlatformRecord, PlatformRecordStore, RunCreatedRecord, RunLifecycleKind, RunLifecycleRecord,
};
@ -52,10 +44,6 @@ use petri_store::{RunKey, RunStore};
use tokio::fs;
use tokio::time::sleep;
const HOST_PLUGIN: &str = "sandbox-driver-host";
const HOST_PLUGIN_OVERRIDE: &str = "PETRI_SANDBOX_HOST_PLUGIN";
const REQUIRE_ENV: &str = "FABRO_REQUIRE_SANDBOX_PLUGINS";
const COMMAND_WORKFLOW: &str = r#"digraph Command {
graph [goal="Run one command"]
start [shape=Mdiamond]
@ -127,24 +115,6 @@ fn described_gate_workflow(markers: &Path) -> String {
)
}
fn host_plugin() -> Option<PathBuf> {
let found = env::var_os(HOST_PLUGIN_OVERRIDE)
.map(PathBuf::from)
.or_else(|| {
env::split_paths(&env::var_os("PATH")?)
.map(|dir| dir.join(HOST_PLUGIN))
.find(|candidate| candidate.is_file())
});
if found.is_none() {
assert!(
env::var_os(REQUIRE_ENV).is_none(),
"{REQUIRE_ENV} is set, but {HOST_PLUGIN} is not on PATH and {HOST_PLUGIN_OVERRIDE} is unset"
);
eprintln!("skipping: {HOST_PLUGIN} is not on PATH and {HOST_PLUGIN_OVERRIDE} is unset");
}
found
}
/// A fresh in-memory database with every table the projection touches.
fn pool() -> DbPool {
test_support::in_memory_pool_with(&[
@ -236,7 +206,11 @@ async fn run_workflow(
workflow: &Path,
stubs: bool,
) {
let runtime = Runtime::standard().frontend(Fabro::new());
let runtime = Runtime::standard()
.in_process_providers(providers::built_in_providers(
&SandboxProviderConfig::default(),
))
.frontend(Fabro::new());
let runtime = if stubs {
petri_attractor_steps::register_stubs(runtime)
} else {
@ -466,9 +440,6 @@ async fn stage_states(pool: &DbPool, run_id: RunId) -> Vec<(String, StageState)>
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_hello_bundle_projects_live_as_it_rebuilds() {
if host_plugin().is_none() {
return;
}
let scenario = hello_scenario().await;
run_live(&scenario).await;
assert_view_equals_rebuild(&scenario.pool, scenario.run_id).await;
@ -495,9 +466,6 @@ async fn the_hello_bundle_projects_live_as_it_rebuilds() {
/// reference, never as the bytes the live log accumulated.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_large_output_projects_as_its_blob_reference() {
if host_plugin().is_none() {
return;
}
let scenario = large_output_scenario().await;
run_live(&scenario).await;
assert_view_equals_rebuild(&scenario.pool, scenario.run_id).await;
@ -519,9 +487,6 @@ async fn a_large_output_projects_as_its_blob_reference() {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_command_workflow_projects_live_as_it_rebuilds() {
if host_plugin().is_none() {
return;
}
let scenario = command_scenario().await;
run_live(&scenario).await;
assert_view_equals_rebuild(&scenario.pool, scenario.run_id).await;
@ -548,9 +513,6 @@ async fn a_command_workflow_projects_live_as_it_rebuilds() {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_parallel_workflow_projects_its_branches_as_child_executions() {
if host_plugin().is_none() {
return;
}
let scenario = parallel_scenario().await;
run_live(&scenario).await;
assert_view_equals_rebuild(&scenario.pool, scenario.run_id).await;
@ -587,9 +549,6 @@ async fn a_parallel_workflow_projects_its_branches_as_child_executions() {
/// folds everything.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn dropped_wake_ups_are_caught_up_by_the_next_signal() {
if host_plugin().is_none() {
return;
}
let scenario = command_scenario().await;
run_unobserved(&scenario).await;
assert!(
@ -614,9 +573,6 @@ async fn dropped_wake_ups_are_caught_up_by_the_next_signal() {
/// The same, through the startup pass.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_startup_pass_catches_up_a_view_nobody_signalled() {
if host_plugin().is_none() {
return;
}
let scenario = command_scenario().await;
run_unobserved(&scenario).await;
let projector = Projector::new(scenario.pool.clone(), scenario.pool.clone());
@ -638,9 +594,6 @@ async fn the_startup_pass_catches_up_a_view_nobody_signalled() {
/// a duplicate.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn concurrent_passes_over_one_run_commit_one_contiguous_stream() {
if host_plugin().is_none() {
return;
}
let scenario = parallel_scenario().await;
run_unobserved(&scenario).await;
let projector = Projector::new(scenario.pool.clone(), scenario.pool.clone());
@ -745,9 +698,6 @@ async fn copy_run_without_records(source: &DbPool, run_id: RunId) -> DbPool {
/// sequence continuing from where the committed view stood.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_crash_between_the_record_commit_and_the_view_applies_only_the_suffix() {
if host_plugin().is_none() {
return;
}
let scenario = parallel_scenario().await;
run_unobserved(&scenario).await;
let rows = petri_rows(&scenario.pool, scenario.run_id).await;
@ -863,9 +813,6 @@ async fn a_crash_between_the_record_commit_and_the_view_applies_only_the_suffix(
/// agree with a projector that saw the run whole.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_restarted_projector_agrees_over_nested_child_executions() {
if host_plugin().is_none() {
return;
}
let scenario = parallel_scenario().await;
run_unobserved(&scenario).await;
let rows = petri_rows(&scenario.pool, scenario.run_id).await;
@ -928,9 +875,6 @@ async fn a_restarted_projector_agrees_over_nested_child_executions() {
/// reported incomplete with the reason.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_torn_tail_holds_the_view_and_reports_the_run_incomplete() {
if host_plugin().is_none() {
return;
}
let scenario = command_scenario().await;
run_unobserved(&scenario).await;
let projector = Projector::new(scenario.pool.clone(), scenario.pool.clone());
@ -1035,9 +979,6 @@ async fn committed_pass(projector: &Projector, run_id: RunId) -> projector::Pass
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_pass_over_a_live_run_costs_its_new_records_not_the_run() {
const BATCH: usize = 7;
if host_plugin().is_none() {
return;
}
let scenario = parallel_scenario().await;
run_unobserved(&scenario).await;
let rows = petri_rows(&scenario.pool, scenario.run_id).await;
@ -1088,9 +1029,6 @@ async fn a_pass_over_a_live_run_costs_its_new_records_not_the_run() {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_restart_and_the_idle_period_drop_the_cache_and_one_full_replay_rebuilds_it() {
const BATCH: usize = 5;
if host_plugin().is_none() {
return;
}
let scenario = parallel_scenario().await;
run_unobserved(&scenario).await;
let rows = petri_rows(&scenario.pool, scenario.run_id).await;
@ -1272,9 +1210,6 @@ impl GateRun {
/// rebuilds the same.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn an_expired_question_is_pending_while_the_gate_waits_and_closes_on_the_expiry() {
if host_plugin().is_none() {
return;
}
let gate = Arc::new(gate_run(r#", timeout="1500ms", human.default_choice="no""#).await);
let running = {
let gate = Arc::clone(&gate);
@ -1354,9 +1289,6 @@ async fn an_expired_question_is_pending_while_the_gate_waits_and_closes_on_the_e
/// on a choice that has none.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_pending_question_carries_its_choice_descriptions_previews_and_context() {
if host_plugin().is_none() {
return;
}
let gate = Arc::new(gate_run_of(described_gate_workflow).await);
let running = {
let gate = Arc::clone(&gate);
@ -1401,9 +1333,6 @@ async fn a_pending_question_carries_its_choice_descriptions_previews_and_context
/// the view rebuilds the same.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn an_auto_approved_answer_closes_the_question_in_the_projection() {
if host_plugin().is_none() {
return;
}
let gate = gate_run("").await;
gate.run(Approval::Auto).await;

40
lib/components/fabro-petri/tests/prune.rs
View file

@ -3,8 +3,7 @@
//! tombstoned in the run's record, a second prune has nothing to do, and a
//! run a live handle holds is refused.
//!
//! The run takes its scope through the sandbox-driver host plugin, so the
//! test skips, and says why, when the executable is not found.
//! The Host scope runs in process without a plugin executable.
#![expect(
clippy::disallowed_methods,
@ -18,12 +17,13 @@ use std::sync::Arc;
use fabro_petri::check::Launch;
use fabro_petri::engine::{self, RunStatus};
use fabro_petri::providers::SandboxProviderConfig;
use fabro_petri::prune::{PruneError, PruneRequest, prune};
use fabro_petri::runtime::RuntimeSpec;
use fabro_petri::{SqliteRunStore, petri};
use fabro_store::test_support;
use fabro_types::SandboxProviderKind;
use support::{Silent, admit, host_plugin, no_questions, run_request};
use support::{Silent, admit, no_questions, run_request};
/// A command-only workflow whose one stage writes a file into its
/// workspace.
@ -96,9 +96,6 @@ async fn lease_states(store: &SqliteRunStore, run_id: &str) -> Vec<String> {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_finished_runs_host_workspace_is_deleted_once_and_a_held_run_is_refused() {
if host_plugin().is_none() {
return;
}
let root = tempfile::tempdir().expect("a temp dir");
let run_dir = root.path().join("run");
let pool = test_support::in_memory_pool_with(&[
@ -137,7 +134,37 @@ async fn a_finished_runs_host_workspace_is_deleted_once_and_a_held_run_is_refuse
);
assert_eq!(lease_states(&store, "prune").await, ["stopped"]);
let logs = petri::RunStore::open(
store.as_ref(),
&petri::RunKey::new("prune"),
petri::Access::Read,
)
.await
.expect("the resources open");
let resources = logs
.read(&petri::LogId::Resources)
.await
.expect("the resources read");
let legacy_fingerprint = format!(
"host:{}",
run_dir
.join("host-registry")
.canonicalize()
.expect("canonical registry")
.display()
);
let allocating = resources
.iter()
.find(|record| {
record.record["body"]["state"] == "allocating"
&& record.record["body"]["fingerprint"].is_string()
})
.expect("the allocation is recorded");
assert_eq!(allocating.record["body"]["fingerprint"], legacy_fingerprint);
drop(logs);
let request = || PruneRequest {
sandbox: SandboxProviderConfig::default(),
run_id: "prune".to_string(),
run_dir: run_dir.clone(),
store: store.clone(),
@ -187,6 +214,7 @@ async fn a_finished_runs_host_workspace_is_deleted_once_and_a_held_run_is_refuse
async fn a_provider_petri_does_not_serve_is_refused_before_the_store_is_opened() {
let store = Arc::new(petri_store::MemoryRunStore::new());
let error = prune(PruneRequest {
sandbox: SandboxProviderConfig::default(),
run_id: "e2b-run".to_string(),
run_dir: std::env::temp_dir().join("fabro-petri-prune-e2b"),
store,

70
lib/components/fabro-petri/tests/runs.rs
View file

@ -2,22 +2,13 @@
//! memory on the stub registry, and a command-only workflow on the host
//! sandbox through the real step registry.
//!
//! Every run, stubbed or real, acquires its scope's environment through the
//! sandbox-driver host plugin, so both tests skip when that executable is not
//! found, unless `FABRO_REQUIRE_SANDBOX_PLUGINS` is set. Fabro's CI installs
//! the plugin on `PATH` in the sandbox-plugins job and requires it there.
//! Built-in Host scopes run in process without a plugin executable.
#![expect(
clippy::disallowed_methods,
reason = "the tests locate the plugin executable through the process environment"
)]
#![expect(clippy::print_stderr, reason = "a skipped test says why on its stderr")]
use std::env;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;
use fabro_petri::providers::{self, SandboxProviderConfig};
use petri_execution::host::{self, HostRun};
use petri_execution::inspect::{self, RunInspection};
use petri_frontend_fabro::Fabro;
@ -28,10 +19,6 @@ use petri_runtime::{RunOptions, Runtime};
use petri_store::{Access, MemoryRunStore, RunKey, RunStore as _};
use tokio::fs;
const HOST_PLUGIN: &str = "sandbox-driver-host";
const HOST_PLUGIN_OVERRIDE: &str = "PETRI_SANDBOX_HOST_PLUGIN";
const REQUIRE_ENV: &str = "FABRO_REQUIRE_SANDBOX_PLUGINS";
/// A command-only workflow: one script stage between start and exit.
const COMMAND_WORKFLOW: &str = r#"digraph Command {
graph [goal="Run one command"]
@ -48,27 +35,6 @@ fn hello_bundle() -> PathBuf {
Path::new(env!("CARGO_MANIFEST_DIR")).join("../../../.fabro/workflows/hello")
}
/// The host plugin as Petri's lookup finds it: the override variable, else
/// the executable on `PATH`. `None`, after saying so, when the test should
/// skip; a panic when the environment forbids a skip.
fn host_plugin() -> Option<PathBuf> {
let found = env::var_os(HOST_PLUGIN_OVERRIDE)
.map(PathBuf::from)
.or_else(|| {
env::split_paths(&env::var_os("PATH")?)
.map(|dir| dir.join(HOST_PLUGIN))
.find(|candidate| candidate.is_file())
});
if found.is_none() {
assert!(
env::var_os(REQUIRE_ENV).is_none(),
"{REQUIRE_ENV} is set, but {HOST_PLUGIN} is not on PATH and {HOST_PLUGIN_OVERRIDE} is unset"
);
eprintln!("skipping: {HOST_PLUGIN} is not on PATH and {HOST_PLUGIN_OVERRIDE} is unset");
}
found
}
/// Write a bundle's files into `<root>/.fabro/workflows/<name>` so the
/// frontend sees a bundle root of its own, with no project settings layer
/// above it. Returns the workflow file.
@ -131,12 +97,9 @@ async fn run_workflow(
/// The `hello` bundle, whose one stage is a prompt, completes on the stub
/// registry with no model, and its record in the memory store says so. The
/// stubbed stages never run a command, but the run still takes its host
/// scope through the plugin.
/// scope through the in-process provider.
#[tokio::test]
async fn the_hello_bundle_runs_in_memory_on_the_stub_registry() {
if host_plugin().is_none() {
return;
}
let root = tempfile::tempdir().expect("a temp dir");
let bundle = hello_bundle();
let workflow_text = fs::read_to_string(bundle.join("workflow.fabro"))
@ -151,9 +114,15 @@ async fn the_hello_bundle_runs_in_memory_on_the_stub_registry() {
])
.await;
let store = Arc::new(MemoryRunStore::new());
let rt = petri_attractor_steps::register_stubs(Runtime::standard().frontend(Fabro::new()))
.store(store.clone())
.options(run_options(&root.path().join("run"), "hello"));
let rt = petri_attractor_steps::register_stubs(
Runtime::standard()
.in_process_providers(providers::built_in_providers(
&SandboxProviderConfig::default(),
))
.frontend(Fabro::new()),
)
.store(store.clone())
.options(run_options(&root.path().join("run"), "hello"));
let inspection = run_workflow(&rt, &store, "hello", &workflow).await;
@ -167,9 +136,6 @@ async fn the_hello_bundle_runs_in_memory_on_the_stub_registry() {
/// real step registry, and its record in the memory store says so.
#[tokio::test]
async fn a_command_workflow_runs_on_the_host_sandbox() {
if host_plugin().is_none() {
return;
}
let root = tempfile::tempdir().expect("a temp dir");
let workflow = install_bundle(root.path(), "command", &[
("workflow.fabro", COMMAND_WORKFLOW),
@ -177,9 +143,15 @@ async fn a_command_workflow_runs_on_the_host_sandbox() {
])
.await;
let store = Arc::new(MemoryRunStore::new());
let rt = petri_attractor_steps::register(Runtime::standard().frontend(Fabro::new()))
.store(store.clone())
.options(run_options(&root.path().join("run"), "command"));
let rt = petri_attractor_steps::register(
Runtime::standard()
.in_process_providers(providers::built_in_providers(
&SandboxProviderConfig::default(),
))
.frontend(Fabro::new()),
)
.store(store.clone())
.options(run_options(&root.path().join("run"), "command"));
let inspection = run_workflow(&rt, &store, "command", &workflow).await;

12
lib/components/fabro-petri/tests/secrets.rs
View file

@ -2,9 +2,7 @@
//! command's environment, and the value never reaches `petri_records`:
//! Petri masks every record before it is appended.
//!
//! The run takes its host scope through the sandbox-driver host plugin, so
//! the test skips, and says why, when the executable is not found, unless
//! `FABRO_REQUIRE_SANDBOX_PLUGINS` is set.
//! Built-in Host scopes run in process without a plugin executable.
mod support;
@ -19,7 +17,7 @@ use fabro_petri::secrets::VaultSecrets;
use fabro_store::test_support;
use fabro_types::SecretType;
use fabro_vault::Vault;
use support::{Silent, admit, host_plugin, no_questions, run_request};
use support::{Silent, admit, no_questions, run_request};
const TOKEN: &str = "hunter2-hunter2-hunter2";
@ -50,9 +48,6 @@ TOKEN = "{{ secrets.TOKEN }}"
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_secret_reaches_the_command_and_is_masked_in_every_record() {
if host_plugin().is_none() {
return;
}
let root = tempfile::tempdir().expect("a temp dir");
let pool = test_support::in_memory_pool_with(&[
fabro_db::BLOBS_MIGRATION_SQL,
@ -106,9 +101,6 @@ async fn a_secret_reaches_the_command_and_is_masked_in_every_record() {
/// ends the way Fabro's failure policy for a command ends it.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_secret_nobody_provides_fails_the_command() {
if host_plugin().is_none() {
return;
}
let root = tempfile::tempdir().expect("a temp dir");
let store = Arc::new(petri_store::MemoryRunStore::new());
let runtime = RuntimeSpec::default();

33
lib/components/fabro-petri/tests/support/mod.rs
View file

@ -1,4 +1,4 @@
//! What the adapter tests share: the host plugin lookup, a bundle admitted
//! What the adapter tests share: a bundle admitted
//! through `check`, a run request over the engine assembly, and the run's
//! records read back from its store.
@ -8,7 +8,6 @@
)]
use std::collections::BTreeMap;
use std::env;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::{Duration, Instant};
@ -25,39 +24,9 @@ use petri_store::{Access, LogId, RunKey, RunStore};
use tokio::time::sleep;
use tokio_util::sync::CancellationToken;
const HOST_PLUGIN: &str = "sandbox-driver-host";
const HOST_PLUGIN_OVERRIDE: &str = "PETRI_SANDBOX_HOST_PLUGIN";
const REQUIRE_ENV: &str = "FABRO_REQUIRE_SANDBOX_PLUGINS";
pub(crate) const POLL: Duration = Duration::from_millis(10);
pub(crate) const PATIENCE: Duration = Duration::from_secs(30);
/// The host plugin as Petri's lookup finds it: the override variable, else
/// the executable on `PATH`. `None`, after saying so, when the test should
/// skip; a panic when the environment forbids a skip.
#[expect(
clippy::disallowed_methods,
reason = "the tests locate the plugin executable through the process environment"
)]
#[expect(clippy::print_stderr, reason = "a skipped test says why on its stderr")]
pub(crate) fn host_plugin() -> Option<PathBuf> {
let found = env::var_os(HOST_PLUGIN_OVERRIDE)
.map(PathBuf::from)
.or_else(|| {
env::split_paths(&env::var_os("PATH")?)
.map(|dir| dir.join(HOST_PLUGIN))
.find(|candidate| candidate.is_file())
});
if found.is_none() {
assert!(
env::var_os(REQUIRE_ENV).is_none(),
"{REQUIRE_ENV} is set, but {HOST_PLUGIN} is not on PATH and {HOST_PLUGIN_OVERRIDE} is unset"
);
eprintln!("skipping: {HOST_PLUGIN} is not on PATH and {HOST_PLUGIN_OVERRIDE} is unset");
}
found
}
/// The `.fabro/workflows/hello` bundle checked into this repository.
pub(crate) fn hello_bundle() -> PathBuf {
Path::new(env!("CARGO_MANIFEST_DIR")).join("../../../.fabro/workflows/hello")

2
lib/foundation/fabro-static/src/env_vars.rs
View file

@ -168,6 +168,7 @@ impl EnvVars {
pub const DAYTONA_API_KEY: &'static str = "DAYTONA_API_KEY";
pub const DAYTONA_API_URL: &'static str = "DAYTONA_API_URL";
pub const DAYTONA_ORGANIZATION_ID: &'static str = "DAYTONA_ORGANIZATION_ID";
pub const DAYTONA_TARGET: &'static str = "DAYTONA_TARGET";
pub const DAYTONA_SERVER_URL: &'static str = "DAYTONA_SERVER_URL";
pub const SESSION_SECRET: &'static str = "SESSION_SECRET";
@ -330,6 +331,7 @@ mod tests {
EnvVars::DAYTONA_API_URL,
EnvVars::DAYTONA_ORGANIZATION_ID,
EnvVars::DAYTONA_SERVER_URL,
EnvVars::DAYTONA_TARGET,
EnvVars::SESSION_SECRET,
EnvVars::CARGO_BIN_EXE_FABRO,
EnvVars::CARGO_CFG_TARGET_OS,