"""T3.8 回归：并行审记账并发安全（add-only，无 flush 重入）。

旧 bug（T3.7 真 pg E2E 暴露）：三审作为同一 LangGraph superstep 的并行分支，各自
`gateway.run()` → 共用**同一请求 session** 的 `SqlAlchemyLedgerSink.record`
(`session.add` + `await session.flush()`)。`AsyncSession` 非并发安全：`await flush()`
是并行路径里唯一让出控制权的 DB-IO，第二/三审的 flush 撞上「Session is already
flushing」（flush 重入）→ 被 `run_review` 失败隔离吞成 `incomplete` → foreshadow/pace
结果静默丢失。

本测试用一个会**侦测 flush 重入**的假 AsyncSession：`flush` 异步（`await asyncio.sleep`
让出控制权），若两协程同时在 flush 中即抛 `RuntimeError("Session is already flushing")`
——精确复刻 SQLAlchemy 的报错。在旧 add-then-flush 实现下，`asyncio.gather` 三并发
`record` 会触发该重入；修复后 `record` 只 `add`（同步、不让出），不再调 flush，故三行
全部入 session 且零重入。
"""

from __future__ import annotations

import asyncio
import uuid

import pytest
from ww_llm_gateway.ledger import SqlAlchemyLedgerSink
from ww_llm_gateway.types import Scope, Usage


class _FlushReentryDetectingSession:
    """假 AsyncSession：记录 `add` 的行；`flush` 异步且侦测并发重入。

    `add` 同步追加（镜像真 session）。`flush` 进入时若已有协程在 flush 中 → 抛
    `RuntimeError("Session is already flushing")`（复刻 SQLAlchemy 行为），否则
    `await asyncio.sleep(0)` 让出控制权模拟真实 DB-IO 让步点。
    """

    def __init__(self) -> None:
        self.added: list[object] = []
        self.flush_calls = 0
        self._flushing = False

    def add(self, row: object) -> None:
        self.added.append(row)

    async def flush(self) -> None:
        self.flush_calls += 1
        if self._flushing:
            raise RuntimeError("Session is already flushing")
        self._flushing = True
        try:
            await asyncio.sleep(0)  # 让出控制权——并行协程在此交错
        finally:
            self._flushing = False


def _usage(input_tokens: int) -> Usage:
    return Usage(
        provider="deepseek",
        model="deepseek-chat",
        input_tokens=input_tokens,
        output_tokens=1,
        cost_minor=0,
        currency="USD",
    )


def _scope() -> Scope:
    return Scope(user_id=uuid.UUID(int=1), project_id=uuid.UUID(int=2))


async def test_concurrent_record_adds_all_rows_without_flush_reentry() -> None:
    """三并发 `record`（模拟三审并行）→ 三行全部入 session，绝不抛 flush 重入。

    旧 add-then-flush 实现在此会因 `await flush()` 让出控制权而 flush 重入抛错；
    add-only 实现下 `record` 不让出控制权，三行确定性入 session。
    """
    session = _FlushReentryDetectingSession()
    sink = SqlAlchemyLedgerSink(session)  # type: ignore[arg-type]

    await asyncio.gather(
        sink.record(_scope(), _usage(100)),
        sink.record(_scope(), _usage(200)),
        sink.record(_scope(), _usage(300)),
    )

    # 三审用量全部入 session（一次调用一行，记账不丢）
    assert len(session.added) == 3
    input_tokens = sorted(row.input_tokens for row in session.added)  # type: ignore[attr-defined]
    assert input_tokens == [100, 200, 300]
    # add-only：sink 不在并行路径 flush（持久化靠端点/事务 commit）
    assert session.flush_calls == 0


async def test_record_does_not_flush_so_caller_commit_persists() -> None:
    """单次 `record` 只 add 不 flush——契约「网关 ledger 只 add、调用方负责 commit」。"""
    session = _FlushReentryDetectingSession()
    sink = SqlAlchemyLedgerSink(session)  # type: ignore[arg-type]

    await sink.record(_scope(), _usage(42))

    assert len(session.added) == 1
    assert session.flush_calls == 0


@pytest.mark.parametrize("n_concurrent", [2, 5, 10])
async def test_concurrent_record_scales_without_loss(n_concurrent: int) -> None:
    """N 并发 `record` 全部落 session，无丢失、无 flush 重入（不论并发度）。"""
    session = _FlushReentryDetectingSession()
    sink = SqlAlchemyLedgerSink(session)  # type: ignore[arg-type]

    await asyncio.gather(*(sink.record(_scope(), _usage(i + 1)) for i in range(n_concurrent)))

    assert len(session.added) == n_concurrent
    assert session.flush_calls == 0