Code / libs/huleedu_nlp_shared/src/huleedu_nlp_shared/feature_contracts/provenance.py
libs/huleedu_nlp_shared/src/huleedu_nlp_shared/feature_contracts/provenance.py
1
"""Feature contract hash and identity helpers.
2
3
Purpose:
4
Build deterministic identities for feature schemas, input contracts,
5
runtime fingerprints, and concrete artifacts without importing extractor
6
runtimes.
7
8
Relationships:
9
- Used by `feature_contracts.current` to stamp the canonical registry.
10
- Used by artifact manifests and future adapters to keep schema, input,
11
runtime, and payload identities separate.
12
"""
13
14
from __future__ import annotations
15
16
import hashlib
17
import json
18
from collections.abc import Sequence
19
20
from pydantic import BaseModel
21
22
from huleedu_nlp_shared.feature_contracts.models import (
23
FeatureFamilySpecV1,
24
FeatureSetSpecV1,
25
)
26
27
28
def stable_sha256(payload: object) -> str:
29
"""Return a deterministic SHA-256 digest for a JSON-compatible payload."""
30
31
encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
32
return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
33
34
35
def model_sha256(model: BaseModel) -> str:
36
"""Return a deterministic digest for a Pydantic contract model."""
37
38
return stable_sha256(model.model_dump(mode="json", exclude_none=True))
39
40
41
def schema_hash_for_parts(
42
*,
43
registry_id: str,
44
registry_version: str,
45
feature_set_id: str,
46
feature_set_version: str,
47
family_parts: Sequence[dict[str, object]],
48
ordered_feature_names: Sequence[str],
49
) -> str:
50
"""Hash feature names, order, family versions, and meaning versions."""
51
52
return stable_sha256(
53
{
54
"registry_id": registry_id,
55
"registry_version": registry_version,
56
"feature_set_id": feature_set_id,
57
"feature_set_version": feature_set_version,
58
"families": list(family_parts),
59
"ordered_feature_names": list(ordered_feature_names),
60
}
61
)
62
63
64
def feature_set_family_parts(
65
families: Sequence[FeatureFamilySpecV1],
66
) -> tuple[dict[str, object], ...]:
67
"""Return schema-hash payload parts for ordered families."""
68
69
parts: list[dict[str, object]] = []
70
for family in families:
71
if family.generated_vector is not None:
72
vector = family.generated_vector
73
parts.append(
74
{
75
"family_id": family.family_id,
76
"family_version": family.family_version,
77
"value_kind": family.value_kind,
78
"meaning_version": vector.meaning_version,
79
"vector_id": vector.vector_id,
80
"name_prefix": vector.name_prefix,
81
"default_dim": vector.default_dim,
82
"index_width": vector.index_width,
83
"dtype": vector.dtype,
84
"model_name": vector.model_name,
85
"model_revision": vector.model_revision,
86
"tokenizer_revision": vector.tokenizer_revision,
87
"pooling": vector.pooling,
88
"max_length": vector.max_length,
89
"normalization_policy": vector.normalization_policy,
90
"truncate_policy": vector.truncate_policy,
91
"runtime_requirements": [
92
runtime.model_dump(mode="json", exclude_none=True)
93
for runtime in family.runtime_requirements
94
],
95
}
96
)
97
continue
98
parts.append(
99
{
100
"family_id": family.family_id,
101
"family_version": family.family_version,
102
"value_kind": family.value_kind,
103
"features": [
104
{
105
"name": feature.name,
106
"dtype": feature.dtype,
107
"unit": feature.unit,
108
"order": feature.order,
109
"input_role_ids": list(feature.input_role_ids),
110
"denominator_role_id": feature.denominator_role_id,
111
"meaning_version": feature.meaning_version,
112
}
113
for feature in family.features
114
],
115
"runtime_requirements": [
116
runtime.model_dump(mode="json", exclude_none=True)
117
for runtime in family.runtime_requirements
118
],
119
}
120
)
121
return tuple(parts)
122
123
124
def input_contract_hash_for_families(families: Sequence[FeatureFamilySpecV1]) -> str:
125
"""Hash input roles, role bindings, degradation, and stream contracts."""
126
127
return stable_sha256(
128
{
129
"families": [
130
{
131
"family_id": family.family_id,
132
"family_version": family.family_version,
133
"degradation_policy": family.degradation_policy,
134
"roles": [
135
role.model_dump(mode="json", exclude_none=True)
136
for role in family.input_roles
137
],
138
"feature_bindings": [
139
{
140
"name": feature.name,
141
"input_role_ids": list(feature.input_role_ids),
142
"denominator_role_id": feature.denominator_role_id,
143
}
144
for feature in family.features
145
],
146
"generated_vector_input_role": (
147
family.generated_vector.input_role_id
148
if family.generated_vector is not None
149
else ""
150
),
151
"runtime_requirements": [
152
runtime.model_dump(mode="json", exclude_none=True)
153
for runtime in family.runtime_requirements
154
],
155
}
156
for family in families
157
]
158
}
159
)
160
161
162
def registry_schema_hash(feature_sets: Sequence[FeatureSetSpecV1]) -> str:
163
"""Hash registered feature-set recipes into one registry schema hash."""
164
165
return stable_sha256(
166
{
167
"feature_sets": [
168
{
169
"feature_set_id": feature_set.feature_set_id,
170
"feature_set_version": feature_set.feature_set_version,
171
"lane": feature_set.lane.value,
172
"family_ids": list(feature_set.family_ids),
173
}
174
for feature_set in feature_sets
175
]
176
}
177
)