From 48c60ef89c166f682b19c6367c590540d2109482 Mon Sep 17 00:00:00 2001 From: Josh Knapp Date: Fri, 13 Nov 2026 12:02:35 -0800 Subject: [PATCH] Add the app-side triple-c-agent client as ContainerTarget::Agent host/agent.rs implements every row of the protocol spec's client mapping over reqwest (HTTP) and tokio-tungstenite (attach WebSocket): one-shot execs rebuilt into docker/exec.rs's own OneshotOutput (413 passed through verbatim), detached execs and exit polling with Docker's budget, attach with stdout/stderr channel mapping, exit-frame recording and close_stdin on shutdown, resize, uploads, streamed archive download and health. AgentEndpoint's Debug redacts the token, and no error carries it. docker/exec.rs shares its host-file reader, uid probe, one-shot reassembly and exit-poll budget instead of the agent duplicating them; Docker behaviour is unchanged. Lifecycle stop/remove refuse an agent target, and has_exited is false for one (only the host can confirm). Tests run the client against a small fake agent built with axum. Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 6 +- app/src-tauri/Cargo.lock | 2 + app/src-tauri/Cargo.toml | 7 + app/src-tauri/src/docker/exec.rs | 183 +++--- app/src-tauri/src/host/agent.rs | 824 ++++++++++++++++++++++++++ app/src-tauri/src/host/agent/tests.rs | 560 +++++++++++++++++ app/src-tauri/src/host/lifecycle.rs | 9 + app/src-tauri/src/host/mod.rs | 46 +- app/src-tauri/src/host/watch.rs | 5 + 9 files changed, 1563 insertions(+), 79 deletions(-) create mode 100644 app/src-tauri/src/host/agent.rs create mode 100644 app/src-tauri/src/host/agent/tests.rs diff --git a/CLAUDE.md b/CLAUDE.md index 439ac46..b5665ae 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -283,8 +283,12 @@ docker exec stdout → tokio task → emit("terminal-output-{sessionId}") → li feature (containers labelled `triple-c.mcp-server`, `triple-c-net-*` networks). Deletable once users have migrated. - **`host/`** — Where a project's container runs, and the only way into it. `ContainerTarget` - (one `Docker` variant today; AWS planned — `docs/superpowers/specs/2026-11-13-aws-ecs-host-design.md`) + (`Docker`, and `Agent` — reached through the in-container `triple-c-agent` by `agent.rs`, per + `docs/superpowers/specs/2026-11-13-triple-c-agent-protocol.md`; AWS planned — + `docs/superpowers/specs/2026-11-13-aws-ecs-host-design.md`) wraps execs, file transfer and `is_running`; `sessions.rs` holds `ExecSessionManager`. + `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 `get_docker()`** — `host::tests::no_container_interior_bollard_outside_the_host_layer` fails the build on it. `lifecycle.rs`'s `ProjectHost` owns create/start/stop/find/remove and diff --git a/app/src-tauri/Cargo.lock b/app/src-tauri/Cargo.lock index 27b38b6..663962f 100644 --- a/app/src-tauri/Cargo.lock +++ b/app/src-tauri/Cargo.lock @@ -6642,6 +6642,8 @@ dependencies = [ "tauri-plugin-opener", "tempfile", "tokio", + "tokio-tungstenite", + "tokio-util", "tower-http", "url", "uuid", diff --git a/app/src-tauri/Cargo.toml b/app/src-tauri/Cargo.toml index d2f8c1d..8579095 100644 --- a/app/src-tauri/Cargo.toml +++ b/app/src-tauri/Cargo.toml @@ -33,6 +33,13 @@ reqwest = { version = "0.12", default-features = false, features = ["json", "rus iana-time-zone = "0.1" sha2 = "0.10" axum = { version = "0.8", features = ["ws"] } +# Client side of triple-c-agent's two WebSocket endpoints (host/agent.rs). The +# same version axum's `ws` already pulls in; plain ws:// only (the transport is +# loopback or an SSM port-forward), so no TLS features. +tokio-tungstenite = { version = "0.28", default-features = false, features = ["connect"] } +# `PollSender`: the agent attach's stdin `AsyncWrite` over a bounded channel. +# Already in the tree through tokio's ecosystem. +tokio-util = "0.7" tower-http = { version = "0.6", features = ["cors"] } base64 = "0.22" rand = "0.9" diff --git a/app/src-tauri/src/docker/exec.rs b/app/src-tauri/src/docker/exec.rs index 3a7ab60..19a23f4 100644 --- a/app/src-tauri/src/docker/exec.rs +++ b/app/src-tauri/src/docker/exec.rs @@ -206,31 +206,8 @@ pub async fn upload_host_file_with_ids( let mtime = now_epoch_secs(); let tar_buf = tokio::task::spawn_blocking(move || -> Result, String> { - // The caller resolved this path (`resolve_host_read_path`); opening it - // is a second trip through the same directories, so the descriptor is - // checked against the path that was validated before its bytes are - // packed into anything. Two paths reach here: the terminal's drop - // target, and the Files pane's upload via `upload_host_file_with_ids`. - // Between them they are how host bytes enter a container. - let file = std::fs::File::open(&host_path) - .map_err(|e| format!("Failed to read {}: {}", host_path, e))?; - crate::commands::file_commands::verify_opened_path( - &file, - std::path::Path::new(&host_path), - )?; - let mut data = Vec::new(); - std::io::Read::read_to_end( - &mut std::io::Read::take(file, MAX_DROP_BYTES.saturating_add(1)), - &mut data, - ) - .map_err(|e| format!("Failed to read {}: {}", host_path, e))?; - if data.len() as u64 > MAX_DROP_BYTES { - return Err(format!( - "File too large to upload (limit {} MB)", - MAX_DROP_BYTES / (1024 * 1024) - )); - } - build_single_file_tar(&dest_for_blk, &data[..], 0o644, uid, gid, mtime) + let data = read_host_file_for_upload(&host_path)?; + build_single_file_tar(&dest_for_blk, &data[..], HOST_FILE_UPLOAD_MODE, uid, gid, mtime) }) .await .map_err(|e| format!("Upload task panicked: {}", e))??; @@ -251,13 +228,46 @@ pub async fn upload_host_file_with_ids( Ok(container_join(dest_dir, &dest_name)) } +/// The mode a host file lands with in the container, on every host. +pub(crate) const HOST_FILE_UPLOAD_MODE: u32 = 0o644; + +/// Read a host file that is about to be copied into a container. Blocking: +/// call it from `spawn_blocking`. Shared by every host's upload so the checks +/// cannot drift apart. +/// +/// The caller resolved this path (`resolve_host_read_path`); opening it is a +/// second trip through the same directories, so the descriptor is checked +/// against the path that was validated before its bytes are packed into +/// anything. Two paths reach here: the terminal's drop target, and the Files +/// pane's upload via `upload_host_file_with_ids`. Between them they are how +/// host bytes enter a container. [`MAX_DROP_BYTES`] is applied to what the +/// open descriptor actually yields, not to an earlier `stat`. +pub(crate) fn read_host_file_for_upload(host_path: &str) -> Result, String> { + let file = std::fs::File::open(host_path) + .map_err(|e| format!("Failed to read {}: {}", host_path, e))?; + crate::commands::file_commands::verify_opened_path(&file, std::path::Path::new(host_path))?; + let mut data = Vec::new(); + std::io::Read::read_to_end( + &mut std::io::Read::take(file, MAX_DROP_BYTES.saturating_add(1)), + &mut data, + ) + .map_err(|e| format!("Failed to read {}: {}", host_path, e))?; + if data.len() as u64 > MAX_DROP_BYTES { + return Err(format!( + "File too large to upload (limit {} MB)", + MAX_DROP_BYTES / (1024 * 1024) + )); + } + Ok(data) +} + /// Join a container directory to a name that may itself carry separators. /// /// Only the *reported* path — the bytes have already landed by the time this is /// called — but that path is what the terminal echoes and what the Files pane /// puts in its toast, so `/tmp//x` reading back as a different file than `/tmp/x` /// is worth the four lines. `"/"` trims to `""` and yields `/x`. -fn container_join(dir: &str, name: &str) -> String { +pub(crate) fn container_join(dir: &str, name: &str) -> String { format!("{}/{}", dir.trim_end_matches('/'), name.trim_start_matches('/')) } @@ -362,12 +372,26 @@ pub async fn container_user_ids(container_id: &str) -> (u64, u64) { let (out, _) = exec_oneshot_limited( container_id, "claude", - vec!["sh".to_string(), "-c".to_string(), "id -u; id -g".to_string()], - 256, + user_ids_cmd(), + USER_IDS_OUTPUT_LIMIT, ) .await .unwrap_or_default(); + parse_user_ids(&out) +} +/// The command [`container_user_ids`] runs (as `claude`, under +/// [`USER_IDS_OUTPUT_LIMIT`]); every host asks the same question. +pub(crate) fn user_ids_cmd() -> Vec { + vec!["sh".to_string(), "-c".to_string(), "id -u; id -g".to_string()] +} + +/// Output ceiling for the uid probe: two short numbers. +pub(crate) const USER_IDS_OUTPUT_LIMIT: usize = 256; + +/// `id -u; id -g` output -> `(uid, gid)`, falling back to the image's +/// build-time `(1000, 1000)`. +pub(crate) fn parse_user_ids(out: &str) -> (u64, u64) { let mut ids = out.lines().filter_map(|l| l.trim().parse::().ok()); match (ids.next(), ids.next()) { (Some(uid), Some(gid)) => (uid, gid), @@ -495,21 +519,54 @@ pub async fn exec_oneshot_as( /// fail". `stdout_ranges` indexes the parts of it that came from stdout, so a /// caller that is *parsing* output can have just that without the buffer being /// held twice. -struct OneshotOutput { +pub(crate) struct OneshotOutput { combined: Vec, stdout_ranges: Vec<(usize, usize)>, - exit_code: i64, + pub(crate) exit_code: i64, +} + +/// The refusal for output past `limit`, starting with [`OUTPUT_LIMIT_MARKER`]. +pub(crate) fn output_limit_error(limit: usize) -> String { + format!( + "{}: Command output exceeded {} bytes and was abandoned", + OUTPUT_LIMIT_MARKER, limit + ) } impl OneshotOutput { + pub(crate) fn new() -> Self { + Self { + combined: Vec::new(), + stdout_ranges: Vec::new(), + exit_code: 0, + } + } + + /// Append one chunk in arrival order, or refuse it (appending nothing) + /// once the combined output would pass `limit`. The caller must stop + /// reading on an error rather than truncate silently: every caller parses + /// this output, and a half-read manifest or JSON array is worse than an + /// error. + pub(crate) fn push(&mut self, from_stdout: bool, bytes: &[u8], limit: usize) -> Result<(), String> { + match push_capped(&mut self.combined, bytes, limit) { + Some(range) => { + if from_stdout { + self.stdout_ranges.push(range); + } + Ok(()) + } + None => Err(output_limit_error(limit)), + } + } + /// Everything the command printed, in the order it printed it. - fn text(&self) -> String { + pub(crate) fn text(&self) -> String { String::from_utf8_lossy(&self.combined).into_owned() } /// stdout alone — for callers that parse it, where a diagnostic spliced in /// mid-record is a parse error at best. - fn stdout(&self) -> String { + pub(crate) fn stdout(&self) -> String { let mut out = Vec::with_capacity(self.combined.len()); for (start, end) in &self.stdout_ranges { out.extend_from_slice(&self.combined[*start..*end]); @@ -518,7 +575,7 @@ impl OneshotOutput { } /// stderr alone — the complement of [`Self::stdout`], i.e. the diagnostics. - fn stderr(&self) -> String { + pub(crate) fn stderr(&self) -> String { let mut out = Vec::with_capacity(self.combined.len()); let mut cursor = 0usize; for (start, end) in &self.stdout_ranges { @@ -587,8 +644,7 @@ async fn exec_oneshot_raw( .await .map_err(|e| format!("Failed to start exec: {}", e))?; - let mut combined: Vec = Vec::new(); - let mut stdout_ranges: Vec<(usize, usize)> = Vec::new(); + let mut out = OneshotOutput::new(); match result { StartExecResults::Attached { mut output, .. } => { while let Some(msg) = output.next().await { @@ -596,23 +652,9 @@ async fn exec_oneshot_raw( Ok(data) => { let from_stdout = matches!(data, LogOutput::StdOut { .. }); let bytes = data.into_bytes(); - match push_capped(&mut combined, &bytes, limit) { - Some(range) => { - if from_stdout { - stdout_ranges.push(range); - } - } - // Stop reading rather than truncate silently: every - // caller parses this output, and a half-read - // manifest or JSON array is worse than an error. - // Dropping `output` kills the exec's stream. - None => { - return Err(format!( - "{}: Command output exceeded {} bytes and was abandoned", - OUTPUT_LIMIT_MARKER, limit - )) - } - } + // Over the limit: returning drops `output`, which + // kills the exec's stream. + out.push(from_stdout, &bytes, limit)?; } Err(e) => return Err(format!("Exec output error: {}", e)), } @@ -623,13 +665,8 @@ async fn exec_oneshot_raw( // The output stream draining doesn't strictly guarantee inspect_exec has the // final exit_code populated yet, so poll until the exec reports finished. - let exit_code = require_exit_code(wait_for_exec_exit(&exec.id).await)?; - - Ok(OneshotOutput { - combined, - stdout_ranges, - exit_code, - }) + out.exit_code = require_exit_code(wait_for_exec_exit(&exec.id).await)?; + Ok(out) } /// Turn "the exit code could not be determined" into an error rather than a 0. @@ -662,7 +699,7 @@ fn require_exit_code(code: Option) -> Result { /// spurious "the rename failed" and a real one. pub async fn wait_for_exec_exit(exec_id: &str) -> Option { let docker = get_docker().ok()?; - for _ in 0..200 { + for _ in 0..EXEC_EXIT_POLLS { match docker.inspect_exec(exec_id).await { Ok(info) => { if info.running != Some(true) { @@ -681,30 +718,27 @@ pub async fn wait_for_exec_exit(exec_id: &str) -> Option { } Err(_) => return None, } - tokio::time::sleep(std::time::Duration::from_millis(25)).await; + tokio::time::sleep(EXEC_EXIT_POLL_INTERVAL).await; } None } +/// [`wait_for_exec_exit`]'s budget: polls, and the pause between them. Every +/// host's `wait_exec_exit` uses the same numbers. +pub(crate) const EXEC_EXIT_POLLS: usize = 200; +pub(crate) const EXEC_EXIT_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_millis(25); + #[cfg(test)] mod tests { use super::*; /// The frames a demultiplexed exec hands back, as `(is_stdout, bytes)`. fn collect(frames: &[(bool, &[u8])]) -> OneshotOutput { - let mut combined = Vec::new(); - let mut stdout_ranges = Vec::new(); + let mut out = OneshotOutput::new(); for (from_stdout, bytes) in frames { - let range = push_capped(&mut combined, bytes, usize::MAX).unwrap(); - if *from_stdout { - stdout_ranges.push(range); - } - } - OneshotOutput { - combined, - stdout_ranges, - exit_code: 0, + out.push(*from_stdout, bytes, usize::MAX).unwrap(); } + out } #[test] @@ -763,10 +797,7 @@ mod tests { // The marker is what lets `list_container_files` say "too many entries" // instead, which is the thing that actually happened. assert!(!OUTPUT_LIMIT_MARKER.is_empty()); - let refusal = format!( - "{}: Command output exceeded {} bytes and was abandoned", - OUTPUT_LIMIT_MARKER, MAX_ONESHOT_OUTPUT - ); + let refusal = output_limit_error(MAX_ONESHOT_OUTPUT); assert!(refusal.starts_with(OUTPUT_LIMIT_MARKER)); } diff --git a/app/src-tauri/src/host/agent.rs b/app/src-tauri/src/host/agent.rs new file mode 100644 index 0000000..80837fb --- /dev/null +++ b/app/src-tauri/src/host/agent.rs @@ -0,0 +1,824 @@ +//! The app's side of `triple-c-agent`, the small server that runs inside a +//! project container and does over HTTP what the Docker variant does through +//! the Docker API. +//! +//! Wire protocol: `docs/superpowers/specs/2026-11-13-triple-c-agent-protocol.md`. +//! Its "Client mapping" table is what this file implements, one function per +//! [`ContainerTarget`](super::ContainerTarget) method. The contract is Docker +//! parity: every function here returns exactly what its `docker/exec.rs` +//! counterpart returns for the same command, including the shape of one-shot +//! output (combined in arrival order, or stdout/stderr split), the +//! `OUTPUT_LIMIT` refusal text, and the path an upload reports. +//! +//! The token never leaves this file except as the `Authorization` header: +//! [`AgentEndpoint`]'s `Debug` redacts it and no error string is built from a +//! request (reqwest and tungstenite errors name the URL, which carries no +//! secret, never the headers). + +use std::collections::{HashMap, VecDeque}; +use std::io; +use std::pin::Pin; +use std::sync::{Mutex, OnceLock}; +use std::task::{Context, Poll}; +use std::time::Duration; + +use base64::Engine; +use bytes::Bytes; +use futures_util::{SinkExt, StreamExt}; +use serde::Deserialize; +use tokio::io::AsyncWrite; +use tokio_tungstenite::tungstenite::client::IntoClientRequest; +use tokio_tungstenite::tungstenite::http::HeaderValue; +use tokio_tungstenite::tungstenite::Message; +use tokio_util::sync::PollSender; + +use super::{ArchiveStream, AttachedExec, ExecChunk, ExecSpec}; +use crate::docker::exec::{ + container_join, output_limit_error, parse_user_ids, read_host_file_for_upload, user_ids_cmd, + OneshotOutput, EXEC_EXIT_POLLS, EXEC_EXIT_POLL_INTERVAL, HOST_FILE_UPLOAD_MODE, + MAX_ONESHOT_OUTPUT, USER_IDS_OUTPUT_LIMIT, +}; + +/// Connecting (TCP, and the WebSocket handshake) must not hang a caller on an +/// agent that is not there. Nothing else has an overall timeout: a one-shot +/// exec, an attach or an archive download can legitimately run for minutes. +const CONNECT_TIMEOUT: Duration = Duration::from_secs(5); +/// `is_running` is a liveness probe; it answers quickly or says "no". +const HEALTH_TIMEOUT: Duration = Duration::from_secs(3); +/// How long an attach waits for the agent's `started` reply after the +/// handshake. Spawning a process is fast; this only bounds a wedged agent. +const ATTACH_START_TIMEOUT: Duration = Duration::from_secs(30); +/// In-flight stdin frames per attach before `poll_write` applies backpressure. +const STDIN_QUEUE: usize = 64; + +/// Where an agent listens and the bearer token it wants. +/// +/// `base_url` is `http://127.0.0.1:/`: the local end of an SSM +/// port-forward on AWS, or a published loopback port in the local harness. +#[derive(Clone, PartialEq, Eq, Hash)] +pub struct AgentEndpoint { + pub base_url: url::Url, + token: String, +} + +impl std::fmt::Debug for AgentEndpoint { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("AgentEndpoint") + .field("base_url", &self.base_url.as_str()) + .field("token", &"") + .finish() + } +} + +impl AgentEndpoint { + /// An endpoint at `base_url`, which must be a plain `http://` URL with a + /// host. A trailing `/` is added so endpoint paths join beneath it. + pub fn new(mut base_url: url::Url, token: impl Into) -> Result { + if base_url.scheme() != "http" || base_url.host().is_none() { + return Err(format!( + "An agent endpoint must be an http:// URL with a host, not {}", + base_url + )); + } + if !base_url.path().ends_with('/') { + let path = format!("{}/", base_url.path()); + base_url.set_path(&path); + } + base_url.set_query(None); + base_url.set_fragment(None); + Ok(Self { + base_url, + token: token.into(), + }) + } + + /// The agent on `127.0.0.1:`. + pub fn loopback(port: u16, token: impl Into) -> Self { + let url = url::Url::parse(&format!("http://127.0.0.1:{}/", port)) + .expect("a loopback URL always parses"); + Self::new(url, token).expect("a loopback URL is a valid endpoint") + } + + fn url(&self, path: &str) -> url::Url { + self.base_url + .join(path) + .expect("endpoint paths are fixed relative paths") + } + + fn ws_url(&self, path: &str) -> url::Url { + let mut u = self.url(path); + u.set_scheme("ws").expect("http -> ws is a valid scheme change"); + u + } + + fn bearer(&self) -> String { + format!("Bearer {}", self.token) + } +} + +// ── HTTP plumbing ──────────────────────────────────────────────────────────── + +/// One client for every agent: connection pooling is per host, and building a +/// client per call would redo TLS-less setup for nothing. No proxy — the agent +/// is always on loopback, and a corporate `HTTP_PROXY` must not capture it. +fn http() -> &'static reqwest::Client { + static CLIENT: OnceLock = OnceLock::new(); + CLIENT.get_or_init(|| { + reqwest::Client::builder() + .connect_timeout(CONNECT_TIMEOUT) + .no_proxy() + .build() + .expect("a reqwest client with no TLS config always builds") + }) +} + +#[derive(Deserialize)] +struct ErrorBody { + error: String, +} + +/// The agent's `{"error": …}` text, if the body has one. +async fn error_text(resp: reqwest::Response) -> Option { + resp.json::().await.ok().map(|b| b.error) +} + +/// `": "` for a non-success response. +async fn status_error(context: &str, resp: reqwest::Response) -> String { + let status = resp.status(); + if status == reqwest::StatusCode::UNAUTHORIZED { + return format!("{}: the agent rejected the credentials (401)", context); + } + match error_text(resp).await { + Some(e) => format!("{}: {}", context, e), + None => format!("{}: agent returned HTTP {}", context, status.as_u16()), + } +} + +async fn send( + context: &str, + req: reqwest::RequestBuilder, +) -> Result { + req.send().await.map_err(|e| format!("{}: {}", context, e)) +} + +/// Read a response body into memory, refusing past `cap` bytes. +async fn read_capped(mut resp: reqwest::Response, cap: usize) -> Result>, String> { + let mut body = Vec::new(); + while let Some(chunk) = resp + .chunk() + .await + .map_err(|e| format!("Exec output error: {}", e))? + { + if body.len() + chunk.len() > cap { + return Ok(None); + } + body.extend_from_slice(&chunk); + } + Ok(Some(body)) +} + +/// ExecSpec as the protocol spells it. +fn spec_json(spec: &ExecSpec) -> serde_json::Map { + let v = serde_json::json!({ + "cmd": spec.cmd, + "user": spec.user, + "working_dir": spec.working_dir, + "env": spec.env, + "tty": spec.tty, + }); + match v { + serde_json::Value::Object(m) => m, + _ => unreachable!("json! of an object literal is an object"), + } +} + +// ── one-shot execs ─────────────────────────────────────────────────────────── + +#[derive(Deserialize)] +struct OneshotChunk { + s: String, + d: String, +} + +#[derive(Deserialize)] +struct OneshotResponse { + chunks: Vec, + exit_code: i64, +} + +/// `POST /v1/exec`, reassembled into the same [`OneshotOutput`] the Docker +/// path builds. Like Docker's one-shot, there is no tty, stdin is empty, and +/// the working directory is the container's default. +async fn exec_oneshot_raw( + ep: &AgentEndpoint, + user: &str, + cmd: Vec, + env: Vec, + limit: usize, +) -> Result { + const CONTEXT: &str = "Failed to start exec"; + let spec = ExecSpec::new(cmd).user(user).working_dir(None).env(env).tty(false); + let mut body = spec_json(&spec); + body.insert("limit".into(), serde_json::json!(limit)); + + let resp = send( + CONTEXT, + http() + .post(ep.url("v1/exec")) + .header(reqwest::header::AUTHORIZATION, ep.bearer()) + .json(&body), + ) + .await?; + + let status = resp.status(); + if status == reqwest::StatusCode::PAYLOAD_TOO_LARGE { + // Verbatim: it starts with OUTPUT_LIMIT_MARKER, which callers match. + return Err(error_text(resp) + .await + .unwrap_or_else(|| output_limit_error(limit))); + } + if !status.is_success() { + return Err(status_error(CONTEXT, resp).await); + } + + // The agent enforces `limit`, but this is container-controlled input + // being buffered into the host, so the client bounds it too: base64 is + // 4/3 of the payload and each chunk carries a few bytes of JSON, so a + // body past this is over the limit however it was chunked. + let cap = limit.saturating_mul(8).saturating_add(64 * 1024); + let raw = read_capped(resp, cap) + .await? + .ok_or_else(|| output_limit_error(limit))?; + let parsed: OneshotResponse = serde_json::from_slice(&raw) + .map_err(|e| format!("Exec output error: malformed agent response: {}", e))?; + + let mut out = OneshotOutput::new(); + let b64 = base64::engine::general_purpose::STANDARD; + for chunk in parsed.chunks { + let from_stdout = match chunk.s.as_str() { + "out" => true, + "err" => false, + other => { + return Err(format!( + "Exec output error: unknown stream {:?} in agent response", + other + )) + } + }; + let bytes = b64 + .decode(chunk.d.as_bytes()) + .map_err(|e| format!("Exec output error: bad base64 from agent: {}", e))?; + out.push(from_stdout, &bytes, limit)?; + } + out.exit_code = parsed.exit_code; + Ok(out) +} + +/// Combined output and exit code under `limit` — `exec_oneshot_limited` and, +/// with [`MAX_ONESHOT_OUTPUT`], every other combined-output variant. +pub async fn exec_oneshot_limited( + ep: &AgentEndpoint, + user: &str, + cmd: Vec, + env: Vec, + limit: usize, +) -> Result<(String, i64), String> { + let out = exec_oneshot_raw(ep, user, cmd, env, limit).await?; + Ok((out.text(), out.exit_code)) +} + +/// `exec_oneshot_as`: combined output under [`MAX_ONESHOT_OUTPUT`]. +pub async fn exec_oneshot_as( + ep: &AgentEndpoint, + user: &str, + cmd: Vec, + env: Vec, +) -> Result<(String, i64), String> { + exec_oneshot_limited(ep, user, cmd, env, MAX_ONESHOT_OUTPUT).await +} + +/// `exec_oneshot_streams_as`: `(stdout, stderr, exit_code)`. +pub async fn exec_oneshot_streams_as( + ep: &AgentEndpoint, + user: &str, + cmd: Vec, + env: Vec, +) -> Result<(String, String, i64), String> { + let out = exec_oneshot_raw(ep, user, cmd, env, MAX_ONESHOT_OUTPUT).await?; + Ok((out.stdout(), out.stderr(), out.exit_code)) +} + +// ── detached execs, status, resize ─────────────────────────────────────────── + +#[derive(Deserialize)] +struct ExecIdResponse { + exec_id: String, +} + +pub async fn exec_detached(ep: &AgentEndpoint, spec: &ExecSpec) -> Result { + const CONTEXT: &str = "Failed to start exec"; + let resp = send( + CONTEXT, + http() + .post(ep.url("v1/exec/detached")) + .header(reqwest::header::AUTHORIZATION, ep.bearer()) + .json(&spec_json(spec)), + ) + .await?; + if !resp.status().is_success() { + return Err(status_error(CONTEXT, resp).await); + } + resp.json::() + .await + .map(|r| r.exec_id) + .map_err(|e| format!("{}: malformed agent response: {}", CONTEXT, e)) +} + +#[derive(Deserialize)] +struct ExecStatus { + running: bool, + exit_code: Option, +} + +/// Exit codes attach streams saw in their `exit` frame, so `wait_exec_exit` +/// can answer without a round trip. Bounded: a long-running app opens many +/// sessions, and the agent keeps the same answer for ten minutes anyway. +struct ExitBook { + codes: HashMap<(String, String), i64>, + order: VecDeque<(String, String)>, +} + +const EXIT_BOOK_CAP: usize = 256; + +fn exit_book() -> &'static Mutex { + static BOOK: OnceLock> = OnceLock::new(); + BOOK.get_or_init(|| { + Mutex::new(ExitBook { + codes: HashMap::new(), + order: VecDeque::new(), + }) + }) +} + +fn record_exit(base_url: &str, exec_id: &str, code: i64) { + let key = (base_url.to_string(), exec_id.to_string()); + let mut book = exit_book().lock().unwrap_or_else(|p| p.into_inner()); + if book.codes.insert(key.clone(), code).is_none() { + book.order.push_back(key); + while book.order.len() > EXIT_BOOK_CAP { + if let Some(old) = book.order.pop_front() { + book.codes.remove(&old); + } + } + } +} + +fn recorded_exit(base_url: &str, exec_id: &str) -> Option { + let book = exit_book().lock().unwrap_or_else(|p| p.into_inner()); + book.codes + .get(&(base_url.to_string(), exec_id.to_string())) + .copied() +} + +/// Poll `GET /v1/exec/{id}` with Docker's budget. `None` on any error, an +/// unknown id, a finished exec with no code, or the budget running out — +/// exactly where `docker::exec::wait_for_exec_exit` says `None`. +pub async fn wait_exec_exit(ep: &AgentEndpoint, exec_id: &str) -> Option { + if let Some(code) = recorded_exit(ep.base_url.as_str(), exec_id) { + return Some(code); + } + let url = ep.url(&format!("v1/exec/{}", urlencode_segment(exec_id))); + for _ in 0..EXEC_EXIT_POLLS { + if let Some(code) = recorded_exit(ep.base_url.as_str(), exec_id) { + return Some(code); + } + let resp = http() + .get(url.clone()) + .header(reqwest::header::AUTHORIZATION, ep.bearer()) + .send() + .await + .ok()?; + if !resp.status().is_success() { + return None; + } + let status: ExecStatus = resp.json().await.ok()?; + if !status.running { + return status.exit_code; + } + tokio::time::sleep(EXEC_EXIT_POLL_INTERVAL).await; + } + None +} + +pub async fn resize_exec(ep: &AgentEndpoint, exec_id: &str, cols: u16, rows: u16) -> Result<(), String> { + const CONTEXT: &str = "Failed to resize exec"; + let resp = send( + CONTEXT, + http() + .post(ep.url(&format!("v1/exec/{}/resize", urlencode_segment(exec_id)))) + .header(reqwest::header::AUTHORIZATION, ep.bearer()) + .json(&serde_json::json!({ "cols": cols, "rows": rows })), + ) + .await?; + if !resp.status().is_success() { + return Err(status_error(CONTEXT, resp).await); + } + Ok(()) +} + +/// Exec ids come from the agent; escape them anyway so one can never turn +/// into a different path. +fn urlencode_segment(s: &str) -> String { + url::form_urlencoded::byte_serialize(s.as_bytes()).collect() +} + +// ── attach ─────────────────────────────────────────────────────────────────── + +type Ws = tokio_tungstenite::WebSocketStream< + tokio_tungstenite::MaybeTlsStream, +>; + +#[derive(Deserialize)] +struct ControlFrame { + #[serde(rename = "type")] + kind: String, + exec_id: Option, + code: Option, + error: Option, +} + +/// What the stdin writer hands the socket task. +enum StdinCmd { + Data(Bytes), + Close, +} + +/// Open `GET /v1/attach` and authenticate. Shared by attach (and, later, +/// tunnels). +async fn connect_ws(ep: &AgentEndpoint, path: &str, context: &str) -> Result { + let mut req = ep + .ws_url(path) + .as_str() + .into_client_request() + .map_err(|e| format!("{}: {}", context, e))?; + let auth = HeaderValue::from_str(&ep.bearer()) + .map_err(|_| format!("{}: the agent token is not a valid header value", context))?; + req.headers_mut() + .insert(tokio_tungstenite::tungstenite::http::header::AUTHORIZATION, auth); + match tokio::time::timeout(CONNECT_TIMEOUT, tokio_tungstenite::connect_async(req)).await { + Err(_) => Err(format!("{}: timed out connecting to the agent", context)), + Ok(Err(tokio_tungstenite::tungstenite::Error::Http(resp))) + if resp.status() == tokio_tungstenite::tungstenite::http::StatusCode::UNAUTHORIZED => + { + Err(format!("{}: the agent rejected the credentials (401)", context)) + } + Ok(Err(e)) => Err(format!("{}: {}", context, e)), + Ok(Ok((ws, _))) => Ok(ws), + } +} + +/// `GET /v1/attach`: the same [`AttachedExec`] Docker hands back. +/// +/// Output: channel 1 → [`ExecChunk::StdOut`], 2 → [`ExecChunk::StdErr`]; the +/// `exit` frame ends the stream and is remembered for [`wait_exec_exit`]. +/// Input: each write is one binary frame; `shutdown` sends `close_stdin`. +/// Dropping both halves closes the socket. +pub async fn attach(ep: &AgentEndpoint, spec: &ExecSpec) -> Result { + const CONTEXT: &str = "Failed to start exec"; + let mut ws = connect_ws(ep, "v1/attach", CONTEXT).await?; + + let mut hello = spec_json(spec); + hello.insert("attach_stdin".into(), serde_json::json!(spec.attach_stdin)); + let hello = serde_json::Value::Object(hello).to_string(); + ws.send(Message::text(hello)) + .await + .map_err(|e| format!("{}: {}", CONTEXT, e))?; + + let exec_id = match tokio::time::timeout(ATTACH_START_TIMEOUT, wait_started(&mut ws)).await { + Err(_) => return Err(format!("{}: the agent did not start the command", CONTEXT)), + Ok(r) => r.map_err(|e| format!("{}: {}", CONTEXT, e))?, + }; + + let (mut sink, stream) = ws.split(); + + // stdin: a bounded queue drained by one task that owns the socket's write + // half. The task ends — dropping the write half — when the writer is + // dropped or the socket fails. + let (tx, mut rx) = tokio::sync::mpsc::channel::(STDIN_QUEUE); + tokio::spawn(async move { + while let Some(cmd) = rx.recv().await { + let msg = match cmd { + StdinCmd::Data(b) => Message::binary(b), + StdinCmd::Close => Message::text(r#"{"type":"close_stdin"}"#), + }; + if sink.send(msg).await.is_err() { + break; + } + } + }); + + let base_url = ep.base_url.as_str().to_string(); + let id_for_stream = exec_id.clone(); + let output = futures_util::stream::unfold(Some(stream), move |state| { + let base_url = base_url.clone(); + let exec_id = id_for_stream.clone(); + async move { + let mut stream = state?; + loop { + match stream.next().await? { + Err(e) => return Some((Err(format!("Exec output error: {}", e)), None)), + Ok(Message::Binary(b)) => { + if b.is_empty() { + continue; + } + let payload = b.slice(1..); + let chunk = match b[0] { + 1 => ExecChunk::StdOut(payload), + 2 => ExecChunk::StdErr(payload), + _ => ExecChunk::Other(payload), + }; + return Some((Ok(chunk), Some(stream))); + } + Ok(Message::Text(t)) => { + let frame: ControlFrame = match serde_json::from_str(t.as_str()) { + Ok(f) => f, + Err(_) => continue, + }; + match frame.kind.as_str() { + "exit" => { + if let Some(code) = frame.code { + record_exit(&base_url, &exec_id, code); + } + return None; + } + "error" => { + let e = frame.error.unwrap_or_else(|| "agent error".into()); + return Some((Err(format!("Exec output error: {}", e)), None)); + } + _ => continue, + } + } + Ok(Message::Close(_)) => return None, + Ok(_) => continue, + } + } + } + }); + + Ok(AttachedExec { + exec_id, + output: Box::pin(output), + input: Box::pin(AttachInput { + tx: PollSender::new(tx), + close_sent: false, + }), + }) +} + +/// Read until the agent says `started` (its exec id) or `error`. +async fn wait_started(ws: &mut Ws) -> Result { + while let Some(msg) = ws.next().await { + match msg.map_err(|e| e.to_string())? { + Message::Text(t) => { + let frame: ControlFrame = serde_json::from_str(t.as_str()) + .map_err(|e| format!("malformed agent frame: {}", e))?; + match frame.kind.as_str() { + "started" => { + return frame + .exec_id + .ok_or_else(|| "agent sent no exec id".to_string()) + } + "error" => { + return Err(frame.error.unwrap_or_else(|| "agent error".into())) + } + _ => continue, + } + } + Message::Close(_) => break, + _ => continue, + } + } + Err("the agent closed the connection before starting the command".to_string()) +} + +/// An attach's stdin. +struct AttachInput { + tx: PollSender, + close_sent: bool, +} + +fn broken_pipe() -> io::Error { + io::Error::new(io::ErrorKind::BrokenPipe, "agent connection closed") +} + +impl AsyncWrite for AttachInput { + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + let this = self.get_mut(); + if buf.is_empty() { + return Poll::Ready(Ok(0)); + } + if this.close_sent { + return Poll::Ready(Err(broken_pipe())); + } + match this.tx.poll_reserve(cx) { + Poll::Pending => Poll::Pending, + Poll::Ready(Err(_)) => Poll::Ready(Err(broken_pipe())), + Poll::Ready(Ok(())) => { + this.tx + .send_item(StdinCmd::Data(Bytes::copy_from_slice(buf))) + .map_err(|_| broken_pipe())?; + Poll::Ready(Ok(buf.len())) + } + } + } + + /// Frames are flushed by the socket task as it sends them. + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + /// EOF on the child's stdin: `{"type":"close_stdin"}`, once. + fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let this = self.get_mut(); + if this.close_sent { + return Poll::Ready(Ok(())); + } + match this.tx.poll_reserve(cx) { + Poll::Pending => Poll::Pending, + Poll::Ready(Err(_)) => Poll::Ready(Err(broken_pipe())), + Poll::Ready(Ok(())) => { + this.tx.send_item(StdinCmd::Close).map_err(|_| broken_pipe())?; + this.close_sent = true; + Poll::Ready(Ok(())) + } + } + } +} + +// ── files ──────────────────────────────────────────────────────────────────── + +/// `(uid, gid)` of `claude`, falling back to `(1000, 1000)` — the same probe +/// and parse as Docker's `container_user_ids`. +pub async fn user_ids(ep: &AgentEndpoint) -> (u64, u64) { + let (out, _) = exec_oneshot_limited(ep, "claude", user_ids_cmd(), Vec::new(), USER_IDS_OUTPUT_LIMIT) + .await + .unwrap_or_default(); + parse_user_ids(&out) +} + +/// Split Docker's tar-entry `name` (which may carry a relative directory, as +/// the terminal drop's `triple-c-drops/` does) into the `dir` and +/// single-component `name` `PUT /v1/files` takes. Docker's extractor would +/// put the file at the same place. +fn split_upload_target(dest_dir: &str, file_name: &str) -> Result<(String, String), String> { + let parts: Vec<&str> = file_name.split('/').filter(|p| !p.is_empty()).collect(); + let Some((name, parents)) = parts.split_last() else { + return Err(format!("Failed to upload file to container: empty file name {:?}", file_name)); + }; + if parts.iter().any(|p| *p == "." || *p == "..") { + return Err(format!( + "Failed to upload file to container: {:?} is not a plain relative name", + file_name + )); + } + let mut dir = dest_dir.trim_end_matches('/').to_string(); + for p in parents { + dir.push('/'); + dir.push_str(p); + } + if dir.is_empty() { + dir.push('/'); + } + Ok((dir, name.to_string())) +} + +async fn put_file( + ep: &AgentEndpoint, + dest_dir: &str, + file_name: &str, + data: Vec, + mode: u32, + (uid, gid): (u64, u64), +) -> Result<(), String> { + const CONTEXT: &str = "Failed to upload file to container"; + let (dir, name) = split_upload_target(dest_dir, file_name)?; + let mut url = ep.url("v1/files"); + url.query_pairs_mut() + .append_pair("dir", &dir) + .append_pair("name", &name) + .append_pair("mode", &format!("{:o}", mode)) + .append_pair("uid", &uid.to_string()) + .append_pair("gid", &gid.to_string()); + let resp = send( + CONTEXT, + http() + .put(url) + .header(reqwest::header::AUTHORIZATION, ep.bearer()) + .header(reqwest::header::CONTENT_TYPE, "application/octet-stream") + .body(data), + ) + .await?; + if !resp.status().is_success() { + return Err(status_error(CONTEXT, resp).await); + } + Ok(()) +} + +/// Docker's `upload_bytes_with_ids`: same reported path. +pub async fn upload_bytes_with_ids( + ep: &AgentEndpoint, + dest_dir: &str, + file_name: &str, + data: &[u8], + mode: u32, + ids: (u64, u64), +) -> Result { + put_file(ep, dest_dir, file_name, data.to_vec(), mode, ids).await?; + Ok(format!("{}/{}", dest_dir.trim_end_matches('/'), file_name)) +} + +/// Docker's `upload_host_file_with_ids`: the host file is read (and checked) +/// by the same function, mode 0644, same reported path. +pub async fn upload_host_file_with_ids( + ep: &AgentEndpoint, + host_path: &str, + dest_dir: &str, + dest_name: &str, + ids: (u64, u64), +) -> Result { + let path = host_path.to_string(); + let data = tokio::task::spawn_blocking(move || read_host_file_for_upload(&path)) + .await + .map_err(|e| format!("Upload task panicked: {}", e))??; + put_file(ep, dest_dir, dest_name, data, HOST_FILE_UPLOAD_MODE, ids).await?; + Ok(container_join(dest_dir, dest_name)) +} + +/// `GET /v1/archive`, streamed. Dropping the stream drops the response, +/// which closes the connection and cancels the read on the agent. +pub fn download_archive(ep: &AgentEndpoint, path: &str) -> ArchiveStream { + const CONTEXT: &str = "Failed to download archive"; + let mut url = ep.url("v1/archive"); + url.query_pairs_mut().append_pair("path", path); + let req = http() + .get(url) + .header(reqwest::header::AUTHORIZATION, ep.bearer()); + + enum State { + Start(reqwest::RequestBuilder), + Body(reqwest::Response), + } + Box::pin(futures_util::stream::unfold( + Some(State::Start(req)), + |state| async move { + let mut resp = match state? { + State::Start(req) => match req.send().await { + Err(e) => return Some((Err(format!("{}: {}", CONTEXT, e)), None)), + Ok(r) if !r.status().is_success() => { + return Some((Err(status_error(CONTEXT, r).await), None)) + } + Ok(r) => r, + }, + State::Body(r) => r, + }; + loop { + match resp.chunk().await { + Err(e) => return Some((Err(format!("{}: {}", CONTEXT, e)), None)), + Ok(None) => return None, + Ok(Some(b)) if b.is_empty() => continue, + Ok(Some(b)) => return Some((Ok(b), Some(State::Body(resp)))), + } + } + }, + )) +} + +// ── state ──────────────────────────────────────────────────────────────────── + +/// `GET /v1/health` answers → `true`. Can't connect (or times out) → `false`, +/// the same "not running" an inspect failure gives on Docker. An agent that +/// answers with an error status is reachable but unusable, which is an +/// `Err` rather than a quiet `false`. +pub async fn is_running(ep: &AgentEndpoint) -> Result { + let resp = match http() + .get(ep.url("v1/health")) + .header(reqwest::header::AUTHORIZATION, ep.bearer()) + .timeout(HEALTH_TIMEOUT) + .send() + .await + { + Ok(r) => r, + Err(_) => return Ok(false), + }; + if !resp.status().is_success() { + return Err(status_error("Agent health check failed", resp).await); + } + Ok(true) +} + +#[cfg(test)] +mod tests; diff --git a/app/src-tauri/src/host/agent/tests.rs b/app/src-tauri/src/host/agent/tests.rs new file mode 100644 index 0000000..66ad559 --- /dev/null +++ b/app/src-tauri/src/host/agent/tests.rs @@ -0,0 +1,560 @@ +//! The agent client against a **fake** agent: a few axum handlers that speak +//! just enough of the v1 protocol to exercise each client path, and record +//! what they were sent. It is not the real server (that lives in `agent/`) +//! and does not try to behave like one beyond the wire shapes. + +use super::*; +use crate::host::{ContainerTarget, OUTPUT_LIMIT_MARKER}; +use axum::extract::ws::{Message as WsMessage, WebSocket, WebSocketUpgrade}; +use axum::extract::{Path, Query, State}; +use axum::http::{HeaderMap, StatusCode}; +use axum::response::{IntoResponse, Response}; +use axum::routing::{get, post, put}; +use axum::{Json, Router}; +use serde_json::{json, Value}; +use std::sync::Arc; +use tokio::io::AsyncWriteExt; + +/// Obviously fake; long enough to be unmistakable in an error string. +const TOKEN: &str = "fake-agent-token-for-unit-tests-only-0000"; + +#[derive(Default)] +struct Seen { + auth: Vec, + exec_bodies: Vec, + detached_bodies: Vec, + status_polls: u32, + resizes: Vec<(String, Value)>, + uploads: Vec<(HashMap, Vec)>, + attach_hello: Option, + stdin: Vec, + close_stdin: bool, +} + +struct Fake { + seen: std::sync::Mutex, +} + +type St = State>; + +impl Fake { + fn authed(&self, h: &HeaderMap) -> bool { + let got = h + .get("authorization") + .and_then(|v| v.to_str().ok()) + .unwrap_or("") + .to_string(); + let ok = got == format!("Bearer {}", TOKEN); + self.seen.lock().unwrap().auth.push(got); + ok + } +} + +fn unauthorized() -> Response { + StatusCode::UNAUTHORIZED.into_response() +} + +fn b64(bytes: &[u8]) -> String { + base64::engine::general_purpose::STANDARD.encode(bytes) +} + +async fn exec(State(f): St, h: HeaderMap, Json(body): Json) -> Response { + if !f.authed(&h) { + return unauthorized(); + } + f.seen.lock().unwrap().exec_bodies.push(body.clone()); + match body["cmd"][0].as_str().unwrap_or("") { + "sh" => Json(json!({"chunks": [{"s": "out", "d": b64(b"1000\n1001\n")}], "exit_code": 0})) + .into_response(), + // stdout and stderr interleaved; `ü` split across two stdout chunks; + // one byte that is not UTF-8 at all on stderr. + "mixed" => Json(json!({ + "chunks": [ + {"s": "out", "d": b64(b"first ")}, + {"s": "err", "d": b64(b"find: /x: denied\n")}, + {"s": "out", "d": b64(&[0xc3])}, + {"s": "out", "d": b64(&[0xbc, b'!'])}, + {"s": "err", "d": b64(&[0xff])}, + ], + "exit_code": 3 + })) + .into_response(), + "big" => ( + StatusCode::PAYLOAD_TOO_LARGE, + Json(json!({"error": "OUTPUT_LIMIT: Command output exceeded 5 bytes and was abandoned"})), + ) + .into_response(), + _ => (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"error": "boom"}))).into_response(), + } +} + +async fn detached(State(f): St, h: HeaderMap, Json(body): Json) -> Response { + if !f.authed(&h) { + return unauthorized(); + } + f.seen.lock().unwrap().detached_bodies.push(body); + Json(json!({"exec_id": "d1"})).into_response() +} + +/// `d1` reports running for two polls, then exit 7. Everything else is 404 — +/// including the attach's `a1`, so an attach's exit can only have come from +/// the client's own record of the `exit` frame. +async fn status(State(f): St, h: HeaderMap, Path(id): Path) -> Response { + if !f.authed(&h) { + return unauthorized(); + } + if id != "d1" { + return (StatusCode::NOT_FOUND, Json(json!({"error": "no such exec"}))).into_response(); + } + let mut s = f.seen.lock().unwrap(); + s.status_polls += 1; + if s.status_polls < 3 { + Json(json!({"running": true, "exit_code": null})).into_response() + } else { + Json(json!({"running": false, "exit_code": 7})).into_response() + } +} + +async fn resize(State(f): St, h: HeaderMap, Path(id): Path, Json(body): Json) -> Response { + if !f.authed(&h) { + return unauthorized(); + } + f.seen.lock().unwrap().resizes.push((id, body)); + StatusCode::NO_CONTENT.into_response() +} + +async fn files( + State(f): St, + h: HeaderMap, + Query(q): Query>, + body: Bytes, +) -> Response { + if !f.authed(&h) { + return unauthorized(); + } + let path = format!("{}/{}", q["dir"], q["name"]); + f.seen.lock().unwrap().uploads.push((q, body.to_vec())); + Json(json!({"path": path})).into_response() +} + +async fn archive(State(f): St, h: HeaderMap, Query(q): Query>) -> Response { + if !f.authed(&h) { + return unauthorized(); + } + if q["path"] == "/missing" { + return (StatusCode::NOT_FOUND, Json(json!({"error": "no such path"}))).into_response(); + } + let parts: Vec> = vec![ + Ok(Bytes::from_static(b"tar-part-1;")), + Ok(Bytes::from_static(b"tar-part-2;")), + Ok(Bytes::from(q["path"].clone().into_bytes())), + ]; + axum::body::Body::from_stream(futures_util::stream::iter(parts)).into_response() +} + +async fn health(State(f): St, h: HeaderMap) -> Response { + if !f.authed(&h) { + return unauthorized(); + } + Json(json!({"agent": "triple-c-agent", "version": "0.0.0-fake", "protocol": 1})).into_response() +} + +async fn attach_ws(State(f): St, h: HeaderMap, ws: WebSocketUpgrade) -> Response { + if !f.authed(&h) { + return unauthorized(); + } + ws.on_upgrade(move |socket| fake_attach(f, socket)) +} + +/// started → stdout, stderr, an unknown channel → collect stdin until +/// `close_stdin` → echo it on stdout → exit 5 → close. +async fn fake_attach(f: Arc, mut ws: WebSocket) { + let Some(Ok(WsMessage::Text(hello))) = ws.recv().await else { return }; + let hello: Value = serde_json::from_str(hello.as_str()).unwrap(); + let refuse = hello["cmd"][0] == "nope"; + f.seen.lock().unwrap().attach_hello = Some(hello); + if refuse { + let _ = ws + .send(WsMessage::Text(r#"{"type":"error","error":"no such user"}"#.into())) + .await; + return; + } + let _ = ws + .send(WsMessage::Text(r#"{"type":"started","exec_id":"a1"}"#.into())) + .await; + for frame in [&b"\x01out"[..], b"\x02err", b"\x09odd"] { + let _ = ws.send(WsMessage::Binary(Bytes::copy_from_slice(frame))).await; + } + while let Some(Ok(msg)) = ws.recv().await { + match msg { + WsMessage::Binary(b) => f.seen.lock().unwrap().stdin.extend_from_slice(&b), + WsMessage::Text(t) if t.as_str().contains("close_stdin") => { + f.seen.lock().unwrap().close_stdin = true; + break; + } + _ => {} + } + } + let mut echo = vec![1u8]; + echo.extend_from_slice(&f.seen.lock().unwrap().stdin); + let _ = ws.send(WsMessage::Binary(echo.into())).await; + let _ = ws.send(WsMessage::Text(r#"{"type":"exit","code":5}"#.into())).await; + let _ = ws.close().await; +} + +async fn fake_agent() -> (AgentEndpoint, Arc) { + let fake = Arc::new(Fake { + seen: std::sync::Mutex::new(Seen::default()), + }); + let app = Router::new() + .route("/v1/health", get(health)) + .route("/v1/exec", post(exec)) + .route("/v1/exec/detached", post(detached)) + .route("/v1/exec/{id}", get(status)) + .route("/v1/exec/{id}/resize", post(resize)) + .route("/v1/files", put(files)) + .route("/v1/archive", get(archive)) + .route("/v1/attach", get(attach_ws)) + .with_state(fake.clone()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + (AgentEndpoint::loopback(port, TOKEN), fake) +} + +fn target(ep: &AgentEndpoint) -> ContainerTarget { + ContainerTarget::Agent(ep.clone()) +} + +fn cmd(c: &str) -> Vec { + vec![c.to_string()] +} + +// ── endpoint ───────────────────────────────────────────────────────────────── + +#[test] +fn the_token_never_appears_in_debug_output() { + let ep = AgentEndpoint::loopback(47900, TOKEN); + let dbg = format!("{:?} {:?}", ep, ContainerTarget::Agent(ep.clone())); + assert!(!dbg.contains(TOKEN), "{}", dbg); + assert!(dbg.contains("127.0.0.1:47900")); + assert!(dbg.contains("redacted")); +} + +#[test] +fn an_agent_target_is_identified_by_its_base_url() { + let a = ContainerTarget::Agent(AgentEndpoint::loopback(47900, TOKEN)); + assert_eq!(a.id(), "http://127.0.0.1:47900/"); + assert_eq!(a, ContainerTarget::Agent(AgentEndpoint::loopback(47900, TOKEN))); + assert_ne!(a, ContainerTarget::Agent(AgentEndpoint::loopback(47901, TOKEN))); + let mut set = std::collections::HashSet::new(); + set.insert(a.clone()); + assert!(set.contains(&a)); +} + +#[test] +fn only_plain_http_endpoints_are_accepted() { + let ok = AgentEndpoint::new(url::Url::parse("http://127.0.0.1:1/x").unwrap(), TOKEN).unwrap(); + assert_eq!(ok.base_url.as_str(), "http://127.0.0.1:1/x/"); + assert_eq!(ok.url("v1/health").as_str(), "http://127.0.0.1:1/x/v1/health"); + assert_eq!(ok.ws_url("v1/attach").as_str(), "ws://127.0.0.1:1/x/v1/attach"); + assert!(AgentEndpoint::new(url::Url::parse("https://h/").unwrap(), TOKEN).is_err()); + assert!(AgentEndpoint::new(url::Url::parse("file:///tmp").unwrap(), TOKEN).is_err()); +} + +// ── auth ───────────────────────────────────────────────────────────────────── + +#[tokio::test] +async fn every_request_carries_the_bearer_token() { + let (ep, fake) = fake_agent().await; + let t = target(&ep); + assert!(t.is_running().await.unwrap()); + t.exec_oneshot(cmd("sh")).await.unwrap(); + let auth = fake.seen.lock().unwrap().auth.clone(); + assert_eq!(auth.len(), 2); + assert!(auth.iter().all(|a| a == &format!("Bearer {}", TOKEN))); +} + +#[tokio::test] +async fn a_wrong_token_is_an_error_that_does_not_carry_any_token() { + let (ep, _) = fake_agent().await; + let wrong_token = "wrong-fake-token-for-unit-tests-only-1111"; + let bad = ContainerTarget::Agent(AgentEndpoint::new(ep.base_url.clone(), wrong_token).unwrap()); + + let e = bad.exec_oneshot(cmd("sh")).await.unwrap_err(); + assert!(e.contains("401"), "{}", e); + assert!(e.starts_with("Failed to start exec"), "{}", e); + let Err(attach_err) = bad.attach(ExecSpec::new(cmd("sh"))).await else { + panic!("attach with a wrong token succeeded") + }; + assert!(attach_err.contains("401"), "{}", attach_err); + let health_err = bad.is_running().await.unwrap_err(); + let up_err = bad.upload_bytes("/tmp", "x", b"x", 0o644).await.unwrap_err(); + for e in [&e, &attach_err, &health_err, &up_err] { + assert!(!e.contains(wrong_token) && !e.contains(TOKEN), "{}", e); + } +} + +#[tokio::test] +async fn other_errors_carry_the_agents_message_and_no_token() { + let (ep, _) = fake_agent().await; + let e = target(&ep).exec_oneshot(cmd("fail")).await.unwrap_err(); + assert_eq!(e, "Failed to start exec: boom"); +} + +// ── one-shot execs ─────────────────────────────────────────────────────────── + +#[tokio::test] +async fn every_oneshot_variant_reassembles_what_docker_would_return() { + let (ep, fake) = fake_agent().await; + let t = target(&ep); + // Arrival order, `ü` rejoined across chunks, the stray 0xff lossy. + let combined = "first find: /x: denied\nü!\u{fffd}"; + let env = vec!["K=V".to_string()]; + + assert_eq!(t.exec_oneshot(cmd("mixed")).await.unwrap(), combined); + assert_eq!(t.exec_oneshot_env(cmd("mixed"), env.clone()).await.unwrap(), combined); + assert_eq!( + t.exec_oneshot_env_status(cmd("mixed"), env.clone()).await.unwrap(), + (combined.to_string(), 3) + ); + assert_eq!( + t.exec_oneshot_as("root", cmd("mixed"), env.clone()).await.unwrap(), + (combined.to_string(), 3) + ); + assert_eq!( + t.exec_oneshot_limited("root", cmd("mixed"), 4096).await.unwrap(), + (combined.to_string(), 3) + ); + assert_eq!( + t.exec_oneshot_streams_as("root", cmd("mixed"), env.clone()).await.unwrap(), + ("first ü!".to_string(), "find: /x: denied\n\u{fffd}".to_string(), 3) + ); + + let bodies = fake.seen.lock().unwrap().exec_bodies.clone(); + let max = MAX_ONESHOT_OUTPUT as u64; + let expect = [ + ("claude", json!([]), max), + ("claude", json!(["K=V"]), max), + ("claude", json!(["K=V"]), max), + ("root", json!(["K=V"]), max), + ("root", json!([]), 4096), + ("root", json!(["K=V"]), max), + ]; + assert_eq!(bodies.len(), expect.len()); + for (b, (user, env, limit)) in bodies.iter().zip(expect) { + assert_eq!(b["user"], user); + assert_eq!(b["env"], env); + assert_eq!(b["limit"].as_u64(), Some(limit)); + assert_eq!(b["cmd"], json!(["mixed"])); + // Docker's one-shot names no working dir and no tty. + assert_eq!(b["working_dir"], Value::Null); + assert_eq!(b["tty"], false); + } +} + +#[tokio::test] +async fn an_output_limit_refusal_passes_through_verbatim() { + let (ep, _) = fake_agent().await; + let e = target(&ep).exec_oneshot_limited("claude", cmd("big"), 5).await.unwrap_err(); + assert_eq!(e, "OUTPUT_LIMIT: Command output exceeded 5 bytes and was abandoned"); + assert!(e.starts_with(OUTPUT_LIMIT_MARKER)); +} + +#[tokio::test] +async fn the_client_also_enforces_the_limit_on_what_it_buffers() { + // An agent that ignored `limit` still cannot make the client buffer past it. + let (ep, _) = fake_agent().await; + let e = target(&ep).exec_oneshot_limited("claude", cmd("mixed"), 8).await.unwrap_err(); + assert!(e.starts_with(OUTPUT_LIMIT_MARKER), "{}", e); +} + +#[tokio::test] +async fn user_ids_come_from_the_same_probe_as_docker() { + let (ep, fake) = fake_agent().await; + assert_eq!(target(&ep).user_ids().await, (1000, 1001)); + let b = fake.seen.lock().unwrap().exec_bodies[0].clone(); + assert_eq!(b["cmd"], json!(["sh", "-c", "id -u; id -g"])); + assert_eq!(b["user"], "claude"); + assert_eq!(b["limit"], 256); +} + +// ── detached, status, resize ───────────────────────────────────────────────── + +#[tokio::test] +async fn a_detached_exec_is_polled_until_it_reports_its_exit() { + let (ep, fake) = fake_agent().await; + let t = target(&ep); + let spec = ExecSpec::new(cmd("sleep")).user("root").working_dir(None).no_stdin(); + let id = t.exec_detached(spec).await.unwrap(); + assert_eq!(id, "d1"); + assert_eq!(t.wait_exec_exit(&id).await, Some(7)); + let s = fake.seen.lock().unwrap(); + assert_eq!(s.status_polls, 3); + assert_eq!(s.detached_bodies[0]["user"], "root"); + assert_eq!(s.detached_bodies[0]["working_dir"], Value::Null); + drop(s); + assert_eq!(t.wait_exec_exit("unknown").await, None); +} + +#[tokio::test] +async fn resize_posts_cols_and_rows() { + let (ep, fake) = fake_agent().await; + target(&ep).resize_exec("a1", 120, 40).await.unwrap(); + let r = fake.seen.lock().unwrap().resizes.clone(); + assert_eq!(r, vec![("a1".to_string(), json!({"cols": 120, "rows": 40}))]); +} + +// ── attach ─────────────────────────────────────────────────────────────────── + +#[tokio::test] +async fn attach_maps_channels_forwards_stdin_and_ends_on_exit() { + let (ep, fake) = fake_agent().await; + let t = target(&ep); + let spec = ExecSpec::new(cmd("cat")).env(vec!["A=1".into()]).tty(true); + let AttachedExec { + exec_id, + mut output, + mut input, + } = t.attach(spec).await.unwrap(); + assert_eq!(exec_id, "a1"); + + assert_eq!(output.next().await.unwrap().unwrap(), ExecChunk::StdOut(Bytes::from_static(b"out"))); + assert_eq!(output.next().await.unwrap().unwrap(), ExecChunk::StdErr(Bytes::from_static(b"err"))); + assert_eq!(output.next().await.unwrap().unwrap(), ExecChunk::Other(Bytes::from_static(b"odd"))); + + input.write_all(b"hello ").await.unwrap(); + input.write_all(b"agent").await.unwrap(); + input.shutdown().await.unwrap(); + + assert_eq!( + output.next().await.unwrap().unwrap(), + ExecChunk::StdOut(Bytes::from_static(b"hello agent")) + ); + // The exit frame ends the stream… + assert!(output.next().await.is_none()); + // …and is what answers wait_exec_exit (the fake 404s `a1`). + assert_eq!(t.wait_exec_exit(&exec_id).await, Some(5)); + + let s = fake.seen.lock().unwrap(); + assert!(s.close_stdin); + assert_eq!(s.stdin, b"hello agent"); + let hello = s.attach_hello.clone().unwrap(); + assert_eq!(hello["cmd"], json!(["cat"])); + assert_eq!(hello["user"], "claude"); + assert_eq!(hello["working_dir"], "/workspace"); + assert_eq!(hello["env"], json!(["A=1"])); + assert_eq!(hello["tty"], true); + assert_eq!(hello["attach_stdin"], true); +} + +#[tokio::test] +async fn an_attach_the_agent_refuses_is_an_error() { + let (ep, _) = fake_agent().await; + let Err(e) = target(&ep).attach(ExecSpec::new(cmd("nope"))).await else { + panic!("a refused attach succeeded") + }; + assert_eq!(e, "Failed to start exec: no such user"); +} + +// ── files ──────────────────────────────────────────────────────────────────── + +#[tokio::test] +async fn uploads_put_the_bytes_with_dir_name_mode_and_owner() { + let (ep, fake) = fake_agent().await; + let t = target(&ep); + assert_eq!( + t.upload_bytes_with_ids("/tmp/", "x.txt", b"hi", 0o755, (1000, 1001)) + .await + .unwrap(), + "/tmp/x.txt" + ); + assert_eq!( + t.upload_bytes("/home/claude/in", "p.tar", b"tar", 0o644).await.unwrap(), + "/home/claude/in/p.tar" + ); + // The terminal drop's nested name: Docker's tar entry would land it in the + // subdirectory, so the agent is asked for that directory. + assert_eq!( + t.upload_bytes_with_ids("/tmp", "triple-c-drops/a.txt", b"a", 0o644, (1, 2)) + .await + .unwrap(), + "/tmp/triple-c-drops/a.txt" + ); + assert!(t.upload_bytes("/tmp", "../etc/x", b"", 0o644).await.is_err()); + + let up = fake.seen.lock().unwrap().uploads.clone(); + let q = |i: usize, k: &str| up[i].0[k].clone(); + assert_eq!((q(0, "dir"), q(0, "name"), q(0, "mode")), ("/tmp".into(), "x.txt".into(), "755".into())); + assert_eq!((q(0, "uid"), q(0, "gid")), ("1000".into(), "1001".into())); + assert_eq!(up[0].1, b"hi"); + assert_eq!((q(1, "uid"), q(1, "gid"), q(1, "mode")), ("0".into(), "0".into(), "644".into())); + assert_eq!((q(2, "dir"), q(2, "name")), ("/tmp/triple-c-drops".into(), "a.txt".into())); + assert_eq!(up.len(), 3); +} + +#[tokio::test] +async fn a_host_file_upload_reads_through_the_shared_checked_reader() { + let (ep, fake) = fake_agent().await; + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("notes.txt"); + std::fs::write(&path, b"from the host").unwrap(); + let path = path.canonicalize().unwrap(); + let t = target(&ep); + let got = t + .upload_host_file_with_ids(path.to_str().unwrap(), "/workspace/", "notes.txt", (7, 8)) + .await + .unwrap(); + assert_eq!(got, "/workspace/notes.txt"); + // Without ids: resolved through the uid probe first. + t.upload_host_file(path.to_str().unwrap(), "/tmp", "n2.txt").await.unwrap(); + + let up = fake.seen.lock().unwrap().uploads.clone(); + assert_eq!(up[0].1, b"from the host"); + assert_eq!(up[0].0["mode"], "644"); + assert_eq!((up[0].0["uid"].as_str(), up[0].0["gid"].as_str()), ("7", "8")); + assert_eq!((up[1].0["uid"].as_str(), up[1].0["gid"].as_str()), ("1000", "1001")); + + let missing = dir.path().join("gone.txt"); + let e = t + .upload_host_file_with_ids(missing.to_str().unwrap(), "/tmp", "g", (1, 1)) + .await + .unwrap_err(); + assert!(e.starts_with("Failed to read"), "{}", e); +} + +#[tokio::test] +async fn an_archive_is_streamed_and_a_missing_path_is_an_error() { + let (ep, _) = fake_agent().await; + let t = target(&ep); + let mut s = t.download_archive("/workspace/a dir"); + let mut got = Vec::new(); + while let Some(chunk) = s.next().await { + got.extend_from_slice(&chunk.unwrap()); + } + assert_eq!(got, b"tar-part-1;tar-part-2;/workspace/a dir"); + + let mut s = t.download_archive("/missing"); + let e = s.next().await.unwrap().unwrap_err(); + assert_eq!(e, "Failed to download archive: no such path"); + assert!(s.next().await.is_none()); +} + +// ── state ──────────────────────────────────────────────────────────────────── + +#[tokio::test] +async fn is_running_follows_whether_the_agent_answers() { + let (ep, _) = fake_agent().await; + assert!(target(&ep).is_running().await.unwrap()); + assert!(!target(&ep).has_exited().await); + + let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let port = l.local_addr().unwrap().port(); + drop(l); + let gone = ContainerTarget::Agent(AgentEndpoint::loopback(port, TOKEN)); + assert!(!gone.is_running().await.unwrap()); + // Silence is not a positive exit. + assert!(!gone.has_exited().await); +} diff --git a/app/src-tauri/src/host/lifecycle.rs b/app/src-tauri/src/host/lifecycle.rs index 1196d57..0e63b0a 100644 --- a/app/src-tauri/src/host/lifecycle.rs +++ b/app/src-tauri/src/host/lifecycle.rs @@ -25,6 +25,13 @@ pub enum ProjectHost { Docker, } +/// An agent target is a way *into* a container, not a handle on its +/// lifecycle: whatever runs the container (ECS, the local harness) stops and +/// removes it. Reaching here means a caller paired the Docker host with an +/// agent target, which is a bug to surface, not a panic. +const AGENT_LIFECYCLE: &str = + "{op} a container is not supported for an agent target; the host that runs it owns its lifecycle"; + /// What a host could not clean up when asked to discard a project's state. /// Docker: the snapshot image and the named volumes. #[derive(Debug, Default, Clone, PartialEq, Eq)] @@ -71,6 +78,7 @@ impl ProjectHost { 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")), } } @@ -81,6 +89,7 @@ impl ProjectHost { let _ = docker::stop_container(container_id).await; docker::remove_container(container_id).await } + ContainerTarget::Agent(_) => Err(AGENT_LIFECYCLE.replace("{op}", "Removing")), } } diff --git a/app/src-tauri/src/host/mod.rs b/app/src-tauri/src/host/mod.rs index 86359d4..b41ba19 100644 --- a/app/src-tauri/src/host/mod.rs +++ b/app/src-tauri/src/host/mod.rs @@ -7,8 +7,9 @@ //! "is it running" — goes through [`ContainerTarget`] instead of naming //! `docker::exec` or bollard directly. Today the only variant is //! [`ContainerTarget::Docker`], which delegates to the existing code in -//! `docker/exec.rs` and `docker/container.rs` with no change in behaviour. An -//! AWS variant (reached through the in-container agent) slots in beside it. +//! `docker/exec.rs` and `docker/container.rs` with no change in behaviour. +//! [`ContainerTarget::Agent`] reaches a container through the in-container +//! `triple-c-agent` instead (`agent.rs`), which is how the AWS host will get in. //! //! ## Why an enum and not a trait object //! @@ -26,6 +27,7 @@ //! `no_container_interior_bollard_outside_the_host_layer` test pins which files //! may still do so. +pub mod agent; pub mod lifecycle; pub mod sessions; pub mod watch; @@ -33,6 +35,7 @@ pub mod watch; pub use crate::docker::exec::{ MAX_DROP_BYTES, OUTPUT_LIMIT_MARKER, PROC_NET_OUTPUT_LIMIT, }; +pub use agent::AgentEndpoint; pub use sessions::ExecSessionManager; use bytes::Bytes; @@ -132,6 +135,12 @@ pub type ArchiveStream = Pin> + Send pub enum ContainerTarget { /// A container on the local Docker daemon, by id or name. Docker { container_id: String }, + /// A container reached through the in-container `triple-c-agent` + /// (`host/agent.rs`; protocol in + /// `docs/superpowers/specs/2026-11-13-triple-c-agent-protocol.md`). Used + /// by the AWS host (later) and the local agent harness. The host that + /// runs the container, not this target, owns its lifecycle. + Agent(AgentEndpoint), } impl ContainerTarget { @@ -151,6 +160,7 @@ impl ContainerTarget { pub fn id(&self) -> &str { match self { ContainerTarget::Docker { container_id } => container_id, + ContainerTarget::Agent(ep) => ep.base_url.as_str(), } } @@ -163,6 +173,7 @@ impl ContainerTarget { ContainerTarget::Docker { container_id } => { docker::container::is_container_running(container_id).await } + ContainerTarget::Agent(ep) => agent::is_running(ep).await, } } @@ -175,6 +186,9 @@ impl ContainerTarget { ContainerTarget::Docker { container_id } => { docker::exec::exec_oneshot(container_id, cmd).await } + ContainerTarget::Agent(ep) => agent::exec_oneshot_as(ep, "claude", cmd, Vec::new()) + .await + .map(|(out, _)| out), } } @@ -187,6 +201,9 @@ impl ContainerTarget { ContainerTarget::Docker { container_id } => { docker::exec::exec_oneshot_env(container_id, cmd, env).await } + ContainerTarget::Agent(ep) => agent::exec_oneshot_as(ep, "claude", cmd, env) + .await + .map(|(out, _)| out), } } @@ -199,6 +216,7 @@ impl ContainerTarget { ContainerTarget::Docker { container_id } => { docker::exec::exec_oneshot_env_status(container_id, cmd, env).await } + ContainerTarget::Agent(ep) => agent::exec_oneshot_as(ep, "claude", cmd, env).await, } } @@ -212,6 +230,7 @@ impl ContainerTarget { ContainerTarget::Docker { container_id } => { docker::exec::exec_oneshot_as(container_id, user, cmd, env).await } + ContainerTarget::Agent(ep) => agent::exec_oneshot_as(ep, user, cmd, env).await, } } @@ -225,6 +244,9 @@ impl ContainerTarget { ContainerTarget::Docker { container_id } => { docker::exec::exec_oneshot_limited(container_id, user, cmd, limit).await } + ContainerTarget::Agent(ep) => { + agent::exec_oneshot_limited(ep, user, cmd, Vec::new(), limit).await + } } } @@ -239,6 +261,7 @@ impl ContainerTarget { ContainerTarget::Docker { container_id } => { docker::exec::exec_oneshot_streams_as(container_id, user, cmd, env).await } + ContainerTarget::Agent(ep) => agent::exec_oneshot_streams_as(ep, user, cmd, env).await, } } @@ -249,6 +272,7 @@ impl ContainerTarget { ContainerTarget::Docker { container_id } => { docker::exec::exec_detached(container_id, &spec).await } + ContainerTarget::Agent(ep) => agent::exec_detached(ep, &spec).await, } } @@ -264,6 +288,7 @@ impl ContainerTarget { ContainerTarget::Docker { container_id } => { docker::exec::create_attached_exec_spec(container_id, &spec).await } + ContainerTarget::Agent(ep) => agent::attach(ep, &spec).await, } } @@ -272,12 +297,14 @@ impl ContainerTarget { pub async fn wait_exec_exit(&self, exec_id: &str) -> Option { match self { ContainerTarget::Docker { .. } => docker::exec::wait_for_exec_exit(exec_id).await, + ContainerTarget::Agent(ep) => agent::wait_exec_exit(ep, exec_id).await, } } pub async fn resize_exec(&self, exec_id: &str, cols: u16, rows: u16) -> Result<(), String> { match self { ContainerTarget::Docker { .. } => docker::exec::resize_exec(exec_id, cols, rows).await, + ContainerTarget::Agent(ep) => agent::resize_exec(ep, exec_id, cols, rows).await, } } @@ -289,6 +316,7 @@ impl ContainerTarget { ContainerTarget::Docker { container_id } => { docker::exec::container_user_ids(container_id).await } + ContainerTarget::Agent(ep) => agent::user_ids(ep).await, } } @@ -311,6 +339,9 @@ impl ContainerTarget { ) .await } + ContainerTarget::Agent(ep) => { + agent::upload_bytes_with_ids(ep, dest_dir, file_name, data, mode, (0, 0)).await + } } } @@ -335,6 +366,9 @@ impl ContainerTarget { ) .await } + ContainerTarget::Agent(ep) => { + agent::upload_bytes_with_ids(ep, dest_dir, file_name, data, mode, ids).await + } } } @@ -355,6 +389,10 @@ impl ContainerTarget { ) .await } + ContainerTarget::Agent(ep) => { + let ids = agent::user_ids(ep).await; + agent::upload_host_file_with_ids(ep, host_path, dest_dir, dest_name, ids).await + } } } @@ -377,6 +415,9 @@ impl ContainerTarget { ) .await } + ContainerTarget::Agent(ep) => { + agent::upload_host_file_with_ids(ep, host_path, dest_dir, dest_name, ids).await + } } } @@ -387,6 +428,7 @@ impl ContainerTarget { ContainerTarget::Docker { container_id } => { docker::exec::download_archive(container_id, path) } + ContainerTarget::Agent(ep) => agent::download_archive(ep, path), } } } diff --git a/app/src-tauri/src/host/watch.rs b/app/src-tauri/src/host/watch.rs index 7ba6ea5..2c60254 100644 --- a/app/src-tauri/src/host/watch.rs +++ b/app/src-tauri/src/host/watch.rs @@ -50,6 +50,11 @@ impl ContainerTarget { pub async fn has_exited(&self) -> bool { match self { 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, } } }