from __future__ import annotations import asyncio import json import threading from types import SimpleNamespace import pytest from fastapi import FastAPI, HTTPException from langchain_core.messages import HumanMessage from starlette.requests import Request from app.gateway.services import ( _with_knowledge_retrieval_progress_stream, enforce_knowledge_retrieval_mode_run_config, resolve_knowledge_retrieval_mode, ) from deerflow.integrations.weknora.client import WeKnoraClient, WeKnoraError from deerflow.tools.builtins import llmwiki_reasoning_tools as tools class _SettingsStore: def __init__(self, enabled: bool) -> None: self.enabled = enabled async def get(self) -> bool: return self.enabled class _MappingStore: async def get_authorized( self, mapping_id: str, user_id: str, *, write: bool, is_admin: bool, ) -> dict | None: assert user_id == "user-1" assert write is False if mapping_id != "mapping-1": return None return { "id": "mapping-1", "weknora_id": "remote-1", "name": "制度库", } class _Client: async def search(self, query: str, knowledge_base_ids: list[str]) -> list[dict]: assert knowledge_base_ids == ["remote-1"] return [ { "chunk_id": "chunk-1", "knowledge_id": "doc-1", "knowledge_base_id": "remote-1", "title": "采购制度", "content": "采购金额超过十万元时需要集体审议。", "score": 0.8 if query == "采购金额" else 0.9, } ] async def list_documents(self, knowledge_base_id: str, *, page: int, page_size: int) -> dict: assert knowledge_base_id == "remote-1" return { "items": [{"id": "doc-1", "title": "采购制度", "updated_at": "2026-01-01"}], "total": 1, } async def list_chunks(self, knowledge_id: str, *, page: int, page_size: int) -> dict: assert knowledge_id == "doc-1" return { "items": [ {"id": "chunk-1", "content": "条款 A-17：采购金额超过十万元。", "chunk_index": 2}, {"id": "chunk-2", "content": "普通说明。", "chunk_index": 3}, ], "total": 2, } async def grep_chunks(self, query: str, knowledge_base_ids: list[str]) -> dict: assert query == r"A-17|十万元" assert knowledge_base_ids == ["remote-1"] return { "items": [ { "chunk_id": "chunk-1", "knowledge_id": "doc-1", "knowledge_base_id": "remote-1", "title": "采购制度", "content": "条款 A-17：采购金额超过十万元。", "match_count": 2, "chunk_index": 2, } ], "contract": "weknora-v0.7.2-grep-v1", "compatibility_fallback": False, "partial": False, "warnings": [], "successful_knowledge_base_ids": ["remote-1"], "failed_knowledge_base_ids": [], "scanned_chunks": 2, "scan_truncated": False, } def _request(store: _SettingsStore | None) -> Request: app = FastAPI() if store is not None: app.state.knowledge_retrieval_settings_store = store return Request( { "type": "http", "method": "POST", "path": "/api/threads/t/runs", "headers": [], "query_string": b"", "app": app, } ) @pytest.mark.asyncio async def test_database_gate_defaults_classic_and_rejects_disabled_smart_mode() -> None: assert await resolve_knowledge_retrieval_mode( _request(None), SimpleNamespace(context={}), ) == "classic" with pytest.raises(HTTPException) as caught: await resolve_knowledge_retrieval_mode( _request(_SettingsStore(False)), SimpleNamespace(context={"knowledge_retrieval_mode": "smart_reasoning"}), ) assert caught.value.status_code == 409 assert caught.value.detail["code"] == "SMART_REASONING_DISABLED" assert await resolve_knowledge_retrieval_mode( _request(_SettingsStore(True)), SimpleNamespace(context={"knowledge_retrieval_mode": "smart_reasoning"}), ) == "smart_reasoning" def test_validated_mode_overwrites_untrusted_preseeded_config() -> None: config = { "configurable": { "knowledge_retrieval_mode": "smart_reasoning", "disable_tools": True, "excluded_tools": ["ordinary_wiki_search"], }, "context": { "knowledge_retrieval_mode": "smart_reasoning", "disable_tools": True, "excluded_tools": ["ordinary_knowledge_search"], }, "metadata": {"rag_mode_enabled": True, "rag_mode": True}, } enforce_knowledge_retrieval_mode_run_config(config, "classic") assert config["configurable"]["knowledge_retrieval_mode"] == "classic" assert config["context"]["knowledge_retrieval_mode"] == "classic" enforce_knowledge_retrieval_mode_run_config(config, "smart_reasoning") assert config["configurable"]["rag_mode_enabled"] is False assert config["context"]["rag_mode_enabled"] is False assert config["configurable"]["knowledge_rag_enabled"] is False assert config["context"]["knowledge_rag_enabled"] is False assert config["metadata"]["rag_mode_enabled"] is False assert config["metadata"]["rag_mode"] is False assert config["configurable"]["disable_tools"] is False assert config["context"]["disable_tools"] is False assert set(config["configurable"]["excluded_tools"]) == set( tools.CLASSIC_LLMWIKI_TOOL_NAMES ) assert set(config["context"]["excluded_tools"]) == set( tools.CLASSIC_LLMWIKI_TOOL_NAMES ) def test_smart_mode_always_enables_custom_progress_stream() -> None: assert _with_knowledge_retrieval_progress_stream( ["values", "messages-tuple"], "smart_reasoning", ) == ["values", "messages-tuple", "custom"] assert _with_knowledge_retrieval_progress_stream( ["values", "custom", "custom"], "smart_reasoning", ) == ["values", "custom"] assert _with_knowledge_retrieval_progress_stream( ["values"], "classic", ) == ["values"] def test_numbered_citation_prompt_is_classic_only() -> None: from app.gateway import services graph_input = {"messages": [HumanMessage(content="查制度")]} smart = services._apply_classic_rag_citation_prompt( graph_input, enabled=True, knowledge_retrieval_mode="smart_reasoning", ) smart_extra = smart["messages"][0].additional_kwargs assert services._RAG_CITATION_PROMPT not in str( smart_extra.get("prompt_prefix") or "" ) classic = services._apply_classic_rag_citation_prompt( graph_input, enabled=True, knowledge_retrieval_mode="classic", ) assert services._RAG_CITATION_PROMPT in str( classic["messages"][0].additional_kwargs.get("prompt_prefix") or "" ) @pytest.mark.asyncio @pytest.mark.parametrize("assistant_id", [None, "deep-research-planner", "deep-research-collector"]) async def test_smart_scope_binds_this_library_to_authorized_database_metadata( monkeypatch: pytest.MonkeyPatch, assistant_id: str | None, ) -> None: from app.gateway import llmwiki_rag class ScopeStore: async def get_authorized( self, mapping_id: str, user_id: str, *, write: bool, is_admin: bool, ) -> dict | None: assert (mapping_id, user_id, write, is_admin) == ( "mapping-news", "user-1", False, False, ) return { "id": mapping_id, "weknora_id": "remote-news", "name": "国内新闻资料库", "description": "收录国内重要新闻和政策发布。", "wiki_focus_instructions": "突出时间、地点和关键数据", "wiki_directory_paths": [["年度", "重大活动"], ["政策", "发布会"]], } app = FastAPI() app.state.config = SimpleNamespace() app.state.llmwiki_store = ScopeStore() request = Request( { "type": "http", "method": "POST", "path": "/api/threads/t/runs", "headers": [], "query_string": b"", "app": app, } ) monkeypatch.setattr( llmwiki_rag, "get_resolved_llmwiki_runtime", lambda _config: SimpleNamespace(weknora_enabled=True), ) async def actor(_request: Request) -> tuple[str, bool]: return "user-1", False monkeypatch.setattr(llmwiki_rag, "_actor", actor) rag = await llmwiki_rag.build_llmwiki_smart_reasoning_context( request=request, body=SimpleNamespace( context={"llmwiki_knowledge_base_ids": ["mapping-news"]}, metadata={}, assistant_id=assistant_id, ), graph_input={"messages": [HumanMessage(content="这个库讲了什么")]}, ) assert rag.knowledge_base_context == [ { "id": "mapping-news", "name": "国内新闻资料库", "description": "收录国内重要新闻和政策发布。", "focus": "突出时间、地点和关键数据", "directory_hints": ["年度 / 重大活动", "政策 / 发布会"], } ] assert "这个库" in rag.prompt assert "国内新闻资料库" in rag.prompt assert "ordinary_knowledge_search" in rag.prompt assert "/mnt/knowledge" in rag.prompt assert "不要追问" in rag.prompt attached = llmwiki_rag.attach_llmwiki_rag_to_graph_input( {"messages": [HumanMessage(content="这个库讲了什么")]}, rag, ) extra = attached["messages"][0].additional_kwargs assert extra["llmwiki_knowledge_base_context"] == rag.knowledge_base_context assert "这个库" in extra["smart_knowledge_scope_prompt"] assert "prompt_prefix" not in extra def test_smart_scope_keeps_all_identities_under_one_global_prompt_budget() -> None: from app.gateway import llmwiki_rag rows = [ { "id": f"mapping-{index:03d}", "name": f"第{index:03d}个知识库" + ("长名称" * 40), "description": "详细简介" * 300, "wiki_focus_instructions": "抽取重点" * 300, "wiki_directory_paths": [ [f"一级目录{index}", "二级目录" * 80] for _ in range(12) ], } for index in range(100) ] contexts = llmwiki_rag._smart_knowledge_base_context(rows) serialized = json.dumps(contexts, ensure_ascii=False, separators=(",", ":")) prompt = llmwiki_rag._build_smart_scope_prompt(contexts) assert len(contexts) == 100 assert [item["id"] for item in contexts] == [ f"mapping-{index:03d}" for index in range(100) ] assert all(item["name"] for item in contexts) assert len(serialized) <= llmwiki_rag._MAX_SCOPE_CONTEXT_JSON_CHARS assert len(prompt) <= llmwiki_rag._MAX_SMART_SCOPE_PROMPT_CHARS assert any(item.get("description") for item in contexts) assert any("description" not in item for item in contexts) def test_smart_metadata_disables_reference_mode_while_classic_keeps_it() -> None: from app.gateway import services smart_run = {"rag_mode_enabled": True, "rag_mode": True} smart_record = {"rag_mode_enabled": True, "rag_mode": True} smart_thread = {"rag_mode_enabled": True, "rag_mode": True} smart = services._sync_retrieval_rag_metadata( run_metadata=smart_run, record_metadata=smart_record, thread_metadata_update=smart_thread, knowledge_retrieval_mode="smart_reasoning", knowledge_scope_active=True, selection_declared=True, explicit_rag_requested=True, ) assert smart == {"rag_mode_enabled": False, "rag_mode": False} assert smart_run == smart assert smart_record == smart assert smart_thread == smart classic_run: dict = {} classic_record: dict = {} classic_thread: dict = {} classic = services._sync_retrieval_rag_metadata( run_metadata=classic_run, record_metadata=classic_record, thread_metadata_update=classic_thread, knowledge_retrieval_mode="classic", knowledge_scope_active=True, selection_declared=True, explicit_rag_requested=False, ) assert classic == {"rag_mode_enabled": True, "rag_mode": True} assert classic_run == classic assert classic_record == classic assert classic_thread == classic @pytest.mark.asyncio async def test_smart_run_stops_before_launch_when_mode_metadata_cannot_persist( monkeypatch: pytest.MonkeyPatch, ) -> None: from app.gateway import analysis_projects, services from app.gateway.llmwiki_rag import LlmWikiRagContext class ThreadStore: def __init__(self) -> None: self.statuses: list[str] = [] self.metadata_writes = 0 async def get(self, _thread_id: str) -> dict: return {"metadata": {}, "assistant_id": None} async def update_status(self, _thread_id: str, status: str) -> None: self.statuses.append(status) async def update_metadata(self, _thread_id: str, _metadata: dict) -> None: self.metadata_writes += 1 raise RuntimeError("database unavailable") record = SimpleNamespace(run_id="run-1", metadata={}, task=None) class RunManager: def __init__(self) -> None: self.terminal_status: object | None = None async def create_or_reject(self, *_args, **kwargs): record.metadata = dict(kwargs["metadata"]) return record async def set_status(self, _run_id, status, **_kwargs) -> None: self.terminal_status = status thread_store = ThreadStore() run_manager = RunManager() run_context = SimpleNamespace(app_config=object(), thread_store=thread_store) async def prepare_project_run(body, _thread_id, _request): return body, None async def unexpected_fixed_answer_lookup(**_kwargs): raise AssertionError("smart retrieval must bypass fixed answers") async def smart_context(**_kwargs): return LlmWikiRagContext(["kb-1"], [], 0, "", [], "") launched = False async def launch(**_kwargs): nonlocal launched launched = True monkeypatch.setattr(analysis_projects, "prepare_project_run", prepare_project_run) monkeypatch.setattr(services, "get_stream_bridge", lambda _request: object()) monkeypatch.setattr(services, "get_run_manager", lambda _request: run_manager) monkeypatch.setattr(services, "get_run_context", lambda _request: run_context) monkeypatch.setattr(services, "get_concurrency_gate", lambda _request: None) monkeypatch.setattr(services, "get_effective_user_id", lambda: "user-1") monkeypatch.setattr( services, "_find_fixed_answer", unexpected_fixed_answer_lookup, ) monkeypatch.setattr(services, "_resolve_effective_model_name", lambda *_args, **_kwargs: "model-1") monkeypatch.setattr(services, "_get_model_concurrency_limit", lambda _name: None) monkeypatch.setattr(services, "_get_user_model_concurrency_limit", lambda _name: None) monkeypatch.setattr(services, "resolve_agent_factory", lambda _assistant_id: object()) monkeypatch.setattr(services, "build_llmwiki_smart_reasoning_context", smart_context) monkeypatch.setattr(services, "_run_agent_with_concurrency_lease", launch) body = SimpleNamespace( input={"messages": [{"role": "user", "content": "查制度"}]}, config={}, context={"knowledge_retrieval_mode": "smart_reasoning"}, metadata={}, assistant_id=None, on_disconnect="continue", multitask_strategy="reject", ) request = SimpleNamespace( app=SimpleNamespace( state=SimpleNamespace( knowledge_retrieval_settings_store=_SettingsStore(True), ) ), client=None, headers={}, ) with pytest.raises(RuntimeError, match="database unavailable"): await services.start_run(body, "thread-1", request, enforce_access=False) assert thread_store.metadata_writes == 1 assert thread_store.statuses == ["running", "error"] assert str(run_manager.terminal_status).endswith("error") assert launched is False @pytest.mark.asyncio async def test_multi_query_search_merges_same_chunk_and_preserves_query_evidence(monkeypatch) -> None: monkeypatch.setattr(tools, "get_llmwiki_store", lambda: _MappingStore()) monkeypatch.setattr(tools, "get_effective_user_id", lambda: "user-1") monkeypatch.setattr(tools, "get_current_user", lambda: SimpleNamespace(system_role="user")) monkeypatch.setattr(tools, "_client", lambda: _Client()) payload = json.loads( await tools._knowledge_search_impl( ["采购金额", "十万元审议"], { "knowledge_retrieval_mode": "smart_reasoning", "llmwiki_knowledge_base_ids": ["mapping-1"], }, ) ) assert payload["ok"] is True assert payload["result_count"] == 1 assert payload["results"][0]["score"] == pytest.approx(0.9) assert payload["results"][0]["matched_queries"] == ["采购金额", "十万元审议"] assert payload["results"][0]["knowledge_base_id"] == "mapping-1" @pytest.mark.asyncio async def test_vector_tool_streams_public_query_and_aggregate_result_progress( monkeypatch, ) -> None: events: list[dict] = [] context = { "knowledge_retrieval_mode": "smart_reasoning", "llmwiki_knowledge_base_ids": ["mapping-1", "mapping-2"], } runtime = SimpleNamespace( context=context, stream_writer=events.append, tool_call_id="call-vector-1", ) async def fake_search(queries, received_context): assert queries == ["国内新闻", "政策发布"] assert received_context is context return tools._result( operation="knowledge_search", results=[ { "chunk_id": "chunk-secret", "knowledge_id": "doc-secret", "knowledge_base_id": "mapping-1", "knowledge_base": "国内新闻库", "title": "新闻原文", "content": "抽屉中显示的安全新闻正文预览", } ], queries=queries, partial=True, warnings=[{"code": "PROVIDER_QUERY_FAILED", "query_index": 2}], ) monkeypatch.setattr(tools, "_knowledge_search_impl", fake_search) clock = iter([10.0, 10.456]) monkeypatch.setattr(tools.time, "perf_counter", lambda: next(clock)) output = await tools.ordinary_knowledge_search_tool.coroutine( queries=["国内新闻", "政策发布"], runtime=runtime, ) output_payload = json.loads(output) assert output_payload["result_count"] == 1 assert output_payload["elapsed_ms"] == 456 assert [(event["stage"], event["status"]) for event in events] == [ ("query_rewrite", "completed"), ("vector_search", "started"), ("vector_search", "completed"), ] assert all( event["type"] == "knowledge_retrieval_progress" and event["version"] == 1 and event["retrieval_id"] == "call-vector-1" for event in events ) assert events[0]["queries"] == ["国内新闻", "政策发布"] assert events[1]["knowledge_base_count"] == 2 assert events[-1]["result_count"] == 1 assert events[-1]["partial"] is True assert events[-1]["elapsed_ms"] == 456 assert events[-1]["warning_codes"] == ["PROVIDER_QUERY_FAILED"] assert events[-1]["result_items"] == [ { "kind": "document_chunk", "knowledge_base_id": "mapping-1", "knowledge_base": "国内新闻库", "title": "新闻原文", "chunk_id": "chunk-secret", "knowledge_id": "doc-secret", "content": "抽屉中显示的安全新闻正文预览", "content_truncated": False, } ] assert events[-1]["result_items_truncated"] is False @pytest.mark.asyncio async def test_wiki_tool_streams_channel_coverage_and_scan_count(monkeypatch) -> None: events: list[dict] = [] context = { "knowledge_retrieval_mode": "smart_reasoning", "llmwiki_knowledge_base_ids": ["mapping-1"], } runtime = SimpleNamespace( context=context, stream_writer=events.append, tool_call_id="call-wiki-1", ) async def fake_wiki(pattern, received_context): assert pattern == "新闻|发布会" assert received_context is context return tools._result( operation="wiki_search", results=[ { "page_ref": tools._page_ref("mapping-1", "news/release"), "knowledge_base_id": "mapping-1", "knowledge_base": "国内新闻库", "title": "发布会", "slug": "news/release", "summary": "可在抽屉中展示的 Wiki 摘要", } ], pattern=pattern, channel="provider_native_regex", partial=False, native_regex_supported=True, unsupported_knowledge_base_count=0, coverage=[ { "knowledge_base": "国内新闻库", "channel": "provider_native_regex", "status": "searched", } ], scanned_pages=17, ) monkeypatch.setattr(tools, "_wiki_search_impl", fake_wiki) await tools.ordinary_wiki_search_tool.coroutine( pattern="新闻|发布会", runtime=runtime, ) assert [(event["stage"], event["status"]) for event in events] == [ ("query_rewrite", "completed"), ("wiki_regex_search", "started"), ("wiki_regex_search", "completed"), ] completed = events[-1] assert completed["pattern"] == "新闻|发布会" assert completed["channel"] == "provider_native_regex" assert completed["scanned_pages"] == 17 assert completed["result_count"] == 1 assert completed["native_regex_supported"] is True assert completed["unsupported_knowledge_base_count"] == 0 assert completed["coverage"] == [ { "knowledge_base": "国内新闻库", "channel": "provider_native_regex", "status": "searched", } ] assert completed["result_items"] == [ { "kind": "wiki_page", "knowledge_base_id": "mapping-1", "knowledge_base": "国内新闻库", "title": "发布会", "page_ref": tools._page_ref("mapping-1", "news/release"), "slug": "news/release", "summary": "可在抽屉中展示的 Wiki 摘要", } ] def test_public_result_items_are_scope_filtered_and_bounded() -> None: payload = { "operation": "knowledge_search", "results": [ { "knowledge_base_id": "mapping-1", "provider_knowledge_base_id": "remote-must-not-leak", "knowledge_base": "制度库", "title": f"文档 {index}", "chunk_id": f"chunk-{index}", "knowledge_id": f"doc-{index}", "content": "甲" * 2_000, } for index in range(22) ] + [ { "knowledge_base_id": "mapping-foreign", "title": "越权结果", "chunk_id": "foreign-chunk", "content": "不得进入事件", } ], } items, truncated = tools._public_result_items( payload, authorized_mapping_ids={"mapping-1"}, ) assert len(items) == tools._MAX_PROGRESS_RESULT_ITEMS assert truncated is True assert all(item["knowledge_base_id"] == "mapping-1" for item in items) assert all(len(item["content"]) <= tools._MAX_PROGRESS_RAW_CONTENT_CHARS for item in items) rendered = json.dumps(items, ensure_ascii=False) assert "remote-must-not-leak" not in rendered assert "mapping-foreign" not in rendered assert "foreign-chunk" not in rendered def test_oversized_nested_tool_metadata_still_returns_valid_json() -> None: output = tools._result( operation="wiki_browse_index", results=[{"title": "入口"}], groups=[{"type": "summary", "items": [{"summary": "甲" * 60_000}]}], ) payload = json.loads(output) assert payload == { "ok": True, "operation": "wiki_browse_index", "result_count": 0, "results": [], "output_truncated": True, } @pytest.mark.asyncio async def test_progress_failure_hides_raw_exception_and_keeps_tool_contract( monkeypatch, ) -> None: events: list[dict] = [] runtime = SimpleNamespace( context={ "knowledge_retrieval_mode": "smart_reasoning", "llmwiki_knowledge_base_ids": ["mapping-1"], }, stream_writer=events.append, tool_call_id="call-failed-1", ) async def failed_search(_queries, _context): raise RuntimeError("provider-token-must-never-leak") monkeypatch.setattr(tools, "_knowledge_search_impl", failed_search) clock = iter([20.0, 20.125]) monkeypatch.setattr(tools.time, "perf_counter", lambda: next(clock)) output = await tools.ordinary_knowledge_search_tool.coroutine( queries=["国内新闻"], runtime=runtime, ) payload = json.loads(output) assert payload["ok"] is False assert payload["error"]["code"] == "SMART_RETRIEVAL_FAILED" assert payload["elapsed_ms"] == 125 assert events[-1]["status"] == "failed" assert events[-1]["elapsed_ms"] == 125 assert events[-1]["error"] == payload["error"] assert "provider-token-must-never-leak" not in json.dumps( events, ensure_ascii=False, ) @pytest.mark.asyncio async def test_chunk_regex_search_is_case_insensitive_and_bounded(monkeypatch) -> None: monkeypatch.setattr(tools, "get_llmwiki_store", lambda: _MappingStore()) monkeypatch.setattr(tools, "get_effective_user_id", lambda: "user-1") monkeypatch.setattr(tools, "get_current_user", lambda: SimpleNamespace(system_role="user")) monkeypatch.setattr(tools, "_client", lambda: _Client()) payload = json.loads( await tools._chunk_regex_impl( r"A-17|十万元", { "knowledge_retrieval_mode": "smart_reasoning", "llmwiki_knowledge_base_ids": ["mapping-1"], }, ) ) assert payload["result_count"] == 1 assert payload["results"][0]["chunk_id"] == "chunk-1" assert payload["scanned_chunks"] == 2 def test_wiki_page_reference_round_trip_and_regex_validation() -> None: reference = tools._page_ref("mapping-1", "制度/采购") assert tools._decode_page_ref(reference) == ("mapping-1", "制度/采购") assert tools._safe_regex("采购|A-17").search("a-17") with pytest.raises(tools.SmartKnowledgeToolError): tools._safe_regex("(a+)+") class _MultiMappingStore: async def get_authorized( self, mapping_id: str, user_id: str, *, write: bool, is_admin: bool, ) -> dict | None: assert user_id == "user-1" assert write is False if mapping_id not in {"mapping-1", "mapping-2"}: return None suffix = mapping_id.rsplit("-", 1)[-1] return { "id": mapping_id, "weknora_id": f"remote-{suffix}", "name": f"制度库{suffix}", } def _patch_scope(monkeypatch, *, multi: bool = False) -> None: monkeypatch.setattr( tools, "get_llmwiki_store", lambda: _MultiMappingStore() if multi else _MappingStore(), ) monkeypatch.setattr(tools, "get_effective_user_id", lambda: "user-1") monkeypatch.setattr( tools, "get_current_user", lambda: SimpleNamespace(system_role="user"), ) @pytest.mark.asyncio async def test_classic_wiki_index_is_scope_bound_drawer_ready_and_bounded( monkeypatch, ) -> None: _patch_scope(monkeypatch) class Client: async def get_wiki_index( self, knowledge_base_id: str, *, types: list[str], limit: int, cursor: str, ) -> dict: assert knowledge_base_id == "remote-1" assert types == ["summary", "entity"] assert limit == 100 assert cursor == "" return { "intro": "入口见 [[news/timeline|新闻时间线]]。", "groups": [ { "type": "summary", "total": 60, "next_cursor": "next-summary", "items": [ { "id": f"page-{index}", "slug": f"news/page-{index}", "title": f"新闻页面 {index}", "summary": "摘要" * 1_000, "category_path": ["新闻", "国内"], } for index in range(60) ], } ], } monkeypatch.setattr(tools, "_client", Client) payload = json.loads( await tools._wiki_index_impl( "mapping-1", { "knowledge_retrieval_mode": "classic", "llmwiki_knowledge_base_ids": ["mapping-1"], }, types=["summary", "entity"], limit=100, ) ) assert payload["ok"] is True assert payload["operation"] == "wiki_browse_index" assert payload["results_truncated"] is True assert len(payload["groups"][0]["items"]) == tools._MAX_CLASSIC_WIKI_INDEX_ITEMS assert payload["groups"][0]["items_truncated"] is True assert payload["groups"][0]["next_cursor"] == "next-summary" assert "/weknora-source-preview?" in payload["results"][0]["content"] assert "wiki_slug=news%2Ftimeline" in payload["results"][0]["content"] assert len(json.dumps(payload, ensure_ascii=False)) < tools._MAX_TOOL_OUTPUT_CHARS with pytest.raises(tools.SmartKnowledgeToolError) as caught: await tools._wiki_index_impl( "mapping-1", { "knowledge_retrieval_mode": "smart_reasoning", "llmwiki_knowledge_base_ids": ["mapping-1"], }, ) assert caught.value.code == "CLASSIC_MODE_REQUIRED" @pytest.mark.asyncio async def test_classic_wiki_page_read_returns_source_drawer_metadata(monkeypatch) -> None: _patch_scope(monkeypatch) class MissingLocalMirror: async def list_index_status(self, _mapping_ids: list[str]) -> list[dict]: raise RuntimeError("local index disabled") class Client: async def get_wiki_page(self, knowledge_base_id: str, slug: str) -> dict: assert (knowledge_base_id, slug) == ("remote-1", "news/timeline") return { "slug": slug, "title": "新闻时间线", "description": "按日期整理", "markdown": "火炬传递、开幕与闭幕时间线。", "status": "published", } monkeypatch.setattr(tools, "get_local_wiki_index_store", MissingLocalMirror) monkeypatch.setattr(tools, "_client", Client) payload = json.loads( await tools._wiki_read_impl( tools._page_ref("mapping-1", "news/timeline"), { "knowledge_retrieval_mode": "classic", "llmwiki_knowledge_base_ids": ["mapping-1"], }, required_mode="classic", ) ) page = payload["results"][0] assert page["knowledge_base_id"] == "mapping-1" assert page["knowledge_base_name"] == "制度库" assert page["wiki_slug"] == "news/timeline" assert page["type"] == "llmwiki_wiki_page" assert page["content"] == "火炬传递、开幕与闭幕时间线。" assert page["url"].endswith("knowledge_base_id=mapping-1&wiki_slug=news%2Ftimeline") @pytest.mark.asyncio async def test_authorization_read_is_routed_to_the_engine_event_loop( monkeypatch, ) -> None: home_loop = asyncio.new_event_loop() ready = threading.Event() def run_home_loop() -> None: asyncio.set_event_loop(home_loop) ready.set() home_loop.run_forever() home_loop.close() thread = threading.Thread(target=run_home_loop, daemon=True) thread.start() assert ready.wait(timeout=2) observed_loops: list[asyncio.AbstractEventLoop] = [] class Store: async def get_authorized( self, mapping_id: str, user_id: str, *, write: bool, is_admin: bool, ) -> dict: observed_loops.append(asyncio.get_running_loop()) return {"id": mapping_id, "owner_user_id": user_id} monkeypatch.setattr(tools, "get_engine_loop", lambda: home_loop) try: row = await tools._get_authorized_on_engine_loop( Store(), "mapping-1", "user-1", is_admin=False, ) finally: home_loop.call_soon_threadsafe(home_loop.stop) thread.join(timeout=2) assert row == {"id": "mapping-1", "owner_user_id": "user-1"} assert observed_loops == [home_loop] @pytest.mark.asyncio async def test_search_all_failures_are_errors_and_partial_failure_is_structured( monkeypatch, ) -> None: _patch_scope(monkeypatch) class Client: async def search(self, query: str, knowledge_base_ids: list[str]) -> list[dict]: if query == "失败查询": raise WeKnoraError("down") return [ { "chunk_id": "chunk-ok", "knowledge_id": "doc-ok", "knowledge_base_id": "remote-1", "content": "可用证据", "score": 1, } ] monkeypatch.setattr(tools, "_client", Client) failed = json.loads( await tools._knowledge_search_impl( ["失败查询"], { "knowledge_retrieval_mode": "smart_reasoning", "llmwiki_knowledge_base_ids": ["mapping-1"], }, ) ) assert failed["ok"] is False assert failed["error"]["code"] == "PROVIDER_UNAVAILABLE" partial = json.loads( await tools._knowledge_search_impl( ["成功查询", "失败查询"], { "knowledge_retrieval_mode": "smart_reasoning", "llmwiki_knowledge_base_ids": ["mapping-1"], }, ) ) assert partial["ok"] is True assert partial["partial"] is True assert partial["warnings"] == [ {"code": "PROVIDER_QUERY_FAILED", "query_index": 2} ] @pytest.mark.asyncio async def test_wiki_search_uses_native_provider_for_each_mapping_concurrently( monkeypatch, ) -> None: _patch_scope(monkeypatch, multi=True) class LocalIndexMustNotRun: def __init__(self) -> None: raise AssertionError("smart Wiki regex must use the native provider API") class Client: calls: list[tuple[str, str, int]] = [] active = 0 peak = 0 async def search_wiki_pages( self, knowledge_base_id: str, query: str, *, limit: int, ) -> list[dict]: self.calls.append((knowledge_base_id, query, limit)) self.active += 1 self.peak = max(self.peak, self.active) await asyncio.sleep(0.01) self.active -= 1 suffix = knowledge_base_id.rsplit("-", 1)[-1] return [ { "slug": f"news/{suffix}", "title": f"needle 页面 {suffix}", "content": "needle 正文", "status": "published", } ] async def list_wiki_pages(self, *args, **kwargs) -> dict: raise AssertionError("Wiki directory enumeration is forbidden") async def get_wiki_page(self, *args, **kwargs) -> dict: raise AssertionError("search must not fetch each page body") client = Client() monkeypatch.setattr(tools, "get_local_wiki_index_store", LocalIndexMustNotRun) monkeypatch.setattr(tools, "_client", lambda: client) payload = json.loads( await tools._wiki_search_impl( "needle", { "knowledge_retrieval_mode": "smart_reasoning", "llmwiki_knowledge_base_ids": ["mapping-1", "mapping-2"], }, ) ) assert payload["ok"] is True assert payload["native_regex_supported"] is True assert payload["result_count"] == 2 assert {call[:2] for call in client.calls} == { ("remote-1", "needle"), ("remote-2", "needle"), } assert client.peak == 2 assert payload["channel"] == "provider_native_regex" assert [item["status"] for item in payload["coverage"]] == [ "searched", "searched", ] @pytest.mark.asyncio async def test_wiki_search_marks_provider_limit_as_truncated(monkeypatch) -> None: _patch_scope(monkeypatch) class Client: async def search_wiki_pages( self, knowledge_base_id: str, query: str, *, limit: int, ) -> list[dict]: assert (knowledge_base_id, query, limit) == ("remote-1", "needle", 50) return [ { "slug": f"news/{index}", "title": f"needle 页面 {index}", "content": "needle 正文", } for index in range(limit) ] monkeypatch.setattr(tools, "_client", Client) payload = json.loads( await tools._wiki_search_impl( "needle", { "knowledge_retrieval_mode": "smart_reasoning", "llmwiki_knowledge_base_ids": ["mapping-1"], }, ) ) assert payload["scanned_pages"] == 50 assert payload["scan_truncated"] is True assert payload["result_count"] == tools._MAX_RESULTS @pytest.mark.asyncio async def test_wiki_search_handles_supported_and_unsupported_mappings( monkeypatch, ) -> None: _patch_scope(monkeypatch, multi=True) class Client: async def search_wiki_pages( self, knowledge_base_id: str, query: str, *, limit: int, ) -> list[dict]: if knowledge_base_id == "remote-2": raise WeKnoraError("route unavailable", status_code=404) return [{"slug": "news/one", "title": f"{query} 页面"}] monkeypatch.setattr(tools, "_client", Client) payload = json.loads( await tools._wiki_search_impl( "needle", { "knowledge_retrieval_mode": "smart_reasoning", "llmwiki_knowledge_base_ids": ["mapping-1", "mapping-2"], }, ) ) assert payload["ok"] is True assert payload["native_regex_supported"] is True assert payload["unsupported_knowledge_base_count"] == 1 assert payload["partial"] is False assert payload["warnings"] == [] assert [item["status"] for item in payload["coverage"]] == [ "searched", "unsupported", ] @pytest.mark.asyncio async def test_wiki_search_without_native_client_method_is_unsupported( monkeypatch, ) -> None: _patch_scope(monkeypatch) class LegacyClient: async def list_wiki_pages(self, *args, **kwargs) -> dict: raise AssertionError("legacy directory fallback must not run") async def get_wiki_page(self, *args, **kwargs) -> dict: raise AssertionError("legacy page fallback must not run") monkeypatch.setattr(tools, "_client", LegacyClient) payload = json.loads( await tools._wiki_search_impl( "needle", { "knowledge_retrieval_mode": "smart_reasoning", "llmwiki_knowledge_base_ids": ["mapping-1"], }, ) ) assert payload["ok"] is True assert payload["native_regex_supported"] is False assert payload["unsupported_knowledge_base_count"] == 1 assert payload["results"] == [] assert payload["warnings"] == [] @pytest.mark.asyncio @pytest.mark.parametrize("status_code", [404, 405]) async def test_wiki_search_skips_unsupported_native_regex_without_fallback( monkeypatch, status_code: int, ) -> None: _patch_scope(monkeypatch) class Client: async def search_wiki_pages(self, *args, **kwargs) -> list[dict]: raise WeKnoraError("route unavailable", status_code=status_code) async def list_wiki_pages(self, *args, **kwargs) -> dict: raise AssertionError("unsupported native search must not enumerate Wiki pages") async def get_wiki_page(self, *args, **kwargs) -> dict: raise AssertionError("unsupported native search must not read page bodies") monkeypatch.setattr(tools, "_client", Client) payload = json.loads( await tools._wiki_search_impl( "星河工程", { "knowledge_retrieval_mode": "smart_reasoning", "llmwiki_knowledge_base_ids": ["mapping-1"], }, ) ) assert payload["ok"] is True assert payload["result_count"] == 0 assert payload["native_regex_supported"] is False assert payload["unsupported_knowledge_base_count"] == 1 assert payload["partial"] is False assert payload["warnings"] == [] assert payload["coverage"][0]["status"] == "unsupported" @pytest.mark.asyncio async def test_native_wiki_search_result_can_be_read_with_markdown_shape( monkeypatch, ) -> None: _patch_scope(monkeypatch) class EmptyMirror: async def list_index_status(self, mapping_ids: list[str]) -> list[dict]: return [] class Client: search_calls: list[tuple[str, str, int]] = [] async def search_wiki_pages( self, knowledge_base_id: str, query: str, *, limit: int, ) -> list[dict]: self.search_calls.append((knowledge_base_id, query, limit)) return [ { "slug": "news/xinghe", "title": "国内科技新闻", "description": "科技报道汇总", "markdown": "星河工程完成关键节点验收。", "status": "published", } ] async def list_wiki_pages(self, *args, **kwargs) -> dict: raise AssertionError("native search must not enumerate the directory") async def get_wiki_page(self, knowledge_base_id: str, slug: str) -> dict: assert (knowledge_base_id, slug) == ("remote-1", "news/xinghe") return { "slug": slug, "title": "国内科技新闻", "description": "科技报道汇总", "markdown": "星河工程完成关键节点验收。参见 [[news/验收公告]]。", "out_links": ["news/承建单位"], "status": "published", } client = Client() monkeypatch.setattr(tools, "get_local_wiki_index_store", EmptyMirror) monkeypatch.setattr(tools, "_client", lambda: client) payload = json.loads( await tools._wiki_search_impl( "星河工程", { "knowledge_retrieval_mode": "smart_reasoning", "llmwiki_knowledge_base_ids": ["mapping-1"], }, ) ) assert payload["results"][0]["matched_fields"] == ["content"] assert client.search_calls == [("remote-1", "星河工程", 50)] read = json.loads( await tools._wiki_read_impl( payload["results"][0]["page_ref"], { "knowledge_retrieval_mode": "smart_reasoning", "llmwiki_knowledge_base_ids": ["mapping-1"], }, ) ) assert read["results"][0]["content_markdown"] == "星河工程完成关键节点验收。参见 [[news/验收公告]]。" assert read["results"][0]["summary"] == "科技报道汇总" assert read["results"][0]["links"] == ["news/承建单位", "news/验收公告"] @pytest.mark.asyncio async def test_local_wiki_read_exposes_bounded_graph_links_for_adaptive_expansion( monkeypatch, ) -> None: _patch_scope(monkeypatch) class LocalIndex: async def list_index_status(self, mapping_ids: list[str]) -> list[dict]: assert mapping_ids == ["mapping-1"] return [ { "knowledge_base_mapping_id": "mapping-1", "state": "idle", "last_completed_at": "2026-10-14T00:00:00Z", "remote_page_count": 2, "local_page_count": 2, "failed_page_count": 0, "last_error": None, "last_sync_result": {"status": "completed"}, } ] async def get_page(self, mapping_id: str, slug: str) -> dict: assert (mapping_id, slug) == ("mapping-1", "news/xinghe") return { "slug": slug, "title": "星河工程建设进展", "aliases": ["星河计划"], "content_md": "已完成节点验收。", "status": "published", } async def get_adjacent_pages( self, mapping_ids: list[str], seed_slugs: dict[str, list[str]], *, direction: str, limit: int, include_drafts: bool, ) -> list[dict]: assert mapping_ids == ["mapping-1"] assert seed_slugs == {"mapping-1": ["news/xinghe"]} assert direction == "both" assert limit == 6 assert include_drafts is False return [ { "neighbor_slug": "entities/tianheng", "direction": "outgoing", "page": { "slug": "entities/tianheng", "title": "天衡建设集团", }, } ] class Client: async def get_wiki_page(self, *_args, **_kwargs): raise AssertionError("complete local mirror must serve the page") monkeypatch.setattr(tools, "get_local_wiki_index_store", LocalIndex) monkeypatch.setattr(tools, "_client", Client) payload = json.loads( await tools._wiki_read_impl( tools._page_ref("mapping-1", "news/xinghe"), { "knowledge_retrieval_mode": "smart_reasoning", "llmwiki_knowledge_base_ids": ["mapping-1"], }, ) ) assert payload["channel"] == "local_wiki_mirror" assert payload["results"][0]["aliases"] == ["星河计划"] assert payload["results"][0]["links"] == [ { "slug": "entities/tianheng", "title": "天衡建设集团", "direction": "outgoing", } ] @pytest.mark.asyncio async def test_wiki_read_hides_expansion_links_when_database_setting_is_disabled( monkeypatch, ) -> None: _patch_scope(monkeypatch) class DisabledMappingStore(_MappingStore): async def get_authorized(self, *args, **kwargs) -> dict | None: row = await super().get_authorized(*args, **kwargs) return { **(row or {}), "retrieval_settings": { "entity_expansion_enabled": False, "max_related_terms": 6, "statistics_enabled": True, }, } class LocalIndex: async def list_index_status(self, _mapping_ids: list[str]) -> list[dict]: return [ { "knowledge_base_mapping_id": "mapping-1", "state": "idle", "last_completed_at": "2026-10-14T00:00:00Z", "remote_page_count": 1, "local_page_count": 1, "failed_page_count": 0, "last_error": None, "last_sync_result": {"status": "completed"}, } ] async def get_page(self, _mapping_id: str, slug: str) -> dict: return { "slug": slug, "title": "星河工程建设进展", "content_md": "参见 [[entities/tianheng]]。", "out_links": ["entities/tianheng"], "status": "published", } async def get_adjacent_pages(self, *_args, **_kwargs) -> list[dict]: raise AssertionError("disabled expansion must not query the Wiki graph") monkeypatch.setattr(tools, "get_llmwiki_store", DisabledMappingStore) monkeypatch.setattr(tools, "get_local_wiki_index_store", LocalIndex) payload = json.loads( await tools._wiki_read_impl( tools._page_ref("mapping-1", "news/xinghe"), { "knowledge_retrieval_mode": "smart_reasoning", "llmwiki_knowledge_base_ids": ["mapping-1"], }, ) ) assert payload["results"][0]["links"] == [] @pytest.mark.asyncio async def test_wiki_regex_does_not_use_a_stale_local_mirror( monkeypatch, ) -> None: _patch_scope(monkeypatch) class StaleMirror: def __init__(self) -> None: raise AssertionError("smart Wiki regex must not inspect the local mirror") class Client: async def search_wiki_pages( self, knowledge_base_id: str, query: str, *, limit: int, ) -> list[dict]: assert (knowledge_base_id, query, limit) == ( "remote-1", "星河工程", 50, ) return [ { "slug": "news/new", "title": "新报道", "content": "星河工程发布最新进展。", "status": "published", } ] monkeypatch.setattr(tools, "get_local_wiki_index_store", StaleMirror) monkeypatch.setattr(tools, "_client", Client) payload = json.loads( await tools._wiki_search_impl( "星河工程", { "knowledge_retrieval_mode": "smart_reasoning", "llmwiki_knowledge_base_ids": ["mapping-1"], }, ) ) assert payload["result_count"] == 1 assert payload["results"][0]["slug"] == "news/new" assert payload["channel"] == "provider_native_regex" @pytest.mark.asyncio async def test_wiki_provider_all_failures_return_error_and_partial_is_warning( monkeypatch, ) -> None: _patch_scope(monkeypatch, multi=True) class AllFailedClient: async def search_wiki_pages(self, *args, **kwargs) -> list[dict]: raise WeKnoraError("down", status_code=503) monkeypatch.setattr(tools, "_client", AllFailedClient) failed = json.loads( await tools._wiki_search_impl( "needle", { "knowledge_retrieval_mode": "smart_reasoning", "llmwiki_knowledge_base_ids": ["mapping-1", "mapping-2"], }, ) ) assert failed["ok"] is False assert failed["error"]["code"] == "PROVIDER_UNAVAILABLE" class PartiallyFailedClient: async def search_wiki_pages( self, knowledge_base_id: str, query: str, *, limit: int, ) -> list[dict]: if knowledge_base_id == "remote-2": raise WeKnoraError("down", status_code=503) return [{"slug": "ok", "title": "needle", "content": "正文"}] monkeypatch.setattr(tools, "_client", PartiallyFailedClient) partial = json.loads( await tools._wiki_search_impl( "needle", { "knowledge_retrieval_mode": "smart_reasoning", "llmwiki_knowledge_base_ids": ["mapping-1", "mapping-2"], }, ) ) assert partial["ok"] is True assert partial["partial"] is True assert partial["warnings"] == [ {"code": "PROVIDER_WIKI_SEARCH_FAILED", "knowledge_base": "制度库2"} ] @pytest.mark.asyncio async def test_client_grep_compatibility_fallback_is_marked_and_fair() -> None: class Client(WeKnoraClient): async def _request(self, *args, **kwargs): raise WeKnoraError("missing", status_code=404) async def list_documents( self, knowledge_base_id: str, *, page: int, page_size: int, ) -> dict: count = 30 if knowledge_base_id == "remote-1" else 1 rows = [ {"id": f"{knowledge_base_id}-doc-{index}", "title": "needle title"} for index in range(count) ] return {"items": rows, "total": count} async def list_chunks( self, knowledge_id: str, *, page: int, page_size: int, ) -> dict: return { "items": [ { "id": f"{knowledge_id}-chunk", "content": f"needle from {knowledge_id}", "chunk_index": 0, } ], "total": 1, } client = Client(base_url="http://weknora.test") payload = await client.grep_chunks("needle", ["remote-1", "remote-2"]) assert payload["compatibility_fallback"] is True assert payload["contract"] == "weknora-existing-api-compat-v1" assert payload["successful_knowledge_base_ids"] == ["remote-1", "remote-2"] assert {item["knowledge_base_id"] for item in payload["items"]} == { "remote-1", "remote-2", } assert payload["warnings"][0]["code"] == "WEKNORA_GREP_COMPATIBILITY_FALLBACK" @pytest.mark.asyncio async def test_client_grep_prefers_single_cross_scope_provider_request() -> None: class Client(WeKnoraClient): calls: list[tuple[str, str, dict]] = [] async def _request(self, method: str, path: str, *, json=None, **kwargs): self.calls.append((method, path, json)) return { "success": True, "data": { "contract": "weknora-v0.7.2-grep-v1", "chunk_results": [ { "id": "chunk-2", "knowledge_id": "doc-2", "knowledge_base_id": "remote-2", "knowledge_title": "第二库", "content": "needle", } ], }, } client = Client(base_url="http://weknora.test") payload = await client.grep_chunks("needle", ["remote-1", "remote-2"]) assert client.calls == [ ( "POST", "/api/v1/chunks/grep", { "query": "needle", "knowledge_base_ids": ["remote-1", "remote-2"], }, ) ] assert payload["compatibility_fallback"] is False assert payload["items"][0]["knowledge_base_id"] == "remote-2" @pytest.mark.asyncio async def test_client_grep_compatibility_all_scope_failure_raises() -> None: class Client(WeKnoraClient): async def _request(self, *args, **kwargs): raise WeKnoraError("missing", status_code=405) async def list_documents(self, *args, **kwargs) -> dict: raise WeKnoraError("down") client = Client(base_url="http://weknora.test") with pytest.raises(WeKnoraError): await client.grep_chunks("needle", ["remote-1", "remote-2"]) @pytest.mark.asyncio async def test_chunk_grep_all_provider_failure_returns_error(monkeypatch) -> None: _patch_scope(monkeypatch) class Client: async def grep_chunks(self, query: str, knowledge_base_ids: list[str]) -> dict: raise WeKnoraError("down") monkeypatch.setattr(tools, "_client", Client) payload = json.loads( await tools._chunk_regex_impl( "needle", { "knowledge_retrieval_mode": "smart_reasoning", "llmwiki_knowledge_base_ids": ["mapping-1"], }, ) ) assert payload["ok"] is False @pytest.mark.asyncio async def test_chunk_grep_partial_failure_is_structured_and_sanitized(monkeypatch) -> None: _patch_scope(monkeypatch, multi=True) class Client: async def grep_chunks(self, query: str, knowledge_base_ids: list[str]) -> dict: return { "items": [ { "chunk_id": "chunk-1", "knowledge_id": "doc-1", "knowledge_base_id": "remote-1", "content": "needle", } ], "contract": "weknora-existing-api-compat-v1", "compatibility_fallback": True, "partial": True, "warnings": [ {"code": "PROVIDER_SCOPE_FAILED", "knowledge_base_id": "remote-2"} ], "successful_knowledge_base_ids": ["remote-1"], "failed_knowledge_base_ids": ["remote-2"], } monkeypatch.setattr(tools, "_client", Client) payload = json.loads( await tools._chunk_regex_impl( "needle", { "knowledge_retrieval_mode": "smart_reasoning", "llmwiki_knowledge_base_ids": ["mapping-1", "mapping-2"], }, ) ) assert payload["ok"] is True assert payload["partial"] is True assert payload["compatibility_fallback"] is True assert payload["warnings"] == [ {"code": "PROVIDER_SCOPE_FAILED", "knowledge_base": "制度库2"} ] @pytest.mark.asyncio async def test_wiki_read_continuation_reconstructs_long_page_and_binds_cursor( monkeypatch, ) -> None: _patch_scope(monkeypatch) content = "# 第一章\n\n" + "甲" * 35_000 + "\n\n# 第二章\n\n" + "乙" * 35_000 class LocalIndex: async def list_index_status(self, mapping_ids: list[str]) -> list[dict]: return [ { "knowledge_base_mapping_id": "mapping-1", "state": "idle", "last_completed_at": "2026-10-14T00:00:00Z", "remote_page_count": 1, "local_page_count": 1, "failed_page_count": 0, "last_error": None, "last_sync_result": {"status": "completed"}, } ] async def get_page(self, mapping_id: str, slug: str) -> dict: assert (mapping_id, slug) == ("mapping-1", "long-page") return {"title": "长文", "slug": slug, "content_md": content} monkeypatch.setattr(tools, "get_local_wiki_index_store", LocalIndex) page_ref = tools._page_ref("mapping-1", "long-page") cursor = "" parts: list[str] = [] for _ in range(10): payload = json.loads( await tools._wiki_read_impl( page_ref, { "knowledge_retrieval_mode": "smart_reasoning", "llmwiki_knowledge_base_ids": ["mapping-1"], }, cursor=cursor, ) ) page = payload["results"][0] parts.append(page["content_markdown"]) if page["complete"]: assert page["next_cursor"] is None break assert page["content_truncated"] is True cursor = page["next_cursor"] else: pytest.fail("long Wiki page did not complete") assert "".join(parts) == content other_page_ref = tools._page_ref("mapping-1", "other-page") with pytest.raises(tools.SmartKnowledgeToolError) as caught: tools._decode_continuation_cursor(cursor, other_page_ref) assert caught.value.code == "INVALID_CONTINUATION_CURSOR" @pytest.mark.asyncio async def test_deep_read_rejects_cross_scope_ids_and_allows_second_mapping(monkeypatch) -> None: _patch_scope(monkeypatch) class UnauthorizedClient: async def get_chunk(self, chunk_id: str) -> dict: return {"id": chunk_id, "knowledge_base_id": "remote-2"} async def get_document(self, knowledge_id: str) -> dict: return {"id": knowledge_id, "knowledge_base_id": "remote-2"} monkeypatch.setattr(tools, "_client", UnauthorizedClient) context = { "knowledge_retrieval_mode": "smart_reasoning", "llmwiki_knowledge_base_ids": ["mapping-1"], } with pytest.raises(tools.SmartKnowledgeToolError) as chunk_error: await tools._document_chunks_impl( "foreign-chunk", "chunk_id", context, page=1, page_size=20, ) assert chunk_error.value.code == "CHUNK_NOT_FOUND" with pytest.raises(tools.SmartKnowledgeToolError) as document_error: await tools._document_chunks_impl( "foreign-doc", "knowledge_id", context, page=1, page_size=20, ) assert document_error.value.code == "DOCUMENT_NOT_FOUND" _patch_scope(monkeypatch, multi=True) class AuthorizedSecondClient: async def get_document(self, knowledge_id: str) -> dict: return {"id": knowledge_id, "knowledge_base_id": "remote-2"} async def list_chunks( self, knowledge_id: str, *, page: int, page_size: int, ) -> dict: return { "items": [{"id": "chunk-2", "content": "第二库正文"}], "total": 1, } monkeypatch.setattr(tools, "_client", AuthorizedSecondClient) allowed = json.loads( await tools._document_chunks_impl( "doc-2", "knowledge_id", { "knowledge_retrieval_mode": "smart_reasoning", "llmwiki_knowledge_base_ids": ["mapping-1", "mapping-2"], }, page=1, page_size=20, ) ) assert allowed["ok"] is True assert allowed["results"][0]["content"] == "第二库正文" assert allowed["results"][0]["knowledge_base_id"] == "mapping-2" def test_regex_execution_has_a_hard_timeout() -> None: expression = tools._safe_regex("(a|aa)+$") with pytest.raises(tools.SmartKnowledgeToolError) as caught: tools._regex_search(expression, "a" * 10_000 + "!") assert caught.value.code == "REGEX_TIMEOUT" @pytest.mark.asyncio @pytest.mark.parametrize("assistant_id", ["deep-research-planner", "deep-research-collector", "custom-agent"]) async def test_research_smart_scope_cannot_access_unauthorized_library(monkeypatch, assistant_id): from app.gateway import llmwiki_rag request = _request(None) request.app.state.llmwiki_store = _MappingStore() monkeypatch.setattr(llmwiki_rag, "get_resolved_llmwiki_runtime", lambda _: SimpleNamespace(weknora_enabled=True)) monkeypatch.setattr(llmwiki_rag, "_agent_bound_ids", lambda _: []) async def actor(_request): return "user-1", False monkeypatch.setattr(llmwiki_rag, "_actor", actor) with pytest.raises(HTTPException) as caught: await llmwiki_rag.build_llmwiki_smart_reasoning_context( request=request, body=SimpleNamespace(context={"llmwiki_knowledge_base_ids": ["not-authorized"]}, metadata={}, assistant_id=assistant_id), graph_input={"messages": [HumanMessage(content="test")]}, ) assert caught.value.status_code == 404 @pytest.mark.parametrize("agent_id", ["deep-research-planner", "deep-research-collector"]) def test_builtin_research_tools_use_caller_authorized_scope(agent_id): assert tools._allowed_agent_ids({"agent_id": agent_id}) is None from deerflow.tools.builtins.llmwiki_search_tool import _agent_allowed_ids assert _agent_allowed_ids({"agent_id": agent_id}) is None