"""IT-facing alert for a failed API-key auto-grant. Why this module exists (2026-08-19, AK-20260819-346E): The three auto-grant helpers (``bailian_autogrant``, ``bedrock_autogrant``, ``azure_openai_autogrant``) all handle their failures the same way — write ``detail..error`` onto the assignment and return a status string that nobody reads. Every one of those branches is therefore **silent**: the Feishu approval closes as APPROVED, the assignment lands in ``approved`` with ``key_value IS NULL``, the requester's 「我的申请」 says 已通过, and the only record of what went wrong is a JSON blob in a DB column that no page renders. The Bailian per-region sync (PR #2772) broke workspace resolution on 2026-08-15 and the failure was only discovered four days later, by a human asking "why hasn't he got his key". So: every failure branch calls :func:`record_autogrant_failure`, which pushes one card to IT. Deliberately **IT-only** — the requester is not a new recipient here (standing rule: new pushes go to global_admin only); IT fills the key manually and the existing ``apikey.ready`` mail then reaches the requester through the normal path. Best-effort by construction: an alert that raises must never turn a "grant failed, key fillable by hand" into a 500 on the Feishu callback (which Feishu would then retry, re-entering the grant). Every failure in here is caught and logged. """ from __future__ import annotations from typing import TYPE_CHECKING from app.audit.service import AuditService from app.core.exceptions import format_exception_for_log from app.core.safe_logging import get_safe_logger from app.core.txn import best_effort if TYPE_CHECKING: # pragma: no cover - typing only from sqlalchemy.ext.asyncio import AsyncSession from app.domains.apikey.models import ApiKeyAssignment logger = get_safe_logger(__name__) # Human labels for the three auto-grant kinds. Keeps the card readable # without every call site having to pass display copy. _KIND_LABELS: dict[str, str] = { "bailian": "Alibaba Cloud Model Studio", "bedrock": "AWS Bedrock", "azure_openai": "Azure OpenAI", } # Admin deep-link. The API-key admin console is where the fill-key / # reissue actions live, so the card lands the reader on the surface that # can actually resolve the failure. _ADMIN_URL = "/admin/assets/api-keys" async def record_autogrant_failure( db: AsyncSession, assignment: ApiKeyAssignment, *, kind: str, stage: str, error: str, provider_label: str | None = None, ) -> None: """Audit + notify IT that an auto-grant failed. ``stage`` is the machine-readable branch name the helper would have returned (``no-default-workspace``, ``ram-user-create-failed``, …) — it goes into the audit row so a later sweep can group failures by cause without regex-ing the message. The caller is expected to have already stamped ``detail.`` and committed. This function does not mutate ``assignment``. """ request_number = assignment.request_number or str(assignment.id) label = provider_label or _KIND_LABELS.get(kind, kind) logger.error( "API-key auto-grant failed: kind=%s stage=%s assignment=%s " "request_number=%s error=%s", kind, stage, assignment.id, request_number, error, ) # Scrub before persisting / sending: an upstream error body can carry # an AK id or a signed URL, and this text goes to an audit row, a # Feishu card and an email. from app.integrations.service import _scrub_message safe_error = _scrub_message(error) # SAVEPOINT-isolated: the audit row is a flush, and a failed flush # deactivates the WHOLE transaction. Swallowing it bare would leave a # dead session for the notification writes below — they would raise, the # exception would escape into the Feishu callback, and the callback would # 500 on a grant that is otherwise fine. Isolate, then continue. async with best_effort( db, label=f"audit autogrant failure {request_number}", log=logger, ) as audit_step: await AuditService(db).log( trace_id=f"apikey.autogrant_failed.{assignment.id}", actor_type="system", actor_user_id=None, action="api_key.autogrant_failed", status="failure", resource_type="api_key_assignment", resource_id=str(assignment.id), detail={ "kind": kind, "stage": stage, "request_number": request_number, "error": safe_error[:1000], }, ) if not audit_step.ok: logger.warning( "autogrant failure audit write failed for %s: %s", request_number, audit_step.reason(), ) # savepoint-exempt: performs no DB writes of its own — every write inside # is individually SAVEPOINT-isolated (see the per-recipient best_effort in # ``_dispatch_admin_cards``). This catch only stops a non-DB bug here # (a renamed template key, a None recipient attribute) from escaping into # the Feishu callback and 500-ing an otherwise-successful callback. try: await _dispatch_admin_cards( db, assignment=assignment, kind=kind, stage=stage, error=safe_error, label=label, request_number=request_number, ) except Exception as exc: # noqa: BLE001 — notify must not mask the grant logger.warning( "autogrant failure alert dispatch failed for %s: %s", request_number, format_exception_for_log(exc), ) async def _dispatch_admin_cards( db: AsyncSession, *, assignment: ApiKeyAssignment, kind: str, stage: str, error: str, label: str, request_number: str, ) -> None: # Imported lazily: the notification service pulls in the whole # channel stack (Feishu client, SMTP, outbox), which the auto-grant # success path should not pay for. from app.notifications.service import ( CardAction, CardStat, LocalizedText, NotificationEnvelope, NotificationService, ) from app.scheduled_jobs.failure_notifier import list_admin_recipients recipients = await list_admin_recipients(db) if not recipients: logger.warning( "autogrant failure for %s but no active IT-admin recipient " "resolved — the failure is only visible in detail.%s.error", request_number, kind, ) return requester = "" if assignment.assigned_to is not None: from app.identity.person_models import IdentityPerson person = await db.get(IdentityPerson, assignment.assigned_to) if person is not None: requester = person.display_name or person.work_email or "" service = NotificationService(db) # Card CHROME (stat labels, subtitle, button) does not go through the # template layer — a bare Chinese string here would reach an English # or Japanese admin untranslated next to a localized body. stats = [ CardStat( label=LocalizedText(zh="申请单", en="Request", ja="申請番号"), value=request_number, ), CardStat( label=LocalizedText(zh="申请人", en="Requester", ja="申請者"), value=requester or "-", ), CardStat( label=LocalizedText(zh="失败环节", en="Stage", ja="失敗段階"), value=stage, ), ] for admin in recipients: if admin.deleted_at is not None: continue if (admin.employment_status or "").lower() != "active": continue # Per-recipient SAVEPOINT: dispatch writes a notification_outbox row, # so one recipient whose write fails (missing work_email, a duplicate # key) must not take the remaining admins down with it — bare except # around a failed flush leaves every later dispatch writing into a # dead transaction. async with best_effort( db, label=f"autogrant alert to {admin.id}", log=logger, ): await service.dispatch(NotificationEnvelope( recipient=admin, title=f"[Helix] API Key 自动发放失败 · {request_number}", body_text=error, channels=["feishu", "email"], template_key="apikey.autogrant_failed", context={ "provider_name": label, "request_number": request_number, "requester_name": requester or "-", "project_name": assignment.project_name or "-", "stage": stage, "error": error, "portal_url": _ADMIN_URL, }, url=_ADMIN_URL, related_resource_type="api_key_assignment", related_resource_id=assignment.id, card_subtitle=LocalizedText( zh="审批已通过，但凭证没有发出去 —— 需要人工补发", en="Approved, but no credential was issued — needs a " "manual fill", ja="承認済みですが認証情報が発行されていません —— " "手動対応が必要です", ), card_stats=stats, card_actions=[CardAction( label=LocalizedText( zh="去处理", en="Open admin console", ja="対応する", ), url=_ADMIN_URL, style="primary", )], ))