"""Per-person forensic lifecycle timeline.

Aggregates every load-bearing event that touched a single person, in
time-descending order, so the admin "case file" view can answer "what
exactly happened to this employee across all our systems and when?"
without 7 separate console clicks.

This is distinct from the per-person flow endpoint
(:pyfunc:`app.hr_lifecycle.router.person_flow`) — that one renders the
in-flight onboarding/offboarding *pipeline* nodes (account/sso/email/…
phase progression). This one is a flat append-only timeline of every
event from every domain that ever touched ``person.id``, plus 4 KPIs
and an "evidence pack" ZIP for compliance export.

Sources merged:

* ``hr_lifecycle_events``         (onboard / offboard / role_change)
* ``workflow_step_events``         (every workflow journal step)
* ``provisioning_tasks``           (created + completed)
* ``offboarding_processes``        (per-process scanned/ready/sso/complete)
* ``audit_logs``                   (actor_user_id indexed; resource_id
                                    bound by created_at >= person.created_at
                                    to bound the scan)
* ``software_seat_allocations``    (allocated_at + revoked_at as 2 rows)
* ``cmdb_cloud_access_grants``     (granted_at + revoked_at as 2 rows)
* ``identity_external_identities`` (created_at + deleted_at as 2 rows)

Soft-deleted persons are queryable — we do NOT filter on
``IdentityPerson.deleted_at`` because the whole point of a forensic
view is to read the case file of someone who's already left.

Authorization: caller wraps with ``require_admin_read()`` at the router
layer. No scope-narrowing happens inside the service.
"""
from __future__ import annotations

import logging
from dataclasses import dataclass
from datetime import UTC, date, datetime, timedelta
from typing import Any
from uuid import UUID
from zoneinfo import ZoneInfo

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

from app.audit.models import AuditLog
from app.core.txn import best_effort
from app.core.workflow_journal.models import WorkflowRun, WorkflowStepEvent
from app.domains.asset.models import SoftwareSeatAllocation
from app.domains.cmdb.models import CmdbCloudAccessGrant
from app.domains.hr.models import HrLifecycleEvent
from app.domains.offboarding.models import (
    PROC_COMPLETED as OFFBOARDING_STATUS_COMPLETED,
)
from app.domains.offboarding.models import (
    OffboardingProcess,
)
from app.domains.provisioning.models import ProvisioningTask
from app.identity.person_models import IdentityExternalIdentity, IdentityPerson

logger = logging.getLogger(__name__)


# Per-source cap so a person with very long history doesn't OOM the
# response. The frontend renders 100/page so 500 most-recent per source
# is "deep enough" for forensic use; admins drill into the underlying
# admin page if they need more. The evidence-pack ZIP exports the raw
# unbounded slice through a different path so legal/auditor export isn't
# capped here.
_PER_SOURCE_LIMIT = 500

# 7-dim "current resources held" — same dimensions as the onboarding
# cockpit but adapted to work for ANY person (not just new hires whose
# OnboardingProcess row still exists). Each is a real-state check
# against the underlying tables, not a cached blocker row.
_RESOURCE_DIMS = (
    "account",       # employment_status == 'active'
    "sso",           # has any non-deleted external identity (google/feishu)
    "work_email",    # person.work_email is set
    "feishu",        # has non-deleted feishu external identity
    "seat",          # has any active software_seat_allocations row
    "computer",      # has any AssetItem.assigned_to=person and not disposed
    "vpn",           # has UserPermission row tagged vpn (best-effort)
)


def _ensure_aware(dt: datetime | None) -> datetime | None:
    """Coerce a possibly-naive DATETIME (TiDB/MySQL strips tzinfo on
    round-trip; see memory ``feedback_mysql_datetime_naive_on_fresh_load``)
    to UTC-aware so cross-source rows sort without TypeError."""
    if dt is None:
        return None
    if dt.tzinfo is None:
        return dt.replace(tzinfo=UTC)
    return dt


@dataclass
class _TimelineRow:
    """In-memory row used to build the merged timeline. Converted to a
    plain dict before the response is rendered."""

    time: datetime
    event_class: str
    summary: str
    detail_link: str | None
    status_badge: str | None
    source: str
    detail: dict[str, Any]

    def __post_init__(self) -> None:
        # Coerce on construction so every code path lands aware. The
        # sort step then can't trip on a naive row that slipped through.
        coerced = _ensure_aware(self.time)
        if coerced is None:
            raise ValueError("_TimelineRow.time must not be None")
        self.time = coerced

    def as_dict(self) -> dict[str, Any]:
        return {
            "time": self.time.isoformat(),
            "event_class": self.event_class,
            "summary": self.summary,
            "detail_link": self.detail_link,
            "status_badge": self.status_badge,
            "source": self.source,
            "detail": self.detail,
        }


class ForensicTimelineService:
    """Read-only aggregator. Stateless modulo the DB session."""

    def __init__(self, db: AsyncSession) -> None:
        self.db = db

    async def get_person(self, person_id: UUID) -> IdentityPerson | None:
        """Look up the person row, INCLUDING soft-deleted rows. The
        forensic view explicitly serves cases where the employee is
        already gone (``deleted_at IS NOT NULL``)."""
        return await self.db.scalar(
            select(IdentityPerson).where(IdentityPerson.id == person_id)
        )

    async def build_payload(
        self, person: IdentityPerson, *, per_source_limit: int | None = None,
    ) -> dict[str, Any]:
        """Return the full timeline + KPI payload for ``person``.

        ``per_source_limit`` overrides the default 500-row-per-source
        cap. The evidence-pack endpoint passes a larger value so the
        downloadable ZIP genuinely captures the long-tenure case file
        (the JSON API path stays bounded so the admin browser doesn't
        choke on a 10-year tenure).
        """
        limit = per_source_limit or _PER_SOURCE_LIMIT
        rows: list[_TimelineRow] = []
        truncated: dict[str, bool] = {}
        # Each collector returns (rows, was_truncated) so the response
        # can flag sources where the cap actually clipped data — without
        # that signal, an admin reads "row_count=2500" as comprehensive
        # when in fact the oldest decade is invisible.
        for name, collector in (
            ("hr_lifecycle_events", self._collect_hr_events),
            ("workflow_step_events", self._collect_workflow_steps),
            ("provisioning_tasks", self._collect_provisioning_tasks),
            ("offboarding_processes", self._collect_offboarding),
            ("audit_logs", self._collect_audit_logs),
            ("software_seat_allocations", self._collect_seat_allocations),
            ("cmdb_cloud_access_grants", self._collect_cloud_grants),
            ("identity_external_identities", self._collect_external_identities),
        ):
            collected, was_truncated = await collector(person, limit=limit)
            rows.extend(collected)
            truncated[name] = was_truncated

        rows.sort(key=lambda r: r.time, reverse=True)

        kpis = await self._compute_kpis(person)

        return {
            "person": _person_summary(person),
            "kpis": kpis,
            "timeline": [r.as_dict() for r in rows],
            "row_count": len(rows),
            "per_source_limit": limit,
            "truncated_by_source": truncated,
        }

    # ── Sources ─────────────────────────────────────────────────────

    async def _collect_hr_events(
        self, person: IdentityPerson, *, limit: int,
    ) -> tuple[list[_TimelineRow], bool]:
        # +1-trick: fetch one past the cap to detect truncation without
        # a separate COUNT() round-trip.
        rows = (
            await self.db.execute(
                select(HrLifecycleEvent)
                .where(HrLifecycleEvent.person_id == person.id)
                .order_by(HrLifecycleEvent.created_at.desc())
                .limit(limit + 1)
            )
        ).scalars().all()
        truncated = len(rows) > limit
        out = [
            _TimelineRow(
                time=r.created_at,
                event_class="hr_lifecycle",
                summary=f"HRIS {r.event_type} 生效日 {r.effective_date.isoformat()}",
                detail_link=f"/admin/lifecycle/person/{person.id}",
                status_badge=r.event_type,
                source="hr_lifecycle_events",
                detail={
                    "event_type": r.event_type,
                    "effective_date": r.effective_date.isoformat(),
                    "extra": r.detail or {},
                },
            )
            for r in rows[:limit]
        ]
        return out, truncated

    async def _collect_workflow_steps(
        self, person: IdentityPerson, *, limit: int,
    ) -> tuple[list[_TimelineRow], bool]:
        # Join WorkflowStepEvent -> WorkflowRun to filter by person_id;
        # WorkflowRun.person_id is indexed (idx_wf_runs_person_started).
        stmt = (
            select(WorkflowStepEvent, WorkflowRun)
            .join(WorkflowRun, WorkflowStepEvent.workflow_run_id == WorkflowRun.id)
            .where(WorkflowRun.person_id == person.id)
            .order_by(WorkflowStepEvent.started_at.desc())
            .limit(limit + 1)
        )
        rows = (await self.db.execute(stmt)).all()
        truncated = len(rows) > limit
        out: list[_TimelineRow] = []
        for step, run in rows[:limit]:
            ts = step.started_at
            out.append(
                _TimelineRow(
                    time=ts,
                    event_class="workflow",
                    summary=f"{run.workflow_type}: {step.step_name} ({step.status})",
                    detail_link=None,
                    status_badge=step.status,
                    source="workflow_step_events",
                    detail={
                        "workflow_type": run.workflow_type,
                        "step_name": step.step_name,
                        "status": step.status,
                        "attempt": step.attempt,
                        "error_class": step.error_class,
                        "error_message": step.error_message,
                    },
                )
            )
        return out, truncated

    async def _collect_provisioning_tasks(
        self, person: IdentityPerson, *, limit: int,
    ) -> tuple[list[_TimelineRow], bool]:
        rows = (
            await self.db.execute(
                select(ProvisioningTask)
                .where(ProvisioningTask.user_id == person.id)
                .order_by(ProvisioningTask.created_at.desc())
                .limit(limit + 1)
            )
        ).scalars().all()
        truncated = len(rows) > limit
        out: list[_TimelineRow] = []
        for t in rows[:limit]:
            target_label = f"{t.target_type}:{t.target_id}" if t.target_id else t.target_type
            out.append(
                _TimelineRow(
                    time=t.created_at,
                    event_class="provisioning",
                    summary=(
                        f"{t.action} {target_label}"
                        f" via {t.source} → {t.status}"
                    ),
                    detail_link=f"/admin/provisioning/tasks?person_id={person.id}",
                    status_badge=t.status,
                    source="provisioning_tasks",
                    detail={
                        "task_id": str(t.id),
                        "action": t.action,
                        "status": t.status,
                        "source": t.source,
                        "target_type": t.target_type,
                        "target_id": str(t.target_id) if t.target_id else None,
                    },
                )
            )
            if t.completed_at is not None:
                out.append(
                    _TimelineRow(
                        time=t.completed_at,
                        event_class="provisioning",
                        summary=f"{t.action} completed → {t.status}",
                        detail_link=f"/admin/provisioning/tasks?person_id={person.id}",
                        status_badge=t.status,
                        source="provisioning_tasks",
                        detail={
                            "task_id": str(t.id),
                            "action": t.action,
                            "status": t.status,
                            "phase": "completed",
                        },
                    )
                )
        return out, truncated

    async def _collect_offboarding(
        self, person: IdentityPerson, *, limit: int,
    ) -> tuple[list[_TimelineRow], bool]:
        rows = (
            await self.db.execute(
                select(OffboardingProcess)
                .where(OffboardingProcess.person_id == person.id)
                .order_by(OffboardingProcess.created_at.desc())
                .limit(limit + 1)
            )
        ).scalars().all()
        truncated = len(rows) > limit
        out: list[_TimelineRow] = []
        for p in rows[:limit]:
            link = f"/admin/offboarding/process/{p.id}"
            transitions: list[tuple[datetime | None, str]] = [
                (p.created_at, "created"),
                (p.scanned_at, "scanned"),
                (p.ready_at, "ready"),
                (p.sso_suspended_at, "sso_suspended"),
                (p.completed_at, "completed"),
            ]
            for when, phase in transitions:
                if when is None:
                    continue
                out.append(
                    _TimelineRow(
                        time=when,
                        event_class="offboarding",
                        summary=f"离职流程 {phase} (leaver_date={p.leaver_date.isoformat()})",
                        detail_link=link,
                        status_badge=phase,
                        source="offboarding_processes",
                        detail={
                            "process_id": str(p.id),
                            "phase": phase,
                            "leaver_date": p.leaver_date.isoformat(),
                            "status": p.status,
                        },
                    )
                )
        return out, truncated

    async def _collect_audit_logs(
        self, person: IdentityPerson, *, limit: int,
    ) -> tuple[list[_TimelineRow], bool]:
        # Two separate index-keyed queries instead of an OR:
        #   (a) actor_user_id == person.id  — uses ix_audit_logs_actor
        #   (b) resource_id == str(person.id) AND created_at >= floor
        #       — resource_id has no index, so the created_at floor
        #         bounds the scan. We subtract a 5-minute grace from
        #         ``person.created_at`` because HRIS bulk sync can write
        #         the ``hr.person_created`` audit row a few ms BEFORE
        #         the identity_persons INSERT lands; without the grace,
        #         the most important first row in the case file would
        #         silently drop.
        actor_uuid = person.id
        resource_str = str(person.id)
        base_floor = _ensure_aware(person.created_at) or datetime(
            2000, 1, 1, tzinfo=UTC,
        )
        time_floor = base_floor - timedelta(minutes=5)

        actor_rows = (
            await self.db.execute(
                select(AuditLog)
                .where(AuditLog.actor_user_id == actor_uuid)
                .order_by(AuditLog.created_at.desc())
                .limit(limit + 1)
            )
        ).scalars().all()
        actor_truncated = len(actor_rows) > limit

        subject_rows = (
            await self.db.execute(
                select(AuditLog)
                .where(
                    AuditLog.created_at >= time_floor,
                    AuditLog.resource_id == resource_str,
                )
                .order_by(AuditLog.created_at.desc())
                .limit(limit + 1)
            )
        ).scalars().all()
        subject_truncated = len(subject_rows) > limit

        # Merge by id (an audit row can appear in both lists when the
        # person acted on themselves — actor == subject) and trim back
        # to ``limit`` rows total after merging.
        seen: dict[str, AuditLog] = {}
        for r in actor_rows[:limit]:
            seen[str(r.id)] = r
        for r in subject_rows[:limit]:
            seen.setdefault(str(r.id), r)
        merged = sorted(
            seen.values(),
            key=lambda r: r.created_at,
            reverse=True,
        )[:limit]

        out: list[_TimelineRow] = []
        for r in merged:
            actor_role = (
                "actor" if r.actor_user_id == actor_uuid else "subject"
            )
            out.append(
                _TimelineRow(
                    time=r.created_at,
                    event_class="audit",
                    summary=f"{r.action} ({actor_role}, {r.status})",
                    detail_link=f"/admin/audit?trace_id={r.trace_id}",
                    status_badge=r.status,
                    source="audit_logs",
                    detail={
                        "trace_id": str(r.trace_id),
                        "action": r.action,
                        "resource_type": r.resource_type,
                        "resource_id": r.resource_id,
                        "status": r.status,
                        "role": actor_role,
                    },
                )
            )
        return out, (actor_truncated or subject_truncated)

    async def _collect_seat_allocations(
        self, person: IdentityPerson, *, limit: int,
    ) -> tuple[list[_TimelineRow], bool]:
        rows = (
            await self.db.execute(
                select(SoftwareSeatAllocation)
                .where(SoftwareSeatAllocation.user_id == person.id)
                .order_by(SoftwareSeatAllocation.allocated_at.desc())
                .limit(limit + 1)
            )
        ).scalars().all()
        truncated = len(rows) > limit
        out: list[_TimelineRow] = []
        for a in rows[:limit]:
            # ``plan_id`` is a real non-null column on SoftwareSeatAllocation
            # — direct access (project rule: no getattr-default on ORM cols,
            # so a future rename fails loud instead of silently rendering None).
            plan_id = a.plan_id
            # ``/admin/software/seat-allocations`` has never been a
            # route — the seat-allocation surface lives under the asset
            # admin tree. Emitting the wrong prefix 404'd every
            # seat row in the forensic timeline (fixed 2026-07-28).
            link = f"/admin/users/{person.id}?tab=seats"
            out.append(
                _TimelineRow(
                    time=a.allocated_at,
                    event_class="seat",
                    summary="软件席位分配",
                    detail_link=link,
                    status_badge="allocated",
                    source="software_seat_allocations",
                    detail={
                        "allocation_id": str(a.id),
                        "plan_id": str(plan_id) if plan_id else None,
                        "phase": "allocated",
                    },
                )
            )
            if a.revoked_at is not None:
                out.append(
                    _TimelineRow(
                        time=a.revoked_at,
                        event_class="seat",
                        summary="软件席位回收",
                        detail_link=link,
                        status_badge="revoked",
                        source="software_seat_allocations",
                        detail={
                            "allocation_id": str(a.id),
                            "plan_id": str(plan_id) if plan_id else None,
                            "phase": "revoked",
                        },
                    )
                )
        return out, truncated

    async def _collect_cloud_grants(
        self, person: IdentityPerson, *, limit: int,
    ) -> tuple[list[_TimelineRow], bool]:
        # CmdbCloudAccessGrant.person_id is String(36) — see
        # reference_asyncmy_uuid_bind: must bind str(uuid_obj).
        person_str = str(person.id)
        rows = (
            await self.db.execute(
                select(CmdbCloudAccessGrant)
                .where(CmdbCloudAccessGrant.person_id == person_str)
                .order_by(CmdbCloudAccessGrant.granted_at.desc())
                .limit(limit + 1)
            )
        ).scalars().all()
        truncated = len(rows) > limit
        out: list[_TimelineRow] = []
        for g in rows[:limit]:
            # ``/admin/cmdb/grants`` has never been a route. The
            # per-person cloud-access view is the user detail page's
            # cloud tab (fixed 2026-07-28 — this 404'd every cloud-grant
            # row in the forensic timeline).
            link = f"/admin/users/{person.id}?tab=cloudGrants"
            out.append(
                _TimelineRow(
                    time=g.granted_at,
                    event_class="cloud_grant",
                    summary=f"云资源授权 {g.role_display_name or g.role_key} → {g.status}",
                    detail_link=link,
                    status_badge=g.status,
                    source="cmdb_cloud_access_grants",
                    detail={
                        "grant_id": str(g.id),
                        "scope_id": g.scope_id,
                        "role_key": g.role_key,
                        "phase": "granted",
                        "status": g.status,
                    },
                )
            )
            if g.revoked_at is not None:
                out.append(
                    _TimelineRow(
                        time=g.revoked_at,
                        event_class="cloud_grant",
                        summary=f"云资源回收 {g.role_display_name or g.role_key}",
                        detail_link=link,
                        status_badge="revoked",
                        source="cmdb_cloud_access_grants",
                        detail={
                            "grant_id": str(g.id),
                            "scope_id": g.scope_id,
                            "role_key": g.role_key,
                            "phase": "revoked",
                            "reason": g.revoke_reason,
                        },
                    )
                )
        return out, truncated

    async def _collect_external_identities(
        self, person: IdentityPerson, *, limit: int,
    ) -> tuple[list[_TimelineRow], bool]:
        rows = (
            await self.db.execute(
                select(IdentityExternalIdentity)
                .where(IdentityExternalIdentity.person_id == person.id)
                .order_by(IdentityExternalIdentity.created_at.desc())
                .limit(limit + 1)
            )
        ).scalars().all()
        truncated = len(rows) > limit
        out: list[_TimelineRow] = []
        for e in rows[:limit]:
            out.append(
                _TimelineRow(
                    time=e.created_at,
                    event_class="identity",
                    summary=f"绑定 {e.provider} 身份 ({e.provider_email or e.provider_subject})",
                    detail_link=None,
                    status_badge=f"{e.provider}_bound",
                    source="identity_external_identities",
                    detail={
                        "provider": e.provider,
                        "subject": e.provider_subject,
                        "provider_email": e.provider_email,
                        "phase": "created",
                    },
                )
            )
            if e.deleted_at is not None:
                out.append(
                    _TimelineRow(
                        time=e.deleted_at,
                        event_class="identity",
                        summary=f"解绑 {e.provider} 身份",
                        detail_link=None,
                        status_badge=f"{e.provider}_unbound",
                        source="identity_external_identities",
                        detail={
                            "provider": e.provider,
                            "subject": e.provider_subject,
                            "phase": "deleted",
                        },
                    )
                )
        return out, truncated

    # ── KPIs ────────────────────────────────────────────────────────

    async def _compute_kpis(
        self, person: IdentityPerson,
    ) -> dict[str, Any]:
        days = _days_at_company(person)
        held = await self._count_current_resources(person)
        request_count = await self._count_lifetime_requests(person)
        in_flight = await self._detect_in_flight(person)
        return {
            "days_at_company": days,
            "current_resources_held": held,
            "lifetime_request_count": request_count,
            "in_flight": in_flight,
        }

    async def _count_current_resources(
        self, person: IdentityPerson,
    ) -> dict[str, Any]:
        """Returns ``{"held": N, "of": 7, "dims": {...bool...}}``.

        Each dimension is a real-state check, not a cached blocker
        row — so this view works for ex-employees whose OnboardingProcess
        row was archived long ago.
        """
        # account
        account_held = (
            person.employment_status == "active"
            and person.deleted_at is None
        )
        # work_email
        work_email_held = bool(person.work_email)
        # sso / feishu — look at identity_external_identities
        ext_rows = (
            await self.db.execute(
                select(IdentityExternalIdentity)
                .where(
                    IdentityExternalIdentity.person_id == person.id,
                    IdentityExternalIdentity.deleted_at.is_(None),
                )
            )
        ).scalars().all()
        providers = {e.provider for e in ext_rows}
        sso_held = any(p in providers for p in ("google", "feishu"))
        feishu_held = "feishu" in providers
        # seat — any active software_seat_allocations row
        seat_count = (
            await self.db.scalar(
                select(SoftwareSeatAllocation.id)
                .where(
                    SoftwareSeatAllocation.user_id == person.id,
                    SoftwareSeatAllocation.revoked_at.is_(None),
                )
                .limit(1)
            )
        )
        seat_held = seat_count is not None
        # computer — any non-disposed AssetItem assigned_to person
        from app.domains.asset.models import AssetItem
        computer_count = (
            await self.db.scalar(
                select(AssetItem.id)
                .where(
                    AssetItem.assigned_to == person.id,
                    AssetItem.deleted_at.is_(None),
                    AssetItem.disposed_at.is_(None),
                )
                .limit(1)
            )
        )
        computer_held = computer_count is not None
        # vpn — best-effort via UserPermission if the permission domain
        # is installed. Resolve lazily because permission is an optional
        # domain in some test fixtures.
        vpn_held = False
        async with best_effort(
            self.db, label="forensic: vpn permission probe", log=logger,
        ) as vpn_step:
            from app.domains.permission.models import (
                SoftwarePermission,
                UserPermission,
            )
            row = await self.db.scalar(
                select(UserPermission.id)
                .join(
                    SoftwarePermission,
                    UserPermission.permission_id == SoftwarePermission.id,
                )
                .where(
                    UserPermission.user_id == person.id,
                    SoftwarePermission.permission_key.ilike("%vpn%"),
                )
                .limit(1)
            )
            vpn_held = row is not None
        if not vpn_step.ok:
            vpn_held = False

        dims = {
            "account": bool(account_held),
            "sso": bool(sso_held),
            "work_email": bool(work_email_held),
            "feishu": bool(feishu_held),
            "seat": bool(seat_held),
            "computer": bool(computer_held),
            "vpn": bool(vpn_held),
        }
        return {
            "held": sum(1 for v in dims.values() if v),
            "of": len(_RESOURCE_DIMS),
            "dims": dims,
        }

    async def _count_lifetime_requests(
        self, person: IdentityPerson,
    ) -> int:
        """Total EmployeeRequest rows the person ever submitted.

        EmployeeRequest is the user-facing request entity; we count
        every row regardless of status so the KPI reflects "how active
        was this person on the platform".
        """
        try:
            from app.domains.asset.models import EmployeeRequest
        except Exception:  # noqa: BLE001
            return 0
        return int(
            await self.db.scalar(
                select(func.count(EmployeeRequest.id))
                .where(EmployeeRequest.requester_id == person.id)
            )
            or 0
        )

    async def _detect_in_flight(
        self, person: IdentityPerson,
    ) -> dict[str, Any]:
        """Is there a non-terminal offboarding OR role_change in progress?"""
        off_proc = await self.db.scalar(
            select(OffboardingProcess)
            .where(
                OffboardingProcess.person_id == person.id,
                OffboardingProcess.status != OFFBOARDING_STATUS_COMPLETED,
            )
            .order_by(OffboardingProcess.created_at.desc())
            .limit(1)
        )
        # Role-change in flight is harder to detect at this layer because
        # the dedicated coordinator + review_status column lands in a
        # separate PR (issue #623). For now, fall back to detecting an
        # open role_change HrLifecycleEvent in the last 60 days.
        sixty_days_ago = datetime.now(UTC) - timedelta(days=60)
        rc_event = await self.db.scalar(
            select(HrLifecycleEvent)
            .where(
                HrLifecycleEvent.person_id == person.id,
                HrLifecycleEvent.event_type == "role_change",
                HrLifecycleEvent.created_at >= sixty_days_ago,
            )
            .order_by(HrLifecycleEvent.created_at.desc())
            .limit(1)
        )
        return {
            "offboarding": off_proc is not None,
            "offboarding_process_id": str(off_proc.id) if off_proc else None,
            "role_change_recent": rc_event is not None,
        }


# ── Helpers ────────────────────────────────────────────────────────


def _person_summary(person: IdentityPerson) -> dict[str, Any]:
    """Minimal person info — enough for the timeline header. The
    forensic view exposes hire/termination dates and deletion state
    because the audit story depends on knowing the person was wiped."""
    return {
        "id": str(person.id),
        "display_name": person.display_name,
        "work_email": person.work_email,
        "employee_no": person.employee_no,
        "employment_status": person.employment_status,
        "hire_date": person.hire_date.isoformat() if person.hire_date else None,
        "termination_date": (
            person.termination_date.isoformat()
            if person.termination_date else None
        ),
        "deleted_at": (
            person.deleted_at.isoformat() if person.deleted_at else None
        ),
        "manager_person_id": (
            str(person.manager_person_id)
            if person.manager_person_id else None
        ),
    }


def _resolve_person_tz(person: IdentityPerson) -> ZoneInfo:
    """Use the same helper the rest of the platform uses; fall back
    to Asia/Shanghai (HQ default) on missing TZ data."""
    try:
        from app.core.working_hours import resolve_timezone
        tz = resolve_timezone(person)
        if tz is not None:
            return tz
    except Exception:  # noqa: BLE001
        pass
    return ZoneInfo("Asia/Shanghai")


def build_evidence_pack_bytes(
    payload: dict[str, Any], person: IdentityPerson,
) -> bytes:
    """Render the forensic payload as a downloadable ZIP.

    Layout::

        forensic-<person.id>.zip
        ├── manifest.json         — top-level person + KPI summary
        ├── timeline.json         — full merged timeline (time-desc)
        ├── timeline.csv          — same timeline as a flat CSV for
        │                            auditors who paste into Excel
        └── README.txt            — human-readable note: who, when, what

    Caller is responsible for the audit-log row that records the
    download (the router does it).
    """
    import csv
    import io
    import json
    import zipfile

    buf = io.BytesIO()
    generated_at = datetime.now(UTC).isoformat()
    with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
        zf.writestr(
            "manifest.json",
            json.dumps(
                {
                    "person": payload["person"],
                    "kpis": payload["kpis"],
                    "row_count": payload["row_count"],
                    "per_source_limit": payload["per_source_limit"],
                    "generated_at": generated_at,
                },
                ensure_ascii=False, indent=2,
            ),
        )
        zf.writestr(
            "timeline.json",
            json.dumps(
                payload["timeline"], ensure_ascii=False, indent=2,
            ),
        )

        csv_buf = io.StringIO()
        writer = csv.writer(csv_buf)
        writer.writerow(
            ["time", "event_class", "summary", "status_badge", "source", "detail_link"],
        )
        for row in payload["timeline"]:
            writer.writerow(
                [
                    row["time"],
                    row["event_class"],
                    row["summary"],
                    row.get("status_badge") or "",
                    row["source"],
                    row.get("detail_link") or "",
                ],
            )
        zf.writestr("timeline.csv", csv_buf.getvalue())

        zf.writestr(
            "README.txt",
            (
                "Helix Forensic Lifecycle Evidence Pack\n"
                "=======================================\n"
                f"Person ID:    {person.id}\n"
                f"Display name: {person.display_name or '(unknown)'}\n"
                f"Work email:   {person.work_email or '(unknown)'}\n"
                f"Generated at: {generated_at}\n\n"
                "Each row in timeline.json/csv has fields: time,\n"
                "event_class, summary, status_badge, source, detail_link.\n"
                "manifest.json carries the person summary + 4 KPIs + a\n"
                "``truncated_by_source`` map: keys are source table names,\n"
                "values are booleans — True means the per-source cap was\n"
                "hit and the oldest rows past the cap are NOT in this\n"
                "pack. Cap is set to 5000 per source for the evidence\n"
                "pack (500 for the interactive JSON endpoint).\n"
                "Soft-deleted persons are intentionally included.\n"
            ),
        )
    return buf.getvalue()


def _days_at_company(person: IdentityPerson) -> int | None:
    """Days from hire_date to (termination_date or today) inclusive,
    computed in the person's local TZ. ``None`` if hire_date unknown.

    Inclusive on both ends so a hire+leave on the same day counts as 1
    day rather than 0 (HR-side reporting convention)."""
    if person.hire_date is None:
        return None
    tz = _resolve_person_tz(person)
    end: date
    if person.termination_date is not None:
        end = person.termination_date
    else:
        end = datetime.now(tz).date()
    delta = (end - person.hire_date).days + 1
    return max(0, delta)