from __future__ import annotations

import asyncio
import json
from types import SimpleNamespace
from typing import Any

import httpx
import pytest

from app.gateway import wiki_directory_constraints as constraints
from app.gateway.wiki_directory_constraints import (
    ensure_directory_paths,
    ensure_fixed_parent_directories,
    normalize_directory_paths,
    normalize_parent_directories,
    reconcile_fixed_parent_directories,
)
from deerflow.integrations.weknora.client import WeKnoraClient


class FakeConflict(RuntimeError):
    status_code = 409


class FakeWeKnora:
    def __init__(
        self,
        *,
        folders: list[dict[str, Any]] | None = None,
        pages: list[dict[str, Any]] | None = None,
        folder_envelope: str = "folders",
    ) -> None:
        self.folders = [dict(item) for item in (folders or [])]
        self.pages = [dict(item) for item in (pages or [])]
        self.folder_envelope = folder_envelope
        self.calls: list[tuple[Any, ...]] = []
        self._next_id = 1

    async def list_wiki_folders(
        self,
        knowledge_base_id: str,
        *,
        parent_id: str = "",
        page_types: list[str] | None = None,
    ) -> Any:
        self.calls.append(("list_folders", knowledge_base_id, parent_id))
        rows = [dict(item) for item in self.folders if str(item.get("parent_id") or "") == parent_id]
        if self.folder_envelope == "data-list":
            return {"data": rows}
        if self.folder_envelope == "data-folders":
            return {"data": {"folders": rows}}
        return {"parent_id": parent_id, "folders": rows}

    async def create_wiki_folder(self, knowledge_base_id: str, *, name: str, parent_id: str = "") -> dict[str, Any]:
        self.calls.append(("create_folder", knowledge_base_id, name, parent_id))
        if any(str(item.get("parent_id") or "") == parent_id and item.get("name") == name for item in self.folders):
            raise FakeConflict("folder already exists")
        folder = {"id": f"created-{self._next_id}", "name": name, "parent_id": parent_id}
        self._next_id += 1
        self.folders.append(folder)
        return dict(folder)

    async def update_wiki_folder(self, knowledge_base_id: str, folder_id: str, changes: dict[str, Any]) -> dict[str, Any]:
        self.calls.append(("update_folder", knowledge_base_id, folder_id, dict(changes)))
        folder = next(item for item in self.folders if item["id"] == folder_id)
        target_parent = str(changes.get("parent_id") if changes.get("move_parent") else folder.get("parent_id") or "")
        target_name = str(changes.get("name", folder.get("name") or ""))
        if any(
            item["id"] != folder_id
            and str(item.get("parent_id") or "") == target_parent
            and item.get("name") == target_name
            for item in self.folders
        ):
            raise FakeConflict("folder conflict")
        if "name" in changes:
            folder["name"] = changes["name"]
        if changes.get("move_parent"):
            folder["parent_id"] = str(changes.get("parent_id") or "")
        return dict(folder)

    async def delete_wiki_folder(self, knowledge_base_id: str, folder_id: str) -> None:
        self.calls.append(("delete_folder", knowledge_base_id, folder_id))
        self.folders = [item for item in self.folders if item["id"] != folder_id]

    async def list_wiki_pages(
        self,
        knowledge_base_id: str,
        *,
        page: int,
        page_size: int,
        folder_id: str | None = None,
    ) -> dict[str, Any]:
        self.calls.append(("list_pages", knowledge_base_id, page, page_size, folder_id))
        rows = self.pages if folder_id is None else [item for item in self.pages if str(item.get("folder_id") or "") == folder_id]
        start = (page - 1) * page_size
        batch = [dict(item) for item in rows[start : start + page_size]]
        return {"data": {"pages": batch, "total": len(rows), "page": page, "page_size": page_size}}

    async def move_wiki_page(self, knowledge_base_id: str, slug: str, folder_id: str) -> dict[str, Any]:
        self.calls.append(("move_page", knowledge_base_id, slug, folder_id))
        page = next(item for item in self.pages if item["slug"] == slug)
        page["folder_id"] = folder_id
        return dict(page)

    async def get_wiki_page(
        self,
        knowledge_base_id: str,
        slug: str,
    ) -> dict[str, Any]:
        self.calls.append(("get_page", knowledge_base_id, slug))
        return dict(next(item for item in self.pages if item["slug"] == slug))

    async def update_wiki_page(
        self,
        knowledge_base_id: str,
        slug: str,
        changes: dict[str, Any],
    ) -> dict[str, Any]:
        self.calls.append(("update_page", knowledge_base_id, slug, dict(changes)))
        page = next(item for item in self.pages if item["slug"] == slug)
        # The production WeKnora version binds PUT to a fresh WikiPage and
        # replaces these fields.  Missing keys therefore become zero values;
        # modelling PATCH via ``dict.update`` would hide destructive callers.
        defaults: dict[str, Any] = {
            "title": "",
            "content": "",
            "summary": "",
            "page_type": "",
            "status": "",
            "aliases": [],
            "parent_slug": "",
            "folder_id": "",
            "sort_order": 0,
            "source_refs": [],
            "chunk_refs": [],
            "page_metadata": {},
        }
        page.update(
            {
                key: changes.get(key, default)
                for key, default in defaults.items()
            }
        )
        return dict(page)

    async def delete_wiki_page(self, knowledge_base_id: str, slug: str) -> None:
        self.calls.append(("delete_page", knowledge_base_id, slug))
        self.pages = [item for item in self.pages if item["slug"] != slug]

    async def rebuild_wiki_links(self, knowledge_base_id: str) -> None:
        self.calls.append(("rebuild_links", knowledge_base_id))


def test_normalize_parent_directories_preserves_order_and_enforces_limits() -> None:
    assert normalize_parent_directories(["  政治  ", "", "经济", "政治", " 军事 "]) == ["政治", "经济", "军事"]
    assert normalize_parent_directories(None) == []

    for unsafe in ("政治/制度", "军事\\体系", "经济|产业", "社会／文化", "科技｜创新", "含\t制表符"):
        with pytest.raises(ValueError):
            normalize_parent_directories([unsafe])
    with pytest.raises(ValueError, match="100 项"):
        normalize_parent_directories([f"目录{index}" for index in range(101)])
    with pytest.raises(ValueError, match="100 个字符"):
        normalize_parent_directories(["章" * 101])
    with pytest.raises(ValueError, match="6000"):
        normalize_parent_directories(["章" * 100 for _ in range(61)])


def test_normalize_directory_paths_builds_canonical_parent_first_tree() -> None:
    assert normalize_directory_paths(
        [
            [" 政策法规 ", " 国内政策 "],
            ["政策法规"],
            ["政策法规", "国内政策", "申报流程"],
            ["政策法规", "国内政策"],
        ]
    ) == [
        ["政策法规"],
        ["政策法规", "国内政策"],
        ["政策法规", "国内政策", "申报流程"],
    ]
    # Equivalent root spellings keep the first display form for descendants.
    assert normalize_directory_paths([["Ａ"], ["a", "二级"]]) == [
        ["A"],
        ["A", "二级"],
    ]

    with pytest.raises(ValueError, match="1～3"):
        normalize_directory_paths([["一", "二", "三", "四"]])
    with pytest.raises(ValueError, match="不能是"):
        normalize_directory_paths([[".."]])


def test_focus_classifier_parser_ignores_in_band_reasoning_json() -> None:
    text = (
        '<think>Do not parse this example: {"decisions":{}}.</think>\n'
        '{"decisions":{"page:noise":{"keep":false,"directory":""}}}'
    )

    assert constraints._parse_focus_classifier_response(
        text,
        {"page:noise"},
        {"政策 / 目标"},
    ) == {"page:noise": {"keep": False, "directory": ""}}


@pytest.mark.asyncio
async def test_default_focus_classifier_retries_only_omitted_items(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    prompts: list[str] = []
    responses = [
        {
            "decisions": {
                "page:one": {
                    "keep": True,
                    "directory": "政策 / 目标",
                }
            }
        },
        {
            "decisions": {
                "page:two": {"keep": False, "directory": ""},
                "page:three": {
                    "keep": True,
                    "directory": "政策 / 目标",
                },
            }
        },
    ]

    class Model:
        async def ainvoke(self, prompt: str) -> SimpleNamespace:
            prompts.append(prompt)
            return SimpleNamespace(
                content=json.dumps(responses.pop(0), ensure_ascii=False)
            )

    import deerflow.models

    monkeypatch.setattr(
        deerflow.models,
        "create_chat_model",
        lambda **_: Model(),
    )
    items = [
        {
            "item_id": f"page:{name}",
            "kind": "page",
            "title": name,
        }
        for name in ("one", "two", "three")
    ]

    result = await constraints._default_focus_classifier(
        items,
        "只保留目标",
        ["政策 / 目标"],
    )

    assert result == {
        "page:one": {"keep": True, "directory": "政策 / 目标"},
        "page:two": {"keep": False, "directory": ""},
        "page:three": {"keep": True, "directory": "政策 / 目标"},
    }
    assert len(prompts) == 2
    assert "page:one" in prompts[0]
    assert "page:one" not in prompts[1]
    assert "page:two" in prompts[1]
    assert "page:three" in prompts[1]


@pytest.mark.asyncio
async def test_default_focus_classifier_resolves_stubborn_omission_individually(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    prompts: list[str] = []
    responses = [
        {"decisions": {}},
        {"decisions": {}},
        {"keep": False, "directory": ""},
    ]

    class Model:
        async def ainvoke(self, prompt: str) -> SimpleNamespace:
            prompts.append(prompt)
            return SimpleNamespace(
                content=json.dumps(responses.pop(0), ensure_ascii=False)
            )

    import deerflow.models

    monkeypatch.setattr(
        deerflow.models,
        "create_chat_model",
        lambda **_: Model(),
    )

    result = await constraints._default_focus_classifier(
        [{"item_id": "page:noise", "kind": "page", "title": "无关噪声"}],
        "只保留量化目标",
        ["政策 / 目标"],
    )

    assert result == {"page:noise": {"keep": False, "directory": ""}}
    assert len(prompts) == 3
    assert "There is exactly one page" in prompts[-1]


@pytest.mark.asyncio
async def test_ensure_directory_paths_recursively_creates_each_level_once() -> None:
    client = FakeWeKnora()
    paths = [
        ["政策法规"],
        ["政策法规", "国内政策"],
        ["政策法规", "国内政策", "申报流程"],
    ]

    first = await ensure_directory_paths(client, "kb-one", paths)
    second = await ensure_directory_paths(client, "kb-one", paths)

    assert first == second
    assert list(first) == [
        "政策法规",
        "政策法规 / 国内政策",
        "政策法规 / 国内政策 / 申报流程",
    ]
    assert [call for call in client.calls if call[0] == "create_folder"] == [
        ("create_folder", "kb-one", "政策法规", ""),
        ("create_folder", "kb-one", "国内政策", first["政策法规"]),
        (
            "create_folder",
            "kb-one",
            "申报流程",
            first["政策法规 / 国内政策"],
        ),
    ]


@pytest.mark.asyncio
async def test_ensure_directory_paths_adopts_empty_folders_after_conflict() -> None:
    """WeKnora hides empty folders outside its merged page-type view.

    The first read for each parent also simulates a concurrent-create race:
    list returns no row, create reports 409, and the conflict re-read exposes
    the already-created empty folder.
    """

    class EmptyFolderRaceClient:
        def __init__(self) -> None:
            self.folders = [
                {
                    "id": "existing-root",
                    "name": "国内新闻",
                    "parent_id": "",
                },
                {
                    "id": "existing-child",
                    "name": "政策发布",
                    "parent_id": "existing-root",
                },
            ]
            self.list_count: dict[str, int] = {}
            self.create_calls: list[tuple[str, str]] = []

        async def list_wiki_folders(
            self,
            knowledge_base_id: str,
            *,
            parent_id: str = "",
            page_types: list[str] | None = None,
        ) -> dict[str, Any]:
            assert knowledge_base_id == "kb-news"
            assert page_types == [
                "summary",
                "entity",
                "concept",
                "index",
                "synthesis",
                "comparison",
                "article",
            ]
            count = self.list_count.get(parent_id, 0) + 1
            self.list_count[parent_id] = count
            rows = [
                dict(item)
                for item in self.folders
                if str(item.get("parent_id") or "") == parent_id
            ]
            # A stale first read creates the exact list/create race that the
            # conflict recovery path must tolerate.
            if count == 1:
                rows = []
            return {"parent_id": parent_id, "folders": rows}

        async def create_wiki_folder(
            self,
            knowledge_base_id: str,
            *,
            name: str,
            parent_id: str = "",
        ) -> dict[str, Any]:
            assert knowledge_base_id == "kb-news"
            self.create_calls.append((name, parent_id))
            raise FakeConflict("wiki folder name conflict")

    client = EmptyFolderRaceClient()
    paths = [["国内新闻"], ["国内新闻", "政策发布"]]

    first = await ensure_directory_paths(client, "kb-news", paths)
    second = await ensure_directory_paths(client, "kb-news", paths)

    assert first == second == {
        "国内新闻": "existing-root",
        "国内新闻 / 政策发布": "existing-child",
    }
    assert client.create_calls == [
        ("国内新闻", ""),
        ("政策发布", "existing-root"),
    ]


@pytest.mark.asyncio
async def test_ensure_directory_paths_matches_existing_name_by_nfkc_casefold() -> None:
    client = FakeWeKnora(
        folders=[{"id": "root-a", "name": "Ａ", "parent_id": ""}],
    )

    result = await ensure_directory_paths(client, "kb-one", [["a"]])

    assert result == {"a": "root-a"}
    assert not [call for call in client.calls if call[0] == "create_folder"]


@pytest.mark.asyncio
async def test_empty_constraints_make_no_remote_calls() -> None:
    client = FakeWeKnora()

    assert await ensure_fixed_parent_directories(client, "kb-one", None) == {}
    result = await reconcile_fixed_parent_directories(
        client,
        {"weknora_id": "kb-one", "wiki_parent_directories": []},
    )

    assert result == {
        "roots": {},
        "created_roots": 0,
        "moved_folders": 0,
        "renamed_folders": 0,
        "moved_pages": 0,
    }
    assert client.calls == []


@pytest.mark.asyncio
@pytest.mark.parametrize("folder_envelope", ["folders", "data-folders", "data-list"])
async def test_ensure_fixed_roots_rebuilds_only_missing_names(folder_envelope: str) -> None:
    client = FakeWeKnora(
        folders=[{"id": "root-a", "name": "政治", "parent_id": ""}],
        folder_envelope=folder_envelope,
    )

    first = await ensure_fixed_parent_directories(client, "kb-one", ["政治", "经济"])
    second = await ensure_fixed_parent_directories(client, "kb-one", ["政治", "经济"])

    assert list(first) == ["政治", "经济"]
    assert first == second
    assert [item["name"] for item in client.folders if not item.get("parent_id")] == ["政治", "经济"]
    assert [call for call in client.calls if call[0] == "create_folder"] == [
        ("create_folder", "kb-one", "经济", "")
    ]


@pytest.mark.asyncio
async def test_legacy_directory_reference_does_not_move_folders_or_pages() -> None:
    client = FakeWeKnora(
        folders=[
            {"id": "politics", "name": "政治", "parent_id": ""},
            {"id": "existing-defense", "name": "军事", "parent_id": "politics"},
            {"id": "extra-defense", "name": "军事", "parent_id": ""},
            {"id": "extra-defense-child", "name": "装备", "parent_id": "extra-defense"},
            {"id": "extra-market", "name": "市场", "parent_id": ""},
        ],
        pages=[
            {"slug": "index", "title": "Wiki Index", "page_type": "index", "folder_id": ""},
            {"slug": "directory", "title": "目录", "page_type": "summary", "folder_id": ""},
            {"slug": "weekly-report", "title": "每周经济报告", "page_type": "summary", "folder_id": ""},
        ],
    )
    classified: list[str] = []

    async def classifier(item: dict[str, Any], roots: list[str]) -> str:
        classified.append(str(item["item_id"]))
        assert roots == ["政治", "经济"]
        if item.get("name") == "军事":
            return "政治"
        return "经济"

    first = await reconcile_fixed_parent_directories(
        client,
        {"weknora_id": "kb-one", "wiki_parent_directories": ["政治", "经济"]},
        classifier=classifier,
    )
    second = await reconcile_fixed_parent_directories(
        client,
        {"weknora_id": "kb-one", "wiki_parent_directories": ["政治", "经济"]},
        classifier=classifier,
    )

    assert first["created_roots"] == 0
    assert first["moved_folders"] == 0
    assert first["renamed_folders"] == 0
    assert first["moved_pages"] == 0
    assert second == first
    assert classified == []

    top_level = {item["name"]: item["id"] for item in client.folders if not item.get("parent_id")}
    assert set(top_level) == {"政治", "军事", "市场"}
    moved_defense = next(item for item in client.folders if item["id"] == "extra-defense")
    assert moved_defense["parent_id"] == ""
    assert moved_defense["name"] == "军事"
    assert next(item for item in client.folders if item["id"] == "extra-defense-child")["parent_id"] == "extra-defense"
    assert next(item for item in client.folders if item["id"] == "extra-market")["parent_id"] == ""
    assert next(item for item in client.pages if item["slug"] == "weekly-report")["folder_id"] == ""
    assert next(item for item in client.pages if item["slug"] == "index")["folder_id"] == ""
    assert next(item for item in client.pages if item["slug"] == "directory")["folder_id"] == ""


@pytest.mark.asyncio
async def test_directory_reference_preserves_article_only_extra_folder() -> None:
    class PageTypeFilteringFolders(FakeWeKnora):
        async def list_wiki_folders(
            self,
            knowledge_base_id: str,
            *,
            parent_id: str = "",
            page_types: list[str] | None = None,
        ) -> dict[str, Any]:
            self.calls.append(("list_folders", knowledge_base_id, parent_id))
            allowed_types = set(page_types or [])
            rows = []
            for folder in self.folders:
                if str(folder.get("parent_id") or "") != parent_id:
                    continue
                folder_id = str(folder.get("id") or "")
                if any(
                    str(page.get("folder_id") or "") == folder_id
                    and str(page.get("page_type") or "") in allowed_types
                    for page in self.pages
                ):
                    rows.append(dict(folder))
            return {"parent_id": parent_id, "folders": rows}

    client = PageTypeFilteringFolders(
        folders=[
            {
                "id": "manual-extra",
                "name": "人工目录",
                "parent_id": "",
            }
        ],
        pages=[
            {
                "slug": "manual-article",
                "title": "人工维护页",
                "content": "由管理员手工创建。",
                "page_type": "article",
                "folder_id": "manual-extra",
            }
        ],
    )

    result = await reconcile_fixed_parent_directories(
        client,
        {
            "weknora_id": "kb-one",
            "wiki_directory_paths": [["允许目录"]],
            "wiki_focus_instructions": "",
        },
    )

    assert result["moved_pages"] == 0
    assert {folder["name"] for folder in client.folders} == {"人工目录"}
    assert client.pages[0]["folder_id"] == "manual-extra"
    assert client.calls == []


@pytest.mark.asyncio
async def test_directory_reference_does_not_limit_strict_focus_filter() -> None:
    client = FakeWeKnora(
        folders=[
            {"id": "root-policy", "name": "政策", "parent_id": ""},
            {"id": "leaf-target", "name": "目标", "parent_id": "root-policy"},
            {"id": "extra-root", "name": "石窟寺", "parent_id": ""},
            {"id": "extra-child", "name": "保护利用", "parent_id": "extra-root"},
        ],
        pages=[
            {
                "slug": "concept/charging-target",
                "title": "2027年充电设施目标",
                "summary": "设施总量与公共充电容量",
                "page_type": "concept",
                "folder_id": "extra-root",
                "source_refs": ["doc-a"],
            },
            {
                "slug": "concept/cave-temple",
                "title": "石窟寺保护利用",
                "summary": "文物保护背景材料",
                "page_type": "concept",
                "folder_id": "extra-child",
                "source_refs": ["doc-b"],
            },
            {
                "slug": "extra-directory",
                "title": "石窟寺目录",
                "page_type": "directory",
                "folder_id": "extra-root",
            },
        ],
    )

    focus_calls = 0

    async def focus_classifier(
        items: list[dict[str, Any]],
        focus: str,
        allowed_paths: list[str],
    ) -> dict[str, dict[str, Any]]:
        nonlocal focus_calls
        focus_calls += 1
        assert focus == "只提取明确年份和充电设施量化目标"
        assert allowed_paths == []
        assert {item["item_id"] for item in items} == {
            "page:concept/charging-target",
            "page:concept/cave-temple",
        }
        return {
            "page:concept/charging-target": {
                "keep": True,
                "directory": "",
            },
            "page:concept/cave-temple": {"keep": False, "directory": ""},
        }

    result = await reconcile_fixed_parent_directories(
        client,
        {
            "weknora_id": "kb-one",
            "wiki_directory_paths": [["政策"], ["政策", "目标"]],
            "wiki_focus_instructions": "只提取明确年份和充电设施量化目标",
        },
        focus_classifier=focus_classifier,
    )
    second = await reconcile_fixed_parent_directories(
        client,
        {
            "weknora_id": "kb-one",
            "wiki_directory_paths": [["政策"], ["政策", "目标"]],
            "wiki_focus_instructions": "只提取明确年份和充电设施量化目标",
        },
        focus_classifier=focus_classifier,
    )

    assert result["focus_enforced"] is True
    assert result["strict_hierarchy_enforced"] is False
    assert result["moved_pages"] == 0
    assert result["deleted_pages"] == 1
    assert result["deleted_folders"] == 0
    assert {folder["id"] for folder in client.folders} == {
        "root-policy",
        "leaf-target",
        "extra-root",
        "extra-child",
    }
    assert focus_calls == 1
    assert second["moved_pages"] == 0
    assert second["deleted_pages"] == 0
    assert second["deleted_folders"] == 0
    assert [page["slug"] for page in client.pages] == [
        "concept/charging-target",
        "extra-directory",
    ]
    kept_page = client.pages[0]
    assert kept_page["folder_id"] == "extra-root"
    marker = kept_page["page_metadata"]["deerflow_generation_constraints"]
    assert marker["schema"] == 1
    assert marker["keep"] is True
    assert marker["directory"] == ""
    assert len([call for call in client.calls if call[0] == "update_page"]) == 1
    assert ("rebuild_links", "kb-one") in client.calls


@pytest.mark.asyncio
async def test_focus_marker_preserves_full_page_across_replacement_put() -> None:
    original = {
        "slug": "concept/data-infrastructure",
        "title": "数据基础设施",
        "content": "到2029年建成主体结构，到2035年形成完善体系。",
        "summary": "建设目标与时间表",
        "page_type": "concept",
        "status": "published",
        "aliases": ["数据底座"],
        "parent_slug": "concept/digital-china",
        "folder_id": "target-folder",
        "sort_order": 7,
        "source_refs": ["doc-one|建设指引"],
        "chunk_refs": ["chunk-one"],
        "page_metadata": {"language": "zh-CN"},
    }
    client = FakeWeKnora(
        folders=[{"id": "target-folder", "name": "建设指引", "parent_id": ""}],
        pages=[original],
    )

    async def keep_page(
        items: list[dict[str, Any]],
        _: str,
        __: list[str],
    ) -> dict[str, dict[str, Any]]:
        return {
            str(items[0]["item_id"]): {
                "keep": True,
                "directory": "建设指引",
            }
        }

    await reconcile_fixed_parent_directories(
        client,
        {
            "weknora_id": "kb-one",
            "wiki_directory_paths": [["建设指引"]],
            "wiki_focus_instructions": "只保留建设目标和明确时间节点",
        },
        focus_classifier=keep_page,
    )

    page = client.pages[0]
    for key in (
        "title",
        "content",
        "summary",
        "page_type",
        "status",
        "aliases",
        "parent_slug",
        "folder_id",
        "sort_order",
        "source_refs",
        "chunk_refs",
    ):
        assert page[key] == original[key]
    assert page["page_metadata"]["language"] == "zh-CN"
    assert (
        page["page_metadata"]["deerflow_generation_constraints"]["keep"]
        is True
    )

    update_calls = [call for call in client.calls if call[0] == "update_page"]
    assert len(update_calls) == 1
    payload = update_calls[0][3]
    assert {
        "title",
        "content",
        "summary",
        "page_type",
        "status",
        "aliases",
        "parent_slug",
        "folder_id",
        "sort_order",
        "source_refs",
        "chunk_refs",
        "page_metadata",
    } <= payload.keys()


@pytest.mark.asyncio
async def test_focus_only_constraint_filters_pages_without_changing_directories() -> None:
    client = FakeWeKnora(
        folders=[{"id": "existing-folder", "name": "原有目录", "parent_id": ""}],
        pages=[
            {
                "slug": "keep",
                "title": "2029目标",
                "folder_id": "existing-folder",
                "source_refs": ["doc-one"],
            },
            {
                "slug": "drop",
                "title": "无关人物",
                "folder_id": "existing-folder",
                "source_refs": ["doc-one"],
            },
        ]
    )

    async def focus_classifier(
        items: list[dict[str, Any]],
        focus: str,
        allowed_paths: list[str],
    ) -> dict[str, dict[str, Any]]:
        assert focus == "只提取2029年建设目标"
        assert allowed_paths == []
        return {
            "page:keep": {"keep": True, "directory": ""},
            "page:drop": {"keep": False, "directory": ""},
        }

    result = await reconcile_fixed_parent_directories(
        client,
        {
            "weknora_id": "kb-one",
            "wiki_directory_paths": [],
            "wiki_focus_instructions": "只提取2029年建设目标",
        },
        focus_classifier=focus_classifier,
    )

    assert [page["slug"] for page in client.pages] == ["keep"]
    assert client.pages[0]["folder_id"] == "existing-folder"
    assert [folder["id"] for folder in client.folders] == ["existing-folder"]
    assert result["deleted_pages"] == 1
    assert result["moved_pages"] == 0
    assert result["deleted_folders"] == 0


@pytest.mark.asyncio
async def test_focus_uses_full_page_body_invalidates_marker_and_protects_manual_pages() -> None:
    class SummaryOnlyListing(FakeWeKnora):
        async def list_wiki_pages(
            self,
            knowledge_base_id: str,
            *,
            page: int,
            page_size: int,
            folder_id: str | None = None,
        ) -> dict[str, Any]:
            result = await super().list_wiki_pages(
                knowledge_base_id,
                page=page,
                page_size=page_size,
                folder_id=folder_id,
            )
            for row in result["data"]["pages"]:
                row.pop("content", None)
            return result

    client = SummaryOnlyListing(
        pages=[
            {
                "slug": "generated",
                "title": "综合说明",
                "summary": "正文包含具体目标",
                "content": "到2029年完成数据基础设施建设。",
                "page_type": "concept",
                "source_refs": ["doc-one"],
            },
            {
                "slug": "manual",
                "title": "人工维护页",
                "summary": "管理员手工保存",
                "content": "与重点无关，但不得由重点约束删除。",
                "page_type": "article",
            },
        ]
    )
    classified_content: list[str] = []

    async def classify(
        items: list[dict[str, Any]],
        _: str,
        __: list[str],
    ) -> dict[str, dict[str, Any]]:
        assert [item["item_id"] for item in items] == ["page:generated"]
        content = str(items[0]["content_excerpt"])
        classified_content.append(content)
        return {
            "page:generated": {
                "keep": "2029" in content,
                "directory": "",
            }
        }

    mapping = {
        "weknora_id": "kb-one",
        "wiki_directory_paths": [],
        "wiki_focus_instructions": "只保留2029年建设目标",
    }
    first = await reconcile_fixed_parent_directories(
        client,
        mapping,
        focus_classifier=classify,
    )
    assert first["deleted_pages"] == 0
    assert classified_content == ["到2029年完成数据基础设施建设。"]
    assert [call for call in client.calls if call[0] == "get_page"] == [
        ("get_page", "kb-one", "generated")
    ]

    generated = next(page for page in client.pages if page["slug"] == "generated")
    generated["content"] = "正文现已改为石窟寺保护背景材料。"
    second = await reconcile_fixed_parent_directories(
        client,
        mapping,
        focus_classifier=classify,
    )

    assert second["deleted_pages"] == 1
    assert classified_content == [
        "到2029年完成数据基础设施建设。",
        "正文现已改为石窟寺保护背景材料。",
    ]
    assert [page["slug"] for page in client.pages] == ["manual"]
    assert [call for call in client.calls if call[0] == "get_page"] == [
        ("get_page", "kb-one", "generated"),
        ("get_page", "kb-one", "generated"),
    ]


@pytest.mark.asyncio
async def test_focus_page_details_use_bounded_concurrency() -> None:
    class SlowDetails(FakeWeKnora):
        in_flight = 0
        max_in_flight = 0

        async def get_wiki_page(
            self,
            knowledge_base_id: str,
            slug: str,
        ) -> dict[str, Any]:
            self.in_flight += 1
            self.max_in_flight = max(self.max_in_flight, self.in_flight)
            try:
                await asyncio.sleep(0.001)
                return await super().get_wiki_page(knowledge_base_id, slug)
            finally:
                self.in_flight -= 1

    client = SlowDetails(
        pages=[
            {
                "slug": f"generated-{index:02d}",
                "title": f"目标 {index}",
                "content": "2029年建设目标",
                "page_type": "concept",
                "source_refs": [f"doc-{index}"],
            }
            for index in range(12)
        ]
    )

    async def keep_all(
        items: list[dict[str, Any]],
        _: str,
        __: list[str],
    ) -> dict[str, dict[str, Any]]:
        return {
            str(item["item_id"]): {"keep": True, "directory": ""}
            for item in items
        }

    await reconcile_fixed_parent_directories(
        client,
        {
            "weknora_id": "kb-one",
            "wiki_directory_paths": [],
            "wiki_focus_instructions": "只保留建设目标",
        },
        focus_classifier=keep_all,
    )

    assert 1 < client.max_in_flight <= constraints._PAGE_DETAIL_CONCURRENCY


@pytest.mark.asyncio
async def test_incomplete_focus_decisions_do_not_delete_or_move_pages() -> None:
    client = FakeWeKnora(
        folders=[{"id": "root", "name": "目标", "parent_id": ""}],
        pages=[
            {
                "slug": "one",
                "title": "一",
                "folder_id": "",
                "source_refs": ["doc-one"],
            },
            {
                "slug": "two",
                "title": "二",
                "folder_id": "",
                "source_refs": ["doc-two"],
            },
        ],
    )

    async def incomplete(*_: Any) -> dict[str, dict[str, Any]]:
        return {"page:one": {"keep": False, "directory": "目标"}}

    with pytest.raises(RuntimeError, match="incomplete"):
        await reconcile_fixed_parent_directories(
            client,
            {
                "weknora_id": "kb-one",
                "wiki_directory_paths": [["目标"]],
                "wiki_focus_instructions": "硬约束",
            },
            focus_classifier=incomplete,
        )

    assert {page["slug"] for page in client.pages} == {"one", "two"}
    assert not [call for call in client.calls if call[0] in {"delete_page", "move_page"}]


@pytest.mark.asyncio
async def test_completed_focus_batches_are_cached_before_a_later_batch_fails(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    client = FakeWeKnora(
        folders=[{"id": "leaf", "name": "目标", "parent_id": ""}],
        pages=[
            {
                "slug": "one",
                "title": "2029目标",
                "folder_id": "",
                "source_refs": ["doc-one"],
            },
            {
                "slug": "two",
                "title": "无关噪声",
                "folder_id": "",
                "source_refs": ["doc-two"],
            },
        ],
    )
    monkeypatch.setattr(constraints, "_FOCUS_CLASSIFIER_BATCH_SIZE", 1)
    calls: list[list[str]] = []

    async def fail_second_batch(
        items: list[dict[str, Any]],
        *_: Any,
        **__: Any,
    ) -> dict[str, dict[str, Any]]:
        item_ids = [str(item["item_id"]) for item in items]
        calls.append(item_ids)
        if item_ids == ["page:one"]:
            return {
                "page:one": {"keep": True, "directory": "目标"}
            }
        raise RuntimeError("later batch failed")

    monkeypatch.setattr(
        constraints,
        "_classify_focus_pages",
        fail_second_batch,
    )
    mapping = {
        "weknora_id": "kb-one",
        "wiki_directory_paths": [["目标"]],
        "wiki_focus_instructions": "只保留2029目标",
    }

    with pytest.raises(RuntimeError, match="later batch failed"):
        await reconcile_fixed_parent_directories(client, mapping)

    assert calls == [["page:one"], ["page:two"]]
    assert (
        client.pages[0]["page_metadata"]["deerflow_generation_constraints"][
            "keep"
        ]
        is True
    )
    assert not [
        call for call in client.calls if call[0] in {"delete_page", "move_page"}
    ]

    calls.clear()

    async def finish_remaining(
        items: list[dict[str, Any]],
        *_: Any,
        **__: Any,
    ) -> dict[str, dict[str, Any]]:
        item_ids = [str(item["item_id"]) for item in items]
        calls.append(item_ids)
        return {"page:two": {"keep": False, "directory": ""}}

    monkeypatch.setattr(
        constraints,
        "_classify_focus_pages",
        finish_remaining,
    )
    result = await reconcile_fixed_parent_directories(client, mapping)

    assert calls == [["page:two"]]
    assert result["deleted_pages"] == 1
    assert [page["slug"] for page in client.pages] == ["one"]
    assert client.pages[0]["folder_id"] == ""


@pytest.mark.asyncio
async def test_legacy_multi_root_reference_does_not_invoke_classifier(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    client = FakeWeKnora(
        folders=[
            {"id": "root-a", "name": "甲", "parent_id": ""},
            {"id": "root-b", "name": "乙", "parent_id": ""},
            {"id": "extra", "name": "应归乙", "parent_id": ""},
            {"id": "fallback", "name": "未知", "parent_id": ""},
        ],
        pages=[{"slug": "page-one", "title": "乙类页面", "page_type": "summary", "folder_id": ""}],
    )
    batches: list[list[str]] = []

    async def fake_default_classifier(items: list[dict[str, Any]], roots: list[str]) -> dict[str, str]:
        assert roots == ["甲", "乙"]
        batches.append([str(item["item_id"]) for item in items])
        return {
            "folder:extra": "乙",
            "folder:fallback": "不存在的根",
            "page:page-one": "乙",
        }

    monkeypatch.setattr(constraints, "_default_classifier", fake_default_classifier)
    result = await reconcile_fixed_parent_directories(
        client,
        {"weknora_id": "kb-one", "wiki_parent_directories": ["甲", "乙"]},
    )

    assert result["moved_folders"] == 0
    assert result["moved_pages"] == 0
    assert batches == []
    assert next(item for item in client.folders if item["id"] == "extra")["parent_id"] == ""
    assert next(item for item in client.folders if item["id"] == "fallback")["parent_id"] == ""
    assert client.pages[0]["folder_id"] == ""


@pytest.mark.asyncio
async def test_weknora_client_folder_and_page_placement_contracts() -> None:
    seen: list[tuple[str, str]] = []

    async def handler(request: httpx.Request) -> httpx.Response:
        path = request.url.raw_path.decode().split("?", 1)[0]
        seen.append((request.method, path))
        if request.method == "GET" and path.endswith("/wiki/folders"):
            assert request.url.params["parent_id"] == "parent-one"
            assert request.url.params["page_types"] == "entity,concept"
            return httpx.Response(200, json={"data": [{"id": "child", "name": "Child"}]})
        if request.method == "PUT" and path.endswith("/wiki/folders/folder%2Fone"):
            assert json.loads(request.content) == {"parent_id": "root-one", "move_parent": True}
            return httpx.Response(200, json={"data": {"id": "folder/one", "parent_id": "root-one"}})
        if request.method == "DELETE" and path.endswith("/wiki/folders/folder%2Fone"):
            return httpx.Response(204)
        if request.method == "PUT" and path.endswith("/wiki/move-page"):
            assert json.loads(request.content) == {"slug": "entity/acme", "folder_id": "root-one"}
            return httpx.Response(200, json={"data": {"slug": "entity/acme", "folder_id": "root-one"}})
        if request.method == "GET" and path.endswith("/wiki/pages"):
            if request.url.params.get("query") == "unfiltered":
                assert "folder_id" not in request.url.params
            else:
                assert "folder_id" in request.url.params
                assert request.url.params["folder_id"] == ""
            return httpx.Response(200, json={"pages": [], "total": 0, "page": 1, "page_size": 50})
        return httpx.Response(404)

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client:
        client = WeKnoraClient(base_url="http://weknora.local:8080", http_client=http_client)
        folders = await client.list_wiki_folders("kb-one", parent_id="parent-one", page_types=["entity", "concept"])
        updated = await client.update_wiki_folder("kb-one", "folder/one", {"parent_id": "root-one"})
        await client.delete_wiki_folder("kb-one", "folder/one")
        moved = await client.move_wiki_page("kb-one", "entity/acme", "root-one")
        await client.list_wiki_pages("kb-one", folder_id="")
        await client.list_wiki_pages("kb-one", query="unfiltered")

    assert folders["folders"] == [{"id": "child", "name": "Child"}]
    assert updated["parent_id"] == "root-one"
    assert moved["folder_id"] == "root-one"
    assert seen == [
        ("GET", "/api/v1/knowledgebase/kb-one/wiki/folders"),
        ("PUT", "/api/v1/knowledgebase/kb-one/wiki/folders/folder%2Fone"),
        ("DELETE", "/api/v1/knowledgebase/kb-one/wiki/folders/folder%2Fone"),
        ("PUT", "/api/v1/knowledgebase/kb-one/wiki/move-page"),
        ("GET", "/api/v1/knowledgebase/kb-one/wiki/pages"),
        ("GET", "/api/v1/knowledgebase/kb-one/wiki/pages"),
    ]


@pytest.mark.asyncio
async def test_weknora_client_preserves_folder_conflict_status() -> None:
    async def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(
            409,
            json={"message": "wiki folder name conflict"},
        )

    async with httpx.AsyncClient(
        transport=httpx.MockTransport(handler)
    ) as http_client:
        client = WeKnoraClient(
            base_url="http://weknora.local:8080",
            http_client=http_client,
        )
        with pytest.raises(Exception) as caught:
            await client.create_wiki_folder(
                "kb-one",
                name="国内新闻",
                parent_id="",
            )

    assert getattr(caught.value, "status_code", None) == 409