agents/tdd_runner.py

"""
tdd_runner.py — S-GAP3: Auto-Testing post-code execution.

Dopo ogni run_python con codice complesso (>=8 righe, def/class presente),
l'agente genera un test minimale, lo esegue, e verifica PASS/FAIL.
Questo chiude il ciclo: codice scritto → codice verificato.

Integrazione: chiamato da unified_loop_tools.py dopo run_python.
Zero overhead su task semplici (direct_response, query, output non-code).
"""
from __future__ import annotations

import asyncio
import logging
import re
from typing import Any

_logger = logging.getLogger("agente_ai.tdd")

_MIN_LINES = 8
_CODE_RE = re.compile(r"\b(def |class |function |return |async def )", re.MULTILINE)
_SUPPORTED = {"python", "py", ""}


def _should_test(code: str, language: str = "python") -> bool:
    if language.lower() not in _SUPPORTED:
        return False
    lines = [l for l in code.splitlines() if l.strip() and not l.strip().startswith("#")]
    return len(lines) >= _MIN_LINES and bool(_CODE_RE.search(code))


async def _gen_test(code: str, llm: Any) -> str | None:
    try:
        from models.role_router import RoleRouter, Role
        coder = RoleRouter.get_client(Role.CODER)
    except Exception:
        coder = llm
    prompt = (
        "Dato questo codice Python, scrivi un micro-test con assert statements "
        "(NON pytest/unittest — solo assert + print). "
        "Verifica il comportamento principale. Rispondi SOLO con il codice.\n\n"
        f"```python\n{code[:1800]}\n```"
    )
    try:
        raw = await asyncio.wait_for(
            coder.chat(
                [
                    {"role": "system", "content": "Sei un tester Python. Solo codice, niente testo."},
                    {"role": "user", "content": prompt},
                ],
                temperature=0.1, max_tokens=400,
            ),
            timeout=18.0,
        )
        m = re.search(r"```(?:python)?\n?([\s\S]+?)```", raw)
        return m.group(1).strip() if m else raw.strip()
    except Exception as e:
        _logger.warning("TDD gen failed: %s", e)
        return None


async def run_tdd_check(code: str, executor: Any, llm: Any, language: str = "python") -> dict:
    """
    Ciclo TDD completo.
    Returns: { ran, passed, output, test_code }
    """
    out = {"ran": False, "passed": False, "output": "", "test_code": ""}
    if not _should_test(code, language):
        return out

    test_code = await _gen_test(code, llm)
    if not test_code:
        return out

    out["ran"] = True
    out["test_code"] = test_code
    combined = f"{code}\n\n# === AUTO-TEST S-GAP3 ===\n{test_code}"

    try:
        r = await asyncio.wait_for(
            executor.run_tool("run_python", {"code": combined}),
            timeout=28.0,
        )
        stdout  = r.get("output", r.get("stdout", ""))
        stderr  = r.get("stderr", "")
        ec      = r.get("exit_code", -1)
        passed  = ec == 0 and not stderr.strip()
        out["passed"] = passed
        out["output"] = (stdout or "")[:400]
        if stderr:
            out["output"] += f"\n[stderr] {stderr[:150]}"
        _logger.info("TDD: passed=%s exit=%s", passed, ec)
    except asyncio.TimeoutError:
        out["output"] = "⏱ timeout 28s"
    except Exception as e:
        out["output"] = f"⚠️ {e}"
    return out

# ── COG-3: TypeScript TDD ─────────────────────────────────────────────────────

_TS_FILE_RE = re.compile(r'\.(ts|tsx)$', re.IGNORECASE)
_TS_CODE_RE = re.compile(
    r'(interface |type |const |function |class |export |import |=>|async )',
    re.MULTILINE,
)


def _should_test_ts(content: str, path: str = '') -> bool:
    """COG-3: decides whether to type-check a TypeScript file change."""
    if path and not _TS_FILE_RE.search(path):
        return False
    lines = [l for l in content.splitlines() if l.strip() and not l.strip().startswith('//')]
    return len(lines) >= 5 and bool(_TS_CODE_RE.search(content))


async def run_tdd_check_ts(
    content: str,
    path: str,
    executor: Any,
    on_warn: Any = None,
) -> dict:
    """
    COG-3: TypeScript TDD — esegue type_check dopo apply_patch / write_file su .ts/.tsx.

    Pipeline:
      1. _should_test_ts() — guard rapido, zero overhead su file non-TS
      2. executor.run_tool('type_check', {'path': path}) — tsc --noEmit
      3. Se errori TypeScript rilevati: emette warning via on_warn callback

    Returns: { ran, passed, output, path }
    """
    out = {'ran': False, 'passed': False, 'output': '', 'path': path}

    if not _should_test_ts(content, path):
        return out

    if not executor:
        return out

    out['ran'] = True
    try:
        r = await asyncio.wait_for(
            executor.run_tool('type_check', {'path': path}),
            timeout=20.0,
        )
        raw_out = r.get('output', {})
        stdout  = raw_out.get('stdout', '') if isinstance(raw_out, dict) else str(raw_out)
        stderr  = raw_out.get('stderr', '') if isinstance(raw_out, dict) else ''
        ec      = r.get('exit_code', r.get('output', {}).get('exit_code', -1) if isinstance(r.get('output'), dict) else -1)

        has_errors = (
            'error TS' in (stdout + stderr)
            or (isinstance(ec, int) and ec != 0)
        )
        out['passed'] = not has_errors
        out['output'] = (stdout + stderr)[:500]

        if not out['passed'] and on_warn:
            ts_errors = [l for l in (stdout + stderr).splitlines() if 'error TS' in l][:5]
            warn_msg = f'TypeScript errors in {path}:\n' + '\n'.join(ts_errors)
            try:
                import asyncio as _asyncio
                _v = on_warn({'action': 'tdd_ts', 'status': 'warn', 'output': warn_msg})
                if _asyncio.iscoroutine(_v):
                    await _v
            except Exception as _exc:
                _logger.debug("[tdd_runner] silenced %s", type(_exc).__name__)  # noqa: BLE001

        _logger.info('COG-3 TS TDD: path=%s passed=%s ec=%s', path, out['passed'], ec)
    except asyncio.TimeoutError:
        out['output'] = 'TS type_check timeout 20s'
        _logger.warning('COG-3 TS TDD timeout for %s', path)
    except Exception as exc:
        out['output'] = f'TS TDD error: {exc}'
        _logger.warning('COG-3 TS TDD exception: %s', exc)

    return out