fix: session resume, config_set schema, metrics chart, subagent logging

- Fix session resume: remove messages.length>1 guard so sessionStorage ID is used as resumeSessionId on first message (prevents new session creation when user sends "继续") - Fix config_set tool schema: add missing `items: {}` to array type in oneOf (Azure OpenAI strict validation rejects it otherwise) - Metrics chart: use actual target set filename from trigger text instead of generic "target_subset" key - Add sub-agent skill call logging to agent_runtime.py (full prompt/response, no truncation) - Various metric enrichment and dedup fixes in server.py Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-05-27 17:02:24 +08:00
parent 89e77140e7
commit 85ce8c0e69
8 changed files with 411 additions and 40 deletions
@@ -1751,11 +1751,17 @@ def create_app(state: AgentState) -> FastAPI:
        if isinstance(run_id, str) and run_id.strip():
            rid = run_id.strip()
            per_round = _resolve_run_dic_per_round(state_entries)
-            # augment 产物的 runDic = resolve_run_ids.sh 在当轮 eval 落盘后取到的
-            # max(已落盘 workflow) = 当轮 eval 自身的 workflow id，即 per_round[R{n}]
-            # （不是 R{n-1}，脚本取的是 eval 之后的 max，此时 max 已经包含当轮 eval）
+            # 正确标号: R{n}·Train 和 R{n}·Analysis 共用同一 run_id，
+            # per_round[R{n}] 从 Analysis 的 watch/log 中解析到 runDic。
+            # 兼容旧 session 错误标号: agent 每阶段递增 run_id（R7=train, R6=analysis），
+            # 导致 per_round[R_train] = None，需 fallback 到 R{n-1}。
            if step == 'augment':
-                pass  # 直接走下面 per_round.get(rid) 取本轮 runDic
+                run_dic = per_round.get(rid)
+                if run_dic is None:
+                    run_index = _parse_run_index(rid)
+                    if run_index is not None and run_index >= 1:
+                        prev_rid = f'R{run_index - 1}'
+                        run_dic = per_round.get(prev_rid)
            if run_dic is None:
                run_dic = per_round.get(rid)
        if run_dic is None:
@@ -2001,7 +2007,14 @@ def create_app(state: AgentState) -> FastAPI:
    ) -> Response:
        sessions_dir = state.account_paths(account_id)['sessions']
        state_path = _session_state_path(sessions_dir, session_id)
+        await asyncio.to_thread(
+            _sync_remote_program_state, state, account_id, session_id, state_path,
+        )
        state_entries, _ = _read_program_state(state_path)
+        trigger = _extract_trigger_from_state(state_entries) or _extract_trigger_from_session(
+            sessions_dir, session_id
+        )
+        _target_set_name = _parse_target_set(trigger, model_config=state.model_config_for(account_id))
        iter_count = await asyncio.to_thread(
            _read_iteration_log_count,
            session_id,
@@ -2013,6 +2026,7 @@ def create_app(state: AgentState) -> FastAPI:
            session_id,
            agent_state=state,
            account_id=account_id,
+            target_set_name=_target_set_name,
        )
        payload = await asyncio.to_thread(
            _hardcoded_autoresearch_pipeline,
@@ -2060,6 +2074,12 @@ def create_app(state: AgentState) -> FastAPI:
        async def event_gen():
            last_signature: str | None = None
            last_heartbeat = time.monotonic()
+            last_remote_sync = 0.0
+            _state_entries_init, _ = _read_program_state(state_path)
+            _trigger_init = _extract_trigger_from_state(_state_entries_init) or _extract_trigger_from_session(
+                sessions_dir, session_id
+            )
+            _stream_target_set = _parse_target_set(_trigger_init, model_config=state.model_config_for(account_id))
            refresher = asyncio.create_task(
                _iter_count_refresh_loop(
                    session_id,
@@ -2072,12 +2092,19 @@ def create_app(state: AgentState) -> FastAPI:
                    session_id,
                    agent_state=state,
                    account_id=account_id,
+                    target_set_name=_stream_target_set,
                )
            )
            try:
                while True:
                    if await request.is_disconnected():
                        break
+                    now_mono = time.monotonic()
+                    if now_mono - last_remote_sync >= 5.0:
+                        await asyncio.to_thread(
+                            _sync_remote_program_state, state, account_id, session_id, state_path,
+                        )
+                        last_remote_sync = now_mono
                    state_entries, state_mtime = _read_program_state(state_path)
                    iter_count = await asyncio.to_thread(
                        _read_iteration_log_count,
@@ -2090,6 +2117,7 @@ def create_app(state: AgentState) -> FastAPI:
                        session_id,
                        agent_state=state,
                        account_id=account_id,
+                        target_set_name=_stream_target_set,
                    )
                    payload = await asyncio.to_thread(
                        _hardcoded_autoresearch_pipeline,
@@ -5714,11 +5742,43 @@ def _sync_remote_program_state(
    new_content = result.stdout or ''
    if not new_content.strip():
        return
-    # Skip rewrite if local already matches remote (avoid touching mtime, which
+    # Preserve local-only synthetic gate entries that remote doesn't have.
+    local_synthetics: list[str] = []
+    try:
+        if local_state_path.is_file():
+            # Parse remote gate keys once
+            remote_gate_keys: set[str] = set()
+            for raw_line in new_content.splitlines():
+                if not raw_line.strip():
+                    continue
+                try:
+                    r = json.loads(raw_line)
+                    if r.get('step') and not r.get('_synthetic'):
+                        remote_gate_keys.add(f"{r.get('run_id','')}:{r.get('step','')}")
+                except (json.JSONDecodeError, ValueError):
+                    pass
+            for line in local_state_path.read_text(encoding='utf-8').splitlines():
+                if '"_synthetic"' not in line:
+                    continue
+                try:
+                    obj = json.loads(line)
+                    if obj.get('_synthetic'):
+                        gate_key = f"{obj.get('run_id','')}:{obj.get('step','')}"
+                        if gate_key not in remote_gate_keys:
+                            local_synthetics.append(line)
+                except (json.JSONDecodeError, AttributeError):
+                    pass
+    except OSError:
+        pass
+    merged = new_content.rstrip('\n')
+    if local_synthetics:
+        merged += '\n' + '\n'.join(local_synthetics)
+    merged += '\n'
+    # Skip rewrite if local already matches (avoid touching mtime, which
    # SSE uses to detect changes).
    try:
        if local_state_path.is_file() and (
-            local_state_path.read_text(encoding='utf-8') == new_content
+            local_state_path.read_text(encoding='utf-8') == merged
        ):
            return
    except OSError:
@@ -5726,7 +5786,7 @@ def _sync_remote_program_state(
    try:
        local_state_path.parent.mkdir(parents=True, exist_ok=True)
        tmp = local_state_path.with_suffix('.jsonl.sync-tmp')
-        tmp.write_text(new_content, encoding='utf-8')
+        tmp.write_text(merged, encoding='utf-8')
        tmp.replace(local_state_path)
    except OSError:
        return
@@ -6055,7 +6115,7 @@ _STEP_KIND: dict[str, str] = {
    'dist-analysis': 'analysis',
    'hypothesis':    'analysis',
    'log':           'analysis',
-    'augment':       'train',
+    'augment':       'analysis',
    'verify':        'train',
    'sft':           'train',
    'human-check':   'gate',
@@ -6083,8 +6143,10 @@ _CARD_META: dict[tuple[str, str], dict[str, Any]] = {
    ('dist-analysis', 'baseline'): {'icon': 'trending-up',    'title': '分层结果分析 & 报告', 'subtitle': '根因归类 + workflow<runDic>.md'},
    ('hypothesis', 'baseline'):    {'icon': 'dna',            'title': '形成假设',         'subtitle': 'iteration_log hypothesis 字段'},
    ('log', 'baseline'):           {'icon': 'clipboard-list', 'title': '记录迭代日志',     'subtitle': 'iteration_log.jsonl'},
+    # augment — lives under previous round's analysis/baseline
+    ('augment', 'baseline'):       {'icon': 'flask',         'title': '数据增强',         'subtitle': 'augment_<runDic>.jsonl'},
+    ('augment', 'analysis'):       {'icon': 'flask',         'title': '数据增强',         'subtitle': 'augment_<runDic>.jsonl'},
    # R{n}·Train
-    ('augment', 'train'):    {'icon': 'flask',         'title': '数据增强',         'subtitle': 'augment_<runDic>.jsonl'},
    ('verify', 'train'):     {'icon': 'shield',        'title': '修改返回验证',     'subtitle': 'sanity + 人审'},
    ('sft', 'train'):        {'icon': 'graduation-cap','title': 'SFT 训练',        'subtitle': 'submit_sft_via_cml.sh'},
    # R{n}·Analysis
@@ -6103,16 +6165,17 @@ _CARD_ORDER: dict[tuple[str, str], int] = {
    ('dist-analysis', 'baseline'): 2,
    ('hypothesis', 'baseline'): 3,
    ('log', 'baseline'): 4,
+    ('augment', 'baseline'): 5,
    # train
-    ('augment', 'train'): 0,
-    ('verify', 'train'): 1,
-    ('sft', 'train'): 2,
+    ('verify', 'train'): 0,
+    ('sft', 'train'): 1,
    # analysis
    ('cml', 'analysis'): 0,
    ('gold-drift', 'analysis'): 1,
    ('dist-analysis', 'analysis'): 2,
    ('hypothesis', 'analysis'): 3,
    ('log', 'analysis'): 4,
+    ('augment', 'analysis'): 5,
 }

 _SECTION_LABELS: dict[str, str] = {
@@ -6253,9 +6316,9 @@ def _derive_section_status(card_statuses: list[str]) -> str:
        return 'pending'
    if all(s == 'complete' for s in card_statuses):
        return 'complete'
-    if 'failed' in card_statuses and not any(s in ('running', 'complete') for s in card_statuses):
+    if 'failed' in card_statuses and not any(s in ('running', 'complete', 'waiting') for s in card_statuses):
        return 'failed'
-    if 'running' in card_statuses or 'complete' in card_statuses:
+    if any(s in ('running', 'complete', 'waiting') for s in card_statuses):
        return 'running'
    return 'pending'

@@ -6315,6 +6378,11 @@ def _build_pipeline_items(
        if not run_id:
            continue
        last_non_gate_run_id = run_id
+        # augment belongs to the previous round's analysis/baseline section
+        if step == 'augment':
+            aug_index = _parse_run_index(run_id)
+            if aug_index is not None and aug_index >= 1:
+                run_id = f'R{aug_index - 1}'
        run_index = _parse_run_index(run_id)
        if run_index is None:
            continue
@@ -6375,6 +6443,11 @@ def _build_pipeline_items(
                card['progress'] = max(0.0, min(1.0, value))
            except (TypeError, ValueError):
                pass
+        if step == 'augment' and 'count' in entry:
+            try:
+                card['subtitle'] = f'{int(entry["count"])} 条新增样本'
+            except (TypeError, ValueError):
+                pass
        sections[section_key]['cards'].append(card)

    # 3. Sort cards within each section.
@@ -6451,7 +6524,7 @@ def _build_pipeline_items(
            'summary': entry.get('summary'),
            'proposal': entry.get('proposal'),
            'ask': entry.get('ask'),
-            '_order_key': (run_index, 99, gate_position_index.get(gate_key, 0)),
+            '_order_key': entry.get('ts', ''),
        })

    # 6. Filter: drop empty/pending sections (no cards or all pending). Running
@@ -6589,12 +6662,18 @@ def _compute_in_flight(
                progress = max(0.0, min(1.0, value))
            except (TypeError, ValueError):
                pass
+        subtitle = meta['subtitle']
+        if step == 'augment' and 'count' in entry:
+            try:
+                subtitle = f'{int(entry["count"])} 条新增样本'
+            except (TypeError, ValueError):
+                pass
        cards.append({
            'key': f'{target_run_id}:{step}',
            'step': step,
            'icon': meta['icon'],
            'title': meta['title'],
-            'subtitle': meta['subtitle'],
+            'subtitle': subtitle,
            'status': status,
            'progress': progress,
        })
@@ -6679,7 +6758,7 @@ def _maybe_inject_gate_fallback(
        'status': 'running',
        'run_id': run_id or 'R0',
        'reason': last_text[-300:],
-        'ts': datetime_utc_iso(),
+        'ts': time.strftime('%Y-%m-%dT%H:%M:%S+08:00', time.localtime()),
        '_synthetic': True,
    }
    try:
@@ -7033,6 +7112,7 @@ def _read_iteration_log_metrics_history(
    *,
    agent_state: 'AgentState | None' = None,
    account_id: str | None = None,
+    target_set_name: str | None = None,
 ) -> dict[str, Any] | None:
    """Cached read of full iteration_log.jsonl as metrics time-series."""
    cache_key = _iter_count_cache_key(session_id, account_id)
@@ -7040,6 +7120,7 @@ def _read_iteration_log_metrics_history(
        return _METRICS_HISTORY_CACHE[cache_key]
    value = _read_iteration_log_metrics_history_uncached(
        session_id, agent_state=agent_state, account_id=account_id,
+        target_set_name=target_set_name,
    )
    _METRICS_HISTORY_CACHE[cache_key] = value
    return value
@@ -7050,6 +7131,7 @@ async def _metrics_history_refresh_loop(
    *,
    agent_state: 'AgentState | None',
    account_id: str | None,
+    target_set_name: str | None = None,
    interval_seconds: float = 3.0,
 ) -> None:
    cache_key = _iter_count_cache_key(session_id, account_id)
@@ -7060,6 +7142,7 @@ async def _metrics_history_refresh_loop(
                session_id,
                agent_state=agent_state,
                account_id=account_id,
+                target_set_name=target_set_name,
            )
            _METRICS_HISTORY_CACHE[cache_key] = value
        except Exception:  # noqa: BLE001
@@ -7072,6 +7155,7 @@ def _read_iteration_log_metrics_history_uncached(
    *,
    agent_state: 'AgentState | None' = None,
    account_id: str | None = None,
+    target_set_name: str | None = None,
 ) -> dict[str, Any] | None:
    """Parse iteration_log.jsonl into {'metrics': [keys], 'rounds': [...]}.

@@ -7164,8 +7248,70 @@ def _read_iteration_log_metrics_history_uncached(
            'values': values,
        })

+    # Deduplicate: keep only the last entry per (iteration, runDic).
+    # Agent may write multiple entries for the same round (e.g. re-running
+    # dist-analysis); the last one has the most complete metrics.
+    seen_keys: dict[tuple[int, int], int] = {}
+    for idx, r in enumerate(rounds):
+        key = (r['iteration'], r['runDic'])
+        seen_keys[key] = idx
+    rounds = [rounds[i] for i in sorted(seen_keys.values())]
+
+    # Rebuild metrics_order from surviving rounds only (dedup may have removed
+    # entries whose keys shouldn't appear in the chart).
+    metrics_order = []
+    metrics_seen = set()
+    for r in rounds:
+        for k in r['values']:
+            if k not in metrics_seen:
+                metrics_seen.add(k)
+                metrics_order.append(k)
+
    if not rounds:
        return None
+
+    # Enrich each round with KPI-equivalent metrics from lark_template.json.
+    # This ensures dapan_car / specific_test / target_subset are always present
+    # even if the agent omitted them from iteration_log.
+    target_key = target_set_name if target_set_name and target_set_name != '—' else 'target_subset'
+    _KPI_METRIC_MAP = {
+        'specific_test_pass_rate': 'specific_test',
+        'overall_car_pass_rate': 'dapan_car',
+        'target_set_pass_rate': target_key,
+    }
+    # Rename agent-written 'target_subset' to the actual target set name first,
+    # so enrichment doesn't create duplicates.
+    if target_key != 'target_subset':
+        if 'target_subset' in metrics_seen:
+            metrics_order = [target_key if m == 'target_subset' else m for m in metrics_order]
+            metrics_seen.discard('target_subset')
+            metrics_seen.add(target_key)
+        for r in rounds:
+            if 'target_subset' in r['values']:
+                r['values'][target_key] = r['values'].pop('target_subset')
+
+    for r in rounds:
+        run_dic = r.get('runDic')
+        if not isinstance(run_dic, int) or run_dic == 0:
+            continue
+        metric_diff = _read_workflow_metric_diff(
+            run_dic,
+            agent_state=agent_state,
+            account_id=account_id,
+            session_id=session_id,
+        )
+        if metric_diff is None:
+            continue
+        _enrich_metrics_from_eval_output(metric_diff, None)
+        for src_key, dst_key in _KPI_METRIC_MAP.items():
+            if dst_key not in r['values']:
+                raw = metric_diff.get(src_key)
+                if isinstance(raw, (int, float)):
+                    r['values'][dst_key] = round(raw * 100, 2)
+                    if dst_key not in metrics_seen:
+                        metrics_seen.add(dst_key)
+                        metrics_order.append(dst_key)
+
    return {'metrics': metrics_order, 'rounds': rounds}


@@ -7385,7 +7531,8 @@ def _parse_target_set(
 ) -> str | None:
    """Sync, non-blocking. Reads cache only — never makes the LLM call inline.
    Cache is filled by the watcher scanner background loop via
-    _refresh_target_set_cache_blocking (which is run in a thread)."""
+    _refresh_target_set_cache_blocking (which is run in a thread).
+    Falls back to regex extraction of .csv filenames from trigger text."""
    if not trigger:
        return None
    text = trigger.strip()
@@ -7394,7 +7541,14 @@ def _parse_target_set(
    if len(text) > 4000:
        text = text[:4000]
    cache_key = hashlib.sha256(text.encode('utf-8')).hexdigest()
-    return _TARGET_SET_LLM_CACHE.get(cache_key)
+    cached = _TARGET_SET_LLM_CACHE.get(cache_key)
+    if cached is not None:
+        return cached
+    # Regex fallback: extract .csv filename from trigger text.
+    m = re.search(r'(?:(?<=[的/，,\s])|(?<=^))([^\s，,。；的中是在]+\.csv)', text)
+    if m:
+        return m.group(1)
+    return None


 def _refresh_target_set_cache_blocking(
@@ -66,8 +66,7 @@ export const Assistant = ({ initialSessionId }: AssistantProps = {}) => {
 					const lastSessionId = getLastSessionId(messages);
 					const pendingWorkspaceSessionId =
 						messages.length <= 1 ? readPendingWorkspaceSessionId() : null;
-					const selectedResumeSessionId =
-						messages.length > 1 ? selectedSessionId : null;
+					const selectedResumeSessionId = selectedSessionId;
 					const outgoingSessionId =
 						lastSessionId ??
 						selectedResumeSessionId ??
@@ -28,9 +28,14 @@ const METRIC_LABELS: Record<string, string> = {
 	req_set_car: "需求集合(车载)",
 	dapan_car: "大盘车载",
 	specific_test: "Specific Test",
+	target_subset: "目标集合",
+	icl_test: "ICL Test",
+	overall: "大盘整体",
 	triage_err_rate: "Triage 错误率",
 	bvt_nav: "导航BVT",
 	talkable_controllable: "可聊可控",
+	badcase_2025: "Badcase 2025",
+	autotask: "自动任务",
 	target_subset_count: "目标子集数量",
 	target_subset_wrong: "目标子集错误数",
 	target_subset_correct: "目标子集正确数",
@@ -87,8 +87,8 @@ when_to_use: |
 - [ ] §4.0.1 Step B：量级判定基于 label-master 处理后的**残留候选量**走对应支线（≤50 自动按 label-master 推荐改 / 51-200 全量人审 / >200 触发 #3，全量交付）
 - [ ] §4.0.1 Step C：备份 `.bak` 文件
 - [ ] §4.1 输入准备 / §4.2 GPT 调用 / §4.3 sanity check / §4.4 写入
- [ ] **§4.5 label-master 标签复核（落盘后必做，强制）**：H1 `modified_samples.jsonl` 只跑层 1（格式）—— 语义已在 §4.0.1 Step A.5 完成；H2 `augment_<runDic>.jsonl` 跑两层（`validate_label_output.py` 格式 + Skill 调用 `label-master` 语义），**逐条 1:1 全量覆盖**（`augment` 多少行 `label_master_review.jsonl` H2 部分就多少行，禁止抽样 / spot-check / X/X pass 外推），verdict 写 `results/data_clean_<runDic>/label_master_review.jsonl`，**不通过必须 = 0**（任何不通过必须修正后重新 review 直到全 pass 才能进 Step 5）
- [ ] **§4.5 层 2 语义复核必须调用 `Skill(skill="label-master")`，禁止正则/规则脚本替代**：层 2 的本质是"用 label-master 知识体系对每条 (query, label) 做独立语义判定"。label-master 已标记 `repeatable: true`，可在同一 run 内多次调用。**禁止**：纯正则/关键词匹配脚本、"target 都一样所以直接 pass"逻辑、批量写 pass 不看 query 内容。已踩坑：R1 agent 写了个正则脚本充当"layer 2"，完全没走语义判断。
+- [ ] **§4.5 label-master 标签复核（落盘后必做，强制，H1+H2 都要）**：H1 `modified_samples.jsonl` **和** H2 `augment_<runDic>.jsonl` 都必须跑两层（`validate_label_output.py` 格式 + Skill 调用 `label-master` 语义），**逐条 1:1 全量覆盖**。§4.0.1 A.5 的预审是粗筛，**不能替代落盘前的 layer2 语义复核**。verdict 写 `results/data_clean_<runDic>/label_master_review.jsonl`，**不通过必须 = 0**（任何不通过必须修正后重新 review 直到全 pass 才能进 Step 5）
+- [ ] **§4.5 层 2 语义复核必须通过子 agent 调用 `Skill(skill="label-master")`，禁止主 agent 直接调、禁止正则/规则脚本替代**：主 agent 把待审列表写入 `scratchpad/lm_input.jsonl`，起子 agent（`delegate_agent(prompt="...", allow_shell=true)`）逐条调 label-master，结果写 `scratchpad/lm_output.jsonl`，主 agent 读取汇总。子 agent prompt 里给**绝对路径**。**禁止**：主 agent 直接调 Skill(skill="label-master")、纯正则/关键词匹配脚本、"target 都一样所以直接 pass"逻辑、批量写 pass 不看 query 内容。

 ### Step 4 → Step 5 边界（**NEVER STOP 硬连接**，反复踩坑）
 - [ ] 写完 `augment=complete` 那一刻，**同一轮 bash 不许结束**：紧接着跑 §4.5 label-master 复核 → 写 `sft=running` → 调 `submit_sft.sh` → 挂 watcher（评测在 SFT `_SUCCESS` 落盘后的下一轮单独用 `submit_cml_eval.sh` 起，不要在 SFT bg 里串接评测）
@@ -128,17 +128,45 @@ when_to_use: |
 > echo '{"step":"cml","status":"complete","ts":"'$(date -Iseconds)'"}' >> "$SESSION_OUTPUT/program-state.jsonl"
 > ```

-### R1.5 — 推荐每条 state entry 带 run_id（让 UI 准确归位 round）
+### R1.5 — 每条 state entry 必须带 run_id（让 UI 准确归位 round）

-新版 UI 按 R0 / R1 / R2 切分 sections。**强烈建议**每行加 `run_id`：
+新版 UI 按 R0 / R1 / R2 切分 sections。每个 R{n>=1} 同时包含 **Train** 和 **Analysis** 两个 section。**必须**每行带 `run_id`。
+
+⚠️ **核心规则：run_id 只在 hypothesis=complete 之后递增。** 一个完整迭代（augment → sft → eval → analysis → hypothesis）全程使用同一个 run_id。
+
+完整示例（一轮完整迭代 R1）：

 ```bash
-echo '{"step":"cml","status":"running","run_id":"R0","ts":"'$(date -Iseconds)'"}' >> $S
-echo '{"step":"hypothesis","status":"complete","run_id":"R0","ts":"..."}' >> $S    # hypothesis 跟当前轮（R0），不是 R1
-echo '{"step":"augment","status":"running","run_id":"R1","ts":"..."}' >> $S        # augment 才是 R1·Train 起点
+# ─── R0·Baseline ───
+echo '{"step":"cml","status":"running","run_id":"R0"}' >> $S
+echo '{"step":"dist-analysis","status":"complete","run_id":"R0"}' >> $S
+echo '{"step":"hypothesis","status":"complete","run_id":"R0"}' >> $S
+# ← hypothesis=complete → 递增 run_id
+
+# ─── R1·Train ───
+echo '{"step":"augment","status":"running","run_id":"R1"}' >> $S
+echo '{"step":"augment","status":"complete","run_id":"R1","count":80}' >> $S
+echo '{"step":"sft","status":"running","run_id":"R1"}' >> $S
+echo '{"step":"sft","status":"complete","run_id":"R1"}' >> $S
+
+# ─── R1·Analysis（SFT 后的 eval 仍然是 R1，不是 R2！）───
+echo '{"step":"cml","status":"running","run_id":"R1"}' >> $S
+echo '{"step":"cml","status":"complete","run_id":"R1"}' >> $S
+echo '{"step":"dist-analysis","status":"complete","run_id":"R1"}' >> $S
+echo '{"step":"hypothesis","status":"complete","run_id":"R1"}' >> $S
+# ← hypothesis=complete → 递增 run_id
+
+# ─── R2·Train ───
+echo '{"step":"augment","status":"running","run_id":"R2"}' >> $S
+# ...
 ```

-run_id 缺省时后端会按 `iteration_log.jsonl` 行数自动推断（analysis 类 step 含 hypothesis → R{count}，train 类 step augment/verify/sft → R{count+1}），但显式写更准——尤其是并发跨轮、补写历史 entry、或要在 R0 强制注入 baseline 时。
+**错误（禁止）**：eval 后递增 run_id，导致 train 和 analysis 分属不同 round：
+```
+R1: augment+sft → R2: cml+analysis+hypothesis → R3: augment+sft → R4: cml+analysis ...
+```
+
+**为什么重要**：后端通过 `per_round[R{n}]` 从 R{n}·Analysis 的 watch/log entry 解析 runDic，再用这个 runDic 去找 R{n}·Train 的 augment 产物文件。如果 train 和 analysis 用了不同 run_id，augment 的产物文件找不到。

 ### R2 — 写入即生效，最后一行覆盖前面

@@ -465,6 +465,11 @@ def cross_iter_tag(case_h: str, last_runDic: int, registry: dict) -> str:

 #### 4.0 原始训练数据清洗（增强前必做）

+> 🚨 **强制执行：每轮 augment 必须先跑 §4.0→§4.0.1，不得跳过直接做 H2 仿写。**
+> 即使 hypothesis 归因为"训练集缺数据"，也必须先逐 pattern 检索训练集确认是否存在 mislabel。只有当 §4.0.1 Step A 扫描结果 + Step A.5 label-master 预审后残留候选 = 0 时，才能判定"无需 H1 修改"并跳到 H2。**"这轮只需要加数据"不是跳过 H1 检查的合法理由**——上一轮加的新数据可能引入了新的标签冲突，必须每轮重新检索确认。
+>
+> 缺少 H1 检查的 augment 视为不完整：`modified_samples.jsonl` 可以为空（代表确认无需修改），但 `data_clean_<runDic>.log` 必须记录"H1 扫描完成，0 候选"的结论，否则 §5.0 准入检查拒绝启动 SFT。
+
 数据增强不仅仅是加数据，还必须对原数据集中的错误/不一致标签进行清洗，否则新增数据和旧数据矛盾，模型学不好。

 **清洗原则（case 驱动，逐类分析）**：
@@ -542,9 +547,12 @@ with open(out_csv, 'w', encoding='utf-8-sig') as fp:
 具体步骤：

 1. 抽出每条候选的 `(file, line, query, old_label, suspected_new_label)`
-2. **逐条**调 `Skill(skill="label-master", args=...)`——每次只传**一条** `(query, old_label)`（**不**给 suspected_new_label，避免锚定），让 label-master 按 §决策流程 / §候选召回索引 / §高频混淆边界 输出 → `{verdict: 通过 | 不通过, 推荐标签, 排除理由, 易混淆边界}`。
-   - ⚠️ **禁止把多条 query 塞进同一次 Skill 调用**——批量调用会让 label-master 在多条之间相互锚定，分错率显著升高（已踩坑）
-   - 可在同一轮 assistant response 里并发多次 Skill 调用（建议 ≤ 8 路并发）以提升吞吐
+2. **用子 agent 调用 label-master**——**禁止在主 agent 上下文里直接调 Skill(skill="label-master")**，避免大量输出污染主 agent 上下文。做法：
+   - 主 agent 把待审列表写入 jsonl 文件（如 `scratchpad/lm_input.jsonl`，每行 `{query, old_label}`）
+   - 起子 agent：`delegate_agent(prompt="读取 <workspace>/scratchpad/lm_input.jsonl，对每条逐一调用 Skill(skill='label-master', args='query: ... label: ...')，把结果逐行 append 到 <workspace>/scratchpad/lm_output.jsonl（每行 {query, verdict, 推荐标签, 理由}）。注意：每次 Skill 调用只传一条 query，禁止批量。", allow_shell=true)`
+   - 子 agent 完成后主 agent 读取 `lm_output.jsonl` 汇总结果
+   - ⚠️ **禁止把多条 query 塞进同一次 Skill 调用**——批量调用会让 label-master 在多条之间相互锚定（已踩坑）
+   - ⚠️ 子 agent prompt 里必须给出**完整的 workspace 绝对路径**（`$AUTORESEARCH_CHAT_ROOT/scratchpad/...`），因为子 agent 没有父的环境变量
 3. 把 label-master 的"推荐标签"**回写到** `relabel_candidates_<runDic>.csv` 覆盖原 `建议新label` 列；新增列 `verdict`、`label_master_理由`，便于回查
 4. 收尾时按以下规则筛 candidate list（**残留候选 = 真正进入 Step B 的列表**）：
   - `推荐标签 == old_label`：label-master 不认同要改 → 从候选里**剔除**（这条原 label 可能本来就是对的）
@@ -571,7 +579,13 @@ done

 同时落归档到 `results/data_clean_<runDic>/`：
 - `deleted_samples.jsonl`：被删样本原文
- `modified_samples.jsonl`：被改样本（含 before/after output）
+- `modified_samples.jsonl`：被改样本，**必须包含以下字段**：
+  - `line_idx`：原文件行号
+  - `file`：原文件路径
+  - `query`：**完整 query**（用 `extract_query(instruction)` 从原始 instruction 提取，**禁止截断**）
+  - `old_output`：修改前 output
+  - `new_output`：修改后 output
+  - 可选：`instruction_query_excerpt`（仅供人工快速浏览，允许截断，但**不得作为 §4.5 复核的 query 来源**）
 - `data_clean_<runDic>.log`：摘要 + 影响 pattern + 样本数

 **Step D：修改执行（按文件批量，避免重复读写）**
@@ -882,10 +896,35 @@ JSONL 行里的 `sub_cate` 字段仅用于内部路由/去重，归档时丢弃

 | 文件 | 层 1（格式） | 层 2（语义，`Skill(skill="label-master")`） |
 |---|---|---|
-| H1 `modified_samples.jsonl` | ✅ `validate_label_output.py --field output_after` | ❌ 已在 §4.0.1 A.5 完成 |
+| H1 `modified_samples.jsonl` | ✅ `validate_label_output.py --field output_after` | ✅ 逐条调 label-master（1:1 全量）— §4.0.1 A.5 是粗筛，**不能替代落盘前的 layer2** |
 | H2 `augment_<runDic>.jsonl` | ✅ `validate_label_output.py --field output` | ✅ 逐条调 label-master（1:1 全量） |

-⛔ **层 2 禁止用正则/规则脚本替代**，必须调 `Skill(skill="label-master")`（`repeatable: true`）。每次只传一条 `(query, label)`，禁止批量。
+⛔ **层 2 禁止用正则/规则脚本替代**，必须通过**子 agent** 调 `Skill(skill="label-master")`（`repeatable: true`）。每次只传一条 `(query, label)`，禁止批量。**禁止主 agent 直接调用 label-master**——上下文污染会导致主 agent 后续推理质量下降。子 agent 通过文件交换结果（主 agent 写 input jsonl → 子 agent 读取并逐条调 Skill → 写 output jsonl → 主 agent 读取汇总），见 §4.0.1 Step A.5 的详细做法。
+
+⚠️ **层 2 必须校验完整 label（含 tag），不能只校验 complex 维度**。
+
+子 agent 调用 label-master 时的 args 格式：`query: <完整query> label: <完整output>`，例如：
+```
+Skill(skill="label-master", args="query: 顺路再去个加油站 label: Agent(tag=\"地图导航\")")
+```
+
+label-master 会按其决策流程判断该 query 的 **tag 归属**是否正确（如应该是"地图导航"还是"充电加油"），同时判断 **complex**（Agent vs ComplexTask）和**输出格式**。
+
+**禁止只传 complex=true/false 让 label-master 做二分类**——这不是 label-master 的设计用途。必须传完整的 `Agent(tag="xxx")` 或 `ComplexTask(tag="xxx")`，让 label-master 走完整的"候选召回→标签卡片→边界判定→推荐标签"流程。
+
+label-master 的 verdict 必须同时覆盖：
+1. **tag 是否正确**：label-master 推荐的 tag 与当前 label 中的 tag 是否一致，不一致则 verdict=不通过
+2. **complex 是否正确**：Agent vs ComplexTask 是否正确
+3. **输出格式是否合规**：`Agent(tag="xxx")` / `ComplexTask(tag="xxx")` 格式是否规范
+
+`label_master_review.jsonl` 每行必须包含字段：`query`、`label`（完整 output，如 `Agent(tag="地图导航")`）、`verdict`（通过/不通过）、`recommended_label`（label-master 推荐的完整 output）、`reason`（判断依据，需说明 tag 判定理由）。
+
+**以下情况视为不合格 review，§5.0 准入检查拒绝启动 SFT**：
+- review 行中缺少 `label` 或 `recommended_label` 字段
+- `label` 字段只含 complex=true/false 而非完整 output
+- `reason` 中只提及 complex 判定而未提及 tag 归属判断
+
+⚠️ **H1 query 来源**：review 时传给 label-master 的 `query` **必须**取自 `modified_samples.jsonl` 的 `query` 字段（Step C 已要求写入完整 query）。**禁止**从 `instruction_query_excerpt` 提取——该字段可能被截断导致 `extract_query` 返回空值。如果 `query` 字段缺失（旧格式兼容），必须用 `line_idx` + `file` 回原始训练文件读取完整 instruction 再 `extract_query`。

 **覆盖率硬规则**：`label_master_review.jsonl` H2 行数 = `augment_<runDic>.jsonl` 行数（§5.0 `wc -l` 断言会拦）。不通过 > 0 则必须修正后重新 review 直到全 pass。verdict 文件不存在 → Step 5 拒绝启动。

@@ -893,6 +932,12 @@ JSONL 行里的 `sub_cate` 字段仅用于内部路由/去重，归档时丢弃

 🚨 **Step 4 → Step 5 硬连接**：`augment=complete` 后同一轮**紧接着**：① 写 `sft=running` ② §4.5 复核 ③ 复核全过 → `submit_sft.sh` + 挂 watcher。禁止 turn 结束、禁止写简报、禁止等回调。评测在 SFT `_SUCCESS` 落盘后的下一轮单独用 `submit_cml_eval.sh` 起，不串进 SFT bg。

+⚠️ **写 `augment=complete` 时必须附带 `"count"` 字段**，值为本轮最终写入 `augment_<runDic>.jsonl` 的样本行数（经 dedup + label-master 过滤后的实际数）。示例：
+```jsonl
+{"step":"augment","status":"complete","run_id":"R5","count":49,"ts":"2026-05-26T21:51:32+08:00"}
+```
+Pipeline panel 用此字段展示增强条数；缺失则只显示文件名。
+
 使用 `prepare_and_train_sft.py` 完成数据组装和训练。

 #### 5.0 环境准备
@@ -922,6 +967,10 @@ if [ -f "$AUG" ]; then
  fi
 fi

+# 1c. H1 检查日志必须存在（§4.0 强制要求，即使无修改也要记录扫描结论）
+CLEAN_LOG="$AUTORESEARCH_CHAT_ROOT/results/data_clean_${RUNDIC}/data_clean_${RUNDIC}.log"
+[ -f "$CLEAN_LOG" ] || { echo "data_clean 日志不存在，说明 §4.0 H1 检查未执行，回 §4.0"; exit 1; }
+
 # 2. zk_trainer 仓库已 clone（URL 写死，不要换命名空间，clone 失败先看 ssh key）
 cd "$AUTORESEARCH_CHAT_ROOT"
 [ -d zk_trainer ] || git clone git@git.n.xiaomi.com:wangsenhao/zk_trainer.git
@@ -1029,6 +1078,14 @@ eval "$(./scripts/resolve_run_ids.sh)"

 每轮写入 `results/iteration_log.jsonl` 一行，schema：

+> 🚨 **`results` 字段必须包含以下固定指标（每轮都要，不可遗漏）**：
+> - `specific_test`：Specific Test 准确率
+> - `dapan_car`：大盘车载准确率
+> - `target_subset`：目标集合准确率（= 当前需求子集）
+> - `icl_test`：ICL Test 准确率
+>
+> 这些指标从 `lark_template.json` 的 metric_diff 中提取。如果某个指标在评测结果中确实不存在（如首次评测缺少某子集），写 `null` 而不是省略 key——前端 metrics 折线图依赖每轮 key 的一致性，缺 key 会导致数据点丢失。
+
 ```json
 {
  "iteration": 12,
@@ -1041,8 +1098,9 @@ eval "$(./scripts/resolve_run_ids.sh)"
  },
  "prediction": {"req_set_car": 93.0, "triage_err_rate": 0.15},
  "results": {
-    "req_set_car": 92.1, "dapan_car": 96.30,
-    "specific_test": 95.10, "triage_err_rate": 0.18
+    "specific_test": 95.10, "dapan_car": 96.30,
+    "target_subset": 92.1, "icl_test": 86.11,
+    "triage_err_rate": 0.18
  },
  "verdict": "partial",
  "root_cause_findings": [
@@ -154,6 +154,119 @@ def _prepend_runtime_context(prompt: str, runtime_context: str | None) -> str:
    )


+def _log_child_skill_calls(
+    child_result: 'AgentRunResult',
+    child_agent: 'LocalCodingAgent',
+    subtask_label: str,
+) -> None:
+    """Extract Skill tool calls from a child agent's transcript and write to a log file.
+
+    Logs each Skill invocation with the args sent and the agent's response,
+    so we can audit whether sub-agents actually called skills like label-master.
+    """
+    from .agent_types import AgentRunResult  # noqa: F811
+
+    transcript = child_result.transcript
+    if not transcript:
+        return
+    log_dir = child_agent.runtime_config.session_directory / '_subagent_logs'
+    try:
+        log_dir.mkdir(parents=True, exist_ok=True)
+    except OSError:
+        return
+    session_id = child_result.session_id or 'unknown'
+    log_path = log_dir / f'{session_id}.jsonl'
+
+    skill_calls: list[dict[str, object]] = []
+    # Build a map of tool_call_id -> skill args from assistant messages with tool_calls
+    pending_skills: dict[str, dict[str, str]] = {}
+    for entry in transcript:
+        if not isinstance(entry, dict):
+            continue
+        role = entry.get('role')
+        if role == 'assistant':
+            tool_calls = entry.get('tool_calls')
+            if isinstance(tool_calls, list):
+                for tc in tool_calls:
+                    if not isinstance(tc, dict):
+                        continue
+                    fn = tc.get('function') or {}
+                    if not isinstance(fn, dict):
+                        continue
+                    if fn.get('name') == 'Skill':
+                        tc_id = tc.get('id', '')
+                        raw_args = fn.get('arguments', '{}')
+                        try:
+                            args_dict = json.loads(raw_args) if isinstance(raw_args, str) else raw_args
+                        except (json.JSONDecodeError, TypeError):
+                            args_dict = {'raw': raw_args}
+                        pending_skills[tc_id] = {
+                            'skill': args_dict.get('skill', ''),
+                            'args': args_dict.get('args', ''),
+                        }
+        elif role == 'tool':
+            tc_id = entry.get('tool_call_id', '')
+            if tc_id in pending_skills:
+                skill_info = pending_skills.pop(tc_id)
+                skill_calls.append({
+                    'skill': skill_info['skill'],
+                    'args': skill_info['args'],
+                    'prompt_injected': entry.get('content') or '',
+                    'ts': datetime.now(timezone.utc).isoformat(),
+                })
+
+    # Also capture the assistant responses that follow skill injections
+    # by looking at assistant messages after tool messages
+    assistant_after_skill: list[str] = []
+    saw_skill_tool = False
+    for entry in transcript:
+        if not isinstance(entry, dict):
+            continue
+        role = entry.get('role')
+        if role == 'tool' and entry.get('tool_call_id', '') in [
+            tc.get('id', '') for tc in _all_skill_tool_call_ids(transcript)
+        ]:
+            saw_skill_tool = True
+        elif role == 'assistant' and saw_skill_tool:
+            content = entry.get('content', '')
+            if content:
+                assistant_after_skill.append(content)
+            saw_skill_tool = False
+
+    # Merge assistant responses into skill_calls
+    for i, response_text in enumerate(assistant_after_skill):
+        if i < len(skill_calls):
+            skill_calls[i]['agent_response'] = response_text
+
+    if not skill_calls:
+        return
+    try:
+        with open(log_path, 'a', encoding='utf-8') as f:
+            for call in skill_calls:
+                call['subtask_label'] = subtask_label
+                f.write(json.dumps(call, ensure_ascii=False) + '\n')
+    except OSError:
+        pass
+
+
+def _all_skill_tool_call_ids(transcript: tuple[dict[str, object], ...]) -> list[dict[str, str]]:
+    """Collect all tool_call entries for Skill from assistant messages."""
+    results = []
+    for entry in transcript:
+        if not isinstance(entry, dict) or entry.get('role') != 'assistant':
+            continue
+        tool_calls = entry.get('tool_calls')
+        if not isinstance(tool_calls, list):
+            continue
+        for tc in tool_calls:
+            if not isinstance(tc, dict):
+                continue
+            fn = tc.get('function') or {}
+            if isinstance(fn, dict) and fn.get('name') == 'Skill':
+                results.append(tc)
+    return results
+
+
@dataclass
 class LocalCodingAgent:
    model_config: ModelConfig
@@ -3052,6 +3165,11 @@ class LocalCodingAgent:
                    managed_child_index=index,
                    managed_label=subtask_label,
                )
+                if self.tool_context.jupyter_runtime is not None:
+                    child_agent.tool_context = replace(
+                        child_agent.tool_context,
+                        jupyter_runtime=self.tool_context.jupyter_runtime,
+                    )
                if group_id is not None and child_agent.managed_agent_id is not None:
                    self.agent_manager.register_group_child(
                        group_id,
@@ -3118,9 +3236,11 @@ class LocalCodingAgent:
                            break
                        continue
                    child_result = child_agent.resume(child_prompt, stored_child_session)
+                    _log_child_skill_calls(child_result, child_agent, subtask_label)
                    resume_used = True
                else:
                    child_result = child_agent.run(child_prompt)
+                _log_child_skill_calls(child_result, child_agent, subtask_label)
                if group_id is not None and child_agent.managed_agent_id is not None:
                    self.agent_manager.register_group_child(
                        group_id,
@@ -340,7 +340,7 @@ def default_tool_registry() -> dict[str, AgentTool]:
                            {'type': 'number'},
                            {'type': 'integer'},
                            {'type': 'boolean'},
-                            {'type': 'array'},
+                            {'type': 'array', 'items': {}},
                            {'type': 'object'},
                            {'type': 'null'},
                        ]
@@ -268,10 +268,12 @@ class JupyterRuntimeSession:
                '远端 jupyter pod 未挂载 /mnt/wangsenhao 或不可写：'
                + (probe.stdout.strip() or probe.stderr.strip() or 'unknown error')
            )
+        ws_output = f'{shlex.quote(self.binding.workspace_cwd)}/output'
+        chat_output_link = f'{shlex.quote(chat_root)}/output'
        result = self.run_command(
            (
                'mkdir -p '
-                f'{shlex.quote(self.binding.workspace_cwd)}/output '
+                f'{ws_output} '
                f'{shlex.quote(self.binding.workspace_cwd)}/input '
                f'{shlex.quote(self.binding.workspace_cwd)}/scratchpad '
                f'{shlex.quote(self.binding.workspace_root)}/.runtime/uploads '
@@ -279,7 +281,12 @@ class JupyterRuntimeSession:
                f'{shlex.quote(self.account_runtime_root)}/python '
                f'{shlex.quote(chat_root)}/results '
                f'{shlex.quote(chat_root)}/sft_output '
-                f'{shlex.quote(chat_root)}/scripts'
+                f'{shlex.quote(chat_root)}/scripts && '
+                f'if [ -d {chat_output_link} ] && [ ! -L {chat_output_link} ]; then '
+                f'cp -a {chat_output_link}/* {ws_output}/ 2>/dev/null; '
+                f'rm -rf {chat_output_link}; '
+                f'fi && '
+                f'ln -sfnT {ws_output} {chat_output_link}'
            ),
            timeout_seconds=timeout_seconds,
            max_output_chars=4000,