"""Scheduled-job handler for Feishu seat sync. ``feishu.seat_sync`` walks every active SoftwareLicensePlan whose connector binds to a Feishu ``provider='feishu_group'`` integration in a roster-based mode, and mirrors the Feishu side's roster into ``software_seat_allocations``. Roster-based modes are supported (binding's ``operation``): * **Full-tenant roster** — ``tenant_users`` / ``user_suspend`` / ``user_create`` (``_TENANT_ROSTER_OPERATIONS`` in the coordinator). Every active Feishu account in the tenant is one seat. Equivalent in shape to Phase A of ``google.seat_sync`` (mirror "全员账号 = 基础 坐席台账"). The driver walks every accessible department + de-dupes by ``user_id``. All three share this roster because the seat universe is identical — only the offboarding *write* differs; see the read/write-decoupling note below. * ``group_member`` (niche) — only members of one specific Feishu user group are seats. The driver lists that group's ``simplelist`` + reverse-resolves emails via ``users/batch``. Both modes follow the same back-fill / drift contract: * **Import.** Members on the Feishu side whose email matches a Helix person via ``identity_persons.work_email`` or ``email_aliases`` get an ``active`` SoftwareSeatAllocation if they don't already have one. * **Drift cleanup with grace window** (pre-launch hardening 2026-05-05; previously import-only). Helix-active seats whose member is no longer on the Feishu side get their ``drift_observed_at`` stamped on the first run that misses them and are revoked after ``drift_grace_days`` consecutive runs of absence (default 30 days, configurable via the ``feishu.seat_sync.drift_grace_days`` platform setting; **set to 0 to keep the historical pure-import-only behaviour**). The grace window protects against Feishu admins temporarily removing / re-adding accounts for leave / secondment / frozen reasons Helix doesn't model — only sustained absence revokes. Jira / Zoom / Slack syncs revoke immediately on drift; Feishu uses the grace window because its roster mutates more frequently for non- offboarding reasons. Operators may still close Feishu seats explicitly via SeatCancel or admin revoke regardless of grace. * **Partial-roster safety**: the drift cleanup branch calls ``safe_to_drift_revoke`` before mass-revoking, refusing when the matched count falls below 50% of existing active seats. This handles the case where the driver returned a partial roster (some departments out of app scope, some users/batch chunks failed) — the ratio shrinks correspondingly and the safety helper refuses, leaving a defensive-refusal message in ``errors`` (filtered out of ``raise_if_total_failure`` via ``_DEFENSIVE_REFUSAL_FRAGMENTS``). * **Read/write decoupling (2026-06-30)**: ``user_suspend`` / ``user_create`` bindings are seat-synced via the full-tenant roster just like ``tenant_users``. They were previously excluded on the theory that "those bindings are about offboarding, not a roster" — but the roster READ is operation-agnostic and the exclusion silently froze the seat-reconciliation panel for a month after prod's Feishu connector was flipped to ``user_suspend`` for real offboarding (P01452). The snapshot ``product_key`` is normalised to ``tenant_users`` for all three so the panel keeps one stat row per plan. Drift cleanup applies uniformly — a frozen/deleted leaver drops off the tenant roster and the seat is drift-revoked after the grace window, same as before. * **Reconciliation buckets (2026-07-29)**: every roster member is classified into exactly one of ``holders_matched`` (active employee), ``leaver_residual`` (email matches a ``left`` / ``leaver_pending`` person — an offboarding leak), ``external`` (no Helix person) or ``no_email``. The leaver bucket used to be unreachable here (the person lookup excluded leavers, so they fell into ``external`` and the panel's 「离职残留」 was a constant 0 that read as "verified clean"). Import + drift-revoke still consult the active-only subset, so a leaver never gets a seat auto-imported. Idempotent: re-running yields zero new inserts in steady state. Per- plan failures are collected; one bad plan doesn't abort the rest. Per-department / per-batch failures inside the driver's roster walk are isolated and surfaced as informational warnings in ``per_plan_summary[i].departments_failed`` / ``per_plan_summary[i].batches_failed`` — a single bad upstream unit no longer zeroes the whole sync (incident 2026-05-28). """ from __future__ import annotations import logging import uuid from collections.abc import Awaitable from datetime import UTC, datetime from typing import Any from sqlalchemy import select from sqlalchemy.exc import OperationalError from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import format_exception_for_log from app.core.safe_logging import sanitize_log_text from app.core.txn import savepoint from app.scheduled_jobs._seat_sync_helpers import ( ExternalHolderObservation, import_dedup_filter, raise_if_total_failure, record_seat_reconciliation, safe_to_drift_revoke, write_drift_event_audit, write_sync_audit, ) from app.scheduled_jobs.registry import register_scheduled_job logger = logging.getLogger(__name__) # Hard cap on the per-plan ``errors`` array so a pathological plan # (network flake against 100s of departments) can't crowd out other # plans' messages in the result summary. Matches the cap used by # jira_handlers / zoom_handlers. _MAX_ERRORS_PER_PLAN = 30 # Drift-cleanup grace window for Feishu seats. Feishu's tenant # roster is administered by humans (HR / IT) who may temporarily # remove and re-add accounts (leave, secondment, frozen) for reasons # Helix doesn't model. Auto-revoking immediately would close real # seats by mistake; auto-revoking after 30 days catches genuine # drift while honouring the cooling-off window. Configurable via # ``feishu.seat_sync.drift_grace_days`` platform setting (set to 0 # to keep historical "import-only" behaviour). _DEFAULT_FEISHU_DRIFT_GRACE_DAYS = 30 # Helix-side employment states that mean "this person is gone (or on # their way out)". A vendor account still held by one of them is an # offboarding leak → the ``leaver_residual`` reconciliation bucket, and # is never eligible for seat import / re-import. Same pair every other # domain inlines; named here so the classification branch and the # import-eligibility filter can never drift to different tuples. _LEAVER_EMPLOYMENT_STATUSES: frozenset[str] = frozenset( {"left", "leaver_pending"}, ) @register_scheduled_job( key="feishu.seat_sync", display_name="飞书 · 坐席同步", description=( "每天把飞书租户/用户组的成员名单镜像到 software_seat_allocations。" "支持两种 binding 模式 — tenant_users(全员=坐席,走 " "/contact/v3/departments + /contact/v3/users/find_by_department)" "和 group_member(用户组,走 /contact/v3/group/{id}/member/simplelist " "+ /contact/v3/users/batch)。**导入 + 漂移回收**:飞书侧消失的成员" "对应的 Helix 坐席在满 drift_grace_days(默认 30 天,平台设置 " "feishu.seat_sync.drift_grace_days 可调;设 0 退回纯导入不回收)后" "自动 revoke,期间 30 天冷却以容忍请假/借调/冻结等飞书侧临时移除。" "需要 contact:user.base:" "readonly + contact:user.email:readonly + contact:user.employee_id:" "readonly + contact:department.base:readonly + 通讯录范围 = 全部" "成员;group_member 模式额外需要 contact:group.member:readonly。" ), ) async def run_feishu_seat_sync( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: # Deadlock-victim restart on fresh sessions — same incident class # as jira.seat_sync 2026-06-12 (worker restart double-run racing # INSERTs). The reconcile is idempotent: import has the 25h-grace # dedup filter, revoke has safe_to_drift_revoke + grace window. from app.scheduled_jobs.handlers import run_with_lock_retry return await run_with_lock_retry( db, lambda session: _run_feishu_seat_sync_once(session, config), label="feishu.seat_sync", ) async def _run_feishu_seat_sync_once( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: from app.core.coordinators.feishu_seat_coordinator import ( _TENANT_ROSTER_OPERATIONS, get_feishu_misconfigured_seat_sync_plans, get_feishu_seat_sync_plans, load_feishu_software_driver, ) from app.domains.asset.models import ( SoftwareExternalHolder, SoftwareSeatAllocation, SoftwareSeatSyncStat, ) from app.identity.email_lookup import build_person_by_email_index from app.identity.leaver_notice import is_within_leaver_notice_period driver = await load_feishu_software_driver(db) if driver is None: # Round-42 fix: returning a dict with non-empty ``errors`` leaves # ``scheduled_job_runs.status='ok'`` (only RAISED exceptions land # ``failed``) so the dedup-aware failure-notifier stays silent # and admins see the job perpetually "green" while no seats # sync. Per ``feedback_handler_status_must_raise.md``, raise. raise RuntimeError( "Feishu software-provisioning integration " "(provider='feishu_group') not configured or not enabled — " "seat sync cannot proceed.", ) plans = await get_feishu_seat_sync_plans(db) # Misconfigured plans are surfaced even when the eligible set is # empty — a Feishu integration with ONLY malformed bindings would # otherwise look like "feature disabled" instead of "config bug". # (Round-55 visibility fix paired with the driver's strict # ``_resolve_operation``; previously the coordinator silently # dropped these rows and ``scheduled_job_runs.status`` stayed green # forever while no real syncing happened.) misconfigured = await get_feishu_misconfigured_seat_sync_plans(db) misconfigured_errors = [ ( f"{plan.name} (plan {plan.id}): misconfigured Feishu seat-sync " f"binding — {reason}. Fix the row in 软件目录 → 飞书 → 连接器配置." ) for plan, reason in misconfigured ] if not plans: # Zero eligible plans has TWO very different meanings and we must # not collapse them to one silent "ok": # # (a) No active connector wired at all → genuine "feature # disabled" posture (tenant uses Feishu for chat-approval # only). Return cleanly so the run reports OK; admins # discover the absence via the plan-management page. # # (b) An active connector to the enabled feishu_group # integration DOES exist, yet it yields NEITHER an eligible # plan NOR a misconfigured one — meaning the connector's # plan or software went inactive (the eligible + misconfig # queries both join on status='active'). This is the # genuinely-INVISIBLE silent-freeze trap: no error is # emitted at all, the cron stays green, and the panel sits # frozen at the last snapshot. Raise so the run flips to # status='failed' and the dedup failure-notifier pages ops # (per feedback_handler_status_must_raise). # # Misconfigured plans (case below) are DELIBERATELY non-raising # — they already surface in summary['errors'] and are whitelisted # so a static config bug doesn't page every tick (ADR 2026-05-30); # the panel staleness flag is their safety net. # # (After the 2026-06-30 read/write decoupling, user_suspend / # user_create bindings ARE seat-syncable, so a normally-wired # Feishu connector lands in ``plans`` and never reaches here.) from app.domains.provisioning.models import SoftwareConnector from app.integrations.models import Integration has_active_connector = ( await db.execute( select(SoftwareConnector.id) .join( Integration, Integration.id == SoftwareConnector.integration_id, ) .where( SoftwareConnector.is_active.is_(True), Integration.kind == "software_provisioning", Integration.provider == "feishu_group", Integration.enabled.is_(True), ) .limit(1) ) ).first() is not None if has_active_connector and not misconfigured_errors: raise RuntimeError( "Feishu seat sync produced 0 eligible plans while an " "active feishu_group connector is configured — its plan " "or software is likely inactive, so the " "seat-reconciliation panel would silently freeze at the " "last snapshot. Check 软件目录 → 飞书 → 连接器配置 " "(plan + software status).", ) return { "plans_synced": 0, "seats_imported": 0, "errors": [ *misconfigured_errors, "No active Feishu seat-syncable license plans found. " "Create a software_connectors row that points at the " "Feishu integration with binding_config " "{operation: 'tenant_users'} or " "{operation: 'group_member', group_chat_id: '...'}.", ], } total_imported = 0 total_revoked = 0 errors: list[str] = list(misconfigured_errors) per_plan_summary: list[dict[str, Any]] = [ { "plan": plan.name, "operation": None, "skipped": "misconfigured", "reason": reason, } for plan, reason in misconfigured ] now = datetime.now(UTC) # Resolve the configurable grace window once for the whole run. # 0 = keep import-only behaviour; >0 = revoke seats whose vendor # presence has been gone for at least N days. The first time a # seat is "missing", we don't revoke — only drift_observed_at # gets stamped. Subsequent runs that find the same seat still # missing AND drift_observed_at older than grace days revoke it. from app.integrations.service import get_platform_setting raw_grace = await get_platform_setting( "feishu.seat_sync.drift_grace_days", default=_DEFAULT_FEISHU_DRIFT_GRACE_DAYS, session=db, ) try: grace_days = max(int(raw_grace), 0) except (TypeError, ValueError): grace_days = _DEFAULT_FEISHU_DRIFT_GRACE_DAYS # When the operator disables the drift cleanup (grace_days=0), # CLEAR every existing drift_observed_at stamp on this handler's # plans. Otherwise a later re-enable would immediately revoke # any seat that had accumulated a stale stamp during the # disabled window — that would violate the ADR-promised # "flipping back doesn't revoke a backlog" invariant. # # Pre-launch hardening 2026-05-05 (P2 from LS2): the previous # shape ran an unscoped UPDATE that wiped drift_observed_at # across ALL plans. Today only Feishu writes that column so the # bug was latent, but the moment a future vendor handler adopts # the same grace-window pattern, the unscoped wipe would silently # nuke the other vendor's drift state. The UPDATE is now scoped # to ``plan_id IN (this handler's plans)``. if grace_days == 0: from sqlalchemy import update as _sql_update from app.domains.asset.models import ( SoftwareSeatAllocation as _SeatAlloc, ) feishu_plan_ids = [p.id for p, _b in plans] if feishu_plan_ids: await db.execute( _sql_update(_SeatAlloc) .where( _SeatAlloc.plan_id.in_(feishu_plan_ids), _SeatAlloc.drift_observed_at.is_not(None), _SeatAlloc.status == "active", ) .values(drift_observed_at=None), ) for plan, binding in plans: plan_name = plan.name # No silent default: a missing ``operation`` would historically # collapse to ``tenant_users`` and silently mirror the entire # Feishu tenant into seat allocations on this plan. That broke # the driver's fail-loud contract (round-55). Empty string # now falls through to the ``unsupported_operation`` else # branch below and is reported via ``errors`` + per-plan summary. operation = str(binding.get("operation") or "").strip() # ── Step 1: fetch the Feishu-side roster for this plan ─ members: list[dict[str, Any]] roster_label: str # used in audit reason + per_plan summary # Stable reconciliation key independent of the lifecycle # operation: every full-tenant-roster binding (tenant_users / # user_suspend / user_create) writes the SAME ``tenant_users`` # snapshot product_key so the panel stays on one stat row per # plan regardless of which offboarding mode the binding uses # (read/write decoupling, 2026-06-30). group_member keeps its # own product_key. roster_product_key: str # tenant_users mode also returns a per-dept diagnostics envelope # so the per_plan summary can surface "X/Y departments out of # app scope" — the actionable signal when a Feishu admin # narrows 通讯录范围 mid-quarter (incident 2026-05-28). # group_member mode returns a symmetric per-batch diagnostics # envelope for the same reason (a single bad users/batch chunk # would historically zero a group sync). tenant_departments_attempted = 0 tenant_departments_failed = 0 tenant_failure_examples: list[str] = [] # Deliberate roster exclusions (frozen / resigned accounts), so # ``members_listed`` can be reconciled against the Feishu admin # console without a manual API round (2026-07-29). ``None`` for # group_member mode — that walk has no status filter to report. tenant_roster_total: int | None = None tenant_skipped_frozen: int | None = None tenant_skipped_resigned: int | None = None group_batches_attempted = 0 group_batches_failed = 0 group_failure_examples: list[str] = [] try: if operation in _TENANT_ROSTER_OPERATIONS: walk = await driver.list_all_tenant_users_with_emails() members = walk["users"] tenant_departments_attempted = int(walk.get("departments_attempted") or 0) tenant_departments_failed = int(walk.get("departments_failed") or 0) tenant_failure_examples = list(walk.get("failure_examples") or []) tenant_roster_total = int(walk.get("roster_total") or 0) tenant_skipped_frozen = int(walk.get("skipped_frozen") or 0) tenant_skipped_resigned = int(walk.get("skipped_resigned") or 0) roster_label = "Feishu tenant (active accounts)" roster_product_key = "tenant_users" elif operation == "group_member": group_chat_id = str(binding.get("group_chat_id") or "").strip() if not group_chat_id: # Coordinator already excludes empty group_chat_id; # this is defensive in case the predicate loosens # in a future refactor. errors.append( f"{plan_name}: group_member binding missing group_chat_id; skipped", ) per_plan_summary.append( { "plan": plan_name, "operation": operation, "skipped": "missing_group_chat_id", } ) continue group_walk = await driver.list_group_members_with_emails( group_chat_id, ) members = group_walk["users"] group_batches_attempted = int(group_walk.get("batches_attempted") or 0) group_batches_failed = int(group_walk.get("batches_failed") or 0) group_failure_examples = list(group_walk.get("failure_examples") or []) roster_label = f"Feishu user group {group_chat_id}" roster_product_key = "group_member" else: # Coordinator filters this out, but if a future change # admits more modes the handler shouldn't silently # mirror nothing — surface it. errors.append( f"{plan_name}: unsupported operation {operation!r}; skipped", ) per_plan_summary.append( { "plan": plan_name, "operation": operation, "skipped": "unsupported_operation", } ) continue except Exception as e: # noqa: BLE001 — per-plan isolation errors.append( f"{plan_name} (operation={operation}): list failed: {format_exception_for_log(e)}", ) per_plan_summary.append( { "plan": plan_name, "operation": operation, "error": format_exception_for_log(e), } ) continue # ── Step 2: build email → person lookup ─ members_by_email: dict[str, dict[str, Any]] = {} members_without_email = 0 # Feishu accounts with NO email at all (typical: hand-created # consultant/外部顾问 accounts). They can't match a person and # can't key the email-based registry, so they're persisted as # ``category='no_email'`` holder rows keyed on the Feishu # user_id — otherwise they'd exist only as an anonymous counter # and the operator could never see WHO holds those seats. no_email_members: list[dict[str, Any]] = [] for m in members: email = (m.get("email") or "").strip().lower() if not email: members_without_email += 1 if str(m.get("user_id") or "").strip(): no_email_members.append(m) continue members_by_email[email] = m # Two indexes off ONE query (2026-07-29): # # * ``person_index_all`` — every non-deleted person, leavers # included. Drives CLASSIFICATION only: a vendor account whose # email matches a ``left`` / ``leaver_pending`` person is an # offboarding leak (``leaver_residual``), not an anonymous # external account. Before this, Feishu passed # ``exclude_employment_statuses`` here and therefore never # emitted the ``leaver_residual`` bucket at all — leavers whose # Feishu account outlived them were silently filed under # 「外部账号」 while the panel's 「离职残留」 read a constant 0. # That constant 0 is worse than a wrong number: it reads as # "verified clean" on a vendor where Helix was not even looking # (jira / jamf DO emit the bucket, so the panels disagreed on # what the same column meant). # # * ``person_index`` — active subset, byte-identical to the old # behaviour. Still the ONLY index the import + drift-revoke # loops below consult, so a leaver never gets a fresh seat # auto-imported and their existing seat still falls through to # drift cleanup via "user_id not in members_present_user_ids". person_index_all = await build_person_by_email_index( db, members_by_email.keys(), ) # Notice-period exception (2026-08-11): a ``leaver_pending`` # person whose last working day has not passed is still # working and keeps their seat — excluding them drift-revoked # the allocation row weeks early and, once the person # re-matched, re-imported it as a NEW row whose id defeated # offboarding's per-seat revoke-task dedup (duplicate cancel # events). Same invariant as ``leaver_notice_grace`` in # ``build_person_by_email_index``. person_index = { email: p for email, p in person_index_all.items() if p.employment_status not in _LEAVER_EMPLOYMENT_STATUSES or is_within_leaver_notice_period(p) } # ── Step 3: load existing seats in the (plan, user) unique- # index scope. Active OR pending OR recently-revoked — see # zoom_handlers comment. Avoids IntegrityError under migration # 119's partial unique index AND prevents the import branch # from racing a recent admin-revoke / leaver_blocked flip # (round 12 hardening 2026-05-08). existing_all = list( ( await db.execute( select(SoftwareSeatAllocation).where( SoftwareSeatAllocation.plan_id == plan.id, import_dedup_filter(SoftwareSeatAllocation, now=now), ) ) ) .scalars() .all() ) existing_user_ids: set = {s.user_id for s in existing_all} # ── Step 4: import new members ─ seats_imported = 0 external_unmatched = 0 # Per-email matched counter. Mirrors the pattern in jira/slack # handlers — feeds the snapshot's ``holders_matched``. The # ``members_present_user_ids`` set below stays as the per-PERSON # set used by drift-revoke (one revoke per person, not per email). matched_email_count = 0 # Same per-email counting rule for the leaver bucket — it feeds # the snapshot's ``holders_leaver_residual`` and the recorder's # identity check compares it against the number of # ``category='leaver_residual'`` observations. leaver_residual_count = 0 external_observations: list[ExternalHolderObservation] = [] for _nm in no_email_members: _uid = str(_nm.get("user_id") or "").strip() external_observations.append( ExternalHolderObservation( email="", vendor_account_id=_uid, account_status="active", display_name=_nm.get("name") or None, category="no_email", # No email to key on — the Feishu user_id is the only # stable identity. Prefixed so the key is self-describing # in the DB and in the external-accounts registry. holder_key=f"feishu-user:{_uid}", ) ) per_plan_errors = 0 truncated = False for email in members_by_email: person = person_index.get(email) if person is None: _m = members_by_email[email] leaver = person_index_all.get(email) if leaver is not None: # Matched a Helix person who is already gone (or # mid-offboarding) yet still holds a live Feishu # account — an offboarding leak, not an external # account. ``person_id`` lets the drawer deep-link # to the offboarding view. leaver_residual_count += 1 external_observations.append( ExternalHolderObservation( email=email, vendor_account_id=_m.get("user_id"), account_status="active", display_name=(leaver.display_name or _m.get("name") or None), category="leaver_residual", person_id=leaver.id, ), ) continue # Feishu account whose email doesn't match any Helix # person — shared mailbox, contractor, HRIS-absent. external_unmatched += 1 external_observations.append( ExternalHolderObservation( email=email, vendor_account_id=_m.get("user_id"), account_status="active", display_name=_m.get("name") or None, ) ) continue matched_email_count += 1 if person.id in existing_user_ids: continue try: # Per-row SAVEPOINT — one bad INSERT (FK / unique-index # collision) rolls back just this row, not the session. async with db.begin_nested(): db.add( SoftwareSeatAllocation( plan_id=plan.id, user_id=person.id, status="active", allocated_at=now, reason=(f"auto-imported from {roster_label} via feishu.seat_sync"), ) ) await db.flush() seats_imported += 1 # Mark in-memory so a duplicate email same plan run # (shouldn't happen — dict is keyed by email — but # defensive against future caller changes) doesn't # double-INSERT. existing_user_ids.add(person.id) except OperationalError: # Deadlock (1213) killed the whole transaction — # propagate so the fresh-session lock-retry wrapper # restarts the run. raise except Exception as e: # noqa: BLE001 per_plan_errors += 1 errors.append( f"{plan_name}/{email}: insert failed: {e}", ) if per_plan_errors >= _MAX_ERRORS_PER_PLAN: errors.append( f"{plan_name}: insert errors truncated at {_MAX_ERRORS_PER_PLAN}", ) truncated = True break # ── Step 5: drift cleanup with grace window ─ # Pre-launch hardening 2026-05-05: previously the Feishu sync # was strictly import-only ("Helix-active seats whose member # is no longer on the Feishu side are left alone") which # overcharged for offboarded seats indefinitely. Now we record # the first observation of a missing seat in # ``drift_observed_at`` and revoke after ``drift_grace_days`` # (default 30, configurable via platform_setting). Set the # setting to 0 to disable revoke and keep historical behaviour. seats_revoked_this_plan = 0 members_present_user_ids: set = { person.id for email, m in members_by_email.items() if (person := person_index.get(email)) is not None } # Re-load existing seats since we just inserted some — those # newly-imported rows are already in members_present_user_ids # so they'll fall through cleanly without being marked drifted. all_seats_now = list( ( await db.execute( select(SoftwareSeatAllocation).where( SoftwareSeatAllocation.plan_id == plan.id, SoftwareSeatAllocation.status == "active", ) ) ) .scalars() .all() ) # Safety guard via shared helper — refuses cleanup when match # ratio is below 50%. Belt-and-braces against transient # Feishu API outages where the tenant walk returned a small # subset. safe_drift, refuse_reason = safe_to_drift_revoke( # matched_count = total API roster size (all Feishu users # returned by the tenant walk, regardless of Helix match). # want_count = Helix-matched subset for this plan. Passing # the same value for both was a bug: when the API returns # users but none match Helix persons (all contractors), the # empty-vendor guard (#1) would fire with a misleading # "0 matched users" message instead of the correct # empty-want guard (#3) that names the predicate-feed failure. matched_count=len(members_by_email), existing_count=len(all_seats_now), label=f"Feishu seat sync ({plan_name})", want_count=len(members_present_user_ids), ) # Round-2 adversarial-review fix (2026-05-28): when the driver # walk produced a partial roster (some departments out of app # scope, some users/batch chunks failed), we DO NOT KNOW which # users we missed. Their existing Helix seats would land in # `not in members_present_user_ids` → drift_observed_at stamped # → after grace_days of sustained partial failure → auto-revoked # for users who never actually left the tenant. The 50% ratio # guard does NOT save us: losing 1 dept out of 150 (~10 users # out of 477) still passes >97% ratio. Conservatively refuse # drift cleanup entirely on ANY upstream walk failure — the # next clean run will resume drift mechanics for genuinely- # absent users. Surfaced as a defensive refusal (filtered out # of raise_if_total_failure). partial_roster = tenant_departments_failed > 0 or group_batches_failed > 0 if grace_days > 0 and partial_roster: errors.append( f"{plan_name}: refusing drift cleanup — driver returned " f"a partial roster (" f"{tenant_departments_failed} dept(s) and " f"{group_batches_failed} batch(es) failed); cannot " f"safely tell genuine absence from API gap. Next clean " f"run will resume drift mechanics." ) elif grace_days > 0 and not safe_drift: errors.append(refuse_reason or "drift cleanup refused") elif grace_days > 0: from datetime import timedelta cutoff = now - timedelta(days=grace_days) for seat in all_seats_now: if seat.user_id in members_present_user_ids: # Vendor still has them — clear any prior drift # observation so a temporary flicker doesn't # accumulate toward revoke. if seat.drift_observed_at is not None: seat.drift_observed_at = None continue # Skip fresh seats (in-flight grant race window). allocated = seat.allocated_at if allocated is not None: if allocated.tzinfo is None: allocated = allocated.replace(tzinfo=UTC) if (now - allocated).total_seconds() < 7200: continue if seat.drift_observed_at is None: # First time we've noticed it missing — stamp and # wait for the grace window before revoking. seat.drift_observed_at = now continue observed = seat.drift_observed_at if observed.tzinfo is None: observed = observed.replace(tzinfo=UTC) if observed <= cutoff: # rollback expires modified instances; capture first seat_id, seat_user_id, seat_plan_id = ( seat.id, seat.user_id, seat.plan_id, ) try: # Per-seat SAVEPOINT — one bad flush must not # poison the session and kill the rest of the run. async with db.begin_nested(): seat.status = "revoked" seat.revoked_at = now seat.reason = ( (seat.reason or "") + f"\n[auto-revoked by feishu.seat_sync: " f"user absent from {roster_label} for " f"≥{grace_days} days]" ).strip() await write_drift_event_audit( db, job_key="feishu.seat_sync", seat_id=seat_id, user_id=seat_user_id, plan_id=seat_plan_id, event="revoked", detail={ "reason": ( f"absent from {roster_label} for ≥{grace_days} days" ), }, ) await db.flush() seats_revoked_this_plan += 1 except OperationalError: # Whole-transaction rollback (deadlock) — see # the import loop above. raise except Exception as e: # noqa: BLE001 errors.append( f"{plan_name}/seat {seat_id}: drift revoke " f"failed: {format_exception_for_log(e)}", ) total_imported += seats_imported total_revoked += seats_revoked_this_plan plan_summary: dict[str, Any] = { "plan": plan_name, "operation": operation, "members_listed": len(members), "members_with_email": len(members_by_email), "members_without_email": members_without_email, "external_unmatched": external_unmatched, "leaver_residual": leaver_residual_count, "seats_imported": seats_imported, "seats_revoked": seats_revoked_this_plan, "drift_grace_days": grace_days, } if tenant_roster_total is not None: # Roster arithmetic, always emitted (not only on anomaly) so # "后台 446 vs 同步 445" is answerable from the run result # alone — see the driver docstring for the console mapping. plan_summary["roster_total"] = tenant_roster_total plan_summary["skipped_frozen"] = tenant_skipped_frozen plan_summary["skipped_resigned"] = tenant_skipped_resigned # The number the Feishu admin console shows as 总人数 # (active + frozen; 已离职 sits on its own tab), so an # operator can compare like-for-like without doing the # subtraction by hand. plan_summary["feishu_console_total_equivalent"] = len(members) + ( tenant_skipped_frozen or 0 ) if operation == "group_member": plan_summary["group_chat_id"] = str(binding.get("group_chat_id") or "").strip() if operation in _TENANT_ROSTER_OPERATIONS and tenant_departments_failed > 0: # Surface the scope-drift signal in the per-plan summary # AND the top-level errors list. Admins watching the # scheduled-jobs page see the failed dept count and the # remediation hint without combing worker logs. Skipped # departments don't zero the run (only TOTAL failure does # — see driver), so the run lands ``status='ok'`` with the # warning visible. plan_summary["departments_attempted"] = tenant_departments_attempted plan_summary["departments_failed"] = tenant_departments_failed if tenant_failure_examples: plan_summary["departments_failed_examples"] = tenant_failure_examples errors.append( f"{plan_name}: {tenant_departments_failed}/" f"{tenant_departments_attempted} departments unreachable " f"(likely outside app 通讯录范围 — fix on Feishu admin " f"side). First: " f"{tenant_failure_examples[0] if tenant_failure_examples else 'n/a'}" ) if operation == "group_member" and group_batches_failed > 0: # Symmetric partial-failure surfacing for group_member # mode. Same defensive-fragment whitelist ("batches # unreachable" / "departments unreachable" both match # "unreachable" — see _DEFENSIVE_REFUSAL_FRAGMENTS). plan_summary["batches_attempted"] = group_batches_attempted plan_summary["batches_failed"] = group_batches_failed if group_failure_examples: plan_summary["batches_failed_examples"] = group_failure_examples errors.append( f"{plan_name}: {group_batches_failed}/" f"{group_batches_attempted} batches unreachable " f"(likely transient Feishu API error — re-run on next " f"tick). First: " f"{group_failure_examples[0] if group_failure_examples else 'n/a'}" ) if truncated: plan_summary["insert_errors_truncated"] = True per_plan_summary.append(plan_summary) # Round-2 (2026-05-28): skip reconciliation snapshot writes on # a partial roster. `record_seat_reconciliation` flips every # external holder NOT in this run's observed set to # resolved=True. With a partial roster, the failed-dept / # failed-batch holders are absent NOT because they left the # vendor but because we couldn't query them — flipping them to # resolved would lose observation history (first_observed_at # resets on the next clean run) AND understate the # `holders_external` snapshot. Same root cause as the drift- # cleanup skip above. The next clean run re-establishes the # full reconciliation state; until then, the previous snapshot # row remains visible in the admin UI (correctly). if partial_roster: logger.warning( "feishu.seat_sync plan=%s: skipping reconciliation " "snapshot — partial roster (%d dept(s), %d batch(es) " "failed); previous snapshot preserved.", plan_name, tenant_departments_failed, group_batches_failed, ) else: await record_seat_reconciliation( db, holder_model=SoftwareExternalHolder, stat_model=SoftwareSeatSyncStat, integration_id=uuid.UUID(driver.integration_id), software_id=plan.software_id, plan_id=plan.id, provider="feishu", product_key=roster_product_key, site=( str(binding.get("group_chat_id") or "").strip() or None if operation == "group_member" else None ), # Post-email-dedupe billable identity count and per-email # matched count (NOT per-person via members_present_user_ids) # so the reconciliation identity holds even when two Feishu # accounts share an email (rare) or two emails alias to one # Helix person — see jira_handlers.py for the off-by-N # rationale this pattern catches. vendor_license_holders=(len(members_by_email) + members_without_email), holders_matched=matched_email_count, holders_no_email=members_without_email, holders_leaver_residual=leaver_residual_count, externals=external_observations, now=now, ) summary = { "plans_synced": len(plans), "seats_imported": total_imported, "seats_revoked": total_revoked, "drift_grace_days": grace_days, "per_plan": per_plan_summary, "errors": errors, } await write_sync_audit(db, job_key="feishu.seat_sync", summary=summary) await db.commit() raise_if_total_failure( job_key="feishu.seat_sync", summary=summary, progress_keys=("seats_imported", "seats_revoked"), ) return summary # Feishu terminal status codes that should trigger a replay of the # missed callback. Approved + Cancelled both go through the same # idempotent dispatcher — the per-domain handler decides what # "approved" vs "cancelled" means for its own child row. Module-level # so ruff's N806 (uppercase-in-function) stays happy and the constant # is reusable from tests / reaper-extension call sites. _TERMINAL_FEISHU_STATUSES_FOR_REPLAY = frozenset( { "APPROVED", "COMPLETED", "PASS", "CANCELED", "CANCELLED", "REJECTED", "TIMEOUT", }, ) @register_scheduled_job( key="feishu.approval_pending_reaper", display_name="Feishu approval pending 回收 + 漂移对账", description=( "每日 04:00 UTC 扫描所有 status='pending' 且超过阈值天数的" " ApprovalRequest, 重新请求 Feishu 实例状态。对每行:" " 若 Feishu 已 APPROVED/COMPLETED → replay missed callback," " 触发 auto-execute 副作用并把 Helix 行翻成 approved/completed;" " 若 Feishu 已 CANCELED/REJECTED/TIMEOUT → 同样 replay (per-domain" " 服务负责拒绝侧的清理);若仍 PENDING 且超过 hard_expire_after_days" " → 强制 expired 并通知申请人。解决 webhook 丢失/审批人离职导致的" " pending 永远卡住。默认 expire_after_days=14,hard_expire_after_days=30。" ), ) async def run_approval_pending_reaper( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Reap stale pending Feishu approval rows + reconcile missed webhooks. Scans `approval_requests` for pending instances older than `expire_after_days`. For each, fetches the Feishu instance state and: * Helix already-pending + Feishu APPROVED / COMPLETED / CANCELLED / REJECTED / TIMEOUT → replay the missed callback via ``ApprovalService.replay_callback_from_feishu`` so the per-domain dispatcher runs (idempotent — terminal-status guard short-circuits if a concurrent webhook beat us). This covers the "Feishu webhook dropped" case that previously left rows stuck pending forever (the original 2026-05-04 reaper had a "leave alone, callback should land" branch for COMPLETED — fixed 2026-05-28). * Helix already-pending + Feishu still PENDING and age beyond ``hard_expire_after_days`` (default 30) → flip Helix to ``expired`` regardless and notify applicant; admin must re-submit if still wanted. Failure on a single row never aborts the batch. """ from datetime import timedelta from app.approval.models import ApprovalRequest from app.approval.service import ApprovalService from app.auth.tokens.service import _as_aware_utc soft_days = int(config.get("expire_after_days") or 14) hard_days = int(config.get("hard_expire_after_days") or 30) soft_cutoff = datetime.now(UTC) - timedelta(days=soft_days) hard_cutoff = datetime.now(UTC) - timedelta(days=hard_days) rows = ( ( await db.execute( select(ApprovalRequest) .where( ApprovalRequest.status == "pending", ApprovalRequest.created_at < soft_cutoff, ) .limit(500) ) ) .scalars() .all() ) if not rows: return {"checked": 0, "expired": 0, "fetch_errors": 0} service = ApprovalService(db) expired = 0 fetch_errors = 0 async def _record_terminal( req: Any, new_status: str, *, reason: str, ) -> None: """Audit + notify the applicant when the reaper flips a request to a terminal state. Without this, applicants would never learn their request was auto-expired (the Feishu card is gone, the Helix UI just shows ``expired``) — leaving a silently-cancelled request the user might be waiting on. """ from app.audit.service import AuditService try: # SAVEPOINT-isolate the audit flush: this helper runs inside a # per-row loop on one shared session, and a swallowed flush # failure here would poison the session and discard the whole # day's batch at the outer ``db.commit()``. begin_nested rolls # back just this write on failure, keeping the session usable. async with db.begin_nested(): await AuditService(db).log( trace_id=f"feishu.approval_pending_reaper:{req.id}", actor_type="system", actor_user_id=None, action=f"approval.reaper_{new_status}", status="success", resource_type="approval_request", resource_id=str(req.id), detail={ "external_id": req.external_id, "new_status": new_status, "reason": reason, # ``req.created_at`` reads back naive from TiDB/MySQL # even though the column is declared # ``DateTime(timezone=True)`` — the driver strips the # tz. Comparing against ``datetime.now(UTC)`` (aware) # raises ``TypeError`` and poisons the audit log call. # Coerce to aware-UTC via the shared helper; same # pattern as auth token service and identity_sync. # Production incident 2026-05-16. "age_days": (datetime.now(UTC) - _as_aware_utc(req.created_at)).days if req.created_at else None, }, ) except OperationalError: raise except Exception: # noqa: BLE001 logger.exception("reaper audit failed for %s", req.id) # Notify the applicant. Best-effort — failure here must not # roll back the status flip. try: from app.identity.person_models import IdentityPerson from app.notifications.service import ( NotificationEnvelope, NotificationService, ) # Direct attribute access — ``ApprovalRequest`` only has # ``applicant_id``; the previous ``or getattr(req, # "applicant_person_id", None)`` was dead code that hides # the real column name behind a silent-None fallback. Bug # class memo'd in CLAUDE.md (feedback_no_getattr_default_ # for_model_columns). applicant_id = req.applicant_id if applicant_id: applicant = await db.get(IdentityPerson, applicant_id) if applicant is not None: notif = NotificationService(db) await notif.dispatch( NotificationEnvelope( recipient=applicant, title=("[Helix] 您的审批请求已自动关闭"), body_text=( f"您于 {req.created_at:%Y-%m-%d} 提交的审批请求" f"已被系统自动标记为 {new_status}。\n\n" f"原因: {reason}\n\n" "如仍需此审批,请重新提交。" ), channels=["feishu", "email"], template_key="approval.reaper_terminal", url="/my/requests", context={ "new_status": new_status, "reason": reason, "request_id": str(req.id), # The template renders "您于 {submitted_at} # 提交的…" — the date has to live on the # context, not be baked into the title # above, or the bell re-render (which only # has the frozen context) shows a literal # ``{submitted_at}``. "submitted_at": ( f"{req.created_at:%Y-%m-%d}" if req.created_at else "-" ), }, related_resource_type="approval_request", related_resource_id=req.id, ), trace_id=f"feishu.approval_pending_reaper:{req.id}", ) except Exception: # noqa: BLE001 logger.exception( "reaper notify failed for %s; status still flipped to %s", req.id, new_status, ) replayed = 0 replay_failures = 0 async def _replay(req: Any, instance: dict[str, Any]) -> None: """Dispatch the missed callback through the idempotent replay path. Counts successes / no-ops / failures into the outer counters so the cron summary distinguishes "Feishu unreachable" from "domain handler crashed during replay." ``instance`` is passed explicitly (rather than captured as a free variable from the for-loop) so the inner function documents its dependencies and a future refactor that moves the call site won't silently shadow it. """ nonlocal replayed, replay_failures try: # SAVEPOINT-isolate the replay: replay_callback_from_feishu runs # per-domain side-effects (asset return / seat activation / scope # grant / DNS write) and flushes on this shared per-row-loop # session. Without the savepoint, one IntegrityError on replay # poisons the session and the outer ``db.commit()`` discards the # entire batch of expiries + replays. begin_nested rolls back just # this row's replay. A deadlock (OperationalError) propagates out # of the reaper: this handler is NOT wrapped in run_with_lock_retry, # so the run is marked ``failed`` and retried on the next daily tick # (the reaper is idempotent — terminal rows short-circuit, expiry is # a re-scan), rather than restarted in a fresh session. async with db.begin_nested(): result = await service.replay_callback_from_feishu( instance_code=req.external_id, actor_user_id=None, prefetched_instance_detail=instance, trace_id_prefix="feishu.approval_pending_reaper", ) if result.get("replayed"): replayed += 1 else: # Concurrent webhook landed between fetch and replay → # row is already terminal; count as a clean no-op, not # a failure. logger.debug( "reaper replay no-op for %s: %s", req.external_id, result.get("reason"), ) except OperationalError: raise except Exception: # noqa: BLE001 replay_failures += 1 logger.exception( "reaper: replay failed for %s; leaving row pending", req.external_id, ) # PER-ROW COMMIT (login-brownout fix, #1837/#1940 class): this reaper # fetches Feishu state (HTTP) and dispatches applicant notifications # (outbox INSERT → FK shared-lock on identity_persons) for up to 500 # rows. A single trailing commit held that fan-out of FK locks across # the whole batch of multi-second external calls — during which any of # those applicants' SSO login (which writes their identity FK children) # would 1205 lock-wait → login brown-out. Commit after each row so no # lock spans the next row's Feishu fetch. Committing expires loaded # instances, so capture ids first and re-fetch each row fresh (also # re-checks it's still pending → skips rows a concurrent webhook already # handled, avoiding double-processing). req_ids = [r.id for r in rows] for req_id in req_ids: req = await db.get(ApprovalRequest, req_id) if req is None or req.status != "pending": continue try: instance = await service.fetch_feishu_instance_state( req.external_id, ) except Exception as exc: # noqa: BLE001 fetch_errors += 1 logger.warning( "approval_pending_reaper: fetch failed for %s: %s", req.external_id, exc, ) # Hard-expire only when we still can't see it after # ``hard_days``. Don't punish a transient Feishu outage. # ``req.created_at`` is naive when read back from MySQL/TiDB # — see the audit-log block above for the explanation. if _as_aware_utc(req.created_at) < hard_cutoff: req.status = "expired" expired += 1 await _record_terminal( req, "expired", reason=(f"Feishu instance unreachable for ≥{hard_days} days; auto-expired"), ) await db.commit() continue # ``get_instance_detail`` returns Feishu's ``data`` block; the # status field is keyed ``status`` (not ``instance_status``). # The original implementation read ``instance_status`` which # never resolved — every status-based branch below was a # silent no-op since PR #164 launched the reaper. The only # branch that actually fired was the fetch-error hard-expire. # 2026-05-28: keyed off ``status`` to restore the documented # behaviour AND added a replay branch for terminal Feishu # states (APPROVED / COMPLETED) so a missed webhook gets # reconciled by the daily cron instead of stuck forever. feishu_status = (instance.get("status") or "").upper() if feishu_status in _TERMINAL_FEISHU_STATUSES_FOR_REPLAY: # Either side of the terminal binary (approved / cancelled / # rejected / timeout) replays through the same dispatcher # — handle_feishu_callback owns the branching to the right # per-domain handler. Without the replay, a webhook drop on # an APPROVED instance left the row pending forever AND the # auto-execute side-effects (asset return, seat activation, # scope grant, DNS write, …) never fired. await _replay(req, instance) elif feishu_status == "PENDING" and _as_aware_utc(req.created_at) < hard_cutoff: req.status = "expired" expired += 1 await _record_terminal( req, "expired", reason=(f"Feishu instance still PENDING after {hard_days} days; auto-expired"), ) await db.commit() # If every row failed (Feishu open-API outage / credential rotation / # systematic dispatcher bug crashing every replay), the reaper # appears healthy but isn't doing its job. Raise so the failure- # notifier surfaces the outage. Counting BOTH fetch_errors and # replay_failures lets a 100%-fail caused by a dispatcher regression # (rather than a Feishu outage) still page ops on the same day — # the previous "fetch_errors only" guard let a systematic replay # crash silently rot for a week. if rows and (fetch_errors + replay_failures) == len(rows): raise RuntimeError( f"feishu.approval_pending_reaper: all {len(rows)} rows failed " f"(fetch_errors={fetch_errors}, replay_failures={replay_failures})" ) return { "checked": len(rows), "expired": expired, "replayed": replayed, "replay_failures": replay_failures, "fetch_errors": fetch_errors, } @register_scheduled_job( key="feishu.approval_drift_reconciler", display_name="Feishu approval 漂移快速对账", description=( "每 15 分钟扫描 status='pending' 且提交超过 min_age_minutes" "(默认 10 分钟)的 ApprovalRequest,逐行拉取 Feishu 实例状态;" "若 Feishu 已终态(APPROVED/REJECTED/CANCELED/TIMEOUT)则 replay" " missed callback,让状态翻转 + auto-execute 副作用在分钟级到位。" "本轮 durable intent 恢复刚物化的非资产审批会优先同 tick replay," "但仍与老化行共享 batch_limit 总上限。" "背景: Feishu webhook 实践上是 at-most-once,终态事件丢失时列表" "状态会卡在待处理,直到有人点开详情。每日 04:00 的" " approval_pending_reaper 只扫 14 天以上的行,补不了这个窗口。" "本 cron 是快速通道:不做过期回收(交给 reaper),只做终态 replay。" "config: min_age_minutes(默认 10)、batch_limit(默认 200)。" ), ) async def run_approval_drift_reconciler( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: """Fast-path reconcile for missed terminal Feishu webhooks. Complements ``feishu.approval_pending_reaper``: same replay machinery, but runs every 15 minutes over pending rows older than ``min_age_minutes`` plus exact non-asset rows materialized by this run's durable-intent recovery. Current-run rows take priority while sharing the same total ``batch_limit``; unrelated young rows retain the age gate. No expiry logic here — the daily reaper stays the single owner of soft/hard expiry so the two crons can't disagree about when a row dies. Per-row failures are SAVEPOINT-isolated exactly like the reaper; an all-rows-failed tick raises so the failure notifier fires. """ from datetime import timedelta from app.approval.models import ApprovalRequest from app.approval.service import ApprovalService from app.approval.submission_intents import ( # noqa: PLC0415 collect_materialized_approval_request_ids, ) min_age_minutes = int(config.get("min_age_minutes") or 10) batch_limit = int(config.get("batch_limit") or 200) age_cutoff = datetime.now(UTC) - timedelta(minutes=min_age_minutes) intent_min_age_minutes = int(config.get("intent_min_age_minutes") or 2) intent_batch_limit = int(config.get("intent_batch_limit") or 100) from app.domains.ai_tool_application.service import ( # noqa: PLC0415 AiToolApplicationService, ) from app.domains.announcement_request.service import ( # noqa: PLC0415 AnnouncementRequestService, ) from app.domains.apikey.service import ApiKeyService # noqa: PLC0415 from app.domains.cmdb.dns_request_service import ( # noqa: PLC0415 DnsRequestService, ) from app.domains.cmdb.scope_request_service import ( # noqa: PLC0415 CloudScopeRequestService, ) from app.domains.data_access.service import DataAccessService # noqa: PLC0415 from app.domains.domain_registration.service import ( # noqa: PLC0415 DomainRegistrationService, ) from app.domains.sso.service import SsoGrantService # noqa: PLC0415 from app.domains.vpn_application.service import ( # noqa: PLC0415 VpnApplicationService, ) from app.identity.consumers.request_service import ( # noqa: PLC0415 ConsumerRequestService, ) service = ApprovalService(db) summaries: list[dict[str, int]] = [] async def collect_intent_summary( domain: str, operation: Awaitable[dict[str, int]], ) -> None: try: summaries.append(await operation) except Exception as exc: # noqa: BLE001 await db.rollback() logger.error( "approval_drift_reconciler: intent domain failed domain=%s error_type=%s", domain, type(exc).__name__, ) summaries.append({"failed": 1, "domain_failures": 1}) # Capture the typed ApprovalRequest links written by successful non-asset # finalizers in this exact task/run. Context-local collection is narrower # than a created_at/completed_at window: a concurrent HTTP submission may # create an equally-young pending row, but it must stay outside this drift # fetch and retain the normal min-age protection. with collect_materialized_approval_request_ids() as materialized_ids: await collect_intent_summary( "asset_request", service.reconcile_submission_intents( min_age_minutes=intent_min_age_minutes, batch_limit=intent_batch_limit, ), ) for domain_service in ( AnnouncementRequestService(db), VpnApplicationService(db), AiToolApplicationService(db), DomainRegistrationService(db), SsoGrantService(db), ConsumerRequestService(db), ): await collect_intent_summary( type(domain_service).__name__, domain_service.reconcile_submission_intents( min_age_minutes=intent_min_age_minutes, batch_limit=intent_batch_limit, ), ) await collect_intent_summary( "DataAccessService", DataAccessService().reconcile_submission_intents( db, min_age_minutes=intent_min_age_minutes, batch_limit=intent_batch_limit, ), ) for keyed_domain_service in ( ApiKeyService(db), CloudScopeRequestService(db), DnsRequestService(db), ): await collect_intent_summary( type(keyed_domain_service).__name__, keyed_domain_service.reconcile_submission_intents( min_age_minutes=intent_min_age_minutes, batch_limit=intent_batch_limit, ), ) intent_summary = { key: sum(summary.get(key, 0) for summary in summaries) for key in set().union(*(summary.keys() for summary in summaries)) } def raise_if_intent_reconciliation_degraded() -> None: """Fail the job only after every recovery surface had a chance to run.""" domain_failures = int(intent_summary.get("domain_failures", 0)) if domain_failures == len(summaries): raise RuntimeError( f"feishu.approval_drift_reconciler: all {len(summaries)} intent domains failed" ) checked = int(intent_summary.get("checked", 0)) # A method-level crash contributes one synthetic ``failed`` so the # summary remains useful, but it is not a checked intent. Subtract # those sentinels before deciding whether every processed item failed. item_failures = max( 0, int(intent_summary.get("failed", 0)) - domain_failures, ) if checked > 0 and item_failures >= checked: raise RuntimeError( f"feishu.approval_drift_reconciler: all {checked} checked intents failed" ) # Current-run materializations bypass only the age threshold. The ID set # is bounded by the eleven intent reconcilers' existing batch limits and # cannot contain an unrelated young HTTP submission. current_run_rows = ( list( ( await db.scalars( select(ApprovalRequest) .where( ApprovalRequest.status == "pending", ApprovalRequest.id.in_(materialized_ids), ) .order_by(ApprovalRequest.created_at.asc(), ApprovalRequest.id.asc()) .limit(batch_limit) ) ).all() ) if materialized_ids else [] ) remaining_capacity = batch_limit - len(current_run_rows) aged_rows: list[ApprovalRequest] = [] if remaining_capacity > 0: aged_predicates = [ ApprovalRequest.status == "pending", ApprovalRequest.created_at < age_cutoff, ] current_run_ids = {row.id for row in current_run_rows} if current_run_ids: aged_predicates.append(ApprovalRequest.id.not_in(current_run_ids)) aged_rows = list( ( await db.scalars( select(ApprovalRequest) .where(*aged_predicates) # Newest first: fresh rows are the ones users are actively # watching; ancient stragglers are the daily reaper's # problem anyway (expiry) and can wait a tick if the batch # cap ever bites. .order_by(ApprovalRequest.created_at.desc()) .limit(remaining_capacity) ) ).all() ) rows = [*current_run_rows, *aged_rows] if not rows: raise_if_intent_reconciliation_degraded() return { "checked": 0, "replayed": 0, "still_pending": 0, "fetch_errors": 0, "replay_failures": 0, "submission_intents": intent_summary, } replayed = 0 still_pending = 0 fetch_errors = 0 replay_failures = 0 # PER-ROW COMMIT (login-brownout fix, same class as the reaper): the # replay dispatches applicant notifications (outbox INSERT → FK lock on # identity_persons); a single trailing commit held that fan-out of locks # across up to 200 Feishu fetches every 15 min → repeated SSO login # lock-waits. Iterate over captured external_ids (this loop never mutates # ``req``, so no re-fetch needed — capturing before the first commit # avoids touching expired instances) and commit after each replay. external_ids = [r.external_id for r in rows] for external_id in external_ids: try: instance = await service.fetch_feishu_instance_state( external_id, ) except Exception as exc: # noqa: BLE001 fetch_errors += 1 logger.warning( "approval_drift_reconciler: fetch failed for %s: %s", external_id, sanitize_log_text( format_exception_for_log(exc), max_len=500, ), ) continue feishu_status = (instance.get("status") or "").upper() if feishu_status not in _TERMINAL_FEISHU_STATUSES_FOR_REPLAY: still_pending += 1 continue try: # SAVEPOINT (2026-08-06 review): scope this iteration's writes so a # DB-level failure rolls back only its own work. ``savepoint`` # re-raises everything, so the handler chain below keeps its exact # semantics — it only stops a failure from poisoning the session # for the remaining items and the run's trailing commit. async with savepoint(db, label="intent reconcile item"): # SAVEPOINT-isolate the replay — same rationale as the # reaper: one poisoned side-effect must not discard the # row's own write, and a deadlock (OperationalError) # propagates to mark the run failed. async with db.begin_nested(): result = await service.replay_callback_from_feishu( instance_code=external_id, actor_user_id=None, prefetched_instance_detail=instance, trace_id_prefix="feishu.approval_drift_reconciler", ) if result.get("replayed"): replayed += 1 # Commit this row's replay before the next external fetch so # its outbox FK locks don't span the rest of the batch. await db.commit() except OperationalError: raise except Exception: # noqa: BLE001 replay_failures += 1 logger.exception( "approval_drift_reconciler: replay failed for %s; leaving row pending", external_id, ) if rows and (fetch_errors + replay_failures) == len(rows): raise RuntimeError( f"feishu.approval_drift_reconciler: all {len(rows)} rows failed " f"(fetch_errors={fetch_errors}, replay_failures={replay_failures})" ) # No forward progress on ANY terminal row is an outage too, even when # the batch looks mostly healthy because the other rows are genuinely # still pending at Feishu. The old all-rows-failed guard let exactly # this hide: one approved-at-Feishu seat request failed replay every # 15 min for 20h while the job reported ``last_status=ok``, because # 1 failure out of 8 checked rows never tripped it (2026-08-06). if replay_failures and not replayed: raise RuntimeError( "feishu.approval_drift_reconciler: no terminal row could be " f"replayed (replay_failures={replay_failures}, " f"still_pending={still_pending}, fetch_errors={fetch_errors})" ) raise_if_intent_reconciliation_degraded() return { "checked": len(rows), "replayed": replayed, "still_pending": still_pending, "fetch_errors": fetch_errors, "replay_failures": replay_failures, "submission_intents": intent_summary, }