"""AI tool / function introduction request flow. Same shape as ``app.domains.announcement_request.service``, but with the multi-field security review form. After approval the row appears on the Approved AI Tools listing. Feishu flow_key: ``ai_tool_application_request``. The external_code (template GUID) is registered via ``/admin/integrations//flows``. """ from __future__ import annotations import json import logging import uuid from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.approval.cancel_desync import notify_feishu_cancel_desync from app.approval.feishu_client import FeishuApprovalClient from app.approval.models import ApprovalRequest, ApprovalSubmissionIntent from app.approval.submission_intents import ( compact_terminal_intent_payload, defer_submission_intent_retry, reconciliation_due_predicate, reconciliation_order, submit_with_durable_intent, terminal_intent_resource_id, ) from app.audit.service import AuditService from app.core.exceptions import AppException, NotFoundError from app.domains.ai_tool_application.models import ( AI_TOOL_DATA_LEVEL_OPTIONS, AI_TOOL_HOW_TO_USE_OPTIONS, AI_TOOL_STATUS_APPROVED, AI_TOOL_STATUS_CANCELLED, AI_TOOL_STATUS_PENDING, AI_TOOL_STATUS_REJECTED, AI_TOOL_STATUS_REVOKED, AiToolApplication, ) from app.identity.person_models import IdentityPerson from app.integrations.compat import get_feishu_config from app.integrations.service import get_approval_flow logger = logging.getLogger(__name__) _FEISHU_FLOW_KEY = "ai_tool_application_request" # Substring match, case-insensitive — keys mirror the Feishu widget # name / label so we can ship a stable form payload regardless of # whether IT renames a widget on the template side. _AI_FIELD_KEYWORDS: dict[str, tuple[str, ...]] = { "ai_name": ("ai", "name", "function", "product", "名称", "产品"), "how_to_use": ("how to use", "use", "方式", "使用方式"), "data_level": ("data level", "level", "等级", "数据等级"), "purpose": ("purpose", "用途", "目的"), "data_type": ("data type", "type", "数据类型"), "privacy_policy": ("privacy", "policy", "隐私", "url"), } _AI_SKIP_WIDGET_TYPES = frozenset( {"text", "section", "image", "attachmentV2"}, ) def _match_widget(widget_name: str, field_key: str) -> bool: name_lower = (widget_name or "").lower() return any(kw.lower() in name_lower for kw in _AI_FIELD_KEYWORDS.get(field_key, ())) def _resolve_multi_choice_codes( widget: dict[str, Any], display_texts: list[str], ) -> list[str]: """Map each picked label to the option's internal ``value`` code. Feishu's ``checkboxV2`` widget rejects label text — its ``value`` field needs the option's stored ``value`` ID (``"mhlooiug-7zva1j5uapi-0"``-style strings). We resolve each user-supplied label by case-insensitive text match. """ options = widget.get("option") or widget.get("options") or [] resolved: list[str] = [] for label in display_texts: needle = label.strip().lower() if not needle: continue match = None for opt in options: if not isinstance(opt, dict): continue opt_text = str(opt.get("text", "")).strip().lower() if opt_text == needle or needle in opt_text: match = str(opt.get("value", "")) break if match: resolved.append(match) return resolved def _build_ai_form_payload( widgets: list[dict[str, Any]], values: dict[str, Any], ) -> str: # Honour Feishu ``display_condition`` — hidden widgets skipped. from app.integrations.feishu_form_visibility import ( evaluate_display_conditions, ) visibility = evaluate_display_conditions( widgets, {}, assume_visible_when_answers_unsourced=True ) serialized: list[dict[str, Any]] = [] for widget in widgets: wtype = widget.get("type", "input") if wtype in _AI_SKIP_WIDGET_TYPES: continue wid = widget.get("id") if not wid: continue if not visibility.get(wid, True): continue wname = widget.get("name") or "" matched_value: Any = None for field_key in _AI_FIELD_KEYWORDS: if _match_widget(wname, field_key) and field_key in values: matched_value = values[field_key] break if matched_value in (None, "", [], {}): continue if wtype in ("checkboxV2", "checkbox", "multiselect"): picked = ( matched_value if isinstance(matched_value, list) else [s.strip() for s in str(matched_value).split(",") if s.strip()] ) picked = _resolve_multi_choice_codes(widget, picked) # Feishu's checkboxV2 widget accepts the value as a plain # array (NOT a JSON-stringified array). The outer # ``form`` field is what gets JSON-serialized — the inner # ``value`` for a multi-select stays a native list. serialized.append({"id": wid, "type": wtype, "value": picked}) continue elif isinstance(matched_value, list): text = ", ".join(str(v) for v in matched_value) else: text = str(matched_value) serialized.append({"id": wid, "type": wtype, "value": text}) return json.dumps(serialized, ensure_ascii=False) def _parse_template_widgets(raw_form: Any) -> list[dict[str, Any]]: if isinstance(raw_form, str): return json.loads(raw_form or "[]") if isinstance(raw_form, list): return raw_form return [] def _normalize_csv(values: list[str], allowed: tuple[str, ...]) -> str: """Validate each picked value against ``allowed`` and return the canonical comma-separated form. Raises if a value is unknown.""" cleaned: list[str] = [] seen: set[str] = set() for v in values: s = (v or "").strip() if not s: continue if s not in allowed: raise AppException( code="AI_TOOL_OPTION_INVALID", message=(f"{s!r} is not one of {list(allowed)}."), status_code=400, ) if s not in seen: seen.add(s) cleaned.append(s) if not cleaned: raise AppException( code="AI_TOOL_OPTION_REQUIRED", message="Please pick at least one option.", status_code=400, ) return ", ".join(cleaned) class AiToolApplicationService: # Outcome of the most recent ``cancel()`` call's best-effort Feishu # side-cancel. Routers / MCP tools read it after ``cancel()`` to # surface ``feishu_cancel_synced`` to the requester. last_feishu_cancel_synced: bool = True def __init__(self, db: AsyncSession) -> None: self.db = db self.audit = AuditService(db) self._feishu_client: FeishuApprovalClient | None = None async def _client(self) -> FeishuApprovalClient: if self._feishu_client is None: cfg = await get_feishu_config(session=self.db) self._feishu_client = FeishuApprovalClient( app_id=str(cfg.get("app_id") or ""), app_secret=str(cfg.get("app_secret") or ""), ) return self._feishu_client # ── Submit ────────────────────────────────────────────────── async def submit( self, *, requester: IdentityPerson, ai_name: str, how_to_use: list[str], data_level: list[str], purpose: str, data_type: str, privacy_policy: str, trace_id: str, ) -> AiToolApplication: # Round 2b audit fix-up 2026-05-04: service-layer leaver gate # to mirror the router-level dependency for MCP-direct callers. if requester.employment_status in ("leaver_pending", "left"): raise AppException( code="AI_TOOL_REQUESTER_NOT_ACTIVE", message=( "Your account is in offboarding; new AI tool applications are not accepted." ), status_code=403, ) clean_name = (ai_name or "").strip() if not clean_name: raise AppException( code="AI_TOOL_NAME_REQUIRED", message="Please name the AI function or product.", status_code=400, ) clean_how = _normalize_csv(how_to_use, AI_TOOL_HOW_TO_USE_OPTIONS) clean_level = _normalize_csv(data_level, AI_TOOL_DATA_LEVEL_OPTIONS) clean_purpose = (purpose or "").strip() clean_dtype = (data_type or "").strip() clean_privacy = (privacy_policy or "").strip() for label, value in ( ("purpose", clean_purpose), ("data_type", clean_dtype), ("privacy_policy", clean_privacy), ): if not value: raise AppException( code=f"AI_TOOL_{label.upper()}_REQUIRED", message=f"Please fill in {label.replace('_', ' ')}.", status_code=400, ) # Atomicity (review batch 33): the duplicate-pending guard # below is check-then-act with no lock and no backing unique # constraint, so two concurrent submits (double-click / agent # retry) could both pass the check and each fire a Feishu # approval instance. Mirror the canonical ``apikey.submit`` # pattern: take a per-requester distributed lock and re-check # pending INSIDE the lock so only one submission proceeds. from app.core.distributed_lock import ( # noqa: PLC0415 LockBusy, distributed_lock, ) lock_key = f"ai_tool_submit:{requester.id}" try: async with distributed_lock(lock_key, ttl=120): # Pre-launch hardening 2026-05-04: reject duplicate pending # submissions by the same requester (double-click / agent retry). existing_pending = await self.db.scalar( select(AiToolApplication.id) .where( AiToolApplication.person_id == str(requester.id), AiToolApplication.status == AI_TOOL_STATUS_PENDING, ) .limit(1) .with_for_update() ) if existing_pending: raise AppException( code="AI_TOOL_ALREADY_PENDING", message=( "You already have a pending AI tool application. " "Cancel it first or wait for it to be processed." ), status_code=409, ) flow = await get_approval_flow(_FEISHU_FLOW_KEY, session=self.db) approval_code = flow.external_code client = await self._client() definition = await client.get_approval_definition(approval_code) widgets = _parse_template_widgets(definition.get("form", "[]")) field_values: dict[str, Any] = { "ai_name": clean_name, "how_to_use": [s.strip() for s in clean_how.split(",") if s.strip()], "data_level": [s.strip() for s in clean_level.split(",") if s.strip()], "purpose": clean_purpose, "data_type": clean_dtype, "privacy_policy": clean_privacy, } form_payload = _build_ai_form_payload(widgets, field_values) feishu_user_id = await client.resolve_user_id_by_email( requester.work_email, ) app_id = str(uuid.uuid4()) approval_request_id = str(uuid.uuid4()) intent = await submit_with_durable_intent( self.db, client=client, scope=_FEISHU_FLOW_KEY, applicant_id=requester.id, semantic_payload={ "ai_name": clean_name, "how_to_use": clean_how, "data_level": clean_level, "purpose": clean_purpose, "data_type": clean_dtype, "privacy_policy": clean_privacy, }, provider_payload={ "approval_code": approval_code, "identity_type": "user_id", "identity_value": feishu_user_id, "form": form_payload, }, finalization_payload={ "application_id": app_id, "approval_request_id": approval_request_id, "approval_code": approval_code, "ai_name": clean_name, "how_to_use": clean_how, "data_level": clean_level, "purpose": clean_purpose, "data_type": clean_dtype, "privacy_policy": clean_privacy, }, trace_id=trace_id, ) return await self._finalize_submission_intent( intent.id, recovered=intent.attempt_count > 1, ) except LockBusy as exc: raise AppException( code="AI_TOOL_SUBMIT_IN_FLIGHT", message=( "Another submission of yours is being processed. Please retry in a few seconds." ), status_code=409, ) from exc async def _finalize_submission_intent( self, intent_id: uuid.UUID, *, recovered: bool, ) -> AiToolApplication: 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.scope != _FEISHU_FLOW_KEY: raise RuntimeError("AI-tool submission intent disappeared") if intent.state == "completed": app_id = terminal_intent_resource_id( intent, resource_type="ai_tool_application", ) existing = await self.db.get(AiToolApplication, app_id) if existing is None: raise RuntimeError("completed AI-tool intent has no row") return existing payload = dict((intent.payload or {}).get("finalization") or {}) app_id = str(payload["application_id"]) existing = await self.db.get(AiToolApplication, app_id) if intent.state != "provider_created" or not intent.external_id: raise RuntimeError(f"cannot finalize AI-tool intent in {intent.state!r}") now = datetime.now(UTC) if existing is None: row = AiToolApplication( id=app_id, person_id=str(intent.applicant_id), ai_name=str(payload["ai_name"]), how_to_use=str(payload["how_to_use"]), data_level=str(payload["data_level"]), purpose=str(payload["purpose"]), data_type=str(payload["data_type"]), privacy_policy=str(payload["privacy_policy"]), status=AI_TOOL_STATUS_PENDING, feishu_instance_code=intent.external_id, created_at=now, updated_at=now, ) self.db.add(row) await self.db.flush() approval_request = ApprovalRequest( id=uuid.UUID(str(payload["approval_request_id"])), external_id=intent.external_id, approval_type=_FEISHU_FLOW_KEY, applicant_id=intent.applicant_id, status=AI_TOOL_STATUS_PENDING, source_payload={ "ai_tool_application_id": app_id, "approval_code": payload["approval_code"], "ai_name": payload["ai_name"], }, ) self.db.add(approval_request) await self.db.flush() row.approval_request_id = str(approval_request.id) row.updated_at = now await self.db.flush() else: row = existing intent.approval_request_id = uuid.UUID(str(payload["approval_request_id"])) intent.state = "completed" intent.completed_at = now intent.last_error = None await self.audit.log( trace_id=( f"approval-intent-reconcile:{intent.id}" if recovered else str((intent.payload or {}).get("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 "ai_tool_application.submit" ), status="success", resource_type="ai_tool_application", resource_id=app_id, detail={ "intent_id": str(intent.id), "provider_request_uuid": intent.provider_request_uuid, "feishu_instance_code": intent.external_id, "ai_name": payload["ai_name"], "recovered": recovered, }, ) compact_terminal_intent_payload( intent, resource_type="ai_tool_application", resource_id=app_id, ) await self.db.commit() await self.db.refresh(row) return row async def reconcile_submission_intents( self, *, min_age_minutes: int = 2, batch_limit: int = 100, ) -> dict[str, int]: cutoff = datetime.now(UTC) - timedelta(minutes=min_age_minutes) states = ("prepared", "submitting", "unknown", "provider_created") intent_refs = list( ( await self.db.execute( select( ApprovalSubmissionIntent.id, ApprovalSubmissionIntent.applicant_id, ) .where( ApprovalSubmissionIntent.scope == _FEISHU_FLOW_KEY, ApprovalSubmissionIntent.state.in_(states), (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_refs), "reconciled": 0, "still_unknown": 0, "failed": 0} if not intent_refs: return summary client = await self._client() from app.core.distributed_lock import ( # noqa: PLC0415 LockBusy, distributed_lock, ) for intent_id, applicant_id in intent_refs: try: async with distributed_lock( f"ai_tool_submit:{applicant_id}", ttl=120, ): intent = await self.db.get(ApprovalSubmissionIntent, intent_id) if intent is None or intent.state == "completed": continue stored = dict(intent.payload or {}) intent = await submit_with_durable_intent( self.db, client=client, scope=intent.scope, applicant_id=intent.applicant_id, recovery_key=stored.get("recovery_key"), semantic_payload=dict(stored.get("semantic") or {}), provider_payload=dict(stored.get("provider") or {}), finalization_payload=dict(stored.get("finalization") or {}), trace_id=str(stored.get("trace_id") or f"approval-intent:{intent.id}"), ) await self._finalize_submission_intent(intent.id, recovered=True) summary["reconciled"] += 1 except LockBusy as exc: await defer_submission_intent_retry(self.db, intent_id, exc) summary["still_unknown"] += 1 except AppException as exc: if exc.code == "APPROVAL_SUBMISSION_RESULT_UNKNOWN": await self.db.rollback() else: await defer_submission_intent_retry(self.db, intent_id, exc) summary[ "still_unknown" if exc.code == "APPROVAL_SUBMISSION_RESULT_UNKNOWN" else "failed" ] += 1 except Exception as exc: # noqa: BLE001 await defer_submission_intent_retry(self.db, intent_id, exc) summary["failed"] += 1 return summary # ── Cancel ────────────────────────────────────────────────── async def cancel( self, *, app_id: str, actor: IdentityPerson, trace_id: str, ) -> AiToolApplication: row = await self._lock(app_id) if row.person_id != str(actor.id): raise AppException( code="AI_TOOL_NOT_OWNER", message="Only the original requester can cancel.", status_code=403, ) if row.status != AI_TOOL_STATUS_PENDING: raise AppException( code="AI_TOOL_INVALID_STATE", message=(f"Cannot cancel an application in {row.status!r} state."), status_code=409, ) self.last_feishu_cancel_synced = True if row.feishu_instance_code: try: client = await self._client() flow = await get_approval_flow( _FEISHU_FLOW_KEY, session=self.db, ) feishu_user_id = await client.resolve_user_id_by_email( actor.work_email, ) await client.cancel_instance( instance_code=row.feishu_instance_code, user_id=feishu_user_id, approval_code=flow.external_code, ) except Exception as e: # noqa: BLE001 logger.warning( "Feishu cancel failed for ai_tool_application %s (instance=%s): %s", row.id, row.feishu_instance_code, e, ) self.last_feishu_cancel_synced = False await notify_feishu_cancel_desync( self.db, flow_label="AI 工具申请", resource_type="ai_tool_application", resource_id=str(row.id), instance_code=row.feishu_instance_code, error=e, trace_id=trace_id, ) row.status = AI_TOOL_STATUS_CANCELLED row.updated_at = datetime.now(UTC) await self._mirror_approval_request_status(row, "cancelled") await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor.id, action="ai_tool_application.cancel", status="success", resource_type="ai_tool_application", resource_id=row.id, ) return row # ── Admin revoke (offboarding T+0 force-revoke) ───────────── async def revoke( self, *, app_id: str, actor: IdentityPerson, reason: str | None, trace_id: str, ) -> AiToolApplication: """Flip an APPROVED AI tool application to REVOKED. Two callers: the offboarding force-*revoke* path (leaver held an approved seat) and the standalone admin revoke endpoint / ``revoke_ai_tool_application`` MCP tool (2026-06-13). Distinct from ``cancel`` (requester-initiated, PENDING-only) and ``handover`` (transfer to a peer). Mirrors :meth:`VpnApplicationService.revoke`. """ row = await self._lock(app_id) if row.status != AI_TOOL_STATUS_APPROVED: raise AppException( code="AI_TOOL_NOT_REVOKABLE", message=(f"Cannot revoke an application in {row.status!r} state."), status_code=409, ) now = datetime.now(UTC) row.status = AI_TOOL_STATUS_REVOKED row.updated_at = now await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor.id, action="ai_tool_application.revoke", status="success", resource_type="ai_tool_application", resource_id=row.id, detail={"reason": (reason or "").strip()[:255] or None}, ) return row # ── Handover (peer transfer + offboarding successor) ──────── async def handover( self, *, app_id: str, successor_person_id: str, actor: IdentityPerson, trace_id: str, ) -> AiToolApplication: """Transfer an APPROVED AI tool application to a peer. Lifecycle-verb parity with domain_registration / sso handover: a paid named AI-tool seat shouldn't be silently lost when its holder changes role or leaves — it can be handed to an active colleague. Mirrors ``DomainRegistrationService.handover`` (same active-employee successor gate so we never hand a seat to a leaver). Reused by both the requester self-service path and the offboarding resolver. """ row = await self._lock(app_id) if row.status != AI_TOOL_STATUS_APPROVED: raise AppException( code="AI_TOOL_NOT_HANDOVERABLE", message=(f"Cannot hand over an application in {row.status!r} state."), status_code=409, ) successor = await self.db.get(IdentityPerson, uuid.UUID(successor_person_id)) if successor is None or successor.deleted_at is not None: raise NotFoundError("IdentityPerson", successor_person_id) if str(successor.id) == str(row.person_id): raise AppException( code="AI_TOOL_SAME_OWNER", message="Successor must differ from the current holder.", status_code=400, ) successor_status = (successor.employment_status or "").lower() if successor_status in {"left", "leaver_pending"}: raise AppException( code="AI_TOOL_SUCCESSOR_INACTIVE", message=( f"Cannot hand over AI tool to " f"{successor.work_email or successor.display_name} — " f"they are {successor_status or 'inactive'}. " "Pick an active employee." ), status_code=409, ) prior_person_id = row.person_id row.person_id = str(successor.id) row.updated_at = datetime.now(UTC) await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor.id, action="ai_tool_application.handover", status="success", resource_type="ai_tool_application", resource_id=row.id, detail={ "from_person_id": prior_person_id, "to_person_id": str(successor.id), }, ) return row # ── Feishu callback ───────────────────────────────────────── async def handle_feishu_callback( self, *, approval_request: ApprovalRequest, instance_detail: dict[str, Any], ) -> None: source = approval_request.source_payload or {} app_id = source.get("ai_tool_application_id") if not app_id: logger.warning( "ai_tool_application callback missing id: ar=%s", approval_request.id, ) return feishu_status = (instance_detail.get("status") or "").upper() try: row = await self._lock(app_id) except NotFoundError: logger.warning( "ai_tool_application callback for missing row %s — ignoring", app_id, ) return if row.status != AI_TOOL_STATUS_PENDING: return now = datetime.now(UTC) if feishu_status in ("APPROVED", "PASS"): # Pre-launch hardening 2026-05-06: re-validate requester is # still active. Feishu approvals can sit pending for days, # and an offboarded employee getting AI-tool access provisioned # post-departure is exactly the soft-launch we must prevent. requester = await self.db.get(IdentityPerson, row.person_id) if ( requester is None or requester.deleted_at is not None or (requester.employment_status in ("left", "leaver_pending")) ): row.status = AI_TOOL_STATUS_REJECTED row.updated_at = now requester_status = requester.employment_status if requester else None await self.audit.log( trace_id=f"feishu_callback:{approval_request.external_id}", actor_type="integration", actor_user_id=None, action="ai_tool_application.callback_refused_leaver", status="failure", resource_type="ai_tool_application", resource_id=row.id, detail={ "instance_code": approval_request.external_id, "feishu_status": feishu_status, "requester_status": requester_status, "reason": "requester_offboarded_during_pending", }, ) return row.status = AI_TOOL_STATUS_APPROVED row.granted_at = now elif feishu_status in ("REJECTED", "REFUSED"): row.status = AI_TOOL_STATUS_REJECTED elif feishu_status in ("CANCELED", "CANCELLED", "DELETED"): row.status = AI_TOOL_STATUS_CANCELLED else: return row.updated_at = now await self.audit.log( trace_id=f"feishu_callback:{approval_request.external_id}", actor_type="integration", actor_user_id=None, action=f"ai_tool_application.{row.status}", status="success", resource_type="ai_tool_application", resource_id=row.id, detail={ "instance_code": approval_request.external_id, "feishu_status": feishu_status, }, ) # ── Helpers ───────────────────────────────────────────────── async def _lock(self, app_id: str) -> AiToolApplication: # SELECT ... FOR UPDATE so a concurrent cancel + Feishu approval # webhook can't race into a split-brain. Previously this was a # plain ``db.get`` — the helper's name implied a lock that # wasn't there. r = ( await self.db.execute( select(AiToolApplication).where(AiToolApplication.id == app_id).with_for_update() ) ).scalar_one_or_none() if r is None: raise NotFoundError("AiToolApplication", app_id) return r async def _mirror_approval_request_status( self, row: AiToolApplication, new_status: str, ) -> None: if not row.feishu_instance_code: return ar = ( await self.db.execute( select(ApprovalRequest).where( ApprovalRequest.external_id == row.feishu_instance_code, ApprovalRequest.approval_type == _FEISHU_FLOW_KEY, ) ) ).scalar_one_or_none() if ar is None: return ar.status = new_status ar.updated_at = datetime.now(UTC)