From e05f7e3cd8d38802bd3a0273de512efecfe709a8 Mon Sep 17 00:00:00 2001 From: Josh Knapp Date: Sun, 15 Nov 2026 09:15:37 -0800 Subject: [PATCH] feat(aws): start, attach and stop a project on its AWS environment ProjectHost::Aws and the tunnel registry: start looks for the project's live task before preparing (a running task is attached to with the token read back from its secret, never rotated), waits for a new access point, runs, waits, tunnels and checks the agent; a task the start ran is stopped on any later failure. Stop, remove (resources; data only on request), reconcile (background re-attach) and the AWS death watch, with exits confirmed through the registry. Docker-only Reset, migration, staleness and container info refuse or skip AWS projects; an AWS project with resources can't be moved off its host. Shutdown closes every tunnel. Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 6 + app/src-tauri/src/aws/mod.rs | 5 +- app/src-tauri/src/aws/tasks.rs | 3 + app/src-tauri/src/aws/workspace.rs | 1 + app/src-tauri/src/commands/docker_commands.rs | 5 + .../src/commands/migration_commands.rs | 8 + .../src/commands/project_commands.rs | 165 +++- .../src/commands/terminal_commands.rs | 4 +- app/src-tauri/src/container_watch.rs | 17 +- app/src-tauri/src/docker/container.rs | 2 +- app/src-tauri/src/docker/workspace_env.rs | 33 + app/src-tauri/src/host/aws_host.rs | 833 ++++++++++++++++++ app/src-tauri/src/host/aws_host/tests.rs | 543 ++++++++++++ app/src-tauri/src/host/lifecycle.rs | 64 +- app/src-tauri/src/host/mod.rs | 24 +- app/src-tauri/src/host/ssm.rs | 9 + app/src-tauri/src/host/tunnels.rs | 301 +++++++ app/src-tauri/src/host/watch.rs | 10 +- app/src-tauri/src/lib.rs | 16 + app/src-tauri/src/web_terminal/ws_handler.rs | 3 +- 20 files changed, 2023 insertions(+), 29 deletions(-) create mode 100644 app/src-tauri/src/host/aws_host.rs create mode 100644 app/src-tauri/src/host/aws_host/tests.rs create mode 100644 app/src-tauri/src/host/tunnels.rs diff --git a/CLAUDE.md b/CLAUDE.md index a41b63d..649da53 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -292,6 +292,12 @@ docker exec stdout → tokio task → emit("terminal-output-{sessionId}") → li when it drops, so an `AgentEndpoint` stays valid. The `StartSession` token goes to the plugin through its environment, never argv, and the plugin's environment is stripped of every inherited `AWS_*` variable before the session's own credentials are added. + `aws_host.rs` is `ProjectHost::Aws` (an ECS Fargate task per project; a running task is + attached to with the token read back from its secret, never rotated; a task a failed start ran + is stopped again) and `tunnels.rs` the registry of open tunnels by project — an AWS project's + `ContainerTarget::for_project` is its tunnel's endpoint, its stored `container_id` is that + endpoint's URL, and the AWS death watch confirms exits through the registry. Shutdown closes + every tunnel. AWS SDK calls stay in `aws/` (`runtime.rs`, `tasks.rs`, `workspace.rs`). `agent.rs` reuses `docker/exec.rs`'s `OneshotOutput`, host-file reader and uid probe so the two variants cannot drift; an `Agent` target's token never appears in `Debug` or error text. **Container-interior code must go through `ContainerTarget`, never `docker::exec`, bollard or diff --git a/app/src-tauri/src/aws/mod.rs b/app/src-tauri/src/aws/mod.rs index 90bbfe1..90b011a 100644 --- a/app/src-tauri/src/aws/mod.rs +++ b/app/src-tauri/src/aws/mod.rs @@ -13,13 +13,10 @@ pub mod permissions; pub mod policies; #[cfg(test)] mod policy_eval; -// Used by the AWS host (phase 5, Task 6); until then only by its tests. -#[allow(dead_code)] +pub mod runtime; pub mod tasks; pub mod tofu; pub mod tofu_modules; -// Used by the AWS host (phase 5, Task 6); until then only by its tests. -#[allow(dead_code)] pub mod workspace; pub use connection::{sdk_config, sdk_config_for_keys}; diff --git a/app/src-tauri/src/aws/tasks.rs b/app/src-tauri/src/aws/tasks.rs index 5ac3dc9..41b0e89 100644 --- a/app/src-tauri/src/aws/tasks.rs +++ b/app/src-tauri/src/aws/tasks.rs @@ -337,6 +337,7 @@ pub fn quota_or(e: AwsError) -> AwsError { } } +#[allow(dead_code)] // for the UI's start errors (Task 8) pub fn is_quota_error(e: &AwsError) -> bool { matches!(e, AwsError::QuotaExceeded { .. }) } @@ -480,6 +481,7 @@ async fn stopped_error(api: &A, logs: &TaskLogs, t: &TaskInfo) -> Aw // ----------------------------------------------------------- stop / watch /// Why a task stopped, as the app reports it (spec, "Lifecycle and status"). +#[allow(dead_code)] // the stop reasons the UI shows (Task 8) #[derive(Debug, Clone, PartialEq, Eq)] pub enum StopKind { /// Exited on its own (the essential container ended, idle stop, or AWS @@ -493,6 +495,7 @@ pub enum StopKind { UserInitiated, } +#[allow(dead_code)] // Task 8 pub fn classify_stop(info: &TaskInfo) -> StopKind { match info.stop_code.as_deref() { Some("SpotInterruption") => StopKind::Spot, diff --git a/app/src-tauri/src/aws/workspace.rs b/app/src-tauri/src/aws/workspace.rs index e45a995..2da99b5 100644 --- a/app/src-tauri/src/aws/workspace.rs +++ b/app/src-tauri/src/aws/workspace.rs @@ -204,6 +204,7 @@ fn serialize_payload(m: &BTreeMap) -> Result { /// `TRIPLEC_REPOS`, `TRIPLEC_WIP_PUSH` = `1`/`0`, `TRIPLEC_IDLE_STOP_MINUTES` /// when on, `TRIPLEC_STORAGE` = `ephemeral`/`hybrid`), keys sorted. Fails /// when it would exceed 65,536 bytes. +#[cfg_attr(not(test), allow(dead_code))] pub fn project_secret_payload( env_pairs: &[(String, String)], agent_token: &str, diff --git a/app/src-tauri/src/commands/docker_commands.rs b/app/src-tauri/src/commands/docker_commands.rs index 598a42a..e0a561b 100644 --- a/app/src-tauri/src/commands/docker_commands.rs +++ b/app/src-tauri/src/commands/docker_commands.rs @@ -34,6 +34,11 @@ pub async fn get_container_info( .projects_store .get(&project_id) .ok_or_else(|| format!("Project {} not found", project_id))?; + // An AWS project has no Docker container (its own name could match a + // stale one from before it moved to AWS). + if project.is_aws() { + return Ok(None); + } docker::get_container_info(&project).await } diff --git a/app/src-tauri/src/commands/migration_commands.rs b/app/src-tauri/src/commands/migration_commands.rs index 6771e37..593ab98 100644 --- a/app/src-tauri/src/commands/migration_commands.rs +++ b/app/src-tauri/src/commands/migration_commands.rs @@ -432,6 +432,11 @@ pub async fn get_container_staleness( .projects_store .get(&project_id) .ok_or_else(|| format!("Project {} not found", project_id))?; + // An AWS task has no local container or snapshot to compare: no claim + // either way ("unknown"), and nothing is probed. + if project.is_aws() { + return Ok(ContainerStaleness::default()); + } let settings = state.settings_store.get(); let base_image = crate::models::container_config::resolve_image_name( &settings.image_source, @@ -708,6 +713,9 @@ pub async fn migrate_project_to_base( app_handle: tauri::AppHandle, state: State<'_, AppState>, ) -> Result { + if let Some(p) = state.projects_store.get(&project_id) { + crate::commands::project_commands::refuse_aws(&p, "Updating to the current base image")?; + } let _guard = match ActiveGuard::acquire(&project_id) { Ok(guard) => guard, Err(busy) => return Ok(MigrationReport::failed_preflight(&busy)), diff --git a/app/src-tauri/src/commands/project_commands.rs b/app/src-tauri/src/commands/project_commands.rs index 842c457..9264838 100644 --- a/app/src-tauri/src/commands/project_commands.rs +++ b/app/src-tauri/src/commands/project_commands.rs @@ -655,6 +655,7 @@ pub async fn add_project( #[tauri::command] pub async fn remove_project( project_id: String, + discard_data: Option, state: State<'_, AppState>, ) -> Result { // **H-2: the only writer of these three categories that held nothing.** @@ -673,6 +674,32 @@ pub async fn remove_project( let _guard = crate::project_lock::try_acquire(&project_id, crate::project_lock::ProjectOp::Destroy)?; + // An AWS project's task and resources go first, and a failure keeps the + // project: its record is the only handle on a billed task and its + // secret. Its data (the EFS access point) goes only when asked. + if let Some(project) = state.projects_store.get(&project_id).filter(|p| p.is_aws()) { + let settings = state.settings_store.get(); + if let Some(target) = ContainerTarget::for_project(&project) { + state.exec_manager.close_sessions_for_container(target.id()).await; + } + let connection_exists = project + .host_id + .as_deref() + .is_some_and(|id| settings.hosts.iter().any(|h| h.id == id)); + if connection_exists { + ProjectHost::for_project(&project) + .remove_aws_resources(&project, &settings, discard_data.unwrap_or(false)) + .await?; + } else { + log::warn!( + "Project {}: its AWS connection no longer exists, so its AWS resources can't be \ + removed from here", + project_id + ); + crate::host::tunnels::close(&project_id).await; + } + } + // Release any host loopback ports the auth bridge holds for this project // before the container (and the project record) go away. state.auth_bridge.stop(&project_id).await; @@ -1150,11 +1177,23 @@ fn validate_aws_update( ) } } - if (stored.host_id != project.host_id || stored.aws_placement != project.aws_placement) - && stored.status != ProjectStatus::Stopped - { + let moved = stored.host_id != project.host_id || stored.aws_placement != project.aws_placement; + if moved && stored.status != ProjectStatus::Stopped { return Err("Stop the project before changing its host or placement".into()); } + // The task, access point and task definition live in the stored + // connection's account and region; moving the project would orphan them. + if moved + && stored.aws.as_ref().is_some_and(|a| { + a.task_arn.is_some() || a.access_point_id.is_some() || a.task_definition_arn.is_some() + }) + { + return Err( + "This project still has AWS resources (a task, its data's access point or task \ + definitions) where it runs now; remove the workspace's AWS resources first" + .into(), + ); + } Ok(()) } @@ -1359,6 +1398,21 @@ async fn start_project_container_locked( // Update project with container info using granular methods (Issue 14: TOCTOU) state.projects_store.set_container_id(&project_id, Some(container_id.clone()))?; + // AWS: the task's handles, store-owned like the container id. The target + // id stored above is the tunnel's endpoint URL, which is what the death + // watch reports a stopped task as. + if project.is_aws() { + if let Some((task_arn, access_point_id, task_definition_arn)) = + crate::host::tunnels::runtime(&project_id) + { + state.projects_store.set_aws_runtime( + &project_id, + Some(task_arn), + access_point_id, + Some(task_definition_arn), + )?; + } + } state.projects_store.update_status(&project_id, ProjectStatus::Running)?; // Arm the auth bridge if this project opted in. Purely host-side, so it @@ -1404,7 +1458,30 @@ pub async fn stop_project_container( // Drop host listeners first: they only make sense while the container runs. state.auth_bridge.stop(&project_id).await; - if let Some(container) = ContainerTarget::for_project(&project) { + if project.is_aws() { + // A failed StopTask leaves a billed task running, so unlike Docker it + // is not papered over: the project keeps its status and the user + // sees why (expired SSO, typically), and can retry. + emit_progress(&app_handle, &project_id, "Stopping task..."); + if let Some(container) = ContainerTarget::for_project(&project) { + state.exec_manager.close_sessions_for_container(container.id()).await; + } + let settings = state.settings_store.get(); + if let Err(e) = ProjectHost::for_project(&project).stop_project(&project, &settings).await { + log::warn!("Stop failed for project {}: {}", project_id, e); + let _ = state.projects_store.update_status(&project_id, project.status.clone()); + return Err(format!("Could not stop the task: {e}")); + } + if let Some(a) = project.aws.as_ref() { + state.projects_store.set_aws_runtime( + &project_id, + None, + a.access_point_id.clone(), + a.task_definition_arn.clone(), + )?; + } + state.projects_store.set_container_id(&project_id, None)?; + } else if let Some(container) = ContainerTarget::for_project(&project) { // Close exec sessions for this project emit_progress(&app_handle, &project_id, "Stopping container..."); state.exec_manager.close_sessions_for_container(container.id()).await; @@ -1438,6 +1515,7 @@ pub async fn rebuild_project_container( .projects_store .get(&project_id) .ok_or_else(|| format!("Project {} not found", project_id))?; + refuse_aws(&project, "Reset")?; // Reset supersedes any migration decision that was still pending: the // snapshot image and both volumes are about to go, so a surviving record @@ -1513,6 +1591,20 @@ pub async fn rebuild_project_container( /// [`crate::commands::migration_commands::reconcile_migration`]. The migration /// pass runs over *every* project, not just the Running ones, because a project /// whose container was removed mid-migration reports Stopped. +/// How long reconcile waits on AWS for one project. +const AWS_RECONCILE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(15); + +/// Refuse a Docker-only operation on an AWS project. +pub(crate) fn refuse_aws(project: &Project, what: &str) -> Result<(), String> { + if project.is_aws() { + return Err(format!( + "{what} is not available for a project that runs on AWS; it applies to local \ + Docker containers only" + )); + } + Ok(()) +} + #[tauri::command] pub async fn reconcile_project_statuses( app_handle: tauri::AppHandle, @@ -1520,7 +1612,7 @@ pub async fn reconcile_project_statuses( ) -> Result, String> { let projects = state.projects_store.list(); - for project in &projects { + for project in projects.iter().filter(|p| !p.is_aws()) { crate::commands::migration_commands::reconcile_migration(project, &app_handle).await; } @@ -1554,6 +1646,28 @@ pub async fn reconcile_project_statuses( continue; } + // AWS: ECS is the authority. One DescribeTasks per project, bounded + // so a slow or unreachable AWS can't hold the app's start; a live + // task's tunnel is re-opened in the background. + if project.is_aws() { + let settings = state.settings_store.get(); + let store = state.projects_store.clone(); + if tokio::time::timeout( + AWS_RECONCILE_TIMEOUT, + crate::host::aws_host::reconcile(project, &settings, store), + ) + .await + .is_err() + { + log::warn!( + "Project {}: AWS did not answer within {}s; status left as is", + project.id, + AWS_RECONCILE_TIMEOUT.as_secs() + ); + } + continue; + } + let is_running = if let Some(container) = ContainerTarget::for_project(project) { container.is_running().await.unwrap_or(false) } else { @@ -2279,4 +2393,45 @@ mod tests { let mut edited = running.clone(); edited.aws.as_mut().unwrap().spot = true; assert!(validate_aws_update(&running, &edited, &hosts).is_ok()); } + + #[test] + fn a_project_with_aws_resources_cannot_be_moved_off_its_host() { + let hosts = vec![host("h1"), host("h2")]; + let mut stored = aws_project(); + stored.aws.as_mut().unwrap().access_point_id = Some("fsap-1".into()); + let mut local = stored.clone(); + local.host_id = None; + local.aws_placement = None; + local.aws = None; + let e = validate_aws_update(&stored, &local, &hosts).unwrap_err(); + assert!(e.contains("remove the workspace's AWS resources first"), "{e}"); + let mut other = stored.clone(); + other.host_id = Some("h2".into()); + assert!(validate_aws_update(&stored, &other, &hosts).is_err()); + let mut moved = stored.clone(); + moved.aws_placement.as_mut().unwrap().environment = "prod".into(); + assert!(validate_aws_update(&stored, &moved, &hosts).is_err()); + for set in [ + |a: &mut crate::models::AwsWorkspaceConfig| a.task_arn = Some("t".into()), + |a: &mut crate::models::AwsWorkspaceConfig| a.task_definition_arn = Some("td".into()), + ] { + let mut st = aws_project(); + set(st.aws.as_mut().unwrap()); + assert!(validate_aws_update(&st, &local, &hosts).is_err()); + } + // No handles: free to move. Other edits are fine with handles. + assert!(validate_aws_update(&aws_project(), &local, &hosts).is_ok()); + let mut edited = stored.clone(); + edited.aws.as_mut().unwrap().cpu = 2048; + edited.aws.as_mut().unwrap().memory_mib = 8192; + assert!(validate_aws_update(&stored, &edited, &hosts).is_ok()); + } + + #[test] + fn docker_only_operations_refuse_aws_projects() { + assert!(refuse_aws(&aws_project(), "Reset").unwrap_err().contains("Reset is not available")); + let mut local = aws_project(); + local.host_id = None; + assert!(refuse_aws(&local, "Reset").is_ok()); + } } diff --git a/app/src-tauri/src/commands/terminal_commands.rs b/app/src-tauri/src/commands/terminal_commands.rs index a1ddfd7..9bbeb05 100644 --- a/app/src-tauri/src/commands/terminal_commands.rs +++ b/app/src-tauri/src/commands/terminal_commands.rs @@ -160,7 +160,9 @@ pub async fn open_terminal_session( .get(&project_id) .ok_or_else(|| format!("Project {} not found", project_id))?; - let target = ContainerTarget::for_project(&project) + // Waits while an AWS project's tunnel is being re-opened after a restart. + let target = ContainerTarget::for_project_waiting(&project) + .await .ok_or_else(|| "Container not running".to_string())?; if let Some(id) = resume_session_id.as_deref() { diff --git a/app/src-tauri/src/container_watch.rs b/app/src-tauri/src/container_watch.rs index 3f3887c..aedefde 100644 --- a/app/src-tauri/src/container_watch.rs +++ b/app/src-tauri/src/container_watch.rs @@ -219,13 +219,28 @@ pub fn spawn( cancel: watch::Receiver, ) -> tauri::async_runtime::JoinHandle<()> { tauri::async_runtime::spawn(async move { - run(store, source, cancel, |p, reason| { + let store_for_aws = store.clone(); + run(store, source, cancel, move |p, reason| { log::warn!( "Project '{}' ({}): its container stopped without the app asking ({:?}) — marking Stopped", p.name, p.id, reason ); + // An AWS task that stopped: its tunnel leads nowhere now, and its + // task ARN is history (the data's access point stays). + if p.is_aws() { + if let Some(a) = p.aws.as_ref() { + let _ = store_for_aws.set_aws_runtime( + &p.id, + None, + a.access_point_id.clone(), + a.task_definition_arn.clone(), + ); + } + let id = p.id.clone(); + tauri::async_runtime::spawn(async move { crate::host::tunnels::close(&id).await }); + } let _ = app.emit(CONTAINER_EXITED_EVENT, ContainerExited { project: p, reason }); }) .await; diff --git a/app/src-tauri/src/docker/container.rs b/app/src-tauri/src/docker/container.rs index b506a23..701026c 100644 --- a/app/src-tauri/src/docker/container.rs +++ b/app/src-tauri/src/docker/container.rs @@ -402,7 +402,7 @@ pub(super) fn compute_env_fingerprint(custom_env_vars: &[EnvVar]) -> String { /// /// The returned token is never logged and never leaves this module except as /// the env var value handed to Docker. -fn shared_claude_auth(project: &Project) -> Option<(String, String)> { +pub(super) fn shared_claude_auth(project: &Project) -> Option<(String, String)> { if project.backend != Backend::Anthropic || !project.use_shared_auth_token { return None; } diff --git a/app/src-tauri/src/docker/workspace_env.rs b/app/src-tauri/src/docker/workspace_env.rs index cd8f3d9..046eb76 100644 --- a/app/src-tauri/src/docker/workspace_env.rs +++ b/app/src-tauri/src/docker/workspace_env.rs @@ -419,6 +419,39 @@ pub(crate) fn workspace_env( env } + +/// [`workspace_env`] with its inputs resolved from `settings` exactly as +/// `create_container` resolves them (the CA from disk, the shared Claude +/// token from the keychain), for a host that does not create a Docker +/// container. Also returns the resolved CA, whose files the caller ships. +pub(crate) fn workspace_env_from_settings( + project: &Project, + settings: &crate::models::AppSettings, +) -> Result { + let shared_claude = super::container::shared_claude_auth(project); + let effective_ca_path = + resolve_with_global(project.ca_cert_path.as_deref(), settings.ca_cert_path.as_deref()); + let resolved_ca = ca_certs::resolve(effective_ca_path)?; + let pairs = workspace_env( + project, + &WorkspaceEnvInputs { + global_aws: &settings.global_aws, + global_ollama: &settings.global_ollama, + global_llamacpp: &settings.global_llamacpp, + global_openai_compatible: &settings.global_openai_compatible, + global_claude_instructions: settings.global_claude_instructions.as_deref(), + global_custom_env_vars: &settings.global_custom_env_vars, + timezone: settings.timezone.as_deref(), + global_claude_code_settings: settings.global_claude_code_settings.as_ref(), + default_git_user_name: settings.default_git_user_name.as_deref(), + default_git_user_email: settings.default_git_user_email.as_deref(), + resolved_ca: resolved_ca.as_ref(), + shared_claude_token: shared_claude.as_ref().map(|(token, _)| token.as_str()), + }, + ); + Ok((pairs, resolved_ca)) +} + #[cfg(test)] mod tests { use super::super::ca_certs::ResolvedCa; diff --git a/app/src-tauri/src/host/aws_host.rs b/app/src-tauri/src/host/aws_host.rs new file mode 100644 index 0000000..c784d76 --- /dev/null +++ b/app/src-tauri/src/host/aws_host.rs @@ -0,0 +1,833 @@ +//! `ProjectHost::Aws`: a project's container is an ECS Fargate task in one +//! of the account's environments, reached through an SSM port-forward to +//! the in-task agent (`ssm.rs`, `agent.rs`). +//! +//! Start: connection → `SdkConfig` → caller identity → environment record → +//! onboarding → the project's live task, if there is one (its token is read +//! back from the project secret, P2 — **a running task's token is never +//! rotated**); otherwise `prepare_workspace` (which rotates the token) → +//! access point available → RunTask → ready → tunnel → agent health → +//! registry ([`super::tunnels`]). A task this start ran is stopped again on +//! any later failure, so nothing billed is left behind. +//! +//! The decisions are made by [`start_core`] over the `aws/` traits and a +//! [`Connector`], so fakes exercise them; AWS SDK calls stay in `aws/`. + +use std::future::Future; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::Duration; + +use aws_config::SdkConfig; +use futures_util::future::BoxFuture; + +use super::agent::{self, AgentEndpoint}; +use super::ssm::{EcsTask, PortForward, AGENT_PORT}; +use super::tunnels::{self, LiveTunnel, TunnelHandle}; +use super::watch::{AwsTaskLookup, TaskGroup, WatchedTask}; +use super::ContainerTarget; +use crate::aws::environments::EnvironmentInfo; +use crate::aws::runtime::{ + check_onboarded, cluster_arn_of_task, cluster_of_task, read_agent_token, read_environment, + remove_workspace_resources, wait_access_point, PollTiming, RemoveRequest, RuntimeApi, + SdkRuntimeApi, +}; +use crate::aws::tasks::{ + find_or_run, find_project_task, stopped_among, task_log_stream, wait_ready, SdkTaskApi, + TaskApi, TaskInfo, TaskLogs, WaitTiming, LOG_TAIL_LINES, STOP_REASON, +}; +use crate::aws::tofu_modules::{log_stream_prefix, owner_key, task_family}; +use crate::aws::workspace::{prepare_workspace, SdkWorkspaceApi, WorkspaceApi}; +use crate::aws::{caller_identity, sdk_config, CallerIdentity}; +use crate::docker::ca_certs::{container_cert_name, ResolvedCa}; +use crate::models::app_settings::ImageSource; +use crate::models::container_config::REGISTRY_IMAGE; +use crate::models::{AppSettings, Project, ProjectStatus}; +use crate::storage::projects_store::ProjectsStore; +use crate::storage::settings_store::SettingsStore; + +// Progress strings the start view shows, in order (Task 8 matches on none +// of them; they are for the user). +pub const PROGRESS_READING: &str = "Reading environment…"; +pub const PROGRESS_PREPARING: &str = "Preparing workspace…"; +pub const PROGRESS_STARTING: &str = "Starting task…"; +pub const PROGRESS_WAITING: &str = "Waiting for the task (about 2 minutes)…"; +pub const PROGRESS_CONNECTING: &str = "Connecting…"; + +/// A concurrent start won the RunTask race: the secret now holds a token +/// that task never saw. +pub const START_RACE: &str = + "Another start of this workspace is in progress; try again in a minute"; + +/// Per-file cap on what is shipped in the project secret (R1). +pub const MAX_FILE_BYTES: u64 = 64 * 1024; +/// How long the agent gets to answer through a fresh tunnel. +const HEALTH_TIMEOUT: Duration = Duration::from_secs(30); +const HEALTH_INTERVAL: Duration = Duration::from_secs(1); +/// How long a terminal waits for a background re-attach. +pub const REATTACH_WAIT: Duration = Duration::from_secs(90); + +fn s(e: E) -> String { + e.to_string() +} + +// ─────────────────────────────────────────────────────────────── inputs + +/// The image an AWS task runs (P5): registry images only. +pub fn aws_image(settings: &AppSettings) -> Result { + match settings.image_source { + ImageSource::Registry => Ok(REGISTRY_IMAGE.to_string()), + ImageSource::LocalBuild => { + log::info!("AWS projects run the registry image; the local build exists only on this machine"); + Ok(REGISTRY_IMAGE.to_string()) + } + ImageSource::Custom => { + let name = settings.custom_image_name.as_deref().unwrap_or("").trim(); + if is_registry_reference(name) { + Ok(name.to_string()) + } else { + Err(format!( + "The custom image \"{}\" is not a registry reference, so AWS can't pull it. \ + Use a name like registry.example.com/team/image:tag (Settings → Docker).", + name.escape_debug() + )) + } + } + } +} + +/// `host/path[:tag]` or `user/repo[:tag]`: something Fargate can pull. A +/// bare name (a local tag) or a loopback registry is not. +fn is_registry_reference(name: &str) -> bool { + if name.is_empty() || name.len() > 512 || name.chars().any(|c| c.is_whitespace() || c.is_control()) { + return false; + } + let Some((first, rest)) = name.split_once('/') else { + return false; + }; + if first.is_empty() || rest.is_empty() || rest.starts_with('/') { + return false; + } + let host = first.split(':').next().unwrap_or(""); + !(host == "localhost" || host.starts_with("127.") || host == "[::1]") +} + +/// [`aws_env`]'s answer: the pairs, and the resolved CA whose files ship too. +pub type AwsEnv = (Vec<(String, String)>, Option); + +/// The environment pairs an AWS task gets: [`workspace_env`] with the same +/// inputs Docker resolves, and Claude Code's sandbox forced off (R4: Fargate +/// support for bubblewrap is unverified). Returns the resolved CA too. +/// +/// [`workspace_env`]: crate::docker::workspace_env::workspace_env +pub fn aws_env(project: &Project, settings: &AppSettings) -> Result { + let mut p = project.clone(); + if p.sandbox_mode_enabled { + log::info!( + "Project {}: Claude Code's sandbox is turned off on AWS (not yet supported on Fargate)", + project.id + ); + p.sandbox_mode_enabled = false; + } + crate::docker::workspace_env::workspace_env_from_settings(&p, settings) +} + +/// One file read for the secret, refusing anything over [`MAX_FILE_BYTES`]. +fn read_capped(path: &Path, shown: &str) -> Result, String> { + let len = std::fs::metadata(path) + .map_err(|e| format!("Could not read {shown}: {e}"))? + .len(); + if len > MAX_FILE_BYTES { + return Err(format!( + "{shown} is {len} bytes; files sent to an AWS workspace are limited to {} KiB each", + MAX_FILE_BYTES / 1024 + )); + } + std::fs::read(path).map_err(|e| format!("Could not read {shown}: {e}")) +} + +/// The regular files directly in `dir`: subdirectories skipped, and a +/// symlink only when it resolves to a file inside `dir`. +fn files_in(dir: &Path) -> Result, String> { + let root = std::fs::canonicalize(dir) + .map_err(|e| format!("Could not read {}: {e}", dir.display()))?; + let mut out = Vec::new(); + let entries = + std::fs::read_dir(&root).map_err(|e| format!("Could not read {}: {e}", dir.display()))?; + for entry in entries.flatten() { + let path = entry.path(); + let Some(name) = entry.file_name().to_str().map(str::to_string) else { + log::warn!("Skipping {} (not a UTF-8 name)", path.display()); + continue; + }; + let Ok(ft) = entry.file_type() else { continue }; + let real = if ft.is_file() { + path.clone() + } else if ft.is_symlink() { + match std::fs::canonicalize(&path) { + Ok(t) if t.starts_with(&root) && t.is_file() => t, + _ => { + log::warn!("Skipping {} (a link out of the folder, or not to a file)", path.display()); + continue; + } + } + } else { + continue; + }; + out.push((name, real)); + } + out.sort(); + Ok(out) +} + +/// The files the entrypoint writes to `/tmp/.host-ssh` and `/tmp/.host-ca` +/// (R1), as `ssh/` / `ca/` pairs: the project's effective SSH +/// key folder (or file) and the resolved CA, the same sources Docker mounts. +pub fn workspace_files( + ssh_path: Option<&str>, + ca: Option<&ResolvedCa>, +) -> Result)>, String> { + let mut out = Vec::new(); + if let Some(ssh) = ssh_path.map(str::trim).filter(|p| !p.is_empty()) { + crate::commands::project_commands::validate_mounted_host_path( + "the SSH key folder", + None, + Some(ssh), + )?; + let path = Path::new(ssh); + if path.is_dir() { + for (name, real) in files_in(path)? { + let bytes = read_capped(&real, &format!("SSH file {name}"))?; + out.push((format!("ssh/{name}"), bytes)); + } + } else if path.is_file() { + let name = path + .file_name() + .and_then(|n| n.to_str()) + .ok_or_else(|| format!("The SSH key path {ssh} has no usable file name"))?; + out.push((format!("ssh/{name}"), read_capped(path, &format!("SSH file {name}"))?)); + } else { + return Err(format!("The SSH key path {ssh} does not exist")); + } + } + if let Some(ca) = ca { + crate::commands::project_commands::validate_mounted_host_path( + "the CA certificate path", + None, + Some(&ca.host_path), + )?; + if ca.is_dir { + let allowed: Vec = ca.cert_files.clone(); + for (name, real) in files_in(Path::new(&ca.host_path))? { + // Only what `collect_cert_files` picked (by extension). + let picked = allowed.iter().any(|f| { + f.file_name().and_then(|n| n.to_str()) == Some(name.as_str()) + }); + if picked { + out.push((format!("ca/{name}"), read_capped(&real, &format!("certificate {name}"))?)); + } + } + } else { + let path = Path::new(&ca.host_path); + let name = container_cert_name( + &path.file_name().map(|n| n.to_string_lossy().to_string()).unwrap_or_default(), + ); + out.push((format!("ca/{name}"), read_capped(path, &format!("certificate {name}"))?)); + } + } + Ok(out) +} + +// ──────────────────────────────────────────────────────────── connector + +/// Opens the tunnel to a ready task and checks its agent answers. +pub trait Connector: Send + Sync { + fn connect( + &self, + task: &TaskInfo, + token: &str, + ) -> impl Future, AgentEndpoint), String>> + Send; +} + +/// The real one: an SSM port-forward, then `/v1/health` through it. +pub struct SsmConnector { + pub sdk: SdkConfig, +} + +/// The SSM target of a ready task's `workspace` container. +pub fn ecs_task(task: &TaskInfo) -> Result { + let cluster = cluster_of_task(&task.arn) + .ok_or_else(|| format!("{} is not a task ARN with a cluster", task.arn))?; + let task_id = task.arn.rsplit('/').next().unwrap_or_default().to_string(); + let runtime_id = task + .runtime_id + .clone() + .ok_or_else(|| "The task's container has no runtime id yet".to_string())?; + Ok(EcsTask { cluster: cluster.to_string(), task_id, runtime_id }) +} + +/// Wait for the agent behind `ep` to answer. A refusal (wrong token) is +/// final; silence is retried until [`HEALTH_TIMEOUT`]. +async fn wait_healthy(ep: &AgentEndpoint) -> Result<(), String> { + let deadline = tokio::time::Instant::now() + HEALTH_TIMEOUT; + loop { + match agent::is_running(ep).await { + Ok(true) => return Ok(()), + Ok(false) => {} + Err(e) => return Err(format!("The workspace's agent refused the connection: {e}")), + } + if tokio::time::Instant::now() + HEALTH_INTERVAL > deadline { + return Err( + "The workspace's agent did not answer through the tunnel. Check that the image \ + includes triple-c-agent, then start the project again." + .into(), + ); + } + tokio::time::sleep(HEALTH_INTERVAL).await; + } +} + +impl Connector for SsmConnector { + async fn connect( + &self, + task: &TaskInfo, + token: &str, + ) -> Result<(Box, AgentEndpoint), String> { + let target = ecs_task(task)?; + let forward = PortForward::open(&self.sdk, &target, AGENT_PORT).await?; + let ep = forward.endpoint(token); + if let Err(e) = wait_healthy(&ep).await { + forward.close().await; + return Err(e); + } + Ok((Box::new(forward), ep)) + } +} + +// ─────────────────────────────────────────────────────────────── start + +/// What [`start_core`] needs besides the APIs. +pub struct StartInputs<'a> { + pub project: &'a Project, + pub who: &'a CallerIdentity, + pub env_pairs: &'a [(String, String)], + pub files: &'a [(String, Vec)], + pub image: &'a str, + pub wait: WaitTiming, + pub access_point_wait: PollTiming, +} + +/// A connected task. +pub struct Started { + pub env: EnvironmentInfo, + pub task: TaskInfo, + pub forward: Box, + pub endpoint: AgentEndpoint, + /// The access point the task mounts (Hybrid). `None` when an existing + /// task was attached to: the store's value stands. + pub access_point_id: Option, + /// This start ran the task (rather than attaching to a running one). + pub created: bool, +} + +/// Append the task's last log lines to a wait failure that lacks them. +async fn with_log_tail(api: &T, logs: &TaskLogs, arn: &str, err: String) -> String { + if err.contains("Last log lines") || err.contains("wrote no log lines") || err.contains("log could not be read") { + return err; + } + match api.log_tail(&logs.group, &task_log_stream(&logs.stream_prefix, arn), LOG_TAIL_LINES).await { + Ok(lines) if !lines.is_empty() => format!("{err}\nLast log lines:\n{}", lines.join("\n")), + _ => err, + } +} + +/// Stop a task this start ran, best-effort, so nothing billed is orphaned. +async fn stop_created(api: &T, cluster: &str, arn: &str) { + if let Err(e) = api.stop_task(cluster, arn, STOP_REASON).await { + log::warn!("Could not stop task {arn} after a failed start: {e}"); + } +} + +/// The start sequence; see the module docs. +pub async fn start_core( + tasks: &T, + work: &W, + rt: &R, + connector: &C, + inp: &StartInputs<'_>, + progress: &(dyn Fn(&str) + Send + Sync), +) -> Result { + let project = inp.project; + let (placement, cfg) = match (&project.aws_placement, &project.aws) { + (Some(p), Some(c)) if project.is_aws() => (p, c), + _ => return Err("The project is not set up to run on AWS".into()), + }; + let region = placement.region.as_str(); + let ok = owner_key(region, &inp.who.user_id); + + progress(PROGRESS_READING); + let env = read_environment(rt, &placement.environment, &inp.who.account, region) + .await + .map_err(s)?; + check_onboarded(rt, &env.name, &ok).await.map_err(s)?; + let family = task_family(&env.name, &ok, &project.id); + let logs = TaskLogs { + group: env.log_group.clone(), + stream_prefix: log_stream_prefix(&ok, &project.id), + }; + + let existing = find_project_task(tasks, &env.cluster_arn, &family, &inp.who.user_id, &project.id) + .await + .map_err(s)?; + + let (task, token, access_point_id, created) = if let Some(task) = existing { + // Attach: never rotate a running task's token (P2). + log::info!("Project {}: attaching to its running task {}", project.id, task.arn); + let token = read_agent_token(rt, &env.name, &ok, &project.id).await.map_err(s)?; + progress(PROGRESS_WAITING); + let task = match wait_ready(tasks, &env.cluster_arn, &task.arn, &logs, inp.wait, |_| {}).await { + Ok(t) => t, + Err(e) => return Err(with_log_tail(tasks, &logs, &task.arn, e.to_string()).await), + }; + (task, token, None, false) + } else { + progress(PROGRESS_PREPARING); + let prepared = prepare_workspace( + work, + inp.who, + &env, + project, + inp.env_pairs, + inp.files, + inp.image, + ) + .await + .map_err(s)?; + if let Some(ap) = &prepared.access_point_id { + wait_access_point(rt, ap, inp.access_point_wait).await.map_err(s)?; + } + progress(PROGRESS_STARTING); + let start = find_or_run( + tasks, + &env, + &inp.who.user_id, + &project.id, + &prepared.task_definition_arn, + cfg.spot, + ) + .await + .map_err(s)?; + if start.reused { + return Err(START_RACE.into()); + } + progress(PROGRESS_WAITING); + let arn = start.task.arn.clone(); + let task = match wait_ready(tasks, &env.cluster_arn, &arn, &logs, inp.wait, |_| {}).await { + Ok(t) => t, + Err(e) => { + let msg = with_log_tail(tasks, &logs, &arn, e.to_string()).await; + stop_created(tasks, &env.cluster_arn, &arn).await; + return Err(msg); + } + }; + (task, prepared.agent_token, prepared.access_point_id, true) + }; + + progress(PROGRESS_CONNECTING); + match connector.connect(&task, &token).await { + Ok((forward, endpoint)) => Ok(Started { + env, + task, + forward, + endpoint, + access_point_id, + created, + }), + Err(e) => { + if created { + stop_created(tasks, &env.cluster_arn, &task.arn).await; + } + Err(e) + } + } +} + +/// The connection, region and SDK config of an AWS project. +async fn project_sdk(project: &Project, settings: &AppSettings) -> Result<(String, String, SdkConfig), String> { + let host_id = project.host_id.as_deref().ok_or("The project has no host")?; + let region = project + .aws_placement + .as_ref() + .map(|p| p.region.clone()) + .ok_or("The project has no AWS placement")?; + let conn = crate::commands::host_commands::aws_conn(host_id, settings)?; + let sdk = sdk_config(&conn, ®ion).await.map_err(s)?; + Ok((host_id.to_string(), region, sdk)) +} + +/// `ProjectHost::Aws::ensure_started`. +pub async fn ensure_started( + project: &Project, + settings: &AppSettings, + progress: &(dyn Fn(&str) + Send + Sync), +) -> Result { + // Already connected and answering: nothing to do. + if let Some(ep) = tunnels::get_endpoint(&project.id) { + if matches!(agent::is_running(&ep).await, Ok(true)) { + return Ok(ContainerTarget::Agent(ep)); + } + tunnels::close(&project.id).await; + } + + // Everything local first: a bad image or file fails before any AWS call. + let image = aws_image(settings)?; + let (env_pairs, ca) = aws_env(project, settings)?; + let ssh = project + .ssh_key_path + .as_deref() + .or(settings.default_ssh_key_path.as_deref()); + let files = workspace_files(ssh, ca.as_ref())?; + + let (connection_id, region, sdk) = project_sdk(project, settings).await?; + let who = caller_identity(&sdk).await.map_err(s)?; + let started = start_core( + &SdkTaskApi::new(&sdk), + &SdkWorkspaceApi::new(&sdk), + &SdkRuntimeApi::new(&sdk), + &SsmConnector { sdk: sdk.clone() }, + &StartInputs { + project, + who: &who, + env_pairs: &env_pairs, + files: &files, + image: &image, + wait: WaitTiming::default(), + access_point_wait: PollTiming::default(), + }, + progress, + ) + .await?; + + log::info!( + "Project {}: {} task {}", + project.id, + if started.created { "started" } else { "attached to" }, + started.task.arn + ); + let endpoint = started.endpoint.clone(); + let access_point_id = started + .access_point_id + .or_else(|| project.aws.as_ref().and_then(|a| a.access_point_id.clone())); + let ecs = ecs_task(&started.task)?; + if let Some(old) = tunnels::register( + &project.id, + LiveTunnel { + task: ecs, + task_arn: started.task.arn.clone(), + cluster_arn: started.env.cluster_arn.clone(), + connection_id, + region, + sdk, + forward: started.forward, + endpoint: endpoint.clone(), + access_point_id, + task_definition_arn: started.task.task_definition_arn.clone(), + }, + ) { + old.forward.close().await; + } + Ok(ContainerTarget::Agent(endpoint)) +} + +// ──────────────────────────────────────────────────────────────── stop + +/// StopTask the task in `task_arn`'s cluster. +pub async fn stop_task_arn(api: &T, task_arn: &str) -> Result<(), String> { + let cluster = cluster_of_task(task_arn) + .ok_or_else(|| format!("{task_arn} is not a task ARN with a cluster"))?; + api.stop_task(cluster, task_arn, STOP_REASON).await.map_err(s) +} + +/// `ProjectHost::stop` for an agent target: the task behind a registered +/// tunnel. `None` when `ep` is not one of ours. +pub async fn stop_endpoint(ep: &AgentEndpoint) -> Option> { + let (_connection, _region, cluster, arn) = tunnels::task_of(ep)?; + let (project_id, sdk) = tunnels::owner_of(ep)?; + let res = SdkTaskApi::new(&sdk) + .stop_task(&cluster, &arn, STOP_REASON) + .await + .map_err(s); + if res.is_ok() { + tunnels::close(&project_id).await; + } + Some(res) +} + +/// Stop an AWS project's task: through its tunnel when there is one, else +/// the stored task ARN, else whatever live task the project has (a crash +/// between RunTask and the store write). Closes the tunnel. +pub async fn stop_project(project: &Project, settings: &AppSettings) -> Result<(), String> { + if let Some(ep) = tunnels::get_endpoint(&project.id) { + if let Some(res) = stop_endpoint(&ep).await { + return res; + } + } + let (_, region, sdk) = project_sdk(project, settings).await?; + let api = SdkTaskApi::new(&sdk); + let stored = project.aws.as_ref().and_then(|a| a.task_arn.clone()); + let arn = match stored { + Some(a) => Some(a), + None => { + let env = project.aws_placement.as_ref().map(|p| p.environment.clone()).unwrap_or_default(); + let who = caller_identity(&sdk).await.map_err(s)?; + let ok = owner_key(®ion, &who.user_id); + find_project_task(&api, &format!("triple-c-{env}"), &task_family(&env, &ok, &project.id), &who.user_id, &project.id) + .await + .map_err(s)? + .map(|t| t.arn) + } + }; + if let Some(arn) = arn { + stop_task_arn(&api, &arn).await?; + } + tunnels::close(&project.id).await; + Ok(()) +} + +// ────────────────────────────────────────────────────────────── remove + +/// Stop the task and remove the project's AWS resources (task definition +/// revisions, the secret with a 7-day window, and with `discard_data` the +/// access point). +pub async fn remove_resources( + project: &Project, + settings: &AppSettings, + discard_data: bool, +) -> Result<(), String> { + tunnels::close(&project.id).await; + let (_, region, sdk) = project_sdk(project, settings).await?; + let who = caller_identity(&sdk).await.map_err(s)?; + let ok = owner_key(®ion, &who.user_id); + let env = project.aws_placement.as_ref().map(|p| p.environment.clone()).unwrap_or_default(); + let aws = project.aws.as_ref(); + remove_workspace_resources( + &SdkTaskApi::new(&sdk), + &SdkWorkspaceApi::new(&sdk), + &SdkRuntimeApi::new(&sdk), + &RemoveRequest { + env: &env, + region: ®ion, + owner_user_id: &who.user_id, + owner_key: &ok, + project_id: &project.id, + stored_task_arn: aws.and_then(|a| a.task_arn.as_deref()), + stored_access_point_id: aws.and_then(|a| a.access_point_id.as_deref()), + discard_data, + }, + ) + .await + .map_err(s) +} + +// ─────────────────────────────────────────────────────────── reconcile + +/// What reconcile found for a Running AWS project. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Reconciled { + /// The task is up (or on its way): re-open the tunnel. + Live(Box), + /// The task stopped, or ECS no longer knows it: the project is Stopped. + Gone, +} + +/// Ask ECS about a project's stored task. No stored task is `Gone`. +pub async fn reconcile_task(api: &T, task_arn: Option<&str>) -> Result { + let Some(arn) = task_arn else { + return Ok(Reconciled::Gone); + }; + let Some(cluster) = cluster_of_task(arn) else { + return Ok(Reconciled::Gone); + }; + let found = api.describe(cluster, &[arn.to_string()]).await.map_err(s)?; + Ok(match found.into_iter().find(|t| t.arn == arn) { + Some(t) + if t.desired_status == "RUNNING" + && matches!(t.last_status.as_str(), "PROVISIONING" | "PENDING" | "ACTIVATING" | "RUNNING") => + { + Reconciled::Live(Box::new(t)) + } + _ => Reconciled::Gone, + }) +} + +/// Reconcile one AWS project the store has as Running (or Starting/ +/// Stopping/Error left by a crash): a gone task makes it Stopped (its task +/// ARN cleared); a live one keeps it Running. An ECS that can't be asked +/// changes nothing — marking Stopped would hide a billed task. +pub async fn reconcile_project( + api: &T, + store: &ProjectsStore, + project: &Project, +) -> Result { + let arn = project.aws.as_ref().and_then(|a| a.task_arn.clone()); + let r = reconcile_task(api, arn.as_deref()).await?; + match &r { + Reconciled::Gone => { + store.update_status(&project.id, ProjectStatus::Stopped)?; + if let Some(a) = project.aws.as_ref() { + store.set_aws_runtime(&project.id, None, a.access_point_id.clone(), a.task_definition_arn.clone())?; + } + } + Reconciled::Live(_) => { + if project.status != ProjectStatus::Running { + store.update_status(&project.id, ProjectStatus::Running)?; + } + } + } + Ok(r) +} + +/// `reconcile_project_statuses` for an AWS project: decide, and when the +/// task is live re-open its tunnel in the background (terminals wait for +/// it through [`tunnels::wait_endpoint`]). Never blocks on the tunnel. +pub async fn reconcile(project: &Project, settings: &AppSettings, store: Arc) { + let (connection_id, region, sdk) = match project_sdk(project, settings).await { + Ok(v) => v, + Err(e) => { + log::warn!("Project {}: could not reach AWS to reconcile ({e}); status left as is", project.id); + return; + } + }; + let api = SdkTaskApi::new(&sdk); + match reconcile_project(&api, &store, project).await { + Ok(Reconciled::Live(task)) => { + tunnels::set_pending(&project.id, true); + let project = project.clone(); + tauri::async_runtime::spawn(async move { + if let Err(e) = reattach(&project, *task, connection_id, region, sdk, &store).await { + log::warn!("Project {}: could not re-open the tunnel to its task: {e}", project.id); + tunnels::set_pending(&project.id, false); + } + }); + } + Ok(Reconciled::Gone) => { + log::info!("Project '{}' ({}): its task is gone — set to Stopped", project.name, project.id) + } + Err(e) => log::warn!("Project {}: could not check its task ({e}); status left as is", project.id), + } +} + +/// Re-open the tunnel to a live task found at startup. +async fn reattach( + project: &Project, + task: TaskInfo, + connection_id: String, + region: String, + sdk: SdkConfig, + store: &ProjectsStore, +) -> Result<(), String> { + let env = project.aws_placement.as_ref().map(|p| p.environment.clone()).unwrap_or_default(); + let who = caller_identity(&sdk).await.map_err(s)?; + let ok = owner_key(®ion, &who.user_id); + let rt = SdkRuntimeApi::new(&sdk); + let api = SdkTaskApi::new(&sdk); + let token = read_agent_token(&rt, &env, &ok, &project.id).await.map_err(s)?; + let cluster_arn = cluster_arn_of_task(&task.arn).ok_or("bad task ARN")?; + let logs = TaskLogs { group: format!("/triple-c/{env}"), stream_prefix: log_stream_prefix(&ok, &project.id) }; + let task = wait_ready(&api, &cluster_arn, &task.arn, &logs, WaitTiming::default(), |_| {}) + .await + .map_err(s)?; + let (forward, endpoint) = SsmConnector { sdk: sdk.clone() }.connect(&task, &token).await?; + if let Some(old) = tunnels::register( + &project.id, + LiveTunnel { + task: ecs_task(&task)?, + task_arn: task.arn.clone(), + cluster_arn, + connection_id, + region, + sdk, + forward, + endpoint: endpoint.clone(), + access_point_id: project.aws.as_ref().and_then(|a| a.access_point_id.clone()), + task_definition_arn: task.task_definition_arn.clone(), + }, + ) { + old.forward.close().await; + } + // The local port is new, so the stored target id is too. + store.set_container_id(&project.id, Some(endpoint.base_url.to_string()))?; + log::info!("Project {}: tunnel to task {} re-opened", project.id, task.arn); + Ok(()) +} + +// ─────────────────────────────────────────────────────────────── watch + +/// The AWS death watch's lookup: Running AWS projects from the store, their +/// targets from the registry (or, while unattached, an endpoint with the +/// stored id so `container_watch::settle` still matches), and +/// `stopped_among` per connection/region/cluster. +pub struct StoreTaskLookup { + pub projects: Arc, + pub settings: Arc, +} + +/// A Running AWS project's watched task, if it has one. +pub fn watched_task(p: &Project) -> Option { + if p.status != ProjectStatus::Running || !p.is_aws() { + return None; + } + let arn = p.aws.as_ref()?.task_arn.clone()?; + let target = match tunnels::get_endpoint(&p.id) { + Some(ep) => ep, + None => AgentEndpoint::new(url::Url::parse(p.container_id.as_deref()?).ok()?, "unattached").ok()?, + }; + if Some(target.base_url.as_str()) != p.container_id.as_deref() { + return None; + } + Some(WatchedTask { + target: ContainerTarget::Agent(target), + group: TaskGroup { + connection_id: p.host_id.clone()?, + region: p.aws_placement.as_ref()?.region.clone(), + cluster_arn: cluster_arn_of_task(&arn)?, + }, + task_arn: arn, + }) +} + +impl AwsTaskLookup for StoreTaskLookup { + fn running_tasks(&self) -> Vec { + self.projects.list().iter().filter_map(watched_task).collect() + } + + fn stopped<'a>( + &'a self, + group: &'a TaskGroup, + arns: &'a [String], + ) -> BoxFuture<'a, Result, String>> { + Box::pin(async move { + let settings = self.settings.get(); + let conn = crate::commands::host_commands::aws_conn(&group.connection_id, &settings)?; + let sdk = sdk_config(&conn, &group.region).await.map_err(s)?; + let stopped = stopped_among(&SdkTaskApi::new(&sdk), &group.cluster_arn, arns) + .await + .map_err(s)?; + record_stopped(&self.projects.list(), &stopped); + Ok(stopped) + }) + } +} + +/// Tell the registry which targets' tasks ECS reports stopped, so +/// `has_exited` confirms them (R3). +pub fn record_stopped(projects: &[Project], stopped: &[String]) { + for w in projects.iter().filter_map(watched_task) { + if stopped.contains(&w.task_arn) { + tunnels::mark_stopped(w.target.id()); + } + } +} + +#[cfg(test)] +mod tests; diff --git a/app/src-tauri/src/host/aws_host/tests.rs b/app/src-tauri/src/host/aws_host/tests.rs new file mode 100644 index 0000000..afa31b7 --- /dev/null +++ b/app/src-tauri/src/host/aws_host/tests.rs @@ -0,0 +1,543 @@ +use std::collections::HashMap; +use std::sync::atomic::AtomicUsize; +use std::sync::Mutex; + +use aws_sdk_ecs::operation::register_task_definition::RegisterTaskDefinitionInput; +use aws_sdk_ecs::operation::run_task::RunTaskInput; + +use super::*; +use crate::aws::environments::Knowable; +use crate::aws::tasks::TaskInfo; +use crate::aws::workspace::{AccessPointRef, FileSystemRef, TaskDefinitionRef}; +use crate::aws::AwsError; +use crate::host::lifecycle::ProjectHost; +use crate::host::tunnels::tests::{fake_tunnel, FakeTunnel}; +use crate::models::{AwsPlacement, AwsWorkspaceConfig, StorageMode}; + +const REGION: &str = "eu-west-1"; +const ACCOUNT: &str = "111122223333"; +const USER: &str = "AROAEXAMPLEID:alice@example.com"; + +fn who() -> CallerIdentity { + CallerIdentity { + account: ACCOUNT.into(), + arn: format!("arn:aws:sts::{ACCOUNT}:assumed-role/dev/alice"), + user_id: USER.into(), + } +} + +fn aws_project(storage: StorageMode) -> Project { + let mut p = Project::new("aws".into(), Vec::new()); + p.host_id = Some("c1".into()); + p.aws_placement = Some(AwsPlacement { region: REGION.into(), environment: "dev".into() }); + p.aws = Some(AwsWorkspaceConfig { storage, ..Default::default() }); + p +} + +fn family(p: &Project) -> String { + task_family("dev", &owner_key(REGION, USER), &p.id) +} + +fn task_arn(id: &str) -> String { + format!("arn:aws:ecs:{REGION}:{ACCOUNT}:task/triple-c-dev/{id}") +} + +fn cluster_arn() -> String { + format!("arn:aws:ecs:{REGION}:{ACCOUNT}:cluster/triple-c-dev") +} + +fn task(p: &Project, id: &str, last: &str, ready: bool) -> TaskInfo { + TaskInfo { + arn: task_arn(id), + last_status: last.into(), + desired_status: if last == "STOPPED" { "STOPPED" } else { "RUNNING" }.into(), + health: None, + stop_code: (last == "STOPPED").then(|| "EssentialContainerExited".into()), + stopped_reason: (last == "STOPPED").then(|| "Essential container exited".into()), + task_definition_arn: format!("arn:aws:ecs:{REGION}:{ACCOUNT}:task-definition/{}:1", family(p)), + runtime_id: ready.then(|| "rt-1".into()), + exec_agent_running: ready, + owner: Some(USER.into()), + project: Some(p.id.clone()), + } +} + +fn record() -> String { + serde_json::json!({ + "schema": 1, "name": "dev", "template_version": 1, + "cluster_arn": cluster_arn(), + "security_group_id": "sg-0123456789abcdef0", + "subnet_ids": ["subnet-0a", "subnet-0b"], + "assign_public_ip": true, + "kms_key_arn": format!("arn:aws:kms:{REGION}:{ACCOUNT}:key/1234-abcd"), + "log_group": "/triple-c/dev", "network_mode": "public", "tenancy": "strict" + }) + .to_string() +} + +/// One fake for all three AWS traits and the connector, logging calls. +#[derive(Default)] +struct Fake { + calls: Mutex>, + /// ListTasks answers in turn; the last repeats. + listed: Mutex>>, + tasks: Mutex>, + /// What RunTask starts. + run: Option, + logs: Vec, + role: Option, + connect_fails: bool, + connect_tokens: Mutex>, + secret_json: Mutex>, +} + +impl Fake { + fn log(&self, s: impl Into) { + self.calls.lock().unwrap().push(s.into()); + } + fn calls(&self) -> Vec { + self.calls.lock().unwrap().clone() + } + fn called(&self, prefix: &str) -> bool { + self.calls().iter().any(|c| c.starts_with(prefix)) + } + fn with_task(self, t: TaskInfo) -> Self { + self.tasks.lock().unwrap().insert(t.arn.clone(), t); + self + } +} + +impl TaskApi for Fake { + async fn run_task(&self, _req: RunTaskInput) -> Result { + self.log("run"); + let t = self.run.clone().expect("a run was expected"); + self.tasks.lock().unwrap().insert(t.arn.clone(), t.clone()); + Ok(t) + } + async fn describe(&self, _c: &str, arns: &[String]) -> Result, AwsError> { + let t = self.tasks.lock().unwrap(); + Ok(arns.iter().filter_map(|a| t.get(a).cloned()).collect()) + } + async fn list_project_tasks(&self, _c: &str, _f: &str) -> Result, AwsError> { + self.log("list"); + let mut l = self.listed.lock().unwrap(); + Ok(if l.len() > 1 { l.remove(0) } else { l.first().cloned().unwrap_or_default() }) + } + async fn stop_task(&self, _c: &str, arn: &str, reason: &str) -> Result<(), AwsError> { + self.log(format!("stop {arn} {reason}")); + Ok(()) + } + async fn log_tail(&self, _g: &str, _s: &str, _l: i32) -> Result, AwsError> { + Ok(self.logs.clone()) + } +} + +impl WorkspaceApi for Fake { + async fn find_file_system(&self, _t: &str) -> Result, AwsError> { + Ok(Some(FileSystemRef { id: "fs-1".into(), owner: Some(USER.into()) })) + } + async fn access_points(&self, _fs: &str) -> Result, AwsError> { + Ok(Vec::new()) + } + async fn create_access_point( + &self, + _fs: &str, + _root: &str, + _tags: &[(String, String)], + ) -> Result { + self.log("create-ap"); + Ok("fsap-new".into()) + } + async fn put_project_secret( + &self, + _name: &str, + _kms: &str, + json: &str, + _tags: &[(String, String)], + ) -> Result { + self.log("put-secret"); + *self.secret_json.lock().unwrap() = Some(json.into()); + Ok(format!("arn:aws:secretsmanager:{REGION}:{ACCOUNT}:secret:triple-c/dev/x-AbCdEf")) + } + async fn task_definition_revisions(&self, _f: &str) -> Result, AwsError> { + Ok(Vec::new()) + } + async fn describe_task_definition( + &self, + _arn: &str, + ) -> Result, AwsError> { + Ok(None) + } + async fn register_task_definition( + &self, + req: RegisterTaskDefinitionInput, + ) -> Result { + self.log("register"); + Ok(format!( + "arn:aws:ecs:{REGION}:{ACCOUNT}:task-definition/{}:1", + req.family().unwrap() + )) + } +} + +impl RuntimeApi for Fake { + async fn environment_record(&self, env: &str) -> Result, AwsError> { + self.log(format!("record {env}")); + Ok(Some(record())) + } + async fn role_exists(&self, _name: &str) -> Result { + self.log("role"); + Ok(self.role.clone().unwrap_or(Knowable::Yes)) + } + async fn secret_string(&self, _name: &str) -> Result { + self.log("get-secret"); + Ok(r#"{"TRIPLEC_AGENT_TOKEN":"token-of-the-running-task"}"#.into()) + } + async fn access_point_state(&self, id: &str) -> Result, AwsError> { + self.log(format!("ap-state {id}")); + Ok(Some("available".into())) + } + async fn deregister_task_definition(&self, _arn: &str) -> Result<(), AwsError> { + Ok(()) + } + async fn delete_task_definitions(&self, _arns: &[String]) -> Result<(), AwsError> { + Ok(()) + } + async fn delete_secret(&self, _name: &str) -> Result<(), AwsError> { + Ok(()) + } + async fn delete_access_point(&self, _id: &str) -> Result<(), AwsError> { + Ok(()) + } +} + +impl Connector for Fake { + async fn connect( + &self, + _task: &TaskInfo, + token: &str, + ) -> Result<(Box, AgentEndpoint), String> { + self.log("connect"); + self.connect_tokens.lock().unwrap().push(token.into()); + if self.connect_fails { + return Err("session-manager-plugin was not found".into()); + } + Ok(( + Box::new(FakeTunnel(std::sync::Arc::new(AtomicUsize::new(0)))), + AgentEndpoint::loopback(61100, token), + )) + } +} + +fn inputs<'a>(p: &'a Project, who: &'a CallerIdentity) -> StartInputs<'a> { + StartInputs { + project: p, + who, + env_pairs: &[], + files: &[], + image: REGISTRY_IMAGE, + wait: WaitTiming { + interval: Duration::from_millis(1), + timeout: Duration::from_millis(30), + }, + access_point_wait: PollTiming { + interval: Duration::from_millis(1), + timeout: Duration::from_millis(30), + }, + } +} + +async fn start(fake: &Fake, p: &Project) -> (Result, Vec) { + let w = who(); + let steps = Mutex::new(Vec::new()); + let progress = |m: &str| steps.lock().unwrap().push(m.to_string()); + let r = start_core(fake, fake, fake, fake, &inputs(p, &w), &progress).await; + (r, steps.into_inner().unwrap()) +} + +#[tokio::test] +async fn a_first_start_prepares_runs_waits_and_connects_in_that_order() { + let p = aws_project(StorageMode::Hybrid); + let fake = Fake { run: Some(task(&p, "new", "RUNNING", true)), ..Default::default() }; + let (r, steps) = start(&fake, &p).await; + let started = r.unwrap(); + assert!(started.created); + assert_eq!(started.access_point_id.as_deref(), Some("fsap-new")); + assert_eq!(started.task.arn, task_arn("new")); + assert_eq!( + fake.calls(), + vec![ + "record dev", "role", "list", "create-ap", "put-secret", "register", + "ap-state fsap-new", "list", "run", "connect" + ], + "look for a live task before rotating the token; the access point is available before RunTask" + ); + assert_eq!(steps, vec![PROGRESS_READING, PROGRESS_PREPARING, PROGRESS_STARTING, PROGRESS_WAITING, PROGRESS_CONNECTING]); + // The connector got the token the secret was just written with. + let secret: serde_json::Value = + serde_json::from_str(fake.secret_json.lock().unwrap().as_deref().unwrap()).unwrap(); + assert_eq!( + fake.connect_tokens.lock().unwrap()[0], + secret["TRIPLEC_AGENT_TOKEN"].as_str().unwrap() + ); +} + +#[tokio::test] +async fn a_running_task_is_attached_with_its_own_token_and_nothing_is_rewritten() { + let p = aws_project(StorageMode::Hybrid); + let live = task(&p, "live", "RUNNING", true); + let fake = Fake { + listed: Mutex::new(vec![vec![live.arn.clone()]]), + ..Default::default() + } + .with_task(live); + let (r, steps) = start(&fake, &p).await; + let started = r.unwrap(); + assert!(!started.created); + assert_eq!(started.access_point_id, None); + assert_eq!(fake.calls(), vec!["record dev", "role", "list", "get-secret", "connect"]); + assert_eq!(fake.connect_tokens.lock().unwrap()[0], "token-of-the-running-task"); + assert_eq!(steps, vec![PROGRESS_READING, PROGRESS_WAITING, PROGRESS_CONNECTING]); +} + +#[tokio::test] +async fn losing_the_run_race_is_a_clear_error_and_nothing_is_stopped() { + let p = aws_project(StorageMode::Ephemeral); + let other = task(&p, "other", "RUNNING", true); + let fake = Fake { + // Nothing at the first look; a concurrent start's task at the second. + listed: Mutex::new(vec![vec![], vec![other.arn.clone()]]), + ..Default::default() + } + .with_task(other); + let (r, _) = start(&fake, &p).await; + assert_eq!(r.err().as_deref(), Some(START_RACE)); + assert!(!fake.called("run") && !fake.called("stop") && !fake.called("connect")); +} + +#[tokio::test] +async fn a_task_that_stops_while_starting_is_reported_with_its_log() { + let p = aws_project(StorageMode::Ephemeral); + let fake = Fake { + run: Some(task(&p, "dies", "STOPPED", false)), + logs: vec!["entrypoint: cannot decode TRIPLEC_FILES".into()], + ..Default::default() + }; + let (r, _) = start(&fake, &p).await; + let e = r.err().unwrap(); + assert!(e.contains("Essential container exited") && e.contains("cannot decode"), "{e}"); + assert!(!fake.called("connect")); +} + +#[tokio::test] +async fn a_task_that_never_gets_ready_is_stopped_and_its_log_shown() { + let p = aws_project(StorageMode::Ephemeral); + let fake = Fake { + run: Some(task(&p, "slow", "PENDING", false)), + logs: vec!["pulling image".into()], + ..Default::default() + }; + let (r, _) = start(&fake, &p).await; + let e = r.err().unwrap(); + assert!(e.contains("did not become ready") && e.contains("pulling image"), "{e}"); + assert!(fake.calls().contains(&format!("stop {} {STOP_REASON}", task_arn("slow")))); +} + +#[tokio::test] +async fn a_tunnel_failure_stops_a_task_this_start_ran_but_never_an_attached_one() { + let p = aws_project(StorageMode::Ephemeral); + let fake = Fake { + run: Some(task(&p, "new", "RUNNING", true)), + connect_fails: true, + ..Default::default() + }; + let (r, _) = start(&fake, &p).await; + assert!(r.err().unwrap().contains("session-manager-plugin")); + assert!(fake.calls().contains(&format!("stop {} {STOP_REASON}", task_arn("new")))); + + let live = task(&p, "live", "RUNNING", true); + let attached = Fake { + listed: Mutex::new(vec![vec![live.arn.clone()]]), + connect_fails: true, + ..Default::default() + } + .with_task(live); + let (r, _) = start(&attached, &p).await; + assert!(r.is_err()); + assert!(!attached.called("stop")); +} + +#[tokio::test] +async fn a_user_who_has_not_joined_is_told_before_anything_is_written() { + let p = aws_project(StorageMode::Hybrid); + let fake = Fake { role: Some(Knowable::No), ..Default::default() }; + let (r, _) = start(&fake, &p).await; + assert!(r.err().unwrap().contains("not joined")); + assert_eq!(fake.calls(), vec!["record dev", "role"]); +} + +// ──────────────────────────────────────────────────────────────── inputs + +#[test] +fn aws_runs_registry_images_only() { + let mut s = AppSettings::default(); + s.image_source = ImageSource::LocalBuild; + assert_eq!(aws_image(&s).unwrap(), REGISTRY_IMAGE); + s.image_source = ImageSource::Registry; + assert_eq!(aws_image(&s).unwrap(), REGISTRY_IMAGE); + s.image_source = ImageSource::Custom; + for good in ["registry.example.com/team/img:1", "ghcr.io/x/y", "user/repo:tag", "reg:5000/a"] { + s.custom_image_name = Some(good.into()); + assert_eq!(aws_image(&s).unwrap(), good); + } + for bad in ["", "triple-c:latest", "ubuntu", "localhost/x", "127.0.0.1:5000/x", "a /b", "/x"] { + s.custom_image_name = Some(bad.into()); + assert!(aws_image(&s).is_err(), "{bad}"); + } +} + +#[test] +fn the_sandbox_is_forced_off_on_aws() { + let mut p = aws_project(StorageMode::Ephemeral); + p.sandbox_mode_enabled = true; + let (pairs, _) = aws_env(&p, &AppSettings::default()).unwrap(); + let settings = &pairs.iter().find(|(k, _)| k == "CLAUDE_CODE_SETTINGS_JSON").unwrap().1; + let v: serde_json::Value = serde_json::from_str(settings).unwrap(); + assert_eq!(v["sandbox"]["enabled"], false, "{settings}"); +} + +#[test] +fn files_for_the_secret_come_from_the_ssh_folder_without_leaving_it() { + let base = tempfile::tempdir().unwrap(); + let ssh = base.path().join("ssh"); + std::fs::create_dir(&ssh).unwrap(); + std::fs::write(ssh.join("id_ed25519"), b"KEY").unwrap(); + std::fs::write(ssh.join("config"), b"Host *").unwrap(); + std::fs::create_dir(ssh.join("sub")).unwrap(); + std::fs::write(ssh.join("sub/nested"), b"x").unwrap(); + std::fs::write(base.path().join("outside"), b"SECRET-OUTSIDE").unwrap(); + #[cfg(unix)] + { + std::os::unix::fs::symlink(base.path().join("outside"), ssh.join("escape")).unwrap(); + std::os::unix::fs::symlink(ssh.join("config"), ssh.join("alias")).unwrap(); + } + let got = workspace_files(Some(ssh.to_str().unwrap()), None).unwrap(); + let names: Vec<&str> = got.iter().map(|(k, _)| k.as_str()).collect(); + #[cfg(unix)] + assert_eq!(names, vec!["ssh/alias", "ssh/config", "ssh/id_ed25519"]); + #[cfg(not(unix))] + assert_eq!(names, vec!["ssh/config", "ssh/id_ed25519"]); + assert!(got.iter().all(|(_, v)| v != b"SECRET-OUTSIDE")); + + std::fs::write(ssh.join("huge"), vec![b'x'; MAX_FILE_BYTES as usize + 1]).unwrap(); + let e = workspace_files(Some(ssh.to_str().unwrap()), None).unwrap_err(); + assert!(e.contains("huge") && e.contains("64 KiB"), "{e}"); + + assert!(workspace_files(Some("/"), None).is_err(), "a filesystem root is refused"); + assert!(workspace_files(None, None).unwrap().is_empty()); +} + +#[test] +fn ca_files_are_what_docker_would_mount() { + let base = tempfile::tempdir().unwrap(); + let dir = base.path().join("ca"); + std::fs::create_dir(&dir).unwrap(); + std::fs::write(dir.join("corp.pem"), b"PEM").unwrap(); + std::fs::write(dir.join("README"), b"not a cert").unwrap(); + let ca = crate::docker::ca_certs::resolve(Some(dir.to_str().unwrap())).unwrap().unwrap(); + let got = workspace_files(None, Some(&ca)).unwrap(); + assert_eq!(got, vec![("ca/corp.pem".to_string(), b"PEM".to_vec())]); + + let one = base.path().join("Root CA.pem"); + std::fs::write(&one, b"ONE").unwrap(); + let ca = crate::docker::ca_certs::resolve(Some(one.to_str().unwrap())).unwrap().unwrap(); + let got = workspace_files(None, Some(&ca)).unwrap(); + assert_eq!(got, vec![("ca/Root_CA.crt".to_string(), b"ONE".to_vec())]); +} + +// ───────────────────────────────────────────────── targets and reconcile + +#[tokio::test] +async fn an_aws_projects_target_is_its_registered_tunnel() { + let mut p = aws_project(StorageMode::Ephemeral); + p.container_id = Some("http://127.0.0.1:61200/".into()); + assert_eq!(ProjectHost::for_project(&p), ProjectHost::Aws); + assert_eq!(ContainerTarget::for_project(&p), None, "no tunnel, no target, whatever is stored"); + tunnels::register(&p.id, fake_tunnel(61200, std::sync::Arc::new(AtomicUsize::new(0)))); + assert_eq!( + ContainerTarget::for_project(&p), + Some(ContainerTarget::Agent(AgentEndpoint::loopback(61200, "tok"))) + ); + assert_eq!( + ContainerTarget::for_project_waiting(&p).await, + ContainerTarget::for_project(&p) + ); + tunnels::close(&p.id).await; + + let local = Project::new("local".into(), Vec::new()); + assert_eq!(ProjectHost::for_project(&local), ProjectHost::Docker); +} + +fn store_with(p: &Project) -> (tempfile::TempDir, ProjectsStore) { + let dir = tempfile::tempdir().unwrap(); + let store = ProjectsStore::over_file_for_test(dir.path().join("projects.json"), vec![p.clone()]); + (dir, store) +} + +#[tokio::test] +async fn reconcile_marks_a_project_whose_task_stopped_as_stopped() { + let mut p = aws_project(StorageMode::Hybrid); + p.status = ProjectStatus::Running; + { + let a = p.aws.as_mut().unwrap(); + a.task_arn = Some(task_arn("gone")); + a.access_point_id = Some("fsap-1".into()); + a.task_definition_arn = Some("arn:td:1".into()); + } + let fake = Fake::default().with_task(task(&p, "gone", "STOPPED", false)); + let (_d, store) = store_with(&p); + assert_eq!(reconcile_project(&fake, &store, &p).await.unwrap(), Reconciled::Gone); + let after = store.get(&p.id).unwrap(); + assert_eq!(after.status, ProjectStatus::Stopped); + let a = after.aws.unwrap(); + assert_eq!(a.task_arn, None, "a stopped task's ARN is cleared"); + assert_eq!(a.access_point_id.as_deref(), Some("fsap-1"), "the data's access point stays"); + + // ECS has forgotten it: also gone. + let forgotten = Fake::default(); + let (_d, store) = store_with(&p); + assert_eq!(reconcile_project(&forgotten, &store, &p).await.unwrap(), Reconciled::Gone); + assert_eq!(store.get(&p.id).unwrap().status, ProjectStatus::Stopped); +} + +#[tokio::test] +async fn reconcile_keeps_a_project_with_a_live_task_running() { + let mut p = aws_project(StorageMode::Hybrid); + p.status = ProjectStatus::Running; + p.aws.as_mut().unwrap().task_arn = Some(task_arn("up")); + let fake = Fake::default().with_task(task(&p, "up", "RUNNING", true)); + let (_d, store) = store_with(&p); + assert!(matches!(reconcile_project(&fake, &store, &p).await.unwrap(), Reconciled::Live(_))); + assert_eq!(store.get(&p.id).unwrap().status, ProjectStatus::Running); +} + +#[tokio::test] +async fn the_watch_reports_a_running_aws_project_as_its_stored_target_and_confirms_a_recorded_stop() { + let mut p = aws_project(StorageMode::Ephemeral); + p.status = ProjectStatus::Running; + p.container_id = Some("http://127.0.0.1:61300/".into()); + p.aws.as_mut().unwrap().task_arn = Some(task_arn("w1")); + let w = watched_task(&p).expect("watched"); + assert_eq!(w.target.id(), "http://127.0.0.1:61300/", "settle's running_on matches the stored id"); + assert_eq!(w.group.cluster_arn, cluster_arn()); + assert_eq!(w.group.connection_id, "c1"); + assert!(!w.target.has_exited().await, "nothing recorded yet"); + record_stopped(std::slice::from_ref(&p), &[task_arn("w1")]); + assert!(w.target.has_exited().await); + + let mut stopped = p.clone(); + stopped.status = ProjectStatus::Stopped; + assert!(watched_task(&stopped).is_none(), "a Stopped project is not polled again"); + let local = Project::new("l".into(), Vec::new()); + assert!(watched_task(&local).is_none()); +} diff --git a/app/src-tauri/src/host/lifecycle.rs b/app/src-tauri/src/host/lifecycle.rs index 0e63b0a..d38ee78 100644 --- a/app/src-tauri/src/host/lifecycle.rs +++ b/app/src-tauri/src/host/lifecycle.rs @@ -18,11 +18,13 @@ use crate::docker; use crate::models::{AppSettings, Project}; use crate::models::container_config; -/// The host a project runs on. One variant today; an AWS variant will carry -/// the connection, region and environment (see the spec's data model). +/// The host a project runs on: the local Docker daemon, or an ECS Fargate +/// task in an AWS environment (`aws_host.rs`; the connection, region and +/// environment come from the project). #[derive(Debug, Clone, PartialEq, Eq)] pub enum ProjectHost { Docker, + Aws, } /// An agent target is a way *into* a container, not a handle on its @@ -41,8 +43,12 @@ pub struct DiscardLeftovers { } impl ProjectHost { - pub fn for_project(_project: &Project) -> Self { - ProjectHost::Docker + pub fn for_project(project: &Project) -> Self { + if project.is_aws() { + ProjectHost::Aws + } else { + ProjectHost::Docker + } } /// Make sure the project's container exists, matches the project's @@ -60,6 +66,7 @@ impl ProjectHost { ProjectHost::Docker => docker_ensure_started(project, settings, progress) .await .map(ContainerTarget::docker), + ProjectHost::Aws => super::aws_host::ensure_started(project, settings, progress).await, } } @@ -72,24 +79,66 @@ impl ProjectHost { ProjectHost::Docker => Ok(docker::find_existing_container(project) .await? .map(ContainerTarget::docker)), + // The tunnel registry: an AWS project's target exists while its + // tunnel is open. + ProjectHost::Aws => Ok(ContainerTarget::for_project(project)), } } + /// Stop the container behind `target`. For an agent target that is the + /// ECS task behind a registered tunnel (`StopTask`, reason "Stopped from + /// Triple-C"), and the tunnel is closed. pub async fn stop(&self, target: &ContainerTarget) -> Result<(), String> { match target { ContainerTarget::Docker { container_id } => docker::stop_container(container_id).await, - ContainerTarget::Agent(_) => Err(AGENT_LIFECYCLE.replace("{op}", "Stopping")), + ContainerTarget::Agent(ep) => super::aws_host::stop_endpoint(ep) + .await + .unwrap_or_else(|| Err(AGENT_LIFECYCLE.replace("{op}", "Stopping"))), + } + } + + /// Stop a project's container whether or not the app holds a target for + /// it. Docker: the stored container. AWS: the tunnel's task, else the + /// stored task ARN, else the project's live task — a project whose + /// tunnel could not be re-opened must still be stoppable. + pub async fn stop_project(&self, project: &Project, settings: &AppSettings) -> Result<(), String> { + match self { + ProjectHost::Docker => match ContainerTarget::for_project(project) { + Some(t) => self.stop(&t).await, + None => Ok(()), + }, + ProjectHost::Aws => super::aws_host::stop_project(project, settings).await, } } /// Stop (errors ignored — a stopped container is the goal) and remove. + /// An agent target is stopped; its resources go with + /// [`Self::remove_aws_resources`]. pub async fn remove(&self, target: &ContainerTarget) -> Result<(), String> { match target { ContainerTarget::Docker { container_id } => { let _ = docker::stop_container(container_id).await; docker::remove_container(container_id).await } - ContainerTarget::Agent(_) => Err(AGENT_LIFECYCLE.replace("{op}", "Removing")), + ContainerTarget::Agent(ep) => super::aws_host::stop_endpoint(ep) + .await + .unwrap_or_else(|| Err(AGENT_LIFECYCLE.replace("{op}", "Removing"))), + } + } + + /// AWS: stop the task and remove the project's resources — task + /// definition revisions, the secret (7-day recovery window) and, only + /// with `discard_data`, the EFS access point. Docker: nothing (its state + /// goes with [`Self::discard_project_state`]). + pub async fn remove_aws_resources( + &self, + project: &Project, + settings: &AppSettings, + discard_data: bool, + ) -> Result<(), String> { + match self { + ProjectHost::Docker => Ok(()), + ProjectHost::Aws => super::aws_host::remove_resources(project, settings, discard_data).await, } } @@ -107,6 +156,8 @@ impl ProjectHost { left.volumes = docker::remove_project_volumes(project).await; left } + // Removed by `remove_aws_resources`, which needs the settings. + ProjectHost::Aws => DiscardLeftovers::default(), } } @@ -118,6 +169,7 @@ impl ProjectHost { docker::remove_legacy_mcp_containers(&project.id).await; docker::remove_legacy_project_network(&project.id).await; } + ProjectHost::Aws => {} } } } diff --git a/app/src-tauri/src/host/mod.rs b/app/src-tauri/src/host/mod.rs index 2a1bf8f..b149f02 100644 --- a/app/src-tauri/src/host/mod.rs +++ b/app/src-tauri/src/host/mod.rs @@ -28,12 +28,11 @@ //! may still do so. pub mod agent; +pub mod aws_host; pub mod lifecycle; pub mod sessions; -// Nothing in the app opens an SSM tunnel until the AWS host lands (phase 5 of -// the AWS spec); its tests exercise it meanwhile. Remove this allow then. -#[allow(dead_code)] pub mod ssm; +pub mod tunnels; pub mod watch; pub use crate::docker::exec::{ @@ -154,11 +153,28 @@ impl ContainerTarget { } } - /// The target for a project's container, if it has one yet. + /// The target for a project's container, if it has one yet. An AWS + /// project's is its registered tunnel's agent endpoint (`tunnels.rs`): + /// none while no tunnel is open. pub fn for_project(project: &Project) -> Option { + if project.is_aws() { + return tunnels::get_endpoint(&project.id).map(ContainerTarget::Agent); + } project.container_id.as_deref().map(Self::docker) } + /// [`Self::for_project`], waiting (up to `aws_host::REATTACH_WAIT`) for an + /// AWS project whose tunnel is being re-opened in the background after a + /// restart. For a Docker project it is `for_project`. + pub async fn for_project_waiting(project: &Project) -> Option { + if project.is_aws() { + return tunnels::wait_endpoint(&project.id, aws_host::REATTACH_WAIT) + .await + .map(ContainerTarget::Agent); + } + Self::for_project(project) + } + /// The host-specific identifier (a Docker container id today). For logs /// and for keying per-container state; never parse it. pub fn id(&self) -> &str { diff --git a/app/src-tauri/src/host/ssm.rs b/app/src-tauri/src/host/ssm.rs index 43f35b8..a76e6d2 100644 --- a/app/src-tauri/src/host/ssm.rs +++ b/app/src-tauri/src/host/ssm.rs @@ -115,6 +115,10 @@ impl EcsTask { /// ARN), and check it can take a session now: the task is running and /// its ECS Exec agent is up. Fargate starts that agent a few seconds after /// the container, so "not ready yet" is a normal answer just after start. + /// + /// The AWS host builds its target from `aws::tasks::wait_ready`'s answer + /// instead (`aws_host::ecs_task`); this stays for the live tests. + #[allow(dead_code)] pub async fn resolve( ecs: &aws_sdk_ecs::Client, cluster: &str, @@ -175,6 +179,7 @@ impl EcsTask { /// The part of an ARN after the last `marker` (`.../cluster/name` → `name`), /// or the whole string when it is not an ARN. +#[allow(dead_code)] // `EcsTask::resolve` fn last_arn_segment<'a>(s: &'a str, marker: &str) -> &'a str { if !s.starts_with("arn:") { return s; @@ -606,6 +611,7 @@ pub enum TunnelState { pub struct PortForward { local_port: u16, state: watch::Receiver, + #[allow(dead_code)] // `plugin_pid()`, for diagnostics and the tests plugin_pid: watch::Receiver>, supervisor: Option>, _cancel: DropGuard, @@ -678,6 +684,7 @@ impl PortForward { } } + #[allow(dead_code)] // diagnostics and the tests (Task 8 may show the state) pub fn local_port(&self) -> u16 { self.local_port } @@ -688,11 +695,13 @@ impl PortForward { AgentEndpoint::loopback(self.local_port, token) } + #[allow(dead_code)] // diagnostics and the tests (Task 8 may show the state) pub fn state(&self) -> watch::Receiver { self.state.clone() } /// The running plugin's process id, if one is running. + #[allow(dead_code)] // diagnostics and the tests (Task 8 may show the state) pub fn plugin_pid(&self) -> Option { *self.plugin_pid.borrow() } diff --git a/app/src-tauri/src/host/tunnels.rs b/app/src-tauri/src/host/tunnels.rs new file mode 100644 index 0000000..3c71721 --- /dev/null +++ b/app/src-tauri/src/host/tunnels.rs @@ -0,0 +1,301 @@ +//! The live SSM tunnels to AWS projects' agents, by project id. +//! +//! An AWS project's [`ContainerTarget`](super::ContainerTarget) is an agent +//! endpoint at the loopback end of a [`PortForward`](super::ssm::PortForward); +//! this registry is where `ContainerTarget::for_project` finds it, where +//! `ProjectHost::stop` finds the task behind an endpoint, and what the +//! shutdown path closes (every forward runs a `session-manager-plugin` child +//! that would otherwise outlive the app — see `ssm.rs`). +//! +//! It also remembers which targets the AWS death watch found **stopped**, so +//! `ContainerTarget::has_exited` can answer for an agent target (R3): only +//! the host that runs the task can confirm a death, never the agent. + +use std::collections::{HashMap, HashSet}; +use std::sync::Mutex; +use std::time::Duration; + +use aws_config::SdkConfig; +use futures_util::future::BoxFuture; +use tokio::sync::watch; + +use super::agent::AgentEndpoint; +use super::ssm::{EcsTask, PortForward}; + +/// Something that owns a tunnel's resources: the real [`PortForward`], or a +/// test stand-in. +pub trait TunnelHandle: Send + Sync { + fn close(self: Box) -> BoxFuture<'static, ()>; +} + +impl TunnelHandle for PortForward { + fn close(self: Box) -> BoxFuture<'static, ()> { + Box::pin(async move { (*self).close().await }) + } +} + +/// One project's tunnel and the task it reaches. +pub struct LiveTunnel { + /// The SSM target (kept for diagnostics; the forward already uses it). + #[allow(dead_code)] + pub task: EcsTask, + pub task_arn: String, + pub cluster_arn: String, + pub connection_id: String, + pub region: String, + /// The connection's SDK config, so a stop needs no settings lookup. + pub sdk: SdkConfig, + pub forward: Box, + pub endpoint: AgentEndpoint, + /// Runtime handles for the store (`set_aws_runtime`). + pub access_point_id: Option, + pub task_definition_arn: String, +} + +/// Where an endpoint's task is: `(connection, region, cluster, task ARN)`. +pub type TaskOf = (String, String, String, String); + +#[derive(Default)] +struct Registry { + live: HashMap, + /// Target ids (endpoint base URLs) whose task the AWS watch saw STOPPED. + stopped: HashSet, + /// Projects whose tunnel is being re-opened in the background. + pending: HashSet, +} + +fn registry() -> &'static Mutex { + static R: std::sync::OnceLock> = std::sync::OnceLock::new(); + R.get_or_init(Default::default) +} + +fn lock() -> std::sync::MutexGuard<'static, Registry> { + registry().lock().unwrap_or_else(|e| e.into_inner()) +} + +/// Wakes [`wait_endpoint`] callers when the registry changes. +fn changed() -> &'static watch::Sender { + static C: std::sync::OnceLock> = std::sync::OnceLock::new(); + C.get_or_init(|| watch::channel(0).0) +} + +fn notify() { + changed().send_modify(|n| *n = n.wrapping_add(1)); +} + +/// Register `tunnel` for `project_id`, returning the one it replaces (the +/// caller closes it). +pub fn register(project_id: &str, tunnel: LiveTunnel) -> Option { + let old = { + let mut r = lock(); + r.stopped.remove(tunnel.endpoint.base_url.as_str()); + r.pending.remove(project_id); + r.live.insert(project_id.to_string(), tunnel) + }; + notify(); + old +} + +pub fn get_endpoint(project_id: &str) -> Option { + lock().live.get(project_id).map(|t| t.endpoint.clone()) +} + +/// `(task ARN, access point id, task definition ARN)` of the project's tunnel. +pub fn runtime(project_id: &str) -> Option<(String, Option, String)> { + lock().live.get(project_id).map(|t| { + ( + t.task_arn.clone(), + t.access_point_id.clone(), + t.task_definition_arn.clone(), + ) + }) +} + +/// The project whose tunnel is `ep` and its task. +pub fn task_of(ep: &AgentEndpoint) -> Option { + lock().live.values().find(|t| t.endpoint.base_url == ep.base_url).map(|t| { + ( + t.connection_id.clone(), + t.region.clone(), + t.cluster_arn.clone(), + t.task_arn.clone(), + ) + }) +} + +/// The project id and SDK config behind `ep`. +pub(crate) fn owner_of(ep: &AgentEndpoint) -> Option<(String, SdkConfig)> { + lock() + .live + .iter() + .find(|(_, t)| t.endpoint.base_url == ep.base_url) + .map(|(id, t)| (id.clone(), t.sdk.clone())) +} + +/// Take the project's tunnel out and close it. +pub async fn close(project_id: &str) { + let taken = { + let mut r = lock(); + r.pending.remove(project_id); + let t = r.live.remove(project_id); + if let Some(ref t) = t { + r.stopped.remove(t.endpoint.base_url.as_str()); + } + t + }; + notify(); + if let Some(t) = taken { + t.forward.close().await; + } +} + +/// Close every tunnel, concurrently. The shutdown path's step. +pub async fn close_all() { + let all: Vec = { + let mut r = lock(); + r.pending.clear(); + r.stopped.clear(); + r.live.drain().map(|(_, t)| t).collect() + }; + notify(); + futures_util::future::join_all(all.into_iter().map(|t| t.forward.close())).await; +} + +/// The AWS watch saw the task behind target `target_id` STOPPED. +pub fn mark_stopped(target_id: &str) { + lock().stopped.insert(target_id.to_string()); +} + +/// Whether the AWS watch has confirmed the task behind `target_id` stopped. +pub fn is_stopped(target_id: &str) -> bool { + lock().stopped.contains(target_id) +} + +/// A background re-attach for this project has begun / ended without a tunnel. +pub fn set_pending(project_id: &str, pending: bool) { + { + let mut r = lock(); + if pending { + r.pending.insert(project_id.to_string()); + } else { + r.pending.remove(project_id); + } + } + notify(); +} + +pub fn is_pending(project_id: &str) -> bool { + lock().pending.contains(project_id) +} + +/// The project's endpoint, waiting up to `timeout` while a background +/// re-attach is still opening it. `None` when there is none. +pub async fn wait_endpoint(project_id: &str, timeout: Duration) -> Option { + let mut rx = changed().subscribe(); + let deadline = tokio::time::Instant::now() + timeout; + loop { + if let Some(ep) = get_endpoint(project_id) { + return Some(ep); + } + if !is_pending(project_id) { + return None; + } + if tokio::time::timeout_at(deadline, rx.changed()).await.is_err() { + return None; + } + } +} + +#[cfg(test)] +pub(crate) mod tests { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + + /// A stand-in tunnel that counts its closes. + pub(crate) struct FakeTunnel(pub Arc); + impl TunnelHandle for FakeTunnel { + fn close(self: Box) -> BoxFuture<'static, ()> { + let n = self.0.clone(); + Box::pin(async move { + n.fetch_add(1, Ordering::SeqCst); + }) + } + } + + pub(crate) fn fake_tunnel(port: u16, closes: Arc) -> LiveTunnel { + LiveTunnel { + task: EcsTask { + cluster: "triple-c-dev".into(), + task_id: format!("t{port}"), + runtime_id: "r".into(), + }, + task_arn: format!("arn:aws:ecs:eu-west-1:111122223333:task/triple-c-dev/t{port}"), + cluster_arn: "arn:aws:ecs:eu-west-1:111122223333:cluster/triple-c-dev".into(), + connection_id: "c1".into(), + region: "eu-west-1".into(), + sdk: SdkConfig::builder().build(), + forward: Box::new(FakeTunnel(closes)), + endpoint: AgentEndpoint::loopback(port, "tok"), + access_point_id: Some("fsap-1".into()), + task_definition_arn: "arn:td:1".into(), + } + } + + fn pid(tag: &str) -> String { + format!("tunnels-test-{tag}-{}", uuid::Uuid::new_v4()) + } + + #[tokio::test] + async fn a_registered_tunnel_is_found_by_project_and_by_endpoint_and_closed_once() { + let closes = Arc::new(AtomicUsize::new(0)); + let p = pid("reg"); + assert!(get_endpoint(&p).is_none()); + assert!(register(&p, fake_tunnel(61001, closes.clone())).is_none()); + let ep = get_endpoint(&p).unwrap(); + assert_eq!(ep, AgentEndpoint::loopback(61001, "tok")); + let (conn, region, _, arn) = task_of(&ep).unwrap(); + assert_eq!((conn.as_str(), region.as_str()), ("c1", "eu-west-1")); + assert!(arn.ends_with("/t61001")); + assert_eq!(runtime(&p).unwrap().1.as_deref(), Some("fsap-1")); + close(&p).await; + close(&p).await; + assert_eq!(closes.load(Ordering::SeqCst), 1); + assert!(get_endpoint(&p).is_none() && task_of(&ep).is_none()); + } + + #[tokio::test] + async fn re_registering_hands_back_the_old_tunnel_and_clears_a_stale_stop() { + let closes = Arc::new(AtomicUsize::new(0)); + let p = pid("rereg"); + register(&p, fake_tunnel(61002, closes.clone())); + let id = AgentEndpoint::loopback(61003, "tok").base_url.to_string(); + mark_stopped(&id); + let old = register(&p, fake_tunnel(61003, closes.clone())).unwrap(); + assert_eq!(old.endpoint.base_url.port(), Some(61002)); + assert!(!is_stopped(&id), "a new task on a reused port is not stopped"); + old.forward.close().await; + close(&p).await; + assert_eq!(closes.load(Ordering::SeqCst), 2); + } + + #[tokio::test] + async fn waiting_for_an_endpoint_ends_when_it_is_registered_or_the_reattach_gives_up() { + let p = pid("wait"); + assert!(wait_endpoint(&p, Duration::from_secs(5)).await.is_none(), "nothing pending"); + set_pending(&p, true); + let p2 = p.clone(); + let waiter = tokio::spawn(async move { wait_endpoint(&p2, Duration::from_secs(5)).await }); + tokio::time::sleep(Duration::from_millis(20)).await; + register(&p, fake_tunnel(61004, Arc::new(AtomicUsize::new(0)))); + assert_eq!(waiter.await.unwrap().unwrap().base_url.port(), Some(61004)); + close(&p).await; + + set_pending(&p, true); + let p2 = p.clone(); + let waiter = tokio::spawn(async move { wait_endpoint(&p2, Duration::from_secs(5)).await }); + tokio::time::sleep(Duration::from_millis(20)).await; + set_pending(&p, false); + assert!(waiter.await.unwrap().is_none()); + } +} diff --git a/app/src-tauri/src/host/watch.rs b/app/src-tauri/src/host/watch.rs index 837dfb1..d269268 100644 --- a/app/src-tauri/src/host/watch.rs +++ b/app/src-tauri/src/host/watch.rs @@ -27,8 +27,6 @@ pub const AWS_POLL_INTERVAL: Duration = Duration::from_secs(15); #[derive(Clone)] pub enum ExitSource { Docker, - // Constructed by the AWS host (phase 5, Task 6); until then only by tests. - #[allow(dead_code)] Aws(AwsWatch), } @@ -80,7 +78,6 @@ pub struct AwsWatch { interval: Duration, } -#[allow(dead_code)] // Task 6 impl AwsWatch { pub fn new(lookup: Arc) -> Self { Self::every(lookup, AWS_POLL_INTERVAL) @@ -180,9 +177,10 @@ impl ContainerTarget { ContainerTarget::Docker { container_id } => docker::events::has_exited(container_id).await, // The agent can only say "I answer" or "I don't"; silence is not a // positive exit (an SSM blip looks the same). Only the host that - // runs the container can confirm a death — an AWS host answers - // from ECS — so through the agent alone this never confirms one. - ContainerTarget::Agent(_) => false, + // runs the container can confirm a death: the AWS watch records + // the targets whose task ECS reports STOPPED in the tunnel + // registry (R3), and nothing else ever confirms one. + ContainerTarget::Agent(ep) => super::tunnels::is_stopped(ep.base_url.as_str()), } } } diff --git a/app/src-tauri/src/lib.rs b/app/src-tauri/src/lib.rs index 10569a0..d199661 100644 --- a/app/src-tauri/src/lib.rs +++ b/app/src-tauri/src/lib.rs @@ -332,6 +332,19 @@ pub fn run() { crate::host::watch::ExitSource::Docker, lifecycle_setup.cancellation(), )); + // The same for AWS projects: a DescribeTasks poll every 15 s over + // the tasks of Running AWS projects (none: no AWS calls at all). + lifecycle_setup.track(container_watch::spawn( + app.handle().clone(), + projects_store_setup.clone(), + crate::host::watch::ExitSource::Aws(crate::host::watch::AwsWatch::new( + std::sync::Arc::new(crate::host::aws_host::StoreTaskLookup { + projects: projects_store_setup.clone(), + settings: settings_store_setup.clone(), + }), + )), + lifecycle_setup.cancellation(), + )); // Marketplaces: refresh each once at startup, in the background. // Failures are logged, not toasted — the Marketplace tab shows them. @@ -509,6 +522,9 @@ pub fn run() { exec_manager.close_all_sessions(), auth_bridge.stop_all(), browser_view::manager().stop_all(), + // Each AWS tunnel is a session-manager-plugin + // child that `exit` would not end. + crate::host::tunnels::close_all(), ); }; diff --git a/app/src-tauri/src/web_terminal/ws_handler.rs b/app/src-tauri/src/web_terminal/ws_handler.rs index 8d60578..df85e36 100644 --- a/app/src-tauri/src/web_terminal/ws_handler.rs +++ b/app/src-tauri/src/web_terminal/ws_handler.rs @@ -295,7 +295,8 @@ async fn handle_open( return Err(format!("Project '{}' is not running", project.name)); } - let container = crate::host::ContainerTarget::for_project(&project) + let container = crate::host::ContainerTarget::for_project_waiting(&project) + .await .ok_or_else(|| "Container not running".to_string())?; let cmd = match session_type {