#
#  Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.
#

"""KB-wide wiki / artifact compilation.

Extracted from ``rag.svr.task_executor_refactor.task_handler`` where the
same pipeline previously lived as a set of ``_wiki_*`` / ``_persist_wiki_*``
methods and one ``_run_wiki`` orchestrator. The public entry point is
:func:`run_wiki`.

The pipeline runs MAP per (doc, template) — each MAP call resumes from
its own ``artifact_map_extract`` ES rows — then REDUCE / PLAN / REFINE
KB-wide via ``rag.advanced_rag.knowlege_compile.wiki``. Refined pages
land in ES as searchable ``artifact_page`` rows, one
``artifact_page_topic`` row per topic, and ``artifact_entity`` /
``artifact_relation`` rows for the dataset Artifact tab's canvas graph.

Design notes:

* ``load_chunks_for_doc`` is injected rather than imported to keep the
  module decoupled from ``TaskHandler``'s streaming chunk loader.
* The eligibility loop resolves each doc's
  ``parser_config.compilation_template_group_id`` to a template list
  via the shared parser-config helper and
  ``CompilationTemplateGroupService.resolve_template_ids``.
* The persistence helpers (``persist_wiki_pages_to_es`` etc.) are
  exposed at module level for testing but are only called from
  :func:`run_wiki` in production.
"""

from __future__ import annotations

import asyncio
import json
import logging
import re
from typing import AsyncIterator, Callable, Dict, List, Optional

import xxhash

from common import settings
from common.constants import LLMType
from common.misc_utils import thread_pool_exec
from rag.nlp import search
from rag.advanced_rag.knowlege_compile.wiki import (
    wiki_map_from_chunks,
    wiki_plan_from_reduction,
    wiki_reduce_from_extracts,
    wiki_refine_from_plan,
)
from rag.svr.task_executor_refactor.task_context import TaskContext


# ----- tunables ------------------------------------------------------
# Artifact-MAP tuning: how many chunks to feed per ``wiki_map_from_chunks``
# invocation. The function does its own per-call resume-set load + ES
# persist, so smaller batches mean more (small) ES round-trips but a flat
# memory footprint. 64 keeps the resume-set re-reads cheap while leaving
# room for the function's internal split_chunks packing to do real work.
WIKI_MAP_BATCH_CHUNKS = 64

# Per-node cap on ``source_chunk_ids`` carried by the canvas graph blob.
# Pages can accumulate hundreds of source chunks; the graph response is
# meant for fast canvas rendering, not full provenance audit, so we trim
# each node's list. The full per-page list is still available on the
# ``artifact_page`` row the UI deep-links into.
WIKI_GRAPH_MAX_CHUNK_IDS_PER_NODE = 64

# Title + comments stamped on every ``artifact_commit`` row produced by
# :func:`run_wiki`'s regeneration path (as opposed to the dialog-edit
# path, which carries the user's own title/comments).
WIKI_REGEN_COMMIT_TITLE = "Regenerated by artifact compilation"
WIKI_REGEN_COMMIT_COMMENTS_TEMPLATE = "Auto-update via run_wiki (action={action})"
WIKI_PAGE_COMPILE_KWD = "artifact_page"
WIKI_PAGE_TOPIC_COMPILE_KWD = "artifact_page_topic"


# ----- helpers -------------------------------------------------------


def _parser_config_compilation_template_ids(parser_config, tenant_id: str) -> list[str]:
    """Resolve a doc's ``parser_config`` to compile-template ids by
    looking up configured groups. Returns ``[]`` if no group resolves."""
    from rag.svr.task_executor_refactor.chunk_post_processor import (
        _parser_config_compilation_template_group_ids,
    )
    from api.db.services.compilation_template_group_service import (
        CompilationTemplateGroupService,
    )

    template_ids: list[str] = []
    seen: set[str] = set()
    for group_id in _parser_config_compilation_template_group_ids(parser_config):
        for template_id in CompilationTemplateGroupService.resolve_template_ids(group_id, tenant_id):
            if template_id in seen:
                continue
            seen.add(template_id)
            template_ids.append(template_id)
    return template_ids


def _wiki_topic_from_page(page: Dict, fallback: str = "") -> str:
    for key in ("topic", "title", "page_type"):
        value = page.get(key)
        if isinstance(value, str) and value.strip():
            return value.strip()
    return fallback.strip()


def _wiki_topic_slug(topic: str) -> str:
    slug = re.sub(r"[^a-z0-9]+", "-", topic.lower()).strip("-")
    if not slug:
        slug = xxhash.xxh64(topic.encode("utf-8", "surrogatepass")).hexdigest()[:12]
    return f"topic/{slug[:80]}"


async def _ensure_wiki_topic_rows(
    ctx: TaskContext,
    index: str,
    kb_id_str: str,
    topics_by_name: dict[str, str],
) -> None:
    if not topics_by_name:
        return

    from common.doc_store.doc_store_base import OrderByExpr
    from rag.nlp import rag_tokenizer

    topics = list(topics_by_name.keys())
    existing: set[str] = set()
    try:
        res = await thread_pool_exec(
            settings.docStoreConn.search,
            ["id", "topic_kwd"],
            [],
            {
                "compile_kwd": [WIKI_PAGE_TOPIC_COMPILE_KWD],
                "topic_kwd": topics,
            },
            [],
            OrderByExpr(),
            0,
            max(len(topics), 1),
            index,
            [ctx.kb_id],
        )
        field_map = settings.docStoreConn.get_fields(res, ["id", "topic_kwd"])
        for row in (field_map or {}).values():
            raw = row.get("topic_kwd")
            if isinstance(raw, list):
                existing.update(t for t in raw if isinstance(t, str) and t)
            elif isinstance(raw, str) and raw:
                existing.add(raw)
    except Exception:
        logging.exception(
            "wiki_persist: topic existence read failed for kb=%s; inserting stable topic ids",
            kb_id_str,
        )

    rows: list[dict] = []
    for topic, slug in topics_by_name.items():
        if topic in existing:
            continue
        topic_id = xxhash.xxh64(
            f"{kb_id_str}:{WIKI_PAGE_TOPIC_COMPILE_KWD}:{topic}".encode(
                "utf-8",
                "surrogatepass",
            ),
        ).hexdigest()
        content_ltks = rag_tokenizer.tokenize(topic)
        rows.append(
            {
                "id": topic_id,
                "kb_id": kb_id_str,
                "doc_id": kb_id_str,
                "compile_kwd": WIKI_PAGE_TOPIC_COMPILE_KWD,
                "topic_kwd": topic,
                "title_kwd": topic,
                "slug_kwd": slug,
                "content_with_weight": topic,
                "content_ltks": content_ltks,
                "content_sm_ltks": rag_tokenizer.fine_grained_tokenize(content_ltks),
                "available_int": 1,
            }
        )

    if rows:
        await thread_pool_exec(settings.docStoreConn.insert, rows, index, ctx.kb_id)


# ----- persistence ---------------------------------------------------


async def persist_wiki_pages_to_es(
    ctx: TaskContext,
    pages: List[Dict],
    embd_mdl,
) -> None:
    """Insert one ES row per generated artifact page using the
    knowledge-compilation schema:

      id                  xxh64(kb_id + ":" + slug)
      compile_kwd         "artifact_page"
      slug_kwd            page.slug
      title_kwd           page.title
      page_type_kwd       page.page_type
      topic_kwd           page.topic
      entity_names_kwd    page.entity_names
      outlinks_kwd        page.outlinks
      related_kb_pages_kwd page.related_kb_pages
      source_chunk_ids    page.source_chunk_ids
      source_doc_ids      page.source_doc_ids
      kb_id               ctx.kb_id
      content_with_weight rendered markdown
      content_ltks /
      content_sm_ltks     tokenize(content_md + summary)
      q_<dim>_vec         embed(summary)

    ``action`` is intentionally not stored — it's a planner artifact
    and has no meaning post-write.
    """
    if not pages:
        return

    from rag.nlp import rag_tokenizer
    from common.doc_store.doc_store_base import OrderByExpr
    from api.db.services.file_commit_service import (
        FileCommitService as WikiCommitService,
    )

    index = search.index_name(ctx.tenant_id)
    kb_id_str = str(ctx.kb_id)

    # Capture the prior rendered content for every slug we're about to
    # overwrite, so the per-page commit row downstream has a real diff
    # baseline. Single batch read by slug_kwd IN [...] — one round-trip
    # regardless of page count. Failures here degrade gracefully.
    target_slugs: list[str] = [(p.get("slug") or "").strip() for p in pages if isinstance(p.get("slug"), str) and p.get("slug")]
    prior_by_slug: dict[str, str] = {}
    if target_slugs:
        try:
            res = await thread_pool_exec(
                settings.docStoreConn.search,
                ["id", "slug_kwd", "content_with_weight"],
                [],
                {"compile_kwd": [WIKI_PAGE_COMPILE_KWD], "slug_kwd": list(target_slugs)},
                [],
                OrderByExpr(),
                0,
                max(len(target_slugs), 1),
                index,
                [ctx.kb_id],
            )
            field_map = settings.docStoreConn.get_fields(
                res,
                ["id", "slug_kwd", "content_with_weight"],
            )
            for row in (field_map or {}).values():
                s = row.get("slug_kwd")
                c = row.get("content_with_weight")
                if isinstance(s, str) and isinstance(c, str):
                    prior_by_slug[s] = c
        except Exception:
            logging.exception(
                "wiki_persist: prior-content read failed for kb=%s; commit audit will treat all pages as creations",
                kb_id_str,
            )

    # Batch the summary embeddings in one model call. Empty strings are
    # swapped for a single space so the encoder doesn't reject them —
    # they still yield a vector but contribute nothing meaningful.
    summaries = [(p.get("summary") or "").strip() for p in pages]
    embed_inputs = [s if s else " " for s in summaries]
    try:
        embeddings, _ = await thread_pool_exec(embd_mdl.encode, embed_inputs)
    except Exception:
        logging.exception(
            "wiki_persist: summary embedding batch failed for kb=%s",
            kb_id_str,
        )
        return
    try:
        n_emb = len(embeddings) if embeddings is not None else 0
    except TypeError:
        n_emb = 0
    if n_emb != len(pages):
        logging.warning(
            "artifact_persist: embedding count %d != pages %d for kb=%s; aborting",
            n_emb,
            len(pages),
            kb_id_str,
        )
        return

    rows: List[Dict] = []
    topics_by_name: dict[str, str] = {}
    for page, vec in zip(pages, embeddings):
        slug = page.get("slug") or ""
        if not slug:
            continue
        title = page.get("title") or slug
        topic = _wiki_topic_from_page(page, title)
        summary = page.get("summary") or ""
        content_md = page.get("content_md_rendered") or page.get("content_md") or page.get("content_md_raw") or ""

        vec_list = vec.tolist() if hasattr(vec, "tolist") else list(vec)
        if not vec_list:
            logging.warning(
                "artifact_persist: empty embedding for slug=%s; skipping",
                slug,
            )
            continue

        text_for_search = (content_md + "\n\n" + summary).strip()
        content_ltks = rag_tokenizer.tokenize(text_for_search) if text_for_search else ""
        content_sm_ltks = rag_tokenizer.fine_grained_tokenize(content_ltks) if content_ltks else ""

        row_id = xxhash.xxh64(
            f"{kb_id_str}:{slug}".encode("utf-8", "surrogatepass"),
        ).hexdigest()

        rows.append(
            {
                "id": row_id,
                "kb_id": kb_id_str,
                "doc_id": kb_id_str,  # sentinel; KB-scoped row, real provenance in source_doc_ids
                "compile_kwd": WIKI_PAGE_COMPILE_KWD,
                "slug_kwd": slug,
                "title_kwd": title,
                "page_type_kwd": page.get("page_type") or "concept",
                "topic_kwd": topic,
                "entity_names_kwd": list(page.get("entity_names") or []),
                "outlinks_kwd": list(page.get("outlinks") or []),
                "outlinks_int": len(list(page.get("outlinks") or [])),
                "related_kb_pages_kwd": list(page.get("related_kb_pages") or []),
                "source_chunk_ids": list(page.get("source_chunk_ids") or []),
                "source_doc_ids": list(page.get("source_doc_ids") or []),
                "content_with_weight": content_md,
                # Summary kept verbatim alongside the rendered body so the
                # viewer can render it as a distinct (smaller) block above
                # the main content.
                "summary_with_weight": summary,
                "content_ltks": content_ltks,
                "content_sm_ltks": content_sm_ltks,
                f"q_{len(vec_list)}_vec": vec_list,
                "available_int": 1,
            }
        )
        if topic:
            topics_by_name.setdefault(topic, _wiki_topic_slug(topic))

    if not rows:
        return

    try:
        await thread_pool_exec(settings.docStoreConn.insert, rows, index, ctx.kb_id)
    except Exception:
        logging.exception(
            "wiki_persist: bulk insert failed for kb=%s (rows=%d)",
            kb_id_str,
            len(rows),
        )
        return

    if topics_by_name:
        try:
            await _ensure_wiki_topic_rows(ctx, index, kb_id_str, topics_by_name)
        except Exception:
            logging.exception(
                "wiki_persist: topic row insert failed for kb=%s",
                kb_id_str,
            )

    # Audit trail: one ArtifactCommit row per page whose rendered
    # content actually changed (record_edit silently skips empty diffs).
    # Best-effort — commit failures log but don't fail the artifact
    # compile.
    for page in pages:
        slug = (page.get("slug") or "").strip()
        if not slug:
            continue
        if not prior_by_slug.get(slug, ""):
            continue
        content_md = page.get("content_md_rendered") or page.get("content_md") or page.get("content_md_raw") or ""
        action = (page.get("action") or "CREATE").upper()
        try:
            WikiCommitService.record_page_edit(
                tenant_id=ctx.tenant_id,
                kb_id=ctx.kb_id,
                page_type=page.get("page_type") or "concept",
                slug=slug,
                content_before=prior_by_slug.get(slug, ""),
                content_after=content_md,
                title=WIKI_REGEN_COMMIT_TITLE,
                comments=WIKI_REGEN_COMMIT_COMMENTS_TEMPLATE.format(action=action),
                user_id=None,  # system commit — no human author
            )
        except Exception:
            logging.exception(
                "wiki_persist: commit record failed for kb=%s slug=%s",
                kb_id_str,
                slug,
            )


def build_wiki_page_graph(
    pages: List[Dict],
    kb_id: str,
) -> tuple[List[Dict], List[Dict]]:
    """Project the REFINE-emitted page list onto per-entity and
    per-relation ES rows.

    Returns:
        (entity_rows, relation_rows)

    Both lists are ES-ready docs (one per node / per surviving edge)
    using the standard artifact envelope. They are BM25-only (no
    ``q_<dim>_vec``) — entities carry ``content_ltks`` derived from
    ``slug + " " + summary`` so name/summary lookups hit the lexical
    index.

    ``by_slug`` is kept internally only to filter dangling outlinks
    (a target slug not present as a node in this KB).
    """
    from rag.nlp import rag_tokenizer

    by_slug: Dict[str, Dict] = {}
    entity_rows: List[Dict] = []
    for p in pages or []:
        slug = (p.get("slug") or "").strip()
        if not slug:
            continue
        outlinks_raw = p.get("outlinks") or []
        # ``weight`` is the page's outlink count. Drives node size on
        # the canvas. Computed on the raw outlink list before dangling-
        # target filtering so visual weight reflects what the writer
        # actually emitted.
        weight = len(outlinks_raw) if isinstance(outlinks_raw, list) else 0

        # Per-node provenance: union of source chunks REFINE attributed
        # to this page. Dedup preserves first-seen order; cap the list
        # to keep the graph blob small.
        raw_chunk_ids = p.get("source_chunk_ids") or []
        seen_chunk_ids: dict[str, None] = {}
        for cid in raw_chunk_ids:
            if isinstance(cid, str) and cid and cid not in seen_chunk_ids:
                seen_chunk_ids[cid] = None
                if len(seen_chunk_ids) >= WIKI_GRAPH_MAX_CHUNK_IDS_PER_NODE:
                    break
        capped_chunk_ids = list(seen_chunk_ids.keys())

        page_type = p.get("page_type") or "concept"
        description = p.get("summary") or ""
        name = p.get("title") or slug
        aliases = list(p.get("entity_names") or [])

        by_slug[slug] = {
            "slug": slug,
            "name": name,
            "aliases": aliases,
            "description": description,
            "type": page_type,
            "weight": weight,
            "source_chunk_ids": capped_chunk_ids,
        }

        # Per-entity ES row. content_ltks is built from slug + summary
        # so BM25 hits both the deep-link key and human prose.
        content_text = (slug + " " + description).strip()
        entity_payload = {
            "slug": slug,
            "name": name,
            "aliases": aliases,
            "description": description,
            "type": page_type,
            "weight": weight,
        }
        entity_rows.append(
            {
                "id": xxhash.xxh64(
                    f"artifact_entity:{kb_id}:{slug}".encode("utf-8", "surrogatepass"),
                ).hexdigest(),
                "kb_id": kb_id,
                "doc_id": kb_id,  # KB-scoped sentinel
                "available_int": 1,
                "compile_kwd": "artifact_entity",
                "type_kwd": "artifact_" + page_type,
                "slug_kwd": slug,
                "weight_int": int(weight),
                "source_chunk_ids": capped_chunk_ids,
                "content_ltks": rag_tokenizer.tokenize(content_text) if content_text else "",
                "content_with_weight": json.dumps(entity_payload, ensure_ascii=False),
            }
        )

    relation_rows: List[Dict] = []
    for p in pages or []:
        src = (p.get("slug") or "").strip()
        if not src or src not in by_slug:
            continue
        for raw_target in p.get("outlinks") or []:
            if isinstance(raw_target, str):
                tgt = raw_target.strip()
            elif isinstance(raw_target, dict):
                tgt = str(raw_target.get("slug") or "").strip()
            else:
                tgt = ""
            if not tgt or tgt == src or tgt not in by_slug:
                continue
            relation_payload = {"from": src, "to": tgt}
            relation_rows.append(
                {
                    "id": xxhash.xxh64(
                        f"artifact_relation:{kb_id}:{src}:{tgt}".encode("utf-8", "surrogatepass"),
                    ).hexdigest(),
                    "kb_id": kb_id,
                    "doc_id": kb_id,
                    "available_int": 1,
                    "compile_kwd": "artifact_relation",
                    "type_kwd": "artifact_relation",
                    "from_kwd": src,
                    "to_kwd": tgt,
                    "content_with_weight": json.dumps(relation_payload, ensure_ascii=False),
                }
            )

    return entity_rows, relation_rows


async def persist_wiki_page_graph_to_es(
    ctx: TaskContext,
    pages: List[Dict],
) -> None:
    """Materialize and store the per-entity / per-relation ES rows
    derived from artifact pages.

    Writes two row types — both delete-then-insert for idempotent
    re-runs:

    1. ``compile_kwd="artifact_entity"`` — one row per page node,
       BM25-only via ``content_ltks``.
    2. ``compile_kwd="artifact_relation"`` — one row per surviving
       edge (dangling outlinks dropped by the builder).

    Also sweeps any leftover legacy ``artifact_page_graph`` blob so
    the index doesn't accumulate stale state.
    """
    kb_id_str = str(ctx.kb_id)
    entity_rows, relation_rows = build_wiki_page_graph(pages or [], kb_id_str)

    index = search.index_name(ctx.tenant_id)

    async def _replace_bucket(kwd: str, rows: List[Dict]) -> None:
        try:
            await thread_pool_exec(
                settings.docStoreConn.delete,
                {"compile_kwd": kwd},
                index,
                ctx.kb_id,
            )
        except Exception:
            logging.debug(
                "%s: prior delete failed; relying on id-upsert",
                kwd,
            )
        if not rows:
            return
        try:
            await thread_pool_exec(
                settings.docStoreConn.insert,
                rows,
                index,
                ctx.kb_id,
            )
        except Exception:
            logging.exception(
                "%s: insert failed for kb=%s (%d rows)",
                kwd,
                kb_id_str,
                len(rows),
            )

    async def _sweep_legacy_blob() -> None:
        try:
            await thread_pool_exec(
                settings.docStoreConn.delete,
                {"compile_kwd": "artifact_page_graph"},
                index,
                ctx.kb_id,
            )
        except Exception:
            logging.debug(
                "artifact_page_graph: legacy blob sweep failed for kb=%s",
                kb_id_str,
            )

    await asyncio.gather(
        _replace_bucket("artifact_entity", entity_rows),
        _replace_bucket("artifact_relation", relation_rows),
        _sweep_legacy_blob(),
    )


# ----- main entry ----------------------------------------------------


async def run_wiki(
    ctx: TaskContext,
    embedding_model,
    load_chunks_for_doc: Callable[..., AsyncIterator[list[dict]]],
) -> None:
    """KB-wide artifact compilation task.

    Runs after the user clicks the "Artifact" button in the dataset
    generate menu. Iterates every doc in the KB whose parser_config
    has a compilation template group resolving to an artifacts-kind
    child, runs MAP per-doc (which uses ES-stored resume rows to skip
    chunks already processed in a previous run), then runs REDUCE /
    PLAN / REFINE KB-wide and persists pages.

    Batching: each MAP call uses ``batch_size_cap=8`` and
    ``window_fraction=0.5`` — i.e. roll over to a new batch when the
    current batch reaches 8 chunks OR its accumulated token count
    exceeds 50% of the chat model's ``max_length``.
    """
    # Local imports so this module doesn't drag in the API service
    # layer at import time — that's a source of circular-import risk
    # given how much lives under ``api.db.services``.
    from api.db.services.document_service import DocumentService
    from api.db.services.knowledgebase_service import KnowledgebaseService
    from api.db.services.compilation_template_service import CompilationTemplateService
    from api.db.services.llm_service import LLMBundle
    from api.db.joint_services.tenant_model_service import (
        get_tenant_default_model_by_type,
        resolve_model_config,
    )
    from api.apps.restful_apis.chunk_api import _compilation_template_kind

    progress = ctx.progress_cb
    progress(0.0, "Loading documents for wiki compilation...")

    # 1. Resolve KB metadata for PLAN.
    ok, kb = KnowledgebaseService.get_by_id(ctx.kb_id)
    if not ok:
        progress(-1, f"KB {ctx.kb_id} not found.")
        return
    kb_name = kb.name
    kb_description = kb.description

    # 2. Pick docs eligible for artifact compilation (those whose
    # configured template group resolves to at least one artifacts-kind
    # child). The frontend Artifact button targets the KB, but the
    # per-doc opt-in is what gates inclusion.
    all_docs, _ = await thread_pool_exec(
        DocumentService.get_by_kb_id,
        kb_id=ctx.kb_id,
        page_number=0,
        items_per_page=0,
        orderby="create_time",
        desc=False,
        keywords="",
        run_status=[],
        types=[],
        suffix=[],
    )
    eligible = []
    for d in all_docs or []:
        pc = d.get("parser_config") or {}
        for template_id in _parser_config_compilation_template_ids(pc, ctx.tenant_id):
            template = CompilationTemplateService.get_saved(template_id, ctx.tenant_id)
            config = template.get("config") if template else {}
            kind = _compilation_template_kind(config.get("kind") if isinstance(config, dict) else "")
            if kind == "artifacts":
                eligible.append((d, template_id))
                break
    if not eligible:
        progress(1.0, "No documents are configured for wiki compilation.")
        return

    # 3. Resolve chat models. MAP is per-(doc, template) so each pair
    # uses its template's own ``llm_id``. REDUCE / PLAN / REFINE are
    # KB-wide and need exactly one model — we pick the first eligible
    # template's ``llm_id`` as the canonical KB chat model.
    llm_bundle_cache: dict[str, LLMBundle] = {}

    def _bundle_for(llm_id: str | None) -> LLMBundle:
        key = (llm_id or "").strip() or "__tenant_default__"
        cached = llm_bundle_cache.get(key)
        if cached is not None:
            return cached
        try:
            if key == "__tenant_default__":
                cfg = get_tenant_default_model_by_type(ctx.tenant_id, LLMType.CHAT)
            else:
                cfg = resolve_model_config(
                    ctx.tenant_id,
                    LLMType.CHAT,
                    key,
                )
        except Exception:
            logging.exception(
                "wiki: chat model resolution failed for llm_id=%s (kb=%s); falling back to tenant default",
                key,
                ctx.kb_id,
            )
            cfg = get_tenant_default_model_by_type(ctx.tenant_id, LLMType.CHAT)
            key = "__tenant_default__"
            cached = llm_bundle_cache.get(key)
            if cached is not None:
                return cached
        bundle = LLMBundle(ctx.tenant_id, cfg, lang=ctx.language)
        llm_bundle_cache[key] = bundle
        return bundle

    def _stage_cb(prefix: str):
        def _cb(*args, **kwargs):
            try:
                if args and isinstance(args[0], (int, float)):
                    msg = args[1] if len(args) > 1 else kwargs.get("msg", "")
                    progress(msg=f"{prefix} {msg}")
                else:
                    msg = kwargs.get("msg") or (args[0] if args else "")
                    progress(msg=f"{prefix} {msg}")
            except Exception:
                logging.exception("wiki: progress callback failed")

        return _cb

    # 4. MAP per eligible doc. Each MAP call's own resume mechanism
    # (artifact_map_extract rows keyed by chunk_id) skips chunks that
    # were already processed in a prior run — this is the incremental
    # behavior the user asked for.
    #
    # ``kb_chat_llm_id`` is captured from the first eligible template
    # and used as the canonical chat model for REDUCE/PLAN/REFINE.
    # ``kb_writer_example`` follows the same first-template-wins rule:
    # the REFINE writer's page-structure section is pulled from the
    # first eligible artifact template's ``parser_config.example``
    # override (None falls back to the built-in ``WIKI_TEMPLATE_EXAMPLE``).
    kb_chat_llm_id: Optional[str] = None
    kb_writer_example: Optional[str] = None
    n_docs = len(eligible)
    for i, (doc, template_id) in enumerate(eligible):
        doc_id = doc["id"]
        progress(
            0.05 + 0.6 * (i / n_docs),
            f"MAP {i + 1}/{n_docs}: {doc.get('name', doc_id)}",
        )

        template = CompilationTemplateService.get_saved(template_id, ctx.tenant_id)
        if not template:
            logging.warning(
                "artifact: template %s not found for doc %s; skipping",
                template_id,
                doc_id,
            )
            continue
        parser_cfg = template.get("config") or {}

        map_llm_id = (parser_cfg.get("llm_id") or "").strip() if isinstance(parser_cfg, dict) else ""
        map_chat_mdl = _bundle_for(map_llm_id)
        if kb_chat_llm_id is None:
            # First eligible template wins — canonical for KB-wide
            # REDUCE/PLAN/REFINE further down.
            kb_chat_llm_id = map_llm_id or None
            tmpl_example = parser_cfg.get("example") if isinstance(parser_cfg, dict) else None
            if isinstance(tmpl_example, str) and tmpl_example.strip():
                kb_writer_example = tmpl_example

        # Stream the doc's chunks in batches and call MAP per batch so
        # peak memory stays bounded for long docs.
        agg = {"entities": 0, "concepts": 0, "claims": 0, "relations": 0}
        agg_delta = {"new": 0, "changed": 0, "unchanged": 0, "deleted": 0}
        doc_had_delta = False
        saw_any = False
        batch_no = 0
        async for batch in load_chunks_for_doc(
            ctx.tenant_id,
            ctx.kb_id,
            doc_id,
            batch_size=WIKI_MAP_BATCH_CHUNKS,
        ):
            saw_any = True
            batch_no += 1
            try:
                phase1 = await wiki_map_from_chunks(
                    chunks=batch,
                    chat_mdl=map_chat_mdl,
                    embd_mdl=embedding_model,
                    doc_id=doc_id,
                    tenant_id=ctx.tenant_id,
                    kb_id=ctx.kb_id,
                    language=ctx.language,
                    callback=_stage_cb(f"[wiki MAP {i + 1}/{n_docs} b{batch_no}]"),
                    parser_config=parser_cfg,
                    batch_size_cap=8,
                    window_fraction=0.5,
                )
            except Exception:
                logging.exception(
                    "wiki: MAP failed for doc %s batch %d",
                    doc_id,
                    batch_no,
                )
                continue
            for k in agg.keys():
                agg[k] += len(phase1.get(k) or [])
            meta = phase1.get("_meta") or {}
            if isinstance(meta, dict):
                for k in agg_delta.keys():
                    agg_delta[k] += int(meta.get(k, 0) or 0)
                if meta.get("had_delta"):
                    doc_had_delta = True

        if not saw_any:
            logging.info("wiki: no chunks for doc %s; skipping", doc_id)
            continue
        logging.info(
            "wiki: MAP doc=%s entities=%d concepts=%d claims=%d relations=%d (batches=%d, new=%d changed=%d unchanged=%d deleted=%d, delta=%s)",
            doc_id,
            agg["entities"],
            agg["concepts"],
            agg["claims"],
            agg["relations"],
            batch_no,
            agg_delta["new"],
            agg_delta["changed"],
            agg_delta["unchanged"],
            agg_delta["deleted"],
            doc_had_delta,
        )

    # 5. REDUCE / PLAN / REFINE KB-wide. Each phase has its own
    # input_hash gate (REDUCE keys off the MAP-state hash, PLAN off
    # REDUCE's hash, REFINE off PLAN's hash) so re-runs without an
    # upstream delta short-circuit at the cache layer.
    kb_chat_mdl = _bundle_for(kb_chat_llm_id)
    try:
        progress(0.65, "Reducing extracts KB-wide...")
        await wiki_reduce_from_extracts(
            chat_mdl=kb_chat_mdl,
            embd_mdl=embedding_model,
            tenant_id=ctx.tenant_id,
            kb_id=ctx.kb_id,
            callback=_stage_cb("[wiki REDUCE]"),
        )

        progress(0.75, "Planning wiki pages...")
        await wiki_plan_from_reduction(
            chat_mdl=kb_chat_mdl,
            embd_mdl=embedding_model,
            tenant_id=ctx.tenant_id,
            kb_id=ctx.kb_id,
            kb_name=kb_name,
            kb_description=kb_description,
            callback=_stage_cb("[wiki PLAN]"),
        )

        progress(0.85, "Refining pages...")
        pages = await wiki_refine_from_plan(
            chat_mdl=kb_chat_mdl,
            embd_mdl=embedding_model,
            tenant_id=ctx.tenant_id,
            kb_id=ctx.kb_id,
            callback=_stage_cb("[wiki REFINE]"),
            example=kb_writer_example,
        )
    except Exception:
        logging.exception("wiki: REDUCE/PLAN/REFINE failed for kb %s", ctx.kb_id)
        progress(-1, "Wiki pipeline failed during REDUCE/PLAN/REFINE.")
        return

    # 6. Persist searchable artifact_page rows.
    try:
        await persist_wiki_pages_to_es(ctx, pages or [], embedding_model)
    except Exception:
        logging.exception("wiki: ES persist failed for kb %s", ctx.kb_id)

    # 7. Materialize the canvas graph from the refined pages.
    try:
        await persist_wiki_page_graph_to_es(ctx, pages or [])
    except Exception:
        logging.exception("wiki: page-graph persist failed for kb %s", ctx.kb_id)

    progress(1.0, f"Wiki compiled {len(pages or [])} page(s).")
