Spaces:
Sleeping
Sleeping
| """Canonical, optional runtime transparency metadata for model packages. | |
| The legacy training tuple and required model-manifest fields intentionally stay | |
| unchanged. New producers place this validated sidecar under the optional | |
| ``execution_metadata`` manifest key; consumers must continue to accept packages | |
| where that key is absent. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import os | |
| import re | |
| import tempfile | |
| import zipfile | |
| from datetime import datetime, timezone | |
| from pathlib import Path, PurePosixPath | |
| from typing import Any, Mapping, Sequence | |
| from ui_contracts import ( | |
| DecisionSource, | |
| DecisionTrace, | |
| ExperimentReflection, | |
| FeatureExtractionReport, | |
| ResearchPlan, | |
| ) | |
| EXECUTION_METADATA_KEY = "execution_metadata" | |
| EXECUTION_METADATA_SCHEMA_VERSION = 1 | |
| APPROVED_RESEARCH_KEY = "approved_research" | |
| TRUSTED_EXTRACTOR_KEY = "trusted_extractor" | |
| EXPERIMENT_REFLECTION_KEY = "experiment_reflection" | |
| _EXTRACTOR_ID_PATTERN = re.compile(r"^[a-z][a-z0-9_]{2,63}$") | |
| _EXTRACTOR_VERSION_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._+-]{0,63}$") | |
| _SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$") | |
| _TRUSTED_EXTRACTOR_FIELDS = { | |
| "schema_version", | |
| "extractor_id", | |
| "version", | |
| "configuration", | |
| "bundle_hash", | |
| "required_assets", | |
| } | |
| _TRUSTED_ASSET_FIELDS = {"asset_id", "sha256", "size_bytes"} | |
| _APPROVED_RESEARCH_FIELDS = { | |
| "schema_version", | |
| "research_plan", | |
| "approved_candidate_id", | |
| "pipeline_config", | |
| "evidence_ids", | |
| "approval_trace", | |
| } | |
| _PIPELINE_CONFIG_FIELDS = { | |
| "schema_version", | |
| "candidate_id", | |
| "task_type", | |
| "data_modality", | |
| "feature_selection_tool", | |
| "transform_tool", | |
| "ml_tool", | |
| "image_backend", | |
| "image_backbone", | |
| "extractor_configuration", | |
| } | |
| def _mapping(value: Any, name: str) -> Mapping[str, Any]: | |
| if not isinstance(value, Mapping): | |
| raise ValueError(f"{name} must be a mapping") | |
| return value | |
| def make_feature_extraction_report( | |
| *, | |
| report_id: str, | |
| modality: str, | |
| method: str, | |
| input_shape: Sequence[int] = (), | |
| output_shape: Sequence[int] = (), | |
| features_per_sample: int | None = None, | |
| input_feature_count: int | None = None, | |
| output_feature_count: int | None = None, | |
| selected_features: Sequence[str] = (), | |
| generated_features: Sequence[str] = (), | |
| dropped_features: Sequence[str] = (), | |
| parameters: Mapping[str, Any] | None = None, | |
| warnings: Sequence[str] = (), | |
| provenance: Mapping[str, Any] | None = None, | |
| extensions: Mapping[str, Any] | None = None, | |
| status: str = "completed", | |
| ) -> FeatureExtractionReport: | |
| """Build one schema-validated report from actual producer values.""" | |
| return FeatureExtractionReport.from_mapping( | |
| { | |
| "schema_version": 1, | |
| "report_id": report_id, | |
| "status": status, | |
| "modality": modality, | |
| "method": method, | |
| "input_shape": list(input_shape), | |
| "output_shape": list(output_shape), | |
| "features_per_sample": features_per_sample, | |
| "input_feature_count": input_feature_count, | |
| "output_feature_count": output_feature_count, | |
| "selected_features": list(selected_features), | |
| "generated_features": list(generated_features), | |
| "dropped_features": list(dropped_features), | |
| "parameters": dict(parameters or {}), | |
| "warnings": list(warnings), | |
| "provenance": {"phase": "execution", **dict(provenance or {})}, | |
| "extensions": dict(extensions or {}), | |
| } | |
| ) | |
| def make_decision_trace( | |
| *, | |
| trace_id: str, | |
| decision_type: str, | |
| stage: str, | |
| source: str, | |
| selected: Any, | |
| options: Sequence[Any] = (), | |
| rationale: str, | |
| parameters: Mapping[str, Any] | None = None, | |
| actor: str = "runtime", | |
| context_summary: str = "", | |
| evidence_ids: Sequence[str] = (), | |
| warnings: Sequence[str] = (), | |
| provenance: Mapping[str, Any] | None = None, | |
| extensions: Mapping[str, Any] | None = None, | |
| timestamp_utc: str | None = None, | |
| ) -> DecisionTrace: | |
| """Build a reviewable trace while rejecting hidden internal reasoning.""" | |
| return DecisionTrace.from_mapping( | |
| { | |
| "schema_version": 1, | |
| "trace_id": trace_id, | |
| "decision_type": decision_type, | |
| "stage": stage, | |
| "decision_source": source, | |
| "actor": actor, | |
| "context_summary": context_summary, | |
| "options": list(options), | |
| "selected": selected, | |
| "rationale": rationale, | |
| "parameters": dict(parameters or {}), | |
| "evidence_ids": list(evidence_ids), | |
| "warnings": list(warnings), | |
| "provenance": {"phase": "execution", **dict(provenance or {})}, | |
| "timestamp_utc": timestamp_utc | |
| or datetime.now(timezone.utc).isoformat(), | |
| "extensions": dict(extensions or {}), | |
| } | |
| ) | |
| def make_execution_metadata( | |
| feature_report: FeatureExtractionReport | Mapping[str, Any] | None = None, | |
| decision_traces: Sequence[DecisionTrace | Mapping[str, Any]] = (), | |
| *, | |
| approved_research: Mapping[str, Any] | None = None, | |
| trusted_extractor: Mapping[str, Any] | None = None, | |
| experiment_reflection: ExperimentReflection | Mapping[str, Any] | None = None, | |
| ) -> dict[str, Any]: | |
| """Return the canonical JSON-safe execution metadata container.""" | |
| report = ( | |
| None | |
| if feature_report is None | |
| else feature_report | |
| if isinstance(feature_report, FeatureExtractionReport) | |
| else FeatureExtractionReport.from_mapping( | |
| _mapping(feature_report, "feature_extraction_report") | |
| ) | |
| ) | |
| traces = [ | |
| trace | |
| if isinstance(trace, DecisionTrace) | |
| else DecisionTrace.from_mapping(_mapping(trace, "decision trace")) | |
| for trace in decision_traces | |
| ] | |
| approved = ( | |
| normalize_approved_research(approved_research) | |
| if approved_research | |
| else {} | |
| ) | |
| extractor_binding = ( | |
| normalize_trusted_extractor_binding(trusted_extractor) | |
| if trusted_extractor | |
| else {} | |
| ) | |
| reflection = ( | |
| None | |
| if experiment_reflection is None | |
| else experiment_reflection | |
| if isinstance(experiment_reflection, ExperimentReflection) | |
| else ExperimentReflection.from_mapping( | |
| _mapping(experiment_reflection, EXPERIMENT_REFLECTION_KEY) | |
| ) | |
| ) | |
| if approved: | |
| approval_trace = DecisionTrace.from_mapping(approved["approval_trace"]) | |
| if all(trace.trace_id != approval_trace.trace_id for trace in traces): | |
| traces.insert(0, approval_trace) | |
| if reflection is not None and reflection.decision_trace is not None: | |
| if all( | |
| trace.trace_id != reflection.decision_trace.trace_id for trace in traces | |
| ): | |
| traces.append(reflection.decision_trace) | |
| if ( | |
| report is None | |
| and not traces | |
| and not approved | |
| and not extractor_binding | |
| and reflection is None | |
| ): | |
| return {} | |
| result = { | |
| "schema_version": EXECUTION_METADATA_SCHEMA_VERSION, | |
| "feature_extraction_report": report.to_dict() if report else None, | |
| "decision_traces": [trace.to_dict() for trace in traces], | |
| } | |
| if approved: | |
| result[APPROVED_RESEARCH_KEY] = approved | |
| if extractor_binding: | |
| result[TRUSTED_EXTRACTOR_KEY] = extractor_binding | |
| if reflection is not None: | |
| result[EXPERIMENT_REFLECTION_KEY] = reflection.to_dict() | |
| return result | |
| def normalize_trusted_extractor_binding(value: Mapping[str, Any]) -> dict[str, Any]: | |
| """Validate a package-safe binding without treating it as executable authority.""" | |
| data = _mapping(value, TRUSTED_EXTRACTOR_KEY) | |
| if set(data) != _TRUSTED_EXTRACTOR_FIELDS or data.get("schema_version") != 1: | |
| raise ValueError("trusted_extractor has an unsupported shape") | |
| extractor_id = str(data.get("extractor_id") or "").strip().lower() | |
| version = str(data.get("version") or "").strip() | |
| bundle_hash = str(data.get("bundle_hash") or "").strip().lower() | |
| if not _EXTRACTOR_ID_PATTERN.fullmatch(extractor_id): | |
| raise ValueError("trusted_extractor.extractor_id is invalid") | |
| if not _EXTRACTOR_VERSION_PATTERN.fullmatch(version): | |
| raise ValueError("trusted_extractor.version is invalid") | |
| if not _SHA256_PATTERN.fullmatch(bundle_hash): | |
| raise ValueError("trusted_extractor.bundle_hash is invalid") | |
| configuration = data.get("configuration") | |
| if not isinstance(configuration, Mapping) or not all( | |
| isinstance(key, str) for key in configuration | |
| ): | |
| raise ValueError("trusted_extractor.configuration must be a mapping") | |
| try: | |
| configuration_copy = json.loads( | |
| json.dumps(dict(configuration), allow_nan=False) | |
| ) | |
| except (TypeError, ValueError, OverflowError, RecursionError) as exc: | |
| raise ValueError("trusted_extractor.configuration must be JSON-safe") from exc | |
| if len(json.dumps(configuration_copy).encode("utf-8")) > 32_768: | |
| raise ValueError("trusted_extractor.configuration is too large") | |
| raw_assets = data.get("required_assets") | |
| if not isinstance(raw_assets, (list, tuple)) or len(raw_assets) > 64: | |
| raise ValueError("trusted_extractor.required_assets must be a bounded array") | |
| assets: list[dict[str, Any]] = [] | |
| asset_ids: set[str] = set() | |
| for raw in raw_assets: | |
| asset = _mapping(raw, "trusted extractor asset") | |
| if set(asset) != _TRUSTED_ASSET_FIELDS: | |
| raise ValueError("trusted extractor asset has an unsupported shape") | |
| asset_id = str(asset.get("asset_id") or "").strip() | |
| digest = str(asset.get("sha256") or "").strip().lower() | |
| size = asset.get("size_bytes") | |
| if ( | |
| not asset_id | |
| or len(asset_id) > 128 | |
| or asset_id in {".", ".."} | |
| or "/" in asset_id | |
| or "\\" in asset_id | |
| or any(ord(character) < 32 for character in asset_id) | |
| or asset_id in asset_ids | |
| ): | |
| raise ValueError("trusted extractor asset_id is invalid or duplicated") | |
| if not _SHA256_PATTERN.fullmatch(digest): | |
| raise ValueError("trusted extractor asset sha256 is invalid") | |
| if isinstance(size, bool) or not isinstance(size, int) or size <= 0: | |
| raise ValueError("trusted extractor asset size_bytes must be positive") | |
| asset_ids.add(asset_id) | |
| assets.append({"asset_id": asset_id, "sha256": digest, "size_bytes": size}) | |
| return { | |
| "schema_version": 1, | |
| "extractor_id": extractor_id, | |
| "version": version, | |
| "configuration": configuration_copy, | |
| "bundle_hash": bundle_hash, | |
| "required_assets": assets, | |
| } | |
| def normalize_approved_research(value: Mapping[str, Any]) -> dict[str, Any]: | |
| """Validate the optional approved plan carried into trusted execution.""" | |
| data = _mapping(value, APPROVED_RESEARCH_KEY) | |
| if set(data) - _APPROVED_RESEARCH_FIELDS: | |
| raise ValueError("approved_research contains unsupported fields") | |
| if data.get("schema_version", 1) != 1: | |
| raise ValueError("unsupported approved_research schema_version") | |
| plan = ResearchPlan.from_mapping( | |
| _mapping(data.get("research_plan"), "approved_research.research_plan") | |
| ) | |
| if not 1 <= len(plan.candidates) <= 3: | |
| raise ValueError("approved research plan must contain one to three candidates") | |
| approved_id = str(data.get("approved_candidate_id") or "").strip() | |
| candidate = next( | |
| (item for item in plan.candidates if item.candidate_id == approved_id), | |
| None, | |
| ) | |
| if candidate is None: | |
| raise ValueError("approved candidate is not present in the research plan") | |
| config = _mapping( | |
| data.get("pipeline_config"), "approved_research.pipeline_config" | |
| ) | |
| if not all(isinstance(key, str) for key in config): | |
| raise ValueError("approved pipeline config keys must be strings") | |
| if set(config) - _PIPELINE_CONFIG_FIELDS: | |
| raise ValueError("approved pipeline config contains unsupported fields") | |
| if config.get("schema_version", 1) != 1: | |
| raise ValueError("unsupported approved pipeline config schema_version") | |
| if config.get("candidate_id") not in (None, approved_id): | |
| raise ValueError("approved pipeline config references another candidate") | |
| try: | |
| config_copy = json.loads(json.dumps(dict(config), allow_nan=False)) | |
| except (TypeError, ValueError, RecursionError) as exc: | |
| raise ValueError("approved pipeline config must be JSON-safe") from exc | |
| evidence_ids = data.get("evidence_ids") or () | |
| if not isinstance(evidence_ids, (list, tuple)) or not all( | |
| isinstance(item, str) and 0 < len(item) <= 200 for item in evidence_ids | |
| ): | |
| raise ValueError("approved research evidence_ids must be an array of strings") | |
| if len(evidence_ids) != len(set(evidence_ids)): | |
| raise ValueError("approved research evidence_ids must be unique") | |
| known_evidence = { | |
| str(item.get("evidence_id")) | |
| for item in plan.evidence | |
| if isinstance(item, Mapping) and item.get("evidence_id") | |
| } | |
| if not set(evidence_ids) <= known_evidence or not set(evidence_ids) <= set( | |
| candidate.evidence_ids | |
| ): | |
| raise ValueError("approved research references unknown candidate evidence") | |
| trace = DecisionTrace.from_mapping( | |
| _mapping(data.get("approval_trace"), "approved_research.approval_trace") | |
| ) | |
| if trace.decision_source is not DecisionSource.RESEARCH_AGENT: | |
| raise ValueError("approval trace must come from the research agent") | |
| if trace.selected != approved_id: | |
| raise ValueError("approval trace does not match the approved candidate") | |
| return { | |
| "schema_version": 1, | |
| "research_plan": plan.to_dict(), | |
| "approved_candidate_id": approved_id, | |
| "pipeline_config": config_copy, | |
| "evidence_ids": list(evidence_ids), | |
| "approval_trace": trace.to_dict(), | |
| } | |
| def normalize_execution_metadata(value: Mapping[str, Any]) -> dict[str, Any]: | |
| """Validate and canonicalize metadata received from a producer/package.""" | |
| data = _mapping(value, EXECUTION_METADATA_KEY) | |
| if not data: | |
| return {} | |
| version = data.get("schema_version", 1) | |
| if version != EXECUTION_METADATA_SCHEMA_VERSION: | |
| raise ValueError(f"unsupported execution metadata schema_version {version!r}") | |
| traces = data.get("decision_traces", ()) | |
| if not isinstance(traces, (list, tuple)): | |
| raise ValueError("execution_metadata.decision_traces must be an array") | |
| return make_execution_metadata( | |
| data.get("feature_extraction_report"), | |
| traces, | |
| approved_research=data.get(APPROVED_RESEARCH_KEY), | |
| trusted_extractor=data.get(TRUSTED_EXTRACTOR_KEY), | |
| experiment_reflection=data.get(EXPERIMENT_REFLECTION_KEY), | |
| ) | |
| def merge_execution_metadata( | |
| *values: Mapping[str, Any], | |
| feature_report: FeatureExtractionReport | Mapping[str, Any] | None = None, | |
| decision_traces: Sequence[DecisionTrace | Mapping[str, Any]] = (), | |
| approved_research: Mapping[str, Any] | None = None, | |
| trusted_extractor: Mapping[str, Any] | None = None, | |
| experiment_reflection: ExperimentReflection | Mapping[str, Any] | None = None, | |
| ) -> dict[str, Any]: | |
| """Merge producer metadata without duplicating stable trace identifiers.""" | |
| report = feature_report | |
| traces: list[DecisionTrace | Mapping[str, Any]] = [] | |
| approved = approved_research | |
| reflection = experiment_reflection | |
| extractor_binding: Mapping[str, Any] | None = ( | |
| normalize_trusted_extractor_binding(trusted_extractor) | |
| if trusted_extractor | |
| else None | |
| ) | |
| for value in values: | |
| canonical = normalize_execution_metadata(value) if value else {} | |
| if canonical.get("feature_extraction_report") is not None: | |
| report = canonical["feature_extraction_report"] | |
| traces.extend(canonical.get("decision_traces", ())) | |
| if canonical.get(APPROVED_RESEARCH_KEY): | |
| approved = canonical[APPROVED_RESEARCH_KEY] | |
| if canonical.get(TRUSTED_EXTRACTOR_KEY): | |
| incoming_binding = canonical[TRUSTED_EXTRACTOR_KEY] | |
| if extractor_binding is not None and dict(extractor_binding) != dict(incoming_binding): | |
| raise ValueError("cannot merge conflicting trusted extractor bindings") | |
| extractor_binding = incoming_binding | |
| if canonical.get(EXPERIMENT_REFLECTION_KEY): | |
| reflection = canonical[EXPERIMENT_REFLECTION_KEY] | |
| traces.extend(decision_traces) | |
| deduplicated: dict[str, DecisionTrace] = {} | |
| anonymous: list[DecisionTrace] = [] | |
| for raw in traces: | |
| trace = ( | |
| raw | |
| if isinstance(raw, DecisionTrace) | |
| else DecisionTrace.from_mapping(_mapping(raw, "decision trace")) | |
| ) | |
| if trace.trace_id: | |
| deduplicated[trace.trace_id] = trace | |
| else: | |
| anonymous.append(trace) | |
| return make_execution_metadata( | |
| report, | |
| [*deduplicated.values(), *anonymous], | |
| approved_research=approved, | |
| trusted_extractor=extractor_binding, | |
| experiment_reflection=reflection, | |
| ) | |
| def merge_approved_research_into_package( | |
| package_path: str | os.PathLike[str], | |
| approved_research: Mapping[str, Any], | |
| ) -> dict[str, Any]: | |
| """Atomically add approved research to one locally produced ZIP manifest.""" | |
| approved = normalize_approved_research(approved_research) | |
| return _merge_optional_metadata_into_package( | |
| package_path, approved_research=approved | |
| ) | |
| def merge_experiment_reflection_into_package( | |
| package_path: str | os.PathLike[str], | |
| experiment_reflection: ExperimentReflection | Mapping[str, Any], | |
| ) -> dict[str, Any]: | |
| """Atomically persist one validated, optional reflection in a package.""" | |
| reflection = ( | |
| experiment_reflection | |
| if isinstance(experiment_reflection, ExperimentReflection) | |
| else ExperimentReflection.from_mapping( | |
| _mapping(experiment_reflection, EXPERIMENT_REFLECTION_KEY) | |
| ) | |
| ) | |
| return _merge_optional_metadata_into_package( | |
| package_path, experiment_reflection=reflection | |
| ) | |
| def _merge_optional_metadata_into_package( | |
| package_path: str | os.PathLike[str], | |
| *, | |
| approved_research: Mapping[str, Any] | None = None, | |
| experiment_reflection: ExperimentReflection | Mapping[str, Any] | None = None, | |
| ) -> dict[str, Any]: | |
| """Rewrite exactly one trusted manifest while retaining all ZIP members.""" | |
| source = Path(package_path) | |
| if not source.is_file(): | |
| raise OSError("model package is unavailable") | |
| descriptor, temp_name = tempfile.mkstemp( | |
| prefix=f".{source.name}.", suffix=".tmp", dir=str(source.parent) | |
| ) | |
| os.close(descriptor) | |
| temp_path = Path(temp_name) | |
| manifest_count = 0 | |
| merged: dict[str, Any] = {} | |
| try: | |
| with zipfile.ZipFile(source, "r") as incoming, zipfile.ZipFile( | |
| temp_path, "w" | |
| ) as outgoing: | |
| for info in incoming.infolist(): | |
| payload = incoming.read(info) | |
| # The trusted package contract designates exactly one | |
| # automl_model_manifest.json as the model manifest. Image | |
| # artifacts may also contain a dataset-level manifest.json; | |
| # that file is not the package manifest and must remain | |
| # untouched. | |
| if ( | |
| PurePosixPath(info.filename).name | |
| == "automl_model_manifest.json" | |
| ): | |
| manifest_count += 1 | |
| if manifest_count > 1: | |
| raise ValueError("model package contains multiple manifests") | |
| try: | |
| manifest = json.loads(payload.decode("utf-8")) | |
| except (UnicodeDecodeError, json.JSONDecodeError) as exc: | |
| raise ValueError("model package manifest is invalid") from exc | |
| if not isinstance(manifest, Mapping): | |
| raise ValueError("model package manifest must be a mapping") | |
| merged = merge_execution_metadata( | |
| execution_metadata_from_manifest(manifest), | |
| approved_research=approved_research, | |
| experiment_reflection=experiment_reflection, | |
| ) | |
| payload = json.dumps( | |
| manifest_with_execution_metadata(manifest, merged), | |
| indent=2, | |
| sort_keys=True, | |
| ).encode("utf-8") | |
| outgoing.writestr(info, payload) | |
| if manifest_count != 1: | |
| raise ValueError("model package manifest is missing") | |
| os.replace(temp_path, source) | |
| except Exception: | |
| try: | |
| temp_path.unlink() | |
| except OSError: | |
| pass | |
| raise | |
| return merged | |
| def execution_metadata_from_manifest(manifest: Mapping[str, Any]) -> dict[str, Any]: | |
| """Read optional metadata; absence is the legacy-compatible empty value.""" | |
| value = _mapping(manifest, "manifest").get(EXECUTION_METADATA_KEY) | |
| if value is None: | |
| return {} | |
| return normalize_execution_metadata(_mapping(value, EXECUTION_METADATA_KEY)) | |
| def trusted_extractor_from_manifest(manifest: Mapping[str, Any]) -> dict[str, Any]: | |
| """Read the optional package binding; legacy packages return an empty mapping.""" | |
| metadata = execution_metadata_from_manifest(manifest) | |
| return dict(metadata.get(TRUSTED_EXTRACTOR_KEY) or {}) | |
| def manifest_with_execution_metadata( | |
| manifest: Mapping[str, Any], metadata: Mapping[str, Any] | |
| ) -> dict[str, Any]: | |
| """Return a manifest copy with validated metadata only when it is present.""" | |
| result = dict(_mapping(manifest, "manifest")) | |
| canonical = normalize_execution_metadata(metadata) if metadata else {} | |
| if canonical: | |
| result[EXECUTION_METADATA_KEY] = canonical | |
| return result | |
| __all__ = [ | |
| "EXECUTION_METADATA_KEY", | |
| "EXECUTION_METADATA_SCHEMA_VERSION", | |
| "APPROVED_RESEARCH_KEY", | |
| "EXPERIMENT_REFLECTION_KEY", | |
| "TRUSTED_EXTRACTOR_KEY", | |
| "execution_metadata_from_manifest", | |
| "make_decision_trace", | |
| "make_execution_metadata", | |
| "make_feature_extraction_report", | |
| "manifest_with_execution_metadata", | |
| "merge_approved_research_into_package", | |
| "merge_experiment_reflection_into_package", | |
| "merge_execution_metadata", | |
| "normalize_execution_metadata", | |
| "normalize_approved_research", | |
| "normalize_trusted_extractor_binding", | |
| "trusted_extractor_from_manifest", | |
| ] | |