"""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") @dataclass(frozen=True) 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")) @classmethod 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), } @dataclass(frozen=True) 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")) @classmethod 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) @dataclass(frozen=True) 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)) @classmethod def from_mapping(cls, payload: Mapping[str, Any]) -> "CandidatePipeline": return cls._from_mapping(payload) @classmethod def from_legacy_research_agent( cls, payload: Mapping[str, Any] ) -> "CandidatePipeline": return cls._from_mapping( payload, generic_metrics_kind="expected", adapter_name="research_agent" ) @classmethod def from_legacy_benchmark( cls, payload: Mapping[str, Any] ) -> "CandidatePipeline": return cls._from_mapping( payload, generic_metrics_kind="measured", adapter_name="benchmark" ) @classmethod 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"} @dataclass(frozen=True) 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")) @classmethod 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), } @dataclass(frozen=True) 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)) @classmethod 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), } @dataclass(frozen=True) 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")) @classmethod 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), } @dataclass(frozen=True) 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)) @classmethod 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), } @dataclass(frozen=True) 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)) @classmethod 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", ]