Final_Assignment_Template

Sleeping

Add two-stage per-Task timeout with Forced Answer (Part B)
6a102b8 3 months ago
91 kB
"""
GAIA Solver Pro — Autonomous Multi-Tool Agent
Engine: DeepSeek API (deepseek-chat)
Designed for the GAIA Benchmark with CSV output (Asynchronous/Concurrent Execution)
"""

# Import operating system utilities (for file paths and environment variables)
import os

# Import regular expressions library for text manipulation and string cleaning
import re

# Import JSON parsing library for handling structured data
import json

# Import time module to measure task execution duration
import time

# Import random module for jittered retry backoff delays
import random

# Import shutil for safe archive extraction (streaming member writes)
import shutil

# Import queue module for thread-safe worker pooling in parallel execution
import queue

# Import subprocess for running ffmpeg/ffprobe (video frame extraction)
import subprocess

# Import threading module for the shared vision-model lock
import threading

# Import HTTP requests library to make API calls and download files
import requests

# Import pandas for data processing, CSV exporting, and Excel reading
import pandas as pd

# Import PIL Image for handling image files when working with visual models
from PIL import Image

# Import thread pool executor tools for concurrent multithreaded processing
from concurrent.futures import ThreadPoolExecutor

# Import core agent tools and framework wrappers from the smolagents framework
from smolagents import CodeAgent, OpenAIServerModel, DuckDuckGoSearchTool, VisitWebpageTool, tool

# Import dotenv to load environment variables from a local .env file
from dotenv import load_dotenv

# Load secret environment variables (like API keys) into os.environ
load_dotenv()

# ─── DeepSeek API Configuration ──────────────────────────────────────
# Set default endpoint URL where the GAIA benchmark questions & files are hosted
DEFAULT_API_URL = os.getenv("GAIA_API_URL", "https://agents-course-unit4-scoring.hf.space")

# Retrieve the DeepSeek API key from environment variables
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")

# Set base URL for the DeepSeek API (OpenAI-compatible endpoint)
DEEPSEEK_API_BASE = os.getenv("DEEPSEEK_API_BASE", "https://api.deepseek.com/v1")

# Specify the main LLM model name used for agent reasoning and execution
AGENT_MODEL = os.getenv("DEEPSEEK_AGENT_MODEL", "deepseek-chat")

# Specify the LLM model name used for formatting the final string answer
CLEANING_MODEL = os.getenv("DEEPSEEK_CLEANING_MODEL", "deepseek-chat")

# Number of parallel worker agents used by the pipeline
NUM_WORKERS = int(os.getenv("GAIA_NUM_WORKERS", "4"))

# Maximum tokens allowed for the answer-cleaning pass (raised so long
# comma-separated list Answers are not truncated)
CLEANING_MAX_TOKENS = int(os.getenv("CLEANING_MAX_TOKENS", "1024"))

# Two-stage per-Task timeout (Part B, decisions B1-B4): at the SOFT deadline
# the Agent is stopped and a best-effort Answer is synthesized from its partial
# Worklog (status "forced"); at the HARD deadline the Task is terminated and
# recorded as "timeout". Hard is enforced as a total-time ceiling: the synthesis
# call is clamped to the remaining hard-soft budget, so a Task ends no later
# than ~hard seconds after it starts (a hard < soft config yields a zero
# synthesis budget, i.e. a plain timeout).
TASK_SOFT_TIMEOUT_SEC = float(os.getenv("GAIA_TASK_SOFT_TIMEOUT", "300"))
TASK_HARD_TIMEOUT_SEC = float(os.getenv("GAIA_TASK_HARD_TIMEOUT", "360"))

# Bounds the forced-answer synthesis call so soft + synthesis stays under the
# hard cap (the worker also clamps it to the remaining hard-soft budget).
FORCE_ANSWER_TIMEOUT_SEC = float(os.getenv("GAIA_FORCE_ANSWER_TIMEOUT", "45"))

# Local vision model used by the 'inspect_image' tool for image attachments
VISION_MODEL_ID = os.getenv("VISION_MODEL_ID", "Qwen/Qwen2-VL-2B-Instruct")

# ─── Multimodal config (ADR-0002: local-only) ─────────────────────────
# Local compute device for the multimodal models (Qwen vision + faster-whisper).
# One of: auto | cpu | mps | cuda (default 'auto'). 'auto' probes torch for an
# accelerator — Apple MPS first, then CUDA — and falls back to 'cpu'. Note:
# faster-whisper clamps mps->cpu because CTranslate2 has no MPS backend (see
# _whisper_device_config).
LOCAL_DEVICE = os.getenv("LOCAL_DEVICE", "auto").strip().lower()

# Local audio model used by the 'transcribe_audio' tool (faster-whisper).
# Lightweight by default; use a larger model (e.g. 'small'/'medium') for
# better accuracy at the cost of speed and memory.
WHISPER_MODEL = os.getenv("WHISPER_MODEL", "tiny")

# Set USE_WHISPER_API=1 to prefer the hosted Whisper API (OpenAI or
# OpenRouter) over the local faster-whisper model. The API path is also used
# as a documented fallback when the local model fails and a key is configured.
USE_WHISPER_API = os.getenv("USE_WHISPER_API", "").strip().lower() in ("1", "true", "yes", "on")

# Number of evenly-spaced frames the 'extract_video_frames' tool samples from
# a video; each frame is fed to the local vision model via 'inspect_image'.
VIDEO_FRAMES_COUNT = int(os.getenv("VIDEO_FRAMES_COUNT", "4"))

# ─── Answer bundle & Run outputs (ticket 03) ────────────────────────────
# The committed artifact the Space submits from: a JSON file keyed by task_id,
# each entry holding {question, level, file_name, answer, worklog, timestamp}.
# Generated by local Runs and committed to the repo (per the spec's Answer
# bundle contract / ADR-0001). It is deliberately NOT gitignored.
ANSWER_BUNDLE_PATH = os.getenv(
    "ANSWER_BUNDLE_PATH",
    os.path.join(os.path.dirname(os.path.abspath(__file__)), "answer_bundle.json"),
)

# Local Run results CSV output path (configurable so tests and custom Runs can
# write elsewhere without touching the repo's gaia_results.csv)
RESULTS_CSV_PATH = os.getenv("GAIA_RESULTS_CSV", "gaia_results.csv")

# ─── Prompt for Final Answer Cleaning ──────────────────────────────────
# System prompt template that instructs LLM how to format outputs strictly for benchmark submission
FINAL_ANSWER_PROMPT = """You are an expert assistant in generating final answers with the highest possible precision.
Given a question and an initial answer, your task is to generate the clean final answer.
**Rules:**
- The final answer must be a number and/or text string OR as few words as possible OR a comma-separated list.
- If a number is requested, DO NOT use commas to separate thousands NOR units like USD, $, percentage, or % unless specified.
- If a string is requested, DO NOT use articles or abbreviations, write digits in plain text unless specified.
- If a comma-separated list is requested, apply the previous rules depending on whether it is a number or a string.
- If the final answer is a number, use a number, not a word.
- If the final answer is a string, start with a capital letter.
- If the final answer is a comma-separated list of numbers, put a space after each comma.
- If the final answer is a comma-separated list of strings, put a space after each comma and start with a lowercase letter.
- DO NOT add content that is not in the initial answer.
**Examples:**
Question: What is the largest city in California?
Initial Answer: The largest city in California is Los Angeles.
Final Answer: Los Angeles
Question: How many 'r's are in strawberry?
Initial Answer: There are 3 'r's in strawberry.
Final Answer: 3
Question: {question}
Initial Answer: {answer}
Final Answer: """

# System prompt template for the forced-answer path (Part B): when a Task hits
# the soft deadline, a best-effort final Answer is synthesized from the Agent's
# partial reasoning trace so something real is still submitted (status "forced").
FORCED_ANSWER_PROMPT = """You are an expert assistant that must produce a best-effort final answer from an INCOMPLETE reasoning trace, because the agent hit its time limit.
**Rules:**
- Use ONLY evidence present in the trace; do not invent facts.
- If the trace already contains a candidate answer or concrete data, prefer it.
- If there is not enough evidence, give the most reasonable concise answer you can.
- Apply the same formatting rules as a normal benchmark answer: a number and/or short text string, or a comma-separated list; no units/percentages unless specified; no extra commentary.
QUESTION:
{question}
PARTIAL REASONING TRACE:
{trace}
FINAL ANSWER: """

# Tuples of prefixes commonly outputted by reasoners to strip during cleanup
ANSWER_PREFIXES = (
    "Final Answer:",
    "FINAL ANSWER:",
    "Answer:",
    "Respuesta Final:",
)

def _extract_youtube_video_id(youtube_url: str) -> str | None:
    """Extracts a YouTube video ID from watch, short, embed, or live URLs."""
    # Import URL parsing components inside function scope
    from urllib.parse import parse_qs, urlparse

    # Parse the input YouTube URL string into structured URL components
    parsed = urlparse(youtube_url)
    
    # Get domain host name in lowercase (e.g., youtube.com or youtu.be)
    host = (parsed.netloc or "").lower()
    
    # Get the URL path string (e.g., /watch or /shorts/xyz)
    path = parsed.path or ""

    # Check if the URL is a short link (youtu.be/ID)
    if "youtu.be" in host:
        # Extract ID from the path segment
        video_id = path.strip("/").split("/")[0]
        return video_id or None

    # Check if URL belongs to standard youtube.com domains
    if "youtube.com" in host:
        # Try extracting query parameter 'v' from ?v=VIDEO_ID
        query_id = parse_qs(parsed.query).get("v", [None])[0]
        if query_id:
            return query_id
        
        # Check path-based formats like /shorts/, /embed/, /live/
        for prefix in ("/shorts/", "/embed/", "/live/"):
            if path.startswith(prefix):
                video_id = path[len(prefix):].split("/")[0]
                return video_id or None

    # Fallback: regex search for standard 11-character video IDs
    match = re.search(r"(?:v=|youtu\.be/)([a-zA-Z0-9_-]{11})", youtube_url)
    # Return matched ID if found, otherwise None
    return match.group(1) if match else None

def _get_transcript_text(transcript_data) -> str:
    """Normalizes different versions of youtube_transcript_api into a single string."""
    # Convert transcript object to raw list of dictionaries if method exists
    if hasattr(transcript_data, "to_raw_data"):
        segments = transcript_data.to_raw_data()
    else:
        segments = transcript_data

    # List to hold individual text lines
    parts = []
    # Loop over every subtitle segment retrieved
    for segment in segments:
        # Extract text property or dictionary key depending on API version return type
        if hasattr(segment, "text"):
            text = segment.text
        elif isinstance(segment, dict):
            text = segment.get("text", "")
        else:
            text = str(segment)
        
        # Trim leading and trailing whitespace
        text = str(text).strip()
        # If segment contains text, append it to collector list
        if text:
            parts.append(text)
            
    # Combine all segments into a single space-delimited text paragraph
    return " ".join(parts).strip()


# Define local folder path where all attachment files are stored (relative to this script)
LOCAL_FILES_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "files")

# Folder where the 'extract_video_frames' tool writes sampled frames during a
# Run (kept under files/ so it stays out of version control)
VIDEO_FRAMES_DIR = os.getenv(
    "VIDEO_FRAMES_DIR",
    os.path.join(LOCAL_FILES_DIR, "_video_frames"),
)

def _get_local_task_file(task_id: str, file_name: str) -> str:
    """Locates a pre-downloaded file in the local files directory.
    
    Checks both original file_name and task_id-based filenames.
    """
    if not os.path.exists(LOCAL_FILES_DIR):
        return ""

    # Strategy 1: Check by exact file_name if present
    if file_name and str(file_name).strip():
        safe_name = os.path.basename(file_name)
        p1 = os.path.join(LOCAL_FILES_DIR, safe_name)
        if os.path.exists(p1):
            return p1

    # Strategy 2: Check by task_id + extension from file_name (e.g., 2b3ef98c-....mp3)
    if task_id:
        if file_name and "." in file_name:
            ext = file_name.lower().split(".")[-1]
            p2 = os.path.join(LOCAL_FILES_DIR, f"{task_id}.{ext}")
            if os.path.exists(p2):
                return p2

        # Strategy 3: Check by task_id alone
        p3 = os.path.join(LOCAL_FILES_DIR, task_id)
        if os.path.exists(p3):
            return p3

        # Strategy 4: Search for any file in the folder starting with task_id
        for existing_file in os.listdir(LOCAL_FILES_DIR):
            if existing_file.startswith(task_id):
                return os.path.join(LOCAL_FILES_DIR, existing_file)

    return ""


def _normalize_answer_for_submission(answer: str | None) -> str:
    """Ensures only the clean answer is processed by removing thinking tags."""
    # Handle None value safely and strip whitespace from answer string
    text = "" if answer is None else str(answer).strip()
    
    # Strip <think>...</think> reasoning blocks from DeepSeek R1 models using regex
    text = re.sub(r'<think>.*?</think>', '', text, flags=re.DOTALL).strip()
    
    # Clean orphaned closing tags if leftover
    text = re.sub(r'</think>', '', text).strip()
    
    # Clean orphaned opening tags if leftover
    text = re.sub(r'<think>', '', text).strip()

    # Iterate over standard answer prefixes and remove them if present
    for prefix in ANSWER_PREFIXES:
        if text.startswith(prefix):
            text = text[len(prefix):].strip()
            
    # Strip residual code ticks and bold asterisks surrounding answer
    text = text.strip("`*")
    
    # Return normalized clean text string
    return text


# Sentinel placeholders used to interpolate values into FINAL_ANSWER_PROMPT.
# Swapping the template's {question}/{answer} markers for these control-char
# sentinels first means a Question or Answer that itself contains the literal
# substrings "{question}"/"{answer}" cannot be double-substituted.
_PROMPT_QUESTION_SENTINEL = "\x00GAIA_QUESTION\x00"
_PROMPT_ANSWER_SENTINEL = "\x00GAIA_ANSWER\x00"

def _build_cleaning_prompt(question: str | None, answer: str | None) -> str:
    """Builds the final-answer cleaning prompt.
    Interpolates the Question and raw Answer via string replacement (never
    ``str.format``), so Answers containing ``{`` or ``}`` (e.g. JSON or sets)
    do not crash the cleaning pass.
    """
    prompt = (
        FINAL_ANSWER_PROMPT
        .replace("{question}", _PROMPT_QUESTION_SENTINEL)
        .replace("{answer}", _PROMPT_ANSWER_SENTINEL)
        .replace(_PROMPT_QUESTION_SENTINEL, "" if question is None else str(question))
        .replace(_PROMPT_ANSWER_SENTINEL, "" if answer is None else str(answer))
    )
    return prompt


# Sentinels for the forced-answer prompt (Part B); same rationale as the
# cleaning prompt: a partial trace containing "{"/"}" (code, JSON) must not
# crash the synthesis call or double-substitute.
_FORCE_QUESTION_SENTINEL = "\x00GAIA_FORCE_QUESTION\x00"
_FORCE_TRACE_SENTINEL = "\x00GAIA_FORCE_TRACE\x00"


def _build_forced_answer_prompt(question: str | None, partial_trace: str | None) -> str:
    """Builds the forced-answer synthesis prompt.
    Interpolates the Question and partial reasoning trace via string
    replacement (never ``str.format``) so traces containing ``{``/``}`` do not
    crash the synthesis call.
    """
    return (
        FORCED_ANSWER_PROMPT
        .replace("{question}", _FORCE_QUESTION_SENTINEL)
        .replace("{trace}", _FORCE_TRACE_SENTINEL)
        .replace(_FORCE_QUESTION_SENTINEL, "" if question is None else str(question))
        .replace(_FORCE_TRACE_SENTINEL, "" if partial_trace is None else str(partial_trace))
    )


def _chat_answer(model_id: str, prompt: str, *, description: str,
                 max_tokens: int | None = None) -> str:
    """Runs a zero-temperature DeepSeek chat completion and strips it to a bare answer.
    Shared by the answer-cleaning pass and the forced-answer synthesis (Part B):
    both need the same retried, `<think>`-stripped, prefix-stripped text. Raises
    on failure — callers decide whether to fall back (cleaning) or treat it as
    a timeout (forced synthesis).
    """
    import openai

    client = openai.OpenAI(base_url=DEEPSEEK_API_BASE, api_key=DEEPSEEK_API_KEY)

    def _request():
        return client.chat.completions.create(
            model=model_id,
            messages=[{"role": "user", "content": prompt}],
            max_tokens=max_tokens if max_tokens is not None else CLEANING_MAX_TOKENS,
            temperature=0.0,  # Deterministic zero-temperature for consistent answers
        )

    completion = _call_with_retry(_request, description=description)

    # Extract response text content from LLM choice completion
    text = completion.choices[0].message.content
    text = "" if text is None else str(text).strip()

    # Remove internal reasoning <think> blocks if present in output
    text = re.sub(r'<think>.*?</think>', '', text, flags=re.DOTALL).strip()

    # Strip prefixes, formatting marks, and code blocks from the output
    for prefix in [*ANSWER_PREFIXES, "**", "```"]:
        if text.startswith(prefix):
            text = text[len(prefix):].strip()

    # Return the string without trailing formatting chars
    return text.rstrip("*`")


# ─── Retry-with-backoff ───────────────────────────────────────────────
# Transient model/network failures (429 rate limits, connection drops,
# timeouts, 5xx) must not fail a Task and cost a point. smolagents retries
# rate-limit errors internally, but not network errors; the answer-cleaning
# call uses the raw OpenAI client and has no built-in retry at all. These
# helpers give both paths a uniform retry-with-backoff.
_RETRYABLE_STATUS_CODES = {408, 429, 500, 502, 503, 504}

# Message markers that indicate a transient failure when the exception type
# cannot be classified directly (e.g. wrapped/unknown errors).
_RETRYABLE_MESSAGE_MARKERS = (
    "429",
    "rate limit",
    "rate_limit",
    "too many requests",
    "500",
    "502",
    "503",
    "504",
    "internal server error",
    "service unavailable",
    "connection",
    "timeout",
    "timed out",
    "temporarily unavailable",
)

def _is_retryable_error(exc: BaseException) -> bool:
    """Returns True when ``exc`` is a transient model/network failure worth retrying."""
    try:
        from openai import APIConnectionError, APITimeoutError, RateLimitError, APIStatusError
        retryable_classes = (APIConnectionError, APITimeoutError, RateLimitError)
    except Exception:
        retryable_classes = ()
        APIStatusError = None

    # OpenAI SDK transient error classes
    if retryable_classes and isinstance(exc, retryable_classes):
        return True
    if APIStatusError is not None and isinstance(exc, APIStatusError):
        status = getattr(exc, "status_code", None)
        if status is not None:
            return status in _RETRYABLE_STATUS_CODES

    # requests-level errors: retry connection drops and timeouts, and HTTP
    # 5xx/429 surfaced by raise_for_status(), but not permanent 4xx client errors
    if isinstance(exc, requests.exceptions.HTTPError):
        status = getattr(exc.response, "status_code", None)
        if status is not None:
            return status in _RETRYABLE_STATUS_CODES
        return False
    if isinstance(exc, (requests.exceptions.ConnectionError, requests.exceptions.Timeout)):
        return True

    # Fallback heuristic: recognizable markers in the message text
    message = str(exc).lower()
    return any(marker in message for marker in _RETRYABLE_MESSAGE_MARKERS)

def _call_with_retry(fn, *, max_attempts: int = 4, base_delay: float = 1.0,
                     max_delay: float = 30.0, jitter: float = 0.5,
                     description: str = "call"):
    """Calls ``fn``, retrying transient errors with exponential backoff + jitter.
    Args:
        fn: The callable to invoke (must be re-invocable).
        max_attempts: Total attempts before giving up (default 4).
        base_delay: Base backoff seconds, doubled after each failure.
        max_delay: Ceiling on the backoff delay.
        jitter: +/- fraction of the delay applied as random jitter.
        description: Human-readable label used in retry log messages.
    Returns:
        Whatever ``fn`` returns on success.
    Raises:
        The last exception raised by ``fn`` once retries are exhausted or the
        error is not transient.
    """
    attempt = 0
    while True:
        attempt += 1
        try:
            return fn()
        except Exception as exc:
            if attempt >= max_attempts or not _is_retryable_error(exc):
                raise
            delay = min(max_delay, base_delay * (2 ** (attempt - 1)))
            delay *= 1.0 + jitter * (2.0 * random.random() - 1.0)
            print(f"⚠️ {description} failed (attempt {attempt}/{max_attempts}): {exc}; "
                  f"retrying in {max(0.0, delay):.1f}s")
            time.sleep(max(0.0, delay))


def _run_with_timeout(fn, timeout: float):
    """Runs ``fn`` on a daemon helper thread, bounding its runtime to ``timeout``.
    Python cannot forcibly kill a thread blocked in a call (e.g. a hung HTTP
    request), so the call runs on a short-lived **daemon** thread that the
    caller joins with a timeout. If the call is still alive when the timeout
    elapses, ``TimeoutError`` is raised and the caller moves on; the wedged
    helper thread cannot block the worker or the process at exit (daemon).
    Returns:
        Whatever ``fn`` returns on success.
    Raises:
        ``TimeoutError`` if ``fn`` does not finish within ``timeout`` seconds.
        Any exception raised by ``fn`` otherwise.
    """
    result_box = {}

    def _target():
        try:
            result_box["value"] = fn()
        except Exception as exc:
            result_box["error"] = exc

    helper = threading.Thread(target=_target, daemon=True)
    helper.start()
    helper.join(timeout)

    if helper.is_alive():
        raise TimeoutError(f"Task exceeded {timeout:.0f}s")
    if "error" in result_box:
        raise result_box["error"]
    return result_box["value"]

# ─── Worklog capture & Answer bundle (ticket 03) ────────────────────────
# A Worklog is the step-by-step trace of an Agent's Run — its reasoning steps
# (thought), tool calls, and observations. Each local Run serializes the full
# trace plus a per-tool summary with timing into a structured Worklog, and
# writes a committed Answer bundle keyed by task_id so the Space can submit
# without re-solving (ADR-0001 / spec's Answer bundle contract).

def _now_iso() -> str:
    """Returns the current UTC time in ISO-8601 (seconds) format."""
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())


def _step_duration_sec(step) -> float:
    """Returns a step's duration in seconds from its smolagents Timing, or 0.0."""
    timing = getattr(step, "timing", None)
    if timing is None:
        return 0.0
    duration = getattr(timing, "duration", None)
    if duration is None:
        return 0.0
    try:
        return round(float(duration), 3)
    except (TypeError, ValueError):
        return 0.0


def _iter_action_steps(steps):
    """Yields only the Agent Action steps (skipping Task/Planning steps)."""
    for step in steps:
        # Action steps carry a step_number; Task/Planning steps do not.
        if getattr(step, "step_number", None) is not None:
            yield step


def _coerce_thought(model_output) -> str:
    """Coerces a step's model_output (str or list-of-dicts) into a single string."""
    if model_output is None:
        return ""
    if isinstance(model_output, str):
        return model_output
    if isinstance(model_output, (list, tuple)):
        parts = []
        for item in model_output:
            if isinstance(item, dict):
                parts.append(str(item.get("text", item)))
            else:
                parts.append(str(item))
        return "\n".join(parts)
    return str(model_output)


def _coerce_observations(observations) -> str:
    """Coerces a step's observations (str or list) into a single string."""
    if observations is None:
        return ""
    if isinstance(observations, (list, tuple)):
        return "\n".join(str(o) for o in observations)
    return str(observations)


def _coerce_tool_arguments(arguments):
    """Coerces a ToolCall's arguments into a JSON-serializable value.
    smolagents ToolCall.arguments is usually already a dict, but older/newer
    paths can carry a JSON string; parse it so the Worklog is stable JSON.
    """
    if arguments is None:
        return ""
    if isinstance(arguments, str):
        try:
            return json.loads(arguments)
        except (ValueError, TypeError):
            return arguments
    return arguments


def _serialize_action_step(step) -> dict:
    """Serializes one Agent Action step into a JSON-safe Worklog trace entry."""
    tool_calls = []
    for call in getattr(step, "tool_calls", None) or []:
        tool_calls.append({
            "name": getattr(call, "name", "") or "unknown",
            "arguments": _coerce_tool_arguments(getattr(call, "arguments", None)),
        })

    entry = {
        "step_number": getattr(step, "step_number", None),
        "type": "final" if getattr(step, "is_final_answer", False) else "action",
        "thought": _coerce_thought(getattr(step, "model_output", None)),
        "tool_calls": tool_calls,
        "observations": _coerce_observations(getattr(step, "observations", None)),
        "code_action": getattr(step, "code_action", None) or "",
        "duration_sec": _step_duration_sec(step),
    }

    error = getattr(step, "error", None)
    if error is not None:
        entry["error"] = str(error)
    action_output = getattr(step, "action_output", None)
    if action_output is not None:
        entry["action_output"] = str(action_output)[:2000]
    return entry


def _serialize_tool_summary(steps) -> list[dict]:
    """Computes a per-tool summary (call count + timing) from Agent steps.
    Each step's duration is split evenly across the tools called in that step
    (smolagents only exposes per-step timing), so per-tool totals do not
    double-count time. Sorted by total time, then name, for a stable order.
    """
    per_tool: dict[str, dict] = {}
    for step in _iter_action_steps(steps):
        tool_calls = getattr(step, "tool_calls", None) or []
        if not tool_calls:
            continue
        share = _step_duration_sec(step) / len(tool_calls)
        for call in tool_calls:
            name = getattr(call, "name", "") or "unknown"
            bucket = per_tool.setdefault(name, {"calls": 0, "total_sec": 0.0})
            bucket["calls"] += 1
            bucket["total_sec"] += share

    summary = []
    for name in sorted(per_tool, key=lambda n: (-per_tool[n]["total_sec"], n)):
        bucket = per_tool[name]
        total = round(bucket["total_sec"], 3)
        calls = bucket["calls"]
        summary.append({
            "name": name,
            "calls": calls,
            "total_sec": total,
            "avg_sec": round(total / calls, 3) if calls else 0.0,
        })
    return summary


def _serialize_worklog(steps, *, status: str = "completed") -> dict:
    """Serializes the Agent's recorded steps into a structured Worklog.
    The Worklog holds the full trace (thought / tool call / observation per
    step), a per-tool summary with timing, and the total Run duration for the
    Task. ``status`` is "completed" on success, "forced" when the Answer was
    synthesized from a partial trace at the soft deadline, or "error" on
    failure.
    """
    action_steps = list(_iter_action_steps(steps))
    return {
        "steps": [_serialize_action_step(s) for s in action_steps],
        "tool_summary": _serialize_tool_summary(action_steps),
        "total_duration_sec": round(sum(_step_duration_sec(s) for s in action_steps), 3),
        "status": status,
    }


def _make_bundle_entry(task_id: str, question: str, level, file_name: str,
                       answer: str, worklog: dict, timestamp: str | None = None) -> dict:
    """Builds one Answer bundle entry, keyed by ``task_id``.
    The entry carries the Task's final Answer and its structured Worklog, plus
    the Question/Level/file_name so bundled submissions are self-describing and
    consistent with the results CSV (spec's Answer bundle contract).
    """
    return {
        "task_id": task_id,
        "question": question,
        "level": level if level is not None else "?",
        "file_name": file_name if file_name else "None",
        "answer": answer,
        "worklog": worklog,
        "timestamp": timestamp or _now_iso(),
    }


def _load_answer_bundle(path: str | None = None) -> dict:
    """Loads the committed Answer bundle from ``path`` (default ANSWER_BUNDLE_PATH).
    Returns ``{}`` when the bundle is absent or unreadable, so a Run degrades
    to a full solve rather than crashing.
    """
    path = path or ANSWER_BUNDLE_PATH
    try:
        with open(path, "r", encoding="utf-8") as f:
            data = json.load(f)
        if isinstance(data, dict):
            return data
    except (OSError, ValueError, TypeError):
        pass
    return {}


def _save_answer_bundle(bundle: dict, path: str | None = None) -> str:
    """Atomically writes the Answer bundle to ``path`` (default ANSWER_BUNDLE_PATH).
    Writes to a sibling temp file and renames over the target, so a crash
    mid-write cannot corrupt the committed bundle.
    """
    path = path or ANSWER_BUNDLE_PATH
    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
    tmp_path = f"{path}.tmp"
    with open(tmp_path, "w", encoding="utf-8") as f:
        json.dump(bundle, f, indent=2, ensure_ascii=False)
    os.replace(tmp_path, path)
    return path

# ─── Local device resolution (Part A: device auto-selection) ──────────
# A tiny lazy-import seam so the device resolver is unit-testable without
# torch installed (tests inject a fake module).
def _import_torch():
    """Lazily imports and returns the ``torch`` module (or raises ImportError)."""
    import torch
    return torch


_VALID_LOCAL_DEVICES = ("cpu", "mps", "cuda")


def _is_accelerator(local_device: str) -> bool:
    """True if ``local_device`` is a torch accelerator (mps or cuda)."""
    return local_device in ("mps", "cuda")


def _resolve_local_device() -> str:
    """Resolves the local compute device for the multimodal models.
    Reads the module-level ``LOCAL_DEVICE`` setting ('auto' | 'cpu' | 'mps' |
    'cuda', default 'auto'):
    - An explicit ``cpu``/``mps``/``cuda`` is returned as-is (no torch probe).
    - ``auto`` (or any unrecognized value) probes torch for an accelerator,
      preferring Apple MPS, then CUDA, and finally falls back to ``cpu``. If
      torch is unavailable, ``cpu`` is returned.
    """
    if LOCAL_DEVICE in _VALID_LOCAL_DEVICES:
        return LOCAL_DEVICE
    try:
        torch = _import_torch()
        if torch.backends.mps.is_available():
            return "mps"
        if torch.cuda.is_available():
            return "cuda"
    except Exception:
        pass
    return "cpu"


def _whisper_device_config(local_device: str):
    """Maps a resolved local device to faster-whisper's ``(device, compute_type)``.
    CTranslate2 (faster-whisper's backend) accepts only ``"cpu"`` or ``"cuda"``
    — there is no MPS backend — so ``mps`` is clamped to ``cpu``. ``compute_type``
    is ``"int8"`` on cpu (lightweight) and ``"float16"`` on cuda.
    """
    if local_device == "cuda":
        return "cuda", "float16"
    return "cpu", "int8"


def _vision_load_config(torch, local_device: str):
    """Returns ``(torch_dtype, device_map)`` for the Qwen vision model.
    ``torch_dtype`` is ``float16`` on accelerators (mps/cuda) to halve memory
    and ``float32`` on cpu. ``device_map='auto'`` is used only for cuda, because
    transformers' device_map has no MPS backend.
    """
    dtype = torch.float16 if _is_accelerator(local_device) else torch.float32
    device_map = "auto" if local_device == "cuda" else None
    return dtype, device_map


# ─── Agent Tools ──────────────────────────────────────────────────

# Shared faster-whisper model singleton (loaded once, reused across workers)
_whisper_model = None
_whisper_lock = threading.Lock()

def _transcribe_audio_local(audio_path: str) -> str:
    """Transcribes ``audio_path`` with the local faster-whisper model.
    The model (configurable via the WHISPER_MODEL setting, default 'tiny') is
    loaded once and reused across worker threads. Returns the transcript text,
    or an ``"Error: ..."`` string on failure so the Agent can react without
    crashing the Run (matching the other tools).
    """
    global _whisper_model

    try:
        # Import the lightweight local model only when the tool is first used
        from faster_whisper import WhisperModel

        # Load the shared model once (double-checked locking, like the vision model)
        if _whisper_model is None:
            with _whisper_lock:
                if _whisper_model is None:
                    # Route through the shared local-device resolver. faster-whisper
                    # has no MPS backend, so mps is clamped to cpu (int8); cuda uses
                    # float16. See _whisper_device_config.
                    device, compute_type = _whisper_device_config(_resolve_local_device())
                    _whisper_model = WhisperModel(WHISPER_MODEL, device=device, compute_type=compute_type)

        # Beam size 1 keeps latency low for the Agent loop
        segments, _info = _whisper_model.transcribe(audio_path, beam_size=1)
        parts = [
            segment.text.strip()
            for segment in segments
            if segment.text and segment.text.strip()
        ]
        transcript = " ".join(parts).strip()
        if not transcript:
            return f"Error: Local transcription returned no text for '{audio_path}'."
        return transcript
    except Exception as e:
        return f"Error: Local transcription failed (model '{WHISPER_MODEL}'): {e}"


def _transcribe_audio_api(audio_path: str) -> str:
    """Transcribes ``audio_path`` with the hosted Whisper API (documented fallback).
    Requires ``OPENAI_API_KEY`` (whisper-1) or ``OPENROUTER_API_KEY``
    (openai/whisper-large-v3-turbo). Returns the transcript text, or an
    ``"Error: ..."`` string on failure.
    """
    try:
        # Import openai library inside tool execution logic
        import openai

        # Check for OpenAI or OpenRouter API keys (for Whisper transcription only;
        # separate from the DeepSeek agent config since DeepSeek has no audio API)
        api_key = os.getenv("OPENAI_API_KEY") or os.getenv("OPENROUTER_API_KEY")

        # Select base API endpoint depending on which service key is set
        base_url = "https://api.openai.com/v1" if os.getenv("OPENAI_API_KEY") else "https://openrouter.ai/api/v1"

        # Choose Whisper model based on API provider
        model = "whisper-1" if os.getenv("OPENAI_API_KEY") else "openai/whisper-large-v3-turbo"

        # Return warning message if neither API key is present
        if not api_key:
            return "Error: Audio transcription fallback requires configuring OPENAI_API_KEY or OPENROUTER_API_KEY."

        # Instantiate OpenAI client with configured API host & key
        client = openai.OpenAI(base_url=base_url, api_key=api_key)

        # Open target audio file in binary reading mode
        with open(audio_path, "rb") as audio_file:
            # Request audio transcription from Whisper model endpoint
            transcription = client.audio.transcriptions.create(
                model=model,
                file=audio_file,
            )

        # Return transcript text content
        return transcription.text
    except Exception as e:
        # Return error string if transcription fails
        return f"Error transcribing audio via API: {e}"

@tool
def transcribe_audio(audio_path: str) -> str:
    """Transcribes an audio file (MP3, WAV, M4A, etc.).
    Uses the lightweight local faster-whisper model (configurable via the
    WHISPER_MODEL setting, default 'tiny'). If USE_WHISPER_API=1, or the local
    model fails and an API key is configured, it falls back to the hosted
    Whisper API (OpenAI or OpenRouter) as a documented fallback.
    Args:
        audio_path: The local path to the audio file (e.g., 'downloads/file.mp3').
    """
    if not USE_WHISPER_API:
        # Local-first per ADR-0002; fall back to the API only when the local
        # model fails AND a hosted key is configured
        local = _transcribe_audio_local(audio_path)
        if not local.startswith("Error:"):
            return local
        if os.getenv("OPENAI_API_KEY") or os.getenv("OPENROUTER_API_KEY"):
            return _transcribe_audio_api(audio_path)
        return local
    # Explicit opt-in to the hosted API
    return _transcribe_audio_api(audio_path)

@tool
def get_youtube_transcript(youtube_url: str) -> str:
    """Gets the official transcript/subtitles of a YouTube video.
    Args:
        youtube_url: The full URL of the YouTube video (e.g., 'https://www.youtube.com/watch?v=...').
    """
    try:
        # Import YouTube transcript API library
        from youtube_transcript_api import YouTubeTranscriptApi
        
        # Extract video ID from input URL string using helper function
        video_id = _extract_youtube_video_id(youtube_url)
        if not video_id:
            return f"Error: Could not extract video ID from URL: {youtube_url}"

        # Preference list of subtitle languages to query
        preferred_languages = ['en', 'es', 'fr', 'de']
        try:
            # Attempt direct fetching of preferred language subtitles
            transcript = YouTubeTranscriptApi.get_transcript(video_id, languages=preferred_languages)
        except Exception:
            # Fallback: list available transcripts and fetch best matching language
            transcript_list = YouTubeTranscriptApi.list_transcripts(video_id)
            transcript = transcript_list.find_transcript(preferred_languages).fetch()

        # Normalize fetched transcript object into single text string
        return _get_transcript_text(transcript)
    except Exception as e:
        # Return error description string if retrieval fails
        return f"Error obtaining YouTube transcript: {e}."


def _video_frame_timestamps(duration: float, num_frames: int) -> list[float]:
    """Computes N evenly-spaced frame timestamps within a video's duration.
    Timestamps are the centers of N equal slices of ``[0, duration]``, so the
    first and last frames are never the exact first/last instant (often a
    frozen title card or a black frame). Clamped to ``[0, duration]``.
    Args:
        duration: The video duration in seconds.
        num_frames: The number of evenly-spaced frames to sample (values below
            1 are clamped to 1).
    Returns:
        A list of timestamps in seconds, in ascending order.
    """
    n = max(1, int(num_frames))
    if duration <= 0:
        return [0.0]
    step = duration / n
    return [round((i + 0.5) * step, 3) for i in range(n)]


def _probe_video_duration(video_path: str) -> float:
    """Returns the duration (seconds) of a video file via ffprobe."""
    result = subprocess.run(
        ["ffprobe", "-v", "error", "-show_entries", "format=duration",
         "-of", "default=noprint_wrappers=1:nokey=1", video_path],
        capture_output=True, text=True, timeout=60,
    )
    if result.returncode != 0:
        raise RuntimeError(f"ffprobe failed: {result.stderr.strip()}")
    return float(result.stdout.strip())


def _resolve_frame_count(num_frames) -> int:
    """Coerces an LLM-supplied frame count to a valid positive int.
    The Agent calls the tool with whatever it inferred, so ``num_frames`` may
    be None, a float-string ("5.5"), or a word. Anything that is not a
    positive int falls back to the VIDEO_FRAMES_COUNT setting rather than
    failing the whole tool call.
    """
    try:
        n = int(num_frames)
        if n > 0:
            return n
    except (TypeError, ValueError):
        pass
    return VIDEO_FRAMES_COUNT


def _extract_frames_from_video(video_path: str, num_frames: int, job_id: str,
                               out_dir: str | None = None) -> list[str]:
    """Probes ``video_path`` and extracts N evenly-spaced JPEG frames.
    Frames are written as ``frame_{job_id}_NN.jpg`` into ``out_dir`` (defaults
    to the video-frames folder) and returned as absolute paths. Requires
    ffmpeg/ffprobe on PATH. Raises on failure so the caller can clean up any
    partial files.
    """
    out_dir = out_dir or VIDEO_FRAMES_DIR
    os.makedirs(out_dir, exist_ok=True)

    duration = _probe_video_duration(video_path)
    timestamps = _video_frame_timestamps(duration, num_frames)

    frame_paths = []
    for i, ts in enumerate(timestamps, start=1):
        frame_path = os.path.join(out_dir, f"frame_{job_id}_{i:02d}.jpg")
        subprocess.run(
            ["ffmpeg", "-y", "-ss", f"{ts:.3f}", "-i", video_path,
             "-frames:v", "1", "-q:v", "2", frame_path],
            capture_output=True, text=True, timeout=180, check=True,
        )
        frame_paths.append(frame_path)
    return frame_paths


def _cleanup_job_artifacts(job_id: str) -> None:
    """Removes any partial files for ``job_id`` from the video-frames folder.
    Covers the downloaded video (and its ``.part``/``.ytdl`` temp files, which
    share the ``video_{job_id}`` prefix) plus any frames already written before
    a failure, so a failed extraction never orphans files on disk.
    """
    if not os.path.isdir(VIDEO_FRAMES_DIR):
        return
    for name in os.listdir(VIDEO_FRAMES_DIR):
        if name.startswith(f"video_{job_id}") or name.startswith(f"frame_{job_id}"):
            try:
                os.remove(os.path.join(VIDEO_FRAMES_DIR, name))
            except OSError:
                pass


@tool
def extract_video_frames(video_url: str, num_frames: int | None = None) -> str:
    """Downloads a video and extracts N evenly-spaced frames to local files.
    Use this for the VISUAL content of a video (charts, diagrams, on-screen
    text, people/objects). For the SPOKEN content use 'get_youtube_transcript'
    instead. The sampled frames are written to the video-frames folder and can
    be read one-by-one with 'inspect_image'. Requires yt-dlp and
    ffmpeg/ffprobe on PATH.
    Args:
        video_url: The full URL of the video (e.g., 'https://www.youtube.com/watch?v=...').
        num_frames: Number of evenly-spaced frames to extract (defaults to the
            VIDEO_FRAMES_COUNT setting, usually 4).
    """
    job_id = None
    try:
        import yt_dlp

        n = _resolve_frame_count(num_frames)
        os.makedirs(VIDEO_FRAMES_DIR, exist_ok=True)
        job_id = str(int(time.time() * 1000))
        base_prefix = os.path.join(VIDEO_FRAMES_DIR, f"video_{job_id}")

        # Download the video. Bounding the height keeps the download small and
        # fast; noplaylist avoids pulling an entire playlist for one clip.
        ydl_opts = {
            "format": "bv*[height<=720]+ba/b[height<=720]/b",
            "outtmpl": base_prefix + ".%(ext)s",
            "noplaylist": True,
            "quiet": True,
            "no_warnings": True,
            "merge_output_format": "mp4",
        }
        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
            ydl.extract_info(video_url, download=True)

        # Locate the actual downloaded file (the final extension depends on the
        # stream/merge chosen by yt-dlp, so glob for our prefix)
        downloaded = [
            os.path.join(VIDEO_FRAMES_DIR, f)
            for f in os.listdir(VIDEO_FRAMES_DIR)
            if f.startswith(os.path.basename(base_prefix))
        ]
        if not downloaded:
            return f"Error: Download completed but no video file was found for {video_url}"
        video_path = downloaded[0]

        try:
            # Probe the duration, then sample N evenly-spaced frame timestamps
            frame_paths = _extract_frames_from_video(video_path, n, job_id)
        finally:
            # The source video is no longer needed once frames are extracted
            try:
                os.remove(video_path)
            except OSError:
                pass

        lines = [f"Extracted {len(frame_paths)} evenly-spaced frames from the video:"]
        lines += [f"  - {p}" for p in frame_paths]
        lines.append("Inspect each frame with 'inspect_image' to read its visual content.")
        return "\n".join(lines)
    except Exception as e:
        # Clean up this job's partial artifacts so failures never orphan files
        if job_id:
            _cleanup_job_artifacts(job_id)
        return f"Error extracting video frames: {e}"

@tool
def inspect_pdf(pdf_path: str) -> str:
    """Extracts all text from a PDF file in a structured format.
    Args:
        pdf_path: The local path to the PDF file (e.g., 'downloads/document.pdf').
    """
    try:
        # Import pdfplumber library for text extraction
        import pdfplumber
        
        # Container to accumulate text page by page
        texts = []
        
        # Open PDF file in pdfplumber handle context
        with pdfplumber.open(pdf_path) as pdf:
            # Iterate through each page of document with index
            for i, page in enumerate(pdf.pages):
                # Extract text content from current page
                content = page.extract_text()
                if content:
                    # Append formatted header with page content
                    texts.append(f"--- Page {i+1} ---\n{content}")
                    
        # Join extracted page texts and truncate to 60,000 chars to avoid prompt context blowup
        return "\n".join(texts)[:60000]
    except Exception as e:
        # Return error message if reading PDF fails
        return f"Error reading PDF: {e}"

@tool
def inspect_excel(file_path: str) -> str:
    """Reads an Excel file (.xlsx, .xls) and returns a summary of its sheets.
    Args:
        file_path: The local path to the Excel file (e.g., 'downloads/data.xlsx').
    """
    try:
        # Load Excel workbook structure into pandas ExcelFile reader
        xl = pd.ExcelFile(file_path)
        
        # Start summary list with sheet names found
        summary = [f"Sheets found: {xl.sheet_names}"]
        
        # Loop through each sheet name inside workbook
        for sheet in xl.sheet_names:
            # Read first 20 rows of current sheet into DataFrame
            df = pd.read_excel(xl, sheet_name=sheet).head(20)
            # Format sheet header and sample string table into summary
            summary.append(f"\n--- Sheet: {sheet} (Sample) ---\n{df.to_string()}")
            
        # Return joined summary string capped at 50,000 characters max
        return "\n".join(summary)[:50000]
    except Exception as e:
        # Return error message string if Excel read fails
        return f"Error reading Excel: {e}"

@tool
def read_file_as_text(file_path: str) -> str:
    """Reads a plain text file like CSV, JSON, TXT, etc.
    Args:
        file_path: The local path to the text file (e.g., 'downloads/report.csv').
    """
    try:
        # Open target text file with UTF-8 encoding, replacing invalid characters
        with open(file_path, "r", encoding="utf-8", errors="replace") as f:
            content = f.read()
            
        # Truncate content to 50,000 characters if file is too large
        if len(content) > 50000:
            content = content[:50000] + "\n...[CONTENT TRUNCATED]..."
            
        # Return file text contents
        return content
    except Exception as e:
        # Return error message string if text read fails
        return f"Error reading file: {e}"

# Shared vision model singletons (loaded once, reused across all worker threads)
_vision_model = None
_vision_processor = None
_vision_lock = threading.Lock()

@tool
def inspect_image(image_path: str, question: str = "") -> str:
    """Extracts the content of an image using a local vision model (Qwen2-VL-2B) plus OCR.
    Use this tool for image attachments (.png, .jpg, .jpeg, etc.). Pass the task
    question so the model focuses on the exact detail needed for the answer.
    Args:
        image_path: The local path to the image file (e.g., 'downloads/image.png').
        question: The task question to focus the extraction on (optional).
    """
    global _vision_model, _vision_processor

    try:
        # Import heavy libraries only when the tool is first invoked
        torch = _import_torch()
        from transformers import AutoProcessor

        # Resolve the correct auto class (newer transformers name, with older fallback)
        try:
            from transformers import AutoModelForImageTextToText as _AutoVL
        except ImportError:
            from transformers import AutoModelForVision2Seq as _AutoVL

        # Shared local-device resolver (LOCAL_DEVICE env: auto|cpu|mps|cuda)
        device = _resolve_local_device()

        # ── Load the shared model + processor once (double-checked locking) ──
        if _vision_model is None:
            with _vision_lock:
                if _vision_model is None:
                    dtype, device_map = _vision_load_config(torch, device)

                    _vision_processor = AutoProcessor.from_pretrained(VISION_MODEL_ID)
                    _vision_model = _AutoVL.from_pretrained(
                        VISION_MODEL_ID,
                        torch_dtype=dtype,
                        device_map=device_map,
                    )
                    _vision_model.eval()
                    if device == "cpu":
                        _vision_model.to("cpu")
                    elif device == "mps":
                        _vision_model.to("mps")

        # ── Open the image and normalize to RGB ──
        image = Image.open(image_path).convert("RGB")

        # ── Build a question-directed prompt for the VLM ──
        if question and str(question).strip():
            prompt_text = (
                "Look carefully at this image and answer the following question. "
                "Extract the exact text, numbers, names, or details shown in the image "
                "that are needed to answer. Be precise and do not guess.\n"
                f"Question: {question}"
            )
        else:
            prompt_text = (
                "Describe the contents of this image in detail. "
                "Include all visible text, numbers, labels, and names exactly as shown."
            )

        # ── Run inference (serialized across worker threads) ──
        with _vision_lock:
            try:
                # Preferred path: chat template (Qwen2-VL style)
                messages = [{
                    "role": "user",
                    "content": [
                        {"type": "image"},
                        {"type": "text", "text": prompt_text},
                    ],
                }]
                text = _vision_processor.apply_chat_template(
                    messages, tokenize=False, add_generation_prompt=True
                )
                inputs = _vision_processor(
                    text=[text], images=[image], return_tensors="pt"
                )
            except (AttributeError, NotImplementedError):
                # Fallback path: plain image + text inputs
                inputs = _vision_processor(
                    images=image, text=prompt_text, return_tensors="pt"
                )

            # Move inputs to the compute device (accelerators only)
            if _is_accelerator(device):
                inputs = {
                    k: v.to(device) for k, v in inputs.items()
                    if isinstance(v, torch.Tensor)
                }

            with torch.no_grad():
                outputs = _vision_model.generate(
                    **inputs,
                    max_new_tokens=512,
                    do_sample=False,
                )

            # Decode only the newly generated tokens
            input_len = inputs["input_ids"].shape[1]
            generated = _vision_processor.batch_decode(
                outputs[:, input_len:], skip_special_tokens=True
            )
            vlm_text = " ".join(generated).strip()

        # ── Best-effort OCR pass for literal text ──
        ocr_text = ""
        try:
            import pytesseract
            ocr_text = pytesseract.image_to_string(image).strip()
            if len(ocr_text) > 3000:
                ocr_text = ocr_text[:3000] + "\n...[OCR TRUNCATED]..."
        except Exception:
            ocr_text = ""  # OCR unavailable (e.g., Tesseract binary not installed)

        # ── Combine results and cap the total length ──
        result = f"VISION MODEL OUTPUT:\n{vlm_text}"
        if ocr_text:
            result += f"\n\nOCR TEXT:\n{ocr_text}"
        return result[:8000]
    except Exception as e:
        return f"Error analyzing image: {e}"

@tool
def inspect_docx(file_path: str) -> str:
    """Extracts all text (paragraphs and tables) from a Word .docx file.
    Args:
        file_path: The local path to the .docx file (e.g., 'downloads/document.docx').
    """
    try:
        # Import python-docx library inside tool execution logic
        import docx

        # Open the Word document
        document = docx.Document(file_path)

        # Container for extracted text parts
        parts = []

        # Collect non-empty paragraphs
        for para in document.paragraphs:
            text = para.text.strip()
            if text:
                parts.append(text)

        # Collect table contents row by row
        for t_idx, table in enumerate(document.tables):
            parts.append(f"\n--- Table {t_idx + 1} ---")
            for row in table.rows:
                cells = [cell.text.strip() for cell in row.cells]
                parts.append(" | ".join(cells))

        # Return joined text capped at 50,000 characters
        return "\n".join(parts)[:50000]
    except Exception as e:
        # Return error message if .docx reading fails
        return f"Error reading .docx: {e}"

@tool
def inspect_pptx(file_path: str) -> str:
    """Extracts all text from each slide of a PowerPoint .pptx file.
    Args:
        file_path: The local path to the .pptx file (e.g., 'downloads/presentation.pptx').
    """
    try:
        # Import python-pptx library inside tool execution logic
        from pptx import Presentation

        # Open the PowerPoint presentation
        prs = Presentation(file_path)

        # Container for extracted slide text
        parts = []

        # Loop over every slide with its index
        for s_idx, slide in enumerate(prs.slides):
            slide_texts = []
            # Iterate shapes (text boxes, tables, etc.) in the slide
            for shape in slide.shapes:
                if hasattr(shape, "text") and shape.text.strip():
                    slide_texts.append(shape.text.strip())
            if slide_texts:
                # Append formatted slide header with its text lines
                parts.append(f"--- Slide {s_idx + 1} ---\n" + "\n".join(slide_texts))

        # Return joined slide text capped at 50,000 characters
        return "\n\n".join(parts)[:50000]
    except Exception as e:
        # Return error message if .pptx reading fails
        return f"Error reading .pptx: {e}"

def _safe_extract_all(zf, extract_dir: str) -> tuple[list[str], list[str]]:
    """Extracts every member of an open ``zipfile.ZipFile`` without path traversal.
    Never uses ``extractall()``: each member's resolved destination is checked
    to stay inside ``extract_dir``, absolute paths and ``..`` traversal are
    rejected, and members are streamed out as plain files so symlink entries
    cannot be followed outside the extraction folder.
    Returns:
        A ``(extracted, skipped)`` pair: the names of members written, and the
        names of unsafe members that were rejected.
    """
    base = os.path.realpath(extract_dir)
    os.makedirs(base, exist_ok=True)

    extracted: list[str] = []
    skipped: list[str] = []

    for member in zf.infolist():
        # Normalize Windows separators and reject absolute/escaping paths
        member_path = member.filename.replace("\\", "/")
        if member_path.startswith("/") or ".." in member_path.split("/"):
            skipped.append(member.filename)
            continue

        target = os.path.realpath(os.path.join(base, member_path))
        # Guard: resolved target must stay inside the extraction folder
        if not (target == base or target.startswith(base + os.sep)):
            skipped.append(member.filename)
            continue

        if member.is_dir():
            os.makedirs(target, exist_ok=True)
            extracted.append(member.filename)
            continue

        # Stream the member out as a plain file (never follow symlinks)
        os.makedirs(os.path.dirname(target) or base, exist_ok=True)
        with zf.open(member, "r") as src, open(target, "wb") as dst:
            shutil.copyfileobj(src, dst)
        extracted.append(member.filename)

    return extracted, skipped

@tool
def extract_zip(zip_path: str, extract_dir: str = "") -> str:
    """Lists and extracts the contents of a .zip archive into a folder.
    Args:
        zip_path: The local path to the .zip file (e.g., 'downloads/files.zip').
        extract_dir: The folder to extract into (defaults to the files directory).
    """
    try:
        # Import zipfile library inside tool execution logic
        import zipfile

        # Default extraction folder next to the attachment files
        if not extract_dir:
            extract_dir = os.path.join(LOCAL_FILES_DIR, "_extracted")

        # Open the archive and extract entries safely (no extractall / zip-slip)
        with zipfile.ZipFile(zip_path, "r") as zf:
            extracted, skipped = _safe_extract_all(zf, extract_dir)

        # Build summary listing every extracted entry
        summary = [f"Extracted {len(extracted)} entries to '{extract_dir}':"]
        for name in extracted:
            summary.append(f"  - {name}")

        # Surface any rejected unsafe entries so the Agent knows they were skipped
        if skipped:
            summary.append(f"\n⚠️ Skipped {len(skipped)} unsafe entries (path traversal):")
            for name in skipped:
                summary.append(f"  - {name}")

        # Return listing capped at 20,000 characters
        return "\n".join(summary)[:20000]
    except Exception as e:
        # Return error message if .zip extraction fails
        return f"Error extracting .zip: {e}"

# ─── Main Agent Class ──────────────────────────────────────────────
class _RetryingOpenAIServerModel(OpenAIServerModel):
    """OpenAIServerModel that retries transient model/network errors with backoff.
    smolagents already retries rate-limit errors internally; this wrapper adds
    retries for connection/timeout/5xx failures so a transient network blip
    cannot fail a Task. The Agent's step memory is preserved across retries
    (unlike retrying the whole ``agent.run``), so no tool side-effects repeat.
    """

    def generate(self, messages, **kwargs):
        # Zero-arg super() only works directly in a method body — it is NOT
        # propagated into nested lambdas (no __class__ cell), so capture it
        # here and let the retry lambda call the bound base method.
        base = super()
        return _call_with_retry(
            lambda: base.generate(messages, **kwargs),
            description="model call",
        )

class GAIASolverAgent:
    def __init__(self, *, multimodal: bool = True):
        """Builds the ReAct solver Agent.
        Args:
            multimodal: When True (default, local harness), the full toolset is
                registered, including the local vision model (inspect_image),
                local audio transcription (transcribe_audio), and video frame
                sampling (extract_video_frames). When False (live-on-Space
                fallback), only text/web + document tools are registered, so no
                heavy local model is ever loaded — per ADR-0002 the Space
                fallback is text/web-only (no GPU, no hosted key).
        """
        # Validate that DEEPSEEK_API_KEY is available before proceeding
        if not DEEPSEEK_API_KEY:
            raise ValueError("⚠️ The environment variable 'DEEPSEEK_API_KEY' is not configured.")

        # Initialize retrying model pointing to configured API base and model
        self.main_model = _RetryingOpenAIServerModel(
            model_id=AGENT_MODEL,
            api_base=DEEPSEEK_API_BASE,
            api_key=DEEPSEEK_API_KEY,
        )

        # Text/web-only base + document tools, present in both modes
        tools = [
            DuckDuckGoSearchTool(), # Web search tool
            VisitWebpageTool(),     # Web scraping/browsing tool
            get_youtube_transcript, # YouTube transcript extraction tool (spoken content)
            inspect_pdf,            # PDF text extraction tool
            inspect_excel,          # Excel sheet inspector tool
            read_file_as_text,      # General text/CSV/JSON reader tool
            inspect_docx,           # Word .docx text extraction tool
            inspect_pptx,           # PowerPoint .pptx text extraction tool
            extract_zip,            # Zip archive extraction tool
        ]
        if multimodal:
            # Local-model tools (ADR-0002): vision, audio, and video frames.
            # Excluded from the Space fallback so Qwen2-VL / faster-whisper are
            # never loaded on the constrained free Space.
            tools += [
                transcribe_audio,       # Audio transcription tool (local faster-whisper)
                extract_video_frames,   # Video frame-sampling tool (visual content)
                inspect_image,          # Image vision tool (Qwen2-VL-2B + OCR)
            ]

        # Instantiate CodeAgent from smolagents framework
        self.agent = CodeAgent(
            tools=tools,
            model=self.main_model,
            # Explicit Python module imports authorized for code-execution agent actions
            additional_authorized_imports=[
                # ── Standard Python Libraries ──
                "sys", "os", "pathlib", "glob", "shutil", "tempfile", 
                "hashlib", "base64", "math", "statistics", "random", 
                "re", "string", "datetime", "time", "json", "csv", 
                "xml", "html", "urllib", "urllib.parse", "io", "copy", 
                "collections", "itertools", "functools", "operator", 
                "typing", "dataclasses", "unicodedata", "wave", "zipfile", 
                "tarfile", "sqlite3", "uuid", "ast", "inspect",

                # ── Data Science & Math ──
                "numpy", "pandas", "scipy", "scipy.stats", "scipy.optimize",
                "matplotlib", "matplotlib.pyplot", "seaborn", "sympy", 
                "sklearn", "stat",

                # ── File & Image Processing ──
                "PIL", "PIL.Image", "PIL.ImageDraw", "cv2", 
                "openpyxl", "xlrd", "pdfplumber", "PyPDF2", "pypdf", 
                "fitz", "docx", "pptx",

                # ── Web, Scraping & Networking ──
                "requests", "bs4", "lxml", "queue"
            ],
            # Set upper limit on agent reasoning steps per task execution run
            max_steps=25,
        )

    def _clean_answer(self, question: str, raw_answer: str) -> str:
        """Uses the DeepSeek model to structure the clean final answer synchronously per thread."""
        try:
            # Build the cleaning prompt via replacement interpolation (safe for
            # Answers/Questions containing braces) and raise the token budget so
            # long comma-separated list Answers are not truncated
            prompt = _build_cleaning_prompt(question, raw_answer)
            return _chat_answer(CLEANING_MODEL, prompt, description="answer cleaning")
        except Exception:
            # Fallback to standard normalization regex if API formatting call fails
            return _normalize_answer_for_submission(raw_answer)

    def __call__(self, task_id: str, question: str, file_name: str) -> str:
        """Solves a Task and returns just the final Answer (see ``solve``)."""
        return self.solve(task_id, question, file_name)[0]

    def solve(self, task_id: str, question: str, file_name: str,
              file_path: str | None = None) -> tuple[str, dict]:
        """Solves a Task and returns ``(final_answer, worklog)``.
        ``worklog`` is the structured serialization of this Agent's recorded
        ReAct steps (thought / tool call / observation per step) plus a per-tool
        summary with timing, captured fresh from the Agent's memory after the
        run. On failure it returns an ``Error: ...`` Answer and an empty Worklog
        so the Run continues; the Task stays out of the Answer bundle and is
        re-solved on the next Run.
        ``file_path`` is the local path to the Task's Attachment. When omitted,
        the pre-downloaded file is located via ``_get_local_task_file`` (local
        harness behavior); when provided (Space live fallback, where the
        Attachment was just downloaded from the server), it is used directly.
        """
        # Ensure the Worklog only reflects THIS Task. Agent instances are reused
        # across Tasks in the worker pool, and run() resets memory only once it
        # actually starts, so clear it before any work can fail (a failure before
        # run() would otherwise serialize the previous Task's stale trace).
        try:
            self.agent.memory.reset()
        except Exception:
            pass  # fresh agent: memory is already empty

        try:
            # Locate the Attachment: an explicitly-provided download path (Space
            # live fallback) wins; otherwise look for the pre-downloaded file in
            # the local files directory (local harness)
            if file_path is None:
                file_path = _get_local_task_file(task_id, file_name)

            # Construct agent task prompt
            prompt = f"TASK: {question}\n\n"

            prompt += "### FORMATTING RULES:\n"
            prompt += "- If you need to write or run Python code, you MUST wrap the code block strictly inside <code>\n[your python code]\n</code> tags.\n"
            prompt += "- Ensure there is a newline after the opening tag and before the closing tag.\n\n"

            
            if file_path:
                ext = file_path.lower().split('.')[-1]
                if ext in ('png', 'jpg', 'jpeg', 'gif', 'bmp', 'webp'):
                    prompt += f"🖼️ IMAGE saved at '{file_path}'. Use 'inspect_image' with the task question to extract its content.\n\n"
                elif ext in ('mp3', 'wav', 'm4a', 'ogg', 'flac'):
                    prompt += f"🎙️ AUDIO saved at '{file_path}'. Use 'transcribe_audio' (local faster-whisper).\n\n"
                elif ext in ('pdf',):
                    prompt += f"📄 PDF saved at '{file_path}'. Use 'inspect_pdf' or Python libraries like pdfplumber/fitz.\n\n"
                elif ext in ('xlsx', 'xls'):
                    prompt += f"📊 EXCEL saved at '{file_path}'. Use 'inspect_excel' or Python pandas.\n\n"
                elif ext in ('docx',):
                    prompt += f"📝 WORD saved at '{file_path}'. Use 'inspect_docx' to extract text and tables.\n\n"
                elif ext in ('pptx',):
                    prompt += f"📽️ POWERPOINT saved at '{file_path}'. Use 'inspect_pptx' to extract slide text.\n\n"
                elif ext in ('zip',):
                    prompt += f"🗜️ ZIP ARCHIVE saved at '{file_path}'. Use 'extract_zip' to list and extract its contents.\n\n"
                else:
                    prompt += f"📎 FILE saved at '{file_path}'. Use 'read_file_as_text' or open with Python.\n\n"
            elif file_name and str(file_name).strip():
                prompt += f"⚠️ Note: Attachment '{file_name}' was specified but not found locally in files directory.\n\n"

            if "youtube.com" in question.lower() or "youtu.be" in question.lower():
                prompt += (
                    "📹 YOUTUBE: Use 'get_youtube_transcript' for the SPOKEN content, "
                    "and 'extract_video_frames' to sample frames if the answer depends "
                    "on VISUAL content (charts, diagrams, on-screen text). Use whichever "
                    "the Question needs — or both.\n\n"
                )

            prompt += "CRITICAL INSTRUCTIONS:\n1. Answer concisely.\n2. Provide the exact data required.\n"

            # Run agent loop (text-only; image attachments are handled via the
            # 'inspect_image' tool). Transient model/network failures are retried
            # inside _RetryingOpenAIServerModel.generate, which preserves the
            # Agent's step memory (retrying the whole loop would repeat tool
            # side-effects)
            raw_answer = self.agent.run(prompt)
            raw_answer_str = str(raw_answer).strip()
            raw_answer_str = re.sub(r'<think>.*?</think>', '', raw_answer_str, flags=re.DOTALL).strip()

            final_answer = _normalize_answer_for_submission(
                self._clean_answer(question, raw_answer_str)
            )
            # Capture the Agent's recorded steps (thought / tool call /
            # observation) into the structured Worklog before anything else
            # can mutate this Agent's memory
            worklog = _serialize_worklog(self.agent.memory.steps, status="completed")
            return final_answer, worklog
        except Exception as e:
            # Keep the partial trace gathered before the failure so the Worklog
            # still shows how far the Agent got (status "error" + the reason)
            memory = getattr(self.agent, "memory", None)
            steps = memory.steps if memory is not None else []
            worklog = _serialize_worklog(steps, status="error")
            worklog["error"] = str(e)
            return f"Error: {e}", worklog

# ─── Forced-answer synthesis helpers (Part B) ─────────────────────────
# When a Task hits the soft deadline, its partial Worklog is snapshotted and a
# best-effort Answer is synthesized from it (decisions B1-B4) so something real
# is still submitted with status "forced".

def _format_partial_trace(worklog: dict) -> str:
    """Formats a (partial) Worklog into a compact readable trace for synthesis."""
    lines = []
    for step in worklog.get("steps", []):
        thought = (step.get("thought") or "").strip()
        if thought:
            lines.append(f"Thought: {thought}")
        for call in step.get("tool_calls") or []:
            name = call.get("name", "unknown")
            args = _coerce_tool_arguments(call.get("arguments", ""))
            lines.append(f"Tool: {name}({json.dumps(args, ensure_ascii=False)})")
        obs = (step.get("observations") or "").strip()
        if obs:
            lines.append(f"Observation: {obs[:2000]}")
    return "\n".join(lines)


def _synthesize_forced_answer(question: str, partial_trace: str) -> str:
    """Synthesizes a best-effort final Answer from a partial reasoning trace.
    Uses the cleaning-model path (DeepSeek, retried via ``_call_with_retry``)
    with the forced-answer prompt. The caller bounds this call (via
    ``_run_with_timeout``) so the total stays under the hard cap. Raises on
    model failure — the worker then records the Task as ``timeout`` (decision
    B2: a failed synthesis is NOT submitted).
    """
    prompt = _build_forced_answer_prompt(question, partial_trace)
    text = _chat_answer(CLEANING_MODEL, prompt, description="forced answer")
    if not text:
        raise ValueError("Forced-answer synthesis returned an empty answer")
    return _normalize_answer_for_submission(text)


def _forced_answer_budget() -> float:
    """Seconds available for forced-answer synthesis before the hard cap.
    Clamped to ``FORCE_ANSWER_TIMEOUT_SEC`` and to the remaining hard-soft
    budget; 0 (or negative) when the hard cap leaves no room, which means the
    Task must be recorded as a plain timeout instead.
    """
    remaining = TASK_HARD_TIMEOUT_SEC - TASK_SOFT_TIMEOUT_SEC
    return max(0.0, min(FORCE_ANSWER_TIMEOUT_SEC, remaining))


def _snapshot_partial_worklog(agent) -> dict:
    """Best-effort snapshot of an Agent's partial ReAct trace at the soft deadline.
    The wedged ``solve`` daemon helper thread may still be mutating
    ``agent.agent.memory.steps`` while this runs, so the read is deliberately
    racy: it serializes whatever steps are visible and, if the in-flight
    mutation makes serialization fail, falls back to an empty trace (the
    synthesis then produces a best-effort Answer from the Question alone).
    """
    try:
        memory = getattr(getattr(agent, "agent", None), "memory", None)
        steps = memory.steps if memory is not None else []
        return _serialize_worklog(steps, status="forced")
    except Exception:
        worklog = _serialize_worklog([], status="forced")
        worklog["error"] = "Partial Worklog unavailable at the soft deadline"
        return worklog


# ─── Asynchronous Threaded Console Script Execution ──────────────────────
def _make_result(idx, task, answer, elapsed, *, worklog=None, source="live",
                 status="completed", timestamp=None):
    """Builds a structured result record for a Task.
    ``source`` is "live" (solved by the Agent this Run) or "bundle" (submitted
    from the committed Answer bundle). ``status`` is "completed", "forced"
    (Answer synthesized from a partial trace at the soft deadline), "error",
    or "timeout". The record stays consistent with the Answer bundle entry for
    the same Task (same task_id / answer / worklog / timestamp).
    """
    return {
        "index": idx + 1,
        "task_id": task.get("task_id", "???"),
        "level": task.get("Level", "?"),
        "file_name": task.get("file_name", "") or "None",
        "question": task.get("question", ""),
        "answer": answer,
        "execution_time_sec": round(elapsed, 2),
        "source": source,
        "status": status,
        "timestamp": timestamp or _now_iso(),
        "worklog": worklog,
    }


def _make_bundled_result(idx, task, entry):
    """Builds a result record for a Task submitted from the Answer bundle.
    The bundle is authoritative for what was actually submitted (Answer,
    Worklog, timestamp, and the bundled Question/Level/file_name); the server's
    Task metadata is used only as a fallback for missing fields.
    """
    result = _make_result(
        idx, task, entry.get("answer", ""), 0.0,
        worklog=entry.get("worklog"), source="bundle", status="completed",
        timestamp=entry.get("timestamp", _now_iso()),
    )
    result["level"] = entry.get("level", result["level"])
    result["file_name"] = entry.get("file_name", result["file_name"])
    result["question"] = entry.get("question", result["question"])
    return result


def _timeout_result(idx, task, t0) -> dict:
    """Builds the hard-cap timeout result record (not submitted, decision B2/B3)."""
    return _make_result(
        idx, task,
        f"Error: Task timed out after {TASK_HARD_TIMEOUT_SEC:.0f}s",
        time.time() - t0, worklog=None, source="live", status="timeout",
    )


def _run_task(agent, idx, task) -> tuple[dict, bool]:
    """Solves one Task under the two-stage soft/hard timeout.
    Returns ``(result, replace_agent)`` where ``result`` is the structured
    result record (status ``completed``, ``forced``, ``timeout``, or ``error``)
    and ``replace_agent`` is True only when the Agent's ``solve`` was
    interrupted at the soft deadline — its daemon helper thread may still be
    wedged, so it must not be reused (the caller swaps in a fresh Agent).
    The two-stage budget (decisions B1-B4):
    - Within ``TASK_SOFT_TIMEOUT_SEC`` the normal result is used.
    - Otherwise the partial Worklog is snapshotted (best-effort — the wedged
      thread may still be mutating ``agent.memory.steps``) and a best-effort
      Answer is synthesized from it, bounded by ``_forced_answer_budget()`` so
      the total stays under the hard cap. Synthesis success → status ``forced``
      (submitted); failure or hard cap → status ``timeout`` (not submitted).
    """
    t0 = time.time()
    try:
        answer, worklog = _run_with_timeout(
            lambda: agent.solve(
                task.get("task_id", "???"),
                task.get("question", ""),
                task.get("file_name", ""),
            ),
            TASK_SOFT_TIMEOUT_SEC,
        )
        return (
            _make_result(idx, task, answer, time.time() - t0,
                         worklog=worklog, source="live",
                         status=worklog.get("status", "completed")),
            False,
        )
    except TimeoutError:
        # Soft deadline reached. The wedged Agent must always be discarded
        # (replace_agent=True), whether the synthesis succeeds (forced) or not
        # (timeout).
        budget = _forced_answer_budget()
        if budget <= 0:
            # No room left before the hard cap: record a plain timeout.
            return _timeout_result(idx, task, t0), True
        try:
            partial_worklog = _snapshot_partial_worklog(agent)
            answer = _run_with_timeout(
                lambda: _synthesize_forced_answer(
                    task.get("question", ""),
                    _format_partial_trace(partial_worklog),
                ),
                budget,
            )
            return (
                _make_result(idx, task, answer, time.time() - t0,
                             worklog=partial_worklog, source="live",
                             status="forced"),
                True,
            )
        except Exception:
            # Forced-answer synthesis failed or hit the hard cap: record a
            # plain timeout (not submitted, per decision B2/B3).
            return _timeout_result(idx, task, t0), True
    except Exception as e:
        # A throwing Task must not lose the worker: record an error Answer so
        # the Run still completes for every Task. The Agent is still reusable.
        return (
            _make_result(idx, task, f"Error: {e}", time.time() - t0,
                         worklog=None, source="live", status="error"),
            False,
        )


def run_pipeline_and_save_csv(force: bool = False):
    """Fetches questions, solves them, and writes the Answer bundle + results CSV.
    Args:
        force: When True, re-solve every Task from scratch, ignoring the
            committed Answer bundle. When False (default), Tasks already solved
            in the bundle are submitted from it without calling the Agent;
            only Tasks missing from the bundle are solved live. Every Task that
            solves successfully is recorded into the Answer bundle, so the next
            Run solves strictly fewer Tasks.
    """
    print("Base: Connecting to the GAIA server to fetch questions...")
    try:
        # Send HTTP GET request to download questions dataset JSON from server,
        # retrying transient network failures with backoff
        def _fetch_questions():
            response = requests.get(f"{DEFAULT_API_URL}/questions", timeout=15)
            response.raise_for_status()
            return response.json()

        questions_data = _call_with_retry(_fetch_questions, description="questions fetch")

        # If question list is empty, exit pipeline
        if not questions_data:
            print("⚠️ No questions were retrieved from the server.")
            return
    except Exception as e:
        # Print error message and cancel run if question retrieval fails
        print(f"❌ Error downloading the questions dataset: {e}")
        return

    total = len(questions_data)
    print(f"📋 {total} questions loaded.")

    # Load the committed Answer bundle (empty on the first Run) and classify
    # Tasks: bundled Tasks are submitted from the bundle; the rest are solved
    # live below (re-Runs only solve Tasks missing from the bundle, per the
    # spec / ADR-0001).
    bundle = _load_answer_bundle()
    bundled_ids = set(bundle)

    completed_results = []
    pending = []
    for i, task in enumerate(questions_data):
        task_id = task.get("task_id")
        if not task_id:
            continue
        if not force and task_id in bundled_ids:
            completed_results.append(_make_bundled_result(i, task, bundle[task_id]))
        else:
            pending.append((i, task))

    bundled_count = len(completed_results)
    solved_count = len(pending)
    if bundled_count:
        print(f"📦 {bundled_count}/{total} Tasks already in the Answer bundle — "
              f"submitting from the bundle (Agent not called).")
    if solved_count:
        print(f"🚀 Solving {solved_count}/{total} Tasks live (missing from the bundle) "
              f"with up to {NUM_WORKERS} parallel agents (DeepSeek)...")

    # ── Solve the pending Tasks with the Agent pool (lazy: no Agent is built
    # ── unless there is actually something to solve, so a fully-bundled Run
    # ── does not need a DeepSeek key) ─────────────────────────────────────
    if pending:
        try:
            agents = [GAIASolverAgent() for _ in range(min(NUM_WORKERS, len(pending)))]
        except Exception as e:
            print(f"❌ Error creating the agent pool: {e}")
            # Record an error row per pending Task so the Run still completes
            for i, task in pending:
                completed_results.append(
                    _make_result(i, task, f"Error: {e}", 0.0,
                                 worklog=None, source="live", status="error")
                )
        else:
            # Create thread-safe queue to manage available GAIASolverAgent
            # instances across threads
            agent_queue = queue.Queue()
            for a in agents:
                agent_queue.put(a)  # Put each worker agent into queue

            def _resolve(idx, task):
                """Worker thread task execution wrapper."""
                # Pop an available agent instance from queue (blocking if all are busy)
                my_agent = agent_queue.get()
                t0 = time.time()
                replaced = False
                try:
                    # Delegate the two-stage soft/hard timeout (and forced-answer
                    # synthesis at the soft deadline) to _run_task
                    result, must_replace = _run_task(my_agent, idx, task)
                    replaced = must_replace
                    if must_replace:
                        # The Agent may still be wedged in a hung call on a
                        # daemon helper thread. Never reuse a possibly-corrupted
                        # Agent: drop it and put a fresh one in its place so the
                        # pool stays at full strength
                        try:
                            agent_queue.put(GAIASolverAgent())
                        except Exception as e:
                            print(f"⚠️ Could not replace a timed-out Agent: {e}")
                    return result
                except Exception as e:
                    # A throwing Task must not lose the worker: record an error
                    # Answer for this Task so the Run still completes for every
                    # Task
                    return _make_result(idx, task, f"Error: {e}", time.time() - t0,
                                        worklog=None, source="live", status="error")
                finally:
                    # Always return the Agent to the pool (unless it was
                    # replaced above) so no worker is ever lost and the queue
                    # can never deadlock, even when a Task throws
                    if not replaced:
                        agent_queue.put(my_agent)

            print(f"\n🚀 Processing {len(pending)} Tasks asynchronously in parallel...")

            # Initialize ThreadPoolExecutor with maximum configured thread workers
            executor = ThreadPoolExecutor(max_workers=NUM_WORKERS)
            futures = {
                executor.submit(_resolve, i, task): (i, task)
                for i, task in pending
            }
            total_tasks = len(futures)

            completed = 0
            try:
                # Consume results in submission order. Each worker enforces its
                # own soft/hard timeout internally, so every future resolves in
                # bounded time
                for future, (i, task) in futures.items():
                    try:
                        res = future.result()
                    except Exception as e:
                        # Unexpected executor-level failure (worker already
                        # handles Task-level errors); still record an Answer
                        res = _make_result(i, task, f"Error: {e}", 0.0,
                                           worklog=None, source="live", status="error")
                    completed += 1
                    completed_results.append(res)  # Store result record

                    # Fold a successfully solved Task into the Answer bundle so
                    # the next Run submits it without re-solving. Completed and
                    # forced Tasks (a synthesized soft-deadline Answer that IS
                    # submitted — decision B3) are folded; error/timeout Tasks
                    # stay out and are retried next Run.
                    if res["status"] in ("completed", "forced"):
                        bundle[res["task_id"]] = _make_bundle_entry(
                            task_id=res["task_id"],
                            question=res["question"],
                            level=res["level"],
                            file_name=res["file_name"],
                            answer=res["answer"],
                            worklog=res["worklog"],
                            timestamp=res["timestamp"],
                        )

                    # Print live task progress log to stdout
                    print(f"✅ [{completed}/{total_tasks}] Task #{res['index']} "
                          f"({res['source']}) {res['status']} in "
                          f"{res['execution_time_sec']}s | Answer: {res['answer']}")
            finally:
                # Worker threads finish their current Task within
                # TASK_SOFT_TIMEOUT_SEC + FORCE_ANSWER_TIMEOUT_SEC, so
                # shutdown(wait=True) completes promptly; wedged daemon helper
                # threads cannot block the process at exit
                executor.shutdown(wait=True, cancel_futures=True)

    # Re-sort results list by original dataset question index ordering
    completed_results = sorted(completed_results, key=lambda x: x["index"])

    # Keep the bundle consistent with this Run and the served Question set:
    # (1) a force re-solve that now FAILS a previously-bundled Task must not
    # leave a stale entry behind (the next non-forced Run would resubmit the
    # old Answer while this Run's CSV shows an error) — drop it so it is
    # retried. A forced (soft-deadline) result is kept, not dropped: it was
    # already folded above and IS submitted and scored (decision B3), so it
    # must not be treated like an error/timeout; (2) Tasks the server no
    # longer serves are pruned so the bundle stays consistent with the results
    # CSV and never outgrows the served set.
    if force:
        for res in completed_results:
            if res["source"] == "live" and res["status"] not in ("completed", "forced"):
                bundle.pop(res["task_id"], None)
    served_ids = {t.get("task_id") for t in questions_data if t.get("task_id")}
    for task_id in list(bundle):
        if task_id not in served_ids:
            bundle.pop(task_id, None)

    # Write the committed Answer bundle (JSON keyed by task_id) so the Space
    # can submit without re-solving, then the results CSV
    _save_answer_bundle(bundle, ANSWER_BUNDLE_PATH)

    csv_columns = [
        "index", "task_id", "level", "file_name", "question", "answer",
        "execution_time_sec", "source", "status", "timestamp",
    ]
    df_output = pd.DataFrame(completed_results, columns=csv_columns)
    df_output.to_csv(RESULTS_CSV_PATH, index=False, encoding="utf-8")

    # Print completion summary message and target output paths
    print(f"\n{'═'*50}")
    print(f"📊 PROCESS COMPLETED SUCCESSFULLY")
    print(f"💾 Answer bundle ({len(bundle)} entries) saved to: {os.path.abspath(ANSWER_BUNDLE_PATH)}")
    print(f"💾 Results saved to: {os.path.abspath(RESULTS_CSV_PATH)}")
    print(f"{'═'*50}\n")

# Python script entry point check
if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(
        description="Run the local GAIA solver and write the committed Answer "
                    "bundle + results CSV."
    )
    parser.add_argument(
        "--force", action="store_true",
        help="Force a full re-solve, ignoring the committed Answer bundle.",
    )
    args = parser.parse_args()
    # Execute full pipeline when script is run directly from command line
    run_pipeline_and_save_csv(force=args.force)