From 6bc1020d348ca9293fc4c644fdb71c90fa5f8733 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E4=B8=80=E5=B8=86?= Date: Thu, 15 Oct 2026 15:10:39 +0800 Subject: [PATCH] feat(knowledge): decouple scope and read-only catalogs (phase 2) (#6904) * feat(knowledge): decouple scope and read-only catalogs * docs(knowledge): keep frontend guidance within repository budget * fix(knowledge): align execution identity and harden provider boundaries * docs: keep knowledge guidance within CI size budgets --------- Co-authored-by: Willem Jiang --- README.md | 14 + backend/app/gateway/AGENTS.md | 2 +- .../app/gateway/knowledge_scope_admission.py | 14 +- backend/app/gateway/langgraph_auth.py | 16 +- backend/app/gateway/routers/features.py | 16 +- backend/app/gateway/routers/knowledge.py | 217 +++------ backend/app/gateway/services.py | 1 + backend/docs/KNOWLEDGE.md | 129 +++++- .../deerflow_extension_api/__init__.py | 2 +- .../deerflow_extension_api/contracts.py | 10 + backend/packages/extension-api/pyproject.toml | 2 +- .../deerflow/community/lightrag/tools.py | 13 +- .../deerflow/community/ragflow/AGENTS.md | 5 + .../deerflow/community/ragflow/client.py | 6 + .../deerflow/community/ragflow/formatting.py | 2 +- .../deerflow/community/ragflow/provider.py | 96 ++++ .../harness/deerflow/extensions/AGENTS.md | 11 + .../harness/deerflow/extensions/registry.py | 17 + .../harness/deerflow/knowledge/__init__.py | 32 +- .../harness/deerflow/knowledge/contract.py | 59 ++- .../harness/deerflow/knowledge/providers.py | 106 +++++ .../harness/deerflow/knowledge/tools.py | 75 ++++ backend/packages/harness/pyproject.toml | 2 +- .../blocking_io/test_knowledge_router.py | 9 +- backend/tests/test_extension_api_contracts.py | 2 +- .../test_gateway_knowledge_scope_admission.py | 2 +- backend/tests/test_knowledge_contract.py | 22 + backend/tests/test_knowledge_provider.py | 417 ++++++++++++++++++ .../tests/test_knowledge_provider_router.py | 117 +++++ backend/tests/test_knowledge_router.py | 5 +- backend/tests/test_langgraph_auth.py | 19 + backend/tests/test_lightrag_tools.py | 28 ++ backend/tests/test_ragflow_client.py | 32 ++ .../tests/test_retrieval_plugin_artifacts.py | 39 ++ backend/tests/test_subagent_executor.py | 30 ++ backend/uv.lock | 2 +- examples/deerflow-extension-knowledge/LICENSE | 22 + .../deerflow-extension-knowledge/README.md | 44 ++ .../deerflow_extension_knowledge/__init__.py | 51 +++ .../pyproject.toml | 25 ++ frontend/src/AGENTS.md | 14 +- .../workspace/knowledge-scope-selector.tsx | 65 ++- frontend/src/core/extensions/catalog.ts | 5 + frontend/src/core/i18n/locales/en-US.ts | 5 + frontend/src/core/i18n/locales/types.ts | 1 + frontend/src/core/i18n/locales/zh-CN.ts | 4 + frontend/src/core/knowledge/scope-api.ts | 28 ++ frontend/tests/e2e/bookmark-plugin.spec.ts | 62 ++- frontend/tests/e2e/utils/mock-api.ts | 18 + .../agent-tool-groups-knowledge.dom.test.tsx | 39 +- .../citations/knowledge-evidence.dom.test.tsx | 55 +++ .../knowledge-scope-selector.dom.test.tsx | 94 ++++ .../tests/unit/core/knowledge/scope.test.ts | 35 ++ 53 files changed, 1910 insertions(+), 228 deletions(-) create mode 100644 backend/packages/harness/deerflow/community/ragflow/provider.py create mode 100644 backend/packages/harness/deerflow/knowledge/providers.py create mode 100644 backend/packages/harness/deerflow/knowledge/tools.py create mode 100644 backend/tests/test_knowledge_provider.py create mode 100644 backend/tests/test_knowledge_provider_router.py create mode 100644 examples/deerflow-extension-knowledge/LICENSE create mode 100644 examples/deerflow-extension-knowledge/README.md create mode 100644 examples/deerflow-extension-knowledge/deerflow_extension_knowledge/__init__.py create mode 100644 examples/deerflow-extension-knowledge/pyproject.toml create mode 100644 frontend/tests/unit/components/workspace/citations/knowledge-evidence.dom.test.tsx diff --git a/README.md b/README.md index 7f17701392..6a412f151b 100644 --- a/README.md +++ b/README.md @@ -1390,6 +1390,20 @@ Existing web citations and the text-only Python `knowledge_search(query)` APIs remain supported. No knowledge management service or page is required. See [the knowledge contract guide](backend/docs/KNOWLEDGE.md). +Knowledge scope selection also works with operator-installed read-only Provider +extensions. The shared selector shows only the filters a Provider supports: +knowledge-base selection requires a catalog and dataset filtering; file +selection additionally requires document listing and document filtering. +Unsupported or inaccessible filters fail without expanding the search scope. +Choosing no knowledge search stops retrieval before loading the Provider. +Provider resource IDs are preserved up to 256 Unicode code points; longer IDs +are rejected without truncation. +RAGFlow keeps its existing configuration and citations, and search-only +Providers remain usable without a catalog or management page. See the +[offline Provider example](examples/deerflow-extension-knowledge/README.md) +for extension registration and configuration. WeKnora's adapter is planned for +the next phase. + Skills are what make DeerFlow do *almost anything*. A standard Agent Skill is a structured capability module — a Markdown file that defines a workflow, best practices, and references to supporting resources. DeerFlow ships with built-in skills for research, report generation, slide creation, web pages, image and video generation, and more. But the real power is extensibility: add your own skills, replace the built-in ones, or combine them into compound workflows. diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index 277423ef79..adf6655908 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -119,7 +119,7 @@ owner-scoped assistant version selection remains enabled. | **Models** (`/api/models`) | `GET /` - list models; `GET /{name}` - model details | | **Features** (`/api/features`) | `GET /` - UI capabilities: hot-reloaded agents, guarded browser, startup MCP tasks, separate batch repository/worker states so history stays readable without a worker, `conversation_references` (`read_conversation` configured, per-run reference cap), knowledge scope selection, `scheduled_tasks`, and `thread_activity` | | **Thread Activity** (`/api/thread-activity`) | `GET /?cursor=` - per-user feed over `runs.change_seq`: no cursor seeds; lists threads of the caller's server-originated runs, passes interactive rows; `read_version`. `POST /api/threads/{id}/read` and search `unread` use `thread_read_markers`. SQL only | -| **Knowledge** (`/api/knowledge/retrieval-catalog`) | Authenticated, allowlist-safe, read-only dataset/document catalog used only by main and custom-agent chat scope selection; knowledge management remains in RAGFlow | +| **Knowledge** (`/api/knowledge`) | Authenticated, capability-driven read-only retrieval catalog and per-assistant `/capabilities`; provider adapters own resource authorization and mapping. Unsupported filters fail admission/execution without widening scope; management stays outside core | | **Console** (`/api/console`) | Read-only cross-thread observability for the current user: `GET /stats` - headline counters (runs/threads/agents/tokens/cost); `GET /runs` - paginated run history joined with thread titles (per-run cost); `GET /usage` - zero-filled daily token series + per-model breakdown with spend. Queries `runs`/`threads_meta` directly (no new `RunStore` methods); SQL only (503 on `database.backend: memory`). Real-cost estimation reads optional `models[*].pricing` (`currency`, `input_per_million`, `output_per_million`, `input_cache_hit_per_million`; `ModelConfig` is `extra="allow"`, so no schema change) and prices each run from its `token_usage_by_model` input/output split. Pricing is **cache-aware**: `RunJournal` accumulates prompt-cache hits from `usage_metadata.input_token_details.cache_read` into a sparse `cache_read_tokens` bucket key (also threaded through `SubagentTokenCollector` → `record_external_llm_usage_records`), and cache-hit input tokens are billed at `input_cache_hit_per_million` (omitted → billed at the miss price, a conservative upper bound). Mixed currencies null cost/currency fields. Legacy rows fall back to run-level totals at `model_name`; unpriced models yield `cost: null` and cost fields are null when no pricing is configured | | **MCP** (`/api/mcp`) | GET /config - raw/masked; PUT /config - bulk; PATCH /config - toggle; POST /config/servers - add; PUT /config/server - replace; DELETE /config/servers/{server_name:path} - bodyless. POST /cache/reset. Validate expanded, save raw; reload/reset; invalid -> 400. | | **MCP Tasks** (`/api/threads/{id}/mcp-tasks`) | `GET /` - current user's durable tasks for one owned thread; `GET /{task_id}` - bounded result/input/status-error/cancellation-error detail, including cancellation attempt count, without remote task IDs or driver configuration | diff --git a/backend/app/gateway/knowledge_scope_admission.py b/backend/app/gateway/knowledge_scope_admission.py index d031a6308c..fb20500f07 100644 --- a/backend/app/gateway/knowledge_scope_admission.py +++ b/backend/app/gateway/knowledge_scope_admission.py @@ -8,13 +8,14 @@ from fastapi import HTTPException from langchain_core.messages import BaseMessage, HumanMessage from pydantic import ValidationError +from deerflow.knowledge.providers import RAGFLOW_KNOWLEDGE_TOOL, get_knowledge_capabilities, validate_scope_capabilities from deerflow.knowledge_scope import ( KNOWLEDGE_SCOPE_KEY, canonicalize_knowledge_scope, execution_scope, ) -RAGFLOW_KNOWLEDGE_SEARCH_PROVIDER = "deerflow.community.ragflow.tools:knowledge_search_tool" +RAGFLOW_KNOWLEDGE_SEARCH_PROVIDER = RAGFLOW_KNOWLEDGE_TOOL # Legacy import compatibility. def assistant_supports_knowledge_scope( @@ -22,6 +23,7 @@ def assistant_supports_knowledge_scope( assistant_id: str | None, app_config: Any, agent_config: Any | None, + extensions: Any = None, ) -> bool: """Return whether this exact assistant/provider pairing is supported.""" if not assistant_id: @@ -29,9 +31,7 @@ def assistant_supports_knowledge_scope( knowledge_base = getattr(app_config, "knowledge_base", None) if not getattr(knowledge_base, "enabled", False): return False - get_tool_config = getattr(app_config, "get_tool_config", None) - tool = get_tool_config("knowledge_search") if callable(get_tool_config) else None - if getattr(tool, "use", None) != RAGFLOW_KNOWLEDGE_SEARCH_PROVIDER: + if not get_knowledge_capabilities(app_config, extensions=extensions).supports("search"): return False # The main assistant has no custom-agent config row. Its knowledge tool is # controlled solely by the app-level provider configuration. @@ -60,6 +60,7 @@ def admit_message_knowledge_scope( agent_config: Any | None, recovery_scope: object | None = None, recovery: bool = False, + extensions: Any = None, ) -> dict[str, Any] | None: """Canonicalize the sole eligible HumanMessage and return execution scope. @@ -115,6 +116,7 @@ def admit_message_knowledge_scope( assistant_id=assistant_id, app_config=app_config, agent_config=agent_config, + extensions=extensions, ): raise HTTPException( status_code=422, @@ -127,6 +129,10 @@ def admit_message_knowledge_scope( status_code=422, detail=f"Invalid knowledge_scope: {exc.errors()[0]['msg']}", ) from exc + try: + validate_scope_capabilities(canonical, get_knowledge_capabilities(app_config, extensions=extensions)) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from None elif scoped_indexes and not recovery: raise HTTPException(status_code=422, detail="knowledge_scope must be an object") diff --git a/backend/app/gateway/langgraph_auth.py b/backend/app/gateway/langgraph_auth.py index c7745d2398..64b9bfaead 100644 --- a/backend/app/gateway/langgraph_auth.py +++ b/backend/app/gateway/langgraph_auth.py @@ -71,7 +71,9 @@ def _scrub_run_incarnation(value: dict) -> None: if not isinstance(context, dict): context = {} kwargs["context"] = context - runtime_keys = (*lifecycle_keys, "user_id", "thread_id", "run_id") + # Child execution identity is installed only by SubagentExecutor. Public + # runs must not impersonate a child when providers evaluate agent access. + runtime_keys = (*lifecycle_keys, "user_id", "thread_id", "run_id", "is_subagent", "agent_id") for key in runtime_keys: context.pop(key, None) config = kwargs.get("config") @@ -198,6 +200,10 @@ async def _bind_standalone_run_incarnation(ctx, value: dict) -> None: assert is_valid_thread_incarnation(incarnation) context = value["kwargs"]["context"] + # Assistant defaults are merged after admission. Explicit primary-run + # values also override forged fields saved by older server versions. + context["is_subagent"] = False + context["agent_id"] = None context["user_id"] = ctx.user.identity if thread_id is not None: context["thread_id"] = str(thread_id) @@ -330,6 +336,14 @@ async def add_owner_filter(ctx: Auth.types.AuthContext, value: dict): metadata.pop(THREAD_INCARNATION_CONTEXT_KEY, None) if ctx.resource == "assistants" and ctx.action in {"create", "update"}: metadata["created_by"] = "user" + sections = [value.get("context")] + config = value.get("config") + if isinstance(config, dict): + sections.extend(config.get(key) for key in ("context", "configurable", "metadata")) + for section in sections: + if isinstance(section, dict): + section.pop("is_subagent", None) + section.pop("agent_id", None) # Return filter dict — LangGraph applies it to search/read/delete return {"user_id": ctx.user.identity} diff --git a/backend/app/gateway/routers/features.py b/backend/app/gateway/routers/features.py index a155b5ab18..f42e5af214 100644 --- a/backend/app/gateway/routers/features.py +++ b/backend/app/gateway/routers/features.py @@ -13,10 +13,10 @@ from pydantic import BaseModel, Field from app.gateway.browser_capability import browser_capability from app.gateway.conversation_access import conversation_references_enabled from app.gateway.deps import get_config -from app.gateway.knowledge_scope_admission import RAGFLOW_KNOWLEDGE_SEARCH_PROVIDER from app.gateway.routers.thread_activity import thread_activity_available from app.gateway.run_models import MAX_CONVERSATION_REFERENCES from deerflow.config.app_config import AppConfig +from deerflow.knowledge.providers import get_knowledge_capabilities from deerflow.scheduler.runtime import scheduler_tools_enabled from deerflow.subagents.capacity import configured_subagent_max_running @@ -58,11 +58,11 @@ class ConversationReferencesFeature(BaseModel): class KnowledgeBaseFeature(BaseModel): - """Availability of RAGFlow retrieval scope selection in chat.""" + """Availability of capability-driven retrieval scope selection in chat.""" scope_selection_enabled: bool = Field( ..., - description="Whether chat may select a per-message RAGFlow retrieval scope", + description="Whether chat may select a per-message knowledge retrieval scope", ) @@ -129,7 +129,7 @@ async def list_features(request: Request, config: AppConfig = Depends(get_config max_references=MAX_CONVERSATION_REFERENCES, ), knowledge_base=KnowledgeBaseFeature( - scope_selection_enabled=_knowledge_scope_selection_enabled(config), + scope_selection_enabled=_knowledge_scope_selection_enabled(config, getattr(request.app.state, "extensions", None)), ), scheduled_tasks=_scheduled_tasks_feature(request, config), # Startup-scoped: the read repository exists only with SQL persistence. @@ -154,10 +154,10 @@ def _scheduled_tasks_feature(request: Request, config: AppConfig) -> ScheduledTa ) -def _knowledge_scope_selection_enabled(config: AppConfig) -> bool: - """Fail closed unless the effective knowledge_search entry is RAGFlow.""" +def _knowledge_scope_selection_enabled(config: AppConfig, extensions=None) -> bool: + """Selection UI requires both a catalog and dataset filtering.""" settings = config.knowledge_base if not settings.enabled or not settings.scope_selection_enabled: return False - tool = config.get_tool_config("knowledge_search") - return tool is not None and tool.use == RAGFLOW_KNOWLEDGE_SEARCH_PROVIDER + capabilities = get_knowledge_capabilities(config, extensions=extensions) + return capabilities.supports("catalog") and capabilities.supports("dataset_filter") diff --git a/backend/app/gateway/routers/knowledge.py b/backend/app/gateway/routers/knowledge.py index a60ea4998c..276c16f456 100644 --- a/backend/app/gateway/routers/knowledge.py +++ b/backend/app/gateway/routers/knowledge.py @@ -1,4 +1,4 @@ -"""Authenticated, read-only RAGFlow catalog for chat retrieval scope.""" +"""Authenticated, provider-neutral, read-only retrieval catalog.""" from __future__ import annotations @@ -8,106 +8,79 @@ from collections.abc import Awaitable from typing import Annotated, Any from fastapi import APIRouter, Depends, HTTPException, Path, Query, Request +from pydantic import ValidationError from app.gateway.authz import require_permission from app.gateway.deps import get_config from app.gateway.knowledge_scope_admission import assistant_supports_knowledge_scope -from deerflow.community.ragflow.client import ( - RAGFlowAPIError, - RAGFlowConnectionError, - RAGFlowProtocolError, -) -from deerflow.community.ragflow.tools import ( - build_ragflow_retrieval_client as _build_retrieval_client, -) -from deerflow.community.ragflow.tools import ( - resolve_ragflow_datasets, - resolve_ragflow_retrieval_settings, -) from deerflow.config.agents_config import load_agent_config from deerflow.config.app_config import AppConfig +from deerflow.knowledge import KnowledgeRequestContext, RetrievalCatalogPage +from deerflow.knowledge.providers import get_knowledge_capabilities, resolve_knowledge_provider from deerflow.runtime.user_context import get_effective_user_id logger = logging.getLogger(__name__) - router = APIRouter(prefix="/api/knowledge", tags=["knowledge"]) -_DatasetId = Annotated[ - str, - Path(min_length=1, max_length=256, pattern=r"^[A-Za-z0-9_-]+$"), -] +# Scope IDs are provider-owned opaque strings, not RAGFlow UUIDs. +_DatasetId = Annotated[str, Path(min_length=1, max_length=256)] _SCOPE_UNAVAILABLE_DETAIL = "Knowledge scope selection is unavailable for this assistant." -async def _catalog_result[Result](operation: Awaitable[Result]) -> Result: - """Normalize provider failures without exposing credentials or raw payloads.""" +async def _catalog_result(operation: Awaitable[Any]) -> dict[str, Any]: + """Validate the public projection and contain provider exception payloads.""" try: - return await operation - except RAGFlowAPIError as exc: - logger.warning("RAGFlow rejected a retrieval-catalog request (code=%s)", exc.code) - raise HTTPException( - status_code=502, - detail="RAGFlow rejected the retrieval-catalog request.", - ) from None - except RAGFlowConnectionError: - logger.warning("RAGFlow retrieval catalog could not connect") - raise HTTPException(status_code=502, detail="Unable to connect to RAGFlow.") from None - except RAGFlowProtocolError: - logger.warning("RAGFlow returned an invalid retrieval-catalog response") - raise HTTPException( - status_code=502, - detail="RAGFlow returned an invalid retrieval-catalog response.", - ) from None + result = await operation + payload = result.model_dump() if isinstance(result, RetrievalCatalogPage) else result + return RetrievalCatalogPage.model_validate(payload).model_dump() + except PermissionError: + raise HTTPException(status_code=403, detail="Knowledge resource access denied.") from None + except FileNotFoundError: + raise HTTPException(status_code=404, detail="Knowledge base not found.") from None + except ValidationError: + raise HTTPException(status_code=502, detail="Knowledge provider returned an invalid catalog.") from None + except ValueError: + raise HTTPException(status_code=409, detail="The configured knowledge-base scope is unavailable.") from None except Exception as exc: - logger.error("Unexpected RAGFlow retrieval-catalog failure (%s)", type(exc).__name__) - raise HTTPException(status_code=502, detail="RAGFlow request failed.") from None + logger.warning("Knowledge retrieval-catalog request failed (%s)", type(exc).__name__) + raise HTTPException(status_code=502, detail="Knowledge provider request failed.") from None -async def _scope_catalog(config: AppConfig, agent_name: str): - knowledge_base = config.knowledge_base +async def _scope_capabilities(config: AppConfig, agent_name: str, extensions: Any = None): agent_config = None if agent_name != "lead_agent": try: - # The agent store reads files or makes a sync DB round trip. - agent_config = await asyncio.to_thread( - load_agent_config, - agent_name, - user_id=get_effective_user_id(), - ) + agent_config = await asyncio.to_thread(load_agent_config, agent_name, user_id=get_effective_user_id()) except (FileNotFoundError, ValueError): raise HTTPException(status_code=404, detail="Custom agent not found.") from None - if ( - not knowledge_base.enabled - or not knowledge_base.scope_selection_enabled - or not assistant_supports_knowledge_scope( - assistant_id=agent_name, - app_config=config, - agent_config=agent_config, - ) - ): + if not assistant_supports_knowledge_scope(assistant_id=agent_name, app_config=config, agent_config=agent_config, extensions=extensions): raise HTTPException(status_code=409, detail=_SCOPE_UNAVAILABLE_DETAIL) - settings, error = resolve_ragflow_retrieval_settings(config) - if settings is None: - logger.warning( - "RAGFlow retrieval catalog settings are unavailable (%s)", - error, - ) - raise HTTPException( - status_code=503, - detail="Knowledge retrieval is not configured.", - ) - return settings + return get_knowledge_capabilities(config, extensions=extensions) -def _catalog_page( - items: list[dict[str, Any]], - *, - page: int, - page_size: int, -) -> tuple[list[dict[str, Any]], int]: - total = len(items) - start = (page - 1) * page_size - return items[start : start + page_size], total +async def _scope_catalog(config: AppConfig, agent_name: str, request: Request, operation: str): + extensions = getattr(request.app.state, "extensions", None) + capabilities = await _scope_capabilities(config, agent_name, extensions) + if not config.knowledge_base.scope_selection_enabled or not capabilities.supports(operation): + raise HTTPException(status_code=409, detail=_SCOPE_UNAVAILABLE_DETAIL) + try: + return resolve_knowledge_provider(config, extensions=extensions) + except LookupError: + raise HTTPException(status_code=503, detail="Knowledge retrieval is not configured.") from None + + +@router.get("/capabilities") +@require_permission("threads", "read") +async def knowledge_capabilities( + request: Request, + agent_name: Annotated[str, Query(min_length=1, max_length=128)], + config: AppConfig = Depends(get_config), +) -> dict[str, Any]: + capabilities = await _scope_capabilities(config, agent_name, getattr(request.app.state, "extensions", None)) + return { + "operations": sorted(capabilities.operations), + "scope_selection_enabled": config.knowledge_base.scope_selection_enabled and capabilities.supports("catalog") and capabilities.supports("dataset_filter"), + } @router.get("/retrieval-catalog/datasets") @@ -120,42 +93,11 @@ async def list_retrieval_catalog_datasets( search: Annotated[str, Query(max_length=256)] = "", config: AppConfig = Depends(get_config), ) -> dict[str, Any]: - """Return only datasets that the operator permits this agent to retrieve.""" - settings = await _scope_catalog(config, agent_name) - client = _build_retrieval_client(settings) - datasets, error = await _catalog_result( - resolve_ragflow_datasets(client, settings), - ) - if datasets is None: - logger.warning( - "RAGFlow retrieval catalog could not resolve operator scope (%s)", - error, - ) - raise HTTPException( - status_code=409, - detail="The configured knowledge-base scope is unavailable.", - ) - - needle = search.strip().casefold() - entries = [ - { - "id": dataset.dataset_id, - "name": dataset.name, - "selectable": bool(dataset.embedding_model) and dataset.chunk_count != 0, - } - for dataset in datasets - if not needle or needle in dataset.name.casefold() - ] - selected, total = _catalog_page(entries, page=page, page_size=page_size) - return { - "items": selected, - "page": page, - "page_size": page_size, - "total": total, - } + provider = await _scope_catalog(config, agent_name, request, "catalog") + return await _catalog_result(provider.catalog(context=KnowledgeRequestContext(user_id=get_effective_user_id(), agent_name=agent_name), page=page, page_size=page_size, search=search)) -@router.get("/retrieval-catalog/datasets/{dataset_id}/documents") +@router.get("/retrieval-catalog/datasets/{dataset_id:path}/documents") @require_permission("threads", "read") async def list_retrieval_catalog_documents( dataset_id: _DatasetId, @@ -166,58 +108,5 @@ async def list_retrieval_catalog_documents( search: Annotated[str, Query(max_length=256)] = "", config: AppConfig = Depends(get_config), ) -> dict[str, Any]: - """Return a normalized, read-only document page inside operator scope.""" - settings = await _scope_catalog(config, agent_name) - if settings.datasets is not None and dataset_id not in set(settings.datasets): - raise HTTPException(status_code=404, detail="Knowledge base not found.") - client = _build_retrieval_client(settings) - resolved, error = await _catalog_result( - resolve_ragflow_datasets(client, settings, [dataset_id]), - ) - if not resolved: - logger.warning( - "RAGFlow retrieval catalog dataset is unavailable (dataset_id=%s, reason=%s)", - dataset_id, - error, - ) - raise HTTPException(status_code=404, detail="Knowledge base not found.") - - params = [("page", str(page)), ("page_size", str(page_size))] - if search.strip(): - params.append(("keywords", search.strip())) - payload = await _catalog_result( - client.list_documents(dataset_id, params=params), - ) - data = payload.get("data") - docs = data.get("docs") if isinstance(data, dict) else None - if not isinstance(docs, list): - raise HTTPException( - status_code=502, - detail="RAGFlow returned an invalid document list.", - ) - items = [] - for document in docs: - if not isinstance(document, dict): - continue - document_id = document.get("id") - if not isinstance(document_id, str) or not document_id.strip(): - continue - chunk_count = document.get("chunk_count") - searchable = isinstance(chunk_count, int) and not isinstance(chunk_count, bool) and chunk_count > 0 - parsed = document.get("run") == "DONE" - items.append( - { - "id": document_id.strip(), - "name": str(document.get("name") or "Unnamed document"), - "selectable": parsed and searchable, - } - ) - total = data.get("total") - if not isinstance(total, int) or isinstance(total, bool) or total < 0: - total = len(items) - return { - "items": items, - "page": page, - "page_size": page_size, - "total": total, - } + provider = await _scope_catalog(config, agent_name, request, "documents") + return await _catalog_result(provider.documents(dataset_id, context=KnowledgeRequestContext(user_id=get_effective_user_id(), agent_name=agent_name), page=page, page_size=page_size, search=search)) diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py index 47aa629a5a..a5a7096366 100644 --- a/backend/app/gateway/services.py +++ b/backend/app/gateway/services.py @@ -2071,6 +2071,7 @@ async def start_run( agent_config=agent_config, recovery_scope=recovery_scope, recovery=is_scope_recovery, + extensions=getattr(request.app.state, "extensions", None), ) if admitted_knowledge_scope is not None: await _validate_scope_thread_binding( diff --git a/backend/docs/KNOWLEDGE.md b/backend/docs/KNOWLEDGE.md index 169cc84be7..7ec9ec5aae 100644 --- a/backend/docs/KNOWLEDGE.md +++ b/backend/docs/KNOWLEDGE.md @@ -1,10 +1,131 @@ -# Read-only knowledge contract (#6269, phase 1) +# Read-only knowledge contract (#6269, phases 1–2) `deerflow.knowledge` owns `RetrievalResult`, `RetrievalHit`, `KnowledgeSource`, `KnowledgeCapabilities`, and the search-only `RetrievalProvider` protocol. -These depend on no knowledge management service, resource store, or catalog. -Existing RAGFlow scope admission and selection UI remain authoritative. -The shared contract does not add scope selection or WeKnora support to LightRAG. +Search-only providers need no knowledge management service, resource store, +catalog, or scope-filter implementation. Scope admission, the read-only catalog +and selection UI consume the same operation-level capabilities. LightRAG still +does not advertise dataset/document filters; WeKnora remains phase 3. + +## Provider registration and optional operations + +An operator-installed extension calls +`registry.knowledge_provider(name, provider)` during `install()`. The method +returns `False` on older hosts and is supported by extension-api **0.2.7**. +The provider implements the existing `RetrievalProvider.search(query)` and +advertises `KnowledgeCapabilities`. Registration validates each advertised +operation, rejects duplicate/reserved names, and participates in source +attribution and positional install rollback. Connection settings, response +mapping and credentials belong to that extension's `plugins[].config`. +The contract package remains dependency-free; providers use the published +`deerflow.knowledge` models from the harness. + +| Operation | Optional interface | Behavior | +| --- | --- | --- | +| `search` | `RetrievalProvider.search` | Required; returns `RetrievalResult` | +| `sources` | Existing result/source models | Advertises attributable evidence | +| `catalog` | `RetrievalCatalogProvider.catalog` | Paginated accessible datasets | +| `documents` | `RetrievalDocumentsProvider.documents` | Paginated documents in an accessible dataset | +| `dataset_filter` | `ScopedRetrievalProvider.search_scoped` | Enforces selected dataset IDs | +| `document_filter` | `ScopedRetrievalProvider.search_scoped` | Enforces document IDs within selected datasets | + +Both catalog methods receive `context`, `page`, `page_size` and `search`, and +return `RetrievalCatalogPage` with `{id, name, selectable}` items. IDs are opaque +provider identifiers, including non-UUID identifiers. Gateway validates this +bounded public projection and never returns raw provider payloads. +Dataset and document IDs are limited to **256 Unicode code points** throughout +catalog output, URL parameters and scope snapshots. Overlong IDs are rejected, +never truncated. A phase 3 WeKnora adapter must map space-qualified resources +to stable, unambiguous IDs within this bound and resolve them inside the +provider; the earlier `ks1_` encoding allowing 512 characters is not the public +contract. Reuse provider API/mapping logic without adding a second discovery +protocol or a DeerFlow ACL store. + +`KnowledgeRequestContext` supplies the host user ID, Agent name and optional +execution-only scope. A provider implementing `search_scoped` receives context +on every search, including unrestricted searches. A search-only implementation +may retain `search(query)` when it uses an operator-owned connection with a +fixed resource visibility. User/tenant mappings and resource authorization +remain provider responsibilities; DeerFlow does not infer tenants from client +metadata or create a second ACL store. +`agent_name` identifies the current executor: the selected assistant name, +defaulting to `lead_agent`, or the executing subagent's configured name from +its server runtime `agent_id`. It does not identify the parent assistant. +Runtime context takes precedence over legacy `configurable.agent_name` for +primary runs, matching Agent assembly. Public Gateway and standalone LangGraph +run admission discard caller-supplied `is_subagent` and `agent_id`; only the +internal executor can assign child identity. +Standalone primary-run admission explicitly overrides persisted assistant +child-identity defaults, and assistant writes discard these reserved fields. +Catalog requests use the selected assistant name under the same convention; +direct provider callers outside the host runtime may omit it. Delegation +preserves the parent user identity and execution scope, while the provider +must enforce access for the executing agent on every call. + +The Agent tool entry for registered providers is: + +```yaml +knowledge_base: + enabled: true + scope_selection_enabled: true +tools: + - name: knowledge_search + group: knowledge + use: deerflow.knowledge.tools:knowledge_search_tool + provider: my-provider +``` + +`provider` selects an already registered operator-controlled name. The core tool +validates normalized results and uses phase 1's source formatter/budget. The +run's AppConfig and immutable extension snapshot take precedence over global +configuration and the global registry; +ordinary and batch subagents retain their inherited execution scope. Existing +RAGFlow and LightRAG tool paths remain valid. RAGFlow's compatibility adapter +owns its API calls, operator allowlist, searchable-document checks and catalog +mapping; native citation writers and direct text APIs remain unchanged. +Its HTTP client accepts only ASCII letters, digits, underscores and hyphens +in dataset IDs used as path segments. Traversal and encoded URL delimiters +are rejected before any document-list request, for both browsing and scope +validation. Other providers retain their own opaque-ID syntax. + +See [the packaged synthetic Provider example](../../examples/deerflow-extension-knowledge/README.md) +for an offline catalog → selected document → retrieval → captured source flow. +The existing `ModelTool(response_format="content_and_artifact")` and +`PluginToolResult` channel from #6861 is unchanged. Custom ModelTools can still +emit the phase 1 envelope; artifact evidence must pass core validation. + +## Scope admission and selection + +`GET /api/knowledge/capabilities?agent_name=...` returns supported operations for +the exact assistant and whether selection UI is enabled. `/api/features` enables +the shared selector only when `knowledge_base.enabled` and +`scope_selection_enabled` are true and the provider supports both `catalog` and +`dataset_filter`. The selector exposes files only with both `documents` and +`document_filter`. It preserves older selected documents and blocks Apply if +the current provider cannot enforce them. Capability failures do not enable +catalog queries. Historical scope labels remain display-only. + +The existing authenticated endpoints remain: + +- `GET /api/knowledge/retrieval-catalog/datasets` +- `GET /api/knowledge/retrieval-catalog/datasets/{dataset_id}/documents` + +Catalog operations require `threads:read`, the assistant's knowledge tool +access, the operator UI flag and the advertised operation. Providers return +only resources accessible to their connection/current user. A denied resource +returns 403; missing datasets return 404; unsupported operations return 409; +unconfigured retrieval returns 503; invalid provider output/failures return 502. +Errors do not disclose provider credentials or raw exception messages. + +Scope stays in the existing version-one HumanMessage snapshot and server-owned +runtime key. Admission rejects unsupported dataset/document filters with 422, +independently of UI availability. Execution checks capabilities again before +calling the provider. Selected IDs constrain retrieval but never grant access; +missing/inaccessible resources fail the complete constrained request, with no +unrestricted retry. `all` means the provider's accessible operator scope and +`disabled` blocks knowledge search. Search-only providers can accept these +nonfiltering modes without catalog support. Regeneration and resume recover +the original server snapshot; they do not adopt a changed Agent default. ## Sources and citations diff --git a/backend/packages/extension-api/deerflow_extension_api/__init__.py b/backend/packages/extension-api/deerflow_extension_api/__init__.py index 9f68b8acc0..4ff7481f28 100644 --- a/backend/packages/extension-api/deerflow_extension_api/__init__.py +++ b/backend/packages/extension-api/deerflow_extension_api/__init__.py @@ -99,7 +99,7 @@ from deerflow_extension_api.state import ExtensionData #: Contract version. Before 1.0, minors may break and patches are additive. #: From 1.0 on, bump the major for breaking changes. -API_VERSION = "0.2.6" +API_VERSION = "0.2.7" __all__ = [ "AGENT_RUNS_CONTEXT_KEY", diff --git a/backend/packages/extension-api/deerflow_extension_api/contracts.py b/backend/packages/extension-api/deerflow_extension_api/contracts.py index 98010dff30..3ef0e849d2 100644 --- a/backend/packages/extension-api/deerflow_extension_api/contracts.py +++ b/backend/packages/extension-api/deerflow_extension_api/contracts.py @@ -196,6 +196,16 @@ class ExtensionRegistry(Protocol): """Return True when accepted; False means this host lacks plugin UI support.""" return False + def knowledge_provider(self, name: str, provider: Any) -> bool: + """Register a read-only ``deerflow.knowledge.RetrievalProvider``. + + The host validates its KnowledgeCapabilities and optional catalog, + documents and contextual search implementations. The value stays Any + here to keep extension-api independent of the harness and Pydantic. + False means the host does not support this contribution. + """ + return False + def middlewares(self, contributor: MiddlewareContributor) -> None: return None diff --git a/backend/packages/extension-api/pyproject.toml b/backend/packages/extension-api/pyproject.toml index 04e2410206..223b4dda19 100644 --- a/backend/packages/extension-api/pyproject.toml +++ b/backend/packages/extension-api/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "deerflow-extension-api" -version = "0.2.6" +version = "0.2.7" description = "Public contracts for DeerFlow extensions" requires-python = ">=3.12" # Keep the contract package import-light and independent from the host. Public diff --git a/backend/packages/harness/deerflow/community/lightrag/tools.py b/backend/packages/harness/deerflow/community/lightrag/tools.py index b980a41f20..7e4d82f63e 100644 --- a/backend/packages/harness/deerflow/community/lightrag/tools.py +++ b/backend/packages/harness/deerflow/community/lightrag/tools.py @@ -12,7 +12,10 @@ from pydantic import AnyHttpUrl, BaseModel, ConfigDict, Field, SecretStr, Valida from deerflow.community.provider_credentials import redact_secret, reject_url_userinfo, resolve_api_key from deerflow.config import get_app_config from deerflow.knowledge import format_result +from deerflow.knowledge.providers import validate_scope_capabilities from deerflow.knowledge.sources import redact_strings, source_artifact +from deerflow.knowledge_scope import KNOWLEDGE_SCOPE_RUNTIME_KEY +from deerflow.tools.types import Runtime from .client import LightRAGAPIError, LightRAGClient, LightRAGConnectionError, LightRAGProtocolError from .formatting import CAPABILITIES, format_retrieval_result, normalize_retrieval_result @@ -147,12 +150,20 @@ def _tool_description() -> str: return f"{base} Internal identifiers are never shown to the model." -async def _knowledge_search_entrypoint(query: str) -> tuple[str, dict]: +async def _knowledge_search_entrypoint(query: str, runtime: Runtime = None) -> tuple[str, dict | None]: """Search the operator-configured LightRAG knowledge base. Args: query: Specific question or search terms to retrieve from the configured private documents. """ + context = getattr(runtime, "context", None) + scope = context.get(KNOWLEDGE_SCOPE_RUNTIME_KEY) if isinstance(context, Mapping) else None + try: + validated = validate_scope_capabilities(scope, CAPABILITIES) if scope is not None else None + if validated is not None and validated["mode"] == "disabled": + return "Error: Knowledge retrieval is disabled for this turn.", None + except ValueError: + return "Error: LightRAG does not support the selected knowledge scope.", None result = await _search(query, with_sources=True) return result if isinstance(result, tuple) else (result, source_artifact([])) diff --git a/backend/packages/harness/deerflow/community/ragflow/AGENTS.md b/backend/packages/harness/deerflow/community/ragflow/AGENTS.md index e1f1d0bc7d..f4aa1b45fc 100644 --- a/backend/packages/harness/deerflow/community/ragflow/AGENTS.md +++ b/backend/packages/harness/deerflow/community/ragflow/AGENTS.md @@ -32,6 +32,11 @@ both before and after the synopsis head/tail split. ## Document validation +`RAGFlowClient.list_documents` validates dataset path segments with +`[A-Za-z0-9_-]+` before HTTP requests. Keep this provider-specific boundary +shared by catalog browsing and selected-document validation; do not apply +RAGFlow ID syntax to the generic knowledge contract. + `tools.py` validates each dataset's selected documents in batches of at most 100 IDs. All batches share the `_bounded_gather` concurrency limit of four. Merge validated IDs in input order to retain the complete retrieval scope; diff --git a/backend/packages/harness/deerflow/community/ragflow/client.py b/backend/packages/harness/deerflow/community/ragflow/client.py index fcee9c88e4..a4782fc8c7 100644 --- a/backend/packages/harness/deerflow/community/ragflow/client.py +++ b/backend/packages/harness/deerflow/community/ragflow/client.py @@ -2,6 +2,7 @@ from __future__ import annotations +import re from typing import Any import httpx @@ -163,6 +164,11 @@ class RAGFlowClient: params: list[tuple[str, str]], ) -> dict[str, Any]: """Proxy one document-list request for a dataset.""" + # Generic knowledge IDs are opaque. RAGFlow IDs must remain a single + # literal segment: reject traversal, URL delimiters and encoded paths + # here so both catalog browsing and selected-document validation agree. + if not isinstance(dataset_id, str) or re.fullmatch(r"[A-Za-z0-9_-]+", dataset_id) is None: + raise ValueError("Invalid RAGFlow dataset ID.") return await self._request("GET", f"/datasets/{dataset_id}/documents", params=params) async def retrieve( diff --git a/backend/packages/harness/deerflow/community/ragflow/formatting.py b/backend/packages/harness/deerflow/community/ragflow/formatting.py index 80a09844a6..ddff2afd16 100644 --- a/backend/packages/harness/deerflow/community/ragflow/formatting.py +++ b/backend/packages/harness/deerflow/community/ragflow/formatting.py @@ -9,7 +9,7 @@ from typing import Any from deerflow.knowledge import KnowledgeCapabilities, KnowledgeSource, RetrievalHit, RetrievalResult from deerflow.knowledge.sources import truncate as _truncate -CAPABILITIES = KnowledgeCapabilities(operations=frozenset({"search", "sources"})) +CAPABILITIES = KnowledgeCapabilities(operations=frozenset({"search", "sources", "catalog", "documents", "dataset_filter", "document_filter"})) def normalize_retrieval_result(result: Mapping[str, Any], *, dataset_names_by_id: Mapping[str, str]) -> RetrievalResult: diff --git a/backend/packages/harness/deerflow/community/ragflow/provider.py b/backend/packages/harness/deerflow/community/ragflow/provider.py new file mode 100644 index 0000000000..335cca7ee7 --- /dev/null +++ b/backend/packages/harness/deerflow/community/ragflow/provider.py @@ -0,0 +1,96 @@ +"""Compatibility adapter for RAGFlow read-only catalog and scope operations.""" + +from typing import Any + +from deerflow.knowledge import KnowledgeRequestContext, RetrievalCatalogItem, RetrievalCatalogPage, RetrievalResult +from deerflow.knowledge.sources import redact_strings + +from .client import RAGFlowProtocolError +from .formatting import CAPABILITIES, normalize_retrieval_result +from .tools import ( + _api_key, + _group_scoped_datasets, + _retrieve_dataset_groups, + _validate_document_filters, + build_ragflow_retrieval_client, + resolve_ragflow_datasets, + resolve_ragflow_retrieval_settings, +) + + +class RAGFlowKnowledgeProvider: + capabilities = CAPABILITIES + + def __init__(self, app_config: Any): + # Construction performs no I/O; credentials remain in the adapter. + self.settings, _ = resolve_ragflow_retrieval_settings(app_config) + if self.settings is None: + raise LookupError("Knowledge retrieval is not configured") + + async def search(self, query: str) -> RetrievalResult: + return await self._search(query, None) + + async def search_scoped(self, query: str, *, context: KnowledgeRequestContext) -> RetrievalResult: + return await self._search(query, context.scope) + + async def _search(self, query: str, scope: dict | None) -> RetrievalResult: + from deerflow.knowledge.providers import validate_scope_capabilities + + if scope is not None: + scope = validate_scope_capabilities(scope, self.capabilities) + if scope["mode"] == "disabled": + raise ValueError("Knowledge search is disabled for this turn") + selected = scope is not None and scope["mode"] == "selected" + client = build_ragflow_retrieval_client(self.settings) + datasets, error = await resolve_ragflow_datasets(client, self.settings, scope["dataset_ids"] if selected else None) + if not datasets or error: + raise PermissionError("Knowledge scope is unavailable") + filters, error = await _validate_document_filters(client, list(scope.get("document_filters") or []) if selected else []) + if error: + raise PermissionError("Knowledge scope is unavailable") + groups = _group_scoped_datasets(datasets, filters or {}) + if not groups: + return RetrievalResult() + result = await _retrieve_dataset_groups(client, self.settings, query, groups) + key = _api_key(self.settings) + return normalize_retrieval_result(redact_strings(result, key), dataset_names_by_id=redact_strings({item.dataset_id: item.name for item in datasets}, key)) + + async def catalog(self, *, context: KnowledgeRequestContext, page: int, page_size: int, search: str) -> RetrievalCatalogPage: + client = build_ragflow_retrieval_client(self.settings) + datasets, error = await resolve_ragflow_datasets(client, self.settings) + if datasets is None or error: + raise ValueError("The configured knowledge-base scope is unavailable") + needle = search.strip().casefold() + items = [RetrievalCatalogItem(id=item.dataset_id, name=item.name, selectable=bool(item.embedding_model) and item.chunk_count != 0) for item in datasets if not needle or needle in item.name.casefold()] + start = (page - 1) * page_size + return RetrievalCatalogPage(items=items[start : start + page_size], page=page, page_size=page_size, total=len(items)) + + async def documents(self, dataset_id: str, *, context: KnowledgeRequestContext, page: int, page_size: int, search: str) -> RetrievalCatalogPage: + if self.settings.datasets is not None and dataset_id not in self.settings.datasets: + raise FileNotFoundError("Knowledge base not found") + client = build_ragflow_retrieval_client(self.settings) + datasets, error = await resolve_ragflow_datasets(client, self.settings, [dataset_id]) + if not datasets or error: + raise FileNotFoundError("Knowledge base not found") + params = [("page", str(page)), ("page_size", str(page_size))] + if search.strip(): + params.append(("keywords", search.strip())) + payload = await client.list_documents(dataset_id, params=params) + data = payload.get("data") + docs = data.get("docs") if isinstance(data, dict) else None + if not isinstance(docs, list): + raise RAGFlowProtocolError("Invalid document list") + items = [] + for document in docs: + if not isinstance(document, dict): + continue + document_id = document.get("id") + if not isinstance(document_id, str) or not document_id.strip(): + continue + count = document.get("chunk_count") + searchable = isinstance(count, int) and not isinstance(count, bool) and count > 0 + items.append(RetrievalCatalogItem(id=document_id.strip(), name=str(document.get("name") or "Unnamed document"), selectable=document.get("run") == "DONE" and searchable)) + total = data.get("total") + if not isinstance(total, int) or isinstance(total, bool) or total < 0: + total = len(items) + return RetrievalCatalogPage(items=items, page=page, page_size=page_size, total=total) diff --git a/backend/packages/harness/deerflow/extensions/AGENTS.md b/backend/packages/harness/deerflow/extensions/AGENTS.md index deb6852106..43846b7f20 100644 --- a/backend/packages/harness/deerflow/extensions/AGENTS.md +++ b/backend/packages/harness/deerflow/extensions/AGENTS.md @@ -1,5 +1,16 @@ ### Python Extension System (Runtime and Distribution) +`registry.knowledge_provider(name, provider)` (extension-api 0.2.7) adds an +operator-installed read-only `deerflow.knowledge.RetrievalProvider`. The host +validates advertised async operations, attributes entries and rolls them back +positionally with every other contribution. Provider configuration, API calls +and response mapping stay in extensions; core owns capabilities, Scope +admission and source validation. Catalog and execution resolve the same loaded +snapshot, with run-bound snapshots taking precedence over the singleton. +See [the contract](../../../../docs/KNOWLEDGE.md) and the packaged +`examples/deerflow-extension-knowledge` sample. Older registry implementations +return False; never silently ignore unsupported registration. + Third-party Python packages can expose an `install(registry, config)` function and be loaded, in deterministic order, from the startup-only top-level `plugins:` list in `config.yaml`. Keep this list out of `extensions_config.json`: the latter is writable diff --git a/backend/packages/harness/deerflow/extensions/registry.py b/backend/packages/harness/deerflow/extensions/registry.py index e123f9e34d..8ddf7cc0d6 100644 --- a/backend/packages/harness/deerflow/extensions/registry.py +++ b/backend/packages/harness/deerflow/extensions/registry.py @@ -56,6 +56,7 @@ class LoadedExtensions: services: tuple[tuple[str, ExtensionService], ...] = () routers: tuple[tuple[str, Any], ...] = () plugins: tuple[tuple[str, PluginContribution], ...] = () + knowledge_providers: tuple[tuple[str, tuple[str, Any]], ...] = () # Precomputed attributes, not methods: hook sites read one attribute to # short-circuit, so the zero-extension path constructs nothing. @@ -84,6 +85,7 @@ class ExtensionRegistry(ExtensionRegistryContract): self._services: list[_Entry] = [] self._routers: list[_Entry] = [] self._plugins: list[_Entry] = [] + self._knowledge_providers: list[_Entry] = [] self._current_source: str | None = None self._model_access: ModelInvocationScope | None = None @@ -149,6 +151,17 @@ class ExtensionRegistry(ExtensionRegistryContract): def middlewares(self, contributor: MiddlewareContributor) -> None: self._middlewares.append((self._source(), contributor)) + def knowledge_provider(self, name: str, provider: Any) -> bool: + from deerflow.knowledge.providers import validate_provider + + if not isinstance(name, str) or not re.fullmatch(r"[a-z][a-z0-9_-]{0,63}", name) or name in {"ragflow", "lightrag"}: + raise ValueError("Invalid or reserved knowledge provider name") + validate_provider(provider) + if any(existing == name for _, (existing, _) in self._knowledge_providers): + raise ValueError("Duplicate knowledge provider name") + self._knowledge_providers.append((self._source(), (name, provider))) + return True + def task_lifecycle(self, contributor: TaskLifecycleContributor) -> None: self._task_lifecycle.append((self._source(), contributor)) @@ -192,6 +205,7 @@ class ExtensionRegistry(ExtensionRegistryContract): self._services, self._routers, self._plugins, + self._knowledge_providers, ): bucket[:] = [entry for entry in bucket if entry[0] != source] @@ -206,6 +220,7 @@ class ExtensionRegistry(ExtensionRegistryContract): len(self._services), len(self._routers), len(self._plugins), + len(self._knowledge_providers), ) def rollback_to(self, mark: tuple[int, ...]) -> None: @@ -225,6 +240,7 @@ class ExtensionRegistry(ExtensionRegistryContract): self._services, self._routers, self._plugins, + self._knowledge_providers, ), mark, strict=True, @@ -242,6 +258,7 @@ class ExtensionRegistry(ExtensionRegistryContract): services=tuple(self._services), routers=tuple(self._routers), plugins=tuple(self._plugins), + knowledge_providers=tuple(self._knowledge_providers), has_middleware_contributors=bool(self._middlewares), has_task_lifecycle=bool(self._task_lifecycle), has_system_model_observers=bool(self._system_model_observers), diff --git a/backend/packages/harness/deerflow/knowledge/__init__.py b/backend/packages/harness/deerflow/knowledge/__init__.py index 9904987770..6a3a7f3fae 100644 --- a/backend/packages/harness/deerflow/knowledge/__init__.py +++ b/backend/packages/harness/deerflow/knowledge/__init__.py @@ -1,6 +1,32 @@ -"""Read-only knowledge contracts; no catalog or management dependency.""" +"""Read-only knowledge contracts with optional catalog and filter capabilities.""" -from .contract import KnowledgeCapabilities, KnowledgeSource, RetrievalHit, RetrievalProvider, RetrievalResult +from .contract import ( + KnowledgeCapabilities, + KnowledgeRequestContext, + KnowledgeSource, + RetrievalCatalogItem, + RetrievalCatalogPage, + RetrievalCatalogProvider, + RetrievalDocumentsProvider, + RetrievalHit, + RetrievalProvider, + RetrievalResult, + ScopedRetrievalProvider, +) from .sources import format_result, sources_from_messages -__all__ = ["KnowledgeCapabilities", "KnowledgeSource", "RetrievalHit", "RetrievalProvider", "RetrievalResult", "format_result", "sources_from_messages"] +__all__ = [ + "KnowledgeCapabilities", + "KnowledgeRequestContext", + "KnowledgeSource", + "RetrievalCatalogItem", + "RetrievalCatalogPage", + "RetrievalCatalogProvider", + "RetrievalDocumentsProvider", + "RetrievalHit", + "RetrievalProvider", + "RetrievalResult", + "ScopedRetrievalProvider", + "format_result", + "sources_from_messages", +] diff --git a/backend/packages/harness/deerflow/knowledge/contract.py b/backend/packages/harness/deerflow/knowledge/contract.py index bd176084e1..06da9b1ba6 100644 --- a/backend/packages/harness/deerflow/knowledge/contract.py +++ b/backend/packages/harness/deerflow/knowledge/contract.py @@ -4,7 +4,8 @@ from __future__ import annotations import hashlib import json -from typing import Literal, Protocol +from dataclasses import dataclass +from typing import Any, Literal, Protocol from pydantic import BaseModel, ConfigDict, Field, HttpUrl, computed_field, field_validator @@ -13,11 +14,12 @@ class KnowledgeCapabilities(BaseModel): """An operation is supported only when explicitly advertised. Search-only providers need no catalog, scope, or management implementation. - Scope admission and adapters for these optional operations belong to phase 2. """ operations: frozenset[Literal["search", "sources", "catalog", "documents", "dataset_filter", "document_filter"]] = frozenset({"search"}) + model_config = ConfigDict(frozen=True) + def supports(self, operation: str) -> bool: return operation in self.operations @@ -66,3 +68,56 @@ class RetrievalProvider(Protocol): capabilities: KnowledgeCapabilities async def search(self, query: str) -> RetrievalResult: ... + + +@dataclass(frozen=True) +class KnowledgeRequestContext: + """Host identity and execution constraints, never provider authorization. + + Providers must enforce their own permissions on every operation. ``scope`` + is a fresh execution-only snapshot; display labels never reach providers. + No tenant or credential is inferred from caller-supplied metadata. + ``agent_name`` identifies the current executor: the selected assistant + (``lead_agent`` by default), or the executing subagent's configured name. + It is not the parent assistant name. Direct provider callers may omit it. + """ + + user_id: str + agent_name: str | None = None + scope: dict[str, Any] | None = None + + +class RetrievalCatalogItem(BaseModel): + model_config = ConfigDict(extra="forbid", str_strip_whitespace=True) + + id: str = Field(min_length=1, max_length=256) + name: str = Field(min_length=1, max_length=500) + selectable: bool = False + + +class RetrievalCatalogPage(BaseModel): + model_config = ConfigDict(extra="forbid") + + items: list[RetrievalCatalogItem] = Field(default_factory=list, max_length=100) + page: int = Field(ge=1, strict=True) + page_size: int = Field(ge=1, le=100, strict=True) + total: int = Field(ge=0, strict=True) + + +class ScopedRetrievalProvider(Protocol): + """Optional contextual search; required for either filter capability. + + A missing/inaccessible selection must fail as a whole. Never retry a + constrained request as an unconstrained search. Context is also supplied + for all-scope searches when this method is implemented. + """ + + async def search_scoped(self, query: str, *, context: KnowledgeRequestContext) -> RetrievalResult: ... + + +class RetrievalCatalogProvider(Protocol): + async def catalog(self, *, context: KnowledgeRequestContext, page: int, page_size: int, search: str) -> RetrievalCatalogPage: ... + + +class RetrievalDocumentsProvider(Protocol): + async def documents(self, dataset_id: str, *, context: KnowledgeRequestContext, page: int, page_size: int, search: str) -> RetrievalCatalogPage: ... diff --git a/backend/packages/harness/deerflow/knowledge/providers.py b/backend/packages/harness/deerflow/knowledge/providers.py new file mode 100644 index 0000000000..832bd3d210 --- /dev/null +++ b/backend/packages/harness/deerflow/knowledge/providers.py @@ -0,0 +1,106 @@ +"""Resolve read-only providers and enforce operation-level scope capabilities. + +No provider configuration, HTTP calls or resource permissions live here. +Legacy tool paths select compatibility adapters; new providers are installed +through the operator-controlled extension registry and the generic core tool. +""" + +from __future__ import annotations + +import inspect +from typing import Any + +from deerflow.knowledge_scope import execution_scope + +from .contract import KnowledgeCapabilities, KnowledgeRequestContext, RetrievalProvider, RetrievalResult + +RAGFLOW_KNOWLEDGE_TOOL = "deerflow.community.ragflow.tools:knowledge_search_tool" +LIGHTRAG_KNOWLEDGE_TOOL = "deerflow.community.lightrag.tools:knowledge_search_tool" +GENERIC_KNOWLEDGE_TOOL = "deerflow.knowledge.tools:knowledge_search_tool" + + +def validate_provider(provider: Any) -> None: + capabilities = getattr(provider, "capabilities", None) + if not isinstance(capabilities, KnowledgeCapabilities) or not capabilities.supports("search"): + raise ValueError("Knowledge providers require KnowledgeCapabilities with search") + methods = {"search"} + if capabilities.supports("catalog"): + methods.add("catalog") + if capabilities.supports("documents"): + methods.add("documents") + if capabilities.supports("dataset_filter") or capabilities.supports("document_filter"): + methods.add("search_scoped") + for method in methods: + if not inspect.iscoroutinefunction(getattr(provider, method, None)): + raise ValueError(f"Knowledge provider capability requires async {method}") + + +def get_knowledge_capabilities(app_config: Any, *, extensions: Any = None) -> KnowledgeCapabilities: + get_tool = getattr(app_config, "get_tool_config", None) + tool = get_tool("knowledge_search") if callable(get_tool) else None + use = getattr(tool, "use", None) + if use == RAGFLOW_KNOWLEDGE_TOOL: + from deerflow.community.ragflow.formatting import CAPABILITIES + + return CAPABILITIES + if use == LIGHTRAG_KNOWLEDGE_TOOL: + from deerflow.community.lightrag.formatting import CAPABILITIES + + return CAPABILITIES + provider = _registered_provider(tool, extensions=extensions) + return provider.capabilities if provider is not None else KnowledgeCapabilities(operations=frozenset()) + + +def _registered_provider(tool: Any, *, extensions: Any = None) -> RetrievalProvider | None: + if getattr(tool, "use", None) != GENERIC_KNOWLEDGE_TOOL: + return None + from deerflow.extensions import get_loaded_extensions + + loaded = extensions if extensions is not None else get_loaded_extensions() + name = (getattr(tool, "model_extra", None) or {}).get("provider") + for _, (registered_name, provider) in loaded.knowledge_providers: + if registered_name == name: + return provider + return None + + +def resolve_knowledge_provider(app_config: Any, *, extensions: Any = None) -> RetrievalProvider: + get_tool = getattr(app_config, "get_tool_config", None) + tool = get_tool("knowledge_search") if callable(get_tool) else None + if getattr(tool, "use", None) == RAGFLOW_KNOWLEDGE_TOOL: + from deerflow.community.ragflow.provider import RAGFlowKnowledgeProvider + + return RAGFlowKnowledgeProvider(app_config) + provider = _registered_provider(tool, extensions=extensions) + if provider is None: + raise LookupError("Knowledge provider is not configured") + return provider + + +def validate_scope_capabilities(scope: object, capabilities: KnowledgeCapabilities) -> dict[str, Any]: + canonical = execution_scope(scope) + if not capabilities.supports("search"): + raise ValueError("Knowledge retrieval is unavailable") + if canonical["mode"] == "selected": + if not capabilities.supports("dataset_filter"): + raise ValueError("Knowledge provider does not support dataset_filter") + if canonical.get("document_filters") and not capabilities.supports("document_filter"): + raise ValueError("Knowledge provider does not support document_filter") + return canonical + + +async def search_provider(provider: RetrievalProvider, query: str, *, scope: object | None, user_id: str, agent_name: str | None) -> RetrievalResult: + canonical = validate_scope_capabilities(scope, provider.capabilities) if scope is not None else None + if canonical is not None and canonical["mode"] == "disabled": + raise ValueError("Knowledge search is disabled for this turn") + context = KnowledgeRequestContext(user_id=user_id, agent_name=agent_name, scope=canonical) + contextual_search = getattr(provider, "search_scoped", None) + if callable(contextual_search): + result = await contextual_search(query, context=context) + elif canonical is not None and canonical["mode"] == "selected": + raise ValueError("Knowledge provider cannot apply the selected scope") + else: + result = await provider.search(query) + # Validate provider values before they can become model-visible evidence. + payload = result.model_dump() if isinstance(result, RetrievalResult) else result + return RetrievalResult.model_validate(payload) diff --git a/backend/packages/harness/deerflow/knowledge/tools.py b/backend/packages/harness/deerflow/knowledge/tools.py new file mode 100644 index 0000000000..5153872839 --- /dev/null +++ b/backend/packages/harness/deerflow/knowledge/tools.py @@ -0,0 +1,75 @@ +"""Stable Agent entry point for registered read-only knowledge providers.""" + +import logging +from collections.abc import Mapping + +from langchain_core.tools import StructuredTool + +from deerflow.config import get_app_config +from deerflow.config.app_config import AppConfig +from deerflow.extensions import resolve_run_extensions +from deerflow.knowledge.providers import resolve_knowledge_provider, search_provider +from deerflow.knowledge.sources import format_result +from deerflow.knowledge_scope import KNOWLEDGE_SCOPE_RUNTIME_KEY, execution_scope +from deerflow.runtime.user_context import resolve_runtime_user_id +from deerflow.tools.types import Runtime + +logger = logging.getLogger(__name__) + + +def _execution_agent_name(context: Mapping, config: object = None) -> str: + """Identify the current executor, rather than a delegated run's parent.""" + if context.get("is_subagent") is True: + name = context.get("agent_id") + else: + configurable = config.get("configurable") if isinstance(config, Mapping) else None + legacy_name = configurable.get("agent_name") if isinstance(configurable, Mapping) else None + name = context.get("agent_name", legacy_name) + if name is None and context.get("is_subagent") is not True: + return "lead_agent" + if not isinstance(name, str) or not name.strip(): + raise ValueError("Invalid knowledge execution agent identity") + return name + + +async def _knowledge_search_entrypoint(query: str, runtime: Runtime) -> tuple[str, dict | None]: + """Retrieve evidence from the configured knowledge provider. + + Args: + query: Specific question or search terms for private documents. + """ + if not query.strip(): + return "Error: query must not be empty.", None + context = runtime.context if isinstance(runtime.context, Mapping) else {} + try: + raw_scope = context.get(KNOWLEDGE_SCOPE_RUNTIME_KEY) + scope = execution_scope(raw_scope) if raw_scope is not None else None + if scope is not None and scope["mode"] == "disabled": + return "Error: Knowledge search is disabled for this turn.", None + app_config = context.get("app_config") + if not isinstance(app_config, AppConfig): + app_config = get_app_config() + provider = resolve_knowledge_provider(app_config, extensions=resolve_run_extensions(context)) + result = await search_provider( + provider, + query.strip(), + scope=scope, + user_id=resolve_runtime_user_id(runtime), + agent_name=_execution_agent_name(context, getattr(runtime, "config", None)), + ) + return format_result(result) + except PermissionError: + return "Error: The selected knowledge scope is not accessible.", None + except Exception as exc: + # Provider exceptions may contain credentials or resource identifiers. + logger.warning("Knowledge retrieval failed (%s)", type(exc).__name__) + return "Error: Knowledge retrieval failed; check the provider and selected scope.", None + + +knowledge_search_tool = StructuredTool.from_function( + coroutine=_knowledge_search_entrypoint, + name="knowledge_search", + response_format="content_and_artifact", + description="Search accessible knowledge within this turn's scope. Cite only supplied [citation:Title](#knowledge-source-ID) links exactly; never invent source locators.", + parse_docstring=True, +) diff --git a/backend/packages/harness/pyproject.toml b/backend/packages/harness/pyproject.toml index de54df2222..d3133ac2df 100644 --- a/backend/packages/harness/pyproject.toml +++ b/backend/packages/harness/pyproject.toml @@ -13,7 +13,7 @@ dependencies = [ # the contract version it implements, extensions declare ranges. A range # here would let pip resolve a newer contract package than this harness # implements, making newer extensions look supported at runtime. - "deerflow-extension-api==0.2.6", + "deerflow-extension-api==0.2.7", "dotenv>=0.9.9", "exa-py>=1.0.0", "httpx>=0.28.0", diff --git a/backend/tests/blocking_io/test_knowledge_router.py b/backend/tests/blocking_io/test_knowledge_router.py index 6467dc650b..41d7745554 100644 --- a/backend/tests/blocking_io/test_knowledge_router.py +++ b/backend/tests/blocking_io/test_knowledge_router.py @@ -25,6 +25,7 @@ from sqlalchemy import create_engine from app.gateway.knowledge_scope_admission import RAGFLOW_KNOWLEDGE_SEARCH_PROVIDER from app.gateway.routers import knowledge +from deerflow.community.ragflow import provider as ragflow_provider from deerflow.config.paths import get_paths from deerflow.persistence.agents.model import AgentRow from deerflow.persistence.agents.sql import SqlAgentStore @@ -114,7 +115,7 @@ def _fake_ragflow(monkeypatch) -> SimpleNamespace: } ), ) - monkeypatch.setattr(knowledge, "_build_retrieval_client", lambda settings: ragflow) + monkeypatch.setattr(ragflow_provider, "build_ragflow_retrieval_client", lambda settings: ragflow) return ragflow @@ -123,7 +124,7 @@ async def test_catalog_handlers_do_not_block_event_loop(monkeypatch) -> None: _fake_ragflow(monkeypatch) datasets = await _list_datasets( - request=None, + request=SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace())), agent_name=_AGENT, page=1, page_size=20, @@ -132,7 +133,7 @@ async def test_catalog_handlers_do_not_block_event_loop(monkeypatch) -> None: ) documents = await _list_documents( "dataset-1", - request=None, + request=SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace())), agent_name=_AGENT, page=1, page_size=20, @@ -149,7 +150,7 @@ async def test_unknown_agent_is_not_found_without_blocking_event_loop(monkeypatc with pytest.raises(HTTPException) as exc_info: await _list_datasets( - request=None, + request=SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace())), agent_name="missing-agent", page=1, page_size=20, diff --git a/backend/tests/test_extension_api_contracts.py b/backend/tests/test_extension_api_contracts.py index 4a429a3514..e4e104b9bd 100644 --- a/backend/tests/test_extension_api_contracts.py +++ b/backend/tests/test_extension_api_contracts.py @@ -342,7 +342,7 @@ def test_runtime_api_version_matches_the_installed_contract_package(): """Every additive contract slice bumps both gates together.""" from importlib.metadata import version - assert API_VERSION == "0.2.6" + assert API_VERSION == "0.2.7" assert API_VERSION == version("deerflow-extension-api") diff --git a/backend/tests/test_gateway_knowledge_scope_admission.py b/backend/tests/test_gateway_knowledge_scope_admission.py index 4367efc967..80cece8162 100644 --- a/backend/tests/test_gateway_knowledge_scope_admission.py +++ b/backend/tests/test_gateway_knowledge_scope_admission.py @@ -63,7 +63,7 @@ def test_admission_canonicalizes_custom_agent_human_message() -> None: ("assistant_id", "provider", "tool_groups"), [ (None, "deerflow.community.ragflow.tools:knowledge_search_tool", None), - ("agent", "deerflow.community.lightrag.tools:knowledge_search_tool", None), + ("agent", "unknown:knowledge_search_tool", None), ("agent", "deerflow.community.ragflow.tools:knowledge_search_tool", []), ("agent", "deerflow.community.ragflow.tools:knowledge_search_tool", ["web"]), ], diff --git a/backend/tests/test_knowledge_contract.py b/backend/tests/test_knowledge_contract.py index 6c7344325b..b4c0102f60 100644 --- a/backend/tests/test_knowledge_contract.py +++ b/backend/tests/test_knowledge_contract.py @@ -10,6 +10,28 @@ from pydantic import ValidationError from deerflow.knowledge import KnowledgeSource, RetrievalHit, RetrievalResult, format_result, sources_from_messages +@pytest.mark.parametrize("length", [256, 257]) +def test_catalog_and_scope_share_opaque_id_length_limit(length): + from deerflow.knowledge import RetrievalCatalogItem + from deerflow.knowledge_scope import canonicalize_knowledge_scope + + dataset_id = "界" * length + document_id = "😀" * length + scope = {"version": 1, "mode": "selected", "dataset_ids": [dataset_id], "document_filters": [{"dataset_id": dataset_id, "document_ids": [document_id]}]} + if length == 256: + assert RetrievalCatalogItem(id=dataset_id, name="Dataset").id == dataset_id + assert RetrievalCatalogItem(id=document_id, name="Document").id == document_id + assert canonicalize_knowledge_scope(scope) == scope + else: + for resource_id in (dataset_id, document_id): + with pytest.raises(ValidationError): + RetrievalCatalogItem(id=resource_id, name="Resource") + with pytest.raises(ValidationError): + canonicalize_knowledge_scope(scope) + with pytest.raises(ValidationError): + canonicalize_knowledge_scope({**scope, "dataset_ids": ["valid"], "document_filters": [{"dataset_id": "valid", "document_ids": [document_id]}]}) + + def test_source_schema_matches_cross_component_contract(): from jsonschema import Draft202012Validator diff --git a/backend/tests/test_knowledge_provider.py b/backend/tests/test_knowledge_provider.py new file mode 100644 index 0000000000..1a410fefd7 --- /dev/null +++ b/backend/tests/test_knowledge_provider.py @@ -0,0 +1,417 @@ +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +from fastapi import HTTPException +from langchain_core.messages import HumanMessage + +from app.gateway.knowledge_scope_admission import admit_message_knowledge_scope +from deerflow.extensions import ExtensionRegistry, reset_loaded_extensions, set_loaded_extensions +from deerflow.knowledge import KnowledgeCapabilities, RetrievalResult +from deerflow.knowledge.providers import GENERIC_KNOWLEDGE_TOOL, get_knowledge_capabilities, search_provider + + +@pytest.fixture(autouse=True) +def clear_registry(): + reset_loaded_extensions() + yield + reset_loaded_extensions() + + +def provider_config(name="test"): + tool = SimpleNamespace(use=GENERIC_KNOWLEDGE_TOOL, model_extra={"provider": name}) + return SimpleNamespace( + knowledge_base=SimpleNamespace(enabled=True, scope_selection_enabled=True), + get_tool_config=lambda _: tool, + ) + + +def install_provider(operations): + provider = SimpleNamespace( + capabilities=KnowledgeCapabilities(operations=frozenset(operations)), + search=AsyncMock(return_value=RetrievalResult()), + search_scoped=AsyncMock(return_value=RetrievalResult()), + catalog=AsyncMock(), + documents=AsyncMock(), + ) + registry = ExtensionRegistry() + with registry.attributed_to("test:install"): + assert registry.knowledge_provider("test", provider) + set_loaded_extensions(registry.build()) + return provider + + +def test_search_only_provider_needs_no_catalog_or_scoped_search(): + provider = SimpleNamespace(capabilities=KnowledgeCapabilities(), search=AsyncMock()) + registry = ExtensionRegistry() + with registry.attributed_to("test:install"): + registry.knowledge_provider("test", provider) + set_loaded_extensions(registry.build()) + assert get_knowledge_capabilities(provider_config()).operations == {"search"} + + +@pytest.mark.parametrize("operations", [{"search"}, {"search", "catalog"}]) +def test_admission_rejects_unadvertised_dataset_filter(operations): + provider = install_provider(operations) + graph_input = { + "messages": [ + HumanMessage( + content="q", + additional_kwargs={ + "knowledge_scope": {"version": 1, "mode": "selected", "dataset_ids": ["private"]}, + }, + ) + ] + } + with pytest.raises(HTTPException) as exc: + admit_message_knowledge_scope(graph_input, assistant_id="lead_agent", app_config=provider_config(), agent_config=None) + assert exc.value.status_code == 422 + provider.search.assert_not_called() + + +def test_filter_admission_does_not_require_catalog_and_recovery_is_authoritative(): + install_provider({"search", "dataset_filter"}) + graph_input = { + "messages": [ + HumanMessage( + content="q", + additional_kwargs={ + "knowledge_scope": {"version": 1, "mode": "all"}, + }, + ) + ] + } + recovered = {"version": 1, "mode": "selected", "dataset_ids": ["private"]} + assert ( + admit_message_knowledge_scope( + graph_input, + assistant_id="lead_agent", + app_config=provider_config(), + agent_config=None, + recovery=True, + recovery_scope=recovered, + ) + == recovered + ) + + +@pytest.mark.asyncio +async def test_execution_rechecks_filter_capabilities_without_falling_back(): + provider = install_provider({"search", "dataset_filter"}) + scope = {"version": 1, "mode": "selected", "dataset_ids": ["private"], "document_filters": [{"dataset_id": "private", "document_ids": ["secret"]}]} + with pytest.raises(ValueError, match="document_filter"): + await search_provider(provider, "q", scope=scope, user_id="alice", agent_name="lead_agent") + provider.search.assert_not_awaited() + provider.search_scoped.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_provider_receives_only_execution_scope_and_host_identity(): + provider = install_provider({"search", "dataset_filter"}) + scope = {"version": 1, "mode": "selected", "dataset_ids": ["private"], "display": {"datasets": [{"id": "private", "name": "untrusted label"}]}} + await search_provider(provider, "q", scope=scope, user_id="alice", agent_name="researcher") + context = provider.search_scoped.await_args.kwargs["context"] + assert context.user_id == "alice" + assert context.agent_name == "researcher" + assert context.scope == {"version": 1, "mode": "selected", "dataset_ids": ["private"]} + provider.search.assert_not_awaited() + + +def test_registration_rollback_and_duplicate_names(): + registry = ExtensionRegistry() + provider = SimpleNamespace(capabilities=KnowledgeCapabilities(), search=AsyncMock()) + with registry.attributed_to("test:install"): + registry.knowledge_provider("test", provider) + mark = registry.mark() + registry.knowledge_provider("second", provider) + with pytest.raises(ValueError, match="Duplicate"): + registry.knowledge_provider("test", provider) + registry.rollback_to(mark) + assert [name for _, (name, _) in registry.build().knowledge_providers] == ["test"] + + +def test_registration_rejects_capabilities_without_implementation(): + registry = ExtensionRegistry() + provider = SimpleNamespace(capabilities=KnowledgeCapabilities(operations={"search", "dataset_filter"}), search=AsyncMock()) + with registry.attributed_to("test:install"), pytest.raises(ValueError, match="search_scoped"): + registry.knowledge_provider("test", provider) + + +@pytest.mark.asyncio +async def test_search_only_provider_and_disabled_scope_do_not_need_catalog(): + provider = SimpleNamespace(capabilities=KnowledgeCapabilities(), search=AsyncMock(return_value=RetrievalResult())) + assert await search_provider(provider, "q", scope=None, user_id="alice", agent_name=None) == RetrievalResult() + provider.search.assert_awaited_once_with("q") + with pytest.raises(ValueError, match="disabled"): + await search_provider(provider, "q", scope={"version": 1, "mode": "disabled"}, user_id="alice", agent_name=None) + assert provider.search.await_count == 1 + + +@pytest.mark.asyncio +async def test_permission_denial_never_retries_unconstrained(): + provider = install_provider({"search", "dataset_filter"}) + provider.search_scoped.side_effect = PermissionError("private provider details") + with pytest.raises(PermissionError): + await search_provider(provider, "q", scope={"version": 1, "mode": "selected", "dataset_ids": ["private"]}, user_id="alice", agent_name="lead_agent") + provider.search.assert_not_awaited() + assert provider.search_scoped.await_count == 1 + + +@pytest.mark.asyncio +async def test_generic_tool_uses_run_snapshot_and_validates_source_artifact(monkeypatch): + from langchain_core.messages import ToolMessage + + from deerflow.extensions import EXTENSION_SNAPSHOT_CONTEXT_KEY, get_loaded_extensions + from deerflow.knowledge import KnowledgeSource, RetrievalHit, sources_from_messages, tools + from deerflow.knowledge.sources import prune_sources + from deerflow.runtime.serialization import serialize_channel_values_for_api + + provider = install_provider({"search", "sources", "dataset_filter"}) + source = KnowledgeSource(provider="test", title="Private manual", excerpt="Actual provider evidence.") + provider.search_scoped.return_value = RetrievalResult(hits=[RetrievalHit(content=source.excerpt, source=source)]) + loaded = get_loaded_extensions() + reset_loaded_extensions() # A newer singleton must not replace the running snapshot. + monkeypatch.setattr(tools, "get_app_config", provider_config) + runtime = SimpleNamespace( + context={ + "user_id": "alice", + "agent_name": "researcher", + EXTENSION_SNAPSHOT_CONTEXT_KEY: loaded, + "__knowledge_scope_execution": {"version": 1, "mode": "selected", "dataset_ids": ["private"]}, + } + ) + content, artifact = await tools._knowledge_search_entrypoint("q", runtime) + message = ToolMessage(content=content, artifact=artifact, tool_call_id="retrieval-1", name="knowledge_search") + import json + + persisted = json.loads(json.dumps(serialize_channel_values_for_api({"messages": [message]}))) + assert sources_from_messages(persisted["messages"]) == [source] + assert provider.search_scoped.await_args.kwargs["context"].user_id == "alice" + # A truncated tool message cannot retain evidence that is no longer visible. + _, pruned = prune_sources("truncated", artifact) + assert pruned["knowledge_sources"] == [] + + +@pytest.mark.asyncio +async def test_generic_tool_keeps_run_config_and_authenticated_identity(monkeypatch): + from deerflow.config.app_config import AppConfig + from deerflow.config.sandbox_config import SandboxConfig + from deerflow.config.tool_config import ToolConfig + from deerflow.extensions import EXTENSION_SNAPSHOT_CONTEXT_KEY, get_loaded_extensions + from deerflow.knowledge import tools + + provider = install_provider({"search", "dataset_filter"}) + config = AppConfig( + tools=[ToolConfig(name="knowledge_search", group="knowledge", use=GENERIC_KNOWLEDGE_TOOL, provider="test")], + sandbox=SandboxConfig(use="example.remote:Sandbox"), + ) + monkeypatch.setattr(tools, "get_app_config", lambda: provider_config("replacement")) + runtime = SimpleNamespace( + context={"app_config": config, "user_id": "forged", EXTENSION_SNAPSHOT_CONTEXT_KEY: get_loaded_extensions()}, + server_info=SimpleNamespace(user=SimpleNamespace(identity="authenticated")), + ) + content, artifact = await tools._knowledge_search_entrypoint("q", runtime) + assert not content.startswith("Error:") + assert artifact is not None + assert provider.search_scoped.await_args.kwargs["context"].user_id == "authenticated" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("assistant_id", [None, "lead_agent", "researcher"]) +@pytest.mark.parametrize("request_config", [None, {"configurable": {"agent_name": "researcher"}}, {"context": {"agent_name": "writer"}}]) +async def test_generic_tool_identity_matches_real_gateway_run_config(monkeypatch, assistant_id, request_config): + from app.gateway.services import build_run_config + from deerflow.knowledge import tools + + provider = install_provider({"search", "dataset_filter"}) + monkeypatch.setattr(tools, "get_app_config", provider_config) + run_config = build_run_config("knowledge-thread", request_config, None, assistant_id=assistant_id) + runtime = SimpleNamespace(context=run_config.get("context", {}), config=run_config) + content, _ = await tools._knowledge_search_entrypoint("q", runtime) + assert not content.startswith("Error:") + merged = {**run_config.get("configurable", {}), **run_config.get("context", {})} + assert provider.search_scoped.await_args.kwargs["context"].agent_name == (merged.get("agent_name") or "lead_agent") + + +@pytest.mark.asyncio +async def test_subagent_identity_takes_precedence_over_inherited_parent_name(monkeypatch): + from deerflow.knowledge import tools + + provider = install_provider({"search", "dataset_filter"}) + monkeypatch.setattr(tools, "get_app_config", provider_config) + runtime = SimpleNamespace(context={"is_subagent": True, "agent_id": "child", "agent_name": "parent"}) + content, _ = await tools._knowledge_search_entrypoint("q", runtime) + assert not content.startswith("Error:") + assert provider.search_scoped.await_args.kwargs["context"].agent_name == "child" + + +@pytest.mark.asyncio +async def test_missing_subagent_identity_does_not_fall_back_to_lead_agent(monkeypatch): + from deerflow.knowledge import tools + + provider = install_provider({"search", "dataset_filter"}) + monkeypatch.setattr(tools, "get_app_config", provider_config) + runtime = SimpleNamespace(context={"is_subagent": True, "agent_name": "parent"}) + content, artifact = await tools._knowledge_search_entrypoint("q", runtime) + assert content.startswith("Error:") + assert artifact is None + provider.search_scoped.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_standalone_admission_cannot_forge_knowledge_subagent_identity(monkeypatch): + from app.gateway.langgraph_auth import _scrub_run_incarnation + from deerflow.knowledge import tools + + provider = install_provider({"search", "dataset_filter"}) + monkeypatch.setattr(tools, "get_app_config", provider_config) + request = {"kwargs": {"context": {"is_subagent": True, "agent_id": "privileged-child"}}} + _scrub_run_incarnation(request) + runtime = SimpleNamespace(context=request["kwargs"]["context"], config=request["kwargs"]["config"]) + content, _ = await tools._knowledge_search_entrypoint("q", runtime) + assert not content.startswith("Error:") + assert provider.search_scoped.await_args.kwargs["context"].agent_name == "lead_agent" + + +@pytest.mark.asyncio +async def test_standalone_admission_overrides_persisted_assistant_child_identity(monkeypatch): + from langgraph_sdk import Auth + + from app.gateway.langgraph_auth import add_owner_filter + from deerflow.knowledge import tools + + provider = install_provider({"search", "dataset_filter"}) + monkeypatch.setattr(tools, "get_app_config", provider_config) + defaults = {"is_subagent": True, "agent_id": "privileged-child"} + request = {"kwargs": {"context": {}}, "assistant_id": "lead_agent"} + auth_context = Auth.types.AuthContext(resource="threads", action="create_run", user=SimpleNamespace(identity="alice"), permissions=[]) + await add_owner_filter(auth_context, request) + # The standalone runtime merges stored assistant defaults AFTER admission. + effective_context = {**defaults, **request["kwargs"]["context"]} + runtime = SimpleNamespace(context=effective_context, config=request["kwargs"]["config"]) + content, _ = await tools._knowledge_search_entrypoint("q", runtime) + assert not content.startswith("Error:") + assert provider.search_scoped.await_args.kwargs["context"].agent_name == "lead_agent" + assert provider.search_scoped.await_args.kwargs["context"].user_id == "alice" + + +@pytest.mark.asyncio +async def test_disabled_scope_returns_specific_error_before_loading_provider(monkeypatch): + from deerflow.knowledge import tools + + provider = install_provider({"search", "dataset_filter"}) + + def unavailable_config(): + pytest.fail("disabled search must not load configuration or resolve a provider") + + monkeypatch.setattr(tools, "get_app_config", unavailable_config) + runtime = SimpleNamespace(context={"__knowledge_scope_execution": {"version": 1, "mode": "disabled"}}) + assert await tools._knowledge_search_entrypoint("q", runtime) == ("Error: Knowledge search is disabled for this turn.", None) + provider.search.assert_not_awaited() + provider.search_scoped.assert_not_awaited() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("scope", [{"version": 2, "mode": "disabled"}, {"version": 1, "mode": "selected", "dataset_ids": []}]) +async def test_malformed_scope_is_rejected_before_loading_provider(monkeypatch, scope): + from deerflow.knowledge import tools + + provider = install_provider({"search", "dataset_filter"}) + monkeypatch.setattr(tools, "get_app_config", lambda: pytest.fail("invalid scope must fail before loading config")) + runtime = SimpleNamespace(context={"__knowledge_scope_execution": scope}) + content, artifact = await tools._knowledge_search_entrypoint("q", runtime) + assert content == "Error: Knowledge retrieval failed; check the provider and selected scope." + assert artifact is None + provider.search.assert_not_awaited() + provider.search_scoped.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_generic_tool_node_injects_scope_and_emits_shared_evidence(monkeypatch): + from langchain_core.messages import AIMessage, ToolMessage + from langgraph.graph import END, START, MessagesState, StateGraph + from langgraph.prebuilt import ToolNode + + from deerflow.extensions import EXTENSION_SNAPSHOT_CONTEXT_KEY, get_loaded_extensions + from deerflow.knowledge import KnowledgeSource, RetrievalHit, sources_from_messages, tools + + provider = install_provider({"search", "sources", "dataset_filter"}) + source = KnowledgeSource(provider="test", title="Guide", excerpt="Actual private evidence") + provider.search_scoped.return_value = RetrievalResult(hits=[RetrievalHit(content=source.excerpt, source=source)]) + monkeypatch.setattr(tools, "get_app_config", provider_config) + scope = {"version": 1, "mode": "selected", "dataset_ids": ["private"]} + 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": "q"}}])]}, + context={"user_id": "alice", "__knowledge_scope_execution": scope, EXTENSION_SNAPSHOT_CONTEXT_KEY: get_loaded_extensions()}, + ) + message = result["messages"][-1] + assert isinstance(message, ToolMessage) + assert sources_from_messages([message]) == [source] + context = provider.search_scoped.await_args.kwargs["context"] + assert context.user_id == "alice" + assert context.scope == scope + provider.search.assert_not_awaited() + + +@pytest.mark.parametrize("mode", ["all", "disabled"]) +def test_lightrag_accepts_only_nonfiltering_scopes(mode): + config = provider_config() + config.get_tool_config = lambda _: SimpleNamespace(use="deerflow.community.lightrag.tools:knowledge_search_tool") + graph_input = {"messages": [HumanMessage(content="q", additional_kwargs={"knowledge_scope": {"version": 1, "mode": mode}})]} + assert admit_message_knowledge_scope(graph_input, assistant_id="lead_agent", app_config=config, agent_config=None) == {"version": 1, "mode": mode} + + +@pytest.mark.asyncio +async def test_packaged_provider_catalog_scope_and_sources(monkeypatch): + from pathlib import Path + + from deerflow.extensions import load_extensions + from deerflow.extensions.loader import ExtensionSpec + from deerflow.knowledge import KnowledgeRequestContext, format_result, sources_from_messages + from deerflow.knowledge.providers import resolve_knowledge_provider + + monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[2] / "examples/deerflow-extension-knowledge")) + loaded, diagnostics = load_extensions([ExtensionSpec(use="deerflow_extension_knowledge:install")]) + assert diagnostics == [] + provider = resolve_knowledge_provider(provider_config("synthetic-example"), extensions=loaded) + context = KnowledgeRequestContext(user_id="alice", agent_name="lead_agent") + catalog = await provider.catalog(context=context, page=1, page_size=20, search="") + dataset = catalog.items[0] + documents = await provider.documents(dataset.id, context=context, page=1, page_size=20, search="") + scope = {"version": 1, "mode": "selected", "dataset_ids": [dataset.id], "document_filters": [{"dataset_id": dataset.id, "document_ids": [documents.items[0].id]}]} + result = await search_provider(provider, "Cedar", scope=scope, user_id="alice", agent_name="lead_agent") + content, artifact = format_result(result) + sources = sources_from_messages([{"type": "tool", "content": content, "artifact": artifact}]) + assert len(sources) == 1 and sources[0].provider == "synthetic-example" + assert sources[0].page is None and sources[0].url is None + with pytest.raises(PermissionError): + await search_provider(provider, "Cedar", scope={"version": 1, "mode": "selected", "dataset_ids": ["missing"]}, user_id="alice", agent_name="lead_agent") + + +@pytest.mark.asyncio +async def test_ragflow_adapter_keeps_selected_documents_allowlist_and_redaction(monkeypatch): + from test_ragflow_tools import DATASET_ID_1, DATASET_ID_2, FakeRAGFlowClient, _config, _dataset + + from deerflow.community.ragflow import provider as ragflow_provider + + fake = FakeRAGFlowClient( + datasets_by_id={DATASET_ID_1: [_dataset(DATASET_ID_1, "Private guide")]}, + documents_by_dataset_id={DATASET_ID_1: [{"id": "doc-1", "run": "DONE", "chunk_count": 1}]}, + retrieval={"chunks": [{"dataset_id": DATASET_ID_1, "document_keyword": "guide.pdf", "content": "Evidence ragflow-secret"}]}, + ) + monkeypatch.setattr(ragflow_provider, "build_ragflow_retrieval_client", lambda _: fake) + provider = ragflow_provider.RAGFlowKnowledgeProvider(_config(datasets=[DATASET_ID_1])) + scope = {"version": 1, "mode": "selected", "dataset_ids": [DATASET_ID_1], "document_filters": [{"dataset_id": DATASET_ID_1, "document_ids": ["doc-1"]}]} + result = await search_provider(provider, "q", scope=scope, user_id="alice", agent_name="lead_agent") + assert fake.retrieve_calls[0][1]["dataset_ids"] == [DATASET_ID_1] + assert fake.retrieve_calls[0][1]["document_ids"] == ["doc-1"] + assert result.hits[0].source.dataset_name == "Private guide" + assert "ragflow-secret" not in str(result.model_dump()) + before = len(fake.retrieve_calls) + with pytest.raises(PermissionError): + await search_provider(provider, "q", scope={"version": 1, "mode": "selected", "dataset_ids": [DATASET_ID_2]}, user_id="alice", agent_name="lead_agent") + assert len(fake.retrieve_calls) == before diff --git a/backend/tests/test_knowledge_provider_router.py b/backend/tests/test_knowledge_provider_router.py new file mode 100644 index 0000000000..70e9e27fdb --- /dev/null +++ b/backend/tests/test_knowledge_provider_router.py @@ -0,0 +1,117 @@ +from types import SimpleNamespace +from unittest.mock import AsyncMock +from urllib.parse import quote + +import pytest +from _router_auth_helpers import make_authed_test_app +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.extensions import ExtensionRegistry, reset_loaded_extensions +from deerflow.knowledge import KnowledgeCapabilities, RetrievalCatalogPage +from deerflow.knowledge.providers import GENERIC_KNOWLEDGE_TOOL + + +def setup_provider(operations): + provider = SimpleNamespace( + capabilities=KnowledgeCapabilities(operations=frozenset(operations)), + search=AsyncMock(), + search_scoped=AsyncMock(), + catalog=AsyncMock(return_value=RetrievalCatalogPage(items=[{"id": "tenant:private/kb", "name": "Private", "selectable": True}], page=1, page_size=20, total=1)), + documents=AsyncMock(return_value=RetrievalCatalogPage(items=[], page=1, page_size=20, total=0)), + ) + registry = ExtensionRegistry() + with registry.attributed_to("test:install"): + registry.knowledge_provider("test", provider) + user = User(email="catalog@example.com", password_hash="x", system_role="user") + app = make_authed_test_app(user_factory=lambda: user, bind_current_user=True) + app.state.extensions = registry.build() + app.include_router(knowledge.router) + tool = SimpleNamespace(use=GENERIC_KNOWLEDGE_TOOL, model_extra={"provider": "test"}) + config = SimpleNamespace(knowledge_base=SimpleNamespace(enabled=True, scope_selection_enabled=True), get_tool_config=lambda _: tool) + app.dependency_overrides[get_config] = lambda: config + return app, provider, user + + +@pytest.fixture(autouse=True) +def no_singleton_provider(): + reset_loaded_extensions() + yield + reset_loaded_extensions() + + +def test_catalog_uses_app_snapshot_and_authenticated_identity(): + app, provider, user = setup_provider({"search", "catalog", "dataset_filter"}) + with TestClient(app) as client: + cap = client.get("/api/knowledge/capabilities?agent_name=lead_agent") + response = client.get("/api/knowledge/retrieval-catalog/datasets?agent_name=lead_agent&user_id=mallory") + assert cap.json() == {"operations": ["catalog", "dataset_filter", "search"], "scope_selection_enabled": True} + assert response.status_code == 200 + assert response.json()["items"][0]["id"] == "tenant:private/kb" + context = provider.catalog.await_args.kwargs["context"] + assert context.user_id == str(user.id) + assert context.user_id != "mallory" + assert context.agent_name == "lead_agent" + assert context.scope is None + + +def test_search_only_provider_is_available_without_catalog(): + app, provider, _ = setup_provider({"search"}) + with TestClient(app) as client: + response = client.get("/api/knowledge/capabilities?agent_name=lead_agent") + assert response.json() == {"operations": ["search"], "scope_selection_enabled": False} + assert client.get("/api/knowledge/retrieval-catalog/datasets?agent_name=lead_agent").status_code == 409 + provider.catalog.assert_not_awaited() + + +def test_catalog_without_document_capability_never_calls_documents(): + app, provider, _ = setup_provider({"search", "catalog", "dataset_filter"}) + with TestClient(app) as client: + response = client.get("/api/knowledge/retrieval-catalog/datasets/private/documents?agent_name=lead_agent") + assert response.status_code == 409 + provider.documents.assert_not_awaited() + + +def test_provider_opaque_ids_are_not_constrained_to_ragflow_uuid_syntax(): + app, provider, _ = setup_provider({"search", "catalog", "documents", "dataset_filter", "document_filter"}) + dataset_id = "tenant:private/kb with spaces" + with TestClient(app) as client: + response = client.get(f"/api/knowledge/retrieval-catalog/datasets/{quote(dataset_id, safe='')}/documents?agent_name=lead_agent") + assert response.status_code == 200 + assert provider.documents.await_args.args == (dataset_id,) + + +@pytest.mark.parametrize(("length", "status"), [(256, 200), (257, 422)]) +def test_document_route_preserves_bounded_opaque_id_or_rejects_it(length, status): + app, provider, _ = setup_provider({"search", "catalog", "documents", "dataset_filter", "document_filter"}) + dataset_id = "😀" * length + with TestClient(app) as client: + response = client.get(f"/api/knowledge/retrieval-catalog/datasets/{quote(dataset_id, safe='')}/documents?agent_name=lead_agent") + assert response.status_code == status + if status == 200: + assert provider.documents.await_args.args == (dataset_id,) + else: + provider.documents.assert_not_awaited() + + +@pytest.mark.parametrize(("error", "status"), [(PermissionError("credential=secret"), 403), (FileNotFoundError("private"), 404), (RuntimeError("credential=secret"), 502)]) +def test_catalog_errors_are_safe_and_do_not_retry(error, status): + app, provider, _ = setup_provider({"search", "catalog", "dataset_filter"}) + provider.catalog.side_effect = error + with TestClient(app) as client: + response = client.get("/api/knowledge/retrieval-catalog/datasets?agent_name=lead_agent") + assert response.status_code == status + assert "secret" not in response.text + assert provider.catalog.await_count == 1 + provider.search.assert_not_awaited() + + +def test_malformed_catalog_cannot_expose_provider_fields(): + app, provider, _ = setup_provider({"search", "catalog", "dataset_filter"}) + provider.catalog.return_value = {"items": [], "page": 1, "page_size": 20, "total": 0, "api_key": "secret"} + with TestClient(app) as client: + response = client.get("/api/knowledge/retrieval-catalog/datasets?agent_name=lead_agent") + assert response.status_code == 502 + assert "secret" not in response.text diff --git a/backend/tests/test_knowledge_router.py b/backend/tests/test_knowledge_router.py index 6aa1f033dc..b934e81e7c 100644 --- a/backend/tests/test_knowledge_router.py +++ b/backend/tests/test_knowledge_router.py @@ -10,6 +10,7 @@ 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.community.ragflow import provider as ragflow_provider def _config( @@ -56,8 +57,8 @@ def _app( app.include_router(knowledge.router) app.dependency_overrides[get_config] = lambda: config or _config() monkeypatch.setattr( - knowledge, - "_build_retrieval_client", + ragflow_provider, + "build_ragflow_retrieval_client", lambda settings: client, ) return app diff --git a/backend/tests/test_langgraph_auth.py b/backend/tests/test_langgraph_auth.py index 9c53adda0f..640776b737 100644 --- a/backend/tests/test_langgraph_auth.py +++ b/backend/tests/test_langgraph_auth.py @@ -304,6 +304,9 @@ def test_run_admission_uses_persisted_incarnation_and_scrubs_client_values(): "if_not_exists": "reject", } + for section in (value["kwargs"]["context"], *value["kwargs"]["config"].values()): + section.update(is_subagent=True, agent_id="privileged-child") + with patch.object( auth_module, "_read_standalone_thread", @@ -325,6 +328,11 @@ def test_run_admission_uses_persisted_incarnation_and_scrubs_client_values(): assert value["kwargs"]["context"]["user_id"] == "user-a" assert value["kwargs"]["context"]["thread_id"] == str(thread_id) assert value["kwargs"]["context"]["run_id"] == str(run_id) + assert value["kwargs"]["context"]["is_subagent"] is False + assert value["kwargs"]["context"]["agent_id"] is None + for section in value["kwargs"]["config"].values(): + assert "is_subagent" not in section + assert "agent_id" not in section assert THREAD_INCARNATION_CONTEXT_KEY not in value["metadata"] assert THREAD_INCARNATION_METADATA_GUARD_KEY not in value["metadata"] assert THREAD_INCARNATION_CONTEXT_KEY not in value["kwargs"]["config"]["context"] @@ -338,6 +346,17 @@ def test_run_admission_uses_persisted_incarnation_and_scrubs_client_values(): assert THREAD_INCARNATION_METADATA_GUARD_KEY not in value["kwargs"]["config"]["configurable"] +@pytest.mark.parametrize("action", ["create", "update"]) +def test_assistant_writes_cannot_persist_child_execution_identity(action): + forged = {"is_subagent": True, "agent_id": "privileged-child", "agent_name": "researcher"} + value = {"context": dict(forged), "config": {"context": dict(forged), "configurable": dict(forged)}} + asyncio.run(add_owner_filter(_make_ctx("alice", resource="assistants", action=action), value)) + for section in (value["context"], *value["config"].values()): + assert "is_subagent" not in section + assert "agent_id" not in section + assert section["agent_name"] == "researcher" + + def test_existing_legacy_thread_uses_explicit_none_without_backfill(): with patch.object( auth_module, diff --git a/backend/tests/test_lightrag_tools.py b/backend/tests/test_lightrag_tools.py index 46ff7cb2d3..755d6c5b8b 100644 --- a/backend/tests/test_lightrag_tools.py +++ b/backend/tests/test_lightrag_tools.py @@ -2,6 +2,7 @@ import logging from pathlib import Path from types import SimpleNamespace from typing import Any +from unittest.mock import AsyncMock import pytest @@ -418,3 +419,30 @@ def test_lightrag_package_has_explicit_init_file() -> None: package_dir = Path(lightrag_tools.__file__).resolve().parent assert (package_dir / "__init__.py").is_file() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", ["selected", "disabled"]) +async def test_runtime_scope_never_falls_back_to_unconstrained_lightrag(monkeypatch, mode): + search = AsyncMock() + monkeypatch.setattr(lightrag_tools, "_search", search) + scope = {"version": 1, "mode": mode} + if mode == "selected": + scope["dataset_ids"] = ["private"] + runtime = SimpleNamespace(context={"__knowledge_scope_execution": scope}) + content, artifact = await lightrag_tools._knowledge_search_entrypoint("q", runtime=runtime) + assert content.startswith("Error:") + assert artifact is None + search.assert_not_awaited() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("scope", [None, {"version": 1, "mode": "all"}]) +async def test_runtime_unfiltered_scope_preserves_lightrag_search(monkeypatch, scope): + search = AsyncMock(return_value=("evidence", {"knowledge_sources": []})) + monkeypatch.setattr(lightrag_tools, "_search", search) + runtime = SimpleNamespace(context={"__knowledge_scope_execution": scope}) + content, artifact = await lightrag_tools._knowledge_search_entrypoint("q", runtime=runtime) + assert content == "evidence" + assert artifact is not None + search.assert_awaited_once_with("q", with_sources=True) diff --git a/backend/tests/test_ragflow_client.py b/backend/tests/test_ragflow_client.py index a942ec8ad5..b28af26eb4 100644 --- a/backend/tests/test_ragflow_client.py +++ b/backend/tests/test_ragflow_client.py @@ -11,6 +11,38 @@ from deerflow.community.ragflow.client import ( ) +@pytest.mark.anyio +@pytest.mark.parametrize("dataset_id", ["", ".", "..", "../other", "a/../../other", "a/b", "a\\b", "%2e%2e", "%252f", "a%2fb", "a?x=y", "a#fragment", "a b", " a", "a\n", "资料"]) +async def test_list_documents_rejects_unsafe_path_ids_without_http_request(dataset_id): + requests = [] + + async def handler(request): + requests.append(request) + return httpx.Response(200, json={"code": 0, "data": {"docs": []}}) + + client = RAGFlowClient(base_url="http://ragflow.test", api_key="secret", transport=httpx.MockTransport(handler)) + with pytest.raises(ValueError, match="Invalid RAGFlow dataset ID") as exc: + await client.list_documents(dataset_id, params=[]) + assert str(exc.value) == "Invalid RAGFlow dataset ID." + assert requests == [] + + +@pytest.mark.anyio +@pytest.mark.parametrize("dataset_id", ["dataset-1", "kb_ABC123", "0123456789abcdef0123456789abcdef"]) +async def test_list_documents_keeps_valid_ids_in_one_path_segment(dataset_id): + requests = [] + + async def handler(request): + requests.append(request) + assert request.url.path == f"/api/v1/datasets/{dataset_id}/documents" + assert dict(request.url.params) == {"page": "1"} + return httpx.Response(200, json={"code": 0, "data": {"docs": []}}) + + client = RAGFlowClient(base_url="http://ragflow.test", api_key="secret", transport=httpx.MockTransport(handler)) + assert await client.list_documents(dataset_id, params=[("page", "1")]) == {"code": 0, "data": {"docs": []}} + assert len(requests) == 1 + + @pytest.mark.anyio async def test_list_datasets_filters_by_bound_id_in_one_request() -> None: requests: list[httpx.Request] = [] diff --git a/backend/tests/test_retrieval_plugin_artifacts.py b/backend/tests/test_retrieval_plugin_artifacts.py index dee07aad8a..1ecbdb88a5 100644 --- a/backend/tests/test_retrieval_plugin_artifacts.py +++ b/backend/tests/test_retrieval_plugin_artifacts.py @@ -129,3 +129,42 @@ async def test_no_matches_is_a_successful_empty_result(retrieval_plugin): assert message.status == "success" assert "No matching synthetic documents" in message.content assert message.artifact == {"schema": "example.retrieval.v1", "synthetic": True, "matches": []} + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", ["full", "delta"]) +@pytest.mark.parametrize("limit", [120, 10000]) +async def test_shared_knowledge_sources_from_plugin_survive_budget_and_replay(tmp_path, retrieval_plugin, mode, limit): + from dataclasses import replace + + from deerflow_extension_api import PluginToolResult + + from deerflow.knowledge import KnowledgeSource, RetrievalHit, RetrievalResult, format_result, sources_from_messages + + source = KnowledgeSource(provider="plugin-test", title="Synthetic captured guide", excerpt="Synthetic retrieved evidence from the provider.") + content, artifact = format_result(RetrievalResult(hits=[RetrievalHit(content=source.excerpt, source=source)])) + + async def handler(payload, context): + return PluginToolResult(content, artifact) + + loaded = retrieval_plugin({"enabled": True}) + attribution, plugin = loaded.plugins[0] + plugin = replace(plugin, tools=(replace(plugin.tools[0], handler=handler),)) + (tool,) = build_plugin_tools(replace(loaded, plugins=((attribution, plugin),))) + config = {"configurable": {"thread_id": "plugin-knowledge-replay"}} + inject_checkpoint_mode(config, mode) + database = str(tmp_path / "knowledge.db") + async with AsyncSqliteSaver.from_conn_string(database) as saver: + graph = _graph(tool, saver, mode=mode, limit=limit) + result = await graph.ainvoke({"messages": [_call(tool, {"query": "guide"})]}, config, context={"user_id": "alice"}) + message = result["messages"][-1] + assert message.status == "success" + expected = [source] if limit == 10000 else [] + assert sources_from_messages([message.model_dump()]) == expected + + async with AsyncSqliteSaver.from_conn_string(database) as saver: + graph = _graph(tool, saver, mode=mode, limit=limit) + snapshot = await CheckpointStateAccessor.bind(graph, saver, mode=mode).aget(config) + restored = json.loads(json.dumps(serialize_channel_values_for_api(snapshot.values))) + assert sources_from_messages(restored["messages"]) == expected + assert restored["messages"][-1]["artifact"] == message.artifact diff --git a/backend/tests/test_subagent_executor.py b/backend/tests/test_subagent_executor.py index 2100ed8644..a39667fcbe 100644 --- a/backend/tests/test_subagent_executor.py +++ b/backend/tests/test_subagent_executor.py @@ -87,6 +87,36 @@ def test_knowledge_accumulator_scans_only_changed_tail(_setup_executor_classes, assert scanned == [[first], [second], [compacted]] +@pytest.mark.anyio +async def test_knowledge_provider_receives_real_subagent_execution_identity(classes, base_config, monkeypatch): + from langchain_core.messages import AIMessage + + from deerflow.knowledge import KnowledgeCapabilities, RetrievalResult, tools + + provider = SimpleNamespace(capabilities=KnowledgeCapabilities(operations={"search", "dataset_filter"}), search_scoped=AsyncMock(return_value=RetrievalResult())) + monkeypatch.setattr(tools, "get_app_config", _default_app_config) + monkeypatch.setattr(tools, "resolve_knowledge_provider", lambda *args, **kwargs: provider) + scope = {"version": 1, "mode": "selected", "dataset_ids": ["private"]} + + async def capturing_astream(state, *, context, **kwargs): + assert context["is_subagent"] is True + assert context["agent_id"] == base_config.name + content, _ = await tools._knowledge_search_entrypoint("q", SimpleNamespace(context=context)) + assert not content.startswith("Error:") + yield {"messages": [AIMessage(content="Done", id="knowledge-result")]} + + agent = MagicMock() + agent.astream = capturing_astream + executor = classes["SubagentExecutor"](config=base_config, tools=[], thread_id="knowledge-thread", user_id="alice", knowledge_scope=scope) + with patch.object(executor, "_create_agent", return_value=agent): + result = await executor._aexecute("Search private knowledge") + assert result.status == classes["SubagentStatus"].COMPLETED + context = provider.search_scoped.await_args.kwargs["context"] + assert context.agent_name == base_config.name + assert context.user_id == "alice" + assert context.scope == scope + + def _default_app_config(): return SimpleNamespace( tool_search=SimpleNamespace(enabled=False), diff --git a/backend/uv.lock b/backend/uv.lock index df60e27145..978615fdea 100644 --- a/backend/uv.lock +++ b/backend/uv.lock @@ -924,7 +924,7 @@ extensions = [] [[package]] name = "deerflow-extension-api" -version = "0.2.6" +version = "0.2.7" source = { editable = "packages/extension-api" } [[package]] diff --git a/examples/deerflow-extension-knowledge/LICENSE b/examples/deerflow-extension-knowledge/LICENSE new file mode 100644 index 0000000000..9dc98a4a6b --- /dev/null +++ b/examples/deerflow-extension-knowledge/LICENSE @@ -0,0 +1,22 @@ +MIT License + +Copyright (c) 2025 Bytedance Ltd. and/or its affiliates +Copyright (c) 2025-2026 DeerFlow Authors + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/examples/deerflow-extension-knowledge/README.md b/examples/deerflow-extension-knowledge/README.md new file mode 100644 index 0000000000..e905b6d7ed --- /dev/null +++ b/examples/deerflow-extension-knowledge/README.md @@ -0,0 +1,44 @@ +# Synthetic read-only knowledge provider + +This independently packaged, offline example registers a `RetrievalProvider` +with optional catalog, documents and scope filtering. It has one invented +document and no external API, management service, or browser bundle. The corpus +is deliberately public to all users; production adapters must map the host +identity in `KnowledgeRequestContext` to provider permissions on every request. + +Install it from an absolute local path using DeerFlow's extension manager, then +configure the existing `knowledge_search` entry and restart Gateway: + +```yaml +plugins: + - use: deerflow_extension_knowledge:install + enabled: true + required: true + +knowledge_base: + enabled: true + scope_selection_enabled: true + +tool_groups: + - name: knowledge + +tools: + - name: knowledge_search + group: knowledge + use: deerflow.knowledge.tools:knowledge_search_tool + provider: synthetic-example +``` + +Merge these entries with your existing configuration. The provider name is an +operator-selected registry key, never a user-supplied module path. Select +**Synthetic guides**, optionally select **Synthetic knowledge note**, and ask +the Agent to search for `Cedar`. The returned source is captured evidence of +the invented example text, with no fabricated page or URL. + +Unknown datasets or documents fail rather than widening scope. The core +creates the source artifact, applies evidence validation and budgets, and +preserves the existing message/runtime scope boundary. The separate +`deerflow-extension-retrieval` example continues to demonstrate +`ModelTool(response_format="content_and_artifact")` and `PluginToolResult`; +this example consumes the same public source semantics through the stable +`knowledge_search` entry point. diff --git a/examples/deerflow-extension-knowledge/deerflow_extension_knowledge/__init__.py b/examples/deerflow-extension-knowledge/deerflow_extension_knowledge/__init__.py new file mode 100644 index 0000000000..a4e0daf442 --- /dev/null +++ b/examples/deerflow-extension-knowledge/deerflow_extension_knowledge/__init__.py @@ -0,0 +1,51 @@ +"""One synthetic corpus exercising the public read-only knowledge contract.""" + +from deerflow.knowledge import KnowledgeCapabilities, KnowledgeRequestContext, KnowledgeSource, RetrievalCatalogPage, RetrievalHit, RetrievalResult +from deerflow_extension_api import extension + +_DATASET = "synthetic-guides" +_DOCUMENT = "demo-note" +_TITLE = "Synthetic knowledge note" +_TEXT = "The fictional Cedar team keeps its invented demonstration notes in a read-only knowledge base. This text is synthetic test evidence." + + +class SyntheticKnowledgeProvider: + capabilities = KnowledgeCapabilities(operations={"search", "sources", "catalog", "documents", "dataset_filter", "document_filter"}) + + async def search(self, query: str) -> RetrievalResult: + hits = [] + if query.strip().casefold() in (_TITLE + " " + _TEXT).casefold(): + source = KnowledgeSource(provider="synthetic-example", title=_TITLE, excerpt=_TEXT, dataset_name="Synthetic guides") + hits.append(RetrievalHit(content=_TEXT, source=source)) + return RetrievalResult(hits=hits) + + async def search_scoped(self, query: str, *, context: KnowledgeRequestContext) -> RetrievalResult: + scope = context.scope + if scope is not None: + if scope["mode"] == "disabled": + raise ValueError("Knowledge search is disabled") + if scope["mode"] == "selected": + if scope["dataset_ids"] != [_DATASET]: + raise PermissionError("Selected dataset is unavailable") + for item in scope.get("document_filters", []): + if item["dataset_id"] != _DATASET or item["document_ids"] != [_DOCUMENT]: + raise PermissionError("Selected document is unavailable") + return await self.search(query) + + async def catalog(self, *, context: KnowledgeRequestContext, page: int, page_size: int, search: str) -> RetrievalCatalogPage: + items = [{"id": _DATASET, "name": "Synthetic guides", "selectable": True}] if search.strip().casefold() in "synthetic guides" else [] + start = (page - 1) * page_size + return RetrievalCatalogPage(items=items[start : start + page_size], page=page, page_size=page_size, total=len(items)) + + async def documents(self, dataset_id: str, *, context: KnowledgeRequestContext, page: int, page_size: int, search: str) -> RetrievalCatalogPage: + if dataset_id != _DATASET: + raise FileNotFoundError("Dataset unavailable") + items = [{"id": _DOCUMENT, "name": _TITLE, "selectable": True}] if search.strip().casefold() in _TITLE.casefold() else [] + start = (page - 1) * page_size + return RetrievalCatalogPage(items=items[start : start + page_size], page=page, page_size=page_size, total=len(items)) + + +@extension(api="0.2.7", name="knowledge-example") +def install(registry, config): + if not registry.knowledge_provider("synthetic-example", SyntheticKnowledgeProvider()): + raise RuntimeError("This host does not support knowledge provider registration") diff --git a/examples/deerflow-extension-knowledge/pyproject.toml b/examples/deerflow-extension-knowledge/pyproject.toml new file mode 100644 index 0000000000..6b5a6b3894 --- /dev/null +++ b/examples/deerflow-extension-knowledge/pyproject.toml @@ -0,0 +1,25 @@ +[project] +name = "deerflow-extension-knowledge" +version = "0.1.0" +description = "Offline synthetic read-only knowledge provider for DeerFlow" +readme = "README.md" +requires-python = ">=3.12" +license = "MIT" +dependencies = ["deerflow-extension-api>=0.2.7,<0.3", "deerflow-harness>=2.1,<3"] + +[project.entry-points."deerflow.extensions"] +knowledge-example = "deerflow_extension_knowledge:install" + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.hatch.build.targets.wheel] +packages = ["deerflow_extension_knowledge"] + +[tool.ruff] +line-length = 240 +target-version = "py312" + +[tool.ruff.lint] +select = ["E", "F", "I", "UP"] diff --git a/frontend/src/AGENTS.md b/frontend/src/AGENTS.md index 810a36aa36..a19e399488 100644 --- a/frontend/src/AGENTS.md +++ b/frontend/src/AGENTS.md @@ -238,13 +238,13 @@ Markdown. Source records retain retrieval-time evidence, not live documents. Resolve knowledge destinations before testing the label in message and artifact link renderers: Sources lists use ordinary titles without a `citation:` prefix. -Custom-agent settings store an optional `knowledge_scope` retrieval default. -`knowledgeScopeToSelection` restores all execution IDs; missing display names -fall back to IDs. Save only edited scopes, except removing knowledge tool access -must atomically clear the default. Catalog selection needs saved and draft access. -Chats initialize from saved bindings; explicit page-local overrides survive -new-thread route replacement and reset on conversation changes. Gateway supplies -defaults for clients without a selector; UI visibility is not runtime enforcement. +Custom Agents save edited `knowledge_scope` defaults; removing knowledge access +clears them atomically. `knowledgeScopeToSelection` keeps IDs, using IDs for +missing labels. Catalog needs saved/draft access and exact-Agent capabilities. +`catalog` + `dataset_filter` enable bases; `documents` + `document_filter` files. +Pending/failed capabilities block queries and unsupported Apply; keep old file +filters. Chats seed defaults; local overrides survive new-thread replacement and +reset per conversation. Gateway supplies defaults and enforces scope. MCP hooks and the editor default to personal scope. Mutation-option builders require an explicit scope. MCP control accessible names include the localized ownership section, so same-named shared and personal connections remain distinct without region context. Query keys include scope and authenticated user ID; remount each editor when that ID changes. Platform provided uses the deployment editor only for administrators outside static mode, restoring add, edit, toggle and delete through the existing admin-only MCP API. Ordinary users see shared entries read-only. My plugins contains personal configuration templates, Lark account authorization and the personal MCP editor; a deployment installation must not mark a personal template configured. diff --git a/frontend/src/components/workspace/knowledge-scope-selector.tsx b/frontend/src/components/workspace/knowledge-scope-selector.tsx index 0e9786749f..f5c4543c2e 100644 --- a/frontend/src/components/workspace/knowledge-scope-selector.tsx +++ b/frontend/src/components/workspace/knowledge-scope-selector.tsx @@ -25,6 +25,7 @@ import { buildKnowledgeScopeSnapshot, cloneKnowledgeScopeSelection, countKnowledgeScopeSelection, + fetchKnowledgeProviderCapabilities, listRetrievalCatalogDatasets, listRetrievalCatalogDocuments, type KnowledgeScopeDatasetSelection, @@ -263,8 +264,32 @@ export function KnowledgeScopeSelector({ ); const [page, setPage] = useState(1); const [search, setSearch] = useState(""); + const capabilities = useQuery({ + queryKey: ["knowledge", "capabilities", agentName], + queryFn: () => fetchKnowledgeProviderCapabilities(agentName), + enabled: open && !unavailableReason, + retry: false, + staleTime: 0, + }); + const operations = capabilities.data?.operations ?? []; + const canSelectDatasets = + capabilities.isSuccess && + capabilities.data.scope_selection_enabled && + operations.includes("catalog") && + operations.includes("dataset_filter"); + const canSelectDocuments = + canSelectDatasets && + operations.includes("documents") && + operations.includes("document_filter"); const query = useQuery({ - queryKey: ["knowledge", "retrieval-catalog", agentName, page, search], + queryKey: [ + "knowledge", + "retrieval-catalog", + agentName, + operations, + page, + search, + ], queryFn: () => listRetrievalCatalogDatasets({ agentName, @@ -272,7 +297,11 @@ export function KnowledgeScopeSelector({ pageSize: PAGE_SIZE, search, }), - enabled: open && draft.mode === "selected" && !unavailableReason, + enabled: + open && + draft.mode === "selected" && + canSelectDatasets && + !unavailableReason, retry: false, }); @@ -301,6 +330,13 @@ export function KnowledgeScopeSelector({ dataset.documents.mode === "selected" && dataset.documents.items.length === 0, )); + const unsupportedSelection = + draft.mode === "selected" && + (!canSelectDatasets || + (!canSelectDocuments && + draft.datasets.some( + (dataset) => dataset.documents.mode === "selected", + ))); const draftExceedsLimits = useMemo(() => { if (draftInvalid) return false; try { @@ -369,7 +405,9 @@ export function KnowledgeScopeSelector({