CyberCoveLLC/Triple-C
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 47 Wiki Activity

Merge branch 'worktree-agent-a8d5071f2fd1991d6' into feat/triple-c-agent

Browse Source
This commit is contained in:
shadowdao committed 2026-11-13 12:03:00 -08:00
2 parents 20a25fc518 48c60ef89c
commit 55ed2d17d0
9 files changed
+1563 -79
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

CLAUDE.md
+5 -1
View File
@@ -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
app/src-tauri/Cargo.lock
Generated
+2
View File
@@ -6642,6 +6642,8 @@ dependencies = [
"tauri-plugin-opener",
"tempfile",
"tokio",
"tokio-tungstenite",
"tokio-util",
"tower-http",
"url",
"uuid",
app/src-tauri/Cargo.toml
+7
View File
@@ -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"
app/src-tauri/src/docker/exec.rs
+107 -76
View File
@@ -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<Vec<u8>, 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<Vec<u8>, 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<String> {
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::<u64>().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<u8>,
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<u8> = 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<i64>) -> Result<i64, String> {
/// spurious "the rename failed" and a real one.
pub async fn wait_for_exec_exit(exec_id: &str) -> Option<i64> {
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<i64> {
}
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));
}
app/src-tauri/src/host/agent.rs
+824
View File
@@ -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:<port>/`: 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", &"<redacted>")
.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<String>) -> Result<Self, String> {
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:<port>`.
pub fn loopback(port: u16, token: impl Into<String>) -> 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<reqwest::Client> = 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<String> {
resp.json::<ErrorBody>().await.ok().map(|b| b.error)
}
/// `"<context>: <why>"` 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<reqwest::Response, String> {
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<Option<Vec<u8>>, 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<String, serde_json::Value> {
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<OneshotChunk>,
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<String>,
env: Vec<String>,
limit: usize,
) -> Result<OneshotOutput, String> {
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<String>,
env: Vec<String>,
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<String>,
env: Vec<String>,
) -> 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<String>,
env: Vec<String>,
) -> 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<String, String> {
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::<ExecIdResponse>()
.await
.map(|r| r.exec_id)
.map_err(|e| format!("{}: malformed agent response: {}", CONTEXT, e))
}
#[derive(Deserialize)]
struct ExecStatus {
running: bool,
exit_code: Option<i64>,
}
/// 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<ExitBook> {
static BOOK: OnceLock<Mutex<ExitBook>> = 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<i64> {
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<i64> {
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<tokio::net::TcpStream>,
>;
#[derive(Deserialize)]
struct ControlFrame {
#[serde(rename = "type")]
kind: String,
exec_id: Option<String>,
code: Option<i64>,
error: Option<String>,
}
/// 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<Ws, String> {
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<AttachedExec, String> {
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::<StdinCmd>(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<String, String> {
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<StdinCmd>,
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<io::Result<usize>> {
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<io::Result<()>> {
Poll::Ready(Ok(()))
}
/// EOF on the child's stdin: `{"type":"close_stdin"}`, once.
fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
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/<file>` 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<u8>,
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<String, String> {
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<String, String> {
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<bool, String> {
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;
app/src-tauri/src/host/agent/tests.rs
+560
View File
@@ -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<String>,
exec_bodies: Vec<Value>,
detached_bodies: Vec<Value>,
status_polls: u32,
resizes: Vec<(String, Value)>,
uploads: Vec<(HashMap<String, String>, Vec<u8>)>,
attach_hello: Option<Value>,
stdin: Vec<u8>,
close_stdin: bool,
}
struct Fake {
seen: std::sync::Mutex<Seen>,
}
type St = State<Arc<Fake>>;
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<Value>) -> 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<Value>) -> 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<String>) -> 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<String>, Json(body): Json<Value>) -> 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<HashMap<String, String>>,
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<HashMap<String, String>>) -> 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<Result<Bytes, std::io::Error>> = 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<Fake>, 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<Fake>) {
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<String> {
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);
}
app/src-tauri/src/host/lifecycle.rs
+9
View File
@@ -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")),
}
}
app/src-tauri/src/host/mod.rs
+44 -2
View File
@@ -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<Box<dyn Stream<Item = Result<Bytes, String>> + 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<i64> {
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),
}
}
}
app/src-tauri/src/host/watch.rs
+5
View File
@@ -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,
}
}
}