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 <willem.jiang@gmail.com>
This commit is contained in:
王一帆 2026-10-15 15:10:39 +08:00 committed by GitHub
parent edcb78fb87
commit 6bc1020d34
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
53 changed files with 1910 additions and 228 deletions

View File

@ -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.

View File

@ -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 |

View File

@ -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")

View File

@ -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}

View File

@ -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")

View File

@ -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))

View File

@ -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(

View File

@ -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

View File

@ -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",

View File

@ -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

View File

@ -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

View File

@ -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([]))

View File

@ -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;

View File

@ -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(

View File

@ -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:

View File

@ -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)

View File

@ -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

View File

@ -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),

View File

@ -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",
]

View File

@ -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: ...

View File

@ -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)

View File

@ -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,
)

View File

@ -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",

View File

@ -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,

View File

@ -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")

View File

@ -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"]),
],

View File

@ -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

View File

@ -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

View File

@ -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

View File

@ -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

View File

@ -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,

View File

@ -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)

View File

@ -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] = []

View File

@ -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

View File

@ -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),

2
backend/uv.lock generated
View File

@ -924,7 +924,7 @@ extensions = []
[[package]]
name = "deerflow-extension-api"
version = "0.2.6"
version = "0.2.7"
source = { editable = "packages/extension-api" }
[[package]]

View File

@ -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.

View File

@ -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.

View File

@ -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")

View File

@ -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"]

View File

@ -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.

View File

@ -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({
<label className="flex items-center gap-2" key={mode}>
<input
checked={draft.mode === mode}
disabled={disabled}
disabled={
disabled || (mode === "selected" && !canSelectDatasets)
}
name="knowledge-scope-mode"
type="radio"
onChange={() =>
@ -404,7 +442,12 @@ export function KnowledgeScopeSelector({
{t.knowledge.scope.selectedCount(draft.datasets.length)}
</div>
<ScrollArea className="min-h-48 flex-1 pr-3">
{query.isPending ? (
{capabilities.isError ||
(capabilities.isSuccess && !canSelectDatasets) ? (
<p className="text-destructive p-3 text-sm">
{t.knowledge.scope.loadFailed}
</p>
) : capabilities.isPending || query.isPending ? (
<Loader2Icon className="text-muted-foreground mx-auto mt-8 size-5 animate-spin" />
) : query.isError ? (
<p className="text-destructive p-3 text-sm">
@ -443,7 +486,7 @@ export function KnowledgeScopeSelector({
{t.knowledge.scope.notSearchable}
</span>
)}
{selected && (
{selected && canSelectDocuments && (
<Button
aria-expanded={expanded}
size="sm"
@ -465,7 +508,7 @@ export function KnowledgeScopeSelector({
</Button>
)}
</div>
{selected && expanded && (
{selected && expanded && canSelectDocuments && (
<DocumentSelector
agentName={agentName}
dataset={selected}
@ -491,7 +534,8 @@ export function KnowledgeScopeSelector({
</ScrollArea>
</div>
)}
{draftExceedsLimits && (
{(draftExceedsLimits ||
(capabilities.isSuccess && unsupportedSelection)) && (
<p className="text-destructive text-xs">
{t.knowledge.scope.selectionInvalid}
</p>
@ -505,7 +549,12 @@ export function KnowledgeScopeSelector({
{t.common.cancel}
</Button>
<Button
disabled={disabled || draftInvalid || draftExceedsLimits}
disabled={
disabled ||
draftInvalid ||
draftExceedsLimits ||
unsupportedSelection
}
type="button"
onClick={() => {
onChange(cloneKnowledgeScopeSelection(draft));

View File

@ -40,6 +40,11 @@ export const extensionCatalog = [
package: "deerflow-extension-retrieval",
namespace: "example.retrieval",
},
{
key: "knowledge",
package: "deerflow-extension-knowledge",
namespace: null,
},
{ key: "example", package: "deerflow-extension-example", namespace: null },
] as const;

View File

@ -80,6 +80,11 @@ export const enUS: Translations = {
description:
"Demonstrate read-only retrieval and structured tool results using fixed sample documents, without an external knowledge service.",
},
knowledge: {
title: "Knowledge scope example",
description:
"Search sample documents within selected knowledge bases or files and inspect source citations.",
},
},
pageUnavailable: "Extension page unavailable",
pageUnavailableHint:

View File

@ -34,6 +34,7 @@ export interface Translations {
| "classify"
| "screening"
| "retrieval"
| "knowledge"
| "example",
{ title: string; description: string }
>;

View File

@ -76,6 +76,10 @@ export const zhCN: Translations = {
description:
"使用固定示例文档演示只读检索与结构化工具结果,不连接外部知识库。",
},
knowledge: {
title: "知识范围选择示例",
description: "用示例文档演示知识库和文件范围选择、检索及来源引用。",
},
},
pageUnavailable: "扩展页面不可用",
pageUnavailableHint: "此页面未注册,或插件已停用、未能加载。",

View File

@ -15,6 +15,34 @@ export type RetrievalCatalogPage = {
total: number;
};
export type KnowledgeOperation =
| "search"
| "sources"
| "catalog"
| "documents"
| "dataset_filter"
| "document_filter";
export type KnowledgeProviderCapabilities = {
operations: KnowledgeOperation[];
scope_selection_enabled: boolean;
};
export async function fetchKnowledgeProviderCapabilities(
agentName: string,
): Promise<KnowledgeProviderCapabilities> {
const query = new URLSearchParams({ agent_name: agentName });
const response = await fetch(
`${getBackendBaseURL()}/api/knowledge/capabilities?${query}`,
);
if (!response.ok)
await throwGatewayApiError(
response,
"Failed to load knowledge capabilities.",
);
return (await response.json()) as KnowledgeProviderCapabilities;
}
function catalogUrl(path: string): string {
return `${getBackendBaseURL()}/api/knowledge/retrieval-catalog${path}`;
}

View File

@ -20,7 +20,13 @@ test.beforeAll(async ({ request }) => {
gatewayURL = `http://127.0.0.1:${address.port}`;
const backend = path.resolve(process.cwd(), "../backend");
gateway = spawn(
path.join(backend, ".venv/bin/python"),
process.env.PLAYWRIGHT_PYTHON ??
path.join(
backend,
process.platform === "win32"
? ".venv/Scripts/python.exe"
: ".venv/bin/python",
),
[
"-m",
"extension_test_fixtures.bookmark_plugin_gateway",
@ -41,7 +47,7 @@ test.beforeAll(async ({ request }) => {
.then((r) => r.status())
.catch(() => 0);
},
{ timeout: 20_000 },
{ timeout: 60_000 },
)
.toBe(200);
});
@ -442,7 +448,37 @@ for (const locale of ["en-US", "zh-CN"]) {
exact: true,
}),
).toBeVisible();
await expect(page.getByRole("article")).toHaveCount(8);
await expect(page.getByRole("article")).toHaveCount(9);
const knowledgeTitle = zh ? "知识范围选择示例" : "Knowledge scope example";
const knowledge = page
.getByRole("article")
.filter({ hasText: knowledgeTitle });
await expect(knowledge).toHaveCount(1);
await expect(knowledge).toContainText(
zh
? "用示例文档演示知识库和文件范围选择、检索及来源引用。"
: "Search sample documents within selected knowledge bases or files and inspect source citations.",
);
await knowledge
.getByRole("button", {
name: zh ? `查看 ${knowledgeTitle}` : `View ${knowledgeTitle}`,
exact: true,
})
.click();
await expect(
page.getByRole("heading", { name: knowledgeTitle, exact: true }),
).toBeVisible();
await expect(
page.getByRole("link", {
name: zh ? "查看安装说明" : "Installation guide",
}),
).toHaveAttribute("href", /deerflow-extension-knowledge#readme$/);
await page
.getByRole("button", {
name: zh ? "全部扩展" : "All extensions",
exact: true,
})
.click();
const workflow = page.getByRole("article").filter({
hasText: zh ? "脚本工作流" : "Script workflows",
});
@ -602,11 +638,27 @@ test("catalog remains discoverable when runtime discovery fails", async ({
await expect(
page.getByRole("alert").filter({ hasText: "Extensions unavailable." }),
).toBeVisible();
await expect(page.getByRole("article")).toHaveCount(8);
await expect(page.getByRole("article")).toHaveCount(9);
await expect(
page.getByRole("article").filter({ hasText: "Catalog extension" }),
).toHaveCount(8);
).toHaveCount(9);
await expect(
page.getByRole("article").filter({ hasText: "Script workflows" }),
).toHaveCount(1);
const knowledge = page
.getByRole("article")
.filter({ hasText: "Knowledge scope example" });
await expect(knowledge).toHaveCount(1);
await expect(knowledge).toContainText(
"Search sample documents within selected knowledge bases or files and inspect source citations.",
);
await knowledge
.getByRole("button", { name: "View Knowledge scope example", exact: true })
.click();
await expect(
page.getByRole("heading", { name: "Knowledge scope example", exact: true }),
).toBeVisible();
await expect(
page.getByRole("link", { name: "Installation guide" }),
).toHaveAttribute("href", /deerflow-extension-knowledge#readme$/);
});

View File

@ -2390,6 +2390,24 @@ export function mockLangGraphAPI(
return route.fallback();
});
void page.route("**/api/knowledge/capabilities?*", (route) =>
route.fulfill({
json: {
operations: featureFlags.knowledgeScopeSelectionEnabled
? [
"search",
"sources",
"catalog",
"documents",
"dataset_filter",
"document_filter",
]
: ["search"],
scope_selection_enabled: featureFlags.knowledgeScopeSelectionEnabled,
},
}),
);
void page.route("**/api/mcp/config", (route) =>
route.fulfill({ json: { mcp_servers: {} } }),
);

View File

@ -183,14 +183,32 @@ describe("tool access and default knowledge", () => {
);
it("loads the catalog after knowledge access is persisted", async () => {
const fetch = rs.spyOn(globalThis, "fetch").mockImplementation(() =>
const fetch = rs.spyOn(globalThis, "fetch").mockImplementation((input) =>
Promise.resolve(
Response.json({
items: [{ id: "policies", name: "Policies", selectable: true }],
page: 1,
page_size: 100,
total: 1,
}),
Response.json(
(typeof input === "string"
? input
: input instanceof URL
? input.href
: input.url
).includes("/capabilities?")
? {
operations: [
"search",
"catalog",
"documents",
"dataset_filter",
"document_filter",
],
scope_selection_enabled: true,
}
: {
items: [{ id: "policies", name: "Policies", selectable: true }],
page: 1,
page_size: 100,
total: 1,
},
),
),
);
const { rerender } = renderSettings({
@ -210,6 +228,13 @@ describe("tool access and default knowledge", () => {
const trigger = screen.getByTestId("knowledge-scope-trigger");
expect((trigger as HTMLButtonElement).disabled).toBe(false);
fireEvent.click(trigger);
await waitFor(() =>
expect(
screen.getByRole<HTMLInputElement>("radio", {
name: "Selected knowledge bases",
}).disabled,
).toBe(false),
);
fireEvent.click(
screen.getByRole("radio", { name: "Selected knowledge bases" }),
);

View File

@ -0,0 +1,55 @@
import { afterEach, describe, expect, it } from "@rstest/core";
import { cleanup, render, screen } from "@testing-library/react";
import { KnowledgeEvidence } from "@/components/workspace/citations/knowledge-citation";
import { readKnowledgeSources } from "@/core/citations/sources";
import { I18nProvider } from "@/core/i18n/context";
afterEach(cleanup);
describe("registered provider evidence replay", () => {
it("renders persisted plugin sources without fabricating pages or URLs", () => {
const source = {
id: "a".repeat(32),
provider: "catalog-test-provider",
title: "Captured manual",
excerpt: "Actual retrieved excerpt.",
};
const persisted: unknown = JSON.parse(
JSON.stringify([
{
type: "tool",
name: "ext_example_search",
content: `[citation:Captured manual](#knowledge-source-${source.id})\n${source.excerpt}`,
artifact: {
knowledge_sources_version: 1,
knowledge_sources: [source],
},
},
]),
);
const sources = readKnowledgeSources(persisted as unknown[]);
expect(sources).toHaveLength(1);
render(
<I18nProvider initialLocale="en-US">
<KnowledgeEvidence source={sources[0]!} />
</I18nProvider>,
);
expect(screen.getByText("catalog-test-provider")).toBeDefined();
expect(screen.getByText(source.excerpt)).toBeDefined();
expect(screen.queryByText(/Pages/)).toBeNull();
expect(screen.queryByRole("link")).toBeNull();
expect(
readKnowledgeSources([
{
type: "tool",
content: "unrelated",
artifact: {
knowledge_sources_version: 1,
knowledge_sources: [source],
},
},
]),
).toEqual([]);
});
});

View File

@ -74,6 +74,19 @@ describe("KnowledgeScopeSelector trigger", () => {
it("does not load documents while an expanded dataset still uses all files", async () => {
const fetch = rs.spyOn(globalThis, "fetch").mockImplementation((input) => {
const url = requestUrl(input);
if (url.includes("/capabilities?"))
return Promise.resolve(
Response.json({
operations: [
"search",
"catalog",
"documents",
"dataset_filter",
"document_filter",
],
scope_selection_enabled: true,
}),
);
return Promise.resolve(
Response.json({
items: url.includes("/documents?")
@ -118,4 +131,85 @@ describe("KnowledgeScopeSelector trigger", () => {
).toBe(true);
});
});
it("allows dataset selection without exposing unsupported document filtering", async () => {
const fetch = rs.spyOn(globalThis, "fetch").mockImplementation((input) => {
const url = requestUrl(input);
return Promise.resolve(
Response.json(
url.includes("/capabilities?")
? {
operations: ["search", "catalog", "dataset_filter"],
scope_selection_enabled: true,
}
: {
items: [
{ id: "dataset-1", name: "Policies", selectable: true },
],
page: 1,
page_size: 100,
total: 1,
},
),
);
});
renderSelector({
mode: "selected",
datasets: [
{
id: "dataset-1",
name: "Policies",
documents: { mode: "all" },
},
],
});
fireEvent.click(screen.getByRole("button", { name: "Knowledge · 1 base" }));
await screen.findByText("Policies");
expect(screen.queryByRole("button", { name: "Files" })).toBeNull();
expect(
fetch.mock.calls.some(([input]) =>
requestUrl(input).includes("/documents?"),
),
).toBe(false);
});
it("preserves a stale document selection and blocks Apply when filtering is unsupported", async () => {
rs.spyOn(globalThis, "fetch").mockImplementation((input) =>
Promise.resolve(
Response.json(
requestUrl(input).includes("/capabilities?")
? {
operations: ["search", "catalog", "dataset_filter"],
scope_selection_enabled: true,
}
: {
items: [
{ id: "dataset-1", name: "Policies", selectable: true },
],
page: 1,
page_size: 100,
total: 1,
},
),
),
);
renderSelector({
mode: "selected",
datasets: [
{
id: "dataset-1",
name: "Policies",
documents: {
mode: "selected",
items: [{ id: "doc-1", name: "Private file" }],
},
},
],
});
fireEvent.click(screen.getByTestId("knowledge-scope-trigger"));
await screen.findByText("Policies");
expect(
screen.getByRole<HTMLButtonElement>("button", { name: "Apply" }).disabled,
).toBe(true);
});
});

View File

@ -9,6 +9,41 @@ import {
} from "@/core/knowledge/scope";
describe("knowledge scope snapshots", () => {
test("preserves 256-code-point dataset and document ids and rejects 257", () => {
const selection = (
datasetId: string,
documentId: string,
): KnowledgeScopeSelection => ({
mode: "selected",
datasets: [
{
id: datasetId,
name: "Dataset",
documents: {
mode: "selected",
items: [{ id: documentId, name: "Document" }],
},
},
],
});
const datasetId = "界".repeat(256);
const documentId = "😀".repeat(256);
const snapshot = buildKnowledgeScopeSnapshot(
selection(datasetId, documentId),
);
expect(snapshot.dataset_ids).toEqual([datasetId]);
expect(snapshot.document_filters).toEqual([
{ dataset_id: datasetId, document_ids: [documentId] },
]);
expect(readKnowledgeScopeSnapshot(snapshot)).toEqual(snapshot);
for (const invalid of [
selection(datasetId + "界", documentId),
selection(datasetId, documentId + "😀"),
]) {
expect(() => buildKnowledgeScopeSnapshot(invalid)).toThrow();
}
});
test("builds selected datasets and file filters with bounded display labels", () => {
const snapshot = buildKnowledgeScopeSnapshot({
mode: "selected",