"""Read-side queries for the cloud efficiency / cost-optimization surface.

Pure reads over ``cmdb_server_efficiency`` (+ ``cmdb_servers`` for names /
owner scoping, ``integrations`` for the cloud label, and
``cmdb_server_metrics_daily`` for the detail sparkline). The efficiency
rows themselves are produced by ``efficiency.compute_efficiency_for_integration``.
"""
from __future__ import annotations

from datetime import UTC, datetime, timedelta
from decimal import Decimal
from typing import Any

from sqlalchemy import and_, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import aliased

from app.core.txn import best_effort
from app.domains.cmdb.models import (
    EFFICIENCY_RECO_NONE,
    EFFICIENCY_TRIAGE_SNOOZED,
    CmdbCloudScope,
    CmdbCostFact,
    CmdbServer,
    CmdbServerEfficiency,
    CmdbServerEfficiencyTriage,
    CmdbServerMetricsDaily,
)
from app.domains.cmdb.visibility import OwnerVisibility, visibility_clause
from app.identity.person_models import IdentityOrgMembership
from app.integrations.models import Integration

# Tenant-root markers stripped from org paths (mirrors
# IdentityPerson.department_chain).
_TENANT_ROOTS = {"pingcap", "pingcap-ati"}
_UNASSIGNED = "__unassigned__"

# Verdicts that carry an owner-actionable recommendation (idle → stop,
# oversized → downsize). Drives the "with_recommendation" / actionable
# counts and the ``actionable_only`` list filter. Keep in sync with the
# closed vocabulary in models.py.
_ACTIONABLE_VERDICTS = ("idle", "oversized")
_METRICS_PROVIDERS = ("aws", "aliyun", "alicloud_intl", "azure", "gcp")


def _actionable_conds() -> list[Any]:
    """A recommendation is owner-actionable when the verdict is idle/oversized
    AND it still carries a live recommendation. A box the compute pass
    NEUTRALIZED because it is already stopped/terminated keeps its idle/
    oversized verdict for visibility but has ``recommendation='none'`` and
    $0 savings (see efficiency.compute — 'instance_not_running'). Such a box
    must NOT count as actionable/pending or the owner gets nagged to stop a
    machine that is already stopped and the '待处理'/可优化机器 counts inflate.
    Single source of truth for every actionable/pending predicate."""
    return [
        CmdbServerEfficiency.verdict.in_(_ACTIONABLE_VERDICTS),
        CmdbServerEfficiency.recommendation != EFFICIENCY_RECO_NONE,
    ]


def _owner_match_clause(owner_person_id: str) -> Any:
    """Match instances owned by ``owner_person_id`` — the SAME 'effective
    owner' the 归属 column shows: the stored server owner, OR (for servers
    with no explicit owner) the owner inherited from the cloud subscription
    the server is attributed to. The inherited link is the cost-facts
    attribution (``resource_id == instance_id`` → ``scope_id``), the same
    link resolve_inherited_server_owners uses. Uses an aliased sub-select so
    the paginated query stays a single statement. (GCP-standard-export boxes,
    which have no cost facts, are only matched by their stored owner — they
    also can't be priced, so they never surface an inherited owner either.)"""
    owned_scopes = (
        select(CmdbCloudScope.id)
        .where(
            CmdbCloudScope.owner_person_id == owner_person_id,
            CmdbCloudScope.deleted_at.is_(None),
        )
        .scalar_subquery()
    )
    srv = aliased(CmdbServer)
    inherited_ids = (
        select(srv.id)
        .join(
            CmdbCostFact,
            and_(
                CmdbCostFact.integration_id == srv.integration_id,
                CmdbCostFact.resource_id == srv.instance_id,
            ),
        )
        .where(
            srv.owner_person_id.is_(None),
            CmdbCostFact.scope_id.in_(owned_scopes),
        )
    )
    return or_(
        CmdbServer.owner_person_id == owner_person_id,
        CmdbServer.id.in_(inherited_ids),
    )


def _dept_chain(org_path: str | None, org_node_name: str | None) -> list[str]:
    raw = org_path or org_node_name
    if not raw:
        return []
    parts = [p.strip() for p in raw.split("/") if p.strip()]
    if parts and parts[0].lower() in _TENANT_ROOTS:
        parts = parts[1:]
    if not parts and org_node_name:
        parts = [org_node_name]
    return parts


def _f(v: Any) -> float | None:
    if v is None:
        return None
    if isinstance(v, Decimal):
        return float(v)
    return float(v)


def _scope_server_exists(scope: CmdbCloudScope) -> Any:
    """Correlated EXISTS: the outer ``CmdbServer`` has >=1 cost-fact line
    attributing it to this subscription. Servers carry no direct scope FK,
    so ``cmdb_cost_facts`` (``resource_id == instance_id`` within the same
    integration → ``scope_id``) is the established server→subscription
    attribution path — the same one ``cost_service`` uses for chargeback.
    A server with no billing line under the scope simply doesn't roll up
    here (it still shows on the standalone efficiency page / its owner's
    server list)."""
    return (
        select(1)
        .select_from(CmdbCostFact)
        .where(
            CmdbCostFact.scope_id == str(scope.id),
            CmdbCostFact.integration_id == scope.integration_id,
            CmdbCostFact.resource_id == CmdbServer.instance_id,
        )
        .correlate(CmdbServer)
        .exists()
    )


def _base_conditions(
    visibility: OwnerVisibility | None,
    integration_id: str | None,
    scope: CmdbCloudScope | None = None,
) -> list[Any]:
    conds: list[Any] = [
        CmdbServer.deleted_at.is_(None),
        # Unsupported providers are coverage=N/A, not "no metrics". Keeping
        # their legacy efficiency rows in this page created a permanent red
        # queue nobody could fix. The directory still lists those resources
        # and labels monitoring as not_supported.
        CmdbServer.integration_id.in_(select(Integration.id).where(
            Integration.provider.in_(_METRICS_PROVIDERS),
        )),
    ]
    if visibility is not None:
        conds.append(visibility_clause(visibility))
    if integration_id:
        conds.append(CmdbServerEfficiency.integration_id == integration_id)
    if scope is not None:
        conds.append(_scope_server_exists(scope))
    return conds


_EFF_SERVER_JOIN = CmdbServerEfficiency.__table__.join(
    CmdbServer.__table__,
    CmdbServer.id == CmdbServerEfficiency.server_id,
)


def _effective_open_clause() -> Any:
    """SQL predicate: this recommendation is still awaiting owner triage.
    True when there is no triage row, when the stored triage went stale
    (the verdict drifted away from ``verdict_at_decision``), or when a
    snooze has elapsed. Mirrors ``_effective_triage`` computed in Python
    for row rendering — keep the two in sync."""
    t = CmdbServerEfficiencyTriage
    return or_(
        t.id.is_(None),
        t.verdict_at_decision != CmdbServerEfficiency.verdict,
        and_(
            t.status == EFFICIENCY_TRIAGE_SNOOZED,
            t.snooze_until.isnot(None),
            t.snooze_until <= func.now(),
        ),
    )


def _effective_triage(
    eff: CmdbServerEfficiency,
    triage: CmdbServerEfficiencyTriage | None,
    now: datetime,
) -> dict[str, Any]:
    """Freshness-aware view of a server's triage. A stored decision only
    counts while the verdict it was made against still holds; a drifted
    verdict or an elapsed snooze collapses back to ``open``."""
    if triage is None:
        return {"triage_status": "open"}
    stale = triage.verdict_at_decision != eff.verdict
    snoozed_elapsed = (
        triage.status == EFFICIENCY_TRIAGE_SNOOZED
        and triage.snooze_until is not None
        and triage.snooze_until <= now
    )
    if stale or snoozed_elapsed:
        return {"triage_status": "open", "triage_stale": stale}
    return {
        "triage_status": triage.status,
        "triage_reason": triage.reason,
        "triage_decided_by_person_id": triage.decided_by_person_id,
        "triage_decided_at": (
            triage.decided_at.isoformat() if triage.decided_at else None
        ),
        "triage_snooze_until": (
            triage.snooze_until.isoformat() if triage.snooze_until else None
        ),
    }


async def efficiency_summary(
    db: AsyncSession,
    *,
    visibility: OwnerVisibility | None = None,
    integration_id: str | None = None,
    scope: CmdbCloudScope | None = None,
) -> dict[str, Any]:
    """KPI rollup: counts by verdict, total attributed cost + estimated
    savings, and a per-cloud breakdown. ``visibility`` scopes to a
    self-service reader's set (server owner ∪ owned-subscription
    attribution); ``None`` = org-wide (admin-tier). ``scope`` narrows to
    the servers attributed to one subscription."""
    conds = _base_conditions(visibility, integration_id, scope)

    # By verdict.
    vq = (
        select(
            CmdbServerEfficiency.verdict,
            func.count().label("cnt"),
            func.coalesce(
                func.sum(CmdbServerEfficiency.estimated_monthly_savings_usd), 0,
            ).label("savings"),
        )
        .select_from(_EFF_SERVER_JOIN)
        .where(*conds)
        .group_by(CmdbServerEfficiency.verdict)
    )
    by_verdict = [
        {"verdict": v, "count": int(c), "estimated_monthly_savings_usd": _f(s)}
        for v, c, s in (await db.execute(vq)).all()
    ]

    # Totals.
    tq = (
        select(
            func.count().label("cnt"),
            func.coalesce(
                func.sum(CmdbServerEfficiency.estimated_monthly_savings_usd), 0,
            ),
            func.coalesce(
                func.sum(CmdbServerEfficiency.monthly_cost_usd), 0,
            ),
        )
        .select_from(_EFF_SERVER_JOIN)
        .where(*conds)
    )
    trow = (await db.execute(tq)).one()
    total_servers = int(trow[0])
    total_savings = _f(trow[1])
    total_cost = _f(trow[2])

    # "可优化机器" = actionable recommendations. Counted with a dedicated
    # query (not summed from by_verdict) so NEUTRALIZED stopped boxes — which
    # keep an idle/oversized verdict but have recommendation='none' — are
    # excluded. Summing by_verdict would re-include them and inflate the KPI.
    wq = (
        select(func.count())
        .select_from(_EFF_SERVER_JOIN)
        .where(*conds, *_actionable_conds())
    )
    with_reco = int((await db.execute(wq)).scalar() or 0)

    # Pending = actionable recommendations still awaiting owner triage
    # (no live keep/ack/snooze decision). This is the "待处理" queue count.
    pq = (
        select(func.count())
        .select_from(
            _EFF_SERVER_JOIN.join(
                CmdbServerEfficiencyTriage.__table__,
                CmdbServerEfficiencyTriage.server_id
                == CmdbServerEfficiency.server_id,
                isouter=True,
            )
        )
        .where(
            *conds,
            *_actionable_conds(),
            _effective_open_clause(),
        )
    )
    pending = int((await db.execute(pq)).scalar() or 0)

    # "仅订阅级成本" — these machines belong to a priced cloud scope/connector,
    # but the provider export has no exact per-instance grain. This is a
    # precision limitation, not financially unattributed spend. Keep the old
    # response key for clients during rollout, but the product-facing metric
    # is ``cost_scope_only``.
    uaq = (
        select(func.count())
        .select_from(_EFF_SERVER_JOIN)
        .where(
            *conds,
            *_actionable_conds(),
            CmdbServerEfficiency.monthly_cost_usd.is_(None),
        )
    )
    cost_scope_only = int((await db.execute(uaq)).scalar() or 0)

    # By cloud (integration).
    cq = (
        select(
            CmdbServerEfficiency.integration_id,
            Integration.name,
            Integration.provider,
            func.count().label("cnt"),
            func.coalesce(
                func.sum(CmdbServerEfficiency.estimated_monthly_savings_usd), 0,
            ).label("savings"),
        )
        .select_from(
            _EFF_SERVER_JOIN.join(
                Integration.__table__,
                Integration.id == CmdbServerEfficiency.integration_id,
                isouter=True,
            )
        )
        .where(*conds)
        .group_by(
            CmdbServerEfficiency.integration_id,
            Integration.name,
            Integration.provider,
        )
        .order_by(func.coalesce(
            func.sum(CmdbServerEfficiency.estimated_monthly_savings_usd), 0,
        ).desc())
    )
    by_cloud = [
        {
            "integration_id": iid,
            "name": name,
            "provider": provider,
            "count": int(cnt),
            "estimated_monthly_savings_usd": _f(savings),
        }
        for iid, name, provider, cnt, savings in (await db.execute(cq)).all()
    ]

    return {
        "totals": {
            "servers": total_servers,
            "with_recommendation": with_reco,
            "pending": pending,
            "cost_scope_only": cost_scope_only,
            "cost_unattributed": cost_scope_only,
            "estimated_monthly_savings_usd": total_savings,
            "attributed_monthly_cost_usd": total_cost,
        },
        "by_verdict": by_verdict,
        "by_cloud": by_cloud,
    }


def _instance_row(
    eff: CmdbServerEfficiency,
    srv: CmdbServer,
    integ: Integration | None,
    triage: CmdbServerEfficiencyTriage | None = None,
    now: datetime | None = None,
) -> dict[str, Any]:
    now = now or datetime.now(UTC)
    row = {
        "server_id": str(eff.server_id),
        "integration_id": str(eff.integration_id),
        "cloud_name": integ.name if integ else None,
        "provider": integ.provider if integ else None,
        "instance_id": srv.instance_id,
        "name": srv.name,
        "region": srv.region,
        "instance_type": srv.instance_type,
        "vcpu": srv.vcpu,
        "memory_mb": srv.memory_mb,
        "status": srv.status,
        "owner_person_id": srv.owner_person_id,
        "window_days": eff.window_days,
        "data_days": eff.data_days,
        "cpu_avg": _f(eff.cpu_avg),
        "cpu_p95": _f(eff.cpu_p95),
        "cpu_max": _f(eff.cpu_max),
        "mem_avg": _f(eff.mem_avg),
        "mem_p95": _f(eff.mem_p95),
        "net_in_avg": eff.net_in_avg,
        "net_out_avg": eff.net_out_avg,
        "verdict": eff.verdict,
        "verdict_reasons": eff.verdict_reasons or [],
        "recommendation": eff.recommendation,
        "recommended_action_detail": eff.recommended_action_detail,
        "monthly_cost_usd": _f(eff.monthly_cost_usd),
        "estimated_monthly_savings_usd": _f(eff.estimated_monthly_savings_usd),
        "cost_currency": eff.cost_currency,
        "ai_narrative": eff.ai_narrative,
        "computed_at": eff.computed_at.isoformat() if eff.computed_at else None,
    }
    row.update(_effective_triage(eff, triage, now))
    # Neutralized (already-stopped) boxes keep an idle/oversized verdict but
    # recommendation='none' — they are NOT actionable/pending (mirrors
    # _actionable_conds; keeps the row visible without nagging the owner).
    row["actionable"] = (
        eff.verdict in _ACTIONABLE_VERDICTS
        and eff.recommendation != EFFICIENCY_RECO_NONE
    )
    row["is_pending"] = row["actionable"] and row["triage_status"] == "open"
    return row


async def list_instances(
    db: AsyncSession,
    *,
    visibility: OwnerVisibility | None = None,
    integration_id: str | None = None,
    scope: CmdbCloudScope | None = None,
    verdict: str | None = None,
    recommendation: str | None = None,
    min_savings: float | None = None,
    owner_person_id: str | None = None,
    actionable_only: bool = False,
    pending_only: bool = False,
    sort: str = "savings",
    page: int = 1,
    page_size: int = 20,
) -> tuple[list[dict[str, Any]], int]:
    conds = _base_conditions(visibility, integration_id, scope)
    if actionable_only or pending_only:
        conds.extend(_actionable_conds())
    if pending_only:
        conds.append(_effective_open_clause())
    if owner_person_id:
        conds.append(_owner_match_clause(owner_person_id))
    if verdict:
        conds.append(CmdbServerEfficiency.verdict == verdict)
    if recommendation:
        conds.append(CmdbServerEfficiency.recommendation == recommendation)
    if min_savings is not None:
        conds.append(
            CmdbServerEfficiency.estimated_monthly_savings_usd >= min_savings,
        )

    # LEFT JOIN triage so the row can render its (freshness-aware) status
    # and ``pending_only`` can filter on it.
    triage_join = _EFF_SERVER_JOIN.join(
        CmdbServerEfficiencyTriage.__table__,
        CmdbServerEfficiencyTriage.server_id == CmdbServerEfficiency.server_id,
        isouter=True,
    )

    count_stmt = (
        select(func.count())
        .select_from(triage_join)
        .where(*conds)
    )
    total = int((await db.execute(count_stmt)).scalar() or 0)

    order: list[Any]
    if sort == "cost":
        order = [CmdbServerEfficiency.monthly_cost_usd.desc()]
    elif sort == "cpu":
        # "least CPU first" — but no-metrics / insufficient-data rows carry a
        # NULL cpu_p95, and MySQL/TiDB sort NULLs FIRST on ASC, which would
        # push unknown-utilization boxes above the genuinely idle ones. Force
        # NULLs last with the portable ``is_(None)`` prefix (NULLS LAST syntax
        # isn't valid on MySQL/TiDB — see the CI TiDB-parity guard).
        order = [
            CmdbServerEfficiency.cpu_p95.is_(None),
            CmdbServerEfficiency.cpu_p95.asc(),
        ]
    else:
        order = [CmdbServerEfficiency.estimated_monthly_savings_usd.desc()]

    stmt = (
        select(
            CmdbServerEfficiency, CmdbServer, Integration,
            CmdbServerEfficiencyTriage,
        )
        .join(CmdbServer, CmdbServer.id == CmdbServerEfficiency.server_id)
        .join(
            Integration,
            Integration.id == CmdbServerEfficiency.integration_id,
            isouter=True,
        )
        .join(
            CmdbServerEfficiencyTriage,
            CmdbServerEfficiencyTriage.server_id
            == CmdbServerEfficiency.server_id,
            isouter=True,
        )
        .where(*conds)
        .order_by(*order, CmdbServerEfficiency.server_id)
        .limit(page_size)
        .offset((page - 1) * page_size)
    )
    now = datetime.now(UTC)
    rows = (await db.execute(stmt)).all()
    items = [
        _instance_row(eff, srv, integ, triage, now)
        for eff, srv, integ, triage in rows
    ]
    await _augment_inherited_owner_fields(
        db, items, [srv for _, srv, _, _ in rows],
    )
    return items, total


async def _augment_inherited_owner_fields(
    db: AsyncSession, items: list[dict[str, Any]], servers: list[CmdbServer],
) -> None:
    """Attach the resolved owner PERSON to each row so the 归属 column shows
    who to actually contact — not just a subscription tag. A stored
    ``owner_person_id`` wins; otherwise resolve the owner inherited from the
    cloud subscription (live, never stored) and flag ``owner_inherited`` with
    the subscription breadcrumbs. BOTH cases expose the same
    ``effective_owner_{label,employee_no,employment_status}`` fields
    (mirroring the servers list) so a leaver owner reads as departed and the
    row is searchable by owner. Mutates ``items`` in place; keyed by
    ``server_id``. Batched: one resolver pass + one person lookup."""
    from app.domains.cmdb.service import (  # noqa: PLC0415 — sibling module
        resolve_inherited_server_owners,
    )
    from app.identity.person_models import IdentityPerson

    by_id = {str(s.id): s for s in servers}
    # Subscription-inherited owner only matters for servers with no explicit
    # owner; resolve those in one batched pass.
    unowned = [s for s in servers if not s.owner_person_id]
    inherited = (
        await resolve_inherited_server_owners(db, unowned) if unowned else {}
    )

    # One person lookup covering every owner we need to label (stored +
    # inherited).
    pids: set[str] = {
        str(s.owner_person_id) for s in servers if s.owner_person_id
    }
    pids |= {
        str(v["person_id"]) for v in inherited.values() if v.get("person_id")
    }
    people: dict[str, IdentityPerson] = {}
    if pids:
        prow = await db.execute(
            select(IdentityPerson).where(IdentityPerson.id.in_(pids))
        )
        people = {str(p.id): p for p in prow.scalars()}

    def _apply(
        item: dict[str, Any], person_id: Any,
        *, scope_info: dict[str, str | None] | None = None,
    ) -> None:
        p = people.get(str(person_id)) if person_id else None
        item["effective_owner_person_id"] = str(person_id) if person_id else None
        item["effective_owner_label"] = p.display_name if p else None
        item["effective_owner_employee_no"] = p.employee_no if p else None
        item["effective_owner_employment_status"] = (
            p.employment_status if p else None
        )
        item["owner_inherited"] = scope_info is not None
        item["inherited_from_scope_id"] = (
            scope_info["scope_id"] if scope_info else None
        )
        item["inherited_from_scope_name"] = (
            scope_info["scope_name"] if scope_info else None
        )

    for item in items:
        srv = by_id.get(str(item.get("server_id")))
        if srv is None:
            continue
        if srv.owner_person_id:
            _apply(item, srv.owner_person_id)
            continue
        info = inherited.get(str(srv.id))
        if info and info.get("person_id"):
            _apply(item, info["person_id"], scope_info=info)
        else:
            _apply(item, None)


async def get_instance_detail(
    db: AsyncSession, server_id: str,
) -> dict[str, Any] | None:
    stmt = (
        select(
            CmdbServerEfficiency, CmdbServer, Integration,
            CmdbServerEfficiencyTriage,
        )
        .join(CmdbServer, CmdbServer.id == CmdbServerEfficiency.server_id)
        .join(
            Integration,
            Integration.id == CmdbServerEfficiency.integration_id,
            isouter=True,
        )
        .join(
            CmdbServerEfficiencyTriage,
            CmdbServerEfficiencyTriage.server_id
            == CmdbServerEfficiency.server_id,
            isouter=True,
        )
        .where(
            CmdbServerEfficiency.server_id == server_id,
            # A soft-deleted / decommissioned server must not be returned
            # (or remain triageable) — every list/summary path already
            # filters this via ``_base_conditions``; the single-server
            # detail read was the one place that didn't.
            CmdbServer.deleted_at.is_(None),
        )
    )
    row = (await db.execute(stmt)).first()
    if row is None:
        return None
    eff, srv, integ, triage = row
    detail = _instance_row(eff, srv, integ, triage)
    await _augment_inherited_owner_fields(db, [detail], [srv])

    # Bound the sparkline: the previous query returned the server's ENTIRE
    # retained daily-metrics history (grows ~1 row/day forever). Cap to the
    # most-recent 90 rows (desc + reverse) — ample for a sparkline and well
    # above any verdict window — so the per-request payload can't grow
    # unbounded.
    series_stmt = (
        select(CmdbServerMetricsDaily)
        .where(CmdbServerMetricsDaily.server_id == server_id)
        .order_by(CmdbServerMetricsDaily.metric_date.desc())
        .limit(90)
    )
    series_rows = list(
        reversed((await db.execute(series_stmt)).scalars().all())
    )
    detail["metrics_series"] = [
        {
            "date": r.metric_date.isoformat(),
            "cpu_avg": _f(r.cpu_pct_avg),
            "cpu_p95": _f(r.cpu_pct_p95),
            "cpu_max": _f(r.cpu_pct_max),
            "mem_avg": _f(r.mem_pct_avg),
            "disk_used_avg": _f(r.disk_used_pct_avg),
            "net_in_avg": r.net_in_bytes_avg,
            "net_out_avg": r.net_out_bytes_avg,
            "sample_count": r.sample_count,
        }
        for r in series_rows
    ]
    return detail


async def _inherited_dept_chains(
    db: AsyncSession, server_ids: list[str],
) -> dict[str, list[str]]:
    """For servers whose stored owner produced no department, resolve the
    subscription-inherited owner (live, same path the server list uses)
    and return ``{server_id: dept_chain}`` for the ones that resolve to a
    person with a primary org membership."""
    from app.domains.cmdb.service import (  # noqa: PLC0415 — sibling module
        resolve_inherited_server_owners,
    )

    servers = (
        await db.execute(
            select(CmdbServer).where(
                CmdbServer.id.in_(server_ids),
                CmdbServer.deleted_at.is_(None),
            ),
        )
    ).scalars().all()
    inherited = await resolve_inherited_server_owners(db, list(servers))
    person_ids = {
        info["person_id"] for info in inherited.values() if info.get("person_id")
    }
    if not person_ids:
        return {}
    om = IdentityOrgMembership
    mrows = await db.execute(
        select(om.person_id, om.org_path, om.org_node_name).where(
            om.person_id.in_(person_ids),
            om.is_primary.is_(True),
            om.deleted_at.is_(None),
        ),
    )
    person_chain: dict[str, list[str]] = {}
    for pid, org_path, org_node_name in mrows.all():
        # First primary membership wins — same dedup stance as the join.
        person_chain.setdefault(str(pid), _dept_chain(org_path, org_node_name))
    out: dict[str, list[str]] = {}
    for sid, info in inherited.items():
        chain = person_chain.get(info.get("person_id") or "")
        if chain:
            out[sid] = chain
    return out


async def savings_by_department(
    db: AsyncSession,
    *,
    visibility: OwnerVisibility | None = None,
    integration_id: str | None = None,
    limit: int = 20,
) -> list[dict[str, Any]]:
    """Roll up estimated savings + attributed cost + counts by the owning
    person's department. Each ancestor in the department chain is bumped so
    a parent shows subtree totals (mirrors IdentityPerson.department_chain).

    Owner resolution matches what the server list displays (PR #2023):
    the stored ``owner_person_id`` wins (an explicit transfer re-homes the
    machine immediately); a server with NO stored owner inherits its cloud
    subscription's owner, live — so a subscription with an owner never
    dumps its whole fleet into "unassigned". Only servers with neither
    (no owner, no scope attribution, or a scope without an owner) and
    owners without a primary org membership bucket under ``unassigned``.
    Sorted by savings desc, top-N."""
    conds = _base_conditions(visibility, integration_id)
    om = IdentityOrgMembership
    stmt = (
        select(
            CmdbServerEfficiency.server_id,
            CmdbServerEfficiency.estimated_monthly_savings_usd,
            CmdbServerEfficiency.monthly_cost_usd,
            CmdbServerEfficiency.verdict,
            om.org_path,
            om.org_node_name,
        )
        .join(CmdbServer, CmdbServer.id == CmdbServerEfficiency.server_id)
        .join(
            om,
            and_(
                om.person_id == CmdbServer.owner_person_id,
                om.is_primary.is_(True),
                om.deleted_at.is_(None),
            ),
            isouter=True,
        )
        .where(*conds)
    )
    rows = (await db.execute(stmt)).all()

    agg: dict[str, dict[str, Any]] = {}

    def _bump(dept: str, savings, cost, verdict) -> None:
        b = agg.setdefault(
            dept,
            {"department": dept, "servers": 0, "savings": Decimal("0"),
             "cost": Decimal("0"), "idle": 0, "oversized": 0},
        )
        b["servers"] += 1
        if savings is not None:
            b["savings"] += Decimal(str(savings))
        if cost is not None:
            b["cost"] += Decimal(str(cost))
        if verdict == "idle":
            b["idle"] += 1
        elif verdict == "oversized":
            b["oversized"] += 1

    # A person may (rarely) have >1 primary membership row → the LEFT JOIN
    # would emit the same server twice and double-count it. Dedup per
    # server_id (first row wins); within-chain ancestor bumping stays.
    seen_servers: set[str] = set()
    entries: list[tuple[str, Any, Any, Any, list[str]]] = []
    chainless: list[str] = []
    for server_id, savings, cost, verdict, org_path, org_node_name in rows:
        sid = str(server_id)
        if sid in seen_servers:
            continue
        seen_servers.add(sid)
        chain = _dept_chain(org_path, org_node_name)
        if not chain:
            chainless.append(sid)
        entries.append((sid, savings, cost, verdict, chain))

    # Second pass for servers the join couldn't attribute: resolve the
    # owner they inherit from their cloud subscription and use THAT
    # person's department. Servers with a stored owner (who merely lacks
    # a membership row) are skipped inside the resolver — an explicit
    # assignment always wins over inheritance.
    inherited_chains = (
        await _inherited_dept_chains(db, chainless) if chainless else {}
    )
    for sid, savings, cost, verdict, chain in entries:
        if not chain:
            chain = inherited_chains.get(sid, [])
        if not chain:
            _bump(_UNASSIGNED, savings, cost, verdict)
            continue
        # Bump every ancestor so a parent shows subtree totals.
        for dept in chain:
            _bump(dept, savings, cost, verdict)

    out = [
        {
            "department": b["department"],
            "servers": b["servers"],
            "idle": b["idle"],
            "oversized": b["oversized"],
            "estimated_monthly_savings_usd": _f(b["savings"]),
            "attributed_monthly_cost_usd": _f(b["cost"]),
        }
        for b in agg.values()
    ]
    out.sort(key=lambda x: (x["estimated_monthly_savings_usd"] or 0), reverse=True)
    return out[:limit]


# ─── Subscription-scoped reads + owner triage ──────────────────────

async def get_scope(
    db: AsyncSession, scope_id: str,
) -> CmdbCloudScope | None:
    """Fetch a cloud subscription by id (returns None on bad/absent id)."""
    # No module logger here; best_effort falls back to its own.
    async with best_effort(
        db, label="efficiency_service: get_scope",
    ) as be_813_step:
        stmt = select(CmdbCloudScope).where(CmdbCloudScope.id == scope_id)
        return (await db.execute(stmt)).scalar_one_or_none()
    if not be_813_step.ok:
        return None


_TRIAGE_ACTION_STATUS = {
    "keep": "kept",
    "acknowledge": "acknowledged",
    "snooze": "snoozed",
}


async def submit_triage(
    db: AsyncSession,
    *,
    server_id: str,
    action: str,
    reason: str | None,
    snooze_days: int | None,
    actor_person_id: str | None,
) -> dict[str, Any] | None:
    """Record (upsert) an owner's non-destructive triage decision on a
    server's current recommendation. Returns the refreshed instance row,
    or None if the server has no efficiency profile yet. NEVER touches the
    cloud — this only records intent + clears the item from the queue."""
    status = _TRIAGE_ACTION_STATUS.get(action)
    if status is None:
        raise ValueError(f"unknown triage action {action!r}")

    eff = (
        await db.execute(
            select(CmdbServerEfficiency).where(
                CmdbServerEfficiency.server_id == server_id,
            )
        )
    ).scalar_one_or_none()
    if eff is None:
        return None

    now = datetime.now(UTC)
    snooze_until = None
    if status == EFFICIENCY_TRIAGE_SNOOZED:
        snooze_until = now + timedelta(days=snooze_days or 30)

    triage = (
        await db.execute(
            select(CmdbServerEfficiencyTriage).where(
                CmdbServerEfficiencyTriage.server_id == server_id,
            )
        )
    ).scalar_one_or_none()
    if triage is None:
        triage = CmdbServerEfficiencyTriage(server_id=server_id)
        db.add(triage)
    triage.status = status
    triage.reason = reason
    triage.verdict_at_decision = eff.verdict
    triage.decided_by_person_id = actor_person_id
    triage.decided_at = now
    triage.snooze_until = snooze_until
    await db.flush()

    return await get_instance_detail(db, server_id)