@@ -0,0 +1,606 @@
# Agent 执行环境修复(Spec-1) 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: ** 修复 AgentLoop 两个工程缺陷(deepseek 输出变体解析失败 0 步阵亡、LLM 瞬时异常无步级重试)与 view_node 摘要吞实体问题。
**Architecture: ** 三处独立小改动:(A1) `_parse_response` 前置围栏剥除 + action.args 平铺收拢;(A2) `run()` Phase 1 增加步级重试循环(显式可重试异常元组,默认 `(TimeoutError, OSError)` , 20s/40s 退避);(B) `TreeEnvironment` 新增结构化实体字段提取,`SearchToolDispatcher._handle_view_node` 在摘要后确定性追加 `[实体]` /`[画面文字]` 区块。
**Tech Stack: ** Python 3.11 / pytest + pytest-asyncio / json_repair。设计文档:`research-wiki/designs/2026-07-11-agent-runtime-fixes-design.md` 。
**设计变更备忘 ** :设计文档 A2 提到 openai SDK 传输异常入可重试元组——实现时收窄为默认 `(TimeoutError, OSError)` ( `ssl.SSLError` 、`ConnectionError` 均为 OSError 子类,覆盖实测穿透案例 796-3; openai API 类异常由 GovernedLLMClient 内部重试栈负责,且 core/ 不得依赖 openai)。元组保留为构造参数,未来可在组合根扩展。
---
### Task 1: A1 解析容错——围栏剥除 + args 平铺收拢
**Files: **
- Modify: `core/agent/loop.py` ( `_parse_response` ,约 265-300 行;模块顶部加正则常量)
- Test: `tests/unit/test_agent_loop.py` (追加测试类)
- [ ] **Step 1: 写失败测试(用 637-3 生产环境真实坏输出的结构等价样本) **
在 `tests/unit/test_agent_loop.py` 末尾追加:
``` python
# ── A1 解析容错测试(Spec-1)──────────────────────────────────
# 生产真实样本结构:尾部围栏残留 + action.args 平铺(开头围栏场景由
# test_leading_json_fence 单独覆盖)
_REAL_FLAT_FENCED = """ {
" plan " : {
" goal " : " 从三个L1根节点开始建立全局认知 " ,
" tool " : " view_node " ,
" reason " : " 三个L1节点覆盖整个视频 "
},
" action " : {
" tool " : " view_node " ,
" node_id " : " J5Npf2xJpag_L1_000 " ,
" question " : " What is the overall topic of this video? "
}
}
``` """
class TestParseNormalization :
""" deepseek 输出变体(args 平铺 + ```json 围栏)归一化。 """
def _parse ( self , content : str ) :
loop = AgentLoop ( llm = AsyncMock ( ) , max_steps = 10 )
return loop . _parse_response ( _make_response ( content ) )
def test_flat_args_with_trailing_fence ( self ) - > None :
""" 生产样本:action 平铺 node_id/question + 尾部围栏。 """
parsed = self . _parse ( _REAL_FLAT_FENCED )
assert parsed is not None
action = parsed [ 4 ]
assert action [ " tool " ] == " view_node "
assert action [ " args " ] == {
" node_id " : " J5Npf2xJpag_L1_000 " ,
" question " : " What is the overall topic of this video? " ,
}
def test_leading_json_fence ( self ) - > None :
content = ' ```json \n { " reflect " : {} , " plan " : {} , " action " : { " tool " : " submit_answer " , " args " : { " answer " : " A " }}} \n ``` '
parsed = self . _parse ( content )
assert parsed is not None
assert parsed [ 4 ] [ " args " ] == { " answer " : " A " }
def test_nested_args_unchanged ( self ) - > None :
""" 标准嵌套结构不受归一化影响。 """
parsed = self . _parse ( _submit_json ( " B " ) )
assert parsed is not None
assert parsed [ 4 ] == { " tool " : " submit_answer " , " args " : { " answer " : " B " } }
def test_action_missing_tool_still_rejected ( self ) - > None :
content = json . dumps ( { " reflect " : { } , " plan " : { } , " action " : { " node_id " : " x " } } )
assert self . _parse ( content ) is None
def test_empty_content_still_rejected ( self ) - > None :
assert self . _parse ( " " ) is None
```
- [ ] **Step 2: 运行测试确认失败 **
Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_agent_loop.py::TestParseNormalization -v`
Expected: `test_flat_args_with_trailing_fence` FAIL(返回 None);`test_nested_args_unchanged` 等可能已 PASS。
- [ ] **Step 3: 实现归一化 **
`core/agent/loop.py` 模块顶部(`import re` 如缺则加,紧邻其他 import):
``` python
# deepseek 等模型稳定输出变体:```json 围栏包裹 JSON 体
_CODE_FENCE_RE = re . compile ( r " ^ \ s*```(?:json)? \ s* \ n?| \ n? \ s*``` \ s*$ " )
```
`_parse_response` 中两处修改。其一,repair 前剥围栏:
``` python
repaired = repair_json ( _CODE_FENCE_RE . sub ( " " , content ) . strip ( ) )
```
其二,action 校验前收拢平铺参数(替换原 `action = data["action"]` 与校验之间):
``` python
action = data [ " action " ]
# deepseek 变体:args 平铺在 action 下(缺 args 嵌套),确定性收拢
if isinstance ( action , dict ) and " tool " in action and " args " not in action :
flat_args = { k : v for k , v in action . items ( ) if k != " tool " }
action = { " tool " : action [ " tool " ] , " args " : flat_args }
if not isinstance ( action , dict ) or " tool " not in action or " args " not in action :
return None
```
- [ ] **Step 4: 运行测试确认通过 **
Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_agent_loop.py -v`
Expected: 全部 PASS(含原有 9 个测试,确认无回归)。
- [ ] **Step 5: Commit **
``` bash
git add core/agent/loop.py tests/unit/test_agent_loop.py
git commit -m "fix(agent): normalize fenced and flat-args LLM outputs in parser"
```
---
### Task 2: A2 步级重试
**Files: **
- Modify: `core/agent/loop.py` ( `__init__` 第 72-77 行;`run()` Phase 1 约 117-129 行;模块顶部加 `import asyncio` ——当前缺失,Task 2 测试的红灯即源于此)
- Test: `tests/unit/test_agent_loop.py` (追加测试类)
- [ ] **Step 1: 写失败测试 **
``` python
# ── A2 步级重试测试(Spec-1)──────────────────────────────────
import ssl
class TestStepLevelRetry :
""" LLM 瞬时异常的步级重试:可重试元组 / 退避 / fail-fast。 """
def _make_loop ( self , chat_side_effects : list ) - > AgentLoop :
llm = AsyncMock ( )
llm . chat = AsyncMock ( side_effect = chat_side_effects )
return AgentLoop ( llm = llm , max_steps = 10 )
@pytest.mark.asyncio
async def test_transient_error_retried_then_succeeds ( self , monkeypatch ) - > None :
delays : list [ float ] = [ ]
async def _fake_sleep ( seconds : float ) - > None :
delays . append ( seconds )
monkeypatch . setattr ( " core.agent.loop.asyncio.sleep " , _fake_sleep )
loop = self . _make_loop (
[
ssl . SSLError ( " SSLV3_ALERT_BAD_RECORD_MAC " ) ,
TimeoutError ( " watchdog " ) ,
_make_response ( _submit_json ( ) ) ,
]
)
result = await loop . run ( " sys " , " user " , _StubDispatcher ( ) )
assert result . stop_reason == " finished "
assert delays == [ 20.0 , 40.0 ]
@pytest.mark.asyncio
async def test_retry_exhausted_terminates_with_error ( self , monkeypatch ) - > None :
async def _fake_sleep ( seconds : float ) - > None :
pass
monkeypatch . setattr ( " core.agent.loop.asyncio.sleep " , _fake_sleep )
loop = self . _make_loop ( [ TimeoutError ( " t1 " ) , TimeoutError ( " t2 " ) , TimeoutError ( " t3 " ) ] )
result = await loop . run ( " sys " , " user " , _StubDispatcher ( ) )
assert result . stop_reason == " error "
assert loop . _llm . chat . await_count == 3 # 首次 + 2 次重试
@pytest.mark.asyncio
async def test_non_retryable_fails_fast ( self , monkeypatch ) - > None :
sleep_mock = AsyncMock ( )
monkeypatch . setattr ( " core.agent.loop.asyncio.sleep " , sleep_mock )
loop = self . _make_loop ( [ RuntimeError ( " programming bug " ) ] )
result = await loop . run ( " sys " , " user " , _StubDispatcher ( ) )
assert result . stop_reason == " error "
sleep_mock . assert_not_awaited ( )
assert loop . _llm . chat . await_count == 1
```
- [ ] **Step 2: 运行测试确认失败 **
Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_agent_loop.py::TestStepLevelRetry -v`
Expected: 三个测试全部 ERROR——`monkeypatch.setattr("core.agent.loop.asyncio.sleep", ...)` 抛 AttributeError,因为当前 `loop.py` 未 import asyncio。这就是正确的红灯(实现步会加入 `import asyncio` ,此后 monkeypatch 路径有效,绿灯判断以行为断言为准)。
- [ ] **Step 3: 实现步级重试 **
`__init__` 签名扩展(保留既有参数不动):
``` python
def __init__ (
self ,
llm : LLMProvider ,
max_steps : int ,
max_retries : int = 3 ,
* ,
step_retries : int = 2 ,
step_retry_delays : tuple [ float , . . . ] = ( 20.0 , 40.0 ) ,
retryable_exceptions : tuple [ type [ BaseException ] , . . . ] = ( TimeoutError , OSError ) ,
) - > None :
self . _llm = llm
self . _max_steps = max_steps
self . _max_retries = max_retries
self . _step_retries = step_retries
self . _step_retry_delays = step_retry_delays
self . _retryable_exceptions = retryable_exceptions
```
`run()` Phase 1 整段替换(原 117-129 行 try/except):
``` python
# Phase 1: LLM 调用(步级重试:防穿透 GovernedLLMClient 的瞬时异常)
llm_error : Exception | None = None
step_attempt = 0
while True :
try :
response = await self . _call_llm (
messages , token_usage , session_id = session_id
)
break
except self . _retryable_exceptions as e :
step_attempt + = 1
if step_attempt > self . _step_retries :
llm_error = e
break
delay = self . _step_retry_delays [
min ( step_attempt - 1 , len ( self . _step_retry_delays ) - 1 )
]
logger . warning (
" LLM 瞬时异常,步级重试 {} / {} ( {} s 后重发): {} " ,
step_attempt , self . _step_retries , delay , e ,
)
await asyncio . sleep ( delay )
except Exception as e :
llm_error = e
break
if llm_error is not None :
logger . error ( " LLM API 调用失败: {} " , llm_error )
result = LoopResult (
steps = steps ,
steps_used = step_count ,
token_usage = token_usage ,
stop_reason = " error " ,
)
await _call_hook ( pm . hook . on_finish , result = result )
return result
```
注意:`asyncio.CancelledError` 继承 `BaseException` ,两个 except 均不会捕获——取消信号天然穿透,符合设计。失败尝试的 error 遥测由 `GovernedLLMClient` 内部负责(已有),此处仅 loguru 记录。
- [ ] **Step 4: 运行测试确认通过 **
Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_agent_loop.py -v`
Expected: 全部 PASS(原有 9 个 + Task 1 的 5 个 + 本任务 3 个)。
- [ ] **Step 5: Commit **
``` bash
git add core/agent/loop.py tests/unit/test_agent_loop.py
git commit -m "feat(agent): step-level retry for transient LLM errors (20s/40s backoff)"
```
---
### Task 3: B 摘要附带实体原文
**Files: **
- Modify: `app/tree/environment.py` ( `TreeEnvironment` 新增方法,放在 `view_node` 之后约 216 行处)
- Modify: `app/search/tools.py` ( `_handle_view_node` Phase 2/3 之间,约 196-204 行)
- Test: `tests/unit/test_tree_environment.py` 、`tests/unit/test_search_tools.py` (各追加)
- [ ] **Step 1: 写 TreeEnvironment 失败测试 **
`tests/unit/test_tree_environment.py` 追加(该文件已 import 全部 Card/Node 类型与 IndexMeta/TreeIndex,见文件头 11-20 行;构造模式对齐现有 `_make_test_index()` ):
``` python
# ── node_entity_fields 测试(Spec-1 B)───────────────────────
def _make_entity_test_index ( ) - > TreeIndex :
""" 带实体字段的最小三层树(含一个空字段 L2)。 """
l3 = L3Node (
id = " vid_L1_000_L2_000_L3_000 " ,
card = L3Card (
frame_summary = " 一名男子戴耳机 " ,
visible_entities = [ " Bluetooth headset (both ears) " , " man " ] ,
ongoing_actions = [ " talking " ] ,
visible_text = [ " EARPHONE BOTTLE OPENER " ] ,
spatial_layout = " man center " ,
visual_attributes = { } ,
) ,
timestamp = 10.0 ,
)
l2 = L2Node (
id = " vid_L1_000_L2_000 " ,
card = L2Card (
event_description = " 产品评测 " ,
entities = [ " Bluetooth headset (both ears) " , " reviewer " ] ,
actions = [ " reviewing " ] ,
action_subjects = [ " reviewer " ] ,
visible_text = [ " $9.99 " ] ,
spatial_relations = " " ,
state_changes = None ,
) ,
time_range = ( 0.0 , 60.0 ) ,
children = [ l3 ] ,
)
l2_empty = L2Node (
id = " vid_L1_000_L2_001 " ,
card = L2Card (
event_description = " 空镜 " ,
entities = [ ] ,
actions = [ ] ,
action_subjects = [ ] ,
visible_text = [ ] ,
spatial_relations = " " ,
state_changes = None ,
) ,
time_range = ( 60.0 , 120.0 ) ,
)
l1 = L1Node (
id = " vid_L1_000 " ,
card = L1Card (
scene_summary = " 评测场景 " ,
main_setting = " 室内 " ,
key_entities = [ " reviewer " ] ,
main_actions = [ " 评测 " ] ,
topic_keywords = [ " 数码 " ] ,
visible_text = [ ] ,
temporal_flow = " 线性 " ,
) ,
time_range = ( 0.0 , 120.0 ) ,
children = [ l2 , l2_empty ] ,
)
return TreeIndex ( metadata = IndexMeta ( " /test.mp4 " , " video " ) , roots = [ l1 ] )
class TestNodeEntityFields :
def test_l2_entities_and_visible_text ( self ) - > None :
env = TreeEnvironment ( _make_entity_test_index ( ) )
fields = env . node_entity_fields ( " vid_L1_000_L2_000 " )
assert " Bluetooth headset (both ears) " in fields [ " 实体 " ]
assert " $9.99 " in fields [ " 画面文字 " ]
def test_l3_visible_entities ( self ) - > None :
env = TreeEnvironment ( _make_entity_test_index ( ) )
fields = env . node_entity_fields ( " vid_L1_000_L2_000_L3_000 " )
assert " Bluetooth headset (both ears) " in fields [ " 实体 " ]
assert " EARPHONE BOTTLE OPENER " in fields [ " 画面文字 " ]
def test_empty_fields_omitted ( self ) - > None :
env = TreeEnvironment ( _make_entity_test_index ( ) )
assert env . node_entity_fields ( " vid_L1_000_L2_001 " ) == { }
def test_unknown_node_raises ( self ) - > None :
env = TreeEnvironment ( _make_entity_test_index ( ) )
with pytest . raises ( KeyError ) :
env . node_entity_fields ( " nonexistent " )
```
- [ ] **Step 2: 运行确认失败 **
Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_tree_environment.py::TestNodeEntityFields -v`
Expected: FAIL( `AttributeError: node_entity_fields` )。
- [ ] **Step 3: 实现 TreeEnvironment.node_entity_fields **
`app/tree/environment.py` , `view_node` 方法之后追加;模块级常量放 `_SUBTITLE_SKIP` 附近:
``` python
# 各层级 card 的实体字段名(B 修复:dispatcher 追加原文用)
_ENTITY_FIELDS_BY_LEVEL : dict [ str , tuple [ str , . . . ] ] = {
" L1 " : ( " key_entities " , ) ,
" L2 " : ( " entities " , ) ,
" L3 " : ( " visible_entities " , ) ,
}
```
``` python
def node_entity_fields ( self , node_id : str ) - > dict [ str , str ] :
""" 返回节点 card 的实体/画面文字字段原文。
供 dispatcher 在按题摘要后确定性追加,防止 LLM 摘要吞掉
entities/visible_text 信号(benchmark 错题 M1 恶化因素)。
参数:
node_id: 节点 ID。
返回:
{ " 实体 " : " ... " , " 画面文字 " : " ... " },空字段不含对应键。
异常:
KeyError: 节点不存在。
"""
node = self . _id_to_node . get ( node_id )
if node is None :
raise KeyError ( f " 节点不存在: { node_id } " )
level = _node_level ( node )
out : dict [ str , str ] = { }
entity_values : list [ str ] = [ ]
for field_name in _ENTITY_FIELDS_BY_LEVEL [ level ] :
for value in getattr ( node . card , field_name ) or [ ] :
if isinstance ( value , str ) and value . strip ( ) :
entity_values . append ( value . strip ( ) )
if entity_values :
out [ " 实体 " ] = " ; " . join ( dict . fromkeys ( entity_values ) )
text_values = [
v . strip ( )
for v in ( getattr ( node . card , " visible_text " , None ) or [ ] )
if isinstance ( v , str ) and v . strip ( )
]
if text_values :
out [ " 画面文字 " ] = " ; " . join ( dict . fromkeys ( text_values ) )
return out
```
已核实:`_node_level` ( `environment.py:35-48` )返回 `"L1"/"L2"/"L3"` 字符串,与 `_LEVEL_LABEL` 键一致,`_ENTITY_FIELDS_BY_LEVEL` 直接以此为键。
- [ ] **Step 4: 运行确认通过 **
Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_tree_environment.py -v`
Expected: 全部 PASS。
- [ ] **Step 5: 写 dispatcher 失败测试 **
`tests/unit/test_search_tools.py` 追加。注意:现有 `dispatcher` fixture(第 190 行)的树 entities 只有 `["person"]` ,**不复用**——新增专用 fixture 注入带实体的树;summarize 无现成 stub 模式,用 monkeypatch 新建:
``` python
# ── view_node 实体追加测试(Spec-1 B)────────────────────────
def _make_entity_tree ( ) - > TreeIndex :
""" L2 带实体字段的最小树(与 _make_test_tree 同构,仅换 card 内容)。 """
l2 = L2Node (
id = " vid_L1_000_L2_000 " ,
card = L2Card (
event_description = " 产品评测 " ,
entities = [ " Bluetooth headset (both ears) " , " reviewer " ] ,
actions = [ " reviewing " ] ,
action_subjects = [ " reviewer " ] ,
visible_text = [ " $9.99 " ] ,
spatial_relations = " " ,
state_changes = None ,
) ,
time_range = ( 5.0 , 15.0 ) ,
children = [ ] ,
)
l1 = L1Node (
id = " vid_L1_000 " ,
card = L1Card (
scene_summary = " 评测场景 " ,
main_setting = " 室内 " ,
key_entities = [ " reviewer " ] ,
main_actions = [ " 评测 " ] ,
topic_keywords = [ " 数码 " ] ,
visible_text = [ ] ,
temporal_flow = " 线性 " ,
) ,
time_range = ( 0.0 , 30.0 ) ,
children = [ l2 ] ,
)
return TreeIndex (
metadata = IndexMeta ( source_path = " test.mp4 " , modality = " video " ) ,
roots = [ l1 ] ,
)
@pytest.fixture ( )
def entity_dispatcher (
prompts_dir : Path ,
skills_registry : SkillRegistry ,
) - > SearchToolDispatcher :
""" 树含实体字段的 dispatcher(其余配置与 dispatcher fixture 一致)。 """
return SearchToolDispatcher (
env = TreeEnvironment ( _make_entity_tree ( ) ) ,
tool_llm = FakeLLM ( ) ,
vlm = FakeVLM ( ) ,
ocr = FakeOCR ( ) ,
prompts_dir = prompts_dir ,
skills = skills_registry ,
embed_fn = _fake_embed_fn ,
verify_vision = False ,
anchor = False ,
assemble_mode = " ids " ,
)
class TestViewNodeEntityAppendix :
@pytest.mark.asyncio
async def test_view_node_appends_entity_blocks (
self , monkeypatch , entity_dispatcher : SearchToolDispatcher
) - > None :
""" 摘要后必须出现 [实体]/[画面文字] 区块(确定性追加,不经 LLM)。 """
async def _stub_summarize ( * args , * * kwargs ) - > str :
return " [内容摘要] 与问题无关的摘要 "
monkeypatch . setattr ( " app.search.tools.summarize_node " , _stub_summarize )
result = await entity_dispatcher . dispatch (
" view_node " ,
{ " node_id " : " vid_L1_000_L2_000 " , " question " : " 耳机戴哪只耳? " } ,
context = { } ,
)
assert " [实体] " in result
assert " Bluetooth headset (both ears) " in result
assert " [画面文字] " in result
assert " $9.99 " in result
```
( L2 无 children → `summarize_children` 不会被触发,无需 stub。)
- [ ] **Step 6: 运行确认失败 **
Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_search_tools.py::TestViewNodeEntityAppendix -v`
Expected: FAIL(输出无 `[实体]` 区块)。
- [ ] **Step 7: 实现 dispatcher 追加 **
`app/search/tools.py` `_handle_view_node` , Phase 2 摘要之后、Phase 3 子节点概览之前:
``` python
parts : list [ str ] = [
f " [节点] { node_id } | { level_label } | { time_str } " ,
" " ,
summary ,
]
# Phase 2.5: 确定性追加实体/画面文字原文(防按题摘要吞噬,Spec-1 B)
for label , text in self . _env . node_entity_fields ( node_id ) . items ( ) :
parts . append ( f " [ { label } ] { text } " )
```
- [ ] **Step 8: 运行确认通过 + 全量回归 **
Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_search_tools.py tests/unit/test_tree_environment.py -v`
Expected: 全部 PASS。
Run: `make test`
Expected: 全绿,覆盖率不降。
- [ ] **Step 9: Commit **
``` bash
git add app/tree/environment.py app/search/tools.py tests/unit/test_tree_environment.py tests/unit/test_search_tools.py
git commit -m "feat(search): append raw entity fields after view_node summary"
```
---
### Task 4: 端到端验证(真实 benchmark 抽样重跑)
**Files: ** 无新文件(验证性任务)
- [ ] **Step 1: 抽样重跑(24 题,tmux + 无缓存日志) **
``` bash
tmux new-session -d -s spec1check
tmux send-keys -t spec1check "cd /home/iomgaa/Projects/Video-Tree-TRM5 && CUDA_VISIBLE_DEVICES=0 N_SAMPLES=24 RUN_ID=spec1check bash scripts/infer_video_mme.sh" Enter
```
注:`infer_video_mme.sh` 不支持 RUN_ID 环境变量时,直接以 `conda run -n Video-Tree-TRM python main.py --workspace-dir workspaces/default --store-dir store --mode infer --concurrency 24 --max-steps 40 --skill-mode auto --n-samples 24 --questions benchmarks/Video-MME --run-id spec1check --skills-version v1 --prompts-version v1` 运行。
- [ ] **Step 2: 验证三项指标 **
``` bash
sqlite3 workspaces/default/harness.db "SELECT stop_reason, COUNT(*) FROM predictions WHERE run_id='infer_spec1check' GROUP BY stop_reason;"
```
Expected: 无 `parse_error` ( A1 生效);`error` 为 0 或仅真实网络故障(A2 生效)。
``` bash
sqlite3 workspaces/default/harness.db "SELECT steps_json FROM predictions WHERE run_id='infer_spec1check' LIMIT 1;" | grep -c "\[实体\]"
```
Expected: ≥1( B 生效:view_node 输出含实体区块)。
- [ ] **Step 3: 收尾 **
Run: `make lint && make test`
Expected: 全绿。
``` bash
git status # 确认无未预期改动
```
---
## Self-Review 记录
1. **Spec 覆盖 ** : A1( Task 1)、A2 含 20s/40s 与显式异常元组(Task 2)、B 含 dispatcher 侧追加与 TreeEnvironment 结构化提取(Task 3)、验证三件套(Task 4 + 各任务单测)——设计四节全覆盖。
2. **占位符扫描 ** : Task 3 Step 1 的 `_build_env_with_node(s)` 指向 `test_tree_environment.py` 现有构造模式,属"复用现有 fixture"指令而非 TBD;其余步骤均含完整代码/命令。
3. **类型一致性 ** : `node_entity_fields` 在 Task 3 Step 3 定义、Step 7 调用,签名一致;`step_retry_delays` 构造参数与测试断言 `[20.0, 40.0]` 一致。
## 核心算法保真校验
本计划涉及**算法 #10 Agent Loop**( `core/agent/loop.py` ):A1/A2 均为解析与异常路径的加固,不触碰 Thinking+JSON 协议、json_repair 兜底链、pluggy hook 时序与步数语义(解析失败重试不计步、工具无效不计步的现状行为在测试中有回归覆盖)。对照参考 `/home/iomgaa/Projects/Video-Tree-TRM4/core/loop.py` :本改动为 TRM5 新增韧性层,无迁移简化。其余 12 项算法不涉及。