Files

T

iomgaa 09100dcbd4 docs: core/evolution/ implementation plan — 9 tasks, TDD, algorithm fidelity checks

2026-07-07 07:55:21 -04:00

44 KiB

Raw Permalink Blame History

core/evolution/ 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: Migrate TRM4 evolution engine into a Clean Architecture extractable kernel (core/evolution/).

Architecture: 7 files, dependency-ordered. Pure decision logic only — no DB writes, no filesystem versioning, no inference orchestration. All external I/O through 3 Protocols + 1 existing LLMProvider. Async-first (asyncio.gather + Semaphore).

Tech Stack: Python 3.11, asyncio, scipy.special, json_repair, loguru, pluggy (already in project)

Design doc: research-wiki/designs/2026-07-07-core-evolution-design.md

TRM4 source: /home/iomgaa/Projects/Video-Tree-TRM4/core/harness/

File Structure

File	Lines (est.)	Creates	Depends on
`core/evolution/protocols.py`	60	New	`core/evolution/types.py` (TYPE_CHECKING)
`core/evolution/types.py`	280	New	—
`core/evolution/gate.py`	170	New	`types.py`
`core/evolution/patch.py`	440	New	— (loguru only)
`core/evolution/validate.py`	70	New	`gate.py`, `types.py`
`core/evolution/diagnose.py`	1200	New	`types.py`, `protocols.py`, `core/protocols.py`
`core/evolution/evolve.py`	900	New	`types.py`, `protocols.py`, `patch.py`, `core/protocols.py`
`core/evolution/__init__.py`	30	Modify	all above
`tests/unit/test_gate.py`	200	New	—
`tests/unit/test_patch.py`	350	New	—
`tests/unit/test_validate.py`	120	New	—
`tests/unit/test_diagnose.py`	400	New	—
`tests/unit/test_evolve.py`	400	New	—

Task 1: protocols.py + types.py（基础层）

Files:

Create: core/evolution/protocols.py
Create: core/evolution/types.py
Test: tests/unit/test_evolution_types.py
Step 1: Write type construction tests

"""tests/unit/test_evolution_types.py"""
from core.evolution.types import (
    GateParams, GateVerdict, SpanMetrics, SkillStepAdherence,
    QuestionMetrics, ErrorAttribution, CaseSample,
    SkillCasePack, SystemCasePack, ToolCasePack, DiagnosisResult,
    EvolutionRecord, RejectedEdit, EvolutionResult,
    PairResult, QuadrantClassification,
    DiagnosePrompts, EvolvePrompts,
)

def test_gate_params_frozen():
    p = GateParams(e_confirm=20.0, e_provisional=3.0, w_net_min=2,
                   delta_min=0.02, lambda_dir=-0.642, e_rollback=10.0)
    assert p.e_confirm == 20.0
    import pytest
    with pytest.raises(AttributeError):
        p.e_confirm = 1.0

def test_evolution_record_mutable():
    r = EvolutionRecord(
        target_file="test.md", target_type="skill",
        original_content="a", evolved_content="b",
        reason="test", status="accepted", source_version="v1",
        suggestions=[], edits=[], apply_report=[], clip_info={},
    )
    r.status = "rejected"
    assert r.status == "rejected"

def test_diagnose_prompts_frozen():
    dp = DiagnosePrompts(
        defect_vs_lapse="p1", reasoning_sub="p2",
        span_eval_system="p3", span_eval_user="p4",
        missed_nodes="p5", skill_adherence="p6",
        confirmation_bias="p7", evidence_sufficiency="p8",
    )
    assert dp.defect_vs_lapse == "p1"

Step 2: Run test — expect FAIL (ImportError)

Run: conda activate Video-Tree-TRM & pytest tests/unit/test_evolution_types.py -v

Step 3: Implement types.py

从 TRM4 迁移所有 dataclass。关键变更：

TRM4 位置	TRM5 变更
`eprocess.py::GateParams/GateVerdict`	原样迁移
`diagnose.py::SpanMetrics` 等 9 个	全部标 `frozen=True`（TRM4 中 QuestionMetrics 非 frozen，TRM5 一次性构造）
`evolve.py::EvolutionRecord`	保持 mutable，新增 `result_version: str
`evolve.py::RejectedEdit`	原样迁移，frozen
`evolve.py::EvolutionResult`	移除 `skills_version`/`prompts_version`（app/ 职责）
`validate.py::ValidationOutcome/Probation/InferenceRunConfig`	不迁——属 app/harness/
新增 `PairResult`/`QuadrantClassification`	块验证纯决策输出
新增 `DiagnosePrompts`/`EvolvePrompts`	模板束

保真校验点：逐字段对比 TRM4 dataclass，确保无遗漏字段。特别注意 DiagnosisResult 的完整字段列表（约 20 个字段）。

Step 4: Implement protocols.py

"""core/evolution/protocols.py — 3 个只读 Protocol"""
from __future__ import annotations
from typing import Any, Protocol, runtime_checkable

@runtime_checkable
class SkillStore(Protocol):
    def read_skill(self, filename: str) -> str: ...
    def list_skill_files(self) -> list[str]: ...

@runtime_checkable
class PromptStore(Protocol):
    def read_prompt(self, filename: str) -> str: ...
    def list_prompt_files(self) -> list[str]: ...

@runtime_checkable
class RunLog(Protocol):
    async def get_predictions(
        self, run_id: str, *, question_ids: list[str] | None = None,
    ) -> list[dict[str, Any]]: ...
    async def get_traces(
        self, run_id: str, *, question_ids: list[str] | None = None,
    ) -> list[dict[str, Any]]: ...

Step 5: Run test — expect PASS

Run: conda activate Video-Tree-TRM & pytest tests/unit/test_evolution_types.py -v

Step 6: Commit

git add core/evolution/types.py core/evolution/protocols.py tests/unit/test_evolution_types.py
git commit -m "feat(evolution): types.py + protocols.py — foundation dataclasses and Protocol ports"

Task 2: gate.py — CE-Gate e-process（算法 #5）

Files:

Create: core/evolution/gate.py
Test: tests/unit/test_gate.py
Source: TRM4 eprocess.py (163 行，原样迁移，零有意变更)
Step 1: Write gate tests

"""tests/unit/test_gate.py"""
import math
import pytest
from core.evolution.gate import compute_e_value, gate_decision, probation_verdict
from core.evolution.types import GateParams, GateVerdict

_PARAMS = GateParams(
    e_confirm=20.0, e_provisional=3.0, w_net_min=2,
    delta_min=0.02, lambda_dir=-0.642, e_rollback=10.0,
)

class TestComputeEValue:
    def test_zero_zero_returns_one(self):
        assert compute_e_value(0, 0) == pytest.approx(1.0)

    def test_negative_w_raises(self):
        with pytest.raises(ValueError):
            compute_e_value(-1, 0)

    def test_negative_l_raises(self):
        with pytest.raises(ValueError):
            compute_e_value(0, -1)

    def test_heavy_loss_returns_near_zero(self):
        assert compute_e_value(0, 20) < 0.01

    def test_heavy_win_returns_large(self):
        assert compute_e_value(10, 0) > 100

    def test_symmetric(self):
        e_5_3 = compute_e_value(5, 3)
        e_3_5 = compute_e_value(3, 5)
        assert e_5_3 > e_3_5

class TestGateDecision:
    def test_confirmed_needs_both_e_and_delta(self):
        v = gate_decision(10, 0, 10, 10, params=_PARAMS)
        assert v.decision == "accept_confirmed"

    def test_continue_on_balanced(self):
        v = gate_decision(3, 3, 6, 20, params=_PARAMS)
        assert v.decision == "continue"

    def test_reject_inertia_on_exhaustion(self):
        v = gate_decision(1, 1, 2, 0, params=_PARAMS)
        assert v.decision == "reject_inertia"

    def test_n_used_zero_raises(self):
        with pytest.raises(ValueError):
            gate_decision(0, 0, 0, 10, params=_PARAMS)

    def test_n_remaining_negative_raises(self):
        with pytest.raises(ValueError):
            gate_decision(1, 0, 1, -1, params=_PARAMS)

class TestProbationVerdict:
    def test_strong_win_confirmed(self):
        assert probation_verdict(10, 0, params=_PARAMS) == "confirmed"

    def test_strong_loss_rollback(self):
        assert probation_verdict(0, 10, params=_PARAMS) == "rollback"

    def test_balanced_unverified(self):
        assert probation_verdict(3, 3, params=_PARAMS) == "unverified"

Step 2: Run test — expect FAIL

Run: conda activate Video-Tree-TRM & pytest tests/unit/test_gate.py -v

Step 3: Implement gate.py

从 TRM4 eprocess.py 原样迁移全部代码（163 行）。变更仅限：

import 路径：core.harness.eprocess → core.evolution.gate
GateParams/GateVerdict 从 core.evolution.types 导入（不在 gate.py 定义）

保真校验：逐行比对 TRM4 eprocess.py，确保 _WALD_WIN/_WALD_LOSS/_SHRINK_PSEUDO 常量值、log 空间公式、对称性技巧、四出口优先级链、futility best-case 检查全部保留。

Step 4: Run test — expect PASS

Run: conda activate Video-Tree-TRM & pytest tests/unit/test_gate.py -v

Step 5: Commit

git add core/evolution/gate.py tests/unit/test_gate.py
git commit -m "feat(evolution): gate.py — CE-Gate e-process (#5)"

Task 3: patch.py — 补丁引擎（算法 #9 局部）

Files:

Create: core/evolution/patch.py
Test: tests/unit/test_patch.py
Source: TRM4 patch.py (427 行，原样迁移，零有意变更)
Step 1: Write patch tests

"""tests/unit/test_patch.py"""
import pytest
from core.evolution.patch import (
    APPENDIX_START, APPENDIX_END, MOMENTUM_START, MOMENTUM_END,
    appendix_region_bounds, momentum_region_bounds, momentum_inner,
    append_to_appendix, extract_appendix_notes, replace_appendix_notes,
    replace_momentum, apply_patch_with_report,
)

class TestRegionBounds:
    def test_no_markers_returns_none(self):
        assert appendix_region_bounds("hello") is None
        assert momentum_region_bounds("hello") is None

    def test_both_markers_returns_range(self):
        text = f"head\n{APPENDIX_START}\nbody\n{APPENDIX_END}\ntail"
        start, end = appendix_region_bounds(text)
        assert text[start:end].startswith(APPENDIX_START)
        assert text[start:end].endswith(APPENDIX_END)

    def test_single_marker_raises(self):
        with pytest.raises(ValueError):
            appendix_region_bounds(f"head\n{APPENDIX_START}\nbody")
        with pytest.raises(ValueError):
            momentum_region_bounds(f"head\n{MOMENTUM_END}\nbody")

class TestAppendix:
    def test_append_creates_region(self):
        result = append_to_appendix("content", ["note1"])
        assert APPENDIX_START in result
        assert "- note1" in result

    def test_extract_notes(self):
        text = f"{APPENDIX_START}\n- a\n- b\n{APPENDIX_END}"
        assert extract_appendix_notes(text) == ["a", "b"]

    def test_replace_empty_deletes_region(self):
        text = f"head\n{APPENDIX_START}\n- old\n{APPENDIX_END}\ntail"
        result = replace_appendix_notes(text, [])
        assert APPENDIX_START not in result

class TestMomentum:
    def test_replace_creates_region(self):
        result = replace_momentum("content", "guidance text")
        assert MOMENTUM_START in result
        assert "guidance text" in result

    def test_marker_injection_raises(self):
        with pytest.raises(ValueError):
            replace_momentum("content", f"evil {MOMENTUM_START}")

    def test_empty_guidance_clears(self):
        text = replace_momentum("content", "old")
        result = replace_momentum(text, "")
        inner = momentum_inner(result)
        assert inner == ""

class TestApplyPatch:
    def test_append_before_protected(self):
        content = f"body\n{APPENDIX_START}\nprotected\n{APPENDIX_END}"
        edits = [{"op": "append", "target": "", "content": "new line"}]
        new, report = apply_patch_with_report(content, edits, [f"{APPENDIX_START}\nprotected\n{APPENDIX_END}"])
        assert report[0]["status"].startswith("applied")
        assert new.index("new line") < new.index(APPENDIX_START)

    def test_replace_in_protected_skipped(self):
        protected = f"{APPENDIX_START}\nprotected\n{APPENDIX_END}"
        content = f"body\n{protected}"
        edits = [{"op": "replace", "target": "protected", "content": "replaced"}]
        new, report = apply_patch_with_report(content, edits, [protected])
        assert report[0]["status"] == "skipped_protected"
        assert "protected" in new

    def test_insert_after_fallback(self):
        content = "line1\nline2"
        edits = [{"op": "insert_after", "target": "nonexistent", "content": "new"}]
        new, report = apply_patch_with_report(content, edits)
        assert "applied_insert_after_fallback" in report[0]["status"]

    def test_delete_first_occurrence(self):
        content = "a\nb\na\nc"
        edits = [{"op": "delete", "target": "a", "content": ""}]
        new, report = apply_patch_with_report(content, edits)
        assert new.count("a") == 1

Step 2: Run test — expect FAIL

Run: conda activate Video-Tree-TRM & pytest tests/unit/test_patch.py -v

Step 3: Implement patch.py

从 TRM4 patch.py 原样迁移全部代码（427 行）。零有意变更——import 路径调整除外。

保真校验：

7 个常量值完全一致
_protected_ranges 半开区间语义 [start, end)
_in_ranges 用 start <= pos < end
append op 的 start > 0 过滤（跳过 frontmatter）
insert_after 三结果（成功 / 降级 append / skip）
target 不 strip，payload strip
每条 edit 前重算 ranges
report 字段 truncation（target[:200], content[:200]）
Step 4: Run test — expect PASS

Run: conda activate Video-Tree-TRM & pytest tests/unit/test_patch.py -v

Step 5: Commit

git add core/evolution/patch.py tests/unit/test_patch.py
git commit -m "feat(evolution): patch.py — patch engine with protected regions (#9)"

Task 4: validate.py — 纯决策函数（算法 #7 局部）

Files:

Create: core/evolution/validate.py
Test: tests/unit/test_validate.py
Step 1: Write validate tests

"""tests/unit/test_validate.py"""
from core.evolution.validate import pair_block, classify_quadrants, compute_accuracy

class TestPairBlock:
    def test_basic_flips(self):
        baseline = {"q1": False, "q2": True, "q3": True}
        candidate = {"q1": True, "q2": False, "q3": True}
        result = pair_block(baseline, candidate, ["q1", "q2", "q3"])
        assert result.w == 1  # q1: wrong→right
        assert result.l == 1  # q2: right→wrong
        assert result.observed == {
            "q1": (False, True), "q2": (True, False), "q3": (True, True)
        }

    def test_empty(self):
        result = pair_block({}, {}, [])
        assert result.w == 0 and result.l == 0

class TestClassifyQuadrants:
    def test_all_four(self):
        observed = {
            "q1": (False, True),   # improved
            "q2": (True, False),   # regressed
            "q3": (False, False),  # persistent_fail
            "q4": (True, True),    # stable_success
        }
        qc = classify_quadrants(observed)
        assert qc.improvements == ["q1"]
        assert qc.regressions == ["q2"]
        assert qc.persistent_fails == ["q3"]
        assert qc.stable_successes == ["q4"]

    def test_sorted_within_quadrant(self):
        observed = {"z": (False, True), "a": (False, True)}
        qc = classify_quadrants(observed)
        assert qc.improvements == ["a", "z"]

class TestComputeAccuracy:
    def test_basic(self):
        assert compute_accuracy({"q1": True, "q2": False}, ["q1", "q2"]) == 0.5

    def test_empty_raises(self):
        import pytest
        with pytest.raises(ZeroDivisionError):
            compute_accuracy({}, [])

Step 2: Run test — expect FAIL

Run: conda activate Video-Tree-TRM & pytest tests/unit/test_validate.py -v

Step 3: Implement validate.py

三个纯函数，约 70 行。从 TRM4 validate.py 的 _pair_block 和 _classify_quadrants 提取纯逻辑。

"""core/evolution/validate.py — 块验证纯决策函数。"""
from core.evolution.types import PairResult, QuadrantClassification

def pair_block(
    baseline: dict[str, bool],
    candidate: dict[str, bool],
    question_ids: list[str],
) -> PairResult:
    """逐题比对基线与候选对错，统计翻转。"""
    w = l = 0
    observed: dict[str, tuple[bool, bool]] = {}
    for qid in question_ids:
        b, c = baseline[qid], candidate[qid]
        observed[qid] = (b, c)
        if not b and c:
            w += 1
        elif b and not c:
            l += 1
    return PairResult(w=w, l=l, observed=observed)

def classify_quadrants(
    observed: dict[str, tuple[bool, bool]],
) -> QuadrantClassification:
    """按 (baseline, candidate) 四组分类，各组内 sorted。"""
    improvements, regressions, persistent_fails, stable_successes = [], [], [], []
    for qid, (prev, curr) in observed.items():
        if not prev and curr:
            improvements.append(qid)
        elif prev and not curr:
            regressions.append(qid)
        elif not prev and not curr:
            persistent_fails.append(qid)
        else:
            stable_successes.append(qid)
    return QuadrantClassification(
        improvements=sorted(improvements),
        regressions=sorted(regressions),
        persistent_fails=sorted(persistent_fails),
        stable_successes=sorted(stable_successes),
    )

def compute_accuracy(
    correctness: dict[str, bool],
    question_ids: list[str],
) -> float:
    """纯算术：sum(correct) / len(ids)。"""
    return sum(correctness[qid] for qid in question_ids) / len(question_ids)

Step 4: Run test — expect PASS

Run: conda activate Video-Tree-TRM & pytest tests/unit/test_validate.py -v

Step 5: Commit

git add core/evolution/validate.py tests/unit/test_validate.py
git commit -m "feat(evolution): validate.py — pure block validation decision functions (#7)"

Task 5: diagnose.py — 指标计算与 judge 辅助函数

Files:

Create: core/evolution/diagnose.py（本 Task 写 metrics 部分，约 500 行）
Test: tests/unit/test_diagnose.py（本 Task 写 metrics 测试）
Source: TRM4 metrics.py + diagnose.py 的 attribute_error/classify_defect_vs_lapse
Step 1: Write metrics + attribution tests

"""tests/unit/test_diagnose.py"""
import pytest
from core.evolution.diagnose import (
    calc_format_compliance, calc_budget_usage,
    calc_confidence_calibration, calc_repeat_visit_rate,
    calc_search_keyword_repetition, calc_level_jump_pattern,
    calc_tool_usage, extract_json_from_response,
    attribute_error, question_soft_score, aggregate_soft,
)

class TestRuleMetrics:
    def test_format_compliance_empty_returns_one(self):
        assert calc_format_compliance([]) == 1.0

    def test_budget_usage(self):
        assert calc_budget_usage(5, 15) == pytest.approx(1/3)

    def test_confidence_calibration(self):
        assert calc_confidence_calibration(0.9, False) == "high_conf_wrong"
        assert calc_confidence_calibration(0.3, True) == "low_conf_right"
        assert calc_confidence_calibration(0.6, True) == "calibrated"

    def test_repeat_visit_empty(self):
        assert calc_repeat_visit_rate([]) == 0.0

    def test_repeat_visit_all_unique(self):
        assert calc_repeat_visit_rate(["a", "b", "c"]) == 0.0

    def test_repeat_visit_all_same(self):
        assert calc_repeat_visit_rate(["a", "a", "a"]) == pytest.approx(2/3)

    def test_keyword_repetition_lt2(self):
        assert calc_search_keyword_repetition(["one"]) == 0.0

    def test_keyword_repetition_max_jaccard(self):
        val = calc_search_keyword_repetition(["abcdef", "abcxyz"])
        assert 0.0 < val < 1.0

    def test_level_jump_pattern(self):
        assert "L1" in calc_level_jump_pattern(["seg_L1_000", "seg_L2_001"])

    def test_tool_usage_counts(self):
        assert calc_tool_usage(["view_node", "view_node", "search_similar"]) == {
            "view_node": 2, "search_similar": 1,
        }

class TestJsonExtraction:
    def test_fenced_block(self):
        raw = '```json\n{"key": "val"}\n```'
        assert extract_json_from_response(raw) == {"key": "val"}

    def test_outermost_braces(self):
        raw = 'prefix {"key": 1} suffix'
        assert extract_json_from_response(raw) == {"key": 1}

    def test_non_dict_raises(self):
        with pytest.raises(ValueError):
            extract_json_from_response("[1,2,3]")

    def test_garbage_raises(self):
        with pytest.raises(ValueError):
            extract_json_from_response("not json at all")

class TestAttributeError:
    def test_extraction_failure(self):
        from core.evolution.types import QuestionMetrics, SpanMetrics
        span = SpanMetrics(step=1, tool_name="view_node",
            extraction_completeness=0.3, hallucination_rate=0.0,
            missed_info_tags=[], hallucination_tags=[])
        qm = _make_qm(correct=False, span_metrics=[span], missed_nodes=[],
                       evidence_sufficient=True)
        ea = attribute_error(qm)
        assert ea.error_type == "extraction_failure"

    def test_search_failure(self):
        qm = _make_qm(correct=False, span_metrics=[], missed_nodes=["L2_001"],
                       evidence_sufficient=False)
        ea = attribute_error(qm)
        assert ea.error_type == "search_failure"

    def test_reasoning_failure(self):
        qm = _make_qm(correct=False, span_metrics=[], missed_nodes=[],
                       evidence_sufficient=True)
        ea = attribute_error(qm)
        assert ea.error_type == "reasoning_failure"

    def test_mixed_fallback(self):
        qm = _make_qm(correct=False, span_metrics=[], missed_nodes=[],
                       evidence_sufficient=False)
        ea = attribute_error(qm)
        assert ea.error_type == "mixed"

class TestSoftScore:
    def test_no_spans_returns_none(self):
        assert question_soft_score([]) is None

    def test_aggregate_skips_none(self):
        assert aggregate_soft([0.8, None, 0.6]) == pytest.approx(0.7)

    def test_aggregate_all_none(self):
        assert aggregate_soft([None, None]) is None

注：_make_qm 是测试辅助工厂函数，构造 QuestionMetrics 并为非关键字段填充合理默认值。

Step 2: Run test — expect FAIL

Run: conda activate Video-Tree-TRM & pytest tests/unit/test_diagnose.py -v

Step 3: Implement diagnose.py metrics 部分

从 TRM4 metrics.py 迁移以下函数组：

extract_rule_metrics（从 prediction dict + raw_contents 提取 7 个规则指标；confidence 优先取末步 JSON 的 reflect.confidence，否则 prediction["answer_confidence"] 默认 0.5）
7 个规则指标函数（calc_format_compliance 等）+ _trigrams + _parse_json_object + _extract_last_confidence
extract_json_from_response（三级解析：fenced → outermost {} → json_repair）
_call_judge（async 化，max_retries=2 即共 3 次，仅 ValueError 重试，API 错误直传）
question_soft_score + aggregate_soft
5 个 judge 函数（evaluate_span 等，async 化），prompt 文件名：diagnose_span.md/diagnose_missed_nodes.md/diagnose_skill_adherence.md/diagnose_confirmation_bias.md/diagnose_evidence_sufficiency.md
compute_question_metrics（async 化）
_format_trace_text（metrics 版：thought[:100], output[:200]）

从 TRM4 diagnose.py 迁移：

attribute_error（归因瀑布，纯函数）
classify_defect_vs_lapse（async 化，LLMProvider 替代 LLMClient）
_make_degraded_metrics（worker 抛 ValueError 时生成 degraded=True 的 QuestionMetrics，judge 字段 None/空列表；其他异常直传）

关键变更：

LLMClient → LLMProvider；response.choices[0].message.content → response.content
_call_judge 变 async：await llm.chat(messages)
judge 函数均变 async
load_diagnose_prompt(prompts_dir, filename) → 直接从 DiagnosePrompts 束取属性

保真校验：

_SPAN_EVAL_TOOLS = {"view_node", "search_similar", "observe_frame"}
trigram 是字符级，取 MAX（非 mean）
calc_format_compliance 空返回 1.0；calc_budget_usage 无除零 guard（P5）
confidence 阈值：>=0.7 且错 → high_conf_wrong，<0.5 且对 → low_conf_right
calc_level_jump_pattern regex r"_L(\d+)_"，用 → 连接
_call_judge max_retries=2（共 3 次），API 错误直传
归因瀑布精确顺序：extraction → search → reasoning → mixed
defect_vs_lapse 解析失败降级 "lapse"
_extract_last_confidence 任意异常返回 0.5
judge 返回值默认：span completeness/hallucination 默认 0.0，tags 用 list()，missed_nodes 非 list 返 []
Step 4: Run test — expect PASS

Run: conda activate Video-Tree-TRM & pytest tests/unit/test_diagnose.py -v

Step 5: Commit

git add core/evolution/diagnose.py tests/unit/test_diagnose.py
git commit -m "feat(evolution): diagnose.py metrics + attribution (Stage 1)"

Task 6: diagnose.py — 聚合 + 案例包 + 入口

Files:

Modify: core/evolution/diagnose.py（追加 ~700 行）
Modify: tests/unit/test_diagnose.py（追加聚合 + 入口测试）
Step 1: Write aggregation + case pack tests

# 追加到 tests/unit/test_diagnose.py
from unittest.mock import AsyncMock
from core.evolution.types import (
    QuestionMetrics, ErrorAttribution, SpanMetrics,
    SkillCasePack, SystemCasePack, ToolCasePack, DiagnosisResult,
)
from core.evolution.diagnose import (
    aggregate_d2, aggregate_d3, aggregate_d4, aggregate_d5,
    merge_system_packs, merge_tool_packs, run_diagnosis,
)

class TestAggregation:
    def test_d2_empty(self):
        assert aggregate_d2([]) == {}

    def test_d5_empty_returns_zero_structure(self):
        result = aggregate_d5([])
        assert "early_submit_rate" in result
        assert result["early_submit_rate"] == 0.0

class TestMerge:
    def test_merge_system_packs_none_on_empty(self):
        assert merge_system_packs([]) is None

    def test_merge_system_packs_wraps_stats(self):
        pack = SystemCasePack(
            stats={"a": 1}, failure_cases=[], success_cases=[],
        )
        merged = merge_system_packs([pack, pack])
        assert "per_step" in merged.stats
        assert len(merged.stats["per_step"]) == 2

class TestRunDiagnosis:
    def test_empty_predictions_returns_empty_result(self):
        import asyncio
        from core.evolution.types import DiagnosePrompts
        mock_log = AsyncMock()
        mock_log.get_predictions.return_value = []
        mock_log.get_traces.return_value = []
        mock_llm = AsyncMock()
        mock_store = MagicMock()
        mock_store.list_skill_files.return_value = []
        prompts = DiagnosePrompts(
            defect_vs_lapse="", reasoning_sub="",
            span_eval_system="", span_eval_user="",
            missed_nodes="", skill_adherence="",
            confirmation_bias="", evidence_sufficiency="",
        )
        result = asyncio.run(run_diagnosis(
            "run1", [], {}, mock_llm, mock_log, mock_store, prompts,
            concurrency=1,
        ))
        assert isinstance(result, DiagnosisResult)

Step 2: Run test — expect FAIL

Run: conda activate Video-Tree-TRM & pytest tests/unit/test_diagnose.py::TestAggregation tests/unit/test_diagnose.py::TestMerge tests/unit/test_diagnose.py::TestRunDiagnosis -v

Step 3: Implement aggregation + case packs + run_diagnosis

从 TRM4 diagnose.py 迁移：

_mean, _percentile 辅助函数
aggregate_d2/d3/d4/d5
_build_skill_case_packs（含 severity 函数、C3 lapse routing、single-failure fallback；成功案例 n_success=max(2, len(failures)//2)，acc≤0.3 按 budget 升序否则按 adherence 降序）
_build_system_case_pack（_MIN_PATTERN_COUNT=3，3 种行为模式；成功案例要求 correct+calibrated+no_bias+0.3≤budget≤0.8，按 abs(budget-0.5) 排序）
_build_tool_case_packs（_TOOL_TARGET_FILES 映射；低 completeness 先选最多 4 条，高 hallucination 补到总数 4 上限；成功 span 要求 completeness≥0.9 且 hallucination==0.0）
merge_system_packs/merge_tool_packs（stats 用 {"per_step": [...]} 包裹）
_classify_reasoning_failure（串行 pass，prompt diagnose_reasoning_failure.md，JSON key type，解析失败 → reasoning_failure_type=None 不中断）
run_diagnosis 入口（async，Semaphore 限并发，reasoning_failure 串行 pass）

注：resolve_skill_file 定义在 evolve.py（Task 7），diagnose.py 从 evolve 导入。

关键变更：

ThreadPoolExecutor → asyncio.gather + Semaphore(concurrency)
HarnessLog → RunLog Protocol（get_predictions/get_traces）
不写 DB（_ensure_diagnosis_tables/_clear_existing/_insert_* 全部移除）
不写 JSON 文件（write analyses/... 移除）
树数据从参数传入（非 _load_tree_cache 文件读）
skill 内容从 SkillStore 读
INFRA 统计按 task/video/question 过滤范围重算，不受 stop_reason 过滤影响

保真校验：

_INFRA_STOP_REASONS = frozenset({"error", "parse_error"})
案例包选择规则（见上述各函数描述）
single-failure fallback：1 个 defect → lapse_note（fallback 文本 "复核该类已有规则，避免重复此类单例失败"）
lapse_note 空白过滤（strip 后空则丢弃）
_format_trace_text（diagnose 版不截断，与 metrics 版不同！）
D3 avg_steps key 实际存 budget_usage mean（TRM4 命名不一致，保留）
_make_case_sample metrics 子字典固定 key：correct/error_type/budget_usage/confidence_calibration/repeat_visit_rate/tool_usage/missed_nodes/adherence_rate/confirmation_bias/evidence_sufficient
Step 4: Run test — expect PASS

Run: conda activate Video-Tree-TRM & pytest tests/unit/test_diagnose.py -v

Step 5: Commit

git add core/evolution/diagnose.py tests/unit/test_diagnose.py
git commit -m "feat(evolution): diagnose.py aggregation + case packs + run_diagnosis (#8)"

Task 7: evolve.py — 验证 + 辅助函数

Files:

Create: core/evolution/evolve.py（本 Task 写验证 + 辅助部分，约 400 行）
Test: tests/unit/test_evolve.py
Source: TRM4 evolve.py 的 validate_*/rank_and_clip/edit_budget_at/_resolve_skill_file 等
Step 1: Write evolve validation + helpers tests

"""tests/unit/test_evolve.py"""
import pytest
from core.evolution.evolve import (
    validate_skill, validate_system, validate_tool,
    edit_budget_at, resolve_skill_file,
)

class TestValidateSkill:
    def test_identical_passes(self):
        content = "---\nname: test\ndescription: d\ntask_type: t\n---\nbody"
        result = validate_skill(content, content)
        assert result.passed

    def test_changed_frontmatter_fails(self):
        orig = "---\nname: a\ndescription: d\ntask_type: t\n---\nbody"
        evol = "---\nname: b\ndescription: d\ntask_type: t\n---\nbody"
        result = validate_skill(orig, evol)
        assert not result.passed

    def test_length_ratio_too_short_fails(self):
        orig = "---\nname: a\ndescription: d\ntask_type: t\n---\n" + "x" * 1000
        evol = "---\nname: a\ndescription: d\ntask_type: t\n---\nshort"
        result = validate_skill(orig, evol)
        assert not result.passed

class TestValidateSystem:
    def test_identical_passes(self):
        content = "intro\n## 能力边界\nfrozen\n## 输出格式\nfrozen2\n## other\nbody"
        result = validate_system(content, content)
        assert result.passed

    def test_changed_frozen_section_fails(self):
        orig = "intro\n## 能力边界\noriginal\n## other\nbody"
        evol = "intro\n## 能力边界\nchanged\n## other\nbody"
        result = validate_system(orig, evol)
        assert not result.passed

class TestValidateTool:
    def test_identical_passes(self):
        extract = "## 输出格式\nfixed\n## other\nbody"
        verify = "## 输出格式\nfixed2\n## other\nbody2"
        result = validate_tool(extract, extract, verify, verify)
        assert result.passed

    def test_no_code_block_check(self):
        extract = "## 输出格式\nfixed\n```\nunclosed"
        result = validate_tool(extract, extract, "v", "v")
        assert result.passed  # tool 不检查代码块闭合

class TestEditBudget:
    def test_start_at_zero(self):
        assert edit_budget_at(0, 100, 5, 2) == 5

    def test_end_at_total(self):
        assert edit_budget_at(100, 100, 5, 2) == 2

    def test_total_steps_one(self):
        assert edit_budget_at(0, 1, 5, 2) == 5

    def test_start_less_than_end_asserts(self):
        with pytest.raises(AssertionError):
            edit_budget_at(0, 100, 2, 5)

class TestResolveSkillFile:
    def test_direct_match(self):
        class FakeStore:
            def list_skill_files(self): return ["action-reasoning.md", "default-strategy.md"]
            def read_skill(self, f): return ""
        assert resolve_skill_file(FakeStore(), "Action Reasoning") == "action-reasoning.md"

    def test_fallback_to_default(self):
        class FakeStore:
            def list_skill_files(self): return ["default-strategy.md"]
            def read_skill(self, f): return ""
        assert resolve_skill_file(FakeStore(), "Unknown Type") == "default-strategy.md"

Step 2: Run test — expect FAIL

Run: conda activate Video-Tree-TRM & pytest tests/unit/test_evolve.py -v

Step 3: Implement evolve.py validation + helpers

从 TRM4 evolve.py 迁移：

_parse_frontmatter、_strip_appendix_region（宽容版）、_strip_momentum_region（严格版）、_strip_protected_regions
_check_length（去 appendix+momentum 后比较，ratio [0.3, 2.0]）
_check_code_blocks
_extract_section
_skill_protected_spans/_system_protected_spans/_tool_protected_spans
validate_skill/validate_system/validate_tool → 返回 ValidationResult（定义在此文件内部，非 types.py）
edit_budget_at（纯数学，保持 TRM4 断言和 banker's rounding）
rank_and_clip（async 化，type(idx) is int 排除 bool）
_select_top_edits
_parse_llm_json（两级：fenced → json.loads，失败返回 None）
resolve_skill_file（接受 SkillStore 而非 Path）
_format_case_samples（tool_output[:500] 截断）
_format_spans（tool_output[:500] 截断）
_format_rejected_edits

关键变更：

LLMClient → LLMProvider
_resolve_skill_file(skills_dir: Path, ...) → resolve_skill_file(skill_store: SkillStore, ...)
rank_and_clip 变 async

保真校验：

冻结区配置：Skill(frontmatter+appendix+momentum)、System(3 sections+appendix)、Tool(输出格式+appendix)
frontmatter 三字段：name/description/task_type
_parse_frontmatter regex 必须从文件开头匹配 ^---\n...\n---，yaml.safe_load 失败返回 None
_strip_appendix_region 宽容 vs _strip_momentum_region 严格（不对称保留）
_parse_llm_json：只匹配 ```json fenced block（非 ``` 不带 json）→ json.loads，失败返回 None（与 metrics 的三级不同！）
validate_tool 不检查代码块闭合（与 skill/system 不同）
type(idx) is int（非 isinstance）
rank_and_clip/_request_rank_indices：rank LLM ValueError 降级，API 异常不捕获
rewrite_from_suggestions：prompt evolve_rewrite.md，JSON key rewritten，重写不得长于原文，只捕 ValueError/KeyError/TypeError/AttributeError
_format_case_samples tool_output[:500] 截断
_format_rejected_edits gate 证据格式 W=... L=... E={:.2f} δ̂={:+.3f}
Step 4: Run test — expect PASS

Run: conda activate Video-Tree-TRM & pytest tests/unit/test_evolve.py -v

Step 5: Commit

git add core/evolution/evolve.py tests/unit/test_evolve.py
git commit -m "feat(evolution): evolve.py validation + helpers"

Task 8: evolve.py — per-target 进化

Files:

Modify: core/evolution/evolve.py（追加 ~500 行）
Modify: tests/unit/test_evolve.py（追加进化测试）
Step 1: Write evolution loop tests

# 追加到 tests/unit/test_evolve.py
import asyncio
from unittest.mock import AsyncMock, MagicMock
from core.evolution.types import (
    SkillCasePack, SystemCasePack, ToolCasePack,
    EvolutionRecord, EvolvePrompts,
)
from core.evolution.evolve import (
    evolve_single_skill, evolve_system_prompt, evolve_single_tool,
    consolidate_appendix,
)

_PROMPTS = EvolvePrompts(
    evolve_skill="sk", evolve_system="sys", evolve_tool="tool",
    evolve_rank="rank", consolidate_system="cons",
)

def _make_fake_llm(response_content: str):
    """构造返回固定内容的假 LLMProvider。"""
    from core.types import LLMResponse
    mock = AsyncMock()
    mock.chat.return_value = LLMResponse(
        content=response_content, thinking="", model="test",
        provider="test", prompt_tokens=0, completion_tokens=0,
        latency_ms=0, ttft_ms=None, max_inter_token_ms=None,
        cache_hit=False, call_id="test-id",
    )
    return mock

class TestEvolveSingleSkill:
    def test_empty_pack_skipped(self):
        pack = SkillCasePack(
            task_type="test", target_file="test.md",
            stats={}, failure_cases=[], success_cases=[], lapse_notes=[],
        )
        store = MagicMock()
        store.read_skill.return_value = "---\nname: t\ndescription: d\ntask_type: t\n---\nbody"
        store.list_skill_files.return_value = ["test.md"]
        llm = _make_fake_llm('{"suggestions":[],"edits":[]}')
        record = asyncio.run(evolve_single_skill(
            llm, pack, store, _PROMPTS, "v1", 5, 6,
        ))
        assert record.status in ("rejected", "skipped")

class TestEvolveSystemPrompt:
    def test_no_failures_returns_skipped(self):
        pack = SystemCasePack(stats={}, failure_cases=[], success_cases=[])
        store = MagicMock()
        store.read_prompt.return_value = "## 能力边界\nfixed\n## 输出格式\nfixed\n## 视频树结构\nfixed\nbody"
        llm = _make_fake_llm('{"suggestions":[],"edits":[]}')
        record = asyncio.run(evolve_system_prompt(
            llm, pack, store, _PROMPTS, "v1", 5,
        ))
        assert record.status in ("rejected", "skipped")

class TestEvolveSingleTool:
    def test_evolved_content_is_json(self):
        pack = ToolCasePack(
            tool_name="view_node",
            target_files=["view_node_extract.md", "view_node_verify.md"],
            stats={}, failure_spans=[], success_spans=[],
        )
        store = MagicMock()
        store.read_prompt.return_value = "## 输出格式\nfixed\nbody"
        llm = _make_fake_llm('{"suggestions":[],"edits":[]}')
        record = asyncio.run(evolve_single_tool(
            llm, pack, store, _PROMPTS, "v1", 5,
        ))
        import json
        parsed = json.loads(record.evolved_content)
        assert "extract" in parsed and "verify" in parsed

class TestConsolidateAppendix:
    def test_single_note_passthrough(self):
        llm = _make_fake_llm("")
        result = asyncio.run(consolidate_appendix(llm, ["note1"]))
        assert result == ["note1"]

    def test_exception_returns_original(self):
        llm = AsyncMock()
        llm.chat.side_effect = RuntimeError("boom")
        result = asyncio.run(consolidate_appendix(llm, ["a", "b", "c"]))
        assert result == ["a", "b", "c"]

Step 2: Run test — expect FAIL

Run: conda activate Video-Tree-TRM & pytest tests/unit/test_evolve.py::TestEvolveSingleSkill tests/unit/test_evolve.py::TestEvolveSystemPrompt tests/unit/test_evolve.py::TestEvolveSingleTool tests/unit/test_evolve.py::TestConsolidateAppendix -v

Step 3: Implement per-target evolution

从 TRM4 evolve.py 迁移：

_run_patch_evolution_loop（async 化，range(2) 两轮，三种失败反馈）
_build_lapse_only_attempt（合成 applied_append report）
evolve_single_skill（三分支：lapse-only/rewrite/patch + appendix 追加 + consolidation）
evolve_system_prompt（无 lapse、无 rewrite、无 appendix）
evolve_single_tool（extract+verify 合池 _src 标记、shared budget、JSON evolved_content）
consolidate_appendix（async 化，四守卫）
rewrite_from_suggestions（async 化，3 个拒绝条件）
_append_lapse_with_consolidation（>= threshold 触发、G4 >= 拒绝等长）

关键变更：

全部 client.chat() → await llm.chat(messages)
response.choices[0].message.content → response.content
skills_dir / target_file → skill_store.read_skill(target_file)
prompts_dir / filename → prompt_store.read_prompt(filename)
版本写入（advance_version/copytree）全部移除——返回 EvolutionRecord
run_evolution 编排移除——per-target 函数是最高粒度

保真校验：

Skill 三分支精确条件和行为
_build_lapse_only_attempt 合成 applied_append 状态
rank_and_clip 三级降级
Tool _src 标记合池/拆回
consolidate 四守卫（G4 在调用方）
rewrite 长度限制（重写不得长于原文）、异常类型（仅捕 ValueError/KeyError/TypeError/AttributeError）
两轮重试反馈文本
Step 4: Run test — expect PASS

Run: conda activate Video-Tree-TRM & pytest tests/unit/test_evolve.py -v

Step 5: Commit

git add core/evolution/evolve.py tests/unit/test_evolve.py
git commit -m "feat(evolution): evolve.py per-target evolution — skill/system/tool (#9)"

Task 9: init.py + 集成 + lint

Files:

Modify: core/evolution/__init__.py
Run: lint + dependency check
Step 1: Write __init__.py public API

"""core/evolution/ — 自进化循环决策内核。"""
from core.evolution.gate import compute_e_value, gate_decision, probation_verdict
from core.evolution.patch import (
    apply_patch_with_report, append_to_appendix,
    extract_appendix_notes, replace_appendix_notes,
    replace_momentum, momentum_inner,
)
from core.evolution.validate import pair_block, classify_quadrants, compute_accuracy
from core.evolution.diagnose import run_diagnosis
from core.evolution.evolve import (
    evolve_single_skill, evolve_system_prompt, evolve_single_tool,
    edit_budget_at, resolve_skill_file,
)

__all__ = [
    "compute_e_value", "gate_decision", "probation_verdict",
    "apply_patch_with_report", "append_to_appendix",
    "extract_appendix_notes", "replace_appendix_notes",
    "replace_momentum", "momentum_inner",
    "pair_block", "classify_quadrants", "compute_accuracy",
    "run_diagnosis",
    "evolve_single_skill", "evolve_system_prompt", "evolve_single_tool",
    "edit_budget_at", "resolve_skill_file",
]

Step 2: Run lint

conda activate Video-Tree-TRM & ruff check core/evolution/ --fix
conda activate Video-Tree-TRM & ruff format core/evolution/

Step 3: Dependency direction check

# core/evolution/ 不得 import app/ 或 adapters/
grep -rn "from app\." core/evolution/ && echo "VIOLATION" || echo "OK"
grep -rn "from adapters\." core/evolution/ && echo "VIOLATION" || echo "OK"
grep -rn "import app\." core/evolution/ && echo "VIOLATION" || echo "OK"
grep -rn "import adapters\." core/evolution/ && echo "VIOLATION" || echo "OK"

Expected: 全部 OK

Step 4: Update ARCHITECTURE.md §3.1

将 SkillStore/PromptStore/RunLog 的 Protocol 定义从"含写方法"修订为"core/ 只读，写方法在 app/ 实现类"。同步设计文档 §3 的修订说明。

Step 5: Run full test suite

conda activate Video-Tree-TRM & pytest tests/unit/test_gate.py tests/unit/test_patch.py tests/unit/test_validate.py tests/unit/test_diagnose.py tests/unit/test_evolve.py tests/unit/test_evolution_types.py -v --tb=short

Step 6: Commit

git add core/evolution/__init__.py research-wiki/ARCHITECTURE.md
git commit -m "feat(evolution): __init__.py public API + ARCHITECTURE.md Protocol update"

Algorithm Fidelity Check

本计划涉及 4 项核心算法迁移：

#	算法	计划 Task	保真措施
5	CE-Gate e-process	Task 2	原样迁移 163 行，零有意变更；测试覆盖边界（W=L=0、负值、重 win/loss）
7	块顺序验证	Task 4	纯决策函数提取；编排留 app/harness/（Design B 约定）
8	诊断瀑布	Task 5-6	归因瀑布顺序、defect/lapse 分类、案例包选择规则逐条保真
9	进化 patch 引擎	Task 3, 7-8	patch.py 原样迁移；evolve.py 三分支/rank_clip/consolidate 四守卫逐条保真

不涉及的算法：#1-4（建树/检索器）、#6（信息阶梯，app/harness/）、#10（mini-batch，app/harness/）、#11（Agent Loop，已迁移）、#12（树环境语义搜索，已迁移）、#13（训练循环编排，app/harness/）。

44 KiB Raw Permalink Blame History Unescape Escape

core/evolution/ Implementation Plan

File Structure

Task 1: protocols.py + types.py（基础层）

Task 2: gate.py — CE-Gate e-process（算法 #5）

Task 3: patch.py — 补丁引擎（算法 #9 局部）

Task 4: validate.py — 纯决策函数（算法 #7 局部）

Task 5: diagnose.py — 指标计算与 judge 辅助函数

Task 6: diagnose.py — 聚合 + 案例包 + 入口

Task 7: evolve.py — 验证 + 辅助函数

Task 8: evolve.py — per-target 进化

Task 9: __init__.py + 集成 + lint

Algorithm Fidelity Check

44 KiB

Raw Permalink Blame History

Task 9: init.py + 集成 + lint