Coverage for / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / tools / model_integration / manifest.py: 80%
194 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-10-25 05:45 +0800
1# Copyright 2026 Huawei Technologies Co., Ltd
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14# ============================================================================
15"""Strict local-only manifest loading for integration and validation runs."""
17from __future__ import annotations
19import copy
20import hashlib
21import importlib
22import json
23from collections.abc import Mapping
24from dataclasses import dataclass
25from pathlib import Path
26from typing import Any, Callable, Optional
28import yaml
30from hyper_parallel.tools.model_integration.schemas import SCHEMA_VERSION
33class ManifestError(ValueError):
34 """Raised when a model-integration manifest violates its public schema."""
37def _mapping(value: Any, path: str, required: bool = True) -> dict[str, Any]:
38 if value is None and not required:
39 return {}
40 if not isinstance(value, Mapping):
41 raise ManifestError(f"{path} must be a mapping")
42 return dict(value)
45def _local_path(value: Any, base_dir: Path, field_path: str) -> Optional[Path]:
46 if value is None:
47 return None
48 if not isinstance(value, str) or not value:
49 raise ManifestError(f"{field_path} must be a non-empty local path or null")
50 if value.startswith(("http://", "https://", "hf://")):
51 raise ManifestError(
52 f"{field_path} must be local; model-integration tools never download implicitly"
53 )
54 path = Path(value).expanduser()
55 return (base_dir / path).resolve() if not path.is_absolute() else path.resolve()
58@dataclass(frozen=True)
59class ValidationManifest:
60 """Resolved manifest with path normalization and immutable raw content."""
62 path: Path
63 raw: dict[str, Any]
64 model: dict[str, Any]
65 reference: dict[str, Any]
66 matrix: dict[str, Any]
67 acceptance: dict[str, Any]
68 launcher: dict[str, Any]
69 output_dir: Path
71 @property
72 def family(self) -> str:
73 """Return the adapter/model family lookup key."""
74 family = self.model.get("adapter") or self.model.get("model_type")
75 if not isinstance(family, str) or not family:
76 raise ManifestError("model.adapter or model.model_type must be a non-empty string")
77 return family
79 @property
80 def model_path(self) -> Optional[Path]:
81 """Return the local authoritative model asset path when declared."""
82 return _local_path(self.model.get("id_or_path"), self.path.parent, "model.id_or_path")
84 @property
85 def reference_path(self) -> Optional[Path]:
86 """Return the local reference repository path when declared."""
87 return _local_path(self.reference.get("source_path"), self.path.parent, "reference.source_path")
89 @property
90 def integration_handoff_path(self) -> Optional[Path]:
91 """Return the prior integration handoff required by precision validation."""
92 return _local_path(
93 self.raw.get("integration_handoff"),
94 self.path.parent,
95 "integration_handoff",
96 )
98 @property
99 def launcher_config_path(self) -> Optional[Path]:
100 """Return the Trainer recipe path relative to the validation manifest."""
101 return _local_path(
102 self.launcher.get("config"),
103 self.path.parent,
104 "launcher.config",
105 )
107 def to_dict(self) -> dict[str, Any]:
108 """Return a deep copy safe for evidence serialization."""
109 payload = copy.deepcopy(self.raw)
110 payload["output_dir"] = str(self.output_dir)
111 return payload
113 def import_callable(self, target: str) -> Callable[..., Any]:
114 """Import an explicit dotted callable without mutating ``sys.path``."""
115 if not isinstance(target, str) or "." not in target:
116 raise ManifestError(f"callable target must be a dotted path, got {target!r}")
117 module_name, attribute_name = target.rsplit(".", 1)
118 try:
119 value = getattr(importlib.import_module(module_name), attribute_name)
120 except (ImportError, AttributeError) as exc:
121 raise ManifestError(f"could not import callable {target!r}: {exc}") from exc
122 if not callable(value):
123 raise ManifestError(f"manifest target {target!r} is not callable")
124 return value
126 def fingerprint_paths(self) -> dict[str, Any]:
127 """Hash bounded metadata from declared model and reference sources."""
128 fingerprints = {}
129 path_fields = {
130 "model.id_or_path": self.model.get("id_or_path"),
131 "reference.source_path": self.reference.get("source_path"),
132 }
133 for field_path, value in path_fields.items():
134 path = _local_path(value, self.path.parent, field_path)
135 if path is None:
136 continue
137 if not path.exists():
138 fingerprints[field_path] = {"path": str(path), "exists": False}
139 continue
140 if path.is_dir():
141 metadata_names = {
142 "config.json",
143 "generation_config.json",
144 "preprocessor_config.json",
145 "processor_config.json",
146 "special_tokens_map.json",
147 "tokenizer.json",
148 "tokenizer_config.json",
149 }
150 metadata_files = [
151 candidate
152 for candidate in path.iterdir()
153 if candidate.is_file()
154 and (candidate.name in metadata_names or candidate.name.endswith(".index.json"))
155 ]
156 for relative_name in (
157 "inference/model.py",
158 "inference/engram.py",
159 "inference/vision.py",
160 ):
161 candidate = path / relative_name
162 if candidate.is_file():
163 metadata_files.append(candidate)
164 metadata_files.sort(key=lambda candidate: str(candidate.relative_to(path)))
165 fingerprints[field_path] = {
166 "path": str(path),
167 "exists": True,
168 "kind": "directory",
169 "metadata": {
170 str(candidate.relative_to(path)): self._fingerprint_file(candidate)
171 for candidate in metadata_files
172 },
173 }
174 continue
175 fingerprints[field_path] = {
176 "path": str(path),
177 "exists": True,
178 "kind": "file",
179 **self._fingerprint_file(path),
180 }
181 return fingerprints
183 @staticmethod
184 def _fingerprint_file(path: Path) -> dict[str, Any]:
185 """Return a stable SHA-256 and size for one explicit local file."""
186 digest = hashlib.sha256()
187 with path.open("rb") as input_file:
188 for chunk in iter(lambda: input_file.read(1024 * 1024), b""):
189 digest.update(chunk)
190 return {"sha256": digest.hexdigest(), "size": path.stat().st_size}
192 def _require_matrix_steps(self) -> None:
193 """Require an explicit baseline and a positive step budget."""
194 if not self.matrix.get("baseline"):
195 raise ManifestError("matrix.baseline is required for validate")
196 steps = self.matrix.get("steps")
197 if not isinstance(steps, int) or steps <= 0:
198 raise ManifestError("matrix.steps must be a positive integer")
200 def _require_launcher_contract(self) -> None:
201 """Require an existing local Trainer config and a module path."""
202 trainer_config = self.launcher.get("config")
203 trainer_module = self.launcher.get("module")
204 if not isinstance(trainer_config, str) or not trainer_config:
205 raise ManifestError("launcher.config must be a non-empty path")
206 trainer_config_path = self.launcher_config_path
207 if trainer_config_path is None or not trainer_config_path.is_file():
208 raise ManifestError(
209 f"launcher.config does not exist: {trainer_config_path}"
210 )
211 if not isinstance(trainer_module, str) or not trainer_module:
212 raise ManifestError("launcher.module must be a non-empty dotted module")
214 def _require_matrix_resources(self) -> None:
215 """Require a positive device count and a bool warm-start flag."""
216 devices = self.matrix.get("devices")
217 if not isinstance(devices, int) or devices <= 0:
218 raise ManifestError("matrix.devices must be a positive integer")
219 shared_initial_checkpoint = self.matrix.get("shared_initial_checkpoint", False)
220 if not isinstance(shared_initial_checkpoint, bool):
221 raise ManifestError("matrix.shared_initial_checkpoint must be a bool")
223 def _require_launcher_timeout(self) -> None:
224 """Require any declared launcher timeout to be positive."""
225 timeout = self.launcher.get("timeout_seconds")
226 if timeout is not None and (
227 isinstance(timeout, bool)
228 or not isinstance(timeout, (int, float))
229 or timeout <= 0
230 ):
231 raise ManifestError("launcher.timeout_seconds must be positive")
233 def _require_resume_split_step(self) -> None:
234 """Bound the resume split step whenever resume cases are declared."""
235 steps = self.matrix.get("steps")
236 has_resume_case = bool(
237 self.matrix.get("same_topology_resume")
238 or self.matrix.get("cross_topology_resume")
239 )
240 split_step = self.matrix.get("resume_split_step")
241 if has_resume_case and (
242 isinstance(split_step, bool)
243 or not isinstance(split_step, int)
244 or split_step <= 0
245 or split_step >= steps
246 ):
247 raise ManifestError(
248 "resume cases require 0 < matrix.resume_split_step < matrix.steps"
249 )
250 shared_initial_checkpoint = self.matrix.get("shared_initial_checkpoint", False)
251 if has_resume_case and shared_initial_checkpoint and split_step <= 1:
252 raise ManifestError(
253 "resume cases with matrix.shared_initial_checkpoint=true require "
254 "matrix.resume_split_step > 1 because the shared warm-start is global step 1"
255 )
257 def _require_integration_handoff(self) -> None:
258 """Require a PASS handoff for the same adapter family when declared."""
259 handoff = self.integration_handoff_path
260 if handoff is not None and not handoff.is_file():
261 raise ManifestError(f"integration_handoff does not exist: {handoff}")
262 if handoff is None:
263 return
264 try:
265 handoff_payload = yaml.safe_load(handoff.read_text(encoding="utf-8"))
266 except (OSError, yaml.YAMLError) as exc:
267 raise ManifestError(f"could not parse integration_handoff: {exc}") from exc
268 if not isinstance(handoff_payload, Mapping):
269 raise ManifestError("integration_handoff must contain a mapping")
270 if handoff_payload.get("schema_version") != SCHEMA_VERSION:
271 raise ManifestError(
272 f"integration_handoff schema_version must be {SCHEMA_VERSION}"
273 )
274 if handoff_payload.get("status") != "PASS":
275 raise ManifestError("integration_handoff status must be PASS")
276 if handoff_payload.get("family") != self.family:
277 raise ManifestError(
278 "integration_handoff family does not match model adapter"
279 )
281 def validate_for(self, command: str) -> None:
282 """Validate command-specific fields without silently adding defaults."""
283 _ = self.model_path
284 _ = self.reference_path
285 _ = self.integration_handoff_path
286 if command == "parity" and not self.reference:
287 raise ManifestError(f"reference is required for {command}")
288 if command != "validate":
289 return
290 self._require_matrix_steps()
291 self._require_launcher_contract()
292 self._require_matrix_resources()
293 self._require_launcher_timeout()
294 self._require_resume_split_step()
295 self._require_integration_handoff()
298def load_manifest(path: str | Path, output_dir: str | Path | None = None) -> ValidationManifest:
299 """Load and validate a local YAML/JSON integration manifest.
301 Args:
302 path: Manifest path.
303 output_dir: Optional CLI-only evidence directory override.
305 Returns:
306 Resolved immutable manifest.
308 Raises:
309 ManifestError: If schema, types, or local-only constraints are invalid.
310 """
311 manifest_path = Path(path).expanduser().resolve()
312 if not manifest_path.is_file():
313 raise ManifestError(f"manifest does not exist: {manifest_path}")
314 try:
315 if manifest_path.suffix.lower() == ".json":
316 loaded = json.loads(manifest_path.read_text(encoding="utf-8"))
317 else:
318 loaded = yaml.safe_load(manifest_path.read_text(encoding="utf-8"))
319 except (OSError, json.JSONDecodeError, yaml.YAMLError) as exc:
320 raise ManifestError(f"could not parse manifest {manifest_path}: {exc}") from exc
321 raw = _mapping(loaded, "$")
322 if raw.get("schema_version") != SCHEMA_VERSION:
323 raise ManifestError(f"schema_version must be {SCHEMA_VERSION}")
324 model = _mapping(raw.get("model"), "model")
325 if not (model.get("adapter") or model.get("model_type")):
326 raise ManifestError("model.adapter or model.model_type is required")
327 reference = _mapping(raw.get("reference"), "reference", required=False)
328 matrix = _mapping(raw.get("matrix"), "matrix", required=False)
329 acceptance = _mapping(raw.get("acceptance"), "acceptance", required=False)
330 launcher = _mapping(raw.get("launcher"), "launcher", required=False)
331 configured_output = output_dir or raw.get("output_dir") or (
332 manifest_path.parent / "output" / "model_integration" / str(model.get("adapter"))
333 )
334 resolved_output = Path(configured_output).expanduser()
335 if not resolved_output.is_absolute():
336 resolved_output = (manifest_path.parent / resolved_output).resolve()
337 else:
338 resolved_output = resolved_output.resolve()
339 return ValidationManifest(
340 path=manifest_path,
341 raw=raw,
342 model=model,
343 reference=reference,
344 matrix=matrix,
345 acceptance=acceptance,
346 launcher=launcher,
347 output_dir=resolved_output,
348 )
351__all__ = ["ManifestError", "ValidationManifest", "load_manifest"]