[TOW-1299] App Isolation traits

Cargo.toml

@@ -15,6 +15,7 @@ repository = "https://github.com/tower/tower-cli"
 aes-gcm = "0.10"
 anyhow = "1.0.95"
 async-compression = { version = "0.4", features = ["tokio", "gzip"] }
+async-trait = "0.1.89"
 async_zip = { version = "0.0.16", features = ["tokio", "tokio-fs", "deflate"] }
 axum = "0.8.4"
 base64 = "0.22"

crates/config/src/session.rs

@@ -21,7 +21,9 @@ fn extract_aid_from_jwt(jwt: &str) -> Option<String> {
     let payload = parts[1];
     let decoded = URL_SAFE_NO_PAD.decode(payload).ok()?;
     let json: serde_json::Value = serde_json::from_slice(&decoded).ok()?;
-    json.get("https://tower.dev/aid")?.as_str().map(String::from)
+    json.get("https://tower.dev/aid")?
+        .as_str()
+        .map(String::from)
 }
 
 const DEFAULT_TOWER_URL: &str = "https://api.tower.dev";

crates/tower-cmd/src/run.rs

@@ -5,18 +5,24 @@ use std::collections::HashMap;
 use std::path::PathBuf;
 use tower_api::models::Run;
 use tower_package::{Package, PackageSpec};
-use tower_runtime::{local::LocalApp, App, AppLauncher, OutputReceiver, Status};
+use tower_runtime::{OutputReceiver, Status};
 use tower_telemetry::{debug, Context};
 
+use crate::{api, output, util::dates};
 use std::sync::Arc;
+use std::time::{SystemTime, UNIX_EPOCH};
 use tokio::sync::{
-    mpsc::{unbounded_channel, Receiver as MpscReceiver},
+    mpsc::Receiver as MpscReceiver,
     oneshot::{self, Receiver as OneshotReceiver},
     Mutex,
 };
 use tokio::time::{sleep, timeout, Duration};
-
-use crate::{api, output, util::dates};
+use tower_runtime::execution::ExecutionHandle;
+use tower_runtime::execution::{
+    CacheBackend, CacheConfig, CacheIsolation, ExecutionBackend, ExecutionSpec, PackageRef,
+    ResourceLimits, RuntimeConfig as ExecRuntimeConfig,
+};
+use tower_runtime::subprocess::SubprocessBackend;
 
 pub fn run_cmd() -> Command {
     Command::new("run")
@@ -148,7 +154,7 @@ where
     env_vars.insert("TOWER_URL".to_string(), config.tower_url.to_string());
 
     // There should always be a session, if there isn't one then I'm not sure how we got here?
-    let session = config.session.ok_or(Error::NoSession)?;
+    let session = config.session.as_ref().ok_or(Error::NoSession)?;
 
     env_vars.insert("TOWER_JWT".to_string(), session.token.jwt.to_string());
 
@@ -164,32 +170,34 @@ where
 
     // Build the package
     let mut package = build_package(&towerfile).await?;
-
     // Unpack the package
     package.unpack().await?;
-
-    let (sender, receiver) = unbounded_channel();
-
     output::success(&format!("Launching app `{}`", towerfile.app.name));
-    let output_task = tokio::spawn(output_handler(receiver));
-
-    let mut launcher: AppLauncher<LocalApp> = AppLauncher::default();
-    launcher
-        .launch(
-            Context::new(),
-            sender,
-            package,
-            env.to_string(),
-            secrets,
+    let backend = SubprocessBackend::new(config.cache_dir.clone());
+    let run_id = format!(
+        "cli-run-{}",
+        SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap()
+            .as_nanos()
+    );
+    let handle = backend
+        .create(build_cli_execution_spec(
+            config,
+            env,
             params,
+            secrets,
             env_vars,
-            config.cache_dir,
-        )
+            &mut package,
+            run_id,
+        ))
         .await?;
+    let receiver = handle.logs().await?;
+    let output_task = tokio::spawn(output_handler(receiver));
 
-    // Monitor app output and status concurrently
-    let app = Arc::new(Mutex::new(launcher.app.unwrap()));
-    let status_task = tokio::spawn(monitor_local_status(Arc::clone(&app)));
+    // Monitor app status concurrently
+    let handle = Arc::new(Mutex::new(handle));
+    let status_task = tokio::spawn(monitor_cli_status(Arc::clone(&handle)));
 
     // Wait for app to complete or SIGTERM
     let status_result = tokio::select! {
@@ -199,7 +207,7 @@ where
         },
         _ = tokio::signal::ctrl_c(), if !output::get_output_mode().is_mcp() => {
             output::write("\nReceived Ctrl+C, stopping local run...\n");
-            app.lock().await.terminate().await.ok();
+            handle.lock().await.terminate().await.ok();
             return Ok(output_task.await.unwrap());
         }
     };
@@ -222,6 +230,57 @@ where
     Ok(final_result)
 }
 
+fn build_cli_execution_spec(
+    config: Config,
+    env: &str,
+    params: HashMap<String, String>,
+    secrets: HashMap<String, String>,
+    env_vars: HashMap<String, String>,
+    package: &mut Package,
+    run_id: String,
+) -> ExecutionSpec {
+    let spec = ExecutionSpec {
+        id: run_id,
+        package: PackageRef::Local {
+            path: package
+                .unpacked_path
+                .clone()
+                .expect("Package must be unpacked before execution"),
+        },
+        runtime: ExecRuntimeConfig {
+            image: "local".to_string(),
+            version: None,
+            cache: CacheConfig {
+                enable_bundle_cache: true,
+                enable_runtime_cache: true,
+                enable_dependency_cache: true,
+                backend: match config.cache_dir.clone() {
+                    Some(dir) => CacheBackend::Local { cache_dir: dir },
+                    None => CacheBackend::None,
+                },
+                isolation: CacheIsolation::None,
+            },
+            entrypoint: None,
+            command: None,
+        },
+        environment: env.to_string(),
+        secrets,
+        parameters: params,
+        env_vars,
+        resources: ResourceLimits {
+            cpu_millicores: None,
+            memory_mb: None,
+            storage_mb: None,
+            max_pids: None,
+            gpu_count: 0,
+            timeout_seconds: 3600,
+        },
+        networking: None,
+        telemetry_ctx: Context::new(),
+    };
+    spec
+}
+
 /// do_run_local is the entrypoint for running an app locally. It will load the Towerfile, build
 /// the package, and launch the app. The relevant package is cleaned up after execution is
 /// complete.
@@ -595,51 +654,52 @@ async fn monitor_output(mut output: OutputReceiver) {
 
 /// monitor_local_status is a helper function that will monitor the status of a given app and waits for
 /// it to progress to a terminal state.
-async fn monitor_local_status(app: Arc<Mutex<LocalApp>>) -> Status {
-    debug!("Starting status monitoring for LocalApp");
+async fn monitor_cli_status(
+    handle: Arc<Mutex<tower_runtime::subprocess::SubprocessHandle>>,
+) -> Status {
+    use tower_runtime::execution::ExecutionHandle as _;
+
+    debug!("Starting status monitoring for CLI execution");
     let mut check_count = 0;
     let mut err_count = 0;
 
     loop {
         check_count += 1;
 
         debug!(
-            "Status check #{}, attempting to get app status",
+            "Status check #{}, attempting to get CLI handle status",
             check_count
         );
 
-        match app.lock().await.status().await {
+        match handle.lock().await.status().await {
             Ok(status) => {
                 // We reset the error count to indicate that we can intermittently get statuses.
                 err_count = 0;
 
                 match status {
                     Status::Exited => {
                         debug!("Run exited cleanly, stopping status monitoring");
-
-                        // We're done. Exit this loop and function.
                         return status;
                     }
                     Status::Crashed { .. } => {
                         debug!("Run crashed, stopping status monitoring");
-
-                        // We're done. Exit this loop and function.
                         return status;
                     }
                     _ => {
+                        debug!("Handle status: other, continuing to monitor");
                         sleep(Duration::from_millis(100)).await;
                     }
                 }
             }
             Err(e) => {
-                debug!("Failed to get app status: {:?}", e);
+                debug!("Failed to get handle status: {:?}", e);
                 err_count += 1;
 
                 // If we get five errors in a row, we abandon monitoring.
                 if err_count >= 5 {
-                    debug!("Failed to get app status after 5 attempts, giving up");
+                    debug!("Failed to get handle status after 5 attempts, giving up");
                     output::error("An error occured while monitoring your local run status!");
-                    return tower_runtime::Status::Crashed { code: -1 };
+                    return Status::Crashed { code: -1 };
                 }
 
                 // Otherwise, keep on keepin' on.

crates/tower-runtime/Cargo.toml

@@ -7,14 +7,15 @@ rust-version = { workspace = true }
 license = { workspace = true }
 
 [dependencies]
+async-trait = { workspace = true }
 chrono = { workspace = true }
 nix = { workspace = true }
-snafu = { workspace = true } 
+snafu = { workspace = true }
 tokio = { workspace = true }
 tokio-util = { workspace = true }
-tower-package = { workspace = true } 
-tower-telemetry = { workspace = true } 
-tower-uv = { workspace = true } 
+tower-package = { workspace = true }
+tower-telemetry = { workspace = true }
+tower-uv = { workspace = true }
 
 [dev-dependencies]
 config = { workspace = true }

crates/tower-runtime/src/errors.rs

@@ -64,6 +64,15 @@ pub enum Error {
 
     #[snafu(display("cancelled"))]
     Cancelled,
+
+    #[snafu(display("app not started"))]
+    AppNotStarted,
+
+    #[snafu(display("no execution handle"))]
+    NoHandle,
+
+    #[snafu(display("invalid package"))]
+    InvalidPackage,
 }
 
 impl From<std::io::Error> for Error {