alien-bindings 3.3.41

//! AWS sandbox provider: Lambda MicroVMs, reached over the agent protocol.
//!
//! AWS gives a transport and nothing on the other end — a MicroVM is reachable only through its
//! HTTPS endpoint, and the agent Alien ships in the image is what answers there.
//!
//! Authorization is the endpoint token, not an Alien capability. `CreateMicrovmAuthToken` is
//! minted with the workload's own IAM identity and scoped to one MicroVM, an explicit port set
//! and an expiry — a request to a port outside it is refused at the proxy. One MicroVM is one
//! sandbox, so that scope is exactly the one a capability would express.

use std::collections::BTreeMap;
use std::sync::Arc;

use async_trait::async_trait;
use futures::stream::BoxStream;

use crate::error::{ErrorData, Result};
use crate::providers::sandbox::agent_protocol::{self, AgentTransport, AGENT_PORT};
use crate::providers::sandbox::refusal::Unreachable;
use crate::traits::{
    Binding, CommandOutput, CreateSandboxRequest, JobPoll, JobStart, PreviewCapability,
    ResolvedSandbox, RunCommandRequest, Sandbox, SandboxInstance, SandboxState,
};
use alien_aws_clients::aws::lambda_microvms::{LambdaMicrovmsApi, Microvm, MAX_AUTH_TOKEN_MINUTES};
use alien_core::{Platform, SandboxCapabilities};
use alien_error::{AlienError, Context, ContextError, IntoAlienError};
use tracing::warn;

/// Header the proxy reads to decide which port inside the MicroVM a request reaches.
const PROXY_PORT_HEADER: &str = "X-aws-proxy-port";

/// How long `create` waits for a MicroVM to become servable.
///
/// AWS answers 502 at the proxy "during the first few seconds after the MicroVM is run while the
/// snapshot is being restored", so the window is short; this is generous enough that a slow
/// restore reads as slow rather than broken.
#[cfg(not(test))]
const SANDBOX_READY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60);
#[cfg(not(test))]
const SANDBOX_READY_POLL: std::time::Duration = std::time::Duration::from_millis(500);

// A unit test has no reachable agent, so the budget only decides how long it takes to say so.
#[cfg(test)]
const SANDBOX_READY_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(20);
#[cfg(test)]
const SANDBOX_READY_POLL: std::time::Duration = std::time::Duration::from_millis(5);

/// How long `terminate` waits for a MicroVM to reach `TERMINATED` after AWS accepts the request.
///
/// Alien's images declare no terminate hook, so there is no in-guest ceiling to wait out; this
/// is the same bound Azure's terminate gives its asynchronous delete. A caller waiting forever is
/// its own outage, and an unconfirmed termination is reported as unconfirmed rather than done.
#[cfg(not(test))]
const SANDBOX_TERMINATED_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
#[cfg(not(test))]
const SANDBOX_TERMINATED_POLL: std::time::Duration = std::time::Duration::from_secs(2);

#[cfg(test)]
const SANDBOX_TERMINATED_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(40);
#[cfg(test)]
const SANDBOX_TERMINATED_POLL: std::time::Duration = std::time::Duration::from_millis(5);

/// Side-effect free, which is what makes it safe to repeat while `run_command` is not.
const HEALTH_PATH: &str = "/v1/health";

/// Life of a token minted to talk to the agent.
///
/// Short because it is minted per request anyway: the mint response carries no expiry, so there
/// is nothing to cache against and a long-lived token would only widen the window if one leaked.
const AGENT_TOKEN_MINUTES: u32 = 5;

/// Life of a preview capability handed to a caller.
///
/// Clamped where it is reported, not only where it is requested: AWS caps the mint at 60
/// minutes, so an unclamped figure here would promise a caller more life than the token has.
const PREVIEW_TOKEN_MINUTES: u32 = 30;

/// What a caller is told a preview capability is good for.
fn preview_lifetime_seconds() -> u64 {
    u64::from(PREVIEW_TOKEN_MINUTES.min(MAX_AUTH_TOKEN_MINUTES)) * 60
}

/// A Sandbox backed by Lambda MicroVMs.
#[derive(Debug)]
pub struct AwsSandbox {
    egress: Option<alien_core::SandboxEgress>,
    egress_manager: Option<Arc<crate::sandbox_egress::SandboxEgressManager>>,
    default_policy: Option<alien_core::EgressPolicy>,
    microvms: Arc<dyn LambdaMicrovmsApi>,
    image_identifier: String,
    image_version: String,
    /// Connectors every sandbox starts with. Empty means the public internet is reachable, so
    /// `deny` is a connector rather than the absence of one.
    egress_connector_arns: Vec<String>,
    /// Ports preview may be minted for. `CreateMicrovmAuthToken` carries no port condition key
    /// and grants whatever port it is asked for, so this list bounds callers that go through this
    /// provider; a Remote Bindings caller holding the raw credential is not bounded by it.
    preview_ports: Vec<u16>,
    /// Idle seconds before AWS suspends the MicroVM, where the declaration asked for it.
    idle_pause_seconds: Option<u32>,
    /// Wall-clock ceiling on a sandbox, where the declaration asked for one. Lambda terminates
    /// the MicroVM when it elapses.
    max_lifetime_seconds: Option<u32>,
    agent: reqwest::Client,
}

impl AwsSandbox {
    /// Supplies the declaration carried by the resolved binding.
    pub fn with_egress(mut self, egress: Option<alien_core::SandboxEgress>) -> Self {
        self.egress = egress;
        self
    }

    /// Enables gateway enforcement once deployment setup has provided its gateway and signing vault.
    pub fn with_egress_manager(
        mut self,
        manager: Arc<crate::sandbox_egress::SandboxEgressManager>,
    ) -> Self {
        self.egress_manager = Some(manager);
        self
    }

    pub fn with_default_policy(mut self, policy: Option<alien_core::EgressPolicy>) -> Self {
        self.default_policy = policy;
        self
    }
    fn validate_egress(&self, requested: Option<&alien_core::EgressPolicy>) -> Result<()> {
        let effective = requested.or(self.default_policy.as_ref());
        if self.egress_manager.is_some() {
            super::egress::validate(
                effective.unwrap_or(&alien_core::EgressPolicy::Internet),
                self.egress.as_ref(),
            )
        } else {
            super::egress::require_default(effective, self.egress.as_ref())
        }
    }

    /// Deliver identity only after the gateway has accepted the initial policy.
    async fn install_egress(
        &self,
        sandbox_id: &str,
        delivery: crate::sandbox_egress::EgressDelivery,
    ) -> Result<()> {
        self.deliver_egress(sandbox_id, Some(&delivery.token), &delivery.policy)
            .await
    }

    async fn deliver_egress(
        &self,
        sandbox_id: &str,
        token: Option<&str>,
        policy: &str,
    ) -> Result<()> {
        let mut body = serde_json::json!({"policy":policy});
        if let Some(token) = token {
            body["token"] = token.into();
        }
        let deadline = std::time::Instant::now() + SANDBOX_READY_TIMEOUT;
        loop {
            if self
                .owned_microvm("sandbox.egress", sandbox_id)
                .await?
                .is_some_and(|vm| vm.endpoint.is_some())
            {
                break;
            }
            if std::time::Instant::now() >= deadline {
                return Err(AlienError::new(ErrorData::SandboxUnreachable {
                    operation: "sandbox.egress".to_string(),
                    reason: "sandbox published no agent endpoint".to_string(),
                }));
            }
            tokio::time::sleep(SANDBOX_READY_POLL).await;
        }
        let request = self
            .authorized_request(sandbox_id, reqwest::Method::PUT, "/v1/egress")
            .await?
            .json(&body)
            .timeout(SANDBOX_READY_TIMEOUT);
        Self::poll_egress(request, deadline, token, policy).await
    }

    async fn poll_egress(
        request: reqwest::RequestBuilder,
        deadline: std::time::Instant,
        token: Option<&str>,
        policy: &str,
    ) -> Result<()> {
        let mut delay = SANDBOX_READY_POLL;
        let mut last_seen = "the endpoint gave no answer".to_string();
        let delivery = async {
            loop {
                let sent = request
                    .try_clone()
                    .ok_or_else(|| {
                        AlienError::new(ErrorData::SandboxUnreachable {
                            operation: "sandbox.egress".to_string(),
                            reason: "could not repeat identity delivery".to_string(),
                        })
                    })?
                    .send()
                    .await;
                let mut throttled = false;
                match sent {
                    Ok(response) if response.status().is_success() => return Ok(true),
                    Ok(mut response) => {
                        let status = response.status();
                        last_seen = format!("HTTP {status}");
                        let reason =
                            Self::endpoint_refusal_reason(&mut response, token, policy).await?;
                        if !reason.is_empty() {
                            last_seen.push_str(&format!(": {reason}"));
                        }
                        // Only endpoint throttles and restore failures can repeat a response.
                        if !matches!(status.as_u16(), 429 | 502 | 503) {
                            return Err(AlienError::new(ErrorData::SandboxUnreachable {
                                operation: "sandbox.egress".to_string(),
                                reason: format!("agent refused egress identity: {last_seen}"),
                            }));
                        }
                        throttled = status == reqwest::StatusCode::TOO_MANY_REQUESTS;
                    }
                    // Do not retain request errors: they may contain the identity URL.
                    Err(_) => last_seen = "identity delivery connection failed".to_string(),
                }
                if !endpoint_backoff(&mut delay, deadline, throttled).await {
                    return Ok(false);
                }
            }
        };
        if let Ok(result) =
            tokio::time::timeout_at(tokio::time::Instant::from_std(deadline), delivery).await
        {
            if result? {
                return Ok(());
            }
        }
        Err(AlienError::new(ErrorData::SandboxUnreachable {
            operation: "sandbox.egress".into(),
            reason: format!("identity delivery exceeded the readiness deadline: {last_seen}"),
        }))
    }

    /// Read a refusal once, keeping bounded, redacted diagnostics. On health and egress,
    /// only the agent's typed JSON error distinguishes an agent 429 from an endpoint throttle.
    async fn endpoint_refusal_reason(
        response: &mut reqwest::Response,
        token: Option<&str>,
        policy: &str,
    ) -> Result<String> {
        let status = response.status();
        tokio::time::timeout(SANDBOX_READY_TIMEOUT, async {
            let mut body = Vec::new();
            while let Some(chunk) = response.chunk().await.into_alien_error().context(
                ErrorData::SandboxUnreachable {
                    operation: "sandbox.endpoint".into(),
                    reason: format!("HTTP {status}; response body could not be read"),
                },
            )? {
                let oversized = body.len() + chunk.len() > 4096;
                body.extend_from_slice(&chunk[..chunk.len().min(4096 - body.len())]);
                // Recognize the self-check refusal as soon as its prefix arrives.
                if status == reqwest::StatusCode::SERVICE_UNAVAILABLE
                    && body.starts_with(b"EGRESS_PATH_UNEXPECTED:")
                {
                    return Err(AlienError::new(ErrorData::SandboxEgressPathUnexpected));
                }
                if oversized {
                    return Ok("agent refusal body exceeded 4096 bytes".to_string());
                }
            }
            let agent_throttle = status == reqwest::StatusCode::TOO_MANY_REQUESTS
                && serde_json::from_slice::<agent_protocol::JobErrorResponse>(&body).is_ok();
            let mut body = String::from_utf8_lossy(&body).into_owned();
            if !policy.is_empty() {
                body = body.replace(policy, "[redacted]");
            }
            if let Some(token) = token {
                body = body.replace(token, "[redacted]");
            }
            let body: String = body.chars().filter(|c| !c.is_control()).collect();
            let body = body.trim().to_string();
            if agent_throttle {
                return Err(AlienError::new(ErrorData::SandboxCommandFailed {
                    failure: "agentRefused".into(),
                    reason: format!("agent returned HTTP {status}: {body}"),
                }));
            }
            Ok(body)
        })
        .await
        .into_alien_error()
        .context(ErrorData::SandboxUnreachable {
            operation: "sandbox.endpoint".into(),
            reason: format!("HTTP {status}; response body timed out"),
        })?
    }

    async fn renew_egress(&self, sandbox_id: &str) -> Result<()> {
        if let Some(manager) = &self.egress_manager {
            let vm = self
                .owned_microvm("sandbox.egress", sandbox_id)
                .await?
                .ok_or_else(|| {
                    AlienError::new(ErrorData::SandboxUnreachable {
                        operation: "sandbox.egress".into(),
                        reason: "sandbox does not belong to this resource".into(),
                    })
                })?;
            let started = vm
                .started_at
                .as_ref()
                .and_then(serde_json::Number::as_f64)
                .filter(|value| value.is_finite() && *value >= 0.0 && *value < i64::MAX as f64)
                .ok_or_else(|| {
                    AlienError::new(ErrorData::SandboxUnreachable {
                        operation: "sandbox.egress".into(),
                        reason: "AWS returned no valid sandbox start time".into(),
                    })
                })? as i64;
            let lifetime = vm
                .maximum_duration_in_seconds
                .filter(|value| *value > 0)
                .ok_or_else(|| {
                    AlienError::new(ErrorData::SandboxUnreachable {
                        operation: "sandbox.egress".into(),
                        reason: "AWS returned no sandbox lifetime".into(),
                    })
                })?;
            let expires_at = started.saturating_add(lifetime.min(28_800));
            self.install_egress(sandbox_id, manager.renew(sandbox_id, expires_at).await?)
                .await?;
        }
        Ok(())
    }

    /// Builds a provider over the MicroVMs API.
    pub fn new(
        microvms: Arc<dyn LambdaMicrovmsApi>,
        image_identifier: impl Into<String>,
        image_version: impl Into<String>,
        egress_connector_arns: Vec<String>,
        preview_ports: Vec<u16>,
        idle_pause_seconds: Option<u32>,
        max_lifetime_seconds: Option<u32>,
    ) -> Self {
        Self {
            egress: None,
            egress_manager: None,
            default_policy: None,
            microvms,
            image_identifier: image_identifier.into(),
            image_version: image_version.into(),
            egress_connector_arns,
            preview_ports,
            idle_pause_seconds,
            max_lifetime_seconds,
            agent: reqwest::Client::new(),
        }
    }

    /// The lifetime `RunMicrovm` is asked for: what the caller asked for, never above what the
    /// declaration allows. AWS cannot move a running MicroVM's deadline, so this is the only
    /// point at which it can be chosen.
    fn lifetime_seconds(&self, timeout_ms: Option<u64>, operation: &str) -> Result<Option<u32>> {
        match timeout_ms {
            Some(timeout_ms) => {
                super::requested_lifetime_seconds(timeout_ms, self.max_lifetime_seconds, operation)
                    .map(Some)
            }
            None => Ok(self.max_lifetime_seconds),
        }
    }

    /// Reads a sandbox's record, with no ownership check: absent is `None`, and whose it is
    /// stays for the caller to ask [`Self::owns`]. Read off the sandbox itself rather than by
    /// enumerating the image, because listing would need an account-wide grant.
    async fn fetched_microvm(&self, operation: &str, sandbox_id: &str) -> Result<Option<Microvm>> {
        let microvm = match self.microvms.get_microvm(sandbox_id).await {
            Ok(microvm) => microvm,
            // A sandbox id that names nothing is absent, not a failure — `get` reports that as
            // `None`. Read as the variant rather than as `http_status_code`: the client's own
            // status-bearing variant declares no status of its own, so every error would arrive
            // as 500 and this arm would never match.
            Err(error)
                if matches!(
                    &error.error,
                    Some(alien_client_core::ErrorData::RemoteResourceNotFound { .. })
                ) =>
            {
                return Ok(None)
            }
            Err(error) => {
                return Err(error)
                    .unreachable(operation, &format!("could not read sandbox '{sandbox_id}'"))
            }
        };

        Ok(Some(microvm))
    }

    /// Whether a fetched record is one of this binding's own sandboxes.
    ///
    /// The image ARN is the boundary — one per declared sandbox by construction, and IAM does not
    /// draw this line: the token mint is scoped to `microvm-image:<stack prefix>-*`, which matches
    /// every sibling. An absent `imageArn` is refused rather than assumed to match, so a response
    /// the client failed to parse never passes as ownership.
    fn owns(&self, microvm: &Microvm) -> bool {
        microvm
            .image_arn
            .as_deref()
            .is_some_and(|image| image == self.image_identifier)
    }

    /// The sandbox's record if it exists *and* is this binding's own; absent and someone else's
    /// collapse to `None`, which is what every operation except `terminate` and `resume` wants.
    async fn owned_microvm(&self, operation: &str, sandbox_id: &str) -> Result<Option<Microvm>> {
        Ok(self
            .fetched_microvm(operation, sandbox_id)
            .await?
            .filter(|microvm| self.owns(microvm)))
    }

    /// The sandbox's record, refusing one that is absent or not one of this binding's own.
    async fn ensure_owned(&self, operation: &str, sandbox_id: &str) -> Result<Microvm> {
        self.owned_microvm(operation, sandbox_id)
            .await?
            .ok_or_else(|| {
                AlienError::new(ErrorData::SandboxUnreachable {
                    operation: operation.to_string(),
                    reason: format!(
                        "sandbox '{sandbox_id}' does not belong to this sandbox resource"
                    ),
                })
            })
    }

    /// Builds a request to the agent inside one sandbox, authorised and port-scoped.
    ///
    /// This is the whole of what AWS does differently; everything after it is the shared agent
    /// protocol. Two AWS calls per request, because the mint response carries no expiry — without
    /// one, caching a token means guessing how long it stays valid.
    async fn authorized_request(
        &self,
        sandbox_id: &str,
        method: reqwest::Method,
        path: &str,
    ) -> Result<reqwest::RequestBuilder> {
        // One read serves both: the record that proves the sandbox is ours also carries the
        // endpoint to reach it.
        let microvm = self
            .owned_microvm("sandbox.agent", sandbox_id)
            .await?
            .ok_or_else(|| {
                AlienError::new(ErrorData::SandboxUnreachable {
                    operation: "sandbox.agent".to_string(),
                    reason: format!(
                        "sandbox '{sandbox_id}' does not belong to this sandbox resource"
                    ),
                })
            })?;

        let endpoint = microvm.endpoint.ok_or_else(|| {
            AlienError::new(ErrorData::SandboxUnreachable {
                operation: "sandbox.agent".to_string(),
                reason: format!("MicroVM '{sandbox_id}' has no endpoint yet"),
            })
        })?;

        let token = self
            .microvms
            .create_microvm_auth_token(sandbox_id, vec![AGENT_PORT], AGENT_TOKEN_MINUTES)
            .await
            .unreachable(
                "sandbox.agent",
                &format!("could not mint an endpoint token for '{sandbox_id}'"),
            )?;

        let mut request = self
            .agent
            .request(method, format!("https://{endpoint}{path}"))
            .header(PROXY_PORT_HEADER, AGENT_PORT.to_string());

        // The mint returns a header map, not a bearer string. Sending it as `Authorization:
        // Bearer` yields a 403 that reads like a permissions problem.
        for (name, value) in token.auth_token {
            request = request.header(name, value);
        }

        Ok(request)
    }

    fn instance(&self, microvm_id: String, state: Option<String>) -> SandboxInstance {
        SandboxInstance {
            sandbox_id: microvm_id,
            state: sandbox_state(state.as_deref()),
            // Terminate destroys the MicroVM rather than fencing it, so a sandbox never outlives
            // its own generation and there is nothing for a second one to mean.
            generation: 1,
        }
    }
}

/// Only endpoint throttles back off exponentially. Restore and connection failures reset
/// to fixed polling. Stop immediately if the next attempt cannot start within the budget.
async fn endpoint_backoff(
    delay: &mut std::time::Duration,
    deadline: std::time::Instant,
    throttled: bool,
) -> bool {
    if !throttled {
        *delay = SANDBOX_READY_POLL;
    }
    if *delay >= deadline.saturating_duration_since(std::time::Instant::now()) {
        return false;
    }
    tokio::time::sleep(*delay).await;
    if throttled {
        *delay = delay.saturating_mul(2).min(SANDBOX_READY_POLL * 8);
    }
    std::time::Instant::now() < deadline
}

/// Whether `error` is an operation refusing a sandbox because it has terminated.
fn is_sandbox_terminated(error: &AlienError<ErrorData>) -> bool {
    matches!(
        &error.error,
        Some(ErrorData::SandboxCommandFailed { failure, .. }) if failure == "sandboxTerminated"
    )
}

/// Maps a MicroVM lifecycle state onto the binding's.
fn sandbox_state(state: Option<&str>) -> SandboxState {
    match state {
        Some("RUNNING") => SandboxState::Running,
        Some("SUSPENDED") => SandboxState::Paused,
        Some("TERMINATED") | Some("TERMINATING") => SandboxState::Terminated,
        // Anything else is a MicroVM on its way up. Reporting Running would tell a caller to
        // start sending commands to something that cannot answer yet.
        _ => SandboxState::Starting,
    }
}

impl AwsSandbox {
    /// Reads one of this binding's own sandboxes, or `None` if there is no such sandbox.
    ///
    /// `operation` is the verb the caller invoked, not this lookup: `get_or_create` reconnects
    /// through here, and reporting its failure as `sandbox.get` would name a call the caller
    /// never made.
    async fn fetch(&self, sandbox_id: &str, operation: &str) -> Result<Option<SandboxInstance>> {
        let Some(microvm) = self.owned_microvm(operation, sandbox_id).await? else {
            return Ok(None);
        };

        // Echoing the caller's own id when the response carried none would report a sandbox the
        // client could not parse as a sandbox it read — the same substitution `owned_microvm`
        // refuses for the image.
        let microvm_id = microvm.microvm_id.ok_or_else(|| {
            AlienError::new(ErrorData::SandboxUnreachable {
                operation: operation.to_string(),
                reason: format!("the record for sandbox '{sandbox_id}' carried no id"),
            })
        })?;

        Ok(Some(self.instance(microvm_id, microvm.state)))
    }

    /// Hands back a live sandbox `get_or_create` found, applying the requested policy first.
    async fn reconnect(
        &self,
        existing: SandboxInstance,
        egress: Option<alien_core::EgressPolicy>,
    ) -> Result<SandboxInstance> {
        let id = existing.sandbox_id.clone();
        if let Some(policy) = egress {
            self.update_egress(&id, policy).await?;
        }
        // Reaching a sandbox someone else started still has to mean what `create` means, or the
        // guarantee holds only for whoever won the race. Waited for here rather than in `get`,
        // which reports a sandbox's state and does not promise one.
        if matches!(existing.state, SandboxState::Starting) {
            self.wait_until_servable("sandbox.create", &id).await?;
            return Ok(self.instance(id, Some("RUNNING".to_string())));
        }
        Ok(existing)
    }

    /// Blocks until the agent answers, so `create` and `resume` return a sandbox that can take
    /// work. `operation` is the verb the caller invoked, which a failure here is reported under.
    async fn wait_until_servable(&self, operation: &str, sandbox_id: &str) -> Result<()> {
        let deadline = std::time::Instant::now() + SANDBOX_READY_TIMEOUT;

        // Only the endpoint's absence is worth waiting on. A refused token mint, a sandbox that is
        // not ours, an API error — none of those resolve by waiting, and folding them into the
        // timeout would report a permission problem as a slow boot a minute later.
        let probe = loop {
            let published = self
                .owned_microvm(operation, sandbox_id)
                .await?
                .is_some_and(|microvm| microvm.endpoint.is_some());

            if published {
                break self
                    .authorized_request(sandbox_id, reqwest::Method::GET, HEALTH_PATH)
                    .await?;
            }
            if std::time::Instant::now() >= deadline {
                return Err(AlienError::new(ErrorData::SandboxUnreachable {
                    operation: operation.to_string(),
                    reason: format!(
                        "MicroVM '{sandbox_id}' published no endpoint within {}s",
                        SANDBOX_READY_TIMEOUT.as_secs()
                    ),
                }));
            }
            tokio::time::sleep(SANDBOX_READY_POLL).await;
        };

        Self::poll_until_healthy(probe, deadline, operation, sandbox_id).await
    }

    /// Repeats the health probe until it answers or the deadline passes.
    ///
    /// Separated from the endpoint wait so the restore window this absorbs — AWS answering 502
    /// while a snapshot restores — can be exercised without a MicroVM.
    async fn poll_until_healthy(
        probe: reqwest::RequestBuilder,
        deadline: std::time::Instant,
        operation: &str,
        sandbox_id: &str,
    ) -> Result<()> {
        let mut last_seen = String::new();
        let mut delay = SANDBOX_READY_POLL;
        loop {
            // Cloned rather than rebuilt: the token outlives this wait, so the health poll costs
            // no further describes or mints.
            let request = probe.try_clone().ok_or_else(|| {
                AlienError::new(ErrorData::SandboxUnreachable {
                    operation: operation.to_string(),
                    reason: "the readiness probe could not be repeated".to_string(),
                })
            })?;

            // Bounded by the deadline: a connection the proxy accepts and never answers would
            // otherwise hold the wait past it, and the caller would hear nothing at all.
            let sent =
                tokio::time::timeout_at(tokio::time::Instant::from_std(deadline), request.send())
                    .await;
            let mut throttled = false;
            match sent {
                Ok(Ok(response)) if response.status().is_success() => return Ok(()),
                Ok(Ok(mut response)) => {
                    let status = response.status();
                    last_seen = format!("HTTP {status}");
                    let reason = tokio::time::timeout_at(
                        tokio::time::Instant::from_std(deadline),
                        Self::endpoint_refusal_reason(&mut response, None, ""),
                    )
                    .await
                    .into_alien_error()
                    .context(ErrorData::SandboxUnreachable {
                        operation: operation.into(),
                        reason: format!("MicroVM '{sandbox_id}' response exceeded the readiness deadline: {last_seen}"),
                    })??;
                    if !reason.is_empty() {
                        last_seen.push_str(&format!(": {reason}"));
                    }
                    throttled = status == reqwest::StatusCode::TOO_MANY_REQUESTS;
                }
                Ok(Err(error)) => last_seen = error.without_url().to_string(),
                // Keeps an earlier answer, which says more about the sandbox than the deadline.
                Err(_) if last_seen.is_empty() => {
                    last_seen = "the probe got no answer before the deadline".to_string()
                }
                Err(_) => {}
            }

            if !endpoint_backoff(&mut delay, deadline, throttled).await {
                return Err(AlienError::new(ErrorData::SandboxUnreachable {
                    operation: operation.to_string(),
                    reason: format!(
                        "MicroVM '{sandbox_id}' did not become servable in time: {last_seen}"
                    ),
                }));
            }
        }
    }
}

#[async_trait]
impl AgentTransport for AwsSandbox {
    async fn request(
        &self,
        sandbox_id: &str,
        method: reqwest::Method,
        path: &str,
    ) -> Result<reqwest::RequestBuilder> {
        self.authorized_request(sandbox_id, method, path).await
    }

    fn provider(&self) -> &'static str {
        "aws-sandbox"
    }
}

impl Binding for AwsSandbox {}

/// Refused rather than dropped: `RunMicrovm` has nowhere to put either, and a sandbox-level
/// value that silently never applies is worse than no sandbox. Per-command `env` on
/// `RunCommandRequest` is the path that works here.
fn refuse_unsupported_create_fields(request: &CreateSandboxRequest, operation: &str) -> Result<()> {
    if !request.env.is_empty() {
        return Err(AlienError::new(ErrorData::OperationNotSupported {
            operation: operation.to_string(),
            reason: "AWS sandboxes take no sandbox-level env; set env per command instead"
                .to_string(),
        }));
    }
    if request.tenant_key.is_some() {
        return Err(AlienError::new(ErrorData::OperationNotSupported {
            operation: operation.to_string(),
            reason: "AWS sandboxes take no tenantKey; a MicroVM is already single-tenant"
                .to_string(),
        }));
    }
    Ok(())
}

#[async_trait]
impl Sandbox for AwsSandbox {
    fn egress_declaration(&self) -> Option<&alien_core::SandboxEgress> {
        self.egress.as_ref()
    }
    /// Narrows the platform ceiling to this instance: `preview` is false with no declared
    /// `preview_ports`, since `preview()` would refuse every port anyway and callers are meant
    /// to branch on this flag rather than call and fail.
    fn capabilities(&self) -> SandboxCapabilities {
        let mut capabilities =
            SandboxCapabilities::for_platform(Platform::Aws).expect("AWS has a sandbox backend");
        capabilities.preview = !self.preview_ports.is_empty();
        capabilities
    }

    /// Starts a MicroVM.
    ///
    /// The client token is fresh per attempt and is **never** the caller's `sandbox_id`. AWS
    /// returns the MicroVM a token previously created even after it has been terminated, so a
    /// caller reusing a sandbox id would receive a dead MicroVM and wait for one that will never
    /// start — observed against the live API. Reconnecting to an existing sandbox is
    /// [`Sandbox::get`]'s job, not an idempotency key's.
    async fn create(&self, request: CreateSandboxRequest) -> Result<SandboxInstance> {
        self.validate_egress(request.egress.as_ref())?;
        let _ = request.sandbox_id;
        refuse_unsupported_create_fields(&request, "sandbox.create")?;
        let max_lifetime_seconds = self.lifetime_seconds(request.timeout_ms, "sandbox.create")?;
        // AWS has an eight-hour hard lifetime ceiling, including when no smaller limit is set.
        let expires_at = crate::sandbox_egress::now_unix()?.saturating_add(i64::from(
            max_lifetime_seconds.unwrap_or(28_800).min(28_800),
        ));
        let client_token = uuid::Uuid::new_v4().simple().to_string();

        let microvm = self
            .microvms
            .run_microvm(
                &self.image_identifier,
                &self.image_version,
                &client_token,
                // Never a role: a sandbox reads an attached role's credentials from instance
                // metadata, which the egress connector does not govern.
                None,
                self.egress_connector_arns.clone(),
                self.idle_pause_seconds,
                max_lifetime_seconds,
                self.egress_manager
                    .as_ref()
                    .map(|manager| manager.bootstrap_payload()),
            )
            .await
            .unreachable(
                "sandbox.create",
                &format!("could not start a MicroVM from '{}'", self.image_identifier),
            )?;

        let microvm_id = microvm.microvm_id.ok_or_else(|| {
            AlienError::new(ErrorData::UnexpectedResponseFormat {
                provider: "aws-sandbox".to_string(),
                binding_name: "sandbox.create".to_string(),
                field: "microvmId".to_string(),
                response_json: "RunMicrovm returned no MicroVM id".to_string(),
            })
        })?;

        // RunMicrovm returns once the MicroVM is accepted, not once it can serve: AWS restores the
        // snapshot afterwards and its proxy answers 502 until that finishes. Returning here hands
        // back a sandbox whose first command races that window, which is invisible to a caller and
        // fails a fraction of the time. Local, Azure and Kubernetes all return a sandbox that can
        // already serve, so this is what makes AWS mean the same thing.
        let ready = async {
            if let Some(manager) = &self.egress_manager {
                let policy = request
                    .egress
                    .as_ref()
                    .or(self.default_policy.as_ref())
                    .unwrap_or(&alien_core::EgressPolicy::Internet);
                let delivery = manager.create(&microvm_id, policy, expires_at).await?;
                self.install_egress(&microvm_id, delivery).await?;
            }
            self.wait_until_servable("sandbox.create", &microvm_id)
                .await
        }
        .await;
        if let Err(mut error) = ready {
            if let Some(manager) = &self.egress_manager {
                if let Err(cleanup) = manager.delete(&microvm_id).await {
                    error = cleanup.context(ErrorData::SandboxUnreachable {
                        operation: "sandbox.create".into(),
                        reason: format!(
                            "create failed ({error}); policy cleanup failed for '{microvm_id}'"
                        ),
                    });
                }
            }
            return Err(match self.microvms.terminate_microvm(&microvm_id).await {
                Ok(()) => error,
                // The wait can end on a MicroVM that already ended, which AWS may have reaped.
                Err(cleanup)
                    if matches!(
                        &cleanup.error,
                        Some(alien_client_core::ErrorData::RemoteResourceNotFound { .. })
                    ) =>
                {
                    error
                }
                Err(cleanup) => {
                    warn!(
                        microvm = %microvm_id,
                        %cleanup,
                        "could not terminate a MicroVM that never became servable"
                    );
                    // Names the leak and refuses a retry: no caller holds this MicroVM's id, so
                    // honouring the wait's retryable error would mint another orphan beside it.
                    error.context(ErrorData::SandboxCommandFailed {
                        failure: "sandboxLeftBehind".to_string(),
                        reason: format!(
                            "MicroVM '{microvm_id}' was not handed to its caller and could not be \
                             terminated, so it is still running"
                        ),
                    })
                }
            });
        }

        Ok(self.instance(microvm_id, Some("RUNNING".to_string())))
    }

    async fn update_egress(
        &self,
        sandbox_id: &str,
        policy: alien_core::EgressPolicy,
    ) -> Result<()> {
        self.validate_egress(Some(&policy))?;
        let manager = self
            .egress_manager
            .as_ref()
            .ok_or_else(|| super::egress::unsupported("update_egress"))?;
        let sandbox = self
            .fetch(sandbox_id, "sandbox.updateEgress")
            .await?
            .ok_or_else(|| {
                AlienError::new(ErrorData::SandboxUnreachable {
                    operation: "sandbox.updateEgress".into(),
                    reason: format!(
                        "sandbox '{sandbox_id}' does not belong to this sandbox resource"
                    ),
                })
            })?;
        // A terminated MicroVM never runs again, so a policy written for it governs nothing and
        // would re-create the gateway record its end should have removed.
        if sandbox.state == SandboxState::Terminated {
            return Err(AlienError::new(ErrorData::SandboxCommandFailed {
                failure: "sandboxTerminated".to_string(),
                reason: format!("sandbox '{sandbox_id}' has terminated"),
            }));
        }
        manager.update(sandbox_id, &policy).await?;
        if sandbox.state == SandboxState::Running {
            // Keep the live identity unchanged while refreshing the agent's copy for
            // recovery after total gateway loss. Paused agents receive it on resume.
            let record = manager.policy_record(sandbox_id).await?;
            self.deliver_egress(sandbox_id, None, &record).await?;
        }
        Ok(())
    }

    async fn get(&self, sandbox_id: &str) -> Result<Option<SandboxInstance>> {
        self.fetch(sandbox_id, "sandbox.get").await
    }

    async fn get_or_create(&self, request: CreateSandboxRequest) -> Result<ResolvedSandbox> {
        self.validate_egress(request.egress.as_ref())?;
        // Refused on the reconnect path too: an existing sandbox honors these fields no more
        // than a fresh one would.
        refuse_unsupported_create_fields(&request, "sandbox.getOrCreate")?;
        if let Some(id) = request.sandbox_id.as_deref() {
            // A MicroVM that is terminating or terminated never serves again, so the id is
            // answered with a fresh sandbox, as on GCP and Azure. It can also end between the read
            // here and the one `update_egress` makes, which refuses it as `sandboxTerminated`; that
            // is the same answer reached a moment later.
            let terminated = match self.fetch(id, "sandbox.getOrCreate").await? {
                Some(existing) if existing.state == SandboxState::Terminated => true,
                Some(existing) => match self.reconnect(existing, request.egress.clone()).await {
                    Ok(found) => return Ok(ResolvedSandbox::found(found)),
                    Err(error) if is_sandbox_terminated(&error) => true,
                    Err(error) => return Err(error),
                },
                None => false,
            };
            // Lambda ends a MicroVM at its maximum lifetime without anyone calling `terminate`,
            // so the gateway can still hold this id's policy; it is removed here, as `terminate`
            // would have.
            if terminated {
                if let Some(manager) = &self.egress_manager {
                    manager.delete(id).await?;
                }
            }
        }

        self.create(request).await.map(ResolvedSandbox::created)
    }

    /// Not offered, as on Azure.
    ///
    /// Enumerating would mean `lambda:ListMicrovms`, which AWS authorizes against no resource
    /// type — the grant could only be account-wide, on the management profile any stack with a
    /// sandbox holds. The
    /// reason to accept that would be recovering a MicroVM whose `RunMicrovm` response never
    /// arrived, since nobody holds its id. AWS suspends a MicroVM for lack of traffic only under a
    /// declared idle policy, so without one such an orphan runs until its lifetime ends, eight
    /// hours at the latest. Reconnecting to a sandbox whose id *is* known is `get`, which reads it
    /// directly.
    async fn list(&self) -> Result<Vec<SandboxInstance>> {
        Err(AlienError::new(ErrorData::OperationNotSupported {
            operation: "sandbox.list".to_string(),
            reason:
                "enumerating sandboxes would need an account-wide grant; reach a known sandbox \
                     with get"
                    .to_string(),
        }))
    }

    async fn run_command(
        &self,
        sandbox_id: &str,
        request: RunCommandRequest,
    ) -> Result<BoxStream<'static, Result<CommandOutput>>> {
        agent_protocol::run_command(self, sandbox_id, request).await
    }

    async fn start_job(&self, sandbox_id: &str, request: RunCommandRequest) -> Result<JobStart> {
        agent_protocol::start_job(self, sandbox_id, request).await
    }

    async fn poll_job(
        &self,
        sandbox_id: &str,
        job_id: &str,
        since_seq: Option<u64>,
    ) -> Result<JobPoll> {
        agent_protocol::poll_job(self, sandbox_id, job_id, since_seq).await
    }

    async fn cancel_job(&self, sandbox_id: &str, job_id: &str) -> Result<()> {
        agent_protocol::cancel_job(self, sandbox_id, job_id).await
    }

    async fn read_file(&self, sandbox_id: &str, path: &str) -> Result<Vec<u8>> {
        agent_protocol::read_file(self, sandbox_id, path).await
    }

    async fn write_files(&self, sandbox_id: &str, files: BTreeMap<String, Vec<u8>>) -> Result<()> {
        agent_protocol::write_files(self, sandbox_id, files).await
    }

    /// Mints a capability to reach one port inside the sandbox.
    ///
    /// The endpoint is never returned bare: a caller cannot reach it without the token headers
    /// and the port header, and handing over a URL would push them into building the auth
    /// themselves.
    async fn preview(&self, sandbox_id: &str, port: u16) -> Result<PreviewCapability> {
        // Port first, ownership second: an undeclared port is refused without spending a call.
        if !self.preview_ports.contains(&port) {
            return Err(AlienError::new(ErrorData::OperationNotSupported {
                operation: "sandbox.preview".to_string(),
                reason: format!(
                    "port {port} is not one of this sandbox's declared preview ports {:?}; a \
                     minted token would grant ingress the stack never asked for",
                    self.preview_ports
                ),
            }));
        }

        // One read again: ownership and the endpoint come off the same record.
        let microvm = self
            .owned_microvm("sandbox.preview", sandbox_id)
            .await?
            .ok_or_else(|| {
                AlienError::new(ErrorData::SandboxUnreachable {
                    operation: "sandbox.preview".to_string(),
                    reason: format!(
                        "sandbox '{sandbox_id}' does not belong to this sandbox resource"
                    ),
                })
            })?;

        let endpoint = microvm.endpoint.ok_or_else(|| {
            AlienError::new(ErrorData::SandboxUnreachable {
                operation: "sandbox.preview".to_string(),
                reason: format!("MicroVM '{sandbox_id}' has no endpoint yet"),
            })
        })?;

        let token = self
            .microvms
            .create_microvm_auth_token(sandbox_id, vec![port], PREVIEW_TOKEN_MINUTES)
            .await
            .unreachable(
                "sandbox.preview",
                &format!("could not mint a preview token for port {port}"),
            )?;

        let mut headers: BTreeMap<String, String> = token.auth_token.into_iter().collect();
        headers.insert(PROXY_PORT_HEADER.to_string(), port.to_string());

        Ok(PreviewCapability {
            endpoint: format!("https://{endpoint}"),
            headers,
            allowed_ports: vec![port],
            expires_in_seconds: preview_lifetime_seconds(),
        })
    }

    async fn pause(&self, sandbox_id: &str) -> Result<()> {
        self.ensure_owned("sandbox.pause", sandbox_id).await?;

        self.microvms.suspend_microvm(sandbox_id).await.unreachable(
            "sandbox.pause",
            &format!("could not pause MicroVM '{sandbox_id}'"),
        )
    }

    async fn resume(&self, sandbox_id: &str) -> Result<()> {
        // An ended MicroVM never serves again, so resuming it cannot succeed and a retryable
        // answer would invite a retry. AWS drops a terminated MicroVM's record, so absence ends
        // it too.
        let ended = |what: &str| {
            AlienError::new(ErrorData::SandboxCommandFailed {
                failure: "sandboxTerminated".to_string(),
                reason: format!("sandbox.resume: MicroVM '{sandbox_id}' {what}"),
            })
        };
        let microvm = match self.fetched_microvm("sandbox.resume", sandbox_id).await? {
            None => {
                return Err(ended(
                    "does not exist: it was terminated and reaped, or never created",
                ))
            }
            Some(microvm) if !self.owns(&microvm) => {
                return Err(AlienError::new(ErrorData::SandboxUnreachable {
                    operation: "sandbox.resume".to_string(),
                    reason: format!(
                        "sandbox '{sandbox_id}' does not belong to this sandbox resource"
                    ),
                }))
            }
            Some(microvm) => microvm,
        };
        if let Some(state @ ("TERMINATING" | "TERMINATED")) = microvm.state.as_deref() {
            return Err(ended(&format!("is {state}")));
        }

        // A retry after a resume whose wait gave up finds the MicroVM already running. Asking
        // AWS to resume it again could be refused for a sandbox that is up.
        if microvm.state.as_deref() != Some("RUNNING") {
            self.microvms.resume_microvm(sandbox_id).await.unreachable(
                "sandbox.resume",
                &format!("could not resume MicroVM '{sandbox_id}'"),
            )?;
        }
        self.renew_egress(sandbox_id).await?;

        // ResumeMicrovm is accepted while the MicroVM is still SUSPENDED; it takes traffic only
        // once its /resume hook returns, so a command sent before then hits the proxy's 502. Not
        // terminated on a failed wait, unlike `create`: the caller holds this sandbox's id.
        self.wait_until_servable("sandbox.resume", sandbox_id).await
    }

    async fn snapshot(&self, _sandbox_id: &str) -> Result<String> {
        Err(AlienError::new(ErrorData::OperationNotSupported {
            operation: "sandbox.snapshot".to_string(),
            reason: "Lambda MicroVMs expose no snapshot API".to_string(),
        }))
    }

    /// Idempotent per the trait: a sandbox AWS has already reaped is terminated, not an error.
    /// Absence and "someone else's" part ways here and in `resume`; every other operation needs
    /// the sandbox to exist, so for them the two are the same refusal.
    async fn terminate(&self, sandbox_id: &str) -> Result<()> {
        let state = match self
            .fetched_microvm("sandbox.terminate", sandbox_id)
            .await?
        {
            None => {
                if let Some(manager) = &self.egress_manager {
                    manager.delete(sandbox_id).await?;
                }
                return Ok(());
            }
            Some(microvm) if !self.owns(&microvm) => {
                return Err(AlienError::new(ErrorData::SandboxUnreachable {
                    operation: "sandbox.terminate".to_string(),
                    reason: format!(
                        "sandbox '{sandbox_id}' does not belong to this sandbox resource"
                    ),
                }))
            }
            Some(microvm) => microvm.state,
        };

        let policy_cleanup = match &self.egress_manager {
            Some(manager) => manager.delete(sandbox_id).await,
            None => Ok(()),
        };

        // A retry after an unconfirmed terminate finds the request already in flight or done.
        // Asking AWS again could be refused for a sandbox that is gone or going.
        if !matches!(state.as_deref(), Some("TERMINATING" | "TERMINATED")) {
            match self.microvms.terminate_microvm(sandbox_id).await {
                Ok(()) => {}
                // Reaped between the read and the request; the poll below confirms it.
                Err(error)
                    if matches!(
                        &error.error,
                        Some(alien_client_core::ErrorData::RemoteResourceNotFound { .. })
                    ) => {}
                Err(error) => {
                    return Err::<(), _>(error).unreachable(
                        "sandbox.terminate",
                        &format!("could not terminate MicroVM '{sandbox_id}'"),
                    )
                }
            }
        }

        // TerminateMicrovm returns once AWS accepts it, not once the MicroVM is gone. Returning
        // here would report containment while the code inside is still running. Only `TERMINATED`
        // or an absent record confirms it. Each read is bounded by the deadline; one that fails
        // (after the client's own retries) says nothing about the sandbox, so it becomes the
        // error's source rather than a confirmation.
        let deadline = std::time::Instant::now() + SANDBOX_TERMINATED_TIMEOUT;
        let unconfirmed = |last_seen: &str| ErrorData::SandboxUnreachable {
            operation: "sandbox.terminate".to_string(),
            reason: format!(
                "termination of MicroVM '{sandbox_id}' was accepted, but {last_seen}; it may \
                 still be running"
            ),
        };
        let mut last_state: Option<String> = None;
        loop {
            // Checked before each read as well as after it: a sleep that ends past the deadline
            // would otherwise send a read with no time left, and report that rather than the
            // state last seen.
            if let Some(state) = &last_state {
                if std::time::Instant::now() >= deadline {
                    return Err(AlienError::new(unconfirmed(&format!(
                        "after {}s it was still {state}",
                        SANDBOX_TERMINATED_TIMEOUT.as_secs()
                    ))));
                }
            }
            let read = tokio::time::timeout_at(
                tokio::time::Instant::from_std(deadline),
                self.fetched_microvm("sandbox.terminate", sandbox_id),
            )
            .await;
            let state = match read {
                Ok(Ok(None)) => break,
                Ok(Ok(Some(microvm))) => microvm.state,
                Ok(Err(error)) => {
                    return Err(error).context(unconfirmed("its record could not be read"))
                }
                Err(_) => {
                    return Err(AlienError::new(unconfirmed(
                        "its record could not be read before the deadline",
                    )))
                }
            };
            if state.as_deref() == Some("TERMINATED") {
                break;
            }
            last_state = Some(state.unwrap_or_else(|| "in a state it did not report".to_string()));
            tokio::time::sleep(SANDBOX_TERMINATED_POLL).await;
        }
        policy_cleanup
    }

    fn as_any(&self) -> &dyn std::any::Any {
        self
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use alien_aws_clients::aws::lambda_microvms::{
        Microvm, MicrovmAuthToken, MockLambdaMicrovmsApi,
    };
    use alien_error::Context;
    use std::time::Duration;

    /// Exercise both retrying endpoint operations over actual HTTP, including request replay.
    #[rstest::rstest]
    #[case::identity(false)]
    #[case::health(true)]
    #[tokio::test]
    async fn endpoint_429_retries_with_exponential_backoff(#[case] health: bool) {
        let attempts = Arc::new(std::sync::Mutex::new(Vec::new()));
        let handler = {
            let attempts = attempts.clone();
            move |method: axum::http::Method, body: String| {
                let attempts = attempts.clone();
                async move {
                    if !health {
                        assert_eq!(method, axum::http::Method::PUT);
                        assert_eq!(
                            body,
                            r#"{"policy":"signed-policy","token":"identity-token"}"#
                        );
                    }
                    let mut attempts = attempts.lock().expect("attempts");
                    attempts.push(std::time::Instant::now());
                    match attempts.len() {
                        1 | 3 => (axum::http::StatusCode::TOO_MANY_REQUESTS, ""),
                        2 => (
                            axum::http::StatusCode::TOO_MANY_REQUESTS,
                            "Rate limit exceeded",
                        ),
                        4 => (
                            axum::http::StatusCode::TOO_MANY_REQUESTS,
                            "Endpoint quota exceeded; try again",
                        ),
                        5 => (
                            axum::http::StatusCode::TOO_MANY_REQUESTS,
                            r#"{"message":"Too many requests"}"#,
                        ),
                        _ => (axum::http::StatusCode::OK, ""),
                    }
                }
            }
        };
        let (url, server) =
            endpoint_server(axum::Router::new().route("/", axum::routing::any(handler))).await;
        poll_endpoint(
            endpoint_request(&url, health),
            health,
            Duration::from_secs(2),
        )
        .await
        .expect("endpoint recovered");
        server.abort();
        let attempts = attempts.lock().expect("attempts");
        assert_eq!(attempts.len(), 6);
        for (index, pair) in attempts.windows(2).enumerate() {
            let expected = SANDBOX_READY_POLL * (1 << index.min(3));
            assert!(
                pair[1].duration_since(pair[0]) >= expected,
                "attempt {index} must back off by {expected:?}"
            );
        }
    }

    #[rstest::rstest]
    #[case::identity(false, "Rate limit exceeded")]
    #[case::health(true, "Rate limit exceeded")]
    #[case::identity_empty(false, "")]
    #[case::health_empty(true, "")]
    #[tokio::test]
    async fn endpoint_429_exhausts_budget_with_retryable_error(
        #[case] health: bool,
        #[case] reason: &'static str,
    ) {
        let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let handler = {
            let attempts = attempts.clone();
            move || {
                attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                async move { (axum::http::StatusCode::TOO_MANY_REQUESTS, reason) }
            }
        };
        let (url, server) =
            endpoint_server(axum::Router::new().route("/", axum::routing::any(handler))).await;
        let request = endpoint_request(&url, health);
        let started = std::time::Instant::now();
        let error = tokio::time::timeout(
            Duration::from_millis(500),
            poll_endpoint(request, health, Duration::from_millis(70)),
        )
        .await
        .expect("must respect the readiness budget")
        .expect_err("permanent throttle");
        server.abort();
        assert_eq!(error.code, "SANDBOX_UNREACHABLE");
        assert!(error.retryable, "quota can recover: {error}");
        assert!(started.elapsed() < Duration::from_millis(70));
        assert!(
            error.message.contains("HTTP 429 Too Many Requests"),
            "{error}"
        );
        assert!(error.message.contains(reason), "{error}");
        assert!(!error.message.ends_with(": "), "{error}");
        assert!(attempts.load(std::sync::atomic::Ordering::SeqCst) > 1);
    }

    #[rstest::rstest]
    #[case::identity(false)]
    #[case::health(true)]
    #[tokio::test]
    async fn endpoint_429_preserves_agent_refusal_without_retry(#[case] health: bool) {
        let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let handler = {
            let attempts = attempts.clone();
            move || {
                attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                async {
                    (
                        axum::http::StatusCode::TOO_MANY_REQUESTS,
                        r#"{"code":"JOB_LIMIT_REACHED","message":"No job slots available"}"#,
                    )
                }
            }
        };
        let (url, server) =
            endpoint_server(axum::Router::new().route("/", axum::routing::any(handler))).await;
        let error = poll_endpoint(
            endpoint_request(&url, health),
            health,
            Duration::from_secs(2),
        )
        .await
        .expect_err("agent refused");
        server.abort();
        assert_eq!(error.code, "SANDBOX_COMMAND_FAILED");
        assert!(error.message.contains("JOB_LIMIT_REACHED"));
        assert!(error.message.contains("No job slots available"));
        assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 1);
    }

    #[rstest::rstest]
    #[case::identity(false)]
    #[case::health(true)]
    #[tokio::test]
    async fn endpoint_deadline_preserves_throttle_before_a_silent_attempt(#[case] health: bool) {
        let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let handler = {
            let attempts = attempts.clone();
            move || {
                let attempt = attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                async move {
                    if attempt > 0 {
                        std::future::pending::<()>().await;
                    }
                    (
                        axum::http::StatusCode::TOO_MANY_REQUESTS,
                        "Endpoint quota exceeded",
                    )
                }
            }
        };
        let (url, server) =
            endpoint_server(axum::Router::new().route("/", axum::routing::any(handler))).await;
        let error = tokio::time::timeout(
            Duration::from_millis(500),
            poll_endpoint(
                endpoint_request(&url, health),
                health,
                Duration::from_millis(70),
            ),
        )
        .await
        .expect("a silent retry must respect the readiness deadline")
        .expect_err("silent endpoint");
        server.abort();
        assert_eq!(error.code, "SANDBOX_UNREACHABLE");
        assert!(error.retryable);
        assert!(
            error.message.contains("HTTP 429 Too Many Requests"),
            "{error}"
        );
        assert!(error.message.contains("Endpoint quota exceeded"), "{error}");
        assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 2);
    }

    /// A real HTTP restore window must fit the fixed-poll budget. Exponential restore
    /// delays (or failing to reset after a throttle) exhaust this budget before recovery.
    #[rstest::rstest]
    #[case::identity_restore(false, false)]
    #[case::health_restore(true, false)]
    #[case::identity_mixed(false, true)]
    #[case::health_mixed(true, true)]
    #[tokio::test]
    async fn endpoint_restore_resets_to_fixed_poll(#[case] health: bool, #[case] mixed: bool) {
        let attempts = Arc::new(std::sync::Mutex::new(Vec::new()));
        let handler = {
            let attempts = attempts.clone();
            move || {
                let mut attempts = attempts.lock().expect("attempts");
                attempts.push(std::time::Instant::now());
                let attempt = attempts.len();
                async move {
                    let status = if attempt == 21 {
                        axum::http::StatusCode::OK
                    } else if mixed && attempt % 3 != 0 {
                        axum::http::StatusCode::TOO_MANY_REQUESTS
                    } else if mixed && attempt % 2 == 0 {
                        axum::http::StatusCode::SERVICE_UNAVAILABLE
                    } else {
                        axum::http::StatusCode::BAD_GATEWAY
                    };
                    (status, "")
                }
            }
        };
        let (url, server) =
            endpoint_server(axum::Router::new().route("/", axum::routing::any(handler))).await;
        poll_endpoint(
            endpoint_request(&url, health),
            health,
            Duration::from_millis(400),
        )
        .await
        .expect("restore polling must recover within the fixed-cadence budget");
        server.abort();
        let attempts = attempts.lock().expect("attempts");
        assert_eq!(attempts.len(), 21);
        for (index, pair) in attempts.windows(2).enumerate() {
            let expected = if mixed && (index + 1) % 3 == 2 {
                SANDBOX_READY_POLL * 2
            } else {
                SANDBOX_READY_POLL
            };
            assert!(pair[1].duration_since(pair[0]) >= expected);
        }
    }

    async fn endpoint_server(router: axum::Router) -> (String, tokio::task::JoinHandle<()>) {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind");
        let address = listener.local_addr().expect("address");
        let server =
            tokio::spawn(async move { axum::serve(listener, router).await.expect("serve") });
        (format!("http://{address}/"), server)
    }

    fn endpoint_request(url: &str, health: bool) -> reqwest::RequestBuilder {
        let client = reqwest::Client::new();
        if health {
            client.get(url)
        } else {
            client
                .put(url)
                .json(&serde_json::json!({"policy":"signed-policy", "token":"identity-token"}))
        }
    }

    async fn poll_endpoint(
        request: reqwest::RequestBuilder,
        health: bool,
        budget: Duration,
    ) -> Result<()> {
        let deadline = std::time::Instant::now() + budget;
        if health {
            AwsSandbox::poll_until_healthy(request, deadline, "sandbox.create", "mvm-throttled")
                .await
        } else {
            AwsSandbox::poll_egress(request, deadline, Some("identity-token"), "signed-policy")
                .await
        }
    }

    fn image_version(version: &str) -> alien_aws_clients::aws::lambda_microvms::MicrovmImage {
        alien_aws_clients::aws::lambda_microvms::MicrovmImage {
            description: None,
            image_identifier: Some("sbx-image".to_string()),
            image_arn: None,
            image_version: Some(version.to_string()),
            state: Some("CREATED".to_string()),
        }
    }

    /// A MicroVM belonging to this sandbox's image. Ownership is now a field on the record, so
    /// every fixture has to say whose sandbox it is.
    fn owned(id: &str, state: &str) -> Microvm {
        Microvm {
            started_at: None,
            maximum_duration_in_seconds: None,
            microvm_id: Some(id.to_string()),
            endpoint: None,
            state: Some(state.to_string()),
            image_arn: Some("sbx-image".to_string()),
            image_version: Some("1".to_string()),
        }
    }

    fn sandbox(client: MockLambdaMicrovmsApi) -> AwsSandbox {
        sandbox_previewing(client, Vec::new())
    }

    fn sandbox_previewing(client: MockLambdaMicrovmsApi, preview_ports: Vec<u16>) -> AwsSandbox {
        AwsSandbox::new(
            Arc::new(client),
            "sbx-image",
            "3",
            Vec::new(),
            preview_ports,
            None,
            None,
        )
    }

    /// AWS reaps the microvm record after termination, so pinning idempotent-on-absent keeps a
    /// caller's retry loop from flaking once the record is gone.
    #[tokio::test]
    async fn terminating_an_absent_sandbox_succeeds() {
        let mut client = MockLambdaMicrovmsApi::new();
        client.expect_get_microvm().returning(|id| {
            Err(alien_error::AlienError::new(
                alien_client_core::ErrorData::RemoteResourceNotFound {
                    resource_type: "Microvm".to_string(),
                    resource_name: id.to_string(),
                },
            ))
        });
        client.expect_terminate_microvm().never();

        let sandbox = sandbox(client);

        sandbox
            .terminate("already-reaped")
            .await
            .expect("an absent sandbox is already terminated");
    }

    /// `TerminateMicrovm` returns once AWS accepts it, not once the MicroVM is gone. A caller told
    /// the sandbox is terminated while its record still says `RUNNING` would treat code that is
    /// still executing as contained.
    #[tokio::test]
    async fn terminate_does_not_report_a_sandbox_gone_while_it_still_runs() {
        let mut client = MockLambdaMicrovmsApi::new();
        client
            .expect_get_microvm()
            .returning(|id| Ok(owned(id, "RUNNING")));
        client
            .expect_terminate_microvm()
            .times(1)
            .returning(|_| Ok(()));

        let error = sandbox(client)
            .terminate("still-running")
            .await
            .expect_err("a sandbox whose record still says RUNNING is not terminated");

        assert_eq!(error.code, "SANDBOX_UNREACHABLE", "{error}");
        assert!(
            error.to_string().contains("still-running") && error.message.contains("still RUNNING"),
            "the failure names the sandbox and the state it was last seen in: {error}"
        );
    }

    /// The record is read once for ownership and then until it says `TERMINATED`. The read count
    /// is the assertion that `TERMINATING` is not taken for confirmation: accepting it would end
    /// the poll one read early.
    #[tokio::test]
    async fn terminate_returns_once_the_record_says_terminated() {
        let mut client = MockLambdaMicrovmsApi::new();
        let mut reads = 0;
        client.expect_get_microvm().times(4).returning(move |id| {
            reads += 1;
            let state = match reads {
                1 | 2 => "RUNNING",
                3 => "TERMINATING",
                _ => "TERMINATED",
            };
            Ok(owned(id, state))
        });
        client
            .expect_terminate_microvm()
            .times(1)
            .returning(|_| Ok(()));

        sandbox(client)
            .terminate("winding-down")
            .await
            .expect("a record that reaches TERMINATED confirms the sandbox is gone");
    }

    /// AWS may reap the record before the poll reads `TERMINATED`; absent after an accepted
    /// terminate is the sandbox gone, the same answer `terminating_an_absent_sandbox_succeeds`
    /// pins before one.
    #[tokio::test]
    async fn terminate_returns_once_the_record_is_gone() {
        let mut client = MockLambdaMicrovmsApi::new();
        let mut reads = 0;
        client.expect_get_microvm().returning(move |id| {
            reads += 1;
            if reads == 1 {
                return Ok(owned(id, "RUNNING"));
            }
            Err(alien_error::AlienError::new(
                alien_client_core::ErrorData::RemoteResourceNotFound {
                    resource_type: "Microvm".to_string(),
                    resource_name: id.to_string(),
                },
            ))
        });
        client
            .expect_terminate_microvm()
            .times(1)
            .returning(|_| Ok(()));

        sandbox(client)
            .terminate("reaped-quickly")
            .await
            .expect("a record AWS has reaped confirms the sandbox is gone");
    }

    /// What a retry after an unconfirmed terminate finds: the end state already reached. A second
    /// TerminateMicrovm for it could be refused, failing the one verb that promises idempotence.
    #[tokio::test]
    async fn terminate_of_a_terminated_record_sends_no_second_request() {
        let mut client = MockLambdaMicrovmsApi::new();
        client
            .expect_get_microvm()
            .returning(|id| Ok(owned(id, "TERMINATED")));
        client.expect_terminate_microvm().never();

        sandbox(client)
            .terminate("already-down")
            .await
            .expect("a terminated record is the end state terminate wants");
    }

    /// A request already in flight is waited on rather than repeated.
    #[tokio::test]
    async fn terminate_of_a_terminating_record_waits_without_asking_again() {
        let mut client = MockLambdaMicrovmsApi::new();
        let mut reads = 0;
        client.expect_get_microvm().returning(move |id| {
            reads += 1;
            Ok(owned(
                id,
                if reads == 1 {
                    "TERMINATING"
                } else {
                    "TERMINATED"
                },
            ))
        });
        client.expect_terminate_microvm().never();

        sandbox(client)
            .terminate("going-down")
            .await
            .expect("the in-flight request reaching TERMINATED confirms it");
    }

    /// A read that still fails after the client's own retries is reported, not taken for a
    /// confirmation, and its visibility carries over: what the caller sees is the read's own
    /// error as the source, not its text copied into a public message.
    #[tokio::test]
    async fn terminate_reports_a_failed_read_as_unconfirmed_with_its_source() {
        let mut client = MockLambdaMicrovmsApi::new();
        let mut reads = 0;
        client.expect_get_microvm().returning(move |id| {
            reads += 1;
            if reads == 1 {
                return Ok(owned(id, "RUNNING"));
            }
            Err(alien_error::AlienError::new(
                alien_client_core::ErrorData::RemoteServiceUnavailable {
                    message: "GetMicrovm was throttled".to_string(),
                },
            ))
        });
        client
            .expect_terminate_microvm()
            .times(1)
            .returning(|_| Ok(()));

        let error = sandbox(client)
            .terminate("unreadable-after")
            .await
            .expect_err("a record that cannot be read does not confirm termination");
        assert_eq!(error.code, "SANDBOX_UNREACHABLE", "{error}");
        assert!(
            error.message.contains("may still be running") && error.source.is_some(),
            "the unconfirmed outcome is the message, the failed read its source: {error}"
        );
        assert!(
            !error.message.contains("throttled"),
            "the read's own text stays in the source, not the public message: {error}"
        );
    }

    /// A MicroVM AWS reaped between the read and the request is gone, as terminate wanted; the
    /// read that follows is what confirms it.
    #[tokio::test]
    async fn terminate_treats_a_not_found_request_as_gone() {
        let mut client = MockLambdaMicrovmsApi::new();
        let mut reads = 0;
        client.expect_get_microvm().returning(move |id| {
            reads += 1;
            if reads == 1 {
                return Ok(owned(id, "RUNNING"));
            }
            Err(alien_error::AlienError::new(
                alien_client_core::ErrorData::RemoteResourceNotFound {
                    resource_type: "Microvm".to_string(),
                    resource_name: id.to_string(),
                },
            ))
        });
        client.expect_terminate_microvm().times(1).returning(|id| {
            Err(alien_error::AlienError::new(
                alien_client_core::ErrorData::RemoteResourceNotFound {
                    resource_type: "Microvm".to_string(),
                    resource_name: id.to_string(),
                },
            ))
        });

        sandbox(client)
            .terminate("reaped-in-between")
            .await
            .expect("a MicroVM already gone is terminated");
    }

    /// An ended MicroVM is refused before anything is asked of AWS, and as an answer a retry
    /// cannot change.
    #[tokio::test]
    async fn resume_of_an_ended_sandbox_is_refused_without_asking_aws() {
        let mut client = MockLambdaMicrovmsApi::new();
        client
            .expect_get_microvm()
            .returning(|id| Ok(owned(id, "TERMINATED")));
        client.expect_resume_microvm().never();

        let error = sandbox(client)
            .resume("already-ended")
            .await
            .expect_err("an ended sandbox cannot be resumed");
        assert_eq!(error.code, "SANDBOX_COMMAND_FAILED", "{error}");
        assert!(!error.retryable, "a retry cannot bring it back: {error}");
    }

    /// AWS drops a terminated MicroVM's record, so absence is how an ended sandbox usually looks.
    #[tokio::test]
    async fn resume_of_a_sandbox_aws_no_longer_knows_is_refused_without_asking_aws() {
        let mut client = MockLambdaMicrovmsApi::new();
        client.expect_get_microvm().returning(|id| {
            Err(alien_error::AlienError::new(
                alien_client_core::ErrorData::RemoteResourceNotFound {
                    resource_type: "Microvm".to_string(),
                    resource_name: id.to_string(),
                },
            ))
        });
        client.expect_resume_microvm().never();

        let error = sandbox(client)
            .resume("long-gone")
            .await
            .expect_err("a sandbox with no record cannot be resumed");
        assert_eq!(error.code, "SANDBOX_COMMAND_FAILED", "{error}");
        assert!(!error.retryable, "a retry cannot bring it back: {error}");
    }

    /// A retry after a resume whose wait gave up finds the MicroVM running; it is waited on, not
    /// resumed again.
    #[tokio::test]
    async fn resume_of_a_running_sandbox_does_not_ask_aws_again() {
        let mut client = MockLambdaMicrovmsApi::new();
        client
            .expect_get_microvm()
            .returning(|id| Ok(owned(id, "RUNNING")));
        client.expect_resume_microvm().never();

        let error = sandbox(client)
            .resume("already-up")
            .await
            .expect_err("a running sandbox that publishes no endpoint still cannot be served");
        assert_eq!(error.code, "SANDBOX_UNREACHABLE", "{error}");
    }

    /// Reconnecting must not report a sandbox this call did not make, or a caller runs its
    /// first-run setup a second time on a sandbox that already had it.
    #[tokio::test]
    async fn reconnecting_to_a_running_sandbox_reports_found() {
        let mut client = MockLambdaMicrovmsApi::new();
        client
            .expect_get_microvm()
            .returning(|id| Ok(owned(id, "RUNNING")));
        client.expect_run_microvm().never();

        let resolved = sandbox(client)
            .get_or_create(CreateSandboxRequest {
                sandbox_id: Some("already-up".to_string()),
                ..Default::default()
            })
            .await
            .expect("a running sandbox is handed back");

        assert_eq!(resolved.sandbox.sandbox_id, "already-up");
        assert!(
            !resolved.created,
            "the sandbox was already running, so this call did not create it"
        );
    }

    /// A paused sandbox is still the caller's: only a terminated one is replaced, so this is
    /// handed back as found, not replaced and not resumed.
    #[tokio::test]
    async fn reconnecting_to_a_paused_sandbox_reports_found() {
        let mut client = MockLambdaMicrovmsApi::new();
        client
            .expect_get_microvm()
            .returning(|id| Ok(owned(id, "SUSPENDED")));
        client.expect_run_microvm().never();
        client.expect_resume_microvm().never();

        let resolved = sandbox(client)
            .get_or_create(CreateSandboxRequest {
                sandbox_id: Some("asleep".to_string()),
                ..Default::default()
            })
            .await
            .expect("a paused sandbox is handed back");

        assert_eq!(resolved.sandbox.sandbox_id, "asleep");
        assert_eq!(resolved.sandbox.state, SandboxState::Paused);
        assert!(!resolved.created);
    }

    /// The reconnect read is `get_or_create`'s, not `get`'s. Naming the helper it happens to
    /// share would point a reader at a call the caller never made.
    #[tokio::test]
    async fn a_failed_reconnect_reports_the_verb_the_caller_used() {
        let mut client = MockLambdaMicrovmsApi::new();
        client.expect_get_microvm().returning(|_| {
            Err(alien_error::AlienError::new(
                alien_client_core::ErrorData::RemoteServiceUnavailable {
                    message: "GetMicrovm was throttled".to_string(),
                },
            ))
        });
        client.expect_run_microvm().never();

        let error = sandbox(client)
            .get_or_create(CreateSandboxRequest {
                sandbox_id: Some("unreadable".to_string()),
                ..Default::default()
            })
            .await
            .expect_err("a read that fails is not a sandbox that is absent");

        assert!(
            error.to_string().contains("sandbox.getOrCreate"),
            "the reconnect failure names getOrCreate, not get: {error}"
        );
    }

    /// The branch that waits: a MicroVM still coming up is waited for, and the wait failing is
    /// answered with the failure rather than with a second MicroVM nobody holds an id for.
    ///
    /// Pins only the never-create half. Carrying the wait through to success needs a TLS listener,
    /// because `authorized_request` builds an `https://` URL — the readiness poll itself is covered
    /// directly by `the_readiness_poll_outlasts_the_snapshot_restore_window`.
    #[tokio::test]
    async fn a_booting_sandbox_is_never_answered_with_a_second_microvm() {
        let mut client = MockLambdaMicrovmsApi::new();
        // Coming up, and with no endpoint published yet, so the wait runs and then gives up.
        client
            .expect_get_microvm()
            .returning(|id| Ok(owned(id, "PENDING")));
        client.expect_run_microvm().never();

        let error = sandbox(client)
            .get_or_create(CreateSandboxRequest {
                sandbox_id: Some("still-booting".to_string()),
                ..Default::default()
            })
            .await
            .expect_err("a sandbox that never publishes an endpoint cannot be served");

        assert_eq!(error.code, "SANDBOX_UNREACHABLE");
        assert!(
            error.to_string().contains("still-booting"),
            "the failure names the sandbox it waited on: {error}"
        );
    }

    /// `ResumeMicrovm` is accepted while the MicroVM is still SUSPENDED and stays so until its
    /// `/resume` hook returns, so the acceptance is not a sandbox that can take a command. Covers
    /// the failure path only: the success path needs TLS, as
    /// `a_booting_sandbox_is_never_answered_with_a_second_microvm` explains. The caller holds this
    /// sandbox's id, so a wait that gives up must not terminate it.
    #[tokio::test]
    async fn resume_does_not_return_before_the_sandbox_can_serve() {
        let mut client = MockLambdaMicrovmsApi::new();
        client
            .expect_get_microvm()
            .returning(|id| Ok(owned(id, "SUSPENDED")));
        client
            .expect_resume_microvm()
            .times(1)
            .returning(|_| Ok(()));
        client.expect_terminate_microvm().never();

        let error = sandbox(client)
            .resume("asleep")
            .await
            .expect_err("a resumed sandbox that never publishes an endpoint cannot be served");

        assert_eq!(error.code, "SANDBOX_UNREACHABLE");
        assert!(
            error.to_string().contains("sandbox.resume") && error.to_string().contains("asleep"),
            "the failure names the verb the caller used and the sandbox it waited on: {error}"
        );
    }

    /// See `capabilities()` for why this tracks `preview_ports`.
    #[tokio::test]
    async fn capabilities_reflect_the_declared_preview_ports() {
        assert!(
            !sandbox(MockLambdaMicrovmsApi::new()).capabilities().preview,
            "no declared ports means no preview capability"
        );
        assert!(
            sandbox_previewing(MockLambdaMicrovmsApi::new(), vec![8080])
                .capabilities()
                .preview,
            "a declared port makes the capability real"
        );
    }

    /// See `create()` for why these are refused rather than silently dropped.
    #[tokio::test]
    async fn unsupported_sandbox_fields_are_refused_not_dropped() {
        let mut client = MockLambdaMicrovmsApi::new();
        client.expect_run_microvm().never();

        let sandbox = sandbox(client);

        let with_env = CreateSandboxRequest {
            env: [("KEY".to_string(), "value".to_string())].into(),
            ..Default::default()
        };
        let error = sandbox
            .create(with_env)
            .await
            .expect_err("a sandbox-level env must be refused");
        assert_eq!(error.code, "OPERATION_NOT_SUPPORTED");

        let with_tenant = CreateSandboxRequest {
            tenant_key: Some("tenant-1".to_string()),
            ..Default::default()
        };
        let error = sandbox
            .create(with_tenant)
            .await
            .expect_err("a tenant key must be refused");
        assert_eq!(error.code, "OPERATION_NOT_SUPPORTED");
    }

    /// IAM cannot draw this line: the stack binding scopes the token mint to
    /// `microvm-image:<stack prefix>-*`, which matches every sibling sandbox in the stack, so a
    /// workload passing a sibling's sandbox id would be authorised for it. The sandbox's own
    /// `imageArn` is what says whose it is.
    #[tokio::test]
    async fn a_sandbox_from_another_declaration_is_refused_before_anything_is_minted() {
        let mut client = MockLambdaMicrovmsApi::new();
        client.expect_get_microvm().returning(|id| {
            Ok(Microvm {
                started_at: None,
                maximum_duration_in_seconds: None,
                microvm_id: Some(id.to_string()),
                endpoint: Some("vm.example.invalid".to_string()),
                state: Some("RUNNING".to_string()),
                // A live sandbox, reachable, running — and belonging to a different declaration.
                image_arn: Some("someone-elses-image".to_string()),
                image_version: Some("1".to_string()),
            })
        });
        client.expect_create_microvm_auth_token().never();
        client.expect_terminate_microvm().never();
        client.expect_suspend_microvm().never();
        client.expect_resume_microvm().never();

        let sandbox = sandbox_previewing(client, vec![8080]);

        for outcome in [
            sandbox.preview("a-siblings-sandbox", 8080).await.err(),
            sandbox.terminate("a-siblings-sandbox").await.err(),
            sandbox.pause("a-siblings-sandbox").await.err(),
            sandbox.resume("a-siblings-sandbox").await.err(),
        ] {
            let error = outcome.expect("a sandbox this sandbox does not own is refused");
            assert!(
                error
                    .to_string()
                    .contains("does not belong to this sandbox"),
                "names the reason: {error}"
            );
        }

        assert!(
            sandbox
                .get("a-siblings-sandbox")
                .await
                .expect("reading it is not an error")
                .is_none(),
            "a sibling's sandbox reads as absent rather than as one of ours"
        );
    }

    /// The absent-sandbox path, built the way the client builds it rather than by hand. `get`
    /// must report a sandbox that does not exist as `None`, because `get_or_create` reads that
    /// answer to decide whether to create one — an error there means a caller supplying a fresh
    /// id can never create a sandbox at all.
    #[tokio::test]
    async fn a_sandbox_that_does_not_exist_reads_as_absent_rather_than_as_a_failure() {
        let mut client = MockLambdaMicrovmsApi::new();
        client.expect_get_microvm().returning(|_| {
            Err(alien_error::AlienError::new(
                alien_client_core::ErrorData::RemoteResourceNotFound {
                    resource_type: "Microvm".to_string(),
                    resource_name: "GetMicrovm".to_string(),
                },
            ))
        });

        assert!(sandbox(client)
            .get("never-existed")
            .await
            .expect("an absent sandbox is an answer, not an error")
            .is_none());
    }

    /// A read that genuinely failed is not an absent sandbox. Flattening it into `None` would
    /// have `get_or_create` start a second sandbox while the first is still running.
    #[tokio::test]
    async fn a_failed_read_is_not_reported_as_an_absent_sandbox() {
        let mut client = MockLambdaMicrovmsApi::new();
        client.expect_get_microvm().returning(|_| {
            Err(alien_error::AlienError::new(
                alien_client_core::ErrorData::RateLimitExceeded {
                    message: "throttled".to_string(),
                },
            ))
        });

        sandbox(client)
            .get("ours")
            .await
            .expect_err("a throttle is not an absent sandbox");
    }

    /// A response the client could not parse an image out of must not pass as ours. Defaulting
    /// the other way would make every unparsed sandbox belong to whoever asked.
    #[tokio::test]
    async fn a_sandbox_with_no_image_is_not_assumed_to_be_ours() {
        let mut client = MockLambdaMicrovmsApi::new();
        client.expect_get_microvm().returning(|id| {
            Ok(Microvm {
                started_at: None,
                maximum_duration_in_seconds: None,
                microvm_id: Some(id.to_string()),
                endpoint: Some("vm.example.invalid".to_string()),
                state: Some("RUNNING".to_string()),
                image_arn: None,
                image_version: None,
            })
        });
        client.expect_create_microvm_auth_token().never();

        let error = sandbox_previewing(client, vec![8080])
            .preview("unlabelled", 8080)
            .await
            .expect_err("an unattributable sandbox is refused");
        assert!(error
            .to_string()
            .contains("does not belong to this sandbox"));
    }

    /// The check costs one `GetMicrovm`, which `sandbox/execute` grants scoped to this image.
    /// Enumerating instead would need `ListMicrovms`, which that set does not carry — an app
    /// linked to a sandbox would fail on its first command.
    #[tokio::test]
    async fn reaching_a_sandbox_does_not_enumerate_the_image() {
        let mut client = MockLambdaMicrovmsApi::new();
        client.expect_list_microvms().never();
        client.expect_list_microvm_image_versions().never();
        client
            .expect_get_microvm()
            .returning(|id| Ok(owned(id, "RUNNING")));

        let instance = sandbox(client)
            .get("ours")
            .await
            .expect("reading our own sandbox succeeds")
            .expect("it is present");
        assert_eq!(instance.sandbox_id, "ours");
    }

    /// A bare URL would be unusable: the endpoint refuses anything without the token headers and
    /// the port header, so a caller handed only a string would have to rebuild the auth.
    /// AWS answers 502 at the proxy while a MicroVM's snapshot restores, so the wait exists to
    /// outlast that window rather than to hand the first command a sandbox that cannot serve.
    /// Served over plain HTTP against a local listener: what is under test is the polling, not
    /// the transport that reaches a real MicroVM.
    #[tokio::test]
    async fn the_readiness_poll_outlasts_the_snapshot_restore_window() {
        use std::sync::atomic::{AtomicUsize, Ordering};

        let attempts = std::sync::Arc::new(AtomicUsize::new(0));
        let seen = attempts.clone();
        let handler = move || {
            let seen = seen.clone();
            async move {
                match seen.fetch_add(1, Ordering::SeqCst) {
                    0 => (axum::http::StatusCode::BAD_GATEWAY, ""),
                    1 => (
                        axum::http::StatusCode::SERVICE_UNAVAILABLE,
                        "snapshot restoring",
                    ),
                    _ => (axum::http::StatusCode::OK, ""),
                }
            }
        };
        let router = axum::Router::new().route(HEALTH_PATH, axum::routing::get(handler));
        let listener = tokio::net::TcpListener::bind::<std::net::SocketAddr>(
            "127.0.0.1:0".parse().expect("a loopback address"),
        )
        .await
        .expect("bind");
        let address = listener.local_addr().expect("address");
        tokio::spawn(async move { axum::serve(listener, router).await.expect("serve") });

        let probe = reqwest::Client::new().get(format!("http://{address}{HEALTH_PATH}"));
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);

        AwsSandbox::poll_until_healthy(probe, deadline, "sandbox.create", "mvm-restoring")
            .await
            .expect("the wait must outlast generic restore 502 and 503 responses");
        assert_eq!(
            attempts.load(Ordering::SeqCst),
            3,
            "it must keep probing through generic restore responses"
        );
    }

    /// The counterpart: a MicroVM that never answers has to fail, and say which sandbox.
    #[tokio::test]
    async fn the_readiness_poll_gives_up_on_a_sandbox_that_never_answers() {
        let router = axum::Router::new().route(
            HEALTH_PATH,
            axum::routing::get(|| async { axum::http::StatusCode::BAD_GATEWAY }),
        );
        let listener = tokio::net::TcpListener::bind::<std::net::SocketAddr>(
            "127.0.0.1:0".parse().expect("a loopback address"),
        )
        .await
        .expect("bind");
        let address = listener.local_addr().expect("address");
        tokio::spawn(async move { axum::serve(listener, router).await.expect("serve") });

        let probe = reqwest::Client::new().get(format!("http://{address}{HEALTH_PATH}"));
        let deadline = std::time::Instant::now() + std::time::Duration::from_millis(30);

        let error = AwsSandbox::poll_until_healthy(probe, deadline, "sandbox.create", "mvm-dead")
            .await
            .expect_err("a sandbox that never answers must not be reported as ready");
        assert_eq!(error.code, "SANDBOX_UNREACHABLE");
        assert!(
            error.to_string().contains("mvm-dead") && error.to_string().contains("502"),
            "the failure has to name the sandbox and what it last saw: {error}"
        );
    }

    /// A connection the proxy accepts and never answers must not hold the wait past its deadline.
    #[tokio::test]
    async fn the_readiness_poll_is_bounded_by_its_deadline_on_a_silent_connection() {
        let listener = tokio::net::TcpListener::bind::<std::net::SocketAddr>(
            "127.0.0.1:0".parse().expect("a loopback address"),
        )
        .await
        .expect("bind");
        let address = listener.local_addr().expect("address");
        tokio::spawn(async move {
            let mut held = Vec::new();
            while let Ok((socket, _)) = listener.accept().await {
                held.push(socket);
            }
        });

        let probe = reqwest::Client::new().get(format!("http://{address}{HEALTH_PATH}"));
        let deadline = std::time::Instant::now() + std::time::Duration::from_millis(50);

        let error = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            AwsSandbox::poll_until_healthy(probe, deadline, "sandbox.resume", "mvm-silent"),
        )
        .await
        .expect("the wait must end at its own deadline, not hang on the silent connection")
        .expect_err("a sandbox that never answers is not ready");
        assert_eq!(error.code, "SANDBOX_UNREACHABLE");
    }

    #[tokio::test]
    async fn preview_returns_the_headers_a_caller_cannot_construct() {
        let mut client = MockLambdaMicrovmsApi::new();
        client
            .expect_list_microvm_image_versions()
            .returning(|_| Ok(vec![image_version("3")]));
        client.expect_list_microvms().returning(|_, _| {
            Ok(vec![Microvm {
                started_at: None,
                maximum_duration_in_seconds: None,
                microvm_id: Some("mvm-1".into()),
                endpoint: None,
                state: Some("RUNNING".into()),
                image_arn: Some("sbx-image".to_string()),
                image_version: Some("1".to_string()),
            }])
        });
        client.expect_get_microvm().returning(|_| {
            Ok(Microvm {
                started_at: None,
                maximum_duration_in_seconds: None,
                microvm_id: Some("mvm-1".to_string()),
                endpoint: Some("mvm-1.lambda-microvms.aws".to_string()),
                state: Some("RUNNING".to_string()),
                image_arn: Some("sbx-image".to_string()),
                image_version: Some("1".to_string()),
            })
        });
        client
            .expect_create_microvm_auth_token()
            .withf(|_, ports, minutes| {
                ports.as_slice() == [8080] && *minutes == PREVIEW_TOKEN_MINUTES
            })
            .returning(|_, _, _| {
                Ok(MicrovmAuthToken {
                    auth_token: std::collections::HashMap::from([(
                        "X-aws-proxy-auth".to_string(),
                        "jwe-value".to_string(),
                    )]),
                })
            });

        let capability = sandbox_previewing(client, vec![8080])
            .preview("mvm-1", 8080)
            .await
            .expect("mints");

        assert_eq!(capability.endpoint, "https://mvm-1.lambda-microvms.aws");
        assert_eq!(
            capability
                .headers
                .get("X-aws-proxy-auth")
                .map(String::as_str),
            Some("jwe-value")
        );
        assert_eq!(
            capability
                .headers
                .get(PROXY_PORT_HEADER)
                .map(String::as_str),
            Some("8080")
        );
        assert_eq!(capability.allowed_ports, vec![8080]);
        assert_eq!(capability.expires_in_seconds, 1800);
    }

    /// The declared list is what bounds ingress for callers on this path: `CreateMicrovmAuthToken`
    /// mints a token for whatever port it is handed and has no port condition key, so an unlisted
    /// port must be refused before the call rather than after it.
    #[tokio::test]
    async fn a_port_the_stack_did_not_declare_is_refused_before_a_token_exists() {
        let mut client = MockLambdaMicrovmsApi::new();
        client.expect_get_microvm().never();
        client.expect_create_microvm_auth_token().never();

        let error = sandbox_previewing(client, vec![8080])
            .preview("mvm-1", 22)
            .await
            .expect_err("port 22 was never declared");

        assert!(
            error.to_string().contains("22"),
            "the refusal must name the port asked for: {error}"
        );
    }

    /// The figure handed to a caller must not outrun the token behind it: AWS caps the mint at
    /// 60 minutes, so an unclamped 30-minute promise would still be honest, but a raised
    /// `PREVIEW_TOKEN_MINUTES` past the cap would not.
    #[test]
    fn a_reported_preview_lifetime_never_exceeds_what_aws_will_mint() {
        assert_eq!(preview_lifetime_seconds(), 1800);
        assert!(
            preview_lifetime_seconds() <= u64::from(MAX_AUTH_TOKEN_MINUTES) * 60,
            "the reported lifetime must not outrun the cap the client sends"
        );
    }

    /// The lifecycle states AWS reports, mapped onto the binding's. Read through `get`, which is
    /// the only way a sandbox is reached now that enumeration is gone.
    #[tokio::test]
    async fn a_microvm_that_is_not_running_yet_is_reported_as_starting() {
        for (aws_state, expected) in [
            ("PENDING", SandboxState::Starting),
            ("RUNNING", SandboxState::Running),
            ("SUSPENDED", SandboxState::Paused),
            ("TERMINATED", SandboxState::Terminated),
        ] {
            let mut client = MockLambdaMicrovmsApi::new();
            client
                .expect_get_microvm()
                .returning(move |id| Ok(owned(id, aws_state)));

            let instance = sandbox(client)
                .get("s1")
                .await
                .expect("reads")
                .expect("present");
            assert_eq!(instance.state, expected, "AWS state {aws_state}");
        }
    }

    /// The declared ceiling has to survive the last hop as well as the first: the binding carries
    /// it onto `AwsSandbox`, and only this call puts it on the wire. A field dropped here would
    /// leave a sandbox running past a limit its stack declared, with every other test still green.
    #[tokio::test]
    async fn the_declared_lifetime_reaches_the_run_call() {
        let mut client = MockLambdaMicrovmsApi::new();
        client
            .expect_run_microvm()
            .withf(|_, _, _, _, _, _, max_lifetime, _| *max_lifetime == Some(1800))
            .returning(|_, _, _, _, _, _, _, _| Ok(owned("mvm-1", "PENDING")));
        // The wait reads the sandbox back; with no endpoint published it stays unreachable,
        // which is all a unit test can offer. Create then terminates what it started.
        client
            .expect_get_microvm()
            .returning(|id| Ok(owned(id, "RUNNING")));
        client
            .expect_terminate_microvm()
            .times(1)
            .returning(|_| Ok(()));

        let result = AwsSandbox::new(
            std::sync::Arc::new(client),
            "sbx-image",
            "3",
            vec!["connector".to_string()],
            Vec::new(),
            None,
            Some(1800),
        )
        .create(CreateSandboxRequest {
            sandbox_id: None,
            tenant_key: None,
            env: BTreeMap::new(),
            ..Default::default()
        })
        .await;

        // `withf` above is the assertion: a run carrying the wrong ceiling matches no
        // expectation and panics. Create then waits for an agent no unit test can serve.
        let error = result.expect_err("no agent answers in a unit test");
        assert_eq!(error.code, "SANDBOX_UNREACHABLE");
        assert!(
            error.to_string().contains("published no endpoint"),
            "the failure has to name the readiness wait, not any error: {error}"
        );
    }

    /// AWS cannot move a running MicroVM's deadline, so a lifetime the caller asked for is only
    /// ever applied here. Pins the unit as well as the value: `timeoutMs` is milliseconds and
    /// `maximumDurationInSeconds` is seconds, and a missed conversion is 1000x either way.
    #[tokio::test]
    async fn a_requested_lifetime_reaches_the_run_call_in_seconds() {
        let mut client = MockLambdaMicrovmsApi::new();
        client
            .expect_run_microvm()
            .withf(|_, _, _, _, _, _, max_lifetime, _| *max_lifetime == Some(90))
            .returning(|_, _, _, _, _, _, _, _| Ok(owned("mvm-1", "PENDING")));
        client
            .expect_get_microvm()
            .returning(|id| Ok(owned(id, "RUNNING")));
        client
            .expect_terminate_microvm()
            .times(1)
            .returning(|_| Ok(()));

        let error = sandbox(client)
            .create(CreateSandboxRequest {
                timeout_ms: Some(90_000),
                ..Default::default()
            })
            .await
            .expect_err("no agent answers in a unit test");

        // `withf` above is the assertion: a run carrying anything but 90 seconds matches no
        // expectation and panics. Create then waits for an agent a unit test cannot serve.
        assert_eq!(error.code, "SANDBOX_UNREACHABLE");
    }

    /// The declared ceiling is the deployment's, not the caller's. A request that could raise it
    /// would let an application outlive the limit its own stack declared, which is the one
    /// direction this field must never move.
    #[tokio::test]
    async fn a_requested_lifetime_cannot_raise_the_declared_ceiling() {
        let mut client = MockLambdaMicrovmsApi::new();
        client
            .expect_run_microvm()
            .withf(|_, _, _, _, _, _, max_lifetime, _| *max_lifetime == Some(1800))
            .returning(|_, _, _, _, _, _, _, _| Ok(owned("mvm-1", "PENDING")));
        client
            .expect_get_microvm()
            .returning(|id| Ok(owned(id, "RUNNING")));
        client
            .expect_terminate_microvm()
            .times(1)
            .returning(|_| Ok(()));

        let error = AwsSandbox::new(
            std::sync::Arc::new(client),
            "sbx-image",
            "3",
            vec!["connector".to_string()],
            Vec::new(),
            None,
            Some(1800),
        )
        .create(CreateSandboxRequest {
            timeout_ms: Some(7_200_000),
            ..Default::default()
        })
        .await
        .expect_err("no agent answers in a unit test");

        assert_eq!(error.code, "SANDBOX_UNREACHABLE");
    }

    /// A MicroVM that never became servable and could not be terminated is still running, and
    /// only this error carries its id. The wait's own failure is retryable, so returning it would
    /// invite a retry that mints a second MicroVM beside the first.
    #[tokio::test]
    async fn a_microvm_that_could_not_be_terminated_is_reported_by_id() {
        let mut client = MockLambdaMicrovmsApi::new();
        client
            .expect_run_microvm()
            .returning(|_, _, _, _, _, _, _, _| Ok(owned("mvm-orphan", "PENDING")));
        // Published no endpoint, so the readiness wait gives up, and the cleanup it then tries
        // is refused — one missing grant refuses both verbs in practice.
        client
            .expect_get_microvm()
            .returning(|id| Ok(owned(id, "RUNNING")));
        client.expect_terminate_microvm().times(1).returning(|_| {
            Err(alien_error::AlienError::new(
                alien_client_core::ErrorData::RemoteServiceUnavailable {
                    message: "TerminateMicrovm was refused".to_string(),
                },
            ))
        });

        let error = sandbox(client)
            .create(CreateSandboxRequest::default())
            .await
            .expect_err("a MicroVM that never became servable is not a sandbox");

        assert_eq!(error.code, "SANDBOX_COMMAND_FAILED", "{error}");
        assert!(
            !error.retryable,
            "a retry would mint another MicroVM nobody can reach: {error}"
        );
        assert!(
            error.message.contains("sandboxLeftBehind"),
            "the leak is reported under the label the siblings use: {error}"
        );
        // The outermost message, not the rendered chain: the wait's own error names the MicroVM
        // too, so the chain would read as green with the leak unnamed.
        assert!(
            error.message.contains("mvm-orphan"),
            "only this error can send an operator to the MicroVM left running: {error}"
        );
    }

    /// Observed live: AWS returns the MicroVM a client token previously created **even after it
    /// is terminated**. Using the caller's sandbox id as that token hands back a dead MicroVM
    /// and then waits for it to start, which is a hang, not an error.
    #[tokio::test]
    async fn a_caller_supplied_sandbox_id_is_never_the_client_token() {
        let mut client = MockLambdaMicrovmsApi::new();
        client
            .expect_run_microvm()
            .withf(|image, version, token, _, _, _, _, _| {
                image == "sbx-image" && version == "3" && token != "caller-chosen"
            })
            .returning(|_, _, _, _, _, _, _, _| {
                Ok(Microvm {
                    started_at: None,
                    maximum_duration_in_seconds: None,
                    microvm_id: Some("mvm-9".to_string()),
                    endpoint: None,
                    state: Some("PENDING".to_string()),
                    image_arn: Some("sbx-image".to_string()),
                    image_version: Some("1".to_string()),
                })
            });
        // The wait reads the sandbox back; with no endpoint published it stays unreachable,
        // which is all a unit test can offer. Create then terminates what it started.
        client
            .expect_get_microvm()
            // AWS assigns the id and `create` has to carry that one forward, not the caller's.
            .withf(|id| id == "mvm-9")
            .returning(|id| Ok(owned(id, "RUNNING")));
        client
            .expect_terminate_microvm()
            .withf(|id| id == "mvm-9")
            .times(1)
            .returning(|_| Ok(()));

        let result = sandbox(client)
            .create(CreateSandboxRequest {
                sandbox_id: Some("caller-chosen".to_string()),
                tenant_key: None,
                env: BTreeMap::new(),
                ..Default::default()
            })
            .await;

        // The client token assertion is `withf` above; a run reusing the caller's id matches no
        // expectation and panics. Create then waits for an agent a unit test cannot serve.
        let error = result.expect_err("no agent answers in a unit test");
        assert_eq!(error.code, "SANDBOX_UNREACHABLE");
        assert!(
            error.to_string().contains("published no endpoint"),
            "the failure has to name the readiness wait, not any error: {error}"
        );
    }

    #[tokio::test]
    async fn a_command_without_a_timeout_is_refused_before_any_aws_call() {
        // No expectations set: a call to AWS here would fail the mock, which is the assertion.
        let outcome = sandbox(MockLambdaMicrovmsApi::new())
            .run_command(
                "mvm-1",
                RunCommandRequest {
                    command: "/bin/echo".to_string(),
                    args: Vec::new(),
                    cwd: None,
                    env: BTreeMap::new(),
                    timeout: Duration::ZERO,
                },
            )
            .await;

        match outcome {
            Ok(_) => panic!("a zero timeout must be refused"),
            Err(error) => assert!(error.to_string().contains("non-zero timeout"), "{error}"),
        }
    }

    /// A rolled image version does not end the sandboxes running on the previous one, and does
    /// not change whose they are. Comparing the version as well as the image would make `get`
    /// return None for a live sandbox after a roll, which a caller reads as expired — the exact
    /// false negative GCP's capability set refuses to ship.
    #[tokio::test]
    async fn a_sandbox_on_a_previous_image_version_is_still_ours() {
        let mut client = MockLambdaMicrovmsApi::new();
        client.expect_get_microvm().returning(|id| {
            Ok(Microvm {
                started_at: None,
                maximum_duration_in_seconds: None,
                microvm_id: Some(id.to_string()),
                endpoint: None,
                state: Some("RUNNING".to_string()),
                image_arn: Some("sbx-image".to_string()),
                // The binding is pinned to version 3; this sandbox predates the roll.
                image_version: Some("2".to_string()),
            })
        });

        let found = sandbox(client)
            .get("older")
            .await
            .expect("reads")
            .expect("a sandbox on the previous version is still live and still ours");

        assert_eq!(found.sandbox_id, "older");
    }

    /// Enumeration would cost an account-wide `ListMicrovms`, and the case it would serve is a
    /// `RunMicrovm` whose response never arrived, leaving a MicroVM nobody holds the id for.
    #[tokio::test]
    async fn sandboxes_are_not_enumerable_and_nothing_asks_aws_to_be() {
        let mut client = MockLambdaMicrovmsApi::new();
        client.expect_list_microvms().never();
        client.expect_list_microvm_image_versions().never();

        let error = sandbox(client)
            .list()
            .await
            .expect_err("listing is not offered on AWS");
        assert!(
            error.to_string().contains("get"),
            "points the caller at what does work: {error}"
        );
    }

    /// A `RunMicrovm` refused for a missing IAM action, shaped as `LambdaMicrovmsClient::send`
    /// shapes one: the transport records the response body, and `classify` wraps a non-404 as
    /// its own generic failure.
    fn refused_run() -> Result<Microvm, alien_client_core::ErrorData> {
        Err(AlienError::new(
            alien_client_core::ErrorData::HttpResponseError {
                message: "Request failed with HTTP 403: Forbidden".to_string(),
                url: "https://lambda.us-east-2.amazonaws.com/2025-09-09/microvms".to_string(),
                http_status: 403,
                http_request_text: None,
                http_response_text: Some(
                    r#"{"Message":"User: arn:aws:sts::123456789012:assumed-role/stack-access/session is not authorized to perform: lambda:PassNetworkConnector on resource: arn:aws:lambda:us-east-2:aws:network-connector:aws-network-connector:INTERNET_EGRESS"}"#
                        .to_string(),
                ),
            },
        ))
        .context(alien_client_core::ErrorData::GenericError {
            message: "Lambda MicroVMs RunMicrovm failed".to_string(),
        })
    }

    /// The refused action is what sends a reader to the role rather than to this code, and the
    /// wire format past this binding is a flat message string — so `reason` is the only place a
    /// structured consumer sees it. It reaches an operator's log alone: an IAM identity makes the
    /// whole error internal, and `into_external` replaces it.
    #[tokio::test]
    async fn a_refused_create_reports_what_aws_refused_it_with() {
        let mut client = MockLambdaMicrovmsApi::new();
        client
            .expect_run_microvm()
            .returning(|_, _, _, _, _, _, _, _| refused_run());
        // Nothing was started, so nothing is cleaned up.
        client.expect_terminate_microvm().never();

        let error = sandbox(client)
            .create(CreateSandboxRequest {
                sandbox_id: None,
                tenant_key: None,
                env: BTreeMap::new(),
                ..Default::default()
            })
            .await
            .expect_err("a refused RunMicrovm cannot produce a sandbox");

        assert_eq!(error.code, "SANDBOX_UNREACHABLE");
        assert!(
            error
                .message
                .contains("could not start a MicroVM from 'sbx-image'"),
            "the binding still says which call it was: {}",
            error.message
        );
        assert!(
            error
                .message
                .contains("is not authorized to perform: lambda:PassNetworkConnector"),
            "and AWS's own sentence is what tells the operator why: {}",
            error.message
        );
        assert!(
            error.internal,
            "an IAM identity in the message makes the error internal: {error}"
        );
        assert_eq!(
            error.into_external().message,
            "Internal server error",
            "so none of it is published to the caller"
        );
    }
}

#[cfg(all(test, feature = "local"))]
#[path = "aws_egress_tests.rs"]
mod egress_tests;