diff --git a/runs/frontier-code-trace-v0/README.md b/runs/frontier-code-trace-v0/README.md new file mode 100644 index 0000000..4abc76b --- /dev/null +++ b/runs/frontier-code-trace-v0/README.md @@ -0,0 +1,118 @@ +# Frontier code-trace campaign handoff + +本目录已经准备好无 GPU 的 data preflight、512→16 参数化映射、prefill-only 转换和 `max_model_len` 显式适配。当前没有启动或探测 `dash1`–`dash4`。 + +完整设计与 gate 见 [`experiment-card.md`](experiment-card.md)。 + +## 当前已知阻塞 + +本机 `/home/gahow/ali-trace/trace-glm5.1-formatted/` 不存在。仓库历史记录的远端路径是： + +```text +/home/admin/cpfs/wjh/ali-trace/trace-glm5.1-formatted/ +``` + +有机器后先确认用户给出的 `~/ali-trace/...` 是否解析到同一目录，再运行以下命令。 + +## 1. 审计所有 1h+ code source + +在持有 trace 的机器、repo 根目录执行： + +```bash +python3 runs/frontier-code-trace-v0/audit_code_trace.py \ + --trace-root ~/ali-trace/trace-glm5.1-formatted \ + --output runs/frontier-code-trace-v0/inputs/code-audit.json +``` + +如果目录里混有非 request JSONL，先只读列举文件，再用多个 `--source` 显式指定。审计输出必须满足： + +```text +data_gate = PASS +selected.hash_contract.exact_source_block_size != null +max_model_len_recommendation != null +selected.selected_window_stats.max_model_len_coverage[推荐值].coverage = 1.0 +``` + +旧记录预计 source block size 为 512、推荐 max model len 为 131072，但禁止把这两个值写死为实验事实。 + +## 2. 物化稳定窗口 + +```bash +python3 runs/frontier-code-trace-v0/prepare_code_window.py \ + --audit runs/frontier-code-trace-v0/inputs/code-audit.json \ + --output-root runs/frontier-code-trace-v0/inputs/code-window +``` + +输出是 60–75min `code-raw-window.jsonl` 和 manifest。source 文件不修改。 + +## 3. 生成 P+D paired trace + +若没有 prompt sidecar，先生成 shape/prefix-faithful synthetic prompts： + +```bash +python3 runs/frontier-s3-real-v0/remap_hash_blocks.py \ + --input runs/frontier-code-trace-v0/inputs/code-window/code-raw-window.jsonl \ + --output-root runs/frontier-code-trace-v0/inputs/code-pd-rho-max \ + --source-block-size 512 \ + --workload-mode prefill_decode \ + --rho 1.0 \ + --max-total-tokens 131072 \ + --validate-parents +``` + +命令中的 `512` 和 `131072` 必须替换为 audit manifest 值。若存在对齐 prompt sidecar，加 `--prompt` 与 `--tokenizer`，并要求 synthetic fallback 为 0。 + +正式 rho 不能直接用 1.0；先从最大 remap cache 按 session-coherent `sampling_u` 过滤，分别标定 low/mid/near-knee。 + +## 4. 生成 prefill-only paired trace + +对 chat/code 使用同一个转换接口： + +```bash +python3 runs/frontier-s3-real-v0/remap_hash_blocks.py \ + --input INPUT_WINDOW.jsonl \ + --output-root OUTPUT_ROOT \ + --source-block-size SOURCE_BLOCK_SIZE \ + --workload-mode prefill_only \ + --rho RHO \ + --max-total-tokens MAX_MODEL_LEN \ + --validate-parents +``` + +该模式会同时把 Frontier `num_decode_tokens`、real request `min/max_tokens` 和 remapped row 的 `output_length` 固定为 1。 + +## 5. max-model-len 真机 gate + +现有 real runner 新增了三个显式环境变量，chat 默认行为不变： + +```bash +MAX_MODEL_LEN=131072 \ +TRACE_INPUT_ROOT=/absolute/path/to/materialized/code-cell \ +ALLOW_SYNTHETIC_PROMPTS=true \ +OUTPUT_ROOT=/absolute/path/to/new/output \ +bash runs/frontier-s3-real-v0/run_full_real.sh RHO_LABEL tp4_mns16 1 PORT +``` + +- `MAX_MODEL_LEN` 必须等于 manifest 推荐值； +- `TRACE_INPUT_ROOT` 内必须有 `real_requests.jsonl` 和 `manifest.json`； +- synthetic prompt 默认拒绝，只有在 experiment card 明确降级 claim 后才设为 `true`； +- runner 会在启动前扫描 paired requests，若任何 `ISL+OSL` 超 cap 立即失败。 + +正式 full job 前，先按 experiment card 的 G4 补 32k–128k attention profile，再做 TP4→TP2 的 p50/p99/max 单请求与 5min canary。 + +## 6. decode-only + +当前 materializer 故意不提供 `decode_only` 选项。严格 decode-only 需要 initial-KV state，而不是把 prompt 改短。只有 real `DecodeBenchConnector`（或等价能力）与 Frontier initial-KV contract 都通过 G7 后，才创建 decode-only jobs。 + +## 本地验证 + +```bash +python3 -m unittest -v \ + runs/frontier-code-trace-v0/test_code_trace_preflight.py \ + runs/frontier-s3-real-v0/test_remap_hash_blocks.py \ + runs/frontier-s3-real-v0/test_select_chat_window.py +python3 -m py_compile \ + runs/frontier-code-trace-v0/*.py \ + runs/frontier-s3-real-v0/*.py +bash -n runs/frontier-s3-real-v0/run_full_real.sh +``` diff --git a/runs/frontier-code-trace-v0/audit_code_trace.py b/runs/frontier-code-trace-v0/audit_code_trace.py new file mode 100644 index 0000000..250b47f --- /dev/null +++ b/runs/frontier-code-trace-v0/audit_code_trace.py @@ -0,0 +1,361 @@ +#!/usr/bin/env python3 +"""Audit long code traces before choosing a replay window and max model length.""" + +from __future__ import annotations + +import argparse +import json +import math +import statistics +from collections import Counter +from pathlib import Path +from typing import Any, Iterable, Sequence + + +BLOCK_SIZE_CANDIDATES = (16, 32, 64, 128, 256, 512, 1024) +MAX_MODEL_LEN_CANDIDATES = (40960, 65536, 98304, 131072, 262144) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--trace-root", type=Path) + parser.add_argument("--source", type=Path, action="append") + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--min-minutes", type=int, default=60) + parser.add_argument("--max-minutes", type=int, default=75) + parser.add_argument("--bin-seconds", type=int, default=60) + parser.add_argument("--max-acceptable-gap-s", type=float, default=5.0) + parser.add_argument("--model-position-limit", type=int, default=262144) + return parser.parse_args() + + +def percentile(values: Sequence[int | float], fraction: float) -> float | None: + if not values: + return None + ordered = sorted(float(value) for value in values) + position = (len(ordered) - 1) * fraction + lower = math.floor(position) + upper = math.ceil(position) + if lower == upper: + return ordered[lower] + return ordered[lower] * (upper - position) + ordered[upper] * (position - lower) + + +def distribution(values: Sequence[int | float]) -> dict[str, int | float | None]: + return { + "count": len(values), + "min": min(values) if values else None, + "p50": percentile(values, 0.50), + "p90": percentile(values, 0.90), + "p95": percentile(values, 0.95), + "p99": percentile(values, 0.99), + "max": max(values) if values else None, + "mean": statistics.fmean(values) if values else None, + } + + +def parse_hash_ids(value: Any) -> list[Any]: + if isinstance(value, list): + return value + if isinstance(value, str): + stripped = value.strip() + if not stripped: + return [] + if stripped.startswith("["): + decoded = json.loads(stripped) + if not isinstance(decoded, list): + raise ValueError("hash_ids JSON must decode to a list") + return decoded + delimiter = "|" if "|" in stripped else "," + return [part for part in stripped.split(delimiter) if part.strip()] + if value is None: + return [] + return [value] + + +def iter_jsonl(path: Path) -> Iterable[tuple[int, dict[str, Any]]]: + with path.open() as stream: + for line_number, line in enumerate(stream, 1): + if not line.strip(): + continue + row = json.loads(line) + if not isinstance(row, dict): + raise ValueError(f"{path}:{line_number}: row must be an object") + yield line_number, row + + +def choose_window( + *, + counts: Sequence[int], + max_gaps: Sequence[float], + first_timestamp: float, + min_minutes: int, + max_minutes: int, + bin_seconds: int, + max_acceptable_gap_s: float, +) -> dict[str, Any] | None: + candidates = [] + for minutes in range(max_minutes, min_minutes - 1, -1): + bins = math.ceil(minutes * 60 / bin_seconds) + for start_bin in range(0, len(counts) - bins + 1): + selected = counts[start_bin : start_bin + bins] + mean = statistics.fmean(selected) + cv = statistics.pstdev(selected) / mean if mean else math.inf + max_gap = max(max_gaps[start_bin : start_bin + bins], default=0.0) + candidates.append( + { + "_score": ( + max_gap > max_acceptable_gap_s, + cv, + max_gap, + -minutes, + start_bin, + ), + "start_bin": start_bin, + "minutes": minutes, + "count_mean_per_bin": mean, + "count_cv": cv, + "count_min_per_bin": min(selected), + "count_max_per_bin": max(selected), + "max_gap_s": max_gap, + } + ) + if not candidates: + return None + chosen = min(candidates, key=lambda item: item["_score"]) + chosen.pop("_score") + chosen["start_timestamp"] = first_timestamp + chosen["start_bin"] * bin_seconds + chosen["end_timestamp"] = chosen["start_timestamp"] + chosen["minutes"] * 60 + return chosen + + +def scan_source(path: Path, args: argparse.Namespace) -> dict[str, Any]: + rows = 0 + first_timestamp = None + last_timestamp = None + previous_timestamp = None + counts: Counter[int] = Counter() + max_gaps: dict[int, float] = {} + input_lengths: list[int] = [] + output_lengths: list[int] = [] + total_lengths: list[int] = [] + hash_rows = 0 + hash_matches = Counter() + prompt_rows = 0 + sampling_rows = 0 + schema_keys: Counter[str] = Counter() + for line_number, row in iter_jsonl(path): + missing = [ + key + for key in ("timestamp", "input_length", "output_length") + if key not in row + ] + if missing: + raise ValueError(f"{path}:{line_number}: missing required fields {missing}") + timestamp = float(row["timestamp"]) + if first_timestamp is None: + first_timestamp = timestamp + if previous_timestamp is not None and timestamp < previous_timestamp: + raise ValueError( + f"{path}:{line_number}: timestamp {timestamp} < {previous_timestamp}" + ) + bin_index = math.floor((timestamp - first_timestamp) / args.bin_seconds) + counts[bin_index] += 1 + if previous_timestamp is not None: + previous_bin = math.floor( + (previous_timestamp - first_timestamp) / args.bin_seconds + ) + max_gaps[previous_bin] = max( + max_gaps.get(previous_bin, 0.0), + timestamp - previous_timestamp, + ) + input_tokens = int(row["input_length"]) + output_tokens = max(1, int(row["output_length"])) + if input_tokens <= 0: + raise ValueError(f"{path}:{line_number}: input_length must be positive") + input_lengths.append(input_tokens) + output_lengths.append(output_tokens) + total_lengths.append(input_tokens + output_tokens) + hashes = parse_hash_ids(row.get("hash_ids")) + if hashes: + hash_rows += 1 + for block_size in BLOCK_SIZE_CANDIDATES: + if len(hashes) == math.ceil(input_tokens / block_size): + hash_matches[block_size] += 1 + prompt_rows += int( + isinstance(row.get("prompt"), (str, list)) and bool(row.get("prompt")) + ) + sampling_rows += int("sampling_u" in row) + schema_keys.update(row.keys()) + rows += 1 + previous_timestamp = timestamp + last_timestamp = timestamp + if not rows or first_timestamp is None or last_timestamp is None: + raise ValueError(f"{path}: empty trace") + total_bins = math.floor((last_timestamp - first_timestamp) / args.bin_seconds) + 1 + chosen = choose_window( + counts=[counts[index] for index in range(total_bins)], + max_gaps=[max_gaps.get(index, 0.0) for index in range(total_bins)], + first_timestamp=first_timestamp, + min_minutes=args.min_minutes, + max_minutes=args.max_minutes, + bin_seconds=args.bin_seconds, + max_acceptable_gap_s=args.max_acceptable_gap_s, + ) + return { + "source": str(path.resolve()), + "rows": rows, + "first_timestamp": first_timestamp, + "last_timestamp": last_timestamp, + "span_s": last_timestamp - first_timestamp, + "request_rate_per_s": rows / max(last_timestamp - first_timestamp, 1.0), + "input_length": distribution(input_lengths), + "output_length": distribution(output_lengths), + "total_length": distribution(total_lengths), + "over_max_model_len": { + str(limit): { + "requests": sum(value > limit for value in total_lengths), + "fraction": sum(value > limit for value in total_lengths) / rows, + } + for limit in MAX_MODEL_LEN_CANDIDATES + }, + "hash_contract": { + "rows_with_hash_ids": hash_rows, + "candidate_exact_match_rows": { + str(size): hash_matches[size] for size in BLOCK_SIZE_CANDIDATES + }, + "exact_source_block_size": next( + ( + size + for size in BLOCK_SIZE_CANDIDATES + if hash_rows and hash_matches[size] == hash_rows + ), + None, + ), + }, + "prompt_rows": prompt_rows, + "sampling_u_rows": sampling_rows, + "schema_field_counts": dict(sorted(schema_keys.items())), + "stable_window": chosen, + } + + +def scan_window(source: Path, window: dict[str, Any]) -> dict[str, Any]: + start = float(window["start_timestamp"]) + end = float(window["end_timestamp"]) + inputs: list[int] = [] + outputs: list[int] = [] + totals: list[int] = [] + for _, row in iter_jsonl(source): + timestamp = float(row["timestamp"]) + if timestamp < start: + continue + if timestamp >= end: + break + input_tokens = int(row["input_length"]) + output_tokens = max(1, int(row["output_length"])) + inputs.append(input_tokens) + outputs.append(output_tokens) + totals.append(input_tokens + output_tokens) + return { + "requests": len(totals), + "input_length": distribution(inputs), + "output_length": distribution(outputs), + "total_length": distribution(totals), + "max_model_len_coverage": { + str(limit): { + "covered_requests": sum(value <= limit for value in totals), + "excluded_requests": sum(value > limit for value in totals), + "coverage": sum(value <= limit for value in totals) / len(totals), + } + for limit in MAX_MODEL_LEN_CANDIDATES + }, + } + + +def resolve_sources(args: argparse.Namespace) -> list[Path]: + if args.source: + return [path.resolve() for path in args.source] + if args.trace_root is None: + raise ValueError("provide --trace-root or one or more --source") + sources = sorted( + path.resolve() + for path in args.trace_root.glob("*.jsonl") + if "prompt" not in path.stem.lower() + ) + if not sources: + raise FileNotFoundError(f"no non-prompt JSONL files under {args.trace_root}") + return sources + + +def main() -> None: + args = parse_args() + if not 0 < args.min_minutes <= args.max_minutes: + raise ValueError("require 0 < min_minutes <= max_minutes") + sources = resolve_sources(args) + files = [scan_source(path, args) for path in sources] + eligible = [item for item in files if item["stable_window"] is not None] + if not eligible: + chosen = None + data_gate = "BLOCKED_NO_1H_WINDOW" + else: + chosen = min( + eligible, + key=lambda item: ( + item["stable_window"]["max_gap_s"] > args.max_acceptable_gap_s, + item["stable_window"]["count_cv"], + -item["stable_window"]["minutes"], + item["source"], + ), + ) + chosen["selected_window_stats"] = scan_window( + Path(chosen["source"]), chosen["stable_window"] + ) + exact_block_size = chosen["hash_contract"]["exact_source_block_size"] + max_total = chosen["selected_window_stats"]["total_length"]["max"] + data_gate = ( + "PASS" + if exact_block_size is not None + and max_total is not None + and max_total <= args.model_position_limit + else "BLOCKED_HASH_OR_POSITION_CONTRACT" + ) + recommendation = None + if chosen is not None: + maximum = chosen["selected_window_stats"]["total_length"]["max"] + recommendation = next( + ( + limit + for limit in MAX_MODEL_LEN_CANDIDATES + if maximum <= limit <= args.model_position_limit + ), + None, + ) + payload = { + "schema": "frontier-code-trace-audit-v1", + "trace_root": str(args.trace_root.resolve()) if args.trace_root else None, + "sources": [str(path) for path in sources], + "window_policy": { + "min_minutes": args.min_minutes, + "max_minutes": args.max_minutes, + "bin_seconds": args.bin_seconds, + "max_acceptable_gap_s": args.max_acceptable_gap_s, + "selection": "lowest density CV after rejecting anomalous-gap windows", + }, + "model_position_limit": args.model_position_limit, + "files": files, + "selected": chosen, + "max_model_len_recommendation": recommendation, + "data_gate": data_gate, + "runtime_gate": ( + "PENDING: vLLM startup must prove enough KV blocks and nonzero " + "max concurrency at the recommended max_model_len for each TP" + ), + } + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") + print(json.dumps({"data_gate": data_gate, "output": str(args.output)})) + + +if __name__ == "__main__": + main() diff --git a/runs/frontier-code-trace-v0/experiment-card.md b/runs/frontier-code-trace-v0/experiment-card.md new file mode 100644 index 0000000..ec77a14 --- /dev/null +++ b/runs/frontier-code-trace-v0/experiment-card.md @@ -0,0 +1,204 @@ +# EXP-CODE-TRACE：从 chat 1h trace 扩展到 code 与 phase-separated replay + +> **状态：READY_FOR_DATA PREFLIGHT，未启动 GPU。** 当前只完成本地适配与实验冻结；`dash1`--`dash4` 有整机空闲后按本文 gate 顺序推进。禁止使用 `dash0`。 + +## 目标与成功定义 + +当前 1h+ 证据只覆盖 Qwen3-30B-A3B 的生产 chat trace、prefill+decode（P+D）和亚临界负载。本 campaign 分两步扩展： + +1. **主任务：** 使用 `~/ali-trace/trace-glm5.1-formatted/` 中的 1h+ code trace，先完成 P+D real-vs-Frontier 回放； +2. **后续 phase matrix：** 对 chat/code 都补 prefill-only 和严格 decode-only。 + +本轮不是只看“能否跑完”。每个正式 cell 必须满足：同一 request vector、同一 arrival、同一 token shape、同一 prefix/initial-KV 合约、real 零失败、无持续 backlog，并同时报告 TTFT/TPOT/E2E、queue/batch、KV/prefix state 与 5min 分窗漂移。 + +## 三种 workload mode 的冻结定义 + +| Mode | 保留 | 改写 | 主指标 | 明确不声称 | +|---|---|---|---|---| +| P+D | 原 ISL/OSL、arrival、session/prefix | 仅做 source block→16-token runtime block 映射 | TTFT、TPOT、E2E、hit ratio、batch/queue | 不代表 PD 分离 | +| prefill-only | 原 ISL、arrival、session/prefix | OSL 固定为 1，real `min_tokens=max_tokens=1`，sim decode tokens=1 | TTFT、prefill service/tokens/s、prefix hit、queue | TPOT 不定义；1-token decode 只用于完成请求 | +| strict decode-only | 原 OSL、context length、arrival burst | arrival 定义为 **KV-ready time**；请求进入 decode 时已有 ISL 长度的 initial KV | TPOT、decode tokens/s、batch/queue、preemption | 不包含 prefill 与 KV transfer latency，不把短 prompt proxy 称为 decode-only | + +strict decode-only 必须同时具备： + +- real：vLLM `DecodeBenchConnector`（或等价、经验证的 initial-KV 注入）； +- sim：Frontier request 在 admission 时已拥有相同长度/块布局的 computed KV； +- 两侧都不在 decode critical path 重做 prefill； +- arrival 以 KV-ready time 对齐。若只保留原 trace 的相对到达形状，结论限定为 decode engine compute/scheduling fidelity。 + +在该合约完成前，只允许跑并标注为 **decode-dominant proxy**，不能进入 strict decode-only 结果表。 + +## 为什么 code P+D 不能直接复用 chat 配置 + +已知历史探查显示 code trace ISL p90 约 81.9k，约 32.6% 请求超过旧 `40960` 上限；真实数值必须由本 campaign 重新审计。至少有四个独立适配面： + +1. **Serving cap：** `max_model_len` 必须覆盖 `ISL+OSL`，不能只看 ISL，也不能静默丢掉超长请求； +2. **KV capacity：** Qwen3-30B 模型 position limit 为 262144，但 TP1/2/4 在 H20 上是否有足够 KV blocks 是 runtime gate，不由 config.json 自动保证； +3. **Prefix block：** code source hash 预计为 512-token block，chat harness 原先固定 64→16； +4. **Profile support：** 当前修复后的 attention profile 只覆盖到约 32k KV context。即使 vLLM 能跑 128k，Frontier 对 32k–128k 仍会出 profile 支撑域；在补 long-context 网格前只能做诊断 replay，不能做 fidelity claim。 + +## Hypotheses + +- **H-code-generalizes：** 在补齐 long-context profile 支撑域后，code P+D 的 TTFT/TPOT/E2E 分布统计偏差仍处于当前 chat 量级，且 1h 残差不发散。 +- **H-longctx-gap：** code 的主要新增 gap 来自 32k 以上 KV-context 外推；补到 trace p99/max 对应的网格后，TTFT bias 随 ISL 的二次项显著收敛。 +- **H-phase-specific：** prefill-only 主要暴露 long-context/profile gap；strict decode-only 主要暴露 batch-conditioned whole-layer service 与 scheduler fixed-point gap。二者不能用 P+D 的误差抵消来互相证明准确。 + +## Preflight gates（按顺序，任一失败即停止后续真机矩阵） + +### G0：数据位置与 provenance + +- 只读列举 `trace-glm5.1-formatted/*.jsonl`，记录文件大小与 SHA256； +- 确认至少两个独立日期段：一个作为 development，一个 held-out； +- 本机当前没有该目录；仓库历史记录的远端位置为 + `/home/admin/cpfs/wjh/ali-trace/trace-glm5.1-formatted/`。恢复机器后先确认 `~/ali-trace/...` 是否为同一路径/软链，不能假设。 + +### G1：1h window、schema 与 block contract + +运行 `audit_code_trace.py`，要求： + +- timestamp 单调，存在 60–75min 连续稳定窗口； +- `timestamp/input_length/output_length` 全行存在； +- `hash_ids` 数量与某个 source block size 在全行严格满足 + `ceil(ISL/source_block_size)`；预计值 512，但以审计结果为准； +- 记录 ISL/OSL/ISL+OSL 的 p50/p90/p95/p99/max、gap、request rate、prompt/sampling 字段覆盖。 + +选择窗口后用 `prepare_code_window.py` 物化只读派生文件，并按 session root 生成确定性的 `sampling_u`。另一日期段不参与 rho 与 profile 选择。 + +### G2：`max_model_len` data gate + +候选 cap 固定为 `40960/65536/98304/131072/262144`，选能 **100% 覆盖选中窗口 `ISL+OSL`** 的最小值。规则： + +- 若 max≤131072，主路径使用 131072 或更小的审计推荐值； +- 若存在 >131072 请求，不允许悄悄过滤。优先验证 262144；若 runtime 不可行，必须预注册过滤比例，并把 claim 改为“≤131072 子群”； +- Frontier 的 trace max tokens、predictor max tokens/request、vLLM `--max-model-len` 三处使用同一个 manifest 值。 + +### G3：prompt 与 prefix fidelity + +优先级： + +1. 有对齐 prompt text sidecar：用 Qwen tokenizer 重分词，要求 token length 与 trace ISL 全行一致； +2. trace 内已有 prompt text/token IDs：同样做长度与 hash relation 检查； +3. 两者都没有：允许用 source hash 确定性展开为 synthetic Qwen token IDs，但结果降级为 **length/arrival/prefix-shape faithful**，不声称 prompt-content 或 MoE routing faithful。 + +不论走哪条路径，source→16 映射冲突、runtime identity collision、parent prefix violation 都必须为 0。P+D/prefill-only 两侧 prefix caching 同开；先用 5–10min TP4/MNS16 做 hit-ratio audit。 + +### G4：long-context profile support + +现有 profile-v5 的 KV context 上界约 32k，对 code 不足。根据 development window 的 uncached-ISL 分布生成 profile-v6-code-longctx： + +- full chunk：`q8k`，context 至少覆盖 32k/48k/64k/80k/96k/112k/120k； +- tail chunk：从真实 `ISL mod 8192` 的 p50/p90 选择 2–4k/4–6k 代表点； +- TP1/2/4 分开采集，复测 `q1ks8k/q8ks32k` anchor； +- 每点至少两次 fresh-process repeat；CV≤5%，anchor drift≤10%； +- profile max context 必须 ≥ development window p99；正式 max claim 要求 ≥ max。若只覆盖 p99，max 以上请求单独列为 out-of-support，不进入总体准确度数字。 + +这是 code P+D 正式 fidelity 的硬 gate。可以先用旧 profile 跑 diagnostic sim 来估 load，但不得与真机组成最终 gap。 + +### G5：vLLM max-length/KV runtime gate + +对每个候选 topology（先 TP4，再 TP2，TP1 后置）： + +1. fresh server，以 manifest cap 启动； +2. 记录 vLLM 版本、model config、GPU KV blocks、maximum concurrency、启动日志； +3. 发 3 个单请求：ISL p50、p99、max（OSL=1），usage 必须逐 token 对齐； +4. 发 5min sampled P+D canary，零 OOM/timeout/preemption storm； +5. 只有 maximum concurrency>1 且 canary drain tail≤窗口时长 10% 才进入 rho calibration。 + +`max_model_len` 变大不等于每个请求都预占最大 KV，但会改变启动合法性与可表达的单请求上界；实际 KV 压力仍由并发 token state 决定。 + +### G6：每种 mode 独立标定 rho + +不能复用 P+D rho： + +- P+D 同时按 raw/prefix-adjusted prefill tokens/s 与 decode tokens/s 看 knee； +- prefill-only 因 OSL=1，重新按 prefill work 标定； +- strict decode-only 因无 prefill，按 decode tokens/s 和 batch fixed point 标定。 + +每种 workload×mode 选择 `low/mid/near-knee` 三点；正式点必须亚临界：全请求完成、无持续 backlog、drain tail≤10%、waiting p99 不单调随时间增长。跨 knee 点若运行，只作为 overload boundary，不支持“不发散”结论。 + +### G7：strict decode-only capability gate + +先在 10min synthetic trace 上验证： + +- real connector 确认没有执行 prefill kernel； +- Frontier ledger 第一个阶段就是 decode，computed tokens=ISL； +- 相同 context length 下两侧 KV block count 一致； +- connector preload/transfer 时间独立记账，不混入 TPOT； +- decode batch telemetry 能覆盖 b1 到目标 batch。 + +若 vLLM 0.20 community stack 没有等价 connector，严格 case 保持 BLOCKED；可另跑 decode-dominant proxy，但单独命名和汇报。 + +## 正式实验矩阵与推进顺序 + +### Phase A：code P+D（第一优先级） + +1. **A0 CPU/data：** G0–G4； +2. **A1 max-len smoke：** TP4→TP2；TP1 只在 KV gate 通过后加入； +3. **A2 paired 10min canary：** TP4/MNS16，low rho，real+sim； +4. **A3 calibration：** 各 rho 只先跑 sim，冻结 low/mid/near-knee； +5. **A4 full：** TP4/MNS16、TP2/MNS16 × 3 rho × 2 trial × 60–75min； +6. **A5 held-out：** 只在 development window 判据冻结后，对第二日期段跑 TP4 的 mid/near-knee。 + +若某 topology 的 near-knee 过载，像现有 chat TP2/ρ0.01 一样排除，不为凑齐矩阵强跑。 + +### Phase B：chat/code prefill-only + +- 复用各自已物化 window，只把 OSL 改为 1； +- primary：TP4/MNS16、TP2/MNS16 × 3 独立 rho × 2 trial； +- 报 TTFT/CDF/quantiles、prefill tokens/s、prefix hit、waiting、chunk/context 分带 residual； +- TPOT 记为 N/A，E2E 仅作为“一 token completion”辅助值； +- code 必须继续使用 profile-v6 long-context；chat 使用已验证 profile-v5。 + +### Phase C：chat/code strict decode-only + +先做 batch-sensitive screening，再决定是否铺满： + +- **C0 capability canary：** 两 workload × TP4 × MNS{16,128}，10min； +- **C1 core full：** TP{2,4} × MNS{16,128} × rho{low,near-knee} × 2 trial； +- **C2 conditional expansion：** 只有当 C1 的 batch 分布从 b≤8 跨到 b>8，或 accuracy gap 随 MNS 改变>5pp，才补 MNS{32,64} 与 mid rho。 + +decode profile/serving anchors 至少覆盖实际 batch p99。当前 whole-layer grid 只对少数 b≤8 有证据，且 b6 有长尾；在 MNS128 case 前必须补 b{1,2,4,8,16,32,64,128} 或实际访问 bucket，不能把 b8 常数外推到 b128。 + +## 指标与判据 + +共同口径： + +- 分布统计偏差：`(sim statistic-real statistic)/real statistic`，不是 per-request MAPE； +- mean/p50/p90/p99 与 empirical CDF； +- 5min 分窗，前 15min warmup 不进漂移 slope； +- batch histogram、time-weighted running/waiting、drain tail、preemption； +- 两 trial pooled 结果和 trial-to-trial noise floor 分开报告。 + +判据分两层： + +1. **准确度：** primary latency mean/p90/p99 的 |bias|≤15% 为强通过，15–30% 为有界但需标注 correction，>30% 立 bad case；任何 topology 排序或 SLO feasibility 翻转都单独判 failure，不能被平均值掩盖。 +2. **长时稳定：** `|residual Theil–Sen slope|×12 / real noise floor < 1` 为 H-BOUNDED；只适用于亚临界 cell。 + +mode-specific： + +- P+D：TTFT/TPOT/E2E 全部 primary； +- prefill-only：TTFT primary，TPOT N/A； +- strict decode-only：TPOT primary，TTFT 仅表示 admission/connector overhead，不进入 compute-fidelity gate。 + +## 成本与调度 + +- Phase A core：12 个 60–75min jobs（2 topology×3 load×2 trial），约 15 host-hours；按 TP 加权约 45 H20-GPU-hours，加 2–4 个 smoke/canary； +- Phase B 两 workload：24 个 full jobs，按相同 75min 上界约 90 H20-GPU-hours； +- Phase C 不一次铺满。C0 4 个 10min canary；C1 32 个 full jobs；C2 按触发条件追加。 + +每个 job fresh server；只在 `dash1`–`dash4` 全 8 卡 idle/healthy 时启动。即使 TP2/TP4 job 只用部分 GPU，也不在同一 host 并跑，避免 fresh-server 空窗竞态。每一批使用新的 jobs TOML，现有 dispatcher 非幂等。 + +## 预期产物 + +- `inputs/code-audit.json`、`inputs/code-window/window-manifest.json`； +- P+D/prefill-only 的 paired `frontier.csv`、`real_requests.jsonl` 与 manifest； +- profile-v6-code-longctx raw/merged profile 与 variance report； +- 每 cell real/sim request metrics、server telemetry、stage ledger； +- `results/code-pd-fidelity.md`； +- 最终 `chat/code × P+D/prefill-only/decode-only` compatibility table。 + +## 已知边界 + +- code trace 来自 GLM5.1 业务，serving model 是 Qwen3-30B；若无原 prompt text，测试只能保持 shape/prefix 结构，不能证明内容相关 routing fidelity； +- `max_model_len=128k/256k` 解决的是接入上界，不自动解决 32k 以上 profile 外推； +- strict decode-only 只测 decode engine；完整 PD 分离还需要单独建模 prefill、KV transfer、backpressure 与 KV-ready arrival。 diff --git a/runs/frontier-code-trace-v0/prepare_code_window.py b/runs/frontier-code-trace-v0/prepare_code_window.py new file mode 100644 index 0000000..28065d9 --- /dev/null +++ b/runs/frontier-code-trace-v0/prepare_code_window.py @@ -0,0 +1,114 @@ +#!/usr/bin/env python3 +"""Materialize the stable code window selected by audit_code_trace.py.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +from pathlib import Path +from typing import Any + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--audit", type=Path, required=True) + parser.add_argument("--output-root", type=Path, required=True) + parser.add_argument("--sample-seed", type=int, default=20260723) + return parser.parse_args() + + +def session_uniform(seed: int, window_id: str, session_root: Any) -> float: + payload = json.dumps( + {"seed": seed, "window_id": window_id, "session_root": session_root}, + sort_keys=True, + separators=(",", ":"), + ).encode() + return int.from_bytes(hashlib.blake2b(payload, digest_size=8).digest(), "big") / ( + 1 << 64 + ) + + +def sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + for chunk in iter(lambda: stream.read(1 << 20), b""): + digest.update(chunk) + return digest.hexdigest() + + +def main() -> None: + args = parse_args() + audit = json.loads(args.audit.read_text()) + if audit["data_gate"] != "PASS": + raise ValueError(f"trace data gate is not PASS: {audit['data_gate']}") + selected = audit["selected"] + source = Path(selected["source"]) + window = selected["stable_window"] + start = float(window["start_timestamp"]) + end = float(window["end_timestamp"]) + if args.output_root.exists(): + raise ValueError(f"refusing to overwrite {args.output_root}") + args.output_root.mkdir(parents=True) + destination = args.output_root / "code-raw-window.jsonl" + root_of: dict[Any, Any] = {} + request_count = 0 + with source.open() as input_stream, destination.open("w") as output_stream: + for source_index, line in enumerate(input_stream): + if not line.strip(): + continue + row = json.loads(line) + timestamp = float(row["timestamp"]) + if timestamp < start: + continue + if timestamp >= end: + break + chat = row.get("chat_id", source_index) + parent = row.get("parent_chat_id") + has_parent = parent not in (None, "", -1, "-1") + session_root = root_of.get(parent, parent) if has_parent else chat + root_of[chat] = session_root + materialized = { + **row, + "source_index": source_index, + "session_root": session_root, + "sampling_u": session_uniform( + args.sample_seed, + f"code-{start:.6f}-{end:.6f}", + session_root, + ), + } + output_stream.write( + json.dumps(materialized, ensure_ascii=False, separators=(",", ":")) + + "\n" + ) + request_count += 1 + expected = int(selected["selected_window_stats"]["requests"]) + if request_count != expected: + raise ValueError(f"window request mismatch: materialized={request_count}, audit={expected}") + manifest = { + "schema": "frontier-code-window-v1", + "audit": str(args.audit.resolve()), + "audit_sha256": sha256(args.audit), + "source": str(source.resolve()), + "source_block_size": selected["hash_contract"]["exact_source_block_size"], + "target_block_size": 16, + "start_timestamp": start, + "end_timestamp": end, + "duration_s": end - start, + "requests": request_count, + "sample_seed": args.sample_seed, + "sampling_rule": "session-coherent deterministic sampling_u", + "max_model_len": audit["max_model_len_recommendation"], + "window_stats": selected["selected_window_stats"], + "raw_window": str(destination.resolve()), + "raw_window_sha256": sha256(destination), + } + (args.output_root / "window-manifest.json").write_text( + json.dumps(manifest, indent=2, sort_keys=True) + "\n" + ) + print(json.dumps({"requests": request_count, "output_root": str(args.output_root)})) + + +if __name__ == "__main__": + main() diff --git a/runs/frontier-code-trace-v0/test_code_trace_preflight.py b/runs/frontier-code-trace-v0/test_code_trace_preflight.py new file mode 100644 index 0000000..0f5a19f --- /dev/null +++ b/runs/frontier-code-trace-v0/test_code_trace_preflight.py @@ -0,0 +1,80 @@ +#!/usr/bin/env python3 + +from __future__ import annotations + +import json +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + + +ROOT = Path(__file__).resolve().parent + + +class CodeTracePreflightTest(unittest.TestCase): + def test_audit_and_materialize_512_block_window(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + source = root / "051315-051317.jsonl" + with source.open("w") as stream: + for index in range(4501): + input_tokens = 513 if index % 2 else 512 + stream.write( + json.dumps( + { + "chat_id": index, + "parent_chat_id": index - 1 if index % 2 else -1, + "timestamp": float(index), + "input_length": input_tokens, + "output_length": 32, + "hash_ids": [index // 2] + if input_tokens == 512 + else [index // 2, 100000 + index], + } + ) + + "\n" + ) + audit = root / "audit.json" + subprocess.run( + [ + sys.executable, + str(ROOT / "audit_code_trace.py"), + "--source", + str(source), + "--output", + str(audit), + ], + check=True, + ) + payload = json.loads(audit.read_text()) + self.assertEqual(payload["data_gate"], "PASS") + self.assertEqual( + payload["selected"]["hash_contract"]["exact_source_block_size"], 512 + ) + self.assertEqual(payload["max_model_len_recommendation"], 40960) + output = root / "window" + subprocess.run( + [ + sys.executable, + str(ROOT / "prepare_code_window.py"), + "--audit", + str(audit), + "--output-root", + str(output), + ], + check=True, + ) + manifest = json.loads((output / "window-manifest.json").read_text()) + rows = [ + json.loads(line) + for line in (output / "code-raw-window.jsonl").read_text().splitlines() + ] + self.assertEqual(manifest["requests"], len(rows)) + self.assertGreaterEqual(manifest["duration_s"], 3600) + self.assertEqual(rows[0]["sampling_u"], rows[1]["sampling_u"]) + + +if __name__ == "__main__": + unittest.main()