Spaces:
Sleeping
Sleeping
| """Versioned, immutable UI-facing contracts for explainable AutoML planning. | |
| This module deliberately has no dependency on the UI, feature extractors, or a | |
| research-agent implementation. It accepts the small set of legacy shapes used | |
| by the project while always serializing one canonical version-1 shape. | |
| """ | |
| from __future__ import annotations | |
| import math | |
| from dataclasses import dataclass, field | |
| from enum import Enum | |
| from types import MappingProxyType | |
| from typing import Any, Mapping, Protocol, Sequence | |
| SCHEMA_VERSION = 1 | |
| _MISSING = object() | |
| _INTERNAL_REASONING_KEYS = { | |
| "analysis", | |
| "chain_of_thought", | |
| "hidden_reasoning", | |
| "internal_reasoning", | |
| } | |
| class DecisionSource(str, Enum): | |
| """Auditable origin of a decision, without hidden reasoning content.""" | |
| USER = "user" | |
| PRESET = "preset" | |
| DETERMINISTIC_RULE = "deterministic_rule" | |
| BENCHMARK = "benchmark" | |
| LLM = "llm" | |
| RESEARCH_AGENT = "research_agent" | |
| FALLBACK = "fallback" | |
| def _mapping(value: Any, field_name: str = "payload") -> Mapping[str, Any]: | |
| if not isinstance(value, Mapping): | |
| raise ValueError(f"{field_name} must be a mapping") | |
| if not all(isinstance(key, str) for key in value): | |
| raise ValueError(f"{field_name} keys must be strings") | |
| return value | |
| def _plain(value: Any) -> Any: | |
| if isinstance(value, Enum): | |
| return value.value | |
| if isinstance(value, Mapping): | |
| return {key: _plain(item) for key, item in value.items()} | |
| if isinstance(value, (list, tuple)): | |
| return [_plain(item) for item in value] | |
| return value | |
| def _freeze_json(value: Any, field_name: str = "value") -> Any: | |
| """Deep-freeze a JSON-compatible value so frozen contracts stay immutable.""" | |
| if isinstance(value, Enum): | |
| value = value.value | |
| if value is None or isinstance(value, (str, bool, int)): | |
| return value | |
| if isinstance(value, float): | |
| if not math.isfinite(value): | |
| raise ValueError(f"{field_name} must not contain non-finite numbers") | |
| return value | |
| if isinstance(value, Mapping): | |
| if not all(isinstance(key, str) for key in value): | |
| raise ValueError(f"{field_name} keys must be strings") | |
| return MappingProxyType( | |
| { | |
| key: _freeze_json(item, f"{field_name}.{key}") | |
| for key, item in value.items() | |
| } | |
| ) | |
| if isinstance(value, (list, tuple)): | |
| return tuple( | |
| _freeze_json(item, f"{field_name}[{index}]") | |
| for index, item in enumerate(value) | |
| ) | |
| raise ValueError(f"{field_name} must contain only JSON-compatible values") | |
| def _thaw_json(value: Any) -> Any: | |
| if isinstance(value, Enum): | |
| return value.value | |
| if isinstance(value, Mapping): | |
| return {key: _thaw_json(item) for key, item in value.items()} | |
| if isinstance(value, tuple): | |
| return [_thaw_json(item) for item in value] | |
| return value | |
| def _version(data: Mapping[str, Any]) -> int: | |
| raw = data.get("schema_version", 0) | |
| if isinstance(raw, bool) or not isinstance(raw, int) or raw not in (0, 1): | |
| raise ValueError( | |
| f"unsupported schema_version {raw!r}; supported versions are 0 and 1" | |
| ) | |
| return raw | |
| def _assert_no_internal_reasoning(data: Mapping[str, Any]) -> None: | |
| def walk(value: Any) -> bool: | |
| if isinstance(value, Mapping): | |
| return any( | |
| key in _INTERNAL_REASONING_KEYS or walk(item) | |
| for key, item in value.items() | |
| ) | |
| if isinstance(value, (list, tuple)): | |
| return any(walk(item) for item in value) | |
| return False | |
| if walk(data): | |
| raise ValueError( | |
| "chain-of-thought or other hidden internal reasoning is not part of " | |
| "the UI contract; provide a concise rationale instead" | |
| ) | |
| def _path_value(data: Mapping[str, Any], path: tuple[str, ...]) -> Any: | |
| current: Any = data | |
| for part in path: | |
| if not isinstance(current, Mapping) or part not in current: | |
| return _MISSING | |
| current = current[part] | |
| return current | |
| def _resolve( | |
| data: Mapping[str, Any], | |
| canonical: str, | |
| aliases: Sequence[str | tuple[str, ...]] = (), | |
| *, | |
| default: Any = _MISSING, | |
| required: bool = False, | |
| ) -> tuple[Any, set[str]]: | |
| """Resolve canonical/legacy names and reject silently lossy conflicts.""" | |
| paths = [(canonical,)] + [ | |
| (alias,) if isinstance(alias, str) else alias for alias in aliases | |
| ] | |
| found: list[tuple[tuple[str, ...], Any]] = [] | |
| consumed: set[str] = set() | |
| for path in paths: | |
| value = _path_value(data, path) | |
| if value is not _MISSING: | |
| found.append((path, value)) | |
| consumed.add(path[0]) | |
| if found: | |
| first_path, first_value = found[0] | |
| for other_path, other_value in found[1:]: | |
| if _plain(first_value) != _plain(other_value): | |
| raise ValueError( | |
| "conflicting values for " | |
| f"{canonical!r}: {'.'.join(first_path)!r} and " | |
| f"{'.'.join(other_path)!r}" | |
| ) | |
| return first_value, consumed | |
| if required: | |
| raise ValueError(f"{canonical} is required") | |
| return default, consumed | |
| def _text(value: Any, field_name: str, *, required: bool = False) -> str: | |
| if value is None and not required: | |
| return "" | |
| if not isinstance(value, str): | |
| raise ValueError(f"{field_name} must be a string") | |
| result = " ".join(value.split()) | |
| if required and not result: | |
| raise ValueError(f"{field_name} is required") | |
| return result | |
| def _rationale(value: Any) -> str: | |
| result = _text(value, "rationale") | |
| if len(result) > 2_000: | |
| raise ValueError("rationale must be a concise summary of at most 2000 characters") | |
| return result | |
| def _optional_count(value: Any, field_name: str) -> int | None: | |
| if value is None: | |
| return None | |
| if isinstance(value, bool) or not isinstance(value, int) or value < 0: | |
| raise ValueError(f"{field_name} must be a non-negative integer or null") | |
| return value | |
| def _shape(value: Any, field_name: str) -> tuple[int, ...]: | |
| if value is None: | |
| return () | |
| if not isinstance(value, (list, tuple)): | |
| raise ValueError(f"{field_name} must be an array of non-negative integers") | |
| result: list[int] = [] | |
| for dimension in value: | |
| if isinstance(dimension, bool) or not isinstance(dimension, int) or dimension < 0: | |
| raise ValueError( | |
| f"{field_name} must be an array of non-negative integers" | |
| ) | |
| result.append(dimension) | |
| return tuple(result) | |
| def _strings(value: Any, field_name: str) -> tuple[str, ...]: | |
| if value is None: | |
| return () | |
| if isinstance(value, str): | |
| value = [value] | |
| if not isinstance(value, (list, tuple)): | |
| raise ValueError(f"{field_name} must be an array of strings") | |
| return tuple(_text(item, field_name, required=True) for item in value) | |
| def _frozen_mapping(value: Any, field_name: str) -> Mapping[str, Any]: | |
| if value is None: | |
| value = {} | |
| return _freeze_json(_mapping(value, field_name), field_name) | |
| def _extensions( | |
| data: Mapping[str, Any], consumed: set[str], extra: Mapping[str, Any] | None = None | |
| ) -> Mapping[str, Any]: | |
| raw = data.get("extensions", {}) | |
| result = dict(_mapping(raw, "extensions")) | |
| for key, value in data.items(): | |
| if key not in consumed and key not in {"schema_version", "extensions"}: | |
| if key in result and _plain(result[key]) != _plain(value): | |
| raise ValueError(f"conflicting values for extension {key!r}") | |
| result[key] = value | |
| for key, value in (extra or {}).items(): | |
| if key in result and _plain(result[key]) != _plain(value): | |
| raise ValueError(f"conflicting values for extension {key!r}") | |
| result[key] = value | |
| return _frozen_mapping(result, "extensions") | |
| class FeatureExtractionReport: | |
| """Stable summary of feature extraction outputs and shapes.""" | |
| schema_version: int = SCHEMA_VERSION | |
| report_id: str = "" | |
| status: str = "unknown" | |
| modality: str = "" | |
| method: str = "" | |
| input_shape: tuple[int, ...] = () | |
| output_shape: tuple[int, ...] = () | |
| features_per_sample: int | None = None | |
| input_feature_count: int | None = None | |
| output_feature_count: int | None = None | |
| selected_features: tuple[str, ...] = () | |
| generated_features: tuple[str, ...] = () | |
| dropped_features: tuple[str, ...] = () | |
| parameters: Mapping[str, Any] = field(default_factory=dict) | |
| warnings: tuple[str, ...] = () | |
| provenance: Mapping[str, Any] = field(default_factory=dict) | |
| extensions: Mapping[str, Any] = field(default_factory=dict) | |
| def __post_init__(self) -> None: | |
| if self.schema_version != SCHEMA_VERSION: | |
| raise ValueError(f"unsupported schema_version {self.schema_version!r}") | |
| object.__setattr__(self, "input_shape", _shape(self.input_shape, "input_shape")) | |
| object.__setattr__(self, "output_shape", _shape(self.output_shape, "output_shape")) | |
| object.__setattr__( | |
| self, | |
| "features_per_sample", | |
| _optional_count(self.features_per_sample, "features_per_sample"), | |
| ) | |
| object.__setattr__( | |
| self, | |
| "input_feature_count", | |
| _optional_count(self.input_feature_count, "input_feature_count"), | |
| ) | |
| object.__setattr__( | |
| self, | |
| "output_feature_count", | |
| _optional_count(self.output_feature_count, "output_feature_count"), | |
| ) | |
| for name in ("selected_features", "generated_features", "dropped_features"): | |
| object.__setattr__(self, name, _strings(getattr(self, name), name)) | |
| for name in ("parameters", "provenance", "extensions"): | |
| object.__setattr__(self, name, _frozen_mapping(getattr(self, name), name)) | |
| object.__setattr__(self, "warnings", _strings(self.warnings, "warnings")) | |
| def from_mapping(cls, payload: Mapping[str, Any]) -> "FeatureExtractionReport": | |
| data = _mapping(payload) | |
| _version(data) | |
| consumed: set[str] = set() | |
| def get(name: str, aliases: Sequence[str] = (), default: Any = None) -> Any: | |
| value, used = _resolve(data, name, aliases, default=default) | |
| consumed.update(used) | |
| return value | |
| return cls( | |
| report_id=_text(get("report_id", ("id",), ""), "report_id"), | |
| status=_text(get("status", (), "unknown"), "status") or "unknown", | |
| modality=_text(get("modality", ("data_type",), ""), "modality"), | |
| method=_text(get("method", ("tool_name", "extractor"), ""), "method"), | |
| input_shape=_shape(get("input_shape", (), ()), "input_shape"), | |
| output_shape=_shape(get("output_shape", (), ()), "output_shape"), | |
| features_per_sample=_optional_count( | |
| get("features_per_sample", ("feature_dim",), None), | |
| "features_per_sample", | |
| ), | |
| input_feature_count=_optional_count( | |
| get("input_feature_count", ("n_original",), None), | |
| "input_feature_count", | |
| ), | |
| output_feature_count=_optional_count( | |
| get("output_feature_count", ("n_selected",), None), | |
| "output_feature_count", | |
| ), | |
| selected_features=_strings( | |
| get("selected_features", ("selected_feature_names",), ()), | |
| "selected_features", | |
| ), | |
| generated_features=_strings( | |
| get("generated_features", ("generated_feature_names",), ()), | |
| "generated_features", | |
| ), | |
| dropped_features=_strings( | |
| get("dropped_features", ("dropped_feature_names",), ()), | |
| "dropped_features", | |
| ), | |
| parameters=_frozen_mapping( | |
| get("parameters", ("params",), {}), "parameters" | |
| ), | |
| warnings=_strings(get("warnings", (), ()), "warnings"), | |
| provenance=_frozen_mapping(get("provenance", (), {}), "provenance"), | |
| extensions=_extensions(data, consumed), | |
| ) | |
| def to_dict(self) -> dict[str, Any]: | |
| return { | |
| "schema_version": self.schema_version, | |
| "report_id": self.report_id, | |
| "status": self.status, | |
| "modality": self.modality, | |
| "method": self.method, | |
| "input_shape": list(self.input_shape), | |
| "output_shape": list(self.output_shape), | |
| "features_per_sample": self.features_per_sample, | |
| "input_feature_count": self.input_feature_count, | |
| "output_feature_count": self.output_feature_count, | |
| "selected_features": list(self.selected_features), | |
| "generated_features": list(self.generated_features), | |
| "dropped_features": list(self.dropped_features), | |
| "parameters": _thaw_json(self.parameters), | |
| "warnings": list(self.warnings), | |
| "provenance": _thaw_json(self.provenance), | |
| "extensions": _thaw_json(self.extensions), | |
| } | |
| class DecisionTrace: | |
| """Reviewable decision provenance containing concise rationale only.""" | |
| schema_version: int = SCHEMA_VERSION | |
| trace_id: str = "" | |
| decision_type: str = "" | |
| stage: str = "" | |
| decision_source: DecisionSource = DecisionSource.DETERMINISTIC_RULE | |
| actor: str = "" | |
| context_summary: str = "" | |
| options: tuple[Any, ...] = () | |
| selected: Any = None | |
| rationale: str = "" | |
| parameters: Mapping[str, Any] = field(default_factory=dict) | |
| evidence_ids: tuple[str, ...] = () | |
| warnings: tuple[str, ...] = () | |
| provenance: Mapping[str, Any] = field(default_factory=dict) | |
| timestamp_utc: str = "" | |
| extensions: Mapping[str, Any] = field(default_factory=dict) | |
| def __post_init__(self) -> None: | |
| if self.schema_version != SCHEMA_VERSION: | |
| raise ValueError(f"unsupported schema_version {self.schema_version!r}") | |
| try: | |
| source = DecisionSource(self.decision_source) | |
| except (TypeError, ValueError) as exc: | |
| raise ValueError(f"invalid decision_source {self.decision_source!r}") from exc | |
| object.__setattr__(self, "decision_source", source) | |
| object.__setattr__(self, "options", _freeze_json(self.options, "options")) | |
| object.__setattr__(self, "selected", _freeze_json(self.selected, "selected")) | |
| object.__setattr__(self, "rationale", _rationale(self.rationale)) | |
| for name in ("parameters", "provenance", "extensions"): | |
| object.__setattr__(self, name, _frozen_mapping(getattr(self, name), name)) | |
| object.__setattr__(self, "evidence_ids", _strings(self.evidence_ids, "evidence_ids")) | |
| object.__setattr__(self, "warnings", _strings(self.warnings, "warnings")) | |
| def from_mapping(cls, payload: Mapping[str, Any]) -> "DecisionTrace": | |
| data = _mapping(payload) | |
| incoming_version = _version(data) | |
| _assert_no_internal_reasoning(data) | |
| consumed: set[str] = set() | |
| def get( | |
| name: str, | |
| aliases: Sequence[str | tuple[str, ...]] = (), | |
| default: Any = None, | |
| required: bool = False, | |
| ) -> Any: | |
| value, used = _resolve( | |
| data, name, aliases, default=default, required=required | |
| ) | |
| consumed.update(used) | |
| return value | |
| metadata = data.get("metadata", {}) | |
| if metadata is not None: | |
| metadata = _mapping(metadata, "metadata") | |
| source_raw = get("decision_source", ("source",), _MISSING) | |
| if source_raw is _MISSING: | |
| if incoming_version == SCHEMA_VERSION: | |
| raise ValueError("decision_source is required for schema_version 1") | |
| source_raw = ( | |
| DecisionSource.FALLBACK.value | |
| if metadata.get("fallback") is True | |
| else DecisionSource.DETERMINISTIC_RULE.value | |
| ) | |
| try: | |
| source = DecisionSource(source_raw) | |
| except (TypeError, ValueError) as exc: | |
| raise ValueError(f"invalid decision_source {source_raw!r}") from exc | |
| parameters = get( | |
| "parameters", ("params", ("metadata", "parameters")), {} | |
| ) | |
| legacy_metadata = { | |
| key: value | |
| for key, value in metadata.items() | |
| if key not in {"fallback", "parameters"} | |
| } | |
| if "metadata" in data: | |
| consumed.add("metadata") | |
| extension_extra = ( | |
| {"legacy_metadata": legacy_metadata} if legacy_metadata else None | |
| ) | |
| return cls( | |
| trace_id=_text(get("trace_id", ("id",), ""), "trace_id"), | |
| decision_type=_text( | |
| get("decision_type", ("type",), ""), "decision_type" | |
| ), | |
| stage=_text(get("stage", (), ""), "stage"), | |
| decision_source=source, | |
| actor=_text(get("actor", (), ""), "actor"), | |
| context_summary=_text( | |
| get("context_summary", ("context",), ""), "context_summary" | |
| ), | |
| options=_decision_options(get("options", ("candidates",), ())), | |
| selected=get("selected", ("decision",), None), | |
| rationale=_rationale(get("rationale", ("reasoning",), "")), | |
| parameters=_frozen_mapping(parameters, "parameters"), | |
| evidence_ids=_strings( | |
| get("evidence_ids", ("evidence",), ()), "evidence_ids" | |
| ), | |
| warnings=_strings(get("warnings", (), ()), "warnings"), | |
| provenance=_frozen_mapping(get("provenance", (), {}), "provenance"), | |
| timestamp_utc=_text( | |
| get("timestamp_utc", ("timestamp",), ""), "timestamp_utc" | |
| ), | |
| extensions=_extensions(data, consumed, extension_extra), | |
| ) | |
| def to_dict(self) -> dict[str, Any]: | |
| return { | |
| "schema_version": self.schema_version, | |
| "trace_id": self.trace_id, | |
| "decision_type": self.decision_type, | |
| "stage": self.stage, | |
| "decision_source": self.decision_source.value, | |
| "actor": self.actor, | |
| "context_summary": self.context_summary, | |
| "options": _thaw_json(self.options), | |
| "selected": _thaw_json(self.selected), | |
| "rationale": self.rationale, | |
| "parameters": _thaw_json(self.parameters), | |
| "evidence_ids": list(self.evidence_ids), | |
| "warnings": list(self.warnings), | |
| "provenance": _thaw_json(self.provenance), | |
| "timestamp_utc": self.timestamp_utc, | |
| "extensions": _thaw_json(self.extensions), | |
| } | |
| def _metrics(value: Any, field_name: str, kind: str) -> Mapping[str, Any]: | |
| raw = _mapping(value or {}, field_name) | |
| result: dict[str, Any] = {} | |
| for metric_name, metric_value in raw.items(): | |
| if isinstance(metric_value, Mapping): | |
| entry = dict(metric_value) | |
| existing_kind = entry.get("kind") | |
| if existing_kind is not None and existing_kind != kind: | |
| raise ValueError( | |
| f"{field_name}.{metric_name} has kind {existing_kind!r}; " | |
| f"{field_name} requires kind {kind!r}" | |
| ) | |
| entry["kind"] = kind | |
| else: | |
| entry = { | |
| "kind": kind, | |
| "estimate" if kind == "expected" else "value": metric_value, | |
| } | |
| result[metric_name] = entry | |
| return _frozen_mapping(result, field_name) | |
| def _decision_options(value: Any) -> tuple[Any, ...]: | |
| if value is None: | |
| return () | |
| if not isinstance(value, (list, tuple)): | |
| raise ValueError("options must be an array") | |
| return tuple(value) | |
| class CandidatePipeline: | |
| """One candidate plan with estimates and measurements kept separate.""" | |
| schema_version: int = SCHEMA_VERSION | |
| candidate_id: str = "" | |
| name: str = "" | |
| task_type: str = "" | |
| stages: tuple[Mapping[str, Any], ...] = () | |
| rationale: str = "" | |
| evidence_ids: tuple[str, ...] = () | |
| assumptions: tuple[str, ...] = () | |
| expected_tradeoffs: tuple[str, ...] = () | |
| expected_metrics: Mapping[str, Any] = field(default_factory=dict) | |
| measured_metrics: Mapping[str, Any] = field(default_factory=dict) | |
| warnings: tuple[str, ...] = () | |
| decision_trace_ids: tuple[str, ...] = () | |
| provenance: Mapping[str, Any] = field(default_factory=dict) | |
| extensions: Mapping[str, Any] = field(default_factory=dict) | |
| def __post_init__(self) -> None: | |
| if self.schema_version != SCHEMA_VERSION: | |
| raise ValueError(f"unsupported schema_version {self.schema_version!r}") | |
| object.__setattr__( | |
| self, | |
| "candidate_id", | |
| _text(self.candidate_id, "candidate_id", required=True), | |
| ) | |
| object.__setattr__(self, "stages", _freeze_json(self.stages, "stages")) | |
| if not all(isinstance(stage, Mapping) for stage in self.stages): | |
| raise ValueError("stages must be an array of mappings") | |
| object.__setattr__(self, "rationale", _rationale(self.rationale)) | |
| for name in ( | |
| "evidence_ids", | |
| "assumptions", | |
| "expected_tradeoffs", | |
| "warnings", | |
| "decision_trace_ids", | |
| ): | |
| object.__setattr__(self, name, _strings(getattr(self, name), name)) | |
| object.__setattr__( | |
| self, | |
| "expected_metrics", | |
| _metrics(self.expected_metrics, "expected_metrics", "expected"), | |
| ) | |
| object.__setattr__( | |
| self, | |
| "measured_metrics", | |
| _metrics(self.measured_metrics, "measured_metrics", "measured"), | |
| ) | |
| for name in ("provenance", "extensions"): | |
| object.__setattr__(self, name, _frozen_mapping(getattr(self, name), name)) | |
| def from_mapping(cls, payload: Mapping[str, Any]) -> "CandidatePipeline": | |
| return cls._from_mapping(payload) | |
| def from_legacy_research_agent( | |
| cls, payload: Mapping[str, Any] | |
| ) -> "CandidatePipeline": | |
| return cls._from_mapping( | |
| payload, generic_metrics_kind="expected", adapter_name="research_agent" | |
| ) | |
| def from_legacy_benchmark( | |
| cls, payload: Mapping[str, Any] | |
| ) -> "CandidatePipeline": | |
| return cls._from_mapping( | |
| payload, generic_metrics_kind="measured", adapter_name="benchmark" | |
| ) | |
| def _from_mapping( | |
| cls, | |
| payload: Mapping[str, Any], | |
| *, | |
| generic_metrics_kind: str | None = None, | |
| adapter_name: str | None = None, | |
| ) -> "CandidatePipeline": | |
| data = _mapping(payload) | |
| _version(data) | |
| _assert_no_internal_reasoning(data) | |
| consumed: set[str] = set() | |
| def get( | |
| name: str, | |
| aliases: Sequence[str] = (), | |
| default: Any = None, | |
| required: bool = False, | |
| ) -> Any: | |
| value, used = _resolve( | |
| data, name, aliases, default=default, required=required | |
| ) | |
| consumed.update(used) | |
| return value | |
| expected = get( | |
| "expected_metrics", ("estimated_metrics", "predicted_metrics", "projected_metrics"), {} | |
| ) | |
| measured = get( | |
| "measured_metrics", ("evaluation_metrics", "observed_metrics"), {} | |
| ) | |
| if "metrics" in data: | |
| if generic_metrics_kind is None: | |
| raise ValueError( | |
| "generic metrics are ambiguous; use expected_metrics or " | |
| "measured_metrics, or a source-specific legacy adapter" | |
| ) | |
| generic = data["metrics"] | |
| consumed.add("metrics") | |
| target = expected if generic_metrics_kind == "expected" else measured | |
| if target and _plain(target) != _plain(generic): | |
| field_name = ( | |
| "expected_metrics" | |
| if generic_metrics_kind == "expected" | |
| else "measured_metrics" | |
| ) | |
| raise ValueError( | |
| f"conflicting values for {field_name!r} and legacy 'metrics'" | |
| ) | |
| if generic_metrics_kind == "expected": | |
| expected = generic | |
| else: | |
| measured = generic | |
| provenance = dict(_mapping(get("provenance", (), {}), "provenance")) | |
| if adapter_name: | |
| prior = provenance.get("legacy_adapter") | |
| if prior is not None and prior != adapter_name: | |
| raise ValueError("conflicting values for provenance.legacy_adapter") | |
| provenance["legacy_adapter"] = adapter_name | |
| stages_raw = get("stages", ("pipeline_steps",), ()) | |
| if not isinstance(stages_raw, (list, tuple)): | |
| raise ValueError("stages must be an array of mappings") | |
| if not all(isinstance(stage, Mapping) for stage in stages_raw): | |
| raise ValueError("stages must be an array of mappings") | |
| return cls( | |
| candidate_id=_text( | |
| get("candidate_id", ("id",), required=True), | |
| "candidate_id", | |
| required=True, | |
| ), | |
| name=_text(get("name", ("tool_name",), ""), "name"), | |
| task_type=_text(get("task_type", ("problem_type",), ""), "task_type"), | |
| stages=tuple(stages_raw), | |
| rationale=_rationale(get("rationale", ("reasoning",), "")), | |
| evidence_ids=_strings( | |
| get("evidence_ids", ("evidence",), ()), "evidence_ids" | |
| ), | |
| assumptions=_strings(get("assumptions", (), ()), "assumptions"), | |
| expected_tradeoffs=_strings( | |
| get("expected_tradeoffs", ("tradeoffs",), ()), | |
| "expected_tradeoffs", | |
| ), | |
| expected_metrics=_metrics(expected, "expected_metrics", "expected"), | |
| measured_metrics=_metrics(measured, "measured_metrics", "measured"), | |
| warnings=_strings(get("warnings", (), ()), "warnings"), | |
| decision_trace_ids=_strings( | |
| get("decision_trace_ids", ("decision_ids",), ()), | |
| "decision_trace_ids", | |
| ), | |
| provenance=_frozen_mapping(provenance, "provenance"), | |
| extensions=_extensions(data, consumed), | |
| ) | |
| def to_dict(self) -> dict[str, Any]: | |
| return { | |
| "schema_version": self.schema_version, | |
| "candidate_id": self.candidate_id, | |
| "name": self.name, | |
| "task_type": self.task_type, | |
| "stages": _thaw_json(self.stages), | |
| "rationale": self.rationale, | |
| "evidence_ids": list(self.evidence_ids), | |
| "assumptions": list(self.assumptions), | |
| "expected_tradeoffs": list(self.expected_tradeoffs), | |
| "expected_metrics": _thaw_json(self.expected_metrics), | |
| "measured_metrics": _thaw_json(self.measured_metrics), | |
| "warnings": list(self.warnings), | |
| "decision_trace_ids": list(self.decision_trace_ids), | |
| "provenance": _thaw_json(self.provenance), | |
| "extensions": _thaw_json(self.extensions), | |
| } | |
| _VERIFICATION_STATUSES = {"approved", "revised", "incompatible", "rejected"} | |
| _VERIFICATION_SOURCES = {DecisionSource.LLM, DecisionSource.FALLBACK} | |
| _EXPECTATION_RESULTS = {"met", "partially_met", "not_met", "unknown"} | |
| class CandidateVerification: | |
| """Bounded public critic result for one immutable pipeline candidate.""" | |
| schema_version: int = SCHEMA_VERSION | |
| candidate_id: str = "" | |
| status: str = "approved" | |
| compatible: bool = True | |
| rationale: str = "" | |
| evidence_ids: tuple[str, ...] = () | |
| assumptions: tuple[str, ...] = () | |
| weaknesses: tuple[str, ...] = () | |
| warnings: tuple[str, ...] = () | |
| revised_rationale: str = "" | |
| revised_tradeoffs: tuple[str, ...] = () | |
| extensions: Mapping[str, Any] = field(default_factory=dict) | |
| def __post_init__(self) -> None: | |
| if self.schema_version != SCHEMA_VERSION: | |
| raise ValueError(f"unsupported schema_version {self.schema_version!r}") | |
| object.__setattr__( | |
| self, "candidate_id", _text(self.candidate_id, "candidate_id", required=True) | |
| ) | |
| status = _text(self.status, "status", required=True).lower() | |
| if status not in _VERIFICATION_STATUSES: | |
| raise ValueError(f"invalid candidate verification status {status!r}") | |
| if not isinstance(self.compatible, bool): | |
| raise ValueError("compatible must be boolean") | |
| if status in {"incompatible", "rejected"} and self.compatible: | |
| raise ValueError("rejected or incompatible candidates cannot be compatible") | |
| if status in {"approved", "revised"} and not self.compatible: | |
| raise ValueError("approved or revised candidates must be compatible") | |
| object.__setattr__(self, "status", status) | |
| object.__setattr__(self, "rationale", _rationale(self.rationale)) | |
| object.__setattr__( | |
| self, "revised_rationale", _rationale(self.revised_rationale) | |
| ) | |
| for name in ( | |
| "evidence_ids", | |
| "assumptions", | |
| "weaknesses", | |
| "warnings", | |
| "revised_tradeoffs", | |
| ): | |
| values = _strings(getattr(self, name), name) | |
| if len(values) > 32 or len(values) != len(set(values)): | |
| raise ValueError(f"{name} must be a bounded unique array") | |
| object.__setattr__(self, name, values) | |
| object.__setattr__(self, "extensions", _frozen_mapping(self.extensions, "extensions")) | |
| def from_mapping(cls, payload: Mapping[str, Any]) -> "CandidateVerification": | |
| data = _mapping(payload, "candidate verification") | |
| _version(data) | |
| _assert_no_internal_reasoning(data) | |
| allowed = { | |
| "schema_version", | |
| "candidate_id", | |
| "status", | |
| "compatible", | |
| "rationale", | |
| "evidence_ids", | |
| "assumptions", | |
| "weaknesses", | |
| "warnings", | |
| "revised_rationale", | |
| "revised_tradeoffs", | |
| "extensions", | |
| } | |
| if set(data) - allowed: | |
| raise ValueError("candidate verification contains unsupported fields") | |
| return cls( | |
| candidate_id=data.get("candidate_id", ""), | |
| status=data.get("status", "approved"), | |
| compatible=data.get("compatible", True), | |
| rationale=data.get("rationale", ""), | |
| evidence_ids=tuple(data.get("evidence_ids") or ()), | |
| assumptions=tuple(data.get("assumptions") or ()), | |
| weaknesses=tuple(data.get("weaknesses") or ()), | |
| warnings=tuple(data.get("warnings") or ()), | |
| revised_rationale=data.get("revised_rationale", ""), | |
| revised_tradeoffs=tuple(data.get("revised_tradeoffs") or ()), | |
| extensions=_frozen_mapping(data.get("extensions", {}), "extensions"), | |
| ) | |
| def to_dict(self) -> dict[str, Any]: | |
| return { | |
| "schema_version": self.schema_version, | |
| "candidate_id": self.candidate_id, | |
| "status": self.status, | |
| "compatible": self.compatible, | |
| "rationale": self.rationale, | |
| "evidence_ids": list(self.evidence_ids), | |
| "assumptions": list(self.assumptions), | |
| "weaknesses": list(self.weaknesses), | |
| "warnings": list(self.warnings), | |
| "revised_rationale": self.revised_rationale, | |
| "revised_tradeoffs": list(self.revised_tradeoffs), | |
| "extensions": _thaw_json(self.extensions), | |
| } | |
| class ResearchVerification: | |
| """Versioned verifier output plus one concise, auditable decision trace.""" | |
| schema_version: int = SCHEMA_VERSION | |
| verification_id: str = "" | |
| source: DecisionSource = DecisionSource.FALLBACK | |
| status: str = "approved" | |
| summary: str = "" | |
| candidate_results: tuple[CandidateVerification, ...] = () | |
| evidence_ids: tuple[str, ...] = () | |
| warnings: tuple[str, ...] = () | |
| decision_trace: DecisionTrace | None = None | |
| provider_provenance: Mapping[str, Any] = field(default_factory=dict) | |
| extensions: Mapping[str, Any] = field(default_factory=dict) | |
| def __post_init__(self) -> None: | |
| if self.schema_version != SCHEMA_VERSION: | |
| raise ValueError(f"unsupported schema_version {self.schema_version!r}") | |
| object.__setattr__( | |
| self, | |
| "verification_id", | |
| _text(self.verification_id, "verification_id", required=True), | |
| ) | |
| try: | |
| source = DecisionSource(self.source) | |
| except (TypeError, ValueError) as exc: | |
| raise ValueError(f"invalid verification source {self.source!r}") from exc | |
| if source not in _VERIFICATION_SOURCES: | |
| raise ValueError("verification source must be llm or fallback") | |
| object.__setattr__(self, "source", source) | |
| status = _text(self.status, "status", required=True).lower() | |
| if status not in {"approved", "revised", "partial", "rejected"}: | |
| raise ValueError(f"invalid research verification status {status!r}") | |
| object.__setattr__(self, "status", status) | |
| object.__setattr__(self, "summary", _rationale(self.summary)) | |
| if not 1 <= len(self.candidate_results) <= 3 or not all( | |
| isinstance(item, CandidateVerification) for item in self.candidate_results | |
| ): | |
| raise ValueError("candidate_results must contain one to three validations") | |
| identifiers = [item.candidate_id for item in self.candidate_results] | |
| if len(identifiers) != len(set(identifiers)): | |
| raise ValueError("candidate verification IDs must be unique") | |
| object.__setattr__(self, "candidate_results", tuple(self.candidate_results)) | |
| for name in ("evidence_ids", "warnings"): | |
| values = _strings(getattr(self, name), name) | |
| if len(values) > 64 or len(values) != len(set(values)): | |
| raise ValueError(f"{name} must be a bounded unique array") | |
| object.__setattr__(self, name, values) | |
| if self.decision_trace is not None and not isinstance( | |
| self.decision_trace, DecisionTrace | |
| ): | |
| raise ValueError("decision_trace must be a DecisionTrace or null") | |
| if self.decision_trace is not None and self.decision_trace.decision_source != source: | |
| raise ValueError("verification trace source must match verification source") | |
| for name in ("provider_provenance", "extensions"): | |
| object.__setattr__(self, name, _frozen_mapping(getattr(self, name), name)) | |
| def from_mapping(cls, payload: Mapping[str, Any]) -> "ResearchVerification": | |
| data = _mapping(payload, "research verification") | |
| _version(data) | |
| _assert_no_internal_reasoning(data) | |
| allowed = { | |
| "schema_version", | |
| "verification_id", | |
| "source", | |
| "status", | |
| "summary", | |
| "candidate_results", | |
| "evidence_ids", | |
| "warnings", | |
| "decision_trace", | |
| "provider_provenance", | |
| "extensions", | |
| } | |
| if set(data) - allowed: | |
| raise ValueError("research verification contains unsupported fields") | |
| raw_results = data.get("candidate_results") or () | |
| if not isinstance(raw_results, (list, tuple)): | |
| raise ValueError("candidate_results must be an array") | |
| raw_trace = data.get("decision_trace") | |
| return cls( | |
| verification_id=data.get("verification_id", ""), | |
| source=data.get("source", DecisionSource.FALLBACK.value), | |
| status=data.get("status", "approved"), | |
| summary=data.get("summary", ""), | |
| candidate_results=tuple( | |
| item | |
| if isinstance(item, CandidateVerification) | |
| else CandidateVerification.from_mapping(_mapping(item, "candidate verification")) | |
| for item in raw_results | |
| ), | |
| evidence_ids=tuple(data.get("evidence_ids") or ()), | |
| warnings=tuple(data.get("warnings") or ()), | |
| decision_trace=( | |
| None | |
| if raw_trace is None | |
| else raw_trace | |
| if isinstance(raw_trace, DecisionTrace) | |
| else DecisionTrace.from_mapping(_mapping(raw_trace, "decision trace")) | |
| ), | |
| provider_provenance=_frozen_mapping( | |
| data.get("provider_provenance", {}), "provider_provenance" | |
| ), | |
| extensions=_frozen_mapping(data.get("extensions", {}), "extensions"), | |
| ) | |
| def to_dict(self) -> dict[str, Any]: | |
| return { | |
| "schema_version": self.schema_version, | |
| "verification_id": self.verification_id, | |
| "source": self.source.value, | |
| "status": self.status, | |
| "summary": self.summary, | |
| "candidate_results": [item.to_dict() for item in self.candidate_results], | |
| "evidence_ids": list(self.evidence_ids), | |
| "warnings": list(self.warnings), | |
| "decision_trace": self.decision_trace.to_dict() if self.decision_trace else None, | |
| "provider_provenance": _thaw_json(self.provider_provenance), | |
| "extensions": _thaw_json(self.extensions), | |
| } | |
| class NextExperimentSuggestion: | |
| """One non-executing next-step proposal emitted by experiment reflection.""" | |
| schema_version: int = SCHEMA_VERSION | |
| suggestion_id: str = "" | |
| title: str = "" | |
| task_type: str = "" | |
| stages: tuple[Mapping[str, Any], ...] = () | |
| rationale: str = "" | |
| expected_tradeoffs: tuple[str, ...] = () | |
| expected_metrics: Mapping[str, Any] = field(default_factory=dict) | |
| evidence_ids: tuple[str, ...] = () | |
| executable: bool = False | |
| suggestion_only: bool = True | |
| validation_status: str = "suggestion_only" | |
| warnings: tuple[str, ...] = () | |
| extensions: Mapping[str, Any] = field(default_factory=dict) | |
| def __post_init__(self) -> None: | |
| if self.schema_version != SCHEMA_VERSION: | |
| raise ValueError(f"unsupported schema_version {self.schema_version!r}") | |
| for name in ("suggestion_id", "title", "task_type"): | |
| object.__setattr__(self, name, _text(getattr(self, name), name, required=True)) | |
| stages = _freeze_json(self.stages, "stages") | |
| if not all(isinstance(stage, Mapping) for stage in stages): | |
| raise ValueError("stages must be an array of mappings") | |
| object.__setattr__(self, "stages", stages) | |
| object.__setattr__(self, "rationale", _rationale(self.rationale)) | |
| for name in ("expected_tradeoffs", "evidence_ids", "warnings"): | |
| values = _strings(getattr(self, name), name) | |
| if len(values) > 32 or len(values) != len(set(values)): | |
| raise ValueError(f"{name} must be a bounded unique array") | |
| object.__setattr__(self, name, values) | |
| object.__setattr__( | |
| self, | |
| "expected_metrics", | |
| _metrics(self.expected_metrics, "expected_metrics", "expected"), | |
| ) | |
| if not isinstance(self.executable, bool) or not isinstance(self.suggestion_only, bool): | |
| raise ValueError("executable and suggestion_only must be boolean") | |
| if self.executable == self.suggestion_only: | |
| raise ValueError("suggestion must be executable or suggestion-only, not both") | |
| validation_status = _text( | |
| self.validation_status, "validation_status", required=True | |
| ).lower() | |
| if validation_status not in {"compatible", "incompatible", "suggestion_only"}: | |
| raise ValueError("invalid suggestion validation_status") | |
| if self.executable and validation_status != "compatible": | |
| raise ValueError("executable suggestions must be compatible") | |
| object.__setattr__(self, "validation_status", validation_status) | |
| object.__setattr__(self, "extensions", _frozen_mapping(self.extensions, "extensions")) | |
| def from_mapping(cls, payload: Mapping[str, Any]) -> "NextExperimentSuggestion": | |
| data = _mapping(payload, "next experiment suggestion") | |
| _version(data) | |
| _assert_no_internal_reasoning(data) | |
| allowed = { | |
| "schema_version", | |
| "suggestion_id", | |
| "title", | |
| "task_type", | |
| "stages", | |
| "rationale", | |
| "expected_tradeoffs", | |
| "expected_metrics", | |
| "evidence_ids", | |
| "executable", | |
| "suggestion_only", | |
| "validation_status", | |
| "warnings", | |
| "extensions", | |
| } | |
| if set(data) - allowed: | |
| raise ValueError("next experiment suggestion contains unsupported fields") | |
| return cls( | |
| suggestion_id=data.get("suggestion_id", ""), | |
| title=data.get("title", ""), | |
| task_type=data.get("task_type", ""), | |
| stages=tuple(data.get("stages") or ()), | |
| rationale=data.get("rationale", ""), | |
| expected_tradeoffs=tuple(data.get("expected_tradeoffs") or ()), | |
| expected_metrics=data.get("expected_metrics") or {}, | |
| evidence_ids=tuple(data.get("evidence_ids") or ()), | |
| executable=data.get("executable", False), | |
| suggestion_only=data.get("suggestion_only", True), | |
| validation_status=data.get("validation_status", "suggestion_only"), | |
| warnings=tuple(data.get("warnings") or ()), | |
| extensions=_frozen_mapping(data.get("extensions", {}), "extensions"), | |
| ) | |
| def to_dict(self) -> dict[str, Any]: | |
| return { | |
| "schema_version": self.schema_version, | |
| "suggestion_id": self.suggestion_id, | |
| "title": self.title, | |
| "task_type": self.task_type, | |
| "stages": _thaw_json(self.stages), | |
| "rationale": self.rationale, | |
| "expected_tradeoffs": list(self.expected_tradeoffs), | |
| "expected_metrics": _thaw_json(self.expected_metrics), | |
| "evidence_ids": list(self.evidence_ids), | |
| "executable": self.executable, | |
| "suggestion_only": self.suggestion_only, | |
| "validation_status": self.validation_status, | |
| "warnings": list(self.warnings), | |
| "extensions": _thaw_json(self.extensions), | |
| } | |
| class ExperimentReflection: | |
| """Sanitized post-experiment interpretation with no execution authority.""" | |
| schema_version: int = SCHEMA_VERSION | |
| reflection_id: str = "" | |
| plan_id: str = "" | |
| source: DecisionSource = DecisionSource.FALLBACK | |
| summary: str = "" | |
| expectations_met: str = "unknown" | |
| limitations: tuple[str, ...] = () | |
| interpretation: str = "" | |
| measured_metrics: Mapping[str, Any] = field(default_factory=dict) | |
| evidence_ids: tuple[str, ...] = () | |
| suggestions: tuple[NextExperimentSuggestion, ...] = () | |
| warnings: tuple[str, ...] = () | |
| decision_trace: DecisionTrace | None = None | |
| provider_provenance: Mapping[str, Any] = field(default_factory=dict) | |
| extensions: Mapping[str, Any] = field(default_factory=dict) | |
| def __post_init__(self) -> None: | |
| if self.schema_version != SCHEMA_VERSION: | |
| raise ValueError(f"unsupported schema_version {self.schema_version!r}") | |
| for name in ("reflection_id", "plan_id"): | |
| object.__setattr__(self, name, _text(getattr(self, name), name, required=True)) | |
| try: | |
| source = DecisionSource(self.source) | |
| except (TypeError, ValueError) as exc: | |
| raise ValueError(f"invalid reflection source {self.source!r}") from exc | |
| if source not in _VERIFICATION_SOURCES: | |
| raise ValueError("reflection source must be llm or fallback") | |
| object.__setattr__(self, "source", source) | |
| object.__setattr__(self, "summary", _rationale(self.summary)) | |
| result = _text(self.expectations_met, "expectations_met", required=True).lower() | |
| if result not in _EXPECTATION_RESULTS: | |
| raise ValueError("invalid expectations_met value") | |
| object.__setattr__(self, "expectations_met", result) | |
| object.__setattr__(self, "interpretation", _rationale(self.interpretation)) | |
| object.__setattr__( | |
| self, | |
| "measured_metrics", | |
| _metrics(self.measured_metrics, "measured_metrics", "measured"), | |
| ) | |
| for name in ("limitations", "evidence_ids", "warnings"): | |
| values = _strings(getattr(self, name), name) | |
| if len(values) > 32 or len(values) != len(set(values)): | |
| raise ValueError(f"{name} must be a bounded unique array") | |
| object.__setattr__(self, name, values) | |
| if not 1 <= len(self.suggestions) <= 3 or not all( | |
| isinstance(item, NextExperimentSuggestion) for item in self.suggestions | |
| ): | |
| raise ValueError("suggestions must contain one to three next experiments") | |
| suggestion_ids = [item.suggestion_id for item in self.suggestions] | |
| if len(suggestion_ids) != len(set(suggestion_ids)): | |
| raise ValueError("suggestion IDs must be unique") | |
| object.__setattr__(self, "suggestions", tuple(self.suggestions)) | |
| if self.decision_trace is not None and not isinstance( | |
| self.decision_trace, DecisionTrace | |
| ): | |
| raise ValueError("decision_trace must be a DecisionTrace or null") | |
| if self.decision_trace is not None and self.decision_trace.decision_source != source: | |
| raise ValueError("reflection trace source must match reflection source") | |
| for name in ("provider_provenance", "extensions"): | |
| object.__setattr__(self, name, _frozen_mapping(getattr(self, name), name)) | |
| def from_mapping(cls, payload: Mapping[str, Any]) -> "ExperimentReflection": | |
| data = _mapping(payload, "experiment reflection") | |
| _version(data) | |
| _assert_no_internal_reasoning(data) | |
| allowed = { | |
| "schema_version", | |
| "reflection_id", | |
| "plan_id", | |
| "source", | |
| "summary", | |
| "expectations_met", | |
| "limitations", | |
| "interpretation", | |
| "measured_metrics", | |
| "evidence_ids", | |
| "suggestions", | |
| "warnings", | |
| "decision_trace", | |
| "provider_provenance", | |
| "extensions", | |
| } | |
| if set(data) - allowed: | |
| raise ValueError("experiment reflection contains unsupported fields") | |
| raw_suggestions = data.get("suggestions") or () | |
| if not isinstance(raw_suggestions, (list, tuple)): | |
| raise ValueError("suggestions must be an array") | |
| raw_trace = data.get("decision_trace") | |
| return cls( | |
| reflection_id=data.get("reflection_id", ""), | |
| plan_id=data.get("plan_id", ""), | |
| source=data.get("source", DecisionSource.FALLBACK.value), | |
| summary=data.get("summary", ""), | |
| expectations_met=data.get("expectations_met", "unknown"), | |
| limitations=tuple(data.get("limitations") or ()), | |
| interpretation=data.get("interpretation", ""), | |
| measured_metrics=data.get("measured_metrics") or {}, | |
| evidence_ids=tuple(data.get("evidence_ids") or ()), | |
| suggestions=tuple( | |
| item | |
| if isinstance(item, NextExperimentSuggestion) | |
| else NextExperimentSuggestion.from_mapping( | |
| _mapping(item, "next experiment suggestion") | |
| ) | |
| for item in raw_suggestions | |
| ), | |
| warnings=tuple(data.get("warnings") or ()), | |
| decision_trace=( | |
| None | |
| if raw_trace is None | |
| else raw_trace | |
| if isinstance(raw_trace, DecisionTrace) | |
| else DecisionTrace.from_mapping(_mapping(raw_trace, "decision trace")) | |
| ), | |
| provider_provenance=_frozen_mapping( | |
| data.get("provider_provenance", {}), "provider_provenance" | |
| ), | |
| extensions=_frozen_mapping(data.get("extensions", {}), "extensions"), | |
| ) | |
| def to_dict(self) -> dict[str, Any]: | |
| return { | |
| "schema_version": self.schema_version, | |
| "reflection_id": self.reflection_id, | |
| "plan_id": self.plan_id, | |
| "source": self.source.value, | |
| "summary": self.summary, | |
| "expectations_met": self.expectations_met, | |
| "limitations": list(self.limitations), | |
| "interpretation": self.interpretation, | |
| "measured_metrics": _thaw_json(self.measured_metrics), | |
| "evidence_ids": list(self.evidence_ids), | |
| "suggestions": [item.to_dict() for item in self.suggestions], | |
| "warnings": list(self.warnings), | |
| "decision_trace": self.decision_trace.to_dict() if self.decision_trace else None, | |
| "provider_provenance": _thaw_json(self.provider_provenance), | |
| "extensions": _thaw_json(self.extensions), | |
| } | |
| class ResearchPlan: | |
| """UI-ready collection of candidate pipelines and recommendation evidence.""" | |
| schema_version: int = SCHEMA_VERSION | |
| plan_id: str = "" | |
| created_at_utc: str = "" | |
| task_summary: str = "" | |
| candidates: tuple[CandidatePipeline, ...] = () | |
| evidence: tuple[Mapping[str, Any], ...] = () | |
| assumptions: tuple[str, ...] = () | |
| expected_tradeoffs: tuple[str, ...] = () | |
| recommendation: Mapping[str, Any] = field(default_factory=dict) | |
| warnings: tuple[str, ...] = () | |
| decision_provenance: tuple[DecisionTrace, ...] = () | |
| verification: ResearchVerification | None = None | |
| feature_extraction_report: FeatureExtractionReport | None = None | |
| provider_provenance: Mapping[str, Any] = field(default_factory=dict) | |
| extensions: Mapping[str, Any] = field(default_factory=dict) | |
| def __post_init__(self) -> None: | |
| if self.schema_version != SCHEMA_VERSION: | |
| raise ValueError(f"unsupported schema_version {self.schema_version!r}") | |
| if not all(isinstance(item, CandidatePipeline) for item in self.candidates): | |
| raise ValueError("candidates must contain CandidatePipeline values") | |
| ids = [candidate.candidate_id for candidate in self.candidates] | |
| if len(ids) != len(set(ids)): | |
| raise ValueError("candidate_id values must be unique within a ResearchPlan") | |
| object.__setattr__(self, "candidates", tuple(self.candidates)) | |
| frozen_evidence = _freeze_json(self.evidence, "evidence") | |
| if not all(isinstance(item, Mapping) for item in frozen_evidence): | |
| raise ValueError("evidence must be an array of mappings") | |
| object.__setattr__(self, "evidence", frozen_evidence) | |
| for name in ("assumptions", "expected_tradeoffs", "warnings"): | |
| object.__setattr__(self, name, _strings(getattr(self, name), name)) | |
| recommendation = _frozen_mapping(self.recommendation, "recommendation") | |
| recommended_id = recommendation.get("candidate_id") | |
| if recommendation and ( | |
| not isinstance(recommended_id, str) or not recommended_id.strip() | |
| ): | |
| raise ValueError("recommendation must include a non-empty candidate_id") | |
| if recommended_id and recommended_id not in set(ids): | |
| raise ValueError( | |
| f"recommendation references unknown candidate_id {recommended_id!r}" | |
| ) | |
| object.__setattr__(self, "recommendation", recommendation) | |
| if not all(isinstance(item, DecisionTrace) for item in self.decision_provenance): | |
| raise ValueError("decision_provenance must contain DecisionTrace values") | |
| object.__setattr__(self, "decision_provenance", tuple(self.decision_provenance)) | |
| if self.verification is not None and not isinstance( | |
| self.verification, ResearchVerification | |
| ): | |
| raise ValueError("verification must be a ResearchVerification or null") | |
| if self.verification is not None: | |
| candidate_ids = {item.candidate_id for item in self.candidates} | |
| verified_ids = { | |
| item.candidate_id for item in self.verification.candidate_results | |
| } | |
| if not verified_ids <= candidate_ids: | |
| raise ValueError("verification references an unknown candidate") | |
| if self.feature_extraction_report is not None and not isinstance( | |
| self.feature_extraction_report, FeatureExtractionReport | |
| ): | |
| raise ValueError( | |
| "feature_extraction_report must be a FeatureExtractionReport or null" | |
| ) | |
| for name in ("provider_provenance", "extensions"): | |
| object.__setattr__(self, name, _frozen_mapping(getattr(self, name), name)) | |
| def from_mapping(cls, payload: Mapping[str, Any]) -> "ResearchPlan": | |
| data = _mapping(payload) | |
| _version(data) | |
| _assert_no_internal_reasoning(data) | |
| consumed: set[str] = set() | |
| def get( | |
| name: str, | |
| aliases: Sequence[str] = (), | |
| default: Any = None, | |
| ) -> Any: | |
| value, used = _resolve(data, name, aliases, default=default) | |
| consumed.update(used) | |
| return value | |
| candidates_raw = get("candidates", ("candidate_pipelines",), ()) | |
| if not isinstance(candidates_raw, (list, tuple)): | |
| raise ValueError("candidates must be an array") | |
| candidates = tuple( | |
| candidate | |
| if isinstance(candidate, CandidatePipeline) | |
| else CandidatePipeline.from_mapping(_mapping(candidate, "candidate")) | |
| for candidate in candidates_raw | |
| ) | |
| recommendation_raw = get("recommendation", (), _MISSING) | |
| legacy_id = get("recommended_candidate_id", (), _MISSING) | |
| legacy_reason = get("recommendation_reason", (), _MISSING) | |
| if recommendation_raw is _MISSING: | |
| recommendation: dict[str, Any] = {} | |
| if legacy_id is not _MISSING: | |
| recommendation["candidate_id"] = legacy_id | |
| if legacy_reason is not _MISSING: | |
| recommendation["rationale"] = legacy_reason | |
| else: | |
| recommendation = dict(_mapping(recommendation_raw, "recommendation")) | |
| comparisons = ( | |
| ("candidate_id", legacy_id, "recommended_candidate_id"), | |
| ("rationale", legacy_reason, "recommendation_reason"), | |
| ) | |
| for canonical_key, legacy_value, legacy_key in comparisons: | |
| if legacy_value is _MISSING: | |
| continue | |
| if ( | |
| canonical_key in recommendation | |
| and _plain(recommendation[canonical_key]) != _plain(legacy_value) | |
| ): | |
| raise ValueError( | |
| "conflicting values for recommendation." | |
| f"{canonical_key!s} and {legacy_key!r}" | |
| ) | |
| recommendation[canonical_key] = legacy_value | |
| if "rationale" in recommendation: | |
| recommendation["rationale"] = _rationale(recommendation["rationale"]) | |
| evidence_raw = get("evidence", ("supporting_evidence",), ()) | |
| if not isinstance(evidence_raw, (list, tuple)) or not all( | |
| isinstance(item, Mapping) for item in evidence_raw | |
| ): | |
| raise ValueError("evidence must be an array of mappings") | |
| decisions_raw = get( | |
| "decision_provenance", ("decision_trace", "decision_traces"), () | |
| ) | |
| if isinstance(decisions_raw, Mapping): | |
| decisions_raw = [decisions_raw] | |
| if not isinstance(decisions_raw, (list, tuple)): | |
| raise ValueError("decision_provenance must be an array") | |
| decisions = tuple( | |
| trace | |
| if isinstance(trace, DecisionTrace) | |
| else DecisionTrace.from_mapping(_mapping(trace, "decision trace")) | |
| for trace in decisions_raw | |
| ) | |
| feature_raw = get( | |
| "feature_extraction_report", ("feature_report",), None | |
| ) | |
| feature_report = ( | |
| None | |
| if feature_raw is None | |
| else feature_raw | |
| if isinstance(feature_raw, FeatureExtractionReport) | |
| else FeatureExtractionReport.from_mapping( | |
| _mapping(feature_raw, "feature_extraction_report") | |
| ) | |
| ) | |
| verification_raw = get("verification", ("research_verification",), None) | |
| verification = ( | |
| None | |
| if verification_raw is None | |
| else verification_raw | |
| if isinstance(verification_raw, ResearchVerification) | |
| else ResearchVerification.from_mapping( | |
| _mapping(verification_raw, "verification") | |
| ) | |
| ) | |
| return cls( | |
| plan_id=_text(get("plan_id", ("id",), ""), "plan_id"), | |
| created_at_utc=_text( | |
| get("created_at_utc", ("created_at",), ""), "created_at_utc" | |
| ), | |
| task_summary=_text( | |
| get("task_summary", ("objective",), ""), "task_summary" | |
| ), | |
| candidates=candidates, | |
| evidence=tuple(evidence_raw), | |
| assumptions=_strings(get("assumptions", (), ()), "assumptions"), | |
| expected_tradeoffs=_strings( | |
| get("expected_tradeoffs", ("tradeoffs",), ()), | |
| "expected_tradeoffs", | |
| ), | |
| recommendation=recommendation, | |
| warnings=_strings(get("warnings", (), ()), "warnings"), | |
| decision_provenance=decisions, | |
| verification=verification, | |
| feature_extraction_report=feature_report, | |
| provider_provenance=_frozen_mapping( | |
| get("provider_provenance", ("provider",), {}), | |
| "provider_provenance", | |
| ), | |
| extensions=_extensions(data, consumed), | |
| ) | |
| def to_dict(self) -> dict[str, Any]: | |
| return { | |
| "schema_version": self.schema_version, | |
| "plan_id": self.plan_id, | |
| "created_at_utc": self.created_at_utc, | |
| "task_summary": self.task_summary, | |
| "candidates": [candidate.to_dict() for candidate in self.candidates], | |
| "evidence": _thaw_json(self.evidence), | |
| "assumptions": list(self.assumptions), | |
| "expected_tradeoffs": list(self.expected_tradeoffs), | |
| "recommendation": _thaw_json(self.recommendation), | |
| "warnings": list(self.warnings), | |
| "decision_provenance": [ | |
| trace.to_dict() for trace in self.decision_provenance | |
| ], | |
| "verification": ( | |
| self.verification.to_dict() if self.verification is not None else None | |
| ), | |
| "feature_extraction_report": ( | |
| self.feature_extraction_report.to_dict() | |
| if self.feature_extraction_report is not None | |
| else None | |
| ), | |
| "provider_provenance": _thaw_json(self.provider_provenance), | |
| "extensions": _thaw_json(self.extensions), | |
| } | |
| class ResearchPlanProvider(Protocol): | |
| """Minimal provider boundary for a future research-agent integration.""" | |
| def create_plan(self, context: Mapping[str, Any] | None = None) -> ResearchPlan: | |
| """Return one validated research plan.""" | |
| class MockResearchPlanProvider: | |
| """Deterministic provider for UI development without an LLM or network call.""" | |
| def create_plan(self, context: Mapping[str, Any] | None = None) -> ResearchPlan: | |
| supplied = _mapping(context or {}, "context") | |
| dataset_name = _text(supplied.get("dataset_name", "dataset"), "dataset_name") | |
| task_type = _text( | |
| supplied.get("task_type", "classification"), "task_type" | |
| ) | |
| target = _text(supplied.get("target_column", "target"), "target_column") | |
| modality = _text(supplied.get("modality", "tabular"), "modality").lower() | |
| if modality not in {"tabular", "image"}: | |
| modality = "tabular" | |
| if modality == "image": | |
| candidates = [ | |
| { | |
| "candidate_id": "mock-image-balanced", | |
| "name": "Balanced deep features", | |
| "task_type": task_type, | |
| "stages": [ | |
| { | |
| "stage": "preprocess", | |
| "component": "automatic_image_preprocessing", | |
| }, | |
| { | |
| "stage": "feature_extractor", | |
| "component": "frozen_cnn_embeddings", | |
| }, | |
| {"stage": "model", "component": "automatic_classifier"}, | |
| ], | |
| "rationale": "A balanced image baseline using frozen deep features.", | |
| "evidence_ids": ["mock-image-heuristic"], | |
| "assumptions": [ | |
| "Image labels and file paths pass backend validation." | |
| ], | |
| "expected_tradeoffs": [ | |
| "Stronger representation capacity with moderate GPU startup cost." | |
| ], | |
| "expected_metrics": { | |
| "relative_quality": { | |
| "kind": "expected", | |
| "estimate": "medium-high", | |
| "basis": "deterministic mock image heuristic", | |
| } | |
| }, | |
| "measured_metrics": {}, | |
| }, | |
| { | |
| "candidate_id": "mock-image-efficient", | |
| "name": "Efficient image baseline", | |
| "task_type": task_type, | |
| "stages": [ | |
| {"stage": "preprocess", "component": "resize_and_normalize"}, | |
| { | |
| "stage": "feature_extractor", | |
| "component": "resnet18_frozen_features", | |
| }, | |
| {"stage": "model", "component": "linear_probe"}, | |
| ], | |
| "rationale": "An inspectable image baseline with bounded training cost.", | |
| "evidence_ids": ["mock-image-heuristic"], | |
| "assumptions": [ | |
| "Frozen pretrained features are suitable for the image domain." | |
| ], | |
| "expected_tradeoffs": [ | |
| "Faster and easier to inspect, but may underfit domain-specific patterns." | |
| ], | |
| "expected_metrics": { | |
| "relative_quality": { | |
| "kind": "expected", | |
| "estimate": "medium", | |
| "basis": "deterministic mock image heuristic", | |
| } | |
| }, | |
| "measured_metrics": {}, | |
| }, | |
| ] | |
| evidence = [ | |
| { | |
| "evidence_id": "mock-image-heuristic", | |
| "source": "mock", | |
| "summary": "Placeholder image-workload heuristic for UI development.", | |
| } | |
| ] | |
| recommendation = { | |
| "candidate_id": "mock-image-balanced", | |
| "rationale": "Recommended by the deterministic image mock rule.", | |
| } | |
| selected_ids = ["mock-image-balanced", "mock-image-efficient"] | |
| else: | |
| candidates = [ | |
| { | |
| "candidate_id": "mock-balanced-ensemble", | |
| "name": "Balanced ensemble", | |
| "task_type": task_type, | |
| "stages": [ | |
| {"stage": "preprocess", "component": "automatic"}, | |
| {"stage": "model", "component": "tree_ensemble"}, | |
| ], | |
| "rationale": "A balanced baseline for mixed feature types.", | |
| "evidence_ids": ["mock-project-heuristic"], | |
| "assumptions": ["The supplied target is usable for training."], | |
| "expected_tradeoffs": [ | |
| "Moderate runtime with less interpretability than a linear model." | |
| ], | |
| "expected_metrics": { | |
| "relative_quality": { | |
| "kind": "expected", | |
| "estimate": "medium-high", | |
| "basis": "deterministic mock heuristic", | |
| } | |
| }, | |
| "measured_metrics": {}, | |
| }, | |
| { | |
| "candidate_id": "mock-interpretable-baseline", | |
| "name": "Interpretable baseline", | |
| "task_type": task_type, | |
| "stages": [ | |
| {"stage": "preprocess", "component": "automatic"}, | |
| {"stage": "model", "component": "linear_baseline"}, | |
| ], | |
| "rationale": "A transparent comparison candidate.", | |
| "evidence_ids": ["mock-project-heuristic"], | |
| "assumptions": [ | |
| "A simpler decision boundary is a useful baseline." | |
| ], | |
| "expected_tradeoffs": [ | |
| "Easier to inspect but may underfit nonlinear relationships." | |
| ], | |
| "expected_metrics": { | |
| "relative_quality": { | |
| "kind": "expected", | |
| "estimate": "medium", | |
| "basis": "deterministic mock heuristic", | |
| } | |
| }, | |
| "measured_metrics": {}, | |
| }, | |
| ] | |
| evidence = [ | |
| { | |
| "evidence_id": "mock-project-heuristic", | |
| "source": "mock", | |
| "summary": "Placeholder project heuristic for UI development.", | |
| } | |
| ] | |
| recommendation = { | |
| "candidate_id": "mock-balanced-ensemble", | |
| "rationale": "Recommended by a deterministic mock rule for UI development.", | |
| } | |
| selected_ids = ["mock-balanced-ensemble", "mock-interpretable-baseline"] | |
| return ResearchPlan.from_mapping( | |
| { | |
| "schema_version": 1, | |
| "plan_id": "mock-research-plan-v1", | |
| "created_at_utc": "2000-01-01T00:00:00Z", | |
| "task_summary": ( | |
| f"Model {target or 'target'} for {dataset_name or 'dataset'} " | |
| f"as {task_type or 'classification'}." | |
| ), | |
| "candidates": candidates, | |
| "evidence": evidence, | |
| "assumptions": ["Training data and labels will be validated separately."], | |
| "expected_tradeoffs": [ | |
| "Expected quality, runtime, and interpretability must be validated." | |
| ], | |
| "recommendation": recommendation, | |
| "warnings": [ | |
| "All candidate metrics are expected estimates and are not measured results." | |
| ], | |
| "decision_provenance": [ | |
| { | |
| "schema_version": 1, | |
| "trace_id": "mock-candidate-preset", | |
| "decision_type": "candidate_set", | |
| "decision_source": "preset", | |
| "selected": selected_ids, | |
| "rationale": "The mock provider exposes two stable UI fixtures.", | |
| }, | |
| { | |
| "schema_version": 1, | |
| "trace_id": "mock-recommendation-rule", | |
| "decision_type": "recommend_candidate", | |
| "decision_source": "deterministic_rule", | |
| "selected": recommendation["candidate_id"], | |
| "options": selected_ids, | |
| "rationale": "The balanced fixture is the deterministic default.", | |
| }, | |
| ], | |
| "provider_provenance": { | |
| "provider": "mock", | |
| "provider_version": 1, | |
| "modality": modality, | |
| "network_used": False, | |
| "llm_used": False, | |
| }, | |
| } | |
| ) | |
| __all__ = [ | |
| "CandidateVerification", | |
| "CandidatePipeline", | |
| "DecisionSource", | |
| "DecisionTrace", | |
| "ExperimentReflection", | |
| "FeatureExtractionReport", | |
| "MockResearchPlanProvider", | |
| "NextExperimentSuggestion", | |
| "ResearchPlan", | |
| "ResearchPlanProvider", | |
| "ResearchVerification", | |
| "SCHEMA_VERSION", | |
| ] | |