"""Jira (Atlassian Cloud) software-provisioning driver.

Helix maps the company's Atlassian workspace to several
SoftwareLicensePlans — Jira Software, Jira Service Management, and (per
migration 206) two Confluence subscriptions (tidb + pingcap-us) — each
backed by its own ``software_connectors`` binding pointing at this driver.

Two distinct concerns, deliberately separated:

  * **Seat COUNT** is taken from the Atlassian Org Admin API
    ``product_access`` (who really holds the license), scoped per binding
    by ``product_key`` + ``site_urls`` — see
    ``app.integrations.atlassian_activity`` and the ``jira.seat_sync``
    handler. Group membership is NOT the seat count (a group can grant a
    product to people who hold it via other groups too, and some groups —
    notably ``jira-servicedesk-users`` — grant nothing at all).
  * **grant / revoke** still operates on the binding's ``group_name``,
    which must be a *conferring* group for the product (verified via
    ``/rest/api/3/applicationrole``). Granting invites the user (if
    needed) and adds them to the group; revoking removes them (account
    preserved by default so issue history stays attributed). JSM's
    conferring default group is ``jira-servicemanagement-users-pingcap-ticket``
    — NOT ``jira-servicedesk-users``.

The same driver instance serves multiple bindings — the binding config is
what differentiates them.

Atlassian's REST endpoints split across two hosts; this driver speaks
both, and **each host needs its own credential** (this is not optional —
they are different token types from different generation pages):

  * ``https://<site>.atlassian.net/rest/api/3/`` — site-level user +
    group management. Auth: HTTP Basic ``admin_email:api_token`` where
    ``api_token`` is a **user-level** token from
    ``id.atlassian.com/manage-profile/security/api-tokens``. All seat-
    management calls land here. Always required.
  * ``https://api.atlassian.com/admin/v1/orgs/{org_id}/`` — optional
    org-level lifecycle calls. Only used when ``org_id`` AND
    ``org_admin_api_key`` are configured AND
    ``action_on_offboard='delete'`` is requested. Auth: Bearer
    ``org_admin_api_key`` where the key is generated at
    ``admin.atlassian.com/o/{orgId}/api-keys`` (Atlassian Guard /
    Access plans only). The user ``api_token`` does NOT work here —
    org admin keys and user API tokens are separate credentials.

Reference
    https://developer.atlassian.com/cloud/jira/platform/rest/v3/intro/
    https://developer.atlassian.com/cloud/admin/organization/rest/intro/
"""
from __future__ import annotations

import asyncio
import base64
import logging
import re
from typing import Any
from urllib.parse import quote as _url_quote

import httpx

from app.integrations.driver_base import (
    ProvisioningContext,
    ProvisioningResult,
    SoftwareProvisioningDriver,
)
from app.integrations.registry import register_driver
from app.integrations.schemas import (
    IntegrationFieldInfo,
    IntegrationTestResult,
)

logger = logging.getLogger(__name__)

# ── Error tags ───────────────────────────────────────────────────
#
# Mirrors the Zoom / Google / Feishu tag convention. Pairs with
# entries in ``app.domains.provisioning.service._PROVISIONING_REMEDIATION_HINTS``.
JIRA_TAG_ORG_ADMIN_AUTH_FAILED = "JIRA_ORG_ADMIN_AUTH_FAILED"
JIRA_TAG_ORG_ADMIN_PERMISSION_DENIED = "JIRA_ORG_ADMIN_PERMISSION_DENIED"

# ── Protected org-wide identity groups ───────────────────────────
#
# Some Atlassian products (notably Confluence at PingCAP) are licensed
# *org-wide*: the conferring set includes company-wide identity groups
# that EVERY employee belongs to (``employee_in_china``,
# ``All users from G Suite``). These groups are mirrored from the IdP
# directory via SCIM / Atlassian "default groups", so removing a single
# user from them to reclaim a product seat is (a) actively harmful — it
# strips a core directory group — and (b) futile: the next SCIM sign-in
# sync re-adds the user, the product re-appears in ``product_access``,
# and ``jira.recently_revoked_audit`` flags an endless "bounce" that
# pages IT every week (the 2026-06-15 incident: 13 active employees,
# auto re-revoked ≥3× then escalated, recurring weekly — fixed here).
#
# Rule: a product conferred *only* via a protected identity group is an
# **org-wide entitlement**, NOT individually revocable. ``revoke_product_access``
# refuses to touch these groups and returns ``not_individually_revocable``
# so the seat-reclaim / audit pipeline treats it as terminal (no retry,
# no alert) instead of churning forever. The defaults below are merged
# with the admin-configurable
# ``software_seat.product_access_revoke.protected_identity_groups``
# platform_setting (injected into ``ctx.extra``), so a site with
# differently-named "everyone" groups can extend the list without a
# code change. Match is case-insensitive on the trimmed group name.
DEFAULT_PROTECTED_IDENTITY_GROUPS: frozenset[str] = frozenset({
    "employee_in_china",
    "all users from g suite",
})


def _normalize_group_name(name: str | None) -> str:
    """Case-insensitive, whitespace-trimmed key for group-name compares."""
    return (name or "").strip().casefold()


def _redact_secrets(blob: str) -> str:
    """Strip Atlassian / generic Authorization headers from error
    snippets before they land in ``provisioning_tasks.error_message``
    or ``raw_response.remove_access_error``. Mirrors the Slack
    driver's ``_redact_secrets`` (Round-3 security scan MEDIUM).

    Atlassian error responses occasionally echo the request's Bearer
    header (especially 401 / "invalid token" responses). Without this
    scrub, an admin reading the failure-DM card would see a partial
    or full org_admin_api_key in their Feishu chat history.
    """
    if not blob:
        return blob
    # Authorization: Bearer xxx → "Authorization: Bearer ***"
    blob = re.sub(
        r"(?i)(authorization\s*:?\s*(?:bearer|basic)\s+)[A-Za-z0-9._\-+/=]+",
        r"\1***",
        blob,
    )
    # Atlassian Cloud API tokens (ATCTT prefix on org-admin keys).
    blob = re.sub(r"\bATCTT[A-Za-z0-9_\-]{10,}\b", "ATCTT***", blob)
    return blob


def _redact_jira_body(text: str | None, *, max_len: int = 300) -> str:
    """Strip Authorization headers + Atlassian token shapes from a Jira
    response body before it lands in an error message.

    Round-49 — Atlassian's proxies occasionally echo the inbound request
    envelope on auth failures. With Basic auth that's
    ``Authorization: Basic <base64(email:api_token)>`` — decoding yields
    the user's API token. Strip the credential material before any
    ``error_message`` interpolation lands the body in
    ``provisioning_tasks.error_message`` (which surfaces in admin DMs
    via ``_dispatch_revoke_failure_dms``).

    Complement to ``_redact_secrets``: ``_redact_secrets`` is the
    pre-existing helper used at lifecycle/raw_response sites; this
    one is used at the Round-49 error-message sites and also enforces
    a hard ``max_len`` cap so a multi-KB Atlassian error page can't
    bloat the audit row.
    """
    if not text:
        return ""
    import re as _re
    redacted = _re.sub(
        r"(?i)Authorization\s*:\s*Basic\s+[A-Za-z0-9+/=]+",
        "Authorization: Basic [REDACTED]",
        text,
    )
    # Atlassian user-API-token shapes
    redacted = _re.sub(
        r"\bATATT[A-Za-z0-9_\-]{20,}\b",
        "[REDACTED-ATATT-token]",
        redacted,
    )
    redacted = _re.sub(
        r"\bATCTT[A-Za-z0-9_\-]{20,}\b",
        "[REDACTED-ATCTT-token]",
        redacted,
    )
    if len(redacted) > max_len:
        return redacted[:max_len] + "..."
    return redacted

_HTTP_TIMEOUT = httpx.Timeout(30.0, connect=10.0)
# Hard cap on pagination iterations — see slack_software.py for the
# rationale.
_MAX_PAGINATOR_PAGES = 200
_ADMIN_API_BASE = "https://api.atlassian.com/admin/v1"

# Transient upstream statuses on the Atlassian Admin API. The org-wide
# ``/orgs/{orgId}/users`` roster pull is the heaviest call we make (the
# whole managed-account directory in one cursor walk), and Atlassian's
# gateway sporadically 502/503/504s on large orgs and 429s under its
# per-org rate limit. These are retryable — a single transient blip
# must not fail the whole ``jira.seat_sync`` / idle-reclaim cron, which
# would otherwise surface an empty roster and make the idle job think
# nobody is active. 401/403/404 and other 4xx are NOT here: those are
# terminal config/permission errors where retrying only wastes time.
_TRANSIENT_HTTP_STATUSES: frozenset[int] = frozenset({429, 502, 503, 504})
_MAX_TRANSIENT_RETRIES = 4
_TRANSIENT_BACKOFF_BASE_SECONDS = 1.0
_TRANSIENT_BACKOFF_MAX_SECONDS = 20.0


def _transient_backoff_seconds(
    resp: httpx.Response | None,
    attempt: int,
) -> float:
    """Delay before the next retry. Honour a server-supplied
    ``Retry-After`` (seconds form, as Atlassian sends on 429) when
    present and sane, otherwise capped exponential backoff keyed on the
    zero-based ``attempt`` number.
    """
    if resp is not None:
        raw = resp.headers.get("Retry-After")
        if raw:
            try:
                secs = float(raw)
            except ValueError:
                secs = -1.0
            if secs >= 0:
                return min(secs, _TRANSIENT_BACKOFF_MAX_SECONDS)
    return float(min(
        _TRANSIENT_BACKOFF_BASE_SECONDS * (2 ** attempt),
        _TRANSIENT_BACKOFF_MAX_SECONDS,
    ))


@register_driver
class JiraSoftwareDriver(SoftwareProvisioningDriver):
    provider = "jira"
    display_name = "Jira (Atlassian)"
    description = (
        "Atlassian Cloud Jira / Confluence 工作区。坐席 = 某个 product 组的"
        "成员资格。grant 时调用 /rest/api/3/user/search 找用户、不存在则 POST "
        "/rest/api/3/user 邀请,然后 POST /rest/api/3/group/user 加入绑定指定"
        "的 group;revoke 时 DELETE /rest/api/3/group/user 移出 group(默认"
        "保留 Atlassian 账号,issue 历史不丢)。一个 Jira integration 可挂多"
        "个 SoftwareLicensePlan(每个产品/group 一个),每个 plan 自己的"
        "connector binding 通过 group_name 区分。"
    )
    supports_seat = True
    supports_permission = False
    supports_multi_instance = True
    # Offboarding revokes a leaver's ENTIRE Jira footprint by removing
    # them from every group they belong to (not just one bound product
    # group, and never by deactivating the account — Helix policy keeps
    # the managed account so issue history stays attributed). Enabled here
    # so the provisioning ``account`` dispatch routes leaver tasks to
    # :meth:`revoke_all_access`.
    supports_full_access_revoke = True
    # Active employees self-cancelling a seat (or admin-cancel on behalf)
    # route through :meth:`revoke_product_access` — intersect the user's
    # current groups with the product's ``/rest/api/3/applicationrole``
    # ``groups`` (conferring groups) and DELETE that intersection. Keeps
    # identity groups (``employee_in_china`` / ``All users from G Suite``)
    # and unrelated product accesses (Confluence when only cancelling
    # Jira) intact, unlike :meth:`revoke_all_access`. The single-group
    # :meth:`revoke_seat` it replaces was insufficient: at PingCAP a
    # Jira-software license is conferred by ``jira-software-users`` AND
    # team groups like ``jira-dev-engineers`` / ``RD Leader``; removing
    # from only the bound group still left ``product_access`` populated
    # so the next ``jira.seat_sync`` cron re-imported the allocation
    # (the loop that stuck shengbo.ma / sunxiaoguang).
    supports_product_access_revoke = True
    config_schema = [
        IntegrationFieldInfo(
            key="site_url", label="Jira 站点 URL",
            is_secret=False, required=True,
            placeholder="tidb.atlassian.net",
            hint=(
                "去掉 https:// 和末尾斜杠;只填主机名。所有 /rest/api/3/ "
                "调用都会拼到这个域名上。"
            ),
        ),
        IntegrationFieldInfo(
            key="admin_email", label="管理员邮箱",
            is_secret=False, required=True,
            placeholder="it-admin@yourcompany.com",
            hint=(
                "API Token 绑定的 Atlassian 账号邮箱。Atlassian REST API "
                "用 Basic email:token 认证,缺这个会 401。该用户必须是 "
                "Jira 站点 admin / 组管理员才能改组成员。"
            ),
        ),
        IntegrationFieldInfo(
            key="api_token", label="用户 API Token",
            is_secret=True, required=True,
            placeholder="ATATT... 或 ATCTT...",
            hint=(
                "**必须在 id.atlassian.com/manage-profile/security/api-tokens "
                "生成**(以 admin_email 这个账号登录 → Create API token)。"
                "Helix 用 Basic auth (admin_email:api_token) 走 site REST API。"
                "❌ 不要用 admin.atlassian.com/o/{orgId}/api-keys 那个页面"
                "生成的 org admin API key — 那是另一种 token,只能 Bearer auth "
                "走 api.atlassian.com/admin/v1/,在这里会 401。"
            ),
        ),
        IntegrationFieldInfo(
            key="org_id", label="Organization ID (可选)",
            is_secret=False, required=False,
            placeholder="7b9308d0-...-0927k2c082a5",
            hint=(
                "Atlassian Organization 的 UUID(从 admin.atlassian.com URL "
                "里取)。仅在离职策略选 delete + 想走 Atlassian Admin API "
                "彻底停用账号时需要;只走 group 移除操作可以留空。"
                "必须配合 org_admin_api_key 一起填。"
            ),
        ),
        IntegrationFieldInfo(
            key="org_admin_api_key", label="Org Admin API Key — 只读 (可选)",
            is_secret=True, required=False,
            placeholder="ATCTT... (admin.atlassian.com,可带 scope)",
            hint=(
                "**在 admin.atlassian.com/o/{orgId}/api-keys 生成**的 org "
                "API key,Bearer auth 走 api.atlassian.com/admin/v1/。Helix "
                "用它做**只读**离职对账(拉 org 托管账号名册 + product_access)。"
                "带 `read:accounts:admin` 等 granular scope 的 key 就够 —— "
                "最小权限,日常同步用这把。撤离职访问的写操作另用 "
                "org_unscoped_api_key(见下)。需要 Atlassian Guard 订阅。"
            ),
        ),
        IntegrationFieldInfo(
            key="org_unscoped_api_key", label="Org Unscoped API Key — 撤访问写 (可选)",
            is_secret=True, required=False,
            placeholder="ATCTT... (创建时一个 scope 都不要勾)",
            hint=(
                "在 admin.atlassian.com/o/{orgId}/api-keys 新建、**创建时不选"
                "任何 scope** 的 org API key(Atlassian:『无 scope 的 key 可"
                "访问全部 Admin API』)。撤离职访问的端点("
                "directory/users 的 remove-access / suspend-access)**不在 "
                "granular scope 体系里**,带 scope 的 key 会 401 "
                "INSUFFICIENT_SCOPE —— 必须用这把无 scope 的全权 key。Helix "
                "只在『撤离职残留访问』这一处用它(action_on_offboard=delete "
                "及非 group 残留自动清理),日常只读仍走 org_admin_api_key。"
                "⚠️ 全 org 管理员权限,务必视为高敏密钥。"
            ),
        ),
        IntegrationFieldInfo(
            key="suspend_account_on_offboard",
            label="离职时同时停用 Atlassian 登录", is_secret=False,
            required=False, default="true",
            options=[
                {"value": "true",
                 "label": "是 —— 移出所有 group 后再停用登录 (推荐)"},
                {"value": "false",
                 "label": "否 —— 只移出 group,账号仍可登录"},
            ],
            hint=(
                "开启后,离职撤权在移出所有 group、撤掉组织级产品访问之外,"
                "再调 directory 的 suspend-access **停用该账号的登录**"
                "(需要 org_id + org_unscoped_api_key)。"
                "**不会删除账号** —— issue 历史照旧归属,`/restore-access` "
                "随时可恢复。\n"
                "为什么默认开:把离职者账号置为 inactive 的原本一直是组织侧"
                "的身份目录同步(Guard 的 Google Workspace / SCIM 用户供给),"
                "那条链路 Helix 既不拥有也看不见 —— 2026-08-13 它被一次设置页"
                "保存静默关掉、16 天无人发现,期间所有离职者账号全都活着。"
                "开启后同步降级为冗余而不是唯一路径,并且覆盖了同步永远够不到"
                "的那类人(公司邮箱被保留、Google 账号没被删的离职者)。"
            ),
        ),
    ]
    binding_schema = [
        IntegrationFieldInfo(
            key="group_name", label="Jira 产品 Group (grant/revoke 用)",
            is_secret=False, required=True,
            placeholder="jira-software-users-pingcap-ticket",
            hint=(
                "grant/revoke 时把账号加入 / 移出的 Atlassian group。**必须是该"
                "产品的 conferring group**(/rest/api/3/applicationrole 的 "
                "groups/defaultGroups 列表里的)。坑:Jira Software 用 "
                "jira-software-users(或默认组 jira-software-users-pingcap-ticket);"
                "JSM 必须用 jira-servicemanagement-users-pingcap-ticket —— "
                "**不是 jira-servicedesk-users**(后者根本不发 JSM 许可,往里加人"
                "不会占座);Confluence 用 confluence-users。坐席『统计』口径见 "
                "product_key / site_urls(走 product_access,与本组成员数无关)。"
            ),
        ),
        IntegrationFieldInfo(
            key="action_on_offboard", label="离职动作",
            is_secret=False, required=True,
            default="deactivate",
            options=[
                {"value": "deactivate", "label": "移出 group (保留账号,推荐)"},
                {"value": "delete",
                 "label": "额外撤销组织级产品访问 (需 org_id + org_unscoped_api_key)"},
            ],
            hint=(
                "deactivate 把人从这一个 group 里移出 — issue 历史保留、可继续"
                "被分配。delete 在此之外再调 Atlassian Admin API 的 "
                "`DELETE directory/users` 撤掉该账号的**组织级产品访问**"
                "(Trello 工作区、组织默认访问、直接授权这些 group 清不掉的),"
                "需要 integration 同时配置 org_id + org_unscoped_api_key + "
                "Atlassian Guard 订阅。"
                "⚠️ 两个选项都**不会停用 Atlassian 账号** —— Helix 的既定策略"
                "是永远保留账号以维持 issue 历史归属(见 revoke_all_access 的"
                "注释)。真正把离职者账号置为 inactive 的是组织侧身份目录同步"
                "(Guard 的 Google Workspace / SCIM 用户供给),不是 Helix。"
                "此前 delete 的标签写作『停用 Atlassian 账号』,与代码实际行为"
                "不符,已按实际行为更正。"
            ),
        ),
        IntegrationFieldInfo(
            key="product_key", label="产品 (坐席统计口径)",
            is_secret=False, required=False,
            options=[
                {"value": "jira-software", "label": "Jira Software"},
                {"value": "jira-servicedesk", "label": "Jira Service Management"},
                {"value": "confluence", "label": "Confluence"},
            ],
            hint=(
                "该 plan 统计哪一个 Atlassian 产品的 license。坐席数 = Org "
                "Admin API product_access 里 key=该产品 且在下面 site_urls 站点"
                "上、account 为 active 的真实持有者(不是 group_name 的成员数 —"
                "组成员 ≠ 许可)。**留空 = 该 plan 不参与坐席统计**(用于让 "
                "Tempo / ScriptRunner / eazyBI 等 Marketplace 插件目录复用同一 "
                "Jira connector 却不被当成 Jira Software 坐席误计)。"
            ),
        ),
        IntegrationFieldInfo(
            key="site_urls", label="付费生产站点 (逗号分隔)",
            is_secret=False, required=False,
            placeholder="tidb.atlassian.net",
            hint=(
                "该 plan 统计哪些付费生产站点(逗号分隔主机名)。Jira / JSM 填 "
                "tidb.atlassian.net;Confluence 两个独立订阅分别填 "
                "tidb.atlassian.net 与 pingcap-us.atlassian.net(各建一个 plan)。"
                "用于把 sandbox / 外部个人站点排除在坐席统计之外。留空则用该产品"
                "的默认生产站点白名单。grant/revoke 仍走 integration 配的 "
                "site_url + group_name。"
            ),
        ),
        IntegrationFieldInfo(
            key="seat_revocable", label="可逐人回收 (闲置自动回收开关)",
            is_secret=False, required=False,
            default="true",
            options=[
                {"value": "true",
                 "label": "是 — 真实产品 plan,可按人回收 (默认)"},
                {"value": "false",
                 "label": "否 — 分摊计费插件 / tier 授权,整体跳过闲置回收"},
            ],
            hint=(
                "仅影响『闲置坐席自动回收』(software_seat.idle_reclaim)。"
                "true = 该 plan 的绑定组就是这个产品的 conferring group,移除"
                "某人 = 真正收回该产品(Jira Software / JSM / Confluence 选 "
                "true)。false = 该 plan 是按 Jira 用户分摊计费的 Marketplace "
                "插件(Tempo / ScriptRunner / eazyBI),复用 jira-software-users "
                "组只为统计 —— 对它做逐人回收会把人从 Jira 本体踢掉,所以闲置"
                "回收整体跳过(计入 skipped)。**坐席统计(jira.seat_sync)不看"
                "此字段**,仍按 product_key 走 product_access。留空 = true。"
            ),
        ),
    ]

    # ── Config accessors ──────────────────────────────────────────

    async def _site_url(self) -> str:
        v = (self.config.get("site_url") or "").strip()
        if not v:
            raise ValueError("Jira integration missing site_url")
        # Tolerate users pasting a full URL or one with a path/query —
        # we only want the host. Strip scheme first, then split on '/'
        # (kills any trailing path) and on '?' (kills any query string).
        v = v.removeprefix("https://").removeprefix("http://")
        v = v.split("/", 1)[0].split("?", 1)[0].strip()
        if not v:
            raise ValueError(
                "Jira integration site_url is empty after normalisation",
            )
        return v

    async def _site_transport(self):
        """Pin the admin-configured Jira origin for one HTTP client."""
        from app.integrations._url_safety import safe_async_http_transport
        return await safe_async_http_transport(
            f"https://{await self._site_url()}",
            allowed_suffixes=("atlassian.net",),
        )

    def _admin_email(self) -> str:
        v = (self.config.get("admin_email") or "").strip()
        if not v:
            raise ValueError("Jira integration missing admin_email")
        return v

    def _api_token(self) -> str:
        v = self.config.get("api_token")
        if not v:
            raise ValueError("Jira integration missing api_token")
        return str(v)

    def _org_id(self) -> str | None:
        v = (self.config.get("org_id") or "").strip()
        return v or None

    def _org_admin_api_key(self) -> str | None:
        v = self.config.get("org_admin_api_key")
        if not v:
            return None
        s = str(v).strip()
        return s or None

    def _org_unscoped_api_key(self) -> str | None:
        """Unscoped org API key — the ONLY credential that can call the
        user-access write endpoints (remove-access / suspend-access).
        Those endpoints are not covered by Atlassian's granular scope
        system, so a scoped key 401s INSUFFICIENT_SCOPE on them
        regardless of which scopes it holds. This is a distinct config
        field from ``org_admin_api_key`` (the scoped read key) with no
        fallback between them — see :meth:`_unscoped_bearer_headers`."""
        v = self.config.get("org_unscoped_api_key")
        if not v:
            return None
        s = str(v).strip()
        return s or None

    def _suspend_on_offboard_enabled(self) -> bool:
        """Whether a leaver revoke also suspends the Atlassian sign-in.

        Defaults to **on**. The historical default (group removal only)
        is what left every leaver with a live, loginable account whenever
        the org-side directory sync hiccupped, so "on" is the safe
        default and turning it off is the decision that needs a
        deliberate act. Accepts the usual falsy spellings so an operator
        who types ``no`` gets what they meant.
        """
        raw = self.config.get("suspend_account_on_offboard")
        if raw is None or str(raw).strip() == "":
            return True
        return str(raw).strip().lower() not in ("false", "0", "no", "off")

    def _basic_headers(self) -> dict[str, str]:
        token = base64.b64encode(
            f"{self._admin_email()}:{self._api_token()}".encode(),
        ).decode()
        return {
            "Authorization": f"Basic {token}",
            "Accept": "application/json",
            "Content-Type": "application/json",
        }

    def _admin_bearer_headers(self) -> dict[str, str]:
        """Bearer-auth headers for ``api.atlassian.com/admin/v1/``.

        The org admin API key is a different credential from the
        site-level user API token used in :meth:`_basic_headers`.
        Caller must check :meth:`_org_admin_api_key` is not ``None``
        before invoking this — we raise loudly rather than fall back
        to the user API token (which would 401 on the admin API).
        """
        org_key = self._org_admin_api_key()
        if not org_key:
            raise ValueError(
                "Jira integration missing org_admin_api_key — required "
                "for org Admin API calls (action_on_offboard=delete). "
                "Generate one at admin.atlassian.com/o/{orgId}/api-keys.",
            )
        return {
            "Authorization": f"Bearer {org_key}",
            "Accept": "application/json",
            "Content-Type": "application/json",
        }

    def _unscoped_bearer_headers(self) -> dict[str, str]:
        """Bearer-auth headers for the user-access WRITE endpoints using
        the unscoped org API key.

        Uses ``org_unscoped_api_key`` ONLY — deliberately no fallback to
        ``org_admin_api_key``. The two keys are not interchangeable for
        writes and their values are indistinguishable, so a fallback
        would let a *scoped* admin key silently drive these endpoints and
        401 INSUFFICIENT_SCOPE — the exact footgun this split exists to
        prevent. Keeping the check to this one field means the audit
        gate, the offboarding precondition, and this header builder all
        agree on a single credential."""
        key = self._org_unscoped_api_key()
        if not key:
            raise ValueError(
                "Jira integration missing org_unscoped_api_key — required "
                "to revoke org-level user access (remove-access / "
                "suspend-access). Create an API key with NO scopes at "
                "admin.atlassian.com/o/{orgId}/api-keys (scoped keys 401 "
                "on these endpoints).",
            )
        return {
            "Authorization": f"Bearer {key}",
            "Accept": "application/json",
            "Content-Type": "application/json",
        }

    async def remove_org_user_access(self, account_id: str) -> dict[str, Any]:
        """Remove ALL of a managed account's product access org-wide.

        ``DELETE /admin/v1/orgs/{orgId}/directory/users/{accountId}`` —
        the authoritative way to strip access that group removal can't
        touch (Trello workspace membership, org default access, direct
        product grants). Keeps the Atlassian account itself, so issue
        history stays attributed (Helix policy) — it only revokes access.

        Requires ``org_id`` + an UNSCOPED org API key (see
        :meth:`_unscoped_bearer_headers`). Idempotent: a 404 means the
        account is already gone / has no managed access — treated as
        success so callers can close the blocker.

        Returns a dict ``{ok, status, removed, already_absent, error}``
        rather than raising on HTTP status, so the audit can aggregate
        per-leaver outcomes without try/except around each call.
        """
        org_id = self._org_id()
        if not org_id:
            raise ValueError(
                "Jira integration missing org_id — required to remove "
                "org-level user access via the Atlassian Admin API.",
            )
        url = (
            f"{_ADMIN_API_BASE}/orgs/{_url_quote(org_id, safe='')}"
            f"/directory/users/{_url_quote(str(account_id), safe='')}"
        )
        try:
            async with httpx.AsyncClient(
                timeout=_HTTP_TIMEOUT, trust_env=False,
            ) as client:
                resp = await self._request_with_retry(
                    client, "DELETE", url,
                    headers=self._unscoped_bearer_headers(),
                    what="Jira remove-org-access",
                )
        except httpx.HTTPError as e:
            return {
                "ok": False, "status": None, "removed": False,
                "already_absent": False,
                "error": f"{type(e).__name__}: {e}",
            }
        if resp.status_code in (200, 204):
            logger.info(
                "jira remove-org-access: accountId=%s status=%s",
                account_id, resp.status_code,
            )
            return {
                "ok": True, "status": resp.status_code, "removed": True,
                "already_absent": False, "error": None,
            }
        if resp.status_code == 404:
            # Disambiguate the two 404s. A *route* 404 ("Request failed
            # to match any route") means a wrong/changed endpoint or
            # org_id — that's the silent-false-positive class this PR
            # killed for lifecycle/disable, so treat it as a HARD error,
            # never "already absent". A *resource* 404 (this account has
            # no managed access) is the legitimate idempotent success.
            body_l = (resp.text or "").lower()
            if "match any route" in body_l or "no route" in body_l:
                logger.error(
                    "jira remove-org-access: ROUTE 404 for accountId=%s "
                    "— endpoint/org_id likely wrong, NOT treating as "
                    "already-absent: %s",
                    account_id, _redact_secrets(resp.text[:200]),
                )
                return {
                    "ok": False, "status": 404, "removed": False,
                    "already_absent": False,
                    "error": _redact_secrets(resp.text[:300]),
                }
            return {
                "ok": True, "status": 404, "removed": False,
                "already_absent": True, "error": None,
            }
        # 401/403 = key is scoped (not unscoped) / revoked / lacks org
        # admin; anything else = unexpected. Surface, don't mask.
        return {
            "ok": False, "status": resp.status_code, "removed": False,
            "already_absent": False,
            "error": _redact_secrets(resp.text[:300]),
        }

    async def set_org_user_access_suspended(
        self, account_id: str, *, suspended: bool,
    ) -> dict[str, Any]:
        """Suspend (or restore) a managed account's ability to sign in.

        ``POST /admin/v1/orgs/{orgId}/directory/users/{accountId}/suspend-access``
        and its ``/restore-access`` twin. This is the lever that produces
        ``account_status: inactive`` — the same end state the org's IdP
        directory sync produces when the upstream Google account goes
        away.

        **Why this does not contradict "Helix never deactivates the
        account".** That rule is about *deletion*: an Atlassian account
        must survive so issue history stays attributed to a real person
        rather than collapsing to "former user". Suspending access does
        not delete anything — the account, its authored issues and its
        comment history are untouched, and ``/restore-access`` puts it
        back. What it removes is exactly what offboarding is supposed to
        remove: the ability to log in.

        **Why Helix does it at all**, rather than leaving it to the IdP
        sync as before: a chain Helix does not own cannot be the only
        thing standing between a leaver and a live account. On
        2026-08-13 that chain was switched off by a single settings-page
        save and stayed off for 16 days without a sound. With this call
        in the offboarding path the sync becomes redundancy instead of
        the sole mechanism, and it also covers the leavers the sync can
        never reach — the ones whose upstream Google account is
        deliberately retained.

        Requires ``org_id`` + the UNSCOPED org API key (same credential
        and the same scope caveat as :meth:`remove_org_user_access`: the
        directory user-access endpoints are outside Atlassian's granular
        scope system, so a scoped key 401s here).

        Returns ``{ok, status, changed, already_absent, error}`` instead
        of raising, so a sweep can aggregate per-leaver outcomes without
        wrapping every call. A 404 is idempotent success in the suspend
        direction (no managed account = cannot sign in), but a *route*
        404 is a hard error — the same silent-false-positive class that
        made the old ``lifecycle/disable`` call look like it worked for
        months while doing nothing.
        """
        org_id = self._org_id()
        if not org_id:
            raise ValueError(
                "Jira integration missing org_id — required to suspend "
                "org-level access via the Atlassian Admin API.",
            )
        if not self._org_unscoped_api_key():
            return {
                "ok": False, "status": None, "changed": False,
                "already_absent": False,
                "error": (
                    "org_unscoped_api_key not configured — the directory "
                    "suspend-access / restore-access endpoints are not in "
                    "Atlassian's granular scope system and 401 with a "
                    "scoped key."
                ),
            }
        verb = "suspend-access" if suspended else "restore-access"
        url = (
            f"{_ADMIN_API_BASE}/orgs/{_url_quote(org_id, safe='')}"
            f"/directory/users/{_url_quote(str(account_id), safe='')}/{verb}"
        )
        try:
            async with httpx.AsyncClient(
                timeout=_HTTP_TIMEOUT, trust_env=False,
            ) as client:
                resp = await self._request_with_retry(
                    client, "POST", url,
                    headers=self._unscoped_bearer_headers(),
                    what=f"Jira {verb}",
                )
        except httpx.HTTPError as e:
            return {
                "ok": False, "status": None, "changed": False,
                "already_absent": False,
                "error": f"{type(e).__name__}: {e}",
            }
        if resp.status_code in (200, 201, 204):
            logger.info(
                "jira %s: accountId=%s status=%s",
                verb, account_id, resp.status_code,
            )
            return {
                "ok": True, "status": resp.status_code, "changed": True,
                "already_absent": False, "error": None,
            }
        if resp.status_code == 404:
            body_l = (resp.text or "").lower()
            if "match any route" in body_l or "no route" in body_l:
                logger.error(
                    "jira %s: ROUTE 404 for accountId=%s — endpoint/org_id "
                    "likely wrong, NOT treating as already-absent: %s",
                    verb, account_id, _redact_secrets(resp.text[:200]),
                )
                return {
                    "ok": False, "status": 404, "changed": False,
                    "already_absent": False,
                    "error": _redact_secrets(resp.text[:300]),
                }
            # Resource 404: no managed account under this org. In the
            # suspend direction the desired end state (cannot sign in)
            # already holds. In the RESTORE direction it does not — there
            # is nothing to restore, and silently calling that success
            # would tell an operator their un-suspend worked when the
            # account is simply not there.
            return {
                "ok": bool(suspended), "status": 404, "changed": False,
                "already_absent": True,
                "error": None if suspended else (
                    "no managed account under this org — nothing to restore"
                ),
            }
        return {
            "ok": False, "status": resp.status_code, "changed": False,
            "already_absent": False,
            "error": _redact_secrets(resp.text[:300]),
        }

    async def list_org_events(
        self, *, action: str, limit: int = 10,
    ) -> list[dict[str, Any]]:
        """Newest-first org audit events for ONE action type.

        ``GET /admin/v1/orgs/{orgId}/events?action=<name>``. Reads fine
        with the ordinary **scoped** ``org_admin_api_key`` — no unscoped
        credential needed, because this is the read side.

        Filtering by ``action`` is not an optimisation, it is the only
        workable form: the unfiltered feed of this org is dominated by
        per-issue and per-tool events, so paging it to reach a
        once-per-4-hours record burns dozens of requests and still may
        not get there. (The documented ``from``/``to`` window params were
        observed NOT to filter reliably — ``to`` applies, ``from`` is
        ignored — so they are not used here.)

        The action names this platform cares about:
          ``scim_auto_sync_complete``  the real sync cadence — its newest
              record IS the heartbeat.
          ``scim_sync_disabled`` / ``scim_sync_enabled``  who turned user
              provisioning off, and when.
          ``scim_user_delete`` / ``scim_user_create``  what a given sync
              actually did (the console's own created/deleted counters
              were observed to report 0 while the audit showed 4 creates
              and 6 deletes — trust these, not those).

        Returns the raw ``attributes`` dicts. Never raises on HTTP
        status: a heartbeat sweep that dies on a 500 is a heartbeat that
        reports nothing, so failures come back as an empty list and the
        caller records ``unknown`` rather than a false ``ok``.
        """
        org_id = self._org_id()
        key = self._org_admin_api_key()
        if not org_id or not key:
            return []
        url = (
            f"{_ADMIN_API_BASE}/orgs/{_url_quote(org_id, safe='')}/events"
            f"?limit={max(1, min(int(limit), 100))}"
            f"&action={_url_quote(action, safe='')}"
        )
        try:
            async with httpx.AsyncClient(
                timeout=_HTTP_TIMEOUT, trust_env=False,
            ) as client:
                resp = await self._request_with_retry(
                    client, "GET", url,
                    headers=self._admin_bearer_headers(),
                    what=f"Jira org events ({action})",
                )
        except httpx.HTTPError as e:
            logger.warning(
                "jira list_org_events(%s) transport error: %s: %s",
                action, type(e).__name__, e,
            )
            return []
        if resp.status_code != 200:
            logger.warning(
                "jira list_org_events(%s) HTTP %s: %s",
                action, resp.status_code, _redact_secrets(resp.text[:200]),
            )
            return []
        try:
            body = resp.json()
        except ValueError:
            return []
        out: list[dict[str, Any]] = []
        for item in (body.get("data") or []):
            if isinstance(item, dict) and isinstance(
                item.get("attributes"), dict,
            ):
                out.append(item["attributes"])
        return out

    async def read_org_user_state(self, email: str) -> dict[str, Any] | None:
        """Read one managed account's org-level state by email.

        Read-back for the write above, and the per-person refresh the
        assurance page uses. Returns the normalised
        ``{account_id, account_status, product_access, access_billable}``
        or ``None`` when the org directory has no account for that email.

        ⚠️ **This reads the same aggregated org listing as
        ``list_org_users_with_activity``, which Atlassian refreshes with a
        lag of up to ~24h.** A suspend that returned 204 can therefore
        still read back ``active`` for a while. Callers must treat that as
        *unverified*, never as failure — recording it as failure would
        turn a working write into a permanent red row, and retrying the
        write would hammer the API for nothing. The daily audit re-reads
        and settles it.
        """
        target = (email or "").strip().lower()
        if not target:
            return None
        for raw in await self.list_org_users_with_activity():
            if not isinstance(raw, dict):
                continue
            raw_attrs = raw.get("attributes")
            attrs: dict[str, Any] = (
                raw_attrs if isinstance(raw_attrs, dict) else {}
            )

            def _pick(
                *keys: str,
                _r: dict[str, Any] = raw,
                _a: dict[str, Any] = attrs,
            ) -> Any:
                for k in keys:
                    if _r.get(k) is not None:
                        return _r.get(k)
                    if _a.get(k) is not None:
                        return _a.get(k)
                return None

            if str(_pick("email") or "").strip().lower() != target:
                continue
            pa = _pick("product_access") or []
            return {
                "account_id": _pick("account_id", "id"),
                "account_status": _pick("account_status", "status"),
                "product_access": pa if isinstance(pa, list) else [],
                "access_billable": _pick("access_billable"),
            }
        return None

    async def _site_base(self) -> str:
        return f"https://{await self._site_url()}/rest/api/3"

    # ── test_connection ──────────────────────────────────────────

    async def test_connection(self) -> IntegrationTestResult:
        """Probe ``GET /rest/api/3/myself`` — the cheapest authenticated
        call Atlassian exposes. Confirms the token + email pair works
        and surfaces the resolved account so the admin can sanity-check.
        """
        try:
            async with httpx.AsyncClient(
                timeout=_HTTP_TIMEOUT,
                trust_env=False,
                transport=await self._site_transport(),
            ) as client:
                resp = await self._request_with_retry(
                    client, "GET", f"{await self._site_base()}/myself",
                    headers=self._basic_headers(), what="Jira /myself",
                )
        except httpx.HTTPError as e:
            return IntegrationTestResult(
                ok=False,
                message=f"Jira /myself failed: {type(e).__name__}: {e}",
            )
        except Exception as e:  # noqa: BLE001 — config errors flow here
            return IntegrationTestResult(
                ok=False,
                message=f"Jira /myself failed: {type(e).__name__}: {e}",
            )
        if resp.status_code == 401:
            return IntegrationTestResult(
                ok=False,
                message=(
                    "Jira 401 Unauthorized — Basic auth 失败。最常见的原因: "
                    "你把 admin.atlassian.com/o/{orgId}/api-keys 那个页面"
                    "生成的 **org admin API key** 当 api_token 填进来了 — "
                    "那种 token 只能走 api.atlassian.com/admin/v1/ 的 Bearer "
                    "auth,不能走 site REST API。请用 admin_email 这个账号"
                    "登录 https://id.atlassian.com/manage-profile/security/"
                    "api-tokens 重新生成一个**用户级** API token,把它填到 "
                    "api_token 字段。原来那个 org admin key 如果你需要走离职 "
                    "delete 路径,可以填到可选字段 org_admin_api_key 里。"
                ),
            )
        if resp.status_code == 403:
            return IntegrationTestResult(
                ok=False,
                message=(
                    "Jira 403 Forbidden — 该账号无 Jira 站点访问权限。请确"
                    "认它是站点成员且持有 Site administrator 角色。"
                ),
            )
        if resp.status_code != 200:
            return IntegrationTestResult(
                ok=False,
                message=f"Jira /myself HTTP {resp.status_code}: {_redact_jira_body(resp.text, max_len=200)}",
            )
        data = resp.json() or {}
        account_id = data.get("accountId") or ""
        display = data.get("displayName") or ""
        email = data.get("emailAddress") or ""
        return IntegrationTestResult(
            ok=True,
            message=f"OK · {display} <{email}> · accountId={account_id[:16]}…",
            detail={
                "account_id": account_id,
                "display_name": display,
                "email": email,
                "site_url": await self._site_url(),
            },
        )

    @staticmethod
    def _is_already_member_response(resp: httpx.Response) -> bool:
        """Detect Atlassian's "user is already a member of …" 400.

        Atlassian's REST API surfaces this either in plain ``resp.text``
        or wrapped in JSON ``{"errorMessages": [...]}`` depending on
        product version. We check both, case-insensitive.
        """
        body = resp.text or ""
        if "already a member" in body.lower():
            return True
        ctype = resp.headers.get("content-type", "")
        if not ctype.startswith("application/json"):
            return False
        try:
            data = resp.json()
        except Exception:  # noqa: BLE001 — malformed JSON, treat as no
            return False
        msgs = (data or {}).get("errorMessages") or []
        return any("already a member" in str(m).lower() for m in msgs)

    @staticmethod
    def _is_not_a_member_response(resp: httpx.Response) -> bool:
        """Detect Atlassian's "user is not a member of …" 400 on revoke.

        Inverse of ``_is_already_member_response``. Atlassian returns
        ``HTTP 400`` with body ``Cannot remove user from group 'X'
        since user is not a member of 'X'`` (text or JSON
        ``errorMessages``) when DELETE /group/user targets a user who
        is not in the group. The desired end-state is already met, so
        revoke_seat treats this the same as 404 (idempotent
        already-absent) instead of bubbling a failure that keeps the
        offboarding blocker open.
        """
        body = resp.text or ""
        if "not a member" in body.lower():
            return True
        ctype = resp.headers.get("content-type", "")
        if not ctype.startswith("application/json"):
            return False
        try:
            data = resp.json()
        except Exception:  # noqa: BLE001 — malformed JSON, treat as no
            return False
        msgs = (data or {}).get("errorMessages") or []
        return any("not a member" in str(m).lower() for m in msgs)

    # ── User lookup ───────────────────────────────────────────────

    async def find_user_by_email(self, email: str) -> dict[str, Any] | None:
        """Return the Atlassian user record for ``email`` or ``None``.

        Uses ``/rest/api/3/user/search?query=…`` with cursor pagination.
        Atlassian's user-search is eventually-consistent and the target
        user may appear on a page beyond the first when the email prefix
        matches multiple accounts (e.g. shared display-name prefix).
        Without pagination the revoke path would declare the user absent
        and return ok=True while the account remains live.
        Inactive accounts are returned so callers can decide whether to
        reactivate or treat as already-absent.
        """
        email_norm = (email or "").strip().lower()
        if not email_norm:
            return None
        site_base = await self._site_base()
        start_at = 0
        page_size = 50
        for _page in range(_MAX_PAGINATOR_PAGES):
            try:
                async with httpx.AsyncClient(
                    timeout=_HTTP_TIMEOUT,
                    trust_env=False,
                    transport=await self._site_transport(),
                ) as client:
                    resp = await self._request_with_retry(
                        client, "GET", f"{site_base}/user/search",
                        headers=self._basic_headers(),
                        params={
                            "query": email_norm,
                            "startAt": start_at,
                            "maxResults": page_size,
                        },
                        what="Jira user/search",
                    )
            except httpx.HTTPError as e:
                raise RuntimeError(
                    f"Jira user/search failed: {type(e).__name__}: {e}",
                ) from e
            if resp.status_code != 200:
                raise RuntimeError(
                    f"Jira user/search HTTP {resp.status_code}: {_redact_jira_body(resp.text, max_len=200)}",
                )
            results = resp.json() or []
            if not results:
                break
            for u in results:
                if not isinstance(u, dict):
                    continue
                if (u.get("emailAddress") or "").lower() == email_norm:
                    return u
            if len(results) < page_size:
                break
            start_at += page_size
        return None

    async def get_approximate_license_count(
        self, product_key: str,
    ) -> int | None:
        """Atlassian's authoritative per-product license user count.

        Calls ``GET /rest/api/3/license/approximateLicenseCount/product/{key}``
        (Basic auth on the paid Jira site) which Atlassian documents as
        "the approximate number of users assigned a license for the
        product on this site". This is the single best public-API match
        for the number Atlassian admin's Apps / Subscriptions page
        displays — supports ``jira-software`` / ``jira-servicedesk`` /
        ``jira-core`` / ``jira-product-discovery``. **Atlassian does NOT
        expose a public endpoint for Confluence's license count** —
        callers must fall back to product_access counting for confluence.

        Prod 2026-05-30 numbers from this endpoint (tidb.atlassian.net):

          * ``jira-software`` → 367   (Atlassian Apps page: 372 Tempo Std)
          * ``jira-servicedesk`` → 73 (Atlassian Apps page: 71 JSM Prem)

        Both within 1–5 of Atlassian's own Admin UI — closer than what
        product_access counting alone (`360` / `70`) achieves, because
        this endpoint reflects Atlassian's internal license_assigned
        accounting rather than externally-visible product_access. The
        remaining 1–5 gap is Atlassian's own subsystem drift (the same
        number from /admin and /rest reports differs by a few) and is
        irreducible from the customer side.

        Returns the integer count on 200 OK, or ``None`` for any error
        (404 for unsupported product, network error, malformed body).
        Caller decides the fallback strategy — typically subtract
        ``holders_off_paid_site`` from the org-wide ``len(all_holders)``
        for confluence and unsupported families.
        """
        key = (product_key or "").strip().lower()
        if not key:
            return None
        site_base = await self._site_base()
        url = (
            f"{site_base}/license/approximateLicenseCount/product/"
            f"{_url_quote(key, safe='')}"
        )
        try:
            async with httpx.AsyncClient(
                timeout=_HTTP_TIMEOUT,
                trust_env=False,
                transport=await self._site_transport(),
            ) as client:
                resp = await self._request_with_retry(
                    client, "GET", url, headers=self._basic_headers(),
                    what="Jira approximateLicenseCount",
                )
        except httpx.HTTPError:
            return None
        if resp.status_code != 200:
            return None
        try:
            body = resp.json() or {}
        except (ValueError, TypeError):
            return None
        raw = body.get("value")
        if raw is None:
            return None
        try:
            return int(str(raw).strip())
        except (TypeError, ValueError):
            return None

    async def list_group_members(
        self, group_name: str, *, include_inactive: bool = False,
    ) -> list[dict[str, Any]]:
        """Return every account in ``group_name``.

        Uses ``/rest/api/3/group/member`` with cursor pagination
        (``startAt`` + ``isLast``). Active members only by default —
        Atlassian's ``includeInactiveUsers`` toggle pulls deactivated
        rows in too, useful for drift reconciliation but noisy for
        seat sync (a deactivated user is no longer "occupying" a seat
        in the license sense).
        """
        out: list[dict[str, Any]] = []
        start = 0
        page_size = 50
        # Max-page cap (P1 from CR2). 200 pages × 50/page = 10k members
        # — comfortable headroom for any realistic group.
        for _page in range(_MAX_PAGINATOR_PAGES):
            params: dict[str, Any] = {
                "groupname": group_name,
                "startAt": start,
                "maxResults": page_size,
                "includeInactiveUsers": str(include_inactive).lower(),
            }
            try:
                async with httpx.AsyncClient(
                    timeout=_HTTP_TIMEOUT,
                    trust_env=False,
                    transport=await self._site_transport(),
                ) as client:
                    resp = await self._request_with_retry(
                        client, "GET",
                        f"{await self._site_base()}/group/member",
                        headers=self._basic_headers(),
                        params=params,
                        what="Jira group/member",
                    )
            except httpx.HTTPError as e:
                raise RuntimeError(
                    f"Jira group/member failed: {type(e).__name__}: {e}",
                ) from e
            if resp.status_code == 404:
                # Group doesn't exist — distinguish from empty group.
                raise RuntimeError(
                    f"Jira group {group_name!r} not found (HTTP 404). "
                    "Check binding's group_name or create the group "
                    "in Jira admin first.",
                )
            if resp.status_code != 200:
                raise RuntimeError(
                    f"Jira group/member HTTP {resp.status_code}: "
                    f"{_redact_jira_body(resp.text, max_len=200)}",
                )
            data = resp.json() or {}
            values = data.get("values") or []
            for u in values:
                if isinstance(u, dict):
                    out.append(u)
            if data.get("isLast", False) or len(values) < page_size:
                break
            start += page_size
        else:
            raise RuntimeError(
                f"Jira group/member paginator did not terminate after "
                f"{_MAX_PAGINATOR_PAGES} pages for group {group_name!r}.",
            )
        return out

    # ── Transient-failure retry layer ─────────────────────────────

    async def _request_with_retry(
        self,
        client: httpx.AsyncClient,
        method: str,
        url: str,
        *,
        what: str,
        **kwargs: Any,
    ) -> httpx.Response:
        """Transparent retry wrapper around ``client.request`` for every
        Atlassian call this driver makes.

        Retries transient upstream failures — 429 / 502 / 503 / 504 and
        client-side connect/read timeouts (``httpx.TransportError``) —
        with capped exponential backoff, honouring ``Retry-After`` on 429.
        Then it hands control back to the caller **unchanged**:

          * a terminal response (2xx, or any 4xx other than 429) returns
            immediately;
          * a transient HTTP status that persists past the retry budget
            is *returned as-is* — the caller's existing status handling
            reacts to it (so 401/403/404 messaging, "already member",
            per-leaver aggregation, etc. all keep working);
          * a transport error that persists past the budget is *re-raised
            as the original* ``httpx`` exception — the caller's existing
            ``except httpx.HTTPError`` clause reacts to it.

        Safe for reads (idempotent), for group/access removals (idempotent
        — re-DELETE of an already-removed member returns 404 / not-a-member,
        both handled), and for the two mutating POSTs in ``grant_seat``
        (user invite + add-to-group): each already treats an "already
        done" response (409 / "already exists" / already-member) as
        success, so a retry after a 504 whose first attempt actually
        landed converges instead of duplicating.
        """
        last_exc: httpx.HTTPError | None = None
        resp: httpx.Response | None = None
        for attempt in range(_MAX_TRANSIENT_RETRIES + 1):
            try:
                resp = await client.request(method, url, **kwargs)
            except httpx.HTTPError as e:
                if not isinstance(e, httpx.TransportError):
                    # e.g. TooManyRedirects / DecodingError — not a
                    # transient network blip, so propagate now (the
                    # caller's own httpx.HTTPError handler sees it).
                    raise
                # Timeout / connect reset / protocol error — retryable.
                last_exc, resp = e, None
            else:
                last_exc = None
                if resp.status_code not in _TRANSIENT_HTTP_STATUSES:
                    return resp
            if attempt >= _MAX_TRANSIENT_RETRIES:
                break
            delay = _transient_backoff_seconds(resp, attempt)
            logger.warning(
                "%s transient failure (%s); retrying in %.1fs "
                "(attempt %d/%d)",
                what,
                (
                    f"HTTP {resp.status_code}"
                    if resp is not None
                    else type(last_exc).__name__
                ),
                delay,
                attempt + 1,
                _MAX_TRANSIENT_RETRIES,
            )
            await asyncio.sleep(delay)
        if resp is not None:
            # Transient status persisted — let the caller classify it.
            logger.warning(
                "%s still failing (HTTP %s) after %d retries; "
                "returning to caller for terminal handling.",
                what, resp.status_code, _MAX_TRANSIENT_RETRIES,
            )
            return resp
        # Transport error persisted — re-raise the original httpx error.
        assert last_exc is not None
        raise last_exc

    # ── Org Admin API: directory + activity ───────────────────────

    async def list_org_users_with_activity(self) -> list[dict[str, Any]]:
        """Return every managed account in the Atlassian org, with its
        ``last_active`` + per-product ``product_access`` activity.

        Pulls ``GET /admin/v1/orgs/{orgId}/users`` (Bearer auth via the
        org admin API key) using Atlassian's ``data`` + ``links.next``
        cursor pagination. Each item carries (shape can be flat or
        nested under ``attributes`` depending on API version):

          * ``account_id`` / ``id``
          * ``email`` (managed-account email)
          * ``account_status`` — ``active`` / ``inactive`` / ``closed``
          * ``last_active`` — org-wide last-seen, ISO-8601 (nanosecond
            precision UTC, e.g. ``2026-05-26T14:24:25.994292581Z``)
          * ``product_access`` — list of ``{key, name, url, last_active}``
            entries, one per (product × site). ``url`` is the site host;
            ``last_active`` is the per-site-per-product last-seen.
          * ``access_billable`` / ``roles`` (org-admin / site-admin)

        This is the directory the idle-seat-reclaim job (and any future
        license-utilisation report) cross-references against Helix
        seat allocations. Requires ``org_id`` AND ``org_admin_api_key``
        (Atlassian Guard / Access) — raises ``RuntimeError`` naming the
        missing field otherwise so the caller can surface a config gap
        instead of silently returning an empty roster (which would make
        the idle job think nobody is active).
        """
        org_id = self._org_id()
        if not org_id:
            raise RuntimeError(
                "Jira org_id not configured — required for the Atlassian "
                "Admin API user-activity pull (the Organization UUID from "
                "admin.atlassian.com). Set it on the jira integration row.",
            )
        # Raises ValueError if org_admin_api_key is missing; normalise to
        # RuntimeError so the seat-sync-style error collectors treat it
        # uniformly.
        try:
            headers = self._admin_bearer_headers()
        except ValueError as e:
            raise RuntimeError(str(e)) from e

        out: list[dict[str, Any]] = []
        next_url: str | None = (
            f"{_ADMIN_API_BASE}/orgs/{_url_quote(org_id, safe='')}/users"
        )
        async with httpx.AsyncClient(timeout=_HTTP_TIMEOUT, trust_env=False) as client:
            for _page in range(_MAX_PAGINATOR_PAGES):
                if not next_url:
                    break
                resp = await self._request_with_retry(
                    client, "GET", next_url,
                    headers=headers, what="Jira admin /users",
                )
                if resp.status_code in (401, 403):
                    raise RuntimeError(
                        f"Jira admin /users HTTP {resp.status_code} — the "
                        "org_admin_api_key lacks permission or has been "
                        "rotated/revoked (Atlassian Guard / Access plan "
                        "required for the Org Admin API). "
                        f"{_redact_jira_body(resp.text, max_len=200)}",
                    )
                if resp.status_code != 200:
                    raise RuntimeError(
                        f"Jira admin /users HTTP {resp.status_code}: "
                        f"{_redact_jira_body(resp.text, max_len=200)}",
                    )
                body = resp.json() or {}
                data = body.get("data") or []
                for u in data:
                    if isinstance(u, dict):
                        out.append(u)
                links = body.get("links") or {}
                nxt = links.get("next")
                next_url = str(nxt) if nxt else None
                # Only ever follow a pagination cursor that stays on the
                # Atlassian admin host — we send a Bearer org-admin key on
                # every hop, and blindly following an off-host ``next``
                # (if the API response were ever tampered/misconfigured)
                # would leak that credential to an arbitrary origin.
                if next_url and not next_url.startswith(
                    "https://api.atlassian.com/",
                ):
                    raise RuntimeError(
                        "Jira admin /users pagination returned an off-host "
                        "`next` link; refusing to follow it with the org "
                        "admin bearer token.",
                    )
            else:
                raise RuntimeError(
                    "Jira admin /users paginator did not terminate after "
                    f"{_MAX_PAGINATOR_PAGES} pages.",
                )
        return out

    async def get_user_groups(self, account_id: str) -> list[str]:
        """Return the names of every group ``account_id`` belongs to on
        the production site (``/rest/api/3/user/groups``, Basic auth).

        Used by the idle-reclaim job's multi-group safety check: removing
        a user from the binding's product group is only a *real* reclaim
        if no other group on the site also grants the same product.
        Atlassian returns the full group list in one response (no
        pagination), each entry ``{name, groupId}``. A 404 (account
        unknown to this site) yields an empty list — the caller treats
        that as "no overlapping groups here".
        """
        account = (account_id or "").strip()
        if not account:
            return []
        site_base = await self._site_base()
        try:
            async with httpx.AsyncClient(
                timeout=_HTTP_TIMEOUT,
                trust_env=False,
                transport=await self._site_transport(),
            ) as client:
                resp = await self._request_with_retry(
                    client, "GET", f"{site_base}/user/groups",
                    headers=self._basic_headers(),
                    params={"accountId": account},
                    what="Jira user/groups",
                )
        except httpx.HTTPError as e:
            raise RuntimeError(
                f"Jira user/groups failed: {type(e).__name__}: {e}",
            ) from e
        if resp.status_code == 404:
            return []
        if resp.status_code != 200:
            raise RuntimeError(
                f"Jira user/groups HTTP {resp.status_code}: "
                f"{_redact_jira_body(resp.text, max_len=200)}",
            )
        data = resp.json() or []
        names: list[str] = []
        for g in data:
            if not isinstance(g, dict):
                continue
            name = (g.get("name") or "").strip()
            if name:
                names.append(name)
        return names

    async def list_conferring_groups(
        self, product_key: str,
    ) -> frozenset[str]:
        """Return the set of group names that confer ``product_key`` on
        this site, per Atlassian's ``/rest/api/3/applicationrole``.

        This is the authoritative "which groups grant access to this
        product?" view that Atlassian's own Admin UI uses when an
        operator clicks "Remove access" for a single product on a user
        — it removes them from every group in this set that they are
        currently in, leaving identity groups (employee_in_china,
        synced-from-G-Suite) and unrelated product accesses alone. Helix
        re-implements that UI action via :meth:`revoke_product_access`
        because the Atlassian organization-level Admin API does NOT
        expose a single-product revoke endpoint (only whole-account
        suspend); we have to do the group-set math ourselves.

        Endpoint returns ``[{key, groups, defaultGroups, ...}, ...]`` —
        one entry per application role on the site. We union ``groups``
        and ``defaultGroups`` because both confer access; some Atlassian
        configs only populate ``defaultGroups`` (the one the role auto-
        assigns to new users) and leave the broader ``groups`` array
        empty, so taking only one of the two would miss real conferring
        groups. Uses the site REST API (Basic auth) — same credential
        grant/revoke already use, no org-admin authority needed.

        Returns an empty set when:
          * Atlassian returns no role matching ``product_key`` (config
            mismatch between the binding's declared product and what
            the site actually exposes — the caller surfaces this as a
            real failure rather than silently revoking nothing).
          * The matched role has empty ``groups`` AND ``defaultGroups``
            (no conferring groups at all on this site — same caller
            handling).
        Raises ``RuntimeError`` on a hard API failure so the caller
        refuses to "revoke product access" off an empty group set,
        which would silently no-op the revoke and let the seat sync
        re-import the allocation on the next tick.
        """
        key_norm = (product_key or "").strip().lower()
        if not key_norm:
            raise RuntimeError(
                "Jira list_conferring_groups called without a product_key — "
                "refusing to enumerate every role's groups (would over-revoke).",
            )
        site_base = await self._site_base()
        try:
            async with httpx.AsyncClient(
                timeout=_HTTP_TIMEOUT,
                trust_env=False,
                transport=await self._site_transport(),
            ) as client:
                resp = await self._request_with_retry(
                    client, "GET", f"{site_base}/applicationrole",
                    headers=self._basic_headers(),
                    what="Jira applicationrole",
                )
        except httpx.HTTPError as e:
            raise RuntimeError(
                f"Jira applicationrole failed: {type(e).__name__}: {e}",
            ) from e
        if resp.status_code != 200:
            raise RuntimeError(
                f"Jira applicationrole HTTP {resp.status_code}: "
                f"{_redact_jira_body(resp.text, max_len=200)}",
            )
        try:
            data = resp.json()
        except ValueError as e:
            raise RuntimeError(
                f"Jira applicationrole returned non-JSON body: {e}",
            ) from e
        if not isinstance(data, list):
            raise RuntimeError(
                "Jira applicationrole did not return a JSON array; "
                f"got {type(data).__name__}",
            )
        for role in data:
            if not isinstance(role, dict):
                continue
            role_key = str(role.get("key") or "").strip().lower()
            if role_key != key_norm:
                continue
            groups: set[str] = set()
            for src_field in ("groups", "defaultGroups"):
                src = role.get(src_field)
                if not isinstance(src, list):
                    continue
                for g in src:
                    if isinstance(g, str):
                        name = g.strip()
                        if name:
                            groups.add(name)
                    elif isinstance(g, dict):
                        # Atlassian's "Details" variant returns
                        # ``[{groupId, name, self}]`` instead of bare
                        # strings — accept both shapes so a site that
                        # serves either response stays supported.
                        name = str(g.get("name") or "").strip()
                        if name:
                            groups.add(name)
            return frozenset(groups)
        # No role matched. Return empty so the caller (revoke_product_access)
        # decides what to do — silently revoking 0 groups would be a worse
        # outcome than failing loudly with "binding's product_key does not
        # match any applicationrole on this site".
        return frozenset()

    async def revoke_all_access(
        self, ctx: ProvisioningContext,
    ) -> ProvisioningResult:
        """Offboarding: remove the leaver from EVERY group they belong to.

        A person's Jira/Confluence access can come from arbitrarily many
        groups (jira-administrators, SOC2, per-site product groups, …),
        so removing them from one bound group never covers everything —
        that was the original offboarding leak. The reliable action is to
        enumerate ALL of the account's groups via :meth:`get_user_groups`
        and DELETE the membership for each. The Atlassian account itself
        is never DELETED (Helix policy — issue history has to stay
        attributed to a real person).

        Since 2026-08-30 this also **suspends the account's sign-in**
        (:meth:`set_org_user_access_suspended`) when the integration has
        an unscoped org key and ``suspend_account_on_offboard`` is on.
        Stripping every group leaves the leaver with zero product access,
        but an account that can still authenticate is not an offboarded
        account — and the org-side IdP directory sync that used to
        provide that half was off for 16 days in August 2026 with nobody
        the wiser. Suspension is reversible (``/restore-access``) and
        deletes nothing, so it costs the history-attribution policy
        nothing. A suspend failure never fails the whole revoke: the
        group removals already succeeded and are the larger prize; the
        outcome is reported in ``raw_response.account_suspend`` and the
        daily assurance sweep re-attempts it.

        Uses only the site REST API (Basic auth) — the same credential
        grant/revoke already use, no org-admin write authority needed.
        Idempotent: a group the user is already absent from (404 / "not a
        member") counts as success, so re-runs and the many-seats →
        one-action dedup never dead-letter the offboarding blocker.

        Returns ``ok=False`` if ANY group removal failed for a real reason
        (so the blocker stays open and the leaver stays visible) while
        still reporting which groups WERE removed for the forensic trail.
        """
        email_norm = (ctx.user_email or "").strip().lower()
        # The offboarding leaver audit hands us the authoritative account_id
        # from the Org Admin directory listing via ``context_data`` →
        # ``ctx.extra``. We need it because site user-search is GDPR-blind:
        # ``/rest/api/3/user/search`` only returns accounts whose
        # ``emailAddress`` is visible, so a managed / deactivated leaver
        # (whose email the site search omits) resolves to ``None`` even
        # though the account still holds org-wide product access and group
        # memberships. Without this fallback those leavers were silently
        # reported as ``no_account`` (a no-op "success") and their groups
        # were never stripped — the perpetual false "persistent bounce".
        hint_account_id = (
            str((ctx.extra or {}).get("account_id") or "").strip() or None
        )
        if not email_norm and not hint_account_id:
            return ProvisioningResult(
                ok=False,
                error_message=(
                    "Jira revoke_all_access called with empty user_email and "
                    "no account_id hint — refusing to act without a target."
                ),
            )
        account_id: str | None = None
        if email_norm:
            try:
                user = await self.find_user_by_email(email_norm)
            except RuntimeError as e:
                return ProvisioningResult(ok=False, error_message=str(e))
            if user is not None:
                account_id = user.get("accountId")
                if not account_id:
                    return ProvisioningResult(
                        ok=False,
                        error_message=(
                            f"Jira lookup for {ctx.user_email!r} returned a "
                            "record without accountId"
                        ),
                    )
        if account_id is None and hint_account_id:
            # Site search couldn't see the account by email — fall back to
            # the account_id the audit resolved via the Org Admin directory.
            account_id = hint_account_id
        if account_id is None:
            # No Atlassian account at all → nothing to revoke.
            return ProvisioningResult(
                ok=True,
                external_ref=f"jira:{email_norm}:no_account",
                raw_response={"no_account": True},
            )
        try:
            groups = await self.get_user_groups(str(account_id))
        except RuntimeError as e:
            return ProvisioningResult(ok=False, error_message=str(e))

        site_base = await self._site_base()
        removed: list[str] = []
        already_absent: list[str] = []
        failed: list[str] = []
        async with httpx.AsyncClient(
            timeout=_HTTP_TIMEOUT,
            trust_env=False,
            transport=await self._site_transport(),
        ) as client:
            for group_name in groups:
                try:
                    rm = await self._request_with_retry(
                        client, "DELETE", f"{site_base}/group/user",
                        headers=self._basic_headers(),
                        params={
                            "groupname": group_name,
                            "accountId": account_id,
                        },
                        what="Jira group/user DELETE (revoke_all_access)",
                    )
                except httpx.HTTPError as e:
                    failed.append(f"{group_name}: {type(e).__name__}: {e}")
                    continue
                not_a_member = (
                    rm.status_code == 400 and self._is_not_a_member_response(rm)
                )
                if rm.status_code in (200, 204):
                    removed.append(group_name)
                    logger.info(
                        "jira group/user removed (offboarding all-access): "
                        "group=%s accountId=%s", group_name, account_id,
                    )
                elif rm.status_code == 404 or not_a_member:
                    already_absent.append(group_name)
                else:
                    failed.append(
                        f"{group_name}: HTTP {rm.status_code}: "
                        f"{_redact_jira_body(rm.text, max_len=150)}"
                    )

        if failed:
            return ProvisioningResult(
                ok=False,
                external_ref=str(account_id),
                error_message=(
                    f"Jira revoke_all_access: removed {len(removed)} of "
                    f"{len(groups)} group(s) for {email_norm}, "
                    f"{len(failed)} failed: {'; '.join(failed[:5])}"
                ),
                raw_response={
                    "account_id": str(account_id),
                    "groups_total": len(groups),
                    "removed": removed,
                    "already_absent": already_absent,
                    "failed": failed[:20],
                },
            )
        suspend_outcome = await self._suspend_account_after_revoke(
            account_id, ctx=ctx,
        )
        return ProvisioningResult(
            ok=True,
            external_ref=str(account_id),
            raw_response={
                "account_id": str(account_id),
                "groups_total": len(groups),
                "removed": removed,
                "already_absent": already_absent,
                "account_suspend": suspend_outcome,
            },
        )

    async def _suspend_account_after_revoke(
        self, account_id: str, *, ctx: ProvisioningContext,
    ) -> dict[str, Any]:
        """Best-effort sign-in suspension at the end of a leaver revoke.

        Deliberately **never raises and never flips the revoke to
        ok=False**. The group removals have already landed; failing the
        task here would dead-letter it and re-run those DELETEs on retry
        (each then a 400 "not a member" loop) for no gain. Instead the
        verdict is recorded, and the account-assurance sweep — which
        reads vendor truth daily rather than trusting a single HTTP
        response — is what actually closes the loop.

        Skips (each reported by name, never silently):
          * ``disabled``  — operator turned ``suspend_account_on_offboard``
            off on the integration.
          * ``no_unscoped_key`` — the credential these endpoints require
            is not configured; the alert tells IT exactly which key.
          * ``no_org_id`` — integration incompletely configured.
        """
        if not self._suspend_on_offboard_enabled():
            return {"status": "disabled"}
        if not self._org_id():
            return {"status": "no_org_id"}
        if not self._org_unscoped_api_key():
            return {"status": "no_unscoped_key"}
        try:
            res = await self.set_org_user_access_suspended(
                str(account_id), suspended=True,
            )
        except Exception as exc:  # noqa: BLE001 — never fail the revoke
            logger.warning(
                "jira revoke_all_access: suspend-access raised for %s: %s",
                account_id, exc,
            )
            return {"status": "error", "error": f"{type(exc).__name__}: {exc}"}
        if res.get("ok"):
            logger.info(
                "jira revoke_all_access: account sign-in suspended "
                "accountId=%s (user=%s)", account_id, ctx.user_email,
            )
            return {
                "status": "suspended",
                "http_status": res.get("status"),
                "already_absent": res.get("already_absent"),
            }
        return {
            "status": "failed",
            "http_status": res.get("status"),
            "error": res.get("error"),
        }

    async def revoke_product_access(
        self, ctx: ProvisioningContext,
    ) -> ProvisioningResult:
        """Active-employee self-cancel: strip ONE product family's access.

        Difference vs the other two revokes on this driver:

          * :meth:`revoke_seat` removes from a single bound
            ``group_name`` — insufficient when the user holds the
            product via any other conferring group (jira-administrators,
            team-specific groups, …). PingCAP's real users hold Jira
            via ``jira-dev-engineers`` / ``RD Leader``, not the
            connector-bound ``jira-software-users`` — so revoke_seat
            silently returned "not a member" while the seat stayed
            occupied and ``jira.seat_sync`` re-imported the allocation
            the next morning. This method fixes that loop.
          * :meth:`revoke_all_access` removes from EVERY group —
            correct for a leaver (scorched-earth) but over-revokes an
            active employee, stripping identity groups
            (``employee_in_china``, ``All users from G Suite``) and
            unrelated product accesses (Confluence) that the employee
            still needs.

        The intent here matches Atlassian's own Admin UI "Remove
        access" button for a single product: enumerate the product's
        conferring groups via :meth:`list_conferring_groups`,
        intersect with the user's current groups via
        :meth:`get_user_groups`, and DELETE that intersection. After
        the call ``product_access`` for ``product_key`` drops from the
        user's record, the next ``jira.seat_sync`` sees them gone from
        the license-holder roster, and the allocation stays revoked
        (no re-import bounce). Identity groups and other product
        accesses are preserved.

        ``ctx.extra`` must carry ``product_key`` (e.g. ``jira-software``,
        ``jira-servicedesk``, ``confluence``); ``sites`` is informational
        (the seat-sync's site filter, not used at revoke time since the
        site REST API is per-site anyway and the driver's configured
        ``site_url`` defines that site). Idempotent: groups the user is
        already absent from count as success (404 / "not a member" →
        already_absent), so re-runs are safe. Returns ``ok=False`` if
        ANY removal failed for a real reason — caller keeps the task
        open and surfaces the discrepancy to admin.
        """
        email_norm = (ctx.user_email or "").strip().lower()
        # #1651: site user-search is GDPR-blind — a managed/deactivated
        # leaver whose email the site omits resolves to None even though the
        # account still holds the product seat. Fall back to the authoritative
        # account_id the offboarding audit resolved via the Org Admin
        # directory (ctx.extra), exactly like revoke_all_access, so those
        # leavers aren't silently reported "no_account" while keeping a seat.
        hint_account_id = (
            str((ctx.extra or {}).get("account_id") or "").strip() or None
        )
        if not email_norm and not hint_account_id:
            return ProvisioningResult(
                ok=False,
                error_message=(
                    "Jira revoke_product_access called with empty user_email "
                    "and no account_id hint — refusing to act without a target."
                ),
            )
        product_key = str(
            (ctx.extra or {}).get("product_key") or "",
        ).strip().lower()
        if not product_key:
            return ProvisioningResult(
                ok=False,
                error_message=(
                    "Jira revoke_product_access called without product_key in "
                    "ctx.extra — dispatcher must set it from the source "
                    "connector binding's product_key field."
                ),
            )
        account_id: str | None = None
        if email_norm:
            try:
                user = await self.find_user_by_email(email_norm)
            except RuntimeError as e:
                return ProvisioningResult(ok=False, error_message=str(e))
            if user is not None:
                account_id = user.get("accountId")
                if not account_id:
                    return ProvisioningResult(
                        ok=False,
                        error_message=(
                            f"Jira lookup for {ctx.user_email!r} returned a "
                            "record without accountId"
                        ),
                    )
        if account_id is None and hint_account_id:
            # Site search couldn't see the account by email — use the
            # account_id the audit resolved via the Org Admin directory.
            account_id = hint_account_id
        if account_id is None:
            # No Atlassian account at all → nothing to revoke. Same
            # success-shape as revoke_all_access so the task flips
            # to completed and the Helix-side allocation cascade fires.
            return ProvisioningResult(
                ok=True,
                external_ref=f"jira:{email_norm}:no_account",
                raw_response={"no_account": True, "product_key": product_key},
            )

        applicationrole_error: str | None = None
        try:
            conferring = await self.list_conferring_groups(product_key)
        except RuntimeError as e:
            # Don't bail yet — the admin-declared list below may carry
            # the answer even when applicationrole is down /
            # misconfigured. Remember the error so we can attribute
            # later if BOTH paths come up empty.
            conferring = frozenset()
            applicationrole_error = str(e)
        # ── Admin-declared conferring groups for products
        #    ``/rest/api/3/applicationrole`` doesn't cover ───────────
        #
        # applicationrole is **Jira-only** on Atlassian Cloud. Confluence
        # (and any future non-Jira Atlassian product on the same site)
        # has no equivalent endpoint, and the Atlassian Org Admin API
        # does not expose per-group product-access metadata either (we
        # verified — the public 2022 community thread confirming "no
        # per-product revoke API" still stands as of 2026-05). So
        # admin-declared lists are the **primary** mechanism for these
        # products, NOT a fallback masquerading as a feature.
        #
        # ``ctx.extra['admin_declared_conferring_groups']`` carries a
        # list of group names per ``product_key``, populated by the
        # dispatcher from platform_setting
        # ``software_seat.product_access_revoke.admin_declared_conferring_groups``
        # (admin PUTs once via the standard platform-settings endpoint).
        # ``ctx.extra['admin_declared_groups_load_error']`` is set when
        # the dispatcher detected a malformed setting (DB outage,
        # value not a JSON object, list-of-non-strings, etc.) — the
        # driver propagates that diagnostic into the refusal message
        # so the admin re-PUTting broken JSON gets a precise breadcrumb
        # instead of "no fallback configured" looping forever.
        #
        # The driver treats the admin-declared list EXACTLY like
        # applicationrole output: intersect with user's actual groups,
        # DELETE the intersection. Refusal-on-empty still applies.
        # When BOTH applicationrole AND admin-declared have entries we
        # union — applicationrole is canonical but admin extras cover
        # custom groups conferring access via Atlassian Access
        # user-provisioning rules that don't surface in applicationrole.
        admin_declared_raw = (ctx.extra or {}).get(
            "admin_declared_conferring_groups",
        )
        admin_declared_groups: set[str] = set()
        if isinstance(admin_declared_raw, list):
            for g in admin_declared_raw:
                if isinstance(g, str):
                    name = g.strip()
                    if name:
                        admin_declared_groups.add(name)
        admin_declared_load_error = (ctx.extra or {}).get(
            "admin_declared_groups_load_error",
        )
        if not conferring and admin_declared_groups:
            conferring = frozenset(admin_declared_groups)
            conferring_source = "admin_declared"
            logger.info(
                "Jira revoke_product_access(%s): applicationrole returned no "
                "conferring groups, using admin-declared list (%d group(s))",
                product_key, len(admin_declared_groups),
            )
        elif conferring and admin_declared_groups:
            conferring = frozenset(set(conferring) | admin_declared_groups)
            conferring_source = "union"
        elif conferring:
            conferring_source = "applicationrole"
        else:
            conferring_source = "none"
        if not conferring:
            # Hard refusal — caller must NOT mark the seat revoked
            # against an empty target set (would be the same silent
            # no-op as the bug this method was written to fix).
            # Round 3 audit P1.4: explicit attribution between the
            # three possible refusal causes:
            #   (a) admin never wrote the setting → "configure it"
            #   (b) admin wrote the setting but it's malformed →
            #       "your setting exists but shape is wrong, here's why"
            #   (c) applicationrole hit a real error AND no admin list →
            #       both root causes in one message
            if admin_declared_load_error:
                # Case (b): setting present but broken.
                detail = (
                    f"Jira revoke_product_access for product_key="
                    f"{product_key!r} refused — applicationrole returned "
                    "no conferring groups (Jira-only endpoint) AND the "
                    "admin-declared list at platform_setting "
                    "``software_seat.product_access_revoke.admin_declared_"
                    f"conferring_groups`` failed to load: "
                    f"{admin_declared_load_error}"
                )
            else:
                # Case (a) or (c): setting absent.
                detail = (
                    f"Jira applicationrole returned no conferring groups "
                    f"for product_key={product_key!r} on this site, and "
                    "no admin-declared list is configured in "
                    "platform_setting "
                    "``software_seat.product_access_revoke."
                    "admin_declared_conferring_groups`` for this product. "
                    "For Atlassian products that applicationrole doesn't "
                    "cover (Confluence and similar non-Jira products on "
                    "the same site), the admin must enumerate the "
                    "conferring groups ONCE from Atlassian Admin UI "
                    "(Directory → Groups → filter on app access) and PUT "
                    "them at "
                    "``/api/v1/platform-settings/software_seat.product_"
                    "access_revoke.admin_declared_conferring_groups`` as "
                    "``{<product_key>: [<group_name>, ...]}``."
                )
            if applicationrole_error:
                detail += (
                    f" applicationrole error: {applicationrole_error}"
                )
            return ProvisioningResult(
                ok=False,
                error_message=detail,
                raw_response={
                    "product_key": product_key,
                    "conferring_source": "none",
                    "applicationrole_error": applicationrole_error,
                    "admin_declared_groups_load_error": (
                        admin_declared_load_error
                    ),
                },
            )

        try:
            user_groups = await self.get_user_groups(str(account_id))
        except RuntimeError as e:
            return ProvisioningResult(ok=False, error_message=str(e))

        # Intersection — only groups the user is actually in that ALSO
        # confer the target product. Stable order (sorted) so audit
        # logs and tests can compare deterministically.
        user_conferring = set(user_groups) & conferring

        # ── Org-wide-entitlement guard ────────────────────────────────
        #
        # Never remove an active employee from a protected org-wide
        # identity group (``employee_in_china`` / ``All users from G
        # Suite`` / admin-configured extras) to reclaim a product seat.
        # If the user holds the product via ANY such group, the product
        # is an org-wide entitlement — removing the removable conferring
        # groups would not clear ``product_access`` (the identity group
        # still confers it) and SCIM would re-add the identity group
        # anyway, so the "revoke" would bounce forever. Refuse to touch
        # anything and report ``not_individually_revocable`` so the
        # reclaim / audit pipeline treats the seat as terminal — no
        # retry, no weekly bounce page. See ``DEFAULT_PROTECTED_IDENTITY_GROUPS``.
        protected_norm = set(DEFAULT_PROTECTED_IDENTITY_GROUPS)
        extra_protected = (ctx.extra or {}).get("protected_identity_groups")
        if isinstance(extra_protected, list):
            for g in extra_protected:
                if isinstance(g, str) and g.strip():
                    protected_norm.add(_normalize_group_name(g))
        protected_present = sorted(
            g for g in user_conferring
            if _normalize_group_name(g) in protected_norm
        )
        if protected_present:
            logger.warning(
                "Jira revoke_product_access(%s): %s holds the product via "
                "protected org-wide identity group(s) %s — org-wide "
                "entitlement, NOT individually revocable. Refusing to "
                "remove any group (would strip a directory group + bounce "
                "via SCIM). Mark the plan seat_revocable=false.",
                product_key, email_norm, protected_present,
            )
            return ProvisioningResult(
                ok=True,
                external_ref=str(account_id),
                raw_response={
                    "account_id": str(account_id),
                    "product_key": product_key,
                    "conferring_source": conferring_source,
                    "conferring_groups_total": len(conferring),
                    "user_groups_total": len(user_groups),
                    "targets": [],
                    "removed": [],
                    "already_absent": [],
                    "not_individually_revocable": True,
                    "org_wide_entitlement": True,
                    "protected_conferring_groups": protected_present,
                },
            )

        targets = sorted(user_conferring)

        site_base = await self._site_base()
        removed: list[str] = []
        already_absent: list[str] = []
        failed: list[str] = []
        if targets:
            async with httpx.AsyncClient(
                timeout=_HTTP_TIMEOUT,
                trust_env=False,
                transport=await self._site_transport(),
            ) as client:
                for group_name in targets:
                    try:
                        rm = await self._request_with_retry(
                            client, "DELETE", f"{site_base}/group/user",
                            headers=self._basic_headers(),
                            params={
                                "groupname": group_name,
                                "accountId": account_id,
                            },
                            what="Jira group/user DELETE (revoke_product_access)",
                        )
                    except httpx.HTTPError as e:
                        failed.append(
                            f"{group_name}: {type(e).__name__}: {e}",
                        )
                        continue
                    not_a_member = (
                        rm.status_code == 400
                        and self._is_not_a_member_response(rm)
                    )
                    if rm.status_code in (200, 204):
                        removed.append(group_name)
                        logger.info(
                            "jira group/user removed (product_access revoke "
                            "for %s): group=%s accountId=%s",
                            product_key, group_name, account_id,
                        )
                    elif rm.status_code == 404 or not_a_member:
                        already_absent.append(group_name)
                    else:
                        failed.append(
                            f"{group_name}: HTTP {rm.status_code}: "
                            f"{_redact_jira_body(rm.text, max_len=150)}",
                        )

        # Diagnostic raw_response always carries the intersection size
        # so the audit log explains WHY zero groups were removed — was
        # it "user never in any conferring group" (intersection empty)
        # vs "user is in groups but all already absent" (404 path).
        # Round 3 audit P2.5: attribute WHICH source produced the
        # conferring set so post-mortem ("why did we delete group X?")
        # can be answered from the audit row alone. ``source`` is one
        # of ``applicationrole`` (Jira-only canonical), ``admin_declared``
        # (Confluence + non-Jira products via platform_setting), or
        # ``union`` (both contributed).
        raw_response: dict[str, Any] = {
            "account_id": str(account_id),
            "product_key": product_key,
            "conferring_source": conferring_source,
            "conferring_groups_total": len(conferring),
            "user_groups_total": len(user_groups),
            "targets": targets,
            "removed": removed,
            "already_absent": already_absent,
        }
        if applicationrole_error:
            raw_response["applicationrole_error"] = applicationrole_error
        if failed:
            raw_response["failed"] = failed[:20]
            return ProvisioningResult(
                ok=False,
                external_ref=str(account_id),
                error_message=(
                    f"Jira revoke_product_access({product_key}): removed "
                    f"{len(removed)} of {len(targets)} group(s) for "
                    f"{email_norm}, {len(failed)} failed: "
                    f"{'; '.join(failed[:5])}"
                ),
                raw_response=raw_response,
            )
        return ProvisioningResult(
            ok=True,
            external_ref=str(account_id),
            raw_response=raw_response,
        )

    # ── grant_seat / revoke_seat ──────────────────────────────────

    async def grant_seat(self, ctx: ProvisioningContext) -> ProvisioningResult:
        """Add the user to the binding's ``group_name``.

        Sequence:
          1. Look up user by email. If absent, POST ``/rest/api/3/user``
             to invite (Atlassian sends an email + creates the account).
          2. POST ``/rest/api/3/group/user?groupname=…`` with
             ``{"accountId": "…"}`` to add to the product group.
          3. ``external_ref`` = the Atlassian accountId so future
             revoke calls can target the exact account.

        Idempotent: a 400 from /group/user with body containing
        "user is already a member" is treated as success.
        """
        email_norm = (ctx.user_email or "").strip().lower()
        if not email_norm:
            return ProvisioningResult(
                ok=False,
                error_message=(
                    "Jira grant_seat called with empty user_email — "
                    "refusing to invite with no recipient address."
                ),
            )
        group_name = str(
            (ctx.extra or {}).get("group_name") or "",
        ).strip()
        if not group_name:
            return ProvisioningResult(
                ok=False,
                error_message=(
                    "Jira binding missing group_name — set it on the "
                    "software_connectors row binding_config."
                ),
            )

        try:
            user = await self.find_user_by_email(email_norm)
        except RuntimeError as e:
            return ProvisioningResult(ok=False, error_message=str(e))

        account_id: str | None = None
        if user is not None:
            account_id = user.get("accountId")
        else:
            # Invite the user. Atlassian site-admins can create accounts;
            # the user receives an email with a sign-up link. The first
            # time they accept, the Atlassian account becomes active.
            invite_payload: dict[str, Any] = {
                "emailAddress": email_norm,
                "products": [],  # admin manages products via groups
            }
            display = (ctx.user_display_name or "").strip()
            if display:
                invite_payload["displayName"] = display
            try:
                async with httpx.AsyncClient(
                    timeout=_HTTP_TIMEOUT,
                    trust_env=False,
                    transport=await self._site_transport(),
                ) as client:
                    inv = await self._request_with_retry(
                        client, "POST", f"{await self._site_base()}/user",
                        headers=self._basic_headers(),
                        json=invite_payload,
                        what="Jira invite",
                    )
            except httpx.HTTPError as e:
                return ProvisioningResult(
                    ok=False,
                    error_message=f"Jira invite failed: {type(e).__name__}: {e}",
                )
            if inv.status_code in (200, 201):
                inv_data = inv.json() or {}
                account_id = inv_data.get("accountId")
            elif (
                inv.status_code == 400 and "already exists" in inv.text.lower()
            ) or inv.status_code == 409:
                # Re-lookup — race window where another process invited
                # in parallel, or Atlassian's search index lagged behind
                # the user's actual existence. Atlassian returns 400
                # "already exists" OR 409 (newer API / concurrent invite)
                # for an existing user; both must resolve to the existing
                # accountId to keep grant idempotent, not hard-fail.
                try:
                    user = await self.find_user_by_email(email_norm)
                except RuntimeError as e:
                    return ProvisioningResult(ok=False, error_message=str(e))
                if user is not None:
                    account_id = user.get("accountId")
            else:
                return ProvisioningResult(
                    ok=False,
                    error_message=(
                        f"Jira invite HTTP {inv.status_code}: "
                        f"{_redact_jira_body(inv.text, max_len=300)}"
                    ),
                )

        if not account_id:
            return ProvisioningResult(
                ok=False,
                error_message=(
                    f"Jira: could not resolve accountId for "
                    f"{ctx.user_email!r} after invite"
                ),
            )

        try:
            async with httpx.AsyncClient(
                timeout=_HTTP_TIMEOUT,
                trust_env=False,
                transport=await self._site_transport(),
            ) as client:
                add = await self._request_with_retry(
                    client, "POST", f"{await self._site_base()}/group/user",
                    headers=self._basic_headers(),
                    params={"groupname": group_name},
                    json={"accountId": account_id},
                    what="Jira group/user POST (add)",
                )
        except httpx.HTTPError as e:
            return ProvisioningResult(
                ok=False,
                error_message=f"Jira add-to-group failed: {type(e).__name__}: {e}",
            )
        if add.status_code in (200, 201):
            return ProvisioningResult(
                ok=True,
                external_ref=str(account_id),
                raw_response={"group": group_name, "added": True},
            )
        if add.status_code == 400 and self._is_already_member_response(add):
            return ProvisioningResult(
                ok=True,
                external_ref=str(account_id),
                raw_response={"group": group_name, "already_member": True},
            )
        if add.status_code == 404:
            return ProvisioningResult(
                ok=False,
                error_message=(
                    f"Jira group {group_name!r} not found — check binding's "
                    "group_name and that the group exists in Jira admin."
                ),
            )
        return ProvisioningResult(
            ok=False,
            error_message=(
                f"Jira group/user HTTP {add.status_code}: {_redact_jira_body(add.text, max_len=300)}"
            ),
        )

    async def revoke_seat(self, ctx: ProvisioningContext) -> ProvisioningResult:
        """Remove the user from the binding's ``group_name``.

        ``action_on_offboard='delete'`` additionally calls Atlassian
        Admin API to disable the account org-wide (requires ``org_id``
        + Atlassian Guard).

        Idempotent: 404 from /group/user (group missing or user not
        in group) AND 400 with body "since user is not a member of …"
        (Atlassian's actual response when DELETEing a non-member) are
        both treated as already-absent success — we still want the
        offboarding flow to mark the blocker resolved.

        Asymmetry with ``grant_seat`` is deliberate: a 404 there means
        the bound group does not exist and adding to it must fail loud
        (the binding is misconfigured); here a 404 may mean the same
        group is gone, but the desired end-state for offboarding is
        "user not in this group" and a deleted group trivially
        satisfies that — closing the blocker is correct.
        """
        email_norm = (ctx.user_email or "").strip().lower()
        # #1651: fall back to the Org-Admin-directory account_id when the
        # site can't resolve the leaver by email (GDPR-hidden / deactivated),
        # so a leaver who still holds the seat isn't mis-reported
        # already_absent (a no-op "success") and left occupying a paid seat.
        hint_account_id = (
            str((ctx.extra or {}).get("account_id") or "").strip() or None
        )
        if not email_norm and not hint_account_id:
            # Treating an empty email as "already absent" would silently
            # close an offboarding blocker without ever talking to Jira.
            # Surface it as a real failure so the operator can fix the
            # upstream record.
            return ProvisioningResult(
                ok=False,
                error_message=(
                    "Jira revoke_seat called with empty user_email and no "
                    "account_id hint — refusing to close the seat without a "
                    "target."
                ),
            )
        group_name = str(
            (ctx.extra or {}).get("group_name") or "",
        ).strip()
        if not group_name:
            return ProvisioningResult(
                ok=False,
                error_message=(
                    "Jira binding missing group_name — set it on the "
                    "software_connectors row binding_config."
                ),
            )

        raw_action = (ctx.extra or {}).get("action_on_offboard")
        if raw_action is None or str(raw_action).strip() == "":
            return ProvisioningResult(
                ok=False,
                error_message=(
                    "Jira binding missing action_on_offboard — set it "
                    "on the software_connectors row binding_config "
                    "(deactivate|delete)."
                ),
            )
        action = str(raw_action).strip().lower()
        if action not in ("deactivate", "delete"):
            return ProvisioningResult(
                ok=False,
                error_message=(
                    f"Jira binding invalid action_on_offboard "
                    f"{raw_action!r} (expected deactivate|delete)"
                ),
            )

        account_id: str | None = None
        if email_norm:
            try:
                user = await self.find_user_by_email(email_norm)
            except RuntimeError as e:
                return ProvisioningResult(ok=False, error_message=str(e))
            if user is not None:
                account_id = user.get("accountId")
                if not account_id:
                    return ProvisioningResult(
                        ok=False,
                        error_message=(
                            f"Jira lookup for {ctx.user_email!r} returned "
                            "record without accountId"
                        ),
                        raw_response=user,
                    )
        if account_id is None and hint_account_id:
            account_id = hint_account_id
        if account_id is None:
            return ProvisioningResult(
                ok=True,
                external_ref=f"jira:{email_norm}:already_absent",
                raw_response={"already_absent": True},
            )

        try:
            async with httpx.AsyncClient(
                timeout=_HTTP_TIMEOUT,
                trust_env=False,
                transport=await self._site_transport(),
            ) as client:
                rm = await self._request_with_retry(
                    client, "DELETE",
                    f"{await self._site_base()}/group/user",
                    headers=self._basic_headers(),
                    params={
                        "groupname": group_name,
                        "accountId": account_id,
                    },
                    what="Jira group/user DELETE (revoke_seat)",
                )
        except httpx.HTTPError as e:
            return ProvisioningResult(
                ok=False,
                error_message=f"Jira remove-from-group failed: {type(e).__name__}: {e}",
            )
        not_a_member_400 = (
            rm.status_code == 400 and self._is_not_a_member_response(rm)
        )
        if rm.status_code not in (200, 204, 404) and not not_a_member_400:
            return ProvisioningResult(
                ok=False,
                error_message=(
                    f"Jira group/user DELETE HTTP {rm.status_code}: "
                    f"{_redact_jira_body(rm.text, max_len=300)}"
                ),
            )
        already_absent = rm.status_code == 404 or not_a_member_400
        removed_now = rm.status_code in (200, 204)

        # Round-1 audit HIGH #5: log destructive op for forensic trail.
        if removed_now:
            logger.info(
                "jira group/user removed: group=%s accountId=%s",
                group_name, account_id,
            )

        if action == "deactivate":
            return ProvisioningResult(
                ok=True,
                external_ref=str(account_id),
                raw_response={
                    "group": group_name,
                    "removed": removed_now,
                    "already_absent": already_absent,
                },
            )

        # action == "delete" — additionally strip the account's org-wide
        # product access via the Organizations REST API
        # (DELETE directory/users), clearing access that group removal
        # can't (Trello workspace membership, org default access, direct
        # grants). Keeps the Atlassian account (issue history preserved);
        # requires org_id + an UNSCOPED org API key (org_unscoped_api_key).
        #
        # NOTE (2026-06): the previous implementation called
        # ``/admin/v1/orgs/{org}/users/{acct}/manage/lifecycle/disable``,
        # which is NOT a valid Atlassian route (404) and, worse, the 404
        # was mis-handled as "already disabled = success" — a silent
        # false-positive. Verified against prod: lifecycle/disable lives
        # under the User-Management API and needs ``manage:org`` (not
        # grantable to API keys); the working revoke is directory
        # remove-access with an unscoped key.
        org_id = self._org_id()
        if not org_id:
            return ProvisioningResult(
                ok=False,
                error_message=(
                    "Jira binding action_on_offboard=delete requires "
                    "integration org_id to be set (so Helix can call the "
                    "Atlassian Admin API to remove org access)."
                ),
            )
        if not self._org_unscoped_api_key():
            return ProvisioningResult(
                ok=False,
                error_message=(
                    "Jira binding action_on_offboard=delete requires "
                    "integration org_unscoped_api_key — an UNSCOPED org "
                    "API key (admin.atlassian.com/o/{orgId}/api-keys, "
                    "create with NO scopes). Scoped keys 401 on the "
                    "user-access endpoints."
                ),
            )
        try:
            access = await self.remove_org_user_access(account_id)
        except ValueError as e:
            return ProvisioningResult(ok=False, error_message=str(e))

        if access["ok"]:
            return ProvisioningResult(
                ok=True,
                external_ref=str(account_id),
                raw_response={
                    "group": group_name,
                    "removed_from_group": removed_now,
                    "already_absent_from_group": already_absent,
                    "org_access_removed": access["removed"],
                    "org_access_already_absent": access["already_absent"],
                    "remove_access_status": access["status"],
                },
            )
        # 401/403 = permanent auth/scope problem (key still scoped /
        # revoked / lacks org admin) — retrying won't help. The group
        # removal in phase-1 ALREADY SUCCEEDED, so the user is off the
        # licensed group seat. Returning ok=False would dead-letter the
        # offboarding task and re-run the group DELETE on retry (→ 400
        # "not a member" loop). Surface ok=True with a tagged partial
        # flag so audit + failure-DM show the deferred state and an admin
        # can finish it (configure the unscoped key) — never auto-retry.
        if access["status"] in (401, 403):
            tag = (
                JIRA_TAG_ORG_ADMIN_AUTH_FAILED
                if access["status"] == 401
                else JIRA_TAG_ORG_ADMIN_PERMISSION_DENIED
            )
            hint = (
                "org_unscoped_api_key is missing/scoped/revoked. Create an "
                "UNSCOPED org API key (no scopes) at admin.atlassian.com → "
                "Settings → API keys — scoped keys 401 on remove-access."
            )
            logger.warning(
                "jira remove-org-access refused %s for accountId=%s (group "
                "removal already done, org access NOT removed): %s",
                access["status"], account_id, hint,
            )
            return ProvisioningResult(
                ok=True,
                external_ref=str(account_id),
                raw_response={
                    "partial_success_tag": tag,
                    "group": group_name,
                    "removed_from_group": removed_now,
                    "already_absent_from_group": already_absent,
                    "org_access_removed": False,
                    "org_access_remove_deferred": True,
                    "remove_access_status": access["status"],
                    "remove_access_error": access["error"],
                    "deferred_reason": hint,
                },
            )
        # Anything else (5xx / network error = mostly transient; a route
        # 404 = misconfig, not transient) — return ok=False so the task
        # retries up to the dispatcher's max-attempts cap and then
        # dead-letters with an admin-visible audit row (no infinite
        # loop). Preserve the phase-1 group-removal state in raw_response
        # so audit can tell phase-1 from phase-2 failures.
        return ProvisioningResult(
            ok=False,
            error_message=(
                f"Jira remove-org-access HTTP {access['status']}: "
                f"{access['error']} (group removal already done)"
            ),
            raw_response={
                "group": group_name,
                "removed_from_group": removed_now,
                "already_absent_from_group": already_absent,
                "org_access_removed": False,
                "remove_access_status": access["status"],
                "remove_access_error": access["error"],
            },
        )