diff --git a/.research/experiments/qwen235-fixed-pd-state-matched-diagnosis-20260719.md b/.research/experiments/qwen235-fixed-pd-state-matched-diagnosis-20260719.md new file mode 100644 index 0000000..fd45f4a --- /dev/null +++ b/.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。 diff --git a/runs/frontier-fidelity-envelope-v1/plot_qwen235_state_matched_diagnosis_mock.py b/runs/frontier-fidelity-envelope-v1/plot_qwen235_state_matched_diagnosis_mock.py new file mode 100644 index 0000000..39300ad --- /dev/null +++ b/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() diff --git a/runs/frontier-fidelity-envelope-v1/qwen235-state-matched-diagnosis-mock.png b/runs/frontier-fidelity-envelope-v1/qwen235-state-matched-diagnosis-mock.png new file mode 100644 index 0000000..bed9a02 Binary files /dev/null and b/runs/frontier-fidelity-envelope-v1/qwen235-state-matched-diagnosis-mock.png differ diff --git a/runs/frontier-fidelity-envelope-v1/run_qwen235_fixed_pd_state_replay.py b/runs/frontier-fidelity-envelope-v1/run_qwen235_fixed_pd_state_replay.py new file mode 100644 index 0000000..995b185 --- /dev/null +++ b/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() diff --git a/runs/frontier-fidelity-envelope-v1/test_fidelity_envelope.py b/runs/frontier-fidelity-envelope-v1/test_fidelity_envelope.py index 0c4c298..d045f95 100644 --- a/runs/frontier-fidelity-envelope-v1/test_fidelity_envelope.py +++ b/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: