api.py

# ============================================================
# 🚀 DFE-AutoML Streaming Backend v17.0
# ------------------------------------------------------------
# Research-aligned stack:
# RL • Explainability • Two-Speed Fairness • Mapper • Drift
# + Fairness Scope Manager • Streaming Fairness Measurement
# + Dynamic Fairness Envelope • Chunked Streaming Execution
# + Hard Thread Safety • CO2 Tracking (CodeCarbon)
#
# v17.0 upgrades:
# ✅ Fixed controller fairness control mismatch
# ✅ Controller now uses SFM-aligned fairness control signal
# ✅ Added DynamicFairnessEnvelope integration
# ✅ Added SFM telemetry to controller + response contract
# ✅ Added DFE telemetry to controller + response contract
# ✅ Preserved legacy outputs during transition
# ✅ Preserved scope-aware end-to-end decision_context propagation
# ============================================================

from __future__ import annotations

import copy
import inspect
import logging
import os
import platform
import threading
import time
import traceback
import uuid
from collections import deque
from dataclasses import MISSING, dataclass, field, fields as dataclass_fields
from typing import Any, Dict, List, Optional, Tuple

import numpy as np
import pandas as pd
from flask import Flask, jsonify, request
from river import drift, metrics

from autostream_core import AutoStreamCore
from adaptive_controller import AdaptiveFairnessController
from dynamic_fairness_envelope import DynamicFairnessEnvelope, DynamicFairnessEnvelopeConfig
from explainability_engine import StreamingExplainabilityEngine
from fairness_engine import FairnessEngineConfig, StreamingFairnessEngine
from metrics_stream import StreamingFairnessMetric
from topological_bias_engine import TopologicalBiasEngine
from llm_auditor import StreamingLLMAuditor
from fairness_scope_manager import FairnessScopeManager
from config_space import ConfigSpaceSettings, StreamingConfigSpace
from candidate_evaluator import CandidateEvaluationResult, CandidateEvaluator, CandidateEvaluatorConfig
from osmac_optimizer import OSMACOptimizer, OSMACOptimizerConfig
from offline_benchmarks import (
    StreamingBenchmarkSpec,
    run_streaming_fairbo_baseline,
)


# ============================================================
# Version + Schema
# ============================================================

BACKEND_VERSION = "17.0"
SCHEMA_VERSION = "v17.0"


# ============================================================
# Limits / Constants
# ============================================================

MAX_HISTORY = int(os.getenv("FAIRASML_MAX_HISTORY", "5000") or "5000")
DEFAULT_CHUNK_SIZE = int(os.getenv("FAIRASML_DEFAULT_CHUNK", "200") or "200")

MAPPER_UPDATE_FREQUENCY = int(os.getenv("FAIRASML_MAPPER_FREQ", "100") or "100")
LLM_COOLDOWN = int(os.getenv("FAIRASML_LLM_COOLDOWN", "300") or "300")

# Ollama
OLLAMA_MODEL = os.getenv("FAIRASML_OLLAMA_MODEL", "llama3:latest")
OLLAMA_URL = os.getenv("FAIRASML_OLLAMA_URL", "http://localhost:11434/api/generate")
OLLAMA_TIMEOUT_SECONDS = float(os.getenv("FAIRASML_OLLAMA_TIMEOUT", "300") or "300")
OLLAMA_MAX_RETRIES = int(os.getenv("FAIRASML_OLLAMA_MAX_RETRIES", "2") or "2")
OLLAMA_CALL_COOLDOWN = int(os.getenv("FAIRASML_OLLAMA_CALL_COOLDOWN", "15") or "15")
OLLAMA_UNAVAILABLE_COOLDOWN = int(os.getenv("FAIRASML_OLLAMA_UNAVAILABLE_COOLDOWN", "120") or "120")

STABILITY_WARNING_THRESHOLD = float(os.getenv("FAIRASML_STABILITY_WARN", "0.15") or "0.15")
PERF_DRIFT_LATCH_STEPS = int(os.getenv("FAIRASML_PERF_LATCH", "40") or "40")
PERF_DRIFT_DELTA = float(os.getenv("FAIRASML_PERF_DRIFT_DELTA", "0.002") or "0.002")

MAX_RISE_PER_GROUP = int(os.getenv("FAIRASML_MAX_RISE_PER_GROUP", "500") or "500")
MAX_GROUPS_RISE = int(os.getenv("FAIRASML_MAX_GROUPS_RISE", "64") or "64")

EXPERIMENT_TTL_SECONDS = int(os.getenv("FAIRASML_EXPERIMENT_TTL", "0") or "0")

MAX_DATASET_ROWS = int(os.getenv("FAIRASML_MAX_DATASET_ROWS", "200000") or "200000")
MAX_DATASET_COLS = int(os.getenv("FAIRASML_MAX_DATASET_COLS", "500") or "500")

SCOPE_REEVAL_FREQUENCY = int(os.getenv("FAIRASML_SCOPE_REEVAL_FREQ", "50") or "50")
OSMAC_MIN_WINDOW = int(os.getenv("FAIRASML_OSMAC_MIN_WINDOW", "25") or "25")
OSMAC_MAX_WINDOW = int(os.getenv("FAIRASML_OSMAC_MAX_WINDOW", "250") or "250")
OSMAC_HISTORY_TAIL = int(os.getenv("FAIRASML_OSMAC_HISTORY_TAIL", "40") or "40")
MAX_OPTIMIZER_CANDIDATE_HISTORY = int(
    os.getenv("FAIRASML_OPTIMIZER_CANDIDATE_HISTORY", "2048") or "2048"
)
DEFAULT_STATIC_BENCHMARK_BUDGET = int(os.getenv("FAIRASML_STATIC_BENCHMARK_BUDGET", "12") or "12")
DEFAULT_STATIC_BENCHMARK_POOL = int(os.getenv("FAIRASML_STATIC_BENCHMARK_POOL", "96") or "96")


# ============================================================
# Logging
# ============================================================

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("fairasml.api")


# ============================================================
# Flask + Global Registry
# ============================================================

app = Flask(__name__)

EXPERIMENTS: Dict[str, "ExperimentState"] = {}
EXPERIMENTS_LOCK = threading.Lock()


# ============================================================
# Optional: CodeCarbon
# ============================================================

try:
    from codecarbon import EmissionsTracker  # type: ignore
    CODECARBON_AVAILABLE = True
except Exception:
    EmissionsTracker = None  # type: ignore
    CODECARBON_AVAILABLE = False


# ============================================================
# Utilities
# ============================================================

def _now() -> float:
    return float(time.time())


def safe_float(x: Any, default: float = 0.0) -> float:
    try:
        if x is None:
            return default
        v = float(x)
        if np.isnan(v) or np.isinf(v):
            return default
        return v
    except Exception:
        return default


def safe_int(x: Any, default: int = 0) -> int:
    try:
        if x is None:
            return default
        if isinstance(x, bool):
            return int(x)
        if isinstance(x, (np.integer, int)):
            return int(x)
        return int(float(x))
    except Exception:
        return default


def safe_bool(x: Any, default: bool = False) -> bool:
    if x is None:
        return bool(default)
    if isinstance(x, bool):
        return bool(x)
    if isinstance(x, (int, np.integer)):
        return bool(int(x))
    if isinstance(x, (float, np.floating)):
        return bool(safe_float(x, 1.0 if default else 0.0) >= 0.5)
    s = str(x).strip().lower()
    if s in {"1", "true", "yes", "y", "on"}:
        return True
    if s in {"0", "false", "no", "n", "off"}:
        return False
    return bool(default)


def safe_str(x: Any, default: str = "") -> str:
    if x is None:
        return str(default)
    return str(x)


def bounded_append(lst: List[Any], value: Any, max_len: int = MAX_HISTORY) -> None:
    lst.append(value)
    if len(lst) > max_len:
        del lst[0]


def make_json_safe(obj: Any) -> Any:
    if obj is None:
        return None

    if isinstance(obj, dict):
        return {str(k): make_json_safe(v) for k, v in obj.items()}

    if isinstance(obj, (list, tuple, set, deque)):
        return [make_json_safe(v) for v in obj]

    if isinstance(obj, np.ndarray):
        return obj.tolist()

    if isinstance(obj, np.generic):
        obj = obj.item()

    if isinstance(obj, (np.floating, float)):
        return safe_float(obj)

    if isinstance(obj, (np.integer, int)):
        return int(obj)

    if isinstance(obj, (str, bool)):
        return obj

    return str(obj)


def _err_payload(
    code: str,
    message: str,
    http_status: int = 400,
    details: Optional[Dict[str, Any]] = None,
):
    payload: Dict[str, Any] = {
        "status": "error",
        "error": code,
        "message": message,
        "backend_version": BACKEND_VERSION,
        "schema_version": SCHEMA_VERSION,
    }
    if details:
        payload["details"] = make_json_safe(details)
    return jsonify(payload), http_status


def _normalize_value(v: Any) -> Any:
    if v is None:
        return "unknown"
    if isinstance(v, (np.integer, int)):
        return int(v)
    if isinstance(v, (np.floating, float)):
        fv = safe_float(v, 0.0)
        if abs(fv - round(fv)) < 1e-9:
            return int(round(fv))
        return fv
    return str(v)


def _normalize_sensitive_dict(xi_full: Dict[str, Any], sensitive_cols: List[str]) -> Dict[str, Any]:
    out: Dict[str, Any] = {}
    for s in sensitive_cols:
        out[s] = _normalize_value(xi_full.get(s, None))
    return out


def _coerce_binary_target(v: Any) -> int:
    if v is None:
        return 0

    if isinstance(v, bool):
        return int(v)

    if isinstance(v, (int, np.integer)):
        return 1 if int(v) == 1 else 0

    if isinstance(v, (float, np.floating)):
        fv = safe_float(v, 0.0)
        return 1 if fv >= 1.0 else 0

    s = str(v).strip().lower()
    if s in {"1", "true", "yes", "y", ">50k", "positive", "pos", "good", "approved", "accept"}:
        return 1
    if s in {"0", "false", "no", "n", "<=50k", "negative", "neg", "bad", "rejected", "reject"}:
        return 0
    return 0


def _build_encoded_feature_matrix(
    df: pd.DataFrame,
    target_col: str,
    sensitive_cols: List[str],
) -> Tuple[List[Dict[str, float]], Dict[str, Any]]:
    excluded = set([target_col] + list(sensitive_cols))
    base_cols = [c for c in df.columns if c not in excluded]

    feature_df = df[base_cols].copy()

    numeric_cols: List[str] = []
    categorical_cols: List[str] = []
    dropped_cols: List[str] = []

    for col in list(feature_df.columns):
        s = feature_df[col]

        if pd.api.types.is_bool_dtype(s):
            feature_df[col] = s.astype(int)
            numeric_cols.append(str(col))
            continue

        if pd.api.types.is_numeric_dtype(s):
            feature_df[col] = pd.to_numeric(s, errors="coerce")
            numeric_cols.append(str(col))
            continue

        try:
            feature_df[col] = s.astype("string").fillna("__MISSING__")
            categorical_cols.append(str(col))
        except Exception:
            dropped_cols.append(str(col))

    usable_cols = [c for c in feature_df.columns if c not in dropped_cols]
    feature_df = feature_df[usable_cols]

    if categorical_cols:
        encoded = pd.get_dummies(
            feature_df,
            columns=[c for c in categorical_cols if c in feature_df.columns],
            dummy_na=False,
        )
    else:
        encoded = feature_df.copy()

    encoded = encoded.replace([np.inf, -np.inf], np.nan).fillna(0.0)

    for c in encoded.columns:
        encoded[c] = pd.to_numeric(encoded[c], errors="coerce").fillna(0.0).astype(float)

    X_model = encoded.to_dict(orient="records")

    feature_manifest = {
        "base_feature_columns": [str(c) for c in base_cols],
        "numeric_feature_columns": numeric_cols,
        "categorical_feature_columns": categorical_cols,
        "encoded_feature_columns": [str(c) for c in encoded.columns],
        "dropped_feature_columns": dropped_cols,
        "excluded_columns": sorted(list(excluded)),
        "encoded_feature_count": int(len(encoded.columns)),
    }

    return X_model, feature_manifest


def _get_proba_1(proba: Dict[Any, float]) -> float:
    if not isinstance(proba, dict) or not proba:
        return 0.5

    normalized: Dict[str, float] = {}
    raw_numeric_one: Optional[float] = None
    raw_numeric_zero: Optional[float] = None

    for k, v in proba.items():
        pv = float(np.clip(safe_float(v, 0.0), 0.0, 1.0))

        if k == 1:
            raw_numeric_one = pv
        elif k == 0:
            raw_numeric_zero = pv
        elif k is True:
            raw_numeric_one = pv
        elif k is False:
            raw_numeric_zero = pv

        ks = str(k).strip().lower()
        normalized[ks] = pv

    positive_aliases = {"1", "true", "yes", "positive", "pos"}
    negative_aliases = {"0", "false", "no", "negative", "neg"}

    if raw_numeric_one is not None:
        return float(np.clip(raw_numeric_one, 0.0, 1.0))

    for key in positive_aliases:
        if key in normalized:
            return float(np.clip(normalized[key], 0.0, 1.0))

    if raw_numeric_zero is not None:
        return float(np.clip(1.0 - raw_numeric_zero, 0.0, 1.0))

    for key in negative_aliases:
        if key in normalized:
            return float(np.clip(1.0 - normalized[key], 0.0, 1.0))

    vals = [float(np.clip(safe_float(v, 0.0), 0.0, 1.0)) for v in proba.values()]
    if len(vals) == 2:
        s = sum(vals)
        if 0.95 <= s <= 1.05:
            return 0.5

    return 0.5


def _get_control_fairness_value(snapshot: Dict[str, Any], default: float = 0.0) -> float:
    if not isinstance(snapshot, dict):
        return default
    return safe_float(snapshot.get("primary_metric", snapshot.get("ddsp", default)), default)


def _get_sfm_payload(snapshot: Dict[str, Any]) -> Dict[str, Any]:
    if not isinstance(snapshot, dict):
        return {}
    sfm = snapshot.get("sfm_payload", {})
    if isinstance(sfm, dict) and sfm:
        return sfm

    if any(k in snapshot for k in ("fm_raw", "fm_smooth", "fm_uncertainty")):
        return {
            "fm_raw": safe_float(snapshot.get("fm_raw", snapshot.get("primary_metric", 0.0)), 0.0),
            "fm_smooth": safe_float(snapshot.get("fm_smooth", snapshot.get("primary_metric", 0.0)), 0.0),
            "fm_drift": safe_float(snapshot.get("fm_drift", 0.0), 0.0),
            "fm_smooth_drift": safe_float(snapshot.get("fm_smooth_drift", snapshot.get("fm_drift", 0.0)), 0.0),
            "fm_uncertainty": safe_float(snapshot.get("fm_uncertainty", 1.0), 1.0),
            "confidence": safe_float(snapshot.get("confidence", 0.0), 0.0),
            "is_reliable": bool(snapshot.get("is_reliable", False)),
            "insufficient_support": bool(snapshot.get("insufficient_support", True)),
            "metric_name": str(snapshot.get("primary_metric_name", "ddsp@global")),
            "metric_key": str(snapshot.get("metric_key", "ddsp")),
            "metric_target": str(snapshot.get("metric_target", "global")),
            "window_support": safe_int(snapshot.get("window_support", 0), 0),
            "effective_support": safe_int(snapshot.get("effective_support", 0), 0),
            "eligible_groups": safe_int(snapshot.get("eligible_groups", 0), 0),
            "groups_tracked": safe_int(snapshot.get("groups_tracked", 0), 0),
            "group_counts": dict(snapshot.get("group_support", {})) if isinstance(snapshot.get("group_support"), dict) else {},
            "low_support_groups": list(snapshot.get("low_support_groups", [])) if isinstance(snapshot.get("low_support_groups"), list) else [],
            "sfm": snapshot.get("sfm", None),
            "sfm_state": copy.deepcopy(snapshot.get("sfm_state", {})) if isinstance(snapshot.get("sfm_state"), dict) else {},
            "value": safe_float(snapshot.get("fm_smooth", snapshot.get("primary_metric", 0.0)), 0.0),
            "primary_metric": safe_float(snapshot.get("fm_smooth", snapshot.get("primary_metric", 0.0)), 0.0),
        }

    return {}


def _normalize_primary_metric_key(metric_key: Any) -> str:
    key = str(metric_key or "ddsp").strip().lower()
    alias = {
        "eo": "eopp",
        "equal_opportunity": "eopp",
        "equalized_odds": "eodds",
        "equalized_odds_proxy": "eodds",
        "ppv": "predictive_parity",
        "predictive_parity_gap": "predictive_parity",
    }
    key = alias.get(key, key)
    if key not in {"ddsp", "error_rate", "eopp", "eodds", "predictive_parity"}:
        return "ddsp"
    return key


def _parse_primary_metric_name(metric_name: Any) -> Tuple[str, str]:
    raw = str(metric_name or "").strip()
    if not raw:
        return "ddsp", "global"
    metric_key, _, metric_target = raw.partition("@")
    return _normalize_primary_metric_key(metric_key), (metric_target or "global").strip()


def _select_metric_diagnostic(
    metrics_snapshot: Dict[str, Any],
    metric_key: str,
    metric_target: str,
) -> Dict[str, Any]:
    if not isinstance(metrics_snapshot, dict):
        return {}

    global_key_map = {
        "ddsp": "ddsp",
        "error_rate": "error_rate",
        "eopp": "equal_opportunity",
        "eodds": "equalized_odds_proxy",
        "predictive_parity": "predictive_parity",
    }
    scoped_key_map = {
        "ddsp": "ddsp",
        "error_rate": "error_rate",
        "eopp": "eopp",
        "eodds": "eodds",
        "predictive_parity": "predictive_parity",
    }

    metric_key = _normalize_primary_metric_key(metric_key)
    metric_target = str(metric_target or "global").strip()

    if metric_target.startswith("attribute:"):
        attr = metric_target.split(":", 1)[1].strip()
        attr_metrics = metrics_snapshot.get("attribute_metrics", {})
        if isinstance(attr_metrics, dict):
            attr_payload = attr_metrics.get(attr, {})
            if isinstance(attr_payload, dict):
                diag = attr_payload.get("diagnostic_uncertainty", {})
                if isinstance(diag, dict):
                    sel = diag.get(scoped_key_map[metric_key], {})
                    return sel if isinstance(sel, dict) else {}

    if metric_target.startswith("intersection:"):
        inter = metric_target.split(":", 1)[1].strip()
        inter_metrics = metrics_snapshot.get("intersection_metrics", {})
        if isinstance(inter_metrics, dict):
            inter_payload = inter_metrics.get(inter, {})
            if isinstance(inter_payload, dict):
                diag = inter_payload.get("diagnostic_uncertainty", {})
                if isinstance(diag, dict):
                    sel = diag.get(scoped_key_map[metric_key], {})
                    return sel if isinstance(sel, dict) else {}

    diag = metrics_snapshot.get("diagnostic_uncertainty", {})
    if not isinstance(diag, dict):
        return {}
    sel = diag.get(global_key_map[metric_key], {})
    return sel if isinstance(sel, dict) else {}


def _build_sfm_payload(
    fairness_snapshot: Dict[str, Any],
    metrics_snapshot: Dict[str, Any],
    prev_fm_smooth: Optional[float] = None,
    smooth_alpha: float = 0.35,
) -> Dict[str, Any]:
    fairness_snapshot = fairness_snapshot if isinstance(fairness_snapshot, dict) else {}
    metrics_snapshot = metrics_snapshot if isinstance(metrics_snapshot, dict) else {}

    existing = _get_sfm_payload(fairness_snapshot)
    if existing:
        return existing

    metric_key, metric_target = _parse_primary_metric_name(fairness_snapshot.get("primary_metric_name", "ddsp@global"))
    metric_name = f"{metric_key}@{metric_target}"

    fm_raw = safe_float(
        fairness_snapshot.get("primary_metric", fairness_snapshot.get("ddsp", 0.0)),
        0.0,
    )
    prev_smooth = None if prev_fm_smooth is None else safe_float(prev_fm_smooth, fm_raw)
    if prev_smooth is None:
        fm_smooth = fm_raw
        fm_drift = 0.0
    else:
        alpha = float(np.clip(smooth_alpha, 0.0, 1.0))
        fm_smooth = (alpha * fm_raw) + ((1.0 - alpha) * prev_smooth)
        fm_drift = fm_smooth - prev_smooth

    diag = _select_metric_diagnostic(metrics_snapshot, metric_key=metric_key, metric_target=metric_target)
    diag_support = diag.get("group_support", {})
    group_counts = dict(diag_support) if isinstance(diag_support, dict) else {}
    low_support_groups = list(diag.get("low_support_groups", [])) if isinstance(diag.get("low_support_groups"), list) else []

    uncertainty_default = 1.0 if not diag else 0.0
    fm_uncertainty = float(np.clip(safe_float(diag.get("uncertainty", uncertainty_default), uncertainty_default), 0.0, 1.0))
    confidence = float(np.clip(
        safe_float(diag.get("confidence", 1.0 - min(1.0, fm_uncertainty)), 1.0 - min(1.0, fm_uncertainty)),
        0.0,
        1.0,
    ))
    is_reliable = bool(diag.get("is_reliable", False))
    insufficient_support = bool(diag.get("insufficient_support", not is_reliable))
    effective_support = safe_int(diag.get("effective_support", 0), 0)
    eligible_groups = safe_int(diag.get("eligible_groups", 0), 0)
    groups_tracked = safe_int(diag.get("groups_tracked", len(group_counts)), len(group_counts))
    window_support = safe_int(diag.get("window_support", fairness_snapshot.get("groups_tracked", 0)), 0)

    sfm_triplet = [
        float(fm_smooth),
        float(fm_drift),
        float(fm_uncertainty),
    ]

    return {
        "metric_name": metric_name,
        "metric_key": metric_key,
        "metric_target": metric_target,
        "fm_raw": float(fm_raw),
        "fm_smooth": float(fm_smooth),
        "fm_drift": float(fm_drift),
        "fm_smooth_drift": float(fm_drift),
        "fm_uncertainty": float(fm_uncertainty),
        "confidence": float(confidence),
        "is_reliable": bool(is_reliable),
        "insufficient_support": bool(insufficient_support),
        "window_support": int(window_support),
        "effective_support": int(effective_support),
        "eligible_groups": int(eligible_groups),
        "groups_tracked": int(groups_tracked),
        "group_counts": group_counts,
        "low_support_groups": low_support_groups,
        "sfm": sfm_triplet,
        "sfm_state": {
            "fm_stable": float(fm_smooth),
            "delta_fm": float(fm_drift),
            "sigma": float(fm_uncertainty),
            "confidence": float(confidence),
            "is_reliable": bool(is_reliable),
        },
        "value": float(fm_smooth),
        "primary_metric": float(fm_smooth),
    }


def _update_y_prob_debug(exp: "ExperimentState", y_prob: float) -> None:
    bounded_append(
        exp.y_prob_history,
        float(np.clip(safe_float(y_prob, 0.5), 0.0, 1.0)),
        max_len=MAX_HISTORY,
    )

    arr = np.asarray(exp.y_prob_history, dtype=float)
    if arr.size == 0:
        exp.y_prob_debug_snapshot = {
            "count": 0,
            "mean": 0.0,
            "std": 0.0,
            "min": 0.0,
            "max": 0.0,
            "share_gt_05": 0.0,
            "share_near_05_pm_005": 0.0,
        }
        return

    exp.y_prob_debug_snapshot = {
        "count": int(arr.size),
        "mean": safe_float(arr.mean(), 0.0),
        "std": safe_float(arr.std(), 0.0),
        "min": safe_float(arr.min(), 0.0),
        "max": safe_float(arr.max(), 0.0),
        "share_gt_05": safe_float(np.mean(arr > 0.5), 0.0),
        "share_near_05_pm_005": safe_float(np.mean(np.abs(arr - 0.5) <= 0.05), 0.0),
    }


def _detector_drift_flag(detector: Any) -> bool:
    for attr in ("drift_detected", "change_detected"):
        if hasattr(detector, attr):
            try:
                return bool(getattr(detector, attr))
            except Exception:
                return False
    if hasattr(detector, "warning_detected"):
        try:
            return bool(getattr(detector, "warning_detected"))
        except Exception:
            return False
    return False


def _cleanup_expired_experiments() -> None:
    if EXPERIMENT_TTL_SECONDS <= 0:
        return
    now = _now()
    expired: List[str] = []
    with EXPERIMENTS_LOCK:
        for eid, exp in list(EXPERIMENTS.items()):
            if (now - float(exp.last_poll_ts)) > EXPERIMENT_TTL_SECONDS:
                expired.append(eid)
        for eid in expired:
            EXPERIMENTS.pop(eid, None)
    if expired:
        logger.info("Evicted %s experiments (TTL=%ss).", len(expired), EXPERIMENT_TTL_SECONDS)


def _call_with_optional_kwargs(fn: Any, *args: Any, **kwargs: Any) -> Any:
    try:
        return fn(*args, **kwargs)
    except TypeError:
        pass

    try:
        sig = inspect.signature(fn)
        allowed = {k: v for k, v in kwargs.items() if k in sig.parameters}
        return fn(*args, **allowed)
    except Exception:
        pass

    return fn(*args)


def _build_dataclass_config(config_cls: Any, payload: Any) -> Any:
    if not isinstance(payload, dict):
        return config_cls()

    kwargs: Dict[str, Any] = {}
    for f in dataclass_fields(config_cls):
        if f.name not in payload:
            continue

        raw = payload.get(f.name)
        default = None if f.default is MISSING else f.default

        if isinstance(default, bool):
            kwargs[f.name] = safe_bool(raw, default)
        elif isinstance(default, int) and not isinstance(default, bool):
            kwargs[f.name] = safe_int(raw, default)
        elif isinstance(default, float):
            kwargs[f.name] = safe_float(raw, default)
        elif isinstance(default, str):
            kwargs[f.name] = str(raw)
        else:
            kwargs[f.name] = raw

    return config_cls(**kwargs)


# ============================================================
# Fairness scope helpers
# ============================================================

def _default_scope_config(
    sensitive_attr: List[str],
    window_size: int,
) -> Dict[str, Any]:
    return {
        "mode": "manual" if sensitive_attr else "adaptive",
        "approved_candidates": list(sensitive_attr),
        "blocked_candidates": [],
        "max_active_attributes": max(1, min(2, len(sensitive_attr) if sensitive_attr else 1)),
        "allow_intersections": True,
        "fairness_threshold": 0.05,
        "window_size": window_size,
    }


def _resolve_scope_config(
    payload_scope_cfg: Any,
    sensitive_attr: List[str],
    window_size: int,
) -> Dict[str, Any]:
    cfg = _default_scope_config(sensitive_attr, window_size)

    if not isinstance(payload_scope_cfg, dict):
        return cfg

    mode = str(payload_scope_cfg.get("mode", cfg["mode"])).strip().lower()
    if mode not in {"manual", "assisted", "adaptive"}:
        mode = cfg["mode"]

    approved_candidates = payload_scope_cfg.get("approved_candidates", cfg["approved_candidates"])
    if approved_candidates and not isinstance(approved_candidates, list):
        approved_candidates = [approved_candidates]
    approved_candidates = [str(x) for x in approved_candidates] if approved_candidates else list(cfg["approved_candidates"])

    blocked_candidates = payload_scope_cfg.get("blocked_candidates", [])
    if blocked_candidates and not isinstance(blocked_candidates, list):
        blocked_candidates = [blocked_candidates]
    blocked_candidates = [str(x) for x in blocked_candidates] if blocked_candidates else []

    return {
        "mode": mode,
        "approved_candidates": approved_candidates,
        "blocked_candidates": blocked_candidates,
        "max_active_attributes": max(
            1,
            safe_int(payload_scope_cfg.get("max_active_attributes", cfg["max_active_attributes"]), cfg["max_active_attributes"]),
        ),
        "allow_intersections": bool(payload_scope_cfg.get("allow_intersections", cfg["allow_intersections"])),
        "fairness_threshold": float(np.clip(
            safe_float(payload_scope_cfg.get("fairness_threshold", cfg["fairness_threshold"]), cfg["fairness_threshold"]),
            0.0,
            1.0,
        )),
        "window_size": max(10, safe_int(payload_scope_cfg.get("window_size", window_size), window_size)),
    }


def _default_optimizer_runtime_config(window_size: int) -> Dict[str, Any]:
    eval_window = max(OSMAC_MIN_WINDOW, min(OSMAC_MAX_WINDOW, int(window_size)))
    return {
        "enabled": True,
        "window_size": int(eval_window),
        "min_samples": int(max(OSMAC_MIN_WINDOW, min(eval_window, 50))),
        "step_frequency": int(eval_window),
        "include_incumbent": True,
        "force_exploration_on_violation": True,
        "history_tail": int(max(5, min(OSMAC_HISTORY_TAIL, 200))),
    }


def _resolve_optimizer_runtime_config(
    runtime_payload: Any,
    window_size: int,
) -> Dict[str, Any]:
    cfg = _default_optimizer_runtime_config(window_size)
    incoming = runtime_payload if isinstance(runtime_payload, dict) else {}

    cfg["enabled"] = safe_bool(incoming.get("enabled", cfg["enabled"]), cfg["enabled"])
    cfg["window_size"] = max(
        OSMAC_MIN_WINDOW,
        min(OSMAC_MAX_WINDOW, safe_int(incoming.get("window_size", cfg["window_size"]), cfg["window_size"])),
    )
    cfg["min_samples"] = max(
        2,
        min(cfg["window_size"], safe_int(incoming.get("min_samples", cfg["min_samples"]), cfg["min_samples"])),
    )
    cfg["step_frequency"] = max(
        1,
        safe_int(incoming.get("step_frequency", cfg["step_frequency"]), cfg["step_frequency"]),
    )
    cfg["include_incumbent"] = safe_bool(
        incoming.get("include_incumbent", cfg["include_incumbent"]),
        cfg["include_incumbent"],
    )
    cfg["force_exploration_on_violation"] = safe_bool(
        incoming.get("force_exploration_on_violation", cfg["force_exploration_on_violation"]),
        cfg["force_exploration_on_violation"],
    )
    cfg["history_tail"] = max(
        5,
        min(200, safe_int(incoming.get("history_tail", cfg["history_tail"]), cfg["history_tail"])),
    )
    return cfg


def _build_envelope_config(
    fairness_scope_config: Dict[str, Any],
) -> DynamicFairnessEnvelopeConfig:
    threshold = float(np.clip(safe_float(fairness_scope_config.get("fairness_threshold", 0.05), 0.05), 0.02, 0.10))
    min_upper = float(threshold)
    max_upper = float(np.clip(max(threshold, threshold * 2.0), threshold, 0.10))
    beta_relax = float(min(0.003, max(0.001, threshold * 0.05)))
    max_lower = float(min(0.02, threshold * 0.5))
    enforce_lower_bound = safe_bool(fairness_scope_config.get("enforce_lower_bound", False), False)

    return DynamicFairnessEnvelopeConfig(
        initial_upper=threshold,
        initial_lower=0.0,
        min_lower=0.0,
        max_lower=max_lower,
        min_upper=min_upper,
        max_upper=max_upper,
        alpha_drift=0.25,
        alpha_uncertainty=0.15,
        alpha_violation=0.35,
        alpha_lower_drift=0.05,
        alpha_lower_uncertainty=0.03,
        beta_relax=beta_relax,
        beta_lower_relax=0.0005,
        warning_margin_ratio=0.10,
        trend_alert_ratio=0.08,
        repeat_violation_tighten_bonus=0.015,
        consecutive_warning_tighten_bonus=0.003,
        enforce_lower_bound=enforce_lower_bound,
    )


def _resolve_streaming_optimizer_method(value: Any) -> str:
    method = safe_str(value, "").strip().lower()
    aliases = {
        "dfe-automl": "online_fairbo_ensemble",
        "dfe_automl": "online_fairbo_ensemble",
        "dfeautoml": "online_fairbo_ensemble",
        "onlinefairbo": "online_fairbo_basic",
        "online_fairbo": "online_fairbo_basic",
        "onlinefairbo-basic": "online_fairbo_basic",
        "online_fairbo_basic": "online_fairbo_basic",
        "onlinefairbo_basic": "online_fairbo_basic",
        "onlinefairbobasic": "online_fairbo_basic",
        "adaptivefairbo": "online_fairbo_ensemble",
        "adaptive_fairbo": "online_fairbo_ensemble",
        "onlinefairbo-ensemble": "online_fairbo_ensemble",
        "online_fairbo_ensemble": "online_fairbo_ensemble",
        "onlinefairbo_ensemble": "online_fairbo_ensemble",
        "onlinefairboensemble": "online_fairbo_ensemble",
    }
    return aliases.get(method, "online_fairbo_ensemble")


def _initial_scope_state(
    scope_manager: Optional[FairnessScopeManager],
    scope_cfg: Dict[str, Any],
    sensitive_attr: List[str],
) -> Dict[str, Any]:
    active_attributes: List[str] = []
    active_intersections: List[Tuple[str, ...]] = []
    scores: Dict[str, float] = {}
    selection_reason = "bootstrap"

    if scope_cfg.get("mode") == "manual":
        active_attributes = list(sensitive_attr)
        if scope_cfg.get("allow_intersections", True) and len(active_attributes) >= 2:
            active_intersections = [tuple(active_attributes[:2])]
        selection_reason = "manual"

    elif scope_manager is not None and getattr(scope_manager, "profiles", None):
        ranked = sorted(
            scope_manager.profiles.values(),
            key=lambda p: (
                safe_float(getattr(p, "static_score", 0.0), 0.0),
                -safe_int(getattr(p, "unique_vals", 0), 0),
            ),
            reverse=True,
        )
        active_attributes = [p.name for p in ranked[: max(1, safe_int(scope_cfg.get("max_active_attributes", 1), 1))]]
        if scope_cfg.get("allow_intersections", True) and len(active_attributes) >= 2:
            active_intersections = [tuple(active_attributes[:2])]
        scores = {p.name: safe_float(getattr(p, "static_score", 0.0), 0.0) for p in ranked}
        selection_reason = "static_bootstrap"

    primary = active_attributes[0] if active_attributes else (sensitive_attr[0] if sensitive_attr else "sensitive_attribute")

    return {
        "mode": scope_cfg.get("mode", "manual"),
        "active_attributes": active_attributes,
        "active_intersections": active_intersections,
        "primary_attribute": primary,
        "scores": scores,
        "selection_reason": selection_reason,
        "last_rescore_step": 0,
    }


def _active_sensitive_from_scope(
    xi_full: Dict[str, Any],
    all_sensitive_dict: Dict[str, Any],
    scope_state: Dict[str, Any],
) -> Dict[str, Any]:
    active_attrs = scope_state.get("active_attributes", []) if isinstance(scope_state, dict) else []
    if not active_attrs:
        return dict(all_sensitive_dict)

    out: Dict[str, Any] = {}
    for attr in active_attrs:
        if attr in all_sensitive_dict:
            out[attr] = all_sensitive_dict[attr]
        else:
            out[attr] = _normalize_value(xi_full.get(attr, None))
    return out


def _safe_scope_snapshot(scope_manager: Any) -> Dict[str, Any]:
    if scope_manager is None:
        return {}
    try:
        snap = scope_manager.snapshot()
        return snap if isinstance(snap, dict) else {}
    except Exception:
        return {}


def _build_decision_context(
    xi_full: Dict[str, Any],
    all_sensitive_dict: Dict[str, Any],
    active_sensitive_dict: Dict[str, Any],
    scope_state: Dict[str, Any],
    scope_snapshot: Dict[str, Any],
    active_mitigation: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    primary_attr = scope_state.get("primary_attribute", "sensitive_attribute") if isinstance(scope_state, dict) else "sensitive_attribute"
    active_intersections = scope_state.get("active_intersections", []) if isinstance(scope_state, dict) else []
    primary_value = active_sensitive_dict.get(primary_attr, all_sensitive_dict.get(primary_attr, "unknown"))

    intersection_values: Dict[str, Any] = {}
    if active_intersections:
        for inter in active_intersections:
            if isinstance(inter, (list, tuple)):
                key = "|".join(map(str, inter))
                intersection_values[key] = "|".join(str(all_sensitive_dict.get(a, "unknown")) for a in inter)

    mitigation = active_mitigation if isinstance(active_mitigation, dict) else {}
    mitigation_params = mitigation.get("params", {}) if isinstance(mitigation.get("params", {}), dict) else {}

    routing: Dict[str, Any] = {
        "mode": "scope_aware",
        "prediction_mode": mitigation_params.get("prediction_mode"),
        "candidate_weights": mitigation_params.get("candidate_weights", {}),
        "explore_boost": mitigation_params.get("explore_boost", 1.0),
    }

    active_scope = {
        "mode": scope_state.get("mode", "unknown"),
        "primary_attribute": primary_attr,
        "active_attributes": copy.deepcopy(scope_state.get("active_attributes", [])),
        "active_intersections": copy.deepcopy(active_intersections),
    }

    ctx = {
        "primary_attribute": primary_attr,
        "primary_group_value": primary_value,
        "all_sensitive_dict": dict(all_sensitive_dict),
        "active_sensitive_dict": dict(active_sensitive_dict),
        "active_intersections": copy.deepcopy(active_intersections),
        "intersection_values": intersection_values,
        "raw_row_sensitive_view": {k: xi_full.get(k) for k in all_sensitive_dict.keys()},
        "scope_state": copy.deepcopy(scope_state),
        "fairness_scope_snapshot": copy.deepcopy(scope_snapshot),
        "active_scope": active_scope,
        "mitigation": copy.deepcopy(mitigation),
        "routing": routing,
    }

    if mitigation_params:
        if "fairness_weight" in mitigation_params:
            ctx["fairness_weight"] = safe_float(mitigation_params.get("fairness_weight", 1.0), 1.0)
        if "performance_weight" in mitigation_params:
            ctx["performance_weight"] = safe_float(mitigation_params.get("performance_weight", 1.0), 1.0)
        if "sample_weight" in mitigation_params:
            ctx["sample_weight"] = safe_float(mitigation_params.get("sample_weight", 1.0), 1.0)

    return ctx


def _resolve_scope_weight(
    active_mitigation: Dict[str, Any],
    active_sensitive_dict: Dict[str, Any],
    decision_context: Dict[str, Any],
    fallback_primary_attr: str,
) -> float:
    if not isinstance(active_mitigation, dict):
        return 1.0

    mitigation_params = active_mitigation.get("params", {})
    mitigation_params = mitigation_params if isinstance(mitigation_params, dict) else {}

    weighting_active = bool(
        active_mitigation.get("weighting_active")
        or mitigation_params.get("weighting_active", False)
    )
    if not weighting_active:
        return 1.0

    weighting_rule = mitigation_params.get("weighting_rule", {})
    weighting_rule = weighting_rule if isinstance(weighting_rule, dict) else {}

    target_kind = str(
        weighting_rule.get("target_kind", mitigation_params.get("target_kind", "attribute"))
    ).strip().lower()

    target_attr = str(
        weighting_rule.get("target_attr", mitigation_params.get("target_attr", fallback_primary_attr))
    ).strip()

    target_group = weighting_rule.get("target_group", mitigation_params.get("target_group", None))
    target_group_key = weighting_rule.get("target_group_key", mitigation_params.get("target_group_key", None))
    target_intersection = weighting_rule.get("target_intersection", mitigation_params.get("target_intersection", None))

    default_weight = safe_float(weighting_rule.get("default_weight", 1.0), 1.0)
    target_weight = safe_float(weighting_rule.get("target_weight", 2.0), 2.0)

    if target_kind == "attribute":
        group_val = active_sensitive_dict.get(target_attr, "unknown")
        if target_group is not None and str(group_val) == str(target_group):
            return target_weight
        if target_group is None and str(group_val) == "0":
            return target_weight
        return default_weight

    if target_kind == "group_key":
        active_group_key = None
        active_sensitive_dict_ctx = decision_context.get("active_sensitive_dict", {})
        if isinstance(active_sensitive_dict_ctx, dict) and active_sensitive_dict_ctx:
            active_group_key = "|".join(str(v) for _, v in sorted(active_sensitive_dict_ctx.items()))

        if target_group_key is not None and active_group_key is not None:
            if str(active_group_key) == str(target_group_key):
                return target_weight
        return default_weight

    if target_kind == "intersection":
        intersection_values = decision_context.get("intersection_values", {})
        if isinstance(intersection_values, dict) and intersection_values:
            key = None
            if isinstance(target_intersection, (list, tuple)):
                key = "|".join(map(str, target_intersection))
            elif target_intersection is not None:
                key = str(target_intersection)

            if key is not None and key in intersection_values:
                inter_val = str(intersection_values.get(key, "unknown"))
                if target_group_key is not None and inter_val == str(target_group_key):
                    return target_weight
                if target_group is not None and inter_val == str(target_group):
                    return target_weight

        return default_weight

    return 1.0


def _sync_scope_to_controller(exp: "ExperimentState") -> None:
    primary = exp.scope_state.get("primary_attribute", exp.primary_sensitive_attr)
    exp.primary_sensitive_attr = str(primary)

    if hasattr(exp.controller, "sensitive_attr_name"):
        try:
            exp.controller.sensitive_attr_name = exp.primary_sensitive_attr
        except Exception:
            pass

    if hasattr(exp.controller, "registry") and hasattr(exp.controller.registry, "sensitive_attr_name"):
        try:
            exp.controller.registry.sensitive_attr_name = exp.primary_sensitive_attr
        except Exception:
            pass


def _sync_scope_context_to_modules(
    exp: "ExperimentState",
    active_sensitive_dict: Dict[str, Any],
) -> Dict[str, Any]:
    scope_snapshot = _safe_scope_snapshot(exp.fairness_scope_manager)

    for component_name in (
        "fairness_metric",
        "topological_engine",
        "explain_engine",
    ):
        component = getattr(exp, component_name, None)
        if component is not None and hasattr(component, "update_scope_context"):
            try:
                component.update_scope_context(
                    scope_state=copy.deepcopy(exp.scope_state),
                    fairness_scope_snapshot=copy.deepcopy(scope_snapshot),
                    active_sensitive_dict=copy.deepcopy(active_sensitive_dict),
                )
            except Exception:
                logger.debug("Failed scope context sync for %s", component_name)

    fairness_engine = getattr(exp, "fairness_engine", None)
    if fairness_engine is not None:
        if hasattr(fairness_engine, "scope_state"):
            try:
                fairness_engine.scope_state = copy.deepcopy(exp.scope_state)
                fairness_engine.fairness_scope_snapshot = copy.deepcopy(scope_snapshot)
                fairness_engine.active_sensitive_dict = copy.deepcopy(active_sensitive_dict)
            except Exception:
                logger.debug("Failed fairness_engine scope field sync")

        detector = getattr(fairness_engine, "drift_detector", None)
        if detector is not None and hasattr(detector, "update_scope_context"):
            try:
                detector.update_scope_context(
                    scope_state=copy.deepcopy(exp.scope_state),
                    fairness_scope_snapshot=copy.deepcopy(scope_snapshot),
                    active_sensitive_dict=copy.deepcopy(active_sensitive_dict),
                )
            except Exception:
                logger.debug("Failed fairness_drift scope sync")

    return scope_snapshot


# ============================================================
# CodeCarbon wrapper
# ============================================================

@dataclass
class CarbonState:
    enabled: bool = False
    tracker: Any = None
    running: bool = False

    co2_total_kg: float = 0.0
    energy_total_kwh: float = 0.0

    co2_surface_kg: List[float] = field(default_factory=list)
    energy_surface_kwh: List[float] = field(default_factory=list)

    def start_chunk(self) -> None:
        if not self.enabled or self.tracker is None or self.running:
            return
        try:
            self.tracker.start()
            self.running = True
        except Exception:
            self.running = False

    def _extract_energy_kwh_best_effort(self, tracker_obj: Any) -> float:
        fed = getattr(tracker_obj, "final_emissions_data", None)
        if fed is not None:
            e = safe_float(getattr(fed, "energy_consumed", 0.0), 0.0)
            if e > 0:
                return e

        last = getattr(tracker_obj, "_last_emissions_data", None)
        if last is not None:
            e = safe_float(getattr(last, "energy_consumed", 0.0), 0.0)
            if e > 0:
                return e

        return 0.0

    def end_chunk(self) -> None:
        if not self.enabled or self.tracker is None or not self.running:
            return
        try:
            result = self.tracker.stop()

            co2 = 0.0
            if isinstance(result, (float, int, np.floating, np.integer)):
                co2 = safe_float(result, 0.0)
            elif result is not None:
                co2 = safe_float(getattr(result, "emissions", 0.0), 0.0)

            self.co2_total_kg += co2
            bounded_append(self.co2_surface_kg, co2, max_len=MAX_HISTORY)

            e = self._extract_energy_kwh_best_effort(self.tracker)
            if e > 0:
                self.energy_total_kwh += e
                bounded_append(self.energy_surface_kwh, e, max_len=MAX_HISTORY)

        except Exception:
            pass
        finally:
            self.running = False

    def finalize(self) -> None:
        if self.running:
            self.end_chunk()


# ============================================================
# Experiment State
# ============================================================

@dataclass
class ExperimentState:
    exp_id: str

    compute_lock: threading.Lock = field(default_factory=threading.Lock)
    state_lock: threading.Lock = field(default_factory=threading.Lock)

    created_at: float = field(default_factory=_now)
    last_poll_ts: float = field(default_factory=_now)
    last_error: str = ""

    window_size: int = 500
    sample_budget: int = 0
    dataset_name: str = ""
    sensitive_attr: List[str] = field(default_factory=list)
    primary_sensitive_attr: str = "sensitive_attribute"
    target: str = "target"

    fairness_scope_config: Dict[str, Any] = field(default_factory=dict)
    scope_state: Dict[str, Any] = field(default_factory=dict)
    scope_snapshot: Dict[str, Any] = field(default_factory=dict)

    X_model: List[Dict[str, float]] = field(default_factory=list)
    X_full: List[Dict[str, Any]] = field(default_factory=list)
    y: List[int] = field(default_factory=list)
    index: int = 0

    feature_manifest: Dict[str, Any] = field(default_factory=dict)

    model: Any = field(default_factory=AutoStreamCore)
    controller: Any = None
    fairness_engine: Any = None
    fairness_metric: Any = None
    fairness_envelope: Any = None
    performance_drift: Any = None
    explain_engine: Any = None
    topological_engine: Any = None
    llm_auditor: Any = None
    fairness_scope_manager: Any = None
    config_space: Any = None
    candidate_evaluator: Any = None
    optimizer: Any = None

    llm_auditor_output: Dict[str, Any] = field(default_factory=lambda: {
        "narrative": "",
        "root_cause": "",
        "scope_summary": "",
        "mitigation_verification": "",
        "incident_report": "",
        "memory": [],
    })
    llm_is_processing: bool = False
    last_llm_trigger_step: int = 0
    last_response_cache: Dict[str, Any] = field(default_factory=dict)

    active_mitigation_state: Dict[str, Any] = field(default_factory=dict)
    decision_context: Dict[str, Any] = field(default_factory=dict)

    accuracy_metric: Any = field(default_factory=metrics.Accuracy)
    accuracy_over_time: List[float] = field(default_factory=list)

    fairness_over_time: List[float] = field(default_factory=list)
    cumulative_fairness_over_time: List[float] = field(default_factory=list)
    fairness_stability_surface: List[float] = field(default_factory=list)
    equal_opportunity_over_time: List[float] = field(default_factory=list)
    equalized_odds_over_time: List[float] = field(default_factory=list)

    # New SFM surfaces
    fm_raw_over_time: List[float] = field(default_factory=list)
    fm_smooth_over_time: List[float] = field(default_factory=list)
    fm_drift_over_time: List[float] = field(default_factory=list)
    fm_uncertainty_over_time: List[float] = field(default_factory=list)

    # New DFE surfaces
    envelope_lower_over_time: List[float] = field(default_factory=list)
    envelope_upper_over_time: List[float] = field(default_factory=list)
    envelope_width_over_time: List[float] = field(default_factory=list)
    envelope_severity_over_time: List[float] = field(default_factory=list)
    envelope_violation_over_time: List[int] = field(default_factory=list)
    envelope_pressure_over_time: List[float] = field(default_factory=list)

    explanation_drift_surface: List[float] = field(default_factory=list)

    fairness_drift_event_points: List[int] = field(default_factory=list)
    warning_event_points: List[int] = field(default_factory=list)
    perf_drift_event_points: List[int] = field(default_factory=list)
    perf_drift_monitor_event_points: List[int] = field(default_factory=list)

    fairness_drift_state_over_time: List[int] = field(default_factory=list)
    warning_state_over_time: List[int] = field(default_factory=list)
    perf_drift_state_over_time: List[int] = field(default_factory=list)
    perf_drift_monitor_state_over_time: List[int] = field(default_factory=list)

    # New envelope event points
    envelope_warning_event_points: List[int] = field(default_factory=list)
    envelope_violation_event_points: List[int] = field(default_factory=list)
    envelope_tighten_event_points: List[int] = field(default_factory=list)
    envelope_relax_event_points: List[int] = field(default_factory=list)

    last_warning_state: bool = False
    perf_drift_latch: int = 0
    perf_drift_latch_steps: int = PERF_DRIFT_LATCH_STEPS
    perf_drift_min_separation: int = 0
    last_perf_drift_step: int = -10**9
    performance_drift_monitor: Any = None
    perf_drift_monitor_latch: int = 0
    perf_drift_monitor_latch_steps: int = 0
    perf_drift_monitor_min_separation: int = 0
    last_perf_drift_monitor_step: int = -10**9
    perf_drift_monitor_short_window: int = 0
    perf_drift_monitor_long_window: int = 0
    perf_drift_monitor_drop_threshold: float = 0.0
    perf_drift_monitor_short_errors: deque = field(default_factory=deque)
    perf_drift_monitor_long_errors: deque = field(default_factory=deque)

    rise_residuals_by_group_raw: Dict[str, List[float]] = field(default_factory=dict)

    latest_mapper_graph: Optional[Dict[str, Any]] = None
    mapper_config: Dict[str, Any] = field(default_factory=dict)

    latest_fairness_snapshot: Dict[str, Any] = field(default_factory=dict)
    latest_metrics_snapshot: Dict[str, Any] = field(default_factory=dict)
    latest_explainability_snapshot: Dict[str, Any] = field(default_factory=dict)
    latest_topology_snapshot: Dict[str, Any] = field(default_factory=dict)
    latest_envelope_snapshot: Dict[str, Any] = field(default_factory=dict)
    latest_sfm_state: Dict[str, Any] = field(default_factory=dict)
    config_space_summary: Dict[str, Any] = field(default_factory=dict)
    config_space_manifest: Dict[str, Any] = field(default_factory=dict)
    candidate_evaluator_config: Dict[str, Any] = field(default_factory=dict)
    optimizer_config: Dict[str, Any] = field(default_factory=dict)
    optimizer_runtime_config: Dict[str, Any] = field(default_factory=dict)
    latest_optimizer_batch: Dict[str, Any] = field(default_factory=dict)
    latest_candidate_evaluations: List[Dict[str, Any]] = field(default_factory=list)
    optimizer_candidate_history: List[Dict[str, Any]] = field(default_factory=list)
    latest_live_handoff: Dict[str, Any] = field(default_factory=dict)
    latest_optimizer_error: str = ""
    last_optimizer_eval_step: int = 0
    static_benchmark_config: Dict[str, Any] = field(default_factory=dict)
    static_benchmark_results: Dict[str, Any] = field(default_factory=dict)
    static_benchmark_error: str = ""
    static_benchmark_status: str = "disabled"
    static_benchmark_started_at: float = 0.0
    static_benchmark_finished_at: float = 0.0

    y_prob_history: List[float] = field(default_factory=list)
    y_prob_debug_snapshot: Dict[str, Any] = field(default_factory=dict)

    carbon: CarbonState = field(default_factory=CarbonState)
    start_time: float = field(default_factory=_now)


# ============================================================
# Payload validation + experiment creation
# ============================================================

def _validate_start_payload(data: Dict[str, Any]) -> Tuple[bool, str, Dict[str, Any]]:
    if not isinstance(data, dict):
        return False, "payload_not_dict", {}

    dataset = data.get("dataset", None)
    if dataset is None:
        return False, "missing_dataset", {}
    if not isinstance(dataset, list) or len(dataset) == 0:
        return False, "dataset_not_list_or_empty", {
            "type": str(type(dataset)),
            "len": (len(dataset) if isinstance(dataset, list) else None),
        }

    if len(dataset) > MAX_DATASET_ROWS:
        return False, "dataset_too_large", {"max_rows": MAX_DATASET_ROWS, "rows": len(dataset)}

    if not all(isinstance(r, dict) for r in dataset[: min(50, len(dataset))]):
        return False, "dataset_rows_not_dict", {}

    target = data.get("target", None)
    if target is None or str(target).strip() == "":
        return False, "missing_target", {}

    fcfg = data.get("fairness_scope_config", None)
    if fcfg is not None and not isinstance(fcfg, dict):
        return False, "bad_fairness_scope_config", {"type": str(type(fcfg))}

    return True, "ok", {}


def _infer_static_benchmark_dataset_name(
    df: pd.DataFrame,
    *,
    target: str,
    sensitive_attr: List[str],
    explicit_name: Any = None,
) -> str:
    explicit = safe_str(explicit_name, "").strip().lower()
    if explicit in {"adult", "german", "compas"}:
        return explicit

    cols = {str(col).strip().lower() for col in df.columns}
    sens = {str(col).strip().lower() for col in sensitive_attr}
    target_l = str(target).strip().lower()

    if {"income", "sex"}.issubset(cols) and (target_l == "income" or "sex" in sens):
        return "adult"
    if (
        {"sex", "class"}.issubset(cols)
        or {"gender", "class"}.issubset(cols)
        or {"sex", "credit_risk"}.issubset(cols)
        or {"gender", "credit_risk"}.issubset(cols)
    ):
        return "german"
    if (
        {"race", "two_year_recid"}.issubset(cols)
        or {"ethnicity", "two_year_recid"}.issubset(cols)
        or {"race", "is_recid"}.issubset(cols)
    ):
        return "compas"
    return ""


def _resolve_static_benchmark_config(raw_cfg: Any, *, dataset_name: str) -> Dict[str, Any]:
    cfg = dict(raw_cfg or {}) if isinstance(raw_cfg, dict) else {}
    methods = cfg.get("methods", ["Streaming_Aware_FairBO"])
    if not isinstance(methods, list) or not methods:
        methods = ["Streaming_Aware_FairBO"]

    normalized_methods: List[str] = []
    for method in methods:
        name = safe_str(method, "").strip()
        if not name:
            continue
        upper = name.upper()
        if upper in {"STREAMING_AWARE_FAIRBO", "STREAMING-AWARE-FAIRBO", "FAIRBO", "FAIR-BO"}:
            normalized_methods.append("Streaming_Aware_FairBO")

    if not normalized_methods:
        normalized_methods = ["Streaming_Aware_FairBO"]

    enabled = safe_bool(cfg.get("enabled", False), False)
    return {
        "enabled": bool(enabled),
        "dataset_name": str(dataset_name),
        "methods": normalized_methods,
        "budget": max(3, safe_int(cfg.get("budget", DEFAULT_STATIC_BENCHMARK_BUDGET), DEFAULT_STATIC_BENCHMARK_BUDGET)),
        "fairness_threshold": float(np.clip(safe_float(cfg.get("fairness_threshold", 0.05), 0.05), 0.0, 1.0)),
        "random_state": max(0, safe_int(cfg.get("random_state", 42), 42)),
        "initial_design": max(2, safe_int(cfg.get("initial_design", 5), 5)),
        "candidate_pool_size": max(16, safe_int(cfg.get("candidate_pool_size", DEFAULT_STATIC_BENCHMARK_POOL), DEFAULT_STATIC_BENCHMARK_POOL)),
        "history_window": max(0, safe_int(cfg.get("history_window", 0), 0)),
        "max_train_samples": max(0, safe_int(cfg.get("max_train_samples", 0), 0)),
        "available_datasets": [],
    }


def _basic_live_handoff_policy(dataset_name: str, live_baseline: float) -> Dict[str, float]:
    dataset_key = str(dataset_name or "").strip().lower()
    baseline = safe_float(live_baseline, 0.0)

    if dataset_key == "german":
        if baseline >= 0.80:
            improvement_margin = 0.0025
        elif baseline >= 0.75:
            improvement_margin = 0.0035
        else:
            improvement_margin = 0.0050
        minimum_adopt_accuracy = 0.79 if baseline >= 0.78 else 0.77
        stable_freeze_accuracy = 0.79
    else:
        improvement_margin = 0.020 if baseline >= 0.80 else 0.010
        minimum_adopt_accuracy = 0.90 if baseline >= 0.88 else 0.85
        stable_freeze_accuracy = 0.85

    return {
        "improvement_margin": float(improvement_margin),
        "minimum_adopt_accuracy": float(minimum_adopt_accuracy),
        "stable_freeze_accuracy": float(stable_freeze_accuracy),
    }


def _german_live_handoff_score(
    *,
    accuracy: float,
    balanced_accuracy: float,
    constrained_utility: float,
) -> float:
    acc = safe_float(accuracy, 0.0)
    bal = safe_float(balanced_accuracy, acc)
    util = safe_float(constrained_utility, acc)
    return float((0.50 * acc) + (0.35 * bal) + (0.15 * util))


def _summarize_static_method_records(records: List[Dict[str, Any]]) -> Dict[str, Any]:
    if not records:
        return {}
    last = dict(records[-1])
    feasible_records = [r for r in records if bool(r.get("feasible", False))]
    best_feasible_error = None
    best_feasible_dsp = None
    best_feasible_accuracy = None
    if feasible_records:
        best_feasible_error = min(safe_float(r.get("validation_error", 1.0), 1.0) for r in feasible_records)
        best_feasible_dsp = min(safe_float(r.get("dsp", 1.0), 1.0) for r in feasible_records)
        best_feasible_accuracy = 1.0 - float(best_feasible_error)

    agg = {
        "privileged": {"count": 0.0, "pred_pos": 0.0, "tp": 0.0, "fn": 0.0, "fp": 0.0, "tn": 0.0},
        "unprivileged": {"count": 0.0, "pred_pos": 0.0, "tp": 0.0, "fn": 0.0, "fp": 0.0, "tn": 0.0},
    }
    for record in records:
        meta = dict(record.get("metadata", {}) or {}) if isinstance(record, dict) else {}
        group_conf = dict(meta.get("group_confusion", {}) or {}) if isinstance(meta, dict) else {}
        for side in ("privileged", "unprivileged"):
            side_block = dict(group_conf.get(side, {}) or {}) if isinstance(group_conf, dict) else {}
            for key in ("count", "pred_pos", "tp", "fn", "fp", "tn"):
                agg[side][key] += safe_float(side_block.get(key, 0.0), 0.0)

    def _rate(num: float, den: float) -> float:
        return float(num / den) if den > 0 else 0.0

    priv = agg["privileged"]
    unpriv = agg["unprivileged"]
    priv_rate = _rate(priv["pred_pos"], priv["count"])
    unpriv_rate = _rate(unpriv["pred_pos"], unpriv["count"])
    overall_dsp = abs(priv_rate - unpriv_rate)
    priv_tpr = _rate(priv["tp"], priv["tp"] + priv["fn"])
    unpriv_tpr = _rate(unpriv["tp"], unpriv["tp"] + unpriv["fn"])
    priv_fpr = _rate(priv["fp"], priv["fp"] + priv["tn"])
    unpriv_fpr = _rate(unpriv["fp"], unpriv["fp"] + unpriv["tn"])
    overall_eo = abs(priv_tpr - unpriv_tpr)
    overall_eodds = max(overall_eo, abs(priv_fpr - unpriv_fpr))

    return {
        "iterations": len(records),
        "feasible_rate": float(sum(1 for r in records if bool(r.get("feasible", False))) / max(1, len(records))),
        "final_validation_accuracy": safe_float(last.get("validation_accuracy", 0.0), 0.0),
        "final_validation_error": safe_float(last.get("validation_error", 1.0), 1.0),
        "final_dsp": safe_float(last.get("dsp", 1.0), 1.0),
        "final_equal_opportunity": safe_float(last.get("equal_opportunity", 0.0), 0.0),
        "final_equalized_odds": safe_float(last.get("equalized_odds", 0.0), 0.0),
        "overall_dsp": float(overall_dsp),
        "overall_equal_opportunity": float(overall_eo),
        "overall_equalized_odds": float(overall_eodds),
        "best_feasible_accuracy": None if best_feasible_accuracy is None else float(best_feasible_accuracy),
        "best_feasible_error": None if best_feasible_error is None else float(best_feasible_error),
        "best_feasible_dsp": None if best_feasible_dsp is None else float(best_feasible_dsp),
        "total_perf_drift_events": safe_int(last.get("total_perf_drift_events", 0), 0),
        "total_fairness_drift_events": safe_int(last.get("total_fairness_drift_events", 0), 0),
    }


def _run_static_benchmark_suite(
    df: pd.DataFrame,
    *,
    dataset_name: str,
    config: Dict[str, Any],
) -> Tuple[Dict[str, Any], str]:
    if not bool(config.get("enabled", False)):
        return {}, ""

    target_column = safe_str(config.get("target_column", ""), "").strip()
    sensitive_column = safe_str(config.get("sensitive_column", ""), "").strip()
    if not target_column or not sensitive_column:
        return {}, "Streaming benchmark requires target_column and sensitive_column"

    window_size = max(20, safe_int(config.get("window_size", 200), 200))
    sample_budget = max(0, safe_int(config.get("sample_budget", len(df)), len(df)))
    spec = StreamingBenchmarkSpec(
        dataset_name=str(dataset_name),
        target_column=target_column,
        sensitive_column=sensitive_column,
        model_family="random_forest",
        fairness_metric="dsp",
        fairness_threshold=safe_float(config.get("fairness_threshold", 0.05), 0.05),
        window_size=window_size,
        min_history=max(window_size, int(2 * window_size)),
        history_window=max(0, safe_int(config.get("history_window", 0), 0)),
        step_size=window_size,
        sample_budget=sample_budget,
        max_train_samples=max(0, safe_int(config.get("max_train_samples", 0), 0)),
        random_state=safe_int(config.get("random_state", 42), 42),
        budget=safe_int(config.get("budget", DEFAULT_STATIC_BENCHMARK_BUDGET), DEFAULT_STATIC_BENCHMARK_BUDGET),
    )

    methods = list(config.get("methods", ["Streaming_Aware_FairBO"]))
    initial_design = safe_int(config.get("initial_design", 5), 5)
    candidate_pool_size = safe_int(config.get("candidate_pool_size", DEFAULT_STATIC_BENCHMARK_POOL), DEFAULT_STATIC_BENCHMARK_POOL)

    runners = {
        "Streaming_Aware_FairBO": lambda: run_streaming_fairbo_baseline(
            spec,
            df=df,
            initial_design=initial_design,
            candidate_pool_size=candidate_pool_size,
        ),
    }

    methods_payload: Dict[str, Any] = {}
    method_errors: Dict[str, str] = {}
    for method in methods:
        runner = runners.get(method)
        if runner is None:
            continue
        try:
            artifact = runner()
            artifact_dict = artifact.to_dict()
            records = artifact_dict.get("records", []) if isinstance(artifact_dict, dict) else []
            methods_payload[method] = {
                **artifact_dict,
                "summary": _summarize_static_method_records(records if isinstance(records, list) else []),
            }
        except Exception as exc:
            method_errors[method] = f"{type(exc).__name__}: {exc}"

    payload = {
        "enabled": True,
        "dataset_name": str(dataset_name),
        "benchmark_kind": "streaming_aware",
        "spec": spec.to_dict(),
        "methods": methods_payload,
        "method_order": list(methods_payload.keys()),
        "method_errors": method_errors,
    }
    overall_error = ""
    if not methods_payload and method_errors:
        overall_error = "; ".join(f"{key}: {value}" for key, value in sorted(method_errors.items()))
    return payload, overall_error


def _start_static_benchmark_async(
    exp_id: str,
    *,
    df: pd.DataFrame,
    dataset_name: str,
    config: Dict[str, Any],
) -> None:
    df_local = df.copy(deep=True)
    cfg_local = dict(config)

    def _worker() -> None:
        with EXPERIMENTS_LOCK:
            exp = EXPERIMENTS.get(exp_id)
        if exp is None:
            return

        with exp.state_lock:
            exp.static_benchmark_status = "running"
            exp.static_benchmark_started_at = _now()
            exp.static_benchmark_finished_at = 0.0
            exp.static_benchmark_error = ""

        try:
            payload, error = _run_static_benchmark_suite(
                df_local,
                dataset_name=dataset_name,
                config=cfg_local,
            )
            status = "completed" if payload else ("error" if error else "completed")
        except Exception as exc:
            payload = {}
            error = f"{type(exc).__name__}: {exc}"
            status = "error"

        with EXPERIMENTS_LOCK:
            exp = EXPERIMENTS.get(exp_id)
        if exp is None:
            return

        with exp.state_lock:
            exp.static_benchmark_results = make_json_safe(payload)
            exp.static_benchmark_error = str(error or "")
            exp.static_benchmark_status = str(status)
            exp.static_benchmark_finished_at = _now()

        logger.info(
            "Static benchmark %s for experiment %s finished with status=%s methods=%s error=%s",
            dataset_name,
            exp_id,
            status,
            list((payload or {}).get("methods", {}).keys()) if isinstance(payload, dict) else [],
            error,
        )

    thread = threading.Thread(
        target=_worker,
        name=f"static-benchmark-{exp_id}",
        daemon=True,
    )
    thread.start()


def create_experiment(data: Dict[str, Any]) -> Tuple[bool, str, Optional[str], Optional[Dict[str, Any]]]:
    ok, reason, details = _validate_start_payload(data)
    if not ok:
        return False, reason, None, details

    exp_id = str(uuid.uuid4())

    try:
        df = pd.DataFrame(data["dataset"])
        if df.shape[1] > MAX_DATASET_COLS:
            return False, "too_many_columns", None, {"max_cols": MAX_DATASET_COLS, "cols": int(df.shape[1])}

        target = str(data["target"])
        if target not in df.columns:
            return False, "target_not_in_dataset", None, {"target": target, "columns": list(map(str, df.columns))[:50]}

        sensitive_attr = data.get("sensitive_attr", [])
        if sensitive_attr and not isinstance(sensitive_attr, list):
            sensitive_attr = [sensitive_attr]
        sensitive_attr = [str(s) for s in sensitive_attr] if sensitive_attr else []

        window_size = max(10, safe_int(data.get("window_size", 500), 500))
        sample_budget = safe_int(data.get("sample_budget", len(df)), len(df))
        dataset_name = _infer_static_benchmark_dataset_name(
            df,
            target=target,
            sensitive_attr=sensitive_attr,
            explicit_name=data.get("dataset_name"),
        )
        static_benchmark_config = _resolve_static_benchmark_config(
            data.get("baseline_benchmark_config", {}),
            dataset_name=dataset_name,
        )
        if not safe_str(static_benchmark_config.get("dataset_name", ""), "").strip():
            static_benchmark_config["dataset_name"] = safe_str(data.get("dataset_name", ""), "").strip() or "uploaded_stream"
        static_benchmark_config["target_column"] = str(target)
        static_benchmark_config["sensitive_column"] = str(sensitive_attr[0]) if sensitive_attr else ""
        static_benchmark_config["window_size"] = int(window_size)
        static_benchmark_config["sample_budget"] = int(sample_budget)
        if safe_int(static_benchmark_config.get("history_window", 0), 0) <= 0:
            static_benchmark_config["history_window"] = int(max(window_size * 2, 200))
        if safe_int(static_benchmark_config.get("max_train_samples", 0), 0) <= 0:
            hist = safe_int(static_benchmark_config.get("history_window", 0), 0)
            static_benchmark_config["max_train_samples"] = int(max(55, round(0.155 * max(hist, window_size))))

        fairness_scope_config = _resolve_scope_config(
            data.get("fairness_scope_config", {}),
            sensitive_attr=sensitive_attr,
            window_size=window_size,
        )
        optimizer_runtime_config = _resolve_optimizer_runtime_config(
            data.get("osmac_runtime_config", data.get("optimizer_runtime_config", {})),
            window_size=window_size,
        )

        config_space_payload = dict(data.get("config_space_config", {}) or {})
        candidate_eval_payload = dict(data.get("candidate_evaluator_config", {}) or {})
        optimizer_payload = dict(
            data.get("osmac_optimizer_config", data.get("optimizer_config", {})) or {}
        )
        streaming_optimizer_method = _resolve_streaming_optimizer_method(
            data.get("streaming_optimizer_method", optimizer_payload.get("streaming_method", "online_fairbo_ensemble"))
        )
        optimizer_payload["streaming_method"] = streaming_optimizer_method

        if streaming_optimizer_method == "online_fairbo_basic":
            optimizer_payload.setdefault("acquisition_mode", "cei_static")
            optimizer_payload.setdefault("objective_mode", "maximize_accuracy")
            optimizer_payload.setdefault("incumbent_metric", "constrained_utility")
            candidate_eval_payload.setdefault("optimizer_constraint_target", "static_threshold")
            candidate_eval_payload.setdefault("scoring_burn_in_fraction", 0.20)
            candidate_eval_payload.setdefault("scoring_burn_in_min_samples", max(20, int(window_size // 8)))
            optimizer_payload.setdefault("feasibility_weight", 1.15)
            optimizer_payload.setdefault("risk_penalty_weight", 0.45)
            optimizer_payload.setdefault("envelope_violation_penalty", 0.50)
            optimizer_payload.setdefault("recent_feasible_bonus", 0.18)
            optimizer_payload.setdefault("feasible_priority_bonus", 1.35)
            optimizer_payload.setdefault("infeasible_priority_penalty", 0.40)
            optimizer_runtime_config["step_frequency"] = max(75, int(window_size // 2))
            optimizer_runtime_config["min_samples"] = max(OSMAC_MIN_WINDOW, int(window_size // 3))
            optimizer_runtime_config["include_incumbent"] = True
            optimizer_runtime_config["force_exploration_on_violation"] = False
            autostream_model_config = {
                "prediction_mode": "best",
                "ensemble_size": 1,
                "exploration_window": max(300, int(2.0 * window_size)),
                "budget": 16,
                "drift_delta": 0.001,
                "drift_cooldown_steps": max(250, int(1.5 * window_size)),
                "drift_accuracy_guard": 0.865,
                "scheduled_cooldown_after_drift_steps": max(150, int(window_size)),
                "scheduled_accuracy_guard": 0.835,
                "search_random_fraction": 0.25,
                "search_allowed_models": [
                    "srp_classifier",
                    "gaussian_nb",
                    "logistic_regression",
                ],
                "search_allowed_preprocessors": [
                    "standard_scaler",
                ],
                "search_allowed_feature_selectors": [],
                "search_allow_feature_selectors": False,
                "fairness_weight": 0.0,
                "feasibility_weight": 0.0,
                "uncertainty_weight": 0.0,
                "topology_weight": 0.0,
                "fairness_penalty_mode": "subtract",
            }
        else:
            optimizer_payload.setdefault("acquisition_mode", "cei_envelope")
            optimizer_payload.setdefault("objective_mode", "maximize_accuracy")
            optimizer_payload.setdefault("incumbent_metric", "constrained_utility")
            candidate_eval_payload.setdefault("optimizer_constraint_target", "envelope")
            candidate_eval_payload.setdefault("scoring_burn_in_fraction", 0.15)
            candidate_eval_payload.setdefault("scoring_burn_in_min_samples", max(15, int(window_size // 10)))
            optimizer_payload.setdefault("feasibility_weight", 1.90)
            optimizer_payload.setdefault("risk_penalty_weight", 0.95)
            optimizer_payload.setdefault("envelope_violation_penalty", 1.10)
            optimizer_payload.setdefault("recent_feasible_bonus", 0.35)
            optimizer_payload.setdefault("feasible_priority_bonus", 2.30)
            optimizer_payload.setdefault("infeasible_priority_penalty", 0.90)
            optimizer_runtime_config["step_frequency"] = max(50, int(window_size // 2))
            optimizer_runtime_config["include_incumbent"] = True
            optimizer_runtime_config["force_exploration_on_violation"] = True
            autostream_model_config = {
                "prediction_mode": "ensemble",
                "ensemble_size": 3,
                "exploration_window": max(400, 3 * int(window_size)),
                "budget": 12,
                "fairness_weight": 0.12,
                "feasibility_weight": 0.22,
                "uncertainty_weight": 0.08,
                "topology_weight": 0.03,
                "fairness_penalty_mode": "subtract",
            }

        if "static_fairness_threshold" not in optimizer_payload:
            optimizer_payload["static_fairness_threshold"] = safe_float(
                fairness_scope_config.get("fairness_threshold", 0.10),
                0.10,
            )
        if "feasibility_weight" not in optimizer_payload:
            optimizer_payload["feasibility_weight"] = 1.5
        if "risk_penalty_weight" not in optimizer_payload:
            optimizer_payload["risk_penalty_weight"] = 0.85
        if "envelope_violation_penalty" not in optimizer_payload:
            optimizer_payload["envelope_violation_penalty"] = 1.00
        if "recent_feasible_bonus" not in optimizer_payload:
            optimizer_payload["recent_feasible_bonus"] = 0.25
        if "feasible_priority_bonus" not in optimizer_payload:
            optimizer_payload["feasible_priority_bonus"] = 2.0
        if "infeasible_priority_penalty" not in optimizer_payload:
            optimizer_payload["infeasible_priority_penalty"] = 1.0
        if "default_envelope_upper" not in candidate_eval_payload:
            candidate_eval_payload["default_envelope_upper"] = safe_float(
                optimizer_payload.get("static_fairness_threshold", 0.05),
                0.05,
            )
        if "optimizer_constraint_threshold" not in candidate_eval_payload:
            candidate_eval_payload["optimizer_constraint_threshold"] = safe_float(
                optimizer_payload.get("static_fairness_threshold", 0.10),
                0.10,
            )
        if dataset_name == "german":
            config_space_payload.setdefault(
                "allowed_models",
                ["logistic_regression", "gaussian_nb", "srp_classifier"],
            )
            config_space_payload.setdefault(
                "allowed_preprocessors",
                ["standard_scaler"],
            )
            config_space_payload.setdefault(
                "allowed_feature_selectors",
                ["none"],
            )
            config_space_payload.setdefault("enable_class_weight_candidates", True)
            config_space_payload.setdefault(
                "class_weight_profile_choices",
                ["none", "balanced_sqrt"],
            )
            candidate_eval_payload.setdefault("enable_threshold_selection", True)
            candidate_eval_payload.setdefault("threshold_selection_min", 0.35)
            candidate_eval_payload.setdefault("threshold_selection_max", 0.75)
            candidate_eval_payload.setdefault("threshold_selection_step", 0.025)
            candidate_eval_payload.setdefault("threshold_selection_ddsp_weight", 1.35)
            candidate_eval_payload.setdefault("threshold_selection_eopp_weight", 0.45)
            candidate_eval_payload.setdefault("threshold_selection_eodds_weight", 0.65)
            candidate_eval_payload.setdefault("threshold_selection_center_penalty", 0.04)
            candidate_eval_payload.setdefault("class_weight_cap", 3.5)
            candidate_eval_payload.setdefault("class_weight_warmup_samples", max(10, int(window_size // 12)))
            autostream_model_config["search_allowed_models"] = [
                "logistic_regression",
                "gaussian_nb",
                "srp_classifier",
            ]
            autostream_model_config["search_allowed_preprocessors"] = [
                "standard_scaler",
            ]
            autostream_model_config["search_allowed_feature_selectors"] = []
            autostream_model_config["search_allow_feature_selectors"] = False
            autostream_model_config["drift_cooldown_steps"] = max(
                safe_int(autostream_model_config.get("drift_cooldown_steps", 0), 0),
                max(900, 4 * int(window_size)),
            )
            autostream_model_config["scheduled_cooldown_after_drift_steps"] = max(
                safe_int(autostream_model_config.get("scheduled_cooldown_after_drift_steps", 0), 0),
                max(700, 3 * int(window_size)),
            )
            autostream_model_config["drift_accuracy_guard"] = max(
                safe_float(autostream_model_config.get("drift_accuracy_guard", 0.0), 0.0),
                0.80,
            )
            autostream_model_config["scheduled_accuracy_guard"] = max(
                safe_float(autostream_model_config.get("scheduled_accuracy_guard", 0.0), 0.0),
                0.79,
            )
            autostream_model_config["search_random_fraction"] = min(
                safe_float(autostream_model_config.get("search_random_fraction", 0.20), 0.20),
                0.12,
            )
            if streaming_optimizer_method == "online_fairbo_ensemble":
                autostream_model_config["prediction_mode"] = "best"
                autostream_model_config["ensemble_size"] = 1
                autostream_model_config["budget"] = max(10, safe_int(autostream_model_config.get("budget", 12), 12))
                autostream_model_config["search_random_fraction"] = 0.20

        primary_sensitive_attr = sensitive_attr[0] if sensitive_attr else "sensitive_attribute"

        mapper_in = data.get("mapper_config", {})
        if not isinstance(mapper_in, dict):
            mapper_in = {}
        mapper_config = {
            "cover_intervals": max(2, min(24, safe_int(mapper_in.get("cover_intervals", 8), 8))),
            "cover_overlap": float(np.clip(safe_float(mapper_in.get("cover_overlap", 0.35), 0.35), 0.0, 0.9)),
            "min_samples": max(2, min(25, safe_int(mapper_in.get("min_samples", 8), 8))),
            "edge_jaccard_threshold": float(np.clip(safe_float(mapper_in.get("edge_jaccard_threshold", 0.06), 0.06), 0.0, 0.5)),
            "eps_scale": max(0.05, min(5.0, safe_float(mapper_in.get("eps_scale", 1.25), 1.25))),
            "eps_floor": max(1e-6, min(5.0, safe_float(mapper_in.get("eps_floor", 0.25), 0.25))),
            "knn_k": max(2, min(64, safe_int(mapper_in.get("knn_k", 12), 12))),
            "lens_name": str(mapper_in.get("lens_name", "prob")).lower().strip(),
            "min_verified": max(10, min(max(10, window_size), safe_int(mapper_in.get("min_verified", 60), 60))),
            "min_group_support": max(1, min(100, safe_int(mapper_in.get("min_group_support", 12), 12))),
        }
        if mapper_config["lens_name"] not in {"prob", "residual"}:
            mapper_config["lens_name"] = "prob"

        scope_manager = FairnessScopeManager(
            target_col=target,
            approved_candidates=fairness_scope_config.get("approved_candidates", []),
            blocked_candidates=fairness_scope_config.get("blocked_candidates", []),
            fairness_threshold=fairness_scope_config.get("fairness_threshold", 0.05),
            window_size=fairness_scope_config.get("window_size", window_size),
            max_active_attributes=fairness_scope_config.get("max_active_attributes", 2),
            allow_intersections=fairness_scope_config.get("allow_intersections", True),
        )

        try:
            scope_manager.fit(df)
        except Exception as e:
            logger.warning("FairnessScopeManager.fit failed during init: %s", e)

        ordered_sensitive_cols: List[str] = []

        for c in sensitive_attr:
            if c not in ordered_sensitive_cols:
                ordered_sensitive_cols.append(c)

        for c in fairness_scope_config.get("approved_candidates", []):
            if c not in ordered_sensitive_cols:
                ordered_sensitive_cols.append(c)

        if getattr(scope_manager, "profiles", None):
            for c in scope_manager.profiles.keys():
                if c not in ordered_sensitive_cols:
                    ordered_sensitive_cols.append(c)

        records = df.to_dict(orient="records")

        y: List[int] = []
        X_full: List[Dict[str, Any]] = []

        for row in records:
            yi = _coerce_binary_target(row.get(target, 0))
            y.append(int(yi))
            X_full.append(dict(row))

        X_model, feature_manifest = _build_encoded_feature_matrix(
            df=df,
            target_col=target,
            sensitive_cols=ordered_sensitive_cols,
        )

        config_space_settings = _build_dataclass_config(ConfigSpaceSettings, config_space_payload)
        candidate_evaluator_cfg = _build_dataclass_config(CandidateEvaluatorConfig, candidate_eval_payload)
        osmac_cfg = _build_dataclass_config(OSMACOptimizerConfig, optimizer_payload)

        sample_budget = max(0, min(int(sample_budget), len(X_model)))

        exp = ExperimentState(
            exp_id=exp_id,
            window_size=window_size,
            sample_budget=sample_budget,
            dataset_name=dataset_name,
            sensitive_attr=ordered_sensitive_cols,
            primary_sensitive_attr=primary_sensitive_attr,
            target=target,
            X_model=X_model,
            X_full=X_full,
            y=y,
            mapper_config=mapper_config,
            fairness_scope_config=fairness_scope_config,
            feature_manifest=feature_manifest,
        )

        exp.model = AutoStreamCore(**autostream_model_config)
        exp.controller = AdaptiveFairnessController(
            sensitive_attr_name=primary_sensitive_attr
        )
        exp.config_space = StreamingConfigSpace(settings=config_space_settings)
        exp.candidate_evaluator = CandidateEvaluator(
            config_space=exp.config_space,
            config=candidate_evaluator_cfg,
        )
        exp.optimizer = OSMACOptimizer(
            config_space=exp.config_space,
            config=osmac_cfg,
        )

        fairness_engine_kwargs = {
            "window_size": window_size,
        }
        if streaming_optimizer_method == "online_fairbo_ensemble":
            fairness_engine_kwargs.update(
                {
                    "drift_delta": 0.004,
                    "drift_warning_threshold": 0.009,
                    "drift_threshold": 0.018,
                    "drift_rolling_window": max(18, min(30, window_size // 2)),
                    "drift_min_separation": max(15, min(30, window_size // 3)),
                    "drift_hysteresis": 0.003,
                    "drift_warmup_steps": max(25, min(50, window_size // 4)),
                    "drift_latch_steps": 6,
                    "drift_promote_warning_to_drift": True,
                    "drift_warning_promotion_ratio": 0.8,
                    "drift_direct_promotion_ratio": 0.95,
                }
            )
        else:
            fairness_engine_kwargs.update(
                {
                    "drift_delta": 0.006,
                    "drift_warning_threshold": 0.015,
                    "drift_threshold": 0.030,
                    "drift_rolling_window": max(24, min(48, window_size // 2)),
                    "drift_min_separation": max(80, int(window_size)),
                    "drift_hysteresis": 0.006,
                    "drift_warmup_steps": max(60, min(120, window_size // 2)),
                    "drift_latch_steps": 5,
                    "drift_promote_warning_to_drift": True,
                    "drift_warning_promotion_ratio": 0.90,
                    "drift_direct_promotion_ratio": 0.98,
                    "drift_adaptive_thresholds": True,
                    "drift_adapt_window": max(100, 2 * window_size),
                    "drift_adapt_q_warning": 0.88,
                    "drift_adapt_q_drift": 0.95,
                    "drift_adapt_floor_warning": 0.010,
                    "drift_adapt_floor_drift": 0.020,
                }
            )
        fairness_engine_cfg = FairnessEngineConfig(**fairness_engine_kwargs)
        exp.fairness_engine = StreamingFairnessEngine(config=fairness_engine_cfg)
        exp.fairness_metric = StreamingFairnessMetric(window_size=window_size, history_size=window_size)
        exp.fairness_envelope = DynamicFairnessEnvelope(config=_build_envelope_config(fairness_scope_config))
        perf_drift_delta = PERF_DRIFT_DELTA
        exp.perf_drift_latch_steps = PERF_DRIFT_LATCH_STEPS
        exp.perf_drift_min_separation = max(30, window_size // 2)
        perf_drift_monitor_delta = max(0.004, PERF_DRIFT_DELTA)
        exp.perf_drift_monitor_latch_steps = max(12, window_size // 8)
        exp.perf_drift_monitor_min_separation = max(60, window_size // 2)
        exp.perf_drift_monitor_short_window = max(40, window_size // 2)
        exp.perf_drift_monitor_long_window = max(160, 2 * window_size)
        exp.perf_drift_monitor_drop_threshold = 0.012
        if streaming_optimizer_method == "online_fairbo_basic":
            perf_drift_delta = 0.0008
            exp.perf_drift_latch_steps = max(20, window_size // 6)
            exp.perf_drift_min_separation = max(120, window_size)
            # Monitor-only detector: more sensitive than the operational path, but it never
            # feeds the live controller. This is for publication-style drift visualization.
            perf_drift_monitor_delta = 0.015
            exp.perf_drift_monitor_latch_steps = max(18, window_size // 7)
            exp.perf_drift_monitor_min_separation = max(500, int(2.5 * window_size))
            exp.perf_drift_monitor_short_window = max(32, window_size // 3)
            exp.perf_drift_monitor_long_window = max(100, int(1.5 * window_size))
            exp.perf_drift_monitor_drop_threshold = 0.010
        elif streaming_optimizer_method == "online_fairbo_ensemble":
            exp.perf_drift_latch_steps = max(30, window_size // 5)
            exp.perf_drift_min_separation = max(60, window_size // 2)
            perf_drift_monitor_delta = 0.012
            exp.perf_drift_monitor_latch_steps = max(20, window_size // 7)
            exp.perf_drift_monitor_min_separation = max(60, window_size // 3)
            exp.perf_drift_monitor_short_window = max(25, window_size // 2)
            exp.perf_drift_monitor_long_window = max(100, 2 * window_size)
            exp.perf_drift_monitor_drop_threshold = 0.009
        if dataset_name == "german":
            exp.perf_drift_min_separation = max(int(exp.perf_drift_min_separation), max(180, int(1.5 * window_size)))
            exp.perf_drift_monitor_min_separation = max(int(exp.perf_drift_monitor_min_separation), max(700, 3 * int(window_size)))
            exp.perf_drift_monitor_short_window = max(int(exp.perf_drift_monitor_short_window), max(45, int(0.75 * window_size)))
            exp.perf_drift_monitor_long_window = max(int(exp.perf_drift_monitor_long_window), max(180, 3 * int(window_size)))
            exp.perf_drift_monitor_drop_threshold = max(float(exp.perf_drift_monitor_drop_threshold), 0.014)
        exp.performance_drift = drift.ADWIN(delta=perf_drift_delta)
        exp.performance_drift_monitor = drift.ADWIN(delta=perf_drift_monitor_delta)
        exp.perf_drift_monitor_short_errors = deque(maxlen=exp.perf_drift_monitor_short_window)
        exp.perf_drift_monitor_long_errors = deque(maxlen=exp.perf_drift_monitor_long_window)

        exp.explain_engine = StreamingExplainabilityEngine(window_size=window_size)
        exp.topological_engine = TopologicalBiasEngine(
            window_size=window_size,
            cover_intervals=mapper_config["cover_intervals"],
            cover_overlap=mapper_config["cover_overlap"],
            min_samples=mapper_config["min_samples"],
            edge_jaccard_threshold=mapper_config["edge_jaccard_threshold"],
            eps_scale=mapper_config["eps_scale"],
            eps_floor=mapper_config["eps_floor"],
            knn_k=mapper_config["knn_k"],
            lens_name=mapper_config["lens_name"],
            min_verified=mapper_config["min_verified"],
            min_group_support=mapper_config["min_group_support"],
        )
        exp.llm_auditor = StreamingLLMAuditor(
            model_name=OLLAMA_MODEL,
            ollama_url=OLLAMA_URL,
            timeout=OLLAMA_TIMEOUT_SECONDS,
            max_retries=OLLAMA_MAX_RETRIES,
            cooldown_seconds=OLLAMA_CALL_COOLDOWN,
            unavailable_cooldown_seconds=OLLAMA_UNAVAILABLE_COOLDOWN,
        )

        exp.fairness_scope_manager = scope_manager
        exp.config_space_summary = make_json_safe(exp.config_space.summary()) if exp.config_space is not None else {}
        exp.config_space_manifest = make_json_safe(exp.config_space.manifest()) if exp.config_space is not None else {}
        exp.candidate_evaluator_config = make_json_safe(candidate_evaluator_cfg.__dict__)
        exp.optimizer_config = make_json_safe(osmac_cfg.__dict__)
        exp.optimizer_runtime_config = make_json_safe(optimizer_runtime_config)
        exp.static_benchmark_config = make_json_safe(static_benchmark_config)
        exp.static_benchmark_status = "queued" if bool(static_benchmark_config.get("enabled", False)) else "disabled"

        exp.scope_state = _initial_scope_state(
            scope_manager=exp.fairness_scope_manager,
            scope_cfg=fairness_scope_config,
            sensitive_attr=sensitive_attr,
        )
        exp.scope_snapshot = _safe_scope_snapshot(exp.fairness_scope_manager)

        _sync_scope_to_controller(exp)

        if CODECARBON_AVAILABLE:
            try:
                exp.carbon.enabled = True
                exp.carbon.tracker = EmissionsTracker(
                    project_name=f"FairASML_{exp_id}",
                    measure_power_secs=1,
                    save_to_file=False,
                    log_level="error",
                    allow_multiple_runs=True,
                    tracking_mode="process",
                )
            except Exception:
                exp.carbon.enabled = False
                exp.carbon.tracker = None
        else:
            exp.carbon.enabled = False
            exp.carbon.tracker = None

        logger.info(
            "Experiment %s initialized: encoded_feature_count=%s numeric=%s categorical=%s dropped=%s",
            exp_id,
            exp.feature_manifest.get("encoded_feature_count", 0),
            len(exp.feature_manifest.get("numeric_feature_columns", [])),
            len(exp.feature_manifest.get("categorical_feature_columns", [])),
            len(exp.feature_manifest.get("dropped_feature_columns", [])),
        )
        exp.last_response_cache = make_json_safe(build_response(exp, False))

        with EXPERIMENTS_LOCK:
            EXPERIMENTS[exp_id] = exp

        if bool(static_benchmark_config.get("enabled", False)):
            _start_static_benchmark_async(
                exp_id,
                df=df,
                dataset_name=safe_str(static_benchmark_config.get("dataset_name", dataset_name), "").strip() or dataset_name,
                config=static_benchmark_config,
            )

        return True, "ok", exp_id, None

    except Exception as e:
        logger.error("Error creating experiment: %s", e)
        logger.error(traceback.format_exc())
        return False, "init_exception", None, {"exception": str(e)}


# ============================================================
# Async LLM worker
# ============================================================

def async_llm_audit(
    exp: ExperimentState,
    current_step: int,
    fairness_snapshot: Dict[str, Any],
    metrics_snapshot: Dict[str, Any],
    topology_snapshot: Dict[str, Any],
    explainability_snapshot: Dict[str, Any],
    envelope_snapshot: Dict[str, Any],
    active_mitigation: Dict[str, Any],
    recent_fair_drift_events: List[int],
    acc: float,
) -> None:
    try:
        auditor = exp.llm_auditor
        if auditor is None:
            return

        outputs: Dict[str, Optional[str]] = {
            "narrative": None,
            "root_cause": None,
            "scope_summary": None,
            "mitigation_verification": None,
            "incident_report": None,
            "memory": None,
        }

        mapper_nodes = topology_snapshot.get("node_metrics", []) if isinstance(topology_snapshot, dict) else []
        current_fairness = safe_float(
            fairness_snapshot.get("fm_smooth", fairness_snapshot.get("primary_metric", 0.0)),
            0.0,
        )

        context_bundle = {
            "active_scope": fairness_snapshot.get("active_scope", {}),
            "scope_state": fairness_snapshot.get("scope_state", exp.scope_state),
            "fairness_scope_snapshot": fairness_snapshot.get("fairness_scope_snapshot", exp.scope_snapshot),
            "attribute_metrics": fairness_snapshot.get("attribute_metrics", {}) or metrics_snapshot.get("attribute_metrics", {}),
            "intersection_metrics": fairness_snapshot.get("intersection_metrics", {}) or metrics_snapshot.get("intersection_metrics", {}),
            "fairness_envelope": envelope_snapshot,
            "fm_raw": fairness_snapshot.get("fm_raw"),
            "fm_smooth": fairness_snapshot.get("fm_smooth"),
            "fm_drift": fairness_snapshot.get("fm_drift"),
            "fm_uncertainty": fairness_snapshot.get("fm_uncertainty"),
        }

        if mapper_nodes:
            res = auditor.generate_narrative(
                mapper_nodes=mapper_nodes,
                current_fairness=current_fairness,
                context=context_bundle,
            )
            if isinstance(res, dict) and res.get("status") == "ok":
                outputs["narrative"] = res.get("response")

        if recent_fair_drift_events:
            res = auditor.generate_root_cause(
                drift_events=recent_fair_drift_events,
                active_mitigation=active_mitigation,
                context=context_bundle,
            )
            if isinstance(res, dict) and res.get("status") == "ok":
                outputs["root_cause"] = res.get("response")

        res = auditor.generate_scope_summary(
            fairness_snapshot=fairness_snapshot,
            metrics_snapshot=metrics_snapshot,
            topo_snapshot=topology_snapshot,
            explainability_snapshot=explainability_snapshot,
        )
        if isinstance(res, dict) and res.get("status") == "ok":
            outputs["scope_summary"] = res.get("response")

        if active_mitigation:
            res = auditor.verify_mitigation(
                mitigation=active_mitigation,
                fairness_snapshot=fairness_snapshot,
                metrics_snapshot=metrics_snapshot,
                topo_snapshot=topology_snapshot,
                explainability_snapshot=explainability_snapshot,
            )
            if isinstance(res, dict) and res.get("status") == "ok":
                outputs["mitigation_verification"] = res.get("response")

        res = auditor.generate_incident_report(
            fairness_snapshot=fairness_snapshot,
            metrics_snapshot=metrics_snapshot,
            topo_snapshot=topology_snapshot,
            explainability_snapshot=explainability_snapshot,
            mitigation=active_mitigation,
        )
        if isinstance(res, dict) and res.get("status") == "ok":
            outputs["incident_report"] = res.get("response")

        if recent_fair_drift_events and active_mitigation:
            action = active_mitigation.get("action_taken", "Monitoring")
            res = auditor.build_memory_entry(
                timestep=current_step,
                accuracy=acc,
                fairness=current_fairness,
                action_taken=action,
                context=context_bundle,
            )
            if isinstance(res, dict) and res.get("status") == "ok":
                outputs["memory"] = res.get("response")

        with exp.state_lock:
            for key in ("narrative", "root_cause", "scope_summary", "mitigation_verification", "incident_report"):
                if outputs[key]:
                    exp.llm_auditor_output[key] = outputs[key]

            if outputs["memory"]:
                exp.llm_auditor_output["memory"].append(outputs["memory"])

    except Exception as e:
        logger.error("Background LLM Error: %s", e)
    finally:
        with exp.state_lock:
            exp.llm_is_processing = False


# ============================================================
# OSMAC sidecar loop
# ============================================================

def _optimizer_protected_attributes(exp: ExperimentState) -> List[str]:
    scope_state = exp.scope_state if isinstance(exp.scope_state, dict) else {}
    active_attrs = scope_state.get("active_attributes", []) if isinstance(scope_state.get("active_attributes", []), list) else []
    attrs = [str(v) for v in active_attrs if str(v).strip()]
    if attrs:
        return attrs
    return [str(v) for v in exp.sensitive_attr if str(v).strip()]


def _build_optimizer_fairness_context(exp: ExperimentState) -> Dict[str, Any]:
    fairness_snapshot = exp.latest_fairness_snapshot if isinstance(exp.latest_fairness_snapshot, dict) else {}
    envelope_snapshot = exp.latest_envelope_snapshot if isinstance(exp.latest_envelope_snapshot, dict) else {}
    topology_snapshot = exp.latest_topology_snapshot if isinstance(exp.latest_topology_snapshot, dict) else {}

    return {
        "fm_raw": safe_float(fairness_snapshot.get("fm_raw", fairness_snapshot.get("primary_metric", 0.0)), 0.0),
        "fm_smooth": safe_float(fairness_snapshot.get("fm_smooth", fairness_snapshot.get("primary_metric", 0.0)), 0.0),
        "fm_drift": safe_float(fairness_snapshot.get("fm_drift", 0.0), 0.0),
        "fm_uncertainty": safe_float(fairness_snapshot.get("fm_uncertainty", 1.0), 1.0),
        "confidence": safe_float(fairness_snapshot.get("confidence", 0.0), 0.0),
        "is_reliable": bool(fairness_snapshot.get("is_reliable", False)),
        "effective_support": safe_int(fairness_snapshot.get("effective_support", 0), 0),
        "sfm_state": copy.deepcopy(fairness_snapshot.get("sfm_state", {})) if isinstance(fairness_snapshot.get("sfm_state"), dict) else {},
        "envelope_lower": safe_float(envelope_snapshot.get("lower", envelope_snapshot.get("envelope_lower", 0.0)), 0.0),
        "envelope_upper": safe_float(envelope_snapshot.get("upper", envelope_snapshot.get("envelope_upper", 0.0)), 0.0),
        "envelope_violation": bool(envelope_snapshot.get("is_violation", envelope_snapshot.get("envelope_violation", False))),
        "envelope_severity": safe_float(envelope_snapshot.get("severity", envelope_snapshot.get("envelope_severity", 0.0)), 0.0),
        "envelope_pressure": safe_float(envelope_snapshot.get("envelope_pressure", 0.0), 0.0),
        "trend_to_boundary": bool(envelope_snapshot.get("trend_to_boundary", False)),
        "lower_bound_enforced": bool(envelope_snapshot.get("lower_bound_enforced", False)),
        "topo_hotspot_score": safe_float(topology_snapshot.get("hotspot_score", 0.0), 0.0),
        "topo_structural_drift": safe_float(topology_snapshot.get("structural_drift", 0.0), 0.0),
    }


def _optimizer_model_family(model_name: Any) -> str:
    name = str(model_name or "").strip().lower()
    mapping = {
        "logistic_regression": "Logistic Regression",
        "gaussian_nb": "Gaussian NB",
        "hoeffding_tree": "Hoeffding Tree",
        "knn": "KNN",
        "srp_classifier": "SRP Ensemble",
    }
    return mapping.get(name, str(model_name or "Unknown"))


def _optimizer_pipeline_label(config: Dict[str, Any]) -> str:
    pre = str(config.get("preprocessor", "none"))
    fs = str(config.get("feature_selector", "none"))
    model = str(config.get("model", "unknown"))
    parts: List[str] = []
    if pre and pre != "none":
        parts.append(pre)
    if fs and fs != "none":
        parts.append(fs)
    parts.append(model)
    return " | ".join(parts)


def _summarize_optimizer_candidate(
    payload: Dict[str, Any],
    *,
    optimizer_iteration: int,
    candidate_rank: int,
    evaluation_window: Dict[str, Any],
) -> Dict[str, Any]:
    config = payload.get("config", {}) if isinstance(payload.get("config"), dict) else {}
    predictive = payload.get("predictive_metrics", {}) if isinstance(payload.get("predictive_metrics"), dict) else {}
    fairness = payload.get("fairness_metrics", {}) if isinstance(payload.get("fairness_metrics"), dict) else {}
    feasibility = payload.get("feasibility", {}) if isinstance(payload.get("feasibility"), dict) else {}
    score_payload = payload.get("score_payload", {}) if isinstance(payload.get("score_payload"), dict) else {}

    accuracy = safe_float(predictive.get("accuracy", score_payload.get("objective_value", 0.0)), 0.0)
    unfairness = safe_float(
        fairness.get("fm_smooth", fairness.get("constraint_value", fairness.get("primary_metric", 0.0))),
        0.0,
    )
    constrained_utility = safe_float(
        score_payload.get("constrained_utility", score_payload.get("utility_score", accuracy)),
        accuracy,
    )
    return {
        "optimizer_iteration": int(optimizer_iteration),
        "candidate_rank": int(candidate_rank),
        "config_id": str(payload.get("config_id", "")),
        "pipeline_label": _optimizer_pipeline_label(config),
        "model_name": str(config.get("model", "unknown")),
        "model_family": _optimizer_model_family(config.get("model")),
        "preprocessor": str(config.get("preprocessor", "none")),
        "feature_selector": str(config.get("feature_selector", "none")),
        "accuracy": float(accuracy),
        "validation_error": float(max(0.0, 1.0 - accuracy)),
        "log_loss": safe_float(predictive.get("log_loss", 0.0), 0.0),
        "unfairness": float(unfairness),
        "fairness_metric": float(unfairness),
        "fairness_metric_name": str(fairness.get("primary_metric_name", fairness.get("constraint_name", "fairness"))),
        "feasible": bool(fairness.get("constraint_feasible", feasibility.get("feasible_hard", False))),
        "envelope_violation": bool(fairness.get("envelope_violation", False)),
        "feasibility_probability": safe_float(
            fairness.get("feasibility_probability", feasibility.get("feasibility_probability", 0.0)),
            0.0,
        ),
        "fairness_risk": safe_float(fairness.get("fairness_risk", feasibility.get("fairness_risk", 0.0)), 0.0),
        "constrained_utility": float(constrained_utility),
        "objective_value": safe_float(score_payload.get("objective_value", accuracy), accuracy),
        "constraint_value": safe_float(score_payload.get("constraint_value", unfairness), unfairness),
        "window_index": safe_int(evaluation_window.get("window_index", optimizer_iteration), optimizer_iteration),
        "processed_step": safe_int(evaluation_window.get("end", 0), 0),
        "window_start": safe_int(evaluation_window.get("start", 0), 0),
        "window_end": safe_int(evaluation_window.get("end", 0), 0),
    }


def _run_optimizer_cycle(exp: ExperimentState, upto_index: int) -> None:
    runtime_cfg = exp.optimizer_runtime_config if isinstance(exp.optimizer_runtime_config, dict) else {}
    if not safe_bool(runtime_cfg.get("enabled", False), False):
        return
    if exp.config_space is None or exp.candidate_evaluator is None or exp.optimizer is None:
        return

    processed = max(0, min(int(upto_index), int(exp.sample_budget)))
    min_samples = max(2, safe_int(runtime_cfg.get("min_samples", OSMAC_MIN_WINDOW), OSMAC_MIN_WINDOW))
    if processed < min_samples:
        return

    step_frequency = max(1, safe_int(runtime_cfg.get("step_frequency", min_samples), min_samples))
    if exp.last_optimizer_eval_step and (processed - int(exp.last_optimizer_eval_step)) < step_frequency:
        return

    eval_window_size = max(
        min_samples,
        safe_int(runtime_cfg.get("window_size", exp.window_size), exp.window_size),
    )
    start_idx = max(0, processed - eval_window_size)
    if (processed - start_idx) < min_samples:
        return

    X_window = [dict(x or {}) for x in exp.X_model[start_idx:processed]]
    y_window = list(exp.y[start_idx:processed])
    sensitive_window = [
        _normalize_sensitive_dict(exp.X_full[j], exp.sensitive_attr)
        for j in range(start_idx, processed)
    ]

    fairness_context = _build_optimizer_fairness_context(exp)
    include_incumbent = safe_bool(runtime_cfg.get("include_incumbent", True), True)
    force_exploration = bool(
        safe_bool(runtime_cfg.get("force_exploration_on_violation", True), True)
        and fairness_context.get("envelope_violation", False)
    )
    protected_attributes = _optimizer_protected_attributes(exp)
    history_tail = max(1, safe_int(runtime_cfg.get("history_tail", OSMAC_HISTORY_TAIL), OSMAC_HISTORY_TAIL))
    streaming_method = str(exp.optimizer_config.get("streaming_method", "online_fairbo_ensemble")).strip().lower()
    if streaming_method == "online_fairbo_basic" and exp.model is not None:
        current_live_acc = 0.0
        if exp.accuracy_over_time:
            tail = exp.accuracy_over_time[-max(25, int(eval_window_size // 2)):]
            if tail:
                current_live_acc = float(np.mean([safe_float(v, 0.0) for v in tail]))
        current_fm = safe_float(exp.latest_fairness_snapshot.get("current_bias", 0.0), 0.0) if isinstance(exp.latest_fairness_snapshot, dict) else 0.0
        current_env_upper = safe_float(
            exp.latest_envelope_snapshot.get("upper", exp.latest_envelope_snapshot.get("envelope_upper", 0.05)),
            0.05,
        ) if isinstance(exp.latest_envelope_snapshot, dict) else 0.05
        latest_handoff = exp.latest_live_handoff if isinstance(exp.latest_live_handoff, dict) else {}
        handoff_acc = safe_float(latest_handoff.get("accuracy", 0.0), 0.0)
        handoff_feasible = bool(latest_handoff.get("constraint_feasible", False))
        handoff_policy = _basic_live_handoff_policy(exp.dataset_name, current_live_acc)
        keep_search_frozen = (
            bool(latest_handoff.get("adopted", False))
            and handoff_feasible
            and handoff_acc >= safe_float(handoff_policy.get("stable_freeze_accuracy", 0.85), 0.85)
            and current_live_acc >= max(0.0, handoff_acc - 0.08)
            and current_fm <= max(current_env_upper, 0.05)
        )
        if keep_search_frozen and hasattr(exp.model, "freeze_search"):
            exp.model.freeze_search()
        elif hasattr(exp.model, "unfreeze_search"):
            exp.model.unfreeze_search()
        basic_live_handoff_enabled = current_live_acc < 0.88
    else:
        basic_live_handoff_enabled = True

    try:
        proposal_batch = exp.optimizer.propose_next(
            include_incumbent=include_incumbent,
            fairness_context=fairness_context,
            force_exploration=force_exploration,
        )

        evaluated_results: List[Dict[str, Any]] = []
        evaluated_objects: List[CandidateEvaluationResult] = []
        active_window_index = safe_int(getattr(exp.optimizer, "window_index", 0), 0)

        for rank, cfg in enumerate(proposal_batch.proposed_configs):
            candidate_name = (
                exp.config_space.to_candidate_name(cfg, prefix=f"osmac_w{active_window_index}_r{rank}")
                if exp.config_space is not None
                else f"osmac_w{active_window_index}_r{rank}"
            )
            result_obj = exp.candidate_evaluator.evaluate(
                config=cfg,
                X_window=X_window,
                y_window=y_window,
                sensitive_window=sensitive_window,
                protected_attributes=protected_attributes,
                scope_state=copy.deepcopy(exp.scope_state),
                fairness_scope_snapshot=copy.deepcopy(exp.scope_snapshot),
                timestep_start=start_idx,
                candidate_name=candidate_name,
                extra_metadata={
                    "window_start": int(start_idx),
                    "window_end": int(processed),
                    "active_scope": copy.deepcopy(exp.scope_state),
                    "fairness_context": copy.deepcopy(fairness_context),
                },
            )
            evaluated_objects.append(result_obj)
            evaluated_results.append(make_json_safe(result_obj.to_dict()))

        register_out = exp.optimizer.register_batch(
            evaluated_results,
            timestep=processed,
            window_index=active_window_index,
        )
        optimizer_state_after_register = make_json_safe(exp.optimizer.get_state())
        exp.optimizer.step_window()
        optimizer_state_after_step = make_json_safe(exp.optimizer.get_state())

        exp.last_optimizer_eval_step = int(processed)
        exp.latest_candidate_evaluations = list(evaluated_results[-history_tail:])
        evaluation_window = {
            "start": int(start_idx),
            "end": int(processed),
            "size": int(processed - start_idx),
            "protected_attributes": list(protected_attributes),
            "window_index": int(active_window_index),
        }
        exp.latest_optimizer_batch = make_json_safe({
            **proposal_batch.to_dict(),
            "evaluation_window": evaluation_window,
            "fairness_context": copy.deepcopy(fairness_context),
            "register_out": register_out,
            "optimizer_state_after_register": optimizer_state_after_register,
            "optimizer_state_after_step": optimizer_state_after_step,
        })

        adopted_payload: Dict[str, Any] = {}
        if evaluated_objects and exp.model is not None:
            feasible_objs = [
                obj for obj in evaluated_objects
                if bool(obj.fairness_metrics.get("constraint_feasible", obj.feasibility.get("feasible_hard", False)))
            ]
            adoption_pool = feasible_objs if feasible_objs else list(evaluated_objects)
            if streaming_method == "online_fairbo_basic":
                preferred_models = {"srp_classifier", "gaussian_nb", "logistic_regression"}
                preferred_preprocessors = {"standard_scaler"}
                restricted_pool = []
                for obj in adoption_pool:
                    cfg = obj.config if isinstance(obj.config, dict) else {}
                    model_name = str(cfg.get("model", "")).strip().lower()
                    pre_name = str(cfg.get("preprocessor", "")).strip().lower()
                    fs_name = str(cfg.get("feature_selector", "none")).strip().lower()
                    if (
                        model_name in preferred_models
                        and pre_name in preferred_preprocessors
                        and fs_name in {"none", ""}
                    ):
                        restricted_pool.append(obj)
                if restricted_pool:
                    adoption_pool = restricted_pool
                if str(exp.dataset_name or "").strip().lower() == "german":
                    adopt_obj = max(
                        adoption_pool,
                        key=lambda obj: (
                            safe_float(obj.predictive_metrics.get("accuracy", 0.0), 0.0),
                            _german_live_handoff_score(
                                accuracy=safe_float(obj.predictive_metrics.get("accuracy", 0.0), 0.0),
                                balanced_accuracy=safe_float(
                                    obj.predictive_metrics.get("balanced_accuracy", obj.predictive_metrics.get("accuracy", 0.0)),
                                    0.0,
                                ),
                                constrained_utility=safe_float(obj.score_payload.get("constrained_utility", 0.0), 0.0),
                            ),
                            safe_float(obj.predictive_metrics.get("balanced_accuracy", 0.0), 0.0),
                        ),
                    )
                else:
                    adopt_obj = max(
                        adoption_pool,
                        key=lambda obj: (
                            safe_float(obj.predictive_metrics.get("accuracy", 0.0), 0.0),
                            safe_float(obj.score_payload.get("constrained_utility", 0.0), 0.0),
                        ),
                    )
            else:
                adopt_obj = max(
                    adoption_pool,
                    key=lambda obj: (
                        safe_float(obj.score_payload.get("constrained_utility", 0.0), 0.0),
                        safe_float(obj.predictive_metrics.get("accuracy", 0.0), 0.0),
                    ),
                )
            score_hint = safe_float(
                adopt_obj.score_payload.get("constrained_utility", adopt_obj.predictive_metrics.get("accuracy", 0.0)),
                0.0,
            )
            should_adopt = True
            recent_live_acc = 0.0
            if exp.accuracy_over_time:
                tail = exp.accuracy_over_time[-max(25, int(eval_window_size // 2)):]
                if tail:
                    recent_live_acc = float(np.mean([safe_float(v, 0.0) for v in tail]))
            adopt_acc = safe_float(adopt_obj.predictive_metrics.get("accuracy", 0.0), 0.0)
            adopt_bal_acc = safe_float(adopt_obj.predictive_metrics.get("balanced_accuracy", adopt_acc), adopt_acc)
            adopt_live_score = _german_live_handoff_score(
                accuracy=adopt_acc,
                balanced_accuracy=adopt_bal_acc,
                constrained_utility=score_hint,
            )
            if streaming_method == "online_fairbo_basic":
                latest_handoff = exp.latest_live_handoff if isinstance(exp.latest_live_handoff, dict) else {}
                current_handoff_acc = safe_float(latest_handoff.get("accuracy", 0.0), 0.0) if bool(latest_handoff.get("adopted", False)) else 0.0
                live_baseline = max(recent_live_acc, current_handoff_acc)
                feasible_flag = bool(adopt_obj.fairness_metrics.get("constraint_feasible", False))
                if str(exp.dataset_name or "").strip().lower() == "german":
                    current_handoff_score = safe_float(
                        latest_handoff.get("live_handoff_score", latest_handoff.get("accuracy", 0.0)),
                        0.0,
                    ) if bool(latest_handoff.get("adopted", False)) else 0.0
                    live_baseline_score = max(recent_live_acc, current_handoff_score)
                    score_margin = 0.0020 if live_baseline_score >= 0.78 else 0.0010
                    minimum_adopt_accuracy = 0.765 if live_baseline >= 0.77 else 0.755
                    minimum_balanced_accuracy = 0.59 if live_baseline >= 0.78 else 0.57
                    should_adopt = bool(basic_live_handoff_enabled) and feasible_flag and (
                        (
                            adopt_live_score >= max(0.74, live_baseline_score + score_margin)
                            and adopt_acc >= minimum_adopt_accuracy
                            and adopt_bal_acc >= minimum_balanced_accuracy
                        )
                        or live_baseline_score <= 0.0
                    )
                else:
                    handoff_policy = _basic_live_handoff_policy(exp.dataset_name, live_baseline)
                    improvement_margin = safe_float(handoff_policy.get("improvement_margin", 0.010), 0.010)
                    minimum_adopt_accuracy = safe_float(handoff_policy.get("minimum_adopt_accuracy", 0.85), 0.85)
                    should_adopt = bool(basic_live_handoff_enabled) and feasible_flag and (
                        adopt_acc >= max(minimum_adopt_accuracy, live_baseline + improvement_margin)
                        or live_baseline <= 0.0
                    )
            adopted_idx = None
            if should_adopt and hasattr(exp.model, "adopt_external_candidate"):
                prefer_steps = max(100, 2 * int(eval_window_size))
                if streaming_method == "online_fairbo_basic":
                    prefer_steps = max(600, 6 * int(eval_window_size))
                elif streaming_method == "online_fairbo_ensemble":
                    prefer_steps = max(250, 4 * int(eval_window_size))
                if str(exp.dataset_name or "").strip().lower() == "german":
                    prefer_steps = max(prefer_steps, max(1200, 8 * int(eval_window_size)))
                adopted_idx = _call_with_optional_kwargs(
                    exp.model.adopt_external_candidate,
                    adopt_obj.estimator_object,
                    candidate_name=str(adopt_obj.raw_artifacts.get("candidate_name", adopt_obj.config_id)),
                    score_hint=float(score_hint),
                    accuracy_hint=safe_float(adopt_obj.predictive_metrics.get("accuracy", 0.0), 0.0),
                    decision_threshold=safe_float(adopt_obj.raw_artifacts.get("selected_threshold", 0.5), 0.5),
                    class_weight_profile=str(
                        adopt_obj.raw_artifacts.get(
                            "class_weight_profile",
                            (
                                dict(adopt_obj.config.get("model_params", {}) or {}).get("class_weight_profile", "none")
                                if isinstance(adopt_obj.config, dict) else "none"
                            ),
                        )
                    ),
                    prefer_live_steps=prefer_steps,
                )
            adopted_payload = {
                "config_id": str(adopt_obj.config_id),
                "candidate_name": str(adopt_obj.raw_artifacts.get("candidate_name", adopt_obj.config_id)),
                "accuracy": adopt_acc,
                "balanced_accuracy": float(adopt_bal_acc),
                "live_handoff_score": float(adopt_live_score),
                "unfairness": safe_float(adopt_obj.fairness_metrics.get("fm_smooth", 0.0), 0.0),
                "constraint_feasible": bool(adopt_obj.fairness_metrics.get("constraint_feasible", False)),
                "constrained_utility": float(score_hint),
                "decision_threshold": safe_float(adopt_obj.raw_artifacts.get("selected_threshold", 0.5), 0.5),
                "class_weight_profile": str(
                    adopt_obj.raw_artifacts.get(
                        "class_weight_profile",
                        (
                            dict(adopt_obj.config.get("model_params", {}) or {}).get("class_weight_profile", "none")
                            if isinstance(adopt_obj.config, dict) else "none"
                        ),
                    )
                ),
                "adopted_candidate_index": safe_int(adopted_idx, -1),
                "adopted": bool(should_adopt),
                "recent_live_accuracy": float(recent_live_acc),
                "window_index": int(active_window_index),
                "processed_step": int(processed),
            }
            exp.latest_live_handoff = make_json_safe(adopted_payload)
            if isinstance(exp.latest_optimizer_batch, dict):
                exp.latest_optimizer_batch["live_handoff"] = copy.deepcopy(exp.latest_live_handoff)
            if (
                streaming_method == "online_fairbo_basic"
                and bool(adopted_payload.get("adopted", False))
                and bool(adopted_payload.get("constraint_feasible", False))
                and safe_float(adopted_payload.get("accuracy", 0.0), 0.0) >= safe_float(
                    _basic_live_handoff_policy(exp.dataset_name, recent_live_acc).get("stable_freeze_accuracy", 0.85),
                    0.85,
                )
                and hasattr(exp.model, "freeze_search")
            ):
                exp.model.freeze_search()

        for rank, payload in enumerate(evaluated_results):
            bounded_append(
                exp.optimizer_candidate_history,
                make_json_safe(
                    _summarize_optimizer_candidate(
                        payload,
                        optimizer_iteration=active_window_index,
                        candidate_rank=rank,
                        evaluation_window=evaluation_window,
                    )
                ),
                max_len=MAX_OPTIMIZER_CANDIDATE_HISTORY,
            )
        exp.latest_optimizer_error = ""

    except Exception as e:
        exp.latest_optimizer_error = f"{type(e).__name__}: {e}"
        logger.warning("OSMAC sidecar cycle failed at step %s: %s", processed, e)


# ============================================================
# Stream loop
# ============================================================

def evaluate_stream(exp: ExperimentState, start: int, end: int) -> None:
    exp.carbon.start_chunk()

    for i in range(start, end):
        xi_model = exp.X_model[i]
        xi_full = exp.X_full[i]
        yi = exp.y[i]

        model = exp.model
        controller = exp.controller
        active_mitigation = exp.active_mitigation_state if isinstance(exp.active_mitigation_state, dict) else {}

        all_sensitive_vals = _normalize_sensitive_dict(xi_full, exp.sensitive_attr) if exp.sensitive_attr else {}

        # scope reevaluation only; do not inject fake observations
        try:
            if exp.fairness_scope_manager is not None and all_sensitive_vals:
                scope_mode = exp.fairness_scope_config.get("mode", "manual")
                should_rescore = (
                    scope_mode in {"assisted", "adaptive"}
                    and (i == 0 or i % SCOPE_REEVAL_FREQUENCY == 0)
                )

                if should_rescore:
                    selected = _call_with_optional_kwargs(exp.fairness_scope_manager.select_scope) or {}
                    selected_attrs = selected.get("active_attributes", []) or []
                    selected_intersections = selected.get("active_intersections", []) or []

                    if scope_mode == "assisted" and exp.fairness_scope_config.get("approved_candidates"):
                        approved = set(exp.fairness_scope_config.get("approved_candidates", []))
                        selected_attrs = [a for a in selected_attrs if a in approved]

                    if selected_attrs:
                        exp.scope_state = {
                            "mode": scope_mode,
                            "active_attributes": list(selected_attrs),
                            "active_intersections": list(selected_intersections),
                            "primary_attribute": str(selected_attrs[0]),
                            "scores": make_json_safe(selected.get("scores", {})),
                            "selection_reason": "stream_rescore",
                            "last_rescore_step": int(i),
                        }
                        _sync_scope_to_controller(exp)
        except Exception as e:
            logger.debug("Scope manager update failed at step %s: %s", i, e)

        active_sensitive_vals = _active_sensitive_from_scope(
            xi_full=xi_full,
            all_sensitive_dict=all_sensitive_vals,
            scope_state=exp.scope_state,
        )
        exp.scope_snapshot = _sync_scope_context_to_modules(exp, active_sensitive_vals)

        exp.decision_context = _build_decision_context(
            xi_full=xi_full,
            all_sensitive_dict=all_sensitive_vals,
            active_sensitive_dict=active_sensitive_vals,
            scope_state=exp.scope_state,
            scope_snapshot=exp.scope_snapshot,
            active_mitigation=active_mitigation,
        )

        # ---------------- 1) Predict + Learn ----------------
        uncertainty_entropy = 0.0
        y_pred, y_prob, acc = 0, 0.5, 0.0

        try:
            y_pred = _call_with_optional_kwargs(
                model.predict_one,
                xi_model,
                decision_context=exp.decision_context,
            )
            if y_pred is None:
                y_pred = 0

            proba = _call_with_optional_kwargs(
                model.predict_proba_one,
                xi_model,
                decision_context=exp.decision_context,
            ) or {}
            y_prob = _get_proba_1(proba)
            _update_y_prob_debug(exp, y_prob)

            p1 = float(np.clip(y_prob, 1e-12, 1.0 - 1e-12))
            uncertainty_entropy = float(-(p1 * np.log(p1) + (1.0 - p1) * np.log(1.0 - p1)))

            exp.accuracy_metric.update(yi, y_pred)

            weight = _resolve_scope_weight(
                active_mitigation=active_mitigation,
                active_sensitive_dict=active_sensitive_vals,
                decision_context=exp.decision_context,
                fallback_primary_attr=exp.scope_state.get("primary_attribute", exp.primary_sensitive_attr),
            )

            _call_with_optional_kwargs(
                model.learn_one,
                xi_model,
                yi,
                w=weight,
                decision_context=exp.decision_context,
            )

            acc = safe_float(exp.accuracy_metric.get())
            bounded_append(exp.accuracy_over_time, acc)

        except Exception:
            logger.debug("Predict/learn failure at step %s", i)
            y_pred, y_prob, acc = 0, 0.5, 0.0
            uncertainty_entropy = 0.0
            _update_y_prob_debug(exp, 0.5)
            bounded_append(exp.accuracy_over_time, 0.0)

        if i % MAPPER_UPDATE_FREQUENCY == 0 and i > 0:
            logger.info("Step %s y_prob stats: %s", i, exp.y_prob_debug_snapshot)

        # refresh scope manager with actual prediction only
        try:
            if exp.fairness_scope_manager is not None and all_sensitive_vals:
                _call_with_optional_kwargs(
                    exp.fairness_scope_manager.observe,
                    row=xi_full,
                    y_true=yi,
                    y_pred=int(y_pred),
                    prob=float(y_prob),
                )
        except Exception:
            logger.debug("Post-predict scope observe failed at step %s", i)

        # ---------------- 2) Performance Drift ----------------
        perf_drift_event = False
        perf_drift_state = False
        perf_drift_monitor_event = False
        perf_drift_monitor_state = False
        try:
            is_error = 1 if int(y_pred) != int(yi) else 0
            exp.performance_drift.update(is_error)
            perf_drift_event = _detector_drift_flag(exp.performance_drift)
            if exp.performance_drift_monitor is not None:
                exp.performance_drift_monitor.update(is_error)
                perf_drift_monitor_event = _detector_drift_flag(exp.performance_drift_monitor)
            if exp.perf_drift_monitor_short_window > 0:
                exp.perf_drift_monitor_short_errors.append(is_error)
            if exp.perf_drift_monitor_long_window > 0:
                exp.perf_drift_monitor_long_errors.append(is_error)
            if (
                len(exp.perf_drift_monitor_short_errors) >= exp.perf_drift_monitor_short_window
                and len(exp.perf_drift_monitor_long_errors) >= exp.perf_drift_monitor_long_window
            ):
                short_err = float(np.mean(exp.perf_drift_monitor_short_errors))
                long_err = float(np.mean(exp.perf_drift_monitor_long_errors))
                if (short_err - long_err) >= float(exp.perf_drift_monitor_drop_threshold):
                    perf_drift_monitor_event = True
        except Exception:
            perf_drift_event = False
            perf_drift_monitor_event = False

        if perf_drift_event:
            if (i - int(exp.last_perf_drift_step)) >= max(1, int(exp.perf_drift_min_separation)):
                exp.perf_drift_latch = int(exp.perf_drift_latch_steps)
                exp.last_perf_drift_step = int(i)
                exp.perf_drift_event_points.append(i)
            else:
                perf_drift_event = False

        if exp.perf_drift_latch > 0:
            perf_drift_state = True
            exp.perf_drift_latch -= 1

        if perf_drift_monitor_event:
            if (i - int(exp.last_perf_drift_monitor_step)) >= max(1, int(exp.perf_drift_monitor_min_separation)):
                exp.perf_drift_monitor_latch = int(exp.perf_drift_monitor_latch_steps)
                exp.last_perf_drift_monitor_step = int(i)
                exp.perf_drift_monitor_event_points.append(i)
            else:
                perf_drift_monitor_event = False

        if exp.perf_drift_monitor_latch > 0:
            perf_drift_monitor_state = True
            exp.perf_drift_monitor_latch -= 1

        bounded_append(exp.perf_drift_state_over_time, 1 if perf_drift_state else 0)
        bounded_append(exp.perf_drift_monitor_state_over_time, 1 if perf_drift_monitor_state else 0)

        # ---------------- 3) Fairness + SFM + DFE ----------------
        fairness_value = 0.0
        fairness_stability = 0.0
        fair_drift_event = False
        fair_drift_state = False
        fm_raw = 0.0
        fm_smooth = 0.0
        fm_drift = 0.0
        fm_uncertainty = 1.0
        sfm_payload: Dict[str, Any] = {}
        envelope_snapshot: Dict[str, Any] = {}
        controller_telemetry: Dict[str, Any] = {}

        try:
            if active_sensitive_vals:
                add_out = _call_with_optional_kwargs(
                    exp.fairness_engine.add_record,
                    y_true=yi,
                    y_pred=y_pred,
                    sensitive=active_sensitive_vals,
                    prob=y_prob,
                ) or {}

                fair_drift_event = bool(add_out.get("drift_event", False))
                if fair_drift_event:
                    exp.fairness_drift_event_points.append(i)

                controller_telemetry = {
                    "fairness_drift_severity": safe_float(
                        add_out.get("drift_severity", add_out.get("severity", 0.0)),
                        0.0,
                    ),
                    "harmed_group_error": add_out.get("harmed_group_error"),
                    "advantaged_group_error": add_out.get("advantaged_group_error"),
                    "harmed_group_dp": add_out.get("harmed_group_dp"),
                    "advantaged_group_dp": add_out.get("advantaged_group_dp"),
                    "fairness_drift_event": bool(fair_drift_event),
                    "uncertainty_entropy": float(uncertainty_entropy),
                }

                fairness_metric_snapshot = (
                    copy.deepcopy(exp.latest_metrics_snapshot)
                    if isinstance(exp.latest_metrics_snapshot, dict)
                    else {}
                )
                if exp.fairness_metric is not None:
                    try:
                        _call_with_optional_kwargs(
                            exp.fairness_metric.add_record,
                            y_true=yi,
                            y_pred=y_pred,
                            prob=y_prob,
                            sensitive=active_sensitive_vals,
                            scope_state=copy.deepcopy(exp.scope_state),
                            fairness_scope_snapshot=copy.deepcopy(exp.scope_snapshot),
                            active_sensitive_dict=copy.deepcopy(active_sensitive_vals),
                        )
                        fairness_metric_snapshot = _call_with_optional_kwargs(exp.fairness_metric.snapshot) or {}
                        exp.latest_metrics_snapshot = make_json_safe(fairness_metric_snapshot)

                        conf_gap = safe_float(
                            getattr(exp.fairness_metric, "get_ddsp_confidence_gap", lambda: 0.0)(),
                            0.0,
                        )
                        group_er = _call_with_optional_kwargs(exp.fairness_metric.group_error_rates) or {}

                        controller_telemetry["ddsp_confidence_gap"] = conf_gap
                        controller_telemetry["group_error_rates"] = group_er
                        controller_telemetry["verified_support"] = safe_int(getattr(exp.fairness_metric, "ver_total", 0), 0)
                        controller_telemetry["hb_total"] = safe_int(getattr(exp.fairness_metric, "pred_total", 0), 0)

                        if isinstance(group_er, dict) and group_er:
                            harmed = max(group_er, key=group_er.get)
                            advantaged = min(group_er, key=group_er.get)
                            controller_telemetry["highest_error_group"] = harmed
                            controller_telemetry["lowest_error_group"] = advantaged
                    except Exception:
                        logger.debug("fairness_metric failure at step %s", i)

                snap = _call_with_optional_kwargs(exp.fairness_engine.snapshot, exp.accuracy_over_time) or {}
                prev_fm_smooth = exp.fm_smooth_over_time[-1] if exp.fm_smooth_over_time else None
                sfm_payload = _build_sfm_payload(
                    fairness_snapshot=snap,
                    metrics_snapshot=fairness_metric_snapshot,
                    prev_fm_smooth=prev_fm_smooth,
                )
                if sfm_payload:
                    snap["sfm_payload"] = copy.deepcopy(sfm_payload)
                    snap["sfm"] = list(sfm_payload.get("sfm", []))
                    snap["fm_raw"] = safe_float(sfm_payload.get("fm_raw", snap.get("primary_metric", 0.0)), 0.0)
                    snap["fm_smooth"] = safe_float(sfm_payload.get("fm_smooth", snap.get("primary_metric", 0.0)), 0.0)
                    snap["fm_drift"] = safe_float(sfm_payload.get("fm_drift", 0.0), 0.0)
                    snap["fm_smooth_drift"] = safe_float(sfm_payload.get("fm_smooth_drift", snap["fm_drift"]), snap["fm_drift"])
                    snap["fm_uncertainty"] = safe_float(sfm_payload.get("fm_uncertainty", 1.0), 1.0)
                    snap["confidence"] = safe_float(sfm_payload.get("confidence", 0.0), 0.0)
                    snap["is_reliable"] = bool(sfm_payload.get("is_reliable", False))
                    snap["insufficient_support"] = bool(sfm_payload.get("insufficient_support", True))
                    snap["window_support"] = safe_int(sfm_payload.get("window_support", 0), 0)
                    snap["effective_support"] = safe_int(sfm_payload.get("effective_support", 0), 0)
                    snap["eligible_groups"] = safe_int(sfm_payload.get("eligible_groups", 0), 0)
                    snap["group_support"] = dict(sfm_payload.get("group_counts", {})) if isinstance(sfm_payload.get("group_counts"), dict) else {}
                    snap["low_support_groups"] = list(sfm_payload.get("low_support_groups", [])) if isinstance(sfm_payload.get("low_support_groups"), list) else []
                    snap["sfm_state"] = copy.deepcopy(sfm_payload.get("sfm_state", {})) if isinstance(sfm_payload.get("sfm_state"), dict) else {}

                    # Backward-compatible aliases for older frontend builds.
                    snap["raw_fairness"] = snap["fm_raw"]
                    snap["smoothed_fairness"] = snap["fm_smooth"]
                    snap["fairness_uncertainty"] = snap["fm_uncertainty"]

                exp.latest_fairness_snapshot = make_json_safe(snap)
                exp.latest_sfm_state = make_json_safe(snap.get("sfm_state", {})) if isinstance(snap.get("sfm_state"), dict) else {}

                sfm_payload = _get_sfm_payload(snap)
                fm_raw = safe_float(sfm_payload.get("fm_raw", snap.get("primary_metric", 0.0)), 0.0)
                fm_smooth = safe_float(sfm_payload.get("fm_smooth", fm_raw), fm_raw)
                fm_drift = safe_float(sfm_payload.get("fm_drift", 0.0), 0.0)
                fm_uncertainty = safe_float(sfm_payload.get("fm_uncertainty", 1.0), 1.0)

                # Controller fairness signal is now SFM-aligned
                fairness_value = float(fm_smooth)
                fairness_stability = safe_float(snap.get("stability", 0.0), 0.0)

                # Envelope update
                if exp.fairness_envelope is not None:
                    try:
                        envelope_snapshot = _call_with_optional_kwargs(
                            exp.fairness_envelope.update_from_sfm,
                            sfm_payload=copy.deepcopy(sfm_payload),
                            timestep=i,
                        ) or {}
                    except Exception:
                        envelope_snapshot = {}
                exp.latest_envelope_snapshot = make_json_safe(envelope_snapshot)

                # Preserve legacy curves + add new curves
                bounded_append(exp.fairness_over_time, fairness_value)
                bounded_append(exp.fm_raw_over_time, fm_raw)
                bounded_append(exp.fm_smooth_over_time, fm_smooth)
                bounded_append(exp.fm_drift_over_time, fm_drift)
                bounded_append(exp.fm_uncertainty_over_time, fm_uncertainty)

                bounded_append(exp.fairness_stability_surface, fairness_stability)
                bounded_append(exp.equal_opportunity_over_time, safe_float(snap.get("equal_opportunity", 0.0), 0.0))
                bounded_append(exp.equalized_odds_over_time, safe_float(snap.get("equalized_odds_proxy", 0.0), 0.0))

                cum_fair = safe_float(snap.get("cumulative_ddsp", 0.0), 0.0)
                bounded_append(exp.cumulative_fairness_over_time, cum_fair)

                # Envelope surfaces
                if envelope_snapshot:
                    bounded_append(exp.envelope_lower_over_time, safe_float(envelope_snapshot.get("lower", 0.0), 0.0))
                    bounded_append(exp.envelope_upper_over_time, safe_float(envelope_snapshot.get("upper", 0.0), 0.0))
                    bounded_append(exp.envelope_width_over_time, safe_float(envelope_snapshot.get("width", 0.0), 0.0))
                    bounded_append(exp.envelope_severity_over_time, safe_float(envelope_snapshot.get("severity", 0.0), 0.0))
                    bounded_append(exp.envelope_violation_over_time, 1 if bool(envelope_snapshot.get("is_violation", False)) else 0)
                    bounded_append(exp.envelope_pressure_over_time, safe_float(envelope_snapshot.get("envelope_pressure", 0.0), 0.0))

                    if bool(envelope_snapshot.get("warning_event", False)):
                        exp.envelope_warning_event_points.append(i)
                    if bool(envelope_snapshot.get("violation_event", False)):
                        exp.envelope_violation_event_points.append(i)
                    if bool(envelope_snapshot.get("tightened", False)):
                        exp.envelope_tighten_event_points.append(i)
                    if bool(envelope_snapshot.get("relaxed", False)):
                        exp.envelope_relax_event_points.append(i)
                else:
                    bounded_append(exp.envelope_lower_over_time, 0.0)
                    bounded_append(exp.envelope_upper_over_time, 0.0)
                    bounded_append(exp.envelope_width_over_time, 0.0)
                    bounded_append(exp.envelope_severity_over_time, 0.0)
                    bounded_append(exp.envelope_violation_over_time, 0)
                    bounded_append(exp.envelope_pressure_over_time, 0.0)

                fstate = _call_with_optional_kwargs(exp.fairness_engine.get_fairness_state) or {}
                fair_drift_state = bool(fstate.get("is_drifting", False))

                residual = abs(float(yi) - float(y_prob))
                for s_key, s_val in active_sensitive_vals.items():
                    group_name = f"{s_key}: {s_val}"
                    if group_name not in exp.rise_residuals_by_group_raw and len(exp.rise_residuals_by_group_raw) >= MAX_GROUPS_RISE:
                        continue
                    exp.rise_residuals_by_group_raw.setdefault(group_name, [])
                    bounded_append(exp.rise_residuals_by_group_raw[group_name], residual, max_len=MAX_RISE_PER_GROUP)

                # Add SFM + DFE telemetry to controller
                controller_telemetry.update({
                    "fm_raw": float(fm_raw),
                    "fm_smooth": float(fm_smooth),
                    "fm_drift": float(fm_drift),
                    "fm_uncertainty": float(fm_uncertainty),
                    "sfm_payload": copy.deepcopy(sfm_payload),
                    "sfm": snap.get("sfm", (fm_smooth, fm_drift, fm_uncertainty)),

                    "fairness_envelope": copy.deepcopy(envelope_snapshot),
                    "envelope_lower": safe_float(envelope_snapshot.get("lower", 0.0), 0.0),
                    "envelope_upper": safe_float(envelope_snapshot.get("upper", 0.0), 0.0),
                    "envelope_width": safe_float(envelope_snapshot.get("width", 0.0), 0.0),
                    "envelope_violation": bool(envelope_snapshot.get("is_violation", False)),
                    "envelope_severity": safe_float(envelope_snapshot.get("severity", 0.0), 0.0),
                    "envelope_margin": safe_float(envelope_snapshot.get("envelope_margin", 0.0), 0.0),
                    "envelope_pressure": safe_float(envelope_snapshot.get("envelope_pressure", 0.0), 0.0),
                    "envelope_action_zone": str(envelope_snapshot.get("action_zone", "safe")),
                    "envelope_reason": str(envelope_snapshot.get("reason", "none")),
                    "envelope_trend_to_boundary": bool(envelope_snapshot.get("trend_to_boundary", False)),
                })

        except Exception:
            logger.debug("fairness_engine failure at step %s", i)
            bounded_append(exp.envelope_lower_over_time, 0.0)
            bounded_append(exp.envelope_upper_over_time, 0.0)
            bounded_append(exp.envelope_width_over_time, 0.0)
            bounded_append(exp.envelope_severity_over_time, 0.0)
            bounded_append(exp.envelope_violation_over_time, 0)
            bounded_append(exp.envelope_pressure_over_time, 0.0)

        bounded_append(exp.fairness_drift_state_over_time, 1 if fair_drift_state else 0)

        # ---------------- 3b) Stability Warning ----------------
        warning_state = fairness_stability > STABILITY_WARNING_THRESHOLD
        bounded_append(exp.warning_state_over_time, 1 if warning_state else 0)
        warning_event = warning_state and (not exp.last_warning_state)
        if warning_event:
            exp.warning_event_points.append(i)
        exp.last_warning_state = warning_state

        controller_telemetry["warning_event"] = bool(warning_event)
        controller_telemetry["perf_drift_event"] = bool(perf_drift_event)

        controller_telemetry["scope_state"] = copy.deepcopy(exp.scope_state)
        controller_telemetry["scope_mode"] = exp.fairness_scope_config.get("mode", "manual")
        controller_telemetry["active_sensitive_dict"] = copy.deepcopy(active_sensitive_vals)
        controller_telemetry["decision_context"] = copy.deepcopy(exp.decision_context)
        controller_telemetry["fairness_scope_snapshot"] = copy.deepcopy(exp.scope_snapshot)
        controller_telemetry["y_prob_debug_snapshot"] = copy.deepcopy(exp.y_prob_debug_snapshot)

        # ---------------- 4) Explainability ----------------
        try:
            explain_snapshot = _call_with_optional_kwargs(
                exp.explain_engine.update,
                model=model,
                x=xi_model,
                y_true=yi,
                sensitive=active_sensitive_vals,
                node_id=None,
                scope_state=copy.deepcopy(exp.scope_state),
                fairness_scope_snapshot=copy.deepcopy(exp.scope_snapshot),
                active_sensitive_dict=copy.deepcopy(active_sensitive_vals),
            ) or {}

            # optional control-state annotation
            if isinstance(explain_snapshot, dict):
                explain_snapshot["fairness_control_state"] = {
                    "fm_raw": float(fm_raw),
                    "fm_smooth": float(fm_smooth),
                    "fm_drift": float(fm_drift),
                    "fm_uncertainty": float(fm_uncertainty),
                    "envelope_action_zone": str(envelope_snapshot.get("action_zone", "safe")),
                    "envelope_violation": bool(envelope_snapshot.get("is_violation", False)),
                    "envelope_severity": safe_float(envelope_snapshot.get("severity", 0.0), 0.0),
                }

            exp.latest_explainability_snapshot = make_json_safe(explain_snapshot)

            drift_val = safe_float(explain_snapshot.get("global_drift", 0.0), 0.0)
            bounded_append(exp.explanation_drift_surface, drift_val)

            controller_telemetry["explainability_snapshot"] = exp.latest_explainability_snapshot
            controller_telemetry["explainability_drift"] = drift_val
        except Exception:
            bounded_append(exp.explanation_drift_surface, 0.0)

        # ---------------- 5) Topological Mapper ----------------
        try:
            _call_with_optional_kwargs(
                exp.topological_engine.add_record,
                x=xi_model,
                y_true=yi,
                y_pred=y_pred,
                prob=y_prob,
                sensitive=active_sensitive_vals,
                scope_state=copy.deepcopy(exp.scope_state),
                fairness_scope_snapshot=copy.deepcopy(exp.scope_snapshot),
                active_sensitive_dict=copy.deepcopy(active_sensitive_vals),
            )

            if i % MAPPER_UPDATE_FREQUENCY == 0 and i > 0:
                topo_snapshot = _call_with_optional_kwargs(exp.topological_engine.snapshot) or {}

                if isinstance(topo_snapshot, dict):
                    topo_snapshot["fairness_control_state"] = {
                        "fm_raw": float(fm_raw),
                        "fm_smooth": float(fm_smooth),
                        "fm_drift": float(fm_drift),
                        "fm_uncertainty": float(fm_uncertainty),
                        "envelope_lower": safe_float(envelope_snapshot.get("lower", 0.0), 0.0),
                        "envelope_upper": safe_float(envelope_snapshot.get("upper", 0.0), 0.0),
                        "envelope_action_zone": str(envelope_snapshot.get("action_zone", "safe")),
                        "envelope_violation": bool(envelope_snapshot.get("is_violation", False)),
                        "envelope_severity": safe_float(envelope_snapshot.get("severity", 0.0), 0.0),
                    }

                exp.latest_topology_snapshot = make_json_safe(topo_snapshot)

                if isinstance(topo_snapshot, dict) and topo_snapshot.get("node_metrics"):
                    exp.latest_mapper_graph = make_json_safe({
                        "node_metrics": topo_snapshot.get("node_metrics", []),
                        "edges": topo_snapshot.get("edges", []),
                        "bias_pockets": topo_snapshot.get("bias_pockets", []),
                        "hotspot_ranking": topo_snapshot.get("hotspot_ranking", []),
                        "structural_drift": topo_snapshot.get("structural_drift", 0.0),
                        "component_count": topo_snapshot.get("component_count", 0),
                        "lens_name": topo_snapshot.get("lens_name", "prob"),
                        "projection_info": topo_snapshot.get("projection_info", {}),
                        "attribute_metrics": topo_snapshot.get("attribute_metrics", {}),
                        "intersection_metrics": topo_snapshot.get("intersection_metrics", {}),
                        "active_scope": topo_snapshot.get("active_scope", {}),
                        "scope_snapshot": topo_snapshot.get("scope_snapshot", {}),
                        "fairness_control_state": topo_snapshot.get("fairness_control_state", {}),
                    })
        except Exception:
            logger.debug("topological_engine failure at step %s", i)

        if exp.latest_mapper_graph:
            try:
                controller_telemetry["latest_mapper_graph"] = exp.latest_mapper_graph
                controller_telemetry["topo_structural_drift"] = safe_float(exp.latest_mapper_graph.get("structural_drift", 0.0), 0.0)
                hr = exp.latest_mapper_graph.get("hotspot_ranking", [])
                if isinstance(hr, list) and hr:
                    controller_telemetry["topo_hotspot_score"] = safe_float(
                        max(safe_float(x.get("hotspot_score", 0.0), 0.0) for x in hr if isinstance(x, dict)),
                        0.0,
                    )
            except Exception:
                pass

        controller_telemetry["fairness_snapshot"] = copy.deepcopy(exp.latest_fairness_snapshot)
        controller_telemetry["metrics_snapshot"] = copy.deepcopy(exp.latest_metrics_snapshot)
        controller_telemetry["topology_snapshot"] = copy.deepcopy(exp.latest_topology_snapshot)
        controller_telemetry["envelope_snapshot"] = copy.deepcopy(exp.latest_envelope_snapshot)

        try:
            _call_with_optional_kwargs(
                exp.model.set_fairness_context,
                {
                    "fm_smooth": float(fm_smooth),
                    "fm_drift": float(fm_drift),
                    "fm_uncertainty": float(fm_uncertainty),
                    "envelope_violation": bool(envelope_snapshot.get("is_violation", False)),
                    "envelope_severity": safe_float(envelope_snapshot.get("severity", 0.0), 0.0),
                    "envelope_pressure": safe_float(envelope_snapshot.get("envelope_pressure", 0.0), 0.0),
                    "trend_to_boundary": bool(envelope_snapshot.get("trend_to_boundary", False)),
                    "confidence": safe_float(sfm_payload.get("confidence", 0.0), 0.0),
                    "is_reliable": bool(sfm_payload.get("is_reliable", False)),
                    "effective_support": safe_int(sfm_payload.get("effective_support", 0), 0),
                    "topo_hotspot_score": safe_float(controller_telemetry.get("topo_hotspot_score", 0.0), 0.0),
                    "topo_structural_drift": safe_float(controller_telemetry.get("topo_structural_drift", 0.0), 0.0),
                },
            )
        except Exception:
            logger.debug("model fairness-context sync failure at step %s", i)

        # ---------------- 6) RL Control ----------------
        try:
            streaming_method = str(exp.optimizer_config.get("streaming_method", "online_fairbo_ensemble")).strip().lower()
            if streaming_method == "online_fairbo_basic":
                exp.active_mitigation_state = {
                    "action_taken": "monitor_only",
                    "reason": "basic_mode_monitoring",
                    "weighting_active": False,
                    "scope_state": copy.deepcopy(exp.scope_state),
                    "active_scope": copy.deepcopy(exp.decision_context.get("active_scope", {})),
                    "fairness_control_state": {
                        "fm_raw": float(fm_raw),
                        "fm_smooth": float(fm_smooth),
                        "fm_drift": float(fm_drift),
                        "fm_uncertainty": float(fm_uncertainty),
                    },
                    "fairness_envelope": copy.deepcopy(exp.latest_envelope_snapshot),
                }
                exp.decision_context = _build_decision_context(
                    xi_full=xi_full,
                    all_sensitive_dict=all_sensitive_vals,
                    active_sensitive_dict=active_sensitive_vals,
                    scope_state=exp.scope_state,
                    scope_snapshot=exp.scope_snapshot,
                    active_mitigation=exp.active_mitigation_state,
                )
            else:
                new_model, info = _call_with_optional_kwargs(
                    controller.evaluate_and_adapt,
                    model=model,
                    timestep=i,
                    current_accuracy=acc,
                    fairness_metric=fm_smooth,              # SFM-aligned control signal
                    fairness_stability=fairness_stability,
                    perf_drift_state=perf_drift_state,
                    fairness_drift_state=fair_drift_state,
                    warning_state=warning_state,
                    telemetry=controller_telemetry,
                )
                exp.model = new_model

                if info and info.get("adapted"):
                    info_meta = info.get("metadata") or {}
                    info_params = info_meta.get("params", {}) if isinstance(info_meta, dict) else {}
                    exp.active_mitigation_state = {
                        **info_meta,
                        "action_taken": info.get("action_taken"),
                        "reason": info.get("reason"),
                        "scope_state": copy.deepcopy(exp.scope_state),
                        "active_scope": copy.deepcopy(exp.decision_context.get("active_scope", {})),
                        "fairness_control_state": {
                            "fm_raw": float(fm_raw),
                            "fm_smooth": float(fm_smooth),
                            "fm_drift": float(fm_drift),
                            "fm_uncertainty": float(fm_uncertainty),
                        },
                        "fairness_envelope": copy.deepcopy(exp.latest_envelope_snapshot),
                    }
                    if isinstance(info_params, dict) and "weighting_active" in info_params:
                        exp.active_mitigation_state["weighting_active"] = bool(info_params.get("weighting_active"))

                    exp.decision_context = _build_decision_context(
                        xi_full=xi_full,
                        all_sensitive_dict=all_sensitive_vals,
                        active_sensitive_dict=active_sensitive_vals,
                        scope_state=exp.scope_state,
                        scope_snapshot=exp.scope_snapshot,
                        active_mitigation=exp.active_mitigation_state,
                    )

                    logger.info("RL Action Taken: %s at step %s", info.get("action_taken"), i)
        except Exception:
            logger.debug("controller failure at step %s", i)

        # ---------------- 7) LLM Auditor Trigger ----------------
        try:
            time_since_last_llm = i - exp.last_llm_trigger_step
            should_trigger_llm = bool(
                fair_drift_event
                or perf_drift_event
                or bool(envelope_snapshot.get("violation_event", False))
                or (i % MAPPER_UPDATE_FREQUENCY == 0 and time_since_last_llm > LLM_COOLDOWN)
            )

            if should_trigger_llm and not exp.llm_is_processing:
                exp.llm_is_processing = True
                exp.last_llm_trigger_step = i

                recent_fair_drift_events = copy.deepcopy(exp.fairness_drift_event_points[-5:]) if exp.fairness_drift_event_points else []
                active_mit_copy = copy.deepcopy(exp.active_mitigation_state)

                threading.Thread(
                    target=async_llm_audit,
                    args=(
                        exp,
                        i,
                        copy.deepcopy(exp.latest_fairness_snapshot),
                        copy.deepcopy(exp.latest_metrics_snapshot),
                        copy.deepcopy(exp.latest_topology_snapshot),
                        copy.deepcopy(exp.latest_explainability_snapshot),
                        copy.deepcopy(exp.latest_envelope_snapshot),
                        active_mit_copy,
                        recent_fair_drift_events,
                        acc,
                    ),
                    daemon=True,
                ).start()
        except Exception:
            logger.debug("LLM trigger failure at step %s", i)

    _run_optimizer_cycle(exp, end)
    exp.carbon.end_chunk()


# ============================================================
# Process chunk
# ============================================================

def process_experiment(exp_id: str, chunk_size: int = DEFAULT_CHUNK_SIZE) -> Tuple[bool, Dict[str, Any]]:
    with EXPERIMENTS_LOCK:
        exp = EXPERIMENTS.get(exp_id)
    if exp is None:
        raise KeyError("invalid_id")

    with exp.state_lock:
        exp.last_poll_ts = _now()
        exp.last_error = ""
        if exp.index >= exp.sample_budget:
            exp.carbon.finalize()
            if exp.last_response_cache:
                return True, copy.deepcopy(exp.last_response_cache)
            resp_data = make_json_safe(build_response(exp, True))
            exp.last_response_cache = copy.deepcopy(resp_data)
            return True, resp_data

    if not exp.compute_lock.acquire(timeout=0.0):
        with exp.state_lock:
            if exp.last_response_cache:
                return False, copy.deepcopy(exp.last_response_cache)
            resp_data = make_json_safe(build_response(exp, False))
            exp.last_response_cache = copy.deepcopy(resp_data)
            return False, resp_data

    try:
        with exp.state_lock:
            if exp.index >= exp.sample_budget:
                exp.carbon.finalize()
                if exp.last_response_cache:
                    return True, copy.deepcopy(exp.last_response_cache)
                resp_data = make_json_safe(build_response(exp, True))
                exp.last_response_cache = copy.deepcopy(resp_data)
                return True, resp_data

            start = int(exp.index)
            end = int(min(start + int(chunk_size), exp.sample_budget))

        evaluate_stream(exp, start, end)

        with exp.state_lock:
            exp.index = end
            finished = exp.index >= exp.sample_budget
            if finished:
                exp.carbon.finalize()
        resp_data = make_json_safe(build_response(exp, finished))
        with exp.state_lock:
            exp.last_response_cache = copy.deepcopy(resp_data)
        return bool(finished), resp_data

    finally:
        exp.compute_lock.release()


# ============================================================
# Response builder
# ============================================================

def build_response(exp: ExperimentState, finished: bool) -> Dict[str, Any]:
    rise_sorted: Dict[str, List[float]] = {}
    for group, residuals in list(exp.rise_residuals_by_group_raw.items()):
        try:
            rise_sorted[group] = sorted(list(residuals))
        except Exception:
            rise_sorted[group] = list(residuals)

    controller = exp.controller
    fairness_metric_snapshot = copy.deepcopy(exp.latest_metrics_snapshot) if isinstance(exp.latest_metrics_snapshot, dict) else None
    scope_snapshot = copy.deepcopy(exp.scope_snapshot) if isinstance(exp.scope_snapshot, dict) else {}
    fairness_snapshot = copy.deepcopy(exp.latest_fairness_snapshot) if isinstance(exp.latest_fairness_snapshot, dict) else {}
    envelope_snapshot = copy.deepcopy(exp.latest_envelope_snapshot) if isinstance(exp.latest_envelope_snapshot, dict) else {}
    optimizer_get_state = getattr(exp.optimizer, "get_state", None)
    model_get_state = getattr(exp.model, "get_system_state", None)
    optimizer_state = make_json_safe(_call_with_optional_kwargs(optimizer_get_state)) if optimizer_get_state is not None else {}
    autostream_state = make_json_safe(_call_with_optional_kwargs(model_get_state)) if model_get_state is not None else {}

    resp: Dict[str, Any] = {
        "status": "ok",
        "schema_version": SCHEMA_VERSION,
        "backend_version": BACKEND_VERSION,
        "experiment_id": exp.exp_id,
        "dataset_name": exp.dataset_name,

        "processed_steps": int(exp.index),
        "sample_budget": int(exp.sample_budget),
        "finished": bool(finished),

        # Core curves
        "accuracy_over_time": list(exp.accuracy_over_time),
        "fairness_over_time": list(exp.fairness_over_time),
        "cumulative_fairness_over_time": list(exp.cumulative_fairness_over_time),
        "fairness_stability_surface": list(exp.fairness_stability_surface),
        "equal_opportunity_over_time": list(exp.equal_opportunity_over_time),
        "equalized_odds_over_time": list(exp.equalized_odds_over_time),
        "explanation_drift_surface": list(exp.explanation_drift_surface),

        # New SFM curves
        "fm_raw_over_time": list(exp.fm_raw_over_time),
        "fm_smooth_over_time": list(exp.fm_smooth_over_time),
        "fm_drift_over_time": list(exp.fm_drift_over_time),
        "fm_uncertainty_over_time": list(exp.fm_uncertainty_over_time),
        "raw_fairness_over_time": list(exp.fm_raw_over_time),
        "smoothed_fairness_over_time": list(exp.fm_smooth_over_time),
        "fairness_uncertainty_over_time": list(exp.fm_uncertainty_over_time),

        # New DFE curves
        "envelope_lower_over_time": list(exp.envelope_lower_over_time),
        "envelope_upper_over_time": list(exp.envelope_upper_over_time),
        "envelope_width_over_time": list(exp.envelope_width_over_time),
        "envelope_severity_over_time": list(exp.envelope_severity_over_time),
        "envelope_violation_over_time": list(exp.envelope_violation_over_time),
        "envelope_pressure_over_time": list(exp.envelope_pressure_over_time),

        # Latest SFM/DFE fields
        "fm_raw": safe_float(fairness_snapshot.get("fm_raw", 0.0), 0.0),
        "fm_smooth": safe_float(fairness_snapshot.get("fm_smooth", 0.0), 0.0),
        "fm_drift": safe_float(fairness_snapshot.get("fm_drift", 0.0), 0.0),
        "fm_uncertainty": safe_float(fairness_snapshot.get("fm_uncertainty", 1.0), 1.0),
        "raw_fairness": safe_float(fairness_snapshot.get("raw_fairness", fairness_snapshot.get("fm_raw", 0.0)), 0.0),
        "smoothed_fairness": safe_float(fairness_snapshot.get("smoothed_fairness", fairness_snapshot.get("fm_smooth", 0.0)), 0.0),
        "fairness_uncertainty": safe_float(
            fairness_snapshot.get("fairness_uncertainty", fairness_snapshot.get("fm_uncertainty", 1.0)),
            1.0,
        ),
        "sfm": fairness_snapshot.get("sfm", None),
        "sfm_payload": fairness_snapshot.get("sfm_payload", {}),
        "sfm_state": copy.deepcopy(exp.latest_sfm_state),

        "fairness_envelope": envelope_snapshot,
        "latest_envelope_snapshot": envelope_snapshot,
        "envelope_lower": safe_float(envelope_snapshot.get("envelope_lower", envelope_snapshot.get("lower", 0.0)), 0.0),
        "envelope_upper": safe_float(envelope_snapshot.get("envelope_upper", envelope_snapshot.get("upper", 0.0)), 0.0),
        "envelope_width": safe_float(envelope_snapshot.get("envelope_width", envelope_snapshot.get("width", 0.0)), 0.0),
        "envelope_violation": bool(envelope_snapshot.get("envelope_violation", envelope_snapshot.get("is_violation", False))),
        "envelope_severity": safe_float(envelope_snapshot.get("envelope_severity", envelope_snapshot.get("severity", 0.0)), 0.0),
        "envelope_action_zone": str(envelope_snapshot.get("envelope_action_zone", envelope_snapshot.get("action_zone", "safe"))),
        "envelope_reason": str(envelope_snapshot.get("envelope_reason", envelope_snapshot.get("reason", "none"))),
        "envelope_margin": safe_float(envelope_snapshot.get("envelope_margin", 0.0), 0.0),
        "envelope_pressure": safe_float(envelope_snapshot.get("envelope_pressure", 0.0), 0.0),
        "margin_to_upper": safe_float(envelope_snapshot.get("margin_to_upper", 0.0), 0.0),
        "margin_to_lower": safe_float(envelope_snapshot.get("margin_to_lower", 0.0), 0.0),
        "closest_boundary": str(envelope_snapshot.get("closest_boundary", "upper")),
        "lower_bound_enforced": bool(envelope_snapshot.get("lower_bound_enforced", False)),

        # Events vs State
        "fairness_drift_event_points": list(exp.fairness_drift_event_points),
        "warning_event_points": list(exp.warning_event_points),
        "perf_drift_event_points": list(exp.perf_drift_event_points),
        "perf_drift_monitor_event_points": list(exp.perf_drift_monitor_event_points),

        # New DFE events
        "envelope_warning_event_points": list(exp.envelope_warning_event_points),
        "envelope_violation_event_points": list(exp.envelope_violation_event_points),
        "envelope_tighten_event_points": list(exp.envelope_tighten_event_points),
        "envelope_relax_event_points": list(exp.envelope_relax_event_points),

        "total_fairness_drift_events": int(len(exp.fairness_drift_event_points)),
        "total_perf_drift_events": int(len(exp.perf_drift_event_points)),
        "total_perf_drift_monitor_events": int(len(exp.perf_drift_monitor_event_points)),
        "total_warning_events": int(len(exp.warning_event_points)),
        "total_envelope_warning_events": int(len(exp.envelope_warning_event_points)),
        "total_envelope_violation_events": int(len(exp.envelope_violation_event_points)),

        "fairness_drift_state_over_time": list(exp.fairness_drift_state_over_time),
        "warning_state_over_time": list(exp.warning_state_over_time),
        "perf_drift_state_over_time": list(exp.perf_drift_state_over_time),
        "perf_drift_monitor_state_over_time": list(exp.perf_drift_monitor_state_over_time),

        # Scope automation
        "fairness_scope_config": copy.deepcopy(exp.fairness_scope_config),
        "scope_state": copy.deepcopy(exp.scope_state),
        "fairness_scope_snapshot": copy.deepcopy(scope_snapshot),
        "decision_context": copy.deepcopy(exp.decision_context),

        # Latest module snapshots
        "latest_fairness_snapshot": fairness_snapshot,
        "fairness_metric_snapshot": fairness_metric_snapshot,
        "latest_explainability_snapshot": copy.deepcopy(exp.latest_explainability_snapshot),
        "latest_topology_snapshot": copy.deepcopy(exp.latest_topology_snapshot),
        "config_space_summary": copy.deepcopy(exp.config_space_summary),
        "candidate_evaluator_config": copy.deepcopy(exp.candidate_evaluator_config),
        "optimizer_config": copy.deepcopy(exp.optimizer_config),
        "optimizer_runtime_config": copy.deepcopy(exp.optimizer_runtime_config),
        "osmac_optimizer_state": optimizer_state,
        "optimizer_state": optimizer_state,
        "osmac_incumbent": optimizer_state.get("incumbent") if isinstance(optimizer_state, dict) else None,
        "optimizer_incumbent": optimizer_state.get("incumbent") if isinstance(optimizer_state, dict) else None,
        "latest_optimizer_batch": copy.deepcopy(exp.latest_optimizer_batch),
        "latest_candidate_evaluations": copy.deepcopy(exp.latest_candidate_evaluations),
        "optimizer_candidate_history": copy.deepcopy(exp.optimizer_candidate_history),
        "latest_live_handoff": copy.deepcopy(exp.latest_live_handoff),
        "latest_optimizer_error": str(exp.latest_optimizer_error or ""),
        "static_benchmark_config": copy.deepcopy(exp.static_benchmark_config),
        "static_benchmark_results": copy.deepcopy(exp.static_benchmark_results),
        "static_benchmark_error": str(exp.static_benchmark_error or ""),
        "static_benchmark_status": str(exp.static_benchmark_status or "disabled"),
        "static_benchmark_started_at": safe_float(exp.static_benchmark_started_at, 0.0),
        "static_benchmark_finished_at": safe_float(exp.static_benchmark_finished_at, 0.0),
        "autostream_state": autostream_state,

        # Debug / diagnostics
        "feature_manifest": copy.deepcopy(exp.feature_manifest),
        "y_prob_debug_snapshot": copy.deepcopy(exp.y_prob_debug_snapshot),

        # RISE
        "rise_residuals_by_group": rise_sorted,

        # Mapper
        "latest_mapper_graph": exp.latest_mapper_graph,
        "mapper_config": copy.deepcopy(exp.mapper_config),

        # LLM
        "llm_auditor_output": copy.deepcopy(exp.llm_auditor_output),

        # RL
        "lambda_history": controller.get_lambda_history() if controller else [],
        "reward_decomposition": controller.get_reward_decomposition() if controller else [],
        "adaptation_log": controller.get_adaptation_log() if controller else [],
        "active_mitigation": exp.active_mitigation_state,

        # Carbon
        "carbon_available": bool(exp.carbon.enabled),
        "co2_total_kg": safe_float(exp.carbon.co2_total_kg),
        "energy_total_kwh": safe_float(exp.carbon.energy_total_kwh),
        "co2_surface_kg": list(exp.carbon.co2_surface_kg),
        "energy_surface_kwh": list(exp.carbon.energy_surface_kwh),

        # Runtime + meta
        "runtime_sec": safe_float(_now() - exp.start_time),
        "created_at": safe_float(exp.created_at),
        "last_poll_ts": safe_float(exp.last_poll_ts),
        "last_error": str(exp.last_error or ""),
    }
    return resp


# ============================================================
# Mapper node explanation helpers
# ============================================================

def _format_pct(v: Any) -> str:
    return f"{100.0 * safe_float(v, 0.0):.1f}%"


def _build_node_analytic_explanation(
    node: Dict[str, Any],
    graph: Dict[str, Any],
    current_accuracy: float,
    current_fairness: float,
) -> str:
    size = safe_int(node.get("size", 0), 0)
    fair_gap = safe_float(node.get("fairness_gap", node.get("prob_parity_gap", 0.0)), 0.0)
    err = safe_float(node.get("error_rate", 0.0), 0.0)
    err_gap = safe_float(node.get("error_gap", 0.0), 0.0)
    theil = safe_float(node.get("theil_index", 0.0), 0.0)
    hotspot = safe_float(node.get("hotspot_score", 0.0), 0.0)

    group_dist = node.get("group_distribution", {}) if isinstance(node.get("group_distribution"), dict) else {}
    top_groups = sorted(group_dist.items(), key=lambda kv: safe_float(kv[1], 0.0), reverse=True)[:2]
    group_txt = ", ".join([f"{k}={_format_pct(v)}" for k, v in top_groups]) if top_groups else "n/a"

    s_feats = node.get("salient_features", []) if isinstance(node.get("salient_features"), list) else []
    feat_txt = ", ".join(
        f"{f.get('feature')} (z={safe_float(f.get('z_shift', 0.0), 0.0):.2f})"
        for f in s_feats[:3]
        if isinstance(f, dict)
    ) or "n/a"

    risk = "high" if (fair_gap >= 0.15 or err_gap >= 0.15) else ("moderate" if (fair_gap >= 0.08 or err_gap >= 0.08) else "low")

    return (
        f"Node {safe_int(node.get('node_id', -1), -1)} has {size} samples and {risk} fairness-risk profile. "
        f"Within-node fairness gap={fair_gap:.4f}, error_rate={err:.4f}, error_gap={err_gap:.4f}, theil={theil:.4f}, hotspot={hotspot:.4f}. "
        f"Dominant groups: {group_txt}. Top shifted features: {feat_txt}. "
        f"Global context now: accuracy={current_accuracy:.4f}, fairness_gap={current_fairness:.4f}, structural_drift={safe_float(graph.get('structural_drift', 0.0), 0.0):.4f}."
    )


# ============================================================
# Routes
# ============================================================

@app.before_request
def _pre() -> None:
    _cleanup_expired_experiments()


@app.route("/health", methods=["GET"])
def health():
    return jsonify(make_json_safe({
        "status": "ok",
        "schema_version": SCHEMA_VERSION,
        "backend_version": BACKEND_VERSION,
        "experiments_active": len(EXPERIMENTS),
        "codecarbon_available": bool(CODECARBON_AVAILABLE),
        "python": platform.python_version(),
        "platform": platform.platform(),
        "ttl_seconds": EXPERIMENT_TTL_SECONDS,
    }))


@app.route("/experiments", methods=["GET"])
def list_experiments():
    with EXPERIMENTS_LOCK:
        ids = list(EXPERIMENTS.keys())
    return jsonify({
        "status": "ok",
        "schema_version": SCHEMA_VERSION,
        "backend_version": BACKEND_VERSION,
        "experiments": ids,
        "count": len(ids),
    })


@app.route("/experiment_meta/<exp_id>", methods=["GET"])
def experiment_meta(exp_id: str):
    with EXPERIMENTS_LOCK:
        exp = EXPERIMENTS.get(exp_id)
    if exp is None:
        return _err_payload("invalid_id", "Experiment not found", 404)

    with exp.state_lock:
        optimizer_get_state = getattr(exp.optimizer, "get_state", None)
        optimizer_state = make_json_safe(_call_with_optional_kwargs(optimizer_get_state)) if optimizer_get_state is not None else {}
        meta = {
            "status": "ok",
            "schema_version": SCHEMA_VERSION,
            "backend_version": BACKEND_VERSION,
            "experiment_id": exp.exp_id,
            "dataset_name": exp.dataset_name,
            "created_at": exp.created_at,
            "last_poll_ts": exp.last_poll_ts,
            "processed_steps": exp.index,
            "sample_budget": exp.sample_budget,
            "target": exp.target,
            "sensitive_attr": exp.sensitive_attr,
            "primary_sensitive_attr": exp.primary_sensitive_attr,
            "window_size": exp.window_size,
            "mapper_config": copy.deepcopy(exp.mapper_config),
            "fairness_scope_config": copy.deepcopy(exp.fairness_scope_config),
            "scope_state": copy.deepcopy(exp.scope_state),
            "fairness_scope_snapshot": copy.deepcopy(exp.scope_snapshot),
            "feature_manifest": copy.deepcopy(exp.feature_manifest),
            "config_space_summary": copy.deepcopy(exp.config_space_summary),
            "optimizer_runtime_config": copy.deepcopy(exp.optimizer_runtime_config),
            "candidate_evaluator_config": copy.deepcopy(exp.candidate_evaluator_config),
            "optimizer_config": copy.deepcopy(exp.optimizer_config),
            "osmac_optimizer_state": optimizer_state,
            "latest_optimizer_batch": copy.deepcopy(exp.latest_optimizer_batch),
            "latest_candidate_evaluations": copy.deepcopy(exp.latest_candidate_evaluations),
            "optimizer_candidate_history": copy.deepcopy(exp.optimizer_candidate_history),
            "latest_live_handoff": copy.deepcopy(exp.latest_live_handoff),
            "latest_optimizer_error": str(exp.latest_optimizer_error or ""),
            "static_benchmark_config": copy.deepcopy(exp.static_benchmark_config),
            "static_benchmark_results": copy.deepcopy(exp.static_benchmark_results),
            "static_benchmark_error": str(exp.static_benchmark_error or ""),
            "static_benchmark_status": str(exp.static_benchmark_status or "disabled"),
            "static_benchmark_started_at": safe_float(exp.static_benchmark_started_at, 0.0),
            "static_benchmark_finished_at": safe_float(exp.static_benchmark_finished_at, 0.0),
            "y_prob_debug_snapshot": copy.deepcopy(exp.y_prob_debug_snapshot),
            "latest_envelope_snapshot": copy.deepcopy(exp.latest_envelope_snapshot),
            "last_error": exp.last_error,
        }
    return jsonify(make_json_safe(meta))


@app.route("/explain_mapper_node/<exp_id>", methods=["POST"])
def explain_mapper_node(exp_id: str):
    with EXPERIMENTS_LOCK:
        exp = EXPERIMENTS.get(exp_id)
    if exp is None:
        return _err_payload("invalid_id", "Experiment not found", 404)

    payload = request.get_json(silent=True) or {}
    node_id = safe_int(payload.get("node_id", -1), -1)
    use_llm = bool(payload.get("use_llm", True))
    if node_id < 0:
        return _err_payload("bad_node_id", "node_id must be a non-negative integer", 400)

    with exp.state_lock:
        graph = copy.deepcopy(exp.latest_mapper_graph) if isinstance(exp.latest_mapper_graph, dict) else {}
        current_accuracy = safe_float(exp.accuracy_over_time[-1], 0.0) if exp.accuracy_over_time else 0.0
        current_fairness = safe_float(exp.fm_smooth_over_time[-1], 0.0) if exp.fm_smooth_over_time else (
            safe_float(exp.fairness_over_time[-1], 0.0) if exp.fairness_over_time else 0.0
        )
        mapper_nodes = graph.get("node_metrics", []) if isinstance(graph, dict) else []
        node = None
        if isinstance(mapper_nodes, list):
            for n in mapper_nodes:
                if isinstance(n, dict) and safe_int(n.get("node_id", -1), -1) == node_id:
                    node = copy.deepcopy(n)
                    break

    if not isinstance(graph, dict) or not graph:
        return _err_payload("mapper_unavailable", "Mapper graph not ready yet", 409)
    if not isinstance(node, dict):
        return _err_payload("node_not_found", "Requested node_id not found in latest mapper snapshot", 404)

    analytic = _build_node_analytic_explanation(node, graph, current_accuracy, current_fairness)

    llm_out = None
    if use_llm and exp.llm_auditor is not None:
        try:
            llm_out = exp.llm_auditor.generate_node_explanation(
                node_metrics=node,
                global_context={
                    "current_accuracy": current_accuracy,
                    "current_fairness_gap": current_fairness,
                    "structural_drift": safe_float(graph.get("structural_drift", 0.0), 0.0),
                    "lens_name": graph.get("lens_name", "prob"),
                    "active_scope": exp.latest_fairness_snapshot.get("active_scope", {}),
                    "attribute_metrics": graph.get("attribute_metrics", {}),
                    "intersection_metrics": graph.get("intersection_metrics", {}),
                    "fairness_envelope": exp.latest_envelope_snapshot,
                },
            )
        except Exception as e:
            llm_out = {"status": "error", "response": f"LLM explain failed: {e}"}

    return jsonify(make_json_safe({
        "status": "ok",
        "schema_version": SCHEMA_VERSION,
        "backend_version": BACKEND_VERSION,
        "experiment_id": exp.exp_id,
        "node_id": node_id,
        "analytic_explanation": analytic,
        "node_metrics": node,
        "llm": llm_out,
    }))


@app.route("/start_experiment", methods=["POST"])
def start_experiment():
    payload = request.get_json(silent=True)
    if payload is None:
        return _err_payload("bad_json", "Request body must be valid JSON", 400)

    ok, reason, exp_id, details = create_experiment(payload)
    if not ok or exp_id is None:
        return _err_payload("init_failed", "Failed to initialize experiment", 400, {"reason": reason, **(details or {})})

    return jsonify({
        "status": "ok",
        "schema_version": SCHEMA_VERSION,
        "backend_version": BACKEND_VERSION,
        "experiment_id": exp_id,
        "codecarbon_available": bool(CODECARBON_AVAILABLE),
    })


@app.route("/poll_experiment/<exp_id>", methods=["GET"])
def poll_experiment(exp_id: str):
    with EXPERIMENTS_LOCK:
        exp = EXPERIMENTS.get(exp_id)
    if exp is None:
        return _err_payload("invalid_id", "Experiment not found", 404)

    try:
        chunk_size = safe_int(request.args.get("chunk_size", DEFAULT_CHUNK_SIZE), DEFAULT_CHUNK_SIZE)
        chunk_size = max(1, min(5000, int(chunk_size)))

        finished, resp_data = process_experiment(exp_id, chunk_size=chunk_size)
        return jsonify(make_json_safe(resp_data))

    except KeyError:
        return _err_payload("invalid_id", "Experiment not found", 404)
    except Exception as e:
        logger.error("Polling error: %s", e)
        logger.error(traceback.format_exc())
        with exp.state_lock:
            exp.last_error = str(e)
        return _err_payload("poll_failed", "Internal server error during polling", 500, {"exception": str(e)})


@app.route("/finalize_experiment/<exp_id>", methods=["GET"])
def finalize_experiment(exp_id: str):
    with EXPERIMENTS_LOCK:
        exp = EXPERIMENTS.pop(exp_id, None)
    if exp is None:
        return _err_payload("invalid_id", "Experiment not found", 404)

    with exp.state_lock:
        exp.carbon.finalize()
        final = build_response(exp, True)
        final.update({"accuracy": safe_float(exp.accuracy_metric.get())})
    return jsonify(make_json_safe(final))


@app.route("/reset_experiment/<exp_id>", methods=["POST"])
def reset_experiment(exp_id: str):
    with EXPERIMENTS_LOCK:
        removed = EXPERIMENTS.pop(exp_id, None)
    if removed is None:
        return _err_payload("invalid_id", "Experiment not found", 404)
    return jsonify({
        "status": "ok",
        "schema_version": SCHEMA_VERSION,
        "backend_version": BACKEND_VERSION,
        "experiment_id": exp_id,
    })


if __name__ == "__main__":
    app.run(debug=True, use_reloader=False, port=5000, threaded=True)