feat(knowledge): add optional platform connections and hierarchical spaces (phase 3) (#7417)

* feat(knowledge): add optional platform connections and hierarchical spaces

* fix(knowledge): return actionable errors for unavailable policy storage

* perf(knowledge): resolve space policy once per snapshot
This commit is contained in:
王一帆 2026-10-25 19:52:31 +08:00 committed by GitHub
parent 95ffb29e50
commit bd62f96732
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
24 changed files with 1793 additions and 10 deletions

View File

@ -95,12 +95,11 @@ hard-code business logic into core. Working examples:
AgentRuns; POSIX, one process/store; approval and session-authorized recovery; uncertain admissions need reconciliation;
no multi-Gateway failover or hard currency budget.
Knowledge management is an optional extension at
`examples/deerflow-extension-knowledge-management/` (see its `AGENTS.md` and
`CONTRACT.md`). It uses existing plugin/router/principal extension points;
retrieval has no management dependency. Writes require explicit administrator
resource/operation grants and upstream authorization. Provider-managed stages
are not completion evidence; uncertain writes must be reconciled, not retried.
Knowledge management/space policy: `examples/deerflow-extension-knowledge-{management,platform}/`
(scoped `AGENTS.md`, `CONTRACT.md`). Use existing plugin/router/principal contracts;
retrieval stays independent. Space policy intersects administrator/resource/operation
and upstream grants, never grants chat access. SQLite holds plugin-owned space policy.
Provider stages are not completion evidence; reconcile uncertain writes, never retry.
Root runtime configs: copy `config.example.yaml` → `config.yaml` and
`extensions_config.example.json` → `extensions_config.json` (MCP/skills).

View File

@ -1648,6 +1648,11 @@ retrieval citations. Configure explicit user-to-workspace credential bindings;
resource authorization remains with WeKnora. The extension directory includes
its installation guide.
The optional [knowledge platform extension](examples/deerflow-extension-knowledge-platform/README.md)
adds connection policies, hierarchical spaces, members and explicit resource sharing.
Space-routed management retains existing administrator and Provider grants. It is
disabled by default and includes no management page or global chat retrieval ACL.
The optional [knowledge management extension](examples/deerflow-extension-knowledge-management/README.md)
adds administrator-authorized file, URL and manual ingestion, processing status,
reprocessing and cancellation through WeKnora. Parsing, indexing and optional

View File

@ -744,6 +744,10 @@ DeerFlow 可连接租户级 RAGFlow,并通过 `knowledge_search` 按 embedding
需显式配置 DeerFlow 用户到 WeKnora 工作空间的凭据绑定;资源授权仍由 WeKnora 执行。
安装与配置步骤见扩展目录中的英文指南。
可选的[知识平台扩展](examples/deerflow-extension-knowledge-platform/README.md)
提供连接策略、多层级空间、成员和显式资源共享。空间内管理调用仍需现有管理员与
Provider 授权。默认关闭,不包含管理页面,也不改变聊天检索的全局权限。
可选的[知识入库与处理扩展](examples/deerflow-extension-knowledge-management/README.md)
通过 WeKnora 提供管理员授权的文件、网址和手工文本入库、处理状态查询、重新处理与取消。
解析、索引及可选的 Wiki/图谱生成由上游负责。扩展默认关闭,需明确配置资源与操作授权,

View File

@ -0,0 +1,37 @@
"""Real SQLite and policy serialization stay outside async HTTP execution."""
import asyncio
from pathlib import Path
import httpx
import pytest
from deerflow_extension_api.auth import ExtensionPrincipal
from fastapi import FastAPI
@pytest.mark.asyncio
async def test_platform_policy_and_effective_http_do_not_block(tmp_path, monkeypatch):
root = Path(__file__).resolve().parents[3]
for package in ("knowledge-platform", "knowledge-management"):
monkeypatch.syspath_prepend(str(root / "examples" / f"deerflow-extension-{package}"))
from deerflow_extension_knowledge_platform.models import Settings, resource_key
from deerflow_extension_knowledge_platform.router import build_router
from deerflow_extension_knowledge_platform.service import Platform
from deerflow_extension_knowledge_platform.store import Store
settings = Settings.model_validate({"enabled": True, "storage_path": str(tmp_path / "platform.db"), "connections": [{"id": "test", "provider": "weknora", "resources": [{"base_id": "kb", "dataset_id": "wk1_synthetic"}]}]})
store = await asyncio.to_thread(Store, settings)
app = FastAPI()
app.include_router(build_router(Platform(settings, store, {})))
app.state.deerflow_extension_principal_resolver = lambda request: ExtensionPrincipal("admin", is_admin=True)
async def authorize(request, namespace, scope):
return True
app.state.deerflow_extension_plugin_authz_resolver_async = authorize
policy = {"spaces": {"root": {"title": "Synthetic", "resources": [resource_key("weknora", "test", "kb")], "members": {"admin": "manager"}}}}
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
assert (await client.put("/api/extensions/knowledge-platform/policy", json={"revision": 0, "policy": policy})).status_code == 200
assert (await client.get("/api/extensions/knowledge-platform/policy")).json()["revision"] == 1
assert len((await client.get("/api/extensions/knowledge-platform/spaces")).json()["items"]) == 1
assert (await client.get("/api/extensions/knowledge-platform/spaces/root")).json()["resources"][0]["dataset_id"] == "wk1_synthetic"

View File

@ -62,6 +62,7 @@ EXPECTED_GUIDANCE_PATHS = {
"examples/deerflow-extension-workflows/AGENTS.md",
"examples/deerflow-extension-weknora/AGENTS.md",
"examples/deerflow-extension-knowledge-management/AGENTS.md",
"examples/deerflow-extension-knowledge-platform/AGENTS.md",
"examples/deerflow-extension-lightrag/AGENTS.md",
}

View File

@ -0,0 +1,740 @@
"""Synthetic platform policy, persistence and management composition regressions."""
import asyncio
from pathlib import Path
import pytest
from deerflow_extension_api.auth import ExtensionPrincipal
@pytest.fixture
def platform(tmp_path, monkeypatch):
root = Path(__file__).resolve().parents[2]
for package in ("knowledge-platform", "knowledge-management"):
monkeypatch.syspath_prepend(str(root / "examples" / f"deerflow-extension-{package}"))
from deerflow_extension_knowledge_platform.models import Settings, resource_key
from deerflow_extension_knowledge_platform.store import Store
resource = resource_key("weknora", "local", "kb")
settings = Settings.model_validate(
{"enabled": True, "storage_path": str(tmp_path / "platform.db"), "connections": [{"id": "local", "provider": "weknora", "provider_config": {}, "resources": [{"base_id": "kb", "dataset_id": "wk1_fixture"}]}]}
)
store = Store(settings)
return store, settings, resource
ADMIN = ExtensionPrincipal("alice", is_admin=True)
MEMBER = ExtensionPrincipal("bob")
def document(resource):
return {
"connections": {"local": {}},
"spaces": {
"org": {"title": "Organization", "resources": [resource], "members": {"alice": "manager", "bob": "viewer"}},
"division": {"title": "Division", "parent_id": "org", "processing": {"summary_enabled": True}},
"team": {"title": "Team", "parent_id": "division", "operations": ["capabilities", "status", "manual"], "processing": {"summary_enabled": False}},
"project": {"title": "Project", "parent_id": "team"},
},
}
def test_four_level_inheritance_and_restart(platform):
from deerflow_extension_knowledge_platform.store import Store
store, settings, resource = platform
result = store.replace(ADMIN, document(resource), 0)
assert result["revision"] == 1
effective = Store(settings).effective(ADMIN, "project")
assert effective["resources"][0]["dataset_id"] == "wk1_fixture"
assert effective["resources"][0]["processing"] == {"summary_enabled": False}
assert set(effective["resources"][0]["operations"]) == {"capabilities", "status", "manual"}
assert effective["role"] == "manager"
def deep_policy(settings, directory, resources_per_connection=40, connections=5):
from deerflow_extension_knowledge_platform.models import Settings, resource_key
from deerflow_extension_knowledge_platform.store import Store
config = settings.model_dump()
config["storage_path"] = str(directory / "deep.db")
config["connections"] = [{"id": f"c{i}", "provider": "weknora", "resources": [{"base_id": f"b{j}", "dataset_id": f"wk1_{i}_{j}"} for j in range(resources_per_connection)]} for i in range(connections)]
store = Store(Settings.model_validate(config))
spaces = {f"s{i}": {"title": f"Space {i}", "parent_id": f"s{i - 1}"} for i in range(1, 500)}
spaces["s0"] = {"title": "Root", "resources": list(store.resources), "members": {"alice": "manager", "bob": "viewer"}}
spaces["s250"]["processing"] = {"chunk_size": 256, "chunk_overlap": 32}
spaces["s499"].update(processing={"summary_enabled": False}, default_tool=None, tools=[])
# Resolve leaves before parents to exercise arbitrary policy JSON order.
policy = {
"defaults": {"default_tool": "knowledge_search", "processing": {"summary_enabled": True}},
"connections": {f"c{i}": {"processing": {"graph_enabled": bool(i % 2)}, "default_tool": "knowledge_entries_search" if i == 4 else "knowledge_search"} for i in range(connections)},
"spaces": dict(reversed(list(spaces.items()))),
}
return store, policy, resource_key("weknora", "c4", "b0")
@pytest.mark.parametrize(("connections", "resources_per_connection"), [(5, 1), (5, 40), (20, 200)])
def test_deep_policy_does_not_reserialize_processing_per_resource_ancestor(platform, tmp_path, monkeypatch, connections, resources_per_connection):
from deerflow_extension_knowledge_management.contracts import ProcessingOptions
_, settings, _ = platform
store, policy, resource = deep_policy(settings, tmp_path, resources_per_connection, connections)
original = ProcessingOptions.model_dump
dumps = 0
def counted_dump(self, *args, **kwargs):
nonlocal dumps
dumps += 1
# A depth/resource product used to perform 25M dumps for this valid tree.
assert dumps <= 2000, "Processing serialization multiplied by resource count and ancestor depth"
return original(self, *args, **kwargs)
monkeypatch.setattr(ProcessingOptions, "model_dump", counted_dump)
store.replace(ADMIN, policy, 0)
effective = store.effective(MEMBER, "s499")
assert effective["role"] == "viewer"
assert len(effective["resources"]) == resources_per_connection * connections
item = next(item for item in effective["resources"] if item["resource_id"] == resource)
assert item["processing"] == {"summary_enabled": False, "graph_enabled": False, "chunk_size": 256, "chunk_overlap": 32}
assert item["default_tool"] is None and item["tools"] == []
assert item["operations"] == ["capabilities", "status"]
inherited = next(item for item in store.effective(MEMBER, "s498")["resources"] if item["resource_id"] == resource)
assert inherited["default_tool"] == "knowledge_entries_search"
def test_listing_deep_spaces_does_not_build_processing_or_resource_projections(platform, tmp_path, monkeypatch):
from deerflow_extension_knowledge_management.contracts import ProcessingOptions
_, settings, _ = platform
store, policy, _ = deep_policy(settings, tmp_path, resources_per_connection=1)
store.replace(ADMIN, policy, 0)
def unexpected(*args, **kwargs):
pytest.fail("Space listing materialized management processing/resource projections")
monkeypatch.setattr(store, "_effective", unexpected)
monkeypatch.setattr(ProcessingOptions, "model_dump", unexpected)
result = store.spaces(MEMBER)
assert result["revision"] == 1
assert len(result["items"]) == 500
assert all(set(item) == {"space_id", "title", "role"} and item["role"] == "viewer" for item in result["items"])
@pytest.mark.parametrize("change", ["resource_removed", "tool_revoked", "connection_disabled", "ancestor_disabled", "member_blocked", "ancestor_revoked"])
def test_listing_keeps_detail_admission_after_policy_or_operator_changes(platform, change):
from deerflow_extension_knowledge_platform.models import Settings
from deerflow_extension_knowledge_platform.store import Store
store, settings, resource = platform
data = document(resource)
data["defaults"] = {"default_tool": "knowledge_search"}
data["spaces"]["unrelated"] = {"title": "Unrelated", "resources": [], "members": {"bob": "viewer"}}
expected = {"org", "division", "team", "project", "unrelated"}
if change == "connection_disabled":
data["connections"]["local"]["enabled"] = False
elif change == "ancestor_disabled":
data["spaces"]["division"]["enabled"] = False
expected -= {"division", "team", "project"}
elif change == "member_blocked":
data["spaces"]["team"]["members"] = {"bob": "blocked"}
expected -= {"team", "project"}
elif change == "ancestor_revoked":
del data["spaces"]["org"]["members"]["bob"]
expected = {"unrelated"}
store.replace(ADMIN, data, 0)
changed = settings.model_dump()
if change == "resource_removed":
changed["connections"][0]["resources"] = [{"base_id": "replacement", "dataset_id": "wk1_replacement"}]
expected = {"unrelated"}
elif change == "tool_revoked":
changed["connections"][0]["resources"][0]["tools"] = []
expected = {"unrelated"}
store = Store(Settings.model_validate(changed))
assert {item["space_id"] for item in store.spaces(MEMBER)["items"]} == expected
for space_id in data["spaces"]:
if space_id in expected:
assert store.effective(MEMBER, space_id)["role"] == "viewer"
else:
with pytest.raises(PermissionError):
store.effective(MEMBER, space_id)
def test_blocked_members_cannot_be_unblocked_below_or_leak_into_siblings(platform):
store, _, resource = platform
data = document(resource)
data["spaces"]["team"]["members"] = {"bob": "blocked"}
data["spaces"]["sibling"] = {"title": "Sibling", "parent_id": "division"}
store.replace(ADMIN, data, 0)
assert {item["space_id"] for item in store.spaces(MEMBER)["items"]} == {"org", "division", "sibling"}
data["spaces"]["project"]["members"] = {"bob": "viewer"}
with pytest.raises(ValueError):
store.replace(ADMIN, data, 1)
assert store.read(ADMIN)["revision"] == 1
def test_members_cannot_edit_or_escalate(platform):
store, _, resource = platform
store.replace(ADMIN, document(resource), 0)
with pytest.raises(PermissionError):
store.replace(MEMBER, document(resource), 1)
effective = store.effective(MEMBER, "project")
assert set(effective["resources"][0]["operations"]) == {"capabilities", "status"}
with pytest.raises(PermissionError):
store.effective(ExtensionPrincipal("outsider", is_admin=True), "project")
@pytest.mark.parametrize("change", ["cycle", "missing_parent", "resource_widening", "operation_widening", "role_widening", "unknown_connection"])
def test_invalid_graph_rolls_back(platform, change):
store, _, resource = platform
data = document(resource)
store.replace(ADMIN, data, 0)
if change == "cycle":
data["spaces"]["org"]["parent_id"] = "project"
elif change == "missing_parent":
data["spaces"]["team"]["parent_id"] = "missing"
elif change == "resource_widening":
data["spaces"]["team"]["resources"] = ["unmapped"]
elif change == "operation_widening":
data["spaces"]["project"]["operations"] = ["upload"]
elif change == "role_widening":
data["spaces"]["project"]["members"] = {"bob": "manager"}
else:
data["connections"]["evil"] = {}
with pytest.raises(ValueError):
store.replace(ADMIN, data, 1)
assert store.read(ADMIN)["revision"] == 1
def test_ancestor_revocation_and_connection_disable(platform):
store, _, resource = platform
data = document(resource)
store.replace(ADMIN, data, 0)
del data["spaces"]["org"]["members"]["bob"]
store.replace(ADMIN, data, 1)
with pytest.raises(PermissionError):
store.effective(MEMBER, "project")
data["connections"]["local"]["enabled"] = False
store.replace(ADMIN, data, 2)
assert store.effective(ADMIN, "project")["resources"] == []
def test_sharing_is_explicit_and_same_named_resources_do_not_alias(platform):
from deerflow_extension_knowledge_platform.models import resource_key
store, _, resource = platform
assert resource != resource_key("weknora", "other", "kb")
data = document(resource)
data["spaces"]["other"] = {"title": "Other organization", "resources": [], "members": {"bob": "viewer"}}
store.replace(ADMIN, data, 0)
assert store.effective(MEMBER, "other")["resources"] == []
data["spaces"]["other"]["resources"] = [resource]
store.replace(ADMIN, data, 1)
assert store.effective(MEMBER, "other")["resources"][0]["resource_id"] == resource
def test_concurrent_writers_conflict_instead_of_losing_update(platform):
from concurrent.futures import ThreadPoolExecutor
from deerflow_extension_knowledge_platform.store import RevisionConflict
store, _, resource = platform
def write():
try:
store.replace(ADMIN, document(resource), 0)
return "saved"
except RevisionConflict:
return "conflict"
with ThreadPoolExecutor(max_workers=2) as pool:
assert sorted(pool.map(lambda _: write(), range(2))) == ["conflict", "saved"]
def _process_writer(settings_data, resource):
from deerflow_extension_knowledge_platform.models import Settings
from deerflow_extension_knowledge_platform.store import RevisionConflict, Store
try:
Store(Settings.model_validate(settings_data)).replace(ADMIN, document(resource), 0)
return "saved"
except RevisionConflict:
return "conflict"
def test_independent_processes_share_the_sqlite_revision_fence(platform):
import multiprocessing
from concurrent.futures import ProcessPoolExecutor
_, settings, resource = platform
with ProcessPoolExecutor(max_workers=2, mp_context=multiprocessing.get_context("spawn")) as pool:
futures = [pool.submit(_process_writer, settings.model_dump(), resource) for _ in range(2)]
assert sorted(future.result(timeout=30) for future in futures) == ["conflict", "saved"]
@pytest.mark.asyncio
async def test_space_management_preserves_principal_and_applies_defaults(platform):
from deerflow_extension_knowledge_management.contracts import Capabilities, IngestRequest, ProcessingResult
from deerflow_extension_knowledge_platform.service import Platform
store, settings, resource = platform
data = document(resource)
store.replace(ADMIN, data, 0)
calls = []
class Provider:
async def capabilities(self, context, connection, base):
return Capabilities(operations={"capabilities", "manual", "status"}, processing_options={"summary_enabled"}, pipeline={})
async def ingest(self, context, connection, base, body):
calls.append((context.principal, connection, base, body.processing))
return ProcessingResult(provider="weknora", connection_id="local", dataset_id="wk1_fixture", document_id="doc", state="queued", provider_state="pending", searchable=False)
service = Platform(settings, store, {"local": Provider()})
body = IngestRequest(kind="manual", title="Synthetic", content="Cedar")
await service.invoke(ADMIN, "project", resource, "manual", body)
assert calls[0][:3] == (ADMIN, "local", "kb")
assert calls[0][3].summary_enabled is False
for principal, operation in [(MEMBER, "manual"), (ExtensionPrincipal("alice", is_admin=True, is_internal=True), "manual"), (ADMIN, "upload")]:
with pytest.raises(PermissionError):
await service.invoke(principal, "project", resource, operation, body)
assert len(calls) == 1
@pytest.mark.asyncio
async def test_policy_denial_before_any_provider_call(platform):
from deerflow_extension_knowledge_platform.service import Platform
store, settings, resource = platform
store.replace(ADMIN, document(resource), 0)
class Provider:
def __getattr__(self, name):
pytest.fail("Denied request contacted provider")
service = Platform(settings, store, {"local": Provider()})
with pytest.raises(PermissionError):
await service.invoke(ADMIN, "project", "unknown-resource", "status", "doc")
@pytest.mark.asyncio
async def test_http_trusted_auth_and_bounded_input(platform):
import httpx
from deerflow_extension_knowledge_platform.router import build_router
from deerflow_extension_knowledge_platform.service import Platform
from fastapi import FastAPI
store, settings, resource = platform
app = FastAPI()
app.include_router(build_router(Platform(settings, store, {})))
principal = ADMIN
scopes = []
app.state.deerflow_extension_principal_resolver = lambda request: principal
async def authorize(request, namespace, scope):
scopes.append((namespace, scope))
return True
app.state.deerflow_extension_plugin_authz_resolver_async = authorize
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
response = await client.put("/api/extensions/knowledge-platform/policy", json={"revision": 0, "policy": document(resource)})
assert response.status_code == 200
response = await client.put("/api/extensions/knowledge-platform/policy", json={"revision": 0, "policy": document(resource)})
assert response.status_code == 409
principal = MEMBER
assert (await client.get("/api/extensions/knowledge-platform/spaces/project")).status_code == 200
assert (await client.get("/api/extensions/knowledge-platform/policy")).status_code == 403
assert (await client.post(f"/api/extensions/knowledge-platform/spaces/project/resources/{resource}/documents", json={"user_id": "alice"})).status_code == 403
principal = ADMIN
assert (await client.put("/api/extensions/knowledge-platform/policy", content=b"x" * (1024 * 1024 + 1))).status_code == 413
assert (await client.put("/api/extensions/knowledge-platform/policy", json={"revision": 1, "policy": document(resource), "user_id": "alice"})).status_code == 422
principal = None
assert (await client.get("/api/extensions/knowledge-platform/spaces")).status_code == 403
assert ("knowledge-platform", "write") in scopes
@pytest.mark.asyncio
@pytest.mark.parametrize("method", ["GET", "PUT"])
async def test_policy_storage_lock_returns_503_without_mutating(platform, monkeypatch, method):
import sqlite3
from contextlib import closing
import httpx
from deerflow_extension_knowledge_platform.router import build_router
from deerflow_extension_knowledge_platform.service import Platform
from fastapi import FastAPI
store, settings, resource = platform
app = FastAPI()
app.include_router(build_router(Platform(settings, store, {})))
app.state.deerflow_extension_principal_resolver = lambda request: ADMIN
async def authorize(*args):
return True
app.state.deerflow_extension_plugin_authz_resolver_async = authorize
# Exercise the real busy-timeout error without waiting the production ten seconds.
monkeypatch.setattr(store, "connect", lambda: sqlite3.connect(store.path, timeout=0.01))
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app, raise_app_exceptions=False), base_url="http://test") as client:
with closing(sqlite3.connect(store.path)) as owner:
owner.execute("BEGIN EXCLUSIVE")
response = await client.request(method, "/api/extensions/knowledge-platform/policy", **({"json": {"revision": 0, "policy": document(resource)}} if method == "PUT" else {}))
assert response.status_code == 503
assert response.json() == {"detail": "Policy storage is busy or unavailable; reload policy before retrying"}
assert str(store.path) not in response.text
snapshot = await client.get("/api/extensions/knowledge-platform/policy")
assert snapshot.status_code == 200 and snapshot.json()["revision"] == 0
saved = await client.put("/api/extensions/knowledge-platform/policy", json={"revision": 0, "policy": document(resource)})
assert saved.status_code == 200 and saved.json()["revision"] == 1
@pytest.mark.asyncio
async def test_unavailable_policy_database_returns_sanitized_503(platform):
import httpx
from deerflow_extension_knowledge_platform.router import build_router
from deerflow_extension_knowledge_platform.service import Platform
from fastapi import FastAPI
store, settings, _ = platform
store.path.write_bytes(b"invalid SQLite database with private diagnostic content")
app = FastAPI()
app.include_router(build_router(Platform(settings, store, {})))
app.state.deerflow_extension_principal_resolver = lambda request: ADMIN
async def authorize(*args):
return True
app.state.deerflow_extension_plugin_authz_resolver_async = authorize
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app, raise_app_exceptions=False), base_url="http://test") as client:
response = await client.get("/api/extensions/knowledge-platform/policy")
assert response.status_code == 503
assert response.json() == {"detail": "Policy storage is busy or unavailable; reload policy before retrying"}
assert "private diagnostic" not in response.text and str(store.path) not in response.text
@pytest.mark.asyncio
async def test_platform_database_work_runs_off_loop(platform, monkeypatch):
from deerflow_extension_knowledge_platform.service import Platform
store, settings, _ = platform
original = store.read
def read(principal):
with pytest.raises(RuntimeError):
asyncio.get_running_loop()
return original(principal)
monkeypatch.setattr(store, "read", read)
assert (await Platform(settings, store, {}).policy(ADMIN))["revision"] == 0
@pytest.mark.asyncio
async def test_revocation_during_capability_lookup_prevents_write(platform):
from deerflow_extension_knowledge_management.contracts import Capabilities, IngestRequest
from deerflow_extension_knowledge_platform.service import Platform
store, settings, resource = platform
data = document(resource)
store.replace(ADMIN, data, 0)
class Provider:
async def capabilities(self, *args):
data["connections"]["local"]["enabled"] = False
await asyncio.to_thread(store.replace, ADMIN, data, 1)
return Capabilities(operations={"manual"}, processing_options={"summary_enabled"}, pipeline={})
async def ingest(self, *args):
pytest.fail("Revoked policy submitted an upstream write")
with pytest.raises(PermissionError):
await Platform(settings, store, {"local": Provider()}).invoke(ADMIN, "project", resource, "manual", IngestRequest(kind="manual", title="Test", content="Test"))
@pytest.mark.asyncio
async def test_real_management_adapter_composition(platform, monkeypatch):
import hashlib
import json
import httpx
from deerflow_extension_knowledge_management.contracts import IngestRequest
from deerflow_extension_knowledge_management.weknora import WeKnoraManagement
from deerflow_extension_knowledge_platform.service import Platform
store, settings, resource = platform
dataset = "wk1_" + hashlib.sha256(json.dumps(["https://weknora.invalid", "1", "kb"], ensure_ascii=False).encode()).hexdigest()
updated = settings.model_dump()
updated["connections"][0]["resources"][0]["dataset_id"] = dataset
from deerflow_extension_knowledge_platform.models import Settings
from deerflow_extension_knowledge_platform.store import Store
settings = Settings.model_validate(updated)
store = Store(settings)
store.replace(ADMIN, document(resource), 0)
monkeypatch.setenv("PLATFORM_SYNTHETIC_KEY", "synthetic-secret")
calls, code = [], [200]
def handler(request):
calls.append(request)
assert request.headers["X-API-Key"] == "synthetic-secret"
if code[0] != 200:
return httpx.Response(code[0], text="synthetic-secret")
if request.url.path == "/api/v1/knowledge-bases/kb":
data = {"id": "kb", "capabilities": {"wiki": True}, "chunking_config": {"chunk_size": 1024, "chunk_overlap": 128}}
else:
data = {"id": "doc", "knowledge_base_id": "kb", "parse_status": "pending", "enable_status": "enabled"}
return httpx.Response(200, json={"success": True, "data": data})
provider = WeKnoraManagement(
{
"connections": [
{
"id": "local",
"base_url": "https://weknora.invalid",
"tenant_id": "1",
"api_key_env": "PLATFORM_SYNTHETIC_KEY",
"grants": [{"user_id": "alice", "knowledge_base_ids": ["kb"], "operations": ["capabilities", "manual", "status"]}],
}
]
},
transport=httpx.MockTransport(handler),
)
service = Platform(settings, store, {"local": provider})
body = IngestRequest(kind="manual", title="Synthetic", content="Cedar")
result = await service.invoke(ADMIN, "project", resource, "manual", body)
assert result.dataset_id == dataset and result.state == "queued"
assert json.loads(calls[-1].content)["process_config"]["summary_enabled"] is False
status = await service.invoke(ADMIN, "project", resource, "status", "doc")
assert status.document_id == "doc"
code[0] = 403
before = len(calls)
with pytest.raises(PermissionError):
await service.invoke(ADMIN, "project", resource, "manual", body)
assert len(calls) == before + 1
# Correct space policy still cannot bypass the existing provider's exact-user grant.
data = document(resource)
data["spaces"]["org"]["members"]["unbound"] = "manager"
store.replace(ADMIN, data, 1)
before = len(calls)
with pytest.raises(PermissionError):
await service.invoke(ExtensionPrincipal("unbound", is_admin=True), "project", resource, "manual", body)
assert len(calls) == before
@pytest.mark.asyncio
async def test_mismatched_result_identity_is_not_remapped(platform):
from deerflow_extension_knowledge_management.contracts import ProcessingResult
from deerflow_extension_knowledge_platform.service import Platform
store, settings, resource = platform
store.replace(ADMIN, document(resource), 0)
class Provider:
async def status(self, *args):
return ProcessingResult(provider="weknora", connection_id="other", dataset_id="wrong", document_id="doc", state="completed", provider_state="completed", searchable=True)
with pytest.raises(RuntimeError, match="identity"):
await Platform(settings, store, {"local": Provider()}).invoke(ADMIN, "project", resource, "status", "doc")
def test_connections_and_tools_limit_effective_capabilities(platform):
store, _, resource = platform
data = document(resource)
data["connections"]["local"] = {"operations": ["status"], "tools": ["knowledge_search"], "default_tool": "knowledge_search"}
store.replace(ADMIN, data, 0)
item = store.effective(ADMIN, "project")["resources"][0]
assert item["operations"] == ["status"]
assert item["tools"] == ["knowledge_search"]
data["spaces"]["project"]["default_tool"] = "entity_relations_search"
with pytest.raises(ValueError):
store.replace(ADMIN, data, 1)
def test_platform_defaults_are_inherited_with_a_permission_ceiling(platform):
store, _, resource = platform
data = document(resource)
data["defaults"] = {"operations": ["capabilities", "status", "manual"], "processing": {"graph_enabled": False}, "tools": ["knowledge_search"], "default_tool": "knowledge_search"}
store.replace(ADMIN, data, 0)
item = store.effective(ADMIN, "project")["resources"][0]
assert item["processing"] == {"graph_enabled": False, "summary_enabled": False}
assert item["tools"] == ["knowledge_search"]
assert item["default_tool"] == "knowledge_search"
assert set(item["operations"]) == {"capabilities", "status", "manual"}
def test_child_can_explicitly_clear_inherited_default_tool(platform):
store, _, resource = platform
data = document(resource)
data["defaults"] = {"default_tool": "knowledge_search"}
data["spaces"]["project"].update(tools=[], default_tool=None)
store.replace(ADMIN, data, 0)
item = store.effective(ADMIN, "project")["resources"][0]
assert item["tools"] == [] and item["default_tool"] is None
snapshot = store.read(ADMIN)
store.replace(ADMIN, snapshot["policy"], 1)
assert store.effective(ADMIN, "team")["resources"][0]["default_tool"] == "knowledge_search"
assert store.effective(ADMIN, "project")["resources"][0]["default_tool"] is None
def test_operator_resource_removal_can_be_reconciled_without_global_outage(platform):
from deerflow_extension_knowledge_platform.models import Settings
from deerflow_extension_knowledge_platform.store import Store
store, settings, resource = platform
data = document(resource)
data["spaces"]["unrelated"] = {"title": "Unrelated", "resources": [], "members": {"alice": "manager"}}
store.replace(ADMIN, data, 0)
changed = settings.model_dump()
changed["connections"][0]["resources"] = [{"base_id": "replacement", "dataset_id": "wk1_replacement"}]
store = Store(Settings.model_validate(changed))
snapshot = store.read(ADMIN)
assert snapshot["revision"] == 1
assert store.effective(ADMIN, "unrelated")["resources"] == []
with pytest.raises(PermissionError):
store.effective(ADMIN, "project")
assert [s["space_id"] for s in store.spaces(ADMIN)["items"]] == ["unrelated"]
data["spaces"]["org"]["resources"] = []
assert store.replace(ADMIN, data, 1)["revision"] == 2
def test_operator_tool_revocation_cannot_leave_an_invalid_default(platform):
from deerflow_extension_knowledge_platform.models import Settings
from deerflow_extension_knowledge_platform.store import Store
store, settings, resource = platform
data = document(resource)
data["defaults"] = {"default_tool": "knowledge_search"}
store.replace(ADMIN, data, 0)
changed = settings.model_dump()
changed["connections"][0]["resources"][0]["tools"] = []
store = Store(Settings.model_validate(changed))
assert store.read(ADMIN)["revision"] == 1
with pytest.raises(PermissionError):
store.effective(ADMIN, "project")
def test_management_namespace_can_be_registered_without_direct_routes(platform):
from deerflow_extension_knowledge_management import install
class Registry:
plugins, mounted = [], []
def plugin(self, contribution):
self.plugins.append(contribution)
return True
def routers(self, routes):
self.mounted.extend(routes)
registry = Registry()
install(registry, {"enabled": True, "routes_enabled": False})
assert len(registry.plugins) == 1
assert registry.plugins[0].namespace == "knowledge-management"
assert not registry.mounted
def test_policy_never_serializes_operator_configuration(platform):
import json
store, _, _ = platform
assert "provider_config" not in json.dumps(store.read(ADMIN))
@pytest.mark.asyncio
async def test_dual_namespace_authorization_is_required(platform):
import httpx
from deerflow_extension_knowledge_platform.router import build_router
from deerflow_extension_knowledge_platform.service import Platform
from fastapi import FastAPI
store, settings, resource = platform
store.replace(ADMIN, document(resource), 0)
app = FastAPI()
app.include_router(build_router(Platform(settings, store, {})))
app.state.deerflow_extension_principal_resolver = lambda request: ADMIN
async def authorize(request, namespace, scope):
return namespace == "knowledge-platform"
app.state.deerflow_extension_plugin_authz_resolver_async = authorize
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
response = await client.get(f"/api/extensions/knowledge-platform/spaces/project/resources/{resource}/capabilities")
assert response.status_code == 403
@pytest.mark.asyncio
@pytest.mark.parametrize("operation", ["upload", "reprocess", "cancel"])
async def test_remaining_management_operations_preserve_contract(platform, operation):
from deerflow_extension_knowledge_management.contracts import Capabilities, ProcessingOptions, ProcessingResult
from deerflow_extension_knowledge_platform.service import Platform
store, settings, resource = platform
data = document(resource)
del data["spaces"]["team"]["operations"]
store.replace(ADMIN, data, 0)
calls = []
result = ProcessingResult(provider="weknora", connection_id="local", dataset_id="wk1_fixture", document_id="doc", state="unknown", provider_state="new_upstream_state", searchable=False)
class Provider:
async def capabilities(self, *args):
return Capabilities(operations={"upload", "reprocess", "cancel"}, processing_options={"summary_enabled"}, pipeline={})
async def upload(self, context, connection, base, filename, content, options):
calls.append((filename, content, options.model_dump(exclude_none=True)))
return result
async def process(self, context, connection, base, document, action, options):
calls.append((document, action, options.model_dump(exclude_none=True)))
return result
value = await Platform(settings, store, {"local": Provider()}).invoke(ADMIN, "project", resource, operation, ProcessingOptions(), document="doc", filename="guide.txt", content=b"synthetic")
assert value == result # No fabricated success for an unknown upstream state.
assert calls == ([("guide.txt", b"synthetic", {"summary_enabled": False})] if operation == "upload" else [("doc", operation, {})])
@pytest.mark.asyncio
async def test_unsupported_options_do_not_mutate_or_retry(platform):
from deerflow_extension_knowledge_management.contracts import Capabilities, IngestRequest
from deerflow_extension_knowledge_platform.service import Platform
store, settings, resource = platform
store.replace(ADMIN, document(resource), 0)
calls = []
class Provider:
async def capabilities(self, *args):
calls.append("capabilities")
return Capabilities(operations={"manual"}, processing_options=set(), pipeline={})
async def ingest(self, *args):
pytest.fail("Unsupported option submitted a write")
with pytest.raises(NotImplementedError):
await Platform(settings, store, {"local": Provider()}).invoke(ADMIN, "project", resource, "manual", IngestRequest(kind="manual", title="Synthetic", content="Test"))
assert calls == ["capabilities"]
def test_install_is_opt_in_and_uses_only_operator_factory(platform, monkeypatch):
from types import SimpleNamespace
from deerflow_extension_knowledge_platform import install
_, settings, _ = platform
calls = []
class Registry:
plugins, mounted = [], []
def plugin(self, contribution):
self.plugins.append(contribution)
return True
def routers(self, routes):
self.mounted.extend(routes)
registry = Registry()
install(registry, {})
assert not registry.plugins
monkeypatch.setattr("deerflow_extension_knowledge_platform.import_module", lambda module: SimpleNamespace(WeKnoraManagement=lambda config: calls.append(config) or object()))
install(registry, settings.model_dump())
assert len(registry.plugins) == len(registry.mounted) == len(calls) == 1
assert registry.plugins[0].namespace == "knowledge-platform"
assert registry.plugins[0].frontend is None

View File

@ -45,3 +45,7 @@ Capability options must reflect the fetched base: omit graph controls without
graph support and chunk overrides for missing/parent-child chunking config.
Local credential validation raises ManagementCredentialUnavailable before HTTP;
map it to 503 without uncertain-write reconciliation guidance.
`routes_enabled: false` preserves the enabled plugin namespace for downstream
authorization while mounting no direct routes or Provider. Keep the default true
for compatibility; disabling the entire install also removes the namespace.

View File

@ -43,6 +43,8 @@ Every route requires all of:
This initial authorization mode is **explicit administrator resource grants with service credentials**. It does not claim upstream user impersonation, general organization RBAC, Agent-delegated writes, or platform-space authorization. Existing WeKnora retrieval `user_ids` confer no management rights. The management grants do not grant retrieval access. Client-provided user, Agent, tenant and space fields are rejected. Future multi-user/space management must supply a separately reviewed authorization implementation; do not remove the administrator floor to imply those capabilities exist.
To retain the registered `knowledge-management` namespace for space-routed authorization while disabling these direct routes, set `enabled: true` and `routes_enabled: false`. In that mode no Provider factory is instantiated; the platform extension supplies its own operator-configured Provider. The default `routes_enabled: true` preserves existing installations.
## HTTP contract, version 1
All routes have this prefix:

View File

@ -13,6 +13,7 @@ from .router import NAMESPACE, build_router
class Settings(Model):
enabled: bool = Field(default=False, strict=True)
routes_enabled: bool = Field(default=True, strict=True)
provider_factory: str = "deerflow_extension_knowledge_management.weknora:WeKnoraManagement"
provider_config: dict[str, Any] = Field(default_factory=dict)
@ -28,9 +29,10 @@ def install(registry, config):
settings = Settings.model_validate(dict(config))
if not settings.enabled:
return
# Executable factory paths are operator-owned configuration, never HTTP input.
module, attribute = settings.provider_factory.split(":", 1)
provider = getattr(import_module(module), attribute)(settings.provider_config)
if not registry.plugin(PluginContribution(namespace=NAMESPACE, title="Knowledge management", enabled=True, backend=(BackendAction("contract", describe_contract),))):
raise RuntimeError("Host does not support knowledge management plugin admission")
registry.routers((build_router(provider),))
if settings.routes_enabled:
# Executable factory paths are operator-owned configuration, never HTTP input.
module, attribute = settings.provider_factory.split(":", 1)
provider = getattr(import_module(module), attribute)(settings.provider_config)
registry.routers((build_router(provider),))

View File

@ -0,0 +1,34 @@
# Knowledge platform
Keep platform connections, logical spaces and upstream tenants separate. This
optional plugin uses public principal/plugin authorization plus the existing
management Provider; do not introduce a second ingestion implementation or core
knowledge/ACL registry. No host-internal imports.
All policy writes require the host administrator floor and platform write grant.
Every space-routed management request additionally requires the management
namespace grant, explicit space membership, resource/operation policy and the
unchanged Provider's exact-user/base/operation authorization. Never manufacture
an administrator, accept client identities, silently widen filters, substitute
credentials or retry a mutation. Config/defaults and resource references are not
grants; inherited child sets/members must remain within ancestor limits.
Executable Provider factories/configuration are operator-only. HTTP policy must
never persist credentials, endpoints or factory paths. Use absolute private
SQLite storage, transactional revision checks and bounded validated snapshots.
Run all filesystem/SQLite work off async request loops; add strict blocking-I/O
regressions. Do not hold a database transaction across Provider HTTP calls.
Map SQLite failures to a sanitized 503; require policy reload before retrying
an update and never automatically retry Provider mutations.
Resolve ancestors incrementally within one policy snapshot. Never cache grants
across requests. Space listing must share detail admission checks without
materializing per-resource processing responses or repeating ancestor merges.
Keep resource identity provider/connection/native-base-qualified and bounded;
preserve retrieval dataset IDs and validate management result identities. Space
policies apply only to this plugin's routes. Do not claim global retrieval ACL,
upstream user impersonation, management pages, Trace or multi-host replication.
Never infer processing success or progress from policy/capability presence.
Tests use real SQLite and HTTP routing, mock only upstream transport. Real
upstream acceptance must be explicitly distinguished from synthetic coverage.

View File

@ -0,0 +1,21 @@
# Platform-space handoff, version 1
This is an extension-owned contract, separate from core retrieval and the management Provider protocol. Resource references are not grants. A logical space is not an upstream tenant, connection, dataset or user's identity.
## Management and pages
Read the administrator policy using its revision, edit it, then submit an atomic replacement. `409` requires reloading and reconciling; never silently overwrite another administrator's change. The policy can enable/disable declared connections and constrain operations/tools/defaults, not create new executable Provider configurations. Provider settings and credentials stay operator-owned.
`GET /spaces` and `/spaces/{space}` use the authenticated host principal and return only authorized projections. `operations` is a **space policy ceiling**, not proof of Provider support or host/admin authorization. Actual capabilities require the space resource capability endpoint, which rechecks both plugin namespaces, the administrator floor, space membership and the existing Provider's resource grant. Pages must distinguish a visible policy from permission to submit a write. Hiding a page is never a server authorization check.
Processing/default configuration is inherited, while resource/operation/tool permissions can only narrow. Child membership cannot exceed ancestor membership. An ancestor restriction that invalidates descendant declarations must narrow those declarations in the same atomic update; disabled ancestors/connections deny all new requests. Editing the policy is restricted to host administrators with platform write permission, regardless of a member's `manager` label.
`resource_id` is a bounded hash of `(provider, connection_id, native_base_id)`, not a replacement retrieval scope ID. Keep `dataset_id` unchanged and preserve the existing management response's `document_id`. Management results cannot create citations or retrieval permissions. Retries of uncertain writes still need upstream reconciliation.
## Trace and retrieval consumers
This stage does not emit processing spans, run IDs or fabricated task history. Use real management snapshots and the original Provider contract. A future Trace implementation must independently authorize access and consume actual events; it cannot infer step success from platform capability, default configuration or terminal document state.
No current chat tool consumes these space grants. Any future retrieval consumer must explicitly bind trusted user/executor/space context, intersect operator and upstream authorization, honor all selected document filters and validate sources through existing contracts. A client space ID, raw resource mapping or tool name alone must never confer access.
The policy revision identifies a checked snapshot, not a durable execution lease. Mutations recheck policy after the upstream capability lookup before submission. The last platform admission check is the boundary: later revocation does not cancel an admitted Provider call, including its internal HTTP checks or upstream processing. There is no exactly-once admission promise or cross-host policy replication.

View File

@ -0,0 +1,127 @@
# Optional knowledge platform extension
Stage 3 of [#6269's five-stage follow-up](https://github.com/bytedance/deer-flow/issues/6269#issuecomment-6281785410): platform connection policies, hierarchical logical spaces, members and explicit resource sharing. It composes the management contracts delivered by #7293 and installs no page, model tool, parser, index or Trace service. The existing retrieval and management extensions continue to work independently.
## Install and configure
From the repository root, install the local management package first so its source is available to the dependency resolver, then this package:
```sh
make extension-install SOURCE=./examples/deerflow-extension-knowledge-management
make extension-install SOURCE=./examples/deerflow-extension-knowledge-platform
```
The management namespace must be registered for the host's second authorization check. Enable its entry point with `routes_enabled: false` to retain that namespace without exposing direct management routes or requiring a second Provider configuration. Configure the platform entry point in operator-owned `config.yaml` and restart Gateway. Use an absolute private SQLite path on local storage. Multiple workers on the same machine can share that file; this is not a multi-host database service.
```yaml
plugins:
- use: deerflow_extension_knowledge_management:install
config:
enabled: true
routes_enabled: false
- use: deerflow_extension_knowledge_platform:install
config:
enabled: true
storage_path: /absolute/private/path/knowledge-platform.db
connections:
- id: reference
provider: weknora
# Another trusted module:factory can implement ManagementProvider.
provider_factory: deerflow_extension_knowledge_management.weknora:WeKnoraManagement
provider_config:
connections:
- id: reference
base_url: https://weknora.example
tenant_id: "1"
api_key_env: WEKNORA_MANAGEMENT_KEY
grants:
- user_id: exact-deerflow-admin-user-id
knowledge_base_ids: [native-base-id]
operations: [capabilities, status, manual, url, upload, reprocess, cancel]
resources:
- base_id: native-base-id
# Obtain the exact ID from the existing retrieval catalog or a
# management snapshot for this endpoint/tenant/native base.
dataset_id: wk1_replace_with_the_existing_dataset_id
operations: [capabilities, status, manual, url, upload, reprocess, cancel]
tools: [knowledge_search, knowledge_entries_search]
```
Set the key only in the Gateway environment. IDs, Provider factories, endpoints, grants, resource allowlists and credential references are deployment configuration. HTTP can change connection display names, enabled state and policy within that envelope; it cannot create executable factories, change endpoints or grant new upstream credentials. Adding another Provider connection or upstream base requires an operator config change and restart. Do not reuse a connection ID for a different upstream endpoint/tenant: allocate a new ID and explicitly remap spaces. An invalid dataset mapping is rejected when validating the management response; after a mutation this may be an uncertain write requiring reconciliation.
Installation is disabled by default. Removing the entry point or setting `enabled: false` and restarting removes the routes without deleting the private policy database. Back up that database with normal SQLite backup procedures; do not commit it or expose it as a downloadable browser asset.
## Policy and API
All routes are under `/api/extensions/knowledge-platform`. Gateway authentication, CSRF protection for writes and the host's namespace-specific plugin management authorization apply. No request body supplies a user, Agent, tenant or credential. The plugin registers its namespace through `PluginContribution`; no new core registry is needed.
| Route | Purpose |
| --- | --- |
| `GET /policy` | Administrator-only policy snapshot, revision and operator resource identities; no provider config or secrets |
| `PUT /policy` | Administrator-only atomic replacement with the expected revision; `409` on a stale revision |
| `GET /spaces` | Current member's accessible space IDs, titles and roles; no other members' identities |
| `GET /spaces/{space}` | Current member's effective resources, operation/tool policy and processing defaults |
| `GET /spaces/{space}/resources/{resource}/capabilities` | Space policy intersected with actual management Provider capabilities |
| `POST /spaces/{space}/resources/{resource}/documents` | Existing URL/manual `IngestRequest`; `202` is a processing snapshot |
| `POST /spaces/{space}/resources/{resource}/files?filename=guide.txt` | Raw octet-stream, at most 16 MiB; optional `X-Processing-Options` |
| `GET /spaces/{space}/resources/{resource}/documents/{document}` | Existing `ProcessingResult`, including unknown states |
| `POST /spaces/{space}/resources/{resource}/documents/{document}/{reprocess\|cancel}` | Existing processing options and uncertainty semantics |
First read `/policy` to discover the `kp1_...` resource ID derived from `(provider, connection_id, native_base_id)`. This local mapping identity is separate from the retrieval `dataset_id`. Two connections with the same native base name do not alias. The server checks management response provider, connection, dataset and requested document identities; it never rewrites an unrelated response into the selected resource.
Submit the whole policy with the last observed revision. The initial revision is `0`. This example creates four levels; the protocol has no fixed hierarchy depth. The bounded store allows 500 total spaces, 200 members per space and 1 MiB of policy JSON. Space/connection names use bounded ASCII IDs; titles and exact host user IDs support Unicode.
```json
{
"revision": 0,
"policy": {
"defaults": {"default_tool": "knowledge_search"},
"connections": {"reference": {"title": "Team reference", "enabled": true}},
"spaces": {
"organization": {
"title": "Organization",
"resources": ["kp1_replace_with_resource_id_from_GET_policy"],
"members": {"exact-deerflow-admin-user-id": "manager", "another-host-user-id": "viewer"}
},
"division": {"title": "Division", "parent_id": "organization"},
"team": {"title": "Team", "parent_id": "division", "operations": ["capabilities", "status", "manual"]},
"project": {"title": "Project", "parent_id": "team", "processing": {"summary_enabled": false}}
}
}
}
```
To share a base with another organization, an administrator explicitly adds the same resource ID to that root's envelope and assigns its members. Sharing no longer needs a duplicate base or an alias to a tenant. Child spaces inherit resources, operations, tools and members when fields are absent. An empty set removes access; an explicit child set must be a subset of the parent. Child membership can only narrow ancestor grants. Roles are `blocked`, `viewer`, `editor`, `manager`; viewers have only status/capability operations. Editors and managers can request configured management operations, but all actual management requests retain the additional administrator floor below. The `manager` label does not grant host administrator status or policy-edit privileges.
The entire updated tree is checked before publication: cycles, missing parents, unknown resources/connections and child permission expansion reject the transaction. Reparenting, deletion and ancestor restrictions must leave descendants valid; when necessary narrow descendant grants in the same replacement. Disabling an ancestor or connection denies future admissions. Policy writes use `BEGIN IMMEDIATE` and compare revisions inside the transaction, preventing lost updates across independent connections/processes. A Provider call already admitted by the platform is in flight and is not cancelled by a later policy change, including its internal HTTP checks. Removing operator resources denies affected spaces while administrators can still read the old policy/revision and `unavailable_resources` to reconcile it; unrelated spaces remain available.
Processing defaults merge in this order: platform → connection → ancestors → selected space → explicit ingestion options. These are defaults, not an authorization grant. Operation and tool permissions intersect; configuration inheritance cannot expand those permissions. Unsupported options fail before mutation. Empty reprocess options retain the existing document's upstream policy; new-document defaults are not silently applied to old documents. A selected default retrieval tool must remain in the effective tool envelope. An absent `default_tool` inherits; explicit `null` clears it, including when a child disables all tools. Policy round trips preserve this distinction.
## Authorization boundary
Space-routed management requires **all** of:
1. Host `knowledge-platform` plugin read/write authorization.
2. Host `knowledge-management` plugin read/write authorization, even if its direct routes are disabled.
3. A host-resolved human administrator and an effective space membership/operation grant.
4. The unchanged management adapter's exact-user, native-base and operation grants.
5. Upstream resource authorization, including document/base membership checks.
The same principal is passed to the Provider. No member is promoted to an administrator, and no fallback credential or automatic write retry exists. A space grant alone cannot authorize an upstream mutation. Non-admin members can inspect their effective policy when the host grants platform read permission; they cannot invoke the initial administrator-only management backend. Enabling non-admin management or upstream per-user delegation requires a separate reviewed authorization change.
**This plugin enforces space policy on its own routes.** Tool/default policy in its effective projection is for future consumers; installing it does not reroute `knowledge_search`, modify saved `knowledge_scope`, add a chat space selector or change existing citations. Existing direct management routes keep their independent administrator grants. If every management request must cross space admission, keep its namespace entry point enabled with `routes_enabled: false` and expose only this plugin's management routes. Space sharing neither grants chat retrieval nor removes an independently configured retrieval grant. Do not present this plugin as a global retrieval ACL.
Management responses preserve #7293's contract: capability is not observed execution, `202` is not completion, searchable is not success of every Wiki/graph step, and unknown stays unknown. Writes are never retried. On timeouts, lost/malformed responses or identity mismatches, reconcile with the selected upstream base before submitting again. There is no exactly-once ledger. Errors omit provider messages/credentials; database or Provider exceptions are not exposed to clients. See [CONTRACT.md](CONTRACT.md) for the fourth/fifth-stage handoff.
SQLite failures, including an exhausted busy timeout when workers share the policy file, return a sanitized `503` without database paths or exception details. Reload the policy and its revision before deciding whether to retry a policy update; ingestion and processing writes are never automatically retried.
Policy evaluation resolves each ancestor once within the current snapshot, combines processing defaults per connection and reuses resource-envelope checks. Space listing performs the same admission checks as space details without constructing their resource/processing responses. These temporary calculations are discarded after each call; grants are never cached across requests or revisions.
## Verification
```sh
uv run --no-sync --project backend pytest backend/tests/test_knowledge_platform_extension.py -q
uv run --no-sync --project backend pytest backend/tests/blocking_io/test_knowledge_platform_extension.py -q
```
The offline tests use real SQLite, FastAPI routing and the existing WeKnora management adapter with synthetic HTTP responses. They establish policy enforcement and contract composition, not real ingestion/extraction quality or production identity integration. [VALIDATION.md](VALIDATION.md) records the checks and current acceptance boundary. The earlier `superbuzzy/deer-flow` branch informed the space/resource separation; its core persistence, old Provider loader and 512-character scope IDs are not transplanted into main.

View File

@ -0,0 +1,154 @@
# Stage 3 validation record
Date: 2026-10-24. Base: upstream `9a11f9ae`. Environment: native Windows,
CPython 3.12.15, Node 24.16.0, pnpm 10.26.2. This records the platform/space
backend in the new five-stage follow-up to #6269, after #7171 and #7293.
## Verified behavior
The new backend tests use real SQLite and FastAPI routing. The management bridge
also exercises the existing WeKnora adapter through `httpx.MockTransport` with
synthetic resources and responses. It does not contact a real WeKnora service.
- Four-level spaces, inherited defaults, explicit sharing, exact member roles,
administrator-only policy changes and restart persistence.
- Rejection of cycles, missing parents, child resource/operation/member expansion,
stale revisions and unauthorized resources before Provider submission.
- Transactional revision fencing across threads and independently spawned
processes; one writer wins and the stale writer receives a conflict.
- Ancestor revocation, connection disablement, removed operator resources,
administrator recovery and continued access to unaffected spaces.
- Tool ceilings, inherited default tools, explicit `null` clearing across policy
round trips, and rejection when operator tool revocation invalidates a default.
- The same authenticated administrator reaches the original management Provider;
both plugin namespace grants, space policy and upstream grants remain required.
- Capability/default composition, upload, ingestion, status, reprocess and cancel;
unsupported options fail before mutation and uncertain writes are not retried.
- Response identity checks preserve dataset/document identity. Unknown processing
states remain unknown. Revocation during capability discovery prevents a write.
- Bounded request parsing, rejection of forged identity/configuration fields,
secret-free policy responses and SQLite offloading under the blocking-I/O gate.
## Passing checks
From `backend/`, the final combined regression command passed **272 tests**:
```sh
uv run pytest tests/test_extension_route_principal.py tests/test_extension_gateway_wiring.py tests/test_extension_registry.py tests/test_plugin_targets.py tests/test_knowledge_platform_extension.py tests/test_knowledge_management_extension.py tests/blocking_io/test_knowledge_platform_extension.py tests/blocking_io/test_knowledge_management_extension.py tests/test_agent_guidance_check.py -q --tb=short
```
This includes **32 new platform tests and one strict blocking-I/O test**. Backend
`ruff check .` and `ruff format --check .` passed, as did Ruff on both example
packages. `python scripts/check_agent_guidance.py` reported 0 errors and 15 existing
soft budget warnings. Both extension wheels passed `scripts/check_extension.py`.
The real `ExtensionManager.install(..., yes=True)` source-snapshot workflow also
installed management and platform, in that order, into a separate temporary
project and virtual environment with extension-api 0.2.9. Installed entry points
loaded successfully: both were inert by default; enabling the management
namespace without direct routes and then platform registered both namespaces,
nine platform routes and a policy that survived reopening its SQLite store.
The probe used the real WeKnora factory but made no upstream requests.
Frontend `pnpm check`, formatting checks for changed files and the extension
catalog unit suite passed (**10 tests**). A production build passed using
`BETTER_AUTH_SECRET=local-dev-secret pnpm exec next build --webpack`.
The default Turbopack build could not use the native SWC binary because Windows
Application Control blocked it; its WASM fallback does not support Turbopack.
Against that production build, Chromium ran the existing extension-directory
cases for English, Chinese and failed runtime discovery: **3 passed**. An initial
English attempt saw duplicate search inputs during loading; its isolated rerun
and the subsequent complete three-case run passed without changing the tests.
These cases validate discovery metadata, not a knowledge management page.
## Full-suite results and baseline comparison
The full suites were run, but are **not reported as green** on this Windows host.
- Four backend offline shards initially reported 35,872 passed, 105 failed,
21 errors and 1,847 skipped. That run overlapped implementation and included
one platform test whose fake response was subsequently corrected to the
validated management result contract. The final platform/management regression
selection above passes.
- Replaying the other failing/error nodes on an untouched `9a11f9ae` worktree
reproduced 96 failures and 21 errors; eight nodes passed on that replay.
Failures include Windows filesystem/process behavior and existing test fixture
errors. A narrower replay on this branch passed eight transient cases; the
goal-worker fixture error also reproduced separately on the untouched base.
This is evidence of baseline and transient failures, not a claim that every
full-suite failure has been conclusively classified.
- The full strict blocking-I/O suite reported 274 passed, 8 skipped and one
existing dynamic-context memory timeout failure. That timeout test passed
separately on both this branch and the untouched base. The new platform
blocking-I/O test passes in the combined regression selection.
- The full frontend suite reported 3,477 passed and one task-form pending-capability
timeout. That test file passed separately on this branch and the untouched
base: 37 passed in each run.
CI on the PR is a separate result; these local checks do not substitute for it.
## Review follow-up: policy storage failures
The low-severity review on PR #7417 identified uncaught SQLite busy/unavailable
errors. Three new HTTP regressions first reproduced `500` on the previous code:
real exclusive locks exhaust the busy timeout on policy GET/PUT, and a damaged
database raises a non-operational SQLite error. With the fix all return a
sanitized `503` directing the caller to reload policy before retrying. Lock
failures leave revision 0 intact; after releasing the lock, reads and an explicit
policy update succeed. No automatic retry was introduced.
The combined regression command above now passes **275 tests**, including **35
platform tests and one strict blocking-I/O test**. The narrower platform and
management selection passes **99 tests**. Ruff and the platform wheel check pass.
The original commit `d64411f2` also passed all validation workflows after
maintainer approval; CI on subsequent commits must be assessed independently.
## Review follow-up: deep-tree policy evaluation
The second non-blocking review identified repeated ancestor/default evaluation
for every selected resource, and full resource projection during member space
listing. Three regression cases first failed on `a96d2e7b`. Evaluation now uses
an iterative tree scoped to one snapshot, incremental ancestor processing,
connection-level default composition and shared resource-envelope checks. Listing
uses the same admission checks as details without building processing/resource
responses. No cache survives the call or a policy revision.
The regression selection above now passes **286 tests**, including **46 platform
tests and one strict blocking-I/O test**. New cases exercise a 500-level tree
with 5, 200 and 4,000 resources, bound serialization work instead of asserting
wall-clock timing, and reject processing/resource materialization during listing.
They also cover ancestor revocation/disablement, blocked members, sibling
isolation, operator resource/tool removal and list/detail admission parity.
Independent architecture/security review found no regression; an additional
comparison against the old implementation matched 3,840 effective projections
across 80 generated trees and three user identities.
Same-host single-run measurements used real SQLite, `Store.replace`, member
`Store.spaces` and member `Store.effective` on a 500-space chain. Each connection
has distinct defaults, with overrides midway and at the leaf. Times include
parsing/validation or serialization performed by those methods, but not HTTP
transport. They are local measurements, not production latency guarantees.
| Version / resources | Policy replacement | Space listing | Leaf detail |
| --- | --- | --- | --- |
| `a96d2e7b`, 5 connections / 200 resources | 40.907 s | 39.492 s | 0.146 s |
| Optimized, same 5 / 200 | 0.0146 s | 0.0039 s | 0.0035 s |
| Optimized, maximum 20 / 4,000 | 0.0309 s | 0.0082 s | 0.0092 s |
CI on `a96d2e7b` passed all validation workflows. The subsequent performance
commit has independent CI results; the earlier pass does not cover this change.
## Acceptance boundary
This is an opt-in backend policy layer, with discovery text in the existing
extension directory. It does not deliver the fourth-stage management page or
fifth-stage Trace, and it does not enforce a global chat retrieval ACL. Tool
policy is an effective-policy projection for future consumers. Existing chat
tools and persisted citations retain their current contracts.
No real upstream ingestion, model answer, production SSO, per-user upstream
delegation, multi-host policy replication or cancellation of already admitted
Provider calls was verified here. The second-stage real-service acceptance
record is independent evidence and is not reclassified as third-stage live
acceptance. Use [CONTRACT.md](CONTRACT.md) for the next-stage integration boundary.

View File

@ -0,0 +1,32 @@
"""Optional platform/space policy. No frontend, retrieval or core modifications."""
from importlib import import_module
from deerflow_extension_api import extension
from deerflow_extension_api.plugins import BackendAction, PluginContribution
from .models import Settings
from .router import NAMESPACE, build_router
from .service import Platform
from .store import Store, administrator
async def contract(payload, context):
administrator(context.principal)
return {"version": 1, "authorization": "space-intersection-with-explicit-admin-provider-grants", "storage": "plugin-owned-sqlite", "retrieval_acl": False}
@extension(api="0.2.9", name="knowledge-platform")
def install(registry, config):
if config.get("enabled", False) is False:
return
settings = Settings.model_validate(dict(config))
providers = {}
for connection in settings.connections:
module, attribute = connection.provider_factory.split(":", 1)
providers[connection.id] = getattr(import_module(module), attribute)(connection.provider_config)
store = Store(settings)
platform = Platform(settings, store, providers)
if not registry.plugin(PluginContribution(namespace=NAMESPACE, title="Knowledge platform", enabled=True, backend=(BackendAction("contract", contract),))):
raise RuntimeError("Host does not support platform plugin admission")
registry.routers((build_router(platform),))

View File

@ -0,0 +1,88 @@
"""Extension-owned policy; executable configuration never comes from HTTP."""
import hashlib
import json
from typing import Any, Literal
from deerflow_extension_knowledge_management.contracts import Model, Operation, ProcessingOptions
from pydantic import Field, field_validator, model_validator
OPERATIONS = frozenset({"capabilities", "status", "upload", "url", "manual", "reprocess", "cancel"})
TOOLS = frozenset({"knowledge_search", "knowledge_entries_search", "entity_relations_search"})
Role = Literal["blocked", "viewer", "editor", "manager"]
RANK = {"blocked": 0, "viewer": 1, "editor": 2, "manager": 3}
Tool = Literal["knowledge_search", "knowledge_entries_search", "entity_relations_search"]
def resource_key(provider: str, connection: str, base: str) -> str:
return "kp1_" + hashlib.sha256(json.dumps([provider, connection, base], ensure_ascii=False, separators=(",", ":")).encode("utf-8")).hexdigest()
class Resource(Model):
base_id: str = Field(min_length=1, max_length=256)
dataset_id: str = Field(min_length=1, max_length=256)
operations: frozenset[Operation] = OPERATIONS
tools: frozenset[Tool] = TOOLS
class Connection(Model):
id: str = Field(pattern=r"^[A-Za-z0-9_-]{1,64}$")
provider: str = Field(pattern=r"^[a-z][a-z0-9_-]{0,63}$")
provider_factory: str = "deerflow_extension_knowledge_management.weknora:WeKnoraManagement"
provider_config: dict[str, Any] = Field(default_factory=dict, repr=False)
resources: tuple[Resource, ...] = Field(min_length=1, max_length=200)
@model_validator(mode="after")
def unique(self):
if len({r.base_id for r in self.resources}) != len(self.resources):
raise ValueError("Duplicate native base identity")
return self
class Settings(Model):
enabled: bool = Field(default=False, strict=True)
storage_path: str
connections: tuple[Connection, ...] = Field(min_length=1, max_length=20)
@model_validator(mode="after")
def unique(self):
if len({c.id for c in self.connections}) != len(self.connections):
raise ValueError("Duplicate connection identity")
return self
class Policy(Model):
enabled: bool = Field(default=True, strict=True)
operations: frozenset[Operation] | None = None
tools: frozenset[Tool] | None = None
default_tool: Tool | None = None
processing: ProcessingOptions = Field(default_factory=ProcessingOptions)
class ConnectionPolicy(Policy):
title: str = Field(default="", max_length=200)
class Space(Policy):
title: str = Field(min_length=1, max_length=200)
parent_id: str | None = Field(default=None, pattern=r"^[A-Za-z0-9_-]{1,64}$")
resources: frozenset[str] | None = None
members: dict[str, Role] = Field(default_factory=dict, max_length=200)
@field_validator("members")
@classmethod
def exact_users(cls, members):
if any(not user or user != user.strip() or len(user) > 256 for user in members):
raise ValueError("Use exact bounded host user identities")
return members
class PlatformPolicy(Model):
defaults: Policy = Field(default_factory=Policy)
connections: dict[str, ConnectionPolicy] = Field(default_factory=dict, max_length=20)
spaces: dict[str, Space] = Field(default_factory=dict, max_length=500)
class Update(Model):
revision: int = Field(ge=0, strict=True)
policy: PlatformPolicy

View File

@ -0,0 +1,133 @@
"""Host-authenticated platform routes; no caller-provided identities or credentials."""
import logging
import sqlite3
from typing import Literal
from deerflow_extension_api.auth import arequire_plugin_management
from deerflow_extension_knowledge_management.contracts import IngestRequest, ManagementCredentialUnavailable, ProcessingOptions
from fastapi import APIRouter, Depends, HTTPException, Request
from pydantic import ValidationError
from .models import Update
from .store import RevisionConflict, administrator
NAMESPACE = "knowledge-platform"
logger = logging.getLogger(__name__)
async def principal(request: Request):
try:
result = await arequire_plugin_management(request, NAMESPACE, scope="read" if request.method == "GET" else "write")
if result.is_internal or not result.user_id:
raise PermissionError
return result
except PermissionError:
raise HTTPException(403, "Knowledge platform access denied") from None
async def manager(request: Request, caller=Depends(principal)):
try:
administrator(caller)
await arequire_plugin_management(request, "knowledge-management", scope="read" if request.method == "GET" else "write")
return caller
except PermissionError:
raise HTTPException(403, "Knowledge management access denied") from None
async def invoke(awaitable):
try:
return await awaitable
except PermissionError:
raise HTTPException(403, "Space or provider access denied") from None
except RevisionConflict:
raise HTTPException(409, "Policy changed; reload and reconcile before retrying") from None
except FileNotFoundError:
raise HTTPException(404, "Provider resource unavailable") from None
except FileExistsError:
raise HTTPException(409, "Resource conflict; reconcile before retrying") from None
except NotImplementedError:
raise HTTPException(422, "Unsupported capability or processing option") from None
except ValueError:
raise HTTPException(422, "Invalid platform policy or request") from None
except ManagementCredentialUnavailable:
raise HTTPException(503, "Management credential is not configured") from None
except sqlite3.Error as exc:
logger.warning("Knowledge platform storage failed: %s", type(exc).__name__)
raise HTTPException(503, "Policy storage is busy or unavailable; reload policy before retrying") from None
except RuntimeError as exc:
logger.warning("Knowledge platform provider failed: %s", type(exc).__name__)
raise HTTPException(502, "Provider request failed; reconcile before retrying a write") from None
async def bounded(request, limit):
data = bytearray()
async for chunk in request.stream():
if len(data) + len(chunk) > limit:
raise HTTPException(413, "Platform request exceeds size limit")
data.extend(chunk)
return bytes(data)
async def body(request, model):
data = await bounded(request, 1024 * 1024)
try:
return model.model_validate_json(data)
except ValidationError:
raise HTTPException(422, "Invalid platform request") from None
def build_router(platform):
router = APIRouter(prefix="/api/extensions/knowledge-platform", tags=[NAMESPACE])
@router.get("/policy")
async def policy(caller=Depends(principal)):
return await invoke(platform.policy(caller))
@router.put("/policy")
async def replace(request: Request, caller=Depends(principal)):
if not caller.is_admin:
raise HTTPException(403, "Policy changes require an administrator")
update = await body(request, Update)
return await invoke(platform.replace(caller, update.policy, update.revision))
@router.get("/spaces")
async def spaces(caller=Depends(principal)):
return await invoke(platform.spaces(caller))
@router.get("/spaces/{space}")
async def effective(space: str, caller=Depends(principal)):
return await invoke(platform.effective(caller, space))
prefix = "/spaces/{space}/resources/{resource}"
@router.get(prefix + "/capabilities")
async def capabilities(space: str, resource: str, caller=Depends(manager)):
return await invoke(platform.invoke(caller, space, resource, "capabilities"))
@router.post(prefix + "/documents", status_code=202)
async def ingest(request: Request, space: str, resource: str, caller=Depends(manager)):
data = await body(request, IngestRequest)
return await invoke(platform.invoke(caller, space, resource, data.kind, data))
@router.post(prefix + "/files", status_code=202)
async def upload(request: Request, space: str, resource: str, filename: str, caller=Depends(manager)):
if request.headers.get("content-type", "").split(";", 1)[0].strip().lower() != "application/octet-stream":
raise HTTPException(415, "Use application/octet-stream")
try:
options = ProcessingOptions.model_validate_json(request.headers.get("x-processing-options", "{}"))
except ValidationError:
raise HTTPException(422, "Invalid processing options") from None
data = await bounded(request, 16 * 1024 * 1024)
return await invoke(platform.invoke(caller, space, resource, "upload", options, filename=filename, content=data))
@router.get(prefix + "/documents/{document}")
async def status(space: str, resource: str, document: str, caller=Depends(manager)):
return await invoke(platform.invoke(caller, space, resource, "status", document))
@router.post(prefix + "/documents/{document}/{action}", status_code=202)
async def process(request: Request, space: str, resource: str, document: str, action: Literal["reprocess", "cancel"], caller=Depends(manager)):
options = await body(request, ProcessingOptions)
return await invoke(platform.invoke(caller, space, resource, action, options, document=document))
return router

View File

@ -0,0 +1,76 @@
"""Space policy is an additional restriction on the existing management provider."""
import asyncio
from deerflow_extension_knowledge_management.contracts import ManagementContext, ProcessingOptions, ProcessingResult
from .store import RevisionConflict, administrator
class Platform:
def __init__(self, settings, store, providers):
self.settings, self.store, self.providers = settings, store, providers
async def policy(self, principal):
return await asyncio.to_thread(self.store.read, principal)
async def replace(self, principal, policy, revision):
return await asyncio.to_thread(self.store.replace, principal, policy, revision)
async def spaces(self, principal):
return await asyncio.to_thread(self.store.spaces, principal)
async def effective(self, principal, space_id):
return await asyncio.to_thread(self.store.effective, principal, space_id)
async def _admit(self, principal, space_id, resource, operation):
# Keep phase 2's explicit administrator floor; never impersonate a member.
administrator(principal)
effective = await self.effective(principal, space_id)
for item in effective["resources"]:
if item["resource_id"] == resource and operation in item["operations"]:
return effective["revision"], item
raise PermissionError("Space resource operation denied")
@staticmethod
def _result(result, item, document=None):
try:
result = ProcessingResult.model_validate(result)
except ValueError:
raise RuntimeError("Invalid management result") from None
if (result.provider, result.connection_id, result.dataset_id) != (item["provider"], item["connection_id"], item["dataset_id"]) or (document is not None and result.document_id != document):
raise RuntimeError("Management result identity does not match space resource")
return result
async def invoke(self, principal, space_id, resource, operation, payload=None, *, document=None, filename=None, content=None):
revision, item = await self._admit(principal, space_id, resource, operation)
provider = self.providers[item["connection_id"]]
context = ManagementContext(principal=principal)
connection, base = item["connection_id"], item["native_base_id"]
if operation == "status":
return self._result(await provider.status(context, connection, base, payload), item, payload)
capabilities = await provider.capabilities(context, connection, base)
if operation == "capabilities":
return capabilities.model_copy(update={"operations": capabilities.operations & set(item["operations"])})
if operation not in capabilities.operations:
raise NotImplementedError("Provider does not support this operation")
if operation in {"manual", "url", "upload"}:
# Defaults are for NEW documents; an empty reprocess preserves upstream policy.
supplied = payload.processing if operation != "upload" else payload
options = ProcessingOptions.model_validate({**item["processing"], **supplied.model_dump(exclude_none=True)})
else:
options = payload
if not options.model_dump(exclude_none=True).keys() <= capabilities.processing_options:
raise NotImplementedError("Provider does not support selected processing options")
# Capabilities may involve upstream I/O. Recheck all policy before submitting a write.
latest, _ = await self._admit(principal, space_id, resource, operation)
if latest != revision:
raise RevisionConflict("Policy changed during admission; reconcile before retrying")
if operation in {"manual", "url"}:
result = await provider.ingest(context, connection, base, payload.model_copy(update={"processing": options}))
return self._result(result, item)
if operation == "upload":
return self._result(await provider.upload(context, connection, base, filename, content, options), item)
if operation in {"reprocess", "cancel"}:
return self._result(await provider.process(context, connection, base, document, operation, options), item, document)
raise ValueError("Unknown management operation")

View File

@ -0,0 +1,265 @@
"""Transactional bounded policy snapshots, independent of host persistence."""
import json
import re
import sqlite3
from contextlib import closing
from dataclasses import dataclass
from pathlib import Path
from deerflow_extension_knowledge_management.contracts import ProcessingOptions
from .models import OPERATIONS, RANK, TOOLS, ConnectionPolicy, PlatformPolicy, Settings, Space, resource_key
class RevisionConflict(Exception):
pass
def administrator(principal):
if not principal.user_id or not principal.is_admin or principal.is_internal:
raise PermissionError("Platform policy requires a human administrator")
def inherited_defaults(policies):
values, default = {}, None
for policy in policies:
values.update(policy.processing.model_dump(exclude_none=True))
if "default_tool" in policy.model_fields_set:
default = policy.default_tool
return values, default
@dataclass(frozen=True)
class _SpaceState:
space_id: str
space: Space
parent: "_SpaceState | None"
resources: frozenset[str]
operations: frozenset[str]
tools: frozenset[str]
members: dict[str, str]
enabled: bool
default_tool: str | None
default_overridden: bool
class _PolicyTree:
"""One snapshot only: reuse ancestor state, never cache grants across requests."""
def __init__(self, policy, resources):
self.policy, self.resources = policy, resources
self.states, self.processing, self.groups, self.connection_defaults = {}, {}, {}, {}
self.connections = {}
for connection, _ in resources.values():
if connection.id not in self.connections:
self.connections[connection.id] = policy.connections.get(connection.id, ConnectionPolicy())
def resolve(self, space_id):
pending, visited = [], set()
cursor = space_id
while cursor is not None and cursor not in self.states:
if cursor in visited or cursor not in self.policy.spaces:
raise ValueError("Cyclic or missing space parent")
if not re.fullmatch(r"[A-Za-z0-9_-]{1,64}", cursor):
raise ValueError("Invalid space identity")
visited.add(cursor)
pending.append(cursor)
cursor = self.policy.spaces[cursor].parent_id
for key in reversed(pending):
space = self.policy.spaces[key]
parent = self.states.get(space.parent_id)
if parent is None and space.resources is None:
raise ValueError("Root requires an explicit resource envelope")
values = {}
for field, initial in (("resources", frozenset()), ("operations", OPERATIONS), ("tools", TOOLS)):
ceiling = getattr(parent, field) if parent else initial
value = getattr(space, field)
if parent is not None and value is not None and not value <= ceiling:
raise ValueError(f"Child {field} exceed parent envelope")
values[field] = ceiling if value is None else value
members = dict(parent.members) if parent else {}
if parent and any(RANK[role] > RANK[members.get(user, "blocked")] for user, role in space.members.items()):
raise ValueError("Child membership exceeds ancestor grant")
for user, role in space.members.items():
if role == "blocked":
members.pop(user, None)
else:
members[user] = role
overridden = "default_tool" in space.model_fields_set
self.states[key] = _SpaceState(
space_id=key,
space=space,
parent=parent,
**values,
members=members,
enabled=space.enabled and (parent.enabled if parent else True),
default_tool=space.default_tool if overridden else (parent.default_tool if parent else None),
default_overridden=overridden or (parent.default_overridden if parent else False),
)
return self.states[space_id]
def resource_groups(self, selected):
if selected not in self.groups:
groups = {}
for key in selected:
if key not in self.resources:
raise ValueError("Unknown operator resource")
connection, base = self.resources[key]
previous = groups.get(connection.id, (connection, TOOLS))[1]
groups[connection.id] = (connection, previous & base.tools)
self.groups[selected] = groups
return self.groups[selected]
def default_tool(self, state, connection):
if state.default_overridden:
return state.default_tool
cp = self.connections[connection]
return cp.default_tool if "default_tool" in cp.model_fields_set else self.policy.defaults.default_tool
def tools(self, state, connection):
cp = self.connections[connection]
return state.tools & (self.policy.defaults.tools if self.policy.defaults.tools is not None else TOOLS) & (cp.tools if cp.tools is not None else TOOLS)
def defaults(self, state, connection):
if connection not in self.connection_defaults:
self.connection_defaults[connection] = inherited_defaults([self.policy.defaults, self.connections[connection]])[0]
pending, cursor = [], state
while cursor is not None and cursor.space_id not in self.processing:
pending.append(cursor)
cursor = cursor.parent
values = self.processing[cursor.space_id] if cursor else {}
for item in reversed(pending):
values = {**values, **item.space.processing.model_dump(exclude_none=True)}
self.processing[item.space_id] = values
return {**self.connection_defaults[connection], **self.processing[state.space_id]}
class Store:
def __init__(self, settings: Settings):
self.settings = settings
self.path = Path(settings.storage_path)
if not self.path.is_absolute():
raise ValueError("storage_path must be an absolute deployment-owned path")
self.resources = {resource_key(c.provider, c.id, r.base_id): (c, r) for c in settings.connections for r in c.resources}
self.path.parent.mkdir(parents=True, exist_ok=True)
with closing(self.connect()) as db, db:
db.execute("CREATE TABLE IF NOT EXISTS platform_policy (id INTEGER PRIMARY KEY CHECK(id=1), revision INTEGER NOT NULL, data TEXT NOT NULL)")
db.execute("INSERT OR IGNORE INTO platform_policy VALUES (1, 0, ?)", (PlatformPolicy().model_dump_json(),))
def connect(self):
return sqlite3.connect(self.path, timeout=10)
def _snapshot(self, db):
revision, data = db.execute("SELECT revision, data FROM platform_policy WHERE id=1").fetchone()
policy = PlatformPolicy.model_validate_json(data)
return revision, policy
def validate(self, policy):
if not policy.connections.keys() <= {c.id for c in self.settings.connections}:
raise ValueError("Unknown operator connection")
tree = _PolicyTree(policy, self.resources)
for space_id in policy.spaces:
state = tree.resolve(space_id)
for connection, base_tools in tree.resource_groups(state.resources).values():
ProcessingOptions.model_validate(tree.defaults(state, connection.id))
default = tree.default_tool(state, connection.id)
if default is not None and default not in tree.tools(state, connection.id) & base_tools:
raise ValueError("Default tool is outside the effective tool envelope")
def read(self, principal):
administrator(principal)
with closing(self.connect()) as db:
revision, policy = self._snapshot(db)
# Operator configuration, endpoints and credential references never leave Store.
return {
"revision": revision,
"policy": policy.model_dump(mode="json", exclude_unset=True),
"unavailable_resources": sorted({resource for space in policy.spaces.values() for resource in (space.resources or ()) if resource not in self.resources}),
"resources": [{"resource_id": key, "provider": c.provider, "connection_id": c.id, "native_base_id": r.base_id, "dataset_id": r.dataset_id} for key, (c, r) in self.resources.items()],
}
def replace(self, principal, data, expected_revision):
administrator(principal)
policy = PlatformPolicy.model_validate(data)
self.validate(policy)
# Preserve absent-vs-null default_tool semantics across persistence/round trips.
serialized = policy.model_dump_json(exclude_unset=True)
if len(serialized.encode("utf-8")) > 1024 * 1024:
raise ValueError("Policy exceeds 1 MiB")
with closing(self.connect()) as db, db:
db.execute("BEGIN IMMEDIATE")
revision = db.execute("SELECT revision FROM platform_policy WHERE id=1").fetchone()[0]
if revision != expected_revision:
raise RevisionConflict("Policy changed; reload before retrying")
db.execute("UPDATE platform_policy SET revision=?, data=? WHERE id=1", (revision + 1, serialized))
return {"revision": revision + 1, "policy": json.loads(serialized)}
def _access(self, principal, space_id, tree):
if not principal.user_id or principal.is_internal or not tree.policy.defaults.enabled:
raise PermissionError("Space access denied")
if space_id not in tree.policy.spaces:
raise PermissionError("Space access denied")
state = tree.resolve(space_id)
role = state.members.get(principal.user_id, "blocked")
if not state.enabled or not RANK[role]:
raise PermissionError("Space access denied")
try:
groups = tree.resource_groups(state.resources)
except ValueError:
raise PermissionError("Space contains an unavailable operator resource") from None
# Keep listing and detail admission identical after operator tool revocation.
# Intersect base tools per connection without constructing resource responses.
for connection, base_tools in groups.values():
default = tree.default_tool(state, connection.id)
if tree.connections[connection.id].enabled and default is not None and default not in tree.tools(state, connection.id) & base_tools:
raise PermissionError("Space default tool is no longer available")
return state, role
def _effective(self, principal, space_id, revision, policy):
tree = _PolicyTree(policy, self.resources)
state, role = self._access(principal, space_id, tree)
operations = state.operations & (policy.defaults.operations if policy.defaults.operations is not None else OPERATIONS)
if role == "viewer":
operations &= {"capabilities", "status"}
items, defaults = [], {}
for key in sorted(state.resources):
conn, base = self.resources[key]
cp = tree.connections[conn.id]
if not cp.enabled:
continue
if conn.id not in defaults:
defaults[conn.id] = tree.defaults(state, conn.id)
resource_tools = tree.tools(state, conn.id) & base.tools
items.append(
{
"resource_id": key,
"provider": conn.provider,
"connection_id": conn.id,
"native_base_id": base.base_id,
"dataset_id": base.dataset_id,
"operations": sorted(operations & base.operations & (cp.operations if cp.operations is not None else OPERATIONS)),
"tools": sorted(resource_tools),
"default_tool": tree.default_tool(state, conn.id),
"processing": defaults[conn.id],
}
)
return {"version": 1, "revision": revision, "space_id": space_id, "title": state.space.title, "role": role, "resources": items}
def effective(self, principal, space_id):
with closing(self.connect()) as db:
revision, policy = self._snapshot(db)
return self._effective(principal, space_id, revision, policy)
def spaces(self, principal):
with closing(self.connect()) as db:
revision, policy = self._snapshot(db)
tree = _PolicyTree(policy, self.resources)
items = []
for space_id in policy.spaces:
try:
state, role = self._access(principal, space_id, tree)
except PermissionError:
continue
items.append({"space_id": space_id, "title": state.space.title, "role": role})
return {"revision": revision, "items": items}

View File

@ -0,0 +1,25 @@
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "deerflow-extension-knowledge-platform"
version = "0.1.0"
description = "Optional platform connection and hierarchical knowledge-space policy"
readme = "README.md"
license = "MIT"
requires-python = ">=3.12"
dependencies = ["deerflow-extension-api>=0.2.9,<0.3", "deerflow-extension-knowledge-management>=0.1.0,<0.2", "fastapi>=0.115,<1", "pydantic>=2,<3"]
[project.entry-points."deerflow.extensions"]
knowledge-platform = "deerflow_extension_knowledge_platform:install"
[tool.hatch.build.targets.wheel]
packages = ["deerflow_extension_knowledge_platform"]
[tool.ruff]
line-length = 240
target-version = "py312"
[tool.ruff.lint]
select = ["E", "F", "I", "UP"]

View File

@ -55,6 +55,11 @@ export const extensionCatalog = [
package: "deerflow-extension-knowledge-management",
namespace: "knowledge-management",
},
{
key: "knowledgePlatform",
package: "deerflow-extension-knowledge-platform",
namespace: "knowledge-platform",
},
{ key: "lightrag", package: "deerflow-extension-lightrag", namespace: null },
{ key: "weknora", package: "deerflow-extension-weknora", namespace: null },
{ key: "example", package: "deerflow-extension-example", namespace: null },

View File

@ -90,6 +90,11 @@ export const enUS: Translations = {
description:
"Retrieve documents and entity relations from LightRAG with shared source citations.",
},
knowledgePlatform: {
title: "Knowledge spaces and connections",
description:
"Manage hierarchical spaces, resource sharing and inherited policies with explicit administrator grants. Requires installation and configuration; no management page or global retrieval ACL is included.",
},
knowledgeManagement: {
title: "Knowledge ingestion and processing",
description:

View File

@ -37,6 +37,7 @@ export interface Translations {
| "retrieval"
| "knowledge"
| "knowledgeManagement"
| "knowledgePlatform"
| "weknora"
| "lightrag"
| "example",

View File

@ -84,6 +84,11 @@ export const zhCN: Translations = {
title: "LightRAG",
description: "从 LightRAG 检索文档与实体关系,并使用统一的来源引用。",
},
knowledgePlatform: {
title: "知识空间与连接",
description:
"在管理员明确授权后管理多层级空间、资源共享与继承策略。需要安装和配置,不包含管理页面,也不改变全局检索权限。",
},
knowledgeManagement: {
title: "知识入库与处理",
description:

View File

@ -195,3 +195,21 @@ test("knowledge management discovery merges its namespace without implying activ
expect(matches[0]?.loaded).toBe(management);
}
});
test("knowledge platform discovery is localized and does not imply a management page", () => {
for (const [copy, title] of [
[enUS.extensions.catalog, "Knowledge spaces and connections"],
[zhCN.extensions.catalog, "知识空间与连接"],
] as const) {
const entry = extensionDirectory([], copy).find(
(item) => item.package === "deerflow-extension-knowledge-platform",
);
expect(entry?.id).toBe("knowledge-platform");
expect(entry?.title).toBe(title);
expect(entry?.description).toBe(copy.knowledgePlatform.description);
expect(entry?.guide).toBe(
"https://github.com/bytedance/deer-flow/tree/main/examples/deerflow-extension-knowledge-platform#readme",
);
expect(entry?.loaded).toBeUndefined();
}
});