"""soak harness 纯函数层测试(M2 设计 §8;T10/T11)。

语料还原器与不变量断言函数是纯函数,用真实 schema 的二次构造微型库测试
(真实场景优先: schema 与 data/soak 实库逐字一致)。
"""

import json
import sqlite3

import pytest

from tools.soak.corpus import (
    assemble_frame_messages,
    chs_image_messages,
    load_replay_payloads,
    load_trace_chains,
)
from tools.soak.scenarios import weighted_mix


def _mini_harness_db(path):
    """predictions 表 schema 与 data/soak/harness.db 逐字一致的微型库。"""
    conn = sqlite3.connect(path)
    conn.execute(
        "CREATE TABLE predictions (run_id TEXT, timestamp TEXT, video_id TEXT,"
        " question_id TEXT, task_type TEXT, prediction TEXT, answer TEXT, evidence TEXT,"
        " reasoning TEXT, steps_used INTEGER, prompt_tokens INTEGER,"
        " completion_tokens INTEGER, stop_reason TEXT, steps_json TEXT)"
    )
    steps = [
        {
            "thought": f"思考{i}",
            "tool_call": {"tool": "view_node", "args": {"i": i}},
            "tool_output": f"输出{i}" * 50,
        }
        for i in range(3)
    ]
    conn.execute(
        "INSERT INTO predictions VALUES ('r1','t','vid1','q1','synopsis','A','A','','',3,1,1,'stop',?)",
        (json.dumps(steps, ensure_ascii=False),),
    )
    conn.execute(
        "INSERT INTO predictions VALUES ('r1','t','vid2','q2','synopsis','B','B','','',0,1,1,'stop','[]')"
    )
    conn.commit()
    conn.close()


def _mini_telemetry_db(path):
    conn = sqlite3.connect(path)
    conn.execute(
        "CREATE TABLE llm_calls (call_id TEXT PRIMARY KEY, parent_call_id TEXT,"
        " session_id TEXT, model_name TEXT, provider TEXT, messages TEXT, response TEXT,"
        " thinking TEXT, prompt_tokens INTEGER, completion_tokens INTEGER, latency_ms INTEGER,"
        " ttft_ms REAL, max_inter_token_ms REAL, cache_hit INTEGER, error TEXT, created_at TEXT)"
    )
    msgs = json.dumps([{"role": "user", "content": "回放负载"}])
    conn.execute(
        "INSERT INTO llm_calls VALUES ('c1',NULL,'s1','m','p',?,'r','',1,1,1,NULL,NULL,0,NULL,'t')",
        (msgs,),
    )
    conn.execute(
        "INSERT INTO llm_calls VALUES ('c2',NULL,'s1','m','p','not-json','r','',1,1,1,NULL,NULL,0,NULL,'t')"
    )
    conn.commit()
    conn.close()


class TestTraceChains:
    def test_cumulative_chain_shapes(self, tmp_path):
        db = tmp_path / "h.db"
        _mini_harness_db(db)
        chains = load_trace_chains(db)
        assert len(chains) == 1  # steps_used=0 的行剔除
        chain = chains[0]
        assert len(chain) == 3  # 每步一个累积快照
        # 快照单调增长: system+user 起步,每步追加 assistant+user
        assert [len(snap) for snap in chain] == [4, 6, 8]
        first = chain[0]
        assert first[0]["role"] == "system" and first[1]["role"] == "user"
        assert first[2]["role"] == "assistant" and "思考0" in first[2]["content"]
        assert first[3]["role"] == "user" and "输出0" in first[3]["content"]
        # 后续快照是前缀扩展(累积语义)
        assert chain[1][:4] == chain[0]

    def test_max_steps_cap(self, tmp_path):
        db = tmp_path / "h.db"
        _mini_harness_db(db)
        assert [len(s) for s in load_trace_chains(db, max_steps=2)[0]] == [4, 6]


class TestReplayPayloads:
    def test_loads_and_skips_unparseable(self, tmp_path):
        db = tmp_path / "t.db"
        _mini_telemetry_db(db)
        payloads = load_replay_payloads(db)
        assert len(payloads) == 1  # 坏 JSON 行剔除
        assert payloads[0][0]["content"] == "回放负载"


class TestImageAssembly:
    def test_frame_messages_two_tiers(self, tmp_path):
        vid = tmp_path / "vid1" / "frames"
        vid.mkdir(parents=True)
        for i in range(6):
            (vid / f"f{i}.jpg").write_bytes(b"\xff\xd8fakejpeg" + bytes([i]))
        msgs = assemble_frame_messages(tmp_path, n_frames=4, rng=lambda: 0.0)
        parts = msgs[0]["content"]
        images = [p for p in parts if p["type"] == "image_url"]
        assert len(images) == 4
        assert images[0]["image_url"]["url"].startswith("data:image/jpeg;base64,")

    def test_chs_single_image_with_instruction(self, tmp_path):
        (tmp_path / "chs_0001.jpg").write_bytes(b"\xff\xd8fake")
        msgs = chs_image_messages(tmp_path / "chs_0001.jpg")
        parts = msgs[0]["content"]
        assert any(p["type"] == "image_url" for p in parts)
        text = next(p["text"] for p in parts if p["type"] == "text")
        assert "JSON" in text  # 固定结构化指令


class TestWeightedMix:
    def test_ratio_approximation(self):
        import random

        rng = random.Random(42).random
        weights = {"P1": 0.1, "P2": 0.2, "P3": 0.5, "P5": 0.2}
        picks = [weighted_mix(weights, rng) for _ in range(10_000)]
        for key, w in weights.items():
            assert picks.count(key) / 10_000 == pytest.approx(w, abs=0.05)

    def test_invalid_weights_rejected(self):
        with pytest.raises(ValueError):
            weighted_mix({"P1": 0.0}, lambda: 0.5)


# ═══════════ T11: 记分板硬不变量(纯函数) ═══════════

from tools.soak.scoreboard import (  # noqa: E402
    inv_call_ids_unique,
    inv_rows_match_calls,
    inv_rpm_never_exceeded,
    inv_rss_stable,
    structured_success_rate,
)


def _row(
    call_id,
    *,
    source="s1",
    created_at="2026-07-20T10:00:00",
    error=None,
    session="r-p3",
    latency_ms=0,
    cache_hit=0,
):
    return {
        "call_id": call_id,
        "source_name": source,
        "created_at": created_at,
        "error": error,
        "session_id": session,
        "latency_ms": latency_ms,
        "cache_hit": cache_hit,
    }


class TestInvariants:
    def test_rows_match_calls(self):
        rows = [_row("a"), _row("b")]
        inv_rows_match_calls(rows, expected_calls=2)
        with pytest.raises(AssertionError):
            inv_rows_match_calls(rows, expected_calls=3)

    def test_call_ids_unique(self):
        inv_call_ids_unique([_row("a"), _row("b")])
        with pytest.raises(AssertionError):
            inv_call_ids_unique([_row("a"), _row("a")])

    def test_rpm_minute_bucket(self):
        # 固定分钟窗口口径(与限流器 int(sec/60) 同源;滑动窗会对合法的
        # 跨窗背靠背流量误报,不采用)。击穿用例取窗口中部,避免贴边歧义。
        rows = [
            _row("a", created_at="2026-07-20T10:00:10"),
            _row("b", created_at="2026-07-20T10:00:50"),
            _row("c", created_at="2026-07-20T10:01:10"),
        ]
        inv_rpm_never_exceeded(rows, {"s1": 2})
        rows.append(_row("d", created_at="2026-07-20T10:00:30"))
        with pytest.raises(AssertionError):
            inv_rpm_never_exceeded(rows, {"s1": 2})

    def test_rpm_ignores_unlimited_sources(self):
        inv_rpm_never_exceeded([_row(str(i)) for i in range(100)], {"s1": 0})

    def test_rpm_buckets_by_admit_time_not_completion(self):
        # 准入相隔 100s(各窗 1 次),长延迟让完成时刻挤进同一分钟——
        # 完成时刻聚桶会误报(P6 伪击穿教训之一)
        rows = [
            _row("a", created_at="2026-07-20T10:02:30", latency_ms=120_000),  # 准入 10:00:30
            _row("b", created_at="2026-07-20T10:02:35", latency_ms=25_000),  # 准入 10:02:10
        ]
        inv_rpm_never_exceeded(rows, {"s1": 1})

    def test_rpm_server_clock_offset_dedistorts(self):
        # P6 实跑复现: Redis 服务器钟快 31s,本机分钟内 10 次准入
        # 实为两个服务器窗口各 5 次(+31 后: 37..59 前窗 | 63..90 后窗)
        secs = [6, 19, 21, 26, 28, 32, 42, 47, 52, 59]
        rows = [_row(str(i), created_at=f"2026-07-20T10:00:{s:02d}") for i, s in enumerate(secs)]
        inv_rpm_never_exceeded(rows, {"s1": 5}, clock_offset_s=31.0)

    def test_rpm_true_breach_still_caught(self):
        # 6 次准入全落窗口中部(无贴边歧义),限额 5 → 必须击穿
        rows = [
            _row(str(i), created_at=f"2026-07-20T10:00:{s:02d}")
            for i, s in enumerate([10, 20, 25, 30, 40, 50])
        ]
        with pytest.raises(AssertionError):
            inv_rpm_never_exceeded(rows, {"s1": 5})

    def test_rpm_excludes_cache_hits(self):
        # P6 实跑教训之二: 缓存命中在限流闸之前返回,未耗 RPM 名额也未打
        # 网关,但遥测按"遥测必录"记行且署原源名——不得计入 RPM 口径
        rows = [
            _row(str(i), created_at=f"2026-07-20T10:00:{10 + i * 5:02d}", cache_hit=1)
            for i in range(8)
        ]
        rows.append(_row("real", created_at="2026-07-20T10:00:30"))
        inv_rpm_never_exceeded(rows, {"s1": 1})

    def test_rpm_boundary_jitter_exempted(self):
        # 贴边行(±2s 采样噪声)在邻窗有余量时可归邻窗,不算击穿
        rows = [
            _row(str(i), created_at=f"2026-07-20T10:00:{s:02d}")
            for i, s in enumerate([59, 10, 20, 30, 40, 50])
        ]
        inv_rpm_never_exceeded(rows, {"s1": 5})

    def test_rpm_jitter_is_side_aware(self):
        # verifier 对抗样例①: 左贴边行只能借左邻余量——左邻已满、右邻
        # 全空时,7 行(2 行贴左界)仍是真实击穿,不得借右邻豁免
        rows = [
            _row(f"L{i}", created_at=f"2026-07-20T10:00:{s:02d}")
            for i, s in enumerate([10, 20, 30, 40, 50])  # 左邻(10:00)满额 5
        ]
        rows += [
            _row(f"M{i}", created_at=f"2026-07-20T10:01:{s:02d}")
            for i, s in enumerate([0, 1, 10, 20, 30, 40, 50])  # 本窗 7 行,2 行贴左界
        ]
        with pytest.raises(AssertionError):
            inv_rpm_never_exceeded(rows, {"s1": 5})

    def test_rpm_jitter_room_not_double_claimed(self):
        # verifier 对抗样例②: 两个超限窗不得重复认领中间窗的同 1 个余量
        rows = [
            _row(f"A{i}", created_at=f"2026-07-20T10:00:{s:02d}")
            for i, s in enumerate([10, 20, 30, 40, 50, 59])  # 6 行,1 行贴右界
        ]
        rows += [
            _row(f"B{i}", created_at=f"2026-07-20T10:01:{s:02d}")
            for i, s in enumerate([10, 20, 30, 40])  # 中间窗 4 行,仅 1 余量
        ]
        rows += [
            _row(f"C{i}", created_at=f"2026-07-20T10:02:{s:02d}")
            for i, s in enumerate([1, 10, 20, 30, 40, 50])  # 6 行,1 行贴左界
        ]
        with pytest.raises(AssertionError):
            inv_rpm_never_exceeded(rows, {"s1": 5})

    def test_rss_stable(self):
        inv_rss_stable([100.0, 105.0, 110.0], max_growth_mb=50.0)
        with pytest.raises(AssertionError):
            inv_rss_stable([100.0, 400.0], max_growth_mb=50.0)

    def test_structured_success_rate(self):
        rows = [
            _row("a"),
            _row("b", error="ResultInvalidError: x"),
            _row("c", session="r-p1"),  # 非 P3 剔除
        ]
        assert structured_success_rate(rows, session_suffix="-p3") == pytest.approx(0.5)


class TestHarnessMeasurement:
    """P6 伪击穿修复(2026-07-21): RSS 采当前值、分发有界。"""

    def test_rss_mb_reports_current_not_peak(self):
        import resource
        import sys as _sys

        from tools.soak.run_soak import _rss_mb

        ballast = bytearray(300 * 1024 * 1024)
        ballast[::4096] = b"x" * len(ballast[::4096])  # 触页,确保计入 RSS
        divisor = 1e6 if _sys.platform == "darwin" else 1024
        peak_mb = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / divisor
        del ballast
        # ru_maxrss 含 300MB 压舱物峰值;当前 RSS 应已显著回落
        assert _rss_mb() < peak_mb - 100

    async def test_paced_dispatch_bounds_inflight_and_honors_stop(self):
        import asyncio

        from tools.soak.run_soak import _paced_dispatch

        gauge = {"now": 0, "max": 0, "done": 0}
        dispatched: list[int] = []

        async def _work():
            gauge["now"] += 1
            gauge["max"] = max(gauge["max"], gauge["now"])
            await asyncio.sleep(0.02)
            gauge["now"] -= 1
            gauge["done"] += 1

        async def _gen():
            for i in range(10):
                yield "chat", {"i": i}

        inflight = await _paced_dispatch(
            _gen(),
            sem=asyncio.Semaphore(2),
            spawn=lambda kind, kwargs: _work(),
            should_stop=lambda: len(dispatched) >= 6,
            on_dispatched=lambda: dispatched.append(1),
        )
        await asyncio.gather(*inflight, return_exceptions=True)
        assert gauge["max"] <= 2  # 并发名额先占后建任务
        assert gauge["done"] == 6  # should_stop 在预算命中处截停分发


class TestLiveInvariantsAndCaps:
    def test_capped_budget_clamps_to_signed_values(self):
        from tools.soak.scoreboard import capped_budget

        assert capped_budget("P1", 9999) == 500
        assert capped_budget("P6", 100) == 100

    async def test_gate_reenterable_detects_hung_probe(self):
        from tools.soak.scoreboard import inv_gate_reenterable

        class _Decision:
            def __init__(self, allowed, state, is_probe=False):
                self.allowed = allowed
                self.state = state
                self.is_probe = is_probe
                self.retry_after_s = 30.0

        class _Gate:
            def __init__(self, decision):
                self._d = decision
                self.released = 0

            async def try_enter(self, name, owner):
                return self._d

            async def release_probe(self, entry):
                self.released += 1

        healthy = _Gate(_Decision(True, "closed"))
        await inv_gate_reenterable(healthy, ["s1"])
        probe_gate = _Gate(_Decision(True, "half_open", is_probe=True))
        await inv_gate_reenterable(probe_gate, ["s1"])
        assert probe_gate.released == 1  # 探针当场归还,不留新悬挂
        await inv_gate_reenterable(_Gate(_Decision(False, "open")), ["s1"])  # 冷却合法
        with pytest.raises(AssertionError, match="悬挂"):
            await inv_gate_reenterable(_Gate(_Decision(False, "half_open")), ["s1"])

    def test_fault_errors_presence(self):
        from tools.soak.scoreboard import inv_fault_errors_present

        rows = [_row("a", source="bad_1", error="SourceDeadError: 401"), _row("b")]
        inv_fault_errors_present(rows, fault_source_names=["bad_1"])
        inv_fault_errors_present(rows, fault_source_names=[])  # 未配故障源直接通过
        with pytest.raises(AssertionError):
            inv_fault_errors_present([_row("c")], fault_source_names=["bad_1"])


class TestAnyErrors:
    def test_presence_and_absence(self):
        from tools.soak.scoreboard import inv_any_errors

        inv_any_errors([_row("a", error="TransientError: x"), _row("b")])
        with pytest.raises(AssertionError, match="故障未生效"):
            inv_any_errors([_row("c")])
