lightrag-cn-git/tests/pipeline/test_doc_status_chunk_preservation.py

import asyncio
from datetime import datetime, timezone
from types import MethodType
from uuid import uuid4

import numpy as np
import pytest

import lightrag.lightrag as lightrag_module
import lightrag.pipeline as pipeline_module
from lightrag.base import DocStatus
from lightrag.constants import GRAPH_FIELD_SEP
from lightrag.kg.shared_storage import get_namespace_data, get_namespace_lock
from lightrag.lightrag import LightRAG
from lightrag.utils import (
    EmbeddingFunc,
    Tokenizer,
    compute_mdhash_id,
    make_relation_chunk_key,
)

pytestmark = pytest.mark.offline


class _SimpleTokenizerImpl:
    def encode(self, content: str) -> list[int]:
        return [ord(ch) for ch in content]

    def decode(self, tokens: list[int]) -> str:
        return "".join(chr(t) for t in tokens)


async def _dummy_embedding(texts: list[str]) -> np.ndarray:
    return np.ones((len(texts), 8), dtype=float)


async def _dummy_llm(*args, **kwargs) -> str:
    return "ok"


def _deterministic_chunking(
    tokenizer,
    content: str,
    split_by_character,
    split_by_character_only: bool,
    chunk_overlap_token_size: int,
    chunk_token_size: int,
) -> list[dict]:
    return [
        {"tokens": 1, "content": f"{content}::chunk1", "chunk_order_index": 0},
        {"tokens": 1, "content": f"{content}::chunk2", "chunk_order_index": 1},
    ]


def _failing_chunking(
    tokenizer,
    content: str,
    split_by_character,
    split_by_character_only: bool,
    chunk_overlap_token_size: int,
    chunk_token_size: int,
) -> list[dict]:
    raise RuntimeError("chunking fail sentinel")


def _status_to_text(status: object) -> str:
    if isinstance(status, DocStatus):
        return status.value
    return str(status).replace("DocStatus.", "").lower()


async def _build_rag(
    tmp_path,
    test_name: str,
    chunking_func,
    *,
    max_parallel_insert: int = 1,
) -> LightRAG:
    workspace = f"{test_name}_{uuid4().hex[:8]}"
    rag = LightRAG(
        working_dir=str(tmp_path / test_name),
        workspace=workspace,
        llm_model_func=_dummy_llm,
        embedding_func=EmbeddingFunc(
            embedding_dim=8,
            max_token_size=8192,
            func=_dummy_embedding,
        ),
        tokenizer=Tokenizer("test-tokenizer", _SimpleTokenizerImpl()),
        chunking_func=chunking_func,
        max_parallel_insert=max_parallel_insert,
    )
    await rag.initialize_storages()
    return rag


async def _seed_chunk_cache_entries(
    rag: LightRAG, chunk_ids: list[str], prefix: str
) -> list[str]:
    updates = {}
    cache_records = {}
    cache_ids: list[str] = []

    for idx, chunk_id in enumerate(chunk_ids):
        chunk_data = await rag.text_chunks.get_by_id(chunk_id)
        assert chunk_data is not None
        cache_id = f"{prefix}-cache-{idx}"
        chunk_data["llm_cache_list"] = [cache_id]
        updates[chunk_id] = chunk_data
        cache_records[cache_id] = {"cache_type": "extract", "return": f"cached-{idx}"}
        cache_ids.append(cache_id)

    await rag.text_chunks.upsert(updates)
    await rag.llm_response_cache.upsert(cache_records)
    return cache_ids


async def _seed_delete_retry_state(
    rag: LightRAG,
    *,
    doc_id: str,
    status_chunk_ids: list[str],
    tracking_chunk_ids: list[str],
    chunk_owners: dict[str, str],
    metadata: dict | None = None,
) -> dict[str, str]:
    entity_a = "ENTITY-A"
    entity_b = "ENTITY-B"
    relation_key = make_relation_chunk_key(entity_a, entity_b)
    source_id = GRAPH_FIELD_SEP.join(tracking_chunk_ids)
    now = datetime.now(timezone.utc).isoformat()

    await rag.full_docs.upsert(
        {doc_id: {"content": "delete retry state doc", "file_path": "delete_retry.txt"}}
    )
    await rag.doc_status.upsert(
        {
            doc_id: {
                "status": DocStatus.PROCESSED,
                "content_summary": "delete retry state",
                "content_length": 22,
                "chunks_count": len(status_chunk_ids),
                "chunks_list": status_chunk_ids,
                "created_at": now,
                "updated_at": now,
                "file_path": "delete_retry.txt",
                "track_id": f"track-{doc_id}",
                "error_msg": "",
                "metadata": metadata or {},
            }
        }
    )

    chunk_payload = {}
    for chunk_id, owner_doc_id in chunk_owners.items():
        chunk_payload[chunk_id] = {
            "content": f"{chunk_id} content",
            "file_path": f"{chunk_id}.txt",
            "full_doc_id": owner_doc_id,
        }

    if chunk_payload:
        await rag.text_chunks.upsert(chunk_payload)
        await rag.chunks_vdb.upsert(chunk_payload)

    await rag.full_entities.upsert({doc_id: {"entity_names": [entity_a, entity_b]}})
    await rag.full_relations.upsert(
        {doc_id: {"relation_pairs": [(entity_a, entity_b)]}}
    )
    await rag.entity_chunks.upsert(
        {
            entity_a: {
                "chunk_ids": tracking_chunk_ids,
                "count": len(tracking_chunk_ids),
            },
            entity_b: {
                "chunk_ids": tracking_chunk_ids,
                "count": len(tracking_chunk_ids),
            },
        }
    )
    await rag.relation_chunks.upsert(
        {
            relation_key: {
                "chunk_ids": tracking_chunk_ids,
                "count": len(tracking_chunk_ids),
            }
        }
    )

    created_at = int(datetime.now(timezone.utc).timestamp())
    for entity_name in [entity_a, entity_b]:
        await rag.chunk_entity_relation_graph.upsert_node(
            entity_name,
            {
                "entity_id": entity_name,
                "source_id": source_id,
                "description": f"{entity_name} description",
                "entity_type": "test",
                "file_path": "delete_retry.txt",
                "created_at": created_at,
                "truncate": "",
            },
        )

    await rag.chunk_entity_relation_graph.upsert_edge(
        entity_a,
        entity_b,
        {
            "source": entity_a,
            "target": entity_b,
            "source_id": source_id,
            "description": "related",
            "keywords": "test",
            "weight": 1.0,
            "file_path": "delete_retry.txt",
        },
    )

    await rag.entities_vdb.upsert(
        {
            compute_mdhash_id(entity_a, prefix="ent-"): {
                "content": f"{entity_a}\n{entity_a} description",
                "entity_name": entity_a,
                "source_id": source_id,
                "description": f"{entity_a} description",
                "entity_type": "test",
                "file_path": "delete_retry.txt",
            },
            compute_mdhash_id(entity_b, prefix="ent-"): {
                "content": f"{entity_b}\n{entity_b} description",
                "entity_name": entity_b,
                "source_id": source_id,
                "description": f"{entity_b} description",
                "entity_type": "test",
                "file_path": "delete_retry.txt",
            },
        }
    )
    await rag.relationships_vdb.upsert(
        {
            compute_mdhash_id(entity_a + entity_b, prefix="rel-"): {
                "content": f"test\t{entity_a}\n{entity_b}\nrelated",
                "src_id": entity_a,
                "tgt_id": entity_b,
                "source_id": source_id,
                "description": "related",
                "keywords": "test",
                "weight": 1.0,
                "file_path": "delete_retry.txt",
            }
        }
    )

    return {
        "entity_a": entity_a,
        "entity_b": entity_b,
        "relation_key": relation_key,
    }


async def _succeed_rebuild_from_remaining_chunks(
    entities_to_rebuild,
    relationships_to_rebuild,
    knowledge_graph_inst,
    entities_vdb,
    relationships_vdb,
    **kwargs,
):
    for entity_name, remaining_chunk_ids in entities_to_rebuild.items():
        node = await knowledge_graph_inst.get_node(entity_name)
        assert node is not None
        updated_node = {
            **node,
            "source_id": GRAPH_FIELD_SEP.join(remaining_chunk_ids),
        }
        await knowledge_graph_inst.upsert_node(entity_name, updated_node)
        await entities_vdb.upsert(
            {
                compute_mdhash_id(entity_name, prefix="ent-"): {
                    "content": f"{entity_name}\n{updated_node['description']}",
                    "entity_name": entity_name,
                    "source_id": updated_node["source_id"],
                    "description": updated_node["description"],
                    "entity_type": updated_node["entity_type"],
                    "file_path": updated_node["file_path"],
                }
            }
        )

    for (src, tgt), remaining_chunk_ids in relationships_to_rebuild.items():
        edge = await knowledge_graph_inst.get_edge(src, tgt)
        assert edge is not None
        updated_edge = {
            **edge,
            "source_id": GRAPH_FIELD_SEP.join(remaining_chunk_ids),
        }
        await knowledge_graph_inst.upsert_edge(src, tgt, updated_edge)
        await relationships_vdb.upsert(
            {
                compute_mdhash_id(src + tgt, prefix="rel-"): {
                    "content": f"{updated_edge['keywords']}\t{src}\n{tgt}\n{updated_edge['description']}",
                    "src_id": src,
                    "tgt_id": tgt,
                    "source_id": updated_edge["source_id"],
                    "description": updated_edge["description"],
                    "keywords": updated_edge["keywords"],
                    "weight": updated_edge["weight"],
                    "file_path": updated_edge["file_path"],
                }
            }
        )


@pytest.mark.asyncio
async def test_extract_failure_preserves_chunks_and_allows_delete_with_cache_cleanup(
    tmp_path, monkeypatch
):
    rag = await _build_rag(tmp_path, "extract_failure_cleanup", _deterministic_chunking)
    try:
        content = "extract failure document"
        file_path = "extract_failure.txt"
        doc_id = compute_mdhash_id(file_path, prefix="doc-")
        await rag.apipeline_enqueue_documents(input=content, file_paths=file_path)

        async def fail_extract(self, chunks, pipeline_status, pipeline_status_lock):
            raise RuntimeError("extract fail sentinel")

        rag._process_extract_entities = MethodType(fail_extract, rag)

        await rag.apipeline_process_enqueue_documents()

        doc_status = await rag.doc_status.get_by_id(doc_id)
        assert doc_status is not None
        assert _status_to_text(doc_status["status"]) == "failed"
        chunk_ids = doc_status.get("chunks_list", [])
        assert len(chunk_ids) == 2
        assert doc_status.get("chunks_count") == 2

        cache_ids = await _seed_chunk_cache_entries(rag, chunk_ids, "extract")

        result = await rag.adelete_by_doc_id(doc_id, delete_llm_cache=True)
        assert result.status == "success"

        deleted_chunks = await rag.text_chunks.get_by_ids(chunk_ids)
        assert all(item is None for item in deleted_chunks)
        deleted_cache = [
            await rag.llm_response_cache.get_by_id(cid) for cid in cache_ids
        ]
        assert all(item is None for item in deleted_cache)
        assert await rag.doc_status.get_by_id(doc_id) is None
    finally:
        await rag.finalize_storages()


@pytest.mark.asyncio
async def test_extract_failure_before_chunking_clears_stale_chunk_snapshot(
    tmp_path,
):
    """The resume branch of ``apipeline_process_enqueue_documents`` purges
    any stale ``chunks_list`` from a previous interrupted run *before*
    chunking starts (so the new run does not mix old and new chunks).
    Therefore, when chunking subsequently fails on the retry, the failed
    doc_status reflects the post-purge state — the previous snapshot is
    intentionally not preserved any more.

    Earlier this test asserted the opposite ("preserve previous snapshot
    across failure"), which conflicted with the documented resume rule
    that "已抽取文档一律删掉所有的文本块，重新走多模态分析和实体关系提取".
    """
    rag = await _build_rag(tmp_path, "extract_failure_pre_chunking", _failing_chunking)
    try:
        content = "chunking failure document"
        file_path = "chunking_failure.txt"
        doc_id = compute_mdhash_id(file_path, prefix="doc-")
        await rag.apipeline_enqueue_documents(input=content, file_paths=file_path)

        previous_chunks = ["chunk-old-1", "chunk-old-2", "chunk-old-3"]
        existing = await rag.doc_status.get_by_id(doc_id)
        assert existing is not None
        await rag.doc_status.upsert(
            {
                doc_id: {
                    "status": DocStatus.FAILED,
                    "content_summary": existing["content_summary"],
                    "content_length": existing["content_length"],
                    "chunks_count": len(previous_chunks),
                    "chunks_list": previous_chunks,
                    "created_at": existing["created_at"],
                    "updated_at": datetime.now(timezone.utc).isoformat(),
                    "file_path": existing["file_path"],
                    "track_id": existing["track_id"],
                    "error_msg": "previous failure",
                    "metadata": {"source": "test"},
                }
            }
        )

        await rag.apipeline_process_enqueue_documents()

        failed_status = await rag.doc_status.get_by_id(doc_id)
        assert failed_status is not None
        assert _status_to_text(failed_status["status"]) == "failed"
        # Resume purged the stale list before chunking; the failure record
        # therefore shows zero chunks rather than the previous snapshot.
        assert failed_status.get("chunks_list") == []
        assert failed_status.get("chunks_count") == 0
        assert "chunking fail sentinel" in (failed_status.get("error_msg") or "")
    finally:
        await rag.finalize_storages()


@pytest.mark.asyncio
async def test_merge_failure_preserves_chunks_and_skip_cache_cleanup_when_disabled(
    tmp_path, monkeypatch
):
    rag = await _build_rag(
        tmp_path, "merge_failure_keep_cache", _deterministic_chunking
    )
    try:
        content = "merge failure document"
        file_path = "merge_failure.txt"
        doc_id = compute_mdhash_id(file_path, prefix="doc-")
        await rag.apipeline_enqueue_documents(input=content, file_paths=file_path)

        async def ok_extract(self, chunks, pipeline_status, pipeline_status_lock):
            return {"chunk_count": len(chunks)}

        async def fail_merge(**kwargs):
            raise RuntimeError("merge fail sentinel")

        rag._process_extract_entities = MethodType(ok_extract, rag)
        monkeypatch.setattr(pipeline_module, "merge_nodes_and_edges", fail_merge)

        await rag.apipeline_process_enqueue_documents()

        doc_status = await rag.doc_status.get_by_id(doc_id)
        assert doc_status is not None
        assert _status_to_text(doc_status["status"]) == "failed"
        chunk_ids = doc_status.get("chunks_list", [])
        assert len(chunk_ids) == 2
        assert doc_status.get("chunks_count") == 2

        cache_ids = await _seed_chunk_cache_entries(rag, chunk_ids, "merge")
        result = await rag.adelete_by_doc_id(doc_id, delete_llm_cache=False)
        assert result.status == "success"

        remaining_cache = [
            await rag.llm_response_cache.get_by_id(cid) for cid in cache_ids
        ]
        assert all(item is not None for item in remaining_cache)
    finally:
        await rag.finalize_storages()


@pytest.mark.asyncio
async def test_delete_rebuild_failure_prunes_chunk_tracking_before_abort(
    tmp_path, monkeypatch
):
    rag = await _build_rag(
        tmp_path, "delete_rebuild_failure_chunk_tracking", _deterministic_chunking
    )
    try:
        doc_id = "doc-delete-rebuild-failure"
        keep_chunk_id = "chunk-keep"
        drop_chunk_id = "chunk-drop"
        seeded = await _seed_delete_retry_state(
            rag,
            doc_id=doc_id,
            status_chunk_ids=[drop_chunk_id],
            tracking_chunk_ids=[keep_chunk_id, drop_chunk_id],
            chunk_owners={
                keep_chunk_id: "doc-keep",
                drop_chunk_id: doc_id,
            },
        )
        entity_a = seeded["entity_a"]
        entity_b = seeded["entity_b"]
        relation_key = seeded["relation_key"]

        async def fail_rebuild(**kwargs):
            raise RuntimeError("rebuild fail sentinel")

        monkeypatch.setattr(
            lightrag_module, "rebuild_knowledge_from_chunks", fail_rebuild
        )

        result = await rag.adelete_by_doc_id(doc_id)

        entity_tracking = await rag.entity_chunks.get_by_id(entity_a)
        relation_tracking = await rag.relation_chunks.get_by_id(relation_key)
        failed_status = await rag.doc_status.get_by_id(doc_id)

        assert result.status == "fail"
        assert "rebuild fail sentinel" in result.message
        assert await rag.text_chunks.get_by_id(drop_chunk_id) is None
        assert await rag.text_chunks.get_by_id(keep_chunk_id) is not None
        assert failed_status is not None
        assert failed_status["chunks_list"] == [drop_chunk_id]
        assert failed_status["metadata"]["deletion_failed"] is True
        assert (
            failed_status["metadata"]["deletion_failure_stage"]
            == "rebuild_knowledge_graph"
        )
        assert "rebuild fail sentinel" in failed_status["error_msg"]
        assert entity_tracking is not None
        assert entity_tracking["chunk_ids"] == [keep_chunk_id]
        assert entity_tracking["count"] == 1
        assert relation_tracking is not None
        assert relation_tracking["chunk_ids"] == [keep_chunk_id]
        assert relation_tracking["count"] == 1
        assert (
            await rag.chunk_entity_relation_graph.get_edge(entity_a, entity_b)
            is not None
        )
    finally:
        await rag.finalize_storages()


@pytest.mark.asyncio
async def test_delete_retry_succeeds_after_rebuild_failure(tmp_path, monkeypatch):
    rag = await _build_rag(
        tmp_path, "delete_retry_after_failure", _deterministic_chunking
    )
    try:
        doc_id = "doc-delete-retry-success"
        keep_chunk_id = "chunk-keep"
        drop_chunk_id = "chunk-drop"
        seeded = await _seed_delete_retry_state(
            rag,
            doc_id=doc_id,
            status_chunk_ids=[drop_chunk_id],
            tracking_chunk_ids=[keep_chunk_id, drop_chunk_id],
            chunk_owners={
                keep_chunk_id: "doc-keep",
                drop_chunk_id: doc_id,
            },
        )
        entity_a = seeded["entity_a"]
        entity_b = seeded["entity_b"]
        relation_key = seeded["relation_key"]

        async def fail_rebuild(**kwargs):
            raise RuntimeError("rebuild fail sentinel")

        monkeypatch.setattr(
            lightrag_module, "rebuild_knowledge_from_chunks", fail_rebuild
        )
        first_result = await rag.adelete_by_doc_id(doc_id)
        assert first_result.status == "fail"

        monkeypatch.setattr(
            lightrag_module,
            "rebuild_knowledge_from_chunks",
            _succeed_rebuild_from_remaining_chunks,
        )
        second_result = await rag.adelete_by_doc_id(doc_id)

        assert second_result.status == "success"
        assert await rag.doc_status.get_by_id(doc_id) is None
        assert await rag.full_docs.get_by_id(doc_id) is None
        assert await rag.full_entities.get_by_id(doc_id) is None
        assert await rag.full_relations.get_by_id(doc_id) is None
        assert await rag.text_chunks.get_by_id(drop_chunk_id) is None
        assert await rag.text_chunks.get_by_id(keep_chunk_id) is not None

        entity_a_tracking = await rag.entity_chunks.get_by_id(entity_a)
        entity_b_tracking = await rag.entity_chunks.get_by_id(entity_b)
        relation_tracking = await rag.relation_chunks.get_by_id(relation_key)
        edge = await rag.chunk_entity_relation_graph.get_edge(entity_a, entity_b)

        assert entity_a_tracking is not None
        assert entity_a_tracking["chunk_ids"] == [keep_chunk_id]
        assert entity_b_tracking is not None
        assert entity_b_tracking["chunk_ids"] == [keep_chunk_id]
        assert relation_tracking is not None
        assert relation_tracking["chunk_ids"] == [keep_chunk_id]
        assert edge is not None
        assert edge["source_id"] == keep_chunk_id
    finally:
        await rag.finalize_storages()


@pytest.mark.asyncio
async def test_delete_retry_cleans_llm_cache_after_rebuild_failure(
    tmp_path, monkeypatch
):
    rag = await _build_rag(
        tmp_path, "delete_retry_cleans_llm_cache", _deterministic_chunking
    )
    try:
        doc_id = "doc-delete-retry-cache-cleanup"
        keep_chunk_id = "chunk-keep"
        drop_chunk_id = "chunk-drop"
        seeded = await _seed_delete_retry_state(
            rag,
            doc_id=doc_id,
            status_chunk_ids=[drop_chunk_id],
            tracking_chunk_ids=[keep_chunk_id, drop_chunk_id],
            chunk_owners={
                keep_chunk_id: "doc-keep",
                drop_chunk_id: doc_id,
            },
        )
        entity_a = seeded["entity_a"]
        entity_b = seeded["entity_b"]
        cache_ids = await _seed_chunk_cache_entries(rag, [drop_chunk_id], "retry")

        async def fail_rebuild(**kwargs):
            raise RuntimeError("rebuild fail sentinel")

        monkeypatch.setattr(
            lightrag_module, "rebuild_knowledge_from_chunks", fail_rebuild
        )
        first_result = await rag.adelete_by_doc_id(doc_id, delete_llm_cache=True)
        assert first_result.status == "fail"

        failed_status = await rag.doc_status.get_by_id(doc_id)
        assert failed_status is not None
        assert failed_status["metadata"]["deletion_llm_cache_ids"] == cache_ids
        assert await rag.text_chunks.get_by_id(drop_chunk_id) is None

        monkeypatch.setattr(
            lightrag_module,
            "rebuild_knowledge_from_chunks",
            _succeed_rebuild_from_remaining_chunks,
        )
        second_result = await rag.adelete_by_doc_id(doc_id, delete_llm_cache=True)

        assert second_result.status == "success"
        assert await rag.doc_status.get_by_id(doc_id) is None
        assert await rag.full_docs.get_by_id(doc_id) is None
        assert await rag.llm_response_cache.get_by_id(cache_ids[0]) is None
        edge = await rag.chunk_entity_relation_graph.get_edge(entity_a, entity_b)
        assert edge is not None
        assert edge["source_id"] == keep_chunk_id
    finally:
        await rag.finalize_storages()


@pytest.mark.asyncio
async def test_delete_retry_cleans_llm_cache_when_enabled_on_retry(
    tmp_path, monkeypatch
):
    rag = await _build_rag(
        tmp_path, "delete_retry_cache_cleanup_flag_change", _deterministic_chunking
    )
    try:
        doc_id = "doc-delete-retry-cache-flag-change"
        keep_chunk_id = "chunk-keep"
        drop_chunk_id = "chunk-drop"
        await _seed_delete_retry_state(
            rag,
            doc_id=doc_id,
            status_chunk_ids=[drop_chunk_id],
            tracking_chunk_ids=[keep_chunk_id, drop_chunk_id],
            chunk_owners={
                keep_chunk_id: "doc-keep",
                drop_chunk_id: doc_id,
            },
        )
        cache_ids = await _seed_chunk_cache_entries(rag, [drop_chunk_id], "retry-flag")

        async def fail_rebuild(**kwargs):
            raise RuntimeError("rebuild fail sentinel")

        monkeypatch.setattr(
            lightrag_module, "rebuild_knowledge_from_chunks", fail_rebuild
        )
        first_result = await rag.adelete_by_doc_id(doc_id, delete_llm_cache=False)

        assert first_result.status == "fail"
        failed_status = await rag.doc_status.get_by_id(doc_id)
        assert failed_status is not None
        assert failed_status["metadata"]["deletion_llm_cache_ids"] == cache_ids
        assert await rag.text_chunks.get_by_id(drop_chunk_id) is None

        monkeypatch.setattr(
            lightrag_module,
            "rebuild_knowledge_from_chunks",
            _succeed_rebuild_from_remaining_chunks,
        )
        second_result = await rag.adelete_by_doc_id(doc_id, delete_llm_cache=True)

        assert second_result.status == "success"
        assert await rag.doc_status.get_by_id(doc_id) is None
        assert await rag.full_docs.get_by_id(doc_id) is None
        assert await rag.llm_response_cache.get_by_id(cache_ids[0]) is None
    finally:
        await rag.finalize_storages()


@pytest.mark.asyncio
async def test_delete_retry_collects_cache_ids_without_cache_storage(
    tmp_path, monkeypatch
):
    rag = await _build_rag(
        tmp_path,
        "delete_retry_collect_cache_ids_without_storage",
        _deterministic_chunking,
    )
    try:
        doc_id = "doc-delete-retry-collect-cache-ids"
        keep_chunk_id = "chunk-keep"
        drop_chunk_id = "chunk-drop"
        await _seed_delete_retry_state(
            rag,
            doc_id=doc_id,
            status_chunk_ids=[drop_chunk_id],
            tracking_chunk_ids=[keep_chunk_id, drop_chunk_id],
            chunk_owners={
                keep_chunk_id: "doc-keep",
                drop_chunk_id: doc_id,
            },
        )
        cache_ids = await _seed_chunk_cache_entries(
            rag, [drop_chunk_id], "collect-without-storage"
        )

        async def fail_rebuild(**kwargs):
            raise RuntimeError("rebuild fail sentinel")

        cache_storage = rag.llm_response_cache
        rag.llm_response_cache = None
        monkeypatch.setattr(
            lightrag_module, "rebuild_knowledge_from_chunks", fail_rebuild
        )
        first_result = await rag.adelete_by_doc_id(doc_id, delete_llm_cache=False)

        assert first_result.status == "fail"
        failed_status = await rag.doc_status.get_by_id(doc_id)
        assert failed_status is not None
        assert failed_status["metadata"]["deletion_llm_cache_ids"] == cache_ids
        assert await rag.text_chunks.get_by_id(drop_chunk_id) is None

        rag.llm_response_cache = cache_storage
        monkeypatch.setattr(
            lightrag_module,
            "rebuild_knowledge_from_chunks",
            _succeed_rebuild_from_remaining_chunks,
        )
        second_result = await rag.adelete_by_doc_id(doc_id, delete_llm_cache=True)

        assert second_result.status == "success"
        assert await rag.doc_status.get_by_id(doc_id) is None
        assert await rag.full_docs.get_by_id(doc_id) is None
        assert await rag.llm_response_cache.get_by_id(cache_ids[0]) is None
    finally:
        await rag.finalize_storages()


@pytest.mark.asyncio
async def test_delete_retry_succeeds_after_llm_cache_cleanup_failure(
    tmp_path, monkeypatch
):
    rag = await _build_rag(
        tmp_path, "delete_retry_after_cache_cleanup_failure", _deterministic_chunking
    )
    try:
        doc_id = "doc-delete-cache-cleanup-failure"
        drop_chunk_id = "chunk-drop"
        await _seed_delete_retry_state(
            rag,
            doc_id=doc_id,
            status_chunk_ids=[drop_chunk_id],
            tracking_chunk_ids=[drop_chunk_id],
            chunk_owners={drop_chunk_id: doc_id},
        )
        cache_ids = await _seed_chunk_cache_entries(
            rag, [drop_chunk_id], "cache-cleanup-failure"
        )

        async def fail_cache_delete(self, ids):
            raise RuntimeError("llm cache delete fail sentinel")

        monkeypatch.setattr(
            rag.llm_response_cache,
            "delete",
            MethodType(fail_cache_delete, rag.llm_response_cache),
        )
        first_result = await rag.adelete_by_doc_id(doc_id, delete_llm_cache=True)

        assert first_result.status == "fail"
        assert "Failed to delete LLM cache" in first_result.message
        assert await rag.doc_status.get_by_id(doc_id) is not None
        assert await rag.full_docs.get_by_id(doc_id) is not None
        assert await rag.llm_response_cache.get_by_id(cache_ids[0]) is not None

        monkeypatch.undo()
        second_result = await rag.adelete_by_doc_id(doc_id, delete_llm_cache=True)

        assert second_result.status == "success"
        assert await rag.doc_status.get_by_id(doc_id) is None
        assert await rag.full_docs.get_by_id(doc_id) is None
        assert await rag.llm_response_cache.get_by_id(cache_ids[0]) is None
    finally:
        await rag.finalize_storages()


@pytest.mark.asyncio
async def test_delete_retry_succeeds_after_silent_llm_cache_cleanup_failure(
    tmp_path, monkeypatch
):
    rag = await _build_rag(
        tmp_path,
        "delete_retry_after_silent_cache_cleanup_failure",
        _deterministic_chunking,
    )
    try:
        doc_id = "doc-delete-silent-cache-cleanup-failure"
        drop_chunk_id = "chunk-drop"
        await _seed_delete_retry_state(
            rag,
            doc_id=doc_id,
            status_chunk_ids=[drop_chunk_id],
            tracking_chunk_ids=[drop_chunk_id],
            chunk_owners={drop_chunk_id: doc_id},
        )
        cache_ids = await _seed_chunk_cache_entries(
            rag, [drop_chunk_id], "silent-cache-cleanup-failure"
        )

        async def silently_fail_cache_delete(self, ids):
            return None

        monkeypatch.setattr(
            rag.llm_response_cache,
            "delete",
            MethodType(silently_fail_cache_delete, rag.llm_response_cache),
        )
        first_result = await rag.adelete_by_doc_id(doc_id, delete_llm_cache=True)

        assert first_result.status == "fail"
        assert "still exist after delete" in first_result.message
        assert await rag.doc_status.get_by_id(doc_id) is not None
        assert await rag.full_docs.get_by_id(doc_id) is not None
        assert await rag.llm_response_cache.get_by_id(cache_ids[0]) is not None

        monkeypatch.undo()
        second_result = await rag.adelete_by_doc_id(doc_id, delete_llm_cache=True)

        assert second_result.status == "success"
        assert await rag.doc_status.get_by_id(doc_id) is None
        assert await rag.full_docs.get_by_id(doc_id) is None
        assert await rag.llm_response_cache.get_by_id(cache_ids[0]) is None
    finally:
        await rag.finalize_storages()


@pytest.mark.asyncio
async def test_delete_retry_preserves_cache_cleanup_state_when_cache_storage_unavailable(
    tmp_path,
):
    rag = await _build_rag(
        tmp_path, "delete_retry_cache_storage_unavailable", _deterministic_chunking
    )
    try:
        doc_id = "doc-delete-cache-storage-unavailable"
        drop_chunk_id = "chunk-drop"
        await _seed_delete_retry_state(
            rag,
            doc_id=doc_id,
            status_chunk_ids=[drop_chunk_id],
            tracking_chunk_ids=[drop_chunk_id],
            chunk_owners={drop_chunk_id: doc_id},
        )
        cache_ids = await _seed_chunk_cache_entries(
            rag, [drop_chunk_id], "cache-storage-unavailable"
        )

        status_doc = await rag.doc_status.get_by_id(doc_id)
        assert status_doc is not None
        status_doc["metadata"] = {"deletion_llm_cache_ids": cache_ids}
        await rag.doc_status.upsert({doc_id: status_doc})

        cache_storage = rag.llm_response_cache
        rag.llm_response_cache = None

        first_result = await rag.adelete_by_doc_id(doc_id, delete_llm_cache=True)

        assert first_result.status == "fail"
        assert "cache storage is unavailable" in first_result.message
        failed_status = await rag.doc_status.get_by_id(doc_id)
        assert failed_status is not None
        assert failed_status["metadata"]["deletion_llm_cache_ids"] == cache_ids
        assert await rag.full_docs.get_by_id(doc_id) is not None

        rag.llm_response_cache = cache_storage
        second_result = await rag.adelete_by_doc_id(doc_id, delete_llm_cache=True)

        assert second_result.status == "success"
        assert await rag.doc_status.get_by_id(doc_id) is None
        assert await rag.full_docs.get_by_id(doc_id) is None
        assert await rag.llm_response_cache.get_by_id(cache_ids[0]) is None
    finally:
        await rag.finalize_storages()


@pytest.mark.asyncio
async def test_delete_succeeds_when_chunks_list_missing(tmp_path):
    rag = await _build_rag(
        tmp_path, "delete_missing_chunks_list_rejected", _deterministic_chunking
    )
    try:
        doc_id = "doc-delete-missing-chunks-list"
        drop_chunk_id = "chunk-drop-only"
        seeded = await _seed_delete_retry_state(
            rag,
            doc_id=doc_id,
            status_chunk_ids=[],
            tracking_chunk_ids=[drop_chunk_id],
            chunk_owners={drop_chunk_id: doc_id},
        )
        entity_a = seeded["entity_a"]
        entity_b = seeded["entity_b"]
        relation_key = seeded["relation_key"]

        result = await rag.adelete_by_doc_id(doc_id)

        assert result.status == "success"
        assert "without associated chunks" in result.message
        assert await rag.doc_status.get_by_id(doc_id) is None
        assert await rag.full_docs.get_by_id(doc_id) is None
        assert await rag.full_entities.get_by_id(doc_id) is not None
        assert await rag.full_relations.get_by_id(doc_id) is not None
        assert await rag.text_chunks.get_by_id(drop_chunk_id) is not None
        assert await rag.chunks_vdb.get_by_id(drop_chunk_id) is not None
        assert await rag.chunk_entity_relation_graph.get_node(entity_a) is not None
        assert await rag.chunk_entity_relation_graph.get_node(entity_b) is not None
        assert (
            await rag.chunk_entity_relation_graph.get_edge(entity_a, entity_b)
            is not None
        )
        assert await rag.entity_chunks.get_by_id(entity_a) is not None
        assert await rag.entity_chunks.get_by_id(entity_b) is not None
        assert await rag.relation_chunks.get_by_id(relation_key) is not None
        assert (
            await rag.entities_vdb.get_by_id(compute_mdhash_id(entity_a, prefix="ent-"))
            is not None
        )
        assert (
            await rag.entities_vdb.get_by_id(compute_mdhash_id(entity_b, prefix="ent-"))
            is not None
        )
        assert (
            await rag.relationships_vdb.get_by_id(
                compute_mdhash_id(entity_a + entity_b, prefix="rel-")
            )
            is not None
        )
    finally:
        await rag.finalize_storages()


@pytest.mark.asyncio
async def test_delete_ignores_stale_graph_source_ids_when_tracking_exists(tmp_path):
    rag = await _build_rag(
        tmp_path, "delete_ignore_stale_graph_sources", _deterministic_chunking
    )
    try:
        doc_id = "doc-delete-stale-graph-sources"
        drop_chunk_id = "chunk-drop"
        stale_chunk_id = "chunk-stale"
        seeded = await _seed_delete_retry_state(
            rag,
            doc_id=doc_id,
            status_chunk_ids=[drop_chunk_id],
            tracking_chunk_ids=[drop_chunk_id],
            chunk_owners={drop_chunk_id: doc_id},
        )
        entity_a = seeded["entity_a"]
        entity_b = seeded["entity_b"]
        relation_key = seeded["relation_key"]
        stale_source_id = GRAPH_FIELD_SEP.join([stale_chunk_id, drop_chunk_id])

        await rag.chunk_entity_relation_graph.upsert_node(
            entity_a,
            {
                "entity_id": entity_a,
                "source_id": stale_source_id,
                "description": f"{entity_a} description",
                "entity_type": "test",
                "file_path": "delete_retry.txt",
                "created_at": 1,
                "truncate": "",
            },
        )
        await rag.chunk_entity_relation_graph.upsert_node(
            entity_b,
            {
                "entity_id": entity_b,
                "source_id": stale_source_id,
                "description": f"{entity_b} description",
                "entity_type": "test",
                "file_path": "delete_retry.txt",
                "created_at": 1,
                "truncate": "",
            },
        )
        await rag.chunk_entity_relation_graph.upsert_edge(
            entity_a,
            entity_b,
            {
                "source": entity_a,
                "target": entity_b,
                "source_id": stale_source_id,
                "description": "related",
                "keywords": "test",
                "weight": 1.0,
                "file_path": "delete_retry.txt",
            },
        )

        result = await rag.adelete_by_doc_id(doc_id)

        assert result.status == "success"
        assert await rag.doc_status.get_by_id(doc_id) is None
        assert await rag.full_docs.get_by_id(doc_id) is None
        assert await rag.text_chunks.get_by_id(drop_chunk_id) is None
        assert await rag.chunk_entity_relation_graph.get_node(entity_a) is None
        assert await rag.chunk_entity_relation_graph.get_node(entity_b) is None
        assert (
            await rag.chunk_entity_relation_graph.get_edge(entity_a, entity_b) is None
        )
        assert await rag.entity_chunks.get_by_id(entity_a) is None
        assert await rag.entity_chunks.get_by_id(entity_b) is None
        assert await rag.relation_chunks.get_by_id(relation_key) is None
    finally:
        await rag.finalize_storages()


@pytest.mark.asyncio
async def test_validate_and_fix_consistency_preserves_chunks_on_reset(tmp_path):
    rag = await _build_rag(tmp_path, "reset_preserve_chunks", _deterministic_chunking)
    try:
        failed_doc_id = "doc-failed-reset"
        processing_doc_id = "doc-processing-reset"
        inferred_count_doc_id = "doc-inferred-count-reset"

        now = datetime.now(timezone.utc).isoformat()
        await rag.full_docs.upsert(
            {
                failed_doc_id: {"content": "failed doc", "file_path": "failed.txt"},
                processing_doc_id: {
                    "content": "processing doc",
                    "file_path": "processing.txt",
                },
                inferred_count_doc_id: {
                    "content": "inferred count doc",
                    "file_path": "inferred.txt",
                },
            }
        )
        await rag.doc_status.upsert(
            {
                failed_doc_id: {
                    "status": DocStatus.FAILED,
                    "content_summary": "failed",
                    "content_length": 10,
                    "chunks_count": 2,
                    "chunks_list": ["f-1", "f-2"],
                    "created_at": now,
                    "updated_at": now,
                    "file_path": "failed.txt",
                    "track_id": "track-1",
                    "error_msg": "old error",
                    "metadata": {"old": True},
                },
                processing_doc_id: {
                    "status": DocStatus.PROCESSING,
                    "content_summary": "processing",
                    "content_length": 12,
                    "chunks_count": 1,
                    "chunks_list": ["p-1"],
                    "created_at": now,
                    "updated_at": now,
                    "file_path": "processing.txt",
                    "track_id": "track-2",
                    "error_msg": "old error",
                    "metadata": {"old": True},
                },
                inferred_count_doc_id: {
                    "status": DocStatus.FAILED,
                    "content_summary": "inferred",
                    "content_length": 14,
                    "chunks_list": ["i-1", "i-2", "i-3"],
                    "created_at": now,
                    "updated_at": now,
                    "file_path": "inferred.txt",
                    "track_id": "track-3",
                    "error_msg": "old error",
                    "metadata": {"old": True},
                },
            }
        )

        failed_docs = await rag.doc_status.get_docs_by_status(DocStatus.FAILED)
        processing_docs = await rag.doc_status.get_docs_by_status(DocStatus.PROCESSING)
        to_process_docs = {**failed_docs, **processing_docs}

        pipeline_status = {"latest_message": "", "history_messages": []}
        await rag._validate_and_fix_document_consistency(
            to_process_docs=to_process_docs,
            pipeline_status=pipeline_status,
            pipeline_status_lock=asyncio.Lock(),
        )

        failed_reset = await rag.doc_status.get_by_id(failed_doc_id)
        assert failed_reset is not None
        assert _status_to_text(failed_reset["status"]) == "pending"
        assert failed_reset.get("chunks_list") == ["f-1", "f-2"]
        assert failed_reset.get("chunks_count") == 2

        processing_reset = await rag.doc_status.get_by_id(processing_doc_id)
        assert processing_reset is not None
        assert _status_to_text(processing_reset["status"]) == "pending"
        assert processing_reset.get("chunks_list") == ["p-1"]
        assert processing_reset.get("chunks_count") == 1

        inferred_count_reset = await rag.doc_status.get_by_id(inferred_count_doc_id)
        assert inferred_count_reset is not None
        assert _status_to_text(inferred_count_reset["status"]) == "pending"
        assert inferred_count_reset.get("chunks_list") == ["i-1", "i-2", "i-3"]
        assert inferred_count_reset.get("chunks_count") == 3
    finally:
        await rag.finalize_storages()


@pytest.mark.asyncio
async def test_validate_and_fix_consistency_repairs_unknown_file_path_from_full_docs(
    tmp_path,
):
    rag = await _build_rag(tmp_path, "repair_reset_file_path", _deterministic_chunking)
    try:
        doc_id = "doc-repair-reset"
        now = datetime.now(timezone.utc).isoformat()
        await rag.full_docs.upsert(
            {
                doc_id: {
                    "content": "repair path doc",
                    "file_path": "repaired-from-full-docs.md",
                }
            }
        )
        await rag.doc_status.upsert(
            {
                doc_id: {
                    "status": DocStatus.FAILED,
                    "content_summary": "repair path",
                    "content_length": 15,
                    "chunks_count": 0,
                    "chunks_list": [],
                    "created_at": now,
                    "updated_at": now,
                    "file_path": "unknown_source",
                    "track_id": "track-repair",
                    "error_msg": "old error",
                    "metadata": {"old": True},
                }
            }
        )

        failed_docs = await rag.doc_status.get_docs_by_status(DocStatus.FAILED)
        pipeline_status = {"latest_message": "", "history_messages": []}
        await rag._validate_and_fix_document_consistency(
            to_process_docs=failed_docs,
            pipeline_status=pipeline_status,
            pipeline_status_lock=asyncio.Lock(),
        )

        repaired_status = await rag.doc_status.get_by_id(doc_id)
        assert repaired_status is not None
        assert _status_to_text(repaired_status["status"]) == "pending"
        assert repaired_status["file_path"] == "repaired-from-full-docs.md"
    finally:
        await rag.finalize_storages()


@pytest.mark.asyncio
async def test_pipeline_cancellation_preserves_file_path_for_queued_docs(
    tmp_path, monkeypatch
):
    rag = await _build_rag(
        tmp_path,
        "cancel_preserve_file_path",
        _deterministic_chunking,
        max_parallel_insert=1,
    )
    try:
        contents = ["first queued content", "second queued content"]
        file_paths = ["first.md", "second.md"]
        await rag.apipeline_enqueue_documents(input=contents, file_paths=file_paths)

        extraction_started = asyncio.Event()
        release_first_doc = asyncio.Event()

        async def _blocking_extract(
            self, chunks, pipeline_status, pipeline_status_lock
        ):
            extraction_started.set()
            await release_first_doc.wait()
            return []

        monkeypatch.setattr(
            rag,
            "_process_extract_entities",
            MethodType(_blocking_extract, rag),
        )

        pipeline_task = asyncio.create_task(rag.apipeline_process_enqueue_documents())
        await asyncio.wait_for(extraction_started.wait(), timeout=5)

        pipeline_status = await get_namespace_data(
            "pipeline_status", workspace=rag.workspace
        )
        pipeline_status_lock = get_namespace_lock(
            "pipeline_status", workspace=rag.workspace
        )
        async with pipeline_status_lock:
            pipeline_status["cancellation_requested"] = True

        release_first_doc.set()
        await asyncio.wait_for(pipeline_task, timeout=5)

        second_doc_id = compute_mdhash_id(file_paths[1], prefix="doc-")
        second_status = await rag.doc_status.get_by_id(second_doc_id)
        assert second_status is not None
        assert _status_to_text(second_status["status"]) == "failed"
        assert second_status["file_path"] == "second.md"
        assert second_status["error_msg"] == "User cancelled"
    finally:
        await rag.finalize_storages()


@pytest.mark.asyncio
async def test_pipeline_cancellation_repairs_placeholder_file_path_for_queued_docs(
    tmp_path, monkeypatch
):
    rag = await _build_rag(
        tmp_path,
        "cancel_repair_placeholder_file_path",
        _deterministic_chunking,
        max_parallel_insert=1,
    )
    try:
        contents = ["first queued content", "second queued content"]
        file_paths = ["first.md", "second.md"]
        await rag.apipeline_enqueue_documents(input=contents, file_paths=file_paths)

        second_doc_id = compute_mdhash_id(file_paths[1], prefix="doc-")
        second_status = await rag.doc_status.get_by_id(second_doc_id)
        assert second_status is not None
        second_status["file_path"] = "unknown_source"
        await rag.doc_status.upsert({second_doc_id: second_status})

        extraction_started = asyncio.Event()
        release_first_doc = asyncio.Event()

        async def _blocking_extract(
            self, chunks, pipeline_status, pipeline_status_lock
        ):
            extraction_started.set()
            await release_first_doc.wait()
            return []

        monkeypatch.setattr(
            rag,
            "_process_extract_entities",
            MethodType(_blocking_extract, rag),
        )

        pipeline_task = asyncio.create_task(rag.apipeline_process_enqueue_documents())
        await asyncio.wait_for(extraction_started.wait(), timeout=5)

        pipeline_status = await get_namespace_data(
            "pipeline_status", workspace=rag.workspace
        )
        pipeline_status_lock = get_namespace_lock(
            "pipeline_status", workspace=rag.workspace
        )
        async with pipeline_status_lock:
            pipeline_status["cancellation_requested"] = True

        release_first_doc.set()
        await asyncio.wait_for(pipeline_task, timeout=5)

        repaired_status = await rag.doc_status.get_by_id(second_doc_id)
        assert repaired_status is not None
        assert _status_to_text(repaired_status["status"]) == "failed"
        assert repaired_status["file_path"] == "second.md"
        assert repaired_status["error_msg"] == "User cancelled"
    finally:
        await rag.finalize_storages()


@pytest.mark.asyncio
async def test_delete_doc_entries_guard_prevents_zombie_record(tmp_path, monkeypatch):
    """When doc_status.delete fails, the guard must not re-create a zombie record.

    The exception handler skips _update_delete_retry_state when deletion_stage is
    "delete_doc_entries". doc_status.delete runs first, so full_docs is still intact
    on failure. This test confirms: (a) the result is status="fail", and (b) the
    exception handler does not upsert a new record with deletion_failed=True.
    """
    rag = await _build_rag(
        tmp_path, "delete_doc_entries_guard", _deterministic_chunking
    )
    try:
        doc_id = "doc-delete-entries-guard"
        drop_chunk_id = "chunk-drop"
        await _seed_delete_retry_state(
            rag,
            doc_id=doc_id,
            status_chunk_ids=[drop_chunk_id],
            tracking_chunk_ids=[drop_chunk_id],
            chunk_owners={drop_chunk_id: doc_id},
        )

        original_delete = rag.doc_status.delete

        call_count = 0

        async def fail_doc_status_delete(ids):
            nonlocal call_count
            call_count += 1
            raise RuntimeError("doc_status delete fail sentinel")

        monkeypatch.setattr(rag.doc_status, "delete", fail_doc_status_delete)

        result = await rag.adelete_by_doc_id(doc_id)

        monkeypatch.setattr(rag.doc_status, "delete", original_delete)

        assert result.status == "fail"
        assert "doc_status delete fail sentinel" in result.message
        # doc_status.delete is now called first (before full_docs.delete). The patch
        # makes it fail, so doc_status still exists. The guard must not call
        # _update_delete_retry_state (which would upsert the record with
        # deletion_failed=True) when deletion_stage is "delete_doc_entries".
        status_record = await rag.doc_status.get_by_id(doc_id)
        assert (
            status_record is not None
        ), "doc_status should still exist (delete failed)"
        metadata = status_record.get("metadata", {})
        assert not metadata.get(
            "deletion_failed"
        ), "guard failed: zombie record written with deletion_failed=True"
    finally:
        await rag.finalize_storages()


@pytest.mark.asyncio
async def test_retry_state_write_failure_in_exception_handler_still_returns_fail(
    tmp_path, monkeypatch
):
    """If _update_delete_retry_state itself fails inside the exception handler,
    the caller must still receive a well-formed DeletionResult with status="fail"
    and the original error message — not an unhandled exception.
    """
    rag = await _build_rag(
        tmp_path, "retry_state_write_failure", _deterministic_chunking
    )
    try:
        doc_id = "doc-retry-state-write-fail"
        keep_chunk_id = "chunk-keep"
        drop_chunk_id = "chunk-drop"
        # Include a keep_chunk_id so entities survive and rebuild is triggered
        await _seed_delete_retry_state(
            rag,
            doc_id=doc_id,
            status_chunk_ids=[drop_chunk_id],
            tracking_chunk_ids=[keep_chunk_id, drop_chunk_id],
            chunk_owners={
                keep_chunk_id: "doc-keep",
                drop_chunk_id: doc_id,
            },
        )

        async def fail_rebuild(**kwargs):
            raise RuntimeError("rebuild fail sentinel")

        async def fail_upsert(data):
            raise RuntimeError("doc_status upsert fail sentinel")

        monkeypatch.setattr(
            lightrag_module, "rebuild_knowledge_from_chunks", fail_rebuild
        )
        monkeypatch.setattr(rag.doc_status, "upsert", fail_upsert)

        result = await rag.adelete_by_doc_id(doc_id)

        assert result.status == "fail"
        # The original rebuild error must be present in the message; the compound
        # "Additionally, failed to persist retry state" suffix may also be present.
        assert "rebuild fail sentinel" in result.message
    finally:
        await rag.finalize_storages()


@pytest.mark.asyncio
async def test_persist_pre_rebuild_failure_records_stage_and_allows_retry(
    tmp_path, monkeypatch
):
    """A failure in _insert_done (persist_pre_rebuild_changes stage) records the
    correct stage in doc_status metadata, and a subsequent retry completes the
    deletion successfully.
    """
    rag = await _build_rag(
        tmp_path, "persist_pre_rebuild_failure", _deterministic_chunking
    )
    try:
        doc_id = "doc-persist-pre-rebuild-failure"
        keep_chunk_id = "chunk-keep"
        drop_chunk_id = "chunk-drop"
        seeded = await _seed_delete_retry_state(
            rag,
            doc_id=doc_id,
            status_chunk_ids=[drop_chunk_id],
            tracking_chunk_ids=[keep_chunk_id, drop_chunk_id],
            chunk_owners={
                keep_chunk_id: "doc-keep",
                drop_chunk_id: doc_id,
            },
        )
        entity_a = seeded["entity_a"]
        # entity_b = seeded["entity_b"]

        insert_done_calls = 0
        original_insert_done = rag._insert_done

        async def fail_first_insert_done():
            nonlocal insert_done_calls
            insert_done_calls += 1
            if insert_done_calls == 1:
                raise RuntimeError("insert_done fail sentinel")
            await original_insert_done()

        monkeypatch.setattr(rag, "_insert_done", fail_first_insert_done)

        first_result = await rag.adelete_by_doc_id(doc_id)

        failed_status = await rag.doc_status.get_by_id(doc_id)
        assert first_result.status == "fail"
        assert "insert_done fail sentinel" in first_result.message
        assert failed_status is not None
        assert failed_status["metadata"]["deletion_failed"] is True
        assert (
            failed_status["metadata"]["deletion_failure_stage"]
            == "persist_pre_rebuild_changes"
        )

        monkeypatch.undo()
        monkeypatch.setattr(
            lightrag_module,
            "rebuild_knowledge_from_chunks",
            _succeed_rebuild_from_remaining_chunks,
        )
        second_result = await rag.adelete_by_doc_id(doc_id)

        assert second_result.status == "success"
        assert await rag.doc_status.get_by_id(doc_id) is None
        assert await rag.full_docs.get_by_id(doc_id) is None
        assert await rag.text_chunks.get_by_id(drop_chunk_id) is None
        assert await rag.text_chunks.get_by_id(keep_chunk_id) is not None
        entity_a_tracking = await rag.entity_chunks.get_by_id(entity_a)
        assert entity_a_tracking is not None
        assert entity_a_tracking["chunk_ids"] == [keep_chunk_id]
    finally:
        await rag.finalize_storages()


@pytest.mark.asyncio
async def test_delete_doc_graph_metadata_failure_records_stage_and_allows_retry(
    tmp_path, monkeypatch
):
    """A failure in full_relations.delete (delete_doc_graph_metadata stage) records
    the correct stage in retry metadata, and a subsequent retry completes the deletion.
    """
    rag = await _build_rag(
        tmp_path, "delete_doc_graph_metadata_failure", _deterministic_chunking
    )
    try:
        doc_id = "doc-graph-metadata-failure"
        drop_chunk_id = "chunk-drop"
        await _seed_delete_retry_state(
            rag,
            doc_id=doc_id,
            status_chunk_ids=[drop_chunk_id],
            tracking_chunk_ids=[drop_chunk_id],
            chunk_owners={drop_chunk_id: doc_id},
        )

        relations_delete_calls = 0
        original_relations_delete = rag.full_relations.delete

        async def fail_first_relations_delete(ids):
            nonlocal relations_delete_calls
            relations_delete_calls += 1
            if relations_delete_calls == 1:
                raise RuntimeError("full_relations delete fail sentinel")
            await original_relations_delete(ids)

        monkeypatch.setattr(rag.full_relations, "delete", fail_first_relations_delete)

        first_result = await rag.adelete_by_doc_id(doc_id)

        failed_status = await rag.doc_status.get_by_id(doc_id)
        assert first_result.status == "fail"
        assert "full_relations delete fail sentinel" in first_result.message
        assert failed_status is not None
        assert failed_status["metadata"]["deletion_failed"] is True
        assert (
            failed_status["metadata"]["deletion_failure_stage"]
            == "delete_doc_graph_metadata"
        )

        monkeypatch.undo()
        second_result = await rag.adelete_by_doc_id(doc_id)

        assert second_result.status == "success"
        assert await rag.doc_status.get_by_id(doc_id) is None
        assert await rag.full_docs.get_by_id(doc_id) is None
        assert await rag.full_entities.get_by_id(doc_id) is None
        assert await rag.full_relations.get_by_id(doc_id) is None
    finally:
        await rag.finalize_storages()


@pytest.mark.asyncio
async def test_deletion_fully_completed_prevents_success_override_in_finally(
    tmp_path, monkeypatch
):
    """When deletion completes successfully but _insert_done fails in the finally
    block, the already-returned success result must not be overridden with a failure.
    This covers both the no-chunk path and the full deletion path.
    """
    for scenario in ("no_chunk", "full"):
        rag = await _build_rag(
            tmp_path,
            f"deletion_fully_completed_{scenario}",
            _deterministic_chunking,
        )
        try:
            doc_id = f"doc-fully-completed-{scenario}"

            if scenario == "no_chunk":
                # Seed a doc with no chunks: doc_status.chunks_list is empty.
                now = datetime.now(timezone.utc).isoformat()
                await rag.full_docs.upsert(
                    {doc_id: {"content": "no chunk doc", "file_path": "nc.txt"}}
                )
                await rag.doc_status.upsert(
                    {
                        doc_id: {
                            "status": DocStatus.PROCESSED,
                            "content_summary": "no chunks",
                            "content_length": 11,
                            "chunks_count": 0,
                            "chunks_list": [],
                            "created_at": now,
                            "updated_at": now,
                            "file_path": "nc.txt",
                            "track_id": f"track-{doc_id}",
                            "error_msg": "",
                            "metadata": {},
                        }
                    }
                )
            else:
                drop_chunk_id = "chunk-drop-fc"
                await _seed_delete_retry_state(
                    rag,
                    doc_id=doc_id,
                    status_chunk_ids=[drop_chunk_id],
                    tracking_chunk_ids=[drop_chunk_id],
                    chunk_owners={drop_chunk_id: doc_id},
                )

            insert_done_calls = 0
            original_insert_done = rag._insert_done

            async def fail_later_insert_done():
                nonlocal insert_done_calls
                insert_done_calls += 1
                # Let the first call (persist_pre_rebuild_changes) succeed for the
                # full path; only fail the finally-block call.
                if insert_done_calls <= (1 if scenario == "full" else 0):
                    await original_insert_done()
                else:
                    raise RuntimeError("finally insert_done fail sentinel")

            monkeypatch.setattr(rag, "_insert_done", fail_later_insert_done)

            result = await rag.adelete_by_doc_id(doc_id)

            assert (
                result.status == "success"
            ), f"[{scenario}] expected success but got {result.status}: {result.message}"
        finally:
            monkeypatch.undo()
            await rag.finalize_storages()