"""Privacy-safe deterministic fact bundle for one cloud subscription."""

from __future__ import annotations

import calendar
import uuid
from collections import Counter
from dataclasses import dataclass
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal
from typing import Any

from sqlalchemy import func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.exceptions import NotFoundError
from app.domains.cmdb.cost_service import detect_anomalies
from app.domains.cmdb.cost_view import (
    COST_MEASURE_LIST,
    engineering_row_clause,
    fact_engineering_amount,
)
from app.domains.cmdb.models import (
    EFFICIENCY_VERDICT_BUSY,
    EFFICIENCY_VERDICT_HEALTHY,
    EFFICIENCY_VERDICT_IDLE,
    EFFICIENCY_VERDICT_OVERSIZED,
    GRANT_STATUS_ACTIVE,
    GRANT_STATUS_PENDING,
    SCOPE_STATUS_ACTIVE,
    SCOPE_TYPE_AZURE_RESOURCE_GROUP,
    CmdbCloudAccessGrant,
    CmdbCloudResource,
    CmdbCloudRole,
    CmdbCloudScope,
    CmdbCostFact,
    CmdbResourceGovernance,
    CmdbServerEfficiency,
)
from app.domains.cmdb.resource_directory_coverage import resolve_resource_coverage
from app.domains.cmdb.resource_directory_ownership import resolve_effective_resource_owners
from app.domains.cmdb.scope_billing_detail import scope_fact_filters
from app.domains.cmdb.service import (
    resolve_inherited_server_owners,
    resolve_scope_servers,
)
from app.identity.person_models import IdentityPerson
from app.integrations.cloud_role_governance import (
    TIER_ELEVATED,
    TIER_RESTRICTED,
    TIER_UNGRANTABLE,
    most_severe,
)
from app.intelligence.cloud_scope_scoring import score_cloud_scope
from app.intelligence.privacy import EvidenceObject, assert_llm_safe, build_reference_map
from app.intelligence.scoring import MetricInput, MetricKind, json_value

_CLASSIFIED = {
    EFFICIENCY_VERDICT_IDLE,
    EFFICIENCY_VERDICT_OVERSIZED,
    EFFICIENCY_VERDICT_HEALTHY,
    EFFICIENCY_VERDICT_BUSY,
}
_ACTIONABLE = {EFFICIENCY_VERDICT_IDLE, EFFICIENCY_VERDICT_OVERSIZED}
_PRIVILEGED_TIERS = {TIER_ELEVATED, TIER_RESTRICTED, TIER_UNGRANTABLE}
_DIRECTORY_FACT_BATCH_SIZE = 500
_DIRECTORY_EVIDENCE_LIMIT = 20


@dataclass(frozen=True)
class CloudScopeFactBundle:
    health: dict[str, Any]
    aggregate_facts: dict[str, Any]
    reference_map: dict[str, Any]
    data_as_of: datetime


def _utc(value: datetime) -> datetime:
    return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)


def _pct(numerator: Decimal | int, denominator: Decimal | int) -> Decimal:
    if Decimal(denominator) <= 0:
        return Decimal("0")
    return (Decimal(numerator) * 100 / Decimal(denominator)).quantize(Decimal("0.01"))


def _missed_days(value: datetime | date | None, now: datetime) -> int:
    if value is None:
        return 3
    day = value.date() if isinstance(value, datetime) else value
    return max(0, (now.date() - day).days)


def _month_start_months_ago(anchor: date, months: int) -> date:
    absolute_month = anchor.year * 12 + anchor.month - 1 - months
    year, zero_based_month = divmod(absolute_month, 12)
    return date(year, zero_based_month + 1, 1)


def _reference_keys(
    reverse: dict[tuple[str, str], str],
    kind: str,
    object_ids: list[Any],
) -> tuple[str, ...]:
    return tuple(
        reverse[(kind, str(object_id))]
        for object_id in object_ids
        if (kind, str(object_id)) in reverse
    )


class ScopeAnomalyIndex:
    """Per-run memo of ``detect_anomalies``, keyed by (integration, day).

    ``detect_anomalies`` is scoped to an **integration**, not to one
    subscription: its ``resource_keys`` argument only filters rows in
    Python *after* the scan (``cost_service`` line ~3812), and the scan
    itself is the widest GROUP BY in the cost domain (integration ×
    resource × date over 30 days). Calling it once per subscription
    therefore re-runs an identical query once per sibling scope —
    aws-primary has 96 active scopes, so its 30-day cost facts were
    scanned 96 times per fleet refresh.

    Measured against prod 2026-08-18: ~51s per scope average, i.e.
    ~3.9 hours for the 271 active scopes. That is why
    ``intelligence.cloud_scope_baseline_daily`` had never once completed
    since it was seeded, and why the bootstrap row that shares this code
    path outlived the worker watchdog's 300s grace and crash-looped prod
    for ~16h (see the 2026-08-18 CHANGELOG entry).

    Memoising the **output** is cheap: it holds only rows that already
    cleared the spike threshold, not the per-resource series that the
    scan builds and discards. One index instance lives for one fleet
    refresh; nothing is cached across runs, so a stale index can never
    outlive the ``run_at`` anchor it was keyed on.
    """

    __slots__ = ("_by_integration",)

    def __init__(self) -> None:
        self._by_integration: dict[tuple[str, date], list[dict[str, Any]]] = {}

    async def for_integration(
        self,
        db: AsyncSession,
        integration_id: str,
        *,
        as_of_date: date,
    ) -> list[dict[str, Any]]:
        """Unfiltered anomalies for one integration; callers filter per scope.

        Deliberately called WITHOUT ``resource_keys`` so a single result
        serves every sibling scope. Equivalent to the per-scope call
        because the key filter is post-hoc on
        ``(integration_id, resource_id)``, ``limit`` is ``None`` here (no
        truncation to interact with), the sort is stable under filtering,
        and each resource's median/IQR is computed from its own series
        only — no cross-resource coupling.
        """
        key = (integration_id, as_of_date)
        cached = self._by_integration.get(key)
        if cached is None:
            cached = await detect_anomalies(
                db,
                days=30,
                integration_ids=[integration_id],
                limit=None,
                measure=COST_MEASURE_LIST,
                as_of_date=as_of_date,
            )
            self._by_integration[key] = cached
        return cached


async def collect_cloud_scope_facts(
    db: AsyncSession,
    scope_id: str,
    *,
    now: datetime | None = None,
    anomaly_index: ScopeAnomalyIndex | None = None,
) -> CloudScopeFactBundle:
    """Collect shared/list-cost facts only; never reads invoice-only columns.

    ``anomaly_index`` is an optional per-run memo used by the fleet-wide
    refresh. Single-scope callers (the subscription analysis route) leave
    it ``None`` and keep the narrower per-scope query, which filters
    before building the in-process series and so touches less memory.
    """
    anchor = _utc(now or datetime.now(UTC))
    scope = (
        await db.execute(
            select(CmdbCloudScope).where(
                CmdbCloudScope.id == scope_id,
                CmdbCloudScope.deleted_at.is_(None),
            )
        )
    ).scalar_one_or_none()
    if scope is None:
        raise NotFoundError("Cloud subscription", scope_id)
    cost_filters, cost_routing = scope_fact_filters(scope)
    if not cost_filters:
        # An RG scope without an external identifier has no safe attribution
        # key. A false empty result would look like zero spend, so fail closed.
        cost_filters = [CmdbCostFact.id.is_(None)]

    grant_rows = list(
        (
            await db.execute(
                select(CmdbCloudAccessGrant).where(
                    CmdbCloudAccessGrant.scope_id == str(scope.id),
                    CmdbCloudAccessGrant.status.in_(
                        (GRANT_STATUS_ACTIVE, GRANT_STATUS_PENDING),
                    ),
                )
            )
        ).scalars()
    )
    grants = [row for row in grant_rows if row.status == GRANT_STATUS_ACTIVE]
    pending_grants = [row for row in grant_rows if row.status == GRANT_STATUS_PENDING]
    person_ids = {str(row.person_id) for row in grants if row.person_id}
    if scope.owner_person_id:
        person_ids.add(str(scope.owner_person_id))
    active_people: set[str] = (
        {
            str(value)
            for value in (
                await db.execute(
                    select(IdentityPerson.id).where(
                        IdentityPerson.id.in_(person_ids),
                        IdentityPerson.deleted_at.is_(None),
                        IdentityPerson.employment_status == "active",
                    )
                )
            ).scalars()
        }
        if person_ids
        else set()
    )

    roles = list(
        (
            await db.execute(
                select(CmdbCloudRole).where(
                    CmdbCloudRole.integration_id == str(scope.integration_id),
                    or_(
                        CmdbCloudRole.scope_id.is_(None),
                        CmdbCloudRole.scope_id == str(scope.id),
                    ),
                    CmdbCloudRole.deleted_at.is_(None),
                )
            )
        ).scalars()
    )
    role_tiers: dict[str, str] = {}
    for role in roles:
        current = role_tiers.get(role.role_key)
        role_tiers[role.role_key] = (
            most_severe(current, role.governance_tier or "normal")
            if current
            else role.governance_tier or "normal"
        )

    identity_gaps = [
        row for row in grants if row.person_id is None or str(row.person_id) not in active_people
    ]
    unmapped_roles = [row for row in grants if row.role_key not in role_tiers]
    permanent = [row for row in grants if row.expires_at is None]
    privileged = [
        row for row in grants if role_tiers.get(row.role_key, "normal") in _PRIVILEGED_TIERS
    ]
    privileged_pending_review = [row for row in privileged if row.review_status == "pending_review"]
    governed = [
        row
        for row in grants
        if row not in identity_gaps
        and row not in unmapped_roles
        and (row.expires_at is None or _utc(row.expires_at) > anchor)
        and row.review_status != "pending_review"
    ]
    expiring_30d = [
        row
        for row in grants
        if row.expires_at is not None
        and anchor < _utc(row.expires_at) <= anchor + timedelta(days=30)
    ]

    # Exactly six calendar labels including the current partial month, matching
    # the subscription billing drawer rather than leaking a seventh stub month.
    window_start = _month_start_months_ago(anchor.date(), 5)
    last_complete_day = anchor.date() - timedelta(days=1)
    amount = fact_engineering_amount()
    daily_rows = list(
        (
            await db.execute(
                select(CmdbCostFact.date, func.coalesce(func.sum(amount), 0).label("cost"))
                .where(
                    *cost_filters,
                    CmdbCostFact.date >= window_start,
                    CmdbCostFact.date <= last_complete_day,
                    engineering_row_clause(),
                )
                .group_by(CmdbCostFact.date)
                .order_by(CmdbCostFact.date)
            )
        ).all()
    )
    monthly: dict[str, Decimal] = {}
    for row in daily_rows:
        key = row.date.strftime("%Y-%m")
        monthly[key] = monthly.get(key, Decimal("0")) + Decimal(str(row.cost or 0))
    month_key = anchor.strftime("%Y-%m")
    current_cost = monthly.get(month_key, Decimal("0"))
    elapsed_days = max(1, anchor.day - 1)
    days_in_month = calendar.monthrange(anchor.year, anchor.month)[1]
    forecast = (current_cost * days_in_month / elapsed_days).quantize(Decimal("0.01"))
    # Shared analysis may compare only against a business/list-basis budget.
    # A legacy net budget belongs to the privileged invoice audience and must
    # neither affect this score nor be inferable from a burn percentage.
    budget = (
        Decimal(str(scope.budget_usd_monthly))
        if scope.budget_usd_monthly and scope.budget_measure == COST_MEASURE_LIST
        else None
    )
    overrun_rate = (
        max(Decimal("0"), forecast * 100 / budget - 100).quantize(Decimal("0.01"))
        if budget and budget > 0
        else Decimal("0")
    )

    resource_rows = list(
        (
            await db.execute(
                select(CmdbCostFact.resource_id)
                .where(
                    *cost_filters,
                    CmdbCostFact.date >= window_start,
                    CmdbCostFact.date <= last_complete_day,
                    CmdbCostFact.resource_id.is_not(None),
                    engineering_row_clause(),
                )
                .group_by(CmdbCostFact.resource_id)
                .add_columns(func.count(func.distinct(CmdbCostFact.date)).label("data_days"))
            )
        ).all()
    )
    resource_ids = {str(row.resource_id) for row in resource_rows}
    anomaly_start = last_complete_day - timedelta(days=29)
    anomaly_coverage = list(
        (
            await db.execute(
                select(
                    CmdbCostFact.resource_id,
                    func.count(func.distinct(CmdbCostFact.date)).label("data_days"),
                )
                .where(
                    *cost_filters,
                    CmdbCostFact.date >= anomaly_start,
                    CmdbCostFact.date <= last_complete_day,
                    CmdbCostFact.resource_id.is_not(None),
                    engineering_row_clause(),
                )
                .group_by(CmdbCostFact.resource_id)
            )
        ).all()
    )
    evaluable_keys = {
        (str(scope.integration_id), str(row.resource_id))
        for row in anomaly_coverage
        if int(row.data_days or 0) >= 7
    }
    if not evaluable_keys:
        anomalies: list[dict[str, Any]] = []
    elif anomaly_index is not None:
        # Fleet refresh: one scan per integration, filtered per scope here.
        # Same key the in-scan filter uses, so the result set is identical.
        anomalies = [
            row
            for row in await anomaly_index.for_integration(
                db,
                str(scope.integration_id),
                as_of_date=anchor.date(),
            )
            if (str(row["integration_id"]), str(row["resource_id"])) in evaluable_keys
        ]
    else:
        anomalies = await detect_anomalies(
            db,
            days=30,
            integration_ids=[str(scope.integration_id)],
            limit=None,
            measure=COST_MEASURE_LIST,
            as_of_date=anchor.date(),
            resource_keys=evaluable_keys,
        )
    anomaly_ids = {str(row["resource_id"]) for row in anomalies}

    # Resource efficiency follows the actual cloud-inventory relationship,
    # not only resource IDs present in the billing export.  A newly-created,
    # stopped, free-tier, or delayed-cost machine is still a real machine and
    # must be evaluated when the provider inventory already identifies it as
    # belonging to this subscription.
    server_resolution = await resolve_scope_servers(db, scope)
    servers = server_resolution.servers
    server_ids = {str(row.id) for row in servers}
    efficiency = (
        list(
            (
                await db.execute(
                    select(CmdbServerEfficiency).where(
                        CmdbServerEfficiency.server_id.in_(server_ids),
                    )
                )
            ).scalars()
        )
        if server_ids
        else []
    )
    classified = [row for row in efficiency if row.verdict in _CLASSIFIED]
    actionable = [row for row in classified if row.verdict in _ACTIONABLE]
    server_owner_ids = {str(row.owner_person_id) for row in servers if row.owner_person_id}
    active_server_owners = (
        {
            str(value)
            for value in (
                await db.execute(
                    select(IdentityPerson.id).where(
                        IdentityPerson.id.in_(server_owner_ids),
                        IdentityPerson.deleted_at.is_(None),
                        IdentityPerson.employment_status == "active",
                    )
                )
            ).scalars()
        }
        if server_owner_ids
        else set()
    )
    owner_active = bool(scope.owner_person_id and str(scope.owner_person_id) in active_people)
    # Keep per-subscription analysis on the same effective-owner resolver as
    # the platform-wide cloud/network provider. A server's stored owner FK is
    # intentionally nullable when it inherits the current subscription owner;
    # checking only that column here would reintroduce the old false gap for
    # cloud-scope snapshots and their AI recommendations.
    inherited_owners = await resolve_inherited_server_owners(db, servers) if servers else {}
    inherited_owner_ids = {
        str(value["person_id"])
        for value in inherited_owners.values()
        if value.get("person_id") is not None
    }
    active_inherited_owners = (
        {
            str(value)
            for value in (
                await db.execute(
                    select(IdentityPerson.id).where(
                        IdentityPerson.id.in_(inherited_owner_ids),
                        IdentityPerson.deleted_at.is_(None),
                        IdentityPerson.employment_status == "active",
                    )
                )
            ).scalars()
        }
        if inherited_owner_ids
        else set()
    )
    unowned_servers = [
        row
        for row in servers
        if not owner_active
        and str(row.owner_person_id) not in active_server_owners
        and str((inherited_owners.get(str(row.id)) or {}).get("person_id"))
        not in active_inherited_owners
    ]
    # The subscription intelligence contract covers the unified resource
    # directory, not only virtual machines.  Azure resource groups are stored
    # in raw provider metadata while ordinary scopes use the canonical scope
    # column, matching the directory visibility/ownership rules.
    scope_resource_clause = (
        or_(
            CmdbCloudResource.scope_external_id == scope.external_id,
            CmdbCloudResource.raw_metadata["resource_group"].as_string() == scope.external_id,
        )
        if scope.scope_type == SCOPE_TYPE_AZURE_RESOURCE_GROUP
        else CmdbCloudResource.scope_external_id == scope.external_id
    )
    # A subscription can contain tens of thousands of broad-inventory rows.
    # Keep the worker's memory and every coverage IN-clause bounded: only
    # counters plus a small evidence sample survive each keyset page.
    directory_resource_total = 0
    unowned_resource_count = 0
    monitoring_evaluable_count = 0
    monitoring_gap_count = 0
    cost_gap_count = 0
    changed_30d_count = 0
    unowned_resources: list[CmdbCloudResource] = []
    monitoring_gaps: list[CmdbCloudResource] = []
    cost_gaps: list[CmdbCloudResource] = []
    changed_30d: list[CmdbCloudResource] = []
    by_category: Counter[str] = Counter()
    by_resource_type: Counter[str] = Counter()
    monitoring_states: Counter[str] = Counter()
    cost_states: Counter[str] = Counter()
    directory_latest_seen: datetime | None = None
    directory_cursor: uuid.UUID | None = None
    while True:
        batch_query = select(CmdbCloudResource).where(
            CmdbCloudResource.integration_id == str(scope.integration_id),
            scope_resource_clause,
            CmdbCloudResource.deleted_at.is_(None),
        )
        if directory_cursor is not None:
            batch_query = batch_query.where(CmdbCloudResource.id > directory_cursor)
        batch = list(
            (
                await db.scalars(
                    batch_query.order_by(
                        CmdbCloudResource.id,
                    ).limit(_DIRECTORY_FACT_BATCH_SIZE)
                )
            ).all()
        )
        if not batch:
            break

        batch_ids = {str(row.id) for row in batch}
        batch_owners = await resolve_effective_resource_owners(db, batch)
        batch_governance = {
            str(row.resource_id): row
            for row in (
                await db.scalars(
                    select(CmdbResourceGovernance).where(
                        CmdbResourceGovernance.resource_id.in_(batch_ids),
                    )
                )
            ).all()
        }
        batch_coverage = await resolve_resource_coverage(db, batch)
        for resource in batch:
            resource_id = str(resource.id)
            governance = batch_governance.get(resource_id)
            coverage = batch_coverage[resource_id]
            directory_resource_total += 1
            by_category[resource.category] += 1
            by_resource_type[resource.resource_type] += 1
            monitoring_states[coverage.monitoring_status] += 1
            cost_state = (
                coverage.cost_source if coverage.cost_status == "covered" else coverage.cost_status
            ) or "unknown"
            cost_states[cost_state] += 1
            directory_latest_seen = max(
                directory_latest_seen or resource.last_seen_at,
                resource.last_seen_at,
            )

            resource_governed = bool(
                governance
                and (governance.business_owner_person_id or governance.technical_owner_person_id)
            )
            if resource_id not in batch_owners and not resource_governed:
                unowned_resource_count += 1
                if len(unowned_resources) < _DIRECTORY_EVIDENCE_LIMIT:
                    unowned_resources.append(resource)
            if coverage.monitoring_status in {"covered", "missing"}:
                monitoring_evaluable_count += 1
            if coverage.monitoring_status == "missing":
                monitoring_gap_count += 1
                if len(monitoring_gaps) < _DIRECTORY_EVIDENCE_LIMIT:
                    monitoring_gaps.append(resource)
            if coverage.cost_status == "not_available":
                cost_gap_count += 1
                if len(cost_gaps) < _DIRECTORY_EVIDENCE_LIMIT:
                    cost_gaps.append(resource)
            if resource.last_changed_at and _utc(resource.last_changed_at) >= anchor - timedelta(
                days=30
            ):
                changed_30d_count += 1
                if len(changed_30d) < _DIRECTORY_EVIDENCE_LIMIT:
                    changed_30d.append(resource)

        directory_cursor = batch[-1].id
        if len(batch) < _DIRECTORY_FACT_BATCH_SIZE:
            break

    top_services = list(
        (
            await db.execute(
                select(CmdbCostFact.service, func.coalesce(func.sum(amount), 0).label("cost"))
                .where(
                    *cost_filters,
                    CmdbCostFact.date >= anchor.date() - timedelta(days=29),
                    CmdbCostFact.date <= last_complete_day,
                    engineering_row_clause(),
                )
                .group_by(CmdbCostFact.service)
                .order_by(func.sum(amount).desc())
                .limit(10)
            )
        ).all()
    )
    top_resources = list(
        (
            await db.execute(
                select(CmdbCostFact.resource_id, func.coalesce(func.sum(amount), 0).label("cost"))
                .where(
                    *cost_filters,
                    CmdbCostFact.date >= anchor.date() - timedelta(days=29),
                    CmdbCostFact.date <= last_complete_day,
                    CmdbCostFact.resource_id.is_not(None),
                    engineering_row_clause(),
                )
                .group_by(CmdbCostFact.resource_id)
                .order_by(func.sum(amount).desc())
                .limit(10)
            )
        ).all()
    )

    evidence = [EvidenceObject("scope", str(scope.id))]
    evidence.extend(EvidenceObject("grant", str(row.id)) for row in identity_gaps)
    evidence.extend(EvidenceObject("grant", str(row.id)) for row in unmapped_roles)
    evidence.extend(EvidenceObject("grant", str(row.id)) for row in permanent)
    evidence.extend(EvidenceObject("grant", str(row.id)) for row in privileged_pending_review)
    evidence.extend(EvidenceObject("grant", str(row.id)) for row in pending_grants)
    evidence.extend(EvidenceObject("grant", str(row.id)) for row in expiring_30d)
    evidence.extend(EvidenceObject("server", str(row.server_id)) for row in actionable)
    evidence.extend(EvidenceObject("server", str(row.id)) for row in unowned_servers)
    evidence.extend(EvidenceObject("resource", str(row.id)) for row in unowned_resources[:20])
    evidence.extend(EvidenceObject("resource", str(row.id)) for row in monitoring_gaps[:20])
    evidence.extend(EvidenceObject("resource", str(row.id)) for row in cost_gaps[:20])
    evidence.extend(EvidenceObject("resource", str(row.id)) for row in changed_30d[:20])
    evidence.extend(
        EvidenceObject("costservice", f"{scope.integration_id}:{row.service or 'unclassified'}")
        for row in top_services
    )
    evidence.extend(
        EvidenceObject("costresource", f"{scope.integration_id}:{value}") for value in anomaly_ids
    )
    evidence.extend(
        EvidenceObject("costresource", f"{scope.integration_id}:{row.resource_id}")
        for row in top_resources
    )
    _, reference_map = build_reference_map(evidence)
    reverse = {(value["kind"], value["object_id"]): key for key, value in reference_map.items()}

    def refs(kind: str, rows: list[Any]) -> tuple[str, ...]:
        return _reference_keys(reverse, kind, rows)

    cost_as_of = daily_rows[-1].date if daily_rows else None
    efficiency_as_of = max((row.computed_at for row in efficiency), default=None)
    confirmation_cycles = (
        3
        if scope.last_owner_confirmed_at is None
        else max(0, (anchor - _utc(scope.last_owner_confirmed_at)).days // 180)
    )
    cost_days = len(
        {row.date for row in daily_rows if row.date >= anchor.date() - timedelta(days=29)}
    )
    metrics = [
        MetricInput(
            "cloud.cost.budget_coverage",
            MetricKind.COVERAGE,
            100 if budget else 0,
            confidence=100,
            evidence_refs=refs("scope", [scope.id]),
            data_as_of=scope.updated_at,
        ),
        MetricInput(
            "cloud.cost.budget_overrun_rate",
            MetricKind.FAILURE_RATE,
            overrun_rate,
            confidence=100 if budget and cost_days else 0,
            evidence_refs=refs("scope", [scope.id]),
            data_as_of=datetime.combine(cost_as_of, datetime.min.time(), tzinfo=UTC)
            if cost_as_of
            else None,
        ),
        MetricInput(
            "cloud.cost.anomaly_rate",
            MetricKind.FAILURE_RATE,
            _pct(len(anomaly_ids), len(evaluable_keys)),
            confidence=_pct(len(evaluable_keys), len(resource_ids)) if resource_ids else 0,
            evidence_refs=refs(
                "costresource", [f"{scope.integration_id}:{value}" for value in anomaly_ids]
            ),
            data_as_of=datetime.combine(cost_as_of, datetime.min.time(), tzinfo=UTC)
            if cost_as_of
            else None,
        ),
        MetricInput(
            "cloud.cost.data_coverage",
            MetricKind.COVERAGE,
            _pct(cost_days, 30),
            confidence=100 if daily_rows else 0,
            data_as_of=datetime.combine(cost_as_of, datetime.min.time(), tzinfo=UTC)
            if cost_as_of
            else None,
        ),
        MetricInput(
            "cloud.resource.classified_coverage",
            MetricKind.COVERAGE,
            _pct(len(classified), len(servers)) if servers else 100,
            confidence=100 if servers else 0,
            data_as_of=efficiency_as_of or scope.last_synced_at,
        ),
        MetricInput(
            "cloud.resource.actionable_rate",
            MetricKind.FAILURE_RATE,
            _pct(len(actionable), len(classified)) if classified else 0,
            confidence=_pct(len(classified), len(servers)) if servers else 0,
            evidence_refs=refs("server", [row.server_id for row in actionable]),
            data_as_of=efficiency_as_of or scope.last_synced_at,
        ),
        MetricInput(
            "cloud.resource.owner_coverage",
            MetricKind.COVERAGE,
            _pct(directory_resource_total - unowned_resource_count, directory_resource_total)
            if directory_resource_total
            else 100,
            confidence=100 if directory_resource_total else 0,
            evidence_refs=refs("resource", [row.id for row in unowned_resources]),
            data_as_of=scope.last_synced_at,
        ),
        MetricInput(
            "cloud.resource.monitoring_coverage",
            MetricKind.COVERAGE,
            _pct(
                monitoring_evaluable_count - monitoring_gap_count,
                monitoring_evaluable_count,
            )
            if monitoring_evaluable_count
            else 100,
            confidence=100 if directory_resource_total else 0,
            evidence_refs=refs("resource", [row.id for row in monitoring_gaps]),
            data_as_of=scope.last_synced_at,
        ),
        MetricInput(
            "cloud.access.governed_coverage",
            MetricKind.COVERAGE,
            _pct(len(governed), len(grants)) if grants else 100,
            confidence=100,
            evidence_refs=refs("grant", [row.id for row in [*identity_gaps, *unmapped_roles]]),
            data_as_of=max((row.updated_at for row in grants), default=scope.last_synced_at),
        ),
        MetricInput(
            "cloud.access.permanent_rate",
            MetricKind.FAILURE_RATE,
            _pct(len(permanent), len(grants)) if grants else 0,
            confidence=100,
            evidence_refs=refs("grant", [row.id for row in permanent]),
            data_as_of=max((row.updated_at for row in grants), default=scope.last_synced_at),
        ),
        MetricInput(
            "cloud.access.privileged_review_coverage",
            MetricKind.COVERAGE,
            _pct(len(privileged) - len(privileged_pending_review), len(privileged))
            if privileged
            else 100,
            confidence=100,
            evidence_refs=refs("grant", [row.id for row in privileged_pending_review]),
            data_as_of=max((row.updated_at for row in privileged), default=scope.last_synced_at),
        ),
        MetricInput(
            "cloud.access.identity_gap_rate",
            MetricKind.FAILURE_RATE,
            _pct(len(identity_gaps), len(grants)) if grants else 0,
            confidence=100,
            evidence_refs=refs("grant", [row.id for row in identity_gaps]),
            data_as_of=max((row.updated_at for row in grants), default=scope.last_synced_at),
        ),
        MetricInput(
            "cloud.owner.active_coverage",
            MetricKind.COVERAGE,
            100 if owner_active else 0,
            confidence=100,
            evidence_refs=refs("scope", [scope.id]),
            data_as_of=scope.updated_at,
        ),
        MetricInput(
            "cloud.owner.confirmation_freshness",
            MetricKind.FRESHNESS,
            confirmation_cycles,
            confidence=100,
            evidence_refs=refs("scope", [scope.id]),
            data_as_of=scope.last_owner_confirmed_at or scope.created_at,
        ),
        MetricInput(
            "cloud.lifecycle.state_score",
            MetricKind.DIRECT,
            100 if scope.status == SCOPE_STATUS_ACTIVE else 20,
            confidence=100,
            evidence_refs=refs("scope", [scope.id]),
            data_as_of=scope.updated_at,
        ),
        MetricInput(
            "cloud.sync.scope_freshness",
            MetricKind.FRESHNESS,
            _missed_days(scope.last_synced_at, anchor),
            confidence=100,
            evidence_refs=refs("scope", [scope.id]),
            data_as_of=scope.last_synced_at,
        ),
        MetricInput(
            "cloud.sync.cost_freshness",
            MetricKind.FRESHNESS,
            _missed_days(cost_as_of, anchor),
            confidence=100 if cost_as_of else 0,
            data_as_of=datetime.combine(cost_as_of, datetime.min.time(), tzinfo=UTC)
            if cost_as_of
            else None,
        ),
        MetricInput(
            "cloud.sync.efficiency_freshness",
            MetricKind.FRESHNESS,
            _missed_days(efficiency_as_of, anchor),
            confidence=100 if efficiency_as_of else 0,
            data_as_of=efficiency_as_of,
        ),
    ]
    health = score_cloud_scope(metrics)
    monthly_rows = [
        {"month": key, "business_cost_usd": str(value.quantize(Decimal("0.01")))}
        for key, value in sorted(monthly.items())
    ]
    # This is deliberately deterministic and aggregate-only.  A confidence
    # score tells a user *that* the analysis is incomplete; this contract
    # tells them exactly which source is missing, what Helix did find, and
    # whether the source is not applicable rather than silently treating it
    # as a failure.  It is safe to show to every authorized scope member.
    efficiency_state = (
        "not_applicable"
        if not servers
        else "ready"
        if len(classified) == len(servers)
        else "missing"
        if not efficiency
        else "partial"
    )
    cost_state = "missing" if not daily_rows else "ready" if cost_days >= 7 else "partial"
    inventory_state = "ready" if directory_resource_total else "missing"
    data_quality = {
        "sources": [
            {
                "key": "inventory",
                "state": inventory_state,
                "reason_code": (
                    "inventory_resolved" if directory_resource_total else "no_linked_inventory"
                ),
                "available_count": directory_resource_total,
                "total_count": None,
                "data_as_of": scope.last_synced_at,
            },
            {
                "key": "efficiency",
                "state": efficiency_state,
                "reason_code": (
                    "no_inventory"
                    if not servers
                    else "efficiency_ready"
                    if efficiency_state == "ready"
                    else "no_efficiency_result"
                    if not efficiency
                    else "some_machines_unclassified"
                ),
                "available_count": len(classified),
                "total_count": len(servers),
                "data_as_of": efficiency_as_of,
            },
            {
                "key": "cost",
                "state": cost_state,
                "reason_code": (
                    "cost_ready"
                    if cost_state == "ready"
                    else "no_cost_facts"
                    if cost_state == "missing"
                    else "cost_history_too_short"
                ),
                "available_count": cost_days,
                "total_count": 30,
                "data_as_of": (
                    datetime.combine(cost_as_of, datetime.min.time(), tzinfo=UTC)
                    if cost_as_of
                    else None
                ),
            },
            {
                "key": "scope_sync",
                "state": "ready" if _missed_days(scope.last_synced_at, anchor) <= 1 else "partial",
                "reason_code": (
                    "scope_sync_current"
                    if _missed_days(scope.last_synced_at, anchor) <= 1
                    else "scope_sync_stale"
                ),
                "available_count": None,
                "total_count": None,
                "data_as_of": scope.last_synced_at,
            },
        ],
    }
    aggregate_facts = {
        "scope_state": {
            "status": scope.status,
            "environment": scope.environment,
            "business_budget_configured": budget is not None,
            "budget_basis_mismatch": bool(
                scope.budget_usd_monthly and scope.budget_measure != COST_MEASURE_LIST
            ),
        },
        "costs": {
            "currency": "USD",
            "routing": cost_routing["routing"],
            "has_cost_data": bool(daily_rows),
            "zero_consumption": bool(daily_rows and current_cost == 0),
            "current_month_business_cost": str(current_cost.quantize(Decimal("0.01"))),
            "month_end_forecast": str(forecast),
            "budget_burn_percent": str(_pct(forecast, budget)) if budget else None,
            "monthly_trend": monthly_rows,
            "top_service_categories": [
                {
                    "ref": reverse.get(
                        ("costservice", f"{scope.integration_id}:{row.service or 'unclassified'}")
                    ),
                    "business_cost": str(Decimal(str(row.cost or 0)).quantize(Decimal("0.01"))),
                }
                for row in top_services
            ],
            "top_resources": [
                {
                    "ref": reverse.get(
                        ("costresource", f"{scope.integration_id}:{row.resource_id}")
                    ),
                    "business_cost": str(Decimal(str(row.cost or 0)).quantize(Decimal("0.01"))),
                }
                for row in top_resources
            ],
            "anomaly_refs": list(
                refs(
                    "costresource",
                    [f"{scope.integration_id}:{value}" for value in anomaly_ids],
                )
            ),
        },
        "resources": {
            "total": directory_resource_total,
            "server_total": len(servers),
            "attribution": {
                "direct_inventory": server_resolution.direct_count,
                "cost_fallback": server_resolution.cost_fallback_count,
                "coverage_state": ("resolved" if servers else "no_linked_inventory"),
            },
            "classified": len(classified),
            "actionable": len(actionable),
            "without_owner": unowned_resource_count,
            "without_monitoring": monitoring_gap_count,
            "without_cost_data": cost_gap_count,
            "changed_30d": changed_30d_count,
            "by_category": [
                {"key": key, "count": value} for key, value in by_category.most_common(20)
            ],
            "by_resource_type": [
                {"key": key, "count": value} for key, value in by_resource_type.most_common(30)
            ],
            "monitoring_coverage": dict(sorted(monitoring_states.items())),
            "cost_coverage": dict(sorted(cost_states.items())),
            "estimated_monthly_savings": str(
                sum(
                    (Decimal(str(row.estimated_monthly_savings_usd or 0)) for row in actionable),
                    Decimal("0"),
                ).quantize(Decimal("0.01"))
            ),
            "actionable_refs": list(refs("server", [row.server_id for row in actionable])),
            "owner_gap_refs": list(refs("resource", [row.id for row in unowned_resources])),
            "monitoring_gap_refs": list(refs("resource", [row.id for row in monitoring_gaps])),
            "cost_gap_refs": list(refs("resource", [row.id for row in cost_gaps])),
            "changed_refs": list(refs("resource", [row.id for row in changed_30d])),
        },
        "access": {
            "active_grants": len(grants),
            "pending_grants": len(pending_grants),
            "permanent_grants": len(permanent),
            "privileged_grants": len(privileged),
            "pending_privileged_reviews": len(privileged_pending_review),
            "identity_gaps": len(identity_gaps),
            "unmapped_roles": len(unmapped_roles),
            "expiring_30d": len(expiring_30d),
            "pending_refs": list(refs("grant", [row.id for row in pending_grants])),
            "identity_gap_refs": list(refs("grant", [row.id for row in identity_gaps])),
            "unmapped_role_refs": list(refs("grant", [row.id for row in unmapped_roles])),
            "privileged_review_refs": list(
                refs("grant", [row.id for row in privileged_pending_review])
            ),
            "expiring_refs": list(refs("grant", [row.id for row in expiring_30d])),
        },
        "governance": {
            "active_owner": owner_active,
            "owner_confirmation_missed_cycles": confirmation_cycles,
        },
        "sync": {
            "scope_missed_days": _missed_days(scope.last_synced_at, anchor),
            "cost_missed_days": _missed_days(cost_as_of, anchor),
            "efficiency_missed_days": _missed_days(efficiency_as_of, anchor),
        },
        "data_quality": data_quality,
    }
    payload = json_value(aggregate_facts, decimal_as_string=True)
    assert_llm_safe(payload)
    source_times = [
        scope.last_synced_at or scope.updated_at or scope.created_at or anchor,
    ]
    if cost_as_of:
        source_times.append(datetime.combine(cost_as_of, datetime.min.time(), tzinfo=UTC))
    if efficiency_as_of:
        source_times.append(efficiency_as_of)
    if directory_latest_seen:
        source_times.append(directory_latest_seen)
    return CloudScopeFactBundle(
        health=health,
        aggregate_facts=payload,
        reference_map=reference_map,
        data_as_of=min(_utc(value) for value in source_times),
    )