"""The Feishu sweep: what it discovers, what it refuses to overwrite,
and the wire facts it is built on.

The stub client enforces the real endpoints' contracts rather than
politely accepting anything — a test double that does not enforce the
vendor's limits is how a wire-format bug ships green (the 50-char
``uuid`` cap on Feishu messages did exactly that).
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta

import pytest

from app.domains.approval_registry import constants as const
from app.domains.approval_registry import discovery
from app.domains.approval_registry import service as svc
from app.identity.person_models import IdentityPerson

CODE_A = "AAAAAAAA-1111-2222-3333-444444444444"
CODE_B = "BBBBBBBB-1111-2222-3333-444444444444"


def _ms(dt: datetime) -> str:
    return str(int(dt.timestamp() * 1000))


class StubClient:
    """Enough of ``FeishuApprovalClient`` to drive the sweep."""

    def __init__(
        self, *, instances=None, tasks=None, definitions=None, users=None,
        lookup_failures=None,
    ):
        self._instances = instances or {}
        self._tasks = tasks or {}
        self._definitions = definitions or {}
        self._users = users or {}
        # user_ids whose contact lookup BROKE (timeout / 5xx), as
        # opposed to ones Feishu says are gone. The real client keeps
        # these apart and so must the double: collapsing them is exactly
        # the bug the guard below exists for.
        self._lookup_failures = set(lookup_failures or ())
        self.definition_calls: list[tuple[str, bool]] = []
        self.admin_id_flags: list[bool] = []
        self.page_sizes: list[int] = []

    async def batch_resolve_user_ids_by_email(self, emails):
        # Feishu caps the batch at 50; a double that ignores the cap
        # would let a caller ship a request the real API rejects whole.
        assert len(emails) <= 500
        return {e.lower(): f"uid-{e.split('@')[0]}" for e in emails}

    async def query_instances_for_user(
        self, user_id, *, page_size=200, max_pages=30,
        start_time_ms=None, end_time_ms=None,
    ):
        assert page_size <= 200, "Feishu rejects page_size > 200"
        # Feishu rejects a half-open window outright.
        assert (start_time_ms is None) == (end_time_ms is None)
        self.page_sizes.append(page_size)
        return list(self._instances.get(user_id, [])), False

    async def search_tasks_for_user(self, user_id, *, page_size=200, max_pages=30):
        assert page_size <= 200
        return list(self._tasks.get(user_id, [])), False

    async def get_approval_definition(
        self, code, *, subscribe=True, with_admin_id=False,
        user_id_type="user_id",
    ):
        self.definition_calls.append((code, subscribe))
        self.admin_id_flags.append(with_admin_id)
        payload = self._definitions.get(code)
        if payload is None:
            from app.core.exceptions import AppException

            raise AppException(
                code="FEISHU_API_ERROR",
                message="Feishu API error 1390002: approval not found",
                status_code=502,
            )
        # THE contract this double exists to enforce. Feishu omits
        # ``approval_admin_ids`` from the response entirely unless the
        # request carries ``with_admin_id=true`` — a double that handed
        # the field over unconditionally would let a caller that forgot
        # the param pass green, which is exactly how the register
        # published 「还没人认领」 for 32 flows whose managers Feishu knew.
        out = dict(payload)
        if not with_admin_id:
            out.pop("approval_admin_ids", None)
        return out

    async def batch_get_user_profiles(self, user_ids):
        return (
            {
                u: self._users[u] for u in user_ids
                if u in self._users and u not in self._lookup_failures
            },
            {u for u in user_ids if u in self._lookup_failures},
        )


def _instance(code, name, group, *, instance_code, started=None):
    return {
        "approval": {"code": code, "name": name},
        "group": {"name": group},
        "instance": {
            "code": instance_code,
            "start_time": _ms(started or datetime.now(UTC)),
            "status": "APPROVED",
        },
    }


def _task(code, *, status="APPROVED", ended=None):
    return {
        "approval": {"code": code, "name": "x"},
        # Deliberately present and WRONG: the sweep must never read a
        # group from a task. Measured on the live tenant, 13 of 25
        # definitions returned conflicting group names here.
        "group": {"name": "WRONG-GROUP"},
        "task": {
            "status": status,
            "end_time": _ms(ended or datetime.now(UTC)),
        },
    }


async def _person(db_session, email, *, name="P", status="active"):
    person = IdentityPerson(
        id=uuid.uuid4(), work_email=email, display_name=name,
        employment_status=status, source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:10]}",
    )
    db_session.add(person)
    await db_session.flush()
    await db_session.commit()
    return person


def _install(monkeypatch, stub):
    async def _fake_client(_db):
        return stub

    monkeypatch.setattr(discovery, "_client", _fake_client)


# ── Discovery ───────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_discovery_registers_flows_and_takes_group_from_instances(
    db_session, monkeypatch,
):
    person = await _person(db_session, f"a-{uuid.uuid4().hex[:6]}@test.com")
    uid = f"uid-{person.work_email.split('@')[0]}"
    stub = StubClient(
        instances={
            uid: [
                _instance(CODE_A, "差旅报销", "Finance", instance_code="i-1"),
                _instance(CODE_A, "差旅报销", "Finance", instance_code="i-2"),
                _instance(CODE_B, "入职申请", "HR&Admin", instance_code="i-3"),
            ],
        },
        definitions={
            CODE_A: {"approval_name": "差旅报销", "status": "ACTIVE"},
            CODE_B: {"approval_name": "入职申请", "status": "ACTIVE"},
        },
    )
    _install(monkeypatch, stub)

    result = await discovery.run_sync(db_session, infer_owners=False)
    assert result["flows_discovered"] == 2
    assert result["flows_created"] == 2

    flow = await svc.get_by_code(db_session, CODE_A)
    assert flow is not None
    assert flow.source_group == "Finance"
    # Seeded from upstream, and editable afterwards — the console's
    # taxonomy and the API's disagree in the real tenant.
    assert flow.category == "Finance"
    assert flow.observed_instances == 2
    assert flow.source == const.SOURCE_DISCOVERED


@pytest.mark.asyncio
async def test_the_same_instance_seen_by_two_people_counts_once(
    db_session, monkeypatch,
):
    """An instance surfaces once per participant. Counting rows would
    make a flow with many approvers look busier than one with many
    applicants purely because more people can see it."""
    one = await _person(db_session, f"b1-{uuid.uuid4().hex[:6]}@test.com")
    two = await _person(db_session, f"b2-{uuid.uuid4().hex[:6]}@test.com")
    shared = _instance(CODE_A, "共享实例", "IT", instance_code="i-shared")
    stub = StubClient(
        instances={
            f"uid-{one.work_email.split('@')[0]}": [shared],
            f"uid-{two.work_email.split('@')[0]}": [shared],
        },
        definitions={CODE_A: {"approval_name": "共享实例", "status": "ACTIVE"}},
    )
    _install(monkeypatch, stub)

    await discovery.run_sync(db_session, infer_owners=False)
    flow = await svc.get_by_code(db_session, CODE_A)
    assert flow.observed_instances == 1


@pytest.mark.asyncio
async def test_definition_sync_never_subscribes_to_other_teams_flows(
    db_session, monkeypatch,
):
    """Subscribing would point every HR / finance / legal instance event
    at Helix's webhook."""
    person = await _person(db_session, f"c-{uuid.uuid4().hex[:6]}@test.com")
    uid = f"uid-{person.work_email.split('@')[0]}"
    stub = StubClient(
        instances={uid: [_instance(CODE_A, "薪资调整", "HR", instance_code="i-9")]},
        definitions={CODE_A: {"approval_name": "薪资调整", "status": "ACTIVE"}},
    )
    _install(monkeypatch, stub)

    await discovery.run_sync(db_session, infer_owners=False)
    assert stub.definition_calls
    assert all(subscribe is False for _code, subscribe in stub.definition_calls)


@pytest.mark.asyncio
async def test_upstream_rename_and_stop_are_events_not_silent_writes(
    db_session, monkeypatch,
):
    person = await _person(db_session, f"d-{uuid.uuid4().hex[:6]}@test.com")
    uid = f"uid-{person.work_email.split('@')[0]}"
    stub = StubClient(
        instances={uid: [_instance(CODE_A, "旧名字", "IT", instance_code="i-1")]},
        definitions={CODE_A: {"approval_name": "旧名字", "status": "ACTIVE"}},
    )
    _install(monkeypatch, stub)
    await discovery.run_sync(db_session, infer_owners=False)

    stub._definitions[CODE_A] = {"approval_name": "新名字", "status": "INACTIVE"}
    await discovery.run_sync(db_session, discover=False, infer_owners=False)

    flow = await svc.get_by_code(db_session, CODE_A)
    assert flow.name == "新名字"
    assert flow.status == const.STATUS_INACTIVE
    assert flow.feishu_status == "INACTIVE"
    events = await svc.list_events(db_session, flow.id)
    kinds = {e.event_type for e in events}
    assert const.EVENT_RENAMED in kinds
    assert const.EVENT_STATUS_CHANGED in kinds


@pytest.mark.asyncio
async def test_an_unmapped_upstream_status_leaves_ours_alone(
    db_session, monkeypatch,
):
    """Guessing "active" for a value Feishu invents later would
    resurrect a stopped flow on the all-staff page."""
    person = await _person(db_session, f"e-{uuid.uuid4().hex[:6]}@test.com")
    uid = f"uid-{person.work_email.split('@')[0]}"
    stub = StubClient(
        instances={uid: [_instance(CODE_A, "流程", "IT", instance_code="i-1")]},
        definitions={CODE_A: {"approval_name": "流程", "status": "ACTIVE"}},
    )
    _install(monkeypatch, stub)
    await discovery.run_sync(db_session, infer_owners=False)
    flow = await svc.get_by_code(db_session, CODE_A)
    await svc.update_flow(db_session, flow, status=const.STATUS_INACTIVE)
    await db_session.commit()

    stub._definitions[CODE_A] = {"approval_name": "流程", "status": "SOMETHING_NEW"}
    await discovery.run_sync(db_session, discover=False, infer_owners=False)

    flow = await svc.get_by_code(db_session, CODE_A)
    assert flow.status == const.STATUS_INACTIVE
    assert flow.feishu_status == "SOMETHING_NEW"


@pytest.mark.asyncio
async def test_a_definition_that_404s_is_recorded_not_swallowed(
    db_session, monkeypatch,
):
    person = await _person(db_session, f"f-{uuid.uuid4().hex[:6]}@test.com")
    uid = f"uid-{person.work_email.split('@')[0]}"
    stub = StubClient(
        instances={uid: [_instance(CODE_A, "流程", "IT", instance_code="i-1")]},
        definitions={},  # every GET raises
    )
    _install(monkeypatch, stub)
    result = await discovery.run_sync(db_session, infer_owners=False)

    assert result["definitions_failed"] == 1
    flow = await svc.get_by_code(db_session, CODE_A)
    assert flow.last_sync_error is not None
    assert flow.last_synced_at is None
    # The flow still exists and still has its discovered name: an
    # unreadable definition is a signal, not a reason to drop the row.
    assert flow.name == "流程"


# ── Owner inference ─────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_a_dominant_approver_becomes_a_candidate_owner(
    db_session, monkeypatch,
):
    boss = await _person(db_session, f"g-{uuid.uuid4().hex[:6]}@test.com", name="老板")
    uid = f"uid-{boss.work_email.split('@')[0]}"
    stub = StubClient(
        instances={uid: [_instance(CODE_A, "预算审批", "Finance", instance_code="i-1")]},
        tasks={uid: [_task(CODE_A) for _ in range(5)]},
        definitions={CODE_A: {"approval_name": "预算审批", "status": "ACTIVE"}},
    )
    _install(monkeypatch, stub)
    result = await discovery.run_sync(db_session)

    assert result["owners_inferred"] == 1
    flow = await svc.get_by_code(db_session, CODE_A)
    assert flow.owner_person_id == boss.id
    assert flow.owner_source == const.OWNER_SOURCE_INFERRED
    # Never presented as settled: no confirmation stamp, and the
    # evidence travels with the name.
    assert flow.owner_confirmed_at is None
    assert flow.inference_meta["task_count"] == 5
    assert flow.inference_meta["share_pct"] == 100.0


@pytest.mark.asyncio
async def test_a_flow_everybody_approves_stays_unclaimed(db_session, monkeypatch):
    """A "line manager approves" flow spreads decisions over dozens of
    people; the busiest of them has no say over the flow itself."""
    people = [
        await _person(db_session, f"h{i}-{uuid.uuid4().hex[:6]}@test.com")
        for i in range(4)
    ]
    uids = [f"uid-{p.work_email.split('@')[0]}" for p in people]
    stub = StubClient(
        instances={
            uids[0]: [_instance(CODE_A, "请假", "HR", instance_code="i-1")],
        },
        tasks={u: [_task(CODE_A) for _ in range(3)] for u in uids},
        definitions={CODE_A: {"approval_name": "请假", "status": "ACTIVE"}},
    )
    _install(monkeypatch, stub)
    result = await discovery.run_sync(db_session)

    assert result["owners_inferred"] == 0
    flow = await svc.get_by_code(db_session, CODE_A)
    assert flow.owner_person_id is None
    assert flow.owner_source == const.OWNER_SOURCE_UNASSIGNED


@pytest.mark.asyncio
async def test_pending_tasks_do_not_hand_anyone_a_flow(db_session, monkeypatch):
    """A pending task says the flow was routed to them, not that they
    decided anything."""
    person = await _person(db_session, f"i-{uuid.uuid4().hex[:6]}@test.com")
    uid = f"uid-{person.work_email.split('@')[0]}"
    stub = StubClient(
        instances={uid: [_instance(CODE_A, "流程", "IT", instance_code="i-1")]},
        tasks={uid: [_task(CODE_A, status="PENDING") for _ in range(9)]},
        definitions={CODE_A: {"approval_name": "流程", "status": "ACTIVE"}},
    )
    _install(monkeypatch, stub)
    result = await discovery.run_sync(db_session)

    assert result["owners_inferred"] == 0
    flow = await svc.get_by_code(db_session, CODE_A)
    assert flow.owner_person_id is None


@pytest.mark.asyncio
async def test_decisions_older_than_the_window_do_not_count(
    db_session, monkeypatch,
):
    person = await _person(db_session, f"j-{uuid.uuid4().hex[:6]}@test.com")
    uid = f"uid-{person.work_email.split('@')[0]}"
    long_ago = datetime.now(UTC) - timedelta(
        days=const.INFERENCE_WINDOW_DAYS + 30,
    )
    stub = StubClient(
        instances={uid: [_instance(CODE_A, "流程", "IT", instance_code="i-1")]},
        tasks={uid: [_task(CODE_A, ended=long_ago) for _ in range(8)]},
        definitions={CODE_A: {"approval_name": "流程", "status": "ACTIVE"}},
    )
    _install(monkeypatch, stub)
    result = await discovery.run_sync(db_session)

    assert result["owners_inferred"] == 0


@pytest.mark.asyncio
async def test_inference_never_overwrites_a_human_decision(
    db_session, monkeypatch,
):
    """The sweep fills blanks; it does not re-litigate decisions."""
    named = await _person(db_session, f"k1-{uuid.uuid4().hex[:6]}@test.com", name="被指派的")
    approver = await _person(db_session, f"k2-{uuid.uuid4().hex[:6]}@test.com", name="常批的")
    flow = await svc.create_flow(
        db_session, approval_code=CODE_A, name="预置流程",
        owner_person_id=named.id,
    )
    await db_session.commit()
    assert flow.owner_source == const.OWNER_SOURCE_MANUAL

    uid = f"uid-{approver.work_email.split('@')[0]}"
    stub = StubClient(
        instances={uid: [_instance(CODE_A, "预置流程", "IT", instance_code="i-1")]},
        tasks={uid: [_task(CODE_A) for _ in range(20)]},
        definitions={CODE_A: {"approval_name": "预置流程", "status": "ACTIVE"}},
    )
    _install(monkeypatch, stub)
    result = await discovery.run_sync(db_session)

    assert result["owners_inferred"] == 0
    flow = await svc.get_by_code(db_session, CODE_A)
    assert flow.owner_person_id == named.id
    assert flow.owner_source == const.OWNER_SOURCE_MANUAL


@pytest.mark.asyncio
async def test_inference_refuses_to_name_someone_who_has_left(db_session):
    """Inferring a leaver is worse than inferring nobody: it fills the
    blank the coverage read would otherwise chase.

    Driven straight at the write rather than through the sweep: the
    roster already excludes departed people, so a run_sync-based test
    would pass with this guard deleted — the fixture would have made the
    guarded branch unreachable.
    """
    gone = await _person(
        db_session, f"l-{uuid.uuid4().hex[:6]}@test.com",
        status="leaver_pending",
    )
    flow = await svc.create_flow(
        db_session, approval_code=CODE_A, name="无主流程",
    )
    await db_session.commit()

    changed = await svc.apply_inferred_owner(
        db_session, flow, person=gone,
        evidence={"task_count": 99, "share_pct": 100.0},
    )
    assert changed is False
    assert flow.owner_person_id is None
    assert flow.owner_source == const.OWNER_SOURCE_UNASSIGNED


@pytest.mark.asyncio
async def test_client_skips_subscription_when_asked_to(monkeypatch):
    """The caller passing ``subscribe=False`` only helps if the client
    honours it — the two are verified separately because a stub client
    cannot prove anything about the real one."""
    from app.approval.feishu_client import FeishuApprovalClient

    client = FeishuApprovalClient(app_id="cli_x", app_secret="secret")
    calls: list[str] = []

    async def _fake_request(method, path, **kwargs):
        return {"approval_name": "流程", "status": "ACTIVE"}

    async def _fake_subscribe(code):
        calls.append(code)

    monkeypatch.setattr(client, "_request", _fake_request)
    monkeypatch.setattr(client, "ensure_subscribed", _fake_subscribe)

    await client.get_approval_definition(CODE_A, subscribe=False)
    assert calls == []

    # The platform's own flows still subscribe — the default is unchanged.
    await client.get_approval_definition(CODE_A)
    assert calls == [CODE_A]


# ── 流程管理员 (approval_admin_ids) ──────────────────────────────────
#
# The register's first release published 「还没人认领」 on 32 of 49 flows
# because the definition sweep called Feishu without ``with_admin_id``,
# saw no ``approval_admin_ids`` key, and concluded the concept did not
# exist. These tests pin both halves: that the param is sent, and that
# what comes back is read the way the domain says it should be.


def _profile(name, email, *, resigned=False):
    return {"name": name, "email": email, "resigned": resigned}


@pytest.mark.asyncio
async def test_the_sweep_asks_feishu_for_the_flow_admins(
    db_session, monkeypatch,
):
    """Regression for the whole bug: without the query param the
    response has no admin key at all, and the register reads that
    absence as 「飞书没有负责人」."""
    person = await _person(db_session, f"a-{uuid.uuid4().hex[:6]}@test.com")
    uid = f"uid-{person.work_email.split('@')[0]}"
    stub = StubClient(
        instances={uid: [_instance(CODE_A, "流程", "IT", instance_code="i-1")]},
        definitions={
            CODE_A: {
                "approval_name": "流程", "status": "ACTIVE",
                "approval_admin_ids": ["fs-owner"],
            },
        },
        users={"fs-owner": _profile("管理员", "admin@test.com")},
    )
    _install(monkeypatch, stub)
    await discovery.run_sync(db_session, infer_owners=False)

    assert stub.admin_id_flags and all(stub.admin_id_flags)
    flow = await svc.get_by_code(db_session, CODE_A)
    assert [a["name"] for a in flow.feishu_admins] == ["管理员"]
    # 「asked and got none」 must be distinguishable from 「never asked」.
    assert flow.feishu_admins_synced_at is not None


@pytest.mark.asyncio
async def test_a_lone_admin_who_works_here_becomes_the_candidate_owner(
    db_session, monkeypatch,
):
    admin = await _person(
        db_session, f"solo-{uuid.uuid4().hex[:6]}@test.com", name="唯一管理员",
    )
    person = await _person(db_session, f"b-{uuid.uuid4().hex[:6]}@test.com")
    uid = f"uid-{person.work_email.split('@')[0]}"
    stub = StubClient(
        instances={uid: [_instance(CODE_A, "流程", "IT", instance_code="i-1")]},
        definitions={
            CODE_A: {
                "approval_name": "流程", "status": "ACTIVE",
                "approval_admin_ids": ["fs-solo"],
            },
        },
        users={"fs-solo": _profile("唯一管理员", admin.work_email)},
    )
    _install(monkeypatch, stub)
    summary = await discovery.run_sync(db_session, infer_owners=False)

    assert summary["owners_from_feishu_admin"] == 1
    flow = await svc.get_by_code(db_session, CODE_A)
    assert flow.owner_person_id == admin.id
    assert flow.owner_source == const.OWNER_SOURCE_FEISHU_ADMIN
    # A declaration in the Feishu console is not the person agreeing to
    # answer for the flow here.
    assert flow.owner_confirmed_at is None

    from app.domains.approval_registry.serializers import owner_state
    flow = await svc.get_flow(db_session, flow.id)
    assert owner_state(flow) == "unconfirmed"


@pytest.mark.asyncio
async def test_several_admins_still_produce_one_owner_and_the_full_list(
    db_session, monkeypatch,
):
    """User decision 2026-08-27: every flow Feishu names an admin for
    gets an owner, team or not. An empty owner column means nobody gets
    the half-year card and nobody is blocked at offboarding, and one
    named person who can answer 「不是我」 beats nobody to ask."""
    one = await _person(db_session, f"x1-{uuid.uuid4().hex[:6]}@test.com")
    two = await _person(db_session, f"x2-{uuid.uuid4().hex[:6]}@test.com")
    person = await _person(db_session, f"c-{uuid.uuid4().hex[:6]}@test.com")
    uid = f"uid-{person.work_email.split('@')[0]}"
    stub = StubClient(
        instances={uid: [_instance(CODE_A, "流程", "IT", instance_code="i-1")]},
        definitions={
            CODE_A: {
                "approval_name": "流程", "status": "ACTIVE",
                "approval_admin_ids": ["fs-2", "fs-1"],
            },
        },
        users={
            "fs-1": _profile("甲", one.work_email),
            "fs-2": _profile("乙", two.work_email),
        },
    )
    _install(monkeypatch, stub)
    summary = await discovery.run_sync(db_session, infer_owners=False)

    assert summary["owners_from_feishu_admin"] == 1
    flow = await svc.get_by_code(db_session, CODE_A)
    # ``fs-1`` despite Feishu listing ``fs-2`` first — the pick is
    # user_id-ordered precisely because Feishu's order is not a fact.
    assert flow.owner_person_id == one.id
    assert flow.owner_source == const.OWNER_SOURCE_FEISHU_ADMIN
    # The whole list still publishes: one of the team owns the row, the
    # page still answers 「找谁」 with everybody.
    assert {a["person_id"] for a in flow.feishu_admins} == {
        str(one.id), str(two.id),
    }


@pytest.mark.asyncio
async def test_the_owner_does_not_rotate_when_feishu_reorders_the_list(
    db_session, monkeypatch,
):
    """THE reason the pick cannot be ``admin_ids[0]``. Measured against
    the live tenant: five reads of 名片申请 (7 admins) returned four
    different orders and four different heads — Feishu rotates the
    array on every call. Index 0 would re-point the flow most nights,
    log a transfer each time, and send December's card to whoever won
    that morning's roll."""
    a = await _person(db_session, f"r1-{uuid.uuid4().hex[:6]}@test.com", name="甲")
    bb = await _person(db_session, f"r2-{uuid.uuid4().hex[:6]}@test.com", name="乙")
    cc = await _person(db_session, f"r3-{uuid.uuid4().hex[:6]}@test.com", name="丙")
    person = await _person(db_session, f"j-{uuid.uuid4().hex[:6]}@test.com")
    uid = f"uid-{person.work_email.split('@')[0]}"
    users = {
        "fs-a": _profile("甲", a.work_email),
        "fs-b": _profile("乙", bb.work_email),
        "fs-c": _profile("丙", cc.work_email),
    }
    definitions = {
        CODE_A: {
            "approval_name": "流程", "status": "ACTIVE",
            "approval_admin_ids": ["fs-b", "fs-c", "fs-a"],
        },
    }
    stub = StubClient(
        instances={uid: [_instance(CODE_A, "流程", "IT", instance_code="i-1")]},
        definitions=definitions, users=users,
    )
    _install(monkeypatch, stub)
    await discovery.run_sync(db_session, infer_owners=False)
    flow = await svc.get_by_code(db_session, CODE_A)
    first_owner = flow.owner_person_id
    assert first_owner == a.id

    # Feishu rotates the ring three more times. The owner must not move.
    for rotation in (
        ["fs-c", "fs-a", "fs-b"],
        ["fs-a", "fs-b", "fs-c"],
        ["fs-b", "fs-c", "fs-a"],
    ):
        definitions[CODE_A]["approval_admin_ids"] = rotation
        summary = await discovery.run_sync(
            db_session, discover=False, infer_owners=False,
        )
        assert summary["owners_from_feishu_admin"] == 0
        flow = await svc.get_by_code(db_session, CODE_A)
        assert flow.owner_person_id == first_owner

    events = await svc.list_events(db_session, flow.id)
    assert len([
        e for e in events
        if e.event_type == const.EVENT_OWNER_FROM_FEISHU_ADMIN
    ]) == 1
    # Membership never changed either, so no false 「管理员变更」 churn.
    assert len([
        e for e in events if e.event_type == const.EVENT_FEISHU_ADMINS_CHANGED
    ]) == 1


@pytest.mark.asyncio
async def test_a_newly_added_admin_does_not_take_the_flow_off_the_incumbent(
    db_session, monkeypatch,
):
    """The case the stable sort alone does NOT cover, and the reason the
    incumbent check exists: somebody is added as a 流程管理员 in Feishu
    and happens to sort ahead of the person who has been answering for
    the flow. Re-pointing on that would move accountability — and
    December's confirmation card — because of an unrelated console edit.
    """
    incumbent = await _person(
        db_session, f"t2-{uuid.uuid4().hex[:6]}@test.com", name="在任",
    )
    newcomer = await _person(
        db_session, f"t1-{uuid.uuid4().hex[:6]}@test.com", name="新来的",
    )
    person = await _person(db_session, f"m-{uuid.uuid4().hex[:6]}@test.com")
    uid = f"uid-{person.work_email.split('@')[0]}"
    users = {"fs-zz": _profile("在任", incumbent.work_email)}
    definitions = {
        CODE_A: {
            "approval_name": "流程", "status": "ACTIVE",
            "approval_admin_ids": ["fs-zz"],
        },
    }
    stub = StubClient(
        instances={uid: [_instance(CODE_A, "流程", "IT", instance_code="i-1")]},
        definitions=definitions, users=users,
    )
    _install(monkeypatch, stub)
    await discovery.run_sync(db_session, infer_owners=False)
    flow = await svc.get_by_code(db_session, CODE_A)
    assert flow.owner_person_id == incumbent.id

    # 「fs-aa」 sorts before 「fs-zz」. Without the incumbent check the
    # flow would silently change hands.
    users["fs-aa"] = _profile("新来的", newcomer.work_email)
    definitions[CODE_A]["approval_admin_ids"] = ["fs-aa", "fs-zz"]
    summary = await discovery.run_sync(
        db_session, discover=False, infer_owners=False,
    )

    assert summary["owners_from_feishu_admin"] == 0
    flow = await svc.get_by_code(db_session, CODE_A)
    assert flow.owner_person_id == incumbent.id
    # The newcomer IS published as an admin — they just did not take
    # the row over.
    assert {a["name"] for a in flow.feishu_admins} == {"在任", "新来的"}


@pytest.mark.asyncio
async def test_the_owner_moves_on_only_when_they_drop_off_the_list(
    db_session, monkeypatch,
):
    """Sticky, not frozen. An incumbent who is removed from the Feishu
    list is no longer an administrator of that flow, and keeping them
    would make the register argue with the console."""
    a = await _person(db_session, f"s1-{uuid.uuid4().hex[:6]}@test.com", name="甲")
    bb = await _person(db_session, f"s2-{uuid.uuid4().hex[:6]}@test.com", name="乙")
    person = await _person(db_session, f"k-{uuid.uuid4().hex[:6]}@test.com")
    uid = f"uid-{person.work_email.split('@')[0]}"
    users = {
        "fs-a": _profile("甲", a.work_email),
        "fs-b": _profile("乙", bb.work_email),
    }
    definitions = {
        CODE_A: {
            "approval_name": "流程", "status": "ACTIVE",
            "approval_admin_ids": ["fs-a", "fs-b"],
        },
    }
    stub = StubClient(
        instances={uid: [_instance(CODE_A, "流程", "IT", instance_code="i-1")]},
        definitions=definitions, users=users,
    )
    _install(monkeypatch, stub)
    await discovery.run_sync(db_session, infer_owners=False)
    flow = await svc.get_by_code(db_session, CODE_A)
    assert flow.owner_person_id == a.id

    definitions[CODE_A]["approval_admin_ids"] = ["fs-b"]
    summary = await discovery.run_sync(
        db_session, discover=False, infer_owners=False,
    )
    assert summary["owners_from_feishu_admin"] == 1
    flow = await svc.get_by_code(db_session, CODE_A)
    assert flow.owner_person_id == bb.id


@pytest.mark.asyncio
async def test_a_departed_admin_is_flagged_not_dropped_and_never_adopted(
    db_session, monkeypatch,
):
    gone = await _person(
        db_session, f"gone-{uuid.uuid4().hex[:6]}@test.com",
        name="已离职", status="left",
    )
    person = await _person(db_session, f"d-{uuid.uuid4().hex[:6]}@test.com")
    uid = f"uid-{person.work_email.split('@')[0]}"
    stub = StubClient(
        instances={uid: [_instance(CODE_A, "流程", "IT", instance_code="i-1")]},
        definitions={
            CODE_A: {
                "approval_name": "流程", "status": "ACTIVE",
                "approval_admin_ids": ["fs-gone"],
            },
        },
        users={"fs-gone": _profile("已离职", gone.work_email, resigned=True)},
    )
    _install(monkeypatch, stub)
    summary = await discovery.run_sync(db_session, infer_owners=False)

    assert summary["owners_from_feishu_admin"] == 0
    flow = await svc.get_by_code(db_session, CODE_A)
    assert flow.owner_person_id is None
    # Kept, and marked. 「管理员已经离职」 is the actionable answer;
    # dropping the row turns it back into a silent 「还没人认领」.
    assert [(a["name"], a["active"]) for a in flow.feishu_admins] == [
        ("已离职", False),
    ]


@pytest.mark.asyncio
async def test_an_admin_registered_under_the_sibling_domain_still_matches(
    db_session, monkeypatch,
):
    """Feishu accounts here live under pingcap.com OR pingcap.com.cn
    while ``work_email`` holds one of them. Exact-match-only lookup
    dropped 8 of 37 real admins, including two flows' only one."""
    admin = await _person(
        db_session, f"sib-{uuid.uuid4().hex[:6]}@pingcap.com", name="孟凡生",
    )
    feishu_address = admin.work_email.replace("@pingcap.com", "@pingcap.com.cn")
    person = await _person(db_session, f"e-{uuid.uuid4().hex[:6]}@test.com")
    uid = f"uid-{person.work_email.split('@')[0]}"
    stub = StubClient(
        instances={uid: [_instance(CODE_A, "流程", "IT", instance_code="i-1")]},
        definitions={
            CODE_A: {
                "approval_name": "流程", "status": "ACTIVE",
                "approval_admin_ids": ["fs-sib"],
            },
        },
        users={"fs-sib": _profile("孟凡生", feishu_address)},
    )
    _install(monkeypatch, stub)
    await discovery.run_sync(db_session, infer_owners=False)

    flow = await svc.get_by_code(db_session, CODE_A)
    assert flow.owner_person_id == admin.id
    assert flow.feishu_admins[0]["person_id"] == str(admin.id)


@pytest.mark.asyncio
async def test_an_admin_with_no_helix_person_is_still_published_by_name(
    db_session, monkeypatch,
):
    person = await _person(db_session, f"f-{uuid.uuid4().hex[:6]}@test.com")
    uid = f"uid-{person.work_email.split('@')[0]}"
    stub = StubClient(
        instances={uid: [_instance(CODE_A, "流程", "IT", instance_code="i-1")]},
        definitions={
            CODE_A: {
                "approval_name": "流程", "status": "ACTIVE",
                "approval_admin_ids": ["fs-outsider"],
            },
        },
        users={
            "fs-outsider": _profile("外部管理员", "outsider@elsewhere.example"),
        },
    )
    _install(monkeypatch, stub)
    await discovery.run_sync(db_session, infer_owners=False)

    flow = await svc.get_by_code(db_session, CODE_A)
    assert flow.feishu_admins == [{
        "user_id": "fs-outsider", "name": "外部管理员",
        "email": "outsider@elsewhere.example", "person_id": None,
        "active": True,
    }]
    # No person row means nobody to make accountable — the name is the
    # answer to 「找谁」 and nothing more.
    assert flow.owner_person_id is None


@pytest.mark.asyncio
async def test_task_inference_does_not_overwrite_a_declared_admin(
    db_session, monkeypatch,
):
    """Two machine sweeps that each overwrote the other would re-point
    the row on alternating nights. A declaration outranks a statistic."""
    admin = await _person(
        db_session, f"decl-{uuid.uuid4().hex[:6]}@test.com", name="声明的管理员",
    )
    busy = await _person(db_session, f"busy-{uuid.uuid4().hex[:6]}@test.com")
    uid = f"uid-{busy.work_email.split('@')[0]}"
    stub = StubClient(
        instances={uid: [_instance(CODE_A, "流程", "IT", instance_code="i-1")]},
        tasks={uid: [_task(CODE_A) for _ in range(9)]},
        definitions={
            CODE_A: {
                "approval_name": "流程", "status": "ACTIVE",
                "approval_admin_ids": ["fs-decl"],
            },
        },
        users={"fs-decl": _profile("声明的管理员", admin.work_email)},
    )
    _install(monkeypatch, stub)
    await discovery.run_sync(db_session)

    flow = await svc.get_by_code(db_session, CODE_A)
    assert flow.owner_person_id == admin.id
    assert flow.owner_source == const.OWNER_SOURCE_FEISHU_ADMIN


@pytest.mark.asyncio
async def test_a_declared_admin_replaces_a_weaker_inferred_guess(
    db_session, monkeypatch,
):
    admin = await _person(
        db_session, f"real-{uuid.uuid4().hex[:6]}@test.com", name="真管理员",
    )
    guessed = await _person(
        db_session, f"guess-{uuid.uuid4().hex[:6]}@test.com", name="被猜的人",
    )
    flow = await svc.create_flow(db_session, approval_code=CODE_A, name="流程")
    # Through ``load_person``: ``IdentityPerson.department`` is a
    # property over a relationship, and both writers read it.
    admin = await svc.load_person(db_session, admin.id)
    guessed = await svc.load_person(db_session, guessed.id)
    await svc.apply_inferred_owner(
        db_session, flow, person=guessed, evidence={"task_count": 9},
    )
    await db_session.flush()
    assert flow.owner_source == const.OWNER_SOURCE_INFERRED

    changed = await svc.apply_feishu_admin_owner(db_session, flow, person=admin)
    await db_session.flush()
    assert changed is True
    assert flow.owner_person_id == admin.id
    assert flow.owner_source == const.OWNER_SOURCE_FEISHU_ADMIN


@pytest.mark.asyncio
async def test_a_declared_admin_never_overwrites_a_human_decision(
    db_session, monkeypatch,
):
    it_pick = await _person(
        db_session, f"it-{uuid.uuid4().hex[:6]}@test.com", name="IT 指派的",
    )
    admin = await _person(
        db_session, f"fs-{uuid.uuid4().hex[:6]}@test.com", name="飞书管理员",
    )
    flow = await svc.create_flow(
        db_session, approval_code=CODE_B, name="流程",
        owner_person_id=it_pick.id, announce=None,
    )
    await db_session.flush()
    assert flow.owner_source == const.OWNER_SOURCE_MANUAL

    changed = await svc.apply_feishu_admin_owner(db_session, flow, person=admin)
    assert changed is False
    assert flow.owner_person_id == it_pick.id


@pytest.mark.asyncio
async def test_a_failed_definition_read_keeps_yesterdays_admins(
    db_session, monkeypatch,
):
    """Blanking the list on a transient 502 would put 「还没人认领」 back
    on the all-staff page every time Feishu rate-limited one call."""
    admin = await _person(
        db_session, f"keep-{uuid.uuid4().hex[:6]}@test.com", name="常驻管理员",
    )
    person = await _person(db_session, f"g-{uuid.uuid4().hex[:6]}@test.com")
    uid = f"uid-{person.work_email.split('@')[0]}"
    definitions = {
        CODE_A: {
            "approval_name": "流程", "status": "ACTIVE",
            "approval_admin_ids": ["fs-keep"],
        },
    }
    stub = StubClient(
        instances={uid: [_instance(CODE_A, "流程", "IT", instance_code="i-1")]},
        definitions=definitions,
        users={"fs-keep": _profile("常驻管理员", admin.work_email)},
    )
    _install(monkeypatch, stub)
    await discovery.run_sync(db_session, infer_owners=False)

    definitions.clear()  # every GET now raises
    await discovery.run_sync(db_session, discover=False, infer_owners=False)

    flow = await svc.get_by_code(db_session, CODE_A)
    assert flow.last_sync_error is not None
    assert [a["name"] for a in flow.feishu_admins] == ["常驻管理员"]


@pytest.mark.asyncio
async def test_a_broken_contact_lookup_does_not_delete_a_live_admin(
    db_session, monkeypatch,
):
    """Caught by measuring production, not by reasoning: one admin id
    resolved on one pass and errored on the next, which flipped a flow
    from 「有唯一管理员」 to 「一个人都没有」 between two runs minutes
    apart. Writing the survivors of a half-failed lookup deletes a live
    administrator off the all-staff page and logs a false 「管理员变更」."""
    a = await _person(db_session, f"la-{uuid.uuid4().hex[:6]}@test.com", name="甲")
    b = await _person(db_session, f"lb-{uuid.uuid4().hex[:6]}@test.com", name="乙")
    person = await _person(db_session, f"h-{uuid.uuid4().hex[:6]}@test.com")
    uid = f"uid-{person.work_email.split('@')[0]}"
    users = {
        "fs-a": _profile("甲", a.work_email),
        "fs-b": _profile("乙", b.work_email),
    }
    definitions = {
        CODE_A: {
            "approval_name": "流程", "status": "ACTIVE",
            "approval_admin_ids": ["fs-a", "fs-b"],
        },
    }
    stub = StubClient(
        instances={uid: [_instance(CODE_A, "流程", "IT", instance_code="i-1")]},
        definitions=definitions, users=users,
    )
    _install(monkeypatch, stub)
    await discovery.run_sync(db_session, infer_owners=False)
    flow = await svc.get_by_code(db_session, CODE_A)
    assert {x["name"] for x in flow.feishu_admins} == {"甲", "乙"}

    # Next night: 乙's contact lookup times out. 乙 has not left.
    stub._lookup_failures = {"fs-b"}
    summary = await discovery.run_sync(
        db_session, discover=False, infer_owners=False,
    )

    assert summary["admins_kept_stale"] == 1
    assert summary["admin_ids_lookup_failed"] == 1
    # NOT counted as gone — that number drives 「管理员已离职」 triage.
    assert summary["admin_ids_unresolved"] == 0
    flow = await svc.get_by_code(db_session, CODE_A)
    assert {x["name"] for x in flow.feishu_admins} == {"甲", "乙"}
    events = await svc.list_events(db_session, flow.id)
    assert len([
        e for e in events if e.event_type == const.EVENT_FEISHU_ADMINS_CHANGED
    ]) == 1


@pytest.mark.asyncio
async def test_an_id_feishu_itself_rejects_still_counts_as_gone(
    db_session, monkeypatch,
):
    """The other half of the distinction: Feishu answering 「no such
    user」 IS a departed administrator, and must not be hidden behind
    the transient-failure guard."""
    person = await _person(db_session, f"i-{uuid.uuid4().hex[:6]}@test.com")
    uid = f"uid-{person.work_email.split('@')[0]}"
    stub = StubClient(
        instances={uid: [_instance(CODE_A, "流程", "IT", instance_code="i-1")]},
        definitions={
            CODE_A: {
                "approval_name": "流程", "status": "ACTIVE",
                "approval_admin_ids": ["fs-deleted"],
            },
        },
        users={},  # Feishu rejects the id outright
    )
    _install(monkeypatch, stub)
    summary = await discovery.run_sync(db_session, infer_owners=False)

    assert summary["admin_ids_unresolved"] == 1
    assert summary["admins_kept_stale"] == 0
    flow = await svc.get_by_code(db_session, CODE_A)
    assert flow.feishu_admins == []
    # Asked, and Feishu named nobody reachable — different from never
    # asked, and the timestamp is how the two stay distinguishable.
    assert flow.feishu_admins_synced_at is not None


@pytest.mark.asyncio
async def test_the_client_only_sends_the_param_when_asked(monkeypatch):
    """The stub proves the sweep passes the flag; only the real client
    can prove the flag becomes a query param."""
    from app.approval.feishu_client import FeishuApprovalClient

    client = FeishuApprovalClient(app_id="cli_x", app_secret="secret")
    paths: list[str] = []

    async def _fake_request(method, path, **kwargs):
        paths.append(path)
        return {"approval_name": "流程", "status": "ACTIVE"}

    monkeypatch.setattr(client, "_request", _fake_request)

    await client.get_approval_definition(CODE_A, subscribe=False)
    await client.get_approval_definition(
        CODE_A, subscribe=False, with_admin_id=True,
    )

    assert "with_admin_id" not in paths[0]
    assert "with_admin_id=true" in paths[1]
    assert "user_id_type=user_id" in paths[1]


@pytest.mark.asyncio
async def test_a_second_run_is_idempotent(db_session, monkeypatch):
    person = await _person(db_session, f"m-{uuid.uuid4().hex[:6]}@test.com")
    uid = f"uid-{person.work_email.split('@')[0]}"
    stub = StubClient(
        instances={uid: [_instance(CODE_A, "流程", "IT", instance_code="i-1")]},
        tasks={uid: [_task(CODE_A) for _ in range(4)]},
        definitions={CODE_A: {"approval_name": "流程", "status": "ACTIVE"}},
    )
    _install(monkeypatch, stub)
    first = await discovery.run_sync(db_session)
    second = await discovery.run_sync(db_session)

    assert first["flows_created"] == 1
    assert second["flows_created"] == 0
    # Re-inferring the same person is not a second ownership event.
    assert second["owners_inferred"] == 0
    flow = await svc.get_by_code(db_session, CODE_A)
    events = await svc.list_events(db_session, flow.id)
    assert len([e for e in events if e.event_type == const.EVENT_OWNER_INFERRED]) == 1


@pytest.mark.asyncio
async def test_window_is_clamped_to_the_feishu_ceiling(db_session, monkeypatch):
    """Feishu rejects an instance window wider than 30 days."""
    person = await _person(db_session, f"n-{uuid.uuid4().hex[:6]}@test.com")
    uid = f"uid-{person.work_email.split('@')[0]}"
    seen: dict[str, str] = {}

    class WindowStub(StubClient):
        async def query_instances_for_user(
            self, user_id, *, page_size=200, max_pages=30,
            start_time_ms=None, end_time_ms=None,
        ):
            seen["from"] = start_time_ms
            seen["to"] = end_time_ms
            return [], False

    stub = WindowStub(instances={uid: []})
    _install(monkeypatch, stub)
    await discovery.run_sync(
        db_session, window_days=90, sync_definitions=False, infer_owners=False,
    )

    span_days = (int(seen["to"]) - int(seen["from"])) / 86_400_000
    assert span_days <= discovery.MAX_WINDOW_DAYS