Result: ' + escape(status) + + '
' + summary_html + index_html) + atomic_text(out / "report.md", markdown) + atomic_text(out / "report.html", html) + atomic_text(out / "report.json", payload) + except Exception as exc: + report["passed"] = False + report["status"] = "FAIL" + report.setdefault("report_errors", []).append(f"{type(exc).__name__}: {exc}") + # A human may open a view directly; remove already-written PASS views + # as well as JSON when any required report fails to publish. + for filename in ("report.md", "report.html", "report.json"): + try: + (out / filename).unlink(missing_ok=True) + except OSError as cleanup_error: + report["report_errors"].append(f"cleanup {filename}: {cleanup_error}") + target = out / "report.json" + try: + target.unlink(missing_ok=True) + atomic_text(target, json.dumps(report, ensure_ascii=False, indent=2, allow_nan=False) + "\n") + except Exception: + pass + raise diff --git a/probes/w3_full_preflight/README.md b/probes/w3_full_preflight/README.md new file mode 100644 index 0000000..02661f8 --- /dev/null +++ b/probes/w3_full_preflight/README.md @@ -0,0 +1,15 @@ +# W3 完整资产与静态尺寸预检 + +复现统一采用 **Ubuntu 24.04 x86_64、Bash、Python 3.12**。先完成 [Linux 环境与资产准备](../../docs/llm-deploy-v1.0/LINUX_REPRODUCTION.md),再在同一个 Bash 会话、仓库根目录执行命令;该指南设置 `SCRATCHV_PYTHON`、`SCRATCHV_CC` 和 `SCRATCHV_QEMU`。 + +~~~bash +set -euo pipefail +W3_MODEL=/absolute/path/to/qwen3-0.6b-onnx +"$SCRATCHV_PYTHON" -B -X utf8 probes/w3_full_preflight/run.py --model-dir "$W3_MODEL" --output-dir output/w3-full-preflight-new +~~~ + +离线流式校验 W1 固定 manifest 的模型和三个 external-data 文件,复用 ONNX 静态契约/范围检查,统计 initializer、嵌入/外部 tensor、具名 value 与输出的逻辑字节。不加载外部数组进行 IR 运算,也不执行完整 ORT。 + +报告始终标记 full_ir_executed=false、w3_exit_accepted=false。PASS 只代表准备检查通过;sum_named_value_logical_bytes 包含别名、可能重复,不能作为所需物理内存上限或峰值承诺。process_peak_rss_bytes 只属于这个 metadata-only 预检进程。 + +完整数值验收仍需 W1 人工前置确认、中等与真实子图基线、完整 IR/ORT <1e-4、三人复现和 Nightly。参见 [W3 准备说明](../../docs/llm-deploy-v1.0/W3/README.md)。 diff --git a/probes/w3_full_preflight/run.py b/probes/w3_full_preflight/run.py new file mode 100644 index 0000000..8943d32 --- /dev/null +++ b/probes/w3_full_preflight/run.py @@ -0,0 +1,101 @@ +"""Verify fixed full-model assets and size accounting without executing full IR.""" +from __future__ import annotations +import argparse +from collections import Counter +import math +from pathlib import Path +import sys +import time +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) +import numpy as np +import onnx +from probes.w1_qwen3_export.run import MANIFEST, verify_files, inspect_model, tensors +from probes.w3_common import new_output_dir, source_evidence, write_reports, process_peak_rss_bytes + +def tensor_size(tensor): + return math.prod(tensor.dims) * np.dtype(onnx.helper.tensor_dtype_to_np_dtype(tensor.data_type)).itemsize + +def value_size(value): + kind = value.type.tensor_type + if not kind.HasField("shape"): + return None + dimensions = [] + for dim in kind.shape.dim: + if not dim.HasField("dim_value"): + return None + dimensions.append(dim.dim_value) + return math.prod(dimensions) * np.dtype(onnx.helper.tensor_dtype_to_np_dtype(kind.elem_type)).itemsize + +def size_accounting(model): + names = {} + unknown = [] + for value in (*model.graph.input, *model.graph.value_info, *model.graph.output): + count = value_size(value) + if count is None: + unknown.append(value.name) + else: + if value.name in names and names[value.name] != count: + raise ValueError(f"Conflicting shape information: {value.name}") + names[value.name] = count + initializer_bytes = sum(tensor_size(t) for t in model.graph.initializer) + all_tensor_bytes = sum(tensor_size(t) for t in tensors(model)) + graph_names = {v.name for v in model.graph.input} | {t.name for t in model.graph.initializer} + graph_names.update(name for node in model.graph.node for name in node.output if name) + known = set(names) | {t.name for t in model.graph.initializer} + return { + "initializer_logical_bytes": initializer_bytes, + "all_embedded_and_external_tensor_logical_bytes": all_tensor_bytes, + "sum_named_value_logical_bytes": sum(names.values()), + "named_values_with_static_size": len(names), "unknown_size_values": unknown, + "graph_value_count": len(graph_names), + "values_without_persisted_static_size": sorted(graph_names - known), + "output_logical_bytes": {v.name:value_size(v) for v in model.graph.output}, + "largest_named_values": sorted(names.items(), key=lambda x:(-x[1],x[0]))[:10], + "scope": "Static ONNX logical sizes, including aliases. Not measured IR live storage or a process RAM requirement.", + } + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--model-dir", type=Path, required=True) + parser.add_argument("--output-dir", type=Path, required=True) + args = parser.parse_args(argv) + out = new_output_dir(args.output_dir) + report = { + "gate":"preflight:w3-full-assets", "status":"FAIL", "passed":False, "stage":"source_evidence", + "full_ir_executed":False, "w3_exit_accepted":False, + "scope":"Asset integrity and static size preparation only; no ORT/IR/full-model numerical acceptance.", + "required_before_full_acceptance":[ + "W1 human acceptance and interface/risk decisions", + "Medium and pretrained subgraph numeric/memory baselines", + "Measured host capacity and bounded full IR/ORT runner", + "Full IR vs ORT <1e-4, three independent reproductions and successful Nightly", + ], + } + start = time.perf_counter() + try: + report.update(source_evidence()) + report["stage"] = "asset_hashes" + files = verify_files(args.model_dir) + report["files"] = files + report["source_revision"] = MANIFEST["revision"] + report["stage"] = "onnx_contract" + report["onnx_contract"] = inspect_model(args.model_dir, files) + report["stage"] = "static_size_accounting" + model = onnx.load(str(args.model_dir / "model.onnx"), load_external_data=False) + report["node_count"] = len(model.graph.node) + report["operator_counts"] = dict(Counter(node.op_type for node in model.graph.node)) + report["sizes"] = size_accounting(model) + report["weights_file_bytes"] = sum(item["bytes"] for item in files if item["name"] != "model.onnx") + report.update(passed=True, status="PASS", stage="complete") + except Exception as exc: + report["error"] = f"{type(exc).__name__}: {exc}" + report.update(elapsed_seconds=time.perf_counter()-start, + process_peak_rss_bytes=process_peak_rss_bytes(), + peak_scope="This metadata-only preflight process, not full parser/IR execution") + write_reports(out,report) + return 0 if report["passed"] else 1 + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/probes/w3_layer_diff/README.md b/probes/w3_layer_diff/README.md new file mode 100644 index 0000000..94c18dd --- /dev/null +++ b/probes/w3_layer_diff/README.md @@ -0,0 +1,65 @@ +# W3 offline checkpoint comparison + +Reproduction targets **Ubuntu 24.04 x86_64, Bash and Python 3.12**. Complete the [Linux environment and asset setup](../../docs/llm-deploy-v1.0/LINUX_REPRODUCTION.md) first. Run commands from the repository root in the same Bash session; the setup exports `SCRATCHV_PYTHON`, `SCRATCHV_CC` and `SCRATCHV_QEMU`. + +This tool compares recorded NPZ tensors without executing a model. It reuses +`probes.w2_qwen3_small.diagnostics.compare_outputs`: exact shape and dtype, +finite numeric values, and strict `max_abs < atol` (default `1e-5`). + +```bash +set -euo pipefail +"$SCRATCHV_PYTHON" probes/w3_layer_diff/run.py --actual output/w3/medium/traces/full_seed_0/ir_diagnostic_none.npz --reference output/w3/medium/traces/full_seed_0/reference.npz --schema output/w3/medium/trace_schema.json --out output/w3/layerdiff-full +``` + +`--out` must be new. The shared W3 publisher writes `report.json`, `report.md` +and `report.html` atomically per file, with JSON last. Publication errors fail +the command. No network, pickle loading or model execution is involved. + +The schema is UTF-8 JSON with an explicit ordered checkpoint list: + +```json +{"version":1,"checkpoints":[{"name":"layer_0.hidden","shape":[1,256,128],"dtype":"float32","layer":0,"checkpoint":"hidden","sequence_axis":1}]} +``` + +Only `name`, `shape`, and `dtype` are mandatory in each entry. `layer` is a +nonnegative integer; `checkpoint` is a local selector name; `sequence_axis` is +an optional valid axis or null. `--reference-schema` can provide a second schema +which must match exactly, including checkpoint order and metadata. Every NPZ +must have exactly the schema names, no missing/extra arrays or duplicate ZIP +members. Duplicate JSON keys/checkpoint names, object arrays, shape/type +disagreements and nonfinite values fail before comparison. ZIP expanded size +and schema logical size are limited by `--max-bytes` (default 2 GiB per NPZ); +schema size is limited to 8 MiB. NPY v1/v2 headers and payload sizes are checked +against the schema before NumPy allocates an array. This is validation of local evidence, not a +claim to safely process arbitrary hostile archives. + +Repeat `--layer N` or `--checkpoint NAME` to select checkpoints. A checkpoint +selector matches an exact name or the optional local `checkpoint` field. +Layer and checkpoint filters intersect; unknown, duplicate and empty selections +fail. All evidence names, shapes, dtypes and finite values are still validated. +Every explicitly selected comparison reports **PARTIAL**, even when the +selection happens to cover every checkpoint; it cannot produce a full PASS. + +Exit codes: `0` = all recorded checkpoints compared and passed; `1` = numeric +difference, invalid evidence or report publication failure; `2` = selected +checkpoints passed with partial coverage. Reports include ordered per-checkpoint +errors, the first divergence, the worst absolute-error element, source evidence +and artifact SHA-256 hashes. A full checkpoint PASS establishes no claim about +whole-model correctness, causal/padding tests or backend acceptance. + +## IR memory observations + +`IRInterpreter.run(..., collect_memory_stats=True)` populates +`ExecutionResult.memory_stats`; it is `None` by default. The optional observation +records input/initializer logical bytes, peak retained logical bytes, peak +deduplicated NumPy backing storage, and retained value counts. Inputs retain +their existing copy semantics; initializers count referenced bound globals +(including scalar global constants). Views count in logical bytes but share one +storage owner. ALLOCA buffers are counted. Final return copies are included in +peaks and separately reported, but excluded from retained counts. + +These are instruction-boundary snapshots of values retained by the interpreter, +not RSS or total process allocation. They exclude caller arrays, transient +NumPy kernel buffers, Python object sizes and allocator overhead. The observer +does not release values, change kernels, alter control flow or add tensor copies. +Use the separate W3 process RSS metric for process-level memory evidence. diff --git a/probes/w3_layer_diff/__init__.py b/probes/w3_layer_diff/__init__.py new file mode 100644 index 0000000..be43338 --- /dev/null +++ b/probes/w3_layer_diff/__init__.py @@ -0,0 +1 @@ +"""Offline W3 checkpoint evidence comparison.""" diff --git a/probes/w3_layer_diff/run.py b/probes/w3_layer_diff/run.py new file mode 100644 index 0000000..7be1e81 --- /dev/null +++ b/probes/w3_layer_diff/run.py @@ -0,0 +1,260 @@ +#!/usr/bin/env python3 +"""Compare recorded checkpoint arrays without executing a model or loading pickle.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import math +from pathlib import Path +import sys +import zipfile + +import numpy as np + +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from probes.w2_qwen3_small.diagnostics import compare_outputs, tensor_diff +from probes.w3_common import source_evidence, write_reports + +DEFAULT_MAX_BYTES = 2 * 1024**3 + + +class EvidenceError(ValueError): + def __init__(self, message, checkpoint=None, details=None): + super().__init__(message) + self.checkpoint = checkpoint + self.details = details + + +def _unique_object(pairs): + result = {} + for key, value in pairs: + if key in result: + raise EvidenceError(f"duplicate JSON key: {key}") + result[key] = value + return result + + +def _integer(value): + return isinstance(value, int) and not isinstance(value, bool) + + +def load_schema(path: Path) -> list[dict]: + if path.stat().st_size > 8 * 1024**2: + raise EvidenceError("schema exceeds 8 MiB limit") + document = json.loads(path.read_text(encoding="utf-8"), object_pairs_hook=_unique_object, + parse_constant=lambda value: (_ for _ in ()).throw( + EvidenceError(f"nonfinite JSON constant: {value}"))) + if (not isinstance(document, dict) or not _integer(document.get("version")) + or document["version"] != 1): + raise EvidenceError("schema must be an object with version=1") + entries = document.get("checkpoints") + if not isinstance(entries, list) or not entries: + raise EvidenceError("schema checkpoints must be a nonempty ordered list") + seen = set() + for entry in entries: + if not isinstance(entry, dict) or not {"name", "shape", "dtype"} <= set(entry): + raise EvidenceError("each checkpoint requires name, shape and dtype") + name = entry["name"] + if not isinstance(name, str) or not name or name in seen: + raise EvidenceError(f"empty, invalid or duplicate checkpoint name: {name!r}") + seen.add(name) + shape = entry["shape"] + if (not isinstance(shape, list) + or any(not _integer(dim) or dim < 0 for dim in shape)): + raise EvidenceError("shape must contain nonnegative integer dimensions", name) + try: + if not isinstance(entry["dtype"], str): + raise TypeError("dtype must be a string") + dtype = np.dtype(entry["dtype"]) + except (TypeError, ValueError) as exc: + raise EvidenceError(f"invalid dtype: {entry['dtype']!r}", name) from exc + if dtype.kind not in "biuf" or dtype.hasobject: + raise EvidenceError("dtype must be real numeric or boolean, without objects", name) + if "layer" in entry and (not _integer(entry["layer"]) or entry["layer"] < 0): + raise EvidenceError("layer must be a nonnegative integer", name) + if "checkpoint" in entry and (not isinstance(entry["checkpoint"], str) + or not entry["checkpoint"]): + raise EvidenceError("checkpoint must be a nonempty string", name) + axis = entry.get("sequence_axis") + if axis is not None and (not _integer(axis) or not -len(shape) <= axis < len(shape)): + raise EvidenceError("sequence_axis is outside checkpoint rank", name) + return entries + + +def load_arrays(path: Path, entries: list[dict], max_bytes=DEFAULT_MAX_BYTES) -> dict: + """Preflight ZIP membership/size, then load numeric NPY arrays with no pickle.""" + if not _integer(max_bytes) or max_bytes < 1: + raise EvidenceError("max_bytes must be a positive integer") + names = [entry["name"] for entry in entries] + by_name = {entry["name"]: entry for entry in entries} + with zipfile.ZipFile(path) as archive: + members = archive.infolist() + raw_names = [member.filename for member in members] + if len(set(raw_names)) != len(raw_names): + raise EvidenceError("NPZ contains duplicate ZIP members") + if any(not name.endswith(".npy") or name == ".npy" for name in raw_names): + raise EvidenceError("NPZ must contain only named .npy array members") + keys = [name[:-4] for name in raw_names] + # NpzFile accepts both member paths and names without the .npy suffix; + # keys such as x and x.npy would otherwise resolve to the same member. + if set(keys) & set(raw_names): + raise EvidenceError("NPZ contains ambiguous checkpoint/member name aliases") + if len(set(keys)) != len(keys) or set(keys) != set(names): + missing, extra = sorted(set(names) - set(keys)), sorted(set(keys) - set(names)) + first = next((name for name in names if name not in keys), None) + raise EvidenceError(f"NPZ names must exactly match schema; missing={missing}, extra={extra}", first) + if sum(member.file_size for member in members) > max_bytes: + raise EvidenceError("NPZ expanded data exceeds max_bytes") + if sum(math.prod(entry["shape"]) * np.dtype(entry["dtype"]).itemsize + for entry in entries) > max_bytes: + raise EvidenceError("schema logical data exceeds max_bytes") + # Inspect headers before NumPy can allocate arrays. A tiny forged member + # claiming a huge shape must not bypass the expanded ZIP size limit. + for member in members: + name = member.filename[:-4] + entry = by_name[name] + with archive.open(member) as stream: + version = np.lib.format.read_magic(stream) + if version == (1, 0): + shape, _, dtype = np.lib.format.read_array_header_1_0(stream, max_header_size=10000) + elif version == (2, 0): + shape, _, dtype = np.lib.format.read_array_header_2_0(stream, max_header_size=10000) + else: + raise EvidenceError(f"unsupported NPY version: {version}", name) + if list(shape) != entry["shape"] or dtype != np.dtype(entry["dtype"]): + raise EvidenceError(f"{name}: NPY header shape/dtype disagrees with schema", name) + if dtype.hasobject or dtype.kind not in "biuf": + raise EvidenceError(f"{name}: NPY dtype is not safe numeric data", name) + payload_bytes = math.prod(shape) * dtype.itemsize + if stream.tell() + payload_bytes != member.file_size: + raise EvidenceError(f"{name}: NPY payload size disagrees with header", name) + result = {} + with np.load(path, allow_pickle=False, max_header_size=10000) as archive: + for entry in entries: + name = entry["name"] + try: + array = archive[name] + except (ValueError, TypeError, EOFError) as exc: + raise EvidenceError(f"cannot safely load checkpoint {name}: {exc}", name) from exc + if list(array.shape) != entry["shape"]: + raise EvidenceError(f"{name}: array shape {list(array.shape)} disagrees with schema {entry['shape']}", name) + if array.dtype != np.dtype(entry["dtype"]): + raise EvidenceError(f"{name}: array dtype {array.dtype} disagrees with schema {entry['dtype']}", name) + if array.dtype.kind not in "biuf" or not np.isfinite(array).all(): + diagnostic = tensor_diff(array, array) + raise EvidenceError(f"{name}: nonfinite or unsupported array", name, + dict(name=name, evidence_path=str(path.resolve()), + worst_index=diagnostic["worst_index"], + value=diagnostic["actual_value"], + reason=diagnostic["reason"])) + result[name] = array + return result + + +def select_names(entries: list[dict], layers=(), checkpoints=()) -> list[str]: + if len(set(layers)) != len(layers) or len(set(checkpoints)) != len(checkpoints): + raise EvidenceError("duplicate layer/checkpoint selections are not allowed") + known_layers = {entry["layer"] for entry in entries if "layer" in entry} + if any(not _integer(layer) or layer < 0 or layer not in known_layers for layer in layers): + raise EvidenceError(f"layer selection is outside schema layers {sorted(known_layers)}") + known_checkpoints = {entry["name"] for entry in entries} + known_checkpoints.update(entry["checkpoint"] for entry in entries if "checkpoint" in entry) + if set(checkpoints) - known_checkpoints: + raise EvidenceError(f"unknown checkpoint selections: {sorted(set(checkpoints) - known_checkpoints)}") + names = [entry["name"] for entry in entries + if (not layers or entry.get("layer") in layers) + and (not checkpoints or entry["name"] in checkpoints + or entry.get("checkpoint") in checkpoints)] + if not names: + raise EvidenceError("layer/checkpoint selection has an empty intersection") + return names + + +def compare_evidence(actual_path, reference_path, schema_path, *, reference_schema_path=None, + layers=(), checkpoints=(), atol=1e-5, max_bytes=DEFAULT_MAX_BYTES): + # tensor_diff owns the W2 finite positive tolerance contract, even for bad inputs. + tensor_diff(np.array([], dtype="float32"), np.array([], dtype="float32"), atol) + entries = load_schema(Path(schema_path)) + reference_entries = load_schema(Path(reference_schema_path)) if reference_schema_path else entries + if entries != reference_entries: + raise EvidenceError("actual/reference schemas must match exactly in checkpoint order and metadata") + names = select_names(entries, layers, checkpoints) + # Always validate every member, including unselected evidence, before comparison. + actual = load_arrays(Path(actual_path), entries, max_bytes) + expected = load_arrays(Path(reference_path), reference_entries, max_bytes) + comparison = compare_outputs({name: actual[name] for name in names}, + {name: expected[name] for name in names}, names, atol=atol) + partial = bool(layers or checkpoints) + passing = comparison["passed"] + worst = max(comparison["checkpoints"], + key=lambda row: (math.inf if row["reason"] == "absolute error overflow" + else row["max_abs"] if row["max_abs"] is not None else -1), + default=None) + return dict(gate="w3_layer_diff", status="FAIL" if not passing else ("PARTIAL" if partial else "PASS"), + passed=passing and not partial, selected_passed=passing, partial=partial, + coverage=dict(total_checkpoints=len(entries), compared_checkpoints=len(names), + complete=not partial, selected_names=names, + evidence_validated_checkpoints=len(entries)), + atol=float(atol), first_divergence=comparison["first_divergence"], + worst_element=worst, checkpoints=comparison["checkpoints"], + claim="Comparison covers recorded checkpoints only; no whole-model or backend acceptance is implied.") + + +def _file_evidence(path): + path = Path(path) + digest = hashlib.sha256() + with path.open("rb") as stream: + for chunk in iter(lambda: stream.read(1024**2), b""): + digest.update(chunk) + return dict(path=str(path.resolve()), size_bytes=path.stat().st_size, sha256=digest.hexdigest()) + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--actual", required=True, type=Path) + parser.add_argument("--reference", required=True, type=Path) + parser.add_argument("--schema", required=True, type=Path) + parser.add_argument("--reference-schema", type=Path) + parser.add_argument("--out", required=True, type=Path) + parser.add_argument("--layer", type=int, action="append", default=[]) + parser.add_argument("--checkpoint", action="append", default=[]) + parser.add_argument("--atol", type=float, default=1e-5) + parser.add_argument("--max-bytes", type=int, default=DEFAULT_MAX_BYTES) + args = parser.parse_args(argv) + report = dict(gate="w3_layer_diff", status="FAIL", passed=False) + try: + # Existing evidence is never overwritten, including on report errors. + args.out.mkdir(parents=True, exist_ok=False) + except OSError as exc: + print(f"FAIL: cannot create fresh output directory: {exc}", file=sys.stderr) + return 1 + try: + report.update(source_evidence()) + report["artifacts"] = {name: _file_evidence(path) for name, path in + (("actual", args.actual), ("reference", args.reference), + ("schema", args.schema), ("reference_schema", args.reference_schema)) + if path is not None} + report.update(compare_evidence(args.actual, args.reference, args.schema, + reference_schema_path=args.reference_schema, + layers=args.layer, checkpoints=args.checkpoint, + atol=args.atol, max_bytes=args.max_bytes)) + except Exception as exc: + report.update(status="FAIL", passed=False, error=f"{type(exc).__name__}: {exc}", + first_divergence=getattr(exc, "checkpoint", None), + worst_element=getattr(exc, "details", None)) + try: + write_reports(args.out, report) + except Exception as exc: + print(f"FAIL: report publication failed: {type(exc).__name__}: {exc}", file=sys.stderr) + return 1 + print(f"{report['status']}: {args.out / 'report.json'}") + return 0 if report["passed"] else (2 if report["status"] == "PARTIAL" else 1) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/probes/w3_qwen3_full/EVIDENCE.md b/probes/w3_qwen3_full/EVIDENCE.md new file mode 100644 index 0000000..83e78aa --- /dev/null +++ b/probes/w3_qwen3_full/EVIDENCE.md @@ -0,0 +1,58 @@ +# 完整 W3 结果的离线复核 + +复现统一采用 **Ubuntu 24.04 x86_64、Bash、Python 3.12**。先完成 [Linux 环境与资产准备](../../docs/llm-deploy-v1.0/LINUX_REPRODUCTION.md),再在同一个 Bash 会话、仓库根目录执行命令;该指南设置 `SCRATCHV_PYTHON`、`SCRATCHV_CC` 和 `SCRATCHV_QEMU`。 + +`scripts/verify_w3_evidence.py` 对**已有七案例原始产物**重新计算误差,不执行模型、不下载资产。 +它复用完整模型 runner 的 `validate_worker`、`compare_case`、`invariants`,以及安全 NPZ 读取器。 + +在仓库根目录,使用已有 Python 3.12 和项目锁定依赖运行: + +```bash +set -euo pipefail +"$SCRATCHV_PYTHON" -B scripts/verify_w3_evidence.py \ + --evidence-dir output/downloaded-w3-run/raw/w3-full \ + --output-dir output/w3-evidence-audit-01 +``` + +输出目录必须不存在;不会覆盖被审核的原报告。 +输出为 `report.json`、`report.md`、`report.html`,退出码 `0` 表示复核通过,`1` 表示证据无效或不足。 +非法参数、已有输出目录等启动错误同样返回非零。 + +## 所需内容与检查 + +需要 full runner 的完整输出目录:顶层报告,以及每个案例的 `inputs.npz`、两个后端的报告、 +普通/诊断 `logits.npy`、`diagnostic_logits.npy`、`checkpoints.npz` 和 `checkpoint_schema.json`。 +仅下载 reports artifact 不足以复核,必须同时保留 raw artifact 并保持目录结构。CI raw artifact 内含 `w3-full/` 前缀:指定 `gh run download ... -n w3-full-numeric-raw --dir output/downloaded-w3-run/raw` 后,证据根是 `output/downloaded-w3-run/raw/w3-full`。完整下载和报告 SHA256 获取方法见 [Linux CI 产物说明](../../docs/llm-deploy-v1.0/W3/Linux-CI与Nightly.md#产物与失败定位)。 + +- 七组案例完整、顺序固定,输入 dtype/shape/有限值以及每个数值均符合确定性输入生成规则。 +- 模型文件描述符与固定 manifest 一致。可选 `--model-dir PATH` 进一步核对本地模型文件哈希,仍不执行模型。 +- worker 报告哈希、产物大小/哈希、checkpoint schema、执行步数、耗时和 ORT 配置符合 runner 的验证规则。 +- 每组普通/诊断 logits 一致,IR/ORT 对全部 256 个位置重新比较,严格 `max_abs < 1e-4`,`rtol=0`。 +- 30 个检查点的 dtype、shape、有限值和误差重新计算;其误差用于诊断定位,不额外替代或改变 logits 验收门槛。 +- 两个后端各自的未来 token 隔离、padding 隔离,共四项 invariant 重新计算。 +- 重新计算结果必须与原报告一致;仅有 `passed: true` 不能通过。 +- 解释器数值模式和策略一致,并受当前验证器支持。当前已提交 reference 契约为 `numpy-fp32-reference-v4`,所有 IR worker 的 `cpu_strategy` 必须一致,且完整 profile 的字段和值须与 `avx2-fma3` 或 `avx512` 的规范契约匹配;不是仅检查一个策略名字。 + +v4 复核读取生产报告保存的 `cpu_strategy`,不从审计机器 CPU 或 `SCRATCHV_FP32_REFERENCE_CPU` 环境变量推断生产策略。完整 profile 校验包括 `matmul_k_block`、`matmul_row_tile`、`matmul_row_tail`、`matmul_column_tail`、`matmul_layout` 和 `matmul_vector_policy`,不能缺字段或仅靠版本名字通过。因此可以在另一种 CPU 上核对已保存的数组,无需控制或执行本机 ORT。这里的跨机器数组核对不证明该机器实际执行模型也能通过。 + +当前 v4 验证器不接受历史 v2/v3 profile,须使用报告对应源码版本中的验证器,或重新生成 v4 完整证据。不能手动改旧报告的 profile、CPU 策略或源码哈希。`SCRATCHV_FP32_REFERENCE_CPU=auto|avx2-fma3|avx512` 是实际 reference 执行的选项,默认 `auto`;它不用于把旧证据转换到新契约,也不控制 ORT dispatch。 + +## 结论边界 + +`audit:w3-full-saved-evidence` 的 PASS 只说明**保存的文件和数值结果自洽**。 +它不证明生产者身份,也不代表第二人独立执行模型、当前源码已运行、Linux 通过、Nightly 通过或团队验收。 +本工具无法通过保存文件独立证明报告中的历史耗时与执行声明确实发生。 + +默认 SHA-256 检查发现文件与报告不一致;若报告和数据同时被替换,它不能充当数字签名。 +可从可信渠道预先独立保留原始顶层报告哈希,然后传入: + +```bash +set -euo pipefail +"$SCRATCHV_PYTHON" -B scripts/verify_w3_evidence.py --evidence-dir output/downloaded-w3-run/raw/w3-full --output-dir output/w3-evidence-audit-02 --expected-report-sha256 ORIGINAL_REPORT_SHA256 +``` + +`source_comparison` 明确列出原报告中生产源码与当前审计源码的更改、新增、缺失。 +源码不一致时,只要保存的数值契约仍受当前验证器支持,就可复核旧数组;但**不能将旧报告升级成当前源码执行的证明**。不受支持的旧 profile 仍会失败。 +报告同时保存 producer 环境、审计器环境和版本化数值策略。生产环境中的 `environment.numeric_runtime.blas_environment.OPENBLAS_CORETYPE` 是环境变量原值;`numpy_openblas` 的 `status/libraries` 及可用库项的 `core_name/config` 是可选运行诊断。审计不得以本机 CPU SIMD 或 BLAS 内核替填生产环境,也不能从该环境变量推断生产库已采用某个内核。诊断不可用或历史字段缺失不豁免任何数组数值门槛。 + +当前已提交 v4 的[第四轮 Linux 执行](https://github.com/yuki-328/ScratchV/actions/runs/37299235103)(`1c49e8acbcd9174491b2d0c2a5a0072178ae5f9a`)已在七组完整执行、preparation 与手动汇总通过;完整 raw 数组已下载验真、离线 audit 通过。实际执行 PASS 和保存证据 audit PASS 分别记录,实际 audit 报告及 SHA256、原样保留的源码换行差异说明见 [第四轮实测记录](../../docs/llm-deploy-v1.0/W3/Linux-CI与Nightly.md)。本说明本身不替代该原始记录。 diff --git a/probes/w3_qwen3_full/LOCALIZE.md b/probes/w3_qwen3_full/LOCALIZE.md new file mode 100644 index 0000000..0af0093 --- /dev/null +++ b/probes/w3_qwen3_full/LOCALIZE.md @@ -0,0 +1,46 @@ +# 完整模型的局部数值定位 + +复现统一采用 **Ubuntu 24.04 x86_64、Bash、Python 3.12**。先完成 [Linux 环境与资产准备](../../docs/llm-deploy-v1.0/LINUX_REPRODUCTION.md),再在同一个 Bash 会话、仓库根目录执行命令;该指南设置 `SCRATCHV_PYTHON`、`SCRATCHV_CC` 和 `SCRATCHV_QEMU`。 + +`localize.py` 将同一个完整 ORT 层输入送入原图切片的 ORT 和 ScratchV IR,帮助区分当前层局部误差与上游累计误差。它保留完整 FP32 运算、真实权重和所有 padding query。不会接受 W3 完整数值 gate,也不会修改固定模型或门槛。 + +先用本目录的 `run.py` 生成完整模型证据,再运行: + +```bash +set -euo pipefail +W3_MODEL=/absolute/path/to/qwen3-0.6b-onnx +export OMP_NUM_THREADS='1' +export OPENBLAS_NUM_THREADS='1' +export MKL_NUM_THREADS='1' +export PYTHONIOENCODING='utf-8' +export SCRATCHV_FP32_REFERENCE_CPU='auto' +"$SCRATCHV_PYTHON" -B -X utf8 probes/w3_qwen3_full/localize.py \ + --model-dir "$W3_MODEL" \ + --case-dir output/full-run/short_17 \ + --output-dir output/full-localize-new \ + --layer 0 --layer 21 +``` + +依赖使用完整探测的固定 CPU 环境。输出目录必须不存在;不指定 `--layer` 时仅诊断第一层,层编号从 0 到 27,可重复指定不同层。原始 case 必须包含 `inputs.npz` 和 `ort/`、`ir/` 下完整 worker 的报告、logits、检查点及 schema。工具重新核验固定资产 SHA256、输入/mask、报告记录的数组哈希、普通/诊断 logits 一致性和固定图的 30 检查点结构。历史 IR/ORT 的源码指纹必须彼此一致,但不要求与新增诊断工具后的源码集相同;旧报告和当前诊断的源码指纹分别保留。 + +诊断沿用原 IR worker 记录的 `fp32_mode`,并核对完整 reference profile;无模式字段的旧报告按 `native` 解释。当前已提交 reference 版本为 `numpy-fp32-reference-v4`,`SCRATCHV_FP32_REFERENCE_CPU` 可设为 `auto`(默认)、`avx2-fma3` 或 `avx512`。本机自动解析出的策略必须与原报告 `ir.fp32_profile.cpu_strategy` 一致;若跨 CPU 诊断,先读取原报告,在启动 Python 前显式设置同一策略。否则工具会拒绝不一致的 profile,不会静默换用当前机器策略。 + +复现时另外保留原报告与本机 `environment.numeric_runtime` 中的 `blas_environment.OPENBLAS_CORETYPE`、`numpy_openblas.status/libraries` 及可用库项的 `core_name/config`。CPU SIMD 支持、参考策略和实际 BLAS 内核不同;环境变量原值不证明库已采用该内核。可选诊断不可用时保留原说明,不把它作为跳过数值检查的理由。 + +这个环境变量只改变本次 IR reference 算术策略,不控制本机 ORT dispatch;局部诊断会实际执行本机 ORT/IR,因此不能把“重放保存的 IR 策略”当作两后端必然一致的保证。这与只读取原始数组的 [离线 audit](EVIDENCE.md) 不同,后者无需在当前机器选择或执行生产 CPU 策略。 + +未知、互相矛盾或不受当前实现支持的 profile 会失败,不能自动换成最新策略。历史 v2/v3 的局部诊断应使用对应的代码版本,或重新生成 v4 完整证据;原报告的模式、profile 和源码哈希不得改写。 + +v4 的 profile 还包含 MatMul 分块、尾行/单列处理与布局字段;即使 `cpu_strategy` 相同,也必须与保存证据的完整契约一致。当前 v4 实现不能直接用 v3 证据运行并宣称是在复现原算术策略。 + +所选层按固定图的依赖关系切片;计算节点与权重不重写,以保存的前一层 ORT hidden state 作为共同输入。每层记录 13 个检查点,包含 Q/K RMSNorm、RoPE、Attention probabilities/context、残差及 MLP。第一次归约的 `Pow/ReduceMean/epsilon/rsqrt` 在选择第 0 层时额外记录。 + +选择第 0 层时还自动运行五项相同输入原语实验:input RMSNorm 的 ReduceMean、Q/K 投影 MatMul、Attention QK MatMul、Softmax。FP64 仅用于估计原语的数学误差,不参与 IR/ORT 执行、不替换任何输出或正式参考。Softmax 另记录不可见 key 概率和行和偏差。 + +`report.json`、`report.md`、`report.html` 保存误差、最差索引、有效/padding 分项、耗时、进程峰值 RSS 和源码/资产指纹。`layer-XX-values.npz` 保存原始检查点,`op-*-values.npz` 保存原语输入和双方输出,`.onnx` 是本次使用的实际 FP32 局部图。大文件仅在本地输出目录生成。 + +退出码 0 与 `status=COMPLETE` 仅表示诊断完成;局部误差超过 `1e-4` 会照实记录为比较不通过,不伪装成工具执行失败。资产、输入、schema、执行或有限值检查失败时退出 1。无论诊断完成与否,`numeric_gate_passed=false`、`w3_exit_accepted=false`,原完整模型失败报告保持原样。 + +不能单凭一个大值中间张量超 `1e-4` 判定语义错误。例如 6000 量级 FP32 数值的一个 ULP 已可能超过该阈值。应同时查看同输入对照、原语误差、值域和原始 mask 概率,最终仍以完整 runner 的全部 logits `<1e-4` 验收。 + +当前已提交 v4 的[第四轮 Linux 完整运行](https://github.com/yuki-328/ScratchV/actions/runs/37299235103)(`1c49e8acbcd9174491b2d0c2a5a0072178ae5f9a`)已七组完整执行通过;raw 数组 audit 也已通过。该结论来自完整 runner,局部诊断和本说明不替代完整 gate 或团队验收。 diff --git a/probes/w3_qwen3_full/README.md b/probes/w3_qwen3_full/README.md new file mode 100644 index 0000000..5b51e48 --- /dev/null +++ b/probes/w3_qwen3_full/README.md @@ -0,0 +1,87 @@ +# 完整 Qwen3 IR 数值验证 + +复现统一采用 **Ubuntu 24.04 x86_64、Bash、Python 3.12**。先完成 [Linux 环境与资产准备](../../docs/llm-deploy-v1.0/LINUX_REPRODUCTION.md),再在同一个 Bash 会话、仓库根目录执行命令;该指南设置 `SCRATCHV_PYTHON`、`SCRATCHV_CC` 和 `SCRATCHV_QEMU`。 + +该入口实际执行固定发布的 28 层 Qwen3-0.6B FP32 ONNX,比较 ScratchV IR 与 ORT 的完整 logits,包括右侧 padding query。它使用优化前 IR(none),不会把资产检查、两层模型或部分输入通过视为完整模型数值验收。 + +## 验证范围 + +默认运行七组输入:两组满长随机输入、单 token、17/255 个有效 token,以及改变未来 token、改变 padding token 的输入。固定形状为 INT64 input_ids `[1,256]` 和 FP32 加性 mask `[1,1,256,256]`,输出为 FP32 `[1,256,151936]`。输入是可复现的原始 embedding ID,用于数值测试,不作为 tokenizer 或语言生成质量证据。 + +每组输入分别启动 ORT 和 IR 独立进程;每个后端运行普通与诊断两次,要求两份完整 logits 逐元素完全一致。跨后端对普通、诊断 logits 分别执行严格的 `max_abs < 1e-4`、`rtol=0`,同时报告有效位置和 padding 位置的误差。因果性和 padding 隔离另有四项跨输入检查。 + +诊断保存 embedding、28 层 residual 输出、final norm 共 30 个检查点,按固定图的实际连接定位。中间张量的 `1e-4` 标记用于定位误差,正式数值门槛作用于完整 logits;某些中间激活很大,单个 FP32 ULP 已可能大于 `1e-4`。报告保留这些标记,不将它们隐藏,也不把中间误差与最终输出误差混为一谈。 + +## 运行 + +在 W3 工作树根目录,使用 [固定依赖](../../requirements/qwen3-small-probe.txt)。模型目录必须包含已发布且匹配 manifest 的完整 ONNX 与三份权重分片;入口不下载、不修改模型。 + +完整入口默认使用 `--fp32-mode reference`,当前已提交策略为 `numpy-fp32-reference-v4`;请在复现命令中显式写出模式。解释器 API 默认仍为 `native`,追加 `--fp32-mode native` 才复现原 NumPy 策略。两种模式、不同版本和不同 CPU 策略的 profile 必须分别记录与验收,不能将一份报告的通过结论用于另一种模式。该参考模式不是跨 CPU、BLAS 或 ORT 版本逐元素一致的保证。 + +环境变量 `SCRATCHV_FP32_REFERENCE_CPU` 只作用于 `reference` IR 的算术策略,允许以下取值: + +| 值 | 行为 | +|---|---| +| `auto`(未设置时的默认值) | 从 NumPy 报告的可用 CPU 能力推断;具有 AVX2/FMA3 时,根据 AVX512F 是否可用选择下列一种策略 | +| `avx2-fma3` | 显式采用对应的参考算术顺序,profile 记录该值 | +| `avx512` | 显式采用对应的参考算术顺序,profile 记录该值 | + +显式设置不会控制 ORT 的 CPU dispatch,也不是启用或禁用真实硬件指令的开关;两条参考路径仍由 NumPy 实现。跨机器重放保存的算术策略时,可以在启动 Python 前显式指定报告中的 `cpu_strategy`,但必须重新比较本机 ORT 输出。未知变量取值会报错;`auto` 模式无法识别所需 CPU 能力,或发现 `NPY_DISABLE_CPU_FEATURES` 遮蔽相关 AVX/FMA 能力时也会明确报错,要求显式选择并记录策略。 + +v4 把矩阵乘的分块与内存布局纳入版本化 profile;数值策略名字与 CPU 策略相同仍不足以替代完整字段核对: + +| profile 字段 | `avx2-fma3` | `avx512` | +|---|---|---| +| `matmul_k_block` | 128 | 128;K≤128 时保留原 NumPy 路径 | +| `matmul_row_tile` | 2 | `null`,保留原宽矩阵路径 | +| `matmul_row_tail` | `zero-pad-to-two-then-trim` | `native` | +| `matmul_column_tail` | `zero-pad-single-column-to-two-then-trim` | `native` | +| `matmul_layout` | `contiguous-matrix-operands-per-k-block` | `native-strides` | +| `matmul_vector_policy` | `native-numpy-vector-promotion` | `native-numpy-vector-promotion` | + +AVX2 矩阵路径按两个输出行计算,尾行和单列临时补零后裁回原输出形状,每个 K 分块的两个矩阵操作数使用 C 连续切片;向量提升继续遵循 NumPy。以上是参考计算契约,不改模型图、权重、输出形状或误差门槛,也不是性能优化验收。本节描述已提交的 v4 实现;第四轮 Linux 已取得同提交七组完整运行 PASS,普通/诊断完整 logits 及四项不变量最大误差均为 0;这是 v4 的新实测证据,历史 v3 或单例结果不能替代它。 + +~~~bash +set -euo pipefail +W3_MODEL=/absolute/path/to/qwen3-0.6b-onnx +export SCRATCHV_FP32_REFERENCE_CPU=auto +"$SCRATCHV_PYTHON" -B -X utf8 probes/w3_qwen3_full/run.py \ + --model-dir "$W3_MODEL" --output-dir output/w3-full-new \ + --fp32-mode reference --worker-timeout 1800 --max-worker-memory-gib 10 +~~~ + +需要复现单个失败时追加 `--case short_17`。成功但仅覆盖部分 case 返回 PARTIAL/2,完整七组和不变量全部通过才返回 PASS/0;实际执行失败或数值超标返回 FAIL/1。中断返回 130。输出目录必须是新目录。 + +每个 worker 的超时和采样内存上限覆盖本次拥有的整棵进程树。Linux 使用独立进程组,采样整个进程树的 RSS,并在超时、超限或中断时清理本次拥有的进程。采样间隔为 0.2 秒,共享内存页可能按进程重复计数。它是超限终止措施,不是操作系统的硬内存配额,短暂峰值仍可能超过界限。失败保留阶段进度、日志、已完成产物和报告;中断、超时及超限会清理本次拥有的进程树。 + +## 证据与时长 + +统一 `report.json`、`report.md` 和 `report.html` 包含每组输入、每个 worker、完整比较和四项不变量。`cases[].comparison.logits` 是最终精度验收数据,`checkpoints` 及 `first_divergence` 定位最早超过诊断阈值的检查点。 + +Summary 将普通图与诊断图的最终 logits 分开,明确实际 `native/reference` 模式,并报告最大绝对误差、余弦相似度和相对 L2 误差;逐用例索引默认折叠,完整检查点和机器证据仅保存在 `report.json`,不重复嵌入摘要。后两项由实际输出和参考数组分块计算,使用 FP64 累积,不改变原有通过门槛。不能把参考兼容模式的通过结果归给默认 NumPy 模式。旧报告不补写或重新命名为本轮运行;离线重算保存数组也不代表重新执行了模型。 + +PPL(真实文本预测的困惑度)和 zero-shot(无示例任务评分)本轮未评测。这里的随机 token 输入用于数值对齐,不能作为语言质量语料;加入这类评测前需要固定真实文本、Tokenizer、L=256 的分窗及计分方式。 + +每组目录内有 `inputs.npz`、ORT/IR 日志,以及两个后端各自的完整 logits、诊断 logits、检查点 NPZ、schema、报告和阶段进度。报告包含固定模型、输入、输出及源码哈希;统一入口重新校验原始数组与报告一致性。 + +`cases[].workers[].stages_seconds` 可查看模型解析、普通执行、诊断执行和写盘时间;统一 `elapsed_seconds` 包括所有 worker 及文件核对。`process_peak_rss_bytes` 为该 worker 的全生命周期峰值,`resource_monitor` 是父进程采样观测。两者测量口径不同。 + +每个 IR worker 的 `ir.fp32_profile` 记录实际数值策略,包括 `name=numpy-fp32-reference-v4` 与已解析的 `cpu_strategy`;`auto` 是选择方式,不是保存后的策略名称。`environment.numeric_runtime` 补充 CPU 架构、NumPy 构建/运行诊断和线程环境。`blas_environment.OPENBLAS_CORETYPE` 保存环境变量原值,未设置时为 `null`,不是实际生效内核的证明;`numpy_openblas` 保存可选诊断的总体 `status`、`libraries`,可用库项包含 `path`、`status`、`core_name`、`config`,不可用时保存 `reason` 或 `error`。该诊断只读取已装载的 NumPy 私有 OpenBLAS 库,系统 BLAS/MKL 等可明确为不可用。 + +CPU SIMD 能力、IR 的 `cpu_strategy` 与实际 BLAS 内核须分别看待;上述字段也不证明 ORT 使用了哪一种内部内核。可选诊断不可用或旧证据缺少该字段,不能掩盖数值失败,所有门槛照常执行。独立复现应一并保留这些字段、环境变量原值、依赖版本与源码哈希。 + +离线 audit 按生产报告中保存的 CPU 策略核对整个 profile,不从审计机器 CPU 推断生产策略,也不会重新运行模型。当前 v4 工具严格验证 v4 契约;历史 v2/v3 原始证据应在其对应源码版本下核对,不能把报告字段改为 v4 或直接沿用旧版本的 PASS。每个提交和策略的完整七组以对应运行结果验收。 + +完整执行使用只读映射权重、显式借用 initializer、最后一次使用后释放中间 binding;默认解释器 API 仍复制和保留原有数据。只读映射文件在调用期间必须保持不变。解释器的 `peak_numpy_storage_bytes` 计入当时已绑定的借用 buffer,排除已解除绑定但仍由调用方持有的数组、内核临时缓冲和观察回调的拷贝;请同时参考实际进程 RSS。observer 是受信任的同步回调,只读 view 用于防止误写,不提供隔离沙箱。 + +需要继续定位失败时,使用 [同输入逐层与原语诊断](LOCALIZE.md)。该工具读取已保存的完整执行证据,保存层内检查点和局部 FP64 参考误差;诊断完成不代表完整数值验收通过。 + +## CI 与阶段结论 + +[完整模型手动 CI](../../.github/workflows/w3-full-numeric.yml) 已随 PR #96 发布。2026-10-05,作者在提交 `1c49e8acbcd9174491b2d0c2a5a0072178ae5f9a` 的[第四轮 Linux 手动编排](https://github.com/yuki-328/ScratchV/actions/runs/37299235103)中通过 preparation、完整七组 full 和汇总;v4/avx2-fma3、OpenBLAS Haswell 环境下普通/诊断 logits、30 检查点与四不变量最大误差均为 0,完整 runner 用时 `1477.314971525 s`(约 24.6 分钟)。reports 已下载并核对,完整 raw 已下载验真、离线 audit 通过。前三轮失败及旧 v2/v3 记录保留在 [Linux 实测记录](../../docs/llm-deploy-v1.0/W3/Linux-CI与Nightly.md)。实际完整数值失败仍使 job 失败;没有自动忽略 padding、调大阈值、启用低精度或回退到小模型。 + +该流程同时支持 `workflow_call`,供显式启用的 [Nightly 编排](../../.github/workflows/w3-nightly.yml) 复用;两套自动检查均成功才能通过汇总。Linux 运行、定时启用和资源限制说明见 [Linux CI 与 Nightly](../../docs/llm-deploy-v1.0/W3/Linux-CI与Nightly.md)。收到完整保存目录后,可按 [离线证据复核](EVIDENCE.md) 重新检查数组;这不执行模型,也不替代独立复现。 + +CI 将报告、哈希和 schema 放入 `w3-full-numeric-reports`(保留 30 天),完整原始输入、logits 和检查点另放入 `w3-full-numeric-raw`(保留 3 天,也包含报告)。需要独立复核原始数值时,应在过期前下载 raw 产物;其解压目录中保留 `w3-full/` 前缀,所以 `--evidence-dir` 应为 `下载目录/w3-full`。下载命令、报告 SHA256 获取与完整 audit 示例见 [产物与失败定位](../../docs/llm-deploy-v1.0/W3/Linux-CI与Nightly.md#产物与失败定位)。仅有摘要和哈希不足以重新比较数组。失败或中断运行的产物可能不完整,须先核对报告状态与覆盖范围。 + +即使本地数值 PASS,`w3_exit_accepted` 仍为 false:W1 人工前置验收、团队确认、E1/E2/E5 独立复现及 Nightly 状态由团队单独核对。[Linux 完整执行记录](../../docs/llm-deploy-v1.0/W3/W3-完整模型执行与数值诊断报告.md) 保留固定提交、模式和数据,不代表之后版本已经重跑。当前结论以本次完整运行生成的 `report.json`、实际 profile 和源码指纹为准。 diff --git a/probes/w3_qwen3_full/__init__.py b/probes/w3_qwen3_full/__init__.py new file mode 100644 index 0000000..67f94da --- /dev/null +++ b/probes/w3_qwen3_full/__init__.py @@ -0,0 +1 @@ +"""Bounded full Qwen3 numerical verification workers and gate.""" diff --git a/probes/w3_qwen3_full/cases.py b/probes/w3_qwen3_full/cases.py new file mode 100644 index 0000000..aa5b542 --- /dev/null +++ b/probes/w3_qwen3_full/cases.py @@ -0,0 +1,29 @@ +"""Fixed numerical input suite for the released FP32 L256 model.""" +from __future__ import annotations + +import numpy as np + +from scratchv.runtime.llm_inputs import prepare_inputs, QWEN3_VOCAB_SIZE + +CASE_NAMES = ("full_seed_0", "full_seed_42", "one_token", "short_17", "short_255", + "changed_future", "changed_padding") +SEQ = 256 + + +def input_cases(): + """Raw embedding IDs; this suite makes no tokenizer or language-quality claim.""" + result = [] + for name, seed, length in ((CASE_NAMES[0], 0, 256), (CASE_NAMES[1], 42, 256), + (CASE_NAMES[2], 7, 1), (CASE_NAMES[3], 7, 17), + (CASE_NAMES[4], 7, 255)): + tokens = np.random.default_rng(seed).integers(1, QWEN3_VOCAB_SIZE, length, dtype=np.int64) + feed = prepare_inputs(tokens, pad_token_id=0).as_feed() + result.append((name, length, feed)) + future = {key: value.copy() for key, value in result[0][2].items()} + future["input_ids"][:, 64:] = future["input_ids"][:, 64:] % (QWEN3_VOCAB_SIZE - 1) + 1 + result.append((CASE_NAMES[5], 256, future)) + padding = {key: value.copy() for key, value in result[3][2].items()} + padding["input_ids"][:, 17:] = np.random.default_rng(99).integers( + 1, QWEN3_VOCAB_SIZE, (1, 239), dtype=np.int64) + result.append((CASE_NAMES[6], 17, padding)) + return result diff --git a/probes/w3_qwen3_full/comparison.py b/probes/w3_qwen3_full/comparison.py new file mode 100644 index 0000000..478b864 --- /dev/null +++ b/probes/w3_qwen3_full/comparison.py @@ -0,0 +1,62 @@ +"""Bounded-memory absolute-error comparisons, including padded query rows.""" +from __future__ import annotations + +import math +import numpy as np + +from scratchv.verification.numeric_metrics import numeric_metrics, undefined_numeric_metrics + +ATOL = 1e-4 + + +def compare_tensor(actual, reference, *, atol=ATOL, chunk_elements=262144): + if not math.isfinite(atol) or atol <= 0: + raise ValueError("atol must be finite and positive") + if type(chunk_elements) is not int or chunk_elements < 1: + raise ValueError("chunk_elements must be a positive integer") + row = {"passed": False, "atol": atol, "rtol": 0, + "shape": list(actual.shape), "reference_shape": list(reference.shape), + "dtype": str(actual.dtype), "reference_dtype": str(reference.dtype), + "elements": int(actual.size), "max_abs": None, "finite": False} + row.update(undefined_numeric_metrics("comparison is invalid")) + if actual.shape != reference.shape or actual.dtype != np.float32 or reference.dtype != np.float32: + row["reason"] = "shape or FP32 dtype mismatch" + row.update(undefined_numeric_metrics(row["reason"])) + return row + if not actual.size: + row["reason"] = "empty comparison is not numerical evidence" + row.update(undefined_numeric_metrics(row["reason"])) + return row + # nditer handles strided slices without a full-size ravel/copy allocation. + chunks = np.nditer([actual, reference], flags=["external_loop", "buffered"], + op_flags=[["readonly"], ["readonly"]], order="C", buffersize=chunk_elements) + maximum, offset, worst = -1.0, 0, None + for a, b in chunks: + finite = np.isfinite(a) & np.isfinite(b) + if not finite.all(): + index = offset + int(np.flatnonzero(~finite)[0]) + row.update(reason="nonfinite value", worst_index=list(np.unravel_index(index, actual.shape))) + row["worst_index"] = [int(x) for x in row["worst_index"]] + row.update(undefined_numeric_metrics(row["reason"])) + return row + difference = np.abs(a.astype(np.float64) - b.astype(np.float64)) + local = int(np.argmax(difference)) + if float(difference[local]) > maximum: + maximum = float(difference[local]) + worst = (offset + local, float(a[local]), float(b[local])) + offset += a.size + row.update(passed=maximum < atol, finite=True, max_abs=maximum, + worst_index=[int(x) for x in np.unravel_index(worst[0], actual.shape)], + actual_value=worst[1], reference_value=worst[2]) + row.update(numeric_metrics(actual, reference, block_elements=chunk_elements)) + return row + + +def compare_positions(actual, reference, valid_length, *, atol=ATOL): + if actual.ndim != 3 or reference.ndim != 3 or not 1 <= valid_length <= actual.shape[1]: + raise ValueError("expected [batch, sequence, width] and a valid query length") + total = compare_tensor(actual, reference, atol=atol) + total["valid_queries"] = compare_tensor(actual[:, :valid_length], reference[:, :valid_length], atol=atol) + total["padding_queries"] = (compare_tensor(actual[:, valid_length:], reference[:, valid_length:], atol=atol) + if valid_length < actual.shape[1] else None) + return total diff --git a/probes/w3_qwen3_full/localize.py b/probes/w3_qwen3_full/localize.py new file mode 100644 index 0000000..c4fd787 --- /dev/null +++ b/probes/w3_qwen3_full/localize.py @@ -0,0 +1,390 @@ +"""Localize full-Qwen3 numerical differences using identical layer inputs. + +A completed diagnosis is not a numerical gate: large intermediate activations +may differ by more than 1e-4 in a few FP32 ULPs. The full-model runner remains +the authority for full logits. This tool never changes execution precision. +""" +from __future__ import annotations + +import argparse +import copy +import gc +import json +from pathlib import Path +import sys +import time + +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +import numpy as np +import onnx + +from probes.w1_qwen3_export.run import verify_files +from probes.w3_common import (new_output_dir, process_peak_rss_bytes, sha256_file, + source_evidence, write_reports) +from probes.w3_layer_diff.run import load_arrays, load_schema +from probes.w3_qwen3_full.comparison import compare_tensor +from probes.w3_qwen3_full.run import validate_worker +from probes.w3_qwen3_full.worker import checkpoint_schema, load_inputs + + +def require(value, message): + if not value: + raise ValueError(message) + + +def selected_layers(values): + values = [0] if values is None else list(values) + require(bool(values) and all(type(v) is int and 0 <= v < 28 for v in values), + "Layers must be integer indices in [0, 27]") + require(len(values) == len(set(values)), "Duplicate layer selection") + return sorted(values) + + +def compare_queries(actual, reference, valid, axis): + require(actual.ndim > axis and actual.shape[axis] == 256 and 1 <= valid <= 256, + "Expected complete L256 query axis and valid length") + row = compare_tensor(actual, reference) + prefix = [slice(None)] * actual.ndim + suffix = prefix.copy() + prefix[axis], suffix[axis] = slice(0, valid), slice(valid, None) + row["valid_queries"] = compare_tensor(actual[tuple(prefix)], reference[tuple(prefix)]) + row["padding_queries"] = (compare_tensor(actual[tuple(suffix)], reference[tuple(suffix)]) + if valid < 256 else None) + row["query_axis"] = axis + return row + + +def probability_diagnostics(probabilities, mask): + require(probabilities.dtype == np.float32 and probabilities.ndim == 4 + and probabilities.shape[0] == 1 and probabilities.shape[2:] == (256, 256) + and np.isfinite(probabilities).all(), "Invalid attention probabilities") + require(mask.shape == (1, 1, 256, 256) and mask.dtype == np.float32, + "Invalid attention mask") + invisible = np.broadcast_to(mask < 0, probabilities.shape) + return {"masked_probability_max_abs": float(np.abs(probabilities[invisible]).max()), + "row_sum_max_abs_error": float(np.abs(probabilities.sum(-1, dtype=np.float64) - 1).max()), + "minimum_probability": float(probabilities.min())} + + +def fp64_error_oracle(op, feed, ir_value, ort_value): + """Double precision is only an independent local error estimate.""" + require(ir_value.dtype == ort_value.dtype == np.float32, + "Execution outputs must remain FP32") + if op == "MatMul": + truth = np.matmul(feed["a"].astype(np.float64), feed["b"].astype(np.float64)) + elif op == "ReduceMean": + truth = feed["x"].astype(np.float64).mean(-1, keepdims=True) + elif op == "Softmax": + x = feed["x"].astype(np.float64) + exp = np.exp(x - x.max(-1, keepdims=True)) + truth = exp / exp.sum(-1, keepdims=True) + else: + raise ValueError("Unsupported local diagnostic operation") + return {"scope": "diagnostic only; neither execution result nor acceptance output", + "ir_max_abs": float(np.abs(ir_value - truth).max()), + "ort_max_abs": float(np.abs(ort_value - truth).max())} + + +class LayerSlice: + """Extract exact pinned-graph ancestors; never reconstruct model arithmetic.""" + + def __init__(self, model, model_dir): + self.model, self.model_dir = model, Path(model_dir) + self.nodes = list(model.graph.node) + self.producers = {name: node for node in self.nodes for name in node.output} + self.info = {v.name: v for v in [*model.graph.value_info, *model.graph.input, *model.graph.output]} + self.initializers = {v.name: v for v in model.graph.initializer} + require(len(self.producers) == sum(len(n.output) for n in self.nodes), + "Duplicate graph values") + + @staticmethod + def one(nodes, description): + rows = list(nodes) + require(len(rows) == 1, f"Expected unique {description}") + return rows[0] + + def norm(self, weight): + node = self.one((n for n in self.nodes if n.op_type == "Mul" and weight in n.input), weight) + value = next(x for x in node.input if x != weight) + cast = self.producers[value] + require(cast.op_type == "Cast", "Unexpected normalization cast") + mul = self.producers[cast.input[0]] + require(mul.op_type == "Mul", "Unexpected normalization product") + cast = self.producers[mul.input[0]] + require(cast.op_type == "Cast", "Unexpected normalization input") + return cast.input[0], node.output[0] + + def projection(self, weight): + tr = self.one((n for n in self.nodes if n.op_type == "Transpose" and weight in n.input), weight) + mat = self.one((n for n in self.nodes if n.op_type == "MatMul" and tr.output[0] in n.input), weight + " MatMul") + return mat.input[0], mat.output[0] + + def describe(self, index): + p = f"model.model.layers.{index}." + hidden, input_norm = self.norm(p + "input_layernorm.weight") + residual, postnorm = self.norm(p + "post_attention_layernorm.weight") + _, qnorm = self.norm(p + "self_attn.q_norm.weight") + _, knorm = self.norm(p + "self_attn.k_norm.weight") + _, vproj = self.projection(p + "self_attn.v_proj.weight") + context, attn_output = self.projection(p + "self_attn.o_proj.weight") + _, mlp = self.projection(p + "mlp.down_proj.weight") + final = self.one((n for n in self.nodes if n.op_type == "Add" and mlp in n.input), "MLP residual").output[0] + begin, end = self.nodes.index(self.producers[input_norm]), self.nodes.index(self.producers[final]) + block = self.nodes[begin:end + 1] + softmax = self.one((n for n in block if n.op_type == "Softmax"), "layer Softmax").output[0] + probs = self.one((n for n in block if n.op_type == "Cast" and softmax in n.input), "Softmax cast").output[0] + + def rope(norm): + tr = self.one((n for n in block if n.op_type == "Transpose" and norm in n.input), "norm transpose") + mul = self.one((n for n in block if n.op_type == "Mul" and tr.output[0] in n.input), "RoPE cosine product") + return self.one((n for n in block if n.op_type == "Add" and mul.output[0] in n.input), "RoPE sum").output[0] + + return hidden, {"input_norm": input_norm, "q_norm": qnorm, "k_norm": knorm, "v_proj": vproj, + "rope_q": rope(qnorm), "rope_k": rope(knorm), "attn_probs": probs, + "attn_context": context, "attn_output": attn_output, "attn_residual": residual, + "post_attention_norm": postnorm, "mlp": mlp, "residual": final} + + def build(self, hidden, outputs): + stop, needed, queue, chosen = {hidden, "attention_mask"}, set(outputs), list(outputs), set() + while queue: + value = queue.pop() + if value in stop or value not in self.producers: + continue + node = self.producers[value] + if node.name in chosen: + continue + chosen.add(node.name) + for name in node.input: + if name and name not in needed: + needed.add(name) + queue.append(name) + nodes = [copy.deepcopy(n) for n in self.nodes if n.name in chosen] + weights = [onnx.numpy_helper.from_array(onnx.numpy_helper.to_array(t, base_dir=str(self.model_dir)), name) + for name, t in self.initializers.items() if name in needed and name not in stop] + for node in nodes: + for attr in node.attribute: + if attr.type == onnx.AttributeProto.TENSOR and attr.t.external_data: + attr.t.CopyFrom(onnx.numpy_helper.from_array( + onnx.numpy_helper.to_array(attr.t, base_dir=str(self.model_dir)), attr.t.name)) + graph = onnx.helper.make_graph(nodes, "pinned_layer_slice", + [copy.deepcopy(self.info[x]) for x in sorted(stop) if x in needed], + [copy.deepcopy(self.info[x]) for x in outputs], initializer=weights) + result = onnx.helper.make_model(graph, opset_imports=self.model.opset_import) + result.ir_version = self.model.ir_version + onnx.checker.check_model(result) + return result + + +def run_pair(model, feed, outputs, path, *, fp32_mode="native"): + """Actual ORT and production Parser/IR execution, with FP32 feeds unchanged.""" + import onnxruntime as ort + from scratchv.frontend.onnx_parser import ONNXParser + from scratchv.verification.ir_interpreter import IRInterpreter + + require(all(v.dtype == np.float32 and np.isfinite(v).all() for v in feed.values()), + "Local execution feeds must be finite FP32") + options = ort.SessionOptions() + options.intra_op_num_threads = options.inter_op_num_threads = 1 + options.enable_cpu_mem_arena = False + options.log_severity_level = 3 + options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_DISABLE_ALL + diagnostic_path = path.with_name(path.stem + "-diagnostic.onnx") + onnx.save_model(model, diagnostic_path) + session = ort.InferenceSession(model.SerializeToString(), sess_options=options, providers=["CPUExecutionProvider"]) + ort_values = dict(zip(outputs, session.run(list(outputs.values()), feed))) + ordinary = copy.deepcopy(model) + wanted = outputs[next(reversed(outputs))] + last = next(v for v in ordinary.graph.output if v.name == wanted) + last = copy.deepcopy(last) + del ordinary.graph.output[:] + ordinary.graph.output.append(last) + onnx.save_model(ordinary, path) + parser = ONNXParser() + program = parser.parse(str(path)) + for value in parser.initializers.values(): + value.setflags(write=False) + bindings = {parser._value_map[value].name: name for name, value in outputs.items()} + require(len(bindings) == len(outputs), "Aliased/ambiguous diagnostic checkpoints") + ir_values = {} + + def observe(name, value): + if name in bindings: + require(bindings[name] not in ir_values, "Checkpoint executed twice") + ir_values[bindings[name]] = value.copy() + + result = IRInterpreter(program).run(feed, initializers=parser.initializers, + memory_mode="last_use", copy_initializers=False, observer=observe, + fp32_mode=fp32_mode) + require(set(ir_values) == set(outputs), "Missing executed checkpoint") + require(all(v.dtype == np.float32 and np.isfinite(v).all() for v in [*ir_values.values(), *ort_values.values()]), + "Nonfinite/non-FP32 execution output") + return ir_values, ort_values, {"executed_steps": result.executed_steps, + "fp32_mode": fp32_mode, + "model_sha256": sha256_file(path), "model_file": path.name, + "ort_model_file": diagnostic_path.name, + "ort_model_sha256": sha256_file(diagnostic_path), + "checkpoint_onnx_names": dict(outputs)} + + +def primitive(op, feed, shape): + inputs, initializers, attrs = list(feed), [], {} + if op == "ReduceMean": + inputs += ["axes"] + initializers = [onnx.numpy_helper.from_array(np.array([-1], dtype=np.int64), "axes")] + attrs["keepdims"] = 1 + elif op == "Softmax": + attrs["axis"] = -1 + graph = onnx.helper.make_graph([onnx.helper.make_node(op, inputs, ["y"], **attrs)], op, + [onnx.helper.make_tensor_value_info(k, 1, v.shape) for k, v in feed.items()], + [onnx.helper.make_tensor_value_info("y", 1, shape)], initializer=initializers) + model = onnx.helper.make_model(graph, opset_imports=[onnx.helper.make_opsetid("", 18)]) + model.ir_version = 10 + return model + + +def local_primitives(extract, ort_values, mask, valid, out, *, fp32_mode="native"): + rows = [] + + def run(name, op, feed, shape, axis=1): + ir, ort_values, evidence = run_pair(primitive(op, feed, shape), feed, {"result": "y"}, + out / f"op-{name}.onnx", fp32_mode=fp32_mode) + a, b = ir["result"], ort_values["result"] + row = {"name": name, "op": op, "comparison": compare_queries(a, b, valid, axis), + "fp64_error_oracle": fp64_error_oracle(op, feed, a, b), **evidence} + if op == "Softmax": + row["probabilities"] = {"ir": probability_diagnostics(a, mask), "ort": probability_diagnostics(b, mask)} + rows.append(row) + values_path = out / f"op-{name}-values.npz" + np.savez(values_path, ir=a, ort=b, **feed) + row["values"] = {"file": values_path.name, "bytes": values_path.stat().st_size, + "sha256": sha256_file(values_path)} + return b + + run("input_norm_reduce_mean", "ReduceMean", {"x": ort_values["square"]}, [1, 256, 1]) + for key, width in (("q", 2048), ("k", 1024)): + name = f"model.model.layers.0.self_attn.{key}_proj.weight" + weight = onnx.numpy_helper.to_array(extract.initializers[name], base_dir=str(extract.model_dir)).T.copy() + run(f"projection_{key}", "MatMul", {"a": ort_values["input_norm"], "b": weight}, [1, 256, width]) + q, k = ort_values["rope_q"], np.repeat(ort_values["rope_k"], 2, axis=1) + scores = run("attention_scores", "MatMul", {"a": q, "b": k.transpose(0, 1, 3, 2).copy()}, [1, 16, 256, 256], 2) + scores = scores * np.float32(128 ** -0.5) + mask + run("attention_softmax", "Softmax", {"x": scores}, [1, 16, 256, 256], 2) + return rows + + +def validate_reference_metadata(record, schema, expected_schema, files, input_hash): + require(record.get("gate") == "numeric:w3-full-worker" and record.get("backend") == "ort" + and record.get("passed") is True and record.get("status") == "PASS", "Expected executed ORT worker") + require(record.get("files") == files, "Reference model assets differ") + require(record.get("input", {}).get("sha256") == input_hash, "Reference inputs differ") + require(schema == expected_schema and record.get("checkpoints") == schema, + "Reference schema differs from fixed graph dataflow") + + +def validate_producer_pair(ort_record, ir_record): + require(bool(ort_record.get("source_sha256")) + and ort_record["source_sha256"] == ir_record.get("source_sha256"), + "Historical IR/ORT producer source fingerprints differ") + + +def producer_fp32_mode(ir_record): + details = ir_record.get("ir", {}) + mode = details.get("fp32_mode", "native") + require(mode in ("native", "reference"), "Unknown producer FP32 mode") + require(ir_record.get("fp32_mode", mode) == mode, "Producer FP32 modes disagree") + if mode == "reference": + from scratchv.verification.fp32_reference import profile + require(details.get("fp32_profile") == profile(), + "Unsupported historical FP32 reference profile") + else: + require("fp32_profile" not in details or details["fp32_profile"] == {"name": "numpy-native"}, + "Unsupported historical native FP32 profile") + return mode + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--model-dir", type=Path, required=True) + parser.add_argument("--case-dir", type=Path, required=True) + parser.add_argument("--output-dir", type=Path, required=True) + parser.add_argument("--layer", type=int, action="append", help="Repeat for selected indices 0..27; default 0") + args = parser.parse_args(argv) + try: + layers = selected_layers(args.layer) + except ValueError as exc: + parser.error(str(exc)) + out = new_output_dir(args.output_dir) + report = {"gate": "diagnostic:w3-full-localize", "status": "FAIL", "passed": False, + "diagnostic_complete": False, "numeric_gate_passed": False, "w3_exit_accepted": False, + "scope": "Selected exact layer slices and equal-input FP32 primitives; cannot accept the full-model gate", + "selected_layers": layers, "layers": [], "primitives": [], + "execution_dtype": "float32", "fp64_scope": "local error oracle only"} + start = time.perf_counter() + try: + report.update(source_evidence()) + files = verify_files(args.model_dir) + feed, valid = load_inputs(args.case_dir / "inputs.npz") + model = onnx.load(args.model_dir / "model.onnx", load_external_data=False) + expected = checkpoint_schema(model) + folder = args.case_dir / "ort" + schema = load_schema(folder / "checkpoint_schema.json") + record = json.loads((folder / "report.json").read_text(encoding="utf-8")) + ir_folder = args.case_dir / "ir" + ir_record = json.loads((ir_folder / "report.json").read_text(encoding="utf-8")) + validate_producer_pair(record, ir_record) + input_hash = sha256_file(args.case_dir / "inputs.npz") + validate_reference_metadata(record, schema, expected, files, input_hash) + # Historical source identity is retained rather than falsely claiming a + # new localizer file existed when that complete reference was generated. + validate_worker(folder, "ort", args.case_dir / "inputs.npz", record["source_sha256"], files) + validate_worker(ir_folder, "ir", args.case_dir / "inputs.npz", record["source_sha256"], files) + fp32_mode = producer_fp32_mode(ir_record) + report["fp32_mode"] = fp32_mode + states = load_arrays(folder / "checkpoints.npz", schema, max_bytes=64 * 1024**2) + report.update(files=files, valid_length=valid, input_sha256=input_hash, + reference_report_sha256=sha256_file(folder / "report.json"), + paired_ir_report_sha256=sha256_file(ir_folder / "report.json"), + reference_source_sha256=record["source_sha256"]) + extract = LayerSlice(model, args.model_dir) + for index in layers: + hidden, names = extract.describe(index) + require(hidden == expected[index]["onnx_name"] and names["residual"] == expected[index + 1]["onnx_name"], + "Extracted layer boundary differs from audited graph") + if index == 0: + names = {"square": "pow_1", "mean": "mean", "epsilon_add": "add", "rsqrt": "rsqrt", **names} + state = states["embedding" if index == 0 else f"layer_{index - 1}.output"] + pair_feed = {hidden: state, "attention_mask": feed["attention_mask"]} + graph = extract.build(hidden, list(names.values())) + ir, ort_values, evidence = run_pair(graph, pair_feed, names, out / f"layer-{index:02d}.onnx", + fp32_mode=fp32_mode) + comparisons = {name: compare_queries(ir[name], ort_values[name], valid, + 2 if name in ("rope_q", "rope_k", "attn_probs") else 1) for name in names} + report["layers"].append({"layer": index, "same_input": comparisons, **evidence, + "probabilities": {"ir": probability_diagnostics(ir["attn_probs"], feed["attention_mask"]), + "ort": probability_diagnostics(ort_values["attn_probs"], feed["attention_mask"])}}) + values_path = out / f"layer-{index:02d}-values.npz" + np.savez(values_path, **{"ir_" + k: v for k, v in ir.items()}, + **{"ort_" + k: v for k, v in ort_values.items()}) + report["layers"][-1]["values"] = {"file": values_path.name, "bytes": values_path.stat().st_size, + "sha256": sha256_file(values_path)} + if index == 0: + report["primitives"] = local_primitives(extract, ort_values, feed["attention_mask"], valid, + out, fp32_mode=fp32_mode) + del graph, ir, ort_values + gc.collect() + require(source_evidence()["source_sha256"] == report["source_sha256"], "Sources changed during diagnosis") + report.update(status="COMPLETE", passed=True, diagnostic_complete=True) + except KeyboardInterrupt: + report.update(interrupted=True, error="KeyboardInterrupt: local diagnosis cancelled") + except Exception as exc: + report["error"] = f"{type(exc).__name__}: {exc}" + report.update(elapsed_seconds=time.perf_counter() - start, process_peak_rss_bytes=process_peak_rss_bytes()) + write_reports(out, report) + return 130 if report.get("interrupted") else (0 if report["diagnostic_complete"] else 1) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/probes/w3_qwen3_full/resources.py b/probes/w3_qwen3_full/resources.py new file mode 100644 index 0000000..4821ad9 --- /dev/null +++ b/probes/w3_qwen3_full/resources.py @@ -0,0 +1,403 @@ +"""Observe and bound an owned full-model worker; no extra monitor dependency.""" +from __future__ import annotations + +import ctypes +import os +from pathlib import Path +import subprocess +import time + +from scripts.run_w2_acceptance import terminate_process_tree as _legacy_terminate + + +class _ResidentMemoryUnavailable(OSError): + """Linux can drop VmRSS while exit is still in progress, before zombie state.""" + + +_RSS_EXIT_GRACE_SECONDS = 0.2 +_RSS_EXIT_RECHECK_SECONDS = 0.01 + + +class _VerifiedProcessHandle: + """A handle already verified as a member of the owned Windows Job.""" + def __init__(self, handle): + self.handle = handle + + +class _WindowsJob: + """Own descendants before the suspended launcher executes any user code. + + Windows 8+ supports nesting under an existing sandbox Job. No breakaway + flags or UI limits are used. Assignment failure is fatal, not a fallback + to monitoring only the launcher. Handles are non-inheritable. + """ + def __init__(self): + from ctypes import wintypes as w + self.k = k = ctypes.WinDLL("kernel32", use_last_error=True) + prototypes = { + "CreateJobObjectW": ([ctypes.c_void_p, w.LPCWSTR], w.HANDLE), + "SetInformationJobObject": ([w.HANDLE, ctypes.c_int, ctypes.c_void_p, w.DWORD], w.BOOL), + "AssignProcessToJobObject": ([w.HANDLE, w.HANDLE], w.BOOL), + "QueryInformationJobObject": ([w.HANDLE, ctypes.c_int, ctypes.c_void_p, w.DWORD, ctypes.c_void_p], w.BOOL), + "IsProcessInJob": ([w.HANDLE, w.HANDLE, ctypes.POINTER(w.BOOL)], w.BOOL), + "TerminateJobObject": ([w.HANDLE, w.UINT], w.BOOL), + "OpenProcess": ([w.DWORD, w.BOOL, w.DWORD], w.HANDLE), + "GetExitCodeProcess": ([w.HANDLE, ctypes.POINTER(w.DWORD)], w.BOOL), + "CloseHandle": ([w.HANDLE], w.BOOL), + "CreateToolhelp32Snapshot": ([w.DWORD, w.DWORD], w.HANDLE), + "Thread32First": ([w.HANDLE, ctypes.c_void_p], w.BOOL), + "Thread32Next": ([w.HANDLE, ctypes.c_void_p], w.BOOL), + "OpenThread": ([w.DWORD, w.BOOL, w.DWORD], w.HANDLE), + "GetProcessIdOfThread": ([w.HANDLE], w.DWORD), + "ResumeThread": ([w.HANDLE], w.DWORD), + } + for name, (args, result) in prototypes.items(): + getattr(k, name).argtypes, getattr(k, name).restype = args, result + + class Basic(ctypes.Structure): + _fields_ = [("process_time", ctypes.c_longlong), ("job_time", ctypes.c_longlong), + ("flags", w.DWORD), ("min_working", ctypes.c_size_t), + ("max_working", ctypes.c_size_t), ("active", w.DWORD), + ("affinity", ctypes.c_size_t), ("priority", w.DWORD), ("scheduling", w.DWORD)] + class Limits(ctypes.Structure): + _fields_ = [("basic", Basic), ("io", ctypes.c_ulonglong * 6), + ("process_memory", ctypes.c_size_t), ("job_memory", ctypes.c_size_t), + ("peak_process", ctypes.c_size_t), ("peak_job", ctypes.c_size_t)] + self.handle = k.CreateJobObjectW(None, None) + if not self.handle: + raise ctypes.WinError(ctypes.get_last_error()) + limits = Limits() + limits.basic.flags = 0x2000 # JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE + if not k.SetInformationJobObject(self.handle, 9, ctypes.byref(limits), ctypes.sizeof(limits)): + error = ctypes.WinError(ctypes.get_last_error()) + self.close() + raise error + + def assign_and_resume(self, process): + from ctypes import wintypes as w + if not self.k.AssignProcessToJobObject(self.handle, int(process._handle)): + raise ctypes.WinError(ctypes.get_last_error()) + class ThreadEntry(ctypes.Structure): + _fields_ = [("size", w.DWORD), ("usage", w.DWORD), ("tid", w.DWORD), + ("pid", w.DWORD), ("priority", w.LONG), ("delta", w.LONG), ("flags", w.DWORD)] + snapshot = self.k.CreateToolhelp32Snapshot(4, 0) # TH32CS_SNAPTHREAD + if snapshot == ctypes.c_void_p(-1).value: + raise ctypes.WinError(ctypes.get_last_error()) + tids = [] + try: + entry = ThreadEntry() + entry.size = ctypes.sizeof(entry) + ok = self.k.Thread32First(snapshot, ctypes.byref(entry)) + while ok: + if entry.pid == process.pid: + tids.append(entry.tid) + ok = self.k.Thread32Next(snapshot, ctypes.byref(entry)) + if ctypes.get_last_error() != 18: # ERROR_NO_MORE_FILES + raise ctypes.WinError(ctypes.get_last_error()) + finally: + self.k.CloseHandle(snapshot) + if len(tids) != 1: + raise OSError("Suspended launcher must have exactly one primary thread") + thread = self.k.OpenThread(0x0002 | 0x0040, False, tids[0]) + if not thread: + raise ctypes.WinError(ctypes.get_last_error()) + try: + if self.k.GetProcessIdOfThread(thread) != process.pid: + raise OSError("Suspended primary thread ownership changed") + if self.k.ResumeThread(thread) != 1: + raise OSError("Suspended launcher did not have exactly one suspension") + finally: + self.k.CloseHandle(thread) + + def pids(self): + # QueryInformationJobObject includes descendant nested Jobs. Retry if + # more processes arrived between sizing and reading the PID list. + for capacity in (16, 64, 256, 1024, 4096, 16384): + buffer = ctypes.create_string_buffer(8 + capacity * ctypes.sizeof(ctypes.c_size_t)) + ok = self.k.QueryInformationJobObject(self.handle, 3, buffer, len(buffer), None) + assigned, count = (ctypes.c_ulong * 2).from_buffer(buffer) + if ok and count == assigned and count <= capacity: + return list((ctypes.c_size_t * count).from_buffer(buffer, 8)) + if not ok and ctypes.get_last_error() != 234: # ERROR_MORE_DATA + raise ctypes.WinError(ctypes.get_last_error()) + raise OSError("Owned worker Job exceeds bounded process enumeration") + + def sample(self): + from ctypes import wintypes as w + total = {"rss_bytes": 0, "private_commit_bytes": 0, "pids": []} + for pid in self.pids(): + handle = self.k.OpenProcess(0x1010, False, pid) + if not handle: + if pid not in self.pids(): + continue # Exited after the Job snapshot. + raise ctypes.WinError(ctypes.get_last_error()) + try: + member = w.BOOL() + if not self.k.IsProcessInJob(handle, self.handle, ctypes.byref(member)): + raise ctypes.WinError(ctypes.get_last_error()) + if not member.value: + # PID reuse cannot substitute an unrelated process: never + # inspect or terminate the new process's memory. + continue + try: + value = process_memory(_VerifiedProcessHandle(handle)) + except OSError: + code = w.DWORD() + if self.k.GetExitCodeProcess(handle, ctypes.byref(code)) and code.value != 259: + continue + raise + total["rss_bytes"] += value["rss_bytes"] + total["private_commit_bytes"] += value["private_commit_bytes"] or 0 + total["pids"].append(pid) + finally: + self.k.CloseHandle(handle) + return total + + def terminate(self): + if not self.k.TerminateJobObject(self.handle, 1): + raise ctypes.WinError(ctypes.get_last_error()) + deadline = time.monotonic() + 5 + while self.pids(): + if time.monotonic() > deadline: + raise OSError("Owned worker Job still has active descendants after termination") + time.sleep(0.01) + + def close(self): + if self.handle: + handle, self.handle = self.handle, None + if not self.k.CloseHandle(handle): + raise ctypes.WinError(ctypes.get_last_error()) + + +def spawn_owned(command, **kwargs): + """Create a worker tree whose ownership precedes all executed child code.""" + if os.name != "nt": + kwargs["start_new_session"] = True + process = subprocess.Popen(command, **kwargs) + process._scratchv_owned_group = True + return process + flags = kwargs.pop("creationflags", 0) + if flags & (0x01000000 | 0x00000004): + raise ValueError("Caller cannot request breakaway or manage suspended owned workers") + job, process = _WindowsJob(), None + try: + process = subprocess.Popen(command, creationflags=flags | 0x00000004, **kwargs) + process._scratchv_job = job + job.assign_and_resume(process) + return process + except BaseException as original: + # Before ResumeThread the launcher cannot have started descendants; + # after assignment closing the Job owns every descendant regardless. + try: + job.close() + except BaseException as cleanup: + original.add_note(f"Closing failed worker Job: {cleanup}") + try: + if process is not None: + if process.poll() is None: + process.kill() + process.wait(timeout=5) + except BaseException as cleanup: + original.add_note(f"Reaping failed launcher: {cleanup}") + raise + + +def terminate_process_tree(process): + job = getattr(process, "_scratchv_job", None) + if job is not None: + try: + job.terminate() + finally: + job.close() + process._scratchv_job = None + process.wait(timeout=5) + else: + _legacy_terminate(process) + + +class _PosixGroup: + """Linux session/group membership, including children outliving launcher.""" + def __init__(self, process): + self.process = process + self.rss_exit_rechecks = 0 + + @staticmethod + def identity(pid): + text = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8") + fields = text[text.rindex(")") + 2:].split() + return fields[0], int(fields[2]), int(fields[3]), int(fields[19]) + + def pids(self): + found = [] + for entry in Path("/proc").iterdir(): + if not entry.name.isdecimal(): + continue + try: + state, group, session, _ = self.identity(int(entry.name)) + except (FileNotFoundError, ProcessLookupError): + continue + if group == session == self.process.pid and state not in {"Z", "X", "x"}: + found.append(int(entry.name)) + return found + + def _member_memory(self, pid, before): + # exit_mm() can remove VmRSS before the process becomes a zombie or + # waitpid observes its exit. Only this specific missing-field case gets + # a short grace period; permission errors and malformed RSS fail closed. + deadline = None + while True: + try: + value = process_memory(pid) + except _ResidentMemoryUnavailable: + after = self.identity(pid) + if before[1:] != after[1:] or after[0] in {"Z", "X", "x"}: + return None + if deadline is None: + deadline = time.perf_counter() + _RSS_EXIT_GRACE_SECONDS + if time.perf_counter() >= deadline: + raise # Still the same live process and still unobservable. + self.rss_exit_rechecks += 1 + time.sleep(_RSS_EXIT_RECHECK_SECONDS) + continue + after = self.identity(pid) + if before[1:] != after[1:] or after[0] in {"Z", "X", "x"}: + return None + return value + + def sample(self): + total = {"rss_bytes": 0, "private_commit_bytes": None, "pids": []} + for pid in self.pids(): + try: + before = self.identity(pid) + if (before[1] != self.process.pid or before[2] != self.process.pid + or before[0] in {"Z", "X", "x"}): + continue + value = self._member_memory(pid, before) + if value is None: + continue # PID was reused/exited during sampling. + except (FileNotFoundError, ProcessLookupError): + continue + except OSError: + try: + after = self.identity(pid) + except (FileNotFoundError, ProcessLookupError): + continue + if after[0] in {"Z", "X", "x"}: + continue + raise + total["rss_bytes"] += value["rss_bytes"] + total["pids"].append(pid) + return total + + +def process_memory(pid): + """Current RSS, and private committed bytes on Windows (not virtual size).""" + if os.name == "nt": + from ctypes import wintypes + class Counters(ctypes.Structure): + _fields_ = [("cb", wintypes.DWORD), ("faults", wintypes.DWORD), + ("peak_rss", ctypes.c_size_t), ("rss", ctypes.c_size_t), + ("peak_paged", ctypes.c_size_t), ("paged", ctypes.c_size_t), + ("peak_nonpaged", ctypes.c_size_t), ("nonpaged", ctypes.c_size_t), + ("pagefile", ctypes.c_size_t), ("peak_pagefile", ctypes.c_size_t), + ("private", ctypes.c_size_t)] + kernel = ctypes.WinDLL("kernel32", use_last_error=True) + kernel.OpenProcess.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD] + kernel.OpenProcess.restype = wintypes.HANDLE + kernel.CloseHandle.argtypes = [wintypes.HANDLE] + psapi = ctypes.WinDLL("psapi", use_last_error=True) + psapi.GetProcessMemoryInfo.argtypes = [wintypes.HANDLE, ctypes.POINTER(Counters), wintypes.DWORD] + psapi.GetProcessMemoryInfo.restype = wintypes.BOOL + borrowed = isinstance(pid, _VerifiedProcessHandle) + handle = pid.handle if borrowed else kernel.OpenProcess(0x1010, False, pid) + if not handle: + raise OSError(ctypes.get_last_error(), "Cannot observe worker memory") + try: + values = Counters() + values.cb = ctypes.sizeof(values) + if not psapi.GetProcessMemoryInfo(handle, ctypes.byref(values), values.cb): + raise OSError(ctypes.get_last_error(), "Cannot read worker memory") + return {"rss_bytes": int(values.rss), "private_commit_bytes": int(values.private)} + finally: + if not borrowed: + kernel.CloseHandle(handle) + status = Path(f"/proc/{pid}/status").read_text(encoding="utf-8") + rows = dict(line.split(":", 1) for line in status.splitlines() if ":" in line) + if "VmRSS" not in rows: + raise _ResidentMemoryUnavailable(f"Cannot observe worker {pid} resident memory") + try: + resident = rows["VmRSS"].split() + if len(resident) != 2 or resident[1] != "kB": + raise ValueError("VmRSS must use the Linux kB unit") + rss = int(resident[0]) * 1024 + if rss < 0: + raise ValueError("VmRSS cannot be negative") + except ValueError as exc: + raise OSError(f"Cannot observe worker {pid} resident memory") from exc + return {"rss_bytes": rss, "private_commit_bytes": None} + + +def wait_bounded(process, *, timeout, max_memory_bytes, row, interval=0.2): + """A sampled guard, not an OS hard limit. Stop the owned tree on any failure.""" + started = time.perf_counter() + job = getattr(process, "_scratchv_job", None) + group = bool(getattr(process, "_scratchv_owned_group", False)) + tree = job if job is not None else _PosixGroup(process) if group else None + observation = {"samples": 0, "rss_peak_sampled_bytes": 0, "private_commit_peak_sampled_bytes": None, + "interval_seconds": interval, "limit_bytes": max_memory_bytes, + "ownership": "windows_job" if job else "posix_process_group", + "observed_pids": [], "peak_sampled_processes": 0, + "scope": ("Owned Windows Job and descendants; summed sampled RSS/private commit, shared pages may be counted per process, not a hard OS quota" + if job else "Owned Linux process group/session; summed sampled RSS, shared pages may be counted per process, not a hard OS quota")} + row["resource_monitor"] = observation + try: + if os.name == "nt" and isinstance(process, subprocess.Popen) and job is None: + raise ValueError("Windows workers must be created with spawn_owned before execution") + while process.poll() is None or (tree is not None and tree.pids()): + if time.perf_counter() - started > timeout: + raise TimeoutError(f"Full worker exceeded {timeout} seconds") + try: + values = tree.sample() if tree is not None else process_memory(process.pid) + except OSError: + if process.poll() is not None and (tree is None or not tree.pids()): + break + raise + observation["samples"] += 1 + observation["rss_peak_sampled_bytes"] = max(observation["rss_peak_sampled_bytes"], values["rss_bytes"]) + private = values["private_commit_bytes"] + pids = values.get("pids", [process.pid]) + observation["observed_pids"] = sorted(set(observation["observed_pids"]) | set(pids)) + observation["peak_sampled_processes"] = max(observation["peak_sampled_processes"], len(pids)) + if private is not None: + observation["private_commit_peak_sampled_bytes"] = max( + observation["private_commit_peak_sampled_bytes"] or 0, private) + if max(values["rss_bytes"], private or 0) > max_memory_bytes: + raise MemoryError(f"Full worker exceeded sampled memory limit {max_memory_bytes} bytes") + if process.poll() is not None: + time.sleep(interval) # Job children can outlive their launcher. + else: + try: + process.wait(timeout=interval) + except subprocess.TimeoutExpired: + pass + return process.wait(timeout=5) + except BaseException: + if process.poll() is None or job is not None or group: + try: + terminate_process_tree(process) + except (OSError, subprocess.SubprocessError) as cleanup: + # Preserve timeout/interruption rather than replacing its + # cause with any secondary cleanup diagnostic. + row["cleanup_error"] = f"{type(cleanup).__name__}: {cleanup}" + raise + finally: + if isinstance(tree, _PosixGroup): + observation["rss_exit_rechecks"] = tree.rss_exit_rechecks + if getattr(process, "_scratchv_job", None) is not None: + try: + job.close() + except OSError as cleanup: + row["cleanup_error"] = f"{type(cleanup).__name__}: {cleanup}" + process._scratchv_job = None + row["returncode"] = process.poll() + row["elapsed_seconds"] = time.perf_counter() - started diff --git a/probes/w3_qwen3_full/run.py b/probes/w3_qwen3_full/run.py new file mode 100644 index 0000000..8695e80 --- /dev/null +++ b/probes/w3_qwen3_full/run.py @@ -0,0 +1,295 @@ +"""Full pinned Qwen3 IR/ORT comparison with isolated, resource-bounded workers.""" +from __future__ import annotations + +import argparse +import json +import math +import os +from pathlib import Path +import subprocess +import sys +import time + +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +import numpy as np +from probes.w1_qwen3_export.run import verify_files +from probes.w3_common import new_output_dir, sha256_file, source_evidence, write_reports +from probes.w3_layer_diff.run import load_schema, load_arrays +from probes.w3_qwen3_full.cases import CASE_NAMES, input_cases +from probes.w3_qwen3_full.comparison import ATOL, compare_tensor, compare_positions +from probes.w3_qwen3_full.resources import spawn_owned, wait_bounded + +CHECKPOINTS = ("embedding", *(f"layer_{i}.output" for i in range(28)), "final_norm") +LOGIT_SHAPE = (1, 256, 151936) + + +def require(condition, message): + if not condition: + raise ValueError(message) + + +def checked_artifact(folder, descriptor, expected_name): + require(isinstance(descriptor, dict), f"Missing artifact {expected_name}") + require(descriptor.get("path") == expected_name, f"Unexpected artifact path: {expected_name}") + path = folder / expected_name + require(path.is_file() and path.resolve().parent == folder.resolve(), f"Missing/external artifact: {expected_name}") + require(path.stat().st_size == descriptor.get("bytes") and sha256_file(path) == descriptor.get("sha256"), + f"Altered artifact: {expected_name}") + return path + + +def load_logits(path): + value = np.load(path, mmap_mode="r", allow_pickle=False, max_header_size=10000) + require(value.shape == LOGIT_SHAPE and value.dtype == np.float32, "Wrong full logits shape/dtype") + return value + + +def saved_arithmetic_profile(mode, recorded): + """Validate saved arithmetic without guessing it from the auditing host CPU.""" + require(isinstance(recorded, dict), "Missing saved FP32 arithmetic profile") + if mode == "reference": + from scratchv.verification.fp32_reference import profile + strategy = recorded.get("cpu_strategy") + require(strategy in ("avx2-fma3", "avx512"), "Unsupported saved FP32 CPU profile") + canonical = profile(cpu_strategy=strategy) + else: + require(mode == "native", "Unsupported saved FP32 arithmetic profile") + canonical = {"name": "numpy-native"} + require(recorded == canonical and all(type(recorded[key]) is type(value) + for key, value in canonical.items()), "Noncanonical saved FP32 arithmetic profile") + return canonical + + +def validate_worker(folder, backend, input_path, sources, assets, *, expected_fp32_mode=None, + expected_fp32_profile=None): + require(expected_fp32_profile is None or expected_fp32_mode is not None, + "Saved FP32 profile requires an explicit mode") + for name in ("report.json", "report.md", "report.html"): + require((folder/name).is_file(), f"Missing worker {name}") + report = json.loads((folder/"report.json").read_text(encoding="utf-8")) + require(report.get("gate") == "numeric:w3-full-worker" and report.get("schema_version") == 1, + "Wrong worker identity") + require(report.get("backend") == backend and report.get("optimization_level") == "none", "Wrong backend") + require(report.get("passed") is True and report.get("status") == "PASS", "Worker not PASS") + require(report.get("full_model_executed") is True and report.get("full_ir_executed") is (backend == "ir"), + "Full model was not actually executed") + require(report.get("w3_exit_accepted") is False, "A worker cannot accept the team milestone") + require(report.get("source_sha256") == sources, "Worker production sources differ") + require(report.get("files") == assets, "Worker assets differ from fixed model") + require(report.get("input", {}).get("sha256") == sha256_file(input_path), "Worker input differs") + require(type(report.get("process_peak_rss_bytes")) is int and report["process_peak_rss_bytes"] > 0, + "Missing measured peak RSS") + timings = report.get("stages_seconds") + require(isinstance(timings, dict), "Missing worker execution timings") + for mode in ("ordinary", "diagnostic"): + name = f"{backend}_{mode}_execute" + duration = timings.get(name) + require(type(duration) in (int, float) and math.isfinite(duration) and duration > 0, + f"Missing/invalid worker execution timing: {name}") + if backend == "ort": + runtime = report.get("ort") + expected_runtime = {"provider": "CPUExecutionProvider", "graph_optimization": "disabled", + "intra_op_threads": 1, "inter_op_threads": 1, "cpu_mem_arena": False} + require(isinstance(runtime, dict) and all(type(runtime.get(key)) is type(value) + and runtime.get(key) == value for key, value in expected_runtime.items()), + "Worker ORT execution profile differs") + comparison = report.get("ordinary_diagnostic", {}) + require(comparison.get("passed") is True and comparison.get("comparison") == "exact" + and comparison.get("max_abs") == 0.0, "Diagnostic execution changed logits") + names = {"logits": "logits.npy", "diagnostic_logits": "diagnostic_logits.npy", + "checkpoints": "checkpoints.npz", "checkpoint_schema": "checkpoint_schema.json"} + artifacts = report.get("artifacts", {}) + for key, filename in names.items(): + checked_artifact(folder, artifacts.get(key), filename) + schema = load_schema(folder/"checkpoint_schema.json") + require(tuple(x["name"] for x in schema) == CHECKPOINTS, "Incomplete/out-of-order full checkpoints") + require(all(x["shape"] == [1, 256, 1024] and x["dtype"] == "float32" and x.get("sequence_axis") == 1 + for x in schema), "Wrong full checkpoint contract") + require(report.get("checkpoints") == schema, "Worker report/schema disagree") + ordinary, diagnostic = load_logits(folder/"logits.npy"), load_logits(folder/"diagnostic_logits.npy") + exact = compare_tensor(ordinary, diagnostic) + require(exact["passed"] and exact["max_abs"] == 0, "Ordinary/diagnostic arrays actually differ") + del ordinary, diagnostic + if backend == "ir": + ir = report.get("ir") + require(isinstance(ir, dict), "Missing IR execution profile") + instruction_count = ir.get("instruction_count") + require(type(instruction_count) is int and instruction_count > 0, + "Missing/invalid IR instruction count") + if expected_fp32_mode is not None: + from scratchv.verification.fp32_reference import profile + require(report.get("fp32_mode") == expected_fp32_mode, + "Worker FP32 execution mode differs") + # Live runs require the selected local strategy. Saved-array audits + # can explicitly supply a known producer strategy from another CPU; + # the complete canonical profile still has to match. + expected_profile = (saved_arithmetic_profile(expected_fp32_mode, expected_fp32_profile) + if expected_fp32_profile is not None else + profile() if expected_fp32_mode == "reference" else {"name": "numpy-native"}) + require(ir.get("fp32_mode") == expected_fp32_mode + and ir.get("fp32_profile") == expected_profile, + "Worker FP32 arithmetic profile differs") + for mode in ("ordinary", "diagnostic"): + execution = report.get("executions", {}).get(mode, {}) + # The pinned full graph is one straight-line block, including RETURN; + # every declared instruction must execute once in both runs. + require(type(execution.get("executed_steps")) is int + and execution["executed_steps"] == instruction_count, + "IR execution steps do not cover the declared instruction count") + memory = execution.get("memory_stats", {}) + require(type(memory.get("peak_numpy_storage_bytes")) is int and memory["peak_numpy_storage_bytes"] > 0, + "Missing IR storage measurement") + return report + + +def compare_case(folder, valid_length): + rows = [] + for filename in ("logits.npy", "diagnostic_logits.npy"): + actual, reference = load_logits(folder/"ir"/filename), load_logits(folder/"ort"/filename) + rows.append({"name": filename, **compare_positions(actual, reference, valid_length)}) + del actual, reference + schema = load_schema(folder/"ir/checkpoint_schema.json") + require(schema == load_schema(folder/"ort/checkpoint_schema.json"), "IR/ORT checkpoint schema differs") + actual = load_arrays(folder/"ir/checkpoints.npz", schema, max_bytes=64*1024**2) + reference = load_arrays(folder/"ort/checkpoints.npz", schema, max_bytes=64*1024**2) + checkpoints = [{"name": entry["name"], **compare_positions(actual[entry["name"]], reference[entry["name"]], valid_length)} + for entry in schema] + all_rows = checkpoints + rows + worst = max((r for r in all_rows if r["max_abs"] is not None), key=lambda x: x["max_abs"], default=None) + return {"passed": all(r["passed"] for r in rows), "logits": rows, "checkpoints": checkpoints, + "max_logits_abs": max(r["max_abs"] for r in rows), + "checkpoint_threshold_scope": "Diagnostic location only: large intermediate activations can exceed 1e-4 in one FP32 ULP; acceptance applies to complete output logits", + "first_divergence": next((r for r in all_rows if not r["passed"]), None), + "worst": worst, "atol": ATOL, "rtol": 0} + + +def invariants(out, successful): + rows = [] + for name, reference, prefix in (("changed_future", "full_seed_0", 64), ("changed_padding", "short_17", 17)): + if name not in successful or reference not in successful: + continue + for backend in ("ort", "ir"): + a = load_logits(out/name/backend/"logits.npy") + b = load_logits(out/reference/backend/"logits.npy") + rows.append({"case": name, "reference": reference, "prefix": prefix, "backend": backend, + **compare_tensor(a[:, :prefix], b[:, :prefix])}) + del a, b + return rows + + +def run_worker(args, folder, backend, input_path, row=None): + command = [str(Path(args.python).resolve()), "-B", "-X", "utf8", + "probes/w3_qwen3_full/worker.py", "--backend", backend, + "--model-dir", str(args.model_dir.resolve()), "--input-file", str(input_path), + "--output-dir", str(folder/backend), "--optimization-level", "none", + "--fp32-mode", getattr(args, "fp32_mode", "reference")] + row = {} if row is None else row + row.update(backend=backend, passed=False, command=command) + env = dict(os.environ, PYTHONIOENCODING="utf-8", OMP_NUM_THREADS="1", OPENBLAS_NUM_THREADS="1", MKL_NUM_THREADS="1") + options = {"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP} if os.name == "nt" else {"start_new_session": True} + with (folder/(backend+".log")).open("w", encoding="utf-8") as stream: + process = spawn_owned(command, cwd=ROOT, env=env, stdout=stream, stderr=subprocess.STDOUT, **options) + try: + row["returncode"] = wait_bounded(process, timeout=args.worker_timeout, + max_memory_bytes=int(args.max_worker_memory_gib*1024**3), row=row) + require(row["returncode"] == 0, f"{backend} worker exited {row['returncode']}") + row["passed"] = True + except Exception as exc: + row["error"] = f"{type(exc).__name__}: {exc}" + for filename in ("report.json", "progress.json"): + try: + child = json.loads((folder/backend/filename).read_text(encoding="utf-8")) + row["child_stage"] = child.get("stage") + row["child_error"] = child.get("error") + break + except (OSError, ValueError): + pass + return row + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--model-dir", required=True, type=Path) + parser.add_argument("--output-dir", required=True, type=Path) + parser.add_argument("--python", default=sys.executable) + parser.add_argument("--worker-timeout", type=float, default=1800) + parser.add_argument("--max-worker-memory-gib", type=float, default=10) + parser.add_argument("--case", action="append", choices=CASE_NAMES) + parser.add_argument("--fp32-mode", choices=("native", "reference"), default="reference", + help="Explicit IR FP32 arithmetic policy; native reproduces the original NumPy baseline") + args = parser.parse_args(argv) + if any(not math.isfinite(v) or v <= 0 for v in (args.worker_timeout, args.max_worker_memory_gib)): + parser.error("Resource limits must be finite and positive") + if args.case and len(args.case) != len(set(args.case)): + parser.error("Duplicate case selection") + out = new_output_dir(args.output_dir) + selected = set(args.case or CASE_NAMES) + start = time.perf_counter() + report = {"gate": "numeric:ir-full-qwen3", "status": "FAIL", "passed": False, + "optimization_level": "none", "atol": ATOL, "rtol": 0, + "fp32_mode": args.fp32_mode, + "full_ir_executed": False, "w3_exit_accepted": False, + "scope": "Pinned 28-layer FP32 L256 IR vs ORT; strict 1e-4 on all ordinary/diagnostic logits including padding; 30 checkpoints localize errors", + "cases": [], "invariants": [], "required_cases": list(CASE_NAMES), + "selected_cases": [x for x in CASE_NAMES if x in selected], "stage": "sources"} + try: + report.update(source_evidence()) + report["stage"] = "assets" + assets = verify_files(args.model_dir) + report["files"] = assets + successful = set() + for name, valid, feed in input_cases(): + if name not in selected: + continue + report["stage"] = name + folder = out/name + folder.mkdir() + input_path = folder/"inputs.npz" + np.savez(input_path, **feed) + row = {"name": name, "valid_length": valid, "passed": False, "workers": [], + "input_sha256": sha256_file(input_path)} + report["cases"].append(row) + for backend in ("ort", "ir"): + worker = {"backend": backend, "passed": False} + row["workers"].append(worker) + run_worker(args, folder, backend, input_path, worker) + if worker["passed"]: + try: + evidence = validate_worker(folder/backend, backend, input_path, + report["source_sha256"], assets, + expected_fp32_mode=args.fp32_mode) + require(evidence["input"]["valid_length"] == valid, "Worker valid length differs") + worker.update(report_sha256=sha256_file(folder/backend/"report.json"), + process_peak_rss_bytes=evidence["process_peak_rss_bytes"], + stages_seconds=evidence["stages_seconds"]) + if backend == "ir": + report["full_ir_executed"] = True + except Exception as exc: + worker.update(passed=False, error=f"{type(exc).__name__}: {exc}") + print(f"[{name}/{backend}] {'EXECUTED' if worker['passed'] else 'FAIL'}", flush=True) + if all(w["passed"] for w in row["workers"]): + row["comparison"] = compare_case(folder, valid) + row["passed"] = row["comparison"]["passed"] + successful.add(name) + print(f"[{name}/compare] {'PASS' if row['passed'] else 'FAIL'}; logits_max_abs={row['comparison']['max_logits_abs']}; diagnostic_max_abs={row['comparison']['worst']['max_abs']}", flush=True) + write_reports(out, report) + report["invariants"] = invariants(out, successful) + require(source_evidence()["source_sha256"] == report["source_sha256"], "Sources changed during full execution") + coverage = set(selected) == set(CASE_NAMES) and len(report["cases"]) == len(CASE_NAMES) and len(report["invariants"]) == 4 + numeric = len(report["cases"]) == len(selected) and all(row["passed"] for row in report["cases"]) and all(x["passed"] for x in report["invariants"]) + report.update(coverage_complete=coverage, selected_numerical_passed=numeric, + passed=coverage and numeric, status="PASS" if coverage and numeric else ("PARTIAL" if numeric else "FAIL"), stage="complete") + except KeyboardInterrupt: + report.update(status="FAIL", passed=False, interrupted=True, error="KeyboardInterrupt: full verification cancelled") + except Exception as exc: + report.update(status="FAIL", passed=False, error=f"{type(exc).__name__}: {exc}") + report["elapsed_seconds"] = time.perf_counter() - start + write_reports(out, report) + return 130 if report.get("interrupted") else (0 if report["passed"] else (2 if report["status"] == "PARTIAL" else 1)) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/probes/w3_qwen3_full/worker.py b/probes/w3_qwen3_full/worker.py new file mode 100644 index 0000000..e86e297 --- /dev/null +++ b/probes/w3_qwen3_full/worker.py @@ -0,0 +1,327 @@ +"""Execute one complete pinned Qwen3 backend in an isolated process. + +A worker PASS proves execution and ordinary/diagnostic consistency only. The +parent compares independent IR and ORT results; no worker accepts W3 itself. +""" +from __future__ import annotations + +import argparse +import gc +import json +import os +from pathlib import Path +import sys +import tempfile +import time +import traceback + +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +import numpy as np +import onnx + +from probes.w1_qwen3_export import run as assets +from probes.w2_qwen3_parse.validation import audit_graph_structure +from probes.w3_common import (atomic_text, new_output_dir, process_peak_rss_bytes, + sha256_file, source_evidence, write_reports) +from probes.w3_layer_diff.run import load_arrays + +GATE = "numeric:w3-full-worker" +INPUT_SCHEMA = [ + {"name": "input_ids", "shape": [1, 256], "dtype": "int64"}, + {"name": "attention_mask", "shape": [1, 1, 256, 256], "dtype": "float32"}, +] +LOGITS_SHAPE = (1, 256, 151936) + + +def load_inputs(path): + """Validate the pinned export's finite additive, right-padded causal mask.""" + feed = load_arrays(Path(path), INPUT_SCHEMA, max_bytes=1024 * 1024) + ids, mask = feed["input_ids"], feed["attention_mask"] + if np.any(ids < 0) or np.any(ids >= 151936): + raise ValueError("input_ids must be in the fixed Qwen3 vocabulary") + length = int(np.count_nonzero(mask[0, 0, -1] == 0)) + if not 1 <= length <= 256: + raise ValueError("attention_mask must expose at least one valid token") + query, key = np.arange(256)[:, None], np.arange(256)[None, :] + wanted = np.where((key <= query) & (key < length), np.float32(0), + np.finfo(np.float32).min).reshape(1, 1, 256, 256) + if not np.array_equal(mask, wanted): + raise ValueError("attention_mask is not the pinned causal/right-padding additive mask") + return feed, length + + +def checkpoint_schema(model, audit=None): + """Locate checkpoints from audited dataflow, never layer-name substrings.""" + audit = audit_graph_structure(model) if audit is None else audit + layers = audit.get("layers", []) + if (audit.get("passed") is not True or audit.get("layer_count") != 28 + or [row.get("layer") for row in layers] != list(range(28))): + raise ValueError("Checkpoint coverage requires the complete ordered 28-layer audit") + by_name = {} + for node in model.graph.node: + by_name.setdefault(node.name, []).append(node) + + def output(name): + nodes = by_name.get(name, []) + if len(nodes) != 1 or len(nodes[0].output) != 1: + raise ValueError(f"Checkpoint node must be unique with one output: {name}") + return nodes[0].output[0] + + rows = [{"name": "embedding", "onnx_name": output(audit["embedding"]), + "checkpoint": "embedding"}] + rows += [{"name": f"layer_{row['layer']}.output", "onnx_name": row["output"], + "layer": row["layer"], "checkpoint": "output"} for row in layers] + rows += [{"name": "final_norm", "onnx_name": output(audit["final_norm"]), + "checkpoint": "final_norm"}] + definitions = {name for node in model.graph.node for name in node.output} + names = [row["onnx_name"] for row in rows] + if len(names) != len(set(names)) or not set(names) <= definitions: + raise ValueError("Checkpoint tensors must be unique graph-produced values") + for row in rows: + row.update(shape=[1, 256, 1024], dtype="float32", sequence_axis=1) + return rows + + +def checked_tensor(array, shape, name): + if not isinstance(array, np.ndarray) or array.shape != tuple(shape) or array.dtype != np.float32: + raise ValueError(f"{name}: expected FP32 {tuple(shape)}, got " + f"{getattr(array, 'shape', None)}/{getattr(array, 'dtype', None)}") + for start in range(0, shape[1], 8): + if not np.isfinite(array[:, start:start + 8]).all(): + raise ValueError(f"{name}: nonfinite output") + return array + + +def save_array(path, value, *, archive=False): + """Atomically persist numeric arrays; leave existing evidence untouched.""" + path = Path(path) + if path.exists(): + raise FileExistsError(path) + fd, temporary = tempfile.mkstemp(prefix="." + path.name, suffix=".tmp", dir=path.parent) + try: + with os.fdopen(fd, "wb") as stream: + if archive: + np.savez(stream, **value) + else: + np.save(stream, value, allow_pickle=False) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, path) + finally: + if os.path.exists(temporary): + os.unlink(temporary) + + +def evidence(path, *, array=None): + path = Path(path) + record = {"path": path.name, "sha256": sha256_file(path), "bytes": path.stat().st_size} + if array is not None: + record.update(shape=list(array.shape), dtype=str(array.dtype), all_finite=True) + return record + + +def save_logits(out, name, array, report): + checked_tensor(array, LOGITS_SHAPE, name) + path = out / f"{name}.npy" + save_array(path, array) + report["artifacts"][name] = evidence(path, array=array) + + +def save_checkpoints(out, arrays, schema, report): + if set(arrays) != {row["name"] for row in schema} or len(arrays) != 30: + raise ValueError("Diagnostic output omitted/duplicated a required checkpoint") + for row in schema: + checked_tensor(arrays[row["name"]], row["shape"], row["name"]) + path = out / "checkpoints.npz" + save_array(path, arrays, archive=True) + report["artifacts"]["checkpoints"] = evidence(path) + + +def exact_ordinary_diagnostic(out): + ordinary = np.load(out / "logits.npy", mmap_mode="r", allow_pickle=False) + diagnostic = np.load(out / "diagnostic_logits.npy", mmap_mode="r", allow_pickle=False) + checked_tensor(ordinary, LOGITS_SHAPE, "ordinary logits") + checked_tensor(diagnostic, LOGITS_SHAPE, "diagnostic logits") + for start in range(0, LOGITS_SHAPE[1], 8): + if not np.array_equal(ordinary[:, start:start + 8], diagnostic[:, start:start + 8]): + raise ValueError(f"Ordinary and diagnostic logits differ at sequence chunk {start}") + return {"passed": True, "comparison": "exact", "max_abs": 0.0, + "positions_compared": 256, "elements_compared": int(np.prod(LOGITS_SHAPE))} + + +def diagnostic_graph(model, schema, model_dir, out): + """Add graph outputs without re-exporting or copying the external weights.""" + model = onnx.ModelProto.FromString(model.SerializeToString()) + for row in schema: + model.graph.output.append(onnx.helper.make_tensor_value_info( + row["onnx_name"], onnx.TensorProto.FLOAT, row["shape"])) + # External references are now relative to the derived metadata's directory. + # Source manifest verification already bounded and validated these files. + for tensor in assets.tensors(model): + if tensor.data_location == onnx.TensorProto.EXTERNAL: + for entry in tensor.external_data: + if entry.key == "location": + original = (model_dir / entry.value).resolve() + if not original.is_relative_to(model_dir.resolve()) or not original.is_file(): + raise ValueError("Invalid external tensor location in diagnostic model") + entry.value = os.path.relpath(original, out).replace("\\", "/") + target = out / "diagnostic.onnx" + target.write_bytes(model.SerializeToString()) + return target + + +def execute_ort(model_dir, out, feed, schema, model, report, stage): + import onnxruntime as ort + + def session(path): + options = ort.SessionOptions() + options.intra_op_num_threads = 1 + options.inter_op_num_threads = 1 + options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_DISABLE_ALL + options.enable_cpu_mem_arena = False + return ort.InferenceSession(str(path), options, providers=["CPUExecutionProvider"]) + + runtime = stage("ort_ordinary_session", lambda: session(model_dir / "model.onnx")) + logits = stage("ort_ordinary_execute", lambda: runtime.run(["logits"], feed)[0]) + stage("ordinary_save", lambda: save_logits(out, "logits", logits, report)) + del logits, runtime + gc.collect() + diagnostic = stage("diagnostic_graph", lambda: diagnostic_graph(model, schema, model_dir, out)) + report["artifacts"]["diagnostic_model"] = evidence(diagnostic) + runtime = stage("ort_diagnostic_session", lambda: session(diagnostic)) + names = ["logits", *[row["onnx_name"] for row in schema]] + results = stage("ort_diagnostic_execute", lambda: runtime.run(names, feed)) + if len(results) != 31: + raise ValueError("ORT diagnostic did not return all 31 outputs") + stage("diagnostic_save", lambda: save_logits(out, "diagnostic_logits", results[0], report)) + arrays = {row["name"]: value for row, value in zip(schema, results[1:])} + stage("checkpoints_save", lambda: save_checkpoints(out, arrays, schema, report)) + report["ort"] = {"provider": "CPUExecutionProvider", "graph_optimization": "disabled", + "intra_op_threads": 1, "inter_op_threads": 1, "cpu_mem_arena": False} + + +def execute_ir(model_dir, out, feed, schema, model, report, stage): + from scratchv.frontend.onnx_parser import ONNXParser + from scratchv.verification.ir_interpreter import IRInterpreter + from scratchv.verification.fp32_reference import profile + + fp32_mode = report.get("fp32_mode", "reference") + + parser = ONNXParser() + program = stage("scratchv_parse", lambda: parser.parse(str(model_dir / "model.onnx"), + mmap_external_data=True)) + for value in parser.initializers.values(): + value.setflags(write=False) + mapping = {} + for row in schema: + value = parser._value_map.get(row["onnx_name"]) + if value is None or value.name in mapping: + raise ValueError(f"Missing/non-unique IR checkpoint mapping: {row['name']}") + mapping[value.name] = row + report["ir_checkpoint_bindings"] = {name: row["name"] for name, row in mapping.items()} + report["executions"] = {} + for kind in ("ordinary", "diagnostic"): + arrays = {} + + def observe(name, value): + if name in mapping: + row = mapping[name] + if row["name"] in arrays: + raise ValueError(f"Checkpoint executed multiple times: {row['name']}") + arrays[row["name"]] = checked_tensor(value, row["shape"], row["name"]).copy() + + result = stage(f"ir_{kind}_execute", lambda: IRInterpreter(program).run( + feed, initializers=parser.initializers, collect_memory_stats=True, + memory_mode="last_use", copy_initializers=False, + fp32_mode=fp32_mode, + observer=observe if kind == "diagnostic" else None)) + report["executions"][kind] = {"executed_steps": result.executed_steps, + "memory_stats": result.memory_stats} + label = "logits" if kind == "ordinary" else "diagnostic_logits" + stage(f"{kind}_save", lambda: save_logits(out, label, result.return_value, report)) + if kind == "diagnostic": + stage("checkpoints_save", lambda: save_checkpoints(out, arrays, schema, report)) + del result, arrays + gc.collect() + report["ir"] = {"optimization_level": "none", "memory_mode": "last_use", + "fp32_mode": fp32_mode, + "fp32_profile": profile() if fp32_mode == "reference" else {"name": "numpy-native"}, + "copy_initializers": False, "mmap_external_data": True, + "instruction_count": sum(len(b.instructions) for f in program.functions for b in f.blocks)} + + +def main(argv=None): + cli = argparse.ArgumentParser(description=__doc__) + cli.add_argument("--backend", choices=("ort", "ir"), required=True) + cli.add_argument("--model-dir", type=Path, required=True) + cli.add_argument("--input-file", type=Path, required=True) + cli.add_argument("--output-dir", type=Path, required=True) + cli.add_argument("--optimization-level", choices=("none",), default="none") + cli.add_argument("--fp32-mode", choices=("native", "reference"), default="reference") + args = cli.parse_args(argv) + out = new_output_dir(args.output_dir) + started = time.perf_counter() + report = {"schema_version": 1, "gate": GATE, "backend": args.backend, + "optimization_level": args.optimization_level, "status": "FAIL", "passed": False, + "fp32_mode": args.fp32_mode if args.backend == "ir" else None, + "stage": "initialization", "stages_seconds": {}, "artifacts": {}, + "full_ir_executed": False, "full_model_executed": False, "w3_exit_accepted": False, + "scope": "Complete single-backend execution and exact ordinary/diagnostic consistency. " + "Independent IR/ORT comparison belongs to the parent gate; not W3 team acceptance.", + "model_id": assets.MANIFEST["model_id"], "revision": assets.MANIFEST["revision"]} + + def stage(name, operation): + report["stage"] = name + atomic_text(out / "progress.json", json.dumps({"stage": name, + "elapsed_seconds": time.perf_counter() - started, + "process_peak_rss_bytes": process_peak_rss_bytes()}) + "\n") + print(f"[{args.backend}] {name}", flush=True) + begin = time.perf_counter() + try: + return operation() + finally: + report["stages_seconds"][name] = time.perf_counter() - begin + + try: + report.update(stage("source_evidence", source_evidence)) + model_dir = args.model_dir.resolve() + report["model_dir"] = str(model_dir) + report["files"] = stage("asset_hashes", lambda: assets.verify_files(model_dir)) + report["onnx_contract"] = stage("onnx_contract", lambda: assets.inspect_model(model_dir, report["files"])) + feed, length = stage("input_validation", lambda: load_inputs(args.input_file)) + report["input"] = evidence(args.input_file) + report["input"].update(path=str(args.input_file.resolve()), valid_length=length) + model = stage("metadata_load", lambda: onnx.load(str(model_dir / "model.onnx"), load_external_data=False)) + report["graph_audit"] = stage("graph_audit", lambda: audit_graph_structure(model)) + schema = stage("checkpoint_selection", lambda: checkpoint_schema(model, report["graph_audit"])) + report["checkpoints"] = schema + atomic_text(out / "checkpoint_schema.json", json.dumps({"version": 1, "checkpoints": schema}, indent=2) + "\n") + report["artifacts"]["checkpoint_schema"] = evidence(out / "checkpoint_schema.json") + executor = execute_ort if args.backend == "ort" else execute_ir + executor(model_dir, out, feed, schema, model, report, stage) + report["full_model_executed"] = True + report["full_ir_executed"] = args.backend == "ir" + report["ordinary_diagnostic"] = stage("ordinary_diagnostic_consistency", lambda: exact_ordinary_diagnostic(out)) + final_sources = stage("source_recheck", source_evidence)["source_sha256"] + if final_sources != report["source_sha256"]: + raise ValueError("Production sources changed during numerical execution") + report.update(passed=True, status="PASS", stage="complete") + except BaseException as exc: + report["error"] = f"{type(exc).__name__}: {exc}" + traceback.print_exc() + if isinstance(exc, KeyboardInterrupt): + report["interrupted"] = True + report.update(elapsed_seconds=time.perf_counter() - started, + process_peak_rss_bytes=process_peak_rss_bytes(), + peak_scope="Lifetime resident high-water mark of this single backend worker, " + "including both ordinary and diagnostic runs; not isolated kernel memory") + write_reports(out, report) + print(f"[{GATE}] {report['status']}; {out / 'report.json'}", flush=True) + return 130 if report.get("interrupted") else 0 if report["passed"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/probes/w3_qwen3_medium/README.md b/probes/w3_qwen3_medium/README.md new file mode 100644 index 0000000..c06072d --- /dev/null +++ b/probes/w3_qwen3_medium/README.md @@ -0,0 +1,110 @@ +# W3 medium Qwen3 host numerical probe + +Reproduction targets **Ubuntu 24.04 x86_64, Bash and Python 3.12**. Complete the [Linux environment and asset setup](../../docs/llm-deploy-v1.0/LINUX_REPRODUCTION.md) first. Run commands from the repository root in the same Bash session; the setup exports `SCRATCHV_PYTHON`, `SCRATCHV_CC` and `SCRATCHV_QEMU`. + +This is a fixed candidate for the W3 six-layer experiment. The plan fixes six +layers; the dimensions below are the candidate chosen for this experiment, +not a claim that the plan mandated a specific hidden or feed-forward width. + +| Property | W2 small default | W3 medium candidate | +|---|---:|---:| +| Decoder layers | 2 | 6 | +| Hidden width | 32 | 64 | +| FFN intermediate width | 96 | 192 | +| Query heads / KV heads | 4 / 2 | 4 / 2 | +| Head dimension | 16 | 16 | +| Vocabulary / sequence length | 128 / 256 | 128 / 256 | +| Batch / precision | 1 / FP32 | 1 / FP32 | +| Named checkpoints | 29 | 81 | + +Weights are fixed-seed random initialization of the official +`transformers==4.51.3` `Qwen3ForCausalLM`. No weights or tokenizer are downloaded. +This gate measures architecture and numerical semantics. It does not establish +pretrained language quality, full Qwen3-0.6B execution, or RISC-V/QEMU correctness. +Attention is eager, positions are `0..255`, embeddings and LM head are tied, and +there is no KV cache. All numerical backends use one CPU thread. + +## Run + +Use Python 3.12 and the existing pinned +`requirements/qwen3-small-probe.txt` environment. From the W3 worktree: + +```bash +set -euo pipefail +export OMP_NUM_THREADS='1' +export OPENBLAS_NUM_THREADS='1' +export MKL_NUM_THREADS='1' +"$SCRATCHV_PYTHON" -m probes.w3_qwen3_medium.run --output-dir output/w3-medium --model-seed 0 +``` + +The output directory must not already exist, even if it is empty. Preserve a +failed directory and choose a new path for a retry. The runner exits 0 only if the entire gate passes; numeric, +parsing, optimization, execution, and missing-reference failures exit 1. + +## Acceptance and evidence + +The gate uses the existing seven W2 inputs: two full random sequences, lengths +1, 17, and 255 with right padding, an altered future suffix, and altered masked +padding tokens. It exports the official model with named observations, then +produces ordinary logits-only and diagnostic packed-output ONNX graphs. + +Both graphs run in ORT with graph optimization disabled and in ScratchV IR at +`none`, `basic`, and `all`. Every IR path compares against both PyTorch and its +own ORT graph. Ordinary graph, diagnostic graph, and every optimization level +record independent results. One failure cannot be masked by another passing +path. Independent NumPy GQA/probability/context and blocked-attention checks +also run against PyTorch. + +The report uses uppercase `PASS` / `FAIL`. Its `coverage_complete` field requires +the exact seven named cases and lengths, both ORT graphs, all six IR variants +per case, and all 18 named/backend/graph isolation checks. Missing or duplicated +evidence cannot pass by reducing the set of checks being summarized. + +All comparisons require matching shape and dtype, finite values, and strict +`max_abs < 1e-5`, with `rtol=0`. The threshold is fixed and is not a CLI option. +All positions are included, with valid-token and padding-query errors reported +separately. Causal and padding isolation compare the valid prefix at **every +checkpoint**, for PyTorch, both ORT graphs, and all six IR variants. + +Checkpoint count is `3 + 13 * num_hidden_layers`: embedding, 13 observations per +layer, final norm, and logits. Shapes are derived from the actual model config, +including query/KV head count, head dimension, hidden width, vocabulary, and +sequence length. `probes/w2_qwen3_small/model.py` retains its two-layer defaults +and accepts explicit `model_config` overrides plus `sequence_length`. + +The output contains: + +- `model.onnx`, `checkpoints.onnx`, and `diagnostics.onnx`. +- `config.json`, `checkpoint_metadata.json`, packed `checkpoints.json`, and + versioned named-NPZ `trace_schema.json` / `logits_schema.json`. +- `inputs_' + + escape(coverage) + '
' + escape(explanation) + + '
" + escape(description) + "
" + if comparisons: + md += compact_metric_table(comparisons) + "\n\n" + page += compact_metric_table_html(comparisons) + md += "\n\n".join(escape(note) for note in notes) + "\n\n" + page += "".join("" + escape(note) + "
" for note in notes) + return md, page diff --git a/pyproject.toml b/pyproject.toml index 670aa63..b4320d5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,7 +7,7 @@ name = "scratchv" version = "0.3.0" description = "A compiler from ONNX models to RISC-V assembly and LLVM IR" readme = "README.md" -license = {text = "MIT"} +license = {text = "MIT AND MPL-2.0"} keywords = ["compiler", "risc-v", "onnx", "llvm", "machine-learning"] requires-python = ">=3.8" dependencies = [ @@ -20,6 +20,7 @@ classifiers = [ "Intended Audience :: Developers", "Intended Audience :: Education", "License :: OSI Approved :: MIT License", + "License :: OSI Approved :: Mozilla Public License 2.0 (MPL 2.0)", "Programming Language :: Python :: 3", "Programming Language :: Python :: 3.8", "Programming Language :: Python :: 3.9", @@ -46,6 +47,9 @@ Documentation = "https://github.com/kinsomwang/ScratchV/tree/main/docs/INDEX.md" [project.scripts] scratchv = "scratchv.main:main" +[tool.setuptools] +license-files = ["LICENSE", "LICENSES/*.txt", "THIRD_PARTY_NOTICES.txt"] + [tool.setuptools.packages.find] include = ["scratchv", "scratchv.*", "scratchv_dag", "scratchv_dag.*", "benchmarks", "benchmarks.*"] diff --git a/scratchv/frontend/onnx_parser.py b/scratchv/frontend/onnx_parser.py index 6bcfacc..364dc0f 100644 --- a/scratchv/frontend/onnx_parser.py +++ b/scratchv/frontend/onnx_parser.py @@ -3,7 +3,7 @@ from __future__ import annotations import math -from pathlib import Path +from pathlib import Path, PureWindowsPath from typing import TYPE_CHECKING if TYPE_CHECKING: @@ -34,9 +34,20 @@ def __init__(self): self._producers = {} self._constant_cache = {} self._base_dir = "" + self._mmap_external_data = False - def parse(self, model_path: str) -> Program: - """Parse an ONNX model file and return an IR Program.""" + def parse(self, model_path: str, *, mmap_external_data: bool = False) -> Program: + """Parse ONNX, optionally mapping external weights as read-only arrays. + + The default loader is unchanged. The explicit mmap mode avoids + materializing external weights in TensorProto.raw_data. External files + must be within the model directory, with valid byte ranges matching + the tensor shape and dtype. The caller must keep those files unchanged + while this parser's initializer arrays are in use. Mapping is not an + integrity check; model/weight hashes must be validated separately. + """ + if not isinstance(mmap_external_data, bool): + raise ONNXParseError("mmap_external_data must be boolean") try: import onnx except ImportError: @@ -52,6 +63,7 @@ def parse(self, model_path: str) -> Program: self._producers = {} self._constant_cache = {} self._base_dir = str(Path(model_path).resolve().parent) + self._mmap_external_data = mmap_external_data # Inspect/infer the graph before materializing external weights. This # avoids serializing a >2 GiB protobuf just to infer tensor shapes. model = onnx.load(model_path, load_external_data=False) @@ -93,7 +105,10 @@ def parse(self, model_path: str) -> Program: # Map ONNX initializers (constants) to IR values for init in graph.initializer: - arr = onnx.numpy_helper.to_array(init, base_dir=self._base_dir) + if mmap_external_data and onnx.external_data_helper.uses_external_data(init): + arr = self._map_external_initializer(init) + else: + arr = onnx.numpy_helper.to_array(init, base_dir=self._base_dir) self._bind_constant(init.name, arr) # Map graph inputs to function params @@ -127,6 +142,66 @@ def parse(self, model_path: str) -> Program: return self.builder.program + def _map_external_initializer(self, tensor): + """Validate before mapping; never populate the TensorProto raw_data.""" + import numpy as np + from onnx import helper + + self._dtype(tensor.data_type) # Keep the frontend's supported dtype set. + fields = {} + for entry in tensor.external_data: + if entry.key in fields: + raise ONNXParseError(f"Duplicate external field for {tensor.name}: {entry.key}") + fields[entry.key] = entry.value + unknown = set(fields) - {"location", "offset", "length", "checksum"} + if unknown: + raise ONNXParseError(f"Unsupported external fields for {tensor.name}: {sorted(unknown)}") + location = fields.get("location", "") + windows = PureWindowsPath(location) + path = Path(location) + if (not location or path.is_absolute() or windows.is_absolute() + or windows.drive or ".." in path.parts or ".." in windows.parts): + raise ONNXParseError(f"External path must stay within the model directory: {tensor.name}") + # Recognize either separator on every host; resolve symlinks as well. + base = Path(self._base_dir).resolve() + candidate = (base / Path(*windows.parts)).resolve() + try: + candidate.relative_to(base) + except ValueError as exc: + raise ONNXParseError(f"External path escapes the model directory: {tensor.name}") from exc + if candidate == base: + raise ONNXParseError(f"External path escapes the model directory: {tensor.name}") + if tensor.HasField("raw_data") or tensor.HasField("segment"): + raise ONNXParseError(f"Ambiguous raw or segmented external tensor: {tensor.name}") + shape = tuple(tensor.dims) + if any(dimension < 0 for dimension in shape): + raise ONNXParseError(f"Negative external tensor dimension: {tensor.name}") + dtype = np.dtype(helper.tensor_dtype_to_np_dtype(tensor.data_type)).newbyteorder("<") + expected = math.prod(shape) * dtype.itemsize + + def size_field(name, default): + value = fields.get(name) + if value is None: + return default + if not value or not value.isascii() or not value.isdecimal(): + raise ONNXParseError(f"External {name} must be a nonnegative integer: {tensor.name}") + return int(value) + + offset = size_field("offset", 0) + length = size_field("length", expected) + if length != expected: + raise ONNXParseError(f"External length disagrees with tensor shape/dtype: {tensor.name}") + try: + if not candidate.is_file() or offset + expected > candidate.stat().st_size: + raise ONNXParseError(f"External tensor byte range exceeds file: {tensor.name}") + if expected == 0: + array = np.empty(shape, dtype=dtype) + array.setflags(write=False) + return array + return np.memmap(candidate, mode="r", dtype=dtype, offset=offset, shape=shape, order="C") + except (OSError, ValueError, OverflowError) as exc: + raise ONNXParseError(f"Cannot map external tensor {tensor.name}: {exc}") from exc + def _translate_node(self, node, output_names: set[str]) -> None: """Translate a single ONNX node to IR instructions.""" op_type = node.op_type @@ -244,13 +319,16 @@ def _constant_ints(self, value: Value) -> tuple[int, ...]: def _handle_constant(self, node, inputs, outputs): import numpy as np - from onnx import numpy_helper + from onnx import external_data_helper, numpy_helper attrs = self._attributes(node) if len(attrs) != 1 or len(outputs) != 1 or inputs: raise ONNXParseError("Constant requires one value attribute and one output") key, value = next(iter(attrs.items())) if key == "value": - data = numpy_helper.to_array(value, base_dir=self._base_dir) + if self._mmap_external_data and external_data_helper.uses_external_data(value): + data = self._map_external_initializer(value) + else: + data = numpy_helper.to_array(value, base_dir=self._base_dir) elif key in ("value_int", "value_ints"): data = np.asarray(value, dtype=np.int64) elif key in ("value_float", "value_floats"): diff --git a/scratchv/verification/fp32_eigen.py b/scratchv/verification/fp32_eigen.py new file mode 100644 index 0000000..c53960d --- /dev/null +++ b/scratchv/verification/fp32_eigen.py @@ -0,0 +1,156 @@ +# SPDX-License-Identifier: MPL-2.0 +# Adapted from Eigen (https://gitlab.com/libeigen/eigen), commit +# 1d8b82b0740839c0de7f1242a3585e3390ff5f33: +# Eigen/src/Core/Redux.h +# Eigen/src/Core/arch/SSE/PacketMath.h +# Eigen/src/Core/arch/Default/GenericPacketMathFunctions.h +# Copyright (C) 2006-2008 Benoit Jacob