frontend/local_server.py

from __future__ import annotations

import asyncio
import base64
import datetime as _dt
import json
import os
import re
import shutil
import tempfile
import threading
import time
import traceback
import zipfile
from pathlib import Path
from typing import Any
from urllib.parse import unquote
from uuid import uuid4

from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from starlette.background import BackgroundTask

from agent_base.react_agent import MultiTurnReactAgent, default_llm_config
from agent_base.utils import (
    MissingRequiredEnvError,
    PROJECT_ROOT,
    append_saved_image_paths_to_prompt,
    image_input_content_parts,
    load_default_dotenvs,
    require_required_env,
    safe_jsonable,
    stage_image_bytes_for_input,
)


STATIC_DIR = Path(__file__).resolve().parent / "static"
MAX_UPLOAD_IMAGES = 12
MAX_IMAGE_BYTES = 12 * 1024 * 1024
MAX_WORKSPACE_DOWNLOAD_BYTES = 100 * 1024 * 1024
MAX_WORKSPACE_DOWNLOAD_FILES = 5000
FRONTEND_MANAGED_RUNS_DIR: str | None = None
FRONTEND_CLEANUP_RETENTION_SECONDS = 6 * 60 * 60
FRONTEND_CLEANUP_MAX_RUNS = 40
FRONTEND_CLEANUP_INTERVAL_SECONDS = 15 * 60
FRONTEND_COLLECTION_ENABLED = True
FRONTEND_COLLECTION_DATASET_REPO = "InternScience/ResearchHarness-Data"
FRONTEND_COLLECTION_BATCH_SIZE = 5
FRONTEND_COLLECTION_MAX_BUNDLE_BYTES = 20 * 1024 * 1024
_CLEANUP_THREAD_STARTED = False
_ACTIVE_MANAGED_RUNS: set[str] = set()
_ACTIVE_MANAGED_RUNS_LOCK = threading.Lock()
_DOWNLOAD_WORKSPACES: dict[str, str] = {}
_DOWNLOAD_WORKSPACES_LOCK = threading.Lock()
_COLLECTION_LOCK = threading.Lock()
_COLLECTION_CONFIG_WARNED: set[str] = set()

app = FastAPI(title="ResearchHarness Space UI")
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="frontend-static")


def configure_frontend(
    *,
    managed_runs_dir: str | None = None,
    cleanup_retention_seconds: int | None = None,
    cleanup_max_runs: int | None = None,
    cleanup_interval_seconds: int | None = None,
    collection_enabled: bool | None = None,
    collection_dataset_repo: str | None = None,
    collection_batch_size: int | None = None,
    collection_max_bundle_bytes: int | None = None,
) -> None:
    global FRONTEND_MANAGED_RUNS_DIR
    global FRONTEND_CLEANUP_RETENTION_SECONDS, FRONTEND_CLEANUP_MAX_RUNS, FRONTEND_CLEANUP_INTERVAL_SECONDS
    global FRONTEND_COLLECTION_ENABLED, FRONTEND_COLLECTION_DATASET_REPO
    global FRONTEND_COLLECTION_BATCH_SIZE, FRONTEND_COLLECTION_MAX_BUNDLE_BYTES
    if collection_enabled is not None:
        FRONTEND_COLLECTION_ENABLED = bool(collection_enabled)
    if collection_dataset_repo is not None:
        FRONTEND_COLLECTION_DATASET_REPO = str(collection_dataset_repo or "").strip()
    if collection_batch_size is not None:
        FRONTEND_COLLECTION_BATCH_SIZE = max(1, int(collection_batch_size))
    if collection_max_bundle_bytes is not None:
        FRONTEND_COLLECTION_MAX_BUNDLE_BYTES = max(1, int(collection_max_bundle_bytes))
    if not managed_runs_dir:
        raise ValueError("managed_runs_dir is required for the Space frontend")
    path = Path(managed_runs_dir).expanduser()
    if path.exists() and not path.is_dir():
        raise ValueError(f"managed-runs-dir is not a directory: {path}")
    path.mkdir(parents=True, exist_ok=True)
    FRONTEND_MANAGED_RUNS_DIR = str(path)
    if cleanup_retention_seconds is not None:
        FRONTEND_CLEANUP_RETENTION_SECONDS = max(60, int(cleanup_retention_seconds))
    if cleanup_max_runs is not None:
        FRONTEND_CLEANUP_MAX_RUNS = max(1, int(cleanup_max_runs))
    if cleanup_interval_seconds is not None:
        FRONTEND_CLEANUP_INTERVAL_SECONDS = max(60, int(cleanup_interval_seconds))
    _collection_root()
    cleanup_managed_runs_once()
    _start_managed_cleanup_thread()


class FrontendRunBridge:
    def __init__(self, *, loop: asyncio.AbstractEventLoop):
        self.loop = loop
        self.outbound: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
        self.cancelled = threading.Event()
        self.conversation_messages: list[dict[str, Any]] | None = None
        self.conversation_workspace_root: str = ""
        self.managed_run_root: str = ""
        self.managed_workspace_root: str = ""
        self.managed_trace_dir: str = ""
        self.download_token: str = ""
        self._pending_answers: dict[str, str] = {}
        self._pending_events: dict[str, threading.Event] = {}
        self._lock = threading.Lock()

    def send(self, payload: dict[str, Any]) -> None:
        self.loop.call_soon_threadsafe(self.outbound.put_nowait, safe_jsonable(payload))

    def trace_event(self, row: dict[str, Any]) -> None:
        self.send({"type": "trace", "row": row})

    def submit_answer(self, request_id: str, answer: str) -> bool:
        with self._lock:
            event = self._pending_events.get(request_id)
            if event is None:
                return False
            self._pending_answers[request_id] = str(answer)
            event.set()
            return True

    def ask_user(self, *, question: str, context: str = "") -> str:
        request_id = uuid4().hex
        event = threading.Event()
        with self._lock:
            self._pending_events[request_id] = event
        self.send(
            {
                "type": "ask_user",
                "request_id": request_id,
                "question": question,
                "context": context,
            }
        )
        while not event.wait(0.2):
            if self.cancelled.is_set():
                return "[AskUser] Cancelled before user answer was received."
        with self._lock:
            answer = self._pending_answers.pop(request_id, "")
            self._pending_events.pop(request_id, None)
        answer = str(answer).strip()
        if not answer:
            return "[AskUser] User answer was empty."
        return f"[AskUser] User answer:\n{answer}"


def _managed_runs_root() -> Path | None:
    if not FRONTEND_MANAGED_RUNS_DIR:
        return None
    return Path(FRONTEND_MANAGED_RUNS_DIR).expanduser().resolve()


def _new_managed_run_root() -> Path:
    root = _managed_runs_root()
    if root is None:
        raise ValueError("managed workspace mode is not configured")
    timestamp = _dt.datetime.now().strftime("%Y%m%d_%H%M%S")
    return root / f"run_{timestamp}_{uuid4().hex[:8]}"


def _mark_managed_run_active(run_root: Path) -> None:
    with _ACTIVE_MANAGED_RUNS_LOCK:
        _ACTIVE_MANAGED_RUNS.add(str(run_root.resolve()))


def _register_download_workspace(workspace_root: Path) -> str:
    token = uuid4().hex
    with _DOWNLOAD_WORKSPACES_LOCK:
        _DOWNLOAD_WORKSPACES[token] = str(workspace_root.resolve())
    return token


def _unregister_download_workspace(token: str) -> None:
    if not token:
        return
    with _DOWNLOAD_WORKSPACES_LOCK:
        _DOWNLOAD_WORKSPACES.pop(token, None)


def _download_workspace_for_token(token: str) -> Path:
    with _DOWNLOAD_WORKSPACES_LOCK:
        workspace_text = _DOWNLOAD_WORKSPACES.get(str(token or ""))
    if not workspace_text:
        raise HTTPException(status_code=404, detail="No downloadable workspace is available for this chat.")
    workspace_root = Path(workspace_text).resolve()
    if not workspace_root.is_dir():
        raise HTTPException(status_code=404, detail="The workspace is no longer available.")
    return workspace_root


def _resolve_workspace_file_path(workspace_root: Path, raw_path: str) -> Path:
    text = str(raw_path or "").strip()
    if text.startswith("file://"):
        text = text[7:]
    text = unquote(text)
    if not text:
        raise HTTPException(status_code=400, detail="workspace file path is required")
    candidate = Path(text)
    if not candidate.is_absolute():
        candidate = workspace_root / text
    resolved = candidate.resolve()
    try:
        resolved.relative_to(workspace_root.resolve())
    except ValueError as exc:
        raise HTTPException(status_code=403, detail="workspace file path is outside the workspace") from exc
    if not resolved.is_file():
        raise HTTPException(status_code=404, detail="workspace file does not exist")
    if resolved.suffix.lower() not in {".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".svg"}:
        raise HTTPException(status_code=415, detail="only workspace image files can be displayed inline")
    return resolved


def _release_managed_run(bridge: FrontendRunBridge) -> None:
    _unregister_download_workspace(bridge.download_token)
    if bridge.managed_run_root:
        with _ACTIVE_MANAGED_RUNS_LOCK:
            _ACTIVE_MANAGED_RUNS.discard(str(Path(bridge.managed_run_root).resolve()))
    bridge.managed_run_root = ""
    bridge.managed_workspace_root = ""
    bridge.managed_trace_dir = ""
    bridge.download_token = ""


def _create_managed_run(bridge: FrontendRunBridge) -> tuple[Path, str]:
    run_root = _new_managed_run_root()
    workspace_root = run_root / "agent_workspace"
    trace_dir = run_root / "agent_trace"
    workspace_root.mkdir(parents=True, exist_ok=True)
    trace_dir.mkdir(parents=True, exist_ok=True)
    bridge.managed_run_root = str(run_root)
    bridge.managed_workspace_root = str(workspace_root)
    bridge.managed_trace_dir = str(trace_dir)
    bridge.download_token = _register_download_workspace(workspace_root)
    _mark_managed_run_active(run_root)
    return workspace_root, str(trace_dir)


def cleanup_managed_runs_once() -> None:
    root = _managed_runs_root()
    if root is None or not root.exists():
        return
    now = time.time()
    with _ACTIVE_MANAGED_RUNS_LOCK:
        active = set(_ACTIVE_MANAGED_RUNS)
    runs = []
    for child in root.iterdir():
        if not child.is_dir() or not child.name.startswith("run_"):
            continue
        try:
            resolved = str(child.resolve())
            mtime = child.stat().st_mtime
        except OSError:
            continue
        runs.append((mtime, child, resolved))

    for mtime, child, resolved in runs:
        if resolved in active:
            continue
        if FRONTEND_CLEANUP_RETENTION_SECONDS and now - mtime > FRONTEND_CLEANUP_RETENTION_SECONDS:
            shutil.rmtree(child, ignore_errors=True)

    remaining = []
    with _ACTIVE_MANAGED_RUNS_LOCK:
        active = set(_ACTIVE_MANAGED_RUNS)
    for child in root.iterdir():
        if not child.is_dir() or not child.name.startswith("run_"):
            continue
        try:
            remaining.append((child.stat().st_mtime, child, str(child.resolve())))
        except OSError:
            continue
    remaining.sort(reverse=True, key=lambda item: item[0])
    for _, child, resolved in remaining[FRONTEND_CLEANUP_MAX_RUNS:]:
        if resolved not in active:
            shutil.rmtree(child, ignore_errors=True)


def _managed_cleanup_loop() -> None:
    while True:
        time.sleep(FRONTEND_CLEANUP_INTERVAL_SECONDS)
        cleanup_managed_runs_once()


def _start_managed_cleanup_thread() -> None:
    global _CLEANUP_THREAD_STARTED
    if _CLEANUP_THREAD_STARTED:
        return
    thread = threading.Thread(target=_managed_cleanup_loop, daemon=True)
    thread.start()
    _CLEANUP_THREAD_STARTED = True


def _collection_root() -> Path | None:
    root = _managed_runs_root()
    if root is None:
        return None
    collection_root = root / "_collection"
    (collection_root / "pending").mkdir(parents=True, exist_ok=True)
    return collection_root


def _collection_token() -> str:
    for name in ("HF_TOKEN", "HUGGINGFACE_HUB_TOKEN", "HUGGING_FACE_HUB_TOKEN"):
        value = os.getenv(name, "").strip()
        if value:
            return value
    return ""


def _warn_collection_once(key: str, message: str) -> None:
    if key in _COLLECTION_CONFIG_WARNED:
        return
    _COLLECTION_CONFIG_WARNED.add(key)
    print(f"[ResearchHarness Space collection] {message}", flush=True)


def _collection_ready() -> bool:
    if not FRONTEND_COLLECTION_ENABLED:
        return False
    if not FRONTEND_COLLECTION_DATASET_REPO:
        _warn_collection_once("missing_repo", "disabled because RH_COLLECTION_DATASET_REPO is empty.")
        return False
    if not _collection_token():
        _warn_collection_once("missing_token", "disabled because HF_TOKEN is not configured.")
        return False
    return _collection_root() is not None


class _CollectionBundleTooLarge(RuntimeError):
    pass


def _iter_collection_files(run_root: Path) -> list[tuple[Path, str]]:
    files: list[tuple[Path, str]] = []
    for dirname in ("agent_trace", "agent_workspace"):
        base = run_root / dirname
        if not base.exists() or not base.is_dir():
            continue
        for path in sorted(base.rglob("*")):
            if path.is_symlink() or not path.is_file():
                continue
            arcname = str(Path(dirname) / path.relative_to(base))
            files.append((path, arcname))
    return files


def _write_collection_bundle(run_root: Path, result: dict[str, Any]) -> Path | None:
    collection_root = _collection_root()
    if collection_root is None:
        return None
    pending_dir = collection_root / "pending"
    bundle_id = f"{run_root.name}_{_dt.datetime.utcnow().strftime('%Y%m%dT%H%M%SZ')}_{uuid4().hex[:8]}"
    zip_path = pending_dir / f"{bundle_id}.zip"
    meta_path = pending_dir / f"{bundle_id}.json"
    files = _iter_collection_files(run_root)
    skipped: list[dict[str, str]] = []
    manifest = {
        "bundle_id": bundle_id,
        "run_id": run_root.name,
        "created_at_utc": _dt.datetime.utcnow().isoformat(timespec="seconds") + "Z",
        "source": "ResearchHarness HuggingFace Space",
        "max_bundle_bytes": FRONTEND_COLLECTION_MAX_BUNDLE_BYTES,
        "file_count": len(files),
        "result_text": str(result.get("result_text", "")),
        "termination": str(result.get("termination", "")),
    }
    try:
        with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as archive:
            for path, arcname in files:
                try:
                    archive.write(path, arcname)
                except OSError as exc:
                    skipped.append({"path": str(path), "error": str(exc)})
                    continue
                if zip_path.stat().st_size > FRONTEND_COLLECTION_MAX_BUNDLE_BYTES:
                    raise _CollectionBundleTooLarge
            manifest["skipped_files"] = skipped
            archive.writestr("manifest.json", json.dumps(safe_jsonable(manifest), ensure_ascii=False, indent=2))
            if zip_path.stat().st_size > FRONTEND_COLLECTION_MAX_BUNDLE_BYTES:
                raise _CollectionBundleTooLarge
    except _CollectionBundleTooLarge:
        zip_path.unlink(missing_ok=True)
        meta_path.unlink(missing_ok=True)
        print(
            f"[ResearchHarness Space collection] dropped oversized bundle for {run_root.name}; "
            f"limit={FRONTEND_COLLECTION_MAX_BUNDLE_BYTES} bytes",
            flush=True,
        )
        return None
    except Exception:
        zip_path.unlink(missing_ok=True)
        meta_path.unlink(missing_ok=True)
        print("[ResearchHarness Space collection] failed to create bundle", flush=True)
        traceback.print_exc()
        return None

    meta = dict(manifest)
    meta["bundle_bytes"] = zip_path.stat().st_size
    meta_path.write_text(json.dumps(safe_jsonable(meta), ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"[ResearchHarness Space collection] queued bundle {zip_path.name}", flush=True)
    return zip_path


def _record_collection_upload_error(collection_root: Path, error: str) -> None:
    payload = {
        "created_at_utc": _dt.datetime.utcnow().isoformat(timespec="seconds") + "Z",
        "error": error,
    }
    (collection_root / "last_upload_error.json").write_text(
        json.dumps(payload, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )


def _create_dataset_pr_for_bundles(bundle_paths: list[Path]) -> str:
    from huggingface_hub import CommitOperationAdd, HfApi

    batch_id = f"batch_{_dt.datetime.utcnow().strftime('%Y%m%dT%H%M%SZ')}_{uuid4().hex[:8]}"
    operations = []
    for bundle_path in bundle_paths:
        operations.append(
            CommitOperationAdd(
                path_in_repo=f"batches/{batch_id}/{bundle_path.name}",
                path_or_fileobj=str(bundle_path),
            )
        )
        sidecar = bundle_path.with_suffix(".json")
        if sidecar.exists():
            operations.append(
                CommitOperationAdd(
                    path_in_repo=f"batches/{batch_id}/{sidecar.name}",
                    path_or_fileobj=str(sidecar),
                )
            )
    info = HfApi(token=_collection_token()).create_commit(
        repo_id=FRONTEND_COLLECTION_DATASET_REPO,
        repo_type="dataset",
        operations=operations,
        commit_message=f"Add ResearchHarness traces {batch_id}",
        commit_description="Automatically collected ResearchHarness Space trajectories.",
        create_pr=True,
    )
    return str(getattr(info, "pr_url", "") or getattr(info, "commit_url", "") or info)


def _flush_collection_batches() -> None:
    if not _collection_ready():
        return
    collection_root = _collection_root()
    if collection_root is None:
        return
    with _COLLECTION_LOCK:
        pending_dir = collection_root / "pending"
        while True:
            bundles = sorted(pending_dir.glob("*.zip"), key=lambda path: path.stat().st_mtime)
            if len(bundles) < FRONTEND_COLLECTION_BATCH_SIZE:
                return
            selected = bundles[:FRONTEND_COLLECTION_BATCH_SIZE]
            try:
                pr_url = _create_dataset_pr_for_bundles(selected)
            except Exception as exc:
                _record_collection_upload_error(collection_root, str(exc))
                print("[ResearchHarness Space collection] failed to create dataset PR", flush=True)
                traceback.print_exc()
                return
            for bundle_path in selected:
                bundle_path.unlink(missing_ok=True)
                bundle_path.with_suffix(".json").unlink(missing_ok=True)
            (collection_root / "last_upload_error.json").unlink(missing_ok=True)
            print(
                f"[ResearchHarness Space collection] created dataset PR for {len(selected)} bundles: {pr_url}",
                flush=True,
            )


def _collect_finished_managed_run(run_root_text: str, result: dict[str, Any]) -> None:
    if not _collection_ready() or not run_root_text:
        return
    run_root = Path(run_root_text)
    if not run_root.exists() or not run_root.is_dir():
        return
    bundle = _write_collection_bundle(run_root, result)
    if bundle is None:
        return
    threading.Thread(target=_flush_collection_batches, daemon=True).start()


def _workspace_download_files(workspace_root: Path) -> list[Path]:
    files: list[Path] = []
    total_bytes = 0
    for path in sorted(workspace_root.rglob("*")):
        if path.is_symlink() or not path.is_file():
            continue
        try:
            resolved = path.resolve()
            resolved.relative_to(workspace_root)
            size = resolved.stat().st_size
        except (OSError, ValueError):
            continue
        files.append(resolved)
        total_bytes += size
        if len(files) > MAX_WORKSPACE_DOWNLOAD_FILES:
            raise HTTPException(status_code=413, detail="Workspace has too many files to download as one zip.")
        if total_bytes > MAX_WORKSPACE_DOWNLOAD_BYTES:
            raise HTTPException(status_code=413, detail="Workspace is too large to download as one zip.")
    if not files:
        raise HTTPException(status_code=404, detail="The agent workspace has no downloadable files yet.")
    return files


def _create_workspace_zip(workspace_root: Path) -> Path:
    files = _workspace_download_files(workspace_root)
    handle = tempfile.NamedTemporaryFile(prefix="rh_workspace_", suffix=".zip", delete=False)
    zip_path = Path(handle.name)
    handle.close()
    try:
        with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as archive:
            for path in files:
                archive.write(path, path.relative_to(workspace_root).as_posix())
        if zip_path.stat().st_size > MAX_WORKSPACE_DOWNLOAD_BYTES:
            raise HTTPException(status_code=413, detail="Workspace zip is too large to download.")
    except Exception:
        zip_path.unlink(missing_ok=True)
        raise
    return zip_path


class FrontendInteractiveAgent(MultiTurnReactAgent):
    def __init__(self, *, bridge: FrontendRunBridge, **kwargs: Any):
        super().__init__(**kwargs)
        self.bridge = bridge

    def custom_call_tool(self, tool_name: str, tool_args: Any, **kwargs: Any):
        if tool_name != "AskUser":
            return super().custom_call_tool(tool_name, tool_args, **kwargs)
        tool = self.tool_map.get("AskUser")
        if tool is None:
            return "[AskUser] Tool is not available in this run."
        try:
            parsed = tool.parse_json_args(tool_args)
        except ValueError as exc:
            return f"[AskUser] {exc}"
        question = str(parsed.get("question", "")).strip()
        context = str(parsed.get("context", "") or "").strip()
        if not question:
            return "[AskUser] question must be a non-empty string."
        return self.bridge.ask_user(question=question, context=context)


def _safe_image_suffix(mime: str, filename: str = "") -> str:
    suffix = Path(filename).suffix.lower()
    if suffix in {".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp"}:
        return suffix
    mapping = {
        "image/png": ".png",
        "image/jpeg": ".jpg",
        "image/gif": ".gif",
        "image/webp": ".webp",
        "image/bmp": ".bmp",
    }
    return mapping.get(mime.lower(), ".png")


def decode_image_data_url(data_url: str, *, filename: str = "") -> tuple[str, bytes]:
    match = re.fullmatch(r"data:(image/[A-Za-z0-9.+-]+);base64,(.*)", str(data_url), flags=re.DOTALL)
    if not match:
        raise ValueError("image must be a data:image/...;base64,... URL")
    mime = match.group(1)
    try:
        raw = base64.b64decode(match.group(2), validate=True)
    except ValueError as exc:
        raise ValueError(f"invalid base64 image data: {exc}") from exc
    if not raw:
        raise ValueError("image upload is empty")
    if len(raw) > MAX_IMAGE_BYTES:
        raise ValueError(f"image upload exceeds {MAX_IMAGE_BYTES} bytes")
    return _safe_image_suffix(mime, filename), raw


def save_uploaded_images(workspace_root: Path, images: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], list[str]]:
    if len(images) > MAX_UPLOAD_IMAGES:
        raise ValueError(f"at most {MAX_UPLOAD_IMAGES} images are supported per run")
    if not images:
        return [], []
    timestamp = _dt.datetime.now().strftime("%Y%m%d_%H%M%S")
    content_parts: list[dict[str, Any]] = []
    saved_paths: list[str] = []
    for idx, item in enumerate(images, start=1):
        if not isinstance(item, dict):
            raise ValueError("each image item must be an object")
        data_url = str(item.get("data_url", "")).strip()
        filename = str(item.get("name", "") or f"image_{idx}")
        suffix, raw = decode_image_data_url(data_url, filename=filename)
        saved_path = stage_image_bytes_for_input(
            raw,
            workspace_root=workspace_root,
            filename=f"{timestamp}_{filename}",
            image_index=idx - 1,
            suffix=suffix,
        )
        saved_paths.append(saved_path)
        content_parts.extend(image_input_content_parts(data_url, saved_path))
    return content_parts, saved_paths


def _prompt_with_uploaded_image_paths(prompt: str, saved_paths: list[str]) -> str:
    return append_saved_image_paths_to_prompt(prompt, saved_paths)


def _run_agent_thread(
    *,
    bridge: FrontendRunBridge,
    prompt: str,
    workspace_root: Path,
    initial_content_parts: list[dict[str, Any]],
    trace_dir: str,
    prior_messages: list[dict[str, Any]] | None = None,
    managed_run_root: str = "",
    model_name: str = "",
) -> None:
    try:
        load_default_dotenvs()
        require_required_env("ResearchHarness frontend")
        agent = FrontendInteractiveAgent(
            bridge=bridge,
            llm=default_llm_config(model_name=model_name or None),
            trace_dir=trace_dir,
        )
        bridge.send(
            {
                "type": "run_started",
                "model": agent.model,
                "workspace_root": str(workspace_root),
                "trace_dir": trace_dir,
                "download_token": bridge.download_token,
            }
        )
        result = agent._run_session(
            prompt,
            workspace_root=str(workspace_root),
            event_callback=bridge.trace_event,
            initial_content_parts=initial_content_parts or None,
            prior_messages=prior_messages,
            interrupt_event=bridge.cancelled,
        )
        bridge.conversation_messages = result.get("messages", [])
        bridge.conversation_workspace_root = str(workspace_root)
        if managed_run_root:
            _collect_finished_managed_run(managed_run_root, result)
        bridge.send(
            {
                "type": "run_finished",
                "result_text": result.get("result_text", ""),
                "termination": result.get("termination", ""),
            }
        )
    except (MissingRequiredEnvError, ValueError) as exc:
        bridge.send({"type": "run_error", "error": str(exc)})
    except Exception as exc:
        bridge.send({"type": "run_error", "error": str(exc), "traceback": traceback.format_exc()})


@app.get("/")
def index() -> FileResponse:
    return FileResponse(STATIC_DIR / "index.html")


@app.get("/favicon.ico")
def favicon() -> FileResponse:
    return FileResponse(STATIC_DIR / "favicon.svg", media_type="image/svg+xml")


@app.get("/api/workspace.zip")
def download_workspace_zip(token: str) -> FileResponse:
    workspace_root = _download_workspace_for_token(token)
    zip_path = _create_workspace_zip(workspace_root)
    filename = f"{workspace_root.parent.name}_agent_workspace.zip"
    return FileResponse(
        zip_path,
        media_type="application/zip",
        filename=filename,
        background=BackgroundTask(lambda path: Path(path).unlink(missing_ok=True), str(zip_path)),
    )


@app.get("/api/workspace-file")
def workspace_file(token: str, path: str) -> FileResponse:
    workspace_root = _download_workspace_for_token(token)
    return FileResponse(_resolve_workspace_file_path(workspace_root, path))


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket) -> None:
    await websocket.accept()
    bridge = FrontendRunBridge(loop=asyncio.get_running_loop())
    run_thread: threading.Thread | None = None

    async def sender() -> None:
        while True:
            payload = await bridge.outbound.get()
            await websocket.send_json(payload)

    sender_task = asyncio.create_task(sender())
    try:
        await websocket.send_json({"type": "ready", "managed_workspace": True})
        while True:
            message = await websocket.receive_json()
            message_type = str(message.get("type", "")).strip()
            if message_type == "start":
                if run_thread is not None and run_thread.is_alive():
                    bridge.send({"type": "run_error", "error": "A run is already active. Wait for it to finish before starting a new conversation."})
                    continue
                prompt = str(message.get("prompt", "")).strip()
                if not prompt:
                    bridge.send({"type": "run_error", "error": "Prompt is required."})
                    continue
                try:
                    continue_conversation = bool(message.get("continue_conversation"))
                    model_name = str(message.get("model_name", "") or "").strip()
                    prior_messages = None
                    if continue_conversation:
                        if not bridge.conversation_messages or not bridge.managed_workspace_root:
                            bridge.send({"type": "run_error", "error": "No active conversation is available on the server. Click New chat and start again."})
                            continue
                        workspace_root = Path(bridge.managed_workspace_root)
                        effective_trace_dir = bridge.managed_trace_dir
                        prior_messages = bridge.conversation_messages
                    else:
                        _release_managed_run(bridge)
                        workspace_root, effective_trace_dir = _create_managed_run(bridge)
                    image_parts, saved_paths = save_uploaded_images(
                        workspace_root,
                        message.get("images", []) if isinstance(message.get("images", []), list) else [],
                    )
                    run_prompt = _prompt_with_uploaded_image_paths(prompt, saved_paths)
                except ValueError as exc:
                    bridge.send({"type": "run_error", "error": str(exc)})
                    continue
                bridge.cancelled.clear()
                if not continue_conversation:
                    bridge.conversation_messages = None
                    bridge.conversation_workspace_root = str(workspace_root)
                    bridge.send({"type": "conversation_reset"})
                if saved_paths:
                    bridge.send({"type": "uploaded_images", "paths": saved_paths})
                run_thread = threading.Thread(
                    target=_run_agent_thread,
                    kwargs={
                        "bridge": bridge,
                        "prompt": run_prompt,
                        "workspace_root": workspace_root,
                        "initial_content_parts": image_parts,
                        "trace_dir": effective_trace_dir,
                        "prior_messages": prior_messages,
                        "managed_run_root": bridge.managed_run_root,
                        "model_name": model_name,
                    },
                    daemon=True,
                )
                run_thread.start()
            elif message_type == "ask_user_answer":
                ok = bridge.submit_answer(str(message.get("request_id", "")), str(message.get("answer", "")))
                if not ok:
                    bridge.send({"type": "run_error", "error": "No pending AskUser request matched that answer."})
            elif message_type == "interrupt":
                if run_thread is not None and run_thread.is_alive():
                    bridge.cancelled.set()
                    bridge.send({"type": "interrupt_requested"})
                else:
                    bridge.send({"type": "run_error", "error": "No active run is available to interrupt."})
            elif message_type == "new":
                if run_thread is not None and run_thread.is_alive():
                    bridge.send({"type": "run_error", "error": "The current run is still active. Start a new conversation after it finishes."})
                else:
                    _release_managed_run(bridge)
                    bridge.conversation_messages = None
                    bridge.conversation_workspace_root = ""
                    bridge.send({"type": "conversation_reset"})
            else:
                bridge.send({"type": "run_error", "error": f"Unknown websocket message type: {message_type}"})
    except WebSocketDisconnect:
        bridge.cancelled.set()
    finally:
        bridge.cancelled.set()
        _release_managed_run(bridge)
        sender_task.cancel()