diff --git a/Cargo.lock b/Cargo.lock index c5544df46..cf215d63e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1208,7 +1208,7 @@ checksum = "d7a1e2f27636f116493b8b860f5546edb47c8d8f8ea73e1d2a20be88e28d1fea" [[package]] name = "daytona-api-client" version = "0.1.0" -source = "git+https://github.com/brynary/daytona-sdk-rust?rev=06033ca#06033caaf5d9e12918ae68396048c5c0a98822c6" +source = "git+https://github.com/brynary/daytona-sdk-rust?rev=4bbe218#4bbe2188bb617823a6eed57b8db9c5eab701f7e7" dependencies = [ "reqwest 0.12.28", "reqwest-middleware", @@ -1222,7 +1222,7 @@ dependencies = [ [[package]] name = "daytona-sdk" version = "0.1.0" -source = "git+https://github.com/brynary/daytona-sdk-rust?rev=06033ca#06033caaf5d9e12918ae68396048c5c0a98822c6" +source = "git+https://github.com/brynary/daytona-sdk-rust?rev=4bbe218#4bbe2188bb617823a6eed57b8db9c5eab701f7e7" dependencies = [ "daytona-api-client", "daytona-toolbox-client", @@ -1241,7 +1241,7 @@ dependencies = [ [[package]] name = "daytona-toolbox-client" version = "0.1.0" -source = "git+https://github.com/brynary/daytona-sdk-rust?rev=06033ca#06033caaf5d9e12918ae68396048c5c0a98822c6" +source = "git+https://github.com/brynary/daytona-sdk-rust?rev=4bbe218#4bbe2188bb617823a6eed57b8db9c5eab701f7e7" dependencies = [ "reqwest 0.12.28", "reqwest-middleware", diff --git a/Cargo.toml b/Cargo.toml index f93fef499..dc77a2354 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -68,8 +68,8 @@ twin-openai = { path = "test/twin/openai" } twin-github = { path = "test/twin/github" } tokio-tungstenite = { version = "0.26", features = ["rustls-tls-webpki-roots"] } futures-util = "0.3" -daytona-sdk = { git = "https://github.com/brynary/daytona-sdk-rust", rev = "06033ca", package = "daytona-sdk" } -daytona-api-client = { git = "https://github.com/brynary/daytona-sdk-rust", rev = "06033ca", package = "daytona-api-client" } +daytona-sdk = { git = "https://github.com/brynary/daytona-sdk-rust", rev = "4bbe218", package = "daytona-sdk" } +daytona-api-client = { git = "https://github.com/brynary/daytona-sdk-rust", rev = "4bbe218", package = "daytona-api-client" } sentry = { version = "0.35", default-features = false, features = ["backtrace", "contexts", "ureq", "rustls"] } fork = "0.2" exec = "0.3" diff --git a/lib/crates/fabro-sandbox/src/daytona/mod.rs b/lib/crates/fabro-sandbox/src/daytona/mod.rs index 8d99ed323..195eecf67 100644 --- a/lib/crates/fabro-sandbox/src/daytona/mod.rs +++ b/lib/crates/fabro-sandbox/src/daytona/mod.rs @@ -1,14 +1,17 @@ use std::collections::HashMap; use std::fmt::Write; use std::path::Path; -use std::time::Instant; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::{Duration, Instant}; use async_trait::async_trait; use daytona_sdk::api_types::SignedPortPreviewUrl; +use daytona_sdk::{DaytonaError, SessionCommandLogsResult}; use fabro_github::GitHubCredentials; use fabro_types::{CommandOutputStream, CommandTermination, RunId}; use rand::Rng; -use tokio::sync::OnceCell; +use tokio::sync::{Mutex, OnceCell}; use tokio::{fs, time}; use tokio_util::sync::CancellationToken; @@ -1158,27 +1161,232 @@ impl Sandbox for DaytonaSandbox { cancel_token: Option, output_callback: CommandOutputCallback, ) -> crate::Result { - let result = self - .exec_command(command, timeout_ms, working_dir, env_vars, cancel_token) - .await?; - if !result.stdout.is_empty() { - output_callback( + let sandbox = self.sandbox()?; + let start = Instant::now(); + let cwd = + working_dir.map_or_else(|| WORKING_DIRECTORY.to_string(), |d| self.resolve_path(d)); + + let process_svc = sandbox + .process() + .await + .map_err(|e| crate::Error::context("Failed to get process service", e))?; + + let session_id = format!("fabro-{:016x}", rand::rng().random::()); + process_svc + .create_session(&session_id) + .await + .map_err(|e| crate::Error::context("Failed to create Daytona session", e))?; + + let session_command = build_session_command(command, &cwd, env_vars); + let session_exec = match process_svc + .execute_session_command(&session_id, &session_command, true, true) + .await + { + Ok(result) => result, + Err(err) => { + if let Err(delete_err) = process_svc.delete_session(&session_id).await { + tracing::warn!( + error = %delete_err, + session_id, + "failed to delete Daytona session after command start failure" + ); + } + return Err(crate::Error::context( + "Failed to execute Daytona session command", + err, + )); + } + }; + let command_id = session_exec.cmd_id; + + let stream_process_svc = match sandbox.process().await { + Ok(process_svc) => process_svc, + Err(err) => { + if let Err(delete_err) = process_svc.delete_session(&session_id).await { + tracing::warn!( + error = %delete_err, + session_id, + "failed to delete Daytona session after stream setup failure" + ); + } + return Err(crate::Error::context("Failed to get process service", err)); + } + }; + let stdout_seen = Arc::new(Mutex::new(Vec::new())); + let stderr_seen = Arc::new(Mutex::new(Vec::new())); + let saw_live_chunk = Arc::new(AtomicBool::new(false)); + + let stream_session_id = session_id.clone(); + let stream_command_id = command_id.clone(); + let stdout_callback = output_callback.clone(); + let stderr_callback = output_callback.clone(); + let stdout_seen_for_stream = Arc::clone(&stdout_seen); + let stderr_seen_for_stream = Arc::clone(&stderr_seen); + let stdout_live = Arc::clone(&saw_live_chunk); + let stderr_live = Arc::clone(&saw_live_chunk); + let mut stream_task = tokio::spawn(async move { + stream_process_svc + .get_session_command_logs_stream( + &stream_session_id, + &stream_command_id, + move |chunk| { + let callback = stdout_callback.clone(); + let stdout_seen = Arc::clone(&stdout_seen_for_stream); + let saw_live_chunk = Arc::clone(&stdout_live); + async move { + let bytes = chunk.into_bytes(); + if !bytes.is_empty() { + saw_live_chunk.store(true, Ordering::Relaxed); + stdout_seen.lock().await.extend_from_slice(&bytes); + callback(CommandOutputStream::Stdout, bytes) + .await + .map_err(daytona_callback_error)?; + } + Ok(()) + } + }, + move |chunk| { + let callback = stderr_callback.clone(); + let stderr_seen = Arc::clone(&stderr_seen_for_stream); + let saw_live_chunk = Arc::clone(&stderr_live); + async move { + let bytes = chunk.into_bytes(); + if !bytes.is_empty() { + saw_live_chunk.store(true, Ordering::Relaxed); + stderr_seen.lock().await.extend_from_slice(&bytes); + callback(CommandOutputStream::Stderr, bytes) + .await + .map_err(daytona_callback_error)?; + } + Ok(()) + } + }, + ) + .await + }); + + let timeout_duration = Duration::from_millis(timeout_ms); + let token = cancel_token.unwrap_or_default(); + let mut exit_code = session_exec.exit_code; + let mut termination = CommandTermination::Exited; + let mut final_logs = None; + + if exit_code.is_none() { + let timeout_sleep = time::sleep(timeout_duration); + tokio::pin!(timeout_sleep); + loop { + tokio::select! { + () = time::sleep(Duration::from_millis(250)) => { + let status = match process_svc + .get_session_command(&session_id, &command_id) + .await + { + Ok(status) => status, + Err(err) => { + stream_task.abort(); + if let Err(delete_err) = process_svc.delete_session(&session_id).await { + tracing::warn!( + error = %delete_err, + session_id, + "failed to delete Daytona session after status poll failure" + ); + } + return Err(crate::Error::context("Failed to get Daytona session command status", err)); + } + }; + if let Some(code) = status.exit_code { + exit_code = Some(code); + break; + } + } + () = &mut timeout_sleep => { + termination = CommandTermination::TimedOut; + final_logs = fetch_daytona_session_logs(&process_svc, &session_id, &command_id).await; + break; + } + () = token.cancelled() => { + termination = CommandTermination::Cancelled; + final_logs = fetch_daytona_session_logs(&process_svc, &session_id, &command_id).await; + break; + } + } + } + } + + if termination != CommandTermination::Exited { + if let Err(err) = process_svc.delete_session(&session_id).await { + tracing::warn!( + error = %err, + session_id, + "failed to delete Daytona session after terminal command state" + ); + } + } + + let stream_succeeded = match finish_daytona_log_stream(&mut stream_task).await { + Ok(stream_succeeded) => stream_succeeded, + Err(err) => { + if termination == CommandTermination::Exited { + if let Err(delete_err) = process_svc.delete_session(&session_id).await { + tracing::warn!( + error = %delete_err, + session_id, + "failed to delete Daytona session after log stream failure" + ); + } + } + return Err(err); + } + }; + + if final_logs.is_none() { + final_logs = fetch_daytona_session_logs(&process_svc, &session_id, &command_id).await; + } + + if termination == CommandTermination::Exited { + if let Err(err) = process_svc.delete_session(&session_id).await { + tracing::warn!( + error = %err, + session_id, + "failed to delete Daytona session after command completion" + ); + } + } + + let mut streams_separated = stream_succeeded; + if let Some(logs) = final_logs.as_ref() { + streams_separated |= logs.streams_separated; + append_missing_log_suffix( CommandOutputStream::Stdout, - result.stdout.as_bytes().to_vec(), + logs.stdout.as_bytes(), + &stdout_seen, + &output_callback, ) .await?; - } - if !result.stderr.is_empty() { - output_callback( + append_missing_log_suffix( CommandOutputStream::Stderr, - result.stderr.as_bytes().to_vec(), + logs.stderr.as_bytes(), + &stderr_seen, + &output_callback, ) .await?; } + + let stdout = String::from_utf8_lossy(&stdout_seen.lock().await).into_owned(); + let stderr = String::from_utf8_lossy(&stderr_seen.lock().await).into_owned(); + Ok(ExecStreamingResult { - result, - streams_separated: false, - live_streaming: false, + result: ExecResult { + stdout, + stderr, + exit_code: (termination == CommandTermination::Exited) + .then_some(exit_code) + .flatten(), + termination, + duration_ms: elapsed_ms(&start), + }, + streams_separated, + live_streaming: saw_live_chunk.load(Ordering::Relaxed), }) } @@ -1284,6 +1492,122 @@ impl Sandbox for DaytonaSandbox { } } +fn daytona_callback_error(err: crate::Error) -> DaytonaError { + DaytonaError::general(format!("output callback failed: {err}")) +} + +async fn finish_daytona_log_stream( + stream_task: &mut tokio::task::JoinHandle>, +) -> crate::Result { + match time::timeout(Duration::from_secs(2), stream_task).await { + Ok(Ok(Ok(()))) => Ok(true), + Ok(Ok(Err(err))) => { + let message = err.to_string(); + if message.contains("output callback failed") { + return Err(crate::Error::context( + "Daytona log stream callback failed", + err, + )); + } + tracing::warn!(error = %message, "Daytona log stream ended with an error"); + Ok(false) + } + Ok(Err(err)) => { + tracing::warn!(error = %err, "Daytona log stream task failed"); + Ok(false) + } + Err(_) => { + stream_task.abort(); + tracing::warn!("Daytona log stream did not close after command completion"); + Ok(false) + } + } +} + +async fn fetch_daytona_session_logs( + process_svc: &daytona_sdk::ProcessService, + session_id: &str, + command_id: &str, +) -> Option { + match process_svc + .get_session_command_logs(session_id, command_id) + .await + { + Ok(logs) => Some(logs), + Err(err) => { + tracing::warn!( + error = %err, + session_id, + command_id, + "failed to fetch Daytona session command logs" + ); + None + } + } +} + +async fn append_missing_log_suffix( + stream: CommandOutputStream, + final_bytes: &[u8], + seen: &Arc>>, + output_callback: &CommandOutputCallback, +) -> crate::Result<()> { + if final_bytes.is_empty() { + return Ok(()); + } + + let mut seen = seen.lock().await; + let offset = missing_log_suffix_offset(&seen, final_bytes); + if offset >= final_bytes.len() { + return Ok(()); + } + + let missing = final_bytes[offset..].to_vec(); + seen.extend_from_slice(&missing); + drop(seen); + output_callback(stream, missing).await +} + +fn missing_log_suffix_offset(seen: &[u8], final_bytes: &[u8]) -> usize { + if final_bytes.starts_with(seen) { + return seen.len(); + } + if seen.starts_with(final_bytes) { + return final_bytes.len(); + } + + let max_overlap = seen.len().min(final_bytes.len()); + for overlap in (1..=max_overlap).rev() { + if seen[seen.len() - overlap..] == final_bytes[..overlap] { + return overlap; + } + } + 0 +} + +fn build_session_command( + command: &str, + cwd: &str, + env_vars: Option<&HashMap>, +) -> String { + let mut lines = vec![format!("cd {} || exit $?", shell_quote(cwd))]; + + if let Some(vars) = env_vars { + let mut entries: Vec<_> = vars.iter().collect(); + entries.sort_by(|(left, _), (right, _)| left.cmp(right)); + for (key, value) in entries { + lines.push(format!( + "export {}={}", + shell_quote(key), + shell_quote(value) + )); + } + } + + lines.push(command.to_string()); + lines.join("\n") +} + /// Wrap a command string with `bash -c '...'`, escaping single quotes. /// /// The Daytona API uses direct exec (not a shell), so pipes, env vars, @@ -1375,6 +1699,33 @@ mod tests { ); } + #[test] + fn build_session_command_adds_cwd_and_sorted_exports() { + let env = HashMap::from([ + ("BETA".to_string(), "two words".to_string()), + ("ALPHA".to_string(), "one".to_string()), + ]); + + let command = build_session_command("echo $ALPHA $BETA", "/tmp/with space", Some(&env)); + + assert_eq!( + command, + "cd '/tmp/with space' || exit $?\n\ + export ALPHA=one\n\ + export BETA='two words'\n\ + echo $ALPHA $BETA" + ); + } + + #[test] + fn missing_log_suffix_offset_handles_prefix_overlap() { + assert_eq!(missing_log_suffix_offset(b"hello", b"hello world"), 5); + assert_eq!(missing_log_suffix_offset(b"hello wor", b"hello world"), 9); + assert_eq!(missing_log_suffix_offset(b"abcxyz", b"xyz123"), 3); + assert_eq!(missing_log_suffix_offset(b"hello world", b"hello"), 5); + assert_eq!(missing_log_suffix_offset(b"abc", b"def"), 0); + } + #[test] fn detect_git_remote_from_repo() { let dir = tempfile::tempdir().unwrap(); diff --git a/lib/crates/fabro-sandbox/src/sandbox.rs b/lib/crates/fabro-sandbox/src/sandbox.rs index 83ec73bef..5a8558011 100644 --- a/lib/crates/fabro-sandbox/src/sandbox.rs +++ b/lib/crates/fabro-sandbox/src/sandbox.rs @@ -760,11 +760,7 @@ pub(crate) async fn fetch_source_run_ref( let fetch = sandbox .exec_command(&fetch_cmd, 30_000, None, None, None) .await?; - if !fetch.is_success() { - last_error = fetch - .into_exec_error("git fetch source run ref") - .to_string(); - } else { + if fetch.is_success() { let check = sandbox .exec_command(&check_cmd, 10_000, None, None, None) .await?; @@ -776,6 +772,10 @@ pub(crate) async fn fetch_source_run_ref( "checkpoint {checkpoint_sha} is not reachable from {remote_ref}" )) .to_string(); + } else { + last_error = fetch + .into_exec_error("git fetch source run ref") + .to_string(); } time::sleep(Duration::from_millis(500)).await; }