diff --git a/docs/internal/events-strategy.md b/docs/internal/events-strategy.md index 0d62bcaf6..b3aea81c0 100644 --- a/docs/internal/events-strategy.md +++ b/docs/internal/events-strategy.md @@ -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 diff --git a/lib/apps/fabro-cli/src/commands/run/run_progress/event.rs b/lib/apps/fabro-cli/src/commands/run/run_progress/event.rs index a8596e6f1..b80c977d6 100644 --- a/lib/apps/fabro-cli/src/commands/run/run_progress/event.rs +++ b/lib/apps/fabro-cli/src/commands/run/run_progress/event.rs @@ -53,8 +53,6 @@ pub(super) enum ProgressEvent { provider: String, duration_ms: u64, name: Option, - cpu: Option, - memory: Option, url: Option, }, SandboxFailed { @@ -255,8 +253,6 @@ pub(super) fn from_run_event(stored: &RunEvent) -> Option { 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 { 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(), diff --git a/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs b/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs index b3a00ba80..ece815fe1 100644 --- a/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs +++ b/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs @@ -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(), diff --git a/lib/apps/fabro-cli/src/commands/run/run_progress/setup_display.rs b/lib/apps/fabro-cli/src/commands/run/run_progress/setup_display.rs index ac509d948..03000bbe4 100644 --- a/lib/apps/fabro-cli/src/commands/run/run_progress/setup_display.rs +++ b/lib/apps/fabro-cli/src/commands/run/run_progress/setup_display.rs @@ -56,20 +56,10 @@ impl SetupDisplay { provider: &str, duration_ms: u64, name: Option<&str>, - cpu: Option, - memory: Option, 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 { diff --git a/lib/apps/fabro-cli/src/commands/run/run_progress/styles.rs b/lib/apps/fabro-cli/src/commands/run/run_progress/styles.rs index 44ce942df..d0f379f0f 100644 --- a/lib/apps/fabro-cli/src/commands/run/run_progress/styles.rs +++ b/lib/apps/fabro-cli/src/commands/run/run_progress/styles.rs @@ -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::>().join(" "); if single_line.len() > max { diff --git a/lib/components/fabro-agent/src/lib.rs b/lib/components/fabro-agent/src/lib.rs index 6768e57d2..fdb4786ce 100644 --- a/lib/components/fabro-agent/src/lib.rs +++ b/lib/components/fabro-agent/src/lib.rs @@ -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, diff --git a/lib/components/fabro-agent/src/sandbox.rs b/lib/components/fabro-agent/src/sandbox.rs index b4ff2317c..2f99a5e5f 100644 --- a/lib/components/fabro-agent/src/sandbox.rs +++ b/lib/components/fabro-agent/src/sandbox.rs @@ -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, }; diff --git a/lib/components/fabro-sandbox/src/daytona.rs b/lib/components/fabro-sandbox/src/daytona.rs index 0ee8cf0ee..99c6354e6 100644 --- a/lib/components/fabro-sandbox/src/daytona.rs +++ b/lib/components/fabro-sandbox/src/daytona.rs @@ -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, ) -> 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 { + async fn prepare(&self, events: Option) -> crate::Result { 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 diff --git a/lib/components/fabro-sandbox/src/driver_sandbox.rs b/lib/components/fabro-sandbox/src/driver_sandbox.rs index 4db4b78e1..db5be371e 100644 --- a/lib/components/fabro-sandbox/src/driver_sandbox.rs +++ b/lib/components/fabro-sandbox/src/driver_sandbox.rs @@ -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) -> crate::Result { + 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, + events: Option, +) -> crate::Result { 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) -> 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) -> 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, /// The provider snapshot the sandbox is created from, when the provider /// has that concept; recorded on the run. pub(crate) snapshot: Option, } /// 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; + async fn prepare(&self, events: Option) -> crate::Result; } /// 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 { + async fn prepare(&self, _events: Option) -> crate::Result { 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>, - pending: Option, - workspace: Option, - env_policy: ExplicitEnvPolicy, - event_callback: Option, + handle: OnceCell>, + pending: Option, + workspace: Option, + 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, /// `(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, + snapshot: OnceLock, } impl RunSandbox { @@ -366,7 +369,6 @@ impl RunSandbox { kind: SandboxProviderKind, provider: Arc, spec: DriverSpec, - source: Option, 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> { 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, - callback: Option, - pull_started: Mutex>, -} - -impl CreateProgress { - fn new(source: Option, callback: Option) -> 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 { + 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>> = 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>); - 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 + )), + ) + .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 = 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] diff --git a/lib/components/fabro-sandbox/src/lib.rs b/lib/components/fabro-sandbox/src/lib.rs index 1819a166e..81a5f344e 100644 --- a/lib/components/fabro-sandbox/src/lib.rs +++ b/lib/components/fabro-sandbox/src/lib.rs @@ -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 diff --git a/lib/components/fabro-sandbox/src/provider_sandbox.rs b/lib/components/fabro-sandbox/src/provider_sandbox.rs index d5e445699..65e02add6 100644 --- a/lib/components/fabro-sandbox/src/provider_sandbox.rs +++ b/lib/components/fabro-sandbox/src/provider_sandbox.rs @@ -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, run_id: Option, + events: Option, ) -> crate::Result { 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, diff --git a/lib/components/fabro-sandbox/src/reconnect.rs b/lib/components/fabro-sandbox/src/reconnect.rs index a9e6432e3..54841474b 100644 --- a/lib/components/fabro-sandbox/src/reconnect.rs +++ b/lib/components/fabro-sandbox/src/reconnect.rs @@ -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, ) -> Result { - 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, - event_callback: Option, + events: Option, ) -> Result { - 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, - event_callback: Option, + events: Option, ) -> Result { 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) } diff --git a/lib/components/fabro-sandbox/src/sandbox.rs b/lib/components/fabro-sandbox/src/sandbox.rs index 26f92aeba..2a4361401 100644 --- a/lib/components/fabro-sandbox/src/sandbox.rs +++ b/lib/components/fabro-sandbox/src/sandbox.rs @@ -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, - cpu: Option, - memory: Option, - url: Option, - }, - InitializeFailed { - provider: String, - error: String, - #[serde(default, skip_serializing_if = "Vec::is_empty")] - causes: Vec, - 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, - }, - StartStarted { - provider: String, - }, - StartCompleted { - provider: String, - duration_ms: u64, - }, - StartFailed { - provider: String, - error: String, - #[serde(default, skip_serializing_if = "Vec::is_empty")] - causes: Vec, - }, - StopStarted { - provider: String, - }, - StopCompleted { - provider: String, - duration_ms: u64, - }, - StopFailed { - provider: String, - error: String, - #[serde(default, skip_serializing_if = "Vec::is_empty")] - causes: Vec, - }, - DeleteStarted { - provider: String, - }, - DeleteCompleted { - provider: String, - duration_ms: u64, - }, - DeleteFailed { - provider: String, - error: String, - #[serde(default, skip_serializing_if = "Vec::is_empty")] - causes: Vec, - }, - - // -- 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, - }, - - // -- Daytona git -- - GitCloneStarted { - url: String, - branch: Option, - }, - GitCloneCompleted { - url: String, - duration_ms: u64, - }, - GitCloneFailed { - url: String, - error: String, - #[serde(default, skip_serializing_if = "Vec::is_empty")] - causes: Vec, - }, -} - -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; - /// 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); diff --git a/lib/components/fabro-sandbox/src/sandbox_spec.rs b/lib/components/fabro-sandbox/src/sandbox_spec.rs index f3b6c4985..0efe6611e 100644 --- a/lib/components/fabro-sandbox/src/sandbox_spec.rs +++ b/lib/components/fabro-sandbox/src/sandbox_spec.rs @@ -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 { + 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, + events: Option, ) -> Result, 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)) } diff --git a/lib/components/fabro-sandbox/src/test_support.rs b/lib/components/fabro-sandbox/src/test_support.rs index 059c6d5b7..9e532d807 100644 --- a/lib/components/fabro-sandbox/src/test_support.rs +++ b/lib/components/fabro-sandbox/src/test_support.rs @@ -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, - pub event_callback: Option, pub stdio_process: Option, pub stdio_process_error: Option, /// 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, 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, diff --git a/lib/components/fabro-workflow/Cargo.toml b/lib/components/fabro-workflow/Cargo.toml index 298d7dc7b..8a87fbece 100644 --- a/lib/components/fabro-workflow/Cargo.toml +++ b/lib/components/fabro-workflow/Cargo.toml @@ -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 diff --git a/lib/components/fabro-workflow/src/event.rs b/lib/components/fabro-workflow/src/event.rs index 8fc910726..e20385e9b 100644 --- a/lib/components/fabro-workflow/src/event.rs +++ b/lib/components/fabro-workflow/src/event.rs @@ -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, diff --git a/lib/components/fabro-workflow/src/event/convert.rs b/lib/components/fabro-workflow/src/event/convert.rs index 015d1f4ed..caa767d35 100644 --- a/lib/components/fabro-workflow/src/event/convert.rs +++ b/lib/components/fabro-workflow/src/event/convert.rs @@ -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()], diff --git a/lib/components/fabro-workflow/src/event/events.rs b/lib/components/fabro-workflow/src/event/events.rs index f23bc5c35..f8ff5094d 100644 --- a/lib/components/fabro-workflow/src/event/events.rs +++ b/lib/components/fabro-workflow/src/event/events.rs @@ -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, + #[serde(default, skip_serializing_if = "Option::is_none")] + url: Option, + }, + InitializeFailed { + provider: String, + error: String, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + causes: Vec, + 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, + }, + StopStarted { + provider: String, + }, + StopCompleted { + provider: String, + duration_ms: u64, + }, + StopFailed { + provider: String, + error: String, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + causes: Vec, + }, + DeleteStarted { + provider: String, + }, + DeleteCompleted { + provider: String, + duration_ms: u64, + }, + DeleteFailed { + provider: String, + error: String, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + causes: Vec, + }, + /// 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, + }, +} + +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, diff --git a/lib/components/fabro-workflow/src/event/names.rs b/lib/components/fabro-workflow/src/event/names.rs index 12a26d1a9..a63d4f647 100644 --- a/lib/components/fabro-workflow/src/event/names.rs +++ b/lib/components/fabro-workflow/src/event/names.rs @@ -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", diff --git a/lib/components/fabro-workflow/src/event/sandbox_bridge.rs b/lib/components/fabro-workflow/src/event/sandbox_bridge.rs new file mode 100644 index 000000000..d00446a57 --- /dev/null +++ b/lib/components/fabro-workflow/src/event/sandbox_bridge.rs @@ -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, + /// 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, + /// Creates that pulled an image, by operation, with when the pull began. + pulls: Mutex>, +} + +impl SandboxEventBridge { + pub fn new(emitter: Arc, provider: impl Into, image: Option) -> 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 { + 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 { + 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 { + self.pulls + .lock() + .unwrap_or_else(PoisonError::into_inner) + .remove(operation_id?) + } +} + +fn failed( + error: &ErrorReport, + build: impl FnOnce(String, Vec) -> 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 }); + } + } +} diff --git a/lib/components/fabro-workflow/src/pipeline/initialize.rs b/lib/components/fabro-workflow/src/pipeline/initialize.rs index 4d8c1e52a..7664def97 100644 --- a/lib/components/fabro-workflow/src/pipeline/initialize.rs +++ b/lib/components/fabro-workflow/src/pipeline/initialize.rs @@ -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()); diff --git a/lib/foundation/fabro-types/src/run_event/infra.rs b/lib/foundation/fabro-types/src/run_event/infra.rs index 2230e950f..203375b12 100644 --- a/lib/foundation/fabro-types/src/run_event/infra.rs +++ b/lib/foundation/fabro-types/src/run_event/infra.rs @@ -195,34 +195,11 @@ pub struct SandboxReadyProps { #[serde(default, skip_serializing_if = "Option::is_none")] pub name: Option, #[serde(default, skip_serializing_if = "Option::is_none")] - pub cpu: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub memory: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] pub url: Option, } 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, -} - #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct SandboxStartStartedProps { pub provider: String, @@ -299,27 +276,6 @@ pub struct SnapshotFailedProps { pub causes: Vec, } -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -pub struct GitCloneStartedProps { - pub url: String, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub branch: Option, -} - -#[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, -} - #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct SandboxInitializedProps { pub working_directory: String, diff --git a/lib/foundation/fabro-types/src/run_event/mod.rs b/lib/foundation/fabro-types/src/run_event/mod.rs index a0450dea6..dbd910b08 100644 --- a/lib/foundation/fabro-types/src/run_event/mod.rs +++ b/lib/foundation/fabro-types/src/run_event/mod.rs @@ -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",