Update app.py

import glob
import json
import os
import re
import threading
import time
from collections import deque
from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED

import gradio as gr
import pandas as pd
import requests
from smolagents import (
CodeAgent,
DuckDuckGoSearchTool,
InferenceClientModel,
LiteLLMModel,
VisitWebpageTool,
tool,
)

# ZeroGPU 하드웨어에서 시작 오류를 막기 위한 더미 함수 (CPU 하드웨어에서는 무시됨)
try:
import spaces

@spaces.GPU
def _gpu_placeholder():
return "ok"
except ImportError:
pass

# ---------------------------------------------------------------------------
# 설정 (Space의 Variables에서 조정 가능)
# ---------------------------------------------------------------------------
DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
FILES_DIR = "/tmp/task_files"
CACHE_PATH = "/tmp/answers_cache.json"
os.makedirs(FILES_DIR, exist_ok=True)

WORKERS = int(os.getenv("WORKERS", "2")) # 동시 실행 문항 수
QUESTION_TIMEOUT = int(os.getenv("QUESTION_TIMEOUT", "300")) # 문항당 제한 시간(초)
MAX_STEPS = int(os.getenv("MAX_STEPS", "8")) # 에이전트 최대 단계
MODEL_TIMEOUT = int(os.getenv("MODEL_TIMEOUT", "90")) # LLM 1회 호출 제한(초)
HEARTBEAT = 3 # 화면 갱신 주기(초)

ANSWER_RULES = """You are a careful research and reasoning agent solving GAIA-style questions.
Rules:
1. Use the tools (web search, webpage visit, file reading, python code) whenever the
question needs facts you are not 100% sure about. Do not guess.
2. The final answer is graded by EXACT MATCH. Return ONLY the answer value, with no
explanation, no units unless asked, no trailing period, no "FINAL ANSWER:" prefix.
3. If the answer is a number: no thousands separators, no currency/percent symbols
unless the question asks for them.
4. If the answer is a list: comma-separated, in the order the question requests.
5. If the answer is a string: no articles (a, an, the) and no abbreviations unless
the question asks for them.
6. If a file path is given, read it with the read_local_file tool or with python code.
7. Be efficient: stop researching as soon as you have a reliable answer.
8. Finish by calling final_answer(...) with the answer only."""

# ---------------------------------------------------------------------------
# 이벤트 로그 (화면 상태창에 최근 기록을 보여주기 위함)
# ---------------------------------------------------------------------------
_events = deque(maxlen=10)
_active: dict = {} # task_id -> 시작 시각
_state_lock = threading.Lock()


def log(msg: str):
line = f"[{time.strftime('%H:%M:%S')}] {msg}"
print(line, flush=True)
with _state_lock:
_events.append(line)


def recent_log() -> str:
with _state_lock:
return "\n".join(_events)


# ---------------------------------------------------------------------------
# 캐시
# ---------------------------------------------------------------------------
_cache_lock = threading.Lock()


def load_cache() -> dict:
try:
with open(CACHE_PATH, "r", encoding="utf-8") as f:
return json.load(f)
except Exception: # noqa: BLE001
return {}


def save_cache_entry(task_id: str, answer: str):
with _cache_lock:
cache = load_cache()
cache[task_id] = answer
with open(CACHE_PATH, "w", encoding="utf-8") as f:
json.dump(cache, f, ensure_ascii=False)


# ---------------------------------------------------------------------------
# 도구
# ---------------------------------------------------------------------------
@tool
def read_local_file(path: str) -> str:
"""Reads a local file (txt, csv, json, py, md, xlsx) and returns its content as text.

Args:
path: Absolute or relative path of the file to read.
"""
try:
if not os.path.exists(path):
basename = os.path.basename(path)
matches = glob.glob(os.path.join(FILES_DIR, f"*{basename}"))
if matches:
path = matches[0]
lower = path.lower()
if lower.endswith((".xlsx", ".xls")):
sheets = pd.read_excel(path, sheet_name=None)
return "\n\n".join(
f"## Sheet: {n}\n{df.to_string(max_rows=200)}" for n, df in sheets.items()
)
if lower.endswith(".csv"):
return pd.read_csv(path).to_string(max_rows=200)
with open(path, "r", encoding="utf-8", errors="ignore") as f:
return f.read()[:20000]
except Exception as e: # noqa: BLE001
return f"Could not read file: {e}"


# ---------------------------------------------------------------------------
# 모델 / 에이전트
# ---------------------------------------------------------------------------
def gemini_model_id() -> str:
return os.getenv("GEMINI_MODEL", "gemini-3.8-flash").replace("gemini/", "")


def build_model():
gemini_key = os.getenv("GEMINI_API_KEY")
if gemini_key:
return LiteLLMModel(
model_id=f"gemini/{gemini_model_id()}",
api_key=gemini_key,
timeout=MODEL_TIMEOUT, # LLM 호출이 무한정 멈추지 않도록 제한
num_retries=2,
)
openai_key = os.getenv("OPENAI_API_KEY")
if openai_key:
return LiteLLMModel(
model_id=f"openai/{os.getenv('OPENAI_MODEL', 'gpt-4o-mini')}",
api_key=openai_key,
timeout=MODEL_TIMEOUT,
num_retries=2,
)
return InferenceClientModel(
model_id=os.getenv("HF_MODEL", "Qwen/Qwen2.5-Coder-32B-Instruct"),
token=os.getenv("HF_TOKEN"),
)


def build_agent() -> CodeAgent:
return CodeAgent(
tools=[DuckDuckGoSearchTool(), VisitWebpageTool(), read_local_file],
model=build_model(),
additional_authorized_imports=[
"pandas", "numpy", "re", "json", "math", "datetime",
"collections", "itertools", "statistics", "requests", "bs4",
],
max_steps=MAX_STEPS,
verbosity_level=1,
)


def clean_answer(answer) -> str:
text = str(answer).strip()
text = re.sub(r"^\s*final answer\s*:\s*", "", text, flags=re.IGNORECASE)
return text.strip().rstrip(".").strip()


# ---------------------------------------------------------------------------
# 채점 서버 통신
# ---------------------------------------------------------------------------
def fetch_questions(api_url: str) -> list[dict]:
resp = requests.get(f"{api_url}/questions", timeout=30)
resp.raise_for_status()
data = resp.json()
if not data:
raise ValueError("질문 목록이 비어 있습니다.")
return data


def download_task_file(api_url: str, task_id: str, file_name: str):
if not file_name:
return None
try:
resp = requests.get(f"{api_url}/files/{task_id}", timeout=60)
resp.raise_for_status()
path = os.path.join(FILES_DIR, f"{task_id}_{file_name}")
with open(path, "wb") as f:
f.write(resp.content)
return path
except Exception as e: # noqa: BLE001
log(f"파일 다운로드 실패 {task_id[:8]}: {e}")
return None


# ---------------------------------------------------------------------------
# 문항 풀이 (캐시 + 시간 제한)
# ---------------------------------------------------------------------------
def solve_one(item: dict, api_url: str) -> str:
prompt = ANSWER_RULES + "\n\nQuestion:\n" + item["question"]
file_path = download_task_file(api_url, item["task_id"], item.get("file_name", ""))
if file_path:
prompt += f"\n\nAn attached file is available at: {file_path}"
return clean_answer(build_agent().run(prompt))


def solve_with_timeout(item: dict, api_url: str):
"""(답변, 성공 여부, 비고)를 반환합니다."""
task_id = item["task_id"]
short = task_id[:8]
cached = load_cache().get(task_id)
if cached:
log(f"캐시 사용 {short} -> {cached}")
return cached, True, "cache"

box: dict = {}

def target():
try:
box["answer"] = solve_one(item, api_url)
except Exception as e: # noqa: BLE001
box["error"] = f"{type(e).__name__}: {e}"[:300]

log(f"시작 {short}")
with _state_lock:
_active[task_id] = time.time()
t = threading.Thread(target=target, daemon=True)
t.start()
t.join(QUESTION_TIMEOUT)
with _state_lock:
_active.pop(task_id, None)

if t.is_alive():
log(f"시간 초과 {short} ({QUESTION_TIMEOUT}초)")
return "unknown", False, f"timeout {QUESTION_TIMEOUT}s"
if "error" in box:
log(f"오류 {short}: {box['error']}")
return "unknown", False, box["error"]

save_cache_entry(task_id, box["answer"])
log(f"완료 {short} -> {box['answer']}")
return box["answer"], True, ""


def iter_solve(questions: list[dict], api_url: str):
"""끝난 문항은 (item, answer, ok, note)로, 대기 중에는 주기적으로 None을 내보냅니다."""
valid = [q for q in questions if q.get("task_id") and q.get("question") is not None]
pool = ThreadPoolExecutor(max_workers=WORKERS)
try:
pending = {pool.submit(solve_with_timeout, q, api_url): q for q in valid}
while pending:
done, _ = wait(list(pending), timeout=HEARTBEAT, return_when=FIRST_COMPLETED)
if not done:
yield None
continue
for fut in done:
item = pending.pop(fut)
answer, ok, note = fut.result()
yield item, answer, ok, note
finally:
pool.shutdown(wait=False)


def make_df(rows: dict, order: list[str]) -> pd.DataFrame:
data = [rows[tid] for tid in order if tid in rows]
return pd.DataFrame(data, columns=["Task ID", "Question", "Submitted Answer", "OK", "Note"])


def status_text(label: str, done: int, total: int, start: float) -> str:
with _state_lock:
running = len(_active)
oldest = int(time.time() - min

app.py

@@ -1,9 +1,12 @@
+import glob
 import json
 import os
 import re
 import threading
 import time
-from concurrent.futures import ThreadPoolExecutor, as_completed
+from collections import deque
+from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED
+
 import gradio as gr
 import pandas as pd
 import requests
@@ -16,9 +19,10 @@ from smolagents import (
     tool,
 )
 
-# ZeroGPU 하드웨어에서 시작 오류를 막기 위한 안전한 더미 함수
+# ZeroGPU 하드웨어에서 시작 오류를 막기 위한 더미 함수 (CPU 하드웨어에서는 무시됨)
 try:
     import spaces
+
     @spaces.GPU
     def _gpu_placeholder():
         return "ok"
@@ -26,16 +30,18 @@ except ImportError:
     pass
 
 # ---------------------------------------------------------------------------
-# 설정 (Space의 Variables 또는 Secrets에서 조정 가능)
+# 설정 (Space의 Variables에서 조정 가능)
 # ---------------------------------------------------------------------------
 DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 FILES_DIR = "/tmp/task_files"
 CACHE_PATH = "/tmp/answers_cache.json"
 os.makedirs(FILES_DIR, exist_ok=True)
 
-WORKERS = int(os.getenv("WORKERS", "3"))                      # 동시 실행 문항 수
-QUESTION_TIMEOUT = int(os.getenv("QUESTION_TIMEOUT", "240"))  # 문항당 제한 시간(초)
-MAX_STEPS = int(os.getenv("MAX_STEPS", "12"))                 # 에이전트 최대 단계
+WORKERS = int(os.getenv("WORKERS", "2"))                      # 동시 실행 문항 수
+QUESTION_TIMEOUT = int(os.getenv("QUESTION_TIMEOUT", "300"))  # 문항당 제한 시간(초)
+MAX_STEPS = int(os.getenv("MAX_STEPS", "8"))                  # 에이전트 최대 단계
+MODEL_TIMEOUT = int(os.getenv("MODEL_TIMEOUT", "90"))         # LLM 1회 호출 제한(초)
+HEARTBEAT = 3                                                 # 화면 갱신 주기(초)
 
 ANSWER_RULES = """You are a careful research and reasoning agent solving GAIA-style questions.
 Rules:
@@ -53,10 +59,31 @@ Rules:
 8. Finish by calling final_answer(...) with the answer only."""
 
 # ---------------------------------------------------------------------------
-# 캐시 관리
+# 이벤트 로그 (화면 상태창에 최근 기록을 보여주기 위함)
+# ---------------------------------------------------------------------------
+_events = deque(maxlen=10)
+_active: dict = {}  # task_id -> 시작 시각
+_state_lock = threading.Lock()
+
+
+def log(msg: str):
+    line = f"[{time.strftime('%H:%M:%S')}] {msg}"
+    print(line, flush=True)
+    with _state_lock:
+        _events.append(line)
+
+
+def recent_log() -> str:
+    with _state_lock:
+        return "\n".join(_events)
+
+
+# ---------------------------------------------------------------------------
+# 캐시
 # ---------------------------------------------------------------------------
 _cache_lock = threading.Lock()
 
+
 def load_cache() -> dict:
     try:
         with open(CACHE_PATH, "r", encoding="utf-8") as f:
@@ -64,6 +91,7 @@ def load_cache() -> dict:
     except Exception:  # noqa: BLE001
         return {}
 
+
 def save_cache_entry(task_id: str, answer: str):
     with _cache_lock:
         cache = load_cache()
@@ -71,23 +99,23 @@ def save_cache_entry(task_id: str, answer: str):
         with open(CACHE_PATH, "w", encoding="utf-8") as f:
             json.dump(cache, f, ensure_ascii=False)
 
+
 # ---------------------------------------------------------------------------
-# 도구 (파일 경로 보완 기능 포함)
+# 도구
 # ---------------------------------------------------------------------------
 @tool
 def read_local_file(path: str) -> str:
     """Reads a local file (txt, csv, json, py, md, xlsx) and returns its content as text.
+
     Args:
         path: Absolute or relative path of the file to read.
     """
     try:
-        # 에이전트가 파일명만 입력하거나 경로가 잘못된 경우 FILES_DIR에서 탐색
         if not os.path.exists(path):
             basename = os.path.basename(path)
-            candidate = os.path.join(FILES_DIR, basename)
-            if os.path.exists(candidate):
-                path = candidate
-
+            matches = glob.glob(os.path.join(FILES_DIR, f"*{basename}"))
+            if matches:
+                path = matches[0]
         lower = path.lower()
         if lower.endswith((".xlsx", ".xls")):
             sheets = pd.read_excel(path, sheet_name=None)
@@ -101,28 +129,38 @@ def read_local_file(path: str) -> str:
     except Exception as e:  # noqa: BLE001
         return f"Could not read file: {e}"
 
+
 # ---------------------------------------------------------------------------
-# 에이전트 및 모델 설정 (gemini-3.8-flash 반영)
+# 모델 / 에이전트
 # ---------------------------------------------------------------------------
+def gemini_model_id() -> str:
+    return os.getenv("GEMINI_MODEL", "gemini-3.8-flash").replace("gemini/", "")
+
+
 def build_model():
     gemini_key = os.getenv("GEMINI_API_KEY")
     if gemini_key:
-        model_id = os.getenv("GEMINI_MODEL", "gemini-3.8-flash")
-        print(f"[model] Gemini: {model_id}")
-        return LiteLLMModel(model_id=f"gemini/{model_id}", api_key=gemini_key)
-    
+        return LiteLLMModel(
+            model_id=f"gemini/{gemini_model_id()}",
+            api_key=gemini_key,
+            timeout=MODEL_TIMEOUT,   # LLM 호출이 무한정 멈추지 않도록 제한
+            num_retries=2,
+        )
     openai_key = os.getenv("OPENAI_API_KEY")
     if openai_key:
-        model_id = os.getenv("OPENAI_MODEL", "gpt-4o-mini")
-        print(f"[model] OpenAI: {model_id}")
-        return LiteLLMModel(model_id=f"openai/{model_id}", api_key=openai_key)
-    
-    model_id = os.getenv("HF_MODEL", "Qwen/Qwen2.5-Coder-32B-Instruct")
-    print(f"[model] HF Inference: {model_id}")
-    return InferenceClientModel(model_id=model_id, token=os.getenv("HF_TOKEN"))
+        return LiteLLMModel(
+            model_id=f"openai/{os.getenv('OPENAI_MODEL', 'gpt-4o-mini')}",
+            api_key=openai_key,
+            timeout=MODEL_TIMEOUT,
+            num_retries=2,
+        )
+    return InferenceClientModel(
+        model_id=os.getenv("HF_MODEL", "Qwen/Qwen2.5-Coder-32B-Instruct"),
+        token=os.getenv("HF_TOKEN"),
+    )
+
 
 def build_agent() -> CodeAgent:
-    # 스레드 안전 및 문항별 독립 컨텍스트 유지를 위해 매번 새 에이전트를 생성합니다.
     return CodeAgent(
         tools=[DuckDuckGoSearchTool(), VisitWebpageTool(), read_local_file],
         model=build_model(),
@@ -134,13 +172,15 @@ def build_agent() -> CodeAgent:
         verbosity_level=1,
     )
 
+
 def clean_answer(answer) -> str:
     text = str(answer).strip()
     text = re.sub(r"^\s*final answer\s*:\s*", "", text, flags=re.IGNORECASE)
     return text.strip().rstrip(".").strip()
 
+
 # ---------------------------------------------------------------------------
-# 채점 서버 통신 및 파일 다운로드
+# 채점 서버 통신
 # ---------------------------------------------------------------------------
 def fetch_questions(api_url: str) -> list[dict]:
     resp = requests.get(f"{api_url}/questions", timeout=30)
@@ -150,6 +190,7 @@ def fetch_questions(api_url: str) -> list[dict]:
         raise ValueError("질문 목록이 비어 있습니다.")
     return data
 
+
 def download_task_file(api_url: str, task_id: str, file_name: str):
     if not file_name:
         return None
@@ -161,66 +202,118 @@ def download_task_file(api_url: str, task_id: str, file_name: str):
             f.write(resp.content)
         return path
     except Exception as e:  # noqa: BLE001
-        print(f"파일 다운로드 실패 ({task_id}): {e}")
+        log(f"파일 다운로드 실패 {task_id[:8]}: {e}")
         return None
 
+
 # ---------------------------------------------------------------------------
-# 개별 문항 풀이 (캐시 + 타임아웃)
+# 문항 풀이 (캐시 + 시간 제한)
 # ---------------------------------------------------------------------------
 def solve_one(item: dict, api_url: str) -> str:
     prompt = ANSWER_RULES + "\n\nQuestion:\n" + item["question"]
     file_path = download_task_file(api_url, item["task_id"], item.get("file_name", ""))
     if file_path:
         prompt += f"\n\nAn attached file is available at: {file_path}"
-    
-    agent = build_agent()
-    result = agent.run(prompt)
-    return clean_answer(result)
+    return clean_answer(build_agent().run(prompt))
 
-def solve_with_timeout(item: dict, api_url: str) -> tuple[str, bool]:
+
+def solve_with_timeout(item: dict, api_url: str):
+    """(답변, 성공 여부, 비고)를 반환합니다."""
     task_id = item["task_id"]
+    short = task_id[:8]
     cached = load_cache().get(task_id)
     if cached:
-        print(f"[cache] {task_id} -> {cached}")
-        return cached, True
+        log(f"캐시 사용 {short} -> {cached}")
+        return cached, True, "cache"
 
     box: dict = {}
+
     def target():
         try:
             box["answer"] = solve_one(item, api_url)
         except Exception as e:  # noqa: BLE001
-            box["error"] = str(e)
+            box["error"] = f"{type(e).__name__}: {e}"[:300]
 
-    print(f"[start] {task_id}")
+    log(f"시작 {short}")
+    with _state_lock:
+        _active[task_id] = time.time()
     t = threading.Thread(target=target, daemon=True)
     t.start()
     t.join(QUESTION_TIMEOUT)
+    with _state_lock:
+        _active.pop(task_id, None)
 
     if t.is_alive():
-        print(f"[timeout] {task_id}")
-        return "unknown", False
+        log(f"시간 초과 {short} ({QUESTION_TIMEOUT}초)")
+        return "unknown", False, f"timeout {QUESTION_TIMEOUT}s"
     if "error" in box:
-        print(f"[error] {task_id}: {box['error']}")
-        return "unknown", False
+        log(f"오류 {short}: {box['error']}")
+        return "unknown", False, box["error"]
 
     save_cache_entry(task_id, box["answer"])
-    print(f"[done] {task_id} -> {box['answer']}")
-    return box["answer"], True
+    log(f"완료 {short} -> {box['answer']}")
+    return box["answer"], True, ""
+
 
 def iter_solve(questions: list[dict], api_url: str):
+    """끝난 문항은 (item, answer, ok, note)로, 대기 중에는 주기적으로 None을 내보냅니다."""
     valid = [q for q in questions if q.get("task_id") and q.get("question") is not None]
-    with ThreadPoolExecutor(max_workers=WORKERS) as pool:
-        futures = {pool.submit(solve_with_timeout, q, api_url): q for q in valid}
-        for fut in as_completed(futures):
-            answer, ok = fut.result()
-            yield futures[fut], answer, ok
+    pool = ThreadPoolExecutor(max_workers=WORKERS)
+    try:
+        pending = {pool.submit(solve_with_timeout, q, api_url): q for q in valid}
+        while pending:
+            done, _ = wait(list(pending), timeout=HEARTBEAT, return_when=FIRST_COMPLETED)
+            if not done:
+                yield None
+                continue
+            for fut in done:
+                item = pending.pop(fut)
+                answer, ok, note = fut.result()
+                yield item, answer, ok, note
+    finally:
+        pool.shutdown(wait=False)
+
 
 def make_df(rows: dict, order: list[str]) -> pd.DataFrame:
     data = [rows[tid] for tid in order if tid in rows]
-    return pd.DataFrame(data, columns=["Task ID", "Question", "Submitted Answer", "OK"])
+    return pd.DataFrame(data, columns=["Task ID", "Question", "Submitted Answer", "OK", "Note"])
+
+
+def status_text(label: str, done: int, total: int, start: float) -> str:
+    with _state_lock:
+        running = len(_active)
+        oldest = int(time.time() - min(_active.values())) if _active else 0
+    return (
+        f"{label}: {done}/{total} 완료 | 실행 중 {running}개 "
+        f"(가장 오래된 문항 {oldest}초) | 총 경과 {int(time.time() - start)}초\n"
+        f"--- 최근 기록 ---\n{recent_log()}"
+    )
+
+
+def run_batch(questions: list[dict], api_url: str, label: str):
+    """진행 상황을 화면에 내보내고, 마지막에 (rows, order)를 반환합니다."""
+    order = [q["task_id"] for q in questions if q.get("task_id")]
+    rows: dict = {}
+    start = time.time()
+    log(f"{label} 시작: 질문 {len(order)}개, WORKERS={WORKERS}, 모델={gemini_model_id()}")
+    yield status_text(label, 0, len(order), start), None
+
+    for event in iter_solve(questions, api_url):
+        if event is not None:
+            item, answer, ok, note = event
+            rows[item["task_id"]] = {
+                "Task ID": item["task_id"],
+                "Question": item["question"],
+                "Submitted Answer": answer,
+                "OK": "O" if ok else "X",
+                "Note": note,
+            }
+        yield status_text(label, len(rows), len(order), start), make_df(rows, order)
+    return rows, order
+
 
 # ---------------------------------------------------------------------------
-# Gradio 콜백 (실시간 진행 상황 제너레이터 반영)
+# Gradio 콜백
 # ---------------------------------------------------------------------------
 def run_and_submit_all(profile: gr.OAuthProfile | None):
     space_id = os.getenv("SPACE_ID")
@@ -235,27 +328,14 @@ def run_and_submit_all(profile: gr.OAuthProfile | None):
     api_url = DEFAULT_API_URL
     agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main"
 
-    yield "질문을 가져오는 중...", None
+    yield "질문을 가져오는 중... (최대 30초)", None
     try:
         questions = fetch_questions(api_url)
     except Exception as e:  # noqa: BLE001
         yield f"질문 가져오기 오류: {e}", None
         return
 
-    order = [q["task_id"] for q in questions if q.get("task_id")]
-    total = len(order)
-    rows: dict = {}
-    start = time.time()
-
-    for item, answer, ok in iter_solve(questions, api_url):
-        rows[item["task_id"]] = {
-            "Task ID": item["task_id"],
-            "Question": item["question"],
-            "Submitted Answer": answer,
-            "OK": "O" if ok else "X",
-        }
-        elapsed = int(time.time() - start)
-        yield f"진행 {len(rows)}/{total} 완료 (경과 {elapsed}초)", make_df(rows, order)
+    rows, order = yield from run_batch(questions, api_url, "본 실행")
 
     answers_payload = [
         {"task_id": tid, "submitted_answer": rows[tid]["Submitted Answer"]}
@@ -285,35 +365,74 @@ def run_and_submit_all(profile: gr.OAuthProfile | None):
         status = f"제출 실패: {detail}"
     except Exception as e:  # noqa: BLE001
         status = f"제출 중 오류: {e}"
-
     yield status, make_df(rows, order)
 
+
 def run_test_only(n_questions: int):
     api_url = DEFAULT_API_URL
-    yield "질문을 가져오는 중...", None
+    yield "질문을 가져오는 중... (최대 30초)", None
     try:
         questions = fetch_questions(api_url)[: int(n_questions)]
     except Exception as e:  # noqa: BLE001
         yield f"준비 중 오류: {e}", None
         return
+    rows, order = yield from run_batch(questions, api_url, "테스트")
+    yield f"테스트 완료 (제출 안 함)\n--- 최근 기록 ---\n{recent_log()}", make_df(rows, order)
 
-    order = [q["task_id"] for q in questions if q.get("task_id")]
-    rows: dict = {}
-    start = time.time()
 
-    for item, answer, ok in iter_solve(questions, api_url):
-        rows[item["task_id"]] = {
-            "Task ID": item["task_id"],
-            "Question": item["question"],
-            "Submitted Answer": answer,
-            "OK": "O" if ok else "X",
-        }
-        yield (
-            f"테스트 진행 {len(rows)}/{len(order)} (경과 {int(time.time() - start)}초)",
-            make_df(rows, order),
-        )
+def _call_with_timeout(fn, seconds: int):
+    box: dict = {}
+
+    def target():
+        try:
+            box["r"] = fn()
+        except Exception as e:  # noqa: BLE001
+            box["e"] = f"{type(e).__name__}: {e}"[:300]
+
+    th = threading.Thread(target=target, daemon=True)
+    th.start()
+    th.join(seconds)
+    if th.is_alive():
+        return False, f"{seconds}초 안에 응답 없음"
+    if "e" in box:
+        return False, box["e"]
+    return True, str(box.get("r"))[:150]
+
+
+def diagnose():
+    """채점 서버 / Gemini / 웹 검색 연결을 각각 점검합니다."""
+    lines = [
+        f"설정: 모델={gemini_model_id()}, GEMINI_API_KEY={'있음' if os.getenv('GEMINI_API_KEY') else '없음'}, "
+        f"WORKERS={WORKERS}, MAX_STEPS={MAX_STEPS}, QUESTION_TIMEOUT={QUESTION_TIMEOUT}"
+    ]
+    yield "\n".join(lines + ["", "1/3 채점 서버 확인 중..."])
+
+    ok, msg = _call_with_timeout(lambda: len(fetch_questions(DEFAULT_API_URL)), 35)
+    lines.append(f"1) 채점 서버 질문 목록: {'정상' if ok else '실패'} - {msg}")
+    yield "\n".join(lines + ["", "2/3 Gemini 호출 확인 중..."])
+
+    key = os.getenv("GEMINI_API_KEY")
+    if key:
+        def ping():
+            import litellm
+            r = litellm.completion(
+                model=f"gemini/{gemini_model_id()}",
+                api_key=key,
+                messages=[{"role": "user", "content": "Reply with the single word OK."}],
+                timeout=45,
+            )
+            return r.choices[0].message.content
+
+        ok, msg = _call_with_timeout(ping, 60)
+        lines.append(f"2) Gemini 호출: {'정상' if ok else '실패'} - {msg}")
+    else:
+        lines.append("2) Gemini 호출: GEMINI_API_KEY가 설정되지 않아 건너뜀")
+    yield "\n".join(lines + ["", "3/3 웹 검색 확인 중..."])
+
+    ok, msg = _call_with_timeout(lambda: DuckDuckGoSearchTool()("Hugging Face"), 40)
+    lines.append(f"3) 웹 검색(DuckDuckGo): {'정상' if ok else '실패'} - {msg}")
+    yield "\n".join(lines)
 
-    yield f"테스트 완료 ({int(time.time() - start)}초, 제출 안 함)", make_df(rows, order)
 
 def clear_cache():
     try:
@@ -322,28 +441,31 @@ def clear_cache():
     except FileNotFoundError:
         return "삭제할 캐시가 없습니다."
 
+
 # ---------------------------------------------------------------------------
-# Gradio UI 구성
+# UI
 # ---------------------------------------------------------------------------
 with gr.Blocks() as demo:
     gr.Markdown("# Basic Agent Evaluation Runner (Gemini 3.8 Flash)")
     gr.Markdown(
         """**Instructions:**
-1. 이 Space를 복제(Duplicate)한 뒤 코드를 수정해 에이전트 로직을 정의합니다.
-2. 아래 버튼으로 Hugging Face에 로그인합니다. (HF 사용자명으로 제출됩니다)
-3. 먼저 '테스트 실행'으로 소수 문항을 확인하고, 문제없으면 'Run Evaluation & Submit All Answers'를 누릅니다."""
+1. 먼저 '연결 진단'으로 채점 서버, Gemini, 웹 검색이 정상인지 확인합니다.
+2. 로그인 후 '테스트 실행'으로 소수 문항을 확인합니다. (제출 안 함)
+3. 문제없으면 'Run Evaluation & Submit All Answers'를 누릅니다."""
     )
     gr.LoginButton()
-    
+
     with gr.Row():
+        diag_button = gr.Button("연결 진단")
         test_n = gr.Number(value=2, label="테스트할 문항 수 (제출 안 함)", precision=0)
         test_button = gr.Button("테스트 실행 (제출 안 함)")
         cache_button = gr.Button("캐시 초기화")
-        
+
     run_button = gr.Button("Run Evaluation & Submit All Answers")
-    status_output = gr.Textbox(label="Run Status / Submission Result", lines=6, interactive=False)
+    status_output = gr.Textbox(label="Run Status / Submission Result", lines=14, interactive=False)
     results_table = gr.DataFrame(label="Questions and Agent Answers", wrap=True)
 
+    diag_button.click(fn=diagnose, outputs=[status_output])
     test_button.click(fn=run_test_only, inputs=[test_n], outputs=[status_output, results_table])
     run_button.click(fn=run_and_submit_all, outputs=[status_output, results_table])
     cache_button.click(fn=clear_cache, outputs=[status_output])