#!/usr/bin/env python3
"""Collect an account-scoped AWS cost and inventory snapshot without cloud writes.

The snapshot is raw deterministic input for the multi-cloud normalizer. Required billing
and ownership failures stop a complete report; deeper audit failures remain explicit evidence
gaps. Only decision-useful Name/cp9 tags and description categories are retained.

Run `python3 tools/cloud-cost-aws.py check` offline after collector changes. Its API-shape
replay guards pagination, storage units, failed reads versus empty inventory, image-status
coverage and tag normalization: errors here would silently misstate cleanup evidence.
The command has no credentials and performs no AWS calls.
"""

from __future__ import annotations

import argparse
import collections
import json
import re
import subprocess
import sys
from concurrent.futures import ThreadPoolExecutor
from contextlib import closing
from datetime import date, datetime, timedelta, timezone
from decimal import Decimal
from pathlib import Path
from typing import Any, Callable

import boto3
from botocore.config import Config
from botocore.exceptions import BotoCoreError, ClientError


VERSION = "cp9.cloud-cost-aws-snapshot/v3"
AWS_ACCOUNT = re.compile(r"^[0-9]{12}$")
SCOPE_ID = re.compile(r"^[a-z0-9][a-z0-9-]{2,79}$")
BILLING_METRICS = {
    "amortized": ("NetAmortizedCost", "NetUnblendedCost"),
    "unblended": ("NetUnblendedCost", "NetAmortizedCost"),
}
INVENTORY_SURFACES = (
    "vpc",
    "eks",
    "compute",
    "ebs",
    "nat",
    "loadbalancer",
    "rds",
    "vpc-endpoint",
    "public-ipv4",
    "msk",
)
EVIDENCE_SURFACES = (
    "kms-key",
    "private-ca",
    "ecr-repository",
    "opensearch-domain",
    "elasticache-cluster",
    "config-recorder",
    "config-conformance-pack",
    "quicksight-user",
    "vpn",
    "vpc-lattice-service-network",
)


class AWSCallError(RuntimeError):
    pass


def aws_json(*args: str, region: str | None = None) -> Any:
    command = ["aws", "--no-cli-pager", "--output", "json"]
    if region:
        command += ["--region", region]
    command += list(args)
    try:
        result = subprocess.run(command, capture_output=True, text=True, timeout=180)
    except (OSError, subprocess.TimeoutExpired) as exc:
        raise AWSCallError(f"{' '.join(args[:2])}: {exc}") from exc
    if result.returncode != 0:
        error_code = re.search(r"An error occurred \(([^)]+)\)", result.stderr)
        detail = error_code.group(1) if error_code else "AWS CLI failed"
        raise AWSCallError(f"{' '.join(args[:2])}: {detail}")
    try:
        return json.loads(result.stdout or "null")
    except json.JSONDecodeError as exc:
        raise AWSCallError(f"{' '.join(args[:2])}: invalid JSON response") from exc


def window_bounds(today: date) -> dict[str, dict[str, str]]:
    # Thursday's run sees the week ending Monday after the agreed 72-hour billing delay.
    cutoff = today - timedelta(days=3)
    current_end = cutoff - timedelta(days=cutoff.weekday())
    current_start = current_end - timedelta(days=7)
    previous_start = current_start - timedelta(days=7)
    return {
        "previous": {"start": previous_start.isoformat(), "end": current_start.isoformat()},
        "current": {"start": current_start.isoformat(), "end": current_end.isoformat()},
    }


def enabled_regions() -> list[str]:
    response = aws_json("ec2", "describe-regions", "--all-regions", region="us-east-1")
    return sorted(
        item["RegionName"]
        for item in response.get("Regions", [])
        if item.get("OptInStatus") in {"opt-in-not-required", "opted-in"}
    )


def resource(
    kind: str, region: str, resource_id: str, weight: int = 1, **fields: Any
) -> dict[str, Any]:
    for key, value in decision_tags(fields.pop("tags", None)).items():
        fields.setdefault(
            {"Name": "name", "cp9-owner": "owner", "cp9-session": "session"}[key], value
        )
    return {"kind": kind, "region": region, "id": resource_id, "weight": weight, **fields}


def decision_tags(value: Any) -> dict[str, str]:
    tags = {
        item.get("Key"): item.get("Value")
        for item in value or []
        if isinstance(item, dict)
    }
    return {
        key: tags[key]
        for key in ("Name", "cp9-owner", "cp9-session")
        if isinstance(tags.get(key), str) and tags[key]
    }


def vpc_resource(region: str, item: dict[str, Any]) -> dict[str, Any]:
    tags = decision_tags(item.get("Tags"))
    return resource(
        "vpc",
        region,
        item["VpcId"],
        vpc_id=item["VpcId"],
        name=tags.get("Name"),
        owner=tags.get("cp9-owner"),
        session=tags.get("cp9-session"),
        state=item.get("State"),
    )


def attach_cluster_ids(rows: list[dict[str, Any]]) -> None:
    cluster_by_vpc: dict[str, list[str]] = collections.defaultdict(list)
    for item in rows:
        if item["kind"] == "eks" and item.get("vpc_id"):
            cluster_by_vpc[item["vpc_id"]].append(item["id"])
    for item in rows:
        item["cluster_ids"] = (
            [item["id"]]
            if item["kind"] == "eks"
            else sorted(cluster_by_vpc.get(item.get("vpc_id", ""), []))
        )


def collect_region(region: str) -> tuple[list[dict[str, Any]], list[str]]:
    rows: list[dict[str, Any]] = []
    gaps: list[str] = []
    instance_vpc: dict[str, str] = {}
    subnet_vpc: dict[str, str] = {}

    def read(surface: str, loader: Callable[[], None]) -> None:
        try:
            loader()
        except (AWSCallError, KeyError, TypeError) as exc:
            gaps.append(f"{region}:{surface}: {exc}")

    def eks() -> None:
        for name in aws_json("eks", "list-clusters", region=region).get("clusters", []):
            cluster = aws_json("eks", "describe-cluster", "--name", name, region=region)["cluster"]
            rows.append(
                resource(
                    "eks",
                    region,
                    name,
                    vpc_id=cluster["resourcesVpcConfig"]["vpcId"],
                    created=str(cluster.get("createdAt", ""))[:10],
                )
            )

    def vpcs() -> None:
        for item in aws_json("ec2", "describe-vpcs", region=region).get("Vpcs", []):
            rows.append(vpc_resource(region, item))

    def compute() -> None:
        response = aws_json(
            "ec2",
            "describe-instances",
            "--filters",
            "Name=instance-state-name,Values=pending,running,stopping,stopped",
            region=region,
        )
        for reservation in response.get("Reservations", []):
            for item in reservation.get("Instances", []):
                instance_vpc[item["InstanceId"]] = item.get("VpcId", "")
                state = (item.get("State") or {}).get("Name")
                if state not in {"pending", "running"}:
                    continue
                rows.append(
                    resource(
                        "compute",
                        region,
                        item["InstanceId"],
                        vpc_id=item.get("VpcId"),
                        created=str(item.get("LaunchTime", ""))[:10],
                        subtype=item.get("InstanceType"),
                        state=state,
                    )
                )

    def subnets() -> None:
        for item in aws_json("ec2", "describe-subnets", region=region).get("Subnets", []):
            subnet_vpc[item["SubnetId"]] = item["VpcId"]

    def volumes() -> None:
        for item in aws_json("ec2", "describe-volumes", region=region).get("Volumes", []):
            attachments = item.get("Attachments") or []
            attached_instance = attachments[0].get("InstanceId") if attachments else None
            rows.append(
                resource(
                    "ebs",
                    region,
                    item["VolumeId"],
                    weight=item.get("Size") or 1,
                    vpc_id=instance_vpc.get(attached_instance or ""),
                    created=str(item.get("CreateTime", ""))[:10],
                    subtype=item.get("VolumeType"),
                    size_gib=item.get("Size"),
                    state=item.get("State"),
                    tags=item.get("Tags"),
                    source=(
                        next(
                            (
                                tag["Value"]
                                for tag in item.get("Tags", [])
                                if tag.get("Key") == "kubernetes.io/created-for/pvc/name"
                            ),
                            None,
                        )
                    ),
                )
            )

    def nat_gateways() -> None:
        response = aws_json(
            "ec2",
            "describe-nat-gateways",
            "--filter",
            "Name=state,Values=pending,available",
            region=region,
        )
        for item in response.get("NatGateways", []):
            rows.append(
                resource(
                    "nat",
                    region,
                    item["NatGatewayId"],
                    vpc_id=item.get("VpcId"),
                    created=str(item.get("CreateTime", ""))[:10],
                )
            )

    def load_balancers() -> None:
        modern = aws_json("elbv2", "describe-load-balancers", region=region).get(
            "LoadBalancers", []
        )
        target_groups = aws_json("elbv2", "describe-target-groups", region=region).get(
            "TargetGroups", []
        )
        groups_by_lb: dict[str, list[str]] = collections.defaultdict(list)
        for target_group in target_groups:
            for load_balancer_arn in target_group.get("LoadBalancerArns", []):
                groups_by_lb[load_balancer_arn].append(target_group["TargetGroupArn"])
        health_by_lb: dict[str, tuple[int, int]] = {}
        for load_balancer_arn, group_arns in groups_by_lb.items():
            registered = healthy = 0
            for group_arn in group_arns:
                health = aws_json(
                    "elbv2",
                    "describe-target-health",
                    "--target-group-arn",
                    group_arn,
                    region=region,
                )
                targets = health.get("TargetHealthDescriptions", [])
                registered += len(targets)
                healthy += sum(
                    (target.get("TargetHealth") or {}).get("State") == "healthy"
                    for target in targets
                )
            health_by_lb[load_balancer_arn] = (registered, healthy)
        for item in modern:
            arn = item["LoadBalancerArn"]
            registered, healthy = health_by_lb.get(arn, (0, 0))
            rows.append(
                resource(
                    "loadbalancer",
                    region,
                    arn,
                    vpc_id=item.get("VpcId"),
                    created=str(item.get("CreatedTime", ""))[:10],
                    subtype=item.get("Type"),
                    target_groups=len(groups_by_lb.get(arn, [])),
                    registered_targets=registered,
                    healthy_targets=healthy,
                    name=item.get("LoadBalancerName"),
                )
            )
        for item in aws_json("elb", "describe-load-balancers", region=region).get(
            "LoadBalancerDescriptions", []
        ):
            instances = item.get("Instances") or []
            healthy = 0
            if instances:
                states = aws_json(
                    "elb",
                    "describe-instance-health",
                    "--load-balancer-name",
                    item["LoadBalancerName"],
                    region=region,
                ).get("InstanceStates", [])
                healthy = sum(state.get("State") == "InService" for state in states)
            rows.append(
                resource(
                    "loadbalancer",
                    region,
                    item["LoadBalancerName"],
                    vpc_id=item.get("VPCId"),
                    created=str(item.get("CreatedTime", ""))[:10],
                    subtype="classic",
                    target_groups=0,
                    registered_targets=len(instances),
                    healthy_targets=healthy,
                    name=item.get("LoadBalancerName"),
                )
            )

    def databases() -> None:
        for item in aws_json("rds", "describe-db-instances", region=region).get(
            "DBInstances", []
        ):
            subnets = ((item.get("DBSubnetGroup") or {}).get("Subnets") or [])
            subnet_id = subnets[0].get("SubnetIdentifier") if subnets else ""
            rows.append(
                resource(
                    "rds",
                    region,
                    item["DBInstanceArn"],
                    vpc_id=subnet_vpc.get(subnet_id),
                    created=str(item.get("InstanceCreateTime", ""))[:10],
                    subtype=item.get("DBInstanceClass"),
                )
            )

    def endpoints() -> None:
        for item in aws_json("ec2", "describe-vpc-endpoints", region=region).get(
            "VpcEndpoints", []
        ):
            rows.append(
                resource(
                    "vpc-endpoint",
                    region,
                    item["VpcEndpointId"],
                    vpc_id=item.get("VpcId"),
                    created=str(item.get("CreationTimestamp", ""))[:10],
                )
            )

    def public_ipv4() -> None:
        addresses = aws_json("ec2", "describe-addresses", region=region).get("Addresses", [])
        elastic_ips = {item.get("PublicIp") for item in addresses}
        interfaces = aws_json("ec2", "describe-network-interfaces", region=region).get(
            "NetworkInterfaces", []
        )
        interface_by_public_ip = {
            public_ip: item
            for item in interfaces
            if (public_ip := (item.get("Association") or {}).get("PublicIp"))
        }
        for item in addresses:
            interface = interface_by_public_ip.get(item.get("PublicIp"), {})
            rows.append(
                resource(
                    "public-ipv4",
                    region,
                    item.get("AllocationId") or item["PublicIp"],
                    vpc_id=interface.get("VpcId"),
                    subtype="elastic-ip",
                    state="associated" if item.get("AssociationId") else "unassociated",
                    tags=item.get("Tags"),
                )
            )
        for item in interfaces:
            public_ip = (item.get("Association") or {}).get("PublicIp")
            if public_ip and public_ip not in elastic_ips:
                rows.append(
                    resource(
                        "public-ipv4",
                        region,
                        item["NetworkInterfaceId"],
                        vpc_id=item.get("VpcId"),
                        subtype=item.get("InterfaceType"),
                        state="associated",
                    )
                )

    def kafka() -> None:
        for item in aws_json("kafka", "list-clusters-v2", region=region).get(
            "ClusterInfoList", []
        ):
            provisioned = item.get("Provisioned") or {}
            serverless = item.get("Serverless") or {}
            subnet_ids = (provisioned.get("BrokerNodeGroupInfo") or {}).get(
                "ClientSubnets", []
            ) or (serverless.get("VpcConfigs") or [{}])[0].get("SubnetIds", [])
            rows.append(
                resource(
                    "msk",
                    region,
                    item["ClusterArn"],
                    weight=provisioned.get("NumberOfBrokerNodes") or 1,
                    vpc_id=subnet_vpc.get(subnet_ids[0]) if subnet_ids else None,
                    created=str(item.get("CreationTime", ""))[:10],
                    subtype=(provisioned.get("BrokerNodeGroupInfo") or {}).get(
                        "InstanceType", "provisioned"
                    )
                    if provisioned
                    else "serverless",
                )
            )

    read("vpc", vpcs)
    read("eks", eks)
    read("compute", compute)
    read("subnets", subnets)
    read("ebs", volumes)
    read("nat", nat_gateways)
    read("loadbalancer", load_balancers)
    read("rds", databases)
    read("vpc-endpoint", endpoints)
    read("public-ipv4", public_ipv4)
    read("msk", kafka)
    return rows, gaps


def collect_cost_groups(
    windows: dict[str, dict[str, str]],
    metric: str,
    comparison_metric: str,
    dimensions: tuple[str, str],
) -> tuple[dict[str, Any], list[str]]:
    request = [
        "ce",
        "get-cost-and-usage",
        "--time-period",
        f"Start={windows['previous']['start']},End={windows['current']['end']}",
        "--granularity",
        "DAILY",
        "--metrics",
        metric,
        comparison_metric,
        "UsageQuantity",
        "--group-by",
        *(f"Type=DIMENSION,Key={dimension}" for dimension in dimensions),
    ]
    days_by_start: dict[str, dict[str, Any]] = {}
    seen_tokens: set[str] = set()
    next_token: str | None = None
    page_count = 0
    try:
        while True:
            page = aws_json(
                *request,
                *(["--next-page-token", next_token] if next_token else []),
                region="us-east-1",
            )
            page_count += 1
            results = page.get("ResultsByTime")
            if not isinstance(results, list):
                raise AWSCallError("ce get-cost-and-usage: missing ResultsByTime")
            for item in results:
                start = item["TimePeriod"]["Start"]
                day = days_by_start.setdefault(
                    start,
                    {
                        "TimePeriod": item["TimePeriod"],
                        "Estimated": False,
                        "Groups": [],
                    },
                )
                day["Estimated"] = day["Estimated"] or item.get("Estimated") is True
                day["Groups"].extend(item.get("Groups", []))
            next_token = page.get("NextPageToken")
            if not next_token:
                break
            if next_token in seen_tokens:
                raise AWSCallError("ce get-cost-and-usage: repeated NextPageToken")
            seen_tokens.add(next_token)
    except AWSCallError as exc:
        return {"page_count": page_count, "days": []}, [f"billing: {exc}"]
    starts = sorted(days_by_start)
    days = [days_by_start[start] for start in starts]
    gaps = [
        f"billing:{item.get('TimePeriod', {}).get('Start', 'unknown')}: estimated"
        for item in days
        if item.get("Estimated") is True
    ]
    first_day = date.fromisoformat(windows["previous"]["start"])
    expected_starts = [(first_day + timedelta(days=offset)).isoformat() for offset in range(14)]
    if starts != expected_starts:
        gaps.append("billing: expected 14 contiguous daily results")
    return {
        "page_count": page_count,
        "days": days,
    }, gaps


def collect_cost_history(
    windows: dict[str, dict[str, str]], metric: str, comparison_metric: str
) -> tuple[dict[str, Any], list[str]]:
    end = date.fromisoformat(windows["current"]["end"])
    start = end - timedelta(days=56)
    try:
        response = aws_json(
            "ce",
            "get-cost-and-usage",
            "--time-period",
            f"Start={start.isoformat()},End={end.isoformat()}",
            "--granularity",
            "DAILY",
            "--metrics",
            metric,
            comparison_metric,
            region="us-east-1",
        )
        days = response.get("ResultsByTime")
        if not isinstance(days, list):
            raise AWSCallError("ce get-cost-and-usage history: missing ResultsByTime")
    except AWSCallError as exc:
        return {"start": start.isoformat(), "end": end.isoformat(), "days": []}, [
            f"billing-history: {exc}"
        ]
    expected = [(start + timedelta(days=offset)).isoformat() for offset in range(56)]
    starts = [item.get("TimePeriod", {}).get("Start") for item in days]
    gaps = []
    if starts != expected:
        gaps.append("billing-history: expected 56 contiguous daily results")
    if any(item.get("Estimated") is True for item in days):
        gaps.append("billing-history: contains estimated results")
    return {"start": start.isoformat(), "end": end.isoformat(), "days": days}, gaps


def collect_savings_plan_coverage(
    windows: dict[str, dict[str, str]]
) -> tuple[dict[str, Any], list[str]]:
    request = [
        "ce",
        "get-savings-plans-coverage",
        "--time-period",
        f"Start={windows['previous']['start']},End={windows['current']['end']}",
        "--group-by",
        "Type=DIMENSION,Key=SERVICE",
        "--metrics",
        "SpendCoveredBySavingsPlans",
        "--max-results",
        "100",
        "--no-paginate",
    ]
    rows: list[dict[str, Any]] = []
    seen_tokens: set[str] = set()
    next_token: str | None = None
    page_count = 0
    try:
        while True:
            page = aws_json(
                *request,
                *(["--next-token", next_token] if next_token else []),
                region="us-east-1",
            )
            page_count += 1
            values = page.get("SavingsPlansCoverages")
            if not isinstance(values, list):
                raise AWSCallError(
                    "ce get-savings-plans-coverage: missing SavingsPlansCoverages"
                )
            rows.extend(values)
            next_token = page.get("NextToken")
            if not next_token:
                break
            if next_token in seen_tokens:
                raise AWSCallError(
                    "ce get-savings-plans-coverage: repeated NextToken"
                )
            seen_tokens.add(next_token)
    except AWSCallError as exc:
        return {"page_count": page_count, "rows": []}, [f"savings-plans: {exc}"]
    return {"page_count": page_count, "rows": rows}, []


def collect_cost(
    windows: dict[str, dict[str, str]], billing_source: str
) -> tuple[dict[str, Any], list[str]]:
    metric, comparison_metric = BILLING_METRICS[billing_source]
    regional, regional_gaps = collect_cost_groups(
        windows, metric, comparison_metric, ("REGION", "USAGE_TYPE")
    )
    service, service_gaps = collect_cost_groups(
        windows, metric, comparison_metric, ("SERVICE", "USAGE_TYPE")
    )
    history, history_gaps = collect_cost_history(windows, metric, comparison_metric)
    savings_plans, savings_plan_gaps = collect_savings_plan_coverage(windows)
    return {
        "metric": metric,
        "comparison_metric": comparison_metric,
        "page_count": regional["page_count"],
        "days": regional["days"],
        "service_usage_page_count": service["page_count"],
        "service_usage_days": service["days"],
        "history": history,
        "savings_plans": savings_plans,
    }, (
        regional_gaps
        + [f"service-usage:{gap}" for gap in service_gaps]
        + history_gaps
        + savings_plan_gaps
    )


def bucket_metric_values(
    batch: list[tuple[str, str]], results: list[dict[str, Any]]
) -> list[tuple[str, Decimal, str]]:
    by_id = {item["Id"]: item for item in results}
    values = []
    for index, (bucket, _storage_type) in enumerate(batch):
        result = by_id.get(f"m{index}", {})
        if result.get("StatusCode") not in {None, "Complete"}:
            raise AWSCallError(f"cloudwatch get-metric-data: {result.get('StatusCode')}")
        if result.get("Values") and result.get("Timestamps"):
            values.append(
                (bucket, Decimal(str(result["Values"][0])), str(result["Timestamps"][0]))
            )
    return values


def collect_object_storage(collected_at: datetime) -> tuple[dict[str, Any], list[str]]:
    try:
        bucket_rows = aws_json("s3api", "list-buckets", "--page-size", "1000").get(
            "Buckets", []
        )
        buckets: dict[str, dict[str, str]] = {}
        for item in bucket_rows:
            bucket = item["Name"]
            region = item.get("BucketRegion")
            if not region:
                location = aws_json(
                    "s3api", "get-bucket-location", "--bucket", bucket
                ).get("LocationConstraint")
                region = {None: "us-east-1", "EU": "eu-west-1"}.get(location, location)
            buckets[bucket] = {
                "region": region,
                "created": str(item.get("CreationDate", ""))[:10],
            }

        values: dict[str, Decimal] = collections.defaultdict(Decimal)
        observed: dict[str, str] = {}
        for region in sorted({item["region"] for item in buckets.values()}):
            metrics = aws_json(
                "cloudwatch",
                "list-metrics",
                "--namespace",
                "AWS/S3",
                "--metric-name",
                "BucketSizeBytes",
                region=region,
            ).get("Metrics", [])
            selected = []
            for metric in metrics:
                dimensions = {
                    item["Name"]: item["Value"] for item in metric.get("Dimensions", [])
                }
                bucket = dimensions.get("BucketName")
                storage_type = dimensions.get("StorageType")
                if (buckets.get(bucket) or {}).get("region") == region and storage_type:
                    selected.append((bucket, storage_type))
            selected = sorted(set(selected))
            for offset in range(0, len(selected), 100):
                batch = selected[offset : offset + 100]
                queries = [
                    {
                        "Id": f"m{index}",
                        "MetricStat": {
                            "Metric": {
                                "Namespace": "AWS/S3",
                                "MetricName": "BucketSizeBytes",
                                "Dimensions": [
                                    {"Name": "BucketName", "Value": bucket},
                                    {"Name": "StorageType", "Value": storage_type},
                                ],
                            },
                            "Period": 86400,
                            "Stat": "Average",
                        },
                        "ReturnData": True,
                    }
                    for index, (bucket, storage_type) in enumerate(batch)
                ]
                response = aws_json(
                    "cloudwatch",
                    "get-metric-data",
                    "--metric-data-queries",
                    json.dumps(queries, separators=(",", ":")),
                    "--start-time",
                    (collected_at - timedelta(days=7)).isoformat(),
                    "--end-time",
                    collected_at.isoformat(),
                    "--scan-by",
                    "TimestampDescending",
                    region=region,
                )
                for bucket, amount, timestamp in bucket_metric_values(
                    batch, response.get("MetricDataResults", [])
                ):
                    values[bucket] += amount
                    observed[bucket] = min(observed.get(bucket, timestamp), timestamp)

        def audit_bucket(bucket: str) -> tuple[str, dict[str, Any]]:
            audit: dict[str, Any] = {
                "empty": None,
                "lifecycle_state": "unknown",
                "lifecycle_actions": [],
                "audit_errors": [],
            }
            region = buckets[bucket]["region"]
            try:
                versions = aws_json(
                    "s3api",
                    "list-object-versions",
                    "--bucket",
                    bucket,
                    "--max-keys",
                    "1",
                    "--no-paginate",
                    region=region,
                )
                if versions.get("Name") != bucket:
                    raise AWSCallError("version listing did not identify the bucket")
                audit["empty"] = not (
                    versions.get("Versions")
                    or versions.get("DeleteMarkers")
                    or versions.get("IsTruncated")
                )
            except AWSCallError as exc:
                audit["audit_errors"].append(f"empty-state: {exc}")
            try:
                rules = aws_json(
                    "s3api",
                    "get-bucket-lifecycle-configuration",
                    "--bucket",
                    bucket,
                    region=region,
                ).get("Rules", [])
                enabled = [rule for rule in rules if rule.get("Status") == "Enabled"]
                audit["lifecycle_state"] = "enabled" if enabled else "disabled"
                audit["lifecycle_actions"] = sorted(
                    {
                        label
                        for rule in enabled
                        for key, label in (
                            ("Expiration", "expire-current"),
                            ("NoncurrentVersionExpiration", "expire-noncurrent"),
                            ("Transitions", "transition-current"),
                            ("NoncurrentVersionTransitions", "transition-noncurrent"),
                            ("AbortIncompleteMultipartUpload", "abort-multipart"),
                        )
                        if key in rule
                    }
                )
            except AWSCallError as exc:
                if "NoSuchLifecycleConfiguration" in str(exc):
                    audit["lifecycle_state"] = "none"
                else:
                    audit["audit_errors"].append(f"lifecycle: {exc}")
            return bucket, audit

        with ThreadPoolExecutor(max_workers=8) as executor:
            audits = dict(executor.map(audit_bucket, sorted(buckets)))
        return {
            "bucket_count": len(buckets),
            "buckets": [
                {
                    "bucket": bucket,
                    **buckets[bucket],
                    "bytes": (
                        int(values[bucket].to_integral_value()) if bucket in values else None
                    ),
                    "observed_at": observed.get(bucket),
                    **audits[bucket],
                }
                for bucket in sorted(buckets)
            ],
        }, []
    except (AWSCallError, KeyError, TypeError, ValueError) as exc:
        return {"bucket_count": 0, "buckets": []}, [f"object-storage: {exc}"]


def collect_deep_region(
    region: str, account_id: str, collected_at: datetime
) -> tuple[list[dict[str, Any]], list[str]]:
    rows: list[dict[str, Any]] = []
    gaps: list[str] = []

    def read(surface: str, loader: Callable[[], None]) -> None:
        try:
            loader()
        except (AWSCallError, KeyError, TypeError, ValueError) as exc:
            gaps.append(f"{region}:{surface}: {exc}")

    def kms() -> None:
        keys = aws_json("kms", "list-keys", "--page-size", "100", region=region).get(
            "Keys", []
        )
        aliases = aws_json(
            "kms", "list-aliases", "--page-size", "100", region=region
        ).get("Aliases", [])
        aliases_by_key: dict[str, list[str]] = collections.defaultdict(list)
        for item in aliases:
            if item.get("TargetKeyId") and item.get("AliasName"):
                aliases_by_key[item["TargetKeyId"]].append(item["AliasName"])
        unaliased = sorted(
            item["KeyId"] for item in keys if not aliases_by_key.get(item.get("KeyId", ""))
        )
        sampled = set(unaliased[:4])
        metadata = {
            key_id: aws_json("kms", "describe-key", "--key-id", key_id, region=region)[
                "KeyMetadata"
            ]
            for key_id in sampled
        }
        for item in keys:
            key_id = item["KeyId"]
            key_aliases = sorted(aliases_by_key.get(key_id, []))
            custom_aliases = [
                alias for alias in key_aliases if not alias.startswith("alias/aws/")
            ]
            key_metadata = metadata.get(key_id, {})
            rows.append(
                resource(
                    "kms-key",
                    region,
                    key_id,
                    name=custom_aliases[0] if custom_aliases else None,
                    alias_count=len(custom_aliases),
                    aws_managed=any(alias.startswith("alias/aws/") for alias in key_aliases),
                    sampled=key_id in sampled,
                    description_present=(
                        bool(key_metadata.get("Description")) if key_id in sampled else None
                    ),
                    description_kind=(
                        "account-service organization key"
                        if str(key_metadata.get("Description", "")).startswith(
                            "account dbuser for org:"
                        )
                        else (
                            "named DBaaS key"
                            if str(key_metadata.get("Description", ""))
                            .lower()
                            .startswith("cmk for dbaas ")
                            else (
                                "other description"
                                if key_metadata.get("Description")
                                else ("no description" if key_id in sampled else "not sampled")
                            )
                        )
                    ),
                    created=str(key_metadata.get("CreationDate", ""))[:10],
                    state=key_metadata.get("KeyState"),
                )
            )

    def private_ca() -> None:
        authorities = aws_json(
            "acm-pca", "list-certificate-authorities", region=region
        ).get("CertificateAuthorities", [])
        ca_rows: dict[str, dict[str, Any]] = {}
        for item in authorities:
            arn = item["Arn"]
            tags = decision_tags(
                aws_json(
                    "acm-pca",
                    "list-tags",
                    "--certificate-authority-arn",
                    arn,
                    region=region,
                ).get("Tags", [])
            )
            row = resource(
                "private-ca",
                region,
                arn,
                name=tags.get("Name"),
                owner=tags.get("cp9-owner"),
                session=tags.get("cp9-session"),
                created=str(item.get("CreatedAt", ""))[:10],
                state=item.get("Status"),
                subtype=item.get("UsageMode"),
                ca_type=item.get("Type"),
                recent_issue_events=0,
                recent_issue_callers=[],
            )
            rows.append(row)
            ca_rows[arn] = row
        if not ca_rows:
            return
        events = aws_json(
            "cloudtrail",
            "lookup-events",
            "--lookup-attributes",
            "AttributeKey=EventName,AttributeValue=IssueCertificate",
            "--start-time",
            (collected_at - timedelta(days=90)).isoformat(),
            "--end-time",
            collected_at.isoformat(),
            "--max-results",
            "50",
            "--no-paginate",
            region=region,
        ).get("Events", [])
        callers: dict[str, set[str]] = collections.defaultdict(set)
        event_counts: collections.Counter[str] = collections.Counter()
        for event in events:
            try:
                detail = json.loads(event.get("CloudTrailEvent") or "{}")
            except json.JSONDecodeError:
                continue
            request = detail.get("requestParameters") or {}
            arn = request.get("certificateAuthorityArn")
            if arn not in ca_rows:
                continue
            identity = detail.get("userIdentity") or {}
            caller = ((identity.get("sessionContext") or {}).get("sessionIssuer") or {}).get(
                "arn"
            ) or identity.get("arn")
            if isinstance(caller, str) and caller:
                callers[arn].add(caller)
            event_counts[arn] += 1
        for arn, row in ca_rows.items():
            row["recent_issue_events"] = event_counts[arn]
            row["recent_issue_callers"] = sorted(callers[arn])

    def ecr() -> None:
        repositories = aws_json(
            "ecr", "describe-repositories", "--page-size", "100", region=region
        ).get("repositories", [])

        if not repositories:
            return
        try:
            client = boto3.session.Session().client(
                "ecr",
                region_name=region,
                config=Config(
                    max_pool_connections=8,
                    connect_timeout=5,
                    read_timeout=30,
                    retries={"mode": "standard", "total_max_attempts": 3},
                ),
            )
        except BotoCoreError as exc:
            raise AWSCallError(f"ecr client: {type(exc).__name__}") from exc
        paginator = client.get_paginator("describe_images")
        print(
            f"cloud-cost-aws: {region} ECR images: {len(repositories)} repositories",
            file=sys.stderr,
            flush=True,
        )

        def images(item: dict[str, Any]) -> dict[str, Any]:
            row = resource(
                "ecr-repository",
                region,
                item["repositoryArn"],
                name=item.get("repositoryName"),
                created=str(item.get("createdAt", ""))[:10],
            )
            try:
                details = []
                for page in paginator.paginate(
                    repositoryName=item["repositoryName"],
                    filter={"imageStatus": "ANY"},
                    PaginationConfig={"PageSize": 1000},
                ):
                    page_details = page.get("imageDetails")
                    if not isinstance(page_details, list) or any(
                        not isinstance(image, dict)
                        or not isinstance(image.get("imageSizeInBytes"), int)
                        or isinstance(image["imageSizeInBytes"], bool)
                        or image["imageSizeInBytes"] < 0
                        for image in page_details
                    ):
                        raise AWSCallError("image listing omitted valid size evidence")
                    details.extend(page_details)
                row["image_count"] = len(details)
                row["image_bytes"] = sum(image["imageSizeInBytes"] for image in details)
                row["untagged_images"] = sum(not image.get("imageTags") for image in details)
                pushed = sorted(
                    str(image["imagePushedAt"])[:10]
                    for image in details
                    if image.get("imagePushedAt")
                )
                pulled = sorted(
                    str(image["lastRecordedPullTime"])[:10]
                    for image in details
                    if image.get("lastRecordedPullTime")
                )
                row["oldest_image"] = pushed[0] if pushed else None
                row["latest_pull"] = pulled[-1] if pulled else None
            except ClientError as exc:
                row["image_read_error"] = f"ecr describe-images: {exc.response['Error']['Code']}"
            except BotoCoreError as exc:
                row["image_read_error"] = f"ecr describe-images: {type(exc).__name__}"
            except AWSCallError as exc:
                row["image_read_error"] = str(exc)
            return row

        repository_rows = []
        with closing(client), ThreadPoolExecutor(max_workers=8) as executor:
            for row in executor.map(images, repositories):
                repository_rows.append(row)
                if len(repository_rows) % 500 == 0 or len(repository_rows) == len(repositories):
                    print(
                        f"cloud-cost-aws: {region} ECR images: "
                        f"{len(repository_rows)}/{len(repositories)} collected",
                        file=sys.stderr,
                        flush=True,
                    )
        # ponytail: policies for the largest ten observed repositories per region; widen only if needed.
        for row in sorted(
            repository_rows, key=lambda row: (-row.get("image_bytes", -1), row["id"])
        )[:10]:
            try:
                aws_json(
                    "ecr",
                    "get-lifecycle-policy",
                    "--repository-name",
                    row["name"],
                    region=region,
                )
                row["lifecycle_state"] = "present"
            except AWSCallError as exc:
                row["lifecycle_state"] = (
                    "none" if "LifecyclePolicyNotFoundException" in str(exc) else "unknown"
                )
        rows.extend(repository_rows)

    def opensearch() -> None:
        for item in aws_json("opensearch", "list-domain-names", region=region).get(
            "DomainNames", []
        ):
            rows.append(
                resource(
                    "opensearch-domain",
                    region,
                    item["DomainName"],
                    name=item["DomainName"],
                    subtype=item.get("EngineType"),
                )
            )

    def elasticache() -> None:
        for item in aws_json(
            "elasticache", "describe-cache-clusters", region=region
        ).get("CacheClusters", []):
            rows.append(
                resource(
                    "elasticache-cluster",
                    region,
                    item["CacheClusterId"],
                    weight=item.get("NumCacheNodes") or 1,
                    name=item["CacheClusterId"],
                    subtype=f"{item.get('Engine', 'unknown')} {item.get('CacheNodeType', 'unknown')}",
                    state=item.get("CacheClusterStatus"),
                )
            )

    def config() -> None:
        for item in aws_json(
            "configservice", "describe-configuration-recorders", region=region
        ).get("ConfigurationRecorders", []):
            rows.append(
                resource(
                    "config-recorder",
                    region,
                    item["name"],
                    name=item["name"],
                )
            )
        for item in aws_json(
            "configservice", "describe-conformance-packs", region=region
        ).get("ConformancePackDetails", []):
            rows.append(
                resource(
                    "config-conformance-pack",
                    region,
                    item["ConformancePackArn"],
                    name=item.get("ConformancePackName"),
                )
            )

    def quicksight() -> None:
        if region != "us-east-1":
            return
        for item in aws_json(
            "quicksight",
            "list-users",
            "--aws-account-id",
            account_id,
            "--namespace",
            "default",
            region=region,
        ).get("UserList", []):
            rows.append(
                resource(
                    "quicksight-user",
                    region,
                    item["Arn"],
                    name=item.get("UserName"),
                    subtype=item.get("Role"),
                    state=item.get("Active"),
                )
            )

    def vpn() -> None:
        for item in aws_json("ec2", "describe-vpn-connections", region=region).get(
            "VpnConnections", []
        ):
            rows.append(
                resource(
                    "vpn",
                    region,
                    item["VpnConnectionId"],
                    state=item.get("State"),
                    subtype=item.get("Type"),
                )
            )

    def vpc_lattice() -> None:
        for item in aws_json("vpc-lattice", "list-service-networks", region=region).get(
            "items", []
        ):
            rows.append(
                resource(
                    "vpc-lattice-service-network",
                    region,
                    item["arn"],
                    name=item.get("name"),
                    created=str(item.get("createdAt", ""))[:10],
                )
            )

    read("kms-key", kms)
    read("private-ca", private_ca)
    read("ecr-repository", ecr)
    read("opensearch-domain", opensearch)
    read("elasticache-cluster", elasticache)
    read("config", config)
    read("quicksight-user", quicksight)
    read("vpn", vpn)
    read("vpc-lattice-service-network", vpc_lattice)
    return rows, gaps


def collect(scope_id: str, account_id: str, billing_source: str) -> dict[str, Any]:
    caller = aws_json("sts", "get-caller-identity", region="us-east-1")
    if caller.get("Account") != account_id:
        raise AWSCallError(
            f"caller account {caller.get('Account')!r} does not match configured {account_id!r}"
        )

    collected_at = datetime.now(timezone.utc)
    windows = window_bounds(collected_at.date())
    gaps: list[str] = []
    print("cloud-cost-aws: collecting billing", file=sys.stderr, flush=True)
    billing, billing_gaps = collect_cost(windows, billing_source)
    gaps.extend(billing_gaps)
    print("cloud-cost-aws: collecting S3 metrics and bucket audits", file=sys.stderr, flush=True)
    object_storage, object_storage_gaps = collect_object_storage(collected_at)
    gaps.extend(object_storage_gaps)

    try:
        regions = enabled_regions()
    except AWSCallError as exc:
        regions = []
        gaps.append(f"regions: {exc}")

    rows: list[dict[str, Any]] = []
    evidence_gaps: list[str] = []
    if regions:
        print("cloud-cost-aws: collecting regional inventory", file=sys.stderr, flush=True)
        with ThreadPoolExecutor(max_workers=4) as executor:
            for region_rows, region_gaps in executor.map(collect_region, regions):
                rows.extend(region_rows)
                gaps.extend(region_gaps)
        print("cloud-cost-aws: collecting deep evidence", file=sys.stderr, flush=True)
        with ThreadPoolExecutor(max_workers=4) as executor:
            for region_rows, region_gaps in executor.map(
                lambda region: collect_deep_region(region, account_id, collected_at),
                regions,
            ):
                rows.extend(region_rows)
                evidence_gaps.extend(region_gaps)

    attach_cluster_ids(rows)

    observed_counts = collections.Counter(item["kind"] for item in rows)
    surfaces = INVENTORY_SURFACES + EVIDENCE_SURFACES
    counts = {surface: observed_counts[surface] for surface in surfaces}
    return {
        "version": VERSION,
        "scope_id": scope_id,
        "provider": "aws",
        "account_id": account_id,
        "billing_source": billing_source,
        "collected_at": collected_at.isoformat(),
        "windows": windows,
        "coverage": {"status": "incomplete" if gaps else "complete", "gaps": sorted(gaps)},
        "billing": billing,
        "inventory": {
            "baseline": "no_inventory_baseline",
            "regions": regions,
            "surfaces": list(surfaces),
            "evidence_coverage": {
                "status": "incomplete" if evidence_gaps else "complete",
                "gaps": sorted(evidence_gaps),
            },
            "counts": counts,
            "resources": sorted(rows, key=lambda item: (item["kind"], item["region"], item["id"])),
            "object_storage": object_storage,
        },
    }


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--scope-id", required=True)
    parser.add_argument("--account-id", required=True)
    parser.add_argument("--billing-source", choices=tuple(BILLING_METRICS), required=True)
    parser.add_argument("--output", type=Path, required=True)
    return parser.parse_args()


def check_evidence_reads() -> None:
    """Offline API-shape replay: keep read failures unknown and preserve ownership evidence."""
    global aws_json
    from unittest.mock import patch

    from botocore.exceptions import ReadTimeoutError
    from botocore.stub import Stubber

    original = aws_json
    denied = False

    def fake(*args: str, region: str | None = None) -> Any:
        operation = args[:2]
        if operation == ("s3api", "list-buckets"):
            return {"Buckets": [{"Name": "sample", "BucketRegion": "us-west-2"}]}
        if operation == ("s3api", "list-object-versions"):
            if denied:
                raise AWSCallError("AccessDenied")
            return {
                "Name": "sample",
                "Versions": [],
                "DeleteMarkers": [],
                "IsTruncated": False,
            }
        if operation == ("s3api", "get-bucket-lifecycle-configuration"):
            raise AWSCallError("AccessDenied" if denied else "NoSuchLifecycleConfiguration")
        if operation == ("kms", "list-keys"):
            return {"Keys": [{"KeyId": "sample"}]}
        if operation == ("kms", "describe-key"):
            return {
                "KeyMetadata": {
                    "Description": "account dbuser for org:example",
                    "CreationDate": "2021-01-01",
                    "KeyState": "Enabled",
                }
            }
        if operation == ("ecr", "describe-repositories"):
            return {
                "repositories": [
                    {
                        "repositoryArn": "arn:aws:ecr:us-west-2:123456789012:repository/sample",
                        "repositoryName": "sample",
                    }
                ]
            }
        if operation == ("ecr", "get-lifecycle-policy"):
            raise AWSCallError("LifecyclePolicyNotFoundException")
        return {}

    client = boto3.session.Session().client(
        "ecr",
        region_name="us-west-2",
        aws_access_key_id="offline",
        aws_secret_access_key="offline",
    )
    request = {"repositoryName": "sample", "filter": {"imageStatus": "ANY"}, "maxResults": 1000}
    first_page = {
        "imageDetails": [{"imageSizeInBytes": 100, "imageTags": ["tag"]}],
        "nextToken": "next",
    }
    stubber = Stubber(client)
    stubber.add_response("describe_images", first_page, request)
    stubber.add_response(
        "describe_images",
        {"imageDetails": [{"imageSizeInBytes": 20}]},
        {**request, "nextToken": "next"},
    )
    stubber.add_response("describe_images", first_page, request)
    stubber.add_client_error(
        "describe_images",
        service_error_code="AccessDeniedException",
        expected_params={**request, "nextToken": "next"},
    )
    stubber.add_response("describe_images", {"imageDetails": [{"imageTags": ["tag"]}]}, request)
    stubber.activate()
    client_patch = patch("boto3.session.Session.client", return_value=client)
    client_patch.start()
    try:
        aws_json = fake
        storage, gaps = collect_object_storage(datetime(2026, 1, 20, tzinfo=timezone.utc))
        bucket = storage["buckets"][0]
        assert not gaps and bucket["empty"] is True and bucket["lifecycle_state"] == "none"
        rows, gaps = collect_deep_region(
            "us-west-2", "123456789012", datetime(2026, 1, 20, tzinfo=timezone.utc)
        )
        assert not gaps
        repository = next(item for item in rows if item["kind"] == "ecr-repository")
        assert (
            repository["image_bytes"] == 120
            and repository["image_count"] == 2
            and repository["untagged_images"] == 1
        )
        assert (
            next(item for item in rows if item["kind"] == "kms-key")["description_kind"]
            == "account-service organization key"
        )
        denied = True
        storage, gaps = collect_object_storage(datetime(2026, 1, 20, tzinfo=timezone.utc))
        assert (
            storage["buckets"][0]["empty"] is None
            and storage["buckets"][0]["lifecycle_state"] == "unknown"
        )
        rows, gaps = collect_deep_region(
            "us-west-2", "123456789012", datetime(2026, 1, 20, tzinfo=timezone.utc)
        )
        repository = next(item for item in rows if item["kind"] == "ecr-repository")
        assert "image_count" not in repository and "image_read_error" in repository
        denied = False
        rows, _ = collect_deep_region(
            "us-west-2", "123456789012", datetime(2026, 1, 20, tzinfo=timezone.utc)
        )
        repository = next(item for item in rows if item["kind"] == "ecr-repository")
        assert "image_count" not in repository and "image_read_error" in repository
        stubber.assert_no_pending_responses()
        with patch.object(
            client, "_make_api_call", side_effect=ReadTimeoutError(endpoint_url="offline")
        ):
            rows, _ = collect_deep_region(
                "us-west-2", "123456789012", datetime(2026, 1, 20, tzinfo=timezone.utc)
            )
        repository = next(item for item in rows if item["kind"] == "ecr-repository")
        assert "image_count" not in repository and "ReadTimeoutError" in repository["image_read_error"]
        tagged = resource(
            "ebs",
            "us-west-2",
            "vol-test",
            tags=[{"Key": "cp9-owner", "Value": "owner"}],
        )
        assert tagged["owner"] == "owner" and "cp9-owner" not in tagged
    finally:
        aws_json = original
        client_patch.stop()
        stubber.deactivate()
        client.close()


def main() -> int:
    if sys.argv[1:] == ["check"]:
        check_evidence_reads()
        rows = bucket_metric_values(
            [("bucket-a", "StandardStorage"), ("bucket-a", "GlacierStorage")],
            [
                {
                    "Id": "m0",
                    "StatusCode": "Complete",
                    "Values": [100.0],
                    "Timestamps": ["2026-01-01T00:00:00+00:00"],
                },
                {
                    "Id": "m1",
                    "StatusCode": "Complete",
                    "Values": [25.0],
                    "Timestamps": ["2026-01-01T00:00:00+00:00"],
                },
            ],
        )
        if sum((amount for _bucket, amount, _timestamp in rows), Decimal(0)) != 125:
            print("cloud-cost-aws: bucket metric check failed", file=sys.stderr)
            return 1
        if decision_tags(
            [
                {"Key": "Name", "Value": "demo"},
                {"Key": "cp9-owner", "Value": "owner"},
                {"Key": "secret", "Value": "not-retained"},
            ]
        ) != {"Name": "demo", "cp9-owner": "owner"}:
            print("cloud-cost-aws: decision tag check failed", file=sys.stderr)
            return 1
        inventory = [
            vpc_resource(
                "us-west-2",
                {
                    "VpcId": "vpc-test",
                    "State": "available",
                    "Tags": [{"Key": "Name", "Value": "test-vpc"}],
                },
            ),
            resource("eks", "us-west-2", "cluster-test", vpc_id="vpc-test"),
        ]
        attach_cluster_ids(inventory)
        if inventory[0]["cluster_ids"] != ["cluster-test"]:
            print("cloud-cost-aws: VPC cluster association check failed", file=sys.stderr)
            return 1
        print("cloud cost AWS collector check: ok")
        return 0
    args = parse_args()
    if not SCOPE_ID.fullmatch(args.scope_id) or not AWS_ACCOUNT.fullmatch(args.account_id):
        print("cloud-cost-aws: invalid scope or account id", file=sys.stderr)
        return 1
    try:
        snapshot = collect(args.scope_id, args.account_id, args.billing_source)
        args.output.parent.mkdir(parents=True, exist_ok=True)
        args.output.write_text(json.dumps(snapshot, indent=2, sort_keys=True) + "\n")
    except (AWSCallError, OSError) as exc:
        print(f"cloud-cost-aws: {exc}", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())