Coverage for / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / tools / model_integration / checkpoint_coverage.py: 69%
75 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"""Complete checkpoint coverage reporting for finalized model state."""
17from __future__ import annotations
19import fnmatch
20import re
21from typing import Any, Optional
23from hyper_parallel.tools.model_integration.schemas import (
24 FindingSeverity,
25 ModelIntegrationFinding,
26)
27from hyper_parallel.models.validation_spec import CheckpointValidationSpec
30def _matches_target(transform: Any, target_name: str) -> bool:
31 """Match processed WeightTransform regexes, not adapter-owned FQN globs."""
32 scope = getattr(transform, "scope_prefix", None)
33 scoped_name = target_name
34 if scope and target_name.startswith(f"{scope}."):
35 scoped_name = target_name.removeprefix(f"{scope}.")
36 for pattern in getattr(transform, "target_patterns", ()):
37 normalized = str(pattern).replace(".*.", ".0.")
38 if re.search(normalized, scoped_name):
39 return True
40 return False
43def _matches_any(name: str, patterns: tuple[str, ...]) -> bool:
44 return any(fnmatch.fnmatchcase(name, pattern) for pattern in patterns)
47def _non_reversible_transforms(transforms: tuple[Any, ...]) -> list[str]:
48 """Probe each declared transform for an executable reverse transform."""
49 non_reversible = []
50 for transform in transforms:
51 reverse_transform = getattr(transform, "reverse_transform", None)
52 if not callable(reverse_transform):
53 non_reversible.append(type(transform).__name__)
54 continue
55 try:
56 reverse_transform()
57 except (AttributeError, NotImplementedError, TypeError, ValueError):
58 non_reversible.append(type(transform).__name__)
59 return non_reversible
62def _target_category(
63 target_name: str,
64 matching: list[Any],
65 loaded: set[str],
66 intentional_missing: set[str],
67 unsupported_missing: set[str],
68) -> str:
69 """Classify one finalized persistent target key."""
70 if target_name in loaded:
71 if not matching:
72 return "direct_load"
73 if any(hasattr(transform, "operations") for transform in matching):
74 return "reversible_transform"
75 return "same_shape_rename"
76 if target_name in intentional_missing:
77 return "training_only_initialization"
78 if target_name in unsupported_missing:
79 return "unsupported_missing_target"
80 return "alias_or_non_persistent"
83def _coverage_entries(
84 model: Any,
85 transforms: tuple[Any, ...],
86 loaded: set[str],
87 intentional_missing: set[str],
88 unsupported_missing: set[str],
89 unexpected: set[str],
90 inference_only: set[str],
91) -> list[dict[str, Any]]:
92 """Build one coverage entry per target and per unexpected source key."""
93 entries = []
94 for target_name in sorted(model.state_dict()):
95 matching = [transform for transform in transforms if _matches_target(transform, target_name)]
96 entries.append(
97 {
98 "target_key": target_name,
99 "category": _target_category(
100 target_name,
101 matching,
102 loaded,
103 intentional_missing,
104 unsupported_missing,
105 ),
106 "transforms": [type(transform).__name__ for transform in matching],
107 }
108 )
109 for source_name in sorted(unexpected):
110 entries.append(
111 {
112 "source_key": source_name,
113 "category": (
114 "inference_only_exclusion"
115 if source_name in inference_only
116 else "unsupported_architectural_mismatch"
117 ),
118 }
119 )
120 return entries
123def _coverage_findings(
124 unsupported_missing: set[str],
125 unsupported_unexpected: set[str],
126 non_reversible: list[str],
127) -> list[ModelIntegrationFinding]:
128 """Emit ERROR findings for unsupported keys and one-way transforms."""
129 findings = []
130 if unsupported_missing or unsupported_unexpected:
131 findings.append(
132 ModelIntegrationFinding(
133 phase="D5",
134 owner_fqn="<root>",
135 severity=FindingSeverity.ERROR,
136 message="checkpoint coverage is incomplete after finalization",
137 facts={
138 "missing_targets": sorted(unsupported_missing),
139 "unexpected_sources": sorted(unsupported_unexpected),
140 },
141 why_unsafe=(
142 "strict=False logs do not prove that every authoritative tensor has an "
143 "intentional load, transform, initialization, or exclusion policy"
144 ),
145 remediation=(
146 "add a checkpoint mapping/transform or an explicit training-only/inference-only "
147 "declaration, then rebuild with strict finalization"
148 ),
149 related_config=("model.load_base_model", "model adapter checkpoint"),
150 )
151 )
152 if non_reversible:
153 findings.append(
154 ModelIntegrationFinding(
155 phase="D5",
156 owner_fqn="<root>",
157 severity=FindingSeverity.ERROR,
158 message="checkpoint transform has no executable reverse transform",
159 facts={"transforms": non_reversible},
160 why_unsafe=(
161 "a one-way transform cannot preserve the authoritative namespace and values "
162 "when exporting or validating a source-target-source round trip"
163 ),
164 remediation="provide inverse conversion operations and add a tensor round-trip parity case",
165 related_config=("model adapter checkpoint",),
166 )
167 )
168 return findings
171def build_checkpoint_coverage(
172 model: Any,
173 validation_spec: Optional[CheckpointValidationSpec] = None,
174) -> tuple[dict[str, Any], list[ModelIntegrationFinding]]:
175 """Classify every finalized persistent state key using the last load report."""
176 report = getattr(model, "_hp_checkpoint_load_report", None)
177 if report is None:
178 return {
179 "status": "NOT_LOADED",
180 "entries": [],
181 "reason": "model was initialized from scratch or no load report was retained",
182 }, []
183 loaded = set(getattr(report, "loaded_keys", ()))
184 missing = set(getattr(report, "missing_keys", ()))
185 unexpected = set(getattr(report, "unexpected_keys", ()))
186 validation_spec = validation_spec or CheckpointValidationSpec()
187 intentional_missing = {
188 name
189 for name in missing
190 if _matches_any(name, validation_spec.training_only_target_patterns)
191 }
192 inference_only = {
193 name
194 for name in unexpected
195 if _matches_any(name, validation_spec.inference_only_source_patterns)
196 }
197 unsupported_missing = missing - intentional_missing
198 unsupported_unexpected = unexpected - inference_only
199 transforms = tuple(getattr(model, "_weight_conversions", ()) or ())
200 non_reversible = _non_reversible_transforms(transforms)
201 entries = _coverage_entries(
202 model,
203 transforms,
204 loaded,
205 intentional_missing,
206 unsupported_missing,
207 unexpected,
208 inference_only,
209 )
210 findings = _coverage_findings(
211 unsupported_missing,
212 unsupported_unexpected,
213 non_reversible,
214 )
215 return {
216 "status": "FAIL" if findings else "PASS",
217 "loaded": len(loaded),
218 "missing": len(missing),
219 "unexpected": len(unexpected),
220 "intentional_training_only": len(intentional_missing),
221 "intentional_inference_only": len(inference_only),
222 "entries": entries,
223 }, findings
226__all__ = ["build_checkpoint_coverage"]