scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-15 23:32:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Consume the sandbox driver's events directly in the workflow

Browse source 
Fabro-sandbox carried its own SandboxEvent enum and a callback for it.
The run sandbox wrapped every lifecycle call to emit a start, completed,
or failed variant with its own clock, and re-described the driver's
create-time progress as snapshot events through an observer that lived
next to the run sandbox. The workflow then converted that enum to the
wire. The driver already reports every operation it performs, so the
enum was a second, hand-maintained copy of that stream.

The workflow now observes the driver's events directly. A run's sandbox
is created or attached with a driver EventContext whose observer is the
new SandboxEventBridge in the workflow's event module. The bridge turns
the driver's start, stop, and delete operations, its image pull inside a
create, and its snapshot builds into the workflow's SandboxLifecycle
events, stamping fabro's provider name so the run keeps recording
`local` rather than the driver's `host`. The pipeline emits the
initializing, ready, and failed events itself around bringing the sandbox
up, since that composite step — create, activate, prepare the workspace
— is the pipeline's, not the driver's. Fabro-sandbox emits no events of
its own any more; the run sandbox gained console_url for the ready
event, and a local sandbox can be created with an event context.

The wire keeps every name the CLI reads. Two families go: the cleanup
events, which only the server's manifest validation could have produced
and it passed no callback, and the git clone events, which nothing read
and whose facts the sandbox.initialized event and tracing already carry.
The ready event drops the cpu and memory fields no provider ever
populated. Daytona snapshot events now come from the driver's ensure
call, so a snapshot that already exists and is active reports nothing
rather than a creating-and-ready pair that did no work.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-09-10 00:37:42 -06:00
parent 6e874aff5b
commit 3f721dd032
No known key found for this signature in database
24 changed files with 707 additions and 964 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
docs/internal/events-strategy.md
View file

@ -125,7 +125,12 @@ Never build the same `RunEvent` twice if multiple sinks receive it.
### 1. Add the typed event
Add a variant to `Event`, `AgentEvent`, or `SandboxEvent` as appropriate.
Add a variant to `Event`, `AgentEvent`, or `SandboxLifecycle` as appropriate. Sandbox
lifecycle facts come from two places: the pipeline emits `Initializing`, `Ready`, and
`InitializeFailed` around bringing the sandbox up, and `SandboxEventBridge` (in the
`fabro-workflow::event` module) translates the sandbox driver's own events — start,
stop, delete, image pulls, snapshot builds — into the rest. Fabro-sandbox emits no
events of its own.
### 2. Add tracing

20
lib/apps/fabro-cli/src/commands/run/run_progress/event.rs
View file

@ -53,8 +53,6 @@ pub(super) enum ProgressEvent {
provider: String,
duration_ms: u64,
name: Option<String>,
cpu: Option<f64>,
memory: Option<f64>,
url: Option<String>,
},
SandboxFailed {
@ -255,8 +253,6 @@ pub(super) fn from_run_event(stored: &RunEvent) -> Option<ProgressEvent> {
provider: props.provider.clone(),
duration_ms: props.duration_ms,
name: props.name.clone(),
cpu: props.cpu,
memory: props.memory,
url: props.url.clone(),
}),
EventBody::SandboxFailed(props) => Some(ProgressEvent::SandboxFailed {
@ -538,7 +534,7 @@ fn display_value(value: &Value) -> Option<String> {
mod tests {
use fabro_agent::AgentEvent;
use fabro_types::{MetadataSnapshotFailureKind, MetadataSnapshotPhase, fixtures};
use fabro_workflow::event::{Event, RunNoticeCode, to_run_event};
use fabro_workflow::event::{Event, RunNoticeCode, SandboxLifecycle, to_run_event};
use super::*;
@ -748,12 +744,10 @@ mod tests {
#[test]
fn round_trip_sandbox_ready() {
let event = Event::Sandbox {
event: fabro_agent::SandboxEvent::Ready {
event: SandboxLifecycle::Ready {
provider: "daytona".into(),
duration_ms: 2500,
name: Some("sandbox-1".into()),
cpu: Some(4.0),
memory: Some(8.0),
url: Some("https://example.test".into()),
},
};
@ -774,7 +768,7 @@ mod tests {
#[test]
fn round_trip_sandbox_failed() {
let event = Event::Sandbox {
event: fabro_agent::SandboxEvent::InitializeFailed {
event: SandboxLifecycle::InitializeFailed {
provider: "docker".into(),
error: "pull failed".into(),
causes: Vec::new(),
@ -794,23 +788,23 @@ mod tests {
#[test]
fn round_trip_snapshot_lifecycle_events() {
let pulling = to_run_event(&fixtures::RUN_1, &Event::Sandbox {
event: fabro_agent::SandboxEvent::SnapshotPulling {
event: SandboxLifecycle::SnapshotPulling {
name: "buildpack-deps:noble".into(),
},
});
let creating = to_run_event(&fixtures::RUN_1, &Event::Sandbox {
event: fabro_agent::SandboxEvent::SnapshotCreating {
event: SandboxLifecycle::SnapshotCreating {
name: "fabro-v9".into(),
},
});
let ready = to_run_event(&fixtures::RUN_1, &Event::Sandbox {
event: fabro_agent::SandboxEvent::SnapshotReady {
event: SandboxLifecycle::SnapshotReady {
name: "buildpack-deps:noble".into(),
duration_ms: 1200,
},
});
let failed = to_run_event(&fixtures::RUN_1, &Event::Sandbox {
event: fabro_agent::SandboxEvent::SnapshotFailed {
event: SandboxLifecycle::SnapshotFailed {
name: "fabro-v9".into(),
error: "build failed".into(),
causes: Vec::new(),

80
lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs
View file

@ -140,8 +140,6 @@ impl ProgressUI {
provider,
duration_ms,
name,
cpu,
memory,
url,
} => {
self.setup.on_sandbox_ready(
@ -149,8 +147,6 @@ impl ProgressUI {
&provider,
duration_ms,
name.as_deref(),
cpu,
memory,
url.as_deref(),
);
}
@ -457,7 +453,7 @@ mod tests {
use std::sync::{Arc, Mutex};
use chrono::{DateTime, Utc};
use fabro_agent::{AgentEvent, SandboxEvent};
use fabro_agent::AgentEvent;
use fabro_llm::types::TokenCounts;
use fabro_model::{Catalog, ModelRef, ProviderId};
use fabro_types::run_event::CliEnsureCompletedProps;
@ -465,7 +461,9 @@ mod tests {
MetadataSnapshotFailureKind, MetadataSnapshotPhase, ParallelBranchId, SandboxProviderKind,
StageId, fixtures,
};
use fabro_workflow::event::{Event, RunNoticeLevel, to_run_event, to_run_event_at};
use fabro_workflow::event::{
Event, RunNoticeLevel, SandboxLifecycle, to_run_event, to_run_event_at,
};
use fabro_workflow::outcome::billed_model_usage_from_llm;
use super::*;
@ -1048,17 +1046,15 @@ mod tests {
let (mut ui, buffer) = capture_ui(false);
emit(&mut ui, Event::Sandbox {
event: SandboxEvent::Initializing {
event: SandboxLifecycle::Initializing {
provider: "daytona".into(),
},
});
emit(&mut ui, Event::Sandbox {
event: SandboxEvent::Ready {
event: SandboxLifecycle::Ready {
provider: "daytona".into(),
duration_ms: 2500,
name: Some("sandbox-1".into()),
cpu: Some(4.0),
memory: Some(8.0),
url: None,
},
});
@ -1077,12 +1073,12 @@ mod tests {
duration_ms: 600,
}),
);
insta::assert_snapshot!(rendered(&buffer), @r"
Sandbox: daytona (ready in 2s)
sandbox-1 (4 cpu, 8 GB)
ssh daytona@example
Setup: 2 commands (8s)
CLI: gh (installed, 600ms)
insta::assert_snapshot!(rendered(&buffer), @"
Sandbox: daytona (ready in 2s)
sandbox-1
ssh daytona@example
Setup: 2 commands (8s)
CLI: gh (installed, 600ms)
");
}
@ -1091,36 +1087,34 @@ mod tests {
let (mut ui, buffer) = capture_ui(false);
emit(&mut ui, Event::Sandbox {
event: SandboxEvent::Initializing {
event: SandboxLifecycle::Initializing {
provider: "daytona".into(),
},
});
emit(&mut ui, Event::Sandbox {
event: SandboxEvent::SnapshotCreating {
event: SandboxLifecycle::SnapshotCreating {
name: "fabro-v9-test".into(),
},
});
emit(&mut ui, Event::Sandbox {
event: SandboxEvent::SnapshotReady {
event: SandboxLifecycle::SnapshotReady {
name: "fabro-v9-test".into(),
duration_ms: 210_000,
},
});
emit(&mut ui, Event::Sandbox {
event: SandboxEvent::Ready {
event: SandboxLifecycle::Ready {
provider: "daytona".into(),
duration_ms: 212_000,
name: Some("sandbox-1".into()),
cpu: Some(4.0),
memory: Some(8.0),
url: None,
},
});
insta::assert_snapshot!(rendered(&buffer), @r"
Sandbox: building fabro-v9-test...
Sandbox: daytona (ready in 3m32s)
sandbox-1 (4 cpu, 8 GB)
insta::assert_snapshot!(rendered(&buffer), @"
Sandbox: building fabro-v9-test...
Sandbox: daytona (ready in 3m32s)
sandbox-1
");
}
@ -1129,28 +1123,26 @@ mod tests {
let (mut ui, buffer) = capture_ui(false);
emit(&mut ui, Event::Sandbox {
event: SandboxEvent::Initializing {
event: SandboxLifecycle::Initializing {
provider: "docker".into(),
},
});
emit(&mut ui, Event::Sandbox {
event: SandboxEvent::SnapshotPulling {
event: SandboxLifecycle::SnapshotPulling {
name: "buildpack-deps:noble".into(),
},
});
emit(&mut ui, Event::Sandbox {
event: SandboxEvent::SnapshotReady {
event: SandboxLifecycle::SnapshotReady {
name: "buildpack-deps:noble".into(),
duration_ms: 8_200,
},
});
emit(&mut ui, Event::Sandbox {
event: SandboxEvent::Ready {
event: SandboxLifecycle::Ready {
provider: "docker".into(),
duration_ms: 9_000,
name: None,
cpu: None,
memory: None,
url: None,
},
});
@ -1166,17 +1158,15 @@ mod tests {
let (mut ui, buffer) = capture_ui(false);
emit(&mut ui, Event::Sandbox {
event: SandboxEvent::Initializing {
event: SandboxLifecycle::Initializing {
provider: "docker".into(),
},
});
emit(&mut ui, Event::Sandbox {
event: SandboxEvent::Ready {
event: SandboxLifecycle::Ready {
provider: "docker".into(),
duration_ms: 20,
name: None,
cpu: None,
memory: None,
url: None,
},
});
@ -1189,19 +1179,19 @@ mod tests {
let (mut ui, buffer) = capture_ui(false);
emit(&mut ui, Event::Sandbox {
event: SandboxEvent::Initializing {
event: SandboxLifecycle::Initializing {
provider: "docker".into(),
},
});
emit(&mut ui, Event::Sandbox {
event: SandboxEvent::SnapshotFailed {
event: SandboxLifecycle::SnapshotFailed {
name: "buildpack-deps:noble".into(),
error: "pull failed".into(),
causes: Vec::new(),
},
});
emit(&mut ui, Event::Sandbox {
event: SandboxEvent::InitializeFailed {
event: SandboxLifecycle::InitializeFailed {
provider: "docker".into(),
error: "pull failed".into(),
causes: Vec::new(),
@ -1220,14 +1210,14 @@ mod tests {
let mut ui = ProgressUI::new(true, false);
emit(&mut ui, Event::Sandbox {
event: SandboxEvent::Initializing {
event: SandboxLifecycle::Initializing {
provider: "docker".into(),
},
});
assert!(ui.setup.sandbox_bar.is_some());
emit(&mut ui, Event::Sandbox {
event: SandboxEvent::SnapshotReady {
event: SandboxLifecycle::SnapshotReady {
name: "buildpack-deps:noble".into(),
duration_ms: 10,
},
@ -1235,12 +1225,10 @@ mod tests {
assert!(ui.setup.sandbox_bar.is_some());
emit(&mut ui, Event::Sandbox {
event: SandboxEvent::Ready {
event: SandboxLifecycle::Ready {
provider: "docker".into(),
duration_ms: 20,
name: None,
cpu: None,
memory: None,
url: None,
},
});
@ -1252,14 +1240,14 @@ mod tests {
let mut ui = ProgressUI::new(true, false);
emit(&mut ui, Event::Sandbox {
event: SandboxEvent::Initializing {
event: SandboxLifecycle::Initializing {
provider: "docker".into(),
},
});
assert!(ui.setup.sandbox_bar.is_some());
emit(&mut ui, Event::Sandbox {
event: SandboxEvent::InitializeFailed {
event: SandboxLifecycle::InitializeFailed {
provider: "docker".into(),
error: "pull failed".into(),
causes: Vec::new(),

12
lib/apps/fabro-cli/src/commands/run/run_progress/setup_display.rs
View file

@ -56,20 +56,10 @@ impl SetupDisplay {
provider: &str,
duration_ms: u64,
name: Option<&str>,
cpu: Option<f64>,
memory: Option<f64>,
url: Option<&str>,
) {
let dur = format_duration_ms(duration_ms);
let detail = match (name, cpu, memory) {
(Some(name), Some(cpu), Some(memory)) => Some(format!(
"{name} ({} cpu, {} GB)",
styles::format_number(cpu),
styles::format_number(memory)
)),
(Some(name), _, _) => Some(name.to_string()),
_ => None,
};
let detail = name.map(str::to_string);
if renderer.is_tty() {
let display_provider = match url {

13
lib/apps/fabro-cli/src/commands/run/run_progress/styles.rs
View file

@ -68,19 +68,6 @@ pub(super) fn terminal_hyperlink(url: &str, text: &str) -> String {
format!("\x1b]8;;{url}\x1b\\{text}\x1b]8;;\x1b\\")
}
pub(super) fn format_number(n: f64) -> String {
if (n - n.round()).abs() < f64::EPSILON {
#[allow(
clippy::cast_possible_truncation,
reason = "Whole-number display intentionally narrows to i64 for formatting."
)]
let i = n as i64;
format!("{i}")
} else {
format!("{n:.1}")
}
}
pub(super) fn truncate(s: &str, max: usize) -> String {
let single_line = s.split_whitespace().collect::<Vec<_>>().join(" ");
if single_line.len() > max {

5
lib/components/fabro-agent/src/lib.rs
View file

@ -57,9 +57,8 @@ pub use question_tools::{
pub use sandbox::{
CommandOutputCallback, DirEntry, ExecResult, ExecStreamingRequest, ExecStreamingResult,
FileKind, GrepMatch, GrepOptions, OutputCaptureStats, RefreshOutcome, RemoteCredentialAction,
RunSandbox, SandboxEvent, SandboxEventCallback, SandboxFile, StderrCollector, StdioProcess,
StdioProcessHandle, TokenProvenance, TokenSnapshot, WalkOptions, format_lines_numbered,
shell_quote,
RunSandbox, SandboxFile, StderrCollector, StdioProcess, StdioProcessHandle, TokenProvenance,
TokenSnapshot, WalkOptions, format_lines_numbered, shell_quote,
};
pub use session::{
CompletionCoordinator, Session, SessionControlHandle, SessionInputTiming,

6
lib/components/fabro-agent/src/sandbox.rs
View file

@ -2,7 +2,7 @@
pub use fabro_sandbox::{
CommandOutputCallback, DirEntry, ExecResult, ExecStreamingRequest, ExecStreamingResult,
FileKind, GrepMatch, GrepOptions, OutputCaptureStats, RefreshOutcome, RemoteCredentialAction,
RunSandbox, SandboxEvent, SandboxEventCallback, SandboxFile, StderrCollector, StdioProcess,
StdioProcessHandle, StdioProcessTermination, TokenProvenance, TokenSnapshot, WalkOptions,
format_lines_numbered, shell_quote,
RunSandbox, SandboxFile, StderrCollector, StdioProcess, StdioProcessHandle,
StdioProcessTermination, TokenProvenance, TokenSnapshot, WalkOptions, format_lines_numbered,
shell_quote,
};

49
lib/components/fabro-sandbox/src/daytona.rs
View file

@ -16,7 +16,7 @@ use async_trait::async_trait;
use fabro_types::settings::server::ServerSandboxProviderSettings;
use fabro_types::{RunId, SandboxProviderKind};
use sandbox_driver::{
HealthStatus, LifecycleTimers, Resources, SandboxProvider, SandboxSource,
EventContext, HealthStatus, LifecycleTimers, Resources, SandboxProvider, SandboxSource,
SandboxSpec as DriverSpec, SnapshotId, SnapshotSource, SnapshotSpec,
};
use tokio::time;
@ -25,7 +25,6 @@ pub use crate::driver::DaytonaCredentials;
use crate::driver::{ProviderConnectOptions, connect_provider};
use crate::driver_sandbox::{CreatePlan, PreparedCreate, WorkspaceLayout};
use crate::options::SandboxOptions;
use crate::sandbox::SandboxEvent;
pub(crate) const WORKING_DIRECTORY: &str = "/home/daytona/workspace";
pub(crate) const REPOS_ROOT: &str = "/home/daytona/repos";
@ -332,6 +331,7 @@ async fn ensure_snapshot(
provider: &dyn SandboxProvider,
api_key: &str,
inputs: &SnapshotInputs<'_>,
events: Option<EventContext>,
) -> crate::Result<(SnapshotId, String)> {
let name = snapshot_identity::snapshot_name(api_key, inputs)?;
let snapshots = provider.snapshots().ok_or_else(|| {
@ -341,7 +341,7 @@ async fn ensure_snapshot(
.ensure(
&snapshot_spec(&name, inputs),
DAYTONA_SNAPSHOT_ACTIVE_TIMEOUT,
None,
events,
)
.await
.map_err(|error| {
@ -402,38 +402,12 @@ pub(crate) fn create_plan(
#[async_trait]
impl CreatePlan for DaytonaCreatePlan {
async fn prepare(
&self,
emit: &(dyn Fn(SandboxEvent) + Send + Sync),
) -> crate::Result<PreparedCreate> {
async fn prepare(&self, events: Option<EventContext>) -> crate::Result<PreparedCreate> {
let (snapshot_id, snapshot_name) = match snapshot_inputs(&self.options) {
// The driver finds, activates, builds, or waits for the snapshot
// as needed, and reports that work through `events`.
Some(inputs) => {
let started = time::Instant::now();
// The driver finds, activates, builds, or waits for the
// snapshot as needed; fabro reports the step around it.
let name = snapshot_identity::snapshot_name(&self.api_key, &inputs)?;
emit(SandboxEvent::SnapshotCreating { name });
let result = ensure_snapshot(self.provider.as_ref(), &self.api_key, &inputs).await;
match result {
Ok((id, name)) => {
emit(SandboxEvent::SnapshotReady {
name: name.clone(),
duration_ms: u64::try_from(started.elapsed().as_millis())
.unwrap_or(u64::MAX),
});
(id, name)
}
Err(error) => {
let name = snapshot_identity::snapshot_name(&self.api_key, &inputs)
.unwrap_or_default();
emit(SandboxEvent::SnapshotFailed {
name,
error: error.to_string(),
causes: error.causes(),
});
return Err(error);
}
}
ensure_snapshot(self.provider.as_ref(), &self.api_key, &inputs, events).await?
}
None => (
SnapshotId::try_new(DEFAULT_SNAPSHOT).expect("the default snapshot name is valid"),
@ -447,7 +421,6 @@ impl CreatePlan for DaytonaCreatePlan {
self.run_id.as_ref(),
&snapshot_id,
),
source: Some(snapshot_name.clone()),
snapshot: Some(snapshot_name),
})
}
@ -804,13 +777,7 @@ mod wire_gate {
let snapshot = SnapshotId::try_new(DEFAULT_SNAPSHOT).expect("snapshot id");
let options = SandboxOptions::default();
let spec = overlay(base_spec(&options, None), &options, None, &snapshot);
let sandbox = RunSandbox::pending(
SandboxProviderKind::DAYTONA,
remote,
spec,
Some(DEFAULT_SNAPSHOT.to_string()),
workspace,
);
let sandbox = RunSandbox::pending(SandboxProviderKind::DAYTONA, remote, spec, workspace);
sandbox
.initialize()
.await

443
lib/components/fabro-sandbox/src/driver_sandbox.rs
View file

@ -7,13 +7,14 @@
//! the handle or whether it runs in-process or over the plugin wire.
//!
//! What stays fabro's: the exec ladder, the credential filter on explicit
//! environment variables, the lifecycle events fabro records on a run, and
//! the run-facing conventions (`platform` names, grep line format, walk
//! results relative to a caller-declared base).
//! environment variables, and the run-facing conventions (`platform` names,
//! grep line format, walk results relative to a caller-declared base). The
//! driver reports lifecycle events itself, through the [`EventContext`] a
//! sandbox is created or attached with.
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, OnceLock, PoisonError};
use std::sync::{Arc, OnceLock};
use std::time::{Duration, Instant};
use async_trait::async_trait;
@ -22,10 +23,9 @@ use fabro_github::token_source::InstallationTokenSource;
use fabro_types::SandboxProviderKind;
use fabro_util::workspace_glob::WorkspaceGlob;
use sandbox_driver::{
Action, DirEntry, Event, EventBody, EventContext, EventObserver, FileKind, GrepMatch,
GrepOptions, LifecycleTimers, ProgressCode, PtyOptions, PtySize, Sandbox as DriverHandle,
SandboxProvider as DriverProvider, SandboxSource, SandboxSpec as DriverSpec, SandboxState,
Search as _, WaitOptions, WalkOptions,
DirEntry, EventContext, FileKind, GrepMatch, GrepOptions, LifecycleTimers, PtyOptions, PtySize,
Sandbox as DriverHandle, SandboxProvider as DriverProvider, SandboxSource,
SandboxSpec as DriverSpec, SandboxState, Search as _, WaitOptions, WalkOptions,
};
use sandbox_driver_host::HostProvider;
use tokio::fs;
@ -47,6 +47,14 @@ use crate::{GitRunInfo, GitSetupIntent, RefreshOutcome, RetryPlan};
/// later process rebuilds the handle by calling this again with the
/// persisted working directory rather than by id.
pub async fn local_sandbox(working_directory: impl Into<PathBuf>) -> crate::Result<RunSandbox> {
local_sandbox_with_events(working_directory, None).await
}
/// [`local_sandbox`] whose driver lifecycle events reach `events`.
pub async fn local_sandbox_with_events(
working_directory: impl Into<PathBuf>,
events: Option<EventContext>,
) -> crate::Result<RunSandbox> {
let working_directory: PathBuf = working_directory.into();
fs::create_dir_all(&working_directory)
.await
@ -55,7 +63,7 @@ pub async fn local_sandbox(working_directory: impl Into<PathBuf>) -> crate::Resu
let spec = DriverSpec::new(SandboxSource::HostDirectory)
.working_directory(working_directory.display().to_string());
let handle = provider
.create(&spec, None)
.create(&spec, events)
.await
.map_err(|error| crate::Error::context("Failed to create local sandbox", error))?;
let sandbox = RunSandbox::new(SandboxProviderKind::LOCAL, handle);
@ -65,7 +73,7 @@ pub async fn local_sandbox(working_directory: impl Into<PathBuf>) -> crate::Resu
use crate::exec::{ExplicitEnvPolicy, SandboxExec};
use crate::sandbox::{
self, ExecResult, ExecStreamingRequest, ExecStreamingResult, PushError, PushReport,
SandboxEvent, SandboxEventCallback, SandboxFile, SandboxWorkspaceLayout, StdioProcess,
SandboxFile, SandboxWorkspaceLayout, StdioProcess,
};
/// Where a clone-based provider puts its files: the run works under
@ -280,22 +288,17 @@ impl LayoutSource {
#[derive(Clone)]
pub(crate) struct PreparedCreate {
pub(crate) spec: DriverSpec,
/// The image or snapshot named by the spec, for pull progress events.
pub(crate) source: Option<String>,
/// The provider snapshot the sandbox is created from, when the provider
/// has that concept; recorded on the run.
pub(crate) snapshot: Option<String>,
}
/// Settles a create's inputs right before the provider call. A plan may
/// build provider resources first (a Daytona snapshot) and report progress
/// through fabro's events.
/// build provider resources first (a Daytona snapshot); the driver reports
/// that work through `events`.
#[async_trait]
pub(crate) trait CreatePlan: Send + Sync {
async fn prepare(
&self,
emit: &(dyn Fn(SandboxEvent) + Send + Sync),
) -> crate::Result<PreparedCreate>;
async fn prepare(&self, events: Option<EventContext>) -> crate::Result<PreparedCreate>;
}
/// A create whose spec is known up front.
@ -303,10 +306,7 @@ struct SpecPlan(PreparedCreate);
#[async_trait]
impl CreatePlan for SpecPlan {
async fn prepare(
&self,
_emit: &(dyn Fn(SandboxEvent) + Send + Sync),
) -> crate::Result<PreparedCreate> {
async fn prepare(&self, _events: Option<EventContext>) -> crate::Result<PreparedCreate> {
Ok(self.0.clone())
}
}
@ -320,19 +320,22 @@ struct PendingCreate {
/// A fabro sandbox backed by a sandbox-driver handle.
pub struct RunSandbox {
kind: SandboxProviderKind,
kind: SandboxProviderKind,
/// Set at construction for an existing sandbox, at `initialize` for a
/// pending one.
handle: OnceCell<Arc<dyn DriverHandle>>,
pending: Option<PendingCreate>,
workspace: Option<RepoWorkspace>,
env_policy: ExplicitEnvPolicy,
event_callback: Option<SandboxEventCallback>,
handle: OnceCell<Arc<dyn DriverHandle>>,
pending: Option<PendingCreate>,
workspace: Option<RepoWorkspace>,
env_policy: ExplicitEnvPolicy,
/// Where the driver reports the lifecycle of a sandbox this creates.
/// Set before `initialize` on a pending sandbox; an existing handle
/// already carries the context it was created or attached with.
events: Option<EventContext>,
/// `(platform, os_version)` learned from the sandbox at initialize or
/// start; unknown until then.
platform: OnceLock<(String, String)>,
platform: OnceLock<(String, String)>,
/// The provider snapshot the sandbox was created from, when known.
snapshot: OnceLock<String>,
snapshot: OnceLock<String>,
}
impl RunSandbox {
@ -366,7 +369,6 @@ impl RunSandbox {
kind: SandboxProviderKind,
provider: Arc<dyn DriverProvider>,
spec: DriverSpec,
source: Option<String>,
workspace: RepoWorkspace,
) -> Self {
Self::pending_with_plan(
@ -374,7 +376,6 @@ impl RunSandbox {
provider,
Box::new(SpecPlan(PreparedCreate {
spec,
source,
snapshot: None,
})),
workspace,
@ -425,14 +426,17 @@ impl RunSandbox {
pending: None,
workspace: None,
env_policy,
event_callback: None,
events: None,
platform: OnceLock::new(),
snapshot: OnceLock::new(),
}
}
pub fn set_event_callback(&mut self, cb: SandboxEventCallback) {
self.event_callback = Some(cb);
/// Where the driver reports this sandbox's lifecycle once `initialize`
/// creates it. An existing handle reports through the context it was
/// created or attached with, so this only matters for a pending sandbox.
pub fn set_events(&mut self, events: EventContext) {
self.events = Some(events);
}
/// The provider kind fabro persists for this sandbox.
@ -479,17 +483,6 @@ impl RunSandbox {
}
}
fn provider_name(&self) -> String {
self.kind.to_string()
}
fn emit(&self, event: SandboxEvent) {
event.trace();
if let Some(cb) = &self.event_callback {
cb(event);
}
}
fn search(&self) -> crate::Result<sandbox_driver::SearchFacet<'_>> {
self.handle()?.search().ok_or_else(|| {
crate::Error::message(format!(
@ -507,17 +500,13 @@ impl RunSandbox {
let Some(pending) = &self.pending else {
return self.handle().map(|_| ());
};
let prepared = pending.plan.prepare(&|event| self.emit(event)).await?;
let prepared = pending.plan.prepare(self.events.clone()).await?;
if let Some(snapshot) = prepared.snapshot {
let _ = self.snapshot.set(snapshot);
}
let observer = Arc::new(CreateProgress::new(
prepared.source,
self.event_callback.clone(),
));
let handle = pending
.provider
.create(&prepared.spec, Some(EventContext::new(observer)))
.create(&prepared.spec, self.events.clone())
.await
.map_err(|error| {
crate::Error::context(format!("Failed to create {} sandbox", self.kind), error)
@ -569,10 +558,11 @@ impl RunSandbox {
Ok(())
}
WorkspacePlan::Clone(plan) => {
self.emit(SandboxEvent::GitCloneStarted {
url: plan.origin_url.clone(),
branch: plan.branch.clone(),
});
tracing::debug!(
url = plan.origin_url.as_str(),
branch = plan.branch.as_deref().unwrap_or(""),
"Git clone started"
);
let started = Instant::now();
let handle = self.handle()?;
// The clone names every directory it touches, so it runs
@ -598,18 +588,20 @@ impl RunSandbox {
let _ = workspace
.execution_directory
.set(outcome.layout.execution_directory.clone());
self.emit(SandboxEvent::GitCloneCompleted {
url: plan.origin_url.clone(),
duration_ms: elapsed_ms(started),
});
tracing::debug!(
url = plan.origin_url.as_str(),
duration_ms = elapsed_ms(started),
"Git clone completed"
);
Ok(())
}
Err(error) => {
self.emit(SandboxEvent::GitCloneFailed {
url: plan.origin_url.clone(),
error: error.to_string(),
causes: error.causes(),
});
tracing::error!(
url = plan.origin_url.as_str(),
error = %error,
causes = ?error.causes(),
"Git clone failed"
);
Err(error)
}
}
@ -683,92 +675,6 @@ impl RunSandbox {
}
}
/// Turns the driver's create-time progress into fabro's snapshot events:
/// an image pull starts `SnapshotPulling` and the create's completion ends
/// it. A create without a pull emits nothing.
struct CreateProgress {
source: Option<String>,
callback: Option<SandboxEventCallback>,
pull_started: Mutex<Option<Instant>>,
}
impl CreateProgress {
fn new(source: Option<String>, callback: Option<SandboxEventCallback>) -> Self {
Self {
source,
callback,
pull_started: Mutex::new(None),
}
}
fn emit(&self, event: SandboxEvent) {
event.trace();
if let Some(cb) = &self.callback {
cb(event);
}
}
fn name(&self) -> String {
self.source.clone().unwrap_or_default()
}
}
#[async_trait]
impl EventObserver for CreateProgress {
async fn observe(&self, event: Event) {
match &event.body {
EventBody::OperationProgress { progress, .. }
if progress.code.as_str() == ProgressCode::IMAGE_PULL =>
{
let mut started = self
.pull_started
.lock()
.unwrap_or_else(PoisonError::into_inner);
if started.is_none() {
*started = Some(Instant::now());
drop(started);
self.emit(SandboxEvent::SnapshotPulling { name: self.name() });
}
}
EventBody::OperationCompleted {
action: Action::Create,
..
} => {
let started = self
.pull_started
.lock()
.unwrap_or_else(PoisonError::into_inner)
.take();
if let Some(started) = started {
self.emit(SandboxEvent::SnapshotReady {
name: self.name(),
duration_ms: elapsed_ms(started),
});
}
}
EventBody::OperationFailed {
action: Action::Create,
error,
..
} => {
let started = self
.pull_started
.lock()
.unwrap_or_else(PoisonError::into_inner)
.take();
if started.is_some() {
self.emit(SandboxEvent::SnapshotFailed {
name: self.name(),
error: error.message.clone(),
causes: error.causes.clone(),
});
}
}
_ => {}
}
}
}
/// Fabro names the macOS platform `darwin`, as `uname -s` does.
fn fabro_platform_name(os: &str) -> &str {
match os {
@ -1009,46 +915,24 @@ impl RunSandbox {
/// Create the sandbox when it is pending, bring it to `Running`, and
/// prepare fabro's workspace (empty root or clone) on first use.
pub async fn initialize(&self) -> crate::Result<()> {
self.emit(SandboxEvent::Initializing {
provider: self.provider_name(),
});
let started = Instant::now();
let result = async {
self.ensure_created().await?;
self.make_ready().await?;
self.prepare_workspace().await
self.ensure_created().await?;
self.make_ready().await?;
self.prepare_workspace().await
}
/// The provider's console page for this sandbox, when it has one. Best
/// effort: a failed describe reports no page. The local sandbox is the
/// host and has none.
pub async fn console_url(&self) -> Option<String> {
if self.kind.is_local() {
return None;
}
.await;
let duration_ms = elapsed_ms(started);
match &result {
Ok(()) => {
// The provider's console page, when it has one. Best effort:
// a failed describe never fails a successful initialize.
let url = match self.handle() {
Ok(handle) if !self.kind.is_local() => handle
.describe()
.await
.ok()
.and_then(|status| status.web_url),
_ => None,
};
self.emit(SandboxEvent::Ready {
provider: self.provider_name(),
duration_ms,
name: Some(self.sandbox_info()).filter(|name| !name.is_empty()),
cpu: None,
memory: None,
url,
});
}
Err(error) => self.emit(SandboxEvent::InitializeFailed {
provider: self.provider_name(),
error: error.to_string(),
causes: error.causes(),
duration_ms,
}),
}
result
self.handle()
.ok()?
.describe()
.await
.ok()
.and_then(|status| status.web_url)
}
/// Idempotent access-time check: a running sandbox is left alone; a
@ -1062,89 +946,22 @@ impl RunSandbox {
}
pub async fn start(&self) -> crate::Result<()> {
self.emit(SandboxEvent::StartStarted {
provider: self.provider_name(),
});
let started = Instant::now();
let result = self.make_ready().await;
match &result {
Ok(()) => self.emit(SandboxEvent::StartCompleted {
provider: self.provider_name(),
duration_ms: elapsed_ms(started),
}),
Err(error) => self.emit(SandboxEvent::StartFailed {
provider: self.provider_name(),
error: error.to_string(),
causes: error.causes(),
}),
}
result
self.make_ready().await
}
pub async fn stop(&self) -> crate::Result<()> {
self.emit(SandboxEvent::StopStarted {
provider: self.provider_name(),
});
let started = Instant::now();
let result = match self.handle() {
Ok(handle) => handle.stop().await.map_err(crate::Error::from),
Err(error) => Err(error),
};
match &result {
Ok(()) => self.emit(SandboxEvent::StopCompleted {
provider: self.provider_name(),
duration_ms: elapsed_ms(started),
}),
Err(error) => self.emit(SandboxEvent::StopFailed {
provider: self.provider_name(),
error: error.to_string(),
causes: error.causes(),
}),
}
result
self.handle()?.stop().await.map_err(crate::Error::from)
}
pub async fn delete(&self) -> crate::Result<()> {
self.emit(SandboxEvent::DeleteStarted {
provider: self.provider_name(),
});
let started = Instant::now();
let result = self.release().await;
match &result {
Ok(()) => self.emit(SandboxEvent::DeleteCompleted {
provider: self.provider_name(),
duration_ms: elapsed_ms(started),
}),
Err(error) => self.emit(SandboxEvent::DeleteFailed {
provider: self.provider_name(),
error: error.to_string(),
causes: error.causes(),
}),
}
result
self.release().await
}
/// Releases the sandbox. For a designated host directory this frees the
/// handle and leaves the directory in place; for an isolated provider it
/// removes the sandbox.
pub async fn cleanup(&self) -> crate::Result<()> {
self.emit(SandboxEvent::CleanupStarted {
provider: self.provider_name(),
});
let started = Instant::now();
let result = self.release().await;
match &result {
Ok(()) => self.emit(SandboxEvent::CleanupCompleted {
provider: self.provider_name(),
duration_ms: elapsed_ms(started),
}),
Err(error) => self.emit(SandboxEvent::CleanupFailed {
provider: self.provider_name(),
error: error.to_string(),
causes: error.causes(),
}),
}
result
self.release().await
}
/// The directory the run works in: the cloned repository's link for a
@ -1362,6 +1179,8 @@ fn elapsed_ms(started: Instant) -> u64 {
#[cfg(test)]
mod tests {
use std::sync::Mutex;
use fabro_types::CommandTermination;
use sandbox_driver::{SandboxProvider as _, SandboxSource, SandboxSpec};
use sandbox_driver_host::HostProvider;
@ -1582,71 +1401,85 @@ mod tests {
);
}
#[tokio::test]
async fn initialize_emits_lifecycle_events_and_learns_the_platform() {
let mut f = fixture().await;
let events: Arc<Mutex<Vec<SandboxEvent>>> = Arc::new(Mutex::new(Vec::new()));
let captured = Arc::clone(&events);
f.sandbox.set_event_callback(Arc::new(move |event| {
captured.lock().unwrap().push(event);
}));
assert_eq!(f.sandbox.platform(), "unknown");
/// Collects the driver's events for assertions.
struct Recorded(Mutex<Vec<sandbox_driver::Event>>);
f.sandbox.initialize().await.unwrap();
#[async_trait]
impl sandbox_driver::EventObserver for Recorded {
async fn observe(&self, event: sandbox_driver::Event) {
self.0.lock().unwrap().push(event);
}
}
#[tokio::test]
async fn lifecycle_reaches_the_driver_events_and_learns_the_platform() {
let dir = tempfile::tempdir().unwrap();
let recorded = Arc::new(Recorded(Mutex::new(Vec::new())));
let sandbox = local_sandbox_with_events(
dir.path(),
Some(EventContext::new(
Arc::clone(&recorded) as Arc<dyn sandbox_driver::EventObserver>
)),
)
.await
.unwrap();
let expected = if cfg!(target_os = "macos") {
"darwin"
} else {
std::env::consts::OS
};
assert_eq!(f.sandbox.platform(), expected);
assert!(f.sandbox.os_version().starts_with(expected));
assert_eq!(sandbox.platform(), expected);
assert!(sandbox.os_version().starts_with(expected));
assert_eq!(
f.sandbox.sandbox_info(),
sandbox.sandbox_info(),
"",
"local sandboxes are identified by directory"
);
let handle = Arc::clone(f.sandbox.handle().unwrap());
let handle = Arc::clone(sandbox.handle().unwrap());
let isolated = RunSandbox::new(SandboxProviderKind::DOCKER, Arc::clone(&handle));
assert_eq!(isolated.sandbox_info(), handle.id().to_string());
assert_eq!(sandbox.console_url().await, None);
f.sandbox.stop().await.unwrap();
f.sandbox.activate().await.unwrap();
f.sandbox.cleanup().await.unwrap();
sandbox.stop().await.unwrap();
sandbox.activate().await.unwrap();
sandbox.cleanup().await.unwrap();
assert!(
f.dir.path().is_dir(),
dir.path().is_dir(),
"designated directories survive cleanup"
);
let captured = events.lock().unwrap();
let names: Vec<&str> = captured
let captured = recorded.0.lock().unwrap();
let steps: Vec<String> = captured
.iter()
.map(|event| match event {
SandboxEvent::Initializing { .. } => "initializing",
SandboxEvent::Ready { .. } => "ready",
SandboxEvent::StopStarted { .. } => "stop_started",
SandboxEvent::StopCompleted { .. } => "stop_completed",
SandboxEvent::CleanupStarted { .. } => "cleanup_started",
SandboxEvent::CleanupCompleted { .. } => "cleanup_completed",
_ => "other",
.filter_map(|event| match &event.body {
sandbox_driver::EventBody::OperationStarted { action } => {
Some(format!("{action:?} started"))
}
sandbox_driver::EventBody::OperationCompleted { action, .. } => {
Some(format!("{action:?} completed"))
}
sandbox_driver::EventBody::OperationFailed { action, .. } => {
Some(format!("{action:?} failed"))
}
_ => None,
})
.collect();
assert_eq!(names, vec![
"initializing",
"ready",
"stop_started",
"stop_completed",
"cleanup_started",
"cleanup_completed",
assert_eq!(steps, vec![
"Create started",
"Create completed",
"Stop started",
"Stop completed",
"Start started",
"Start completed",
"Delete started",
"Delete completed",
]);
assert!(captured.iter().all(|event| match event {
SandboxEvent::Initializing { provider }
| SandboxEvent::Ready { provider, .. }
| SandboxEvent::StopStarted { provider }
| SandboxEvent::StopCompleted { provider, .. }
| SandboxEvent::CleanupStarted { provider }
| SandboxEvent::CleanupCompleted { provider, .. } => provider == "local",
_ => true,
}));
assert!(
captured
.iter()
.all(|event| event.provider.to_string() == "host"),
"the driver names its own provider"
);
}
#[tokio::test]

9
lib/components/fabro-sandbox/src/lib.rs
View file

@ -58,15 +58,14 @@ pub use provider::{
pub use provider_sandbox::{attach_provider_sandbox, provider_sandbox};
pub use push_credentials::RefreshErrorKind;
pub use reconnect::{
reconnect, reconnect_driver_for_run, reconnect_for_run, reconnect_for_run_with_callback,
reconnect, reconnect_driver_for_run, reconnect_for_run, reconnect_for_run_with_events,
};
pub use sandbox::{
CommandOutputCallback, DEFAULT_EXEC_OUTPUT_TAIL_BYTES, ExecResult, ExecStreamingRequest,
ExecStreamingResult, GitRunInfo, GitSetupIntent, OutputCaptureStats, PushAttempt, PushError,
PushReport, RefreshOutcome, RemoteCredentialAction, SandboxEvent, SandboxEventCallback,
SandboxFile, SandboxWorkspaceLayout, StderrCollector, StdioProcess, StdioProcessHandle,
StdioProcessTermination, format_lines_numbered, redacted_output_tail, setup_git_via_exec,
shell_quote,
PushReport, RefreshOutcome, RemoteCredentialAction, SandboxFile, SandboxWorkspaceLayout,
StderrCollector, StdioProcess, StdioProcessHandle, StdioProcessTermination,
format_lines_numbered, redacted_output_tail, setup_git_via_exec, shell_quote,
};
/// Driver types a run sandbox's file and search operations speak, and the
/// network policy a [`SandboxOptions`] asks for, re-exported so consumers

14
lib/components/fabro-sandbox/src/provider_sandbox.rs
View file

@ -13,7 +13,7 @@ use std::sync::Arc;
use fabro_github::GitHubCredentials;
use fabro_types::{BundledProvider, RunId, SandboxProviderKind};
use sandbox_driver::{OwnedProvider, SandboxId, SandboxProvider};
use sandbox_driver::{EventContext, OwnedProvider, SandboxId, SandboxProvider};
use crate::driver::{ProviderAccess, connect_provider};
use crate::driver_sandbox::{LayoutSource, RepoWorkspace, RunSandbox};
@ -53,8 +53,8 @@ pub async fn provider_sandbox(
let base = options::base_spec(&options, run_id.as_ref());
Ok(match kind.bundled() {
Some(BundledProvider::Docker) => {
let (spec, image) = docker::overlay(base, &options);
RunSandbox::pending(kind, provider, spec, Some(image), workspace)
let (spec, _image) = docker::overlay(base, &options);
RunSandbox::pending(kind, provider, spec, workspace)
}
Some(BundledProvider::Daytona) => {
let credentials = access
@ -78,12 +78,13 @@ pub async fn provider_sandbox(
None => {
let mut spec = base;
spec.network = options::supported_network(spec.network, provider.capabilities());
RunSandbox::pending(kind, provider, spec, options.image.clone(), workspace)
RunSandbox::pending(kind, provider, spec, workspace)
}
})
}
/// Reattach to a run's sandbox on `kind` by its persisted id.
/// Reattach to a run's sandbox on `kind` by its persisted id. The driver
/// reports the sandbox's lifecycle from here on through `events`.
///
/// The sandbox must carry fabro's managed label and, when a run id is
/// known, the matching run label: the provider shares its backend with
@ -98,11 +99,12 @@ pub async fn attach_provider_sandbox(
working_directory: String,
clone_origin_url: Option<String>,
run_id: Option<RunId>,
events: Option<EventContext>,
) -> crate::Result<RunSandbox> {
let provider = connect(&kind, access, run_id.as_ref()).await?;
let id = SandboxId::try_new(sandbox_id)
.map_err(|error| crate::Error::context(format!("Invalid {kind} sandbox id"), error))?;
let handle = provider.attach(&id, None).await.map_err(|error| {
let handle = provider.attach(&id, events).await.map_err(|error| {
crate::Error::context(
format!("Failed to reconnect {kind} sandbox '{sandbox_id}'"),
error,

23
lib/components/fabro-sandbox/src/reconnect.rs
View file

@ -2,10 +2,11 @@ use std::path::PathBuf;
use anyhow::{Context, Result};
use fabro_types::{BundledProvider, RunId, RunSandboxInstance};
use sandbox_driver::EventContext;
use crate::driver::ProviderAccess;
use crate::driver_sandbox::{RunSandbox, local_sandbox};
use crate::{SandboxEventCallback, provider_sandbox};
use crate::driver_sandbox::{RunSandbox, local_sandbox_with_events};
use crate::provider_sandbox;
/// Reconnect to a sandbox from a saved record.
///
@ -20,16 +21,16 @@ pub async fn reconnect_for_run(
access: &ProviderAccess,
run_id: Option<RunId>,
) -> Result<RunSandbox> {
reconnect_for_run_with_callback(record, access, run_id, None).await
reconnect_for_run_with_events(record, access, run_id, None).await
}
pub async fn reconnect_for_run_with_callback(
pub async fn reconnect_for_run_with_events(
record: &RunSandboxInstance,
access: &ProviderAccess,
run_id: Option<RunId>,
event_callback: Option<SandboxEventCallback>,
events: Option<EventContext>,
) -> Result<RunSandbox> {
reconnect_driver_for_run(record, access, run_id, event_callback).await
reconnect_driver_for_run(record, access, run_id, events).await
}
/// Reconnects as the driver-backed sandbox type, for callers that need a
@ -39,14 +40,14 @@ pub async fn reconnect_driver_for_run(
record: &RunSandboxInstance,
access: &ProviderAccess,
run_id: Option<RunId>,
event_callback: Option<SandboxEventCallback>,
events: Option<EventContext>,
) -> Result<RunSandbox> {
let runtime = &record.runtime;
// A local sandbox is its working directory: rebuilding the handle over
// that directory is the reconnect. The per-process Host registry holds
// no state worth attaching to.
let mut sandbox = if record.provider.bundled() == Some(BundledProvider::Local) {
local_sandbox(PathBuf::from(&runtime.working_directory))
let sandbox = if record.provider.bundled() == Some(BundledProvider::Local) {
local_sandbox_with_events(PathBuf::from(&runtime.working_directory), events)
.await
.context("Failed to reconnect local sandbox")?
} else {
@ -64,12 +65,10 @@ pub async fn reconnect_driver_for_run(
runtime.working_directory.clone(),
runtime.clone_origin_url.clone(),
run_id,
events,
)
.await
.with_context(|| format!("Failed to reconnect {} sandbox", record.provider))?
};
if let Some(callback) = event_callback {
sandbox.set_event_callback(callback);
}
Ok(sandbox)
}

311
lib/components/fabro-sandbox/src/sandbox.rs
View file

@ -58,237 +58,6 @@ pub enum GitSetupIntent {
},
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum SandboxEvent {
// -- Common lifecycle --
Initializing {
provider: String,
},
Ready {
provider: String,
duration_ms: u64,
name: Option<String>,
cpu: Option<f64>,
memory: Option<f64>,
url: Option<String>,
},
InitializeFailed {
provider: String,
error: String,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
causes: Vec<String>,
duration_ms: u64,
},
CleanupStarted {
provider: String,
},
CleanupCompleted {
provider: String,
duration_ms: u64,
},
CleanupFailed {
provider: String,
error: String,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
causes: Vec<String>,
},
StartStarted {
provider: String,
},
StartCompleted {
provider: String,
duration_ms: u64,
},
StartFailed {
provider: String,
error: String,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
causes: Vec<String>,
},
StopStarted {
provider: String,
},
StopCompleted {
provider: String,
duration_ms: u64,
},
StopFailed {
provider: String,
error: String,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
causes: Vec<String>,
},
DeleteStarted {
provider: String,
},
DeleteCompleted {
provider: String,
duration_ms: u64,
},
DeleteFailed {
provider: String,
error: String,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
causes: Vec<String>,
},
// -- Snapshot lifecycle --
SnapshotPulling {
name: String,
},
SnapshotCreating {
name: String,
},
SnapshotReady {
name: String,
duration_ms: u64,
},
SnapshotFailed {
name: String,
error: String,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
causes: Vec<String>,
},
// -- Daytona git --
GitCloneStarted {
url: String,
branch: Option<String>,
},
GitCloneCompleted {
url: String,
duration_ms: u64,
},
GitCloneFailed {
url: String,
error: String,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
causes: Vec<String>,
},
}
impl SandboxEvent {
pub fn trace(&self) {
use tracing::{debug, error, info, warn};
match self {
Self::Initializing { provider } => {
debug!(provider, "Sandbox initializing");
}
Self::Ready {
provider,
duration_ms,
..
} => {
info!(provider, duration_ms, "Sandbox ready");
}
Self::InitializeFailed {
provider,
error,
causes,
duration_ms,
} => {
error!(provider, error, causes = ?causes, duration_ms, "Sandbox init failed");
}
Self::CleanupStarted { provider } => {
info!(provider, "Sandbox cleanup started");
}
Self::CleanupCompleted {
provider,
duration_ms,
} => {
info!(provider, duration_ms, "Sandbox cleanup completed");
}
Self::CleanupFailed {
provider,
error,
causes,
} => {
warn!(provider, error, causes = ?causes, "Sandbox cleanup failed");
}
Self::StartStarted { provider } => {
info!(provider, "Sandbox start started");
}
Self::StartCompleted {
provider,
duration_ms,
} => {
info!(provider, duration_ms, "Sandbox start completed");
}
Self::StartFailed {
provider,
error,
causes,
} => {
warn!(provider, error, causes = ?causes, "Sandbox start failed");
}
Self::StopStarted { provider } => {
info!(provider, "Sandbox stop started");
}
Self::StopCompleted {
provider,
duration_ms,
} => {
info!(provider, duration_ms, "Sandbox stop completed");
}
Self::StopFailed {
provider,
error,
causes,
} => {
warn!(provider, error, causes = ?causes, "Sandbox stop failed");
}
Self::DeleteStarted { provider } => {
info!(provider, "Sandbox delete started");
}
Self::DeleteCompleted {
provider,
duration_ms,
} => {
info!(provider, duration_ms, "Sandbox delete completed");
}
Self::DeleteFailed {
provider,
error,
causes,
} => {
warn!(provider, error, causes = ?causes, "Sandbox delete failed");
}
Self::SnapshotPulling { name } => {
debug!(name, "Snapshot pulling");
}
Self::SnapshotCreating { name } => {
debug!(name, "Snapshot creating");
}
Self::SnapshotReady { name, duration_ms } => {
info!(name, duration_ms, "Snapshot ready");
}
Self::SnapshotFailed {
name,
error,
causes,
} => {
error!(name, error, causes = ?causes, "Snapshot failed");
}
Self::GitCloneStarted { url, branch } => {
debug!(
url,
branch = branch.as_deref().unwrap_or(""),
"Git clone started"
);
}
Self::GitCloneCompleted { url, duration_ms } => {
debug!(url, duration_ms, "Git clone completed");
}
Self::GitCloneFailed { url, error, causes } => {
error!(url, error, causes = ?causes, "Git clone failed");
}
}
}
}
/// Callback type for sandbox events.
pub type SandboxEventCallback = Arc<dyn Fn(SandboxEvent) + Send + Sync>;
/// Formats file content with line numbers for display.
///
/// Applies optional offset (1-based starting line number) and limit (max lines
@ -1982,86 +1751,6 @@ mod tests {
);
}
#[test]
fn sandbox_event_serialization_round_trip() {
let events = vec![
SandboxEvent::Initializing {
provider: "local".into(),
},
SandboxEvent::Ready {
provider: "local".into(),
duration_ms: 50,
name: None,
cpu: None,
memory: None,
url: None,
},
SandboxEvent::InitializeFailed {
provider: "docker".into(),
error: "no daemon".into(),
causes: vec!["connection refused".into()],
duration_ms: 100,
},
SandboxEvent::CleanupStarted {
provider: "daytona".into(),
},
SandboxEvent::CleanupCompleted {
provider: "daytona".into(),
duration_ms: 200,
},
SandboxEvent::CleanupFailed {
provider: "docker".into(),
error: "container gone".into(),
causes: Vec::new(),
},
SandboxEvent::SnapshotPulling {
name: "ubuntu:22.04".into(),
},
SandboxEvent::SnapshotCreating {
name: "my-snap".into(),
},
SandboxEvent::SnapshotReady {
name: "my-snap".into(),
duration_ms: 30000,
},
SandboxEvent::SnapshotFailed {
name: "my-snap".into(),
error: "build failed".into(),
causes: Vec::new(),
},
SandboxEvent::GitCloneStarted {
url: "https://github.com/org/repo.git".into(),
branch: Some("main".into()),
},
SandboxEvent::GitCloneCompleted {
url: "https://github.com/org/repo.git".into(),
duration_ms: 8000,
},
SandboxEvent::GitCloneFailed {
url: "https://github.com/org/repo.git".into(),
error: "auth failed".into(),
causes: Vec::new(),
},
];
assert_eq!(events.len(), 13, "should test all 13 variants");
for event in &events {
let json = serde_json::to_string(event).unwrap();
let deserialized: SandboxEvent = serde_json::from_str(&json).unwrap();
let json2 = serde_json::to_string(&deserialized).unwrap();
assert_eq!(json, json2);
}
}
#[test]
fn sandbox_event_callback_type_compiles() {
let cb: SandboxEventCallback = Arc::new(|_event| {});
cb(SandboxEvent::Initializing {
provider: "test".into(),
});
}
#[test]
fn format_lines_numbered_basic() {
let result = format_lines_numbered("hello\nworld\nfoo", None, None);

29
lib/components/fabro-sandbox/src/sandbox_spec.rs
View file

@ -4,11 +4,12 @@ use std::sync::Arc;
use anyhow::Context as _;
use fabro_github::GitHubCredentials;
use fabro_types::{RunId, RunSandboxInstance, RunSandboxRuntime, SandboxProviderKind};
use sandbox_driver::EventContext;
use crate::driver::ProviderAccess;
use crate::driver_sandbox::{LayoutSource, RunSandbox, local_sandbox};
use crate::driver_sandbox::{LayoutSource, RunSandbox, local_sandbox_with_events};
use crate::options::SandboxOptions;
use crate::{SandboxEventCallback, clone_source, provider_sandbox};
use crate::{clone_source, provider_sandbox};
/// Options for sandbox initialization and construction.
#[derive(Clone, Debug)]
@ -49,6 +50,16 @@ impl SandboxSpec {
self.provider().to_string()
}
/// The image the run record names for this sandbox: the environment's,
/// or the provider's default when the environment names none. A local
/// sandbox has no image.
pub fn image(&self) -> Option<String> {
match self {
Self::Local { .. } => None,
Self::Provider(spec) => provider_sandbox::recorded_image(&spec.kind, &spec.options),
}
}
/// Build initialized sandbox metadata for persistence.
pub fn to_run_sandbox_instance(
&self,
@ -143,18 +154,18 @@ impl SandboxSpec {
}
}
/// Builds the sandbox. The driver reports its lifecycle through
/// `events`: the local sandbox's from creation here, a provider
/// sandbox's from `initialize` on.
pub async fn build(
&self,
event_callback: Option<SandboxEventCallback>,
events: Option<EventContext>,
) -> Result<Arc<RunSandbox>, anyhow::Error> {
match self {
Self::Local { working_directory } => {
let mut sandbox = local_sandbox(working_directory.clone())
let sandbox = local_sandbox_with_events(working_directory.clone(), events)
.await
.context("Failed to create local sandbox")?;
if let Some(callback) = event_callback {
sandbox.set_event_callback(callback);
}
Ok(Arc::new(sandbox))
}
Self::Provider(spec) => {
@ -182,8 +193,8 @@ impl SandboxSpec {
)
.await
.with_context(|| format!("Failed to create {kind} sandbox"))?;
if let Some(callback) = event_callback {
sandbox.set_event_callback(callback);
if let Some(events) = events {
sandbox.set_events(events);
}
Ok(Arc::new(sandbox))
}

9
lib/components/fabro-sandbox/src/test_support.rs
View file

@ -17,7 +17,7 @@ pub use sandbox_driver_testing::{ScriptedExec, ScriptedSandbox, ScriptedStdioPro
use tokio::io::DuplexStream;
use crate::driver_sandbox::RunSandbox;
use crate::sandbox::{ExecResult, SandboxEventCallback, SandboxFile, StderrCollector};
use crate::sandbox::{ExecResult, SandboxFile, StderrCollector};
// --- MockSandbox ---
@ -45,7 +45,6 @@ pub struct MockSandbox {
/// Fails `activate` after the sandbox is built, as a sandbox whose
/// Bash contract broke would.
pub activate_error: Option<String>,
pub event_callback: Option<SandboxEventCallback>,
pub stdio_process: Option<MockStdioProcess>,
pub stdio_process_error: Option<String>,
/// Lines every grep returns, as `path:line:content`.
@ -85,7 +84,6 @@ impl Default for MockSandbox {
platform_str: "darwin",
os_version_str: "Darwin 24.0.0".into(),
activate_error: None,
event_callback: None,
stdio_process: None,
stdio_process_error: None,
grep_results: Vec::new(),
@ -166,15 +164,12 @@ impl MockSandbox {
let driver = Arc::new(self.build_driver());
// An isolated provider: explicit environment passes as the
// caller composed it, as it does for Docker and Daytona runs.
let mut run = RunSandbox::new_with_platform(
let run = RunSandbox::new_with_platform(
SandboxProviderKind::DOCKER,
Arc::clone(&driver) as Arc<dyn sandbox_driver::Sandbox>,
self.platform_str,
self.os_version_str.clone(),
);
if let Some(callback) = &self.event_callback {
run.set_event_callback(Arc::clone(callback));
}
Built {
run: Arc::new(run),
driver,

2
lib/components/fabro-workflow/Cargo.toml
View file

@ -30,6 +30,7 @@ fabro-hooks = { path = "../fabro-hooks" }
fabro-validate = { path = "../fabro-validate" }
fabro-dump = { path = "../fabro-dump" }
fabro-sandbox = { path = "../fabro-sandbox" }
sandbox-driver.workspace = true
fabro-mcp = { path = "../fabro-mcp" }
fabro-github = { path = "../fabro-github" }
fabro-interview = { path = "../fabro-interview" }
@ -84,7 +85,6 @@ fabro-workflow = { path = ".", features = ["test-support"] }
fabro-api = { path = "../../foundation/fabro-api" }
fabro-environment = { path = "../fabro-environment" }
fabro-sandbox = { path = "../fabro-sandbox", features = ["test-support"] }
sandbox-driver.workspace = true
fabro-mcp = { path = "../fabro-mcp" }
tokio = { workspace = true, features = ["test-util", "macros"] }
object_store.workspace = true

4
lib/components/fabro-workflow/src/event.rs
View file

@ -3,6 +3,7 @@ mod emitter;
mod events;
mod names;
mod redaction;
mod sandbox_bridge;
mod sink;
mod stored_fields;
#[cfg(test)]
@ -12,11 +13,12 @@ pub use fabro_types::{EventBody, RunNoticeCode, RunNoticeLevel};
pub use self::convert::{to_run_event, to_run_event_at};
pub use self::emitter::Emitter;
pub use self::events::Event;
pub use self::events::{Event, SandboxLifecycle};
pub use self::names::event_name;
pub use self::redaction::{
build_redacted_event_payload, event_payload_from_redacted_json, redacted_event_json,
};
pub use self::sandbox_bridge::SandboxEventBridge;
pub use self::sink::{
RunEventLogger, RunEventPersistenceError, RunEventSink, StoreProgressLogger, append_event,
append_event_if, append_event_to_sink, create_run,

93
lib/components/fabro-workflow/src/event/convert.rs
View file

@ -2,12 +2,12 @@ use ::fabro_types::{
EventBody, RunControlAction, RunEvent, RunId, StageOutcome, run_event as fabro_types,
};
use chrono::Utc;
use fabro_agent::{AgentEvent, SandboxEvent, SkillActivationSource};
use fabro_agent::{AgentEvent, SkillActivationSource};
use fabro_model::UsdMicros;
use uuid::Uuid;
use super::Event;
use super::stored_fields::stored_event_fields;
use super::{Event, SandboxLifecycle};
use crate::outcome::billed_token_counts_from_llm;
use crate::stage_scope::StageScope;
@ -974,27 +974,23 @@ fn event_body_from_event(event: &Event) -> EventBody {
duration_ms: *duration_ms,
}),
Event::Sandbox { event } => match event {
SandboxEvent::Initializing { provider } => {
SandboxLifecycle::Initializing { provider } => {
EventBody::SandboxInitializing(fabro_types::SandboxInitializingProps {
provider: provider.clone(),
})
}
SandboxEvent::Ready {
SandboxLifecycle::Ready {
provider,
duration_ms,
name,
cpu,
memory,
url,
} => EventBody::SandboxReady(fabro_types::SandboxReadyProps {
provider: provider.clone(),
duration_ms: *duration_ms,
name: name.clone(),
cpu: *cpu,
memory: *memory,
url: url.clone(),
}),
SandboxEvent::InitializeFailed {
SandboxLifecycle::InitializeFailed {
provider,
error,
causes,
@ -1005,40 +1001,19 @@ fn event_body_from_event(event: &Event) -> EventBody {
causes: causes.clone(),
duration_ms: *duration_ms,
}),
SandboxEvent::CleanupStarted { provider } => {
EventBody::SandboxCleanupStarted(fabro_types::SandboxCleanupStartedProps {
provider: provider.clone(),
})
}
SandboxEvent::CleanupCompleted {
provider,
duration_ms,
} => EventBody::SandboxCleanupCompleted(fabro_types::SandboxCleanupCompletedProps {
provider: provider.clone(),
duration_ms: *duration_ms,
}),
SandboxEvent::CleanupFailed {
provider,
error,
causes,
} => EventBody::SandboxCleanupFailed(fabro_types::SandboxCleanupFailedProps {
provider: provider.clone(),
error: error.clone(),
causes: causes.clone(),
}),
SandboxEvent::StartStarted { provider } => {
SandboxLifecycle::StartStarted { provider } => {
EventBody::SandboxStartStarted(fabro_types::SandboxStartStartedProps {
provider: provider.clone(),
})
}
SandboxEvent::StartCompleted {
SandboxLifecycle::StartCompleted {
provider,
duration_ms,
} => EventBody::SandboxStartCompleted(fabro_types::SandboxStartCompletedProps {
provider: provider.clone(),
duration_ms: *duration_ms,
}),
SandboxEvent::StartFailed {
SandboxLifecycle::StartFailed {
provider,
error,
causes,
@ -1047,19 +1022,19 @@ fn event_body_from_event(event: &Event) -> EventBody {
error: error.clone(),
causes: causes.clone(),
}),
SandboxEvent::StopStarted { provider } => {
SandboxLifecycle::StopStarted { provider } => {
EventBody::SandboxStopStarted(fabro_types::SandboxStopStartedProps {
provider: provider.clone(),
})
}
SandboxEvent::StopCompleted {
SandboxLifecycle::StopCompleted {
provider,
duration_ms,
} => EventBody::SandboxStopCompleted(fabro_types::SandboxStopCompletedProps {
provider: provider.clone(),
duration_ms: *duration_ms,
}),
SandboxEvent::StopFailed {
SandboxLifecycle::StopFailed {
provider,
error,
causes,
@ -1068,19 +1043,19 @@ fn event_body_from_event(event: &Event) -> EventBody {
error: error.clone(),
causes: causes.clone(),
}),
SandboxEvent::DeleteStarted { provider } => {
SandboxLifecycle::DeleteStarted { provider } => {
EventBody::SandboxDeleteStarted(fabro_types::SandboxDeleteStartedProps {
provider: provider.clone(),
})
}
SandboxEvent::DeleteCompleted {
SandboxLifecycle::DeleteCompleted {
provider,
duration_ms,
} => EventBody::SandboxDeleteCompleted(fabro_types::SandboxDeleteCompletedProps {
provider: provider.clone(),
duration_ms: *duration_ms,
}),
SandboxEvent::DeleteFailed {
SandboxLifecycle::DeleteFailed {
provider,
error,
causes,
@ -1089,19 +1064,19 @@ fn event_body_from_event(event: &Event) -> EventBody {
error: error.clone(),
causes: causes.clone(),
}),
SandboxEvent::SnapshotPulling { name } => {
SandboxLifecycle::SnapshotPulling { name } => {
EventBody::SnapshotPulling(fabro_types::SnapshotNameProps { name: name.clone() })
}
SandboxEvent::SnapshotCreating { name } => {
SandboxLifecycle::SnapshotCreating { name } => {
EventBody::SnapshotCreating(fabro_types::SnapshotNameProps { name: name.clone() })
}
SandboxEvent::SnapshotReady { name, duration_ms } => {
SandboxLifecycle::SnapshotReady { name, duration_ms } => {
EventBody::SnapshotReady(fabro_types::SnapshotCompletedProps {
name: name.clone(),
duration_ms: *duration_ms,
})
}
SandboxEvent::SnapshotFailed {
SandboxLifecycle::SnapshotFailed {
name,
error,
causes,
@ -1110,25 +1085,6 @@ fn event_body_from_event(event: &Event) -> EventBody {
error: error.clone(),
causes: causes.clone(),
}),
SandboxEvent::GitCloneStarted { url, branch } => {
EventBody::GitCloneStarted(fabro_types::GitCloneStartedProps {
url: url.clone(),
branch: branch.clone(),
})
}
SandboxEvent::GitCloneCompleted { url, duration_ms } => {
EventBody::GitCloneCompleted(fabro_types::GitCloneCompletedProps {
url: url.clone(),
duration_ms: *duration_ms,
})
}
SandboxEvent::GitCloneFailed { url, error, causes } => {
EventBody::GitCloneFailed(fabro_types::GitCloneFailedProps {
url: url.clone(),
error: error.clone(),
causes: causes.clone(),
})
}
},
Event::SandboxInitialized {
working_directory,
@ -1469,8 +1425,7 @@ mod tests {
};
use chrono::Utc;
use fabro_agent::{
AgentEvent, McpToolSummary, MemoryFileSummary, SandboxEvent, SkillActivationSource,
SkillSummary,
AgentEvent, McpToolSummary, MemoryFileSummary, SkillActivationSource, SkillSummary,
};
use fabro_llm::types::TokenCounts as LlmTokenCounts;
use fabro_model::{ModelRef, ProviderId};
@ -1707,12 +1662,10 @@ mod tests {
#[test]
fn run_event_sandbox_event_keeps_properties_nested() {
let stored = to_run_event(&fixtures::RUN_5, &Event::Sandbox {
event: SandboxEvent::Ready {
event: SandboxLifecycle::Ready {
provider: "daytona".to_string(),
duration_ms: 2500,
name: Some("sandbox-1".to_string()),
cpu: Some(4.0),
memory: Some(8.0),
url: Some("https://example.test".to_string()),
},
});
@ -1727,13 +1680,13 @@ mod tests {
#[test]
fn run_event_sandbox_stop_and_delete_use_distinct_event_names() {
let stopped = to_run_event(&fixtures::RUN_5, &Event::Sandbox {
event: SandboxEvent::StopCompleted {
event: SandboxLifecycle::StopCompleted {
provider: "docker".to_string(),
duration_ms: 10,
},
});
let deleted = to_run_event(&fixtures::RUN_5, &Event::Sandbox {
event: SandboxEvent::DeleteCompleted {
event: SandboxLifecycle::DeleteCompleted {
provider: "docker".to_string(),
duration_ms: 20,
},
@ -1746,7 +1699,7 @@ mod tests {
#[test]
fn run_event_sandbox_failure_serializes_causes() {
let stored = to_run_event(&fixtures::RUN_5, &Event::Sandbox {
event: SandboxEvent::InitializeFailed {
event: SandboxLifecycle::InitializeFailed {
provider: "docker".to_string(),
error: "Failed to pull Docker image buildpack-deps:noble".to_string(),
causes: vec!["connection refused".to_string()],

185
lib/components/fabro-workflow/src/event/events.rs
View file

@ -9,7 +9,7 @@ use ::fabro_types::{
RunTiming, SandboxProviderKind, StageId, StageOutcome, StageTiming, SuccessReason,
WorkflowVersionId, run_event as fabro_types,
};
use fabro_agent::{AgentEvent, SandboxEvent};
use fabro_agent::AgentEvent;
use fabro_model::{ReasoningEffort, Speed};
use serde::{Deserialize, Serialize};
@ -517,9 +517,10 @@ pub enum Event {
status: String,
duration_ms: u64,
},
/// Forwarded from a sandbox lifecycle operation.
/// A fact about the run's sandbox: the pipeline bringing it up, or a
/// driver operation on it.
Sandbox {
event: SandboxEvent,
event: SandboxLifecycle,
},
/// Emitted after the sandbox has been initialized (by engine lifecycle).
SandboxInitialized {
@ -762,6 +763,181 @@ pub enum Event {
},
}
/// The lifecycle of a run's sandbox as workflow events.
///
/// Initializing, ready, and failed are the pipeline's view of bringing the
/// sandbox up — create, activate, and prepare the workspace as one step.
/// The rest are the sandbox driver's own operations and snapshot work,
/// translated from its events by [`super::SandboxEventBridge`].
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum SandboxLifecycle {
Initializing {
provider: String,
},
Ready {
provider: String,
duration_ms: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
name: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
url: Option<String>,
},
InitializeFailed {
provider: String,
error: String,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
causes: Vec<String>,
duration_ms: u64,
},
StartStarted {
provider: String,
},
StartCompleted {
provider: String,
duration_ms: u64,
},
StartFailed {
provider: String,
error: String,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
causes: Vec<String>,
},
StopStarted {
provider: String,
},
StopCompleted {
provider: String,
duration_ms: u64,
},
StopFailed {
provider: String,
error: String,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
causes: Vec<String>,
},
DeleteStarted {
provider: String,
},
DeleteCompleted {
provider: String,
duration_ms: u64,
},
DeleteFailed {
provider: String,
error: String,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
causes: Vec<String>,
},
/// The provider is pulling the image the sandbox is created from.
SnapshotPulling {
name: String,
},
/// The provider is building or activating the snapshot.
SnapshotCreating {
name: String,
},
SnapshotReady {
name: String,
duration_ms: u64,
},
SnapshotFailed {
name: String,
error: String,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
causes: Vec<String>,
},
}
impl SandboxLifecycle {
pub fn trace(&self) {
use tracing::{debug, error, info, warn};
match self {
Self::Initializing { provider } => {
debug!(provider, "Sandbox initializing");
}
Self::Ready {
provider,
duration_ms,
..
} => {
info!(provider, duration_ms, "Sandbox ready");
}
Self::InitializeFailed {
provider,
error,
causes,
duration_ms,
} => {
error!(provider, error, causes = ?causes, duration_ms, "Sandbox init failed");
}
Self::StartStarted { provider } => {
info!(provider, "Sandbox start started");
}
Self::StartCompleted {
provider,
duration_ms,
} => {
info!(provider, duration_ms, "Sandbox start completed");
}
Self::StartFailed {
provider,
error,
causes,
} => {
warn!(provider, error, causes = ?causes, "Sandbox start failed");
}
Self::StopStarted { provider } => {
info!(provider, "Sandbox stop started");
}
Self::StopCompleted {
provider,
duration_ms,
} => {
info!(provider, duration_ms, "Sandbox stop completed");
}
Self::StopFailed {
provider,
error,
causes,
} => {
warn!(provider, error, causes = ?causes, "Sandbox stop failed");
}
Self::DeleteStarted { provider } => {
info!(provider, "Sandbox delete started");
}
Self::DeleteCompleted {
provider,
duration_ms,
} => {
info!(provider, duration_ms, "Sandbox delete completed");
}
Self::DeleteFailed {
provider,
error,
causes,
} => {
warn!(provider, error, causes = ?causes, "Sandbox delete failed");
}
Self::SnapshotPulling { name } => {
debug!(name, "Snapshot pulling");
}
Self::SnapshotCreating { name } => {
debug!(name, "Snapshot creating");
}
Self::SnapshotReady { name, duration_ms } => {
info!(name, duration_ms, "Snapshot ready");
}
Self::SnapshotFailed {
name,
error,
causes,
} => {
error!(name, error, causes = ?causes, "Snapshot failed");
}
}
}
}
impl Event {
#[must_use]
pub fn workflow_run_failed_from_error(
@ -1311,7 +1487,8 @@ impl Event {
} => {
debug!(node_id, model, provider, "Prompt completed");
}
Self::Agent { .. } | Self::Sandbox { .. } => {}
Self::Agent { .. } => {}
Self::Sandbox { event } => event.trace(),
Self::SandboxInitialized {
working_directory,
provider,

42
lib/components/fabro-workflow/src/event/names.rs
View file

@ -1,6 +1,6 @@
use fabro_agent::{AgentEvent, SandboxEvent};
use fabro_agent::AgentEvent;
use super::Event;
use super::{Event, SandboxLifecycle};
#[must_use]
pub fn event_name(event: &Event) -> &'static str {
@ -102,28 +102,22 @@ pub fn event_name(event: &Event) -> &'static str {
Event::SubgraphStarted { .. } => "subgraph.started",
Event::SubgraphCompleted { .. } => "subgraph.completed",
Event::Sandbox { event } => match event {
SandboxEvent::Initializing { .. } => "sandbox.initializing",
SandboxEvent::Ready { .. } => "sandbox.ready",
SandboxEvent::InitializeFailed { .. } => "sandbox.failed",
SandboxEvent::CleanupStarted { .. } => "sandbox.cleanup.started",
SandboxEvent::CleanupCompleted { .. } => "sandbox.cleanup.completed",
SandboxEvent::CleanupFailed { .. } => "sandbox.cleanup.failed",
SandboxEvent::StartStarted { .. } => "sandbox.start.started",
SandboxEvent::StartCompleted { .. } => "sandbox.start.completed",
SandboxEvent::StartFailed { .. } => "sandbox.start.failed",
SandboxEvent::StopStarted { .. } => "sandbox.stop.started",
SandboxEvent::StopCompleted { .. } => "sandbox.stop.completed",
SandboxEvent::StopFailed { .. } => "sandbox.stop.failed",
SandboxEvent::DeleteStarted { .. } => "sandbox.delete.started",
SandboxEvent::DeleteCompleted { .. } => "sandbox.delete.completed",
SandboxEvent::DeleteFailed { .. } => "sandbox.delete.failed",
SandboxEvent::SnapshotPulling { .. } => "sandbox.snapshot.pulling",
SandboxEvent::SnapshotCreating { .. } => "sandbox.snapshot.creating",
SandboxEvent::SnapshotReady { .. } => "sandbox.snapshot.ready",
SandboxEvent::SnapshotFailed { .. } => "sandbox.snapshot.failed",
SandboxEvent::GitCloneStarted { .. } => "sandbox.git.started",
SandboxEvent::GitCloneCompleted { .. } => "sandbox.git.completed",
SandboxEvent::GitCloneFailed { .. } => "sandbox.git.failed",
SandboxLifecycle::Initializing { .. } => "sandbox.initializing",
SandboxLifecycle::Ready { .. } => "sandbox.ready",
SandboxLifecycle::InitializeFailed { .. } => "sandbox.failed",
SandboxLifecycle::StartStarted { .. } => "sandbox.start.started",
SandboxLifecycle::StartCompleted { .. } => "sandbox.start.completed",
SandboxLifecycle::StartFailed { .. } => "sandbox.start.failed",
SandboxLifecycle::StopStarted { .. } => "sandbox.stop.started",
SandboxLifecycle::StopCompleted { .. } => "sandbox.stop.completed",
SandboxLifecycle::StopFailed { .. } => "sandbox.stop.failed",
SandboxLifecycle::DeleteStarted { .. } => "sandbox.delete.started",
SandboxLifecycle::DeleteCompleted { .. } => "sandbox.delete.completed",
SandboxLifecycle::DeleteFailed { .. } => "sandbox.delete.failed",
SandboxLifecycle::SnapshotPulling { .. } => "sandbox.snapshot.pulling",
SandboxLifecycle::SnapshotCreating { .. } => "sandbox.snapshot.creating",
SandboxLifecycle::SnapshotReady { .. } => "sandbox.snapshot.ready",
SandboxLifecycle::SnapshotFailed { .. } => "sandbox.snapshot.failed",
},
Event::SandboxInitialized { .. } => "sandbox.initialized",
Event::SetupStarted { .. } => "setup.started",

192
lib/components/fabro-workflow/src/event/sandbox_bridge.rs Normal file
View file

@ -0,0 +1,192 @@
//! The sandbox driver's events for a run's sandbox, as workflow events.
//!
//! A run's sandbox is created or attached with a driver [`EventContext`]
//! whose observer is a [`SandboxEventBridge`]. The driver reports every
//! operation it performs — start, stop, delete, the image pull inside a
//! create, snapshot builds — and the bridge turns the ones fabro records
//! on a run into [`SandboxLifecycle`] events. Everything else the driver
//! reports (state observations, notices, other operations) is not a run
//! event and is dropped here.
use std::collections::HashMap;
use std::sync::{Arc, Mutex, PoisonError};
use std::time::{Duration, Instant};
use async_trait::async_trait;
use sandbox_driver::{
Action, ErrorReport, Event as DriverEvent, EventBody as DriverEventBody, EventObserver,
EventSubject, OperationId, ProgressCode,
};
use super::{Emitter, Event, SandboxLifecycle};
/// Emits the workflow's sandbox lifecycle events from the driver's.
pub struct SandboxEventBridge {
emitter: Arc<Emitter>,
/// Fabro's name for the provider, which is what the run records; the
/// driver's own kind name can differ (`host` for a `local` run).
provider: String,
/// The image the sandbox is created from, named on pull events.
image: Option<String>,
/// Creates that pulled an image, by operation, with when the pull began.
pulls: Mutex<HashMap<OperationId, Instant>>,
}
impl SandboxEventBridge {
pub fn new(emitter: Arc<Emitter>, provider: impl Into<String>, image: Option<String>) -> Self {
Self {
emitter,
provider: provider.into(),
image,
pulls: Mutex::new(HashMap::new()),
}
}
/// The lifecycle event a driver event stands for, if fabro records one.
fn translate(&self, event: &DriverEvent) -> Option<SandboxLifecycle> {
match &event.subject {
EventSubject::Sandbox { .. } => self.translate_sandbox(event),
EventSubject::Snapshot { id, name } => {
let name = name
.clone()
.or_else(|| id.as_ref().map(ToString::to_string))
.unwrap_or_default();
match &event.body {
DriverEventBody::OperationStarted { .. } => {
Some(SandboxLifecycle::SnapshotCreating { name })
}
DriverEventBody::OperationCompleted { duration, .. } => {
Some(SandboxLifecycle::SnapshotReady {
name,
duration_ms: duration_ms(*duration),
})
}
DriverEventBody::OperationFailed { error, .. } => {
Some(SandboxLifecycle::SnapshotFailed {
name,
error: error.message.clone(),
causes: error.causes.clone(),
})
}
_ => None,
}
}
_ => None,
}
}
fn translate_sandbox(&self, event: &DriverEvent) -> Option<SandboxLifecycle> {
let provider = self.provider.clone();
match &event.body {
DriverEventBody::OperationStarted { action } => match action {
Action::Start => Some(SandboxLifecycle::StartStarted { provider }),
Action::Stop => Some(SandboxLifecycle::StopStarted { provider }),
Action::Delete => Some(SandboxLifecycle::DeleteStarted { provider }),
_ => None,
},
DriverEventBody::OperationProgress { action, progress } => {
if *action != Action::Create || progress.code.as_str() != ProgressCode::IMAGE_PULL {
return None;
}
// The first pull report of a create opens the pull; later
// ones are the same pull's progress.
let operation_id = event.operation_id.clone()?;
let mut pulls = self.pulls.lock().unwrap_or_else(PoisonError::into_inner);
if pulls.contains_key(&operation_id) {
return None;
}
pulls.insert(operation_id, Instant::now());
Some(SandboxLifecycle::SnapshotPulling {
name: self
.image
.clone()
.or_else(|| progress.message.clone())
.unwrap_or_default(),
})
}
DriverEventBody::OperationCompleted { action, duration } => match action {
Action::Create => {
let pulled = self.take_pull(event.operation_id.as_ref())?;
Some(SandboxLifecycle::SnapshotReady {
name: self.image.clone().unwrap_or_default(),
duration_ms: duration_ms(pulled.elapsed()),
})
}
Action::Start => Some(SandboxLifecycle::StartCompleted {
provider,
duration_ms: duration_ms(*duration),
}),
Action::Stop => Some(SandboxLifecycle::StopCompleted {
provider,
duration_ms: duration_ms(*duration),
}),
Action::Delete => Some(SandboxLifecycle::DeleteCompleted {
provider,
duration_ms: duration_ms(*duration),
}),
_ => None,
},
DriverEventBody::OperationFailed { action, error, .. } => match action {
Action::Create => {
self.take_pull(event.operation_id.as_ref())?;
Some(SandboxLifecycle::SnapshotFailed {
name: self.image.clone().unwrap_or_default(),
error: error.message.clone(),
causes: error.causes.clone(),
})
}
Action::Start => Some(failed(error, |error, causes| {
SandboxLifecycle::StartFailed {
provider,
error,
causes,
}
})),
Action::Stop => Some(failed(error, |error, causes| {
SandboxLifecycle::StopFailed {
provider,
error,
causes,
}
})),
Action::Delete => Some(failed(error, |error, causes| {
SandboxLifecycle::DeleteFailed {
provider,
error,
causes,
}
})),
_ => None,
},
_ => None,
}
}
/// When the create `operation_id` began pulling its image, if it did.
fn take_pull(&self, operation_id: Option<&OperationId>) -> Option<Instant> {
self.pulls
.lock()
.unwrap_or_else(PoisonError::into_inner)
.remove(operation_id?)
}
}
fn failed(
error: &ErrorReport,
build: impl FnOnce(String, Vec<String>) -> SandboxLifecycle,
) -> SandboxLifecycle {
build(error.message.clone(), error.causes.clone())
}
fn duration_ms(duration: Duration) -> u64 {
u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
}
#[async_trait]
impl EventObserver for SandboxEventBridge {
async fn observe(&self, event: DriverEvent) {
if let Some(lifecycle) = self.translate(&event) {
self.emitter.emit(&Event::Sandbox { event: lifecycle });
}
}
}

61
lib/components/fabro-workflow/src/pipeline/initialize.rs
View file

@ -12,18 +12,20 @@ use fabro_graphviz::graph;
use fabro_hooks::{HookContext, HookDecision, HookEvent, HookExecutionContext, HookRunner};
use fabro_model::Catalog;
use fabro_sandbox::{
DaytonaCredentials, GitSetupIntent, ProviderAccess, SandboxEventCallback, SandboxSpec,
reconnect_for_run_with_callback, shell_quote,
DaytonaCredentials, GitSetupIntent, ProviderAccess, SandboxSpec, reconnect_for_run_with_events,
shell_quote,
};
use fabro_static::EnvVars;
use fabro_types::RunSandboxKind;
use fabro_util::time::elapsed_ms;
use fabro_vault::Vault;
use sandbox_driver::{CorrelationId, EventContext};
use tokio::runtime::Handle;
use tokio::sync::RwLock as AsyncRwLock;
use super::types::{InitOptions, Initialized, LlmSpec, Persisted, SandboxEnvSpec};
use crate::error::Error;
use crate::event::{Event, RunNoticeCode, RunNoticeLevel};
use crate::event::{Event, RunNoticeCode, RunNoticeLevel, SandboxEventBridge, SandboxLifecycle};
use crate::git::GitAuthor;
use crate::git_bridge;
use crate::handler::llm::{AgentAcpBackend, AgentApiBackend, BackendRouter, routing};
@ -396,12 +398,15 @@ pub async fn initialize(
);
}
let sandbox_event_callback: SandboxEventCallback = {
let emitter = Arc::clone(&options.emitter);
Arc::new(move |event| {
emitter.emit(&Event::Sandbox { event });
})
};
// The driver reports what it does to the run's sandbox; the bridge
// records the operations fabro keeps as run events.
let provider_name = options.sandbox.provider_name();
let sandbox_events = EventContext::new(Arc::new(SandboxEventBridge::new(
Arc::clone(&options.emitter),
provider_name.clone(),
options.sandbox.image(),
)))
.correlation_id(CorrelationId::new(options.run_options.run_id.to_string()));
let attach_instance = if is_resume {
let record = options
.run_store
@ -443,11 +448,11 @@ pub async fn initialize(
DaytonaCredentials::from_api_key(api_key.to_string(), process_env_var)
}),
};
let sandbox = reconnect_for_run_with_callback(
let sandbox = reconnect_for_run_with_events(
&instance,
&access,
Some(options.run_options.run_id),
Some(Arc::clone(&sandbox_event_callback)),
Some(sandbox_events.clone()),
)
.await
.map_err(|err| Error::engine_with_anyhow("Failed to reconnect sandbox for resume", err))?;
@ -455,7 +460,7 @@ pub async fn initialize(
} else {
options
.sandbox
.build(Some(Arc::clone(&sandbox_event_callback)))
.build(Some(sandbox_events.clone()))
.await
.map_err(|e| Error::engine_with_anyhow("Failed to build sandbox", e))?
};
@ -477,10 +482,34 @@ pub async fn initialize(
.await
.map_err(|e| Error::engine_with_source("Failed to start sandbox", e))?;
} else {
sandbox
.initialize()
.await
.map_err(|e| Error::engine_with_source("Failed to initialize sandbox", e))?;
options.emitter.emit(&Event::Sandbox {
event: SandboxLifecycle::Initializing {
provider: provider_name.clone(),
},
});
let started = Instant::now();
if let Err(error) = sandbox.initialize().await {
options.emitter.emit(&Event::Sandbox {
event: SandboxLifecycle::InitializeFailed {
provider: provider_name.clone(),
error: error.to_string(),
causes: error.causes(),
duration_ms: elapsed_ms(started),
},
});
return Err(Error::engine_with_source(
"Failed to initialize sandbox",
error,
));
}
options.emitter.emit(&Event::Sandbox {
event: SandboxLifecycle::Ready {
provider: provider_name.clone(),
duration_ms: elapsed_ms(started),
name: Some(sandbox.sandbox_info()).filter(|name| !name.is_empty()),
url: sandbox.console_url().await,
},
});
}
let locations = RunLocations::for_sandbox(host_source_dir, sandbox.as_ref(), run_dir.clone());

44
lib/foundation/fabro-types/src/run_event/infra.rs
View file

@ -195,34 +195,11 @@ pub struct SandboxReadyProps {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub name: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cpu: Option<f64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub memory: Option<f64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub url: Option<String>,
}
pub type SandboxFailedProps = RunSandboxFailure;
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct SandboxCleanupStartedProps {
pub provider: String,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct SandboxCleanupCompletedProps {
pub provider: String,
pub duration_ms: u64,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct SandboxCleanupFailedProps {
pub provider: String,
pub error: String,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub causes: Vec<String>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct SandboxStartStartedProps {
pub provider: String,
@ -299,27 +276,6 @@ pub struct SnapshotFailedProps {
pub causes: Vec<String>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct GitCloneStartedProps {
pub url: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub branch: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct GitCloneCompletedProps {
pub url: String,
pub duration_ms: u64,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct GitCloneFailedProps {
pub url: String,
pub error: String,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub causes: Vec<String>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct SandboxInitializedProps {
pub working_directory: String,

18
lib/foundation/fabro-types/src/run_event/mod.rs
View file

@ -284,12 +284,6 @@ pub enum EventBody {
SandboxReady(SandboxReadyProps),
#[serde(rename = "sandbox.failed")]
SandboxFailed(SandboxFailedProps),
#[serde(rename = "sandbox.cleanup.started")]
SandboxCleanupStarted(SandboxCleanupStartedProps),
#[serde(rename = "sandbox.cleanup.completed")]
SandboxCleanupCompleted(SandboxCleanupCompletedProps),
#[serde(rename = "sandbox.cleanup.failed")]
SandboxCleanupFailed(SandboxCleanupFailedProps),
#[serde(rename = "sandbox.start.started")]
SandboxStartStarted(SandboxStartStartedProps),
#[serde(rename = "sandbox.start.completed")]
@ -316,12 +310,6 @@ pub enum EventBody {
SnapshotReady(SnapshotCompletedProps),
#[serde(rename = "sandbox.snapshot.failed")]
SnapshotFailed(SnapshotFailedProps),
#[serde(rename = "sandbox.git.started")]
GitCloneStarted(GitCloneStartedProps),
#[serde(rename = "sandbox.git.completed")]
GitCloneCompleted(GitCloneCompletedProps),
#[serde(rename = "sandbox.git.failed")]
GitCloneFailed(GitCloneFailedProps),
#[serde(rename = "sandbox.initialized")]
SandboxInitialized(SandboxInitializedProps),
#[serde(rename = "setup.started")]
@ -539,9 +527,6 @@ impl EventBody {
Self::SandboxInitializing(_) => "sandbox.initializing",
Self::SandboxReady(_) => "sandbox.ready",
Self::SandboxFailed(_) => "sandbox.failed",
Self::SandboxCleanupStarted(_) => "sandbox.cleanup.started",
Self::SandboxCleanupCompleted(_) => "sandbox.cleanup.completed",
Self::SandboxCleanupFailed(_) => "sandbox.cleanup.failed",
Self::SandboxStartStarted(_) => "sandbox.start.started",
Self::SandboxStartCompleted(_) => "sandbox.start.completed",
Self::SandboxStartFailed(_) => "sandbox.start.failed",
@ -555,9 +540,6 @@ impl EventBody {
Self::SnapshotCreating(_) => "sandbox.snapshot.creating",
Self::SnapshotReady(_) => "sandbox.snapshot.ready",
Self::SnapshotFailed(_) => "sandbox.snapshot.failed",
Self::GitCloneStarted(_) => "sandbox.git.started",
Self::GitCloneCompleted(_) => "sandbox.git.completed",
Self::GitCloneFailed(_) => "sandbox.git.failed",
Self::SandboxInitialized(_) => "sandbox.initialized",
Self::SetupStarted(_) => "setup.started",
Self::SetupCommandStarted(_) => "setup.command.started",