"""tests/unit/test_gate_unit_scope.py — gate 真实执行路径按 unit 口径跑。 迁移自块序贯版 test_gate_block_unit.py（载体 validate_skill_local，Task 6 删除）： 针对 app/harness/validate.py::validate_skills_concurrent（连续并发 gate 真实路径）， 断言混格阶梯下 gate 按 unit 口径运行：baseline_cache 键含 unit_id、n_used 按 unit 累加、pair_block 折叠 AR pair、逐题 predictions 仍溯源。 核心算法保真 #5（信息阶梯 e-process 口径从 question_id 迁至 unit_id）。 """ from __future__ import annotations from typing import TYPE_CHECKING, Any import pytest from app.harness.gate_ladder import BaselineCache, skill_hash from app.harness.inference import PREDICTIONS_SCHEMA, InferenceResult from app.harness.log import HarnessLog from app.harness.validate import GateSpec, _ladder_units, validate_skills_concurrent from core.evolution import GateParams from core.types import GeneratedQuestion if TYPE_CHECKING: from pathlib import Path _DEFAULT_GATE_PARAMS = GateParams( e_confirm=20.0, e_provisional=3.0, w_net_min=2, delta_min=0.05, lambda_dir=-2.0, e_rollback=10.0, ) def _single(qid: str) -> GeneratedQuestion: """构造非 AR single 题。""" return GeneratedQuestion( question_id=qid, video_id=f"v_{qid}", task_type="temporal", question="Q?", options=("A", "B", "C", "D"), answer="A", source_nodes=(), difficulty="easy", ) def _pair(pair_id: str) -> list[GeneratedQuestion]: """构造 AR 孪生对（original + mirror，共享 pair_id）。""" common = { "video_id": f"v_{pair_id}", "task_type": "temporal", "question": "Q?", "options": ("A", "B", "C", "D"), "answer": "A", "source_nodes": (), "difficulty": "easy", "pair_id": pair_id, "flip_axis": "before_after", } return [ GeneratedQuestion(question_id=f"{pair_id}_o", question_role="pair_original", **common), GeneratedQuestion(question_id=f"{pair_id}_m", question_role="pair_mirror", **common), ] def _setup_workspace(tmp_path: Path) -> Path: """构建最小 workspace（skills/v1/temporal.md）。""" skills_dir = tmp_path / "skills" / "v1" skills_dir.mkdir(parents=True) (skills_dir / "temporal.md").write_text("baseline skill content", encoding="utf-8") return tmp_path def _make_log(workspace: Path, run_id: str = "test_master") -> HarnessLog: """创建 HarnessLog 并建 predictions 表。""" log = HarnessLog(str(workspace / "harness.db"), run_id) log.create_table("predictions", PREDICTIONS_SCHEMA) return log def _insert_predictions(log: HarnessLog, run_id: str, per_q: dict[str, bool]) -> None: """逐题写 predictions（溯源仍逐题）。""" for qid, correct in per_q.items(): log.insert( "predictions", { "run_id": run_id, "video_id": "v0", "question_id": qid, "task_type": "temporal", "prediction": "A" if correct else "Z", "answer": "A", "evidence": "", "reasoning": "", "steps_used": 1, "prompt_tokens": 10, "completion_tokens": 10, "stop_reason": "completed", "steps_json": "[]", }, ) def _make_mock_run_inference( log: HarnessLog, baseline_correct: dict[str, bool], candidate_correct: dict[str, bool], ): """构建 mock RunInferenceFn，按 run_id 的 arm 后缀选基线/候选逐题对错。""" call_log: list[dict[str, Any]] = [] async def mock_fn( questions: list[GeneratedQuestion], *, run_id: str, skills_dir: Path, ) -> InferenceResult: src = baseline_correct if run_id.endswith("_base") else candidate_correct per_q = {q.question_id: src.get(q.question_id, False) for q in questions} _insert_predictions(log, run_id, per_q) call_log.append({"run_id": run_id, "qids": [q.question_id for q in questions]}) total = len(questions) return InferenceResult( run_id=run_id, accuracy=sum(per_q.values()) / total if total else 0.0, total=total, correct=sum(per_q.values()), per_task_type={}, steps_mean=1.0, token_usage={"prompt_tokens": 10, "completion_tokens": 10}, stop_reason_counts={"completed": total}, ) return mock_fn, call_log def _mk_spec(ladder: list[GeneratedQuestion]) -> GateSpec: """由混格阶梯题序构造单题型 GateSpec（units 经 _ladder_units 聚合）。""" return GateSpec( task_type="temporal", target_file="temporal.md", candidate_content="improved skill", base_skill_content="baseline skill content", units=tuple(_ladder_units(ladder)), gate_run_prefix="step1_gate_test", ) async def _run_gate(workspace: Path, spec: GateSpec, mock_fn, log: HarnessLog, cache, params): """跑单 spec 的 validate_skills_concurrent 并返回该题型的 outcome。""" outcomes = await validate_skills_concurrent( workspace_dir=workspace, base_skills_version="v1", specs=[spec], gate_params=params, gate_guard_err=0.5, baseline_cache=cache, prompts_version="p1", run_inference=mock_fn, log=log, concurrency=8, ) return outcomes[spec.task_type] class TestLadderUnits: """_ladder_units：阶梯题序聚合为单元并保持信息阶梯序。""" def test_pure_non_ar_is_identity(self) -> None: """纯非 AR 输入 → 单元序恒等（unit 序==原题序、unit_id==question_id）。""" ladder = [_single("s0"), _single("s1"), _single("s2")] units = _ladder_units(ladder) assert [u.kind for u in units] == ["single", "single", "single"] # unit_id 逐一等于原 question_id，且顺序与输入完全一致 assert [u.unit_id for u in units] == ["s0", "s1", "s2"] assert [q.question_id for u in units for q in u.questions] == ["s0", "s1", "s2"] def test_mixed_preserves_ladder_order(self) -> None: """混格交错输入 → 按单元最早出现下标重排，pair 整锁、不被 single-first 污染。""" p1o, p1m = _pair("p1") # 交错布置：pair 两成员分居 idx 1、3；single 分居 idx 0、2、4 ladder = [_single("s0"), p1o, _single("s1"), p1m, _single("s2")] units = _ladder_units(ladder) # 最早出现下标：s0=0, p1=min(1,3)=1, s1=2, s2=4 → 阶梯序 [s0,p1,s1,s2] assert [u.unit_id for u in units] == ["s0", "p1", "s1", "s2"] # pair 整锁为一个单元（含两成员），不被拆 p1_unit = next(u for u in units if u.unit_id == "p1") assert p1_unit.kind == "pair" assert {q.question_id for q in p1_unit.questions} == {"p1_o", "p1_m"} def test_mixed_differs_from_build_units_default(self) -> None: """混格重排必须纠正 build_units 的 single-first 顺序（否则阶梯序被污染）。""" from app.harness.question_units import build_units p1o, p1m = _pair("p1") ladder = [_single("s0"), p1o, _single("s1"), p1m, _single("s2")] default_order = [u.unit_id for u in build_units(ladder)] ladder_order = [u.unit_id for u in _ladder_units(ladder)] # build_units 把 pair 排到 single 之后；_ladder_units 恢复信息阶梯序 assert default_order == ["s0", "s1", "s2", "p1"] assert ladder_order == ["s0", "p1", "s1", "s2"] assert ladder_order != default_order @pytest.mark.asyncio async def test_gate_n_used_counts_units_not_questions(tmp_path: Path) -> None: """混格阶梯（1 pair + 2 single）→ n_used=3 单元，非 4 题（迁移自块序贯版）。""" workspace = _setup_workspace(tmp_path) log = _make_log(workspace) cache = BaselineCache(workspace / "baseline_cache.json") ladder = [*_pair("p1"), _single("s0"), _single("s1")] # 基线全错、候选全对 → 3 单元齐翻 W=3 baseline = {"p1_o": False, "p1_m": False, "s0": False, "s1": False} candidate = {"p1_o": True, "p1_m": True, "s0": True, "s1": True} mock_fn, _ = _make_mock_run_inference(log, baseline, candidate) accept_params = GateParams( e_confirm=15.0, e_provisional=3.0, w_net_min=2, delta_min=0.05, lambda_dir=-2.0, e_rollback=10.0, ) try: outcome = await _run_gate(workspace, _mk_spec(ladder), mock_fn, log, cache, accept_params) # n_used 按 unit 计（3），W 按 unit 计（3） assert outcome.n_used == 3 assert outcome.w == 3 assert outcome.l == 0 # 证据行按 unit 口径（3 行），ladder_rank 沿阶梯序连续 assert len(outcome.evidence_rows) == 3 assert [r["ladder_rank"] for r in outcome.evidence_rows] == [0, 1, 2] # baseline_cache 键含 unit_id：pair 用 pair_id、single 用 question_id s_hash = skill_hash("baseline skill content") assert cache.get("temporal", s_hash, "p1", "p1") is False assert cache.get("temporal", s_hash, "p1", "s0") is False # 逐题 question_id 不作为 baseline_cache 键（pair 成员未单独缓存） assert cache.get("temporal", s_hash, "p1", "p1_o") is None # 逐题 predictions 仍溯源：候选 per-q 含两 pair 成员 assert set(outcome.candidate_correctness) >= {"p1_o", "p1_m", "s0", "s1"} finally: log.close() @pytest.mark.asyncio async def test_gate_pair_partial_flip_not_counted(tmp_path: Path) -> None: """AR pair 候选仅单向翻(T,F)→单元仍错，W 不被单题污染（迁移自块序贯版）。 前缀逐单元判定下 2 单元小阶梯会在首单元 futility 早停，观测不到 pair 语义； 补 2 个 single 拉长阶梯：4 单元中 3 个 single 翻转 → W=3（pair 不计入）， candidate_acc = 3/4。 """ workspace = _setup_workspace(tmp_path) log = _make_log(workspace) cache = BaselineCache(workspace / "baseline_cache.json") ladder = [*_pair("p1"), _single("s0"), _single("s1"), _single("s2")] baseline = {"p1_o": False, "p1_m": False, "s0": False, "s1": False, "s2": False} # pair 只翻一半（p1_o 对、p1_m 错）→ 单元 AND 仍错；singles 全翻对 candidate = {"p1_o": True, "p1_m": False, "s0": True, "s1": True, "s2": True} mock_fn, _ = _make_mock_run_inference(log, baseline, candidate) try: outcome = await _run_gate( workspace, _mk_spec(ladder), mock_fn, log, cache, _DEFAULT_GATE_PARAMS ) # 只有 single 单元翻转，pair 单元不计 W（保真 #5：不被 P/Q 单题污染） assert outcome.w == 3 assert outcome.l == 0 assert outcome.n_used == 4 # candidate_acc 分母按 unit（4 单元，1 对）→ 3/4 assert outcome.candidate_acc == 0.75 finally: log.close() @pytest.mark.asyncio async def test_gate_baseline_cache_hit_by_unit(tmp_path: Path) -> None: """基线缓存按 unit_id 预填充 → 基线侧全命中不发起推理（迁移自块序贯版）。 阶梯补长到 4 单元避免首单元 futility 早停，覆盖 pair 与 single 两种 unit 键。 """ workspace = _setup_workspace(tmp_path) log = _make_log(workspace) cache = BaselineCache(workspace / "baseline_cache.json") ladder = [*_pair("p1"), _single("s0"), _single("s1"), _single("s2")] s_hash = skill_hash("baseline skill content") # 按 unit_id 预填充（pair→pair_id，single→question_id），全错 for unit_id in ("p1", "s0", "s1", "s2"): cache.put("temporal", s_hash, "p1", unit_id, False) baseline = {"p1_o": False, "p1_m": False, "s0": False, "s1": False, "s2": False} candidate = {"p1_o": True, "p1_m": True, "s0": True, "s1": True, "s2": True} mock_fn, call_log = _make_mock_run_inference(log, baseline, candidate) try: outcome = await _run_gate( workspace, _mk_spec(ladder), mock_fn, log, cache, _DEFAULT_GATE_PARAMS ) base_calls = [c for c in call_log if c["run_id"].endswith("_base")] assert base_calls == [], "unit 键全命中不应发起基线推理" assert outcome.n_used == 4 finally: log.close()