From 3f8a3dfcff7696872aba983e758fa8a17415a05f Mon Sep 17 00:00:00 2001
From: Kevin Hu <kevinhu.sh@gmail.com>
Date: Thu, 30 Jul 2026 18:39:01 +0800
Subject: [PATCH] Feat: Compilation benefit naive rag. (#17555)

### Summary

Compilation benifit naive rag

---------

Co-authored-by: Yingfeng Zhang <yingfeng.zhang@gmail.com>
---
 api/apps/restful_apis/chunk_api.py            |   2 +-
 api/apps/restful_apis/dataset_api.py          |  17 +-
 .../utils/compilation_template_validation.py  |   2 +-
 api/apps/services/dataset_api_service.py      | 402 ++++++++++-----
 .../init_data/compilation_templates/wiki.yaml |   6 +-
 .../compilation_template_group_service.py     |   4 +-
 api/db/services/document_service.py           |   2 +-
 common/constants.py                           |   4 +-
 .../harness/orchestrator/agentic.py           |   2 +-
 .../harness/orchestrator/direct.py            |   4 +-
 rag/advanced_rag/harness/tools/navigation.py  | 169 +++++--
 rag/advanced_rag/harness/tools/search.py      | 473 +++++++++++++++++-
 rag/advanced_rag/knowlege_compile/runner.py   |   2 +-
 rag/svr/task_executor.py                      |   4 +-
 .../dataset_wiki_generator.py                 |   2 +-
 .../task_executor_refactor/task_handler.py    |   2 +-
 16 files changed, 901 insertions(+), 196 deletions(-)

diff --git a/api/apps/restful_apis/chunk_api.py b/api/apps/restful_apis/chunk_api.py
index 63e7a3dff7..c639521624 100644
--- a/api/apps/restful_apis/chunk_api.py
+++ b/api/apps/restful_apis/chunk_api.py
@@ -637,7 +637,7 @@ async def get_document_structure_graph(tenant_id, dataset_id, document_id):
             config = template.get("config") if isinstance(template.get("config"), dict) else {}
             raw_kind = (config.get("kind") if isinstance(config, dict) else "") or template.get("kind") or ""
             kind_norm = _compilation_template_kind(raw_kind)
-            if kind_norm == "artifacts":
+            if kind_norm == "wiki":
                 continue
             seen_configured_ids.add(template_id)
             configured_ids.append(template_id)
diff --git a/api/apps/restful_apis/dataset_api.py b/api/apps/restful_apis/dataset_api.py
index 732d3df4f2..eaaf0d44d3 100644
--- a/api/apps/restful_apis/dataset_api.py
+++ b/api/apps/restful_apis/dataset_api.py
@@ -792,16 +792,21 @@ def delete_dataset_structure(tenant_id, dataset_id):
 @login_required
 @add_tenant_id_to_kwargs
 async def get_wiki_alteration(tenant_id, dataset_id):
-    """Return document drift for the dataset Artifact wiki.
+    """Return document drift for a compiled dataset product.
 
-    GET /api/v1/datasets/<dataset_id>/artifacts/alteration
+    GET /api/v1/datasets/<dataset_id>/artifacts/alteration?kind=<kind>
+    ``kind`` (default ``wiki``) is one of: wiki | graph | mindmap | timeline |
+    tree (tree covers both ``tree`` and ``page_index``).
     Success: {"code": 0, "data": {"removed": int, "newly_uploaded": int, ...}}
     """
+    kind = (request.args.get("kind") or "wiki").strip().lower()
+    if kind not in {"wiki", "graph", "mindmap", "timeline", "tree"}:
+        return get_error_data_result(message=f"Unsupported kind: {kind!r}. Expected one of: wiki, graph, mindmap, timeline, tree.")
     try:
-        success, result = await dataset_api_service.get_wiki_alteration(
-            dataset_id,
-            tenant_id,
-        )
+        if kind == "wiki":
+            success, result = await dataset_api_service.get_wiki_alteration(dataset_id, tenant_id)
+        else:
+            success, result = await dataset_api_service.get_structure_alteration(dataset_id, tenant_id, kind)
         if success:
             return get_result(data=result)
         return get_result(data=False, message=result, code=RetCode.AUTHENTICATION_ERROR)
diff --git a/api/apps/restful_apis/utils/compilation_template_validation.py b/api/apps/restful_apis/utils/compilation_template_validation.py
index 0ee378fcc8..1ac3b6a3c6 100644
--- a/api/apps/restful_apis/utils/compilation_template_validation.py
+++ b/api/apps/restful_apis/utils/compilation_template_validation.py
@@ -56,7 +56,7 @@ def validate_template_payload(req: dict, require_all: bool = True) -> str:
                     return f"{section.capitalize()} field description is too long."
                 if len(str((field or {}).get("rule") or "")) > 1024:
                     return f"{section.capitalize()} field rule is too long."
-        if config.get("kind") == "artifacts" or req.get("kind") == "artifacts":
+        if config.get("kind") == "wiki" or req.get("kind") == "wiki":
             for field in (config.get("claim") or {}).get("fields") or []:
                 if not str((field or {}).get("statement") or "").strip():
                     return "Claim statement is required."
diff --git a/api/apps/services/dataset_api_service.py b/api/apps/services/dataset_api_service.py
index 52965bbaf2..264f6dccfb 100644
--- a/api/apps/services/dataset_api_service.py
+++ b/api/apps/services/dataset_api_service.py
@@ -50,13 +50,13 @@ _STRUCTURE_INDEX_TYPE_TO_KIND = {
 }
 _STRUCTURE_INDEX_TYPES = frozenset(_STRUCTURE_INDEX_TYPE_TO_KIND)
 
-_VALID_INDEX_TYPES = {"graph", "raptor", "mindmap", "artifact", "skill"} | set(_STRUCTURE_INDEX_TYPES)
+_VALID_INDEX_TYPES = {"graph", "raptor", "mindmap", "wiki", "skill"} | set(_STRUCTURE_INDEX_TYPES)
 
 _INDEX_TYPE_TO_TASK_TYPE = {
     "graph": "structure_graph",
     "raptor": "raptor",
     "mindmap": "structure_mindmap",
-    "artifact": "artifact",
+    "wiki": "wiki",
     "skill": "skill",
     # Structure merge types carry their own task_type (== index_type) so the
     # executor can resolve which kind to merge from the task body.
@@ -67,7 +67,7 @@ _INDEX_TYPE_TO_TASK_ID_FIELD = {
     "graph": "graphrag_task_id",
     "raptor": "raptor_task_id",
     "mindmap": "mindmap_task_id",
-    "artifact": "wiki_task_id",
+    "wiki": "wiki_task_id",
     "skill": "skill_task_id",
     **{t: f"{t}_task_id" for t in _STRUCTURE_INDEX_TYPES},
 }
@@ -76,7 +76,7 @@ _INDEX_TYPE_TO_DISPLAY_NAME = {
     "graph": "Graph",
     "raptor": "RAPTOR",
     "mindmap": "Mindmap",
-    "artifact": "Artifact",
+    "wiki": "Wiki",
     "skill": "Skill",
     "structure_graph": "Structure Graph",
     "structure_mindmap": "Structure Mindmap",
@@ -1614,40 +1614,53 @@ def _extract_pipeline_compiler_group_ids(dsl) -> list[str]:
     return group_ids
 
 
-def _template_is_wiki(template: dict | None) -> bool:
+def _normalize_template_kind(raw) -> str:
+    """Lowercase a template's raw kind WITHOUT folding.
+
+    Unlike :func:`_compilation_template_kind` (which folds ``page_index`` /
+    ``knowledge_graph`` into ``timeline``), this keeps kinds distinct so the
+    alteration API can tell ``graph`` / ``timeline`` / ``tree`` apart.
+    """
+    if not isinstance(raw, str):
+        return ""
+    return raw.strip().lower().replace("-", "_")
+
+
+def _template_matches_kind(template: dict | None, accepted: set[str]) -> bool:
     if not isinstance(template, dict):
         return False
     config = template.get("config") if isinstance(template.get("config"), dict) else {}
     raw_kind = config.get("kind") or template.get("kind") or ""
-    return _compilation_template_kind(raw_kind) == "artifacts"
+    return _normalize_template_kind(raw_kind) in accepted
 
 
-def _group_has_wiki_template(group_id: str, tenant_id: str, group_cache: dict[str, bool]) -> bool:
+def _group_has_template_of_kind(group_id: str, tenant_id: str, accepted: set[str], group_cache: dict[str, bool]) -> bool:
     if group_id in group_cache:
         return group_cache[group_id]
     from api.db.services.compilation_template_group_service import CompilationTemplateGroupService
 
     group = CompilationTemplateGroupService.get_saved(group_id, tenant_id)
-    has_wiki = any(_template_is_wiki(template) for template in (group or {}).get("templates") or [])
-    group_cache[group_id] = has_wiki
-    return has_wiki
+    matched = any(_template_matches_kind(template, accepted) for template in (group or {}).get("templates") or [])
+    group_cache[group_id] = matched
+    return matched
 
 
-def _parser_config_has_wiki_template(parser_config, tenant_id: str, template_cache: dict[str, bool]) -> bool:
+def _parser_config_has_template_of_kind(parser_config, tenant_id: str, accepted: set[str], template_cache: dict[str, bool]) -> bool:
     from api.db.services.compilation_template_service import CompilationTemplateService
     from rag.svr.task_executor_refactor.chunk_post_processor import _parser_config_compilation_template_ids
 
     for template_id in _parser_config_compilation_template_ids(parser_config, tenant_id):
         if template_id not in template_cache:
-            template_cache[template_id] = _template_is_wiki(CompilationTemplateService.get_saved(template_id, tenant_id))
+            template_cache[template_id] = _template_matches_kind(CompilationTemplateService.get_saved(template_id, tenant_id), accepted)
         if template_cache[template_id]:
             return True
     return False
 
 
-def _pipeline_has_wiki_compiler(
+def _pipeline_has_compiler_of_kind(
     pipeline_id: str,
     tenant_id: str,
+    accepted: set[str],
     pipeline_cache: dict[str, bool],
     group_cache: dict[str, bool],
 ) -> bool:
@@ -1665,9 +1678,9 @@ def _pipeline_has_wiki_compiler(
         return False
 
     group_ids = _extract_pipeline_compiler_group_ids(getattr(canvas, "dsl", None))
-    has_wiki = any(_group_has_wiki_template(group_id, tenant_id, group_cache) for group_id in group_ids)
-    pipeline_cache[pipeline_id] = has_wiki
-    return has_wiki
+    matched = any(_group_has_template_of_kind(group_id, tenant_id, accepted, group_cache) for group_id in group_ids)
+    pipeline_cache[pipeline_id] = matched
+    return matched
 
 
 def _skill_index_or_none(tenant_id: str, kb_id: str):
@@ -1841,48 +1854,63 @@ async def get_dataset_structure(dataset_id: str, tenant_id: str, kind: str, keyw
     # fields only) to enumerate the distinct template ids whose top-level kind
     # matches the request; ``build_bucket`` below then reads each template's rows.
     meta_fields = ["id", "compile_kwd", "compilation_template_ids", "compilation_template_kind_kwd"]
-    kind_template_ids: list[str] = []
-    seen_tid: set[str] = set()
-    has_templateless = False
-    offset = 0
     page_size = 1000
-    pages = 0
-    while True:
-        pages += 1
-        try:
-            res = await thread_pool_exec(
-                settings.docStoreConn.search,
-                meta_fields,
-                [],
-                {"knowledge_graph_kwd": ["entity", "relation"], "scope_kwd": ["dataset"]},
-                [],
-                OrderByExpr(),
-                offset,
-                page_size,
-                index_nm,
-                [dataset_id],
-            )
-            meta_rows = settings.docStoreConn.get_fields(res, meta_fields) or {}
-        except Exception:
-            logging.exception("get_dataset_structure: docStore discovery failed for kb=%s", dataset_id)
-            return True, empty
-        if not meta_rows:
-            break
-        for row in meta_rows.values():
-            tid = _row_template_id(row)
-            stamped_kind = (row.get("compilation_template_kind_kwd") or "").strip()
-            row_kind = stamped_kind or _template_meta(tid) or ""
-            if _resolve_dataset_structure_kind(row_kind) != resolved_kind:
-                continue
-            if tid:
-                if tid not in seen_tid:
-                    seen_tid.add(tid)
-                    kind_template_ids.append(tid)
-            else:
-                has_templateless = True
-        if len(meta_rows) < page_size:
-            break
-        offset += page_size
+
+    async def _discover_scope_templates(scope_kwd: str) -> tuple[list[str], bool, int]:
+        scope_template_ids: list[str] = []
+        seen_scope_tid: set[str] = set()
+        scope_has_templateless = False
+        offset = 0
+        pages = 0
+        while True:
+            pages += 1
+            try:
+                res = await thread_pool_exec(
+                    settings.docStoreConn.search,
+                    meta_fields,
+                    [],
+                    {"knowledge_graph_kwd": ["entity", "relation"], "scope_kwd": [scope_kwd]},
+                    [],
+                    OrderByExpr(),
+                    offset,
+                    page_size,
+                    index_nm,
+                    [dataset_id],
+                )
+                meta_rows = settings.docStoreConn.get_fields(res, meta_fields) or {}
+            except Exception:
+                logging.exception("get_dataset_structure: docStore discovery failed for kb=%s scope=%s", dataset_id, scope_kwd)
+                return [], False, pages
+            if not meta_rows:
+                break
+            for row in meta_rows.values():
+                tid = _row_template_id(row)
+                stamped_kind = row.get("compilation_template_kind_kwd") or ""
+                if isinstance(stamped_kind, list):
+                    stamped_kind = stamped_kind[0].strip()
+                row_kind = stamped_kind or _template_meta(tid) or ""
+                if _resolve_dataset_structure_kind(row_kind) != resolved_kind:
+                    continue
+                if tid:
+                    if tid not in seen_scope_tid:
+                        seen_scope_tid.add(tid)
+                        scope_template_ids.append(tid)
+                else:
+                    scope_has_templateless = True
+            if len(meta_rows) < page_size:
+                break
+            offset += page_size
+        return scope_template_ids, scope_has_templateless, pages
+
+    dataset_template_ids, has_templateless, dataset_pages = await _discover_scope_templates("dataset")
+    kind_template_ids: list[str] = list(dataset_template_ids)
+    template_scope_by_id: dict[str, str] = {tid: "dataset" for tid in dataset_template_ids}
+
+    doc_template_ids, _, doc_pages = await _discover_scope_templates("doc")
+    for tid in doc_template_ids:
+        if tid not in template_scope_by_id:
+            template_scope_by_id[tid] = "doc"
+            kind_template_ids.append(tid)
 
     # Detect datasets that have ONLY the legacy dataset_graph blob (no
     # entity/relation rows yet) so the fallback path below handles them.
@@ -1907,9 +1935,11 @@ async def get_dataset_structure(dataset_id: str, tenant_id: str, kind: str, keyw
             pass
 
     logging.debug(
-        "get_dataset_structure: discovered %d template(s) in %d page(s) for kb=%s kind=%s",
-        len(kind_template_ids),
-        pages,
+        "get_dataset_structure: discovered %d dataset template(s) and %d doc template(s) in %d/%d page(s) for kb=%s kind=%s",
+        len(dataset_template_ids),
+        len(doc_template_ids),
+        dataset_pages,
+        doc_pages,
         dataset_id,
         resolved_kind,
     )
@@ -1927,17 +1957,20 @@ async def get_dataset_structure(dataset_id: str, tenant_id: str, kind: str, keyw
 
         def _scope_for_template(row: dict):
             tid = _row_template_id(row) or ""
-            stamped = (row.get("compilation_template_kind_kwd") or "").strip()
+            stamped = row.get("compilation_template_kind_kwd") or ""
+            if isinstance(stamped, list):
+                stamped = stamped[0].strip()
+            scope_kwd = template_scope_by_id.get(tid, "dataset") if tid else "dataset"
             meta = _bucket_meta_for(tid, stamped) if tid else {"template_id": f"kind:{resolved_kind}", "template_name": f"kind:{resolved_kind}", "kind": resolved_kind}
             if tid:
-                return meta, {"compilation_template_ids": [tid], "scope_kwd": ["dataset"]}
-            return meta, {"compilation_template_kind_kwd": [stamped], "scope_kwd": ["dataset"]}
+                return meta, {"compilation_template_ids": [tid], "scope_kwd": [scope_kwd]}
+            return meta, {"compilation_template_kind_kwd": [stamped], "scope_kwd": [scope_kwd]}
 
         bucket_meta, kw_entities, kw_relations = await sgc.keyword_subgraph(
             index_nm,
             dataset_id,
             embd_mdl,
-            {"compilation_template_ids": kind_template_ids, "knowledge_graph_kwd": ["entity"], "scope_kwd": ["dataset"]},
+            {"compilation_template_ids": kind_template_ids, "knowledge_graph_kwd": ["entity"], "scope_kwd": ["dataset", "doc"]},
             keywords,
             _scope_for_template,
             log_ctx=f"kb={dataset_id}",
@@ -1956,11 +1989,18 @@ async def get_dataset_structure(dataset_id: str, tenant_id: str, kind: str, keyw
     # ── normal mode: per-template subgraph sampling from raw KB-wide rows. ──
     templates_out: list[dict] = []
     for tid in kind_template_ids:
+        scope_kwd = template_scope_by_id.get(tid, "dataset")
         try:
-            entities, relations = await sgc.build_bucket(index_nm, dataset_id, {"compilation_template_ids": [tid], "scope_kwd": ["dataset"]})
+            entities, relations = await sgc.build_bucket(index_nm, dataset_id, {"compilation_template_ids": [tid], "scope_kwd": [scope_kwd]})
         except Exception:
             logging.exception("get_dataset_structure: bucket build failed for kb=%s template=%s", dataset_id, tid)
             continue
+        if scope_kwd == "dataset" and not entities and not relations:
+            try:
+                entities, relations = await sgc.build_bucket(index_nm, dataset_id, {"compilation_template_ids": [tid], "scope_kwd": ["doc"]})
+            except Exception:
+                logging.exception("get_dataset_structure: doc fallback bucket build failed for kb=%s template=%s", dataset_id, tid)
+                continue
         if resolved_kind in {"knowledge_graph", "mind_map", "timeline"}:
             entities = sgc.filter_entities_with_relations(entities, relations)
             if not entities:
@@ -1993,7 +2033,10 @@ async def get_dataset_structure(dataset_id: str, tenant_id: str, kind: str, keyw
             legacy_rows = {}
         legacy_bucket = {"template_id": f"kind:{resolved_kind}", "template_name": f"kind:{resolved_kind}", "kind": resolved_kind, "entities": [], "relations": []}
         for row in legacy_rows.values():
-            if _resolve_dataset_structure_kind((row.get("compilation_template_kind_kwd") or "").strip()) != resolved_kind:
+            stamped_kind = row.get("compilation_template_kind_kwd") or ""
+            if isinstance(stamped_kind, list):
+                stamped_kind = stamped_kind[0].strip()
+            if _resolve_dataset_structure_kind((stamped_kind or "").strip()) != resolved_kind:
                 continue
             try:
                 graph = json.loads(row.get("content_with_weight") or "{}")
@@ -2012,14 +2055,39 @@ async def get_dataset_structure(dataset_id: str, tenant_id: str, kind: str, keyw
     return True, {"kind": kind, "templates": templates_out}
 
 
-async def get_wiki_alteration(dataset_id: str, tenant_id: str):
-    """Return doc-level drift between current dataset docs and compiled wiki provenance."""
-    if not KnowledgebaseService.accessible(dataset_id, tenant_id):
-        return False, "no authorization"
-    ok, kb = KnowledgebaseService.get_by_id(dataset_id)
-    if not ok:
-        return False, "Invalid Dataset ID"
+# ── artifacts/alteration: per-``kind`` provenance & eligibility mapping ──
+#
+# Alteration is a pure set operation shared by every kind:
+#   removed        = involved (product provenance) − current (dataset docs)
+#   newly_uploaded = eligible (should be compiled) − involved
+# Only two inputs vary by kind: how ``involved`` provenance is gathered, and
+# which template kinds make a doc ``eligible``.
 
+# Non-folded template kinds that make a doc eligible for each API kind.
+_ALTERATION_ELIGIBLE_TEMPLATE_KINDS = {
+    "wiki": {"artifacts"},
+    "graph": {"knowledge_graph"},
+    "mindmap": {"mind_map"},
+    "timeline": {"timeline"},
+    "tree": {"tree", "page_index"},
+}
+
+# Dataset-merged structure kinds carry provenance in ``source_doc_ids`` on
+# ``scope_kwd="dataset"`` rows, discovered by their stamped top-level kind.
+_ALTERATION_KIND_TO_MERGED_ROW_KIND = {
+    "graph": "knowledge_graph",
+    "mindmap": "mind_map",
+    "timeline": "timeline",
+}
+
+# Per-document structure kinds keep one product row per document; provenance is
+# the row's own ``doc_id`` (no dataset merge, no ``source_doc_ids``). ``tree`` and
+# ``page_index`` write distinct ``compile_kwd`` values.
+_ALTERATION_TREE_COMPILE_KWDS = ["tree", "page_index"]
+
+
+async def _current_dataset_docs(dataset_id: str):
+    """Return ``(docs, current_doc_id_set)`` for the dataset."""
     docs, _ = await thread_pool_exec(
         DocumentService.get_by_kb_id,
         kb_id=dataset_id,
@@ -2032,81 +2100,147 @@ async def get_wiki_alteration(dataset_id: str, tenant_id: str):
         types=[],
         suffix=[],
     )
-    current_doc_ids = {str(doc.get("id")) for doc in docs or [] if doc.get("id")}
+    docs = docs or []
+    current_doc_ids = {str(doc.get("id")) for doc in docs if doc.get("id")}
+    return docs, current_doc_ids
 
-    wiki_involved_doc_ids: set[str] = set()
-    pack = _wiki_index_or_none(kb.tenant_id, dataset_id)
-    if pack is not None:
-        index_nm, _ = pack
-        from common.doc_store.doc_store_base import OrderByExpr
 
-        select_fields = ["id", "source_doc_ids"]
-        offset = 0
-        page_size = 1000
-        while True:
-            try:
-                res = await thread_pool_exec(
-                    settings.docStoreConn.search,
-                    select_fields=select_fields,
-                    highlight_fields=[],
-                    condition={"compile_kwd": [WIKI_PAGE_COMPILE_KWD]},
-                    match_expressions=[],
-                    order_by=OrderByExpr(),
-                    offset=offset,
-                    limit=page_size,
-                    index_names=index_nm,
-                    knowledgebase_ids=[dataset_id],
-                )
-                rows = settings.docStoreConn.get_fields(res, select_fields) or {}
-            except Exception:
-                logging.exception("get_wiki_alteration: docStore search failed for kb=%s", dataset_id)
-                rows = {}
+def _alteration_result(current_doc_ids: set, involved_doc_ids: set, eligible_doc_ids: set) -> dict:
+    """Build the drift response dict shared by every ``kind``."""
+    removed_doc_ids = sorted(involved_doc_ids - current_doc_ids)
+    newly_uploaded_doc_ids = sorted(eligible_doc_ids - involved_doc_ids)
+    return {
+        "removed": len(removed_doc_ids),
+        "newly_uploaded": len(newly_uploaded_doc_ids),
+        "removed_doc_ids": removed_doc_ids,
+        "newly_uploaded_doc_ids": newly_uploaded_doc_ids,
+        "involved_doc_ids": sorted(involved_doc_ids),
+        "eligible_doc_ids": sorted(eligible_doc_ids),
+    }
 
-            if not rows:
-                break
-            for row in rows.values():
-                source_doc_ids = row.get("source_doc_ids")
-                if isinstance(source_doc_ids, str):
-                    source_doc_ids = [source_doc_ids]
-                if not isinstance(source_doc_ids, list):
-                    continue
-                wiki_involved_doc_ids.update(str(doc_id) for doc_id in source_doc_ids if doc_id)
-
-            offset += page_size
-            total = settings.docStoreConn.get_total(res)
-            if not total or offset >= int(total):
-                break
 
+def _eligible_doc_ids_for_kind(docs, tenant_id: str, kind: str) -> set:
+    """Doc ids whose parser_config or pipeline carries a template of ``kind``."""
+    accepted = _ALTERATION_ELIGIBLE_TEMPLATE_KINDS.get(kind) or set()
     template_cache: dict[str, bool] = {}
     group_cache: dict[str, bool] = {}
     pipeline_cache: dict[str, bool] = {}
-    eligible_wiki_doc_ids: set[str] = set()
+    eligible: set[str] = set()
     for doc in docs or []:
         doc_id = str(doc.get("id") or "")
         if not doc_id:
             continue
         parser_config = doc.get("parser_config") or {}
-        if _parser_config_has_wiki_template(parser_config, kb.tenant_id, template_cache):
-            eligible_wiki_doc_ids.add(doc_id)
+        if _parser_config_has_template_of_kind(parser_config, tenant_id, accepted, template_cache):
+            eligible.add(doc_id)
             continue
-        if _pipeline_has_wiki_compiler(
-            doc.get("pipeline_id") or "",
-            kb.tenant_id,
-            pipeline_cache,
-            group_cache,
-        ):
-            eligible_wiki_doc_ids.add(doc_id)
+        if _pipeline_has_compiler_of_kind(doc.get("pipeline_id") or "", tenant_id, accepted, pipeline_cache, group_cache):
+            eligible.add(doc_id)
+    return eligible
 
-    removed_doc_ids = sorted(wiki_involved_doc_ids - current_doc_ids)
-    newly_uploaded_doc_ids = sorted(eligible_wiki_doc_ids - wiki_involved_doc_ids)
-    return True, {
-        "removed": len(removed_doc_ids),
-        "newly_uploaded": len(newly_uploaded_doc_ids),
-        "removed_doc_ids": removed_doc_ids,
-        "newly_uploaded_doc_ids": newly_uploaded_doc_ids,
-        "involved_doc_ids": sorted(wiki_involved_doc_ids),
-        "eligible_doc_ids": sorted(eligible_wiki_doc_ids),
-    }
+
+async def _involved_doc_ids_paged(index_nm, dataset_id: str, condition: dict, field: str, from_list: bool) -> set:
+    """Page a docStore search, folding each row's ``field`` into a doc-id set.
+
+    ``from_list`` reads ``field`` as ``source_doc_ids`` (a list of provenance doc
+    ids); otherwise ``field`` is the row's own ``doc_id`` scalar.
+    """
+    from common.doc_store.doc_store_base import OrderByExpr
+
+    involved: set[str] = set()
+    select_fields = ["id", field]
+    offset = 0
+    page_size = 1000
+    while True:
+        try:
+            res = await thread_pool_exec(
+                settings.docStoreConn.search,
+                select_fields=select_fields,
+                highlight_fields=[],
+                condition=condition,
+                match_expressions=[],
+                order_by=OrderByExpr(),
+                offset=offset,
+                limit=page_size,
+                index_names=index_nm,
+                knowledgebase_ids=[dataset_id],
+            )
+            rows = settings.docStoreConn.get_fields(res, select_fields) or {}
+        except Exception:
+            logging.exception("alteration: docStore search failed for kb=%s cond=%s", dataset_id, condition)
+            rows = {}
+
+        if not rows:
+            break
+        for row in rows.values():
+            value = row.get(field)
+            if from_list:
+                if isinstance(value, str):
+                    value = [value]
+                if isinstance(value, list):
+                    involved.update(str(d) for d in value if d)
+            elif value:
+                involved.add(str(value))
+
+        offset += page_size
+        total = settings.docStoreConn.get_total(res)
+        if not total or offset >= int(total):
+            break
+    return involved
+
+
+async def _involved_doc_ids_for_kind(index_nm, dataset_id: str, kind: str) -> set:
+    """Gather the doc ids baked into the compiled product for ``kind``."""
+    if kind == "wiki":
+        return await _involved_doc_ids_paged(index_nm, dataset_id, {"compile_kwd": [WIKI_PAGE_COMPILE_KWD]}, "source_doc_ids", from_list=True)
+    if kind in _ALTERATION_KIND_TO_MERGED_ROW_KIND:
+        condition = {
+            "knowledge_graph_kwd": ["entity", "relation"],
+            "scope_kwd": ["dataset"],
+            "compilation_template_kind_kwd": [_ALTERATION_KIND_TO_MERGED_ROW_KIND[kind]],
+        }
+        return await _involved_doc_ids_paged(index_nm, dataset_id, condition, "source_doc_ids", from_list=True)
+    if kind == "tree":
+        return await _involved_doc_ids_paged(index_nm, dataset_id, {"compile_kwd": list(_ALTERATION_TREE_COMPILE_KWDS)}, "doc_id", from_list=False)
+    return set()
+
+
+async def _get_alteration(dataset_id: str, tenant_id: str, kind: str):
+    """Shared driver: doc-level drift between the ``kind`` product and the dataset."""
+    if not KnowledgebaseService.accessible(dataset_id, tenant_id):
+        return False, "no authorization"
+    ok, kb = KnowledgebaseService.get_by_id(dataset_id)
+    if not ok:
+        return False, "Invalid Dataset ID"
+
+    docs, current_doc_ids = await _current_dataset_docs(dataset_id)
+
+    involved_doc_ids: set[str] = set()
+    pack = _compiled_index_or_none(kb.tenant_id, dataset_id)
+    if pack is not None:
+        index_nm, _ = pack
+        involved_doc_ids = await _involved_doc_ids_for_kind(index_nm, dataset_id, kind)
+
+    eligible_doc_ids = _eligible_doc_ids_for_kind(docs, kb.tenant_id, kind)
+    return True, _alteration_result(current_doc_ids, involved_doc_ids, eligible_doc_ids)
+
+
+async def get_wiki_alteration(dataset_id: str, tenant_id: str):
+    """Return doc-level drift between current dataset docs and compiled wiki provenance."""
+    return await _get_alteration(dataset_id, tenant_id, "wiki")
+
+
+async def get_structure_alteration(dataset_id: str, tenant_id: str, kind: str):
+    """Return doc-level drift for a non-wiki structure ``kind``.
+
+    ``kind`` is one of ``graph`` / ``mindmap`` / ``timeline`` (dataset-merged
+    products, provenance from ``source_doc_ids``) or ``tree`` (per-document
+    products covering ``tree`` + ``page_index``, provenance from ``doc_id``).
+    """
+    kind = (kind or "").strip().lower()
+    if kind not in _ALTERATION_KIND_TO_MERGED_ROW_KIND and kind != "tree":
+        return False, f"Unsupported structure kind: {kind!r}. Expected one of: graph, mindmap, timeline, tree."
+    return await _get_alteration(dataset_id, tenant_id, kind)
 
 
 async def list_wiki_pages(
diff --git a/api/db/init_data/compilation_templates/wiki.yaml b/api/db/init_data/compilation_templates/wiki.yaml
index b0484c4b1c..dc2949d060 100644
--- a/api/db/init_data/compilation_templates/wiki.yaml
+++ b/api/db/init_data/compilation_templates/wiki.yaml
@@ -1,7 +1,7 @@
-kind: artifacts
-display_name: Artifacts — Graph-based wiki
+kind: wiki
+display_name: Wiki — Graph-based wiki
 config:
-  kind: artifacts
+  kind: wiki
   example: |
       - Each page must be a proper encyclopedic article, NOT a flat bullet list:
       - 1. Opening paragraph (2-4 sentences defining what this is). No heading.
diff --git a/api/db/services/compilation_template_group_service.py b/api/db/services/compilation_template_group_service.py
index 00bf0e09dc..7ea104c79a 100644
--- a/api/db/services/compilation_template_group_service.py
+++ b/api/db/services/compilation_template_group_service.py
@@ -39,10 +39,10 @@ def _derive_scope(templates: list[dict]) -> str:
     if not templates:
         raise GroupValidationError("A template group must contain at least one template.")
     kinds = [str((t or {}).get("kind") or "").strip() for t in templates]
-    artifact_count = sum(1 for k in kinds if k == "artifacts")
+    artifact_count = sum(1 for k in kinds if k == "wiki")
     if artifact_count > 0:
         if artifact_count != 1 or len(templates) != 1:
-            raise GroupValidationError("An artifacts template cannot be combined with other templates in the same group.")
+            raise GroupValidationError("A wiki template cannot be combined with other templates in the same group.")
         return SCOPE_DATASET
 
     _enforce_single_rechunk_tree(templates)
diff --git a/api/db/services/document_service.py b/api/db/services/document_service.py
index 48167aff6c..b639bb291c 100644
--- a/api/db/services/document_service.py
+++ b/api/db/services/document_service.py
@@ -1225,7 +1225,7 @@ def queue_raptor_o_graphrag_tasks(sample_doc, ty, priority, fake_doc_id="", doc_
         "graphrag",
         "raptor",
         "mindmap",
-        "artifact",
+        "wiki",
         "skill",
         # KB-wide structure-graph merge task types (rebuild dataset_graph rows).
         "structure_graph",
diff --git a/common/constants.py b/common/constants.py
index 139612e42d..51fb3c7e17 100644
--- a/common/constants.py
+++ b/common/constants.py
@@ -185,7 +185,9 @@ class PipelineTaskType(StrEnum):
     GRAPH_RAG = "GraphRAG"
     MINDMAP = "Mindmap"
     MEMORY = "Memory"
-    ARTIFACT = "Artifact"
+    # Member name kept as ARTIFACT for back-compat; value is "Wiki" so the
+    # runtime task_type (``.lower()`` == "wiki") matches the wiki index/task type.
+    ARTIFACT = "Wiki"
     SKILL = "Skill"
     # KB-wide structure-graph merge tasks (rebuild_dataset_structure_graph_json).
     STRUCTURE_GRAPH = "StructureGraph"
diff --git a/rag/advanced_rag/harness/orchestrator/agentic.py b/rag/advanced_rag/harness/orchestrator/agentic.py
index 19bb9424f0..212cbc110c 100644
--- a/rag/advanced_rag/harness/orchestrator/agentic.py
+++ b/rag/advanced_rag/harness/orchestrator/agentic.py
@@ -310,7 +310,7 @@ async def _add_template_group_compilations(comps: set[str], parser_config: dict,
                 comps.add("page_index")
             elif kind in {"mindmap", "mind_map"}:
                 comps.add("mindmap")
-            elif kind == "artifacts":
+            elif kind == "wiki":
                 comps.add("wiki")
 
 
diff --git a/rag/advanced_rag/harness/orchestrator/direct.py b/rag/advanced_rag/harness/orchestrator/direct.py
index e7cdc32704..d7c1fc43b3 100644
--- a/rag/advanced_rag/harness/orchestrator/direct.py
+++ b/rag/advanced_rag/harness/orchestrator/direct.py
@@ -11,9 +11,9 @@ async def direct_search(state: dict, tools) -> dict:
     """Single hybrid search → merge into kbinfos."""
     question = state.get("question", "")
     keywords = state.get("keywords", "")
-    _LOG.info("[Direct search] Looking up the knowledge base for: \"%s\" (keywords: %s)", question, keywords)
+    _LOG.info('[Direct search] Looking up the knowledge base for: "%s" (keywords: %s)', question, keywords)
 
-    result = await hybrid_search(tools, query=question, keywords=keywords)
+    result = await hybrid_search(tools, query=question, keywords=keywords, use_compiled=True)
     _merge_kbinfos(tools, result)
 
     if not _has_chunks(tools):
diff --git a/rag/advanced_rag/harness/tools/navigation.py b/rag/advanced_rag/harness/tools/navigation.py
index 4d52cc9e8d..f3127452f5 100644
--- a/rag/advanced_rag/harness/tools/navigation.py
+++ b/rag/advanced_rag/harness/tools/navigation.py
@@ -543,13 +543,22 @@ async def dataset_navigation_by_tree(tools, topic: str, keywords: str = "", doc_
 # its way to a small subgraph: seed entities by the question, hop out over
 # relations to the 2nd-degree neighbours, then answer from that subgraph.
 
-_KG_SEEDS = 3  # entities matched directly to the question
-_KG_HOPS = 1  # relation hops out from the seeds (1 => "2nd degree")
-_KG_NEIGHBORS = 10  # neighbour entity rows resolved per hop
+# Scope of the compiled KG rows we search. Without a doc_scope we read the
+# dataset-merged rows (one graph per dataset); with a doc_scope we read the
+# per-document rows and filter by doc_id. Kept local rather than imported from
+# the task-executor layer so the harness has no dependency on it.
+_SCOPE_KWD_DATASET = "dataset"
+_SCOPE_KWD_DOC = "doc"
+
+_KG_SEEDS = 2  # top-N entities matched directly to the question
+_KG_SEED_POOL = 64  # KNN candidate pool before the mention_count_int re-sort
+_KG_SEED_SIM = 0.8  # dense-similarity floor for seed entities
+_KG_HOPS = 2  # relation hops out from the seeds (1 => "2nd degree")
+_KG_NEIGHBORS = 128  # cap on neighbour entity rows resolved per hop
 _KG_REL_LIMIT = 32  # relations fetched per endpoint filter
 
 
-async def _kg_scopes(tools, doc_scope: list[str] | None):
+async def _kg_scopes(tools, doc_scope: list[str] | None = None):
     """Resolve the (kb_id, tenant_id, doc_ids|None) groups to search.
 
     With a ``doc_scope`` the graph is limited to those docs (grouped by their
@@ -567,29 +576,59 @@ async def _kg_scopes(tools, doc_scope: list[str] | None):
     return [(kb.id, kb.tenant_id, None) for kb in getattr(tools, "kbs", []) or []]
 
 
-async def _kg_search(tools, kb_id: str, tenant_id: str, doc_ids, kind: str, text: str = "", top_n: int = 8, extra: dict | None = None) -> list[dict]:
-    """Search the compiled KG rows of one KB and return the raw field maps."""
+async def _kg_search(
+    tools,
+    kb_id: str,
+    tenant_id: str,
+    doc_ids,
+    kind: str,
+    text: str = "",
+    top_n: int = 8,
+    extra: dict | None = None,
+    scope_kwd: str | None = None,
+    order_desc: str | None = None,
+    pool: int | None = None,
+    similarity: float = 0.6,
+) -> list[dict]:
+    """Search the compiled KG rows of one KB and return the raw field maps.
+
+    ``scope_kwd`` narrows to dataset-merged (``"dataset"``) or per-doc (``"doc"``)
+    rows. ``order_desc`` sorts the hits by that field descending; when combined
+    with a dense ``text`` match, ``pool`` sets the KNN candidate count so the
+    re-sort ranks a wider pool than the ``top_n`` finally kept. ``similarity`` is
+    the dense-match floor.
+    """
     from common import settings
     from common.doc_store.doc_store_base import MatchTextExpr, OrderByExpr
     from common.misc_utils import thread_pool_exec
     from rag.nlp import search
 
     condition: dict = {"knowledge_graph_kwd": [kind]}
+    if scope_kwd:
+        condition["scope_kwd"] = [scope_kwd]
     if doc_ids:
         condition["doc_id"] = list(doc_ids)
     if extra:
         condition.update(extra)
 
-    fields = ["content_with_weight", "source_chunk_ids", "doc_id", "docnm_kwd", "from_entity_kwd", "to_entity_kwd"]
+    fields = ["content_with_weight", "source_chunk_ids", "doc_id", "docnm_kwd", "name_kwd", "mention_count_int", "from_entity_kwd", "to_entity_kwd"]
     exprs = []
     if text:
+        knn_topn = pool or top_n
         if getattr(tools, "embed_mdl", None):
             try:
-                exprs.append(await settings.retriever.get_vector(text, tools.embed_mdl, top_n, 0.1))
+                exprs.append(await settings.retriever.get_vector(text, tools.embed_mdl, knn_topn, similarity))
             except Exception:
                 _LOG.exception("[Graph exploration] vector build failed; using keyword match")
         if not exprs:
-            exprs.append(MatchTextExpr(["content_ltks", "content_sm_ltks"], text, top_n))
+            exprs.append(MatchTextExpr(["content_ltks", "content_sm_ltks"], text, knn_topn))
+
+    order_by = OrderByExpr()
+    if order_desc:
+        try:
+            order_by.desc(order_desc)
+        except Exception:
+            order_by = OrderByExpr()
 
     try:
         res = await thread_pool_exec(
@@ -598,7 +637,7 @@ async def _kg_search(tools, kb_id: str, tenant_id: str, doc_ids, kind: str, text
             [],
             condition,
             exprs,
-            OrderByExpr(),
+            order_by,
             0,
             top_n,
             search.index_name(tenant_id),
@@ -651,6 +690,25 @@ def _kg_parse_relation(row: dict) -> dict | None:
     }
 
 
+def _endpoint_terms(names) -> list[str]:
+    """Case variants for matching relation endpoints.
+
+    ``dataset_structure_merger`` lowercases ``from_entity_kwd``/``to_entity_kwd``
+    on merged rows while entity names keep their original case, so hop queries
+    must try both forms. Accepts a single name or an iterable and returns the
+    sorted union of each name's original and lowercased form.
+    """
+    if isinstance(names, str):
+        names = [names]
+    terms: set[str] = set()
+    for n in names or []:
+        n = (n or "").strip()
+        if n:
+            terms.add(n)
+            terms.add(n.lower())
+    return sorted(terms)
+
+
 def _collect_evidence_ids(entities: list[dict], relations: list[dict], relevant_names: list[str]) -> dict:
     """Group the source_chunk_ids of the relevant entities AND relations by doc.
 
@@ -684,23 +742,31 @@ def _collect_evidence_ids(entities: list[dict], relations: list[dict], relevant_
 async def graph_explore(tools, query: str, keywords: str = "", doc_scope: list[str] | None = None) -> dict:
     """Explore the compiled knowledge graph to answer ``query``.
 
-    Searches seed entities for the question, hops out over their relations to the
-    2nd-degree neighbours, asks the chat model to answer from that subgraph, then
-    pulls the source passages behind the entities/relations it found relevant and
-    narrows them with ``keywords``. Falls back to a plain hybrid search when the
-    KB has no compiled graph or no source text is behind the relevant nodes.
+    Seeds the top-``_KG_SEEDS`` entities for the question (dense match above
+    ``_KG_SEED_SIM``, ranked by ``mention_count_int``), hops ``_KG_HOPS`` out over
+    their relations, then asks the chat model whether that subgraph answers the
+    question. When it does, the answer is returned directly; when it doesn't, the
+    source passages behind the entities/relations the model found relevant are
+    returned as evidence (narrowed by ``keywords``) so the caller can continue.
 
-    :returns: ``{"answer": str, "chunks": [...], "doc_aggs": [...]}``
+    Without ``doc_scope`` the dataset-merged graph is searched
+    (``scope_kwd="dataset"``); with it, the per-document rows of those docs
+    (``scope_kwd="doc"``, filtered by ``doc_id``).
+
+    :returns: ``{"answer": str, "chunks": [...], "doc_aggs": [...]}`` — exactly one
+        of ``answer`` / ``chunks`` is populated.
     """
-    from rag.advanced_rag.harness.tools.search import _narrow_by_keywords, hybrid_search
+    from rag.advanced_rag.harness.tools.search import _narrow_by_keywords
 
+    _empty = {"answer": "", "chunks": [], "doc_aggs": []}
     _LOG.info(f'[Graph exploration] Exploring the knowledge graph for "{query}" (keywords: {keywords})')
 
     scopes = await _kg_scopes(tools, doc_scope)
     if not scopes:
-        _LOG.info("[Graph exploration] No knowledge base in scope — falling back to a normal search.")
-        return await hybrid_search(tools, query=query, keywords=keywords, doc_scope=doc_scope)
+        _LOG.info("[Graph exploration] No knowledge base in scope.")
+        return _empty
 
+    scope_kwd = _SCOPE_KWD_DOC if doc_scope else _SCOPE_KWD_DATASET
     text = f"{query} {keywords}".strip()
     entities: list[dict] = []
     relations: list[dict] = []
@@ -718,52 +784,79 @@ async def graph_explore(tools, query: str, keywords: str = "", doc_scope: list[s
         return added
 
     for kb_id, tenant_id, doc_ids in scopes:
-        seed_rows = await _kg_search(tools, kb_id, tenant_id, doc_ids, "entity", text=text, top_n=_KG_SEEDS)
+        # (1) Seeds: condition C — dense match (>= _KG_SEED_SIM) over the scoped
+        # entity rows, ranked by mention_count_int desc, top _KG_SEEDS.
+        seed_rows = await _kg_search(
+            tools,
+            kb_id,
+            tenant_id,
+            doc_ids,
+            "entity",
+            text=text,
+            top_n=_KG_SEEDS,
+            scope_kwd=scope_kwd,
+            order_desc="mention_count_int",
+            pool=_KG_SEED_POOL,
+            similarity=_KG_SEED_SIM,
+        )
         seeds = [e for e in (_kg_parse_entity(r) for r in seed_rows.values()) if e]
         frontier = _add_entities(seeds, kb_id)
         _LOG.info("[Graph exploration] Seeded %d entity(ies): %s", len(frontier), ", ".join(frontier) or "none")
 
+        # (2) Expand _KG_HOPS out, collecting relations and neighbour entities.
         for _hop in range(_KG_HOPS):
             if not frontier:
                 break
-            # Relations touching the current frontier (as source OR target).
+            terms = _endpoint_terms(frontier)  # case variants — merged rows lowercase endpoints
             rel_rows: dict = {}
-            rel_rows.update(await _kg_search(tools, kb_id, tenant_id, doc_ids, "relation", top_n=_KG_REL_LIMIT, extra={"from_entity_kwd": frontier}))
-            rel_rows.update(await _kg_search(tools, kb_id, tenant_id, doc_ids, "relation", top_n=_KG_REL_LIMIT, extra={"to_entity_kwd": frontier}))
+            rel_rows.update(await _kg_search(tools, kb_id, tenant_id, doc_ids, "relation", top_n=_KG_REL_LIMIT, scope_kwd=scope_kwd, extra={"from_entity_kwd": terms}))
+            rel_rows.update(await _kg_search(tools, kb_id, tenant_id, doc_ids, "relation", top_n=_KG_REL_LIMIT, scope_kwd=scope_kwd, extra={"to_entity_kwd": terms}))
             hop_relations = [r for r in (_kg_parse_relation(x) for x in rel_rows.values()) if r]
             relations.extend(hop_relations)
 
-            neighbour_names = {n for r in hop_relations for n in (r["from"], r["to"]) if n.lower() not in ent_names}
-            if not neighbour_names:
+            seen_lower = {k.split(":", 1)[1] for k in ent_names if k.startswith(f"{kb_id}:")}
+            neigh_names = {n.strip() for r in hop_relations for n in (r["from"], r["to"]) if n and n.strip()}
+            neigh_lower_set = {n.lower() for n in neigh_names} - seen_lower
+            if not neigh_lower_set:
                 break
-            neigh_rows = await _kg_search(tools, kb_id, tenant_id, doc_ids, "entity", text=" ".join(neighbour_names), top_n=_KG_NEIGHBORS)
-            wanted = {n.lower() for n in neighbour_names}
-            neighbours = [e for e in (_kg_parse_entity(r) for r in neigh_rows.values()) if e and e["name"].lower() in wanted]
+            neigh_filtered = {n for n in neigh_names if n.lower() in neigh_lower_set}
+            neigh_rows = await _kg_search(
+                tools,
+                kb_id,
+                tenant_id,
+                doc_ids,
+                "entity",
+                top_n=min(max(len(neigh_filtered), 1), _KG_NEIGHBORS),
+                scope_kwd=scope_kwd,
+                extra={"name_kwd": _endpoint_terms(neigh_filtered)},
+            )
+            neighbours = [e for e in (_kg_parse_entity(r) for r in neigh_rows.values()) if e]
             frontier = _add_entities(neighbours, kb_id)
             _LOG.info("[Graph exploration] Hop %d reached %d neighbour entity(ies).", _hop + 1, len(frontier))
 
     if not entities and not relations:
-        _LOG.info("[Graph exploration] No compiled knowledge graph here — falling back to a normal search.")
-        return await hybrid_search(tools, query=query, keywords=keywords, doc_scope=doc_scope)
+        _LOG.info("[Graph exploration] No compiled knowledge graph in scope.")
+        return _empty
 
     _LOG.info("[Graph exploration] Built a subgraph of %d entity(ies) and %d relation(s).", len(entities), len(relations))
 
+    # (3) Does the subgraph answer the question?
     answer, relevant = await _ask_structure(tools, query, entities, relations, "knowledge graph", "Graph exploration")
 
+    # (4a) Sufficient — return the answer, no chunks.
+    if answer:
+        _LOG.info("[Graph exploration] The subgraph answered the question directly.")
+        return {"answer": answer, "chunks": [], "doc_aggs": []}
+
+    # (4b) Insufficient — return the source passages behind the relevant nodes.
     evidence = _collect_evidence_ids(entities, relations, relevant)
     chunks: list[dict] = []
     for doc_id, ids in evidence.items():
         if doc_id and ids:
             chunks.extend(await _load_chunks_by_ids(tools, doc_id, ids))
 
-    if not chunks:
-        _LOG.info("[Graph exploration] No source text behind those nodes — falling back to a normal search.")
-        fallback = await hybrid_search(tools, query=query, keywords=keywords, doc_scope=doc_scope)
-        fallback["answer"] = answer
-        return fallback
-
     before = len(chunks)
     chunks = _narrow_by_keywords(chunks, keywords)
-    _LOG.info("[Graph exploration] Pulled %d source passage(s) behind those nodes, kept %d after keyword filtering.", before, len(chunks))
+    _LOG.info("[Graph exploration] Insufficient; returning %d evidence passage(s) (%d before keyword filtering).", len(chunks), before)
 
-    return {"answer": answer, "chunks": chunks, "doc_aggs": _doc_aggs(chunks)}
+    return {"answer": "", "chunks": chunks, "doc_aggs": _doc_aggs(chunks)}
diff --git a/rag/advanced_rag/harness/tools/search.py b/rag/advanced_rag/harness/tools/search.py
index 351cbba44c..822c7812b1 100644
--- a/rag/advanced_rag/harness/tools/search.py
+++ b/rag/advanced_rag/harness/tools/search.py
@@ -4,6 +4,7 @@ import logging
 import re
 import hashlib
 from common import settings
+from .navigation import _kg_scopes
 
 _LOG = logging.getLogger(__name__)
 
@@ -179,7 +180,7 @@ def _normalize(kbinfos: dict, tenant_ids: list[str] | str | None) -> dict:
     return kbinfos
 
 
-async def hybrid_search(tools, query: str, kb_ids: list[str] | None = None, top_n: int = 12, doc_scope: list[str] | None = None, keywords: str = "") -> dict:
+async def hybrid_search(tools, query: str, kb_ids: list[str] | None = None, top_n: int = 12, doc_scope: list[str] | None = None, keywords: str = "", use_compiled: bool = False) -> dict:
     if not tools.kb_ids and not kb_ids:
         return {"chunks": [], "doc_aggs": []}
     target_ids = kb_ids or tools.kb_ids
@@ -220,6 +221,9 @@ async def hybrid_search(tools, query: str, kb_ids: list[str] | None = None, top_
         length = len(kbinfos["chunks"])
         kbinfos["chunks"] = _narrow_by_keywords(kbinfos.get("chunks", []), keywords)
         _LOG.info(f"[Hybrid search] Kept {len(kbinfos['chunks'])} of {length} passage(s) that actually mention the keywords.")
+    if use_compiled and kbinfos.get("chunks"):
+        _LOG.info("[Hybrid search] Compiled expansion enabled — enriching with page_index/tree/KG navigation.")
+        await _expand_with_compiled(tools, query, keywords, kbinfos)
     if cache is not None:
         cache[cache_key] = kbinfos
     return kbinfos
@@ -277,6 +281,473 @@ async def bm25_search(tools, query: str, kb_ids: list[str] | None = None, top_n:
     return kbinfos
 
 
+# ─── Compiled product expansion (zero-LLM, used by hybrid_search with use_compiled=True) ───
+
+
+async def _expand_with_compiled(tools, query: str, keywords: str, kbinfos: dict) -> None:
+    """Zero-LLM compiled-product expansion: page_index → tree → KG.
+
+    For each bound KB, searches compiled entity rows matching the query,
+    hops 1-hop via relations to find neighbour entities, then appends
+    their source passages to ``kbinfos["chunks"]``.
+    """
+    before = len(kbinfos.get("chunks", []))
+    seen_ids = {c.get("chunk_id") or c.get("id") for c in kbinfos.get("chunks", [])}
+
+    scopes = await _kg_scopes(tools)
+    if not scopes:
+        return
+
+    for kb_id, tenant_id, doc_ids in scopes:
+        # 1-hop entity-graph expansion per template kind.
+        # Each template writes entity/relation rows tagged with
+        # ``compilation_template_kind_kwd`` — search them independently.
+        for label, template_kind in (
+            ("knowledge_graph", "knowledge_graph"),
+            ("mind_map", "mind_map"),
+            ("timeline", "timeline"),
+            ("page_index", "page_index"),
+        ):
+            chunks = await _expand_compiled_strategy(
+                tools,
+                kb_id,
+                tenant_id,
+                doc_ids,
+                query,
+                seen_ids,
+                template_kind=template_kind,
+                max_chunks=5,
+            )
+            if chunks:
+                kbinfos.setdefault("chunks", []).extend(chunks)
+                _LOG.debug("[Compiled expand] %s: +%d chunks", label, len(chunks))
+
+        # Tree structure graph (uses ``compile_kwd``, not template kind).
+        chunks = await _expand_compiled_strategy(
+            tools,
+            kb_id,
+            tenant_id,
+            doc_ids,
+            query,
+            seen_ids,
+            compile_kwd="tree",
+            max_chunks=5,
+        )
+        if chunks:
+            kbinfos.setdefault("chunks", []).extend(chunks)
+            _LOG.debug("[Compiled expand] tree: +%d chunks", len(chunks))
+
+        # Synthesis pages — standalone rendered articles from wiki / session
+        # graph / session essence templates.  Searched directly (no entity-graph nav).
+        for label, ckwd in (
+            ("wiki_page", "wiki_page"),
+            ("artifact_page", "artifact_page"),
+            ("essence", "essence"),
+        ):
+            chunks = await _expand_wiki_page_strategy(
+                tools,
+                kb_id,
+                tenant_id,
+                doc_ids,
+                query,
+                seen_ids,
+                compile_kwd=ckwd,
+                max_chunks=5,
+            )
+            if chunks:
+                kbinfos.setdefault("chunks", []).extend(chunks)
+                _LOG.debug("[Compiled expand] %s: +%d chunks", label, len(chunks))
+
+    # Re-sort so compiled-expansion chunks blend by similarity with regular ones.
+    chunks = kbinfos.get("chunks", [])
+    if chunks:
+        chunks.sort(key=lambda c: c.get("similarity", 0.0), reverse=True)
+
+    after = len(chunks)
+    _LOG.info("[Hybrid search] Compiled expansion added %d chunks.", after - before)
+
+
+async def _search_compiled_rows(
+    tools,
+    kb_id: str,
+    tenant_id: str,
+    doc_ids: list[str] | None,
+    kind: str,
+    *,
+    text: str = "",
+    top_n: int = 8,
+    extra: dict | None = None,
+    compile_kwd: str | None = None,
+    template_kind: str | None = None,
+) -> dict:
+    """Search compiled KG rows in one KB, returning raw field maps.
+
+    *compile_kwd* filters by the ``compile_kwd`` field (e.g. "tree" for tree
+    structure nodes).  *template_kind* filters by ``compilation_template_kind_kwd``
+    (e.g. "knowledge_graph", "mind_map").  Leave both ``None`` to scan all rows.
+    """
+    from common import settings
+    from common.doc_store.doc_store_base import MatchTextExpr, OrderByExpr
+    from common.misc_utils import thread_pool_exec
+    from rag.nlp import search
+
+    condition: dict = {"knowledge_graph_kwd": [kind]}
+    if compile_kwd:
+        condition["compile_kwd"] = compile_kwd
+    if template_kind:
+        condition["compilation_template_kind_kwd"] = template_kind
+    if doc_ids:
+        condition["doc_id"] = list(doc_ids)
+    if extra:
+        condition.update(extra)
+
+    fields = [
+        "content_with_weight",
+        "source_chunk_ids",
+        "doc_id",
+        "docnm_kwd",
+        "from_entity_kwd",
+        "to_entity_kwd",
+        "name_kwd",
+    ]
+    exprs = []
+    if text:
+        embd_mdl = getattr(tools, "embed_mdl", None)
+        if embd_mdl:
+            try:
+                exprs.append(await settings.retriever.get_vector(text, embd_mdl, top_n, 0.1))
+            except Exception:
+                _LOG.exception("[Compiled expand] vector build failed; using keyword match")
+        if not exprs:
+            exprs.append(
+                MatchTextExpr(
+                    ["content_ltks", "content_sm_ltks"],
+                    text,
+                    top_n,
+                )
+            )
+
+    try:
+        res = await thread_pool_exec(
+            settings.docStoreConn.search,
+            fields,
+            [],
+            condition,
+            exprs,
+            OrderByExpr(),
+            0,
+            top_n,
+            search.index_name(tenant_id),
+            [kb_id],
+        )
+        return settings.docStoreConn.get_fields(res, fields) or {}
+    except Exception:
+        _LOG.exception("[Compiled expand] search failed (kind=%s compile_kwd=%s)", kind, compile_kwd)
+        return {}
+
+
+async def _load_chunks_for_doc(tools, doc_id: str, chunk_ids: list[str]) -> list[dict]:
+    """Load chunks by their IDs from the doc store."""
+    if not chunk_ids:
+        return []
+    from common import settings
+    from common.doc_store.doc_store_base import OrderByExpr
+    from common.misc_utils import thread_pool_exec
+    from rag.nlp import search
+
+    resolved = await thread_pool_exec(tools._resolve_doc_tenant, doc_id)
+    if not resolved:
+        return []
+    kb_id, tenant_id = resolved
+
+    fields = ["content_with_weight", "docnm_kwd", "doc_id", "id"]
+    try:
+        res = await thread_pool_exec(
+            settings.docStoreConn.search,
+            fields,
+            [],
+            {"id": list(chunk_ids)},
+            [],
+            OrderByExpr(),
+            0,
+            len(chunk_ids),
+            search.index_name(tenant_id),
+            [kb_id],
+        )
+        rows = settings.docStoreConn.get_fields(res, fields)
+        if not rows:
+            return []
+        return [{**v, "chunk_id": k} for k, v in rows.items()]
+    except Exception:
+        _LOG.exception("[Compiled expand] failed to load chunks for doc_id=%s", doc_id)
+        return []
+
+
+async def _expand_compiled_strategy(
+    tools,
+    kb_id: str,
+    tenant_id: str,
+    doc_ids: list[str] | None,
+    query: str,
+    seen_ids: set[str],
+    *,
+    compile_kwd: str | None = None,
+    template_kind: str | None = None,
+    max_chunks: int = 5,
+) -> list[dict]:
+    """Generic 1-hop compiled expansion: entity search → relation nav → chunk load.
+
+    1. Embedding-match seed entities (filtered by *compile_kwd* or *template_kind*).
+    2. Fetch relations adjacent to seed entities (forward + backward).
+    3. Collect neighbour entity names (1-hop away).
+    4. Look up neighbour entities to get ``source_chunk_ids``.
+    5. Load actual chunks, deduplicate, respect *max_chunks*.
+
+    *compile_kwd* is used for structure graphs (e.g. "tree").
+    *template_kind* is used for entity extraction rows (e.g. "knowledge_graph").
+    """
+    import json
+
+    # -- 1. Seed entities --
+    seed_rows = await _search_compiled_rows(
+        tools,
+        kb_id,
+        tenant_id,
+        doc_ids,
+        "entity",
+        text=query,
+        top_n=5,
+        compile_kwd=compile_kwd,
+        template_kind=template_kind,
+    )
+    if not seed_rows:
+        return []
+
+    seed_names: set[str] = set()
+    for r in seed_rows.values():
+        try:
+            payload = json.loads(r.get("content_with_weight") or "{}")
+        except Exception:
+            continue
+        name = (payload.get("name") or payload.get("title") or "").strip()
+        if name:
+            seed_names.add(name)
+    if not seed_names:
+        return []
+
+    # -- 2. Adjacent relations (outgoing + incoming) --
+    # Provide both original and lowercased names — dataset_structure_merger
+    # lowercases merged-row endpoints while per-doc rows keep original case.
+    seed_list = sorted({n.lower() for n in seed_names} | seed_names)
+    fwd = await _search_compiled_rows(
+        tools,
+        kb_id,
+        tenant_id,
+        doc_ids,
+        "relation",
+        top_n=50,
+        compile_kwd=compile_kwd,
+        template_kind=template_kind,
+        extra={"from_entity_kwd": seed_list},
+    )
+    bwd = await _search_compiled_rows(
+        tools,
+        kb_id,
+        tenant_id,
+        doc_ids,
+        "relation",
+        top_n=50,
+        compile_kwd=compile_kwd,
+        template_kind=template_kind,
+        extra={"to_entity_kwd": seed_list},
+    )
+    all_rels = {**fwd, **bwd}
+
+    # -- 3. Neighbour names (1-hop, exclude seeds) --
+    seed_lower = {n.lower() for n in seed_names}
+    neighbour_names: set[str] = set()
+    for r in all_rels.values():
+        frm = (r.get("from_entity_kwd") or "").strip()
+        frm_lower = frm.lower()
+        to = (r.get("to_entity_kwd") or "").strip()
+        to_lower = to.lower()
+        if frm_lower in seed_lower and to and to_lower not in seed_lower:
+            neighbour_names.add(to)
+        if to_lower in seed_lower and frm and frm_lower not in seed_lower:
+            neighbour_names.add(frm)
+    if not neighbour_names:
+        return []
+
+    # -- 4. Neighbour entity source_chunk_ids --
+    # Provide both original and lowercased — same as seed_list above.
+    neigh_list = sorted({n.lower() for n in neighbour_names} | neighbour_names)
+    if len(neigh_list) > 100:
+        neigh_list = neigh_list[:100]  # reasonable cap for name_kwd search
+    neigh_rows = await _search_compiled_rows(
+        tools,
+        kb_id,
+        tenant_id,
+        doc_ids,
+        "entity",
+        top_n=len(neigh_list),
+        compile_kwd=compile_kwd,
+        template_kind=template_kind,
+        extra={"name_kwd": neigh_list},
+    )
+
+    # Group chunk IDs by doc
+    by_doc: dict[str, set[str]] = {}
+    for r in neigh_rows.values():
+        doc_id = r.get("doc_id") or ""
+        for cid in r.get("source_chunk_ids") or []:
+            if cid and cid not in seen_ids:
+                by_doc.setdefault(doc_id, set()).add(cid)
+
+    # -- 5. Load and return --
+    new_chunks: list[dict] = []
+    for doc_id, cids in by_doc.items():
+        if len(new_chunks) >= max_chunks:
+            break
+        limit = max_chunks - len(new_chunks)
+        chunks = await _load_chunks_for_doc(tools, doc_id, list(cids)[:limit])
+        for c in chunks:
+            cid = c.get("chunk_id") or c.get("id")
+            if cid and cid not in seen_ids:
+                seen_ids.add(cid)
+                new_chunks.append(c)
+
+    return new_chunks
+
+
+async def _search_synthesis_pages(
+    tools,
+    kb_id: str,
+    tenant_id: str,
+    doc_ids: list[str] | None,
+    text: str,
+    *,
+    compile_kwd: str = "wiki_page",
+    top_n: int = 8,
+) -> dict:
+    """Search synthesis-compiled page rows (no knowledge_graph_kwd filter).
+
+    Synthesis pages are standalone articles (wiki_page, artifact_page,
+    essence, etc.) with ``content_with_weight``, keyword index, and
+    vector.  They do NOT carry the ``knowledge_graph_kwd`` field (unlike
+    entity/relation rows from extraction).
+    """
+    from common import settings
+    from common.doc_store.doc_store_base import MatchTextExpr, OrderByExpr
+    from common.misc_utils import thread_pool_exec
+    from rag.nlp import search
+
+    condition: dict = {"compile_kwd": compile_kwd, "available_int": 1}
+    if doc_ids:
+        condition["source_doc_ids"] = list(doc_ids)
+
+    fields = [
+        "content_with_weight",
+        "summary_with_weight",
+        "source_chunk_ids",
+        "doc_id",
+        "title_kwd",
+        "topic_kwd",
+    ]
+
+    exprs = []
+    if text:
+        embd_mdl = getattr(tools, "embed_mdl", None)
+        if embd_mdl:
+            try:
+                exprs.append(await settings.retriever.get_vector(text, embd_mdl, top_n, 0.1))
+            except Exception:
+                _LOG.exception("[Wiki expand] vector build failed; using keyword match")
+        if not exprs:
+            exprs.append(
+                MatchTextExpr(
+                    ["content_ltks", "content_sm_ltks"],
+                    text,
+                    top_n,
+                )
+            )
+
+    try:
+        res = await thread_pool_exec(
+            settings.docStoreConn.search,
+            fields,
+            [],
+            condition,
+            exprs,
+            OrderByExpr(),
+            0,
+            top_n,
+            search.index_name(tenant_id),
+            [kb_id],
+        )
+        return settings.docStoreConn.get_fields(res, fields) or {}
+    except Exception:
+        _LOG.exception("[Wiki expand] search failed for kb=%s", kb_id)
+        return {}
+
+
+async def _expand_wiki_page_strategy(
+    tools,
+    kb_id: str,
+    tenant_id: str,
+    doc_ids: list[str] | None,
+    query: str,
+    seen_ids: set[str],
+    *,
+    compile_kwd: str = "wiki_page",
+    max_chunks: int = 5,
+) -> list[dict]:
+    """Expand synthesis-compiled pages: semantic search → load source chunks.
+
+    Unlike ``_expand_compiled_strategy`` (which does 1-hop entity-graph
+    navigation), synthesis pages are standalone rendered articles — we search
+    them directly and load the referenced source chunks as context.
+
+    *compile_kwd* selects which synthesis type: "wiki_page" (Wiki template),
+    "artifact_page" (Session Graph synthesis), "essence" (Session Essence).
+    """
+    # -- 1. Search synthesis pages --
+    wiki_rows = await _search_synthesis_pages(
+        tools,
+        kb_id,
+        tenant_id,
+        doc_ids,
+        query,
+        compile_kwd=compile_kwd,
+        top_n=5,
+    )
+    if not wiki_rows:
+        return []
+
+    # -- 2. Collect source_chunk_ids from matching pages --
+    by_doc: dict[str, set[str]] = {}
+    for r in wiki_rows.values():
+        doc_id = r.get("doc_id") or ""
+        for cid in r.get("source_chunk_ids") or []:
+            if cid and cid not in seen_ids:
+                by_doc.setdefault(doc_id, set()).add(cid)
+
+    # -- 3. Load chunks, assign high similarity for priority ranking --
+    new_chunks: list[dict] = []
+    for doc_id, cids in by_doc.items():
+        if len(new_chunks) >= max_chunks:
+            break
+        limit = max_chunks - len(new_chunks)
+        chunks = await _load_chunks_for_doc(tools, doc_id, list(cids)[:limit])
+        for c in chunks:
+            cid = c.get("chunk_id") or c.get("id")
+            if cid and cid not in seen_ids:
+                seen_ids.add(cid)
+                c.setdefault("similarity", 0.9)  # wiki pages rank high
+                new_chunks.append(c)
+
+    return new_chunks
+
+
 async def web_search(tools, query: str, keywords: str = "") -> dict:
     if not tools.has_web():
         return {"chunks": [], "doc_aggs": []}
diff --git a/rag/advanced_rag/knowlege_compile/runner.py b/rag/advanced_rag/knowlege_compile/runner.py
index d87dca810d..61d37cf6de 100644
--- a/rag/advanced_rag/knowlege_compile/runner.py
+++ b/rag/advanced_rag/knowlege_compile/runner.py
@@ -135,7 +135,7 @@ def load_active_templates(template_ids, tenant_id: str) -> list[tuple[str, dict]
             logging.warning("document_structure_compile: template %s config is invalid", template_id)
             continue
         kind = _compilation_template_kind(parser_cfg.get("kind"))
-        if not kind or kind == "artifacts":
+        if not kind or kind == "wiki":
             continue
         active_templates.append((template_id, parser_cfg))
     return active_templates
diff --git a/rag/svr/task_executor.py b/rag/svr/task_executor.py
index 626070375c..4f3ac2361c 100644
--- a/rag/svr/task_executor.py
+++ b/rag/svr/task_executor.py
@@ -136,7 +136,7 @@ TASK_TYPE_TO_PIPELINE_TASK_TYPE = {
     "graphrag": PipelineTaskType.GRAPH_RAG,
     "mindmap": PipelineTaskType.MINDMAP,
     "memory": PipelineTaskType.MEMORY,
-    "artifact": PipelineTaskType.ARTIFACT,
+    "wiki": PipelineTaskType.ARTIFACT,
     "skill": PipelineTaskType.SKILL,
     "structure_graph": PipelineTaskType.STRUCTURE_GRAPH,
     "structure_mindmap": PipelineTaskType.STRUCTURE_MINDMAP,
@@ -152,7 +152,7 @@ _KB_FANOUT_TASK_TYPES = [
     "graphrag",
     "raptor",
     "mindmap",
-    "artifact",
+    "wiki",
     "skill",
     "structure_graph",
     "structure_mindmap",
diff --git a/rag/svr/task_executor_refactor/dataset_wiki_generator.py b/rag/svr/task_executor_refactor/dataset_wiki_generator.py
index 8b6e8c3ed3..b81919d0cc 100644
--- a/rag/svr/task_executor_refactor/dataset_wiki_generator.py
+++ b/rag/svr/task_executor_refactor/dataset_wiki_generator.py
@@ -1028,7 +1028,7 @@ async def run_wiki(
             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":
+            if kind == "wiki":
                 eligible.append((d, template_id))
                 break
     if not eligible:
diff --git a/rag/svr/task_executor_refactor/task_handler.py b/rag/svr/task_executor_refactor/task_handler.py
index 5bb0cb0cf4..250523b5b1 100644
--- a/rag/svr/task_executor_refactor/task_handler.py
+++ b/rag/svr/task_executor_refactor/task_handler.py
@@ -255,7 +255,7 @@ class TaskHandler:
                 await self._run_graphrag(embedding_model)
             elif task_type == "mindmap":
                 ctx.progress_cb(1, "place holder")
-            elif task_type == "artifact":
+            elif task_type == "wiki":
                 from rag.svr.task_executor_refactor.dataset_wiki_generator import (
                     run_wiki,
                 )