#!/usr/bin/env python3
"""Validate, normalize, and render synthetic or live cloud-cost analysis windows.

Run `check --scopes .github/cloud-cost-scopes.json --fixture tools/cloud-cost-report-fixture.json`
offline before changing billing allocation or publication. The retained replay catches
cross-service usage collisions, cent closure, incomplete reads, invented resource diffs,
ungrounded editor output and candidate-history corruption; none needs cloud credentials.
"""

from __future__ import annotations

import argparse
import collections
import copy
import json
import re
import sys
from datetime import date, datetime, timedelta, timezone
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Any


MANIFEST_VERSION = "cp9.cloud-cost-scopes/v1"
FIXTURE_VERSION = "cp9.cloud-cost-replay/v1"
AWS_SNAPSHOT_VERSION = "cp9.cloud-cost-aws-snapshot/v3"
AWS_LEGACY_SNAPSHOT_VERSION = "cp9.cloud-cost-aws-snapshot/v2"
ANALYSIS_VERSION = "cp9.cloud-cost-analysis/v5"
PROVIDERS = ("aws", "gcp", "aliyun", "azure")
BILLING_SOURCES = {"invoice", "amortized", "unblended", "list_price", "unpriced"}
AWS_BILLING_METRICS = {
    "amortized": ("NetAmortizedCost", "NetUnblendedCost"),
    "unblended": ("NetUnblendedCost", "NetAmortizedCost"),
}
AWS_INVENTORY_SURFACES = (
    "vpc",
    "eks",
    "compute",
    "ebs",
    "nat",
    "loadbalancer",
    "rds",
    "vpc-endpoint",
    "public-ipv4",
    "msk",
)
AWS_CONFIG_EVIDENCE_SURFACES = ("config-recorder", "config-conformance-pack")
AWS_EVIDENCE_SURFACES = (
    "kms-key",
    "private-ca",
    "ecr-repository",
    "opensearch-domain",
    "elasticache-cluster",
    *AWS_CONFIG_EVIDENCE_SURFACES,
    "quicksight-user",
    "vpn",
    "vpc-lattice-service-network",
)
AWS_EVIDENCE_GAP_KINDS = {
    **{
        surface: (surface,)
        for surface in AWS_EVIDENCE_SURFACES
        if surface not in AWS_CONFIG_EVIDENCE_SURFACES
    },
    "config": AWS_CONFIG_EVIDENCE_SURFACES,
}
AWS_REQUIRED_SURFACES = AWS_INVENTORY_SURFACES + AWS_EVIDENCE_SURFACES
AWS_ACCOUNT_SERVICE_KINDS = AWS_EVIDENCE_SURFACES[2:]
AWS_REPORT_SECTION_PREFIXES = (
    "# 1. Summary",
    "# 2. Grouped by VPC and EKS cluster (sums to 100%)",
    "# 3. Cost belonging to no VPC",
    "# 4. Account-level services",
    "# 5. Reclaim candidates",
    "# 6. Resource changes and recurrence",
    "# 7. Action plan",
)
AWS_REPORT_SUBSECTION_PREFIXES = (
    "## Candidate validation",
    "## Workload and retention review",
    "## Evidence to resolve before cleanup",
)
AWS_REPORT_RAW_SCHEMA_TERMS = (
    "current_total",
    "previous_total",
    "reclaim_candidate_count",
    "coverage.status",
    "new_count",
    "removed_count",
    "confirmed_count",
    "no_inventory_baseline",
)
REPORT_INTERPRETATION_KEYS = tuple(str(index) for index in range(1, 11))
AWS_REPORT_GENERATED_MARKER = "Generated by the controlplane9 cloud-cost-report workflow."
SCOPE_ID = re.compile(r"^[a-z0-9][a-z0-9-]{2,79}$")
AWS_ACCOUNT = re.compile(r"^[0-9]{12}$")
ALIYUN_ACCOUNT = re.compile(r"^[0-9]{16}$")
GCP_PROJECT = re.compile(r"^[a-z][a-z0-9-]{4,28}[a-z0-9]$")
AZURE_SUBSCRIPTION = re.compile(
    r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$"
)
SCOPE_KEYS = {
    "id",
    "label",
    "provider",
    "billing_source",
    "enabled",
    "identity",
    "account_id",
    "project_id",
    "account_name",
    "subscription_id",
}
REQUIRED_SCOPE_KEYS = {"id", "label", "provider", "billing_source", "enabled", "identity"}


class ContractError(ValueError):
    pass


def load_json(path: Path) -> dict[str, Any]:
    try:
        value = json.loads(path.read_text(), parse_float=Decimal)
    except (OSError, json.JSONDecodeError) as exc:
        raise ContractError(f"cannot read {path}: {exc}") from exc
    if not isinstance(value, dict):
        raise ContractError(f"{path} must contain a JSON object")
    return value


def require_keys(value: dict[str, Any], expected: set[str], where: str) -> None:
    extra = set(value) - expected
    if extra:
        raise ContractError(f"{where} has unknown fields: {', '.join(sorted(extra))}")


def validate_manifest(manifest: dict[str, Any]) -> dict[str, dict[str, Any]]:
    require_keys(manifest, {"version", "scopes"}, "scope manifest")
    if manifest.get("version") != MANIFEST_VERSION:
        raise ContractError(f"scope manifest version must be {MANIFEST_VERSION!r}")
    scopes = manifest.get("scopes")
    if not isinstance(scopes, list) or not scopes:
        raise ContractError("scope manifest must contain a non-empty scopes array")

    by_id: dict[str, dict[str, Any]] = {}
    targets: set[tuple[str, str]] = set()
    for index, scope in enumerate(scopes):
        where = f"scope[{index}]"
        if not isinstance(scope, dict):
            raise ContractError(f"{where} must be an object")
        require_keys(scope, SCOPE_KEYS, where)
        missing = REQUIRED_SCOPE_KEYS - set(scope)
        if missing:
            raise ContractError(f"{where} is missing fields: {', '.join(sorted(missing))}")

        scope_id = scope.get("id")
        if not isinstance(scope_id, str) or not SCOPE_ID.fullmatch(scope_id):
            raise ContractError(f"{where}.id must be a lowercase DNS-style identifier")
        if scope_id in by_id:
            raise ContractError(f"duplicate scope id: {scope_id}")

        provider = scope.get("provider")
        if provider not in PROVIDERS:
            raise ContractError(f"{where}.provider must be one of {', '.join(PROVIDERS)}")
        if not isinstance(scope.get("label"), str) or not scope["label"].strip():
            raise ContractError(f"{where}.label must be a non-empty string")
        if scope.get("billing_source") not in BILLING_SOURCES:
            raise ContractError(f"{where}.billing_source is unsupported")
        if not isinstance(scope.get("enabled"), bool):
            raise ContractError(f"{where}.enabled must be boolean")
        identity = scope["identity"]
        if identity is not None and (
            not isinstance(identity, dict)
            or not identity
            or not all(
                isinstance(key, str)
                and key.strip()
                and isinstance(item, str)
                and item.strip()
                for key, item in identity.items()
            )
        ):
            raise ContractError(
                f"{where}.identity must contain non-empty string fields when set"
            )
        if scope["enabled"] and identity is None:
            raise ContractError(f"{where}.identity is required before enabling the scope")

        target = validate_target(scope, where)
        target_key = (provider, target)
        if target_key in targets:
            raise ContractError(f"duplicate {provider} target: {target}")
        targets.add(target_key)
        by_id[scope_id] = scope
    missing_providers = set(PROVIDERS) - {scope["provider"] for scope in scopes}
    if missing_providers:
        raise ContractError(
            f"scope manifest is missing providers: {', '.join(sorted(missing_providers))}"
        )
    return by_id


def validate_target(scope: dict[str, Any], where: str) -> str:
    provider = scope["provider"]
    fields = {
        "aws": (("account_id", AWS_ACCOUNT),),
        "gcp": (("project_id", GCP_PROJECT),),
        "aliyun": (("account_name", None), ("account_id", ALIYUN_ACCOUNT)),
        "azure": (("subscription_id", AZURE_SUBSCRIPTION),),
    }
    wanted = fields[provider]
    target_names = {name for choices in fields.values() for name, _ in choices}
    present = {name for name in target_names if name in scope}
    if present != {name for name, _ in wanted}:
        names = ", ".join(repr(name) for name, _ in wanted)
        raise ContractError(f"{where} must define only provider target fields {names}")
    for name, pattern in wanted:
        target = scope[name]
        if not isinstance(target, str) or not target.strip() or target != target.strip():
            raise ContractError(f"{where}.{name} must be a non-empty string")
        if pattern and not pattern.fullmatch(target):
            raise ContractError(f"{where}.{name} has an invalid format")
    return scope[wanted[0][0]]


def parse_window(value: Any, where: str) -> tuple[date, date]:
    if not isinstance(value, dict) or set(value) != {"start", "end"}:
        raise ContractError(f"{where} must contain only start and end")
    try:
        start = date.fromisoformat(value["start"])
        end = date.fromisoformat(value["end"])
    except (TypeError, ValueError) as exc:
        raise ContractError(f"{where} dates must use YYYY-MM-DD") from exc
    if end - start != timedelta(days=7):
        raise ContractError(f"{where} must be exactly seven days")
    return start, end


def money(value: Any, where: str) -> Decimal:
    try:
        amount = Decimal(str(value))
    except (InvalidOperation, ValueError) as exc:
        raise ContractError(f"{where} must be a decimal amount") from exc
    if not amount.is_finite() or amount < 0:
        raise ContractError(f"{where} must be finite and non-negative")
    return amount


def money_text(value: Decimal) -> str:
    return f"{value.quantize(Decimal('0.01')):.2f}"


def count_text(value: int, singular: str, plural: str | None = None) -> str:
    return f"{value} {singular if value == 1 else plural or singular + 's'}"


def percent_text(part: Decimal, total: Decimal) -> str:
    if total == 0:
        return "0.00"
    return f"{(part * Decimal(100) / total).quantize(Decimal('0.01')):.2f}"


def change_percent_text(delta: Decimal, previous: Decimal) -> str:
    return "n/a" if previous == 0 else percent_text(delta, previous)


def named_costs(value: Any, where: str) -> dict[str, Decimal]:
    if not isinstance(value, list):
        raise ContractError(f"{where} must be an array")
    out: dict[str, Decimal] = {}
    for index, row in enumerate(value):
        row_where = f"{where}[{index}]"
        if not isinstance(row, dict) or set(row) != {"name", "cost"}:
            raise ContractError(f"{row_where} must contain only name and cost")
        name = row["name"]
        if not isinstance(name, str) or not name.strip():
            raise ContractError(f"{row_where}.name must be non-empty")
        if name in out:
            raise ContractError(f"{where} has duplicate name {name!r}")
        out[name] = money(row["cost"], f"{row_where}.cost")
    return out


def validate_candidates(value: Any, where: str) -> list[dict[str, Any]]:
    if not isinstance(value, list):
        raise ContractError(f"{where} must be an array")
    out: list[dict[str, Any]] = []
    seen: set[str] = set()
    expected = {"id", "classification", "signals", "evidence", "estimated_monthly_cost"}
    for index, row in enumerate(value):
        row_where = f"{where}[{index}]"
        if not isinstance(row, dict) or set(row) != expected:
            raise ContractError(f"{row_where} has an invalid candidate shape")
        candidate_id = row["id"]
        if (
            not isinstance(candidate_id, str)
            or not candidate_id.strip()
            or candidate_id != candidate_id.strip()
            or candidate_id in seen
        ):
            raise ContractError(f"{row_where}.id must be unique and non-empty")
        seen.add(candidate_id)
        classification = row["classification"]
        if classification not in {"candidate", "confirmed"}:
            raise ContractError(f"{row_where}.classification is unsupported")
        signals = row["signals"]
        evidence = row["evidence"]
        if (
            not isinstance(signals, list)
            or not signals
            or not all(isinstance(item, str) and item.strip() for item in signals)
        ):
            raise ContractError(f"{row_where}.signals must contain non-empty strings")
        if (
            not isinstance(evidence, list)
            or not evidence
            or not all(isinstance(item, str) and item.strip() for item in evidence)
        ):
            raise ContractError(f"{row_where}.evidence must contain non-empty strings")
        if classification == "confirmed" and len(set(evidence)) < 2:
            raise ContractError(f"{row_where} needs at least two independent evidence entries")
        out.append(
            {
                **row,
                "estimated_monthly_cost": money_text(
                    money(row["estimated_monthly_cost"], f"{row_where}.estimated_monthly_cost")
                ),
            }
        )
    return out


def validate_window_data(value: Any, where: str) -> dict[str, Any]:
    expected = {
        "total_cost",
        "groups",
        "unattributed_cost",
        "other_services",
        "resources",
        "reclaim_candidates",
    }
    if not isinstance(value, dict) or set(value) != expected:
        raise ContractError(f"{where} has an invalid window shape")
    total = money(value["total_cost"], f"{where}.total_cost")
    groups = named_costs(value["groups"], f"{where}.groups")
    unattributed = money(value["unattributed_cost"], f"{where}.unattributed_cost")
    other = named_costs(value["other_services"], f"{where}.other_services")
    if sum(groups.values(), Decimal(0)) + unattributed + sum(other.values(), Decimal(0)) != total:
        raise ContractError(f"{where} groups, unattributed cost, and other services must sum to total_cost")
    resources = value["resources"]
    if not isinstance(resources, list) or not all(
        isinstance(item, str) and item.strip() for item in resources
    ):
        raise ContractError(f"{where}.resources must contain non-empty strings")
    if len(resources) != len(set(resources)):
        raise ContractError(f"{where}.resources must be unique")
    return {
        "total": total,
        "groups": groups,
        "unattributed": unattributed,
        "other": other,
        "resources": set(resources),
        "candidates": validate_candidates(value["reclaim_candidates"], f"{where}.reclaim_candidates"),
    }


def cost_delta(previous: dict[str, Decimal], current: dict[str, Decimal], total: Decimal) -> list[dict[str, str]]:
    return [
        {
            "name": name,
            "previous": money_text(previous.get(name, Decimal(0))),
            "current": money_text(current.get(name, Decimal(0))),
            "delta": money_text(current.get(name, Decimal(0)) - previous.get(name, Decimal(0))),
            "current_share_percent": percent_text(current.get(name, Decimal(0)), total),
        }
        for name in sorted(set(previous) | set(current))
    ]


def build_analysis(manifest: dict[str, Any], fixture: dict[str, Any], scenario_name: str) -> dict[str, Any]:
    scopes = validate_manifest(manifest)
    require_keys(fixture, {"version", "synthetic", "windows", "providers", "scenarios"}, "fixture")
    if fixture.get("version") != FIXTURE_VERSION or fixture.get("synthetic") is not True:
        raise ContractError("replay input must be the declared synthetic v1 fixture")

    windows = fixture.get("windows")
    if not isinstance(windows, dict) or set(windows) != {"previous", "current"}:
        raise ContractError("fixture.windows must contain previous and current")
    previous_start, previous_end = parse_window(windows["previous"], "fixture.windows.previous")
    current_start, current_end = parse_window(windows["current"], "fixture.windows.current")
    if previous_end != current_start:
        raise ContractError("fixture windows must be adjacent")

    scenarios = fixture.get("scenarios")
    if not isinstance(scenarios, dict) or scenario_name not in scenarios:
        raise ContractError(f"unknown replay scenario {scenario_name!r}")
    scenario = scenarios[scenario_name]
    if not isinstance(scenario, dict) or set(scenario) != {"incomplete_providers"}:
        raise ContractError(f"scenario {scenario_name!r} has an invalid shape")
    incomplete = scenario["incomplete_providers"]
    if not isinstance(incomplete, dict) or not set(incomplete).issubset(PROVIDERS):
        raise ContractError(f"scenario {scenario_name!r} has invalid incomplete providers")
    for provider, gaps in incomplete.items():
        if (
            not isinstance(gaps, list)
            or not gaps
            or not all(isinstance(gap, str) and gap.strip() for gap in gaps)
        ):
            raise ContractError(f"scenario gaps for {provider} must contain non-empty strings")

    providers = fixture.get("providers")
    if not isinstance(providers, dict) or set(providers) != set(PROVIDERS):
        raise ContractError(f"fixture.providers must contain exactly {', '.join(PROVIDERS)}")

    provider_output: dict[str, Any] = {}
    for provider in PROVIDERS:
        raw = providers[provider]
        expected = {"currency", "windows"}
        if not isinstance(raw, dict) or set(raw) != expected:
            raise ContractError(f"fixture.providers.{provider} has an invalid shape")
        provider_scopes = [
            {
                "id": scope_id,
                "label": scope["label"],
                "target": validate_target(scope, f"scope {scope_id!r}"),
                "billing_source": scope["billing_source"],
                "enabled": scope["enabled"],
            }
            for scope_id, scope in sorted(scopes.items())
            if scope["provider"] == provider
        ]
        currency = raw["currency"]
        if not isinstance(currency, str) or not re.fullmatch(r"[A-Z]{3}", currency):
            raise ContractError(f"fixture provider {provider} currency must be a three-letter code")

        if provider in incomplete:
            gaps = incomplete[provider]
            provider_output[provider] = {
                "scopes": provider_scopes,
                "currency": currency,
                "coverage": {"status": "incomplete", "gaps": gaps},
            }
            continue

        raw_windows = raw["windows"]
        if not isinstance(raw_windows, dict) or set(raw_windows) != {"previous", "current"}:
            raise ContractError(f"fixture provider {provider} must contain two windows")
        previous = validate_window_data(raw_windows["previous"], f"fixture.providers.{provider}.windows.previous")
        current = validate_window_data(raw_windows["current"], f"fixture.providers.{provider}.windows.current")
        total_delta = current["total"] - previous["total"]
        provider_output[provider] = {
            "scopes": provider_scopes,
            "currency": currency,
            "coverage": {"status": "complete", "gaps": []},
            "cost": {
                "previous_total": money_text(previous["total"]),
                "current_total": money_text(current["total"]),
                "delta": money_text(total_delta),
                "delta_percent": change_percent_text(total_delta, previous["total"]),
                "groups": cost_delta(previous["groups"], current["groups"], current["total"]),
                "unattributed": {
                    "previous": money_text(previous["unattributed"]),
                    "current": money_text(current["unattributed"]),
                    "delta": money_text(current["unattributed"] - previous["unattributed"]),
                    "current_share_percent": percent_text(current["unattributed"], current["total"]),
                },
                "other_services": cost_delta(previous["other"], current["other"], current["total"]),
            },
            "resources": {
                "new": sorted(current["resources"] - previous["resources"]),
                "removed": sorted(previous["resources"] - current["resources"]),
            },
            "reclaim_candidates": current["candidates"],
        }

    return {
        "version": ANALYSIS_VERSION,
        "synthetic": True,
        "mode": "synthetic-replay",
        "report_title": "SYNTHETIC REPLAY: Multi-Cloud Cost Report",
        "scenario": scenario_name,
        "windows": {
            "previous": {"start": previous_start.isoformat(), "end": previous_end.isoformat()},
            "current": {"start": current_start.isoformat(), "end": current_end.isoformat()},
        },
        "decision_units": synthetic_decision_units(provider_output),
        "providers": provider_output,
    }


def aws_resource_key(row: dict[str, Any]) -> str:
    return f"{row['region']}/{row['kind']}/{row['id']}"


def validate_aws_snapshot(
    snapshot: dict[str, Any],
    scope: dict[str, Any],
    where: str,
    *,
    allow_legacy: bool = False,
) -> None:
    require_keys(
        snapshot,
        {
            "version",
            "scope_id",
            "provider",
            "account_id",
            "billing_source",
            "collected_at",
            "windows",
            "coverage",
            "billing",
            "inventory",
        },
        where,
    )
    legacy = snapshot.get("version") == AWS_LEGACY_SNAPSHOT_VERSION
    if snapshot.get("version") != AWS_SNAPSHOT_VERSION and not (allow_legacy and legacy):
        raise ContractError(f"{where}.version must be {AWS_SNAPSHOT_VERSION!r}")
    if (
        snapshot.get("scope_id") != scope["id"]
        or snapshot.get("provider") != "aws"
        or snapshot.get("account_id") != scope["account_id"]
        or snapshot.get("billing_source") != scope["billing_source"]
    ):
        raise ContractError(f"{where} identity does not match the selected scope")
    try:
        if datetime.fromisoformat(snapshot["collected_at"]).tzinfo is None:
            raise ValueError("timestamp needs a timezone")
    except (TypeError, ValueError) as exc:
        raise ContractError(f"{where}.collected_at must be ISO-8601") from exc

    windows = snapshot.get("windows")
    if not isinstance(windows, dict) or set(windows) != {"previous", "current"}:
        raise ContractError(f"{where}.windows must contain previous and current")
    previous_start, previous_end = parse_window(windows["previous"], f"{where}.windows.previous")
    current_start, current_end = parse_window(windows["current"], f"{where}.windows.current")
    if previous_end != current_start:
        raise ContractError(f"{where}.windows must be adjacent")

    coverage = snapshot.get("coverage")
    if not isinstance(coverage, dict) or set(coverage) != {"status", "gaps"}:
        raise ContractError(f"{where}.coverage has an invalid shape")
    if coverage.get("status") not in {"complete", "incomplete"}:
        raise ContractError(f"{where}.coverage.status is unsupported")
    gaps = coverage.get("gaps")
    if not isinstance(gaps, list) or not all(
        isinstance(gap, str) and gap.strip() for gap in gaps
    ):
        raise ContractError(f"{where}.coverage.gaps must contain non-empty strings")
    if (coverage["status"] == "complete") != (not gaps):
        raise ContractError(f"{where}.coverage status and gaps disagree")

    billing = snapshot.get("billing")
    billing_fields = {
        "metric",
        "comparison_metric",
        "page_count",
        "days",
        "service_usage_page_count",
        "service_usage_days",
    }
    if not legacy:
        billing_fields |= {"history", "savings_plans"}
    if not isinstance(billing, dict) or set(billing) != billing_fields:
        raise ContractError(f"{where}.billing has an invalid shape")
    expected_metrics = AWS_BILLING_METRICS.get(scope["billing_source"])
    if not expected_metrics or tuple(
        billing.get(key) for key in ("metric", "comparison_metric")
    ) != expected_metrics:
        raise ContractError(f"{where}.billing metrics do not match billing_source")
    expected_starts = [
        (previous_start + timedelta(days=offset)).isoformat() for offset in range(14)
    ]

    def validate_billing_days(days_key: str, page_count_key: str, dimensions: str) -> None:
        page_count = billing.get(page_count_key)
        if not isinstance(page_count, int) or isinstance(page_count, bool) or page_count < 0:
            raise ContractError(f"{where}.billing.{page_count_key} must be non-negative")
        days = billing.get(days_key)
        if not isinstance(days, list):
            raise ContractError(f"{where}.billing.{days_key} must be an array")
        starts: list[str] = []
        for day_index, item in enumerate(days):
            day_where = f"{where}.billing.{days_key}[{day_index}]"
            if not isinstance(item, dict) or set(item) != {"TimePeriod", "Estimated", "Groups"}:
                raise ContractError(f"{day_where} has an invalid shape")
            try:
                day_start = date.fromisoformat(item["TimePeriod"]["Start"])
                day_end = date.fromisoformat(item["TimePeriod"]["End"])
            except (KeyError, TypeError, ValueError) as exc:
                raise ContractError(f"{day_where}.TimePeriod is invalid") from exc
            if day_end - day_start != timedelta(days=1):
                raise ContractError(f"{day_where} must cover one day")
            starts.append(day_start.isoformat())
            if not isinstance(item["Estimated"], bool) or not isinstance(item["Groups"], list):
                raise ContractError(f"{day_where} Estimated or Groups is invalid")
            for group_index, group in enumerate(item["Groups"]):
                group_where = f"{day_where}.Groups[{group_index}]"
                if not isinstance(group, dict) or set(group) != {"Keys", "Metrics"}:
                    raise ContractError(f"{group_where} has an invalid shape")
                if (
                    not isinstance(group["Keys"], list)
                    or len(group["Keys"]) != 2
                    or not all(isinstance(key, str) for key in group["Keys"])
                ):
                    raise ContractError(f"{group_where}.Keys must contain {dimensions}")
                metrics = group["Metrics"]
                expected = {billing["metric"], billing["comparison_metric"], "UsageQuantity"}
                if not isinstance(metrics, dict) or set(metrics) != expected:
                    raise ContractError(f"{group_where}.Metrics has an invalid shape")
                for metric_name, metric in metrics.items():
                    if not isinstance(metric, dict) or set(metric) != {"Amount", "Unit"}:
                        raise ContractError(f"{group_where}.Metrics.{metric_name} is invalid")
                    money(metric["Amount"], f"{group_where}.Metrics.{metric_name}.Amount")
                    if not isinstance(metric["Unit"], str):
                        raise ContractError(f"{group_where}.Metrics.{metric_name}.Unit is invalid")
                if any(
                    metrics[name]["Unit"] != "USD" for name in expected - {"UsageQuantity"}
                ):
                    raise ContractError(f"{group_where} cost currency must be USD")
        if coverage["status"] == "complete" and (
            starts != expected_starts or any(item["Estimated"] for item in days)
        ):
            raise ContractError(
                f"{where} complete {days_key} must contain 14 exact non-estimated days"
            )

    validate_billing_days("days", "page_count", "region and usage type")
    validate_billing_days(
        "service_usage_days", "service_usage_page_count", "service and usage type"
    )

    if not legacy:
        history = billing["history"]
        if not isinstance(history, dict) or set(history) != {"start", "end", "days"}:
            raise ContractError(f"{where}.billing.history has an invalid shape")
        try:
            history_start = date.fromisoformat(history["start"])
            history_end = date.fromisoformat(history["end"])
        except (TypeError, ValueError) as exc:
            raise ContractError(f"{where}.billing.history dates are invalid") from exc
        if history_end != current_end or history_end - history_start != timedelta(days=56):
            raise ContractError(f"{where}.billing.history must end at the report window after 56 days")
        history_days = history["days"]
        if not isinstance(history_days, list):
            raise ContractError(f"{where}.billing.history.days must be an array")
        history_starts = []
        for index, item in enumerate(history_days):
            item_where = f"{where}.billing.history.days[{index}]"
            if not isinstance(item, dict) or set(item) != {
                "TimePeriod",
                "Estimated",
                "Total",
                "Groups",
            }:
                raise ContractError(f"{item_where} has an invalid shape")
            try:
                item_start = date.fromisoformat(item["TimePeriod"]["Start"])
                item_end = date.fromisoformat(item["TimePeriod"]["End"])
            except (KeyError, TypeError, ValueError) as exc:
                raise ContractError(f"{item_where}.TimePeriod is invalid") from exc
            if item_end - item_start != timedelta(days=1):
                raise ContractError(f"{item_where} must cover one day")
            history_starts.append(item_start.isoformat())
            if not isinstance(item.get("Estimated"), bool) or not isinstance(
                item.get("Total"), dict
            ):
                raise ContractError(f"{item_where} Estimated or Total is invalid")
            for metric_name in (billing["metric"], billing["comparison_metric"]):
                metric = item["Total"].get(metric_name)
                if not isinstance(metric, dict) or set(metric) != {"Amount", "Unit"}:
                    raise ContractError(f"{item_where}.Total.{metric_name} is invalid")
                money(metric["Amount"], f"{item_where}.Total.{metric_name}.Amount")
                if metric["Unit"] != "USD":
                    raise ContractError(f"{item_where}.Total.{metric_name} must use USD")
        expected_history_starts = [
            (history_start + timedelta(days=offset)).isoformat() for offset in range(56)
        ]
        if coverage["status"] == "complete" and (
            history_starts != expected_history_starts
            or any(item["Estimated"] for item in history_days)
        ):
            raise ContractError(
                f"{where} complete billing history must contain 56 exact non-estimated days"
            )
        if coverage["status"] == "complete":
            for window_name, window_start, window_end in (
                ("previous", previous_start, previous_end),
                ("current", current_start, current_end),
            ):
                for metric_name in (billing["metric"], billing["comparison_metric"]):
                    grouped_total = sum(
                        (
                            money(
                                group["Metrics"][metric_name]["Amount"],
                                f"{where}.billing.days:{window_name}:{metric_name}",
                            )
                            for item in billing["days"]
                            if window_start
                            <= date.fromisoformat(item["TimePeriod"]["Start"])
                            < window_end
                            for group in item["Groups"]
                        ),
                        Decimal(0),
                    )
                    history_total = sum(
                        (
                            money(
                                item["Total"][metric_name]["Amount"],
                                f"{where}.billing.history:{window_name}:{metric_name}",
                            )
                            for item in history_days
                            if window_start
                            <= date.fromisoformat(item["TimePeriod"]["Start"])
                            < window_end
                        ),
                        Decimal(0),
                    )
                    if grouped_total.quantize(Decimal("0.01")) != history_total.quantize(
                        Decimal("0.01")
                    ):
                        raise ContractError(
                            f"{where}.billing.history {window_name} {metric_name} "
                            "does not match grouped billing days"
                        )

        savings_plans = billing["savings_plans"]
        if not isinstance(savings_plans, dict) or set(savings_plans) != {
            "page_count",
            "rows",
        }:
            raise ContractError(f"{where}.billing.savings_plans has an invalid shape")
        if (
            not isinstance(savings_plans["page_count"], int)
            or isinstance(savings_plans["page_count"], bool)
            or savings_plans["page_count"] < 0
            or not isinstance(savings_plans["rows"], list)
        ):
            raise ContractError(f"{where}.billing.savings_plans counts are invalid")
        for index, item in enumerate(savings_plans["rows"]):
            item_where = f"{where}.billing.savings_plans.rows[{index}]"
            if not isinstance(item, dict) or set(item) != {
                "Attributes",
                "Coverage",
                "TimePeriod",
            }:
                raise ContractError(f"{item_where} has an invalid shape")
            if not isinstance(item["Attributes"], dict) or not all(
                isinstance(key, str) and (isinstance(value, str) or value is None)
                for key, value in item["Attributes"].items()
            ):
                raise ContractError(f"{item_where}.Attributes is invalid")
            coverage_row = item["Coverage"]
            if not isinstance(coverage_row, dict) or set(coverage_row) != {
                "SpendCoveredBySavingsPlans",
                "OnDemandCost",
                "TotalCost",
                "CoveragePercentage",
            }:
                raise ContractError(f"{item_where}.Coverage has an invalid shape")
            amounts = {
                key: money(coverage_row[key], f"{item_where}.Coverage.{key}")
                for key in ("SpendCoveredBySavingsPlans", "OnDemandCost", "TotalCost")
            }
            if (
                amounts["SpendCoveredBySavingsPlans"] + amounts["OnDemandCost"]
            ).quantize(Decimal("0.01")) != amounts["TotalCost"].quantize(Decimal("0.01")):
                raise ContractError(f"{item_where}.Coverage costs do not close")
            try:
                coverage_percent = Decimal(coverage_row["CoveragePercentage"])
            except (InvalidOperation, TypeError) as exc:
                raise ContractError(
                    f"{item_where}.Coverage.CoveragePercentage is invalid"
                ) from exc
            if not coverage_percent.is_finite() or not 0 <= coverage_percent <= 100:
                raise ContractError(
                    f"{item_where}.Coverage.CoveragePercentage is out of range"
                )
            if item["TimePeriod"] != {
                "Start": windows["previous"]["start"],
                "End": windows["current"]["end"],
            }:
                raise ContractError(f"{item_where}.TimePeriod does not match the report window")

    inventory = snapshot.get("inventory")
    inventory_fields = {
        "baseline",
        "regions",
        "surfaces",
        "counts",
        "resources",
        "object_storage",
    }
    if not legacy:
        inventory_fields.add("evidence_coverage")
    if not isinstance(inventory, dict) or set(inventory) != inventory_fields:
        raise ContractError(f"{where}.inventory has an invalid shape")
    regions = inventory["regions"]
    surfaces = inventory["surfaces"]
    resources = inventory["resources"]
    if not legacy:
        evidence_coverage = inventory["evidence_coverage"]
        if not isinstance(evidence_coverage, dict) or set(evidence_coverage) != {
            "status",
            "gaps",
        }:
            raise ContractError(f"{where}.inventory.evidence_coverage has an invalid shape")
        evidence_gaps = evidence_coverage["gaps"]
        if evidence_coverage["status"] not in {"complete", "incomplete"} or not isinstance(
            evidence_gaps, list
        ) or not all(isinstance(gap, str) and gap for gap in evidence_gaps):
            raise ContractError(f"{where}.inventory.evidence_coverage is invalid")
        if (evidence_coverage["status"] == "complete") != (not evidence_gaps):
            raise ContractError(
                f"{where}.inventory evidence coverage status and gaps disagree"
            )
    if (
        not isinstance(regions, list)
        or regions != sorted(set(regions))
        or not all(isinstance(region, str) and region for region in regions)
    ):
        raise ContractError(f"{where}.inventory.regions must be sorted and unique")
    if not legacy:
        for gap in evidence_gaps:
            parts = gap.split(":", 2)
            if (
                len(parts) != 3
                or parts[0] not in regions
                or parts[1] not in AWS_EVIDENCE_GAP_KINDS
                or not parts[2].strip()
            ):
                raise ContractError(
                    f"{where}.inventory.evidence_coverage.gaps contains an invalid gap"
                )
    if legacy:
        if (
            not isinstance(surfaces, list)
            or len(surfaces) != len(set(surfaces))
            or not all(isinstance(surface, str) and surface for surface in surfaces)
        ):
            raise ContractError(f"{where}.inventory.surfaces must be unique names")
    elif (
        not isinstance(surfaces, list)
        or not all(isinstance(surface, str) for surface in surfaces)
        or len(surfaces) != len(AWS_REQUIRED_SURFACES)
        or set(surfaces) != set(AWS_REQUIRED_SURFACES)
    ):
        raise ContractError(f"{where}.inventory.surfaces must contain the complete v3 surface set")
    if not isinstance(resources, list):
        raise ContractError(f"{where}.inventory.resources must be an array")
    allowed_resource_fields = {
        "kind",
        "region",
        "id",
        "weight",
        "vpc_id",
        "created",
        "subtype",
        "state",
        "size_gib",
        "cluster_ids",
        "target_groups",
        "registered_targets",
        "healthy_targets",
        "name",
        "owner",
        "session",
        "alias_count",
        "aws_managed",
        "sampled",
        "description_present",
        "description_kind",
        "image_count",
        "image_bytes",
        "untagged_images",
        "oldest_image",
        "latest_pull",
        "image_read_error",
        "lifecycle_state",
        "source",
        "ca_type",
        "recent_issue_events",
        "recent_issue_callers",
    }
    observed = collections.Counter()
    keys: list[tuple[str, str, str]] = []
    for row_index, row in enumerate(resources):
        row_where = f"{where}.inventory.resources[{row_index}]"
        if not isinstance(row, dict):
            raise ContractError(f"{row_where} must be an object")
        require_keys(row, allowed_resource_fields, row_where)
        if (
            row.get("kind") not in surfaces
            or row.get("region") not in regions
            or not isinstance(row.get("id"), str)
            or not row["id"]
        ):
            raise ContractError(f"{row_where} kind, region, or id is invalid")
        weight = row.get("weight", 1)
        if not isinstance(weight, int) or isinstance(weight, bool) or weight <= 0:
            raise ContractError(f"{row_where}.weight must be a positive integer")
        cluster_ids = row.get("cluster_ids")
        if (
            not isinstance(cluster_ids, list)
            or cluster_ids != sorted(set(cluster_ids))
            or not all(isinstance(cluster_id, str) and cluster_id for cluster_id in cluster_ids)
        ):
            raise ContractError(f"{row_where}.cluster_ids must be sorted and unique")
        created = row.get("created")
        if created:
            try:
                date.fromisoformat(created)
            except (TypeError, ValueError) as exc:
                raise ContractError(f"{row_where}.created must use YYYY-MM-DD") from exc
        for text_key in (
            "name",
            "owner",
            "session",
            "ca_type",
            "description_kind",
            "oldest_image",
            "latest_pull",
            "image_read_error",
            "source",
        ):
            if text_key in row and row[text_key] is not None and not isinstance(
                row[text_key], str
            ):
                raise ContractError(f"{row_where}.{text_key} must be a string or null")
        for count_key in (
            "alias_count",
            "recent_issue_events",
            "image_count",
            "image_bytes",
            "untagged_images",
        ):
            if count_key in row and (
                not isinstance(row[count_key], int)
                or isinstance(row[count_key], bool)
                or row[count_key] < 0
            ):
                raise ContractError(f"{row_where}.{count_key} must be non-negative")
        if row["kind"] != "ecr-repository" and any(
            field in row for field in ("image_read_error", "lifecycle_state")
        ):
            raise ContractError(f"{row_where} image evidence requires an ECR repository")
        if "description_kind" in row and (
            row["kind"] != "kms-key"
            or row["description_kind"]
            not in {
                "account-service organization key",
                "named DBaaS key",
                "other description",
                "no description",
                "not sampled",
            }
        ):
            raise ContractError(f"{row_where}.description_kind is invalid")
        if "image_count" in row and (
            row["kind"] != "ecr-repository"
            or "image_bytes" not in row
            or "untagged_images" not in row
            or row["untagged_images"] > row["image_count"]
            or "image_read_error" in row
            or (row["image_count"] == 0 and row["image_bytes"] != 0)
        ):
            raise ContractError(f"{row_where} has inconsistent image evidence")
        if (
            any(
                field in row
                for field in (
                    "image_bytes",
                    "untagged_images",
                    "oldest_image",
                    "latest_pull",
                )
            )
            and "image_count" not in row
        ):
            raise ContractError(f"{row_where} image details need a completed image read")
        for field in ("oldest_image", "latest_pull"):
            if row.get(field):
                try:
                    date.fromisoformat(row[field])
                except (TypeError, ValueError) as exc:
                    raise ContractError(f"{row_where}.{field} must be an ISO date") from exc
        for field in ("sampled", "aws_managed", "description_present"):
            if field in row and row[field] is not None and not isinstance(row[field], bool):
                raise ContractError(f"{row_where}.{field} must be boolean")
        if "lifecycle_state" in row and row["lifecycle_state"] not in {
            "present",
            "none",
            "unknown",
        }:
            raise ContractError(f"{row_where}.lifecycle_state is invalid")
        callers = row.get("recent_issue_callers")
        if callers is not None and (
            not isinstance(callers, list)
            or callers != sorted(set(callers))
            or not all(isinstance(caller, str) and caller for caller in callers)
        ):
            raise ContractError(f"{row_where}.recent_issue_callers is invalid")
        observed[row["kind"]] += 1
        keys.append((row["kind"], row["region"], row["id"]))
    if keys != sorted(keys) or len(keys) != len(set(keys)):
        raise ContractError(f"{where}.inventory.resources must be sorted and unique")
    counts = inventory["counts"]
    if not isinstance(counts, dict) or set(counts) != set(surfaces):
        raise ContractError(f"{where}.inventory.counts must contain every surface")
    if any(counts.get(surface, 0) != observed[surface] for surface in surfaces):
        raise ContractError(f"{where}.inventory.counts does not match resources")

    object_storage = inventory["object_storage"]
    expected_storage_fields = (
        {"bucket_count", "bucket_sizes"}
        if legacy
        else {"bucket_count", "buckets"}
    )
    if not isinstance(object_storage, dict) or set(object_storage) != expected_storage_fields:
        raise ContractError(f"{where}.inventory.object_storage has an invalid shape")
    bucket_count = object_storage["bucket_count"]
    bucket_sizes = object_storage["bucket_sizes" if legacy else "buckets"]
    if (
        not isinstance(bucket_count, int)
        or isinstance(bucket_count, bool)
        or bucket_count < 0
        or not isinstance(bucket_sizes, list)
        or len(bucket_sizes) > bucket_count
    ):
        raise ContractError(f"{where}.inventory.object_storage counts are invalid")
    bucket_names = []
    for index, item in enumerate(bucket_sizes):
        item_where = f"{where}.inventory.object_storage.bucket_sizes[{index}]"
        expected_bucket_fields = (
            {"bucket", "region", "bytes", "observed_at"}
            if legacy
            else {"bucket", "region", "created", "bytes", "observed_at"}
        )
        optional_bucket_fields = (
            {"empty", "lifecycle_state", "lifecycle_actions", "audit_errors"}
            if not legacy
            else set()
        )
        if (
            not isinstance(item, dict)
            or not expected_bucket_fields <= set(item)
            or set(item) - expected_bucket_fields - optional_bucket_fields
        ):
            raise ContractError(f"{item_where} has an invalid shape")
        if set(item) & optional_bucket_fields and not optional_bucket_fields <= set(item):
            raise ContractError(f"{item_where} bucket audit fields must appear together")
        if "empty" in item and item["empty"] is not None and not isinstance(item["empty"], bool):
            raise ContractError(f"{item_where}.empty must be boolean or unknown")
        if "lifecycle_state" in item and item["lifecycle_state"] not in {
            "enabled",
            "disabled",
            "none",
            "unknown",
        }:
            raise ContractError(f"{item_where}.lifecycle_state is invalid")
        for field in ("lifecycle_actions", "audit_errors"):
            if field in item and (
                not isinstance(item[field], list)
                or not all(isinstance(value, str) for value in item[field])
            ):
                raise ContractError(f"{item_where}.{field} is invalid")
        if item.get("lifecycle_actions") and item.get("lifecycle_state") != "enabled":
            raise ContractError(f"{item_where} has actions without an enabled lifecycle")
        if set(item.get("lifecycle_actions", [])) - {
            "expire-current",
            "expire-noncurrent",
            "transition-current",
            "transition-noncurrent",
            "abort-multipart",
        }:
            raise ContractError(f"{item_where}.lifecycle_actions is invalid")
        if (
            not isinstance(item["bucket"], str)
            or not item["bucket"]
            or not isinstance(item["region"], str)
            or not item["region"]
        ):
            raise ContractError(f"{item_where} bucket or bytes is invalid")
        if item["bytes"] is not None and (
            not isinstance(item["bytes"], int)
            or isinstance(item["bytes"], bool)
            or item["bytes"] < 0
        ):
            raise ContractError(f"{item_where}.bytes is invalid")
        if (item["bytes"] is None) != (item["observed_at"] is None):
            raise ContractError(f"{item_where} bytes and observed_at must appear together")
        if item["observed_at"] is not None:
            try:
                datetime.fromisoformat(item["observed_at"].replace("Z", "+00:00"))
            except (AttributeError, TypeError, ValueError) as exc:
                raise ContractError(f"{item_where}.observed_at must be ISO-8601") from exc
        if not legacy and item["created"]:
            try:
                date.fromisoformat(item["created"])
            except (TypeError, ValueError) as exc:
                raise ContractError(f"{item_where}.created must use YYYY-MM-DD") from exc
        bucket_names.append(item["bucket"])
    if bucket_names != sorted(set(bucket_names)):
        raise ContractError(f"{where}.inventory.object_storage buckets must be sorted and unique")


def aws_usage_route(usage: str, service: str) -> tuple[str, str, str | None, str | None]:
    if "DataTransfer" in usage or "NatGateway-Bytes" in usage:
        return "unattributed", "data transfer (needs VPC Flow Logs)", None, None
    if service not in {
        "Amazon Elastic Compute Cloud - Compute",
        "EC2 - Other",
        "Amazon Elastic Load Balancing",
        "Amazon Relational Database Service",
        "Amazon Elastic Container Service for Kubernetes",
        "Amazon Virtual Private Cloud",
        "Amazon Managed Streaming for Apache Kafka",
        "Amazon MemoryDB",
    }:
        return "other", "other AWS services", None, None
    if match := re.search(r"(?:BoxUsage|SpotUsage):(.+)$", usage):
        return "resource", "compute", match.group(1), None
    if "NatGateway-Hours" in usage:
        return "resource", "nat", None, None
    if "LoadBalancerUsage" in usage and "LCUUsage" not in usage:
        return "resource", "loadbalancer", None, None
    if "LCUUsage" in usage:
        return "unattributed", "load-balancer capacity (needs per-LB metrics)", None, None
    if match := re.search(r"EBS:VolumeUsage\.([^.]+)$", usage):
        return "resource", "ebs", match.group(1), None
    if "EBS:Volume" in usage:
        return "unattributed", "EBS add-ons (needs provisioned IOPS/throughput)", None, None
    if match := re.search(r"(?:InstanceUsage|Multi-AZUsage):(.+)$", usage):
        return "resource", "rds", match.group(1), None
    if "Aurora:ServerlessV2Usage" in usage:
        return "resource", "rds", "db.serverless", None
    if "Aurora:" in usage or "NodeUsage:" in usage:
        return "unattributed", "database variable usage", None, None
    if "AmazonEKS-Hours" in usage:
        return "resource", "eks", None, None
    if "VpcEndpoint-Hours" in usage:
        return "resource", "vpc-endpoint", None, None
    if "VpcEndpoint" in usage:
        return "unattributed", "VPC endpoint data processing", None, None
    if "PublicIPv4:IdleAddress" in usage:
        return "resource", "public-ipv4", "elastic-ip", "unassociated"
    if "PublicIPv4:InUseAddress" in usage:
        return "resource", "public-ipv4", None, "associated"
    if match := re.search(r"Kafka\.([^.]+(?:\.[^.]+)*)$", usage):
        return "resource", "msk", match.group(1), None
    return "other", "other AWS services", None, None


def aws_resource_present(row: dict[str, Any], window_end: date) -> bool:
    created = row.get("created")
    return not created or date.fromisoformat(created) < window_end


def round_buckets(
    buckets: dict[tuple[str, ...], Decimal], total: Decimal
) -> dict[tuple[str, ...], Decimal]:
    target = total.quantize(Decimal("0.01"))
    if sum(buckets.values(), Decimal(0)).quantize(Decimal("0.01")) != target:
        raise ContractError("AWS cost detail does not close to its expected total")
    rounded = {key: value.quantize(Decimal("0.01")) for key, value in buckets.items()}
    residual = target - sum(rounded.values(), Decimal(0))
    if residual and rounded:
        largest = max(rounded, key=lambda key: abs(rounded[key]))
        rounded[largest] += residual
    return rounded


def analyze_aws_window(
    snapshot: dict[str, Any], window_name: str
) -> dict[str, Any]:
    window = snapshot["windows"][window_name]
    start = date.fromisoformat(window["start"])
    end = date.fromisoformat(window["end"])
    metric = snapshot["billing"]["metric"]
    comparison_metric = snapshot["billing"]["comparison_metric"]
    resources = snapshot["inventory"]["resources"]
    buckets: dict[tuple[str, str], Decimal] = collections.defaultdict(Decimal)
    resource_costs: dict[str, Decimal] = collections.defaultdict(Decimal)
    vpc_costs: dict[tuple[str, str], Decimal] = collections.defaultdict(Decimal)
    vpc_services: dict[tuple[str, str, str], Decimal] = collections.defaultdict(Decimal)
    drilldown_costs: dict[str, Decimal] = collections.defaultdict(Decimal)
    drilldown_usage: dict[str, dict[str, Decimal]] = collections.defaultdict(
        lambda: collections.defaultdict(Decimal)
    )
    def add_non_vpc(category: str, label: str, amount: Decimal, usage: str) -> None:
        buckets[(category, label)] += amount
        drilldown_costs[label] += amount
        drilldown_usage[label][usage] += amount

    routes: dict[str, set[tuple]] = collections.defaultdict(set)
    for day in snapshot["billing"]["service_usage_days"]:
        for group in day["Groups"]:
            service, usage = group["Keys"]
            if Decimal(group["Metrics"][metric]["Amount"]):
                routes[usage].add(aws_usage_route(usage, service))
    comparison_total = Decimal(0)
    total = Decimal(0)
    for day in snapshot["billing"]["days"]:
        day_start = date.fromisoformat(day["TimePeriod"]["Start"])
        if not (start <= day_start < end):
            continue
        for group in day["Groups"]:
            region, usage = group["Keys"]
            amount = money(group["Metrics"][metric]["Amount"], f"{window_name}:{usage}")
            total += amount
            comparison_total += money(
                group["Metrics"][comparison_metric]["Amount"],
                f"{window_name}:{usage}:{comparison_metric}",
            )
            if not amount:
                continue
            choices = routes[usage]
            if not choices:
                raise ContractError(f"AWS usage {usage!r} has no service evidence")
            # No joint region/service dimension: do not invent resource ownership for an ambiguous usage.
            route, label, subtype, state = (
                next(iter(choices))
                if len(choices) == 1
                else ("other", "other AWS services", None, None)
            )
            if route != "resource":
                add_non_vpc(route, label, amount, usage)
                continue
            selected = [
                row
                for row in resources
                if row["kind"] == label
                and row["region"] == region
                and aws_resource_present(row, day_start + timedelta(days=1))
                and (subtype is None or row.get("subtype") == subtype)
                and (state is None or row.get("state") == state)
            ]
            if not selected:
                add_non_vpc(
                    "unattributed",
                    f"{label} without matching inventory",
                    amount,
                    usage,
                )
                continue
            denominator = sum(Decimal(row.get("weight", 1)) for row in selected)
            for row in selected:
                share = amount * Decimal(row.get("weight", 1)) / denominator
                resource_costs[aws_resource_key(row)] += share
                if row.get("vpc_id"):
                    vpc_costs[(row["region"], row["vpc_id"])] += share
                    vpc_services[(row["region"], row["vpc_id"], label)] += share
                cluster_ids = row["cluster_ids"]
                if len(cluster_ids) == 1:
                    buckets[("group", cluster_ids[0])] += share
                elif len(cluster_ids) > 1:
                    buckets[("unattributed", "shared VPC with multiple EKS clusters")] += share
                elif row.get("vpc_id"):
                    buckets[("unattributed", "VPC without an EKS cluster")] += share
                else:
                    add_non_vpc(
                        "unattributed",
                        "account-level or detached resource",
                        share,
                        usage,
                    )
    rounded = round_buckets(buckets, total)
    if sum(rounded.values(), Decimal(0)) != total.quantize(Decimal("0.01")):
        raise ContractError(f"AWS {window_name} cost buckets do not close to total")
    by_category: dict[str, dict[str, Decimal]] = collections.defaultdict(dict)
    for (category, label), amount in rounded.items():
        by_category[category][label] = amount
    drilldown_costs = {
        label: by_category[category][label]
        for category, label in buckets
        if label in drilldown_costs
    }
    return {
        "total": total.quantize(Decimal("0.01")),
        "comparison_total": comparison_total.quantize(Decimal("0.01")),
        "groups": by_category["group"],
        "unattributed": by_category["unattributed"],
        "other": by_category["other"],
        "resource_costs": resource_costs,
        "vpc_costs": vpc_costs,
        "vpc_services": vpc_services,
        "raw_no_vpc_total": total - sum(vpc_costs.values(), Decimal(0)),
        "drilldown_costs": drilldown_costs,
        "drilldown_usage": drilldown_usage,
    }


def analyze_aws_service_usage(
    snapshot: dict[str, Any], window_name: str, expected_total: Decimal
) -> dict[tuple[str, str], Decimal]:
    window = snapshot["windows"][window_name]
    start = date.fromisoformat(window["start"])
    end = date.fromisoformat(window["end"])
    metric = snapshot["billing"]["metric"]
    costs: dict[tuple[str, str], Decimal] = collections.defaultdict(Decimal)
    for day in snapshot["billing"]["service_usage_days"]:
        day_start = date.fromisoformat(day["TimePeriod"]["Start"])
        if not (start <= day_start < end):
            continue
        for group in day["Groups"]:
            service, usage = group["Keys"]
            costs[(service or "Unknown service", usage)] += money(
                group["Metrics"][metric]["Amount"], f"{window_name}:{service}:{usage}"
            )
    if sum(costs.values(), Decimal(0)).quantize(Decimal("0.01")) != expected_total:
        raise ContractError(f"AWS {window_name} service and usage costs do not match total")
    return costs


def live_cost_delta(
    previous: dict[str, Decimal], current: dict[str, Decimal], total: Decimal
) -> list[dict[str, str]]:
    return [
        {
            "name": name,
            "previous": money_text(previous.get(name, Decimal(0))),
            "current": money_text(current.get(name, Decimal(0))),
            "delta": money_text(current.get(name, Decimal(0)) - previous.get(name, Decimal(0))),
            "current_share_percent": percent_text(current.get(name, Decimal(0)), total),
        }
        for name in sorted(set(previous) | set(current))
    ]


def aws_ownership_rows(
    snapshot: dict[str, Any], previous: dict[str, Any], current: dict[str, Any]
) -> dict[str, Any]:
    cluster_by_vpc: dict[tuple[str, str], list[str]] = collections.defaultdict(list)
    for item in snapshot["inventory"]["resources"]:
        if item["kind"] == "eks" and item.get("vpc_id"):
            cluster_by_vpc[(item["region"], item["vpc_id"])].append(item["id"])

    vpc_keys = set(previous["vpc_costs"]) | set(current["vpc_costs"])
    compact_keys = {
        key: key if cluster_by_vpc.get(key) else ("multiple", "Other VPCs without EKS")
        for key in vpc_keys
    }

    def costs(window: dict[str, Any]) -> dict[tuple[str, str], Decimal]:
        values: dict[tuple[str, str], Decimal] = collections.defaultdict(Decimal)
        for key, amount in window["vpc_costs"].items():
            values[compact_keys[key]] += amount
        values[("all", "No VPC")] = window["total"] - sum(
            window["vpc_costs"].values(), Decimal(0)
        )
        return round_buckets(values, window["total"])

    previous_costs = costs(previous)
    current_costs = costs(current)
    rows = []
    for key in set(previous_costs) | set(current_costs):
        current_cost = current_costs.get(key, Decimal(0))
        previous_cost = previous_costs.get(key, Decimal(0))
        clusters = sorted(set(cluster_by_vpc.get(key, [])))
        rows.append(
            {
                "region": key[0],
                "vpc": key[1],
                "eks_cluster": ", ".join(clusters) or "none",
                "previous": money_text(previous_cost),
                "current": money_text(current_cost),
                "delta": money_text(current_cost - previous_cost),
                "current_share_percent": percent_text(current_cost, current["total"]),
            }
        )
    rows.sort(
        key=lambda item: (
            item["vpc"] in {"Other VPCs without EKS", "No VPC"},
            -Decimal(item["current"]),
        )
    )
    summary_amounts = {
        "EKS-linked VPCs": sum(
            (
                Decimal(item["current"])
                for item in rows
                if item["vpc"] not in {"Other VPCs without EKS", "No VPC"}
            ),
            Decimal(0),
        ),
        "Other VPCs without EKS": sum(
            (
                Decimal(item["current"])
                for item in rows
                if item["vpc"] == "Other VPCs without EKS"
            ),
            Decimal(0),
        ),
        "No VPC": sum(
            (Decimal(item["current"]) for item in rows if item["vpc"] == "No VPC"),
            Decimal(0),
        ),
    }
    if sum(summary_amounts.values(), Decimal(0)) != current["total"]:
        raise ContractError("AWS summary ownership slices do not close to current total")
    return {
        "rows": rows,
        "vpc_count": len(vpc_keys),
        "previous_total": money_text(previous["total"]),
        "current_total": money_text(current["total"]),
        "summary_pie": [
            {
                "name": name,
                "current": money_text(amount),
                "current_share_percent": percent_text(amount, current["total"]),
            }
            for name, amount in summary_amounts.items()
        ]
        if current["total"] > 0 and all(amount >= 0 for amount in summary_amounts.values())
        else [],
    }


def aws_non_vpc_category(label: str) -> str:
    if label == "data transfer (needs VPC Flow Logs)":
        return "network traffic"
    if label == "account-level or detached resource" or label.endswith(
        " without matching inventory"
    ):
        return "resources without a VPC or matching inventory"
    if label in {
        "object storage",
        "container registry",
        "KMS",
        "ACM / Private CA",
        "other AWS services",
    }:
        return "account-level services"
    return "resource add-ons without per-resource evidence"


def aws_non_vpc_service_usage(
    previous: dict[str, Any],
    current: dict[str, Any],
    previous_services: dict[tuple[str, str], Decimal],
    current_services: dict[tuple[str, str], Decimal],
    ownership: dict[str, Any],
) -> dict[str, list[dict[str, str]]]:
    def split_by_service(
        services: dict[tuple[str, str], Decimal], usage: str, amount: Decimal
    ) -> dict[str, Decimal]:
        matches = {
            service: cost
            for (service, candidate_usage), cost in services.items()
            if candidate_usage == usage and cost
        }
        if not matches:
            raise ContractError(
                f"AWS usage type {usage!r} has no matching billing service"
            )
        if len(matches) == 1:
            return {next(iter(matches)): amount}
        if sum(matches.values(), Decimal(0)).quantize(Decimal("0.01")) == amount.quantize(
            Decimal("0.01")
        ):
            return matches
        raise ContractError(
            f"AWS usage type {usage!r} spans services and partial inventory attribution"
        )

    def round_detail(
        values: dict[tuple[str, ...], Decimal],
        raw_total: Decimal,
        displayed_total: Decimal,
    ) -> dict[tuple[str, ...], Decimal]:
        rounded = round_buckets(values, raw_total)
        residual = displayed_total - sum(rounded.values(), Decimal(0))
        if residual and rounded:
            rounded[max(rounded, key=lambda key: abs(rounded[key]))] += residual
        return rounded

    def grouped(
        window: dict[str, Any], services: dict[str, Any], total: Decimal
    ) -> dict[tuple[str, ...], Decimal]:
        values: dict[tuple[str, ...], Decimal] = collections.defaultdict(Decimal)
        for label, usage_costs in window["drilldown_usage"].items():
            category = aws_non_vpc_category(label)
            for usage, amount in usage_costs.items():
                for service, service_amount in split_by_service(
                    services, usage, amount
                ).items():
                    values[(category, service, usage)] += service_amount
        return round_detail(values, window["raw_no_vpc_total"], total)

    no_vpc = next(item for item in ownership["rows"] if item["vpc"] == "No VPC")
    previous_costs = grouped(previous, previous_services, Decimal(no_vpc["previous"]))
    current_costs = grouped(current, current_services, Decimal(no_vpc["current"]))
    rows = []
    for category, service, usage in set(previous_costs) | set(current_costs):
        previous_cost = previous_costs.get((category, service, usage), Decimal(0))
        current_cost = current_costs.get((category, service, usage), Decimal(0))
        rows.append(
            {
                "category": category,
                "service": service,
                "usage_type": usage,
                "previous": money_text(previous_cost),
                "current": money_text(current_cost),
                "delta": money_text(current_cost - previous_cost),
                "share_of_no_vpc_percent": percent_text(
                    current_cost, Decimal(no_vpc["current"])
                ),
            }
        )
    rows.sort(
        key=lambda item: (
            item["category"],
            -Decimal(item["current"]),
            item["service"],
            item["usage_type"],
        )
    )
    categories = []
    for name in (
        "network traffic",
        "resources without a VPC or matching inventory",
        "resource add-ons without per-resource evidence",
        "account-level services",
    ):
        previous_cost = sum(
            (Decimal(item["previous"]) for item in rows if item["category"] == name),
            Decimal(0),
        )
        current_cost = sum(
            (Decimal(item["current"]) for item in rows if item["category"] == name),
            Decimal(0),
        )
        categories.append(
            {
                "name": name,
                "previous": money_text(previous_cost),
                "current": money_text(current_cost),
                "delta": money_text(current_cost - previous_cost),
                "current_share_percent": percent_text(current_cost, current["total"]),
            }
        )

    compute_label = "compute without matching inventory"
    compute_total = current["drilldown_costs"].get(compute_label, Decimal(0))
    compute_previous_total = previous["drilldown_costs"].get(compute_label, Decimal(0))
    previous_compute_usage = previous["drilldown_usage"].get(compute_label, {})
    previous_compute = round_detail(
        {
            (service, usage): service_amount
            for usage, amount in previous_compute_usage.items()
            for service, service_amount in split_by_service(
                previous_services, usage, amount
            ).items()
        },
        sum(previous_compute_usage.values(), Decimal(0)),
        compute_previous_total,
    )
    current_compute_usage = current["drilldown_usage"].get(compute_label, {})
    current_compute = round_detail(
        {
            (service, usage): service_amount
            for usage, amount in current_compute_usage.items()
            for service, service_amount in split_by_service(
                current_services, usage, amount
            ).items()
        },
        sum(current_compute_usage.values(), Decimal(0)),
        compute_total,
    )
    compute_rows = []
    for service, usage in set(previous_compute) | set(current_compute):
        previous_cost = previous_compute.get((service, usage), Decimal(0))
        current_cost = current_compute.get((service, usage), Decimal(0))
        compute_rows.append(
            {
                "service": service,
                "usage_type": usage,
                "previous": money_text(previous_cost),
                "current": money_text(current_cost),
                "delta": money_text(current_cost - previous_cost),
                "share_percent": percent_text(current_cost, compute_total),
            }
        )
    compute_rows.sort(key=lambda item: (-Decimal(item["current"]), item["usage_type"]))
    return {
        "categories": categories,
        "rows": rows,
        "compute_without_inventory": compute_rows,
    }


def aws_object_storage(
    snapshot: dict[str, Any],
    previous: dict[tuple[str, str], Decimal],
    current: dict[tuple[str, str], Decimal],
) -> dict[str, Any]:
    raw = snapshot["inventory"]["object_storage"]
    all_rows = raw["buckets"]
    rows = sorted(
        (item for item in all_rows if item["bytes"] is not None),
        key=lambda item: (-item["bytes"], item["bucket"]),
    )
    total_bytes = sum(item["bytes"] for item in rows)
    previous_cost, current_cost = (
        sum(
            (
                amount
                for (service, usage), amount in data.items()
                if service == "Amazon Simple Storage Service"
            ),
            Decimal(0),
        )
        for data in (previous, current)
    )
    prefix_counts: collections.Counter[str] = collections.Counter()
    for item in all_rows:
        parts = re.split(r"[-_.]+", item["bucket"])
        for length in range(2, min(4, len(parts)) + 1):
            prefix_counts["-".join(parts[:length])] += 1

    families: dict[str, dict[str, Any]] = {}
    for item in all_rows:
        parts = re.split(r"[-_.]+", item["bucket"])
        candidates = [
            "-".join(parts[:length])
            for length in range(2, min(4, len(parts)) + 1)
            if prefix_counts["-".join(parts[:length])] > 1
        ]
        family = min(candidates, key=len) if candidates else item["bucket"]
        group = families.setdefault(
            family,
            {"name": family, "bucket_count": 0, "measured_bucket_count": 0, "bytes": 0},
        )
        group["bucket_count"] += 1
        if item["bytes"] is not None:
            group["measured_bucket_count"] += 1
            group["bytes"] += item["bytes"]
    return {
        "previous": money_text(previous_cost),
        "current": money_text(current_cost),
        "delta": money_text(current_cost - previous_cost),
        "bucket_count": raw["bucket_count"],
        "measured_bucket_count": len(rows),
        "unmeasured_bucket_count": raw["bucket_count"] - len(rows),
        "empty_count": sum(item.get("empty") is True for item in all_rows),
        "empty_checked": sum(item.get("empty") is not None for item in all_rows),
        "lifecycle_checked": sum(
            item.get("lifecycle_state", "unknown") != "unknown" for item in all_rows
        ),
        "lifecycle_missing": sum(
            item.get("lifecycle_state") in {"none", "disabled"} for item in all_rows
        ),
        "all_buckets": all_rows,
        "measured_bytes": total_bytes,
        "families": [
            {
                **item,
                "share_of_measured_percent": (
                    f"{Decimal(item['bytes']) * Decimal(100) / Decimal(total_bytes):.2f}"
                    if total_bytes
                    else "0.00"
                ),
            }
            for item in sorted(families.values(), key=lambda item: (-item["bytes"], item["name"]))
        ],
        "buckets": [
            {
                **item,
                "share_of_measured_percent": (
                    f"{Decimal(item['bytes']) * Decimal(100) / Decimal(total_bytes):.2f}"
                    if total_bytes
                    else "0.00"
                ),
            }
            for item in rows
        ],
    }


def aws_weekly_history(snapshot: dict[str, Any]) -> list[dict[str, str]]:
    billing = snapshot["billing"]
    history = billing["history"]
    metric = billing["metric"]
    comparison_metric = billing["comparison_metric"]
    rows = []
    for offset in range(0, 56, 7):
        days = history["days"][offset : offset + 7]
        start = date.fromisoformat(days[0]["TimePeriod"]["Start"])
        end = date.fromisoformat(days[-1]["TimePeriod"]["End"])
        current = sum(
            (money(item["Total"][metric]["Amount"], f"history:{metric}") for item in days),
            Decimal(0),
        )
        comparison = sum(
            (
                money(
                    item["Total"][comparison_metric]["Amount"],
                    f"history:{comparison_metric}",
                )
                for item in days
            ),
            Decimal(0),
        )
        rows.append(
            {
                "start": start.isoformat(),
                "end": end.isoformat(),
                "metric": money_text(current),
                "comparison_metric": money_text(comparison),
            }
        )
    return rows


def aws_savings_plan_coverage(snapshot: dict[str, Any]) -> dict[str, Any]:
    rows = []
    for item in snapshot["billing"]["savings_plans"]["rows"]:
        attributes = item["Attributes"]
        service = attributes.get("SERVICE") or next(
            filter(None, attributes.values()), "Unattributed service"
        )
        coverage = item["Coverage"]
        covered = money(coverage["SpendCoveredBySavingsPlans"], service)
        on_demand = money(coverage["OnDemandCost"], service)
        rows.append(
            {
                "service": service,
                "covered_spend": money_text(covered),
                "on_demand_cost": money_text(on_demand),
                "total_cost": money_text(covered + on_demand),
                "coverage_percent": percent_text(covered, covered + on_demand),
            }
        )
    rows.sort(key=lambda item: (-Decimal(item["covered_spend"]), item["service"]))
    covered = sum((Decimal(item["covered_spend"]) for item in rows), Decimal(0))
    total = sum((Decimal(item["total_cost"]) for item in rows), Decimal(0))
    return {
        "covered_spend": money_text(covered),
        "total_cost": money_text(total),
        "coverage_percent": percent_text(covered, total),
        "services": rows,
    }


def aws_vpc_evidence(
    snapshot: dict[str, Any], current: dict[str, Any], ownership: dict[str, Any]
) -> dict[str, Any]:
    resources = snapshot["inventory"]["resources"]
    vpcs = {
        (item["region"], item["id"]): item
        for item in resources
        if item["kind"] == "vpc"
    }
    resource_counts: dict[tuple[str, str], collections.Counter[str]] = collections.defaultdict(
        collections.Counter
    )
    for item in resources:
        if item["kind"] != "vpc" and item.get("vpc_id"):
            resource_counts[(item["region"], item["vpc_id"])][item["kind"]] += 1
    other = next(
        (
            item
            for item in ownership["rows"]
            if item["vpc"] == "Other VPCs without EKS"
        ),
        {"current": "0.00"},
    )
    raw_costs = {
        key: amount
        for key, amount in current["vpc_costs"].items()
        if not vpcs.get(key, {}).get("cluster_ids")
    }
    rounded = round_buckets(raw_costs, Decimal(other["current"])) if raw_costs else {}
    rows = []
    for key, amount in rounded.items():
        region, vpc_id = key
        vpc = vpcs.get(key, {})
        name = vpc.get("name") or "unnamed"
        if vpc.get("session"):
            signal = "cp9-session tag"
        elif re.search(r"(?:data[-_]?plane|dedicated|nightly)", name, re.IGNORECASE):
            signal = "data-plane name signal"
        elif re.search(r"(?:test|demo|sandbox|playground|personal)", name, re.IGNORECASE):
            signal = "test name signal"
        elif vpc.get("owner"):
            signal = "cp9-owner tag"
        elif name == "unnamed":
            signal = "no Name or cp9 ownership tag"
        else:
            signal = "named; purpose unclassified"
        kinds = resource_counts.get(key, {})
        inventory = "; ".join(
            f"{kind} {count}" for kind, count in sorted(kinds.items(), key=lambda pair: (-pair[1], pair[0]))
        ) or "no matched inventory resources"
        rows.append(
            {
                "vpc_id": vpc_id,
                "name": name,
                "region": region,
                "current": money_text(amount),
                "share_percent": percent_text(amount, Decimal(ownership["current_total"])),
                "purpose_signal": signal,
                "owner": vpc.get("owner"),
                "session": vpc.get("session"),
                "state": vpc.get("state") or "unknown",
                "inventory": inventory,
            }
        )
    rows.sort(key=lambda item: (-Decimal(item["current"]), item["region"], item["vpc_id"]))
    return {"rows": rows, "current_total": other["current"]}


def aws_evidence_gap_surfaces(snapshot: dict[str, Any]) -> set[str]:
    if snapshot["version"] == AWS_LEGACY_SNAPSHOT_VERSION:
        return set()
    return {
        kind
        for gap in snapshot["inventory"]["evidence_coverage"]["gaps"]
        for kind in AWS_EVIDENCE_GAP_KINDS[gap.split(":", 2)[1]]
    }


def aws_deep_inventory_evidence(snapshot: dict[str, Any]) -> dict[str, Any]:
    resources = snapshot["inventory"]["resources"]
    evidence_coverage = snapshot["inventory"]["evidence_coverage"]
    gap_surfaces = aws_evidence_gap_surfaces(snapshot)
    by_kind: dict[str, list[dict[str, Any]]] = collections.defaultdict(list)
    for item in resources:
        by_kind[item["kind"]].append(item)

    customer_keys = [item for item in by_kind["kms-key"] if not item.get("aws_managed")]
    unaliased_keys = [item for item in customer_keys if item.get("alias_count", 0) == 0]
    sampled_keys = [item for item in unaliased_keys if item.get("sampled")]
    old_sample = sum(
        bool(item.get("created")) and date.fromisoformat(item["created"]) < date(2022, 1, 1)
        for item in sampled_keys
    )
    kms = {
        "coverage": "partial" if "kms-key" in gap_surfaces else "complete",
        "customer_key_count": len(customer_keys),
        "custom_aliased_key_count": len(customer_keys) - len(unaliased_keys),
        "unaliased_key_count": len(unaliased_keys),
        "sampled_unaliased_count": len(sampled_keys),
        "sampled_pending_deletion_count": sum(
            item.get("state") == "PendingDeletion" for item in sampled_keys
        ),
        "sampled_enabled_count": sum(item.get("state") == "Enabled" for item in sampled_keys),
        "sampled_without_description_count": sum(
            item.get("description_present") is False for item in sampled_keys
        ),
        "sampled_before_2022_count": old_sample,
        "sampled_unaliased_keys": [
            {
                "region": item["region"],
                "id": item["id"],
                "created": item.get("created") or "unknown",
                "state": item.get("state") or "unknown",
                "description": item.get("description_kind")
                or (
                    "present; source not retained"
                    if item.get("description_present") is True
                    else "absent" if item.get("description_present") is False else "Unknown"
                ),
            }
            for item in sorted(sampled_keys, key=lambda row: (row["region"], row["id"]))
        ],
    }

    authorities = copy.deepcopy(by_kind["private-ca"])
    for item in authorities:
        item["recent_issue_callers"] = sorted(
            {
                re.sub(
                    r"^arn:([^:]+):sts::([0-9]+):assumed-role/(.+)/[^/]+$",
                    r"arn:\1:iam::\2:role/\3",
                    caller,
                )
                for caller in item.get("recent_issue_callers", [])
            }
        )
    callers = sorted(
        {
            caller
            for item in authorities
            for caller in item.get("recent_issue_callers", [])
        }
    )
    private_ca = {
        "coverage": "partial" if "private-ca" in gap_surfaces else "complete",
        "count": len(authorities),
        "active_count": sum(item.get("state") == "ACTIVE" for item in authorities),
        "general_purpose_count": sum(
            item.get("subtype") == "GENERAL_PURPOSE" for item in authorities
        ),
        "short_lived_count": sum(
            item.get("subtype") == "SHORT_LIVED_CERTIFICATE" for item in authorities
        ),
        "recent_issue_events": sum(
            item.get("recent_issue_events", 0) for item in authorities
        ),
        "recent_issue_callers": callers,
        "authorities": [
            {
                "region": item["region"],
                "id": item["id"],
                "state": item.get("state") or "unknown",
                "type": item.get("ca_type") or "unknown",
                "usage_mode": item.get("subtype") or "unknown",
                "created": item.get("created") or "unknown",
                "owner": item.get("owner"),
                "session": item.get("session"),
                "recent_issue_events": item.get("recent_issue_events", 0),
                "recent_issue_callers": item.get("recent_issue_callers", []),
            }
            for item in authorities
        ],
    }

    services = []
    for kind in AWS_ACCOUNT_SERVICE_KINDS:
        items = by_kind[kind]
        states = collections.Counter(str(item["state"]) for item in items if item.get("state") is not None)
        types = collections.Counter(item["subtype"] for item in items if item.get("subtype"))
        created = sorted(item["created"] for item in items if item.get("created"))
        services.append(
            {
                "kind": kind,
                "coverage": (
                    "partial"
                    if kind in gap_surfaces
                    else "complete"
                ),
                "count": len(items),
                "capacity": sum(int(item.get("weight", 1)) for item in items),
                "configuration": "; ".join(
                    [
                        *(f"{name} {count}" for name, count in sorted(types.items())),
                        *(f"state {name} {count}" for name, count in sorted(states.items())),
                        *(f"oldest {created[0]}" for _ in created[:1]),
                    ]
                )
                or "No additional configuration",
                "examples": [
                    item.get("name") or item["id"].rsplit("/", 1)[-1]
                    for item in sorted(items, key=lambda row: (row["region"], row["id"]))[:5]
                ],
            }
        )
    repositories = sorted(
        by_kind["ecr-repository"],
        key=lambda item: (-item.get("image_bytes", -1), item["region"], item["id"]),
    )
    image_reads = sum("image_count" in item for item in repositories)
    image_coverage = {
        "status": (
            "incomplete"
            if image_reads != len(repositories) or "ecr-repository" in gap_surfaces
            else "complete"
        ),
        "checked": image_reads,
        "observed_repositories": len(repositories),
    }
    coverage = copy.deepcopy(evidence_coverage)
    if image_reads != len(repositories):
        coverage["status"] = "incomplete"
        coverage["gaps"].append(
            f"ECR image metadata: {len(repositories) - image_reads} of {len(repositories)} observed repositories remain unmeasured"
        )
    return {
        "coverage": coverage,
        "kms": kms,
        "private_ca": private_ca,
        "account_services": services,
        "repositories": repositories,
        "image_coverage": image_coverage,
    }


def aws_inventory_insights(
    snapshot: dict[str, Any],
    candidates: list[dict[str, Any]],
    has_inventory_baseline: bool,
) -> dict[str, Any]:
    resources_by_key = {
        aws_resource_key(item): item for item in snapshot["inventory"]["resources"]
    }
    by_kind: dict[str, list[dict[str, Any]]] = collections.defaultdict(list)
    for candidate in candidates:
        kind = candidate["id"].split("/", 2)[1]
        by_kind[kind].append(candidate)
    ebs_rows = by_kind["ebs"]
    lb_rows = by_kind["loadbalancer"]
    ip_rows = by_kind["public-ipv4"]

    def monthly(rows: list[dict[str, Any]]) -> str:
        return money_text(
            sum((Decimal(item["estimated_monthly_cost"]) for item in rows), Decimal(0))
        )

    reclaim = {
        "candidate_count": len(candidates),
        "confirmed_count": sum(item["classification"] == "confirmed" for item in candidates),
        "recurring_candidate_count": (
            sum("present in previous snapshot" in item["signals"] for item in candidates)
            if has_inventory_baseline
            else None
        ),
        "estimated_monthly_cost": monthly(candidates),
        "ebs": {
            "count": len(ebs_rows),
            "size_gib": sum(
                int(resources_by_key[item["id"]].get("size_gib") or 0) for item in ebs_rows
            ),
            "estimated_monthly_cost": monthly(ebs_rows),
        },
        "load_balancer": {
            "total_count": snapshot["inventory"]["counts"].get("loadbalancer", 0),
            "zero_healthy_count": len(lb_rows),
            "zero_registered_count": sum(
                resources_by_key[item["id"]].get("registered_targets", 0) == 0
                for item in lb_rows
            ),
            "registered_but_unhealthy_count": sum(
                resources_by_key[item["id"]].get("registered_targets", 0) > 0
                for item in lb_rows
            ),
            "estimated_monthly_cost": monthly(lb_rows),
        },
        "public_ipv4": {
            "total_count": snapshot["inventory"]["counts"].get("public-ipv4", 0),
            "unassociated_count": len(ip_rows),
            "estimated_monthly_cost": monthly(ip_rows),
        },
    }

    return {
        "reclaim": reclaim,
        "current_resource_counts": copy.deepcopy(snapshot["inventory"]["counts"]),
    }


def aws_reclaim_candidates(
    snapshot: dict[str, Any], current: dict[str, Any], previous_keys: set[str] | None
) -> list[dict[str, Any]]:
    candidates: list[dict[str, Any]] = []
    vpcs = {
        (item["region"], item["id"]): item
        for item in snapshot["inventory"]["resources"]
        if item["kind"] == "vpc"
    }
    for row in snapshot["inventory"]["resources"]:
        signals: list[str] = []
        evidence: list[str] = []
        if row["kind"] == "ebs" and row.get("state") == "available":
            signals.append("unattached")
            evidence.append("AWS EBS state is available")
        elif row["kind"] == "loadbalancer" and row.get("healthy_targets") == 0:
            signals.append(
                "no registered targets"
                if row.get("registered_targets", 0) == 0
                else "registered targets are unhealthy"
            )
            evidence.append(
                "AWS target health reports "
                f"0 healthy of {row.get('registered_targets', 0)} registered targets"
            )
        elif row["kind"] == "public-ipv4" and row.get("state") == "unassociated":
            signals.append("unassociated")
            evidence.append("AWS Elastic IP has no association")
        else:
            continue
        key = aws_resource_key(row)
        if previous_keys is not None and key in previous_keys:
            signals.append("present in previous snapshot")
        weekly_cost = current["resource_costs"].get(key, Decimal(0))
        vpc = vpcs.get((row["region"], row.get("vpc_id")), {})
        next_step = (
            "Confirm retention and recovery requirements before removal"
            if row["kind"] == "ebs"
            else (
                "Confirm reserved use and DNS dependencies before release"
                if row["kind"] == "public-ipv4"
                else (
                    "Check DNS and recent traffic; remove only after owner confirms no dependency"
                    if row.get("registered_targets", 0) == 0
                    else "Inspect target registration and health failures; owner chooses repair or retirement"
                )
            )
        )
        evidence.append(
            f"30/7 projection of the current seven-day {snapshot['billing']['metric']} base charge"
        )
        candidates.append(
            {
                "id": key,
                "created": row.get("created") or "unknown",
                "name": row.get("name") or row["id"].split("loadbalancer/", 1)[-1],
                "owner": row.get("owner") or vpc.get("owner") or "Unresolved",
                "source": row.get("source") or row.get("session") or vpc.get("name") or "Unknown",
                "next_step": next_step,
                "cost_estimate_status": (
                    "priced" if key in current["resource_costs"] else "outside matched billing"
                ),
                "classification": "candidate",
                "signals": signals,
                "evidence": evidence,
                "estimated_monthly_cost": money_text(weekly_cost * Decimal(30) / Decimal(7)),
            }
        )
    return sorted(candidates, key=lambda candidate: candidate["id"])


def decision_unit(
    key: str,
    section: str,
    headline: str,
    evidence: str,
    boundary: str,
    decision: str,
    gate: str,
) -> dict[str, str]:
    return {
        "key": key,
        "section": section,
        "headline": headline,
        "evidence": evidence,
        "boundary": boundary,
        "decision": decision,
        "gate": gate,
    }


def candidate_history(
    snapshot: dict[str, Any],
    candidates: list[dict[str, Any]],
    previous: dict[str, Any] | None,
) -> dict[str, Any]:
    history = {
        "version": "cp9.cloud-cost-candidate-history/v1",
        "scope_id": snapshot["scope_id"],
        "account_id": snapshot["account_id"],
        "observations": [],
    }
    now = datetime.fromisoformat(snapshot["collected_at"]).astimezone(timezone.utc)
    if previous is not None:
        if (
            not isinstance(previous, dict)
            or set(previous) != set(history)
            or any(previous[key] != history[key] for key in ("version", "scope_id", "account_id"))
        ):
            raise ContractError("candidate history belongs to another schema or account")
        observations = previous["observations"]
        if not isinstance(observations, list) or len(observations) > 57:
            raise ContractError("candidate history must contain at most 57 daily observations")
        last = None
        for observation in observations:
            if not isinstance(observation, dict) or set(observation) != {
                "collected_at",
                "candidates",
            }:
                raise ContractError("candidate history observation shape is invalid")
            try:
                at = datetime.fromisoformat(observation["collected_at"])
                if at.tzinfo is not None:
                    at = at.astimezone(timezone.utc)
                valid_time = (
                    at.tzinfo is not None and at < now and (last is None or last.date() < at.date())
                )
            except (ValueError, TypeError):
                valid_time = False
            ids = observation["candidates"]
            if (
                not valid_time
                or not isinstance(ids, list)
                or not all(
                    isinstance(key, str)
                    and re.fullmatch(r"[a-z0-9-]+/(?:ebs|loadbalancer|public-ipv4)/[^\s]+", key)
                    for key in ids
                )
                or ids != sorted(set(ids))
            ):
                raise ContractError("candidate history dates or resource identities are invalid")
            last = at
            if at.date() != now.date() and (now - at).days < 56:
                history["observations"].append({"collected_at": at.isoformat(), "candidates": ids})
    history["observations"].append(
        {
            "collected_at": now.isoformat(),
            "candidates": sorted(item["id"] for item in candidates),
        }
    )
    return history


def synthetic_decision_units(providers: dict[str, Any]) -> list[dict[str, str]]:
    sections = (
        "summary",
        "cost-groups",
        "unattributed-cost",
        "account-services",
        "reclaim-candidates",
        "resource-changes",
        "action-plan",
        "coverage-boundary",
        "recurrence-boundary",
        "next-decision",
    )
    coverage = "; ".join(
        f"{name.upper()} {provider['coverage']['status']}"
        for name, provider in sorted(providers.items())
    )
    return [
        decision_unit(
            key,
            section,
            f"The synthetic replay provides deterministic evidence for {section}.",
            coverage,
            "Every cost and resource fact is synthetic; incomplete provider evidence remains unavailable.",
            "Use this replay only to verify report behavior and evidence boundaries.",
            "Fixture reviewer",
        )
        for key, section in zip(REPORT_INTERPRETATION_KEYS, sections)
    ]


def incomplete_aws_decision_units(
    scope: dict[str, Any], gaps: list[str]
) -> list[dict[str, str]]:
    sections = (
        "summary",
        "ownership",
        "no-vpc",
        "object-storage",
        "kms",
        "private-ca",
        "account-services",
        "reclaim",
        "recurrence",
        "action",
    )
    evidence = "; ".join(gaps)
    return [
        decision_unit(
            key,
            section,
            f"Required AWS coverage is incomplete for {section}.",
            evidence,
            "Missing required evidence blocks cost, ownership, resource, and reclaim conclusions.",
            "Restore the required reads before making this decision.",
            f"{scope['label']} owner",
        )
        for key, section in zip(REPORT_INTERPRETATION_KEYS, sections)
    ]


def aws_decision_units(provider: dict[str, Any], scope: dict[str, Any]) -> list[dict[str, str]]:
    cost = provider["cost"]
    ownership = provider["ownership"]
    no_vpc = next(item for item in ownership["rows"] if item["vpc"] == "No VPC")
    largest_eks = max(
        (item for item in ownership["rows"] if item["eks_cluster"] != "none"),
        key=lambda item: Decimal(item["current"]),
        default=None,
    )
    storage = provider["object_storage"]
    top_family = storage["families"][0] if storage["families"] else None
    deep = provider["deep_evidence"]
    kms = deep["kms"]
    private_ca = deep["private_ca"]
    services = [
        item
        for item in deep["account_services"]
        if item["count"] or item["coverage"] == "partial"
    ]
    reclaim = provider["inventory_insights"]["reclaim"]
    recurring = reclaim["recurring_candidate_count"]
    owner = "Unresolved"
    evidence_gaps = deep["coverage"]["gaps"]

    weekly = provider["weekly_history"]
    first_week = weekly[0]
    last_week = weekly[-1]
    summary = decision_unit(
        "1",
        "summary",
        f"Current spend is ${cost['current_total']}; the adjacent-window movement is ${cost['delta']}.",
        f"Eight comparable weekly totals run from ${first_week['metric']} to ${last_week['metric']}; Savings Plans covered spend is ${provider['savings_plan_coverage']['covered_spend']} across the observed services.",
        "Savings Plans coverage is a direct Cost Explorer coverage measure; it is not inferred from the difference between amortized and unblended metrics.",
        "Confirm whether the latest direction is expected and whether the dominant ownership slice has an accountable owner.",
        owner,
    )
    ownership_unit = decision_unit(
        "2",
        "ownership",
        f"No VPC is the largest ownership slice at ${no_vpc['current']} ({no_vpc['current_share_percent']}%).",
        (
            f"The largest EKS-linked row is {largest_eks['eks_cluster']} at ${largest_eks['current']}; "
            f"{len(provider['vpc_evidence']['rows'])} non-EKS VPC rows have deterministic name, tag, and inventory signals."
            if largest_eks
            else f"No EKS-linked row is present; {len(provider['vpc_evidence']['rows'])} non-EKS VPC rows have deterministic name, tag, and inventory signals."
        ),
        "A name or tag is an ownership signal, not proof that a VPC is unused or belongs to a product line.",
        "Route the largest named non-EKS VPCs to their tag owners before any cleanup or resizing decision.",
        owner,
    )
    categories = ", ".join(
        f"{item['name']} ${item['current']}" for item in provider["non_vpc_categories"]
    )
    no_vpc_unit = decision_unit(
        "3",
        "no-vpc",
        f"No-VPC cost is ${no_vpc['current']} and closes through four evidence categories.",
        categories,
        "Billing service and usage type identify the charge, not the workload or resource owner.",
        "Resolve the largest service and usage pairs to inventory or an accountable shared-service owner.",
        owner,
    )
    storage_unit = decision_unit(
        "4",
        "object-storage",
        (
            f"Measured object storage is {report_bytes(storage['measured_bytes'])} across "
            f"{count_text(storage['measured_bucket_count'], 'bucket')} of "
            f"{count_text(storage['bucket_count'], 'bucket')}."
        ),
        (
            f"The largest deterministic bucket family is {top_family['name']} with "
            f"{report_bytes(top_family['bytes'])} ({top_family['share_of_measured_percent']}%) "
            f"across {count_text(top_family['bucket_count'], 'bucket')}."
            if top_family
            else "No recent BucketSizeBytes metric is available."
        ),
        f"Stored bytes do not allocate the ${storage['current']} service cost or request cost to buckets; {count_text(storage['unmeasured_bucket_count'], 'bucket')} have no recent metric.",
        "Confirm retention and lifecycle policy for the largest bucket families.",
        owner,
    )
    kms_unit = decision_unit(
        "5",
        "kms",
        f"{'Observed KMS inventory' if kms['coverage'] == 'partial' else 'KMS inventory'} "
        f"contains {count_text(kms['customer_key_count'], 'customer-managed key')}; "
        f"{count_text(kms['unaliased_key_count'], 'key')} "
        f"{'has' if kms['unaliased_key_count'] == 1 else 'have'} no alias.",
        f"A deterministic regional sample covers {count_text(kms['sampled_unaliased_count'], 'unaliased key')}; "
        f"{count_text(kms['sampled_without_description_count'], 'sample')} "
        f"{'lacks' if kms['sampled_without_description_count'] == 1 else 'lack'} descriptions and "
        f"{kms['sampled_before_2022_count']} predate 2022.",
        (
            "KMS reads are partial, so counts are observed lower bounds. "
            if kms["coverage"] == "partial"
            else ""
        )
        + "The stable key-ID sample is bounded and not statistically representative; key deletion safety is not established.",
        (
            "Have the key owner verify pending-deletion windows and dependencies, including whether cancellation is needed; review enabled unaliased keys separately."
            if kms["sampled_pending_deletion_count"]
            else "Assign owners to enabled unaliased keys and prove ciphertext independence before scheduling deletion; resolve unknown states first."
        ),
        owner,
    )
    ca_unit = decision_unit(
        "6",
        "private-ca",
        f"{'Observed Private CA inventory' if private_ca['coverage'] == 'partial' else 'Private CA inventory'} "
        f"contains {count_text(private_ca['count'], 'authority', 'authorities')}, including {private_ca['active_count']} active.",
        f"The bounded ninety-day CloudTrail lookup observed {count_text(private_ca['recent_issue_events'], 'IssueCertificate event')} from {count_text(len(private_ca['recent_issue_callers']), 'caller')}.",
        (
            "Private CA reads are partial, so counts are observed lower bounds. "
            if private_ca["coverage"] == "partial"
            else ""
        )
        + "A bounded CloudTrail lookup proves observed issuance, not the absence of older or unlogged use.",
        "Treat an active issuing CA as workload-bearing; review rotation frequency before considering lifecycle changes.",
        owner,
    )
    service_evidence = "; ".join(
        f"{item['kind']} {item['count']}{' observed' if item['coverage'] == 'partial' else ''}"
        for item in services
    ) or "No named account-level inventory was observed."
    account_unit = decision_unit(
        "7",
        "account-services",
        f"Account-level billing is ${next(item for item in provider['non_vpc_categories'] if item['name'] == 'account-level services')['current']}.",
        service_evidence,
        (
            f"Deep inventory has {count_text(len(evidence_gaps), 'explicit read gap')}; unavailable surfaces remain unknown."
            if evidence_gaps
            else "Configured named-service inventory and ECR image reads completed; billing does not allocate shared-service cost to a workload."
        ),
        (
            "Restore ECR image reads before ranking retention across repositories; confirm the other named-service owners."
            if deep["image_coverage"]["status"] != "complete"
            else "Confirm the named service inventories and subscription users before changing capacity or seats."
        ),
        owner,
    )
    reclaim_unit = decision_unit(
        "8",
        "reclaim",
        f"The inventory exposes {count_text(reclaim['candidate_count'], 'candidate')} with a ${reclaim['estimated_monthly_cost']} monthly projection; "
        f"{reclaim['confirmed_count']} {'is' if reclaim['confirmed_count'] == 1 else 'are'} confirmed.",
        f"Candidates include {count_text(reclaim['ebs']['count'], 'EBS volume')}, {count_text(reclaim['load_balancer']['zero_healthy_count'], 'load balancer')}, and {count_text(reclaim['public_ipv4']['unassociated_count'], 'public IPv4 address')}.",
        "A candidate signal and projected base charge are not deletion approval or guaranteed saving.",
        "Validate the resource-level queue in cost order and record owner decisions.",
        owner,
    )
    recurrence_unit = decision_unit(
        "9",
        "recurrence",
        (
            "Recurring candidate evidence is unavailable because no compatible baseline exists."
            if recurring is None
            else f"{recurring} of {reclaim['candidate_count']} current candidates also exist in the previous snapshot."
        ),
        (
            "Eight weekly cost totals provide trend context; inventory recurrence has no earlier snapshot."
            if recurring is None
            else f"Eight weekly cost totals provide trend context; inventory recurrence uses only baseline {provider['resources']['baseline']}."
        ),
        "Cost persistence and inventory persistence are signals; neither establishes the creating defect.",
        "Preserve successive compatible snapshots and escalate only repeatable resource patterns.",
        owner,
    )
    action_unit = decision_unit(
        "10",
        "action",
        "The action queue separates no-workload validation, workload review, and defect evidence.",
        f"The queue carries {reclaim['candidate_count']} resource candidates, the largest non-EKS VPC signals, and named shared-service inventory.",
        "This report authorizes no cloud change and does not replace service-owner confirmation.",
        "Return owner decisions, then run the same evidence contract again before cleanup.",
        owner,
    )
    return [
        summary,
        ownership_unit,
        no_vpc_unit,
        storage_unit,
        kms_unit,
        ca_unit,
        account_unit,
        reclaim_unit,
        recurrence_unit,
        action_unit,
    ]


def build_aws_analysis(
    manifest: dict[str, Any],
    snapshot: dict[str, Any],
    previous_snapshot: dict[str, Any] | None,
    previous_history: dict[str, Any] | None = None,
) -> dict[str, Any]:
    scopes = validate_manifest(manifest)
    scope_id = snapshot.get("scope_id")
    if scope_id not in scopes or scopes[scope_id]["provider"] != "aws":
        raise ContractError("AWS snapshot scope is not configured")
    scope = scopes[scope_id]
    validate_aws_snapshot(snapshot, scope, "AWS snapshot")
    previous_keys: set[str] | None = None
    baseline_surfaces: set[str] | None = None
    baseline = "no_inventory_baseline"
    if previous_snapshot is not None:
        validate_aws_snapshot(
            previous_snapshot, scope, "previous AWS snapshot", allow_legacy=True
        )
        if previous_snapshot["coverage"]["status"] != "complete":
            raise ContractError("previous AWS snapshot must have complete coverage")
        if datetime.fromisoformat(previous_snapshot["collected_at"]) >= datetime.fromisoformat(
            snapshot["collected_at"]
        ):
            raise ContractError("previous AWS snapshot must predate the current snapshot")
        baseline_surfaces = set(previous_snapshot["inventory"]["surfaces"]) - (
            aws_evidence_gap_surfaces(previous_snapshot)
        )
        baseline = previous_snapshot["collected_at"]
    public_scope = {
        "id": scope_id,
        "label": scope["label"],
        "target": scope["account_id"],
        "billing_source": scope["billing_source"],
        "enabled": scope["enabled"],
    }
    provider: dict[str, Any] = {
        "scopes": [public_scope],
        "currency": "USD",
        "coverage": copy.deepcopy(snapshot["coverage"]),
    }
    analysis = {
        "version": ANALYSIS_VERSION,
        "synthetic": False,
        "mode": "live-proof",
        "report_title": (
            f"{scope['account_id'][-4:]} Account Cost Breakdown "
            "(Including Savings Plan Coverage)"
        ),
        "report_contract": {
            "reference": "0414 Account Cost Breakdown revision 50",
            "section_prefixes": list(AWS_REPORT_SECTION_PREFIXES),
            "subsection_prefixes": list(AWS_REPORT_SUBSECTION_PREFIXES),
            "interpretation_keys": list(REPORT_INTERPRETATION_KEYS),
            "raw_schema_terms_forbidden": list(AWS_REPORT_RAW_SCHEMA_TERMS),
        },
        "windows": copy.deepcopy(snapshot["windows"]),
        "providers": {"aws": provider},
    }
    if snapshot["coverage"]["status"] != "complete":
        analysis["decision_units"] = incomplete_aws_decision_units(
            public_scope, snapshot["coverage"]["gaps"]
        )
        return analysis

    previous = analyze_aws_window(snapshot, "previous")
    current = analyze_aws_window(snapshot, "current")
    previous_services = analyze_aws_service_usage(snapshot, "previous", previous["total"])
    current_services = analyze_aws_service_usage(snapshot, "current", current["total"])
    previous_unattributed = sum(previous["unattributed"].values(), Decimal(0))
    current_unattributed = sum(current["unattributed"].values(), Decimal(0))
    provider["cost"] = {
        "metric": snapshot["billing"]["metric"],
        "previous_total": money_text(previous["total"]),
        "current_total": money_text(current["total"]),
        "delta": money_text(current["total"] - previous["total"]),
        "delta_percent": change_percent_text(
            current["total"] - previous["total"], previous["total"]
        ),
        "groups": live_cost_delta(previous["groups"], current["groups"], current["total"]),
        "unattributed": {
            "previous": money_text(previous_unattributed),
            "current": money_text(current_unattributed),
            "delta": money_text(current_unattributed - previous_unattributed),
            "current_share_percent": percent_text(current_unattributed, current["total"]),
            "breakdown": live_cost_delta(
                previous["unattributed"], current["unattributed"], current["total"]
            ),
        },
        "comparison": {
            "metric": snapshot["billing"]["comparison_metric"],
            "previous_total": money_text(previous["comparison_total"]),
            "current_total": money_text(current["comparison_total"]),
            "current_difference": money_text(
                current["total"] - current["comparison_total"]
            ),
        },
    }
    provider["ownership"] = aws_ownership_rows(snapshot, previous, current)
    provider["weekly_history"] = aws_weekly_history(snapshot)
    provider["savings_plan_coverage"] = aws_savings_plan_coverage(snapshot)
    provider["vpc_evidence"] = aws_vpc_evidence(
        snapshot, current, provider["ownership"]
    )
    provider["vpc_services"] = [
        {
            "region": region,
            "vpc_id": vpc,
            "service": service,
            "current": money_text(amount),
            "previous": money_text(
                previous["vpc_services"].get((region, vpc, service), Decimal(0))
            ),
        }
        for (region, vpc, service), amount in sorted(
            current["vpc_services"].items(), key=lambda pair: -pair[1]
        )
    ]
    no_vpc_usage = aws_non_vpc_service_usage(
        previous,
        current,
        previous_services,
        current_services,
        provider["ownership"],
    )
    provider["non_vpc_categories"] = no_vpc_usage["categories"]
    provider["no_vpc_service_usage"] = no_vpc_usage["rows"]
    provider["compute_without_inventory_usage"] = no_vpc_usage[
        "compute_without_inventory"
    ]
    provider["account_level_service_usage"] = [
        item
        for item in no_vpc_usage["rows"]
        if item["category"] == "account-level services"
    ]
    provider["object_storage"] = aws_object_storage(snapshot, previous_services, current_services)
    provider["service_totals"] = [
        {
            "service": service,
            "current": money_text(
                sum(
                    (
                        amount
                        for (name, usage), amount in current_services.items()
                        if name == service
                    ),
                    Decimal(0),
                )
            ),
        }
        for service in sorted({name for name, usage in current_services})
    ]
    provider["deep_evidence"] = aws_deep_inventory_evidence(snapshot)
    comparable_surfaces = set(snapshot["inventory"]["surfaces"]) - (
        aws_evidence_gap_surfaces(snapshot)
    )
    if baseline_surfaces is not None:
        comparable_surfaces &= baseline_surfaces
        previous_keys = {
            aws_resource_key(row)
            for row in previous_snapshot["inventory"]["resources"]
            if row["kind"] in comparable_surfaces
        }
    current_keys = {
        aws_resource_key(row)
        for row in snapshot["inventory"]["resources"]
        if row["kind"] in comparable_surfaces
    }
    provider["resources"] = {
        "baseline": baseline,
        "collected_at": snapshot["collected_at"],
        "counts": copy.deepcopy(snapshot["inventory"]["counts"]),
        "new": sorted(current_keys - previous_keys) if previous_keys is not None else [],
        "removed": sorted(previous_keys - current_keys) if previous_keys is not None else [],
    }
    provider["reclaim_candidates"] = aws_reclaim_candidates(
        snapshot, current, previous_keys
    )
    provider["candidate_history"] = candidate_history(
        snapshot, provider["reclaim_candidates"], previous_history
    )
    provider["inventory_insights"] = aws_inventory_insights(
        snapshot,
        provider["reclaim_candidates"],
        previous_keys is not None,
    )
    displayed_total = (
        sum((Decimal(row["current"]) for row in provider["cost"]["groups"]), Decimal(0))
        + Decimal(provider["cost"]["unattributed"]["current"])
        + sum(current["other"].values(), Decimal(0))
    )
    if displayed_total != Decimal(provider["cost"]["current_total"]):
        raise ContractError("AWS normalized current cost buckets do not close to total")
    analysis["decision_units"] = aws_decision_units(provider, public_scope)
    return analysis


def check_aws_normalizer(
    manifest: dict[str, Any], interpretations: dict[str, str]
) -> None:
    scopes = validate_manifest(manifest)
    scope = next(
        scope
        for _, scope in sorted(scopes.items())
        if scope["provider"] == "aws" and scope["identity"] is not None
    )
    metric, comparison = AWS_BILLING_METRICS[scope["billing_source"]]
    first_day = date(2026, 1, 5)

    def metrics(cost: str, comparison_cost: str, quantity: str, unit: str) -> dict[str, Any]:
        return {
            metric: {"Amount": cost, "Unit": "USD"},
            comparison: {"Amount": comparison_cost, "Unit": "USD"},
            "UsageQuantity": {"Amount": quantity, "Unit": unit},
        }

    days = []
    service_usage_days = []
    for offset in range(14):
        start = first_day + timedelta(days=offset)
        time_period = {
            "Start": start.isoformat(),
            "End": (start + timedelta(days=1)).isoformat(),
        }
        groups = [
            {
                "Keys": ["us-west-2", "USW2-AmazonEKS-Hours:perCluster"],
                "Metrics": metrics("1.0005714", "0.8", "24", "Hrs"),
            },
            {
                "Keys": ["us-west-2", "USW2-EBS:VolumeUsage.gp3"],
                "Metrics": metrics("1", "1", "10", "GB-Mo"),
            },
            {
                "Keys": ["us-west-2", "USW2-TimedStorage-ByteHrs"],
                "Metrics": metrics("2.0005714", "2", "1", "GB-Mo"),
            },
        ]
        days.append({"TimePeriod": time_period, "Estimated": False, "Groups": groups})
        service_usage_days.append(
            {
                "TimePeriod": time_period,
                "Estimated": False,
                "Groups": [
                    {
                        "Keys": [
                            "Amazon Elastic Container Service for Kubernetes",
                            "USW2-AmazonEKS-Hours:perCluster",
                        ],
                        "Metrics": metrics("1.0005714", "0.8", "24", "Hrs"),
                    },
                    {
                        "Keys": ["EC2 - Other", "USW2-EBS:VolumeUsage.gp3"],
                        "Metrics": metrics("1", "1", "10", "GB-Mo"),
                    },
                    {
                        "Keys": [
                            "Amazon Simple Storage Service",
                            "USW2-TimedStorage-ByteHrs",
                        ],
                        "Metrics": metrics("2.0005714", "2", "1", "GB-Mo"),
                    },
                ],
            }
        )
    history_days = []
    history_start = date(2025, 11, 24)
    for offset in range(56):
        start = history_start + timedelta(days=offset)
        history_days.append(
            {
                "TimePeriod": {
                    "Start": start.isoformat(),
                    "End": (start + timedelta(days=1)).isoformat(),
                },
                "Estimated": False,
                "Total": {
                    metric: {"Amount": "4.0014286", "Unit": "USD"},
                    comparison: {"Amount": "3.80", "Unit": "USD"},
                },
                "Groups": [],
            }
        )
    cluster = {
        "kind": "eks",
        "region": "us-west-2",
        "id": "test-cluster",
        "weight": 1,
        "vpc_id": "vpc-test",
        "created": "2026-01-01",
        "cluster_ids": ["test-cluster"],
    }
    volume = {
        "kind": "ebs",
        "region": "us-west-2",
        "id": "vol-test",
        "weight": 10,
        "vpc_id": None,
        "created": "2026-01-01",
        "subtype": "gp3",
        "size_gib": 10,
        "state": "available",
        "cluster_ids": [],
    }
    vpc = {
        "kind": "vpc",
        "region": "us-west-2",
        "id": "vpc-test",
        "weight": 1,
        "name": "demo-test-vpc",
        "owner": "owner@example.com",
        "session": None,
        "state": "available",
        "cluster_ids": ["test-cluster"],
    }
    kms_key = {
        "kind": "kms-key",
        "region": "us-west-2",
        "id": "key-test",
        "weight": 1,
        "name": None,
        "alias_count": 0,
        "aws_managed": False,
        "sampled": True,
        "description_present": False,
        "created": "2021-01-01",
        "state": "Enabled",
        "cluster_ids": [],
    }
    private_ca = {
        "kind": "private-ca",
        "region": "us-west-2",
        "id": "arn:aws:acm-pca:us-west-2:000000000000:certificate-authority/test",
        "weight": 1,
        "name": "test-ca",
        "owner": None,
        "session": None,
        "created": "2021-02-01",
        "state": "ACTIVE",
        "subtype": "GENERAL_PURPOSE",
        "ca_type": "ROOT",
        "recent_issue_events": 3,
        "recent_issue_callers": ["arn:aws:iam::000000000000:role/test-caller"],
        "cluster_ids": [],
    }
    ecr = {
        "kind": "ecr-repository",
        "region": "us-west-2",
        "id": "arn:aws:ecr:us-west-2:000000000000:repository/test",
        "weight": 1,
        "name": "test",
        "created": "2025-01-01",
        "cluster_ids": [],
        "image_count": 2,
        "image_bytes": 120,
        "untagged_images": 1,
        "oldest_image": "2025-01-01",
        "latest_pull": None,
        "lifecycle_state": "none",
    }
    snapshot = {
        "version": AWS_SNAPSHOT_VERSION,
        "scope_id": scope["id"],
        "provider": "aws",
        "account_id": scope["account_id"],
        "billing_source": scope["billing_source"],
        "collected_at": "2026-01-20T00:00:00+00:00",
        "windows": {
            "previous": {"start": "2026-01-05", "end": "2026-01-12"},
            "current": {"start": "2026-01-12", "end": "2026-01-19"},
        },
        "coverage": {"status": "complete", "gaps": []},
        "billing": {
            "metric": metric,
            "comparison_metric": comparison,
            "page_count": 2,
            "days": days,
            "service_usage_page_count": 2,
            "service_usage_days": service_usage_days,
            "history": {
                "start": "2025-11-24",
                "end": "2026-01-19",
                "days": history_days,
            },
            "savings_plans": {
                "page_count": 1,
                "rows": [
                    {
                        "Attributes": {
                            "INSTANCE_TYPE_FAMILY": None,
                            "REGION": None,
                            "SERVICE": "Amazon Elastic Compute Cloud - Compute",
                        },
                        "Coverage": {
                            "SpendCoveredBySavingsPlans": "10.00",
                            "OnDemandCost": "4.00",
                            "TotalCost": "14.00",
                            "CoveragePercentage": "71.428571",
                        },
                        "TimePeriod": {"Start": "2026-01-05", "End": "2026-01-19"},
                    }
                ],
            },
        },
        "inventory": {
            "baseline": "no_inventory_baseline",
            "regions": ["us-west-2"],
            "surfaces": list(AWS_REQUIRED_SURFACES),
            "evidence_coverage": {"status": "complete", "gaps": []},
            "counts": {
                **{surface: 0 for surface in AWS_REQUIRED_SURFACES},
                "ebs": 1,
                "ecr-repository": 1,
                "eks": 1,
                "kms-key": 1,
                "private-ca": 1,
                "vpc": 1,
            },
            "resources": [volume, cluster, vpc, kms_key, private_ca, ecr],
            "object_storage": {
                "bucket_count": 3,
                "buckets": [
                    {
                        "bucket": "archive-test",
                        "empty": False,
                        "lifecycle_state": "enabled",
                        "lifecycle_actions": ["expire-noncurrent"],
                        "audit_errors": [],
                        "region": "us-west-2",
                        "created": "2025-01-01",
                        "bytes": 100,
                        "observed_at": "2026-01-19T00:00:00+00:00",
                    },
                    {
                        "bucket": "sample-test",
                        "region": "us-west-2",
                        "created": "2025-01-01",
                        "bytes": 900,
                        "observed_at": "2026-01-19T00:00:00+00:00",
                    },
                    {
                        "bucket": "unmeasured-test",
                        "region": "us-west-2",
                        "created": "2025-01-01",
                        "bytes": None,
                        "observed_at": None,
                    },
                ],
            },
        },
    }
    snapshot["inventory"]["resources"] = sorted(
        snapshot["inventory"]["resources"],
        key=lambda row: (row["kind"], row["region"], row["id"]),
    )
    missing_surface = copy.deepcopy(snapshot)
    missing_surface["inventory"]["surfaces"].remove("vpn")
    missing_surface["inventory"]["counts"].pop("vpn")
    try:
        build_aws_analysis(manifest, missing_surface, None)
    except ContractError:
        pass
    else:
        raise ContractError("AWS v3 validator accepted a missing inventory surface")

    for invalid_gap in (
        "malformed",
        "us-west-2:ecr_repository: denied",
        "us-west-2:config-recorder: denied",
        "not-a-region:ecr-repository: denied",
        "us-west-2:ecr-repository: ",
    ):
        malformed_gap = copy.deepcopy(snapshot)
        malformed_gap["inventory"]["evidence_coverage"] = {
            "status": "incomplete",
            "gaps": [invalid_gap],
        }
        try:
            build_aws_analysis(manifest, malformed_gap, None)
        except ContractError:
            pass
        else:
            raise ContractError("AWS v3 validator accepted an invalid evidence gap")

    for metric_name in (metric, comparison):
        broken_history = copy.deepcopy(snapshot)
        broken_history["billing"]["history"]["days"][-1]["Total"][metric_name][
            "Amount"
        ] = "999"
        try:
            build_aws_analysis(manifest, broken_history, None)
        except ContractError:
            pass
        else:
            raise ContractError(
                f"AWS v3 validator accepted history that disagrees for {metric_name}"
            )

    previous_gap = copy.deepcopy(snapshot)
    previous_gap["collected_at"] = "2026-01-13T00:00:00+00:00"
    previous_gap["inventory"]["evidence_coverage"] = {
        "status": "incomplete",
        "gaps": ["us-west-2:ecr-repository: denied"],
    }
    previous_gap["inventory"]["resources"] = [
        row
        for row in previous_gap["inventory"]["resources"]
        if row["kind"] != "ecr-repository"
    ]
    previous_gap["inventory"]["counts"]["ecr-repository"] = 0
    previous_gap_analysis = build_aws_analysis(manifest, snapshot, previous_gap)
    if previous_gap_analysis["providers"]["aws"]["resources"]["new"]:
        raise ContractError("AWS normalizer turned a previous evidence gap into a new resource")

    previous_config = copy.deepcopy(snapshot)
    previous_config["collected_at"] = "2026-01-13T00:00:00+00:00"
    previous_config["inventory"]["resources"].extend(
        [
            {
                "kind": "config-conformance-pack",
                "region": "us-west-2",
                "id": "pack-test",
                "weight": 1,
                "cluster_ids": [],
            },
            {
                "kind": "config-recorder",
                "region": "us-west-2",
                "id": "recorder-test",
                "weight": 1,
                "cluster_ids": [],
            },
        ]
    )
    previous_config["inventory"]["resources"].sort(
        key=lambda row: (row["kind"], row["region"], row["id"])
    )
    previous_config["inventory"]["counts"]["config-conformance-pack"] = 1
    previous_config["inventory"]["counts"]["config-recorder"] = 1
    current_config_gap = copy.deepcopy(snapshot)
    current_config_gap["inventory"]["evidence_coverage"] = {
        "status": "incomplete",
        "gaps": ["us-west-2:config: denied"],
    }
    current_config_gap_analysis = build_aws_analysis(
        manifest, current_config_gap, previous_config
    )
    if current_config_gap_analysis["providers"]["aws"]["resources"]["removed"]:
        raise ContractError("AWS normalizer turned a current Config gap into removed resources")

    previous_snapshot = copy.deepcopy(snapshot)
    previous_snapshot["version"] = AWS_LEGACY_SNAPSHOT_VERSION
    previous_snapshot["collected_at"] = "2026-01-13T00:00:00+00:00"
    previous_snapshot["billing"].pop("history")
    previous_snapshot["billing"].pop("savings_plans")
    previous_snapshot["inventory"]["resources"] = [cluster]
    previous_snapshot["inventory"]["surfaces"] = ["ebs", "eks"]
    previous_snapshot["inventory"]["counts"] = {"ebs": 0, "eks": 1}
    previous_snapshot["inventory"].pop("evidence_coverage")
    previous_snapshot["inventory"]["object_storage"] = {
        "bucket_count": 0,
        "bucket_sizes": [],
    }

    analysis = build_aws_analysis(manifest, snapshot, previous_snapshot)
    aws = analysis["providers"]["aws"]
    if (
        aws["object_storage"]["lifecycle_checked"] != 1
        or aws["deep_evidence"]["repositories"][0]["image_bytes"] != 120
    ):
        raise ContractError("AWS normalizer discarded retention or image-size evidence")
    if aws["cost"]["current_total"] != "28.01":
        raise ContractError("AWS normalizer produced an unexpected total")
    missing_images = copy.deepcopy(snapshot)
    missing_ecr = next(
        item
        for item in missing_images["inventory"]["resources"]
        if item["kind"] == "ecr-repository"
    )
    for field in (
        "image_count",
        "image_bytes",
        "untagged_images",
        "oldest_image",
        "latest_pull",
        "lifecycle_state",
    ):
        missing_ecr.pop(field, None)
    next(item for item in missing_images["inventory"]["resources"] if item["kind"] == "kms-key")[
        "state"
    ] = "PendingDeletion"
    missing_images["inventory"]["object_storage"]["buckets"][0]["empty"] = None
    missing_analysis = build_aws_analysis(manifest, missing_images, None)
    missing_deep = missing_analysis["providers"]["aws"]["deep_evidence"]
    missing_report = render_report(missing_analysis, interpretations)
    if (
        missing_deep["coverage"]["status"] != "incomplete"
        or missing_deep["image_coverage"]["status"] != "incomplete"
        or "Review ECR image retention" in missing_report
        or "Restore ECR image evidence" not in missing_report
    ):
        raise ContractError("unknown ECR images became complete coverage or a retention ranking")
    if (
        missing_deep["kms"]["sampled_pending_deletion_count"] != 1
        or "Review pending KMS deletions" not in missing_report
    ):
        raise ContractError("pending KMS deletion lost its existing-window review gate")
    if "| None |" in render_evidence(missing_analysis):
        raise ContractError("unknown evidence leaked an internal null value")
    rendered = render_report(analysis, interpretations)
    evidence = render_evidence(analysis)
    if (
        kms_key["id"] in rendered
        or private_ca["id"] in rendered
        or any(
            value not in evidence
            for value in (
                kms_key["id"],
                private_ca["id"],
                private_ca["recent_issue_callers"][0],
                *aws["resources"]["new"],
                *aws["resources"]["removed"],
            )
        )
    ):
        raise ContractError(
            "compact report lost full KMS, CA, caller or inventory-difference evidence"
        )
    shared_usage = copy.deepcopy(snapshot)
    for day in shared_usage["billing"]["service_usage_days"]:
        s3_group = next(
            group for group in day["Groups"] if group["Keys"][0] == "Amazon Simple Storage Service"
        )
        s3_group["Metrics"] = metrics("1.0005714", "1", "1", "GB-Mo")
        day["Groups"].append(
            {
                "Keys": [
                    "Amazon EC2 Container Registry (ECR)",
                    "USW2-TimedStorage-ByteHrs",
                ],
                "Metrics": metrics("1", "1", "1", "GB-Mo"),
            }
        )
    shared_analysis = build_aws_analysis(manifest, shared_usage, None)
    if shared_analysis["providers"]["aws"]["object_storage"]["current"] != "7.00":
        raise ContractError("S3 report included another service's shared Usage Type")
    for service in ("AmazonCloudWatch", "Amazon Simple Queue Service"):
        if aws_usage_route("USW2-TimedStorage-ByteHrs", service)[1] == "object storage":
            raise ContractError("storage-looking usage bypassed service attribution")
    history_snapshot = copy.deepcopy(snapshot)
    history_snapshot["collected_at"] = (
        datetime.fromisoformat(snapshot["collected_at"]) - timedelta(days=7)
    ).isoformat()
    history = candidate_history(history_snapshot, aws["reclaim_candidates"], None)
    combined = candidate_history(snapshot, aws["reclaim_candidates"], history)
    if len(combined["observations"]) != 2:
        raise ContractError("candidate history discarded a prior daily observation")
    invalid_history = copy.deepcopy(history)
    invalid_history["account_id"] = "000000000000"
    try:
        candidate_history(snapshot, aws["reclaim_candidates"], invalid_history)
    except ContractError:
        pass
    else:
        raise ContractError("candidate history accepted another account")
    future = copy.deepcopy(snapshot)
    next(item for item in future["inventory"]["resources"] if item["kind"] == "ebs")["created"] = (
        snapshot["windows"]["current"]["end"]
    )
    future_candidates = build_aws_analysis(manifest, future, None)["providers"]["aws"][
        "reclaim_candidates"
    ]
    if not any(
        item["cost_estimate_status"] == "outside matched billing" for item in future_candidates
    ):
        raise ContractError(
            "newly created reclaim candidates disappeared outside the billing window"
        )
    if aws["resources"]["new"] != ["us-west-2/ebs/vol-test"]:
        raise ContractError("AWS normalizer produced an unexpected inventory diff")
    if (
        len(aws["reclaim_candidates"]) != 1
        or aws["reclaim_candidates"][0]["classification"] != "candidate"
        or aws["reclaim_candidates"][0]["estimated_monthly_cost"] != "30.00"
    ):
        raise ContractError("AWS normalizer weakened the reclaim candidate boundary")
    if aws["ownership"]["rows"][0]["vpc"] != "vpc-test":
        raise ContractError("AWS normalizer lost VPC ownership context")
    no_vpc = next(item for item in aws["ownership"]["rows"] if item["vpc"] == "No VPC")
    if no_vpc["current"] != "21.01" or sum(
        (Decimal(item["current"]) for item in aws["non_vpc_categories"]), Decimal(0)
    ) != Decimal(no_vpc["current"]):
        raise ContractError("AWS no-VPC categories did not close to the ownership remainder")
    account_level = next(
        item
        for item in aws["non_vpc_categories"]
        if item["name"] == "account-level services"
    )
    if (
        aws["account_level_service_usage"][0]["service"]
        != "Amazon Simple Storage Service"
        or sum(
            (
                Decimal(item["current"])
                for item in aws["account_level_service_usage"]
            ),
            Decimal(0),
        )
        != Decimal(account_level["current"])
        or aws["object_storage"]["measured_bytes"] != 1000
        or aws["object_storage"]["unmeasured_bucket_count"] != 1
        or len(aws["weekly_history"]) != 8
        or aws["savings_plan_coverage"]["covered_spend"] != "10.00"
    ):
        raise ContractError("AWS normalizer lost service, usage, or bucket grouping")
    if (
        aws["deep_evidence"]["kms"]["unaliased_key_count"] != 1
        or aws["deep_evidence"]["private_ca"]["recent_issue_events"] != 3
        or len(analysis["decision_units"]) != len(REPORT_INTERPRETATION_KEYS)
        or [item["key"] for item in analysis["decision_units"]]
        != list(REPORT_INTERPRETATION_KEYS)
    ):
        raise ContractError("AWS normalizer lost deterministic decision evidence")
    if sum(
        (Decimal(item["current"]) for item in aws["ownership"]["summary_pie"]),
        Decimal(0),
    ) != Decimal(aws["ownership"]["current_total"]):
        raise ContractError("AWS summary pie did not close to current total")
    if aws["inventory_insights"]["reclaim"]["recurring_candidate_count"] != 0:
        raise ContractError("AWS normalizer lost the available inventory baseline")
    no_baseline = build_aws_analysis(manifest, snapshot, None)
    if (
        no_baseline["providers"]["aws"]["inventory_insights"]["reclaim"][
            "recurring_candidate_count"
        ]
        is not None
    ):
        raise ContractError("AWS normalizer turned an absent inventory baseline into zero")

    partial = copy.deepcopy(snapshot)
    partial["inventory"]["evidence_coverage"] = {
        "status": "incomplete",
        "gaps": [
            "us-west-2:config: denied",
            "us-west-2:ecr-repository: denied",
            "us-west-2:kms-key: denied",
            "us-west-2:private-ca: denied",
        ],
    }
    partial_analysis = build_aws_analysis(manifest, partial, None)
    partial_deep = partial_analysis["providers"]["aws"]["deep_evidence"]
    if (
        partial_deep["kms"]["coverage"] != "partial"
        or partial_deep["private_ca"]["coverage"] != "partial"
        or next(
            item
            for item in partial_deep["account_services"]
            if item["kind"] == "config-recorder"
        )["coverage"]
        != "partial"
        or next(
            item
            for item in partial_deep["account_services"]
            if item["kind"] == "config-conformance-pack"
        )["coverage"]
        != "partial"
        or next(
            item
            for item in partial_deep["account_services"]
            if item["kind"] == "ecr-repository"
        )["coverage"]
        != "partial"
    ):
        raise ContractError("AWS normalizer lost per-surface evidence coverage")

    incomplete = copy.deepcopy(snapshot)
    incomplete["coverage"] = {"status": "incomplete", "gaps": ["billing: denied"]}
    incomplete_analysis = build_aws_analysis(manifest, incomplete, None)
    incomplete_aws = incomplete_analysis["providers"]["aws"]
    if any(
        field in incomplete_aws for field in ("cost", "resources", "reclaim_candidates")
    ):
        raise ContractError("incomplete AWS normalization leaked conclusions")

    for report_analysis in (analysis, no_baseline, partial_analysis, incomplete_analysis):
        report = render_report(report_analysis, interpretations)
        validate_report(report_analysis, interpretations, report)
        if report_analysis is no_baseline and (
            "Recurrence was not evaluated because no earlier inventory snapshot was available."
            not in report
            or "# 5. Reclaim candidates" not in report
            or "same-current-window" not in report.casefold()
            or "Do not infer complete service-level attribution" not in report
            or "# 4. Account-level services" not in report
            or "Object storage by bucket" not in report
            or "Object storage families" not in report
            or "KMS key ownership evidence" not in report
            or "Private CA workload evidence" not in report
            or "Named account-level inventory" not in report
            or "Resource-level validation queue" not in report
            or "Method and data notes" in report
        ):
            raise ContractError("AWS deterministic report lost a required evidence boundary")
        if report_analysis is partial_analysis and (
            "KMS reads are partial, so counts are observed lower bounds" not in report
            or "Private CA reads are partial, so counts are observed lower bounds"
            not in report
            or "| config-recorder | 0 observed | PARTIAL |" not in report
            or "ecr-repository 1 observed" not in report
            or "all counts are observed lower bounds" not in report
        ):
            raise ContractError("AWS report hid partial deep evidence")
    try:
        validate_report(
            analysis,
            interpretations,
            render_report(analysis, interpretations).replace("\\$28.01", "\\$0.01", 1),
        )
    except ContractError:
        pass
    else:
        raise ContractError("AWS report validator accepted a changed deterministic value")


def check_contract(manifest: dict[str, Any], fixture: dict[str, Any]) -> None:
    interpretations = {key: "context" for key in REPORT_INTERPRETATION_KEYS}
    complete = build_analysis(manifest, fixture, "complete")
    if [item["key"] for item in complete["decision_units"]] != list(
        REPORT_INTERPRETATION_KEYS
    ):
        raise ContractError("synthetic replay lost deterministic decision units")
    if any(item["coverage"]["status"] != "complete" for item in complete["providers"].values()):
        raise ContractError("complete scenario did not remain complete")
    if sum(len(item["scopes"]) for item in complete["providers"].values()) != len(
        manifest["scopes"]
    ):
        raise ContractError("complete scenario did not retain every configured scope")
    if complete["providers"]["aws"]["cost"]["delta"] != "18.00":
        raise ContractError("complete scenario produced an unexpected AWS delta")
    if complete["providers"]["aws"]["resources"]["new"] != ["synthetic-aws-new-volume"]:
        raise ContractError("complete scenario produced an unexpected AWS resource diff")
    if change_percent_text(Decimal("1"), Decimal(0)) != "n/a":
        raise ContractError("zero-baseline percentage did not fail closed")

    incomplete = build_analysis(manifest, fixture, "incomplete")
    gcp = incomplete["providers"]["gcp"]
    if gcp["coverage"]["status"] != "incomplete":
        raise ContractError("incomplete scenario did not fail closed")
    if any(field in gcp for field in ("cost", "resources", "reclaim_candidates")):
        raise ContractError("incomplete provider leaked conclusions")
    for report_analysis in (complete, incomplete):
        report = render_report(report_analysis, interpretations)
        validate_report(report_analysis, interpretations, report)
    for copied_number in (
        "The latest spend is the lowest recorded.",
        "assigned public addresses",
        "context extra",
        "",
    ):
        invalid_interpretations = copy.deepcopy(interpretations)
        invalid_interpretations["1"] = copied_number
        try:
            render_report(complete, invalid_interpretations)
        except ContractError:
            pass
        else:
            raise ContractError("report renderer accepted ungrounded Agent prose")

    invalid = copy.deepcopy(manifest)
    invalid["scopes"][0]["enabled"] = True
    invalid["scopes"][0]["identity"] = None
    try:
        validate_manifest(invalid)
    except ContractError:
        pass
    else:
        raise ContractError("enabled scope without identity passed validation")
    check_aws_normalizer(manifest, interpretations)


def validate_decision_units(value: Any) -> None:
    fields = {"key", "section", "headline", "evidence", "boundary", "decision", "gate"}
    if (
        not isinstance(value, list)
        or [item.get("key") for item in value if isinstance(item, dict)]
        != list(REPORT_INTERPRETATION_KEYS)
        or any(
            not isinstance(item, dict)
            or set(item) != fields
            or any(not isinstance(item[field], str) or not item[field] for field in fields)
            for item in value
        )
    ):
        raise ContractError("analysis decision units do not match the deterministic report contract")


def validate_interpretations(value: dict[str, Any]) -> dict[str, str]:
    """The editor selects grounded wording; unverified prose never reaches publication."""
    if set(value) != set(REPORT_INTERPRETATION_KEYS) or any(
        not isinstance(choice, str) or choice not in {"action", "context"}
        for choice in value.values()
    ):
        raise ContractError("Agent must select action or context for exactly keys 1 through 10")
    return value


def markdown_table(headers: list[str], rows: list[list[Any]]) -> str:
    def cell(value: Any) -> str:
        text = " ".join(str(value).splitlines()).strip().replace("`", "'")
        text = text.replace("<", "&lt;").replace(">", "&gt;")
        return re.sub(r"([|\[\]])", r"\\\1", text)

    return "\n".join(
        [
            "| " + " | ".join(cell(value) for value in headers) + " |",
            "| " + " | ".join("---" for _ in headers) + " |",
            *["| " + " | ".join(cell(value) for value in row) + " |" for row in rows],
        ]
    )


def report_currency(currency: str, value: str, signed: bool = False) -> str:
    amount = Decimal(value)
    sign = "+" if signed and amount > 0 else "-" if amount < 0 else ""
    symbol = "\\$" if currency == "USD" else f"{currency} "
    return f"{sign}{symbol}{abs(amount):,.2f}"


def report_percent(value: str, signed: bool = False) -> str:
    if value == "n/a":
        return value
    amount = Decimal(value)
    sign = "+" if signed and amount > 0 else "-" if amount < 0 else ""
    return f"{sign}{abs(amount):.2f}%"


def report_bytes(value: int) -> str:
    amount = Decimal(value)
    for unit in ("bytes", "KiB", "MiB", "GiB", "TiB", "PiB"):
        if amount < 1024 or unit == "PiB":
            return f"{amount:,.0f} {unit}" if unit == "bytes" else f"{amount:,.2f} {unit}"
        amount /= 1024
    raise AssertionError("unreachable")


def append_interpretation(lines: list[str], interpretations: dict[str, str], index: int) -> None:
    lines.extend(["", interpretations[str(index)], ""])


def render_synthetic_report(analysis: dict[str, Any], interpretations: dict[str, str]) -> str:
    providers = analysis["providers"]
    lines = [
        f"# {analysis['report_title']}",
        "",
        "Every cost and resource fact in this report is synthetic.",
        "",
        "## 1. Summary",
        "",
    ]
    summary_rows = []
    for name in PROVIDERS:
        provider = providers[name]
        cost = provider.get("cost")
        summary_rows.append(
            [
                name.upper(),
                provider["coverage"]["status"].upper(),
                report_currency(provider["currency"], cost["current_total"]) if cost else "Unavailable",
                report_currency(provider["currency"], cost["previous_total"]) if cost else "Unavailable",
                report_currency(provider["currency"], cost["delta"], signed=True) if cost else "Unavailable",
                len(provider["scopes"]),
            ]
        )
    lines.append(
        markdown_table(
            ["Provider", "Coverage", "Current", "Previous", "Delta", "Configured scopes"],
            summary_rows,
        )
    )
    append_interpretation(lines, interpretations, 1)

    lines.extend(["## 2. Cost groups", ""])
    group_rows = []
    for name in PROVIDERS:
        provider = providers[name]
        for item in (provider.get("cost") or {}).get("groups", []):
            group_rows.append(
                [
                    name.upper(),
                    item["name"],
                    report_currency(provider["currency"], item["current"]),
                    report_percent(item["current_share_percent"]),
                    report_currency(provider["currency"], item["previous"]),
                    report_currency(provider["currency"], item["delta"], signed=True),
                ]
            )
    lines.append(
        markdown_table(
            ["Provider", "Group", "Current", "Share", "Previous", "Delta"],
            group_rows
            or [["Unavailable", "No complete provider groups", "Unavailable", "Unavailable", "Unavailable", "Unavailable"]],
        )
    )
    append_interpretation(lines, interpretations, 2)

    lines.extend(["## 3. Unattributed cost", ""])
    unattributed_rows = []
    for name in PROVIDERS:
        provider = providers[name]
        item = (provider.get("cost") or {}).get("unattributed")
        if item:
            unattributed_rows.append(
                [
                    name.upper(),
                    report_currency(provider["currency"], item["current"]),
                    report_percent(item["current_share_percent"]),
                    report_currency(provider["currency"], item["previous"]),
                    report_currency(provider["currency"], item["delta"], signed=True),
                ]
            )
    lines.append(
        markdown_table(
            ["Provider", "Current", "Share", "Previous", "Delta"],
            unattributed_rows
            or [["Unavailable", "Unavailable", "Unavailable", "Unavailable", "No complete provider data"]],
        )
    )
    append_interpretation(lines, interpretations, 3)

    lines.extend(["## 4. Account-level services", ""])
    service_rows = []
    for name in PROVIDERS:
        provider = providers[name]
        for item in (provider.get("cost") or {}).get("other_services", []):
            service_rows.append(
                [
                    name.upper(),
                    item["name"],
                    report_currency(provider["currency"], item["current"]),
                    report_percent(item["current_share_percent"]),
                    report_currency(provider["currency"], item["previous"]),
                    report_currency(provider["currency"], item["delta"], signed=True),
                ]
            )
    lines.append(
        markdown_table(
            ["Provider", "Service", "Current", "Share", "Previous", "Delta"],
            service_rows
            or [["Unavailable", "No complete provider services", "Unavailable", "Unavailable", "Unavailable", "Unavailable"]],
        )
    )
    append_interpretation(lines, interpretations, 4)

    lines.extend(["## 5. Reclaim candidates", ""])
    candidate_rows = []
    for name in PROVIDERS:
        provider = providers[name]
        for item in provider.get("reclaim_candidates", []):
            candidate_rows.append(
                [
                    name.upper(),
                    item["id"],
                    item["classification"],
                    "; ".join(item["signals"]),
                    report_currency(provider["currency"], item["estimated_monthly_cost"]),
                ]
            )
    lines.append(
        markdown_table(
            ["Provider", "Resource", "Class", "Signals", "Monthly estimate"],
            candidate_rows or [["Unavailable", "None", "Unavailable", "Unavailable", "Unavailable"]],
        )
    )
    append_interpretation(lines, interpretations, 5)

    lines.extend(["## 6. Resource changes", ""])
    resource_rows = []
    for name in PROVIDERS:
        provider = providers[name]
        resources = provider.get("resources") or {}
        resource_rows.extend([name.upper(), "New", item] for item in resources.get("new", []))
        resource_rows.extend(
            [name.upper(), "Removed", item] for item in resources.get("removed", [])
        )
    lines.append(
        markdown_table(
            ["Provider", "Change", "Resource"],
            resource_rows or [["Unavailable", "None", "No complete resource comparison"]],
        )
    )
    append_interpretation(lines, interpretations, 6)

    lines.extend(["## 7. Action plan", ""])
    action_rows = []
    for name in PROVIDERS:
        provider = providers[name]
        for item in provider.get("reclaim_candidates", []):
            action_rows.append(
                [
                    name.upper(),
                    item["id"],
                    item["classification"],
                    "; ".join(item["evidence"]),
                    "Confirm the evidence before any change",
                ]
            )
    lines.append(
        markdown_table(
            ["Provider", "Resource", "Class", "Evidence", "Next step"],
            action_rows
            or [["Unavailable", "None", "Unavailable", "No complete candidate data", "Collect evidence"]],
        )
    )
    append_interpretation(lines, interpretations, 7)
    for index in range(8, 11):
        append_interpretation(lines, interpretations, index)
    lines.extend([AWS_REPORT_GENERATED_MARKER, ""])
    return "\n".join(lines)


def render_incomplete_aws_report(
    analysis: dict[str, Any], interpretations: dict[str, str]
) -> str:
    aws = analysis["providers"]["aws"]
    scope = aws["scopes"][0]
    windows = analysis["windows"]
    gaps = "; ".join(aws["coverage"]["gaps"])
    unavailable = markdown_table(
        ["Status", "Evidence boundary"],
        [["INCOMPLETE", gaps]],
    )
    lines = [
        AWS_REPORT_SECTION_PREFIXES[0],
        "",
        markdown_table(
            ["Metric", "Current / status", "Previous / comparison", "Delta / note"],
            [
                [
                    "Window",
                    f"{windows['current']['start']} to {windows['current']['end']}",
                    f"{windows['previous']['start']} to {windows['previous']['end']}",
                    "Adjacent billing windows",
                ],
                [
                    "Account / scope",
                    f"{scope['label']}; {scope['id']}; account {scope['target']}",
                    "Same configured scope",
                    scope["billing_source"],
                ],
                ["Coverage", "INCOMPLETE", "Configured calls did not complete", gaps],
            ],
        ),
        "",
        "This report is incomplete. Unsupported totals, deltas, ownership, and reclaim conclusions are omitted.",
    ]
    append_interpretation(lines, interpretations, 1)
    for index, heading in enumerate(AWS_REPORT_SECTION_PREFIXES[1:6], 2):
        lines.extend([heading, "", unavailable])
        append_interpretation(lines, interpretations, index)
    lines.extend([AWS_REPORT_SECTION_PREFIXES[6], ""])
    for subsection in AWS_REPORT_SUBSECTION_PREFIXES:
        lines.extend([subsection, "", unavailable, ""])
    append_interpretation(lines, interpretations, 7)
    for index in range(8, 11):
        append_interpretation(lines, interpretations, index)
    lines.extend([AWS_REPORT_GENERATED_MARKER, ""])
    return "\n".join(lines)


def render_aws_report(analysis: dict[str, Any], interpretations: dict[str, str]) -> str:
    aws = analysis["providers"]["aws"]
    if aws["coverage"]["status"] != "complete":
        return render_incomplete_aws_report(analysis, interpretations)
    scope = aws["scopes"][0]
    currency = aws["currency"]
    cost = aws["cost"]
    ownership = aws["ownership"]
    insights = aws["inventory_insights"]
    reclaim = insights["reclaim"]
    resources = aws["resources"]
    windows = analysis["windows"]

    def money(value: str, signed: bool = False) -> str:
        return report_currency(currency, value, signed)

    recurring = reclaim["recurring_candidate_count"]
    recurrence_status = (
        "Not evaluated; no earlier inventory snapshot"
        if recurring is None
        else f"{recurring} of {reclaim['candidate_count']} current candidates; baseline {resources['baseline']}"
    )
    summary_rows = [
        [
            "Window",
            f"{windows['current']['start']} to {windows['current']['end']}",
            f"{windows['previous']['start']} to {windows['previous']['end']}",
            "Adjacent "
            f"{(date.fromisoformat(windows['current']['end']) - date.fromisoformat(windows['current']['start'])).days}"
            "-day windows",
        ],
        [
            "Account / scope",
            f"{scope['label']}; {scope['id']}; account {scope['target']}",
            "Same configured scope",
            scope["billing_source"],
        ],
        [
            f"Report metric: {cost['metric']}",
            money(cost["current_total"]),
            money(cost["previous_total"]),
            f"{money(cost['delta'], signed=True)} ({report_percent(cost['delta_percent'], signed=True)})",
        ],
        [
            f"Comparison metric: {cost['comparison']['metric']}",
            money(cost["comparison"]["current_total"]),
            money(cost["comparison"]["previous_total"]),
            "Same metric across adjacent windows",
        ],
        [
            "Same-current-window metric difference",
            money(cost["comparison"]["current_difference"], signed=True),
            f"{cost['metric']} versus {cost['comparison']['metric']}",
            "Not a previous-window change or service-level Savings Plan attribution",
        ],
        [
            "Savings Plans coverage",
            money(aws["savings_plan_coverage"]["covered_spend"]),
            f"{report_percent(aws['savings_plan_coverage']['coverage_percent'])} of covered plus on-demand cost",
            "Direct Cost Explorer Savings Plans coverage over both adjacent windows",
        ],
        *[
            [
                item["name"],
                money(item["current"]),
                "Current ownership bucket",
                report_percent(item["current_share_percent"]),
            ]
            for item in ownership["summary_pie"]
        ],
        [
            "Reclaim evidence",
            f"{reclaim['confirmed_count']} confirmed; {reclaim['candidate_count']} candidates",
            f"Monthly candidate estimate {money(reclaim['estimated_monthly_cost'])}",
            "Every candidate remains evidence-gated",
        ],
        ["Inventory recurrence", recurrence_status, "Point-in-time inventory", "Not defect proof"],
    ]
    delta = Decimal(cost["delta"])
    if delta > 0:
        movement = f"increased by {money(cost['delta'])}"
    elif delta < 0:
        movement = f"decreased by {money(str(abs(delta)))}"
    else:
        movement = "did not change"
    lines = [
        AWS_REPORT_SECTION_PREFIXES[0],
        "",
        markdown_table(
            ["Metric", "Current / status", "Previous / comparison", "Delta / note"],
            summary_rows,
        ),
        "",
        f"Current {cost['metric']} spend {movement} "
        f"({report_percent(cost['delta_percent'], signed=True)}) versus the adjacent previous "
        f"window. The same-current-window difference from {cost['comparison']['metric']} is "
        f"{money(cost['comparison']['current_difference'], signed=True)}; it is a metric "
        "comparison, not a window movement or service-level Savings Plan attribution.",
    ]
    if ownership["summary_pie"]:
        lines.extend(
            [
                "",
                '<whiteboard type="mermaid">',
                "pie showData",
                "    title Current cost ownership (100.00%)",
                *[
                    f'    "{item["name"]}" : {item["current"]}'
                    for item in ownership["summary_pie"]
                ],
                "</whiteboard>",
            ]
        )
    lines.extend(
        [
            "",
            "## Eight-week spend history",
            "",
            markdown_table(
                ["Week (UTC)", cost["metric"], cost["comparison"]["metric"]],
                [
                    [
                        f"{item['start']} to {item['end']}",
                        money(item["metric"]),
                        money(item["comparison_metric"]),
                    ]
                    for item in aws["weekly_history"]
                ],
            ),
            "",
            "## Savings Plans coverage by service",
            "",
            markdown_table(
                ["AWS service", "Covered spend", "On-demand cost", "Total cost", "Coverage"],
                [
                    [
                        item["service"],
                        money(item["covered_spend"]),
                        money(item["on_demand_cost"]),
                        money(item["total_cost"]),
                        report_percent(item["coverage_percent"]),
                    ]
                    for item in aws["savings_plan_coverage"]["services"]
                ]
                or [["No Savings Plans-covered service", money("0"), money("0"), money("0"), "0.00%"]],
            ),
        ]
    )
    minimum = min(aws["weekly_history"], key=lambda item: Decimal(item["metric"]))
    storage = aws["object_storage"]
    repositories = aws["deep_evidence"]["repositories"]
    lines.extend(
        [
            "",
            f"The lowest week in the displayed history was {minimum['start']} to {minimum['end']} "
            f"at {money(minimum['metric'])}.",
            "",
            f"Billing and base inventory reads completed. Named-service deep reads: {aws['deep_evidence']['coverage']['status']}; "
            f"{count_text(len(aws['deep_evidence']['coverage']['gaps']), 'read gap')}. S3 retention policies checked: "
            f"{storage['lifecycle_checked']} of {storage['bucket_count']}; version-empty checks: "
            f"{storage['empty_checked']} of {storage['bucket_count']}. ECR image inventories checked: "
            f"{sum('image_count' in item for item in repositories)} of {len(repositories)}. Missing reads remain unknown.",
            "",
            "Full resource identifiers, all candidates, bucket policies and repository details: "
            + (
                f"[report evidence]({analysis['evidence_url']})"
                if analysis.get("evidence_url")
                else "see the companion report-evidence.md artifact."
            ),
        ]
    )
    append_interpretation(lines, interpretations, 1)

    ownership_rows = ownership["rows"]
    lines.extend(
        [
            AWS_REPORT_SECTION_PREFIXES[1],
            "",
            markdown_table(
                ["VPC", "EKS cluster", "Region", "Current cost", "Share", "Previous cost", "Delta"],
                [
                    [
                        item["vpc"],
                        item["eks_cluster"],
                        item["region"],
                        money(item["current"]),
                        report_percent(item["current_share_percent"]),
                        money(item["previous"]),
                        money(item["delta"], signed=True),
                    ]
                    for item in ownership_rows
                ]
                + [
                    [
                        "Total",
                        "",
                        "",
                        money(ownership["current_total"]),
                        "100.00%",
                        money(ownership["previous_total"]),
                        money(cost["delta"], signed=True),
                    ]
                ],
            ),
        ]
    )
    largest = max(ownership_rows, key=lambda item: Decimal(item["current"]))
    eks_rows = [item for item in ownership_rows if item["eks_cluster"] != "none"]
    if eks_rows:
        largest_eks = max(eks_rows, key=lambda item: Decimal(item["current"]))
        ownership_highlight = (
            f"The largest ownership bucket is {largest['vpc']} at {money(largest['current'])} "
            f"({report_percent(largest['current_share_percent'])}). The largest EKS-linked row is "
            f"{largest_eks['eks_cluster']} at {money(largest_eks['current'])}, with an adjacent-window "
            f"change of {money(largest_eks['delta'], signed=True)}."
        )
    else:
        ownership_highlight = (
            f"The largest ownership bucket is {largest['vpc']} at {money(largest['current'])} "
            f"({report_percent(largest['current_share_percent'])}); no EKS-linked row is present."
        )
    vpc_rows = aws["vpc_evidence"]["rows"]
    lines.extend(
        [
            "",
            ownership_highlight,
            "",
            "## Non-EKS VPC decision evidence",
            "",
            markdown_table(
                [
                    "VPC / Name",
                    "Region",
                    "Current cost",
                    "Share",
                    "Purpose or owner signal",
                    "State",
                    "Matched inventory",
                ],
                [
                    [
                        f"{item['vpc_id']} / {item['name']}",
                        item["region"],
                        money(item["current"]),
                        report_percent(item["share_percent"]),
                        "; ".join(
                            value
                            for value in (
                                item["purpose_signal"],
                                f"owner {item['owner']}" if item["owner"] else "",
                                f"session {item['session']}" if item["session"] else "",
                            )
                            if value
                        ),
                        item["state"],
                        item["inventory"],
                    ]
                    for item in vpc_rows[:10]
                ]
                or [
                    [
                        "None",
                        "None",
                        money("0"),
                        "0.00%",
                        "No non-EKS VPC",
                        "Unknown",
                        "None",
                    ]
                ],
            ),
            "",
            "Names and cp9 tags are routing signals only. They do not prove that a VPC is a "
            "leftover, personal, or safe to remove.",
        ]
    )
    composition_rows = []
    for item in sorted(eks_rows, key=lambda item: -Decimal(item["current"]))[:3]:
        components = [
            row
            for row in aws["vpc_services"]
            if row["region"] == item["region"] and row["vpc_id"] == item["vpc"]
        ]
        composition_rows.append(
            [
                item["eks_cluster"],
                "; ".join(f"{row['service']} {money(row['current'])}" for row in components[:4]),
                "Inspect "
                + ", ".join(row["service"] for row in components[:2])
                + " usage before changing capacity",
            ]
        )
    lines.extend(
        [
            "",
            "## Main cluster cost drivers",
            "",
            markdown_table(
                ["Cluster", "Largest allocated service components", "Next decision"],
                composition_rows
                or [["None", "No matched components", "Restore attribution evidence"]],
            ),
            "",
            "Component amounts use the same inventory-weighted allocation as the VPC table; they are not resource-level invoices.",
        ]
    )
    append_interpretation(lines, interpretations, 2)

    no_vpc = next(item for item in ownership_rows if item["vpc"] == "No VPC")
    detail_categories = [
        "network traffic",
        "resources without a VPC or matching inventory",
        "resource add-ons without per-resource evidence",
    ]
    no_vpc_detail_rows = []
    for category in detail_categories:
        category_rows = sorted(
            (
                item
                for item in aws["no_vpc_service_usage"]
                if item["category"] == category
            ),
            key=lambda item: (-Decimal(item["current"]), item["service"], item["usage_type"]),
        )
        for index, item in enumerate(category_rows[:5]):
            no_vpc_detail_rows.append(
                [
                    category if index == 0 else "",
                    item["service"],
                    item["usage_type"],
                    money(item["current"]),
                    report_percent(item["share_of_no_vpc_percent"]),
                    money(item["previous"]),
                    money(item["delta"], signed=True),
                ]
            )
        remainder = category_rows[5:]
        if remainder:
            current_remainder = sum(Decimal(item["current"]) for item in remainder)
            previous_remainder = sum(Decimal(item["previous"]) for item in remainder)
            no_vpc_detail_rows.append(
                [
                    "",
                    f"Remaining {len(remainder)} service and usage pairs",
                    "Grouped remainder",
                    money(money_text(current_remainder)),
                    report_percent(percent_text(current_remainder, Decimal(no_vpc["current"]))),
                    money(money_text(previous_remainder)),
                    money(money_text(current_remainder - previous_remainder), signed=True),
                ]
            )
    lines.extend(
        [
            AWS_REPORT_SECTION_PREFIXES[2],
            "",
            f"No VPC accounts for {money(no_vpc['current'])} "
            f"({report_percent(no_vpc['current_share_percent'])}) of current cost. The four "
            "categories below close exactly to that remainder.",
            "",
            markdown_table(
                ["No VPC category", "Current", "Share", "Previous", "Delta"],
                [
                    [
                        item["name"],
                        money(item["current"]),
                        report_percent(item["current_share_percent"]),
                        money(item["previous"]),
                        money(item["delta"], signed=True),
                    ]
                    for item in aws["non_vpc_categories"]
                ],
            ),
            "",
            markdown_table(
                [
                    "No VPC category",
                    "AWS service",
                    "Usage type",
                    "Current",
                    "Share of No VPC",
                    "Previous",
                    "Delta",
                ],
                no_vpc_detail_rows
                or [["None", "No charges", "No usage", money("0"), "0.00%", money("0"), money("0")]],
            ),
            "",
            "Each detail row assigns a no-VPC charge to an AWS service and usage type. The table "
            "shows the largest pairs for network traffic, unmatched resources, and resource add-ons; "
            "section 4 expands the account-level services row. Billing groups show where AWS charged "
            "the account, but they do not identify a workload owner.",
        ]
    )
    compute_rows = aws["compute_without_inventory_usage"]
    compute_current = sum(Decimal(item["current"]) for item in compute_rows)
    compute_previous = sum(Decimal(item["previous"]) for item in compute_rows)
    compute_delta = compute_current - compute_previous
    if compute_delta > 0:
        compute_movement = f"rose by {money(money_text(compute_delta))}"
    elif compute_delta < 0:
        compute_movement = f"fell by {money(money_text(abs(compute_delta)))}"
    else:
        compute_movement = "did not change"
    if compute_rows:
        lines.extend(
            [
                "",
                f"## Compute without matching inventory ({money(money_text(compute_current))})",
                "",
                markdown_table(
                    ["AWS service", "Usage type", "Current", "Share", "Previous", "Delta"],
                    [
                        [
                            item["service"],
                            item["usage_type"],
                            money(item["current"]),
                            report_percent(item["share_percent"]),
                            money(item["previous"]),
                            money(item["delta"], signed=True),
                        ]
                        for item in compute_rows[:10]
                    ],
                ),
                "",
                f"These compute charges {compute_movement} across the adjacent "
                "windows. They are grouped by usage type because the current inventory has no "
                "matching resource for this billing slice. The amount is already included in the "
                "unmatched-resources category above; an owner must reconcile the usage types to "
                "instances before treating the change as a saving.",
            ]
        )
    append_interpretation(lines, interpretations, 3)

    account_category = next(
        item for item in aws["non_vpc_categories"] if item["name"] == "account-level services"
    )
    account_rows = sorted(
        aws["account_level_service_usage"],
        key=lambda item: (-Decimal(item["current"]), item["service"], item["usage_type"]),
    )
    account_detail_rows = [
        [
            item["service"],
            item["usage_type"],
            money(item["current"]),
            report_percent(
                percent_text(Decimal(item["current"]), Decimal(account_category["current"]))
            ),
            money(item["previous"]),
            money(item["delta"], signed=True),
        ]
        for item in account_rows[:20]
    ]
    account_remainder = account_rows[20:]
    if account_remainder:
        current_remainder = sum(Decimal(item["current"]) for item in account_remainder)
        previous_remainder = sum(Decimal(item["previous"]) for item in account_remainder)
        account_detail_rows.append(
            [
                f"Remaining {len(account_remainder)} service and usage pairs",
                "Grouped remainder",
                money(money_text(current_remainder)),
                report_percent(
                    percent_text(current_remainder, Decimal(account_category["current"]))
                ),
                money(money_text(previous_remainder)),
                money(money_text(current_remainder - previous_remainder), signed=True),
            ]
        )
    if not account_detail_rows:
        account_detail_rows = [["None", "No usage", money("0"), "0.00%", money("0"), money("0")]]

    storage = aws["object_storage"]
    bucket_rows = [
        [
            item["bucket"],
            item["region"],
            item["created"] or "Unknown",
            report_bytes(item["bytes"]),
            report_percent(item["share_of_measured_percent"]),
            item["observed_at"][:10],
        ]
        for item in storage["buckets"][:10]
    ]
    remaining_buckets = storage["buckets"][10:]
    if remaining_buckets:
        remaining_bytes = sum(item["bytes"] for item in remaining_buckets)
        bucket_rows.append(
            [
                f"Remaining {len(remaining_buckets)} measured buckets",
                "Multiple regions",
                "Multiple dates",
                report_bytes(remaining_bytes),
                report_percent(
                    percent_text(Decimal(remaining_bytes), Decimal(storage["measured_bytes"]))
                ),
                "Latest available daily metrics",
            ]
        )
    if not bucket_rows:
        bucket_rows = [
            [
                "No bucket size metrics",
                "Unavailable",
                "Unavailable",
                "Unavailable",
                "Unavailable",
                "Unavailable",
            ]
        ]
    family_rows = [
        [
            item["name"],
            item["bucket_count"],
            item["measured_bucket_count"],
            report_bytes(item["bytes"]),
            report_percent(item["share_of_measured_percent"]),
        ]
        for item in storage["families"][:10]
    ] or [["None", 0, 0, report_bytes(0), "0.00%"]]
    lines.extend(
        [
            AWS_REPORT_SECTION_PREFIXES[3],
            "",
            f"This section expands the account-level services row from section 3: "
            f"{money(account_category['current'])} current, {money(account_category['previous'])} "
            f"previous, and {money(account_category['delta'], signed=True)} change. These are the "
            "same costs, not an additional total.",
            "",
            markdown_table(
                ["AWS service", "Usage type", "Current", "Share", "Previous", "Delta"],
                account_detail_rows,
            ),
            "",
            "The largest service and usage pairs are shown individually; the remaining pairs are "
            "combined in the last row. Together they close to the account-level total.",
            "",
            "## Object storage by bucket",
            "",
            f"S3 service costs are {money(storage['current'])} current and "
            f"{money(storage['previous'])} previous, a "
            f"{money(storage['delta'], signed=True)} change. These S3 charges already appear across "
            "the traffic and account-level rows above, not as an additional total. Cost Explorer does not expose a bucket dimension "
            "in this query, so CloudWatch stored bytes show bucket concentration without assigning "
            "the service cost or request cost to a bucket.",
            "",
            markdown_table(
                [
                    "Bucket",
                    "Region",
                    "Created",
                    "Measured stored data",
                    "Share of measured storage",
                    "Oldest component metric (UTC)",
                ],
                bucket_rows,
            ),
            "",
            f"CloudWatch has recent size metrics for "
            f"{count_text(storage['measured_bucket_count'], 'general-purpose bucket')} of "
            f"{count_text(storage['bucket_count'], 'bucket')}, covering "
            f"{report_bytes(storage['measured_bytes'])}; "
            f"{count_text(storage['unmeasured_bucket_count'], 'bucket')} "
            f"{'has' if storage['unmeasured_bucket_count'] == 1 else 'have'} no recent size metric.",
            "",
            "## Object storage families",
            "",
            markdown_table(
                ["Bucket family", "Buckets", "Measured", "Stored data", "Share"],
                family_rows,
            ),
            "",
            "Families use only shared bucket-name prefixes. A family name is a routing signal, not "
            "proof of ownership, retention intent, lifecycle configuration, or workload safety.",
        ]
    )
    lines.extend(
        [
            "",
            f"Version-empty evidence: {storage['empty_count']} observed empty among "
            f"{storage['empty_checked']} checked; {storage['bucket_count'] - storage['empty_checked']} unknown. This excludes unfinished multipart uploads and is not deletion approval. "
            f"{storage['lifecycle_missing']} checked buckets have no enabled lifecycle policy; {storage['bucket_count'] - storage['lifecycle_checked']} policies are unknown.",
            "",
            markdown_table(
                ["Largest bucket", "Lifecycle", "Enabled actions", "Version-empty"],
                [
                    [
                        item["bucket"],
                        item.get("lifecycle_state", "unknown"),
                        ", ".join(item.get("lifecycle_actions", [])) or "None observed",
                        ("Unknown" if item.get("empty") is None else str(item["empty"]).lower()),
                    ]
                    for item in storage["buckets"][:3]
                ],
            ),
        ]
    )
    append_interpretation(lines, interpretations, 4)

    deep = aws["deep_evidence"]
    kms = deep["kms"]
    lines.extend(
        [
            "## KMS key ownership evidence",
            "",
            markdown_table(
                ["KMS evidence", "Result", "Meaning"],
                [
                    [
                        "Read coverage",
                        kms["coverage"].upper(),
                        (
                            "Partial counts are observed lower bounds"
                            if kms["coverage"] == "partial"
                            else "Configured regional reads completed"
                        ),
                    ],
                    [
                        "Customer-managed keys",
                        kms["customer_key_count"],
                        "AWS-managed aliases excluded",
                    ],
                    [
                        "Keys with a custom alias",
                        kms["custom_aliased_key_count"],
                        "Alias is an ownership signal",
                    ],
                    [
                        "Keys without an alias",
                        kms["unaliased_key_count"],
                        "Owner needs confirmation",
                    ],
                    [
                        "Regional sample",
                        kms["sampled_unaliased_count"],
                        "Lowest key IDs per region",
                    ],
                    [
                        "Sample without a description",
                        kms["sampled_without_description_count"],
                        "Description presence only",
                    ],
                    [
                        "Sample already PendingDeletion",
                        kms["sampled_pending_deletion_count"],
                        "Review the active deletion window and dependencies",
                    ],
                    [
                        "Sample Enabled",
                        kms["sampled_enabled_count"],
                        "Owner and dependency proof precede scheduling deletion",
                    ],
                    [
                        "Sample created before 2022",
                        kms["sampled_before_2022_count"],
                        "Age is not deletion safety",
                    ],
                ],
            ),
            "",
            markdown_table(
                ["Sample description category", "Keys"],
                [
                    [description, count]
                    for description, count in sorted(
                        collections.Counter(
                            item["description"] for item in kms["sampled_unaliased_keys"]
                        ).items()
                    )
                ]
                or [["No sample", 0]],
            ),
            "",
            "Full sampled key IDs and observed metadata are in the companion evidence.",
            "",
            "The sample is deterministic and bounded. It proves only the observed alias, state, "
            "creation date, and description category; it does not prove that ciphertext no longer "
            "depends on a key.",
        ]
    )
    append_interpretation(lines, interpretations, 5)

    private_ca = deep["private_ca"]
    lines.extend(
        [
            "## Private CA workload evidence",
            "",
            markdown_table(
                ["Private CA evidence", "Result", "Meaning"],
                [
                    [
                        "Read coverage",
                        private_ca["coverage"].upper(),
                        (
                            "Partial counts are observed lower bounds"
                            if private_ca["coverage"] == "partial"
                            else "Configured regional reads completed"
                        ),
                    ],
                    [
                        "Authorities",
                        private_ca["count"],
                        "Observed across enabled regions",
                    ],
                    [
                        "Active",
                        private_ca["active_count"],
                        "Active does not prove current issuance",
                    ],
                    [
                        "General purpose",
                        private_ca["general_purpose_count"],
                        "Configured usage mode",
                    ],
                    [
                        "Short-lived certificate",
                        private_ca["short_lived_count"],
                        "Configured usage mode",
                    ],
                    [
                        "Recent IssueCertificate events",
                        private_ca["recent_issue_events"],
                        "Bounded CloudTrail lookup",
                    ],
                    [
                        "Observed roles or identities",
                        len(private_ca["recent_issue_callers"]),
                        "Caller identity is a workload signal",
                    ],
                ],
            ),
            "",
            "Authority IDs and observed role identities are in the companion evidence.",
            "",
            "The CloudTrail query covers the recent bounded lookback only. Observed issuance is "
            "positive workload evidence; no observed event does not prove that an authority is unused.",
        ]
    )
    append_interpretation(lines, interpretations, 6)

    evidence_coverage = deep["coverage"]
    lines.extend(
        [
            "## Named account-level inventory",
            "",
            markdown_table(
                [
                    "Service inventory",
                    "Resources",
                    "Read coverage",
                    "Capacity or seats",
                    "Configuration / state",
                    "Examples",
                ],
                [
                    [
                        item["kind"],
                        (
                            f"{item['count']} observed"
                            if item["coverage"] == "partial"
                            else item["count"]
                        ),
                        item["coverage"].upper(),
                        item["capacity"],
                        item["configuration"],
                        ", ".join(item["examples"]) or "None",
                    ]
                    for item in deep["account_services"]
                ],
            ),
            "",
            markdown_table(
                ["Named-service and image read coverage", "Read gaps", "Boundary"],
                [
                    [
                        evidence_coverage["status"].upper(),
                        "; ".join(evidence_coverage["gaps"][:20]) or "None",
                        "A read gap is unknown, not an observed empty service",
                    ]
                ],
            ),
            "",
            "Named inventory makes shared services routable, but it does not allocate their billing "
            "rows to a workload or establish that a subscription, repository, domain, cluster, or "
            "network is unused.",
        ]
    )
    repositories = deep["repositories"]
    measured_repositories = [item for item in repositories if "image_count" in item]
    image_coverage = deep["image_coverage"]
    ecr_cost = next(
        (
            item["current"]
            for item in aws["service_totals"]
            if item["service"] == "Amazon EC2 Container Registry (ECR)"
        ),
        "0.00",
    )
    lines.extend(
        [
            "",
            "## Container registry storage",
            "",
            f"ECR service cost is {money(ecr_cost)} across all its usage types, already included above. "
            f"Image metadata coverage is {image_coverage['status']}: {image_coverage['checked']} of {image_coverage['observed_repositories']} observed repositories checked. "
            "Ranking covers only the measured subset. "
            "The size sums below describe image manifests, not deduplicated billable storage; shared layers can be counted more than once. "
            "Missing pull timestamps do not prove an image is unused. Lifecycle presence alone does not prove adequate retention.",
            "",
            markdown_table(
                [
                    "Repository / region",
                    "Images",
                    "Sum of image sizes",
                    "Untagged",
                    "Oldest push",
                    "Latest observed pull",
                    "Lifecycle",
                ],
                [
                    [
                        f"{item.get('name') or item['id']} / {item['region']}",
                        item["image_count"],
                        report_bytes(item["image_bytes"]),
                        item["untagged_images"],
                        item.get("oldest_image") or "Unknown",
                        item.get("latest_pull") or "Unknown",
                        item.get("lifecycle_state", "not checked"),
                    ]
                    for item in repositories
                    if "image_count" in item
                ][:5]
                or [
                    [
                        "Not measured",
                        "Unknown",
                        "Unknown",
                        "Unknown",
                        "Unknown",
                        "Unknown",
                        "Unknown",
                    ]
                ],
            ),
        ]
    )
    append_interpretation(lines, interpretations, 7)

    ranked_candidates = sorted(
        aws["reclaim_candidates"],
        key=lambda item: (-Decimal(item["estimated_monthly_cost"]), item["id"]),
    )
    shown_candidates = [
        item
        for kind in ("ebs", "loadbalancer", "public-ipv4")
        for item in [row for row in ranked_candidates if row["id"].split("/", 2)[1] == kind][:3]
    ]
    candidate_detail_rows = [
        [
            item["name"],
            item["id"].split("/", 1)[0],
            "; ".join(item["signals"]),
            item["owner"],
            (
                money(item["estimated_monthly_cost"])
                if item["cost_estimate_status"] == "priced"
                else "Not matched to billing"
            ),
            item["next_step"],
        ]
        for item in shown_candidates
    ] or [["None", "Unknown", "No candidate signal", "Unresolved", "Unknown", "No action"]]
    lines.extend(
        [
            AWS_REPORT_SECTION_PREFIXES[4],
            "",
            markdown_table(
                [
                    "Candidate group",
                    "Inventory signal",
                    "Candidate count",
                    "Monthly estimate",
                    "Evidence gate",
                ],
                [
                    [
                        "EBS volumes",
                        f"{reclaim['ebs']['count']} unattached; {reclaim['ebs']['size_gib']} GiB",
                        reclaim["ebs"]["count"],
                        money(reclaim["ebs"]["estimated_monthly_cost"]),
                        "Confirm attachment and workload use before any change",
                    ],
                    [
                        "Load balancers",
                        f"{reclaim['load_balancer']['zero_healthy_count']} zero healthy of "
                        f"{reclaim['load_balancer']['total_count']}; "
                        f"{reclaim['load_balancer']['zero_registered_count']} zero registered; "
                        f"{reclaim['load_balancer']['registered_but_unhealthy_count']} registered "
                        "but unhealthy",
                        reclaim["load_balancer"]["zero_healthy_count"],
                        money(reclaim["load_balancer"]["estimated_monthly_cost"]),
                        "Confirm traffic, DNS, and target health before any change",
                    ],
                    [
                        "Public IPv4 addresses",
                        f"{reclaim['public_ipv4']['unassociated_count']} unassociated of "
                        f"{reclaim['public_ipv4']['total_count']}",
                        reclaim["public_ipv4"]["unassociated_count"],
                        money(reclaim["public_ipv4"]["estimated_monthly_cost"]),
                        "Confirm pending use before release",
                    ],
                    [
                        "Candidate pool",
                        f"{reclaim['candidate_count']} inventory candidates; {reclaim['confirmed_count']} confirmed",
                        reclaim["candidate_count"],
                        money(reclaim["estimated_monthly_cost"]),
                        "Inventory signals are not reclaim approval",
                    ],
                ],
            ),
            "",
            f"The current inventory has {count_text(reclaim['candidate_count'], 'reclaim candidate')} with a "
            f"priced monthly subtotal of {money(reclaim['estimated_monthly_cost'])}; "
            f"{reclaim['confirmed_count']} "
            f"{'is' if reclaim['confirmed_count'] == 1 else 'are'} confirmed reclaimable.",
            "",
            "## Resource-level validation queue",
            "",
            markdown_table(
                [
                    "Resource",
                    "Region",
                    "Signals",
                    "Owner",
                    "Monthly estimate",
                    "Decision gate",
                ],
                candidate_detail_rows,
            ),
            "",
            f"The table shows up to three per resource type: {count_text(len(shown_candidates), 'candidate')} of "
            f"{count_text(len(ranked_candidates), 'candidate')}. The companion evidence contains the complete queue. A candidate "
            "remains a validation queue item, not "
            "a cleanup instruction or guaranteed saving.",
        ]
    )
    unpriced = sum(item["cost_estimate_status"] != "priced" for item in ranked_candidates)
    lines.extend(
        [
            "",
            f"{unpriced} candidates have no matched charge in the billing window; the projected subtotal excludes them. "
            "Unresolved owners have no usable resource or VPC owner tag. Source names are routing hints, not proof of ownership.",
        ]
    )
    append_interpretation(lines, interpretations, 8)

    recurrence_sentence = (
        "Recurrence was not evaluated because no earlier inventory snapshot was available."
        if recurring is None
        else f"Recurrence was evaluated against inventory collected at {resources['baseline']}; "
        f"{recurring} of {reclaim['candidate_count']} current candidates were present there."
    )
    signal_rows = [
        [
            "Total AWS cost",
            money(cost["current_total"]),
            money(cost["previous_total"]),
            money(cost["delta"], signed=True),
        ],
        *[
            [item["name"], money(item["current"]), money(item["previous"]), money(item["delta"], signed=True)]
            for item in aws["non_vpc_categories"]
        ],
    ]
    lines.extend(
        [
            AWS_REPORT_SECTION_PREFIXES[5],
            "",
            markdown_table(
                ["Recurrence evidence", "Current candidates", "Recurring", "Baseline"],
                [
                    [
                        "Inventory snapshot comparison",
                        reclaim["candidate_count"],
                        recurring if recurring is not None else "Not evaluated",
                        resources["baseline"] if recurring is not None else "No earlier snapshot",
                    ]
                ],
            ),
            "",
            markdown_table(["Cost signal", "Current", "Previous", "Delta"], signal_rows),
            "",
            recurrence_sentence
            + " Cost deltas and candidate persistence are investigation signals, not defect or causal proof.",
        ]
    )
    observations = aws["candidate_history"]["observations"]
    first_seen = {}
    for observation in observations:
        for key in observation["candidates"]:
            first_seen.setdefault(key, observation["collected_at"])
    elapsed = (
        (
            datetime.fromisoformat(resources["collected_at"])
            - datetime.fromisoformat(resources["baseline"])
        ).total_seconds()
        / 3600
        if recurring is not None
        else None
    )
    lines.extend(
        [
            "",
            (
                f"The immediate inventory baseline is {elapsed:.2f} hours earlier."
                if elapsed is not None
                else "No immediate inventory baseline is available."
            ),
            f"The retained candidate history contains {count_text(len(observations), 'daily observation')} from {observations[0]['collected_at']} "
            f"to {observations[-1]['collected_at']}. Observations show presence on those dates, not continuous use or a creating defect.",
            "",
            markdown_table(
                ["Observation (UTC)", "Candidate count"],
                [[item["collected_at"], len(item["candidates"])] for item in observations],
            ),
            "",
            (
                f"Inventory changes across comparable read surfaces: {len(resources['new'])} newly observed and {len(resources['removed'])} no longer observed. "
                "These are snapshot differences, not confirmed creation or deletion events."
                if recurring is not None
                else "Inventory changes were not evaluated without an earlier snapshot."
            ),
            "",
            markdown_table(
                [
                    "Recently created candidate",
                    "Created",
                    "Source signal",
                    "First retained candidate observation",
                ],
                [
                    [
                        item["name"],
                        item["created"],
                        item["source"],
                        first_seen[item["id"]],
                    ]
                    for item in sorted(
                        ranked_candidates,
                        key=lambda item: item["created"],
                        reverse=True,
                    )
                    if item["created"] != "unknown"
                ][:5],
            ),
        ]
    )
    append_interpretation(lines, interpretations, 9)

    owner = "Unresolved"
    candidate_actions = [
        [
            "Validate EBS candidates",
            f"{reclaim['ebs']['count']} volumes; {reclaim['ebs']['size_gib']} GiB",
            money(reclaim["ebs"]["estimated_monthly_cost"]),
            owner,
            "Confirm attachment and workload use",
        ],
        [
            "Validate load balancer candidates",
            f"{reclaim['load_balancer']['zero_healthy_count']} zero healthy of "
            f"{reclaim['load_balancer']['total_count']}",
            money(reclaim["load_balancer"]["estimated_monthly_cost"]),
            owner,
            "Confirm traffic, DNS, and target health",
        ],
        [
            "Validate public IPv4 candidates",
            f"{reclaim['public_ipv4']['unassociated_count']} unassociated of "
            f"{reclaim['public_ipv4']['total_count']}",
            money(reclaim["public_ipv4"]["estimated_monthly_cost"]),
            owner,
            "Confirm pending use before release",
        ],
    ]
    workload_actions = []
    if eks_rows:
        largest_eks = max(eks_rows, key=lambda item: Decimal(item["current"]))
        largest_change = max(eks_rows, key=lambda item: abs(Decimal(item["delta"])))
        workload_actions.extend(
            [
                [
                    "Review largest EKS-linked cluster",
                    f"{largest_eks['eks_cluster']}; {money(largest_eks['current'])}; "
                    f"{money(largest_eks['delta'], signed=True)}",
                    owner,
                    "Inspect the largest allocated service components in section 2 before changing capacity",
                ],
                [
                    "Review largest EKS-linked movement",
                    f"{largest_change['eks_cluster']}; {money(largest_change['current'])}; "
                    f"{money(largest_change['delta'], signed=True)}",
                    owner,
                    "Confirm the workload movement is intentional",
                ],
            ]
        )
    if compute_rows:
        workload_actions.append(
            [
                "Map unmatched cost",
                f"Compute without matching inventory; {money(money_text(compute_current))}; "
                f"{money(money_text(compute_delta), signed=True)}",
                owner,
                "Reconcile usage to inventory before right-sizing",
            ]
        )
    if vpc_rows:
        for item in vpc_rows[:3]:
            workload_actions.append(
                [
                    "Confirm non-EKS VPC purpose",
                    f"{item['vpc_id']} / {item['name']}; {money(item['current'])}; "
                    f"{item['purpose_signal']}",
                    item["owner"] or "Unresolved",
                    "Confirm the named VPC purpose and whether its listed resources still serve that purpose",
                ]
            )
    for item in storage["buckets"][:2]:
        workload_actions.append(
            [
                "Review S3 retention",
                f"{item['bucket']}; {report_bytes(item['bytes'])}; policy {item.get('lifecycle_state', 'unknown')}",
                "Unresolved; bucket owner",
                "Confirm retention need; inspect lifecycle filters and storage-class transitions",
            ]
        )
    if image_coverage["status"] != "complete":
        workload_actions.append(
            [
                "Restore ECR image evidence",
                f"{image_coverage['checked']} of {image_coverage['observed_repositories']} observed repositories checked",
                "Unresolved; registry reader",
                "Complete repository and image reads before ranking retention across the account",
            ]
        )
    for item in measured_repositories[:2]:
        workload_actions.append(
            [
                "Review ECR image retention",
                item.get("name") or item["id"],
                "Unresolved; image publisher",
                "Check deployed digests and retention policy before expiring any images",
            ]
        )
    for item in deep["account_services"]:
        if item["kind"] in {"opensearch-domain", "elasticache-cluster"} and item["count"]:
            workload_actions.append(
                [
                    "Review shared-service utilization",
                    ", ".join(item["examples"]),
                    "Unresolved; service owner",
                    (
                        "Check log retention and shard utilization"
                        if item["kind"] == "opensearch-domain"
                        else "Check memory and connection utilization before downsizing"
                    ),
                ]
            )
    defect_rows = [
        [
            "Confirmed reclaim evidence",
            f"{reclaim['confirmed_count']} confirmed of {reclaim['candidate_count']} candidates",
            owner,
            "Keep candidate validation ahead of cleanup",
        ],
        [
            "Inventory baseline",
            "No earlier snapshot" if recurring is None else resources["baseline"],
            owner,
            "Do not infer creation or deletion without a comparable snapshot",
        ],
        [
            "Coverage boundary",
            (
                "Named-service and image reads completed"
                if evidence_coverage["status"] == "complete"
                else f"{len(evidence_coverage['gaps'])} deep evidence read gaps"
            ),
            owner,
            "Do not infer complete service-level attribution",
        ],
        [
            "Unaliased KMS ownership",
            f"{kms['unaliased_key_count']} keys; {kms['sampled_unaliased_count']} sampled"
            + ("; observed lower bounds" if kms["coverage"] == "partial" else ""),
            owner,
            "For enabled keys, establish owner and ciphertext dependencies before scheduling deletion; resolve unknown states first",
        ],
        [
            "Private CA workload",
            f"{private_ca['active_count']} active; {private_ca['recent_issue_events']} recent issuance events"
            + ("; observed lower bounds" if private_ca["coverage"] == "partial" else ""),
            owner,
            "Review observed callers and rotation frequency",
        ],
    ]
    if kms["sampled_pending_deletion_count"]:
        defect_rows.append(
            [
                "Review pending KMS deletions",
                f"{kms['sampled_pending_deletion_count']} sampled keys are already PendingDeletion",
                owner,
                "Verify deletion windows and dependencies; the owner must cancel unsafe scheduled deletion within its window",
            ]
        )
    partial_services = [
        item for item in deep["account_services"] if item["coverage"] == "partial"
    ]
    if partial_services:
        defect_rows.append(
            [
                "Partial named-service inventory",
                "; ".join(
                    f"{item['kind']} {item['count']} observed" for item in partial_services
                )
                + "; all counts are observed lower bounds",
                owner,
                "Restore each named-service read before treating absence as empty",
            ]
        )
    lines.extend(
        [
            AWS_REPORT_SECTION_PREFIXES[6],
            "",
            AWS_REPORT_SUBSECTION_PREFIXES[0],
            "",
            markdown_table(
                ["Decision", "Evidence", "Monthly estimate", "Who confirms", "Next step"],
                candidate_actions,
            ),
            "",
            AWS_REPORT_SUBSECTION_PREFIXES[1],
            "",
            markdown_table(
                ["Decision", "Evidence", "Who confirms", "Next step"],
                workload_actions
                or [
                    [
                        "No EKS-linked action",
                        "No EKS-linked ownership row",
                        owner,
                        "Collect utilization evidence",
                    ]
                ],
            ),
            "",
            AWS_REPORT_SUBSECTION_PREFIXES[2],
            "",
            markdown_table(
                ["Decision", "Evidence", "Who confirms", "Next step"], defect_rows
            ),
            "",
            "Every action above is a review gate; this report authorizes no cloud change.",
        ]
    )
    append_interpretation(lines, interpretations, 10)
    lines.extend([AWS_REPORT_GENERATED_MARKER, ""])
    return "\n".join(lines)


def render_report(analysis: dict[str, Any], interpretation_input: dict[str, Any]) -> str:
    validate_decision_units(analysis.get("decision_units"))
    choices = validate_interpretations(interpretation_input)
    interpretations = {
        unit["key"]: unit["decision"]
        + (" " + unit["boundary"] if choices[unit["key"]] == "context" else "")
        for unit in analysis["decision_units"]
    }
    if analysis.get("synthetic") is True:
        return render_synthetic_report(analysis, interpretations)
    if analysis.get("synthetic") is not False or set(analysis.get("providers", {})) != {"aws"}:
        raise ContractError("deterministic live report rendering currently supports one AWS scope")
    contract = analysis.get("report_contract")
    if (
        not isinstance(contract, dict)
        or contract.get("section_prefixes") != list(AWS_REPORT_SECTION_PREFIXES)
        or contract.get("subsection_prefixes") != list(AWS_REPORT_SUBSECTION_PREFIXES)
        or contract.get("interpretation_keys") != list(REPORT_INTERPRETATION_KEYS)
    ):
        raise ContractError("analysis is missing the deterministic AWS report contract")
    return render_aws_report(analysis, interpretations)


def validate_report(
    analysis: dict[str, Any], interpretation_input: dict[str, Any], report: str
) -> None:
    if report != render_report(analysis, interpretation_input):
        raise ContractError("report differs from deterministic Python materialization")
    if "`" in report:
        raise ContractError("deterministic report must not render inline code")
    if "—" in report or "–" in report:
        raise ContractError("deterministic report must use prose punctuation, not dashes")
    leaked = [
        term for term in AWS_REPORT_RAW_SCHEMA_TERMS if term.casefold() in report.casefold()
    ]
    if leaked:
        raise ContractError(f"deterministic report exposes raw schema terms: {', '.join(leaked)}")
    if report.count(AWS_REPORT_GENERATED_MARKER) != 1:
        raise ContractError("deterministic report must contain exactly one generated marker")


def write_json(path: Path, value: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n")


def render_evidence(analysis: dict[str, Any]) -> str:
    aws = analysis["providers"].get("aws", {})
    if analysis.get("synthetic") or aws.get("coverage", {}).get("status") != "complete":
        return "# Report evidence\n\nNo complete live AWS evidence is available.\n"
    lines = [
        "# Report evidence",
        "",
        "Resource identifiers and observed states support review only; no cleanup is authorized.",
        "",
        "## Complete candidate queue",
        "",
        markdown_table(
            [
                "Resource ID",
                "Name",
                "Created",
                "Owner signal",
                "Source signal",
                "Signals",
                "Monthly projection",
                "Next step",
            ],
            [
                [
                    item["id"],
                    item["name"],
                    item["created"],
                    item["owner"],
                    item["source"],
                    "; ".join(item["signals"]),
                    (
                        item["estimated_monthly_cost"]
                        if item["cost_estimate_status"] == "priced"
                        else "Unpriced"
                    ),
                    item["next_step"],
                ]
                for item in aws["reclaim_candidates"]
            ],
        ),
        "",
        "## Non-EKS VPCs",
        "",
        markdown_table(
            ["VPC", "Name", "Region", "Cost", "Owner", "Purpose signal", "Inventory"],
            [
                [
                    item["vpc_id"],
                    item["name"],
                    item["region"],
                    item["current"],
                    item["owner"] or "Unresolved",
                    item["purpose_signal"],
                    item["inventory"],
                ]
                for item in aws["vpc_evidence"]["rows"]
            ],
        ),
        "",
        "## Bucket policies and version checks",
        "",
        markdown_table(
            [
                "Bucket",
                "Region",
                "Bytes",
                "Version-empty",
                "Lifecycle",
                "Actions",
                "Read errors",
            ],
            [
                [
                    item["bucket"],
                    item["region"],
                    item["bytes"] if item["bytes"] is not None else "Unknown",
                    "Unknown" if item.get("empty") is None else item["empty"],
                    item.get("lifecycle_state", "unknown"),
                    ", ".join(item.get("lifecycle_actions", [])),
                    "; ".join(item.get("audit_errors", [])),
                ]
                for item in aws["object_storage"]["all_buckets"]
            ],
        ),
        "",
        "## Repository image metadata",
        "",
        "Image-size sums are not deduplicated billable bytes; missing pull times do not establish disuse.",
        "",
        markdown_table(
            [
                "Repository",
                "Region",
                "Images",
                "Image-size sum",
                "Untagged",
                "Oldest push",
                "Latest pull",
                "Lifecycle",
                "Read error",
            ],
            [
                [
                    item["id"],
                    item["region"],
                    item.get("image_count", "Unknown"),
                    item.get("image_bytes", "Unknown"),
                    item.get("untagged_images", "Unknown"),
                    item.get("oldest_image") or "Unknown",
                    item.get("latest_pull") or "Unknown",
                    item.get("lifecycle_state", "not checked"),
                    item.get("image_read_error", ""),
                ]
                for item in aws["deep_evidence"]["repositories"]
            ],
        ),
        "",
        "## KMS sampled keys",
        "",
        "The bounded sample is not statistically representative; key age and description do not prove ciphertext independence.",
        "",
        markdown_table(
            ["Region", "Key ID", "Created", "State", "Description category"],
            [
                [item["region"], item["id"], item["created"], item["state"], item["description"]]
                for item in aws["deep_evidence"]["kms"]["sampled_unaliased_keys"]
            ],
        ),
        "",
        "## Private CA authorities and callers",
        "",
        "CloudTrail observations cover the bounded lookback; absence of an event does not establish disuse.",
        "",
        markdown_table(
            [
                "Region",
                "Authority ID",
                "State",
                "Type",
                "Usage mode",
                "Created",
                "Owner signal",
                "IssueCertificate events",
                "Observed role or identity",
            ],
            [
                [
                    item["region"],
                    item["id"],
                    item["state"],
                    item["type"],
                    item["usage_mode"],
                    item["created"],
                    item["owner"] or item["session"] or "Unresolved",
                    item["recent_issue_events"],
                    "; ".join(item["recent_issue_callers"]) or "None observed",
                ]
                for item in aws["deep_evidence"]["private_ca"]["authorities"]
            ],
        ),
        "",
        "## Comparable inventory differences",
        "",
        "Observed differences are not confirmed creation or deletion events. Only surfaces covered by both snapshots are compared.",
        (
            "No earlier snapshot; differences were not evaluated."
            if aws["resources"]["baseline"] == "no_inventory_baseline"
            else f"Baseline: {aws['resources']['baseline']}; current: {aws['resources']['collected_at']}."
        ),
        "",
        markdown_table(
            ["Observed difference", "Resource ID"],
            [
                [label, resource_id]
                for field, label in (("new", "Newly observed"), ("removed", "No longer observed"))
                for resource_id in aws["resources"][field]
            ],
        ),
        "",
        "## Candidate observations",
        "",
        markdown_table(
            ["Observed at", "Resource ID"],
            [
                [observation["collected_at"], resource_id]
                for observation in aws["candidate_history"]["observations"]
                for resource_id in observation["candidates"]
            ],
        ),
        "",
    ]
    return "\n".join(lines)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    subparsers = parser.add_subparsers(dest="command", required=True)

    validate = subparsers.add_parser("validate-scopes", help="validate the versioned scope allowlist")
    validate.add_argument("--scopes", type=Path, required=True)

    replay = subparsers.add_parser("replay", help="normalize and diff a synthetic two-window scenario")
    replay.add_argument("--scopes", type=Path, required=True)
    replay.add_argument("--fixture", type=Path, required=True)
    replay.add_argument("--scenario", choices=("complete", "incomplete"), required=True)
    replay.add_argument("--output", type=Path, required=True)

    normalize_aws = subparsers.add_parser(
        "normalize-aws", help="normalize one live AWS snapshot without cloud access"
    )
    normalize_aws.add_argument("--scopes", type=Path, required=True)
    normalize_aws.add_argument("--snapshot", type=Path, required=True)
    normalize_aws.add_argument("--previous-snapshot", type=Path)
    normalize_aws.add_argument("--previous-history", type=Path)
    normalize_aws.add_argument("--history-output", type=Path)
    normalize_aws.add_argument("--evidence-url")
    normalize_aws.add_argument("--output", type=Path, required=True)

    check = subparsers.add_parser("check", help="run the retained deterministic replay safety check")
    check.add_argument("--scopes", type=Path, required=True)
    check.add_argument("--fixture", type=Path, required=True)

    validate_report_command = subparsers.add_parser(
        "validate-report", help="verify exact deterministic report materialization"
    )
    validate_report_command.add_argument("--analysis", type=Path, required=True)
    validate_report_command.add_argument("--interpretations", type=Path, required=True)
    validate_report_command.add_argument("--report", type=Path, required=True)

    render_report_command = subparsers.add_parser(
        "render-report", help="materialize the deterministic report with Agent interpretations"
    )
    render_report_command.add_argument("--analysis", type=Path, required=True)
    render_report_command.add_argument("--interpretations", type=Path, required=True)
    render_report_command.add_argument("--output", type=Path, required=True)
    render_report_command.add_argument("--evidence-output", type=Path)
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    try:
        if args.command == "validate-scopes":
            validate_manifest(load_json(args.scopes))
        elif args.command == "replay":
            analysis = build_analysis(load_json(args.scopes), load_json(args.fixture), args.scenario)
            write_json(args.output, analysis)
        elif args.command == "normalize-aws":
            analysis = build_aws_analysis(
                load_json(args.scopes),
                load_json(args.snapshot),
                load_json(args.previous_snapshot) if args.previous_snapshot else None,
                load_json(args.previous_history) if args.previous_history else None,
            )
            if args.evidence_url:
                if not re.fullmatch(
                    r"https://github\.com/[\w.-]+/[\w.-]+/actions/runs/[0-9]+(?:#artifacts)?",
                    args.evidence_url,
                ):
                    raise ContractError("evidence URL must point to a GitHub Actions run")
                analysis["evidence_url"] = args.evidence_url
            write_json(args.output, analysis)
            if (
                args.history_output
                and analysis["providers"]["aws"]["coverage"]["status"] == "complete"
            ):
                write_json(
                    args.history_output,
                    analysis["providers"]["aws"]["candidate_history"],
                )
        elif args.command == "check":
            check_contract(load_json(args.scopes), load_json(args.fixture))
            print("cloud cost report replay check: ok")
        elif args.command == "validate-report":
            try:
                report = args.report.read_text()
            except OSError as exc:
                raise ContractError(f"cannot read {args.report}: {exc}") from exc
            validate_report(
                load_json(args.analysis), load_json(args.interpretations), report
            )
        elif args.command == "render-report":
            try:
                analysis = load_json(args.analysis)
                args.output.write_text(render_report(analysis, load_json(args.interpretations)))
                if args.evidence_output:
                    args.evidence_output.write_text(render_evidence(analysis))
            except OSError as exc:
                raise ContractError(f"cannot render report: {exc}") from exc
        else:
            raise ContractError(f"unsupported command {args.command!r}")
    except ContractError as exc:
        print(f"cloud-cost-report: {exc}", file=sys.stderr)
        return 1
    return 0


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