Video-Tree-TRM5/research-wiki/plans/fix-diagnosis-tree-data-link-plan.md

type, node_id, title, date

type	node_id	title	date
plan	plan:fix-diagnosis-tree-data-link-plan	实现计划: 修复诊断 tree_data 断链 bug	2026-07-15

修复诊断 tree_data 断链 bug Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use subagent-driven-development to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 接通 TRM4→TRM5 迁移时断掉的诊断树加载环，让 evaluate_span 拿到真实 ground_truth、error_type 归因不再坍缩。

Architecture: app 层新增树展平器（递归遍历 tree.json 的嵌套 roots → 扁平 {"nodes":{id:{card,level,time_range}}}），在离线诊断（video_split_cli）与训练循环（runner）两个注入点按诊断涉及的 video 加载填充 tree_data；core 侧把 run_diagnosis 的静默回退改为缺失即 fail-loud。core 只消费 dict，不碰归因瀑布（算法保真 §4.7#7）。

Tech Stack: Python 3.11、pytest、loguru、asyncio；参考 TRM4 core/harness/diagnose.py:1677 的 tree_cache 语义。

---

设计来源

research-wiki/designs/fix-diagnosis-tree-data-link.md（已含 Codex 审查修订）。

文件结构映射

文件	动作	责任
app/harness/tree_nodes.py	新建	树展平器：load_tree_nodes + load_tree_data_for_videos
core/evolution/diagnose.py	改 :2144-2145	视频未覆盖即 raise（fail-loud，唯一 core 改动）
app/harness/video_split_cli.py	改 build_diagnosis_deps(:262-337) + _execute_real(:590-600)	离线注入：wrong_ids→video 加载填充 tree_data
app/harness/runner.py	改 _run_diagnosis(:2163-2187)	训练注入：question_ids→video 加载注入
tests/unit/test_tree_nodes.py	新建	展平器单测
tests/integration/test_baseline_diagnosis.py	改 :99	更新传 tree_data={} 的用例为真实/伪造树
tests/integration/test_diagnosis_tree_link.py	新建	ground_truth 接通 + 依赖方向

关键代码事实（Codex 已核验）

• 仅 L1Node 有 to_dict（app/tree/index.py:260）；L2/L3 为其内部闭包；输出无 level、L3 用 timestamp 无 time_range → 不走对象层，直接遍历 json。
• node_id 累积式 ..._L1_000_L2_000_L3_000 → level 按遍历深度赋值，不解析 node_id。
• GeneratedQuestion.video_id: str（core/types.py:60）；load_questions_by_id(dir) -> dict[str, GeneratedQuestion]（video_split_cli.py:443）。
• run_diagnosis else 分支已支持 {video_id: {...}} 形态（diagnose.py:2068-2074）。
• diagnose.py:2145 的 td = ...get(vid,{}) 在 :2148 try 之前 → 在此处 raise 不会被 :2159 的 except ValueError（judge 降级）吞。
• store 根 = Path("store")，tree.json 在 store/videos/<vid>/tree.json（与 factory.py:85 一致）。

---

Task 1: 树展平器 app/harness/tree_nodes.py

Files:

• Create: app/harness/tree_nodes.py

• Test: tests/unit/test_tree_nodes.py

• Step 1: 写失败测试（正确性 + level + fail-loud）

# tests/unit/test_tree_nodes.py
"""树展平器单测：用真实 store/videos/0RxMZBLeqRI/tree.json 验证展平正确性与 fail-loud。"""
import json
from pathlib import Path

import pytest

from app.harness.tree_nodes import load_tree_data_for_videos, load_tree_nodes

_STORE = Path("store")
_VID = "0RxMZBLeqRI"  # 真实样本，111 节点


def _recursive_count(tree_json: dict) -> int:
    def walk(n: dict) -> int:
        return 1 + sum(walk(c) for c in (n.get("children") or []))
    return sum(walk(r) for r in tree_json["roots"])


def test_load_tree_nodes_flattens_all_nodes():
    result = load_tree_nodes(_STORE, _VID)
    assert set(result.keys()) == {"nodes"}
    nodes = result["nodes"]
    raw = json.loads((_STORE / "videos" / _VID / "tree.json").read_text(encoding="utf-8"))
    assert len(nodes) == _recursive_count(raw)
    sample = next(iter(nodes.values()))
    assert set(sample.keys()) == {"card", "level", "time_range"}
    assert isinstance(sample["card"], dict)


def test_level_assigned_by_depth_not_node_id():
    nodes = load_tree_nodes(_STORE, _VID)["nodes"]
    l1_id = f"{_VID}_L1_000"
    l3_id = f"{_VID}_L1_000_L2_000_L3_000"
    assert nodes[l1_id]["level"] == 1
    assert nodes[l3_id]["level"] == 3  # 若按 node_id 首个 _L\d_ 会误判成 1


def test_missing_tree_raises_file_not_found():
    with pytest.raises(FileNotFoundError):
        load_tree_nodes(_STORE, "__no_such_video__")


def test_empty_roots_raises_value_error(tmp_path):
    vdir = tmp_path / "videos" / "vX"
    vdir.mkdir(parents=True)
    (vdir / "tree.json").write_text(json.dumps({"metadata": {}, "roots": []}), encoding="utf-8")
    with pytest.raises(ValueError):
        load_tree_nodes(tmp_path, "vX")


def test_load_for_videos_dedups():
    data = load_tree_data_for_videos(_STORE, [_VID, _VID])
    assert set(data.keys()) == {_VID}
    assert data[_VID]["nodes"]

• Step 2: 运行确认失败

Run: conda run -n Video-Tree-TRM pytest tests/unit/test_tree_nodes.py -q Expected: FAIL（ModuleNotFoundError: No module named 'app.harness.tree_nodes'）

• Step 3: 实现展平器

# app/harness/tree_nodes.py
"""诊断侧树读取适配：把嵌套 tree.json 展平成诊断消费的扁平 nodes dict。

诊断编排（core/evolution/diagnose.py）期望 tree_data 形如
{"nodes": {node_id: {card, level, time_range}}}，但 TRM5 建树产物
store/videos/<vid>/tree.json 是嵌套 {"metadata","roots":[...]}。本模块递归展平，
接通 TRM4→TRM5 迁移时断掉的 ground_truth 加载环。

不走 TreeIndex 对象层：仅 L1Node 有 to_dict（app/tree/index.py:260），L2/L3 为其内部闭包，
且 to_dict 输出无 level、L3 用 timestamp 无 time_range。直接遍历 json 更省且零改建树模块。
"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any


def load_tree_nodes(store_dir: Path, video_id: str) -> dict[str, Any]:
    """加载单视频 tree.json 并展平成扁平 nodes dict。

    参数:
        store_dir: store 根目录（含 videos/<video_id>/tree.json）。
        video_id: 视频标识。

    返回:
        {"nodes": {node_id: {"card": dict, "level": int, "time_range": list}}}。

    异常:
        FileNotFoundError: tree.json 不存在（沿用 factory.py fail-loud 先例）。
        ValueError: 树无有效 roots、节点缺 id、或展平后 nodes 为空。

    关键实现:
        level 由遍历深度赋值（root=1/child=2/孙=3），不解析 node_id——node_id 累积式
        （..._L1_..._L2_..._L3_）用正则首匹配会把 L2/L3 误判成 1。
        L3 无 time_range，用 timestamp 合成 [t, t]。
    """
    tree_path = store_dir / "videos" / video_id / "tree.json"
    if not tree_path.exists():
        raise FileNotFoundError(f"树索引文件不存在: {tree_path}（诊断需真实树，P5 fail loud）")
    tree = json.loads(tree_path.read_text(encoding="utf-8"))
    roots = tree.get("roots")
    if not isinstance(roots, list) or not roots:
        raise ValueError(f"树无有效 roots: {tree_path}")

    nodes: dict[str, Any] = {}

    def _walk(node: dict[str, Any], level: int) -> None:
        node_id = node.get("id")
        if not isinstance(node_id, str) or not node_id:
            raise ValueError(f"节点缺 id: {tree_path}")
        time_range = node.get("time_range")
        if time_range is None:
            ts = node.get("timestamp")
            time_range = [ts, ts] if ts is not None else [0, 0]
        nodes[node_id] = {
            "card": node.get("card", {}),
            "level": level,
            "time_range": time_range,
        }
        for child in node.get("children", []) or []:
            _walk(child, level + 1)

    for root in roots:
        _walk(root, 1)

    if not nodes:
        raise ValueError(f"展平后 nodes 为空: {tree_path}")
    return {"nodes": nodes}


def load_tree_data_for_videos(store_dir: Path, video_ids: list[str]) -> dict[str, Any]:
    """按一组 video_id 去重加载展平树，供诊断按 video 注入。

    参数:
        store_dir: store 根目录。
        video_ids: 视频标识列表（可含重复，内部按首次出现顺序去重）。

    返回:
        {video_id: {"nodes": {...}}}。

    异常:
        同 load_tree_nodes（任一视频树缺失/无效即 fail-loud）。
    """
    return {vid: load_tree_nodes(store_dir, vid) for vid in dict.fromkeys(video_ids)}

• Step 4: 运行确认通过

Run: conda run -n Video-Tree-TRM pytest tests/unit/test_tree_nodes.py -q Expected: PASS（5 passed）

• Step 5: Commit

git add app/harness/tree_nodes.py tests/unit/test_tree_nodes.py
git commit -m "feat: add tree.json flattener for diagnosis ground_truth"

---

Task 2: core 视频覆盖 fail-loud（diagnose.py）

Files:

• Modify: core/evolution/diagnose.py:2144-2145

• Modify: tests/integration/test_baseline_diagnosis.py:99（更新受影响用例）

• Step 1: 写失败测试（缺树 video → raise，不降级）

在 tests/integration/test_baseline_diagnosis.py 追加：

@pytest.mark.asyncio
async def test_run_diagnosis_raises_when_video_tree_missing():
    """诊断视频未被 tree_data 覆盖时 fail-loud（不静默回退、不走 judge 降级）。"""
    from core.evolution.diagnose import run_diagnosis

    q = _make_one_wrong_question(video_id="vMISS", question_id="vMISS-1")
    with pytest.raises(ValueError, match="诊断视频树未覆盖"):
        await run_diagnosis(
            run_id="infer_adhoc",
            questions=[q],
            tree_data={"vOTHER": {"nodes": {}}},  # 故意不含 vMISS
            llm=_FakeLLM(),
            run_log=_fake_run_log_with_one_wrong(q),
            skill_store=_FakeSkillStore(),
            prompts=_fake_diagnose_prompts(),
            concurrency=1,
            question_ids=["vMISS-1"],
        )

helper（_make_one_wrong_question / _fake_run_log_with_one_wrong / _FakeLLM / _FakeSkillStore / _fake_diagnose_prompts）：复用该测试文件已有的伪造装配；若无 _make_one_wrong_question，构造 GeneratedQuestion(question_id="vMISS-1", video_id="vMISS", task_type="Object Reasoning", ...) 并让 run_log 返回一条 correct=0 且带非空 steps_json（含一个 view_node 步）的 prediction。

• Step 2: 运行确认失败

Run: conda run -n Video-Tree-TRM pytest tests/integration/test_baseline_diagnosis.py::test_run_diagnosis_raises_when_video_tree_missing -q Expected: FAIL（当前静默回退 {}，不抛异常）

• Step 3: 改 core 加 fail-loud

core/evolution/diagnose.py 的 _process_question，把 :2144-2145 的：

            vid = prediction.get("video_id", "")
            td = tree_data_by_video.get(vid, {})

改为（位置在 :2148 的 try 之前，故不被 :2159 的 except ValueError 吞）：

            vid = prediction.get("video_id", "")
            if vid not in tree_data_by_video:
                # P5 fail-loud：诊断需真实树，调用方须为每个诊断视频加载 tree_data；
                # 静默回退空树会让 ground_truth 恒空、error_type 归因坍缩（本次修复的根因）。
                raise ValueError(
                    f"诊断视频树未覆盖: video_id={vid!r} 不在注入的 tree_data 中"
                    "（调用方须为每个诊断视频加载树，P5 fail loud）"
                )
            td = tree_data_by_video[vid]

• Step 4: 更新既有传 tree_data={} 的三处用例

tests/integration/test_baseline_diagnosis.py 有三处 tree_data={}（:99 / :224 / :311，形参分别在 :66 / :197 / :283）。逐处判断：

• 若该用例真诊断题（wrong_ids 非空、进 run_diagnosis）→ 改为覆盖其诊断 video 的伪造树：

          tree_data={"<该用例的 video_id>": {"nodes": {"<node_id>": {"card": {}, "level": 1, "time_range": [0, 0]}}}},
  

  （video_id/node_id 填该用例 prediction 实际用的值。）
• 若该用例期望**"无题诊断"早返回**（wrong_ids=[]，不进 run_diagnosis）→ tree_data={} 可保留，并在该用例加一行注释说明豁免原因。

逐处核对：读每个用例构造的 prediction 是否 correct=0 且被诊断——是则改树，否则注释豁免。

• Step 4b: 全仓兜底扫描其它 tree_data={} 调用点

Run: conda run -n Video-Tree-TRM grep -rn "tree_data={}\|tree_data = {}" tests/ app/ 对每个命中判断：进 run_diagnosis 且诊断非空题的必须提供覆盖树；dry-run fake_deps（video_split_cli.py 内，wrong_ids=[] 早返回）豁免。确保 core fail-loud 不误伤既有用例。

• Step 5: 运行确认通过

Run: conda run -n Video-Tree-TRM pytest tests/integration/test_baseline_diagnosis.py -q Expected: PASS

• Step 6: Commit

git add core/evolution/diagnose.py tests/integration/test_baseline_diagnosis.py
git commit -m "fix: fail loud when diagnosis video tree not covered (algo #7 input)"

---

Task 3: 离线注入（video_split_cli.py）

Files:

• Modify: app/harness/video_split_cli.py（新增 _DEFAULT_STORE_DIR + --store-dir CLI、_resolve_paths 返回 store_dir、改 build_diagnosis_deps 签名与 tree_data、_execute_real 传 store_dir/video_ids、删模块顶层假注释 :19-21）

• Modify: tests/unit/test_video_split_cli.py:269-273 与 :290-292（两处旧调用补 store_dir/video_ids，否则改签名后先抛 TypeError 而非期望的 SystemExit）

• Test: tests/unit/test_video_split_cli_tree_inject.py（新建）

• Step 1: 写失败测试（build_diagnosis_deps 填充真实树）

# tests/unit/test_video_split_cli_tree_inject.py
"""离线诊断注入：build_diagnosis_deps 按 video_ids 填充非空 tree_data。"""
from pathlib import Path
from unittest.mock import patch

from app.harness.video_split_cli import build_diagnosis_deps


def test_build_diagnosis_deps_loads_tree_for_videos():
    with patch("app.harness.video_split_cli._DiagLLMSettings") as S, \
         patch("adapters.llm.GovernedLLMClient"), \
         patch("adapters.telemetry.SQLiteTelemetryRecorder"), \
         patch("app.harness.video_split_cli._build_redis_cache", return_value=None):
        s = S.return_value
        s.search_llm_model = "deepseek-v4-pro"
        s.search_llm_base_url = "http://x"
        s.search_llm_api_key = "k"
        s.llm_circuit_breaker_threshold = 32
        s.llm_circuit_breaker_cooldown = 60
        s.llm_timeout = s.llm_ttft_timeout = s.llm_inter_token_timeout = 60
        s.llm_max_retries = 1
        s.llm_retry_base_delay = s.llm_retry_max_delay = 1
        deps = build_diagnosis_deps(
            harness_db=Path("workspaces/default/harness.db"),
            store_dir=Path("store"),
            video_ids=["0RxMZBLeqRI"],
            concurrency=1,
            expected_model="deepseek-v4-pro",
        )
    assert "0RxMZBLeqRI" in deps.tree_data
    assert deps.tree_data["0RxMZBLeqRI"]["nodes"]

GovernedLLMClient / SQLiteTelemetryRecorder 在 build_diagnosis_deps 内是函数级 import，patch 其源模块（adapters.llm / adapters.telemetry）。测试目的仅验证 deps.tree_data 被真实树填充；若装配桩不足以走到 return，可进一步 patch RunLogImpl/VersionedSkillStore。

• Step 2: 运行确认失败

Run: conda run -n Video-Tree-TRM pytest tests/unit/test_video_split_cli_tree_inject.py -q Expected: FAIL（当前 build_diagnosis_deps 无 store_dir/video_ids 参数 → TypeError）

• Step 3: 加常量 + --store-dir + _resolve_paths + 删模块假注释 + 改签名 + 填 tree_data

(a) app/harness/video_split_cli.py 路径常量区（:65-67 附近）新增：

_DEFAULT_STORE_DIR = Path("store")  # tree.json 在 store/videos/<vid>/

(b) argparse（:732 --out-dir 之后）新增：

    parser.add_argument("--store-dir", type=Path, default=None, dest="store_dir")

(c) _resolve_paths（:582-587）改为返回 4 元组（含 store_dir）：

def _resolve_paths(args: argparse.Namespace) -> tuple[Path, Path, Path, Path]:
    """解析 harness_db / questions_dir / out_dir / store_dir（CLI 覆盖默认工程路径）。"""
    harness_db = args.harness_db or _DEFAULT_HARNESS_DB
    questions_dir = args.questions_dir or _DEFAULT_QUESTIONS_DIR
    out_dir = args.out_dir or _DEFAULT_OUT_DIR
    store_dir = args.store_dir or _DEFAULT_STORE_DIR
    return harness_db, questions_dir, out_dir, store_dir

(d) 删模块顶层假注释：app/harness/video_split_cli.py:19-21 把 “+ tree_data={}（由诊断管线内部按需加载）” 改为 “+ tree_data 按 wrong_ids 涉及 video 预加载（store/videos//tree.json 展平）”。

(e) build_diagnosis_deps 签名（:262-264）改为：

def build_diagnosis_deps(
    *,
    harness_db: Path,
    store_dir: Path,
    video_ids: list[str],
    concurrency: int,
    expected_model: str,
) -> DiagnosisDeps:

函数末尾 return DiagnosisDeps(...)（:330-337）改为按 video 加载（并删 docstring 里"tree_data={} 由诊断管线内部按需加载"假注释）：

    from app.harness.tree_nodes import load_tree_data_for_videos

    return DiagnosisDeps(
        run_log=RunLogImpl(str(harness_db)),
        llm=llm,
        skill_store=VersionedSkillStore(_diagnosis_skills_dir()),
        prompts=_load_diagnose_prompts(),
        tree_data=load_tree_data_for_videos(store_dir, video_ids),
        concurrency=concurrency,
    )

• Step 4: _execute_real 解包 store_dir 并算 video_ids 传入

app/harness/video_split_cli.py:592（解包改 4 元组）+ :595-600：

    harness_db, questions_dir, out_dir, store_dir = _resolve_paths(args)
    if not harness_db.exists():
        raise SystemExit(f"harness.db 不存在: {harness_db}（P5 fail loud）")
    canonical_preds = load_canonical_predictions(harness_db, config.baseline_run_id)
    wrong_ids = select_diagnosable_wrong_ids(canonical_preds)
    questions = load_questions_by_id(questions_dir)
    video_ids = [questions[qid].video_id for qid in wrong_ids]
    deps = build_diagnosis_deps(
        harness_db=harness_db,
        store_dir=store_dir,
        video_ids=video_ids,
        concurrency=args.concurrency,
        expected_model=config.model,
    )

全仓其它 _resolve_paths(args) 解包处（如 dry-run _execute_dry 路径若有）同步改 4 元组解包，避免 ValueError: too many values to unpack。先 grep -n "_resolve_paths(args)" app/harness/video_split_cli.py 逐处核对。

• Step 5: 更新受签名影响的既有单测

tests/unit/test_video_split_cli.py:269-273 与 :290-292 两处 cli.build_diagnosis_deps(...) 调用补必填参数（这俩测试验的是凭证/模型漂移 fail-loud（SystemExit），该校验在 tree_data 加载之前，故 video_ids 传空即可）：

        cli.build_diagnosis_deps(
            harness_db=tmp_path / "h.db",
            store_dir=tmp_path,
            video_ids=[],
            concurrency=2,
            expected_model="deepseek-v4-pro",
        )

（两处调用都照此加 store_dir=tmp_path, video_ids=[]。）

• Step 6: 运行确认通过 + 回归

Run: conda run -n Video-Tree-TRM pytest tests/unit/test_video_split_cli_tree_inject.py tests/unit/test_video_split_cli.py tests/unit/test_generate_questions.py -q Expected: PASS

• Step 7: Commit

git add app/harness/video_split_cli.py tests/unit/test_video_split_cli_tree_inject.py tests/unit/test_video_split_cli.py
git commit -m "fix: load real tree_data for offline diagnosis (video_split_cli)"

---

Task 4: 训练注入（runner._run_diagnosis）

Files:

• Modify: app/harness/runner.py:2163-2187

• Test: tests/unit/test_runner_diag_tree_inject.py

• Step 1: 写失败测试（_run_diagnosis 注入非空树）

# tests/unit/test_runner_diag_tree_inject.py
"""训练循环诊断注入：_run_diagnosis 按 batch question_ids 加载真实树注入 run_diagnosis。"""
from unittest.mock import AsyncMock, patch

import pytest


@pytest.mark.asyncio
async def test_run_diagnosis_injects_tree_data(runner_with_real_store):
    """question_ids 对应的 video 树被加载并作为 tree_data 传入 run_diagnosis。"""
    captured = {}

    async def _fake_run_diagnosis(**kwargs):
        captured["tree_data"] = kwargs["tree_data"]
        return _empty_diagnosis_result()

    with patch("core.evolution.diagnose.run_diagnosis", new=AsyncMock(side_effect=_fake_run_diagnosis)):
        await runner_with_real_store._run_diagnosis("infer_adhoc", question_ids=["604-2"])

    assert "0RxMZBLeqRI" in captured["tree_data"]  # 604-2 属于 0RxMZBLeqRI
    assert captured["tree_data"]["0RxMZBLeqRI"]["nodes"]

runner_with_real_store fixture：构造 self._config.store_dir="store"、self._paths.questions_dir 指向含 604-2 的真实 benchmark 的 runner（复用该测试目录已有 runner helper；若无，最小构造使 load_benchmark 能取到 604-2）。_empty_diagnosis_result()：返回 DiagnosisResult 空壳（error_attributions=[]、infra=[]、degraded=[]）。_run_diagnosis 内 run_diagnosis 为函数级 import，patch 其源符号 core.evolution.diagnose.run_diagnosis。

• Step 2: 运行确认失败

Run: conda run -n Video-Tree-TRM pytest tests/unit/test_runner_diag_tree_inject.py -q Expected: FAIL（当前 tree_data={} → captured 不含 0RxMZBLeqRI）

• Step 3: 改 _run_diagnosis 注入树

app/harness/runner.py:2172-2187，在 questions = load_benchmark(...) 后、run_diagnosis(...) 调用处：

        questions = load_benchmark(self._paths.questions_dir)
        run_log = RunLogImpl(str(self._paths.db_path))
        skill_store = VersionedSkillStore(self._paths.skills_dir)
        diagnose_prompts = self._load_diagnose_prompts()

        from app.harness.tree_nodes import load_tree_data_for_videos

        if question_ids is not None:
            qid_set = set(question_ids)
            video_ids = [q.video_id for q in questions if q.question_id in qid_set]
        else:
            video_ids = [q.video_id for q in questions]
        tree_data = load_tree_data_for_videos(Path(self._config.store_dir), video_ids)

        return await run_diagnosis(
            run_id=run_id,
            questions=questions,
            tree_data=tree_data,
            llm=self._llm,
            run_log=run_log,
            skill_store=skill_store,
            prompts=diagnose_prompts,
            concurrency=self._config.concurrency,
            question_ids=question_ids,
        )

（删 tree_data={}, # tree_data 由诊断管线内部按需加载 假注释；确认文件顶部已 from pathlib import Path，否则补 import。）

• Step 4: 运行确认通过

Run: conda run -n Video-Tree-TRM pytest tests/unit/test_runner_diag_tree_inject.py -q Expected: PASS

• Step 5: Commit

git add app/harness/runner.py tests/unit/test_runner_diag_tree_inject.py
git commit -m "fix: load real tree_data for training-loop diagnosis (runner)"

---

Task 5: 集成验证 ground_truth 接通 + 依赖方向

Files:

• Create: tests/integration/test_diagnosis_tree_link.py

• Step 1: 写 integration 测试（ground_truth 非空 + 依赖方向）

# tests/integration/test_diagnosis_tree_link.py
"""集成验证：真实树注入后 evaluate_span 收到非空 ground_truth；core 不依赖 app。"""
from pathlib import Path

from app.harness.tree_nodes import load_tree_nodes
from core.evolution.diagnose import _get_ground_truth_for_trace


def test_ground_truth_nonempty_with_real_tree():
    """对真实 T2 样本 604-2（video 0RxMZBLeqRI）的 view_node 调用，ground_truth 非空。"""
    td = load_tree_nodes(Path("store"), "0RxMZBLeqRI")
    node_id = "0RxMZBLeqRI_L1_000"  # 该视频真实存在的节点
    gt = _get_ground_truth_for_trace(td, "view_node", {"node_id": node_id})
    assert gt and gt != "{}"  # 拿到该节点 card 的 JSON，非空


def test_core_diagnose_does_not_import_app():
    """算法保真 + 依赖方向：core/evolution/diagnose.py 不 import app。"""
    src = Path("core/evolution/diagnose.py").read_text(encoding="utf-8")
    assert "import app." not in src
    assert "from app." not in src

• Step 2: 运行确认通过

Run: conda run -n Video-Tree-TRM pytest tests/integration/test_diagnosis_tree_link.py -q Expected: PASS

• Step 3: 全量回归

Run: conda run -n Video-Tree-TRM pytest tests/ -q Expected: PASS（无回归；诊断相关用例因 core fail-loud 需补树的已在 Task 2 Step 4 处理）

• Step 4: Commit

git add tests/integration/test_diagnosis_tree_link.py
git commit -m "test: integration for diagnosis tree_data link + core dep direction"

---

重跑与重冻衔接（代码计划外的运行步骤）

代码合入后 git short SHA 变 → diag_fingerprint 变 → 需全量重跑：

1. 重跑诊断 + 重冻切分（一条命令走完两阶段）： CUDA_VISIBLE_DEVICES=0 CONCURRENCY=12 bash scripts/build_video_split.sh
2. 实测校验 tier：重跑后查新 fingerprint 的 baseline_diagnosis，确认 T2≈82 / T1≈152（缓存命中预期）；偏差需归因（设计 §6：C3 异常吞并等非 tree_data 不稳定源）。
3. 观察 error_type 恢复多值：确认 error_type 不再 100% extraction_failure、evolution_target 不再全 tool（软验收，充分性依赖 judge）。
4. 接受 pools.json 成员变化：test/train 具体成员随多样性维恢复而变，floor/代表性 ε 约束仍保证 test 代表性合格（设计 §6，用户已确认重冻覆盖）。

算法保真校验（§4.7）

算法	是否涉及	结论
#7 诊断瀑布	是（仅接通输入）	不改 attribute_error（diagnose.py:910-941）、severity 函数、defect/lapse 判定；Task 2 仅在 _process_question 加 fail-loud 输入护栏。参考 TRM4 core/harness/diagnose.py:1677 tree_cache 语义对齐 ground_truth。
#12 训练循环编排	是（仅换 tree_data 来源）	runner._run_diagnosis 只把 tree_data={} 换成真实加载，不改三级嵌套/慢更新/断点续训。
其余 11 项	否	不涉及。

验收标准

• 5 个 Task 全绿；pytest tests/ 无回归。
• evaluate_span/_get_ground_truth_for_trace 在真实树下拿到非空 ground_truth（Task 5）。
• core/evolution/diagnose.py 不 import app（依赖方向）。
• 两生产注入点（离线 build_diagnosis_deps + 训练 _run_diagnosis）均加载真实树。允许保留 tree_data={} 的豁免场景：dry-run fake_deps（wrong_ids=[] 早返回）、以及测试里走"无题诊断早返回"路径的用例（须带注释说明）。Step 4b 的 grep 兜底确保无遗漏的会真正进 run_diagnosis 的空树调用。