"""Regression: ``preview_full_employee_sync`` must include SF status='f'
users and route them through the offboard branch.

Pre-launch P0 (wangdi PR 2026-05-30): the bootstrap full sync called
``list_active_users()`` (``status eq 't'``), permanently invisibling
every leaver. The fix switches to ``list_all_users()`` and synthesizes
offboard events for ``status='f'`` users so they land in
``identity_persons`` with ``employment_status='left'`` (when endDate
is firmly past) or ``leaver_pending`` (when within 30 days).

These tests pin:

1. Bootstrap calls ``list_all_users()`` (not ``list_active_users()``).
2. Bootstrap fetches EmpEmployment endDates for forfeited userIds.
3. Forfeited user → staged event_type='offboard'.
4. Confirm-batch → IdentityPerson(employment_status='left') for
   historical leavers (>30 day endDate).
5. Confirm-batch → IdentityPerson(employment_status='leaver_pending')
   for recent leavers (≤30 day endDate).
"""
from __future__ import annotations

from datetime import date

import pytest
from sqlalchemy import select

from app.domains.hr.models import HrSyncStagedEvent
from app.domains.hr.service import SuccessFactorsLifecycleService
from app.domains.leaver_blocked.listener import (
    register_listener as _register_leaver_listener,
)
from app.identity.audit_listener import (
    register_listener as _register_identity_audit_listener,
)
from app.identity.person_models import IdentityPerson


@pytest.fixture(autouse=True)
def _wire_prod_listeners():
    """Activate the same SQLAlchemy event listeners production registers
    on boot. Without this, the leaver_blocked listener stays dormant in
    tests and the bootstrap-forfeited path's ``employment_status='left'``
    INSERT decision is never genuinely tested against the listener's
    contract — see review pass 1 finding "30-day-cutoff INSERT-as-'left'
    path is unreachable in production". Both registrations are
    idempotent (``_REGISTERED`` guard) so the fixture is safe to apply
    auto-use.
    """
    _register_leaver_listener()
    _register_identity_audit_listener()


class _FakeFullSyncClient:
    """Fakes the SF client surface for ``preview_full_employee_sync``.

    Returns 3 users: 1 active, 1 long-ago forfeited (wangdi-shape),
    1 recently forfeited. Companies/departments/locations stay empty
    so we don't exercise the org-tree code paths.
    """

    def __init__(self) -> None:
        self.list_all_called = False
        self.list_active_called = False
        self.get_employments_called_with: list[str] | None = None

    async def list_active_companies(self) -> list[dict]:
        return []

    async def list_active_departments(self) -> list[dict]:
        return []

    async def list_active_locations(self) -> list[dict]:
        return []

    async def list_active_positions(self) -> list[dict]:
        return []

    async def list_active_users(self) -> list[dict]:
        # Tests assert this is NOT called by the bootstrap path. If
        # it ever is, the regression has returned.
        self.list_active_called = True
        return []

    async def list_all_users(self) -> list[dict]:
        self.list_all_called = True
        # Active employee. ``hireDate`` 2023-11-14 (PingCAP start)
        # is distinct from the career-start ``originalStartDate``
        # 2018-04-01 emitted by ``get_employments_for_users`` below
        # — the regression for the 年假计算器 work_start_date bug.
        active = {
            "userId": "P_ACTIVE",
            "displayName": "Active Person",
            "email": "active@pingcap.com",
            "empId": "P_ACTIVE",
            "status": "t",
            "hireDate": "/Date(1700000000000)/",  # 2023-11-14
            "lastModifiedDateTime": "/Date(1700000000000+0000)/",
        }
        # wangdi-shape: terminated 2026-03-20 (long ago vs 2026-05-30 today).
        wangdi = {
            "userId": "P0175",
            "displayName": "王迪 Di Wang",
            "email": "wangdi@pingcap.com",
            "empId": "P0175",
            # status='f' is the marker that was silently filtered.
            "status": "f",
            "hireDate": "/Date(1572220800000)/",  # 2019-10-28
            "lastModifiedDateTime": "/Date(1776514800000+0000)/",
        }
        # Recently terminated — should land at leaver_pending (within 30d).
        recent = {
            "userId": "P_RECENT",
            "displayName": "Recent Leaver",
            "email": "recent@pingcap.com",
            "empId": "P_RECENT",
            "status": "f",
            "hireDate": "/Date(1600000000000)/",
            "lastModifiedDateTime": "/Date(1776514800000+0000)/",
        }
        return [active, wangdi, recent]

    async def get_employments_for_users(self, user_ids: list[str]) -> list[dict]:
        self.get_employments_called_with = list(user_ids)
        # endDate 2026-03-20 for wangdi, endDate 2026-05-16 for recent.
        # JS epoch ms: 2026-03-20 = 1773964800000; 2026-05-16 = 1778889600000
        results: list[dict] = []
        if "P_ACTIVE" in user_ids:
            # Active employee: 3 distinct career-related dates exercising
            # the 3-tier fallback chain in ``_upsert_person``:
            #   * firstDateWorked  = 2011-11-11 (real career start)
            #   * originalStartDate = 2020-01-01 (PingCAP-first-stint
            #     anchor, distinct from startDate to model a rehire)
            #   * startDate         = 2023-11-14 (current stint = hireDate)
            #   * seniorityDate     = 2023-11-14 (司龄 anchor reset on rehire)
            # JS epoch ms: 2011-11-11 = 1320969600000;
            #              2020-01-01 = 1577836800000;
            #              2023-11-14 = 1700000000000
            results.append({
                "personIdExternal": "P_ACTIVE",
                "userId": "P_ACTIVE",
                "startDate": "/Date(1700000000000)/",  # 2023-11-14
                "endDate": None,
                "originalStartDate": "/Date(1577836800000)/",  # 2020-01-01
                "firstDateWorked": "/Date(1320969600000)/",  # 2011-11-11
                "seniorityDate": "/Date(1700000000000)/",  # 2023-11-14
                "lastModifiedDateTime": "/Date(1700000000000+0000)/",
            })
        if "P0175" in user_ids:
            results.append({
                "personIdExternal": "P0175",
                "userId": "P0175",
                "startDate": "/Date(1572220800000)/",
                "endDate": "/Date(1773964800000)/",  # 2026-03-20
                "originalStartDate": "/Date(1572220800000)/",
                "lastModifiedDateTime": "/Date(1776514800000+0000)/",
            })
        if "P_RECENT" in user_ids:
            results.append({
                "personIdExternal": "P_RECENT",
                "userId": "P_RECENT",
                "startDate": "/Date(1600000000000)/",
                "endDate": "/Date(1778889600000)/",  # 2026-05-16
                "originalStartDate": "/Date(1600000000000)/",
                "lastModifiedDateTime": "/Date(1776514800000+0000)/",
            })
        return results

    async def get_jobs_for_users(self, user_ids: list[str]) -> list[dict]:
        return []


@pytest.mark.asyncio
async def test_preview_full_sync_calls_list_all_users_not_list_active(
    db_session,
) -> None:
    """The bootstrap MUST use ``list_all_users``. Calling
    ``list_active_users`` is the exact wangdi-regression shape."""
    client = _FakeFullSyncClient()
    service = SuccessFactorsLifecycleService(
        db_session, sf_client=client, today=date(2026, 5, 30),
    )
    await service.preview_full_employee_sync(triggered_by="test")

    assert client.list_all_called, (
        "preview_full_employee_sync did not call list_all_users — the "
        "wangdi regression has returned."
    )
    assert not client.list_active_called, (
        "preview_full_employee_sync incorrectly called list_active_users "
        "— forfeited users will silently disappear from the sync."
    )


@pytest.mark.asyncio
async def test_preview_full_sync_synthesizes_offboard_for_forfeited(
    db_session,
) -> None:
    """``status='f'`` users must produce ``event_type='offboard'``
    staged events; ``status='t'`` users still produce ``onboard``.
    """
    client = _FakeFullSyncClient()
    service = SuccessFactorsLifecycleService(
        db_session, sf_client=client, today=date(2026, 5, 30),
    )
    batch = await service.preview_full_employee_sync(triggered_by="test")

    # EmpEmployment fetch must cover ALL users (not just forfeited) —
    # active users need ``originalStartDate`` for work_start_date
    # (参加工作日期 / 社会工龄). The original "forfeited only" scope
    # was the bug behind 99.5% of prod active rows showing
    # work_start_date == hire_date (2026-06-02).
    called_with = set(client.get_employments_called_with or [])
    assert "P0175" in called_with
    assert "P_RECENT" in called_with
    assert "P_ACTIVE" in called_with, (
        "Active users MUST trigger EmpEmployment lookup so we can pull "
        "originalStartDate (career start) for work_start_date — without "
        "this the synthetic onboard event falls back to user.hireDate, "
        "silently equating 参加工作日期 with 入职日期 for every row."
    )

    staged_rows = (await db_session.execute(
        select(HrSyncStagedEvent).where(HrSyncStagedEvent.batch_id == batch.id)
    )).scalars().all()
    by_user = {r.user_id: r.event_type for r in staged_rows}
    assert by_user.get("P_ACTIVE") == "onboard"
    assert by_user.get("P0175") == "offboard", (
        f"wangdi (status='f', endDate=2026-03-20) must be routed via "
        f"offboard — got {by_user.get('P0175')!r}"
    )
    assert by_user.get("P_RECENT") == "offboard"


@pytest.mark.asyncio
async def test_confirm_full_sync_lands_historical_leaver_as_left(
    db_session,
) -> None:
    """End-to-end: forfeited user with endDate >30 days past →
    IdentityPerson(employment_status='left'). This is the wangdi
    invariant: bootstrap should re-create wangdi as a fully-departed
    row so downstream anchors can re-link without a manual hop.
    """
    client = _FakeFullSyncClient()
    service = SuccessFactorsLifecycleService(
        db_session, sf_client=client, today=date(2026, 5, 30),
    )
    batch = await service.preview_full_employee_sync(triggered_by="test")
    result = await service.confirm_batch(batch.id, confirmed_by="test")
    assert result["error_count"] == 0, (
        f"confirm_batch reported errors: {result}"
    )

    # wangdi → 'left'
    wangdi = (await db_session.execute(
        select(IdentityPerson).where(IdentityPerson.hris_person_id == "P0175")
    )).scalar_one()
    assert wangdi.employment_status == "left", (
        f"wangdi must land at 'left' (endDate=2026-03-20, today=2026-05-30, "
        f"71 days past); got {wangdi.employment_status!r}"
    )
    assert wangdi.termination_date == date(2026, 3, 20)
    assert wangdi.work_email == "wangdi@pingcap.com"
    assert wangdi.hris_record_state == "present"


@pytest.mark.asyncio
async def test_confirm_full_sync_recent_leaver_stays_leaver_pending(
    db_session,
) -> None:
    """endDate within 30 days → stays at 'leaver_pending'. The
    process_advance cron handles the final flip — bypassing it
    would skip in-flight handover actions (account revoke, asset
    return) that ARE still meaningful for a fresh leaver.
    """
    client = _FakeFullSyncClient()
    # today = 2026-05-30; recent endDate = 2026-05-16 → 14 days past (< 30).
    service = SuccessFactorsLifecycleService(
        db_session, sf_client=client, today=date(2026, 5, 30),
    )
    batch = await service.preview_full_employee_sync(triggered_by="test")
    await service.confirm_batch(batch.id, confirmed_by="test")

    recent = (await db_session.execute(
        select(IdentityPerson).where(IdentityPerson.hris_person_id == "P_RECENT")
    )).scalar_one()
    assert recent.employment_status == "leaver_pending", (
        f"Recent leaver (14d) must land at 'leaver_pending'; got "
        f"{recent.employment_status!r}. The 30-day cutoff exists so "
        f"in-flight offboarding actions still fire."
    )
    assert recent.termination_date == date(2026, 5, 16)


@pytest.mark.asyncio
async def test_confirm_full_sync_active_pulls_work_start_date_from_sf(
    db_session,
) -> None:
    """Regression for the 年假计算器 bug (2026-06-02).

    Two-pass narrative:

    1. Initial PR pulled EmpEmployment.originalStartDate for active
       users. That made 47/432 prod rows show a real career start —
       but those 47 turned out to all be PingCAP REHIRES (SF's
       originalStartDate is "first start at THIS company", 司龄 anchor).
       The other 383 still showed work_start_date == hire_date.
    2. Direct SF dump for P0989 revealed PingCAP's true 参加工作日期 is
       **EmpEmployment.firstDateWorked** (2011-11-11 for shuai.feng vs
       hireDate 2022-06-30 = 11-year career outside PingCAP). 328/500
       (66%) populated in prod. The fix flips priority to firstDateWorked
       and keeps originalStartDate / user.hireDate only as fallbacks.

    Assertions pin both the primary source AND the canonical 司龄
    anchor (``seniority_date``, migration 239).
    """
    client = _FakeFullSyncClient()
    service = SuccessFactorsLifecycleService(
        db_session, sf_client=client, today=date(2026, 5, 30),
    )
    batch = await service.preview_full_employee_sync(triggered_by="test")
    result = await service.confirm_batch(batch.id, confirmed_by="test")
    assert result["error_count"] == 0, f"confirm_batch reported errors: {result}"

    active = (await db_session.execute(
        select(IdentityPerson).where(IdentityPerson.hris_person_id == "P_ACTIVE")
    )).scalar_one()
    # hire_date = SF hireDate 2023-11-14 (current employment startDate).
    assert active.hire_date == date(2023, 11, 14), (
        f"hire_date should be SF startDate; got {active.hire_date}"
    )
    # work_start_date MUST come from firstDateWorked (2011-11-11),
    # NOT from originalStartDate (2020-01-01) nor user.hireDate
    # (2023-11-14). This is the load-bearing assertion: if a future
    # change demotes firstDateWorked to a fallback again, every
    # employee with prior career experience silently loses 11+ years
    # of social seniority. The 年假计算器 visibly mis-reports their
    # 福利年假 tier the next FY.
    assert active.work_start_date == date(2011, 11, 11), (
        f"work_start_date must come from EmpEmployment.firstDateWorked "
        f"(2011-11-11), NOT originalStartDate (2020-01-01) nor "
        f"user.hireDate (2023-11-14). Got {active.work_start_date} — "
        f"prod 2026-06-02 regression has returned."
    )
    # seniority_date = SF seniorityDate 2023-11-14 (司龄 anchor =
    # current PingCAP stint start, since this fixture models a rehire
    # where SF resets seniority to the latest employment).
    assert active.seniority_date == date(2023, 11, 14), (
        f"seniority_date should be SF seniorityDate; got "
        f"{active.seniority_date}"
    )


@pytest.mark.asyncio
async def test_full_sync_enrichment_failure_is_best_effort(
    db_session,
) -> None:
    """Regression (2026-06-02 review sweep finding #21).

    The all-users EmpEmployment enrichment fetch (work_start_date /
    originalStartDate source) is best-effort: a raised chunk MUST NOT
    abort the whole sync. Critical leaver detection — the entire reason
    the bootstrap exists (wangdi case) — must still produce the
    forfeited offboard events, with active work_start_date falling back
    to hire_date.

    The forfeited-subset fetch (termination_date source) stays critical
    and is a SEPARATE call over only ``status='f'`` userIds, so it
    succeeds even when the all-users enrichment call raises.
    """
    client = _FakeFullSyncClient()
    orig_get = client.get_employments_for_users

    async def flaky(user_ids):
        # The best-effort enrichment fetch covers ALL users (incl. the
        # active P_ACTIVE). Raise on that call only; let the critical
        # forfeited-only fetch (P0175 / P_RECENT, no P_ACTIVE) succeed.
        if "P_ACTIVE" in user_ids:
            raise RuntimeError("SF EmpEmployment chunk blew up")
        return await orig_get(user_ids)

    client.get_employments_for_users = flaky  # type: ignore[assignment]

    service = SuccessFactorsLifecycleService(
        db_session, sf_client=client, today=date(2026, 5, 30),
    )
    batch = await service.preview_full_employee_sync(triggered_by="test")
    result = await service.confirm_batch(batch.id, confirmed_by="test")
    assert result["error_count"] == 0, f"confirm_batch reported errors: {result}"

    # Critical leaver detection survived the enrichment failure.
    staged_rows = (await db_session.execute(
        select(HrSyncStagedEvent).where(HrSyncStagedEvent.batch_id == batch.id)
    )).scalars().all()
    by_user = {r.user_id: r.event_type for r in staged_rows}
    assert by_user.get("P0175") == "offboard", (
        "Forfeited offboard event must still be produced when the "
        "best-effort work_start_date enrichment fetch raises."
    )
    assert by_user.get("P_RECENT") == "offboard"
    assert by_user.get("P_ACTIVE") == "onboard"

    wangdi = (await db_session.execute(
        select(IdentityPerson).where(IdentityPerson.hris_person_id == "P0175")
    )).scalar_one()
    assert wangdi.employment_status == "left"
    assert wangdi.termination_date == date(2026, 3, 20)

    # Active employee falls back to hire_date for work_start_date
    # because the enrichment dict ended up empty.
    active = (await db_session.execute(
        select(IdentityPerson).where(IdentityPerson.hris_person_id == "P_ACTIVE")
    )).scalar_one()
    assert active.hire_date == date(2023, 11, 14)
    assert active.work_start_date == date(2023, 11, 14), (
        f"With enrichment failed, work_start_date must fall back to "
        f"hire_date (2023-11-14); got {active.work_start_date}."
    )


@pytest.mark.asyncio
async def test_work_start_date_falls_back_to_original_start_when_no_first_worked(
    db_session,
) -> None:
    """Tier 2 of the fallback chain: when SF doesn't carry
    ``firstDateWorked`` but does carry a non-trivial
    ``originalStartDate`` (rehire shape), we still want the most
    historical signal available. This is the 47/432 case from the
    2026-06-02 prod snapshot — rehires whose original PingCAP stint
    started years before the current one.
    """
    # Build a fixture variant: P_ACTIVE with NO firstDateWorked.
    client = _FakeFullSyncClient()
    orig_get = client.get_employments_for_users

    async def patched(user_ids):
        rows = await orig_get(user_ids)
        for r in rows:
            if r["userId"] == "P_ACTIVE":
                r.pop("firstDateWorked", None)
        return rows

    client.get_employments_for_users = patched  # type: ignore[assignment]

    service = SuccessFactorsLifecycleService(
        db_session, sf_client=client, today=date(2026, 5, 30),
    )
    batch = await service.preview_full_employee_sync(triggered_by="test")
    await service.confirm_batch(batch.id, confirmed_by="test")

    active = (await db_session.execute(
        select(IdentityPerson).where(IdentityPerson.hris_person_id == "P_ACTIVE")
    )).scalar_one()
    # Falls back to originalStartDate 2020-01-01 (not hireDate
    # 2023-11-14), preserving the 3+ years of pre-current-stint
    # PingCAP tenure as 社会工龄.
    assert active.work_start_date == date(2020, 1, 1), (
        f"Fallback tier 2: work_start_date should come from "
        f"originalStartDate (2020-01-01) when SF doesn't carry "
        f"firstDateWorked. Got {active.work_start_date}."
    )


class _FakeFutureHireClient(_FakeFullSyncClient):
    """One ``status='f'`` user whose ``hireDate`` is in the FUTURE — a
    not-yet-started hire, NOT a leaver. SF marks pre-start hires inactive
    (``status='f'``), the same flag it uses for forfeited/terminated
    people, so the full sync must not confuse the two.
    """

    async def list_all_users(self) -> list[dict]:
        self.list_all_called = True
        # hireDate 2026-08-01 (future vs today=2026-07-05). status='f'
        # because she hasn't started yet — mirrors Camila P10009.
        return [{
            "userId": "P_FUTURE",
            "displayName": "Future Hire",
            "email": "future@pingcap.com",
            "empId": "P_FUTURE",
            "status": "f",
            "hireDate": "/Date(1785283200000)/",  # 2026-08-01
            "lastModifiedDateTime": "/Date(1783000000000+0000)/",
        }]

    async def get_employments_for_users(self, user_ids):
        self.get_employments_called_with = list(user_ids)
        # A nonsensical endDate that PRECEDES the future startDate — the
        # exact artifact shape that tricked the old logic into staging an
        # offboard for Camila (endDate ~07-01 while hire_date was 07-06).
        return [{
            "personIdExternal": "P_FUTURE",
            "userId": "P_FUTURE",
            "startDate": "/Date(1785283200000)/",   # 2026-08-01
            "endDate": "/Date(1782000000000)/",      # ~2026-06-20
            "lastModifiedDateTime": "/Date(1783000000000+0000)/",
        }]


@pytest.mark.asyncio
async def test_preview_full_sync_skips_future_dated_forfeited(
    db_session,
) -> None:
    """Regression (2026-07-06 Camila P10009). A ``status='f'`` user with a
    FUTURE ``hireDate`` is a not-yet-started hire, not a leaver. The full
    sync must NOT stage an offboard for them — otherwise, in the
    CN+EU mirrored-org deployment, whichever tenant's full sync runs first
    creates the hire as a leaver owned by the WRONG tenant, and the
    owning tenant's onboard can't recover them.
    """
    client = _FakeFutureHireClient()
    service = SuccessFactorsLifecycleService(
        db_session, sf_client=client, today=date(2026, 7, 5),
    )
    batch = await service.preview_full_employee_sync(triggered_by="test")

    staged_rows = (await db_session.execute(
        select(HrSyncStagedEvent).where(HrSyncStagedEvent.batch_id == batch.id)
    )).scalars().all()
    assert staged_rows == [], (
        "A future-dated status='f' hire must not be staged at all — the "
        f"owning instance's onboard event creates them. Got: "
        f"{[(r.user_id, r.event_type, r.action) for r in staged_rows]}"
    )

    # And no leaver row was created.
    person = (await db_session.execute(
        select(IdentityPerson).where(IdentityPerson.hris_person_id == "P_FUTURE")
    )).scalar_one_or_none()
    assert person is None, (
        "Future-dated forfeited hire must not be created as a leaver row."
    )


class _FakeDayOneHireClient(_FakeFullSyncClient):
    """One ``status='f'`` user whose ``hireDate`` is TODAY and who carries
    NO termination EmpEmployment at all — the exact Robert Wood P10013
    shape (2026-07-15). A scheduled full sync that fires in the Asia/Shanghai
    small hours sees ``self.today`` already rolled onto the hire's own start
    date; the pre-fix strict ``>`` boundary let this not-yet-started hire
    fall through to the offboard branch, which then faked a leaver_date from
    lastModifiedDateTime (no real endDate existed).
    """

    async def list_all_users(self) -> list[dict]:
        self.list_all_called = True
        # hireDate 2026-07-13 == today the test injects below. status='f'
        # because the overseas hire hasn't started in their own tz yet.
        return [{
            "userId": "P10013",
            "displayName": "Robert Wood",
            "email": "robert.wood@pingcap.com",
            "empId": "P10013",
            "status": "f",
            "hireDate": "/Date(1783900800000)/",  # 2026-07-13
            "lastModifiedDateTime": "/Date(1783875493000+0000)/",  # 07-12
        }]

    async def get_employments_for_users(self, user_ids):
        # No EmpEmployment / termination record at all — the empty
        # ``_jobs_snapshot`` shape that forced the bogus lastModifiedDateTime
        # fallback for P10013.
        self.get_employments_called_with = list(user_ids)
        return []


@pytest.mark.asyncio
async def test_preview_full_sync_skips_day_one_forfeited_hire(
    db_session,
) -> None:
    """Regression (2026-07-15 Robert Wood P10013). A ``status='f'`` user
    whose ``hireDate == self.today`` (the Shanghai date rolled onto their
    start day while they're still pre-start in their own tz) must NOT be
    staged as an offboard. The bug: full sync used strict ``>`` while its
    incremental twin already used ``>=``; on the boundary day the future-hire
    skip failed and a not-yet-started hire was synthesized as a leaver with a
    faked termination date.
    """
    client = _FakeDayOneHireClient()
    service = SuccessFactorsLifecycleService(
        db_session, sf_client=client, today=date(2026, 7, 13),
    )
    batch = await service.preview_full_employee_sync(triggered_by="test")

    staged_rows = (await db_session.execute(
        select(HrSyncStagedEvent).where(HrSyncStagedEvent.batch_id == batch.id)
    )).scalars().all()
    assert staged_rows == [], (
        "A day-one status='f' hire (hireDate == today) must not be staged — "
        "with strict '>' this regressed into a bogus offboard. Got: "
        f"{[(r.user_id, r.event_type, r.action) for r in staged_rows]}"
    )

    person = (await db_session.execute(
        select(IdentityPerson).where(IdentityPerson.hris_person_id == "P10013")
    )).scalar_one_or_none()
    assert person is None, (
        "Day-one forfeited hire must not be created/flipped to a leaver row."
    )


@pytest.mark.asyncio
async def test_preview_full_sync_still_offboards_past_dated_forfeited(
    db_session,
) -> None:
    """Guard the guard: a genuine past leaver (``status='f'`` with a
    hireDate in the PAST) MUST still be routed through offboard. The
    future-hire skip must not swallow real leavers (the wangdi case)."""
    client = _FakeFullSyncClient()  # P0175 / P_RECENT have past hireDates
    service = SuccessFactorsLifecycleService(
        db_session, sf_client=client, today=date(2026, 5, 30),
    )
    batch = await service.preview_full_employee_sync(triggered_by="test")
    staged_rows = (await db_session.execute(
        select(HrSyncStagedEvent).where(HrSyncStagedEvent.batch_id == batch.id)
    )).scalars().all()
    by_user = {r.user_id: r.event_type for r in staged_rows}
    assert by_user.get("P0175") == "offboard"
    assert by_user.get("P_RECENT") == "offboard"


class _FakePastForfeitedNoEndDateClient(_FakeFullSyncClient):
    """A ``status='f'`` user with a PAST hireDate but NO EmpEmployment endDate
    at all — SF data gap. Pre-fix, full sync FABRICATED a termination_date from
    ``lastModifiedDateTime`` and offboarded them. The systemic fix refuses to
    invent a leaver: no substantiated endDate → stage nothing (surfaced by the
    hris_status_divergence DQ check instead). This aligns full sync with the
    incremental path, which already skipped this shape.
    """

    async def list_all_users(self) -> list[dict]:
        self.list_all_called = True
        return [{
            "userId": "P_GAP",
            "displayName": "Data Gap Leaver",
            "email": "gap@pingcap.com",
            "empId": "P_GAP",
            "status": "f",
            "hireDate": "/Date(1600000000000)/",  # 2020-09-13 (past)
            "lastModifiedDateTime": "/Date(1776514800000+0000)/",
        }]

    async def get_employments_for_users(self, user_ids):
        self.get_employments_called_with = list(user_ids)
        return []  # no endDate-bearing employment → unsubstantiated


@pytest.mark.asyncio
async def test_preview_full_sync_no_fabricated_offboard_without_end_date(
    db_session,
) -> None:
    """Regression (2026-07-15 systemic fix). A past-dated ``status='f'`` user
    with NO real termination endDate must NOT be offboarded — the full-sync
    ``lastModifiedDateTime`` fabrication is removed. Stage nothing; no leaver
    row created."""
    client = _FakePastForfeitedNoEndDateClient()
    service = SuccessFactorsLifecycleService(
        db_session, sf_client=client, today=date(2026, 7, 15),
    )
    batch = await service.preview_full_employee_sync(triggered_by="test")
    staged_rows = (await db_session.execute(
        select(HrSyncStagedEvent).where(HrSyncStagedEvent.batch_id == batch.id)
    )).scalars().all()
    assert staged_rows == [], (
        "Forfeited-without-endDate must NOT be offboarded (no fabrication). "
        f"Got: {[(r.user_id, r.event_type, r.action) for r in staged_rows]}"
    )
    person = (await db_session.execute(
        select(IdentityPerson).where(IdentityPerson.hris_person_id == "P_GAP")
    )).scalar_one_or_none()
    assert person is None


class _FakeImpossibleEndDateClient(_FakeFullSyncClient):
    """``status='f'``, PAST hireDate, but a real endDate that PRECEDES the
    hireDate — a logically impossible leaver. The classifier (and the
    ``_stage_event`` backstop) must refuse the offboard."""

    async def list_all_users(self) -> list[dict]:
        self.list_all_called = True
        return [{
            "userId": "P_IMPOSSIBLE",
            "displayName": "Impossible Dates",
            "email": "impossible@pingcap.com",
            "empId": "P_IMPOSSIBLE",
            "status": "f",
            "hireDate": "/Date(1704067200000)/",  # 2024-01-01
            "lastModifiedDateTime": "/Date(1776514800000+0000)/",
        }]

    async def get_employments_for_users(self, user_ids):
        self.get_employments_called_with = list(user_ids)
        return [{
            "personIdExternal": "P_IMPOSSIBLE",
            "userId": "P_IMPOSSIBLE",
            "startDate": "/Date(1704067200000)/",   # 2024-01-01
            "endDate": "/Date(1700000000000)/",      # 2023-11-14 — before hire
            "lastModifiedDateTime": "/Date(1776514800000+0000)/",
        }]


@pytest.mark.asyncio
async def test_preview_full_sync_refuses_impossible_offboard(
    db_session,
) -> None:
    """Regression (2026-07-15 systemic fix). termination_date < hire_date is
    impossible — refuse the offboard, create no leaver row."""
    client = _FakeImpossibleEndDateClient()
    service = SuccessFactorsLifecycleService(
        db_session, sf_client=client, today=date(2026, 7, 15),
    )
    batch = await service.preview_full_employee_sync(triggered_by="test")
    staged_rows = (await db_session.execute(
        select(HrSyncStagedEvent).where(HrSyncStagedEvent.batch_id == batch.id)
    )).scalars().all()
    offboards = [r for r in staged_rows if r.event_type == "offboard"]
    assert offboards == [], (
        "An offboard whose termination_date precedes hire_date is impossible "
        f"and must be refused. Got: {[(r.user_id, r.action) for r in offboards]}"
    )
    person = (await db_session.execute(
        select(IdentityPerson).where(
            IdentityPerson.hris_person_id == "P_IMPOSSIBLE"
        )
    )).scalar_one_or_none()
    assert person is None