deer-flow/backend/tests/test_weknora_extension.py

"""Synthetic v0.8.3 responses; never contacts an external provider."""

import json
from pathlib import Path

import httpx
import pytest

from deerflow.knowledge import KnowledgeRequestContext, format_result, sources_from_messages


@pytest.fixture
def adapter(monkeypatch):
    monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[2] / "examples/deerflow-extension-weknora"))
    from deerflow_extension_weknora import WeKnoraProvider

    monkeypatch.setenv("WEKNORA_TEST_KEY", "synthetic-secret")
    calls = []
    document = {"id": "doc-1", "knowledge_base_id": "kb-1", "title": "Synthetic guide", "parse_status": "completed", "enable_status": "enabled"}
    state = {"document": document, "hits": [{"knowledge_id": "doc-1", "content": "Cedar is synthetic evidence.", "knowledge_title": "Synthetic guide", "score": 0.8}]}

    def handler(request):
        calls.append(request)
        assert request.headers["X-API-Key"] == "synthetic-secret"
        assert request.headers["X-Tenant-ID"] in {"1", "2"}
        if state.get("status"):
            return httpx.Response(state["status"], text="synthetic-secret upstream body")
        if request.url.path == "/api/v1/knowledge-bases":
            data = state.get("bases", [{"id": "kb-1", "name": "Synthetic base", "capabilities": {"wiki": True}}])
        elif request.url.path == "/api/v1/knowledge-bases/kb-1/knowledge":
            return httpx.Response(200, json={"success": True, "data": [state["document"]], **state.get("pagination", {"total": 1})})
        elif request.url.path == "/api/v1/knowledge/doc-1":
            data = state["document"]
        elif request.url.path == "/api/v1/knowledgebase/kb-1/wiki/search":
            return httpx.Response(200, json={"pages": state.get("pages", [])})
        elif request.url.path == "/api/v1/knowledge-search":
            return httpx.Response(403, json={"success": False, "error": "KB-restricted keys reject global knowledge_ids"})
        elif request.url.path == "/api/v1/knowledge-bases/kb-1/hybrid-search":
            data = state["hits"]
        else:
            return httpx.Response(404, json={"success": False})
        return httpx.Response(200, json={"success": True, "data": data})

    config = {"spaces": [{"base_url": "https://weknora.example", "tenant_id": "1", "api_key_env": "WEKNORA_TEST_KEY", "user_ids": ["alice"], "agent_names": ["lead_agent", "researcher"]}]}
    transport = httpx.MockTransport(handler)
    return WeKnoraProvider(config, transport=transport), calls, state, config, transport


async def selection(provider, *, docs=True):
    context = KnowledgeRequestContext("alice", "lead_agent")
    page = await provider.catalog(context=context, page=1, page_size=20, search="")
    dataset_id = page.items[0].id
    scope = {"version": 1, "mode": "selected", "dataset_ids": [dataset_id]}
    if docs:
        scope["document_filters"] = [{"dataset_id": dataset_id, "document_ids": ["doc-1"]}]
    return KnowledgeRequestContext("alice", "lead_agent", scope)


@pytest.mark.asyncio
async def test_catalog_documents_search_shared_evidence(adapter):
    provider, calls, _, _, _ = adapter
    ctx = await selection(provider)
    page = await provider.documents(ctx.scope["dataset_ids"][0], context=ctx, page=1, page_size=20, search="guide")
    assert page.items[0].selectable and page.items[0].id == "doc-1"
    assert calls[-1].url.params["keyword"] == "guide"
    result = await provider.search_scoped("Cedar", context=ctx)
    body = json.loads(next(r.content for r in calls if r.method == "POST"))
    assert body["knowledge_ids"] == ["doc-1"]
    assert "knowledge_base_ids" not in body
    assert body["query_text"] == "Cedar"
    assert body["skip_context_enrichment"] is True
    assert next(r for r in calls if r.method == "POST").url.path == "/api/v1/knowledge-bases/kb-1/hybrid-search"
    content, artifact = format_result(result)
    sources = sources_from_messages([{"type": "tool", "content": content, "artifact": artifact}])
    assert len(sources) == 1 and sources[0].provider == "weknora"
    assert sources[0].excerpt == "Cedar is synthetic evidence."
    assert sources[0].page is None and sources[0].url is None
    assert sources_from_messages([{"type": "tool", "content": "removed", "artifact": artifact}]) == []


@pytest.mark.asyncio
@pytest.mark.parametrize("user,agent", [("mallory", "lead_agent"), ("alice", "unbound"), ("alice", None)])
async def test_unbound_identity_cannot_make_requests(adapter, user, agent):
    provider, calls, *_ = adapter
    ctx = KnowledgeRequestContext(user, agent)
    with pytest.raises(PermissionError):
        await provider.catalog(context=ctx, page=1, page_size=20, search="")
    with pytest.raises(PermissionError):
        await provider.search_scoped("Cedar", context=ctx)
    assert not calls


@pytest.mark.asyncio
async def test_no_context_search_is_not_an_authorization_bypass(adapter):
    provider, calls, *_ = adapter
    with pytest.raises(PermissionError):
        await provider.search("Cedar")
    assert not calls


@pytest.mark.asyncio
@pytest.mark.parametrize("scope", [{"version": 1, "mode": "disabled"}, {"mode": "selected", "dataset_ids": []}])
async def test_disabled_or_invalid_scope_never_calls_upstream(adapter, scope):
    provider, calls, *_ = adapter
    with pytest.raises(ValueError):
        await provider.search_scoped("Cedar", context=KnowledgeRequestContext("alice", "lead_agent", scope))
    assert not calls


@pytest.mark.asyncio
@pytest.mark.parametrize("change", [{"knowledge_base_id": "other"}, {"id": "other"}, {"enable_status": "disabled"}, {"parse_status": "pending"}])
async def test_document_validation_precedes_search(adapter, change):
    provider, calls, state, *_ = adapter
    ctx = await selection(provider)
    state["document"].update(change)
    with pytest.raises(PermissionError):
        await provider.search_scoped("Cedar", context=ctx)
    assert not any(r.method == "POST" for r in calls)


@pytest.mark.asyncio
async def test_unselected_result_fails_entire_request(adapter):
    provider, calls, state, *_ = adapter
    ctx = await selection(provider)
    state["hits"].append({"knowledge_id": "other", "content": "must never escape"})
    with pytest.raises(PermissionError):
        await provider.search_scoped("Cedar", context=ctx)
    assert len([r for r in calls if r.method == "POST"]) == 1


@pytest.mark.asyncio
async def test_dataset_search_verifies_returned_document_membership(adapter):
    provider, _, state, *_ = adapter
    ctx = await selection(provider, docs=False)
    state["document"]["knowledge_base_id"] = "other"
    with pytest.raises(PermissionError):
        await provider.search_scoped("Cedar", context=ctx)


@pytest.mark.asyncio
@pytest.mark.parametrize("status,error", [(401, PermissionError), (403, PermissionError), (404, FileNotFoundError), (500, RuntimeError), (302, RuntimeError)])
async def test_http_errors_are_redacted_and_never_retried(adapter, status, error):
    provider, calls, state, *_ = adapter
    state["status"] = status
    with pytest.raises(error) as exc:
        await provider.catalog(context=KnowledgeRequestContext("alice", "lead_agent"), page=1, page_size=20, search="")
    assert "synthetic-secret" not in str(exc.value)
    assert len(calls) == 1


@pytest.mark.asyncio
async def test_multi_space_ids_are_stable_bounded_and_isolated(adapter):
    provider, _, _, config, transport = adapter
    from deerflow_extension_weknora import WeKnoraProvider

    config["spaces"].append({**config["spaces"][0], "tenant_id": "2", "user_ids": ["bob"]})
    multi = WeKnoraProvider(config, transport=transport)
    alice = await multi.catalog(context=KnowledgeRequestContext("alice", "lead_agent"), page=1, page_size=20, search="")
    bob = await multi.catalog(context=KnowledgeRequestContext("bob", "lead_agent"), page=1, page_size=20, search="")
    old = await provider.catalog(context=KnowledgeRequestContext("alice", "lead_agent"), page=1, page_size=20, search="")
    assert alice.items[0].id == old.items[0].id != bob.items[0].id
    assert len(alice.items[0].id) <= 256
    with pytest.raises(PermissionError):
        await multi.search_scoped("Cedar", context=KnowledgeRequestContext("bob", "lead_agent", {"version": 1, "mode": "selected", "dataset_ids": [alice.items[0].id]}))


@pytest.mark.asyncio
async def test_all_scope_still_sends_explicit_targets_and_subagent_works(adapter):
    provider, calls, *_ = adapter
    await provider.search_scoped("Cedar", context=KnowledgeRequestContext("alice", "researcher"))
    body = json.loads(next(r.content for r in calls if r.method == "POST"))
    assert "knowledge_base_ids" not in body and "knowledge_ids" not in body
    assert next(r for r in calls if r.method == "POST").url.path == "/api/v1/knowledge-bases/kb-1/hybrid-search"


@pytest.mark.asyncio
async def test_secrets_redacted_and_missing_attribution_not_invented(adapter):
    provider, _, state, *_ = adapter
    state["hits"][0].update(content="synthetic-secret evidence", knowledge_title="")
    state["document"]["title"] = ""
    result = await provider.search_scoped("Cedar", context=KnowledgeRequestContext("alice", "lead_agent"))
    assert "synthetic-secret" not in result.model_dump_json()
    assert result.hits[0].source is None


@pytest.mark.asyncio
@pytest.mark.parametrize("bad_id", ["..", "../secret", "%2e%2e", "a/b", "x" * 257])
async def test_unsafe_native_ids_fail_before_path_request(adapter, bad_id):
    provider, calls, *_ = adapter
    ctx = await selection(provider)
    ctx.scope["document_filters"][0]["document_ids"] = [bad_id]
    with pytest.raises(ValueError):
        await provider.search_scoped("Cedar", context=ctx)
    assert not any("/knowledge/" in r.url.path or r.method == "POST" for r in calls)


def test_configuration_rejects_implicit_authorization_and_credentials(adapter):
    from deerflow_extension_weknora import WeKnoraProvider

    _, _, _, config, _ = adapter
    for change in ({"tenant_id": None}, {"user_ids": []}, {"base_url": "https://user:secret@example.com"}, {"api_key_env": ""}, {"knowledge_base_ids": []}):
        with pytest.raises(ValueError):
            WeKnoraProvider({"spaces": [{**config["spaces"][0], **change}]})


@pytest.mark.asyncio
async def test_packaged_registration_real_tool_replay_budget_and_delegation(adapter, monkeypatch):
    from types import SimpleNamespace

    from langchain_core.messages import AIMessage
    from langgraph.graph import END, START, MessagesState, StateGraph
    from langgraph.prebuilt import ToolNode

    from deerflow.extensions import EXTENSION_SNAPSHOT_CONTEXT_KEY, load_extensions
    from deerflow.extensions.loader import ExtensionSpec
    from deerflow.knowledge import tools
    from deerflow.knowledge.providers import GENERIC_KNOWLEDGE_TOOL, resolve_knowledge_provider
    from deerflow.knowledge.sources import prune_sources
    from deerflow.runtime.serialization import serialize_channel_values_for_api
    from deerflow.tools.builtins.task_tool import _task_result_command

    _, _, _, config, transport = adapter
    loaded, diagnostics = load_extensions([ExtensionSpec(use="deerflow_extension_weknora:install", config=config)])
    assert diagnostics == []
    app_config = SimpleNamespace(get_tool_config=lambda _: SimpleNamespace(use=GENERIC_KNOWLEDGE_TOOL, model_extra={"provider": "weknora"}))
    provider = resolve_knowledge_provider(app_config, extensions=loaded)
    provider._transport = transport
    ctx = await selection(provider)
    monkeypatch.setattr(tools, "get_app_config", lambda: app_config)
    builder = StateGraph(MessagesState, context_schema=dict)
    builder.add_node("tools", ToolNode([tools.knowledge_search_tool]))
    builder.add_edge(START, "tools")
    builder.add_edge("tools", END)
    result = await builder.compile().ainvoke(
        {"messages": [AIMessage(content="", tool_calls=[{"id": "call", "name": "knowledge_search", "args": {"query": "Cedar"}}])]},
        context={"user_id": "alice", "is_subagent": True, "agent_id": "researcher", "__knowledge_scope_execution": ctx.scope, EXTENSION_SNAPSHOT_CONTEXT_KEY: loaded},
    )
    message = result["messages"][-1]
    sources = sources_from_messages([message])
    assert len(sources) == 1 and sources[0].provider == "weknora"
    persisted = json.loads(json.dumps(serialize_channel_values_for_api({"messages": [message]})))
    assert sources_from_messages(persisted["messages"]) == sources
    command = _task_result_command(tool_call_id="task", status="completed", result=message.content, knowledge_sources=[s.model_dump(mode="json", exclude_none=True) for s in sources])
    assert sources_from_messages(command.update["messages"]) == sources
    _, pruned = prune_sources("truncated", message.artifact)
    assert pruned["knowledge_sources"] == []


@pytest.mark.asyncio
async def test_missing_key_fails_closed(adapter, monkeypatch):
    provider, calls, *_ = adapter
    monkeypatch.delenv("WEKNORA_TEST_KEY")
    with pytest.raises(RuntimeError):
        await provider.search_scoped("Cedar", context=KnowledgeRequestContext("alice", "lead_agent"))
    assert not calls


@pytest.mark.asyncio
async def test_allowlist_and_missing_configured_base(adapter):
    from deerflow_extension_weknora import WeKnoraProvider

    _, calls, state, config, transport = adapter
    config["spaces"][0]["knowledge_base_ids"] = ["kb-1"]
    provider = WeKnoraProvider(config, transport=transport)
    state["bases"] = [{"id": "kb-1", "name": "Allowed"}, {"id": "kb-2", "name": "Excluded"}]
    page = await provider.catalog(context=KnowledgeRequestContext("alice", "lead_agent"), page=1, page_size=20, search="")
    assert [item.name for item in page.items] == ["Allowed"]
    state["bases"] = [{"id": "kb-2", "name": "Excluded"}]
    with pytest.raises(PermissionError):
        await provider.search_scoped("Cedar", context=KnowledgeRequestContext("alice", "lead_agent"))
    assert not any(r.method == "POST" for r in calls)


@pytest.mark.asyncio
async def test_empty_catalog_never_sends_unconstrained_search(adapter):
    provider, calls, state, *_ = adapter
    state["bases"] = None
    result = await provider.search_scoped("Cedar", context=KnowledgeRequestContext("alice", "lead_agent"))
    assert result.hits == [] and not any(r.method == "POST" for r in calls)


@pytest.mark.asyncio
@pytest.mark.parametrize("payload", [[], {"success": False, "data": []}, {"success": True}, {"success": True, "data": "bad"}])
async def test_malformed_upstream_response_fails_closed(adapter, payload):
    provider, *_ = adapter
    provider._transport = httpx.MockTransport(lambda _: httpx.Response(200, json=payload))
    with pytest.raises(RuntimeError):
        await provider.catalog(context=KnowledgeRequestContext("alice", "lead_agent"), page=1, page_size=20, search="")


@pytest.mark.asyncio
async def test_response_size_limit(adapter):
    provider, *_ = adapter
    provider._transport = httpx.MockTransport(lambda _: httpx.Response(200, content=b"x" * (4 * 1024 * 1024 + 1)))
    with pytest.raises(RuntimeError, match="size"):
        await provider.catalog(context=KnowledgeRequestContext("alice", "lead_agent"), page=1, page_size=20, search="")


@pytest.mark.asyncio
async def test_all_document_selections_checked_before_any_search(adapter):
    provider, calls, *_ = adapter
    ctx = await selection(provider)
    ctx.scope["document_filters"][0]["document_ids"].append("missing")
    with pytest.raises(FileNotFoundError):
        await provider.search_scoped("Cedar", context=ctx)
    assert not any(r.method == "POST" for r in calls)


def test_gateway_catalog_uses_authenticated_user_with_real_adapter(adapter):
    from _router_auth_helpers import make_authed_test_app
    from deerflow_extension_weknora import WeKnoraProvider
    from fastapi.testclient import TestClient

    from app.gateway.auth.models import User
    from app.gateway.deps import get_config
    from app.gateway.routers import knowledge
    from deerflow.config.app_config import AppConfig
    from deerflow.extensions import ExtensionRegistry

    _, calls, _, config, transport = adapter
    user = User(email="weknora@example.com", password_hash="unused", system_role="user")
    config["spaces"][0]["user_ids"] = [str(user.id)]
    provider = WeKnoraProvider(config, transport=transport)
    registry = ExtensionRegistry()
    with registry.attributed_to("deerflow_extension_weknora:install"):
        registry.knowledge_provider("weknora", provider)
    app = make_authed_test_app(user_factory=lambda: user, bind_current_user=True)
    app.state.extensions = registry.build()
    app.include_router(knowledge.router)
    app_config = AppConfig.model_validate(
        {
            "models": [],
            "sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"},
            "knowledge_base": {"enabled": True, "scope_selection_enabled": True},
            "tools": [{"name": "knowledge_search", "group": "knowledge", "use": "deerflow.knowledge.tools:knowledge_search_tool", "provider": "weknora"}],
        }
    )
    app.dependency_overrides[get_config] = lambda: app_config
    with TestClient(app) as client:
        cap = client.get("/api/knowledge/capabilities?agent_name=lead_agent")
        assert cap.status_code == 200 and cap.json()["scope_selection_enabled"] is True
        catalog = client.get("/api/knowledge/retrieval-catalog/datasets?agent_name=lead_agent&user_id=mallory")
        assert catalog.status_code == 200
        dataset_id = catalog.json()["items"][0]["id"]
        documents = client.get(f"/api/knowledge/retrieval-catalog/datasets/{dataset_id}/documents?agent_name=lead_agent")
        assert documents.status_code == 200 and documents.json()["items"][0]["selectable"] is True
        assert client.get("/api/knowledge/retrieval-catalog/datasets/unknown/documents?agent_name=lead_agent").status_code == 403
    assert all(r.method == "GET" for r in calls)


@pytest.mark.asyncio
async def test_oversized_hit_list_does_not_cause_unbounded_detail_requests(adapter):
    provider, calls, state, *_ = adapter
    state["hits"] = [{"knowledge_id": f"doc-{i}", "content": "synthetic"} for i in range(101)]
    with pytest.raises(RuntimeError, match="result limit"):
        await provider.search_scoped("Cedar", context=KnowledgeRequestContext("alice", "lead_agent"))
    assert not any("/knowledge/" in r.url.path for r in calls)


@pytest.mark.asyncio
@pytest.mark.parametrize("pagination", [{}, {"total": None}, {"total": "1"}, {"total": True}, {"total": -1}, {"total": 1.5}])
async def test_invalid_document_total_is_explicit_provider_error(adapter, pagination):
    provider, calls, state, *_ = adapter
    ctx = await selection(provider)
    state["pagination"] = pagination
    with pytest.raises(RuntimeError, match="Invalid WeKnora document total"):
        await provider.documents(ctx.scope["dataset_ids"][0], context=ctx, page=1, page_size=20, search="")
    assert not any(r.method == "POST" for r in calls)


@pytest.mark.asyncio
@pytest.mark.parametrize("target", ["hits", "document"])
@pytest.mark.parametrize("data", ["bad", [], ["bad"], 42])
async def test_malformed_retrieval_payloads_fail_without_retry(adapter, target, data):
    provider, calls, state, *_ = adapter
    ctx = await selection(provider)
    state[target] = data
    if target == "hits" and data == []:
        assert (await provider.search_scoped("Cedar", context=ctx)).hits == []
    else:
        with pytest.raises(RuntimeError, match="Invalid WeKnora"):
            await provider.search_scoped("Cedar", context=ctx)
    assert len([r for r in calls if r.method == "POST"]) == (1 if target == "hits" else 0)


@pytest.mark.asyncio
async def test_retrieval_response_size_limit_without_retry(adapter):
    provider, calls, _, _, transport = adapter
    ctx = await selection(provider, docs=False)
    posts = []

    async def handler(request):
        if request.method == "POST":
            posts.append(request)
            return httpx.Response(200, content=b"x" * (4 * 1024 * 1024 + 1))
        return await transport.handle_async_request(request)

    provider._transport = httpx.MockTransport(handler)
    with pytest.raises(RuntimeError, match="size"):
        await provider.search_scoped("Cedar", context=ctx)
    assert len(posts) == 1
    assert not any("/knowledge/" in r.url.path for r in calls)


@pytest.mark.asyncio
async def test_published_entries_use_real_wiki_evidence(adapter):
    provider, calls, state, *_ = adapter
    ctx = await selection(provider, docs=False)
    state["pages"] = [{"knowledge_base_id": "kb-1", "slug": "cedar", "title": "Cedar entry", "status": "published", "summary": "Cedar entry evidence."}]
    result = await provider.entries_search("Cedar", context=ctx)
    assert result.hits[0].source.title == "Cedar entry"
    assert result.hits[0].source.excerpt == "Cedar entry evidence."
    assert calls[-1].url.params["q"] == "Cedar"
    content, artifact = format_result(result)
    assert len(sources_from_messages([{"type": "tool", "name": "knowledge_entries_search", "content": content, "artifact": artifact}])) == 1
    state["pages"][0]["status"] = "draft"
    assert not (await provider.entries_search("Cedar", context=ctx)).hits
    state["pages"][0]["knowledge_base_id"] = "foreign"
    with pytest.raises(PermissionError):
        await provider.entries_search("Cedar", context=ctx)


@pytest.mark.asyncio
async def test_entries_reject_document_filter_before_network(adapter):
    provider, calls, *_ = adapter
    ctx = await selection(provider)
    calls.clear()
    with pytest.raises(ValueError, match="document"):
        await provider.entries_search("Cedar", context=ctx)
    assert calls == []


@pytest.mark.asyncio
async def test_entries_skip_non_wiki_bases_and_reject_unbound_identity(adapter):
    provider, calls, state, *_ = adapter
    state["bases"] = [{"id": "kb-1", "name": "Document only", "capabilities": {"wiki": False}}]
    assert not (await provider.entries_search("q", context=KnowledgeRequestContext("alice", "lead_agent"))).hits
    assert not any("/wiki/" in str(request.url) for request in calls)
    calls.clear()
    with pytest.raises(PermissionError):
        await provider.entries_search("q", context=KnowledgeRequestContext("mallory", "lead_agent"))
    assert not calls


@pytest.mark.asyncio
async def test_wiki_empty_phrase_falls_back_to_bounded_keywords_with_same_scope(adapter):
    provider, calls, state, _, transport = adapter
    ctx = await selection(provider, docs=False)
    seen = []
    page = {"knowledge_base_id": "kb-1", "slug": "cedar", "title": "青杉", "status": "published", "summary": "林岚负责青杉计划"}

    async def handler(request):
        if request.url.path.endswith("/wiki/search"):
            seen.append(request)
            return httpx.Response(200, json={"pages": [page] if request.url.params["q"] in {"青杉", "林岚"} else []})
        return await transport.handle_async_request(request)

    provider._transport = httpx.MockTransport(handler)
    result = await provider.entries_search("青杉 计划 负责人 林岚 星河实验室 青杉", context=ctx)
    assert [r.url.params["q"] for r in seen] == ["青杉 计划 负责人 林岚 星河实验室 青杉", "青杉", "计划", "负责人", "林岚"]
    assert all(r.url.path == "/api/v1/knowledgebase/kb-1/wiki/search" for r in seen)
    assert all(r.headers["X-API-Key"] == "synthetic-secret" for r in seen)
    assert len(result.hits) == 1
    assert result.hits[0].source.excerpt == "林岚负责青杉计划"


@pytest.mark.asyncio
@pytest.mark.parametrize("failure", ["foreign", "malformed", "http"])
async def test_wiki_keyword_fallback_preserves_fail_closed_errors(adapter, failure):
    provider, _, _, _, transport = adapter
    seen = []

    async def handler(request):
        if not request.url.path.endswith("/wiki/search"):
            return await transport.handle_async_request(request)
        seen.append(request.url.params["q"])
        if len(seen) == 1:
            return httpx.Response(200, json={"pages": []})
        if failure == "http":
            return httpx.Response(403)
        if failure == "malformed":
            return httpx.Response(200, json={"pages": "bad"})
        return httpx.Response(200, json={"pages": [{"knowledge_base_id": "foreign", "status": "published"}]})

    provider._transport = httpx.MockTransport(handler)
    with pytest.raises((PermissionError, RuntimeError)):
        await provider.entries_search("青杉 林岚", context=KnowledgeRequestContext("alice", "lead_agent"))
    assert seen == ["青杉 林岚", "青杉"]


@pytest.mark.asyncio
async def test_wiki_nonempty_phrase_does_not_fan_out(adapter):
    provider, calls, state, *_ = adapter
    state["pages"] = [{"knowledge_base_id": "kb-1", "status": "published", "title": "青杉", "summary": "evidence"}]
    assert (await provider.entries_search("青杉 林岚", context=KnowledgeRequestContext("alice", "lead_agent"))).hits
    assert len([r for r in calls if r.url.path.endswith("/wiki/search")]) == 1


@pytest.mark.asyncio
@pytest.mark.parametrize(("query", "expected"), [("青杉", ["青杉"]), ("a b", ["a b"]), ("青杉 青杉 林岚", ["青杉 青杉 林岚", "青杉", "林岚"])])
async def test_wiki_empty_fallback_never_repeats_queries_or_sends_short_terms(adapter, query, expected):
    provider, calls, *_ = adapter
    assert not (await provider.entries_search(query, context=KnowledgeRequestContext("alice", "lead_agent"))).hits
    assert [r.url.params["q"] for r in calls if r.url.path.endswith("/wiki/search")] == expected