Add Qwen235 state replay diagnosis

.research/experiments/qwen235-fixed-pd-state-matched-diagnosis-20260719.md

@@ -0,0 +1,45 @@
+# 实验：Qwen235 Fixed-PD state-matched decode diagnosis
+
+> **状态：** 已批准，执行中
+>
+> 本 card 记录 Qwen235 Fixed-PD 在真实 collective profile 后仍保留 30%+ selection regret 的下一层判因实验。
+
+## Claim 与决策
+
+- **Parent claim：** Qwen235 Fixed-PD 的错误排序来自 action-conditioned decode residual，而不是缺失的 TP8 all-reduce profile。
+- **目的：** 区分 simulator 错在 state distribution，还是相同 state 下的 conditional execution-time composition。
+- **Competing hypotheses：** H1：Frontier 生成的 decode batch/context/graph state 与真机不同；H2：state 对齐后 Frontier 仍把 TP8/EP8 预测得更快，误差位于 MoE/EP、graph 或 attention 的 conditional stage model。
+- **事前预测：** 真机 config contrast 为 `TP8-TP4=+6.95 ms/token`，A1 simulator 为 `-20.07 ms/token`。若 H1 成立，用真实 state 重加权后 contrast 应翻正；若 H2 成立，matched-state contrast 仍为负。
+- **判定规则：** 先比较 frozen-real coarse state 与 full-ledger Frontier state。state 明显不匹配则补 iteration telemetry；state 支持重叠且 matched-state predictor 仍反序，才进入 stage breakdown。任何 stage 只有在 measured substitution 能使 winner 翻转时才称为 decision-bearing root cause。
+
+## Setup
+
+- **自变量：** state source（frozen real / Frontier）；后续 matched-state replay 中固定 decode batch、context-length、graph bucket 与 routing load。
+- **控制变量：** Qwen235 FP8、vLLM 0.20.0、H20、Fixed-PD 4096→256、0.2 req/s/GPU、MBT8192、MNS64、TP4/EP1 与 TP8/EP8、Frontier commit、r2 operator profiles、A1 measured collective CSV 全部冻结。
+- **选择 MNS64：** A1 中 MNS64/128 的 TTFT/TPOT/E2E 完全相同；先去掉不提供判别力的重复维度。
+- **第一阶段：** CPU-only 重放原 A1 commands，只打开 `frontier_stage_batch_ledger` 与 individual batch metrics；验证 request metrics 与原 A1 bitwise/score 等价。真机先复用 3 次 frozen server logs 的 10 秒 Running/Waiting/KV samples，明确标为 coarse proxy，不冒充 per-iteration batch。
+- **第二阶段触发条件：** coarse proxy 不足以判断或 state mismatch 显著时，短窗口重跑真机并采集 per-iteration `decode_batch_size/context_length_hist/cudagraph bucket`；否则进入 matched-state whole-decode-step。
+- **Metrics：** decode batch/token distribution、prefill fraction、scheduler steps/s、graph bucket/padding、queue/KV proxy；configuration contrast `TP8-TP4`；stage measured-substitution 后的 winner。
+
+## 预期产物与 review
+
+- **预期数据：** 两个 Frontier full-ledger replays；三次真机日志的 coarse state summary；state overlap/reweighting verdict；必要时的 short-window iteration telemetry。
+- **Figure prototype：** `../../runs/frontier-fidelity-envelope-v1/qwen235-state-matched-diagnosis-mock.png`。左图对比 real/sim state；右图展示 H1 与 H2 下 matched-state contrast 的可区分方向。全部数值标为 schematic/mock。
+- **人工 review：** 已批准（用户在分析方案后要求“推进”）。
+- **Review 意见：** 先做最便宜的 state audit，不直接启动完整 Nsight sweep；每一步只在能改变下一决策时升级证据成本。
+
+## 复现信息
+
+- **Code：** AITuner `feature/sim`；运行 commit 待冻结。Frontier `6e8e0d845bceff11b0b62cb29df3a1a93411fdd4`。
+- **Environment：** dash0；Frontier replay CPU-only；后续真机才使用 4/8×H20。
+- **输入：** `/home/admin/cpfs/wjh/aituner/qwen235-collective-profile-ablation-20260719-r1/sim/fixed-pd` 与 frozen real campaign `/home/admin/cpfs/wjh/aituner/qwen235-v020-fourcase-20260719-r1/real/fixed-pd`。
+- **产物路径：** `/home/admin/cpfs/wjh/aituner/qwen235-fixed-pd-state-diagnosis-20260719-r1`。
+- **已知 deviation：** frozen real logs 的 Running 指标是 10 秒采样的 active-request proxy，不是 scheduler iteration ledger；不能单独支持 matched-state causal claim。
+
+## 结果
+
+- **观察事实：** 待运行。
+- **异常：** 待运行。
+- **含义：** 待运行。
+- **Claim update：** unchanged
+- **下一步：** 运行两项 CPU state replay 与 coarse real-state analysis。

runs/frontier-fidelity-envelope-v1/plot_qwen235_state_matched_diagnosis_mock.py

@@ -0,0 +1,43 @@
+#!/usr/bin/env python3
+"""Render the reviewed schematic for Qwen235 state-matched diagnosis."""
+
+from pathlib import Path
+
+import matplotlib.pyplot as plt
+import numpy as np
+
+
+OUTPUT = Path(__file__).with_name("qwen235-state-matched-diagnosis-mock.png")
+
+
+def main() -> None:
+    figure, axes = plt.subplots(1, 2, figsize=(11.2, 4.4))
+
+    labels = ["TP4/EP1", "TP8/EP8"]
+    x = np.arange(2)
+    width = 0.34
+    axes[0].bar(x - width / 2, [5, 12], width, label="Real proxy (mock)")
+    axes[0].bar(x + width / 2, [8, 8], width, label="Frontier ledger (mock)")
+    axes[0].set_xticks(x, labels)
+    axes[0].set_ylabel("Active decode requests / iteration")
+    axes[0].set_title("(a) Does Frontier reproduce real state?")
+    axes[0].legend(frameon=False)
+
+    scenarios = ["Observed\nA1", "Matched state\nif H1", "Matched state\nif H2", "Real"]
+    contrasts = [-20.07, 4.0, -18.0, 6.95]
+    colors = ["#d95f02", "#1b9e77", "#d95f02", "#1b9e77"]
+    axes[1].bar(np.arange(4), contrasts, color=colors)
+    axes[1].axhline(0, color="black", linewidth=0.9)
+    axes[1].set_xticks(np.arange(4), scenarios)
+    axes[1].set_ylabel("TP8 - TP4 TPOT (ms/token)")
+    axes[1].set_title("(b) State mismatch or execution model?")
+    axes[1].text(0.02, 0.97, "SCHEMATIC / MOCK DATA", transform=axes[1].transAxes,
+                 va="top", fontsize=9, color="#8c2d04")
+
+    figure.suptitle("Qwen235 Fixed-PD state-matched diagnosis", fontsize=13)
+    figure.tight_layout()
+    figure.savefig(OUTPUT, dpi=180, bbox_inches="tight")
+
+
+if __name__ == "__main__":
+    main()

runs/frontier-fidelity-envelope-v1/run_qwen235_fixed_pd_state_replay.py

@@ -0,0 +1,336 @@
+#!/usr/bin/env python3
+"""Replay the two Qwen235 Fixed-PD A1 cells with Frontier state outputs.
+
+The frozen command is the experimental input.  This runner changes only the
+metrics output/run id and the two existing state-observation flags, then checks
+that request-level scorer inputs are byte-identical to the frozen A1 run.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import importlib.util
+import json
+import os
+import subprocess
+import sys
+import time
+from pathlib import Path
+from typing import Any
+
+
+HERE = Path(__file__).resolve().parent
+REPO_ROOT = HERE.parents[1]
+TELEMETRY_ROOT = REPO_ROOT / "runs/telemetry-residual"
+sys.path.insert(0, str(TELEMETRY_ROOT))
+
+from common_state import summarize_frontier  # noqa: E402
+from run_frontier_state import enable_state_outputs, find_state_metrics  # noqa: E402
+
+
+CONFIGS = ("tp4_ep1_mns64", "tp8_ep8_mns64")
+ALLOWED_FLAG_CHANGES = {
+    "--metrics_config_output_dir",
+    "--metrics_config_run_id",
+    "--no-metrics_config_store_frontier_stage_batch_ledger",
+    "--metrics_config_store_frontier_stage_batch_ledger",
+    "--no-metrics_config_keep_individual_batch_metrics",
+    "--metrics_config_keep_individual_batch_metrics",
+}
+
+
+def load_module(name: str, path: Path):
+    spec = importlib.util.spec_from_file_location(name, path)
+    if spec is None or spec.loader is None:
+        raise ImportError(path)
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+def sha256_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as source:
+        for chunk in iter(lambda: source.read(1 << 20), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def atomic_json(path: Path, payload: Any) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(path.suffix + ".tmp")
+    temporary.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
+    os.replace(temporary, path)
+
+
+def replace_option(command: list[str], option: str, value: str) -> list[str]:
+    result = list(command)
+    if result.count(option) != 1:
+        raise ValueError(f"expected exactly one {option}, found {result.count(option)}")
+    index = result.index(option)
+    if index + 1 >= len(result) or result[index + 1].startswith("--"):
+        raise ValueError(f"{option} has no value")
+    result[index + 1] = value
+    return result
+
+
+def option_value(command: list[str], option: str) -> str:
+    if command.count(option) != 1:
+        raise ValueError(f"expected exactly one {option}, found {command.count(option)}")
+    return command[command.index(option) + 1]
+
+
+def _semantic_options(command: list[str]) -> dict[str, tuple[str, ...]]:
+    """Parse CLI tokens sufficiently to audit this controlled command edit."""
+
+    result: dict[str, tuple[str, ...]] = {}
+    index = 0
+    while index < len(command):
+        token = command[index]
+        if not token.startswith("--"):
+            index += 1
+            continue
+        values: list[str] = []
+        index += 1
+        while index < len(command) and not command[index].startswith("--"):
+            values.append(command[index])
+            index += 1
+        if token in result:
+            raise ValueError(f"duplicate option in frozen command: {token}")
+        result[token] = tuple(values)
+    return result
+
+
+def transform_command(
+    command: list[str], *, metrics_root: Path, run_id: str
+) -> list[str]:
+    result = replace_option(
+        command, "--metrics_config_output_dir", str(metrics_root.resolve())
+    )
+    result = replace_option(result, "--metrics_config_run_id", run_id)
+    result = enable_state_outputs(result)
+
+    before = _semantic_options(command)
+    after = _semantic_options(result)
+    changed = {
+        option
+        for option in set(before) | set(after)
+        if before.get(option) != after.get(option)
+    }
+    if not changed <= ALLOWED_FLAG_CHANGES:
+        raise ValueError(f"unapproved command changes: {sorted(changed - ALLOWED_FLAG_CHANGES)}")
+    required = {
+        "--metrics_config_output_dir",
+        "--metrics_config_run_id",
+        "--no-metrics_config_store_frontier_stage_batch_ledger",
+        "--metrics_config_store_frontier_stage_batch_ledger",
+        "--metrics_config_keep_individual_batch_metrics",
+    }
+    if not required <= changed:
+        raise ValueError(f"required controlled changes missing: {sorted(required - changed)}")
+    return result
+
+
+def environment(frontier_source: Path, python_deps: Path) -> dict[str, str]:
+    result = os.environ.copy()
+    result.update(
+        {
+            "PYTHONPATH": ":".join([str(python_deps), str(frontier_source)]),
+            "CUDA_VISIBLE_DEVICES": "",
+            "NVIDIA_VISIBLE_DEVICES": "void",
+            "WANDB_DISABLED": "true",
+            "VIDUR_DISABLE_WANDB": "1",
+            "FRONTIER_LOG_LEVEL": "WARNING",
+            "PYTHONDONTWRITEBYTECODE": "1",
+        }
+    )
+    return result
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--base-sim-root", type=Path, required=True)
+    parser.add_argument("--output-root", type=Path, required=True)
+    parser.add_argument("--frontier-source", type=Path, required=True)
+    parser.add_argument("--python-deps", type=Path, required=True)
+    parser.add_argument("--config", action="append", choices=CONFIGS)
+    parser.add_argument("--timeout-seconds", type=float, default=1200)
+    parser.add_argument("--resume", action="store_true")
+    return parser.parse_args()
+
+
+def run_cell(
+    *,
+    config: str,
+    base_sim_root: Path,
+    output_root: Path,
+    frontier_source: Path,
+    python_deps: Path,
+    timeout_seconds: float,
+    resume: bool,
+    q30: Any,
+) -> dict[str, Any]:
+    base_run = base_sim_root / "runs" / config / "eval"
+    base_command_path = base_run / "command.json"
+    base_result_path = base_run / "result.json"
+    if not base_command_path.is_file() or not base_result_path.is_file():
+        raise FileNotFoundError(f"frozen A1 inputs missing under {base_run}")
+    base_command = json.loads(base_command_path.read_text())
+    base_result = json.loads(base_result_path.read_text())
+    if base_result.get("status") != "completed":
+        raise ValueError(f"frozen A1 result is not completed: {base_result_path}")
+
+    run_root = output_root / config
+    result_path = run_root / "result.json"
+    if resume and result_path.is_file():
+        previous = json.loads(result_path.read_text())
+        if (
+            previous.get("status") == "PASS"
+            and previous.get("inputs", {}).get("base_command_sha256")
+            == sha256_file(base_command_path)
+            and previous.get("inputs", {}).get("base_result_sha256")
+            == sha256_file(base_result_path)
+        ):
+            return previous
+    if run_root.exists() and any(run_root.iterdir()):
+        raise FileExistsError(f"refusing non-empty output: {run_root}")
+    run_root.mkdir(parents=True, exist_ok=True)
+
+    command = transform_command(
+        base_command,
+        metrics_root=run_root / "frontier_metrics",
+        run_id=f"qwen235_fixed_pd_state_{config}",
+    )
+    atomic_json(run_root / "command.json", command)
+    manifest = {
+        "schema": "qwen235-fixed-pd-state-replay-v1",
+        "config": config,
+        "inputs": {
+            "base_run": str(base_run.resolve()),
+            "base_command_sha256": sha256_file(base_command_path),
+            "base_result_sha256": sha256_file(base_result_path),
+            "base_request_metrics_sha256": base_result["request_metrics_sha256"],
+        },
+        "controlled_changes": [
+            "metrics output directory",
+            "metrics run id",
+            "full Frontier stage/batch ledger enabled",
+            "individual batch metrics enabled",
+        ],
+        "command_sha256": sha256_file(run_root / "command.json"),
+        "frontier": {
+            "source": str(frontier_source),
+            "git_head": subprocess.check_output(
+                ["git", "-C", str(frontier_source), "rev-parse", "HEAD"], text=True
+            ).strip(),
+        },
+        "environment": {
+            "PYTHONPATH": ":".join([str(python_deps), str(frontier_source)]),
+            "CUDA_VISIBLE_DEVICES": "",
+            "NVIDIA_VISIBLE_DEVICES": "void",
+            "FRONTIER_LOG_LEVEL": "WARNING",
+        },
+    }
+    atomic_json(run_root / "run_manifest.json", manifest)
+
+    started = time.monotonic()
+    with (run_root / "stdout.log").open("w") as stdout, (
+        run_root / "stderr.log"
+    ).open("w") as stderr:
+        try:
+            completed = subprocess.run(
+                command,
+                cwd=frontier_source,
+                env=environment(frontier_source, python_deps),
+                stdout=stdout,
+                stderr=stderr,
+                timeout=timeout_seconds,
+                check=False,
+            )
+            returncode = int(completed.returncode)
+        except subprocess.TimeoutExpired:
+            returncode = 124
+    elapsed_seconds = time.monotonic() - started
+    if returncode != 0:
+        failure = {
+            "status": "STOP",
+            "config": config,
+            "returncode": returncode,
+            "elapsed_seconds": elapsed_seconds,
+        }
+        atomic_json(run_root / "failure.json", failure)
+        raise RuntimeError(f"state replay failed: {failure}")
+
+    fallback_evidence = q30.collective_fallback_evidence(run_root)
+    if fallback_evidence:
+        raise RuntimeError(f"collective-profile fallback detected: {fallback_evidence}")
+    paths = find_state_metrics(run_root)
+    request_metrics_sha256 = sha256_file(paths["requests"])
+    scorer_equivalent = request_metrics_sha256 == base_result["request_metrics_sha256"]
+    if not scorer_equivalent:
+        raise RuntimeError(
+            f"observation changed scorer input for {config}: "
+            f"{request_metrics_sha256} != {base_result['request_metrics_sha256']}"
+        )
+    state = summarize_frontier(
+        system_metrics_path=paths["system"],
+        request_metrics_path=paths["requests"],
+        batch_metrics_path=paths["batches"],
+        ledger_path=paths["ledger"],
+    )
+    atomic_json(run_root / "common-state.json", state)
+    result = {
+        "schema": "qwen235-fixed-pd-state-replay-result-v1",
+        "status": "PASS",
+        "config": config,
+        "elapsed_seconds": elapsed_seconds,
+        "returncode": returncode,
+        "inputs": manifest["inputs"],
+        "scorer_equivalence": {
+            "request_metrics_byte_identical": scorer_equivalent,
+            "request_metrics_sha256": request_metrics_sha256,
+            "base_metrics": base_result["metrics"],
+        },
+        "state_artifacts": {
+            name: {"path": str(path.resolve()), "sha256": sha256_file(path)}
+            for name, path in paths.items()
+        },
+        "common_state": state,
+        "collective_fallback_evidence": fallback_evidence,
+    }
+    atomic_json(result_path, result)
+    return result
+
+
+def main() -> None:
+    args = parse_args()
+    for name in ("base_sim_root", "output_root", "frontier_source", "python_deps"):
+        setattr(args, name, getattr(args, name).resolve())
+    selected = tuple(args.config or CONFIGS)
+    q30 = load_module("q235_state_q30", HERE / "run_frontier_qwen30_exact_trace_surface.py")
+    results = []
+    for config in selected:
+        result = run_cell(
+            config=config,
+            base_sim_root=args.base_sim_root,
+            output_root=args.output_root,
+            frontier_source=args.frontier_source,
+            python_deps=args.python_deps,
+            timeout_seconds=args.timeout_seconds,
+            resume=args.resume,
+            q30=q30,
+        )
+        results.append(result)
+        print(json.dumps({"config": config, "status": result["status"]}), flush=True)
+    aggregate = {
+        "schema": "qwen235-fixed-pd-state-replay-aggregate-v1",
+        "status": "PASS" if all(result["status"] == "PASS" for result in results) else "STOP",
+        "results": results,
+    }
+    atomic_json(args.output_root / "state_replay.json", aggregate)
+
+
+if __name__ == "__main__":
+    main()

runs/frontier-fidelity-envelope-v1/test_fidelity_envelope.py

@@ -135,6 +135,58 @@ class FidelityEnvelopeTest(unittest.TestCase):
             self.assertEqual(len(evidence), 1)
             self.assertEqual(evidence[0]["log"], "stdout.log")
 
+    def test_qwen235_state_replay_changes_observation_flags_only(self) -> None:
+        module = load("run_qwen235_fixed_pd_state_replay.py")
+        base = [
+            "/usr/bin/python3",
+            "-m",
+            "frontier.main",
+            "--metrics_config_output_dir",
+            "/frozen/metrics",
+            "--metrics_config_run_id",
+            "frozen",
+            "--no-metrics_config_store_frontier_stage_batch_ledger",
+            "--attn_tensor_parallel_size",
+            "8",
+            "--moe_expert_parallel_size",
+            "8",
+            "--communication_collective_profile_path",
+            "/profiles/real-allreduce.csv",
+        ]
+        transformed = module.transform_command(
+            base, metrics_root=Path("/new/metrics"), run_id="state-run"
+        )
+        self.assertEqual(
+            module.option_value(transformed, "--attn_tensor_parallel_size"), "8"
+        )
+        self.assertEqual(
+            module.option_value(transformed, "--communication_collective_profile_path"),
+            "/profiles/real-allreduce.csv",
+        )
+        self.assertIn(
+            "--metrics_config_store_frontier_stage_batch_ledger", transformed
+        )
+        self.assertIn("--metrics_config_keep_individual_batch_metrics", transformed)
+        self.assertNotIn(
+            "--no-metrics_config_store_frontier_stage_batch_ledger", transformed
+        )
+
+    def test_qwen235_state_replay_rejects_implicit_ledger_base(self) -> None:
+        module = load("run_qwen235_fixed_pd_state_replay.py")
+        base = [
+            "/usr/bin/python3",
+            "-m",
+            "frontier.main",
+            "--metrics_config_output_dir",
+            "/frozen/metrics",
+            "--metrics_config_run_id",
+            "frozen",
+        ]
+        with self.assertRaisesRegex(ValueError, "explicitly disable one full Frontier ledger"):
+            module.transform_command(
+                base, metrics_root=Path("/new/metrics"), run_id="state-run"
+            )
+
     def test_materialize_qwen235_allreduce_requires_serving_contract(self) -> None:
         module = load("materialize_qwen235_v020_allreduce.py")
         with tempfile.TemporaryDirectory() as temporary: