"""_gate_batch_skills 并行装配的纯逻辑护栏 + runner 级并发编排测试。"""

from __future__ import annotations

import asyncio
import time
from types import SimpleNamespace
from typing import TYPE_CHECKING

import pytest

from app.harness import runner as runner_mod
from app.harness.question_units import build_units
from app.harness.runner import Runner, _assert_disjoint_target_files
from app.harness.validate import ValidationOutcome
from core.types import GeneratedQuestion

if TYPE_CHECKING:
    from pathlib import Path


def test_disjoint_target_files_pass() -> None:
    """各题型映射不同文件:通过。"""
    _assert_disjoint_target_files(
        {"Action Reasoning": "action-reasoning.md", "Counting Problem": "counting-problem.md"}
    )


def test_shared_target_file_fails_fast() -> None:
    """两题型 fallback 到同一文件:并行进化会互相覆盖,必须 fail-fast。"""
    with pytest.raises(RuntimeError, match="default-strategy.md"):
        _assert_disjoint_target_files(
            {"OCR Problems": "default-strategy.md", "Spatial Reasoning": "default-strategy.md"}
        )


# ---------------------------------------------------------------------------
# runner 级并发编排测试(Codex 计划审 I6):
# 用 Runner.__new__ 裸实例 + 假依赖驱动 _gate_batch_skills 四阶段,
# 断言 Phase A gather 并行、Phase D 字母序落账、accept/reject 正确分派。
# ---------------------------------------------------------------------------

_TYPE_A = "Action Reasoning"
_TYPE_C = "Counting Problem"
_TARGET_FILES = {_TYPE_A: "action-reasoning.md", _TYPE_C: "counting-problem.md"}


def _question(qid: str, task_type: str) -> GeneratedQuestion:
    """构造一条真实结构的 single 题目(unit_id 由 __post_init__ 回填)。"""
    return GeneratedQuestion(
        question_id=qid,
        video_id="video-001",
        task_type=task_type,
        question="视频中主角最先做了什么?",
        options=("A. 开门", "B. 关灯", "C. 坐下", "D. 起身"),
        answer="A",
        source_nodes=("L3_0001",),
        difficulty="medium",
    )


def _record(task_type: str) -> SimpleNamespace:
    """构造 EvolutionRecord 替身(仅含 _gate_batch_skills 消费的属性)。"""
    return SimpleNamespace(
        status="accepted",
        original_content="旧 skill 内容",
        evolved_content=f"进化后 skill 内容({task_type})",
        target_file=_TARGET_FILES[task_type],
        clip_info={},
    )


def _outcome(accepted: bool) -> ValidationOutcome:
    """构造真实 ValidationOutcome(一 accept 一 reject 分派用)。"""
    return ValidationOutcome(
        action="accept_confirmed" if accepted else "reject",
        accepted=accepted,
        stop_reason="confirmed" if accepted else "futility",
        e_value=25.0 if accepted else 0.4,
        w=3,
        l=0 if accepted else 3,
        n_used=4,
        delta_hat=0.3 if accepted else -0.2,
        delta_shrunk=0.2 if accepted else -0.1,
        baseline_acc=0.5,
        candidate_acc=0.8 if accepted else 0.3,
        evidence_rows=[{"question_id": "q", "stop_reason": "answered"}],
    )


class _FakeHarnessLog:
    """HarnessLog no-op 替身(上下文管理器协议)。"""

    def __init__(self, *args: object, **kwargs: object) -> None:
        self.args = args

    def __enter__(self) -> _FakeHarnessLog:
        return self

    def __exit__(self, *exc: object) -> bool:
        return False


def _build_runner(tmp_path: Path) -> tuple[Runner, SimpleNamespace, SimpleNamespace]:
    """构造裸 Runner 实例与 state/pools 替身(不触发真实 __init__)。"""
    skills_dir = tmp_path / "skills" / "v1"
    skills_dir.mkdir(parents=True)
    for target in _TARGET_FILES.values():
        (skills_dir / target).write_text("旧 skill 内容", encoding="utf-8")

    runner = Runner.__new__(Runner)
    runner._config = SimpleNamespace(
        workspace_dir=tmp_path,
        edit_budget_start=4,
        edit_budget_end=2,
        appendix_consolidate_threshold=3,
        skill_update_mode="rewrite",
        gate_p_low=0.3,
        gate_p_high=0.85,
        gate_n_max=8,
        gate_e_confirm=20.0,
        gate_e_provisional=5.0,
        gate_w_net_min=2,
        gate_delta_min=0.05,
        gate_lambda_dir=0.5,
        gate_e_rollback=0.05,
        gate_guard_err=0.34,
        concurrency=4,
        max_steps=10,
        skill_mode="live",
    )
    runner._paths = SimpleNamespace(
        skills_dir=skills_dir,
        prompts_dir=tmp_path / "prompts",
        db_path=tmp_path / "harness.db",
    )
    runner._llm = object()
    runner._evolve_llm = object()
    runner._load_evolve_prompts = lambda: None
    runner._current_version = lambda kind: "v1"
    runner._class_baseline_acc = lambda *a, **k: 0.5
    runner._record_run = lambda run_id: None

    questions = {t: _question(f"q-{t[:2].lower()}", t) for t in _TARGET_FILES}
    units = {t: build_units([q])[0] for t, q in questions.items()}
    runner._gate_questions_by_id = {q.question_id: q for q in questions.values()}
    runner._gate_units_by_id = {u.unit_id: u for u in units.values()}
    unit_ids_by_type = {t: [u.unit_id] for t, u in units.items()}

    state = SimpleNamespace(
        gate_cooldown={},
        rejected_buffer={},
        global_step=0,
        correctness={},
        gate_epoch_observed=True,
        baseline_cache=object(),
        gate_pools=SimpleNamespace(
            ladder_for=lambda task_type, exclude, *, p_low, p_high, cold: unit_ids_by_type[
                task_type
            ]
        ),
    )
    pools = SimpleNamespace(baseline_run_id="baseline-run", validation=[])
    return runner, state, pools


def test_gate_batch_parallel_evolve_and_alphabetical_settle(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Phase A 两题型进化时间窗重叠(并行),Phase D 按字母序 accept/reject 分派。"""
    import core.evolution as core_evolution

    runner, state, pools = _build_runner(tmp_path)
    diagnosis = SimpleNamespace(
        skill_case_packs={
            # 故意逆字母序插入,验证排序不是插入序的巧合
            _TYPE_C: SimpleNamespace(task_type=_TYPE_C, failure_cases=[], success_cases=[]),
            _TYPE_A: SimpleNamespace(task_type=_TYPE_A, failure_cases=[], success_cases=[]),
        }
    )
    records = {t: _record(t) for t in _TARGET_FILES}
    outcomes = {_TYPE_A: _outcome(accepted=True), _TYPE_C: _outcome(accepted=False)}

    evolve_windows: dict[str, tuple[float, float]] = {}

    async def fake_evolve_single_skill(
        llm, pack, skill_store, prompts, version, budget, threshold, **kwargs
    ):
        start = time.monotonic()
        await asyncio.sleep(0.05)
        evolve_windows[pack.task_type] = (start, time.monotonic())
        return records[pack.task_type]

    captured: dict[str, object] = {}

    async def fake_validate_skills_concurrent(**kwargs):
        captured.update(kwargs)
        # 逆字母序返回,验证 Phase D 落账顺序来自 sorted 而非 dict 插入序
        return {
            _TYPE_C: outcomes[_TYPE_C],
            _TYPE_A: outcomes[_TYPE_A],
        }

    settle_calls: list[tuple[str, str]] = []
    runner._accept_skill = lambda task_type, *a: settle_calls.append(("accept", task_type))
    runner._record_rejected_skill = lambda buf, task_type, *a: settle_calls.append(
        ("reject", task_type)
    )

    monkeypatch.setattr(core_evolution, "evolve_single_skill", fake_evolve_single_skill)
    monkeypatch.setattr(runner_mod, "validate_skills_concurrent", fake_validate_skills_concurrent)
    monkeypatch.setattr(runner_mod, "HarnessLog", _FakeHarnessLog)
    monkeypatch.setattr(runner_mod, "write_gate_evidence", lambda *a, **k: None)
    monkeypatch.setattr(runner_mod, "write_step_report", lambda *a, **k: None)
    monkeypatch.setattr(runner_mod, "write_quadrant_pairs", lambda *a, **k: None)
    monkeypatch.setattr(runner_mod, "_outcome_to_quadrant_pairs", lambda t, o: [])
    monkeypatch.setattr(runner_mod, "_write_skip_report", lambda *a, **k: None)

    asyncio.run(runner._gate_batch_skills(1, 0, diagnosis, 3, pools, state))

    # (a) 进化时间窗重叠 = gather 真并行(串行时前者 end <= 后者 start)
    win_a, win_c = evolve_windows[_TYPE_A], evolve_windows[_TYPE_C]
    assert win_a[0] < win_c[1] and win_c[0] < win_a[1], f"进化未并行: {evolve_windows}"

    # (b) Phase D 落账顺序 == sorted(题型),且 (c) accept/reject 分派与 outcome 一致
    assert settle_calls == [("accept", _TYPE_A), ("reject", _TYPE_C)]

    # Phase B 装配的 GateSpec 与 Phase C 共享 log 抽查
    specs = captured["specs"]
    assert [s.task_type for s in specs] == sorted(_TARGET_FILES)
    for spec in specs:
        assert spec.target_file == _TARGET_FILES[spec.task_type]
        assert spec.base_skill_content == "旧 skill 内容"
        assert spec.candidate_content == records[spec.task_type].evolved_content
        assert len(spec.units) == 1
        assert "_gate_" in spec.gate_run_prefix
    assert isinstance(captured["log"], _FakeHarnessLog)
    assert callable(captured["run_inference"])


# ---------------------------------------------------------------------------
# 共享 gate_log 的 run_id 契约(真 SQLite,Codex 质量审 C1):
# HarnessLog.insert 缺省用实例 run_id 填充;record 自带 run_id 必须覆盖它,
# 否则连续并发 gate 下所有臂的 predictions 会落成 step 级 run_id,
# validate 按臂 run_id 回读为空 → gate 静默废掉。
# ---------------------------------------------------------------------------


def test_harness_log_insert_record_run_id_overrides_instance(tmp_path: Path) -> None:
    """record 自带 run_id 覆盖实例 run_id;缺省时回落实例 run_id(锁死 enriched.update 语义)。"""
    from app.harness.inference import PREDICTIONS_SCHEMA
    from app.harness.log import HarnessLog

    with HarnessLog(str(tmp_path / "harness.db"), "gate_e1_s0") as log:
        log.create_table("predictions", PREDICTIONS_SCHEMA)
        log.insert(
            "predictions",
            {"run_id": "run_e1_s0_gate_a_base_u0", "question_id": "q1", "prediction": "A"},
        )
        log.insert("predictions", {"question_id": "q2", "prediction": "B"})
        rows = log.query("SELECT question_id, run_id FROM predictions ORDER BY question_id")
    assert [(r["question_id"], r["run_id"]) for r in rows] == [
        ("q1", "run_e1_s0_gate_a_base_u0"),
        ("q2", "gate_e1_s0"),
    ]


def test_inference_prediction_row_carries_arm_run_id(tmp_path: Path) -> None:
    """经共享 gate_log 落库的 prediction 行 run_id 必须是臂 run_id 而非实例 run_id。

    prompt_builder 抛错走异常路径即落库,无需真实 LLM;
    该路径与成功路径共用同一 record 初始 dict,契约一致。
    """
    from app.harness.inference import PREDICTIONS_SCHEMA, _run_single_question
    from app.harness.log import HarnessLog

    def _broken_prompt_builder(qa: GeneratedQuestion) -> tuple[str, str]:
        raise RuntimeError("测试注入:跳过真实推理")

    async def _noop_dispatch(tool_name: str, args: dict, *, context: dict) -> str:
        raise NotImplementedError

    with HarnessLog(str(tmp_path / "harness.db"), "gate_e1_s0") as gate_log:
        gate_log.create_table("predictions", PREDICTIONS_SCHEMA)
        asyncio.run(
            _run_single_question(
                _question("q-arm", _TYPE_A),
                llm=object(),  # prompt_builder 先抛错,不会触达
                tool_dispatch_fn=_noop_dispatch,
                prompt_builder=_broken_prompt_builder,
                log=gate_log,
                max_steps=3,
                plugins=[],
                run_id="run_e1_s0_gate_action-reasoning_cand_u0",
            )
        )
        rows = gate_log.query("SELECT run_id, stop_reason FROM predictions")
    assert len(rows) == 1
    assert rows[0]["run_id"] == "run_e1_s0_gate_action-reasoning_cand_u0"
    assert rows[0]["stop_reason"] == "error"