import re import uuid from datetime import date as date_type from datetime import datetime from decimal import Decimal from typing import Any from sqlalchemy import ( JSON, Boolean, Date, DateTime, ForeignKey, Index, Integer, Numeric, String, Text, UniqueConstraint, text, ) from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy.sql import func from app.core.database import Base from app.core.sql_types import GUID as Uuid class ApiKeyProvider(Base): """Admin-maintained catalog of available AI API key sources.""" __tablename__ = "api_key_providers" __mapper_args__ = {"eager_defaults": True} id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) name: Mapped[str] = mapped_column(String(128), nullable=False, unique=True) description: Mapped[str | None] = mapped_column(Text) is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, server_default=text("1")) sort_order: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("0")) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now(), ) assignments: Mapped[list["ApiKeyAssignment"]] = relationship(back_populates="provider") class ApiKeyAssignment(Base): """An API key assigned to an employee, created via Feishu approval flow.""" __tablename__ = "api_key_assignments" __mapper_args__ = {"eager_defaults": True} __table_args__ = ( Index("ix_api_key_assignments_assigned_to", "assigned_to"), Index("ix_api_key_assignments_provider_id", "provider_id"), Index("ix_api_key_assignments_status", "status"), Index( "ix_apikey_assignments_revoke_actor_at", "revoke_actor_role", "revoked_at", ), # Powers the ``apikey.rotation_retire`` cron's due-rotation scan # (WHERE rotation_retire_at <= now). Mirrors migration 401. Index( "ix_api_key_assignments_rotation_retire_at", "rotation_retire_at", ), ) id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) provider_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("api_key_providers.id"), nullable=False, ) assigned_to: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("identity_persons.id"), nullable=False, ) status: Mapped[str] = mapped_column( String(32), nullable=False, server_default=text("'pending_approval'"), ) # Approval tracking approval_request_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("approval_requests.id"), ) # Back-link to the unified employee_requests row created alongside # the assignment. Added in migration 030. Kept nullable so pre-030 # rows that predate the dual-write stay valid until backfilled. employee_request_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("employee_requests.id"), ) request_number: Mapped[str] = mapped_column(String(64), nullable=False, unique=True) # Application form data project_name: Mapped[str] = mapped_column(String(256), nullable=False) reason: Mapped[str] = mapped_column(Text, nullable=False) monthly_budget_usd: Mapped[Decimal] = mapped_column(Numeric(10, 2), nullable=False) expected_usage: Mapped[str | None] = mapped_column(Text) # Key value (filled by admin after approval) key_value: Mapped[str | None] = mapped_column(Text) # Validity valid_from: Mapped[date_type | None] = mapped_column(Date) valid_until: Mapped[date_type | None] = mapped_column(Date) extended_count: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("0")) # ── IAM / RAM binding ──────────────────────────────────────── # Populated for cloud providers whose auto-grant mints a dedicated # IAM/RAM user (AWS Bedrock, Aliyun Bailian) so the admin UI can # show "this assignment is bound to IAM user X in account Y" and # the rebind endpoint has a structured target instead of having # to parse detail.bedrock JSON. # # All three columns are nullable: legacy rows have None, and # non-cloud providers (OpenAI, Azure OpenAI, …) never set them. iam_user_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("cmdb_iam_users.id", ondelete="SET NULL"), ) iam_user_arn: Mapped[str | None] = mapped_column(String(512)) iam_integration_id: Mapped[str | None] = mapped_column( String(36), ForeignKey("integrations.id", ondelete="SET NULL"), ) # ── Cloud-scope binding (migration 078) ────────────────────── # Used when an assignment's "AI workload" lives inside a cloud # sub-account / subscription (Azure OpenAI sub, AWS SG, GCP AI # project) rather than tied to a single IAM user. The dashboard # uses this to chargeback the scope's monthly billing to the # employees whose API keys point at it. Mutually compatible with # ``iam_user_id`` — an Azure OpenAI key issued from a SP can # legitimately have BOTH set (the IAM user owns the key, the # scope is where the cost is). Either or both may be NULL on # legacy / external (OpenAI / Anthropic SaaS) rows. cloud_scope_id: Mapped[str | None] = mapped_column( String(36), ForeignKey("cmdb_cloud_scopes.id", ondelete="SET NULL"), ) # ── Credential rotation retire schedule (migration 401) ────── # When an assignment is mid-rotation (a fresh cloud credential has # been minted and is live alongside the old one during a grace # window), this column holds the UTC moment the *previous* # credential should be torn down. The full rotation state — # provider-specific handle of the old credential to delete, who # initiated, grace days, the stashed prior ``key_value`` for a # cancel/revert — lives on ``detail..rotation``; this # column is the denormalised timestamp so the ``apikey.rotation_retire`` # cron scans due rotations with an index and the UI renders the # "轮换中 · 旧 key X 停用" badge without decrypting the key or # parsing the detail blob. # # NULL on every assignment not currently mid-rotation. Set at # rotation start; cleared back to NULL when the retire cron / # force-complete / cancel resolves the rotation, and on revoke # (the whole cloud identity is torn down, so any pending retire is # moot). rotation_retire_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True, ) # Feishu detail detail: Mapped[Any | None] = mapped_column(JSON, default=dict) # Per-assignment budget-alert recipient override (migration 146). # # Semantics: # * ``None`` (default for existing rows) — use the default # "applicant + their HRIS direct manager" derivation. This is # what every legacy row falls through to and what the # ``apikey.budget_check`` cron has always done. # * ``[]`` (empty list) — admin override "no recipients". Budget # alerts are suppressed for this assignment. Dangerous but # legitimate (paused project, deliberate burn-down). The # suppression is logged in the audit row at edit time so # compliance can spot it. # * ``[, ...]`` — exactly these ``identity_persons.id`` # values (stringified UUIDs) receive in-app + Feishu alerts. # The applicant is NOT auto-included; admins explicitly choose # every recipient. # # Stored as JSON (not a join table) — the cron reads this column # once per assignment per tick, no fan-out query, and recipients # are dereferenced against ``identity_persons`` at dispatch time so # a deleted person id silently drops out. UUIDs are stored as # strings (not int) because ``identity_persons.id`` is a UUID # column; the service layer canonicalises to lowercase # str(uuid.UUID(...)) before write so equality lookups are stable. notify_recipients: Mapped[list[str] | None] = mapped_column( JSON, nullable=True, ) # ── Revocation metadata (migration 183) ────────────────────── # Denormalised forensic columns mirroring the audit row written at # revoke time. Lets the admin table show "who / when / why / which # surface" inline without a JSON predicate scan over audit_logs. # See migration 183's module docstring for the full rationale and # the per-column semantics — including why the revoked_by column # holds the assignee (NOT the leader) on self_with_approval rows. # # NULL on rows that are not in status='revoked'. Backfilled on # historical revoked rows only for ``revoked_at`` (= updated_at) and # ``revoke_actor_role`` (= 'admin'); reason_code / reason_note / # revoked_by stay NULL for pre-PR rows — the truth lives in # audit_logs and we don't risk a JSON misparse. revoked_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True, ) revoked_by: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"), nullable=True, ) revoke_reason_code: Mapped[str | None] = mapped_column( String(32), nullable=True, ) revoke_reason_note: Mapped[str | None] = mapped_column( Text, nullable=True, ) # ``admin`` | ``self_with_approval`` | ``system_offboarding``. # ``self_with_approval`` reads as a single token (not just ``self``) # because it captures BOTH facts: the employee initiated it AND a # leader signed off — without the suffix a future ``self`` row # written by some bypass path would be indistinguishable on the # report. Future revocation surfaces (e.g. CI-driven leak sweep) # introduce new tokens here. revoke_actor_role: Mapped[str | None] = mapped_column( String(24), nullable=True, ) # ── Transfer ancestry (migration 184) ──────────────────────── # The *most recent* prior owner. NULL on a row that has never been # transferred. Multi-hop history (key A → B → C → D) lives in # ``audit_logs`` under ``apikey.transfer.complete``; this column is # the cheap denormalised hint so the admin table and the assignee's # self-service view can answer "where did this key come from?" # without an audit-log drill-down. Pairs with the existing # ``assigned_to`` (current owner) — the two together describe the # last hop of the chain. transferred_from: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"), nullable=True, ) # Timestamps created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now(), ) provider: Mapped["ApiKeyProvider"] = relationship(back_populates="assignments") class ApiKeyAssignmentMember(Base): """Shared-access roster row for an :class:`ApiKeyAssignment`. Until migration 200, an API Key was a strict 1:1 between the applicant (``ApiKeyAssignment.assigned_to``) and a single human; only that person — plus admins — could read the key's MTD spend / usage detail / cost breakdown via the ``/spend`` family of endpoints. In practice keys are issued for a *team*: the project lead applies, then the colleagues actually using the same SDK credential need to see the budget burn. The lead had to either screenshot the spend panel every week or hand admin rights to a teammate — both bad. This row is the per-(assignment, person) entry on the roster. The Owner (``assigned_to``) maintains the list; every listed person inherits read access to the same surfaces the Owner reads — ``get_assignment_spend`` / ``get_assignment_spend_detail`` / ``reveal_key`` / inclusion in ``get_my_api_keys`` / detail page — but NOT the owner-only mutations (member add / remove, transfer, release, revoke, IAM rebind, extend, edit budget). See :meth:`ApiKeyService.is_owner_or_member` for the unified authz helper every read-side endpoint consults. Schema notes (mirror :func:`alembic.versions.200_apikey_assignment_members.upgrade`): * ``UNIQUE(assignment_id, person_id)`` — a person appears at most once on a key's roster. Service-layer ``add_member`` upserts on this constraint so a double-click is idempotent. * ``person_id`` FK with ``ON DELETE CASCADE`` — a hard-deleted person carries no useful state on the roster; soft-deleted / ``left`` employees are filtered at read/write time by the application layer (``employment_status != 'left'``) so the Owner doesn't see a leaver in the member list and the leaver runtime gate already blocks mutations. * ``added_by`` FK with ``ON DELETE SET NULL`` — forensic; survives a hard-deleted Owner so "who added this teammate?" is still answerable from the row. * Indexes: ``(person_id)`` powers the "my-as-member api keys" augment on ``GET /apikeys/my``; ``(assignment_id)`` powers the "list this key's members" panel on the detail page. """ __tablename__ = "apikey_assignment_members" __mapper_args__ = {"eager_defaults": True} __table_args__ = ( UniqueConstraint( "assignment_id", "person_id", name="uq_apikey_assignment_members_assign_person", ), Index("ix_apikey_assignment_members_person_id", "person_id"), Index( "ix_apikey_assignment_members_assignment_id", "assignment_id", ), ) id: Mapped[uuid.UUID] = mapped_column( Uuid, primary_key=True, default=uuid.uuid4, ) assignment_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("api_key_assignments.id", ondelete="CASCADE"), nullable=False, ) person_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="CASCADE"), nullable=False, ) added_by: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"), nullable=True, ) added_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) class ApiKeyBudgetAlertsDedup(Base): """Idempotency window for the ``apikey.budget_check`` cron. UNIQUE on ``(assignment_id, billing_month, threshold_pct)``. The cron runs daily; each :class:`ApiKeyAssignment` carries a ``monthly_budget_usd`` cap. We fire two thresholds — 80% (warning) and 100% (over-budget) — and need each to land at most once per calendar month per assignment, regardless of how many times the cron sees the same overrun. Mirrors :class:`CmdbBillingAlertsDedup` structurally — the model lives here (apikey domain) instead of cmdb because the dedup key is per-assignment, not per-cloud-tenant. Decision matrix at fire time: * No row exists for (assignment, month, threshold_pct) → send. * Row exists → SKIP (the threshold already fired this month). Why the schema is simpler than the cmdb dedup row: each threshold is monotonic — once 80% fires for May, the next interesting signal is 100% (a different ``threshold_pct`` → a different row). Re-firing 80% mid-month after a refund-driven dip would re-spam finance and isn't actionable. So we land one row per (assignment, month, threshold) and never update it. The ``last_alerted_*`` columns are kept for audit / debugging only. ``assignment_id`` is a UUID FK to ``api_key_assignments`` with ``ON DELETE CASCADE`` — if the assignment is deleted the dedup history is meaningless. ``billing_month`` is the YYYY-MM string the cron formatted via ``%Y-%m`` (Asia/Shanghai-anchored, same convention as ``software_billing.auto_close``). """ __tablename__ = "apikey_budget_alerts_dedup" __mapper_args__ = {"eager_defaults": True} __table_args__ = ( # ``UniqueConstraint`` (not Index(..., unique=True)) so the # alembic-autogen DDL matches migration 143's # ``UniqueConstraint`` exactly — otherwise a future autogen # diff would think the model and migration disagree and emit # a churn ALTER. Same constraint, more honest model. UniqueConstraint( "assignment_id", "billing_month", "threshold_pct", name="uq_apikey_budget_alerts_dedup_assign_month_pct", ), Index( "ix_apikey_budget_alerts_dedup_month", "billing_month", ), ) id: Mapped[uuid.UUID] = mapped_column( Uuid, primary_key=True, default=uuid.uuid4, ) assignment_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("api_key_assignments.id", ondelete="CASCADE"), nullable=False, ) billing_month: Mapped[str] = mapped_column(String(7), nullable=False) # ``80`` or ``100`` — the threshold percentage that fired. Stored # as a small integer rather than a float so duplicate rows for # near-equal thresholds (e.g. 80.0 vs 80.0001 from float drift) # cannot bypass the UNIQUE constraint. threshold_pct: Mapped[int] = mapped_column(Integer, nullable=False) last_alerted_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, ) last_alerted_amount_usd: Mapped[Decimal] = mapped_column( Numeric(14, 2), nullable=False, ) last_alerted_budget_usd: Mapped[Decimal] = mapped_column( Numeric(14, 2), nullable=False, ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) class ApiKeyExpiryAlertsDedup(Base): """Idempotency window for the ``apikey.expiry_check`` cron (PR3-2). UNIQUE on ``(assignment_id, valid_until_iso, threshold_days)``. The cron runs daily and fires two thresholds — 7 days out (advance reminder) and 1 day out (final warning) — and needs each to land at most once per assignment per validity window. Re-firing the same threshold tomorrow would spam the assignee; re-firing it after the admin extends the validity is desirable (different ``valid_until`` → different dedup row). Structure mirrors :class:`ApiKeyBudgetAlertsDedup` exactly: simple ``(scope_key, period_key, threshold_key)`` triple with monotonic thresholds. ``valid_until_iso`` is stored as a ``DATE`` column (the same shape as ``api_key_assignments.valid_until``) so a date drift between worker timezones doesn't change the dedup key. The cron anchors "today" to Asia/Shanghai and computes ``days_left = assignment.valid_until - today``; the threshold key is one of ``7`` / ``1`` so the cron skips firing 7-day when 1-day is more urgent. """ __tablename__ = "apikey_expiry_alerts_dedup" __mapper_args__ = {"eager_defaults": True} __table_args__ = ( UniqueConstraint( "assignment_id", "valid_until_iso", "threshold_days", name="uq_apikey_expiry_alerts_dedup_assign_until_threshold", ), Index( "ix_apikey_expiry_alerts_dedup_valid_until", "valid_until_iso", ), ) id: Mapped[uuid.UUID] = mapped_column( Uuid, primary_key=True, default=uuid.uuid4, ) assignment_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("api_key_assignments.id", ondelete="CASCADE"), nullable=False, ) valid_until_iso: Mapped[date_type] = mapped_column(Date, nullable=False) threshold_days: Mapped[int] = mapped_column(Integer, nullable=False) last_alerted_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) class ApiKeyIdleAlertsDedup(Base): """Idempotency window for the ``apikey.idle_check`` cron (migration 603). UNIQUE on ``(assignment_id, billing_month)``: the monthly sweep asks "was this key idle last month?" at most once per key per month. The row exists to record that the question was ASKED — delivery failures do not roll it back (same posture as the budget / expiry dedups), and ``employee_notified`` records whether the holder was actually paged (that leg is behind ``apikey.idle_alert.enabled_for_employees``, so a row with ``employee_notified=False`` was IT-digest only). ``consecutive_months`` is the running streak at the time of the row — the digest sorts on it so a key that has been idle for four months outranks one that just went quiet. """ __tablename__ = "apikey_idle_alerts_dedup" __mapper_args__ = {"eager_defaults": True} __table_args__ = ( UniqueConstraint( "assignment_id", "billing_month", name="uq_apikey_idle_alerts_assignment_month", ), ) id: Mapped[uuid.UUID] = mapped_column( Uuid, primary_key=True, default=uuid.uuid4, ) assignment_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("api_key_assignments.id", ondelete="CASCADE"), nullable=False, ) billing_month: Mapped[str] = mapped_column(String(7), nullable=False) consecutive_months: Mapped[int] = mapped_column( Integer, nullable=False, default=1, server_default=text("1"), ) employee_notified: Mapped[bool] = mapped_column( Boolean, nullable=False, default=False, server_default=text("0"), ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) class ApiKeyReleaseRequest(Base): """Self-service release-approval row — Helix-internal, no Feishu template. The employee submits a release via :meth:`ApiKeyService.submit_release_request`; the row is inserted in ``status='pending'``, the parent :class:`ApiKeyAssignment` flips to ``pending_release``, and the assignee's direct manager (``identity_persons.manager_person_id``) receives a push card linking to the Helix Web UI approval page. The manager — or any IT Admin — clicks Approve / Reject inside Helix; no Feishu approval template, no callback round-trip. On approve we call :meth:`ApiKeyService.revoke_key` with ``actor_role=self_with_approval`` (so the audit trail makes clear the employee initiated it and a leader signed off). On reject we revert the assignment to its prior status (captured here as ``previous_status``) and notify the requester. Why a dedicated table and not a reuse of ``approval_requests``: the ``approval_requests`` table is the lifecycle wrapper for the Feishu-approval-template flows (api_key_request, dns_request, etc.) where ``external_id = feishu instance_code``. Our release flow has no Feishu instance, no external_id, and a different status enum. A dedicated table keeps the schema honest — ``external_id`` does not become a polymorphic ``NULL or instance_code`` column, and other approval-callback machinery (``approval/service.py``) does not have to grow a Helix-internal branch. """ __tablename__ = "api_key_release_requests" __mapper_args__ = {"eager_defaults": True} __table_args__ = ( # "My pending releases as requester" + "my pending releases as # approver" both want a (person_id, status) composite scan. Index( "ix_apikey_release_requests_requester_status", "requester_id", "status", ), Index( "ix_apikey_release_requests_approver_status", "approver_id", "status", ), # Cross-state guard / "find the open request for this AK" — only # one pending request per assignment is allowed; the lookup is # by assignment_id + status. Index( "ix_apikey_release_requests_assignment_status", "assignment_id", "status", ), ) id: Mapped[uuid.UUID] = mapped_column( Uuid, primary_key=True, default=uuid.uuid4, ) assignment_id: Mapped[uuid.UUID] = mapped_column( Uuid, ForeignKey("api_key_assignments.id", ondelete="CASCADE"), nullable=False, ) # SET NULL not CASCADE: the request row survives the # requester's hard-delete (extremely rare; soft-delete is the # norm) so compliance reports can still answer "what # self-service flows did this person initiate?". The column is # nullable in the DB so the FK SET NULL clause is valid SQL on # TiDB / MySQL ≥ 5.7 (NOT NULL + SET NULL is rejected with # ERROR 1830). The application layer enforces non-null on # insert — every code path that inserts an # ApiKeyReleaseRequestModel passes ``requester_id=current_user.id`` # in :meth:`ApiKeyService.submit_release_request`. requester_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"), nullable=True, ) # The manager / IT admin who decided. NULL while pending; set on # approve / reject. SET NULL on person delete so the request row # survives forensically even if the approver is later hard-deleted. approver_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"), nullable=True, ) # Enum (no DB ENUM — string column matches existing apikey pattern): # ``no_longer_needed`` | ``project_ended`` | ``wrong_provider`` | ``other``. # ``other`` requires ``reason_note`` (validated at schema layer). reason_code: Mapped[str] = mapped_column(String(32), nullable=False) reason_note: Mapped[str | None] = mapped_column(Text, nullable=True) # Status enum: ``pending`` | ``approved`` | ``rejected`` | ``cancelled``. # ``cancelled`` is requester-initiated withdrawal; ``rejected`` is # approver-initiated denial. status: Mapped[str] = mapped_column( String(24), nullable=False, server_default=text("'pending'"), ) # Free-text supplement from the approver on reject. Optional on # approve. Kept separate from ``reason_note`` (requester-supplied) # so an "approver decision" predicate scan does not have to # disambiguate which side wrote which note. decision_note: Mapped[str | None] = mapped_column(Text, nullable=True) # Stash of ``api_key_assignments.status`` at submit time, so a # reject / cancel restores the row to its pre-release state. Kept # as a column (not ``assignment.detail.release.previous_status``) # because the assignment-row JSON is touched by many other flows and # a structured column survives JSON-merge mishaps. Also indexed # implicitly via the ``(assignment_id, status)`` composite. previous_status: Mapped[str] = mapped_column(String(32), nullable=False) requested_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) responded_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True, ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now(), ) _WS_RUN_RE = re.compile(r"\s+") def normalize_provider_match_name(name: str | None) -> str: """Lowercase + collapse whitespace runs so substring matchers in the cloud auto-grant routers (Bedrock / Bailian / Azure OpenAI) survive admin-typed quirks like double spaces, NBSP (\\u00a0), full-width space (\\u3000), and tabs. Reason: AK-20260509-EF8A landed in prod with provider name ``"Azure OpenAI"`` (two spaces). The matcher's canonical token ``"azure openai"`` (one space) is not a substring of ``"azure openai"``, so ``maybe_auto_grant_azure_openai`` short- circuited at the first line and never minted the RG / AOAI account / chargeback rule. Python's ``\\s`` is Unicode-aware for str patterns, so this normalisation also covers NBSP / full-width space / tabs. """ if not name: return "" return _WS_RUN_RE.sub(" ", name.lower()).strip()