"""EmailGroupService — coordinates Helix rows + Google Workspace calls.

Flow contract for every write op:

    1. Validate + rate-limit against DB (fail fast).
    2. Begin savepoint.
    3. Write local row(s).
    4. Call Google API via :class:`GoogleWorkspaceDriver`.
    5. On Google failure, roll the savepoint back (no orphan local
       rows). On success, commit and enqueue audit + notification.

This keeps local and Google in sync: if Google rejects, Helix forgets
about it; if Helix crashes between the call and commit, a reconcile
job catches the drift on next run.

The reads (public directory, per-person owned groups) never touch
Google — they hit local rows only, so the directory page loads fast
even when Google is slow.
"""
from __future__ import annotations

import asyncio
import re
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from typing import TYPE_CHECKING
from uuid import UUID, uuid4

from sqlalchemy import String as sa_String
from sqlalchemy import and_, asc, func, or_, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from app.audit.service import AuditService, _esc
from app.core.corporate_domains import DEFAULT_CORPORATE_EMAIL_DOMAINS
from app.core.exceptions import AppException, NotFoundError
from app.core.rbac import (
    GLOBAL_ADMIN_ROLE,
    can_alias_for_others,
    can_delete_any_mail_group,
    has_role,
)
from app.domains.asset.models import EmployeeRequest
from app.domains.email_group.models import (
    EmailGroup,
    EmailGroupJoinRequest,
    EmailGroupOwner,
)
from app.domains.email_group.schemas import (
    EmailAliasCreateRequest,
    EmailAliasItem,
    EmailAliasListResponse,
    EmailGroupCreateRequest,
    MemberGroupItem,
    MemberGroupListResponse,
    PasswordResetRequest,
)
from app.identity.person_models import IdentityPerson
from app.integrations.drivers.google_workspace import (
    GoogleWorkspaceDriver,
)
from app.integrations.models import Integration
from app.integrations.service import (
    _resolve,  # noqa: PLC2701 — intentional reuse
    get_platform_setting,
)
from app.notifications.service import NotificationEnvelope, NotificationService

if TYPE_CHECKING:
    pass

from app.core.safe_logging import get_safe_logger

logger = get_safe_logger(__name__)


# Single source of truth lives in app.core.corporate_domains — the same
# company-owned-domains list the identity work_email guard trusts. Kept
# as a sorted list here for the existing list-typed return contract.
_DEFAULT_SUFFIX_WHITELIST = sorted(DEFAULT_CORPORATE_EMAIL_DOMAINS)
_DEFAULT_PASSWORD_RESET_HOURS = 2
_DEFAULT_MAIL_GROUP_CREATE_HOURS = 2

_EMAIL_RE = re.compile(r"^[a-z0-9][a-z0-9._+\-]{0,62}[a-z0-9]@[a-z0-9.\-]+$|^[a-z0-9]@[a-z0-9.\-]+$")


@dataclass(slots=True)
class MailGroupCreateResult:
    request: EmployeeRequest
    group: EmailGroup


@dataclass(slots=True)
class AliasCreateResult:
    request: EmployeeRequest
    alias_email: str


@dataclass(slots=True)
class PasswordResetResult:
    request: EmployeeRequest
    temp_password: str


class EmailGroupService:
    def __init__(self, db: AsyncSession) -> None:
        self.db = db
        self.audit = AuditService(db)
        self.notify = NotificationService(db)

    # ─── Driver resolution ────────────────────────────────────────

    async def _get_driver(self) -> GoogleWorkspaceDriver:
        row = (
            await self.db.execute(
                select(Integration).where(
                    Integration.kind == "software_provisioning",
                    Integration.provider == "google_workspace",
                    Integration.enabled.is_(True),
                )
                .order_by(
                    Integration.is_primary.desc(),
                    Integration.created_at,
                )
            )
        ).scalars().first()
        if row is None:
            raise AppException(
                code="GOOGLE_WORKSPACE_NOT_CONFIGURED",
                message=(
                    "Google Workspace 集成未配置或已禁用。请联系 IT Admin "
                    "在 /admin/integrations 中添加并启用。"
                ),
                status_code=503,
            )
        resolved = await _resolve(self.db, row)
        driver = GoogleWorkspaceDriver(
            integration_id=str(row.id), config=resolved.config,
        )
        return driver

    # ─── Configuration readers ────────────────────────────────────

    async def get_suffix_whitelist(self) -> list[str]:
        """Read the suffix whitelist from ``platform_settings``.

        Returns the built-in default if the setting row is missing or
        malformed — avoids blocking the feature on a seed-data glitch.
        """
        raw = await get_platform_setting(
            "google.mail_group.suffix_whitelist",
            default=None,
            session=self.db,
        )
        if isinstance(raw, list) and all(isinstance(x, str) for x in raw):
            return [s.strip().lower() for s in raw if s.strip()]
        return list(_DEFAULT_SUFFIX_WHITELIST)

    async def _password_reset_interval_hours(self) -> int:
        raw = await get_platform_setting(
            "google.password_reset.rate_limit_hours",
            default=_DEFAULT_PASSWORD_RESET_HOURS,
            session=self.db,
        )
        try:
            return max(int(raw), 0)
        except (TypeError, ValueError):
            return _DEFAULT_PASSWORD_RESET_HOURS

    async def _mail_group_create_interval_hours(self) -> int:
        raw = await get_platform_setting(
            "google.mail_group.create_rate_limit_hours",
            default=_DEFAULT_MAIL_GROUP_CREATE_HOURS,
            session=self.db,
        )
        try:
            return max(int(raw), 0)
        except (TypeError, ValueError):
            return _DEFAULT_MAIL_GROUP_CREATE_HOURS

    # ─── Public directory reads ───────────────────────────────────

    async def list_groups(
        self,
        *,
        search: str | None = None,
        suffix: str | None = None,
        owner_person_id: UUID | None = None,
        only_unowned: bool = False,
        created_from: datetime | None = None,
        created_to: datetime | None = None,
        page: int = 1,
        page_size: int = 50,
    ) -> tuple[list[EmailGroup], int, dict[UUID, IdentityPerson]]:
        """Paginated active-groups list with owner eager-load + filters.

        Filter semantics (all composable, AND'd together):

        * ``search`` — case-insensitive match on email / name / description.
        * ``suffix`` — exact domain after ``@`` (e.g. ``pingcap.com``).
        * ``owner_person_id`` — only groups where this person is an
          owner. The UI uses this for the "我拥有的" checkbox (pass the
          current user's id) and for an owner-picker filter.
        * ``only_unowned`` — governance view: groups with ZERO owner
          rows. Mutually exclusive with ``owner_person_id`` — if both
          are set, ``only_unowned`` wins (the UI shouldn't send both,
          but the service guards anyway).
        * ``created_from`` / ``created_to`` — inclusive range on
          ``created_at``. Datetime at the service layer so the router
          can accept ``YYYY-MM-DD`` and parse to start/end-of-day.

        Owners are returned in a dict keyed by person_id so the caller
        can serialize without N+1.
        """
        page = max(page, 1)
        page_size = max(min(page_size, 200), 1)

        conds: list = [EmailGroup.status == "active"]
        if search:
            # Escaped, with ``escape=`` on every comparison. The admin
            # twin of this filter (``manage_service._like``) was fixed to
            # do this and THIS half was left as-is — which is how a
            # paired contract ends up with the bug simply relocated: an
            # unescaped ``_`` is LIKE's single-character wildcard, so an
            # employee searching ``g_all`` silently matches ``g-all`` and
            # a ``%`` matches everything. Same helper, so the two halves
            # cannot drift again.
            pattern = f"%{_esc(search.strip().lower())}%"
            # Aliases live in a JSON column (list of lowercase strings).
            # Cast-to-CHAR + LIKE is the portable trick that works on
            # both MySQL 8 and TiDB without needing JSON_SEARCH (which
            # has subtly different argument shapes across the two).
            # ``CAST(NULL AS CHAR)`` is NULL → COALESCE to '[]' so the
            # LIKE doesn't drop unsynced rows from the result set even
            # when their alias / name happens to match.
            aliases_text = func.lower(
                func.coalesce(func.cast(EmailGroup.aliases, sa_String), "[]"),
            )
            conds.append(or_(
                func.lower(EmailGroup.email).like(pattern, escape="\\"),
                func.lower(EmailGroup.name).like(pattern, escape="\\"),
                func.lower(func.coalesce(EmailGroup.description, "")).like(
                    pattern, escape="\\",
                ),
                aliases_text.like(pattern, escape="\\"),
            ))
        if suffix:
            suffix_clean = suffix.strip().lower()
            if suffix_clean:
                # ``LOWER(email) LIKE '%@suffix'`` — pins the suffix
                # to the right side so ``pingcap.com`` doesn't match
                # ``pingcap.com.cn``. Using LIKE (not substring-eq) so
                # the existing email index on ``email`` stays usable
                # for the prefix.
                conds.append(
                    func.lower(EmailGroup.email).like(
                        f"%@{_esc(suffix_clean)}", escape="\\",
                    ),
                )
        if created_from is not None:
            conds.append(EmailGroup.created_at >= created_from)
        if created_to is not None:
            conds.append(EmailGroup.created_at <= created_to)

        # Owner-based filters need a join / anti-join. Keep them last
        # so the simpler WHERE-only path above stays fast when no
        # owner filter is set.
        if only_unowned:
            # Anti-join: groups with NO owner rows.
            subq = select(EmailGroupOwner.email_group_id).distinct()
            conds.append(~EmailGroup.id.in_(subq))
        elif owner_person_id is not None:
            subq = (
                select(EmailGroupOwner.email_group_id)
                .where(EmailGroupOwner.person_id == owner_person_id)
            )
            conds.append(EmailGroup.id.in_(subq))

        stmt = (
            select(EmailGroup)
            .where(*conds)
            .options(selectinload(EmailGroup.owners))
            .order_by(asc(EmailGroup.email))
        )
        count_stmt = select(func.count(EmailGroup.id)).where(*conds)

        total = (await self.db.execute(count_stmt)).scalar() or 0
        rows = (
            await self.db.execute(
                stmt.offset((page - 1) * page_size).limit(page_size)
            )
        ).scalars().all()

        person_ids: set[UUID] = set()
        for g in rows:
            for o in g.owners:
                person_ids.add(o.person_id)
        persons_map: dict[UUID, IdentityPerson] = {}
        if person_ids:
            persons = (
                await self.db.execute(
                    select(IdentityPerson).where(IdentityPerson.id.in_(person_ids))
                )
            ).scalars().all()
            persons_map = {p.id: p for p in persons}

        return list(rows), int(total), persons_map

    async def list_groups_owned_by(self, person_id: UUID) -> list[EmailGroup]:
        """Groups where ``person_id`` is listed as an owner and the
        group is still active. Used by offboarding blocker scan.
        """
        rows = (
            await self.db.execute(
                select(EmailGroup)
                .join(EmailGroupOwner, EmailGroupOwner.email_group_id == EmailGroup.id)
                .where(
                    and_(
                        EmailGroupOwner.person_id == person_id,
                        EmailGroup.status == "active",
                    )
                )
                .options(selectinload(EmailGroup.owners))
                .distinct()
            )
        ).scalars().all()
        return list(rows)

    async def is_owner(self, *, group_id: UUID, person_id: UUID) -> bool:
        row = (
            await self.db.execute(
                select(EmailGroupOwner).where(
                    and_(
                        EmailGroupOwner.email_group_id == group_id,
                        EmailGroupOwner.person_id == person_id,
                    )
                )
            )
        ).scalars().first()
        return row is not None

    async def find_group_by_email(
        self, email: str, *, include_deleted: bool = False,
    ) -> EmailGroup | None:
        """Look up a group by email.

        ``include_deleted=False`` (default) skips soft-deleted rows —
        used wherever only live groups matter (directory listing,
        membership ops). NOTE: address *reuse* after delete is NOT
        actually supported: the ``email`` column's UNIQUE index covers
        tombstones too, so ``create_mail_group`` must pass
        ``include_deleted=True`` to catch a deleted-row collision and
        return a clean 409 instead of a DB-level 500. Pass
        ``include_deleted=True`` when the caller needs to see ghosts
        (creation pre-flight, audit-review tooling).
        """
        e = (email or "").strip().lower()
        if not e:
            return None
        stmt = (
            select(EmailGroup).where(EmailGroup.email == e)
            .options(selectinload(EmailGroup.owners))
        )
        if not include_deleted:
            stmt = stmt.where(EmailGroup.status == "active")
        return (await self.db.execute(stmt)).scalars().first()

    # ─── Create mail group ────────────────────────────────────────

    async def create_mail_group(
        self,
        *,
        body: EmailGroupCreateRequest,
        requester: IdentityPerson,
        trace_id: str,
        admin_mode: bool = False,
        actor_roles: list[str] | None = None,
    ) -> MailGroupCreateResult:
        """Mail group creation. Self-service by default.

        Writes both ``employee_requests`` and ``email_groups`` rows,
        then calls Google. If Google fails the whole savepoint rolls
        back. Idempotent at the API level via the standard
        ``Idempotency-Key`` header (enforced upstream).

        ``admin_mode`` (HR / global_admin, via the 邮件组管理 desk)
        changes exactly two things, and nothing else — the Google calls,
        the owner-eligibility gate, the compensation-on-failure path and
        the audit row are all shared, so an admin-created group is
        indistinguishable from a self-service one afterwards:

        * **No per-requester rate limit.** The limit exists to stop an
          employee spraying groups; an admin standing up five team lists
          in one sitting is the tool working as intended.
        * **The creator is not forced to be an owner.** Self-service
          pins the requester as an owner because they are the person
          asking for the group. An admin creating a list *for another
          team* must be able to name that team's owners without becoming
          responsible for the group themselves. ``additional_owner_ids``
          is therefore the complete owner set in admin mode, and at
          least one entry is required — Google demands ≥1 owner, and an
          ownerless group is an offboarding blocker waiting to happen.

        ``admin_mode=True`` REQUIRES ``actor_roles`` and enforces
        ``can_manage_email_groups`` here, not at the router. Two reasons
        a router-body check is not good enough: ``@idempotent`` returns
        a cached envelope *before* the handler body runs, so a replay
        would skip it entirely; and a second caller of this method (an
        MCP tool, a callback) would silently inherit the rate-limit
        bypass and the "creator need not be an owner" rule with no gate
        at all. Putting it here makes it unforgettable.
        """
        if admin_mode:
            from app.core.rbac import can_manage_email_groups

            if not can_manage_email_groups(actor_roles or []):
                raise AppException(
                    code="NOT_AUTHORIZED",
                    message="只有 HR / IT Admin 才能以管理员身份创建邮件组。",
                    status_code=403,
                )
        suffix = body.suffix.strip().lower()
        whitelist = await self.get_suffix_whitelist()
        if suffix not in whitelist:
            raise AppException(
                code="SUFFIX_NOT_ALLOWED",
                message=(
                    f"后缀 {suffix!r} 不在白名单中。允许的后缀:"
                    f" {', '.join(whitelist)}"
                ),
                status_code=400,
            )

        local_part = body.local_part.strip().lower()
        group_email = f"{local_part}@{suffix}"

        # Rate-limit on the REQUESTER (not a global lock). Uses
        # ``SELECT … FOR UPDATE`` on the requester row to serialise
        # concurrent checks from the same user — otherwise two tabs
        # could both slip through with count=0 and create two groups
        # within the rate-limit window.
        hours = 0 if admin_mode else (
            await self._mail_group_create_interval_hours()
        )
        if hours > 0:
            cutoff = datetime.now(UTC) - timedelta(hours=hours)
            locked_requester = (
                await self.db.execute(
                    select(IdentityPerson)
                    .where(IdentityPerson.id == requester.id)
                    .with_for_update()
                )
            ).scalars().first()
            if locked_requester is None:
                raise AppException(
                    code="REQUESTER_NOT_FOUND",
                    message="Requester record not found.",
                    status_code=404,
                )
            recent = (
                await self.db.execute(
                    select(func.count(EmployeeRequest.id)).where(
                        and_(
                            EmployeeRequest.requester_id == requester.id,
                            EmployeeRequest.request_type == "google_mail_group",
                            EmployeeRequest.created_at >= cutoff,
                            # Admin-desk creations are exempt from the
                            # limit, so they must not COUNT toward it
                            # either — otherwise an HR admin who stands
                            # up five team lists at the desk is then
                            # 429'd on their own self-service request,
                            # for groups they created for other teams.
                            func.coalesce(
                                func.json_unquote(
                                    func.json_extract(
                                        EmployeeRequest.detail, "$.via",
                                    ),
                                ),
                                "self_service",
                            ) != "admin",
                        )
                    )
                )
            ).scalar() or 0
            if int(recent) > 0:
                raise AppException(
                    code="RATE_LIMITED",
                    message=(
                        f"你在最近 {hours} 小时内已经创建过邮件组。"
                        "为防止滥用,请稍后再试。"
                    ),
                    status_code=429,
                )

        # Pre-flight: email uniqueness. Hard fail — don't try Google
        # first and hope for 409, because a soft-deleted row would
        # collide at the DB level even if Google would accept.
        existing = await self.find_group_by_email(group_email)
        if existing is not None:
            raise AppException(
                code="EMAIL_GROUP_EXISTS",
                message=f"邮件组 {group_email} 已存在,请先到「邮件组目录」查看。",
                status_code=409,
            )
        # The ``email`` column carries a plain UNIQUE index that also
        # covers soft-deleted (status='deleted') tombstones, so a deleted
        # group still holds its address at the DB level. Catch that here
        # with a clean 409 — otherwise the INSERT below hits errno 1062
        # and surfaces as a bare 500 (the active-only check above can't
        # see the tombstone). Address reuse is intentionally NOT supported
        # under the current schema.
        tombstone = await self.find_group_by_email(
            group_email, include_deleted=True,
        )
        if tombstone is not None:
            raise AppException(
                code="EMAIL_GROUP_EXISTS",
                message=(
                    f"邮件组 {group_email} 曾被创建并已删除,该地址仍被占用。"
                    "请联系 IT 彻底清除后再创建,或改用其它地址。"
                ),
                status_code=409,
            )

        # Validate owners. Self-service pins the requester as an owner;
        # admin mode takes ``additional_owner_ids`` as the complete set
        # (see the ``admin_mode`` note in the docstring).
        owner_ids: list[UUID] = [] if admin_mode else [requester.id]
        seen: set[UUID] = set(owner_ids)
        if admin_mode and not body.additional_owner_ids:
            raise AppException(
                code="OWNER_REQUIRED",
                message=(
                    "管理员建组必须指定至少一个负责人 —— Google 要求每个组"
                    "至少有一个负责人,无负责人的组还会变成离职卡点。"
                ),
                status_code=400,
            )
        for pid in body.additional_owner_ids:
            if pid in seen:
                continue
            seen.add(pid)
            owner_ids.append(pid)
        # Confirm all owners are ACTIVE, non-soft-deleted persons — the
        # previous query only checked existence, so a leaver_pending / left
        # / soft-deleted person id passed (work_email still present) and got
        # written to Google as OWNER + a local owners row, i.e. a departing
        # employee retained owner control of a live list (SOC2 access-gov
        # hole). Mirror ``transfer_owner_on_demand``'s owner-eligibility gate.
        persons = (
            await self.db.execute(
                select(IdentityPerson).where(
                    IdentityPerson.id.in_(owner_ids),
                    IdentityPerson.deleted_at.is_(None),
                    IdentityPerson.employment_status == "active",
                )
            )
        ).scalars().all()
        if len(persons) != len(owner_ids):
            raise AppException(
                code="OWNER_NOT_FOUND",
                message="部分 owner 不存在 / 已删除 / 非在职,请检查成员列表。",
                status_code=400,
            )

        # Google side. If Google says "create OK" but the following
        # local DB flush explodes (disk full, uuid collision, …), we
        # own a group on Google with zero Helix trace — every later
        # flow treats it as non-existent. So we MUST be able to
        # compensate: after Google create succeeds, wrap everything
        # below in try/except and call delete_group on failure.
        driver = await self._get_driver()
        description = (body.description or body.reason or "").strip() or None
        create_resp = await driver.create_group(
            email=group_email, name=body.name, description=description,
        )
        if not create_resp.ok:
            raise AppException(
                code="GOOGLE_CREATE_GROUP_FAILED",
                message=f"Google 创建邮件组失败: {create_resp.error_message}",
                status_code=502,
            )

        try:
            # Add every owner to the group with role=OWNER. We track
            # which ones succeeded so the local owner table matches
            # Google truth (finding #9 of review pass 1): if Google
            # rejects an owner add, we do NOT insert the local owner
            # row — otherwise the directory would falsely show them
            # as an owner.
            owner_persons_by_id = {p.id: p for p in persons}
            successful_owner_ids: list[UUID] = []
            owner_errors: list[str] = []
            for pid in owner_ids:
                person = owner_persons_by_id[pid]
                if not person.work_email:
                    owner_errors.append(
                        f"{person.display_name or pid} 缺少工作邮箱,无法加为 owner"
                    )
                    continue
                # Google Directory API has eventual consistency on a
                # freshly created group: members().insert() right after
                # groups().insert() can hit a replica that hasn't seen
                # the group yet and return 404 "Resource Not Found:
                # groupKey". Retry on 404 with backoff (~7.5s total)
                # before giving up. We deliberately do NOT retry inside
                # the driver — transfer-owner callers want 404 to mean
                # "group is really gone", not a propagation hiccup.
                resp = None
                for attempt, delay in enumerate((0.0, 0.5, 1.0, 2.0, 4.0)):
                    if delay:
                        await asyncio.sleep(delay)
                    resp = await driver.set_group_owner(
                        group_email=group_email,
                        member_email=person.work_email,
                    )
                    if resp.ok or resp.status != 404:
                        break
                    logger.info(
                        "set_group_owner 404 on freshly-created group %s "
                        "for %s (attempt %d) — Google propagation delay, "
                        "retrying",
                        group_email, person.work_email, attempt + 1,
                    )
                if not resp.ok:
                    owner_errors.append(
                        f"{person.display_name or person.work_email}: "
                        f"{resp.error_message}"
                    )
                    continue
                successful_owner_ids.append(pid)
            if not successful_owner_ids:
                # All owners failed — Google requires ≥1 owner per
                # group, so the group is in an unusable state. Roll
                # back by deleting it on Google; surface the errors.
                raise AppException(
                    code="GOOGLE_NO_VALID_OWNER",
                    message=(
                        "未能给任何申请 owner 成功授权,邮件组已被回滚。"
                        f" 详情: {'; '.join(owner_errors)}"
                    ),
                    status_code=502,
                )
            if owner_errors:
                logger.warning(
                    "create_mail_group owner add partial errors for %s: %s",
                    group_email, owner_errors,
                )

            # Local rows — only after Google ops have settled.
            request_number = _generate_request_number("MG")
            er = EmployeeRequest(
                id=uuid4(),
                request_number=request_number,
                request_type="google_mail_group",
                status="completed",
                requester_id=requester.id,
                reason=body.reason,
                detail={
                    "group_email": group_email,
                    "group_name": body.name,
                    "description": description,
                    "owner_ids": [str(x) for x in successful_owner_ids],
                    "owner_errors": owner_errors or None,
                    # Read by the rate-limit filter above, and by the
                    # request history so an HR admin can tell the groups
                    # they stood up for other teams from their own.
                    "via": "admin" if admin_mode else "self_service",
                },
                processed_at=datetime.now(UTC),
            )
            self.db.add(er)
            await self.db.flush()

            group = EmailGroup(
                id=uuid4(),
                email=group_email,
                name=body.name,
                description=description,
                status="active",
                google_group_id=(
                    str(create_resp.data.get("id") or "") or None
                ),
                created_by=requester.id,
                employee_request_id=er.id,
                # NOT stamped. ``last_synced_at`` means "Helix read this
                # group's WHOLE state from Google", and a group created
                # through this path has ``aliases`` and
                # ``direct_members_count`` both NULL — nothing has been
                # read. The cleanup workbench renders this column as
                # staleness evidence beside exactly those two fields, so
                # stamping it here makes a brand-new row look like the
                # freshest-confirmed one in the estate while being the
                # least confirmed. ``_refresh_member_count``,
                # ``transfer_group_owner`` and ``leave_group_for_person``
                # all refuse to stamp it for the same reason; this was
                # the last writer that still did.
                #
                # The first ``google.group_sync`` tick fills it in
                # together with the values it describes.
                last_synced_at=None,
                # A group we just created through the Directory API is
                # hand-maintained by construction — Cloud Identity
                # dynamic groups are created by
                # ``dynamic_service.create`` (which sets ``True``).
                # Leaving this NULL meant "kind never determined", and
                # the 邮件组管理 desk refuses every member / owner /
                # display / alias operation on an undetermined group
                # (``GROUP_KIND_UNKNOWN``, 409) because it might be
                # rule-driven. That made create-then-populate — the
                # desk's headline flow — dead until the next 6-hourly
                # sync tick.
                is_dynamic=False,
            )
            self.db.add(group)
            await self.db.flush()
            for pid in successful_owner_ids:
                self.db.add(
                    EmailGroupOwner(
                        id=uuid4(),
                        email_group_id=group.id,
                        person_id=pid,
                    )
                )
            await self.db.flush()
        except IntegrityError as int_exc:
            # A UNIQUE(email) collision means a parallel request won
            # the race — the group already exists on Google AND has a
            # valid local row owned by that other request. We must
            # NOT call delete_group here (that would destroy the
            # other request's legitimate group). Return a clean 409
            # so the client retries or shows "already exists".
            # Code-review pass 2 finding C1.
            logger.info(
                "create_mail_group lost race for %s: %s", group_email, int_exc,
            )
            # Re-raise as AppException so the router maps to 409.
            raise AppException(
                code="EMAIL_GROUP_EXISTS",
                message=(
                    f"邮件组 {group_email} 已被另一个并发请求创建,"
                    "请刷新目录后重试。"
                ),
                status_code=409,
            ) from int_exc
        except Exception:
            # Any non-collision failure (DB flush error on our owned
            # rows, notification failure, etc.) → compensate by
            # deleting the Google group so we don't orphan it.
            try:
                await driver.delete_group(email=group_email)
            except Exception as compensate_exc:  # noqa: BLE001
                logger.error(
                    "create_mail_group compensation failed for %s: %s. "
                    "Orphan Google group may remain; operator must clean up.",
                    group_email, compensate_exc,
                )
            raise

        # Audit
        await self.audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=requester.id,
            action="email_group.create",
            status="success",
            resource_type="email_group",
            resource_id=str(group.id),
            detail={
                "group_email": group_email,
                "owners": [str(x) for x in successful_owner_ids],
                "already_existed_on_google": create_resp.already_exists,
                # So the operation log can tell an HR-created group from
                # a self-service one without re-deriving it from the
                # actor's roles at read time (roles change; this row
                # must not).
                "via": "admin" if admin_mode else "self_service",
            },
        )

        # Notify each owner (including the requester)
        # The Google Groups admin link format is
        #   https://groups.google.com/a/<domain>/g/<local>
        # so split the email into local-part + domain explicitly and
        # map each by its semantic name to avoid a later refactor
        # flipping them back (see review finding #4).
        group_local, _, group_domain = group_email.partition("@")
        owner_names = ", ".join(
            owner_persons_by_id[pid].display_name or owner_persons_by_id[pid].work_email or str(pid)
            for pid in successful_owner_ids
        )
        for pid in successful_owner_ids:
            person = owner_persons_by_id[pid]
            env = NotificationEnvelope(
                recipient=person,
                title=f"邮件组 {group_email} 已创建",
                body_text=(
                    f"邮件组 {group_email} 已创建成功,你是 owner 之一。\n"
                    f"管理链接: https://groups.google.com/a/{group_domain}/g/{group_local}"
                ),
                template_key="mail_group_created",
                context={
                    "person_name": person.display_name or person.work_email or "",
                    "group_email": group_email,
                    "group_name": body.name,
                    # Em-dash rather than "" for the optional fields: the
                    # template renders "描述:{group_description}" on its own
                    # line and a format string can't drop the label, so an
                    # empty value ships a dangling "描述:" to the owner.
                    "group_description": description or "—",
                    "group_owners": owner_names,
                    "group_domain": group_domain,
                    "group_local": group_local,
                },
                related_resource_type="email_group",
                related_resource_id=group.id,
            )
            await self.notify.dispatch(
                env, trace_id=trace_id, actor_user_id=requester.id,
            )

        return MailGroupCreateResult(request=er, group=group)

    # ─── Delete mail group ────────────────────────────────────────

    async def delete_mail_group(
        self,
        *,
        group_id: UUID,
        actor: IdentityPerson,
        actor_roles: list[str],
        trace_id: str,
        reason: str | None = None,
        acknowledge_blocking: bool = False,
    ) -> EmailGroup:
        """Delete a mail group. Must be owner, or HR, or global admin.

        Local soft-delete (``status='deleted'``) + Google hard-delete.
        If Google delete fails we roll back so the user can retry; a
        half-deleted state (local gone, Google alive) is the harder
        reconciliation. ``reason`` is required by the router (kept
        optional at the service layer for backward compat with
        internal callers) and lands in the audit ``detail`` payload.

        **Blocking-signal gate.** Google's delete is irreversible, and
        the 邮件组管理 desk grew a cleanup workbench that will not sweep
        a group something else depends on — nested inside another group,
        carrying a real second address, holding a live BigQuery dataset
        grant, or named in Helix's own configuration. This per-group
        path sat on the SAME screen with none of those checks, so every
        gate the workbench builds was bypassable one group at a time.
        It now runs the identical signal computation and refuses on a
        BLOCKING verdict unless the caller explicitly acknowledges it
        (which only ``can_delete_any_mail_group`` holders may do — an
        ordinary owner disbanding their own group cannot wave away a
        live dataset grant they may not even know about).

        The signals are recorded in the audit ``detail`` either way, so
        a deliberate override is reconstructable rather than
        indistinguishable from a routine delete.
        """
        # ``select(...).options(...).execute()`` with
        # ``populate_existing()``, NOT ``db.get(..., options=[...])``.
        #
        # ``db.get`` returns the instance straight from the session's
        # IDENTITY MAP when it is already there, and loader options are
        # not applied to an already-present instance — so ``group.owners``
        # below would lazy-load, which under the async session raises
        # ``MissingGreenlet``. Production has never hit it (every request
        # and every MCP call gets a fresh session, so the map is empty),
        # which is exactly what makes it a trap: it is invisible until
        # something shares a session, and then it fails on the
        # OWNERSHIP CHECK of an irreversible delete.
        group = (await self.db.execute(
            select(EmailGroup)
            .where(EmailGroup.id == group_id)
            .options(selectinload(EmailGroup.owners))
            .execution_options(populate_existing=True)
        )).scalar_one_or_none()
        if group is None or group.status != "active":
            raise NotFoundError("EmailGroup", str(group_id))

        is_owner = any(o.person_id == actor.id for o in group.owners)
        if not (is_owner or can_delete_any_mail_group(actor_roles)):
            raise AppException(
                code="NOT_AUTHORIZED",
                message="只有该邮件组的 owner / HR / IT Admin 才能删除。",
                status_code=403,
            )

        # Same checks as the batch workbench, same source. Imported
        # here rather than at module scope: ``cleanup_service`` imports
        # ``manage_service``, which imports this module's models, and a
        # top-level import would close that cycle.
        from app.domains.email_group.cleanup_service import (  # noqa: PLC0415
            EmailGroupCleanupService,
        )

        signals, risk = await EmailGroupCleanupService(
            self.db,
        ).blocking_signals_for(group)
        blocking = [s["key"] for s in signals if s["level"] == "blocking"]
        may_override = can_delete_any_mail_group(actor_roles)
        if blocking and not (acknowledge_blocking and may_override):
            raise AppException(
                code="DELETE_BLOCKED_BY_SIGNALS",
                message=(
                    f"删除邮件组 {group.email} 会影响其它东西,已拒绝:"
                    f"{', '.join(blocking)}。"
                    "请到「邮件组管理 → 清理」查看每条信号的证据;"
                    + (
                        "确认无误后可以在删除时勾选「已确认阻断信号」强制执行。"
                        if may_override
                        else "如确需删除,请联系 IT 管理员。"
                    )
                ),
                status_code=409,
            )

        driver = await self._get_driver()
        resp = await driver.delete_group(email=group.email)
        if not resp.ok:
            raise AppException(
                code="GOOGLE_DELETE_GROUP_FAILED",
                message=f"Google 删除邮件组失败: {resp.error_message}",
                status_code=502,
            )

        group.status = "deleted"
        group.updated_at = datetime.now(UTC)
        await self.db.flush()

        await self.audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=actor.id,
            action="email_group.delete",
            status="success",
            resource_type="email_group",
            resource_id=str(group.id),
            detail={
                "group_email": group.email,
                # Recorded on EVERY delete, so an override is
                # reconstructable instead of looking routine.
                "risk": risk,
                "signals": [s["key"] for s in signals],
                "blocking_signals": blocking,
                "blocking_acknowledged": bool(blocking),
                "already_absent_on_google": bool(
                    resp.data.get("already_absent")
                ),
                "reason": reason,
            },
        )
        return group

    # ─── Alias ───────────────────────────────────────────────────

    async def create_alias(
        self,
        *,
        body: EmailAliasCreateRequest,
        requester: IdentityPerson,
        requester_roles: list[str],
        trace_id: str,
    ) -> AliasCreateResult:
        """Add an alias on either a user mailbox or a mail group.

        Employees may only alias: (a) their own work email, or (b) a
        group where they are an owner. HR and global admin may alias
        any target.
        """
        suffix = body.alias_suffix.strip().lower()
        whitelist = await self.get_suffix_whitelist()
        if suffix not in whitelist:
            raise AppException(
                code="SUFFIX_NOT_ALLOWED",
                message=(
                    f"后缀 {suffix!r} 不在白名单中。允许的后缀:"
                    f" {', '.join(whitelist)}"
                ),
                status_code=400,
            )

        alias_email = f"{body.alias_local_part.strip().lower()}@{suffix}"
        target_email = body.target_email.strip().lower()

        # Permission check
        authorised = False
        if can_alias_for_others(requester_roles):
            authorised = True
        else:
            if body.target_kind == "user":
                if (requester.work_email or "").lower() == target_email:
                    authorised = True
            elif body.target_kind == "group":
                group = await self.find_group_by_email(target_email)
                if group is not None and any(
                    o.person_id == requester.id for o in group.owners
                ):
                    authorised = True
        if not authorised:
            raise AppException(
                code="NOT_AUTHORIZED",
                message=(
                    "员工只能给自己邮箱或自己是 owner 的邮件组添加别名。"
                    "如需给其他对象添加,请联系 HR 或 IT Admin。"
                ),
                status_code=403,
            )

        # Pre-flight: alias must not collide with an existing group
        existing = await self.find_group_by_email(alias_email)
        if existing is not None:
            raise AppException(
                code="ALIAS_COLLIDES_WITH_GROUP",
                message=(
                    f"别名 {alias_email} 与已有邮件组冲突,请换一个别名。"
                ),
                status_code=409,
            )

        driver = await self._get_driver()
        resp = await driver.add_alias(
            target_email=target_email,
            alias_email=alias_email,
            target_kind=body.target_kind,
        )
        if not resp.ok:
            raise AppException(
                code="GOOGLE_ADD_ALIAS_FAILED",
                message=f"Google 添加别名失败: {resp.error_message}",
                status_code=502,
            )

        er = EmployeeRequest(
            id=uuid4(),
            request_number=_generate_request_number("AL"),
            request_type="google_email_alias",
            status="completed",
            requester_id=requester.id,
            reason=body.reason,
            detail={
                "target_kind": body.target_kind,
                "target_email": target_email,
                "alias_email": alias_email,
                "already_existed_on_google": resp.already_exists,
            },
            processed_at=datetime.now(UTC),
        )
        self.db.add(er)
        await self.db.flush()

        await self.audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=requester.id,
            action="email_alias.create",
            status="success",
            resource_type="email_alias",
            resource_id=alias_email,
            detail={
                "target_kind": body.target_kind,
                "target_email": target_email,
                "alias_email": alias_email,
            },
        )

        # When the alias targets a GROUP, mirror the event onto the
        # group's own timeline. Two rows, two resources — the alias has
        # its own lifecycle (``email_alias`` / ``alias_email``) and the
        # group has one, and "what happened to eng@?" must include "an
        # alias was pointed at it". Without the mirror the 邮件组管理
        # operation log (which filters ``resource_type='email_group'``)
        # silently omits every alias ever added through this
        # self-service path. Keeping BOTH rows preserves the existing
        # alias audit trail rather than moving it.
        if body.target_kind == "group":
            aliased_group = await self.find_group_by_email(target_email)
            if aliased_group is not None:
                # Mirror the local alias list too, so the management
                # list shows the new alias before the next sync tick.
                # Only when a mirror exists: writing ``[alias]`` onto a
                # NULL column would claim "synced, and this is the whole
                # set" about a group nobody has alias-synced.
                if aliased_group.aliases is not None:
                    aliased_group.aliases = sorted({
                        *aliased_group.aliases, alias_email,
                    })
                await self.audit.log(
                    trace_id=trace_id,
                    actor_type="user",
                    actor_user_id=requester.id,
                    action="email_group.alias_added",
                    status="success",
                    resource_type="email_group",
                    resource_id=str(aliased_group.id),
                    detail={
                        "group_email": target_email,
                        "alias": alias_email,
                        "via": "self_service_alias_request",
                        "already_existed_on_google": resp.already_exists,
                    },
                )

        # Notify the requester and (if different) the target mailbox
        # owner. For a group target we notify the current owners.
        notify_targets: list[IdentityPerson] = [requester]
        if body.target_kind == "user" and target_email != (requester.work_email or "").lower():
            target_person = (
                await self.db.execute(
                    select(IdentityPerson).where(
                        func.lower(IdentityPerson.work_email) == target_email,
                    )
                )
            ).scalars().first()
            if target_person is not None:
                notify_targets.append(target_person)
        elif body.target_kind == "group":
            group = await self.find_group_by_email(target_email)
            if group is not None:
                owner_pids = [o.person_id for o in group.owners]
                if owner_pids:
                    owners = (
                        await self.db.execute(
                            select(IdentityPerson).where(
                                IdentityPerson.id.in_(owner_pids)
                            )
                        )
                    ).scalars().all()
                    for o in owners:
                        if all(t.id != o.id for t in notify_targets):
                            notify_targets.append(o)

        for person in notify_targets:
            env = NotificationEnvelope(
                recipient=person,
                title=f"邮箱别名 {alias_email} 已创建",
                body_text=(
                    f"别名 {alias_email} 已指向 {target_email}。"
                ),
                template_key="alias_created",
                context={
                    "person_name": person.display_name or person.work_email or "",
                    "target_email": target_email,
                    "alias_email": alias_email,
                    "actor_name": (
                        requester.display_name or requester.work_email or ""
                    ),
                },
                url="/me/settings?tab=aliases",
                related_resource_type="email_alias",
                related_resource_id=er.id,
            )
            await self.notify.dispatch(
                env, trace_id=trace_id, actor_user_id=requester.id,
            )

        return AliasCreateResult(request=er, alias_email=alias_email)

    # ─── Alias listing + deletion (self-service + admin) ─────────

    async def list_aliases_for_person(
        self,
        *,
        person: IdentityPerson,
    ) -> EmailAliasListResponse:
        """Return the alias list for one person.

        Hits Google ``users().get`` once so we can split the merged
        ``aliases`` (editable) from ``nonEditableAliases`` (Google
        domain aliases — read-only, the API rejects deletion). The DB
        column ``identity_persons.email_aliases`` is opportunistically
        refreshed with the union, matching the daily ``google.dept_sync``
        write path so the column stays current between scheduled runs.

        ``person`` is the IdentityPerson row whose aliases to fetch —
        callers are responsible for the authorisation check (employee
        passes their own row; admins may pass any row).
        """
        primary_email = (person.work_email or "").strip().lower()
        if not primary_email:
            # Person has no work email yet — no Google identity, so no
            # aliases. Return an empty list rather than calling Google
            # with a blank userKey (which 400s).
            return EmailAliasListResponse(
                person_id=person.id,
                primary_email="",
                aliases=[],
            )

        driver = await self._get_driver()
        profile = await driver.get_user_profile(primary_email)
        if profile is None:
            # The Helix person isn't in Google (contractor with no
            # workspace identity). Surface an empty list rather than
            # error — the UI shows "No aliases" which is correct.
            return EmailAliasListResponse(
                person_id=person.id,
                primary_email=primary_email,
                aliases=[],
            )

        editable_raw = list(profile.get("aliases") or [])
        non_editable_raw = list(profile.get("nonEditableAliases") or [])

        editable_set = {
            a.strip().lower() for a in editable_raw if a and a.strip()
        }
        non_editable_set = {
            a.strip().lower() for a in non_editable_raw if a and a.strip()
        }
        # If Google returns the same string under both lists (rare but
        # observed when the workspace had a domain alias added then a
        # user-level duplicate was created), prefer the more permissive
        # editable=True so the user can clean it up.
        non_editable_set -= editable_set

        items = [
            EmailAliasItem(alias_email=a, editable=True)
            for a in sorted(editable_set)
        ] + [
            EmailAliasItem(alias_email=a, editable=False)
            for a in sorted(non_editable_set)
        ]

        # Refresh the DB column so list_my_aliases stays cheap on future
        # reads from sync handlers / cross-system lookup.
        new_union = sorted(editable_set | non_editable_set)
        current = sorted(person.email_aliases or [])
        if new_union != current:
            person.email_aliases = new_union or None
            await self.db.flush()

        return EmailAliasListResponse(
            person_id=person.id,
            primary_email=primary_email,
            aliases=items,
        )

    async def delete_alias_for_person(
        self,
        *,
        person: IdentityPerson,
        alias_email: str,
        actor: IdentityPerson,
        actor_roles: list[str],
        reason: str | None,
        trace_id: str,
    ) -> EmployeeRequest:
        """Remove an alias from a person's mailbox.

        Permission contract (verified again here even though the router
        also gates):
          * Self-delete — ``actor.id == person.id``.
          * Admin delete — caller satisfies ``can_alias_for_others``.

        Returns the EmployeeRequest row created for the action so the
        caller can render the request_number in the response. Mirrors
        the ``create_alias`` flow which also writes an EmployeeRequest
        for symmetry on the "我的申请" page.
        """
        primary_email = (person.work_email or "").strip().lower()
        if not primary_email:
            raise AppException(
                code="NO_PRIMARY_EMAIL",
                message=(
                    "目标人员没有 work_email,无法操作邮箱别名。"
                ),
                status_code=400,
            )
        alias = (alias_email or "").strip().lower()
        if "@" not in alias:
            raise AppException(
                code="INVALID_ALIAS",
                message="alias_email must be a full email address",
                status_code=400,
            )

        # Authorisation: self OR admin. Router does its own check too,
        # but the service layer is the source of truth.
        is_self = actor.id == person.id
        if not is_self and not can_alias_for_others(actor_roles):
            raise AppException(
                code="NOT_AUTHORIZED",
                message=(
                    "员工只能删除自己的邮箱别名。"
                    "如需删除其他人的别名,请联系 HR 或 IT Admin。"
                ),
                status_code=403,
            )

        # Pre-flight: check the alias exists on the user, AND that it
        # is editable. We hit Google directly so we don't trust the
        # potentially-stale DB column.
        driver = await self._get_driver()
        profile = await driver.get_user_profile(primary_email)
        if profile is None:
            raise AppException(
                code="GOOGLE_USER_NOT_FOUND",
                message=(
                    f"Google Workspace 中找不到 {primary_email},无法操作"
                    "别名。请联系 IT Admin 检查 HRIS 同步状态。"
                ),
                status_code=404,
            )
        editable_set = {
            a.strip().lower()
            for a in (profile.get("aliases") or [])
            if a and a.strip()
        }
        non_editable_set = {
            a.strip().lower()
            for a in (profile.get("nonEditableAliases") or [])
            if a and a.strip()
        }
        if alias in non_editable_set:
            # nonEditableAliases come from a domain-level alias on the
            # Google Workspace tenant. Deletion fails on the API with
            # 403; reject up-front with a clear message.
            raise AppException(
                code="ALIAS_NOT_EDITABLE",
                message=(
                    f"别名 {alias} 是 Google 域名级自动别名,"
                    "需要 Workspace Admin 在 Google 控制台删除域名别名"
                    "(/admin/domains)。"
                ),
                status_code=409,
            )
        if alias not in editable_set:
            raise AppException(
                code="ALIAS_NOT_FOUND",
                message=(
                    f"别名 {alias} 不在 {primary_email} 的别名列表中。"
                ),
                status_code=404,
            )

        # Drop on Google. ``remove_alias`` is idempotent on 404 — if a
        # parallel race already deleted it, we still record the
        # EmployeeRequest as completed (with already_absent=True) since
        # the desired state is reached.
        resp = await driver.remove_alias(
            target_email=primary_email,
            alias_email=alias,
            target_kind="user",
        )
        if not resp.ok:
            raise AppException(
                code="GOOGLE_REMOVE_ALIAS_FAILED",
                message=f"Google 删除别名失败: {resp.error_message}",
                status_code=502,
            )

        # Refresh the DB column so subsequent reads don't show the
        # stale alias. We re-fetch instead of mutating in-place because
        # nonEditableAliases may have changed concurrently.
        refreshed = await driver.get_user_profile(primary_email)
        if refreshed is not None:
            new_union = sorted({
                a.strip().lower()
                for a in (
                    list(refreshed.get("aliases") or [])
                    + list(refreshed.get("nonEditableAliases") or [])
                )
                if a and a.strip()
            })
            person.email_aliases = new_union or None
        else:
            # Fallback: drop the just-removed alias from the column.
            current = list(person.email_aliases or [])
            person.email_aliases = sorted(
                a for a in current if a != alias
            ) or None

        # Persist the operation as an EmployeeRequest (status=completed)
        # so it shows up on "我的申请" alongside the corresponding
        # ``google_email_alias`` create row.
        er = EmployeeRequest(
            id=uuid4(),
            request_number=_generate_request_number("AL"),
            request_type="google_email_alias",
            status="completed",
            requester_id=actor.id,
            reason=reason,
            detail={
                "action": "delete",
                "target_kind": "user",
                "target_email": primary_email,
                "alias_email": alias,
                "target_person_id": str(person.id),
                "already_absent_on_google": bool(
                    resp.data.get("already_absent")
                ),
            },
            processed_at=datetime.now(UTC),
        )
        self.db.add(er)
        await self.db.flush()

        await self.audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=actor.id,
            action="email_alias.delete",
            status="success",
            resource_type="email_alias",
            resource_id=alias,
            detail={
                "target_kind": "user",
                "target_email": primary_email,
                "target_person_id": str(person.id),
                "alias_email": alias,
                "self_service": is_self,
                "reason": reason,
            },
        )

        # Notification: tell the alias owner the deletion happened. For
        # self-service that's the actor; for admin-initiated we notify
        # the affected mailbox owner so they don't get a "missing alias"
        # surprise next time they look.
        env = NotificationEnvelope(
            recipient=person,
            title=f"邮箱别名 {alias} 已删除",
            body_text=(
                f"别名 {alias} 已从 {primary_email} 中删除。"
                + (f"操作原因:{reason}" if reason else "")
            ),
            template_key="alias_deleted",
            context={
                "person_name": person.display_name or person.work_email or "",
                "target_email": primary_email,
                "alias_email": alias,
                "actor_name": actor.display_name or actor.work_email or "",
                "self_service": is_self,
            },
            url="/me/settings?tab=aliases",
            related_resource_type="email_alias",
            related_resource_id=er.id,
        )
        await self.notify.dispatch(
            env, trace_id=trace_id, actor_user_id=actor.id,
        )
        return er

    # ─── Group memberships (self-service "我的邮件组") ────────────

    async def list_member_groups_for_person(
        self,
        *,
        person: IdentityPerson,
    ) -> MemberGroupListResponse:
        """Return the mail groups ``person`` is a member of.

        Membership is read LIVE from Google (``groups.list(userKey)``) —
        Helix only syncs group metadata + owners, never the full member
        roster, so there is no local table to read. Owner/sole-owner
        flags are then derived from the synced ``email_group_owners``
        rows by joining on the group email, so we avoid an extra
        ``members.get`` Google call per group.

        Callers handle authorisation (employee passes their own row).
        """
        primary_email = (person.work_email or "").strip().lower()
        if not primary_email:
            return MemberGroupListResponse(
                person_id=person.id, primary_email="", groups=[],
            )

        driver = await self._get_driver()
        raw_groups = await driver.list_groups_for_user(primary_email)

        # Map Google group email → local mirror row (active only) so we
        # can attach owner info + group_id. One IN query, no N+1.
        emails = sorted({
            (g.get("email") or "").strip().lower()
            for g in raw_groups
            if g.get("email")
        })
        local_by_email: dict[str, EmailGroup] = {}
        if emails:
            rows = (
                await self.db.execute(
                    select(EmailGroup)
                    .where(
                        func.lower(EmailGroup.email).in_(emails),
                        EmailGroup.status == "active",
                    )
                    .options(selectinload(EmailGroup.owners))
                )
            ).scalars().all()
            local_by_email = {r.email.strip().lower(): r for r in rows}

        items: list[MemberGroupItem] = []
        for g in raw_groups:
            email = (g.get("email") or "").strip().lower()
            if not email:
                continue
            local = local_by_email.get(email)
            owner_ids = (
                [o.person_id for o in local.owners] if local is not None else []
            )
            is_owner = person.id in owner_ids
            sole_owner = is_owner and len(owner_ids) == 1
            items.append(
                MemberGroupItem(
                    email=email,
                    name=g.get("name") or (local.name if local else None),
                    is_owner=is_owner,
                    sole_owner=sole_owner,
                    group_id=local.id if local is not None else None,
                    # NULL is_dynamic (never determined) → treat as static
                    # so we don't block leave on an unconfigured scope.
                    is_dynamic=bool(local.is_dynamic) if local is not None else False,
                )
            )
        items.sort(key=lambda i: i.email)
        return MemberGroupListResponse(
            person_id=person.id,
            primary_email=primary_email,
            groups=items,
        )

    async def leave_group_for_person(
        self,
        *,
        person: IdentityPerson,
        group_email: str,
        actor: IdentityPerson,
        reason: str | None,
        trace_id: str,
    ) -> EmployeeRequest:
        """Remove ``person`` from a mail group they belong to.

        Self-service "退出邮件组". The actor is the leaver themselves
        (the router only exposes this on the ``/me`` surface). Flow:

        1. Sole-owner guard — if ``person`` is the group's ONLY owner
           **per Google's live roster** (not the ``email_group_owners``
           mirror, which only holds owners resolvable to an
           ``identity_persons`` row), refuse: Google requires ≥1 owner
           and the directory's owner invariant would break. Tell them to
           transfer ownership first.
        2. Google ``members.delete`` (idempotent on 404).
        3. Drop the local owner row if one existed (co-owner leaving).
        4. Record an EmployeeRequest + audit row.
        5. Notify the group's remaining owners.
        """
        primary_email = (person.work_email or "").strip().lower()
        if not primary_email:
            raise AppException(
                code="NO_PRIMARY_EMAIL",
                message="你没有工作邮箱,无法操作邮件组成员关系。",
                status_code=400,
            )
        group_email = (group_email or "").strip().lower()
        if "@" not in group_email:
            raise AppException(
                code="INVALID_GROUP_EMAIL",
                message="group_email must be a full email address",
                status_code=400,
            )

        # Local mirror (if synced) — owner info + notification targets.
        local = await self.find_group_by_email(group_email)
        owner_ids = [o.person_id for o in local.owners] if local else []
        is_owner = person.id in owner_ids

        # Dynamic group → self-service leave is a no-op (Google re-computes
        # membership / rejects the delete). Refuse up-front and point the
        # user to IT. ``is_dynamic`` is NULL until the Cloud Identity scope
        # is configured — only block on an explicit True so an
        # unconfigured scope never traps a user.
        if local is not None and local.is_dynamic is True:
            raise AppException(
                code="DYNAMIC_GROUP_CANNOT_LEAVE",
                message=(
                    f"邮件组 {group_email} 是动态组,成员由规则自动管理,"
                    "自助退出无效。如需移出,请联系 IT 管理员。"
                ),
                status_code=409,
            )

        driver = await self._get_driver()

        # Sole-owner guard, decided on the LIVE roster — always, for
        # every group, synced or not.
        #
        # It used to be decided from ``email_group_owners`` whenever a
        # local mirror existed, falling back to the live roster only for
        # an unsynced group. But that mirror is not a faithful copy of
        # Google's owner set: it only holds owners that resolve to an
        # ``identity_persons`` row by ``work_email``, so an external
        # owner (contractor, vendor address) or a departed employee is
        # invisible in it. That made the guard wrong in both directions,
        # and one of them strands a group:
        #
        #  * mirror shows 2, Google has 1 — a stale co-owner row (the
        #    other owner removed in the Google console) let the real sole
        #    owner walk out and leave the group with ZERO owners: the
        #    exact "无人管理" state this guard exists to prevent, and
        #    exactly the shape the 邮件组管理 desk then reports as an
        #    ownerless group nobody can explain.
        #  * mirror shows 0/1, Google has 2 — a co-owner Helix cannot
        #    resolve turned a perfectly safe leave into a hard 409.
        #
        # ``EmailGroupAdminService.revoke_owner`` decides the same
        # question from the live roster too; two paths answering "is this
        # the last owner" from two different sources is how two surfaces
        # end up disagreeing about the same group. One Google call on a
        # rare self-service action is the right price for one answer.
        live_owners = await driver.list_group_owners(group_email)
        if live_owners is None:
            # INDETERMINATE, not "no owners". ``list_group_owners``
            # answers ``None`` for every non-404 failure — 403, 5xx,
            # transport, paginator overrun — and it needs a scope
            # (``group.member.readonly``) the leave itself does not.
            #
            # Refusing outright here would be a serious over-reach: this
            # check now runs on EVERY self-service leave, so one Google
            # blip would make 退出邮件组 return "请联系 IT 管理员" to
            # ordinary members of groups where no ownership question
            # exists at all. Fall back to the mirror instead, and only in
            # the SAFE direction:
            #
            #  * mirror says they hold ownership (or the group is not
            #    mirrored at all, so the mirror can tell us nothing) →
            #    refuse, exactly as before this check existed.
            #  * mirror says they are not an owner → allow. This is the
            #    pre-existing behaviour for a synced group, so an outage
            #    cannot make things worse than they were; the residual
            #    risk (an owner row the mirror never captured) is the one
            #    the live read exists to catch, and it is strictly
            #    smaller than breaking every leave.
            if local is None or (is_owner and len(owner_ids) <= 1):
                raise AppException(
                    code="GROUP_OWNERSHIP_UNVERIFIABLE",
                    message=(
                        f"无法向 Google 核验邮件组 {group_email} 的 owner "
                        "信息,而本地记录显示你可能是唯一的 owner。"
                        "为避免邮件组无人管理,请联系 IT 管理员处理退出。"
                    ),
                    status_code=409,
                )
        else:
            owner_emails = {
                str(o.get("email") or "").strip().lower() for o in live_owners
            }
            owner_emails.discard("")
            # ``is_owner`` stays the LOCAL fact — it drives the mirror
            # cleanup below, which can only delete a row that exists —
            # but the refusal is decided live.
            if primary_email in owner_emails and len(owner_emails) == 1:
                raise AppException(
                    code="SOLE_OWNER_CANNOT_LEAVE",
                    message=(
                        f"你是邮件组 {group_email} 的唯一 owner,直接退出会让"
                        "该组无人管理。请先到「邮件组目录」把 owner 转交给"
                        "其他人,再退出。"
                    ),
                    status_code=409,
                )

        resp = await driver.remove_group_member(
            group_email=group_email,
            member_email=primary_email,
        )
        if not resp.ok:
            raise AppException(
                code="GOOGLE_LEAVE_GROUP_FAILED",
                message=f"Google 退出邮件组失败: {resp.error_message}",
                status_code=502,
            )
        already_absent = bool(resp.data.get("already_absent"))

        # Keep the local owner mirror in lock-step: a co-owner who left
        # is no longer an owner. (Sole owners were blocked above, so we
        # never strand a group with zero local owners here.)
        if local is not None and is_owner:
            owner_row = (
                await self.db.execute(
                    select(EmailGroupOwner).where(
                        and_(
                            EmailGroupOwner.email_group_id == local.id,
                            EmailGroupOwner.person_id == person.id,
                        )
                    )
                )
            ).scalars().first()
            if owner_row is not None:
                await self.db.delete(owner_row)
                # NOT ``last_synced_at``: that column means "when did Helix
                # last read this group's whole state FROM Google", and the
                # cleanup workbench renders it as staleness evidence beside a
                # member count and an alias list this write never touched.
                # Stamping it here makes a one-owner edit certify the whole
                # row as fresh, and the operator's freshest-looking rows
                # become the ones whose data is least confirmed.
                local.updated_at = datetime.now(UTC)
                await self.db.flush()

        er = EmployeeRequest(
            id=uuid4(),
            request_number=_generate_request_number("GL"),
            request_type="google_group_leave",
            status="completed",
            requester_id=actor.id,
            reason=reason,
            detail={
                "group_email": group_email,
                "group_id": str(local.id) if local is not None else None,
                "was_owner": is_owner,
                "already_absent_on_google": already_absent,
            },
            processed_at=datetime.now(UTC),
        )
        self.db.add(er)
        await self.db.flush()

        await self.audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=actor.id,
            action="email_group.leave",
            status="success",
            resource_type="email_group",
            resource_id=str(local.id) if local is not None else group_email,
            detail={
                "group_email": group_email,
                "person_id": str(person.id),
                "was_owner": is_owner,
                "already_absent_on_google": already_absent,
                "reason": reason,
            },
        )

        # Notify the group's remaining owners (exclude the leaver). They
        # manage the group, so a self-service departure is theirs to know
        # about. If the group isn't synced locally we have no owners to
        # notify — the audit row still captures the event.
        remaining_owner_ids = [pid for pid in owner_ids if pid != person.id]
        if remaining_owner_ids:
            owners = (
                await self.db.execute(
                    select(IdentityPerson).where(
                        IdentityPerson.id.in_(remaining_owner_ids)
                    )
                )
            ).scalars().all()
            leaver_label = (
                person.display_name or person.work_email or str(person.id)
            )
            for owner in owners:
                body_text = (
                    f"{leaver_label} 已退出邮件组 {group_email}。"
                )
                if reason:
                    body_text += f"\n\n退出原因:{reason}"
                env = NotificationEnvelope(
                    recipient=owner,
                    title=f"{leaver_label} 已退出邮件组 {group_email}",
                    body_text=body_text,
                    template_key="mail_group_member_left",
                    context={
                        "person_name": (
                            owner.display_name or owner.work_email or ""
                        ),
                        "group_email": group_email,
                        "group_name": (local.name if local else group_email),
                        "leaver_name": leaver_label,
                        "leaver_email": primary_email,
                        # Em-dash, not "" — the template's "退出原因:{reason}"
                        # line can't be conditionally dropped by a format
                        # string (see mail_group_created above).
                        "reason": reason or "—",
                    },
                    related_resource_type="email_group",
                    related_resource_id=local.id if local is not None else None,
                )
                await self.notify.dispatch(
                    env, trace_id=trace_id, actor_user_id=actor.id,
                )

        return er

    # ─── Password reset ──────────────────────────────────────────

    async def reset_own_password(
        self,
        *,
        body: PasswordResetRequest,
        requester: IdentityPerson,
        trace_id: str,
    ) -> PasswordResetResult:
        """Self-service Google password reset with a per-user throttle.

        The new password is generated server-side by the driver (CSPRNG),
        returned once to the caller (for Feishu/email delivery), then
        immediately discarded — never persisted.
        """
        if not (requester.work_email or "").strip():
            raise AppException(
                code="NO_WORK_EMAIL",
                message="你的工作邮箱为空,无法重置 Google 密码。",
                status_code=400,
            )

        hours = await self._password_reset_interval_hours()
        if hours > 0:
            # Race-proof rate limit (pass-2 review finding C2).
            # Without the lock, two parallel tabs could both pass the
            # count-is-zero check and both trigger a reset; the user
            # ends up with two different temp passwords. Take a
            # ``SELECT … FOR UPDATE`` on the most recent reset row
            # (if any) to serialise concurrent checks on the same
            # requester. If no row exists yet, lock a sentinel (the
            # IdentityPerson row) as a barrier.
            cutoff = datetime.now(UTC) - timedelta(hours=hours)
            locked_requester = (
                await self.db.execute(
                    select(IdentityPerson)
                    .where(IdentityPerson.id == requester.id)
                    .with_for_update()
                )
            ).scalars().first()
            if locked_requester is None:
                raise AppException(
                    code="REQUESTER_NOT_FOUND",
                    message="Requester record not found.",
                    status_code=404,
                )
            recent = (
                await self.db.execute(
                    select(func.count(EmployeeRequest.id)).where(
                        and_(
                            EmployeeRequest.requester_id == requester.id,
                            EmployeeRequest.request_type == "google_password_reset",
                            EmployeeRequest.created_at >= cutoff,
                        )
                    )
                )
            ).scalar() or 0
            if int(recent) > 0:
                raise AppException(
                    code="RATE_LIMITED",
                    message=(
                        f"你在最近 {hours} 小时内已经重置过一次密码。"
                        "为防止滥用,请稍后再试。"
                    ),
                    status_code=429,
                )

        return await self._do_password_reset(
            target_person=requester,
            actor=requester,
            reason=body.reason or "self-service reset",
            trace_id=trace_id,
            admin_reset=False,
        )

    async def admin_reset_password(
        self,
        *,
        target_person: IdentityPerson,
        actor: IdentityPerson,
        actor_roles: list[str],
        reason: str,
        trace_id: str,
    ) -> PasswordResetResult:
        """IT Admin resets someone else's password.

        Bypasses the per-user rate limit (it's a human decision with
        justification) but still audit-logs the actor + reason.
        """
        if not has_role(actor_roles, GLOBAL_ADMIN_ROLE):
            raise AppException(
                code="NOT_AUTHORIZED",
                message="只有 IT Admin 可以代其他人重置密码。",
                status_code=403,
            )
        if not (target_person.work_email or "").strip():
            raise AppException(
                code="NO_WORK_EMAIL",
                message="目标用户的工作邮箱为空,无法重置 Google 密码。",
                status_code=400,
            )
        return await self._do_password_reset(
            target_person=target_person,
            actor=actor,
            reason=reason,
            trace_id=trace_id,
            admin_reset=True,
        )

    async def _do_password_reset(
        self,
        *,
        target_person: IdentityPerson,
        actor: IdentityPerson,
        reason: str,
        trace_id: str,
        admin_reset: bool,
    ) -> PasswordResetResult:
        driver = await self._get_driver()
        # Generate the plaintext password HERE (caller-side) instead of
        # relying on the driver to surface it back through ``resp.data``.
        # The driver deliberately strips ``new_password`` from its return
        # payload so the secret never lands in ``provisioning_tasks
        # .raw_response``. Generating here keeps the plaintext in-memory
        # for the notification fan-out without any DB persistence path.
        new_password = GoogleWorkspaceDriver._generate_strong_password()
        resp = await driver.reset_user_password(
            user_email=str(target_person.work_email),
            new_password=new_password,
        )
        if not resp.ok:
            raise AppException(
                code="GOOGLE_PASSWORD_RESET_FAILED",
                message=f"Google 密码重置失败: {resp.error_message}",
                status_code=502,
            )

        er = EmployeeRequest(
            id=uuid4(),
            request_number=_generate_request_number("PW"),
            request_type="google_password_reset",
            status="completed",
            requester_id=target_person.id,
            reason=reason,
            detail={
                "admin_reset": admin_reset,
                "actor_id": str(actor.id),
            },
            processed_at=datetime.now(UTC),
        )
        self.db.add(er)
        await self.db.flush()

        # CRITICAL: the audit log records the EVENT, never the password.
        await self.audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=actor.id,
            action="google_password.reset",
            status="success",
            resource_type="identity_person",
            resource_id=str(target_person.id),
            detail={
                "target_email": target_person.work_email,
                "admin_reset": admin_reset,
                "reason": reason[:200],
            },
        )

        env = NotificationEnvelope(
            recipient=target_person,
            title="你的 Google 密码已重置",
            body_text=(
                f"你的 Google 密码已重置。临时密码: {new_password}\n"
                "请立即登录并自行修改密码。"
            ),
            template_key="password_reset",
            context={
                "person_name": target_person.display_name or target_person.work_email or "",
                "temp_password": new_password,
            },
            related_resource_type="google_password_reset",
            related_resource_id=er.id,
        )
        await self.notify.dispatch(
            env, trace_id=trace_id, actor_user_id=actor.id,
        )

        # SECURITY: the plaintext password briefly lives on the
        # notification_outbox row (in body_text + context.temp_password)
        # during dispatch. Redact both immediately after — a delivered
        # password doesn't need to be retryable, and a failed delivery
        # is better surfaced as a new reset request than a stored
        # plaintext retry. Finding #8 of code-review pass 1.
        await self._scrub_password_from_outbox(er.id)

        return PasswordResetResult(request=er, temp_password=new_password)

    async def _scrub_password_from_outbox(self, request_id: UUID) -> None:
        """Overwrite ``body_text`` / strip ``temp_password`` on any
        outbox rows tied to a password-reset request.

        The notification service keeps the plaintext body so a retry
        can re-render. For password resets we intentionally forbid
        retries (if delivery fails, user re-runs the reset to mint a
        fresh password). So after the first dispatch attempt we zero
        the plaintext.
        """
        from app.notifications.models import NotificationOutbox

        rows = (
            await self.db.execute(
                select(NotificationOutbox).where(
                    and_(
                        NotificationOutbox.related_resource_type
                        == "google_password_reset",
                        NotificationOutbox.related_resource_id == request_id,
                    )
                )
            )
        ).scalars().all()
        for row in rows:
            row.body_text = "[REDACTED — password already delivered]"
            if isinstance(row.context, dict) and "temp_password" in row.context:
                scrubbed = dict(row.context)
                scrubbed.pop("temp_password", None)
                row.context = scrubbed
            # Mark max_attempts reached so the retry worker doesn't
            # pick it up with the scrubbed body.
            row.attempts = max(row.attempts, row.max_attempts)
        await self.db.flush()

    # ─── On-demand owner transfer (directory page) ───────────────

    async def transfer_owner_on_demand(
        self,
        *,
        group_id: UUID,
        from_person_id: UUID,
        to_person_id: UUID,
        actor: IdentityPerson,
        actor_roles: list[str],
        reason: str,
        trace_id: str,
    ) -> EmailGroup:
        """UI-initiated owner transfer.

        Authorization:
          * The actor must be either (a) one of the group's current
            owners, or (b) HR / global admin. Same gate as delete —
            owners can manage their own groups, admins can manage any.
          * ``from_person_id`` must currently be an owner; ``to_person_id``
            must be an active identity with a work_email.

        Mechanics reuse :meth:`transfer_group_owner` so Google + local
        stay in lock-step (add new owner before removing old;
        savepoint-wrapped so invariant "≥1 local owner" holds even on
        partial failure).
        """
        # Use ``select().execute()`` (not ``session.get()``) so the
        # owner-eager-load always fires even when the row is already in
        # the session's identity map from an earlier unrelated call.
        group = (
            await self.db.execute(
                select(EmailGroup)
                .where(EmailGroup.id == group_id)
                .options(selectinload(EmailGroup.owners))
            )
        ).scalar_one_or_none()
        if group is None or group.status != "active":
            raise NotFoundError("EmailGroup", str(group_id))

        is_owner = any(o.person_id == actor.id for o in group.owners)
        is_admin = can_delete_any_mail_group(actor_roles)
        if not (is_owner or is_admin):
            raise AppException(
                code="NOT_AUTHORIZED",
                message="只有该邮件组的 owner / HR / IT Admin 才能转交 owner。",
                status_code=403,
            )
        # A plain co-owner may only transfer away *their own* ownership.
        # Without this, in a group {A,B,C} owner A could pass
        # ``from_person_id=B`` and silently evict co-owner B (who is
        # actually removed Google-side below) — a lateral takeover with
        # no consent and no admin involvement. Only HR / IT Admin
        # (``can_delete_any_mail_group``) may transfer on behalf of
        # another owner. Fixed 2026-06-19 (self-service authz review).
        if not is_admin and from_person_id != actor.id:
            raise AppException(
                code="NOT_AUTHORIZED",
                message="只能转交你自己持有的 owner 身份；如需替他人转交请联系 HR / IT Admin。",
                status_code=403,
            )
        if from_person_id == to_person_id:
            raise AppException(
                code="SAME_PERSON",
                message="新旧 owner 不能是同一个人。",
                status_code=400,
            )
        if not any(o.person_id == from_person_id for o in group.owners):
            raise AppException(
                code="FROM_NOT_OWNER",
                message=(
                    "from_person_id 当前不是该邮件组的 owner。请刷新后重试。"
                ),
                status_code=400,
            )

        from_person = await self.db.get(IdentityPerson, from_person_id)
        to_person = await self.db.get(IdentityPerson, to_person_id)
        if from_person is None:
            raise NotFoundError("IdentityPerson", str(from_person_id))
        # Reject a non-active recipient: ``deleted_at`` alone misses the
        # ``leaver_pending`` / blocked window (person not yet soft-
        # deleted), which would re-orphan the group at the next
        # offboarding pass. The owners-directory endpoint already filters
        # ``employment_status == 'active'``; mirror it here. Fixed
        # 2026-06-19 (self-service authz review).
        if (
            to_person is None
            or to_person.deleted_at is not None
            or to_person.employment_status != "active"
        ):
            raise AppException(
                code="TO_NOT_ACTIVE",
                message="新 owner 不存在或已离职。",
                status_code=400,
            )
        if not (to_person.work_email or "").strip():
            raise AppException(
                code="NO_WORK_EMAIL",
                message="新 owner 缺少工作邮箱,无法转交。",
                status_code=400,
            )

        await self.transfer_group_owner(
            group=group,
            from_person=from_person,
            to_person=to_person,
            actor_user_id=actor.id,
            trace_id=trace_id,
            reason=reason,
            initiator="on_demand",
        )
        return group

    # ─── Offboarding hook: transfer ownership ────────────────────

    async def transfer_group_owner(
        self,
        *,
        group: EmailGroup,
        from_person: IdentityPerson,
        to_person: IdentityPerson,
        actor_user_id: UUID,
        trace_id: str,
        reason: str | None = None,
        initiator: str = "offboarding",
    ) -> None:
        """Move a single owner seat from ``from_person`` → ``to_person``.

        Called by (a) offboarding advance_service when a leaver's
        group is ``accept``-ed by a new owner, and (b) the directory
        page's on-demand transfer endpoint via
        :meth:`transfer_owner_on_demand`.

        Order of operations matters: we ADD the new owner FIRST, then
        remove the leaver. Google requires at least one owner; doing
        it in the opposite order on a single-owner group would leave
        the group temporarily ownerless and the remove would fail.

        ``reason`` lands in the audit row + new-owner notification so
        the receiving side always has context. ``initiator`` tags the
        audit row (``offboarding`` | ``on_demand``) so the admin can
        filter which transfers were manual vs system-driven.
        """
        driver = await self._get_driver()

        if not (to_person.work_email or "").strip():
            raise AppException(
                code="NO_WORK_EMAIL",
                message="接收人缺少工作邮箱,无法转移邮件组 owner。",
                status_code=400,
            )

        # 1) Add new owner (idempotent)
        add_resp = await driver.set_group_owner(
            group_email=group.email,
            member_email=str(to_person.work_email),
        )
        if not add_resp.ok:
            raise AppException(
                code="GOOGLE_SET_OWNER_FAILED",
                message=f"Google 添加新 owner 失败: {add_resp.error_message}",
                status_code=502,
            )

        # 2) Remove old owner (idempotent on 404)
        if from_person.work_email:
            rm_resp = await driver.remove_group_owner(
                group_email=group.email,
                member_email=str(from_person.work_email),
            )
            if not rm_resp.ok:
                # Partial: new owner added, old not removed. Surface
                # the error so the offboarding UI displays it and
                # admin can retry. DON'T swallow — the invariant is
                # broken otherwise.
                raise AppException(
                    code="GOOGLE_REMOVE_OWNER_FAILED",
                    message=(
                        f"Google 移除旧 owner 失败: {rm_resp.error_message}"
                    ),
                    status_code=502,
                )

        # 3) Update local rows — wrap in a savepoint so a partial
        # failure never leaves the group with zero owners locally
        # (Google already has ≥1, but local drift breaks directory
        # search / offboarding scan). Code-review pass 2 finding D1:
        # the INSERT-new-owner happens BEFORE the DELETE-old-owner
        # inside the savepoint, so even if the delete fails the
        # invariant "at least one owner locally" holds.
        with self.db.no_autoflush:
            async with self.db.begin_nested():
                already_to = (
                    await self.db.execute(
                        select(EmailGroupOwner).where(
                            and_(
                                EmailGroupOwner.email_group_id == group.id,
                                EmailGroupOwner.person_id == to_person.id,
                            )
                        )
                    )
                ).scalars().first()
                if already_to is None:
                    self.db.add(
                        EmailGroupOwner(
                            id=uuid4(),
                            email_group_id=group.id,
                            person_id=to_person.id,
                        )
                    )
                    await self.db.flush()

                owner_row = (
                    await self.db.execute(
                        select(EmailGroupOwner).where(
                            and_(
                                EmailGroupOwner.email_group_id == group.id,
                                EmailGroupOwner.person_id == from_person.id,
                            )
                        )
                    )
                ).scalars().first()
                if owner_row is not None:
                    await self.db.delete(owner_row)

                # NOT ``last_synced_at``: that column means "when did Helix
                # last read this group's whole state FROM Google", and the
                # cleanup workbench renders it as staleness evidence beside a
                # member count and an alias list this write never touched.
                # Stamping it here makes a one-owner edit certify the whole
                # row as fresh, and the operator's freshest-looking rows
                # become the ones whose data is least confirmed.
                group.updated_at = datetime.now(UTC)
                await self.db.flush()

        await self.audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=actor_user_id,
            action="email_group.transfer_owner",
            status="success",
            resource_type="email_group",
            resource_id=str(group.id),
            detail={
                "group_email": group.email,
                "from_person_id": str(from_person.id),
                "to_person_id": str(to_person.id),
                "initiator": initiator,
                "reason": reason,
            },
        )

        # Notify the new owner.
        #
        # Coalesce burst (prod 2026-05-13 P0092 incident): a leaver with N
        # mailing lists fanning out to the same successor previously fired
        # N separate Feishu pushes within seconds — recipient (the CEO,
        # tagged as fallback owner) got pinged 10× in one offboarding
        # round. For ``initiator='offboarding'`` we now look back 5 min
        # for an outbox row already addressed to the same recipient with
        # the same template; if found, append this group to its digest
        # context (so the inbox tile shows the full list) and SKIP a fresh
        # Feishu/email dispatch. First push of the burst still goes out
        # immediately so the recipient is informed in real time; follow-up
        # groups arrive silently in the inbox under the digest.
        domain_full = group.email.split("@", 1)[1] if "@" in group.email else ""
        local_part = group.email.split("@", 1)[0] if "@" in group.email else group.email
        from app.notifications.models import NotificationOutbox  # noqa: PLC0415

        existing: NotificationOutbox | None = None
        if initiator == "offboarding":
            cutoff = datetime.now(UTC) - timedelta(minutes=5)
            existing = (
                await self.db.execute(
                    select(NotificationOutbox)
                    .where(
                        NotificationOutbox.recipient_person_id == to_person.id,
                        NotificationOutbox.template_key
                        == "mail_group_owner_transfer",
                        NotificationOutbox.created_at >= cutoff,
                    )
                    .order_by(NotificationOutbox.created_at.desc())
                    .limit(1)
                )
            ).scalar_one_or_none()
            # Only coalesce when the prior burst came from the SAME leaver
            # — two unrelated transfers happening to land on the same
            # recipient within 5 min shouldn't silently merge.
            if existing is not None:
                prior_from = (existing.context or {}).get("previous_owner_id")
                if prior_from and str(from_person.id) != prior_from:
                    existing = None

        if existing is not None:
            ctx = dict(existing.context or {})
            groups = list(ctx.get("groups") or [])
            if not groups and ctx.get("group_email"):
                groups.append(
                    {
                        "email": ctx["group_email"],
                        "name": ctx.get("group_name"),
                    },
                )
            groups.append({"email": group.email, "name": group.name})
            ctx["groups"] = groups
            ctx["group_count"] = len(groups)
            existing.context = ctx
            existing.title = (
                f"{from_person.display_name or from_person.work_email} 的 "
                f"{len(groups)} 个邮件组 owner 权限已转交给你"
            )
            existing.body_text = (
                f"{from_person.display_name or from_person.work_email} 把以下 "
                f"{len(groups)} 个邮件组的 owner 权限转交给了你:\n"
                + "\n".join(f"  · {g['email']}" for g in groups)
                + (f"\n\n转交原因:{reason}" if reason else "")
            )
            await self.db.flush()
            return

        body_text = (
            f"{from_person.display_name or from_person.work_email} 把 "
            f"邮件组 {group.email} 的 owner 权限转交给了你。"
        )
        if reason:
            body_text += f"\n\n转交原因:{reason}"
        env = NotificationEnvelope(
            recipient=to_person,
            title=f"邮件组 {group.email} 的 owner 权限已转交给你",
            body_text=body_text,
            template_key="mail_group_owner_transfer",
            context={
                "person_name": to_person.display_name or to_person.work_email or "",
                "group_email": group.email,
                "group_name": group.name,
                "previous_owner": from_person.display_name or from_person.work_email or "",
                "previous_owner_id": str(from_person.id),
                "group_domain": domain_full,
                "group_local": local_part,
                "reason": reason or "",
                "initiator": initiator,
            },
            related_resource_type="email_group",
            related_resource_id=group.id,
        )
        await self.notify.dispatch(
            env, trace_id=trace_id, actor_user_id=actor_user_id,
        )

        # …and tell the person who LOST it, when somebody else did it to
        # them.
        #
        # ``transfer_owner_on_demand`` lets HR / a global admin pass
        # ``from_person_id != actor.id``, so an admin can strip a named
        # person's ownership — and this method notified only the
        # recipient, so the ex-owner learned nothing and kept believing
        # they were handling the group's requests. That is the same gap
        # ``_notify_owner_revoked`` was added to the admin desk for; a
        # third door onto the same state change must not be the silent
        # one. Skipped when the actor IS the outgoing owner (they just
        # did it, on purpose) and for the offboarding path (the leaver's
        # account is going away and the offboarding surfaces own that
        # conversation).
        if (
            initiator != "offboarding"
            and actor_user_id is not None
            and from_person.id != actor_user_id
            and from_person.id != to_person.id
        ):
            try:
                loss = NotificationEnvelope(
                    recipient=from_person,
                    title=f"你已不再是邮件组 {group.email} 的负责人",
                    body_text=(
                        f"邮件组「{group.name}」({group.email})的 owner "
                        f"权限已转交给 "
                        f"{to_person.display_name or to_person.work_email}。"
                        + (f"\n\n转交原因:{reason}" if reason else "")
                    ),
                    template_key="mail_group_owner_transferred_away",
                    context={
                        "person_name": (
                            from_person.display_name
                            or from_person.work_email or ""
                        ),
                        "group_email": group.email,
                        "group_name": group.name,
                        "new_owner": (
                            to_person.display_name or to_person.work_email or ""
                        ),
                        "reason": reason or "—",
                    },
                    related_resource_type="email_group",
                    related_resource_id=group.id,
                )
                await self.notify.dispatch(
                    loss, trace_id=trace_id, actor_user_id=actor_user_id,
                )
            except Exception:
                logger.exception(
                    "mail-group owner-transferred-away notify failed "
                    "(non-fatal) for %s", from_person.work_email,
                )

    # ─── Join requests (in-Helix owner approval) ─────────────────

    @staticmethod
    def _manage_link(group_email: str) -> str:
        """Google Groups admin console link for a group's owner.

        ``https://groups.google.com/a/<domain>/g/<local>`` — the surface
        an owner uses to manage members / settings directly. Surfaced in
        owner-facing notifications as the "you can also manage it here"
        hint the user asked for.
        """
        local, _, domain = group_email.partition("@")
        if not domain:
            return "https://groups.google.com"
        return f"https://groups.google.com/a/{domain}/g/{local}"

    async def _serialize_join_request(
        self,
        req: EmailGroupJoinRequest,
        *,
        group: EmailGroup | None = None,
        persons: dict[UUID, IdentityPerson] | None = None,
    ) -> dict:
        """Build the wire dict for one join request (group + person labels).

        ``group`` / ``persons`` may be pre-loaded by the caller to avoid
        per-row queries; missing pieces are fetched on demand.
        """
        if group is None:
            group = await self.db.get(EmailGroup, req.email_group_id)
        persons = persons or {}
        requester = persons.get(req.requester_id) or await self.db.get(
            IdentityPerson, req.requester_id,
        )
        decided_by = None
        if req.decided_by_id is not None:
            decided_by = persons.get(req.decided_by_id) or await self.db.get(
                IdentityPerson, req.decided_by_id,
            )
        return {
            "id": str(req.id),
            "email_group_id": str(req.email_group_id),
            "group_email": group.email if group else "",
            "group_name": group.name if group else None,
            "requester": {
                "id": str(req.requester_id),
                "display_name": requester.display_name if requester else None,
                "work_email": requester.work_email if requester else None,
            },
            "reason": req.reason,
            "status": req.status,
            "decided_by_id": str(req.decided_by_id) if req.decided_by_id else None,
            "decided_by_name": (
                (decided_by.display_name or decided_by.work_email)
                if decided_by else None
            ),
            "decided_at": req.decided_at,
            "decision_note": req.decision_note,
            "created_at": req.created_at,
        }

    async def request_join(
        self,
        *,
        group_id: UUID,
        requester: IdentityPerson,
        reason: str | None,
        trace_id: str,
    ) -> tuple[EmailGroupJoinRequest, bool]:
        """Submit a request to JOIN a mail group; routes to its owner(s).

        Returns ``(request, has_owner)`` — ``has_owner`` is False when the
        group currently has no owner, so the caller can tell the requester
        the request was routed to IT (the HR / IT-Admin 待审批 queue)
        rather than to a named owner.

        Guards (fail fast, before any write):
          * group must be active;
          * NOT a Cloud Identity dynamic group (membership is
            rule-computed — a manual add is a no-op, so there's nothing
            for an owner to approve);
          * requester must have a work_email;
          * requester must not already be an owner or a (live Google)
            member;
          * at most one ``pending`` request per (group, requester) — the
            generated ``open_request_key`` unique index is the race-proof
            backstop; we pre-check for a friendly 409.

        On success: writes a ``pending`` row, notifies every current
        owner (with the reason + the groups.google.com management link),
        and audits. The owner decides via :meth:`decide_join`.
        """
        # ``select().execute()`` (NOT ``session.get(..., options=...)``) so the
        # owner eager-load always fires even when the row is already in the
        # session identity map from an earlier query — get() returns the
        # cached instance without applying selectinload, which then lazy-loads
        # ``owners`` in a sync context and trips MissingGreenlet. Same fix as
        # ``transfer_owner_on_demand``.
        group = (
            await self.db.execute(
                select(EmailGroup)
                .where(EmailGroup.id == group_id)
                .options(selectinload(EmailGroup.owners))
            )
        ).scalar_one_or_none()
        if group is None or group.status != "active":
            raise NotFoundError("EmailGroup", str(group_id))
        if group.is_dynamic is True:
            raise AppException(
                code="DYNAMIC_GROUP_CANNOT_JOIN",
                message=(
                    f"邮件组 {group.email} 是动态组,成员由规则自动管理,"
                    "无法自助申请加入。如需加入,请联系 IT 管理员。"
                ),
                status_code=409,
            )
        primary_email = (requester.work_email or "").strip().lower()
        if not primary_email:
            raise AppException(
                code="NO_WORK_EMAIL",
                message="你没有工作邮箱,无法申请加入邮件组。",
                status_code=400,
            )
        if any(o.person_id == requester.id for o in group.owners):
            raise AppException(
                code="ALREADY_OWNER",
                message=f"你已经是邮件组 {group.email} 的 owner,无需申请加入。",
                status_code=409,
            )

        # Live membership check — mirror the leave flow (Helix never
        # mirrors the full member roster, so ask Google). A driver
        # failure here should not hard-block the request: fall through
        # and let the approving owner's add be the idempotent backstop.
        driver = await self._get_driver()
        try:
            raw_groups = await driver.list_groups_for_user(primary_email)
            member_emails = {
                (g.get("email") or "").strip().lower() for g in raw_groups
            }
            if group.email.strip().lower() in member_emails:
                raise AppException(
                    code="ALREADY_MEMBER",
                    message=f"你已经是邮件组 {group.email} 的成员了。",
                    status_code=409,
                )
        except AppException:
            raise
        except Exception:  # noqa: BLE001
            logger.warning(
                "request_join: membership pre-check failed for %s on %s — "
                "proceeding (owner-approve add is idempotent).",
                primary_email, group.email,
            )

        # Friendly duplicate-pending check (the unique index is the
        # race-proof guarantee).
        existing = (
            await self.db.execute(
                select(EmailGroupJoinRequest).where(
                    and_(
                        EmailGroupJoinRequest.email_group_id == group.id,
                        EmailGroupJoinRequest.requester_id == requester.id,
                        EmailGroupJoinRequest.status == "pending",
                    )
                )
            )
        ).scalars().first()
        if existing is not None:
            raise AppException(
                code="JOIN_REQUEST_PENDING",
                message=(
                    f"你已提交过加入 {group.email} 的申请,正在等待 owner 审批。"
                ),
                status_code=409,
            )

        req = EmailGroupJoinRequest(
            id=uuid4(),
            email_group_id=group.id,
            requester_id=requester.id,
            reason=(reason or "").strip() or None,
            status="pending",
        )
        self.db.add(req)
        try:
            await self.db.flush()
        except IntegrityError as exc:
            # Lost the open_request_key race — a parallel tab already
            # created the pending row. Surface a clean 409.
            raise AppException(
                code="JOIN_REQUEST_PENDING",
                message=(
                    f"你已提交过加入 {group.email} 的申请,正在等待 owner 审批。"
                ),
                status_code=409,
            ) from exc

        await self.audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=requester.id,
            action="email_group.join_requested",
            status="success",
            resource_type="email_group",
            resource_id=str(group.id),
            detail={
                "group_email": group.email,
                "requester_id": str(requester.id),
                "join_request_id": str(req.id),
                "reason": req.reason,
            },
        )

        # Notify every current owner: someone wants in, come decide.
        # If the group has NO owner, there is nobody to push to — the
        # request still lands in the HR / IT-Admin 待审批 queue + action
        # inbox (those surface every pending request org-wide to
        # ``can_delete_any_mail_group``), so it is decided by IT rather
        # than rotting. We deliberately do NOT fan out a fresh push to
        # every admin per request (that violates the no-disturb rule);
        # the passive queue is their actionable surface. The requester is
        # told via ``has_owner=False`` so the UI can set expectations.
        owner_ids = [o.person_id for o in group.owners]
        if owner_ids:
            owners = (
                await self.db.execute(
                    select(IdentityPerson).where(IdentityPerson.id.in_(owner_ids))
                )
            ).scalars().all()
            requester_label = (
                requester.display_name or requester.work_email or str(requester.id)
            )
            manage_link = self._manage_link(group.email)
            for owner in owners:
                body_text = (
                    f"{requester_label} 申请加入邮件组 {group.email},等待你审批。"
                )
                if req.reason:
                    body_text += f"\n\n申请理由:{req.reason}"
                body_text += (
                    "\n\n在 Helix「邮件组目录 → 待审批」中通过或拒绝。"
                    f"\n你也可以直接到 Google Groups 管理该组:{manage_link}"
                )
                env = NotificationEnvelope(
                    recipient=owner,
                    title=f"{requester_label} 申请加入邮件组 {group.email}",
                    body_text=body_text,
                    template_key="mail_group_join_requested",
                    context={
                        "person_name": (
                            owner.display_name or owner.work_email or ""
                        ),
                        "group_email": group.email,
                        "group_name": group.name,
                        "requester_name": requester_label,
                        "requester_email": primary_email,
                        "reason": req.reason or "—",
                        "manage_link": manage_link,
                    },
                    # ``/email-groups`` has never been a route — the
                    # mail-group surface is ``/company/mail-groups``.
                    # Emitted the wrong path until 2026-07-28, so every
                    # owner tapping "去复核" landed on a 404.
                    url="/company/mail-groups?review=1",
                    related_resource_type="email_group",
                    related_resource_id=group.id,
                )
                await self.notify.dispatch(
                    env, trace_id=trace_id, actor_user_id=requester.id,
                )

        return req, bool(owner_ids)

    async def decide_join(
        self,
        *,
        request_id: UUID,
        actor: IdentityPerson,
        actor_roles: list[str],
        decision: str,
        note: str | None,
        trace_id: str,
    ) -> EmailGroupJoinRequest:
        """Owner (or HR / IT Admin) approves or rejects a join request.

        On ``approve``: adds the requester to the Google group as a
        MEMBER (idempotent) then flips the row to ``approved``. On
        ``reject``: flips to ``rejected``, no Google call. Either way the
        requester is notified and the decision is audited.

        Authorization mirrors delete / transfer-owner: the actor must be
        one of the group's current owners, or hold
        ``can_delete_any_mail_group`` (HR / IT Admin).
        """
        # Lock the request row so the pending-check → Google add → status
        # flip serializes against a concurrent approve/reject; without it a
        # racing approve+reject could add the member to Google while the row
        # lands 'rejected', leaving silent access.
        req = await self.db.get(
            EmailGroupJoinRequest, request_id, with_for_update=True,
        )
        if req is None:
            raise NotFoundError("EmailGroupJoinRequest", str(request_id))
        if req.status != "pending":
            raise AppException(
                code="JOIN_REQUEST_NOT_PENDING",
                message="该申请已被处理,请刷新后查看最新状态。",
                status_code=409,
            )
        group = (
            await self.db.execute(
                select(EmailGroup)
                .where(EmailGroup.id == req.email_group_id)
                .options(selectinload(EmailGroup.owners))
            )
        ).scalar_one_or_none()
        if group is None or group.status != "active":
            raise NotFoundError("EmailGroup", str(req.email_group_id))

        is_owner = any(o.person_id == actor.id for o in group.owners)
        if not (is_owner or can_delete_any_mail_group(actor_roles)):
            raise AppException(
                code="NOT_AUTHORIZED",
                message="只有该邮件组的 owner / HR / IT Admin 才能审批加入申请。",
                status_code=403,
            )
        # Four-eyes: nobody APPROVES their OWN join request. The pending
        # queue (list_pending_join_requests_for_owner) already excludes the
        # caller's own requests, but the decide endpoint didn't — so a person
        # who requested to join and was LATER made an owner could
        # self-approve, bypassing the dual-control the list was built to
        # enforce. (Reject-own is harmless — that's just withdrawing.)
        if decision == "approve" and str(req.requester_id) == str(actor.id):
            raise AppException(
                code="JOIN_SELF_APPROVAL_FORBIDDEN",
                message="不能审批自己提交的加入申请(需他人四眼复核)。",
                status_code=403,
            )

        requester = await self.db.get(IdentityPerson, req.requester_id)
        requester_label = (
            (requester.display_name or requester.work_email or str(req.requester_id))
            if requester else str(req.requester_id)
        )

        if decision == "approve":
            # Leaver gate — never auto-add someone mid-offboarding, even
            # if they requested while active. Mirror the SOC2 handler.
            # Includes ``deleted_at`` (a soft-deleted person keeps
            # employment_status='active' per the platform gotcha, so the
            # status-only check let an offboarded person be re-added).
            if requester is not None and (
                requester.deleted_at is not None
                or requester.employment_status in ("leaver_pending", "left")
            ):
                raise AppException(
                    code="REQUESTER_LEFT",
                    message=(
                        f"{requester_label} 正在离职流程中"
                        f"(employment_status={requester.employment_status}),"
                        "拒绝自动加入邮件组。如仍需加入请联系 IT。"
                    ),
                    status_code=409,
                )
            member_email = (requester.work_email or "").strip().lower() if requester else ""
            if not member_email:
                raise AppException(
                    code="NO_WORK_EMAIL",
                    message="申请人没有工作邮箱,无法加入邮件组。",
                    status_code=400,
                )
            driver = await self._get_driver()
            resp = await driver.add_group_member(
                group_email=group.email, member_email=member_email,
            )
            if not resp.ok:
                raise AppException(
                    code="GOOGLE_ADD_MEMBER_FAILED",
                    message=f"Google 添加成员失败: {resp.error_message}",
                    status_code=502,
                )

        req.status = "approved" if decision == "approve" else "rejected"
        req.decided_by_id = actor.id
        req.decided_at = datetime.now(UTC)
        req.decision_note = (note or "").strip() or None
        await self.db.flush()

        await self.audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=actor.id,
            action=f"email_group.join_{req.status}",
            status="success",
            resource_type="email_group",
            resource_id=str(group.id),
            detail={
                "group_email": group.email,
                "join_request_id": str(req.id),
                "requester_id": str(req.requester_id),
                "decision": decision,
                "note": req.decision_note,
            },
        )

        # Notify the requester of the outcome.
        if requester is not None:
            if req.status == "approved":
                title = f"你已加入邮件组 {group.email}"
                body_text = (
                    f"你的加入申请已通过,现在你是邮件组 {group.email} 的成员。"
                )
            else:
                title = f"加入邮件组 {group.email} 的申请被拒绝"
                body_text = (
                    f"你申请加入邮件组 {group.email} 的请求已被 owner 拒绝。"
                )
            if req.decision_note:
                body_text += f"\n\nowner 备注:{req.decision_note}"
            env = NotificationEnvelope(
                recipient=requester,
                title=title,
                body_text=body_text,
                template_key=f"mail_group_join_{req.status}",
                context={
                    "person_name": (
                        requester.display_name or requester.work_email or ""
                    ),
                    "group_email": group.email,
                    "group_name": group.name,
                    "decider_name": actor.display_name or actor.work_email or "",
                    "note": req.decision_note or "",
                },
                url="/me/settings?tab=groups",
                related_resource_type="email_group",
                related_resource_id=group.id,
            )
            await self.notify.dispatch(
                env, trace_id=trace_id, actor_user_id=actor.id,
            )

        return req

    async def cancel_join(
        self,
        *,
        request_id: UUID,
        requester: IdentityPerson,
        trace_id: str,
    ) -> EmailGroupJoinRequest:
        """Requester withdraws their own still-pending join request."""
        req = await self.db.get(EmailGroupJoinRequest, request_id)
        if req is None:
            raise NotFoundError("EmailGroupJoinRequest", str(request_id))
        if req.requester_id != requester.id:
            raise AppException(
                code="NOT_AUTHORIZED",
                message="只能撤回你自己提交的加入申请。",
                status_code=403,
            )
        if req.status != "pending":
            raise AppException(
                code="JOIN_REQUEST_NOT_PENDING",
                message="该申请已被处理,无法撤回。",
                status_code=409,
            )
        req.status = "cancelled"
        req.decided_at = datetime.now(UTC)
        await self.db.flush()
        await self.audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=requester.id,
            action="email_group.join_cancelled",
            status="success",
            resource_type="email_group",
            resource_id=str(req.email_group_id),
            detail={"join_request_id": str(req.id)},
        )
        return req

    async def list_pending_join_requests_for_owner(
        self,
        *,
        person: IdentityPerson,
        actor_roles: list[str],
    ) -> list[dict]:
        """Pending join requests the caller is entitled to decide.

        * HR / IT Admin (``can_delete_any_mail_group``) → every pending
          request org-wide.
        * a plain owner → pending requests only on groups they own,
          excluding their own requests (four-eyes; can't self-approve).

        Data-driven duty, not a role — mirrors the cloud-grant approval
        queue. Returns serialized dicts (group + requester labels).
        """
        conds: list = [EmailGroupJoinRequest.status == "pending"]
        if not can_delete_any_mail_group(actor_roles):
            owned_subq = (
                select(EmailGroupOwner.email_group_id)
                .where(EmailGroupOwner.person_id == person.id)
            )
            conds.append(EmailGroupJoinRequest.email_group_id.in_(owned_subq))
            conds.append(EmailGroupJoinRequest.requester_id != person.id)

        rows = (
            await self.db.execute(
                select(EmailGroupJoinRequest)
                .where(and_(*conds))
                .order_by(EmailGroupJoinRequest.created_at.asc())
            )
        ).scalars().all()
        if not rows:
            return []

        group_ids = {r.email_group_id for r in rows}
        person_ids = {r.requester_id for r in rows} | {
            r.decided_by_id for r in rows if r.decided_by_id
        }
        groups = {
            g.id: g for g in (
                await self.db.execute(
                    select(EmailGroup).where(EmailGroup.id.in_(group_ids))
                )
            ).scalars().all()
        }
        persons = {
            p.id: p for p in (
                await self.db.execute(
                    select(IdentityPerson).where(IdentityPerson.id.in_(person_ids))
                )
            ).scalars().all()
        }
        return [
            await self._serialize_join_request(
                r, group=groups.get(r.email_group_id), persons=persons,
            )
            for r in rows
        ]

    async def list_my_join_requests(
        self, *, requester: IdentityPerson,
    ) -> list[dict]:
        """Every join request the caller submitted (any status), newest
        first — backs the requester's tracking view."""
        rows = (
            await self.db.execute(
                select(EmailGroupJoinRequest)
                .where(EmailGroupJoinRequest.requester_id == requester.id)
                .order_by(EmailGroupJoinRequest.created_at.desc())
            )
        ).scalars().all()
        if not rows:
            return []
        group_ids = {r.email_group_id for r in rows}
        groups = {
            g.id: g for g in (
                await self.db.execute(
                    select(EmailGroup).where(EmailGroup.id.in_(group_ids))
                )
            ).scalars().all()
        }
        persons = {requester.id: requester}
        return [
            await self._serialize_join_request(
                r, group=groups.get(r.email_group_id), persons=persons,
            )
            for r in rows
        ]

    # ─── Group member management (owner / admin) ─────────────────

    async def _load_group_for_member_ops(
        self, *, group_id: UUID, actor: IdentityPerson, actor_roles: list[str],
    ) -> EmailGroup:
        group = (
            await self.db.execute(
                select(EmailGroup)
                .where(EmailGroup.id == group_id)
                .options(selectinload(EmailGroup.owners))
            )
        ).scalar_one_or_none()
        if group is None or group.status != "active":
            raise NotFoundError("EmailGroup", str(group_id))
        is_owner = any(o.person_id == actor.id for o in group.owners)
        if not (is_owner or can_delete_any_mail_group(actor_roles)):
            raise AppException(
                code="NOT_AUTHORIZED",
                message="只有该邮件组的 owner / HR / IT Admin 才能管理成员。",
                status_code=403,
            )
        return group

    async def list_members(
        self, *, group_id: UUID, actor: IdentityPerson, actor_roles: list[str],
    ) -> dict:
        """List the members of a group (owner / admin only).

        Reads live from Google. Members are enriched with a Helix
        display_name when their email matches a known person, so the UI
        can show names rather than bare emails.
        """
        group = await self._load_group_for_member_ops(
            group_id=group_id, actor=actor, actor_roles=actor_roles,
        )
        driver = await self._get_driver()
        raw = await driver.list_group_members(group.email)
        emails = sorted({
            (m.get("email") or "").strip().lower()
            for m in raw if m.get("email")
        })
        name_by_email: dict[str, str] = {}
        if emails:
            persons = (
                await self.db.execute(
                    select(IdentityPerson).where(
                        func.lower(IdentityPerson.work_email).in_(emails)
                    )
                )
            ).scalars().all()
            for p in persons:
                key = (p.work_email or "").strip().lower()
                if key:
                    name_by_email[key] = p.display_name or ""
        # EVERY entry, including the ones with no address. Google
        # represents whole-domain membership as ``type: "CUSTOMER"`` with
        # an ``id`` and no ``email``; dropping those made this endpoint
        # disagree with the estate row above it — the row (which counts
        # raw entries) says 1 member while the drawer opened from that
        # row said 「该组没有直接成员」, on exactly the case the raw-count
        # fix was written for. An entry with no address renders under its
        # Google ``id`` instead, which is the truthful thing to show.
        members = [
            {
                "email": (m.get("email") or "").strip().lower() or None,
                "member_id": m.get("id"),
                "role": m.get("role"),
                "type": m.get("type"),
                "status": m.get("status"),
                "display_name": name_by_email.get(
                    (m.get("email") or "").strip().lower()
                ) or None,
            }
            for m in raw
        ]
        members.sort(
            key=lambda m: (m["role"] != "OWNER", m["email"] or "\uffff"),
        )
        return {
            "group_id": str(group.id),
            "group_email": group.email,
            "members": members,
        }

    async def remove_member(
        self,
        *,
        group_id: UUID,
        member_email: str,
        actor: IdentityPerson,
        actor_roles: list[str],
        reason: str | None,
        trace_id: str,
    ) -> dict:
        """Owner / admin removes a member from a group.

        Guards:
          * NOT a dynamic group (membership is rule-computed).
          * The target must not currently be an OWNER — owner removal
            goes through the transfer flow so the group never goes
            ownerless. (Prevents an owner from evicting a co-owner here.)

        The removed person (if resolvable) is notified. Idempotent on a
        Google 404 (already absent).
        """
        group = await self._load_group_for_member_ops(
            group_id=group_id, actor=actor, actor_roles=actor_roles,
        )
        target = (member_email or "").strip().lower()
        if group.is_dynamic is True:
            raise AppException(
                code="DYNAMIC_GROUP_CANNOT_EDIT",
                message=(
                    f"邮件组 {group.email} 是动态组,成员由规则自动管理,"
                    "无法手动移除。请联系 IT 管理员。"
                ),
                status_code=409,
            )
        # Protect owners: check live roles on Google (the local owner
        # mirror only tracks owners Helix knows about).
        driver = await self._get_driver()
        raw = await driver.list_group_members(group.email)
        role_by_email = {
            (m.get("email") or "").strip().lower(): (m.get("role") or "")
            for m in raw
        }
        if target not in role_by_email:
            raise AppException(
                code="NOT_A_MEMBER",
                message=f"{target} 不是邮件组 {group.email} 的成员。",
                status_code=404,
            )
        if role_by_email.get(target) == "OWNER":
            raise AppException(
                code="CANNOT_REMOVE_OWNER",
                message=(
                    "不能在这里移除 owner。请先用「转交 owner」把 owner 身份"
                    "转给其他人,再移除。"
                ),
                status_code=409,
            )

        resp = await driver.remove_group_member(
            group_email=group.email, member_email=target,
        )
        if not resp.ok:
            raise AppException(
                code="GOOGLE_REMOVE_MEMBER_FAILED",
                message=f"Google 移除成员失败: {resp.error_message}",
                status_code=502,
            )
        already_absent = bool((resp.data or {}).get("already_absent"))

        await self.audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=actor.id,
            action="email_group.member_removed",
            status="success",
            resource_type="email_group",
            resource_id=str(group.id),
            detail={
                "group_email": group.email,
                "member_email": target,
                "already_absent_on_google": already_absent,
                "reason": reason,
            },
        )

        # Notify the removed person if we can resolve them.
        removed = (
            await self.db.execute(
                select(IdentityPerson).where(
                    func.lower(IdentityPerson.work_email) == target
                )
            )
        ).scalars().first()
        if removed is not None:
            body_text = (
                f"你已被移出邮件组 {group.email}。"
            )
            if reason:
                body_text += f"\n\n操作原因:{reason}"
            env = NotificationEnvelope(
                recipient=removed,
                title=f"你已被移出邮件组 {group.email}",
                body_text=body_text,
                template_key="mail_group_member_removed",
                context={
                    "person_name": (
                        removed.display_name or removed.work_email or ""
                    ),
                    "group_email": group.email,
                    "group_name": group.name,
                    "actor_name": actor.display_name or actor.work_email or "",
                    "reason": reason or "—",
                },
                related_resource_type="email_group",
                related_resource_id=group.id,
            )
            await self.notify.dispatch(
                env, trace_id=trace_id, actor_user_id=actor.id,
            )

        return {
            "group_email": group.email,
            "member_email": target,
            "already_absent": already_absent,
        }


def _generate_request_number(prefix: str) -> str:
    ts = datetime.now(UTC).strftime("%Y%m%d")
    return f"{prefix}-{ts}-{uuid4().hex[:4].upper()}"