guided_workbench.py

"""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"}
    )


@dataclass(frozen=True)
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)),
        }


@dataclass(frozen=True)
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


@dataclass(frozen=True)
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))

    @property
    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",
]