from __future__ import annotations import asyncio import hashlib import hmac import json import unicodedata from datetime import UTC, date, datetime, timedelta from decimal import Decimal from typing import Any from uuid import UUID, uuid4 from fastapi.encoders import jsonable_encoder from sqlalchemy import Numeric, and_, func, literal_column, or_, select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from app.approval.buyback_price import ( SUPPORTED_CURRENCIES as BUYBACK_SUPPORTED_CURRENCIES, ) from app.approval.buyback_price import ( calculate_buyback, resolve_default_currency, ) from app.approval.feishu_client import ( FeishuApprovalClient, parse_feishu_error_code, ) from app.approval.models import ApprovalRequest, ApprovalSubmissionIntent from app.approval.schemas import ( LEAVER_SELF_SERVICE_TEMPLATE_KEYS, AdminApprovalDetailResponse, AdminApprovalListItem, AdminApprovalTimelineEntry, AdminBuybackOnBehalfContextResponse, ApprovalBackedEmployeeRequestResponse, ApprovalProgress, ApprovalProgressNode, ApprovalTemplateOption, ApprovalTemplateStage, ApprovalTemplateWidget, AssetApprovalSubmitRequest, AssetApprovalTemplateResponse, BuybackQuoteResponse, FlowStageBundle, ) from app.approval.submission_idempotency import ( normalize_approval_idempotency_key, ) from app.approval.submission_intents import ( compact_terminal_intent_payload, provider_attempt_anchors, reconciliation_due_predicate, reconciliation_order, record_provider_attempt_anchor, ) from app.audit.service import AuditService from app.auth.models import AuthUser from app.core.exceptions import ( AppException, ForbiddenError, NotFoundError, ValidationError, ) from app.core.fixed_currency import SUPPORTED_FIXED_CURRENCIES, to_cny from app.core.grant_eligibility import is_blocked_from_new_grant from app.core.rbac import can_manage_assets from app.core.safe_logging import get_safe_logger, sanitize_log_text from app.core.txn import best_effort, savepoint from app.domains.asset.models import ( ASSET_STATUS_HELD_BY_LEAVER, AssetEmployeeRequest, AssetItem, AssetLegalEntity, ) from app.integrations.compat import get_feishu_config from app.integrations.feishu_widget_values import ( FEISHU_FORM_VALUE_ERROR_CODES, date_interval_wire_value, form_without_date_interval, rejection_targets_date_interval, ) from app.integrations.models import IntegrationApprovalFlow logger = get_safe_logger(__name__) _ASSET_PROVIDER_LOOKUP_ONLY = "provider_lookup_only" _ASSET_PROVIDER_LOOKUP_CURSOR = "provider_lookup_cursor" _ASSET_PROVIDER_ANCHOR_LIMIT = 32 _ASSET_PROVIDER_LOOKUP_LIMIT = 4 _ASSET_PROVIDER_ATTEMPT_LEASE_MINUTES = 5 # ── Feishu widget name aliases (single source of truth) ──────────── # # Feishu admins free-form widget names in whichever language they # prefer, with arbitrary casing. Strict equality on a single spelling # silently drops fields on the next admin rename — a bug class that # has cost us 6+ hotfix rounds (laptop_replacement 2026-05-11/12/18, # asset_return 2026-05-19, offboarding_buyback 2026-05-19/20, Payment # Information security bypass 2026-05-20). # # Every alias set used by ``_collect_request_type_metadata``, # ``_approver_only_widget_names``, ``_augment_answers_with_asset``, # the submit-time detail stash, and the auto-execute path resolves # here. When the Feishu admin renames a widget OR adds a localised # variant, update **only** this section. Verify by pulling the live # ``platform_settings['approval.asset_template_cache']`` payload (or # the prod TiDB row directly) and confirming every shipped # ``widget.name`` for the affected template has a matching alias # entry. Mirror to ``web/src/utils/approvalWidgetI18n.ts`` so the # frontend label resolver stays in lockstep. def _normalise_widget_name(raw: str | None) -> str: """Canonicalise a Feishu widget name before comparing to an alias. Round 4 audit #5 (2026-05-20): a Feishu admin who renames the ``Payment Information`` widget with an NBSP (``\\u00A0``) or a fullwidth ``Ｉ`` (``\\uFF29``) would bypass the unconditional server-side overwrite in ``_augment_answers_with_asset`` — the bypass was the exact bug class the alias set was added to close. NFKC normalisation collapses fullwidth + compatibility variants to their canonical form; whitespace cleanup catches NBSP / ZWSP / tabs that ``.strip()`` ignores. Returns ``""`` on a non-string input so callers can compare ``in {…normalised aliases…}`` without a None-guard. """ if not isinstance(raw, str): return "" normed = unicodedata.normalize("NFKC", raw) # Strip zero-width characters (ZWSP, ZWJ, ZWNJ, BOM) — they survive # NFKC and are invisible to a human reading the rendered form. for invisible in ("​", "‌", "‍", ""): normed = normed.replace(invisible, "") # Collapse internal runs of any-whitespace (NBSP normalises to # regular space under NFKC; runs of spaces/tabs collapse to one). normed = " ".join(normed.split()) return normed def _normalise_legal_entity_name(raw: str | None) -> str: """Return the exact-safe comparison key for a legal-entity name. Historical asset imports commonly differ from the registered entity only by presentation punctuation (for example ``LTD`` versus ``LTD.``). Such a difference must not make a payment amount fall back to another currency. The comparison remains deliberately narrow: it never fuzzy-matches words or abbreviations that could select another company's bank details. """ return _normalise_widget_name(raw).replace(".", "").casefold() # Pre-normalised alias sets — keep the user-readable spellings in the # constants above (so they're easy to grep for) but compare against # the normalised form to defeat NBSP / fullwidth / ZWSP bypasses. WIDGET_ALIASES_PAYMENT_INFO: frozenset[str] = frozenset({ "Payment Information", "Payment information", "付款信息", }) _WIDGET_ALIASES_PAYMENT_INFO_NORMALISED: frozenset[str] = frozenset( _normalise_widget_name(n) for n in WIDGET_ALIASES_PAYMENT_INFO ) # 所属主体 / Owner Entity — a 单选 (radioV2) widget the Feishu admin # added 2026-06-17 to laptop_replacement + offboarding_buyback. It is # the SAME value as the 公司主体 line of Payment Information: the # asset's owning legal entity (``asset.purchase_entity`` → # ``asset_legal_entities.name``). Server-derived + stamped in # ``_augment_answers_with_asset`` so the constrained option always # matches the bank dump the approver sees, rather than a client guess. # Legacy ``资产所属主体`` / ``Owner Entity`` kept so the older # approver-only buyback widget (and any historical callback row) still # resolves. Mirror to ``web/src/components/approvalWidgetEngine.ts`` # (``isOwnerEntityWidget``) so the frontend prefill stays in lockstep. WIDGET_ALIASES_OWNER_ENTITY: tuple[str, ...] = ( "所属主体", "资产所属主体", "Owner Entity", "Owning Entity", ) _WIDGET_ALIASES_OWNER_ENTITY_NORMALISED: frozenset[str] = frozenset( _normalise_widget_name(n) for n in WIDGET_ALIASES_OWNER_ENTITY ) # asset_return — current Feishu widget is ``归还日期`` / ``Return Date``. # Legacy ``离职日期`` / ``Resignation Date`` is included because the # 2025-era asset_return form re-used the resignation-date widget for # what was semantically the *return* date; any historical row stamped # under those names must still resolve. NEVER add ``Last working day`` # here — that is the offboarding_buyback resignation widget and # blending the two would let a buyback callback's resignation date # silently land on an asset_return record. WIDGET_ALIASES_ASSET_RETURN_DATE: tuple[str, ...] = ( "归还日期", "Return Date", "Return date", "离职日期", "Resignation Date", ) # offboarding_buyback — current widget is ``Last working day``. # Legacy ``Resignation Date`` / ``离职日期`` kept for historical rows. WIDGET_ALIASES_RESIGNATION_DATE: tuple[str, ...] = ( "Last working day", "Last Working Day", "最后工作日", "Resignation Date", "离职日期", ) _WIDGET_ALIASES_RESIGNATION_DATE_NORMALISED: frozenset[str] = frozenset( _normalise_widget_name(name) for name in WIDGET_ALIASES_RESIGNATION_DATE ) # laptop_replacement — the 是否回购旧电脑 radio that decides whether the # applicant pays to buy back the device they're replacing. When the # answer is affirmative the (otherwise approver-only) 付款金额 amount # widget becomes the buyback price and we auto-fill it with the policy # quote, exactly like offboarding_buyback. Mirror to # ``web/src/components/approvalWidgetEngine.ts`` (``isBuybackChoiceWidget``) # so the desktop + mobile create forms gate the amount field identically. WIDGET_ALIASES_BUYBACK_CHOICE: tuple[str, ...] = ( "是否回购旧电脑", "是否回购", "Whether to purchase the old computer that is currently in use？", "Whether to purchase the old computer that is currently in use?", ) _WIDGET_ALIASES_BUYBACK_CHOICE_NORMALISED: frozenset[str] = frozenset( _normalise_widget_name(n) for n in WIDGET_ALIASES_BUYBACK_CHOICE ) # Affirmative answers for the 是否回购 radio. Feishu returns the option # *text* on the wire (real prod laptop_replacement instances carry the # literal ``"Yes"`` / ``"No"``; the older designer used ``"是"`` / ``"否"``). # Compared case-folded so ``Yes`` / ``yes`` both match. _BUYBACK_AFFIRMATIVE_VALUES: frozenset[str] = frozenset({ "yes", "y", "true", "1", "是", "是的", "需要", "回购", }) def _parse_iso_date(value: Any) -> date | None: """Best-effort parse of a Feishu-form date string into ``date``. Feishu dates arrive as e.g. ``"2026-04-23"`` or occasionally as an ISO datetime. Returns None for empties / un-parseable input so the caller can fall back to today's date via the service defaults. """ if value in (None, "", [], {}): return None if isinstance(value, date) and not isinstance(value, datetime): return value if isinstance(value, datetime): return value.date() try: return date.fromisoformat(str(value)[:10]) except (ValueError, TypeError): return None class ApprovalService: def __init__(self, db: AsyncSession): self.db = db # Client is built lazily from the primary chat_approval integration # — constructing it synchronously in __init__ would require a DB # roundtrip, which sync-construction forbids. self._client: FeishuApprovalClient | None = None @property def audit(self) -> AuditService: """Audit writer for this service's session. Most of this file constructs a local ``audit = AuditService(self.db)``, but the asset auto-execute path (``_auto_execute_approved_request`` and ``_record_auto_execute_failure``) calls ``self.audit.log(...)`` in seven places — and this attribute never existed, so every one of them raised ``AttributeError``. It went unnoticed because those calls sit inside "swallow and continue" blocks: the asset state-flip had already been written, the exception was logged and dropped, and the failure evidence writer raised the same ``AttributeError`` and was swallowed by its own guard. Net effect in production: **Feishu-driven asset borrow / return / dispose never wrote their audit rows**, and a real auto-execute failure was completely invisible. Found 2026-08-06 when SAVEPOINT-isolating that block turned the silent abort into a visible rollback. A property (rather than seven local vars) also keeps any future ``self.audit`` caller correct by construction. """ return AuditService(self.db) async def _get_client(self) -> FeishuApprovalClient: """Resolve the Feishu client from the primary chat_approval integration's credentials. Cached per-instance so one request only hits the DB once.""" if self._client is None: cfg = await get_feishu_config(session=self.db) self._client = FeishuApprovalClient( app_id=str(cfg.get("app_id") or ""), app_secret=str(cfg.get("app_secret") or ""), ) return self._client @property def client(self) -> FeishuApprovalClient: """Back-compat accessor for tests that set ``service.client`` directly. Production code should ``await self._get_client()``.""" if self._client is None: raise RuntimeError( "FeishuApprovalClient not yet initialized — call " "await _get_client() first", ) return self._client @client.setter def client(self, value: FeishuApprovalClient) -> None: self._client = value async def template_config(self) -> dict[str, dict[str, Any]]: """Asset approval templates driven by ``integration_approval_flows``. Reads the active rows from the primary chat_approval integration (Feishu today). Flow keys managed elsewhere — currently just ``api_key_request`` in the apikey domain — are filtered out. New asset flows can be added via the admin UI under ``/admin/integrations//flows`` without code changes. """ from app.integrations.service import get_approval_flows_map flows = await get_approval_flows_map(session=self.db) asset_flow_keys = { "laptop_replacement", "computer_borrow", "asset_return", "offboarding_buyback", "software_seat", "software_addon", } result: dict[str, dict[str, Any]] = {} for template_key, code in flows.items(): if template_key not in asset_flow_keys or not code: continue result[template_key] = { "approval_code": code, "request_type": template_key, } return result @staticmethod def _approver_only_widget_names() -> dict[str, set[str]]: """Widget names the APPROVER (not the applicant) fills. Used as a **denylist** at template-fetch time so the create- request modal hides fields like ``Payment Information`` that only the IT-side approver completes after the request lands. We deliberately invert from a "visible-name whitelist" to a denylist (2026-05-12): the previous whitelist silently dropped every widget the Feishu admin renamed (e.g. "Request for Computer Model" → "申请电脑机型") AND every static description text block, because their names were not in the hardcoded set. That broke applicant rendering whenever admins edited the Feishu template — the exact failure mode the round-12 audit flagged ("电脑更换少了一堆字段和提示"). Denylist scales: new applicant-facing fields that admins add (config notes, delivery-time notices, FY-year spec sheets) flow through automatically. Only approver-only fields stay enumerated, and we list bilingual aliases so a Chinese rename of an approver-only widget doesn't reopen the leak either. """ # Only templates with KNOWN approver-only widgets are listed. # Anything missing here (``software_seat``, ``computer_borrow``, # ``api_key_request``, and every future flow that hasn't opted # in) hits the no-key passthrough in ``_filter_applicant_widgets`` # and ships every widget Feishu returns. That's the conservative # behaviour — don't second-guess the Feishu template unless an # explicit approver-only widget is identified here. return { "laptop_replacement": { # 2026-05-18: ``Payment Information`` removed from the # approver-only set. The applicant now sees the field # auto-filled (read-only) from the asset's # ``purchase_entity`` → ``asset_legal_entities`` so the # IT admin doesn't have to retype bank info per approval. # # 2026-06-24: ``Payment Amount`` / ``付款金额`` removed # too. In the current Feishu template it is the *buyback # price* of the old computer the applicant chose to # repurchase (是否回购旧电脑 = 是), NOT a free IT-set # figure — so it must be auto-filled with the policy # quote exactly like offboarding_buyback (and # server-enforced against tampering). It flows through to # the applicant form now; ``create_request`` drops it # again whenever buyback was NOT chosen, so the common # no-buyback path never sees a (statically required) # amount widget. See ``_enforce_server_buyback_amount``. "IT资产编码", "IT Asset Code", "Type", "丢失类型", "Police Report Receipt", "报警回执", }, "asset_return": { "快递单号", "Tracking Number", }, "offboarding_buyback": { "回购设备资产号", "Buyback Asset Code", # 2026-05-26 (revert of 2026-05-19): ``Computer asset # number`` / ``电脑资产编号`` is NOT approver-only — the # buyback applicant always picks an asset (server # gate at line ~1047) and we know its tag at submit # time. Denylisting the widget meant # ``_augment_answers_with_asset`` never saw it, so # the Feishu approval card shipped with a blank tag # column — Finance / IT had to manually cross- # reference asset_id from the link instead of just # reading the field. Approvers now see the auto- # filled value on the card (read-only in the # serialized form_payload because the applicant # form pre-stamps it via ``_augment_answers_with_ # asset``). "资产所属主体", "Owner Entity", }, } # ── Template cache (PR-D 2026-04-21) ──────────────────────── # # The original ``list_asset_templates`` did N round-trips to the # Feishu open-API every time the "create request" page rendered # (one per active template, typically 5). That's the dominant # cause of the "创建申请页太卡太慢" complaint — a cold page load # routinely blocked for 3-5 seconds. # # Fix: cache the serialised response in ``platform_settings`` with # a configurable TTL (default 6 h). The template schema is owned # by admins and changes maybe once a quarter; a stale cache is # survivable for that window. Admins can force an immediate # refresh via POST /api/v1/approval/asset/templates/refresh. _CACHE_KEY = "approval.asset_template_cache" _CACHE_TTL_KEY = "approval.asset_template_cache.ttl_minutes" _DEFAULT_CACHE_TTL_MINUTES = 360 # 6 h # All-flow-stages cache — broader than the asset-template cache # (which only covers the 4 hardware/software templates that drive # the create-request form). This one fans across every Feishu- # backed flow on file in ``integration_approval_flows`` (api_key / # google_* / dns / vpn / SSO / cloud_scope / announcement / ...). # Used by the admin "流程节点备注" Modal so IT can review + annotate # every flow in one place, and by non-asset flow create-modals # that want to render the same stages preview as the asset ones. _FLOW_STAGES_CACHE_KEY = "approval.flow_stages_cache" _STAGE_NOTES_KEY = "approval.stage_notes" async def _cache_ttl_minutes(self) -> int: from app.integrations.service import get_platform_setting raw = await get_platform_setting( self._CACHE_TTL_KEY, default=self._DEFAULT_CACHE_TTL_MINUTES, session=self.db, ) try: n = int(raw) except (TypeError, ValueError): return self._DEFAULT_CACHE_TTL_MINUTES return max(n, 0) async def _read_template_cache( self, ) -> list[AssetApprovalTemplateResponse] | None: """Return cached templates if fresh, else None.""" from app.integrations.service import get_platform_setting ttl_minutes = await self._cache_ttl_minutes() if ttl_minutes <= 0: return None entry = await get_platform_setting( self._CACHE_KEY, default=None, session=self.db, ) if not isinstance(entry, dict): return None cached_at_raw = entry.get("cached_at") payload = entry.get("payload") if not isinstance(payload, list) or not isinstance(cached_at_raw, str): return None try: cached_at = datetime.fromisoformat(cached_at_raw) except ValueError: return None if cached_at.tzinfo is None: cached_at = cached_at.replace(tzinfo=UTC) age = datetime.now(UTC) - cached_at if age.total_seconds() > ttl_minutes * 60: return None try: return [ AssetApprovalTemplateResponse.model_validate(item) for item in payload ] except Exception: # noqa: BLE001 — corrupted cache, pretend miss logger.warning( "Approval template cache is corrupted; dropping it.", ) return None async def _write_template_cache( self, templates: list[AssetApprovalTemplateResponse], ) -> None: from app.integrations.models import PlatformSetting payload = { "cached_at": datetime.now(UTC).isoformat(), "payload": [t.model_dump(mode="json") for t in templates], } existing = await self.db.get(PlatformSetting, self._CACHE_KEY) if existing is None: self.db.add( PlatformSetting( key=self._CACHE_KEY, value=payload, description="Feishu 审批模板缓存(自动写入,勿手改)。", ) ) else: existing.value = payload await self.db.flush() async def invalidate_template_cache(self) -> None: """Drop the cached payload so the next read refetches from Feishu. Exposed via ``POST /api/v1/approval/asset/templates/refresh`` for the admin "立即刷新" button. """ from app.integrations.models import PlatformSetting row = await self.db.get(PlatformSetting, self._CACHE_KEY) if row is not None: await self.db.delete(row) await self.db.flush() # ── Stage notes (admin-curated hints on approval flow nodes) ─── # # Storage shape (one ``platform_settings`` row, key # ``approval.stage_notes``): # # { "": { "": "" } } # # Refreshing the stage list from Feishu doesn't touch this map; # when a node_id disappears from a redrawn flow the orphan entry # remains in storage but stops surfacing (the join in # ``_attach_notes_to_stages`` only matches against live node_ids). # Admin-only writes via ``set_stage_note``; reads are public # (notes are designed for end-user consumption). async def _load_stage_notes(self) -> dict[str, dict[str, str]]: """Return the full notes map. Empty dict when unset.""" from app.integrations.service import get_platform_setting raw = await get_platform_setting( self._STAGE_NOTES_KEY, default=None, session=self.db, ) if not isinstance(raw, dict): return {} cleaned: dict[str, dict[str, str]] = {} for code, nodes in raw.items(): if not isinstance(code, str) or not isinstance(nodes, dict): continue inner: dict[str, str] = {} for nid, note in nodes.items(): if isinstance(nid, str) and isinstance(note, str) and note.strip(): inner[nid] = note if inner: cleaned[code] = inner return cleaned @staticmethod def _attach_notes_to_stages( stages: list[ApprovalTemplateStage], approval_code: str, notes_map: dict[str, dict[str, str]], ) -> list[ApprovalTemplateStage]: """Return a copy of ``stages`` with ``note`` populated. Stages whose ``node_id`` matches an entry in ``notes_map[approval_code]`` get the fresh note attached. Stages whose node_id has NO entry in the map get ``note=None`` — this is the deletion path (admin deleted a note that was previously baked into the cache; the cache snapshot still carries the stale note in ``stage.note``, but a re-join with an empty notes_map MUST clear it). Without this clear, a deleted note silently re-appears every read until the Feishu template cache is busted. Stages with no ``node_id`` (anonymous nodes — already filtered upstream in most paths, but possible) silently never receive a note. Always returns a fresh list — callers never see the input list. Each stage is a fresh ``model_copy`` so we never mutate the cached pydantic objects. """ flow_notes = notes_map.get(approval_code) or {} enriched: list[ApprovalTemplateStage] = [] for stage in stages: note = flow_notes.get(stage.node_id) if stage.node_id else None if note: enriched.append(stage.model_copy(update={"note": note})) elif stage.note is not None: # Cache may carry a stale baked-in note (from a # previous write where the admin's note was active). # Clear it so the response reflects the *current* # notes_map state. enriched.append(stage.model_copy(update={"note": None})) else: enriched.append(stage) return enriched async def set_stage_note( self, *, approval_code: str, node_id: str, note: str | None, ) -> None: """Upsert ONE note. ``note=None`` or empty string deletes. Concurrent-safe: the entire (approval_code, node_id) → note map lives in ONE ``platform_settings`` row, so naive read-modify-write (``db.get`` → mutate ``.value`` → ``flush``) loses data when two admins save different nodes at the same instant — Admin B's write clobbers Admin A's. Fixed 2026-05-21 (round-2 review P0): take a row-level pessimistic lock via ``SELECT ... FOR UPDATE`` before the read so the second writer waits for the first to commit before reading. TiDB / MySQL both honour ``FOR UPDATE`` even on JSON-payload rows; the lock is on the row identity, not the column. The lock releases when the outer transaction commits — caller (the PUT /admin/stage-notes endpoint) calls ``db.commit()`` after this method, so the window is short. First-time insert path (no existing row) still races on the ``INSERT`` itself — handled by catching ``IntegrityError`` on the PK constraint and falling back to the lock path. """ from sqlalchemy import select as sa_select from sqlalchemy.exc import IntegrityError from app.integrations.models import PlatformSetting # Read-with-lock. ``with_for_update()`` is a no-op against an # empty row set, so the NULL-row case falls through to INSERT # below — that path catches IntegrityError on duplicate key # and retries with the lock. result = await self.db.execute( sa_select(PlatformSetting) .where(PlatformSetting.key == self._STAGE_NOTES_KEY) .with_for_update() ) existing = result.scalar_one_or_none() current = ( dict(existing.value) if (existing and isinstance(existing.value, dict)) else {} ) flow_map = dict(current.get(approval_code) or {}) if note is None or not note.strip(): flow_map.pop(node_id, None) else: flow_map[node_id] = note.strip() if flow_map: current[approval_code] = flow_map else: current.pop(approval_code, None) if existing is None: new_row = PlatformSetting( key=self._STAGE_NOTES_KEY, value=current, description="管理员为审批节点添加的备注/提示(自动写入)", ) self.db.add(new_row) try: await self.db.flush() except IntegrityError: # Another writer inserted between our SELECT and our # INSERT. Re-take the lock and retry the read-modify # path against the now-existing row. await self.db.rollback() result2 = await self.db.execute( sa_select(PlatformSetting) .where(PlatformSetting.key == self._STAGE_NOTES_KEY) .with_for_update() ) existing2 = result2.scalar_one() cur2 = dict(existing2.value) if isinstance(existing2.value, dict) else {} fm2 = dict(cur2.get(approval_code) or {}) if note is None or not note.strip(): fm2.pop(node_id, None) else: fm2[node_id] = note.strip() if fm2: cur2[approval_code] = fm2 else: cur2.pop(approval_code, None) existing2.value = cur2 await self.db.flush() else: existing.value = current await self.db.flush() # ── All-flow stages (admin Modal + non-asset flow previews) ──── async def _read_flow_stages_cache(self) -> list[FlowStageBundle] | None: """TTL-gated cache read. Empty list is a legitimate result.""" from app.integrations.service import get_platform_setting ttl_minutes = await self._cache_ttl_minutes() if ttl_minutes <= 0: return None entry = await get_platform_setting( self._FLOW_STAGES_CACHE_KEY, default=None, session=self.db, ) if not isinstance(entry, dict): return None cached_at_raw = entry.get("cached_at") payload = entry.get("payload") if not isinstance(payload, list) or not isinstance(cached_at_raw, str): return None try: cached_at = datetime.fromisoformat(cached_at_raw) except ValueError: return None if cached_at.tzinfo is None: cached_at = cached_at.replace(tzinfo=UTC) age = datetime.now(UTC) - cached_at if age.total_seconds() > ttl_minutes * 60: return None try: return [FlowStageBundle.model_validate(item) for item in payload] except Exception: # noqa: BLE001 logger.warning("Flow-stages cache is corrupted; dropping it.") return None async def _write_flow_stages_cache( self, bundles: list[FlowStageBundle], ) -> None: from app.integrations.models import PlatformSetting payload = { "cached_at": datetime.now(UTC).isoformat(), "payload": [b.model_dump(mode="json") for b in bundles], } existing = await self.db.get(PlatformSetting, self._FLOW_STAGES_CACHE_KEY) if existing is None: self.db.add( PlatformSetting( key=self._FLOW_STAGES_CACHE_KEY, value=payload, description=( "全部飞书 approval 流程的节点列表缓存" "(自动写入,勿手改;管理员可在「流程节点备注」Modal 强制刷新)" ), ) ) else: existing.value = payload await self.db.flush() async def invalidate_flow_stages_cache(self) -> None: from app.integrations.models import PlatformSetting row = await self.db.get(PlatformSetting, self._FLOW_STAGES_CACHE_KEY) if row is not None: await self.db.delete(row) await self.db.flush() async def list_all_flow_stages( self, *, force_refresh: bool = False, ) -> list[FlowStageBundle]: """Return every Feishu-backed flow's stage chain + admin notes. Reads cache (TTL gated) when ``force_refresh=False``; otherwise re-fetches all flows in parallel from Feishu, caches the result, and returns it. Notes are joined in at every return (cache hit AND fresh fetch) so an edit to ``approval.stage_notes`` is visible on the next call without busting the stage cache. Per-flow fetch errors are non-fatal: the bundle still appears with ``fetch_error`` set and an empty ``stages`` list so the admin Modal can render the row + show the error rather than silently vanishing the flow. """ notes_map = await self._load_stage_notes() if not force_refresh: cached = await self._read_flow_stages_cache() if cached is not None: return [ b.model_copy( update={ "stages": self._attach_notes_to_stages( b.stages, b.approval_code, notes_map, ) } ) for b in cached ] # Fresh fetch: enumerate every flow on file + fan out. from app.integrations.models import IntegrationApprovalFlow rows = ( await self.db.execute(select(IntegrationApprovalFlow)) ).scalars().all() try: client = await self._get_client() except Exception as e: # noqa: BLE001 logger.warning( "Flow-stages: cannot acquire Feishu client (%s); " "serving cached/empty.", e, ) # Best-effort: try a stale cache read (no TTL gate) so a # transient Feishu outage doesn't blank the admin Modal. stale = await self._read_flow_stages_cache_stale() if stale is not None: return [ b.model_copy( update={ "stages": self._attach_notes_to_stages( b.stages, b.approval_code, notes_map, ) } ) for b in stale ] return [] async def _fetch_one(row: IntegrationApprovalFlow) -> FlowStageBundle: try: defn = await client.get_approval_definition(row.external_code) return FlowStageBundle( flow_key=row.flow_key, label=row.label, approval_code=row.external_code, approval_name=defn.get("approval_name"), stages=self._parse_template_stages(defn), ) except Exception as e: # noqa: BLE001 logger.warning( "Flow-stages fetch failed for flow_key=%s " "(approval_code=%s): %s", row.flow_key, row.external_code, e, ) return FlowStageBundle( flow_key=row.flow_key, label=row.label, approval_code=row.external_code, approval_name=None, stages=[], fetch_error=str(e), ) bundles = await asyncio.gather(*(_fetch_one(r) for r in rows)) # Sort by label for a stable admin UI order. bundles_sorted = sorted(bundles, key=lambda b: b.label or b.flow_key) try: await self._write_flow_stages_cache(bundles_sorted) except Exception as e: # noqa: BLE001 logger.warning("Flow-stages cache write failed: %s", e) # Join notes in for the response (notes are NOT cached; # caller-visible only). return [ b.model_copy( update={ "stages": self._attach_notes_to_stages( b.stages, b.approval_code, notes_map, ) } ) for b in bundles_sorted ] async def _read_flow_stages_cache_stale( self, ) -> list[FlowStageBundle] | None: """TTL-ignoring stale read used as Feishu-outage fallback.""" from app.integrations.service import get_platform_setting entry = await get_platform_setting( self._FLOW_STAGES_CACHE_KEY, default=None, session=self.db, ) if not isinstance(entry, dict): return None payload = entry.get("payload") if not isinstance(payload, list): return None try: return [FlowStageBundle.model_validate(item) for item in payload] except Exception: # noqa: BLE001 return None async def get_flow_stages_by_code( self, approval_code: str, ) -> FlowStageBundle | None: """Return ONE flow's bundle for the public preview endpoint. Used by non-asset create-modals (api_key / vpn / dns / ...) so they can render the same ApprovalStagesPreview as the asset templates. Reads from the cache (no force-fetch) — admin triggers the refresh from the management Modal. """ bundles = await self.list_all_flow_stages() for b in bundles: if b.approval_code == approval_code: return b return None async def get_flow_stages_by_key( self, flow_key: str, ) -> FlowStageBundle | None: """Same as :meth:`get_flow_stages_by_code` but keyed by the Helix-internal ``flow_key`` (``api_key_request`` / ``software_seat`` / ``google_*`` / ...). Non-asset Modals typically know their ``flow_key`` (it's a hardcoded constant in each domain) but NOT the upstream ``approval_code`` (varies per tenant / template revision). This helper resolves the key via the cached bundle list and returns the same shape so the frontend wrapper can drop the approval-flow tab into any Modal with one prop. """ bundles = await self.list_all_flow_stages() for b in bundles: if b.flow_key == flow_key: return b return None async def list_asset_templates( self, *, force_refresh: bool = False, ) -> list[AssetApprovalTemplateResponse]: if not force_refresh: cached = await self._read_template_cache() if cached is not None: # Re-join notes on every cache read so admin edits to # ``approval.stage_notes`` show up immediately without # forcing a Feishu refetch. notes_map = await self._load_stage_notes() return [ t.model_copy( update={ "stages": self._attach_notes_to_stages( t.stages, t.approval_code, notes_map, ) } ) for t in cached ] # Resilience (code-review finding #3 of pass 1): # - If Feishu is unreachable / chat_approval isn't configured, # do NOT hard-fail the create-request page. Return a stale # cache if any exists, otherwise an empty list. This trades # "transparent outage" for "empty picker" — correct for the # goal "一个秒出的申请页". try: templates: list[AssetApprovalTemplateResponse] = [] for template_key, config in (await self.template_config()).items(): definition = await (await self._get_client()).get_approval_definition(config["approval_code"]) widgets = self._filter_applicant_widgets( template_key, self._parse_template_widgets(definition.get("form", "[]")), ) # Parse stages WITHOUT notes — notes are a pure join # layer applied at response time (see below). Keeping # the cache note-free means a deleted note can't # silently re-appear from a stale cache row; the # cache only carries the upstream Feishu definition. stages = self._parse_template_stages(definition) templates.append( AssetApprovalTemplateResponse( template_key=template_key, approval_code=config["approval_code"], approval_name=definition.get("approval_name", template_key), requires_upload=any(widget.type in {"image", "attachmentV2"} for widget in widgets), widgets=widgets, stages=stages, ) ) except Exception as e: # noqa: BLE001 logger.warning( "Approval template fetch from Feishu failed (%s). " "Serving stale or empty cache instead.", e, ) # Retry the cache read WITHOUT the TTL gate, so a stale # payload still ships (better to show yesterday's templates # than an empty picker during a Feishu incident). stale = await self._read_template_cache_stale() if not stale: return [] # Same note-join on the stale path. notes_map = await self._load_stage_notes() return [ t.model_copy( update={ "stages": self._attach_notes_to_stages( t.stages, t.approval_code, notes_map, ) } ) for t in stale ] # Write-through on a fresh fetch. ``templates`` carries # note-free stages at this point (see the construction loop # above) — cache stays note-free, notes are joined at # response time below. try: await self._write_template_cache(templates) except Exception as e: # noqa: BLE001 # Don't fail the request if cache write fails; log and # return the fresh data anyway. logger.warning("Approval template cache write failed: %s", e) # Join notes for the response. The cache row stays clean so # a future deleted note can't silently re-appear from a # baked-in stage.note value. notes_map = await self._load_stage_notes() return [ t.model_copy( update={ "stages": self._attach_notes_to_stages( t.stages, t.approval_code, notes_map, ) } ) for t in templates ] async def _read_template_cache_stale( self, ) -> list[AssetApprovalTemplateResponse] | None: """TTL-ignoring cache read used as fallback on Feishu failure.""" from app.integrations.service import get_platform_setting entry = await get_platform_setting( self._CACHE_KEY, default=None, session=self.db, ) if not isinstance(entry, dict): return None payload = entry.get("payload") if not isinstance(payload, list): return None try: return [ AssetApprovalTemplateResponse.model_validate(item) for item in payload ] except Exception: # noqa: BLE001 return None async def list_asset_approval_requests( self, *, current_user: AuthUser, page: int = 1, page_size: int = 20, status: str | None = None, scope: str | None = None, request_type: list[str] | str | None = None, keyword: str | None = None, requester_id: UUID | None = None, start_date: str | None = None, end_date: str | None = None, ) -> tuple[list[ApprovalBackedEmployeeRequestResponse], int]: """List asset-approval requests. - Non-admins are always scoped to their own requests. - Admins see all requests by default, but can pass ``scope='mine'`` to narrow to their own — this backs the ``/my/requests`` page so an admin browsing their own inbox isn't buried under everyone else's submissions. """ query = select(AssetEmployeeRequest) count_query = select(func.count()).select_from(AssetEmployeeRequest) conditions = [] force_self = scope == "mine" or not can_manage_assets(current_user.role_names) if force_self: conditions.append(AssetEmployeeRequest.requester_id == current_user.id) # Dedicated applicant filter (admin views). Stacks with # ``force_self`` harmlessly — a non-admin passing someone else's # id just AND-conflicts with their own-scope clause → empty set. if requester_id is not None: conditions.append(AssetEmployeeRequest.requester_id == requester_id) if status: conditions.append(AssetEmployeeRequest.status == status) if request_type: # Accept either a single string (legacy) or a list (the # asset-mgmt vs software-mgmt split sends a multi-value list # so each menu only shows its own request kinds). if isinstance(request_type, str): conditions.append(AssetEmployeeRequest.request_type == request_type) else: conditions.append(AssetEmployeeRequest.request_type.in_(request_type)) if keyword: pattern = f"%{keyword}%" # Match the applicant (requester) by name / work_email too, # not just request_number + reason. Uses an IN-subquery # against identity_persons rather than an outer join so the # row count stays correct even when a requester row is # missing — mirrors ``list_admin_approval_requests``. requester_match = select(AuthUser.id).where( or_( AuthUser.display_name.ilike(pattern), AuthUser.work_email.ilike(pattern), ) ) conditions.append( or_( AssetEmployeeRequest.request_number.ilike(pattern), AssetEmployeeRequest.reason.ilike(pattern), AssetEmployeeRequest.requester_id.in_(requester_match), ) ) if start_date: try: start_dt = datetime.combine(date.fromisoformat(start_date), datetime.min.time()) except ValueError as exc: raise ValidationError( f"start_date must be ISO format (YYYY-MM-DD), got {start_date!r}", ) from exc conditions.append(AssetEmployeeRequest.created_at >= start_dt) if end_date: try: end_dt = datetime.combine(date.fromisoformat(end_date), datetime.max.time()) except ValueError as exc: raise ValidationError( f"end_date must be ISO format (YYYY-MM-DD), got {end_date!r}", ) from exc conditions.append(AssetEmployeeRequest.created_at <= end_dt) if conditions: query = query.where(and_(*conditions)) count_query = count_query.where(and_(*conditions)) total = (await self.db.execute(count_query)).scalar() or 0 result = await self.db.execute( query.order_by(AssetEmployeeRequest.created_at.desc()) .offset((page - 1) * page_size) .limit(page_size) ) requests = list(result.scalars().all()) requester_ids = {request.requester_id for request in requests} asset_ids = {request.asset_id for request in requests if request.asset_id} requesters = await self._load_users(requester_ids) assets = await self._load_assets(asset_ids) return [ self._to_request_response(request, requesters, assets) for request in requests ], total async def asset_request_summary( self, *, current_user: AuthUser, scope: str | None = None, request_type: list[str] | str | None = None, window_days: int = 30, ) -> dict[str, Any]: """Scoped aggregate for the employee-request page charts. Mirrors ``list_asset_approval_requests`` scoping: non-admins (or ``scope='mine'``) see only their own requests; admins see all, narrowable by ``request_type`` (the asset-mgmt vs software-mgmt domain split). ``by_status`` / ``by_request_type`` are over the full scoped population; ``daily`` is a windowed created-at trend. Single GROUP BY per slice. """ from app.core.timeseries import day_axis base_conditions = [] force_self = scope == "mine" or not can_manage_assets(current_user.role_names) if force_self: base_conditions.append( AssetEmployeeRequest.requester_id == current_user.id ) if request_type: if isinstance(request_type, str): base_conditions.append( AssetEmployeeRequest.request_type == request_type ) else: base_conditions.append( AssetEmployeeRequest.request_type.in_(request_type) ) def _scoped(stmt): return stmt.where(and_(*base_conditions)) if base_conditions else stmt status_rows = (await self.db.execute(_scoped( select(AssetEmployeeRequest.status, func.count()) .group_by(AssetEmployeeRequest.status) ))).all() by_status = {s: int(n) for s, n in status_rows} type_rows = (await self.db.execute(_scoped( select(AssetEmployeeRequest.request_type, func.count()) .group_by(AssetEmployeeRequest.request_type) ))).all() by_request_type = {s: int(n) for s, n in type_rows} axis = day_axis(window_days=window_days, now=datetime.now(UTC)) daily_rows = (await self.db.execute(_scoped( select(func.date(AssetEmployeeRequest.created_at), func.count()) .where(AssetEmployeeRequest.created_at >= axis.window_start) .group_by(func.date(AssetEmployeeRequest.created_at)) ))).all() daily_map = {str(d): int(n) for d, n in daily_rows} daily = [ {"date": d.isoformat(), "count": daily_map.get(d.isoformat(), 0)} for d in axis.days ] # ── Buyback money aggregate (2026-07-03) ────────────────────── # Sums ``detail.buyback_amount`` (first-class key stamped at # submit; migration 379 backfilled historical rows) grouped by # currency × status. Only pending + approved count — a rejected # / cancelled / expired buyback is money that never moved. # Skipped entirely when the page's request_type pin excludes # both buyback-capable types (e.g. the software-mgmt view). buyback_types = {"offboarding_buyback", "laptop_replacement"} requested_types: set[str] | None = None if request_type: requested_types = ( {request_type} if isinstance(request_type, str) else set(request_type) ) buyback_by_currency: list[dict[str, Any]] = [] if requested_types is None or (requested_types & buyback_types): currency_expr = func.json_unquote( func.json_extract( AssetEmployeeRequest.detail, "$.buyback_amount.currency", ) ) amount_expr = func.cast( func.json_unquote( func.json_extract( AssetEmployeeRequest.detail, "$.buyback_amount.amount", ) ), Numeric(18, 2), ) buyback_rows = (await self.db.execute(_scoped( select( currency_expr, AssetEmployeeRequest.status, func.sum(amount_expr), func.count(), ) .where( AssetEmployeeRequest.request_type.in_(buyback_types), AssetEmployeeRequest.status.in_(["pending", "approved"]), func.json_extract( AssetEmployeeRequest.detail, "$.buyback_amount.amount", ).is_not(None), ) .group_by(currency_expr, AssetEmployeeRequest.status) ))).all() per_currency: dict[str, dict[str, Any]] = {} for currency, row_status, amount_sum, row_count in buyback_rows: cur = (currency or "").upper() or "UNKNOWN" bucket = per_currency.setdefault(cur, { "currency": cur, "pending_total": 0.0, "pending_count": 0, "approved_total": 0.0, "approved_count": 0, }) bucket[f"{row_status}_total"] = float(amount_sum or 0) bucket[f"{row_status}_count"] = int(row_count) buyback_by_currency = sorted( per_currency.values(), key=lambda b: b["approved_total"] + b["pending_total"], reverse=True, ) buyback_total_cny = sum( ( to_cny( Decimal(str(bucket["pending_total"] + bucket["approved_total"])), str(bucket["currency"]), ) for bucket in buyback_by_currency if str(bucket["currency"]) in SUPPORTED_FIXED_CURRENCIES ), Decimal("0"), ) else: buyback_total_cny = Decimal("0") return { "window_days": window_days, "total": sum(by_status.values()), "by_status": by_status, "by_request_type": by_request_type, "daily": daily, "buyback": {"by_currency": buyback_by_currency, "total_cny": float(buyback_total_cny)}, } async def fetch_feishu_instance_state( self, instance_code: str, ) -> dict[str, Any]: """Fetch the live Feishu approval instance state for the ``feishu.approval_pending_reaper`` cron. Returns the raw Feishu ``data`` block from ``GET /approval/v4/instances/{instance_code}``. The status is carried under the key ``status`` (PENDING / APPROVED / COMPLETED / REJECTED / CANCELED / TIMEOUT) — NOT ``instance_status``, which is a separate field that Feishu ships on a different endpoint shape and was the source of a silent-no-op bug in PR #164's original reaper implementation. """ if not instance_code: raise AppException( code="APPROVAL_INSTANCE_CODE_REQUIRED", message="instance_code is required", status_code=400, ) client = await self._get_client() return await client.get_instance_detail(instance_code) or {} async def replay_callback_from_feishu( self, *, instance_code: str, actor_user_id: UUID | None, prefetched_instance_detail: dict[str, Any] | None = None, trace_id_prefix: str = "feishu.reconcile", ) -> dict[str, Any]: """Replay the Feishu callback path when the live webhook was missed. Background: Feishu's webhook delivery is at-most-once in practice — drops, signature mismatches, mid-flight 5xx, or Feishu open-platform outages can leave Helix with ``approval_requests.status='pending'`` even though Feishu's side is fully terminal (APPROVED / COMPLETED / REJECTED / CANCELLED). Without a recovery path the row's auto-execute side-effects (asset return, seat activation, scope grant, DNS write, …) never fire and the applicant's request looks stuck forever in the UI. This method is the recovery path. It is invoked from two call sites: 1. ``get_admin_approval_request_detail`` / ``get_my_approval_request_detail`` (the "刷新流程" button) — after a successful live Feishu fetch, when the response shows a terminal state but the Helix row is still pending, we replay the callback inline. The admin sees the row transition + side-effects on the SAME click that they expected to "just refresh the timeline." Cheaper than a separate "force reconcile" button and matches the user's mental model ("I clicked refresh, Helix should now match reality"). 2. ``feishu.approval_pending_reaper`` (daily cron) — when the reaper observes COMPLETED/APPROVED/PASS on a pending row, it replays here instead of the previous behaviour of "leave alone, callback should land" (which left COMPLETED rows stuck indefinitely if the original webhook was lost). Idempotency: this method is safe to call multiple times on the same row. It re-runs ``handle_feishu_callback``, which owns its own ``SELECT ... FOR UPDATE`` lock + terminal-status guard. A row that already settled to ``approved`` / ``rejected`` / ``cancelled`` short-circuits immediately with no side effects. ``"completed"`` is deliberately NOT in that short-circuit set: it can no longer be produced (COMPLETED now maps to ``approved``), and replaying a legacy ``completed`` row re-fires its orphaned side-effects (per-domain handlers are idempotent, so a genuinely-done re-process is a safe no-op). RBAC: callers are responsible for gating. The admin endpoint only invokes this for write-capable roles (``global_admin`` / ``asset_admin``); auditor + management see the refreshed timeline but never reach this method. Returns a result dict suitable for logging / audit: { "replayed": bool, # True if dispatch actually ran "reason": str, # short reason code "previous_status": str, # row.status before "new_status": str, # row.status after "feishu_status": str | None, # raw Feishu instance_status } Never raises on a "nothing to do" branch — only on infra failures (DB, Feishu) or domain handler exceptions, which propagate so the caller can audit + surface to the user. """ if not instance_code: return { "replayed": False, "reason": "no_instance_code", "previous_status": None, "new_status": None, "feishu_status": None, } row = ( await self.db.execute( select(ApprovalRequest).where( ApprovalRequest.external_id == instance_code, ) ) ).scalar_one_or_none() if row is None: return { "replayed": False, "reason": "row_not_found", "previous_status": None, "new_status": None, "feishu_status": None, } previous_status = row.status # ``"completed"`` deliberately dropped: it can no longer be produced # (COMPLETED now maps to "approved"), and any LEGACY row left at # "completed" was sealed WITHOUT its side-effects by the old mapping — # letting replay re-process it (re-mapping COMPLETED→approved) rescues # the orphaned child request. Per-domain side-effect handlers are # idempotent, so reprocessing a genuinely-done row is a safe no-op. if previous_status in ("approved", "rejected", "cancelled"): return { "replayed": False, "reason": "already_terminal", "previous_status": previous_status, "new_status": previous_status, "feishu_status": None, } # Fetch live Feishu state if the caller didn't already. if prefetched_instance_detail is None: try: client = await self._get_client() prefetched_instance_detail = ( await client.get_instance_detail(instance_code) or {} ) except Exception as fetch_err: # noqa: BLE001 # Surface as a "fetch_failed" no-op so the caller can # decide how to react. We deliberately do NOT raise: # the admin "刷新流程" path already swallows Feishu # transient errors into a yellow banner, and the # reaper has its own all-failed escalation. logger.warning( "replay_callback_from_feishu: Feishu fetch failed for %s (%s)", instance_code, fetch_err, ) return { "replayed": False, "reason": f"feishu_fetch_failed:{fetch_err}", "previous_status": previous_status, "new_status": previous_status, "feishu_status": None, } feishu_status_raw = prefetched_instance_detail.get("status") mapped = self._map_feishu_status(feishu_status_raw) if mapped == "pending": # Feishu also says pending — nothing to reconcile, the # original webhook is genuinely just not done yet. The # reaper's separate hard-expire branch handles the # truly-stuck case. return { "replayed": False, "reason": "feishu_still_pending", "previous_status": previous_status, "new_status": previous_status, "feishu_status": feishu_status_raw, } # Dispatch through the same callback path the webhook would # have taken. The dispatcher acquires its own row lock; if a # concurrent webhook lands at the same instant, one of the # two waits and then short-circuits via the terminal guard. payload = { "event": {"instance_code": instance_code}, "_replay_source": trace_id_prefix, } await self.handle_feishu_callback( payload, prefetched_instance_detail=prefetched_instance_detail, ) # Re-load to learn the post-replay status — handle_feishu_callback # may have flipped it to approved / rejected / cancelled / # completed depending on the Feishu terminal state. await self.db.refresh(row) new_status = row.status # Distinct audit row so a reconcile is traceable apart from a # real webhook (different ``trace_id`` prefix + action name). await AuditService(self.db).log( trace_id=f"{trace_id_prefix}:{instance_code}", actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="approval.reconcile.replay", status="success", resource_type="approval_request", resource_id=str(row.id), detail={ "instance_code": instance_code, "approval_type": row.approval_type, "feishu_status": feishu_status_raw, "previous_status": previous_status, "new_status": new_status, }, ) return { "replayed": True, "reason": "ok", "previous_status": previous_status, "new_status": new_status, "feishu_status": feishu_status_raw, } async def create_asset_approval_request( self, *, body: AssetApprovalSubmitRequest, current_user: AuthUser, trace_id: str, ip_address: str | None, user_agent: str | None, idempotency_key: str, # Pre-launch hardening 2026-05-04 (P0 AP1 fix): plumbing-only # extras moved off the public input schema. ONLY internal # callers may supply these — the public ``/asset/requests`` # router does NOT thread them through, so an external client # cannot inject a victim's ``seat_allocation_id`` to force- # activate someone else's seat. See AssetApprovalSubmitRequest # docstring for context. internal_extra_source_data: dict[str, Any] | None = None, internal_extra_detail: dict[str, Any] | None = None, internal_idempotency_payload: dict[str, Any] | None = None, # Admin "代提回购" (on-behalf buyback) plumbing (2026-06-30). When # set, ``current_user`` is the IT/asset admin physically submitting # the request (and the Feishu *applicant*, because the leaver's # Feishu account is frozen and can't be an applicant), while # ``internal_on_behalf_of_person_id`` is the leaver the buyback is # *for*. The stored ``requester_id`` becomes the leaver so the # auto-execute TOCTOU ownership check (requester == asset.assigned_to) # passes and the disposal is attributed correctly. ONLY internal # callers (the on-behalf router / MCP tool) thread this — the public # ``/asset/requests`` schema cannot, so an employee can't submit a # buyback against a colleague's asset. See the dedicated wrapper # ``create_offboarding_buyback_on_behalf``. internal_on_behalf_of_person_id: UUID | None = None, internal_buyback_currency: str | None = None, internal_base_location: str | None = None, internal_last_working_day: date | None = None, ) -> ApprovalBackedEmployeeRequestResponse: # On-behalf is currently scoped to offboarding_buyback only — the one # flow whose subject (a leaver) literally cannot self-submit. Reject # any other template defensively so a future caller can't repurpose # the requester-rewrite to spoof attribution on, say, a borrow. if ( internal_on_behalf_of_person_id is not None and body.template_key != "offboarding_buyback" ): raise AppException( code="ON_BEHALF_TEMPLATE_UNSUPPORTED", message=( "On-behalf submission is only supported for " "offboarding_buyback." ), status_code=400, ) if ( (internal_base_location is not None or internal_last_working_day is not None) and ( internal_on_behalf_of_person_id is None or body.template_key != "offboarding_buyback" ) ): raise AppException( code="ON_BEHALF_FIELD_OVERRIDE_UNSUPPORTED", message=( "Location and last-working-day overrides are only " "supported by the admin on-behalf buyback flow." ), status_code=400, ) # ── Leaver Blocked carve-out (2026-08-13) ──────────────────── # This endpoint is exempt from the router-level mutate gate # (``_LEAVER_MUTATE_EXACT_ALLOWLIST``) and from the MCP runtime # gate (``leaver_allowed=True`` on ``submit_approval_request``) # because a 离职中 employee has to be able to submit 归还 / # 回购 — those are offboarding obligations. Both gates hand the # per-template decision here so the two surfaces cannot drift. # # A leaver may only submit for THEMSELVES: the admin on-behalf # wrapper rewrites ``requester_id`` to a third party, so a # leaver_pending asset_admin must not reach it (both entry # points already gate it, this is the defence in depth). from app.core.dependencies import raise_leaver_runtime_locked from app.core.rbac import is_active_employee if not is_active_employee(current_user) and ( body.template_key not in LEAVER_SELF_SERVICE_TEMPLATE_KEYS or internal_on_behalf_of_person_id is not None ): raise_leaver_runtime_locked() config = (await self.template_config()).get(body.template_key) if not config: raise AppException( code="UNSUPPORTED_APPROVAL_TEMPLATE", message=f"Unsupported template key: {body.template_key}", status_code=400, ) # ``offboarding_buyback`` MUST have an asset_id — the server-side # amount-enforcement (``_enforce_server_buyback_amount``) needs # the asset row to compute the canonical quote, and the auto- # execute callback needs it to dispose. Without this server-side # check, a client can submit without asset_id, skip the amount # enforcement entirely, and ship a tampered amount to Feishu # — the approver would see a bogus card with no asset link, but # an admin completing the disposal manually would work from a # tampered audit trail. The frontend already requires asset_id # via ``TEMPLATE_ASSET_REQUIREMENT[offboarding_buyback] = true``, # but that's frontend-only; this is the matching server gate. if body.template_key == "offboarding_buyback" and not body.asset_id: raise AppException( code="OFFBOARDING_BUYBACK_REQUIRES_ASSET", message=( "offboarding_buyback submissions must reference an " "asset_id — pick the asset to be bought back." ), status_code=400, ) asset = None if body.asset_id: asset = await self._get_asset(body.asset_id) # CRITICAL IDOR guard (cycle 2 audit fix). Without this an # employee could submit an approval naming a colleague's # asset_id; on Feishu approval the auto-execute path # (``_apply_approved_request_to_asset`` → # ``_auto_execute_approved_request``) calls dispose / # borrow / return on the asset using the requester as # ``to_user_id`` / ``operator_id`` — silently transferring # or disposing of a colleague's asset. Enforce ownership # at the entry point so the approval card never gets # generated against a stranger's asset. await self._assert_asset_caller_authorized( asset=asset, template_key=body.template_key, current_user=current_user, on_behalf_of_person_id=internal_on_behalf_of_person_id, ) # On-behalf buyback: resolve the leaver the request is *for*. The # form (location auto-fill) and the stored requester_id key off this # person, not the admin who is clicking submit. Loaded once here and # reused below. on_behalf_person: AuthUser | None = None if internal_on_behalf_of_person_id is not None: from app.identity.person_models import ( IdentityPerson as _IdentityPerson, # noqa: PLC0415 ) on_behalf_person = await self.db.get( _IdentityPerson, internal_on_behalf_of_person_id, ) if on_behalf_person is None: raise NotFoundError( "IdentityPerson", str(internal_on_behalf_of_person_id), ) key = normalize_approval_idempotency_key(idempotency_key) submission_fingerprint = self._approval_submission_fingerprint( internal_idempotency_payload or { "body": body.model_dump(mode="json"), "on_behalf_of_person_id": ( str(internal_on_behalf_of_person_id) if internal_on_behalf_of_person_id else None ), "buyback_currency": internal_buyback_currency, "base_location": internal_base_location, "last_working_day": ( internal_last_working_day.isoformat() if internal_last_working_day else None ), }, ) existing_intent = await self.db.scalar( select(ApprovalSubmissionIntent).where( ApprovalSubmissionIntent.scope == "asset_request", ApprovalSubmissionIntent.applicant_id == current_user.id, ApprovalSubmissionIntent.idempotency_key == key, ), ) if existing_intent is not None: if existing_intent.payload_fingerprint != submission_fingerprint: raise AppException( code="APPROVAL_IDEMPOTENCY_PAYLOAD_MISMATCH", message=( "This Idempotency-Key was already used with a " "different approval payload." ), status_code=409, ) if existing_intent.state == "completed": return await self._submission_intent_response(existing_intent) if existing_intent.state == "provider_created": existing_intent = await self._finalize_submission_intent( existing_intent.id, recovered=True, ) return await self._submission_intent_response(existing_intent) if existing_intent.state not in { "prepared", "submitting", "unknown", "failed", }: raise AppException( code="APPROVAL_SUBMISSION_IN_PROGRESS", message=( "This approval submission is in progress or awaiting " "provider reconciliation. Retry later with the same key." ), status_code=409, ) # Perf (2026-04-22): the Feishu definition-fetch and the # email→user_id lookup are independent, so run them in parallel # instead of sequentially — saves ~300-500ms of Feishu RTT on # the hot path (the "我的申请 → 创建申请 → 提交" click). client = await self._get_client() definition, feishu_user_id = await asyncio.gather( client.get_approval_definition(config["approval_code"]), client.resolve_user_id_by_email(current_user.work_email), ) # Filter to applicant-visible widgets BEFORE augmenting + serializing. # Feishu marks many widgets ``required=True`` statically while # gating them with ``display_condition`` (e.g. laptop_replacement # "Police Report Receipt" — required only when Request Type=Lost). # The denylist captures the approver-only / system-managed # widgets that should never reach the applicant form; everything # else flows through. Validating hard-required on hidden widgets # would make every conditional submission 400 with # APPROVAL_FIELD_REQUIRED. all_widgets = self._parse_template_widgets(definition.get("form", "[]")) widgets = self._filter_applicant_widgets(body.template_key, all_widgets) # Resolve the asset's purchase_entity → AssetLegalEntity row up # front so ``_augment_answers_with_asset`` can fill the Payment # Information widget (bank info) for laptop_replacement + # offboarding_buyback without a per-widget DB round-trip. Best- # effort: missing entity → leave the widget blank, which is the # pre-2026-05-18 behaviour. The frontend already prefilled it # via ``GET /approval/asset/{id}/payment-info``; this is the # backend backstop so a curl/MCP/replay caller that skipped the # prefill round-trip still gets the same auto-fill. payment_entity = await self._resolve_payment_entity(asset) # ``current_user`` is an ``IdentityPerson`` row (re-exported as # ``AuthUser`` in ``app.auth.models``); its # ``work_location_name`` / ``city`` columns feed the location # auto-fill branch in ``_augment_answers_with_asset`` when the # asset has no ``location`` populated (or no asset linked). # The specialized on-behalf modal does not render the complete generic # widget engine. Honour valid defaults authored in Feishu so adding a # required defaulted radio does not break this server-driven flow. initial_answers = dict(body.answers) if internal_on_behalf_of_person_id is not None: for widget in widgets: if ( widget.default_value not in (None, "", [], {}) and initial_answers.get(widget.id) in (None, "", [], {}) ): initial_answers[widget.id] = widget.default_value resolved_answers = self._augment_answers_with_asset( asset, initial_answers, widgets, payment_entity=payment_entity, # On-behalf: the form represents the leaver, so location / # contact auto-fill should reflect them, not the admin clicking # submit. Falls back to the asset's own location regardless. person=on_behalf_person or current_user, ) # If asset/person metadata could not identify the constrained location, # the admin may choose one in the specialized modal. Always convert the # human label to Feishu's canonical option value. Unknown choices still # reach the existing option-set validator and fail closed. if internal_base_location: for widget in widgets: if self._is_location_widget(widget.name): resolved_answers[widget.id] = ( self._resolve_widget_option_value( widget, internal_base_location, ) or internal_base_location ) # On-behalf buyback: auto-fill the resignation / last-working-day # date widget from the leaver's HR record when the admin didn't # supply one. The leaver can't fill the form themselves, and the # Feishu template marks this date required — without the backfill # every 代提回购 would 400 on a required-field guard. if on_behalf_person is not None: last_working_day = ( on_behalf_person.termination_date or internal_last_working_day ) for w in widgets: if ( w.type == "date" and _normalise_widget_name(w.name) in _WIDGET_ALIASES_RESIGNATION_DATE_NORMALISED and not resolved_answers.get(w.id) and last_working_day is not None ): resolved_answers[w.id] = last_working_day.isoformat() # laptop_replacement: the 付款金额 amount widget is the buyback # price of the old computer and ONLY applies when the applicant # chose 是否回购旧电脑 = 是. When they didn't, drop the widget so # the (statically required) amount field can't 400 the common # no-buyback path and so no buyback figure leaks onto a # keep-my-old-laptop request — independent of whether the Feishu # template happens to carry a display_condition on it. When they # did, the widget stays and ``_enforce_server_buyback_amount`` # below auto-fills the canonical policy quote. if body.template_key == "laptop_replacement": laptop_buyback_chosen = self._laptop_replacement_buyback_chosen( widgets, resolved_answers ) # Server gate mirroring OFFBOARDING_BUYBACK_REQUIRES_ASSET: an # affirmative laptop buyback MUST reference the asset being # repurchased. Without a linked asset, _enforce_server_buyback_amount # below short-circuits on ``asset is None`` and a client-supplied # ¥1 amount would ship to Feishu unmodified — the exact tamper the # buyback enforcement exists to block. The frontend already requires # the asset; this is the matching backend gate so a curl / MCP / # replay caller that omits asset_id can't skip the server quote. if laptop_buyback_chosen and asset is None: raise AppException( code="LAPTOP_BUYBACK_REQUIRES_ASSET", message=( "Buying back the old computer requires selecting the " "asset to repurchase — submit with its asset_id." ), status_code=400, ) if not laptop_buyback_chosen: widgets = [w for w in widgets if w.type != "amount"] # Trust boundary: overwrite any client-supplied buyback Amount # with the canonical server quote BEFORE the form ships to # Feishu. See ``_enforce_server_buyback_amount`` docstring for # threat model — this is the only thing standing between an # employee and a ¥1 laptop buyback. resolved_answers = await self._enforce_server_buyback_amount( template_key=body.template_key, asset=asset, answers=resolved_answers, widgets=widgets, current_user=current_user, override_currency=internal_buyback_currency, ) form_payload = self._serialize_feishu_form(widgets, resolved_answers) # Human-readable form snapshot — every applicant-visible widget # (input AND static description blocks) paired with the value # actually submitted. This is what the admin drawer renders so # IT can audit "what exact form did this person see and what # did they answer" without having to cross-reference widget-id # → widget-name themselves. ``submitted_answers`` / # ``resolved_answers`` are widget-id keyed and useless for that # forensic question. Keep both for back-compat / raw inspection. # # Conditional widgets (``display_condition``): only persist the # ones that were ACTUALLY visible at submit time given the # answers the applicant supplied. Otherwise an admin reviewing # a "No buyback" laptop_replacement row would see a hidden # ``Whether to upload buyback photo`` entry in the audit # snapshot that the applicant never saw on screen, which is # confusing at best and misleading at worst. Text widgets are # gated too — Feishu admins can attach a display_condition to # description blocks (the ``录视频要求`` notice only shows when # buyback=Yes), so they must be filtered the same way. snapshot_visibility = self._evaluate_display_conditions( widgets, resolved_answers, ) form_snapshot = [ { "id": w.id, "name": w.name, "type": w.type, "required": w.required, "description": w.description, "answer": resolved_answers.get(w.id), } for w in widgets if snapshot_visibility.get(w.id, True) ] request_detail = { "submitted_answers": body.answers, "resolved_answers": resolved_answers, "feishu": { "template_key": body.template_key, "approval_code": config["approval_code"], "approval_name": definition.get("approval_name"), "submitted_form": json.loads(form_payload), "form_snapshot": form_snapshot, }, } # Stash flow-critical fields in the detail blob so the auto- # execute callback path doesn't depend on Feishu echoing them # back via form_items. Two failure modes this defends: # (a) dateInterval (computer_borrow) — the borrow window does # ship to Feishu since 2026-08-14, but the submit path # degrades to omitting the widget if Feishu rejects the # value, so Feishu's echo can still be absent. # (b) Feishu callback form_items occasionally arrives empty / # partial under load. Without a Helix-side copy, the # resignation_date / borrow dates fall through to the # T+30 fallback and the asset return / disposal lands # with the wrong date. if body.template_key == "computer_borrow": interval = next( ( resolved_answers.get(w.id) for w in widgets if w.type == "dateInterval" and isinstance(resolved_answers.get(w.id), dict) ), None, ) if isinstance(interval, dict): start = interval.get("start") end = interval.get("end") if start and end: request_detail["borrow_dates"] = { "start": str(start), "end": str(end), "duration_days": interval.get("duration"), } elif body.template_key == "asset_return": # 2026-05-20: the asset_return form's ``date`` widget is # ``归还日期`` / ``Return Date`` — the day the applicant # plans to return the device. Stash it under ``return_date`` # so the auto-execute path (``_auto_execute_approved_request``) # has the user's intended date even when Feishu's callback # form_items arrives partial under load. Legacy # ``resignation_date`` key is preserved as a fallback for # historical rows captured before this rename landed. # # Iterate ALL matching widgets and take the first non-empty # value — a Feishu template rebuild can leave a legacy # ``Resignation Date`` widget alongside the new # ``归还日期``; if the empty stale one iterates first, a # premature ``break`` would skip the real value. for w in widgets: if w.type == "date" and w.name in WIDGET_ALIASES_ASSET_RETURN_DATE: raw = resolved_answers.get(w.id) if isinstance(raw, str) and raw.strip(): request_detail["return_date"] = raw.strip() break elif body.template_key == "offboarding_buyback": # Resignation date — applicant supplies via a ``date`` # widget. Centralised alias set covers ``离职日期`` / # ``Resignation Date`` (legacy) AND ``Last working day`` # (current Feishu template, 2026-05-20). Same "break only # on non-empty" rule as asset_return above to survive # template-rebuild widget co-existence. for w in widgets: if w.type == "date" and w.name in WIDGET_ALIASES_RESIGNATION_DATE: raw = resolved_answers.get(w.id) if isinstance(raw, str) and raw.strip(): request_detail["resignation_date"] = raw.strip() break # Stash the server-enforced buyback quote as a first-class # detail key (2026-07-03). The canonical figure already lives in # ``resolved_answers`` — but that dict is widget-id keyed, so # list renderers and SQL aggregates (回购汇总 on the hardware # request page) can't reach it without re-loading the template. # ``offboarding_buyback`` always carries an amount widget; # ``laptop_replacement`` only when the applicant chose buyback # (``create_request`` drops the widget otherwise, so this loop # simply finds nothing on the no-buyback path). Historical rows # are backfilled by migration 379 from the form snapshot. if body.template_key in ("offboarding_buyback", "laptop_replacement"): for w in widgets: if w.type == "amount": raw = resolved_answers.get(w.id) if isinstance(raw, dict) and raw.get("amount") not in (None, ""): request_detail["buyback_amount"] = { "amount": str(raw["amount"]), "currency": str(raw.get("currency") or ""), } # There is exactly one amount widget on these templates, # but it is not guaranteed to be first (widget order is # admin-configurable). Break only after we've reached the # amount widget — not after the first (unrelated) widget, # which would silently skip the buyback figure entirely. break # Caller-supplied enrichment (e.g. software_seat resolved # software_name / plan_name / sso_provider) is merged into the # detail blob so list-renderers don't have to rejoin downstream # tables on every page load. ``dict.update`` lets extra_detail # keys overwrite the reserved ones (submitted_answers / # resolved_answers / feishu) on collision — in practice the # only caller (``request_seat``) uses a non-conflicting # ``software_seat`` key, but be aware of the precedence if you # add new callers. if internal_extra_detail: request_detail.update(internal_extra_detail) effective_requester_id = ( internal_on_behalf_of_person_id or current_user.id ) source_payload_data = { "asset_id": str(asset.id) if asset else None, "approval_code": config["approval_code"], "approval_name": definition.get("approval_name"), } if internal_extra_source_data: source_payload_data.update(internal_extra_source_data) return await self._submit_asset_request_with_intent( idempotency_key=key, body=body, current_user=current_user, asset=asset, effective_requester_id=effective_requester_id, request_type=config["request_type"], approval_code=config["approval_code"], feishu_user_id=feishu_user_id, form_payload=form_payload, request_detail=request_detail, source_payload_data=source_payload_data, trace_id=trace_id, ip_address=ip_address, user_agent=user_agent, on_behalf_of_person_id=internal_on_behalf_of_person_id, payload_fingerprint=submission_fingerprint, ) async def create_offboarding_buyback_on_behalf( self, *, asset_id: UUID, current_user: AuthUser, trace_id: str, ip_address: str | None, user_agent: str | None, reason: str | None = None, notes: str | None = None, currency: str | None = None, base_location: str | None = None, last_working_day: date | None = None, idempotency_key: str, ) -> ApprovalBackedEmployeeRequestResponse: """Admin 代提回购 — start the normal offboarding-buyback approval for a leaver-held asset, with the admin as the Feishu applicant. The buyback subject (the leaver) is derived from the asset's current holder, NOT passed in — IT picks an asset, not a person, so there's no way to point the disposal at someone who doesn't actually hold the hardware. Everything else (amount, payment info, owner entity, location, last-working-day) is server-derived; the admin only supplies an optional reason/display currency and may fill location or last-working-day only when authoritative asset/HR data cannot. Reuses the exact same submission path as self-service buyback so the Feishu form, amount-tamper enforcement, and auto-execute-on-approval are identical. """ _asset, beneficiary_id, _beneficiary = ( await self._resolve_offboarding_buyback_subject( asset_id=asset_id, current_user=current_user, ) ) body = AssetApprovalSubmitRequest( template_key="offboarding_buyback", asset_id=asset_id, reason=reason, notes=notes, answers={}, ) return await self.create_asset_approval_request( body=body, current_user=current_user, trace_id=trace_id, ip_address=ip_address, user_agent=user_agent, internal_on_behalf_of_person_id=beneficiary_id, internal_buyback_currency=currency, internal_base_location=base_location, internal_last_working_day=last_working_day, idempotency_key=idempotency_key, ) async def get_offboarding_buyback_on_behalf_context( self, *, asset_id: UUID, current_user: AuthUser, ) -> AdminBuybackOnBehalfContextResponse: """Preflight eligibility and HR date for the asset-admin modal. Keeping this on the specialized, role-gated endpoint avoids exposing termination dates on the general asset list/detail responses. """ _asset, _beneficiary_id, beneficiary = ( await self._resolve_offboarding_buyback_subject( asset_id=asset_id, current_user=current_user, ) ) config = (await self.template_config()).get("offboarding_buyback") if config is None: raise AppException( code="APPROVAL_TEMPLATE_NOT_CONFIGURED", message="Offboarding buyback approval is not configured.", status_code=503, ) definition = await (await self._get_client()).get_approval_definition( config["approval_code"], ) widgets = self._filter_applicant_widgets( "offboarding_buyback", self._parse_template_widgets(definition.get("form", "[]")), ) return AdminBuybackOnBehalfContextResponse( last_working_day=beneficiary.termination_date, widgets=widgets, ) async def _resolve_offboarding_buyback_subject( self, *, asset_id: UUID, current_user: AuthUser, ) -> tuple[AssetItem, UUID, Any]: """Resolve and validate the physical holder for preflight + submit.""" if not can_manage_assets(current_user.role_names): raise ForbiddenError( "Only asset administrators may submit a buyback on behalf " "of a leaver.", ) asset = await self._get_asset(asset_id) # The leaver is whoever still holds the asset. ``held_by_leaver`` # preserves ``assigned_to``; a borrowed leaver device keeps # ``borrowed_by``. Either is a valid buyback subject. # Borrower wins because assigned_to remains the original owner while # borrowed_by is the current physical holder. beneficiary_id = asset.borrowed_by or asset.assigned_to if beneficiary_id is None: raise AppException( code="BUYBACK_ON_BEHALF_NO_HOLDER", message=( "This asset has no current holder — there is no leaver " "to buy it back. 代提回购 only applies to assets still " "held by a departed employee." ), status_code=400, ) if asset.status != ASSET_STATUS_HELD_BY_LEAVER: raise AppException( code="ASSET_NOT_HELD_BY_LEAVER", message=( "On-behalf buyback is only valid for an asset currently " "marked held_by_leaver." ), status_code=400, ) from app.identity.person_models import IdentityPerson # noqa: PLC0415 beneficiary = await self.db.get(IdentityPerson, beneficiary_id) if beneficiary is None or beneficiary.employment_status not in { "left", "leaver_pending", }: raise AppException( code="BUYBACK_ON_BEHALF_HOLDER_NOT_LEAVER", message=( "The current holder is not a departed or pending-leaver " "employee. Mark the correct leaver-held asset first." ), status_code=400, ) return asset, beneficiary_id, beneficiary @staticmethod def _approval_submission_fingerprint(payload: dict[str, Any]) -> str: canonical = json.dumps( jsonable_encoder(payload), ensure_ascii=False, sort_keys=True, separators=(",", ":"), ) return hashlib.sha256(canonical.encode("utf-8")).hexdigest() async def _submission_intent_response( self, intent: ApprovalSubmissionIntent, ) -> ApprovalBackedEmployeeRequestResponse: if intent.employee_request_id is None: raise AppException( code="APPROVAL_SUBMISSION_RECOVERY_REQUIRED", message=( "The approval submission is recorded but not fully " "reconciled yet. Retry later with the same Idempotency-Key." ), status_code=409, ) row = await self.db.get(AssetEmployeeRequest, intent.employee_request_id) if row is None: raise RuntimeError( "completed approval submission intent has no employee request" ) requester = await self.db.get(AuthUser, row.requester_id) requesters = {row.requester_id: requester} if requester else {} asset = await self.db.get(AssetItem, row.asset_id) if row.asset_id else None assets = {row.asset_id: asset} if asset and row.asset_id else {} return self._to_request_response(row, requesters, assets) async def _finalize_submission_intent( self, intent_id: UUID, *, recovered: bool, ) -> ApprovalSubmissionIntent: intent = await self.db.scalar( select(ApprovalSubmissionIntent) .where(ApprovalSubmissionIntent.id == intent_id) .with_for_update(), ) if intent is None: raise RuntimeError("approval submission intent disappeared") if intent.state == "completed": return intent if intent.state != "provider_created" or not intent.external_id: raise RuntimeError( f"cannot finalize approval submission intent in {intent.state!r}" ) payload = dict(intent.payload or {}) employee_request_id = UUID(payload["employee_request_id"]) approval_request_id = UUID(payload["approval_request_id"]) existing = await self.db.get(AssetEmployeeRequest, employee_request_id) if existing is None: detail = dict(payload["request_detail"]) feishu_detail = dict(detail.get("feishu") or {}) provider_response = dict(payload.get("provider_response") or {}) provider_status = ( provider_response.get("instance_status") or provider_response.get("status") or "PENDING" ) feishu_detail.update({ "instance_code": intent.external_id, "instance_status": provider_status, }) detail["feishu"] = feishu_detail employee_request = AssetEmployeeRequest( id=employee_request_id, request_number=payload["request_number"], request_type=payload["request_type"], # A recovered provider response may already be terminal. # Materialise the correlation rows as pending first; the # reconciler then replays the canonical callback dispatcher # so domain side effects are never skipped by a pre-terminal # parent row. status=( "pending" if recovered else self._map_feishu_status(provider_status) ), requester_id=UUID(payload["effective_requester_id"]), asset_id=( UUID(payload["asset_id"]) if payload.get("asset_id") else None ), reason=payload.get("reason"), detail=detail, notes=payload.get("notes"), ) self.db.add(employee_request) await self.db.flush() source_payload = dict(payload["source_payload_data"]) source_payload["employee_request_id"] = str(employee_request_id) self.db.add(ApprovalRequest( id=approval_request_id, external_id=intent.external_id, approval_type=payload["template_key"], applicant_id=intent.applicant_id, status=employee_request.status, source_payload=source_payload, )) await self.db.flush() intent.employee_request_id = employee_request_id intent.approval_request_id = approval_request_id intent.state = "completed" intent.completed_at = datetime.now(UTC) intent.last_error = None await AuditService(self.db).log( trace_id=( f"approval-intent-reconcile:{intent.id}" if recovered else payload["trace_id"] ), actor_type="system" if recovered else "user", actor_user_id=(None if recovered else intent.applicant_id), action=( "approval.submission_intent.reconciled" if recovered else "approval.asset_request.create" ), resource_type="employee_request", resource_id=str(employee_request_id), status="success", detail={ "intent_id": str(intent.id), "request_number": payload["request_number"], "request_type": payload["request_type"], "approval_instance_code": intent.external_id, "provider_request_uuid": intent.provider_request_uuid, "recovered": recovered, **( { "on_behalf_of_person_id": payload[ "on_behalf_of_person_id" ], "submitted_by_admin_id": str(intent.applicant_id), } if payload.get("on_behalf_of_person_id") else {} ), }, ip_address=payload.get("ip_address"), user_agent=payload.get("user_agent"), ) compact_terminal_intent_payload( intent, resource_type="employee_request", resource_id=employee_request_id, ) await self.db.commit() await self.db.refresh(intent) return intent async def _submit_asset_request_with_intent( self, *, idempotency_key: str, body: AssetApprovalSubmitRequest, current_user: AuthUser, asset: AssetItem | None, effective_requester_id: UUID, request_type: str, approval_code: str, feishu_user_id: str, form_payload: str, request_detail: dict[str, Any], source_payload_data: dict[str, Any], trace_id: str, ip_address: str | None, user_agent: str | None, on_behalf_of_person_id: UUID | None, payload_fingerprint: str, ) -> ApprovalBackedEmployeeRequestResponse: key = normalize_approval_idempotency_key(idempotency_key) applicant_id = current_user.id intent_payload = { "template_key": body.template_key, "request_type": request_type, "request_number": self._generate_request_number(), "employee_request_id": str(uuid4()), "approval_request_id": str(uuid4()), "effective_requester_id": str(effective_requester_id), "asset_id": str(asset.id) if asset else None, "reason": body.reason, "notes": body.notes, "approval_code": approval_code, "feishu_user_id": feishu_user_id, "form_payload": form_payload, "request_detail": request_detail, "source_payload_data": source_payload_data, "trace_id": trace_id, "ip_address": ip_address, "user_agent": user_agent, "on_behalf_of_person_id": ( str(on_behalf_of_person_id) if on_behalf_of_person_id else None ), } fingerprint = payload_fingerprint intent = await self.db.scalar( select(ApprovalSubmissionIntent) .where( ApprovalSubmissionIntent.scope == "asset_request", ApprovalSubmissionIntent.applicant_id == applicant_id, ApprovalSubmissionIntent.idempotency_key == key, ) .with_for_update(), ) created = False if intent is None: candidate = ApprovalSubmissionIntent( scope="asset_request", applicant_id=applicant_id, idempotency_key=key, payload_fingerprint=fingerprint, provider_request_uuid=str(uuid4()), state="prepared", payload=jsonable_encoder(intent_payload), ) try: async with self.db.begin_nested(): self.db.add(candidate) await self.db.flush() intent = candidate created = True except IntegrityError: intent = await self.db.scalar( select(ApprovalSubmissionIntent) .where( ApprovalSubmissionIntent.scope == "asset_request", ApprovalSubmissionIntent.applicant_id == applicant_id, ApprovalSubmissionIntent.idempotency_key == key, ) .with_for_update(), ) if intent is None: raise AppException( code="APPROVAL_SUBMISSION_IN_PROGRESS", message="This approval submission is still in progress.", status_code=409, ) from None if intent.payload_fingerprint != fingerprint: raise AppException( code="APPROVAL_IDEMPOTENCY_PAYLOAD_MISMATCH", message=( "This Idempotency-Key was already used with a different " "approval payload." ), status_code=409, ) if intent.state == "completed": return await self._submission_intent_response(intent) if intent.state == "provider_created": intent = await self._finalize_submission_intent( intent.id, recovered=True, ) return await self._submission_intent_response(intent) retry_at = intent.next_attempt_at if intent.state == "unknown" and retry_at is not None: if retry_at.tzinfo is None: retry_at = retry_at.replace(tzinfo=UTC) if retry_at > datetime.now(UTC): raise AppException( code="APPROVAL_SUBMISSION_IN_PROGRESS", message=( "This approval submission is awaiting provider " "reconciliation. Retry later with the same key." ), status_code=409, ) if not created and intent.state not in { "prepared", "unknown", "failed", }: raise AppException( code="APPROVAL_SUBMISSION_IN_PROGRESS", message=( "This approval submission is in progress or awaiting " "provider reconciliation. Retry later with the same key." ), status_code=409, ) if intent.state == "failed": # A definite provider rejection has no surviving recovery payload. # Rehydrate it from this fingerprint-matched user retry before the # same provider UUID is attempted again. intent.payload = jsonable_encoder(intent_payload) lookup_anchors = provider_attempt_anchors( dict(intent.payload or {}), fallback=intent.created_at, ) existing_anchors = list( (intent.payload or {}).get("provider_attempt_anchors") or [] ) was_lookup_only = bool( (intent.payload or {}).get(_ASSET_PROVIDER_LOOKUP_ONLY) ) if len(existing_anchors) >= _ASSET_PROVIDER_ANCHOR_LIMIT: payload = dict(intent.payload or {}) payload[_ASSET_PROVIDER_LOOKUP_ONLY] = True intent.payload = payload if not was_lookup_only: logger.error( "approval submission intent %s reached the provider attempt " "cap; switching to bounded lookup-only recovery", intent.id, ) elif not (intent.payload or {}).get(_ASSET_PROVIDER_LOOKUP_ONLY): record_provider_attempt_anchor(intent) stored_payload = dict(intent.payload or {}) intent.state = "submitting" intent.attempt_count += 1 attempt_token = intent.attempt_count intent.last_error = None intent.next_attempt_at = datetime.now(UTC) + timedelta( minutes=_ASSET_PROVIDER_ATTEMPT_LEASE_MINUTES, ) intent_id = intent.id provider_uuid = intent.provider_request_uuid await self.db.commit() # durable BEFORE the external side effect try: instance = await self._create_or_recover_asset_provider_instance( client=await self._get_client(), payload=stored_payload, provider_uuid=provider_uuid, attempt_anchors=lookup_anchors, ) except AppException as exc: unknown = exc.code == "APPROVAL_SUBMISSION_RESULT_UNKNOWN" cause = exc.__cause__ if unknown and exc.__cause__ is not None else exc recorded = await self._record_asset_provider_failure( intent_id, cause, unknown=unknown, trace_id=trace_id, actor_type="user", ip_address=ip_address, user_agent=user_agent, recovery_payload=stored_payload, expected_attempt_count=attempt_token, ) if not recorded: current = await self.db.get(ApprovalSubmissionIntent, intent_id) if current is not None and current.state == "completed": return await self._submission_intent_response(current) raise await self.db.rollback() intent = await self.db.scalar( select(ApprovalSubmissionIntent) .where(ApprovalSubmissionIntent.id == intent_id) .with_for_update() .execution_options(populate_existing=True) ) if intent is None: raise RuntimeError("approval submission intent disappeared") if intent.state != "submitting" or intent.attempt_count != attempt_token: if intent.state == "completed": completed_intent_id = intent.id await self.db.rollback() completed_intent = await self.db.get( ApprovalSubmissionIntent, completed_intent_id, ) if completed_intent is None: raise RuntimeError("approval submission intent disappeared") return await self._submission_intent_response(completed_intent) await self.db.rollback() raise AppException( code="APPROVAL_SUBMISSION_IN_PROGRESS", message="A newer provider recovery attempt owns this submission.", status_code=409, ) payload = dict(intent.payload or {}) payload["provider_response"] = jsonable_encoder(instance) intent.payload = payload intent.external_id = instance["instance_code"] intent.state = "provider_created" intent.next_attempt_at = None await self.db.commit() # external success is now durably correlated intent = await self._finalize_submission_intent(intent_id, recovered=False) return await self._submission_intent_response(intent) @staticmethod def _asset_submission_unknown_error() -> AppException: return AppException( code="APPROVAL_SUBMISSION_RESULT_UNKNOWN", message=( "Feishu may have accepted this approval. Helix retained the " "submission UUID; retry the same Idempotency-Key to reconcile it." ), status_code=503, ) async def _create_instance_with_form_fallback( self, *, client: FeishuApprovalClient, payload: dict[str, Any], provider_uuid: str, ) -> dict[str, Any]: """Create the Feishu instance, degrading the form once if the borrow window is rejected. The ``dateInterval`` value shape shipped by ``_date_interval_wire_value`` is read off live Feishu, but it is the one widget in the asset flows whose accepted write shape we could not exercise end-to-end without submitting a real approval into a production flow. If Feishu rejects the form with a value-validation error, retry ONCE with the interval widget stripped — the shape that shipped between 2026-05-19 and 2026-08-14, which Feishu is known to accept. That makes the worst case "approver's card is missing the dates" (today's behaviour) rather than "nobody can submit a borrow request", which is exactly how the 2026-08-04 amount-widget regression managed to block every non-CNY buyback for nine days. The rejected attempt creates nothing on Feishu's side, so the retry reuses ``provider_uuid`` and the intent's idempotency contract is unchanged. ``payload["form_payload"]`` is updated in place: on the failure path the caller persists that same dict as ``recovery_payload``, so a recovery tick replays the degraded form. On the success path there is no later tick to replay — the instance exists. """ try: return await client.create_instance( approval_code=str(payload["approval_code"]), user_id=str(payload["feishu_user_id"]), form=str(payload["form_payload"]), instance_uuid=provider_uuid, ) except AppException as exc: if parse_feishu_error_code(exc) not in FEISHU_FORM_VALUE_ERROR_CODES: raise # Only degrade when Feishu's ``控件= `` names the # interval widget — a rejection caused by some other widget # stays terminal rather than burning an identical retry. if not rejection_targets_date_interval( str(payload["form_payload"]), exc.message or "", ): raise fallback_form = self._form_without_date_interval( str(payload["form_payload"]), ) if fallback_form is None: raise logger.warning( "Feishu rejected the dateInterval value for approval %s " "(%s); resubmitting without the borrow window. The " "approver's card will not show the dates — the window " "is still on EmployeeRequest.detail.borrow_dates.", payload.get("approval_code"), exc.message, ) payload["form_payload"] = fallback_form return await client.create_instance( approval_code=str(payload["approval_code"]), user_id=str(payload["feishu_user_id"]), form=str(payload["form_payload"]), instance_uuid=provider_uuid, ) async def _create_or_recover_asset_provider_instance( self, *, client: FeishuApprovalClient, payload: dict[str, Any], provider_uuid: str, attempt_anchors: tuple[datetime, ...], ) -> dict[str, Any]: """Create once, or resolve the same UUID after an ambiguous response.""" instance: dict[str, Any] | None lookup_only = bool(payload.get(_ASSET_PROVIDER_LOOKUP_ONLY)) if not lookup_only: try: instance = await self._create_instance_with_form_fallback( client=client, payload=payload, provider_uuid=provider_uuid, ) except AppException as exc: if parse_feishu_error_code(exc) != 60012: # A parsed Feishu business error is a definite rejection. It is # the only create-stage exception safe to classify terminal. raise # This attempt only observed the collision; it cannot be the # generation that created the provider instance. Remove its # anchor and switch future ticks to lookup-only recovery. anchors = list(payload.get("provider_attempt_anchors") or []) if anchors: anchors.pop() payload["provider_attempt_anchors"] = anchors payload[_ASSET_PROVIDER_LOOKUP_ONLY] = True lookup_only = True except Exception as exc: # JSON decoding, response validation and client post-processing can # fail after Feishu accepted the POST. Preserve the UUID generation. raise self._asset_submission_unknown_error() from exc if lookup_only: try: instance = None cursor = int(payload.get(_ASSET_PROVIDER_LOOKUP_CURSOR) or 0) if attempt_anchors: cursor %= len(attempt_anchors) selected = ( attempt_anchors[cursor:cursor + _ASSET_PROVIDER_LOOKUP_LIMIT] or attempt_anchors[:_ASSET_PROVIDER_LOOKUP_LIMIT] ) for anchor in selected: instance = await client.find_instance_by_uuid( approval_code=str(payload["approval_code"]), instance_uuid=provider_uuid, start_time=anchor - timedelta(minutes=5), end_time=min( anchor + timedelta(minutes=5), datetime.now(UTC) + timedelta(minutes=1), ), ) if instance is not None: break if instance is None and attempt_anchors: payload[_ASSET_PROVIDER_LOOKUP_CURSOR] = ( cursor + len(selected) ) % len(attempt_anchors) except Exception as lookup_exc: raise self._asset_submission_unknown_error() from lookup_exc if instance is None: miss = RuntimeError( "Feishu UUID collision was not visible in the bounded window" ) raise self._asset_submission_unknown_error() from miss try: encoded = jsonable_encoder(instance) external_id = str((encoded or {}).get("instance_code") or "").strip() except Exception as exc: raise self._asset_submission_unknown_error() from exc if not external_id: missing = RuntimeError("Feishu create response omitted instance_code") raise self._asset_submission_unknown_error() from missing return dict(encoded) async def _record_asset_provider_failure( self, intent_id: UUID, exc: BaseException, *, unknown: bool, trace_id: str, actor_type: str, ip_address: str | None, user_agent: str | None, recovery_payload: dict[str, Any] | None = None, expected_attempt_count: int, ) -> bool: await self.db.rollback() intent = await self.db.scalar( select(ApprovalSubmissionIntent) .where(ApprovalSubmissionIntent.id == intent_id) .with_for_update() .execution_options(populate_existing=True) ) if intent is None: return False if ( intent.state != "submitting" or intent.attempt_count != expected_attempt_count ): await self.db.rollback() return False if recovery_payload is not None: intent.payload = jsonable_encoder(recovery_payload) intent.state = "unknown" if unknown else "failed" intent.last_error = sanitize_log_text(exc, max_len=512) intent.next_attempt_at = ( datetime.now(UTC) + timedelta(minutes=1) if unknown else None ) await AuditService(self.db).log( trace_id=trace_id, actor_type=actor_type, actor_user_id=intent.applicant_id if actor_type == "user" else None, action=( "approval.submission_intent.unknown" if unknown else "approval.submission_intent.failed" ), resource_type="approval_submission_intent", resource_id=str(intent_id), status="failure", detail={ "provider_request_uuid": intent.provider_request_uuid, "state": intent.state, "error_type": type(exc).__name__, }, ip_address=ip_address, user_agent=user_agent, ) if not unknown: compact_terminal_intent_payload( intent, resource_type="approval_submission", resource_id=intent.id, ) await self.db.commit() return True async def _defer_asset_reconcile_failure( self, intent_id: UUID, exc: BaseException, *, expected_attempt_count: int, ) -> bool: """Fence a reconcile failure so a late worker cannot defer a new lease.""" await self.db.rollback() intent = await self.db.scalar( select(ApprovalSubmissionIntent) .where(ApprovalSubmissionIntent.id == intent_id) .with_for_update() .execution_options(populate_existing=True) ) if ( intent is None or intent.attempt_count != expected_attempt_count or intent.state not in { "prepared", "submitting", "unknown", "provider_created", } ): await self.db.rollback() return False delay_minutes = min( 2 ** min(max(intent.attempt_count - 1, 0), 6), 60, ) intent.next_attempt_at = datetime.now(UTC) + timedelta( minutes=delay_minutes, ) intent.last_error = sanitize_log_text(exc, max_len=512) await self.db.commit() return True async def _asset_attempt_completed_before_followup_failure( self, intent_id: UUID, *, expected_attempt_count: int, ) -> bool: """Detect this attempt's post-completion callback/replay failure.""" await self.db.rollback() intent = await self.db.get(ApprovalSubmissionIntent, intent_id) return bool( intent is not None and intent.state == "completed" and intent.attempt_count == expected_attempt_count ) async def reconcile_submission_intents( self, *, min_age_minutes: int = 2, batch_limit: int = 100, ) -> dict[str, int]: """Recover durable intents left around the Feishu create boundary. Unknown/submitting rows retry create with the persisted Feishu UUID. A 60012 collision is resolved through Feishu's documented bounded approval/time list followed by detail lookups using instance codes. Provider-created rows only need the local atomic finalization replay. """ from datetime import timedelta now = datetime.now(UTC) cutoff = now - timedelta(minutes=min_age_minutes) lease_cutoff = now - timedelta( minutes=_ASSET_PROVIDER_ATTEMPT_LEASE_MINUTES, ) intent_ids = list((await self.db.scalars( select(ApprovalSubmissionIntent.id) .where( ApprovalSubmissionIntent.scope == "asset_request", ApprovalSubmissionIntent.state.in_(( "prepared", "submitting", "unknown", "provider_created", )), or_( ApprovalSubmissionIntent.state == "provider_created", ApprovalSubmissionIntent.updated_at < cutoff, ), reconciliation_due_predicate(datetime.now(UTC)), ) .order_by(*reconciliation_order()) .limit(batch_limit) )).all()) await self.db.rollback() summary = { "checked": len(intent_ids), "reconciled": 0, "still_unknown": 0, "failed": 0, } if not intent_ids: return summary client = await self._get_client() for intent_id in intent_ids: attempt_token: int | None = None # The body commits mid-flight (3×), so a SAVEPOINT would be # released by the first commit and the block exit would then # operate on a freed one. The handlers below already recover the # session themselves — ``_defer_asset_reconcile_failure`` opens # with ``await self.db.rollback()``. Verified 2026-08-06: wrapping # this in ``best_effort`` broke 13 idempotency tests. # savepoint-exempt: commits mid-flight; handlers already rollback try: instance_for_replay: dict[str, Any] | None = None intent = await self.db.scalar( select(ApprovalSubmissionIntent) .where(ApprovalSubmissionIntent.id == intent_id) .with_for_update() .execution_options(populate_existing=True) ) if intent is None or intent.state == "completed": await self.db.rollback() continue attempt_token = intent.attempt_count if intent.state != "provider_created": updated_at = intent.updated_at if updated_at.tzinfo is None: updated_at = updated_at.replace(tzinfo=UTC) retry_at = intent.next_attempt_at if retry_at is not None and retry_at.tzinfo is None: retry_at = retry_at.replace(tzinfo=UTC) fresh_submitting = ( intent.state == "submitting" and updated_at >= lease_cutoff ) too_young = ( intent.state in {"prepared", "unknown"} and updated_at >= cutoff ) retry_not_due = retry_at is not None and retry_at > datetime.now(UTC) if ( intent.state not in {"prepared", "submitting", "unknown"} or fresh_submitting or too_young or retry_not_due ): await self.db.rollback() continue if intent.state != "provider_created": payload = dict(intent.payload or {}) lookup_anchors = provider_attempt_anchors( payload, fallback=intent.created_at, ) anchors = list(payload.get("provider_attempt_anchors") or []) was_lookup_only = bool( payload.get(_ASSET_PROVIDER_LOOKUP_ONLY) ) if len(anchors) >= _ASSET_PROVIDER_ANCHOR_LIMIT: payload[_ASSET_PROVIDER_LOOKUP_ONLY] = True intent.payload = payload if not was_lookup_only: logger.error( "approval submission intent %s reached the provider " "attempt cap; switching to bounded lookup-only recovery", intent.id, ) elif not payload.get(_ASSET_PROVIDER_LOOKUP_ONLY): record_provider_attempt_anchor(intent) payload = dict(intent.payload or {}) intent.state = "submitting" intent.attempt_count += 1 attempt_token = intent.attempt_count intent.last_error = None intent.next_attempt_at = datetime.now(UTC) + timedelta( minutes=_ASSET_PROVIDER_ATTEMPT_LEASE_MINUTES, ) await self.db.commit() try: instance = ( await self._create_or_recover_asset_provider_instance( client=client, payload=payload, provider_uuid=intent.provider_request_uuid, attempt_anchors=lookup_anchors, ) ) except AppException as exc: if exc.code == "APPROVAL_SUBMISSION_RESULT_UNKNOWN": cause = exc.__cause__ or exc recorded = await self._record_asset_provider_failure( intent_id, cause, unknown=True, trace_id=( "approval.submission_intent.reconcile:" f"{intent_id}" ), actor_type="system", ip_address=None, user_agent=None, recovery_payload=payload, expected_attempt_count=attempt_token, ) if not recorded: continue summary["still_unknown"] += 1 logger.warning( "approval submission intent reconcile deferred " "for %s: %s", intent_id, sanitize_log_text(exc, max_len=512), ) continue recorded = await self._record_asset_provider_failure( intent_id, exc, unknown=False, trace_id=( "approval.submission_intent.reconcile:" f"{intent_id}" ), actor_type="system", ip_address=None, user_agent=None, recovery_payload=payload, expected_attempt_count=attempt_token, ) if not recorded: continue summary["failed"] += 1 logger.warning( "approval submission intent reconcile rejected " "for %s: %s", intent_id, sanitize_log_text(exc, max_len=512), ) continue await self.db.rollback() intent = await self.db.scalar( select(ApprovalSubmissionIntent) .where(ApprovalSubmissionIntent.id == intent_id) .with_for_update() .execution_options(populate_existing=True) ) if ( intent is None or intent.state != "submitting" or intent.attempt_count != attempt_token ): await self.db.rollback() continue external_id = instance["instance_code"] payload["provider_response"] = jsonable_encoder(instance) instance_for_replay = instance intent.payload = payload intent.external_id = str(external_id) intent.state = "provider_created" intent.last_error = None intent.next_attempt_at = None await self.db.commit() await self._finalize_submission_intent( intent_id, recovered=True, ) if ( instance_for_replay is not None and str(instance_for_replay.get("status") or "").upper() in {"APPROVED", "COMPLETED", "REJECTED", "CANCELED", "CANCELLED", "TIMEOUT"} ): await self.replay_callback_from_feishu( instance_code=str(instance_for_replay["instance_code"]), actor_user_id=None, prefetched_instance_detail=instance_for_replay, trace_id_prefix="approval.submission_intent", ) await self.db.commit() summary["reconciled"] += 1 except AppException as exc: if attempt_token is None: continue deferred = await self._defer_asset_reconcile_failure( intent_id, exc, expected_attempt_count=attempt_token, ) if not deferred and await self._asset_attempt_completed_before_followup_failure( intent_id, expected_attempt_count=attempt_token, ): summary["failed"] += 1 logger.error( "approval submission intent %s completed but terminal " "callback replay failed (%s): %s", intent_id, type(exc).__name__, sanitize_log_text(exc, max_len=512), ) continue if not deferred: continue summary["still_unknown"] += 1 logger.warning( "approval submission intent reconcile deferred for %s: %s", intent_id, sanitize_log_text(exc, max_len=512), ) except Exception as exc: # noqa: BLE001 if attempt_token is None: continue deferred = await self._defer_asset_reconcile_failure( intent_id, exc, expected_attempt_count=attempt_token, ) if not deferred and await self._asset_attempt_completed_before_followup_failure( intent_id, expected_attempt_count=attempt_token, ): summary["failed"] += 1 logger.error( "approval submission intent %s completed but terminal " "callback replay failed (%s): %s", intent_id, type(exc).__name__, sanitize_log_text(exc, max_len=512), ) continue if not deferred: continue summary["failed"] += 1 logger.error( "approval submission intent reconcile failed for %s " "(%s): %s", intent_id, type(exc).__name__, sanitize_log_text(exc, max_len=512), ) return summary @staticmethod def _is_data_access_approval_type(approval_type: str) -> bool: from app.domains.data_access import constants as data_access_constants return approval_type in ( data_access_constants.APPROVAL_FLOW_DATA_ACCESS_GROUP_GRANT, data_access_constants.APPROVAL_FLOW_DATA_ACCESS_SERVICE_ACCOUNT_GRANT, data_access_constants.APPROVAL_FLOW_DATA_ACCESS_PERSONAL_GRANT, ) async def _data_access_callback_needs_recovery( self, approval_request: ApprovalRequest, ) -> bool: if not self._is_data_access_approval_type(approval_request.approval_type): return False from app.domains.data_access.models import ( REQUEST_STATUS_PENDING_FEISHU, DataAccessRequest, ) request_status = await self.db.scalar( select(DataAccessRequest.status).where( DataAccessRequest.approval_request_id == approval_request.id, ), ) return request_status == REQUEST_STATUS_PENDING_FEISHU async def handle_feishu_callback( self, payload: dict[str, Any], *, prefetched_instance_detail: dict[str, Any] | None = None, ) -> None: """Dispatch a Feishu approval callback to the correct domain handler. ``prefetched_instance_detail`` (added 2026-05-28 for the ``replay_callback_from_feishu`` reconcile path) lets the caller skip the ``get_instance_detail`` round-trip when it has already fetched the live snapshot — e.g. when the admin "刷新流程" button just pulled it for display, or when the ``approval_pending_reaper`` cron already fetched it for the terminal-state check. Webhook delivery callers leave it ``None`` so the lookup-by-instance_code path is unchanged. """ instance_code = ( payload.get("event", {}).get("instance_code") or payload.get("event", {}).get("approval_instance_code") or payload.get("instance_code") ) if not instance_code: logger.warning( "Feishu callback has no instance_code; payload keys=%s event keys=%s", list(payload.keys()), list((payload.get("event") or {}).keys()) if isinstance(payload.get("event"), dict) else None, ) return # Issue #49 — return silently if the request hasn't been committed yet; # Feishu will retry on non-200, but raising NotFoundError would cause a 500. # # Acquire a row-level lock (SELECT ... FOR UPDATE) so a rapid Feishu # redelivery cannot race with an in-flight callback: both would # otherwise observe status='pending', pass the idempotency guard # below, and double-apply the domain action (software seat grant, # asset mutation, etc). The lock is held until transaction commit. result = await self.db.execute( select(ApprovalRequest) .where(ApprovalRequest.external_id == instance_code) .with_for_update() ) approval_request = result.scalar_one_or_none() # When ``synthesized_detail`` is set the synthesizer already # called ``get_instance_detail`` and that snapshot is what # populated ``source_payload.latest_instance_detail`` on the # new row — re-fetching below would (a) waste an HTTP call # and (b) risk operating on a fresher snapshot than the # synthesized row recorded. synthesized_detail: dict[str, Any] | None = None if approval_request is None: # Externally-initiated flow path (PR #256). A flow whose # ``integration_approval_flows.accept_external_origin = # TRUE`` is allowed to enter the dispatcher even though # Helix never created the parent ``approval_requests`` # row (the user submitted directly in Feishu). We must # fetch instance_detail FIRST to learn the approval_code # (template UUID), look up the flow row, and synthesize # the parent. Default behavior for every other flow # remains: silent drop, no synthesis. synthesis = await self._synthesize_external_origin_request( instance_code=instance_code, payload=payload, ) if synthesis is None: logger.warning( "Feishu callback for unknown instance_code %s — ignoring", instance_code, ) return approval_request, synthesized_detail = synthesis logger.info( "feishu callback processing: instance=%s approval_type=%s current_status=%s", instance_code, approval_request.approval_type, approval_request.status, ) # Issue #50 — idempotency guard: skip re-processing if already settled. # Round-22 review: ``cancelled`` and ``completed`` are also terminal # states emitted by ``_map_feishu_status`` and a redelivered # webhook for either must NOT re-run domain side effects (e.g. # writing duplicate audit rows, re-invoking dispatch). Without # them the dispatcher would re-enter, no-op the side effect # because every existing service branches on # ``approval_status == "approved"`` for the actual work, but # would emit a second ``approval..cancelled`` audit row # and (for SOC2) re-fetch instance_detail. # ``"completed"`` dropped from the terminal set: it can no longer be # produced (COMPLETED→"approved" now), and a legacy "completed" row was # sealed without side-effects — re-processing rescues it (idempotent # per-domain handlers make a genuinely-done re-process a safe no-op). if approval_request.status in ("approved", "rejected", "cancelled"): if not await self._data_access_callback_needs_recovery( approval_request, ): logger.info( "Feishu callback duplicate for instance_code %s — already processed, skipping", instance_code, ) return logger.warning( "recovering data-access callback drift: instance=%s " "approval_type=%s parent_status=%s", instance_code, approval_request.approval_type, approval_request.status, ) if prefetched_instance_detail is not None: # Reconcile / reaper path — caller already fetched the live # snapshot. Skip the second round-trip so a backlog of # reaper-replays doesn't multiply Feishu API load. instance_detail = prefetched_instance_detail elif synthesized_detail is not None: instance_detail = synthesized_detail else: instance_detail = await (await self._get_client()).get_instance_detail(instance_code) # Dispatch to domain-specific handler based on approval_type if self._is_data_access_approval_type(approval_request.approval_type): from app.domains.data_access.service import DataAccessService approval_status = self._map_feishu_status( instance_detail.get("status"), ) await DataAccessService().handle_feishu_callback( self.db, approval_request=approval_request, instance_detail=instance_detail, approval_status=approval_status, trace_id=f"data_access_callback:{instance_code}", ) await self.db.flush() logger.info( "data-access Feishu callback applied: instance=%s status=%s", instance_code, instance_detail.get("status"), ) return if approval_request.approval_type == "api_key_request": from app.domains.apikey.service import ApiKeyService apikey_svc = ApiKeyService(self.db) await apikey_svc.handle_feishu_callback(instance_code, instance_detail) await self.db.flush() return # ``api_key_release`` previously routed here for the Feishu # approval-template path; the flow was rewritten to be # Helix-internal (POST /apikeys/release-requests/{id}/approve) # so there is no Feishu instance + no callback. Any stale # ``approval_type='api_key_release'`` rows from before the # cutover land in the generic ``approval/feishu_callback_router`` # dispatch and are no-ops here. if approval_request.approval_type == "domain_registration_request": from app.domains.domain_registration.service import ( DomainRegistrationService, ) approval_status = self._map_feishu_status( instance_detail.get("status"), ) approval_request.status = approval_status source_payload = dict(approval_request.source_payload or {}) source_payload["latest_instance_detail"] = instance_detail approval_request.source_payload = source_payload approval_request.updated_at = datetime.now(UTC) dr_svc = DomainRegistrationService(self.db) await dr_svc.handle_feishu_callback( approval_request=approval_request, instance_detail=instance_detail, ) audit = AuditService(self.db) await audit.log( trace_id=f"feishu_callback:{instance_code}", actor_type="integration", actor_user_id=None, action=f"approval.domain_registration_request.{approval_status}", status="success", resource_type="approval_request", resource_id=str(approval_request.id), detail={ "instance_code": instance_code, "approval_type": approval_request.approval_type, "feishu_status": instance_detail.get("status"), }, ) return if approval_request.approval_type == "identity_consumer_request": from app.identity.consumers.request_service import ( ConsumerRequestService, ) approval_status = self._map_feishu_status( instance_detail.get("status"), ) approval_request.status = approval_status source_payload = dict(approval_request.source_payload or {}) source_payload["latest_instance_detail"] = instance_detail approval_request.source_payload = source_payload approval_request.updated_at = datetime.now(UTC) cr_svc = ConsumerRequestService(self.db) await cr_svc.handle_feishu_callback( approval_request=approval_request, instance_detail=instance_detail, ) audit = AuditService(self.db) await audit.log( trace_id=f"feishu_callback:{instance_code}", actor_type="integration", actor_user_id=None, action=f"approval.identity_consumer_request.{approval_status}", status="success", resource_type="approval_request", resource_id=str(approval_request.id), detail={ "instance_code": instance_code, "approval_type": approval_request.approval_type, "feishu_status": instance_detail.get("status"), }, ) return if approval_request.approval_type == "sso_grant_request": from app.domains.sso.service import SsoGrantService approval_status = self._map_feishu_status( instance_detail.get("status"), ) approval_request.status = approval_status source_payload = dict(approval_request.source_payload or {}) source_payload["latest_instance_detail"] = instance_detail approval_request.source_payload = source_payload approval_request.updated_at = datetime.now(UTC) sso_svc = SsoGrantService(self.db) await sso_svc.handle_feishu_callback( approval_request=approval_request, instance_detail=instance_detail, ) audit = AuditService(self.db) await audit.log( trace_id=f"feishu_callback:{instance_code}", actor_type="integration", actor_user_id=None, action=f"approval.sso_grant_request.{approval_status}", status="success", resource_type="approval_request", resource_id=str(approval_request.id), detail={ "instance_code": instance_code, "approval_type": approval_request.approval_type, "feishu_status": instance_detail.get("status"), }, ) return if approval_request.approval_type == "vpn_application_request": from app.domains.vpn_application.service import ( VpnApplicationService, ) approval_status = self._map_feishu_status( instance_detail.get("status"), ) approval_request.status = approval_status source_payload = dict(approval_request.source_payload or {}) source_payload["latest_instance_detail"] = instance_detail approval_request.source_payload = source_payload approval_request.updated_at = datetime.now(UTC) vpn_svc = VpnApplicationService(self.db) await vpn_svc.handle_feishu_callback( approval_request=approval_request, instance_detail=instance_detail, ) audit = AuditService(self.db) await audit.log( trace_id=f"feishu_callback:{instance_code}", actor_type="integration", actor_user_id=None, action=f"approval.vpn_application_request.{approval_status}", status="success", resource_type="approval_request", resource_id=str(approval_request.id), detail={ "instance_code": instance_code, "approval_type": approval_request.approval_type, "feishu_status": instance_detail.get("status"), }, ) return if approval_request.approval_type == "announcement_request": from app.domains.announcement_request.service import ( AnnouncementRequestService, ) approval_status = self._map_feishu_status( instance_detail.get("status"), ) approval_request.status = approval_status source_payload = dict(approval_request.source_payload or {}) source_payload["latest_instance_detail"] = instance_detail approval_request.source_payload = source_payload approval_request.updated_at = datetime.now(UTC) ann_svc = AnnouncementRequestService(self.db) await ann_svc.handle_feishu_callback( approval_request=approval_request, instance_detail=instance_detail, ) audit = AuditService(self.db) await audit.log( trace_id=f"feishu_callback:{instance_code}", actor_type="integration", actor_user_id=None, action=f"approval.announcement_request.{approval_status}", status="success", resource_type="approval_request", resource_id=str(approval_request.id), detail={ "instance_code": instance_code, "approval_type": approval_request.approval_type, "feishu_status": instance_detail.get("status"), }, ) return if approval_request.approval_type == "ai_tool_application_request": from app.domains.ai_tool_application.service import ( AiToolApplicationService, ) approval_status = self._map_feishu_status( instance_detail.get("status"), ) approval_request.status = approval_status source_payload = dict(approval_request.source_payload or {}) source_payload["latest_instance_detail"] = instance_detail approval_request.source_payload = source_payload approval_request.updated_at = datetime.now(UTC) ai_svc = AiToolApplicationService(self.db) await ai_svc.handle_feishu_callback( approval_request=approval_request, instance_detail=instance_detail, ) audit = AuditService(self.db) await audit.log( trace_id=f"feishu_callback:{instance_code}", actor_type="integration", actor_user_id=None, action=f"approval.ai_tool_application_request.{approval_status}", status="success", resource_type="approval_request", resource_id=str(approval_request.id), detail={ "instance_code": instance_code, "approval_type": approval_request.approval_type, "feishu_status": instance_detail.get("status"), }, ) return if approval_request.approval_type == "cloud_scope_request": # Cloud sub-account / subscription approvals run through # their own handler, mirroring the dns_record_request # pattern: sync the parent ApprovalRequest BEFORE the # domain handler so a redelivered webhook short-circuits # via the idempotency guard above. from app.domains.cmdb.scope_request_service import ( CloudScopeRequestService, ) approval_status = self._map_feishu_status( instance_detail.get("status"), ) approval_request.status = approval_status source_payload = dict(approval_request.source_payload or {}) source_payload["latest_instance_detail"] = instance_detail approval_request.source_payload = source_payload approval_request.updated_at = datetime.now(UTC) # Pre-launch hardening 2026-05-06 (round 3): write the # dispatcher audit BEFORE calling # ``scope_svc.handle_feishu_callback``, because the scope # service's ``approve()`` path calls # ``await self.db.commit()`` mid-flight (breadcrumb for # the cloud-side ``create_scope`` call). If we audited # AFTER, the audit insert would land in a fresh autobegin # tx that could fail independently — leaving # ``approval_request.status='approved'`` durably committed # but the ``approval.cloud_scope_request.`` audit # row missing. Auditing first puts the audit in the same # tx as the approval_request mutations above, so the # breadcrumb commit either persists both or rolls both # back atomically. audit = AuditService(self.db) await audit.log( trace_id=f"feishu_callback:{instance_code}", actor_type="integration", actor_user_id=None, action=f"approval.cloud_scope_request.{approval_status}", status="success", resource_type="approval_request", resource_id=str(approval_request.id), detail={ "instance_code": instance_code, "approval_type": approval_request.approval_type, "feishu_status": instance_detail.get("status"), }, ) scope_svc = CloudScopeRequestService(self.db) await scope_svc.handle_feishu_callback( approval_request=approval_request, instance_detail=instance_detail, ) return if approval_request.approval_type == "soc2_email_group_grant": # Externally-initiated flow (PR #256). The Feishu approval # template is submitted directly in Feishu (not via Helix's # ``instance/create``). On APPROVED the handler resolves # the applicant via Feishu ``contact/v3/users/{open_id}`` # → email, refuses leavers, and calls Google Workspace # ``members.insert`` against ``soc2@pingcap.com``. # Idempotent on Feishu redeliveries via the parent-row # status guard above AND via the driver's 409 → ok=True # behavior. from app.domains.soc2_email_group.service import ( Soc2EmailGroupService, ) approval_status = self._map_feishu_status( instance_detail.get("status"), ) # Round-22 review (P1): call the service BEFORE mutating # ``approval_request.status`` so a service failure leaves # the row in its pre-callback state. Otherwise rollback # would still revert the in-memory mutation, but ordering # service-first means we never DOUBLE-write the audit row # (success path writes once; failure path persists the # failure-audit through a fresh session — see below). soc_svc = Soc2EmailGroupService(self.db) try: await soc_svc.handle_feishu_callback( approval_request=approval_request, instance_detail=instance_detail, ) except Exception as svc_exc: # Failure-audit MUST survive the rollback that the # outer ``get_db()`` triggers when the AppException # bubbles up. Writing into ``self.db`` would land in # the doomed transaction and disappear. Open a fresh # session, commit the audit row standalone, then # re-raise into the request transaction so the parent # status mutation (if any) is rolled back and Feishu # retries. Mirrors the ``failure_notifier`` pattern # for scheduled jobs (server/app/scheduled_jobs/ # failure_notifier.py:211). try: from app.core.database import ( async_session_factory as _afs, ) async with _afs() as _audit_session: _audit = AuditService(_audit_session) await _audit.log( trace_id=f"feishu_callback:{instance_code}", actor_type="integration", actor_user_id=None, action=( f"approval.soc2_email_group_grant." f"{approval_status}" ), status="failure", resource_type="approval_request", resource_id=str(approval_request.id), detail={ "instance_code": instance_code, "approval_type": ( approval_request.approval_type ), "feishu_status": instance_detail.get( "status", ), "error_code": getattr( svc_exc, "code", None, ), "error_message": str( getattr( svc_exc, "message", svc_exc, ), )[:500], }, ) await _audit_session.commit() except Exception: # noqa: BLE001 # Audit-write failure must never mask the real # service failure. Log + drop. logger.exception( "soc2_email_group_grant: failed to persist " "failure audit for instance %s", instance_code, ) raise approval_request.status = approval_status source_payload = dict(approval_request.source_payload or {}) source_payload["latest_instance_detail"] = instance_detail approval_request.source_payload = source_payload approval_request.updated_at = datetime.now(UTC) audit = AuditService(self.db) await audit.log( trace_id=f"feishu_callback:{instance_code}", actor_type="integration", actor_user_id=None, action=f"approval.soc2_email_group_grant.{approval_status}", status="success", resource_type="approval_request", resource_id=str(approval_request.id), detail={ "instance_code": instance_code, "approval_type": approval_request.approval_type, "feishu_status": instance_detail.get("status"), }, ) return if approval_request.approval_type == "github_org_join_request": # 「申请加入 GitHub 组织」。**两个来源共用这一个分支**: # Helix 自助入口发起的单子(source_payload 里带本地申请单 # id),以及员工直接在飞书上提交的单子(accept_external_origin # 合成父行,handle 与组织从表单里现场解析)。飞书那边的表单 # 入口不会在上线当天消失,而且已经有在途单子 —— 少了外来 # 那一半,过渡期里从飞书提交的人会静默地拿不到自动化。 from app.domains.github_org import callback as github_callback approval_status = self._map_feishu_status( instance_detail.get("status"), ) # 与 SOC2 同一条顺序约定:**先调服务,再改父行状态**。 # 服务失败时父行留在回调前的状态,飞书重投 → 重新分发 → # 执行层幂等,收敛到同一结果。 try: await github_callback.handle_feishu_callback( self.db, approval_request=approval_request, instance_detail=instance_detail, approval_status=approval_status, ) except Exception as svc_exc: # 失败审计必须活过外层 get_db() 的 rollback —— 写进 # self.db 会随那个注定回滚的事务一起消失,运维就看不到 # 这次失败。开一个全新 session 单独 commit。 try: from app.core.database import ( async_session_factory as _afs, ) async with _afs() as _audit_session: await AuditService(_audit_session).log( trace_id=f"feishu_callback:{instance_code}", actor_type="integration", actor_user_id=None, action=( f"approval.github_org_join_request." f"{approval_status}" ), status="failure", resource_type="approval_request", resource_id=str(approval_request.id), detail={ "instance_code": instance_code, "approval_type": ( approval_request.approval_type ), "feishu_status": instance_detail.get("status"), "error_code": getattr(svc_exc, "code", None), "error_message": str( getattr(svc_exc, "message", svc_exc), )[:500], }, ) await _audit_session.commit() except Exception: # noqa: BLE001 logger.exception( "github_org_join_request: failed to persist failure " "audit for instance %s", instance_code, ) raise approval_request.status = approval_status source_payload = dict(approval_request.source_payload or {}) source_payload["latest_instance_detail"] = instance_detail approval_request.source_payload = source_payload approval_request.updated_at = datetime.now(UTC) await AuditService(self.db).log( trace_id=f"feishu_callback:{instance_code}", actor_type="integration", actor_user_id=None, action=f"approval.github_org_join_request.{approval_status}", status="success", resource_type="approval_request", resource_id=str(approval_request.id), detail={ "instance_code": instance_code, "approval_type": approval_request.approval_type, "feishu_status": instance_detail.get("status"), }, ) return if approval_request.approval_type == "dns_record_request": # DNS self-service approvals run through their own handler # which (a) updates the cmdb_dns_record_requests row state # machine, (b) on APPROVED, calls the same Tencent / # Route53 / etc. provider write the admin /admin/cloud/dns # path uses, and (c) commits its own work — caller MUST NOT # commit/flush again. Mirror api_key_request: bail out # cleanly after dispatch. from app.domains.cmdb.dns_request_service import DnsRequestService # Sync the parent ApprovalRequest status BEFORE the DNS # handler runs so a redelivered webhook short-circuits via # the idempotency guard above. approval_status = self._map_feishu_status(instance_detail.get("status")) approval_request.status = approval_status source_payload = dict(approval_request.source_payload or {}) source_payload["latest_instance_detail"] = instance_detail approval_request.source_payload = source_payload approval_request.updated_at = datetime.now(UTC) dns_svc = DnsRequestService(self.db) await dns_svc.handle_feishu_callback( approval_request=approval_request, instance_detail=instance_detail, ) audit = AuditService(self.db) await audit.log( trace_id=f"feishu_callback:{instance_code}", actor_type="integration", actor_user_id=None, action=f"approval.dns_record_request.{approval_status}", status="success", resource_type="approval_request", resource_id=str(approval_request.id), detail={ "instance_code": instance_code, "approval_type": approval_request.approval_type, "feishu_status": instance_detail.get("status"), }, ) return approval_status = self._map_feishu_status(instance_detail.get("status")) approval_request.status = approval_status source_payload = dict(approval_request.source_payload or {}) source_payload["latest_instance_detail"] = instance_detail approval_request.source_payload = source_payload approval_request.updated_at = datetime.now(UTC) employee_request_id = source_payload.get("employee_request_id") employee_request = None if employee_request_id: employee_request = await self._get_employee_request(UUID(employee_request_id)) detail = dict(employee_request.detail or {}) feishu_detail = dict(detail.get("feishu") or {}) feishu_detail["instance_status"] = instance_detail.get("status") feishu_detail["latest_instance_detail"] = instance_detail detail["feishu"] = feishu_detail employee_request.detail = detail employee_request.status = approval_status employee_request.processed_at = datetime.now(UTC) if approval_status != "pending" else None if employee_request and approval_status == "approved": if employee_request.request_type == "software_seat": await self._activate_software_seat( employee_request, source_payload, approval_request_id=approval_request.id, ) elif employee_request.request_type == "software_addon": await self._approve_software_addon_charge(employee_request, source_payload) else: await self._apply_approved_request_to_asset(employee_request, source_payload, instance_detail) elif ( employee_request and approval_status in ("rejected", "cancelled") and employee_request.request_type == "software_addon" ): await self._reject_software_addon_charge( employee_request, source_payload, approval_status, ) elif ( employee_request and approval_status in ("rejected", "cancelled") and employee_request.request_type == "software_seat" ): # Pre-launch hardening 2026-05-05: when Feishu rejects or # cancels the approval, the linked SoftwareSeatAllocation # row was previously left in ``status='pending'`` forever — # blocking re-requests via ``check_no_active_seat_request`` # AND being silently billed by ``_fetch_allocations_overlapping`` # because there was no status filter on the bill query. # We now mark the row revoked with ``revoked_at = allocated_at`` # so its billable window collapses to zero (the seat was # never actually active). Idempotent — re-running on an # already-revoked row is a no-op. await self._revoke_pending_seat_on_rejection( employee_request, source_payload, approval_status, ) await self.db.flush() logger.info( "feishu callback applied: instance=%s approval_type=%s new_status=%s", instance_code, approval_request.approval_type, approval_status, ) # Audit trail for the outcome. Actor is Feishu (external # integration), not a Helix user — hence actor_type="integration" # and actor_user_id=None. The trace_id is derived from the Feishu # instance_code so all events for the same approval instance # cluster together in the audit view. audit = AuditService(self.db) await audit.log( trace_id=f"feishu_callback:{instance_code}", actor_type="integration", actor_user_id=None, action=f"approval.{approval_request.approval_type}.{approval_status}", status="success", resource_type="approval_request", resource_id=str(approval_request.id), detail={ "instance_code": instance_code, "approval_type": approval_request.approval_type, "previous_status": "pending", "new_status": approval_status, "employee_request_id": employee_request_id, "feishu_status": instance_detail.get("status"), }, ) async def _synthesize_external_origin_request( self, *, instance_code: str, payload: dict[str, Any], ) -> tuple[ApprovalRequest, dict[str, Any]] | None: """Create an ``ApprovalRequest`` for a webhook that arrived without a Helix-initiated parent row. Used only for flows where the operator has opted in via ``integration_approval_flows.accept_external_origin = TRUE``. Returns ``None`` (silent drop) when: * the Feishu callback's ``approval_code`` doesn't map to a known flow, or * the flow exists but does NOT opt into external origin, or * the Feishu ``instance_detail`` fetch fails (Feishu retries). On success returns ``(approval_request, instance_detail)`` so the dispatcher can skip a second ``get_instance_detail`` round trip AND operate on the same Feishu snapshot the synthesizer used (avoids a rare APPROVED-then-CANCELLED divergence). The row is flushed but NOT committed inside the caller's transaction with ``status='pending'`` and ``applicant_id=None``. Race-safety: when two webhooks for the same unknown ``instance_code`` arrive concurrently (Feishu rarely deduplicates in-flight redeliveries), the first INSERT wins; the second hits ``UNIQUE(external_id)`` on flush. We catch ``IntegrityError`` and re-select the now-existing row inside a SAVEPOINT so the second webhook continues into the dispatcher exactly as if it had won the row-lookup race at line 636 (idempotency guard upstream then short-circuits if the first one already settled the state). """ try: client = await self._get_client() instance_detail = await client.get_instance_detail(instance_code) except Exception as e: # noqa: BLE001 # Network blips or revoked tokens should not silently # drop an event that we'd otherwise process — surface # via warning + return None so Feishu retries. logger.warning( "feishu: failed to fetch instance_detail for unknown instance_code %s — %s", instance_code, e, ) return None approval_code = instance_detail.get("approval_code") if not approval_code: logger.warning( "feishu: instance_detail for %s has no approval_code; cannot route", instance_code, ) return None flow_row = (await self.db.execute( select(IntegrationApprovalFlow) .where(IntegrationApprovalFlow.external_code == approval_code) .where(IntegrationApprovalFlow.active.is_(True)) .where(IntegrationApprovalFlow.accept_external_origin.is_(True)) .limit(1) )).scalar_one_or_none() if flow_row is None: # Either the template isn't registered in Helix, or it # IS but the operator hasn't opted into external-origin # acceptance — silent drop preserves the existing # contract for the 13 Helix-initiated flow_keys. logger.info( "feishu: unknown instance_code %s for approval_code %s not opted " "into accept_external_origin — ignoring", instance_code, approval_code, ) return None # Opportunistic subscribe refresh — covers the lifespan-sweep # dead-end case where boot-time Feishu was unreachable and the # in-process ``_subscribed`` cache stayed empty. A successful # webhook DELIVERY proves we ARE subscribed (else we wouldn't # have got the event), but feeding the cache here means later # ``submit_approval`` / ``get_approval_definition`` calls on # the same template won't re-POST /subscribe needlessly. Best # effort — failures swallowed by ``ensure_subscribed`` itself. try: await client.ensure_subscribed(approval_code) except Exception: # noqa: BLE001 pass # Wrap the INSERT in a SAVEPOINT so a concurrent-webhook # UNIQUE-collision can be unwound WITHOUT nuking the outer # request transaction (which would also discard prior session # state opened by FastAPI middleware). The codebase-wide # pattern is ``async with self.db.begin_nested()`` — same as # the email_group / asset / leaver_blocked / google_handlers # paths that do per-row INSERTs with possible FK violations. approval_request = ApprovalRequest( external_id=instance_code, approval_type=flow_row.flow_key, applicant_id=None, status="pending", source_payload={ "external_origin": True, "approval_code": approval_code, "flow_key": flow_row.flow_key, "webhook_payload": payload, "latest_instance_detail": instance_detail, "synthesized_at": datetime.now(UTC).isoformat(), }, ) try: async with self.db.begin_nested(): self.db.add(approval_request) await self.db.flush() except IntegrityError: # Concurrent webhook for the same instance_code lost the # race. The SAVEPOINT auto-rolled back when the # IntegrityError propagated out of the ``begin_nested`` # context; the outer request transaction is intact. # Re-select the row the winning webhook inserted and # continue into the dispatcher as if our initial lookup # had hit it. Idempotency guard upstream short-circuits # if the winner already settled the status; otherwise the # driver-level 409 ("already a member") makes the # duplicate side effect a no-op. existing = (await self.db.execute( select(ApprovalRequest) .where(ApprovalRequest.external_id == instance_code) .with_for_update() )).scalar_one_or_none() if existing is None: # Should be impossible — UNIQUE collision implies a # row exists. Defensive: silent drop, Feishu retries. logger.warning( "feishu: synthesize race for %s but row not visible after " "SAVEPOINT rollback — letting Feishu redeliver", instance_code, ) return None logger.info( "feishu: synthesize race for %s — using row %s inserted by " "concurrent webhook", instance_code, existing.id, ) return existing, instance_detail logger.info( "feishu: synthesized ApprovalRequest %s for external-origin " "instance %s (flow_key=%s)", approval_request.id, instance_code, flow_row.flow_key, ) return approval_request, instance_detail async def upload_feishu_file( self, widget_type: str, *, filename: str, content: bytes, content_type: str, ) -> dict[str, Any]: # All approval-form uploads (image / attachmentV2) go through # Feishu's special endpoint ``https://www.feishu.cn/approval/ # openapi/v2/file/upload``, which returns ``{"code": ""}`` # — the canonical file identifier the widget value array must # reference. The ``upload_image`` and ``upload_attachment`` # client helpers both hit this endpoint; only ``type`` differs # (``image`` vs ``attachment``). Videos for the image widget # with the 视频 checkbox enabled use ``upload_video`` which # also hits the same endpoint with ``type=image`` — Feishu's # admin-side widget config (the 视频 checkbox) is what gates # video playback, not upload-time type. client = await self._get_client() if widget_type == "image": if content_type.lower().startswith("video/"): return await client.upload_video_bytes( filename=filename, content=content, content_type=content_type, ) return await client.upload_image_bytes( filename=filename, content=content, content_type=content_type, ) if widget_type == "attachmentV2": return await client.upload_attachment_bytes( filename=filename, content=content, content_type=content_type, ) raise AppException( code="UNSUPPORTED_UPLOAD_WIDGET", message=f"Unsupported upload widget type: {widget_type}", status_code=400, ) async def verify_callback_token(self, payload: dict[str, Any]) -> None: """Authenticate a Feishu callback without X-Lark-Signature. Pre-launch hardening 2026-05-05 (P1 from SS3): the docstring previously described a fallback path that "trusted" a payload containing only ``app_id`` (no token anywhere) — a public identifier, not a secret. That paragraph was reading-trap bait for a future maintainer who might re-introduce the unsafe fallback. Behavior here is now strict: token must be present AND match. No app_id fallback; missing token = 401. Token may live at any of: ``payload.token``, ``payload.header.token``, ``payload.event.token`` (Feishu webhook format has shifted across versions). The endpoint rejects the call when any of these holds a value that doesn't match the configured ``event_verification_token`` AND when no token field is present at all. URL-verification proves we own the endpoint at registration time; the per-event token is the ongoing authenticity signal. """ cfg = await get_feishu_config(session=self.db) expected = str(cfg.get("event_verification_token") or "") if not expected: logger.error( "feishu.event_verification_token is not set in 配置中心 — " "refusing all callbacks until configured." ) raise AppException( code="FEISHU_WEBHOOK_NOT_CONFIGURED", message="Feishu webhook verification token is not configured. Contact IT Admin.", status_code=503, ) header = payload.get("header") if isinstance(payload.get("header"), dict) else {} event = payload.get("event") if isinstance(payload.get("event"), dict) else {} token_candidates = [ payload.get("token"), header.get("token") if header else None, event.get("token") if event else None, ] token = next((t for t in token_candidates if t), None) if token: if hmac.compare_digest(expected.encode(), str(token).encode()): # Pre-launch P1 fix (Round 45, CR-103): verify the # inbound payload's app_id matches the configured # Feishu app. If two apps ever share / leak the same # event_verification_token, the wrong app's callbacks # could mutate our data. Compounds with the static # token model — token compromise from ANY app that # ever had it set would otherwise allow forged # approved/rejected callbacks for any matching code. configured_app_id = str(cfg.get("app_id") or "") inbound_app_id = ( payload.get("app_id") or (header.get("app_id") if header else None) or (event.get("app_id") if event else None) ) if ( configured_app_id and inbound_app_id and str(inbound_app_id) != configured_app_id ): logger.warning( "feishu callback app_id mismatch: inbound=%r " "configured=%r — token matched but app diverged", inbound_app_id, configured_app_id, ) raise AppException( code="FEISHU_APP_MISMATCH", message="Feishu app_id does not match configured app.", status_code=401, ) return logger.warning( "feishu callback token mismatch: found token at some location " "but value did not match configured event_verification_token" ) raise AppException( code="FEISHU_TOKEN_INVALID", message="Invalid Feishu webhook token", status_code=401, ) raise AppException( code="FEISHU_TOKEN_INVALID", message="Invalid Feishu webhook token", status_code=401, ) @staticmethod def verify_lark_signature( encrypt_key: str, timestamp: str, nonce: str, body: bytes, signature: str, ) -> bool: """Verify Feishu X-Lark-Signature header (v2 webhook security). The signature is SHA-256 of (timestamp + nonce + encrypt_key) concatenated with the raw request body bytes. """ content = (timestamp + nonce + encrypt_key).encode() + body expected = hashlib.sha256(content).hexdigest() return hmac.compare_digest(expected, signature) async def _apply_approved_request_to_asset( self, employee_request: AssetEmployeeRequest, source_payload: dict[str, Any], instance_detail: dict[str, Any], ) -> None: asset = None asset_id = source_payload.get("asset_id") if asset_id: asset = await self._get_asset(UUID(asset_id)) elif employee_request.asset_id: asset = await self._get_asset(employee_request.asset_id) if not asset: return extracted = self._extract_asset_relevant_fields( employee_request.request_type, instance_detail.get("form", []), ) custom_fields = dict(asset.custom_fields or {}) feishu_approvals = dict(custom_fields.get("feishu_approvals") or {}) request_detail = dict(employee_request.detail or {}) feishu_detail = dict(request_detail.get("feishu") or {}) instance_code = ( source_payload.get("latest_instance_detail", {}).get("instance_code") or feishu_detail.get("instance_code") ) approval_record = { "request_number": employee_request.request_number, "instance_code": instance_code, "status": "approved", "synced_at": datetime.now(UTC).isoformat(), } approval_record.update(extracted) feishu_approvals[employee_request.request_type] = approval_record custom_fields["feishu_approvals"] = feishu_approvals asset.custom_fields = custom_fields asset.updated_at = datetime.now(UTC) # Close the loop: flip the asset state to match the approval. # Before this, admins had to manually click 退库/借出/处置 on the UI # after every Feishu approval — easy to forget, hard to audit. # Each branch logs and swallows errors so a bad auto-execute # doesn't roll back the metadata write or the Feishu callback. # SAVEPOINT-isolate (2026-08-06): "swallows errors so a bad # auto-execute doesn't roll back the metadata write or the Feishu # callback" only holds if the session survives the failure. The # dispose/borrow/return calls below write rows; on a DB-level error # the transaction deactivates and the caller's own # ``await self.db.flush()`` + audit write raise instead — taking the # whole callback down and leaving the approval stuck ``pending``. # This is the exact twin of the software_seat break fixed alongside it. async with best_effort( self.db, label=( f"auto-execute {employee_request.request_type} " f"{employee_request.request_number}" ), log=logger, ) as exec_step: await self._auto_execute_approved_request( employee_request, asset, extracted, instance_code, ) if not exec_step.ok: exc = exec_step.error # A swallowed failure here leaves request=approved + asset # metadata stamped but the state-flip (dispose/borrow/return) # NOT done — e.g. buyback then believes the disposal happened # while the asset is still assigned. Log-only evidence is not # enough for that divergence: write an audit row and page the # global admins (dedup-aware) so someone reconciles the asset # by hand. Deliberately still no rollback — the metadata write # and the Feishu callback must survive. async with best_effort( self.db, label=( "record auto-execute failure evidence for " f"{employee_request.request_number}" ), log=logger, ): await self._record_auto_execute_failure( employee_request, asset, instance_code, exc, ) async def _record_auto_execute_failure( self, employee_request: AssetEmployeeRequest, asset: AssetItem, instance_code: str | None, exc: BaseException | None, ) -> None: """Audit + dedup-aware admin page for a swallowed auto-execute failure — the request shows approved while the asset state-flip (dispose/borrow/return) did not happen.""" from app.core.exceptions import ( # noqa: PLC0415 format_exception_for_log, ) await self.audit.log( trace_id=f"feishu:{instance_code}" if instance_code else "auto", actor_type="integration", actor_user_id=None, action="approval.auto_execute_failed", status="failure", resource_type="asset_item", resource_id=str(asset.id), detail={ "request_number": employee_request.request_number, "request_type": employee_request.request_type, "error": ( format_exception_for_log(exc, max_len=500) if exc is not None else "unknown" ), }, ) from app.approval.cancel_desync import ( # noqa: PLC0415 notify_feishu_cancel_desync, ) await notify_feishu_cancel_desync( self.db, flow_label="资产审批自动执行", resource_type="asset_auto_execute", resource_id=str(asset.id), instance_code=instance_code or "", error=exc if isinstance(exc, Exception) else RuntimeError(str(exc)), trace_id=f"feishu:{instance_code}" if instance_code else "auto", ) async def _auto_execute_approved_request( self, employee_request: AssetEmployeeRequest, asset: AssetItem, extracted: dict[str, Any], instance_code: str | None, ) -> None: """Flip asset state to match the just-approved Feishu request. Each handler is idempotent: if the asset is already in the target terminal state (disposed / borrowed by requester / idle) we skip without raising. If the asset is in a state that prevents the action (e.g. return_asset on an already-idle asset), we log and skip — the approval metadata is still recorded either way. """ from app.domains.asset.schemas import ( AssetBorrowRequest, AssetDisposeRequest, AssetReturnRequest, ) from app.domains.asset.service import AssetService request_type = employee_request.request_type if request_type not in ( "offboarding_buyback", "computer_borrow", "asset_return", ): return # laptop_replacement + anything new: metadata only # TOCTOU revalidation (cycle 3 audit fix). Approval cards can sit # in Feishu for days; if the asset has been reassigned / returned # / disposed between submit and approval, executing the # auto-action against the new state could mutate hardware that # belongs to a successor. ``offboarding_buyback`` calls # ``dispose_asset(force=True)`` which would otherwise bypass the # in-use guard and dispose a successor's hardware. Re-load the # asset (the caller already did, but be defensive) and re-check # ownership. On mismatch we audit + skip the mutation rather # than silently doing the wrong thing. from sqlalchemy import select as _select # noqa: PLC0415 fresh = await self.db.scalar( _select(AssetItem).where(AssetItem.id == asset.id), ) if fresh is None: logger.warning( "auto-execute %s: asset %s vanished between submit and " "approval — skipping", employee_request.request_number, asset.id, ) return # Allowed: requester still owns OR currently borrows. Or, for # ``computer_borrow``, asset is now in stock (no holder). requester_id = employee_request.requester_id ownership_ok = ( fresh.assigned_to == requester_id or fresh.borrowed_by == requester_id ) in_stock_borrow_ok = ( request_type == "computer_borrow" and fresh.assigned_to is None and fresh.borrowed_by is None ) if not ownership_ok and not in_stock_borrow_ok: logger.warning( "auto-execute %s (%s) refused: asset %s ownership drifted " "between submit and approval (assigned_to=%s borrowed_by=%s " "requester=%s) — not mutating", employee_request.request_number, request_type, fresh.asset_tag, fresh.assigned_to, fresh.borrowed_by, requester_id, ) await self.audit.log( trace_id=f"feishu:{instance_code}" if instance_code else "auto", actor_type="integration", actor_user_id=requester_id, action="approval.auto_execute_refused_ownership_drift", status="failure", resource_type="asset_item", resource_id=str(fresh.id), detail={ "request_number": employee_request.request_number, "request_type": request_type, "asset_assigned_to": ( str(fresh.assigned_to) if fresh.assigned_to else None ), "asset_borrowed_by": ( str(fresh.borrowed_by) if fresh.borrowed_by else None ), "requester": str(requester_id), }, ) return # Use the fresh row going forward so any state-flip happens # against the latest persisted snapshot. asset = fresh # Pre-launch P1 fix (Round 45, CR-308): re-check the requester's # employment status at auto-execute time for the ONE flow where # requester-must-be-active is a real precondition: # ``computer_borrow``. The other two paths must NOT apply this # gate: # * ``offboarding_buyback`` — by definition submitted by a # leaver; status will be ``leaver_pending`` at every # auto-execute. Refusing here breaks the entire buyback # flow. # * ``asset_return`` — a return-in-flight on a leaver is the # legitimate "hand back hardware on the way out" path. # The software_seat activation path elsewhere already gates # its own way; this hardware-path gate is scoped accordingly. if request_type == "computer_borrow": from app.identity.person_models import ( IdentityPerson as _IdentityPerson, # noqa: PLC0415 ) requester = await self.db.get(_IdentityPerson, requester_id) requester_status = ( (requester.employment_status or "").lower() if requester else "" ) if requester is None or requester_status in {"left", "leaver_pending"}: logger.warning( "auto-execute %s (%s) refused: requester %s status=%s — " "not mutating", employee_request.request_number, request_type, requester_id, requester_status or "missing", ) await self.audit.log( trace_id=f"feishu:{instance_code}" if instance_code else "auto", actor_type="system", actor_user_id=None, action="approval.auto_execute_refused_requester_inactive", status="failure", resource_type="employee_request", resource_id=str(employee_request.id), detail={ "request_number": employee_request.request_number, "request_type": request_type, "requester_status": requester_status or "missing", "requester_id": str(requester_id), "asset_id": str(asset.id), }, ) return asset_service = AssetService(self.db) # The approval system is the actor of record for auto-executes. # Operator IDs that reference a human require an identity_persons # row; we use the requester as a pragmatic stand-in so the # AssetOperation row has a traceable origin. operator_id = employee_request.requester_id audit_note_tail = ( f" [auto via approval {employee_request.request_number}" + (f" / feishu {instance_code}" if instance_code else "") + "]" ) if request_type == "offboarding_buyback": # Leaver is buying their own device. At approval time the # asset is typically status=assigned holder=leaver, so we # force through the "asset in use" guard and dispose it # as sold. Idempotent: skip if already disposed. if asset.status in ("disposed", "disposal_pending"): logger.info( "offboarding_buyback auto-dispose skipped: asset %s already %s", asset.asset_tag, asset.status, ) return buyback_amount = extracted.get("buyback_amount") # Round 2 audit F12 (defence-in-depth, 2026-05-20): # ``buyback_amount`` is whatever Feishu echoed back from # the form. On the Helix-originated submit path it WAS # server-enforced via ``_enforce_server_buyback_amount`` # at submit time, so any client tamper was stripped before # the request ever reached Feishu. Recompute the canonical # quote here from the asset's purchase columns + policy # and stamp it alongside the claimed value so any tamper # is forensically visible — and so downstream cost # reconciliation can rely on the canonical figure even if # the submit-time enforcer is ever bypassed (e.g. a future # ``accept_external_origin`` opt-in that synthesises an # ``AssetEmployeeRequest`` without re-running the # enforcer). canonical_amount: dict[str, Any] | None = None try: # Resolve target currency the same way the public quote # path does (see ``quote_offboarding_buyback``): # entity-based default → asset.currency → CNY. # # Round 4 audit #2 (2026-05-20): pin to the # entity-derived default; DO NOT honour # ``buyback_amount.currency`` here. If a future bypass # of the submit-time enforcer landed a tampered # currency, accepting it would defeat the forensic # comparison this audit field exists for. The canonical # figure must be reproducible from asset state alone. entity = await self._resolve_payment_entity(asset) # ``AssetLegalEntity.currency`` (defined at # ``server/app/domains/asset/models.py:694``) holds the # entity's banking currency — there is no # ``default_payment_currency`` column. Same attribute # the public ``quote_offboarding_buyback`` path reads # at L3192. entity_currency = entity.currency if entity else None default_currency = resolve_default_currency( entity_currency=entity_currency, asset_currency=asset.currency, ) quote = calculate_buyback( purchase_cost=asset.purchase_cost, purchase_currency=asset.currency, purchase_date=( asset.purchase_date.date() if isinstance(asset.purchase_date, datetime) else asset.purchase_date ), today=datetime.now(UTC).date(), target_currency=default_currency, ) canonical_amount = { "amount": str(quote.amount), "currency": quote.currency, } except ValueError as exc: # Asset missing purchase data (calculator raises # ``MISSING_PURCHASE_COST`` / ``MISSING_PURCHASE_DATE`` # / ``UNSUPPORTED_PURCHASE_CURRENCY`` etc.) — log + # continue with whatever the form claimed. The # disposal itself doesn't depend on the figure; only # the audit record does. logger.warning( "buyback canonical re-quote failed for asset %s: %s", asset.asset_tag, exc, ) # Round 4 audit #3 (2026-05-20): json.dumps the # user-supplied dict before interpolating into the # AssetOperation.notes column. f"...amount={dict}..." would # land newlines / control chars / ANSI escapes verbatim, # corrupting grep / CSV exports of the audit trail. The # canonical figure is stamped separately in audit_logs. try: amount_str = json.dumps( buyback_amount, ensure_ascii=True, default=str, )[:200] except (TypeError, ValueError): amount_str = "" notes = f"Offboarding buyback approved; amount={amount_str}{audit_note_tail}" await asset_service.dispose_asset( asset.id, AssetDisposeRequest(disposal_type="sold", notes=notes), operator_id, force=True, # Buyback is the only legitimate dispose-of-held_by_leaver # path — opt in explicitly so the service-layer guard # against silent leaver-attribution loss stays strict. # Kwarg lives on the service method, NOT on the Pydantic # schema, so an external API caller can't sneak it in. allow_held_by_leaver=True, # Names the reason for downstream copy — the MDM cleanup # card says "离职回购(员工买断)" rather than a bare # "sold", which is the difference between an # administrator understanding the request and guessing. disposal_context="buyback", ) # Audit row (cycle 3 fix): the underlying ``dispose_asset`` # writes ``AssetOperation`` + ``AssetHistory`` but NOT to # ``audit_logs`` (the canonical compliance store; service- # layer doesn't write it because the HTTP router normally # does). On the auto-execute path there's no router, so # without this the auditor cannot filter # ``action LIKE 'asset.dispose%'`` and see Feishu-driven # disposals. await self.audit.log( trace_id=f"feishu:{instance_code}" if instance_code else "auto", actor_type="integration", actor_user_id=operator_id, action="asset.dispose.auto_via_approval", status="success", resource_type="asset_item", resource_id=str(asset.id), detail={ "asset_tag": asset.asset_tag, "disposal_type": "sold", "buyback_amount": buyback_amount, # 2026-05-20 (F12): include the server-recomputed # canonical figure so a forensic comparison against # the claimed figure is one ``audit_logs`` query # away. ``None`` means the asset is missing # purchase-cost data and the canonical value cannot # be computed — in which case the claimed figure is # all we have. "buyback_amount_canonical": canonical_amount, "request_number": employee_request.request_number, "instance_code": instance_code, }, ) return if request_type == "computer_borrow": borrower_id = employee_request.requester_id # Round 2b audit fix-up 2026-05-04: idempotency guard now # uses ``borrowed_by`` (the canonical "current borrower" # column post-115). The previous ``assigned_to == borrower`` # check matched the legacy bug-shape of borrow rows; under # the new model ``assigned_to`` is the preserved original # owner (or NULL) so the guard always missed and duplicate # Feishu callbacks fell through to "not available" warning. if asset.status == "borrowed" and asset.borrowed_by == borrower_id: logger.info( "computer_borrow auto-execute skipped: asset %s already borrowed by %s", asset.asset_tag, borrower_id, ) return # ``borrow_asset`` now also accepts ``claimed`` / ``assigned`` # (lend-out-while-owned), so the dispatch is broader than # idle stock. if asset.status not in ("idle", "available", "claimed", "assigned"): logger.warning( "computer_borrow auto-execute skipped: asset %s is %s (not available)", asset.asset_tag, asset.status, ) return # Date source preference (2026-05-19): # 1. Helix's EmployeeRequest.detail.borrow_dates — the # canonical record of what the applicant submitted, # stashed at submit time in ``create_asset_approval_request``. # Stays the primary source even now that the window # ships to Feishu (2026-08-14) — Feishu's callback # form_items arrives partial under load, and the # submit path may still degrade to omitting the # widget (see ``_create_instance_with_form_fallback``). # 2. ``extracted`` from Feishu form_items — pre-2026-05-19 # path; still tried as fallback in case an admin # edited the dateInterval widget on Feishu's UI before # approving (which DOES round-trip through # ``get_instance`` → form_items). # 3. None → the T+30-days fallback below. stored_dates = (employee_request.detail or {}).get("borrow_dates") or {} start = ( _parse_iso_date(stored_dates.get("start")) or _parse_iso_date(extracted.get("borrow_start")) ) end = ( _parse_iso_date(stored_dates.get("end")) or _parse_iso_date(extracted.get("borrow_end")) ) # ``AssetBorrowRequest.expected_return_date`` is REQUIRED # post-115 + must be strictly in the future. Feishu approvals # can arrive after the originally-requested end date (slow # approvers), so substitute a sensible fallback (T+30 days # so the new ``asset.borrow_overdue`` cron has a signal to # escalate on) instead of passing None — which would 422 at # Pydantic now that the field is required. from datetime import timedelta as _td today = date.today() if end is None or end <= today: fallback = today + _td(days=30) logger.info( "computer_borrow auto-execute: requested end %s is " "missing/stale, using fallback %s", end, fallback, ) end = fallback await asset_service.borrow_asset( asset.id, AssetBorrowRequest( to_user_id=borrower_id, borrow_date=start, expected_return_date=end, notes=f"Auto-borrow from Feishu approval{audit_note_tail}", ), operator_id, ) await self.audit.log( trace_id=f"feishu:{instance_code}" if instance_code else "auto", actor_type="integration", actor_user_id=operator_id, action="asset.borrow.auto_via_approval", status="success", resource_type="asset_item", resource_id=str(asset.id), detail={ "asset_tag": asset.asset_tag, "borrower": str(borrower_id), "borrow_date": start.isoformat() if start else None, "expected_return_date": end.isoformat() if end else None, "request_number": employee_request.request_number, "instance_code": instance_code, }, ) return if request_type == "asset_return": if asset.status == "idle": logger.info( "asset_return auto-execute skipped: asset %s already idle", asset.asset_tag, ) return if asset.status not in ("claimed", "assigned"): logger.warning( "asset_return auto-execute skipped: asset %s is %s (not claimed/assigned)", asset.asset_tag, asset.status, ) return # Ownership drift guard (2026-05-28): on the reaper / refresh # replay path the original Feishu submit may be 3–14 days old. # If the original requester returned the asset manually # (asset → idle) and a NEW user has since claimed it (asset # → claimed, asset.assigned_to = new_user), the # ``asset.status in ('claimed','assigned')`` check above # would PASS — and we would then return the asset out from # under the new owner. Refuse + audit when the current # holder is not the original requester. if ( asset.assigned_to is not None and asset.assigned_to != employee_request.requester_id ): logger.warning( "asset_return auto-execute skipped: asset %s was " "re-assigned (current holder=%s, original requester=%s); " "admin must reconcile manually", asset.asset_tag, asset.assigned_to, employee_request.requester_id, ) await self.audit.log( trace_id=f"feishu:{instance_code}" if instance_code else "auto", actor_type="integration", actor_user_id=operator_id, action="asset.return.auto_via_approval.skipped_ownership_drift", status="success", resource_type="asset_item", resource_id=str(asset.id), detail={ "asset_tag": asset.asset_tag, "current_holder": str(asset.assigned_to), "original_requester": str(employee_request.requester_id), "request_number": employee_request.request_number, "instance_code": instance_code, }, ) return # Prefer Helix's stashed return_date over the Feishu # callback's form_items echo (2026-05-19): the applicant's # original submission is the source of truth — Feishu's # callback form_items can arrive partial under load and # falling back to today() then disposes the wrong-day # return on the asset history. # # 2026-05-20: source key changed from ``resignation_date`` # to ``return_date`` — the asset_return Feishu form ships # ``归还日期`` / ``Return Date``, not a resignation date. # Legacy ``resignation_date`` retained as a fallback for # rows captured before the rename so historical re-execute # paths still resolve. stash = employee_request.detail or {} return_date = ( _parse_iso_date(stash.get("return_date")) or _parse_iso_date(stash.get("resignation_date")) or _parse_iso_date(extracted.get("return_date")) or _parse_iso_date(extracted.get("resignation_date")) ) await asset_service.return_asset( asset.id, AssetReturnRequest( return_date=return_date, notes=f"Auto-return from Feishu approval{audit_note_tail}", ), operator_id, ) await self.audit.log( trace_id=f"feishu:{instance_code}" if instance_code else "auto", actor_type="integration", actor_user_id=operator_id, action="asset.return.auto_via_approval", status="success", resource_type="asset_item", resource_id=str(asset.id), detail={ "asset_tag": asset.asset_tag, "return_date": return_date.isoformat() if return_date else None, "request_number": employee_request.request_number, "instance_code": instance_code, }, ) return async def _revoke_pending_seat_on_rejection( self, employee_request: AssetEmployeeRequest, source_payload: dict[str, Any], approval_status: str, ) -> None: """Flip a pending SoftwareSeatAllocation to revoked when Feishu rejects/cancels the approval. Idempotent. Safe to call from both the live callback and the ``approval_pending_reaper`` cron — both feed through the same ``_apply_callback_state`` path. ``revoked_at`` is set equal to ``allocated_at`` so the row's billable window collapses to zero (the seat was never active). The companion ``_fetch_allocations_overlapping`` filter excludes any row whose ``revoked_at == allocated_at``. """ from app.audit.service import AuditService from app.domains.asset.models import SoftwareSeatAllocation seat_allocation_id = source_payload.get("seat_allocation_id") if not seat_allocation_id: logger.warning( "software_seat rejection callback missing seat_allocation_id " "for request %s — pending row will be cleaned up by " "migration 119 backfill on next deploy", employee_request.request_number, ) return allocation = await self.db.scalar( select(SoftwareSeatAllocation) .where(SoftwareSeatAllocation.id == UUID(seat_allocation_id)), ) if allocation is None: return if allocation.status != "pending": # Already activated and possibly already revoked elsewhere; # don't touch a row in a terminal state. return # Defence-in-depth ownership check (mirror of _activate_software_seat). if allocation.user_id != employee_request.requester_id: logger.error( "software_seat rejection refused for %s: allocation user_id " "(%s) does not match approval applicant (%s)", seat_allocation_id, allocation.user_id, employee_request.requester_id, ) return allocation.status = "revoked" allocation.revoked_at = allocation.allocated_at allocation.reason = ( (allocation.reason or "") + f"\n[auto-revoked by approval callback: Feishu approval {approval_status}; " "revoked_at aligned to allocated_at so the row never enters a bill]" ).strip() await self.db.flush() await AuditService(self.db).log( trace_id=f"approval:{employee_request.request_number}", actor_type="system", actor_user_id=None, action=f"software_seat.rejected_{approval_status}", resource_type="software_seat_allocation", resource_id=str(allocation.id), status="success", detail={ "plan_id": str(allocation.plan_id), "target_user_id": str(allocation.user_id), "approval_status": approval_status, "request_number": employee_request.request_number, "trigger": "feishu_approval_callback", }, ) logger.info( "software_seat allocation %s revoked on approval %s for request %s", allocation.id, approval_status, employee_request.request_number, ) async def _activate_software_seat( self, employee_request: AssetEmployeeRequest, source_payload: dict[str, Any], approval_request_id: UUID | None = None, ) -> None: """Activate a SoftwareSeatAllocation after Feishu approval. Pre-launch hardening: the original implementation only flipped ``status = active``. That worked for the happy path but failed compliance review on three counts — 1. No leaver re-check. Feishu approvals can sit for days; the requester may have left the company in the interim and granting a paid seat to a leaver is an audit incident. 2. No quota re-check. The plan cap may have been tightened or other concurrent activations may have consumed the remaining seats since the request was submitted. 3. No audit log. The request had one, the cron-driven revoke has one, but the activation itself was invisible to the audit trail — so "when did this seat go live, and why?" could only be answered from the Feishu side. All three are now closed. Each failure is non-fatal (the Feishu instance is already approved on their side, so we still commit the existing audit row + transition error notification and leave the allocation in 'pending' for an admin to clean up from /admin/seat-allocations). """ from app.audit.service import AuditService from app.core.coordinators.provisioning_coordinator import ( check_plan_quota, enqueue_grant_for_seat, ) from app.core.exceptions import AppException from app.domains.asset.models import ( SoftwareLicensePlan, SoftwareSeatAllocation, ) from app.identity.person_models import IdentityPerson seat_allocation_id = source_payload.get("seat_allocation_id") if not seat_allocation_id: logger.warning( "software_seat approval callback missing seat_allocation_id for request %s", employee_request.request_number, ) return row = (await self.db.execute( select(SoftwareSeatAllocation, SoftwareLicensePlan) .join( SoftwareLicensePlan, SoftwareSeatAllocation.plan_id == SoftwareLicensePlan.id, ) .where(SoftwareSeatAllocation.id == UUID(seat_allocation_id)) )).first() if row is None: logger.warning( "SoftwareSeatAllocation %s not found for approval callback", seat_allocation_id, ) return allocation, plan = row if allocation.status != "pending": logger.info( "SoftwareSeatAllocation %s already in status %s — skipping activation", seat_allocation_id, allocation.status, ) return # Pre-launch hardening 2026-05-04 (P0 AP2): defence-in-depth # ownership cross-check. The Feishu callback dispatches by # ``seat_allocation_id`` from the source_payload — even after # the public-schema ``extra_source_data`` was locked down, # any future regression that lets a caller seed # ``seat_allocation_id`` for a victim's pending allocation # would activate that victim's seat. Reject any mismatch # between the allocation's user_id and the approval applicant. if allocation.user_id != employee_request.requester_id: logger.error( "software_seat activation refused for %s: allocation " "user_id (%s) does not match approval applicant (%s) — " "possible IDOR attempt (request %s)", seat_allocation_id, allocation.user_id, employee_request.requester_id, employee_request.request_number, ) await AuditService(self.db).log( trace_id=f"approval:{employee_request.request_number}", actor_type="system", actor_user_id=None, action="software_seat.activation_refused_ownership_mismatch", resource_type="software_seat_allocation", resource_id=str(allocation.id), status="failure", detail={ "plan_id": str(allocation.plan_id), "target_user_id": str(allocation.user_id), "applicant_id": str(employee_request.requester_id), "reason": "ownership_mismatch", }, ) return # Leaver re-check — must NOT grant a paid seat to someone who # has left OR whose offboarding has already started in the # interim between request and approval. target = await self.db.get(IdentityPerson, allocation.user_id) if is_blocked_from_new_grant(target): # Pre-launch hardening — log the target's status when known # (target may be None here, hence the explicit conditional; # NOT a getattr-default antipattern). target_status = target.employment_status if target else None logger.warning( "software_seat activation refused for %s: target user " "is deleted/leaver (request %s, status=%s)", seat_allocation_id, employee_request.request_number, target_status, ) await AuditService(self.db).log( trace_id=f"approval:{employee_request.request_number}", actor_type="system", actor_user_id=None, action="software_seat.activation_refused_leaver", resource_type="software_seat_allocation", resource_id=str(allocation.id), status="failure", detail={ "plan_id": str(allocation.plan_id), "target_user_id": str(allocation.user_id), "target_status": target_status, "reason": "target_user_left_or_deleted_or_leaver_pending", }, ) return # Quota re-check at activation time. Mirrors what the no-Feishu # admin_approve and admin_grant paths do. try: await check_plan_quota(self.db, plan.id) except AppException as exc: logger.warning( "software_seat activation refused for %s: plan %s over " "quota (%s)", seat_allocation_id, plan.id, exc.code, ) await AuditService(self.db).log( trace_id=f"approval:{employee_request.request_number}", actor_type="system", actor_user_id=None, action="software_seat.activation_refused_quota", resource_type="software_seat_allocation", resource_id=str(allocation.id), status="failure", detail={ "plan_id": str(allocation.plan_id), "target_user_id": str(allocation.user_id), "reason": exc.code, }, ) return allocation.status = "active" allocation.allocated_at = datetime.now(UTC) await self.db.flush() task_id: UUID | None = None # SAVEPOINT-isolate the enqueue so the "must not block the # activation" promise below actually holds. Without it a failed # INSERT (e.g. the 2026-08-06 FK violation on the task's # approval_request_id) deactivates the surrounding transaction, # and the *next* statement — the audit write, or even the lazy # load of ``employee_request.request_number`` feeding it — # raises instead. That turned a swallowed warning into a hard # failure of the whole callback, leaving the approval row stuck # pending and the seat never provisioned. trace = f"approval:{employee_request.request_number}" # Enqueue failure must not block the activation — the seat is now # active for access control, and an admin can still create the task # from /admin/provisioning. ``best_effort`` also re-raises the # transaction-aborting errnos a bare ``begin_nested`` would have # swallowed into a savepoint cascade. async with best_effort( self.db, label=f"enqueue grant for seat {seat_allocation_id}", log=logger, ) as enqueue_step: task_id = await enqueue_grant_for_seat( self.db, seat=allocation, trace_id=trace, actor_user_id=None, approval_request_id=approval_request_id, ) if not enqueue_step.ok: task_id = None await AuditService(self.db).log( trace_id=trace, actor_type="system", actor_user_id=None, action="software_seat.activated", resource_type="software_seat_allocation", resource_id=str(allocation.id), status="success", detail={ "plan_id": str(allocation.plan_id), "plan_name": plan.name, "target_user_id": str(allocation.user_id), "target_email": target.work_email, "software_name": plan.software.name if plan.software else None, "approval_request_number": employee_request.request_number, "provisioning_task_id": str(task_id) if task_id else None, "trigger": "feishu_approval_callback", }, ) async def _approve_software_addon_charge( self, employee_request: Any, source_payload: dict[str, Any], ) -> None: """Approve a one-time SoftwareAddonCharge after Feishu approval. Parallels ``_activate_software_seat`` but there is no seat to grant and no quota to check — the side-effect is purely "pin the charge to a billing period so close_month bills it". Keeps the same ownership + leaver re-checks: never bill a leaver's (former) department for a purchase that was never delivered. Idempotent: a redelivered webhook on an already-approved charge no-ops. """ from app.audit.service import AuditService from app.domains.asset import addon_service from app.domains.asset.models import SoftwareAddonCharge from app.identity.person_models import IdentityPerson charge_id = source_payload.get("addon_charge_id") if not charge_id: logger.warning( "software_addon approval callback missing addon_charge_id for request %s", employee_request.request_number, ) return charge = (await self.db.execute( select(SoftwareAddonCharge).where(SoftwareAddonCharge.id == UUID(charge_id)) )).scalar_one_or_none() if charge is None: logger.warning( "SoftwareAddonCharge %s not found for approval callback", charge_id, ) return if charge.status != "pending": logger.info( "SoftwareAddonCharge %s already in status %s — skipping approval", charge_id, charge.status, ) return # Ownership cross-check — defence-in-depth against a source_payload # carrying a victim's charge id (mirrors the seat path). if charge.user_id != employee_request.requester_id: logger.error( "software_addon approval refused for %s: charge user_id (%s) " "!= applicant (%s) — possible IDOR (request %s)", charge_id, charge.user_id, employee_request.requester_id, employee_request.request_number, ) await AuditService(self.db).log( trace_id=f"approval:{employee_request.request_number}", actor_type="system", actor_user_id=None, action="software_addon.approval_refused_ownership_mismatch", resource_type="software_addon_charge", resource_id=str(charge.id), status="failure", detail={ "applicant_id": str(employee_request.requester_id), "charge_user_id": str(charge.user_id), "reason": "ownership_mismatch", }, ) return # Leaver re-check — don't bill a former employee's department for a # purchase that won't be delivered. Leave the charge pending for an # admin to resolve (the Feishu side is already approved). target = await self.db.get(IdentityPerson, charge.user_id) if is_blocked_from_new_grant(target): target_status = target.employment_status if target else None logger.warning( "software_addon approval refused for %s: target leaver/deleted " "(request %s, status=%s)", charge_id, employee_request.request_number, target_status, ) await AuditService(self.db).log( trace_id=f"approval:{employee_request.request_number}", actor_type="system", actor_user_id=None, action="software_addon.approval_refused_leaver", resource_type="software_addon_charge", resource_id=str(charge.id), status="failure", detail={ "target_user_id": str(charge.user_id), "target_status": target_status, "reason": "target_user_left_or_deleted_or_leaver_pending", }, ) return await addon_service.mark_charge_approved( self.db, charge, department=target.department, ) await AuditService(self.db).log( trace_id=f"approval:{employee_request.request_number}", actor_type="system", actor_user_id=None, action="software_addon.approved", resource_type="software_addon_charge", resource_id=str(charge.id), status="success", detail={ "item_id": str(charge.item_id), "software_id": str(charge.software_id), "target_user_id": str(charge.user_id), "quantity": charge.quantity, "amount": str(charge.amount), "currency": charge.currency, "period": charge.period, "department": charge.department, "approval_request_number": employee_request.request_number, "trigger": "feishu_approval_callback", }, ) async def _reject_software_addon_charge( self, employee_request: Any, source_payload: dict[str, Any], approval_status: str, ) -> None: """Collapse a SoftwareAddonCharge when Feishu rejects/cancels it, so it is never materialised into a bill. Idempotent.""" from app.audit.service import AuditService from app.domains.asset import addon_service from app.domains.asset.models import SoftwareAddonCharge charge_id = source_payload.get("addon_charge_id") if not charge_id: logger.warning( "software_addon %s callback missing addon_charge_id for request %s", approval_status, employee_request.request_number, ) return charge = (await self.db.execute( select(SoftwareAddonCharge).where(SoftwareAddonCharge.id == UUID(charge_id)) )).scalar_one_or_none() if charge is None or charge.status != "pending": return new_status = "cancelled" if approval_status == "cancelled" else "rejected" await addon_service.mark_charge_rejected(self.db, charge, status=new_status) await AuditService(self.db).log( trace_id=f"approval:{employee_request.request_number}", actor_type="system", actor_user_id=None, action=f"software_addon.{new_status}", resource_type="software_addon_charge", resource_id=str(charge.id), status="success", detail={ "item_id": str(charge.item_id), "target_user_id": str(charge.user_id), "approval_status": approval_status, }, ) @staticmethod def _extract_form_value(form_items: list[dict[str, Any]], names: set[str]) -> Any: for item in form_items or []: if item.get("name") in names: return item.get("value") return None def _extract_asset_relevant_fields( self, request_type: str, form_items: list[dict[str, Any]] | str | None, ) -> dict[str, Any]: # Feishu's ``GET /approval/v4/instances/{code}`` ships ``form`` # as a JSON-encoded STRING, not a list (same shape # ``_synthesize_snapshot_from_instance_form`` in the asset # domain decodes). Until 2026-07-03 this method iterated the # raw string, so every asset-linked APPROVED callback died with # ``AttributeError: 'str' object has no attribute 'get'`` — # webhook 500 → Feishu retry → give-up, leaving the row stuck # ``pending`` until an admin opened the detail drawer (whose # swallowed-exception commit "flipped" the status without # side-effects or audit). Rejections never crashed because the # dispatcher only extracts metadata on approve. Decode # defensively: a malformed payload degrades to "no metadata" # instead of killing the status flip. if isinstance(form_items, str): try: form_items = json.loads(form_items) except (TypeError, ValueError): logger.warning( "asset-field extraction: instance form is a non-JSON string; " "proceeding without form metadata (request_type=%s)", request_type, ) form_items = [] if not isinstance(form_items, list): form_items = [] form_map = { item.get("name"): item.get("value") for item in form_items if isinstance(item, dict) and item.get("name") } # Bilingual lookup — Feishu admins rename widget names freely # (e.g. "Request for Computer Model" → "申请电脑机型"), and the # rename silently broke metadata extraction in the past. Try # every alias for each metadata key and pick the first non-empty # value. Add new aliases here when the Feishu form is updated. def _pick(*aliases: str) -> Any: for alias in aliases: value = form_map.get(alias) if value not in (None, "", [], {}): return value return None if request_type == "laptop_replacement": return self._compact_fields( { "replacement_type": _pick("Request Type", "申请类型"), # 2026-05-20: ``Computer pickup location`` / # ``电脑领取地点`` is the current Feishu widget that # carries the same semantic as the legacy ``Base # Location`` — the office the new laptop ships to. # Keep both old + new aliases so historical rows # still resolve and the current template doesn't # silently drop the field. "base_location": _pick( "Computer pickup location", "Computer Pickup Location", "Computer pick-up location", "电脑领取地点", "领取地点", "Base Location", "Base", "工作地点", "办公地点", ), "requested_model": _pick("Request for Computer Model", "申请电脑机型"), "model_configuration": _pick("Model configuration", "机型配置"), "reason": _pick("Please explain the reason.", "使用电脑说明", "申请理由"), "buyback_requested": _pick( "Whether to purchase the old computer that is currently in use？", "Whether to purchase the old computer that is currently in use?", "是否回购旧电脑", "是否回购", ), # The buyback price the applicant owes when they chose # to repurchase the old computer (是否回购 = 是). Auto- # filled server-side at submit with the policy quote # (see ``_enforce_server_buyback_amount``); captured # here so the admin drawer / audit record carries the # canonical figure. Absent on no-buyback requests. "buyback_amount": self._parse_json_if_possible( _pick( "付款金额", "Payment Amount", "Amount", "回购金额", "Repurchase price", ), ), } ) if request_type == "asset_return": # 2026-05-20: the asset_return Feishu form has NO # Resignation Date widget — the date the applicant supplies # is the ``Return Date`` / ``归还日期`` they intend to drop # the device off. Older code mis-named the field # ``resignation_date`` because the offboarding_buyback flow # used to share this extractor; that alias is preserved as # a legacy fallback for any historical rows still stamped # by the old form. return self._compact_fields( { "return_date": _pick(*WIDGET_ALIASES_ASSET_RETURN_DATE), "return_method": _pick("归还方式", "Return Method"), "return_location": _pick( "归还地点", "Return Location", "Return location", ), "device_serial_number": _pick( "资产序列号", "Device Serial Number", "Computer serial number", "Computer Serial Number", ), "asset_tag": _pick( "资产编号", "Asset Code", "Computer asset number", "Computer Asset Number", "电脑资产编号", ), } ) if request_type == "computer_borrow": interval_value = self._parse_json_if_possible( _pick("DateInterval", "借用时间", "Borrow Period"), ) scenario = _pick( "Requirement Scenario", "需求场景", "借用原因", # 2026-05-20: current Feishu widget renames the field # to the long EN sentence; keep all three aliases so # the alias mismatch can't silently empty the column. "Reason for borrowing and computer model/configuration requirements", # zh-CN rendition of the same widget — a callback whose # form_items arrive in the tenant's Chinese locale would # otherwise empty the column. "借用原因和电脑配置需求", ) if isinstance(interval_value, dict): return self._compact_fields( { "borrow_start": interval_value.get("start"), "borrow_end": interval_value.get("end"), "duration_days": interval_value.get("duration") or interval_value.get("interval"), "scenario": scenario, } ) return self._compact_fields({"scenario": scenario}) if request_type == "offboarding_buyback": return self._compact_fields( { # 2026-05-20: offboarding_buyback Feishu form ships # the resignation date as ``Last working day`` — # the legacy ``Resignation Date`` widget no longer # exists in the current template. Both aliases live # in the shared ``WIDGET_ALIASES_RESIGNATION_DATE`` # constant. "resignation_date": _pick(*WIDGET_ALIASES_RESIGNATION_DATE), # 2026-05-20: ``Computer serial number`` is the # current widget name; legacy ``Buyback Device # Serial Number`` / ``回购设备序列号`` kept as a # fallback for rows captured before the rebuild. "device_serial_number": _pick( "Computer serial number", "Computer Serial Number", "Buyback Device Serial Number", "回购设备序列号", ), # 2026-05-19: ``Computer asset number`` / ``电脑资产编号`` # are the renamed Feishu widget aliases that replaced # the original ``回购设备资产号`` / ``Buyback Asset Code`` # pair when the buyback form was rebuilt. Keep BOTH # historical and current names so the metadata # extractor survives the Feishu admin rename without # silently emitting metadata rows missing ``asset_tag``. "asset_tag": _pick( "回购设备资产号", "Buyback Asset Code", "Computer asset number", "Computer Asset Number", "电脑资产编号", "资产编号", "Asset Code", ), "base_location": _pick( "Base location selection", "Base Location Selection", "Base Location", "Base", "工作地点", "办公地点", ), # 2026-06-17: current Feishu widget ships ``所属主体``; # legacy ``资产所属主体`` / ``Owner Entity`` kept for # prior callback rows. Aliases in # ``WIDGET_ALIASES_OWNER_ENTITY``. "owner_entity": _pick(*WIDGET_ALIASES_OWNER_ENTITY), # 2026-05-20: current Feishu widget ships # ``Repurchase price``; legacy ``Amount`` / ``回购金额`` # kept for prior rows. "buyback_amount": self._parse_json_if_possible( _pick( "Repurchase price", "Repurchase Price", "Amount", "回购金额", "Buyback Amount", "回购价格", ), ), } ) return {} @staticmethod def _compact_fields(payload: dict[str, Any]) -> dict[str, Any]: return {key: value for key, value in payload.items() if value not in (None, "", [], {})} @staticmethod def _parse_json_if_possible(value: Any) -> Any: if not isinstance(value, str): return value value = value.strip() if not value.startswith("{") and not value.startswith("["): return value try: return json.loads(value) except json.JSONDecodeError: return value @staticmethod def _generate_request_number() -> str: return f"REQ-{datetime.now(UTC).strftime('%Y%m%d')}-{uuid4().hex[:4].upper()}" def _filter_applicant_widgets( self, template_key: str, widgets: list[ApprovalTemplateWidget], ) -> list[ApprovalTemplateWidget]: # When the template isn't in the denylist registry at all # (``software_seat``, ``api_key_request``, anything new), pass # widgets through unchanged — matches the pre-2026-05-12 # whitelist's "no key → unfiltered" semantics, so flows that # never opted into a deny / allow list keep their existing # contract (notably any Feishu widget the admin marked # ``visible: false`` intentionally still survives, because we # don't second-guess Feishu for these flows). registry = self._approver_only_widget_names() if template_key not in registry: return list(widgets) deny = registry[template_key] return [ widget for widget in widgets if widget.visible is not False and widget.name not in deny ] @staticmethod def _laptop_replacement_buyback_chosen( widgets: list[ApprovalTemplateWidget], answers: dict[str, Any], ) -> bool: """True when a ``laptop_replacement`` applicant chose to buy back the old computer (是否回购旧电脑 = 是 / Yes). Drives whether the ``付款金额`` amount widget applies: the buyback price is only owed — and only auto-filled + server- enforced — when this is affirmative. Returns False when the buyback radio isn't present or the answer isn't affirmative, so the common "I just want a new laptop, keep my old one" path never carries an amount. Answer resolution is defensive: the radio answer may be keyed by widget id (frontend) or widget name (callback / MCP), and may arrive as the option *text* (``"Yes"`` on the real wire) or, on legacy designer templates, an option value/id — in which case we resolve through the widget's option set and test the matched option's text/value. """ widget = next( ( w for w in widgets if _normalise_widget_name(w.name) in _WIDGET_ALIASES_BUYBACK_CHOICE_NORMALISED ), None, ) if widget is None: return False raw = answers.get(widget.id) if raw in (None, "") and widget.name: raw = answers.get(widget.name) if raw in (None, "", [], {}): return False def _is_affirmative(value: Any) -> bool: return str(value).strip().casefold() in _BUYBACK_AFFIRMATIVE_VALUES if _is_affirmative(raw): return True # Option-id / option-value payloads (legacy designer): map the # answer back to its option and test the human-readable text. for opt in getattr(widget, "options", None) or []: if str(opt.value) == str(raw) or ( opt.text and str(opt.text) == str(raw) ): return _is_affirmative(opt.text) or _is_affirmative(opt.value) return False async def _enforce_server_buyback_amount( self, *, template_key: str, asset: AssetItem | None, answers: dict[str, Any], widgets: list[ApprovalTemplateWidget], current_user: AuthUser, override_currency: str | None = None, ) -> dict[str, Any]: """Overwrite the client-supplied ``Amount`` widget value with the canonical server-computed buyback quote. ``override_currency`` (admin on-behalf path) takes precedence over any client-supplied currency — the leaver isn't present to pick, so the admin's selection in the 代提回购 modal is the source of truth. Applies to ``offboarding_buyback`` (always — every such request is a buyback) and to ``laptop_replacement`` when the applicant chose to buy back the old computer. For ``laptop_replacement`` the caller (``create_request``) has already dropped the amount widget from ``widgets`` when buyback was NOT chosen, so the ``amount_widget is None`` short-circuit below is what scopes the enforcement to the buyback branch — no separate answer re-check is needed here. Without this, the buyback flow is theatre — the frontend renders the amount read-only and writes the server quote into form state, but a determined client (curl, a malicious MCP agent, anyone intercepting the request body) can POST ``answers={"Amount": {"amount": "1", "currency": "CNY"}}`` and ship that tampered value verbatim to Feishu. The approver in Feishu sees only the number the requester sent. Without server-side re-validation an employee could buy back their laptop for 1 CNY. **Name/id-key sibling bypass** (caught by round-2 review, 2026-05-12): a single-key overwrite is NOT enough. ``_serialize_feishu_form`` does ``name_to_id`` resolution at serialise time — so a client who submits BOTH ``answers["amount"] = forged_id_keyed`` AND ``answers["Amount"] = forged_name_keyed`` will see the name-keyed entry CLOBBER our id-keyed canonical value in ``resolved`` (insertion-order: name comes after id). The attack thus reduces to the same outcome as Round-1's P0. Fix: drop **both** the id-keyed and the name-keyed entry (case-insensitive on the name) before writing the canonical one, so there is exactly one Amount answer in the dict and no sibling for the serialiser to resurrect. The employee's choice of *currency* is preserved (the policy explicitly lets the employee switch currency), but the *amount* is forced to whatever the calculator says for that currency. If the client-supplied currency is outside the supported set, the canonical default currency is used instead. Trust boundary: the server NEVER trusts the client for the amount. The audit log records both the client-submitted value (in ``detail.submitted_answers``) and the server-enforced value (in ``detail.resolved_answers`` + the Feishu form payload) so any tamper attempt is forensically visible. """ if ( template_key not in ("offboarding_buyback", "laptop_replacement") or asset is None ): return answers amount_widget = next( (w for w in widgets if w.type == "amount"), None, ) if amount_widget is None: # offboarding_buyback with no amount widget (mis-templated), # or laptop_replacement where ``create_request`` dropped the # amount widget because buyback wasn't chosen — nothing to # enforce. return answers # Collect every key the client might have used to reference the # amount widget — the widget's id, the widget's name, and any # case-variant of the name. Case-insensitive matching defends # against `"AMOUNT"`-style payloads even if ``name_to_id`` # (which is case-sensitive in the serialiser) would not have # resolved them — defence in depth, since a future serialiser # change to fuzzy matching would otherwise reopen the hole. # Use a list (not a set) so the precedence order is # deterministic — important for forensic replay of audit logs # when an attacker submits multiple Amount aliases with # conflicting currencies. Order: widget id (most authoritative) # → exact name → lower → upper. amount_keys: list[str] = [amount_widget.id] if amount_widget.name: for variant in ( amount_widget.name, amount_widget.name.lower(), amount_widget.name.upper(), ): if variant not in amount_keys: amount_keys.append(variant) # Look for a client-supplied currency under ANY of the # known aliases — honour the employee's choice if it's # in the supported set; otherwise let the calculator pick # the default (entity currency → CNY fallback). client_currency: str | None = None for key in amount_keys: client_value = answers.get(key) if isinstance(client_value, dict): raw_currency = client_value.get("currency") if isinstance(raw_currency, str) and raw_currency.strip(): client_currency = raw_currency.strip().upper() break # Admin on-behalf currency wins — the leaver can't pick, so the # admin's modal selection is authoritative over any (absent) client # value. Still validated against the supported set by the quote. if override_currency and override_currency.strip(): client_currency = override_currency.strip().upper() # Compute the canonical quote. Authz was already enforced at # ``_assert_asset_caller_authorized`` upstream; re-quoting here # bypasses the public ``quote_offboarding_buyback`` re-check # because we know the caller is authorised AND we want the same # data-validation errors (missing purchase_cost / date) to # surface as 422 to the submitter, not 403. try: quote = await self.quote_offboarding_buyback( asset_id=asset.id, current_user=current_user, target_currency=client_currency, ) except AppException as exc: # Re-raise unchanged — 422 with stable error codes is the # contract the frontend already handles. raise exc # Strip ALL Amount aliases the client may have planted — # case-insensitive on the widget name to defend against # ``Amount`` / ``amount`` / ``AMOUNT`` siblings — then write # the canonical entry exactly once, keyed by the widget id. # The serialiser's name→id resolution can no longer find an # alternate value to clobber the canonical one. # # Note: whitespace-padded variants (`` Amount ``) and unicode- # confusables (`Amоunt` with Cyrillic о) are NOT caught here — # they slip through case-folding. They ARE dropped by the # serialiser's trailing `widget_ids` filter (see # ``_serialize_feishu_form``), because they don't match any # widget id OR any name in ``name_to_id``. If a future # serialiser refactor relaxes that filter (e.g. fuzzy name # matching), tighten this strip in lockstep — the layered # defence is what makes the bypass un-exploitable today. widget_name_lower = (amount_widget.name or "").lower() resolved_answers = { k: v for k, v in answers.items() if k != amount_widget.id and ( not isinstance(k, str) or k.lower() != widget_name_lower or not widget_name_lower ) } resolved_answers[amount_widget.id] = { "amount": str(quote.amount), "currency": quote.currency, } return resolved_answers async def _resolve_payment_entity( self, asset: AssetItem | None, ) -> AssetLegalEntity | None: """Look up the ``AssetLegalEntity`` for ``asset.purchase_entity``. ``asset.purchase_entity`` is a free-text name (kept for backward compat with pre-2026-05-08 intake rows that don't carry a FK). Matching is exact after safe legal-entity normalisation: both the asset's ``purchase_entity`` and every ``asset_legal_entities.name`` are run through ``_normalise_legal_entity_name`` (NFKC + whitespace cleanup + zero-width strip + presentation-only full-stop cleanup) before equality compare. Why a full table scan + Python filter instead of ``WHERE name = needle``: pre-2026-05-08 intake rows occasionally carry NBSP, fullwidth punctuation, or ZWSP in ``purchase_entity`` (the Excel import preserved whatever the user pasted). A naive equality query against the DB UNIQUE-constrained ``name`` column would miss those rows and the applicant would see the placeholder "[资产 X 未配置采购主体]" + IT admin would have to manually fix the asset row. Loading all entities and normalising in Python catches that pollution without giving up the security property (still exact equality post-normalisation, NOT fuzzy match — bank-info collision would be a security incident). N is small (~10 entities in prod) so the full scan is cheap. On ambiguous matches (two entities normalise to the same canonical form — unlikely given the DB UNIQUE but possible if a future entity uses a confusable name like Latin "A" vs Cyrillic "А"), we abstain with a warning rather than guess. Returns ``None`` when there's no asset, no ``purchase_entity``, no candidate after normalisation, or candidates are ambiguous — the placeholder text in the Payment Info widget is the visible artifact in those cases. """ if asset is None or not asset.purchase_entity: return None # Pre-2026-05-08 intake rows occasionally carry trailing # whitespace, NBSP, or fullwidth punctuation in # ``purchase_entity`` (the Excel import preserved whatever the # user pasted). A literal equality miss here surfaces as the # "[资产 X 未配置采购主体]" placeholder in the Payment Info # widget — applicant submits, IT admin has to re-edit the # asset row. NFKC-normalise both sides on whitespace + width # before comparing so trivial pollution resolves cleanly. We # do NOT fuzzy-match (bank-info collision would be a security # incident) — exact match against the normalised form only. needle = _normalise_legal_entity_name(asset.purchase_entity) if not needle: return None # Pull candidates whose trimmed name COULD match — small N # (single-digit entities in prod), so a full scan is cheap and # we get to apply NFKC normalisation in Python without a # DB-side function index. Exact post-normalisation match wins; # ambiguous matches log a warning and abstain. rows = ( await self.db.execute(select(AssetLegalEntity)) ).scalars().all() candidates = [ r for r in rows if _normalise_legal_entity_name(r.name) == needle ] if not candidates: return None if len(candidates) > 1: logger.warning( "purchase_entity %r matched %d legal entities after safe " "normalisation — abstaining from autofill to avoid emitting " "the wrong bank info. Candidate ids: %s", asset.purchase_entity, len(candidates), [str(r.id) for r in candidates], ) return None return candidates[0] @staticmethod def _format_payment_info(entity: AssetLegalEntity) -> str: """Render an ``AssetLegalEntity`` row as multi-line payment info. Used to fill the laptop_replacement / offboarding_buyback "Payment Information" / "付款信息" widget. Bank fields lead, tax / registration trails — same order the IT admin's procurement sheet uses. Lines with no value are skipped so the rendered text never has empty "开户行：" rows. """ rows: list[tuple[str, str | None]] = [ ("公司主体", entity.name), ("国家/地区", entity.country_or_region), ("税号", entity.tax_identifier), ("注册地址", entity.registered_address), ("电话", entity.phone), ("开户行", entity.bank_name), ("账户名", entity.bank_account_name), ("账号", entity.bank_account), ("开户行地址", entity.bank_address), ("SWIFT", entity.swift_code), ("币种", entity.currency), ("账户类型", entity.account_type), ] return "\n".join( f"{label}：{value}" for label, value in rows if value ) @staticmethod def _resolve_location_default( asset: AssetItem | None, person: Any | None, ) -> str | None: """Pick the best free-text location string for laptop_replacement / offboarding_buyback / asset_return. Priority (user-confirmed 2026-05-21): 1. ``asset.location`` — the device's current physical location (asset_items.location is the source of truth for where the device sits today; laptop_replacement is about replacing THIS device, so the new one ships to the same place by default) 2. ``person.work_location_name`` — applicant's primary office from the SuccessFactors sync; ground truth for employees whose asset row predates the location backfill 3. ``person.city`` — last-resort, only when the SF-resolved ``work_location_name`` is unset Returns ``None`` when every source is empty — the widget stays blank and the applicant fills it manually (which was the pre-fix behaviour, just for everyone). """ if asset is not None: loc = (asset.location or "").strip() if hasattr(asset, "location") else "" if loc: return loc if person is not None: for attr in ("work_location_name", "city"): val = getattr(person, attr, None) if isinstance(val, str) and val.strip(): return val.strip() return None @staticmethod def _is_location_widget(name: str | None) -> bool: """True for the free-text location widget on laptop_replacement + offboarding_buyback + asset_return. Mirrors the alias set ``_extract_asset_relevant_fields`` uses for the same conceptual field, so the auto-fill and the metadata extractor agree on what counts as "the location widget". Excludes payment / entity widgets that happen to contain "address" — those are handled by the Payment-Information overwrite branch above and must NOT be clobbered by location text. """ if not name: return False n = name.strip().lower() if not n: return False # Negative guard: anything that looks like an entity / payment # / bank dump field MUST NOT be auto-filled with location text. # The Payment-Information branch above handles those. if any(k in n for k in ("payment", "付款", "bank", "银行", "tax", "税")): return False # Positive aliases — match the substrings used by # ``_extract_asset_relevant_fields`` for both laptop_replacement # ("Computer pickup location"/"电脑领取地点") and asset_return # ("Return Location"/"归还地点") and offboarding_buyback # ("Base Location"/"工作地点"/"办公地点"). return any( kw in n for kw in ( "pickup location", "pick-up location", "return location", "base location", "工作地点", "办公地点", "电脑领取地点", "领取地点", "归还地点", "地点", ) ) @staticmethod def _resolve_widget_option_value( widget: ApprovalTemplateWidget, candidate: str | None, ) -> str | None: """Resolve a human label/value to a constrained widget's value. Feishu stores opaque option ids while asset/HR rows store readable office names. Normalising both sides lets ``Japan`` match ``japan`` without weakening the existing allowed-option boundary. """ if not candidate or widget.type not in {"radioV2", "select"}: return None wanted = _normalise_widget_name(candidate).casefold() if not wanted: return None for option in widget.options: if wanted in { _normalise_widget_name(option.value).casefold(), _normalise_widget_name(option.text).casefold(), }: return option.value return None @staticmethod def _augment_answers_with_asset( asset: AssetItem | None, answers: dict[str, Any], widgets: list[ApprovalTemplateWidget], *, payment_entity: AssetLegalEntity | None = None, person: Any | None = None, ) -> dict[str, Any]: resolved_answers = dict(answers) if not asset and not person: return resolved_answers # Pre-resolve the location string once — used by the # location-widget branch below. Computed even when both asset # and person are present so we honour the user-confirmed # priority (asset.location → person.work_location_name → city). location_default = ApprovalService._resolve_location_default(asset, person) payment_text: str | None = None if payment_entity is not None: payment_text = ApprovalService._format_payment_info(payment_entity) for widget in widgets: # Payment Information UNCONDITIONAL overwrite (2026-05-19 # security tightening, code-review pass 2): the widget is # rendered ``disabled`` on the frontend, but ``disabled`` is # NOT an authoritative server-side trust boundary — a # malicious client (curl, MCP agent, intercepted XHR) can # POST forged bank info under the widget's id. Without this # overwrite, the approver's card would show # attacker-controlled bank account details that LOOK like a # legitimate auto-fill from purchase_entity. The fix: # ALWAYS overwrite the Payment Information slot with the # server-resolved entity text (or the placeholder), then # short-circuit the rest of the per-widget loop. # # Match the whitespace-stripped widget name against the # explicit alias set (case-sensitive — current Feishu form # ships exactly ``Payment Information`` / ``Payment # information`` / ``付款信息``). Strict equality keeps an # unrelated field whose label happens to contain "payment" # / "付款" from accidentally inheriting the bank dump. # # 2026-05-20 SECURITY FIX: the alias set MUST include the # lowercase ``Payment information`` variant — that is the # actual widget name on the offboarding_buyback Feishu # form. Without the lowercase alias the server-side # overwrite was BYPASSED on every buyback approval, # letting a malicious client (curl / MCP agent / intercepted # XHR) POST forged bank-account text under the widget's id # and have it survive into the approver's review card. # Centralised in ``WIDGET_ALIASES_PAYMENT_INFO`` so future # Feishu renames update a single constant. # # ``Payment Information`` was removed from # ``_approver_only_widget_names`` on 2026-05-18 (it's now # applicant-facing, auto-filled from purchase_entity), so # the two sets intentionally no longer mirror each other — # this overwrite + the asset_legal_entities backstop are # what enforce the no-tamper guarantee. if _normalise_widget_name(widget.name) in _WIDGET_ALIASES_PAYMENT_INFO_NORMALISED: if asset is None: # Person-only auth context (e.g. submitted with no # asset linked) — leave Payment Information alone # rather than printing a misleading "[asset 未配置 # 主体]" placeholder. continue if payment_text: resolved_answers[widget.id] = payment_text else: # No matching entity → write a placeholder that # names the missing purchase_entity. Without this # stamp the required-field guard later in # ``_serialize_feishu_form`` would 400 every # laptop_replacement against an asset whose # purchase_entity hasn't been backfilled (pre- # 2026-05-08 intake rows). The placeholder # doubles as a nudge to the approver to fix the # asset row. resolved_answers[widget.id] = ( f"[资产 {asset.asset_tag or asset.id} 未配置采购主体" + ( f"（{asset.purchase_entity}，未在「公司主体信息」登记）" if asset.purchase_entity else "" ) + "，请联系 IT 补全后再走流程]" ) continue # 所属主体 / Owner Entity single-select (2026-06-17, user- # requested): a 单选 widget that mirrors the 公司主体 line of # Payment Information. Same server-derived value # (asset.purchase_entity → legal entity), so stamp it from # the resolved entity instead of trusting the client — keeps # the constrained option consistent with the bank dump the # approver sees and resists a forged-payload entity swap. # Unlike Payment Information we can't write a free-text # placeholder: radioV2 / select accept ONLY a value in their # option set, and ``_serialize_feishu_form`` 400s anything # else. So when the resolved entity name isn't a registered # option (or no entity resolved) we leave the applicant's own # pick alone and let the required-field / option-set guards # decide. Placed before the empty-only skip below so the # server value authoritatively overwrites a client-sent one. if ( widget.type in {"radioV2", "select"} and _normalise_widget_name(widget.name) in _WIDGET_ALIASES_OWNER_ENTITY_NORMALISED ): if payment_entity is not None and widget.options: # Send the option's canonical ``value`` (Feishu's # radioV2 keys on value), matching either the option # value or its display text against the entity name. match = ApprovalService._resolve_widget_option_value( widget, payment_entity.name, ) if match is not None: resolved_answers[widget.id] = match continue # Current production buyback uses a required constrained # ``Base location selection``. Map the readable asset/person # location to Feishu's opaque option value when it is an exact # normalised match. If it does not match, preserve any template # default and let the specialized modal provide an explicit pick. if ( widget.type in {"radioV2", "select"} and location_default and ApprovalService._is_location_widget(widget.name) ): location_option = ApprovalService._resolve_widget_option_value( widget, location_default, ) if location_option is not None: resolved_answers[widget.id] = location_option continue if resolved_answers.get(widget.id) not in (None, ""): continue # Location widget auto-fill (2026-05-21, user-requested): # Feishu admin changed 地点 from a constrained select to a # free-text input on laptop_replacement, breaking the # implicit "the office defaults to where you sit" UX. # Backstop here: when the location widget is empty AND we # can resolve a default (asset.location → person.work_ # location_name → person.city), stamp it. Frontend also # pre-fills via the same resolution, but the backstop # catches MCP / replay callers and any future client that # skipped the prefill path. # # Only fires for free-text widget types (input / textarea) # — Feishu's radioV2 / select branch ships its own # option-set guard in ``_serialize_feishu_form`` and a # location string almost never matches a constrained # option's value, so writing into a constrained widget # would just be deleted at the boundary. if ( widget.type in {"input", "textarea"} and location_default and ApprovalService._is_location_widget(widget.name) ): resolved_answers[widget.id] = location_default continue # 2026-05-20 (Round 2 audit F6): match the per-widget # heuristics against ``widget.name`` ONLY. The previous # ``f"{widget.name} {widget.id}".lower()`` form let a Feishu # UUID containing substrings like ``model`` / ``serial`` / # ``code`` false-positive an asset prefill onto an unrelated # widget (e.g. a reason textarea whose UUID embedded # ``model`` got auto-stamped with the asset model). The # field is only ever populated when the answer is currently # empty, but MCP-tool callers can land empty answers + a # fresh asset — exactly the path where this used to misfile # asset metadata into the wrong slot. if asset is None: continue field_text = (widget.name or "").lower() if ("serial" in field_text or "序列号" in field_text) and asset.serial_number: resolved_answers[widget.id] = asset.serial_number continue # ``computer asset number`` (2026-05-26) — matches the # post-2026-05-19 buyback Feishu rename ``Computer asset # number`` / ``电脑资产编号``. Outer ``asset`` token guards # against ``phone number`` / ``mobile number`` false-fires. if ( "资产编号" in field_text or "asset code" in field_text or "资产号" in field_text or ("asset" in field_text and "number" in field_text) ) and asset.asset_tag: resolved_answers[widget.id] = asset.asset_tag continue if ("品牌" in field_text or "brand" in field_text) and asset.brand: resolved_answers[widget.id] = asset.brand continue if ("model" in field_text or "型号" in field_text) and asset.model: resolved_answers[widget.id] = asset.model continue return resolved_answers @staticmethod def _parse_template_widgets(raw_form: str | list[dict[str, Any]]) -> list[ApprovalTemplateWidget]: # Feishu-controlled data — harden exactly like the sibling instance # parser ``_extract_asset_relevant_fields`` was after the #1622 # incident: a truncated/malformed definition must NOT 500 every # submission of the flow. Bad JSON → treat as "no widgets"; a # non-list top level or a non-dict element is skipped rather than # raising ``AttributeError`` on ``widget.get``. if isinstance(raw_form, str): try: widgets = json.loads(raw_form or "[]") except (TypeError, ValueError): widgets = [] else: widgets = raw_form if not isinstance(widgets, list): widgets = [] parsed: list[ApprovalTemplateWidget] = [] for widget in widgets: if not isinstance(widget, dict): continue raw_option = widget.get("option") if isinstance(widget.get("option"), dict) else None option_items = widget.get("option") if isinstance(option_items, dict): option_items = option_items.get("options") or option_items.get("value") or [] options = [] if isinstance(option_items, list): options = [ApprovalTemplateOption.model_validate(option) for option in option_items] wtype = widget.get("type", "input") # Extract the Feishu admin's pre-baked default for this # widget. Feishu's v4 form designer stores it under # ``widget_default_value`` with an ``enable_default_value`` # gate flag — discovered 2026-05-21 by dumping the prod # laptop_replacement definition: an admin-baked # "Request Type=Retirement Replacement / 机型=MacBook Pro" # widget ships ``enable_default_value: True`` + ``widget_ # default_value: 'm28rckrw-...'`` (the option's machine # value). Older docs / examples sometimes mention ``default`` # / ``defaultValue`` / ``value`` — keep them as legacy # fallbacks for any tenant still on the v3 schema, but # priority is widget_default_value (the only key the # current PingCAP tenant actually uses). # # For constrained widgets we resolve the raw value against # the registered option set up front: a stale default # (admin renamed an option but forgot to fix the default) # otherwise rides to ``_serialize_feishu_form`` and trips # the APPROVAL_FIELD_INVALID_OPTION backstop. ``None`` # everywhere means "no default" — frontend leaves blank. default_value: Any = None if wtype != "text": # Primary: widget_default_value (gated by # enable_default_value=True; an explicit False means # admin disabled the default even if the value field # carries a leftover value). wdv = widget.get("widget_default_value") enabled = widget.get("enable_default_value") if wdv not in (None, "", []) and enabled is not False: default_value = wdv else: # Legacy fallbacks for older Feishu schemas. for key in ("default", "defaultValue", "value"): candidate = widget.get(key) # ``value`` on radio widgets carries the option # array (already handled above), not a default — # only honour it when it's a scalar / list of # scalars / a string. if key == "value" and isinstance(candidate, (list, dict)): continue if candidate not in (None, "", []): default_value = candidate break # Fallback for radio/checkbox/select: scan the option # list for an entry flagged ``default`` / ``selected``. if default_value is None and isinstance(option_items, list): pre_selected = [ opt.get("value") for opt in option_items if isinstance(opt, dict) and (opt.get("selected") is True or opt.get("default") is True) ] if pre_selected: default_value = ( pre_selected if wtype == "checkboxV2" else pre_selected[0] ) # Constrain to a valid option for single-select widgets # so a stale default doesn't bypass the boundary check # in ``_serialize_feishu_form``. if ( default_value is not None and wtype in {"radioV2", "select"} and options ): allowed = {opt.value for opt in options} | { opt.text for opt in options if opt.text } if str(default_value) not in allowed: default_value = None # checkboxV2: filter to only options that exist; a list # of one bad value otherwise still triggers 1390001. if ( default_value is not None and wtype == "checkboxV2" and isinstance(default_value, list) and options ): allowed = {opt.value for opt in options} | { opt.text for opt in options if opt.text } filtered = [v for v in default_value if str(v) in allowed] default_value = filtered or None # Date / dateInterval widgets: Feishu ships a stringy # ``'0'`` as the "no admin default" sentinel value on # the date types (verified 2026-05-21 against prod # asset_return ``归还日期``, offboarding_buyback # ``Last working day``, computer_borrow # ``DateInterval``). If we pass ``'0'`` straight to # Antd's DatePicker via ``setFieldsValue`` the # component throws on the unparseable value and # crashes the entire create-Modal render. Treat any # falsy / non-date scalar on these widget types as # no-default. # # 2026-05-21 round-2 review P3: also defensively # reject ANY numeric default on date widgets. Feishu # COULD legitimately ship an epoch-millis number in # the future (e.g. an admin set a static date # default); we'd need a different deserialiser to # turn that into the ISO string Antd expects. # Until that path is implemented + tested, drop # numeric defaults rather than crash the modal. if ( default_value is not None and wtype in {"date", "dateInterval"} ): if default_value in (0, "0", "", None): default_value = None elif isinstance(default_value, (int, float)): # Future work: convert epoch-millis to ISO # before surfacing to the frontend. For now, # drop to avoid a DatePicker crash. default_value = None description: str | None = None if wtype == "text": # Feishu's static description widgets carry their # multi-line display content under one of several # schema keys depending on form-designer version # ("value" is the modern field, "extra" / "description" # are legacy, "name" is the fallback when the designer # stores everything on the label). Try each in order # so we don't drop a description block just because a # newer Feishu version moved the field. for key in ("value", "extra", "description"): candidate = widget.get(key) if isinstance(candidate, str) and candidate.strip(): description = candidate break if description is None: # Last resort — some admins author the entire # description into the widget's ``name`` (which # doubles as the heading for non-text widgets). name_val = widget.get("name") if isinstance(name_val, str) and name_val.strip(): description = name_val parsed.append( ApprovalTemplateWidget( id=widget["id"], name=widget.get("name", ""), type=wtype, required=widget.get("required", False), visible=widget.get("visible", True), printable=widget.get("printable", True), display_condition=widget.get("display_condition"), options=options, raw_option=raw_option, description=description, default_value=default_value, ) ) return parsed @staticmethod def _parse_template_stages( definition: dict[str, Any], ) -> list[ApprovalTemplateStage]: """Extract the static approval node chain from a Feishu definition. Feishu's ``GET /approval/v4/approvals/{approval_code}`` ships the form widgets AND the workflow node list — we historically only parsed the former and dropped the latter on the floor. Surfacing stages lets the self-service "create request" UI render a Steps strip before the applicant submits, and lets the post-submit detail modal grey-overlay un-traversed future nodes (instead of only showing nodes the task_list has actually touched). Schema variations across Feishu API versions: * ``node_list`` — modern v4 templates (most common) * ``process_node_list`` — older v3-style payload some tenants still return * ``viewers`` — CC-only entries (skipped here; they don't gate the flow) Each node entry typically carries: ``node_id`` / ``custom_node_id`` — stable id ``name`` / ``node_name`` — display label ``node_type`` — AND / OR / SEQUENCE / CC ``approver_chosen_value`` — pre-resolved approver list (user_id + name when admin statically named approvers) ``approver_uid_list`` — bare user_id list (legacy) We're best-effort here: missing keys silently no-op so a template that uses a dynamic approver resolver (e.g. "the applicant's manager") still surfaces the node name + type even when names aren't known until submission. The frontend shows "由系统决定" for empty approver_names. """ if not isinstance(definition, dict): return [] raw_nodes = ( definition.get("node_list") or definition.get("process_node_list") or [] ) if not isinstance(raw_nodes, list): return [] # Defence-in-depth bound — Feishu workflows rarely exceed # ~10 nodes; clamp to 100 so a malformed response can't bloat # the cached template payload. raw_nodes = raw_nodes[:100] parsed: list[ApprovalTemplateStage] = [] for entry in raw_nodes: if not isinstance(entry, dict): continue node_id_raw = ( entry.get("node_id") or entry.get("custom_node_id") or entry.get("approval_node_id") ) name_raw = ( entry.get("name") or entry.get("node_name") or entry.get("custom_node_name") ) node_type = entry.get("node_type") or entry.get("type") # Resolve approver names from any of the common shapes. # # ``approver_chosen_value`` items can carry any of: # * ``name`` / ``user_name`` — statically-pinned user # * ``role_name`` — role-based resolver # ("财务", "IT 资产管理员") # * ``department_name`` — department resolver # ("财务部") # Older code only matched the first pair, so role/department # nodes (very common on the offboarding_buyback flow) landed # with an empty approver list and rendered "该环节暂无审批 # 人" on the post-submit timeline — user-visible nag bug, # 2026-05-26. Surfacing role / department labels is enough # signal that the employee can answer "who's next" without # having to ask IT. approver_names: list[str] = [] chosen = entry.get("approver_chosen_value") if isinstance(chosen, list): for item in chosen[:50]: if isinstance(item, dict): n = ( item.get("name") or item.get("user_name") or item.get("role_name") or item.get("department_name") ) if isinstance(n, str) and n.strip(): approver_names.append(n.strip()) elif isinstance(item, str) and item.strip(): approver_names.append(item.strip()) # Legacy ``approver_uid_list`` is user_id-only — surface # the bare ids so the frontend at least shows something # rather than rendering an empty node. The detail modal # already calls the user-name enricher for instance data; # for the create-request UI the bare id is acceptable. uid_list = entry.get("approver_uid_list") or entry.get("approver_user_ids") if not approver_names and isinstance(uid_list, list): for uid in uid_list[:50]: if isinstance(uid, str) and uid.strip(): approver_names.append(uid.strip()) parsed.append( ApprovalTemplateStage( node_id=str(node_id_raw) if node_id_raw else None, name=name_raw if isinstance(name_raw, str) and name_raw.strip() else None, node_type=str(node_type).upper() if isinstance(node_type, str) else None, approver_names=approver_names, ) ) return parsed def _serialize_feishu_form( self, widgets: list[ApprovalTemplateWidget], answers: dict[str, Any], ) -> str: serialized: list[dict[str, Any]] = [] # Build name→id map so callers can provide answers by widget name name_to_id = {w.name: w.id for w in widgets if w.name} # Defence-in-depth (added 2026-05-12 alongside the buyback # tamper-bypass fix in ``_enforce_server_buyback_amount``): # process name-keyed entries FIRST, then id-keyed entries — # so an id-keyed canonical value ALWAYS wins over a name-keyed # sibling pointing at the same widget. Without this ordering, # a payload with both keys (``"amount": id_value`` AND # ``"Amount": name_value``) lets the name-keyed entry clobber # the id-keyed one because the latter was assigned first and # the loop's last-write-wins semantics favoured the name. # See buyback security review round 2 for the original attack. resolved: dict[str, Any] = {} widget_ids = {w.id for w in widgets} # Pass 1: name-keyed entries (resolve to id, but only if the # id slot is not yet taken — id wins on collision below). for key, value in answers.items(): if key in name_to_id and name_to_id[key] not in resolved: resolved[name_to_id[key]] = value # Pass 2: id-keyed (or unknown-keyed) entries — overwrite any # name-keyed value with the canonical id-keyed one. for key, value in answers.items(): if key in name_to_id and name_to_id[key] != key: # This was already a name key; skip to keep pass-1 # behaviour unless the same string is ALSO a widget id # (then it's already an id and falls through below). continue resolved[key] = value # Strip stray non-widget-id keys (name-keys that collided with # widget ids, or unknown entries) — defensive but cheap. resolved = {k: v for k, v in resolved.items() if k in widget_ids} # Pre-compute display-condition visibility so the required-check # below skips widgets that Feishu's UI would hide for this set # of answers (e.g. ``Police Report Receipt`` only when Request # Type=Lost; the buyback photo upload only when ``是否回购=是``). # Without this, conditional widgets marked ``required=True`` in # the template definition would 400 every "happy path" # submission with APPROVAL_FIELD_REQUIRED — the exact failure # mode the denylist was a stopgap for. The evaluator is # fail-open (unknown schema → assume visible) so a Feishu # version that changes display_condition shape doesn't block # legitimate submissions; it'd just stop hiding the widget # until the evaluator is extended. visibility = self._evaluate_display_conditions(widgets, resolved) for widget in widgets: if widget.type == "text": continue # ``dateInterval`` (computer_borrow 借用时间). Between # 2026-05-19 and 2026-08-14 this widget was SKIP'd outright # — every value shape we'd tried 400'd with ``1395006: # 控件值不合法或者为空``, and Feishu accepts a submission # that omits the widget even though the template marks it # required. The cost of that workaround: the borrow window # the applicant typed into Helix never reached the # approver's Feishu card, so 电脑借用申请 showed a reason # and a pick-up location and no dates at all. # # ``_date_interval_wire_value`` now ships the object shape # Feishu itself stores for natively-submitted borrow # windows (see its docstring for the read-off matrix — the # old attempts used ``duration`` instead of ``interval`` # and omitted ``timezoneOffset``). A half-filled or # unparseable window still serialises to ``""`` and is # dropped here rather than 400'ing the applicant, and if # Feishu ever rejects the object anyway, # ``_create_or_recover_asset_provider_instance`` resubmits # without the widget — so the worst case is the old # behaviour, never a blocked submission. # # The window is ALSO stashed on # ``EmployeeRequest.detail.borrow_dates`` (see # ``create_request``); the auto-execute path keeps reading # from there, not from Feishu's echo. if widget.type == "dateInterval": if not visibility.get(widget.id, True): continue interval_value = self._date_interval_wire_value( resolved.get(widget.id), ) if interval_value in (None, "", {}): continue serialized.append({ "id": widget.id, "type": widget.type, "value": interval_value, }) continue if not visibility.get(widget.id, True): # Widget is conditionally hidden — don't enforce required, # don't ship its answer to Feishu (Feishu would reject # an answer for a hidden widget on some templates). continue answer = resolved.get(widget.id) if widget.required and answer in (None, "", [], {}): raise AppException( code="APPROVAL_FIELD_REQUIRED", message=f"Missing required approval field: {widget.name}", status_code=400, ) if answer in (None, "", [], {}): continue # Defence-in-depth: ``radioV2`` widgets accept ONLY a value # from the registered option set — Feishu rejects anything # else with API 1390001 ("控件的值不存在单选框的选项中"). The # frontend's asset-prefill path historically pushed # ``asset.model = "Mac Pro" / "MacBook Pro I5"`` (free-form # text stored in CMDB) straight into the laptop_replacement # template's "申请电脑机型" widget (options: MacBook Air / # MacBook Pro / ThinkPad X1 / 非标机型) and the request # exploded at the Feishu boundary instead of at our 4xx # boundary. Reject here so the surface error is a clean # Helix 400 naming the widget + the offending value + # the legal options — and so future callers (MCP tool, # direct API, replay scripts) inherit the same guard # without each one re-implementing it. The frontend # prefill path has been hardened in parallel; this is the # backstop, not the primary defence. if widget.type == "radioV2" and widget.options: allowed = {opt.value for opt in widget.options} | { opt.text for opt in widget.options if opt.text } if str(answer) not in allowed: raise AppException( code="APPROVAL_FIELD_INVALID_OPTION", message=( f"Field '{widget.name}' value '{answer}' is not " f"in the allowed options: " f"{[opt.text or opt.value for opt in widget.options]}" ), status_code=400, ) # Pre-launch P0 fix (Round 45, CR-304): extend the option-set # backstop to multi-select widgets. radioV2 was the original # bug class (PR #344) but checkboxV2 + select share the same # Feishu API-1390001 failure mode — Feishu rejects values # that aren't in the registered option set. MCP/programmatic # callers don't go through the frontend's option resolver, so # they need the backend guard. Each element must be in the # option set; empty list was already filtered above. if widget.type in ("checkboxV2", "select") and widget.options: allowed = {opt.value for opt in widget.options} | { opt.text for opt in widget.options if opt.text } values = ( answer if isinstance(answer, (list, tuple)) else [answer] ) bad = [ str(v) for v in values if str(v) not in allowed ] if bad: raise AppException( code="APPROVAL_FIELD_INVALID_OPTION", message=( f"Field '{widget.name}' values {bad} not in " f"the allowed options: " f"{[opt.text or opt.value for opt in widget.options]}" ), status_code=400, ) # Feishu's ``amount`` widget is multi-currency: the # ``currencyRange`` list in the widget's ``option`` blob # (e.g. offboarding_buyback "Repurchase price" ships # ``["CNY","USD","JPY","SGD","MYR"]``) drives which currency # the card renders. When the form value is a bare number # Feishu falls back to ``currencyRange[0]`` (CNY), which is # why a Japan buyback of JPY 10 400 was mislabelled "CNY". # The currency travels as a SIBLING key on the form item, # never inside ``value`` — see ``_amount_wire_currency``. currency_range: list[str] | None = None if widget.type == "amount" and isinstance(widget.raw_option, dict): raw_range = widget.raw_option.get("currencyRange") if isinstance(raw_range, list) and raw_range: currency_range = [str(c) for c in raw_range] item: dict[str, Any] = { "id": widget.id, "type": widget.type, "value": self._serialize_widget_value(widget.type, answer), } # A currency label on an EMPTY amount is meaningless — an # answer dict that carried a currency but no ``amount`` # serialises to ``""`` and must not drag a stray currency # key onto the wire. if widget.type == "amount" and item["value"] != "": wire_currency = self._amount_wire_currency( answer, currency_range, ) if wire_currency: item["currency"] = wire_currency serialized.append(item) return json.dumps(serialized, ensure_ascii=False) @staticmethod def _date_interval_wire_value(answer: Any) -> dict[str, Any] | str: """Borrow-window wire value — see :func:`app.integrations.feishu_widget_values.date_interval_wire_value` for the shape, where it was read from, and the three details the 2026-05 attempts got wrong. Kept as a method so the serializer and its regression tests have one obvious entry point. """ return date_interval_wire_value(answer) @staticmethod def _form_without_date_interval(form_payload: str) -> str | None: """See :func:`app.integrations.feishu_widget_values.form_without_date_interval`. """ return form_without_date_interval(form_payload) @staticmethod def _amount_wire_currency( answer: Any, currency_range: list[str] | None, ) -> str | None: """Currency code to ship as the form item's sibling ``currency`` key for an ``amount`` widget, or ``None`` to let Feishu apply ``currencyRange[0]``. **Wire contract (verified against live Feishu 2026-08-13** on a throwaway approval definition whose amount widget carried ``currencyRange = ["CNY","USD","JPY"]``, so no production flow was used as a test bed): ============================================== ============= ========== form item create result currency ============================================== ============= ========== ``{"value": "816.61"}`` accepted CNY (= [0]) ``{"value": {"amount":…, "currency":…}}`` **1395006** — ``{"value": "816.61", "ext": "{currency…}"}`` accepted CNY (ignored) ``{"value": "816.61", "ext": {currency…}}`` accepted CNY (ignored) ``{"value": "816.61", "currency": "USD"}`` accepted **USD** ============================================== ============= ========== So the currency is a **sibling key on the form item**, never a member of ``value``. The nested-object shape #2558 inferred from the image-widget precedent is exactly the shape Feishu rejects: every non-CNY buyback since 2026-08-04 died with ``1395006 validate form error 控件值不合法或者为空。控件= widget17284665706080001`` (the "Repurchase price" widget) — the USD/USA 代提回购 report that prompted this fix. The dict shape's rejection was already recorded in ``app/domains/apikey/service.py::_autofill_answers``; the approval domain re-derived it the hard way. Feishu stores the accepted result as ``value: `` + ``ext: {"currency": …}`` — byte-identical in structure to a human's currency pick in the Feishu mobile form, which is why the finance "confirm receipt of payment" **edit** node reads it back fine (the failure mode of the stringified-dict shape in #1204). Emit the key only for a currency that is BOTH non-default and actually offered by the widget: the default is what Feishu applies anyway, and an unknown currency (a stale row, or a currency the widget doesn't list) must not be pushed at Feishu. The CNY path therefore stays byte-for-byte what it has been for every shipped buyback — zero regression surface. """ if not isinstance(answer, dict): return None raw_currency = answer.get("currency") currency = str(raw_currency).strip().upper() if raw_currency else "" allowed = [str(c).strip().upper() for c in (currency_range or [])] default = allowed[0] if allowed else "" if currency and default and currency != default and currency in allowed: return currency return None @staticmethod def _evaluate_display_conditions( widgets: list[ApprovalTemplateWidget], answers: dict[str, Any], ) -> dict[str, bool]: """Return ``{widget_id: is_visible}`` for every widget. Thin pass-through to the shared :mod:`app.integrations.feishu_form_visibility` evaluator — every Feishu-backed flow (cloud_scope_request, dns_request, domain_registration, vpn_application, announcement_request, ai_tool_application, sso_apply, api_key_request) calls the same module so the four schema variants + every operator behave identically across the platform. See the module docstring for the schema variations and fail-open contract. """ from app.integrations.feishu_form_visibility import ( evaluate_display_conditions, ) return evaluate_display_conditions(widgets, answers) @staticmethod def _serialize_widget_value(widget_type: str, answer: Any) -> Any: # Feishu's ``POST /approval/v4/instances`` accepts ``form`` as # a JSON-stringified array of widget dicts. The encoding of # each ``value`` field varies by widget type: # # - input / textarea / radioV2 / date: plain string. # - image / imageV2 / attachmentV2: **RAW JSON array of plain # file-code strings** — e.g. ``"value": ["img_v3_xxx", # "file_v2_yyy"]``. NOT stringified at the inner level, # NOT wrapped in ``{image_key|file_key|file_token: ...}`` # objects. Feishu reads the token prefix internally # (``img_v2_*`` / ``img_v3_*`` / ``file_v2_*``) to render # each entry as photo or video. # - amount: stringified object ``"{\"amount\":..., \"currency\":...}"`` # when the widget is currency-enabled. # - dateInterval: stringified object (currently SKIP'd upstream). # - checkboxV2 / select: raw or stringified array (not yet # verified against live Feishu). # # ``_serialize_widget_value`` returns the NATIVE Python value # the wire form needs (str / list / dict / number); # ``_serialize_feishu_form`` then does ONE outer json.dumps to # produce the final ``form`` string. So for image widgets the # function returns a Python list, which encodes as a real # JSON array on the wire — not a string. # # **History — four wrong shape attempts before locking this # in** (all verified live against open.feishu.cn 2026-05-26): # 1. ``"value": "[{\"image_key\":\"...\"}]"`` (stringified # array of objects, pre-2026-05-26 code from PR #498): # Feishu 1390001 ``控件值不是数组`` — Feishu sees value # as a string and rejects. # 2. ``"value": [{"image_key": "..."}]`` (raw array of # objects, 2026-05-26 attempt #1): Feishu 1390001 # ``Widget value is not string`` — Feishu's misleading # error wording actually means "array elements are not # strings". # 3. ``"value": "[\"img_v3_...\"]"`` (stringified array of # plain strings, 2026-05-26 attempt #2): Feishu 1390001 # ``控件值不是数组`` again — value field still wrapped # in a JSON string, not a real array. # 4. CURRENT — raw JSON array of plain strings. # # Feishu's two 1390001 error variants matter: # - ``控件值不是数组`` / ``Widget value is not an array`` → # the value field itself is not a JSON array (it's a # string or other type). # - ``Widget value is not string`` → the value field IS an # array but its elements aren't strings. # Reading them together: value must be ``[str, str, ...]``. if widget_type == "text": raise AppException( code="APPROVAL_INTERNAL_ERROR", message="text widgets must not be serialised back to Feishu", status_code=500, ) if widget_type in {"input", "textarea", "radioV2"}: return str(answer) if widget_type == "date": # Feishu's ``date`` widget value MUST be RFC3339 with an # explicit timezone offset, per the official 创建审批实例 # widget value spec (e.g. ``"2019-10-01T08:12:01+08:00"``). # A bare ``"YYYY-MM-DD"`` (the shape the asset-request # frontend submits via Antd DatePicker.format='YYYY-MM-DD') # is rejected with ``1395006 validate form error 控件值不 # 合法或者为空 控件= ``. Coerce to noon # Asia/Shanghai so the date in the approval card matches # the date the applicant picked regardless of approver # timezone (DST-safe midpoint, can't drift to the prior/ # next day in either CN or US-based approver locales). # # Idempotent: if the upstream caller (MCP / replay) already # supplied an RFC3339 string with a ``T`` separator we # pass it through unchanged. s = str(answer).strip() if "T" not in s and len(s) == 10: # YYYY-MM-DD → YYYY-MM-DDT12:00:00+08:00 return f"{s}T12:00:00+08:00" return s if widget_type == "dateInterval": return ApprovalService._date_interval_wire_value(answer) if widget_type == "amount": # Feishu's 金额 widget is MULTI-CURRENCY: the widget's # ``option.currencyRange`` (e.g. offboarding_buyback ships # ``["CNY","USD","JPY","SGD","MYR"]``) is the set of pickable # currencies and ``currencyRange[0]`` is the default the card # renders when the form item carries no currency of its own. # # ``value`` is ALWAYS a plain numeric string. The currency # never lives inside ``value`` — it rides as a sibling # ``currency`` key that ``_serialize_feishu_form`` attaches # via ``_amount_wire_currency`` (see that docstring for the # live-Feishu shape matrix and the full bug history: the # nested-object shape is the one Feishu rejects with 1395006, # and the stringified-dict shape is the one that blanks the # finance edit node). if isinstance(answer, dict): amount_value = answer.get("amount") if amount_value is None: return "" return str(amount_value) return str(answer) if widget_type == "image": tokens = answer if isinstance(answer, list) else [answer] return [str(t) for t in tokens] if widget_type == "attachmentV2": tokens = answer if isinstance(answer, list) else [answer] return [str(t) for t in tokens] if isinstance(answer, (dict, list)): return answer return str(answer) @staticmethod def _map_feishu_status(status: str | None) -> str: """Normalise Feishu's instance status to the Helix enum. Feishu emits one of: PENDING / RUNNING / APPROVED / PASS / REJECTED / REFUSED / CANCELED / CANCELLED / DELETED / TIMEOUT / COMPLETED. We map TIMEOUT → ``rejected`` (2026-05-28): TIMEOUT means the Feishu flow ran out the per-template max-duration and Feishu auto-closed it without an approval decision. The pre-2026-05-28 default mapping returned ``pending`` for TIMEOUT, which left rows stuck in pending forever via the callback path AND made the reaper's replay a no-op (the replay's terminal-state guard read ``pending`` and skipped). Treating TIMEOUT as rejected matches the reaper's pre-fix intent (it directly mutated TIMEOUT rows to "rejected") and lets the per-domain handlers run their rejection-side cleanup (revoke pending seat, cancel scope request, …). """ normalized = (status or "PENDING").lower() # ``COMPLETED`` is a positive terminal (the instance finished its last # node) — map it to ``approved`` so every per-domain side-effect gate # (seat activation / scope grant / DNS write / asset apply), which all # check ``approval_status == "approved"``, actually fires. Previously # COMPLETED mapped to its own ``"completed"`` value, which no gate # matched: the row sealed terminal with NO side-effects and the child # request (seat/scope/DNS/asset) was orphaned in pending forever — and # the reaper couldn't rescue it (it re-sealed "completed" on replay), # directly contradicting its own documented intent that COMPLETED # triggers auto-execute. The analytics already bucket approved+completed # together, so this is consistent. if normalized in {"approved", "pass", "approved_status", "completed"}: return "approved" if normalized in {"rejected", "reject", "refused", "timeout"}: return "rejected" if normalized in {"canceled", "cancelled", "deleted"}: return "cancelled" return "pending" async def _load_users(self, user_ids: set[UUID]) -> dict[UUID, AuthUser]: if not user_ids: return {} result = await self.db.execute(select(AuthUser).where(AuthUser.id.in_(user_ids))) return {user.id: user for user in result.scalars().all()} async def _load_assets(self, asset_ids: set[UUID]) -> dict[UUID, AssetItem]: if not asset_ids: return {} result = await self.db.execute(select(AssetItem).where(AssetItem.id.in_(asset_ids))) return {asset.id: asset for asset in result.scalars().all()} async def _get_asset(self, asset_id: UUID) -> AssetItem: # Filter ``deleted_at IS NULL`` (added 2026-05-12): the asset # domain uses soft-delete (migration 115), and every other read # path applies this filter. Without it, approval submissions and # auto-execute callbacks can land on a soft-deleted asset and # silently mutate it (dispose-on-buyback, return, transfer) # while the asset list page treats it as gone. 404 is the # correct response: pending approvals against a soft-deleted # asset are operationally a bug — admins shouldn't soft-delete # assets with in-flight approvals, and if they do, the # approval must surface "asset not found" rather than zombie- # mutate. result = await self.db.execute( select(AssetItem).where( AssetItem.id == asset_id, AssetItem.deleted_at.is_(None), ) ) asset = result.scalar_one_or_none() if not asset: raise NotFoundError("Asset", str(asset_id)) return asset async def quote_offboarding_buyback( self, *, asset_id: UUID, current_user: AuthUser, target_currency: str | None = None, ) -> BuybackQuoteResponse: """Compute the offboarding-buyback amount the employee would owe. Reads the asset's ``purchase_cost`` / ``purchase_date`` / ``purchase_entity`` and applies the policy in :mod:`app.approval.buyback_price`. The default display currency follows the asset's ``purchase_entity`` (resolved through ``asset_legal_entities``) — Singapore-bought hardware quotes in SGD, Japan in JPY, everything else in CNY. Authz mirrors the buyback submission path: only the asset's owner / current borrower may quote it, so employees can't enumerate other people's hardware to scrape purchase prices. IT / asset admins bypass the check. """ asset = await self._get_asset(asset_id) if not can_manage_assets(current_user.role_names): actor_id = current_user.id if ( asset.assigned_to != actor_id and asset.borrowed_by != actor_id ): raise ForbiddenError( "You can only quote buyback for assets you own or " "currently borrow.", ) # Use the same legacy-safe resolution as the bank-information field. # An exact SQL lookup here used to miss harmless ``LTD`` / ``LTD.`` # differences and silently quote the wrong payment currency. entity = await self._resolve_payment_entity(asset) entity_currency = entity.currency if entity is not None else None default_currency = resolve_default_currency( entity_currency=entity_currency, asset_currency=asset.currency, ) # The amount paid to a registered legal entity must always carry that # entity's banking currency. This applies equally to employee, mobile, # and admin-on-behalf paths; ``currency`` only remains a legacy # fallback for assets that have no registered entity currency. if ( entity_currency and entity_currency.upper() in BUYBACK_SUPPORTED_CURRENCIES ): resolved_target = entity_currency.upper() else: resolved_target = (target_currency or default_currency).upper() try: quote = calculate_buyback( purchase_cost=asset.purchase_cost, purchase_currency=asset.currency, purchase_date=( asset.purchase_date.date() if isinstance(asset.purchase_date, datetime) else asset.purchase_date ), today=datetime.now(UTC).date(), target_currency=resolved_target, ) except ValueError as exc: # The calculator raises a stable ``CODE: message`` string; # surface that as a 422 so the frontend can show a precise # hint ("ask IT to backfill purchase_date") rather than # bubbling a 500. code_and_message = str(exc) code = code_and_message.split(":", 1)[0] raise AppException( code=code, message=code_and_message, status_code=422, ) from exc return BuybackQuoteResponse( asset_id=asset.id, amount=quote.amount, currency=quote.currency, default_currency=default_currency, supported_currencies=list(BUYBACK_SUPPORTED_CURRENCIES), mode=quote.mode, tier=quote.tier, months_used=quote.months_used, prices=quote.prices, ) async def _assert_asset_caller_authorized( self, *, asset: AssetItem, template_key: str, current_user: AuthUser, on_behalf_of_person_id: UUID | None = None, ) -> None: """Reject approval submission when ``current_user`` doesn't own / currently borrow ``asset``. Templates that mutate an asset (return / buyback / replacement / re-borrow against a held asset) must be self-service only — otherwise the auto-execute path on Feishu approval silently operates on someone else's hardware. ``computer_borrow`` against an asset already held by the caller (re-extending) is allowed; a fresh borrow targets stock and the public router is the authorized path (no asset_id supplied). **On-behalf buyback (代提回购)**: when ``on_behalf_of_person_id`` is set the caller is an IT/asset admin submitting for a leaver, so the self-ownership rule is replaced by two stricter checks: (1) the caller MUST be an asset manager, and (2) the named person MUST be the asset's current holder (the leaver who still shows as ``assigned_to`` / ``borrowed_by``). This keeps the requester-rewrite from being usable to dispose of an asset whose holder the admin invented. """ if on_behalf_of_person_id is not None: if not can_manage_assets(current_user.role_names): raise ForbiddenError( "Only asset administrators may submit a buyback on " "behalf of a leaver.", ) if ( asset.assigned_to != on_behalf_of_person_id and asset.borrowed_by != on_behalf_of_person_id ): raise ForbiddenError( "On-behalf buyback target must be the asset's current " "holder.", ) return actor_id = current_user.id # Templates whose auto-execute writes to ``asset`` directly — # all of these MUST require caller ownership / borrowership. asset_mutating = { "computer_borrow", "asset_return", "offboarding_buyback", "laptop_replacement", } if template_key not in asset_mutating: return # Allowed: caller is the original owner OR the current borrower. # ``computer_borrow`` against a stranger's claimed asset is the # canonical attack — explicitly forbid. if asset.assigned_to == actor_id or asset.borrowed_by == actor_id: return # ``computer_borrow`` may legitimately target an in_stock asset # the caller wants to claim — but the public path normally omits # asset_id (the system picks). If asset_id IS supplied for # in-stock claim the caller still must not be referencing # someone else's hardware. if ( template_key == "computer_borrow" and asset.assigned_to is None and asset.borrowed_by is None ): return raise ForbiddenError( "You can only submit asset approvals for assets you own or " "currently borrow.", ) async def _get_employee_request(self, request_id: UUID) -> AssetEmployeeRequest: result = await self.db.execute( select(AssetEmployeeRequest).where(AssetEmployeeRequest.id == request_id) ) request = result.scalar_one_or_none() if not request: raise NotFoundError("EmployeeRequest", str(request_id)) return request async def _get_approval_request_by_external_id(self, external_id: str) -> ApprovalRequest: result = await self.db.execute( select(ApprovalRequest).where(ApprovalRequest.external_id == external_id) ) request = result.scalar_one_or_none() if not request: raise NotFoundError("ApprovalRequest", external_id) return request @staticmethod def extract_buyback_amount( request_type: str, detail: dict[str, Any] | None, ) -> dict[str, str] | None: """Resolve the canonical buyback figure for a request row. Newest-first fallback chain — every shape the detail blob has carried across template rebuilds: 1. ``detail.buyback_amount`` — first-class key stamped at submit (2026-07-03) and backfilled by migration 379. 2. ``detail.feishu.form_snapshot`` — per-widget audit snapshot (2026-05-12+); the ``amount``-typed widget's answer. 3. ``detail.feishu.submitted_form`` — the serialized Feishu form payload (list of ``{id, type, value}``). Returns ``{"amount": str, "currency": str}`` or ``None``. Scoped to the two buyback-capable types so a stray amount widget on some future template can't masquerade as a buyback figure. """ if request_type not in ("offboarding_buyback", "laptop_replacement"): return None detail = detail or {} def _norm(value: Any) -> dict[str, str] | None: if isinstance(value, str): value = ApprovalService._parse_json_if_possible(value) if not isinstance(value, dict): return None amount = value.get("amount") if amount in (None, ""): return None return { "amount": str(amount), "currency": str(value.get("currency") or ""), } direct = _norm(detail.get("buyback_amount")) if direct: return direct feishu = detail.get("feishu") or {} snapshot = feishu.get("form_snapshot") if isinstance(snapshot, list): for widget in snapshot: if isinstance(widget, dict) and widget.get("type") == "amount": found = _norm(widget.get("answer")) if found: return found submitted = feishu.get("submitted_form") if isinstance(submitted, str): submitted = ApprovalService._parse_json_if_possible(submitted) if isinstance(submitted, list): for widget in submitted: if isinstance(widget, dict) and widget.get("type") == "amount": found = _norm(widget.get("value")) if found: return found return None def _to_request_response( self, request: AssetEmployeeRequest, requesters: dict[UUID, AuthUser], assets: dict[UUID, AssetItem], ) -> ApprovalBackedEmployeeRequestResponse: requester = requesters.get(request.requester_id) asset = assets.get(request.asset_id) if request.asset_id else None feishu_detail = dict((request.detail or {}).get("feishu") or {}) return ApprovalBackedEmployeeRequestResponse( id=request.id, request_number=request.request_number, request_type=request.request_type, status=request.status, requester_id=request.requester_id, requester_name=requester.display_name if requester else None, requester_email=requester.work_email if requester else None, requester_employee_no=requester.employee_no if requester else None, requester_employment_status=requester.employment_status if requester else None, asset_id=request.asset_id, asset_tag=asset.asset_tag if asset else None, asset_name=asset.asset_name if asset else None, reason=request.reason, notes=request.notes, detail=request.detail, buyback_amount=self.extract_buyback_amount( request.request_type, request.detail, ), approval_code=feishu_detail.get("approval_code"), approval_name=feishu_detail.get("approval_name"), approval_instance_code=feishu_detail.get("instance_code"), approval_status=self._map_feishu_status(feishu_detail.get("instance_status")), created_at=request.created_at, processed_at=request.processed_at, processed_by=request.processed_by, ) # ── Admin all-approvals view ──────────────────────────────── # # Powers the IT-admin "审批中心" page: a unified table of EVERY # employee_requests row (asset hardware, software seat, api-key, # google_*) with search / filter / pagination, plus a detail # endpoint that pulls the live Feishu approval timeline so admins # can see who's pending, who approved when, and any approver # comments — without leaving Helix. # # Distinct from ``list_asset_approval_requests`` because: # * No domain pin (the asset/software menu pages narrow to their # own type sets; this view never narrows by default). # * Keyword search joins ``identity_persons`` so admins can find # a request by requester name / work_email, not just by # request_number / reason. # * Caller is guaranteed to be ``global_admin`` by the router # dependency, so we don't re-check role here — but we DO # document the precondition. async def admin_analytics_summary( self, *, window_days: int = 30, ) -> dict[str, Any]: """Aggregate signal for the ``/admin/approvals`` 分析 tab. Source table is the unified ``employee_requests`` (admin approval surface). All aggregates pivot on ``created_at`` so the window slices match what the list tab shows. Turnaround latency derives from ``processed_at - created_at`` for rows that reached a terminal state inside the window. Defensive scope: read-only; admin gate enforced at the router. """ from sqlalchemy import case as _case from app.core.timeseries import day_axis from app.identity.person_models import IdentityPerson now = datetime.now(UTC) # Round-62 P1 fix (off-by-one analytics drops today): shared # day-axis helper anchors the SQL window and the bucket loop to # the same N day-aligned interval ending on today. Previously # the bucket loop computed [today-N … today-1] from window_start # while the SQL filter included today, so today's approvals # were counted in the headline KPI but absent from the chart. axis = day_axis(window_days=window_days, now=now) window_start = axis.window_start today_start = now.replace(hour=0, minute=0, second=0, microsecond=0) # ── Lifetime + window counters ──────────────────────────── total = ( await self.db.execute( select(func.count()).select_from(AssetEmployeeRequest) ) ).scalar() or 0 total_window = ( await self.db.execute( select(func.count()).select_from(AssetEmployeeRequest) .where(AssetEmployeeRequest.created_at >= window_start) ) ).scalar() or 0 total_today = ( await self.db.execute( select(func.count()).select_from(AssetEmployeeRequest) .where(AssetEmployeeRequest.created_at >= today_start) ) ).scalar() or 0 pending_open = ( await self.db.execute( select(func.count()).select_from(AssetEmployeeRequest) .where(AssetEmployeeRequest.status == "pending") ) ).scalar() or 0 # ── Status breakdown (window) ───────────────────────────── status_rows = ( await self.db.execute( select( AssetEmployeeRequest.status, func.count(), ) .where(AssetEmployeeRequest.created_at >= window_start) .group_by(AssetEmployeeRequest.status) ) ).all() by_status = {row[0]: int(row[1]) for row in status_rows} approved_n = by_status.get("approved", 0) + by_status.get("completed", 0) rejected_n = by_status.get("rejected", 0) decided_n = approved_n + rejected_n # ``None`` when no decisions in window so the KPI renders ``—`` # (neutral) instead of a red 0% which falsely advertises # "everything got rejected" while the truth is "no decisions yet". approval_rate: float | None = ( round(approved_n * 1000 / decided_n) / 10 if decided_n else None ) # ── Request-type breakdown (window) — overlap with approval rate type_rows = ( await self.db.execute( select( AssetEmployeeRequest.request_type, AssetEmployeeRequest.status, func.count(), ) .where(AssetEmployeeRequest.created_at >= window_start) .group_by( AssetEmployeeRequest.request_type, AssetEmployeeRequest.status, ) ) ).all() by_request_type: dict[str, dict[str, int]] = {} for rt, status_v, n in type_rows: bucket = by_request_type.setdefault( rt or "unknown", {"total": 0, "pending": 0, "approved": 0, "rejected": 0, "completed": 0, "cancelled": 0}, ) bucket["total"] += int(n) if status_v in bucket: bucket[status_v] = int(n) # Convert to a sorted list of dicts for the UI bar chart. request_type_breakdown = [] for rt, b in sorted(by_request_type.items(), key=lambda x: -x[1]["total"]): decided = (b["approved"] + b["completed"]) + b["rejected"] # ``None`` (not 0.0) for undecided types so the frontend # detail-table renders ``—`` not a misleading red ``0%`` # — matches the top-level ``totals.approval_rate`` contract. ar: float | None = ( round((b["approved"] + b["completed"]) * 1000 / decided) / 10 if decided else None ) request_type_breakdown.append({ "request_type": rt, "total": b["total"], "pending": b["pending"], "approved": b["approved"], "rejected": b["rejected"], "completed": b["completed"], "cancelled": b["cancelled"], "approval_rate": ar, }) # ── Daily volume (created vs processed) in the window ──── daily_created_rows = ( await self.db.execute( select( func.date(AssetEmployeeRequest.created_at).label("d"), func.count(), ) .where(AssetEmployeeRequest.created_at >= window_start) .group_by("d").order_by("d") ) ).all() daily_processed_rows = ( await self.db.execute( select( func.date(AssetEmployeeRequest.processed_at).label("d"), func.count(), ) .where(AssetEmployeeRequest.processed_at >= window_start) .group_by("d").order_by("d") ) ).all() created_map = { (r[0].isoformat() if hasattr(r[0], "isoformat") else str(r[0])): int(r[1]) for r in daily_created_rows } processed_map = { (r[0].isoformat() if hasattr(r[0], "isoformat") else str(r[0])): int(r[1]) for r in daily_processed_rows } daily_series: list[dict] = [] for day in axis.days: key = day.isoformat() daily_series.append({ "date": key, "created": created_map.get(key, 0), "processed": processed_map.get(key, 0), }) # ── Turnaround (hours between created → processed) ──────── # Use TIMESTAMPDIFF for portability; we only consider rows whose # terminal state is approved/completed/rejected — pending / # cancelled / abandoned distort averages. # # The first argument to TIMESTAMPDIFF is a SQL **keyword** # (MINUTE / HOUR / DAY), NOT a string literal. Passing # ``"MINUTE"`` lets SQLAlchemy bind it as a parameter # (``TIMESTAMPDIFF(%s, ...)``) which TiDB/MySQL reject with # ``1064 You have an error in your SQL syntax``. Wrap it in # ``literal_column`` so it renders as a bare token. minute_unit = literal_column("MINUTE") decided_status = ("approved", "completed", "rejected") turn_rows = ( await self.db.execute( select( AssetEmployeeRequest.request_type, func.avg( func.timestampdiff( minute_unit, AssetEmployeeRequest.created_at, AssetEmployeeRequest.processed_at, ) ).label("avg_min"), func.max( func.timestampdiff( minute_unit, AssetEmployeeRequest.created_at, AssetEmployeeRequest.processed_at, ) ).label("max_min"), func.count().label("n"), ) .where( AssetEmployeeRequest.processed_at.isnot(None), AssetEmployeeRequest.status.in_(decided_status), AssetEmployeeRequest.processed_at >= window_start, ) .group_by(AssetEmployeeRequest.request_type) ) ).all() turnaround_by_type = [ { "request_type": row[0] or "unknown", "avg_hours": ( round(float(row[1]) / 6) / 10 if row[1] is not None else None ), "max_hours": ( round(float(row[2]) / 6) / 10 if row[2] is not None else None ), "decided_count": int(row[3]), } for row in turn_rows ] # Overall turnaround. overall_turn = ( await self.db.execute( select( func.avg( func.timestampdiff( minute_unit, AssetEmployeeRequest.created_at, AssetEmployeeRequest.processed_at, ) ), func.count(), ) .where( AssetEmployeeRequest.processed_at.isnot(None), AssetEmployeeRequest.status.in_(decided_status), AssetEmployeeRequest.processed_at >= window_start, ) ) ).first() if overall_turn and overall_turn[0] is not None: avg_overall_hours = round(float(overall_turn[0]) / 6) / 10 decided_overall = int(overall_turn[1] or 0) else: avg_overall_hours = None decided_overall = 0 # ── Top 10 requesters (window) ──────────────────────────── requester_rows = ( await self.db.execute( select( AssetEmployeeRequest.requester_id, func.count().label("c"), func.sum( _case((AssetEmployeeRequest.status == "rejected", 1), else_=0) ).label("rejected_c"), ) .where(AssetEmployeeRequest.created_at >= window_start) .group_by(AssetEmployeeRequest.requester_id) .order_by(func.count().desc()) .limit(10) ) ).all() requester_ids = [row[0] for row in requester_rows if row[0]] person_map: dict[str, tuple[str | None, str | None, str | None, str | None]] = {} if requester_ids: res = await self.db.execute( select( IdentityPerson.id, IdentityPerson.display_name, IdentityPerson.work_email, IdentityPerson.employee_no, IdentityPerson.employment_status, ).where(IdentityPerson.id.in_(requester_ids)) ) person_map = { str(r.id): (r.display_name, r.work_email, r.employee_no, r.employment_status) for r in res } top_requesters = [] for rid, c, rc in requester_rows: display, email, emp_no, emp_status = person_map.get( str(rid), (None, None, None, None), ) top_requesters.append({ "requester_id": str(rid) if rid else None, "display_name": display, "work_email": email, "employee_no": emp_no, "employment_status": emp_status, "count": int(c), "rejected_count": int(rc or 0), }) return { "window_days": window_days, "checked_at": now.isoformat(), "totals": { "total": int(total), "total_window": int(total_window), "total_today": int(total_today), "pending_open": int(pending_open), "approved_window": approved_n, "rejected_window": rejected_n, "approval_rate": approval_rate, "avg_turnaround_hours": avg_overall_hours, # ``decided_window`` MUST be the same cohort the # ``approval_rate`` was computed over (rows created in # window) — not ``decided_overall`` (rows PROCESSED in # window, which is a different cohort and skews the # KPI's "X% of N decisions" gloss). The turnaround # average's denominator is exposed separately as # ``decided_for_turnaround``. "decided_window": decided_n, "decided_for_turnaround": decided_overall, }, "by_status": by_status, "request_type_breakdown": request_type_breakdown, "daily_series": daily_series, "turnaround_by_type": turnaround_by_type, "top_requesters": top_requesters, } async def list_admin_approval_requests( self, *, page: int = 1, page_size: int = 20, status: str | None = None, request_type: list[str] | None = None, requester_id: UUID | None = None, keyword: str | None = None, start_date: str | None = None, end_date: str | None = None, ) -> tuple[list[AdminApprovalListItem], int]: """Admin-scope list of every employee_requests row. Caller MUST be ``global_admin`` — the router enforces that via ``Depends(require_admin())``. No role check here. """ from sqlalchemy.orm import selectinload # Build conditions on AssetEmployeeRequest first; keyword # search adds a join-bound subcondition that pulls in the # identity_persons table. conditions: list[Any] = [] if status: conditions.append(AssetEmployeeRequest.status == status) if request_type: conditions.append(AssetEmployeeRequest.request_type.in_(request_type)) if requester_id is not None: conditions.append(AssetEmployeeRequest.requester_id == requester_id) if start_date: try: start_dt = datetime.combine(date.fromisoformat(start_date), datetime.min.time()) except ValueError as exc: raise ValidationError( f"start_date must be ISO format (YYYY-MM-DD), got {start_date!r}", ) from exc conditions.append(AssetEmployeeRequest.created_at >= start_dt) if end_date: try: end_dt = datetime.combine(date.fromisoformat(end_date), datetime.max.time()) except ValueError as exc: raise ValidationError( f"end_date must be ISO format (YYYY-MM-DD), got {end_date!r}", ) from exc conditions.append(AssetEmployeeRequest.created_at <= end_dt) query = select(AssetEmployeeRequest) count_query = select(func.count()).select_from(AssetEmployeeRequest) if keyword: pattern = f"%{keyword}%" # Subquery against identity_persons so the requester-name / # email match doesn't need an outer join (which would # change the row count when a request's requester row is # missing for any reason). requester_match = select(AuthUser.id).where( or_( AuthUser.display_name.ilike(pattern), AuthUser.work_email.ilike(pattern), ) ) keyword_clause = or_( AssetEmployeeRequest.request_number.ilike(pattern), AssetEmployeeRequest.reason.ilike(pattern), AssetEmployeeRequest.requester_id.in_(requester_match), ) conditions.append(keyword_clause) if conditions: combined = and_(*conditions) query = query.where(combined) count_query = count_query.where(combined) total = (await self.db.execute(count_query)).scalar() or 0 result = await self.db.execute( query.order_by(AssetEmployeeRequest.created_at.desc()) .offset((page - 1) * page_size) .limit(page_size) ) rows = list(result.scalars().all()) requester_ids = {row.requester_id for row in rows} asset_ids = {row.asset_id for row in rows if row.asset_id} # Eager-load org_memberships so the ``primary_department`` # property doesn't trigger a per-row lazy load (which would # explode under async — IdentityPerson.org_memberships is a # relationship). requesters: dict[UUID, AuthUser] = {} if requester_ids: users_result = await self.db.execute( select(AuthUser) .where(AuthUser.id.in_(requester_ids)) .options(selectinload(AuthUser.org_memberships)) ) requesters = {u.id: u for u in users_result.scalars().all()} assets = await self._load_assets(asset_ids) items: list[AdminApprovalListItem] = [] for row in rows: requester = requesters.get(row.requester_id) asset = assets.get(row.asset_id) if row.asset_id else None feishu_detail = dict((row.detail or {}).get("feishu") or {}) # ``primary_department`` reads the eagerly-loaded # org_memberships collection — safe in async context. department = requester.primary_department if requester else None items.append( AdminApprovalListItem( id=row.id, request_number=row.request_number, request_type=row.request_type, status=row.status, requester_id=row.requester_id, requester_name=requester.display_name if requester else None, requester_email=requester.work_email if requester else None, requester_employee_no=requester.employee_no if requester else None, requester_employment_status=( requester.employment_status if requester else None ), requester_department=department, asset_id=row.asset_id, asset_tag=asset.asset_tag if asset else None, asset_name=asset.asset_name if asset else None, reason=row.reason, approval_name=feishu_detail.get("approval_name"), approval_instance_code=feishu_detail.get("instance_code"), created_at=row.created_at, processed_at=row.processed_at, ) ) return items, total async def get_admin_approval_request_detail( self, *, request_id: UUID, refresh_timeline: bool = True, owner_user_id: UUID | None = None, actor_user_id: UUID | None = None, allow_reconcile: bool = False, ) -> AdminApprovalDetailResponse: """Return one employee_request enriched with the live Feishu timeline. Caller is typically ``global_admin``. The same enrichment is reused for the owner-scoped ``GET /my/requests/{id}`` endpoint by passing ``owner_user_id=current_user.id`` — the method then refuses any row whose ``requester_id`` doesn't match, returning the standard ``NotFoundError`` (no info leak about whether the id exists or not). The timeline is fetched live from Feishu on every call (one HTTP round-trip per click is acceptable for a one-row drawer view). On Feishu failure we fall back to whatever was cached in ``detail.feishu.latest_instance_detail`` and surface the error in ``timeline_error`` so the UI can show "stale" warnings — we never throw, because the rest of the request data is still useful. ``allow_reconcile`` (added 2026-05-28): when True AND the live Feishu fetch succeeded AND Feishu reports a terminal state AND the Helix row is still ``pending``, the method replays the missed Feishu callback via ``replay_callback_from_feishu`` so the row's status + auto- execute side-effects catch up on the same click that refreshed the timeline. Callers MUST gate this — only pass True for write-capable roles (global_admin / asset_admin) or the row owner; auditor / management see the refreshed timeline but no state mutation. ``actor_user_id`` is recorded in the audit row so a reconcile is traceable to the user who clicked refresh. """ from sqlalchemy.orm import selectinload result = await self.db.execute( select(AssetEmployeeRequest) .where(AssetEmployeeRequest.id == request_id) ) row = result.scalar_one_or_none() if row is None: raise NotFoundError("EmployeeRequest", str(request_id)) # Owner-scope gate for the employee endpoint. Surface the same # NotFound error a stranger-id hit would surface — we don't # want a 403 to confirm "yes that id exists but isn't yours." if owner_user_id is not None and row.requester_id != owner_user_id: raise NotFoundError("EmployeeRequest", str(request_id)) # Load requester with org_memberships so we can render the # primary department in the drawer. requester: AuthUser | None = None users_result = await self.db.execute( select(AuthUser) .where(AuthUser.id == row.requester_id) .options(selectinload(AuthUser.org_memberships)) ) requester = users_result.scalar_one_or_none() asset: AssetItem | None = None if row.asset_id: asset_result = await self.db.execute( select(AssetItem).where(AssetItem.id == row.asset_id) ) asset = asset_result.scalar_one_or_none() feishu_detail = dict((row.detail or {}).get("feishu") or {}) instance_code = feishu_detail.get("instance_code") timeline: list[AdminApprovalTimelineEntry] = [] timeline_fetched_at: datetime | None = None timeline_error: str | None = None instance_status_raw: str | None = feishu_detail.get("instance_status") cached_instance_detail = ( feishu_detail.get("latest_instance_detail") if isinstance(feishu_detail.get("latest_instance_detail"), dict) else None ) # The snapshot we'll feed both ``_build_timeline_from_instance`` # and ``_build_progress_from_instance``. Refreshed below if the # live Feishu fetch succeeds; falls back to the cached blob on # failure so the drawer still shows whatever was last seen. progress_source: dict[str, Any] | None = cached_instance_detail # Tracks whether ``progress_source`` came from a successful live # fetch (vs the cached snapshot we should NOT re-persist). An # explicit flag is sturdier than an ``is not cached_instance_detail`` # identity check — a future refactor that mutated ``cached_instance_detail`` # in place during enrichment would silently break identity-based # gating, leaving the persist branch unreachable. live_fetch_succeeded = False if instance_code and refresh_timeline: try: client = await self._get_client() live = await client.get_instance_detail(instance_code) timeline_fetched_at = datetime.now(UTC) # If Feishu returned a malformed (non-dict) payload, # treat it like a fetch failure: surface a timeline # error, fall back to the cached snapshot for BOTH # timeline and progress (otherwise ``timeline=[]`` # would render alongside a stale progress hydrated # from cache — the exact asymmetry round-1 fix #2 # patched). DO NOT persist the bad payload, since # writing ``latest_instance_detail = None`` would # corrupt the cached blob future callers rely on. if not isinstance(live, dict): timeline_error = ( f"Feishu returned a non-dict instance payload " f"({type(live).__name__})" ) else: new_status = live.get("status") if new_status: instance_status_raw = new_status progress_source = live live_fetch_succeeded = True except Exception as fetch_err: # noqa: BLE001 timeline_error = str(fetch_err) or type(fetch_err).__name__ logger.warning( "admin approval detail: live Feishu fetch failed for instance %s (%s); falling back to cached snapshot", instance_code, timeline_error, ) # ── Enrich task_list approver names ────────────────────────── # Feishu's ``/approval/v4/instances/{code}`` response carries # ``open_id`` + ``user_id`` per task but leaves ``user_name`` # null on every tenant we've observed in prod. Resolve via # ``/contact/v3/users/{id}`` once before building the timeline # and progress sub-objects so downstream consumers (and the # persisted snapshot, on the live-success path) see real names # rather than short ids. Mutates ``progress_source`` in place. # Fully best-effort: any lookup failure leaves the entry's # ``user_name`` as Feishu provided it (i.e. null), and the # legacy uid-fallback in ``_build_progress_from_instance`` # / the frontend renderer continues to render the raw id. if progress_source is not None: try: await self._enrich_task_list_user_names(progress_source) except Exception as enrich_err: # noqa: BLE001 logger.warning( "admin approval detail: name-enrichment failed for instance %s (%s); proceeding with raw ids", instance_code, enrich_err, ) # ── Persist refreshed snapshot back (if live succeeded) ────── # Done AFTER enrichment so the persisted snapshot already has # resolved names — future cached-only reads serve real names # without a Feishu round-trip. Guarded by the explicit # ``live_fetch_succeeded`` flag set inside the live branch # above; a malformed live OR fetch failure leaves the flag # False and skips persist, which is correct (writing back the # CACHED snapshot would be a no-op write at best and risks # status corruption at worst). if live_fetch_succeeded and progress_source is not None: async with best_effort( self.db, label="approval: persist refreshed Feishu progress snapshot", log=logger, ) as be_7509_step: detail_blob = dict(row.detail or {}) fb = dict(detail_blob.get("feishu") or {}) fb["latest_instance_detail"] = progress_source # Preserve the previously-known status when Feishu's # response omits ``status`` (rare, but a defensive # guard so a malformed response can't overwrite a # known-good value with None). fb["instance_status"] = ( progress_source.get("status") or fb.get("instance_status") ) detail_blob["feishu"] = fb row.detail = detail_blob await self.db.flush() if not be_7509_step.ok: logger.warning( "admin approval detail: persisting refreshed Feishu detail failed (request=%s): %s", row.id, be_7509_step.reason(), ) # ── Reconcile drift: webhook missed + Feishu now terminal ──── # Feishu webhooks are at-most-once in practice (drops, signature # mismatches, mid-flight 5xx all leave Helix with status='pending' # while Feishu's side has settled). If the live fetch we just did # shows a terminal state but our row is still pending, replay the # missed callback inline so the row's status + auto-execute # side-effects catch up on the SAME click that refreshed the # timeline. Idempotent — replay short-circuits if a concurrent # webhook lands between fetch and replay. if ( allow_reconcile and live_fetch_succeeded and instance_code and row.status == "pending" and self._map_feishu_status(instance_status_raw) != "pending" ): 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(self.db, label="approval replay item"): # SAVEPOINT-isolate the replay (mirrors the reaper's # per-row ``begin_nested``): the dispatcher mutates the # employee_request (status/processed_at) BEFORE running # domain side-effects, and the router commits this # session after we return. Without the savepoint, a # side-effect crash caught below left those mutations # dirty-but-committed — the row showed approved with no # auto-execute, no audit trail, and no notification # (observed in prod up to 2026-07-03 via the form-string # AttributeError). Rolling back to the savepoint keeps # the row honestly ``pending`` so a later replay can # re-run the whole callback atomically. async with self.db.begin_nested(): replay_result = await self.replay_callback_from_feishu( instance_code=instance_code, actor_user_id=actor_user_id, prefetched_instance_detail=progress_source, trace_id_prefix="approval.refresh.reconcile", ) if replay_result.get("replayed"): # ``handle_feishu_callback`` mutated ``AssetEmployeeRequest`` # (the same row we already hold) in-place at line 1847 # via ``_get_employee_request``. Refresh so the response # we build below carries the post-replay status, not # the stale pending we loaded at the top of the # method. await self.db.refresh(row) logger.info( "admin approval detail: reconciled missed callback " "for instance %s (%s → %s)", instance_code, replay_result.get("previous_status"), replay_result.get("new_status"), ) except Exception as reconcile_err: # noqa: BLE001 # A domain-handler failure mid-replay (e.g. asset_return # auto-execute crashes because the asset is in an # unexpected state) should NOT 500 the refresh endpoint. # Surface as a timeline_error so the admin sees a yellow # banner explaining the reconcile attempt failed, and # log for ops triage. The row stays pending; admin can # retry or escalate. logger.exception( "admin approval detail: reconcile replay failed for " "instance %s; row left in pre-replay state", instance_code, ) # The savepoint rollback expired every attribute the # replay mutated; on an AsyncSession a later plain # attribute access on an expired object raises # MissingGreenlet. Re-load explicitly so the response # builder below reads the (pre-replay) committed state. async with best_effort( self.db, label="service best-effort step", log=logger, ) as be_7598_step: await self.db.refresh(row) if not be_7598_step.ok: # ``row.id`` itself may be expired — identify by the # instance_code we already hold as a plain str. logger.warning( "admin approval detail: post-rollback refresh " "failed for instance %s", instance_code, ) if not timeline_error: timeline_error = ( f"Reconcile replay failed: {reconcile_err}" ) # ── Build timeline + progress from the enriched snapshot ───── # Two non-live branches collapse here too: # (a) ``instance_code`` exists but caller passed # ``refresh_timeline=False`` (read-only cached view) — # previously fell through without building the # timeline, so the response had ``timeline=[]`` while # ``progress`` was hydrated from the same cache. That # asymmetry produced the user-visible "审批进度 shows # step 2/3 but 审批流程 says no actions yet" bug. # (b) ``instance_code`` missing but cache somehow exists # (legacy data shape) — fall through and use it too. if progress_source is not None: timeline = self._build_timeline_from_instance(progress_source) department = requester.primary_department if requester else None # Progress (current node / total nodes) — None when no # task_list has been observed yet (instance not yet created # in Feishu, or legacy fake-approval rows from before the # Feishu integration shipped); the frontend hides the Steps # indicator in that case. # # ``static_stages`` (added 2026-05-21) is the cached template # node chain from ``_parse_template_stages``. When supplied, # the progress builder overlays un-traversed future nodes on # the live task_list-derived progress so the detail modal # shows the FULL flow instead of only nodes the approval has # already touched. Asset templates always have stages; other # flow types (api_key, google_*, ...) silently fall through # with ``static_stages=None`` and keep the legacy # task_list-only progress. static_stages = await self._lookup_cached_template_stages( feishu_detail.get("approval_code"), ) # Always call the builder when EITHER live data OR static # stages exist — the builder has fallback branches for both # (instance_detail=None + static_stages → all-WAIT progress # so the requester still sees the full chain even before # Feishu pushes its first task to us). progress = ( self._build_progress_from_instance( progress_source, static_stages=static_stages, ) if (progress_source is not None or static_stages) else None ) return AdminApprovalDetailResponse( id=row.id, request_number=row.request_number, request_type=row.request_type, status=row.status, requester_id=row.requester_id, requester_name=requester.display_name if requester else None, requester_email=requester.work_email if requester else None, requester_employee_no=requester.employee_no if requester else None, requester_employment_status=( requester.employment_status if requester else None ), requester_department=department, asset_id=row.asset_id, asset_tag=asset.asset_tag if asset else None, asset_name=asset.asset_name if asset else None, reason=row.reason, notes=row.notes, detail=row.detail, approval_code=feishu_detail.get("approval_code"), approval_name=feishu_detail.get("approval_name"), approval_instance_code=instance_code, approval_status=self._map_feishu_status(instance_status_raw), instance_status_raw=instance_status_raw, timeline=timeline, timeline_fetched_at=timeline_fetched_at, timeline_error=timeline_error, progress=progress, created_at=row.created_at, processed_at=row.processed_at, ) async def build_live_feishu_flow( self, instance_code: str | None, ) -> tuple[ApprovalProgress | None, list[AdminApprovalTimelineEntry], str | None]: """Live-fetch ONE Feishu approval instance → (progress, timeline, error). The row-agnostic core of :meth:`get_admin_approval_request_detail`, carved out so the unified "我的申请" detail (:func:`app.domains.me_universe.request_detail.build_my_request_detail`) can render the same 审批流 + 当前审批人 view for the non- ``employee_request`` sources (DNS / cloud-scope / VPN / AI-tool / announcement / domain / SSO) that carry a Feishu instance but were previously stuck with a status-only synthesised timeline. Best-effort and self-contained: one HTTP round-trip, names enriched via ``/contact/v3``, then the shared static builders produce the progress (with ``current_approvers`` populated for the pending node) and the merged task/comment timeline. Any failure returns ``(None, [], error_string)`` so the caller falls back to its synthesised timeline without ever throwing. Does NOT persist a snapshot (no owning row to cache onto) and does NOT reconcile missed callbacks — those stay in the asset-row path that owns them. """ if not instance_code: return None, [], None try: client = await self._get_client() live = await client.get_instance_detail(instance_code) except Exception as fetch_err: # noqa: BLE001 err = str(fetch_err) or type(fetch_err).__name__ logger.warning( "live Feishu flow: fetch failed for instance %s (%s)", instance_code, err, ) return None, [], err if not isinstance(live, dict): return None, [], ( f"Feishu returned a non-dict instance payload ({type(live).__name__})" ) try: await self._enrich_task_list_user_names(live) except Exception as enrich_err: # noqa: BLE001 logger.warning( "live Feishu flow: name-enrichment failed for instance %s (%s); " "proceeding with raw ids", instance_code, enrich_err, ) timeline = self._build_timeline_from_instance(live) progress = self._build_progress_from_instance(live) return progress, timeline, None @staticmethod def _build_timeline_from_instance( instance_detail: dict[str, Any], ) -> list[AdminApprovalTimelineEntry]: """Merge Feishu ``task_list`` + ``comment_list`` into a sorted timeline. - Tasks (审批人动作) carry a status (PENDING/APPROVED/REJECTED/...) and start/end times. ``end_time`` is the moment the approver acted; if absent, we fall back to ``start_time`` (still pending tasks are sorted by when they were assigned). - Comments (审批人留言) carry a ``create_time`` and free text. - Both lists may have epoch-millis or ISO-8601 timestamps; we accept both. Unparseable entries land at the end (None sorts last via the key function below). """ if not isinstance(instance_detail, dict): return [] task_list = instance_detail.get("task_list") comment_list = instance_detail.get("comment_list") # Defence-in-depth bound on Feishu-supplied lists (same # rationale as ``_build_progress_from_instance``'s 500-task # cap). Comments are bounded slightly higher because a busy # approval thread can legitimately accrue many free-text # notes; tasks rarely exceed ~50 in real flows. if isinstance(task_list, list): task_list = task_list[:500] if isinstance(comment_list, list): comment_list = comment_list[:500] entries: list[AdminApprovalTimelineEntry] = [] if isinstance(task_list, list): for task in task_list: if not isinstance(task, dict): continue ts = ApprovalService._coerce_feishu_ts( task.get("end_time") or task.get("start_time") ) node_id_raw = ( task.get("node_id") or task.get("approval_node_id") or task.get("custom_node_id") ) entries.append( AdminApprovalTimelineEntry( kind="task", status=task.get("status"), node_id=str(node_id_raw) if node_id_raw else None, node_name=( task.get("node_name") or task.get("approval_node_name") ), user_id=str(task.get("user_id")) if task.get("user_id") else None, open_id=str(task.get("open_id")) if task.get("open_id") else None, user_name=task.get("user_name") or task.get("name"), comment=None, occurred_at=ts, ) ) if isinstance(comment_list, list): for comment in comment_list: if not isinstance(comment, dict): continue ts = ApprovalService._coerce_feishu_ts(comment.get("create_time")) entries.append( AdminApprovalTimelineEntry( kind="comment", status=None, user_id=str(comment.get("user_id")) if comment.get("user_id") else None, open_id=str(comment.get("open_id")) if comment.get("open_id") else None, user_name=comment.get("user_name") or comment.get("name"), comment=comment.get("comment") or comment.get("text"), occurred_at=ts, ) ) entries.sort(key=lambda e: (e.occurred_at is None, e.occurred_at or datetime.min.replace(tzinfo=UTC))) return entries async def _enrich_task_list_user_names( self, instance_detail: dict[str, Any], ) -> None: """Fill in ``task_list[*].user_name`` + ``comment_list[*].user_name`` via Feishu. Feishu's ``/approval/v4/instances/{code}`` response carries both ``open_id`` and ``user_id`` per task / comment entry but leaves ``user_name`` null on every tenant we've observed in prod (verified 2026-05-15 across multiple ``approval_code`` values — pending AND approved instances alike). Without this enrichment, ``_build_progress_from_instance`` falls back to the raw id and the admin modal renders short ids like ``d8d2646c``; the frontend ``ApprovalFlowTimeline`` dedup logic then keys on ``user_name`` and double-renders each row. Mutates ``instance_detail['task_list']`` and ``instance_detail['comment_list']`` in place. Only fills entries whose ``user_name`` / ``name`` is currently empty — existing names (a tenant where Feishu DOES populate them) are left untouched. Best-effort: any lookup miss (deleted user, transient Feishu error) silently leaves that one entry's ``user_name`` as Feishu returned it. The raw Feishu ``timeline`` array (chronological event log) is intentionally NOT walked — no downstream builder reads it, so enrichment there would waste Feishu API quota. Resolution priority is ``open_id`` first, then ``user_id``. The former is per-app-stable; the latter is short and tenant- configurable and occasionally collides across deactivated accounts. """ task_list = instance_detail.get("task_list") comment_list = instance_detail.get("comment_list") if not isinstance(task_list, list) and not isinstance(comment_list, list): return # Collect ids from BOTH ``task_list`` (approver actions — # consumed by ``_build_progress_from_instance`` for node # approver lists AND by ``_build_timeline_from_instance`` for # the task-kind entries) AND ``comment_list`` (approver留言 — # consumed by ``_build_timeline_from_instance`` for the # comment-kind entries). The Feishu raw ``timeline`` array is # a separate chronological event log that no downstream # builder reads, so enriching it would be wasted work. # Defence-in-depth list-size caps mirror the ones in # ``_build_progress_from_instance`` and # ``_build_timeline_from_instance``. open_ids: set[str] = set() user_ids: set[str] = set() def _collect(entries: list[Any]) -> None: for e in entries[:500]: if not isinstance(e, dict): continue name = e.get("user_name") or e.get("name") if isinstance(name, str) and name.strip(): continue oid = e.get("open_id") uid = e.get("user_id") if isinstance(oid, str) and oid.strip(): open_ids.add(oid.strip()) if isinstance(uid, str) and uid.strip(): user_ids.add(uid.strip()) if isinstance(task_list, list): _collect(task_list) if isinstance(comment_list, list): _collect(comment_list) if not open_ids and not user_ids: return client = await self._get_client() names = await client.batch_get_user_names( open_ids=sorted(open_ids), user_ids=sorted(user_ids), ) if not names: return def _fill(entries: list[Any]) -> None: for e in entries[:500]: if not isinstance(e, dict): continue name = e.get("user_name") or e.get("name") if isinstance(name, str) and name.strip(): continue oid = e.get("open_id") uid = e.get("user_id") # open_id first (more stable + globally unique within # an app). Falls back to user_id if open_id didn't # resolve. Writing ``user_name`` is the canonical # downstream-consumer field; we deliberately do NOT # write ``name`` (an alternate spelling some Feishu # endpoints use) — the consumers all check user_name # first, and overwriting both wouldn't change behaviour. if isinstance(oid, str) and oid.strip() in names: e["user_name"] = names[oid.strip()] elif isinstance(uid, str) and uid.strip() in names: e["user_name"] = names[uid.strip()] if isinstance(task_list, list): _fill(task_list) if isinstance(comment_list, list): _fill(comment_list) async def _lookup_cached_template_stages( self, approval_code: str | None, ) -> list[ApprovalTemplateStage]: """Find cached template stages for ``approval_code``. Reads the same ``platform_settings`` cache the create-request UI hydrates from, so this stays cheap (one DB row, no Feishu round-trip) on every detail-modal open. Returns ``[]`` when the approval_code doesn't belong to any asset template (api_key / google_* / vpn / ... flows) — the caller treats empty as "no overlay" and the progress builder falls back to the legacy task_list-only path. """ if not approval_code: return [] # Try fresh cache first; fall back to the TTL-ignoring read # so a Feishu outage that left the cache stale still surfaces # stages (better to overlay yesterday's stage list than to # show nothing). Both reads are non-throwing. cached = await self._read_template_cache() if cached is None: cached = await self._read_template_cache_stale() if not cached: return [] for template in cached: if template.approval_code == approval_code: return list(template.stages) return [] @staticmethod def _build_progress_from_instance( instance_detail: dict[str, Any] | None, *, static_stages: list[ApprovalTemplateStage] | None = None, include_raw_identifiers: bool = True, ) -> ApprovalProgress | None: """Collapse Feishu ``task_list`` into a node-level progress summary. One Feishu task is one (node, approver) pair — a 3-approver OR-node shows up as three task rows with the same ``node_id``. The detail-drawer Steps indicator wants one Step per *node*, so this method groups tasks by ``node_id`` (preserving first-seen order), collapses approver lists, and computes a per-node status as: - PENDING if ANY task at the node is PENDING / TRANSFERRED, - REJECTED if any task is REJECTED (and none pending), - APPROVED / PASS / DONE if every task is in that terminal set, - falls through to the first observed status otherwise. Legacy templates / older Feishu responses sometimes omit ``node_id`` — those rows are bucketed one-per-task so the Steps indicator at least shows the right total count instead of collapsing the whole flow into a single step. Returns ``None`` if there's no usable task_list — the frontend renders the timeline section as the fallback. """ if not isinstance(instance_detail, dict): # Static-stage-only path: when there's no live instance # at all (just-submitted flow before the Feishu callback # lands), render the cached stages so the requester # sees the full chain instead of a blank "no actions yet" # placeholder. The renderer treats ``status=None`` as # WAIT so the whole flow displays as not-yet-started. if static_stages: return ApprovalProgress( total_nodes=len(static_stages), completed_nodes=0, current_node_index=None, current_node_name=None, current_approvers=[], is_finished=False, sync_lag=False, nodes=[ ApprovalProgressNode( node_id=stage.node_id, name=stage.name, status=None, approvers=list(stage.approver_names), ) for stage in static_stages ], ) return None task_list = instance_detail.get("task_list") if not isinstance(task_list, list) or not task_list: # See above — same fallback when the instance exists but # Feishu hasn't pushed any approver tasks yet. if static_stages: return ApprovalProgress( total_nodes=len(static_stages), completed_nodes=0, current_node_index=None, current_node_name=None, current_approvers=[], is_finished=False, sync_lag=False, nodes=[ ApprovalProgressNode( node_id=stage.node_id, name=stage.name, status=None, approvers=list(stage.approver_names), ) for stage in static_stages ], ) return None # Defence-in-depth: clamp the Feishu-supplied list to a # generous bound before iterating. The data source is a # trusted outbound HTTP call to Feishu, but a runaway # template / Feishu-side bug could otherwise bloat the # response and the persisted JSON column. 500 is well past # any realistic multi-stage approval flow (~10 nodes × 5 # approvers). task_list = task_list[:500] terminal_pass = {"APPROVED", "PASS", "DONE"} terminal_reject = {"REJECTED", "REJECT"} pending_statuses = {"PENDING", "TRANSFERRED"} # OR-mode "loser" task statuses Feishu emits when one # approver of a multi-approver node acts and the others' # tasks get auto-cleared. These are NOT decisions in their # own right — they should be filtered out of the collapse # so a node where Alice-APPROVED + Bob-REVOKED renders as # APPROVED rather than falling through to the "first # status" branch. skipped_statuses = {"REVOKED", "SKIPPED", "RECALL", "RECALLED"} order: list[str] = [] nodes: dict[str, dict[str, Any]] = {} fallback_idx = 0 for task in task_list: if not isinstance(task, dict): continue node_id_raw = ( task.get("node_id") or task.get("approval_node_id") or task.get("custom_node_id") ) if node_id_raw: node_id = str(node_id_raw) else: fallback_idx += 1 node_id = f"__fb_{fallback_idx}" if node_id not in nodes: order.append(node_id) nodes[node_id] = { "name": ( task.get("node_name") or task.get("approval_node_name") or None ), # Dedup tracking — semantics: # * ``user_id`` is the strong identifier # (Feishu-side unique per human). When a # row supplies one, that's the dedup key # and any uname carried alongside is # remembered too so a later uname-only row # for the same person collapses. # * When a row has NO uid, it dedups against # the seen-uname set — best-effort for the # uid-less anonymous case. # * Different humans sharing a display name # ("张伟"/USR_A then "张伟"/USR_B) are NOT # collapsed by this scheme — Row 2 enters # the uid path with a fresh uid, doesn't # check seen_unames, and is added. The # resulting two-"张伟" entries in the # approver list are unambiguous-but-ugly; # the prior round-2 attempt that did # cross-pollinate uname matches into uid # dedup silently dropped one approver in # this case, which is a worse failure # mode (an admin would not know two # people were actually required). "seen_uids": set(), "seen_unames": set(), "approvers": [], "task_statuses": [], } node = nodes[node_id] if not node["name"]: node["name"] = ( task.get("node_name") or task.get("approval_node_name") or node["name"] ) uid = task.get("user_id") or task.get("open_id") uname = task.get("user_name") or task.get("name") uid_s = str(uid) if uid else None uname_s = str(uname) if uname else None if uid_s: if uid_s in node["seen_uids"]: is_dup = True else: node["seen_uids"].add(uid_s) # Record the name carried alongside this uid # so a later uname-only row for the same # person collapses via the uname path. if uname_s: node["seen_unames"].add(uname_s) is_dup = False elif uname_s: if uname_s in node["seen_unames"]: is_dup = True else: node["seen_unames"].add(uname_s) is_dup = False else: # Neither identifier — silently skip (nothing to # render and nothing to dedup against). is_dup = True if not is_dup: # List surfaces only show resolved display names. A raw # Feishu user id is useful in an admin detail fallback, but # it is not a meaningful "current approver" to an employee. label = uname_s or (uid_s if include_raw_identifiers else None) if label: node["approvers"].append(label) status_raw = task.get("status") if status_raw: node["task_statuses"].append(str(status_raw).upper()) def _collapse_status(statuses: list[str]) -> str | None: if not statuses: return None # Drop OR-mode skipped tasks first so they can't taint # the AND/OR semantics below. A node consisting entirely # of skipped tasks falls through to "SKIPPED" so the # caller can recognise the state. effective = [s for s in statuses if s not in skipped_statuses] if not effective: return statuses[0] if any(s in pending_statuses for s in effective): return "PENDING" if any(s in terminal_pass for s in effective): # First approver in OR-mode wins — promote the node # to APPROVED even if some other task at the same # node was REJECTED (which Feishu sometimes leaves # in for parallel approvers who acted concurrently). return "APPROVED" if any(s in terminal_reject for s in effective): return "REJECTED" return effective[0] progress_nodes: list[ApprovalProgressNode] = [] completed = 0 current_idx: int | None = None current_name: str | None = None current_approvers: list[str] = [] for idx, node_id in enumerate(order, start=1): node = nodes[node_id] status = _collapse_status(node["task_statuses"]) progress_nodes.append( ApprovalProgressNode( # Surface ``node_id`` so the frontend timeline can # bucket ``AdminApprovalTimelineEntry.node_id`` # entries by stable identifier — legacy templates # can have two nodes with the same display name # (two "部门审批" stages) which broke name-based # bucketing in the new Modal redesign (2026-05-15). # Synthetic ``__fb_{i}`` fallback ids land here # unchanged; the frontend treats them the same as # real Feishu ids since both are stable per call. node_id=node_id, name=node["name"], status=status, approvers=list(node["approvers"]), ) ) if status in terminal_pass or status in terminal_reject: completed += 1 if current_idx is None and status == "PENDING": current_idx = idx current_name = node["name"] current_approvers = list(node["approvers"]) # ── Static-stage overlay (2026-05-21) ───────────────────── # Feishu's task_list only carries nodes the approval has # actually traversed. To answer the user's "我希望提交后能看 # 到整个流程,不只是下一步" feature, overlay any stages from # the cached approval definition that DON'T appear in the # live progress_nodes. Match by node_id when the static # stage carries one; otherwise append at the end in # definition order. Appended stages get ``status=None`` so # ``ApprovalProgressNode`` renders them with the # ``wait`` ui-status in the frontend renderer (grey clock, # no approver action), distinct from PENDING (current). if static_stages: seen_node_ids = { pn.node_id for pn in progress_nodes if pn.node_id } for stage in static_stages: if stage.node_id and stage.node_id in seen_node_ids: continue progress_nodes.append( ApprovalProgressNode( node_id=stage.node_id, name=stage.name, status=None, approvers=list(stage.approver_names), ) ) # ``is_finished`` requires BOTH (a) every node is in a # terminal status AND (b) the instance-level status agrees # — OR the instance status is explicitly terminal (e.g. # CANCELLED before tasks completed). Without the AND gate, # a flow whose final node just transitioned to APPROVED # but whose instance status hasn't propagated yet would # render "已完结" while Feishu still says PENDING. # # ``None`` task statuses count as un-finished (NOT silently # ignored). Feishu emits ``status: null`` for indeterminate # observer / CC nodes — treating them as "done" would mark # a CC-only node as terminal and prematurely flip the # whole flow to finished. instance_status = str(instance_detail.get("status") or "").upper() finished_instance_statuses = { "APPROVED", "PASS", "REJECTED", "REJECT", "DONE", "CANCELED", "CANCELLED", "RECALLED", "REVERTED", } instance_says_finished = instance_status in finished_instance_statuses nodes_say_finished = ( current_idx is None and len(progress_nodes) > 0 and all( (n.status is not None and (n.status in terminal_pass or n.status in terminal_reject)) for n in progress_nodes ) ) is_finished = instance_says_finished or ( nodes_say_finished and (not instance_status or instance_status not in {"PENDING", "RUNNING"}) ) # Sync-lag UX: if no node is PENDING but the flow isn't # finished yet (Feishu instance status hasn't propagated, # or final node has only skipped tasks), point # ``current_node_index`` past the last node and set the # explicit ``sync_lag=True`` flag so the frontend renders # a "等待状态同步" hint. The flag is the source of truth — # inferring sync-lag from null names would false-fire on # legitimately-in-progress flows with anonymous nodes. effective_current_idx: int | None effective_current_name: str | None effective_current_approvers: list[str] sync_lag = False if is_finished: effective_current_idx = None effective_current_name = None effective_current_approvers = [] elif current_idx is not None: effective_current_idx = current_idx effective_current_name = current_name effective_current_approvers = current_approvers else: effective_current_idx = len(order) if order else None effective_current_name = None effective_current_approvers = [] sync_lag = True # ``total_nodes`` now reflects the FULL progress_nodes list # (live + overlaid static stages) so the Steps header reads # "1 / 3" instead of "1 / 1" on a freshly-submitted flow that # has only surfaced its first task. ``completed_nodes`` / # ``current_node_index`` stay derived from live data only — # an un-traversed static stage is not completed and is not # current. return ApprovalProgress( total_nodes=len(progress_nodes), completed_nodes=completed, current_node_index=effective_current_idx, current_node_name=effective_current_name, current_approvers=effective_current_approvers, is_finished=is_finished, sync_lag=sync_lag, nodes=progress_nodes, ) @staticmethod def _coerce_feishu_ts(value: Any) -> datetime | None: """Normalize Feishu timestamps to a tz-aware UTC datetime. Feishu's approval API emits timestamps as either: * Integer/string epoch milliseconds (most common in task_list / comment_list), or * ISO-8601 strings (rare, only on a few endpoints). Anything else returns None — the caller buckets None at the end of the timeline. """ if value in (None, "", 0, "0"): return None if isinstance(value, (int, float)): try: # Feishu uses epoch millis. Values < 10^11 would be # epoch seconds (year 1970–5138); guard either way. ts = float(value) if ts > 10**11: ts = ts / 1000.0 return datetime.fromtimestamp(ts, tz=UTC) except (OverflowError, OSError, ValueError): return None if isinstance(value, str): stripped = value.strip() if not stripped: return None # Try epoch-millis-as-string first. if stripped.isdigit(): try: ts = float(stripped) if ts > 10**11: ts = ts / 1000.0 return datetime.fromtimestamp(ts, tz=UTC) except (OverflowError, OSError, ValueError): return None # Fall back to ISO-8601. try: parsed = datetime.fromisoformat(stripped.replace("Z", "+00:00")) except ValueError: return None if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=UTC) return parsed return None