"""Leaver cloud-IAM residue — detect access a departed colleague still holds.

The gap this closes
===================

Offboarding revokes a leaver's ``cmdb_cloud_access_grants`` rows and asks
the driver to remove the cloud-side assignment. When that driver call
*silently under-delivers*, nothing ever notices:

* the Helix grant row is already ``revoked``, and every existing sweep
  (``cmdb.leaver_grant_sweep``, the drift sweep, the expiry sweep) only
  looks at ``active`` / ``pending`` rows — a revoked row is never
  revisited;
* the cloud-side principal is therefore invisible to Helix except as a
  ``cmdb_iam_users`` row that nobody reads for this purpose.

Prod proof (2026-08-12, Tencent tenant 100002610019): 褚杰 ``chujie``
left on 2026-06-01 and Helix revoked all four of his CAM grants that
night — the driver's ``DetachUserPolicy`` was a no-op because the
policies were **group-inherited** (fixed 2026-06-02, one day too late),
so 2.5 months later the sub-user still carried all four. 唐彭
``tangpeng`` the same. Both were found by hand, by reading the console.

So the invariant this module enforces is deliberately *not* stated in
terms of Helix grants:

    A person who has left must hold **no cloud-side IAM policy or group
    membership**, whatever Helix's own records claim.

``cmdb_iam_users`` is the right evidence base because it is a pure mirror
of the provider — refreshed every 4 h by the IAM capability sync, with no
opinion about what Helix thinks the access should be. Comparing it
against HR's leaver set is the only check that can catch a revoke which
reported success and did nothing.

Design notes
============

**Detection is separate from action.** The sweep only records and
reports; deleting a cloud account is an irreversible operation and stays
behind an explicit human confirmation in the admin UI (which reuses the
existing ``POST /cmdb/iam-users/{id}/cleanup`` chain). This is also why
the sweep itself is not gated on the prod write lock while the UI action
is — see ``prod_write_lock``'s "system automation is not gated" carve-out
and the fact that the sweep performs no cloud writes at all.

**Stale mirrors are reported, never acted on.** If the IAM sync for an
integration has not run recently, its ``attached_policies`` may describe a
world that no longer exists. Such rows are still surfaced (an IAM sync
that stopped running is itself worth seeing) but flagged
``mirror_stale=True`` so nobody hits "revoke" against a guess.

**Self-service policies are not residue.** AWS attaches
``IAMUserChangePassword`` & friends to every console user; counting them
produced 60+ false alerts per leaver in the ``_derive_grants_from_iam_users``
path before it filtered them. Reuse the same predicate.
"""
from __future__ import annotations

import uuid
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
from typing import Any

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

from app.core.database import async_session_factory
from app.core.exceptions import format_exception_for_log
from app.core.safe_logging import get_safe_logger
from app.domains.cmdb.models import (
    RESIDUE_RESOLUTION_CLEARED,
    RESIDUE_RESOLUTION_NOT_A_LEAVER,
    RESIDUE_RESOLUTION_RECLASSIFIED,
    RESIDUE_STATUS_DISMISSED,
    RESIDUE_STATUS_OPEN,
    RESIDUE_STATUS_RESOLVED,
    CmdbIamUser,
    CmdbLeaverIamResidue,
)
from app.identity.leaver_notice import leaver_last_working_day_passed
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration

logger = get_safe_logger(__name__)


def _naive(value: datetime) -> datetime:
    """Drop tzinfo for arithmetic against a naive utcnow.

    These columns are declared ``DateTime(timezone=True)`` and written as
    aware values, but MySQL/TiDB DATETIME strips the zone on round-trip —
    so a row read back is naive while one still live in the session's
    identity map (``expire_on_commit=False``) is not. Mixing the two in a
    subtraction raises TypeError and would take the whole sweep down.
    ``_residue_row_payload`` already normalises the same columns; do it
    here too rather than depending on which path loaded the row.
    """
    return value.replace(tzinfo=None) if value.tzinfo else value

# The mirror is refreshed every 4 h by ``cmdb.sync.<integration>.iam``.
# Three missed ticks is the point where "the data might be a little old"
# becomes "the IAM sync is broken and I should not trust this row".
MIRROR_STALE_AFTER = timedelta(hours=13)

# Blast-radius guard, same shape as ``_LEAVER_SWEEP_MAX_PER_RUN``. A bad
# HRIS sync that mass-flips people to ``left`` must not produce a
# thousand-row alert that nobody can triage; the sweep raises instead so
# the run lands ``failed`` and a human looks.
MAX_RESIDUE_ROWS_PER_RUN = 200


@dataclass
class ResidueFinding:
    """One departed person × one cloud IAM user that still has access."""

    iam_user_id: str
    integration_id: str
    provider: str
    integration_name: str
    user_name: str
    person_id: str
    person_display_name: str
    person_work_email: str | None
    termination_date: Any | None
    policy_names: list[str] = field(default_factory=list)
    group_names: list[str] = field(default_factory=list)
    mirror_synced_at: datetime | None = None
    mirror_stale: bool = False
    # How the IAM row was tied to the person — ``owner`` (the row's
    # ``owner_person_id`` FK) or ``email`` (resolved from user_name/tags).
    # Surfaced so an admin can tell a confident match from an inferred one
    # before authorising an irreversible delete.
    match_basis: str = "owner"

    @property
    def policy_count(self) -> int:
        return len(self.policy_names)

    @property
    def group_count(self) -> int:
        return len(self.group_names)


def _residual_policy_names(row: CmdbIamUser) -> list[str]:
    """Attached policies that actually represent access.

    Filters the AWS console's self-service trio. Accepts both the ``name``
    and ``arn`` shapes the IAM sync persists, since Tencent CAM rows carry
    no ARN and RAM custom policies carry no name.
    """
    from app.domains.cmdb.lifecycle_service import _is_self_service_managed_policy

    out: list[str] = []
    for entry in (row.attached_policies or []):
        if not isinstance(entry, dict):
            continue
        name = str(entry.get("name") or "").strip()
        arn = str(entry.get("arn") or "").strip()
        if _is_self_service_managed_policy(arn, name):
            continue
        label = name or arn
        if label:
            out.append(label)
    return out


def _group_names(row: CmdbIamUser) -> list[str]:
    """Group memberships — residue in their own right.

    Group membership is *why* the 2026-06-01 revoke silently failed: the
    policies were inherited, so detaching them individually did nothing
    and the person kept every permission the group carried. A leaver
    still sitting in a CAM/RAM group is residue even when their directly
    attached policy list is empty.
    """
    return [str(g).strip() for g in (row.groups or []) if str(g).strip()]


async def _leavers_past_last_working_day(
    session: AsyncSession,
) -> dict[str, IdentityPerson]:
    """Everyone HR considers gone, keyed by person id.

    Includes ``leaver_pending`` whose last working day has already passed
    — the window ``cmdb.leaver_grant_sweep`` misses because it filters on
    ``employment_status == 'left'`` alone, and HRIS can lag a day or two
    behind the actual departure. ``leaver_last_working_day_passed`` is
    timezone-correct per person and returns False when
    ``termination_date`` is NULL, so we never act early on someone whose
    date we do not know.
    """
    rows = (await session.execute(
        select(IdentityPerson).where(
            IdentityPerson.employment_status.in_(("left", "leaver_pending")),
            IdentityPerson.deleted_at.is_(None),
        )
    )).scalars().all()
    return {
        str(p.id): p for p in rows
        if p.employment_status == "left" or leaver_last_working_day_passed(p)
    }


async def _iam_rows_for_leavers(
    session: AsyncSession,
    leavers: dict[str, IdentityPerson],
) -> list[tuple[CmdbIamUser, IdentityPerson, str]]:
    """Live IAM mirror rows belonging to a departed person.

    Two match bases, in confidence order:

    1. ``owner_person_id`` — bound by the IAM sync's own resolution.
    2. work_email — for rows the owner-resolver could not bind (it only
       binds **active** people, so a row whose owner left before it was
       ever bound stays NULL forever; that population is exactly the one
       this sweep must not miss).

    Deleted / already-revoked rows are excluded: ``revoked_at`` means an
    admin ran the cleanup chain, and the next IAM sync will drop the row
    entirely when the provider confirms the user is gone.
    """
    from app.domains.cmdb.lifecycle_service import _email_from_iam_user

    rows = (await session.execute(
        select(CmdbIamUser).where(
            CmdbIamUser.deleted_at.is_(None),
            CmdbIamUser.revoked_at.is_(None),
            # An account a human has marked as a robot is not a leaver's
            # residue even when its name matches one. Service accounts
            # outlive the person who created them by design; leaving them
            # in this queue is how a queue stops being read.
            or_(
                CmdbIamUser.account_kind.is_(None),
                CmdbIamUser.account_kind != "service",
            ),
        )
    )).scalars().all()

    email_to_person: dict[str, IdentityPerson] = {}
    for person in leavers.values():
        email = (person.work_email or "").strip().lower()
        if email:
            email_to_person[email] = person

    out: list[tuple[CmdbIamUser, IdentityPerson, str]] = []
    for row in rows:
        owner_id = str(row.owner_person_id or "")
        owned_by: IdentityPerson | None = (
            leavers.get(owner_id) if owner_id else None
        )
        if owned_by is not None:
            out.append((row, owned_by, "owner"))
            continue
        if owner_id:
            # Bound to somebody who is still employed — not residue.
            continue
        resolved_email = _email_from_iam_user(row.user_name or "", row.tags)
        if not resolved_email:
            continue
        by_email = email_to_person.get(resolved_email)
        if by_email is not None:
            out.append((row, by_email, "email"))
    return out


async def detect_leaver_iam_residue(
    session: AsyncSession,
) -> list[ResidueFinding]:
    """Pure detection — no writes, no cloud calls. Safe to call from a
    read-only request path (the admin page's "rescan" button does)."""
    leavers = await _leavers_past_last_working_day(session)
    if not leavers:
        return []
    pairs = await _iam_rows_for_leavers(session, leavers)
    if not pairs:
        return []

    integration_ids = {str(row.integration_id) for row, _, _ in pairs}
    integrations = {
        str(i.id): i for i in (await session.execute(
            select(Integration).where(Integration.id.in_([
                uuid.UUID(i) for i in integration_ids
            ]))
        )).scalars().all()
    }

    now = datetime.now(UTC).replace(tzinfo=None)
    findings: list[ResidueFinding] = []
    for row, person, basis in pairs:
        policies = _residual_policy_names(row)
        groups = _group_names(row)
        if not policies and not groups:
            continue
        integration = integrations.get(str(row.integration_id))
        synced_at = row.last_synced_at
        stale = (
            synced_at is None
            or (now - _naive(synced_at)) > MIRROR_STALE_AFTER
        )
        findings.append(ResidueFinding(
            iam_user_id=str(row.id),
            integration_id=str(row.integration_id),
            provider=(integration.provider if integration else "") or "",
            integration_name=(integration.name if integration else "") or "",
            user_name=row.user_name or "",
            person_id=str(person.id),
            person_display_name=person.display_name or "",
            person_work_email=person.work_email,
            termination_date=person.termination_date,
            policy_names=policies,
            group_names=groups,
            mirror_synced_at=synced_at,
            mirror_stale=stale,
            match_basis=basis,
        ))
    # Longest-standing first: "how long has a departed colleague kept
    # production access" is the number that matters, and it is what the
    # alert's first line should name.
    findings.sort(key=lambda f: (f.termination_date or datetime.max.date()))
    return findings


@dataclass
class ResidueSyncResult:
    detected: int = 0
    opened: int = 0
    still_open: int = 0
    # A dismissed row whose residue grew since the dismissal. Counted
    # apart from ``opened`` so the summary distinguishes "new problem"
    # from "a problem we thought we had accepted just got worse".
    reopened: int = 0
    auto_resolved: int = 0
    newly_opened_ids: list[str] = field(default_factory=list)


async def reconcile_residue_rows(
    session: AsyncSession,
    findings: list[ResidueFinding],
) -> ResidueSyncResult:
    """Fold detection output into ``cmdb_leaver_iam_residue``.

    Row lifecycle:

    * new finding → INSERT ``open`` with ``first_detected_at`` (the clock
      the UI ages against);
    * finding still present → refresh the snapshot + ``last_detected_at``;
    * previously-open row no longer detected → ``resolved`` with
      ``resolution='cleared'``. Someone removed the access (through Helix
      or in the console) and the mirror now agrees.

    ``dismissed`` rows are deliberately NOT re-opened by a later
    detection of the same pair: an admin has said "this is expected"
    (e.g. a shared account being re-owned), and re-nagging every night is
    how a governance queue becomes noise everyone ignores. The dismissal
    carries a reason and an actor, and the row still shows in the UI
    under the 已忽略 filter.
    """
    result = ResidueSyncResult(detected=len(findings))
    existing = {
        (str(r.iam_user_id), str(r.person_id)): r
        for r in (await session.execute(
            select(CmdbLeaverIamResidue).where(
                CmdbLeaverIamResidue.status != RESIDUE_STATUS_RESOLVED,
            )
        )).scalars().all()
    }
    now = datetime.now(UTC).replace(tzinfo=None)
    seen: set[tuple[str, str]] = set()

    for f in findings:
        key = (f.iam_user_id, f.person_id)
        seen.add(key)
        row = existing.get(key)
        snapshot = {
            "policy_names": f.policy_names,
            "group_names": f.group_names,
            "match_basis": f.match_basis,
            "mirror_stale": f.mirror_stale,
            "user_name": f.user_name,
            "provider": f.provider,
            "integration_name": f.integration_name,
        }
        if row is None:
            row = CmdbLeaverIamResidue(
                id=uuid.uuid4(),
                iam_user_id=f.iam_user_id,
                person_id=f.person_id,
                integration_id=f.integration_id,
                user_name=f.user_name,
                provider=f.provider,
                policy_count=f.policy_count,
                group_count=f.group_count,
                detail=snapshot,
                status=RESIDUE_STATUS_OPEN,
                first_detected_at=now,
                last_detected_at=now,
            )
            session.add(row)
            result.opened += 1
            result.newly_opened_ids.append(str(row.id))
        else:
            # A dismissal is a judgement about the residue the admin was
            # SHOWN, not a permanent exemption for the account. If the
            # access has grown since — a new policy, a new group — the row
            # comes back so the alert can fire again. Otherwise dismissing
            # a leaver's read-only account would silently cover it later
            # being granted AdministratorAccess.
            grew = (
                f.policy_count > (row.policy_count or 0)
                or f.group_count > (row.group_count or 0)
            )
            reopened = False
            if row.status == RESIDUE_STATUS_DISMISSED and grew:
                row.status = RESIDUE_STATUS_OPEN
                row.resolution = None
                row.resolved_at = None
                row.resolved_by_person_id = None
                # The dismissal note is kept: it is why somebody thought
                # this was fine, and that context matters when they look
                # at it again.
                reopened = True
            row.policy_count = f.policy_count
            row.group_count = f.group_count
            row.detail = snapshot
            row.user_name = f.user_name
            row.last_detected_at = now
            if reopened:
                result.reopened += 1
                result.newly_opened_ids.append(str(row.id))
            elif row.status == RESIDUE_STATUS_OPEN:
                result.still_open += 1

    # A row can leave the detection set for reasons OTHER than the access
    # being removed, and closing all of them as ``cleared`` would assert
    # something nobody verified. Ask why before writing the reason.
    out_of_set = [(k, r) for k, r in existing.items() if k not in seen]
    reasons = await _closure_reasons(session, [r for _, r in out_of_set])
    for key, row in out_of_set:
        row.status = RESIDUE_STATUS_RESOLVED
        row.resolution = reasons.get(str(row.id), RESIDUE_RESOLUTION_CLEARED)
        row.resolved_at = now
        result.auto_resolved += 1

    await session.commit()
    return result


async def _closure_reasons(
    session: AsyncSession, rows: list[CmdbLeaverIamResidue],
) -> dict[str, str]:
    """Why did each of these rows stop being detected?

    ``cleared`` is a claim that the cloud-side access is gone, and it is
    only true for one of the three ways a row can leave the detection
    set. The other two must not borrow it:

    * the IAM user was declared a service account — the policies are
      still attached, we simply stopped calling it a leaver's residue;
    * the person is no longer a leaver (rehire, or an HRIS sync flipping
      them back to active) — again, the access never moved.

    Anything else — the mirror row is gone, revoked, or its policy and
    group lists are now empty — really is cleared.
    """
    if not rows:
        return {}
    iam_ids = {str(r.iam_user_id) for r in rows}
    person_ids = {str(r.person_id) for r in rows}

    service_ids = {
        str(i) for (i,) in (await session.execute(
            select(CmdbIamUser.id).where(
                CmdbIamUser.id.in_(iam_ids),
                CmdbIamUser.account_kind == "service",
            )
        )).all()
    }
    rehired_ids = {
        str(i) for (i,) in (await session.execute(
            select(IdentityPerson.id).where(
                IdentityPerson.id.in_(person_ids),
                IdentityPerson.employment_status == "active",
            )
        )).all()
    }

    out: dict[str, str] = {}
    for row in rows:
        if str(row.iam_user_id) in service_ids:
            out[str(row.id)] = RESIDUE_RESOLUTION_RECLASSIFIED
        elif str(row.person_id) in rehired_ids:
            out[str(row.id)] = RESIDUE_RESOLUTION_NOT_A_LEAVER
        else:
            out[str(row.id)] = RESIDUE_RESOLUTION_CLEARED
    return out


async def sweep_leaver_iam_residue() -> dict[str, Any]:
    """Cron entry point. Detect → persist → alert IT admins on new rows.

    Performs **zero cloud writes**; the remediation lever is the admin
    UI's confirmed cleanup action.
    """
    async with async_session_factory() as session:
        findings = await detect_leaver_iam_residue(session)
        if len(findings) > MAX_RESIDUE_ROWS_PER_RUN:
            raise RuntimeError(
                f"cmdb.leaver_iam_residue_sweep found {len(findings)} "
                f"residue rows (cap {MAX_RESIDUE_ROWS_PER_RUN}). That is "
                "far past a normal offboarding backlog and usually means "
                "an HRIS sync mass-flipped people to 'left' or an IAM "
                "mirror was rebuilt. Refusing to alert on a list nobody "
                "can triage — verify identity_persons.employment_status "
                "first, then re-run."
            )
        result = await reconcile_residue_rows(session, findings)

        notified = 0
        if result.newly_opened_ids:
            try:
                notified = await _alert_admins(
                    session, findings, result.newly_opened_ids,
                )
            except Exception as exc:  # noqa: BLE001
                # An alerting failure must not discard the detection work
                # already committed above, but it must still be visible.
                logger.warning(
                    "leaver_iam_residue: admin alert failed: %s",
                    format_exception_for_log(exc),
                )
                return {
                    "detected": result.detected,
                    "opened": result.opened,
                    "still_open": result.still_open,
                    "auto_resolved": result.auto_resolved,
                    "admins_notified": 0,
                    "errors": [
                        f"admin alert failed: {format_exception_for_log(exc)}",
                    ],
                }

    return {
        "detected": result.detected,
        "opened": result.opened,
        "still_open": result.still_open,
        "auto_resolved": result.auto_resolved,
        "admins_notified": notified,
        "errors": [],
    }


def _residue_age_days(f: ResidueFinding, now: datetime) -> int | None:
    if f.termination_date is None:
        return None
    term = f.termination_date
    if isinstance(term, datetime):
        term_date = term.date()
    else:
        term_date = term
    return max(0, (now.date() - term_date).days)


async def _alert_admins(
    session: AsyncSession,
    findings: list[ResidueFinding],
    newly_opened_ids: list[str],
) -> int:
    """Actionable alert to IT admins naming the people and the accounts.

    Only fires for **newly opened** rows. A still-open row from last night
    is already sitting in the governance queue with its age counter
    running; re-sending it every 24 h trains admins to mute the channel,
    which is the failure mode that let the original residue sit for 2.5
    months.
    """
    from app.notifications.service import (
        CardAction,
        CardStat,
        LocalizedText,
        NotificationEnvelope,
        NotificationService,
    )
    from app.scheduled_jobs.failure_notifier import list_admin_recipients

    # Map the persisted new row ids back to findings via
    # (iam_user_id, person_id), the row's natural key.
    new_rows = (await session.execute(
        select(CmdbLeaverIamResidue).where(
            CmdbLeaverIamResidue.id.in_([
                uuid.UUID(i) for i in newly_opened_ids
            ]),
        )
    )).scalars().all()
    new_keys = {(str(r.iam_user_id), str(r.person_id)) for r in new_rows}
    new_findings = [
        f for f in findings if (f.iam_user_id, f.person_id) in new_keys
    ]
    if not new_findings:
        return 0

    now = datetime.now(UTC).replace(tzinfo=None)
    recipients = await list_admin_recipients(session)
    if not recipients:
        logger.warning(
            "leaver_iam_residue: %d new residue row(s) but no admin "
            "recipient resolved", len(new_findings),
        )
        return 0

    lines: list[str] = []
    for f in new_findings[:20]:
        age = _residue_age_days(f, now)
        age_txt = f"，离职 {age} 天" if age is not None else ""
        bits: list[str] = []
        if f.policy_count:
            bits.append(f"{f.policy_count} 条策略")
        if f.group_count:
            bits.append(f"{f.group_count} 个用户组")
        lines.append(
            f"· {f.person_display_name or f.person_work_email or f.person_id}"
            f"（{f.integration_name or f.provider}·{f.user_name}）仍持有 "
            f"{'、'.join(bits)}{age_txt}"
        )
    if len(new_findings) > 20:
        lines.append(f"…另有 {len(new_findings) - 20} 条未列出")

    oldest = max(
        (a for a in (_residue_age_days(f, now) for f in new_findings)
         if a is not None),
        default=None,
    )
    # Card CHROME (stat labels, subtitle, button) does NOT go through the
    # template layer — a bare Chinese string here reaches an English or
    # Japanese admin untranslated, next to a body that localized fine.
    stats = [
        CardStat(
            label=LocalizedText(zh="新发现", en="New", ja="新規"),
            value=str(len(new_findings)),
        ),
        CardStat(
            label=LocalizedText(
                zh="涉及离职者", en="People", ja="対象者",
            ),
            value=str(len({f.person_id for f in new_findings})),
        ),
    ]
    if oldest is not None:
        stats.append(CardStat(
            label=LocalizedText(zh="最久", en="Oldest", ja="最長"),
            value=LocalizedText(
                zh=f"{oldest} 天", en=f"{oldest} days", ja=f"{oldest} 日",
            ),
        ))

    # The framing prose lives in the template (zh / en / ja); ``context``
    # carries only the list. Passing a fully-composed Chinese body would
    # embed it inside the English and Japanese renderings — the exact
    # defect the ``compliance_incident_alert`` template comment records.
    residue_list = "\n".join(lines)

    service = NotificationService(session)
    sent = 0
    for admin in recipients:
        if admin.deleted_at is not None:
            continue
        if (admin.employment_status or "").lower() != "active":
            continue
        try:
            await service.dispatch(NotificationEnvelope(
                recipient=admin,
                title=f"离职人员云权限残留 · 新发现 {len(new_findings)} 条",
                body_text=residue_list,
                channels=["feishu", "email"],
                template_key="compliance_leaver_iam_residue",
                context={
                    "count": len(new_findings),
                    "oldest_days": oldest,
                    "residue_list": residue_list,
                },
                url="/admin/cloud/leaver-residue",
                related_resource_type="cmdb_leaver_iam_residue",
                card_subtitle=LocalizedText(
                    zh="云端镜像与 HR 离职名单的每日对账结果",
                    en="Daily reconciliation of the cloud IAM mirror "
                       "against the HR leaver list",
                    ja="クラウド IAM ミラーと HR 退職者リストの日次突合",
                ),
                card_stats=stats,
                card_actions=[CardAction(
                    label=LocalizedText(
                        zh="去处理", en="Review residue", ja="対応する",
                    ),
                    url="/admin/cloud/leaver-residue",
                    style="primary",
                )],
            ))
            sent += 1
        except Exception as exc:  # noqa: BLE001
            logger.warning(
                "leaver_iam_residue: alert to %s failed: %s",
                admin.id, format_exception_for_log(exc),
            )
    return sent


async def summarize_residue(session: AsyncSession) -> dict[str, Any]:
    """Hero-stat payload for the admin page."""
    now = datetime.now(UTC).replace(tzinfo=None)
    open_rows = (await session.execute(
        select(CmdbLeaverIamResidue).where(
            CmdbLeaverIamResidue.status == RESIDUE_STATUS_OPEN,
        )
    )).scalars().all()
    dismissed = (await session.execute(
        select(func.count()).select_from(CmdbLeaverIamResidue).where(
            CmdbLeaverIamResidue.status == RESIDUE_STATUS_DISMISSED,
        )
    )).scalar() or 0
    resolved_30d = (await session.execute(
        select(func.count()).select_from(CmdbLeaverIamResidue).where(
            CmdbLeaverIamResidue.status == RESIDUE_STATUS_RESOLVED,
            CmdbLeaverIamResidue.resolved_at.is_not(None),
            CmdbLeaverIamResidue.resolved_at >= now - timedelta(days=30),
        )
    )).scalar() or 0
    oldest_days = 0
    for row in open_rows:
        if row.first_detected_at is None:
            continue
        oldest_days = max(
            oldest_days, (now - _naive(row.first_detected_at)).days,
        )
    return {
        "open_count": len(open_rows),
        "people_count": len({str(r.person_id) for r in open_rows}),
        "oldest_open_days": oldest_days,
        "dismissed_count": int(dismissed),
        "resolved_30d_count": int(resolved_30d),
    }


__all__ = [
    "MAX_RESIDUE_ROWS_PER_RUN",
    "MIRROR_STALE_AFTER",
    "ResidueFinding",
    "ResidueSyncResult",
    "detect_leaver_iam_residue",
    "reconcile_residue_rows",
    "summarize_residue",
    "sweep_leaver_iam_residue",
]