app.py

"""Public Space UI: reviewable dry-runs and a separate explicit submission."""

from __future__ import annotations

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

import gradio as gr
import pandas as pd
from huggingface_hub import get_token

from agent import GaiaAgent
from agent_code import resolve_agent_code
from cache import AnswerCache, ResultStore
from config import Settings
from evaluation import run_evaluation, submission_answers
from gaia_client import GaiaClient

SETTINGS = Settings.from_env()
PROJECT_ROOT = Path(__file__).resolve().parent
RESULT_STORE = ResultStore(Path(os.getenv("GAIA_RESULTS_PATH", "results/results.json")))


def has_huggingface_login() -> bool:
    """Detect Space OAuth or local CLI auth without retaining or logging the token."""
    if os.getenv("SPACE_ID") or os.getenv("HF_TOKEN"):
        return True
    try:
        return bool(get_token())
    except OSError:
        return False


def _frame(results: list[dict[str, Any]]) -> pd.DataFrame:
    return pd.DataFrame(
        [
            {
                "Task ID": row.get("task_id", ""),
                "Task Type": row.get("task_type", ""),
                "Question": row.get("question", ""),
                "Generated Answer": row.get("answer", ""),
                "Seconds": row.get("duration_seconds", 0),
                "Status": row.get("status", ""),
                "Error": row.get("error", ""),
                "Evidence": json.dumps(row.get("evidence", []), ensure_ascii=False),
            }
            for row in results
        ]
    )


def _ordered_all(client: GaiaClient) -> list[dict[str, Any]]:
    ids = [str(task["task_id"]) for task in client.get_questions()]
    return RESULT_STORE.ordered(ids)


def run_dry_evaluation(force: bool = False) -> tuple[str, pd.DataFrame]:
    """Answer every task and checkpoint results; this function cannot submit."""
    try:
        client = GaiaClient(SETTINGS)
        results = run_evaluation(
            client,
            GaiaAgent(SETTINGS),
            RESULT_STORE,
            force=bool(force),
        )
        completed = sum(row["status"] == "ok" for row in results)
        failed = len(results) - completed
        status = (
            f"Dry run complete: {completed}/{len(results)} answered; {failed} failed. "
            "No answers were submitted."
        )
        if len(results) == 20 and failed == 0:
            status += " All 20 unique tasks are ready for explicit submission."
        return status, _frame(results)
    except Exception as exc:
        return f"Dry run failed: {type(exc).__name__}: {exc}", _frame(
            RESULT_STORE.ordered()
        )


def rerun_task(task_id: str) -> tuple[str, pd.DataFrame]:
    """Force one task to run again while retaining other checkpoints."""
    task_id = str(task_id or "").strip()
    if not task_id:
        return "Enter a Task ID to rerun.", _frame(RESULT_STORE.ordered())
    try:
        client = GaiaClient(SETTINGS)
        run_evaluation(
            client,
            GaiaAgent(
                SETTINGS,
                AnswerCache(SETTINGS.cache_dir / "answers.json", enabled=False),
            ),
            RESULT_STORE,
            force=True,
            task_ids={task_id},
        )
        results = _ordered_all(client)
        row = RESULT_STORE.load()[task_id]
        return f"Rerun {task_id}: {row['status']}", _frame(results)
    except Exception as exc:
        return f"Rerun failed: {type(exc).__name__}: {exc}", _frame(
            RESULT_STORE.ordered()
        )


def submit_cached_answers(profile: gr.OAuthProfile | None) -> tuple[str, pd.DataFrame]:
    """The sole explicit route to the official submission POST."""
    try:
        client = GaiaClient(SETTINGS)
        results = _ordered_all(client)
    except Exception as exc:
        return f"Could not validate cached run: {type(exc).__name__}: {exc}", _frame(
            RESULT_STORE.ordered()
        )
    if profile is None:
        return "Please log in to Hugging Face before submitting.", _frame(results)
    try:
        answers = submission_answers(results, expected_count=20)
        agent_code = resolve_agent_code(
            space_id=os.getenv("SPACE_ID"),
            configured_url=SETTINGS.agent_code_url,
            allow_inline=SETTINGS.allow_inline_agent_code,
            root=PROJECT_ROOT,
        )
        response = client.submit_answers(
            username=str(profile.username).strip(),
            agent_code=agent_code,
            answers=answers,
        )
        status = (
            "Submission successful.\n"
            f"User: {response.get('username', profile.username)}\n"
            f"Overall Score: {response.get('score', 'N/A')}% "
            f"({response.get('correct_count', '?')}/{response.get('total_attempted', '?')} correct)\n"
            f"Message: {response.get('message', 'No message received.')}"
        )
        return status, _frame(results)
    except Exception as exc:
        return f"Submission not sent: {type(exc).__name__}: {exc}", _frame(results)


with gr.Blocks() as demo:
    gr.Markdown("# Modular GAIA Level-1 Agent")
    gr.Markdown(
        "Run or resume a dry evaluation, review all answers/evidence, and rerun individual "
        "tasks. Submission is a distinct action and is enabled logically only when exactly 20 "
        "unique tasks have non-empty successful answers. Keep this Space public."
    )
    # Outside Spaces, Gradio mocks OAuth from the token saved by `hf auth login`.
    # Only test whether one exists; model/ASR credentials remain environment-only.
    if has_huggingface_login():
        gr.LoginButton()
    else:
        gr.Markdown(
            "Local mode: run `hf auth login`, then expose `HF_TOKEN` to this process for "
            "inference. Configure `GAIA_AGENT_CODE_URL`, or explicitly enable inline source."
        )
    with gr.Row():
        force_all = gr.Checkbox(label="Force rerun all tasks", value=False)
        run_button = gr.Button("Run / Resume Dry Evaluation", variant="primary")
        submit_button = gr.Button("Submit Reviewed Complete Run", variant="secondary")
    with gr.Row():
        rerun_id = gr.Textbox(label="Task ID to rerun")
        rerun_button = gr.Button("Rerun Selected Task")
    status_output = gr.Textbox(label="Status", lines=5, interactive=False)
    results_table = gr.DataFrame(label="20-question review", wrap=True)
    run_button.click(
        run_dry_evaluation, inputs=[force_all], outputs=[status_output, results_table]
    )
    rerun_button.click(
        rerun_task, inputs=[rerun_id], outputs=[status_output, results_table]
    )
    submit_button.click(submit_cached_answers, outputs=[status_output, results_table])


if __name__ == "__main__":
    demo.queue(default_concurrency_limit=1).launch(debug=False, share=False)