refactor: introduce WinMLManifest dataclass as single source of truth for build_manifest.json

docs/reference/output-layout.md

@@ -40,7 +40,7 @@ output/
 | `model.onnx.data` | External weight data (only if model ≥ 100 MiB). Must stay alongside `model.onnx`. |
 | `winml_build_config.json` | The complete pipeline config used for this build (includes auto-discovered optimization flags). This file is a **reproducible pipeline specification** — check it into version control or feed it directly to `winml build -c` in a CI/CD pipeline to guarantee identical model processing across machines and runs (set `"auto": false` for fully deterministic builds). |
 | `analyze_result.json` | Static analysis output: EP compatibility, operator classification, detected patterns. |
-| `build_manifest.json` | Build provenance with stage timings. Only generated via the Python API (`build_hf_model`/`build_onnx_model`). |
+| `build_manifest.json` | Build provenance with stage timings. Generated by `winml build` and the Python API (`build_hf_model`/`build_onnx_model`). |
 | `export_htp_metadata.json` | HTP export metadata: module hierarchy, tracing info, tagging coverage. |
 
 ### Intermediate Files (Can Delete After Build)

src/winml/modelkit/build/hf.py

@@ -31,6 +31,7 @@
 from ..export import export_onnx
 from ..onnx import copy_onnx_model
 from ..quant import quantize_onnx
+from ..utils import MANIFEST_FILENAME, ManifestStage, WinMLManifest
 from .common import ensure_pre_quantized_stamped, run_optimize_analyze_loop
 
 
@@ -174,7 +175,7 @@ def _name(base: str) -> str:
     compiled_path = output_dir / _name("compiled.onnx")
     final_path = output_dir / _name("model.onnx")
     config_path = output_dir / _name("winml_build_config.json")
-    manifest_path = output_dir / _name("build_manifest.json")
+    manifest_path = output_dir / _name(MANIFEST_FILENAME)
     analyze_result_path = output_dir / _name("analyze_result.json")
 
     # Check for existing artifact (skip build if present and not rebuilding)
@@ -369,21 +370,7 @@ def _name(base: str) -> str:
     # =========================================================================
     # [7] BUILD MANIFEST — Machine-readable build provenance
     # =========================================================================
-    manifest: dict[str, Any] = {
-        "schema_version": 1,
-        "model_id": model_label,
-        "task": task,
-        "cache_key": cache_key,
-        "config_hash": cache_key.rsplit("_", 1)[-1] if cache_key and "_" in cache_key else None,
-        "timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(),
-        "elapsed_seconds": round(elapsed, 3),
-        "stages": [],
-        "final_artifact": final_path.name,
-        "analyze_iterations": analyze_iterations,
-        "analyze_unsupported_node_count": analyze_unsupported_nodes,
-        "analyze_details": analyze_details,
-    }
-
+    manifest_stages: list[ManifestStage] = []
     stage_filenames = {
         "export": export_path.name,
         "optimize": optimized_path.name,
@@ -392,33 +379,37 @@ def _name(base: str) -> str:
     }
     for stage_name in ["export", "optimize", "quantize", "compile"]:
         if stage_name in stages_completed:
-            entry: dict[str, Any] = {
-                "name": stage_name,
-                "status": "completed",
-                "filename": stage_filenames[stage_name],
-                "elapsed_seconds": round(stage_timings.get(stage_name, 0), 3),
-            }
+            stage = ManifestStage(
+                name=stage_name,
+                status="completed",
+                filename=stage_filenames[stage_name],
+                elapsed_seconds=round(stage_timings.get(stage_name, 0), 3),
+            )
             # Thread QuantizeResult metrics into manifest
             if stage_name == "quantize" and quant_result is not None:
-                entry["nodes_quantized"] = quant_result.nodes_quantized
-                entry["nodes_skipped"] = quant_result.nodes_skipped
-                entry["calibration_time_seconds"] = round(quant_result.calibration_time_seconds, 3)
-                entry["qdq_insertion_time_seconds"] = round(
-                    quant_result.qdq_insertion_time_seconds, 3
-                )
-            manifest["stages"].append(entry)
+                stage.nodes_quantized = quant_result.nodes_quantized
+                stage.nodes_skipped = quant_result.nodes_skipped
+                stage.calibration_time_seconds = round(quant_result.calibration_time_seconds, 3)
+                stage.qdq_insertion_time_seconds = round(quant_result.qdq_insertion_time_seconds, 3)
+            manifest_stages.append(stage)
         elif stage_name in stages_skipped:
-            manifest["stages"].append(
-                {
-                    "name": stage_name,
-                    "status": "skipped",
-                    "filename": None,
-                    "elapsed_seconds": None,
-                }
-            )
+            manifest_stages.append(ManifestStage(name=stage_name, status="skipped"))
 
-    manifest_path.write_text(json.dumps(manifest, indent=2))
-    logger.debug("Build manifest persisted: %s", manifest_path)
+    manifest = WinMLManifest(
+        source="hf",
+        model_id=model_label,
+        task=task,
+        cache_key=cache_key,
+        config_hash=cache_key.rsplit("_", 1)[-1] if cache_key and "_" in cache_key else None,
+        timestamp=datetime.datetime.now(datetime.timezone.utc).isoformat(),
+        elapsed_seconds=round(elapsed, 3),
+        final_artifact=final_path.name,
+        stages=manifest_stages,
+        analyze_iterations=analyze_iterations,
+        analyze_unsupported_node_count=analyze_unsupported_nodes,
+        analyze_details=analyze_details,
+    )
+    manifest.save(manifest_path)
 
     return BuildResult(
         output_dir=output_dir,

src/winml/modelkit/build/onnx.py

@@ -23,6 +23,7 @@
 from ..compiler import compile_onnx
 from ..onnx import copy_onnx_model
 from ..quant import quantize_onnx
+from ..utils import MANIFEST_FILENAME, ManifestStage, WinMLManifest
 from .common import ensure_pre_quantized_stamped, run_optimize_analyze_loop
 from .hf import BuildResult
 
@@ -116,7 +117,7 @@ def _name(base: str) -> str:
     compiled_path = output_dir / _name("compiled.onnx")
     final_path = output_dir / _name("model.onnx")
     config_path = output_dir / _name("winml_build_config.json")
-    manifest_path = output_dir / _name("build_manifest.json")
+    manifest_path = output_dir / _name(MANIFEST_FILENAME)
     analyze_result_path = output_dir / _name("analyze_result.json")
 
     # Check for existing artifact (skip build if present and not rebuilding)
@@ -276,53 +277,42 @@ def _name(base: str) -> str:
     # =========================================================================
     # [5] BUILD MANIFEST — Machine-readable build provenance
     # =========================================================================
-    manifest: dict[str, Any] = {
-        "schema_version": 1,
-        "source": "onnx",
-        "input_onnx": str(onnx_path),
-        "timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(),
-        "elapsed_seconds": round(elapsed, 3),
-        "stages": [],
-        "final_artifact": final_path.name,
-        "analyze_iterations": analyze_iters,
-        "analyze_unsupported_node_count": analyze_unsupported,
-        "analyze_details": analyze_details,
-    }
-
+    manifest_stages: list[ManifestStage] = []
     stage_filenames = {
         "optimize": optimized_path.name,
         "quantize": quantized_path.name,
         "compile": compiled_path.name,
     }
     for stage_name in ["optimize", "quantize", "compile"]:
         if stage_name in stages_completed:
-            entry: dict[str, Any] = {
-                "name": stage_name,
-                "status": "completed",
-                "filename": stage_filenames[stage_name],
-                "elapsed_seconds": round(stage_timings.get(stage_name, 0), 3),
-            }
+            stage = ManifestStage(
+                name=stage_name,
+                status="completed",
+                filename=stage_filenames[stage_name],
+                elapsed_seconds=round(stage_timings.get(stage_name, 0), 3),
+            )
             # Thread QuantizeResult metrics into manifest
             if stage_name == "quantize" and quant_result is not None:
-                entry["nodes_quantized"] = quant_result.nodes_quantized
-                entry["nodes_skipped"] = quant_result.nodes_skipped
-                entry["calibration_time_seconds"] = round(quant_result.calibration_time_seconds, 3)
-                entry["qdq_insertion_time_seconds"] = round(
-                    quant_result.qdq_insertion_time_seconds, 3
-                )
-            manifest["stages"].append(entry)
+                stage.nodes_quantized = quant_result.nodes_quantized
+                stage.nodes_skipped = quant_result.nodes_skipped
+                stage.calibration_time_seconds = round(quant_result.calibration_time_seconds, 3)
+                stage.qdq_insertion_time_seconds = round(quant_result.qdq_insertion_time_seconds, 3)
+            manifest_stages.append(stage)
         elif stage_name in stages_skipped:
-            manifest["stages"].append(
-                {
-                    "name": stage_name,
-                    "status": "skipped",
-                    "filename": None,
-                    "elapsed_seconds": None,
-                }
-            )
-
-    manifest_path.write_text(json.dumps(manifest, indent=2))
-    logger.debug("Build manifest persisted: %s", manifest_path)
+            manifest_stages.append(ManifestStage(name=stage_name, status="skipped"))
+
+    manifest = WinMLManifest(
+        source="onnx",
+        input_onnx=str(onnx_path),
+        timestamp=datetime.datetime.now(datetime.timezone.utc).isoformat(),
+        elapsed_seconds=round(elapsed, 3),
+        final_artifact=final_path.name,
+        stages=manifest_stages,
+        analyze_iterations=analyze_iters,
+        analyze_unsupported_node_count=analyze_unsupported,
+        analyze_details=analyze_details,
+    )
+    manifest.save(manifest_path)
 
     return BuildResult(
         output_dir=output_dir,

src/winml/modelkit/inference/engine.py

@@ -36,6 +36,7 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, cast
 
+from ..utils.manifest import MANIFEST_FILENAME, WinMLManifest
 from .tasks import BINARY_TYPES, TASK_REGISTRY, InputField, PipelineMapping
 from .types import Prediction, PredictionResult
 
@@ -150,7 +151,7 @@ def _find_build_artifacts(build_dir: Path, *, task: str | None = None) -> tuple[
     """
     # Try plain layout first (bare build output)
     plain_onnx = build_dir / "model.onnx"
-    plain_manifest = build_dir / "build_manifest.json"
+    plain_manifest = build_dir / MANIFEST_FILENAME
     if plain_onnx.exists():
         manifest = json.loads(plain_manifest.read_text()) if plain_manifest.exists() else None
         if task is None or manifest is None or manifest.get("task") == task:
@@ -161,7 +162,7 @@ def _find_build_artifacts(build_dir: Path, *, task: str | None = None) -> tuple[
     candidates: list[tuple[Path, dict | None]] = []
     for onnx_path in sorted(build_dir.glob("*_model.onnx")):
         prefix = onnx_path.name.rsplit("_model.onnx", 1)[0]
-        manifest_path = build_dir / f"{prefix}_build_manifest.json"
+        manifest_path = build_dir / f"{prefix}_{MANIFEST_FILENAME}"
         manifest = json.loads(manifest_path.read_text()) if manifest_path.exists() else None
         if task is not None:
             if manifest is None or manifest.get("task") == task:
@@ -977,11 +978,9 @@ def _load_from_build_dir(
     @staticmethod
     def _resolve_model_id_from_dir(build_dir: Path) -> str | None:
         """Extract model_id from any manifest in the directory (task-agnostic)."""
-        for manifest_path in build_dir.glob("*build_manifest.json"):
-            manifest: dict[str, Any] = json.loads(manifest_path.read_text())
-            model_id = manifest.get("model_id")
-            if model_id:
-                return str(model_id)
+        for m in WinMLManifest.find(build_dir):
+            if m.model_id:
+                return m.model_id
         return None
 
     def _load_from_onnx(

src/winml/modelkit/inspect/resolver.py

@@ -493,6 +493,7 @@ def resolve_cache(model_id: str) -> CacheInfo:
     import json
 
     from ..cache import get_cache_dir, get_model_dir
+    from ..utils.manifest import MANIFEST_FILENAME, WinMLManifest
 
     cache_dir = get_cache_dir()
     model_dir = get_model_dir(model_id, cache_dir=cache_dir)
@@ -508,18 +509,20 @@ def resolve_cache(model_id: str) -> CacheInfo:
     # PRIMARY: Manifest-based resolution
     # -------------------------------------------------------------------------
     if model_dir.exists():
-        manifests = list(model_dir.glob("*build_manifest.json"))
-        if manifests:
-            # Use the most recent manifest (by mtime) when multiple variants exist
-            manifest_path = max(manifests, key=lambda p: p.stat().st_mtime)
+        manifest_paths = sorted(
+            model_dir.glob(f"*{MANIFEST_FILENAME}"),
+            key=lambda p: p.stat().st_mtime,
+            reverse=True,
+        )
+        if manifest_paths:
             try:
-                manifest = json.loads(manifest_path.read_text())
-                manifest_stages = {s["name"]: s for s in manifest.get("stages", [])}
+                manifest = WinMLManifest.load(manifest_paths[0])
+                manifest_stages = {s.name: s for s in manifest.stages}
 
                 for stage in pipeline_stages:
                     ms = manifest_stages.get(stage)
-                    if ms and ms.get("status") == "completed":
-                        filename = ms.get("filename")
+                    if ms and ms.status == "completed":
+                        filename = ms.filename
                         artifact = model_dir / filename if filename else None
                         size_bytes = (
                             artifact.stat().st_size if artifact and artifact.exists() else 0
@@ -543,8 +546,8 @@ def resolve_cache(model_id: str) -> CacheInfo:
                     total_cached=total_cached,
                     total_size_mb=round(total_size_mb, 2),
                 )
-            except (json.JSONDecodeError, KeyError, OSError) as exc:
-                logger.debug("Failed to read manifest %s: %s", manifest_path, exc)
+            except (json.JSONDecodeError, KeyError, TypeError, ValueError, OSError) as exc:
+                logger.debug("Failed to read manifest %s: %s", manifest_paths[0], exc)
                 # Fall through to filename scanning
                 stages = []
                 total_cached = 0

src/winml/modelkit/serve/app.py

@@ -741,11 +741,13 @@ async def cli_command(command: str, request: CliRequest) -> CliResponse:
 def _manifest_from_engine(engine: InferenceEngine) -> dict[str, Any]:
     """Build manifest dict from engine, trying build_manifest.json first."""
     if engine.model_path:
-        manifest_file = Path(engine.model_path) / "build_manifest.json"
+        from ..utils.manifest import MANIFEST_FILENAME, WinMLManifest
+
+        manifest_file = Path(engine.model_path) / MANIFEST_FILENAME
         if manifest_file.exists():
             try:
-                return cast("dict[str, Any]", json.loads(manifest_file.read_text()))
-            except (json.JSONDecodeError, OSError) as e:
+                return WinMLManifest.load(manifest_file).to_dict()
+            except (json.JSONDecodeError, TypeError, ValueError, OSError) as e:
                 logger.warning("Failed to load manifest: %s", e)
 
     return {

src/winml/modelkit/utils/__init__.py

@@ -12,6 +12,7 @@
     load_hf_components_from_onnx,
     save_local_model_configs,
 )
+from .manifest import MANIFEST_FILENAME, ManifestStage, WinMLManifest
 from .model_input import (
     ModelInput,
     ModelInputKind,
@@ -21,8 +22,11 @@
 
 
 __all__ = [
+    "MANIFEST_FILENAME",
+    "ManifestStage",
     "ModelInput",
     "ModelInputKind",
+    "WinMLManifest",
     "classify_model_input",
     "inject_hub_metadata",
     "is_hub_model",