llm-in-text/backend/job_handlers.py

import asyncio
import io
import ipaddress
import json
import os
import re
import socket
import time
import zipfile
from contextlib import suppress
from datetime import datetime
from typing import Any, Callable, Awaitable
from urllib.parse import urlparse

import httpx

import markitdown

from audit_store import get_audit_store
from llm import call_ollama, call_vlm_ocr, stream_ollama_events
from media_utils import extract_audio_wav_bytes, is_video_filename
from prompt import (
    build_completion_prompts,
    build_pro_completion_prompts,
    prepare_prompt_context,
)
from risk_config import load_risk_config
from risk_control import RiskIdentity, estimate_tokens, get_risk_controller
from tts_asr import generate_asr_response, generate_tts_response


IMAGE_MARKDOWN_RE = re.compile(r"!\[[^\]]*]\([^)]+\)")
IMAGE_HTML_RE = re.compile(r"<img\b[^>]*>", re.IGNORECASE)
ALLOWED_CONVERT_EXTENSIONS = {".txt", ".docx", ".pptx", ".pdf"}
SEARXNG_BASE_URL = os.getenv("SEARXNG_BASE_URL", "http://searxng:8080").rstrip("/")
SEARXNG_RESULT_LIMIT = int(os.getenv("SEARXNG_RESULT_LIMIT", "10") or "10")
FIRECRAWL_BASE_URL = os.getenv("FIRECRAWL_BASE_URL", "http://firecrawl:3002").rstrip("/")
FIRECRAWL_API_KEY = os.getenv("FIRECRAWL_API_KEY", "").strip() or ""
WEB_SEARCH_QUERY_COUNT = int(os.getenv("WEB_SEARCH_QUERY_COUNT", "4") or "4")
WEB_SEARCH_SELECTED_URL_LIMIT = int(os.getenv("WEB_SEARCH_SELECTED_URL_LIMIT", "10") or "10")
WEB_SEARCH_CRAWL_CONCURRENCY = max(1, min(6, int(os.getenv("WEB_SEARCH_CRAWL_CONCURRENCY", "3") or "3")))
WEB_SEARCH_CRAWL_TIMEOUT_SECONDS = max(10, min(90, int(os.getenv("WEB_SEARCH_CRAWL_TIMEOUT_SECONDS", "35") or "35")))
_markitdown_instance = None
_risk_config = load_risk_config()


def _get_markitdown():
    global _markitdown_instance
    if _markitdown_instance is None:
        _markitdown_instance = markitdown.MarkItDown()
    return _markitdown_instance


def _convert_url_markdown(url: str) -> dict[str, Any]:
    result = _get_markitdown().convert_url(url)
    markdown = _normalize_multiline_text(str(getattr(result, "markdown", "") or ""))
    title = str(getattr(result, "title", "") or "").strip()
    return {
        "url": url,
        "title": title,
        "markdown": markdown,
    }


def _safe_unlink(path: str | None) -> None:
    if not path:
        return
    with suppress(FileNotFoundError):
        os.unlink(path)


def _sanitize_converted_markdown(text: str) -> str:
    value = (text or "").replace("\r\n", "\n").replace("\r", "\n")
    value = IMAGE_MARKDOWN_RE.sub("", value)
    value = IMAGE_HTML_RE.sub("", value)
    value = re.sub(r"\n{3,}", "\n\n", value)
    return value.strip()


def _normalize_multiline_text(value: str) -> str:
    return (value or "").replace("\r\n", "\n").replace("\r", "\n").strip()


def _looks_like_text(raw_bytes: bytes) -> bool:
    sample = raw_bytes[:8192]
    if not sample or b"\x00" in sample:
        return False
    try:
        text = sample.decode("utf-8")
    except UnicodeDecodeError:
        return False
    if not text.strip():
        return False
    control_count = sum(
        1
        for char in text
        if (ord(char) < 32 and char not in "\t\n\r") or ord(char) == 127
    )
    return control_count / max(len(text), 1) < 0.05


def _infer_convert_suffix(raw_bytes: bytes, filename: str) -> str:
    sample = raw_bytes[:1024 * 1024]
    if sample.startswith(b"%PDF-"):
        return ".pdf"
    if sample.startswith((b"PK\x03\x04", b"PK\x05\x06")):
        try:
            with zipfile.ZipFile(io.BytesIO(raw_bytes)) as archive:
                names = set(archive.namelist())
                if any(name.startswith("ppt/") for name in names):
                    return ".pptx"
                if any(name.startswith("word/") for name in names):
                    return ".docx"
        except Exception:
            pass
    if _looks_like_text(sample):
        return ".txt"
    return ""


def _resolve_url_addresses(url: str) -> list[tuple[Any, ...]]:
    parsed = urlparse((url or "").strip())
    host = (parsed.hostname or "").strip().lower()
    if not host:
        return []
    port = parsed.port or (443 if parsed.scheme == "https" else 80)
    return socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)


def _is_blocked_public_url(url: str) -> bool:
    try:
        parsed = urlparse((url or "").strip())
    except Exception:
        return True
    if parsed.scheme not in {"http", "https"}:
        return True
    host = (parsed.hostname or "").strip().lower()
    if not host:
        return True
    if host in {"localhost", "127.0.0.1", "::1"} or host.endswith((".local", ".localhost")):
        return True
    try:
        ip = ipaddress.ip_address(host)
        return not ip.is_global
    except ValueError:
        pass
    try:
        addresses = _resolve_url_addresses(url)
    except Exception:
        return True
    for info in addresses:
        address = info[4][0]
        try:
            ip = ipaddress.ip_address(address)
        except ValueError:
            continue
        if not ip.is_global:
            return True
    return False


def _strip_code_fence(value: str) -> str:
    text = _normalize_multiline_text(value)
    match = re.match(r"^```(?:json)?\s*([\s\S]*?)\s*```$", text, flags=re.IGNORECASE)
    if match:
        return match.group(1).strip()
    return text


def _extract_json_array(value: str) -> list[Any]:
    text = _strip_code_fence(value)
    try:
        parsed = json.loads(text)
        return parsed if isinstance(parsed, list) else []
    except Exception:
        match = re.search(r"\[[\s\S]*\]", text)
        if not match:
            return []
        try:
            parsed = json.loads(match.group(0))
        except Exception:
            return []
        return parsed if isinstance(parsed, list) else []


def _normalize_search_queries(raw: str) -> list[str]:
    items = _extract_json_array(raw)
    queries: list[str] = []
    if items:
        for item in items:
            text = str(item).strip()
            if text and text not in queries:
                queries.append(text)
    else:
        for line in _strip_code_fence(raw).splitlines():
            text = re.sub(r"^\s*(?:[-*]|\d+[.)])\s*", "", line).strip()
            if text and text not in queries:
                queries.append(text)
    return queries[:WEB_SEARCH_QUERY_COUNT]


def _clean_search_query_text(value: str) -> str:
    text = _normalize_multiline_text(value)
    text = re.sub(r"`{1,3}.*?`{1,3}", " ", text)
    text = re.sub(r"[*_#>\[\]\(\){}|]+", " ", text)
    text = re.sub(r"\s+", " ", text).strip(" -:;,./")
    return text


def _build_fallback_search_queries(context: str, primary_queries: list[str]) -> list[str]:
    queries: list[str] = []

    def add(text: str) -> None:
        cleaned = _clean_search_query_text(text)
        if len(cleaned) < 2 or cleaned in queries:
            return
        queries.append(cleaned[:120])

    for item in primary_queries:
        add(item)

    for line in context.splitlines():
        cleaned = _clean_search_query_text(line)
        if not cleaned:
            continue
        add(cleaned)
        if re.search(r"[A-Za-z]", cleaned):
            add(f"{cleaned} official")
            add(f"{cleaned} github")
        else:
            add(f"{cleaned} 官网")
            add(f"{cleaned} GitHub")

    if context:
        compact = _clean_search_query_text(context.replace("\n", " "))
        if compact:
            add(compact)
            if re.search(r"[A-Za-z]", compact):
                add(f"{compact} official documentation")
            else:
                add(f"{compact} 官方文档")

    return queries[: max(WEB_SEARCH_QUERY_COUNT + 4, 8)]


def _build_no_result_content(queries: list[str], reason: str) -> str:
    lines = [
        "未检索到可用公开结果。",
        "",
        f"原因：{reason}",
    ]
    if queries:
        lines.extend(["", "已尝试的检索词："])
        lines.extend([f"- {query}" for query in queries])
    lines.extend([
        "",
        "可以尝试缩短主题、补充专有名词，或直接给出官网、产品名、项目名、作者名等更具体的线索。",
    ])
    return "\n".join(lines)


def _normalize_selected_urls(raw: str) -> list[str]:
    urls: list[str] = []
    for item in _extract_json_array(raw):
        text = str(item).strip()
        if not text or _is_blocked_public_url(text) or text in urls:
            continue
        urls.append(text)
    return urls[:WEB_SEARCH_SELECTED_URL_LIMIT]


def _summarize_search_results(query: str, results: list[dict[str, Any]]) -> str:
    lines = [f"Query: {query}"]
    for index, item in enumerate(results, start=1):
        lines.append(
            f"{index}. title={item.get('title', '')} url={item.get('url', '')} "
            f"score={item.get('score', '')} date={item.get('published_date', '')} snippet={item.get('snippet', '')}"
        )
    return "\n".join(lines)


async def _searxng_search(query: str, *, limit: int) -> list[dict[str, Any]]:
    async with httpx.AsyncClient(timeout=httpx.Timeout(20.0, connect=10.0)) as client:
        response = await client.get(
            f"{SEARXNG_BASE_URL}/search",
            params={"q": query, "format": "json"},
            headers={
                "Accept": "application/json",
                "User-Agent": "llm-in-text-websearch/1.0",
                "X-Forwarded-For": "127.0.0.1",
                "X-Real-IP": "127.0.0.1",
            },
        )
        response.raise_for_status()
        payload = response.json()

    results: list[dict[str, Any]] = []
    for item in payload.get("results") or []:
        url = str(item.get("url") or item.get("link") or "").strip()
        if not url or await asyncio.to_thread(_is_blocked_public_url, url):
            continue
        results.append({
            "title": str(item.get("title") or "").strip(),
            "url": url,
            "score": item.get("score"),
            "published_date": str(item.get("publishedDate") or item.get("published_date") or item.get("published") or "").strip(),
            "snippet": _normalize_multiline_text(str(item.get("content") or item.get("snippet") or "")),
        })
        if len(results) >= limit:
            break
    return results


async def _firecrawl_scrape(url: str) -> dict[str, Any]:
    headers = {"Content-Type": "application/json"}
    if FIRECRAWL_API_KEY:
        headers["Authorization"] = f"Bearer {FIRECRAWL_API_KEY}"
        headers["X-Api-Key"] = FIRECRAWL_API_KEY

    payload = None
    async with httpx.AsyncClient(timeout=httpx.Timeout(20.0, connect=5.0, read=20.0)) as client:
        try:
            response = await client.post(
                f"{FIRECRAWL_BASE_URL}/v1/scrape",
                json={"url": url, "formats": ["markdown"]},
                headers=headers,
            )
            response.raise_for_status()
            payload = response.json()
        except Exception:
            payload = None
    if payload is None:
        return await asyncio.to_thread(_convert_url_markdown, url)

    data = payload.get("data") or {}
    markdown = _normalize_multiline_text(str(data.get("markdown") or data.get("content") or ""))
    metadata = data.get("metadata") or {}
    return {
        "url": url,
        "title": str(metadata.get("title") or "").strip(),
        "markdown": markdown,
    }


def sanitize_inline_completion_content(text: str, prefill: str = "") -> str:
    value = (text or "").strip()
    if not value:
        return ""

    fim_middle = value.rfind("<|fim_middle|>")
    if fim_middle >= 0:
        value = value[fim_middle + len("<|fim_middle|>") :]

    end_index = value.find("<|end|>")
    if end_index >= 0:
        value = value[:end_index]

    quoted = re.findall(r'"([^"]+)"', value)
    if quoted:
        value = quoted[-1]

    marker_index = max(value.rfind("|fim_middle|>"), value.rfind("<|start|>assistant"))
    if marker_index >= 0:
        tail = value.split(">")[-1]
        if tail:
            value = tail

    value = value.strip()
    if prefill and value.startswith(prefill):
        value = value[len(prefill) :]

    return value.strip()


def _payload_identity(payload: dict[str, Any]) -> RiskIdentity:
    risk = payload.get("risk") or {}
    return RiskIdentity(
        request_id=risk.get("request_id") or payload["request_id"],
        session_hash=risk.get("session_hash", ""),
        ip_hash=risk.get("ip_hash", ""),
        route=payload.get("route", payload.get("job_type", payload.get("request_id", ""))),
        method="POST",
    )


async def _enter_llm_execution(payload: dict[str, Any], emit: Callable[[str, dict[str, Any]], Awaitable[None]]) -> tuple[RiskIdentity, dict[str, Any], list[str]]:
    risk = payload.get("risk") or {}
    identity = _payload_identity(payload)
    delay_ms = int(risk.get("delay_ms", 0) or 0)
    policy = risk.get("policy") or {}
    if delay_ms > 0:
        await emit("resource", {"phase": "delay", "delay_ms": delay_ms})
        await asyncio.sleep(delay_ms / 1000.0)
    controller = get_risk_controller(_risk_config)
    lock_keys = await controller.acquire_execution_slot(identity, model=policy.get("model", ""))
    return identity, risk, lock_keys


async def _exit_llm_execution(
    payload: dict[str, Any],
    identity: RiskIdentity,
    risk: dict[str, Any],
    lock_keys: list[str],
    *,
    status: str,
    actual_output_text: str = "",
    error_code: str = "",
    audit_metadata: dict[str, Any] | None = None,
) -> None:
    policy = (risk.get("policy") or {})
    controller = get_risk_controller(_risk_config)
    await controller.release_execution_slot(identity, lock_keys, model=policy.get("model", ""))
    await controller.record_model_result(model=policy.get("model", ""), success=(status == "completed"))
    store = get_audit_store(os.getenv("DATABASE_URL", "").strip() or None)
    estimated_input_tokens = int(risk.get("estimated_input_tokens", 0) or 0)
    profile = policy.get("profile", "completion")
    pricing_out = {
        "completion": _risk_config.completion_output_cost_per_1k,
        "pro": _risk_config.pro_output_cost_per_1k,
        "vision": _risk_config.vision_output_cost_per_1k,
    }.get(profile, _risk_config.completion_output_cost_per_1k)
    actual_output_tokens = estimate_tokens(actual_output_text)
    extra_metadata = dict(audit_metadata or {})
    if profile == "speech_tts":
        actual_cost = round(
            (int(extra_metadata.get("text_chars", 0) or 0) / 1000.0) * _risk_config.speech_tts_input_cost_per_1k_chars
            + (int(extra_metadata.get("duration_ms", 0) or 0) / 60000.0) * _risk_config.speech_tts_output_cost_per_minute_audio,
            8,
        )
    elif profile == "speech_asr":
        actual_cost = round(
            (int(extra_metadata.get("audio_bytes", 0) or 0) / (1024.0 * 1024.0)) * _risk_config.speech_asr_input_cost_per_mb,
            8,
        )
    else:
        actual_cost = round((estimated_input_tokens / 1000.0) * {
            "completion": _risk_config.completion_input_cost_per_1k,
            "pro": _risk_config.pro_input_cost_per_1k,
            "vision": _risk_config.vision_input_cost_per_1k,
        }.get(profile, _risk_config.completion_input_cost_per_1k) + (actual_output_tokens / 1000.0) * pricing_out, 8)
    job_context = payload.get("job_context") or {}
    now_ms = int(time.time() * 1000)
    started_at = int(job_context.get("started_at", 0) or 0)
    created_at = int(job_context.get("created_at", 0) or 0)
    queue_ms = int(job_context.get("queue_ms", 0) or 0)
    run_ms = int(job_context.get("run_ms", 0) or 0)
    total_ms = int(job_context.get("total_ms", 0) or 0)
    if not run_ms and started_at:
        run_ms = max(0, now_ms - started_at)
    if not total_ms:
        total_ms = max(0, now_ms - created_at) if created_at else run_ms
    await asyncio.to_thread(
        store.record_llm_call,
        {
            "request_id": payload["request_id"],
            "session_hash": identity.session_hash,
            "ip_hash": identity.ip_hash,
            "job_type": policy.get("job_type", ""),
            "model": policy.get("model", ""),
            "estimated_input_tokens": estimated_input_tokens,
            "max_output_tokens": int(policy.get("max_output_tokens", 0) or 0),
            "estimated_cost": float(risk.get("estimated_cost", 0.0) or 0.0),
            "actual_output_chars": len(actual_output_text or ""),
            "actual_cost": actual_cost,
            "status": status,
            "error_code": error_code,
            "queue_ms": queue_ms,
            "run_ms": run_ms,
            "total_ms": total_ms,
            "metadata": {"profile": profile, **extra_metadata},
        },
    )


async def completion_handler(
    payload: dict[str, Any],
    emit: Callable[[str, dict[str, Any]], Awaitable[None]],
    is_cancelled: Callable[[], bool],
) -> dict[str, Any]:
    req = payload["request"]
    identity, risk, lock_keys = await _enter_llm_execution(payload, emit)
    system_prompt, user_prompt, prefill = build_completion_prompts(
        req["prefix"],
        req["suffix"],
        req.get("languageId", "markdown"),
        location=payload.get("location", ""),
        thinking_level=req.get("model_thinking", "low"),
        preferences=req.get("user_preferences"),
    )
    policy = risk.get("policy") or {}
    try:
        result = await call_ollama(
            user_prompt,
            system_prompt=system_prompt,
            tag=f'{payload["request_id"][:8]}-completion',
            temperature=float(policy.get("temperature", req.get("temperature", 0.7))),
            thinking=policy.get("thinking"),
            model=policy.get("model"),
            prefill=prefill or None,
            max_output_tokens=int(policy.get("max_output_tokens", 0) or 0),
        )
        content = sanitize_inline_completion_content(result.get("content") or "", prefill=prefill or "")
        if is_cancelled():
            raise asyncio.CancelledError()
        await emit("result", {"content": content})
        await _exit_llm_execution(payload, identity, risk, lock_keys, status="completed", actual_output_text=content)
        return {"content": content, "request_id": payload["request_id"]}
    except asyncio.CancelledError:
        await _exit_llm_execution(payload, identity, risk, lock_keys, status="cancelled", error_code="cancelled")
        raise
    except Exception:
        await _exit_llm_execution(payload, identity, risk, lock_keys, status="failed", error_code="llm_failed")
        raise


async def pro_completion_handler(
    payload: dict[str, Any],
    emit: Callable[[str, dict[str, Any]], Awaitable[None]],
    is_cancelled: Callable[[], bool],
) -> dict[str, Any]:
    req = payload["request"]
    identity, risk, lock_keys = await _enter_llm_execution(payload, emit)
    system_prompt, user_prompt = build_pro_completion_prompts(
        prefix=req["prefix"],
        suffix=req["suffix"],
        instruction=req.get("instruction", ""),
        language_id=req.get("languageId", "markdown"),
        location=payload.get("location", ""),
        pro_thinking_level=req.get("pro_thinking", "medium"),
        preferences=req.get("user_preferences"),
    )
    chunks: list[str] = []
    policy = risk.get("policy") or {}
    try:
        async for event_type, delta in stream_ollama_events(
            user_prompt,
            system_prompt=system_prompt,
            tag=f'{payload["request_id"][:8]}-pro',
            temperature=float(policy.get("temperature", 0.7)),
            thinking=policy.get("thinking"),
            model=policy.get("model"),
            enable_thinking=True,
            max_output_tokens=int(policy.get("max_output_tokens", 0) or 0),
        ):
            if is_cancelled():
                raise asyncio.CancelledError()
            if event_type == "thinking":
                await emit("progress", {"phase": "thinking"})
                continue
            if delta:
                chunks.append(delta)
                await emit("result", {"delta": delta})
        content = "".join(chunks)
        await _exit_llm_execution(payload, identity, risk, lock_keys, status="completed", actual_output_text=content)
        return {"content": content, "request_id": payload["request_id"]}
    except asyncio.CancelledError:
        await _exit_llm_execution(payload, identity, risk, lock_keys, status="cancelled", error_code="cancelled")
        raise
    except Exception:
        await _exit_llm_execution(payload, identity, risk, lock_keys, status="failed", error_code="llm_failed")
        raise


async def web_search_handler(
    payload: dict[str, Any],
    emit: Callable[[str, dict[str, Any]], Awaitable[None]],
    is_cancelled: Callable[[], bool],
) -> dict[str, Any]:
    req = payload["request"]
    identity, risk, lock_keys = await _enter_llm_execution(payload, emit)
    policy = risk.get("policy") or {}
    prefix = _normalize_multiline_text(req.get("prefix", ""))
    suffix = _normalize_multiline_text(req.get("suffix", ""))
    context = "\n\n".join(part for part in [prefix, suffix] if part).strip()

    try:
        await emit("progress", {"phase": "keywords", "message": "正在生成搜索关键词"})
        keyword_prompt = (
            "你是联网研究助手。请根据下面的上下文，生成 3 到 5 个适合在搜索引擎中直接使用的检索关键词或短句。\n"
            "要求：\n"
            "- 只返回 JSON 数组字符串\n"
            "- 每个元素是一个简洁检索词\n"
            "- 不要解释，不要 Markdown\n\n"
            f"上下文：\n{context}"
        )
        keyword_result = await call_ollama(
            keyword_prompt,
            system_prompt="Return only a JSON array of search queries.",
            tag=f'{payload["request_id"][:8]}-webq',
            temperature=float(policy.get("temperature", 0.4)),
            thinking=policy.get("thinking"),
            model=policy.get("model"),
            max_output_tokens=min(int(policy.get("max_output_tokens", 0) or 1024), 1024),
        )
        queries = _normalize_search_queries(keyword_result.get("content") or "")
        if not queries:
            queries = [_normalize_multiline_text(prefix or suffix)[:120] or "general research query"]

        if is_cancelled():
            raise asyncio.CancelledError()

        await emit("progress", {"phase": "searching", "message": "正在通过 SearXNG 搜索"})
        search_sections: list[str] = []
        search_candidates: list[dict[str, Any]] = []
        attempted_queries: list[str] = []
        for query in queries:
            if is_cancelled():
                raise asyncio.CancelledError()
            attempted_queries.append(query)
            results = await _searxng_search(query, limit=SEARXNG_RESULT_LIMIT)
            if not results:
                continue
            search_sections.append(_summarize_search_results(query, results))
            search_candidates.extend(results)

        if not search_candidates:
            fallback_queries = _build_fallback_search_queries(context, queries)
            retry_queries = [query for query in fallback_queries if query not in attempted_queries]
            if retry_queries:
                await emit("progress", {"phase": "searching", "message": "搜索结果较少，正在尝试更宽泛的检索词"})
            for query in retry_queries:
                if is_cancelled():
                    raise asyncio.CancelledError()
                attempted_queries.append(query)
                results = await _searxng_search(query, limit=SEARXNG_RESULT_LIMIT)
                if not results:
                    continue
                search_sections.append(_summarize_search_results(query, results))
                search_candidates.extend(results)

        if not search_candidates:
            content = _build_no_result_content(attempted_queries, "SearXNG 未返回可用结果")
            created_at = str(req.get("created_at") or payload.get("created_at") or "").strip() or datetime.now().isoformat()
            await _exit_llm_execution(payload, identity, risk, lock_keys, status="completed", actual_output_text=content)
            return {"content": content, "request_id": payload["request_id"], "created_at": created_at}

        deduped_candidates: list[dict[str, Any]] = []
        seen_urls: set[str] = set()
        for item in search_candidates:
            url = item["url"]
            if url in seen_urls:
                continue
            seen_urls.add(url)
            deduped_candidates.append(item)

        await emit("progress", {"phase": "selecting_urls", "message": "正在筛选可信网址"})
        search_results_text = "\n\n".join(search_sections)
        selection_prompt = (
            "你是研究检索筛选器。下面是多组搜索结果，请从中挑选 5 到 20 个最可信、最相关、最值得进一步抓取的 URL。\n"
            "优先选择：官方文档、权威机构、原始来源、信息完整且日期清晰的页面。\n"
            "只返回 JSON 数组，元素必须是 URL 字符串。\n\n"
            f"原始上下文：\n{context}\n\n"
            f"搜索结果：\n{search_results_text}"
        )
        selection_result = await call_ollama(
            selection_prompt,
            system_prompt="Return only a JSON array of selected URLs.",
            tag=f'{payload["request_id"][:8]}-webu',
            temperature=0.2,
            thinking=policy.get("thinking"),
            model=policy.get("model"),
            max_output_tokens=min(int(policy.get("max_output_tokens", 0) or 1024), 1024),
        )
        selected_urls = _normalize_selected_urls(selection_result.get("content") or "")
        if not selected_urls:
            selected_urls = [item["url"] for item in deduped_candidates[:WEB_SEARCH_SELECTED_URL_LIMIT]]

        if is_cancelled():
            raise asyncio.CancelledError()

        await emit("progress", {"phase": "crawling", "message": "正在抓取网页内容"})
        selected_url_set = set(selected_urls)
        selected_candidates = [item for item in deduped_candidates if item["url"] in selected_url_set][:WEB_SEARCH_SELECTED_URL_LIMIT]
        crawl_sem = asyncio.Semaphore(WEB_SEARCH_CRAWL_CONCURRENCY)

        async def _crawl_candidate(item: dict[str, Any]) -> dict[str, Any] | None:
            if is_cancelled():
                raise asyncio.CancelledError()
            async with crawl_sem:
                try:
                    scraped = await asyncio.wait_for(
                        _firecrawl_scrape(item["url"]),
                        timeout=WEB_SEARCH_CRAWL_TIMEOUT_SECONDS,
                    )
                except Exception:
                    return None
            if not scraped.get("markdown"):
                return None
            return {
                "url": item["url"],
                "title": item.get("title") or scraped.get("title") or "",
                "score": item.get("score"),
                "published_date": item.get("published_date") or "",
                "content": scraped["markdown"],
            }

        crawl_results = await asyncio.gather(*[_crawl_candidate(item) for item in selected_candidates])
        crawled_pages = [page for page in crawl_results if page]

        if not crawled_pages:
            content = _build_no_result_content(selected_urls, "已检索到候选网页，但未抓取到可用正文")
            created_at = str(req.get("created_at") or payload.get("created_at") or "").strip() or datetime.now().isoformat()
            await _exit_llm_execution(payload, identity, risk, lock_keys, status="completed", actual_output_text=content)
            return {"content": content, "request_id": payload["request_id"], "created_at": created_at}

        await emit("progress", {"phase": "synthesizing", "message": "正在整理搜索结果"})
        page_sections = []
        for index, page in enumerate(crawled_pages, start=1):
            page_sections.append(
                f"[Source {index}]\n"
                f"URL: {page['url']}\n"
                f"Title: {page.get('title', '')}\n"
                f"Score: {page.get('score', '')}\n"
                f"Date: {page.get('published_date', '')}\n"
                f"Content:\n{page['content'][:12000]}"
            )
        crawled_text = "\n\n".join(page_sections)

        synthesis_prompt = (
            "你是研究写作助手。请根据原始上下文和抓取到的网页内容，写出一篇长篇、结构完整、信息密集的 Markdown 正文。\n"
            "要求：\n"
            "- 不要写标题\n"
            "- 不要写引用编号、来源表或 URL 列表\n"
            "- 直接输出最终正文\n"
            "- 如果信息存在不确定性，用审慎措辞表达\n\n"
            f"原始上下文：\n{context}\n\n"
            f"抓取内容：\n{crawled_text}"
        )

        # 流式合成 — 逐 delta emit，前端可实时渲染
        from llm import stream_ollama_events

        accumulated: list[str] = []
        async for event_type, text in stream_ollama_events(
            synthesis_prompt,
            system_prompt="Return only the final markdown body with no title and no citations list.",
            tag=f'{payload["request_id"][:8]}-webf',
            temperature=float(policy.get("temperature", 0.4)),
            thinking=policy.get("thinking"),
            model=policy.get("model"),
            max_output_tokens=int(policy.get("max_output_tokens", 0) or 0),
        ):
            if is_cancelled():
                raise asyncio.CancelledError()
            # 只推送 content delta，不展示 thinking
            if event_type == "content":
                accumulated.append(text)
                await emit("delta", {"text": text})

        content = _normalize_multiline_text("".join(accumulated))
        if not content:
            raise RuntimeError("联网搜索生成了空结果")
        created_at = str(req.get("created_at") or payload.get("created_at") or "").strip() or datetime.now().isoformat()
        await _exit_llm_execution(payload, identity, risk, lock_keys, status="completed", actual_output_text=content)
        return {"content": content, "request_id": payload["request_id"], "created_at": created_at}
    except asyncio.CancelledError:
        await _exit_llm_execution(payload, identity, risk, lock_keys, status="cancelled", error_code="cancelled")
        raise
    except Exception:
        await _exit_llm_execution(payload, identity, risk, lock_keys, status="failed", error_code="web_search_failed")
        raise


async def compress_handler(
    payload: dict[str, Any],
    emit: Callable[[str, dict[str, Any]], Awaitable[None]],
    is_cancelled: Callable[[], bool],
) -> dict[str, Any]:
    content = payload["content"]
    doc_type = payload.get("docType", "txt")
    identity, risk, lock_keys = await _enter_llm_execution(payload, emit)
    system_prompt = (
        f"你是一个专业的文档摘要助手。请将以下 {doc_type} 类型文档内容进行精简压缩，"
        "保留核心信息和关键要点，去除冗余和啰嗦的表述。"
        "请直接输出压缩后的内容，不要添加任何解释性文字。"
    )
    policy = risk.get("policy") or {}
    try:
        result = await call_ollama(
            content,
            system_prompt=system_prompt,
            tag=f'{payload["request_id"][:8]}-compress',
            model=policy.get("model"),
            temperature=float(policy.get("temperature", 0.2)),
            thinking=policy.get("thinking"),
            max_output_tokens=int(policy.get("max_output_tokens", 0) or 0),
        )
        if is_cancelled():
            raise asyncio.CancelledError()
        compressed = result.get("content") or ""
        await emit("result", {"content": compressed})
        await _exit_llm_execution(payload, identity, risk, lock_keys, status="completed", actual_output_text=compressed)
        return {"content": compressed, "request_id": payload["request_id"]}
    except asyncio.CancelledError:
        await _exit_llm_execution(payload, identity, risk, lock_keys, status="cancelled", error_code="cancelled")
        raise
    except Exception:
        await _exit_llm_execution(payload, identity, risk, lock_keys, status="failed", error_code="llm_failed")
        raise


async def ocr_handler(
    payload: dict[str, Any],
    emit: Callable[[str, dict[str, Any]], Awaitable[None]],
    is_cancelled: Callable[[], bool],
) -> dict[str, Any]:
    path = payload["input_path"]
    identity, risk, lock_keys = await _enter_llm_execution(payload, emit)
    try:
        filename = payload.get("filename", "image.jpg")
        language = payload.get("language", "auto")
        media_type = payload.get("media_type", "image")
        mime_type = payload.get("mime_type", "") or ""

        with open(path, "rb") as handle:
            media_bytes = handle.read()

        await emit("progress", {"phase": "ocr", "media_type": media_type})
        ocr_text = await call_vlm_ocr(
            media_bytes,
            language,
            mime_type=mime_type or "application/octet-stream",
            media_type=media_type,
        )
        if is_cancelled():
            raise asyncio.CancelledError()

        result = {
            "text": ocr_text,
            "ocr_text": ocr_text,
            "filename": filename,
            "media_type": media_type,
        }

        if media_type == "video" or is_video_filename(filename, mime_type):
            asr_text = ""
            try:
                await emit("progress", {"phase": "asr", "media_type": media_type})
                audio_bytes = await asyncio.to_thread(extract_audio_wav_bytes, path)
                asr_response = await generate_asr_response(audio_bytes, language)
                asr_text = getattr(asr_response, "text", "") or ""
            except Exception as exc:
                raise RuntimeError(f"音频解析失败: {exc}") from exc
            if ocr_text.strip() or asr_text.strip():
                text_parts = []
                if ocr_text.strip():
                    text_parts.append(f"## 视频画面 OCR\n\n{ocr_text.strip()}")
                if asr_text.strip():
                    text_parts.append(f"## 视频音频 ASR\n\n{asr_text.strip()}")
                result["text"] = "\n\n".join(text_parts)
            result["asr_text"] = asr_text

        await emit("result", result)
        await _exit_llm_execution(
            payload,
            identity,
            risk,
            lock_keys,
            status="completed",
            actual_output_text=result["text"],
        )
        return result
    except asyncio.CancelledError:
        await _exit_llm_execution(payload, identity, risk, lock_keys, status="cancelled", error_code="cancelled")
        raise
    except Exception:
        await _exit_llm_execution(payload, identity, risk, lock_keys, status="failed", error_code="ocr_failed")
        raise
    finally:
        _safe_unlink(path)


async def convert_handler(
    payload: dict[str, Any],
    emit: Callable[[str, dict[str, Any]], Awaitable[None]],
    is_cancelled: Callable[[], bool],
) -> dict[str, Any]:
    path = payload["input_path"]
    filename = payload.get("filename", "document")
    try:
        temp_ext = os.path.splitext(path)[1].lower()
    except Exception:
        temp_ext = ""
    if temp_ext not in ALLOWED_CONVERT_EXTENSIONS:
        _safe_unlink(path)
        raise ValueError("仅支持 txt、docx、pptx、pdf 格式")
    try:
        if temp_ext == ".txt":
            with open(path, "rb") as handle:
                markdown = _sanitize_converted_markdown(handle.read().decode("utf-8", errors="ignore"))
        else:
            md = _get_markitdown()
            result = await asyncio.to_thread(md.convert, path)
            markdown = _sanitize_converted_markdown(result.text_content)
        if is_cancelled():
            raise asyncio.CancelledError()
        await emit("result", {"markdown": markdown})
        return {"markdown": markdown, "filename": filename}
    finally:
        _safe_unlink(path)


async def tts_handler(
    payload: dict[str, Any],
    emit: Callable[[str, dict[str, Any]], Awaitable[None]],
    is_cancelled: Callable[[], bool],
) -> dict[str, Any]:
    text = str(payload.get("text", "") or "").strip()
    if not text:
        raise ValueError("TTS 文本为空")
    identity, risk, lock_keys = await _enter_llm_execution(payload, emit)
    try:
        response = await generate_tts_response(
            text=text,
            instruct=str(payload.get("instruct", "") or ""),
            speaker=str(payload.get("speaker", "Vivian") or "Vivian"),
            output_format=str(payload.get("format", "wav") or "wav"),
        )
        if is_cancelled():
            raise asyncio.CancelledError()

        result = dict(response)
        await emit("result", result)
        await _exit_llm_execution(
            payload,
            identity,
            risk,
            lock_keys,
            status="completed",
            audit_metadata={
                "speaker": result.get("speaker", ""),
                "format": result.get("format", ""),
                "duration_ms": int(result.get("duration_ms", 0) or 0),
                "audio_bytes": int(result.get("audio_bytes", 0) or 0),
                "text_chars": int(result.get("text_chars", len(text)) or len(text)),
                "request_ms": int(result.get("request_ms", 0) or 0),
                "upstream_request_id": result.get("upstream_request_id", ""),
            },
        )
        return result
    except asyncio.CancelledError:
        await _exit_llm_execution(payload, identity, risk, lock_keys, status="cancelled", error_code="cancelled")
        raise
    except Exception:
        await _exit_llm_execution(payload, identity, risk, lock_keys, status="failed", error_code="tts_failed")
        raise


async def asr_handler(
    payload: dict[str, Any],
    emit: Callable[[str, dict[str, Any]], Awaitable[None]],
    is_cancelled: Callable[[], bool],
) -> dict[str, Any]:
    path = payload["input_path"]
    identity, risk, lock_keys = await _enter_llm_execution(payload, emit)
    try:
        with open(path, "rb") as handle:
            audio_bytes = handle.read()
        response = await generate_asr_response(audio_bytes, payload.get("language", "zh-CN"))
        if is_cancelled():
            raise asyncio.CancelledError()
        result = dict(response)
        await emit("result", result)
        await _exit_llm_execution(
            payload,
            identity,
            risk,
            lock_keys,
            status="completed",
            actual_output_text=result.get("text", "") or "",
            audit_metadata={
                "language": result.get("language", ""),
                "audio_bytes": int(result.get("audio_bytes", len(audio_bytes)) or len(audio_bytes)),
                "request_ms": int(result.get("request_ms", 0) or 0),
                "upstream_request_id": result.get("upstream_request_id", ""),
            },
        )
        return result
    except asyncio.CancelledError:
        await _exit_llm_execution(payload, identity, risk, lock_keys, status="cancelled", error_code="cancelled")
        raise
    except Exception:
        await _exit_llm_execution(payload, identity, risk, lock_keys, status="failed", error_code="asr_failed")
        raise
    finally:
        _safe_unlink(path)