agent/web/app.py

#!/usr/bin/env python3
"""Serve a local paper browser with lightweight Ollama-backed actions."""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
import xml.etree.ElementTree as ET
from collections import Counter, defaultdict
from itertools import combinations
from http import HTTPStatus
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any


ROOT = Path(__file__).resolve().parents[1]
WEB_ROOT = ROOT / "web"
STATIC_ROOT = WEB_ROOT / "static"
INDEX_PATH = ROOT / "data" / "index.json"
CACHE_ROOT = WEB_ROOT / "cache"
ARXIV_CACHE = CACHE_ROOT / "arxiv"
AI_CACHE = CACHE_ROOT / "ai"
OLLAMA_URL = os.environ.get("OLLAMA_URL", "http://192.168.1.10:11434").rstrip("/")

DEFAULT_MODELS = {
    "translate": "ChatGPT-5.6:light",
    "summary": "ChatGPT-5.6:fast",
    "deep": "ChatGPT-5.6:large",
    "atlas": "ChatGPT-5.6:fast",
    "topic": "ChatGPT-5.6:large",
    "path": "ChatGPT-5.6:large",
    "compare": "ChatGPT-5.6:large",
    "search_plan": "ChatGPT-5.6:light",
    "search_summary": "ChatGPT-5.6:fast",
    "search_followup": "ChatGPT-5.6:fast",
    "paper_chat": "ChatGPT-5.6:fast",
}

NS = {
    "atom": "http://www.w3.org/2005/Atom",
    "arxiv": "http://arxiv.org/schemas/atom",
}

INDEX_LOCK = threading.Lock()
OLLAMA_LOCK = threading.Lock()
SEARCH_LOCK = threading.Lock()
INDEX_CACHE: dict[str, Any] = {"mtime": 0.0, "papers": [], "by_id": {}}
ATLAS_CACHE: dict[str, Any] = {"mtime": 0.0, "atlas": None}
NOTE_CACHE: dict[str, dict[str, Any]] = {}
SEARCH_JOBS: dict[str, dict[str, Any]] = {}

TOPIC_PROFILES = {
    "agent-evaluation": {
        "label": "Evaluation",
        "stance": "评估已经成为 Agent 研究的中心层：不只是看最终答案，而是看轨迹、工具调用、失败点、成本和安全边界。",
        "question": "如何证明一个 Agent 真的能在长程、不确定、带工具的环境里可靠完成任务？",
    },
    "tool-use": {
        "label": "Tool Use",
        "stance": "工具调用正在从 prompt 技巧变成运行时系统问题：schema、权限、依赖、失败恢复和可观测性都变成核心。",
        "question": "模型什么时候应该调用工具，如何调用，失败后如何恢复，工具链如何被治理？",
    },
    "memory": {
        "label": "Memory",
        "stance": "Memory 不再只是 RAG 的附属，而是长程 Agent 的状态管理、经验沉淀和安全攻击面。",
        "question": "Agent 应该记什么、忘什么、如何写入、如何检索、如何防止记忆污染？",
    },
    "rag": {
        "label": "Agentic RAG",
        "stance": "RAG 正在从相似度检索走向主动检索、证据规划、多跳推理和工具化知识访问。",
        "question": "Agent 如何把检索变成可验证、可迭代、可追踪的推理过程？",
    },
    "planning": {
        "label": "Planning",
        "stance": "Planning 的重点从一次性计划转向执行中的重规划、状态校正和长程任务稳定性。",
        "question": "长程任务里，计划如何与真实环境反馈、工具失败和上下文限制协同？",
    },
    "agent-safety": {
        "label": "Safety",
        "stance": "Agent safety 已经从拒答问题转向运行时风险：权限、注入、记忆污染、越权动作和治理平面。",
        "question": "如何限制 Agent 的行动边界，同时保留足够的自主性和生产力？",
    },
    "coding-agent": {
        "label": "Coding Agent",
        "stance": "Coding Agent 正在从单题修 bug 走向仓库级、长程、交互式的软件工程流程。",
        "question": "如何评估和构建能长期维护真实仓库的编码 Agent？",
    },
    "computer-use": {
        "label": "Computer Use",
        "stance": "GUI/browser/OS Agent 正在成为独立层：视觉状态、动作边界、任务进度和安全确认都需要重新建模。",
        "question": "Agent 如何可靠理解屏幕状态，并在真实界面中安全行动？",
    },
    "multi-agent": {
        "label": "Multi-Agent",
        "stance": "Multi-agent 正在从角色扮演走向共享记忆、协作协议、路由、治理和组织行为。",
        "question": "多个 Agent 如何共享状态、分工、纠错，并避免多数投票和上下文碎片带来的失败？",
    },
    "workflow-agent": {
        "label": "Workflow",
        "stance": "Workflow Agent 把模型能力接进企业流程，核心不在聊天，而在状态、权限、审计和可恢复执行。",
        "question": "怎样把 Agent 放进真实业务流程，并让它可观察、可治理、可回滚？",
    },
    "reasoning": {
        "label": "Reasoning",
        "stance": "Reasoning 在 Agent 中必须和工具、记忆、验证器、环境反馈一起看，而不是孤立的思维链。",
        "question": "推理如何真正改善行动，而不是只让轨迹看起来更复杂？",
    },
    "world-model": {
        "label": "World Model",
        "stance": "World model 让 Agent 从反应式调用走向对环境、规则和未来状态的内部建模。",
        "question": "Agent 需要什么样的世界模型才能更好地预测行动后果？",
    },
    "embodied-agent": {
        "label": "Embodied",
        "stance": "Embodied Agent 把语言、视觉、规划和物理约束连接起来，是 Agent 能力外延的重要方向。",
        "question": "语言模型如何在有物理约束的环境中做可执行决策？",
    },
}

TOPIC_POSITIONS = {
    "agent-evaluation": (51, 37),
    "tool-use": (37, 52),
    "memory": (52, 61),
    "rag": (68, 57),
    "planning": (48, 48),
    "agent-safety": (62, 35),
    "coding-agent": (27, 31),
    "computer-use": (29, 68),
    "multi-agent": (74, 42),
    "workflow-agent": (62, 74),
    "reasoning": (42, 24),
    "world-model": (78, 68),
    "embodied-agent": (82, 82),
}

KNOWLEDGE_ROUTES = [
    {
        "id": "evaluation-runtime",
        "title": "从答案评测到运行时评估",
        "topics": ["agent-evaluation", "tool-use", "agent-safety", "workflow-agent"],
        "summary": "Agent 的评估对象从 final answer 扩展到轨迹、工具调用、权限、失败恢复和上线治理。",
    },
    {
        "id": "memory-state",
        "title": "从 RAG 到状态管理",
        "topics": ["rag", "memory", "planning", "agent-safety"],
        "summary": "长期记忆、主动检索、写入策略和记忆安全正在把 RAG 推成 Agent 的状态层。",
    },
    {
        "id": "coding-long-horizon",
        "title": "从 SWE-Bench 到长程软件工程",
        "topics": ["coding-agent", "agent-evaluation", "memory", "tool-use"],
        "summary": "编码 Agent 的问题从单 bug 修复扩展到仓库理解、长程维护、交互澄清和 CI 反馈循环。",
    },
    {
        "id": "computer-use",
        "title": "从工具 API 到真实电脑操作",
        "topics": ["computer-use", "tool-use", "planning", "agent-safety"],
        "summary": "GUI、浏览器和 OS Agent 让状态识别、动作边界、确认机制和视觉记忆成为新的系统层。",
    },
    {
        "id": "collective-agents",
        "title": "从单体智能到协作系统",
        "topics": ["multi-agent", "memory", "agent-evaluation", "workflow-agent"],
        "summary": "多 Agent 研究正在走向共享记忆、角色分工、路由、群体失败归因和组织行为评估。",
    },
]


def ensure_cache_dirs() -> None:
    ARXIV_CACHE.mkdir(parents=True, exist_ok=True)
    AI_CACHE.mkdir(parents=True, exist_ok=True)


def clean_title(value: Any) -> str:
    title = str(value or "").strip()
    if len(title) >= 2 and title[0] == title[-1] == '"':
        return title[1:-1]
    return title


def item_id(path: str) -> str:
    return hashlib.sha1(path.encode("utf-8")).hexdigest()[:14]


def as_list(value: Any) -> list[str]:
    if value is None:
        return []
    if isinstance(value, list):
        return [str(item) for item in value if str(item).strip()]
    text = str(value).strip()
    return [text] if text else []


def score_value(value: Any) -> int:
    try:
        return int(str(value or "0"))
    except ValueError:
        return 0


def normalize_paper(item: dict[str, Any]) -> dict[str, Any]:
    meta = item.get("meta") or {}
    path = str(item.get("path") or "")
    title = clean_title(meta.get("title") or item.get("title") or Path(path).stem)
    topics = as_list(meta.get("topics"))
    datasets = as_list(meta.get("datasets"))
    authors = str(meta.get("authors") or "").strip()
    paper = {
        "id": item_id(path),
        "path": path,
        "title": title,
        "authors": authors,
        "year": str(meta.get("year") or ""),
        "venue": str(meta.get("venue") or ""),
        "url": str(meta.get("url") or ""),
        "code_url": as_list(meta.get("code_url")),
        "source": str(meta.get("source") or ""),
        "published_at": str(meta.get("published_at") or ""),
        "updated_at": str(meta.get("updated_at") or ""),
        "status": str(meta.get("status") or "unknown"),
        "relevance": str(meta.get("relevance") or ""),
        "topics": topics,
        "datasets": datasets,
        "collection_score": score_value(meta.get("collection_score")),
        "collection_queries": str(meta.get("collection_queries") or ""),
        "meta": meta,
    }
    search_text = " ".join(
        [
            title,
            authors,
            paper["year"],
            paper["venue"],
            paper["url"],
            paper["status"],
            paper["relevance"],
            paper["collection_queries"],
            " ".join(topics),
            " ".join(datasets),
        ]
    )
    paper["_search"] = search_text.lower()
    return paper


def load_papers() -> tuple[list[dict[str, Any]], dict[str, dict[str, Any]]]:
    mtime = INDEX_PATH.stat().st_mtime
    with INDEX_LOCK:
        if INDEX_CACHE["mtime"] == mtime:
            return INDEX_CACHE["papers"], INDEX_CACHE["by_id"]
        raw_items = json.loads(INDEX_PATH.read_text(encoding="utf-8"))
        papers = [
            normalize_paper(item)
            for item in raw_items
            if item.get("collection") == "papers" and item.get("path")
        ]
        by_id = {paper["id"]: paper for paper in papers}
        INDEX_CACHE.update({"mtime": mtime, "papers": papers, "by_id": by_id})
        return papers, by_id


def read_json_body(handler: SimpleHTTPRequestHandler) -> dict[str, Any]:
    length = int(handler.headers.get("content-length") or "0")
    if length <= 0:
        return {}
    raw = handler.rfile.read(length)
    return json.loads(raw.decode("utf-8"))


def write_json(handler: SimpleHTTPRequestHandler, payload: Any, status: int = 200) -> None:
    data = json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8")
    handler.send_response(status)
    handler.send_header("content-type", "application/json; charset=utf-8")
    handler.send_header("cache-control", "no-store")
    handler.send_header("content-length", str(len(data)))
    handler.end_headers()
    handler.wfile.write(data)


def write_error(handler: SimpleHTTPRequestHandler, status: int, message: str, **extra: Any) -> None:
    payload = {"error": message, **extra}
    write_json(handler, payload, status)


def paper_public(paper: dict[str, Any]) -> dict[str, Any]:
    return {key: value for key, value in paper.items() if not key.startswith("_") and key != "meta"}


def paper_card(paper: dict[str, Any]) -> dict[str, Any]:
    return {
        "id": paper["id"],
        "title": paper["title"],
        "year": paper["year"],
        "venue": paper["venue"],
        "url": paper["url"],
        "source": paper["source"],
        "published_at": paper["published_at"],
        "status": paper["status"],
        "relevance": paper["relevance"],
        "topics": paper["topics"],
        "collection_score": paper["collection_score"],
    }


def strip_note_frontmatter(markdown: str) -> str:
    text = str(markdown or "")
    match = re.match(r"^# .+?\n\n---[\s\S]*?\n---\n\n?", text)
    return text[match.end() :] if match else text


def paper_note_text(paper: dict[str, Any]) -> str:
    path = ROOT / paper["path"]
    try:
        mtime = path.stat().st_mtime
    except FileNotFoundError:
        return ""
    cache_key = str(path)
    cached = NOTE_CACHE.get(cache_key)
    if cached and cached.get("mtime") == mtime:
        return str(cached.get("text") or "")
    text = strip_note_frontmatter(path.read_text(encoding="utf-8", errors="replace"))
    text = normalize_space(text)
    NOTE_CACHE[cache_key] = {"mtime": mtime, "text": text}
    return text


def topic_label_map() -> dict[str, str]:
    return {topic: profile_for(topic)["label"] for topic in TOPIC_PROFILES}


def tokenize_query(value: str) -> list[str]:
    text = str(value or "").lower()
    tokens = re.findall(r"[a-z0-9][a-z0-9.+#/-]{1,}|[\u4e00-\u9fff]{2,}", text)
    return [token.strip(".,;:()[]{}") for token in tokens if token.strip(".,;:()[]{}")]


def unique_strings(items: list[Any], limit: int = 20) -> list[str]:
    seen: set[str] = set()
    output: list[str] = []
    for item in items:
        text = str(item or "").strip()
        key = text.lower()
        if not text or key in seen:
            continue
        seen.add(key)
        output.append(text)
        if len(output) >= limit:
            break
    return output


def extract_json_object(text: str) -> dict[str, Any]:
    raw = clean_model_response(text)
    start = raw.find("{")
    end = raw.rfind("}")
    if start < 0 or end <= start:
        raise ValueError("model did not return a JSON object")
    return json.loads(raw[start : end + 1])


QUERY_EXPANSIONS = {
    "强化学习": {
        "keywords_en": [
            "reinforcement learning",
            "RL",
            "policy optimization",
            "reward",
            "rollout",
            "self-feedback",
            "agent reinforcement learning",
            "credit assignment",
            "trajectory optimization",
        ],
        "topics": ["planning", "reasoning", "agent-evaluation", "multi-agent"],
    },
    "评估": {
        "keywords_en": ["evaluation", "benchmark", "trajectory", "judge", "rubric", "agent benchmark"],
        "topics": ["agent-evaluation"],
    },
    "记忆": {
        "keywords_en": ["memory", "long-horizon", "state", "retention", "retrieval memory"],
        "topics": ["memory", "rag"],
    },
    "工具": {
        "keywords_en": ["tool use", "tool calling", "function calling", "tool failure", "tool selection"],
        "topics": ["tool-use"],
    },
    "安全": {
        "keywords_en": ["safety", "security", "privacy", "prompt injection", "guardrail", "permission"],
        "topics": ["agent-safety"],
    },
}

SEARCH_STOP_TERMS = {
    "agent",
    "agents",
    "ai",
    "llm",
    "model",
    "models",
    "system",
    "systems",
    "method",
    "methods",
    "approach",
    "approaches",
    "algorithm",
    "algorithms",
    "application",
    "applications",
    "research",
    "study",
    "paper",
    "learning",
}


def filter_search_keywords(items: list[Any], limit: int = 24) -> list[str]:
    output: list[str] = []
    for item in unique_strings(items, limit * 3):
        key = item.lower().strip()
        if key in SEARCH_STOP_TERMS:
            continue
        if len(key) < 2:
            continue
        output.append(item)
        if len(output) >= limit:
            break
    return output


def heuristic_search_plan(query: str) -> dict[str, Any]:
    tokens = tokenize_query(query)
    keywords_en: list[str] = []
    topics: list[str] = []
    for key, expansion in QUERY_EXPANSIONS.items():
        if key in query:
            keywords_en.extend(expansion["keywords_en"])
            topics.extend(expansion["topics"])
    for topic, label in topic_label_map().items():
        haystack = f"{topic} {label}".lower()
        if any(token.lower() in haystack for token in tokens):
            topics.append(topic)
            keywords_en.append(label)
            keywords_en.append(topic.replace("-", " "))
    return {
        "intent": "topic_search",
        "interpreted_query": query,
        "keywords_zh": [token for token in tokens if re.search(r"[\u4e00-\u9fff]", token)],
        "keywords_en": filter_search_keywords(keywords_en + [token for token in tokens if re.search(r"[a-z0-9]", token)], 18),
        "topics": unique_strings(topics, 8),
        "filters": {},
        "search_rationale": "基于本地规则扩展中文/英文关键词，并映射到已有 Agent 主题。",
    }


def search_plan_prompt(query: str, scope_topic: str = "") -> str:
    topics = [
        {"id": topic, "label": profile_for(topic)["label"], "question": profile_for(topic)["question"]}
        for topic in TOPIC_PROFILES
    ]
    no_think = "不要输出思考过程，不要输出 <think> 标签。"
    return (
        "你是 Agent 论文库的搜索规划器。请理解用户输入，生成用于本地检索的 JSON。\n"
        "只返回 JSON 对象，不要 Markdown。字段：\n"
        "intent: topic_search|paper_lookup|question_answer|trend_scan|method_search|benchmark_search\n"
        "interpreted_query: 中文解释\nkeywords_zh: 中文关键词数组\nkeywords_en: 英文关键词数组\n"
        "topics: 从给定 topic id 中选择相关项数组\nfilters: 可为空对象，支持 recent_months 整数\n"
        "search_rationale: 一句话说明为什么这样搜。\n"
        f"当前 topic scope: {scope_topic or 'none'}\n"
        f"可选 topics:\n{json.dumps(topics, ensure_ascii=False)}\n"
        f"{no_think}\n\n用户输入：{query}"
    )


def plan_search_query(query: str, scope_topic: str = "") -> tuple[dict[str, Any], dict[str, Any] | None]:
    fallback = heuristic_search_plan(query)
    model = DEFAULT_MODELS["search_plan"]
    result: dict[str, Any] | None = None
    try:
        with OLLAMA_LOCK:
            result = call_ollama(model, search_plan_prompt(query, scope_topic), "search_plan")
        plan = extract_json_object(result["response"])
        merged = {
            **fallback,
            **{key: value for key, value in plan.items() if value not in (None, "", [])},
        }
        merged["keywords_zh"] = unique_strings(as_list(merged.get("keywords_zh")) + fallback["keywords_zh"], 16)
        merged["keywords_en"] = filter_search_keywords(as_list(merged.get("keywords_en")) + fallback["keywords_en"], 24)
        merged["topics"] = [topic for topic in unique_strings(as_list(merged.get("topics")) + fallback["topics"], 10) if topic in TOPIC_PROFILES]
        if not isinstance(merged.get("filters"), dict):
            merged["filters"] = {}
        return merged, result
    except Exception as exc:  # noqa: BLE001
        fallback["planner_error"] = str(exc)
        return fallback, result


def term_score(term: str, title: str, meta: str, note: str) -> float:
    if not term:
        return 0.0
    term_lower = term.lower()
    score = 0.0
    if term_lower in title:
        score += 50
    if term_lower in meta:
        score += 18
    if term_lower in note:
        score += min(22, 5 + note.count(term_lower) * 2)
    for token in tokenize_query(term_lower):
        if len(token) < 2:
            continue
        if token in title:
            score += 14
        elif token in meta:
            score += 8
        elif token in note:
            score += 3
    return score


def score_search_matches(
    papers: list[dict[str, Any]],
    query: str,
    plan: dict[str, Any],
    scope_topic: str = "",
    limit: int = 80,
) -> list[dict[str, Any]]:
    raw_terms = [query, *as_list(plan.get("keywords_zh")), *as_list(plan.get("keywords_en"))]
    terms = unique_strings(raw_terms, 32)
    plan_topics = [topic for topic in as_list(plan.get("topics")) if topic in TOPIC_PROFILES]
    scored: list[tuple[float, dict[str, Any], list[str]]] = []

    for paper in papers:
        if scope_topic and scope_topic not in paper["topics"]:
            continue
        title = paper["title"].lower()
        meta = " ".join(
            [
                paper["_search"],
                " ".join(profile_for(topic)["label"] for topic in paper["topics"]),
                " ".join(topic.replace("-", " ") for topic in paper["topics"]),
            ]
        ).lower()
        note = paper_note_text(paper).lower()
        evidence_score = 0.0
        matched_terms: list[str] = []
        for term in terms:
            value = term_score(term, title, meta, note)
            if value > 0:
                evidence_score += value
                matched_terms.append(term)
        topic_overlap = sorted(set(plan_topics) & set(paper["topics"]))
        if topic_overlap:
            evidence_score += 34 + 9 * len(topic_overlap)
            matched_terms.extend(profile_for(topic)["label"] for topic in topic_overlap)
        if query.lower() in title:
            evidence_score += 70
        if evidence_score <= 0:
            continue
        score = evidence_score + float(paper["collection_score"]) * 0.35
        item = paper_card(paper)
        item["match_score"] = round(score, 2)
        item["matched_terms"] = unique_strings(matched_terms, 8)
        scored.append((score, item, matched_terms))

    scored.sort(key=lambda row: (row[0], row[1].get("published_at") or "", row[1]["collection_score"]), reverse=True)
    return [item for _, item, _ in scored[:limit]]


def aggregate_search_results(items: list[dict[str, Any]]) -> dict[str, Any]:
    topic_counts: Counter = Counter()
    month_counts: Counter = Counter()
    year_counts: Counter = Counter()
    for item in items:
        month = month_key(str(item.get("published_at") or ""))
        if month != "unknown":
            month_counts[month] += 1
            year_counts[month[:4]] += 1
        for topic in item.get("topics") or []:
            topic_counts[topic] += 1

    topic_rows = [
        {"topic": topic, "label": profile_for(topic)["label"], "count": count}
        for topic, count in topic_counts.most_common(10)
    ]
    months = [{"month": month, "count": count} for month, count in sorted(month_counts.items())]
    recent_months = months[-6:]
    date_values = [str(item.get("published_at") or "") for item in items if re.match(r"^\d{4}-\d{2}-\d{2}", str(item.get("published_at") or ""))]
    return {
        "total": len(items),
        "topics": topic_rows,
        "months": months,
        "recent_months": recent_months,
        "years": [{"year": year, "count": count} for year, count in sorted(year_counts.items())],
        "date_range": {
            "start": min(date_values) if date_values else "",
            "end": max(date_values) if date_values else "",
        },
    }


def search_summary_prompt(query: str, plan: dict[str, Any], items: list[dict[str, Any]], aggregation: dict[str, Any]) -> str:
    compact_items = [
        {
            "title": item["title"],
            "published_at": item["published_at"],
            "topics": item["topics"],
            "score": item["collection_score"],
            "matched_terms": item.get("matched_terms", []),
        }
        for item in items[:18]
    ]
    context = {
        "query": query,
        "plan": plan,
        "aggregation": aggregation,
        "top_papers": compact_items,
    }
    no_think = "不要输出思考过程，不要输出 <think> 标签。"
    return (
        "你是 Agent 研究助理。请基于本地论文检索结果生成中文搜索简报。\n"
        "只基于给定结果，不要编造论文细节。输出要短，包含：\n"
        "1. 搜索理解\n2. 主要发现\n3. 时间/趋势\n4. 建议先读\n5. 可以继续追问的方向。\n"
        f"{no_think}\n\n{json.dumps(context, ensure_ascii=False, indent=2)[:14000]}"
    )


def fallback_search_summary(query: str, plan: dict[str, Any], aggregation: dict[str, Any], items: list[dict[str, Any]]) -> str:
    topics = "、".join(item["label"] for item in aggregation["topics"][:5]) or "暂无明显主题"
    papers = "；".join(item["title"] for item in items[:3]) or "暂无命中论文"
    keywords = ", ".join(as_list(plan.get("keywords_en"))[:8] + as_list(plan.get("keywords_zh"))[:4])
    return (
        f"搜索理解：你在查「{query}」，系统扩展了 {keywords or '原始关键词'}。\n\n"
        f"主要发现：命中 {aggregation['total']} 篇候选，集中在 {topics}。\n\n"
        f"时间/趋势：日期范围 {aggregation['date_range']['start'] or 'unknown'} 到 {aggregation['date_range']['end'] or 'unknown'}。\n\n"
        f"建议先读：{papers}。\n\n"
        "可以继续追问：只看最近论文、按主题拆分、生成阅读路径、比较工程/理论方向。"
    )


def generate_search_summary(query: str, plan: dict[str, Any], items: list[dict[str, Any]], aggregation: dict[str, Any]) -> dict[str, Any]:
    model = DEFAULT_MODELS["search_summary"]
    try:
        with OLLAMA_LOCK:
            result = call_ollama(model, search_summary_prompt(query, plan, items, aggregation), "search_summary")
        return {
            "text": result["response"],
            "model": result["model"],
            "duration_seconds": result["duration_seconds"],
            "cached": False,
        }
    except Exception as exc:  # noqa: BLE001
        return {
            "text": fallback_search_summary(query, plan, aggregation, items),
            "model": "fallback",
            "error": str(exc),
            "cached": False,
        }


def new_search_job(query: str, scope_topic: str = "") -> dict[str, Any]:
    job_id = hashlib.sha1(f"{query}:{scope_topic}:{time.time()}".encode("utf-8")).hexdigest()[:14]
    now = time.strftime("%Y-%m-%dT%H:%M:%S%z")
    job = {
        "id": job_id,
        "query": query,
        "scope_topic": scope_topic,
        "status": "queued",
        "created_at": now,
        "updated_at": now,
        "steps": [
            {"id": "understand", "label": "理解问题", "status": "pending", "detail": ""},
            {"id": "expand", "label": "扩展关键词", "status": "pending", "detail": ""},
            {"id": "retrieve", "label": "检索论文库", "status": "pending", "detail": ""},
            {"id": "aggregate", "label": "统计主题/趋势", "status": "pending", "detail": ""},
            {"id": "summarize", "label": "生成搜索结论", "status": "pending", "detail": ""},
        ],
        "result": None,
        "error": "",
    }
    with SEARCH_LOCK:
        SEARCH_JOBS[job_id] = job
        prune_search_jobs_locked()
    thread = threading.Thread(target=run_search_job, args=(job_id,), daemon=True)
    thread.start()
    return public_search_job(job)


def public_search_job(job: dict[str, Any]) -> dict[str, Any]:
    return {
        "id": job["id"],
        "query": job["query"],
        "scope_topic": job.get("scope_topic", ""),
        "status": job["status"],
        "created_at": job["created_at"],
        "updated_at": job["updated_at"],
        "steps": job["steps"],
        "result": job.get("result"),
        "error": job.get("error", ""),
    }


def get_search_job(job_id: str) -> dict[str, Any] | None:
    with SEARCH_LOCK:
        job = SEARCH_JOBS.get(job_id)
        return public_search_job(job) if job else None


def update_search_job(job_id: str, *, status: str | None = None, result: Any = None, error: str = "") -> None:
    with SEARCH_LOCK:
        job = SEARCH_JOBS.get(job_id)
        if not job:
            return
        if status:
            job["status"] = status
        if result is not None:
            job["result"] = result
        if error:
            job["error"] = error
        job["updated_at"] = time.strftime("%Y-%m-%dT%H:%M:%S%z")


def update_search_step(job_id: str, step_id: str, status: str, detail: str = "") -> None:
    with SEARCH_LOCK:
        job = SEARCH_JOBS.get(job_id)
        if not job:
            return
        for step in job["steps"]:
            if step["id"] == step_id:
                step["status"] = status
                if detail:
                    step["detail"] = detail
                step["updated_at"] = time.strftime("%H:%M:%S")
                break
        job["updated_at"] = time.strftime("%Y-%m-%dT%H:%M:%S%z")


def prune_search_jobs_locked() -> None:
    if len(SEARCH_JOBS) <= 24:
        return
    ordered = sorted(SEARCH_JOBS.values(), key=lambda item: item.get("created_at", ""))
    for job in ordered[: len(SEARCH_JOBS) - 24]:
        SEARCH_JOBS.pop(job["id"], None)


def run_search_job(job_id: str) -> None:
    job = get_search_job(job_id)
    if not job:
        return
    query = job["query"]
    scope_topic = job.get("scope_topic", "")
    try:
        update_search_job(job_id, status="running")
        update_search_step(job_id, "understand", "running", "调用轻量模型理解搜索意图")
        plan, planner_result = plan_search_query(query, scope_topic)
        update_search_step(job_id, "understand", "done", str(plan.get("interpreted_query") or query))

        keywords = unique_strings(as_list(plan.get("keywords_en")) + as_list(plan.get("keywords_zh")), 12)
        update_search_step(job_id, "expand", "done", " / ".join(keywords[:8]) or "使用原始输入")

        update_search_step(job_id, "retrieve", "running", "扫描标题、主题、collection query 和本地卡片")
        papers, _ = load_papers()
        items = score_search_matches(papers, query, plan, scope_topic, limit=80)
        update_search_step(job_id, "retrieve", "done", f"找到 {len(items)} 篇候选论文")

        update_search_step(job_id, "aggregate", "running", "统计日期、主题和最近趋势")
        aggregation = aggregate_search_results(items)
        update_search_step(
            job_id,
            "aggregate",
            "done",
            f"{len(aggregation['topics'])} 个相关主题，日期到 {aggregation['date_range']['end'] or 'unknown'}",
        )

        update_search_step(job_id, "summarize", "running", "调用 fast 模型生成搜索简报")
        summary = generate_search_summary(query, plan, items, aggregation)
        update_search_step(job_id, "summarize", "done", summary.get("model", "fallback"))

        result = {
            "query": query,
            "scope_topic": scope_topic,
            "plan": plan,
            "planner": {
                "model": planner_result.get("model") if planner_result else "fallback",
                "duration_seconds": planner_result.get("duration_seconds") if planner_result else None,
            },
            "summary": summary,
            "aggregation": aggregation,
            "items": items[:60],
            "total": len(items),
        }
        update_search_job(job_id, status="done", result=result)
    except Exception as exc:  # noqa: BLE001
        update_search_step(job_id, "summarize", "error", str(exc))
        update_search_job(job_id, status="error", error=str(exc))


def search_followup_prompt(job: dict[str, Any], question: str) -> str:
    result = job.get("result") or {}
    compact = {
        "query": result.get("query"),
        "plan": result.get("plan"),
        "summary": (result.get("summary") or {}).get("text"),
        "aggregation": result.get("aggregation"),
        "top_papers": [
            {
                "title": item["title"],
                "published_at": item["published_at"],
                "topics": item["topics"],
                "score": item["collection_score"],
                "matched_terms": item.get("matched_terms", []),
            }
            for item in (result.get("items") or [])[:18]
        ],
    }
    return (
        "你是 Agent 论文库搜索助手。用户正在基于一次搜索结果继续追问。\n"
        "请只基于给定搜索上下文回答，中文，简洁但有判断。不要编造论文细节。\n"
        "不要输出思考过程，不要输出 <think> 标签。\n\n"
        f"搜索上下文：\n{json.dumps(compact, ensure_ascii=False, indent=2)[:14000]}\n\n"
        f"用户追问：{question}"
    )


def paper_chat_prompt(paper: dict[str, Any], abstract: dict[str, Any] | None, question: str, history: list[dict[str, str]]) -> str:
    recent_history = [
        {"role": str(item.get("role") or ""), "content": str(item.get("content") or "")[:1200]}
        for item in history[-6:]
        if item.get("content")
    ]
    context = {
        "paper": {
            "title": paper["title"],
            "authors": paper["authors"],
            "published_at": paper["published_at"],
            "venue": paper["venue"],
            "topics": paper["topics"],
            "collection_score": paper["collection_score"],
            "url": paper["url"],
        },
        "abstract": abstract,
        "local_note": paper_note_text(paper)[:5000],
        "history": recent_history,
        "question": question,
    }
    return (
        "你是 Agent 论文研读助手。请基于论文上下文和对话历史回答用户问题。\n"
        "中文回答，优先给出判断和证据；信息不足就明确说不足。不要编造论文细节。\n"
        "不要输出思考过程，不要输出 <think> 标签。\n\n"
        f"{json.dumps(context, ensure_ascii=False, indent=2)[:15000]}"
    )


def topic_counts(papers: list[dict[str, Any]]) -> list[dict[str, Any]]:
    counts: dict[str, int] = {}
    for paper in papers:
        for topic in paper["topics"]:
            counts[topic] = counts.get(topic, 0) + 1
    return [
        {"topic": topic, "count": count}
        for topic, count in sorted(counts.items(), key=lambda item: (-item[1], item[0]))
    ]


def month_key(value: str) -> str:
    if re.match(r"^\d{4}-\d{2}", value or ""):
        return value[:7]
    return "unknown"


def profile_for(topic: str) -> dict[str, str]:
    fallback = {
        "label": topic.replace("-", " ").title(),
        "stance": f"{topic} 是当前语料中自动识别出的主题，需要进一步人工导读。",
        "question": "这个主题与 Agent 系统能力的关系是什么？",
    }
    return {**fallback, **TOPIC_PROFILES.get(topic, {})}


def top_papers_for(papers: list[dict[str, Any]], limit: int = 8, recent: bool = False) -> list[dict[str, Any]]:
    ranked = list(papers)
    if recent:
        ranked.sort(key=lambda item: (item["published_at"], item["collection_score"]), reverse=True)
    else:
        ranked.sort(key=lambda item: (item["collection_score"], item["published_at"]), reverse=True)
    return [paper_card(paper) for paper in ranked[:limit]]


def build_atlas() -> dict[str, Any]:
    papers, _ = load_papers()
    mtime = INDEX_PATH.stat().st_mtime
    if ATLAS_CACHE["mtime"] == mtime and ATLAS_CACHE["atlas"]:
        return ATLAS_CACHE["atlas"]

    topic_to_papers: dict[str, list[dict[str, Any]]] = defaultdict(list)
    topic_months: dict[str, Counter] = defaultdict(Counter)
    edge_counts: Counter = Counter()
    month_counts: Counter = Counter()

    for paper in papers:
        month = month_key(paper["published_at"])
        month_counts[month] += 1
        topics = sorted(set(paper["topics"]))
        for topic in topics:
            topic_to_papers[topic].append(paper)
            topic_months[topic][month] += 1
        for left, right in combinations(topics, 2):
            edge_counts[(left, right)] += 1

    months = [month for month in sorted(month_counts) if month != "unknown"]
    recent_months = set(months[-3:])
    max_count = max((len(items) for items in topic_to_papers.values()), default=1)
    max_recent = max(
        (sum(topic_months[topic][month] for month in recent_months) for topic in topic_to_papers),
        default=1,
    )

    topics = []
    fallback_index = 0
    for topic, topic_papers in sorted(topic_to_papers.items(), key=lambda item: (-len(item[1]), item[0])):
        profile = profile_for(topic)
        recent_count = sum(topic_months[topic][month] for month in recent_months)
        if topic in TOPIC_POSITIONS:
            x, y = TOPIC_POSITIONS[topic]
        else:
            x = 18 + (fallback_index % 5) * 16
            y = 16 + (fallback_index // 5) * 14
            fallback_index += 1
        topics.append(
            {
                "id": topic,
                "label": profile["label"],
                "count": len(topic_papers),
                "recent_count": recent_count,
                "share": round(len(topic_papers) / max(len(papers), 1), 3),
                "heat": round(recent_count / max_recent, 3),
                "radius": round(16 + 34 * (len(topic_papers) / max_count), 1),
                "x": x,
                "y": y,
                "stance": profile["stance"],
                "question": profile["question"],
            }
        )

    top_topic_ids = {topic["id"] for topic in topics[:14]}
    max_edge_count = max(edge_counts.values(), default=1)
    edges = [
        {
            "source": left,
            "target": right,
            "count": count,
            "strength": round(count / max_edge_count, 3),
        }
        for (left, right), count in edge_counts.most_common(80)
        if left in top_topic_ids and right in top_topic_ids
    ]

    timeline = []
    for month in months:
        row = {"month": month, "total": month_counts[month], "topics": {}}
        for topic in top_topic_ids:
            row["topics"][topic] = topic_months[topic][month]
        timeline.append(row)

    atlas = {
        "generated_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
        "paper_count": len(papers),
        "topic_count": len(topics),
        "months": months,
        "recent_months": sorted(recent_months),
        "topics": topics,
        "edges": edges,
        "timeline": timeline,
        "routes": KNOWLEDGE_ROUTES,
        "key_papers": top_papers_for(papers, 12),
        "recent_papers": top_papers_for(papers, 12, recent=True),
        "design_policy": {
            "precomputed": [
                "主题地图",
                "主题规模",
                "最近热度",
                "主题共现关系",
                "月度趋势",
                "关键论文候选",
                "主题证据列表",
            ],
            "ai_on_demand": [
                "解释当前地图",
                "生成主题导读",
                "生成阅读路径",
                "比较技术路线",
                "单篇论文研读",
            ],
        },
    }
    ATLAS_CACHE.update({"mtime": mtime, "atlas": atlas})
    return atlas


def topic_focus(topic: str) -> dict[str, Any]:
    atlas = build_atlas()
    papers, _ = load_papers()
    topic_papers = [paper for paper in papers if topic in paper["topics"]]
    if not topic_papers:
        raise KeyError(topic)

    months = {row["month"]: row["topics"].get(topic, 0) for row in atlas["timeline"]}
    neighbor_counts: Counter = Counter()
    for paper in topic_papers:
        for other in paper["topics"]:
            if other != topic:
                neighbor_counts[other] += 1

    profile = profile_for(topic)
    routes = [route for route in KNOWLEDGE_ROUTES if topic in route["topics"]]
    focus = {
        "topic": topic,
        "label": profile["label"],
        "stance": profile["stance"],
        "question": profile["question"],
        "count": len(topic_papers),
        "months": months,
        "neighbors": [
            {
                "topic": name,
                "label": profile_for(name)["label"],
                "count": count,
                "stance": profile_for(name)["stance"],
            }
            for name, count in neighbor_counts.most_common(10)
        ],
        "routes": routes,
        "key_papers": top_papers_for(topic_papers, 12),
        "recent_papers": top_papers_for(topic_papers, 12, recent=True),
        "evidence": top_papers_for(topic_papers, 24),
        "reading_lens": [
            "先看主题问题，再看高分论文，不要从论文列表开始迷路。",
            "优先对比相邻主题，它们通常代表真实系统里的交叉问题。",
            "把论文当作证据：它支持了哪条技术路线，暴露了哪个未解决问题？",
        ],
    }
    return focus


def atlas_context(topic: str = "", topic_b: str = "") -> str:
    atlas = build_atlas()
    compact = {
        "paper_count": atlas["paper_count"],
        "recent_months": atlas["recent_months"],
        "top_topics": [
            {
                "topic": item["id"],
                "label": item["label"],
                "count": item["count"],
                "recent_count": item["recent_count"],
                "stance": item["stance"],
            }
            for item in atlas["topics"][:13]
        ],
        "strong_edges": atlas["edges"][:24],
        "routes": atlas["routes"],
        "key_papers": [
            {
                "title": paper["title"],
                "topics": paper["topics"],
                "score": paper["collection_score"],
                "published_at": paper["published_at"],
            }
            for paper in atlas["key_papers"][:10]
        ],
    }
    if topic:
        compact["topic_focus"] = topic_focus(topic)
    if topic_b:
        compact["compare_topic"] = topic_focus(topic_b)
    return json.dumps(compact, ensure_ascii=False, indent=2)[:12000]


def atlas_prompt_for(mode: str, context: str) -> str:
    no_think = "不要输出思考过程，不要输出 <think> 标签，只输出最终结果。"
    if mode == "topic":
        return (
            "你是 Agent 研究编辑。请根据给定知识库统计，为当前主题写一份中文导读：\n"
            "1. 主题定位\n2. 为什么最近重要\n3. 主要技术路线\n4. 关键争议\n"
            "5. 必读论文阅读顺序\n6. 和相邻主题的关系\n7. 可以做的实验。\n"
            f"只基于给定数据，不要编造论文细节。{no_think}\n\n{context}"
        )
    if mode == "path":
        return (
            "你是研究导师。请基于给定知识库，为一个想系统学习 Agent 技术的人生成阅读路径。\n"
            "输出 4 个阶段：入门地图、核心系统、风险与评估、专题深入。每阶段给出主题、阅读目标、代表论文标题。\n"
            f"保持可执行，避免泛泛而谈。{no_think}\n\n{context}"
        )
    if mode == "compare":
        return (
            "你是技术路线分析师。请比较给定两个 Agent 主题：共同问题、差异、交叉区域、代表论文、工程启发。\n"
            f"只使用给定数据。{no_think}\n\n{context}"
        )
    return (
        "你是 Agent 知识库的研究编辑。请解释这张 Agent 论文 Atlas：\n"
        "1. 当前版图的主轴是什么\n2. 哪些主题是中心节点\n3. 哪些方向最近升温\n"
        "4. 哪些主题之间关系最强\n5. 下一步应该优先精读什么。\n"
        f"只基于给定数据。{no_think}\n\n{context}"
    )


def arxiv_id_from_paper(paper: dict[str, Any]) -> str:
    url = str(paper.get("url") or "")
    match = re.search(r"arxiv\.org/(?:abs|html|pdf)/([0-9]{4}\.[0-9]+)", url)
    return match.group(1) if match else ""


def normalize_space(value: str) -> str:
    return re.sub(r"\s+", " ", value).strip()


def fetch_arxiv(arxiv_id: str) -> dict[str, Any]:
    ensure_cache_dirs()
    cache_path = ARXIV_CACHE / f"{arxiv_id}.json"
    if cache_path.exists():
        return json.loads(cache_path.read_text(encoding="utf-8"))

    query = urllib.parse.urlencode({"id_list": arxiv_id})
    request = urllib.request.Request(
        f"https://export.arxiv.org/api/query?{query}",
        headers={"User-Agent": "agent-kb-paper-browser/0.1"},
    )
    with urllib.request.urlopen(request, timeout=20) as response:
        xml_text = response.read().decode("utf-8", "replace")
    root = ET.fromstring(xml_text)
    entry = root.find("atom:entry", NS)
    if entry is None:
        raise ValueError(f"arXiv entry not found: {arxiv_id}")

    authors = [
        normalize_space(author.findtext("atom:name", default="", namespaces=NS))
        for author in entry.findall("atom:author", NS)
    ]
    categories = [
        category.attrib.get("term", "")
        for category in entry.findall("atom:category", NS)
        if category.attrib.get("term")
    ]
    payload = {
        "arxiv_id": arxiv_id,
        "title": normalize_space(entry.findtext("atom:title", default="", namespaces=NS)),
        "abstract": normalize_space(entry.findtext("atom:summary", default="", namespaces=NS)),
        "authors": authors,
        "published": entry.findtext("atom:published", default="", namespaces=NS)[:10],
        "updated": entry.findtext("atom:updated", default="", namespaces=NS)[:10],
        "categories": categories,
        "url": f"https://arxiv.org/abs/{arxiv_id}",
    }
    cache_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    return payload


def paper_context(paper: dict[str, Any], abstract: dict[str, Any] | None) -> str:
    meta = {
        "title": paper["title"],
        "authors": paper["authors"],
        "year": paper["year"],
        "venue": paper["venue"],
        "url": paper["url"],
        "topics": paper["topics"],
        "collection_score": paper["collection_score"],
        "collection_queries": paper["collection_queries"],
    }
    parts = [f"Metadata:\n{json.dumps(meta, ensure_ascii=False, indent=2)}"]
    if abstract and abstract.get("abstract"):
        parts.append(f"Abstract:\n{abstract['abstract']}")
    else:
        content = (ROOT / paper["path"]).read_text(encoding="utf-8")
        parts.append(f"Local note:\n{content[:4000]}")
    return "\n\n".join(parts)[:7000]


def prompt_for(mode: str, context: str) -> str:
    no_think = "不要输出思考过程，不要输出 <think> 标签，只输出最终结果。"
    if mode == "translate":
        return (
            "你是论文摘要翻译助手。把下面论文摘要准确翻译成中文，保留技术术语，"
            f"不要扩写，不要编造。如果没有摘要，只翻译已有材料。{no_think}\n\n"
            f"{context}"
        )
    if mode == "deep":
        return (
            "你是 Agent 研究员。请基于给定论文材料做中文深度研读笔记，输出：\n"
            "1. 一句话定位\n2. 研究问题\n3. 方法/系统设计\n4. 评估方式\n"
            "5. 对 Agent 工程实践的启发\n6. 可疑点或待验证问题\n7. 推荐标签。\n"
            f"只根据材料回答，信息不足就明确写“不足”。{no_think}\n\n"
            f"{context}"
        )
    return (
        "你是 Agent 论文知识库助手。请基于给定论文材料输出中文摘要，结构如下：\n"
        "1. 一句话结论\n2. 解决的问题\n3. 核心方法\n4. 评估/实验\n"
        "5. 和 Agent 知识库的关系\n6. 推荐优先级：P0/P1/P2。\n"
        f"保持精炼，只根据材料回答，禁止编造。{no_think}\n\n"
        f"{context}"
    )


def clean_model_response(value: str) -> str:
    text = re.sub(r"<think>[\s\S]*?</think>", "", value, flags=re.IGNORECASE).strip()
    return text or value.strip()


def call_ollama(model: str, prompt: str, mode: str) -> dict[str, Any]:
    options = {
        "temperature": 0.2,
        "num_predict": 520,
    }
    if mode == "translate":
        options.update({"num_predict": 700, "temperature": 0.1})
    elif mode == "deep":
        options.update({"num_predict": 900, "temperature": 0.25})
    elif mode in {"atlas", "topic", "path", "compare"}:
        options.update({"num_predict": 950, "temperature": 0.22})
    elif mode == "search_plan":
        options.update({"num_predict": 420, "temperature": 0.1})
    elif mode in {"search_summary", "search_followup", "paper_chat"}:
        options.update({"num_predict": 780, "temperature": 0.18})

    payload = {
        "model": model,
        "prompt": prompt,
        "stream": False,
        "keep_alive": "2m",
        "options": options,
    }
    data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
    request = urllib.request.Request(
        f"{OLLAMA_URL}/api/generate",
        data=data,
        headers={"content-type": "application/json"},
        method="POST",
    )
    started = time.time()
    with urllib.request.urlopen(request, timeout=240) as response:
        result = json.loads(response.read().decode("utf-8"))
    return {
        "model": model,
        "mode": mode,
        "response": clean_model_response(str(result.get("response") or "")),
        "duration_seconds": round(time.time() - started, 2),
        "eval_count": result.get("eval_count"),
        "prompt_eval_count": result.get("prompt_eval_count"),
    }


def ai_cache_path(paper_id: str, mode: str, model: str) -> Path:
    digest = hashlib.sha1(f"{paper_id}:{mode}:{model}".encode("utf-8")).hexdigest()[:18]
    return AI_CACHE / f"{digest}.json"


class PaperBrowserHandler(SimpleHTTPRequestHandler):
    server_version = "AgentPaperBrowser/0.1"

    def log_message(self, format: str, *args: Any) -> None:
        print(f"[paper-web] {self.address_string()} - {format % args}")

    def do_GET(self) -> None:
        parsed = urllib.parse.urlparse(self.path)
        path = parsed.path
        query = urllib.parse.parse_qs(parsed.query)

        if path == "/":
            self.serve_file(STATIC_ROOT / "index.html", "text/html; charset=utf-8")
            return
        if path.startswith("/static/"):
            target = STATIC_ROOT / path.removeprefix("/static/")
            content_type = "text/plain; charset=utf-8"
            if target.suffix == ".css":
                content_type = "text/css; charset=utf-8"
            elif target.suffix == ".js":
                content_type = "application/javascript; charset=utf-8"
            self.serve_file(target, content_type)
            return
        if path == "/api/health":
            self.handle_health()
            return
        if path == "/api/models":
            self.handle_models()
            return
        if path == "/api/topics":
            papers, _ = load_papers()
            write_json(self, {"topics": topic_counts(papers)})
            return
        if path == "/api/atlas":
            write_json(self, build_atlas())
            return
        if path == "/api/topic":
            self.handle_topic(query)
            return
        if path == "/api/papers":
            self.handle_paper_search(query)
            return
        if path == "/api/search/agent":
            self.handle_search_job(query)
            return
        if path == "/api/paper":
            self.handle_paper_detail(query)
            return
        if path == "/api/paper/abstract":
            self.handle_abstract(query)
            return

        write_error(self, HTTPStatus.NOT_FOUND, "not found")

    def do_POST(self) -> None:
        parsed = urllib.parse.urlparse(self.path)
        if parsed.path == "/api/ai":
            self.handle_ai()
            return
        if parsed.path == "/api/atlas/ai":
            self.handle_atlas_ai()
            return
        if parsed.path == "/api/search/agent":
            self.handle_search_agent_start()
            return
        if parsed.path == "/api/search/followup":
            self.handle_search_followup()
            return
        if parsed.path == "/api/paper/chat":
            self.handle_paper_chat()
            return
        write_error(self, HTTPStatus.NOT_FOUND, "not found")

    def serve_file(self, target: Path, content_type: str) -> None:
        try:
            resolved = target.resolve()
            if not str(resolved).startswith(str(STATIC_ROOT.resolve())):
                write_error(self, HTTPStatus.FORBIDDEN, "forbidden")
                return
            data = resolved.read_bytes()
        except FileNotFoundError:
            write_error(self, HTTPStatus.NOT_FOUND, "file not found")
            return
        self.send_response(HTTPStatus.OK)
        self.send_header("content-type", content_type)
        self.send_header("cache-control", "no-store, max-age=0")
        self.send_header("content-length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def handle_health(self) -> None:
        papers, _ = load_papers()
        ollama_ok = False
        try:
            request = urllib.request.Request(f"{OLLAMA_URL}/api/tags")
            with urllib.request.urlopen(request, timeout=3) as response:
                ollama_ok = response.status == 200
        except (urllib.error.URLError, TimeoutError):
            ollama_ok = False
        write_json(
            self,
            {
                "ok": True,
                "paper_count": len(papers),
                "ollama_url": OLLAMA_URL,
                "ollama_ok": ollama_ok,
                "default_models": DEFAULT_MODELS,
            },
        )

    def handle_models(self) -> None:
        try:
            request = urllib.request.Request(f"{OLLAMA_URL}/api/tags")
            with urllib.request.urlopen(request, timeout=5) as response:
                payload = json.loads(response.read().decode("utf-8"))
            models = [
                {
                    "name": item.get("name"),
                    "parameter_size": (item.get("details") or {}).get("parameter_size"),
                    "context_length": (item.get("details") or {}).get("context_length"),
                }
                for item in payload.get("models", [])
            ]
            write_json(self, {"models": models, "default_models": DEFAULT_MODELS})
        except Exception as exc:  # noqa: BLE001
            write_error(self, HTTPStatus.BAD_GATEWAY, f"ollama unavailable: {exc}")

    def handle_paper_search(self, query: dict[str, list[str]]) -> None:
        papers, _ = load_papers()
        q = (query.get("q") or [""])[0].strip().lower()
        topic = (query.get("topic") or [""])[0].strip()
        year = (query.get("year") or [""])[0].strip()
        status = (query.get("status") or [""])[0].strip()
        relevance = (query.get("relevance") or [""])[0].strip()
        sort = (query.get("sort") or ["score"])[0].strip()
        limit = max(1, min(100, int((query.get("limit") or ["40"])[0])))
        offset = max(0, int((query.get("offset") or ["0"])[0]))

        filtered = []
        for paper in papers:
            if q and q not in paper["_search"]:
                continue
            if topic and topic not in paper["topics"]:
                continue
            if year and paper["year"] != year:
                continue
            if status and paper["status"] != status:
                continue
            if relevance and paper["relevance"] != relevance:
                continue
            filtered.append(paper)

        if sort == "date":
            filtered.sort(key=lambda item: (item["published_at"], item["collection_score"]), reverse=True)
        elif sort == "title":
            filtered.sort(key=lambda item: item["title"].lower())
        else:
            filtered.sort(key=lambda item: (item["collection_score"], item["published_at"]), reverse=True)

        page = filtered[offset : offset + limit]
        write_json(
            self,
            {
                "total": len(filtered),
                "offset": offset,
                "limit": limit,
                "items": [paper_public(paper) for paper in page],
            },
        )

    def handle_search_agent_start(self) -> None:
        try:
            payload = read_json_body(self)
            query = str(payload.get("query") or "").strip()
            scope_topic = str(payload.get("topic") or "").strip()
            if not query:
                write_error(self, HTTPStatus.BAD_REQUEST, "query is required")
                return
            if scope_topic and scope_topic not in TOPIC_PROFILES:
                scope_topic = ""
            write_json(self, {"job": new_search_job(query, scope_topic)})
        except Exception as exc:  # noqa: BLE001
            write_error(self, HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))

    def handle_search_job(self, query: dict[str, list[str]]) -> None:
        job_id = (query.get("id") or [""])[0].strip()
        if not job_id:
            write_error(self, HTTPStatus.BAD_REQUEST, "id is required")
            return
        job = get_search_job(job_id)
        if not job:
            write_error(self, HTTPStatus.NOT_FOUND, "search job not found")
            return
        write_json(self, {"job": job})

    def handle_search_followup(self) -> None:
        try:
            payload = read_json_body(self)
            job_id = str(payload.get("job_id") or "").strip()
            question = str(payload.get("question") or "").strip()
            model = str(payload.get("model") or DEFAULT_MODELS["search_followup"])
            if not job_id or not question:
                write_error(self, HTTPStatus.BAD_REQUEST, "job_id and question are required")
                return
            job = get_search_job(job_id)
            if not job or job.get("status") != "done":
                write_error(self, HTTPStatus.NOT_FOUND, "completed search job not found")
                return
            if not OLLAMA_LOCK.acquire(blocking=False):
                write_error(self, HTTPStatus.TOO_MANY_REQUESTS, "ollama is busy; try again later")
                return
            try:
                result = call_ollama(model, search_followup_prompt(job, question), "search_followup")
                write_json(
                    self,
                    {
                        "job_id": job_id,
                        "question": question,
                        "answer": result["response"],
                        "model": result["model"],
                        "duration_seconds": result["duration_seconds"],
                    },
                )
            finally:
                OLLAMA_LOCK.release()
        except Exception as exc:  # noqa: BLE001
            write_error(self, HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))

    def handle_paper_detail(self, query: dict[str, list[str]]) -> None:
        paper_id = (query.get("id") or [""])[0]
        _, by_id = load_papers()
        paper = by_id.get(paper_id)
        if not paper:
            write_error(self, HTTPStatus.NOT_FOUND, "paper not found")
            return
        text = (ROOT / paper["path"]).read_text(encoding="utf-8")
        write_json(self, {"paper": paper_public(paper), "markdown": text})

    def handle_topic(self, query: dict[str, list[str]]) -> None:
        topic = (query.get("topic") or [""])[0].strip()
        if not topic:
            write_error(self, HTTPStatus.BAD_REQUEST, "topic is required")
            return
        try:
            write_json(self, topic_focus(topic))
        except KeyError:
            write_error(self, HTTPStatus.NOT_FOUND, "topic not found")

    def handle_abstract(self, query: dict[str, list[str]]) -> None:
        paper_id = (query.get("id") or [""])[0]
        _, by_id = load_papers()
        paper = by_id.get(paper_id)
        if not paper:
            write_error(self, HTTPStatus.NOT_FOUND, "paper not found")
            return
        arxiv_id = arxiv_id_from_paper(paper)
        if not arxiv_id:
            write_json(self, {"abstract": None, "message": "not an arXiv paper"})
            return
        try:
            write_json(self, {"abstract": fetch_arxiv(arxiv_id)})
        except Exception as exc:  # noqa: BLE001
            write_error(self, HTTPStatus.BAD_GATEWAY, f"arXiv fetch failed: {exc}")

    def handle_paper_chat(self) -> None:
        try:
            payload = read_json_body(self)
            paper_id = str(payload.get("id") or "").strip()
            question = str(payload.get("question") or "").strip()
            history = payload.get("history") or []
            model = str(payload.get("model") or DEFAULT_MODELS["paper_chat"])
            if not paper_id or not question:
                write_error(self, HTTPStatus.BAD_REQUEST, "id and question are required")
                return
            if not isinstance(history, list):
                history = []
            _, by_id = load_papers()
            paper = by_id.get(paper_id)
            if not paper:
                write_error(self, HTTPStatus.NOT_FOUND, "paper not found")
                return
            if not OLLAMA_LOCK.acquire(blocking=False):
                write_error(self, HTTPStatus.TOO_MANY_REQUESTS, "ollama is busy; try again later")
                return
            try:
                abstract = None
                arxiv_id = arxiv_id_from_paper(paper)
                if arxiv_id:
                    try:
                        abstract = fetch_arxiv(arxiv_id)
                    except Exception:
                        abstract = None
                result = call_ollama(model, paper_chat_prompt(paper, abstract, question, history), "paper_chat")
                write_json(
                    self,
                    {
                        "id": paper_id,
                        "question": question,
                        "answer": result["response"],
                        "model": result["model"],
                        "duration_seconds": result["duration_seconds"],
                    },
                )
            finally:
                OLLAMA_LOCK.release()
        except Exception as exc:  # noqa: BLE001
            write_error(self, HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))

    def handle_ai(self) -> None:
        try:
            payload = read_json_body(self)
            paper_id = str(payload.get("id") or "")
            mode = str(payload.get("mode") or "summary")
            refresh = bool(payload.get("refresh"))
            model = str(payload.get("model") or DEFAULT_MODELS.get(mode) or DEFAULT_MODELS["summary"])
            if mode not in {"summary", "translate", "deep"}:
                write_error(self, HTTPStatus.BAD_REQUEST, "invalid ai mode")
                return

            _, by_id = load_papers()
            paper = by_id.get(paper_id)
            if not paper:
                write_error(self, HTTPStatus.NOT_FOUND, "paper not found")
                return

            ensure_cache_dirs()
            cache_path = ai_cache_path(paper_id, mode, model)
            if cache_path.exists() and not refresh:
                result = json.loads(cache_path.read_text(encoding="utf-8"))
                result["cached"] = True
                write_json(self, result)
                return

            if not OLLAMA_LOCK.acquire(blocking=False):
                write_error(self, HTTPStatus.TOO_MANY_REQUESTS, "ollama is busy; try again later")
                return
            try:
                abstract = None
                arxiv_id = arxiv_id_from_paper(paper)
                if arxiv_id:
                    abstract = fetch_arxiv(arxiv_id)
                context = paper_context(paper, abstract)
                result = call_ollama(model, prompt_for(mode, context), mode)
                result.update(
                    {
                        "id": paper_id,
                        "title": paper["title"],
                        "cached": False,
                        "created_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
                    }
                )
                cache_path.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
                write_json(self, result)
            finally:
                OLLAMA_LOCK.release()
        except Exception as exc:  # noqa: BLE001
            write_error(self, HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))

    def handle_atlas_ai(self) -> None:
        try:
            payload = read_json_body(self)
            mode = str(payload.get("mode") or "atlas")
            topic = str(payload.get("topic") or "")
            topic_b = str(payload.get("topic_b") or "")
            refresh = bool(payload.get("refresh"))
            model = str(payload.get("model") or DEFAULT_MODELS.get(mode) or DEFAULT_MODELS["atlas"])
            if mode not in {"atlas", "topic", "path", "compare"}:
                write_error(self, HTTPStatus.BAD_REQUEST, "invalid atlas ai mode")
                return
            if mode == "topic" and not topic:
                write_error(self, HTTPStatus.BAD_REQUEST, "topic is required")
                return
            if mode == "compare" and (not topic or not topic_b):
                write_error(self, HTTPStatus.BAD_REQUEST, "topic and topic_b are required")
                return

            ensure_cache_dirs()
            cache_key = f"atlas:{mode}:{topic}:{topic_b}"
            cache_path = ai_cache_path(cache_key, mode, model)
            if cache_path.exists() and not refresh:
                result = json.loads(cache_path.read_text(encoding="utf-8"))
                result["cached"] = True
                write_json(self, result)
                return

            if not OLLAMA_LOCK.acquire(blocking=False):
                write_error(self, HTTPStatus.TOO_MANY_REQUESTS, "ollama is busy; try again later")
                return
            try:
                context = atlas_context(topic, topic_b)
                result = call_ollama(model, atlas_prompt_for(mode, context), mode)
                result.update(
                    {
                        "scope": cache_key,
                        "cached": False,
                        "created_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
                    }
                )
                cache_path.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
                write_json(self, result)
            finally:
                OLLAMA_LOCK.release()
        except Exception as exc:  # noqa: BLE001
            write_error(self, HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("--host", default=os.environ.get("HOST", "0.0.0.0"))
    parser.add_argument("--port", type=int, default=int(os.environ.get("PORT", "18080")))
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    ensure_cache_dirs()
    load_papers()
    server = ThreadingHTTPServer((args.host, args.port), PaperBrowserHandler)
    print(f"Paper browser: http://{args.host}:{args.port}")
    print(f"Ollama: {OLLAMA_URL}")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nStopping paper browser")
    finally:
        server.server_close()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())