【task002_m0_secondary_fixes】【intern_nemontron_review_cc】M0 review 次优先级 3 项修复 (#5)

* 【task002_m0_secondary_fixes】初始化

* 【task002_m0_secondary_fixes】M0 review 次优先级 3 项修复

1. #7 Hermes 多轮：convert_hermes_conversations 不再在第一条 assistant
   时 break，把所有 assistant + tool 轮收进 extra_env_info.expected_trajectory；
   新增 expected_final_content / expected_turn_count。HERMES_ROLE_MAP 支持
   tool / function / function_response / observation 四种 role。input_messages
   仍只包含首条 assistant 前的上下文（不泄漏期望答案）；expected_tool_calls
   / expected_assistant_content 保留首轮值，向后兼容。

2. #8 hermes hf_config：显式锁定 func_calling_singleturn config（与
   general_tool_calling 环境 max_turns=2 一致）；data_registry 注释解释了
   为何 hermes 没有 hf_val_split（singleturn config 没有 holdout split，
   继续走 fallback 警告路径）。

3. #15 aggregate 重复执行：拆出 score_rows（per-row 打分）与
   aggregate_scored_rows（合并指标）两个纯函数；summarize_baselines 缓存
   split 级 scored 行，aggregate 直接拼接 split 结果。对
   code_execution_python 来说 subprocess fork 数量从 2× 降到 1×。
   evaluate_policy 改成 score_rows + aggregate 的薄封装，签名不变。

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* 【task002_m0_secondary_fixes】次优先级修复回归测试

新增 4 个 case：
- hermes converter 保留 multi-turn trajectory 与 final content
- data_registry hermes hf_config 显式为 func_calling_singleturn
- summarize_baselines 不为 aggregate 重新评分（monkeypatch counter 验证）
- score_rows + aggregate_scored_rows 与 evaluate_policy 行为等价

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* 【task002_m0_secondary_fixes】移除占位 WIP.md

---------

Co-authored-by: intern_nemontron_review_cc <intern_nemontron_review_cc@intern.local>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: 3 people

src/nemotron/recipes/super3/milestones/m0_data_env/data_registry.yaml

@@ -60,8 +60,19 @@ datasets:
     environment: general_tool_calling
     domain: general_tool_calling
     hf_dataset: NousResearch/hermes-function-calling-v1
-    hf_config: null
+    # Hermes ships multiple configs (func_calling_singleturn, func_calling,
+    # glaive_func_calling, json_mode_singleturn, json_mode_agentic). Without an
+    # explicit config, `load_dataset` is ambiguous on this dataset. M0 targets
+    # the single-turn function-calling subset because the `general_tool_calling`
+    # environment registry currently runs with max_turns=2. Multi-turn config
+    # (`func_calling`) is reserved for M1 when the agent loop is wired up.
+    hf_config: func_calling_singleturn
     hf_split: train
+    # NousResearch/hermes-function-calling-v1 does not ship train/val splits per
+    # config — there is only `train`. We intentionally leave hf_val_split unset
+    # so prepare_m0_assets falls back to its sequential-slice path and records
+    # the "not a true holdout" warning in manifest.warnings. This is the same
+    # signal the README surfaces.
     hf_revision: dae3e1d28cfbcf4b915c04ea1e072030529b4bda
     source_url: https://huggingface.co/datasets/NousResearch/hermes-function-calling-v1
     license: apache-2.0

src/nemotron/recipes/super3/milestones/m0_data_env/prepare_m0_assets.py

@@ -262,36 +262,78 @@ def strip_tool_call_blocks(text: str) -> str:
     return TOOL_CALL_RE.sub("", text).strip()
 
 
+HERMES_ROLE_MAP = {
+    "system": "system",
+    "human": "user",
+    "user": "user",
+    "gpt": "assistant",
+    "assistant": "assistant",
+    "tool": "tool",
+    "function": "tool",
+    "function_response": "tool",
+    "observation": "tool",
+}
+
+
 def convert_hermes_conversations(conversations: Iterable[Mapping[str, Any]]) -> tuple[list[JsonDict], JsonDict]:
-    role_map = {
-        "system": "system",
-        "human": "user",
-        "user": "user",
-        "gpt": "assistant",
-        "assistant": "assistant",
-    }
+    """Split a Hermes function-calling conversation into model input vs expected trajectory.
+
+    `input_messages` covers the prompt context up to (but not including) the first
+    assistant turn — that is what the policy sees at inference time. Everything from
+    the first assistant turn onward (assistant tool calls, tool observations,
+    follow-up assistant turns, final answer) is captured in `expected_trajectory`
+    so that downstream verifiers can score multi-turn behavior, not just the first
+    tool emission.
+    """
     input_messages: list[JsonDict] = []
+    expected_trajectory: list[JsonDict] = []
     expected_tool_calls: list[JsonDict] = []
     expected_assistant_content = ""
+    first_assistant_seen = False
+    last_assistant_content = ""
+    last_assistant_had_tool_calls = False
 
     for turn in conversations:
         raw_role = str(turn.get("from") or turn.get("role") or "").strip()
-        role = role_map.get(raw_role)
+        role = HERMES_ROLE_MAP.get(raw_role)
         if role is None:
             continue
         content = str(turn.get("value") or turn.get("content") or "")
         if role == "assistant":
-            expected_tool_calls = parse_tool_calls(content)
-            expected_assistant_content = strip_tool_call_blocks(content)
-            break
-        input_messages.append({"role": role, "content": content})
+            tool_calls = parse_tool_calls(content)
+            stripped = strip_tool_call_blocks(content)
+            expected_trajectory.append(
+                {
+                    "role": "assistant",
+                    "content": stripped,
+                    "tool_calls": tool_calls,
+                }
+            )
+            last_assistant_content = stripped
+            last_assistant_had_tool_calls = bool(tool_calls)
+            if not first_assistant_seen:
+                expected_tool_calls = tool_calls
+                expected_assistant_content = stripped
+                first_assistant_seen = True
+        elif role == "tool":
+            expected_trajectory.append({"role": "tool", "content": content, "tool_calls": []})
+        else:  # system / user
+            if first_assistant_seen:
+                # Late system/user turn after the assistant has spoken — record in
+                # the trajectory rather than leaking it into model input.
+                expected_trajectory.append({"role": role, "content": content, "tool_calls": []})
+            else:
+                input_messages.append({"role": role, "content": content})
 
     if not any(message["role"] == "system" for message in input_messages):
         input_messages.insert(0, {"role": "system", "content": SYSTEM_PROMPTS["general_tool_calling"]})
 
     return input_messages, {
         "expected_tool_calls": expected_tool_calls,
         "expected_assistant_content": expected_assistant_content,
+        "expected_trajectory": expected_trajectory,
+        "expected_final_content": last_assistant_content if not last_assistant_had_tool_calls else "",
+        "expected_turn_count": len(expected_trajectory),
     }
 
 
@@ -329,6 +371,9 @@ def transform_hermes_function_calling(row: Mapping[str, Any], spec: Mapping[str,
         extra_env_info={
             "expected_tool_calls": expected["expected_tool_calls"],
             "expected_assistant_content": expected["expected_assistant_content"],
+            "expected_trajectory": expected["expected_trajectory"],
+            "expected_final_content": expected["expected_final_content"],
+            "expected_turn_count": expected["expected_turn_count"],
             "category": row.get("category"),
             "subcategory": row.get("subcategory"),
             "task": row.get("task"),

src/nemotron/recipes/super3/milestones/m0_data_env/run_m0_health_baseline.py

@@ -220,22 +220,20 @@ def candidates_for_policy(record: Mapping[str, Any], policy: str) -> list[Any]:
     raise ValueError(f"unknown baseline policy: {policy}")
 
 
-def evaluate_policy(
+def score_rows(
     rows: Sequence[Mapping[str, Any]],
     *,
     policy: str,
     best_k: int,
     run_code: bool,
-) -> JsonDict:
-    pass_at_1 = 0
-    best_at_k = 0
-    total_score_at_1 = 0.0
-    total_best_score_at_k = 0.0
-    errors: list[JsonDict] = []
-    skipped_rows = 0
-    scored_rows = 0
-    threshold = 1.0
-
+) -> list[JsonDict]:
+    """Score each row once and return raw per-row results.
+
+    Pulled out of `evaluate_policy` so that aggregate metrics can be derived from
+    the same per-row scores as the split metrics, instead of re-invoking the
+    verifier (and, for python_unit_tests, re-spawning subprocesses).
+    """
+    out: list[JsonDict] = []
     for index, record in enumerate(rows):
         candidates = candidates_for_policy(record, policy)
         scores: list[float] = []
@@ -246,8 +244,36 @@ def evaluate_policy(
             if score is None:
                 continue
             scores.append(score)
+        out.append(
+            {
+                "row_index": index,
+                "scores": scores,
+                "diagnostics": diagnostics,
+                "metadata": record.get("metadata", {}),
+                "verifier": record.get("reward_config", {}).get("verifier"),
+            }
+        )
+    return out
+
+
+def aggregate_scored_rows(
+    scored: Sequence[Mapping[str, Any]],
+    *,
+    policy: str,
+    best_k: int,
+) -> JsonDict:
+    pass_at_1 = 0
+    best_at_k = 0
+    total_score_at_1 = 0.0
+    total_best_score_at_k = 0.0
+    errors: list[JsonDict] = []
+    skipped_rows = 0
+    scored_rows = 0
+    threshold = 1.0
+
+    for entry in scored:
+        scores = entry["scores"]
         if not scores:
-            # Every candidate for this row was skipped (e.g. code execution disabled).
             skipped_rows += 1
             continue
         scored_rows += 1
@@ -258,19 +284,19 @@ def evaluate_policy(
         total_score_at_1 += first_score
         total_best_score_at_k += best_score
         if best_score < threshold:
-            metadata = record.get("metadata", {})
+            metadata = entry["metadata"]
             errors.append(
                 {
-                    "row_index": index,
+                    "row_index": entry["row_index"],
                     "source_dataset": metadata.get("source_dataset"),
                     "source_id": metadata.get("source_id"),
                     "source_row_index": metadata.get("source_row_index"),
-                    "verifier": record.get("reward_config", {}).get("verifier"),
-                    "diagnostics": diagnostics,
+                    "verifier": entry["verifier"],
+                    "diagnostics": entry["diagnostics"],
                 }
             )
 
-    total = len(rows)
+    total = len(scored)
     denominator = scored_rows or 1
     return {
         "policy": policy,
@@ -286,6 +312,20 @@ def evaluate_policy(
     }
 
 
+def evaluate_policy(
+    rows: Sequence[Mapping[str, Any]],
+    *,
+    policy: str,
+    best_k: int,
+    run_code: bool,
+) -> JsonDict:
+    """Score and aggregate in one pass. Kept for callers that don't need
+    the per-row breakdown (existing tests, direct CLI users).
+    """
+    scored = score_rows(rows, policy=policy, best_k=best_k, run_code=run_code)
+    return aggregate_scored_rows(scored, policy=policy, best_k=best_k)
+
+
 def load_environment_specs(path: Path) -> dict[str, JsonDict]:
     registry = load_yaml(path)
     specs = {}
@@ -366,16 +406,26 @@ def summarize_baselines(
 ) -> JsonDict:
     summary: JsonDict = {"best_k": best_k, "policies": list(policies), "environments": {}}
     for env_id, splits in rows_by_env.items():
-        env_summary = {"splits": {}, "aggregate": {}}
-        aggregate_rows = []
+        env_summary: JsonDict = {"splits": {}, "aggregate": {}}
+        # Cache per-(split, policy) scored rows so the aggregate metric reuses
+        # the same scores instead of re-running the verifier (which, for
+        # python_unit_tests, halves the number of subprocess forks).
+        scored_cache: dict[tuple[str, str], list[JsonDict]] = {}
         for split, rows in sorted(splits.items()):
-            aggregate_rows.extend(rows)
-            env_summary["splits"][split] = {
-                policy: evaluate_policy(rows, policy=policy, best_k=best_k, run_code=run_code) for policy in policies
-            }
-        env_summary["aggregate"] = {
-            policy: evaluate_policy(aggregate_rows, policy=policy, best_k=best_k, run_code=run_code) for policy in policies
-        }
+            env_summary["splits"][split] = {}
+            for policy in policies:
+                scored = score_rows(rows, policy=policy, best_k=best_k, run_code=run_code)
+                scored_cache[(split, policy)] = scored
+                env_summary["splits"][split][policy] = aggregate_scored_rows(
+                    scored, policy=policy, best_k=best_k
+                )
+        for policy in policies:
+            combined: list[JsonDict] = []
+            for split in sorted(splits):
+                combined.extend(scored_cache[(split, policy)])
+            env_summary["aggregate"][policy] = aggregate_scored_rows(
+                combined, policy=policy, best_k=best_k
+            )
         summary["environments"][env_id] = env_summary
     return summary
 

tests/recipes/super3/test_m0_data_env.py

@@ -158,3 +158,46 @@ def test_data_registry_specifies_hf_val_split_for_holdout_capable_datasets() ->
     assert by_id["m0_reasoning_gsm8k"]["hf_val_split"] == "test"
     assert by_id["m0_coding_mbpp"]["hf_val_split"] == "validation"
     assert by_id["m0_search_hotpotqa"]["hf_val_split"] == "validation"
+
+
+def test_data_registry_pins_explicit_hf_config_for_hermes() -> None:
+    """Regression for review finding #8: hermes hf_config: null is ambiguous on multi-config datasets."""
+    registry = load_yaml(DATA_REGISTRY_PATH)
+    hermes = next(dataset for dataset in registry["datasets"] if dataset["id"] == "m0_tool_calling_hermes")
+
+    assert hermes["hf_config"] == "func_calling_singleturn"
+    assert hermes["hf_config"] is not None
+
+
+def test_hermes_converter_captures_multi_turn_trajectory() -> None:
+    """Regression for review finding #7: convert_hermes_conversations used to drop tool turns and the final answer."""
+    first_tool_call = '<tool_call>{"name":"lookup","arguments":{"query":"weather"}}</tool_call>'
+    final_answer = "The weather in Paris is sunny."
+    conversations = [
+        {"from": "system", "value": "Use tools when useful."},
+        {"from": "human", "value": "What is the weather in Paris?"},
+        {"from": "gpt", "value": first_tool_call},
+        {"from": "tool", "value": '{"city":"Paris","weather":"sunny"}'},
+        {"from": "gpt", "value": final_answer},
+    ]
+
+    input_messages, expected = convert_hermes_conversations(conversations)
+
+    # Input still stops at the first assistant turn — that's what the policy sees.
+    assert [message["role"] for message in input_messages] == ["system", "user"]
+
+    # Trajectory keeps every assistant + tool turn after the first user message.
+    trajectory = expected["expected_trajectory"]
+    assert [turn["role"] for turn in trajectory] == ["assistant", "tool", "assistant"]
+    assert trajectory[0]["tool_calls"][0]["function"]["name"] == "lookup"
+    assert "Paris" in trajectory[1]["content"]
+    assert trajectory[2]["content"] == final_answer
+    assert trajectory[2]["tool_calls"] == []
+
+    # First-turn fields stay backward-compatible.
+    assert expected["expected_tool_calls"][0]["function"]["name"] == "lookup"
+    assert expected["expected_assistant_content"] == ""
+
+    # Final content surfaces the last assistant turn that wasn't a tool call.
+    assert expected["expected_final_content"] == final_answer
+    assert expected["expected_turn_count"] == 3

tests/recipes/super3/test_m0_health_baseline.py

@@ -2,8 +2,10 @@
 
 import pytest
 
+from nemotron.recipes.super3.milestones.m0_data_env import run_m0_health_baseline
 from nemotron.recipes.super3.milestones.m0_data_env.run_m0_health_baseline import (
     MISSING,
+    aggregate_scored_rows,
     build_report,
     evaluate_policy,
     get_path,
@@ -12,8 +14,10 @@
     overall_status,
     run_python_unit_tests,
     score_record,
+    score_rows,
     score_text,
     score_tool_call,
+    summarize_baselines,
     summarize_health,
 )
 
@@ -205,3 +209,56 @@ def test_build_report_raises_for_missing_input_dir(tmp_path: Path) -> None:
 
     with pytest.raises(FileNotFoundError):
         build_report(args)
+
+
+def test_summarize_baselines_does_not_rescore_for_aggregate(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Regression for review finding #15: aggregate used to re-run the verifier per row."""
+    rows_by_env = {
+        "math_reasoning_numeric": {
+            "train": [
+                {"expected_answer": "1", "reward_config": {"verifier": "normalized_numeric_exact_match"}},
+                {"expected_answer": "2", "reward_config": {"verifier": "normalized_numeric_exact_match"}},
+            ],
+            "val": [
+                {"expected_answer": "3", "reward_config": {"verifier": "normalized_numeric_exact_match"}},
+            ],
+        }
+    }
+    calls = {"count": 0}
+    original_score_record = run_m0_health_baseline.score_record
+
+    def counting_score_record(*args, **kwargs):
+        calls["count"] += 1
+        return original_score_record(*args, **kwargs)
+
+    monkeypatch.setattr(run_m0_health_baseline, "score_record", counting_score_record)
+
+    summary = summarize_baselines(
+        rows_by_env,
+        policies=["oracle"],
+        best_k=1,
+        run_code=False,
+    )
+
+    # oracle policy has 1 candidate; total rows = 3; aggregate must reuse split scores.
+    assert calls["count"] == 3
+
+    env = summary["environments"]["math_reasoning_numeric"]
+    assert env["splits"]["train"]["oracle"]["scored_rows"] == 2
+    assert env["splits"]["val"]["oracle"]["scored_rows"] == 1
+    assert env["aggregate"]["oracle"]["scored_rows"] == 3
+    assert env["aggregate"]["oracle"]["pass_at_1"] == 1.0
+
+
+def test_score_rows_and_aggregate_match_evaluate_policy() -> None:
+    """evaluate_policy keeps working as a thin wrapper after the refactor."""
+    rows = [
+        {"expected_answer": "42", "reward_config": {"verifier": "normalized_numeric_exact_match"}},
+        {"expected_answer": "7", "reward_config": {"verifier": "normalized_numeric_exact_match"}},
+    ]
+
+    scored = score_rows(rows, policy="oracle", best_k=2, run_code=False)
+    aggregated = aggregate_scored_rows(scored, policy="oracle", best_k=2)
+    direct = evaluate_policy(rows, policy="oracle", best_k=2, run_code=False)
+
+    assert aggregated == direct