api/quality_guardian.py

"""
quality_guardian.py — Quality Guardian / Test-Resolution Engine (S363, S701, S703, S704)

Quando l'agente genera codice Python:
1. Estrae il blocco codice dall'output LLM
2. Usa Groq-8b (Role.TESTER) per generare un test minimale
3. Esegue il test in subprocess sandboxato (30s max)
4. Se FAIL: loop 3-iter repair (S703):
   - Iter 1: diagnosi BUG/FIX
   - Iter 2: rewrite guidato dalla diagnosi → re-exec
   - Iter 3: versione semplificata → re-exec
5. Emette evento `test_result` via on_event callback

S701: aggiunto browser testing per HTML/JS/React via Playwright
- Se il codice contiene blocchi HTML/JS/React → _browser_quality_check()
- Screenshot DOM check + JS console error interception

S704: screenshot quality gate
- page.screenshot(JPEG quality:30) → file size < 2500B = blank
- document.querySelectorAll('*').length < 5 = sparse DOM
- Fallback silenzioso se screenshot fallisce

Invarianti:
- Mai blocca il loop principale (fire-and-forget da unified_loop)
- Mai lancia eccezioni (fallback silenzioso totale)
- Timeout hard: 45s totali (S703: 30→45 per loop 3-iter)
"""
from __future__ import annotations

import asyncio
import os
import re
import subprocess
import sys
import tempfile
from typing import Any, Awaitable, Callable

import logging
_logger = logging.getLogger("api.quality_guardian")

# ── Regex ─────────────────────────────────────────────────────────────────────
_CODE_FENCE_RE = re.compile(
    r'```(?:python|py)?\s*\n([\s\S]+?)```',
    re.IGNORECASE,
)

_HTML_FENCE_RE = re.compile(
    r'```(?:html|htm|css|javascript|js|jsx|tsx|ts|react|vue|svelte)\s*\n([\s\S]+?)```',
    re.IGNORECASE,
)

_SECURITY_BLOCKLIST_RE = re.compile(
    r'\b(subprocess|import\s+os|shutil|socket|open\s*\(|eval\(|exec\(|__import__)\b'
)

# ── S-QUALITY-UP-1: Error type classifier — sceglie modello per iter 3 ─────────
def _classify_error_type(test_output: str) -> str:
    """Classifica il tipo di errore per scegliere il modello di repair.
    Returns: 'syntax' | 'logic' | 'type_error' | 'assertion' | 'unknown'
    """
    out = (test_output or "").lower()
    if any(k in out for k in ["syntaxerror", "indentationerror", "unexpected token"]):
        return "syntax"
    if any(k in out for k in ["typeerror", "attributeerror", "nameerror", "importerror"]):
        return "type_error"
    if any(k in out for k in ["assertionerror", "assert ", "expected", "got "]):
        return "assertion"
    if any(k in out for k in ["logicerror", "valueerror", "keyerror", "indexerror", "wrong result"]):
        return "logic"
    return "unknown"


# S701: rileva task HTML/JS/React
_HTML_TASK_GOAL_RE = re.compile(
    r'\b(html|css|javascript|react|vue|svelte|dom|component|webpage|landing.?page|frontend|web.?app)\b',
    re.IGNORECASE,
)

_HTML_CODE_FENCE_RE = re.compile(
    r'```(?:html|htm|javascript|js|jsx|tsx|react|vue|svelte)\s*\n',
    re.IGNORECASE,
)

# ── Prompts ───────────────────────────────────────────────────────────────────
_TESTER_SYS = (
    # S698: 2-3 assert coverage (normal+edge+error) vs precedente "shortest possible"
    "You are a code validator. Given a Python function or class, "
    "write 2-3 assert-based tests covering different scenarios.\n"
    "Rules:\n"
    "1. Output ONLY valid Python code, no explanations, no markdown fences\n"
    "2. Use only assert statements — no pytest, no unittest, no extra imports\n"
    "3. Cover: (a) normal case, (b) edge case (empty/zero/None), (c) error or boundary\n"
    "4. The function is already defined above — call it directly, no re-imports\n"
    "5. Every test must finish in <5s, no file I/O, no network, no subprocess\n"
    "6. If the code cannot be tested in isolation, output exactly: # SKIP\n"
)

_DEBUGGER_SYS = (
    "You are a code debugger. Given failing test output, identify the bug.\n"
    "Response format (2 lines max):\n"
    "BUG: <description in Italian>\n"
    "FIX: <minimal code correction>\n"
)

# S703: prompt per iter 2 (rewrite guidato)
_REWRITER_SYS = (
    "You are a code fixer. Given a Python function and its failing test error, "
    "produce a corrected version of the function.\n"
    "Rules:\n"
    "1. Output ONLY the corrected function code — no tests, no markdown, no explanation\n"
    "2. Keep the same function name and signature\n"
    "3. Fix the specific error shown\n"
    "4. Valid Python only — no placeholders\n"
)

# S703: prompt per iter 3 (simplificazione)
_SIMPLIFIER_SYS = (
    "You are a code simplifier. Given a Python function that has bugs after 2 fix attempts, "
    "produce a minimal working version that handles the basic cases correctly.\n"
    "Rules:\n"
    "1. Output ONLY valid Python code — no markdown, no explanation\n"
    "2. Keep the same function name and signature\n"
    "3. Prioritize correctness over completeness — simplify edge cases if needed\n"
)

_HTML_DEBUGGER_SYS = (
    "You are a frontend debugger. Given browser errors on an HTML/JS page, identify the bug.\n"
    "Response format (2 lines max):\n"
    "BUG: <description in Italian>\n"
    "FIX: <minimal code correction>\n"
)


# ── S701+S704: Browser Quality Check (HTML/JS/React) ──────────────────────────

async def _browser_quality_check(
    task_id: str,
    goal: str,
    html_code: str,
    on_event: Callable | None,
) -> dict:
    """
    S701+S704: testa HTML/JS/React via Playwright headless Chromium.
    Checks: DOM non-empty, zero JS errors, screenshot size (S704), element count (S704).
    Timeout: 15s totali.
    Fallback silenzioso se playwright non installato.
    """
    try:
        from playwright.async_api import async_playwright
    except ImportError:
        return {"passed": None, "skipped": True, "reason": "playwright_not_installed"}

    # Wrap in documento HTML completo se necessario
    html_content = html_code
    if not re.search(r'<html|<!DOCTYPE', html_content, re.IGNORECASE):
        html_content = (
            "<!DOCTYPE html>\n<html lang=\"it\">\n"
            "<head><meta charset=\"UTF-8\"><title>Test</title></head>\n"
            f"<body>\n{html_content}\n</body>\n</html>"
        )

    tmp_html: str | None = None
    try:
        with tempfile.NamedTemporaryFile(suffix=".html", mode="w", delete=False, encoding="utf-8") as f:
            f.write(html_content)
            tmp_html = f.name

        js_errors: list[str] = []
        console_errors: list[str] = []

        async with async_playwright() as p:
            browser = await asyncio.wait_for(
                p.chromium.launch(
                    headless=True,
                    args=["--no-sandbox", "--disable-dev-shm-usage", "--disable-gpu"],
                ),
                timeout=10.0,
            )
            try:
                # S704: viewport ridotto per screenshot size check più sensibile
                page = await browser.new_page(
                    viewport={"width": 800, "height": 600},
                )
                page.on("pageerror", lambda err: js_errors.append(str(err)))
                page.on("console", lambda msg: (
                    console_errors.append(msg.text) if msg.type == "error" else None
                ))

                await asyncio.wait_for(
                    page.goto(f"file://{tmp_html}", wait_until="domcontentloaded"),
                    timeout=8.0,
                )
                await page.wait_for_timeout(400)

                body_text = await page.evaluate(
                    "document.body ? document.body.innerText.trim() : ''"
                )
                body_html = await page.evaluate(
                    "document.body ? document.body.innerHTML.trim() : ''"
                )

                # S704: element count — sparse DOM detection
                is_sparse_dom = False
                try:
                    elem_count = await page.evaluate(
                        "document.querySelectorAll('*').length"
                    )
                    is_sparse_dom = int(elem_count) < 5
                except Exception as _exc:
                    _logger.debug("[quality_guardian] silenced %s", type(_exc).__name__)  # noqa: BLE001

                # S704: screenshot size — blank page detection
                is_blank_screenshot: bool | None = None
                try:
                    ss_bytes = await page.screenshot(
                        type="jpeg", quality=30, full_page=False
                    )
                    # JPEG quality:30 blank white 800×600 ≈ 1–2 KB; real content > 3 KB
                    is_blank_screenshot = len(ss_bytes) < 2500
                except Exception:
                    pass  # screenshot non critico — skip silenzioso

            finally:
                await browser.close()

        is_white_screen = (
            (len(body_text) < 5 and len(body_html) < 20)
            or is_blank_screenshot is True
        )
        has_js_errors = bool(js_errors or console_errors)
        passed        = not is_white_screen and not has_js_errors and not is_sparse_dom
        fix_hint: str | None = None

        if not passed:
            all_errors = (js_errors + console_errors)[:3]
            if is_white_screen:
                fix_hint = (
                    "BUG: Pagina bianca — nessun contenuto visibile nel DOM\n"
                    "FIX: Verifica che il codice HTML/JS inserisca contenuto nel <body>"
                )
            elif is_sparse_dom:
                fix_hint = (
                    "BUG: DOM troppo scarso — meno di 5 elementi nella pagina\n"
                    "FIX: Aggiungere elementi visibili nel body (divs, paragrafi, componenti)"
                )
            elif has_js_errors:
                err_txt = "; ".join(all_errors)
                try:
                    from models.role_router import RoleRouter, Role
                    dbg = RoleRouter.get_client(Role.TESTER)
                    fix_hint = await asyncio.wait_for(
                        dbg.chat(
                            [
                                {"role": "system", "content": _HTML_DEBUGGER_SYS},
                                {"role": "user", "content": (
                                    f"Goal: {goal[:300]}\n"
                                    f"JS Errors:\n{err_txt[:600]}\n\n"
                                    f"Code:\n```html\n{html_code[:1200]}\n```"
                                )},
                            ],
                            temperature=0,
                            max_tokens=200,
                        ),
                        timeout=6.0,
                    )
                except Exception:
                    fix_hint = f"BUG: Errori JS — {err_txt[:200]}\nFIX: Correggi gli errori JavaScript"

        if on_event:
            try:
                # S704: includi screenshot_blank e sparse_dom nel evento
                val = on_event({
                    "type":              "test_result",
                    "action":            "test_result",
                    "taskId":            task_id,
                    "passed":            passed,
                    "stdout":            (
                        f"browser DOM: {'OK' if not is_white_screen else 'WHITE_SCREEN'}"
                        f" | JS errors: {len(js_errors)}"
                        f" | sparse_dom: {is_sparse_dom}"
                        f" | blank_ss: {is_blank_screenshot}"
                    ),
                    "stderr":            "; ".join((js_errors + console_errors)[:3])[:500] if not passed else "",
                    "mode":              "browser",
                    "screenshot_blank":  is_blank_screenshot,
                    "sparse_dom":        is_sparse_dom,
                })
                if asyncio.iscoroutine(val):
                    await val
            except Exception as _exc:
                _logger.debug("[quality_guardian] silenced %s", type(_exc).__name__)  # noqa: BLE001

        # S701+S704: telemetry counters
        try:
            from api.state import increment_stat as _inc_stat
            _inc_stat("browser_quality_pass" if passed else "browser_quality_fail")
            if is_blank_screenshot is True:
                _inc_stat("browser_screenshot_blank")
            if is_sparse_dom:
                _inc_stat("browser_dom_sparse")
        except Exception as _exc:
            _logger.debug("[quality_guardian] silenced %s", type(_exc).__name__)  # noqa: BLE001

        return {
            "passed":           passed,
            "skipped":          False,
            "stdout":           f"browser_test: {'PASS' if passed else 'FAIL'}",
            "stderr":           "; ".join((js_errors + console_errors)[:3])[:500],
            "fix_hint":         fix_hint,
            "mode":             "browser",
            "screenshot_blank": is_blank_screenshot,
            "sparse_dom":       is_sparse_dom,
        }

    except asyncio.TimeoutError:
        return {"passed": None, "skipped": True, "reason": "browser_timeout"}
    except Exception as exc:
        return {"passed": None, "skipped": True, "reason": f"browser_error: {str(exc)[:100]}"}
    finally:
        if tmp_html:
            try:
                os.unlink(tmp_html)
            except Exception as _exc:
                _logger.debug("[quality_guardian] silenced %s", type(_exc).__name__)  # noqa: BLE001


# ── Entry point ───────────────────────────────────────────────────────────────

async def run_quality_check(
    task_id: str,
    goal: str,
    llm_output: str,
    on_event: Callable[[dict], Awaitable[None] | None] | None = None,
    session_files: dict | None = None,  # GAP-3-QG: file sessione per multi-file quality check
) -> dict:
    """Fire-and-forget entry. Always returns a safe dict."""
    try:
        return await asyncio.wait_for(
            _check(task_id, goal, llm_output, on_event, session_files),
            timeout=45.0,  # S703: 30→45s per loop 3-iter repair
        )
    except Exception:
        return {"passed": None, "skipped": True, "reason": "guardian_timeout"}


async def _check(task_id, goal, llm_output, on_event, session_files: dict | None = None) -> dict:
    # S701: route HTML/JS/React tasks to browser test
    if _HTML_CODE_FENCE_RE.search(llm_output) or (
        _HTML_TASK_GOAL_RE.search(goal) and re.search(r'<[a-zA-Z][^>]{0,40}>', llm_output)
    ):
        html_blocks = _HTML_FENCE_RE.findall(llm_output)
        if not html_blocks:
            # Prova raw HTML dal testo
            raw = re.findall(
                r'(<(?:html|body|div|script)[^>]*>[\s\S]+?</(?:html|body|div|script)>)',
                llm_output, re.IGNORECASE,
            )
            if raw:
                html_blocks = [raw[0]]
        if html_blocks:
            try:
                return await asyncio.wait_for(
                    _browser_quality_check(task_id, goal, html_blocks[0].strip(), on_event),
                    timeout=20.0,
                )
            except Exception:
                pass  # fallthrough to Python check se browser fallisce

    # 1. Extract Python code — GAP-3: preferisce file .py reali di sessione ai fence block LLM
    _session_py: str | None = None
    if session_files:
        _py_files = {p: c for p, c in session_files.items() if p.endswith(".py")}
        if _py_files:
            _, _session_py = max(_py_files.items(), key=lambda kv: len(kv[1]))
    if _session_py:
        code = _session_py
    else:
        blocks = _CODE_FENCE_RE.findall(llm_output)
        if not blocks:
            return {"passed": None, "skipped": True, "reason": "no_code_blocks"}
        code = blocks[0].strip()

    if _SECURITY_BLOCKLIST_RE.search(code):
        return {"passed": None, "skipped": True, "reason": "security_skip"}

    # 2. Generate minimal test via Groq-8b
    try:
        from models.role_router import RoleRouter, Role
        tester = RoleRouter.get_client(Role.TESTER)
        test_raw = await asyncio.wait_for(
            tester.chat(
                [
                    {"role": "system", "content": _TESTER_SYS},
                    # S589/S597: goal 500 chars
                    {"role": "user", "content": f"Goal: {goal[:500]}\n\n```python\n{code[:2000]}\n```"},
                ],
                temperature=0,
                max_tokens=400,
            ),
            timeout=10.0,
        )
    except Exception:
        return {"passed": None, "skipped": True, "reason": "test_gen_failed"}

    test_code = test_raw.strip()
    if "# SKIP" in test_code[:60]:
        return {"passed": None, "skipped": True, "reason": "guardian_skipped"}

    test_code = re.sub(r'^```\w*\s*', '', test_code)
    test_code = re.sub(r'\s*```$', '', test_code)

    # S698: valida sintassi prima di eseguire
    try:
        compile(test_code, '<generated_test>', 'exec')
    except SyntaxError as _syn_err:
        try:
            _test_raw2 = await asyncio.wait_for(
                tester.chat(
                    [
                        {"role": "system", "content": _TESTER_SYS},
                        {"role": "user", "content": (
                            f"Goal: {goal[:500]}\n\n" + chr(96) * 3 + f"python\n{code[:2000]}\n" + chr(96) * 3
                            + f"\n\nPrevious attempt had SyntaxError: {_syn_err}\n"
                            "Output ONLY valid Python, no markdown, no explanation."
                        )},
                    ],
                    temperature=0,
                    max_tokens=400,
                ),
                timeout=8.0,
            )
            _tc2 = re.sub(r'^[^a-zA-Z_#]*', '', _test_raw2.strip())
            try:
                compile(_tc2, '<generated_test_retry>', 'exec')
                test_code = _tc2
            except SyntaxError:
                return {"passed": None, "skipped": True, "reason": "test_syntax_invalid"}
        except Exception:
            return {"passed": None, "skipped": True, "reason": "test_syntax_invalid"}

    # 3. Run sandboxed subprocess
    exec_result = await asyncio.get_event_loop().run_in_executor(
        None, _run_subprocess, f"{code}\n\n{test_code}"
    )

    passed = exec_result["returncode"] == 0

    # 4. Emit test_result SSE event
    if on_event:
        try:
            val = on_event({
                "type":    "test_result",
                "action":  "test_result",
                "taskId":  task_id,
                "passed":  passed,
                "stdout":  exec_result["stdout"][:500],  # S604
                "stderr":  exec_result["stderr"][:500],  # S597
            })
            if asyncio.iscoroutine(val):
                await val
        except Exception as _exc:
            _logger.debug("[quality_guardian] silenced %s", type(_exc).__name__)  # noqa: BLE001

    # 5. S703: 3-iter repair loop (iter 1 = diagnosi, iter 2 = rewrite, iter 3 = simplify)
    fix_hint: str | None = None
    active_code = code  # aggiornato se un iter produce codice corretto

    if not passed:
        # ── Iter 1: diagnosi BUG/FIX ────────────────────────────────────────
        if exec_result["stderr"]:
            try:
                from models.role_router import RoleRouter, Role
                debugger = RoleRouter.get_client(Role.TESTER)
                fix_hint = await asyncio.wait_for(
                    debugger.chat(
                        [
                            {"role": "system", "content": _DEBUGGER_SYS},
                            {"role": "user", "content": (
                                f"Code:\n```python\n{code[:1200]}\n```\n"
                                f"Error:\n{exec_result['stderr'][:600]}"
                            )},
                        ],
                        temperature=0,
                        max_tokens=300,  # S587
                    ),
                    timeout=6.0,  # S703: 8→6s per lasciare budget a iter 2+3
                )
            except Exception as _exc:
                _logger.debug("[quality_guardian] silenced %s", type(_exc).__name__)  # noqa: BLE001

        # ── Iter 2: rewrite guidato dalla diagnosi (S703) ───────────────────
        if fix_hint and not passed:
            try:
                from models.role_router import RoleRouter, Role
                from api.state import increment_stat as _inc2
                _inc2("repair_iter2_used")
                rewriter = RoleRouter.get_client(Role.TESTER)
                fixed_raw = await asyncio.wait_for(
                    rewriter.chat(
                        [
                            {"role": "system", "content": _REWRITER_SYS},
                            {"role": "user", "content": (
                                f"Goal: {goal[:300]}\n"
                                f"Original code:\n```python\n{code[:1200]}\n```\n"
                                f"Test error:\n{exec_result['stderr'][:400]}\n"
                                f"Diagnosis:\n{fix_hint[:400]}"
                            )},
                        ],
                        temperature=0,
                        max_tokens=600,
                    ),
                    timeout=8.0,
                )
                fixed_code = re.sub(r'^```\w*\s*', '', fixed_raw.strip())
                fixed_code = re.sub(r'\s*```$', '', fixed_code).strip()
                if fixed_code and len(fixed_code) > 20:
                    r2 = await asyncio.get_event_loop().run_in_executor(
                        None, _run_subprocess, f"{fixed_code}\n\n{test_code}"
                    )
                    if r2["returncode"] == 0:
                        passed = True
                        active_code = fixed_code
                        exec_result = r2
                        # Emetti evento aggiornato
                        if on_event:
                            try:
                                val = on_event({
                                    "type":    "test_result",
                                    "action":  "test_result",
                                    "taskId":  task_id,
                                    "passed":  True,
                                    "stdout":  r2["stdout"][:500],
                                    "stderr":  "",
                                    "repairIter": 2,
                                })
                                if asyncio.iscoroutine(val):
                                    await val
                            except Exception as _exc:
                                _logger.debug("[quality_guardian] silenced %s", type(_exc).__name__)  # noqa: BLE001
            except Exception as _exc:
                _logger.debug("[quality_guardian] silenced %s", type(_exc).__name__)  # noqa: BLE001

        # ── Iter 3: semplificazione (S703) — solo se iter 2 fallita ─────────
        if not passed:
            try:
                from models.role_router import RoleRouter, Role
                from api.state import increment_stat as _inc3
                _inc3("repair_iter3_used")
                simplifier = RoleRouter.get_client(Role.TESTER)
                # S-QUALITY-UP-1: usa DeepSeek-R1 free per errori logici/tipo a iter 3 (2026-06-12)
                try:
                    _err3 = _classify_error_type(exec_result.get('stderr', '') + exec_result.get('stdout', ''))
                    if _err3 in ('logic', 'type_error', 'assertion') and os.getenv('OPENROUTER_API_KEY'):
                        from models.ai_client import AIClient as _AI3
                        simplifier = _AI3(provider='openrouter', model='deepseek/deepseek-r1:free')
                except Exception:
                    pass  # fallback a Groq-8b gia' assegnato sopra
                simple_raw = await asyncio.wait_for(
                    simplifier.chat(
                        [
                            {"role": "system", "content": _SIMPLIFIER_SYS},
                            {"role": "user", "content": (
                                f"Goal: {goal[:300]}\n"
                                f"Broken code (2 fix attempts failed):\n"
                                f"```python\n{code[:1000]}\n```\n"
                                f"Errors:\n{exec_result['stderr'][:300]}"
                            )},
                        ],
                        temperature=0.1,
                        max_tokens=500,
                    ),
                    timeout=6.0,
                )
                simple_code = re.sub(r'^```\w*\s*', '', simple_raw.strip())
                simple_code = re.sub(r'\s*```$', '', simple_code).strip()
                if simple_code and len(simple_code) > 20:
                    r3 = await asyncio.get_event_loop().run_in_executor(
                        None, _run_subprocess, f"{simple_code}\n\n{test_code}"
                    )
                    if r3["returncode"] == 0:
                        passed = True
                        active_code = simple_code
                        exec_result = r3
                        if on_event:
                            try:
                                val = on_event({
                                    "type":    "test_result",
                                    "action":  "test_result",
                                    "taskId":  task_id,
                                    "passed":  True,
                                    "stdout":  r3["stdout"][:500],
                                    "stderr":  "",
                                    "repairIter": 3,
                                })
                                if asyncio.iscoroutine(val):
                                    await val
                            except Exception as _exc:
                                _logger.debug("[quality_guardian] silenced %s", type(_exc).__name__)  # noqa: BLE001
            except Exception as _exc:
                _logger.debug("[quality_guardian] silenced %s", type(_exc).__name__)  # noqa: BLE001

    return {
        "passed":   passed,
        "skipped":  False,
        "stdout":   exec_result["stdout"][:500],  # S604
        "stderr":   exec_result["stderr"][:500],  # S598
        "fix_hint": fix_hint,
    }


def _run_subprocess(code: str) -> dict:
    # S365: prefer exec_sandbox
    try:
        from api.exec_sandbox import run_in_sandbox
        return run_in_sandbox(code, task_id="quality_check")
    except ImportError:
        pass

    try:
        proc = subprocess.run(
            [sys.executable, "-c", code],
            capture_output=True,
            text=True,
            timeout=12,
        )
        # S571: stdout/stderr limits allineati a exec_sandbox
        return {
            "returncode": proc.returncode,
            "stdout":     proc.stdout[:8000],
            "stderr":     proc.stderr[:4000],
        }
    except subprocess.TimeoutExpired:
        return {"returncode": 1, "stdout": "", "stderr": "timeout after 12s"}
    except Exception as e:
        return {"returncode": 1, "stdout": "", "stderr": str(e)}