"""AWS Bedrock auto-grant — called from the API key Feishu callback.

When an ``ApiKeyAssignment`` for the "AWS Bedrock" provider transitions
to ``approved`` via the Feishu callback handler, we don't wait for an
IT admin to manually fill the key — we look up the primary cloud
integration whose name / slug matches the "AWS SG" account, mint a
dedicated IAM user with a scoped ``bedrock:InvokeModel`` policy, add it
to the standing ``DBaaS-DevAdmin-Role`` group (or whatever group the
admin has set via the ``bedrock.provisioning_iam_group`` platform_setting)
so the user inherits the org's baseline Bedrock permissions, and store
the returned access-key pair straight onto the assignment so the
employee can reveal it immediately.

The AWS SG tenant is resolved by slug (``aws-sg``) first, then by any
active cloud integration of provider ``aws`` whose name contains
"singapore" / "sg". Tenants can be swapped by renaming or by setting
the ``bedrock.provisioning_integration_slug`` platform_setting.

The IAM user is named after the *project* (sanitised), not the
employee — e.g. project "AI Support" yields
``bedrock-AI_Support-20260429-F446``. This keeps the AWS console /
billing tags organised by project so multiple keys for the same
project surface under a common prefix even when owned by different
employees. See ``_sanitize_user_name`` for the full rules.

Errors leave the assignment in ``approved`` status with a note in
``detail.bedrock`` so an admin can still fill the key manually.
"""
from __future__ import annotations

import logging
import uuid
from typing import Any

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.apikey.autogrant_alerts import record_autogrant_failure
from app.domains.apikey.models import (
    ApiKeyAssignment,
    ApiKeyProvider,
    normalize_provider_match_name,
)
from app.domains.apikey.service import _encrypt_key_value
from app.domains.cmdb.credential_crypto import decrypt_fields
from app.domains.cmdb.iam_upsert import upsert_local_iam_user
from app.domains.cmdb.models import CmdbCostAllocationRule
from app.domains.cmdb.providers.aws import AWSProvider
from app.domains.cmdb.providers.base import ProviderCredentials
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration, PlatformSetting
from app.notifications.service import NotificationEnvelope, NotificationService

logger = logging.getLogger(__name__)

# Inline IAM policy applied to every auto-granted Bedrock user. The
# region condition pins access to ap-southeast-1 (AWS SG) since that's
# where our Bedrock tenant lives — remove the condition if you need
# multi-region or move the tenant. Trim the Action list further if you
# ever want to restrict to a single model family (e.g.
# ``Resource: arn:aws:bedrock:*::foundation-model/anthropic.*``).
_BEDROCK_POLICY = {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "BedrockInvoke",
            "Effect": "Allow",
            "Action": [
                "bedrock:InvokeModel",
                "bedrock:InvokeModelWithResponseStream",
            ],
            # Scope to the ap-southeast-1 model ARN prefix so users
            # cannot invoke Bedrock models in other regions even if the
            # region condition is ever widened.
            "Resource": "arn:aws:bedrock:ap-southeast-1::foundation-model/*",
            "Condition": {
                "StringEquals": {"aws:RequestedRegion": "ap-southeast-1"},
            },
        },
        {
            "Sid": "BedrockList",
            "Effect": "Allow",
            "Action": [
                "bedrock:ListFoundationModels",
                "bedrock:GetFoundationModel",
            ],
            # List/Get are read-only and costless; leave on "*" so they
            # work regardless of model ARN.
            "Resource": "*",
        },
    ],
}


def _is_bedrock_provider(provider: ApiKeyProvider | None) -> bool:
    """Match against a whitespace-normalised provider name so admin
    typos like ``"AWS  Bedrock"`` (double space) or full-width / NBSP
    variants don't silently bypass the auto-grant — same bug class hit
    Azure OpenAI in prod (AK-20260509-EF8A)."""
    if provider is None:
        return False
    name = normalize_provider_match_name(provider.name)
    return "bedrock" in name


async def _resolve_integration(db: AsyncSession) -> Integration | None:
    """Find the AWS cloud integration to provision the IAM user in."""
    # Explicit override
    override = await db.get(
        PlatformSetting, "bedrock.provisioning_integration_slug",
    )
    if override is not None and override.value:
        slug = str(override.value).strip()
        row = (await db.execute(
            select(Integration).where(
                Integration.slug == slug,
                Integration.enabled.is_(True),
                Integration.kind == "cloud",
            )
        )).scalar_one_or_none()
        if row is not None:
            return row
    # Heuristic: AWS SG
    preferred_slugs = ("aws-sg", "aws_sg", "aws-singapore")
    for slug in preferred_slugs:
        row = (await db.execute(
            select(Integration).where(
                Integration.slug == slug,
                Integration.enabled.is_(True),
                Integration.kind == "cloud",
            )
        )).scalar_one_or_none()
        if row is not None:
            return row
    # Fallback: any enabled AWS cloud integration whose name contains
    # "sg" / "singapore". We deliberately do NOT fall back to "any
    # enabled AWS integration" — if the admin has multiple AWS
    # tenants (e.g. prod + research) the wrong one would silently
    # receive Bedrock grants. An unresolved tenant returns None so the
    # caller surfaces ``status='no-integration'`` and the admin sets
    # ``bedrock.provisioning_integration_slug`` explicitly.
    rows = (await db.execute(
        select(Integration).where(
            Integration.kind == "cloud",
            Integration.provider == "aws",
            Integration.enabled.is_(True),
        )
    )).scalars().all()
    for r in rows:
        name = (r.name or "").lower()
        if "sg" in name or "singapore" in name:
            return r
    return None


_IAM_NAME_MAX_LEN = 64
_IAM_NAME_PREFIX = "bedrock-"


def _sanitize_user_name(project_name: str, request_number: str) -> str:
    """Build the IAM username for a Bedrock auto-grant.

    Shape: ``bedrock-<slug>-<YYYYMMDD>-<XXXX>`` (max 64 chars; AWS IAM
    allows ``[A-Za-z0-9+=,.@_-]``). The slug is derived from the
    assignment's ``project_name`` so the AWS IAM console / billing tags
    are organised by project rather than by individual.

    Slug rules:
    - Whitespace runs collapse to a single ``_`` (``"AI  Support "`` →
      ``"AI_Support"``).
    - Characters outside ``[A-Za-z0-9._-]`` are dropped (covers CJK,
      emoji, ``/``, ``+``, …). We deliberately exclude ``+=,.@`` from
      the *valid* set even though IAM accepts them, because ``,`` and
      ``@`` look like field separators in the AWS console and ``+=``
      are URL-unfriendly when copy-pasted.
    - Leading/trailing ``._-`` are stripped.
    - Truncated to fit the 64-char IAM limit, leaving room for the
      ``bedrock-`` prefix and ``-<suffix>`` tail.
    - If sanitisation leaves nothing usable (e.g. an all-CJK project
      name), the slug is omitted entirely and the name falls back to
      ``bedrock-<suffix>``. The suffix alone keeps the name unique
      within the AWS account.

    The suffix is the FULL date+hex portion of the request_number
    (``AK-YYYYMMDD-XXXX`` → ``YYYYMMDD-XXXX``). Without the date, two
    requests whose 4-hex-char random tails collide (~1/65536) would
    target the same IAM username and the second ``iam.create_user``
    fails with ``EntityAlreadyExists``.
    """
    parts = request_number.split("-", 1)
    suffix = parts[1] if len(parts) > 1 else request_number

    # Build the slug. We iterate char-by-char so whitespace handling and
    # filtering happen in one pass. ``prev_underscore`` collapses runs of
    # whitespace / dropped chars so ``"A  B"`` becomes ``"A_B"`` rather
    # than ``"A__B"``.
    #
    # IAM accepts ``[A-Za-z0-9+=,.@_-]`` but we restrict the slug to
    # ASCII ``[A-Za-z0-9._-]`` only — Python's ``str.isalnum`` returns
    # True for CJK / Cyrillic / etc., and we don't want those reaching
    # AWS where they'd render as mojibake in the console.
    slug_chars: list[str] = []
    prev_underscore = False
    for ch in (project_name or ""):
        if ch.isspace():
            if not prev_underscore and slug_chars:
                slug_chars.append("_")
                prev_underscore = True
            continue
        if ("a" <= ch <= "z") or ("A" <= ch <= "Z") or ("0" <= ch <= "9") or ch in "._-":
            slug_chars.append(ch)
            prev_underscore = False
        # Other chars (CJK, emoji, punctuation) drop silently.
    slug = "".join(slug_chars).strip("._-")

    # Reserve room: prefix + slug + "-" + suffix. If suffix alone is too
    # long (only happens if the request-number generator changes), fall
    # back to a hard truncation.
    reserved = len(_IAM_NAME_PREFIX) + 1 + len(suffix)  # "bedrock-" + "-" + suffix
    if reserved >= _IAM_NAME_MAX_LEN:
        return f"{_IAM_NAME_PREFIX}{suffix}"[:_IAM_NAME_MAX_LEN]
    max_slug_len = _IAM_NAME_MAX_LEN - reserved
    slug = slug[:max_slug_len].rstrip("._-")

    if not slug:
        return f"{_IAM_NAME_PREFIX}{suffix}"
    return f"{_IAM_NAME_PREFIX}{slug}-{suffix}"


_DEFAULT_IAM_GROUP = "DBaaS-DevAdmin-Role"


async def _resolve_iam_group(db: AsyncSession) -> str:
    """Return the IAM group name to add new Bedrock users to.

    Defaults to ``DBaaS-DevAdmin-Role`` (the standing dev-admin group
    in the AWS SG account that grants baseline Bedrock + ancillary
    permissions). Admins can override per-environment by setting the
    ``bedrock.provisioning_iam_group`` platform_setting; an empty
    string in that setting disables group attachment entirely (e.g.
    if a future tenant prefers per-user policies only).
    """
    row = await db.get(PlatformSetting, "bedrock.provisioning_iam_group")
    if row is None:
        return _DEFAULT_IAM_GROUP
    raw = row.value
    if raw is None:
        return _DEFAULT_IAM_GROUP
    name = str(raw).strip()
    # Empty string is a deliberate "skip group attachment" signal — the
    # caller checks for this before calling ``add_user_to_group``.
    return name


async def maybe_auto_grant_bedrock(
    db: AsyncSession,
    assignment: ApiKeyAssignment,
) -> dict[str, Any]:
    """Invoked when an assignment becomes ``approved``.

    Safe no-op when the provider isn't Bedrock or when the auto-grant
    has already run (idempotent on ``detail.bedrock.granted`` OR
    ``detail.bedrock.in_flight``). Modifies the passed-in ``assignment``
    in place; caller commits.

    Leak-safety contract: before we touch AWS (which creates durable
    state), we commit a ``detail.bedrock.in_flight`` tombstone so a
    retry after a mid-callback crash sees it and refuses to re-mint.
    Admin can manually clear the tombstone from ``/admin/assets/api-keys`` if
    a recovery workflow needs one.
    """
    provider_row = await db.get(ApiKeyProvider, assignment.provider_id)
    if not _is_bedrock_provider(provider_row):
        return {"status": "not-bedrock"}

    detail = dict(assignment.detail or {})
    bedrock_info = dict(detail.get("bedrock") or {})
    if bedrock_info.get("granted"):
        return {"status": "already-granted"}
    if bedrock_info.get("in_flight"):
        # A previous callback started an AWS call but the Helix DB commit
        # failed. We cannot safely retry — the previous attempt may have
        # already created the IAM user on AWS. Surface for manual
        # reconciliation.
        return {"status": "in-flight-needs-manual-recovery"}

    integration = await _resolve_integration(db)
    if integration is None:
        bedrock_info["granted"] = False
        bedrock_info["error"] = (
            "No enabled AWS cloud integration found (looked for slug "
            "'aws-sg'/override). Create one under /admin/integrations."
        )
        detail["bedrock"] = bedrock_info
        assignment.detail = detail
        logger.warning(
            "Bedrock auto-grant skipped (no AWS integration) for %s",
            assignment.request_number,
        )
        await db.commit()
        await db.refresh(assignment)
        await record_autogrant_failure(
            db, assignment, kind='bedrock', stage='no-integration',
            error=str(bedrock_info.get('error') or 'no-integration'),
            provider_label=getattr(provider_row, 'name', None),
        )
        return {"status": "no-integration"}

    person = await db.get(IdentityPerson, assignment.assigned_to)
    email = (person.work_email or "") if person else ""
    user_name = _sanitize_user_name(
        assignment.project_name, assignment.request_number,
    )
    iam_group = await _resolve_iam_group(db)

    try:
        cred_fields = decrypt_fields(
            integration.secrets_encrypted, integration.secrets_nonce,
            integration_id=str(integration.id),
        ) if integration.secrets_encrypted else {}
    except Exception as exc:  # noqa: BLE001
        bedrock_info["granted"] = False
        bedrock_info["error"] = f"credential decryption failed: {exc}"
        detail["bedrock"] = bedrock_info
        assignment.detail = detail
        await db.commit()
        await db.refresh(assignment)
        await record_autogrant_failure(
            db, assignment, kind='bedrock', stage='decrypt-failed',
            error=str(bedrock_info.get('error') or 'decrypt-failed'),
            provider_label=getattr(provider_row, 'name', None),
        )
        return {"status": "decrypt-failed"}

    merged = {
        k: str(v) for k, v in (integration.config_public or {}).items() if v is not None
    }
    merged.update(cred_fields)
    provider = AWSProvider(ProviderCredentials(
        provider_id=str(integration.id),
        provider_type="aws",
        account_alias=integration.name,
        fields=merged,
    ))
    # Commit a tombstone so a retry after a mid-callback crash won't
    # mint a second IAM user. Admin can clear ``detail.bedrock.in_flight``
    # from /admin/assets/api-keys after manually verifying AWS state if recovery
    # is needed.
    bedrock_info["in_flight"] = True
    bedrock_info["user_name_attempted"] = user_name
    bedrock_info["integration_slug"] = integration.slug
    bedrock_info["iam_group_attempted"] = iam_group or None
    detail["bedrock"] = bedrock_info
    assignment.detail = detail
    await db.commit()
    await db.refresh(assignment)
    detail = dict(assignment.detail or {})
    bedrock_info = dict(detail.get("bedrock") or {})
    # Empty group name (set deliberately via platform_setting) disables
    # group attachment for this tenant; otherwise pass a single-element
    # list so the AWS provider attaches the user before minting the key.
    groups = [iam_group] if iam_group else []
    try:
        # ``Owner`` is the AWS cost-allocation convention. PingCAP has
        # activated ``Owner`` as an IAM principal tag for cost
        # allocation in the AWS SG account, so Cost Explorer groups
        # Bedrock spend by ``iamPrincipal/Owner`` (NOT
        # ``aws:PrincipalTag/Owner`` — that's the IAM-policy evaluator
        # namespace and silently returns empty values when used as a
        # CE GroupBy key). We tag the IAM user itself with the
        # applicant's work email so each employee's Bedrock invocations
        # land in their own row of the ``iamPrincipal/Owner`` breakdown.
        # Earlier revisions tagged ``Owner=<project>`` which collapsed
        # multiple applicants on the same project into a single bucket
        # — losing per-employee attribution and breaking the matching
        # ``CmdbCostAllocationRule`` (the rule keys on the per-applicant
        # tag value, not project).
        # ``helix-project-name`` stays as a separate tag for product-
        # line analytics; ``helix-owner-email`` is preserved for
        # forensics symmetry with Azure OpenAI auto-grant.
        project = (assignment.project_name or "").strip()[:256]
        tags: dict[str, str] = {
            "helix-assignment-id": str(assignment.id),
            "helix-request-number": assignment.request_number,
            "helix-owner-email": email or "unknown",
            "helix-project-name": project,
        }
        # Falls back to the IAM username when the applicant has no
        # work_email on file (extremely rare — HRIS sync sets it).
        # Empty Owner would land cost in the "(unallocated)" bucket
        # which defeats the per-employee chargeback we're after.
        owner_tag_value = email.strip() if email else user_name
        if owner_tag_value:
            tags["Owner"] = owner_tag_value
        result = await provider.create_iam_user_with_policy(
            user_name=user_name,
            policy_name="bedrock-invoke",
            policy_document=_BEDROCK_POLICY,
            tags=tags,
            group_names=groups,
        )
    except Exception as exc:  # noqa: BLE001
        logger.exception("Bedrock IAM user creation failed")
        bedrock_info["granted"] = False
        bedrock_info["error"] = str(exc)[:500]
        bedrock_info["user_name_attempted"] = user_name
        bedrock_info["integration_slug"] = integration.slug
        bedrock_info["iam_group_attempted"] = iam_group or None
        detail["bedrock"] = bedrock_info
        assignment.detail = detail
        await db.commit()
        await db.refresh(assignment)
        await record_autogrant_failure(
            db, assignment, kind='bedrock', stage='aws-failed',
            error=str(bedrock_info.get('error') or 'aws-failed'),
            provider_label=getattr(provider_row, 'name', None),
        )
        return {"status": "aws-failed"}
    finally:
        await provider.close()

    # Store AK/SK as the key_value using a structured format — the
    # reveal endpoint returns the raw string, so by convention Bedrock
    # keys encode AK + SK separated by a newline. Frontend renders
    # them as two copy-to-clipboard lines.
    ak = result.get("access_key_id") or ""
    sk = result.get("secret_access_key") or ""
    # Round-20 audit (A2F6): bind AES-GCM AAD to the owning row's id so
    # an attacker with DB write capability can't swap one assignment's
    # ciphertext blob into another row and have ``_decrypt_key_value``
    # round-trip it. Mirror the pattern Azure OpenAI + service.fill_key
    # already use (see _encrypt_key_value docstring in service.py).
    assignment.key_value = _encrypt_key_value(
        f"ACCESS_KEY_ID={ak}\nSECRET_ACCESS_KEY={sk}\n"
        f"REGION=ap-southeast-1\nUSER_ARN={result.get('user_arn') or ''}",
        assignment_id=assignment.id,
    )
    assignment.status = "active"

    # Populate the structured IAM-binding columns (added in migration
    # 072). Up to this point they were maintained only via the
    # ``detail.bedrock`` JSON, which made the admin "rebind" UI awkward
    # — now the columns are first-class.
    assignment.iam_user_arn = result.get("user_arn")
    assignment.iam_integration_id = str(integration.id)

    # Upsert the just-minted IAM user into ``cmdb_iam_users`` so the
    # CMDB IAM 用户 page reflects it immediately rather than waiting
    # for the next scheduled IAM sync (default cadence: hourly).
    # Idempotent against the sync job's UQ key
    # ``(integration_id, user_name, sub_account_id)`` so a subsequent
    # sync collapses cleanly onto this row instead of inserting a
    # duplicate. Tagged ``__source=helix_autogrant.bedrock`` so admins
    # auditing the IAM page can tell at a glance which rows came from
    # Helix-driven autogrant vs. cloud-side admin actions.
    iam_row = await upsert_local_iam_user(
        db,
        integration_id=str(integration.id),
        user_name=user_name,
        arn=result.get("user_arn"),
        user_id_external=result.get("user_id"),
        attached_policies=[
            {"PolicyName": "bedrock-invoke", "PolicyType": "Inline"},
        ],
        groups=[iam_group] if iam_group else [],
        tags={
            "__source": "helix_autogrant.bedrock",
            "helix-assignment-id": str(assignment.id),
            "helix-request-number": assignment.request_number,
            "helix-owner-email": email or "",
            "helix-project-name": (assignment.project_name or "")[:256],
        },
        raw_data={
            "UserName": user_name,
            "Arn": result.get("user_arn"),
        },
        owner_person_id=(
            str(assignment.assigned_to) if assignment.assigned_to else None
        ),
    )
    assignment.iam_user_id = iam_row.id

    # Auto-write a cost-allocation rule so the cost-ingest cron
    # stamps every Bedrock fact carrying ``tags.Owner=<email>`` with
    # the applicant as cost owner. Priority 10 puts per-applicant
    # rules ahead of any catch-all (default rules sit at priority
    # 100+). Mirrors the Azure OpenAI auto-grant pattern.
    #
    # Idempotency: a previous successful run for this assignment
    # would have committed a rule with the same name. The Feishu
    # callback retries on transient failures, but the durable commit
    # below also flips ``granted=True`` — and the function short-
    # circuits with ``status="already-granted"`` on subsequent calls,
    # so we never reach this block twice for the same assignment.
    # Defensive pre-check still — if an admin manually re-runs the
    # autogrant via /admin/assets/api-keys after clearing ``granted``, the
    # SELECT keeps the unique-by-name invariant.
    rule_id: str | None = None
    if owner_tag_value and assignment.assigned_to:
        rule_name = (
            f"AWS Bedrock · {assignment.request_number} · "
            f"{owner_tag_value}"
        )[:255]
        existing_rule = (await db.execute(
            select(CmdbCostAllocationRule).where(
                CmdbCostAllocationRule.name == rule_name,
            )
        )).scalar_one_or_none()
        if existing_rule is None:
            rule = CmdbCostAllocationRule(
                id=str(uuid.uuid4()),
                name=rule_name,
                priority=10,
                match_integration_ids=[str(integration.id)],
                match_provider="aws",
                match_resource_group=None,
                match_service=None,
                match_tag_key="Owner",
                match_tag_value=owner_tag_value,
                assign_person_id=str(assignment.assigned_to),
                # Department / cost-center deliberately left NULL —
                # the cost-allocation pipeline re-resolves them from
                # the assigned person at ingest time and stamps them
                # onto ``cmdb_cost_facts``. Baking them into the rule
                # row would fall behind on a cost-center / department
                # change for the same applicant.
                assign_department=None,
                assign_cost_center=None,
                enabled=True,
                description=(
                    f"Auto-generated by helix_autogrant.bedrock for "
                    f"assignment {assignment.id}. Charges every AWS "
                    f"line item tagged ``Owner={owner_tag_value}`` "
                    f"on integration {integration.slug or integration.id} "
                    f"to the API key applicant."
                ),
                created_by_id=None,
                updated_by_id=None,
            )
            db.add(rule)
            rule_id = rule.id
        else:
            rule_id = str(existing_rule.id)

    bedrock_info.update({
        "granted": True,
        "in_flight": False,
        "integration_slug": integration.slug,
        "iam_user_name": user_name,
        "user_arn": result.get("user_arn"),
        "region": "ap-southeast-1",
        "policy_name": "bedrock-invoke",
        "iam_group": iam_group or None,
        "owner_tag_value": owner_tag_value or None,
        "allocation_rule_id": rule_id,
    })
    detail["bedrock"] = bedrock_info
    assignment.detail = detail

    # Durability: commit key_value + granted=True NOW so a subsequent
    # rollback in the outer Feishu-callback transaction cannot leave us
    # with an orphan AWS IAM user (created + access-key minted) but no
    # assignment record. Without this commit, on caller rollback the
    # ``in_flight=true`` tombstone persists and the employee sees
    # "in-flight-needs-manual-recovery" forever while AWS has a live
    # credential pair no one can reveal.
    try:
        await db.commit()
        await db.refresh(assignment)
    except Exception:  # noqa: BLE001
        # Commit failed — best we can do is log a reconciliation hint.
        # The AWS-side user + key are live; ops must either delete them
        # or manually reconcile via the SQL snippet in cmdb-ops.md §3.8.
        logger.exception(
            "Bedrock auto-grant: AWS user created (%s, arn=%s) but DB "
            "commit failed. Manual reconciliation required — see "
            "cmdb-ops.md in-flight recovery section.",
            user_name, result.get("user_arn"),
        )
        raise

    # Notify the assignee that the credential is ready. Done after the
    # main commit so a notification-channel failure (Feishu rate-limit /
    # SMTP down) cannot roll back the durable AWS+DB state. Dispatch is
    # durable via the notification_outbox (retry worker picks up partial /
    # failed sends), so we don't block the autogrant return on the
    # channel result. Skipped when the assignee row has been deleted
    # between approval and callback — extremely rare.
    if person is not None:
        provider_label = (
            provider_row.name if provider_row is not None else "AWS Bedrock"
        )
        try:
            await NotificationService(db).dispatch(
                NotificationEnvelope(
                    recipient=person,
                    title="[Helix] Bedrock API Key 已就绪 / API key ready",
                    body_text=(
                        f"Hi {person.display_name or person.work_email or ''},\n\n"
                        f"你申请的 {provider_label} API Key (申请号 "
                        f"{assignment.request_number},项目 "
                        f"{assignment.project_name or '-'}) 已自动开通。\n\n"
                        f"请到 Helix 的「我的 API Keys」页面查看凭证和使用方法:\n"
                        f"  /me/api-keys\n\n"
                        f"这是一对 AWS IAM AccessKey/Secret(AKIA 开头),"
                        f"已绑定 ``bedrock:InvokeModel`` 内联策略,"
                        f"地区固定为 ap-southeast-1。详细使用手册见:"
                        f"/help#bedrock-employee\n\n"
                        f"— Helix IT 平台\n"
                        f"---\n"
                        f"Hi {person.display_name or person.work_email or ''},\n\n"
                        f"Your {provider_label} API key (request "
                        f"{assignment.request_number}, project "
                        f"{assignment.project_name or '-'}) is now active.\n\n"
                        f"View the credential and usage notes on your "
                        f"Helix \"My API Keys\" page:\n"
                        f"  /me/api-keys\n\n"
                        f"This is an AWS IAM AccessKey/Secret pair "
                        f"(AKIA...), bound to a ``bedrock:InvokeModel`` "
                        f"inline policy and pinned to ap-southeast-1. "
                        f"See /help#bedrock-employee for the SDK setup.\n\n"
                        f"— Helix IT Platform"
                    ),
                    template_key="apikey.ready",
                    context={
                        "person_name": (
                            person.display_name or person.work_email or ""
                        ),
                        "person_email": person.work_email or "",
                        "provider_name": provider_label,
                        "request_number": assignment.request_number,
                        "project_name": assignment.project_name or "",
                        "iam_user_name": user_name,
                        "portal_url": "/me/api-keys",
                        "help_anchor": "/help#bedrock-employee",
                    },
                    url="/me/api-keys",
                    related_resource_type="api_key_assignment",
                    related_resource_id=assignment.id,
                ),
            )
        except Exception:  # noqa: BLE001 — never fail the grant on notify error
            logger.exception(
                "Bedrock auto-grant: notification dispatch failed for "
                "assignment %s; credential remains active and is visible "
                "on /me/api-keys.",
                assignment.request_number,
            )

    logger.info(
        "Bedrock auto-grant succeeded: assignment=%s iam_user=%s",
        assignment.id, user_name,
    )
    return {
        "status": "granted",
        "iam_user_name": user_name,
        "user_arn": result.get("user_arn"),
    }


__all__ = ["maybe_auto_grant_bedrock"]