Feat: refine the tree navigation during compilations (#17140)

api/apps/restful_apis/dataset_api.py

@@ -672,6 +672,73 @@ async def get_wiki_graph(tenant_id, dataset_id):
         return get_error_data_result(message="Internal server error")
 
 
+@manager.route("/datasets/<dataset_id>/artifacts_structure", methods=["GET"])  # noqa: F821
+@login_required
+@add_tenant_id_to_kwargs
+async def get_dataset_structure(tenant_id, dataset_id):
+    """Return the dataset-scope (KB-wide) structure graph for one kind.
+
+    GET /api/v1/datasets/<dataset_id>/artifacts_structure?kind=<kind>
+    where ``kind`` is one of:
+      graph | mindmap | timeline | session_essence | session_graph
+
+    These are the non-tree structured artifacts merged across the whole
+    dataset (written when a template has ``dataset_merge`` enabled). Response
+    mirrors the per-document structure graph so the frontend reuses its view::
+
+        {"code": 0, "data": {"kind": "<kind>", "templates": [
+            {"template_id", "template_name", "kind", "entities", "relations"}
+        ]}}
+    """
+    try:
+        kind = request.args.get("kind", "")
+        if isinstance(kind, str):
+            kind = kind.strip()
+        if not kind:
+            return get_error_data_result(
+                message="`kind` is required (one of: graph, mindmap, timeline, session_essence, session_graph).",
+                code=RetCode.ARGUMENT_ERROR,
+            )
+        if dataset_api_service._resolve_dataset_structure_kind(kind) is None:
+            return get_error_data_result(
+                message=f"Unsupported structure kind: {kind!r}. Expected one of: graph, mindmap, timeline, session_essence, session_graph.",
+                code=RetCode.ARGUMENT_ERROR,
+            )
+        success, result = await dataset_api_service.get_dataset_structure(
+            dataset_id,
+            tenant_id,
+            kind,
+        )
+        if success:
+            return get_result(data=result)
+        return get_result(data=False, message=result, code=RetCode.AUTHENTICATION_ERROR)
+    except Exception as e:
+        logging.exception(e)
+        return get_error_data_result(message="Internal server error")
+
+
+@manager.route("/datasets/<dataset_id>/artifacts/alteration", methods=["GET"])  # noqa: F821
+@login_required
+@add_tenant_id_to_kwargs
+async def get_wiki_alteration(tenant_id, dataset_id):
+    """Return document drift for the dataset Artifact wiki.
+
+    GET /api/v1/datasets/<dataset_id>/artifacts/alteration
+    Success: {"code": 0, "data": {"removed": int, "newly_uploaded": int, ...}}
+    """
+    try:
+        success, result = await dataset_api_service.get_wiki_alteration(
+            dataset_id,
+            tenant_id,
+        )
+        if success:
+            return get_result(data=result)
+        return get_result(data=False, message=result, code=RetCode.AUTHENTICATION_ERROR)
+    except Exception as e:
+        logging.exception(e)
+        return get_error_data_result(message="Internal server error")
+
+
 @manager.route("/datasets/<dataset_id>/artifacts", methods=["DELETE"])  # noqa: F821
 @login_required
 @add_tenant_id_to_kwargs
@@ -764,6 +831,28 @@ async def get_skill_tree(tenant_id, dataset_id):
         return get_error_data_result(message="Internal server error")
 
 
+@manager.route("/datasets/<dataset_id>/skills", methods=["DELETE"])  # noqa: F821
+@login_required
+@add_tenant_id_to_kwargs
+async def delete_all_skills(tenant_id, dataset_id):
+    """Delete every compiled skill for this dataset.
+
+    DELETE /api/v1/datasets/<dataset_id>/skills
+    Success: {"code": 0, "data": {"deleted": <n>}}
+    """
+    try:
+        success, result = await dataset_api_service.delete_skills(
+            dataset_id,
+            tenant_id,
+        )
+        if success:
+            return get_result(data=result)
+        return get_result(data=False, message=result, code=RetCode.AUTHENTICATION_ERROR)
+    except Exception as e:
+        logging.exception(e)
+        return get_error_data_result(message="Internal server error")
+
+
 @manager.route("/datasets/<dataset_id>/skills/<path:skill_kwd>", methods=["GET"])  # noqa: F821
 @login_required
 @add_tenant_id_to_kwargs
@@ -787,6 +876,119 @@ async def get_skill_page(tenant_id, dataset_id, skill_kwd):
         return get_error_data_result(message="Internal server error")
 
 
+@manager.route("/datasets/<dataset_id>/nav", methods=["GET"])  # noqa: F821
+@login_required
+@add_tenant_id_to_kwargs
+async def list_dataset_nav(tenant_id, dataset_id):
+    """First level of the dataset navigation tree — the top-level clusters.
+
+    GET /api/v1/datasets/<dataset_id>/nav
+    Success: {"code": 0, "data": {"total": <n>, "items": [{name, description, doc_count, type, has_children}, ...]}}
+    """
+    try:
+        success, result = await dataset_api_service.list_nav_clusters(
+            dataset_id,
+            tenant_id,
+        )
+        if success:
+            return get_result(data=result)
+        return get_result(data=False, message=result, code=RetCode.AUTHENTICATION_ERROR)
+    except Exception as e:
+        logging.exception(e)
+        return get_error_data_result(message="Internal server error")
+
+
+@manager.route("/datasets/<dataset_id>/nav/<path:name>/children", methods=["GET"])  # noqa: F821
+@login_required
+@add_tenant_id_to_kwargs
+async def list_dataset_nav_children(tenant_id, dataset_id, name):
+    """Direct children of a navigation node (hierarchical, one level per call).
+
+    GET /api/v1/datasets/<dataset_id>/nav/<name>/children
+    Success: {"code": 0, "data": {"total": <n>, "items": [{name, description, doc_count, type, doc_id, has_children}, ...]}}
+    """
+    try:
+        success, result = await dataset_api_service.list_nav_children(
+            dataset_id,
+            tenant_id,
+            name,
+        )
+        if success:
+            return get_result(data=result)
+        return get_result(data=False, message=result, code=RetCode.AUTHENTICATION_ERROR)
+    except Exception as e:
+        logging.exception(e)
+        return get_error_data_result(message="Internal server error")
+
+
+@manager.route("/datasets/<dataset_id>/nav", methods=["DELETE"])  # noqa: F821
+@login_required
+@add_tenant_id_to_kwargs
+async def delete_dataset_nav(tenant_id, dataset_id):
+    """Delete the entire dataset navigation tree.
+
+    DELETE /api/v1/datasets/<dataset_id>/nav
+    Success: {"code": 0, "data": {"deleted": <n>}}
+    """
+    try:
+        success, result = await dataset_api_service.delete_nav(
+            dataset_id,
+            tenant_id,
+        )
+        if success:
+            return get_result(data=result)
+        return get_result(data=False, message=result, code=RetCode.AUTHENTICATION_ERROR)
+    except Exception as e:
+        logging.exception(e)
+        return get_error_data_result(message="Internal server error")
+
+
+@manager.route("/datasets/<dataset_id>/nav/<path:name>", methods=["DELETE"])  # noqa: F821
+@login_required
+@add_tenant_id_to_kwargs
+async def delete_dataset_nav_node(tenant_id, dataset_id, name):
+    """Delete one navigation node and its whole subtree.
+
+    DELETE /api/v1/datasets/<dataset_id>/nav/<name>
+    Success: {"code": 0, "data": {"deleted": <n>}}
+    """
+    try:
+        success, result = await dataset_api_service.delete_nav_node(
+            dataset_id,
+            tenant_id,
+            name,
+        )
+        if success:
+            return get_result(data=result)
+        return get_result(data=False, message=result, code=RetCode.AUTHENTICATION_ERROR)
+    except Exception as e:
+        logging.exception(e)
+        return get_error_data_result(message="Internal server error")
+
+
+@manager.route("/datasets/<dataset_id>/skills/<path:skill_kwd>", methods=["DELETE"])  # noqa: F821
+@login_required
+@add_tenant_id_to_kwargs
+async def delete_skill_page(tenant_id, dataset_id, skill_kwd):
+    """Delete one compiled skill node by skill_kwd.
+
+    DELETE /api/v1/datasets/<dataset_id>/skills/<skill_kwd>
+    Success: {"code": 0, "data": {"deleted": <n>}}
+    """
+    try:
+        success, result = await dataset_api_service.delete_skill(
+            dataset_id,
+            tenant_id,
+            skill_kwd,
+        )
+        if success:
+            return get_result(data=result)
+        return get_result(data=False, message=result, code=RetCode.AUTHENTICATION_ERROR)
+    except Exception as e:
+        logging.exception(e)
+        return get_error_data_result(message="Internal server error")
+
+
 # The two artifact-commit endpoints
 #   GET /datasets/<dataset_id>/artifacts/<page_type>/<path:slug>/commits
 #   GET /datasets/<dataset_id>/artifacts/commits/<commit_id>

api/apps/services/dataset_api_service.py

@@ -35,7 +35,22 @@ from api.utils.api_utils import deep_merge, get_parser_config, remap_dictionary_
 from common.misc_utils import thread_pool_exec
 from rag.advanced_rag.knowlege_compile.wiki import WIKI_PAGE_COMPILE_KWD
 
-_VALID_INDEX_TYPES = {"graph", "raptor", "mindmap", "artifact", "skill"}
+# KB-wide structure-graph merge index types. Each (re)builds the ``dataset_graph``
+# rows for one structure kind via ``rebuild_dataset_structure_graph_json``; the
+# task_type equals the index_type and the KB task-id column is ``<type>_task_id``.
+# The value is the friendly kind resolvable through ``_resolve_dataset_structure_kind``
+# (defined below), except ``structure`` which is the merge-all variant.
+_STRUCTURE_INDEX_TYPE_TO_KIND = {
+    "structure_graph": "graph",
+    "structure_mindmap": "mindmap",
+    "timeline": "timeline",
+    "session_graph": "session_graph",
+    "session_essence": "session_essence",
+    "structure": None,  # merge-all: rebuild every dataset-merge kind
+}
+_STRUCTURE_INDEX_TYPES = frozenset(_STRUCTURE_INDEX_TYPE_TO_KIND)
+
+_VALID_INDEX_TYPES = {"graph", "raptor", "mindmap", "artifact", "skill"} | set(_STRUCTURE_INDEX_TYPES)
 
 _INDEX_TYPE_TO_TASK_TYPE = {
     "graph": "graphrag",
@@ -43,6 +58,9 @@ _INDEX_TYPE_TO_TASK_TYPE = {
     "mindmap": "mindmap",
     "artifact": "artifact",
     "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.
+    **{t: t for t in _STRUCTURE_INDEX_TYPES},
 }
 
 _INDEX_TYPE_TO_TASK_ID_FIELD = {
@@ -51,6 +69,7 @@ _INDEX_TYPE_TO_TASK_ID_FIELD = {
     "mindmap": "mindmap_task_id",
     "artifact": "artifact_task_id",
     "skill": "skill_task_id",
+    **{t: f"{t}_task_id" for t in _STRUCTURE_INDEX_TYPES},
 }
 
 _INDEX_TYPE_TO_DISPLAY_NAME = {
@@ -59,6 +78,12 @@ _INDEX_TYPE_TO_DISPLAY_NAME = {
     "mindmap": "Mindmap",
     "artifact": "Artifact",
     "skill": "Skill",
+    "structure_graph": "Structure Graph",
+    "structure_mindmap": "Structure Mindmap",
+    "timeline": "Timeline",
+    "session_graph": "Session Graph",
+    "session_essence": "Session Essence",
+    "structure": "Structure",
 }
 
 
@@ -884,6 +909,18 @@ def delete_index(dataset_id: str, tenant_id: str, index_type: str, wipe: bool =
         from rag.nlp import search
 
         settings.docStoreConn.delete({"compile_kwd": ["skill", "skill_all"]}, search.index_name(kb.tenant_id), dataset_id)
+    elif wipe and index_type in _STRUCTURE_INDEX_TYPES:
+        from rag.nlp import search
+
+        # Wipe the merged KB-wide dataset_graph rows for the requested kind
+        # (all kinds for the merge-all "structure" type). The per-document
+        # entity/relation rows the merge reads from are left intact.
+        friendly = _STRUCTURE_INDEX_TYPE_TO_KIND.get(index_type)
+        resolved_kind = _resolve_dataset_structure_kind(friendly) if friendly else None
+        condition: dict = {"knowledge_graph_kwd": ["dataset_graph"]}
+        if resolved_kind:
+            condition["compilation_template_kind_kwd"] = [resolved_kind]
+        settings.docStoreConn.delete(condition, search.index_name(kb.tenant_id), dataset_id)
 
     KnowledgebaseService.update_by_id(kb.id, {task_id_field: "", task_finish_at_field: None})
     return True, {}
@@ -1502,6 +1539,124 @@ def _wiki_index_or_none(tenant_id: str, kb_id: str):
     return _compiled_index_or_none(tenant_id, kb_id)
 
 
+def _compilation_template_kind(kind) -> str:
+    if not isinstance(kind, str):
+        return ""
+    normalized = kind.strip().lower().replace("-", "_")
+    if normalized in {"pageindex", "page_index", "knowledge_graph"}:
+        return "timeline"
+    return normalized
+
+
+def _normalize_compilation_template_group_ids(raw) -> list[str]:
+    if isinstance(raw, str):
+        raw = [raw]
+    if not isinstance(raw, list):
+        return []
+    ids: list[str] = []
+    seen: set[str] = set()
+    for group_id in raw:
+        if not isinstance(group_id, str):
+            continue
+        group_id = group_id.strip()
+        if group_id and group_id not in seen:
+            seen.add(group_id)
+            ids.append(group_id)
+    return ids
+
+
+def _extract_pipeline_compiler_group_ids(dsl) -> list[str]:
+    if isinstance(dsl, str):
+        try:
+            dsl = json.loads(dsl)
+        except Exception:
+            return []
+    if not isinstance(dsl, dict):
+        return []
+    components = dsl.get("components")
+    if not isinstance(components, dict):
+        return []
+
+    group_ids: list[str] = []
+    seen: set[str] = set()
+    for component in components.values():
+        if not isinstance(component, dict):
+            continue
+        obj = component.get("obj") if isinstance(component.get("obj"), dict) else {}
+        component_name = obj.get("component_name") or component.get("component_name") or component.get("name")
+        if not isinstance(component_name, str) or component_name.lower() != "compiler":
+            continue
+        candidates = [
+            obj.get("params") if isinstance(obj.get("params"), dict) else {},
+            obj,
+            component.get("params") if isinstance(component.get("params"), dict) else {},
+            component,
+        ]
+        for candidate in candidates:
+            for key in ("compilation_template_group_ids", "compilation_template_group_id"):
+                for group_id in _normalize_compilation_template_group_ids(candidate.get(key)):
+                    if group_id not in seen:
+                        seen.add(group_id)
+                        group_ids.append(group_id)
+    return group_ids
+
+
+def _template_is_wiki(template: dict | None) -> 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"
+
+
+def _group_has_wiki_template(group_id: str, tenant_id: 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
+
+
+def _parser_config_has_wiki_template(parser_config, tenant_id: 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))
+        if template_cache[template_id]:
+            return True
+    return False
+
+
+def _pipeline_has_wiki_compiler(
+    pipeline_id: str,
+    tenant_id: str,
+    pipeline_cache: dict[str, bool],
+    group_cache: dict[str, bool],
+) -> bool:
+    pipeline_id = (pipeline_id or "").strip()
+    if not pipeline_id:
+        return False
+    if pipeline_id in pipeline_cache:
+        return pipeline_cache[pipeline_id]
+
+    from api.db.services.canvas_service import UserCanvasService
+
+    ok, canvas = UserCanvasService.get_by_id(pipeline_id)
+    if not ok or not canvas:
+        pipeline_cache[pipeline_id] = False
+        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
+
+
 def _skill_index_or_none(tenant_id: str, kb_id: str):
     return _compiled_index_or_none(tenant_id, kb_id)
 
@@ -1543,6 +1698,253 @@ async def has_any_wiki(dataset_id: str, tenant_id: str):
     return True, {"has": bool(total)}
 
 
+# Dataset-scope structure kinds the artifacts_structure API serves. Keys are the
+# friendly names the frontend passes; values are the template's *top-level* kind
+# as stamped on ``dataset_graph`` rows. The canonical names map to themselves so
+# a caller can pass either form. Note this deliberately does NOT reuse
+# ``_compilation_template_kind`` — that helper folds ``knowledge_graph`` into
+# ``timeline`` and would merge distinct kinds here.
+_DATASET_STRUCTURE_ROW_KWD = "dataset_graph"
+_DATASET_STRUCTURE_KIND_ALIASES = {
+    "graph": "knowledge_graph",
+    "knowledge_graph": "knowledge_graph",
+    "mindmap": "mind_map",
+    "mind_map": "mind_map",
+    "timeline": "timeline",
+    "session_essence": "session_essence",
+    "session_graph": "session_graph",
+}
+
+
+def _resolve_dataset_structure_kind(kind) -> str | None:
+    """Map a friendly/canonical kind string to the stored top-level kind."""
+    if not isinstance(kind, str):
+        return None
+    return _DATASET_STRUCTURE_KIND_ALIASES.get(kind.strip().lower().replace("-", "_"))
+
+
+async def get_dataset_structure(dataset_id: str, tenant_id: str, kind: str):
+    """Load the dataset-scope (KB-wide) structure graph for one ``kind``.
+
+    ``kind`` is one of ``graph`` / ``mindmap`` / ``timeline`` /
+    ``session_essence`` / ``session_graph``. Reads the
+    ``knowledge_graph_kwd="dataset_graph"`` rows written by
+    ``rebuild_dataset_structure_graph_json`` (one per template), filters to the
+    requested kind, and returns them grouped by template — mirroring the
+    per-document ``structure/graph`` response so the frontend graph view is
+    reused unchanged.
+
+    Returns ``(True, {"kind": <kind>, "templates": [...]})`` or
+    ``(False, message)`` on auth/validation failure.
+    """
+    if not KnowledgebaseService.accessible(dataset_id, tenant_id):
+        return False, "No authorization."
+
+    resolved_kind = _resolve_dataset_structure_kind(kind)
+    if not resolved_kind:
+        return False, f"Unsupported structure kind: {kind!r}. Expected one of: graph, mindmap, timeline, session_essence, session_graph."
+
+    _, kb = KnowledgebaseService.get_by_id(dataset_id)
+    empty = {"kind": kind, "templates": []}
+
+    pack = _compiled_index_or_none(kb.tenant_id, dataset_id)
+    if pack is None:
+        return True, empty
+    index_nm, _ = pack
+
+    from common.doc_store.doc_store_base import OrderByExpr
+    from api.db.services.compilation_template_service import CompilationTemplateService
+
+    select_fields = [
+        "content_with_weight",
+        "compile_kwd",
+        "compilation_template_ids",
+        "compilation_template_kind_kwd",
+    ]
+    try:
+        res = await thread_pool_exec(
+            settings.docStoreConn.search,
+            select_fields,
+            [],
+            {"knowledge_graph_kwd": [_DATASET_STRUCTURE_ROW_KWD]},
+            [],
+            OrderByExpr(),
+            0,
+            1000,
+            index_nm,
+            [dataset_id],
+        )
+        rows = settings.docStoreConn.get_fields(res, select_fields) or {}
+    except Exception:
+        logging.exception("get_dataset_structure: docStore search failed for kb=%s", dataset_id)
+        return True, empty
+
+    def _row_template_id(row: dict) -> str | None:
+        raw = row.get("compilation_template_ids")
+        if isinstance(raw, list):
+            for v in raw:
+                if isinstance(v, str) and v.strip():
+                    return v.strip()
+        if isinstance(raw, str) and raw.strip():
+            return raw.strip()
+        return None
+
+    # Resolve a template's top-level kind + display name, memoized. Used both to
+    # label buckets and as a fallback for rows written before the kind stamp
+    # (they carry ``compilation_template_ids`` but no ``compilation_template_kind_kwd``).
+    template_kind_cache: dict[str, str | None] = {}
+    template_name_cache: dict[str, str] = {}
+
+    def _template_meta(tid: str | None) -> str | None:
+        if not tid:
+            return None
+        if tid in template_kind_cache:
+            return template_kind_cache[tid]
+        top_kind = None
+        try:
+            saved = CompilationTemplateService.get_saved(tid, tenant_id)
+            if saved:
+                top_kind = (saved.get("kind") or "").strip() or None
+                template_name_cache[tid] = saved.get("name") or tid
+        except Exception:
+            logging.exception("get_dataset_structure: template lookup failed for %s", tid)
+        template_kind_cache[tid] = top_kind
+        return top_kind
+
+    grouped: dict[str, dict] = {}
+    for row in 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
+
+        try:
+            graph = json.loads(row.get("content_with_weight") or "{}")
+        except Exception:
+            continue
+        if not isinstance(graph, dict):
+            continue
+        entities = graph.get("entities") or []
+        relations = graph.get("relations") or []
+        if not entities and not relations:
+            continue
+
+        bucket_id = tid or f"kind:{resolved_kind}"
+        if bucket_id not in grouped:
+            if tid and tid not in template_name_cache:
+                _template_meta(tid)
+            grouped[bucket_id] = {
+                "template_id": bucket_id,
+                "template_name": template_name_cache.get(tid or "", bucket_id),
+                "kind": row_kind or resolved_kind,
+                "entities": [],
+                "relations": [],
+            }
+        grouped[bucket_id]["entities"].extend(entities)
+        grouped[bucket_id]["relations"].extend(relations)
+
+    templates_out = [g for g in grouped.values() if g["entities"] or g["relations"]]
+    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"
+
+    docs, _ = await thread_pool_exec(
+        DocumentService.get_by_kb_id,
+        kb_id=dataset_id,
+        page_number=0,
+        items_per_page=0,
+        orderby="create_time",
+        desc=False,
+        keywords="",
+        run_status=[],
+        types=[],
+        suffix=[],
+    )
+    current_doc_ids = {str(doc.get("id")) for doc in docs or [] if doc.get("id")}
+
+    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 = {}
+
+            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
+
+    template_cache: dict[str, bool] = {}
+    group_cache: dict[str, bool] = {}
+    pipeline_cache: dict[str, bool] = {}
+    eligible_wiki_doc_ids: 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)
+            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)
+
+    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 list_wiki_pages(
     dataset_id: str,
     tenant_id: str,
@@ -1892,6 +2294,70 @@ async def get_skill_tree(dataset_id: str, tenant_id: str):
     }
 
 
+async def delete_skills(dataset_id: str, tenant_id: str):
+    """Delete every compiled skill row (``skill`` + ``skill_all``) for a dataset.
+
+    Returns ``(True, {"deleted": <n>})`` on success. When the tenant index does
+    not exist yet there is nothing to delete, so it succeeds with ``0``.
+    """
+    if not KnowledgebaseService.accessible(dataset_id, tenant_id):
+        return False, "No authorization."
+    _, kb = KnowledgebaseService.get_by_id(dataset_id)
+
+    pack = _skill_index_or_none(kb.tenant_id, dataset_id)
+    if pack is None:
+        return True, {"deleted": 0}
+    index_nm, _ = pack
+
+    try:
+        deleted = settings.docStoreConn.delete(
+            {"compile_kwd": [_SKILL_COMPILE_KWD, _SKILL_ALL_COMPILE_KWD]},
+            index_nm,
+            dataset_id,
+        )
+    except Exception:
+        logging.exception("delete_skills: docStore delete failed for kb=%s", dataset_id)
+        return False, "Failed to delete skills."
+
+    # Clear the skill compilation markers so the dataset reflects "no skill"
+    # (and a later re-compile isn't short-circuited by a stale task id).
+    try:
+        KnowledgebaseService.update_by_id(kb.id, {"skill_task_id": "", "skill_task_finish_at": None})
+    except Exception:
+        logging.exception("delete_skills: failed clearing skill task markers for kb=%s", dataset_id)
+
+    return True, {"deleted": int(deleted or 0)}
+
+
+async def delete_skill(dataset_id: str, tenant_id: str, skill_kwd: str):
+    """Delete a single compiled skill node identified by ``skill_kwd``.
+
+    Removes the per-node ``skill`` row(s) matching ``skill_kwd`` (the same
+    identity ``get_skill_page`` reads). The aggregate ``skill_all`` tree row is
+    left untouched. Returns ``(True, {"deleted": <n>})``.
+    """
+    if not KnowledgebaseService.accessible(dataset_id, tenant_id):
+        return False, "No authorization."
+    _, kb = KnowledgebaseService.get_by_id(dataset_id)
+
+    pack = _skill_index_or_none(kb.tenant_id, dataset_id)
+    if pack is None:
+        return True, {"deleted": 0}
+    index_nm, _ = pack
+
+    try:
+        deleted = settings.docStoreConn.delete(
+            {"compile_kwd": [_SKILL_COMPILE_KWD], "skill_kwd": [skill_kwd]},
+            index_nm,
+            dataset_id,
+        )
+    except Exception:
+        logging.exception("delete_skill: docStore delete failed for kb=%s skill=%s", dataset_id, skill_kwd)
+        return False, "Failed to delete skill."
+
+    return True, {"deleted": int(deleted or 0)}
+
+
 async def get_skill_page(dataset_id: str, tenant_id: str, skill_kwd: str):
     """Fetch the full markdown body for a single skill node."""
     if not KnowledgebaseService.accessible(dataset_id, tenant_id):
@@ -1957,6 +2423,211 @@ async def get_skill_page(dataset_id: str, tenant_id: str, skill_kwd: str):
     }
 
 
+# ---------------------------------------------------------------------------
+# Dataset navigation tree (written by rag/advanced_rag/knowlege_compile/
+# dataset_nav.py as nav_cluster / nav_doc rows). Loaded hierarchically: the
+# first call returns the top clusters (parent = "root"); clicking a cluster
+# returns its direct children (sub-clusters + document leaves). These rows are
+# ``available_int=0`` so they're read via docStoreConn.search directly.
+# ---------------------------------------------------------------------------
+
+_NAV_COMPILE_KWD = "dataset_nav"
+_NAV_ROOT_PARENT = "root"
+_NAV_FIELDS = [
+    "id",
+    "name",
+    "type_kwd",
+    "content_with_weight",
+    "doc_count_int",
+    "doc_ids_kwd",
+    "doc_id",
+    "depth_int",
+    "parent_kwd",
+]
+
+
+def _nav_item(row: dict) -> dict:
+    """Shape one nav row into a UI node: name, description, doc count, type."""
+    try:
+        payload = json.loads(row.get("content_with_weight") or "{}")
+    except Exception:
+        payload = {}
+    is_cluster = (row.get("type_kwd") or payload.get("type")) == "nav_cluster"
+    return {
+        "name": row.get("name") or "",
+        "description": payload.get("description") or "",
+        # doc_id count under this node: the cluster's tally, or 1 for a leaf.
+        "doc_count": int(row.get("doc_count_int") or 0) if is_cluster else 1,
+        "type": "cluster" if is_cluster else "doc",
+        "doc_id": None if is_cluster else (row.get("doc_id") or row.get("name")),
+        "has_children": is_cluster,
+    }
+
+
+async def _nav_search(dataset_id: str, tenant_id: str, condition: dict, page: int, page_size: int):
+    """Run one nav-tree search and shape the hits into UI nodes."""
+    if not KnowledgebaseService.accessible(dataset_id, tenant_id):
+        return False, "No authorization."
+    _, kb = KnowledgebaseService.get_by_id(dataset_id)
+
+    pack = _compiled_index_or_none(kb.tenant_id, dataset_id)
+    if pack is None:
+        return True, {"total": 0, "items": []}
+    index_nm, _ = pack
+
+    from common.doc_store.doc_store_base import OrderByExpr
+
+    page = max(1, int(page or 1))
+    page_size = max(1, min(int(page_size or 1000), 2000))
+    offset = (page - 1) * page_size
+
+    order_by = OrderByExpr()
+    try:
+        # Biggest clusters first; leaves (no doc_count_int) fall to the end.
+        order_by.desc("doc_count_int")
+    except Exception:
+        order_by = OrderByExpr()
+
+    try:
+        res = settings.docStoreConn.search(
+            select_fields=_NAV_FIELDS,
+            highlight_fields=[],
+            condition=condition,
+            match_expressions=[],
+            order_by=order_by,
+            offset=offset,
+            limit=page_size,
+            index_names=index_nm,
+            knowledgebase_ids=[dataset_id],
+        )
+        field_map = settings.docStoreConn.get_fields(res, _NAV_FIELDS)
+    except Exception:
+        logging.exception("dataset_nav: docStore search failed for kb=%s", dataset_id)
+        return True, {"total": 0, "items": []}
+
+    total = settings.docStoreConn.get_total(res)
+    items = [_nav_item(row) for row in (field_map or {}).values()]
+    return True, {"total": int(total or 0), "items": items}
+
+
+async def list_nav_clusters(dataset_id: str, tenant_id: str, page: int = 1, page_size: int = 1000):
+    """First level of the nav tree: the clusters with no parent."""
+    condition = {
+        "compile_kwd": [_NAV_COMPILE_KWD],
+        "type_kwd": ["nav_cluster"],
+        "parent_kwd": [_NAV_ROOT_PARENT],
+    }
+    return await _nav_search(dataset_id, tenant_id, condition, page, page_size)
+
+
+async def list_nav_children(dataset_id: str, tenant_id: str, name: str, page: int = 1, page_size: int = 1000):
+    """Direct children of the node ``name`` — sub-clusters and document leaves.
+
+    One level at a time (lazy) so the tree loads hierarchically as the user
+    expands each node.
+    """
+    if not isinstance(name, str) or not name.strip():
+        return True, {"total": 0, "items": []}
+    condition = {
+        "compile_kwd": [_NAV_COMPILE_KWD],
+        "parent_kwd": [name.strip()],
+    }
+    return await _nav_search(dataset_id, tenant_id, condition, page, page_size)
+
+
+async def delete_nav(dataset_id: str, tenant_id: str):
+    """Delete the entire dataset navigation tree for a dataset.
+
+    Returns ``(True, {"deleted": <n>})``; succeeds with ``0`` when there is no
+    index yet.
+    """
+    if not KnowledgebaseService.accessible(dataset_id, tenant_id):
+        return False, "No authorization."
+    _, kb = KnowledgebaseService.get_by_id(dataset_id)
+
+    pack = _compiled_index_or_none(kb.tenant_id, dataset_id)
+    if pack is None:
+        return True, {"deleted": 0}
+    index_nm, _ = pack
+
+    try:
+        deleted = settings.docStoreConn.delete(
+            {"compile_kwd": [_NAV_COMPILE_KWD]},
+            index_nm,
+            dataset_id,
+        )
+    except Exception:
+        logging.exception("delete_nav: docStore delete failed for kb=%s", dataset_id)
+        return False, "Failed to delete the navigation tree."
+
+    return True, {"deleted": int(deleted or 0)}
+
+
+async def delete_nav_node(dataset_id: str, tenant_id: str, name: str):
+    """Delete one navigation node (identified by ``name``) and its whole subtree.
+
+    Children reference their parent by ``name`` (``parent_kwd``), so removing a
+    cluster without its descendants would leave them orphaned in the tree view.
+    We therefore walk the subtree top-down and delete every node in it.
+    """
+    if not isinstance(name, str) or not name.strip():
+        return True, {"deleted": 0}
+    name = name.strip()
+
+    if not KnowledgebaseService.accessible(dataset_id, tenant_id):
+        return False, "No authorization."
+    _, kb = KnowledgebaseService.get_by_id(dataset_id)
+
+    pack = _compiled_index_or_none(kb.tenant_id, dataset_id)
+    if pack is None:
+        return True, {"deleted": 0}
+    index_nm, _ = pack
+
+    from common.doc_store.doc_store_base import OrderByExpr
+
+    # Collect the node plus every descendant, level by level via parent_kwd.
+    names: set[str] = {name}
+    frontier: list[str] = [name]
+    for _ in range(64):  # depth guard against a malformed (cyclic) tree
+        if not frontier:
+            break
+        try:
+            res = settings.docStoreConn.search(
+                select_fields=["name"],
+                highlight_fields=[],
+                condition={"compile_kwd": [_NAV_COMPILE_KWD], "parent_kwd": frontier},
+                match_expressions=[],
+                order_by=OrderByExpr(),
+                offset=0,
+                limit=10000,
+                index_names=index_nm,
+                knowledgebase_ids=[dataset_id],
+            )
+            rows = settings.docStoreConn.get_fields(res, ["name"]) or {}
+        except Exception:
+            logging.exception("delete_nav_node: subtree scan failed for kb=%s name=%s", dataset_id, name)
+            break
+        nxt: list[str] = []
+        for row in rows.values():
+            child = row.get("name")
+            if isinstance(child, str) and child and child not in names:
+                names.add(child)
+                nxt.append(child)
+        frontier = nxt
+
+    try:
+        deleted = settings.docStoreConn.delete(
+            {"compile_kwd": [_NAV_COMPILE_KWD], "name": list(names)},
+            index_nm,
+            dataset_id,
+        )
+    except Exception:
+        logging.exception("delete_nav_node: docStore delete failed for kb=%s name=%s", dataset_id, name)
+        return False, "Failed to delete the navigation node."
+
+    return True, {"deleted": int(deleted or 0)}
+
+
 async def update_wiki_page(
     dataset_id: str,
     tenant_id: str,

api/db/__init__.py

@@ -85,6 +85,16 @@ PIPELINE_SPECIAL_PROGRESS_FREEZE_TASK_TYPES = {
     PipelineTaskType.MINDMAP.lower(),
     PipelineTaskType.ARTIFACT.lower(),
     PipelineTaskType.SKILL.lower(),
+    # Structure-graph merge fan-out task types. These are the raw task_type
+    # strings (== the index type), which — unlike the types above — do not equal
+    # their PipelineTaskType value lowercased (e.g. "structure_graph" vs
+    # "structuregraph"), so they are listed literally.
+    "structure_graph",
+    "structure_mindmap",
+    "timeline",
+    "session_graph",
+    "session_essence",
+    "structure",
 }
 
 

api/db/db_models.py

@@ -866,6 +866,21 @@ class Knowledgebase(DataBaseModel):
     artifact_task_finish_at = DateTimeField(null=True)
     skill_task_id = CharField(max_length=32, null=True, help_text="Skill generation task ID", index=True)
     skill_task_finish_at = DateTimeField(null=True)
+    # KB-wide structure-graph merge tasks, one traceable task id per merged kind
+    # (rebuild_dataset_structure_graph_json). ``structure_task_id`` is the
+    # merge-all variant that rebuilds every dataset-merge kind at once.
+    structure_graph_task_id = CharField(max_length=32, null=True, help_text="Structure graph merge task ID", index=True)
+    structure_graph_task_finish_at = DateTimeField(null=True)
+    structure_mindmap_task_id = CharField(max_length=32, null=True, help_text="Structure mindmap merge task ID", index=True)
+    structure_mindmap_task_finish_at = DateTimeField(null=True)
+    timeline_task_id = CharField(max_length=32, null=True, help_text="Timeline merge task ID", index=True)
+    timeline_task_finish_at = DateTimeField(null=True)
+    session_graph_task_id = CharField(max_length=32, null=True, help_text="Session graph merge task ID", index=True)
+    session_graph_task_finish_at = DateTimeField(null=True)
+    session_essence_task_id = CharField(max_length=32, null=True, help_text="Session essence merge task ID", index=True)
+    session_essence_task_finish_at = DateTimeField(null=True)
+    structure_task_id = CharField(max_length=32, null=True, help_text="Structure merge-all task ID", index=True)
+    structure_task_finish_at = DateTimeField(null=True)
 
     status = CharField(max_length=1, null=True, help_text="is it validate(0: wasted, 1: validate)", default="1", index=True)
 
@@ -1464,6 +1479,7 @@ def alter_db_drop_index(migrator, table_name, index_name):
         # logging.critical(f"Failed to rename {settings.DATABASE_TYPE.upper()}.{table_name} column {old_column_name} to {new_column_name}, error: {ex}")
         pass
 
+
 def ensure_model_indexes(migrator):
     """Create indexes declared by the Peewee models when they are missing."""
     members = inspect.getmembers(sys.modules[__name__], inspect.isclass)
@@ -1756,6 +1772,9 @@ def migrate_db():
     alter_db_add_column(migrator, "knowledgebase", "artifact_task_finish_at", DateTimeField(null=True))
     alter_db_add_column(migrator, "knowledgebase", "skill_task_id", CharField(max_length=32, null=True, help_text="Skill generation task ID", index=True))
     alter_db_add_column(migrator, "knowledgebase", "skill_task_finish_at", DateTimeField(null=True))
+    for _structure_type in ("structure_graph", "structure_mindmap", "timeline", "session_graph", "session_essence", "structure"):
+        alter_db_add_column(migrator, "knowledgebase", f"{_structure_type}_task_id", CharField(max_length=32, null=True, help_text=f"{_structure_type} merge task ID", index=True))
+        alter_db_add_column(migrator, "knowledgebase", f"{_structure_type}_task_finish_at", DateTimeField(null=True))
     alter_db_column_type(migrator, "tenant_llm", "api_key", TextField(null=True, help_text="API KEY"))
     alter_db_add_column(migrator, "tenant_llm", "status", CharField(max_length=1, null=False, help_text="is it validate(0: wasted, 1: validate)", default="1", index=True))
     alter_db_add_column(migrator, "connector2kb", "auto_parse", CharField(max_length=1, null=False, default="1", index=False))

api/db/init_data/compilation_templates/empty.yaml

@@ -1,17 +0,0 @@
-kind: empty
-display_name: Empty
-config:
-  kind: empty
-  entity:
-    description: ''
-    fields:
-      - type: ''
-        description: ''
-        rule: ''
-  relation:
-    description: ''
-    fields:
-      - type: ''
-        description: ''
-        rule: ''
-  global_rules: ''

api/db/services/document_service.py

@@ -498,6 +498,15 @@ class DocumentService(CommonService):
         except Exception as e:
             logging.error(f"Failed to delete chunks from doc store for document {doc.id}: {e}")
 
+        # Ref-counted cleanup of wiki/artifact products this doc fed into
+        # (non-critical, log and continue). A product shared by other docs
+        # survives; one this doc solely owned is removed.
+        try:
+            if chunk_index_exists:
+                cls.remove_artifact_products(doc, tenant_id)
+        except Exception as e:
+            logging.warning(f"Failed to clean up artifact products for document {doc.id}: {e}")
+
         # Prune this doc's line from the KB's tree-kind navigation
         # markdown (best-effort — the markdown is a downstream artifact,
         # and failure here must not block the document delete).
@@ -558,6 +567,103 @@ class DocumentService(CommonService):
                     settings.STORAGE_IMPL.rm(doc.kb_id, cid)
             page += 1
 
+    @classmethod
+    def remove_artifact_products(cls, doc, tenant_id):
+        """Reference-counted cleanup of KB-scoped wiki/artifact products
+        in the doc store when a document is deleted.
+
+        Every derived artifact row (pages, entities, relations, drafts,
+        topics, reduce/plan aggregates) carries a ``source_doc_ids`` list
+        of the documents that contributed to it. On delete we detach
+        ``doc.id`` from that list and drop the row only when this document
+        was its sole contributor — a product shared by other docs
+        survives. ``artifact_map_extract`` resume rows are 1:1 with a
+        document and are removed directly by ``doc_id``.
+
+        The compile_kwd set is pulled from the wiki generator so new
+        artifact row types are covered automatically (single source of
+        truth). Deletion is not a hot path, so the module import cost is
+        acceptable here.
+        """
+        from rag.svr.task_executor_refactor.dataset_wiki_generator import (
+            WIKI_MAP_COMPILE_KWD,
+            WIKI_DERIVED_COMPILE_KWDS,
+        )
+
+        index = search.index_name(tenant_id)
+        if not settings.docStoreConn.index_exist(index, doc.kb_id):
+            return
+
+        # 1. Per-doc MAP resume rows are keyed by the real doc_id.
+        settings.docStoreConn.delete(
+            {"compile_kwd": [WIKI_MAP_COMPILE_KWD], "doc_id": doc.id},
+            index,
+            doc.kb_id,
+        )
+
+        # 2. Derived KB-scoped rows: reference-counted via source_doc_ids.
+        # Read every row this doc contributed to, partitioning into rows it
+        # solely owned (delete by id) vs. rows shared with other docs
+        # (detach this doc). Reading first — rather than a blanket
+        # ``must_not exists`` sweep — avoids deleting rows that legitimately
+        # carry no source_doc_ids.
+        derived_kwds = list(WIKI_DERIVED_COMPILE_KWDS)
+        select_fields = ["id", "source_doc_ids"]
+        sole_owner_ids: list[str] = []
+        shared_seen = False
+        offset = 0
+        page_size = 1000
+        while True:
+            res = settings.docStoreConn.search(
+                select_fields,
+                [],
+                {"compile_kwd": derived_kwds, "source_doc_ids": [doc.id]},
+                [],
+                OrderByExpr(),
+                offset,
+                page_size,
+                index,
+                [doc.kb_id],
+            )
+            field_map = settings.docStoreConn.get_fields(res, select_fields) or {}
+            if not field_map:
+                break
+            for row_id, row in field_map.items():
+                raw = row.get("source_doc_ids")
+                if isinstance(raw, str):
+                    owners = [raw] if raw else []
+                elif isinstance(raw, list):
+                    owners = [d for d in raw if isinstance(d, str) and d]
+                else:
+                    owners = []
+                if any(d != doc.id for d in owners):
+                    shared_seen = True
+                else:
+                    sole_owner_ids.append(row_id)
+            if len(field_map) < page_size:
+                break
+            offset += page_size
+
+        # Drop rows this document solely owned (delete by id in batches).
+        for i in range(0, len(sole_owner_ids), page_size):
+            settings.docStoreConn.delete(
+                {"id": sole_owner_ids[i : i + page_size]},
+                index,
+                doc.kb_id,
+            )
+
+        # Detach this document from rows still owned by others. The filter
+        # guarantees source_doc_ids contains doc.id, so the store's
+        # list-remove is safe; any sole-owner rows already deleted above are
+        # simply not matched.
+        if shared_seen:
+            settings.docStoreConn.update(
+                {"compile_kwd": derived_kwds, "source_doc_ids": doc.id},
+                {"remove": {"source_doc_ids": doc.id}},
+                index,
+                doc.kb_id,
+            )
+
     @classmethod
     @DB.connection_context()
     def get_newly_uploaded(cls):
@@ -1101,7 +1207,20 @@ def queue_raptor_o_graphrag_tasks(sample_doc, ty, priority, fake_doc_id="", doc_
     """
     if doc_ids is None:
         doc_ids = []
-    assert ty in ["graphrag", "raptor", "mindmap", "artifact", "skill"], "type should be graphrag, raptor, mindmap, artifact or skill"
+    assert ty in [
+        "graphrag",
+        "raptor",
+        "mindmap",
+        "artifact",
+        "skill",
+        # KB-wide structure-graph merge task types (rebuild dataset_graph rows).
+        "structure_graph",
+        "structure_mindmap",
+        "timeline",
+        "session_graph",
+        "session_essence",
+        "structure",
+    ], f"unsupported task type '{ty}'"
 
     chunking_config = DocumentService.get_chunking_config(sample_doc["id"])
     hasher = xxhash.xxh64()

api/db/services/knowledgebase_service.py

@@ -310,6 +310,18 @@ class KnowledgebaseService(CommonService):
             cls.model.artifact_task_finish_at,
             cls.model.skill_task_id,
             cls.model.skill_task_finish_at,
+            cls.model.structure_graph_task_id,
+            cls.model.structure_graph_task_finish_at,
+            cls.model.structure_mindmap_task_id,
+            cls.model.structure_mindmap_task_finish_at,
+            cls.model.timeline_task_id,
+            cls.model.timeline_task_finish_at,
+            cls.model.session_graph_task_id,
+            cls.model.session_graph_task_finish_at,
+            cls.model.session_essence_task_id,
+            cls.model.session_essence_task_finish_at,
+            cls.model.structure_task_id,
+            cls.model.structure_task_finish_at,
             cls.model.create_time,
             cls.model.update_time,
         ]

api/db/services/pipeline_operation_log_service.py

@@ -32,6 +32,25 @@ from common.misc_utils import get_uuid
 from common.time_utils import current_timestamp, datetime_format
 
 
+# KB-level fan-out pipeline task types (task row carries a fake doc_id; the real
+# participants live in task["doc_ids"]) → the KB ``<type>_task_finish_at`` column
+# stamped when the task completes. Membership also marks a task as KB-scoped so
+# the per-document progress update is skipped.
+_PIPELINE_TASK_TYPE_TO_FINISH_FIELD = {
+    PipelineTaskType.GRAPH_RAG: "graphrag_task_finish_at",
+    PipelineTaskType.RAPTOR: "raptor_task_finish_at",
+    PipelineTaskType.MINDMAP: "mindmap_task_finish_at",
+    PipelineTaskType.ARTIFACT: "artifact_task_finish_at",
+    PipelineTaskType.SKILL: "skill_task_finish_at",
+    PipelineTaskType.STRUCTURE_GRAPH: "structure_graph_task_finish_at",
+    PipelineTaskType.STRUCTURE_MINDMAP: "structure_mindmap_task_finish_at",
+    PipelineTaskType.TIMELINE: "timeline_task_finish_at",
+    PipelineTaskType.SESSION_GRAPH: "session_graph_task_finish_at",
+    PipelineTaskType.SESSION_ESSENCE: "session_essence_task_finish_at",
+    PipelineTaskType.STRUCTURE: "structure_task_finish_at",
+}
+
+
 class PipelineOperationLogService(CommonService):
     model = PipelineOperationLog
 
@@ -99,7 +118,7 @@ class PipelineOperationLogService(CommonService):
             referred_document_id = document_id
 
         # no need to update document for KB-level fan-out tasks
-        if task_type not in [PipelineTaskType.GRAPH_RAG, PipelineTaskType.RAPTOR, PipelineTaskType.MINDMAP, PipelineTaskType.ARTIFACT, PipelineTaskType.SKILL]:
+        if task_type not in _PIPELINE_TASK_TYPE_TO_FINISH_FIELD:
             ok, document = DocumentService.get_by_id(referred_document_id)
             if not ok:
                 logging.warning(f"Document for referred_document_id {referred_document_id} not found")
@@ -137,7 +156,7 @@ class PipelineOperationLogService(CommonService):
         if task_type not in VALID_PIPELINE_TASK_TYPES:
             raise ValueError(f"Invalid task type: {task_type}")
 
-        if task_type in [PipelineTaskType.GRAPH_RAG, PipelineTaskType.RAPTOR, PipelineTaskType.MINDMAP, PipelineTaskType.ARTIFACT, PipelineTaskType.SKILL]:
+        if task_type in _PIPELINE_TASK_TYPE_TO_FINISH_FIELD:
             # query task to get progress information from task
             ok, task = TaskService.get_by_id(task_id)
             if not ok:
@@ -151,31 +170,10 @@ class PipelineOperationLogService(CommonService):
             process_duration = task.process_duration
 
             finish_at = process_begin_at + timedelta(seconds=process_duration)
-            if task_type == PipelineTaskType.GRAPH_RAG:
-                KnowledgebaseService.update_by_id(
-                    document.kb_id,
-                    {"graphrag_task_finish_at": finish_at},
-                )
-            elif task_type == PipelineTaskType.RAPTOR:
-                KnowledgebaseService.update_by_id(
-                    document.kb_id,
-                    {"raptor_task_finish_at": finish_at},
-                )
-            elif task_type == PipelineTaskType.MINDMAP:
-                KnowledgebaseService.update_by_id(
-                    document.kb_id,
-                    {"mindmap_task_finish_at": finish_at},
-                )
-            elif task_type == PipelineTaskType.ARTIFACT:
-                KnowledgebaseService.update_by_id(
-                    document.kb_id,
-                    {"artifact_task_finish_at": finish_at},
-                )
-            elif task_type == PipelineTaskType.SKILL:
-                KnowledgebaseService.update_by_id(
-                    document.kb_id,
-                    {"skill_task_finish_at": finish_at},
-                )
+            KnowledgebaseService.update_by_id(
+                document.kb_id,
+                {_PIPELINE_TASK_TYPE_TO_FINISH_FIELD[task_type]: finish_at},
+            )
 
         log = dict(
             id=get_uuid(),