Spaces:
Sleeping
Sleeping
| """Pure UI state and presentation helpers for the Guided Research Workbench. | |
| The module is intentionally independent from Gradio, gateway routing, and | |
| training code. It consumes only the stable UI contracts so state transitions | |
| and rendering can be tested without waking a backend. | |
| """ | |
| from __future__ import annotations | |
| import html | |
| import json | |
| import uuid | |
| from dataclasses import dataclass, field, replace | |
| from typing import Any, Mapping, Sequence | |
| from execution_metadata import normalize_execution_metadata | |
| from ui_contracts import ( | |
| CandidateVerification, | |
| CandidatePipeline, | |
| DecisionSource, | |
| DecisionTrace, | |
| ExperimentReflection, | |
| FeatureExtractionReport, | |
| MockResearchPlanProvider, | |
| ResearchPlan, | |
| ) | |
| WORKBENCH_STEPS = ( | |
| "data", | |
| "objective", | |
| "research", | |
| "pipeline", | |
| "compute", | |
| "review", | |
| "run", | |
| ) | |
| PIPELINE_NODE_IDS = ( | |
| "input", | |
| "preprocessing", | |
| "feature_extractor", | |
| "postprocessing", | |
| "model", | |
| "evaluation", | |
| ) | |
| PIPELINE_NODE_PRESENTATION = { | |
| "input": ("IN", "Input"), | |
| "preprocessing": ("PR", "Preprocessing"), | |
| "feature_extractor": ("FX", "Feature extractor"), | |
| "postprocessing": ("PO", "Post-processing"), | |
| "model": ("ML", "Model"), | |
| "evaluation": ("EV", "Evaluation"), | |
| } | |
| # Six stages share one row of the pipeline track, so the visible stage chip | |
| # needs labels that survive a ~130px card. The full names above still carry the | |
| # accessible name for each card. | |
| PIPELINE_NODE_SHORT_LABELS = { | |
| "input": "Input", | |
| "preprocessing": "Prepare", | |
| "feature_extractor": "Features", | |
| "postprocessing": "Transform", | |
| "model": "Model", | |
| "evaluation": "Evaluate", | |
| } | |
| SOURCE_LABELS = { | |
| DecisionSource.USER: "User", | |
| DecisionSource.PRESET: "Preset", | |
| DecisionSource.DETERMINISTIC_RULE: "Deterministic Rule", | |
| DecisionSource.BENCHMARK: "Benchmark", | |
| DecisionSource.RESEARCH_AGENT: "Research Agent", | |
| DecisionSource.LLM: "LLM", | |
| DecisionSource.FALLBACK: "Fallback", | |
| } | |
| _DATA_SOURCES = {"managed", "upload"} | |
| _DATA_MODALITIES = {"tabular", "image"} | |
| _AUTOMATIC_VALUES = {"", "auto", "automatic selection"} | |
| def _json_copy(value: Any) -> Any: | |
| return json.loads(json.dumps(value)) | |
| def _clean_text(value: Any, *, limit: int = 2_000) -> str: | |
| text = " ".join(str(value or "").split()) | |
| return text[:limit] | |
| def _positive_int(value: Any, field_name: str) -> int: | |
| if isinstance(value, bool): | |
| raise ValueError(f"{field_name} must be an integer") | |
| try: | |
| result = int(value) | |
| except (TypeError, ValueError) as exc: | |
| raise ValueError(f"{field_name} must be an integer") from exc | |
| if result < 1: | |
| raise ValueError(f"{field_name} must be positive") | |
| return result | |
| def _constraints(value: Any) -> tuple[str, ...]: | |
| if value is None: | |
| return () | |
| if isinstance(value, str): | |
| value = [value] | |
| if not isinstance(value, (list, tuple)): | |
| raise ValueError("constraints must be an array of strings") | |
| cleaned = tuple(_clean_text(item, limit=500) for item in value) | |
| return tuple(item for item in cleaned if item) | |
| def candidate_verification_map( | |
| plan: ResearchPlan, | |
| ) -> dict[str, CandidateVerification]: | |
| """Return the canonical verifier result for each candidate, if present.""" | |
| if plan.verification is None: | |
| return {} | |
| return { | |
| item.candidate_id: item for item in plan.verification.candidate_results | |
| } | |
| def candidate_is_executable(plan: ResearchPlan, candidate_id: str) -> bool: | |
| """Treat verifier rejection as a public UI block, never as execution authority.""" | |
| verification = candidate_verification_map(plan).get(_clean_text(candidate_id)) | |
| return verification is None or ( | |
| verification.compatible | |
| and verification.status in {"approved", "revised"} | |
| ) | |
| class RunVariant: | |
| """Serializable snapshot for one member of a two-run plan.""" | |
| variant_id: str | |
| data_source: str | |
| data_modality: str | |
| candidate_id: str | |
| dataset: str | |
| target: str | |
| trials: int | |
| preset: str | |
| time_limit: int | |
| image_backend: str | |
| research_plan: Mapping[str, Any] | |
| feature_report: Mapping[str, Any] | |
| approved_research: Mapping[str, Any] | |
| def to_dict(self) -> dict[str, Any]: | |
| return { | |
| "variant_id": self.variant_id, | |
| "data_source": self.data_source, | |
| "data_modality": self.data_modality, | |
| "candidate_id": self.candidate_id, | |
| "dataset": self.dataset, | |
| "target": self.target, | |
| "trials": self.trials, | |
| "preset": self.preset, | |
| "time_limit": self.time_limit, | |
| "image_backend": self.image_backend, | |
| "research_plan": _json_copy(dict(self.research_plan)), | |
| "feature_report": _json_copy(dict(self.feature_report)), | |
| "approved_research": _json_copy(dict(self.approved_research)), | |
| } | |
| class PipelineNodeView: | |
| """Presentation-ready projection of one pipeline stage.""" | |
| node_id: str | |
| label: str | |
| component: str | |
| source: DecisionSource | |
| parameters: tuple[tuple[str, str], ...] | |
| input_shape: tuple[int, ...] | |
| output_shape: tuple[int, ...] | |
| compatibility: str | |
| rationale: str | |
| alternatives: tuple[str, ...] | |
| trace: DecisionTrace | |
| feature_report: FeatureExtractionReport | None = None | |
| class WorkbenchState: | |
| """Immutable, JSON-serializable wizard state.""" | |
| schema_version: int = 1 | |
| step: str = "data" | |
| data_source: str = "managed" | |
| data_modality: str = "tabular" | |
| dataset_name: str = "" | |
| upload_type: str = "Tabular CSV" | |
| target: str = "" | |
| task_type: str = "classification" | |
| review_metric: str = "accuracy" | |
| user_objective: str = "" | |
| constraints: tuple[str, ...] = () | |
| research_plan: ResearchPlan | None = None | |
| selected_candidate_id: str = "" | |
| approved_candidate_id: str = "" | |
| approved_pipeline_config: Mapping[str, Any] = field(default_factory=dict) | |
| approved_evidence_ids: tuple[str, ...] = () | |
| approval_trace: DecisionTrace | None = None | |
| approval_error: str = "" | |
| selected_node_id: str = "input" | |
| preset: str = "medium_quality" | |
| time_limit: int = 300 | |
| trials: int = 1 | |
| image_backend: str = "Auto" | |
| feature_tool: str = "Automatic selection" | |
| transform_tool: str = "Automatic selection" | |
| ml_tool: str = "Automatic selection" | |
| variants: tuple[RunVariant, ...] = () | |
| provider_error: str = "" | |
| def __post_init__(self) -> None: | |
| if self.schema_version != 1: | |
| raise ValueError( | |
| f"unsupported workbench schema_version {self.schema_version!r}" | |
| ) | |
| if self.step not in WORKBENCH_STEPS: | |
| raise ValueError(f"unknown workbench step {self.step!r}") | |
| if self.data_source not in _DATA_SOURCES: | |
| raise ValueError(f"unknown data source {self.data_source!r}") | |
| if self.data_modality not in _DATA_MODALITIES: | |
| raise ValueError(f"unknown data modality {self.data_modality!r}") | |
| if self.selected_node_id not in PIPELINE_NODE_IDS: | |
| raise ValueError(f"unknown pipeline node {self.selected_node_id!r}") | |
| object.__setattr__(self, "constraints", _constraints(self.constraints)) | |
| if not isinstance(self.approved_pipeline_config, Mapping): | |
| raise ValueError("approved_pipeline_config must be a mapping") | |
| object.__setattr__( | |
| self, | |
| "approved_pipeline_config", | |
| _json_copy(dict(self.approved_pipeline_config)), | |
| ) | |
| object.__setattr__( | |
| self, | |
| "approved_evidence_ids", | |
| tuple(_clean_text(item, limit=200) for item in self.approved_evidence_ids), | |
| ) | |
| if self.approval_trace is not None and not isinstance( | |
| self.approval_trace, DecisionTrace | |
| ): | |
| raise ValueError("approval_trace must be a DecisionTrace or null") | |
| object.__setattr__(self, "time_limit", _positive_int(self.time_limit, "time_limit")) | |
| object.__setattr__(self, "trials", _positive_int(self.trials, "trials")) | |
| object.__setattr__(self, "variants", tuple(self.variants)) | |
| def current_candidate(self) -> CandidatePipeline: | |
| if self.research_plan is None: | |
| raise ValueError("Generate a research plan before selecting a candidate.") | |
| for candidate in self.research_plan.candidates: | |
| if candidate.candidate_id == self.selected_candidate_id: | |
| return candidate | |
| raise ValueError("The selected candidate is not present in the research plan.") | |
| def to_dict(self) -> dict[str, Any]: | |
| return { | |
| "schema_version": self.schema_version, | |
| "step": self.step, | |
| "data_source": self.data_source, | |
| "data_modality": self.data_modality, | |
| "dataset_name": self.dataset_name, | |
| "upload_type": self.upload_type, | |
| "target": self.target, | |
| "task_type": self.task_type, | |
| "review_metric": self.review_metric, | |
| "user_objective": self.user_objective, | |
| "constraints": list(self.constraints), | |
| "research_plan": ( | |
| self.research_plan.to_dict() if self.research_plan else None | |
| ), | |
| "selected_candidate_id": self.selected_candidate_id, | |
| "approved_candidate_id": self.approved_candidate_id, | |
| "approved_pipeline_config": _json_copy( | |
| dict(self.approved_pipeline_config) | |
| ), | |
| "approved_evidence_ids": list(self.approved_evidence_ids), | |
| "approval_trace": ( | |
| self.approval_trace.to_dict() if self.approval_trace else None | |
| ), | |
| "approval_error": self.approval_error, | |
| "selected_node_id": self.selected_node_id, | |
| "preset": self.preset, | |
| "time_limit": self.time_limit, | |
| "trials": self.trials, | |
| "image_backend": self.image_backend, | |
| "feature_tool": self.feature_tool, | |
| "transform_tool": self.transform_tool, | |
| "ml_tool": self.ml_tool, | |
| "variants": [variant.to_dict() for variant in self.variants], | |
| "provider_error": self.provider_error, | |
| } | |
| def _variant_from_mapping(value: Mapping[str, Any]) -> RunVariant: | |
| if not isinstance(value, Mapping): | |
| raise ValueError("run variant must be a mapping") | |
| data_source = _clean_text(value.get("data_source") or "managed").lower() | |
| if data_source not in _DATA_SOURCES: | |
| raise ValueError("run variant data_source is invalid") | |
| research_plan = value.get("research_plan") or {} | |
| feature_report = value.get("feature_report") or {} | |
| approved_research = value.get("approved_research") or {} | |
| if not isinstance(research_plan, Mapping) or not isinstance( | |
| feature_report, Mapping | |
| ) or not isinstance(approved_research, Mapping): | |
| raise ValueError("run variant snapshots must be mappings") | |
| return RunVariant( | |
| variant_id=_clean_text(value.get("variant_id")), | |
| data_source=data_source, | |
| data_modality=_clean_text( | |
| value.get("data_modality") or "tabular" | |
| ).lower(), | |
| candidate_id=_clean_text(value.get("candidate_id")), | |
| dataset=_clean_text(value.get("dataset")), | |
| target=_clean_text(value.get("target")), | |
| trials=_positive_int(value.get("trials", 1), "trials"), | |
| preset=_clean_text(value.get("preset") or "medium_quality"), | |
| time_limit=_positive_int(value.get("time_limit", 300), "time_limit"), | |
| image_backend=_clean_text(value.get("image_backend") or "Auto"), | |
| research_plan=_json_copy(dict(research_plan)), | |
| feature_report=_json_copy(dict(feature_report)), | |
| approved_research=_json_copy(dict(approved_research)), | |
| ) | |
| def state_from_mapping(value: Mapping[str, Any] | WorkbenchState) -> WorkbenchState: | |
| """Rebuild state from the plain mapping kept by ``gr.State``.""" | |
| if isinstance(value, WorkbenchState): | |
| return value | |
| if not isinstance(value, Mapping): | |
| raise ValueError("workbench state must be a mapping") | |
| version = value.get("schema_version", 1) | |
| if version != 1: | |
| raise ValueError(f"unsupported workbench schema_version {version!r}") | |
| raw_plan = value.get("research_plan") | |
| plan = ( | |
| None | |
| if raw_plan in (None, {}) | |
| else ResearchPlan.from_mapping(raw_plan) | |
| ) | |
| raw_variants = value.get("variants") or () | |
| if not isinstance(raw_variants, (list, tuple)): | |
| raise ValueError("variants must be an array") | |
| raw_approval_trace = value.get("approval_trace") | |
| approval_trace = ( | |
| None | |
| if raw_approval_trace in (None, {}) | |
| else DecisionTrace.from_mapping(raw_approval_trace) | |
| ) | |
| approved_config = value.get("approved_pipeline_config") or {} | |
| if not isinstance(approved_config, Mapping): | |
| raise ValueError("approved_pipeline_config must be a mapping") | |
| raw_evidence_ids = value.get("approved_evidence_ids") or () | |
| if not isinstance(raw_evidence_ids, (list, tuple)) or not all( | |
| isinstance(item, str) for item in raw_evidence_ids | |
| ): | |
| raise ValueError("approved_evidence_ids must be an array of strings") | |
| return WorkbenchState( | |
| step=_clean_text(value.get("step") or "data").lower(), | |
| data_source=_clean_text(value.get("data_source") or "managed").lower(), | |
| data_modality=_clean_text( | |
| value.get("data_modality") or "tabular" | |
| ).lower(), | |
| dataset_name=_clean_text(value.get("dataset_name")), | |
| upload_type=_clean_text(value.get("upload_type") or "Tabular CSV"), | |
| target=_clean_text(value.get("target")), | |
| task_type=_clean_text(value.get("task_type") or "classification"), | |
| review_metric=_clean_text(value.get("review_metric") or "accuracy"), | |
| user_objective=_clean_text(value.get("user_objective"), limit=1_000), | |
| constraints=_constraints(value.get("constraints")), | |
| research_plan=plan, | |
| selected_candidate_id=_clean_text(value.get("selected_candidate_id")), | |
| approved_candidate_id=_clean_text(value.get("approved_candidate_id")), | |
| approved_pipeline_config=_json_copy(dict(approved_config)), | |
| approved_evidence_ids=tuple(raw_evidence_ids), | |
| approval_trace=approval_trace, | |
| approval_error=_clean_text(value.get("approval_error")), | |
| selected_node_id=_clean_text( | |
| value.get("selected_node_id") or "input" | |
| ).lower(), | |
| preset=_clean_text(value.get("preset") or "medium_quality"), | |
| time_limit=_positive_int(value.get("time_limit", 300), "time_limit"), | |
| trials=_positive_int(value.get("trials", 1), "trials"), | |
| image_backend=_clean_text(value.get("image_backend") or "Auto"), | |
| feature_tool=_clean_text( | |
| value.get("feature_tool") or "Automatic selection" | |
| ), | |
| transform_tool=_clean_text( | |
| value.get("transform_tool") or "Automatic selection" | |
| ), | |
| ml_tool=_clean_text(value.get("ml_tool") or "Automatic selection"), | |
| variants=tuple(_variant_from_mapping(item) for item in raw_variants), | |
| provider_error=_clean_text(value.get("provider_error")), | |
| ) | |
| def new_workbench_state( | |
| *, | |
| data_source: str = "managed", | |
| data_modality: str = "tabular", | |
| dataset_name: str = "", | |
| upload_type: str = "Tabular CSV", | |
| target: str = "", | |
| task_type: str = "classification", | |
| user_objective: str = "", | |
| ) -> WorkbenchState: | |
| return WorkbenchState( | |
| data_source=_clean_text(data_source).lower() or "managed", | |
| data_modality=_clean_text(data_modality).lower() or "tabular", | |
| dataset_name=_clean_text(dataset_name), | |
| upload_type=_clean_text(upload_type) or "Tabular CSV", | |
| target=_clean_text(target), | |
| task_type=_clean_text(task_type) or "classification", | |
| user_objective=_clean_text(user_objective, limit=1_000), | |
| ) | |
| def update_context( | |
| state: Mapping[str, Any] | WorkbenchState, **changes: Any | |
| ) -> WorkbenchState: | |
| current = state_from_mapping(state) | |
| allowed = { | |
| "data_source", | |
| "data_modality", | |
| "dataset_name", | |
| "upload_type", | |
| "target", | |
| "task_type", | |
| "review_metric", | |
| "user_objective", | |
| "constraints", | |
| "preset", | |
| "time_limit", | |
| "trials", | |
| "image_backend", | |
| "feature_tool", | |
| "transform_tool", | |
| "ml_tool", | |
| } | |
| unknown = set(changes) - allowed | |
| if unknown: | |
| raise ValueError(f"unknown workbench field {sorted(unknown)[0]!r}") | |
| normalized = dict(changes) | |
| for name in ( | |
| "data_source", | |
| "data_modality", | |
| "dataset_name", | |
| "upload_type", | |
| "target", | |
| "task_type", | |
| "review_metric", | |
| "user_objective", | |
| "preset", | |
| "image_backend", | |
| "feature_tool", | |
| "transform_tool", | |
| "ml_tool", | |
| ): | |
| if name in normalized: | |
| normalized[name] = _clean_text(normalized[name]) | |
| if "data_source" in normalized: | |
| normalized["data_source"] = normalized["data_source"].lower() | |
| if "data_modality" in normalized: | |
| normalized["data_modality"] = normalized["data_modality"].lower() | |
| if "constraints" in normalized: | |
| normalized["constraints"] = _constraints(normalized["constraints"]) | |
| if "time_limit" in normalized: | |
| normalized["time_limit"] = _positive_int( | |
| normalized["time_limit"], "time_limit" | |
| ) | |
| if "trials" in normalized: | |
| normalized["trials"] = _positive_int(normalized["trials"], "trials") | |
| research_fields = { | |
| "data_source", | |
| "data_modality", | |
| "dataset_name", | |
| "upload_type", | |
| "target", | |
| "task_type", | |
| "review_metric", | |
| "user_objective", | |
| "constraints", | |
| "time_limit", | |
| } | |
| if any( | |
| name in normalized and getattr(current, name) != normalized[name] | |
| for name in research_fields | |
| ): | |
| normalized.update( | |
| research_plan=None, | |
| selected_candidate_id="", | |
| approved_candidate_id="", | |
| approved_pipeline_config={}, | |
| approved_evidence_ids=(), | |
| approval_trace=None, | |
| approval_error="", | |
| variants=(), | |
| provider_error="", | |
| ) | |
| approval_fields = { | |
| "image_backend", | |
| "feature_tool", | |
| "transform_tool", | |
| "ml_tool", | |
| } | |
| if any( | |
| name in normalized and getattr(current, name) != normalized[name] | |
| for name in approval_fields | |
| ): | |
| normalized.update( | |
| approved_candidate_id="", | |
| approved_pipeline_config={}, | |
| approved_evidence_ids=(), | |
| approval_trace=None, | |
| approval_error="", | |
| ) | |
| return replace(current, **normalized) | |
| def set_step( | |
| state: Mapping[str, Any] | WorkbenchState, step: str | |
| ) -> WorkbenchState: | |
| current = state_from_mapping(state) | |
| normalized = _clean_text(step).lower() | |
| if normalized not in WORKBENCH_STEPS: | |
| raise ValueError(f"unknown workbench step {normalized!r}") | |
| return replace(current, step=normalized) | |
| def _fallback_plan(context: Mapping[str, Any]) -> ResearchPlan: | |
| target = _clean_text(context.get("target_column") or "target") | |
| task = _clean_text(context.get("task_type") or "classification") | |
| return ResearchPlan.from_mapping( | |
| { | |
| "schema_version": 1, | |
| "plan_id": "fallback-research-plan-v1", | |
| "task_summary": f"Model {target} as {task} with automatic components.", | |
| "candidates": [ | |
| { | |
| "candidate_id": "fallback-automatic-pipeline", | |
| "name": "Conservative automatic pipeline", | |
| "task_type": task, | |
| "stages": [ | |
| {"stage": "preprocess", "component": "automatic"}, | |
| {"stage": "model", "component": "automatic"}, | |
| ], | |
| "rationale": "Preserves safe existing automatic defaults.", | |
| "expected_tradeoffs": [ | |
| "No agent comparison is available for this fallback." | |
| ], | |
| "expected_metrics": { | |
| "relative_quality": { | |
| "kind": "expected", | |
| "estimate": "unavailable", | |
| "basis": "fallback", | |
| } | |
| }, | |
| "measured_metrics": {}, | |
| "warnings": ["Research planning was unavailable."], | |
| } | |
| ], | |
| "recommendation": { | |
| "candidate_id": "fallback-automatic-pipeline", | |
| "rationale": "Use existing automatic defaults.", | |
| }, | |
| "warnings": [ | |
| "This fallback contains no measured results or live agent evidence." | |
| ], | |
| "decision_provenance": [ | |
| { | |
| "schema_version": 1, | |
| "trace_id": "fallback-plan-selection", | |
| "decision_type": "recommend_candidate", | |
| "decision_source": "fallback", | |
| "selected": "fallback-automatic-pipeline", | |
| "options": ["fallback-automatic-pipeline"], | |
| "rationale": "The research provider was unavailable.", | |
| } | |
| ], | |
| "provider_provenance": { | |
| "provider": "fallback", | |
| "network_used": False, | |
| "llm_used": False, | |
| }, | |
| } | |
| ) | |
| def _plan_with_constraint_recommendation( | |
| plan: ResearchPlan, constraints: Sequence[str] | |
| ) -> ResearchPlan: | |
| combined = " ".join(constraints).lower() | |
| if not any(word in combined for word in ("interpret", "runtime", "fast")): | |
| return plan | |
| preferred = next( | |
| ( | |
| candidate | |
| for candidate in plan.candidates | |
| if "interpret" in ( | |
| f"{candidate.candidate_id} {candidate.name}" | |
| ).lower() | |
| ), | |
| None, | |
| ) | |
| if preferred is None: | |
| preferred = next( | |
| ( | |
| candidate | |
| for candidate in plan.candidates | |
| if any( | |
| word in f"{candidate.candidate_id} {candidate.name}".lower() | |
| for word in ("efficient", "inspectable") | |
| ) | |
| ), | |
| None, | |
| ) | |
| if preferred is None: | |
| return plan | |
| data = plan.to_dict() | |
| data["recommendation"] = { | |
| "candidate_id": preferred.candidate_id, | |
| "rationale": ( | |
| "Selected by a deterministic refinement rule for the stated constraint." | |
| ), | |
| } | |
| data["decision_provenance"] = [ | |
| *data.get("decision_provenance", []), | |
| { | |
| "schema_version": 1, | |
| "trace_id": "mock-refinement-rule", | |
| "decision_type": "refine_recommendation", | |
| "decision_source": "deterministic_rule", | |
| "selected": preferred.candidate_id, | |
| "options": [item.candidate_id for item in plan.candidates], | |
| "rationale": "The refinement prioritizes interpretability or runtime.", | |
| }, | |
| ] | |
| return ResearchPlan.from_mapping(data) | |
| def generate_research_plan( | |
| state: Mapping[str, Any] | WorkbenchState, | |
| provider: Any | None = None, | |
| ) -> WorkbenchState: | |
| current = state_from_mapping(state) | |
| selected_provider = provider or MockResearchPlanProvider() | |
| context = { | |
| "dataset_name": current.dataset_name or current.upload_type, | |
| "modality": current.data_modality, | |
| "task_type": current.task_type, | |
| "target_column": current.target, | |
| "constraints": list(current.constraints), | |
| } | |
| try: | |
| plan = selected_provider.create_plan(context) | |
| if not isinstance(plan, ResearchPlan): | |
| plan = ResearchPlan.from_mapping(plan) | |
| plan = _plan_with_constraint_recommendation(plan, current.constraints) | |
| provider_error = "" | |
| except Exception: | |
| plan = _fallback_plan(context) | |
| provider_error = ( | |
| "Research planning was unavailable. " | |
| "A conservative fallback plan is shown." | |
| ) | |
| recommended = _clean_text(plan.recommendation.get("candidate_id")) | |
| executable_ids = [ | |
| item.candidate_id | |
| for item in plan.candidates | |
| if candidate_is_executable(plan, item.candidate_id) | |
| ] | |
| if recommended not in executable_ids: | |
| recommended = executable_ids[0] if executable_ids else "" | |
| return replace( | |
| current, | |
| research_plan=plan, | |
| selected_candidate_id=recommended, | |
| approved_candidate_id="", | |
| approved_pipeline_config={}, | |
| approved_evidence_ids=(), | |
| approval_trace=None, | |
| approval_error="", | |
| provider_error=provider_error, | |
| ) | |
| def apply_research_plan( | |
| state: Mapping[str, Any] | WorkbenchState, | |
| plan_value: Mapping[str, Any] | ResearchPlan, | |
| *, | |
| provider_error: str = "", | |
| ) -> WorkbenchState: | |
| """Apply one validated real plan while invalidating every prior approval.""" | |
| current = state_from_mapping(state) | |
| plan = ( | |
| plan_value | |
| if isinstance(plan_value, ResearchPlan) | |
| else ResearchPlan.from_mapping(plan_value) | |
| ) | |
| if not 1 <= len(plan.candidates) <= 3: | |
| raise ValueError("research plan must contain one to three candidates") | |
| recommended = _clean_text(plan.recommendation.get("candidate_id")) | |
| candidate_ids = {item.candidate_id for item in plan.candidates} | |
| executable_ids = [ | |
| item.candidate_id | |
| for item in plan.candidates | |
| if candidate_is_executable(plan, item.candidate_id) | |
| ] | |
| if recommended not in candidate_ids or recommended not in executable_ids: | |
| recommended = executable_ids[0] if executable_ids else "" | |
| return replace( | |
| current, | |
| research_plan=plan, | |
| approved_candidate_id="", | |
| approved_pipeline_config={}, | |
| approved_evidence_ids=(), | |
| approval_trace=None, | |
| approval_error="", | |
| variants=(), | |
| provider_error=_clean_text(provider_error, limit=500), | |
| selected_candidate_id="", | |
| ) | |
| def apply_research_error( | |
| state: Mapping[str, Any] | WorkbenchState, message: str | |
| ) -> WorkbenchState: | |
| """Fail closed when the real research provider is unavailable or invalid.""" | |
| current = state_from_mapping(state) | |
| return replace( | |
| current, | |
| research_plan=None, | |
| selected_candidate_id="", | |
| approved_candidate_id="", | |
| approved_pipeline_config={}, | |
| approved_evidence_ids=(), | |
| approval_trace=None, | |
| approval_error="", | |
| variants=(), | |
| provider_error=_clean_text(message, limit=500), | |
| ) | |
| def apply_candidate_approval( | |
| state: Mapping[str, Any] | WorkbenchState, | |
| approval_value: Mapping[str, Any], | |
| ) -> WorkbenchState: | |
| """Apply a private-backend registry approval to the selected candidate.""" | |
| current = state_from_mapping(state) | |
| if current.research_plan is None or not current.selected_candidate_id: | |
| raise ValueError("select a research candidate before approval") | |
| if not isinstance(approval_value, Mapping): | |
| raise ValueError("candidate approval must be a mapping") | |
| approved_id = _clean_text(approval_value.get("approved_candidate_id")) | |
| plan_id = _clean_text(approval_value.get("plan_id")) | |
| if approved_id != current.selected_candidate_id: | |
| raise ValueError("approval does not match the selected candidate") | |
| if plan_id != current.research_plan.plan_id: | |
| raise ValueError("approval does not match the active research plan") | |
| if not candidate_is_executable(current.research_plan, approved_id): | |
| raise ValueError("candidate verification does not permit execution") | |
| config = approval_value.get("pipeline_config") | |
| evidence_ids = approval_value.get("evidence_ids") | |
| trace_value = approval_value.get("decision_trace") | |
| if not isinstance(config, Mapping): | |
| raise ValueError("approved pipeline config must be a mapping") | |
| if not isinstance(evidence_ids, (list, tuple)) or not all( | |
| isinstance(item, str) for item in evidence_ids | |
| ): | |
| raise ValueError("approved evidence IDs must be an array of strings") | |
| if not isinstance(trace_value, Mapping): | |
| raise ValueError("approval decision trace is required") | |
| trace = DecisionTrace.from_mapping(trace_value) | |
| if trace.decision_source is not DecisionSource.RESEARCH_AGENT: | |
| raise ValueError("approval trace must come from the research agent") | |
| if trace.selected != approved_id: | |
| raise ValueError("approval trace selected value does not match") | |
| config_copy = _json_copy(dict(config)) | |
| if config_copy.get("candidate_id") not in (None, approved_id): | |
| raise ValueError("approved pipeline config references another candidate") | |
| if config_copy.get("data_modality") not in (None, current.data_modality): | |
| raise ValueError("approved pipeline config has an incompatible modality") | |
| candidate = current.current_candidate | |
| known_evidence = { | |
| str(item.get("evidence_id")) | |
| for item in current.research_plan.evidence | |
| if isinstance(item, Mapping) and item.get("evidence_id") | |
| } | |
| if not set(evidence_ids) <= known_evidence: | |
| raise ValueError("approval references unknown research evidence") | |
| changes: dict[str, Any] = { | |
| "approved_candidate_id": approved_id, | |
| "approved_pipeline_config": config_copy, | |
| "approved_evidence_ids": tuple(evidence_ids), | |
| "approval_trace": trace, | |
| "approval_error": "", | |
| } | |
| control_fields = { | |
| "feature_selection_tool": "feature_tool", | |
| "transform_tool": "transform_tool", | |
| "ml_tool": "ml_tool", | |
| "image_backend": "image_backend", | |
| } | |
| for config_name, state_name in control_fields.items(): | |
| value = config_copy.get(config_name) | |
| if isinstance(value, str) and value.strip(): | |
| changes[state_name] = _clean_text(value) | |
| if not set(candidate.evidence_ids) >= set(evidence_ids): | |
| raise ValueError("approval contains evidence not used by the candidate") | |
| return replace(current, **changes) | |
| def apply_approval_error( | |
| state: Mapping[str, Any] | WorkbenchState, message: str | |
| ) -> WorkbenchState: | |
| current = state_from_mapping(state) | |
| return replace( | |
| current, | |
| approved_candidate_id="", | |
| approved_pipeline_config={}, | |
| approved_evidence_ids=(), | |
| approval_trace=None, | |
| approval_error=_clean_text(message, limit=500), | |
| ) | |
| def approved_research_metadata( | |
| state: Mapping[str, Any] | WorkbenchState, | |
| ) -> dict[str, Any]: | |
| """Return the optional canonical metadata carried into an approved run.""" | |
| current = state_from_mapping(state) | |
| if ( | |
| current.research_plan is None | |
| or current.approved_candidate_id != current.selected_candidate_id | |
| or not current.approved_pipeline_config | |
| or current.approval_trace is None | |
| ): | |
| return {} | |
| return { | |
| "schema_version": 1, | |
| "research_plan": current.research_plan.to_dict(), | |
| "approved_candidate_id": current.approved_candidate_id, | |
| "pipeline_config": _json_copy(dict(current.approved_pipeline_config)), | |
| "evidence_ids": list(current.approved_evidence_ids), | |
| "approval_trace": current.approval_trace.to_dict(), | |
| } | |
| def select_candidate( | |
| state: Mapping[str, Any] | WorkbenchState, candidate_id: str | |
| ) -> WorkbenchState: | |
| current = state_from_mapping(state) | |
| selected = _clean_text(candidate_id) | |
| if current.research_plan is None or selected not in { | |
| candidate.candidate_id for candidate in current.research_plan.candidates | |
| }: | |
| raise ValueError(f"unknown candidate_id {selected!r}") | |
| if not candidate_is_executable(current.research_plan, selected): | |
| raise ValueError("candidate verification does not permit execution") | |
| preset = current.preset | |
| if "interpretable" in selected.lower(): | |
| preset = "interpretable" | |
| elif "balanced" in selected.lower(): | |
| preset = "medium_quality" | |
| return replace( | |
| current, | |
| selected_candidate_id=selected, | |
| preset=preset, | |
| approved_candidate_id="", | |
| approved_pipeline_config={}, | |
| approved_evidence_ids=(), | |
| approval_trace=None, | |
| approval_error="", | |
| ) | |
| def apply_next_experiment_draft( | |
| state: Mapping[str, Any] | WorkbenchState, | |
| reflection: Mapping[str, Any] | ExperimentReflection, | |
| suggestion_id: str, | |
| ) -> WorkbenchState: | |
| """Create a fresh unapproved objective draft without execution authority.""" | |
| current = state_from_mapping(state) | |
| canonical = ( | |
| reflection | |
| if isinstance(reflection, ExperimentReflection) | |
| else ExperimentReflection.from_mapping(reflection) | |
| ) | |
| selected_id = _clean_text(suggestion_id) | |
| suggestion = next( | |
| (item for item in canonical.suggestions if item.suggestion_id == selected_id), | |
| None, | |
| ) | |
| if suggestion is None: | |
| raise ValueError(f"unknown next experiment suggestion {selected_id!r}") | |
| constraints: tuple[str, ...] = () | |
| if suggestion.executable and not suggestion.suggestion_only: | |
| components = tuple( | |
| _clean_text(stage.get("component"), limit=200) | |
| for stage in suggestion.stages | |
| if isinstance(stage, Mapping) and stage.get("component") | |
| ) | |
| constraints = tuple( | |
| dict.fromkeys( | |
| ( | |
| *(f"Trusted component: {component}" for component in components), | |
| *suggestion.expected_tradeoffs, | |
| ) | |
| ) | |
| ) | |
| drafted = update_context( | |
| current, | |
| user_objective=suggestion.title, | |
| constraints=constraints, | |
| ) | |
| # Clear explicitly even if a package repeats the current objective text. | |
| drafted = replace( | |
| drafted, | |
| research_plan=None, | |
| selected_candidate_id="", | |
| approved_candidate_id="", | |
| approved_pipeline_config={}, | |
| approved_evidence_ids=(), | |
| approval_trace=None, | |
| approval_error="", | |
| variants=(), | |
| provider_error="", | |
| ) | |
| return set_step(drafted, "objective") | |
| def apply_refinement( | |
| state: Mapping[str, Any] | WorkbenchState, refinement: str | |
| ) -> WorkbenchState: | |
| current = state_from_mapping(state) | |
| text = _clean_text(refinement, limit=500) | |
| if not text: | |
| return current | |
| refined = replace(current, constraints=(*current.constraints, text)) | |
| return generate_research_plan(refined) | |
| def select_node( | |
| state: Mapping[str, Any] | WorkbenchState, node_id: str | |
| ) -> WorkbenchState: | |
| current = state_from_mapping(state) | |
| selected = _clean_text(node_id).lower() | |
| if selected not in PIPELINE_NODE_IDS: | |
| raise ValueError(f"unknown pipeline node {selected!r}") | |
| return replace(current, selected_node_id=selected) | |
| def validate_state( | |
| state: Mapping[str, Any] | WorkbenchState, | |
| ) -> tuple[str, ...]: | |
| current = state_from_mapping(state) | |
| issues: list[str] = [] | |
| if current.data_source == "managed" and not current.dataset_name: | |
| issues.append("Data: choose a managed dataset.") | |
| if current.data_source == "upload" and not current.upload_type: | |
| issues.append("Data: choose an upload type.") | |
| if not current.target: | |
| issues.append("Objective: choose an outcome to predict.") | |
| if current.research_plan is None: | |
| issues.append("Research: generate a research plan.") | |
| elif not current.selected_candidate_id: | |
| issues.append("Research: select a candidate pipeline.") | |
| else: | |
| candidate_ids = { | |
| candidate.candidate_id for candidate in current.research_plan.candidates | |
| } | |
| if current.selected_candidate_id not in candidate_ids: | |
| issues.append("Research: the selected candidate is no longer available.") | |
| elif not candidate_is_executable( | |
| current.research_plan, current.selected_candidate_id | |
| ): | |
| issues.append("Research: verifier rejected the selected candidate.") | |
| elif ( | |
| current.approved_candidate_id != current.selected_candidate_id | |
| or not current.approved_pipeline_config | |
| or current.approval_trace is None | |
| ): | |
| issues.append("Research: approve the selected candidate before Run.") | |
| if not 60 <= current.time_limit <= 3_600: | |
| issues.append("Compute: time budget must be between 60 and 3600 seconds.") | |
| if not 1 <= current.trials <= 4: | |
| issues.append("Compute: candidate runs must be between 1 and 4.") | |
| return tuple(issues) | |
| def _planned_feature_report(state: WorkbenchState) -> FeatureExtractionReport: | |
| method = state.feature_tool | |
| if method.lower() in _AUTOMATIC_VALUES: | |
| method = "Automatic feature extraction" | |
| return FeatureExtractionReport.from_mapping( | |
| { | |
| "schema_version": 1, | |
| "report_id": "planned-feature-report", | |
| "status": "planned", | |
| "modality": state.data_modality, | |
| "method": method, | |
| "input_shape": [], | |
| "output_shape": [], | |
| "features_per_sample": None, | |
| "parameters": {"selection": state.feature_tool}, | |
| "warnings": [ | |
| "Execution shapes are unavailable until the backend reports them." | |
| ], | |
| "provenance": { | |
| "phase": "planned", | |
| "source": ( | |
| "user" | |
| if state.feature_tool.lower() not in _AUTOMATIC_VALUES | |
| else "deterministic_rule" | |
| ), | |
| }, | |
| } | |
| ) | |
| def _candidate_stage_component( | |
| candidate: CandidatePipeline | None, stage_names: set[str], fallback: str | |
| ) -> str: | |
| if candidate is None: | |
| return fallback | |
| for stage in candidate.stages: | |
| stage_name = _clean_text(stage.get("stage")).lower() | |
| if stage_name in stage_names: | |
| component = _clean_text(stage.get("component")) | |
| if component: | |
| return component.replace("_", " ").title() | |
| return fallback | |
| def _is_user_selection(value: str) -> bool: | |
| return _clean_text(value).lower() not in _AUTOMATIC_VALUES | |
| def _node_trace( | |
| node_id: str, | |
| component: str, | |
| source: DecisionSource, | |
| rationale: str, | |
| alternatives: Sequence[str], | |
| parameters: Sequence[tuple[str, str]], | |
| ) -> DecisionTrace: | |
| return DecisionTrace.from_mapping( | |
| { | |
| "schema_version": 1, | |
| "trace_id": f"planned-{node_id}", | |
| "decision_type": f"select_{node_id}", | |
| "stage": node_id, | |
| "decision_source": source.value, | |
| "selected": component, | |
| "options": list(alternatives), | |
| "rationale": rationale, | |
| "parameters": dict(parameters), | |
| "provenance": {"phase": "planned"}, | |
| } | |
| ) | |
| def pipeline_nodes( | |
| state: Mapping[str, Any] | WorkbenchState, | |
| ) -> tuple[PipelineNodeView, ...]: | |
| current = state_from_mapping(state) | |
| try: | |
| candidate = current.current_candidate | |
| except ValueError: | |
| candidate = None | |
| planned_report = _planned_feature_report(current) | |
| has_plan = candidate is not None | |
| input_component = ( | |
| current.dataset_name | |
| if current.data_source == "managed" | |
| else current.upload_type | |
| ) or "Data selection pending" | |
| preprocessing = _candidate_stage_component( | |
| candidate, | |
| {"preprocess", "preprocessing"}, | |
| "Automatic preprocessing", | |
| ) | |
| feature_component = planned_report.method | |
| postprocessing = ( | |
| current.transform_tool | |
| if _is_user_selection(current.transform_tool) | |
| else "Automatic post-processing" | |
| ) | |
| model = _candidate_stage_component( | |
| candidate, | |
| {"model", "training"}, | |
| current.ml_tool | |
| if _is_user_selection(current.ml_tool) | |
| else "Model selection pending", | |
| ) | |
| evaluation = f"{current.review_metric.title()} review" | |
| definitions = ( | |
| ( | |
| "input", | |
| "Input", | |
| input_component, | |
| DecisionSource.USER, | |
| (("source", current.data_source),), | |
| "The user selected the data source for this run.", | |
| ("Managed dataset", "Upload"), | |
| "compatible" if input_component != "Data selection pending" else "blocked", | |
| ), | |
| ( | |
| "preprocessing", | |
| "Preprocessing", | |
| preprocessing, | |
| DecisionSource.DETERMINISTIC_RULE, | |
| (("task", current.task_type),), | |
| "A deterministic compatibility rule supplies preprocessing defaults.", | |
| ("Automatic preprocessing", "Backend-compatible preprocessing"), | |
| "compatible" if has_plan else "review", | |
| ), | |
| ( | |
| "feature_extractor", | |
| "Feature Extractor", | |
| feature_component, | |
| ( | |
| DecisionSource.USER | |
| if _is_user_selection(current.feature_tool) | |
| else DecisionSource.DETERMINISTIC_RULE | |
| ), | |
| (("selection", current.feature_tool),), | |
| ( | |
| "The user selected this feature method." | |
| if _is_user_selection(current.feature_tool) | |
| else "A deterministic rule defers feature selection to the backend." | |
| ), | |
| ( | |
| "Automatic feature extraction", | |
| "No feature selection", | |
| "T-test feature selection", | |
| ), | |
| "compatible" if has_plan else "review", | |
| ), | |
| ( | |
| "postprocessing", | |
| "Post-processing", | |
| postprocessing, | |
| ( | |
| DecisionSource.USER | |
| if _is_user_selection(current.transform_tool) | |
| else DecisionSource.PRESET | |
| ), | |
| (("selection", current.transform_tool),), | |
| ( | |
| "The user selected this post-processing method." | |
| if _is_user_selection(current.transform_tool) | |
| else "The active preset supplies post-processing defaults." | |
| ), | |
| ( | |
| "Automatic post-processing", | |
| "No transformation", | |
| "PCA transformation", | |
| ), | |
| "compatible" if has_plan else "review", | |
| ), | |
| ( | |
| "model", | |
| "Model", | |
| model, | |
| ( | |
| DecisionSource.USER | |
| if _is_user_selection(current.ml_tool) | |
| else DecisionSource.PRESET | |
| ), | |
| (("quality_profile", current.preset),), | |
| ( | |
| "The user selected the training engine." | |
| if _is_user_selection(current.ml_tool) | |
| else "The selected candidate and quality preset define the model route." | |
| ), | |
| tuple( | |
| item.name for item in current.research_plan.candidates | |
| ) | |
| if current.research_plan | |
| else ("Generate a research plan",), | |
| "compatible" if has_plan else "blocked", | |
| ), | |
| ( | |
| "evaluation", | |
| "Evaluation", | |
| evaluation, | |
| DecisionSource.USER, | |
| (("review_priority", current.review_metric),), | |
| "The user selected which reported metric to emphasize during review.", | |
| ("Accuracy", "F1", "ROC AUC", "Backend-reported primary metric"), | |
| "compatible" if current.target else "blocked", | |
| ), | |
| ) | |
| result: list[PipelineNodeView] = [] | |
| for ( | |
| node_id, | |
| label, | |
| component, | |
| source, | |
| parameters, | |
| rationale, | |
| alternatives, | |
| compatibility, | |
| ) in definitions: | |
| trace = _node_trace( | |
| node_id, | |
| component, | |
| source, | |
| rationale, | |
| alternatives, | |
| parameters, | |
| ) | |
| result.append( | |
| PipelineNodeView( | |
| node_id=node_id, | |
| label=label, | |
| component=component, | |
| source=source, | |
| parameters=tuple(parameters), | |
| input_shape=(), | |
| output_shape=(), | |
| compatibility=compatibility, | |
| rationale=rationale, | |
| alternatives=tuple(alternatives), | |
| trace=trace, | |
| feature_report=( | |
| planned_report if node_id == "feature_extractor" else None | |
| ), | |
| ) | |
| ) | |
| return tuple(result) | |
| def _variant_signature(variant: RunVariant) -> tuple[Any, ...]: | |
| return ( | |
| variant.data_source, | |
| variant.candidate_id, | |
| variant.dataset, | |
| variant.target, | |
| variant.trials, | |
| variant.preset, | |
| variant.time_limit, | |
| variant.image_backend, | |
| ) | |
| def add_variant( | |
| state: Mapping[str, Any] | WorkbenchState, | |
| ) -> WorkbenchState: | |
| current = state_from_mapping(state) | |
| if current.data_source != "managed": | |
| raise ValueError( | |
| "Parallel run plans currently support managed datasets only." | |
| ) | |
| issues = validate_state(current) | |
| if issues: | |
| raise ValueError(" ".join(issues)) | |
| if len(current.variants) >= 2: | |
| raise ValueError("A run plan supports at most two variants.") | |
| feature_report = _planned_feature_report(current) | |
| variant = RunVariant( | |
| variant_id=uuid.uuid4().hex, | |
| data_source=current.data_source, | |
| data_modality=current.data_modality, | |
| candidate_id=current.selected_candidate_id, | |
| dataset=current.dataset_name, | |
| target=current.target, | |
| trials=current.trials, | |
| preset=current.preset, | |
| time_limit=current.time_limit, | |
| image_backend=current.image_backend, | |
| research_plan=current.research_plan.to_dict(), | |
| feature_report=feature_report.to_dict(), | |
| approved_research=approved_research_metadata(current), | |
| ) | |
| if _variant_signature(variant) in { | |
| _variant_signature(item) for item in current.variants | |
| }: | |
| raise ValueError("This configuration is already in the run plan.") | |
| return replace(current, variants=(*current.variants, variant)) | |
| def source_badge_label(source: DecisionSource | str) -> str: | |
| try: | |
| normalized = DecisionSource(source) | |
| except (TypeError, ValueError) as exc: | |
| raise ValueError(f"unknown decision source {source!r}") from exc | |
| return SOURCE_LABELS[normalized] | |
| def _escape(value: Any) -> str: | |
| return html.escape(_clean_text(value), quote=True) | |
| def _list_html(values: Sequence[Any], *, empty: str = "None recorded") -> str: | |
| cleaned = [_clean_text(value) for value in values if _clean_text(value)] | |
| if not cleaned: | |
| return f'<p class="guided-empty-copy">{_escape(empty)}</p>' | |
| return "<ul>" + "".join(f"<li>{_escape(value)}</li>" for value in cleaned) + "</ul>" | |
| def _mapping_rows(value: Mapping[str, Any]) -> str: | |
| if not value: | |
| return '<p class="guided-empty-copy">Unavailable</p>' | |
| rows = [] | |
| for key, item in value.items(): | |
| normalized_key = str(key).lower() | |
| if item is None: | |
| rendered = "Not reported" | |
| elif isinstance(item, (list, tuple)) and not item: | |
| rendered = ( | |
| ("Not reported" if value.get("status") == "completed" else "Available after training") | |
| if "shape" in normalized_key | |
| else "None recorded" | |
| ) | |
| elif isinstance(item, dict) and not item: | |
| rendered = "Not reported" | |
| elif item == "": | |
| rendered = "Not reported" | |
| elif isinstance(item, Mapping): | |
| rendered = "; ".join( | |
| f"{str(nested_key).replace('_', ' ').title()}: " | |
| f"{_clean_text(str(nested_value)) if nested_value is not None else 'Not reported'}" | |
| for nested_key, nested_value in item.items() | |
| ) | |
| elif isinstance(item, (list, tuple)): | |
| separator = " × " if "shape" in normalized_key else ", " | |
| rendered = separator.join(_clean_text(entry) for entry in item) | |
| else: | |
| rendered = str(item) | |
| rows.append( | |
| '<div class="guided-detail-row">' | |
| f'<span>{_escape(str(key).replace("_", " ").title())}</span>' | |
| f'<strong>{_escape(rendered)}</strong>' | |
| "</div>" | |
| ) | |
| return "".join(rows) | |
| def _runtime_contracts( | |
| value: Mapping[str, Any] | None, | |
| ) -> tuple[FeatureExtractionReport | None, tuple[DecisionTrace, ...]]: | |
| """Validate optional terminal execution metadata for presentation.""" | |
| if not isinstance(value, Mapping): | |
| return None, () | |
| try: | |
| normalized = normalize_execution_metadata(value) | |
| except (TypeError, ValueError): | |
| return None, () | |
| report = None | |
| raw_report = normalized.get("feature_extraction_report") | |
| if isinstance(raw_report, Mapping): | |
| try: | |
| report = FeatureExtractionReport.from_mapping(raw_report) | |
| except (TypeError, ValueError): | |
| report = None | |
| traces: list[DecisionTrace] = [] | |
| raw_traces = normalized.get("decision_traces") or () | |
| for raw_trace in raw_traces: | |
| if not isinstance(raw_trace, Mapping): | |
| continue | |
| try: | |
| traces.append(DecisionTrace.from_mapping(raw_trace)) | |
| except (TypeError, ValueError): | |
| continue | |
| return report, tuple(traces) | |
| def _expected_metrics_html(candidate: CandidatePipeline) -> str: | |
| if not candidate.expected_metrics: | |
| return '<p class="guided-empty-copy">No estimates available</p>' | |
| cards = [] | |
| for name, value in candidate.expected_metrics.items(): | |
| estimate = value.get("estimate", "Unavailable") if isinstance(value, Mapping) else value | |
| basis = value.get("basis", "") if isinstance(value, Mapping) else "" | |
| cards.append( | |
| '<div class="guided-estimate">' | |
| '<small class="guided-metric-kind">Estimate · unmeasured</small>' | |
| f'<span>{_escape(str(name).replace("_", " ").title())}</span>' | |
| f'<strong>{_escape(str(estimate) if estimate is not None else "Unavailable")}</strong>' | |
| f'<small>{_escape(basis or "Planning estimate; not measured.")}</small>' | |
| "</div>" | |
| ) | |
| return "".join(cards) | |
| def candidate_display_name(candidate: CandidatePipeline) -> str: | |
| """Readable labels for known deterministic defaults; identity stays canonical.""" | |
| names = { | |
| "no_feature_selection_tool + no_transform_tool + autogluon_tool": "AutoGluon baseline", | |
| "t_test_feature_selection_tool + no_transform_tool + autogluon_tool": "AutoGluon with feature selection", | |
| "no_feature_selection_tool + pca_transform_tool + autogluon_tool": "AutoGluon with PCA", | |
| } | |
| return names.get(candidate.name, candidate.name or candidate.candidate_id) | |
| def candidate_cards_html( | |
| state: Mapping[str, Any] | WorkbenchState, | |
| ) -> str: | |
| current = state_from_mapping(state) | |
| if current.research_plan is None: | |
| return ( | |
| '<section class="guided-empty-state">' | |
| "<h3>Research plan not generated</h3>" | |
| "<p>Complete Data and Objective, then generate a plan.</p>" | |
| "</section>" | |
| ) | |
| constraints = ( | |
| '<div class="guided-plan-context"><h4>Research constraints</h4>' | |
| f'{_list_html(current.constraints, empty="No additional constraints")}</div>' | |
| ) | |
| cards = [] | |
| verifications = candidate_verification_map(current.research_plan) | |
| recommended_id = _clean_text( | |
| current.research_plan.recommendation.get("candidate_id") | |
| ) | |
| for index, candidate in enumerate(current.research_plan.candidates, start=1): | |
| selected = candidate.candidate_id == current.selected_candidate_id | |
| recommended = candidate.candidate_id == recommended_id | |
| verification = verifications.get(candidate.candidate_id) | |
| executable = candidate_is_executable( | |
| current.research_plan, candidate.candidate_id | |
| ) | |
| recommended_badge = ( | |
| '<span class="guided-recommended-badge">Recommended</span>' | |
| if recommended | |
| else "" | |
| ) | |
| verification_badge = "" | |
| verification_details = "" | |
| if verification is not None: | |
| verification_badge = ( | |
| '<span class="guided-verification-badge ' | |
| f'{_escape(verification.status)}">Verifier · ' | |
| f'{_escape(verification.status.title())}</span>' | |
| ) | |
| verification_details = ( | |
| '<div class="guided-candidate-section guided-verification-detail">' | |
| '<h4>Verifier assessment</h4>' | |
| f'<p>{_escape(verification.revised_rationale or verification.rationale)}</p>' | |
| '<div class="guided-candidate-columns">' | |
| '<div><h4>Weaknesses</h4>' | |
| f'{_list_html(verification.weaknesses)}</div>' | |
| '<div><h4>Verifier warnings</h4>' | |
| f'{_list_html(verification.warnings)}</div></div></div>' | |
| ) | |
| execution_status = ( | |
| "Available after explicit approval" | |
| if executable | |
| else "Not executable · verifier rejected or marked incompatible" | |
| ) | |
| cards.append( | |
| f'<article class="guided-candidate-card' | |
| f'{" selected" if selected else ""}' | |
| f'{" recommended" if recommended else ""}' | |
| f'{" non-executable" if not executable else ""}' | |
| f'" data-candidate-id="{_escape(candidate.candidate_id)}">' | |
| f'<div class="guided-candidate-index">Candidate {index}</div>' | |
| f"{recommended_badge}{verification_badge}" | |
| f'<h3>{_escape(candidate_display_name(candidate))}</h3>' | |
| '<p class="guided-candidate-rationale">' | |
| f'{_escape(candidate.rationale or "No rationale provided")}</p>' | |
| '<div class="guided-candidate-section"><h4>Estimated metric values · unmeasured</h4>' | |
| f'<div class="guided-estimate-grid">{_expected_metrics_html(candidate)}</div></div>' | |
| '<details class="guided-candidate-details"><summary>Evidence and assessment</summary>' | |
| '<h4>Registered pipeline</h4>' | |
| f'<p>{_escape(candidate.name or candidate.candidate_id)}</p>' | |
| '<div class="guided-candidate-columns">' | |
| '<div><h4>Assumptions</h4>' | |
| f'{_list_html(candidate.assumptions)}</div>' | |
| '<div><h4>Expected tradeoffs</h4>' | |
| f'{_list_html(candidate.expected_tradeoffs)}</div>' | |
| "</div>" | |
| '<div class="guided-candidate-section"><h4>Evidence references</h4>' | |
| f'{_list_html(candidate.evidence_ids)}</div>' | |
| '<div class="guided-candidate-section"><h4>Warnings</h4>' | |
| f'{_list_html(candidate.warnings)}</div>' | |
| f"{verification_details}</details>" | |
| '<p class="guided-selection-state">' | |
| f'{_escape("Selected · " + execution_status if selected else execution_status)}</p>' | |
| "</article>" | |
| ) | |
| return ( | |
| '<section class="guided-candidate-list" aria-label="Research candidates">' | |
| f"{constraints}{''.join(cards)}</section>" | |
| ) | |
| def approval_status_html( | |
| state: Mapping[str, Any] | WorkbenchState, | |
| ) -> str: | |
| current = state_from_mapping(state) | |
| if current.approval_error: | |
| return ( | |
| '<div class="guided-inline-error">' | |
| f'{_escape(current.approval_error)}</div>' | |
| ) | |
| if ( | |
| current.approved_candidate_id | |
| and current.approved_candidate_id == current.selected_candidate_id | |
| and current.approval_trace is not None | |
| ): | |
| return ( | |
| '<div class="guided-approval-status approved">' | |
| '<strong>Approved</strong><span>Registry validated · ' | |
| f'{_escape(source_badge_label(current.approval_trace.decision_source))}' | |
| "</span></div>" | |
| ) | |
| if current.selected_candidate_id: | |
| return ( | |
| '<div class="guided-approval-status pending">' | |
| '<strong>Approval required</strong>' | |
| '<span>Review and explicitly approve this candidate before Run.</span>' | |
| "</div>" | |
| ) | |
| return ( | |
| '<div class="guided-approval-status pending">' | |
| '<strong>Select a candidate</strong>' | |
| '<span>Run remains disabled until registry validation and approval.</span>' | |
| "</div>" | |
| ) | |
| def _evidence_card_html(item: Mapping[str, Any]) -> str: | |
| evidence_type = _clean_text(item.get("evidence_type") or "catalog") | |
| source = _clean_text(item.get("source") or "unknown") | |
| title = _clean_text( | |
| item.get("title") | |
| or item.get("name") | |
| or item.get("external_id") | |
| or item.get("evidence_id") | |
| ) | |
| detail_fields = ( | |
| "authors", | |
| "date", | |
| "external_id", | |
| "source_url", | |
| "source_reference", | |
| "query", | |
| "pipeline_stages", | |
| "modality", | |
| "task", | |
| "size", | |
| "target", | |
| "classes", | |
| "license", | |
| "authentication_required", | |
| "download_required", | |
| "architecture", | |
| "framework", | |
| "parameter_count", | |
| "memory_bytes", | |
| "compatibility", | |
| "executable", | |
| "suggestion_only", | |
| "status", | |
| ) | |
| details = { | |
| field: item[field] | |
| for field in detail_fields | |
| if field in item and item[field] not in (None, "", [], {}) | |
| } | |
| return ( | |
| '<article class="guided-evidence-card" ' | |
| f'data-evidence-type="{_escape(evidence_type)}">' | |
| '<div class="guided-evidence-heading">' | |
| f'<span>{_escape(evidence_type.replace("_", " ").title())}</span>' | |
| f'<small>{_escape(source.replace("_", " ").title())}</small></div>' | |
| f'<h4>{_escape(title)}</h4>' | |
| f'<p>{_escape(item.get("summary") or "No summary provided")}</p>' | |
| '<div class="guided-evidence-provenance"><strong>Evidence ID</strong>' | |
| f'<code>{_escape(item.get("evidence_id"))}</code></div>' | |
| f'{_mapping_rows(details)}</article>' | |
| ) | |
| def research_dossier_html( | |
| state: Mapping[str, Any] | WorkbenchState, | |
| ) -> str: | |
| current = state_from_mapping(state) | |
| if current.research_plan is None: | |
| provider_error = ( | |
| f'<div class="guided-inline-error">{_escape(current.provider_error)}</div>' | |
| if current.provider_error | |
| else "" | |
| ) | |
| return ( | |
| '<section class="guided-dossier-empty">' | |
| '<div class="guided-empty-agent-heading">' | |
| '<span class="guided-research-icon" aria-hidden="true">RA</span>' | |
| '<div><h3>Research assistant</h3>' | |
| '<small>Private research backend</small></div></div>' | |
| '<p class="guided-agent-intro">Compare viable pipelines for your ' | |
| 'dataset, objective, and compute budget.</p>' | |
| '<article class="guided-empty-insight">' | |
| '<span>Research plan</span><h4>Ready when your objective is set</h4>' | |
| '<p>Generate a plan to review recommendations, tradeoffs, assumptions, ' | |
| 'and supporting evidence.</p></article>' | |
| f"{provider_error}" | |
| '<p class="guided-provider-note">Research is generated only when ' | |
| "the private backend feature is enabled.</p></section>" | |
| ) | |
| plan = current.research_plan | |
| recommendation = plan.recommendation | |
| decision_provenance = tuple( | |
| " · ".join( | |
| part | |
| for part in ( | |
| source_badge_label(trace.decision_source), | |
| _clean_text(trace.stage or trace.decision_type or "Plan decision"), | |
| _clean_text(trace.rationale), | |
| ( | |
| "Alternatives: " | |
| + ", ".join(_clean_text(item) for item in trace.options) | |
| if trace.options | |
| else "" | |
| ), | |
| ( | |
| "Evidence: " + ", ".join(trace.evidence_ids) | |
| if trace.evidence_ids | |
| else "" | |
| ), | |
| ( | |
| "Warnings: " + ", ".join(trace.warnings) | |
| if trace.warnings | |
| else "" | |
| ), | |
| ) | |
| if part | |
| ) | |
| for trace in plan.decision_provenance | |
| ) | |
| provider = _clean_text( | |
| plan.provider_provenance.get("provider") or "research_agent" | |
| ) | |
| planner_source = ( | |
| DecisionSource.LLM | |
| if plan.provider_provenance.get("llm_used") | |
| else DecisionSource.FALLBACK | |
| if plan.provider_provenance.get("fallback_used") | |
| else DecisionSource.RESEARCH_AGENT | |
| ) | |
| planner_model = _clean_text( | |
| plan.provider_provenance.get("llm_model") or "Not configured" | |
| ) | |
| verification = plan.verification | |
| verifier_source = ( | |
| verification.source if verification is not None else DecisionSource.FALLBACK | |
| ) | |
| verifier_model = _clean_text( | |
| ( | |
| verification.provider_provenance.get("model") | |
| if verification is not None | |
| else None | |
| ) | |
| or plan.provider_provenance.get("verification_model") | |
| or "Not configured" | |
| ) | |
| planner_status_html = ( | |
| '<div class="guided-agent-status-grid">' | |
| '<article><span class="guided-source-badge" data-source="' | |
| f'{_escape(planner_source.value)}">{_escape(source_badge_label(planner_source))}</span>' | |
| f'<h4>{_escape(source_badge_label(planner_source))} planner</h4>' | |
| f'<p>Model: {_escape(planner_model)}</p>' | |
| '<small>Planning only · no execution authority</small></article>' | |
| '<article><span class="guided-source-badge" data-source="' | |
| f'{_escape(verifier_source.value)}">{_escape(source_badge_label(verifier_source))}</span>' | |
| f'<h4>{_escape(source_badge_label(verifier_source))} verifier</h4>' | |
| f'<p>Status: {_escape(verification.status.title() if verification else "Unavailable")}</p>' | |
| f'<p>Model: {_escape(verifier_model)}</p>' | |
| '<small>Deterministic validation remains final</small></article></div>' | |
| ) | |
| evidence_cards = "".join( | |
| _evidence_card_html(item) for item in plan.evidence | |
| ) | |
| provider_error = ( | |
| f'<div class="guided-inline-error">{_escape(current.provider_error)}</div>' | |
| if current.provider_error | |
| else "" | |
| ) | |
| try: | |
| selected_candidate = current.current_candidate | |
| except ValueError: | |
| selected_candidate = None | |
| tradeoffs = ( | |
| selected_candidate.expected_tradeoffs | |
| if selected_candidate and selected_candidate.expected_tradeoffs | |
| else plan.expected_tradeoffs | |
| ) | |
| assumptions = ( | |
| selected_candidate.assumptions | |
| if selected_candidate and selected_candidate.assumptions | |
| else plan.assumptions | |
| ) | |
| warnings = tuple(dict.fromkeys( | |
| (*plan.warnings, *(selected_candidate.warnings if selected_candidate else ())) | |
| )) | |
| insight_blocks = [ | |
| _research_insight_html( | |
| "recommendation", | |
| "Recommendation", | |
| recommendation.get("candidate_id") or "Unavailable", | |
| recommendation.get("rationale") or "No rationale provided", | |
| ) | |
| ] | |
| if tradeoffs: | |
| insight_blocks.append( | |
| _research_insight_html( | |
| "tradeoff", "Tradeoff", "Expected tradeoff", tradeoffs[0] | |
| ) | |
| ) | |
| if assumptions: | |
| insight_blocks.append( | |
| _research_insight_html( | |
| "assumption", "Assumption", "Confirm before running", assumptions[0] | |
| ) | |
| ) | |
| for warning in warnings: | |
| insight_blocks.append( | |
| _research_insight_html( | |
| "warning", "Warning", "Review before running", warning | |
| ) | |
| ) | |
| return ( | |
| '<section class="guided-dossier">' | |
| '<div class="guided-dossier-heading"><div><span>Research assistant</span>' | |
| '<h3>Recommendation and evidence</h3></div>' | |
| '<span class="guided-research-provider">' | |
| f'<b class="guided-provider-badge">{_escape(provider.title())}</b>' | |
| '<small>Planning only · no automatic execution</small></span></div>' | |
| f"{provider_error}{planner_status_html}" | |
| '<div class="guided-dossier-block"><h4>Dataset profile</h4>' | |
| f'<p>{_escape(plan.extensions.get("dataset_profile_source") or "Profile source not recorded")}</p>' | |
| f'{_mapping_rows(plan.extensions.get("dataset_profile") or {})}</div>' | |
| f'<div class="guided-research-insights">{"".join(insight_blocks)}</div>' | |
| '<div class="guided-dossier-block"><h4>Verifier summary</h4>' | |
| f'<p>{_escape(verification.summary if verification else "No verifier result is available.")}</p>' | |
| f'{_list_html(verification.warnings if verification else ())}</div>' | |
| '<div class="guided-dossier-block"><h4>Evidence</h4>' | |
| f'<div class="guided-evidence-grid">{evidence_cards}</div></div>' | |
| '<div class="guided-dossier-block"><h4>Decision provenance</h4>' | |
| f'{_list_html(decision_provenance)}</div>' | |
| '<p class="guided-provider-note">External search: ' | |
| + ( | |
| "enabled; partial source failures are shown above." | |
| if plan.provider_provenance.get("external_search_enabled") | |
| else "disabled; deterministic planning and internal evidence remain available." | |
| ) | |
| + "</p>" | |
| "</section>" | |
| ) | |
| def _research_insight_html( | |
| kind: str, | |
| label: str, | |
| heading: str, | |
| body: str, | |
| ) -> str: | |
| return ( | |
| '<article class="guided-research-insight" ' | |
| f'data-insight-type="{_escape(kind)}">' | |
| f'<span class="guided-insight-label">{_escape(label)}</span>' | |
| f'<h4>{_escape(heading)}</h4>' | |
| f'<p>{_escape(body)}</p>' | |
| "</article>" | |
| ) | |
| def _shape_summary(input_shape: Sequence[int], output_shape: Sequence[int]) -> str: | |
| if not input_shape and not output_shape: | |
| return "Unavailable before execution" | |
| before = " x ".join(str(item) for item in input_shape) or "Unavailable" | |
| after = " x ".join(str(item) for item in output_shape) or "Unavailable" | |
| return f"{before} to {after}" | |
| def _parameter_summary(parameters: Sequence[tuple[str, str]]) -> str: | |
| if not parameters: | |
| return "Backend defaults" | |
| return ", ".join( | |
| f"{key.replace('_', ' ').title()}: {value}" for key, value in parameters | |
| ) | |
| def pipeline_node_labels( | |
| state: Mapping[str, Any] | WorkbenchState, | |
| ) -> tuple[str, ...]: | |
| return tuple( | |
| "\n".join( | |
| ( | |
| node.label, | |
| node.component or "Selection pending", | |
| f"Source: {source_badge_label(node.source)}", | |
| f"Parameters: {_parameter_summary(node.parameters)}", | |
| f"Shape: {_shape_summary(node.input_shape, node.output_shape)}", | |
| f"Compatibility: {node.compatibility.title()}", | |
| ) | |
| ) | |
| for node in pipeline_nodes(state) | |
| ) | |
| _COMPATIBILITY_ICONS = { | |
| "compatible": "✓", | |
| "review": "!", | |
| "blocked": "×", | |
| } | |
| def pipeline_node_cards_html( | |
| state: Mapping[str, Any] | WorkbenchState, | |
| ) -> tuple[str, ...]: | |
| """Render the compact, escaped visual contract for every pipeline node.""" | |
| current = state_from_mapping(state) | |
| cards = [] | |
| for index, node in enumerate(pipeline_nodes(current), start=1): | |
| selected = node.node_id == current.selected_node_id | |
| selected_class = " is-selected" if selected else "" | |
| compatibility = node.compatibility | |
| source = source_badge_label(node.source) | |
| component = node.component or "Selection pending" | |
| parameters = _parameter_summary(node.parameters) | |
| shape = _shape_summary(node.input_shape, node.output_shape) | |
| compatibility_icon = _COMPATIBILITY_ICONS[compatibility] | |
| node_icon, stage_name = PIPELINE_NODE_PRESENTATION[node.node_id] | |
| stage_chip = PIPELINE_NODE_SHORT_LABELS[node.node_id] | |
| current_attribute = ' aria-current="true"' if selected else "" | |
| selected_label = ( | |
| '<span class="guided-node-selected">Inspecting</span>' | |
| if selected | |
| else "" | |
| ) | |
| cards.append( | |
| f'<article class="guided-pipeline-card-content {compatibility}' | |
| f'{selected_class}" data-node-id="{_escape(node.node_id)}" ' | |
| f'data-stage-order="{index:02d}"{current_attribute} ' | |
| f'aria-label="{_escape(stage_name)}: {_escape(component)}; ' | |
| f'{_escape(compatibility)}">' | |
| '<div class="guided-node-topline">' | |
| f'<span class="guided-node-icon" aria-hidden="true">{node_icon}</span>' | |
| f'<span class="guided-node-stage-name" title="{_escape(stage_name)}">' | |
| f"{_escape(stage_chip)}</span>" | |
| "</div>" | |
| f"{selected_label}" | |
| f'<h4 title="{_escape(component)}">{_escape(component)}</h4>' | |
| f'<p class="guided-node-parameter" title="{_escape(parameters)}">' | |
| f"{_escape(parameters)}</p>" | |
| f'<p class="guided-node-shape" title="{_escape(shape)}">' | |
| f"{_escape(shape)}</p>" | |
| '<footer class="guided-node-footer">' | |
| f'<span class="guided-node-source">{_escape(source)}</span>' | |
| f'<span class="guided-node-compatibility {compatibility}">' | |
| f'<b aria-hidden="true">{compatibility_icon}</b>' | |
| f"{_escape(compatibility.title())}</span>" | |
| "</footer></article>" | |
| ) | |
| return tuple(cards) | |
| def node_inspector_html( | |
| state: Mapping[str, Any] | WorkbenchState, | |
| ) -> str: | |
| current = state_from_mapping(state) | |
| node = next( | |
| item for item in pipeline_nodes(current) if item.node_id == current.selected_node_id | |
| ) | |
| trace = node.trace.to_dict() | |
| report = node.feature_report.to_dict() if node.feature_report else None | |
| trace_rows = { | |
| key: value | |
| for key, value in trace.items() | |
| if key not in {"extensions", "schema_version"} | |
| } | |
| report_html = ( | |
| '<div class="guided-inspector-contract"><h4>FeatureExtractionReport</h4>' | |
| f'{_mapping_rows(report)}</div>' | |
| if report is not None | |
| else ( | |
| '<div class="guided-inspector-contract"><h4>FeatureExtractionReport</h4>' | |
| '<p class="guided-empty-copy">Not applicable to this node</p></div>' | |
| ) | |
| ) | |
| return ( | |
| '<section class="guided-node-inspector">' | |
| '<div class="guided-inspector-heading"><div>' | |
| f'<span>{_escape(node.label)}</span><h3>{_escape(node.component)}</h3></div>' | |
| f'<span class="guided-compatibility {node.compatibility}">' | |
| f'{_escape(node.compatibility.title())}</span></div>' | |
| '<div class="guided-inspector-grid">' | |
| '<div><h4>Selection reason</h4>' | |
| f'<p>{_escape(node.rationale)}</p></div>' | |
| '<div><h4>Alternatives considered</h4>' | |
| f'{_list_html(node.alternatives)}</div>' | |
| "</div>" | |
| '<div class="guided-inspector-contract"><h4>DecisionTrace</h4>' | |
| '<div class="guided-detail-row"><span>Decision source</span><strong>' | |
| f'{_escape(source_badge_label(node.source))}</strong></div>' | |
| f"{_mapping_rows(trace_rows)}" | |
| "</div>" | |
| f"{report_html}" | |
| "</section>" | |
| ) | |
| def review_html(state: Mapping[str, Any] | WorkbenchState) -> str: | |
| current = state_from_mapping(state) | |
| issues = validate_state(current) | |
| if issues: | |
| return ( | |
| '<section class="guided-review blocked">' | |
| "<span>Needs attention</span><h3>The run plan is not ready</h3>" | |
| f'{_list_html(issues)}</section>' | |
| ) | |
| warnings = current.research_plan.warnings if current.research_plan else () | |
| return ( | |
| '<section class="guided-review ready">' | |
| "<span>Ready to run</span><h3>Review complete</h3>" | |
| '<div class="guided-review-grid">' | |
| '<div><small>Data</small><strong>' | |
| f'{_escape(current.dataset_name or current.upload_type)}</strong></div>' | |
| f'<div><small>Objective</small><strong>{_escape(current.target)}</strong></div>' | |
| '<div><small>Candidate</small><strong>' | |
| f'{_escape(candidate_display_name(current.current_candidate))}</strong></div>' | |
| f'<div><small>Quality profile</small><strong>{_escape(current.preset)}</strong></div>' | |
| f'<div><small>Time budget</small><strong>{current.time_limit} seconds</strong></div>' | |
| f'<div><small>Candidate trials</small><strong>{current.trials}</strong></div>' | |
| f'<div><small>Run variants</small><strong>{max(1, len(current.variants))}</strong></div>' | |
| "</div>" | |
| '<div class="guided-review-warnings"><h4>Warnings to review</h4>' | |
| f'{_list_html(warnings)}</div></section>' | |
| ) | |
| def run_plan_summary_html( | |
| state: Mapping[str, Any] | WorkbenchState, | |
| ) -> str: | |
| current = state_from_mapping(state) | |
| issues = validate_state(current) | |
| run_count = max(1, len(current.variants)) | |
| variant_label = "variant" if run_count == 1 else "variants" | |
| candidate = "Selection pending" | |
| try: | |
| candidate = candidate_display_name(current.current_candidate) | |
| except ValueError: | |
| pass | |
| status = "Ready" if not issues else f"{len(issues)} item(s) need attention" | |
| return ( | |
| '<section class="guided-run-summary">' | |
| '<div class="guided-summary-status"><span>Run plan</span>' | |
| f'<strong>{_escape(status)}</strong></div>' | |
| '<div class="guided-summary-facts">' | |
| f'<span><small>Candidate</small>{_escape(candidate)}</span>' | |
| f'<span><small>Quality</small>{_escape(current.preset)}</span>' | |
| f'<span><small>Budget</small>{current.time_limit} seconds</span>' | |
| f'<span><small>Trials</small>{current.trials}</span>' | |
| f'<span><small>Variants</small>{run_count} {variant_label}</span>' | |
| "</div></section>" | |
| ) | |
| def _research_plan_snapshot(state: WorkbenchState) -> str: | |
| if state.research_plan is None: | |
| return '<p class="guided-empty-copy">Research plan unavailable</p>' | |
| plan = state.research_plan | |
| evidence = tuple( | |
| _clean_text(item.get("summary") or item.get("evidence_id")) | |
| for item in plan.evidence | |
| ) | |
| return ( | |
| '<section class="guided-snapshot guided-research-snapshot">' | |
| '<div class="guided-snapshot-heading"><span>Pre-run record</span>' | |
| '<h3>Research plan</h3></div>' | |
| '<div class="guided-dossier-block"><h4>Recommendation</h4>' | |
| f'<strong>{_escape(plan.recommendation.get("candidate_id") or "Unavailable")}</strong>' | |
| f'<p>{_escape(plan.recommendation.get("rationale") or "No rationale provided")}</p></div>' | |
| '<div class="guided-dossier-block"><h4>Evidence</h4>' | |
| f'{_list_html(evidence)}</div>' | |
| '<div class="guided-dossier-block"><h4>Assumptions</h4>' | |
| f'{_list_html(plan.assumptions)}</div>' | |
| '<div class="guided-dossier-block"><h4>Warnings</h4>' | |
| f'{_list_html(plan.warnings)}</div>' | |
| '<div class="guided-candidate-section"><h4>Candidate estimates</h4>' | |
| f'{candidate_cards_html(state)}</div></section>' | |
| ) | |
| def _pipeline_trace_snapshot( | |
| state: WorkbenchState, | |
| traces: Sequence[DecisionTrace] = (), | |
| ) -> str: | |
| if traces: | |
| entries = [] | |
| for trace in traces: | |
| selected = ( | |
| trace.selected if trace.selected is not None else "Not reported" | |
| ) | |
| trace_rows = { | |
| "Source": source_badge_label(trace.decision_source), | |
| "Stage": trace.stage or trace.decision_type or "Not reported", | |
| "Selected": selected, | |
| "Alternatives": list(trace.options), | |
| "Rationale": trace.rationale or "Not reported", | |
| "Parameters": dict(trace.parameters), | |
| "Evidence references": list(trace.evidence_ids), | |
| "Warnings": list(trace.warnings), | |
| } | |
| entries.append( | |
| '<article class="guided-trace-entry guided-runtime-trace-entry">' | |
| f'<span>{_escape(trace.stage or trace.decision_type or "Decision")}</span>' | |
| f'<h4>{_escape(selected)}</h4>' | |
| f"{_mapping_rows(trace_rows)}" | |
| "</article>" | |
| ) | |
| return ( | |
| '<section class="guided-snapshot guided-trace-snapshot">' | |
| '<div class="guided-snapshot-heading"><span>Execution provenance</span>' | |
| '<h3>Pipeline trace</h3></div>' | |
| f'<div class="guided-trace-list">{"".join(entries)}</div>' | |
| '<div class="guided-dossier-block"><h4>Research assumptions</h4>' | |
| f'{_list_html(state.research_plan.assumptions if state.research_plan else ())}' | |
| '</div></section>' | |
| ) | |
| entries = [] | |
| for node in pipeline_nodes(state): | |
| entries.append( | |
| '<article class="guided-trace-entry">' | |
| f'<span>{_escape(node.label)}</span><h4>{_escape(node.component)}</h4>' | |
| f'<p><strong>Decision source:</strong> {_escape(source_badge_label(node.source))}</p>' | |
| f'<p>{_escape(node.rationale)}</p></article>' | |
| ) | |
| return ( | |
| '<section class="guided-snapshot guided-trace-snapshot">' | |
| '<div class="guided-snapshot-heading"><span>Planned provenance</span>' | |
| '<h3>Pipeline trace</h3></div>' | |
| f'<div class="guided-trace-list">{"".join(entries)}</div></section>' | |
| ) | |
| def _feature_report_snapshot( | |
| state: WorkbenchState, | |
| report: FeatureExtractionReport | None = None, | |
| ) -> str: | |
| if report is not None: | |
| warning_html = ( | |
| '<div class="guided-dossier-block"><h4>Warnings</h4>' | |
| f"{_list_html(report.warnings)}</div>" | |
| if report.warnings | |
| else "" | |
| ) | |
| return ( | |
| '<section class="guided-snapshot guided-feature-snapshot">' | |
| '<div class="guided-snapshot-heading"><span>Execution report</span>' | |
| '<h3>Feature extraction</h3></div>' | |
| f'{_mapping_rows(report.to_dict())}{warning_html}</section>' | |
| ) | |
| report = _planned_feature_report(state) | |
| return ( | |
| '<section class="guided-snapshot guided-feature-snapshot">' | |
| '<div class="guided-snapshot-heading"><span>Planned report</span>' | |
| '<h3>Feature extraction</h3></div>' | |
| '<p class="guided-contract-note">Execution shapes are unavailable ' | |
| "unless the backend reports them.</p>" | |
| f'{_mapping_rows(report.to_dict())}</section>' | |
| ) | |
| def _reflection_metric_cards( | |
| metrics: Mapping[str, Any], *, measured: bool | |
| ) -> str: | |
| if not metrics: | |
| return '<p class="guided-empty-copy">No metrics recorded</p>' | |
| cards = [] | |
| for name, raw in metrics.items(): | |
| value = raw.get("value") if isinstance(raw, Mapping) else raw | |
| estimate = ( | |
| raw.get("estimate") if not measured and isinstance(raw, Mapping) else None | |
| ) | |
| basis = raw.get("basis") if isinstance(raw, Mapping) else "" | |
| if measured and value is None: | |
| continue | |
| displayed = value if measured else ( | |
| estimate if estimate is not None else "Requires validation" | |
| ) | |
| kind = "Measured" if measured else "Estimate · unmeasured" | |
| cards.append( | |
| '<div class="guided-estimate guided-reflection-metric">' | |
| f'<small class="guided-metric-kind">{_escape(kind)}</small>' | |
| f'<span>{_escape(str(name).replace("_", " ").title())}</span>' | |
| f'<strong>{_escape(str(displayed))}</strong>' | |
| f'<small>{_escape(basis)}</small></div>' | |
| ) | |
| return "".join(cards) or '<p class="guided-empty-copy">No metrics recorded</p>' | |
| def experiment_reflection_outputs( | |
| execution_metadata: Mapping[str, Any] | None, | |
| ) -> tuple[str, dict[str, Any], tuple[tuple[str, str], ...]]: | |
| """Project canonical reflection metadata without granting execution authority.""" | |
| empty = ( | |
| '<section class="guided-reflection-empty">' | |
| '<h3>No experiment reflection</h3>' | |
| '<p>A completed governed run may provide a bounded reflection here.</p>' | |
| "</section>" | |
| ) | |
| if not isinstance(execution_metadata, Mapping) or not execution_metadata: | |
| return empty, {}, () | |
| try: | |
| normalized = normalize_execution_metadata(execution_metadata) | |
| raw_reflection = normalized.get("experiment_reflection") | |
| if not isinstance(raw_reflection, Mapping): | |
| return empty, {}, () | |
| reflection = ExperimentReflection.from_mapping(raw_reflection) | |
| except (TypeError, ValueError): | |
| return empty, {}, () | |
| approved = normalized.get("approved_research") | |
| recommendation_id = "Unavailable" | |
| recommendation_rationale = "No original recommendation was recorded." | |
| approved_id = "Unavailable" | |
| pipeline_config: Mapping[str, Any] = {} | |
| if isinstance(approved, Mapping): | |
| raw_plan = approved.get("research_plan") | |
| if isinstance(raw_plan, Mapping): | |
| try: | |
| plan = ResearchPlan.from_mapping(raw_plan) | |
| recommendation_id = _clean_text( | |
| plan.recommendation.get("candidate_id") or "Unavailable" | |
| ) | |
| recommendation_rationale = _clean_text( | |
| plan.recommendation.get("rationale") | |
| or "No original recommendation rationale was recorded." | |
| ) | |
| except (TypeError, ValueError): | |
| pass | |
| approved_id = _clean_text( | |
| approved.get("approved_candidate_id") or "Unavailable" | |
| ) | |
| if isinstance(approved.get("pipeline_config"), Mapping): | |
| pipeline_config = approved["pipeline_config"] | |
| actual_decisions = [] | |
| for raw_trace in normalized.get("decision_traces") or (): | |
| if not isinstance(raw_trace, Mapping): | |
| continue | |
| try: | |
| trace = DecisionTrace.from_mapping(raw_trace) | |
| except (TypeError, ValueError): | |
| continue | |
| if trace.stage in {"research_verification", "candidate_approval", "experiment_reflection"}: | |
| continue | |
| selected = trace.selected if trace.selected is not None else "Not reported" | |
| actual_decisions.append( | |
| f"{trace.stage or trace.decision_type}: {selected} " | |
| f"({source_badge_label(trace.decision_source)})" | |
| ) | |
| suggestion_cards = [] | |
| choices: list[tuple[str, str]] = [] | |
| for suggestion in reflection.suggestions: | |
| choices.append((suggestion.title, suggestion.suggestion_id)) | |
| status = ( | |
| "Executable · fresh approval required" | |
| if suggestion.executable and not suggestion.suggestion_only | |
| else "Suggestion only · cannot execute" | |
| ) | |
| suggestion_cards.append( | |
| '<article class="guided-next-experiment-card ' | |
| f'{"executable" if suggestion.executable else "suggestion-only"}">' | |
| f'<span>{_escape(status)}</span><h4>{_escape(suggestion.title)}</h4>' | |
| f'<p>{_escape(suggestion.rationale)}</p>' | |
| '<div><h5>Expected tradeoffs</h5>' | |
| f'{_list_html(suggestion.expected_tradeoffs)}</div>' | |
| '<div><h5>Expected metric estimates</h5>' | |
| '<div class="guided-estimate-grid">' | |
| f'{_reflection_metric_cards(suggestion.expected_metrics, measured=False)}' | |
| "</div></div>" | |
| '<div><h5>Evidence references</h5>' | |
| f'{_list_html(suggestion.evidence_ids)}</div>' | |
| '<div><h5>Warnings</h5>' | |
| f'{_list_html(suggestion.warnings)}</div></article>' | |
| ) | |
| provider_model = _clean_text( | |
| reflection.provider_provenance.get("model") or "Not configured" | |
| ) | |
| reflection_source = source_badge_label(reflection.source) | |
| html_output = ( | |
| '<section class="guided-experiment-reflection">' | |
| '<div class="guided-snapshot-heading"><span>Post-run interpretation</span>' | |
| '<h3>Experiment reflection</h3></div>' | |
| '<div class="guided-reflection-sections">' | |
| '<article><h4>Original research recommendation</h4>' | |
| f'<strong>{_escape(recommendation_id)}</strong>' | |
| f'<p>{_escape(recommendation_rationale)}</p></article>' | |
| '<article><h4>Actual executed pipeline</h4>' | |
| f'<strong>Approved candidate: {_escape(approved_id)}</strong>' | |
| f'{_mapping_rows(pipeline_config)}' | |
| f'{_list_html(actual_decisions, empty="No actual execution decisions recorded")}</article>' | |
| '<article><h4>Measured results</h4>' | |
| '<div class="guided-estimate-grid">' | |
| f'{_reflection_metric_cards(reflection.measured_metrics, measured=True)}' | |
| "</div></article>" | |
| '<article><h4>Experiment reflection</h4>' | |
| '<span class="guided-source-badge" data-source="' | |
| f'{_escape(reflection.source.value)}">{_escape(reflection_source)}</span>' | |
| f'<p><strong>Expectations: {_escape(reflection.expectations_met.replace("_", " ").title())}</strong></p>' | |
| f'<p>{_escape(reflection.summary)}</p><p>{_escape(reflection.interpretation)}</p>' | |
| '<h5>Observed limitations</h5>' | |
| f'{_list_html(reflection.limitations)}' | |
| '<h5>Evidence references</h5>' | |
| f'{_list_html(reflection.evidence_ids)}' | |
| '<h5>Warnings</h5>' | |
| f'{_list_html(reflection.warnings)}' | |
| f'<small>Model: {_escape(provider_model)} · no automatic execution</small>' | |
| "</article></div>" | |
| '<div class="guided-dossier-block"><h4>Suggested next experiments</h4>' | |
| f'<div class="guided-next-experiment-grid">{"".join(suggestion_cards)}</div>' | |
| '<p class="guided-provider-note">Using a suggestion creates an unapproved ' | |
| "draft only. Fresh research, verification, validation, and explicit approval " | |
| "remain required.</p></div></section>" | |
| ) | |
| return html_output, reflection.to_dict(), tuple(choices) | |
| def snapshot_outputs( | |
| state: Mapping[str, Any] | WorkbenchState, | |
| execution_metadata: Mapping[str, Any] | None = None, | |
| ) -> tuple[str, str, str]: | |
| current = state_from_mapping(state) | |
| report, traces = _runtime_contracts(execution_metadata) | |
| return ( | |
| _research_plan_snapshot(current), | |
| _pipeline_trace_snapshot(current, traces), | |
| _feature_report_snapshot(current, report), | |
| ) | |
| __all__ = [ | |
| "PIPELINE_NODE_IDS", | |
| "SOURCE_LABELS", | |
| "WORKBENCH_STEPS", | |
| "PipelineNodeView", | |
| "RunVariant", | |
| "WorkbenchState", | |
| "add_variant", | |
| "apply_next_experiment_draft", | |
| "apply_refinement", | |
| "candidate_is_executable", | |
| "candidate_verification_map", | |
| "candidate_cards_html", | |
| "experiment_reflection_outputs", | |
| "generate_research_plan", | |
| "new_workbench_state", | |
| "node_inspector_html", | |
| "pipeline_node_cards_html", | |
| "pipeline_node_labels", | |
| "pipeline_nodes", | |
| "research_dossier_html", | |
| "review_html", | |
| "run_plan_summary_html", | |
| "select_candidate", | |
| "select_node", | |
| "set_step", | |
| "snapshot_outputs", | |
| "source_badge_label", | |
| "state_from_mapping", | |
| "update_context", | |
| "validate_state", | |
| ] | |