"""Hire-day IT info push.

Fires on an employee's ``hire_date`` (not before), sends a single
email + Feishu message with the IT info they need on Day-1:

* their SSO login URL + work email
* provisioned software list (from ``provisioning_tasks``)
* any hardware pickup instructions (from ``asset_items.assigned_to``)
* 2FA setup guidance (rendered from a template)
* IT support contact (platform setting)

Idempotent via ``identity_persons.it_info_sent_at``: once sent we
stamp the timestamp so a re-run on the same day (e.g. after a
worker restart) won't double-send.
"""
from __future__ import annotations

import html as _html
import logging
from datetime import UTC, date, datetime, timedelta
from typing import Any, NamedTuple
from uuid import UUID
from zoneinfo import ZoneInfo

from sqlalchemy import inspect as sa_inspect
from sqlalchemy import or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.corporate_domains import (
    email_is_corporate,
    get_corporate_email_domains,
)
from app.core.exceptions import AppException, format_exception_for_log
from app.core.locale import DEFAULT_LANG, normalize_lang, resolve_lang
from app.core.locale import is_en as locale_is_en
from app.core.name_sanitize import localized_name
from app.core.person_filters import is_test_employee_no
from app.core.working_hours import (
    get_provision_lead_hours,
    is_future_hire,
    provision_lead_window_days,
)
from app.domains.asset.models import ASSET_IN_USE_STATUSES, AssetItem
from app.domains.provisioning.models import ProvisioningTask
from app.identity.person_models import IdentityPerson
from app.integrations.service import get_platform_setting
from app.notifications.service import (
    CardAction,
    CardSection,
    CardStat,
    CardText,
    LocalizedText,
    NotificationEnvelope,
    NotificationService,
    resolve_card_text,
)

logger = logging.getLogger(__name__)

# Hard cap on the Day-1 welcome CC list. Bounds a pathological config (a
# 40-admin office master, a setting pasted full of addresses) without
# silently truncating: the caller logs what got dropped.
_WELCOME_CC_CAP = 10


class _WelcomeCc(NamedTuple):
    """Day-1 welcome CC list + whether its HR leg actually resolved.

    ``hr_covered`` gates the ``hr_cc`` stamp, so it must NOT be inferred
    from ``emails`` being non-empty — the IT contact resolves from a
    defaulted setting and would make the list unconditionally non-empty.
    """

    emails: list[str]
    hr_covered: bool


# Account-missing catch-up (Finding 6) bounds. Lookback keeps the sweep to
# genuinely-recent hires (a real day-1 miss is caught within days; anything
# older is a stale-backfill an admin handles manually). The cap is a
# storm-brake: more than this many recent hires with no Google account is
# an abnormal backlog (broken integration / bulk import), not routine.
_ACCOUNT_BACKFILL_LOOKBACK_DAYS = 14
_ACCOUNT_BACKFILL_MAX = 25

# Live SF presence re-check brake: more pre-start hires vanishing from SF
# in a single tick than this is an SF-side anomaly (auth/scope regression,
# or a NEW User.status value invisible to _USER_STATUS_ANY_FILTER — the
# 2026-07-12 active_external_suite trap that hid an entire EU cohort), not
# HR deleting botched entries one by one. Refuse to tombstone; defer the
# whole batch to the next tick and log loudly instead.
_PRESENCE_RECHECK_ABSENT_CAP = 10


class HireDayNotifyService:
    def __init__(self, db: AsyncSession) -> None:
        self.db = db
        self.notify = NotificationService(db)
        # Per-run memo for the live SF presence re-check: verdict per
        # person id ('present' / 'absent' / 'unverifiable' / 'unchecked')
        # so the main loop and the account backfill share one SF query.
        self._sf_presence_verdicts: dict[UUID, str] = {}

    async def run_for_today(
        self, *, today: date | None = None,
    ) -> dict[str, Any]:
        target_date = today or datetime.now(UTC).date()

        company_name = await get_platform_setting(
            "company.display_name", default="公司", session=self.db,
        )
        it_contact = await get_platform_setting(
            "it.contact_email", default="it@example.com", session=self.db,
        )
        sso_url = await get_platform_setting(
            "sso.login_url", default="https://helix.local/login", session=self.db,
        )
        portal_url = await get_platform_setting(
            "app.frontend_base_url",
            default="https://helix.local",
            session=self.db,
        )

        # Candidate window: provisioning opens ``lead_hours`` BEFORE the
        # local hire date (HR-configurable ``onboarding.provision_lead_hours``
        # platform setting, default 24h), so a hire whose UTC date is up to
        # ``provision_lead_window_days(lead)`` days ahead of ``target_date``
        # can already be inside their lead window once local-vs-UTC skew is
        # folded in (e.g. with a 24h lead a China +8 hire's window opens at
        # target_date-1 16:00 UTC, when their hire_date is still
        # target_date+1 in UTC). Scan a superset, then gate each person on
        # THEIR own timezone + the lead via ``is_future_hire``. The bounded
        # window keeps backfill from storming and ``it_info_sent_at`` keeps
        # each send once.
        lead_hours = await get_provision_lead_hours(self.db)
        window_days = provision_lead_window_days(lead_hours)
        rows = list((
            await self.db.execute(
                select(IdentityPerson)
                .where(
                    IdentityPerson.deleted_at.is_(None),
                    IdentityPerson.hire_date >= target_date - timedelta(days=1),
                    IdentityPerson.hire_date
                    <= target_date + timedelta(days=window_days),
                    IdentityPerson.employment_status == "active",
                    # Tombstoned rows (record deleted from the HRIS feed)
                    # must never re-enter the hire-day pipeline — the
                    # 2026-07-11 roger.zhou02 incident minted a mailbox
                    # for an SF-deleted hire through exactly this query.
                    IdentityPerson.hris_record_state != "purged",
                    IdentityPerson.it_info_sent_at.is_(None),
                )
            )
        ).scalars().all())
        rows = [p for p in rows if not is_future_hire(p, lead_hours=lead_hours)]
        # HR test-record band (P99XX): never notify/provision for HR's own
        # test records (defence-in-depth; they're blocked at ingest).
        rows = [
            p for p in rows
            if not is_test_employee_no(p.employee_no)
            and not is_test_employee_no(p.hris_person_id)
        ]

        # Live SF presence re-check (2026-07-16, closes the roger.zhou02
        # residual). The purged-tombstone gate above only sees deletions
        # the daily absent check has already recorded — but that check
        # deliberately exempts pre-start hires (they have no effective
        # EmpJob row yet, so roster absence means nothing). An SF entry
        # hard-deleted 1-2 days before start therefore stays 'present'
        # straight through this 24h-lead mint window. Re-verify each
        # pre-start candidate against the live SF User entity before any
        # account is minted; confirmed-gone hires are tombstoned here
        # with the same semantics the daily check uses.
        rows, presence_purged, presence_deferred = (
            await self._filter_by_live_sf_presence(rows)
        )

        summary: dict[str, Any] = {
            "date": target_date.isoformat(),
            "candidates": len(rows),
            "notified": 0,
            "skipped_no_email": 0,
            # Held back this tick because the hire's real local hire date
            # (resolved after the effective-dated enrich filled country/city)
            # hasn't arrived — sent on a later tick, not skipped forever.
            "deferred_not_local_hire_date": 0,
            "sf_presence_purged": presence_purged,
            "sf_presence_deferred": presence_deferred,
            "errors": [],
        }

        # Persons we successfully notified — Identity Sync Center
        # event hooks fan out to them AFTER the loop's session commits
        # (see end of function for the commit+fire block).
        hire_notified_person_ids: list = []

        # Pre-launch incident 2026-05-28 mirror (the offboarding side
        # fired auto_revoke on future-dated leavers; this is the
        # onboarding side). When ``HrSyncService.confirm_batch`` sees
        # a freshly-confirmed hire whose hire_date is still in the
        # future, the three coordinator hooks (provision_for_new_hire,
        # provision_google_account, send_onboarding_it_info) defer
        # themselves and rely on THIS cron to fire them when the date
        # arrives. Run them once-per-person (the it_info_sent_at filter
        # bounds the per-person work to a single day) BEFORE the IT
        # info dispatch so the welcome email reflects freshly-allocated
        # seats. All three hooks are idempotent — calling them when
        # they've already fired (legacy hires created before this PR
        # whose hire_date was always in the past) is a no-op via their
        # own existence checks.
        # Set each PRE-START hire's department (primary org membership) +
        # manager from their SF EmpJob effective ON the hire date, BEFORE the
        # Feishu account is minted below. The Feishu build reads the primary
        # org membership for ``department_ids``; without this the 24h-lead
        # account lands in the root "0" fallback and only self-heals on the
        # hire date (#1987). The future-dated EmpJob is absent from SF's
        # as-of-today slice, so ``effective_as_of=hire_date`` reads the day-1
        # slice early. Best-effort — an empty pull (SF has no future-dated job
        # yet) leaves the "0" fallback for #1987 to heal on the start date.

        # Read the two identifiers the loop needs BEFORE the two enrichers
        # below run. Both are best-effort and both ``rollback()`` on an SF
        # failure,
        # and a rollback expires every instance in the session regardless of
        # ``expire_on_commit=False`` — so touching ``person.hris_person_id``
        # afterwards would fire a lazy refresh and raise ``MissingGreenlet``
        # in this async session. Iterating plain tuples makes the loop
        # independent of instance state instead of relying on the enrichers
        # never failing.
        hook_targets = [(p.id, p.hris_person_id) for p in rows]

        await self._enrich_prestart_hire_department(rows)

        # Pull each pre-start hire's HRBP / HR Ops (SF 2448 / 2450) effective
        # ON their hire date BEFORE any account is minted below. Those
        # relationship rows are future-dated in SF, so the default
        # as-of-today slice is empty until day-1 —
        # ``effective_as_of=hire_date`` reads the day-1 slice early. Best-
        # effort: an empty pull (HR hasn't entered the future-dated
        # relationship yet) leaves the columns for the hire-day CC catchup
        # (``run_feishu_welcome_catchup``) to fill on the start date.
        #
        # This has to run BEFORE ``provision_google_account`` below, not just
        # before the welcome email: minting the mailbox also sends the
        # first-login credential mail, and THAT mail's CC audience
        # (``account_provisioner._resolve_welcome_cc_additions``) is read off
        # the very same two columns. With the enrich sitting after the loop
        # the credential mail CC'd only the IT contact while the Day-1 mail,
        # composed a few seconds later, CC'd HR correctly — which is exactly
        # how HR Ops noticed (P10027 刘姝婕 Nikita, 2026-09-01).
        await self._enrich_prestart_hire_relationships(rows)

        from app.core.coordinators import hr_sync_coordinator
        for person_id, hris_person_id in hook_targets:
            if not hris_person_id:
                continue
            try:
                await hr_sync_coordinator.provision_for_new_hire(
                    self.db, hris_person_id,
                )
                await hr_sync_coordinator.provision_google_account(
                    self.db, hris_person_id,
                )
                # Feishu account — counterpart to Google, fires within the
                # 24h lead window for a hire deferred at confirm time.
                await hr_sync_coordinator.provision_feishu_account(
                    self.db, hris_person_id,
                )
                # GitHub 组织自动加入 —— 必须**也**在这里重放。确认时那次
                # 调用被 ``_is_future_hire`` 拦下了(提前 30 天录入的人不该
                # 在录入那一刻就进组织),所以这里是它唯一的第二次机会:
                # 少了这一句,每一个未来日期入职的人都会永远不被自动加入,
                # 而且没有任何地方会说这件事。
                from app.core.coordinators import (  # noqa: PLC0415
                    github_org_coordinator,
                )
                await github_org_coordinator.apply_auto_join_for_new_hire(
                    self.db, hris_person_id,
                )
            except Exception as exc:  # noqa: BLE001
                logger.warning(
                    "hire_day: deferred onboarding hooks failed for "
                    "%s — IT info email still attempted below; %s",
                    person_id, format_exception_for_log(exc, max_len=300),
                )

        # Per-person commit: notify_person delivers the email synchronously
        # and then stamps ``it_info_sent_at`` — the stamp must be durable
        # before the next hire is attempted, or a later failure / lost
        # final commit re-sends already-delivered welcome emails on the
        # next tick. On failure, roll back so a poisoned session doesn't
        # break the rest of the batch (instances are re-loaded by id since
        # rollback expires them).
        row_ids = [p.id for p in rows]
        for pid in row_ids:
            person = await self.db.get(IdentityPerson, pid)
            if person is None or person.it_info_sent_at is not None:
                continue
            if not person.work_email:
                summary["skipped_no_email"] += 1
                continue
            # Re-check the local-hire-date gate AFTER the enrich above. The
            # window filter (top of this method) judged ``is_future_hire`` on
            # a hire whose ``country`` was still NULL — so ``resolve_timezone``
            # defaulted to Asia/Shanghai and a NA/EMEA hire read "already
            # started" up to ~12h before their real local midnight, sending
            # the welcome early with incomplete data (2026-08-03). Now that
            # ``_enrich_prestart_hire_department`` has filled country/city from
            # the effective-dated pull, re-judge with the correct timezone and
            # DEFER to the next tick if their local hire date hasn't arrived.
            # A hire whose enrich couldn't fill country (SF slice still not
            # present) keeps the Shanghai default and is not held back — the
            # localized fallbacks then keep the (rare) early render correct.
            if is_future_hire(person, lead_hours=lead_hours):
                summary["deferred_not_local_hire_date"] += 1
                continue
            try:
                await self.notify_person(
                    person,
                    company_name=company_name,
                    it_contact=it_contact,
                    sso_url=sso_url,
                    portal_url=portal_url,
                )
                await self.db.commit()
                summary["notified"] += 1
                hire_notified_person_ids.append(pid)
            except Exception as exc:  # noqa: BLE001
                await self.db.rollback()
                logger.exception(
                    "hire_day_it_notify: failed for %s", pid,
                )
                summary["errors"].append({
                    "person_id": str(pid),
                    "error": format_exception_for_log(exc, max_len=500),
                })

        # Identity Sync Center event hooks — stamps are already committed
        # per person above, so the fresh session inside
        # ``enqueue_person_sync`` observes ``it_info_sent_at``.
        # Fire-and-forget via ``spawn`` so a bulk Day-1 batch doesn't
        # serialize the hire-day notify run behind downstream SCIM
        # rate limits — see advance_service.run_tick for the same
        # pattern + rationale.
        if hire_notified_person_ids:
            from app.core.background import spawn
            from app.domains.identity_sync.events import enqueue_person_sync
            for pid in hire_notified_person_ids:
                # ``spawn`` retains a strong reference until completion —
                # a bare ``create_task`` here could be GC'd mid-flight and
                # silently drop a new hire's Day-1 directory push.
                spawn(
                    enqueue_person_sync(pid, reason="hire"),
                    name=f"hire_day_sync:{pid}",
                )

        # Account-missing catch-up (2026-07-13 review Finding 6). The
        # provisioning above is gated behind ``it_info_sent_at IS NULL``,
        # which conflates "account minted" with "IT-info notified". A hire
        # whose account creation FAILED on every tick inside the narrow
        # [target-1, target+2] window, or whose prior (botched) provisioning
        # stamped ``it_info_sent_at`` but whose account was later cleaned up
        # (王静文 P10012: the 7-11 roger.zhou02 account was deleted, the stamp
        # stayed), is then EXCLUDED from the query above forever and never
        # gets an account. Independently re-attempt for recent, active,
        # present hires who are missing EITHER onboarding account (Google
        # login identity and/or Feishu) — regardless of ``it_info_sent_at``.
        # Idempotent + fully gated (auto_create_on_hire / future / stale /
        # purged), so a false candidate is a benign skip, not a wrong mint.
        backfilled = await self._backfill_missing_onboarding_accounts(
            target_date,
        )
        summary["account_backfilled"] = backfilled["google"]
        summary["feishu_backfilled"] = backfilled["feishu"]
        summary["sf_presence_purged"] += backfilled["sf_presence_purged"]
        summary["sf_presence_deferred"] += backfilled["sf_presence_deferred"]

        return summary

    async def _backfill_missing_onboarding_accounts(
        self, target_date: date,
    ) -> dict[str, int]:
        """Re-attempt onboarding account creation for recent hires who are
        active + present but are missing an onboarding-provisioned account —
        Google (no live ``google`` login identity) and/or Feishu (no linked
        ``feishu_contact`` sync state). Bounded to the last
        ``_ACCOUNT_BACKFILL_LOOKBACK_DAYS`` so it can never storm.

        This is the ONLY onboarding path that reaches a hire whose
        ``it_info_sent_at`` is already stamped (the main loop above filters
        those out). It rescues a hire provisioned during the window between
        their 24h-lead opening and a later deploy of the Feishu-creation hook,
        whose Google minted + IT-info stamped on the first tick but whose
        Feishu was never attempted (P10011 Eddie Jiang, 2026-07-13). Onboarding
        account creation for Google and Feishu are INDEPENDENT steps (NOT the
        directory sync — that is department-scoped; onboarding creation is
        scope-independent), so a hire missing only Feishu is rescued even
        though they already have Google.

        Both provisioners are idempotent and fully gated (future / stale /
        purged / test / active), so a false candidate is a benign skip. The
        Feishu provisioner is fire-and-forget (spawns a background reconcile-
        first push), so its count is "requested", not "confirmed created".
        Returns ``{"google": n_created, "feishu": n_requested,
        "sf_presence_purged": n, "sf_presence_deferred": n}``.
        """
        from app.core.coordinators import hr_sync_coordinator
        from app.core.working_hours import (
            is_future_hire,
            is_stale_backfill_hire,
        )

        lead_hours = await get_provision_lead_hours(self.db)
        window_days = provision_lead_window_days(lead_hours)
        from app.domains.identity_sync.models import (
            IdentitySyncState,
            IdentitySyncTarget,
        )
        from app.identity.person_models import IdentityExternalIdentity

        lookback_start = target_date - timedelta(
            days=_ACCOUNT_BACKFILL_LOOKBACK_DAYS,
        )
        # Persons with an active, non-deleted google login identity — the
        # "Google done" set.
        has_google = (
            select(IdentityExternalIdentity.person_id)
            .where(
                IdentityExternalIdentity.provider == "google",
                IdentityExternalIdentity.deleted_at.is_(None),
                IdentityExternalIdentity.identity_status == "active",
            )
        )
        # "Feishu done" = a linked feishu_contact sync-state row (external_id
        # set). This is the durable, login-independent signal: the Feishu row
        # in identity_external_identities is only written when the hire logs
        # in via Feishu SSO, so it would falsely read "missing" for a just-
        # provisioned hire and re-push every tick. IdentitySyncState.external_id
        # is set the moment the onboarding force-push creates/links the account.
        enabled_feishu_targets = (
            select(IdentitySyncTarget.id)
            .where(
                IdentitySyncTarget.kind == "feishu_contact",
                IdentitySyncTarget.enabled.is_(True),
                IdentitySyncTarget.deleted_at.is_(None),
            )
        )
        has_feishu = (
            select(IdentitySyncState.person_id)
            .where(
                IdentitySyncState.target_id.in_(enabled_feishu_targets),
                IdentitySyncState.external_id.isnot(None),
            )
        )
        rows = list((await self.db.execute(
            select(IdentityPerson).where(
                IdentityPerson.deleted_at.is_(None),
                IdentityPerson.employment_status == "active",
                IdentityPerson.hris_record_state != "purged",
                IdentityPerson.hris_person_id.isnot(None),
                IdentityPerson.hire_date.isnot(None),
                IdentityPerson.hire_date >= lookback_start,
                # Upper bound mirrors run_for_today's superset window
                # (``provision_lead_window_days`` of the configurable lead):
                # a hire whose lead window is already OPEN but whose
                # hire_date is still days out must be rescued NOW, not only
                # once their start date arrives — else a hire missed during
                # their lead window (Google minted + stamped, Feishu skipped)
                # waits a full extra day. The per-person ``not
                # is_future_hire`` filter below is the real gate; this bound
                # is just an index-friendly prefilter.
                IdentityPerson.hire_date
                <= target_date + timedelta(days=window_days),
                or_(
                    IdentityPerson.id.notin_(has_google),
                    IdentityPerson.id.notin_(has_feishu),
                ),
            )
        )).scalars().all())
        # Drop future / stale-backfill hires — provisioning for them is not
        # this cron's job (future defers to their own lead window; stale is
        # an existing employee who already has an account elsewhere).
        candidates = [
            p for p in rows
            if not is_future_hire(p, lead_hours=lead_hours)
            and not is_stale_backfill_hire(p)
            # HR test-record band (P99XX): never backfill an account for
            # HR's own test records (defence-in-depth; blocked at ingest).
            and not is_test_employee_no(p.employee_no)
            and not is_test_employee_no(p.hris_person_id)
        ]
        if not candidates:
            return {
                "google": 0, "feishu": 0,
                "sf_presence_purged": 0, "sf_presence_deferred": 0,
            }
        if len(candidates) > _ACCOUNT_BACKFILL_MAX:
            # A large backlog is abnormal (a broken integration, a bulk
            # import). Log loudly and process only the cap so a real
            # incident doesn't silently mass-provision.
            logger.warning(
                "hire_day: account-backfill found %d candidates (cap %d) — "
                "processing the oldest %d only; investigate the backlog",
                len(candidates), _ACCOUNT_BACKFILL_MAX, _ACCOUNT_BACKFILL_MAX,
            )
            candidates.sort(key=lambda p: p.hire_date or target_date)
            candidates = candidates[:_ACCOUNT_BACKFILL_MAX]

        # Same live SF presence re-check as the main loop: the backfill
        # reaches pre-start hires the main loop excludes (it_info_sent_at
        # already stamped — the 王静文 P10012 shape), so it can mint for an
        # SF-deleted future hire through an identical hole. The per-run
        # memo means a candidate already verified above costs no extra
        # SF call.
        candidates, presence_purged, presence_deferred = (
            await self._filter_by_live_sf_presence(candidates)
        )
        if not candidates:
            return {
                "google": 0, "feishu": 0,
                "sf_presence_purged": presence_purged,
                "sf_presence_deferred": presence_deferred,
            }

        # Which providers each candidate still needs — one query per provider
        # over the (bounded) candidate id set, then set-membership below.
        cand_ids = [p.id for p in candidates]
        google_present = set((await self.db.execute(
            select(IdentityExternalIdentity.person_id).where(
                IdentityExternalIdentity.person_id.in_(cand_ids),
                IdentityExternalIdentity.provider == "google",
                IdentityExternalIdentity.deleted_at.is_(None),
                IdentityExternalIdentity.identity_status == "active",
            )
        )).scalars().all())
        # "Feishu settled" — do NOT re-drive these from onboarding backfill:
        #   * external_id set        → already linked/created (done)
        #   * last_status='skipped'  → deliberate no-op. The feishu runner
        #     returns skipped+no-external ONLY on the admin skip-list
        #     suppress path, so a skipped state means "intentionally not
        #     provisioned", never a transient miss.
        # A genuinely FAILED attempt (possibly transient) and a
        # never-attempted hire are NOT settled → still rescued.
        #
        # Was ``external_id IS NOT NULL`` only, which re-selected every
        # suppressed / skip-listed junk account (P09966 test.test,
        # P09985 test.malaysia, …) on EVERY hourly tick — the N concurrent
        # single-person pushes then fought over the single feishu-contact
        # target lock and N-1 lost with "locked by another sync" forever
        # (the 2026-07-14 sync-lock churn).
        feishu_settled = set((await self.db.execute(
            select(IdentitySyncState.person_id).where(
                IdentitySyncState.person_id.in_(cand_ids),
                IdentitySyncState.target_id.in_(enabled_feishu_targets),
                or_(
                    IdentitySyncState.external_id.isnot(None),
                    IdentitySyncState.last_status == "skipped",
                ),
            )
        )).scalars().all())
        # Admin skip-list (runner_options.skip_provision_person_ids): honour
        # the "never provision this person" decision in the rescue path too,
        # so a freshly skip-listed hire drops out immediately — before their
        # suppressed attempt has even written the 'skipped' state above.
        feishu_settled |= await self._feishu_skip_listed_person_ids(cand_ids)

        # Same pre-mint HRBP / HR Ops resolve the main loop does: this path
        # mints Google accounts too, and minting sends the first-login
        # credential mail whose CC audience is read off those two columns.
        # Skips the SF call for anyone already resolved, so a backfill sweep
        # over settled hires costs nothing. Identifiers are read out first
        # for the same reason as the main loop: the enrich rolls back on an
        # SF failure, and a rollback expires every instance in the session.
        mint_targets = [(p.id, p.hris_person_id) for p in candidates]
        await self._enrich_prestart_hire_relationships(candidates)

        google_created = 0
        feishu_requested = 0
        for person_id, hris_person_id in mint_targets:
            need_google = person_id not in google_present
            need_feishu = person_id not in feishu_settled
            try:
                if need_google:
                    await hr_sync_coordinator.provision_google_account(
                        self.db, hris_person_id,
                    )
                if need_feishu:
                    # Scope-independent onboarding Feishu creation (fire-and-
                    # forget; spawns a reconcile-first background push).
                    await hr_sync_coordinator.provision_feishu_account(
                        self.db, hris_person_id,
                    )
                await self.db.commit()
                if need_google:
                    # Re-check: a Google account now exists iff provisioning
                    # succeeded.
                    exists = await self.db.scalar(
                        select(IdentityExternalIdentity.id).where(
                            IdentityExternalIdentity.person_id == person_id,
                            IdentityExternalIdentity.provider == "google",
                            IdentityExternalIdentity.deleted_at.is_(None),
                            IdentityExternalIdentity.identity_status == "active",
                        ).limit(1)
                    )
                    if exists is not None:
                        google_created += 1
                if need_feishu:
                    feishu_requested += 1
            except Exception as exc:  # noqa: BLE001 — one hire must not abort the sweep
                await self.db.rollback()
                logger.warning(
                    "hire_day: account-backfill failed for %s: %s",
                    hris_person_id,
                    format_exception_for_log(exc, max_len=300),
                )
        return {
            "google": google_created,
            "feishu": feishu_requested,
            "sf_presence_purged": presence_purged,
            "sf_presence_deferred": presence_deferred,
        }

    async def _filter_by_live_sf_presence(
        self, persons: list[IdentityPerson],
    ) -> tuple[list[IdentityPerson], int, int]:
        """Drop pre-start hires whose HRIS record no longer exists in the
        live SF tenant, tombstoning them exactly like the daily
        ``hris.absent_person_check`` would.

        Why a LIVE check: the daily absent check exempts hires whose
        ``hire_date`` hasn't arrived (a future hire has no effective
        EmpJob row, so absence from ``list_current_jobs`` is structural,
        not evidence — the 静文/亦寒 false-tombstone lesson). That
        exemption means an SF entry HR hard-deletes 1-2 days before start
        (the roger.zhou02 shape) stays ``hris_record_state='present'``
        through the whole 24h-lead mint window, and the purged gate
        (#1905) never fires. This re-check queries the SF ``User`` entity
        directly (``get_users`` spans every known status incl.
        ``active_external_suite``), so for a pre-start hire "no row" means
        exactly one thing: the record was deleted at the source.

        Scope: only ``hire_date >= as_of`` (Asia/Shanghai company date —
        the same rule-1 boundary as ``classify_inactive_sf_user``, #1953)
        needs this; anyone already started is the daily check's job.

        Verdict semantics per person:
          * present      → mint proceeds
          * absent       → tombstone (purged) + lifecycle event + audit,
                           excluded from minting
          * unverifiable → SF query failed, or the per-tick absent brake
                           tripped: fail CLOSED — skip this tick, no
                           tombstone; the hourly cron retries well within
                           the 24h lead
          * unchecked    → non-SF source / no resolvable HRIS instance:
                           fail OPEN (legacy rows keep today's behaviour;
                           the #1905 purged gate still backstops)

        Returns ``(allowed, purged_count, deferred_count)``.
        """
        as_of = datetime.now(ZoneInfo("Asia/Shanghai")).date()
        need_check = [
            p for p in persons
            if p.source_system == "successfactors"
            and p.hris_person_id
            and p.hire_date is not None
            and p.hire_date >= as_of
            and p.id not in self._sf_presence_verdicts
        ]
        if need_check:
            await self._verify_sf_presence(need_check)

        allowed: list[IdentityPerson] = []
        absent: list[IdentityPerson] = []
        deferred = 0
        for p in persons:
            verdict = self._sf_presence_verdicts.get(p.id)
            if verdict == "absent":
                absent.append(p)
            elif verdict == "unverifiable":
                deferred += 1
                logger.warning(
                    "hire_day: SF presence for pre-start hire %s (%s) "
                    "could not be verified — deferring provisioning to "
                    "the next tick (no tombstone)",
                    p.hris_person_id, p.work_email,
                )
            else:
                allowed.append(p)
        purged = await self._tombstone_sf_absent_hires(absent) if absent else 0
        return allowed, purged, deferred

    async def _verify_sf_presence(
        self, persons: list[IdentityPerson],
    ) -> None:
        """Query each person's own SF instance for their ``User`` row and
        record a verdict in ``self._sf_presence_verdicts``. Never raises:
        a failed instance query marks its whole group 'unverifiable'
        (fail-closed); an unresolvable instance marks 'unchecked'
        (fail-open, preserves pre-existing behaviour for legacy rows)."""
        # Function-level imports mirror run_hris_absent_person_check so
        # tests patch the exact same targets.
        from app.identity.successfactors_auth import build_auth_from_config
        from app.identity.successfactors_client import SuccessFactorsClient
        from app.integrations.compat import list_hris_instances

        instances = {
            row.id: (row, config)
            for row, config in await list_hris_instances(session=self.db)
        }
        by_instance: dict[Any, list[IdentityPerson]] = {}
        for p in persons:
            if p.hris_instance_id is None or p.hris_instance_id not in instances:
                logger.info(
                    "hire_day: presence re-check skipped for %s — no "
                    "resolvable HRIS instance (instance_id=%s); "
                    "proceeding as before",
                    p.hris_person_id, p.hris_instance_id,
                )
                self._sf_presence_verdicts[p.id] = "unchecked"
                continue
            by_instance.setdefault(p.hris_instance_id, []).append(p)

        fresh_absent: list[IdentityPerson] = []
        for instance_id, group in by_instance.items():
            row, config = instances[instance_id]
            alias = row.slug or str(row.id)
            try:
                base_url = (config.get("base_url") or "").rstrip("/")
                if not base_url:
                    raise ValueError("missing base_url")
                client = SuccessFactorsClient(
                    base_url=base_url,
                    auth=build_auth_from_config(config),
                )
                users = await client.get_users(
                    [p.hris_person_id for p in group],
                )
                present = {
                    str(u["userId"]) for u in users if u.get("userId")
                }
            except Exception as exc:  # noqa: BLE001
                logger.warning(
                    "hire_day: SF presence re-check failed for instance "
                    "%s (%d pre-start hires) — failing closed this tick; "
                    "%s",
                    alias, len(group),
                    format_exception_for_log(exc, max_len=300),
                )
                for p in group:
                    self._sf_presence_verdicts[p.id] = "unverifiable"
                continue
            for p in group:
                if p.hris_person_id in present:
                    self._sf_presence_verdicts[p.id] = "present"
                else:
                    fresh_absent.append(p)

        if len(fresh_absent) > _PRESENCE_RECHECK_ABSENT_CAP:
            logger.error(
                "hire_day: presence re-check found %d pre-start hires "
                "missing from SF in one tick (brake cap %d) — this smells "
                "like an SF auth/scope regression or a new User.status "
                "value, not per-person deletions. Refusing to tombstone; "
                "deferring all of them. Investigate before the lead "
                "window closes.",
                len(fresh_absent), _PRESENCE_RECHECK_ABSENT_CAP,
            )
            for p in fresh_absent:
                self._sf_presence_verdicts[p.id] = "unverifiable"
        else:
            for p in fresh_absent:
                self._sf_presence_verdicts[p.id] = "absent"

    @staticmethod
    def _needs_job_enrich(person: IdentityPerson) -> bool:
        """True iff this hire is still missing any welcome-critical field the
        effective-dated EmpJob pull can fill: department (primary membership),
        manager, work location, job title, or work country (the last drives
        ``is_china_employee`` → welcome LANGUAGE + reimbursement tool, and
        ``resolve_timezone`` → the local-hire-date gate).

        Deliberately NOT gated on ``is_future_hire`` (the #2013/#2015 bug):
        a pre-start NA/EMEA hire has NULL country, so ``resolve_timezone``
        defaults to Asia/Shanghai and mis-judges them "already started" — the
        exact hires whose EmpJob is still future-dated and who most need this
        enrich were the ones it skipped, shipping English welcome emails with
        ``你的团队`` / ``(待定)`` / ``待确认 (TBD)`` (2026-08-03). Selecting on
        MISSING FIELDS instead is both correct and safe: the writes are
        fill-only (write-when-present), so a hire whose data already landed
        skips the SF call, and one whose future-dated slice isn't in SF yet is
        simply left for the hire-date sync to fill.
        """
        if (
            person.manager_person_id is None
            or not person.work_location_name
            or not person.job_title
            or not person.country
        ):
            return True
        # Department lives on the ``org_memberships`` relationship. Consult it
        # ONLY when it is already loaded — the caller's ``rows`` are
        # ``selectin``-loaded, but this sync predicate must never trigger a
        # lazy IO load (MissingGreenlet). When unloaded, the scalar fields
        # above already decided; a genuinely dept-less hire with every other
        # field present is vanishingly rare (same EmpJob supplies both).
        if "org_memberships" not in sa_inspect(person).unloaded:
            return person.primary_department is None
        return False

    async def _enrich_prestart_hire_department(
        self, persons: list[IdentityPerson],
    ) -> int:
        """Best-effort: for each not-yet-complete hire, pull their SF EmpJob
        effective ON the hire date and apply the full job field set (department
        + manager + work location + job title + company + ``is_china_employee``
        + cost-center + job-function/family + 外派 …) BEFORE the welcome email /
        Feishu account is minted.

        The Feishu account's ``department_ids`` comes from the person's primary
        org membership (``org_node_source_id``). At the 24h-lead build a future
        hire's EmpJob is still future-dated → absent from SF's as-of-today slice
        → no membership → Feishu falls back to root dept "0" and only self-heals
        on the hire date (#1987, yihan.mo / Chi Yeung Yip 2026-07-20). Reading
        the slice ``effective_as_of=hire_date`` gets the real day-1 data at
        build so the account lands in the right department AND the welcome email
        renders Team / Location / Manager / language correctly instead of the
        Chinese fallbacks (2026-08-03 NA/EMEA report).

        Reuses ``HrSyncService``'s tested ``_ensure_master_data_loaded`` (seeds
        the FODepartment / FOCompany / FOLocation / position caches) +
        ``_apply_job_snapshot_fields`` (the exact confirm-path field logic) +
        ``_upsert_dept_membership`` + ``_assign_manager`` — no re-implementation.
        Selects on MISSING FIELDS (``_needs_job_enrich``), not
        ``is_future_hire`` — see that helper. Every write is fill-only, so this
        never front-runs a live employee's future transfer (a started employee
        with complete data is skipped) and re-running on the hire date only
        refines. Grouped by (HRIS instance, hire_date); per-instance commit;
        never raises — an empty pull (SF has no future-dated EmpJob yet) leaves
        the fields for the hire-date sync to fill. Returns the number of people
        enriched.
        """
        from app.domains.hr.service import SuccessFactorsLifecycleService
        from app.identity.successfactors_auth import build_auth_from_config
        from app.identity.successfactors_client import SuccessFactorsClient
        from app.integrations.compat import list_hris_instances

        candidates = [
            p for p in persons
            if p.hris_person_id
            and p.hris_instance_id is not None
            and p.hire_date is not None
            and self._needs_job_enrich(p)
        ]
        if not candidates:
            return 0

        instances = {
            row.id: (row, config)
            for row, config in await list_hris_instances(session=self.db)
        }
        by_key: dict[tuple[Any, date], list[IdentityPerson]] = {}
        for p in candidates:
            if p.hris_instance_id not in instances or p.hire_date is None:
                continue
            by_key.setdefault((p.hris_instance_id, p.hire_date), []).append(p)

        assigned = 0
        for (instance_id, hire_date), group in by_key.items():
            row, config = instances[instance_id]
            alias = row.slug or str(row.id)
            try:
                base_url = (config.get("base_url") or "").rstrip("/")
                if not base_url:
                    raise ValueError("missing base_url")
                client = SuccessFactorsClient(
                    base_url=base_url, auth=build_auth_from_config(config),
                )
                uids = [p.hris_person_id for p in group]
                jobs = await client.get_jobs_for_users(
                    uids, effective_as_of=hire_date,
                )
                # First (date-effective) row per userId — the fromDate=toDate
                # window already pins to the slice effective on the hire date.
                jobs_by_uid: dict[str, dict] = {}
                for j in jobs:
                    uid = str(j.get("userId") or "")
                    if uid and uid not in jobs_by_uid:
                        jobs_by_uid[uid] = j
                if not jobs_by_uid:
                    # SF has no future-dated EmpJob yet — leave the "0" fallback
                    # for #1987 to heal on the hire date.
                    continue
                svc = SuccessFactorsLifecycleService(
                    self.db, sf_client=client, hris_instance_id=instance_id,
                )
                await svc._ensure_master_data_loaded()
                for p in group:
                    job = jobs_by_uid.get(p.hris_person_id or "")
                    if not job:
                        continue
                    # Full confirm-path field set: job_title / location /
                    # company / country / is_china_employee / cost-center /
                    # job-function&family / fte / probation / 外派 — the fields
                    # the welcome email + downstream language selection need,
                    # not just dept + manager (the #2013/#2015 gap).
                    svc._apply_job_snapshot_fields(p, job)
                    # Manager is set from the same snapshot even when the
                    # EmpJob carries no department code (rare, but then
                    # ``_upsert_dept_membership`` has nothing to write).
                    await svc._assign_manager(p, job)
                    dept_code = job.get("department")
                    if dept_code:
                        await svc._upsert_dept_membership(p, dept_code)
                    assigned += 1
                await self.db.commit()
            except Exception as exc:  # noqa: BLE001
                await self.db.rollback()
                logger.warning(
                    "hire_day: pre-start department enrich failed for instance "
                    "%s (%d hires, hire_date=%s) — Feishu account may land in "
                    "root '0'; #1987 heals it on the hire date; %s",
                    alias, len(group), hire_date,
                    format_exception_for_log(exc, max_len=300),
                )
        return assigned

    async def _enrich_prestart_hire_relationships(
        self, persons: list[IdentityPerson],
    ) -> int:
        """Best-effort: pull each hire's HRBP / HR Ops (SF 2448 / 2450)
        effective ON their hire date and write ``hris_hrbp_person_id`` /
        ``hris_hrops_person_id`` BEFORE any account is minted for them.

        The 24h-lead build runs before those relationship rows become the
        current SF slice (they're future-dated), so the default sync leaves
        them NULL and the welcome email CC's nobody (xiaojing / HR Ops report
        2026-07-20). Reading the slice ``effective_as_of=hire_date`` fills
        them at build so the CC lands on the same email the hire gets.

        Callers must run this BEFORE ``provision_google_account``, not merely
        before ``notify_person``: minting the mailbox also sends the
        first-login credential mail, and its CC audience
        (``account_provisioner._resolve_welcome_cc_additions``) reads the
        very same two columns. It sat after the mint loop until 2026-09-01,
        which is why HR Ops received the Day-1 mail for a hire but never the
        account-ready mail (P10027 刘姝婕 Nikita).

        Only touches hires still MISSING a relationship column (skips the SF
        call when both are already resolved). Grouped by HRIS instance;
        per-person ``effective_as_of`` since hires in one tick can have
        different start dates. Never raises and never clears on empty — a hire
        whose HR hasn't entered the future-dated relationship yet is simply
        left for the hire-day CC catchup to fill on the start date. Commits
        per instance so the notify loop's re-loaded person sees the writes.
        Returns the number of people whose columns were (re)assigned.
        """
        from app.identity.successfactors_auth import build_auth_from_config
        from app.identity.successfactors_client import SuccessFactorsClient
        from app.identity.successfactors_sync import (
            apply_hris_role_relationships,
        )
        from app.integrations.compat import list_hris_instances

        candidates = [
            p for p in persons
            if p.hris_person_id
            and p.hris_instance_id is not None
            and p.hire_date is not None
            and (
                p.hris_hrops_person_id is None
                or p.hris_hrbp_person_id is None
            )
        ]
        if not candidates:
            return 0

        instances = {
            row.id: (row, config)
            for row, config in await list_hris_instances(session=self.db)
        }
        assigned = 0
        # Group by (instance, hire_date) so a single scoped pull covers every
        # hire that shares a start date on the same tenant.
        by_key: dict[tuple[Any, date], list[IdentityPerson]] = {}
        for p in candidates:
            if p.hris_instance_id not in instances or p.hire_date is None:
                # ``candidates`` already filters hire_date not-None; the
                # explicit guard narrows it for the typed dict key too.
                continue
            by_key.setdefault((p.hris_instance_id, p.hire_date), []).append(p)

        for (instance_id, hire_date), group in by_key.items():
            row, config = instances[instance_id]
            alias = row.slug or str(row.id)
            try:
                base_url = (config.get("base_url") or "").rstrip("/")
                if not base_url:
                    raise ValueError("missing base_url")
                client = SuccessFactorsClient(
                    base_url=base_url,
                    auth=build_auth_from_config(config),
                )
                n = await apply_hris_role_relationships(
                    self.db, client, group,
                    scoped_pull=True, effective_as_of=hire_date,
                )
                if n:
                    await self.db.commit()
                    assigned += n
            except Exception as exc:  # noqa: BLE001
                await self.db.rollback()
                logger.warning(
                    "hire_day: pre-start HRBP/HR Ops enrich failed for "
                    "instance %s (%d hires, hire_date=%s) — welcome CC may be "
                    "empty; hire-day CC catchup will fill it on start date; %s",
                    alias, len(group), hire_date,
                    format_exception_for_log(exc, max_len=300),
                )
        return assigned

    async def _tombstone_sf_absent_hires(
        self, persons: list[IdentityPerson],
    ) -> int:
        """Stamp the standard soft tombstone (same fields, lifecycle event
        and audit action as ``hris.absent_person_check``, migration 222)
        on pre-start hires confirmed gone from SF, then commit. The
        distinct ``hris_purged_reason`` keeps forensics separable; the
        shared state means every downstream consumer — the #1905 purged
        provisioning gates, the admin「已从 HRIS 抹除」badge, login blocking,
        and the sync-side un-purge on re-appearance — behaves identically.
        Returns the number of rows actually flipped present→purged."""
        from app.audit.service import AuditService
        from app.domains.hr.models import HrLifecycleEvent

        audit = AuditService(self.db)
        now = datetime.now(UTC)
        trace_id = f"hire_day.sf_presence_recheck:{now.isoformat()}"
        purged = 0
        for p in persons:
            if p.hris_record_state == "purged":
                continue
            try:
                # SAVEPOINT per row (same rationale as the daily check):
                # one bad audit/lifecycle insert must not poison the
                # session for the remaining rows or the mint loop after.
                async with self.db.begin_nested():
                    p.hris_record_state = "purged"
                    p.hris_purged_at = now
                    p.hris_purged_reason = "absent_at_hire_day_recheck"
                    self.db.add(HrLifecycleEvent(
                        person_id=p.id,
                        event_type="hris_purged",
                        effective_date=now.date(),
                        detail={
                            "hris_person_id": p.hris_person_id,
                            "work_email": p.work_email,
                            "display_name": p.display_name,
                            "hire_date": (
                                p.hire_date.isoformat()
                                if p.hire_date else None
                            ),
                            "reason": "absent_at_hire_day_recheck",
                            "source": "hire_day_presence_recheck",
                        },
                    ))
                    await audit.log(
                        trace_id=trace_id,
                        actor_type="system",
                        actor_user_id=None,
                        action="hris.absent_person_detected",
                        status="success",
                        resource_type="identity_person",
                        resource_id=str(p.id),
                        detail={
                            "hris_person_id": p.hris_person_id,
                            "work_email": p.work_email,
                            "display_name": p.display_name,
                            "hire_date": (
                                p.hire_date.isoformat()
                                if p.hire_date else None
                            ),
                            "source": "hire_day_presence_recheck",
                            "tombstone": "applied",
                        },
                    )
                purged += 1
                logger.warning(
                    "hire_day: SF no longer knows pre-start hire %s "
                    "(%s, hire_date=%s) — tombstoned; NOT minting "
                    "accounts",
                    p.hris_person_id, p.work_email, p.hire_date,
                )
            except Exception:  # noqa: BLE001
                logger.exception(
                    "hire_day: presence-recheck tombstone failed for %s "
                    "— row stays 'present'; the next tick re-checks",
                    p.id,
                )
        # Durable BEFORE any provisioning hook runs: the tombstone is the
        # very gate (#1905) those hooks consult.
        await self.db.commit()
        return purged

    async def _feishu_skip_listed_person_ids(
        self, candidate_ids: list[UUID],
    ) -> set[UUID]:
        """Person ids on any enabled ``feishu_contact`` target's admin
        skip-list (``runner_options.skip_provision_person_ids``), narrowed
        to ``candidate_ids``. These are intentionally never-provision, so
        the onboarding backfill must not re-drive them (they only ever
        resolve to a suppressed 'skipped' — re-pushing hourly just burns a
        target-lock slot). Malformed ids are dropped, not fatal."""
        from app.domains.identity_sync.models import IdentitySyncTarget

        if not candidate_ids:
            return set()
        target_opts = (await self.db.execute(
            select(IdentitySyncTarget.runner_options).where(
                IdentitySyncTarget.kind == "feishu_contact",
                IdentitySyncTarget.enabled.is_(True),
                IdentitySyncTarget.deleted_at.is_(None),
            )
        )).scalars().all()
        cand = set(candidate_ids)
        out: set[UUID] = set()
        for opts in target_opts:
            for raw in (opts or {}).get("skip_provision_person_ids") or []:
                try:
                    pid = UUID(str(raw))
                except (ValueError, TypeError):
                    continue
                if pid in cand:
                    out.add(pid)
        return out

    async def _render_welcome_payload(
        self,
        person: IdentityPerson,
        *,
        company_name: str,
        it_contact: str,
        sso_url: str,
        portal_url: str,
    ) -> dict[str, Any]:
        """Render the Day-1 welcome once (title + text + rich HTML + template
        context), so the employee email (``notify_person``) and the HR CC copy
        (``_send_welcome_hr_copy``) ship byte-identical content.

        Language + reimbursement tool key off the LABOUR-CONTRACT ENTITY
        (``is_china_employee``, HR Ops 2026-07-15), not ``preferred_language``:
        a China-entity hire gets the Chinese welcome + 汇联易(Helios), everyone
        else the English welcome + Navan. People names render language-matched
        (Latin for EN, CJK for zh); the hire's own name falls back to the work
        email when no display_name is synced (never split an email).
        """
        is_china = bool(person.is_china_employee)
        is_en = not is_china

        software_list = await self._software_list(person.id)
        hardware_list = await self._hardware_list(person.id)
        manager_name, hrbp_name = await self._manager_hrbp_names(person, is_en=is_en)
        # Email path — entity axis end to end, deliberately unchanged.
        network_contact = await self._network_owner_name(is_en=is_en)
        person_name = (
            localized_name(person.display_name, is_en=is_en)
            if person.display_name else person.work_email
        )
        department = person.primary_department or _fallback_team(is_en)
        job_title = person.job_title or _fallback_tbd(is_en)
        location = (
            person.work_location_name
            or person.work_location_code
            or _fallback_tbd(is_en)
        )
        blessing_zh, blessing_en, blessing_ja = _blessing_variants(
            person,
            company_name=company_name,
            person_name=person_name,
            department=department,
            manager_name=manager_name,
        )
        body_text = _render_body(
            person=person,
            company_name=company_name,
            it_contact=it_contact,
            sso_url=sso_url,
            portal_url=portal_url,
            software_list=software_list,
            hardware_list=hardware_list,
            is_en=is_en,
        )
        # Rich HTML is built here (not as a template ``body_html`` field) so
        # it stays a unit-testable function instead of a brittle giant
        # ``.format`` string, and renders in the recipient's own language.
        hris_manual_title, hris_manual_url = _hris_manual_for(person)
        body_html = _render_onboarding_html(
            is_en=is_en,
            is_china=is_china,
            company_name=company_name,
            person_name=person_name,
            person_email=person.work_email,
            department=department,
            location=location,
            manager_name=manager_name,
            hrbp_name=hrbp_name,
            sso_url=sso_url,
            portal_url=portal_url,
            network_contact=network_contact,
            it_contact=it_contact,
            seats=software_list,
            blessing=blessing_en if is_en else blessing_zh,
            hris_manual_title=hris_manual_title,
            hris_manual_url=hris_manual_url,
        )
        title = (
            f"🎉 Welcome to {company_name}! Your IT accounts & day-one guide"
            if is_en
            else f"🎉 欢迎加入 {company_name}!你的 IT 账号与第一天指南"
        )
        return {
            "is_en": is_en,
            "title": title,
            "body_text": body_text,
            "body_html": body_html,
            "context": {
                "person_name": person_name,
                "person_email": person.work_email,
                "company_name": company_name,
                "it_contact": it_contact,
                "sso_url": sso_url,
                "portal_url": portal_url,
                "department": department,
                # 职位/岗位 is no longer rendered to the hire (HR: not
                # employee-visible yet), but the key stays in context so an
                # admin-overridden template still referencing {job_title}
                # won't KeyError and blank the whole email.
                "job_title": job_title,
                "location": location,
                "software_list": _format_list(software_list) or "  · (随部门策略陆续发放)",
                "hardware_list": _format_list(hardware_list) or "  · (暂无,如需在 Helix 申请)",
                "manager_name": manager_name,
                "hrbp_name": hrbp_name,
                "network_contact": network_contact,
                "hris_manual_title": hris_manual_title,
                "hris_manual_url": hris_manual_url,
                # Locale-fragment triple — the zh / en / ja template each
                # interpolates its own. Kept as fragments (not resolved to
                # one language here) so the stored context still renders
                # correctly if the reader later switches language.
                "blessing_zh": blessing_zh,
                "blessing_en": blessing_en,
                "blessing_ja": blessing_ja,
            },
        }

    async def notify_person(
        self,
        person: IdentityPerson,
        *,
        company_name: str,
        it_contact: str,
        sso_url: str,
        portal_url: str,
    ) -> None:
        """Send the SINGLE new-hire onboarding IT-info email to one person's
        work mailbox (feishu + email) and stamp ``it_info_sent_at``.

        Shared by the hire_day cron AND the confirm-time coordinator hook
        (``hr_sync_coordinator.send_onboarding_it_info``), so a hire gets
        exactly one such email whether they were confirmed before, on, or
        after their start date. Renders per-language from the
        ``hire_day_it_info`` template (registry → admin-overridable);
        {software_list}/{hardware_list} are pre-formatted here. The stamp
        lands only AFTER dispatch succeeds — the outbox row is the durable
        retry handle, so a transient failure leaves the person unstamped and
        the next tick retries. Raises on failure; callers record it.
        """
        payload = await self._render_welcome_payload(
            person,
            company_name=company_name,
            it_contact=it_contact,
            sso_url=sso_url,
            portal_url=portal_url,
        )
        # CC the hire's HRBP + HR Ops, their local asset admins, and the IT
        # contact (孟凡生) so everyone who has day-one work for this hire sees
        # the same welcome — the audience the account-created mail already
        # reaches (HR Ops request 2026-07-15; asset admins + IT 2026-08-24).
        # At the 24h-lead build the SF work-relationship rows (2448/2450) may
        # still be future-dated and unresolved — ``_enrich_prestart_hire_
        # relationships`` tries to pull them early, but if HR hasn't entered
        # them yet the HR leg is empty. When the HR leg DID resolve we stamp
        # ``hr_cc`` so the catchup won't re-send; when it didn't we leave the
        # stamp off and ``run_feishu_welcome_catchup`` delivers the HR copy on
        # the start date once the relationship syncs (yihan.mo / HR Ops
        # "当地 Ops 没收到抄送" 2026-07-20). Note the gate is ``hr_covered``,
        # not "cc_emails non-empty": the IT contact makes the list non-empty
        # for every hire, so the old test would disarm that catchup forever.
        cc_emails, hr_covered = await self._welcome_cc_emails(person)
        await self.notify.dispatch(
            NotificationEnvelope(
                recipient=person,
                title=payload["title"],
                body_text=payload["body_text"],
                body_html=payload["body_html"],
                cc_emails=cc_emails,
                # Email-only: a day-1 hire has no Feishu account yet (the
                # Feishu open_id lookup fails — "open_id not found"), so the
                # welcome must land in their freshly-minted Google mailbox as
                # the rich HTML (``body_html``), not fan out to a Feishu
                # channel that can never deliver. IT-facing "mailbox created"
                # notices still use Feishu; this is the new-hire welcome only.
                channels=["email"],
                template_key="hire_day_it_info",
                context=payload["context"],
                related_resource_type="identity_person",
                related_resource_id=person.id,
            ),
            trace_id="hire_day_it_notify",
        )
        person.it_info_sent_at = datetime.now(UTC)
        person.it_info_sent_channels = [
            "email", *(["hr_cc"] if hr_covered else []),
        ]
        await self.db.flush()

        # Feishu welcome card — best-effort, immediately after the email.
        # Usually the hire's Feishu account doesn't exist yet at this
        # moment (identity-sync mints it minutes later), so this attempt
        # is a no-op and the ``hr.hire_day_feishu_welcome`` 30-min sweep
        # delivers the card once the account appears. But for hires
        # confirmed AFTER their start date (late confirm / re-hire) the
        # account already exists and the card lands right now. A failure
        # here must never taint the just-sent email path.
        try:
            await self.send_feishu_welcome_card(
                person,
                company_name=company_name,
                it_contact=it_contact,
                sso_url=sso_url,
                portal_url=portal_url,
            )
        except Exception as exc:  # noqa: BLE001
            logger.warning(
                "hire_day: immediate feishu welcome card attempt failed "
                "for %s (sweep will retry): %s",
                person.id, format_exception_for_log(exc, max_len=300),
            )

    async def run_day90_survey(
        self, *, now: datetime | None = None,
    ) -> dict[str, Any]:
        """Invite hires who reached 90 days of tenure to the
        onboarding-experience survey (Feishu card + email, bilingual).

        The 90-day mark is computed on the EMPLOYEE'S local calendar
        (``resolve_timezone``), same discipline as the other lifecycle
        cutoffs. A 90–97-day catch-up window absorbs missed cron ticks;
        anyone past 97 days at rollout is deliberately never invited
        (no push storm at feature launch). Idempotent via
        ``identity_persons.day90_survey_sent_at``; committed per person
        so a mid-batch failure never re-invites already-notified hires.
        """
        from app.core.working_hours import resolve_timezone

        now_dt = now or datetime.now(UTC)
        today_utc = now_dt.date()

        company_name = await get_platform_setting(
            "company.display_name", default="公司", session=self.db,
        )

        # UTC prefilter ±1 day around the local 90–97 window (index-
        # friendly); the per-person local-date gate below is the truth.
        rows = list((
            await self.db.execute(
                select(IdentityPerson).where(
                    IdentityPerson.deleted_at.is_(None),
                    IdentityPerson.employment_status == "active",
                    IdentityPerson.hris_record_state != "purged",
                    IdentityPerson.work_email.isnot(None),
                    IdentityPerson.hire_date.isnot(None),
                    IdentityPerson.day90_survey_sent_at.is_(None),
                    IdentityPerson.hire_date >= today_utc - timedelta(days=98),
                    IdentityPerson.hire_date <= today_utc - timedelta(days=89),
                )
            )
        ).scalars().all())

        summary: dict[str, Any] = {
            "date": today_utc.isoformat(),
            "candidates": 0,
            "notified": 0,
            "errors": [],
        }

        due_ids = []
        for person in rows:
            local_today = now_dt.astimezone(resolve_timezone(person)).date()
            tenure_days = (local_today - person.hire_date).days
            if 90 <= tenure_days <= 97:
                due_ids.append(person.id)
        summary["candidates"] = len(due_ids)

        for pid in due_ids:
            person = await self.db.get(IdentityPerson, pid)
            if person is None or person.day90_survey_sent_at is not None:
                continue
            try:
                await self._send_day90_survey(person, company_name=company_name)
                await self.db.commit()
                summary["notified"] += 1
            except Exception as exc:  # noqa: BLE001
                await self.db.rollback()
                logger.exception("day90_survey: failed for %s", pid)
                summary["errors"].append({
                    "person_id": str(pid),
                    "error": format_exception_for_log(exc, max_len=500),
                })
        return summary

    async def _send_day90_survey(
        self, person: IdentityPerson, *, company_name: str,
    ) -> None:
        """Dispatch the Day-90 survey invite to one person (feishu +
        email, one envelope → one inbox row) and stamp
        ``day90_survey_sent_at``. Raises on failure; caller records it."""
        person_name = person.display_name or person.work_email
        # Shared resolver — a raw ``startswith("zh")`` test read the free-text
        # SF value "chinese" as *not* Chinese and sent Chinese-preferring hires
        # the English survey (``app.core.locale``).
        is_en = locale_is_en(person)
        fields = _build_day90_survey_fields(
            # Reader's language for everything the card shows; ``is_en`` keeps
            # picking the (bilingual) HTML e-mail body.
            lang=resolve_lang(person),
            is_en=is_en,
            company_name=company_name,
            person_name=person_name,
        )
        await self.notify.dispatch(
            NotificationEnvelope(
                recipient=person,
                title=fields["title"],
                body_text=fields["body_text"],
                body_html=fields["body_html"],
                channels=["feishu", "email"],
                template_key="onboarding.day90_survey",
                # Placeholders for the registered 三语 template. The template's
                # zh/en copy mirrors ``_build_day90_survey_fields`` verbatim, so
                # those two recipients see exactly what they see today; ja is
                # the new variant the code branch (is_en / not is_en) can't
                # express. Card stats/sections and the HTML email body still
                # come from the code builder in the is_en language.
                context={
                    "person_name": person_name,
                    "company_name": company_name,
                },
                url=_DAY90_SURVEY_URL,
                card_subtitle=fields["card_subtitle"],
                card_stats=fields["card_stats"],
                card_sections=fields["card_sections"],
                card_actions=fields["card_actions"],
                related_resource_type="identity_person",
                related_resource_id=person.id,
            ),
            trace_id="day90_survey_notify",
        )
        person.day90_survey_sent_at = datetime.now(UTC)
        await self.db.flush()

    async def run_feishu_welcome_catchup(
        self, *, now: datetime | None = None,
    ) -> dict[str, Any]:
        """Deliver, on the hire's actual start date, the two welcome pieces
        that couldn't fire at the 24h-lead build: the Feishu welcome card and
        the HRBP / HR Ops CC copy.

        Both fail structurally in the 24h-lead window and self-heal here:

        * **Feishu card** — the account is minted ~a day early, so an at-build
          DM lands in an account the employee hasn't activated yet and is
          never seen (yihan.mo 2026-07-20). ``send_feishu_welcome_card`` now
          defers until the local start date; this sweep delivers it then.
        * **HR CC copy** — the employee welcome email fires ~24h early, before
          the hire's SF 2448/2450 relationships become the current slice, so
          ``notify_person`` couldn't CC HRBP / HR Ops. Once the relationship
          syncs (on the start date) this sends them the same welcome and
          stamps ``hr_cc`` (yihan.mo「当地 Ops 没收到抄送」).

        Selects hires who got the email (``it_info_sent_at`` within 3 days)
        but are missing EITHER ``"feishu"`` OR ``"hr_cc"`` in
        ``it_info_sent_channels``. Runs every 30 min; idempotent via the two
        channel stamps; the 3-day window retires hires who never get a Feishu
        account / never resolve an HR relationship (external contractors etc.).
        """
        now_dt = now or datetime.now(UTC)
        window_start = now_dt - timedelta(days=3)

        company_name = await get_platform_setting(
            "company.display_name", default="公司", session=self.db,
        )
        it_contact = await get_platform_setting(
            "it.contact_email", default="it@example.com", session=self.db,
        )
        sso_url = await get_platform_setting(
            "sso.login_url", default="https://helix.local/login", session=self.db,
        )
        portal_url = await get_platform_setting(
            "app.frontend_base_url",
            default="https://helix.local",
            session=self.db,
        )

        rows = list((
            await self.db.execute(
                select(IdentityPerson).where(
                    IdentityPerson.deleted_at.is_(None),
                    IdentityPerson.employment_status == "active",
                    IdentityPerson.hris_record_state != "purged",
                    IdentityPerson.work_email.isnot(None),
                    IdentityPerson.it_info_sent_at.isnot(None),
                    IdentityPerson.it_info_sent_at >= window_start,
                )
            )
        ).scalars().all())
        # JSON-contains filtered in Python — the window bounds this to a
        # handful of rows (recent hires only), not worth a dialect-
        # specific JSON_CONTAINS clause. A hire is a candidate if they still
        # need the Feishu card OR the HR CC copy. The HR-CC leg only applies
        # to hires who actually HAVE an HRBP / HR Ops relationship to CC —
        # otherwise a relationship-less hire (contractor) would stay a
        # candidate every tick for the whole 3-day window (and the copy would
        # be a no-op anyway). A relationship that lands mid-window flips them
        # in on the next tick (pending is recomputed from fresh rows).
        pending = []
        for p in rows:
            ch = p.it_info_sent_channels or []
            needs_card = "feishu" not in ch
            needs_hr_cc = (
                "hr_cc" not in ch
                and (
                    p.hris_hrbp_person_id is not None
                    or p.hris_hrops_person_id is not None
                )
            )
            if needs_card or needs_hr_cc:
                pending.append(p)

        summary: dict[str, Any] = {
            "candidates": len(pending),
            "sent": 0,
            "hr_cc_sent": 0,
            "awaiting_account": 0,
            "awaiting_hr_relationship": 0,
            "errors": [],
        }
        # Per-person commit: dispatch() delivers synchronously, so each stamp
        # must be durable BEFORE the next person is attempted. A single
        # end-of-loop commit meant any later failure rolled back the stamp of
        # an already-delivered DM → duplicate welcome cards on every tick. On
        # failure we roll back (poisoned session would otherwise break the
        # rest of the loop) and re-load each person by id since rollback
        # expires instances.
        pending_ids = [p.id for p in pending]
        for pid in pending_ids:
            person = await self.db.get(IdentityPerson, pid)
            if person is None:
                continue
            channels = person.it_info_sent_channels or []
            # ── Feishu welcome card ──
            if "feishu" not in channels:
                try:
                    sent = await self.send_feishu_welcome_card(
                        person,
                        company_name=company_name,
                        it_contact=it_contact,
                        sso_url=sso_url,
                        portal_url=portal_url,
                    )
                    if sent:
                        await self.db.commit()
                        summary["sent"] += 1
                    else:
                        summary["awaiting_account"] += 1
                except Exception as exc:  # noqa: BLE001
                    await self.db.rollback()
                    logger.exception(
                        "hire_day_feishu_welcome: card failed for %s", pid,
                    )
                    summary["errors"].append({
                        "person_id": str(pid),
                        "stage": "feishu_card",
                        "error": format_exception_for_log(exc, max_len=500),
                    })
                    # rollback expired the instance — reload before hr_cc
                    person = await self.db.get(IdentityPerson, pid)
                    if person is None:
                        continue
            # ── HRBP / HR Ops CC copy ──
            if "hr_cc" not in (person.it_info_sent_channels or []):
                try:
                    sent = await self._send_welcome_hr_copy(
                        person,
                        company_name=company_name,
                        it_contact=it_contact,
                        sso_url=sso_url,
                        portal_url=portal_url,
                    )
                    if sent:
                        await self.db.commit()
                        summary["hr_cc_sent"] += 1
                    else:
                        # No HRBP / HR Ops resolvable yet — relationship not
                        # synced (retry next tick) or the hire genuinely has
                        # none. Not stamped; the 3-day window retires it either
                        # way, mirroring ``awaiting_account`` for the card.
                        summary["awaiting_hr_relationship"] += 1
                except Exception as exc:  # noqa: BLE001
                    await self.db.rollback()
                    logger.exception(
                        "hire_day_feishu_welcome: HR CC copy failed for %s",
                        pid,
                    )
                    summary["errors"].append({
                        "person_id": str(pid),
                        "stage": "hr_cc",
                        "error": format_exception_for_log(exc, max_len=500),
                    })
        return summary

    async def send_feishu_welcome_card(
        self,
        person: IdentityPerson,
        *,
        company_name: str,
        it_contact: str,
        sso_url: str,
        portal_url: str,
    ) -> bool:
        """Send the rich Day-1 welcome card to the hire's Feishu DM.

        Returns False when the card can't be delivered YET — either the hire's
        local start date hasn't arrived, or their work_email doesn't resolve
        to a Feishu open_id (account not minted). The caller retries on the
        next sweep tick. On a resolvable account (start date reached) the card
        envelope is dispatched (feishu-only, ``hidden_from_inbox`` — the
        welcome already sits in the inbox as the email row) and ``"feishu"`` is
        stamped into ``it_info_sent_channels`` immediately: from here on the
        outbox row is the delivery/retry handle, and re-dispatching a second
        envelope on a transient send failure would risk double DMs instead.

        **Start-date gate (2026-07-20 fix, yihan.mo)**: with 24h-lead
        provisioning the Feishu account is minted ~a day BEFORE the hire
        starts, so ``resolve_open_id`` succeeds early — the old code then DM'd
        the card ~a day early to an account the employee hadn't activated yet,
        and it never surfaced when they first logged in on day-1 (the card was
        then stamped ``feishu`` and never re-sent → "飞书没收到指引"). Gate the
        card on the employee's LOCAL start date so it lands on day-1 to a live
        account. The 30-min ``run_feishu_welcome_catchup`` (3-day window)
        covers the ≤24h defer.
        """
        if not person.work_email:
            return False
        # Defer until the hire's local start date has arrived — a card DM'd
        # into a not-yet-activated account (the 24h-lead mint window) isn't
        # seen. ``lead_hours=0`` = strictly "before the local hire date".
        if is_future_hire(person, lead_hours=0):
            return False
        open_id = await self._resolve_feishu_open_id(person.work_email)
        if not open_id:
            return False

        # Language + reimbursement tool key off the labour-contract entity
        # (HR Ops 2026-07-15), mirroring the email path — see
        # ``send_person_it_info``.
        is_china = bool(person.is_china_employee)
        is_en = not is_china
        software_list = await self._software_list(person.id)
        # LANGUAGE axis for the card's prose fallbacks — a China-entity hire
        # reading English used to get 「你的团队」 / 「(待定)」 / 「待确认
        # (TBD)」 sitting inside an otherwise-English card.
        card_lang = resolve_lang(person)
        manager_name, hrbp_name = await self._manager_hrbp_names(
            person, is_en=is_en, lang=card_lang,
        )
        network_contact = await self._network_owner_name(
            is_en=is_en, lang=card_lang,
        )
        person_name = (
            localized_name(person.display_name, is_en=is_en)
            if person.display_name else person.work_email
        )
        department = person.primary_department or _fallback_team(is_en, card_lang)
        location = (
            person.work_location_name
            or person.work_location_code
            or _fallback_tbd(is_en, card_lang)
        )
        # The welcome note ships all three variants: ``CardSection.body`` is a
        # ``CardText`` now, so the note resolves for the reader alongside the
        # section title above it. It used to drop ``blessing_ja`` on the floor
        # ("no slot here" — true only while the body was a bare ``str``), which
        # put a Japanese heading over an English note.
        blessing_zh, blessing_en, blessing_ja = _blessing_variants(
            person,
            company_name=company_name,
            person_name=person_name,
            department=department,
            manager_name=manager_name,
        )
        hris_manual_title, hris_manual_url = _hris_manual_for(person)
        fields = _build_welcome_card_fields(
            # Reader's language (preference → country → entity). Distinct from
            # ``is_en``/``is_china``, which stay the HR-Ops content axis.
            lang=resolve_lang(person),
            is_en=is_en,
            is_china=is_china,
            company_name=company_name,
            person_name=person_name,
            person_email=person.work_email,
            department=department,
            location=location,
            manager_name=manager_name,
            hrbp_name=hrbp_name,
            sso_url=sso_url,
            portal_url=portal_url,
            network_contact=network_contact,
            it_contact=it_contact,
            seats=software_list,
            blessing=LocalizedText(
                zh=blessing_zh, en=blessing_en, ja=blessing_ja,
            ),
            hris_manual_title=hris_manual_title,
            hris_manual_url=hris_manual_url,
        )
        await self.notify.dispatch(
            NotificationEnvelope(
                recipient=person,
                title=fields["title"],
                body_text=fields["body_text"],
                channels=["feishu"],
                # Distinct key (no registered template → the envelope's
                # code-built copy ships verbatim): classifies into the
                # lifecycle category via the ``hire_day_it_info`` prefix
                # and picks the indigo card header.
                template_key="hire_day_it_info.feishu_card",
                url=sso_url,
                card_subtitle=fields["card_subtitle"],
                card_stats=fields["card_stats"],
                card_sections=fields["card_sections"],
                card_actions=fields["card_actions"],
                # The welcome already sits in the personal inbox as the
                # email envelope's row — this one is the Feishu push
                # record only.
                hidden_from_inbox=True,
                related_resource_type="identity_person",
                related_resource_id=person.id,
            ),
            trace_id="hire_day_feishu_welcome",
        )
        person.it_info_sent_channels = [
            *(person.it_info_sent_channels or []), "feishu",
        ]
        await self.db.flush()
        return True

    async def _send_welcome_hr_copy(
        self,
        person: IdentityPerson,
        *,
        company_name: str,
        it_contact: str,
        sso_url: str,
        portal_url: str,
    ) -> bool:
        """Email the hire's HRBP + HR Ops a copy of the SAME Day-1 welcome the
        hire got, then stamp ``hr_cc``.

        The 24h-lead welcome email couldn't CC them because the hire's SF
        2448/2450 relationships were still future-dated at build time
        (``notify_person`` left ``hr_cc`` off). Once the relationship syncs —
        on the start date — this delivers the copy so 当地 Ops (= the hire's
        HR Ops, relationshipType 2450) sees the welcome (yihan.mo 2026-07-20).
        Content renders in the HIRE's language via ``_render_welcome_payload``
        so it's byte-identical to what the hire received. Sent as a separate
        email per recipient (mirrors the account-created card fan-out), so a
        recipient who fills two roles gets one copy. ``hidden_from_inbox`` — a
        courtesy email, not an HR Helix-inbox action item.

        Returns True (delivered + stamped) or False (no HRBP / HR Ops
        resolvable yet — relationship not synced, or the hire genuinely has
        none; the caller retries within the 3-day catchup window).
        """
        ids = [
            pid for pid in (
                person.hris_hrbp_person_id, person.hris_hrops_person_id,
            )
            if pid is not None
        ]
        if not ids:
            return False
        rows = (await self.db.execute(
            select(IdentityPerson).where(
                IdentityPerson.id.in_(ids),
                IdentityPerson.deleted_at.is_(None),
                IdentityPerson.employment_status != "left",
            )
        )).scalars().all()
        # Dedup by id, exclude self and empty mailboxes (mirrors
        # ``_welcome_cc_emails``' hygiene).
        recipients = {
            r.id: r for r in rows if r.id != person.id and r.work_email
        }
        if not recipients:
            return False

        payload = await self._render_welcome_payload(
            person,
            company_name=company_name,
            it_contact=it_contact,
            sso_url=sso_url,
            portal_url=portal_url,
        )
        is_en = payload["is_en"]
        hire_name = (
            localized_name(person.display_name, is_en=is_en)
            if person.display_name else (person.work_email or "")
        )
        # Prefix so HR sees at a glance whose welcome this is (it's a copy,
        # not their own onboarding).
        prefix = (
            f"[New hire] {hire_name} · " if is_en
            else f"[新人入职] {hire_name} · "
        )
        title = prefix + payload["title"]
        for hr in recipients.values():
            await self.notify.dispatch(
                NotificationEnvelope(
                    recipient=hr,
                    title=title,
                    body_text=payload["body_text"],
                    body_html=payload["body_html"],
                    channels=["email"],
                    # Distinct, unregistered template_key → the code-built body
                    # ships verbatim (same pattern as the feishu card).
                    template_key="hire_day_it_info.hr_cc",
                    hidden_from_inbox=True,
                    related_resource_type="identity_person",
                    related_resource_id=person.id,
                ),
                trace_id="hire_day_hr_cc",
            )
        person.it_info_sent_channels = [
            *(person.it_info_sent_channels or []), "hr_cc",
        ]
        await self.db.flush()
        return True

    async def _resolve_feishu_open_id(self, email: str) -> str | None:
        """Email → Feishu open_id, or None while the account doesn't
        exist yet (or Feishu is unreachable/unconfigured — the sweep's
        next tick retries either way). Never raises."""
        from app.approval.feishu_client import FeishuApprovalClient
        from app.integrations.compat import get_feishu_config

        try:
            cfg = await get_feishu_config(session=self.db)
        except AppException:
            return None
        client = FeishuApprovalClient(
            app_id=str(cfg.get("app_id") or ""),
            app_secret=str(cfg.get("app_secret") or ""),
        )
        try:
            return await client.resolve_open_id_by_email(email)
        except AppException:
            return None
        except Exception as exc:  # noqa: BLE001 — network flakiness
            logger.warning(
                "hire_day: feishu open_id lookup failed for %s: %s",
                email, exc,
            )
            return None

    async def _manager_hrbp_names(
        self, person: IdentityPerson, *, is_en: bool, lang: str | None = None,
    ) -> tuple[str, str]:
        """Resolve the new hire's direct manager + HRBP names for the welcome
        message, language-matched: the Latin form for English recipients, the
        CJK form for Chinese ones (HR Ops 2026-07-15 — a bilingual
        ``display_name`` like "孟凡生 Jason Meng" must not render its Chinese
        half in an English email). Friendly fallbacks when the relationship
        isn't synced yet (a brand-new hire's HRBP/manager may not have
        landed) — the fallback labels are already bilingual, so they are
        returned verbatim rather than split. ``hris_hrbp_person_id`` ← SF
        relationshipType 2448.
        """
        ids = [
            pid for pid in (person.manager_person_id, person.hris_hrbp_person_id)
            if pid is not None
        ]
        names: dict = {}
        if ids:
            rows = (await self.db.execute(
                select(IdentityPerson.id, IdentityPerson.display_name)
                .where(IdentityPerson.id.in_(ids))
            )).all()
            names = {r[0]: r[1] for r in rows}
        manager_dn = names.get(person.manager_person_id)
        hrbp_dn = names.get(person.hris_hrbp_person_id)
        manager = (
            localized_name(manager_dn, is_en=is_en)
            if manager_dn else _fallback_manager_tbd(is_en, lang)
        )
        # NOTE: ``localized_name``'s Latin-vs-CJK NAME-FORM choice is still on
        # the entity axis and is shared with the email path. It is a genuine
        # product question for ``ja`` (which form does a Japanese reader want
        # for a Chinese colleague's name?), not a mechanical fix, so it is
        # deliberately left alone here. Only the PROSE fallbacks moved.
        # Language axis, like its sibling one line up. This sat right beside
        # ``_fallback_manager_tbd(is_en, lang)`` and did not move with it: a
        # bilingual literal has no ja form at all, so a Japanese reader with an
        # unsynced HRBP got 「HR 团队 / HR team」 — Simplified Chinese — inside
        # an otherwise-Japanese card.
        hrbp = (
            localized_name(hrbp_dn, is_en=is_en) if hrbp_dn
            else _pick_fallback(_FALLBACK_HR_TEAM, is_en, lang)
        )
        return manager, hrbp

    async def _network_owner_name(
        self, *, is_en: bool, lang: str | None = None,
    ) -> str:
        """网络负责人 display name for the welcome "who to reach" section,
        language-matched (Latin for EN, CJK for zh).

        Prefer the designated person record — ``network.owner_person_id``
        (seeded to 孟凡生 by migration 267, editable in admin Settings) — so
        the name has a real ``display_name`` we can split per language. Fall
        back to the free-text ``it.network_contact`` setting (returned
        verbatim — it's whatever an admin typed), then a generic label. This
        also closes a latent source mismatch: the welcome used to read only
        ``it.network_contact``, which the repo never seeds, so an install
        without an admin override rendered the literal default "网络负责人".
        """
        owner_id = await get_platform_setting(
            "network.owner_person_id", default=None, session=self.db,
        )
        if owner_id:
            owner = await self.db.get(IdentityPerson, str(owner_id))
            if (
                owner is not None
                and owner.employment_status == "active"
                and owner.display_name
            ):
                return localized_name(owner.display_name, is_en=is_en)
        configured = await get_platform_setting(
            "it.network_contact", default=None, session=self.db,
        )
        if configured:
            # Whatever an admin typed — verbatim, in whatever language.
            return str(configured)
        # The generic label. Was a hardcoded Chinese default, so an install
        # with no override put 「网络负责人」 inside an English or Japanese
        # card — the same axis mistake as its neighbours, one function over.
        return _pick_fallback(_FALLBACK_NETWORK_OWNER, is_en, lang)

    async def _welcome_cc_emails(self, person: IdentityPerson) -> _WelcomeCc:
        """Work emails to CC on the Day-1 welcome email, plus whether the HR
        leg of that list actually resolved.

        Recipients (2026-08-24 — deliberately the same operational audience
        the account-created welcome mail reaches via
        ``account_provisioner._resolve_welcome_cc_additions``, so both
        new-hire mails land in front of the same people):

        * the hire's **HRBP** (SF relationshipType 2448) and **HR Ops**
          (2450) — HR Ops request 2026-07-15;
        * every active **asset administrator of the hire's local office**,
          resolved with the canonical asset-region rule — they prepare the
          day-one hardware, so they need the same day-1 signal HR gets; and
        * the **IT contact(s)** — 孟凡生 by default, read from the same
          ``google.account_provisioning.created_notify_emails`` setting the
          account-created mail uses so the contact is configured in one place.

        Deduped, self excluded, empties dropped, departed / soft-deleted
        people excluded (a stale relationship pointing at an HRBP who already
        left must not CC a dead mailbox), every address gated on the fail-OPEN
        corporate-domain guard (a mis-typed setting must not mail the new
        hire's details outside the company), and hard-capped at
        ``_WELCOME_CC_CAP`` with a warning naming what was dropped.

        ``hr_covered`` is True only when the HRBP / HR Ops leg produced at
        least one address. It must gate the ``hr_cc`` stamp instead of "CC
        list non-empty": the IT contact comes from a *defaulted* setting, so
        the list is now non-empty for every hire, which would permanently
        disarm the catchup that delivers HR's copy when a pre-start hire's SF
        relationship syncs late (yihan.mo 2026-07-20).

        Feishu-card delivery is a personal DM and carries no CC.
        """
        self_email = (person.work_email or "").strip().lower()
        raw: list[str] = []
        hr_covered = False

        # ── HRBP + HR Ops ────────────────────────────────────────────
        # Kept as its own query (rather than reading it off the coordinator
        # below) for two reasons: it yields ``hr_covered``, and its
        # "not left" predicate is wider than the coordinator's "active" —
        # an HRBP working their notice period (``leaver_pending``) is still
        # the hire's HRBP and was already being CC'd before this change.
        hr_ids = [
            pid for pid in (person.hris_hrbp_person_id, person.hris_hrops_person_id)
            if pid is not None
        ]
        if hr_ids:
            rows = (await self.db.execute(
                select(IdentityPerson.work_email).where(
                    IdentityPerson.id.in_(hr_ids),
                    IdentityPerson.deleted_at.is_(None),
                    IdentityPerson.employment_status != "left",
                )
            )).all()
            for (email,) in rows:
                e = (email or "").strip()
                if e and e.lower() != self_email:
                    raw.append(e)
                    hr_covered = True

        # ── Local asset admins (and HRBP / HR Ops again — deduped below) ──
        # Routed through the coordinator instead of importing the asset
        # domain's office-admin rule here: cross-domain resolution lives in
        # app/core/coordinators (architecture rule 3), and the
        # account-created mail reads the very same helper.
        from app.core.coordinators.new_hire_recipient_resolver import (
            resolve_new_hire_operational_recipients,
        )

        for recipient in await resolve_new_hire_operational_recipients(
            self.db, person,
        ):
            work_email = (recipient.work_email or "").strip()
            if work_email and recipient.id != person.id:
                raw.append(work_email)

        # ── IT contact (孟凡生 by default) ───────────────────────────
        it_contacts = await get_platform_setting(
            "google.account_provisioning.created_notify_emails",
            default=["fansheng.meng@pingcap.com"], session=self.db,
        )
        if isinstance(it_contacts, list):
            raw.extend(
                str(e).strip() for e in it_contacts if str(e).strip()
            )

        corporate_domains = await get_corporate_email_domains(self.db)
        seen: set[str] = set()
        out: list[str] = []
        external = 0
        for candidate in raw:
            addr = candidate.strip().lower()
            if not addr or addr == self_email or addr in seen:
                continue
            if not email_is_corporate(addr, corporate_domains):
                external += 1
                continue
            seen.add(addr)
            out.append(addr)
        if external:
            logger.warning(
                "hire_day: dropped %d non-corporate welcome CC address(es) "
                "for person=%s", external, person.id,
            )
        if len(out) > _WELCOME_CC_CAP:
            logger.warning(
                "hire_day: welcome CC capped at %d for person=%s — dropped %s",
                _WELCOME_CC_CAP, person.id, out[_WELCOME_CC_CAP:],
            )
        return _WelcomeCc(out[:_WELCOME_CC_CAP], hr_covered)

    async def _software_list(self, person_id) -> list[str]:
        """Names of software the new hire has been provisioned into.

        Pre-launch hardening 2026-05-04: ProvisioningTask doesn't
        carry a ``software_name`` column — the previous getattr
        fallback always returned ``software_id`` (a UUID), so the
        hire-day welcome email rendered raw UUIDs in the "你已开通"
        section. Resolve via AssetSoftwareCatalog name.
        """
        from app.domains.asset.models import AssetSoftwareCatalog

        rows = (
            await self.db.execute(
                select(ProvisioningTask.software_id)
                .where(
                    ProvisioningTask.user_id == person_id,
                    ProvisioningTask.source == "onboarding",
                    ProvisioningTask.software_id.isnot(None),
                )
            )
        ).all()
        software_ids = [r[0] for r in rows if r[0]]
        if not software_ids:
            return []
        catalog_rows = (
            await self.db.execute(
                select(
                    AssetSoftwareCatalog.id,
                    AssetSoftwareCatalog.name,
                ).where(
                    AssetSoftwareCatalog.id.in_(software_ids),
                )
            )
        ).all()
        name_by_id = {row[0]: row[1] for row in catalog_rows}
        return [
            (name_by_id.get(sid) or str(sid))
            for sid in software_ids
        ]

    async def _hardware_list(self, person_id) -> list[str]:
        # Only real "devices to pick up": ``assigned_to`` is deliberately
        # preserved on soft-deleted and held_by_leaver/held_by_office rows
        # (audit/attribution), so without these filters a rehire — or any
        # person reusing a Person row — would see a stale/departed device
        # listed in their Day-1 welcome email.
        rows = list((
            await self.db.execute(
                select(AssetItem)
                .where(
                    AssetItem.assigned_to == person_id,
                    AssetItem.deleted_at.is_(None),
                    AssetItem.status.in_(ASSET_IN_USE_STATUSES),
                )
            )
        ).scalars().all())
        return [
            (a.asset_name or a.asset_tag or str(a.id))
            for a in rows
        ]


def _format_list(items: list[str]) -> str:
    return "\n".join(f"  · {s}" for s in items) if items else ""


# Language-matched fallbacks for the Day-1 welcome when a pre-start hire's
# HRIS data hasn't landed yet (department / manager / work location are
# future-dated in SF and the effective-dated enrich couldn't fill them). An
# English welcome must NEVER render the Chinese placeholder — the 2026-08-03
# NA/EMEA report (``Team: 你的团队``, ``Location: (待定)``, ``Manager: 待确认
# (TBD)`` in English emails). The ``_enrich_prestart_hire_department`` pull
# fixes the common case at source; these keep the rare genuine-gap render
# correct in the recipient's own language.
# ``lang`` is the LANGUAGE axis (``resolve_lang``); ``is_en`` is the labour-
# ENTITY axis. These strings are prose the reader reads, so they follow the
# language — pass ``lang`` from any card path. ``is_en`` remains the fallback
# for the email path, which is still entity-keyed end to end.
#
# Getting this wrong is not hypothetical: a China-entity hire reading English
# with an unsynced HRBP used to see 「待确认 (TBD)」 sitting inside two
# otherwise-English sections.
_FALLBACK_TEAM = {"zh": "你的团队", "en": "Your team", "ja": "あなたのチーム"}
_FALLBACK_TBD = {"zh": "(待定)", "en": "TBD", "ja": "(未定)"}
_FALLBACK_NETWORK_OWNER = {
    "zh": "网络负责人", "en": "the network owner", "ja": "ネットワーク担当者",
}
_FALLBACK_HR_TEAM = {
    "zh": "HR 团队", "en": "the HR team", "ja": "HR チーム",
}
_FALLBACK_MANAGER_TBD = {
    "zh": "待确认 (TBD)", "en": "To be confirmed (TBD)", "ja": "確認中 (TBD)",
}


def _pick_fallback(
    table: dict[str, str], is_en: bool, lang: str | None,
) -> str:
    if lang is not None:
        return table[normalize_lang(lang) or DEFAULT_LANG]
    return table["en"] if is_en else table["zh"]


def _fallback_team(is_en: bool, lang: str | None = None) -> str:
    return _pick_fallback(_FALLBACK_TEAM, is_en, lang)


def _fallback_tbd(is_en: bool, lang: str | None = None) -> str:
    return _pick_fallback(_FALLBACK_TBD, is_en, lang)


def _fallback_manager_tbd(is_en: bool, lang: str | None = None) -> str:
    return _pick_fallback(_FALLBACK_MANAGER_TBD, is_en, lang)


def _render_body(
    *,
    person: IdentityPerson,
    company_name: str,
    it_contact: str,
    sso_url: str,
    portal_url: str,
    software_list: list[str],
    hardware_list: list[str],
    is_en: bool,
) -> str:
    """Plain-text fallback for the rich HTML welcome. Language-matched to
    the recipient (English unless explicitly zh) so the text part never
    disagrees with the HTML part."""
    name = person.display_name or person.work_email
    if is_en:
        return (
            f"Hi {name},\n\n"
            f"Welcome to {company_name}! Here are your IT accounts and the "
            "basics for your first day.\n\n"
            "[Day-1 handbook]\n"
            f"  · {_day1_handbook_url(not is_en)}\n\n"
            "[Account]\n"
            f"  · Work email: {person.work_email}\n"
            f"  · SSO sign-in: {sso_url}\n"
            "  · First sign-in: use Google SSO\n\n"
            "[Software assigned]\n"
            f"{_format_list(software_list) or '  · None yet (IT will provision per team policy)'}\n\n"
            "[Devices to pick up]\n"
            f"{_format_list(hardware_list) or '  · None'}\n\n"
            "[Platform]\n"
            f"  · {portal_url}\n\n"
            "[Questions]\n"
            f"  · Contact IT: {it_contact}\n\n"
            "— Helix IT Platform"
        )
    return (
        f"Hi {name}:\n\n"
        f"欢迎加入 {company_name}!这是今天入职日你的 IT 账号和基础信息。\n\n"
        "【入职第一天手册】\n"
        f"  · {_day1_handbook_url(not is_en)}\n\n"
        "【账号】\n"
        f"  · 工作邮箱: {person.work_email}\n"
        f"  · SSO 登录入口: {sso_url}\n"
        f"  · 首次登录请使用 Google SSO\n\n"
        "【已分配的软件】\n"
        f"{_format_list(software_list) or '  · 暂无(IT 将根据部门策略继续发放)'}\n\n"
        "【待领取设备】\n"
        f"{_format_list(hardware_list) or '  · 暂无'}\n\n"
        "【平台入口】\n"
        f"  · {portal_url}\n\n"
        "【如有疑问】\n"
        f"  · 联系 IT 团队: {it_contact}\n\n"
        "— Helix IT 平台"
    )


# ── Warm onboarding note ─────────────────────────────────────────
# Fixed copy provided by HR Ops (2026-07-15), signed off as coming from
# PingCAP (the company), NOT "the Helix IT platform". One note for every
# hire — replaces the former per-hire randomised pool (HR gave a single
# canonical text). ``{person_name}`` is the recipient's language-matched
# name; ``{company_name}`` is both the greeting subject and the closing
# signature (落款). ``{department}``/``{manager_name}`` are unused now but
# accepted by the formatter so call sites need no change.
_WELCOME_NOTE_ZH = (
    "Hi {person_name},\n\n"
    "新的环境总会带来一些新鲜感,也难免会有一点陌生和挑战。别担心,在这里,"
    "你并不是一个人在摸索。遇到问题时,身边的同事都会很乐意与你一起寻找答案。\n\n"
    "希望你能在 {company_name} 收获成长,也收获信任与友谊;期待这里不仅成为你"
    "工作的地方,也能留下许多值得回忆的时光。\n\n"
    "再次欢迎你的加入,祝你开启一段愉快而充实的新旅程!✨\n\n"
    "{company_name}"
)
_WELCOME_NOTE_EN = (
    "{person_name}, welcome to the {company_name} family!\n\n"
    "We're so excited to have you with us. As you begin this new chapter, "
    "remember that you're never on your own — everyone at {company_name} is "
    "here to support you and help you along the way.\n\n"
    "Wishing you a wonderful journey filled with learning, growth, success, "
    "and memorable moments. Welcome aboard! 🌟\n\n"
    "{company_name}"
)
_WELCOME_NOTE_JA = (
    "{person_name} さん、{company_name} へようこそ。\n\n"
    "新しい環境には、新鮮な発見とともに、少しの戸惑いもあるかと思います。"
    "けれども、ここでは決してお一人で悩む必要はありません。"
    "困ったことがあれば、周りのメンバーが一緒に答えを探します。\n\n"
    "{company_name} での日々が、成長はもちろん、信頼できる仲間との出会いに"
    "つながることを願っています。働く場所であると同時に、"
    "心に残る時間を過ごせる場所となりますように。\n\n"
    "あらためて、ご入社おめでとうございます。"
    "実り多い新しい一歩となりますよう、心よりお祈りしております。✨\n\n"
    "{company_name}"
)


def _blessing_variants(
    person: IdentityPerson,
    *,
    company_name: str,
    person_name: str,
    department: str,
    manager_name: str,
) -> tuple[str, str, str]:
    """(zh, en, ja) welcome note. Fixed HR-provided copy (2026-07-15) — the
    SAME text for every hire, so the email and the trailing Feishu card carry
    an identical note. ``person`` / ``department`` / ``manager_name`` are kept
    in the signature for call-site compatibility; the note no longer uses them.

    All three variants are handed to the notification context as the
    ``blessing_zh`` / ``blessing_en`` / ``blessing_ja`` locale-fragment triple,
    so each language's ``hire_day_it_info`` template interpolates its own."""
    fmt = {
        "company_name": company_name,
        "person_name": person_name,
        "department": department,
        "manager_name": manager_name,
    }
    return (
        _WELCOME_NOTE_ZH.format(**fmt),
        _WELCOME_NOTE_EN.format(**fmt),
        _WELCOME_NOTE_JA.format(**fmt),
    )


# Day-1 handbook (Feishu wiki) — the step-by-step first-day guide linked
# from the IT section of the welcome email. TWO regional editions, per HR's
# "Onboarding-IT" wiki doc (中文版=中国区 / 英文版=海外区域): China-entity
# hires read the 中国区 guide, everyone else the Global one. Picked by
# ``is_china`` (the labour-contract entity) — the same HR-Ops content axis
# that already drives the welcome language and the reimbursement tool.
# Before 2026-08-25 both editions pointed at the Global (English) guide, so
# every China hire was handed an overseas-only handbook.
_DAY1_HANDBOOK_URL = "https://pingcap.feishu.cn/wiki/Tu9AwCr2ui4fRtkQCdScG99Unaf"
_DAY1_HANDBOOK_URL_CN = "https://pingcap.feishu.cn/wiki/wikcnKhelzg1rdNVKFUmMXzfvoW"


def _day1_handbook_url(is_china: bool) -> str:
    """The Day-1 IT handbook for this hire's region."""
    return _DAY1_HANDBOOK_URL_CN if is_china else _DAY1_HANDBOOK_URL

# Mandatory onboarding security course/exam (TiLearning).
_SECURITY_EXAM_URL = (
    "https://learning.pingcap.com/o2o/#/project/detail/1935586295680002050"
)

# Day-1 FAQ quick guides (Feishu wiki) — merged from the HR team's own
# Day-1 welcome message (2026-07-03) so a new hire gets ONE consolidated
# welcome instead of two overlapping ones. Same rows are seeded into the
# 制度中心 (platform_policy_notices) for later lookup.
_FAQ_FIRST_MEETING_URL = "https://pingcap.feishu.cn/wiki/Uq6zw8hDhiq7iak5hXvcG32Xnlb"
_FAQ_CALENDAR_URL = "https://pingcap.feishu.cn/wiki/DzcVwON2qiz7ORkmNHgc80yKnLh"
_FAQ_ORG_CHART_URL = "https://pingcap.feishu.cn/wiki/TOZrwz4pliLR4YkM9Z4cFj6VnHg"
_FAQ_DOC_COLLAB_URL = "https://pingcap.feishu.cn/wiki/BhmWwJOsYimzHNkaEqbcBCX6nQt"
_FAQ_ACCESS_REQUEST_URL = "https://pingcap.feishu.cn/wiki/Yk0ZwWfK0iACE3kso03c9HMknub"

# Day-90 onboarding-experience survey (Feishu form) — pushed by
# ``hr.day90_survey_notify`` when a hire reaches 90 days (their local
# calendar), Feishu card + email.
_DAY90_SURVEY_URL = (
    "https://pingcap.feishu.cn/share/base/form/shrcn9oVfSQnbMIEkNPUA30H9Hg"
    "?chunked=false"
)

# Day-1 onboarding-experience survey (Feishu form) — filled ON the first
# day (distinct from the 90-day survey above); sits in the welcome
# card's "today" checklist. Provided by HR 2026-07-03.
_DAY1_SURVEY_URL = (
    "https://pingcap.feishu.cn/share/base/form/shrcnx5vNqWfJ6yUgng9A1C444f"
)

# Feishu IT Help Desk applinks — the "快捷链接 / Quick Link" from HR's
# "Onboarding-IT" wiki doc, one per region. Replaces the former single
# ``_IT_HELPDESK_URL`` (client/helpdesk/open?id=6939814243826991108, the
# prod ``feishu-helpdesk-it`` integration), which was region-blind and sent
# overseas hires into the China desk. Same ``is_china`` axis as the
# handbook above.
_IT_HELPDESK_APPLINK_CN = "https://applink.feishu.cn/T8TQpyK5mVGj"
_IT_HELPDESK_APPLINK_GLOBAL = "https://applink.feishu.cn/T8UNQO6Y5TAl"


def _it_helpdesk_applink(is_china: bool) -> str:
    """The Feishu IT Help Desk quick link for this hire's region."""
    return _IT_HELPDESK_APPLINK_CN if is_china else _IT_HELPDESK_APPLINK_GLOBAL

# Feishu HR helpdesk applink — provided by HR (qingqing.he) review,
# 2026-07-03.
_HR_HELPDESK_URL = "https://applink.feishu.cn/T97IDmdiMsag"

# China-only HR One Pager (五险一金 / 薪酬福利 / 期权 / 商保 …) — appended
# after the leave/HRIS line for the CHINESE version only (HR request
# 2026-07-15). Non-China (English) hires don't use it, and it's a
# China-benefits doc. The shared link's ``?source=…`` tracking suffix is
# dropped — the bare wiki URL is the canonical, stable target.
_HR_ONE_PAGER_URL = "https://pingcap.feishu.cn/wiki/RwZswUtcPiZG3zkm9zDc77egn0g"

# HRIS user manuals — selected by the hire's WORK COUNTRY (HR rule
# 2026-07-03; the same link serves both languages). Keys are the
# ``_hris_manual_for`` buckets, values are (doc title, url).
_HRIS_MANUAL_CN = (
    "HRIS 员工用户手册 - 中国",
    "https://pingcap.feishu.cn/wiki/LRXwwgun4iGIZakKrSlcJut7nmd",
)
_HRIS_MANUAL_JPMYSGUS = (
    "HRIS Employee User Guide - JP/MY/SG/US",
    "https://pingcap.feishu.cn/wiki/CcjdwsESoiQ9Kvka94tc3HflnBc",
)
_HRIS_MANUAL_EOR = (
    "HRIS EOR Staff User Guide - for all staff under EOR",
    "https://pingcap.feishu.cn/wiki/V8IKwzG7Ui8rbRkSTjOcTRbtnOg"
    "?fromScene=spaceOverview",
)


def _hris_manual_for(person: IdentityPerson) -> tuple[str, str]:
    """(title, url) of the HRIS manual for this hire's work country.

    ``residence_country`` (ISO-2 derived from work_location_code — the
    geo axis) first, ``country`` (legal entity) as fallback. CN and
    unknown → the China manual (the dominant population; an unknown
    country is a data gap, not "other countries"); US/JP/MY/SG → the
    JP/MY/SG/US guide; everything else → the EOR staff guide.
    """
    c = (person.residence_country or person.country or "").strip().upper()[:2]
    if c in ("", "CN"):
        return _HRIS_MANUAL_CN
    if c in ("US", "JP", "MY", "SG"):
        return _HRIS_MANUAL_JPMYSGUS
    return _HRIS_MANUAL_EOR


def _doc_link(title: str, url: str, color: str) -> str:
    """Render a doc reference as a link when the URL is known, else as
    plain bold text (no dead links while HR hasn't provided the target)."""
    esc = _html.escape(title)
    if url:
        return f'<a href="{url}" style="color:{color};font-weight:700;">{esc}</a>'
    return f'<b>{esc}</b>'


def _email_bullet(html_text: str, mark: str = "#2563EB") -> str:
    return (
        '<div style="margin:7px 0;color:#334155;font-size:14px;line-height:1.65;">'
        f'<span style="color:{mark};">▸</span>&nbsp; {html_text}</div>'
    )


def _email_subitem(html_text: str) -> str:
    """Indented continuation line under an ``_email_bullet``.

    Used for the IT Help Desk entry points, which carry their own 🔹 marker
    — a second ▸ in front would read as a nested bullet that is not one.
    """
    return (
        '<div style="margin:4px 0 4px 22px;color:#475569;font-size:13px;'
        'line-height:1.6;">'
        f'{html_text}</div>'
    )


def _render_onboarding_html(
    *,
    is_en: bool,
    is_china: bool,
    company_name: str,
    person_name: str,
    person_email: str,
    department: str,
    location: str,
    manager_name: str,
    hrbp_name: str,
    sso_url: str,
    portal_url: str,
    network_contact: str,
    it_contact: str,
    seats: list[str],
    blessing: str,
    hris_manual_title: str = _HRIS_MANUAL_CN[0],
    hris_manual_url: str = _HRIS_MANUAL_CN[1],
) -> str:
    """Render the rich, sectioned Day-1 onboarding email as HTML.

    Sections: Basics / IT / HR / Finance / Legal·Compliance·Security / a
    warm note. Email-safe (inline styles only, table layout, no external
    assets). ``blessing`` is injected verbatim (already language-matched).
    ``is_china`` selects the reimbursement tool (汇联易 vs Navan) independent
    of the email language.
    """
    e = _html.escape
    company_e, name_e = e(company_name), e(person_name)
    # ``portal_url`` no longer appears in the body: the IT block's Helix
    # self-service pitch went away with the 2026-08-25 rewrite. The CTA
    # button still points at ``sso_url``.
    sso_e = e(sso_url)

    # Reimbursement tool by employment entity (HR Ops 2026-07-15): China →
    # 汇联易 (its Feishu Workplace app is named "Helios"); everyone else →
    # Navan. The finance-desk pointer follows the email language.
    fin_tool = ("汇联易" if not is_en else "Huilianyi") if is_china else "Navan"
    fin_app = "Helios" if is_china else "Navan"
    if is_en:
        fin_workplace = "Feishu Workplace" if is_china else "Lark Workplace"
        fin1 = (
            f"Submit travel / daily / procurement reimbursements "
            f"{'in' if is_china else 'via'} <b>{fin_tool}</b> "
            f"({fin_workplace} &gt; Featured Apps &gt; {fin_app})"
        )
        fin2 = (
            "Questions? Check the "
            + ("Feishu Finance Desk" if is_china else "Lark Finance Desk")
        )
        laptop_line = (
            "<b>Pickup your Laptop</b>: If you have not yet collected or received "
            "your Laptop, please reach out to your local HR Ops contact to arrange "
            "pickup. Note: For employees who are not located in China / Japan / "
            "Malaysia / Singapore / US, please follow IT policy to buy and reimburse."
        )
        course_line = (
            "To access the <b>New Employee Orientation Online Course</b>, please go "
            "directly to <b>Lark &gt; Workplace &gt; TiLearning</b>."
        )
        # EN-only HRIS guidance (HR Ops 2026-07-15): non-China hires DO use
        # HRIS — pointed at their own country's guide (the China manual line
        # is CN-version-only). One self-select line covering both buckets.
        hris_guides_line = (
            "For JP/MY/SG/US, please refer to "
            + _doc_link(_HRIS_MANUAL_JPMYSGUS[0], _HRIS_MANUAL_JPMYSGUS[1], "#10B981")
            + " for completing HR related process and requests in "
            "<b>HRIS (SuccessFactors)</b>. For colleagues working in other "
            "countries, please refer to "
            + _doc_link(_HRIS_MANUAL_EOR[0], _HRIS_MANUAL_EOR[1], "#10B981")
            + "."
        )
    else:
        fin1 = (
            f"差旅 / 日常 / 采购报销在 <b>{fin_tool}</b> 提交"
            f"(飞书工作台 - Featured Apps - {fin_app})"
        )
        fin2 = "有疑问请至飞书财务工作台查询"
        laptop_line = (
            "<b>笔记本电脑领取</b>:如尚未领或收到笔记本电脑,请联系当地 HR Ops 联系人领取。"
        )
        course_line = "新员工培训课程入口,请直接从 <b>飞书 → 工作台 → TiLearning</b> 进入。"
        one_pager_line = (
            "你关心的 HR 问题都在这里(五险一金、薪酬福利等):"
            f'<a href="{_HR_ONE_PAGER_URL}" style="color:#10B981;font-weight:700;">'
            "PingCAP 入职 One Pager</a>"
        )

    # IT section — regional targets from HR's "Onboarding-IT" wiki doc.
    # ``is_china`` is the entity axis; ``is_en`` is only its language shadow.
    handbook_e = e(_day1_handbook_url(is_china))
    helpdesk_e = e(_it_helpdesk_applink(is_china))

    if is_en:
        labels = {
            "kicker": "PINGCAP IT &middot; WELCOME ABOARD",
            "h1": f"🎉 Welcome to {company_e}!",
            "sub": f"{name_e}, we&rsquo;re so glad you&rsquo;re here! This guide has everything you need for your first day",
            "intro": 'Three minutes now makes day one a lot smoother 🙂 It&rsquo;s laid out by <b style="color:#0F172A;">Basics &middot; IT &middot; HR &middot; Finance &middot; Legal</b> — jump to whatever you need.',
            "s_basic": "👤 The basics",
            "l_name": "Name", "l_dept": "Team", "l_loc": "Location",
            "l_email": "Work email", "l_mgr": "Manager", "l_hrbp": "HRBP",
            "s_it": "💻 IT &middot; Accounts &amp; everyday tools",
            "it_intro": "To help you get started smoothly, we&rsquo;ve put together some essential IT guides related to your daily work. Please take a moment to review them carefully:",
            "it_mustread": "Must-Read on Your First Day",
            "it_guide": f'<a href="{handbook_e}" style="color:#2563EB;font-weight:700;">Day 1 - IT Guide at PingCAP (Global)</a> — the PingCAP IT Guide for your region',
            "it_hw": "Laptop Pickup",
            "it_desk": "Your One-Stop IT Service Portal",
            "desk_line": f"<b>IT Help Desk</b> is always here for you ({e(it_contact)}).",
            "desk1": "🔹 <b>Lark/Feishu Search</b>: simply type &ldquo;IT Help Desk&rdquo; for instant access",
            "desk2": "🔹 <b>Lark/Feishu Path</b>: Workbench → Service Desk → IT Help Desk",
            "desk3": f'🔹 <b>Quick Link</b>: click <a href="{helpdesk_e}" style="color:#2563EB;font-weight:700;">here</a> to enter directly via Lark/Feishu',
            "s_hr": "🧑‍💼 HR",
            "hr_desk": f'For anything HR, the <a href="{_HR_HELPDESK_URL}" style="color:#10B981;font-weight:700;">HR helpdesk</a> is your first stop',
            "hr_leave": f'View and request leave / time-off in <b>HRIS (SuccessFactors)</b>: {_doc_link(hris_manual_title, hris_manual_url, "#10B981")}',
            "s_survey": "📝 New Hire Onboarding Survey",
            "survey_line": f'Take a few minutes today for the <a href="{_DAY1_SURVEY_URL}" style="color:#0EA5E9;font-weight:700;">New Hire Onboarding Survey</a> — your honest feedback keeps making onboarding better for everyone who joins after you.',
            "s_fin": "💰 Finance &middot; Reimbursement",
            "s_legal": "⚖️ Legal &middot; Compliance &middot; Security",
            "legal0": f'🔐 <b>Required for onboarding</b>: complete the <a href="{_SECURITY_EXAM_URL}" style="color:#8B5CF6;">security training &amp; exam</a> on the learning platform — please pass it on time',
            "legal1": "Please honour the <b>NDA</b> you signed and our information-security policies",
            "legal2": "Need <b>SOC2</b> materials (customer audits, compliance proof, etc.)? Start a &ldquo;SOC2 request&rdquo; approval in the <b>Feishu workspace</b> — no email needed",
            "legal3": "Contract review and IP matters go through the Feishu legal flow",
            "s_bless": "💌 A note for you",
            "cta": "Sign in to Helix →",
            "footer": f"Sent by the {company_e} IT team via the Helix platform. Reply anytime, or raise a ticket at the Feishu / Helix service desk.",
        }
    else:
        labels = {
            "kicker": "PINGCAP IT &middot; WELCOME ABOARD",
            "h1": f"🎉 欢迎加入 {company_e}!",
            "sub": f"{name_e},欢迎加入 {company_e}! —— 这是属于你的第一天指南",
            "intro": '花 3 分钟读完,第一天会顺很多 🙂 下面按 <b style="color:#0F172A;">基础信息 &middot; IT &middot; HR &middot; 财务 &middot; 法务</b> 分好版块,需要什么直接看对应部分。',
            "s_basic": "👤 基础信息",
            "l_name": "姓名", "l_dept": "部门", "l_loc": "工作地",
            "l_email": "公司邮箱", "l_mgr": "直属经理", "l_hrbp": "HRBP",
            "s_it": "💻 IT &middot; 账号与日常工具",
            "it_intro": "为了帮助你顺利上手,我们整理了一些与日常工作相关的必备 IT 指南,请你抽空仔细阅读:",
            "it_mustread": "入职首日必读",
            "it_guide": f'<a href="{handbook_e}" style="color:#2563EB;font-weight:700;">入职首日 - PingCAP IT 指南(中国区)</a>',
            "it_hw": "电脑设备领取",
            "it_desk": "一站式 IT 服务入口",
            "desk_line": f"<b>IT Help Desk</b> 随时为你提供支持({e(it_contact)})。",
            "desk1": "🔹 <b>Lark/飞书搜索</b>:直接搜索「IT Help Desk」即可快速进入",
            "desk2": "🔹 <b>Lark/飞书路径</b>:工作台 → 服务台 → IT Help Desk",
            "desk3": f'🔹 <b>快捷链接</b>:通过 Lark/飞书点击<a href="{helpdesk_e}" style="color:#2563EB;font-weight:700;">此处</a>直接进入',
            "s_hr": "🧑‍💼 HR 相关",
            "hr_desk": f'HR 相关问题都可以优先咨询 <a href="{_HR_HELPDESK_URL}" style="color:#10B981;font-weight:700;">HR 服务台</a>',
            "hr_leave": f'假期 / 调休 在 <b>HRIS (SuccessFactors)</b> 查看与申请:{_doc_link(hris_manual_title, hris_manual_url, "#10B981")}',
            "s_survey": "📝 新员工入职体验问卷",
            "survey_line": f'今天抽几分钟填写 <a href="{_DAY1_SURVEY_URL}" style="color:#0EA5E9;font-weight:700;">新员工入职体验问卷</a> —— 你的真实感受,会让入职流程对之后加入的同学越来越好。',
            "s_fin": "💰 财务 &middot; 报销",
            "s_legal": "⚖️ 法务 &middot; 合规 &middot; 安全",
            "legal0": f'🔐 <b>入职必修</b>:在学习平台完成<a href="{_SECURITY_EXAM_URL}" style="color:#8B5CF6;">安全学习与考试</a>,请务必按时通过',
            "legal1": "请留意遵守已签署的 <b>保密协议(NDA)</b> 与信息安全规范",
            "legal2": "需要 <b>SOC2</b> 合规材料(客户审计、合规证明等)?在<b>飞书工作台</b>发起「SOC2 申请」审批流程即可,无需邮件",
            "legal3": "合同审阅、知识产权相关走飞书法务流程",
            "s_bless": "💌 给你的寄语",
            "cta": "登录 Helix 门户 →",
            "footer": f"本邮件由 {company_e} IT 团队通过 Helix 平台发送。有问题可直接回复,或在飞书 / Helix 服务台提单。",
        }

    def basic_row(label: str, value: str) -> str:
        return (
            f'<tr><td style="padding:4px 0;color:#64748B;width:120px;vertical-align:top;">{label}</td>'
            f'<td style="padding:4px 0;color:#0F172A;">{value}</td></tr>'
        )

    basics = (
        basic_row(labels["l_name"], name_e)
        + basic_row(labels["l_dept"], e(department))
        + basic_row(labels["l_loc"], e(location))
        + basic_row(labels["l_email"], e(person_email))
        + basic_row(labels["l_mgr"], e(manager_name))
        + basic_row(labels["l_hrbp"], e(hrbp_name))
    )

    def card(accent: str, bg: str, border: str, title: str, inner: str) -> str:
        return (
            '<table role="presentation" width="100%" cellpadding="0" cellspacing="0" style="margin:16px 0 0;border-collapse:separate;"><tr>'
            f'<td style="background:{bg};border:1px solid {border};border-left:4px solid {accent};border-radius:12px;padding:18px 22px;">'
            f'<div style="font-size:15px;font-weight:800;color:#0F172A;margin:0 0 12px;">{title}</div>{inner}</td></tr></table>'
        )

    def sub(label: str) -> str:
        return f'<div style="font-size:13px;color:#6366F1;font-weight:700;margin:12px 0 4px;">{label}</div>'

    # IT block (2026-08-25): REPLACED wholesale by HR's "Onboarding-IT" wiki
    # doc — 中文版 for China-entity hires, 英文版 for everyone else. The doc is
    # the content authority for what IT tells a new hire on day one, so the
    # block is exactly its three items (must-read guide / laptop pickup /
    # IT Help Desk with its three entry points). What the block used to carry
    # — Helix + Feishu sign-in steps, the "already set up for you" account and
    # seat list, the Helix self-service request pitch, the Feishu-vs-Google
    # docs rule, the new-hire quick guides, and the network-owner / HR-desk
    # "who to reach" lines — is deliberately gone, not merged: the user asked
    # for a replacement, not an append.
    it_inner = (
        '<div style="margin:2px 0 2px;color:#334155;font-size:14px;'
        f'line-height:1.65;">{labels["it_intro"]}</div>'
        + sub(labels["it_mustread"]) + _email_bullet(labels["it_guide"])
        + sub(labels["it_hw"]) + _email_bullet(laptop_line)
        + sub(labels["it_desk"]) + _email_bullet(labels["desk_line"])
        + _email_subitem(labels["desk1"])
        + _email_subitem(labels["desk2"])
        + _email_subitem(labels["desk3"])
    )
    # HR block (HR Ops 2026-07-15): helpdesk + HRIS(SuccessFactors) leave
    # manual + the TiLearning course entry. The former "新人培训与融入方案 /
    # 30天制定OKR" integration-plan timeline (PID guide + Day1/30d/90d/6mo
    # milestones) was dropped at HR's request — it belongs to HR's own
    # onboarding flow, not the IT day-one welcome.
    # EN: no leave/HRIS-manual line (non-China hires don't use it, HR
    # 2026-07-15). CN: keep it, then the China-only One Pager right after.
    if is_en:
        hr_inner = (
            _email_bullet(labels["hr_desk"], "#10B981")
            + _email_bullet(hris_guides_line, "#10B981")
            + _email_bullet(course_line, "#10B981")
        )
    else:
        hr_inner = (
            _email_bullet(labels["hr_desk"], "#10B981")
            + _email_bullet(labels["hr_leave"], "#10B981")
            + _email_bullet(one_pager_line, "#10B981")
            + _email_bullet(course_line, "#10B981")
        )
    survey_inner = (
        f'<div style="font-size:14px;color:#334155;line-height:1.8;">{labels["survey_line"]}</div>'
    )
    fin_inner = _email_bullet(fin1, "#F59E0B") + _email_bullet(fin2, "#F59E0B")
    legal_inner = (
        _email_bullet(labels["legal0"], "#8B5CF6") + _email_bullet(labels["legal1"], "#8B5CF6")
        + _email_bullet(labels["legal2"], "#8B5CF6") + _email_bullet(labels["legal3"], "#8B5CF6")
    )
    bless_inner = f'<div style="font-size:14px;color:#334155;line-height:1.9;">{blessing}</div>'

    return (
        '<div style="background:#EEF1F6;padding:28px 12px;'
        "font-family:-apple-system,'Segoe UI',Roboto,'Helvetica Neue',Arial,"
        "'PingFang SC','Microsoft YaHei',sans-serif;\">"
        '<table role="presentation" align="center" width="600" cellpadding="0" cellspacing="0" '
        'style="width:600px;max-width:600px;margin:0 auto;background:#ffffff;border-radius:16px;'
        'overflow:hidden;box-shadow:0 2px 10px rgba(15,23,42,.10);">'
        '<tr><td bgcolor="#0E2A47" style="background:#0E2A47;padding:34px 36px 30px;">'
        f'<div style="font-size:12px;letter-spacing:2.5px;color:#7FB0E6;font-weight:700;">{labels["kicker"]}</div>'
        f'<div style="font-size:27px;font-weight:800;color:#ffffff;margin:12px 0 8px;line-height:1.25;">{labels["h1"]}</div>'
        f'<div style="font-size:15px;color:#C7D6E8;line-height:1.5;">{labels["sub"]}</div>'
        '</td></tr>'
        '<tr><td style="padding:28px 36px 4px;">'
        f'<div style="font-size:14px;color:#475569;line-height:1.75;margin:0 0 8px;">{labels["intro"]}</div>'
        + card("#3B82F6", "#F8FAFC", "#E6EBF2", labels["s_basic"],
               f'<table role="presentation" width="100%" cellpadding="0" cellspacing="0" style="font-size:14px;">{basics}</table>')
        + card("#6366F1", "#F5F6FF", "#E2E5F7", labels["s_it"], it_inner)
        + card("#10B981", "#F2FBF6", "#D7EFE2", labels["s_hr"], hr_inner)
        + card("#F59E0B", "#FFFBF2", "#F3E6CC", labels["s_fin"], fin_inner)
        + card("#8B5CF6", "#FAF7FF", "#E8DEF7", labels["s_legal"], legal_inner)
        + card("#F43F5E", "#FFF5F7", "#F7D9E1", labels["s_bless"], bless_inner)
        + card("#0EA5E9", "#F0F9FF", "#D5EAF5", labels["s_survey"], survey_inner)
        + '<table role="presentation" align="center" cellpadding="0" cellspacing="0" style="margin:24px auto 8px;"><tr>'
        '<td bgcolor="#2563EB" style="background:#2563EB;border-radius:10px;">'
        f'<a href="{sso_e}" style="display:inline-block;padding:14px 34px;color:#ffffff;font-size:15px;font-weight:700;text-decoration:none;">{labels["cta"]}</a>'
        '</td></tr></table>'
        '</td></tr>'
        '<tr><td style="padding:18px 36px 28px;border-top:1px solid #EEF1F6;">'
        f'<div style="font-size:12px;color:#94A3B8;line-height:1.7;">{labels["footer"]}</div>'
        '</td></tr>'
        '</table></div>'
    )


def _build_welcome_card_fields(
    *,
    lang: str,
    is_en: bool,
    is_china: bool,
    company_name: str,
    person_name: str,
    person_email: str,
    department: str,
    location: str,
    manager_name: str,
    hrbp_name: str,
    sso_url: str,
    portal_url: str,
    network_contact: str,
    it_contact: str,
    seats: list[str],
    blessing: CardText,
    hris_manual_title: str = _HRIS_MANUAL_CN[0],
    hris_manual_url: str = _HRIS_MANUAL_CN[1],
) -> dict[str, Any]:
    """Compose the Day-1 Feishu welcome card — a 2x2 grey stat grid for the
    basics, a "today" checklist ON TOP (incl. the Day-1 onboarding-experience
    survey), then the per-domain sections mirroring the email (IT / quick
    guides / HR / finance / legal / who to reach), a small-grey note, two CTA
    buttons. ``is_china`` selects the reimbursement tool (汇联易 vs Navan)
    independent of the card language (HR Ops 2026-07-15).

    TWO AXES, and they are not the same axis:

    * ``lang`` — the READER's language (``resolve_lang(person)``: preference →
      country → entity). Everything a reader's eye lands on resolves on this
      axis: title, body, subtitle, stats, sections, buttons.
    * ``is_en`` / ``is_china`` — the labour-contract ENTITY. An HR Ops choice
      about *content*: which HRIS manual, 汇联易/Helios vs Navan, whether the
      hardware-pickup clause names the local-purchase route.

    ``title`` / ``body_text`` used to sit inside the ``is_en`` branch while the
    rest of the card had already moved to ``LocalizedText``, so a China-entity
    hire with ``preferred_language='en'`` — or any Japan-based hire — got a
    Chinese headline stapled over an English/Japanese card. They are composed
    once, in three languages, and returned RESOLVED for ``lang``:
    ``NotificationEnvelope.title`` / ``.body_text`` are ``str`` (the inbox row
    and the e-mail subject read them), so the resolution happens here.

    Layout rules from the review rounds: real lists use ``-`` bullets
    (hanging indent — never mid-sentence wraps), short link rows sit on
    ONE line, nothing from the email is dropped. Markdown constraints:
    ``**bold**`` + ``[text](url)`` survive the renderer's escaping, raw
    HTML doesn't.

    Returns ``{title, body_text, card_subtitle, card_stats,
    card_sections, card_actions}``.
    """
    # ``seats`` is no longer rendered: the "already set up for you" list
    # went away with the 2026-08-25 IT rewrite. The parameter stays so
    # call sites (and the e-mail renderer they share a context with) do
    # not have to branch.
    # IT section — regional targets from HR's "Onboarding-IT" wiki doc:
    # the 中国区 handbook + China IT Help Desk applink for China-entity
    # hires, the Global pair for everyone else.
    handbook_url = _day1_handbook_url(is_china)
    helpdesk_url = _it_helpdesk_applink(is_china)
    # The doc names its two editions differently; a reader who follows the
    # link should land on the document they were promised by name.
    if is_china:
        guide_zh = "入职首日 - PingCAP IT 指南(中国区)"
        guide_en = "Day 1 - PingCAP IT Guide (China)"
        guide_ja = "入社初日 - PingCAP IT ガイド(中国)"
    else:
        guide_zh = "入职首日 - PingCAP IT 指南(全球)"
        guide_en = "Day 1 - IT Guide at PingCAP (Global)"
        guide_ja = "入社初日 - PingCAP IT ガイド(グローバル)"
    # Reimbursement tool by employment entity (HR Ops 2026-07-15): China →
    # 汇联易 (Feishu Workplace app "Helios"); everyone else → Navan.
    # The tool / desk / workplace NAMES used to be picked by ``is_en`` — a pure
    # card-language choice for the same referent. They now carry all three
    # variants so the finance section can render per reader; the ENTITY choice
    # below (汇联易 vs Navan) stays, it is HR policy, not language.
    if is_china:
        _fin_tool_zh, _fin_tool_en, _fin_tool_ja = "汇联易", "Huilianyi", "Huilianyi"
        _fin_app = "Helios"
        _fin_desk_zh, _fin_desk_en, _fin_desk_ja = (
            "飞书财务工作台", "Feishu Finance Desk", "Feishu 財務デスク",
        )
        _fin_workplace_zh, _fin_workplace_en, _fin_workplace_ja = (
            "飞书工作台", "Feishu Workplace", "Feishu ワークプレイス",
        )
    else:
        _fin_tool_zh = _fin_tool_en = _fin_tool_ja = "Navan"
        _fin_app = "Navan"
        _fin_desk_zh, _fin_desk_en, _fin_desk_ja = (
            "飞书财务工作台", "Lark Finance Desk", "Lark 財務デスク",
        )
        _fin_workplace_zh, _fin_workplace_en, _fin_workplace_ja = (
            "飞书工作台", "Lark Workplace", "Lark ワークプレイス",
        )
    # Title / body / subtitle are card CHROME — they resolve per RECIPIENT
    # instead of being written once in whichever language ``is_en`` picked, so
    # they are hoisted out of the branch below (which still owns the
    # entity-specific COPY: manuals, reimbursement tool, hardware clause).
    card_title = LocalizedText(
        zh=f"🎉 欢迎加入 {company_name}!你的 IT 账号与第一天指南",
        en=f"🎉 Welcome to {company_name}! Your IT accounts & day-one guide",
        ja=f"🎉 {company_name} へようこそ!IT アカウントと入社初日ガイド",
    )
    card_body = LocalizedText(
        zh="更详细的版本已发到你的公司邮箱 📮",
        en="The detailed version is already in your work inbox 📮",
        ja="詳しい内容は会社のメールアドレス宛にお送りしています 📮",
    )
    card_subtitle = LocalizedText(
        zh=f"{person_name},欢迎加入 {company_name}! —— 第一天照这张卡走就稳了",
        en=(
            f"{person_name}, we're so glad you're here! "
            "This card has everything you need for your first day"
        ),
        ja=(
            f"{person_name} さん、{company_name} へようこそ! "
            "—— 初日はこのカードのとおりに進めれば安心です"
        ),
    )
    if is_en:
        stats = [
            CardStat(
                label=LocalizedText(zh="部门", en="Team", ja="部署"),
                value=department,
            ),
            CardStat(
                label=LocalizedText(zh="工作地", en="Location", ja="勤務地"),
                value=location,
            ),
            CardStat(
                label=LocalizedText(zh="直属经理", en="Manager", ja="上長"),
                value=manager_name,
            ),
            # "HRBP" is the same token in every language — no LocalizedText.
            CardStat(label="HRBP", value=hrbp_name),
        ]
        hw_item = (
            "**Pickup your Laptop**: if you have not yet collected or received "
            "your Laptop, reach out to your local HR Ops contact to arrange "
            "pickup (outside China / Japan / Malaysia / Singapore / US: follow "
            "IT policy to buy and reimburse)"
        )
        hw_item_zh = (
            "**领取笔记本电脑**:如果你还没有领到或收到笔记本电脑,请联系当地 "
            "HR Ops 联系人安排领取(中国 / 日本 / 马来西亚 / 新加坡 / 美国以外:"
            "按 IT 政策自行购买并报销)"
        )
        hw_item_ja = (
            "**ノートパソコンの受け取り**:まだ受け取っていない場合は、現地の "
            "HR Ops 担当者に連絡して受け取りを手配してください(中国・日本・"
            "マレーシア・シンガポール・米国以外の国では、IT ポリシーに従って"
            "ご自身で購入し、経費精算してください)"
        )
        sections = [
            CardSection(
                title=LocalizedText(
                    zh="✅ 今天要做", en="✅ Today", ja="✅ 本日やること",
                ),
                body=LocalizedText(
                    zh=(
                        f"- 读一遍 [入职第一天手册]({handbook_url}) —— "
                        "第一天的每一步都在里面\n"
                        f"- 用 **Google SSO** 登录 [Helix 门户]({sso_url}),"
                        "设置密码并开启两步验证\n"
                        "- 用**手机号**登录飞书 App —— 沟通、审批、服务台都在这里\n"
                        f"- {hw_item_zh}\n"
                        f"- 和你的 HRBP **{hrbp_name}** 打个招呼\n"
                        f"- 📝 抽几分钟填写 [新员工入职体验问卷]({_DAY1_SURVEY_URL}) "
                        "—— 你的真实反馈会让入职流程越来越好"
                    ),
                    en=(
                        f"- Read the [Day-1 handbook]({handbook_url}) — "
                        "your whole first day, step by step\n"
                        f"- Sign in to the [Helix portal]({sso_url}) with "
                        "**Google SSO**; set your password and 2-step verification\n"
                        "- Log in to the Feishu app with your **phone number** — "
                        "chat, approvals & service desk\n"
                        f"- {hw_item}\n"
                        f"- Say hi to your HRBP **{hrbp_name}**\n"
                        f"- 📝 Take a few minutes for the [New Hire Onboarding "
                        f"Survey]({_DAY1_SURVEY_URL}) — your honest feedback keeps "
                        "making onboarding better"
                    ),
                    ja=(
                        f"- [入社初日ハンドブック]({handbook_url}) に目を"
                        "通してください —— 初日の流れがすべて載っています\n"
                        f"- **Google SSO** で [Helix ポータル]({sso_url}) に"
                        "サインインし、パスワードと 2 段階認証を設定してください\n"
                        "- **電話番号**で Feishu アプリにログインしてください —— "
                        "チャット・承認・サービスデスクはこちらです\n"
                        f"- {hw_item_ja}\n"
                        f"- HRBP の **{hrbp_name}** にごあいさつしましょう\n"
                        f"- 📝 [新入社員オンボーディングアンケート]"
                        f"({_DAY1_SURVEY_URL}) にもぜひご協力ください —— "
                        "率直なご意見がオンボーディングの改善につながります"
                    ),
                ),
            ),
            CardSection(
                title=LocalizedText(
                    zh="💻 IT · 账号与工具",
                    en="💻 IT · Accounts & tools",
                    ja="💻 IT · アカウントとツール",
                ),
                # 2026-08-25: replaced wholesale by HR's "Onboarding-IT"
                # doc. The account / seat list, the Helix self-service pitch
                # and the docs rule are deliberately gone — the doc is the
                # content authority and does not carry them. The laptop
                # line is NOT repeated here: it already sits in the "today"
                # checklist one section above, in HR's own wording.
                body=LocalizedText(
                    zh=(
                        f"- 入职首日必读:[{guide_zh}]({handbook_url})\n"
                        f"- 一站式 IT 服务入口:[IT Help Desk]({helpdesk_url}) "
                        f"随时为你提供支持({it_contact})\n"
                        "  - 飞书搜索「IT Help Desk」\n"
                        "  - 或 工作台 → 服务台 → IT Help Desk"
                    ),
                    en=(
                        f"- Must-read on your first day: [{guide_en}]"
                        f"({handbook_url})\n"
                        f"- Your one-stop IT service portal: [IT Help Desk]"
                        f"({helpdesk_url}) is always here for you "
                        f"({it_contact})\n"
                        '  - Search "IT Help Desk" in Lark/Feishu\n'
                        "  - Or Workbench → Service Desk → IT Help Desk"
                    ),
                    ja=(
                        f"- 入社初日の必読:[{guide_ja}]({handbook_url})\n"
                        f"- IT サービスの総合窓口:[IT Help Desk]({helpdesk_url}) "
                        f"がいつでもサポートします({it_contact})\n"
                        "  - Feishu/Lark で「IT Help Desk」を検索\n"
                        "  - または ワークプレイス → ヘルプデスク → IT Help Desk"
                    ),
                ),
            ),
            CardSection(
                title=LocalizedText(
                    zh="🧭 新手指引",
                    en="🧭 Quick guides",
                    ja="🧭 クイックガイド",
                ),
                body=LocalizedText(
                    zh=(
                        f"[预约第一个会议]({_FAQ_FIRST_MEETING_URL}) · "
                        f"[飞书日历]({_FAQ_CALENDAR_URL}) · "
                        f"[组织架构 / 找同事]({_FAQ_ORG_CHART_URL}) · "
                        f"[文档协作]({_FAQ_DOC_COLLAB_URL}) · "
                        f"[权限申请]({_FAQ_ACCESS_REQUEST_URL})"
                    ),
                    en=(
                        f"[First meeting]({_FAQ_FIRST_MEETING_URL}) · "
                        f"[Lark calendar]({_FAQ_CALENDAR_URL}) · "
                        f"[Org chart]({_FAQ_ORG_CHART_URL}) · "
                        f"[Doc collaboration]({_FAQ_DOC_COLLAB_URL}) · "
                        f"[Access approval]({_FAQ_ACCESS_REQUEST_URL})"
                    ),
                    ja=(
                        f"[初めての会議設定]({_FAQ_FIRST_MEETING_URL}) · "
                        f"[Lark カレンダー]({_FAQ_CALENDAR_URL}) · "
                        f"[組織図]({_FAQ_ORG_CHART_URL}) · "
                        f"[ドキュメント共同編集]({_FAQ_DOC_COLLAB_URL}) · "
                        f"[権限申請]({_FAQ_ACCESS_REQUEST_URL})"
                    ),
                ),
            ),
            CardSection(
                title=LocalizedText(
                    zh="🧑‍💼 HR 相关", en="🧑‍💼 HR", ja="🧑‍💼 HR 関連",
                ),
                body=LocalizedText(
                    zh=(
                        f"- HR 相关的任何问题,都可以先找 [HR 服务台]"
                        f"({_HR_HELPDESK_URL})\n"
                        "- 日本 / 马来西亚 / 新加坡 / 美国的同事,请参考 "
                        f"[{_HRIS_MANUAL_JPMYSGUS[0]}]({_HRIS_MANUAL_JPMYSGUS[1]}) "
                        "在 **HRIS (SuccessFactors)** 中办理 HR 相关流程与申请;"
                        "其他国家的同事请参考 "
                        f"[{_HRIS_MANUAL_EOR[0]}]({_HRIS_MANUAL_EOR[1]})\n"
                        "- **新员工入职培训在线课程**请直接从 "
                        "**Lark → 工作台 → TiLearning** 进入"
                    ),
                    en=(
                        f"- For anything HR, the [HR helpdesk]({_HR_HELPDESK_URL}) "
                        "is your first stop\n"
                        "- For JP/MY/SG/US, please refer to "
                        f"[{_HRIS_MANUAL_JPMYSGUS[0]}]({_HRIS_MANUAL_JPMYSGUS[1]}) for "
                        "completing HR related process and requests in "
                        "**HRIS (SuccessFactors)**. For colleagues working in other "
                        f"countries, please refer to [{_HRIS_MANUAL_EOR[0]}]({_HRIS_MANUAL_EOR[1]})\n"
                        "- To access the **New Employee Orientation Online Course**, "
                        "go directly to **Lark → Workplace → TiLearning**"
                    ),
                    ja=(
                        f"- HR に関するご相談は、まず [HR ヘルプデスク]"
                        f"({_HR_HELPDESK_URL}) へお問い合わせください\n"
                        "- 日本・マレーシア・シンガポール・米国の方は、"
                        f"[{_HRIS_MANUAL_JPMYSGUS[0]}]({_HRIS_MANUAL_JPMYSGUS[1]}) "
                        "を参照して **HRIS (SuccessFactors)** で HR 関連の手続きや"
                        "申請を行ってください。その他の国の方は "
                        f"[{_HRIS_MANUAL_EOR[0]}]({_HRIS_MANUAL_EOR[1]}) を"
                        "ご参照ください\n"
                        "- **新入社員オリエンテーションのオンラインコース**は "
                        "**Lark → ワークプレイス → TiLearning** から直接"
                        "ご受講ください"
                    ),
                ),
            ),
            CardSection(
                title=LocalizedText(
                    zh="💰 财务 · 报销",
                    en="💰 Finance · Reimbursement",
                    ja="💰 経理 · 経費精算",
                ),
                body=LocalizedText(
                    zh=(
                        f"差旅 / 日常 / 采购报销请在 **{_fin_tool_zh}** 提交"
                        f"({_fin_workplace_zh} > Featured Apps > {_fin_app});"
                        f"如有疑问请查询{_fin_desk_zh}。"
                    ),
                    en=(
                        f"Submit travel / daily / procurement reimbursements "
                        f"{'in' if is_china else 'via'} **{_fin_tool_en}** "
                        f"({_fin_workplace_en} > Featured Apps > {_fin_app}). "
                        f"Questions? Check the {_fin_desk_en}."
                    ),
                    ja=(
                        f"出張 / 日常 / 購買の経費精算は **{_fin_tool_ja}**"
                        f"({_fin_workplace_ja} > Featured Apps > {_fin_app})"
                        f"から申請してください。ご不明な点は{_fin_desk_ja}で"
                        "ご確認ください。"
                    ),
                ),
            ),
            CardSection(
                title=LocalizedText(
                    zh="⚖️ 法务 · 合规 · 安全",
                    en="⚖️ Legal · Compliance · Security",
                    ja="⚖️ 法務 · コンプライアンス · セキュリティ",
                ),
                body=LocalizedText(
                    zh=(
                        f"- 🔐 入职必修:请按时通过 [安全学习与考试]"
                        f"({_SECURITY_EXAM_URL})\n"
                        "- 遵守已签署的 **NDA** 与公司信息安全规范\n"
                        "- SOC2 材料、合同审阅走飞书审批"
                    ),
                    en=(
                        f"- 🔐 Required: pass the [security training & exam]"
                        f"({_SECURITY_EXAM_URL}) on time\n"
                        "- Honour your **NDA** and our information-security policies\n"
                        "- SOC2 materials & contract review go through Feishu approvals"
                    ),
                    ja=(
                        f"- 🔐 必須:[セキュリティ研修と試験]"
                        f"({_SECURITY_EXAM_URL}) を期限内に修了してください\n"
                        "- 締結済みの **NDA** と情報セキュリティ規程を"
                        "遵守してください\n"
                        "- SOC2 関連資料と契約レビューは Feishu の承認フローで"
                        "申請してください"
                    ),
                ),
            ),
            CardSection(
                title=LocalizedText(
                    zh="🙋 遇到问题找谁",
                    en="🙋 Who to reach",
                    ja="🙋 困ったときの連絡先",
                ),
                body=LocalizedText(
                    zh=(
                        f"- 网络 / VPN:**{network_contact}**\n"
                        f"- HR 问题:飞书 HR 服务台 / 你的 HRBP **{hrbp_name}**"
                    ),
                    en=(
                        f"- Network / VPN: **{network_contact}**\n"
                        f"- HR questions: the HR helpdesk in Feishu / your HRBP "
                        f"**{hrbp_name}**"
                    ),
                    ja=(
                        f"- ネットワーク / VPN:**{network_contact}**\n"
                        f"- HR に関するお問い合わせ:Feishu の HR ヘルプデスク、"
                        f"または HRBP の **{hrbp_name}** まで"
                    ),
                ),
            ),
            CardSection(
                title=LocalizedText(
                    zh="💌 给你的寄语",
                    en="💌 A note for you",
                    ja="💌 あなたへのメッセージ",
                ),
                # ``blessing`` is per-person copy from the caller. A
                # ``CardText``: the dispatch site hands over all three
                # variants, so the note resolves for the reader instead of
                # sitting in English under a Japanese heading.
                body=blessing,
                tone="hint",
            ),
        ]
        actions = [
            CardAction(
                label=LocalizedText(
                    zh="登录 Helix 门户 →",
                    en="Sign in to Helix →",
                    ja="Helix にサインイン →",
                ),
                url=sso_url,
                style="primary",
            ),
            CardAction(
                label=LocalizedText(
                    zh="入职第一天手册",
                    en="Day-1 handbook",
                    ja="入社初日ハンドブック",
                ),
                url=handbook_url,
            ),
        ]
    else:
        stats = [
            CardStat(
                label=LocalizedText(zh="部门", en="Team", ja="部署"),
                value=department,
            ),
            CardStat(
                label=LocalizedText(zh="工作地", en="Location", ja="勤務地"),
                value=location,
            ),
            CardStat(
                label=LocalizedText(zh="直属经理", en="Manager", ja="上長"),
                value=manager_name,
            ),
            # "HRBP" is the same token in every language — no LocalizedText.
            CardStat(label="HRBP", value=hrbp_name),
        ]
        hw_item = (
            "**笔记本电脑领取**:如尚未领或收到笔记本电脑,请联系当地 HR Ops 联系人领取"
        )
        hw_item_en = (
            "**Pickup your laptop**: if you have not collected or received your "
            "laptop yet, reach out to your local HR Ops contact to arrange pickup"
        )
        hw_item_ja = (
            "**ノートパソコンの受け取り**:まだ受け取っていない場合は、現地の "
            "HR Ops 担当者に連絡して受け取りを手配してください"
        )
        sections = [
            CardSection(
                title=LocalizedText(
                    zh="✅ 今天要做", en="✅ Today", ja="✅ 本日やること",
                ),
                body=LocalizedText(
                    zh=(
                        f"- 读一遍 [入职第一天手册]({handbook_url}) —— "
                        "第一天的一切都在里面\n"
                        f"- 用 **Google SSO** 登录 [Helix 门户]({sso_url}),"
                        "设好密码、开两步验证\n"
                        "- 用**手机号**登录飞书 App —— 沟通、审批、服务台都在这\n"
                        f"- {hw_item}\n"
                        f"- 和你的 HRBP **{hrbp_name}** 打个招呼\n"
                        f"- 📝 抽几分钟填 [新员工入职体验问卷]({_DAY1_SURVEY_URL}) —— "
                        "你的真实感受,会让入职流程越来越好"
                    ),
                    en=(
                        f"- Read the [Day-1 handbook]({handbook_url}) — "
                        "everything about your first day is in there\n"
                        f"- Sign in to the [Helix portal]({sso_url}) with "
                        "**Google SSO**; set your password and turn on 2-step "
                        "verification\n"
                        "- Log in to the Feishu app with your **phone number** — "
                        "chat, approvals and the service desk all live there\n"
                        f"- {hw_item_en}\n"
                        f"- Say hi to your HRBP **{hrbp_name}**\n"
                        f"- 📝 Take a few minutes for the [New Hire Onboarding "
                        f"Survey]({_DAY1_SURVEY_URL}) — how it really felt for you "
                        "is what keeps making onboarding better"
                    ),
                    ja=(
                        f"- [入社初日ハンドブック]({handbook_url}) に目を"
                        "通してください —— 初日のことはすべて載っています\n"
                        f"- **Google SSO** で [Helix ポータル]({sso_url}) に"
                        "サインインし、パスワードの設定と 2 段階認証の有効化を"
                        "行ってください\n"
                        "- **電話番号**で Feishu アプリにログインしてください —— "
                        "チャット・承認・サービスデスクはすべてこちらです\n"
                        f"- {hw_item_ja}\n"
                        f"- HRBP の **{hrbp_name}** にごあいさつしましょう\n"
                        f"- 📝 [新入社員オンボーディングアンケート]"
                        f"({_DAY1_SURVEY_URL}) にもぜひご協力ください —— "
                        "率直なご感想がオンボーディングの改善につながります"
                    ),
                ),
            ),
            CardSection(
                title=LocalizedText(
                    zh="💻 IT · 账号与工具",
                    en="💻 IT · Accounts & tools",
                    ja="💻 IT · アカウントとツール",
                ),
                # 2026-08-25: replaced wholesale by HR's "Onboarding-IT"
                # doc. The account / seat list, the Helix self-service pitch
                # and the docs rule are deliberately gone — the doc is the
                # content authority and does not carry them. The laptop
                # line is NOT repeated here: it already sits in the "today"
                # checklist one section above, in HR's own wording.
                body=LocalizedText(
                    zh=(
                        f"- 入职首日必读:[{guide_zh}]({handbook_url})\n"
                        f"- 一站式 IT 服务入口:[IT Help Desk]({helpdesk_url}) "
                        f"随时为你提供支持({it_contact})\n"
                        "  - 飞书搜索「IT Help Desk」\n"
                        "  - 或 工作台 → 服务台 → IT Help Desk"
                    ),
                    en=(
                        f"- Must-read on your first day: [{guide_en}]"
                        f"({handbook_url})\n"
                        f"- Your one-stop IT service portal: [IT Help Desk]"
                        f"({helpdesk_url}) is always here for you "
                        f"({it_contact})\n"
                        '  - Search "IT Help Desk" in Lark/Feishu\n'
                        "  - Or Workbench → Service Desk → IT Help Desk"
                    ),
                    ja=(
                        f"- 入社初日の必読:[{guide_ja}]({handbook_url})\n"
                        f"- IT サービスの総合窓口:[IT Help Desk]({helpdesk_url}) "
                        f"がいつでもサポートします({it_contact})\n"
                        "  - Feishu/Lark で「IT Help Desk」を検索\n"
                        "  - または ワークプレイス → ヘルプデスク → IT Help Desk"
                    ),
                ),
            ),
            CardSection(
                title=LocalizedText(
                    zh="🧭 新手指引",
                    en="🧭 Quick guides",
                    ja="🧭 クイックガイド",
                ),
                body=LocalizedText(
                    zh=(
                        f"[预约第一个会议]({_FAQ_FIRST_MEETING_URL}) · "
                        f"[飞书日历]({_FAQ_CALENDAR_URL}) · "
                        f"[组织架构 / 找同事]({_FAQ_ORG_CHART_URL}) · "
                        f"[文档协作]({_FAQ_DOC_COLLAB_URL}) · "
                        f"[权限申请]({_FAQ_ACCESS_REQUEST_URL})"
                    ),
                    en=(
                        f"[First meeting]({_FAQ_FIRST_MEETING_URL}) · "
                        f"[Feishu calendar]({_FAQ_CALENDAR_URL}) · "
                        f"[Org chart / find a colleague]({_FAQ_ORG_CHART_URL}) · "
                        f"[Doc collaboration]({_FAQ_DOC_COLLAB_URL}) · "
                        f"[Access request]({_FAQ_ACCESS_REQUEST_URL})"
                    ),
                    ja=(
                        f"[初めての会議設定]({_FAQ_FIRST_MEETING_URL}) · "
                        f"[Feishu カレンダー]({_FAQ_CALENDAR_URL}) · "
                        f"[組織図 / 同僚を探す]({_FAQ_ORG_CHART_URL}) · "
                        f"[ドキュメント共同編集]({_FAQ_DOC_COLLAB_URL}) · "
                        f"[権限申請]({_FAQ_ACCESS_REQUEST_URL})"
                    ),
                ),
            ),
            CardSection(
                title=LocalizedText(
                    zh="🧑‍💼 HR 相关", en="🧑‍💼 HR", ja="🧑‍💼 HR 関連",
                ),
                body=LocalizedText(
                    zh=(
                        f"- HR 相关问题都可以优先咨询 [HR 服务台]({_HR_HELPDESK_URL})\n"
                        "- 假期 / 调休 在 **HRIS (SuccessFactors)** 查看与申请:"
                        f"[{hris_manual_title}]({hris_manual_url})\n"
                        "- 你关心的 HR 问题都在这里(五险一金、薪酬福利等):"
                        f"[PingCAP 入职 One Pager]({_HR_ONE_PAGER_URL})\n"
                        "- 新员工培训课程入口,请直接从 飞书 → 工作台 → TiLearning 进入"
                    ),
                    en=(
                        f"- For anything HR, the [HR helpdesk]({_HR_HELPDESK_URL}) "
                        "is your first stop\n"
                        "- Check and apply for leave / time off in "
                        "**HRIS (SuccessFactors)**: "
                        f"[{hris_manual_title}]({hris_manual_url})\n"
                        "- The HR topics you care about (social insurance, housing "
                        "fund, pay & benefits …) are all here: "
                        f"[PingCAP onboarding One Pager]({_HR_ONE_PAGER_URL})\n"
                        "- For the new-hire training course, go directly to "
                        "Feishu → Workplace → TiLearning"
                    ),
                    ja=(
                        f"- HR に関するご相談は、まず [HR ヘルプデスク]"
                        f"({_HR_HELPDESK_URL}) へお問い合わせください\n"
                        "- 休暇 / 振替休日の確認・申請は "
                        "**HRIS (SuccessFactors)** から:"
                        f"[{hris_manual_title}]({hris_manual_url})\n"
                        "- 社会保険・住宅積立金・給与福利厚生など、気になる HR "
                        "情報はこちらにまとまっています:"
                        f"[PingCAP 入社 One Pager]({_HR_ONE_PAGER_URL})\n"
                        "- 新入社員研修コースは Feishu → ワークプレイス → "
                        "TiLearning から直接ご受講ください"
                    ),
                ),
            ),
            CardSection(
                title=LocalizedText(
                    zh="💰 财务 · 报销",
                    en="💰 Finance · Reimbursement",
                    ja="💰 経理 · 経費精算",
                ),
                body=LocalizedText(
                    zh=(
                        f"差旅 / 日常 / 采购报销在 **{_fin_tool_zh}** 提交"
                        f"({_fin_workplace_zh} - Featured Apps - {_fin_app});"
                        f"有疑问请至{_fin_desk_zh}查询。"
                    ),
                    en=(
                        f"Submit travel / daily / procurement reimbursements "
                        f"{'in' if is_china else 'via'} **{_fin_tool_en}** "
                        f"({_fin_workplace_en} - Featured Apps - {_fin_app}). "
                        f"Questions? Check the {_fin_desk_en}."
                    ),
                    ja=(
                        f"出張 / 日常 / 購買の経費精算は **{_fin_tool_ja}**"
                        f"({_fin_workplace_ja} - Featured Apps - {_fin_app})"
                        f"から申請してください。ご不明な点は{_fin_desk_ja}で"
                        "ご確認ください。"
                    ),
                ),
            ),
            CardSection(
                title=LocalizedText(
                    zh="⚖️ 法务 · 合规 · 安全",
                    en="⚖️ Legal · Compliance · Security",
                    ja="⚖️ 法務 · コンプライアンス · セキュリティ",
                ),
                body=LocalizedText(
                    zh=(
                        f"- 🔐 入职必修:[安全学习与考试]({_SECURITY_EXAM_URL}),"
                        "请按时通过\n"
                        "- 遵守已签署的 **NDA** 与信息安全规范\n"
                        "- SOC2 材料、合同审阅走飞书工作台审批"
                    ),
                    en=(
                        f"- 🔐 Required for every new hire: pass the "
                        f"[security training & exam]({_SECURITY_EXAM_URL}) on time\n"
                        "- Honour the **NDA** you signed and our "
                        "information-security policies\n"
                        "- SOC2 materials & contract review go through the Feishu "
                        "Workplace approvals"
                    ),
                    ja=(
                        f"- 🔐 入社時の必須項目:[セキュリティ研修と試験]"
                        f"({_SECURITY_EXAM_URL}) を期限内に修了してください\n"
                        "- 締結済みの **NDA** と情報セキュリティ規程を"
                        "遵守してください\n"
                        "- SOC2 関連資料と契約レビューは Feishu ワークプレイスの"
                        "承認フローで申請してください"
                    ),
                ),
            ),
            CardSection(
                title=LocalizedText(
                    zh="🙋 遇到问题找谁",
                    en="🙋 Who to reach",
                    ja="🙋 困ったときの連絡先",
                ),
                body=LocalizedText(
                    zh=(
                        f"- 网络 / VPN:**{network_contact}**\n"
                        f"- HR 问题:飞书 HR 服务台 / 你的 HRBP **{hrbp_name}**"
                    ),
                    en=(
                        f"- Network / VPN: **{network_contact}**\n"
                        f"- HR questions: the HR helpdesk in Feishu / your HRBP "
                        f"**{hrbp_name}**"
                    ),
                    ja=(
                        f"- ネットワーク / VPN:**{network_contact}**\n"
                        f"- HR に関するお問い合わせ:Feishu の HR ヘルプデスク、"
                        f"または HRBP の **{hrbp_name}** まで"
                    ),
                ),
            ),
            CardSection(
                title=LocalizedText(
                    zh="💌 给你的寄语",
                    en="💌 A note for you",
                    ja="💌 あなたへのメッセージ",
                ),
                # ``blessing`` is per-person copy from the caller. A
                # ``CardText``: the dispatch site hands over all three
                # variants, so the note resolves for the reader instead of
                # sitting in English under a Japanese heading.
                body=blessing,
                tone="hint",
            ),
        ]
        actions = [
            CardAction(
                label=LocalizedText(
                    zh="登录 Helix 门户 →",
                    en="Sign in to the Helix portal →",
                    ja="Helix ポータルにサインイン →",
                ),
                url=sso_url,
                style="primary",
            ),
            CardAction(
                label=LocalizedText(
                    zh="入职第一天手册",
                    en="Day-1 handbook",
                    ja="入社初日ハンドブック",
                ),
                url=handbook_url,
            ),
        ]
    return {
        # Resolved to ``str`` here: the envelope's title/body are typed ``str``
        # and are read by the inbox row and the e-mail path, not only the card.
        "title": resolve_card_text(card_title, lang),
        "body_text": resolve_card_text(card_body, lang),
        "card_subtitle": card_subtitle,
        "card_stats": stats,
        "card_sections": sections,
        "card_actions": actions,
    }


def _build_day90_survey_fields(
    *,
    lang: str,
    is_en: bool,
    company_name: str,
    person_name: str,
) -> dict[str, Any]:
    """Compose the Day-90 onboarding-experience survey invite — a
    compact Feishu card (short warm body + a 1x2 grey stat row + a
    small-grey confidentiality note + one primary CTA) and a
    single-card HTML email, bilingual.

    ``lang`` is the READER's language and drives every field the card renders
    (title / body / subtitle / stats / sections / button). ``is_en`` still
    picks the HTML e-mail's copy, which is a single hand-laid bilingual
    template — a ja recipient's e-mail degrades to the English card, as it did
    before, and full ja-ification of the HTML is a separate change.

    Returns ``{title, body_text, body_html, card_subtitle, card_stats,
    card_sections, card_actions}``.
    """
    e = _html.escape
    name_e, company_e = e(person_name), e(company_name)
    url = _DAY90_SURVEY_URL

    # Card chrome — resolved per RECIPIENT, so it is hoisted out of the
    # ``is_en`` branch that still owns the email copy below. Title and body
    # stayed behind in that branch through the first sweep, which is how a
    # Japanese-reading hire got an English headline over a Japanese card.
    card_title = LocalizedText(
        zh=f"📝 入职 {company_name} 90 天啦,想听听你的感受",
        en=f"📝 90 days at {company_name} — how has it been?",
        ja=f"📝 {company_name} にご入社から 90 日 —— ご感想を聞かせてください",
    )
    card_body = LocalizedText(
        zh=(
            f"不知不觉你已加入 {company_name} **90 天**了!入职体验如何,"
            "只有你能告诉我们 —— 花几分钟聊聊,让之后加入的 PingCAPer "
            "走得更顺 🙌"
        ),
        en=(
            f"Just like that, you've been with {company_name} for 90 days! "
            "Only you can tell us what onboarding really felt like — "
            "a few minutes of your feedback makes it smoother for every "
            "PingCAPer who joins after you 🙌"
        ),
        ja=(
            f"{company_name} にご入社されてから **90 日** が経ちました!"
            "入社後の体験がどうだったかは、ご本人にしか分かりません。"
            "数分のご協力が、これから入社する PingCAPer のためになります 🙌"
        ),
    )
    card_subtitle = LocalizedText(
        zh=f"{person_name},谢谢这三个月的投入 —— 聊聊你的入职体验吧",
        en=(
            f"{person_name}, thank you for these three months — "
            "we'd love to hear about your onboarding"
        ),
        ja=(
            f"{person_name} さん、この 3 か月間ありがとうございます —— "
            "オンボーディングの体験をぜひ聞かせてください"
        ),
    )
    if is_en:
        cta = "Fill in the survey →"
        header_kicker = "PINGCAP &middot; YOUR FIRST 90 DAYS"
        h1 = "📝 90 days in — tell us how it went"
        sub = f"{name_e}, thank you for these three months"
        html_ask = (
            f"You&rsquo;ve been with {company_e} for <b>90 days</b> — we hope "
            "it&rsquo;s been a great start! Your feedback is essential to how we "
            "improve onboarding: by sharing your experience you directly shape it "
            "for the PingCAPers who join after you. It takes just a few minutes."
        )
        html_conf = (
            "Rest assured — responses are <b>confidential</b> and used for "
            "internal improvement only. Thank you for your valuable input! 💙"
        )
        footer = (
            f"Sent by the {company_e} HR &amp; IT teams via the Helix platform."
            f"<br>— {company_e} HR &amp; IT"
        )
    else:
        cta = "填写问卷 →"
        header_kicker = "PINGCAP &middot; 入职 90 天"
        h1 = "📝 入职 90 天,想听听你的感受"
        sub = f"{name_e},谢谢这三个月的投入"
        html_ask = (
            f"不知不觉你已加入 {company_e} <b>90 天</b>了 —— 希望这是一段美好的"
            "开始!你的反馈对我们持续改进入职流程至关重要:分享你的体验,就是在为"
            "之后加入的 PingCAPer 铺路。只需要几分钟。"
        )
        html_conf = (
            "请放心 —— 问卷回复<b>完全保密</b>,仅用于内部改进。感谢你的宝贵意见!💙"
        )
        footer = (
            f"本邮件由 {company_e} HR 与 IT 团队通过 Helix 平台发送。"
            f"<br>— {company_e} HR &amp; IT"
        )

    # Card chrome is language-independent now: the stat row and the
    # confidentiality note used to be duplicated in both ``is_en`` arms with
    # identical meaning, which meant a ja recipient read them in English.
    # The branch above still owns the email copy (title / body / HTML); these
    # resolve per RECIPIENT.
    stats = [
        CardStat(
            label=LocalizedText(
                zh="你与 PingCAP 已相伴",
                en="You've been with us",
                ja="入社からの日数",
            ),
            value=LocalizedText(zh="90 天", en="90 days", ja="90 日"),
        ),
        CardStat(
            label=LocalizedText(
                zh="填写大约需要",
                en="The survey takes",
                ja="所要時間",
            ),
            value=LocalizedText(zh="3 分钟", en="~3 min", ja="約 3 分"),
        ),
    ]
    sections = [
        CardSection(
            body=LocalizedText(
                zh="问卷回复完全保密,仅用于内部改进。感谢你的宝贵意见!💙",
                en=(
                    "Responses are confidential and used for internal "
                    "improvement only. Thank you for your valuable input! 💙"
                ),
                ja=(
                    "ご回答は機密として扱い、社内の改善のみに使用します。"
                    "貴重なご意見をありがとうございます!💙"
                ),
            ),
            tone="hint",
        ),
    ]

    body_html = (
        '<div style="background:#EEF1F6;padding:28px 12px;'
        "font-family:-apple-system,'Segoe UI',Roboto,'Helvetica Neue',Arial,"
        "'PingFang SC','Microsoft YaHei',sans-serif;\">"
        '<table role="presentation" align="center" width="600" cellpadding="0" cellspacing="0" '
        'style="width:600px;max-width:600px;margin:0 auto;background:#ffffff;border-radius:16px;'
        'overflow:hidden;box-shadow:0 2px 10px rgba(15,23,42,.10);">'
        '<tr><td bgcolor="#0E2A47" style="background:#0E2A47;padding:34px 36px 30px;">'
        f'<div style="font-size:12px;letter-spacing:2.5px;color:#7FB0E6;font-weight:700;">{header_kicker}</div>'
        f'<div style="font-size:26px;font-weight:800;color:#ffffff;margin:12px 0 8px;line-height:1.3;">{h1}</div>'
        f'<div style="font-size:15px;color:#C7D6E8;line-height:1.5;">{sub}</div>'
        '</td></tr>'
        '<tr><td style="padding:30px 36px 8px;">'
        f'<div style="font-size:14px;color:#334155;line-height:1.85;">{html_ask}</div>'
        '<table role="presentation" align="center" cellpadding="0" cellspacing="0" style="margin:26px auto 6px;"><tr>'
        '<td bgcolor="#2563EB" style="background:#2563EB;border-radius:10px;">'
        f'<a href="{url}" style="display:inline-block;padding:14px 34px;color:#ffffff;font-size:15px;font-weight:700;text-decoration:none;">{cta}</a>'
        '</td></tr></table>'
        f'<div style="font-size:13px;color:#64748B;line-height:1.8;margin:18px 0 6px;">{html_conf}</div>'
        '</td></tr>'
        '<tr><td style="padding:18px 36px 28px;border-top:1px solid #EEF1F6;">'
        f'<div style="font-size:12px;color:#94A3B8;line-height:1.7;">{footer}</div>'
        '</td></tr>'
        '</table></div>'
    )
    # ``cta`` still drives the HTML e-mail button (rendered in the e-mail's own
    # language); the card button resolves per recipient.
    actions = [
        CardAction(
            label=LocalizedText(
                zh="填写问卷 →",
                en="Fill in the survey →",
                ja="アンケートに回答する →",
            ),
            url=url,
            style="primary",
        ),
    ]
    return {
        # Resolved to ``str`` here — see ``_build_welcome_card_fields``.
        "title": resolve_card_text(card_title, lang),
        "body_text": resolve_card_text(card_body, lang),
        "body_html": body_html,
        "card_subtitle": card_subtitle,
        "card_stats": stats,
        "card_sections": sections,
        "card_actions": actions,
    }


__all__ = ["HireDayNotifyService"]