"""Feishu (Lark) Contact user runner — outbound user upsert + freeze. Verified against the live PingCAP tenant 2026-05-28: * ``employee_no`` — Helix HRIS ``employee_id`` (``P0001`` style) maps 1:1 onto Feishu ``employee_no`` and is stamped into the create body, but it CANNOT be a lookup key: Feishu has no "find user by employee_no" API (``user_id_type`` accepts only open_id / union_id / user_id; employee_no / employee_id both 99992402, and ``batch_get_id`` resolves by emails / mobiles only). Lookup is therefore **email first, then mobile** — the same mobile+email association the legacy Authing 飞书 sync used — with a **last-resort employee_no match** when both miss: since Feishu can't be queried by employee_no, the runner enumerates the directory ONCE per sweep (lazily, only when needed) into a local ``{employee_no → user_id}`` index (:meth:`_employee_no_index`) and matches on the HRIS natural key. This closes the residual duplicate-account gap for a person with no mobile AND a mismatched Feishu email (email+mobile both miss → POST would create a duplicate). A match routes them through PATCH (self-correcting the email); a genuine miss still POST-creates. * ``department_ids`` on a Feishu user is a list that already carries **non-HRIS depts** populated by other systems / admin (Tokyo Office, SG Office, SV Office, MY Office — geographic-location depts that don't exist in HRIS). Helix MUST NOT overwrite these. The runner GETs the existing list before PATCH, replaces only the slice it manages (depts in :attr:`_helix_dept_to_external` — i.e. HRIS- derived), and keeps everything else verbatim. * ``custom_attrs`` on Feishu's write API expects a LIST of ``{type, id, value: {text}}`` objects, not a dict. The schema ``id`` is opaque (``C-7418413146041008130`` etc.), keyed in Feishu's catalog by ``i18n_name``. The runner translates payload's ``custom_attrs.`` strings (e.g. ``pinyin``, ``company_entity``) into the Feishu wire format via the catalog primed at runner-build time. Unknown names are dropped with a log warning so admins can fix the Feishu side (build the missing schema) without the user-push 400ing every sync. * ``leader_user_id`` is set with ``?user_id_type=user_id`` in the query string so Feishu interprets the value we send as user_id format (vs the default open_id). Translation miss drops the slot. * ``supports_delete=False`` — hard delete lives in ``feishu_software._user_delete`` (offboarding flow). The sync layer only freezes via ``PATCH /contact/v3/users/{id}`` with ``status.is_frozen=true``; ``230002`` (already_frozen) / ``99991663`` / ``230003`` (user-gone) collapse to idempotent success. Coexistence with whatever currently writes Feishu (Authing, manual admin work, etc.) is non-destructive by construction: * lookup by email → mobile → POST only on genuine miss * department_ids slice-replace (NOT full overwrite) * custom_attrs name-filtered against the live catalog * no hard delete, no clear-all-fields, no group reassignment """ from __future__ import annotations import asyncio import re from typing import Any from urllib.parse import quote import httpx from app.approval.feishu_client import ( FeishuApprovalClient, is_feishu_permission_failure, is_systemic_feishu_failure, ) from app.approval.feishu_client import ( parse_feishu_error_code as _parse_feishu_code, ) from app.core.exceptions import AppException from app.core.geo import normalize_country_code from app.core.phone import DIAL_CODE_ISO2 from app.core.safe_logging import get_safe_logger from app.domains.identity_sync.runners.base import PersonOutcome, SyncRunner logger = get_safe_logger(__name__) # Feishu user-API codes that mean "user not in the directory" — treat # as idempotent for freeze (goal already met) and as "fall through to # create" for upsert. NOTE: 230003 is INTENTIONALLY only here, not in # ``_DEPT_NOT_FOUND_CODES`` (Feishu reuses 230003 for both "user not # found" and "department_id not found" depending on which lookup # triggered it; we disambiguate by checking the response body via # :func:`_response_mentions_department` rather than by code alone). _USER_ABSENT_CODES: frozenset[int] = frozenset({ 99991663, # generic resource-not-found 230003, # user not found (alt path) — also overloaded for dept; # caller disambiguates via response message inspection. }) # Already-frozen / already-target-state codes (freeze idempotency). _USER_ALREADY_FROZEN_CODES: frozenset[int] = frozenset({230002}) # Resigned / already-off codes. Feishu ``42006`` ("user has resigned") is # returned by ``PATCH is_frozen=true`` when the account is ALREADY in the # resigned (离职) state — a disposition MORE terminal than frozen (the seat is # already released and access already cut). These are NO LONGER trusted as # blind success: ``deactivate`` re-reads the directory ``status`` and only # claims ``deactivated`` when access is PROVABLY cut (is_frozen / is_resigned / # is_exited). A genuinely-resigned account verifies (is_resigned=true) and # still short-circuits the daily ``account_backstop`` false-alarm this set was # added for; an account that returns 42006 yet is still active surfaces as a # real failure instead of a false green (observed 2026-07-24, P01108). _USER_RESIGNED_CODES: frozenset[int] = frozenset({42006}) # Department-not-found codes returned by POST/PATCH /contact/v3/users # when one of the supplied ``department_ids`` is not a real od-XXX on # the Feishu side. Surface so we can downgrade (re-PATCH without dept). # NOTE: 230003 is shared between this set and ``_USER_ABSENT_CODES``; # callers MUST disambiguate by inspecting the Feishu response message # (it mentions "department" vs "user" in the localised text) — see # :func:`_response_mentions_department`. The PATCH path checks dept- # scope FIRST so an ambiguous 230003 in a request that carried # department_ids retries without dept_ids before treating the user as # absent. _DEPT_NOT_FOUND_CODES: frozenset[int] = frozenset({ 230009, # department_id not found (some tenants) 230018, # department_id invalid }) # Department-scope codes: the dept od IS a real, readable department, but # the app is not permitted to ASSIGN this user into it (it's outside the # app's 通讯录管理范围, or the tenant otherwise refuses the assignment). # Distinct from _DEPT_NOT_FOUND_CODES (dept doesn't exist / stale adoption, # which self-heals when dept sync re-adopts): a "not assigned" rejection # does NOT self-heal by retrying the same resolved dept, so the runner # degrades to SUCCESS (other fields land, person stays usable) with a # per-slot flag for IT — never churns the whole person as failed. This is # the class that froze P10003's every field behind one un-appliable dept # before the ledger even showed which dept (see #1954 observability). _DEPT_NOT_ASSIGNED_CODES: frozenset[int] = frozenset({ 41041, # "department id is not assigned" }) # Feishu's root-department sentinel. A user CREATE requires a non-empty # ``department_ids``; when no real department resolves (future-hire 24h # window before the effective-dated HRIS dept lands, or a 41041 dept-scope # rejection) we fall the user into root ``"0"`` so the account is created # SOMEWHERE. It is only ever a transient placeholder: Feishu lists a user # under EVERY dept in their ``department_ids``, so a lingering ``"0"`` pins # them at the org root even after their real dept resolves (their card # shows the real dept, the org tree still shows them at the company root — # P10011 Eddie Jiang, 2026-07-13). The merge (:func:`_merge_department_ids`) # therefore strips it the moment any real dept is present. _FEISHU_ROOT_DEPT_ID: str = "0" # Mobile-invalid codes returned by POST/PATCH /contact/v3/users when the # ``mobile`` slot fails Feishu's per-country validation (e.g. an 8-digit SG # number sent with a +86 code because the home country was used). Defense in # depth ON TOP OF the phone-country-code fix: if a mobile still can't be # validated (bad HRIS data / a country we can't resolve), retry the write # WITHOUT the mobile slot so the rest of the person's fields still land and # Feishu keeps its existing number — never fail the whole person over one # optional field. Verified live 2026-07-07 (41004 "mobile is invalid"). _MOBILE_INVALID_CODES: frozenset[int] = frozenset({41004}) # Residence countries whose Feishu Multi-Geo data residency is the CHINA # region. Everything else lands in ``jp``. # # HK / MO / TW are here deliberately (user decision 2026-08-31, recorded in # DECISIONS.md): SuccessFactors itself labels the work country "Hong Kong, # China (HKG)", and the residency axis follows that. Before this the rule # was a bare ``residence == "CN"``, so a Hong Kong hire silently landed in # Japan. Changing the mapping does NOT move anyone already created — geo is # write-once at creation (a Directory PATCH ignores ``geo_name`` outright), # so existing accounts need the security_and_compliance user-migration API. _CN_GEO_RESIDENCE_COUNTRIES: frozenset[str] = frozenset({ "CN", "HK", "MO", "TW", }) # Anchored regex matching dept-shaped tokens in a Feishu error body. # Word boundary on the EN side prevents a user display-name containing # "department" (rare but possible) from being misclassified. # The zh-CN side anchors on `部门` followed by a usage phrase Feishu # uses for dept errors (不存在 = "doesn't exist", 无效 = "invalid", # id) — bare `部门` in a name field doesn't trigger. _DEPT_ERROR_RE = re.compile( r"\bdepartment(?:_?id)?\b|部门(?:不存在|无效|id)", re.IGNORECASE, ) def _scope_blocked_outcome( exc: AppException, *, surface: str, request_body: dict[str, Any] | None = None, ) -> PersonOutcome | None: """``blocked`` outcome for "the app lacks this scope", else ``None``. Why ``blocked`` and not ``failed`` (2026-08-31, P10025 Augustine Thia): a missing scope is a CONFIG problem — an admin must grant it at ``open.feishu.cn/app//auth`` **and publish a new app version**. Retrying cannot heal it, so ``failed`` was actively misleading: it reddened the sweep once an hour for 9 hours and read as "transient". ``blocked`` is this codebase's word for exactly this state — run stays green, no retry churn, and ``maybe_notify_blocked_backlog`` pages IT once with the reason instead of every hour with a stack of red rows. The new-hire path is unaffected in loudness: the onboarding coordinator notifies on ``status in {"failed", "blocked"}`` alike, so IT still gets the 「新员工飞书账号创建失败」 card on the first attempt. Deliberately NOT routed through ``is_systemic_feishu_failure`` (which would halt the whole sweep): Feishu grants scopes PER API SURFACE, and on 2026-08-19 the Directory *create* scope was missing while the Contact *update* scope worked fine for all ~500 people. Halting there would have broken updates too. Feishu's 99991672 message always embeds the exact scopes and a self-service grant link, so it is passed through verbatim — that text IS the remediation. """ if not is_feishu_permission_failure(exc): return None logger.error( "feishu_user: %s refused — the Feishu app is missing a required " "scope. Grant it AND publish a new app version; nothing retries " "into a fix. Vendor said: %s", surface, exc.message, ) return PersonOutcome( status="blocked", error=f"{surface} → {exc.code}: {exc.message}", detail={"blocked_reason": "vendor_scope_missing"}, request_body=request_body, ) def _response_mentions_department(exc: AppException) -> bool: """True when the Feishu error message clearly references a department (vs a user). Used to disambiguate code 230003 between "user not found" and "department_id not found" on the user PATCH endpoint, where both interpretations share the same numeric code. """ return bool(_DEPT_ERROR_RE.search(exc.message or "")) # ── Mobile → E.164 (Feishu wants the country calling code) ──────────── # **ISO-2** country code → international calling code (digits, no "+"). # The runner resolves the code from the rendered ``country`` payload # field, which the feishu preset value-maps to ISO-2 (CHN→CN, JPN→JP — # Feishu's own country field is ISO-2). So this map MUST be keyed by # ISO-2 to match, NOT ISO-3. The mapping engine renders {mobile_phone} # as the raw national number; Feishu's ``mobile`` slot wants # ``+``. Legacy Authing did ``(phoneCountryCode||'+86')+phone``; # an empty/unknown country falls back to China (86, same default). A # country SET but not in this table drops the mobile (see # _format_mobile_e164) rather than guessing a wrong code. # ISO-2 → calling-code reference data is shared with the Google recovery- # phone path; this runner keeps only its own default-+86 fallback below. _COUNTRY_DIALING_CODE = DIAL_CODE_ISO2 _DEFAULT_DIALING_CODE = "86" def _format_mobile_e164( raw: str, country_iso2: str | None, *, raw_country_iso3: str | None = None, phone_country_iso2: str | None = None, ) -> str: """National mobile number → Feishu's ``+`` shape. ``phone_country_iso2`` is the mobile's OWN country ISO-2 (SF ``PerPhone.countryCode`` → externalCode, e.g. ``SG``), threaded via the synthetic ``mobile_country_iso2`` payload field. It takes PRECEDENCE over every home-country signal below: an international-assignment employee (home CN) with a host-country phone (+65 SG) must format as ``+65…``, not ``+86…`` (which is an invalid number the vendor rejects, ``41004``). When set-but-unmapped (a phone-country we carry no dialing code for) the slot is dropped rather than guessed — same stance as the home-country path. ``country_iso2`` is the rendered ``country`` payload field, which the feishu preset value-maps to ISO-2 (CN/JP/US/…) — the same code Feishu stores. ``_COUNTRY_DIALING_CODE`` is keyed by ISO-2 to match. Used only when ``phone_country_iso2`` is absent. ``raw_country_iso3`` is the person's PRE-value-map HRIS country (ISO-3, e.g. ``BRA``), threaded via the synthetic ``mobile_country_iso3`` payload field. It exists solely to break a tie the value-mapped ISO-2 can't: the ``country`` rule renders EMPTY both when the country is genuinely unset AND when it's set-but-unmapped (value-map miss, no passthrough). Without the raw signal those two collapse and a set-but-unmapped country would fall through to the ``+86`` default — silently re-introducing exactly the wrong-calling-code guess #1547 removed. See the resolution rules below. * already ``+``-prefixed → returned normalised (digits only after the ``+``), so a second sweep never double-prefixes — idempotent. * ``00`` international prefix → normalised to ``+``. * otherwise: strip the national trunk leading ``0`` (so JP/UK/AU mobiles render correctly) and prepend ``+`` resolved from the ISO-2 country. Calling-code resolution (the ``+`` / ``00`` shapes above already carry their own code and skip this): * ISO-2 country SET and in the dialing-code table → use it. * ISO-2 country SET but NOT in the table → **drop the slot** (return ``""``). We refuse to guess ``+86`` for a known-foreign person. * ISO-2 EMPTY but the raw HRIS country IS set (value-map miss) → **drop the slot**. The person has a known country we simply don't carry a dialing code for; guessing ``+86`` would be wrong. * ISO-2 EMPTY and raw country also empty → default China ``+86`` (CN-majority tenant, matches the legacy Authing ``phoneCountryCode || '+86'`` default). Empty / digit-less input → ``""`` so the caller drops the slot and Feishu keeps its existing value (never NULL-out). """ if not raw: return "" s = raw.strip() has_plus = s.startswith("+") # HRIS free-text mobile fields carry spaces / dashes / parens / dots. digits = re.sub(r"\D", "", s) # Empty OR all-zeros (data error like "000") → drop the slot. The # strip("0") guard also stops the ``00`` international-prefix branch # below from turning "000" into a bogus "+0". if not digits.strip("0"): return "" if has_plus: return "+" + digits if digits.startswith("00"): return "+" + digits[2:] # PRECEDENCE: the phone's OWN country (SF PerPhone.countryCode → ISO-2). # Authoritative — a host-country phone on an international assignee must use # its own dialing code (SG → +65), never the home country's (+86). When # present-but-unmapped we drop the slot rather than fall back to the home # country, since guessing the home code for a known-foreign phone is the # exact wrong-code bug this field exists to fix. phone_cc = (phone_country_iso2 or "").strip().upper() if phone_cc: if phone_cc not in _COUNTRY_DIALING_CODE: return "" cc = _COUNTRY_DIALING_CODE[phone_cc] national = digits.lstrip("0") return ("+" + cc + national) if national else "" country = (country_iso2 or "").strip().upper() used_default_cc = False if country: if country not in _COUNTRY_DIALING_CODE: # Known-foreign country with no calling code mapped — don't guess # +86. Drop so Feishu keeps its existing value. return "" cc = _COUNTRY_DIALING_CODE[country] elif (raw_country_iso3 or "").strip(): # Country IS set in HRIS but the ISO-2 value-map missed it — same # "known-foreign, no dialing code" case as above. Drop; never guess # +86 for a person we know is not (necessarily) in China. return "" else: # Genuinely-empty country → CN-majority default. cc = _DEFAULT_DIALING_CODE used_default_cc = True national = digits.lstrip("0") if not national: return "" # When we GUESSED the +86 default (country genuinely unknown), the number # must actually look like a Chinese mobile (11 digits, leading 1) or the # guess is provably wrong: a foreign national number on a hire whose # country wasn't synced yet (e.g. a TW/MY joiner, "0938769558" → # "+86938769558") is a number Feishu REJECTS with 99992402 "field # validation failed", and — because ``mobile`` rides in the same create # body as everything else — that one bad field FAILS THE ENTIRE ACCOUNT # CREATE (2026-07-13, P10011 Eddie Jiang: no Feishu account minted at all). # Drop the slot so the account is created WITHOUT a phone (correctable # once the country/phone syncs) rather than not created at all. Explicit # phone-country / home-country paths above are trusted and never # second-guessed here — this guard fires only on the blind +86 default. if used_default_cc and not (len(national) == 11 and national.startswith("1")): return "" return "+" + cc + national def _normalise_attr_name(name: str | None) -> str: """Normalise a custom_attr name for catalog lookup. Feishu schemas carry an opaque ``id`` (``C-XXX``) plus an ``i18n_name`` list of locale-tagged strings. Mapping rules reference the schema by a normalised lower-snake name (``pinyin``, ``company_entity``, ``location``, ``job_function``). The runner's catalog index maps every locale value of every schema to its ``id`` under this normalisation so PingCAP's bilingual "Company Entity"/"公司"/"会社" all resolve to the same Feishu schema id. """ if not name: return "" # Replace any run of non-alphanumeric with single underscore, then # lower-case + strip. cleaned = re.sub(r"[^A-Za-z0-9]+", "_", str(name)).strip("_").lower() return cleaned def _normalise_office_name(name: str | None) -> str: """Normalise an office-dept name for symmetric lookup. Mirrors ``FeishuDepartmentRunner._normalise_name`` (whitespace collapse + lower-case + strip). Used on BOTH index-build and lookup so admin typos / case variance ("Tokyo Office" with two spaces, "tokyo office" lowercase) never silently miss. """ if not name: return "" return " ".join(name.split()).strip().lower() def _coerce_department_ids(payload_value: Any) -> list[str]: """Normalise the rendered ``department_ids`` payload slot to a list[str] of HRIS dept_codes. Mapping engine renders strings; admins typically map ``{department_code}`` → ``department_ids`` and we end up with one HRIS code. Accept: * single code string ("1809808") * comma-separated string ("1809808,4006056") * actual list (future mapping extension) """ if payload_value is None or payload_value == "": return [] if isinstance(payload_value, list): return [str(x).strip() for x in payload_value if x and str(x).strip()] if isinstance(payload_value, str): parts = [p.strip() for p in payload_value.split(",") if p.strip()] return parts return [str(payload_value).strip()] class FeishuContactRunner(SyncRunner): """Push HRIS-of-truth users into Feishu's directory. Build-time state plumbed in by the orchestrator (``service._build_runner``): * :attr:`_helix_dept_to_external` — HRIS ``dept_code → od-XXX`` for every dept Helix has adopted (from ``identity_sync_department_state``). Authoritative source for translating user mapping payload's ``department_ids`` (HRIS codes) into Feishu's open_department_id list. Also identifies which dept slots on a Feishu user are Helix-managed (so the merge step doesn't overwrite Authing-built geographic depts). * :attr:`_custom_attr_catalog` — ``normalised_name → {id, type}`` for every Feishu custom_attr schema configured on the tenant. Primed by :meth:`prime_custom_attrs`. Unknown names get dropped. """ supports_deactivate = True supports_delete = False def __init__( self, *, client: FeishuApprovalClient, helix_dept_to_external: dict[str, str] | None = None, office_name_to_external: dict[str, str] | None = None, custom_attr_catalog: dict[str, dict[str, Any]] | None = None, reopen_frozen: bool = False, skip_provision_emails: frozenset[str] | None = None, label: str = "Feishu Contact (user)", ) -> None: self._client = client # Defensive copy. Empty dict (not None) is the "explicitly none # adopted" state — the merge step treats every existing dept on # the Feishu user as not-Helix-managed and leaves it alone, so # we never overwrite anyone's depts when Helix's adoption table # is empty. self._helix_dept_to_external: dict[str, str] = dict( helix_dept_to_external or {}, ) # SECOND dept dimension — Feishu office-dept name → od-XXX. # Populated by the orchestrator at runner-build time by scanning # the target's mapping rules for the ``office_dept_name`` rule # and looking each value_map VALUE up in the live Feishu # catalog (post-Phase-A). Empty dict = no office sync # configured for this target. Keys are normalised through the # same ``_normalise_office_name`` helper used at lookup time, # so admin typos like "Tokyo Office" (extra spaces) or # "tokyo office" (lowercase) match anyway. self._office_name_to_external: dict[str, str] = { _normalise_office_name(k): v for k, v in (office_name_to_external or {}).items() if k and v } # "Helix-managed universe" for the dept_ids merge — union of # the HRIS BU slice AND the office slice. A Feishu dept that # matches either gets replaced wholesale by the latest Helix # render; everything else (e.g. Authing-only / admin-manual # depts) is preserved verbatim. self._helix_external_set: frozenset[str] = ( frozenset(self._helix_dept_to_external.values()) | frozenset(self._office_name_to_external.values()) ) self._custom_attr_catalog: dict[str, dict[str, Any]] | None = ( dict(custom_attr_catalog) if custom_attr_catalog is not None else None ) self._label = label # ── Provisioning policy (Authing-parity: create-missing + # reopen-frozen, both admin-gated). ────────────────────────── # ``reopen_frozen`` (target runner_options, default off): when # True, an in-scope, HRIS-active person who is frozen on Feishu # is reopened (打开) by asserting ``status.is_frozen=false`` on # the PATCH. Default off is the safe stance — some frozen # accounts are intentionally suspended (security / leaver # leftover) and must NOT be blindly reopened. # ``skip_provision_emails``: the admin skip-list, resolved from # ``skip_provision_person_ids`` to lower-cased work_emails by the # service. A listed person is NEVER created (missing → skipped) # and NEVER reopened; an existing active record is still updated. self._reopen_frozen: bool = bool(reopen_frozen) self._skip_provision_emails: frozenset[str] = frozenset( e.lower() for e in (skip_provision_emails or ()) if e ) # Lazy, once-per-sweep {employee_no(upper) → user_id} index, built # from a full directory walk the FIRST time a person's email + mobile # lookups both miss AND they carry an employee_no (the only path that # would otherwise POST-create a possible duplicate). Cached on the # runner for the rest of the sweep; ``None`` = not yet built, ``{}`` = # built but empty. Steady-state sweeps never build it (everyone is # hash-skipped or email/mobile-resolved). See :meth:`_employee_no_index`. self._emp_no_index: dict[str, str] | None = None # Once-per-sweep result of :meth:`_directory_write_scope_missing` # (``None`` = not probed yet). See that method for why it exists. self._directory_scope_error: str | None = None self._directory_scope_probed: bool = False # Cached build failure (see :meth:`_employee_no_index`): once the # directory walk fails this sweep, re-raise WITHOUT re-walking so one # failure can't fan out into an O(N) directory-call storm. self._emp_no_index_error: Exception | None = None # ── Per-person write observability (PR1). ──────────────────── # The FINAL wire body actually sent to Feishu for the current # person, and Feishu's response/error — captured by # :meth:`_write` at the real POST/PATCH call sites (NOT the # lookups) so the ledger records exactly what we sent and what # came back. Reset at the top of every :meth:`upsert` so a # person who never hits the vendor (hash-skip upstream) doesn't # inherit the previous person's wire. The orchestrator reads # them off the runner right after ``upsert`` returns. self._last_write_request: dict[str, Any] | None = None self._last_write_response: dict[str, Any] | None = None async def _write( self, method: str, path: str, *, body: dict[str, Any], ) -> dict[str, Any]: """Single choke-point for every user CREATE/PATCH wire call. Records the exact ``body`` sent and Feishu's response (or the parsed error) onto the runner so the orchestrator can persist them verbatim on the per-person ledger row — the observability that was missing when a 41041 froze a record and the ledger only showed the pre-render payload with a null response. A later write in the same ``upsert`` (e.g. a slot-dropped retry) overwrites these, so the ledger reflects the FINAL attempt. Re-raises on failure — control flow is unchanged. """ self._last_write_request = body try: resp = await self._client.request(method, path, json_body=body) self._last_write_response = { "ok": True, "response": resp if isinstance(resp, dict) else {"raw": resp}, } return resp except AppException as e: self._last_write_response = { "ok": False, "code": e.code, "feishu_code": _parse_feishu_code(e), "message": e.message, } raise async def prime_custom_attrs(self) -> None: """Populate :attr:`_custom_attr_catalog` from Feishu. Calls ``GET /contact/v3/custom_attrs`` and builds an index mapping every locale value of every schema's ``i18n_name`` (and the schema's ``key`` field, if non-null) to ``{id, type}``. Reads only — safe to call on production. Failure-tolerant: any exception sets the catalog to None (not an empty dict) so the runner falls back to "send everything" — per-user 400s on unknown schemas then surface as the diagnostic signal to the admin. Better than silently dropping every custom_attr the admin mapped. """ if self._custom_attr_catalog is not None: return try: data = await self._client.request( "GET", "/contact/v3/custom_attrs?page_size=50", ) except Exception as e: # noqa: BLE001 logger.warning( "feishu_user: prime_custom_attrs failed: %s. Falling back " "to send-everything (per-user 400s on unknown attrs).", e, ) # Leave attr at None — "send everything" mode. return catalog: dict[str, dict[str, Any]] = {} items = (data or {}).get("items") or [] for it in items: if not isinstance(it, dict): continue schema_id = it.get("id") if not isinstance(schema_id, str) or not schema_id: continue attr_type = it.get("type") or "TEXT" entry = {"id": schema_id, "type": attr_type} # Index every locale value of i18n_name + the bare ``key`` # field (if present, but it's NULL on PingCAP today). raw_key = it.get("key") if isinstance(raw_key, str) and raw_key: catalog[_normalise_attr_name(raw_key)] = entry i18n = it.get("i18n_name") if isinstance(i18n, list): for loc in i18n: if not isinstance(loc, dict): continue val = loc.get("value") if isinstance(val, str) and val: norm = _normalise_attr_name(val) if norm and norm not in catalog: catalog[norm] = entry self._custom_attr_catalog = catalog logger.info( "feishu_user: custom_attrs catalog primed: %d schema lookups", len(catalog), ) # ── Helpers ────────────────────────────────────────────────────── async def _resolve_leader_user_id(self, email: str | None) -> str | None: """Email → Feishu user_id, narrowly. Only the documented "leader doesn't yet have a Feishu account" case (``FEISHU_USER_NOT_FOUND``) is swallowed as None. Other AppException codes (rate-limit, auth-fail, 5xx) propagate to the caller so the sweep can short-circuit on systemic issues — otherwise a 30-min Feishu rate-limit silently drops the leader slot for every user, and hash-skip on the next sweep prevents re-push, leaving the org chart stale on Feishu indefinitely. """ if not email: return None try: return await self._client.resolve_user_id_by_email(email) except AppException as e: if e.code == "FEISHU_USER_NOT_FOUND": return None # Re-raise — caller (upsert) decides whether to fail the # whole user-push or just drop the leader slot per its # own policy. Most propagated codes are transient/global # (rate-limit, auth) and should halt the sweep upstream. raise async def _lookup_user_id_by_email(self, email: str) -> str | None: """Fallback lookup. ``FeishuApprovalClient.resolve_user_id_by_email`` raises ``FEISHU_USER_NOT_FOUND`` on miss; we translate to None and re-raise everything else. """ try: return await self._client.resolve_user_id_by_email(email) except AppException as e: if e.code == "FEISHU_USER_NOT_FOUND": return None raise async def _lookup_user_id_by_mobile(self, mobile: str) -> str | None: """Mobile fallback lookup (mirrors :meth:`_lookup_user_id_by_email`). ``FeishuApprovalClient.resolve_user_id_by_mobile`` raises ``FEISHU_USER_NOT_FOUND`` on miss; we translate to None and re-raise everything else (systemic auth / rate-limit). """ try: return await self._client.resolve_user_id_by_mobile(mobile) except AppException as e: if e.code == "FEISHU_USER_NOT_FOUND": return None raise async def _verify_mobile_hit( self, *, mobile_hit: str, payload: dict[str, Any], person_email: str, ) -> str | None: """Hijack guard for a mobile-fallback hit (#7). A shared/recycled HRIS mobile could resolve to a DIFFERENT person's Feishu account; adopting it blindly would overwrite their email / name / depts (the Beisen-collision class). Verify with ONE targeted GET of the matched user (cheap — no directory walk, so the mobile fast-path stays cheap) and compare its ``employee_no`` to this person's: * matched user's ``employee_no`` is present and DIFFERS from this person's → the number belongs to someone else → REJECT (return None → fall through to the employee_no/create path); * matches, or the matched user has no employee_no (Feishu field empty / app lacks the read scope), or this person has no employee_no, or the verify GET fails non-systemically → adopt the hit (can't prove a hijack; preserve the prior best-effort match). Systemic errors (auth / rate-limit) propagate to halt the sweep. """ emp_no = str((payload or {}).get("employee_no") or "").strip().upper() if not emp_no: return mobile_hit # nothing to verify against safe = quote(mobile_hit, safe="") try: data = await self._client.request( "GET", f"/contact/v3/users/{safe}?user_id_type=user_id", ) except AppException as e: if is_systemic_feishu_failure(e): raise logger.info( "feishu_user: could not verify mobile hit %s for %s (GET %s); " "adopting unverified", mobile_hit, person_email, e.code, ) return mobile_hit matched_emp = str( ((data or {}).get("user") or {}).get("employee_no") or "", ).strip().upper() if matched_emp and matched_emp != emp_no: logger.warning( "feishu_user: mobile hit %s for %s (employee_no=%s) belongs to " "a DIFFERENT employee_no %s — REJECTING to avoid hijacking " "their account; falling through to create/employee_no match", mobile_hit, person_email, emp_no, matched_emp, ) return None return mobile_hit async def _employee_no_index(self) -> dict[str, str]: """Lazy, once-per-sweep ``{employee_no(UPPER) → user_id}`` index. Feishu cannot be QUERIED by employee_no (``user_id_type`` rejects it; ``batch_get_id`` is email/mobile only), so the ONLY way to match on the HRIS natural key is to enumerate the directory and index it locally. Built once (cached on the runner) from :func:`fetch_feishu_directory` — the same fail-loud full-tenant walk the vendor-snapshot uses. Only the FIRST caller pays the walk; the rest of the sweep reuses the cache. First-writer-wins on a duplicate employee_no (Feishu shouldn't have dupes; if it does, the earliest wins deterministically rather than last-seen-in-BFS). Raises (systemic Feishu failures / a truncated walk) propagate to the caller, which downgrades to a best-effort miss — a failed index build must never fail the person, only forgo the employee_no match. """ if self._emp_no_index is not None: return self._emp_no_index # A build failure earlier this sweep is cached and re-raised WITHOUT # re-walking — otherwise every subsequent fallback person would re-run # the whole full-tenant walk, turning one failure into an O(N) # directory-call storm (which itself trips the tenant rate-limit). if self._emp_no_index_error is not None: raise self._emp_no_index_error try: users, _depts = await fetch_feishu_directory(self._client) except Exception as e: # cache + propagate (fail-loud contract) self._emp_no_index_error = e raise idx: dict[str, str] = {} for u in users: eno = str((u or {}).get("employee_no") or "").strip().upper() uid = str((u or {}).get("user_id") or (u or {}).get("open_id") or "") if eno and uid and eno not in idx: idx[eno] = uid self._emp_no_index = idx if not idx and users: # Built a NON-EMPTY directory but ZERO employee_no entries ⇒ the # Feishu app almost certainly lacks the ``contact:user.employee_id`` # read scope (the field is omitted, not errored). Without it every # mismatched-email person would silently POST a duplicate — WARN so # ops can grant the scope, instead of a benign-looking info log. logger.warning( "feishu_user: employee_no index is EMPTY across %d directory " "users — the Feishu app likely lacks the employee_no read " "scope (contact:user.employee_id:readonly); the employee_no " "duplicate-guard is INERT until it is granted.", len(users), ) else: logger.info( "feishu_user: built employee_no→user_id index (%d entries) " "from %d directory users", len(idx), len(users), ) return self._emp_no_index def _formatted_mobile_for_lookup( self, payload: dict[str, Any], ) -> str | None: """E.164-format the payload mobile for a ``batch_get_id`` lookup, using the SAME formatting :meth:`_build_user_body` applies before sending it to Feishu — so the number we query equals the number Feishu stores. Returns None when there is no usable mobile (caller then falls back to the create path). """ if not isinstance(payload, dict): return None raw = payload.get("mobile") if not isinstance(raw, str) or not raw: return None return _format_mobile_e164( raw, payload.get("country"), raw_country_iso3=payload.get("mobile_country_iso3"), phone_country_iso2=payload.get("mobile_country_iso2"), ) async def _get_user_current_dept_state( self, user_id: str, ) -> tuple[list[str] | None, str | None]: """Read the user's current ``department_ids`` + primary dept for the merge / primary-department steps. Reads only. Returns ``(department_ids, primary_department_od)``: * ``department_ids`` — ``list[str]`` (possibly empty) on success; ``None`` on read failure — caller MUST skip the ``department_ids`` slot entirely so Feishu's slot-omitted PATCH semantics preserve the existing depts. * ``primary_department_od`` — the od-XXX currently flagged ``is_primary_dept`` in the user's ``orders`` list, or ``None`` when unknown (read failed / no orders / none flagged). Drives the "only re-write ``orders`` when the business dept isn't already primary" guard in :meth:`upsert`, so a correctly-synced user never churns. Why None vs [] on department_ids: Feishu PATCH replaces the slot wholesale when ``department_ids`` is present. If we returned [] on read failure, the caller might inadvertently send empty list and wipe every existing dept membership for that user. Returning None makes the caller MUST handle "skip the slot" explicitly. We catch :class:`AppException` only — TypeError / KeyError from a structural response change SHOULD propagate so a real bug surfaces instead of silently degrading. """ safe = quote(user_id, safe="") try: data = await self._client.request( "GET", f"/contact/v3/users/{safe}" "?user_id_type=user_id&department_id_type=open_department_id", ) except AppException as e: logger.warning( "feishu_user: GET current dept_ids for %s failed: %s. " "Will PATCH without department_ids slot to preserve " "Feishu state.", user_id, e.code, ) return None, None user = (data or {}).get("user") or {} depts = user.get("department_ids") or [] dept_ids = [str(x) for x in depts if isinstance(x, str) and x] primary_od: str | None = None for order in user.get("orders") or []: if not isinstance(order, dict): continue if order.get("is_primary_dept"): pid = order.get("department_id") if isinstance(pid, str) and pid: primary_od = pid break return dept_ids, primary_od def _translate_helix_dept_codes( self, helix_codes: list[str], ) -> tuple[list[str], list[str]]: """HRIS dept_code list → (resolved Feishu od-XXX list, unresolved HRIS codes list). Resolved are pushed into the Feishu payload; unresolved are logged so the admin can see "dept sync hasn't reached these yet — next cron will fix" without per-user noise. """ resolved: list[str] = [] unresolved: list[str] = [] for code in helix_codes: mapped = self._helix_dept_to_external.get(code) if mapped: resolved.append(mapped) else: unresolved.append(code) return resolved, unresolved def _merge_department_ids( self, *, existing_on_feishu: list[str], helix_resolved_externals: list[str], preserve_existing_helix: bool = False, ) -> list[str]: """Slice-replace merge: keep Feishu-only depts that Helix doesn't manage (geographic offices etc.), replace the Helix-managed slice with the freshly resolved list. Order in the result: 1. Helix-managed depts in the freshly-resolved order (so newer HRIS dept changes land at the visible position). 2. Non-Helix-managed existing depts in their original order. ``preserve_existing_helix`` — set when the person has UNRESOLVED HRIS dept codes this sweep (a dept not yet adopted into Feishu). In that case we CANNOT tell "removed from HRIS membership" apart from "not-yet-resolved", so dropping the existing Helix-managed ods would leave the user DEPARTMENT-LESS on Feishu during the gap (the strip PATCHes through before the downgrade-to-failed). So we KEEP the existing Helix-managed ods (stale-but-present) and merely add the resolved ones — the correct set lands once the unresolved codes adopt. Only when EVERYTHING resolved is it safe to drop the Helix-managed ods that genuinely fell out of HRIS membership. Dedup preserved. """ out: list[str] = [] seen: set[str] = set() # Helix-resolved first (with order preserved). for od in helix_resolved_externals: if od and od not in seen: out.append(od) seen.add(od) # Then non-Helix-managed existing depts. helix_set = self._helix_external_set for od in existing_on_feishu: if not od or od in seen: continue if od in helix_set and not preserve_existing_helix: # A Helix-managed dept REMOVED from the user's HRIS # membership — drop it (HRIS authority wins over Helix's # slice). Skipped when there are unresolved codes, to avoid # a transient department-less state (see docstring). continue out.append(od) seen.add(od) # Strip the root sentinel "0" once the user has ANY real department. # "0" is only ever a create-time placeholder (see _FEISHU_ROOT_DEPT_ID); # if it survives alongside a real dept, Feishu keeps listing the user # at the org root (P10011 Eddie Jiang). Drop it here so the next PATCH # re-parents them cleanly. Keep it ONLY when it is the sole membership # — removing the last dept would strand the user department-less, which # Feishu rejects on PATCH. real = [od for od in out if od != _FEISHU_ROOT_DEPT_ID] return real if real else out @staticmethod def _build_dept_orders( dept_ids: list[str], primary_od: str | None, ) -> list[dict[str, Any]]: """Build Feishu's ``orders`` wire list, making ``primary_od`` the user's PRIMARY department (the one their contact card / org-tree placement displays) and every other dept non-primary. Feishu determines the displayed ("主部门") department from ``orders[].is_primary_dept``, NOT from ``department_ids`` order (verified live 2026-07-07: PATCHing department_ids order alone does NOT move the primary). ``orders`` MUST be sent together with ``department_ids`` (API code 44002 otherwise). The primary gets the highest ``department_order`` so it also sorts first, mirroring the shape correctly-synced users already carry. Dedup preserves first occurrence. ``primary_od`` is floated to the front; when it's falsy or absent from ``dept_ids`` no dept is flagged primary (Feishu keeps its own choice). """ ordered = list(dict.fromkeys(d for d in dept_ids if d)) if primary_od and primary_od in ordered: ordered = [primary_od] + [d for d in ordered if d != primary_od] n = len(ordered) return [ { "department_id": od, "department_order": n - i, "is_primary_dept": bool(primary_od and od == primary_od), "user_order": 0, } for i, od in enumerate(ordered) ] def _filter_and_translate_custom_attrs( self, custom_attrs: Any, ) -> tuple[list[dict[str, Any]], list[str]]: """Translate payload's ``custom_attrs.`` dict into Feishu's wire list ``[{type, id, value: {...}}]``. Returns ``(feishu_list, dropped_names)``. Dropped names are logged at info level so admins can see what their mapping rule produced that Feishu doesn't have a schema for. When :attr:`_custom_attr_catalog` is None (probe failed at build), OMIT custom_attrs entirely from the wire body — Feishu PATCH leaves the slot unchanged when absent. The alternative (passing raw names through as ``id=``) makes Feishu 400 the WHOLE user write, breaking every push that follows in the sweep. Better to skip a few per-user custom_attr updates than block the sweep. """ if not isinstance(custom_attrs, dict) or not custom_attrs: return [], [] catalog = self._custom_attr_catalog if catalog is None: # Probe failed at build — degrade gracefully. The runner # has already logged the probe failure at WARN level # (prime_custom_attrs); a per-user log here would spam. return [], [] out: list[dict[str, Any]] = [] dropped: list[str] = [] for raw_name, raw_value in custom_attrs.items(): if raw_value is None or raw_value == "": continue norm = _normalise_attr_name(raw_name) entry = catalog.get(norm) if entry is None: dropped.append(str(raw_name)) continue attr_type = entry.get("type") or "TEXT" # Today only TEXT is fully wired. HREF / ENUMERATION / # GENERIC_USER need richer payload shapes (option_id for # ENUM, {href, text} for HREF, user reference for # GENERIC_USER) — admins who need those map them outside # the runner for now. Log + skip non-TEXT entries. if attr_type != "TEXT": dropped.append(f"{raw_name} (type={attr_type} not yet supported)") continue out.append({ "type": attr_type, "id": entry["id"], "value": {"text": str(raw_value)}, }) if dropped: logger.info( "feishu_user: dropping custom_attrs not in Feishu catalog " "or unsupported type: %s. Configure in 飞书后台 → 通讯录 → " "自定义字段 to start syncing these.", dropped, ) return out, dropped def _build_hrbp_custom_attr( self, hrbp_user_id: str | None, ) -> dict[str, Any] | None: """Build the HRBP **GENERIC_USER** custom-attr wire entry, or None. Feishu's HRBP field is a USER-typed enterprise custom attribute (verified live 2026-06-16 — schema type ``GENERIC_USER``, wire value ``{"generic_user": {"id": , "type": 1}}``). The schema id is discovered from the live catalog by its ``HRBP`` i18n_name (no hardcoded ``C-...`` id), so a tenant that renames or re-creates the field keeps working after the next catalog prime. Returns None (→ HRBP slot omitted, Feishu keeps its existing value) when: no HRBP resolved, the catalog probe failed, or the tenant has no HRBP custom attribute / it isn't GENERIC_USER-typed. """ if not hrbp_user_id: return None catalog = self._custom_attr_catalog if not catalog: return None entry = catalog.get(_normalise_attr_name("HRBP")) if not entry or (entry.get("type") or "") != "GENERIC_USER": return None return { "type": "GENERIC_USER", "id": entry["id"], "value": {"generic_user": {"id": hrbp_user_id, "type": 1}}, } def _build_user_body( self, *, person_email: str, payload: dict[str, Any], leader_user_id: str | None, department_ids_feishu: list[str] | None, hrbp_user_id: str | None = None, orders: list[dict[str, Any]] | None = None, ) -> dict[str, Any]: """Translate the rendered mapping payload into a Feishu ``POST/PATCH /contact/v3/users`` body shape. Aliases: * ``leader_user_id_email`` → drop; replaced by pre-resolved ``leader_user_id``. * ``hrbp_user_id_email`` → drop; replaced by the pre-resolved ``hrbp_user_id`` written as the HRBP GENERIC_USER custom attr. * ``department_ids`` rendered string → pre-merged list[str] (or ``None`` to skip the slot entirely). * ``orders`` → Feishu's per-user department ordering / primary flag. Only set when the caller wants to assert the primary department; Feishu requires it be sent WITH ``department_ids``. * ``custom_attrs.`` dict → translated Feishu list. """ clean = dict(payload) # Private orchestration input, never a Contact API field. It is # consumed only when this runner creates a previously-missing user. residence_country = clean.pop("_helix_residence_country", None) clean.pop("leader_user_id_email", None) clean.pop("hrbp_user_id_email", None) clean.pop("department_ids", None) # ``office_dept_name`` is a synthetic Helix-side field — # consumed in :meth:`upsert` to resolve an office od-XXX and # appended to ``department_ids``; not a real Feishu wire slot. clean.pop("office_dept_name", None) # ``join_time`` — Feishu's wire format is unix SECONDS (int); # the mapping engine renders {hire_date} as "YYYY-MM-DD". # Convert here; an unparseable value is dropped (the rest of # the upsert must not fail over one optional attribute). jt = clean.pop("join_time", None) if jt not in (None, ""): try: if isinstance(jt, (int, float)): clean["join_time"] = int(jt) else: from datetime import UTC as _UTC from datetime import datetime as _dt clean["join_time"] = int( _dt.strptime(str(jt)[:10], "%Y-%m-%d") .replace(tzinfo=_UTC).timestamp() ) except (ValueError, TypeError): logger.info( "feishu_contact: unparseable join_time %r dropped", jt, ) # ``gender`` + ``employee_type`` — Feishu's wire format is an # integer enum; the mapping engine renders the value-mapped code # as a string ("1"/"2"/…). Convert to int here; an empty (value-map # miss) or unparseable value is dropped so the slot is omitted and # Feishu keeps its existing value. Mirrors the join_time handling. for _enum_field in ("gender", "employee_type"): _raw = clean.pop(_enum_field, None) if _raw in (None, ""): continue try: clean[_enum_field] = int(_raw) except (ValueError, TypeError): logger.info( "feishu_contact: unparseable %s %r dropped", _enum_field, _raw, ) # ``mobile`` — Feishu wants the international calling code # (``+86…``); the mapping engine renders {mobile_phone} as the # raw national number. Resolve the code from the person's country # (still in ``clean`` as the {country} rule's output), matching # the legacy Authing 飞书 task's ``(phoneCountryCode||'+86')+phone``. # Empty / digit-less → drop the slot so Feishu keeps its existing # value (never NULL-out). Idempotent on already-"+" numbers. # Synthetic Helix-side field: the person's PRE-value-map HRIS country # (ISO-3). Consumed here to disambiguate the mobile dialing code and # popped so it never reaches Feishu's wire body. Absent on targets # whose mapping_config predates the rule → formatter falls back to the # pre-fix behaviour (empty country → +86), which is safe. _raw_country_iso3 = clean.pop("mobile_country_iso3", None) # Synthetic Helix-side field: the mobile's OWN country ISO-2 (SF # PerPhone.countryCode → externalCode). Popped so it never reaches # Feishu's wire body; takes precedence in the formatter so a # host-country phone on an international assignee uses its real # dialing code. Absent on targets whose mapping predates the rule → # formatter falls back to home-country derivation (prior behaviour). _phone_country_iso2 = clean.pop("mobile_country_iso2", None) _mobile = clean.get("mobile") if isinstance(_mobile, str) and _mobile: _formatted = _format_mobile_e164( _mobile, clean.get("country"), raw_country_iso3=_raw_country_iso3, phone_country_iso2=_phone_country_iso2, ) if _formatted: clean["mobile"] = _formatted else: clean.pop("mobile", None) custom_attrs_list, _ = self._filter_and_translate_custom_attrs( clean.pop("custom_attrs", None), ) # HRBP is a GENERIC_USER custom attr (resolved separately from the # TEXT-only path above, which drops non-TEXT types). Append it when # resolved + the tenant has the schema. hrbp_attr = self._build_hrbp_custom_attr(hrbp_user_id) if hrbp_attr is not None: custom_attrs_list = [*custom_attrs_list, hrbp_attr] body: dict[str, Any] = { "email": (clean.pop("email", None) or person_email or "").lower(), } # Pass through every other rendered field. Feishu wants ``name`` # as a plain string at top level; mapping rules render # ``{full_name}`` → ``name`` directly, no nesting. for k, v in clean.items(): if v is None: continue if isinstance(v, str) and not v: continue body[k] = v if leader_user_id: body["leader_user_id"] = leader_user_id if department_ids_feishu is not None: body["department_ids"] = department_ids_feishu # ``orders`` MUST accompany ``department_ids`` (Feishu 44002) — # only emit it when both the caller supplied orders AND the # department_ids slot is present in this body. if orders and department_ids_feishu is not None: body["orders"] = orders if custom_attrs_list: body["custom_attrs"] = custom_attrs_list if residence_country is not None: body["_helix_residence_country"] = residence_country return body # ── SyncRunner implementation ──────────────────────────────────── async def upsert( self, *, person_email: str, payload: dict[str, Any], prior_external_id: str | None, ) -> PersonOutcome: # Reset per-person write observability — a person resolved without # any vendor write (e.g. suppressed) must not inherit the previous # person's captured request/response. self._last_write_request = None self._last_write_response = None if not person_email: return PersonOutcome( status="failed", error="person_email is empty — cannot push to Feishu", ) # ── Resolve user_id by email, then mobile. ────────────────── # Feishu offers NO way to look a user up by employee_no: the # ``GET /contact/v3/users/{id}`` path accepts ``user_id_type`` ∈ # {open_id, union_id, user_id} ONLY (employee_no / employee_id # both 99992402 "field validation failed"), and # ``batch_get_id`` resolves by emails / mobiles only. So — like # the legacy Authing 飞书 sync, which associated on BOTH mobile # AND email — we resolve by email first, then fall back to # mobile. employee_no still goes into the create body so a fresh # POST stamps the HRIS id, but it can never be a LOOKUP key. user_id: str | None = prior_external_id if not user_id: try: user_id = await self._lookup_user_id_by_email(person_email) except AppException as e: # Auth / rate-limit failures are SYSTEMIC — propagate # so the orchestrator can abort the whole sweep with # one clear error instead of burning quota writing # N "auth failed" state rows under bad creds. if is_systemic_feishu_failure(e): raise return PersonOutcome( status="failed", error=f"email lookup failed: {e.code}: {e.message}", ) if not user_id: # Mobile fallback. An existing Feishu account whose registered # email differs from work_email (a ``jingwen.wang1`` typo, a # .com/.com.cn split, …) is missed by the email lookup. Without # this it falls through to POST; Feishu then matches the account # by mobile SERVER-SIDE and returns the existing id as a # "create" WITHOUT applying our email — so the wrong email is # frozen forever behind hash-skip (P10012 王静文, 2026-07-13). # Resolving the id here routes the person through the PATCH path # below, which pushes email=work_email and self-corrects it. # Mobile is globally unique on Feishu — BUT the mobile we send is # from THIS person's HRIS row, and two HRIS rows can share a number # (data-entry dupes, carrier-recycled numbers, test data). A blind # adopt would PATCH person B's account with person A's email/name/ # depts (the Beisen-collision class). So a mobile hit is verified # against the employee_no directory index before adoption (#7). lookup_mobile = self._formatted_mobile_for_lookup(payload) if lookup_mobile: try: mobile_hit = await self._lookup_user_id_by_mobile(lookup_mobile) if mobile_hit: user_id = await self._verify_mobile_hit( mobile_hit=mobile_hit, payload=payload, person_email=person_email, ) except AppException as e: # Systemic (auth / rate-limit) still halts the sweep. if is_systemic_feishu_failure(e): raise # Otherwise the mobile fallback is best-effort: a # non-systemic miss/error (e.g. a malformed mobile # Feishu rejects) must NOT fail the person — fall # through to the create path (prior behaviour). logger.info( "feishu_user: mobile fallback lookup for %s did not " "resolve (%s: %s); proceeding to create path", person_email, e.code, e.message, ) if not user_id: # Last-resort match by employee_no (the HRIS natural key). Email # AND mobile both missed — the person either genuinely has no # Feishu account, or has one whose registered email differs AND # whose mobile is absent/unmatched (so Feishu's server-side POST # dedup can't save us from a DUPLICATE). Feishu can't be queried # by employee_no, so consult the lazily-built directory index. # A hit routes them through PATCH (self-correcting email), closing # the residual duplicate-account gap; a miss legitimately falls # through to POST (they truly don't exist). # # FAIL-LOUD on a build failure — do NOT swallow-to-POST. This is the # LAST line of defense against a duplicate (unlike the mobile # fallback, whose miss is backstopped by Feishu's server-side POST # mobile-dedup, this fallback has NO backstop). ``fetch_feishu_ # directory`` is deliberately fail-loud so a truncated/partial walk # is never mistaken for "person absent"; honour that here — a failed # index build fails the PERSON (return failed, hash unwritten → # retried next sweep) rather than POSTing a duplicate for someone # who may well be in the directory. Systemic errors re-raise to halt # the whole sweep. emp_no = str((payload or {}).get("employee_no") or "").strip() if emp_no: try: idx = await self._employee_no_index() hit = idx.get(emp_no.upper()) if hit: user_id = hit logger.info( "feishu_user: matched %s by employee_no %s → %s " "(directory index) — routing to PATCH not POST", person_email, emp_no, hit, ) except AppException as e: if is_systemic_feishu_failure(e): raise logger.warning( "feishu_user: employee_no index build failed for %s " "(%s: %s); FAILING the person (not creating) to avoid a " "possible duplicate — retries next sweep", person_email, e.code, e.message, ) return PersonOutcome( status="failed", error=( f"employee_no index build failed ({e.code}: " f"{e.message}); refused create to avoid duplicate" ), ) except Exception as e: # noqa: BLE001 logger.exception( "feishu_user: employee_no index build failed for %s; " "FAILING the person (not creating) to avoid a possible " "duplicate — retries next sweep", person_email, ) return PersonOutcome( status="failed", error=( f"employee_no index build failed " f"({type(e).__name__}: {e}); refused create to " "avoid duplicate" ), ) # ── Resolve leader + translate departments. ───────────────── leader_email = (payload or {}).get("leader_user_id_email") try: leader_user_id = await self._resolve_leader_user_id( leader_email if isinstance(leader_email, str) else None, ) except AppException as e: # Same auth-failure rule for the leader lookup. A rate- # limit hit here used to silently drop the leader slot for # every user — now it propagates so the sweep halts and # admin sees one alert instead of N silent corruptions. if is_systemic_feishu_failure(e): raise # Anything else (e.g. 5xx on a specific leader lookup) — # log + drop the leader slot for this one user. Hash gets # re-computed without the leader so the next sweep retries. logger.warning( "feishu_user: leader %r resolve failed: %s — dropping " "leader slot for this user", leader_email, e.code, ) leader_user_id = None # ── Resolve HRBP (GENERIC_USER custom attr). ──────────────── # Same email→user_id resolver + same systemic-vs-transient policy # as the leader: a rate-limit/auth failure halts the sweep; a # one-off miss just drops the HRBP slot for this user (next sweep # retries via hash recompute). hrbp_email = (payload or {}).get("hrbp_user_id_email") hrbp_user_id: str | None = None try: hrbp_user_id = await self._resolve_leader_user_id( hrbp_email if isinstance(hrbp_email, str) else None, ) except AppException as e: if is_systemic_feishu_failure(e): raise logger.warning( "feishu_user: hrbp %r resolve failed: %s — dropping " "HRBP slot for this user", hrbp_email, e.code, ) hrbp_user_id = None # Surface the SILENT miss too: the HRBP email was set but no Feishu # user resolved (even after the .com/.com.cn sibling fallback), so # the HRBP slot is omitted and Feishu keeps its prior value. Log at # WARN so admins can see WHY a Feishu HRBP didn't update instead of # it vanishing without a trace (2026-07-06). if hrbp_email and not hrbp_user_id: logger.warning( "feishu_user: HRBP %r for %s did not resolve to a Feishu " "user — HRBP field left unchanged (not written)", hrbp_email, person_email, ) helix_dept_codes = _coerce_department_ids( (payload or {}).get("department_ids"), ) resolved_business, unresolved = self._translate_helix_dept_codes( helix_dept_codes, ) resolved_externals = list(resolved_business) # The person's PRIMARY business department (their primary org # membership's leaf dept, rendered first into ``department_ids``) # must become the Feishu PRIMARY dept — otherwise the office node # appended below (SV Office etc.) shadows it on the contact card. # Falls back to None when no business dept resolved (dept sync # hasn't adopted it yet); the primary flag is then left to Feishu # rather than forcing the office node as primary. primary_business_od: str | None = ( resolved_business[0] if resolved_business else None ) # A partial dept push MUST NOT settle as ``success``. The # orchestrator writes ``last_payload_hash`` on success (and clears # ``last_error``), and the payload hashes the HRIS ``dept_code`` — # NOT the Feishu-side ``od`` adoption state. So a person whose dept # was only PARTIALLY pushed this sweep hashes identically next # sweep and is hash-skipped forever, never reconverging (#1737). # Track WHY the dept dimension is incomplete; the tail of this # method downgrades an otherwise-``success`` outcome to ``failed`` # so the next sweep retries until the dept fully resolves. dept_degraded_reason: str | None = None if unresolved: logger.info( "feishu_user: %r — %d HRIS dept_code(s) not yet adopted " "into Feishu (%s); will push only resolved depts and " "preserve non-Helix-managed Feishu depts via merge.", person_email, len(unresolved), unresolved, ) dept_degraded_reason = ( f"{len(unresolved)} HRIS dept_code(s) not yet adopted into " f"Feishu ({unresolved}) — dept membership not fully pushed; " "retrying next sweep once dept sync adopts them" ) # SECOND dept dimension — office mapping. The ``office_dept_name`` # field is rendered by the mapping rule's value_map (e.g. # work_location "Tokyo" → "Tokyo Office"). Look it up in the # office name → od-XXX index built at runner-build time. Miss # is benign (admin hasn't curated this work_location yet or # the office dept doesn't exist on Feishu) — just skip the # office slot, don't fail the user push. office_name = (payload or {}).get("office_dept_name") if isinstance(office_name, str) and office_name: office_od = self._office_name_to_external.get( _normalise_office_name(office_name), ) if office_od and office_od not in resolved_externals: resolved_externals.append(office_od) elif not office_od: logger.info( "feishu_user: %r — office_dept_name=%r not in the " "Feishu catalog; skipping office assignment " "(extend the office mapping or wait for the catalog " "to refresh).", person_email, office_name, ) # Admin skip-list: never CREATE or REOPEN this person. An # existing active record is still UPDATED (below), but a missing # one is not provisioned and a frozen one is not reopened. Keyed # on the same work_email the sweep uses for the Feishu lookup. suppress = person_email.lower() in self._skip_provision_emails # ── PATCH (existing user) — merge dept_ids first. ─────────── if user_id: existing_dept_ids, current_primary_od = ( await self._get_user_current_dept_state(user_id) ) orders: list[dict[str, Any]] | None = None if existing_dept_ids is None: # Read failed — DO NOT include department_ids slot in # the PATCH. Feishu will leave the user's current # depts untouched (slot-omitted semantics on PATCH). merged_for_patch: list[str] | None = None else: merged_for_patch = self._merge_department_ids( existing_on_feishu=existing_dept_ids, helix_resolved_externals=resolved_externals, # Unresolved codes present → keep existing Helix-managed # ods so the user is never left department-less in the gap. preserve_existing_helix=bool(unresolved), ) # Assert the business dept as primary ONLY when it isn't # already — a correctly-synced user (business already # primary) gets no ``orders`` slot and therefore never # churns. A user whose primary is wrong (e.g. the office # node became primary because the business dept was # dropped at create time) gets flipped back. # # We must ALSO re-designate a primary when the merge dropped # the user's current primary from the slot. That happens when # the root sentinel "0" — left as primary by a root-fallback # create — is stripped now that a real dept resolved: Feishu # keeps the old primary only while it is still present in # ``department_ids``, so a removed primary with no ``orders`` # would leave the user with an invalid/undefined primary. Pick # the business dept when known, else the first surviving dept. primary_dropped = ( current_primary_od is not None and current_primary_od not in merged_for_patch ) desired_primary: str | None = None if primary_business_od and primary_business_od in merged_for_patch: desired_primary = primary_business_od elif primary_dropped and merged_for_patch: desired_primary = merged_for_patch[0] if desired_primary and desired_primary != current_primary_od: orders = self._build_dept_orders( merged_for_patch, desired_primary, ) body = self._build_user_body( person_email=person_email, payload=payload, leader_user_id=leader_user_id, department_ids_feishu=merged_for_patch, hrbp_user_id=hrbp_user_id, orders=orders, ) if existing_dept_ids is None: # Current-dept read failed → we deliberately omit the # ``department_ids`` slot to preserve Feishu state, but that # means any pending HRIS dept change was NOT applied this # sweep. Mark degraded so the tail downgrades to ``failed`` # (retry converges once the read succeeds) instead of # freezing on a hash-stored plain success (#1737). dept_degraded_reason = ( "current-dept read failed — department_ids slot omitted " "this PATCH; retrying next sweep to converge dept " "membership" ) if self._reopen_frozen and not suppress: # Reopen (打开): assert active so an in-scope, HRIS-active # person frozen on Feishu is un-frozen. Idempotent — # Feishu no-ops if already active. Gated by the target's # reopen_frozen toggle + the skip-list, so intentionally # suspended accounts are left frozen. body.setdefault("status", {})["is_frozen"] = False outcome = await self._patch_user( user_id=user_id, body=body, wanted_externals=resolved_externals, ) return self._downgrade_if_dept_degraded(outcome, dept_degraded_reason) # ── POST (new user). ──────────────────────────────────────── if suppress: # Skip-list: person is missing on Feishu but the admin marked # them not-to-provision (bad/junk email, leaver leftover, # intentional exclusion). Do NOT create — record as skipped # (ledger shows why; hash stored so it won't churn) so no # duplicate / junk account is ever made. return PersonOutcome( status="skipped", external_id=None, error="admin skip-list: creation suppressed", ) # department_ids must be non-empty on create. Use resolved # externals, or fall back to Feishu's root sentinel "0" so the # user lands SOMEWHERE; next sweep moves them. create_dept_ids = resolved_externals or ["0"] # Pin the business dept as primary at creation so a new hire's # card shows their real department (not the appended office node). create_orders = ( self._build_dept_orders(create_dept_ids, primary_business_od) if primary_business_od else None ) body = self._build_user_body( person_email=person_email, payload=payload, leader_user_id=leader_user_id, department_ids_feishu=create_dept_ids, hrbp_user_id=hrbp_user_id, orders=create_orders, ) outcome = await self._create_user(body=body) return self._downgrade_if_dept_degraded(outcome, dept_degraded_reason) @staticmethod def _downgrade_if_dept_degraded( outcome: PersonOutcome, reason: str | None, ) -> PersonOutcome: """Turn a PLAIN ``success`` outcome into ``failed`` when the dept dimension was only partially pushed (#1737). The Feishu write itself succeeded (name / title / leader landed), but because the orchestrator hashes the HRIS dept_code — not the Feishu-side adoption — a plain ``success`` here would store the hash and hash-skip every future sweep, freezing the person in a stale / stripped dept. ``failed`` keeps ``last_payload_hash`` unwritten so the next sweep retries and converges. The ``external_id`` is preserved so the person's Feishu mapping (and thus the PATCH-not-recreate path) survives the failed status. A success that already carries ``slot_outcomes`` is NOT downgraded: that is the resilient-write path's DELIBERATE degraded success (#1956) — a dropped 41041 dept / invalid mobile the vendor won't accept, where retrying the identical write churns forever (the drift re-verify or a payload change re-attempts it instead). Flipping it to ``failed`` here both re-introduced that churn (a person with one unadopted dept code AND a 41041-dropped slot went red every sweep) and — because the original code built a bare PersonOutcome — destroyed the per-slot ledger record. ``slot_outcomes`` are carried forward on every path so the ledger keeps field-level truth. """ if outcome.status == "success" and reason: if outcome.slot_outcomes: return outcome return PersonOutcome( status="failed", external_id=outcome.external_id, error=f"partial dept push: {reason}", slot_outcomes=outcome.slot_outcomes, ) return outcome async def _patch_user( self, *, user_id: str, body: dict[str, Any], wanted_externals: list[str], allow_recreate: bool = True, ) -> PersonOutcome: """Thin wrapper that stamps the REAL vendor op on the outcome. The ledger labels a row 「已创建」 vs 「已更新」 from ``detail['op']``, falling back to ``was_new`` (= the payload hash was NULL) when a runner doesn't report one. This runner reported ``op`` only on the create path, so every PATCH fell through to the fallback — and a mapping-config change NULLs the hash for the whole population. Result: 2026-07-13 / 07-17 / 08-18 each recorded a few-hundred-row burst of 「已创建」 for people who joined in 2018–2022 and were merely being updated (411 such rows in August alone). Anyone reading the ledger to answer "did Helix create this account?" got a confident wrong answer — it cost this incident's investigation a wrong conclusion about when creation last worked. ``netsuite`` and ``huilianyi`` already stamp ``op``; feishu was the only runner that didn't. Set here rather than at each of ``_patch_user_inner``'s many returns so no future return path can forget. The recreate branch inside delegates to ``_create_user``, which stamps ``op='create'`` itself — an already-set op is never overwritten. """ outcome = await self._patch_user_inner( user_id=user_id, body=body, wanted_externals=wanted_externals, allow_recreate=allow_recreate, ) if outcome.status == "success" and not (outcome.detail or {}).get("op"): outcome.detail = {**(outcome.detail or {}), "op": "update"} return outcome async def _patch_user_inner( self, *, user_id: str, body: dict[str, Any], wanted_externals: list[str], allow_recreate: bool = True, ) -> PersonOutcome: # Do not leak Helix's creation-only residency hint into a normal # Contact PATCH — but KEEP the value: the user-absent branch below # hands this same ``body`` to :meth:`_create_user`, whose Directory # v1 create hard-BLOCKS without it. Popping it unconditionally (as # 3a0adf7e2 did) made every PATCH→recreate return # ``blocked/missing_residence_country`` without ever issuing the # create, so a person whose Feishu account was deleted could never # be rebuilt. body = dict(body) residence_hint = body.pop("_helix_residence_country", None) safe = quote(user_id, safe="") try: await self._write( "PATCH", f"/contact/v3/users/{safe}" "?user_id_type=user_id&department_id_type=open_department_id", body=body, ) return PersonOutcome(status="success", external_id=user_id) except AppException as e: num = _parse_feishu_code(e) # Disambiguate the overloaded 230003 code: if the PATCH # carried department_ids AND the error mentions a dept # (localised "department" / "部门"), treat it as a # dept-not-found and retry without dept_ids — NOT as # "user vanished". Without this, a stale dept_state entry # silently fails the PATCH path until next sweep. dept_not_found = num in _DEPT_NOT_FOUND_CODES or ( num == 230003 and _response_mentions_department(e) ) # ``not_assigned`` is mutually exclusive with ``not_found`` (which # takes precedence — a stale/missing od self-heals on re-adopt and # must stay a transient ``failed``+retry, #1737). Beyond the # explicit 41041, defence-in-depth: any not-yet-enumerated code # whose message clearly references a department (and isn't a # user-absent / mobile code) is treated as a droppable dept-slot # rejection rather than freezing the whole person — never silently; # the slot outcome + ledger record exactly what happened. # # The heuristic REQUIRES a parsed Feishu business code # (``num is not None``): a code-less error is not a vendor # policy verdict but a transport/5xx-shaped failure, and a # transient 5xx whose message merely mentions "department" # must stay failed+retry — classifying it dept_not_assigned # dropped the dept slot, degraded to success, and froze the # wrong state behind the hash-skip. dept_not_assigned = (not dept_not_found) and ( num in _DEPT_NOT_ASSIGNED_CODES or ( num is not None and num not in _USER_ABSENT_CODES and num not in _MOBILE_INVALID_CODES and _response_mentions_department(e) ) ) dept_scope_error = "department_ids" in body and ( dept_not_found or dept_not_assigned ) if dept_scope_error: dropped_depts = body.get("department_ids") logger.warning( "feishu_user: PATCH %s dept-slot rejected (code=%s, " "not_assigned=%s, depts=%s); retrying WITHOUT the " "department_ids slot so the person's other fields still " "apply — a single un-appliable dept must not freeze the " "whole record.", user_id, num, dept_not_assigned, dropped_depts, ) fallback = dict(body) fallback.pop("department_ids", None) # ``orders`` cannot be sent without ``department_ids`` # (Feishu 44002) — drop it too. fallback.pop("orders", None) extra_slots: list[dict[str, Any]] = [] try: await self._write( "PATCH", f"/contact/v3/users/{safe}" "?user_id_type=user_id&department_id_type=open_department_id", body=fallback, ) except AppException as e2: # Chain the degrade (#6b): a person with BOTH a bad dept # AND a bad mobile (realistic for international assignees) # would otherwise hard-fail here. If the dept-dropped retry # now trips the mobile validator, drop mobile too and retry # once more so the remaining fields still land. n2 = _parse_feishu_code(e2) if n2 in _MOBILE_INVALID_CODES and "mobile" in fallback: fallback.pop("mobile", None) try: await self._write( "PATCH", f"/contact/v3/users/{safe}" "?user_id_type=user_id" "&department_id_type=open_department_id", body=fallback, ) extra_slots.append({ "slot": "mobile", "status": "dropped", "feishu_code": n2, }) except AppException as e3: return PersonOutcome( status="failed", external_id=user_id, error=( f"PATCH /users/{user_id} dept+mobile " f"fallback failed: {e3.code}: {e3.message}" ), ) else: return PersonOutcome( status="failed", external_id=user_id, error=( f"PATCH /users/{user_id} dept-slot fallback " f"failed: {e2.code}: {e2.message}" ), ) slot = { "slot": "department_ids", "status": "dropped", "feishu_code": num, "value": dropped_depts, "detail": (e.message or "")[:200], } if dept_not_assigned: # A REAL, resolved dept the vendor won't let us assign — # retrying the same thing won't help, so DON'T churn the # person as failed forever. The other fields landed; the # person is usable. Degrade to success + flag the dept for # IT (surfaced via slot_outcomes on the ledger). If the # underlying permission is later fixed, drift re-verify # (or a payload change) re-attempts the dept. return PersonOutcome( status="success", external_id=user_id, error=( f"partial: department not applied — Feishu " f"rejected the dept assignment (code {num}); " "other fields applied. Flagged for IT: the app " "cannot assign this user into " f"{dropped_depts}." ), slot_outcomes=[slot] + extra_slots, ) # dept-not-found / stale adoption — TRANSIENT: self-heals once # dept sync re-adopts the od. Keep ``failed`` so the hash stays # unwritten and the next sweep retries and converges (#1737). return PersonOutcome( status="failed", external_id=user_id, error=( "partial dept push: department_ids not pushed (a dept " "od is missing on Feishu — stale adoption); other " "fields applied, retrying next sweep once dept sync " "re-adopts" ), slot_outcomes=[slot] + extra_slots, ) # Mobile-invalid (41004): the number failed Feishu's per-country # validation (bad HRIS data, or a country we couldn't resolve). # Retry WITHOUT the mobile slot so the rest of the person's fields # still update and Feishu keeps its existing number — one bad # optional field must never fail the whole person. if num in _MOBILE_INVALID_CODES and "mobile" in body: logger.warning( "feishu_user: PATCH %s mobile invalid (%s); retrying " "without mobile slot (fix the HRIS phone country).", user_id, num, ) fallback = dict(body) fallback.pop("mobile", None) mob_slots: list[dict[str, Any]] = [{ "slot": "mobile", "status": "dropped", "feishu_code": num, "value": body.get("mobile"), "detail": (e.message or "")[:200], }] try: await self._write( "PATCH", f"/contact/v3/users/{safe}" "?user_id_type=user_id&department_id_type=open_department_id", body=fallback, ) return PersonOutcome( status="success", external_id=user_id, error=( "partial: mobile not pushed (invalid for its " "country on Feishu); fix the HRIS phone / country" ), slot_outcomes=mob_slots, ) except AppException as e2: # Chain (#6b): the mobile-dropped retry now trips a dept # scope — drop department_ids + orders too and retry once. n2 = _parse_feishu_code(e2) # Same guard as the primary dept heuristic (#6): only # a PARSED Feishu code may be classified as a dept # scope error — a code-less transport/5xx that merely # mentions "department" must stay a hard failure and # retry, not silently drop the dept slot. n2_dept = n2 in _DEPT_NOT_ASSIGNED_CODES or ( n2 is not None and n2 not in _USER_ABSENT_CODES and _response_mentions_department(e2) ) if n2_dept and "department_ids" in fallback: dd = fallback.get("department_ids") fallback.pop("department_ids", None) fallback.pop("orders", None) try: await self._write( "PATCH", f"/contact/v3/users/{safe}" "?user_id_type=user_id" "&department_id_type=open_department_id", body=fallback, ) return PersonOutcome( status="success", external_id=user_id, error="partial: mobile + department not pushed " "(both rejected); other fields applied", slot_outcomes=mob_slots + [{ "slot": "department_ids", "status": "dropped", "feishu_code": n2, "value": dd, }], ) except AppException as e3: return PersonOutcome( status="failed", external_id=user_id, error=( f"PATCH /users/{user_id} mobile+dept " f"fallback failed: {e3.code}: {e3.message}" ), ) return PersonOutcome( status="failed", external_id=user_id, error=( f"PATCH /users/{user_id} mobile-fallback failed: " f"{e2.code}: {e2.message}" ), ) # Plain user-absent (no dept_ids in this body OR error # doesn't mention dept). User vanished between lookup # and PATCH — fall through to create. department_ids # must be non-empty on POST. if num in _USER_ABSENT_CODES: if not allow_recreate: # This PATCH is the projection that immediately follows a # create, and Feishu says the user it just handed us isn't # there — a read-after-write propagation window, not a # missing account. Recreating here would build a SECOND # account for the same person (and could recurse). The # account DOES exist and its id is known, so report a # retryable failure that keeps the mapping: the next sweep # PATCHes the existing user and converges. logger.warning( "feishu_user: projection PATCH %s right after create " "returned absent code %s; NOT recreating (the account " "exists — next sweep re-applies the projection).", user_id, num, ) return PersonOutcome( status="failed", external_id=user_id, error=( f"created {user_id} but the Contact projection " f"read it as absent ({num}); fields not applied " "yet — next sweep retries" ), ) logger.info( "feishu_user: PATCH %s returned absent code %s; recreating", user_id, num, ) body["department_ids"] = wanted_externals or ["0"] # The merged ``orders`` was built for the PATCH's merged # dept list (which may reference depts not in # ``wanted_externals``); drop it so the recreate can't 44002 # on a mismatched orders/department_ids pair. The next # sweep re-asserts the primary dept on the recreated user. body.pop("orders", None) # Restore the residency hint stripped at the top of this # method — ``_create_user`` requires it to pick the Multi-Geo # ``geo_name`` and blocks outright when it is absent. if residence_hint is not None: body["_helix_residence_country"] = residence_hint return await self._create_user(body=body) # Resigned account on the UPDATE path (#1958 handled this only in # deactivate). The stored external_id points at a Feishu account # that is already resigned (离职) — an HRIS-active person can't be # updated on it (manual dismissal / rehire / lifecycle race). NOT # transient: re-PATCHing gets 42006 every sweep. BLOCK (persists # cheaply, no churn, no daily red) with an actionable reason; # do NOT auto-recreate (the resigned account still owns the # mobile/email, so a create would collide) — needs a manual # restore/rehire decision. if num in _USER_RESIGNED_CODES: return PersonOutcome( status="blocked", external_id=user_id, error=( "the linked Feishu account is resigned (离职, 42006) " "and cannot be updated — it needs a manual restore / " "rehire in Feishu, or clear the stored link so a fresh " "account is created." ), detail={"blocked_reason": "resigned_account_on_update"}, ) # 41041 "department id is not assigned" that did NOT hit the # dept-slot fallback above — i.e. the body had NO department_ids # slot but Feishu still rejects a dept scope (a stale ``orders`` # primary referencing an unassignable dept). Drop ``orders`` and # retry so the rest of the fields land; if it still refuses, the # user sits in a dept outside the app's 通讯录范围 — BLOCK with an # actionable "grant scope" reason instead of churning failed. if num in _DEPT_NOT_ASSIGNED_CODES and "orders" in body: fb = dict(body) fb.pop("orders", None) fb.pop("department_ids", None) try: await self._write( "PATCH", f"/contact/v3/users/{safe}" "?user_id_type=user_id&department_id_type=open_department_id", body=fb, ) return PersonOutcome( status="success", external_id=user_id, error="partial: primary dept (orders) not applied — " "dropped an unassignable dept scope", slot_outcomes=[{ "slot": "orders", "status": "dropped", "feishu_code": num, }], ) except AppException: return PersonOutcome( status="blocked", external_id=user_id, error=( "this user sits in a Feishu department outside the " "app's 通讯录管理范围 (41041) — grant the app scope " "over their department, then it re-syncs." ), detail={"blocked_reason": "dept_outside_app_scope"}, ) scope_block = _scope_blocked_outcome( e, surface=f"PATCH /users/{user_id}", ) if scope_block is not None: scope_block.external_id = user_id return scope_block return PersonOutcome( status="failed", external_id=user_id, error=f"PATCH /users/{user_id} → {e.code}: {e.message}", ) async def _directory_write_scope_missing(self) -> str | None: """Probe the Directory WRITE scope once per sweep; message or None. Why this exists (2026-08-31 post-mortem). #3084 moved account creation from ``POST /contact/v3/users`` (scope ``contact:user:write``, long granted) to ``POST /directory/v1/employees`` (scope ``directory:employee:write``, never granted). Nothing checks a scope until a write needs it, and no new hire needed a create for the next 12 days — so the create surface was dead the whole time while every sweep ran green on updates. It surfaced only when a real person (P10025) started work with no Feishu account. The probe is a PATCH against a deliberately invalid ``employee_id``. It cannot mutate anything, and it is decisive because this endpoint checks scope BEFORE it validates the id (verified against prod): 99991672 → the scope is missing 99992360 → invalid user_id, i.e. we got PAST the scope gate Called from two places: :meth:`_create_user` (so a doomed POST is never sent and the person gets a precise ``blocked``), and the ``identity_sync.feishu_write_scope_preflight`` scheduled job, which runs daily whether or not anyone is being hired and RAISES on a missing scope so the scheduled-job failure notifier pages IT. The job is the part that would actually have caught #3084 on day one: a runner-side check only ever fires once a hire needs it, which is exactly the 12-day blind spot. Cached per runner instance so a sweep costs at most one probe. Failure to probe is never fatal: any unexpected error returns ``None`` (treat as "scope fine") so a blip can't block creation for everyone — the real create still reports the truth via :func:`_scope_blocked_outcome`. """ if self._directory_scope_probed: return self._directory_scope_error self._directory_scope_probed = True try: await self._client.request( "PATCH", "/directory/v1/employees/helix-scope-probe-000" "?employee_id_type=user_id", json_body={"employee": {}}, ) except AppException as e: if is_feishu_permission_failure(e): self._directory_scope_error = e.message or "scope missing" logger.error( "feishu_user: PREFLIGHT — the Feishu app lacks the " "Directory write scope, so NO new hire can be created " "on this sweep. Grant it and publish a new app " "version. Vendor said: %s", e.message, ) except Exception as e: # noqa: BLE001 logger.warning( "feishu_user: Directory scope preflight inconclusive " "(%s) — proceeding; the create itself still reports " "the truth.", e, ) return self._directory_scope_error async def _create_user(self, *, body: dict[str, Any]) -> PersonOutcome: """Create through Directory v1 so Multi-Geo is fixed at creation. Contact v3 has no ``geo_name`` equivalent. Directory creates the same Feishu employee/user, after which the existing Contact PATCH applies Helix's richer directory fields and department assignments. """ # NOTE: deliberately NOT probing the write scope here. The probe # would save exactly one doomed POST per sweep, and buys nothing # for correctness — a real 99991672 from the POST below is already # classified by ``_scope_blocked_outcome`` into the same actionable # ``blocked``. Early detection (the thing this incident actually # lacked) belongs in the daily # ``identity_sync.feishu_write_scope_preflight`` job, which runs # whether or not anyone is being hired. Putting a probe on the hot # create path instead would add a request to every sweep AND an # extra wire call to every create test, for no behavioural gain. body = dict(body) raw_residence = body.pop("_helix_residence_country", None) residence = normalize_country_code(str(raw_residence or "")) if not residence: return PersonOutcome( status="blocked", error=( "cannot create Feishu account: residence_country is missing " "or invalid; data residency must be chosen explicitly" ), detail={"blocked_reason": "missing_residence_country"}, ) geo_name = ( "cn" if residence in _CN_GEO_RESIDENCE_COUNTRIES else "jp" ) name = str(body.get("name") or body.get("email") or "").strip() if not name: return PersonOutcome( status="blocked", error="cannot create Feishu account: name is required by Directory API", detail={"blocked_reason": "missing_name"}, ) employee: dict[str, Any] = { "name": {"name": {"default_value": name}}, "email": body.get("email"), } # Mobile is best-effort here, and unlike the PATCH path this surface # has NO per-code salvage at all: any rejection above returns # ``failed`` verbatim. Two things the retired Contact v3 create body # did and this one does not: # * 41004 "mobile is invalid" (bad HRIS phone / unresolvable # country) retried WITHOUT the mobile slot, so the account was # still built with a ``dropped`` mobile slot on the ledger. The # PATCH side still does this (see ``_MOBILE_INVALID_CODES``); the # create no longer does, so one bad number fails a NEW hire # outright — pinned in # ``test_create_mobile_invalid_is_a_hard_failure``. # * 41010 "no mobile" (tenant requires one, HRIS row has none) # routed to ``blocked`` rather than ``failed`` — a DATA problem # persists cheaply and auto-retries the moment a mobile arrives, # whereas ``failed`` reddens the run every night. # Directory v1's own code for that second case is NOT verified — do # not re-add a 41010 branch off the Contact v3 number without probing # this endpoint first. if body.get("mobile"): employee["mobile"] = body["mobile"] # No ``employee_type`` on this body. Contact v3's POST required it # (99992402 "employee_type is required"), which forced a create-only # default of 1 (正式员工) for a hire provisioned in their 24h-lead # window, whose effective-dated SF ``employee_type_code`` is still # EMPTY until the start date (2026-07-13, P10011 Eddie Jiang). # Directory v1 does not require the field, so no default is injected # and the Contact projection below carries the real value once HRIS # populates the code. directory_body = { "employee": employee, "options": {"geo_name": geo_name}, } # Set when Feishu rejected the ``mobile`` slot (41004) and the create # had to be retried without it — the hire IS created, the number is # recorded as a dropped slot instead of failing the whole person. mobile_dropped_slot: dict[str, Any] | None = None try: resp = await self._client.request( "POST", "/directory/v1/employees?employee_id_type=user_id" "&department_id_type=open_department_id", json_body=directory_body, ) self._last_write_request = directory_body self._last_write_response = resp except AppException as e: num = _parse_feishu_code(e) # Mobile-invalid (41004) on the Directory create: the number # failed Feishu's per-country validation (bad HRIS data, or a # country we couldn't resolve). Retry the create WITHOUT the # mobile slot so the new hire still gets an account — a bare # ``failed`` here leaves them with NO Feishu at all and reddens # every sweep until someone fixes the HRIS phone. Parity with the # PATCH path (which has always dropped the slot); the drop is # recorded on the outcome so the ledger shows the number didn't # land, exactly the deliberate degraded success of #1956. if num not in _MOBILE_INVALID_CODES or "mobile" not in employee: scope_block = _scope_blocked_outcome( e, surface=f"POST /directory/v1/employees (geo={geo_name})", request_body=directory_body, ) if scope_block is not None: return scope_block return PersonOutcome( status="failed", error=f"POST /directory/v1/employees (geo={geo_name}) → {e.code}: {e.message}", request_body=directory_body, ) logger.warning( "feishu_user: Directory create mobile invalid (%s); retrying " "without the mobile slot (fix the HRIS phone / country).", num, ) mobile_dropped_slot = { "slot": "mobile", "status": "dropped", "feishu_code": num, "value": employee.get("mobile"), "detail": (e.message or "")[:200], } employee = {k: v for k, v in employee.items() if k != "mobile"} directory_body = { "employee": employee, "options": {"geo_name": geo_name}, } try: resp = await self._client.request( "POST", "/directory/v1/employees?employee_id_type=user_id" "&department_id_type=open_department_id", json_body=directory_body, ) self._last_write_request = directory_body self._last_write_response = resp except AppException as e2: scope_block = _scope_blocked_outcome( e2, surface=( "POST /directory/v1/employees mobile-fallback " f"(geo={geo_name})" ), request_body=directory_body, ) if scope_block is not None: return scope_block return PersonOutcome( status="failed", error=( "POST /directory/v1/employees mobile-fallback " f"(geo={geo_name}) → {e2.code}: {e2.message}" ), request_body=directory_body, ) created_id = str((resp or {}).get("employee_id") or "") if not created_id: try: created_id = await self._lookup_user_id_by_email(str(body.get("email") or "")) or "" except AppException as e: return PersonOutcome( status="failed", error=f"Directory created employee but user_id lookup failed: {e.code}: {e.message}", request_body=directory_body, response=resp, ) if not created_id: return PersonOutcome( status="failed", error="Directory created employee but returned no resolvable user_id", request_body=directory_body, response=resp, ) # Finish the existing Contact projection immediately (not next daily # sweep), including departments and custom attributes. Directory v1 # is the authoritative creation surface for overseas employees and # receives both the work email and mobile above. Do not add Contact # ``enterprise_email`` here: that is a separately managed Feishu Mail # field and the tenant rejects it when the mailbox domain is absent. # A dept the app is not permitted to assign into (41041) also needs no # create-time retry into root "0" any more: this PATCH's generic # dept-slot drop records it as a dropped slot, so the hire is still # created and the un-appliable dept stays visible to IT. projection_body = dict(body) if mobile_dropped_slot is not None: # Feishu just rejected this exact number on the create; sending it # again on the Contact projection only buys a guaranteed 41004 and # a second fallback round-trip. Drop it here so "dropped" on the # ledger is literally true (the number was never accepted on any # surface). A later sweep re-PATCHes the mobile on the now-existing # user, so a number Feishu would accept is not lost forever. projection_body.pop("mobile", None) outcome = await self._patch_user( user_id=created_id, body=projection_body, wanted_externals=list(projection_body.get("department_ids") or []), # Never let the projection turn into a second create: the account # was just built, so "user absent" here is a propagation window, # not a missing person. allow_recreate=False, ) projection_request = outcome.request_body or self._last_write_request projection_response = outcome.response or self._last_write_response outcome.request_body = { "directory_create": directory_body, "contact_projection": projection_request, } outcome.response = { "directory_create": resp, "contact_projection": projection_response, } outcome.detail = {**(outcome.detail or {}), "op": "create", "geo_name": geo_name} if mobile_dropped_slot is not None: # Carry the create-side drop onto whatever the projection returned # (never replace the projection's own slot ledger — #1956 keeps # field-level truth, and a success carrying slot_outcomes is the # deliberate degraded success that must NOT be downgraded). slots = list(outcome.slot_outcomes or []) if not any(s.get("slot") == "mobile" for s in slots): slots.append(mobile_dropped_slot) outcome.slot_outcomes = slots if outcome.status == "success" and not outcome.error: outcome.error = ( "partial: created without mobile (invalid for its country " "on Feishu); fix the HRIS phone / country" ) return outcome async def _read_user_status(self, user_id: str) -> dict[str, Any] | None: """Read a user's directory ``status`` block ({is_frozen, is_resigned, is_exited, is_activated, is_unjoin}) to VERIFY a freeze actually took. Returns None on ANY read error — the caller treats "unverifiable" as a freeze failure (retry next sweep) rather than a false success. Only the numeric-coded ``user-gone`` errors are surfaced as an empty dict so the caller can distinguish "gone → access cut" from "can't tell". """ safe = quote(user_id, safe="") try: data = await self._client.request( "GET", f"/contact/v3/users/{safe}?user_id_type=user_id", ) except AppException as e: if _parse_feishu_code(e) in _USER_ABSENT_CODES: # User no longer exists — access is cut by definition. return {"is_exited": True, "_absent": True} logger.warning( "feishu freeze verify: status read failed for %s: %s", user_id, e, ) return None except Exception: # noqa: BLE001 — unverifiable → caller fails safe logger.exception( "feishu freeze verify: status read raised for %s", user_id, ) return None return ((data or {}).get("user") or {}).get("status") or {} @staticmethod def _access_is_cut(status: dict[str, Any] | None) -> bool: """True iff the directory status PROVES the leaver can no longer log in — frozen (停用), resigned (离职), or exited/gone. ``is_activated`` alone is NOT access-cut.""" if not status: return False return bool( status.get("is_frozen") or status.get("is_resigned") or status.get("is_exited") ) async def deactivate( self, *, person_email: str, prior_external_id: str | None, ) -> PersonOutcome: """Freeze the Feishu user AND verify the freeze actually took. Feishu has NO per-user session-revoke API, so ``is_frozen=true`` is the ONLY lever that cuts a leaver's access — a false success here silently leaves a departed employee logged in. And Feishu can return SUCCESS on ``PATCH is_frozen=true`` while leaving the account ACTIVE (observed 2026-07-24, P01108: the PATCH returned ok, ``is_frozen`` stayed false — a leaver with an in-flight self-submitted approval). So we never trust the PATCH return (or the 230002/42006 idempotent codes) blindly: we re-read the directory status and only claim ``deactivated`` when access is PROVABLY cut. Otherwise ``failed`` — which rides the existing retry + failure-notifier + backstop path so IT is told access is NOT cut, instead of a false green. """ # Reset per-person write observability — same contract as # ``upsert``: a person resolved without a vendor write (absent → # goal met) must not inherit the previous person's captured wire. self._last_write_request = None self._last_write_response = None user_id = prior_external_id if not user_id: try: user_id = await self._lookup_user_id_by_email(person_email) except AppException as e: # Systemic-failure propagation: same circuit-break rule # as upsert() — token / rate-limit / bad-creds errors # halt the WHOLE offboarding sweep with one alert # instead of writing N "freeze lookup failed" state # rows under the same bad credentials. if is_systemic_feishu_failure(e): raise return PersonOutcome( status="failed", error=f"freeze lookup failed: {e.code}: {e.message}", ) if not user_id: # User not in Feishu — goal already met. return PersonOutcome(status="deactivated", external_id=None) safe = quote(user_id, safe="") freeze_error: str | None = None try: # Through the ``_write`` chokepoint — NOT a bare # ``self._client.request`` — so the freeze PATCH captures the # actual wire request/response for the per-person ledger row # exactly like every other user write (#1954/#1964 contract). # A bare call here left every offboarding freeze (incl. the # 42006 path) undiagnosable: the ledger showed a null # body/response. await self._write( "PATCH", f"/contact/v3/users/{safe}?user_id_type=user_id", body={"status": {"is_frozen": True}}, ) except AppException as e: num = _parse_feishu_code(e) if num in _USER_ABSENT_CODES: # User gone entirely — access is cut, nothing left to verify. return PersonOutcome( status="deactivated", external_id=user_id, request_body=self._last_write_request, response=self._last_write_response, ) # Bad creds / rate limit during an offboarding sweep halt # cleanly (one systemic alert, not N per-record rows). if is_systemic_feishu_failure(e): raise # A per-record error OR a "benign" idempotent code (already- # frozen / resigned). Either way DON'T trust it — fall through # to the directory re-read, keeping the error for the ledger. freeze_error = f"PATCH /users/{user_id} (freeze) → {e.code}: {e.message}" # ── Verify the freeze actually took. ───────────────────────────── status = await self._read_user_status(user_id) detail: dict[str, Any] = {"verified_status": status} if freeze_error: detail["freeze_error"] = freeze_error if self._access_is_cut(status): return PersonOutcome( status="deactivated", external_id=user_id, detail=detail, request_body=self._last_write_request, response=self._last_write_response, ) if status is None: err = freeze_error or "freeze verify: could not read back user status" else: err = ( f"freeze did not take (is_frozen={status.get('is_frozen')}, " f"is_resigned={status.get('is_resigned')}, " f"is_exited={status.get('is_exited')})" + (f"; {freeze_error}" if freeze_error else "; PATCH returned success but directory still active") ) return PersonOutcome( status="failed", external_id=user_id, error=err, detail=detail, request_body=self._last_write_request, response=self._last_write_response, ) # ``delete`` intentionally NOT implemented — supports_delete=False # short-circuits the orchestrator before it ever reaches here. def describe(self) -> str: return self._label # ── Read-only directory snapshot (vendor 对端数据) ─────────────────── # Feishu backend / transport errors that are TRANSIENT for the read-only # directory walk — retry with backoff rather than failing the whole daily # snapshot on a single blip across the hundreds of paginated calls # ``fetch_feishu_directory`` makes (BFS over every dept + find_by_department # per dept). A ~500-person tenant is hundreds of round-trips; across that # many calls an occasional server-side 40003 is statistically expected, and # ``_request`` has no retry of its own. # # 40003 — Feishu backend "Internal Error" (server-side # hiccup; the API ``msg`` is literally "Internal # Error"). Observed 2026-07-18 failing # identity_sync.vendor_snapshot_refresh. # 99991671 / 99991672 — tenant-wide rate / frequency limit. A full # directory BFS is exactly the burst that trips # these; the textbook response is back off and # retry, not hard-fail. (Both are ALSO in # feishu_client's systemic set — if the limit # outlasts our bounded retries, the final raise # still surfaces as a systemic failure, unchanged.) # # Scope: this retry wraps ONLY the read walk. The write path (upsert / # freeze / delete) keeps its own worker-level retry and idempotent-code # short-circuits untouched — we don't want a write silently retried here. _FEISHU_DIRECTORY_TRANSIENT_CODES: frozenset[int] = frozenset({ 40003, 99991671, 99991672, }) # 3 retries after the first attempt (4 total), ~0.5/1.5/3.5s. Well under # the cron budget, enough to ride out a transient backend blip or a brief # rate-limit window. _DIRECTORY_RETRY_BACKOFFS: tuple[float, ...] = (0.5, 1.5, 3.5) async def _request_directory_page( client: Any, path: str, ) -> dict[str, Any]: """``client.request('GET', path)`` with bounded retry on transient Feishu backend / transport errors. Non-transient failures (auth, scope, not-found, malformed) re-raise immediately so the walk stays fail-loud for real problems — only server-internal / rate-limit / network blips are absorbed. """ last_exc: Exception | None = None for backoff in (*_DIRECTORY_RETRY_BACKOFFS, None): try: return await client.request("GET", path) except AppException as e: if _parse_feishu_code(e) not in _FEISHU_DIRECTORY_TRANSIENT_CODES: raise last_exc = e except httpx.HTTPError as e: # Connection reset / read timeout mid-walk — transient by nature. last_exc = e if backoff is None: break logger.warning( "feishu directory: transient error on GET %s — retrying in " "%.1fs: %s", path, backoff, last_exc, ) await asyncio.sleep(backoff) assert last_exc is not None # loop only exits here after a caught error raise last_exc async def fetch_feishu_directory( client: Any, ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: """Read-only walk of the live Feishu tenant directory. Returns ``(users, depts)`` — every active department (BFS from root "0", same traversal contract as :meth:`FeishuDepartmentRunner.prime_catalog`) and every user found via ``find_by_department`` on each dept (root included — Feishu parks dept-less users there), deduped by ``user_id``. Raises on a failed page rather than returning a partial roster — the vendor-snapshot consumer treats the result as the complete truth for vendor-only detection, and a silent gap would read as "this person disappeared from the tenant". """ from urllib.parse import quote as _q # Per-collection page bound — a page_token cycle (has_more=true # forever with recycled tokens) is a protocol-level wedge that # would otherwise hang the admin's refresh request indefinitely. # Mirrors feishu_software's paginator guard. max_pages = 200 depts: list[dict[str, Any]] = [] dept_ids: list[str] = ["0"] seen_dept_ods: set[str] = set() visited: set[str] = set() queue: list[str] = ["0"] steps_remaining = 500 while queue and steps_remaining > 0: steps_remaining -= 1 parent = queue.pop(0) if parent in visited: continue visited.add(parent) page_token: str | None = None for _page in range(max_pages): qs = ( f"parent_department_id={_q(parent, safe='')}" "&department_id_type=open_department_id" "&page_size=50" "&fetch_child=false" ) if page_token: qs += f"&page_token={_q(page_token, safe='')}" data = await _request_directory_page( client, f"/contact/v3/departments?{qs}", ) items = (data or {}).get("items") or [] for d in items: if not isinstance(d, dict): continue if (d.get("status") or {}).get("is_deleted"): continue od = d.get("open_department_id") if isinstance(od, str) and od and od not in seen_dept_ods: seen_dept_ods.add(od) depts.append(d) dept_ids.append(od) queue.append(od) if not (data or {}).get("has_more"): break page_token = (data or {}).get("page_token") if not page_token: break else: raise RuntimeError( f"feishu directory: dept paginator under {parent!r} did " f"not terminate within {max_pages} pages — refusing a " "possibly-cyclic walk." ) if queue: # Same fail-loud contract as a failed page: a silently # truncated dept walk would make every user whose only dept is # unvisited read as "disappeared from the tenant". raise RuntimeError( "feishu directory: dept BFS exceeded the 500-step guard " f"with {len(queue)} dept(s) unvisited — refusing a " "truncated snapshot." ) users: list[dict[str, Any]] = [] seen_users: set[str] = set() for od in dept_ids: page_token = None for _page in range(max_pages): qs = ( f"department_id={_q(od, safe='')}" "&department_id_type=open_department_id" "&user_id_type=user_id" "&page_size=50" ) if page_token: qs += f"&page_token={_q(page_token, safe='')}" data = await _request_directory_page( client, f"/contact/v3/users/find_by_department?{qs}", ) for u in (data or {}).get("items") or []: if not isinstance(u, dict): continue uid = str(u.get("user_id") or u.get("open_id") or "") if not uid or uid in seen_users: continue seen_users.add(uid) users.append(u) if not (data or {}).get("has_more"): break page_token = (data or {}).get("page_token") if not page_token: break else: raise RuntimeError( f"feishu directory: user paginator for dept {od!r} did " f"not terminate within {max_pages} pages — refusing a " "possibly-cyclic walk." ) return users, depts