弱点補強: asyncio構造化並行性（TaskGroup × ExceptionGroup × タイムアウト伝播）— gather(return_exceptions=True)の例外オブジェクトがtruthy評価され「在庫チェック失敗」が「適用可能」に化けた事故を、CheckResult型 × TaskGroup × except* × PENDING_RETRYの4点セットで作り直す（MOps 販促システムチーム campaign-eligibility-checker Bad→Good 7点）

📋 コピー"""campaign_eligibility.py — キャンペーン適用可否判定(構造化並行性版)。

Python 3.12+。asyncio.TaskGroup(3.11+) + asyncio.timeout(3.11+)
+ ExceptionGroup(PEP 654)を使用。
"""
from __future__ import annotations

import asyncio
import logging
from dataclasses import dataclass
from typing import Awaitable, Callable, Literal

logger = logging.getLogger(__name__)

DEFAULT_TIMEOUT_SECONDS: float = 3.0
MAX_RETRY_ATTEMPTS: int = 2
RETRY_BACKOFF_BASE_SECONDS: float = 0.5
ELIGIBLE_MEMBER_RANKS: frozenset[str] = frozenset({"gold", "platinum"})

type EligibilityStatus = Literal["DECIDED", "PENDING_RETRY"]


class TransientServiceError(Exception):
    """一時的なネットワーク/5xxエラー。リトライ対象を型で明示(修正6)。"""


@dataclass(frozen=True, slots=True)
class CheckResult:
    """1つの外部サービス呼び出しの結果。

    成功/失敗を明示的なboolフィールドとして持つことで、
    例外オブジェクトのtruthy評価に依存した誤判定(問題1・2)
    を型レベルで排除する。

    Attributes:
        source: 呼び出し元サービス名(ログ・監視での識別子)。
        ok: 呼び出しが成功したかどうか。
        value: 成功時の戻り値。okがFalseならNone。
        error: 失敗時のエラーメッセージ。okがTrueならNone。
    """
    source: str
    ok: bool
    value: object | None
    error: str | None


@dataclass(frozen=True, slots=True)
class EligibilityDecision:
    """キャンペーン適用可否の最終判定結果。

    Attributes:
        campaign_id: 判定対象のキャンペーンID。
        status: "DECIDED"(全チェック成功) / "PENDING_RETRY"(保留、修正7)。
        eligible: statusが"DECIDED"の場合のみ意味を持つ適用可否。
        failed_sources: 失敗したチェック元の一覧。
    """
    campaign_id: str
    status: EligibilityStatus
    eligible: bool
    failed_sources: tuple[str, ...]


async def _run_check(
    source: str,
    call: Callable[[], Awaitable[object]],
    *,
    timeout: float = DEFAULT_TIMEOUT_SECONDS,
    max_attempts: int = MAX_RETRY_ATTEMPTS,
) -> CheckResult:
    """1つの外部サービス呼び出しをタイムアウト・リトライ付きで実行する。

    このコルーチンは意図的に例外を外へ伝播させない。すべての失敗を
    CheckResult(ok=False, ...)として返すことで、呼び出し元の
    asyncio.TaskGroupがsibling taskを巻き込みキャンセルするのを
    「真に予期しないバグ」の場合だけに限定する(修正3・4・5・6)。
    """
    last_error: str | None = None
    for attempt in range(1, max_attempts + 1):
        try:
            async with asyncio.timeout(timeout):
                value = await call()
            return CheckResult(source=source, ok=True, value=value, error=None)
        except TimeoutError:
            last_error = f"timeout after {timeout}s (attempt {attempt}/{max_attempts})"
            logger.warning(
                "eligibility_check_timeout",
                extra={"source": source, "attempt": attempt},
            )  # 修正5: 失敗理由を必ずログに残す
        except TransientServiceError as exc:
            last_error = str(exc)
            logger.warning(
                "eligibility_check_transient_error",
                extra={"source": source, "attempt": attempt, "error": str(exc)},
            )
            await asyncio.sleep(RETRY_BACKOFF_BASE_SECONDS * attempt)  # 修正6
        except Exception as exc:
            logger.error(
                "eligibility_check_failed",
                extra={"source": source, "error": str(exc)},
            )
            return CheckResult(source=source, ok=False, value=None, error=str(exc))

    return CheckResult(source=source, ok=False, value=None, error=last_error)


async def check_campaign_eligibility(
    user_id: str, campaign_id: str
) -> EligibilityDecision:
    """3つの外部サービスを構造化並行性で呼び出し適用可否を判定する(修正4・7)。"""
    try:
        async with asyncio.TaskGroup() as tg:  # 修正4: 構造化並行性
            coupon_task = tg.create_task(
                _run_check(
                    "coupon",
                    lambda: coupon_client.check_coupon_eligibility(user_id, campaign_id),
                )
            )
            inventory_task = tg.create_task(
                _run_check("inventory", lambda: inventory_client.check_stock(campaign_id))
            )
            member_task = tg.create_task(
                _run_check("member_rank", lambda: member_client.get_member_rank(user_id))
            )
    except* TimeoutError as eg:
        # _run_check内でタイムアウトは既に捕捉済み。ここに来るのは想定外のバグのみ
        logger.critical(
            "eligibility_unexpected_timeout_group",
            extra={"errors": [str(e) for e in eg.exceptions]},
        )
        raise
    except* Exception as eg:
        logger.critical(
            "eligibility_unexpected_failure_group",
            extra={"errors": [str(e) for e in eg.exceptions]},
        )
        raise

    results = (coupon_task.result(), inventory_task.result(), member_task.result())
    failed_sources = tuple(r.source for r in results if not r.ok)

    if failed_sources:
        # 修正1・2・7: 部分的失敗は「不適用」と即断せず保留にする
        logger.error(
            "eligibility_check_partial_failure",
            extra={"campaign_id": campaign_id, "failed_sources": failed_sources},
        )
        return EligibilityDecision(
            campaign_id=campaign_id,
            status="PENDING_RETRY",
            eligible=False,
            failed_sources=failed_sources,
        )

    coupon_result, inventory_result, member_result = results
    is_eligible = (
        bool(coupon_result.value)
        and bool(inventory_result.value)
        and member_result.value in ELIGIBLE_MEMBER_RANKS
    )
    return EligibilityDecision(
        campaign_id=campaign_id, status="DECIDED", eligible=is_eligible, failed_sources=()
    )