pcb-afm-simulation-system/web/backend/app/services/plan_generator.py

"""AI Plan Generator (P3-M3).

Converts natural language requirements into structured simulation plans
using Kimi k3 model, with L0 pre-screening validation.

Output is converted to unified plan schema v2.0 compatible with
src/plan_schema.py (fixed_params + variables with value lists).
"""
import json
import math
from pathlib import Path
from typing import Dict, List, Optional, Any, Tuple

from ..config import PROMPTS_DIR, KIMI_MAX_TOKENS
from ..services.ai_client import get_kimi_client
from ..services.l0_prescreening import L0PreScreeningEngine
from ..services.rule_engine import SCAN_PARAMETERS, get_parameter, recommend_range, BoundaryConditions
from ..services.fixed_params_template import build_default_fixed_params, get_fixed_param_template, CATEGORY_CN

# Single-source strategy registry (mirrors src/plan_schema.py validation) so the
# AI-generated method string is normalized against the platform registry instead
# of being trusted verbatim. Avoids a second definition of the strategy whitelist.
from src.afmcore.strategies import (
    is_registered as _strategy_registered,
    normalize_method as _strategy_normalize,
)


# Mapping from common AI-output variable names to exact Motor-CAD variable names
_VARIABLE_NAME_MAP = {
    "airgap": "Airgap",
    "airgap_mm": "Airgap",
    "air_gap": "Airgap",
    "air_gap_mm": "Airgap",
    "airgap_length": "Airgap",
    "airgap_length_mm": "Airgap",
    "air_gap_length": "Airgap",
    "air_gap_length_mm": "Airgap",
    "gap_length": "Airgap",
    "gap_length_mm": "Airgap",
    "magnet_length": "Magnet_Length",
    "magnet_length_mm": "Magnet_Length",
    "magnet_axial_thickness": "Magnet_Length",
    "magnet_axial_thickness_mm": "Magnet_Length",
    "magnet_axial_length": "Magnet_Length",
    "magnet_axial_length_mm": "Magnet_Length",
    "magnet_thickness": "Magnet_Thickness",
    "magnet_thickness_mm": "Magnet_Thickness",
    "magnet_radial_depth": "Magnet_Thickness",
    "magnet_radial_depth_mm": "Magnet_Thickness",
    "magnet_radial_thickness": "Magnet_Thickness",
    "magnet_radial_thickness_mm": "Magnet_Thickness",
    "magnet_arc": "Magnet_Arc_[ED]",
    "magnet_arc_deg": "Magnet_Arc_[ED]",
    "magnet_arc_[ed]": "Magnet_Arc_[ED]",
    "magnet_pole_arc": "Magnet_Arc_[ED]",
    "magnet_pole_arc_ratio": "Magnet_Arc_[ED]",
    "pole_arc": "Magnet_Arc_[ED]",
    "pole_arc_deg": "Magnet_Arc_[ED]",
    "pole_arc_ratio": "Magnet_Arc_[ED]",
    "pole_arc_coefficient": "Magnet_Arc_[ED]",
    "rms_current": "RMSCurrent",
    "rms_current_a": "RMSCurrent",
    "current_a": "RMSCurrent",
    "current": "RMSCurrent",
    "phase_current": "RMSCurrent",
    "phase_current_a": "RMSCurrent",
    "rated_current": "RMSCurrent",
    "rated_current_a": "RMSCurrent",
    "shaft_speed": "Shaft_Speed",
    "shaft_speed_rpm": "Shaft_Speed",
    "speed_rpm": "Shaft_Speed",
    "speed": "Shaft_Speed",
    "rotational_speed": "Shaft_Speed",
    "rotational_speed_rpm": "Shaft_Speed",
    "rated_speed": "Shaft_Speed",
    "rated_speed_rpm": "Shaft_Speed",
    "magnet_temperature": "Magnet_Temperature",
    "magnet_temperature_c": "Magnet_Temperature",
    "magnet_temp_c": "Magnet_Temperature",
    "magnet_temp": "Magnet_Temperature",
    "torque_points_per_cycle": "TorquePointsPerCycle",
    "torque_points": "TorquePointsPerCycle",
    "sampling_points": "TorquePointsPerCycle",
    "torque_sampling_points": "TorquePointsPerCycle",
    # Geometry / winding variants the AI commonly emits (normalized to the
    # canonical names registered in rule_engine.SCAN_PARAMETERS).
    "stator_outer_diameter": "Stator_Outer_Diameter",
    "stator_outer_diameter_mm": "Stator_Outer_Diameter",
    "stator_outer_dia": "Stator_Outer_Diameter",
    "stator_od": "Stator_Outer_Diameter",
    "stator_lam_outer_dia": "Stator_Outer_Diameter",
    "stator_lam_dia_outer": "Stator_Outer_Diameter",
    "stator_lam_outside_dia": "Stator_Outer_Diameter",
    "stator_lamination_outer_dia": "Stator_Outer_Diameter",
    "stator_inner_diameter": "Stator_Inner_Diameter",
    "stator_inner_dia": "Stator_Inner_Diameter",
    "stator_lam_inner_dia": "Stator_Inner_Diameter",
    "stator_id": "Stator_Inner_Diameter",
    "slot_depth": "Slot_Depth",
    "slot_depth_mm": "Slot_Depth",
    "stator_slot_depth": "Slot_Depth",
    "turns_per_coil": "Turns_per_Coil",
    "coil_turns": "Turns_per_Coil",
    "turnspercoil": "Turns_per_Coil",
    "coil_turns_per_phase": "Turns_per_Coil",
    "turns_per_coil_per_phase": "Turns_per_Coil",
    "number_of_turns": "Turns_per_Coil",
    "wire_diameter": "Wire_Diameter",
    "wire_diameter_mm": "Wire_Diameter",
    "wire_dia": "Wire_Diameter",
    "magnet_wire_diameter": "Wire_Diameter",
    "coil_wire_diameter": "Wire_Diameter",
}


def _normalize_variable_name(name: str) -> str:
    """Map AI-output variable name to exact Motor-CAD variable name."""
    if not name:
        return name
    # Direct match in registry
    if name in SCAN_PARAMETERS:
        return name
    # Case-insensitive lookup in registry
    lower = name.lower()
    for reg_name in SCAN_PARAMETERS:
        if reg_name.lower() == lower:
            return reg_name
    # Mapping table
    mapped = _VARIABLE_NAME_MAP.get(lower)
    if mapped:
        return mapped
    return name


def _extract_range(sv: Dict[str, Any]) -> Tuple[Optional[float], Optional[float], Optional[float]]:
    """Extract min/max/step from a scan variable dict with flexible field names."""
    min_v = None
    max_v = None
    step = None

    # Try various field names for min
    for key in ["min_value", "min", "minVal", "lower", "start", "from", "minimum"]:
        if key in sv and sv[key] is not None:
            try:
                min_v = float(sv[key])
                break
            except (ValueError, TypeError):
                pass

    # Try various field names for max
    for key in ["max_value", "max", "maxVal", "upper", "stop", "end", "to", "maximum"]:
        if key in sv and sv[key] is not None:
            try:
                max_v = float(sv[key])
                break
            except (ValueError, TypeError):
                pass

    # Try step
    for key in ["step", "step_size", "increment", "delta", "resolution"]:
        if key in sv and sv[key] is not None:
            try:
                step = float(sv[key])
                break
            except (ValueError, TypeError):
                pass

    # If range provided as [min, max]
    if (min_v is None or max_v is None) and "range" in sv:
        rng = sv["range"]
        if isinstance(rng, (list, tuple)) and len(rng) >= 2:
            try:
                min_v = float(rng[0]) if min_v is None else min_v
                max_v = float(rng[1]) if max_v is None else max_v
            except (ValueError, TypeError):
                pass

    return min_v, max_v, step


def _generate_values(start: float, stop: float, step: float) -> List[float]:
    """Generate evenly spaced values from start to stop inclusive."""
    if step <= 0 or start is None or stop is None:
        return []
    count = int(math.floor((stop - start) / step + 1e-9)) + 1
    if count <= 0 or count > 500:
        return []
    vals = [round(start + i * step, 6) for i in range(count)]
    if vals and abs(vals[-1] - stop) > 1e-9:
        vals.append(round(stop, 6))
    return vals


def convert_ai_plan_to_unified(ai_plan: Dict[str, Any], boundary_conditions: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    """Convert AI-generated plan format to unified plan schema v2.0.
    ...
    """
    bc = BoundaryConditions.from_dict(boundary_conditions or {})

    # Collect scan variables from either scan_variables or variables field
    raw_vars = ai_plan.get("scan_variables", []) or ai_plan.get("variables", [])

    variables = []
    for sv in raw_vars:
        if not isinstance(sv, dict):
            continue
        raw_name = sv.get("name", "")
        if not raw_name:
            continue

        # Normalize variable name to Motor-CAD exact name
        name = _normalize_variable_name(raw_name)

        # If AI already provided values array, use it directly
        values = sv.get("values", [])
        if not isinstance(values, list):
            values = []

        min_v, max_v, step = _extract_range(sv)

        # If no values but have range, generate
        if not values and min_v is not None and max_v is not None and step:
            values = _generate_values(min_v, max_v, step)

        # If still no values, fall back to rule engine defaults
        if not values:
            param = get_parameter(name)
            if param:
                rng = recommend_range(name, bc)
                min_v = rng["start"]
                max_v = rng["stop"]
                step = rng["step"]
                values = _generate_values(min_v, max_v, step)

        # Get display name and unit from registry if available
        param = get_parameter(name)
        tmpl = get_fixed_param_template(name)
        display_name = sv.get("display_name") or (param.display_name if param else tmpl.get("display_name", raw_name))
        name_cn = sv.get("name_cn") or tmpl.get("name_cn") or (param.display_name if param else raw_name)
        unit = sv.get("unit") or (param.unit if param else tmpl.get("unit", ""))
        category = sv.get("category") or (param.category if param else tmpl.get("category", ""))
        category_cn = sv.get("category_cn") or CATEGORY_CN.get(category, category)
        description = sv.get("description") or tmpl.get("description", "")
        description_cn = sv.get("description_cn") or tmpl.get("description_cn", "")

        variables.append({
            "name": name,
            "display_name": display_name,
            "name_cn": name_cn,
            "unit": unit,
            "start": min_v,
            "stop": max_v,
            "step": step,
            "values": values,
            "category": category,
            "category_cn": category_cn,
            "description": description,
            "description_cn": description_cn,
        })

    # Deduplicate variables by name (keep first)
    seen = set()
    unique_vars = []
    for v in variables:
        if v["name"] not in seen:
            seen.add(v["name"])
            unique_vars.append(v)
    variables = unique_vars

    # Separate standard registry variables from custom variables
    standard_vars = []
    custom_vars = []
    warnings = []
    for v in variables:
        if v["name"] in SCAN_PARAMETERS:
            standard_vars.append(v)
        else:
            custom_vars.append(v)
            warnings.append(f"Variable '{v['name']}' is not a standard Motor-CAD variable, may need manual adjustment")

    # If too many standard variables, keep only top 4 (to avoid combinatorial explosion)
    MAX_VARIABLES = 4
    if len(standard_vars) > MAX_VARIABLES:
        warnings.append(f"Too many scan variables ({len(standard_vars)}), keeping top {MAX_VARIABLES} to avoid combinatorial explosion")
        standard_vars = standard_vars[:MAX_VARIABLES]

    # Use standard variables primarily, append custom ones that have explicit values
    variables = standard_vars
    for cv in custom_vars:
        if cv.get("values") and len(cv["values"]) > 0:
            variables.append(cv)
        else:
            warnings.append(f"Skipping custom variable '{cv['name']}' - no valid values")

    # Hard limit: total variables <= 4
    if len(variables) > MAX_VARIABLES:
        warnings.append(f"Total variables ({len(variables)}) exceeds limit {MAX_VARIABLES}, keeping first {MAX_VARIABLES}")
        variables = variables[:MAX_VARIABLES]

    # Recalculate total points
    total_points = 1
    for v in variables:
        total_points *= len(v.get("values", [])) if v.get("values") else 1

    # If still too many points, trim variables one by one until under threshold
    MAX_POINTS = 200
    while total_points > MAX_POINTS and len(variables) > 1:
        removed = variables.pop()
        warnings.append(f"Removed variable '{removed['name']}' ({len(removed.get('values', []))} values) to reduce total points below {MAX_POINTS}")
        total_points = 1
        for v in variables:
            total_points *= len(v.get("values", [])) if v.get("values") else 1

    if total_points > MAX_POINTS:
        warnings.append(f"Total scan points ({total_points}) exceeds {MAX_POINTS}, consider narrowing ranges")

    # Fixed params: template-driven (ALL parameters from FIXED_PARAM_TEMPLATES,
    # excluding scan variables). Values inferred from boundary conditions where possible.
    # AI does not need to output fixed_params - the template is the single source.
    scan_names = {v["name"].lower() for v in variables}
    fixed_params = build_default_fixed_params(
        scan_variable_names=list(scan_names),
        boundary_conditions=boundary_conditions,
    )

    # Append any custom fixed params the AI explicitly provided that are not
    # already in the template (keeps AI flexibility for special cases).
    existing_names = {fp["name"].lower() for fp in fixed_params}
    for fp in ai_plan.get("fixed_params", []):
        if not isinstance(fp, dict) or not fp.get("name"):
            continue
        fname = fp["name"].lower()
        if fname in existing_names or fname in scan_names:
            continue
        tmpl = get_fixed_param_template(fp["name"])
        val = fp.get("value", 0)
        try:
            val = float(val)
        except (TypeError, ValueError):
            pass
        fixed_params.append({
            "name": fp["name"],
            "display_name": fp.get("display_name") or tmpl.get("display_name", fp["name"]),
            "name_cn": fp.get("name_cn") or tmpl.get("name_cn", fp["name"]),
            "unit": fp.get("unit") or tmpl.get("unit", ""),
            "value": val,
            "category": fp.get("category") or tmpl.get("category", "General"),
            "category_cn": fp.get("category_cn") or CATEGORY_CN.get(tmpl.get("category", "General"), tmpl.get("category", "General")),
            "description": fp.get("description") or tmpl.get("description", ""),
            "description_cn": fp.get("description_cn") or tmpl.get("description_cn", ""),
        })
        existing_names.add(fname)

    # Normalize acceptance_criteria to standard format
    ac = ai_plan.get("acceptance_criteria", {})
    if isinstance(ac, dict):
        hc = ac.get("hard_constraints")
        if isinstance(hc, dict):
            # Nested dict form: {"outer_diameter_mm": "<=110", ...} -> list of
            # "metric op value" strings so the UI can render them directly.
            ac = dict(ac)
            ac["hard_constraints"] = [f"{k} {v}" for k, v in hc.items()]
        if ac and not isinstance(ac.get("hard_constraints"), list):
            # AI may return custom format like {rated_torque_Nm_min: 25, ...}
            hard_constraints = []
            for key, val in ac.items():
                if isinstance(val, (int, float)):
                    if key.endswith("_min"):
                        metric = key.replace("_min", "").replace("_Nm", "").replace("_percent", "").replace("_N", "").replace("_T", "")
                        hard_constraints.append(f"{metric} >= {val}")
                    elif key.endswith("_max"):
                        metric = key.replace("_max", "").replace("_Nm", "").replace("_percent", "").replace("_N", "").replace("_T", "")
                        hard_constraints.append(f"{metric} <= {val}")
            if hard_constraints:
                ac = {
                    "hard_constraints": hard_constraints,
                    "objective_metric": ac.get("objective_metric", "efficiency_pct"),
                    "objective_direction": ac.get("objective_direction", "maximize"),
                    "soft_targets": {},
                }

    # Normalize search_strategy
    ss = ai_plan.get("search_strategy", {})
    if not isinstance(ss, dict):
        ss = {}
    # Normalize the method against the platform strategy registry. The model may
    # emit a non-registered method (e.g. "full_factorial_grid"); fall back to the
    # deterministic full_factorial default and record a warning instead of letting
    # the invalid value propagate to downstream validation.
    raw_method = ss.get("method", "")
    method = _strategy_normalize(raw_method)
    if method and not _strategy_registered(method):
        warnings.append(
            f"Unknown search strategy method '{raw_method}' - defaulting to full_factorial"
        )
        method = "full_factorial"
    if not method:
        method = "full_factorial"
    if ss and "method" not in ss:
        ss = {
            "method": method,
            "initial_samples": 16,
            "batch_size": 4,
            "max_solver_calls": 80,
            "local_trust_region": False,
            "objective_metric": "efficiency_pct",
            "objective_direction": "maximize",
        }
    else:
        ss = dict(ss)
        ss["method"] = method

    # Estimate total points
    total_points = 1
    for v in variables:
        total_points *= len(v.get("values", [])) if v.get("values") else 1

    return {
        "name": ai_plan.get("plan_name", "AI Generated Plan"),
        "topology": ai_plan.get("topology", "SSSR"),
        "model_path": ai_plan.get("model_path", ""),
        "fixed_params": fixed_params,
        "variables": variables,
        "cases": [{"id": "default", "name": "Default operating point", "params": {}}],
        "output_metrics": [
            "tavg_nm", "ripple_pct", "efficiency_pct", "total_losses_w",
            "copper_loss_w", "iron_loss_w", "magnet_loss_w", "back_emf_v",
            "input_power_w", "output_power_w", "shaft_speed_rpm",
        ],
        "search_strategy": ss,
        "acceptance_criteria": ac if ac else None,
        "ai_reasoning": ai_plan.get("reasoning", ""),
        "estimated_points": total_points,
        "estimated_time_min": total_points * 3,
        "warnings": warnings,
    }


class AIPlanGenerator:
    """Generate simulation plans from natural language using AI."""

    def __init__(self, l0_engine: Optional[L0PreScreeningEngine] = None):
        self.ai_client = get_kimi_client()
        self.l0_engine = l0_engine or L0PreScreeningEngine()
        self._prompt_template = None

    def _load_prompt(self) -> str:
        """Load plan generation prompt template."""
        if self._prompt_template is None:
            prompt_path = PROMPTS_DIR / "plan_generation" / "generate.txt"
            if prompt_path.exists():
                with open(prompt_path, "r", encoding="utf-8") as f:
                    self._prompt_template = f.read()
            else:
                self._prompt_template = self._default_prompt()
        return self._prompt_template

    def _default_prompt(self) -> str:
        """Fallback default prompt."""
        return """\u4f60\u662f\u8f74\u5411\u78c1\u901a\u7535\u673a\u4eff\u771f\u65b9\u6848\u4e13\u5bb6\u3002\u6839\u636e\u7528\u6237\u9700\u6c42\u548c\u8fb9\u754c\u6761\u4ef6\u751f\u6210JSON\u683c\u5f0f\u7684\u4eff\u771f\u65b9\u6848\u3002\n\u56fa\u5b9a\u53c2\u6570\u5df2\u7531\u7cfb\u7edf\u7edf\u4e00\u6a21\u677f\u751f\u6210\uff0c\u4f60\u65e0\u9700\u8f93\u51fafixed_params\uff1b\u4f60\u53ea\u9700\u8f93\u51fa\u4ee5\u4e0b\u5b57\u6bb5\uff1a\n- plan_name: \u65b9\u6848\u540d\u79f0\uff08\u7b80\u77ed\u63cf\u8ff0\uff09\n- topology: \u62d3\u6251\u7ed3\u6784\uff08SSSR/DRSS/SDSR\uff09\n- scan_variables: \u626b\u63cf\u53d8\u91cf\u6570\u7ec4\uff0c\u6bcf\u9879\u542bname\uff08\u7528Motor-CAD\u6807\u51c6\u53d8\u91cf\u540d\uff09\u3001start\u3001stop\u3001step\uff0c\u6700\u591a4\u4e2a\n- search_strategy: \u641c\u7d22\u7b56\u7565\uff08method, objective_metric, objective_direction\u7b49\uff09\n- acceptance_criteria: \u9a8c\u6536\u6807\u51c6\uff08\u786c\u7ea6\u675f\u3001\u6027\u80fd\u76ee\u6807\uff09\n- reasoning: \u4e2d\u6587\u8bbe\u8ba1\u601d\u8def\uff0c\u7ea6200\u5b57\uff0c\u53ea\u8bb2\u5173\u952e\u8bbe\u8ba1\u51b3\u7b56\u548c\u6838\u5fc3\u53c2\u6570\u9009\u62e9\u7406\u7531\uff0c\u7b80\u6d01\u76f4\u63a5\n\u8f93\u51fa\u7eafJSON\uff0c\u4e0d\u8981\u591a\u4f59\u6587\u5b57\u3002"""

    def generate(
        self,
        user_requirement: str,
        project_context: Optional[Dict[str, Any]] = None,
        existing_experience: Optional[List[Dict[str, Any]]] = None,
    ) -> Dict[str, Any]:
        """Generate a simulation plan from natural language.

        Args:
            user_requirement: Natural language description of the simulation need.
            project_context: Optional project context (topology, existing boundary conditions).
            existing_experience: Optional similar experience cases for reference.

        Returns:
            Dict with generated plan, validation results, and AI reasoning.
        """
        if not self.ai_client.is_configured:
            raise RuntimeError("KIMI_API_KEY is not configured")

        # Build context message
        context_parts = []
        if project_context:
            context_parts.append(f"\u9879\u76ee\u4e0a\u4e0b\u6587\uff1a{json.dumps(project_context, ensure_ascii=False)}")
        if existing_experience:
            context_parts.append(f"\u53c2\u8003\u7ecf\u9a8c\u6848\u4f8b\uff08{len(existing_experience)}\u4e2a\uff09\uff1a{json.dumps(existing_experience[:3], ensure_ascii=False)}")

        user_message = user_requirement
        if context_parts:
            user_message += "\n\n" + "\n".join(context_parts)

        # Call AI
        system_prompt = self._load_prompt()
        result = self.ai_client.chat_json(
            messages=[{"role": "user", "content": user_message}],
            system_prompt=system_prompt,
            max_tokens=KIMI_MAX_TOKENS,
        )

        # Parse generated plan
        plan = result.get("parsed_json", {})
        if not plan:
            # Try to extract and repair JSON from raw content
            raw = result.get("raw_content", result.get("content", ""))
            plan = self._try_repair_json(raw)
            if not plan:
                plan = {"error": "Failed to parse AI response", "raw_content": raw[:1000]}

        # Convert to unified format
        bc = (project_context or {}).get("boundary_conditions") if project_context else None
        unified_plan = convert_ai_plan_to_unified(plan, boundary_conditions=bc)

        # Validate with L0 pre-screening
        validation = self._validate_plan(plan)

        return {
            "plan": unified_plan,
            "raw_ai_plan": plan,
            "validation": validation,
            "ai_reasoning": plan.get("reasoning", ""),
            "usage": result.get("usage", {}),
            "raw_content": result.get("content", ""),
        }

    def _try_repair_json(self, raw: str) -> Dict[str, Any]:
        """Try to extract and repair JSON from potentially truncated AI output.

        Handles:
        - Markdown code fences
        - Truncated JSON (missing closing braces)
        - Extra text before/after JSON
        """
        if not raw:
            return {}

        text = raw.strip()

        # Remove markdown code fences
        if text.startswith("```"):
            lines = text.split("\n")
            if lines[0].startswith("```"):
                lines = lines[1:]
            if lines and lines[-1].strip() == "```":
                lines = lines[:-1]
            text = "\n".join(lines).strip()

        # Find first { and last }
        first_brace = text.find("{")
        last_brace = text.rfind("}")

        if first_brace == -1:
            return {}

        if last_brace == -1 or last_brace < first_brace:
            # JSON is truncated - try to close it
            json_str = text[first_brace:]
            # Count open braces and close them
            open_braces = json_str.count("{") - json_str.count("}")
            open_brackets = json_str.count("[") - json_str.count("]")
            # Remove trailing incomplete content
            last_comma = max(json_str.rfind(","), json_str.rfind(":"))
            if last_comma > len(json_str) * 0.8:
                json_str = json_str[:last_comma]
            # Close brackets and braces
            json_str += "]" * max(0, open_brackets)
            json_str += "}" * max(0, open_braces)
        else:
            json_str = text[first_brace:last_brace + 1]

        try:
            return json.loads(json_str)
        except json.JSONDecodeError:
            # Try one more repair: remove trailing comma before closing
            try:
                repaired = json_str.rstrip()
                while repaired and repaired[-1] in ", \n\t":
                    repaired = repaired[:-1]
                # Re-count and close
                open_braces = repaired.count("{") - repaired.count("}")
                open_brackets = repaired.count("[") - repaired.count("]")
                repaired += "]" * max(0, open_brackets)
                repaired += "}" * max(0, open_braces)
                return json.loads(repaired)
            except (json.JSONDecodeError, Exception):
                return {}

    def _validate_plan(self, plan: Dict[str, Any]) -> Dict[str, Any]:
        """Validate generated plan with L0 pre-screening.

        Checks:
        1. Boundary conditions feasibility
        2. Scan variable ranges within engineering limits
        3. Sample point count estimation
        4. Required fields presence
        """
        issues = []
        warnings = []
        checks = []

        # Check required fields
        required_fields = ["plan_name", "topology", "scan_variables"]
        for field in required_fields:
            if field not in plan:
                issues.append(f"Missing required field: {field}")

        # Validate topology
        topology = plan.get("topology", "")
        if topology and topology not in ("SSSR", "DRSS", "SDSR"):
            warnings.append(f"Unusual topology: {topology}")

        # Validate boundary conditions with L0
        bc = plan.get("boundary_conditions", {})
        if bc:
            l0_report = self.l0_engine.evaluate(bc)
            checks.append({
                "name": "boundary_conditions_l0",
                "feasible": l0_report.feasible,
                "passed": l0_report.passed_checks,
                "total": l0_report.total_checks,
                "failed_items": [r.name for r in l0_report.results if not r.passed],
            })
            if not l0_report.feasible:
                issues.append(f"Boundary conditions infeasible: {l0_report.failed_checks} checks failed")

        # Validate scan variables
        variables = plan.get("scan_variables", [])
        if not isinstance(variables, list):
            variables = []
        total_points = 1
        normalized_vars = []
        for var in variables:
            if isinstance(var, str):
                var = {"name": var, "min_value": None, "max_value": None}
            if not isinstance(var, dict):
                continue
            normalized_vars.append(var)
            name = var.get("name", "unknown")
            min_v = var.get("min_value")
            max_v = var.get("max_value")
            step = var.get("step")

            if min_v is None or max_v is None:
                issues.append(f"Variable {name}: missing min/max values")
                continue

            if min_v >= max_v:
                issues.append(f"Variable {name}: min ({min_v}) >= max ({max_v})")

            if step and step > 0:
                n_points = int((max_v - min_v) / step) + 1
                total_points *= n_points
                if n_points > 50:
                    warnings.append(f"Variable {name}: {n_points} levels may be too many")
        plan["scan_variables"] = normalized_vars

        checks.append({
            "name": "scan_variables",
            "count": len(variables),
            "estimated_full_factorial_points": total_points,
            "adaptive_search_recommended": total_points > 100,
        })

        if total_points > 500:
            warnings.append(f"Full factorial would require {total_points} points - strongly recommend adaptive search")

        # Validate search strategy
        search = plan.get("search_strategy", {})
        if search:
            max_calls = search.get("max_solver_calls", 80)
            if max_calls < total_points and not search.get("method", "").startswith(("constrained", "active")):
                warnings.append(f"Budget ({max_calls}) < full factorial ({total_points}) but method is not adaptive")

        overall_valid = len(issues) == 0
        return {
            "valid": overall_valid,
            "issues": issues,
            "warnings": warnings,
            "checks": checks,
        }

    def refine_plan(
        self,
        original_plan: Dict[str, Any],
        user_feedback: str,
    ) -> Dict[str, Any]:
        """Refine an existing plan based on user feedback.

        Args:
            original_plan: The previously generated plan.
            user_feedback: Natural language feedback for refinement.

        Returns:
            Refined plan with validation.
        """
        if not self.ai_client.is_configured:
            raise RuntimeError("KIMI_API_KEY is not configured")

        system_prompt = self._load_prompt() + "\n\n\u4f60\u6b63\u5728\u4f18\u5316\u4e00\u4e2a\u5df2\u6709\u7684\u4eff\u771f\u65b9\u6848\u3002\u6839\u636e\u7528\u6237\u53cd\u9988\u8c03\u6574\u65b9\u6848\uff0c\u4fdd\u6301\u5176\u4ed6\u90e8\u5206\u4e0d\u53d8\u3002"

        user_message = f"\u539f\u6709\u65b9\u6848\uff1a\n{json.dumps(original_plan, ensure_ascii=False, indent=2)}\n\n\u7528\u6237\u53cd\u9988\uff1a{user_feedback}\n\n\u8bf7\u8f93\u51fa\u4f18\u5316\u540e\u7684\u5b8c\u6574\u65b9\u6848JSON\u3002"

        result = self.ai_client.chat_json(
            messages=[{"role": "user", "content": user_message}],
            system_prompt=system_prompt,
            max_tokens=KIMI_MAX_TOKENS,
        )

        refined_plan = result.get("parsed_json", {})
        if not refined_plan:
            raw = result.get("raw_content", result.get("content", ""))
            refined_plan = self._try_repair_json(raw)
            if not refined_plan:
                refined_plan = {"error": "Failed to parse AI response", "raw_content": raw[:1000]}

        bc = original_plan.get("boundary_conditions") if isinstance(original_plan, dict) else None
        unified_plan = convert_ai_plan_to_unified(refined_plan, boundary_conditions=bc)
        validation = self._validate_plan(refined_plan)

        return {
            "plan": unified_plan,
            "raw_ai_plan": refined_plan,
            "validation": validation,
            "ai_reasoning": refined_plan.get("reasoning", ""),
            "usage": result.get("usage", {}),
        }


# Global singleton
_generator: Optional[AIPlanGenerator] = None


def get_plan_generator() -> AIPlanGenerator:
    """Get or create global AIPlanGenerator singleton."""
    global _generator
    if _generator is None:
        _generator = AIPlanGenerator()
    return _generator