src/ai_agentic_coder/crewai_wrapper.py

#!/usr/bin/env python

import subprocess
import time
import threading
import re
import json
import os
from pathlib import Path
import gradio as gr

from src.ai_agentic_coder.crew import EngineeringTeam
from src.ai_agentic_coder.tools.python_code_run_tool import RUN_RESULT_FILE


OUTPUT_DIR = Path(__file__).resolve().parent / "output"


def _run_result_path() -> Path:
    return OUTPUT_DIR / RUN_RESULT_FILE


def _clear_run_result() -> None:
    try:
        _run_result_path().unlink()
    except FileNotFoundError:
        pass


def _load_run_result() -> dict[str, str]:
    try:
        data = json.loads(_run_result_path().read_text(encoding="utf-8"))
    except (FileNotFoundError, json.JSONDecodeError):
        return {}

    return {
        "download_url": str(data.get("download_url", "")),
        "live_url": str(data.get("live_url", "")),
    }


def _set_base_url_from_request(request: gr.Request | None) -> None:
    if request is None:
        return

    host = request.headers.get("x-forwarded-host") or request.headers.get("host")
    if not host:
        return

    proto = request.headers.get("x-forwarded-proto") or request.url.scheme or "http"
    os.environ["AI_AGENTIC_CODER_BASE_URL"] = f"{proto}://{host}"

def run_crew(requirements, module_name, class_name, progress=gr.Progress()):
    """Run the crew with the given inputs and return ONLY the raw result text."""
    try:
        # Kill any running processes from previous runs
        for pattern in (
            "python.*src.ai_agentic_coder.output.app",
            "python.*ai_agentic_coder.generated_app_runner",
        ):
            subprocess.run(
                ["pkill", "-f", pattern],
                stderr=subprocess.DEVNULL,
                stdout=subprocess.DEVNULL
            )

        inputs = {
            'requirements': requirements,
            'module_name': f"{module_name}.py",  # Add .py extension
            'class_name': class_name
        }

        # Baseline progress hints (wrapper controls real-time progress)
        progress(0.02, desc="Initializing crew...")
        crew = EngineeringTeam().crew()

        progress(0.05, desc="Running crew...")
        result = crew.kickoff(inputs=inputs)

        raw = str(result.raw) if hasattr(result, 'raw') else str(result)

        return raw

    except Exception as e:
        return f"❌ Error: {str(e)}"

# Generator function to manage state, progress, and output
def run_crew_wrapper(requirements, module_name, class_name, request: gr.Request | None = None):
    _clear_run_result()
    _set_base_url_from_request(request)

    progress = gr.Progress()
    # Immediately disable the button, show Output as progress area, and hide URL boxes
    yield (
        gr.update(value="[░░░░░░░░░░░░░░░░░░░░] 1% - AI is running hard", visible=True, label="Progress"),
        gr.update(value="", visible=False),  # clear Download URL box at start
        gr.update(value="", visible=False),  # clear Live URL box at start
        gr.update(interactive=False, value="Running…")
    )
    # Kick progress so overlay appears immediately
    progress(0.01, desc="Starting…")

    done = threading.Event()
    result_holder: dict[str, str | None] = {"output": None, "error": None}

    def worker():
        try:
            result_holder["output"] = run_crew(requirements, module_name, class_name, progress)
        except Exception as e:  # pragma: no cover
            result_holder["error"] = str(e)
        finally:
            done.set()

    t = threading.Thread(target=worker, daemon=True)
    t.start()

    start = time.time()
    total = 300.0  # ~5 minutes
    last_pct = 1
    while not done.is_set():
        elapsed = time.time() - start
        pct = min(95, max(1, int((elapsed / total) * 95)))
        if pct != last_pct:
            # Overlay progress and textual progress in output box
            progress(pct / 100.0, desc=f"AI is running hard")
            last_pct = pct
            bar_len = 24
            filled = int(bar_len * pct / 100)
            bar = ("█" * filled) + ("░" * (bar_len - filled))
            yield (
                gr.update(value=f"[{bar}] {pct}% - AI is running hard", visible=True, label="Progress"),
                gr.update(visible=False),
                gr.update(visible=False),
                gr.update(interactive=False, value="Running…")
            )
        time.sleep(1.0)

    # Worker is done
    t.join()
    progress(1.0, desc="Finalizing…")

    if result_holder["error"]:
        final_output = f"❌ Error: {result_holder['error']}"
        # Show error in output, keep URL boxes hidden, re-enable button
        yield (
            gr.update(value=final_output, visible=True, label="Error", elem_classes=["card", "error"]),
            gr.update(visible=False),
            gr.update(visible=False),
            gr.update(interactive=True, value="Run AI Coder")
        )
        return

    raw_output = result_holder["output"] or ""
    run_result = _load_run_result()

    if run_result.get("download_url") and run_result.get("live_url"):
        yield (
            gr.update(value="", visible=False, label="Output"),
            gr.update(value=run_result["download_url"], visible=True),
            gr.update(value=run_result["live_url"], visible=True),
            gr.update(interactive=True, value="Run AI Coder")
        )
        return

    # If the output looks like a failure (but no exception was thrown), treat it as an error
    failure_like = bool(re.search(r"\b(failed|failure|error|exception|traceback)\b", raw_output, re.IGNORECASE))

    # Extract and validate URLs strictly (avoid putting random text into URL boxes)
    rough_urls = re.findall(r"https?://[^\s\"'<>]+", raw_output)
    cleaned_urls = [u.rstrip(".,);]") for u in rough_urls]
    valid_urls = [u for u in cleaned_urls if re.match(r"^https?://(?:[A-Za-z0-9.-]+|localhost|(?:\d{1,3}\.){3}\d{1,3})(?::\d+)?(?:/\S*)?$", u)]

    if failure_like or len(valid_urls) == 0:
        final_output = f"❌ Error: {raw_output.strip()}"
        yield (
            gr.update(value=final_output, visible=True, label="Error", elem_classes=["card", "error"]),
            gr.update(value="", visible=False),
            gr.update(value="", visible=False),
            gr.update(interactive=True, value="Run AI Coder")
        )
        return

    # Use up to two validated URLs
    download_url = valid_urls[0] if len(valid_urls) > 0 else ""
    live_url = valid_urls[1] if len(valid_urls) > 1 else ""

    # Success: hide output, show URLs, re-enable button
    yield (
        gr.update(value="", visible=False, label="Output"),
        gr.update(value=download_url, visible=bool(download_url)),
        gr.update(value=live_url, visible=bool(live_url)),
        gr.update(interactive=True, value="Run AI Coder")
    )