diff --git a/tests/integration/test_v3_contract_e2e.py b/tests/integration/test_v3_contract_e2e.py
new file mode 100644
index 0000000..939d14f
--- /dev/null
+++ b/tests/integration/test_v3_contract_e2e.py
@@ -0,0 +1,379 @@
+"""question-gen v3 Phase 1 契约全链路集成回归（Task 13）。
+
+把 T3-T12 建立的 QuestionUnit(single|pair) 契约在 build_pools → build_batches →
+run_inference 全链路上做一次端到端回归护栏。构造「混格题库」（若干 AR pair + 若干
+非 AR single）真实驱动三个生产入口，逐条锁定六条契约：
+
+1. pair 不拆：混格题库经 build_pools 后，任一 pair 的 original+mirror 落在同一池。
+2. 同批：诊断池经 build_batches 后，任一 pair 的两题在同一个 batch。
+3. 双向 AND 聚合：pair P 对 Q 错 → 该 unit 计错；P 对 Q 对 → 计对。
+4. unit 粒度 total/correct：N single + M pair 的库，total == N + M（pair 计 1）。
+5. 孤儿被剔：只有 original 没有 mirror 的悬挂 pair → 聚合剔除 + 告警、不计入 total。
+6. 非 AR byte-identical：纯 single 题库 build_batches 输出与旧逐题黄金参照逐字节一致。
+
+黄金参照 helper（_reference_build_batches / _ids）与最小构造 helper（_single /
+_pair）复用 tests/unit/test_non_ar_byte_identical.py，不重造。fake 只在 LLM/工具/
+prompt 外部依赖层，build_pools/build_batches/run_inference/聚合逻辑全部真跑。
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+from unittest.mock import AsyncMock
+
+import pytest
+from loguru import logger
+
+from app.harness.batching import build_batches
+from app.harness.inference import InferenceResult, run_inference
+from app.harness.log import HarnessLog
+from app.harness.pools import build_pools
+from core.types import GeneratedQuestion, LLMResponse
+from tests.unit.test_non_ar_byte_identical import (
+    _ids,
+    _pair,
+    _reference_build_batches,
+    _single,
+)
+
+# ── 外部依赖 fake（仅 LLM/工具/prompt，被测链路真跑） ──────────────────
+
+
+def _make_question(
+    question_id: str,
+    *,
+    task_type: str = "RETRIEVAL",
+    answer: str = "B",
+    pair_id: str | None = None,
+    question_role: str = "single",
+    flip_axis: str | None = None,
+    video_id: str = "v1",
+) -> GeneratedQuestion:
+    """构造混格题目；pair_id 非空时视为孪生对成员（共享 flip_axis）。"""
+    return GeneratedQuestion(
+        question_id=question_id,
+        video_id=video_id,
+        task_type=task_type,
+        question="测试问题",
+        options=("A. 选项A", "B. 选项B", "C. 选项C", "D. 选项D"),
+        answer=answer,
+        source_nodes=("L1_001",),
+        difficulty="medium",
+        pair_id=pair_id,
+        question_role=question_role,
+        flip_axis=flip_axis if pair_id else None,
+    )
+
+
+def _make_llm_response(answer: str = "B") -> LLMResponse:
+    """构造 submit_answer 场景的 LLMResponse（固定预测某答案）。"""
+    content = json.dumps(
+        {
+            "reflect": {"observation": "找到答案"},
+            "plan": {"next_step": "提交"},
+            "action": {
+                "tool": "submit_answer",
+                "args": {"answer": answer, "evidence": "证据", "reasoning": "推理"},
+            },
+        }
+    )
+    return LLMResponse(
+        content=content,
+        thinking="思考",
+        model="test-model",
+        provider="test",
+        prompt_tokens=100,
+        completion_tokens=50,
+        latency_ms=200,
+        ttft_ms=30.0,
+        max_inter_token_ms=5.0,
+        cache_hit=False,
+        call_id="test-call-001",
+    )
+
+
+async def _stub_tool_dispatch(
+    tool_name: str, args: dict[str, Any], *, context: dict[str, Any]
+) -> str:
+    """测试用工具调度：只认 submit_answer，未知工具 fail-loud。"""
+    if tool_name == "submit_answer":
+        return "答案已提交"
+    raise ValueError(f"未知工具: {tool_name}")
+
+
+def _stub_prompt_builder(qa: GeneratedQuestion) -> tuple[str, str]:
+    """测试用 prompt 构建。"""
+    return "系统提示词", f"用户问题: {qa.question}"
+
+
+@pytest.fixture
+def harness_log(tmp_path: Any, request: Any) -> HarnessLog:
+    """临时 HarnessLog 实例（run_id 标记为 test-run）。"""
+    db_path = str(tmp_path / f"harness_{id(request)}.db")
+    log = HarnessLog(db_path, "test-run")
+    yield log
+    log.close()
+
+
+# ── 断言 helper（把跨池/跨批的成员定位与 pair 不拆判定收口一处） ────────
+
+
+def _membership(labeled_groups: list[tuple[str, list[GeneratedQuestion]]]) -> dict[str, str]:
+    """把带标签的分组（池或 batch）展平成 question_id → 组标签，并断言组间互斥。
+
+    参数:
+        labeled_groups: (标签, 题目列表) 列表；标签即池名或 batch 下标字符串。
+
+    返回:
+        question_id → 所属组标签映射。
+
+    关键实现:
+        同一 question_id 落入两个组即 fail（三池互斥 / 单题不重复入批的前提被破坏）。
+    """
+    location: dict[str, str] = {}
+    for label, group in labeled_groups:
+        for q in group:
+            assert q.question_id not in location, (
+                f"{q.question_id} 同属 {location.get(q.question_id)} 与 {label}（组未互斥）"
+            )
+            location[q.question_id] = label
+    return location
+
+
+def _pair_together(location: dict[str, str], pair_id: str, *, allow_absent: bool) -> bool:
+    """判定某 pair 的 original+mirror 是否整锁同组，返回是否被采样进组。
+
+    参数:
+        location: question_id → 组标签（来自 _membership）。
+        pair_id: 孪生对标识（成员 id 约定为 f"{pair_id}_o" / f"{pair_id}_m"）。
+        allow_absent: True 时允许整对都未进组（build_pools 场景，采样可不命中）；
+            False 时要求必进组（build_batches 全错场景，必进批）。
+
+    返回:
+        该对是否被完整采样进某组（用于调用方统计非空护栏）。
+
+    异常:
+        AssertionError: 半只进组（被拆）或两题分属不同组（跨组）。
+    """
+    o_loc, m_loc = location.get(f"{pair_id}_o"), location.get(f"{pair_id}_m")
+    if allow_absent and o_loc is None and m_loc is None:
+        return False
+    assert o_loc is not None and m_loc is not None, (
+        f"pair {pair_id} 被拆：original@{o_loc} mirror@{m_loc}（半只进组）"
+    )
+    assert o_loc == m_loc, f"pair {pair_id} 跨组：original@{o_loc} mirror@{m_loc}"
+    return True
+
+
+async def _run_capturing_warnings(
+    questions: list[GeneratedQuestion], log: HarnessLog
+) -> tuple[InferenceResult, list[str]]:
+    """用固定预测 "B" 的 fake LLM 真跑 run_inference，并捕获 WARNING 日志。
+
+    参数:
+        questions: 待推理的混格题库。
+        log: HarnessLog 实例。
+
+    返回:
+        (InferenceResult, 捕获到的 WARNING 消息列表)。
+    """
+    llm = AsyncMock()
+    llm.chat.return_value = _make_llm_response(answer="B")
+
+    captured: list[str] = []
+    sink_id = logger.add(captured.append, level="WARNING", format="{message}")
+    try:
+        result = await run_inference(
+            questions,
+            llm=llm,
+            tool_dispatch_fn=_stub_tool_dispatch,
+            prompt_builder=_stub_prompt_builder,
+            log=log,
+            run_id="run-v3-contract",
+            concurrency=4,
+            max_steps=10,
+            skill_mode="auto",
+        )
+    finally:
+        logger.remove(sink_id)
+    return result, captured
+
+
+# ── 契约 1：pair 全程不拆，两题永不跨池 ───────────────────────────────
+
+
+class TestPairNeverSplitAcrossPools:
+    """混格题库经 build_pools 后，任一 pair 的 original+mirror 落在同一池。"""
+
+    def _mixed_benchmark(self) -> tuple[list[GeneratedQuestion], dict[str, bool]]:
+        """24 non-AR single + 8 AR pair 混格库，附单元级基线对错。
+
+        single：偶数下标对、奇数下标错（12 对 12 错）。
+        pair：前 4 对两题皆对（单元对），后 4 对 original 错（单元错）。
+        """
+        singles = [_single(f"s{i}", task_type="RETRIEVAL") for i in range(24)]
+        correctness: dict[str, bool] = {q.question_id: (i % 2 == 0) for i, q in enumerate(singles)}
+
+        pair_questions: list[GeneratedQuestion] = []
+        for i in range(8):
+            original, mirror = _pair(f"p{i}", task_type="AR")
+            pair_questions.extend([original, mirror])
+            correctness[original.question_id] = i < 4
+            correctness[mirror.question_id] = True
+        return singles + pair_questions, correctness
+
+    def test_no_pair_lands_in_two_pools(self) -> None:
+        """任一 pair 两题要么同池、要么都未被采样，绝不分属不同池。"""
+        questions, correctness = self._mixed_benchmark()
+        pools = build_pools(
+            questions,
+            correctness,
+            diag_cfg={
+                "size": 10,
+                "correct_ratio": 0.5,
+                "task_types": None,
+                "seed": 1,
+                "min_per_class": None,
+            },
+            val_cfg={
+                "size": 8,
+                "correct_ratio": 0.5,
+                "task_types": None,
+                "seed": 1,
+                "min_per_class": None,
+            },
+            test_cfg={"size": 6, "seed": 1},
+            baseline_run_id="baseline",
+        )
+
+        pool_of = _membership(
+            [
+                ("diagnosis", pools.diagnosis),
+                ("validation", pools.validation),
+                ("test", pools.test),
+            ]
+        )
+        sampled = sum(_pair_together(pool_of, f"p{i}", allow_absent=True) for i in range(8))
+        # 非空护栏：确保确有 pair 被采样进池，断言不是空转
+        assert sampled > 0, "无 pair 进入任一池，pair-不拆断言未被实质覆盖"
+
+
+# ── 契约 2：pair 整锁同批 ─────────────────────────────────────────────
+
+
+class TestPairStaysInSameBatch:
+    """诊断池经 build_batches 后，任一 pair 的两题落在同一个 batch。"""
+
+    def _all_wrong_diagnosis(self) -> tuple[list[GeneratedQuestion], dict[str, bool]]:
+        """全错混格诊断池：6 non-AR single + 4 AR pair，皆错以确保全部进批。"""
+        singles = [_single(f"s{i}", task_type="RETRIEVAL") for i in range(6)]
+        correctness: dict[str, bool] = {q.question_id: False for q in singles}
+
+        items: list[GeneratedQuestion] = list(singles)
+        for i in range(4):
+            original, mirror = _pair(f"p{i}", task_type="AR")
+            items.extend([original, mirror])
+            correctness[original.question_id] = False
+            correctness[mirror.question_id] = False
+        return items, correctness
+
+    def test_pair_members_share_batch(self) -> None:
+        """全错混格诊断池分批后，每对两题同批（pair 占 2 容量整锁不拆）。"""
+        items, correctness = self._all_wrong_diagnosis()
+        batches, selected = build_batches(
+            items, correctness, batch_size=6, min_class_per_batch=2, seed=3, correct_ratio=0.0
+        )
+
+        assert selected == len(items), "全错单元应全部进批（selected == 展开题数）"
+
+        batch_of = _membership([(str(idx), batch) for idx, batch in enumerate(batches)])
+        in_batches = sum(_pair_together(batch_of, f"p{i}", allow_absent=False) for i in range(4))
+        assert in_batches == 4, "并非全部 pair 都进批，同批断言未实质覆盖"
+
+
+# ── 契约 3+4+5：run_inference 双向 AND + unit 粒度 total + 孤儿剔除 ────
+
+
+class TestInferenceUnitAggregationEndToEnd:
+    """混格题库真跑 run_inference：双向 AND、unit total/correct、孤儿剔除+告警。"""
+
+    def _mixed_questions(self) -> list[GeneratedQuestion]:
+        """3 single（皆对）+ pairA（两题皆对）+ pairB（P 对 Q 错）+ 孤儿 pair。
+
+        LLM 固定预测 "B"：answer=="B" 即对、answer=="A" 即错，从而在单次推理里
+        制造出 unit 级双向 AND 的对/错两种结果。
+        """
+
+        def _pair_member(qid: str, pid: str, role: str, answer: str) -> GeneratedQuestion:
+            return _make_question(
+                qid,
+                task_type="AR",
+                answer=answer,
+                pair_id=pid,
+                question_role=role,
+                flip_axis="before_after",
+            )
+
+        return [
+            _make_question("s1", answer="B"),
+            _make_question("s2", answer="B"),
+            _make_question("s3", answer="B"),
+            # pairA：两题皆答对 → 双向 AND 判对
+            _pair_member("pa_o", "pa", "pair_original", "B"),
+            _pair_member("pa_m", "pa", "pair_mirror", "B"),
+            # pairB：original 对、mirror 错（answer=A）→ 双向 AND 判错
+            _pair_member("pb_o", "pb", "pair_original", "B"),
+            _pair_member("pb_m", "pb", "pair_mirror", "A"),
+            # 孤儿 pair：只有 original，没有 mirror → 聚合剔除 + 告警
+            _pair_member("orphan_o", "orphan_p", "pair_original", "B"),
+        ]
+
+    def _assert_task_type_breakdown(self, result: InferenceResult) -> None:
+        """契约 3 细化到 task_type：AR 两对仅 pairA 对，RETRIEVAL 三题皆对。"""
+        assert result.per_task_type["AR"]["total"] == 2
+        assert result.per_task_type["AR"]["correct"] == 1
+        assert result.per_task_type["RETRIEVAL"]["total"] == 3
+        assert result.per_task_type["RETRIEVAL"]["correct"] == 3
+
+    def _assert_all_persisted(self, log: HarnessLog, questions: list[GeneratedQuestion]) -> None:
+        """逐题溯源保留：含被剔除的孤儿题在内，每题仍逐题落 predictions。"""
+        rows = log.query("SELECT * FROM predictions WHERE run_id = ?", ("test-run",))
+        persisted = {r["question_id"] for r in rows}
+        assert "orphan_o" in persisted, "孤儿题未逐题落库（逐题溯源被破坏）"
+        assert persisted == {q.question_id for q in questions}, "逐题落库题数与输入不符"
+
+    @pytest.mark.asyncio
+    async def test_double_and_unit_total_and_orphan(self, harness_log: HarnessLog) -> None:
+        """一次全链路推理同时锁定契约 3（双向 AND）、4（unit total）、5（孤儿剔除）。"""
+        questions = self._mixed_questions()
+        result, captured = await _run_capturing_warnings(questions, harness_log)
+
+        # 契约 4：3 single + 2 pair = 5 unit（pair 计 1、孤儿不计入 total）
+        assert result.total == 5, f"unit 粒度 total 错误：{result.total}"
+        # 契约 3：single 3 对 + pairA 对 + pairB 错（双向 AND）= 4
+        assert result.correct == 4, f"双向 AND 聚合错误：correct={result.correct}"
+        self._assert_task_type_breakdown(result)
+
+        # 契约 5：孤儿 pair 被告警（不静默）
+        assert any("orphan_p" in msg for msg in captured), "孤儿 pair 未告警（静默剔除）"
+        self._assert_all_persisted(harness_log, questions)
+
+
+# ── 契约 6：纯非 AR build_batches 与旧逐题黄金参照逐字节一致 ────────────
+
+
+class TestNonARByteIdentical:
+    """纯 single 题库 build_batches 输出与引入 QuestionUnit 前的旧逐题逻辑逐字节一致。"""
+
+    def test_pure_single_matches_golden_reference(self) -> None:
+        """混格链路对纯非 AR 输入不得引入任何漂移（黄金参照对照）。"""
+        items = [_single(f"q{i}", task_type=f"t{i % 4}") for i in range(40)]
+        correctness = {f"q{i}": (i % 3 == 0) for i in range(40)}
+
+        got, _ = build_batches(items, correctness, 8, 3, seed=7, correct_ratio=0.5)
+        ref = _reference_build_batches(items, correctness, 8, 3, seed=7, correct_ratio=0.5)
+
+        # 非空护栏：确保是实质性非空比较（防空==空误通过）
+        assert sum(len(b) for b in ref) > 0, "黄金参照为空，byte-identical 断言未实质覆盖"
+        assert _ids(got) == _ids(ref), "纯非 AR build_batches 与旧逐题黄金参照产生漂移"