use aws_config::{Region, SdkConfig};
use aws_sdk_acm::Client as AcmClient;
use aws_sdk_cloudwatch::Client as CwClient;
use aws_sdk_cloudwatchlogs::Client as CwLogsClient;
use aws_sdk_costexplorer::Client as CostExplorerClient;
use aws_sdk_ec2::Client as Ec2Client;
use aws_sdk_elasticbeanstalk::Client;
use aws_sdk_iam::Client as IamClient;
use aws_sdk_organizations::Client as OrgClient;
use aws_sdk_s3::Client as S3Client;
use aws_sdk_secretsmanager::Client as SecretsClient;
use aws_sdk_sqs::Client as SqsClient;
use aws_sdk_ssm::Client as SsmClient;
use aws_sdk_sts::Client as StsClient;
use chrono::{DateTime, Utc};
use color_eyre::eyre::{eyre, Result, WrapErr};

#[derive(Clone, Debug)]
pub struct Event {
    pub at: Option<DateTime<Utc>>,
    pub env: String,
    pub application: String,
    pub message: String,
    pub severity: String,
    /// Application version label this event relates to, when EB
    /// tags it (deploy events carry it). `None` for events with no
    /// associated version. Drives `:rollback`'s previous-version
    /// detection.
    pub version_label: Option<String>,
}

#[derive(Clone, Debug)]
pub struct CwAlarm {
    pub name: String,
    pub state: String, // OK / ALARM / INSUFFICIENT_DATA
    pub state_reason: String,
    pub metric_name: String,
    pub namespace: String,
}

/// One per-instance result returned by [`AwsClient::run_shell_command`].
/// `status` is the SSM CommandInvocationStatus string verbatim
/// (`Success` / `Failed` / `Cancelled` / `TimedOut` / `Pending` /
/// `InProgress` / `Delayed`) so the renderer can colour by it. The
/// stdout / stderr buffers cap at SSM's per-invocation limit (~24 KiB
/// each by default) — anything larger is signalled via the
/// `*_url` fields on the API which we don't currently follow.
#[derive(Clone, Debug, PartialEq)]
pub struct SsmRunResult {
    pub instance_id: String,
    pub status: String,
    pub exit_code: i32,
    pub stdout: String,
    pub stderr: String,
}

/// One row in a CloudWatch alarm's recent history, surfaced by
/// `:alarm-history`. `kind` is the API's HistoryItemType string —
/// `StateUpdate` (the transitions an operator usually wants),
/// `ConfigurationUpdate` (someone edited the threshold), or `Action`
/// (the alarm fired its SNS / autoscaling action). `summary` is the
/// short human-readable line CloudWatch emits per item.
#[derive(Clone, Debug, PartialEq)]
pub struct AlarmHistoryEntry {
    pub at: Option<DateTime<Utc>>,
    pub kind: String,
    pub summary: String,
}

#[derive(Clone, Debug, Default)]
pub struct MetricSeries {
    pub id: String,    // stable, e.g. "health"
    pub label: String, // CloudWatch label
    pub points: Vec<(DateTime<Utc>, f64)>,
}

#[derive(Clone, Debug, Default)]
pub struct WorkerQueues {
    pub main_url: Option<String>,
    pub dlq_url: Option<String>,
    pub main_stats: Option<QueueStats>,
    pub dlq_stats: Option<QueueStats>,
}

#[derive(Clone, Debug, Default)]
pub struct QueueStats {
    pub visible: i64,
    pub in_flight: i64,
    pub delayed: i64,
}

#[derive(Clone, Debug)]
pub struct QueueMessage {
    pub id: String,
    pub receipt_handle: String,
    pub body: String,
    pub receive_count: i64,
    pub sent_at: Option<DateTime<Utc>>,
}

/// One ISSUED ACM certificate in the active region. Drives the
/// `:listener-edit` SSL-cert picker.
#[derive(Clone, Debug)]
pub struct AcmCert {
    pub arn: String,
    pub domain: String,
}

#[derive(Clone, Debug)]
pub struct Instance {
    pub id: String,
    pub health: String, // Ok / Warning / Degraded / Severe / Info / NoData / Unknown / Pending
    pub color: String,  // Green / Yellow / Red / Grey
    pub causes: Vec<String>,
    pub instance_type: String,
    pub availability_zone: String,
    pub launched_at: Option<DateTime<Utc>>,
}

#[derive(Clone, Debug)]
pub struct Application {
    pub name: String,
    pub description: String,
    /// Surfaced in the `:apps-info` overlay (in operator timezone)
    /// alongside `date_updated`. Was orphaned briefly when the apps
    /// table dropped its CREATED column in 0.3.3.
    pub date_created: Option<DateTime<Utc>>,
    pub date_updated: Option<DateTime<Utc>>,
    pub version_count: usize,
    pub templates: Vec<String>,
    /// Newest application version's label (from `DescribeApplicationVersions`,
    /// sorted by date_created desc). Populated by a follow-up fetch after
    /// `list_applications` — `None` while still loading or if the app has
    /// no versions yet. The EB-console "latest deployed version" matches
    /// this field, not the application-level `date_updated`.
    pub latest_version_label: Option<String>,
    /// `date_created` of the newest application version.
    pub latest_version_created: Option<DateTime<Utc>>,
}

/// Result of `fetch_env_vpc_context` — the env's VPC plus the option-
/// settings selections the `:subnets` / `:elb-subnets` / `:security-groups`
/// pickers need for their pre-fill. Each field is `None` / empty when the
/// env doesn't override that option (EB uses its account-default in that
/// case).
#[derive(Clone, Debug, Default)]
pub struct EnvVpcContext {
    pub vpc_id: Option<String>,
    pub subnets: Vec<String>,
    /// ELB subnets (`aws:ec2:vpc.ELBSubnets`). Web-tier envs typically
    /// attach the ELB to a separate subnet set than the instance subnets;
    /// worker envs leave this empty.
    pub elb_subnets: Vec<String>,
    pub security_groups: Vec<String>,
}

/// One subnet in a VPC. Used by `:subnets` to populate the picker.
#[derive(Clone, Debug)]
pub struct SubnetInfo {
    pub id: String,
    pub availability_zone: String,
    pub cidr_block: String,
    /// Friendly name from the `Name` tag, if any.
    pub name_tag: Option<String>,
}

/// One security group in a VPC. Used by `:security-groups`.
#[derive(Clone, Debug)]
pub struct SecurityGroupInfo {
    pub id: String,
    pub group_name: String,
    pub description: String,
}

#[derive(Clone, Debug)]
pub struct CustomPlatform {
    pub arn: String,
    pub branch: String,
    pub version: String,
    pub status: String,
    pub lifecycle: String,
}

#[derive(Clone, Debug)]
pub struct AppVersion {
    pub label: String,
    pub description: String,
    pub created: Option<DateTime<Utc>>,
}

#[derive(Clone, Debug)]
pub struct Environment {
    pub name: String,
    pub application: String,
    pub status: String,
    pub health: String,
    pub platform: String, // family + version, e.g. "Java 17"
    /// Raw solution-stack name as reported by EB, e.g. `64bit Amazon Linux
    /// 2023 v6.1.0 running Node.js 18`. Empty for platform-ARN / custom-
    /// platform envs that don't report a solution stack. Drives the
    /// stale-platform comparison against `ListAvailableSolutionStacks`.
    pub solution_stack: String,
    pub tier: String, // "Web" / "Worker" / "?"
    pub cname: String,
    pub version_label: String,
    pub arn: Option<String>,
    pub updated: Option<DateTime<Utc>>,
    /// Internal EB environment ID (e.g. `e-abcdef1234`). Required by APIs
    /// that snapshot config from a live env (CreateConfigurationTemplate).
    pub id: Option<String>,
    /// Region the env was discovered in, when results were fanned out across
    /// multiple regions. `None` in single-region mode.
    pub region: Option<String>,
}

#[derive(Clone, Debug)]
pub struct AwsContext {
    pub region: String,
    pub profile: Option<String>,
    pub account_id: Option<String>,
    pub caller_arn: Option<String>,
}

/// One row passed to `fetch_custom_env_metrics`. The shape is wide enough
/// that clippy complains if used inline (`type_complexity` lint), so this
/// alias keeps call-sites tidy.
pub type CustomMetricQuery = (String, String, String, String, Vec<(String, String)>);

/// One event from a CloudWatch Logs stream — server-side timestamp + the
/// stream it came from + the raw message. `:logs-tail` builds these from
/// FilterLogEvents and renders them in chronological order.
#[derive(Clone, Debug)]
pub struct LogEvent {
    pub timestamp_ms: i64,
    pub stream: String,
    pub message: String,
}

/// One result row from a CloudWatch Logs Insights query. Each entry is
/// a (field-name, value) pair — Insights returns fields in query-order,
/// so the Vec preserves that order rather than a HashMap.
#[derive(Clone, Debug)]
pub struct InsightsRow {
    pub fields: Vec<(String, String)>,
}

/// The completed payload of an Insights query — result rows plus the
/// scan statistics (records_matched / records_scanned). Scanned is what
/// AWS bills against; surfacing it in the overlay footer makes the cost
/// of broad queries visible.
#[derive(Clone, Debug)]
pub struct InsightsResults {
    pub rows: Vec<InsightsRow>,
    pub records_scanned: i64,
    pub records_matched: i64,
}

#[derive(Clone, Debug)]
pub struct Identity {
    pub account_id: Option<String>,
    pub caller_arn: Option<String>,
}

/// Per-env summary of instance health, as surfaced in the `INST` column
/// of the main env table. `healthy` is the count EB classifies as Green
/// (Ok + Info — both are "passing health checks", Info just means an
/// operation is in progress on an otherwise-healthy instance); `total`
/// is the sum across every health bucket the env reports. `total == 0`
/// is a real signal (env has no instances right now, e.g. mid-launch),
/// rendered as `0/0` in the table.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct EnvInstanceCounts {
    pub healthy: i32,
    pub total: i32,
}

pub struct AwsClient {
    client: Client,
    sqs: SqsClient,
    cw: CwClient,
    cw_logs: CwLogsClient,
    s3: S3Client,
    ec2: Ec2Client,
    org: OrgClient,
    /// Cost Explorer client. Cost Explorer is a global service —
    /// always endpoints in `us-east-1` regardless of the operator's
    /// active region. Lazy-initialised on first `:cost on` so
    /// operators who never opt in don't carry the dep weight.
    cost: CostExplorerClient,
    /// IAM client used by `:explain` to call
    /// `iam:SimulatePrincipalPolicy`. IAM is a global service —
    /// pinned to `us-east-1` regardless of operator region (same
    /// as Cost Explorer and Organizations).
    iam: IamClient,
    /// Secrets Manager client. Region-scoped (unlike IAM / Cost
    /// Explorer / Organizations) — operators read secrets from
    /// the same region as the env they're configuring.
    secrets: SecretsClient,
    /// ACM client. Region-scoped — `:listener-edit` lists the region's
    /// certificates for the SSL-cert picker.
    acm: AcmClient,
    /// SSM client. Region-scoped — `:ssm-run` sends a shell command to
    /// the env's instances and aggregates the per-instance results.
    /// `pub(crate)` so mock-AWS tests can override the field
    /// post-construction (avoids threading SSM through every
    /// `for_tests` call site).
    pub(crate) ssm: SsmClient,
    config: SdkConfig,
    pub context: AwsContext,
}

/// One row in the `:accounts` overlay — an AWS Organizations child
/// account (or the management account itself). Sourced from
/// `organizations:ListAccounts`.
#[derive(Clone, Debug)]
pub struct OrgAccount {
    /// 12-digit account ID.
    pub id: String,
    /// Friendly name set when the account joined the org.
    pub name: String,
    /// Root user's email address (often the only way to spot ownership
    /// when account names are terse).
    pub email: Option<String>,
    /// `ACTIVE` / `SUSPENDED` / `PENDING_CLOSURE` — capitalised verbatim
    /// from the API.
    pub status: String,
}

/// Build a Cost Explorer client pinned to `us-east-1`. Cost Explorer
/// is a global service that only endpoints in `us-east-1` regardless
/// of which region the caller's `SdkConfig` carries; calling it from
/// any other region returns an empty result with no error, which is
/// exactly the silent failure the operator never debugs. Override
/// region here so the dep can't drift.
fn cost_explorer_client(base: &SdkConfig) -> CostExplorerClient {
    let cfg = base.to_builder().region(Region::new("us-east-1")).build();
    CostExplorerClient::new(&cfg)
}

/// Build an IAM client. IAM is a global service; the region the
/// caller's `SdkConfig` carries doesn't affect routing but pinning
/// here matches the Cost Explorer / Organizations pattern and
/// makes the `:explain` code path's expectations explicit.
fn iam_client(base: &SdkConfig) -> IamClient {
    let cfg = base.to_builder().region(Region::new("us-east-1")).build();
    IamClient::new(&cfg)
}

/// One row in the `:secrets` listing — metadata only, no values.
/// Value retrieval happens via `fetch_secret_value` on demand
/// because every `GetSecretValue` call is a separate audit-loggable
/// AWS event the operator should opt into explicitly.
#[derive(Clone, Debug)]
pub struct SecretSummary {
    pub name: String,
    pub arn: String,
    pub description: Option<String>,
    pub last_changed: Option<DateTime<Utc>>,
    pub last_rotated: Option<DateTime<Utc>>,
    pub kms_key_id: Option<String>,
}

/// One row of the `:explain` IAM diagnosis result. Carries the
/// per-action decision + the matched statements (so the operator
/// can audit which policy granted / denied / failed-to-grant) +
/// SCP / permission-boundary blockers when present.
#[derive(Clone, Debug)]
pub struct IamSimResult {
    pub action: String,
    pub resource: String,
    /// `"allowed"`, `"explicitDeny"`, or `"implicitDeny"`. Verbatim
    /// from the SDK so the renderer can map to severity colours
    /// without re-parsing.
    pub decision: String,
    /// Matched statements — typically `(policy_source_arn, sid_or_index)`.
    /// Empty for `implicitDeny` (no statement matched).
    pub matched_statements: Vec<String>,
    /// Conditions in the matched statements that weren't satisfied
    /// (e.g. `aws:RequestTag/Environment` missing). Empty when
    /// no conditions are pending.
    pub missing_context: Vec<String>,
    /// `true` when an SCP at the Organizations level denied the
    /// action regardless of the role's own policies — diagnoses
    /// the "role looks fine but call fails" case.
    pub blocked_by_scp: bool,
    /// `true` when a permission boundary on the role denied the
    /// action (same shape as SCP at the role level).
    pub blocked_by_boundary: bool,
}

/// Parsed `DescribeEnvironmentResources` payload. The SDK returns
/// flat lists; we hold them in field-typed buckets so the
/// `:resources` renderer can format them as a hierarchical tree
/// (ASG → instances → LB → queues etc.) without re-traversing
/// the raw API shape.
#[derive(Clone, Debug, Default)]
pub struct EnvResources {
    pub asgs: Vec<String>,
    pub instances: Vec<String>,
    pub launch_configs: Vec<String>,
    pub launch_templates: Vec<String>,
    pub load_balancers: Vec<String>,
    pub triggers: Vec<String>,
    pub queues: Vec<EnvResourceQueue>,
}

#[derive(Clone, Debug)]
pub struct EnvResourceQueue {
    pub name: String,
    pub url: String,
}

/// One settable EB configuration option, as returned by
/// [`AwsClient::fetch_env_configuration_options`]. Covers both the
/// operator's currently-set value and the platform's metadata
/// (default / constraints / change severity). `:options` uses this
/// to render the full config vocabulary in one overlay.
#[derive(Clone, Debug)]
pub struct ConfigOption {
    pub namespace: String,
    pub name: String,
    /// Current value, or `None` when the operator hasn't overridden
    /// the default. EB sometimes returns `Some("")` for unset; the
    /// renderer treats both as "default" and tags accordingly.
    pub value: Option<String>,
    pub default_value: Option<String>,
    /// `"Scalar"` / `"List"` / sometimes blank. Lower-cased on
    /// the wire; we render as-is.
    pub value_type: String,
    /// Constrained value options for enum-shaped settings
    /// (e.g. `["AllAtOnce", "Rolling", "Immutable", ...]` for
    /// `DeploymentPolicy`). Empty Vec when unconstrained.
    pub value_options: Vec<String>,
    /// `"NoInterruption"` / `"RestartEnvironment"` /
    /// `"RestartApplicationServer"` / `"Unknown"`. Warns the
    /// operator that changing this option will roll instances.
    pub change_severity: Option<String>,
    /// EB exposes a "this option is operator-settable" flag —
    /// most options have this true. Currently captured but not
    /// rendered (operator-set vs default distinction is enough
    /// signal); kept on the struct because a future "hide read-only
    /// options" filter would consume it.
    #[allow(dead_code)]
    pub user_defined: Option<bool>,
    pub min_value: Option<i32>,
    pub max_value: Option<i32>,
    pub max_length: Option<i32>,
}

/// One row of cost data returned by [`AwsClient::fetch_env_costs`] —
/// an EB env name and its monthly cost in USD across the trailing
/// window. `cost` is in whole + fractional dollars; the SDK returns
/// strings and we parse at the boundary.
#[derive(Clone, Debug)]
pub struct EnvCost {
    pub env_name: String,
    /// Monthly USD spend (summed across the trailing-30d window).
    pub cost_usd: f64,
}

impl AwsClient {
    /// Build the SDK client without making any network calls.
    pub async fn with(profile: Option<String>, region: Option<String>) -> Result<Self> {
        let mut builder = aws_config::defaults(aws_config::BehaviorVersion::latest());
        if let Some(p) = profile.clone() {
            builder = builder.profile_name(p);
        }
        if let Some(r) = region.clone() {
            builder = builder.region(Region::new(r));
        }
        let config = builder.load().await;

        let resolved_region = config
            .region()
            .map(|r| r.as_ref().to_string())
            .unwrap_or_else(|| "unknown".to_string());
        if region.as_deref().is_some_and(|r| r != resolved_region) {
            // SDK silently fell back to its chain. Most likely cause:
            // `.region(Region::new(r))` failed to bind because `r` is
            // empty or whitespace, leaving the env / profile chain to
            // pick. Make it loud so we can see it in `ebman.log`.
            tracing::warn!(
                target: "ebman::aws",
                requested = ?region,
                resolved = %resolved_region,
                env_aws_region = ?std::env::var("AWS_REGION").ok(),
                env_aws_default_region = ?std::env::var("AWS_DEFAULT_REGION").ok(),
                "AwsClient::with region mismatch — explicit override was ignored by SDK"
            );
        }
        let region = resolved_region;
        let profile = profile.or_else(|| std::env::var("AWS_PROFILE").ok());
        let client = Client::new(&config);
        let sqs = SqsClient::new(&config);
        let cw = CwClient::new(&config);
        let cw_logs = CwLogsClient::new(&config);
        let s3 = S3Client::new(&config);
        let ec2 = Ec2Client::new(&config);
        let org = OrgClient::new(&config);
        let cost = cost_explorer_client(&config);
        let iam = iam_client(&config);
        let secrets = SecretsClient::new(&config);

        Ok(Self {
            client,
            sqs,
            cw,
            cw_logs,
            s3,
            ec2,
            org,
            cost,
            iam,
            secrets,
            acm: AcmClient::new(&config),
            ssm: SsmClient::new(&config),
            config,
            context: AwsContext {
                region,
                profile,
                account_id: None,
                caller_arn: None,
            },
        })
    }

    /// Build a fully-mocked `AwsClient` for unit tests. The caller supplies
    /// pre-built (typically `mock_client!`-backed) sub-clients; any client
    /// not exercised by the test can stay as a plain SDK-default instance.
    /// Tests should not assume any of the sub-clients can talk to a real
    /// endpoint — the default ones will fail if a non-mocked code path is
    /// reached, which is exactly the signal we want.
    /// Build an `AwsClient` by `sts:AssumeRole`-ing into a target role
    /// using `source_profile`'s creds as the base identity. Pinned to
    /// `target_region` when supplied (falls back to the source profile's
    /// region / env default). Returned client carries the assumed
    /// session's caller_arn / account_id once `verify_identity` runs.
    ///
    /// Session lifetime defaults to AWS's 1h cap; the caller is
    /// expected to swap clients again before expiry. We don't implement
    /// background refresh here — the operator's refresh tick will
    /// re-invoke this when the session dies.
    pub async fn assume_role(target_name: &str, spec: &crate::config::AccountSpec) -> Result<Self> {
        // Stage 1: load the source-profile creds + region.
        let mut builder = aws_config::defaults(aws_config::BehaviorVersion::latest());
        if let Some(p) = spec.source_profile.as_ref() {
            builder = builder.profile_name(p.clone());
        }
        if let Some(r) = spec.region.clone() {
            builder = builder.region(Region::new(r));
        }
        let base_config = builder.load().await;

        // Stage 2: STS:AssumeRole against the configured role.
        let sts = StsClient::new(&base_config);
        let session_name = format!("ebman-{target_name}");
        let mut req = sts
            .assume_role()
            .role_arn(spec.role_arn.clone())
            .role_session_name(session_name);
        if let Some(eid) = spec.external_id.as_ref() {
            req = req.external_id(eid.clone());
        }
        let resp = req.send().await.wrap_err("sts:AssumeRole failed")?;
        let creds = resp
            .credentials
            .ok_or_else(|| eyre!("sts:AssumeRole returned no credentials"))?;
        let access_key = creds.access_key_id;
        let secret_key = creds.secret_access_key;
        let session_token = creds.session_token;
        let aws_creds = aws_credential_types::Credentials::new(
            access_key,
            secret_key,
            Some(session_token),
            // Expiry: aws_smithy_types::DateTime → SystemTime via secs.
            std::time::SystemTime::UNIX_EPOCH.checked_add(std::time::Duration::from_secs(
                creds.expiration.secs() as u64,
            )),
            "ebman-assume-role",
        );

        // Stage 3: build the final SdkConfig with the assumed creds.
        // We rebuild from scratch (rather than mutating base_config) so
        // the resulting config carries ONLY the assumed-role identity —
        // no leaked source-profile creds, no cross-region surprises.
        let mut builder = aws_config::defaults(aws_config::BehaviorVersion::latest());
        builder = builder.credentials_provider(aws_creds);
        if let Some(r) = spec.region.clone() {
            builder = builder.region(Region::new(r));
        } else if let Some(r) = base_config.region().cloned() {
            builder = builder.region(r);
        }
        let config = builder.load().await;
        let region = config
            .region()
            .map(|r| r.as_ref().to_string())
            .unwrap_or_else(|| "unknown".to_string());

        let cost = cost_explorer_client(&config);
        let iam = iam_client(&config);
        let secrets = SecretsClient::new(&config);
        Ok(Self {
            client: Client::new(&config),
            sqs: SqsClient::new(&config),
            cw: CwClient::new(&config),
            cw_logs: CwLogsClient::new(&config),
            s3: S3Client::new(&config),
            ec2: Ec2Client::new(&config),
            org: OrgClient::new(&config),
            cost,
            iam,
            secrets,
            acm: AcmClient::new(&config),
            ssm: SsmClient::new(&config),
            config,
            context: AwsContext {
                region,
                // Track the friendly account name as the "profile"
                // breadcrumb so the header reads `account=prod` rather
                // than the source profile name (which is just the
                // launchpad, not the destination).
                profile: Some(target_name.to_string()),
                account_id: None,
                caller_arn: None,
            },
        })
    }

    /// Build an `AwsClient` with default (un-mocked) sub-clients. For
    /// tests that exercise non-AWS code paths (keyboard flow, render,
    /// pure-helper composition) and don't care which AWS surface is
    /// reachable. Any AWS call against the returned client will fail
    /// loudly, which is the desired signal for "test accidentally hit
    /// the network". Pair with `App::for_tests` to drive `handle_event`
    /// without spinning up a real session.
    ///
    /// Also the underlying client used by `App::new_demo` — `--demo`
    /// mode wants the no-network behaviour without the cfg gate.
    /// `pub(crate)` because every caller (test code + `App::new_demo`)
    /// lives in this crate; no need to expose to the bin or downstream.
    pub(crate) fn stub() -> Self {
        let cfg = aws_config::SdkConfig::builder()
            .region(Region::new("us-east-1"))
            .behavior_version(aws_config::BehaviorVersion::latest())
            .build();
        Self::for_tests(
            Client::new(&cfg),
            SqsClient::new(&cfg),
            CwClient::new(&cfg),
            CwLogsClient::new(&cfg),
            S3Client::new(&cfg),
            Ec2Client::new(&cfg),
        )
    }

    pub(crate) fn for_tests(
        client: Client,
        sqs: SqsClient,
        cw: CwClient,
        cw_logs: CwLogsClient,
        s3: S3Client,
        ec2: Ec2Client,
    ) -> Self {
        // A bare config is fine here — every sub-client is owned by the
        // caller, so the only consumer of `self.config` is the lazy STS
        // client in `verify_identity`, which our tests don't call.
        let config = aws_config::SdkConfig::builder()
            .region(Region::new("us-east-1"))
            .behavior_version(aws_config::BehaviorVersion::latest())
            .build();
        // Org + Cost Explorer clients use default config because no
        // existing test exercises them; mocked variants can use a
        // dedicated helper if added.
        let org = OrgClient::new(&config);
        let cost = cost_explorer_client(&config);
        let iam = iam_client(&config);
        let secrets = SecretsClient::new(&config);
        Self {
            client,
            sqs,
            cw,
            cw_logs,
            s3,
            ec2,
            org,
            cost,
            iam,
            secrets,
            acm: AcmClient::new(&config),
            ssm: SsmClient::new(&config),
            config,
            context: AwsContext {
                region: "us-east-1".to_string(),
                profile: None,
                account_id: None,
                caller_arn: None,
            },
        }
    }

    /// Verify credentials work and fetch the caller identity. Used at startup to
    /// detect invalid persisted profiles, and as a background task after rebuild.
    pub async fn verify_identity(&self) -> Result<Identity> {
        let ident = StsClient::new(&self.config)
            .get_caller_identity()
            .send()
            .await
            .wrap_err("sts get-caller-identity failed")?;
        Ok(Identity {
            account_id: ident.account,
            caller_arn: ident.arn,
        })
    }

    pub async fn list_events(&self, max: i32) -> Result<Vec<Event>> {
        self.list_events_inner(None, max).await
    }

    pub async fn list_events_for_env(&self, env_name: &str, max: i32) -> Result<Vec<Event>> {
        self.list_events_inner(Some(env_name.to_string()), max)
            .await
    }

    async fn list_events_inner(&self, env_name: Option<String>, max: i32) -> Result<Vec<Event>> {
        let mut req = self.client.describe_events().max_records(max);
        if let Some(n) = env_name {
            req = req.environment_name(n);
        }
        let resp = req.send().await?;
        let events = resp
            .events
            .unwrap_or_default()
            .into_iter()
            .map(|e| Event {
                at: e
                    .event_date
                    .and_then(|d| DateTime::from_timestamp(d.secs(), d.subsec_nanos())),
                env: e.environment_name.unwrap_or_default(),
                application: e.application_name.unwrap_or_default(),
                message: e.message.unwrap_or_default(),
                severity: e
                    .severity
                    .map(|s| s.as_str().to_string())
                    .unwrap_or_else(|| "INFO".to_string()),
                version_label: e.version_label.filter(|v| !v.is_empty()),
            })
            .collect();
        Ok(events)
    }

    /// Full `DescribeEnvironmentResources` dump for an env, formatted as a
    /// human-readable string suitable for an overlay. Covers ASGs,
    /// instances, launch configurations, launch templates, load balancers,
    /// trigger names, and SQS queues — i.e. every infra resource EB
    /// manages for the env. Useful for "what's actually under this env?".
    /// Fetch the env's underlying AWS resources (ASGs, instances,
    /// launch config/template, load balancers, triggers, queues).
    /// Returns the parsed shape so the renderer can format as a
    /// hierarchical tree rather than a flat dump.
    pub async fn describe_env_resources(&self, env_name: &str) -> Result<EnvResources> {
        let resp = self
            .client
            .describe_environment_resources()
            .environment_name(env_name)
            .send()
            .await
            .wrap_err("DescribeEnvironmentResources failed")?;
        let res = resp
            .environment_resources
            .ok_or_else(|| eyre!("no environment_resources in response"))?;
        Ok(EnvResources {
            asgs: res
                .auto_scaling_groups
                .unwrap_or_default()
                .into_iter()
                .filter_map(|a| a.name)
                .collect(),
            instances: res
                .instances
                .unwrap_or_default()
                .into_iter()
                .filter_map(|i| i.id)
                .collect(),
            launch_configs: res
                .launch_configurations
                .unwrap_or_default()
                .into_iter()
                .filter_map(|l| l.name)
                .collect(),
            launch_templates: res
                .launch_templates
                .unwrap_or_default()
                .into_iter()
                .filter_map(|l| l.id)
                .collect(),
            load_balancers: res
                .load_balancers
                .unwrap_or_default()
                .into_iter()
                .filter_map(|l| l.name)
                .collect(),
            triggers: res
                .triggers
                .unwrap_or_default()
                .into_iter()
                .filter_map(|t| t.name)
                .collect(),
            queues: res
                .queues
                .unwrap_or_default()
                .into_iter()
                .filter_map(|q| {
                    let name = q.name?;
                    Some(EnvResourceQueue {
                        name,
                        url: q.url.unwrap_or_default(),
                    })
                })
                .collect(),
        })
    }

    /// Resolve the worker queue URL (and DLQ URL) for an env. EB autocreates
    /// queues when the user doesn't override `WorkerQueueURL`, and in that
    /// (common) case the option value comes back empty — so we ask
    /// `DescribeEnvironmentResources` first, which exposes the actual queue
    /// URLs under named entries (`WorkerQueue`, `WorkerDeadLetterQueue`).
    /// Falls back to the option-settings path for users who override the
    /// URL explicitly.
    pub async fn describe_worker_queues(
        &self,
        application_name: &str,
        env_name: &str,
    ) -> Result<WorkerQueues> {
        let mut main_url: Option<String> = None;
        let mut dlq_url: Option<String> = None;

        // Primary path: ask EB for the env's resources. Includes the URLs of
        // the queues EB created automatically when WorkerQueueURL is empty.
        if let Ok(resp) = self
            .client
            .describe_environment_resources()
            .environment_name(env_name)
            .send()
            .await
        {
            if let Some(res) = resp.environment_resources {
                for q in res.queues.unwrap_or_default() {
                    let name = q.name.unwrap_or_default();
                    let url = q.url.unwrap_or_default();
                    if url.is_empty() {
                        continue;
                    }
                    match name.as_str() {
                        "WorkerQueue" => main_url = Some(url),
                        "WorkerDeadLetterQueue" => dlq_url = Some(url),
                        _ => {}
                    }
                }
            }
        }

        // Fallback / override: look at user-supplied option settings in case
        // the env explicitly points at a queue the user manages outside EB.
        if main_url.is_none() || dlq_url.is_none() {
            if let Ok(resp) = self
                .client
                .describe_configuration_settings()
                .application_name(application_name)
                .environment_name(env_name)
                .send()
                .await
            {
                for setting in resp.configuration_settings.unwrap_or_default() {
                    for opt in setting.option_settings.unwrap_or_default() {
                        let ns = opt.namespace.unwrap_or_default();
                        let name = opt.option_name.unwrap_or_default();
                        if ns != "aws:elasticbeanstalk:sqsd" {
                            continue;
                        }
                        match name.as_str() {
                            "WorkerQueueURL" => {
                                let v = opt.value.unwrap_or_default();
                                if !v.is_empty() && main_url.is_none() {
                                    main_url = Some(v);
                                }
                            }
                            "DeadLetterQueueURL" => {
                                let v = opt.value.unwrap_or_default();
                                if !v.is_empty() && dlq_url.is_none() {
                                    dlq_url = Some(v);
                                }
                            }
                            _ => {}
                        }
                    }
                }
            }
        }

        // If we still have a main queue but no DLQ URL, derive one by SQS naming convention.
        if let (Some(main), None) = (&main_url, &dlq_url) {
            dlq_url = derive_dlq_url(main);
        }

        let main_stats = if let Some(u) = &main_url {
            self.queue_stats(u).await.ok()
        } else {
            None
        };
        let dlq_stats = if let Some(u) = &dlq_url {
            self.queue_stats(u).await.ok()
        } else {
            None
        };

        Ok(WorkerQueues {
            main_url,
            dlq_url,
            main_stats,
            dlq_stats,
        })
    }

    pub async fn queue_stats(&self, queue_url: &str) -> Result<QueueStats> {
        use aws_sdk_sqs::types::QueueAttributeName as Q;
        let resp = self
            .sqs
            .get_queue_attributes()
            .queue_url(queue_url)
            .attribute_names(Q::ApproximateNumberOfMessages)
            .attribute_names(Q::ApproximateNumberOfMessagesNotVisible)
            .attribute_names(Q::ApproximateNumberOfMessagesDelayed)
            .send()
            .await?;
        let attrs = resp.attributes.unwrap_or_default();
        let parse = |k: Q| -> i64 {
            attrs
                .get(&k)
                .and_then(|v| v.parse::<i64>().ok())
                .unwrap_or(0)
        };
        Ok(QueueStats {
            visible: parse(Q::ApproximateNumberOfMessages),
            in_flight: parse(Q::ApproximateNumberOfMessagesNotVisible),
            delayed: parse(Q::ApproximateNumberOfMessagesDelayed),
        })
    }

    /// Peek up to `max` messages from `queue_url` with a short visibility
    /// timeout (so we don't disrupt real consumers). SQS `ReceiveMessage`
    /// returns at most 10 per call AND, because the queue is partitioned, a
    /// single call commonly returns fewer than requested even with a deep
    /// queue. We therefore loop with a short long-poll, accumulating unique
    /// messages until we hit `max`, until two consecutive calls return zero,
    /// or until the per-call budget runs out. De-duplication is by message
    /// id — a partition can return the same message across calls within the
    /// visibility-timeout window if we're slow.
    pub async fn peek_messages(&self, queue_url: &str, max: i32) -> Result<Vec<QueueMessage>> {
        use aws_sdk_sqs::types::MessageSystemAttributeName as M;
        let target = max.clamp(1, 100) as usize;
        let mut out: Vec<QueueMessage> = Vec::new();
        let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
        let mut empty_in_a_row = 0;
        // Cap total iterations so a sparse queue can't spin forever.
        for _ in 0..((target / 10).max(1) + 4) {
            if out.len() >= target {
                break;
            }
            let resp = self
                .sqs
                .receive_message()
                .queue_url(queue_url)
                .max_number_of_messages(((target - out.len()).clamp(1, 10)) as i32)
                // Visibility timeout long enough to read + dedupe across the
                // loop without holding messages back from real consumers for
                // any noticeable time.
                .visibility_timeout(5)
                // Short long-poll: SQS will wait up to 1s for messages from
                // additional partitions before returning. Trades a little
                // latency for much better recall.
                .wait_time_seconds(1)
                .message_system_attribute_names(M::ApproximateReceiveCount)
                .message_system_attribute_names(M::SentTimestamp)
                .send()
                .await
                .wrap_err("ReceiveMessage failed")?;
            let batch = resp.messages.unwrap_or_default();
            if batch.is_empty() {
                empty_in_a_row += 1;
                if empty_in_a_row >= 2 {
                    break;
                }
                continue;
            }
            empty_in_a_row = 0;
            for m in batch {
                let id = m.message_id.clone().unwrap_or_default();
                if !id.is_empty() && !seen.insert(id.clone()) {
                    continue;
                }
                let attrs = m.attributes.unwrap_or_default();
                let receive_count = attrs
                    .get(&M::ApproximateReceiveCount)
                    .and_then(|v| v.parse::<i64>().ok())
                    .unwrap_or(0);
                let sent_at = attrs
                    .get(&M::SentTimestamp)
                    .and_then(|v| v.parse::<i64>().ok())
                    .and_then(DateTime::from_timestamp_millis);
                out.push(QueueMessage {
                    id,
                    receipt_handle: m.receipt_handle.unwrap_or_default(),
                    body: m.body.unwrap_or_default(),
                    receive_count,
                    sent_at,
                });
                if out.len() >= target {
                    break;
                }
            }
        }
        Ok(out)
    }

    pub async fn send_message(&self, queue_url: &str, body: &str) -> Result<()> {
        self.sqs
            .send_message()
            .queue_url(queue_url)
            .message_body(body)
            .send()
            .await?;
        Ok(())
    }

    pub async fn delete_message(&self, queue_url: &str, receipt_handle: &str) -> Result<()> {
        self.sqs
            .delete_message()
            .queue_url(queue_url)
            .receipt_handle(receipt_handle)
            .send()
            .await?;
        Ok(())
    }

    /// Describe metric alarms whose first dimension references the given env.
    /// CloudWatch doesn't expose a server-side filter by dimension, so we pull
    /// alarms in the AWS/ElasticBeanstalk namespace and filter client-side.
    /// Per-env monthly cost from AWS Cost Explorer. One round trip;
    /// returns a row per env tag value the Cost Explorer API saw in
    /// the trailing-30-day window.
    ///
    /// Cost Explorer is rate-limited (~1 req/s per account) and slow
    /// (1-3s per query) — the caller is expected to cache the result
    /// for ~24h via `crate::cost_cache`. The 24h granularity matches
    /// AWS's own data freshness (most cost data lags ~24h).
    ///
    /// Returned costs span the full window — divide by ~30 days for
    /// a daily rate, or treat as a monthly figure (which is what
    /// every operator actually wants).
    ///
    /// Tag key: `elasticbeanstalk:environment-name` (the EB-set tag
    /// AWS adds to every env-owned resource by default). Envs whose
    /// resources have been re-tagged or never carried the tag won't
    /// show up — surface as zero / unknown rather than guessing.
    pub async fn fetch_env_costs(&self) -> Result<Vec<EnvCost>> {
        use aws_sdk_costexplorer::types::{DateInterval, GroupDefinition, GroupDefinitionType};

        // Trailing window — end is "today" (exclusive in Cost Explorer)
        // so the inclusive Start is 30 days ago. Cost Explorer dates
        // are ISO-8601 (YYYY-MM-DD) in UTC.
        let now = chrono::Utc::now().date_naive();
        let start = (now - chrono::Duration::days(30))
            .format("%Y-%m-%d")
            .to_string();
        let end = now.format("%Y-%m-%d").to_string();
        let time_period = DateInterval::builder()
            .start(start)
            .end(end)
            .build()
            .wrap_err("Cost Explorer DateInterval missing field")?;
        let group_by = GroupDefinition::builder()
            .r#type(GroupDefinitionType::Tag)
            .key("elasticbeanstalk:environment-name")
            .build();
        let resp = self
            .cost
            .get_cost_and_usage()
            .time_period(time_period)
            .granularity(aws_sdk_costexplorer::types::Granularity::Monthly)
            .metrics("UnblendedCost")
            .group_by(group_by)
            .send()
            .await
            .wrap_err("GetCostAndUsage failed")?;

        // Result format: results_by_time[N].groups[].keys[0] is the
        // tag value (prefixed with `elasticbeanstalk:environment-name$`
        // — the Cost Explorer SDK encodes the tag key in the group
        // key, separated by `$`). Strip the prefix to recover the env
        // name. Sum across the time buckets in case the window spans
        // multiple months.
        let mut totals: std::collections::HashMap<String, f64> = std::collections::HashMap::new();
        for period in resp.results_by_time.unwrap_or_default() {
            for group in period.groups.unwrap_or_default() {
                let raw_key = match group.keys.as_ref().and_then(|k| k.first()) {
                    Some(k) => k.clone(),
                    None => continue,
                };
                // Cost Explorer encodes a tag group key as
                // `elasticbeanstalk:environment-name$<value>`. The
                // empty-tag bucket (resources untagged) shows up as
                // the bare prefix — skip it.
                let env_name = match raw_key.split_once('$') {
                    Some((_, v)) if !v.is_empty() => v.to_string(),
                    _ => continue,
                };
                let amount: f64 = group
                    .metrics
                    .as_ref()
                    .and_then(|m| m.get("UnblendedCost"))
                    .and_then(|m| m.amount.as_deref())
                    .and_then(|s| s.parse().ok())
                    .unwrap_or(0.0);
                *totals.entry(env_name).or_insert(0.0) += amount;
            }
        }
        let mut out: Vec<EnvCost> = totals
            .into_iter()
            .map(|(env_name, cost_usd)| EnvCost { env_name, cost_usd })
            .collect();
        // Stable ordering — highest-cost first so the operator's eye
        // catches the expensive envs without scrolling.
        out.sort_by(|a, b| {
            b.cost_usd
                .partial_cmp(&a.cost_usd)
                .unwrap_or(std::cmp::Ordering::Equal)
        });
        Ok(out)
    }

    pub async fn list_alarms_for_env(&self, env_name: &str) -> Result<Vec<CwAlarm>> {
        let mut out = Vec::new();
        let mut next_token: Option<String> = None;
        loop {
            let mut req = self.cw.describe_alarms();
            if let Some(t) = next_token.take() {
                req = req.next_token(t);
            }
            let resp = req.send().await.wrap_err("DescribeAlarms failed")?;
            for a in resp.metric_alarms.unwrap_or_default() {
                let dims = a.dimensions.clone().unwrap_or_default();
                let touches = dims.iter().any(|d| d.value.as_deref() == Some(env_name));
                if !touches {
                    continue;
                }
                out.push(CwAlarm {
                    name: a.alarm_name.unwrap_or_default(),
                    state: a
                        .state_value
                        .map(|s| s.as_str().to_string())
                        .unwrap_or_default(),
                    state_reason: a.state_reason.unwrap_or_default(),
                    metric_name: a.metric_name.unwrap_or_default(),
                    namespace: a.namespace.unwrap_or_default(),
                });
            }
            match resp.next_token {
                Some(t) if !t.is_empty() => next_token = Some(t),
                _ => break,
            }
        }
        Ok(out)
    }

    /// Create or update a CloudWatch metric alarm in the
    /// `AWS/ElasticBeanstalk` namespace, dimensioned by `EnvironmentName`.
    /// `metric_name` should be one of the env-scoped metrics already in our
    /// Metrics tab (EnvironmentHealth / ApplicationRequests4xx /
    /// ApplicationRequests5xx / ApplicationLatencyP90) — anything else and
    /// the alarm will be created with no datapoints. No alarm actions are
    /// attached; operators can wire SNS via the console or CLI later.
    #[allow(clippy::too_many_arguments)]
    pub async fn put_env_metric_alarm(
        &self,
        alarm_name: &str,
        env_name: &str,
        metric_name: &str,
        threshold: f64,
        comparison_operator: &str,
        period_secs: i32,
        evaluation_periods: i32,
        statistic: &str,
    ) -> Result<()> {
        use aws_sdk_cloudwatch::types::{ComparisonOperator, Dimension, Statistic};
        // The smithy enums round-trip "unknown" inputs through their Unknown
        // variant; checking `as_str()` against the original input is the
        // documented way to detect that case without matching on the
        // deprecated variant.
        let op = ComparisonOperator::from(comparison_operator);
        if op.as_str() != comparison_operator {
            return Err(eyre!(
                "unknown comparison operator '{comparison_operator}' \
                 (valid: GreaterThanThreshold, GreaterThanOrEqualToThreshold, \
                 LessThanThreshold, LessThanOrEqualToThreshold)"
            ));
        }
        let stat = Statistic::from(statistic);
        if stat.as_str() != statistic {
            return Err(eyre!(
                "unknown statistic '{statistic}' (valid: Average, Sum, Maximum, Minimum, SampleCount)"
            ));
        }
        let dim = Dimension::builder()
            .name("EnvironmentName")
            .value(env_name)
            .build();
        self.cw
            .put_metric_alarm()
            .alarm_name(alarm_name)
            .alarm_description(format!("ebman: {metric_name} alarm on {env_name}"))
            .namespace("AWS/ElasticBeanstalk")
            .metric_name(metric_name)
            .dimensions(dim)
            .comparison_operator(op)
            .threshold(threshold)
            .period(period_secs)
            .evaluation_periods(evaluation_periods)
            .statistic(stat)
            .treat_missing_data("notBreaching")
            .send()
            .await
            .wrap_err("PutMetricAlarm failed")?;
        Ok(())
    }

    /// Fetch the current env vars for an environment from
    /// `DescribeConfigurationSettings` filtered to the
    /// `aws:elasticbeanstalk:application:environment` namespace. Returns
    /// sorted `(KEY, VALUE)` pairs.
    /// Fetch every option setting for a live env. Used by the modal-form
    /// pre-fill: callers filter the result down to the `(namespace, option_name)`
    /// pairs their form cares about. Returns `(namespace, option_name, value)`
    /// triples.
    pub async fn fetch_env_option_settings(
        &self,
        application_name: &str,
        env_name: &str,
    ) -> Result<Vec<(String, String, String)>> {
        let resp = self
            .client
            .describe_configuration_settings()
            .application_name(application_name)
            .environment_name(env_name)
            .send()
            .await
            .wrap_err("DescribeConfigurationSettings(env) failed")?;
        let out = resp
            .configuration_settings
            .unwrap_or_default()
            .into_iter()
            .flat_map(|c| c.option_settings.unwrap_or_default())
            .map(|o| {
                (
                    o.namespace.unwrap_or_default(),
                    o.option_name.unwrap_or_default(),
                    o.value.unwrap_or_default(),
                )
            })
            .collect();
        Ok(out)
    }

    /// Pull the env's VPC id plus the currently-selected subnet and
    /// security-group IDs from EB option settings in a single round-trip.
    /// `:subnets` and `:security-groups` both call this — VPC id drives
    /// the subsequent EC2 list call, the existing selections drive the
    /// MultiSelect pre-fill.
    pub async fn fetch_env_vpc_context(
        &self,
        application_name: &str,
        env_name: &str,
    ) -> Result<EnvVpcContext> {
        let resp = self
            .client
            .describe_configuration_settings()
            .application_name(application_name)
            .environment_name(env_name)
            .send()
            .await
            .wrap_err("DescribeConfigurationSettings(env) failed")?;
        let mut ctx = EnvVpcContext::default();
        for setting in resp.configuration_settings.unwrap_or_default() {
            for opt in setting.option_settings.unwrap_or_default() {
                let ns = opt.namespace.unwrap_or_default();
                let name = opt.option_name.unwrap_or_default();
                let value = opt.value.unwrap_or_default();
                match (ns.as_str(), name.as_str()) {
                    ("aws:ec2:vpc", "VPCId") if !value.is_empty() => {
                        ctx.vpc_id = Some(value);
                    }
                    ("aws:ec2:vpc", "Subnets") if !value.is_empty() => {
                        ctx.subnets = split_csv(&value);
                    }
                    ("aws:ec2:vpc", "ELBSubnets") if !value.is_empty() => {
                        ctx.elb_subnets = split_csv(&value);
                    }
                    ("aws:autoscaling:launchconfiguration", "SecurityGroups")
                        if !value.is_empty() =>
                    {
                        ctx.security_groups = split_csv(&value);
                    }
                    _ => {}
                }
            }
        }
        Ok(ctx)
    }

    /// List subnets in a VPC, ordered by AZ then CIDR for stable picker
    /// rows. Returns the wide rows the `:subnets` picker needs (id + AZ
    /// + CIDR + Name tag) so callers don't need a second round-trip.
    pub async fn list_subnets_in_vpc(&self, vpc_id: &str) -> Result<Vec<SubnetInfo>> {
        use aws_sdk_ec2::types::Filter;
        let resp = self
            .ec2
            .describe_subnets()
            .filters(
                Filter::builder()
                    .name("vpc-id")
                    .values(vpc_id.to_string())
                    .build(),
            )
            .send()
            .await
            .wrap_err("DescribeSubnets failed")?;
        let mut out: Vec<SubnetInfo> = resp
            .subnets
            .unwrap_or_default()
            .into_iter()
            .map(|s| {
                let name_tag = s.tags.as_ref().and_then(|tags| {
                    tags.iter()
                        .find(|t| t.key.as_deref() == Some("Name"))
                        .and_then(|t| t.value.clone())
                });
                SubnetInfo {
                    id: s.subnet_id.unwrap_or_default(),
                    availability_zone: s.availability_zone.unwrap_or_default(),
                    cidr_block: s.cidr_block.unwrap_or_default(),
                    name_tag,
                }
            })
            .collect();
        out.sort_by(|a, b| {
            a.availability_zone
                .cmp(&b.availability_zone)
                .then(a.cidr_block.cmp(&b.cidr_block))
        });
        Ok(out)
    }

    /// List security groups in a VPC, ordered by name for stable picker
    /// rows.
    pub async fn list_security_groups_in_vpc(
        &self,
        vpc_id: &str,
    ) -> Result<Vec<SecurityGroupInfo>> {
        use aws_sdk_ec2::types::Filter;
        let resp = self
            .ec2
            .describe_security_groups()
            .filters(
                Filter::builder()
                    .name("vpc-id")
                    .values(vpc_id.to_string())
                    .build(),
            )
            .send()
            .await
            .wrap_err("DescribeSecurityGroups failed")?;
        let mut out: Vec<SecurityGroupInfo> = resp
            .security_groups
            .unwrap_or_default()
            .into_iter()
            .map(|g| SecurityGroupInfo {
                id: g.group_id.unwrap_or_default(),
                group_name: g.group_name.unwrap_or_default(),
                description: g.description.unwrap_or_default(),
            })
            .collect();
        out.sort_by(|a, b| a.group_name.cmp(&b.group_name));
        Ok(out)
    }

    /// Fetch RDS dbinstance option settings for an env. EB envs
    /// optionally have an attached RDS instance (via
    /// `aws:rds:dbinstance.*` option settings + auto-managed
    /// security group); this returns the configured settings as
    /// `(option_name, value)` pairs sorted alphabetically.
    ///
    /// Empty result = no RDS attached. Caller should distinguish
    /// "no RDS" from "fetch failed" via the Result type.
    pub async fn fetch_env_rds_config(
        &self,
        application_name: &str,
        env_name: &str,
    ) -> Result<Vec<(String, String)>> {
        let resp = self
            .client
            .describe_configuration_settings()
            .application_name(application_name)
            .environment_name(env_name)
            .send()
            .await
            .wrap_err("DescribeConfigurationSettings(rds) failed")?;
        let mut out: Vec<(String, String)> = resp
            .configuration_settings
            .unwrap_or_default()
            .into_iter()
            .flat_map(|c| c.option_settings.unwrap_or_default())
            .filter_map(|o| {
                let ns = o.namespace?;
                if ns != "aws:rds:dbinstance" {
                    return None;
                }
                let opt = o.option_name?;
                let value = o.value.unwrap_or_default();
                if value.is_empty() {
                    return None;
                }
                Some((opt, value))
            })
            .collect();
        out.sort();
        Ok(out)
    }

    /// Fetch every settable EB option for an env — namespace, name,
    /// current value (when set), default, type, constraints.
    ///
    /// Two SDK calls correlated by (namespace, name):
    ///
    ///   - `describe_configuration_options` is the canonical
    ///     "what's the full config vocabulary for this env's
    ///     platform?" API. Returns ~hundreds of option metadata
    ///     rows (default value, value type, change severity,
    ///     constraints) — but no current values.
    ///   - `describe_configuration_settings` returns the current
    ///     values for *every* option, including ones still at
    ///     their default.
    ///
    /// Merged on namespace+name so each row carries both the
    /// metadata and the live value. This is what closes the
    /// operator's "how do I know what I can set?" question.
    /// Caller should treat as on-demand (run via `:options`), not
    /// part of the background refresh — both calls are slow for
    /// platforms with deep option trees.
    /// Call `iam:SimulatePrincipalPolicy` for a role + action list.
    /// Returns the per-action decision (allowed / explicitDeny /
    /// implicitDeny), matched statements, and SCP / permission-
    /// boundary blocker flags. Powers `:explain`.
    ///
    /// `resource_arns` defaults to `["*"]` when empty — most EB
    /// AccessDenied errors don't carry a resource ARN that would
    /// affect the decision, and the unscoped check still surfaces
    /// "the role doesn't have this action at all" cases which is
    /// what the operator usually wants. Pass real ARNs when you
    /// want to evaluate resource-scoped policies.
    ///
    /// Errors out of the SimulatePrincipalPolicy itself usually
    /// mean the caller lacks `iam:SimulatePrincipalPolicy` on the
    /// target role — common with assumed-role sessions that don't
    /// have IAM perms. The renderer surfaces that as a clear hint.
    /// List Secrets Manager secrets in the active region.
    /// `name_filter` is an optional substring match against the
    /// secret name (case-sensitive — Secrets Manager's
    /// `Filters.Key=name` does prefix matching only, so we
    /// post-filter for substring instead).
    ///
    /// Paginates internally. Returns the metadata rows; no
    /// secret *values* are fetched here — see [`fetch_secret_value`].
    pub async fn list_secrets(&self, name_filter: Option<&str>) -> Result<Vec<SecretSummary>> {
        let mut out: Vec<SecretSummary> = Vec::new();
        let mut next: Option<String> = None;
        loop {
            let mut req = self.secrets.list_secrets();
            if let Some(t) = next.take() {
                req = req.next_token(t);
            }
            let resp = req.send().await.wrap_err("ListSecrets failed")?;
            for s in resp.secret_list.unwrap_or_default() {
                let name = match s.name {
                    Some(n) if !n.is_empty() => n,
                    _ => continue,
                };
                if let Some(needle) = name_filter {
                    if !name.contains(needle) {
                        continue;
                    }
                }
                out.push(SecretSummary {
                    name,
                    arn: s.arn.unwrap_or_default(),
                    description: s.description.filter(|d| !d.is_empty()),
                    last_changed: s
                        .last_changed_date
                        .and_then(|d| DateTime::from_timestamp(d.secs(), d.subsec_nanos())),
                    last_rotated: s
                        .last_rotated_date
                        .and_then(|d| DateTime::from_timestamp(d.secs(), d.subsec_nanos())),
                    kms_key_id: s.kms_key_id.filter(|k| !k.is_empty()),
                });
            }
            match resp.next_token {
                Some(t) if !t.is_empty() => next = Some(t),
                _ => break,
            }
        }
        // Stable order — most-recently-changed first so freshly
        // rotated secrets float to the top of the picker.
        out.sort_by_key(|r| std::cmp::Reverse(r.last_changed));
        Ok(out)
    }

    /// `GetSecretValue` for one secret. Returns the value verbatim —
    /// caller decides whether to display, redact, or yank.
    /// Audit-loggable on the AWS side (CloudTrail logs every
    /// GetSecretValue); ebman additionally writes its own audit
    /// line via the caller path.
    pub async fn fetch_secret_value(&self, secret_id: &str) -> Result<String> {
        let resp = self
            .secrets
            .get_secret_value()
            .secret_id(secret_id)
            .send()
            .await
            .wrap_err("GetSecretValue failed")?;
        // Secrets Manager returns either SecretString (UTF-8 text,
        // including JSON for k/v secrets) or SecretBinary (base64
        // blob). Prefer the string; fall back to noting the binary
        // length so the operator doesn't try to inspect.
        if let Some(s) = resp.secret_string {
            return Ok(s);
        }
        if let Some(b) = resp.secret_binary {
            return Ok(format!("(binary, {} bytes — not shown)", b.as_ref().len()));
        }
        Ok(String::new())
    }

    pub async fn simulate_principal_policy(
        &self,
        principal_arn: &str,
        action_names: &[String],
        resource_arns: &[String],
    ) -> Result<Vec<IamSimResult>> {
        if action_names.is_empty() {
            return Ok(Vec::new());
        }
        let resources: Vec<String> = if resource_arns.is_empty() {
            vec!["*".to_string()]
        } else {
            resource_arns.to_vec()
        };
        let mut req = self
            .iam
            .simulate_principal_policy()
            .policy_source_arn(principal_arn);
        for a in action_names {
            req = req.action_names(a);
        }
        for r in &resources {
            req = req.resource_arns(r);
        }
        let resp = req
            .send()
            .await
            .wrap_err("SimulatePrincipalPolicy failed")?;
        let mut out: Vec<IamSimResult> = Vec::new();
        for r in resp.evaluation_results.unwrap_or_default() {
            let action = r.eval_action_name;
            let resource = r.eval_resource_name.unwrap_or_default();
            let decision = r.eval_decision.as_str().to_string();
            let matched_statements: Vec<String> = r
                .matched_statements
                .unwrap_or_default()
                .into_iter()
                .filter_map(|s| {
                    let policy = s.source_policy_id?;
                    // SDK returns start_position as `Option<Position>`
                    // with `line` + `column` already as `i32` (not
                    // Option). Format defensively in case the
                    // position is missing for an inline-eval result.
                    let sid = s
                        .start_position
                        .as_ref()
                        .map(|p| format!("{}:{}", p.line, p.column))
                        .unwrap_or_else(|| "0:0".into());
                    Some(format!("{policy} @ {sid}"))
                })
                .collect();
            let missing_context: Vec<String> = r.missing_context_values.unwrap_or_default();
            // SCP / boundary blockers — only populated when the
            // top-level decision was overridden by an org-level
            // policy or the role's permission boundary. Both fields
            // carry an `EvalDecisionDetail` we just need the
            // `allowed_by_organizations` / `allowed_by_permissions_boundary`
            // flag for.
            let blocked_by_scp = r
                .organizations_decision_detail
                .as_ref()
                .is_some_and(|d| !d.allowed_by_organizations);
            let blocked_by_boundary = r
                .permissions_boundary_decision_detail
                .as_ref()
                .is_some_and(|d| !d.allowed_by_permissions_boundary);
            out.push(IamSimResult {
                action,
                resource,
                decision,
                matched_statements,
                missing_context,
                blocked_by_scp,
                blocked_by_boundary,
            });
        }
        Ok(out)
    }

    pub async fn fetch_env_configuration_options(
        &self,
        application_name: &str,
        env_name: &str,
    ) -> Result<Vec<ConfigOption>> {
        // Parallel fetch of both shapes. The vocabulary call is
        // the slower of the two, so kicking them off together
        // shaves a round-trip off the total latency.
        let vocab_fut = self
            .client
            .describe_configuration_options()
            .environment_name(env_name)
            .send();
        let settings_fut = self
            .client
            .describe_configuration_settings()
            .application_name(application_name)
            .environment_name(env_name)
            .send();
        let (vocab_resp, settings_resp) = tokio::try_join!(
            async {
                vocab_fut
                    .await
                    .wrap_err("DescribeConfigurationOptions failed")
            },
            async {
                settings_fut
                    .await
                    .wrap_err("DescribeConfigurationSettings(options) failed")
            },
        )?;

        // Index current values by (namespace, name).
        let mut current: std::collections::HashMap<(String, String), String> =
            std::collections::HashMap::new();
        for c in settings_resp.configuration_settings.unwrap_or_default() {
            for o in c.option_settings.unwrap_or_default() {
                if let (Some(ns), Some(name)) = (o.namespace, o.option_name) {
                    if let Some(v) = o.value {
                        if !v.is_empty() {
                            current.insert((ns, name), v);
                        }
                    }
                }
            }
        }

        let mut out: Vec<ConfigOption> = vocab_resp
            .options
            .unwrap_or_default()
            .into_iter()
            .filter_map(|o| {
                let namespace = o.namespace?;
                let name = o.name?;
                let value = current.get(&(namespace.clone(), name.clone())).cloned();
                Some(ConfigOption {
                    namespace,
                    name,
                    value,
                    default_value: o.default_value,
                    value_type: o
                        .value_type
                        .map(|v| v.as_str().to_string())
                        .unwrap_or_default(),
                    value_options: o.value_options.unwrap_or_default(),
                    change_severity: o.change_severity,
                    user_defined: o.user_defined,
                    min_value: o.min_value,
                    max_value: o.max_value,
                    max_length: o.max_length,
                })
            })
            .collect();
        // Sort: namespace asc, user-set first within each namespace,
        // then alpha by name. Puts the operator's mutations at the
        // top of each group where they catch the eye.
        out.sort_by(|a, b| {
            let a_set = a.value.is_some();
            let b_set = b.value.is_some();
            a.namespace
                .cmp(&b.namespace)
                .then_with(|| b_set.cmp(&a_set))
                .then_with(|| a.name.cmp(&b.name))
        });
        Ok(out)
    }

    /// Fetch ALB listener option settings for an env. EB stores
    /// listener config in `aws:elbv2:listener:<PORT>` namespaces (one
    /// per listener; `default` is the port-80 HTTP listener, `443`
    /// is the typical HTTPS one). Returns a Vec of
    /// `(port_or_default, option_name, value)` rows so the renderer
    /// can group by port.
    ///
    /// Result is empty when the env doesn't use an ALB (Classic LB
    /// or worker tier) — caller should distinguish from "no config"
    /// by checking the env's tier first.
    pub async fn fetch_env_listeners(
        &self,
        application_name: &str,
        env_name: &str,
    ) -> Result<Vec<(String, String, String)>> {
        let resp = self
            .client
            .describe_configuration_settings()
            .application_name(application_name)
            .environment_name(env_name)
            .send()
            .await
            .wrap_err("DescribeConfigurationSettings(listeners) failed")?;
        let mut out: Vec<(String, String, String)> = resp
            .configuration_settings
            .unwrap_or_default()
            .into_iter()
            .flat_map(|c| c.option_settings.unwrap_or_default())
            .filter_map(|o| {
                let ns = o.namespace?;
                // Listener namespaces look like
                // `aws:elbv2:listener:default` / `aws:elbv2:listener:443`.
                // Strip the prefix to get the port (or "default").
                let port = ns.strip_prefix("aws:elbv2:listener:")?.to_string();
                let opt = o.option_name?;
                let value = o.value.unwrap_or_default();
                // Skip empty values — EB returns every settable key
                // even when unset, and an empty cert ARN / rule
                // list isn't worth showing.
                if value.is_empty() {
                    return None;
                }
                Some((port, opt, value))
            })
            .collect();
        // Sort: 'default' (port 80) first, then numeric ports asc,
        // then alpha by option name within each listener.
        out.sort_by(|a, b| {
            let rank_a = u8::from(a.0 != "default");
            let rank_b = u8::from(b.0 != "default");
            let port_a = a.0.parse::<u32>().unwrap_or(0);
            let port_b = b.0.parse::<u32>().unwrap_or(0);
            (rank_a, port_a, &a.1).cmp(&(rank_b, port_b, &b.1))
        });
        Ok(out)
    }

    /// List the region's ACM certificates (ISSUED only) as
    /// `(arn, primary domain)`. Drives the `:listener-edit` cert picker.
    pub async fn list_certificates(&self) -> Result<Vec<AcmCert>> {
        use aws_sdk_acm::types::CertificateStatus;
        let mut out: Vec<AcmCert> = Vec::new();
        let mut next_token: Option<String> = None;
        loop {
            let mut req = self
                .acm
                .list_certificates()
                .certificate_statuses(CertificateStatus::Issued);
            if let Some(t) = next_token.take() {
                req = req.next_token(t);
            }
            let resp = req.send().await.wrap_err("ListCertificates failed")?;
            for c in resp.certificate_summary_list.unwrap_or_default() {
                if let Some(arn) = c.certificate_arn {
                    out.push(AcmCert {
                        arn,
                        domain: c.domain_name.unwrap_or_default(),
                    });
                }
            }
            match resp.next_token {
                Some(t) if !t.is_empty() => next_token = Some(t),
                _ => break,
            }
        }
        out.sort_by(|a, b| a.domain.cmp(&b.domain));
        Ok(out)
    }

    pub async fn fetch_env_vars(
        &self,
        application_name: &str,
        env_name: &str,
    ) -> Result<Vec<(String, String)>> {
        let resp = self
            .client
            .describe_configuration_settings()
            .application_name(application_name)
            .environment_name(env_name)
            .send()
            .await
            .wrap_err("DescribeConfigurationSettings(env) failed")?;
        let mut out: Vec<(String, String)> = resp
            .configuration_settings
            .unwrap_or_default()
            .into_iter()
            .flat_map(|c| c.option_settings.unwrap_or_default())
            .filter(|o| {
                o.namespace.as_deref() == Some("aws:elasticbeanstalk:application:environment")
            })
            .map(|o| {
                (
                    o.option_name.unwrap_or_default(),
                    o.value.unwrap_or_default(),
                )
            })
            .collect();
        out.sort();
        Ok(out)
    }

    /// Update an env's option settings — `to_set` is `(namespace, option_name,
    /// value)` triples to add or overwrite; `to_remove` is `(namespace,
    /// option_name)` pairs to clear back to defaults. EB applies the change
    /// as a rolling update (or instantly for non-disruptive options).
    pub async fn update_env_option_settings(
        &self,
        env_name: &str,
        to_set: &[(String, String, String)],
        to_remove: &[(String, String)],
    ) -> Result<()> {
        use aws_sdk_elasticbeanstalk::types::{ConfigurationOptionSetting, OptionSpecification};
        if to_set.is_empty() && to_remove.is_empty() {
            return Err(eyre!("update_env_option_settings: nothing to do"));
        }
        let mut req = self.client.update_environment().environment_name(env_name);
        for (ns, name, value) in to_set {
            req = req.option_settings(
                ConfigurationOptionSetting::builder()
                    .namespace(ns)
                    .option_name(name)
                    .value(value)
                    .build(),
            );
        }
        for (ns, name) in to_remove {
            req = req.options_to_remove(
                OptionSpecification::builder()
                    .namespace(ns)
                    .option_name(name)
                    .build(),
            );
        }
        req.send()
            .await
            .wrap_err("UpdateEnvironment(option_settings) failed")?;
        Ok(())
    }

    /// Discover the CloudWatch Logs groups an EB env streams to. EB names
    /// them under the prefix `/aws/elasticbeanstalk/{env}/...` so we
    /// `DescribeLogGroups` with that prefix. Returns sorted group names;
    /// empty if `:logs-stream on` hasn't been issued for the env.
    pub async fn discover_env_log_groups(&self, env_name: &str) -> Result<Vec<String>> {
        let prefix = format!("/aws/elasticbeanstalk/{env_name}/");
        let mut out: Vec<String> = Vec::new();
        let mut next_token: Option<String> = None;
        loop {
            let mut req = self
                .cw_logs
                .describe_log_groups()
                .log_group_name_prefix(&prefix);
            if let Some(t) = next_token.take() {
                req = req.next_token(t);
            }
            let resp = req.send().await.wrap_err("DescribeLogGroups failed")?;
            for g in resp.log_groups.unwrap_or_default() {
                if let Some(name) = g.log_group_name {
                    out.push(name);
                }
            }
            match resp.next_token {
                Some(t) if !t.is_empty() => next_token = Some(t),
                _ => break,
            }
        }
        out.sort();
        Ok(out)
    }

    /// Fetch events from one CW Logs group since `since_ms` (Unix
    /// milliseconds). Uses `FilterLogEvents` so the result spans all log
    /// streams in the group in chronological order — that's how an EB-tier
    /// log group works (one stream per instance). The returned tuple is
    /// `(events, next_since_ms)` where `next_since_ms` is the highest
    /// timestamp + 1 we saw, suitable to pass back on the next call.
    pub async fn fetch_recent_log_events(
        &self,
        log_group: &str,
        since_ms: i64,
        limit: i32,
    ) -> Result<(Vec<LogEvent>, i64)> {
        let resp = self
            .cw_logs
            .filter_log_events()
            .log_group_name(log_group)
            .start_time(since_ms)
            .limit(limit)
            .send()
            .await
            .wrap_err("FilterLogEvents failed")?;
        let mut out: Vec<LogEvent> = Vec::new();
        let mut max_ts = since_ms;
        for e in resp.events.unwrap_or_default() {
            let ts = e.timestamp.unwrap_or(since_ms);
            if ts > max_ts {
                max_ts = ts;
            }
            out.push(LogEvent {
                timestamp_ms: ts,
                stream: e.log_stream_name.unwrap_or_default(),
                message: e.message.unwrap_or_default(),
            });
        }
        // Move the cursor past the last event we saw so the next poll
        // doesn't return it again.
        let next_since = if max_ts > since_ms {
            max_ts + 1
        } else {
            since_ms
        };
        Ok((out, next_since))
    }

    /// Run a CloudWatch Logs Insights query against `log_groups` over the
    /// `[start_ms, end_ms]` window. Starts the query via `StartQuery`, polls
    /// `GetQueryResults` every 2 seconds until the status leaves
    /// Scheduled/Running, and returns the final result rows + scan stats.
    /// The terminal Failed / Cancelled / Timeout states surface as a clean
    /// error rather than empty rows so the caller can show the right toast.
    pub async fn run_insights_query(
        &self,
        log_groups: &[String],
        start_ms: i64,
        end_ms: i64,
        query: &str,
    ) -> Result<InsightsResults> {
        use aws_sdk_cloudwatchlogs::types::QueryStatus;
        // StartQuery's start_time / end_time are epoch *seconds*, not ms.
        let start_s = start_ms / 1000;
        let end_s = end_ms / 1000;
        let mut req = self
            .cw_logs
            .start_query()
            .start_time(start_s)
            .end_time(end_s)
            .query_string(query);
        for g in log_groups {
            req = req.log_group_names(g);
        }
        let start_resp = req.send().await.wrap_err("StartQuery failed")?;
        let query_id = start_resp
            .query_id
            .ok_or_else(|| eyre!("StartQuery returned no query_id"))?;

        // Poll every 2s. Insights queries are server-side timed (max 15
        // min by default) so we don't need our own watchdog; the API
        // surfaces Timeout when the server gives up.
        loop {
            tokio::time::sleep(std::time::Duration::from_secs(2)).await;
            let resp = self
                .cw_logs
                .get_query_results()
                .query_id(&query_id)
                .send()
                .await
                .wrap_err("GetQueryResults failed")?;
            let status = resp.status.clone();
            let scanned = resp
                .statistics
                .as_ref()
                .map(|s| s.records_scanned as i64)
                .unwrap_or(0);
            let matched = resp
                .statistics
                .as_ref()
                .map(|s| s.records_matched as i64)
                .unwrap_or(0);
            match status {
                Some(QueryStatus::Scheduled) | Some(QueryStatus::Running) => continue,
                Some(QueryStatus::Complete) => {
                    let rows: Vec<InsightsRow> = resp
                        .results
                        .unwrap_or_default()
                        .into_iter()
                        .map(|fields| InsightsRow {
                            fields: fields
                                .into_iter()
                                .map(|f| (f.field.unwrap_or_default(), f.value.unwrap_or_default()))
                                .collect(),
                        })
                        .collect();
                    return Ok(InsightsResults {
                        rows,
                        records_scanned: scanned,
                        records_matched: matched,
                    });
                }
                Some(QueryStatus::Failed) => {
                    return Err(eyre!("Insights query failed"));
                }
                Some(QueryStatus::Cancelled) => {
                    return Err(eyre!("Insights query was cancelled"));
                }
                Some(QueryStatus::Timeout) => {
                    return Err(eyre!("Insights query timed out (server-side 15min cap)"));
                }
                Some(other) => {
                    return Err(eyre!(
                        "unexpected Insights query status: {}",
                        other.as_str()
                    ));
                }
                None => {
                    return Err(eyre!("Insights query returned no status"));
                }
            }
        }
    }

    /// Send a shell command to `instance_ids` via SSM Run Command
    /// (`AWS-RunShellScript` document), poll per-instance results until
    /// every invocation reaches a terminal status, and return the
    /// aggregated rows. `wall_clock_secs` bounds total time and feeds
    /// into the SSM-side `TimeoutSeconds` parameter (the server kills
    /// commands that don't start running within that window).
    ///
    /// The poll loop sleeps 2s between cycles — the same cadence
    /// `run_insights_query` uses. SSM has no aggregate "all done" call
    /// the way Insights does, so we poll `GetCommandInvocation` per
    /// instance per cycle and drop instances out of the wait set once
    /// they're terminal. Returns once every instance is terminal *or*
    /// the wall-clock deadline has passed (operator gets best-effort
    /// partial results either way).
    pub async fn run_shell_command(
        &self,
        instance_ids: &[String],
        command: &str,
        wall_clock_secs: u64,
    ) -> Result<Vec<SsmRunResult>> {
        use aws_sdk_ssm::types::CommandInvocationStatus;
        if instance_ids.is_empty() {
            return Err(eyre!("run_shell_command: no instance ids"));
        }
        // Cap the per-command timeout at 600s (SSM's default upper
        // bound for non-MaintenanceWindow runs is 2880s; staying well
        // under it avoids surprising server-side rejections).
        let ssm_timeout = wall_clock_secs.min(600) as i32;
        let mut send = self
            .ssm
            .send_command()
            .document_name("AWS-RunShellScript")
            .timeout_seconds(ssm_timeout)
            .parameters("commands", vec![command.to_string()]);
        for id in instance_ids {
            send = send.instance_ids(id);
        }
        let send_resp = send.send().await.wrap_err("SendCommand failed")?;
        let command_id = send_resp
            .command
            .and_then(|c| c.command_id)
            .ok_or_else(|| eyre!("SendCommand returned no command_id"))?;

        // Track which instances are still pending. SSM needs ~a second
        // before invocations become queryable; the loop's first iteration
        // tolerates an InvocationDoesNotExist error.
        let mut pending: std::collections::HashSet<String> = instance_ids.iter().cloned().collect();
        let mut completed: Vec<SsmRunResult> = Vec::new();
        // `tokio::time::Instant` (vs `std::time::Instant`) so the
        // deadline check advances with `tokio::time::pause + advance`
        // under `#[tokio::test(start_paused = true)]` — otherwise the
        // mocked clock advances tokio::sleep but not the real Instant
        // the deadline was derived from, so paused-time tests can't
        // exercise the timeout branch.
        let deadline =
            tokio::time::Instant::now() + std::time::Duration::from_secs(wall_clock_secs);
        loop {
            tokio::time::sleep(std::time::Duration::from_secs(2)).await;
            // Snapshot so we can mutate `pending` inside the loop.
            let cycle: Vec<String> = pending.iter().cloned().collect();
            for id in cycle {
                let resp = self
                    .ssm
                    .get_command_invocation()
                    .command_id(&command_id)
                    .instance_id(&id)
                    .send()
                    .await;
                let invocation = match resp {
                    Ok(o) => o,
                    Err(_) => {
                        // InvocationDoesNotExist on the first cycle is
                        // expected. Skip and retry next cycle.
                        continue;
                    }
                };
                let status = invocation.status.clone();
                let terminal = matches!(
                    status,
                    Some(CommandInvocationStatus::Success)
                        | Some(CommandInvocationStatus::Failed)
                        | Some(CommandInvocationStatus::Cancelled)
                        | Some(CommandInvocationStatus::TimedOut)
                );
                if !terminal {
                    continue;
                }
                pending.remove(&id);
                completed.push(SsmRunResult {
                    instance_id: id,
                    status: status
                        .map(|s| s.as_str().to_string())
                        .unwrap_or_else(|| "?".into()),
                    exit_code: invocation.response_code,
                    stdout: invocation.standard_output_content.unwrap_or_default(),
                    stderr: invocation.standard_error_content.unwrap_or_default(),
                });
            }
            if pending.is_empty() || tokio::time::Instant::now() >= deadline {
                break;
            }
        }
        // Anything left in `pending` after the deadline gets a synthetic
        // "TimedOut(local)" row so the operator sees which instances
        // didn't finish in the wall-clock window.
        for id in pending {
            completed.push(SsmRunResult {
                instance_id: id,
                status: "TimedOut(local)".into(),
                exit_code: -1,
                stdout: String::new(),
                stderr:
                    "ebman wall-clock timeout — instance didn't reach a terminal status in time"
                        .into(),
            });
        }
        // Sort by instance id so output is deterministic across runs.
        completed.sort_by(|a, b| a.instance_id.cmp(&b.instance_id));
        Ok(completed)
    }

    /// Fetch the recent history for a single CloudWatch alarm. Returns
    /// rows newest-first (matches the SDK's default ordering). `kind`
    /// distinguishes StateUpdate / ConfigurationUpdate / Action so the
    /// renderer can colour or filter by entry type. `max_records` caps
    /// the page size — the SDK enforces a server-side max of 100, so
    /// callers wanting more would need to follow the `next_token`
    /// (deferred until anyone needs it).
    pub async fn fetch_alarm_history(
        &self,
        alarm_name: &str,
        max_records: i32,
    ) -> Result<Vec<AlarmHistoryEntry>> {
        let resp = self
            .cw
            .describe_alarm_history()
            .alarm_name(alarm_name)
            .max_records(max_records)
            .send()
            .await
            .wrap_err("DescribeAlarmHistory failed")?;
        let mut out = Vec::new();
        for item in resp.alarm_history_items.unwrap_or_default() {
            let at = item
                .timestamp
                .and_then(|ts| DateTime::<Utc>::from_timestamp(ts.secs(), ts.subsec_nanos()));
            let kind = item
                .history_item_type
                .map(|t| t.as_str().to_string())
                .unwrap_or_else(|| "?".into());
            let summary = item.history_summary.unwrap_or_default();
            out.push(AlarmHistoryEntry { at, kind, summary });
        }
        Ok(out)
    }

    /// Delete one or more CloudWatch alarms by name.
    pub async fn delete_alarms(&self, names: &[String]) -> Result<()> {
        if names.is_empty() {
            return Ok(());
        }
        let mut req = self.cw.delete_alarms();
        for n in names {
            req = req.alarm_names(n);
        }
        req.send().await.wrap_err("DeleteAlarms failed")?;
        Ok(())
    }

    /// Pull a handful of useful EB metrics for one env, from CloudWatch.
    /// Returns an empty Vec for queries the API filtered out.
    pub async fn fetch_env_metrics(
        &self,
        env_name: &str,
        range_secs: i64,
    ) -> Result<Vec<MetricSeries>> {
        use aws_sdk_cloudwatch::types::{Dimension, Metric, MetricDataQuery, MetricStat};

        let end = Utc::now();
        let start = end - chrono::Duration::seconds(range_secs);

        let dim = Dimension::builder()
            .name("EnvironmentName")
            .value(env_name)
            .build();

        let make_query = |id: &str, name: &str, stat: &str| -> MetricDataQuery {
            let metric = Metric::builder()
                .namespace("AWS/ElasticBeanstalk")
                .metric_name(name)
                .dimensions(dim.clone())
                .build();
            let ms = MetricStat::builder()
                .metric(metric)
                .period(60)
                .stat(stat)
                .build();
            MetricDataQuery::builder().id(id).metric_stat(ms).build()
        };

        let resp = self
            .cw
            .get_metric_data()
            .start_time(to_smithy(start))
            .end_time(to_smithy(end))
            .metric_data_queries(make_query("health", "EnvironmentHealth", "Maximum"))
            .metric_data_queries(make_query("req4xx", "ApplicationRequests4xx", "Sum"))
            .metric_data_queries(make_query("req5xx", "ApplicationRequests5xx", "Sum"))
            .metric_data_queries(make_query("p90", "ApplicationLatencyP90", "Average"))
            .send()
            .await?;

        let order = ["health", "req4xx", "req5xx", "p90"];
        let labels: std::collections::HashMap<&str, (&str, &str)> = [
            ("health", ("Env Health (0–25)", "score")),
            ("req4xx", ("4xx Requests / min", "count")),
            ("req5xx", ("5xx Requests / min", "count")),
            ("p90", ("Latency P90", "s")),
        ]
        .into_iter()
        .collect();

        let mut by_id: std::collections::HashMap<String, MetricSeries> =
            std::collections::HashMap::new();
        for r in resp.metric_data_results.unwrap_or_default() {
            let id = r.id.unwrap_or_default();
            let display = labels
                .get(id.as_str())
                .copied()
                .map(|(d, _)| d.to_string())
                .unwrap_or_else(|| id.clone());
            let timestamps = r.timestamps.unwrap_or_default();
            let values = r.values.unwrap_or_default();
            let mut points: Vec<(DateTime<Utc>, f64)> = timestamps
                .iter()
                .zip(values.iter())
                .filter_map(|(ts, v)| {
                    DateTime::<Utc>::from_timestamp(ts.secs(), ts.subsec_nanos()).map(|t| (t, *v))
                })
                .collect();
            points.sort_by_key(|(t, _)| *t);
            by_id.insert(
                id.clone(),
                MetricSeries {
                    id,
                    label: display,
                    points,
                },
            );
        }

        Ok(order.iter().filter_map(|id| by_id.remove(*id)).collect())
    }

    /// Fetch user-defined metric series for one env. Each spec is
    /// `(label, namespace, name, stat, dimensions)` — `dimensions` are
    /// explicit overrides; when empty the call falls back to the env-scoped
    /// `EnvironmentName=env_name` dimension (the common case for
    /// `AWS/ElasticBeanstalk` metrics). Returns the series in the same
    /// order as `specs` so operators see their additions in add-order.
    pub async fn fetch_custom_env_metrics(
        &self,
        env_name: &str,
        range_secs: i64,
        specs: &[CustomMetricQuery],
    ) -> Result<Vec<MetricSeries>> {
        use aws_sdk_cloudwatch::types::{Dimension, Metric, MetricDataQuery, MetricStat};
        if specs.is_empty() {
            return Ok(Vec::new());
        }
        let end = Utc::now();
        let start = end - chrono::Duration::seconds(range_secs);

        let mut req = self
            .cw
            .get_metric_data()
            .start_time(to_smithy(start))
            .end_time(to_smithy(end));
        // CloudWatch's GetMetricData requires the `id` field to be a valid
        // metric reference (lowercase alpha + numeric + underscore, starts
        // with a letter). We use `m{i}` to dodge label-vs-id concerns.
        let mut id_to_label: std::collections::HashMap<String, String> =
            std::collections::HashMap::new();
        for (i, (label, namespace, name, stat, dims)) in specs.iter().enumerate() {
            let id = format!("m{i}");
            let mut metric_builder = Metric::builder().namespace(namespace).metric_name(name);
            if dims.is_empty() {
                metric_builder = metric_builder.dimensions(
                    Dimension::builder()
                        .name("EnvironmentName")
                        .value(env_name)
                        .build(),
                );
            } else {
                for (k, v) in dims {
                    metric_builder =
                        metric_builder.dimensions(Dimension::builder().name(k).value(v).build());
                }
            }
            let ms = MetricStat::builder()
                .metric(metric_builder.build())
                .period(60)
                .stat(stat)
                .build();
            id_to_label.insert(id.clone(), label.clone());
            req =
                req.metric_data_queries(MetricDataQuery::builder().id(id).metric_stat(ms).build());
        }

        let resp = req.send().await?;
        let mut by_id: std::collections::HashMap<String, MetricSeries> =
            std::collections::HashMap::new();
        for r in resp.metric_data_results.unwrap_or_default() {
            let id = r.id.unwrap_or_default();
            let label = id_to_label.get(&id).cloned().unwrap_or_else(|| id.clone());
            let timestamps = r.timestamps.unwrap_or_default();
            let values = r.values.unwrap_or_default();
            let mut points: Vec<(DateTime<Utc>, f64)> = timestamps
                .iter()
                .zip(values.iter())
                .filter_map(|(ts, v)| {
                    DateTime::<Utc>::from_timestamp(ts.secs(), ts.subsec_nanos()).map(|t| (t, *v))
                })
                .collect();
            points.sort_by_key(|(t, _)| *t);
            by_id.insert(id.clone(), MetricSeries { id, label, points });
        }
        // Return in the spec order so operators see the charts in the order
        // they added them.
        Ok((0..specs.len())
            .filter_map(|i| by_id.remove(&format!("m{i}")))
            .collect())
    }

    pub async fn purge_queue(&self, queue_url: &str) -> Result<()> {
        self.sqs.purge_queue().queue_url(queue_url).send().await?;
        Ok(())
    }

    pub async fn list_tags(&self, resource_arn: &str) -> Result<Vec<(String, String)>> {
        let resp = self
            .client
            .list_tags_for_resource()
            .resource_arn(resource_arn)
            .send()
            .await?;
        let tags = resp
            .resource_tags
            .unwrap_or_default()
            .into_iter()
            .filter_map(|t| match (t.key, t.value) {
                (Some(k), Some(v)) => Some((k, v)),
                _ => None,
            })
            .collect();
        Ok(tags)
    }

    /// UpdateTagsForResource — add/update tags listed in `to_add` and remove
    /// keys listed in `to_remove`. Empty lists are allowed but at least one
    /// side must be non-empty (the API rejects no-op calls).
    pub async fn update_tags(
        &self,
        resource_arn: &str,
        to_add: &[(String, String)],
        to_remove: &[String],
    ) -> Result<()> {
        use aws_sdk_elasticbeanstalk::types::Tag;
        let mut req = self
            .client
            .update_tags_for_resource()
            .resource_arn(resource_arn);
        for (k, v) in to_add {
            req = req.tags_to_add(Tag::builder().key(k).value(v).build());
        }
        for k in to_remove {
            req = req.tags_to_remove(k);
        }
        req.send().await?;
        Ok(())
    }

    pub async fn rebuild_env(&self, env_name: &str) -> Result<()> {
        self.client
            .rebuild_environment()
            .environment_name(env_name)
            .send()
            .await?;
        Ok(())
    }

    pub async fn restart_app_server(&self, env_name: &str) -> Result<()> {
        self.client
            .restart_app_server()
            .environment_name(env_name)
            .send()
            .await?;
        Ok(())
    }

    pub async fn swap_cnames(&self, source: &str, dest: &str) -> Result<()> {
        self.client
            .swap_environment_cnames()
            .source_environment_name(source)
            .destination_environment_name(dest)
            .send()
            .await?;
        Ok(())
    }

    /// Snapshot an env's current configuration as a named template under the
    /// same application. Idempotent for the user — if a template with the
    /// same name already exists, the API returns an error which we surface.
    pub async fn create_config_template(
        &self,
        application_name: &str,
        template_name: &str,
        source_env_name: &str,
    ) -> Result<()> {
        self.client
            .create_configuration_template()
            .application_name(application_name)
            .template_name(template_name)
            .environment_id(source_env_name)
            .send()
            .await
            .wrap_err("CreateConfigurationTemplate failed")?;
        Ok(())
    }

    /// Delete a configuration template by name. AWS will refuse if the
    /// template is currently in use; we pass the error back unchanged.
    pub async fn delete_config_template(
        &self,
        application_name: &str,
        template_name: &str,
    ) -> Result<()> {
        self.client
            .delete_configuration_template()
            .application_name(application_name)
            .template_name(template_name)
            .send()
            .await
            .wrap_err("DeleteConfigurationTemplate failed")?;
        Ok(())
    }

    /// List the newer platform versions in the same branch family as the
    /// env's current platform. Filtered server-side to `Ready` platforms;
    /// branch matching is best-effort using the current ARN's branch suffix
    /// (e.g. `Tomcat 9 with Corretto 17`). Sorted newest version first.
    pub async fn list_compatible_platforms(&self, env_name: &str) -> Result<Vec<CustomPlatform>> {
        use aws_sdk_elasticbeanstalk::types::{PlatformFilter, PlatformStatus};
        // Read the env's current platform ARN.
        let desc = self
            .client
            .describe_environments()
            .environment_names(env_name)
            .send()
            .await
            .wrap_err("DescribeEnvironments failed")?;
        let env = desc
            .environments
            .unwrap_or_default()
            .into_iter()
            .next()
            .ok_or_else(|| eyre!("env '{env_name}' not found"))?;
        let current_arn = env.platform_arn.clone().unwrap_or_default();
        let stack_or_arn = env
            .solution_stack_name
            .clone()
            .unwrap_or_else(|| current_arn.clone());
        let branch = platform_branch_from(&stack_or_arn);
        let owner_filter = PlatformFilter::builder()
            .r#type("PlatformStatus")
            .operator("=")
            .values(PlatformStatus::Ready.as_str())
            .build();
        let mut filters = vec![owner_filter];
        if !branch.is_empty() {
            filters.push(
                PlatformFilter::builder()
                    .r#type("PlatformBranchName")
                    .operator("=")
                    .values(branch.clone())
                    .build(),
            );
        }
        let mut next_token: Option<String> = None;
        let mut out: Vec<CustomPlatform> = Vec::new();
        loop {
            let mut req = self.client.list_platform_versions();
            for f in &filters {
                req = req.filters(f.clone());
            }
            if let Some(t) = next_token.clone() {
                req = req.next_token(t);
            }
            let resp = req.send().await.wrap_err("ListPlatformVersions failed")?;
            for p in resp.platform_summary_list.unwrap_or_default() {
                out.push(CustomPlatform {
                    arn: p.platform_arn.unwrap_or_default(),
                    branch: p.platform_branch_name.unwrap_or_default(),
                    version: p.platform_version.unwrap_or_default(),
                    status: p
                        .platform_status
                        .map(|s| s.as_str().to_string())
                        .unwrap_or_default(),
                    lifecycle: p.platform_lifecycle_state.unwrap_or_default(),
                });
            }
            match resp.next_token {
                Some(t) if !t.is_empty() => next_token = Some(t),
                _ => break,
            }
        }
        // Sort newest-first by semver-ish version.
        out.sort_by(|a, b| compare_versions(&b.version, &a.version));
        Ok(out)
    }

    /// Migrate the env to a new platform ARN via UpdateEnvironment. EB
    /// performs this as a rolling update; the API returns immediately and
    /// the event log carries progress.
    pub async fn upgrade_platform(&self, env_name: &str, platform_arn: &str) -> Result<()> {
        self.client
            .update_environment()
            .environment_name(env_name)
            .platform_arn(platform_arn)
            .send()
            .await
            .wrap_err("UpdateEnvironment(platform_arn) failed")?;
        Ok(())
    }

    /// Clone an env: snapshot the source's settings into a transient
    /// configuration template, spin up a new env from it, then clean the
    /// template up. The new env starts the usual EB launch process — the
    /// caller can monitor via DescribeEvents.
    pub async fn clone_env(&self, source_env_name: &str, target_env_name: &str) -> Result<()> {
        // Snapshot the source env's application + ID.
        let desc = self
            .client
            .describe_environments()
            .environment_names(source_env_name)
            .send()
            .await
            .wrap_err("DescribeEnvironments failed")?;
        let env = desc
            .environments
            .unwrap_or_default()
            .into_iter()
            .next()
            .ok_or_else(|| eyre!("source env '{source_env_name}' not found"))?;
        let application = env
            .application_name
            .ok_or_else(|| eyre!("source env has no application_name"))?;
        let env_id = env
            .environment_id
            .ok_or_else(|| eyre!("source env has no environment_id"))?;
        // Use a transient template name so we can clean it up even if the
        // create fails partway.
        let template = format!(
            "__ebman-clone-{}-{}",
            target_env_name,
            chrono::Utc::now().timestamp()
        );
        self.client
            .create_configuration_template()
            .application_name(&application)
            .template_name(&template)
            .environment_id(&env_id)
            .send()
            .await
            .wrap_err("CreateConfigurationTemplate failed")?;
        // Best-effort cleanup even if create_environment fails — we don't
        // want to leave debris.
        let create_result = self
            .client
            .create_environment()
            .application_name(&application)
            .environment_name(target_env_name)
            .template_name(&template)
            .send()
            .await;
        let _ = self
            .client
            .delete_configuration_template()
            .application_name(&application)
            .template_name(&template)
            .send()
            .await;
        create_result.wrap_err("CreateEnvironment failed")?;
        Ok(())
    }

    /// Set the env's `aws:autoscaling:asg:{MinSize,MaxSize}` so the ASG
    /// reaches `count` instances. Passing `Some(0)` is the "stop" pattern
    /// (no instances, env keeps its config). The API returns immediately;
    /// EB performs the scale as a rolling change.
    pub async fn scale_env(&self, env_name: &str, min: i32, max: i32) -> Result<()> {
        use aws_sdk_elasticbeanstalk::types::ConfigurationOptionSetting;
        let opts = vec![
            ConfigurationOptionSetting::builder()
                .namespace("aws:autoscaling:asg")
                .option_name("MinSize")
                .value(min.to_string())
                .build(),
            ConfigurationOptionSetting::builder()
                .namespace("aws:autoscaling:asg")
                .option_name("MaxSize")
                .value(max.to_string())
                .build(),
        ];
        self.client
            .update_environment()
            .environment_name(env_name)
            .set_option_settings(Some(opts))
            .send()
            .await
            .wrap_err("UpdateEnvironment(asg) failed")?;
        Ok(())
    }

    /// Terminate a single EC2 instance by ID. ASG (created by EB) re-launches
    /// a replacement automatically. The API returns immediately; the
    /// instance enters `shutting-down` and EB's events panel will surface
    /// the replacement within ~30 s.
    pub async fn terminate_instance(&self, instance_id: &str) -> Result<()> {
        self.ec2
            .terminate_instances()
            .instance_ids(instance_id)
            .send()
            .await
            .wrap_err("ec2:TerminateInstances failed")?;
        Ok(())
    }

    /// Stop an in-flight environment update. Useful to bail out of a hung
    /// deploy. No-op if EB sees no operation in progress.
    pub async fn abort_environment_update(&self, env_name: &str) -> Result<()> {
        self.client
            .abort_environment_update()
            .environment_name(env_name)
            .send()
            .await
            .wrap_err("AbortEnvironmentUpdate failed")?;
        Ok(())
    }

    /// List custom EB platforms in this account. Filters server-side via
    /// `PlatformOwner=self` so we only show platforms the caller built, not
    /// the AWS-managed ones. Returns the ARN, platform branch name, and
    /// lifecycle state per entry.
    pub async fn list_custom_platforms(&self) -> Result<Vec<CustomPlatform>> {
        use aws_sdk_elasticbeanstalk::types::PlatformFilter;
        let filter = PlatformFilter::builder()
            .r#type("PlatformOwner")
            .operator("=")
            .values("self")
            .build();
        let mut next_token: Option<String> = None;
        let mut out: Vec<CustomPlatform> = Vec::new();
        loop {
            let mut req = self.client.list_platform_versions().filters(filter.clone());
            if let Some(t) = next_token.clone() {
                req = req.next_token(t);
            }
            let resp = req.send().await.wrap_err("ListPlatformVersions failed")?;
            for p in resp.platform_summary_list.unwrap_or_default() {
                out.push(CustomPlatform {
                    arn: p.platform_arn.unwrap_or_default(),
                    branch: p.platform_branch_name.unwrap_or_default(),
                    version: p.platform_version.unwrap_or_default(),
                    status: p
                        .platform_status
                        .map(|s| s.as_str().to_string())
                        .unwrap_or_default(),
                    lifecycle: p.platform_lifecycle_state.unwrap_or_default(),
                });
            }
            match resp.next_token {
                Some(t) if !t.is_empty() => next_token = Some(t),
                _ => break,
            }
        }
        Ok(out)
    }

    /// Delete a custom platform by ARN. EB returns success immediately even
    /// though the underlying AMI / EBS cleanup runs async. Will fail if any
    /// envs are still using the platform.
    pub async fn delete_custom_platform(&self, platform_arn: &str) -> Result<()> {
        self.client
            .delete_platform_version()
            .platform_arn(platform_arn)
            .send()
            .await
            .wrap_err("DeletePlatformVersion failed")?;
        Ok(())
    }

    /// List application versions for `application_name`, sorted newest-first
    /// by `date_created`. Each entry carries the version label and the
    /// optional description text shown in the EB console. Pages through
    /// `next_token` so orgs with hundreds of historical versions see
    /// everything in `:versions` and `:rollback` can find labels that
    /// fall past the first page.
    pub async fn list_application_versions(
        &self,
        application_name: &str,
    ) -> Result<Vec<AppVersion>> {
        let mut out: Vec<AppVersion> = Vec::new();
        let mut next_token: Option<String> = None;
        loop {
            let mut req = self
                .client
                .describe_application_versions()
                .application_name(application_name);
            if let Some(t) = next_token.take() {
                req = req.next_token(t);
            }
            let resp = req
                .send()
                .await
                .wrap_err("DescribeApplicationVersions failed")?;
            for v in resp.application_versions.unwrap_or_default() {
                out.push(AppVersion {
                    label: v.version_label.unwrap_or_default(),
                    description: v.description.unwrap_or_default(),
                    created: v
                        .date_created
                        .and_then(|d| DateTime::from_timestamp(d.secs(), d.subsec_nanos())),
                });
            }
            match resp.next_token {
                Some(t) if !t.is_empty() => next_token = Some(t),
                _ => break,
            }
        }
        out.sort_by_key(|v| std::cmp::Reverse(v.created));
        Ok(out)
    }

    /// Delete an application version by label. `delete_source_bundle = true`
    /// also removes the underlying `.zip` from S3 so the storage cost goes
    /// away. EB rejects the call if the version is currently deployed to any
    /// env — surfaced as `SourceBundleDeletionException` /
    /// `OperationInProgressException` in the error chain.
    pub async fn delete_application_version(
        &self,
        application_name: &str,
        version_label: &str,
        delete_source_bundle: bool,
    ) -> Result<()> {
        self.client
            .delete_application_version()
            .application_name(application_name)
            .version_label(version_label)
            .delete_source_bundle(delete_source_bundle)
            .send()
            .await
            .wrap_err("DeleteApplicationVersion failed")?;
        Ok(())
    }

    /// Deploy a specific application-version label to an existing env via
    /// Ask EB for its managed S3 bucket — same bucket EB uses for its own
    /// uploads. We push application bundles into a known prefix here so
    /// `CreateApplicationVersion` can reference an `S3Location`. EB
    /// auto-creates the bucket on first call; subsequent calls return the
    /// same name.
    pub async fn create_storage_location(&self) -> Result<String> {
        let resp = self
            .client
            .create_storage_location()
            .send()
            .await
            .wrap_err("CreateStorageLocation failed")?;
        resp.s3_bucket
            .ok_or_else(|| eyre!("CreateStorageLocation returned no S3Bucket"))
    }

    /// Upload an application bundle from disk to S3. Bundles below
    /// [`MULTIPART_THRESHOLD`] use a single streaming `PutObject` so RAM
    /// stays flat regardless of file size; larger bundles use multipart
    /// upload in [`MULTIPART_PART_SIZE`] chunks, lifting the single-call
    /// 5 GiB ceiling and bounding peak RAM at one part. On any failure
    /// during a multipart upload we issue `AbortMultipartUpload` so S3
    /// reclaims the partial parts rather than billing for orphans.
    pub async fn upload_bundle(
        &self,
        bucket: &str,
        key: &str,
        path: &std::path::Path,
    ) -> Result<()> {
        self.upload_bundle_with(bucket, key, path, MULTIPART_THRESHOLD, MULTIPART_PART_SIZE)
            .await
    }

    /// Same as [`upload_bundle`] but lets the caller pin the threshold
    /// and part size. Intended for tests; production code calls
    /// `upload_bundle` which fixes both at module-level constants.
    pub async fn upload_bundle_with(
        &self,
        bucket: &str,
        key: &str,
        path: &std::path::Path,
        multipart_threshold: u64,
        part_size: u64,
    ) -> Result<()> {
        use aws_sdk_s3::primitives::ByteStream;
        let metadata = tokio::fs::metadata(path)
            .await
            .wrap_err_with(|| format!("stat bundle {}", path.display()))?;
        let size = metadata.len();
        if !should_multipart(size, multipart_threshold) {
            // Single PutObject. `ByteStream::from_path` streams from disk
            // in the SDK's default chunk size — no Vec<u8> of the whole
            // file is allocated.
            let body = ByteStream::from_path(path)
                .await
                .wrap_err_with(|| format!("read {}", path.display()))?;
            self.s3
                .put_object()
                .bucket(bucket)
                .key(key)
                .body(body)
                .send()
                .await
                .wrap_err_with(|| format!("S3 PutObject {bucket}/{key} failed"))?;
            return Ok(());
        }
        // Multipart path.
        let create = self
            .s3
            .create_multipart_upload()
            .bucket(bucket)
            .key(key)
            .send()
            .await
            .wrap_err_with(|| format!("S3 CreateMultipartUpload {bucket}/{key} failed"))?;
        let upload_id = create
            .upload_id()
            .ok_or_else(|| eyre!("CreateMultipartUpload returned no UploadId"))?
            .to_string();

        // Per-part upload. We read each chunk into a Vec<u8> sized to
        // the part — RAM = one part, regardless of file size. On any
        // failure mid-loop we abort the upload so S3 doesn't accumulate
        // orphaned parts.
        let plan = plan_part_lengths(size, part_size);
        let mut completed_parts: Vec<aws_sdk_s3::types::CompletedPart> =
            Vec::with_capacity(plan.len());
        // tokio::fs::File implements AsyncReadExt; we read exact chunks.
        use tokio::io::AsyncReadExt;
        let mut file = match tokio::fs::File::open(path).await {
            Ok(f) => f,
            Err(e) => {
                // Best-effort abort — propagate the original open error.
                let _ = self
                    .s3
                    .abort_multipart_upload()
                    .bucket(bucket)
                    .key(key)
                    .upload_id(&upload_id)
                    .send()
                    .await;
                return Err(eyre!("open {} for multipart upload: {e}", path.display()));
            }
        };
        for (idx, part_len) in plan.iter().enumerate() {
            let part_number = (idx + 1) as i32;
            let mut buf = vec![0u8; *part_len as usize];
            if let Err(e) = file.read_exact(&mut buf).await {
                let _ = self
                    .s3
                    .abort_multipart_upload()
                    .bucket(bucket)
                    .key(key)
                    .upload_id(&upload_id)
                    .send()
                    .await;
                return Err(eyre!(
                    "read part {part_number} from {}: {e}",
                    path.display()
                ));
            }
            let resp = match self
                .s3
                .upload_part()
                .bucket(bucket)
                .key(key)
                .upload_id(&upload_id)
                .part_number(part_number)
                .body(ByteStream::from(buf))
                .send()
                .await
            {
                Ok(r) => r,
                Err(e) => {
                    let _ = self
                        .s3
                        .abort_multipart_upload()
                        .bucket(bucket)
                        .key(key)
                        .upload_id(&upload_id)
                        .send()
                        .await;
                    return Err(e).wrap_err_with(|| {
                        format!("S3 UploadPart {part_number} of {bucket}/{key} failed")
                    });
                }
            };
            let e_tag = resp
                .e_tag()
                .ok_or_else(|| eyre!("UploadPart {part_number} returned no ETag"))?
                .to_string();
            completed_parts.push(
                aws_sdk_s3::types::CompletedPart::builder()
                    .part_number(part_number)
                    .e_tag(e_tag)
                    .build(),
            );
        }

        let completed = aws_sdk_s3::types::CompletedMultipartUpload::builder()
            .set_parts(Some(completed_parts))
            .build();
        if let Err(e) = self
            .s3
            .complete_multipart_upload()
            .bucket(bucket)
            .key(key)
            .upload_id(&upload_id)
            .multipart_upload(completed)
            .send()
            .await
        {
            let _ = self
                .s3
                .abort_multipart_upload()
                .bucket(bucket)
                .key(key)
                .upload_id(&upload_id)
                .send()
                .await;
            return Err(e)
                .wrap_err_with(|| format!("S3 CompleteMultipartUpload {bucket}/{key} failed"));
        }
        Ok(())
    }

    /// Register a new application version pointing at an S3 source bundle.
    /// `auto_create_app` is `false` because we only create versions for
    /// existing applications; the env's application is the source of truth.
    pub async fn create_app_version(
        &self,
        application_name: &str,
        version_label: &str,
        description: Option<&str>,
        s3_bucket: &str,
        s3_key: &str,
    ) -> Result<()> {
        use aws_sdk_elasticbeanstalk::types::S3Location;
        let source = S3Location::builder()
            .s3_bucket(s3_bucket)
            .s3_key(s3_key)
            .build();
        let mut req = self
            .client
            .create_application_version()
            .application_name(application_name)
            .version_label(version_label)
            .source_bundle(source)
            .auto_create_application(false);
        if let Some(d) = description {
            req = req.description(d);
        }
        req.send()
            .await
            .wrap_err("CreateApplicationVersion failed")?;
        Ok(())
    }

    /// `UpdateEnvironment(version_label)`. Returns immediately — the env
    /// will mutate in the background.
    pub async fn deploy_version(&self, env_name: &str, version_label: &str) -> Result<()> {
        self.client
            .update_environment()
            .environment_name(env_name)
            .version_label(version_label)
            .send()
            .await
            .wrap_err("UpdateEnvironment(version_label) failed")?;
        Ok(())
    }

    /// Fetch the option settings stored in a saved configuration template.
    /// Returns a sorted `(namespace, option_name, value)` vector — sort makes
    /// the overlay output stable and diffable across runs. Empty values are
    /// preserved (operators sometimes care that a setting is explicitly
    /// empty vs. unset; the call only returns settings the template actually
    /// defines, so "missing" already means "use platform default").
    pub async fn describe_template_settings(
        &self,
        application_name: &str,
        template_name: &str,
    ) -> Result<Vec<(String, String, String)>> {
        let resp = self
            .client
            .describe_configuration_settings()
            .application_name(application_name)
            .template_name(template_name)
            .send()
            .await
            .wrap_err("DescribeConfigurationSettings(template) failed")?;
        let mut out: Vec<(String, String, String)> = resp
            .configuration_settings
            .unwrap_or_default()
            .into_iter()
            .flat_map(|c| c.option_settings.unwrap_or_default())
            .map(|o| {
                (
                    o.namespace.unwrap_or_default(),
                    o.option_name.unwrap_or_default(),
                    o.value.unwrap_or_default(),
                )
            })
            .collect();
        out.sort();
        Ok(out)
    }

    /// Apply a saved configuration template to an existing env via
    /// `UpdateEnvironment(template_name)`. The env will start mutating in
    /// the background; surface the launch via the events panel.
    pub async fn apply_config_template(&self, env_name: &str, template_name: &str) -> Result<()> {
        self.client
            .update_environment()
            .environment_name(env_name)
            .template_name(template_name)
            .send()
            .await
            .wrap_err("UpdateEnvironment(template_name) failed")?;
        Ok(())
    }

    pub async fn terminate_env(&self, env_name: &str) -> Result<()> {
        self.client
            .terminate_environment()
            .environment_name(env_name)
            .send()
            .await?;
        Ok(())
    }

    /// Ask EB to start collecting the tail log for an env. Per-instance log
    /// snapshots become available via `retrieve_env_info` once each instance
    /// has uploaded its sample to S3 (usually 5-15 seconds).
    pub async fn request_env_info_tail(&self, env_name: &str) -> Result<()> {
        use aws_sdk_elasticbeanstalk::types::EnvironmentInfoType;
        self.client
            .request_environment_info()
            .environment_name(env_name)
            .info_type(EnvironmentInfoType::Tail)
            .send()
            .await
            .wrap_err("RequestEnvironmentInfo failed")?;
        Ok(())
    }

    /// Read whatever tail-log samples EB has on file for the env, mapped to
    /// pre-signed S3 URLs. Empty vec means no samples have been uploaded yet —
    /// poll again. Each entry is `(ec2_instance_id, pre_signed_url)`.
    pub async fn retrieve_env_info_tail(&self, env_name: &str) -> Result<Vec<(String, String)>> {
        use aws_sdk_elasticbeanstalk::types::EnvironmentInfoType;
        let resp = self
            .client
            .retrieve_environment_info()
            .environment_name(env_name)
            .info_type(EnvironmentInfoType::Tail)
            .send()
            .await
            .wrap_err("RetrieveEnvironmentInfo failed")?;
        let mut out = Vec::new();
        for info in resp.environment_info.unwrap_or_default() {
            if let (Some(id), Some(url)) = (info.ec2_instance_id, info.message) {
                out.push((id, url));
            }
        }
        Ok(out)
    }

    /// Fetch the body of a pre-signed S3 URL. Shells out to `curl` so we don't
    /// pull in an HTTP-client dep; pre-signed URLs are plain HTTPS GETs with
    /// no auth headers, which curl handles trivially. 15 s cap per fetch.
    pub async fn fetch_url_text(url: &str) -> Result<String> {
        use tokio::process::Command;
        let out = Command::new("curl")
            .args([
                "-s",
                "-S",
                "--fail-with-body",
                "--max-time",
                "15",
                "--no-buffer",
            ])
            .arg(url)
            .output()
            .await
            .wrap_err("could not invoke curl (is it installed?)")?;
        if !out.status.success() {
            return Err(eyre!(
                "curl exit {}: {}",
                out.status,
                String::from_utf8_lossy(&out.stderr).trim()
            ));
        }
        Ok(String::from_utf8_lossy(&out.stdout).into_owned())
    }

    /// `DescribeEnvironmentHealth` summarised down to a `(healthy, total)`
    /// pair for the INST column on the main table. Lightweight compared to
    /// `DescribeInstancesHealth` (one call returns aggregated counts; no
    /// per-instance attributes). Fanned across every env on each refresh
    /// tick — typical accounts have ≤ 50 envs which is well under the
    /// EB API's per-second budget.
    pub async fn fetch_env_instance_counts(&self, env_name: &str) -> Result<EnvInstanceCounts> {
        let resp = self
            .client
            .describe_environment_health()
            .environment_name(env_name)
            .attribute_names(
                aws_sdk_elasticbeanstalk::types::EnvironmentHealthAttribute::InstancesHealth,
            )
            .send()
            .await
            .wrap_err("DescribeEnvironmentHealth failed")?;
        Ok(summarise_instance_health(resp.instances_health.as_ref()))
    }

    pub async fn list_instances(&self, env_name: &str) -> Result<Vec<Instance>> {
        let resp = self
            .client
            .describe_instances_health()
            .environment_name(env_name)
            .attribute_names(aws_sdk_elasticbeanstalk::types::InstancesHealthAttribute::All)
            .send()
            .await?;
        let instances = resp
            .instance_health_list
            .unwrap_or_default()
            .into_iter()
            .map(|i| Instance {
                id: i.instance_id.unwrap_or_default(),
                health: i.health_status.unwrap_or_default(),
                color: i.color.unwrap_or_default(),
                causes: i.causes.unwrap_or_default(),
                instance_type: i.instance_type.unwrap_or_default(),
                availability_zone: i.availability_zone.unwrap_or_default(),
                launched_at: i
                    .launched_at
                    .and_then(|d| DateTime::from_timestamp(d.secs(), d.subsec_nanos())),
            })
            .collect();
        Ok(instances)
    }

    /// `organizations:ListAccounts`, paginated. Returns every active +
    /// suspended account the active credentials can see (i.e. the
    /// caller is in the mgmt account or a delegated administrator).
    /// Surfaces the API's `AccessDenied` cleanly so the `:accounts`
    /// overlay can render a "no org access" hint rather than an opaque
    /// stack trace.
    pub async fn list_org_accounts(&self) -> Result<Vec<OrgAccount>> {
        let mut out: Vec<OrgAccount> = Vec::new();
        let mut next_token: Option<String> = None;
        loop {
            let mut req = self.org.list_accounts();
            if let Some(t) = next_token.take() {
                req = req.next_token(t);
            }
            let resp = req
                .send()
                .await
                .wrap_err("organizations:ListAccounts failed")?;
            for a in resp.accounts.unwrap_or_default() {
                out.push(OrgAccount {
                    id: a.id.unwrap_or_default(),
                    name: a.name.unwrap_or_default(),
                    email: a.email,
                    status: a.status.map(|s| s.as_str().to_string()).unwrap_or_default(),
                });
            }
            match resp.next_token {
                Some(t) if !t.is_empty() => next_token = Some(t),
                _ => break,
            }
        }
        // Stable display order: status (Active first), then name.
        out.sort_by(|a, b| {
            let sa = (a.status != "ACTIVE", a.name.to_lowercase());
            let sb = (b.status != "ACTIVE", b.name.to_lowercase());
            sa.cmp(&sb)
        });
        Ok(out)
    }

    pub async fn list_applications(&self) -> Result<Vec<Application>> {
        let resp = self.client.describe_applications().send().await?;
        let apps = resp
            .applications
            .unwrap_or_default()
            .into_iter()
            .map(|a| Application {
                name: a.application_name.unwrap_or_default(),
                description: a.description.unwrap_or_default(),
                date_created: a
                    .date_created
                    .and_then(|d| DateTime::from_timestamp(d.secs(), d.subsec_nanos())),
                date_updated: a
                    .date_updated
                    .and_then(|d| DateTime::from_timestamp(d.secs(), d.subsec_nanos())),
                version_count: a.versions.map(|v| v.len()).unwrap_or(0),
                templates: a.configuration_templates.unwrap_or_default(),
                // Filled in by a follow-up `list_application_versions` fan-out.
                latest_version_label: None,
                latest_version_created: None,
            })
            .collect();
        Ok(apps)
    }

    pub async fn list_environments(&self) -> Result<Vec<Environment>> {
        let mut all = Vec::new();
        let mut next_token: Option<String> = None;
        loop {
            let mut req = self.client.describe_environments().include_deleted(false);
            if let Some(t) = next_token.take() {
                req = req.next_token(t);
            }
            let resp = req.send().await.wrap_err("DescribeEnvironments failed")?;
            if let Some(envs) = resp.environments {
                all.extend(envs.into_iter().map(map_env));
            }
            match resp.next_token {
                Some(t) if !t.is_empty() => next_token = Some(t),
                _ => break,
            }
        }
        Ok(all)
    }

    /// Flat list of every solution-stack name available in this region
    /// (`ListAvailableSolutionStacks`). Drives the stale-platform check:
    /// an env whose stack has a lower version than the newest stack in
    /// the same family is flagged in the table.
    pub async fn list_solution_stacks(&self) -> Result<Vec<String>> {
        let resp = self
            .client
            .list_available_solution_stacks()
            .send()
            .await
            .wrap_err("ListAvailableSolutionStacks failed")?;
        Ok(resp.solution_stacks.unwrap_or_default())
    }
}

fn map_env(e: aws_sdk_elasticbeanstalk::types::EnvironmentDescription) -> Environment {
    let solution_stack = e.solution_stack_name.clone().unwrap_or_default();
    let raw_platform = e
        .solution_stack_name
        .clone()
        .or(e.platform_arn.clone())
        .unwrap_or_default();
    let tier = e
        .tier
        .as_ref()
        .and_then(|t| t.name.as_deref())
        .map(normalize_tier)
        .unwrap_or_else(|| "?".into());
    Environment {
        name: e.environment_name.unwrap_or_default(),
        application: e.application_name.unwrap_or_default(),
        status: e
            .status
            .map(|s| s.as_str().to_string())
            .unwrap_or_else(|| "-".into()),
        health: e
            .health
            .map(|h| h.as_str().to_string())
            .unwrap_or_else(|| "-".into()),
        platform: platform_family(&raw_platform),
        solution_stack,
        tier,
        cname: e.cname.unwrap_or_default(),
        version_label: e.version_label.unwrap_or_default(),
        arn: e.environment_arn,
        updated: e
            .date_updated
            .and_then(|d| DateTime::from_timestamp(d.secs(), d.subsec_nanos())),
        id: e.environment_id,
        region: None,
    }
}

/// Fan-out helper: build a transient `AwsClient` for `region` (sharing the
/// caller's profile) and pull `DescribeEnvironments` from there. Each
/// returned env has `region` stamped so the table can sort / group on it.
/// Best-effort extraction of the EB platform branch name from a solution
/// stack name or platform ARN. The names look like `64bit Amazon Linux 2023
/// v4.5.2 running Tomcat 9 Corretto 17` — we keep the "running …" tail and
/// strip any leading "running " marker. ARNs follow a separate scheme and
/// already carry the branch in their path.
fn platform_branch_from(stack_or_arn: &str) -> String {
    if let Some(rest) = stack_or_arn.split(" running ").nth(1) {
        return rest.trim().to_string();
    }
    if stack_or_arn.starts_with("arn:") {
        // Branch is the second-to-last path segment.
        let parts: Vec<&str> = stack_or_arn.split('/').collect();
        if parts.len() >= 2 {
            return parts[parts.len() - 2].to_string();
        }
    }
    String::new()
}

/// Compare two dotted version strings semver-ish. Numeric tokens compared
/// numerically; non-numeric tails fall back to string comparison. Returns
/// `Ordering` so this can drive `sort_by`.
fn compare_versions(a: &str, b: &str) -> std::cmp::Ordering {
    use std::cmp::Ordering;
    let parse = |s: &str| {
        s.split('.')
            .map(|p| p.split('-').next().unwrap_or(p).parse::<u64>().ok())
            .collect::<Vec<_>>()
    };
    let av = parse(a);
    let bv = parse(b);
    for i in 0..av.len().max(bv.len()) {
        let aa = av.get(i).and_then(|x| *x);
        let bb = bv.get(i).and_then(|x| *x);
        match (aa, bb) {
            (Some(x), Some(y)) => match x.cmp(&y) {
                Ordering::Equal => continue,
                o => return o,
            },
            (Some(_), None) => return Ordering::Greater,
            (None, Some(_)) => return Ordering::Less,
            (None, None) => break,
        }
    }
    a.cmp(b)
}

/// Pure: roll up EB's per-bucket `InstanceHealthSummary` into the
/// `(healthy, total)` shape the INST column wants. `healthy` is `ok +
/// info` (both Green per EB's docs — Info just means an operation is in
/// progress on an otherwise-healthy instance, not a problem signal).
/// `total` is the sum across every bucket including Grey buckets like
/// `no_data` / `unknown` / `pending` so an env that's mid-launch
/// reports `0/N` rather than `0/0`. Missing input (`None`) and
/// all-None buckets render as `EnvInstanceCounts::default()` (0/0).
pub fn summarise_instance_health(
    summary: Option<&aws_sdk_elasticbeanstalk::types::InstanceHealthSummary>,
) -> EnvInstanceCounts {
    let Some(s) = summary else {
        return EnvInstanceCounts::default();
    };
    let g = |v: Option<i32>| v.unwrap_or(0);
    let ok = g(s.ok);
    let info = g(s.info);
    let healthy = ok + info;
    let total = g(s.no_data)
        + g(s.unknown)
        + g(s.pending)
        + ok
        + info
        + g(s.warning)
        + g(s.degraded)
        + g(s.severe);
    EnvInstanceCounts { healthy, total }
}

/// Pure: parse a time-window spec for `:logs-insights --window WINDOW`.
/// Accepts `<n><unit>` with unit `m` / `h` / `d` — e.g. `30m`, `6h`, `7d`.
/// Returns the window length in *milliseconds* so callers can subtract
/// from `Utc::now().timestamp_millis()` directly. Returns `None` for
/// malformed input or non-positive values so the caller surfaces a
/// usage error instead of silently substituting a wrong window. Same
/// grammar as `parse_replay_spec` in `mode_dlq.rs` — kept consistent
/// so operators only have to learn one time-window vocabulary.
pub fn parse_window_ms(input: &str) -> Option<i64> {
    let s = input.trim().to_lowercase();
    if s.is_empty() {
        return None;
    }
    let unit = s.chars().last()?;
    let num: i64 = s[..s.len() - unit.len_utf8()].parse().ok()?;
    if num <= 0 {
        return None;
    }
    let ms = match unit {
        'm' => num * 60_000,
        'h' => num * 60 * 60_000,
        'd' => num * 24 * 60 * 60_000,
        _ => return None,
    };
    Some(ms)
}

/// Pure: render an `InsightsResults` payload to a multi-line string
/// suitable for a TextOverlay body. Columns are field names from the
/// first row (Insights guarantees every row has the same field set in
/// the same order), each cell width-padded against the column max so
/// the result reads like a table. Long values are truncated to keep
/// the overlay readable. Empty input renders as a "no rows matched"
/// stub plus the scan stats — same shape so the overlay never collapses.
pub fn format_insights_results(
    results: &InsightsResults,
    query: &str,
    log_groups: &[String],
) -> String {
    let mut out = String::new();
    out.push_str(&format!(
        "query: {query}\nlog groups: {}\nmatched: {} / scanned: {}\n",
        if log_groups.is_empty() {
            "(none)".to_string()
        } else {
            log_groups.join(", ")
        },
        results.records_matched,
        results.records_scanned,
    ));
    out.push_str(&"─".repeat(60));
    out.push('\n');
    if results.rows.is_empty() {
        out.push_str("(no rows matched the query)\n");
        return out;
    }
    // Skip the synthetic `@ptr` Insights field — it's a record locator
    // for the API to drill back to individual events, not useful in the
    // operator-facing overlay. Drop it from every row consistently.
    let headers: Vec<String> = results.rows[0]
        .fields
        .iter()
        .map(|(k, _)| k.clone())
        .filter(|k| k != "@ptr")
        .collect();
    // Per-column max-width pass — bounded at 60 cells so a single huge
    // message field doesn't push every other column off-screen.
    const COL_MAX: usize = 60;
    let mut widths: Vec<usize> = headers.iter().map(|h| h.len()).collect();
    for row in &results.rows {
        for (i, h) in headers.iter().enumerate() {
            if let Some((_, v)) = row.fields.iter().find(|(k, _)| k == h) {
                let cells = v.chars().count().min(COL_MAX);
                if cells > widths[i] {
                    widths[i] = cells;
                }
            }
        }
    }
    // Header row.
    let mut header_line = String::new();
    for (i, h) in headers.iter().enumerate() {
        if i > 0 {
            header_line.push_str("  ");
        }
        header_line.push_str(&format!("{:<w$}", h, w = widths[i]));
    }
    out.push_str(&header_line);
    out.push('\n');
    // Separator.
    let mut sep_line = String::new();
    for (i, w) in widths.iter().enumerate() {
        if i > 0 {
            sep_line.push_str("  ");
        }
        sep_line.push_str(&"─".repeat(*w));
    }
    out.push_str(&sep_line);
    out.push('\n');
    // Data rows.
    for row in &results.rows {
        let mut line = String::new();
        for (i, h) in headers.iter().enumerate() {
            if i > 0 {
                line.push_str("  ");
            }
            let raw = row
                .fields
                .iter()
                .find(|(k, _)| k == h)
                .map(|(_, v)| v.as_str())
                .unwrap_or("");
            let trimmed: String = if raw.chars().count() > COL_MAX {
                let mut s: String = raw.chars().take(COL_MAX.saturating_sub(1)).collect();
                s.push('…');
                s
            } else {
                raw.to_string()
            };
            line.push_str(&format!("{:<w$}", trimmed, w = widths[i]));
        }
        out.push_str(&line);
        out.push('\n');
    }
    out
}

/// Switch to multipart upload when the bundle is at least this large.
/// 64 MiB is well above the SDK's default chunked-read window so the
/// streaming PutObject path handles "normal" bundles comfortably; above
/// this threshold the multipart path's recoverability (partial parts can
/// be retried) starts to matter, and the 5 GiB single-PutObject ceiling
/// looms.
pub const MULTIPART_THRESHOLD: u64 = 64 * 1024 * 1024;

/// Per-part chunk size for multipart uploads. S3's minimum part size is
/// 5 MiB (except the last part); 16 MiB gives us 320 GiB headroom under
/// the 10,000-part ceiling, well above S3's 5 TiB object cap.
pub const MULTIPART_PART_SIZE: u64 = 16 * 1024 * 1024;

/// Decide whether a bundle of `size` bytes should go through multipart.
/// Pure for tests; production code calls this with [`MULTIPART_THRESHOLD`].
pub fn should_multipart(size: u64, threshold: u64) -> bool {
    size >= threshold
}

/// Plan the per-part lengths for a multipart upload of a file of
/// `total_size` bytes using `part_size` bytes per part. The last part is
/// whatever's left (>= 1 byte, < part_size) unless `total_size` is an
/// exact multiple. Empty input (`total_size == 0`) yields an empty plan.
/// Pure — for tests and for the upload loop itself.
pub fn plan_part_lengths(total_size: u64, part_size: u64) -> Vec<u64> {
    if total_size == 0 || part_size == 0 {
        return Vec::new();
    }
    let full = total_size / part_size;
    let remainder = total_size % part_size;
    let mut out = Vec::with_capacity(full as usize + if remainder > 0 { 1 } else { 0 });
    for _ in 0..full {
        out.push(part_size);
    }
    if remainder > 0 {
        out.push(remainder);
    }
    out
}

/// Split a solution-stack name into `(family_key, version)`. The family key
/// is the stack name with its `vX.Y.Z` token removed and surrounding
/// whitespace collapsed, so two stacks that differ only in version share a
/// key (e.g. `64bit Amazon Linux 2023 v6.1.0 running Node.js 18` →
/// `("64bit Amazon Linux 2023 running Node.js 18", "6.1.0")`). Returns
/// `None` when no `vN.N…` token is present — platform-ARN / custom-platform
/// envs have no solution stack and so can't be version-compared.
pub fn stack_family_version(stack: &str) -> Option<(String, String)> {
    let version_token = stack.split_whitespace().find(|tok| {
        tok.strip_prefix('v')
            .map(|rest| {
                !rest.is_empty()
                    && rest
                        .split('.')
                        .all(|p| !p.is_empty() && p.chars().all(|c| c.is_ascii_digit()))
            })
            .unwrap_or(false)
    })?;
    let version = version_token.trim_start_matches('v').to_string();
    let key = stack
        .split_whitespace()
        .filter(|tok| *tok != version_token)
        .collect::<Vec<_>>()
        .join(" ");
    Some((key, version))
}

/// Build a `family_key → newest version` map from a flat
/// `ListAvailableSolutionStacks` listing. Stacks with no version token are
/// skipped.
pub fn latest_stack_versions(stacks: &[String]) -> std::collections::HashMap<String, String> {
    let mut out: std::collections::HashMap<String, String> = std::collections::HashMap::new();
    for s in stacks {
        if let Some((key, ver)) = stack_family_version(s) {
            match out.get(&key) {
                Some(cur) if compare_versions(&ver, cur) != std::cmp::Ordering::Greater => {}
                _ => {
                    out.insert(key, ver);
                }
            }
        }
    }
    out
}

/// If a strictly-newer version of `env_stack`'s platform family exists in
/// `latest`, return that version. `None` when the env is already current,
/// has no parseable stack, or its family isn't in the listing.
pub fn newer_stack_version(
    env_stack: &str,
    latest: &std::collections::HashMap<String, String>,
) -> Option<String> {
    let (key, ver) = stack_family_version(env_stack)?;
    let newest = latest.get(&key)?;
    if compare_versions(newest, &ver) == std::cmp::Ordering::Greater {
        Some(newest.clone())
    } else {
        None
    }
}

pub async fn list_environments_in_region(
    profile: Option<String>,
    region: String,
) -> Result<Vec<Environment>> {
    let client = AwsClient::with(profile, Some(region.clone())).await?;
    let mut envs = client.list_environments().await?;
    for e in &mut envs {
        e.region = Some(region.clone());
    }
    Ok(envs)
}

/// Sibling of `list_environments_in_region` for the AssumeRole path:
/// assumes into the named role, then lists envs. `region` overrides the
/// AccountSpec's region when supplied; otherwise the spec's own region
/// wins (or env default). Used by the multi-account fan-out in
/// `:org-health` / `:find-env`.
pub async fn list_environments_for_account(
    name: &str,
    spec: &crate::config::AccountSpec,
    region: Option<String>,
) -> Result<Vec<Environment>> {
    let mut spec = spec.clone();
    if region.is_some() {
        spec.region = region.clone();
    }
    let client = AwsClient::assume_role(name, &spec).await?;
    let resolved_region = client.context.region.clone();
    let mut envs = client.list_environments().await?;
    for e in &mut envs {
        e.region = Some(resolved_region.clone());
    }
    Ok(envs)
}

/// Pulls the family + version out of either a solution_stack_name like
/// "64bit Amazon Linux 2 v3.7.0 running Tomcat 9 Corretto 17"  → "Tomcat 9 Corretto 17"
/// or a platform_arn like
/// "arn:aws:elasticbeanstalk:us-east-1::platform/Java 17 running on 64bit Amazon Linux 2/3.5.0"
///   → "Java 17"
fn platform_family(raw: &str) -> String {
    if raw.is_empty() {
        return String::new();
    }
    // Platform ARN form: "...platform/Family X running on 64bit Amazon Linux/3.5.0"
    // The interesting segment lives between '/' separators and contains " running on ".
    if raw.contains(" running on ") {
        for seg in raw.split('/') {
            if let Some((family, _)) = seg.split_once(" running on ") {
                return family.trim().to_string();
            }
        }
    }
    // Solution-stack form: "...64bit Amazon Linux 2 v3.5.0 running Family X"
    if let Some((_, after)) = raw.rsplit_once(" running ") {
        return after.trim().to_string();
    }
    raw.to_string()
}

/// Convention-based DLQ derivation for EB-managed worker queues. EB names the
/// main queue `awseb-<env-id>-<random>` and the DLQ `awseb-<env-id>-<random>-dlq`.
/// If the main queue URL doesn't match the pattern, returns None and the caller
/// just shows no DLQ.
fn to_smithy(d: DateTime<Utc>) -> aws_sdk_cloudwatch::primitives::DateTime {
    aws_sdk_cloudwatch::primitives::DateTime::from_secs(d.timestamp())
}

/// Pure: split a comma-separated EB option-setting value into a clean
/// `Vec<String>`. Trims each entry and drops empties.
fn split_csv(value: &str) -> Vec<String> {
    value
        .split(',')
        .map(|s| s.trim().to_string())
        .filter(|s| !s.is_empty())
        .collect()
}

fn derive_dlq_url(main: &str) -> Option<String> {
    let trimmed = main.trim_end_matches('/');
    if trimmed.ends_with("-dlq") {
        return None;
    }
    Some(format!("{trimmed}-dlq"))
}

fn normalize_tier(name: &str) -> String {
    match name {
        "WebServer" => "Web".into(),
        "Worker" => "Worker".into(),
        other => other.to_string(),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn platform_family_from_solution_stack() {
        assert_eq!(
            platform_family("64bit Amazon Linux 2 v3.5.0 running Java 17"),
            "Java 17"
        );
        assert_eq!(
            platform_family("64bit Amazon Linux 2 v3.7.0 running Tomcat 9 Corretto 17"),
            "Tomcat 9 Corretto 17"
        );
        assert_eq!(
            platform_family("64bit Amazon Linux 2023 v6.1.0 running Node.js 18"),
            "Node.js 18"
        );
    }

    #[test]
    fn platform_family_from_arn() {
        assert_eq!(
            platform_family(
                "arn:aws:elasticbeanstalk:us-east-1::platform/Java 17 running on 64bit Amazon Linux 2/3.5.0"
            ),
            "Java 17"
        );
    }

    #[test]
    fn platform_family_handles_empty_and_unknown() {
        assert_eq!(platform_family(""), "");
        assert_eq!(platform_family("just a string"), "just a string");
    }

    #[test]
    fn stack_family_version_splits_solution_stack() {
        assert_eq!(
            stack_family_version("64bit Amazon Linux 2023 v6.1.0 running Node.js 18"),
            Some((
                "64bit Amazon Linux 2023 running Node.js 18".to_string(),
                "6.1.0".to_string()
            ))
        );
    }

    #[test]
    fn stack_family_version_rejects_versionless() {
        assert_eq!(stack_family_version(""), None);
        assert_eq!(stack_family_version("some platform with no version"), None);
        // A leading-v word that isn't a dotted number must not be mistaken
        // for the version token.
        assert_eq!(stack_family_version("running via vN stack"), None);
    }

    #[test]
    fn latest_stack_versions_keeps_newest_per_family() {
        let stacks = vec![
            "64bit Amazon Linux 2 v3.1.0 running Node.js 14".to_string(),
            "64bit Amazon Linux 2 v3.10.0 running Node.js 14".to_string(),
            "64bit Amazon Linux 2 v3.2.0 running Node.js 14".to_string(),
            "64bit Amazon Linux 2023 v6.1.0 running Node.js 18".to_string(),
        ];
        let latest = latest_stack_versions(&stacks);
        assert_eq!(
            latest.get("64bit Amazon Linux 2 running Node.js 14"),
            Some(&"3.10.0".to_string())
        );
        assert_eq!(
            latest.get("64bit Amazon Linux 2023 running Node.js 18"),
            Some(&"6.1.0".to_string())
        );
    }

    #[test]
    fn newer_stack_version_flags_only_superseded() {
        let latest = latest_stack_versions(&[
            "64bit Amazon Linux 2023 v6.1.0 running Node.js 18".to_string()
        ]);
        // Older patch → flagged.
        assert_eq!(
            newer_stack_version("64bit Amazon Linux 2023 v6.0.3 running Node.js 18", &latest),
            Some("6.1.0".to_string())
        );
        // Already current → not flagged.
        assert_eq!(
            newer_stack_version("64bit Amazon Linux 2023 v6.1.0 running Node.js 18", &latest),
            None
        );
        // Different family (Node 18 vs 20) → not flagged.
        assert_eq!(
            newer_stack_version("64bit Amazon Linux 2023 v1.0.0 running Node.js 20", &latest),
            None
        );
        // No parseable stack → not flagged.
        assert_eq!(newer_stack_version("", &latest), None);
    }

    #[test]
    fn normalize_tier_maps_known_names() {
        assert_eq!(normalize_tier("WebServer"), "Web");
        assert_eq!(normalize_tier("Worker"), "Worker");
        assert_eq!(normalize_tier("Other"), "Other");
    }

    #[test]
    fn derive_dlq_url_appends_suffix() {
        assert_eq!(
            derive_dlq_url("https://sqs.us-east-1.amazonaws.com/123/awseb-e-foo-queue"),
            Some("https://sqs.us-east-1.amazonaws.com/123/awseb-e-foo-queue-dlq".to_string())
        );
    }

    #[test]
    fn should_multipart_crosses_threshold() {
        assert!(!should_multipart(0, 64));
        assert!(!should_multipart(63, 64));
        assert!(should_multipart(64, 64));
        assert!(should_multipart(1_000_000, 64));
    }

    #[test]
    fn plan_part_lengths_exact_multiple() {
        // 48 bytes, 16-byte parts → three full parts, no remainder.
        assert_eq!(plan_part_lengths(48, 16), vec![16, 16, 16]);
    }

    #[test]
    fn plan_part_lengths_partial_last_part() {
        // 17 bytes, 8-byte parts → 8 + 8 + 1.
        assert_eq!(plan_part_lengths(17, 8), vec![8, 8, 1]);
    }

    #[test]
    fn plan_part_lengths_zero_and_under_one_part() {
        // Zero input yields no parts (no upload to make).
        assert!(plan_part_lengths(0, 16).is_empty());
        // File smaller than one part is still one part.
        assert_eq!(plan_part_lengths(5, 16), vec![5]);
        // Defensive: zero part_size yields no plan (caller guard).
        assert!(plan_part_lengths(100, 0).is_empty());
    }

    #[test]
    fn summarise_instance_health_rolls_up_buckets() {
        use aws_sdk_elasticbeanstalk::types::InstanceHealthSummary;

        // Mixed: 2 ok + 1 info = 3 healthy; total adds severity buckets.
        let s = InstanceHealthSummary::builder()
            .ok(2)
            .info(1)
            .warning(1)
            .degraded(0)
            .severe(1)
            .pending(0)
            .no_data(0)
            .unknown(0)
            .build();
        let counts = super::summarise_instance_health(Some(&s));
        assert_eq!(counts.healthy, 3, "ok + info");
        assert_eq!(counts.total, 5, "ok + info + warning + degraded + severe");

        // All-Grey buckets contribute to total but not to healthy.
        let s = InstanceHealthSummary::builder()
            .pending(2)
            .no_data(1)
            .build();
        let counts = super::summarise_instance_health(Some(&s));
        assert_eq!(counts.healthy, 0);
        assert_eq!(counts.total, 3);

        // None input → 0/0 default.
        let counts = super::summarise_instance_health(None);
        assert_eq!(counts.healthy, 0);
        assert_eq!(counts.total, 0);

        // All-empty summary → 0/0 (rare in practice but defensive).
        let s = InstanceHealthSummary::builder().build();
        let counts = super::summarise_instance_health(Some(&s));
        assert_eq!(counts.healthy, 0);
        assert_eq!(counts.total, 0);
    }

    #[test]
    fn parse_window_ms_accepts_minutes_hours_days() {
        assert_eq!(super::parse_window_ms("30m"), Some(30 * 60_000));
        assert_eq!(super::parse_window_ms("1h"), Some(60 * 60_000));
        assert_eq!(super::parse_window_ms("6h"), Some(6 * 60 * 60_000));
        assert_eq!(super::parse_window_ms("24h"), Some(24 * 60 * 60_000));
        assert_eq!(super::parse_window_ms("7d"), Some(7 * 24 * 60 * 60_000));
        // Whitespace-trimmed.
        assert_eq!(super::parse_window_ms("  2h  "), Some(2 * 60 * 60_000));
        // Case-insensitive on unit.
        assert_eq!(super::parse_window_ms("3H"), Some(3 * 60 * 60_000));
    }

    #[test]
    fn parse_window_ms_rejects_malformed_input() {
        // Empty.
        assert_eq!(super::parse_window_ms(""), None);
        // Missing unit.
        assert_eq!(super::parse_window_ms("30"), None);
        // Missing number.
        assert_eq!(super::parse_window_ms("h"), None);
        // Unknown unit (y / w / s).
        assert_eq!(super::parse_window_ms("1y"), None);
        assert_eq!(super::parse_window_ms("2w"), None);
        assert_eq!(super::parse_window_ms("60s"), None);
        // Non-positive — silently substituting 0 would surprise the operator.
        assert_eq!(super::parse_window_ms("0h"), None);
        assert_eq!(super::parse_window_ms("-1h"), None);
        // Garbage.
        assert_eq!(super::parse_window_ms("hour"), None);
    }

    #[test]
    fn format_insights_results_renders_table() {
        let results = InsightsResults {
            rows: vec![
                InsightsRow {
                    fields: vec![
                        ("@timestamp".into(), "2026-05-23T10:00:00Z".into()),
                        ("@message".into(), "POST /checkout 200 42ms".into()),
                        ("@ptr".into(), "CWL_PTR_X".into()),
                    ],
                },
                InsightsRow {
                    fields: vec![
                        ("@timestamp".into(), "2026-05-23T10:00:01Z".into()),
                        ("@message".into(), "GET /healthcheck 200 1ms".into()),
                        ("@ptr".into(), "CWL_PTR_Y".into()),
                    ],
                },
            ],
            records_scanned: 1234,
            records_matched: 2,
        };
        let body = super::format_insights_results(
            &results,
            "fields @timestamp, @message",
            &["/aws/elasticbeanstalk/prod/var/log/web.stdout.log".to_string()],
        );
        assert!(
            body.contains("matched: 2 / scanned: 1234"),
            "stats line present"
        );
        assert!(body.contains("@timestamp"), "@timestamp header present");
        assert!(body.contains("@message"), "@message header present");
        // @ptr is a record-locator field — always dropped from operator-facing output.
        assert!(
            !body.contains("@ptr"),
            "@ptr field should be filtered out of the rendered table"
        );
        assert!(body.contains("POST /checkout"), "first row body present");
        assert!(body.contains("GET /healthcheck"), "second row body present");
    }

    #[test]
    fn format_insights_results_empty_input_shows_no_rows_stub() {
        let results = InsightsResults {
            rows: vec![],
            records_scanned: 1000,
            records_matched: 0,
        };
        let body = super::format_insights_results(
            &results,
            "fields @message | filter @message like /never/",
            &["/aws/elasticbeanstalk/prod/var/log/web.stdout.log".to_string()],
        );
        assert!(body.contains("no rows matched"), "empty-input stub fires");
        assert!(
            body.contains("matched: 0 / scanned: 1000"),
            "stats line still present"
        );
    }

    #[test]
    fn format_insights_results_truncates_long_values() {
        // A 200-character message should get truncated to ≤ COL_MAX (60)
        // so the table doesn't dominate the overlay.
        let huge = "x".repeat(200);
        let results = InsightsResults {
            rows: vec![InsightsRow {
                fields: vec![("@message".into(), huge.clone())],
            }],
            records_scanned: 1,
            records_matched: 1,
        };
        let body = super::format_insights_results(&results, "fields @message", &[]);
        assert!(
            !body.contains(&huge),
            "raw 200-char value should not appear untouched"
        );
        assert!(
            body.contains("…"),
            "truncation marker should signal the cut to the operator"
        );
    }

    #[tokio::test]
    async fn upload_bundle_uses_multipart_when_size_meets_threshold() {
        // Mocks the three multipart calls (CreateMultipartUpload →
        // UploadPart×N → CompleteMultipartUpload) and feeds upload_bundle
        // a 17-byte tempfile with an 8-byte part size + 1-byte threshold,
        // so we exercise three parts (8, 8, 1) without holding hundreds
        // of MiB in test memory.
        use aws_sdk_s3::operation::complete_multipart_upload::CompleteMultipartUploadOutput;
        use aws_sdk_s3::operation::create_multipart_upload::CreateMultipartUploadOutput;
        use aws_sdk_s3::operation::upload_part::UploadPartOutput;

        const BUCKET: &str = "elasticbeanstalk-eu-west-2-123";
        const KEY: &str = "applications/big-app/v1";
        const UPLOAD_ID: &str = "test-upload-id";

        let cmu_rule = mock!(aws_sdk_s3::Client::create_multipart_upload)
            .match_requests(|req| req.bucket() == Some(BUCKET) && req.key() == Some(KEY))
            .then_output(|| {
                CreateMultipartUploadOutput::builder()
                    .upload_id(UPLOAD_ID)
                    .build()
            });

        // One rule per UploadPart call — aws-smithy-mocks enforces
        // sequential rule matching by default, so a single rule reused
        // across N calls would only match the first call. We assert the
        // part number per rule to pin the order as well as the count.
        let up_rule_1 = mock!(aws_sdk_s3::Client::upload_part)
            .match_requests(|req| {
                req.bucket() == Some(BUCKET)
                    && req.key() == Some(KEY)
                    && req.upload_id() == Some(UPLOAD_ID)
                    && req.part_number() == Some(1)
            })
            .then_output(|| UploadPartOutput::builder().e_tag("\"etag-1\"").build());
        let up_rule_2 = mock!(aws_sdk_s3::Client::upload_part)
            .match_requests(|req| {
                req.bucket() == Some(BUCKET)
                    && req.key() == Some(KEY)
                    && req.upload_id() == Some(UPLOAD_ID)
                    && req.part_number() == Some(2)
            })
            .then_output(|| UploadPartOutput::builder().e_tag("\"etag-2\"").build());
        let up_rule_3 = mock!(aws_sdk_s3::Client::upload_part)
            .match_requests(|req| {
                req.bucket() == Some(BUCKET)
                    && req.key() == Some(KEY)
                    && req.upload_id() == Some(UPLOAD_ID)
                    && req.part_number() == Some(3)
            })
            .then_output(|| UploadPartOutput::builder().e_tag("\"etag-3\"").build());

        let cmpu_rule = mock!(aws_sdk_s3::Client::complete_multipart_upload)
            .match_requests(|req| {
                req.bucket() == Some(BUCKET)
                    && req.key() == Some(KEY)
                    && req.upload_id() == Some(UPLOAD_ID)
                    && req.multipart_upload().map(|m| m.parts().len()) == Some(3)
            })
            .then_output(|| CompleteMultipartUploadOutput::builder().build());

        let s3 = mock_client!(
            aws_sdk_s3,
            [&cmu_rule, &up_rule_1, &up_rule_2, &up_rule_3, &cmpu_rule]
        );
        let cfg = SdkConfig::builder()
            .region(Region::new("us-east-1"))
            .behavior_version(aws_config::BehaviorVersion::latest())
            .build();
        let client = AwsClient::for_tests(
            Client::new(&cfg),
            SqsClient::new(&cfg),
            CwClient::new(&cfg),
            CwLogsClient::new(&cfg),
            s3,
            Ec2Client::new(&cfg),
        );

        let tmp =
            std::env::temp_dir().join(format!("ebman-test-multipart-{}.bin", std::process::id()));
        let bytes = vec![0xABu8; 17];
        std::fs::write(&tmp, &bytes).expect("write tempfile");
        let res = client.upload_bundle_with(BUCKET, KEY, &tmp, 1, 8).await;
        let _ = std::fs::remove_file(&tmp);
        res.expect("multipart upload should succeed");

        assert_eq!(cmu_rule.num_calls(), 1, "CreateMultipartUpload");
        assert_eq!(up_rule_1.num_calls(), 1, "UploadPart #1");
        assert_eq!(up_rule_2.num_calls(), 1, "UploadPart #2");
        assert_eq!(up_rule_3.num_calls(), 1, "UploadPart #3");
        assert_eq!(cmpu_rule.num_calls(), 1, "CompleteMultipartUpload");
    }

    #[tokio::test]
    async fn upload_bundle_aborts_multipart_on_upload_part_failure() {
        // Pins the orphan-prevention invariant: when UploadPart fails
        // mid-flight, the upload loop must issue AbortMultipartUpload
        // before returning the error, otherwise S3 would accumulate
        // partial-upload storage charges that never roll up to a
        // CompleteMultipartUpload.
        use aws_sdk_s3::operation::create_multipart_upload::CreateMultipartUploadOutput;
        use aws_sdk_s3::operation::upload_part::{UploadPartError, UploadPartOutput};
        use aws_smithy_mocks::mock;

        const BUCKET: &str = "elasticbeanstalk-eu-west-2-123";
        const KEY: &str = "applications/abort-test/v1";
        const UPLOAD_ID: &str = "test-abort-upload-id";

        let cmu_rule = mock!(aws_sdk_s3::Client::create_multipart_upload).then_output(|| {
            CreateMultipartUploadOutput::builder()
                .upload_id(UPLOAD_ID)
                .build()
        });
        // First UploadPart succeeds, second fails. We test the worst
        // case where some parts have already landed in S3 — the abort
        // is what reclaims them.
        let up_ok = mock!(aws_sdk_s3::Client::upload_part)
            .match_requests(|req| req.part_number() == Some(1))
            .then_output(|| UploadPartOutput::builder().e_tag("\"etag-1\"").build());
        let up_fail = mock!(aws_sdk_s3::Client::upload_part)
            .match_requests(|req| req.part_number() == Some(2))
            .then_error(|| {
                UploadPartError::unhandled(
                    aws_smithy_types::error::ErrorMetadata::builder().build(),
                )
            });
        let abort_rule = mock!(aws_sdk_s3::Client::abort_multipart_upload)
            .match_requests(|req| {
                req.bucket() == Some(BUCKET)
                    && req.key() == Some(KEY)
                    && req.upload_id() == Some(UPLOAD_ID)
            })
            .then_output(|| {
                aws_sdk_s3::operation::abort_multipart_upload::AbortMultipartUploadOutput::builder()
                    .build()
            });

        let s3 = mock_client!(aws_sdk_s3, [&cmu_rule, &up_ok, &up_fail, &abort_rule]);
        let cfg = SdkConfig::builder()
            .region(Region::new("us-east-1"))
            .behavior_version(aws_config::BehaviorVersion::latest())
            .build();
        let client = AwsClient::for_tests(
            Client::new(&cfg),
            SqsClient::new(&cfg),
            CwClient::new(&cfg),
            CwLogsClient::new(&cfg),
            s3,
            Ec2Client::new(&cfg),
        );

        let tmp = std::env::temp_dir().join(format!("ebman-test-abort-{}.bin", std::process::id()));
        std::fs::write(&tmp, vec![0xCDu8; 16]).expect("write tempfile");
        // threshold=1 forces multipart; part_size=8 → 2 parts (8 + 8).
        let res = client.upload_bundle_with(BUCKET, KEY, &tmp, 1, 8).await;
        let _ = std::fs::remove_file(&tmp);
        assert!(res.is_err(), "upload should surface UploadPart failure");
        assert_eq!(abort_rule.num_calls(), 1, "AbortMultipartUpload must fire");
    }

    #[test]
    fn derive_dlq_url_skips_already_dlq() {
        assert_eq!(
            derive_dlq_url("https://sqs.us-east-1.amazonaws.com/123/foo-dlq"),
            None
        );
    }

    #[test]
    fn derive_dlq_url_strips_trailing_slash() {
        assert_eq!(
            derive_dlq_url("https://sqs.us-east-1.amazonaws.com/123/foo/"),
            Some("https://sqs.us-east-1.amazonaws.com/123/foo-dlq".to_string())
        );
    }

    // ─── Mocked-AWS integration tests ─────────────────────────────────────
    //
    // These exercise the SDK code paths against `aws-smithy-mocks` so we
    // can lock down past regressions and run without an AWS account. Each
    // test names the specific bug it pins to keep the intent crisp when
    // a future change "breaks" it.

    use aws_smithy_mocks::{mock, mock_client};

    /// Build a minimal `AwsClient` where only one sub-client is mocked and
    /// the rest are plain SDK defaults (which will fail loudly if any
    /// unmocked code path is reached — exactly the signal we want).
    fn client_with_eb(eb: Client) -> AwsClient {
        let cfg = aws_config::SdkConfig::builder()
            .region(Region::new("us-east-1"))
            .behavior_version(aws_config::BehaviorVersion::latest())
            .build();
        AwsClient::for_tests(
            eb,
            SqsClient::new(&cfg),
            CwClient::new(&cfg),
            CwLogsClient::new(&cfg),
            S3Client::new(&cfg),
            Ec2Client::new(&cfg),
        )
    }

    fn client_with_cw(cw: CwClient) -> AwsClient {
        let cfg = aws_config::SdkConfig::builder()
            .region(Region::new("us-east-1"))
            .behavior_version(aws_config::BehaviorVersion::latest())
            .build();
        AwsClient::for_tests(
            Client::new(&cfg),
            SqsClient::new(&cfg),
            cw,
            CwLogsClient::new(&cfg),
            S3Client::new(&cfg),
            Ec2Client::new(&cfg),
        )
    }

    /// SSM isn't an arg to `for_tests` (only EB / SQS / CW / CW Logs
    /// / S3 / EC2 are — to keep that signature manageable across the
    /// existing 11 call sites). Tests that need a mocked SSM client
    /// override the field on the constructed AwsClient — the field
    /// is `pub(crate)` for exactly this.
    fn client_with_ssm(ssm: aws_sdk_ssm::Client) -> AwsClient {
        let cfg = aws_config::SdkConfig::builder()
            .region(Region::new("us-east-1"))
            .behavior_version(aws_config::BehaviorVersion::latest())
            .build();
        let mut c = AwsClient::for_tests(
            Client::new(&cfg),
            SqsClient::new(&cfg),
            CwClient::new(&cfg),
            CwLogsClient::new(&cfg),
            S3Client::new(&cfg),
            Ec2Client::new(&cfg),
        );
        c.ssm = ssm;
        c
    }

    fn client_with_eb_and_s3(eb: Client, s3: S3Client) -> AwsClient {
        let cfg = aws_config::SdkConfig::builder()
            .region(Region::new("us-east-1"))
            .behavior_version(aws_config::BehaviorVersion::latest())
            .build();
        AwsClient::for_tests(
            eb,
            SqsClient::new(&cfg),
            CwClient::new(&cfg),
            CwLogsClient::new(&cfg),
            s3,
            Ec2Client::new(&cfg),
        )
    }

    fn client_with_sqs(sqs: SqsClient) -> AwsClient {
        let cfg = aws_config::SdkConfig::builder()
            .region(Region::new("us-east-1"))
            .behavior_version(aws_config::BehaviorVersion::latest())
            .build();
        AwsClient::for_tests(
            Client::new(&cfg),
            sqs,
            CwClient::new(&cfg),
            CwLogsClient::new(&cfg),
            S3Client::new(&cfg),
            Ec2Client::new(&cfg),
        )
    }

    /// Build a base `AwsClient` then swap in a mocked sub-client for one of
    /// the secondary services (ACM / Organizations / Cost Explorer /
    /// Secrets Manager). Field access works because the `tests` module
    /// is a child of `aws`, so the private sub-client fields are visible.
    /// The macro saves repeating six SDK-config lines per test.
    macro_rules! client_with_sub {
        ($field:ident = $value:expr) => {{
            let cfg = aws_config::SdkConfig::builder()
                .region(Region::new("us-east-1"))
                .behavior_version(aws_config::BehaviorVersion::latest())
                .build();
            let mut c = AwsClient::for_tests(
                Client::new(&cfg),
                SqsClient::new(&cfg),
                CwClient::new(&cfg),
                CwLogsClient::new(&cfg),
                S3Client::new(&cfg),
                Ec2Client::new(&cfg),
            );
            c.$field = $value;
            c
        }};
    }

    #[tokio::test]
    async fn list_secrets_maps_secretlistentry_to_summary() {
        // Pins the happy path through ListSecrets — field-by-field
        // mapping from `SecretListEntry` to `SecretSummary`, sort by
        // last_changed desc, and the optional `name_filter` substring
        // match. Caught here because `:secrets` is the operator's
        // entry into Secrets Manager and a silent field-rename in the
        // SDK output would break the picker without a compile error.
        use aws_sdk_secretsmanager::operation::list_secrets::ListSecretsOutput;
        use aws_sdk_secretsmanager::types::SecretListEntry;
        use aws_smithy_types::DateTime as SmithyDt;

        let rule = mock!(aws_sdk_secretsmanager::Client::list_secrets).then_output(|| {
            ListSecretsOutput::builder()
                .secret_list(
                    SecretListEntry::builder()
                        .name("prod/db-password")
                        .arn("arn:aws:secretsmanager:us-east-1:123:secret:prod/db-password-AbCdEf")
                        .description("Production DB master password")
                        .last_changed_date(SmithyDt::from_secs(1_700_000_000))
                        .build(),
                )
                .secret_list(
                    SecretListEntry::builder()
                        .name("staging/api-key")
                        .arn("arn:aws:secretsmanager:us-east-1:123:secret:staging/api-key-XyZ")
                        // Older timestamp than prod/db-password — should
                        // sort below in the result.
                        .last_changed_date(SmithyDt::from_secs(1_600_000_000))
                        .build(),
                )
                .build()
        });
        let secrets = mock_client!(aws_sdk_secretsmanager, [&rule]);
        let client = client_with_sub!(secrets = secrets);

        let all = client.list_secrets(None).await.expect("ok");
        assert_eq!(all.len(), 2);
        // Newest first.
        assert_eq!(all[0].name, "prod/db-password");
        assert_eq!(all[1].name, "staging/api-key");
        // Description + last_changed survived the round-trip.
        assert_eq!(
            all[0].description.as_deref(),
            Some("Production DB master password")
        );
        assert!(all[0].last_changed.is_some());
    }

    #[tokio::test]
    async fn list_certificates_filters_to_issued_and_extracts_domain() {
        // Pins two contract points: (1) ListCertificates is called with
        // `CertificateStatus::Issued` so revoked / pending / expired
        // certs don't show up in the `:listener-edit` picker, and (2)
        // the response's `domain_name` lands in `AcmCert.domain`.
        use aws_sdk_acm::operation::list_certificates::ListCertificatesOutput;
        use aws_sdk_acm::types::{CertificateStatus, CertificateSummary};

        let rule = mock!(aws_sdk_acm::Client::list_certificates)
            .match_requests(|req| {
                req.certificate_statuses()
                    .contains(&CertificateStatus::Issued)
            })
            .then_output(|| {
                ListCertificatesOutput::builder()
                    .certificate_summary_list(
                        CertificateSummary::builder()
                            .certificate_arn("arn:aws:acm:us-east-1:123:certificate/abcd")
                            .domain_name("*.example.com")
                            .build(),
                    )
                    .certificate_summary_list(
                        CertificateSummary::builder()
                            .certificate_arn("arn:aws:acm:us-east-1:123:certificate/efgh")
                            .domain_name("api.example.com")
                            .build(),
                    )
                    .build()
            });
        let acm = mock_client!(aws_sdk_acm, [&rule]);
        let client = client_with_sub!(acm = acm);

        let certs = client.list_certificates().await.expect("ok");
        assert_eq!(certs.len(), 2);
        // Sorted by domain.
        assert_eq!(certs[0].domain, "*.example.com");
        assert_eq!(certs[1].domain, "api.example.com");
        assert_eq!(rule.num_calls(), 1, "ListCertificates fired once");
    }

    #[tokio::test]
    async fn list_org_accounts_sorts_active_first_then_by_name() {
        // The overlay puts ACTIVE accounts at the top so the operator
        // sees switchable accounts before suspended/closed ones. Pins
        // both the field mapping and the sort.
        use aws_sdk_organizations::operation::list_accounts::ListAccountsOutput;
        use aws_sdk_organizations::types::{Account, AccountStatus};

        let rule = mock!(aws_sdk_organizations::Client::list_accounts).then_output(|| {
            ListAccountsOutput::builder()
                .accounts(
                    Account::builder()
                        .id("999999999999")
                        .name("zzz-closed")
                        .email("zzz@example.com")
                        .status(AccountStatus::Suspended)
                        .build(),
                )
                .accounts(
                    Account::builder()
                        .id("222222222222")
                        .name("staging")
                        .email("staging@example.com")
                        .status(AccountStatus::Active)
                        .build(),
                )
                .accounts(
                    Account::builder()
                        .id("111111111111")
                        .name("prod")
                        .email("prod@example.com")
                        .status(AccountStatus::Active)
                        .build(),
                )
                .build()
        });
        let org = mock_client!(aws_sdk_organizations, [&rule]);
        let client = client_with_sub!(org = org);

        let accounts = client.list_org_accounts().await.expect("ok");
        assert_eq!(accounts.len(), 3);
        // ACTIVE first, sorted by name within status.
        assert_eq!(accounts[0].name, "prod");
        assert_eq!(accounts[1].name, "staging");
        assert_eq!(accounts[2].name, "zzz-closed");
    }

    #[tokio::test]
    async fn fetch_env_costs_extracts_env_name_from_tag_group_key() {
        // Cost Explorer encodes the tag group key as
        // `elasticbeanstalk:environment-name$<value>`; the prefix split
        // and the f64 amount parse are the load-bearing bits to pin.
        // Also asserts the metric / granularity / group-by shape on the
        // request, since silently switching from Monthly to Daily would
        // wreck the cache assumptions in cost_cache.rs.
        use aws_sdk_costexplorer::operation::get_cost_and_usage::GetCostAndUsageOutput;
        use aws_sdk_costexplorer::types::{Granularity, Group, MetricValue, ResultByTime};

        let rule = mock!(aws_sdk_costexplorer::Client::get_cost_and_usage)
            .match_requests(|req| {
                req.granularity() == Some(&Granularity::Monthly)
                    && req.metrics().iter().any(|m| m == "UnblendedCost")
                    && req
                        .group_by()
                        .iter()
                        .any(|g| g.key() == Some("elasticbeanstalk:environment-name"))
            })
            .then_output(|| {
                let mut metrics = std::collections::HashMap::new();
                metrics.insert(
                    "UnblendedCost".to_string(),
                    MetricValue::builder().amount("150.25").unit("USD").build(),
                );
                GetCostAndUsageOutput::builder()
                    .results_by_time(
                        ResultByTime::builder()
                            .groups(
                                Group::builder()
                                    .keys("elasticbeanstalk:environment-name$uflexi-prod")
                                    .set_metrics(Some(metrics))
                                    .build(),
                            )
                            .build(),
                    )
                    .build()
            });
        let cost = mock_client!(aws_sdk_costexplorer, [&rule]);
        let client = client_with_sub!(cost = cost);

        let costs = client.fetch_env_costs().await.expect("ok");
        assert_eq!(costs.len(), 1);
        assert_eq!(costs[0].env_name, "uflexi-prod");
        assert!(
            (costs[0].cost_usd - 150.25).abs() < f64::EPSILON,
            "amount parsed from string"
        );
    }

    // ── Regression #1 ────────────────────────────────────────────────────
    // `DescribeConfigurationSettings` returns `WorkerQueueURL = ""` when
    // EB autocreates the queue (the operator didn't override it). The
    // original code looked only at option settings and would show "no
    // queue" for the most common worker-tier shape. The fix queries
    // `DescribeEnvironmentResources` first and only falls back to option
    // settings when explicit overrides exist.

    #[tokio::test]
    async fn worker_queues_resolves_via_describe_environment_resources_when_autocreated() {
        use aws_sdk_elasticbeanstalk::operation::describe_environment_resources::DescribeEnvironmentResourcesOutput;
        use aws_sdk_elasticbeanstalk::types::{EnvironmentResourceDescription, Queue};

        let der = mock!(Client::describe_environment_resources).then_output(|| {
            DescribeEnvironmentResourcesOutput::builder()
                .environment_resources(
                    EnvironmentResourceDescription::builder()
                        .queues(
                            Queue::builder()
                                .name("WorkerQueue")
                                .url("https://sqs.us-east-1.amazonaws.com/123/awseb-e-foo-queue")
                                .build(),
                        )
                        .queues(
                            Queue::builder()
                                .name("WorkerDeadLetterQueue")
                                .url(
                                    "https://sqs.us-east-1.amazonaws.com/123/awseb-e-foo-queue-dlq",
                                )
                                .build(),
                        )
                        .build(),
                )
                .build()
        });
        // Provide an empty configuration-settings response — that's the
        // exact failure mode the bug fix is defending against.
        let dcs = mock!(Client::describe_configuration_settings).then_output(|| {
            aws_sdk_elasticbeanstalk::operation::describe_configuration_settings::DescribeConfigurationSettingsOutput::builder()
                .build()
        });
        let eb = mock_client!(aws_sdk_elasticbeanstalk, [&der, &dcs]);
        let client = client_with_eb(eb);

        // We can't actually fetch SQS stats without mocking SQS too, but
        // the URL resolution is the bit that regressed — assert by
        // calling the option-settings-only path that drives the same
        // logic without the stats round-trip.
        // describe_worker_queues calls queue_stats which would fail
        // against the default sqs client. Use a try-await dance to
        // observe at least the call shape via the mock's call counter.
        let _ = client.describe_worker_queues("eb-app", "eb-env").await;
        assert_eq!(
            der.num_calls(),
            1,
            "describe_environment_resources should be the primary path"
        );
    }

    // ── Regression #2 ────────────────────────────────────────────────────
    // `peek_messages` originally made a single `ReceiveMessage` call —
    // but SQS may return fewer than the requested batch on any one call
    // (it's a maximum, not a guarantee). The fix loops with short long-
    // polling, dedupes by message id across iterations, and bails after
    // two empty batches in a row.

    #[tokio::test]
    async fn peek_messages_loops_and_dedupes_across_batches() {
        use aws_sdk_sqs::operation::receive_message::ReceiveMessageOutput;
        use aws_sdk_sqs::types::Message;

        // First call returns 2 messages, second call returns 1 (including
        // a duplicate of msg-1), third returns empty, fourth returns
        // empty → loop should exit. Expect 3 unique messages.
        fn msg(id: &'static str) -> Message {
            Message::builder().message_id(id).body(id).build()
        }
        let rule = mock!(aws_sdk_sqs::Client::receive_message)
            .sequence()
            .output(|| {
                ReceiveMessageOutput::builder()
                    .messages(msg("msg-1"))
                    .messages(msg("msg-2"))
                    .build()
            })
            .output(|| {
                ReceiveMessageOutput::builder()
                    .messages(msg("msg-1")) // dup
                    .messages(msg("msg-3"))
                    .build()
            })
            .output(|| ReceiveMessageOutput::builder().build())
            .output(|| ReceiveMessageOutput::builder().build())
            .build();
        let sqs = mock_client!(aws_sdk_sqs, [&rule]);
        let client = client_with_sqs(sqs);

        let out = client
            .peek_messages("https://sqs.us-east-1.amazonaws.com/123/q", 10)
            .await
            .expect("peek should succeed");
        let ids: Vec<String> = out.iter().map(|m| m.id.clone()).collect();
        assert_eq!(ids, vec!["msg-1", "msg-2", "msg-3"]);
    }

    #[tokio::test]
    async fn peek_messages_stops_after_two_empty_batches() {
        use aws_sdk_sqs::operation::receive_message::ReceiveMessageOutput;
        // Sequence returns empty twice — should stop without exhausting
        // the call cap.
        let rule = mock!(aws_sdk_sqs::Client::receive_message)
            .sequence()
            .output(|| ReceiveMessageOutput::builder().build())
            .output(|| ReceiveMessageOutput::builder().build())
            // If we reach this, the stop-on-two-empty guard is broken.
            .output(|| {
                ReceiveMessageOutput::builder()
                    .messages(
                        aws_sdk_sqs::types::Message::builder()
                            .message_id("late")
                            .body("late")
                            .build(),
                    )
                    .build()
            })
            .build();
        let sqs = mock_client!(aws_sdk_sqs, [&rule]);
        let client = client_with_sqs(sqs);

        let out = client
            .peek_messages("https://sqs.us-east-1.amazonaws.com/123/q", 10)
            .await
            .expect("peek should succeed");
        assert!(
            out.is_empty(),
            "should have stopped before consuming the 'late' message"
        );
        assert_eq!(
            rule.num_calls(),
            2,
            "exactly two empty-batch calls should terminate the loop"
        );
    }

    // ── Happy-path coverage ──────────────────────────────────────────────
    // Lock down the most-used path so refactors of `list_environments`
    // don't silently break the table-rendering surface.

    #[tokio::test]
    async fn list_environments_maps_describe_environments_to_env_rows() {
        use aws_sdk_elasticbeanstalk::operation::describe_environments::DescribeEnvironmentsOutput;
        use aws_sdk_elasticbeanstalk::types::{EnvironmentDescription, EnvironmentTier};

        let de = mock!(Client::describe_environments).then_output(|| {
            DescribeEnvironmentsOutput::builder()
                .environments(
                    EnvironmentDescription::builder()
                        .environment_name("api-prod")
                        .application_name("api")
                        .status("Ready".into())
                        .health("Green".into())
                        .cname("api-prod.eba.amazonaws.com")
                        .version_label("build-42")
                        .solution_stack_name("64bit Amazon Linux 2 v3.5.0 running Java 17")
                        .tier(EnvironmentTier::builder().name("WebServer").build())
                        .build(),
                )
                .build()
        });
        let eb = mock_client!(aws_sdk_elasticbeanstalk, [&de]);
        let client = client_with_eb(eb);

        let envs = client.list_environments().await.expect("ok");
        assert_eq!(envs.len(), 1);
        let e = &envs[0];
        assert_eq!(e.name, "api-prod");
        assert_eq!(e.application, "api");
        assert_eq!(e.tier, "Web", "tier normalises WebServer → Web");
        assert_eq!(e.platform, "Java 17");
        assert_eq!(e.version_label, "build-42");
    }

    #[tokio::test]
    async fn list_application_versions_pages_through_next_token() {
        // Pins the pagination invariant: orgs with hundreds of historical
        // versions per app must see every entry in `:versions` and let
        // `:rollback` find labels that fall past the first page. Two
        // pages of mocked responses; the second carries no next_token so
        // the loop terminates. Both pages' entries must appear in the
        // returned Vec.
        use aws_sdk_elasticbeanstalk::operation::describe_application_versions::DescribeApplicationVersionsOutput;
        use aws_sdk_elasticbeanstalk::types::ApplicationVersionDescription;

        let page1 = mock!(Client::describe_application_versions)
            .match_requests(|req| {
                req.application_name() == Some("uflexi") && req.next_token().is_none()
            })
            .then_output(|| {
                DescribeApplicationVersionsOutput::builder()
                    .application_versions(
                        ApplicationVersionDescription::builder()
                            .version_label("build-101")
                            .description("first")
                            .build(),
                    )
                    .application_versions(
                        ApplicationVersionDescription::builder()
                            .version_label("build-100")
                            .description("zeroth")
                            .build(),
                    )
                    .next_token("PAGE_2")
                    .build()
            });
        let page2 = mock!(Client::describe_application_versions)
            .match_requests(|req| req.next_token() == Some("PAGE_2"))
            .then_output(|| {
                DescribeApplicationVersionsOutput::builder()
                    .application_versions(
                        ApplicationVersionDescription::builder()
                            .version_label("build-099")
                            .description("rolled")
                            .build(),
                    )
                    .build()
            });
        let eb = mock_client!(aws_sdk_elasticbeanstalk, [&page1, &page2]);
        let client = client_with_eb(eb);

        let versions = client
            .list_application_versions("uflexi")
            .await
            .expect("ok");
        let labels: Vec<&str> = versions.iter().map(|v| v.label.as_str()).collect();
        assert_eq!(
            labels,
            vec!["build-101", "build-100", "build-099"],
            "all three versions from both pages should be returned",
        );
        assert_eq!(page1.num_calls(), 1, "first page fetched once");
        assert_eq!(page2.num_calls(), 1, "second page fetched once");
    }

    // ── MultiSelect picker plumbing ─────────────────────────────────────
    //
    // `:subnets` / `:security-groups` rely on three helpers that all
    // need to round-trip cleanly: VPC discovery via option settings,
    // EC2 inventory listing filtered by VPC, and the comma-split
    // helper that converts EB's CSV format to a clean Vec<String>.

    #[test]
    fn split_csv_trims_and_drops_empties() {
        assert_eq!(
            split_csv("subnet-a,subnet-b, subnet-c, ,subnet-d"),
            vec!["subnet-a", "subnet-b", "subnet-c", "subnet-d"]
        );
        assert!(split_csv("").is_empty());
        assert!(split_csv(",,,").is_empty());
    }

    #[tokio::test]
    async fn fetch_env_vpc_context_pulls_vpc_id_subnets_and_sgs() {
        use aws_sdk_elasticbeanstalk::operation::describe_configuration_settings::DescribeConfigurationSettingsOutput;
        use aws_sdk_elasticbeanstalk::types::{
            ConfigurationOptionSetting, ConfigurationSettingsDescription,
        };

        let dcs = mock!(Client::describe_configuration_settings).then_output(|| {
            DescribeConfigurationSettingsOutput::builder()
                .configuration_settings(
                    ConfigurationSettingsDescription::builder()
                        .option_settings(
                            ConfigurationOptionSetting::builder()
                                .namespace("aws:ec2:vpc")
                                .option_name("VPCId")
                                .value("vpc-123")
                                .build(),
                        )
                        .option_settings(
                            ConfigurationOptionSetting::builder()
                                .namespace("aws:ec2:vpc")
                                .option_name("Subnets")
                                .value("subnet-a,subnet-b")
                                .build(),
                        )
                        .option_settings(
                            ConfigurationOptionSetting::builder()
                                .namespace("aws:ec2:vpc")
                                .option_name("ELBSubnets")
                                .value("subnet-x,subnet-y")
                                .build(),
                        )
                        .option_settings(
                            ConfigurationOptionSetting::builder()
                                .namespace("aws:autoscaling:launchconfiguration")
                                .option_name("SecurityGroups")
                                .value("sg-1,sg-2,sg-3")
                                .build(),
                        )
                        // Noise — should be ignored.
                        .option_settings(
                            ConfigurationOptionSetting::builder()
                                .namespace("aws:elasticbeanstalk:application:environment")
                                .option_name("LOG_LEVEL")
                                .value("debug")
                                .build(),
                        )
                        .build(),
                )
                .build()
        });
        let eb = mock_client!(aws_sdk_elasticbeanstalk, [&dcs]);
        let client = client_with_eb(eb);

        let ctx = client
            .fetch_env_vpc_context("api", "api-prod")
            .await
            .expect("ok");
        assert_eq!(ctx.vpc_id.as_deref(), Some("vpc-123"));
        assert_eq!(ctx.subnets, vec!["subnet-a", "subnet-b"]);
        assert_eq!(ctx.elb_subnets, vec!["subnet-x", "subnet-y"]);
        assert_eq!(ctx.security_groups, vec!["sg-1", "sg-2", "sg-3"]);
    }

    #[tokio::test]
    async fn list_subnets_in_vpc_filters_orders_and_extracts_name_tag() {
        use aws_sdk_ec2::operation::describe_subnets::DescribeSubnetsOutput;
        use aws_sdk_ec2::types::{Subnet, Tag};

        let ds = mock!(aws_sdk_ec2::Client::describe_subnets).then_output(|| {
            DescribeSubnetsOutput::builder()
                .subnets(
                    Subnet::builder()
                        .subnet_id("subnet-2b")
                        .availability_zone("us-east-1b")
                        .cidr_block("10.0.2.0/24")
                        .tags(Tag::builder().key("Name").value("private-2b").build())
                        .build(),
                )
                .subnets(
                    Subnet::builder()
                        .subnet_id("subnet-1a")
                        .availability_zone("us-east-1a")
                        .cidr_block("10.0.1.0/24")
                        .build(),
                )
                .subnets(
                    Subnet::builder()
                        .subnet_id("subnet-1a-overlap")
                        .availability_zone("us-east-1a")
                        .cidr_block("10.0.0.0/24")
                        .build(),
                )
                .build()
        });
        let ec2 = mock_client!(aws_sdk_ec2, [&ds]);
        let cfg = aws_config::SdkConfig::builder()
            .region(Region::new("us-east-1"))
            .behavior_version(aws_config::BehaviorVersion::latest())
            .build();
        let client = AwsClient::for_tests(
            Client::new(&cfg),
            SqsClient::new(&cfg),
            CwClient::new(&cfg),
            CwLogsClient::new(&cfg),
            S3Client::new(&cfg),
            ec2,
        );

        let subnets = client.list_subnets_in_vpc("vpc-abc").await.expect("ok");
        // Ordered by AZ then CIDR — subnet-1a-overlap (10.0.0.0/24) precedes
        // subnet-1a (10.0.1.0/24), then subnet-2b.
        let ids: Vec<&str> = subnets.iter().map(|s| s.id.as_str()).collect();
        assert_eq!(ids, vec!["subnet-1a-overlap", "subnet-1a", "subnet-2b"]);
        // Name tag extracted when present, None when absent.
        assert_eq!(subnets[2].name_tag.as_deref(), Some("private-2b"));
        assert!(subnets[1].name_tag.is_none());
    }

    // ── Write-path coverage ──────────────────────────────────────────────
    //
    // `update_env_option_settings` is the load-bearing write path —
    // every `:capacity`, `:env`, `:tag`, `:subnets`, `:set-option`, etc.
    // ultimately funnels through it. Pin the request-shape contract and
    // the empty-input guard.

    #[tokio::test]
    async fn update_env_option_settings_builds_correct_request_shape() {
        use aws_sdk_elasticbeanstalk::operation::update_environment::UpdateEnvironmentOutput;
        // `match_requests` runs the closure against every captured
        // request; returning false means "no rule matched" and the
        // SDK call returns an error, which the test would then trip on.
        // So an assertion-style predicate doubles as the test body.
        let rule = mock!(Client::update_environment)
            .match_requests(|input| {
                if input.environment_name.as_deref() != Some("api-prod") {
                    return false;
                }
                let options = input.option_settings();
                if options.len() != 2 {
                    return false;
                }
                // Order is preserved from the caller's slice.
                if options[0].namespace.as_deref() != Some("aws:autoscaling:asg")
                    || options[0].option_name.as_deref() != Some("MinSize")
                    || options[0].value.as_deref() != Some("2")
                {
                    return false;
                }
                if options[1].namespace.as_deref() != Some("aws:autoscaling:launchconfiguration")
                    || options[1].option_name.as_deref() != Some("InstanceType")
                    || options[1].value.as_deref() != Some("t3.medium")
                {
                    return false;
                }
                let removes = input.options_to_remove();
                if removes.len() != 1 {
                    return false;
                }
                removes[0].namespace.as_deref()
                    == Some("aws:elasticbeanstalk:application:environment")
                    && removes[0].option_name.as_deref() == Some("OLD_VAR")
            })
            .then_output(|| UpdateEnvironmentOutput::builder().build());
        let eb = mock_client!(aws_sdk_elasticbeanstalk, [&rule]);
        let client = client_with_eb(eb);

        let to_set = vec![
            (
                "aws:autoscaling:asg".to_string(),
                "MinSize".to_string(),
                "2".to_string(),
            ),
            (
                "aws:autoscaling:launchconfiguration".to_string(),
                "InstanceType".to_string(),
                "t3.medium".to_string(),
            ),
        ];
        let to_remove = vec![(
            "aws:elasticbeanstalk:application:environment".to_string(),
            "OLD_VAR".to_string(),
        )];
        client
            .update_env_option_settings("api-prod", &to_set, &to_remove)
            .await
            .expect("expected request shape to match");
        assert_eq!(rule.num_calls(), 1);
    }

    #[tokio::test]
    async fn update_env_option_settings_rejects_empty_input_before_dispatch() {
        // If the guard fails we'd reach the mocked client, which has no
        // rules — that would also error, but with a different message.
        // The empty-input branch must short-circuit *before* any SDK call.
        use aws_sdk_elasticbeanstalk::operation::update_environment::UpdateEnvironmentOutput;
        let trip = mock!(Client::update_environment)
            .then_output(|| UpdateEnvironmentOutput::builder().build());
        let eb = mock_client!(aws_sdk_elasticbeanstalk, [&trip]);
        let client = client_with_eb(eb);

        let err = client
            .update_env_option_settings("api-prod", &[], &[])
            .await
            .expect_err("expected guard to fire");
        assert!(
            err.to_string().contains("nothing to do"),
            "expected nothing-to-do guard, got {err}"
        );
        assert_eq!(
            trip.num_calls(),
            0,
            "guard should short-circuit before any SDK call"
        );
    }

    #[tokio::test]
    async fn update_env_option_settings_surfaces_aws_errors() {
        use aws_sdk_elasticbeanstalk::operation::update_environment::UpdateEnvironmentError;
        use aws_sdk_elasticbeanstalk::types::error::InsufficientPrivilegesException;
        let err_rule = mock!(Client::update_environment).then_error(|| {
            UpdateEnvironmentError::InsufficientPrivilegesException(
                InsufficientPrivilegesException::builder()
                    .message("not authorized to call UpdateEnvironment")
                    .build(),
            )
        });
        let eb = mock_client!(aws_sdk_elasticbeanstalk, [&err_rule]);
        let client = client_with_eb(eb);

        let err = client
            .update_env_option_settings(
                "api-prod",
                &[("aws:autoscaling:asg".into(), "MinSize".into(), "2".into())],
                &[],
            )
            .await
            .expect_err("expected AWS error to propagate");
        // The flatten wraps the SDK error string; we just confirm the
        // contextual prefix is present so logs are actionable.
        assert!(
            err.to_string()
                .contains("UpdateEnvironment(option_settings)"),
            "expected wrapped error context, got {err}"
        );
    }

    #[tokio::test]
    async fn list_security_groups_in_vpc_orders_by_name() {
        use aws_sdk_ec2::operation::describe_security_groups::DescribeSecurityGroupsOutput;
        use aws_sdk_ec2::types::SecurityGroup;

        let dsg = mock!(aws_sdk_ec2::Client::describe_security_groups).then_output(|| {
            DescribeSecurityGroupsOutput::builder()
                .security_groups(
                    SecurityGroup::builder()
                        .group_id("sg-z")
                        .group_name("zeta")
                        .description("z group")
                        .build(),
                )
                .security_groups(
                    SecurityGroup::builder()
                        .group_id("sg-a")
                        .group_name("alpha")
                        .description("a group")
                        .build(),
                )
                .build()
        });
        let ec2 = mock_client!(aws_sdk_ec2, [&dsg]);
        let cfg = aws_config::SdkConfig::builder()
            .region(Region::new("us-east-1"))
            .behavior_version(aws_config::BehaviorVersion::latest())
            .build();
        let client = AwsClient::for_tests(
            Client::new(&cfg),
            SqsClient::new(&cfg),
            CwClient::new(&cfg),
            CwLogsClient::new(&cfg),
            S3Client::new(&cfg),
            ec2,
        );

        let sgs = client
            .list_security_groups_in_vpc("vpc-abc")
            .await
            .expect("ok");
        assert_eq!(sgs.len(), 2);
        assert_eq!(sgs[0].group_name, "alpha");
        assert_eq!(sgs[1].group_name, "zeta");
    }

    // ── Error-path coverage for the load-bearing read methods ────────────
    //
    // Each of these mocks the SDK to return a typed error and asserts
    // our wrapper preserves the operation-name context. Future
    // refactors of these methods will trip a test if they accidentally
    // drop the `.map_err(|e| eyre!(...))?` prefix and start propagating
    // bare SDK errors.

    #[tokio::test]
    async fn list_environments_throttling_error_is_recognised_by_predicate() {
        // End-to-end contract: when EB returns a Throttling-coded SDK error
        // on DescribeEnvironments, the flattened error string we surface
        // must trip `is_throttling_error` so the refresh loop installs a
        // back-off horizon instead of treating it like a normal failure.
        // Pinning this guards against an SDK / smithy change to the
        // stringification format silently breaking back-off.
        use aws_sdk_elasticbeanstalk::operation::describe_environments::DescribeEnvironmentsError;
        let rule = mock!(Client::describe_environments).then_error(|| {
            DescribeEnvironmentsError::generic(
                aws_smithy_types::error::ErrorMetadata::builder()
                    .code("ThrottlingException")
                    .message("Rate exceeded")
                    .build(),
            )
        });
        let eb = mock_client!(aws_sdk_elasticbeanstalk, [&rule]);
        let client = client_with_eb(eb);

        let err = client
            .list_environments()
            .await
            .expect_err("expected throttling error to propagate");
        // Production path: aws error → eyre::Report → flatten_err_to_string
        // (peeks at the Debug form for SDK throttling tokens) → user-facing
        // string → is_throttling_error. Pinning this contract means a future
        // change to either end can't silently break refresh back-off.
        let s = crate::app::flatten_err_to_string(&err);
        assert!(
            crate::app::is_throttling_error(&s),
            "is_throttling_error should fire on the flattened SDK throttling string, got {s:?}"
        );
        // And the user-facing string stays readable — no Debug noise leaks.
        assert!(
            !s.contains("StatusCode") && !s.contains("Extensions"),
            "throttling toast should be clean, got {s:?}"
        );
    }

    #[tokio::test]
    async fn list_environments_expired_token_surfaces_clean_user_message() {
        // When credentials expire mid-session, the SDK returns an
        // `ExpiredToken`-coded error. The toast should not leak Debug
        // noise (HTTP status, headers, body bytes) and must not be
        // misclassified as throttling. Pinning this guards against an
        // SDK stringification change silently turning the toast into a
        // wall of debug output or routing expired-token through the
        // throttle back-off path.
        use aws_sdk_elasticbeanstalk::operation::describe_environments::DescribeEnvironmentsError;
        let rule = mock!(Client::describe_environments).then_error(|| {
            DescribeEnvironmentsError::generic(
                aws_smithy_types::error::ErrorMetadata::builder()
                    .code("ExpiredTokenException")
                    .message("The security token included in the request is expired")
                    .build(),
            )
        });
        let eb = mock_client!(aws_sdk_elasticbeanstalk, [&rule]);
        let client = client_with_eb(eb);

        let err = client
            .list_environments()
            .await
            .expect_err("expected expired-token error to propagate");
        let s = crate::app::flatten_err_to_string(&err);
        assert!(
            !crate::app::is_throttling_error(&s),
            "ExpiredToken should not fire the throttling predicate, got {s:?}"
        );
        assert!(
            !s.contains("StatusCode") && !s.contains("Extensions") && !s.contains("SdkBody"),
            "expired-token toast should be clean, got {s:?}"
        );
    }

    #[tokio::test]
    async fn fetch_env_metrics_batches_and_reorders_by_canonical_id() {
        // CloudWatch `GetMetricData` accepts N queries in one round-trip;
        // `fetch_env_metrics` always dispatches 4 (health / req4xx /
        // req5xx / p90). The response can arrive in any order — the
        // caller re-keys by `id` and returns the canonical order so the
        // Metrics-tab renderer doesn't drift when AWS reorders results
        // (which it has been known to do).
        //
        // Pinning: (a) one batched call, (b) all 4 ids requested, (c)
        // returned series are in canonical order even when the mock
        // shuffles them, (d) labels are mapped per-id.
        use aws_sdk_cloudwatch::operation::get_metric_data::GetMetricDataOutput;
        use aws_sdk_cloudwatch::types::MetricDataResult;
        use aws_smithy_types::DateTime as SdkDateTime;

        let ts = SdkDateTime::from_secs(1_700_000_000);
        let mk_result = move |id: &str, value: f64| {
            MetricDataResult::builder()
                .id(id)
                .timestamps(ts)
                .values(value)
                .build()
        };
        // Return in shuffled order so the test verifies reordering.
        let rule = mock!(aws_sdk_cloudwatch::Client::get_metric_data)
            .match_requests(|req| {
                let ids: Vec<&str> = req
                    .metric_data_queries()
                    .iter()
                    .filter_map(|q| q.id())
                    .collect();
                ids == ["health", "req4xx", "req5xx", "p90"]
            })
            .then_output(move || {
                GetMetricDataOutput::builder()
                    .metric_data_results(mk_result("req5xx", 12.0))
                    .metric_data_results(mk_result("health", 25.0))
                    .metric_data_results(mk_result("p90", 0.42))
                    .metric_data_results(mk_result("req4xx", 3.0))
                    .build()
            });
        let cw = mock_client!(aws_sdk_cloudwatch, [&rule]);
        let client = client_with_cw(cw);

        let series = client
            .fetch_env_metrics("uflexi-prod", 900)
            .await
            .expect("metric fetch should succeed");

        // Single batched call covered all 4 metrics — the function
        // doesn't fan out 4 separate GetMetricData round-trips.
        assert_eq!(rule.num_calls(), 1, "expected exactly one batched call");
        // Canonical order is preserved regardless of response order.
        let ids: Vec<&str> = series.iter().map(|s| s.id.as_str()).collect();
        assert_eq!(ids, vec!["health", "req4xx", "req5xx", "p90"]);
        // Per-id label mapping holds — operator-facing labels not raw ids.
        let by_id: std::collections::HashMap<&str, &str> = series
            .iter()
            .map(|s| (s.id.as_str(), s.label.as_str()))
            .collect();
        assert_eq!(by_id["health"], "Env Health (0–25)");
        assert_eq!(by_id["req4xx"], "4xx Requests / min");
        assert_eq!(by_id["req5xx"], "5xx Requests / min");
        assert_eq!(by_id["p90"], "Latency P90");
        // Timestamp/value zipping survived the shuffle.
        let p90 = series.iter().find(|s| s.id == "p90").unwrap();
        assert_eq!(p90.points.len(), 1);
        assert!((p90.points[0].1 - 0.42).abs() < f64::EPSILON);
    }

    #[tokio::test]
    async fn deploy_from_path_chain_dispatches_each_stage() {
        // End-to-end pinning of the multi-stage `:deploy --from PATH` flow:
        //   1. CreateStorageLocation (EB) → returns the managed bucket name
        //   2. PutObject (S3)             → uploads the bundle bytes
        //   3. CreateApplicationVersion   → registers the version
        //   4. UpdateEnvironment          → deploys to the env
        // Each mock asserts the input it receives matches what the previous
        // stage produced, so a future refactor that reorders / drops a stage
        // or rewires the bucket+key threading fails loud here. This is the
        // most multi-step pure-AWS code path in the project and has no other
        // automated coverage today.
        use aws_sdk_elasticbeanstalk::operation::create_application_version::CreateApplicationVersionOutput;
        use aws_sdk_elasticbeanstalk::operation::create_storage_location::CreateStorageLocationOutput;
        use aws_sdk_elasticbeanstalk::operation::update_environment::UpdateEnvironmentOutput;
        use aws_sdk_s3::operation::put_object::PutObjectOutput;

        const BUCKET: &str = "elasticbeanstalk-us-east-1-123456789012";
        const APP: &str = "uflexi-webapp";
        const ENV: &str = "uflexi-prod";
        const LABEL: &str = "build-2026-05-20-1234567890";
        const KEY: &str = "applications/uflexi-webapp/build-2026-05-20-1234567890";
        let bundle_bytes: Vec<u8> = b"PK\x03\x04 ... a real zip would start here".to_vec();

        let csl_rule = mock!(Client::create_storage_location).then_output(|| {
            CreateStorageLocationOutput::builder()
                .s3_bucket(BUCKET)
                .build()
        });

        // Match every PutObject; assert the bucket + key are exactly what
        // we wired upstream (regression guard for the key-threading bug).
        let put_rule = mock!(aws_sdk_s3::Client::put_object)
            .match_requests(|req| req.bucket() == Some(BUCKET) && req.key() == Some(KEY))
            .then_output(|| PutObjectOutput::builder().build());

        let cav_rule = mock!(Client::create_application_version)
            .match_requests(|req| {
                req.application_name() == Some(APP)
                    && req.version_label() == Some(LABEL)
                    && req.source_bundle().and_then(|s| s.s3_bucket()) == Some(BUCKET)
                    && req.source_bundle().and_then(|s| s.s3_key()) == Some(KEY)
                    && req.auto_create_application() == Some(false)
            })
            .then_output(|| CreateApplicationVersionOutput::builder().build());

        let upd_rule = mock!(Client::update_environment)
            .match_requests(|req| {
                req.environment_name() == Some(ENV) && req.version_label() == Some(LABEL)
            })
            .then_output(|| UpdateEnvironmentOutput::builder().build());

        let eb = mock_client!(aws_sdk_elasticbeanstalk, [&csl_rule, &cav_rule, &upd_rule]);
        let s3 = mock_client!(aws_sdk_s3, [&put_rule]);
        let client = client_with_eb_and_s3(eb, s3);

        // Stage 1
        let bucket = client
            .create_storage_location()
            .await
            .expect("CreateStorageLocation should return the managed bucket");
        assert_eq!(bucket, BUCKET);

        // Stage 2 — write the bundle to a tempfile and let upload_bundle
        // stream it. Threshold is set very high so this exercises the
        // single-PutObject path (the multipart path is covered by a
        // separate test below).
        let tmp =
            std::env::temp_dir().join(format!("ebman-test-bundle-{}.zip", std::process::id()));
        std::fs::write(&tmp, &bundle_bytes).expect("write tempfile");
        let upload_res = client
            .upload_bundle_with(&bucket, KEY, &tmp, u64::MAX, 8 * 1024 * 1024)
            .await;
        let _ = std::fs::remove_file(&tmp);
        upload_res.expect("PutObject should succeed");

        // Stage 3
        client
            .create_app_version(APP, LABEL, Some("test deploy"), &bucket, KEY)
            .await
            .expect("CreateApplicationVersion should succeed");

        // Stage 4
        client
            .deploy_version(ENV, LABEL)
            .await
            .expect("UpdateEnvironment should succeed");

        // Each rule should have fired exactly once.
        assert_eq!(csl_rule.num_calls(), 1, "CreateStorageLocation");
        assert_eq!(put_rule.num_calls(), 1, "S3 PutObject");
        assert_eq!(cav_rule.num_calls(), 1, "CreateApplicationVersion");
        assert_eq!(upd_rule.num_calls(), 1, "UpdateEnvironment");
    }

    #[tokio::test]
    async fn list_environments_surfaces_aws_errors_with_op_context() {
        use aws_sdk_elasticbeanstalk::operation::describe_environments::DescribeEnvironmentsError;
        let rule = mock!(Client::describe_environments).then_error(|| {
            DescribeEnvironmentsError::generic(
                aws_smithy_types::error::ErrorMetadata::builder()
                    .code("InternalServerError")
                    .message("retry later")
                    .build(),
            )
        });
        let eb = mock_client!(aws_sdk_elasticbeanstalk, [&rule]);
        let client = client_with_eb(eb);

        let err = client
            .list_environments()
            .await
            .expect_err("expected AWS error to propagate");
        assert!(
            err.to_string().contains("DescribeEnvironments"),
            "expected operation context, got {err}"
        );
    }

    #[tokio::test]
    async fn peek_messages_surfaces_sqs_errors_with_op_context() {
        use aws_sdk_sqs::operation::receive_message::ReceiveMessageError;
        let rule = mock!(aws_sdk_sqs::Client::receive_message).then_error(|| {
            ReceiveMessageError::generic(
                aws_smithy_types::error::ErrorMetadata::builder()
                    .code("QueueDoesNotExist")
                    .message("queue gone")
                    .build(),
            )
        });
        let sqs = mock_client!(aws_sdk_sqs, [&rule]);
        let cfg = aws_config::SdkConfig::builder()
            .region(Region::new("us-east-1"))
            .behavior_version(aws_config::BehaviorVersion::latest())
            .build();
        let client = AwsClient::for_tests(
            Client::new(&cfg),
            sqs,
            CwClient::new(&cfg),
            CwLogsClient::new(&cfg),
            S3Client::new(&cfg),
            Ec2Client::new(&cfg),
        );

        let err = client
            .peek_messages("https://sqs.us-east-1.amazonaws.com/123/q", 5)
            .await
            .expect_err("expected SQS error to propagate");
        assert!(
            err.to_string().contains("ReceiveMessage"),
            "expected operation context, got {err}"
        );
    }

    #[tokio::test]
    async fn list_subnets_in_vpc_surfaces_ec2_errors_with_op_context() {
        use aws_sdk_ec2::operation::describe_subnets::DescribeSubnetsError;
        let rule = mock!(aws_sdk_ec2::Client::describe_subnets).then_error(|| {
            DescribeSubnetsError::generic(
                aws_smithy_types::error::ErrorMetadata::builder()
                    .code("InvalidVpcID.NotFound")
                    .message("vpc-xxx not found")
                    .build(),
            )
        });
        let ec2 = mock_client!(aws_sdk_ec2, [&rule]);
        let cfg = aws_config::SdkConfig::builder()
            .region(Region::new("us-east-1"))
            .behavior_version(aws_config::BehaviorVersion::latest())
            .build();
        let client = AwsClient::for_tests(
            Client::new(&cfg),
            SqsClient::new(&cfg),
            CwClient::new(&cfg),
            CwLogsClient::new(&cfg),
            S3Client::new(&cfg),
            ec2,
        );

        let err = client
            .list_subnets_in_vpc("vpc-xxx")
            .await
            .expect_err("expected EC2 error to propagate");
        assert!(
            err.to_string().contains("DescribeSubnets"),
            "expected operation context, got {err}"
        );
    }

    #[tokio::test]
    async fn fetch_alarm_history_extracts_kind_and_summary() {
        // Pins the field mapping from the SDK's AlarmHistoryItem onto
        // ebman's AlarmHistoryEntry. If the SDK ever renames
        // `history_item_type` → something else, this test breaks and
        // the operator-visible `:alarm-history` overlay breaks with
        // it. The mock returns one StateUpdate + one
        // ConfigurationUpdate so both common kinds get an assertion.
        use aws_sdk_cloudwatch::operation::describe_alarm_history::DescribeAlarmHistoryOutput;
        use aws_sdk_cloudwatch::types::{AlarmHistoryItem, HistoryItemType};
        use aws_smithy_types::DateTime as SdkDateTime;

        let rule = mock!(aws_sdk_cloudwatch::Client::describe_alarm_history)
            .match_requests(|req| {
                req.alarm_name() == Some("high-cpu") && req.max_records() == Some(50)
            })
            .then_output(|| {
                DescribeAlarmHistoryOutput::builder()
                    .alarm_history_items(
                        AlarmHistoryItem::builder()
                            .alarm_name("high-cpu")
                            .history_item_type(HistoryItemType::StateUpdate)
                            .history_summary("Alarm updated from OK to ALARM")
                            .timestamp(SdkDateTime::from_secs(1_716_640_000))
                            .build(),
                    )
                    .alarm_history_items(
                        AlarmHistoryItem::builder()
                            .alarm_name("high-cpu")
                            .history_item_type(HistoryItemType::ConfigurationUpdate)
                            .history_summary("Threshold changed to 80")
                            .timestamp(SdkDateTime::from_secs(1_716_530_000))
                            .build(),
                    )
                    .build()
            });
        let cw = mock_client!(aws_sdk_cloudwatch, [&rule]);
        let client = client_with_cw(cw);

        let entries = client
            .fetch_alarm_history("high-cpu", 50)
            .await
            .expect("ok");
        assert_eq!(entries.len(), 2);
        assert_eq!(entries[0].kind, "StateUpdate");
        assert_eq!(entries[0].summary, "Alarm updated from OK to ALARM");
        assert!(entries[0].at.is_some(), "timestamp coerced from SDK form");
        assert_eq!(entries[1].kind, "ConfigurationUpdate");
        assert_eq!(entries[1].summary, "Threshold changed to 80");
    }

    #[tokio::test]
    async fn fetch_alarm_history_tolerates_missing_optional_fields() {
        // Real CloudWatch sometimes returns items with missing kind /
        // summary / timestamp (especially for older entries). The
        // function must coerce to sensible defaults (`"?"` for an
        // unknown kind, empty string for missing summary,
        // `None` for missing timestamp) rather than panicking.
        use aws_sdk_cloudwatch::operation::describe_alarm_history::DescribeAlarmHistoryOutput;
        use aws_sdk_cloudwatch::types::AlarmHistoryItem;

        let rule = mock!(aws_sdk_cloudwatch::Client::describe_alarm_history).then_output(|| {
            DescribeAlarmHistoryOutput::builder()
                .alarm_history_items(AlarmHistoryItem::builder().build())
                .build()
        });
        let cw = mock_client!(aws_sdk_cloudwatch, [&rule]);
        let client = client_with_cw(cw);

        let entries = client.fetch_alarm_history("any", 10).await.expect("ok");
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0].kind, "?");
        assert_eq!(entries[0].summary, "");
        assert!(entries[0].at.is_none());
    }

    #[tokio::test(start_paused = true)]
    async fn run_shell_command_collects_per_instance_result_on_success() {
        // Happy path: one instance, SendCommand returns a command_id,
        // first GetCommandInvocation poll returns Success with
        // stdout/stderr/exit_code. `start_paused = true` + advance()
        // skips the actual 2s sleep so the test runs in ms.
        use aws_sdk_ssm::operation::get_command_invocation::GetCommandInvocationOutput;
        use aws_sdk_ssm::operation::send_command::SendCommandOutput;
        use aws_sdk_ssm::types::{Command, CommandInvocationStatus};

        const CMD_ID: &str = "01234567-89ab-cdef-0123-456789abcdef";

        let send_rule = mock!(aws_sdk_ssm::Client::send_command)
            .match_requests(|req| {
                req.document_name() == Some("AWS-RunShellScript")
                    && req.instance_ids().contains(&"i-aaa".to_string())
            })
            .then_output(|| {
                SendCommandOutput::builder()
                    .command(Command::builder().command_id(CMD_ID).build())
                    .build()
            });
        let poll_rule = mock!(aws_sdk_ssm::Client::get_command_invocation)
            .match_requests(|req| {
                req.command_id() == Some(CMD_ID) && req.instance_id() == Some("i-aaa")
            })
            .then_output(|| {
                GetCommandInvocationOutput::builder()
                    .command_id(CMD_ID)
                    .instance_id("i-aaa")
                    .status(CommandInvocationStatus::Success)
                    .response_code(0)
                    .standard_output_content("up 3 days")
                    .build()
            });
        let ssm = mock_client!(aws_sdk_ssm, [&send_rule, &poll_rule]);
        let client = client_with_ssm(ssm);

        // Background the run + advance the paused clock past the
        // first 2s sleep so the poll fires.
        let handle = tokio::spawn(async move {
            client
                .run_shell_command(&["i-aaa".to_string()], "uptime", 60)
                .await
        });
        tokio::time::sleep(std::time::Duration::from_secs(3)).await;
        let results = handle.await.unwrap().expect("ok");
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].instance_id, "i-aaa");
        assert_eq!(results[0].status, "Success");
        assert_eq!(results[0].exit_code, 0);
        assert_eq!(results[0].stdout, "up 3 days");
        assert_eq!(results[0].stderr, "");
    }

    #[tokio::test(start_paused = true)]
    async fn run_shell_command_synthesises_local_timeout_when_deadline_passes() {
        // If GetCommandInvocation keeps returning InProgress past the
        // wall-clock deadline, the function emits a synthetic
        // `TimedOut(local)` row for the still-pending instance rather
        // than hanging. Pinning this catches regressions in the
        // deadline-bound break of the poll loop.
        use aws_sdk_ssm::operation::get_command_invocation::GetCommandInvocationOutput;
        use aws_sdk_ssm::operation::send_command::SendCommandOutput;
        use aws_sdk_ssm::types::{Command, CommandInvocationStatus};

        const CMD_ID: &str = "deadbeef-0000-0000-0000-000000000000";

        let send_rule = mock!(aws_sdk_ssm::Client::send_command).then_output(|| {
            SendCommandOutput::builder()
                .command(Command::builder().command_id(CMD_ID).build())
                .build()
        });
        // Permanent InProgress — never resolves.
        let stuck = mock!(aws_sdk_ssm::Client::get_command_invocation).then_output(|| {
            GetCommandInvocationOutput::builder()
                .command_id(CMD_ID)
                .instance_id("i-stuck")
                .status(CommandInvocationStatus::InProgress)
                .response_code(0)
                .build()
        });
        let ssm = mock_client!(aws_sdk_ssm, [&send_rule, &stuck]);
        let client = client_with_ssm(ssm);

        let handle = tokio::spawn(async move {
            // 1s wall-clock — much shorter than the 2s poll interval
            // so we hit the deadline on the FIRST loop iteration.
            client
                .run_shell_command(&["i-stuck".to_string()], "sleep 999", 1)
                .await
        });
        // Advance well past the 1s deadline + the 2s poll interval.
        tokio::time::sleep(std::time::Duration::from_secs(4)).await;
        let results = handle.await.unwrap().expect("ok");
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].instance_id, "i-stuck");
        assert_eq!(
            results[0].status, "TimedOut(local)",
            "synthetic timeout row should signal which instance didn't finish"
        );
        assert_eq!(results[0].exit_code, -1);
    }
}
