backend/planner.py

"""
Autonomous task planner.

Produces structured action plans (list of dicts):
    {"type": "shell"|"python"|"browser"|"git"|"deploy"|"note", ...}

Strategy:
1. Try LLM-based planning with strict JSON output.
2. If LLM fails or returns invalid JSON, use heuristic fallback.
3. Always produces non-empty plan.
"""
from __future__ import annotations

import json
import logging
import re
from typing import Any, Dict, List

from .llm_router import get_router

logger = logging.getLogger("planner")

PLANNER_SYSTEM = (
    "You are an autonomous AI Developer Agent's planner. "
    "Decompose a user task into a JSON list of concrete actions. "
    "Each action is an object with a 'type' field and supporting fields. "
    "Supported types: shell (cmd), python (code), browser (action,url,...), "
    "git (op,args), deploy (target), note (msg). "
    "Return ONLY a JSON array, no commentary. Keep plan under 12 steps."
)


def plan_task(title: str, description: str, context: Dict[str, Any] | None = None) -> List[Dict[str, Any]]:
    """Generate a concrete action plan."""
    router = get_router()
    user_prompt = (
        f"TASK TITLE: {title}\n"
        f"DESCRIPTION:\n{description}\n\n"
        f"CONTEXT:\n{json.dumps(context or {}, indent=2)[:2000]}\n\n"
        "Output a JSON array of action objects. Example:\n"
        '[{"type":"shell","cmd":"echo hi"},{"type":"note","msg":"done"}]'
    )
    messages = [
        {"role": "system", "content": PLANNER_SYSTEM},
        {"role": "user", "content": user_prompt},
    ]
    try:
        raw = router.chat(messages, temperature=0.1, max_tokens=1200, timeout=45.0)
    except Exception as e:
        logger.warning("Planner LLM call failed: %s", e)
        raw = ""

    plan = _parse_plan_json(raw)
    if plan:
        return plan
    logger.info("Planner falling back to heuristic plan")
    return heuristic_plan(title, description)


def _parse_plan_json(raw: str) -> List[Dict[str, Any]]:
    if not raw:
        return []
    # Try direct
    try:
        obj = json.loads(raw)
        if isinstance(obj, list):
            return [a for a in obj if isinstance(a, dict) and "type" in a]
    except Exception:
        pass
    # Find first JSON array in text
    m = re.search(r"\[[\s\S]*\]", raw)
    if m:
        try:
            obj = json.loads(m.group(0))
            if isinstance(obj, list):
                return [a for a in obj if isinstance(a, dict) and "type" in a]
        except Exception:
            return []
    return []


def heuristic_plan(title: str, description: str) -> List[Dict[str, Any]]:
    """Always-valid fallback plan."""
    text = (title + "\n" + description).lower()
    plan: List[Dict[str, Any]] = []

    if any(k in text for k in ["deploy", "deployment", "huggingface", "hf space", "vercel"]):
        plan.append({"type": "note", "msg": "Deployment task detected"})
        if "vercel" in text:
            plan.append({"type": "deploy", "target": "vercel"})
        if "huggingface" in text or "hf" in text:
            plan.append({"type": "deploy", "target": "huggingface"})
        return plan

    if any(k in text for k in ["git", "github", "commit", "push", "pr ", "pull request"]):
        plan.append({"type": "git", "op": "status"})
        plan.append({"type": "note", "msg": "GitHub task detected"})
        return plan

    if any(k in text for k in ["browser", "scrape", "navigate", "click", "open url", "http://", "https://"]):
        url_match = re.search(r"https?://[\w\.\-/?=&%#]+", description)
        url = url_match.group(0) if url_match else "https://example.com"
        plan.append({"type": "browser", "action": "navigate", "url": url})
        plan.append({"type": "browser", "action": "screenshot"})
        return plan

    if any(k in text for k in ["run python", "python script", "execute python"]):
        plan.append({"type": "python", "code": "print('Hello from AI Developer Agent')"})
        return plan

    if any(k in text for k in ["install", "pip ", "npm "]):
        # Try to extract a package name
        m = re.search(r"(?:install|add)\s+([\w\.\-]+)", text)
        pkg = m.group(1) if m else ""
        if "npm" in text:
            plan.append({"type": "shell", "cmd": f"npm install {pkg}".strip()})
        else:
            plan.append({"type": "shell", "cmd": f"pip install {pkg}".strip()})
        return plan

    # Generic fallback: echo the task back as an inspection action.
    plan.append({"type": "note", "msg": f"Plan-fallback for: {title}"})
    plan.append({"type": "shell", "cmd": "uname -a && python3 --version && node --version 2>/dev/null || true"})
    return plan


def repair_plan(error_category: str, detail: str = "") -> List[Dict[str, Any]]:
    from .classifier import ErrorClass
    from .repair import repair_actions
    return repair_actions(ErrorClass(category=error_category, detail=detail, suggested_fix=""))