app.py

from __future__ import annotations

import html
import io
import json
import os
import re
import time
import uuid
import zipfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Literal
from urllib.parse import quote_plus
from xml.etree import ElementTree

import gradio as gr
import httpx
import spaces
from huggingface_hub import InferenceClient
from pydantic import BaseModel, Field, ValidationError, model_validator
from pypdf import PdfReader


BASE_DIR = Path(__file__).resolve().parent
STATIC_DIR = BASE_DIR / "static"
DATA_DIR = Path("/data/research-decks") if Path("/data").exists() else BASE_DIR / "output"
DATA_DIR.mkdir(parents=True, exist_ok=True)
MODEL_ID = os.getenv("HF_MODEL_ID", "deepseek-ai/DeepSeek-V4-Flash")
HF_PROVIDER = os.getenv("HF_PROVIDER", "auto")
SEARCH_CACHE_TTL = 3600
SEARCH_CACHE: dict[tuple[str, str, int], tuple[float, list[dict[str, Any]]]] = {}
SEARCH_LAST_CALL: dict[str, float] = {}


@spaces.GPU(duration=1)
def zerogpu_runtime_probe() -> str:
    """Satisfy ZeroGPU Space validation; production inference uses HF Providers."""
    return "ready"


class ResearchRequest(BaseModel):
    topic: str = Field(min_length=2, max_length=180)
    purpose: str = Field(default="算法团队内部技术分享", max_length=500)
    audience: str = Field(default="熟悉机器学习基础的算法工程师", max_length=300)
    focus: str = Field(default="", max_length=1200)
    date_range: str = Field(default="经典工作至今", max_length=120)
    depth: str = Field(default="standard", pattern="^(quick|standard|deep)$")
    language: str = Field(default="中文", max_length=40)
    paper_urls: list[str] = Field(default_factory=list, max_length=12)


class EvidenceCard(BaseModel):
    paper_id: str = Field(min_length=1)
    title: str = Field(min_length=2)
    year: str = Field(min_length=1)
    relevance: str = Field(min_length=3)
    problem: str = Field(min_length=10)
    core_idea: str = Field(min_length=10)
    delta: str = Field(min_length=10)
    evidence: list[str] = Field(min_length=1, max_length=4)
    limitations: list[str] = Field(min_length=1, max_length=4)


class EvidenceCardsEnvelope(BaseModel):
    cards: list[EvidenceCard] = Field(min_length=1)


SectionLayout = Literal[
    "text", "timeline", "comparison", "process", "equation",
    "evidence", "paper-analysis", "evidence-gap",
]


class SectionPlan(BaseModel):
    id: str = Field(min_length=2)
    title: str = Field(min_length=2, max_length=120)
    purpose: str = Field(min_length=10, max_length=500)
    layout: SectionLayout
    source_ids: list[str] = Field(default_factory=list)


class ReportOutline(BaseModel):
    title: str = Field(min_length=2, max_length=160)
    subtitle: str = Field(min_length=2, max_length=240)
    research_type: str = Field(min_length=3, max_length=80)
    executive_summary: list[str] = Field(min_length=1, max_length=6)
    section_plan: list[SectionPlan] = Field(min_length=3, max_length=12)


class ReportSection(BaseModel):
    id: str = Field(min_length=2)
    title: str = Field(min_length=2, max_length=120)
    purpose: str = Field(min_length=10, max_length=500)
    takeaway: str = Field(min_length=10, max_length=800)
    layout: SectionLayout
    content: dict[str, Any]
    source_ids: list[str] = Field(default_factory=list)

    @model_validator(mode="after")
    def validate_layout_content(self) -> "ReportSection":
        content = self.content
        if not content or len(json.dumps(content, ensure_ascii=False).strip()) < 20:
            raise ValueError("section content must be substantive")
        if self.layout == "timeline" and len(content.get("nodes", [])) < 2:
            raise ValueError("timeline content requires at least two nodes")
        if self.layout == "timeline" and not all(isinstance(node, dict) for node in content.get("nodes", [])):
            raise ValueError("timeline nodes must be objects")
        if self.layout == "comparison" and (
            len(content.get("columns", [])) < 2 or len(content.get("rows", [])) < 2
        ):
            raise ValueError("comparison content requires columns and at least two rows")
        if self.layout == "comparison" and not all(isinstance(row, dict) for row in content.get("rows", [])):
            raise ValueError("comparison rows must be objects")
        if self.layout in {"paper-analysis", "evidence-gap"}:
            meaningful_blocks = [value for value in content.values() if value not in (None, "", [], {})]
            if len(meaningful_blocks) < 2:
                raise ValueError(f"{self.layout} requires at least two substantive information blocks")
        return self


class SectionsEnvelope(BaseModel):
    sections: list[ReportSection] = Field(min_length=1, max_length=4)


class ResearchReport(BaseModel):
    title: str = Field(min_length=2, max_length=160)
    subtitle: str = Field(min_length=2, max_length=240)
    research_type: str = Field(min_length=3, max_length=80)
    executive_summary: list[str] = Field(min_length=1, max_length=6)
    sections: list[ReportSection] = Field(min_length=3, max_length=12)
    references: list[str] = Field(default_factory=list)
    limitations: list[str] = Field(default_factory=list)
    english_understanding: dict[str, Any] = Field(default_factory=dict)


DEPTHS = {
    "quick": {"discover": 24, "screen": 10, "deep": 2, "queries": 2},
    "standard": {"discover": 40, "screen": 20, "deep": 4, "queries": 3},
    "deep": {"discover": 60, "screen": 35, "deep": 8, "queries": 4},
}


def safe_slug(value: str) -> str:
    slug = re.sub(r"[^a-zA-Z0-9]+", "-", value).strip("-").lower()
    return (slug or "algorithm-topic")[:64]


def http_get_with_retries(
    url: str,
    *,
    headers: dict[str, str] | None = None,
    read_timeout: float = 90,
    attempts: int = 3,
) -> httpx.Response:
    last_error: Exception | None = None
    timeout = httpx.Timeout(connect=20, read=read_timeout, write=30, pool=30)
    for attempt in range(attempts):
        try:
            with httpx.Client(
                timeout=timeout,
                follow_redirects=True,
                headers=headers,
                http2=False,
            ) as client:
                response = client.get(url)
                response.raise_for_status()
                return response
        except httpx.HTTPStatusError as exc:
            last_error = exc
            retryable = exc.response.status_code == 429 or exc.response.status_code >= 500
            if not retryable or attempt + 1 >= attempts:
                raise
            retry_after = exc.response.headers.get("retry-after", "")
            delay = float(retry_after) if retry_after.replace(".", "", 1).isdigit() else 3 + attempt * 2
            time.sleep(min(delay, 8))
        except (httpx.TimeoutException, httpx.NetworkError, httpx.RemoteProtocolError) as exc:
            last_error = exc
            if attempt + 1 < attempts:
                time.sleep(2 + attempt * 3)
    raise RuntimeError(f"External source did not respond after {attempts} attempts: {url}") from last_error


def arxiv_search(topic: str, limit: int) -> list[dict[str, Any]]:
    query = quote_plus(f'all:"{topic}"')
    url = (
        "https://export.arxiv.org/api/query"
        f"?search_query={query}&start=0&max_results={limit}"
        "&sortBy=submittedDate&sortOrder=descending"
    )
    headers = {"User-Agent": f"algorithm-research-deck/1.0 ({os.getenv('CONTACT_EMAIL', 'contact-not-set')})"}
    response = http_get_with_retries(url, headers=headers, read_timeout=35, attempts=2)
    ns = {"a": "http://www.w3.org/2005/Atom"}
    root = ElementTree.fromstring(response.text)
    papers = []
    for entry in root.findall("a:entry", ns):
        abs_url = (entry.findtext("a:id", default="", namespaces=ns) or "").strip()
        authors = [x.findtext("a:name", default="", namespaces=ns) for x in entry.findall("a:author", ns)]
        papers.append({
            "id": abs_url.rsplit("/", 1)[-1],
            "title": " ".join((entry.findtext("a:title", default="", namespaces=ns) or "").split()),
            "summary": " ".join((entry.findtext("a:summary", default="", namespaces=ns) or "").split()),
            "authors": authors,
            "published": entry.findtext("a:published", default="", namespaces=ns),
            "updated": entry.findtext("a:updated", default="", namespaces=ns),
            "url": abs_url,
            "pdf_url": abs_url.replace("/abs/", "/pdf/") + ".pdf",
        })
    return papers


def openalex_search(query: str, limit: int) -> list[dict[str, Any]]:
    email = os.getenv("CONTACT_EMAIL", "").strip()
    api_key = os.getenv("OPENALEX_API_KEY", "").strip()
    url = f"https://api.openalex.org/works?search={quote_plus(query)}&per-page={min(limit, 100)}"
    if email:
        url += f"&mailto={quote_plus(email)}"
    if api_key:
        url += f"&api_key={quote_plus(api_key)}"
    response = http_get_with_retries(url, read_timeout=35, attempts=2)
    results = response.json().get("results", [])
    papers = []
    for work in results:
        primary = work.get("primary_location") or {}
        source = primary.get("source") or {}
        open_access = work.get("open_access") or {}
        best_oa = work.get("best_oa_location") or {}
        landing = primary.get("landing_page_url") or work.get("doi") or work.get("id") or ""
        pdf_url = best_oa.get("pdf_url") or primary.get("pdf_url") or ""
        authors = [
            ((item.get("author") or {}).get("display_name") or "")
            for item in work.get("authorships", [])
        ]
        inverted = work.get("abstract_inverted_index") or {}
        ordered_words = sorted(
            ((position, word) for word, positions in inverted.items() for position in positions),
            key=lambda item: item[0],
        )
        abstract = " ".join(word for _, word in ordered_words)
        papers.append({
            "id": (work.get("doi") or work.get("id") or landing).rsplit("/", 1)[-1],
            "title": work.get("title") or "Untitled",
            "summary": abstract,
            "authors": [x for x in authors if x],
            "published": work.get("publication_date") or "",
            "updated": "",
            "url": landing,
            "pdf_url": pdf_url,
            "venue": source.get("display_name") or "",
            "citation_count": work.get("cited_by_count") or 0,
            "open_access": open_access.get("is_oa", False),
            "source": "OpenAlex",
        })
    return papers


def normalize_user_papers(urls: list[str]) -> list[dict[str, Any]]:
    result = []
    for raw in urls:
        match = re.search(r"arxiv\.org/(?:abs|pdf)/([\w.\-/]+)", raw)
        if match:
            paper_id = match.group(1).removesuffix(".pdf")
            result.append({
                "id": paper_id,
                "title": f"User supplied arXiv paper {paper_id}",
                "summary": "User-supplied paper; prioritize during screening.",
                "authors": [],
                "published": "",
                "updated": "",
                "url": f"https://arxiv.org/abs/{paper_id}",
                "pdf_url": f"https://arxiv.org/pdf/{paper_id}.pdf",
                "user_supplied": True,
            })
    return result


def deduplicate(papers: list[dict[str, Any]]) -> list[dict[str, Any]]:
    seen = set()
    output = []
    for paper in papers:
        normalized_title = re.sub(r"[^a-z0-9]+", "", paper.get("title", "").lower())
        key = normalized_title or re.sub(r"v\d+$", "", paper["id"])
        if key not in seen:
            seen.add(key)
            output.append(paper)
    return output


def plan_search_queries(request: ResearchRequest) -> tuple[list[str], dict[str, Any]]:
    prompt = json.dumps({
        "topic": request.topic,
        "purpose": request.purpose,
        "audience": request.audience,
        "focus": request.focus,
        "date_range": request.date_range,
        "language": request.language,
        "instruction": (
            "First translate and normalize every user field into precise English technical meaning without dropping the original intent. "
            "Return english_understanding with topic, purpose, audience, focus, date_range, key_concepts, ambiguities, and scope. "
            "Also return search_queries containing 3 to 5 concise English academic queries. "
            "Cover historical/classical terminology, milestone method families, current terminology, and one limitation or benchmark angle. "
            "Queries must be suitable for arXiv and OpenAlex. Do not include site operators."
        ),
    }, ensure_ascii=False)
    result = chat_json(
        "You plan literature searches for an algorithm engineer. Expand ambiguous topic names into precise academic terminology.",
        prompt,
        650,
    )
    queries = [str(x).strip() for x in result.get("search_queries", []) if str(x).strip()]
    english_understanding = result.get("english_understanding") or {}
    fallback_query = str(english_understanding.get("topic") or request.topic).strip()
    if fallback_query and fallback_query not in queries:
        queries.append(fallback_query)
    return queries[:6], english_understanding


def search_error_summary(exc: Exception) -> str:
    if exc.__cause__ is not None and isinstance(exc.__cause__, Exception):
        return search_error_summary(exc.__cause__)
    if isinstance(exc, httpx.HTTPStatusError):
        return f"HTTP {exc.response.status_code}"
    if isinstance(exc, httpx.TimeoutException) or "timed out" in str(exc).lower():
        return "请求超时"
    return type(exc).__name__


def cached_search(
    provider: str,
    query: str,
    limit: int,
    search_fn: Any,
) -> list[dict[str, Any]]:
    key = (provider, query.strip().lower(), limit)
    cached = SEARCH_CACHE.get(key)
    if cached and time.monotonic() - cached[0] < SEARCH_CACHE_TTL:
        return [dict(paper) for paper in cached[1]]
    minimum_interval = 1.2 if provider == "arxiv" else 0.4
    elapsed = time.monotonic() - SEARCH_LAST_CALL.get(provider, 0.0)
    if elapsed < minimum_interval:
        time.sleep(minimum_interval - elapsed)
    result = search_fn(query, limit)
    SEARCH_LAST_CALL[provider] = time.monotonic()
    SEARCH_CACHE[key] = (time.monotonic(), [dict(paper) for paper in result])
    return result


def search_literature(queries: list[str], total_limit: int) -> tuple[list[dict[str, Any]], list[str]]:
    per_query = max(5, min(20, total_limit // max(1, len(queries))))
    buckets: list[list[dict[str, Any]]] = []
    warnings: list[str] = []
    failures = {"arXiv": 0, "OpenAlex": 0}
    for query in queries:
        query_papers: list[dict[str, Any]] = []
        if failures["arXiv"] < 2:
            try:
                found = cached_search("arxiv", query, per_query, arxiv_search)
                for paper in found:
                    paper["source"] = "arXiv"
                query_papers.extend(found)
                failures["arXiv"] = 0
            except Exception as exc:
                failures["arXiv"] += 1
                warnings.append(f"arXiv {search_error_summary(exc)}：{query}")
        if len(query_papers) < max(2, per_query // 4) and failures["OpenAlex"] < 2:
            try:
                query_papers.extend(cached_search("openalex", query, per_query, openalex_search))
                failures["OpenAlex"] = 0
            except Exception as exc:
                failures["OpenAlex"] += 1
                warnings.append(f"OpenAlex {search_error_summary(exc)}：{query}")
        buckets.append(deduplicate(query_papers))

    for provider, count in failures.items():
        if count >= 2:
            warnings.append(f"{provider} 连续失败，已停止剩余重复请求")

    balanced: list[dict[str, Any]] = []
    max_bucket = max((len(bucket) for bucket in buckets), default=0)
    for index in range(max_bucket):
        for bucket in buckets:
            if index < len(bucket):
                balanced.append(bucket[index])
        balanced = deduplicate(balanced)
        if len(balanced) >= total_limit:
            break
    return deduplicate(balanced)[:total_limit], warnings


def generate_text(system: str, prompt: str, max_tokens: int, structured: bool = True) -> str:
    token = os.getenv("HF_TOKEN")
    if not token:
        raise RuntimeError("Space Secret HF_TOKEN is missing")
    client = InferenceClient(provider=HF_PROVIDER, api_key=token, timeout=180)
    request_args = {
        "model": MODEL_ID,
        "messages": [
            {"role": "system", "content": f"{system}\nUse non-thinking mode. Return only the requested final answer."},
            {"role": "user", "content": prompt},
        ],
        "temperature": 0.2,
        "max_tokens": max(2048, max_tokens),
    }
    if structured:
        try:
            response = client.chat.completions.create(
                **request_args,
                response_format={"type": "json_object"},
            )
        except Exception:
            response = client.chat.completions.create(**request_args)
    else:
        response = client.chat.completions.create(**request_args)
    message = response.choices[0].message
    content = message.content
    if isinstance(content, list):
        content = "".join(
            item.get("text", "") if isinstance(item, dict) else str(item)
            for item in content
        )
    if not content:
        content = getattr(message, "reasoning_content", None) or ""
    return str(content).strip()


def parse_json_object(content: str) -> dict[str, Any] | None:
    candidates = [content]
    fenced = re.sub(r"^```(?:json)?\s*|\s*```$", "", content, flags=re.IGNORECASE)
    if fenced != content:
        candidates.append(fenced)
    line_continuations_removed = re.sub(r"\\\r?\n", "", fenced)
    if line_continuations_removed != fenced:
        candidates.append(line_continuations_removed)
    match = re.search(r"\{[\s\S]*\}", content)
    if match:
        candidates.append(match.group(0))
    for candidate in candidates:
        try:
            parsed = json.loads(candidate)
            if isinstance(parsed, dict):
                return parsed
        except (json.JSONDecodeError, TypeError):
            continue
    return None


def chat_json(system: str, prompt: str, max_tokens: int = 6000) -> dict[str, Any]:
    last_content = ""
    for attempt in range(2):
        retry_rules = ""
        retry_prompt = prompt
        if attempt:
            retry_rules = (
                " The previous response was malformed or degenerated. Regenerate from scratch as compact valid JSON."
                " Never put literal or escaped line breaks in string values. Do not repeat whitespace or characters."
                " Keep titles under 120 characters and every other string under 500 characters. Close every array and object."
            )
            retry_prompt = (
                f"{prompt}\n\nThe previous response could not be parsed. Its beginning was: "
                f"{last_content[:320]!r}\nRegenerate the requested object from scratch."
            )
        last_content = generate_text(system + retry_rules, retry_prompt, max_tokens)
        parsed = parse_json_object(last_content)
        if parsed is not None:
            return parsed
    preview = last_content[:240].replace("\n", " ") if last_content else "<empty response>"
    raise RuntimeError(
        f"{MODEL_ID} did not return a JSON object. Provider={HF_PROVIDER}. Response preview: {preview}"
    )


def select_papers(request: ResearchRequest, papers: list[dict[str, Any]], screen_limit: int, deep_limit: int) -> list[str]:
    compact = [{k: p[k] for k in ("id", "title", "summary", "authors", "published", "url")} for p in papers[:screen_limit]]
    prompt = json.dumps({
        "topic": request.topic,
        "purpose": request.purpose,
        "focus": request.focus,
        "papers": compact,
        "instruction": f"Select at most {deep_limit} pivotal papers. Return JSON with selected_ids and a short selection_reason per id.",
    }, ensure_ascii=False)
    result = chat_json("You select technically pivotal ML papers. Prefer coverage of the user's decision and strong evidence over recency alone.", prompt, 900)
    valid = {p["id"] for p in papers}
    return [x for x in result.get("selected_ids", []) if x in valid][:deep_limit]


def extract_pdf(paper: dict[str, Any], max_chars: int = 14000) -> str:
    if not paper.get("pdf_url"):
        return "No open PDF URL was available from the academic index."
    response = http_get_with_retries(paper["pdf_url"], read_timeout=120, attempts=3)
    if len(response.content) > 20 * 1024 * 1024:
        return "PDF omitted because it exceeds the 20 MB limit."
    reader = PdfReader(io.BytesIO(response.content))
    pages = []
    indexes = list(range(min(8, len(reader.pages))))
    if len(reader.pages) > 8:
        indexes += list(range(max(8, len(reader.pages) - 3), len(reader.pages)))
    for index in dict.fromkeys(indexes):
        text = reader.pages[index].extract_text() or ""
        pages.append(f"[Page {index + 1}]\n{text}")
        if sum(map(len, pages)) >= max_chars:
            break
    return "\n".join(pages)[:max_chars]


def serializable_validation_errors(exc: ValidationError) -> list[dict[str, Any]]:
    """Remove exception objects and raw inputs before sending validation feedback to the model."""
    return exc.errors(include_url=False, include_context=False, include_input=False)


def validate_stage_with_repairs(
    result: dict[str, Any],
    model_type: type[BaseModel],
    stage_name: str,
    source_context: dict[str, Any],
    max_tokens: int,
) -> BaseModel:
    candidate = result
    for attempt in range(3):
        try:
            return model_type.model_validate(candidate)
        except ValidationError as exc:
            if attempt == 2:
                raise RuntimeError(f"{stage_name} remained invalid after two repairs: {exc}") from exc
            candidate = chat_json(
                f"Repair the {stage_name} JSON. Preserve valid content, fill missing fields, and return only the complete corrected object.",
                json.dumps({
                    "validation_errors": serializable_validation_errors(exc),
                    "object_to_repair": candidate,
                    "source_context": source_context,
                    "schema": model_type.model_json_schema(),
                }, ensure_ascii=False),
                max_tokens,
            )
    raise RuntimeError(f"{stage_name} validation failed")


def align_sections_to_plan(result: dict[str, Any], plans: list[SectionPlan]) -> dict[str, Any]:
    """Order model sections by approved IDs and expose omissions to schema repair."""
    candidates = [item for item in result.get("sections", []) if isinstance(item, dict)]
    by_id = {str(item.get("id")): item for item in candidates if item.get("id")}
    plan_ids = {plan.id for plan in plans}
    unused = [item for item in candidates if str(item.get("id", "")) not in plan_ids]
    aligned = []
    for plan in plans:
        candidate = by_id.get(plan.id)
        if candidate is None and unused:
            candidate = unused.pop(0)
        candidate = dict(candidate or {})
        candidate.update({
            "id": plan.id,
            "title": plan.title,
            "purpose": plan.purpose,
            "layout": plan.layout,
        })
        candidate.setdefault("takeaway", "")
        candidate.setdefault("content", {})
        candidate.setdefault("source_ids", [])
        aligned.append(candidate)
    return {"sections": aligned}


def prepare_evidence_inputs(
    papers: list[dict[str, Any]], deep_texts: dict[str, str], limit: int
) -> list[dict[str, Any]]:
    """Prioritize deeply read papers and bound evidence-card output size."""
    selected_ids = set(deep_texts)
    ordered = sorted(papers, key=lambda paper: paper.get("id") not in selected_ids)
    evidence_inputs = []
    for paper in ordered[:limit]:
        item = {k: paper.get(k) for k in ("id", "title", "summary", "authors", "published", "updated", "url")}
        if paper.get("id") in deep_texts:
            item["selected_text_extract"] = deep_texts[paper["id"]][:9000]
        evidence_inputs.append(item)
    return evidence_inputs


def infer_section_source_ids(section: dict[str, Any], papers: list[dict[str, Any]]) -> list[str]:
    """Recover omitted citations only when the section clearly names a retrieved paper or its authors."""
    source_ids = list(section.get("source_ids", []))
    haystack = json.dumps(
        {key: section.get(key) for key in ("title", "takeaway", "content")}, ensure_ascii=False
    ).lower()
    compact_haystack = re.sub(r"\W+", "", haystack)
    for paper in papers:
        paper_id = paper.get("id")
        if not paper_id or paper_id in source_ids:
            continue
        title = re.sub(r"\W+", "", str(paper.get("title", "")).lower())
        year = str(paper.get("published", ""))[:4]
        surnames = {
            re.sub(r"[^a-z0-9]", "", str(author).lower().split()[-1])
            for author in paper.get("authors", []) if str(author).strip()
        }
        mentioned_authors = [name for name in surnames if len(name) >= 4 and name in haystack]
        title_match = len(title) >= 12 and title in compact_haystack
        author_match = len(mentioned_authors) >= 2 or (
            len(mentioned_authors) == 1 and len(mentioned_authors[0]) >= 6 and year and year in haystack
        )
        if title_match or author_match:
            source_ids.append(paper_id)
    return source_ids


def build_report(
    request: ResearchRequest,
    papers: list[dict[str, Any]],
    deep_texts: dict[str, str],
    english_understanding: dict[str, Any],
) -> dict[str, Any]:
    evidence_limit = min(8, DEPTHS[request.depth]["deep"])
    raw_evidence = prepare_evidence_inputs(papers, deep_texts, evidence_limit)

    if raw_evidence:
        card_payload = {
            "request": request.model_dump(),
            "papers": raw_evidence,
            "requirements": {
                "one_card_per_useful_paper": True,
                "paraphrase_instead_of_copy": True,
                "cite_only_supplied_paper_ids": True,
                "schema": EvidenceCardsEnvelope.model_json_schema(),
            },
        }
        card_prompt = json.dumps(card_payload, ensure_ascii=False)
        try:
            cards_result = chat_json(
                "Create one compact evidence card per supplied paper. Keep only decision-relevant experimental conditions "
                "and limitations. Do not invent results, citations, or paper IDs. Return a complete JSON object; stay concise.",
                card_prompt,
                3600,
            )
        except RuntimeError as first_error:
            cards_result = chat_json(
                "The previous evidence-card response was incomplete or invalid JSON. Regenerate it more concisely. "
                "Return exactly one short card per supplied paper as a complete JSON object with no markdown or commentary.",
                json.dumps({
                    "previous_error": str(first_error),
                    "request": request.model_dump(),
                    "papers": [{k: item.get(k) for k in ("id", "title", "summary", "published")} for item in raw_evidence],
                    "schema": EvidenceCardsEnvelope.model_json_schema(),
                }, ensure_ascii=False),
                3000,
            )
        cards = validate_stage_with_repairs(
            cards_result,
            EvidenceCardsEnvelope,
            "evidence cards",
            {"request": request.model_dump(), "papers": raw_evidence},
            3600,
        ).model_dump()["cards"]
    else:
        cards = []

    shared_context = {
        "request": request.model_dump(),
        "english_understanding": english_understanding,
        "research_cutoff": datetime.now(timezone.utc).date().isoformat(),
        "evidence_cards": cards,
        "evidence_rule": (
            "Use only supplied paper IDs for citations. State evidence gaps instead of inventing support."
            if cards else
            "Online evidence is disabled. Do not invent paper IDs or claim that recent literature was verified."
        ),
    }

    outline_result = chat_json(
        "Act as the editor of an algorithm-engineering research presentation. First infer the most useful research type "
        "from the user's topic and intent, then design a topic-specific outline. Return only JSON matching the schema. "
        "Do not force a timeline, comparison, equations, or paper review when they do not fit. Choose each section layout "
        "deliberately. For unfamiliar or weakly evidenced topics, include an evidence-gap section instead of fabricating facts.",
        json.dumps({
            **shared_context,
            "requirements": {
                "research_type": "Choose a precise type such as concept survey, technical evolution, paper analysis, engineering decision, or exploratory evidence-gap report.",
                "executive_summary_items": "1 to 6 substantive conclusions appropriate to available evidence",
                "section_plan_items": "3 to 12; structure follows the topic rather than a fixed template",
                "allowed_layouts": list(SectionLayout.__args__),
                "schema": ReportOutline.model_json_schema(),
            },
        }, ensure_ascii=False),
        3000,
    )
    outline = validate_stage_with_repairs(
        outline_result,
        ReportOutline,
        "report outline",
        shared_context,
        3200,
    )

    section_system = (
        "Write presentation-ready sections for algorithm engineers. Use the layout selected in the approved plan and put "
        "layout-specific data in content. Text/evidence uses "
        "{points:[{heading,detail,subpoints}]} where each point keeps its detail and subpoints together; timeline uses "
        "{nodes:[{year,title,problem,innovation,limitation}]}; comparison uses "
        "{columns:[...],rows:[{name,values:[...]}]}; process uses {steps:[{name,description}]} or "
        "{common_steps:[...],branches:[{name,steps:[...]}]}; equation uses {equation,symbols,interpretation}; "
        "paper-analysis uses {method,evidence,limitations}; evidence-gap uses "
        "{interpretation,related_topics,hypotheses,required_inputs}. Equivalent topic-appropriate field names are allowed; "
        "never add empty placeholder fields merely to match an example. Write for slides, not an essay: one conclusion per "
        "section, concise points, explicit assumptions, comparable dimensions, and no repeated background. Distinguish "
        "retrieved facts from analysis or hypotheses. Never invent sources, papers, metrics, or recency claims."
    )
    generated_sections: list[ReportSection] = []
    plans = outline.section_plan
    for batch_index in range(0, len(plans), 4):
        batch = plans[batch_index:batch_index + 4]
        sections_result = chat_json(
            section_system,
            json.dumps({
                **shared_context,
                "report_identity": {
                    "title": outline.title,
                    "research_type": outline.research_type,
                    "executive_summary": outline.executive_summary,
                },
                "approved_section_plan": [item.model_dump() for item in batch],
                "requirements": {
                    "exactly_one_section_per_plan_item_in_order": True,
                    "preserve_plan_id_title_purpose_layout_exactly": True,
                    "source_ids": "Only IDs present in evidence_cards; otherwise []",
                    "schema": SectionsEnvelope.model_json_schema(),
                },
            }, ensure_ascii=False),
            4200,
        )
        sections_result = align_sections_to_plan(sections_result, batch)
        try:
            validated_batch = validate_stage_with_repairs(
                sections_result,
                SectionsEnvelope,
                f"report sections batch {batch_index // 4 + 1}",
                {**shared_context, "approved_section_plan": [item.model_dump() for item in batch]},
                4200,
            )
        except RuntimeError:
            validated_batch = None
        if validated_batch is None or len(validated_batch.sections) != len(batch):
            recovered_sections: list[ReportSection] = []
            for plan in batch:
                one_result = chat_json(
                    section_system,
                    json.dumps({
                        **shared_context,
                        "report_identity": {
                            "title": outline.title,
                            "research_type": outline.research_type,
                            "executive_summary": outline.executive_summary,
                        },
                        "approved_section_plan": [plan.model_dump()],
                        "requirements": {
                            "return_exactly_one_section": True,
                            "preserve_plan_identity_and_layout": True,
                            "source_ids": "Only IDs present in evidence_cards; otherwise []",
                            "schema": SectionsEnvelope.model_json_schema(),
                        },
                    }, ensure_ascii=False),
                    2600,
                )
                one_result = align_sections_to_plan(one_result, [plan])
                recovered = validate_stage_with_repairs(
                    one_result,
                    SectionsEnvelope,
                    f"missing report section {plan.id}",
                    {**shared_context, "approved_section_plan": [plan.model_dump()]},
                    2800,
                )
                recovered_sections.append(recovered.sections[0])
            validated_batch = SectionsEnvelope(sections=recovered_sections)
        for plan, section in zip(batch, validated_batch.sections):
            normalized = section.model_dump()
            normalized.update({
                "id": plan.id, "title": plan.title, "purpose": plan.purpose, "layout": plan.layout,
            })
            generated_sections.append(ReportSection.model_validate(normalized))

    assembled = {
        "title": outline.title,
        "subtitle": outline.subtitle,
        "research_type": outline.research_type,
        "executive_summary": outline.executive_summary,
        "sections": [item.model_dump() for item in generated_sections],
        "references": sorted({source_id for section in generated_sections for source_id in section.source_ids}),
        "limitations": (["Online literature verification was disabled; current-work claims may be incomplete."] if not cards else []),
        "english_understanding": english_understanding,
    }
    allowed_ids = {paper["id"] for paper in papers}
    for section in assembled["sections"]:
        section["source_ids"] = [
            paper_id for paper_id in infer_section_source_ids(section, papers) if paper_id in allowed_ids
        ]
    assembled["references"] = sorted({
        paper_id for section in assembled["sections"] for paper_id in section["source_ids"]
    })
    return validate_report_with_repairs(assembled, shared_context)


def validate_report_with_repairs(report: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
    candidate = report
    for attempt in range(3):
        try:
            validated = ResearchReport.model_validate(candidate).model_dump()
            allowed_ids = {card.get("paper_id") for card in context.get("evidence_cards", [])}
            for section in validated["sections"]:
                section["source_ids"] = [paper_id for paper_id in section["source_ids"] if paper_id in allowed_ids]
            validated["references"] = [paper_id for paper_id in validated["references"] if paper_id in allowed_ids]
            return validated
        except ValidationError as exc:
            if attempt == 2:
                raise RuntimeError(f"Report remained incomplete after two repairs: {exc}") from exc
            candidate = chat_json(
                "Repair an incomplete algorithm research report. Preserve its chosen research type, outline, layouts, and valid content. "
                "Fix only schema-invalid fields. Return the complete report as one JSON object and never invent source IDs.",
                json.dumps({
                    "validation_errors": serializable_validation_errors(exc),
                    "report_to_repair": candidate,
                    "evidence_cards": context.get("evidence_cards", []),
                    "schema": ResearchReport.model_json_schema(),
                }, ensure_ascii=False),
                5200,
            )
            candidate["english_understanding"] = context.get("english_understanding", {})
    raise RuntimeError("Report validation failed")


def esc(value: Any) -> str:
    return html.escape(str(value or ""))


def field_label(key: Any) -> str:
    labels = {
        "points": "要点", "details": "详细说明", "claim": "结论", "claims": "结论",
        "conditions": "适用条件", "evidence": "证据", "limitations": "局限",
        "method": "方法", "mechanism": "机制", "problem": "问题", "innovation": "创新",
        "tradeoffs": "权衡", "interpretation": "解释", "related_topics": "相关主题",
        "hypotheses": "待验证假设", "required_inputs": "所需补充信息", "findings": "发现",
        "known_evidence": "已有证据", "unknowns": "证据缺口", "next_steps": "下一步验证",
        "open_questions": "待回答问题", "method_summary": "方法概述", "experimental_results": "实验结果",
    }
    normalized = str(key).strip().lower().replace("-", "_")
    return labels.get(normalized, str(key).replace("_", " ").replace("-", " ").strip().title())


def render_value(value: Any, *, depth: int = 0) -> str:
    """Render arbitrary model JSON recursively instead of leaking JSON/Python repr."""
    if value is None or value == "":
        return ""
    if isinstance(value, list):
        if not value:
            return ""
        return '<div class="content-list">' + "".join(
            f'<article class="content-card">{render_value(item, depth=depth + 1)}</article>'
            if isinstance(item, dict) else
            f'<div class="content-item">{render_value(item, depth=depth + 1)}</div>'
            for item in value
        ) + "</div>"
    if isinstance(value, dict):
        if not value:
            return ""
        heading = value.get("heading") or value.get("title") or value.get("name")
        detail = value.get("detail") or value.get("description") or value.get("text")
        subpoints = value.get("subpoints") or value.get("bullets") or value.get("items")
        if detail is not None or subpoints is not None:
            consumed = {"heading", "title", "name", "detail", "description", "text", "subpoints", "bullets", "items"}
            extras = {key: item for key, item in value.items() if key not in consumed}
            return (
                (f'<h4 class="point-heading">{esc(heading)}</h4>' if heading else "")
                + (f'<div class="point-detail">{render_value(detail, depth=depth + 1)}</div>' if detail is not None else "")
                + (f'<div class="point-subpoints">{render_value(subpoints, depth=depth + 1)}</div>' if subpoints is not None else "")
                + (render_value(extras, depth=depth + 1) if extras else "")
            )
        return '<div class="content-object">' + "".join(
            f'<div class="content-field"><h4>{esc(field_label(key))}</h4>{render_value(item, depth=depth + 1)}</div>'
            for key, item in value.items() if item is not None and item != ""
        ) + "</div>"
    if isinstance(value, bool):
        return f'<p>{"是" if value else "否"}</p>'
    return f"<p>{esc(value)}</p>"


def normalize_comparison(content: dict[str, Any]) -> tuple[str, list[str], list[tuple[str, list[Any]]]]:
    columns = [str(column) for column in content.get("columns", [])]
    raw_rows = content.get("rows", [])
    row_keys = ("name", "dimension", "criterion", "metric", "title", "method")
    row_label = "对比维度"

    max_values = max(
        (len(row.get("values", [])) for row in raw_rows if isinstance(row, dict) and isinstance(row.get("values"), list)),
        default=0,
    )
    first_is_label = bool(columns) and (
        any(word in columns[0].lower() for word in ("维度", "指标", "dimension", "criterion", "metric"))
        or (max_values and len(columns) == max_values + 1)
    )
    if first_is_label:
        row_label, columns = columns[0], columns[1:]

    normalized_rows: list[tuple[str, list[Any]]] = []
    for row in raw_rows:
        if not isinstance(row, dict):
            continue
        label = next((str(row[key]) for key in row_keys if row.get(key) not in (None, "")), "")
        values = row.get("values")
        if isinstance(values, dict):
            cells = [values.get(column, "") for column in columns]
        elif isinstance(values, list):
            cells = list(values)
            if len(cells) == len(columns) + 1 and str(cells[0]) == label:
                cells = cells[1:]
        else:
            cells = [row.get(column, "") for column in columns]
        cells = (cells + [""] * len(columns))[:len(columns)]
        normalized_rows.append((label, cells))
    return row_label, columns, normalized_rows


def normalize_process(content: dict[str, Any]) -> tuple[list[Any], list[dict[str, Any]]]:
    if isinstance(content.get("branches"), list):
        common = content.get("common_steps") or ([content["start"]] if content.get("start") else [])
        return common, [branch for branch in content["branches"] if isinstance(branch, dict)]

    steps = content.get("steps", [])
    common: list[Any] = []
    grouped: dict[str, list[Any]] = {}
    for step in steps:
        name = str(step.get("name", "")) if isinstance(step, dict) else str(step)
        match = re.match(r"^([^:：]{2,24})[:：]\s*(.+)$", name)
        if not match:
            common.append(step)
            continue
        lane, short_name = match.groups()
        normalized_step = dict(step) if isinstance(step, dict) else {"name": short_name}
        normalized_step["name"] = short_name
        grouped.setdefault(lane, []).append(normalized_step)
    if len(grouped) < 2:
        return [], [{"name": content.get("name", "流程"), "steps": steps}]
    return common, [{"name": name, "steps": lane_steps} for name, lane_steps in grouped.items()]


def render_process_node(step: Any, index: str) -> str:
    if not isinstance(step, dict):
        return f'<article class="flow-node"><span>{esc(index)}</span><p>{esc(step)}</p></article>'
    name = step.get("name") or step.get("title") or f"步骤 {index}"
    details = {key: value for key, value in step.items() if key not in {"name", "title"}}
    return (
        f'<details class="flow-node"><summary><span>{esc(index)}</span><strong>{esc(name)}</strong></summary>'
        f'<div class="flow-detail">{render_value(details)}</div></details>'
    )


def render_section_content(section: dict[str, Any]) -> str:
    content, layout = section.get("content", {}), section.get("layout", "text")
    if layout == "timeline":
        return '<div class="timeline">' + "".join(
            f'<article class="milestone"><span>{esc(n.get("year") or n.get("stage"))}</span>'
            f'<h3>{esc(n.get("title") or n.get("method") or n.get("name"))}</h3>'
            f'{render_value({k: v for k, v in n.items() if k not in {"year", "stage", "title", "method", "name"}})}</article>'
            for n in content.get("nodes", [])
        ) + "</div>"
    if layout == "comparison":
        row_label, columns, rows = normalize_comparison(content)
        head = "".join(f"<th>{esc(x)}</th>" for x in columns)
        body = "".join(
            f'<tr><th scope="row">{esc(label)}</th>{"".join(f"<td>{render_value(value)}</td>" for value in values)}</tr>'
            for label, values in rows
        )
        return f'<div class="table-wrap"><table class="comparison-table"><thead><tr><th>{esc(row_label)}</th>{head}</tr></thead><tbody>{body}</tbody></table></div>'
    if layout == "process":
        common, branches = normalize_process(content)
        common_html = "".join(render_process_node(step, str(i)) for i, step in enumerate(common, 1))
        branches_html = "".join(
            f'<section class="flow-lane"><h3>{esc(branch.get("name") or branch.get("title") or "分支")}</h3>'
            f'{"".join(render_process_node(step, f"{lane_index}.{step_index}") for step_index, step in enumerate(branch.get("steps", []), 1))}</section>'
            for lane_index, branch in enumerate(branches, 1)
        )
        fork = '<div class="flow-fork" aria-hidden="true">分支</div>' if common and len(branches) > 1 else ""
        return f'<div class="process-flow"><div class="flow-common">{common_html}</div>{fork}<div class="flow-branches">{branches_html}</div></div>'
    if layout == "equation":
        return (f'<pre class="equation">{esc(content.get("equation"))}</pre><p>{esc(content.get("interpretation"))}</p>'
                f'{render_value(content.get("symbols"))}')
    wrapper = "aside" if layout == "evidence-gap" else "div"
    return f'<{wrapper} class="{esc(layout)}">' + "".join(
        f'<h3>{esc(field_label(k))}</h3>{render_value(v)}' for k, v in content.items()
    ) + f"</{wrapper}>"


def render_presentation(report: dict[str, Any], papers: list[dict[str, Any]], cutoff: str) -> str:
    deck_css = "\n".join(
        (STATIC_DIR / name).read_text(encoding="utf-8")
        for name in ("deck.css", "dynamic-content.css")
    )
    deck_js = (STATIC_DIR / "deck.js").read_text(encoding="utf-8").replace("</script", "<\\/script")
    citations = {p["id"]: p for p in papers}
    evidence_mode = (report.get("english_understanding") or {}).get("evidence_mode", "")
    evidence_note = "未联网核验：内容基于模型已有知识，最新进展未经外部验证。" if evidence_mode else "已联网检索，并基于当前可用论文证据生成。"
    evidence_label = "模型知识 · 未联网核验" if evidence_mode else f"论文证据 · {len(report.get('references', []))} 篇引用"
    summary_html = "".join(
        f'<article class="summary-card"><span>{i:02d}</span><p>{esc(item)}</p></article>'
        for i, item in enumerate(report.get("executive_summary", []), 1)
    )
    agenda_items = "".join(
        f'<a href="#section-{i + 1}" data-slide-target="section-{i + 1}"><span>{i + 1:02d}</span>'
        f'<strong>{esc(section.get("title"))}</strong><small>{esc(section.get("layout"))}</small></a>'
        for i, section in enumerate(report.get("sections", []))
    )
    section_html = "".join(
        f'<section class="slide content-slide" id="section-{i + 1}" data-title="{esc(x.get("title"))}">'
        f'<header class="slide-header"><p class="eyebrow">{i + 3:02d} / {esc(x.get("layout"))}</p><h2>{esc(x.get("title"))}</h2>'
        f'<p class="takeaway">{esc(x.get("takeaway"))}</p></header><div class="slide-content">{render_section_content(x)}</div>'
        f'<footer class="slide-footer"><div class="source-chips">'
        f'{"".join(f"<a href=\"#ref-{esc(pid)}\" title=\"查看参考文献\">{esc(pid)}</a>" for pid in x.get("source_ids", [])) or "<span>本页未绑定外部论文</span>"}'
        f'</div><span>{i + 3:02d}</span></footer></section>'
        for i, x in enumerate(report.get("sections", []))
    )
    refs = "".join(
        f'<li id="ref-{esc(pid)}"><span class="ref-index">{i:02d}</span><div><a href="{esc(citations[pid].get("url"))}" target="_blank" rel="noopener noreferrer">{esc(citations[pid].get("title"))}</a>'
        f'<br><span>{esc(", ".join(citations[pid].get("authors", [])[:6]))} · {esc((citations[pid].get("published") or "")[:10])} · {esc(pid)}</span></div></li>'
        for i, pid in enumerate((pid for pid in report.get("references", []) if pid in citations), 1)
    )
    limitations = "".join(f"<li>{esc(x)}</li>" for x in report.get("limitations", []))
    title, subtitle = esc(report.get("title", "Algorithm Research")), esc(report.get("subtitle", ""))
    return f'''<!doctype html><html lang="zh-CN"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
<title>{title}</title><style>{deck_css}</style></head><body>
<nav class="deck-nav"><button id="menu" aria-label="打开目录" aria-expanded="false">☰</button><strong>{title}</strong><span id="counter"></span><button id="prev" aria-label="上一页">←</button><button id="next" aria-label="下一页">→</button><button id="print" title="打印或保存为 PDF">保存 PDF</button></nav>
<aside id="overview" class="overview" aria-hidden="true"><header><strong>演示目录</strong><button id="close-overview" aria-label="关闭目录">×</button></header><a data-slide-target="cover">封面</a><a data-slide-target="summary">核心结论</a><a data-slide-target="agenda">研究结构</a>{agenda_items}<a data-slide-target="references">参考文献与局限</a></aside><div id="scrim" class="scrim"></div>
<main><section class="slide cover active" id="cover" data-title="封面"><div class="cover-orb"></div><p class="eyebrow">ALGORITHM RESEARCH / {esc(cutoff)}</p><h1>{title}</h1><p class="subtitle">{subtitle}</p><div class="cover-meta"><span>{esc(report.get("research_type"))}</span><span>{esc(evidence_label)}</span></div><p class="evidence-note">{esc(evidence_note)}</p><p class="hint">方向键翻页 · 点击左上角查看目录 · 可打印保存</p></section>
<section class="slide" id="summary" data-title="核心结论"><p class="eyebrow">01 / EXECUTIVE SUMMARY</p><h2>先看结论</h2><div class="summary-grid">{summary_html}</div></section>
<section class="slide" id="agenda" data-title="研究结构"><p class="eyebrow">02 / RESEARCH MAP</p><h2>这份调研如何展开</h2><div class="agenda-grid">{agenda_items}</div></section>{section_html}
<section class="slide references" id="references" data-title="参考文献与局限"><p class="eyebrow">SOURCES / {esc(cutoff)}</p><h2>参考文献与边界</h2><p class="evidence-note">{esc(evidence_note)}</p><ol>{refs or '<li class="empty-state">本报告未绑定外部论文来源。</li>'}</ol><ul class="limitations">{limitations}</ul></section></main>
<div class="progress"><span id="progress-bar"></span></div><script>{deck_js}</script></body></html>'''


def create_bundle(job_dir: Path, html_path: Path, report_path: Path) -> Path:
    bundle = job_dir / "research-package.zip"
    with zipfile.ZipFile(bundle, "w", zipfile.ZIP_DEFLATED) as archive:
        archive.write(html_path, html_path.name)
        archive.write(report_path, report_path.name)
    return bundle


def preview_frame(content: str) -> str:
    source = html.escape(content, quote=True)
    return f'<iframe title="Interactive research presentation" srcdoc="{source}" style="width:100%;height:80vh;border:0;border-radius:14px;background:#071018;box-shadow:0 14px 40px #12283c2b" sandbox="allow-scripts allow-popups allow-modals"></iframe>'


def run_research_ui(
    topic: str,
    purpose: str,
    audience: str,
    focus: str,
    date_range: str,
    online_search: bool,
    depth: str,
    language: str,
    paper_urls: str,
    progress=gr.Progress(),
):
    if not topic or len(topic.strip()) < 2:
        raise gr.Error("请先填写至少两个字符的调研主题。")
    request = ResearchRequest(
        topic=topic.strip(),
        purpose=purpose,
        audience=audience,
        focus=focus,
        date_range=date_range,
        depth=depth.lower(),
        language=language,
        paper_urls=[x.strip() for x in paper_urls.splitlines() if x.strip()],
    )
    budget = DEPTHS[request.depth]
    progress(0.05, desc="模型正在理解主题并规划检索词")
    queries, english_understanding = plan_search_queries(request)
    queries = queries[:budget["queries"]]
    search_warnings: list[str] = []
    deep_texts: dict[str, str] = {}
    if online_search:
        progress(0.16, desc="正在联网检索 arXiv 和 OpenAlex")
        user_papers = normalize_user_papers(request.paper_urls)
        discovered_papers, search_warnings = search_literature(queries, budget["discover"])
        papers = deduplicate(user_papers + discovered_papers)
        screen = papers[: budget["screen"]]
        if screen:
            progress(0.28, desc="DeepSeek-V4-Flash 正在筛选关键论文")
            selected_ids = select_papers(request, screen, budget["screen"], budget["deep"])
            if not selected_ids:
                selected_ids = [p["id"] for p in screen[: budget["deep"]]]
            progress(0.48, desc="正在读取关键论文")
            for index, paper in enumerate(screen):
                if paper["id"] in selected_ids:
                    try:
                        deep_texts[paper["id"]] = extract_pdf(paper)
                    except Exception as exc:
                        deep_texts[paper["id"]] = f"PDF extraction failed: {exc}"
                progress(0.48 + 0.16 * (index + 1) / len(screen), desc="正在提取论文证据")
        else:
            search_warnings.append("学术搜索未返回可用论文，已降级为模型知识模式")
            english_understanding["evidence_mode"] = "Academic search returned no usable papers; model knowledge only."
            progress(0.48, desc="检索无结果，已降级为模型知识模式")
    else:
        papers = []
        screen = []
        english_understanding["evidence_mode"] = "Model knowledge only; online paper search was disabled by the user."
        progress(0.45, desc="已关闭联网检索，使用模型已有知识生成")
    progress(0.68, desc="DeepSeek-V4-Flash 正在规划大纲并生成自适应章节")
    try:
        report = build_report(request, screen, deep_texts, english_understanding)
    except (ValidationError, RuntimeError) as exc:
        raise gr.Error(f"报告结构生成失败，模型在两次定向修复后仍未满足完整性要求：{exc}") from exc
    cutoff = datetime.now(timezone.utc).date().isoformat()
    content = render_presentation(report, screen, cutoff)
    job_dir = DATA_DIR / uuid.uuid4().hex
    job_dir.mkdir(parents=True, exist_ok=True)
    slug = safe_slug(request.topic)
    html_path = job_dir / f"{slug}-research.html"
    report_path = job_dir / f"{slug}-research.json"
    html_path.write_text(content, encoding="utf-8")
    report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
    bundle_path = create_bundle(job_dir, html_path, report_path)
    progress(1, desc="完成")
    warning_text = f" 检索降级：{'；'.join(search_warnings)}。" if search_warnings else ""
    if online_search:
        status = (
            f"✅ 已完成：模型规划 {len(queries)} 组检索词，发现 {len(papers)} 篇，"
            f"筛选 {len(screen)} 篇，深读 {len(deep_texts)} 篇。{warning_text}"
        )
    else:
        status = "✅ 已完成：未联网检索，内容来自模型已有知识，论文与最新性未经过外部核验。"
    return status, preview_frame(content), str(html_path), str(report_path), str(bundle_path)


CSS = """
.gradio-container{max-width:none!important;background:linear-gradient(145deg,#f7f9fc,#eef3f8)!important}.app-hero{margin:4px 4px 18px;padding:18px 22px;border:1px solid #dce5ef;border-radius:16px;background:linear-gradient(120deg,#0c1d2d,#153651);color:#fff;box-shadow:0 12px 35px #16334d1a}.app-hero h1{margin:0 0 5px;color:#fff!important;font-size:25px}.app-hero p{margin:0;color:#fff!important}.app-hero span{display:inline-block;margin-top:10px;padding:4px 8px;border:1px solid #ffffff2b;border-radius:999px;color:#82ead8;font-size:11px}
.preview-column{min-width:0}.input-column{max-width:430px;padding:14px!important;border:1px solid #dce5ef;border-radius:14px;background:#ffffffb8;box-shadow:0 10px 30px #1a38500d}
.preview-toolbar{align-items:center!important;justify-content:flex-end!important;gap:6px!important;min-height:38px}
.preview-title{margin-right:auto!important;min-width:180px}.download-chip{flex:0 0 auto!important;min-width:42px!important;max-width:52px!important;position:relative}
.download-chip button,.download-chip a{min-width:42px!important;padding:5px 8px!important;font-size:12px!important}
.download-chip:hover:after{position:absolute;z-index:50;right:0;top:calc(100% + 7px);width:max-content;max-width:230px;padding:7px 9px;border-radius:7px;background:#161d2b;color:#f3f6fb;box-shadow:0 6px 20px #0006;font-size:12px;line-height:1.35;pointer-events:none}
.download-html:hover:after{content:"下载可独立打开的交互式 HTML 演示"}.download-json:hover:after{content:"下载结构化研究内容与技术路线 JSON"}.download-zip:hover:after{content:"下载包含 HTML 和研究 JSON 的完整压缩包"}
"""

with gr.Blocks(title="Algorithm Research Deck") as demo:
    gr.HTML("""<div class="app-hero"><h1>Algorithm Research Deck</h1><p>从问题定义、论文证据到技术判断，生成可保存的交互式研究演示。</p><span>DeepSeek-V4-Flash · arXiv · OpenAlex</span></div>""")
    with gr.Row():
        with gr.Column(scale=1, elem_classes="input-column"):
            topic = gr.Textbox(label="调研主题", placeholder="例如：Diffusion Transformer")
            purpose = gr.Textbox(label="用途", value="算法团队内部技术分享")
            audience = gr.Textbox(label="受众", value="熟悉机器学习基础的算法工程师")
            focus = gr.Textbox(label="重点关注", lines=3, placeholder="技术路线、推理加速、工程部署……")
            with gr.Row():
                depth = gr.Radio(["Quick", "Standard", "Deep"], value="Quick", label="研究深度")
                language = gr.Dropdown(["中文", "English"], value="中文", label="输出语言")
            date_range = gr.Textbox(label="时间范围", value="经典工作至今")
            online_search = gr.Checkbox(label="联网搜索论文（arXiv + OpenAlex）", value=True)
            paper_urls = gr.Textbox(label="指定 arXiv 链接", lines=3, placeholder="每行一个，可选")
            submit = gr.Button("开始调研", variant="primary")
            status = gr.Markdown("等待任务")
        with gr.Column(scale=2, elem_classes="preview-column"):
            with gr.Row(elem_classes="preview-toolbar"):
                gr.Markdown("### 交互演示", elem_classes="preview-title")
                html_file = gr.DownloadButton("HTML", size="sm", elem_classes=["download-chip", "download-html"])
                json_file = gr.DownloadButton("JSON", size="sm", elem_classes=["download-chip", "download-json"])
                zip_file = gr.DownloadButton("ZIP", size="sm", elem_classes=["download-chip", "download-zip"])
            preview = gr.HTML("<div style='height:80vh;display:grid;place-items:center;border-radius:14px;background:#071018;color:#758099'>交互演示将在这里生成</div>")

    submit.click(
        run_research_ui,
        inputs=[topic, purpose, audience, focus, date_range, online_search, depth, language, paper_urls],
        outputs=[status, preview, html_file, json_file, zip_file],
    )

demo.queue(default_concurrency_limit=1, max_size=8)

if __name__ == "__main__":
    demo.launch(theme=gr.themes.Soft(), css=CSS, ssr_mode=False)