app.py

from __future__ import annotations

import json
import os
import threading
from pathlib import Path
from typing import Any

import gradio as gr
import pandas as pd
import requests

from agent import GaiaAgent
import progress as store


DEFAULT_API_URL = os.getenv(
    "GAIA_SCORING_API",
    "https://agents-course-unit4-scoring.hf.space",
)

CACHE_FILE = Path("answers_cache.json")

# Hard cap per question (seconds). 0 disables it.
QUESTION_TIMEOUT = float(os.getenv("QUESTION_TIMEOUT_SECONDS", "240"))
# Sent to the scorer for a question that has no answer ("Submit anyway").
PLACEHOLDER_ANSWER = os.getenv("PLACEHOLDER_ANSWER", "unknown").strip() or "unknown"
# Refuse "Submit anyway" if more questions than this have no answer.
MAX_PLACEHOLDERS = int(os.getenv("MAX_PLACEHOLDERS", "5"))


def solve_with_timeout(agent: GaiaAgent, question: str, task_id: str, file_name: str) -> str:
    """Run agent.solve but give up after QUESTION_TIMEOUT seconds.

    Python cannot kill a thread, so a timed-out question may keep running in
    the background (using some API quota) until it finishes by itself; its
    late result is simply discarded. Each solve() builds its own agent, so it
    cannot interfere with the next question.
    """
    def _call() -> str:
        return agent.solve(question=question, task_id=task_id, file_name=file_name)

    if QUESTION_TIMEOUT <= 0:
        return _call()

    box: dict[str, str] = {}

    def _target() -> None:
        try:
            box["answer"] = _call()
        except Exception as exc:  # noqa: BLE001
            box["answer"] = f"AGENT_ERROR: {type(exc).__name__}: {exc}"

    worker = threading.Thread(target=_target, daemon=True)
    worker.start()
    worker.join(QUESTION_TIMEOUT)
    if worker.is_alive():
        print(f"[TIMEOUT] task {task_id} exceeded {int(QUESTION_TIMEOUT)}s; moving on.")
        return f"AGENT_ERROR: Timeout after {int(QUESTION_TIMEOUT)}s"
    return box.get("answer", "AGENT_ERROR: no answer produced")


def _request_json(method: str, url: str, **kwargs) -> Any:
    response = requests.request(method, url, timeout=120, **kwargs)
    response.raise_for_status()
    return response.json()


def get_questions() -> list[dict]:
    data = _request_json("GET", f"{DEFAULT_API_URL}/questions")
    if not isinstance(data, list):
        raise ValueError("Unexpected /questions response.")
    return data


def get_agent_code_url() -> str:
    explicit = os.getenv("AGENT_CODE_URL", "").strip()
    if explicit:
        return explicit

    space_id = os.getenv("SPACE_ID", "").strip()
    if space_id:
        return f"https://huggingface.co/spaces/{space_id}/tree/main"

    return ""


def get_username(profile: gr.OAuthProfile | None, manual_username: str) -> str:
    if profile and profile.username:
        return profile.username.strip()
    return manual_username.strip()


def run_single(
    task_id: str,
    questions_json: str,
) -> tuple[str, pd.DataFrame | None]:
    try:
        questions = json.loads(questions_json) if questions_json else get_questions()
        if not isinstance(questions, list):
            return "Questions JSON must contain a list.", None

        task_id = task_id.strip()
        task_map = {str(q["task_id"]): q for q in questions}
        if task_id not in task_map:
            return f"Unknown task_id: {task_id}", None

        agent = GaiaAgent(task_map)
        item = task_map[task_id]
        answer = solve_with_timeout(
            agent,
            item["question"],
            task_id,
            item.get("file_name", ""),
        )
        store.save_answer(task_id, answer)  # a full run will reuse good answers
        row = pd.DataFrame(
            [{
                "Task ID": task_id,
                "Question": item["question"],
                "File": item.get("file_name", ""),
                "Answer": answer,
            }]
        )
        return "Single-task run complete.", row
    except Exception as exc:
        return f"Single-task run failed: {type(exc).__name__}: {exc}", None


def clear_progress() -> str:
    store.clear()
    try:
        CACHE_FILE.unlink()
    except FileNotFoundError:
        pass
    return (
        "Saved progress and cached submission cleared. "
        "The next evaluation will solve every question again."
    )


def run_evaluation(
    profile: gr.OAuthProfile | None,
    manual_username: str,
    progress=gr.Progress(),
):
    """Generator: streams the table after every question and resumes from disk."""
    try:
        # A previous completed run must not remain submit-ready if this run
        # fails before producing a fresh, complete answer set.
        try:
            CACHE_FILE.unlink()
        except FileNotFoundError:
            pass

        questions = get_questions()
        if len(questions) != 20:
            status_note = (
                f"Warning: scorer returned {len(questions)} questions, not 20. "
                "The official set is expected to contain 20."
            )
        else:
            status_note = "Loaded the official 20-question subset."

        username = get_username(profile, manual_username)
        if not username:
            yield "Please sign in or enter your HF username.", None, json.dumps(questions)
            return

        task_map = {str(q["task_id"]): q for q in questions}
        # Fail fast with a clear message if no LLM provider is configured.
        # Each question below gets its own GaiaAgent, so a timed-out question
        # that is still running in the background can never share an object
        # with the next one.
        GaiaAgent(task_map)
        saved = store.load()

        results = []
        answers_payload = []
        total = len(questions)

        for index, item in enumerate(questions, start=1):
            task_id = str(item["task_id"])
            question = item.get("question", "")
            file_name = item.get("file_name", "")
            progress((index - 1) / total, desc=f"Question {index}/{total}")

            if task_id in saved and not store.is_error(saved[task_id]):
                answer, source = saved[task_id], "saved"
            else:
                try:
                    question_agent = GaiaAgent(task_map)
                    answer = solve_with_timeout(
                        question_agent, question, task_id, file_name
                    )
                except Exception as exc:
                    answer = f"AGENT_ERROR: {type(exc).__name__}: {exc}"
                source = "solved" if not store.is_error(answer) else "FAILED"
                store.save_answer(task_id, answer)

            answers_payload.append({
                "task_id": task_id,
                "submitted_answer": answer,
            })
            results.append({
                "Task": index,
                "Task ID": task_id,
                "Question": question,
                "File": file_name,
                "Answer": answer,
                "Source": source,
            })
            failed = sum(1 for r in results if r["Source"] == "FAILED")
            yield (
                f"{status_note}\nProgress: {index}/{total} "
                f"({failed} failed so far). Running...",
                pd.DataFrame(results),
                json.dumps(questions),
            )

        cache = {
            "username": username,
            "agent_code": get_agent_code_url(),
            "answers": answers_payload,
        }
        CACHE_FILE.write_text(json.dumps(cache, indent=2), encoding="utf-8")

        code_note = (
            get_agent_code_url()
            or "No public agent_code URL detected yet. Set AGENT_CODE_URL locally or deploy this project as a public HF Space."
        )
        failed = [r["Task"] for r in results if r["Source"] == "FAILED"]
        retry_note = (
            f"Questions with no answer (error/timeout): {failed}. Rerun to retry "
            "only these (solved answers are kept), or use 'Submit anyway' to send "
            f"'{PLACEHOLDER_ANSWER}' for them.\n"
            if failed else ""
        )

        status = (
            f"{status_note}\n"
            f"Generated {len(answers_payload)} answers for {username}.\n"
            f"{retry_note}"
            f"Cached to {CACHE_FILE}.\n"
            f"agent_code: {code_note}\n"
            "Review the table, then use Submit cached answers."
        )
        yield status, pd.DataFrame(results), json.dumps(questions)
    except Exception as exc:
        yield f"Evaluation failed: {type(exc).__name__}: {exc}", None, ""


def submit_cached(
    profile: gr.OAuthProfile | None,
    manual_username: str,
    fill_failed: bool = False,
) -> str:
    try:
        if not CACHE_FILE.exists():
            return "No cached answers found. Run the 20-question evaluation first."

        cache = json.loads(CACHE_FILE.read_text(encoding="utf-8"))
        cached_answers = cache.get("answers")
        if not isinstance(cached_answers, list):
            return "Cached answers are invalid. Run the evaluation again."

        # Never send quota/error placeholders to the exact-match scorer. Failed
        # tasks are deliberately not persisted by progress.py, so rerunning the
        # evaluation retries them while reusing all successful answers.
        failed = [
            str(answer.get("task_id", "?"))
            for answer in cached_answers
            if not isinstance(answer, dict)
            or not answer.get("task_id")
            or not isinstance(answer.get("submitted_answer"), str)
            or not answer.get("submitted_answer", "").strip()
            or store.is_error(answer.get("submitted_answer"))
        ]
        placeholder_note = ""
        if failed and not fill_failed:
            return (
                "Submission blocked because some tasks did not produce an answer. "
                f"Failed task IDs: {', '.join(failed)}. Run the evaluation again, "
                "or use 'Submit anyway' to send a placeholder for them."
            )
        if failed and fill_failed:
            if len(failed) > MAX_PLACEHOLDERS:
                return (
                    f"Refusing to submit: {len(failed)} questions have no answer "
                    f"(limit {MAX_PLACEHOLDERS}). That looks like a quota or setup "
                    "problem, not a few hard questions. Fix it and rerun."
                )
            failed_set = set(failed)
            cached_answers = [
                {**a, "submitted_answer": PLACEHOLDER_ANSWER}
                if isinstance(a, dict) and str(a.get("task_id", "?")) in failed_set
                else a
                for a in cached_answers
            ]
            placeholder_note = (
                f"Sent '{PLACEHOLDER_ANSWER}' for {len(failed)} unanswered "
                f"question(s): {', '.join(failed)}\n"
            )
        if any(not isinstance(a, dict) or not a.get("task_id") for a in cached_answers):
            return "Cached answers are malformed. Run the evaluation again."
        if len(cached_answers) != 20:
            return (
                f"Submission blocked: expected 20 answers but found "
                f"{len(cached_answers)}. Run the full evaluation again."
            )

        username = get_username(profile, manual_username) or cache.get("username", "")
        agent_code = cache.get("agent_code") or get_agent_code_url()

        if not username:
            return "Please sign in or enter your HF username."
        if not agent_code:
            return (
                "No public agent_code URL is available. "
                "Deploy this project as a public Hugging Face Space or set AGENT_CODE_URL."
            )

        payload = {
            "username": username,
            "agent_code": agent_code,
            "answers": cached_answers,
        }

        result = _request_json(
            "POST",
            f"{DEFAULT_API_URL}/submit",
            json=payload,
        )

        return (
            "Submission complete.\n"
            f"{placeholder_note}"
            f"User: {result.get('username', username)}\n"
            f"Score: {result.get('score', 'N/A')}%\n"
            f"Correct: {result.get('correct_count', '?')}/"
            f"{result.get('total_attempted', '?')}\n"
            f"Message: {result.get('message', '')}"
        )
    except requests.HTTPError as exc:
        detail = ""
        try:
            detail = exc.response.json().get("detail", exc.response.text)
        except Exception:
            detail = exc.response.text if exc.response is not None else ""
        return f"Submission failed: HTTP {exc.response.status_code}: {detail}"
    except Exception as exc:
        return f"Submission failed: {type(exc).__name__}: {exc}"


def submit_anyway(
    profile: gr.OAuthProfile | None,
    manual_username: str,
) -> str:
    return submit_cached(profile, manual_username, fill_failed=True)


with gr.Blocks(title="GAIA Agent — Unit 4") as demo:
    gr.Markdown(
        """
# GAIA Agent — Unit 4

A from-scratch `smolagents` CodeAgent for the Hugging Face Agents Course final project.

The evaluator is deliberately separated from the agent so you can inspect answers
before sending them to the official scorer.
"""
    )

    with gr.Row():
        manual_username = gr.Textbox(
            label="HF username (optional when signed in)",
            placeholder="your-huggingface-username",
        )
        gr.LoginButton()

    with gr.Row():
        run_button = gr.Button("Run 20-question evaluation", variant="primary")
        clear_button = gr.Button("Clear saved progress")
    status_output = gr.Textbox(
        label="Evaluation status",
        lines=8,
        interactive=False,
    )
    results_table = gr.DataFrame(
        label="Generated answers",
        wrap=True,
        interactive=False,
    )

    gr.Markdown("## Submit")
    gr.Markdown(
        "The submit button uses the cached answers from the latest full evaluation. "
        "This lets you review/debug without rerunning all 20 tasks."
    )
    with gr.Row():
        submit_button = gr.Button("Submit cached answers")
        submit_anyway_button = gr.Button(
            "Submit anyway (placeholder for unanswered)", variant="secondary"
        )
    submission_output = gr.Textbox(
        label="Submission result",
        lines=8,
        interactive=False,
    )

    gr.Markdown("## Test one task")
    task_id_input = gr.Textbox(
        label="Task ID",
        placeholder="Paste one task_id from the table",
    )
    test_button = gr.Button("Run one task")
    single_status = gr.Textbox(
        label="Single-task status",
        interactive=False,
    )
    single_table = gr.DataFrame(
        label="Single-task result",
        wrap=True,
        interactive=False,
    )

    questions_state = gr.State("")

    run_button.click(
        fn=run_evaluation,
        inputs=[manual_username],
        outputs=[status_output, results_table, questions_state],
    )

    clear_button.click(fn=clear_progress, inputs=[], outputs=[status_output])

    submit_button.click(
        fn=submit_cached,
        inputs=[manual_username],
        outputs=[submission_output],
    )

    submit_anyway_button.click(
        fn=submit_anyway,
        inputs=[manual_username],
        outputs=[submission_output],
    )

    test_button.click(
        fn=run_single,
        inputs=[task_id_input, questions_state],
        outputs=[single_status, single_table],
    )


if __name__ == "__main__":
    demo.launch()