Coverage for  / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / tools / model_integration / data_contract.py: 57%

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"""Data/forward-input ownership and semantic diagnostics.""" 

16 

17from __future__ import annotations 

18 

19import fnmatch 

20from collections.abc import Mapping 

21from typing import Any, Optional 

22 

23from hyper_parallel.tools.model_integration.schemas import ( 

24 FindingSeverity, 

25 ModelIntegrationFinding, 

26) 

27from hyper_parallel.models.validation_spec import ModelValidationSpec 

28 

29 

30_FRAMEWORK_FIELDS = { 

31 "input_ids", 

32 "labels", 

33 "shift_labels", 

34} 

35 

36 

37def build_field_ownership( 

38 runtime_adapter: Any, 

39) -> dict[str, str]: 

40 """Build declared field ownership without executing another data batch. 

41 

42 Args: 

43 runtime_adapter: Optional model-owned runtime-input adapter. 

44 

45 Returns: 

46 Declared field owners and any duplicate ownership claims. 

47 """ 

48 ownership = {name: "framework" for name in sorted(_FRAMEWORK_FIELDS)} 

49 runtime_fields = getattr(runtime_adapter, "runtime_input_fields", lambda: ())() 

50 collisions = {} 

51 for field_name in runtime_fields: 

52 previous = ownership.get(field_name) 

53 if previous is not None: 

54 collisions.setdefault(field_name, [previous]).append("runtime_adapter") 

55 else: 

56 ownership[field_name] = "runtime_adapter" 

57 return {"fields": ownership, "collisions": collisions} 

58 

59 

60def validate_data_contract( 

61 validation_spec: Optional[ModelValidationSpec], 

62 runtime_adapter: Any, 

63 *, 

64 labels_are_shifted: Optional[bool], 

65 source_type: Optional[str], 

66 assets: tuple[Any, ...] = (), 

67) -> tuple[dict[str, Any], list[ModelIntegrationFinding]]: 

68 """Validate stable adapter ownership and model-declared data semantics. 

69 

70 Args: 

71 validation_spec: Optional model-owned validation declaration. 

72 runtime_adapter: Optional model-owned runtime-input adapter. 

73 labels_are_shifted: Resolved get-batch label alignment, when declared. 

74 source_type: Resolved get-batch source type, when declared. 

75 assets: Tokenizer or processor assets used by the data pipeline. 

76 

77 Returns: 

78 Field-ownership evidence and actionable validation findings. 

79 """ 

80 ownership = build_field_ownership(runtime_adapter) 

81 findings = [] 

82 if ownership["collisions"]: 

83 findings.append( 

84 ModelIntegrationFinding( 

85 phase="D2", 

86 owner_fqn="data.forward_fields", 

87 severity=FindingSeverity.ERROR, 

88 message="multiple data stages claim the same model forward field", 

89 facts={"collisions": ownership["collisions"]}, 

90 why_unsafe="a later stage can silently replace authoritative batch semantics", 

91 remediation="assign every forward field to exactly one framework or adapter owner", 

92 related_config=("dataset.data_transform", "dataloader.get_batch"), 

93 ) 

94 ) 

95 data_spec = validation_spec.data if validation_spec is not None else None 

96 if data_spec is not None: 

97 declared_runtime = set(ownership["fields"]) 

98 missing_runtime = sorted(set(data_spec.runtime_fields) - declared_runtime) 

99 shift_mismatch = ( 

100 data_spec.labels_are_shifted is not None 

101 and labels_are_shifted is not None 

102 and data_spec.labels_are_shifted != labels_are_shifted 

103 ) 

104 missing_processor_revision = data_spec.processor_revision_required and not any( 

105 getattr(asset, "name_or_path", None) 

106 or getattr(asset, "_name_or_path", None) 

107 for asset in assets 

108 ) 

109 if missing_runtime or shift_mismatch or missing_processor_revision: 

110 findings.append( 

111 ModelIntegrationFinding( 

112 phase="D2", 

113 owner_fqn="data.semantic_contract", 

114 severity=FindingSeverity.ERROR, 

115 message="resolved data pipeline disagrees with the model validation contract", 

116 facts={ 

117 "missing_runtime_fields": missing_runtime, 

118 "expected_labels_are_shifted": data_spec.labels_are_shifted, 

119 "actual_labels_are_shifted": labels_are_shifted, 

120 "processor_revision_missing": missing_processor_revision, 

121 }, 

122 why_unsafe=( 

123 "label shift, packed boundaries, modality metadata, and processor revision " 

124 "change the mathematical training input" 

125 ), 

126 remediation=( 

127 "fix the dataset/get_batch recipe or adapter declarations before using " 

128 "the run for precision conclusions" 

129 ), 

130 related_config=("dataset", "dataloader", "model adapter validation"), 

131 ) 

132 ) 

133 ownership.update( 

134 { 

135 "source_type": source_type, 

136 "labels_are_shifted": labels_are_shifted, 

137 } 

138 ) 

139 return ownership, findings 

140 

141 

142def validate_observed_forward_fields( 

143 validation_spec: Optional[ModelValidationSpec], 

144 model_inputs: Mapping[str, Any], 

145) -> list[ModelIntegrationFinding]: 

146 """Validate the actual first-step kwargs against required model semantics.""" 

147 data_spec = validation_spec.data if validation_spec is not None else None 

148 if data_spec is None: 

149 return [] 

150 missing = sorted(set(data_spec.required_forward_fields) - set(model_inputs)) 

151 modality_present = sorted(set(data_spec.modality_fields).intersection(model_inputs)) 

152 if not missing: 

153 return [] 

154 return [ 

155 ModelIntegrationFinding( 

156 phase="D7", 

157 owner_fqn="model.forward", 

158 severity=FindingSeverity.ERROR, 

159 message="actual model forward kwargs omit required training fields", 

160 facts={ 

161 "missing": missing, 

162 "observed": sorted(model_inputs), 

163 "modality_fields_present": modality_present, 

164 }, 

165 why_unsafe="the executed model input is not the declared mathematical training sample", 

166 remediation="fix transform/collation/runtime-input ownership for the missing fields", 

167 related_config=("dataset", "dataloader.get_batch"), 

168 ) 

169 ] 

170 

171 

172def modality_gradient_parameters( 

173 validation_spec: Optional[ModelValidationSpec], 

174 model: Any, 

175) -> dict[str, bool]: 

176 """Return whether every declared multimodal group has a finite gradient.""" 

177 data_spec = validation_spec.data if validation_spec is not None else None 

178 if data_spec is None: 

179 return {} 

180 result = {} 

181 named_parameters = tuple(model.named_parameters(remove_duplicate=False)) 

182 for pattern in data_spec.modality_parameter_patterns: 

183 matched = [ 

184 parameter 

185 for parameter_name, parameter in named_parameters 

186 if parameter.requires_grad and fnmatch.fnmatchcase(parameter_name, pattern) 

187 ] 

188 finite_gradients = [] 

189 for parameter in matched: 

190 gradient = getattr(parameter, "main_grad", None) 

191 if gradient is None: 

192 gradient = parameter.grad 

193 local = ( 

194 gradient.to_local() 

195 if callable(getattr(gradient, "to_local", None)) 

196 else gradient 

197 ) 

198 if local is not None: 

199 finite_gradients.append(bool(local.isfinite().all())) 

200 result[pattern] = bool(finite_gradients) and all(finite_gradients) 

201 return result 

202 

203 

204__all__ = [ 

205 "build_field_ownership", 

206 "modality_gradient_parameters", 

207 "validate_data_contract", 

208 "validate_observed_forward_fields", 

209]