"""
api.v1.knowledge — Knowledge Base management endpoints.

GET    /api/v1/knowledge                          List all KBs
POST   /api/v1/knowledge                          Create KB (name, root_dir)
DELETE /api/v1/knowledge/{kb_id}                  Delete KB
GET    /api/v1/knowledge/{kb_id}/files            List root_dir file tree
POST   /api/v1/knowledge/{kb_id}/files            Add files to KB
DELETE /api/v1/knowledge/{kb_id}/files/{file_id}  Remove file from KB
POST   /api/v1/knowledge/{kb_id}/index            Trigger index build job
POST   /api/v1/knowledge/{kb_id}/wiki             Trigger wiki compile job
PUT    /api/v1/knowledge/{kb_id}/wiki/pause       Pause/resume wiki compile
GET    /api/v1/knowledge/{kb_id}/wiki/tree        Get wiki file tree
GET    /api/v1/knowledge/{kb_id}/wiki/{page_path} Get wiki page content
POST   /api/v1/knowledge/{kb_id}/search           Search KB
GET    /api/v1/knowledge/{kb_id}/search/stream    SSE streaming QA
GET    /api/v1/knowledge/{kb_id}/jobs             List jobs for KB
GET    /api/v1/knowledge/{kb_id}/jobs/{job_id}    Get job status + log
"""
from __future__ import annotations
import asyncio
import json
import os
import subprocess
import sys
import tempfile
import time
import threading
from pathlib import Path
from typing import Any, Dict, List, Optional, Set

from fastapi import APIRouter, Depends, HTTPException, Query
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field

from agentpaas.api.deps import get_tenant, get_database
from agentpaas.api.errors import api_error
from agentpaas.api.middleware.auth import TenantContext
from agentpaas.db.models import Database, gen_id, now_utc

router = APIRouter(prefix="/knowledge", tags=["knowledge"])

# ─── Supported file extensions ───
SUPPORTED_EXTS = {".pdf", ".txt", ".docx", ".md", ".html", ".htm", ".csv"}


# ─── Pydantic models ───

class CreateKBRequest(BaseModel):
    name: str = Field(..., max_length=256)
    root_dir: str = Field(..., description="Absolute path to knowledge base directory")
    description: str = Field(default="", max_length=2048)


class AddFilesRequest(BaseModel):
    file_paths: List[str] = Field(..., description="Absolute paths of files to add")


class IndexBuildRequest(BaseModel):
    index_type: str = Field(default="bm25", description="bm25 | vector | graph | all")
    mode: str = Field(default="full", description="full | incremental")
    chunk_strategy: str = Field(default="token", description="token | contextual | page")


class WikiBuildRequest(BaseModel):
    mode: str = Field(default="incremental", description="incremental | rebuild")
    concurrency: int = Field(default=2, ge=1, le=8)


class SearchRequest(BaseModel):
    query: str = Field(..., max_length=4096)
    mode: str = Field(default="lambda", description="lambda | bm25 | vector | wiki | fusion | pageindex")
    top_k: int = Field(default=5, ge=1, le=20)
    weights: Optional[Dict[str, float]] = None  # bm25/vector/graph/wiki overrides


class UpdateIgnoreDirsRequest(BaseModel):
    ignored_dirs: List[str] = Field(default_factory=list, description="List of relative dir paths to ignore")


class WikiScheduleRequest(BaseModel):
    enabled: bool = True
    interval: str = Field(default="weekly", description="daily | weekly")


# ─── Helper: walk root_dir ───

def _walk_dir(
    root_dir: str,
    added_paths: set,
    ignored_dirs: Optional[Set[str]] = None,
    subpath: str = "",
    name_filter: str = "",
    limit: int = 500,
) -> List[Dict]:
    """
    Recursively list files under root_dir (or subpath within root_dir).
    - ignored_dirs: set of relative paths to skip
    - subpath: only scan this subdirectory (relative to root_dir)
    - name_filter: case-insensitive filename substring filter
    - limit: max results (default 500, was 2000)
    """
    ignored = ignored_dirs or set()
    items: List[Dict] = []
    root = Path(root_dir)
    scan_root = (root / subpath) if subpath else root
    if not scan_root.is_dir():
        return items

    name_lower = name_filter.lower() if name_filter else ""

    for dirpath, dirnames, filenames in os.walk(str(scan_root), topdown=True):
        rel_dir = Path(dirpath).relative_to(root)
        parts = rel_dir.parts  # () for root itself

        # If any ancestor or the current dir matches the ignore list, skip entire subtree
        if parts:
            ignored_hit = any(
                parts[i] in ignored or "/".join(parts[:i + 1]) in ignored
                for i in range(len(parts))
            )
            if ignored_hit:
                dirnames.clear()
                continue

        # Prune child dirs that match ignore list
        dirnames[:] = sorted(
            d for d in dirnames
            if d not in ignored and (
                "/".join((*parts, d)) if parts else d
            ) not in ignored
        )

        for fname in sorted(filenames):
            if name_lower and name_lower not in fname.lower():
                continue
            p = Path(dirpath) / fname
            if p.suffix.lower() not in SUPPORTED_EXTS:
                continue
            rel_file = str(p.relative_to(root))
            try:
                stat = p.stat()
            except OSError:
                continue
            items.append({
                "path": str(p),
                "relative_path": rel_file,
                "name": fname,
                "type": p.suffix.lower().lstrip("."),
                "size": stat.st_size,
                "modified": stat.st_mtime,
                "added": str(p) in added_paths,
            })
            if len(items) >= limit:
                return items

    return items


def _list_dir_shallow(root_dir: str, parent_rel: str, ignored: Set[str]) -> List[Dict]:
    """
    List ONE level of subdirectories under parent_rel (relative to root_dir).
    Returns nodes with empty children=[]; has_children indicates whether the
    node has any subdirectory children so the frontend can show an expand arrow.
    This replaces the full-depth _list_dir_tree to support lazy loading.
    """
    root = Path(root_dir)
    scan_path = (root / parent_rel) if parent_rel else root
    result = []
    try:
        for child in sorted(scan_path.iterdir()):
            if not child.is_dir():
                continue
            child_rel = f"{parent_rel}/{child.name}" if parent_rel else child.name
            self_ignored = child.name in ignored or child_rel in ignored
            # Peek one level to know if there are subdirs (cheap: stops at first hit)
            has_children = False
            try:
                for gc in child.iterdir():
                    if gc.is_dir():
                        has_children = True
                        break
            except OSError:
                pass
            result.append({
                "name": child.name,
                "relative_path": child_rel,
                "ignored": self_ignored,
                "has_children": has_children,
                "children": [],      # populated lazily by the frontend
            })
    except OSError:
        pass
    return result


# ─── Helper: run index build in background ───

def _write_instance_yml(kb_root: str) -> str:
    """Write a temporary instance.yml that pins all config paths to kb_root.

    config.py loads agent-config.yml first (which may contain hardcoded remote
    paths like /home/67/knowledge/...) and then overlays instance.yml via deep
    merge.  By setting INSTANCE_CONFIG to this temp file we override those paths
    with the actual local kb_root before any script reads cfg.base_dir.

    Returns the temp file path; caller is responsible for os.unlink() cleanup.
    """
    kb = Path(kb_root)
    # Single-quote YAML scalars — safe for paths without single quotes
    lines = [
        "knowledge:",
        f"  baseDir: '{kb_root}'",
        f"  rawDir: '{kb_root}'",
        f"  processedDir: '{kb / 'processed'}'",
        f"  indexFile: '{kb / 'rag_index.json'}'",
        f"  keywordFile: '{kb / 'rag_index.keywords.json'}'",
        f"  vectorIndex: '{kb / 'rag_vectors_v2.npy'}'",
        f"  vectorMeta: '{kb / 'rag_vectors_meta_v2.pkl'}'",
        "wiki:",
        f"  dir: '{kb / 'wiki'}'",
    ]
    with tempfile.NamedTemporaryFile(mode='w', suffix='.yml', delete=False) as f:
        f.write("\n".join(lines) + "\n")
        return f.name


def _make_script_env(kb_root: str, scripts_dir: Path, extra: Optional[Dict[str, str]] = None) -> Dict[str, str]:
    """Build subprocess env: inherit os.environ + KNOWLEDGE_DIR + PYTHONPATH.

    Does NOT write the instance.yml here (caller manages temp file lifecycle).
    Pass INSTANCE_CONFIG via `extra` if needed.
    """
    env = dict(os.environ)
    env["KNOWLEDGE_DIR"] = kb_root
    # Ensure scripts_dir is in PYTHONPATH so `from config import cfg` works
    existing_py = env.get("PYTHONPATH", "")
    env["PYTHONPATH"] = str(scripts_dir) + (os.pathsep + existing_py if existing_py else "")
    if extra:
        env.update(extra)
    return env


def _popen_stream(
    script: Path,
    env: Dict[str, str],
    cwd: str,
    log_buf: List[str],
    db: Database,
    job_id: str,
    flush_every: int = 8,
    timeout: int = 3600,
) -> int:
    """Run script with Popen, stream stdout to log_buf, flush to DB every N lines.
    Returns process returncode.  stderr is appended to log_buf on exit.
    """
    proc = subprocess.Popen(
        [sys.executable, str(script)],
        cwd=cwd, env=env,
        stdout=subprocess.PIPE, stderr=subprocess.PIPE,
        text=True, bufsize=1,
    )
    flush_count = 0
    try:
        for line in proc.stdout:  # type: ignore[union-attr]
            log_buf.append(line.rstrip("\n"))
            flush_count += 1
            if flush_count % flush_every == 0:
                try:
                    db.execute(
                        "UPDATE kb_jobs SET log=? WHERE id=?",
                        ("\n".join(log_buf)[-8000:], job_id),
                    )
                    db.commit()
                except Exception:
                    pass
        proc.wait(timeout=timeout)
    except subprocess.TimeoutExpired:
        proc.kill()
        log_buf.append("[超时] 进程已终止")
    stderr_out = (proc.stderr.read() or "").strip()  # type: ignore[union-attr]
    if stderr_out:
        log_buf.append("--- stderr ---")
        log_buf.extend(stderr_out[-2000:].splitlines())
    return proc.returncode


def _run_index_job(db: Database, kb_id: str, job_id: str, kb_root: str, index_type: str, mode: str, chunk_strategy: str = "token"):
    """Background thread — extract text then build index, streaming log to DB."""

    def _flush(status: str, log_buf: List[str]):
        try:
            db.execute(
                "UPDATE kb_jobs SET status=?, log=? WHERE id=?",
                (status, "\n".join(log_buf)[-8000:], job_id),
            )
            db.commit()
        except Exception:
            pass

    kb_path = Path(kb_root)
    scripts_dir = _find_scripts_dir(kb_path)
    instance_yml = _write_instance_yml(kb_root) if scripts_dir else None
    log_buf: List[str] = []
    errors: List[str] = []

    _flush("running", log_buf)
    try:
        if index_type in ("bm25", "all"):
            if not scripts_dir:
                log_buf.append("[BM25] 脚本目录未找到 — 已跳过")
            else:
                env = _make_script_env(kb_root, scripts_dir, {"INSTANCE_CONFIG": instance_yml})
                processed_dir = kb_path / "processed"

                # ── Step 1: 文本提取 ──
                extract_script = scripts_dir / "extract_docs.py"
                if extract_script.exists():
                    existing_txts = list(processed_dir.glob("*.txt")) if processed_dir.is_dir() else []
                    if not existing_txts or mode == "full":
                        log_buf.append("[提取] 正在提取原始文档为文本...")
                        _flush("running", log_buf)
                        rc = _popen_stream(extract_script, env, kb_root, log_buf, db, job_id, flush_every=10)
                        if rc != 0:
                            errors.append("文本提取失败（returncode={rc}）")
                    else:
                        log_buf.append(f"[提取] 已有 {len(existing_txts)} 个文本文件，跳过（全量模式可强制重新提取）")
                else:
                    log_buf.append("[提取] extract_docs.py 未找到，跳过（需手动准备 processed/ 目录）")

                # ── Step 2: 构建 BM25 索引 ──
                bm25_script = next(
                    (scripts_dir / n for n in ("rebuild_index.py", "build_index.py") if (scripts_dir / n).exists()),
                    None,
                )
                if bm25_script:
                    log_buf.append(f"[BM25] 正在构建索引（{bm25_script.name}）...")
                    _flush("running", log_buf)
                    rc = _popen_stream(bm25_script, env, kb_root, log_buf, db, job_id, flush_every=5)
                    if rc != 0:
                        errors.append(f"BM25 构建失败（returncode={rc}）")
                    elif chunk_strategy == "contextual":
                        # ── Step 2b: Contextual chunking enrichment ──
                        log_buf.append("[Contextual] 启动上下文增强分块...")
                        _flush("running", log_buf)
                        _contextual_enrich_index(kb_path, scripts_dir, log_buf, db, job_id, _flush)
                    elif chunk_strategy == "page":
                        # ── Step 2c: PageIndex build (page-level, no vector) ──
                        log_buf.append("[PageIndex] 构建页面级索引...")
                        _flush("running", log_buf)
                        _build_page_index(kb_path, log_buf, db, job_id, _flush)
                else:
                    log_buf.append("[BM25] rebuild_index.py / build_index.py 均未找到 — 已跳过")

        if index_type in ("vector", "all"):
            if not scripts_dir:
                log_buf.append("[Vector] 脚本目录未找到 — 已跳过")
            else:
                env = _make_script_env(kb_root, scripts_dir, {"INSTANCE_CONFIG": instance_yml})
                vec_script = scripts_dir / "build_vector_index.py"
                if vec_script.exists():
                    log_buf.append("[Vector] 正在构建向量索引...")
                    _flush("running", log_buf)
                    rc = _popen_stream(vec_script, env, kb_root, log_buf, db, job_id, flush_every=5, timeout=3600)
                    if rc != 0:
                        errors.append(f"向量索引失败（returncode={rc}）")
                else:
                    log_buf.append("[Vector] build_vector_index.py 未找到 — 已跳过")

        if index_type in ("graph", "all"):
            log_buf.append("[Graph] 图谱由 wiki_compile.py 生成 wiki/relations.json — 请先运行 Wiki 编译")

        if errors:
            log_buf.append("\n❌ ERRORS:\n" + "\n".join(errors))
        _flush("failed" if errors else "completed", log_buf)
        db.execute("UPDATE kb_jobs SET completed_at=? WHERE id=?", (now_utc(), job_id))
        db.commit()

    except Exception as exc:
        log_buf.append(f"\n❌ 异常: {exc}")
        _flush("failed", log_buf)
        db.execute(
            "UPDATE kb_jobs SET error=?, completed_at=? WHERE id=?",
            (str(exc)[:500], now_utc(), job_id),
        )
        db.commit()
    finally:
        if instance_yml:
            try:
                os.unlink(instance_yml)
            except Exception:
                pass


def _detect_local_llm_url() -> str:
    """Probe common local LLM endpoints and return the first that responds.

    Priority: VLLM_URL env var → Ollama :11434 → Ollama :11435 → vLLM :8001
    Falls back to the env default so we never block on a missing service here.
    """
    import urllib.request as _ur
    # If explicitly set by operator/user, trust it
    if os.environ.get("VLLM_URL"):
        return os.environ["VLLM_URL"]

    candidates = [
        "http://127.0.0.1:11434/v1/chat/completions",  # Ollama default
        "http://127.0.0.1:11435/v1/chat/completions",  # Ollama alt port
        "http://127.0.0.1:8001/v1/chat/completions",   # vLLM on non-conflicting port
    ]
    for url in candidates:
        try:
            req = _ur.Request(url, data=b'{}', headers={"Content-Type": "application/json"}, method="POST")
            _ur.urlopen(req, timeout=2)
        except Exception as exc:
            # Connection refused / timeout → not available; 4xx means server IS there
            msg = str(exc)
            if "400" in msg or "422" in msg or "405" in msg or "404" in msg:
                return url
        else:
            return url  # 2xx on empty payload — unlikely but accept
    # Nothing found; return Ollama default and let wiki_compile.py fail with a clear message
    return "http://127.0.0.1:11434/v1/chat/completions"


def _run_wiki_job(db: Database, kb_id: str, job_id: str, kb_root: str, mode: str, concurrency: int):
    """Background thread: run wiki_compile.py with live log streaming to DB."""

    def _flush(status: str, log_buf: List[str]):
        try:
            db.execute(
                "UPDATE kb_jobs SET status=?, log=? WHERE id=?",
                (status, "\n".join(log_buf)[-8000:], job_id),
            )
            db.commit()
        except Exception:
            pass

    kb_path = Path(kb_root)
    scripts_dir = _find_scripts_dir(kb_path)
    script = scripts_dir / "wiki_compile.py" if scripts_dir else None
    instance_yml = _write_instance_yml(kb_root) if scripts_dir else None
    log_buf: List[str] = []

    _flush("running", log_buf)
    try:
        if not script or not script.exists():
            db.execute(
                "UPDATE kb_jobs SET status='failed', error=?, completed_at=? WHERE id=?",
                ("wiki_compile.py 未在脚本目录中找到", now_utc(), job_id),
            )
            db.commit()
            return

        llm_url = _detect_local_llm_url()
        log_buf.append(f"[Wiki] LLM 端点: {llm_url}")
        _flush("running", log_buf)

        env = _make_script_env(kb_root, scripts_dir, {
            "INSTANCE_CONFIG": instance_yml,
            "WIKI_DIR": str(kb_path / "wiki"),
            "VLLM_URL": llm_url,
        })
        if mode == "rebuild":
            env["WIKI_REBUILD"] = "1"
        env["WIKI_CONCURRENCY"] = str(concurrency)

        rc = _popen_stream(script, env, kb_root, log_buf, db, job_id, flush_every=3, timeout=7200)

        _flush("failed" if rc != 0 else "completed", log_buf)
        db.execute("UPDATE kb_jobs SET completed_at=? WHERE id=?", (now_utc(), job_id))
        db.commit()

    except Exception as exc:
        log_buf.append(f"\n❌ 异常: {exc}")
        _flush("failed", log_buf)
        db.execute(
            "UPDATE kb_jobs SET error=?, completed_at=? WHERE id=?",
            (str(exc)[:500], now_utc(), job_id),
        )
        db.commit()
    finally:
        if instance_yml:
            try:
                os.unlink(instance_yml)
            except Exception:
                pass


def _find_scripts_dir(kb_path: Path) -> Optional[Path]:
    """Locate the scripts dir containing rebuild_index.py / wiki_compile.py etc.
    Checks kb_path itself (if scripts live alongside data) and the canonical
    project path agentexample/qaagent67lambda/scripts/.
    Uses rebuild_index.py or wiki_compile.py as the presence marker
    (build_index.py is the old hardcoded variant and no longer required).
    """
    _markers = ("rebuild_index.py", "wiki_compile.py", "build_index.py")

    # Check if scripts live inside kb_path or its parent
    for cand in [kb_path, kb_path.parent]:
        if any((cand / m).exists() for m in _markers):
            return cand

    # Walk up from this source file looking for the canonical scripts dir
    here = Path(__file__).resolve()
    for _ in range(8):
        here = here.parent
        cand = here / "agentexample" / "qaagent67lambda" / "scripts"
        if cand.is_dir() and any((cand / m).exists() for m in _markers):
            return cand

    return None


# Subdirs to consult when the index files aren't found directly at root.
# Index-build scripts conventionally drop their output one level below the
# raw-data root (so the KB root_dir can stay focused on source content).
# `KnowledgeSpace` is the canonical name; the others are historical aliases
# kept for robustness with older deployments.
_INDEX_OUTPUT_SUBDIRS = ("KnowledgeSpace", "knowledge_space", ".index", "index")


def _resolve_index_dir(kb_root: str, marker: str) -> Path:
    """Return the path where `marker` (e.g. 'rag_index.json' or 'wiki') is
    actually present, preferring the KB's root_dir but falling back to known
    subdirs where build scripts conventionally write output."""
    root = Path(kb_root)
    direct = root / marker
    if direct.exists():
        return root
    for sub in _INDEX_OUTPUT_SUBDIRS:
        cand = root / sub
        if (cand / marker).exists():
            return cand
    # Nothing found — return root anyway so callers see "exists: false".
    return root


def _get_index_status(kb_root: str) -> Dict[str, Any]:
    """Read index file presence/stats. Indexes may live at kb_root or one
    level below in a conventional output subdir (e.g. KnowledgeSpace/).
    Each artifact is resolved independently — a deployment that has BM25 at
    root but wiki under KnowledgeSpace/ works fine."""

    def _fstat(p: Path) -> Optional[Dict]:
        if p.exists():
            s = p.stat()
            return {"exists": True, "size": s.st_size, "mtime": s.st_mtime,
                    "path": str(p)}
        return {"exists": False}

    bm25_file = _resolve_index_dir(kb_root, "rag_index.json") / "rag_index.json"
    vector_file = _resolve_index_dir(kb_root, "rag_vectors_v2.npy") / "rag_vectors_v2.npy"
    wiki_dir = _resolve_index_dir(kb_root, "wiki") / "wiki"
    page_file = _resolve_index_dir(kb_root, "rag_page_index.json") / "rag_page_index.json"

    bm25 = _fstat(bm25_file)
    if bm25["exists"]:
        try:
            import json as _json
            with open(bm25_file) as f:
                idx = _json.load(f)
            bm25["total_chunks"] = idx.get("total_chunks", 0)
            bm25["total_files"] = idx.get("total_files", 0)
        except Exception:
            pass

    vector = _fstat(vector_file)
    wiki = {"exists": wiki_dir.is_dir()}
    if wiki["exists"]:
        wiki["path"] = str(wiki_dir)
        subdirs = {}
        for sub in ["sources", "entities", "topics", "analyses"]:
            d = wiki_dir / sub
            subdirs[sub] = len(list(d.glob("*.md"))) if d.is_dir() else 0
        wiki["subdirs"] = subdirs
        wiki["total_pages"] = sum(subdirs.values())

    pageindex = _fstat(page_file)
    if pageindex["exists"]:
        try:
            import json as _json
            with open(page_file, encoding="utf-8") as f:
                pages = _json.load(f)
            pageindex["total_pages"] = len(pages)
            pageindex["total_docs"] = len(set(p.get("source", "") for p in pages))
        except Exception:
            pass

    return {"bm25": bm25, "vector": vector, "wiki": wiki, "pageindex": pageindex}


# ─── Chunk index cache ───
# Avoid re-reading a 17 MB JSON on every request; invalidate when file mtime changes.
_chunk_cache: Dict[str, Any] = {}   # kb_id → {"mtime": float, "chunks": list, "total": int}


def _load_chunks(kb_root: str, kb_id: str) -> Optional[List[Dict]]:
    """Load chunks from rag_index.json with mtime-based in-process cache."""
    index_file = Path(kb_root) / "rag_index.json"
    if not index_file.exists():
        return None
    mtime = index_file.stat().st_mtime
    cached = _chunk_cache.get(kb_id)
    if cached and cached["mtime"] == mtime:
        return cached["chunks"]
    try:
        with open(index_file, encoding="utf-8") as f:
            data = json.load(f)
        chunks = data.get("chunks", [])
        _chunk_cache[kb_id] = {"mtime": mtime, "chunks": chunks, "total": len(chunks)}
        return chunks
    except Exception:
        return None


# ─── CRUD Endpoints ───

@router.get("")
async def list_kbs(
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    rows = db.fetchall(
        "SELECT * FROM knowledge_bases WHERE tenant_id=? ORDER BY created_at DESC",
        (tenant.tenant_id,)
    )
    result = []
    for r in rows:
        status = _get_index_status(r["root_dir"])
        result.append({**r, "index_status": status})
    return {"items": result}


@router.post("", status_code=201)
async def create_kb(
    req: CreateKBRequest,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    root = Path(req.root_dir)
    if not root.is_dir():
        raise HTTPException(status_code=400, detail=f"Directory not found: {req.root_dir}")

    kb_id = gen_id("kb_")
    db.execute(
        "INSERT INTO knowledge_bases (id, tenant_id, name, root_dir, description, created_at, updated_at) "
        "VALUES (?, ?, ?, ?, ?, ?, ?)",
        (kb_id, tenant.tenant_id, req.name, str(root.resolve()), req.description, now_utc(), now_utc())
    )
    db.commit()
    return db.fetchone("SELECT * FROM knowledge_bases WHERE id=?", (kb_id,))


@router.get("/{kb_id}")
async def get_kb(
    kb_id: str,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
    if not kb:
        raise HTTPException(status_code=404, detail="Knowledge base not found")
    status = _get_index_status(kb["root_dir"])
    return {**kb, "index_status": status}


@router.delete("/{kb_id}", status_code=204)
async def delete_kb(
    kb_id: str,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
    if not kb:
        raise HTTPException(status_code=404, detail="Knowledge base not found")
    db.execute("DELETE FROM kb_files WHERE kb_id=?", (kb_id,))
    db.execute("DELETE FROM kb_jobs WHERE kb_id=?", (kb_id,))
    db.execute("DELETE FROM knowledge_bases WHERE id=?", (kb_id,))
    db.commit()
    return None


# ─── Chunk Browser ───

@router.get("/{kb_id}/chunks")
async def list_chunks(
    kb_id: str,
    q: str = Query(default="", description="Keyword filter (case-insensitive substring)"),
    source: str = Query(default="", description="Filter by source file name"),
    page: int = Query(default=1, ge=1),
    limit: int = Query(default=20, ge=1, le=100),
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
    if not kb:
        raise HTTPException(status_code=404, detail="Knowledge base not found")

    chunks = _load_chunks(kb["root_dir"], kb_id)
    if chunks is None:
        raise HTTPException(status_code=404, detail="BM25 index not built yet (rag_index.json not found)")

    # Filter
    q_lower = q.strip().lower()
    src_lower = source.strip().lower()
    filtered = chunks
    if q_lower:
        filtered = [c for c in filtered if q_lower in c.get("text", "").lower()]
    if src_lower:
        filtered = [c for c in filtered if src_lower in c.get("source", "").lower()]

    total = len(filtered)
    offset = (page - 1) * limit
    page_chunks = filtered[offset: offset + limit]

    # Return trimmed chunks (avoid huge payloads)
    return {
        "total": total,
        "total_all": len(chunks),
        "page": page,
        "limit": limit,
        "pages": max(1, (total + limit - 1) // limit),
        "chunks": [
            {
                "id": c.get("id", ""),
                "source": c.get("source", ""),
                "chunk_id": c.get("chunk_id", 0),
                "text": c.get("text", ""),
                "length": c.get("length", 0),
            }
            for c in page_chunks
        ],
    }


# ─── File Management ───

@router.get("/{kb_id}/files")
async def list_files(
    kb_id: str,
    path: str = Query(default="", description="Subdirectory relative path to list files in"),
    q: str = Query(default="", description="Filename filter (case-insensitive substring)"),
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
    if not kb:
        raise HTTPException(status_code=404, detail="Knowledge base not found")

    # Load saved ignore list
    raw_ignored = kb.get("ignore_dirs") or "[]"
    try:
        ignored: Set[str] = set(json.loads(raw_ignored))
    except Exception:
        ignored = set()

    added = db.fetchall("SELECT file_path FROM kb_files WHERE kb_id=?", (kb_id,))
    added_paths = {r["file_path"] for r in added}

    root_dir = kb["root_dir"]
    # Run blocking I/O off the event loop
    files, dirs = await asyncio.get_event_loop().run_in_executor(
        None,
        lambda: (
            _walk_dir(root_dir, added_paths, ignored, subpath=path, name_filter=q),
            _list_dir_shallow(root_dir, path, ignored),
        )
    )

    return {
        "kb_id": kb_id,
        "root_dir": root_dir,
        "files": files,
        "dirs": dirs,
        "ignored_dirs": sorted(ignored),
        "current_path": path,
    }


@router.get("/{kb_id}/dirs")
async def list_dirs(
    kb_id: str,
    path: str = Query(default="", description="Parent directory relative path"),
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """Return immediate subdirectories of `path` (lazy tree expansion)."""
    kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
    if not kb:
        raise HTTPException(status_code=404, detail="Knowledge base not found")

    raw_ignored = kb.get("ignore_dirs") or "[]"
    try:
        ignored: Set[str] = set(json.loads(raw_ignored))
    except Exception:
        ignored = set()

    dirs = await asyncio.get_event_loop().run_in_executor(
        None,
        lambda: _list_dir_shallow(kb["root_dir"], path, ignored),
    )
    return {"dirs": dirs, "path": path}


@router.get("/{kb_id}/ignore-dirs")
async def get_ignore_dirs(
    kb_id: str,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
    if not kb:
        raise HTTPException(status_code=404, detail="Knowledge base not found")
    raw = kb.get("ignore_dirs") or "[]"
    try:
        ignored = sorted(set(json.loads(raw)))
    except Exception:
        ignored = []
    return {"ignored_dirs": ignored}


@router.put("/{kb_id}/ignore-dirs")
async def update_ignore_dirs(
    kb_id: str,
    req: UpdateIgnoreDirsRequest,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
    if not kb:
        raise HTTPException(status_code=404, detail="Knowledge base not found")

    # Normalise: strip slashes, dedup, sort
    clean = sorted({p.strip("/").strip() for p in req.ignored_dirs if p.strip()})
    db.execute(
        "UPDATE knowledge_bases SET ignore_dirs=?, updated_at=? WHERE id=?",
        (json.dumps(clean), now_utc(), kb_id)
    )
    db.commit()
    return {"ignored_dirs": clean}


@router.post("/{kb_id}/files", status_code=201)
async def add_files(
    kb_id: str,
    req: AddFilesRequest,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """Register files into a KB.

    audit/AUDIT_2026-06-05.md #27 fix: previously this endpoint accepted
    any host-filesystem path as `file_paths[i]` and INSERTed it into
    `kb_files` without checking containment. A tenant with `kb_id` could
    register `/etc/passwd` or any other server file — subsequent indexer
    jobs would then read and chunk it into the searchable RAG store.
    The same codebase enforces containment correctly at knowledge.py:1050
    for the wiki endpoint (`str(full).startswith(str(wiki_path.resolve()))`);
    `add_files` just forgot.
    """
    kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
    if not kb:
        raise HTTPException(status_code=404, detail="Knowledge base not found")

    # Path containment: every fp must be inside the KB's root_dir. Resolve
    # both sides to absolute, expand symlinks, then check prefix + os.sep
    # (the trailing-sep check defeats /foo vs /foobar prefix collisions).
    kb_root_str = kb.get("root_dir", "")
    if not kb_root_str:
        raise HTTPException(status_code=400, detail="Knowledge base has no root_dir configured")
    import os as _os
    kb_root = Path(kb_root_str).resolve()
    kb_root_with_sep = str(kb_root) + _os.sep

    added = []
    rejected: list = []
    for fp in req.file_paths:
        p = Path(fp)
        # Resolve before file-exists check so a symlink pointing outside
        # the kb_root is rejected based on its target, not its alias.
        try:
            p_resolved = p.resolve()
        except (OSError, RuntimeError):
            rejected.append({"path": str(p), "reason": "unresolvable"})
            continue
        if not (str(p_resolved) == str(kb_root)
                or str(p_resolved).startswith(kb_root_with_sep)):
            rejected.append({"path": str(p), "reason": "outside kb_root"})
            continue
        if not p_resolved.is_file():
            rejected.append({"path": str(p), "reason": "not a file"})
            continue
        existing = db.fetchone(
            "SELECT id FROM kb_files WHERE kb_id=? AND file_path=?",
            (kb_id, str(p_resolved))
        )
        if existing:
            continue
        fid = gen_id("kbf_")
        db.execute(
            "INSERT INTO kb_files (id, kb_id, file_path, file_name, file_type, size, status, added_at) "
            "VALUES (?, ?, ?, ?, ?, ?, 'pending', ?)",
            (fid, kb_id, str(p_resolved), p_resolved.name,
             p_resolved.suffix.lower().lstrip("."), p_resolved.stat().st_size, now_utc())
        )
        added.append(fid)
    db.commit()
    resp: Dict[str, Any] = {"added": len(added)}
    if rejected:
        resp["rejected"] = rejected
    return resp


@router.delete("/{kb_id}/files/{file_id}", status_code=204)
async def remove_file(
    kb_id: str,
    file_id: str,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
    if not kb:
        raise HTTPException(status_code=404, detail="Knowledge base not found")
    db.execute("DELETE FROM kb_files WHERE id=? AND kb_id=?", (file_id, kb_id))
    db.commit()
    return None


# ─── Index Build Jobs ───

@router.post("/{kb_id}/index", status_code=202)
async def trigger_index(
    kb_id: str,
    req: IndexBuildRequest,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
    if not kb:
        raise HTTPException(status_code=404, detail="Knowledge base not found")

    job_id = gen_id("kbj_")
    db.execute(
        "INSERT INTO kb_jobs (id, kb_id, tenant_id, job_type, status, created_at) VALUES (?,?,?,?,?,?)",
        (job_id, kb_id, tenant.tenant_id, f"index:{req.index_type}", "pending", now_utc())
    )
    db.commit()

    t = threading.Thread(
        target=_run_index_job,
        args=(db, kb_id, job_id, kb["root_dir"], req.index_type, req.mode, req.chunk_strategy),
        daemon=True,
    )
    t.start()
    return {"job_id": job_id, "status": "pending", "chunk_strategy": req.chunk_strategy}


# ─── Wiki Compile Jobs ───

@router.post("/{kb_id}/wiki", status_code=202)
async def trigger_wiki(
    kb_id: str,
    req: WikiBuildRequest,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
    if not kb:
        raise HTTPException(status_code=404, detail="Knowledge base not found")

    job_id = gen_id("kbj_")
    db.execute(
        "INSERT INTO kb_jobs (id, kb_id, tenant_id, job_type, status, created_at) VALUES (?,?,?,?,?,?)",
        (job_id, kb_id, tenant.tenant_id, "wiki", "pending", now_utc())
    )
    db.commit()

    t = threading.Thread(
        target=_run_wiki_job,
        args=(db, kb_id, job_id, kb["root_dir"], req.mode, req.concurrency),
        daemon=True,
    )
    t.start()
    return {"job_id": job_id, "status": "pending"}


@router.put("/{kb_id}/wiki/pause", status_code=200)
async def pause_wiki(
    kb_id: str,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
    if not kb:
        raise HTTPException(status_code=404, detail="Knowledge base not found")

    # Write a stop flag that wiki_compile.py checks
    stop_file = Path(kb["root_dir"]) / ".wiki_stop"
    if stop_file.exists():
        stop_file.unlink()
        return {"action": "resumed"}
    else:
        stop_file.touch()
        return {"action": "paused"}


# ─── Wiki Content ───

@router.get("/{kb_id}/wiki/tree")
async def wiki_tree(
    kb_id: str,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
    if not kb:
        raise HTTPException(status_code=404, detail="Knowledge base not found")

    # Honor build scripts that drop output into a subdir like
    # KnowledgeSpace/ instead of straight under root_dir.
    wiki_path = _resolve_index_dir(kb["root_dir"], "wiki") / "wiki"
    result: Dict[str, Any] = {}
    for sub in ["sources", "entities", "topics", "analyses"]:
        d = wiki_path / sub
        pages = []
        if d.is_dir():
            for f in sorted(d.glob("*.md")):
                s = f.stat()
                pages.append({
                    "name": f.stem,
                    "path": f"{sub}/{f.name}",
                    "size": s.st_size,
                    "modified": s.st_mtime,
                })
        result[sub] = pages

    idx = wiki_path / "index.md"
    result["index_content"] = idx.read_text(encoding="utf-8") if idx.exists() else ""
    return result


@router.get("/{kb_id}/wiki/schedule")
async def get_wiki_schedule(
    kb_id: str,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """Return the current auto-recompile schedule for this knowledge base."""
    kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
    if not kb:
        raise HTTPException(status_code=404, detail="Knowledge base not found")
    return _load_wiki_schedule(kb["root_dir"])


@router.put("/{kb_id}/wiki/schedule")
async def update_wiki_schedule(
    kb_id: str,
    req: WikiScheduleRequest,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """Enable / disable the auto-recompile schedule and set the interval."""
    kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
    if not kb:
        raise HTTPException(status_code=404, detail="Knowledge base not found")
    cfg = _load_wiki_schedule(kb["root_dir"])
    cfg["enabled"] = req.enabled
    cfg["interval"] = req.interval
    cfg["next_run"] = _compute_next_run(req.interval) if req.enabled else cfg.get("next_run", "")
    _save_wiki_schedule(kb["root_dir"], cfg)
    return cfg


@router.get("/{kb_id}/wiki/{page_path:path}")
async def wiki_page(
    kb_id: str,
    page_path: str,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
    if not kb:
        raise HTTPException(status_code=404, detail="Knowledge base not found")

    # Honor build scripts that drop output into a subdir like
    # KnowledgeSpace/ instead of straight under root_dir.
    wiki_path = _resolve_index_dir(kb["root_dir"], "wiki") / "wiki"
    full = (wiki_path / page_path).resolve()
    # Path traversal check
    if not str(full).startswith(str(wiki_path.resolve())):
        raise HTTPException(status_code=403, detail="Forbidden")
    if not full.exists():
        raise HTTPException(status_code=404, detail="Page not found")
    return {"path": page_path, "content": full.read_text(encoding="utf-8"), "size": full.stat().st_size}


@router.delete("/{kb_id}/wiki/analyses/{name}", status_code=204)
async def delete_analysis(
    kb_id: str,
    name: str,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
    if not kb:
        raise HTTPException(status_code=404, detail="Knowledge base not found")

    p = Path(kb["root_dir"]) / "wiki" / "analyses" / f"{name}.md"
    if p.exists() and str(p.resolve()).startswith(str((Path(kb["root_dir"]) / "wiki").resolve())):
        p.unlink()
    return None


# ─── P0 helper: sync LLM call (non-streaming) ───

def _call_llm_once(
    chat_url: str,
    headers: dict,
    model: str,
    messages: List[Dict],
    is_anthropic: bool = False,
    max_tokens: int = 256,
    timeout: int = 30,
) -> str:
    """Make a single non-streaming LLM call and return the full response text."""
    import urllib.request as _ur
    if is_anthropic:
        system_parts = [m["content"] for m in messages if m["role"] == "system"]
        user_msgs = [m for m in messages if m["role"] != "system"]
        body_dict: dict = {
            "model": model,
            "messages": user_msgs,
            "max_tokens": max_tokens,
            "stream": False,
        }
        if system_parts:
            body_dict["system"] = "\n\n".join(system_parts)
        body = json.dumps(body_dict, ensure_ascii=False).encode("utf-8")
        req = _ur.Request(chat_url, data=body, headers=headers, method="POST")
        with _ur.urlopen(req, timeout=timeout) as resp:
            d = json.loads(resp.read())
            return d.get("content", [{}])[0].get("text", "")
    else:
        body = json.dumps({
            "model": model,
            "messages": messages,
            "max_tokens": max_tokens,
            "stream": False,
        }, ensure_ascii=False).encode("utf-8")
        req = _ur.Request(chat_url, data=body, headers=headers, method="POST")
        with _ur.urlopen(req, timeout=timeout) as resp:
            d = json.loads(resp.read())
            return d["choices"][0]["message"].get("content", "")


# ─── P0 helper: Reciprocal Rank Fusion ───

def _rrf_fuse(all_rankings: List[List[Dict]], k: int = 60) -> List[Dict]:
    """Reciprocal Rank Fusion: score(d) += 1/(k + rank) across all rankings.

    Returns a deduplicated list sorted by descending RRF score.
    Deduplication key: (source, first 80 chars of text).
    """
    scores: Dict[str, float] = {}
    best: Dict[str, Dict] = {}
    for ranking in all_rankings:
        for rank, chunk in enumerate(ranking, start=1):
            key = f"{chunk.get('source', '')}::{(chunk.get('text') or '')[:80]}"
            scores[key] = scores.get(key, 0.0) + 1.0 / (k + rank)
            if key not in best:
                best[key] = chunk
    return sorted(best.values(), key=lambda c: -scores[
        f"{c.get('source', '')}::{(c.get('text') or '')[:80]}"
    ])


# ─── P0 helper: Wiki schedule persistence ───

def _load_wiki_schedule(kb_root: str) -> dict:
    """Read {kb_root}/.wiki_schedule.json; return defaults if absent."""
    p = Path(kb_root) / ".wiki_schedule.json"
    defaults = {"enabled": False, "interval": "weekly", "next_run": "", "last_run": ""}
    if not p.exists():
        return defaults
    try:
        return {**defaults, **json.loads(p.read_text(encoding="utf-8"))}
    except Exception:
        return defaults


def _save_wiki_schedule(kb_root: str, config: dict):
    """Write {kb_root}/.wiki_schedule.json atomically."""
    p = Path(kb_root) / ".wiki_schedule.json"
    tmp = p.with_suffix(".json.tmp")
    tmp.write_text(json.dumps(config, ensure_ascii=False, indent=2), encoding="utf-8")
    tmp.replace(p)


def _compute_next_run(interval: str) -> str:
    """Return ISO-8601 UTC timestamp for the next run based on interval."""
    import datetime
    now = datetime.datetime.utcnow()
    if interval == "daily":
        delta = datetime.timedelta(days=1)
    else:  # weekly
        delta = datetime.timedelta(weeks=1)
    return (now + delta).strftime("%Y-%m-%dT%H:%M:%SZ")


# ─── P0 helper: Contextual chunking enrichment ───

def _contextual_enrich_index(
    kb_path: Path,
    scripts_dir: Path,
    log_buf: List[str],
    db: Database,
    job_id: str,
    flush_fn,
    provider_id: str = "",
    model_override: str = "",
):
    """After BM25 build: read rag_index.json, group chunks by source doc, generate one LLM
    summary per doc, prepend '[文档背景: {summary}]\\n' to every chunk from that doc, and
    write the enriched index back.

    Falls back gracefully on any error so the main index job is never broken.
    """
    index_file = kb_path / "rag_index.json"
    if not index_file.exists():
        log_buf.append("[Contextual] rag_index.json 不存在，跳过上下文增强")
        return

    try:
        from agentpaas.api.v1.providers_api import DEFAULT_PROVIDERS
        chat_url, headers, model_name, _ = _resolve_llm(provider_id, model_override)
        pdef = DEFAULT_PROVIDERS.get(provider_id, {})
        is_anthropic = pdef.get("type") == "anthropic"
    except Exception as e:
        log_buf.append(f"[Contextual] LLM 解析失败（{e}），跳过上下文增强")
        flush_fn("running", log_buf)
        return

    try:
        raw = json.loads(index_file.read_text(encoding="utf-8"))
    except Exception as e:
        log_buf.append(f"[Contextual] 读取 rag_index.json 失败（{e}）")
        return

    # raw can be list of chunks or dict; normalise
    chunks: List[Dict] = raw if isinstance(raw, list) else raw.get("chunks", [])

    # Group chunks by source document
    from collections import defaultdict
    doc_chunks: Dict[str, List[int]] = defaultdict(list)
    for idx, chunk in enumerate(chunks):
        src = chunk.get("source") or chunk.get("file") or ""
        doc_chunks[src].append(idx)

    log_buf.append(f"[Contextual] 共 {len(doc_chunks)} 篇文档，生成摘要中...")
    flush_fn("running", log_buf)

    summaries: Dict[str, str] = {}
    for src, idxs in doc_chunks.items():
        # Gather up to 1500 chars of text as doc sample
        sample = "\n".join((chunks[i].get("text") or "")[:300] for i in idxs[:5])[:1500]
        prompt = [
            {"role": "system", "content": "你是文档摘要助手，请用1-3句话概括以下文档的主要内容。"},
            {"role": "user", "content": f"文档: {src}\n\n内容摘录:\n{sample}\n\n一句话摘要:"},
        ]
        try:
            summary = _call_llm_once(
                chat_url, headers, model_name, prompt,
                is_anthropic=is_anthropic, max_tokens=128, timeout=20,
            ).strip()
            summaries[src] = summary
        except Exception as e:
            summaries[src] = src  # fallback: just use filename

    # Prepend background context to each chunk
    enriched = 0
    for chunk in chunks:
        src = chunk.get("source") or chunk.get("file") or ""
        summary = summaries.get(src, "")
        if summary and not (chunk.get("text") or "").startswith("[文档背景:"):
            chunk["text"] = f"[文档背景: {summary}]\n{chunk.get('text', '')}"
            enriched += 1

    # Write back
    try:
        if isinstance(raw, list):
            out = chunks
        else:
            raw["chunks"] = chunks
            out = raw
        index_file.write_text(json.dumps(out, ensure_ascii=False, indent=2), encoding="utf-8")
        log_buf.append(f"[Contextual] ✅ 已为 {enriched} 个 chunk 添加文档背景（{len(summaries)} 篇文档）")
    except Exception as e:
        log_buf.append(f"[Contextual] 写回索引失败（{e}）")
    flush_fn("running", log_buf)


# ─── P0 helper: Wiki auto-recompile background scheduler ───

_SCHEDULE_CHECKER_STARTED = False
_SCHEDULE_CHECKER_LOCK = threading.Lock()


def _wiki_schedule_loop():
    """Daemon thread: every 30 min, scan all KB roots and trigger wiki builds if due."""
    import datetime
    from agentpaas.db.session import get_db
    while True:
        time.sleep(1800)  # 30 minutes
        try:
            db: Database = get_db()
            kbs = db.fetchall("SELECT id, root_dir FROM knowledge_bases WHERE root_dir IS NOT NULL")
            now_str = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
            for kb in kbs:
                try:
                    cfg = _load_wiki_schedule(kb["root_dir"])
                    if not cfg.get("enabled"):
                        continue
                    next_run = cfg.get("next_run", "")
                    if not next_run or now_str >= next_run:
                        # Trigger wiki build
                        job_id = gen_id("kbj_")
                        tenant_row = db.fetchone(
                            "SELECT tenant_id FROM knowledge_bases WHERE id=?", (kb["id"],)
                        )
                        tid = tenant_row["tenant_id"] if tenant_row else "system"
                        db.execute(
                            "INSERT INTO kb_jobs (id, kb_id, tenant_id, job_type, status, created_at) VALUES (?,?,?,?,?,?)",
                            (job_id, kb["id"], tid, "wiki:scheduled", "pending", now_utc()),
                        )
                        db.commit()
                        t = threading.Thread(
                            target=_run_wiki_job,
                            args=(db, kb["id"], job_id, kb["root_dir"], "incremental", 2),
                            daemon=True,
                        )
                        t.start()
                        # Update schedule
                        cfg["last_run"] = now_str
                        cfg["next_run"] = _compute_next_run(cfg.get("interval", "weekly"))
                        _save_wiki_schedule(kb["root_dir"], cfg)
                except Exception:
                    pass
        except Exception:
            pass


def _start_schedule_checker():
    """Start the wiki-schedule background thread exactly once."""
    global _SCHEDULE_CHECKER_STARTED
    with _SCHEDULE_CHECKER_LOCK:
        if _SCHEDULE_CHECKER_STARTED:
            return
        _SCHEDULE_CHECKER_STARTED = True
        t = threading.Thread(
            target=_wiki_schedule_loop,
            daemon=True,
            name="wiki-scheduler",
        )
        t.start()


# ─── P1: PageIndex — no-vector page-level RAG ───

def _build_page_index(
    kb_path: Path,
    log_buf: List[str],
    db: "Database",
    job_id: str,
    flush_fn,
    provider_id: str = "",
    model_override: str = "",
):
    """Build rag_page_index.json from processed/ text files.

    Each page becomes one index entry:
      {id, source, page, summary, text, length}

    Pages are detected by '\\f' (form feed) or '--- Page N ---' markers that
    text extractors like extract_docs.py typically emit. Falls back to
    splitting every ~3000 chars when no page markers are found.

    The summary is generated by LLM (one call per page); falls back to the
    first 150 chars of the page text if LLM is unavailable.
    """
    processed_dir = kb_path / "processed"
    if not processed_dir.is_dir():
        log_buf.append("[PageIndex] processed/ 目录不存在，跳过")
        return

    txt_files = list(processed_dir.glob("*.txt"))
    if not txt_files:
        log_buf.append("[PageIndex] processed/ 中没有 .txt 文件，跳过")
        return

    # Try to get LLM for summaries (graceful fallback if unavailable)
    chat_url = headers = model_name = None
    is_anthropic = False
    try:
        from agentpaas.api.v1.providers_api import DEFAULT_PROVIDERS
        chat_url, headers, model_name, _ = _resolve_llm(provider_id, model_override)
        pdef = DEFAULT_PROVIDERS.get(provider_id, {})
        is_anthropic = pdef.get("type") == "anthropic"
    except Exception:
        pass  # No LLM → use truncated text as summary

    log_buf.append(f"[PageIndex] 处理 {len(txt_files)} 个文档，生成页面索引...")
    flush_fn("running", log_buf)

    entries: List[Dict] = []
    for txt_file in sorted(txt_files):
        source = txt_file.stem
        try:
            text = txt_file.read_text(encoding="utf-8", errors="replace")
        except Exception:
            continue

        # Split into pages
        pages: List[str] = []
        if "\f" in text:
            # Form-feed page breaks (PDF extractors)
            pages = [p.strip() for p in text.split("\f") if p.strip()]
        else:
            import re
            # "--- Page N ---" markers
            parts = re.split(r"---\s*[Pp]age\s*\d+\s*---", text)
            if len(parts) > 1:
                pages = [p.strip() for p in parts if p.strip()]
            else:
                # No markers: split every ~3000 chars at paragraph boundary
                chunk_size = 3000
                remaining = text
                while remaining:
                    if len(remaining) <= chunk_size:
                        pages.append(remaining.strip())
                        break
                    split_at = remaining.rfind("\n\n", 0, chunk_size)
                    if split_at == -1:
                        split_at = chunk_size
                    pages.append(remaining[:split_at].strip())
                    remaining = remaining[split_at:].strip()

        for page_num, page_text in enumerate(pages, start=1):
            if not page_text:
                continue
            # Generate summary
            summary = page_text[:150]  # default fallback
            if chat_url:
                try:
                    prompt = [
                        {"role": "system", "content": "请用1-2句话概括以下页面内容，突出关键信息。"},
                        {"role": "user", "content": page_text[:1500]},
                    ]
                    summary = _call_llm_once(
                        chat_url, headers, model_name, prompt,
                        is_anthropic=is_anthropic, max_tokens=100, timeout=15,
                    ).strip() or page_text[:150]
                except Exception:
                    pass

            entries.append({
                "id": f"{source}::p{page_num}",
                "source": source,
                "page": page_num,
                "summary": summary,
                "text": page_text,
                "length": len(page_text),
            })

    if entries:
        index_file = kb_path / "rag_page_index.json"
        index_file.write_text(json.dumps(entries, ensure_ascii=False, indent=2), encoding="utf-8")
        log_buf.append(f"[PageIndex] ✅ 已生成 {len(entries)} 个页面条目（{len(txt_files)} 篇文档）→ rag_page_index.json")
    else:
        log_buf.append("[PageIndex] ⚠️ 未生成任何页面条目")
    flush_fn("running", log_buf)


def _page_index_search(kb_root: str, query: str, top_k: int) -> List[Dict]:
    """BM25-style search over rag_page_index.json (summaries + text).

    Uses a simple TF token overlap score since we avoid importing rank_bm25
    at the top level. Returns results in {id, source, page, summary, text, score} format.
    """
    index_file = Path(kb_root) / "rag_page_index.json"
    if not index_file.exists():
        return []
    try:
        entries = json.loads(index_file.read_text(encoding="utf-8"))
    except Exception:
        return []

    # Tokenise query — CJK 必须切 bigram：原来 `\W+` 分词把整句中文当成
    # 一个长 token（"循环队列怎么判断队满"），在正文里永远找不到整句 →
    # 全部 0 分按原序返回，中文检索形同失效。英文/数字 token 保持原样。
    import re

    def _query_tokens(q: str) -> set:
        tokens: set = set()
        for part in re.sub(r"[^\w一-鿿]+", " ", q.lower()).split():
            if re.search(r"[一-鿿]", part):
                if len(part) == 1:
                    tokens.add(part)
                for i in range(len(part) - 1):
                    tokens.add(part[i:i + 2])  # CJK bigram
            else:
                tokens.add(part)
        return tokens

    query_tokens = _query_tokens(query)
    if not query_tokens:
        return entries[:top_k]

    scored: List[tuple] = []
    for entry in entries:
        haystack = f"{entry.get('summary','')} {entry.get('text','')}".lower()
        score = sum(1 for tok in query_tokens if tok in haystack) / max(len(query_tokens), 1)
        scored.append((score, entry))

    scored.sort(key=lambda x: -x[0])
    results = []
    for score, entry in scored[:top_k]:
        results.append({**entry, "score": round(score, 4)})
    return results


# ─── Search helpers ───

def _search_subprocess(kb_root: str, scripts_dir: Path, mode: str, query: str, top_k: int) -> List[Dict]:
    """Run search in an isolated subprocess to avoid config singleton / module cache issues.

    search_engine.py uses `from config import cfg` at module level.  Once imported
    in-process the singleton is frozen to the first kb_root it saw.  Running as a
    subprocess gives each search a fresh Python interpreter with the correct
    INSTANCE_CONFIG / KNOWLEDGE_DIR env vars.
    """
    instance_yml = _write_instance_yml(kb_root)
    try:
        env = _make_script_env(kb_root, scripts_dir, {"INSTANCE_CONFIG": instance_yml})

        # Build the inline runner script — use json.dumps for safe string embedding
        import_stmt = (
            "from search_engine import search"
            if mode == "bm25"
            else (
                "try:\n    from search_unified import search\n"
                "except ImportError:\n    from search_engine import search"
            )
        )
        runner = (
            f"import sys, os, json\n"
            f"sys.path.insert(0, {json.dumps(str(scripts_dir))})\n"
            f"os.environ['INSTANCE_CONFIG'] = {json.dumps(instance_yml)}\n"
            f"os.environ['KNOWLEDGE_DIR'] = {json.dumps(kb_root)}\n"
            f"{import_stmt}\n"
            f"results = search({json.dumps(query)}, top_k={top_k})\n"
            f"print(json.dumps(results, ensure_ascii=False))\n"
        )
        r = subprocess.run(
            [sys.executable, "-c", runner],
            capture_output=True, text=True, timeout=30, env=env,
        )
        if r.returncode != 0:
            raise RuntimeError(r.stderr[-800:] or "搜索子进程异常退出")
        # Last line of stdout is the JSON result (earlier lines may be progress logs)
        lines = [l for l in r.stdout.strip().splitlines() if l.strip()]
        if not lines:
            return []
        return json.loads(lines[-1])
    finally:
        try:
            os.unlink(instance_yml)
        except Exception:
            pass


def _fusion_search(
    kb_root: str,
    scripts_dir: Path,
    query: str,
    top_k: int,
    n_subqueries: int = 4,
    provider_id: str = "",
    model_override: str = "",
) -> List[Dict]:
    """RAG Fusion: expand query → N sub-queries → BM25 search each → RRF merge.

    If LLM expansion fails, falls back to a single BM25 search on the original query.
    """
    # ── Step 1: query expansion ──
    sub_queries: List[str] = [query]
    try:
        from agentpaas.api.v1.providers_api import DEFAULT_PROVIDERS
        chat_url, headers, model_name, _ = _resolve_llm(provider_id, model_override)
        pdef = DEFAULT_PROVIDERS.get(provider_id, {})
        is_anthropic = pdef.get("type") == "anthropic"
        expansion_prompt = [
            {
                "role": "system",
                "content": (
                    f"请将用户的问题改写为 {n_subqueries} 个语义相近但措辞不同的子问题，"
                    "每行一个，只输出问题，不要编号或其他内容。"
                ),
            },
            {"role": "user", "content": query},
        ]
        raw = _call_llm_once(
            chat_url, headers, model_name, expansion_prompt,
            is_anthropic=is_anthropic, max_tokens=256, timeout=20,
        )
        lines = [l.strip() for l in raw.splitlines() if l.strip()]
        if lines:
            sub_queries = [query] + lines[:n_subqueries]
    except Exception:
        pass  # fallback: single query

    # ── Step 2: search each sub-query ──
    all_rankings: List[List[Dict]] = []
    for sq in sub_queries:
        try:
            results = _search_subprocess(kb_root, scripts_dir, "bm25", sq, top_k)
            if results:
                all_rankings.append(results)
        except Exception:
            pass

    if not all_rankings:
        # Last resort: direct BM25
        try:
            return _search_subprocess(kb_root, scripts_dir, "bm25", query, top_k)
        except Exception:
            return []

    # ── Step 3: RRF fusion ──
    return _rrf_fuse(all_rankings)[:top_k]


# ─── Search ───

@router.post("/{kb_id}/search")
async def search_kb(
    kb_id: str,
    req: SearchRequest,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
    if not kb:
        raise HTTPException(status_code=404, detail="Knowledge base not found")

    kb_root = kb["root_dir"]
    scripts_dir = _find_scripts_dir(Path(kb_root))

    # pageindex 模式只需要 rag_page_index.json（进程内检索，无脚本依赖）。
    # 此前这里无条件要求 BM25 的脚本目录 + rag_index.json，导致纯
    # pageindex 的知识库（如开箱的示例知识库）在页面上搜不了。
    if req.mode == "pageindex":
        if not (Path(kb_root) / "rag_page_index.json").exists():
            raise HTTPException(
                status_code=503,
                detail="PageIndex 不存在，请先在「索引构建」标签中以 page 分块策略构建。")
    else:
        if not scripts_dir:
            raise HTTPException(status_code=503, detail="搜索脚本未找到，请先构建 BM25 索引。")
        index_file = Path(kb_root) / "rag_index.json"
        if not index_file.exists():
            # BM25 索引缺失但有 pageindex → 自动降级，别让用户吃 503。
            if (Path(kb_root) / "rag_page_index.json").exists():
                results = await asyncio.get_event_loop().run_in_executor(
                    None, lambda: _page_index_search(kb_root, req.query, req.top_k))
                return {"query": req.query, "mode": "pageindex(fallback)",
                        "results": results[:req.top_k]}
            raise HTTPException(status_code=503, detail="BM25 索引不存在，请先在「索引构建」标签中触发构建。")

    try:
        if req.mode == "fusion":
            results = await asyncio.get_event_loop().run_in_executor(
                None,
                lambda: _fusion_search(kb_root, scripts_dir, req.query, req.top_k),
            )
        elif req.mode == "pageindex":
            results = await asyncio.get_event_loop().run_in_executor(
                None,
                lambda: _page_index_search(kb_root, req.query, req.top_k),
            )
        else:
            results = await asyncio.get_event_loop().run_in_executor(
                None,
                lambda: _search_subprocess(kb_root, scripts_dir, req.mode, req.query, req.top_k)
            )
        return {
            "query": req.query,
            "mode": req.mode,
            "results": results[:req.top_k],
        }
    except RuntimeError as e:
        raise HTTPException(status_code=500, detail=str(e))
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


def _detect_ollama_model() -> tuple[str, str]:
    """Return (base_url, model_name) for the best available local LLM.
    Probes Ollama at 11434/11435, picks qwen2.5:7b if available, else first model found.
    """
    import urllib.request as _ur
    preferred = ["qwen2.5:7b", "qwen2.5:14b", "qwen2.5:32b", "qwen2.5-coder:7b"]
    for port in (11434, 11435):
        base = f"http://127.0.0.1:{port}"
        try:
            with _ur.urlopen(f"{base}/api/tags", timeout=2) as r:
                data = json.loads(r.read())
            models = [m["name"] for m in data.get("models", [])]
            if not models:
                continue
            chosen = next((m for m in preferred if m in models), models[0])
            return base, chosen
        except Exception:
            continue
    return "http://127.0.0.1:11434", "qwen2.5:7b"


def _resolve_llm(provider_id: str, model: str) -> tuple[str, dict, str, str]:
    """Resolve LLM call parameters from a configured provider.

    Returns (chat_url, headers, model_name, display_label).
    Falls back to local Ollama if provider_id is empty or "ollama".
    """
    from agentpaas.api.v1.providers_api import DEFAULT_PROVIDERS, _load_providers

    # ── Ollama fallback ──────────────────────────────────────────
    if not provider_id or provider_id == "ollama":
        base, auto_model = _detect_ollama_model()
        chosen = model or auto_model
        chat_url = f"{base}/v1/chat/completions"
        return chat_url, {"Content-Type": "application/json"}, chosen, f"Ollama / {chosen}"

    pdef = DEFAULT_PROVIDERS.get(provider_id)
    if not pdef:
        raise ValueError(f"未知提供商: {provider_id}")

    saved = _load_providers()
    entry = saved.get(provider_id, {})
    base_url = (entry.get("base_url") or pdef.get("base_url", "")).rstrip("/")
    api_key = entry.get("api_key") or os.getenv(pdef.get("env_key", ""), "")
    chosen = model or (pdef["models"][0] if pdef.get("models") else "")

    ptype = pdef.get("type", "openai_compatible")

    # ── Ollama type ──────────────────────────────────────────────
    if ptype == "ollama":
        # Detect actual running port from saved/default base_url
        # but try to also auto-probe in case port changed
        try:
            auto_base, auto_model = _detect_ollama_model()
        except Exception:
            auto_base = base_url
            auto_model = chosen
        final_base = entry.get("base_url") or auto_base
        chosen = model or auto_model
        chat_url = f"{final_base.rstrip('/')}/v1/chat/completions"
        return chat_url, {"Content-Type": "application/json"}, chosen, f"Ollama / {chosen}"

    # ── claude-code type (not streamable via HTTP) ────────────────
    if ptype == "claude-code":
        raise ValueError("claude-code 提供商不支持 HTTP 流式调用，请选择其他提供商")

    # ── Anthropic native API ──────────────────────────────────────
    if ptype == "anthropic":
        if not api_key:
            raise ValueError("Anthropic API Key 未配置")
        chat_url = f"{base_url}/v1/messages"
        headers = {
            "Content-Type": "application/json",
            "x-api-key": api_key,
            "anthropic-version": "2023-06-01",
        }
        return chat_url, headers, chosen, f"Anthropic / {chosen}"

    # ── OpenAI / OpenAI-compatible ────────────────────────────────
    if not api_key:
        raise ValueError(f"提供商 {pdef['name']} 未配置 API Key")
    chat_url = f"{base_url}/chat/completions"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {api_key}",
    }
    return chat_url, headers, chosen, f"{pdef['name']} / {chosen}"


def _stream_openai_compat(
    chat_url: str, headers: dict, model: str, messages: list, temperature: float = 0.3
):
    """Generator that yields text delta strings from an OpenAI-compatible streaming endpoint."""
    import urllib.request as _ur
    body = json.dumps({
        "model": model,
        "messages": messages,
        "stream": True,
        "temperature": temperature,
    }, ensure_ascii=False).encode("utf-8")
    req = _ur.Request(chat_url, data=body, headers=headers, method="POST")
    with _ur.urlopen(req, timeout=180) as resp:
        for raw_line in resp:
            line = raw_line.decode("utf-8").strip()
            if not line.startswith("data:"):
                continue
            payload = line[5:].strip()
            if payload == "[DONE]":
                break
            try:
                d = json.loads(payload)
                delta = d["choices"][0]["delta"].get("content", "")
                if delta:
                    yield delta
            except Exception:
                continue


def _stream_anthropic(
    chat_url: str, headers: dict, model: str, messages: list, temperature: float = 0.3
):
    """Generator that yields text delta strings from Anthropic Messages streaming API."""
    import urllib.request as _ur
    # Convert OpenAI-style messages to Anthropic format
    system_parts = [m["content"] for m in messages if m["role"] == "system"]
    user_msgs = [m for m in messages if m["role"] != "system"]
    body_dict: dict = {
        "model": model,
        "messages": user_msgs,
        "max_tokens": 2048,
        "stream": True,
        "temperature": temperature,
    }
    if system_parts:
        body_dict["system"] = "\n\n".join(system_parts)
    body = json.dumps(body_dict, ensure_ascii=False).encode("utf-8")
    req = _ur.Request(chat_url, data=body, headers=headers, method="POST")
    with _ur.urlopen(req, timeout=180) as resp:
        for raw_line in resp:
            line = raw_line.decode("utf-8").strip()
            if not line.startswith("data:"):
                continue
            payload = line[5:].strip()
            try:
                d = json.loads(payload)
                if d.get("type") == "content_block_delta":
                    delta = d.get("delta", {}).get("text", "")
                    if delta:
                        yield delta
            except Exception:
                continue


@router.get("/{kb_id}/search/stream")
async def search_stream(
    kb_id: str,
    q: str = Query(..., description="Question to answer"),
    mode: str = Query(default="bm25"),
    top_k: int = Query(default=5, ge=1, le=20),
    provider_id: str = Query(default="", description="Provider ID from providers config"),
    model: str = Query(default="", description="Model name override"),
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    """SSE streaming QA: retrieve → build prompt → stream LLM answer token by token."""
    kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
    if not kb:
        raise HTTPException(status_code=404, detail="Knowledge base not found")

    kb_root = kb["root_dir"]
    scripts_dir = _find_scripts_dir(Path(kb_root))

    async def event_stream():
        # ── Step 1: retrieve ──
        if mode == "fusion":
            yield f"data: {json.dumps({'type': 'status', 'text': '正在扩写查询并多路检索（RAG Fusion）...'})}\n\n"
        else:
            yield f"data: {json.dumps({'type': 'status', 'text': '正在检索相关文档...'})}\n\n"
        await asyncio.sleep(0)

        chunks: List[Dict] = []
        index_exists = (Path(kb_root) / "rag_index.json").exists()
        page_index_exists = (Path(kb_root) / "rag_page_index.json").exists()

        if mode == "pageindex" and page_index_exists:
            try:
                chunks = await asyncio.get_event_loop().run_in_executor(
                    None,
                    lambda: _page_index_search(kb_root, q, top_k)
                )
            except Exception as e:
                yield f"data: {json.dumps({'type': 'status', 'text': f'PageIndex 检索出错: {e}'})}\n\n"
        elif scripts_dir and index_exists:
            try:
                if mode == "fusion":
                    chunks = await asyncio.get_event_loop().run_in_executor(
                        None,
                        lambda: _fusion_search(kb_root, scripts_dir, q, top_k, provider_id=provider_id, model_override=model)
                    )
                else:
                    chunks = await asyncio.get_event_loop().run_in_executor(
                        None,
                        lambda: _search_subprocess(kb_root, scripts_dir, mode, q, top_k)
                    )
            except Exception as e:
                yield f"data: {json.dumps({'type': 'status', 'text': f'检索出错: {e}'})}\n\n"
        else:
            msg = "PageIndex 不存在，请先以 page 分块策略构建索引" if mode == "pageindex" else "BM25 索引不存在，请先构建索引"
            yield f"data: {json.dumps({'type': 'status', 'text': msg})}\n\n"

        # Emit retrieved chunks so the UI can show sources panel
        yield f"data: {json.dumps({'type': 'chunks', 'chunks': chunks[:top_k]})}\n\n"
        await asyncio.sleep(0)

        if not chunks:
            yield f"data: {json.dumps({'type': 'chunk', 'text': '未检索到相关文档，无法生成回答。'})}\n\n"
            yield f"data: {json.dumps({'type': 'done'})}\n\n"
            return

        # ── Step 2: build prompt ──
        context_parts = []
        for i, c in enumerate(chunks[:top_k]):
            context_parts.append(f"[来源{i+1}: {c.get('source', '?')}]\n{c.get('text', '')}")
        context = "\n\n---\n\n".join(context_parts)[:6000]

        messages = [
            {
                "role": "system",
                "content": (
                    "你是一个知识库问答助手。请严格基于以下检索到的参考资料回答问题，"
                    "若资料中无相关信息请如实说明，回答时用「[来源N]」标注依据。"
                ),
            },
            {
                "role": "user",
                "content": f"## 参考资料\n{context}\n\n## 问题\n{q}\n\n## 回答",
            },
        ]

        # ── Step 3: resolve provider + stream ──
        yield f"data: {json.dumps({'type': 'status', 'text': '正在生成回答...'})}\n\n"
        await asyncio.sleep(0)

        try:
            chat_url, llm_headers, model_name, display_label = _resolve_llm(provider_id, model)
        except Exception as e:
            err_msg = f"提供商配置错误: {e}"
            yield f"data: {json.dumps({'type': 'chunk', 'text': err_msg})}\n\n"
            yield f"data: {json.dumps({'type': 'done'})}\n\n"
            return

        yield f"data: {json.dumps({'type': 'status', 'text': f'模型: {display_label}'})}\n\n"
        await asyncio.sleep(0)

        # Determine stream generator based on provider type
        from agentpaas.api.v1.providers_api import DEFAULT_PROVIDERS
        pdef = DEFAULT_PROVIDERS.get(provider_id, {})
        is_anthropic = pdef.get("type") == "anthropic"

        def _do_stream():
            if is_anthropic:
                yield from _stream_anthropic(chat_url, llm_headers, model_name, messages)
            else:
                yield from _stream_openai_compat(chat_url, llm_headers, model_name, messages)

        try:
            loop = asyncio.get_event_loop()
            # Use asyncio.Queue so the thread can push tokens without blocking the event loop
            # FIX-09: bounded queue prevents OOM from slow clients
            aio_q: asyncio.Queue = asyncio.Queue(maxsize=1000)

            def _fill_q():
                """Runs in a thread — pushes (kind, val) into the asyncio queue."""
                try:
                    for delta in _do_stream():
                        loop.call_soon_threadsafe(aio_q.put_nowait, ("chunk", delta))
                    loop.call_soon_threadsafe(aio_q.put_nowait, ("done", None))
                except Exception as exc:
                    loop.call_soon_threadsafe(aio_q.put_nowait, ("error", str(exc)))

            # Start the blocking stream in a thread
            stream_fut = loop.run_in_executor(None, _fill_q)

            # Drain the async queue in the event loop
            while True:
                kind, val = await aio_q.get()
                if kind == "chunk":
                    yield f"data: {json.dumps({'type': 'chunk', 'text': val})}\n\n"
                elif kind == "error":
                    err_msg = f"\n\n（LLM 错误: {val}）"
                    yield f"data: {json.dumps({'type': 'chunk', 'text': err_msg})}\n\n"
                    break
                else:  # done
                    break

            await stream_fut  # surface thread exceptions
        except Exception as e:
            err_msg = f"\n\n（LLM 错误: {e}）"
            yield f"data: {json.dumps({'type': 'chunk', 'text': err_msg})}\n\n"

        yield f"data: {json.dumps({'type': 'done'})}\n\n"

    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )


# ─── Jobs ───

@router.get("/{kb_id}/jobs")
async def list_jobs(
    kb_id: str,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    kb = db.fetchone("SELECT * FROM knowledge_bases WHERE id=? AND tenant_id=?", (kb_id, tenant.tenant_id))
    if not kb:
        raise HTTPException(status_code=404, detail="Knowledge base not found")

    jobs = db.fetchall(
        "SELECT * FROM kb_jobs WHERE kb_id=? ORDER BY created_at DESC LIMIT 20",
        (kb_id,)
    )
    return {"jobs": jobs}


@router.get("/{kb_id}/jobs/{job_id}")
async def get_job(
    kb_id: str,
    job_id: str,
    tenant: TenantContext = Depends(get_tenant),
    db: Database = Depends(get_database),
):
    job = db.fetchone("SELECT * FROM kb_jobs WHERE id=? AND kb_id=?", (job_id, kb_id))
    if not job:
        raise HTTPException(status_code=404, detail="Job not found")
    return job
