diff --git a/AGENTS.md b/AGENTS.md
index 4808701c68..28da683125 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -91,6 +91,13 @@ 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.
+
 Root runtime configs: copy `config.example.yaml` → `config.yaml` and
 `extensions_config.example.json` → `extensions_config.json` (MCP/skills).
 Both are gitignored and Gateway-editable; schema/resolution:
diff --git a/README.md b/README.md
index b169bc2310..17337e8d27 100644
--- a/README.md
+++ b/README.md
@@ -1598,6 +1598,15 @@ retrieval citations. Configure explicit user-to-workspace credential bindings;
 resource authorization remains with WeKnora. The extension directory includes
 its installation guide.
 
+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
+Wiki/graph generation remain upstream. It is disabled by default, requires
+explicit resource/operation grants, and installs no management page. Its
+provider-independent API is covered by offline contract tests. Real ingestion,
+GLM-5.3/BGE integration, browser citation replay and observed limitations are
+recorded in the [acceptance report](examples/deerflow-extension-knowledge-management/VALIDATION.md).
+
 Skills are what make DeerFlow do *almost anything*.
 
 A standard Agent Skill is a structured capability module — a Markdown file that defines a workflow, best practices, and references to supporting resources. DeerFlow ships with built-in skills for research, report generation, slide creation, web pages, image and video generation, and more. But the real power is extensibility: add your own skills, replace the built-in ones, or combine them into compound workflows.
diff --git a/README_zh.md b/README_zh.md
index 0adf1e253a..de100b5493 100644
--- a/README_zh.md
+++ b/README_zh.md
@@ -744,6 +744,12 @@ DeerFlow 可连接租户级 RAGFlow，并通过 `knowledge_search` 按 embedding
 需显式配置 DeerFlow 用户到 WeKnora 工作空间的凭据绑定；资源授权仍由 WeKnora 执行。
 安装与配置步骤见扩展目录中的英文指南。
 
+可选的[知识入库与处理扩展](examples/deerflow-extension-knowledge-management/README.md)
+通过 WeKnora 提供管理员授权的文件、网址和手工文本入库、处理状态查询、重新处理与取消。
+解析、索引及可选的 Wiki/图谱生成由上游负责。扩展默认关闭，需明确配置资源与操作授权，
+不包含管理页面。Provider 无关的 API 由离线契约测试覆盖；真实入库、GLM-5.3/BGE
+联调、浏览器引用回放及已知限制见[验收报告](examples/deerflow-extension-knowledge-management/VALIDATION.md)。
+
 Gateway 生成后续建议时，现在会先把普通字符串输出和 block/list 风格的富文本内容统一归一化，再去解析 JSON 数组响应，因此不同 provider 的内容包装方式不会再悄悄把建议吞掉。
 
 Web UI 支持从已完成的 assistant 回复分叉出一个新的主对话。自动继承的分叉标题会使用下一个空闲的数字后缀（`标题 (2)`、`标题 (3)`……）；显式指定或手动重命名得到的同名后缀也会占号，即使它没有生成序号 metadata，后续自动分叉也不会与它重名。API 调用方显式提供的标题保持不变；重命名会清除旧的生成序号，因此从新标题继续自动分叉时会重新从 `(2)` 开始。最近对话列表还会把已加载的分叉直接排列在已加载的父对话下方，并显示低干扰的树形连接线。父对话尚未加载、谱系数据错误或成环、父子置顶状态不一致时，分叉会安全地保留在顶层，不会被隐藏或跨越置顶边界移动。新 thread 会保留该轮回复的 checkpoint 以及用户消息之前的重放 checkpoint，因此分叉后可以立即重新生成该回复。对于缺少 checkpoint 父链接的旧历史或导入历史，Gateway 会进行有界的时间顺序查找；如果不存在更早的重放 checkpoint，分叉仍会按旧版单-checkpoint 形态成功创建，但无法重新生成继承的回复。已有的单-checkpoint 分叉会保持不变，不会通过不安全的 checkpoint 复制尝试修复。只有从最新回合分叉时才会尽力复制当前 thread 的工作区文件；从历史回合分叉不会带入后续时间线创建的文件。
diff --git a/backend/tests/blocking_io/test_knowledge_management_extension.py b/backend/tests/blocking_io/test_knowledge_management_extension.py
new file mode 100644
index 0000000000..93a47d1fbc
--- /dev/null
+++ b/backend/tests/blocking_io/test_knowledge_management_extension.py
@@ -0,0 +1,39 @@
+"""Management performs SSL trust loading outside the Gateway event loop."""
+
+import asyncio
+import ssl
+from pathlib import Path
+
+import httpx
+import pytest
+from deerflow_extension_api.auth import ExtensionPrincipal
+
+
+@pytest.mark.asyncio
+async def test_management_http_client_setup_is_off_loop(monkeypatch):
+    monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[3] / "examples/deerflow-extension-knowledge-management"))
+    from deerflow_extension_knowledge_management.contracts import IngestRequest, ManagementContext
+    from deerflow_extension_knowledge_management.weknora import WeKnoraManagement
+
+    monkeypatch.setenv("MANAGEMENT_BLOCKING_KEY", "synthetic")
+    provider = WeKnoraManagement(
+        {"connections": [{"id": "test", "base_url": "https://weknora.invalid", "tenant_id": "1", "api_key_env": "MANAGEMENT_BLOCKING_KEY", "grants": [{"user_id": "alice", "knowledge_base_ids": ["kb"], "operations": ["manual"]}]}]}
+    )
+
+    async def send(self, request, **kwargs):
+        data = {"id": "kb", "capabilities": {}} if request.method == "GET" else {"id": "doc", "knowledge_base_id": "kb", "parse_status": "pending", "enable_status": "enabled"}
+        return httpx.Response(200, request=request, json={"success": True, "data": data})
+
+    original = ssl.create_default_context
+
+    def ssl_context(*args, **kwargs):
+        try:
+            asyncio.get_running_loop()
+        except RuntimeError:
+            return original(*args, **kwargs)
+        raise AssertionError("SSL trust loading must run off the event loop")
+
+    monkeypatch.setattr(ssl, "create_default_context", ssl_context)
+    monkeypatch.setattr(httpx.AsyncClient, "send", send)
+    result = await provider.ingest(ManagementContext(ExtensionPrincipal("alice", is_admin=True)), "test", "kb", IngestRequest(kind="manual", title="Guide", content="Evidence"))
+    assert result.state == "queued"
diff --git a/backend/tests/test_agent_guidance_check.py b/backend/tests/test_agent_guidance_check.py
index d645856868..110693d9ca 100644
--- a/backend/tests/test_agent_guidance_check.py
+++ b/backend/tests/test_agent_guidance_check.py
@@ -60,6 +60,7 @@ EXPECTED_GUIDANCE_PATHS = {
     "examples/deerflow-extension-agent-teams/AGENTS.md",
     "examples/deerflow-extension-workflows/AGENTS.md",
     "examples/deerflow-extension-weknora/AGENTS.md",
+    "examples/deerflow-extension-knowledge-management/AGENTS.md",
     "examples/deerflow-extension-lightrag/AGENTS.md",
 }
 
diff --git a/backend/tests/test_knowledge_management_extension.py b/backend/tests/test_knowledge_management_extension.py
new file mode 100644
index 0000000000..7d6c5d61ec
--- /dev/null
+++ b/backend/tests/test_knowledge_management_extension.py
@@ -0,0 +1,508 @@
+"""Offline management contract and upstream authorization regressions."""
+
+import json
+from pathlib import Path
+
+import httpx
+import pytest
+from deerflow_extension_api.auth import ExtensionPrincipal
+
+
+@pytest.fixture
+def management(monkeypatch):
+    root = Path(__file__).resolve().parents[2]
+    monkeypatch.syspath_prepend(str(root / "examples/deerflow-extension-knowledge-management"))
+    from deerflow_extension_knowledge_management.weknora import WeKnoraManagement
+
+    monkeypatch.setenv("MANAGEMENT_TEST_KEY", "synthetic-secret")
+    calls = []
+    state = {"status": "pending", "base": "kb-1", "code": 200}
+
+    def handler(request):
+        calls.append(request)
+        assert request.headers["X-API-Key"] == "synthetic-secret"
+        if state["code"] != 200:
+            return httpx.Response(state["code"], text="synthetic-secret private error")
+        if request.url.path == "/api/v1/knowledge-bases/kb-1":
+            data = {"id": "kb-1", "capabilities": {"wiki": True, "graph": True, "vector": True}, "chunking_config": state.get("chunking_config", {"chunk_size": 1024, "chunk_overlap": 128})}
+        else:
+            data = {"id": "doc-1", "knowledge_base_id": state["base"], "parse_status": state["status"], "enable_status": "enabled"}
+        return httpx.Response(201 if request.url.path.endswith(("/file", "/url")) else 200, json={"success": True, "data": data})
+
+    config = {
+        "connections": [
+            {
+                "id": "local",
+                "base_url": "https://weknora.example",
+                "tenant_id": "1",
+                "api_key_env": "MANAGEMENT_TEST_KEY",
+                "grants": [{"user_id": "alice", "knowledge_base_ids": ["kb-1"], "operations": ["capabilities", "status", "upload", "url", "manual", "reprocess", "cancel"]}],
+            }
+        ]
+    }
+    return WeKnoraManagement(config, transport=httpx.MockTransport(handler)), calls, state
+
+
+@pytest.mark.asyncio
+async def test_manual_ingestion_status_and_retrieval_identity(management):
+    from deerflow_extension_knowledge_management.contracts import IngestRequest, ManagementContext
+
+    provider, calls, state = management
+    context = ManagementContext(principal=ExtensionPrincipal("alice", is_admin=True))
+    result = await provider.ingest(context, "local", "kb-1", IngestRequest(kind="manual", title="Guide", content="Cedar evidence"))
+    assert result.document_id == "doc-1" and result.state == "queued"
+    assert result.dataset_id.startswith("wk1_") and not result.searchable
+    assert json.loads(calls[-1].content)["status"] == "publish"
+    state["status"] = "completed"
+    result = await provider.status(context, "local", "kb-1", "doc-1")
+    assert result.state == "completed" and result.searchable
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "principal,base", [(ExtensionPrincipal("alice"), "kb-1"), (ExtensionPrincipal("bob", is_admin=True), "kb-1"), (ExtensionPrincipal("alice", is_admin=True), "other"), (ExtensionPrincipal("alice", is_admin=True, is_internal=True), "kb-1")]
+)
+async def test_denials_make_no_upstream_requests(management, principal, base):
+    from deerflow_extension_knowledge_management.contracts import ManagementContext
+
+    provider, calls, _ = management
+    with pytest.raises(PermissionError):
+        await provider.status(ManagementContext(principal=principal), "local", base, "doc-1")
+    assert not calls
+
+
+@pytest.mark.asyncio
+async def test_cross_base_document_cannot_be_reprocessed(management):
+    from deerflow_extension_knowledge_management.contracts import ManagementContext, ProcessingOptions
+
+    provider, calls, state = management
+    state["base"] = "other"
+    with pytest.raises(PermissionError):
+        await provider.process(ManagementContext(principal=ExtensionPrincipal("alice", is_admin=True)), "local", "kb-1", "doc-1", "reprocess", ProcessingOptions())
+    assert all(call.method == "GET" for call in calls)
+
+
+@pytest.mark.asyncio
+async def test_upstream_auth_failure_never_retries_or_leaks(management):
+    from deerflow_extension_knowledge_management.contracts import ManagementContext
+
+    provider, calls, state = management
+    state["code"] = 403
+    with pytest.raises(PermissionError, match="denied") as error:
+        await provider.status(ManagementContext(principal=ExtensionPrincipal("alice", is_admin=True)), "local", "kb-1", "doc-1")
+    assert len(calls) == 1 and "synthetic-secret" not in str(error.value)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("kind", ["file", "url"])
+async def test_ingestion_accepts_created_and_preserves_options(management, kind):
+    from deerflow_extension_knowledge_management.contracts import IngestRequest, ManagementContext, ProcessingOptions
+
+    provider, calls, _ = management
+    context = ManagementContext(ExtensionPrincipal("alice", is_admin=True))
+    options = ProcessingOptions(chunk_size=512, chunk_overlap=64, graph_enabled=True)
+    if kind == "file":
+        result = await provider.upload(context, "local", "kb-1", "guide.txt", b"Cedar evidence", options)
+        assert b'"chunk_size": 512' in calls[-1].content
+        assert b"Cedar evidence" in calls[-1].content
+    else:
+        result = await provider.ingest(context, "local", "kb-1", IngestRequest(kind="url", url="https://example.com/guide", processing=options))
+        assert json.loads(calls[-1].content)["process_config"]["chunking_config"] == {"chunk_size": 512, "chunk_overlap": 64}
+    assert result.state == "queued"
+    assert calls[-1].url.path.endswith("/" + kind)
+
+
+@pytest.mark.asyncio
+async def test_reprocess_and_cancel_keep_original_processing_policy(management):
+    from deerflow_extension_knowledge_management.contracts import ManagementContext, ProcessingOptions
+
+    provider, calls, _ = management
+    context = ManagementContext(ExtensionPrincipal("alice", is_admin=True))
+    for operation, path in [("reprocess", "reparse"), ("cancel", "cancel-parse")]:
+        await provider.process(context, "local", "kb-1", "doc-1", operation, ProcessingOptions())
+        assert calls[-1].url.path.endswith("/" + path)
+        assert calls[-1].content == b""
+
+
+@pytest.mark.asyncio
+async def test_capabilities_are_provider_and_grant_specific(management):
+    from deerflow_extension_knowledge_management.contracts import ManagementContext
+
+    provider, _, _ = management
+    result = await provider.capabilities(ManagementContext(ExtensionPrincipal("alice", is_admin=True)), "local", "kb-1")
+    assert result.pipeline["entries"] == "provider_managed"
+    assert result.pipeline["entity_relations"] == "provider_managed"
+    assert "upload" in result.operations
+    assert "delete" not in result.operations
+
+
+@pytest.mark.asyncio
+async def test_unknown_state_is_not_completed(management):
+    from deerflow_extension_knowledge_management.contracts import ManagementContext
+
+    provider, _, state = management
+    state["status"] = "new_upstream_state"
+    result = await provider.status(ManagementContext(ExtensionPrincipal("alice", is_admin=True)), "local", "kb-1", "doc-1")
+    assert result.state == "unknown" and not result.searchable
+
+
+@pytest.mark.parametrize("payload", [{"chunk_size": 200}, {"chunk_overlap": 50}, {"chunk_size": 100, "chunk_overlap": 100}, {"graph_enabled": "true"}, {"parser_api_key": "secret"}, {"wiki_enabled": True}])
+def test_processing_rejects_unsupported_or_invalid_options(management, payload):
+    from deerflow_extension_knowledge_management.contracts import ProcessingOptions
+
+    with pytest.raises(ValueError):
+        ProcessingOptions.model_validate(payload)
+
+
+@pytest.mark.asyncio
+async def test_unsupported_graph_rejected_before_mutation(management):
+    from deerflow_extension_knowledge_management.contracts import ProcessingOptions
+
+    provider, calls, _ = management
+    with pytest.raises(NotImplementedError):
+        provider._options(ProcessingOptions(graph_enabled=True), {"capabilities": {"graph": False}})
+    assert not calls
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("filename,content", [("../guide.txt", b"text"), ("guide\n.txt", b"text"), ("guide.txt", b""), pytest.param("guide.txt", b"x" * (16 * 1024 * 1024 + 1), id="oversized")])
+async def test_invalid_upload_never_reaches_provider(management, filename, content):
+    from deerflow_extension_knowledge_management.contracts import ManagementContext, ProcessingOptions
+
+    provider, calls, _ = management
+    with pytest.raises(ValueError):
+        await provider.upload(ManagementContext(ExtensionPrincipal("alice", is_admin=True)), "local", "kb-1", filename, content, ProcessingOptions())
+    assert not calls
+
+
+@pytest.fixture
+def api(management):
+    from deerflow_extension_api.auth import EXTENSION_PLUGIN_AUTHZ_RESOLVER_ASYNC_KEY, EXTENSION_PRINCIPAL_RESOLVER_KEY
+    from deerflow_extension_knowledge_management.router import build_router
+    from fastapi import FastAPI
+
+    provider, calls, state = management
+    app = FastAPI()
+    principal = {"value": ExtensionPrincipal("alice", is_admin=True), "allowed": True}
+    setattr(app.state, EXTENSION_PRINCIPAL_RESOLVER_KEY, lambda request: principal["value"])
+
+    async def authorize(request, namespace, scope):
+        assert namespace == "knowledge-management"
+        assert scope == ("read" if request.method == "GET" else "write")
+        return principal["allowed"]
+
+    setattr(app.state, EXTENSION_PLUGIN_AUTHZ_RESOLVER_ASYNC_KEY, authorize)
+    app.include_router(build_router(provider))
+    return app, principal, calls, state
+
+
+API_BASE = "/api/extensions/knowledge-management/connections/local/bases/kb-1"
+
+
+@pytest.mark.asyncio
+async def test_router_enforces_host_policy_and_ignores_no_client_identity(api):
+    app, principal, calls, _ = api
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
+        principal["allowed"] = False
+        response = await client.post(API_BASE + "/documents", json={"kind": "manual", "title": "Guide", "content": "Text"})
+        assert response.status_code == 403 and not calls
+        principal["allowed"] = True
+        response = await client.post(API_BASE + "/documents", json={"kind": "manual", "title": "Guide", "content": "Text", "user_id": "admin"})
+        assert response.status_code == 422 and not calls
+        response = await client.post(API_BASE + "/documents", json={"kind": "manual", "title": "Guide", "content": "Text"})
+        assert response.status_code == 202 and response.json()["state"] == "queued"
+
+
+@pytest.mark.asyncio
+async def test_router_upload_is_bounded_and_errors_are_sanitized(api, monkeypatch):
+    import deerflow_extension_knowledge_management.router as router
+
+    app, _, calls, state = api
+    monkeypatch.setattr(router, "MAX_UPLOAD_BYTES", 8)
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
+        response = await client.post(API_BASE + "/files?filename=guide.txt", content=b"123456789", headers={"content-type": "application/octet-stream"})
+        assert response.status_code == 413 and not calls
+        state["code"] = 500
+        response = await client.get(API_BASE + "/documents/doc-1")
+        assert response.status_code == 502
+        assert "synthetic-secret" not in response.text
+        assert len(calls) == 1
+
+
+@pytest.mark.asyncio
+async def test_router_fails_closed_without_host_resolver(api):
+    from deerflow_extension_api.auth import EXTENSION_PRINCIPAL_RESOLVER_KEY
+
+    app, _, calls, _ = api
+    delattr(app.state, EXTENSION_PRINCIPAL_RESOLVER_KEY)
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
+        response = await client.get(API_BASE + "/capabilities")
+        assert response.status_code == 403 and not calls
+
+
+def test_install_is_opt_in_and_registers_generic_extension_points(management):
+    from deerflow_extension_knowledge_management import install
+
+    class Registry:
+        plugins = []
+        routes = []
+
+        def plugin(self, value):
+            self.plugins.append(value)
+            return True
+
+        def routers(self, value):
+            self.routes.extend(value)
+
+    registry = Registry()
+    install(registry, {})
+    assert registry.plugins == registry.routes == []
+    provider, _, _ = management
+    install(registry, {"enabled": True, "provider_config": provider.settings.model_dump(mode="json")})
+    assert registry.plugins[0].namespace == "knowledge-management"
+    assert len(registry.routes) == 1
+
+
+@pytest.mark.asyncio
+async def test_ingested_document_flows_through_existing_scope_and_citations(management, monkeypatch):
+    from deerflow_extension_knowledge_management.contracts import IngestRequest, ManagementContext
+
+    from deerflow.knowledge import KnowledgeRequestContext, format_result, sources_from_messages
+
+    monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[2] / "examples/deerflow-extension-weknora"))
+    from deerflow_extension_weknora import WeKnoraProvider
+
+    provider, _, state = management
+    result = await provider.ingest(ManagementContext(ExtensionPrincipal("alice", is_admin=True)), "local", "kb-1", IngestRequest(kind="manual", title="Guide", content="Cedar evidence"))
+    state["status"] = "completed"
+    result = await provider.status(ManagementContext(ExtensionPrincipal("alice", is_admin=True)), "local", "kb-1", result.document_id)
+    assert result.searchable
+
+    def read_handler(request):
+        path = request.url.path
+        if path == "/api/v1/knowledge-bases":
+            data = [{"id": "kb-1", "name": "Guide"}]
+        elif path.endswith("hybrid-search"):
+            assert json.loads(request.content)["knowledge_ids"] == [result.document_id]
+            data = [{"knowledge_id": result.document_id, "content": "Cedar evidence", "knowledge_title": "Guide", "score": 0.8}]
+        else:
+            data = {"id": result.document_id, "knowledge_base_id": "kb-1", "parse_status": "completed", "enable_status": "enabled", "title": "Guide"}
+        return httpx.Response(200, json={"success": True, "data": data})
+
+    retrieval = WeKnoraProvider({"spaces": [{"base_url": "https://weknora.example", "tenant_id": "1", "api_key_env": "MANAGEMENT_TEST_KEY", "user_ids": ["alice"]}]}, transport=httpx.MockTransport(read_handler))
+    scope = {"version": 1, "mode": "selected", "dataset_ids": [result.dataset_id], "document_filters": [{"dataset_id": result.dataset_id, "document_ids": [result.document_id]}]}
+    found = await retrieval.search_scoped("Cedar", context=KnowledgeRequestContext("alice", "lead_agent", scope))
+    content, artifact = format_result(found)
+    replay = json.loads(json.dumps({"type": "tool", "content": content, "artifact": artifact}))
+    sources = sources_from_messages([replay])
+    assert len(sources) == 1 and sources[0].provider == "weknora"
+    assert sources[0].excerpt == "Cedar evidence"
+
+
+@pytest.mark.asyncio
+async def test_generic_router_supports_non_weknora_provider(api):
+    from deerflow_extension_api.auth import EXTENSION_PLUGIN_AUTHZ_RESOLVER_ASYNC_KEY, EXTENSION_PRINCIPAL_RESOLVER_KEY
+    from deerflow_extension_knowledge_management.contracts import Capabilities
+    from deerflow_extension_knowledge_management.router import build_router
+    from fastapi import FastAPI
+
+    class SearchOnly:
+        async def capabilities(self, context, connection, base):
+            assert context.principal.user_id == "alice"
+            return Capabilities(operations=frozenset({"capabilities"}), processing_options=frozenset(), pipeline={"parse": "unsupported"})
+
+        async def ingest(self, *args):
+            raise NotImplementedError
+
+    original, _, _, _ = api
+    app = FastAPI()
+    for key in (EXTENSION_PRINCIPAL_RESOLVER_KEY, EXTENSION_PLUGIN_AUTHZ_RESOLVER_ASYNC_KEY):
+        setattr(app.state, key, getattr(original.state, key))
+    app.include_router(build_router(SearchOnly()))
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
+        response = await client.get(API_BASE + "/capabilities")
+        assert response.status_code == 200 and response.json()["operations"] == ["capabilities"]
+        response = await client.post(API_BASE + "/documents", json={"kind": "manual", "title": "Guide", "content": "Text"})
+        assert response.status_code == 422
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("code,expected", [(302, 502), (401, 403), (403, 403), (404, 404), (409, 409), (500, 502)])
+async def test_http_provider_failures_do_not_retry(api, code, expected):
+    app, _, calls, state = api
+    state["code"] = code
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
+        response = await client.post(API_BASE + "/documents", json={"kind": "manual", "title": "Guide", "content": "Text"})
+        assert response.status_code == expected and len(calls) == 1
+        assert "synthetic-secret" not in response.text
+
+
+@pytest.mark.asyncio
+async def test_json_size_limit_precedes_provider_calls(api):
+    app, _, calls, _ = api
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
+        response = await client.post(API_BASE + "/documents", content=b" " * (2 * 1024 * 1024 + 1), headers={"content-type": "application/json"})
+        assert response.status_code == 413 and not calls
+
+
+@pytest.mark.parametrize(
+    "payload",
+    [{"kind": "url", "url": "https://user:password@example.com"}, {"kind": "manual", "title": "Guide", "content": " "}, {"kind": "url", "url": "file:///etc/passwd"}, {"kind": "url", "url": "https://example.com", "content": "unexpected"}],
+)
+def test_ingest_contract_rejects_unsafe_or_ambiguous_payloads(management, payload):
+    from deerflow_extension_knowledge_management.contracts import IngestRequest
+
+    with pytest.raises(ValueError):
+        IngestRequest.model_validate(payload)
+
+
+@pytest.mark.asyncio
+async def test_poll_rechecks_grants_and_never_caches_authorization(management):
+    from deerflow_extension_knowledge_management.contracts import ManagementContext
+
+    provider, calls, _ = management
+    await provider.status(ManagementContext(ExtensionPrincipal("alice", is_admin=True)), "local", "kb-1", "doc-1")
+    count = len(calls)
+    with pytest.raises(PermissionError):
+        await provider.status(ManagementContext(ExtensionPrincipal("mallory", is_admin=True)), "local", "kb-1", "doc-1")
+    assert len(calls) == count
+
+
+@pytest.mark.asyncio
+async def test_missing_credential_never_uses_another_connection(management, monkeypatch):
+    from deerflow_extension_knowledge_management.contracts import ManagementContext
+
+    provider, calls, _ = management
+    monkeypatch.delenv("MANAGEMENT_TEST_KEY")
+    with pytest.raises(RuntimeError, match="credential"):
+        await provider.status(ManagementContext(ExtensionPrincipal("alice", is_admin=True)), "local", "kb-1", "doc-1")
+    assert not calls
+
+
+@pytest.mark.asyncio
+async def test_declared_optional_operations_fail_closed(management):
+    from deerflow_extension_knowledge_management.contracts import ManagementContext, ManagementProvider, ProcessingOptions
+
+    class CatalogOnly(ManagementProvider):
+        pass
+
+    with pytest.raises(NotImplementedError):
+        await CatalogOnly().upload(ManagementContext(ExtensionPrincipal("alice", is_admin=True)), "connection", "base", "file.txt", b"text", ProcessingOptions())
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("operation", ["manual", "url", "upload", "reprocess"])
+async def test_parent_child_chunk_overrides_rejected_before_mutation(management, operation):
+    from deerflow_extension_knowledge_management.contracts import IngestRequest, ManagementContext, ProcessingOptions
+
+    provider, calls, state = management
+    state["chunking_config"] = {"chunk_size": 1024, "chunk_overlap": 128, "enable_parent_child": True, "parent_chunk_size": 4096, "child_chunk_size": 384}
+    context = ManagementContext(ExtensionPrincipal("alice", is_admin=True))
+    options = ProcessingOptions(chunk_size=512, chunk_overlap=64)
+    with pytest.raises(NotImplementedError, match="parent-child"):
+        if operation == "upload":
+            await provider.upload(context, "local", "kb-1", "guide.txt", b"text", options)
+        elif operation == "reprocess":
+            await provider.process(context, "local", "kb-1", "doc-1", operation, options)
+        else:
+            body = {"kind": operation, "title": "Guide", "processing": options}
+            body.update({"content": "text"} if operation == "manual" else {"url": "https://example.com"})
+            await provider.ingest(context, "local", "kb-1", IngestRequest(**body))
+    assert calls and all(call.method == "GET" for call in calls)
+    assert state["chunking_config"]["parent_chunk_size"] == 4096
+
+
+def test_parent_child_without_chunk_overrides_preserves_policy(management):
+    from deerflow_extension_knowledge_management.contracts import ProcessingOptions
+
+    provider, _, _ = management
+    base = {"chunking_config": {"enable_parent_child": True, "parent_chunk_size": 4096, "child_chunk_size": 384}}
+    assert provider._options(ProcessingOptions(), base) == {}
+    assert provider._options(ProcessingOptions(summary_enabled=False), base) == {"summary_enabled": False}
+
+
+def test_weknora_zero_overlap_override_is_not_silently_ignored(management):
+    from deerflow_extension_knowledge_management.contracts import ProcessingOptions
+
+    provider, _, _ = management
+    with pytest.raises(NotImplementedError):
+        provider._options(ProcessingOptions(chunk_size=512, chunk_overlap=0), {"chunking_config": {"chunk_overlap": 128}})
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("failure", ["timeout", "connect", "json", "envelope", "status"])
+async def test_provider_failure_diagnostics_are_sanitized(management, caplog, failure):
+    from deerflow_extension_knowledge_management.contracts import ManagementContext
+    from deerflow_extension_knowledge_management.router import _invoke
+    from fastapi import HTTPException
+
+    provider, _, _ = management
+    secret = "synthetic-secret private-payload https://private.example/token"
+
+    def fail(request):
+        if failure == "timeout":
+            raise httpx.ReadTimeout(secret, request=request)
+        if failure == "connect":
+            raise httpx.ConnectError(secret, request=request)
+        if failure == "json":
+            return httpx.Response(200, text=secret)
+        if failure == "envelope":
+            return httpx.Response(200, json={"error": secret})
+        return httpx.Response(503, text=secret)
+
+    provider.transport = httpx.MockTransport(fail)
+    with caplog.at_level("WARNING"), pytest.raises(HTTPException) as caught:
+        await _invoke(provider.status(ManagementContext(ExtensionPrincipal("alice", is_admin=True)), "local", "kb-1", "doc-1"))
+    assert caught.value.status_code == 502
+    expected = {"timeout": "ReadTimeout", "connect": "ConnectError", "json": "JSONDecodeError", "envelope": "invalid_envelope", "status": "status=503"}[failure]
+    assert expected in caplog.text
+    assert "RuntimeError" in caplog.text
+    for text in (caplog.text, str(caught.value.detail)):
+        assert "synthetic-secret" not in text and "private-payload" not in text and "private.example" not in text
+    assert all(record.exc_info is None for record in caplog.records)
+
+
+@pytest.mark.asyncio
+async def test_custom_provider_exception_text_is_not_logged(management, caplog):
+    from deerflow_extension_knowledge_management.router import _invoke
+    from fastapi import HTTPException
+
+    async def fail():
+        raise RuntimeError("synthetic-secret private document")
+
+    with caplog.at_level("WARNING"), pytest.raises(HTTPException):
+        await _invoke(fail())
+    assert "RuntimeError" in caplog.text
+    assert "synthetic-secret" not in caplog.text and "private document" not in caplog.text
+    assert all(record.exc_info is None for record in caplog.records)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("chunking,graph,expected", [({"enable_parent_child": True}, False, {"summary_enabled"}), (None, True, {"summary_enabled", "graph_enabled"}), ({}, False, {"summary_enabled", "chunk_size", "chunk_overlap"})])
+async def test_capability_options_reflect_base_support(management, chunking, graph, expected):
+    from deerflow_extension_knowledge_management.contracts import ManagementContext
+
+    provider, _, _ = management
+    provider.transport = httpx.MockTransport(lambda request: httpx.Response(200, json={"success": True, "data": {"id": "kb-1", "chunking_config": chunking, "capabilities": {"graph": graph}}}))
+    result = await provider.capabilities(ManagementContext(ExtensionPrincipal("alice", is_admin=True)), "local", "kb-1")
+    assert result.processing_options == expected
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("credential", [None, "", "invalid secret", "非ASCII"])
+async def test_unavailable_credential_is_local_service_failure(api, monkeypatch, caplog, credential):
+    app, _, calls, _ = api
+    if credential is None:
+        monkeypatch.delenv("MANAGEMENT_TEST_KEY")
+    else:
+        monkeypatch.setenv("MANAGEMENT_TEST_KEY", credential)
+    with caplog.at_level("WARNING"):
+        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
+            response = await client.get(API_BASE + "/capabilities")
+    assert response.status_code == 503
+    assert response.json()["detail"] == "Management credential is not configured"
+    assert not calls
+    assert "ManagementCredentialUnavailable" in caplog.text
+    assert "reconcile" not in response.text
+    if credential:
+        assert credential not in caplog.text + response.text
diff --git a/backend/tests/test_knowledge_management_extension_live.py b/backend/tests/test_knowledge_management_extension_live.py
new file mode 100644
index 0000000000..4abe89e560
--- /dev/null
+++ b/backend/tests/test_knowledge_management_extension_live.py
@@ -0,0 +1,65 @@
+"""Explicit opt-in writes to a dedicated, operator-granted test base.
+
+No automatic retries or cleanup: preserve document IDs to reconcile outcomes.
+The API and upstream must be real; session/CSRF credentials come from the caller.
+"""
+
+import asyncio
+import json
+import os
+from pathlib import Path
+from uuid import uuid4
+
+import httpx
+import pytest
+
+pytestmark = [pytest.mark.live, pytest.mark.asyncio]
+
+
+async def test_live_management_ingestion_to_scoped_evidence(monkeypatch):
+    if os.getenv("DEER_FLOW_RUN_LIVE_TESTS") != "1" or os.getenv("KNOWLEDGE_MANAGEMENT_ALLOW_WRITES") != "1":
+        pytest.skip("management writes require two explicit opt-ins")
+    required = ["KNOWLEDGE_GATEWAY_URL", "KNOWLEDGE_SESSION_COOKIE", "KNOWLEDGE_CSRF_TOKEN", "KNOWLEDGE_CONNECTION_ID", "WEKNORA_BASE_URL", "WEKNORA_TENANT_ID", "WEKNORA_KNOWLEDGE_BASE_ID", "WEKNORA_API_KEY"]
+    if any(not os.getenv(key) for key in required):
+        pytest.fail("Set all documented management live-test environment variables")
+    from urllib.parse import quote
+
+    gateway = os.environ["KNOWLEDGE_GATEWAY_URL"].rstrip("/")
+    base = os.environ["WEKNORA_KNOWLEDGE_BASE_ID"]
+    path = f"/api/extensions/knowledge-management/connections/{quote(os.environ['KNOWLEDGE_CONNECTION_ID'], safe='')}/bases/{quote(base, safe='')}"
+    marker = "MANAGEMENT-LIVE-" + uuid4().hex
+    cookies = {"access_token": os.environ["KNOWLEDGE_SESSION_COOKIE"], "csrf_token": os.environ["KNOWLEDGE_CSRF_TOKEN"]}
+    headers = {"Origin": gateway, "X-CSRF-Token": os.environ["KNOWLEDGE_CSRF_TOKEN"]}
+    async with httpx.AsyncClient(base_url=gateway, cookies=cookies, headers=headers, timeout=90) as client:
+        result = await client.get(path + "/capabilities")
+        assert result.status_code == 200
+        assert {"manual", "status"} <= set(result.json()["operations"])
+        response = await client.post(
+            path + "/documents", json={"kind": "manual", "title": marker, "content": f"Synthetic acceptance fixture. Approval marker {marker}. Travel allowance is 735 CNY.", "processing": {"summary_enabled": False}}
+        )
+        assert response.status_code == 202, "Write failed or uncertain: reconcile upstream before running again"
+        document = response.json()
+        # A native ID is safe diagnostic evidence and necessary for reconciliation.
+        print(f"Created synthetic live document: {document['document_id']}")
+        for _ in range(60):
+            response = await client.get(path + "/documents/" + quote(document["document_id"], safe=""))
+            assert response.status_code == 200
+            state = response.json()
+            if state["state"] in {"completed", "failed", "cancelled"}:
+                break
+            await asyncio.sleep(2)
+        else:
+            pytest.fail("Processing timed out; retain the printed document ID for reconciliation")
+        assert state["state"] == "completed" and state["searchable"], "Upstream processing did not complete"
+
+    monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[2] / "examples/deerflow-extension-weknora"))
+    from deerflow_extension_weknora import WeKnoraProvider
+
+    from deerflow.knowledge import KnowledgeRequestContext, format_result, sources_from_messages
+
+    provider = WeKnoraProvider({"spaces": [{"base_url": os.environ["WEKNORA_BASE_URL"], "tenant_id": os.environ["WEKNORA_TENANT_ID"], "api_key_env": "WEKNORA_API_KEY", "user_ids": ["live-probe"], "knowledge_base_ids": [base]}]})
+    scope = {"version": 1, "mode": "selected", "dataset_ids": [document["dataset_id"]], "document_filters": [{"dataset_id": document["dataset_id"], "document_ids": [document["document_id"]]}]}
+    result = await provider.search_scoped(marker, context=KnowledgeRequestContext("live-probe", "lead_agent", scope))
+    content, artifact = format_result(result)
+    sources = sources_from_messages([json.loads(json.dumps({"type": "tool", "content": content, "artifact": artifact}))])
+    assert sources and any(marker in source.excerpt for source in sources)
diff --git a/examples/deerflow-extension-knowledge-management/AGENTS.md b/examples/deerflow-extension-knowledge-management/AGENTS.md
new file mode 100644
index 0000000000..621105086a
--- /dev/null
+++ b/examples/deerflow-extension-knowledge-management/AGENTS.md
@@ -0,0 +1,47 @@
+# Optional knowledge management
+
+Keep management business logic, provider configuration and HTTP API in this
+independently installed extension. Core retrieval must never depend on it.
+Use public principal/plugin authorization APIs; no host-internal imports.
+
+Administration is opt-in, explicit exact-user/base/operation grants plus the
+host administrator floor and upstream authorization. Retrieval bindings do not
+confer writes. Do not accept client user/tenant/Agent/space identities, widen a
+resource selection, retry writes, follow credential-bearing redirects or fall
+back to other credentials. Check document/base membership before mutations.
+
+Pin WeKnora mappings to v0.8.3 commit
+3e22822c1d2397be730c458c7539d0eb8c113a24. Upload/URL may return 201.
+Processing capabilities describe upstream configuration, not observed success.
+Finalizing may be searchable, and unknown states must remain unknown. Never
+claim Wiki/graph completion based on the document's terminal parse status.
+Reject chunk size/overlap overrides before mutation on parent-child bases;
+upstream derives effective sizes and overlap independently. Preserve base
+chunking settings for supported overrides: v0.8.3 treats
+parent-child mode as authoritative and ignores zero integer overrides. Reject
+an unrepresentable zero overlap. Empty reparse options preserve the stored policy. Keep arbitrary credential,
+model and parser configuration out of request bodies.
+
+Tests belong in backend/tests/test_knowledge_management_extension.py and
+backend/tests/blocking_io/test_knowledge_management_extension.py. Use mock
+transport for offline tests and actual HTTPX SSL initialization for blocking-I/O
+coverage. Keep simulated and real validation explicitly separate.
+
+Live acceptance lives in backend/tests/test_knowledge_management_extension_live.py.
+Require both DEER_FLOW_RUN_LIVE_TESTS and KNOWLEDGE_MANAGEMENT_ALLOW_WRITES;
+use a dedicated synthetic base and caller-supplied session/CSRF credentials.
+Preserve native IDs for uncertain-write reconciliation; no automatic write retry
+or cleanup. Record real stage outcomes and lexical Wiki-search limitations in
+VALIDATION.md separately from mock contract coverage.
+
+Frontend discovery uses the knowledge-management namespace with English/Chinese
+copy; catalog presence never activates routes or implies a management page.
+
+Provider failure logs include only exception types, numeric HTTP status and fixed
+response-failure categories. Never log exception text/chains, payloads, URLs or
+credentials; caplog regressions exercise transport and custom-provider failures.
+
+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.
diff --git a/examples/deerflow-extension-knowledge-management/CONTRACT.md b/examples/deerflow-extension-knowledge-management/CONTRACT.md
new file mode 100644
index 0000000000..46e75c344b
--- /dev/null
+++ b/examples/deerflow-extension-knowledge-management/CONTRACT.md
@@ -0,0 +1,34 @@
+# Management handoff, version 1
+
+This contract is extension-owned and experimental. It deliberately avoids a second core registry: the existing `PluginContribution`, contributed router and principal/authorization bridge are sufficient. Third-to-fifth-stage consumers can integrate the HTTP contract without installing a management page. They must not treat this document as a claim that space authorization or a Trace service has been implemented.
+
+## Identity and resources
+
+`ManagementContext.principal` is resolved by the host, not deserialized from JSON. The initial implementation accepts only explicitly granted administrators. The connection ID selects operator configuration; it is neither a tenant credential nor a logical space ID. No executing Agent or space is implied by an interactive HTTP call. Future delegated contexts must be host-verified before resource operations; a client-supplied Agent/space cannot extend permissions.
+
+Resource identity is `(provider, connection_id, native_base_id, document_id)`. `dataset_id` is the retrieval-facing identity, maintained consistently with the corresponding retrieval adapter. A management result does not create a scope grant or a citation; only the existing retrieval evidence validation can do that. Same-named bases on different connections must not be merged.
+
+## Capabilities and requests
+
+Each base returns the intersection of configured management operations and the adapter's supported surface. `pipeline` describes the configured Provider's support, not observed task execution. The presence of a stage does not authorize a write. Unrecognized inputs and unavailable operations must fail rather than widen the request or silently drop options.
+
+Processing options reflect each base: graph controls require graph support; chunk controls require a valid non-parent-child chunking configuration. Local credential validation fails before HTTP with `ManagementCredentialUnavailable`, mapped to `503` without reconciliation guidance.
+
+The supplied WeKnora adapter exposes ingest, status, reprocess and cancel. Upload returns the Provider's native document identity. The Provider owns queues, parser/chunker/embedding work, Wiki generation, entity extraction and index storage. No local scheduler is involved.
+
+## State and uncertainty
+
+Document snapshots normalize upstream state without creating local task IDs. Polls always recheck authorization. A newer upstream state remains `unknown` until its semantics are explicitly mapped. A searchable document can still be `running` while enrichment proceeds. A terminal document state does not imply every optional stage succeeded.
+
+No timestamp, attempt ID, percentage or step success is invented. The future Trace adapter must consume actual Provider events/spans, independently authorize them, and link them by connection/base/document (and upstream attempt when available). Raw source text and upstream error bodies are not part of this contract. Full Trace retention and sensitive-data policy remain with that separate extension.
+
+Submitting a mutation has no automatic retry. A response loss may leave an accepted upstream operation. A consumer must reconcile before retrying; there is no local durable exactly-once promise. For a future durable workflow, add persistent admission/reconciliation before advertising retry-safe request IDs.
+
+Server-side warning logs distinguish transport exception types, HTTP status and
+malformed/oversized response categories. Logs omit exception messages, tracebacks,
+request URLs and provider payloads because these can contain credentials or source
+content. These diagnostics do not provide durable admission or processing Trace.
+
+## Compatibility
+
+No changes are made to `knowledge_scope`, retrieval tools, source artifacts, frontend citation models or stored messages. Offline fixtures in `backend/tests/test_knowledge_management_extension.py` demonstrate shared dataset identity and citation replay with the existing WeKnora read-only provider. Synthetic fixture transitions are not live processing evidence.
diff --git a/examples/deerflow-extension-knowledge-management/README.md b/examples/deerflow-extension-knowledge-management/README.md
new file mode 100644
index 0000000000..76b5ac6d19
--- /dev/null
+++ b/examples/deerflow-extension-knowledge-management/README.md
@@ -0,0 +1,138 @@
+# Optional knowledge management extension
+
+This package implements the first management slice of [#6269's second stage](https://github.com/bytedance/deer-flow/issues/6269#issuecomment-6336870813): document ingestion and provider-owned processing. It installs no frontend, model tools, database, parser, embedding model or index. Existing retrieval works without this package.
+
+The supplied adapter targets **WeKnora v0.8.3**, commit `3e22822c1d2397be730c458c7539d0eb8c113a24`. API mappings were checked against its handlers and `superbuzzy/deer-flow`'s `community/weknora/documents.py`. This is a new independent extension using the current host contracts, not a merge of that branch.
+
+## Installation and authorization
+
+From the repository root:
+
+```sh
+make extension-install SOURCE=./examples/deerflow-extension-knowledge-management
+```
+
+Installation alone does not expose management routes. Configure the installed plugin's `config` in operator-owned `config.yaml`, then restart Gateway:
+
+```yaml
+plugins:
+  - use: deerflow_extension_knowledge_management:install
+    config:
+      enabled: true
+      provider_config:
+        connections:
+          - id: team-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-weknora-base-id]
+                operations: [capabilities, status, upload, url, manual, reprocess, cancel]
+```
+
+Set the credential in the Gateway environment. Use a restricted upstream key with retrieval and ingestion rights only for the intended bases. A retrieval-only key will fail writes; the extension never substitutes a more privileged key. Disable by setting `config.enabled: false` or removing the extension, then restarting Gateway. There is no runtime settings override or separate page switch.
+
+Every route requires all of:
+
+1. Gateway authentication and its normal CSRF protection for writes.
+2. Host `plugin_management` read/write authorization for `knowledge-management`.
+3. A host-resolved administrator principal, excluding internal callers.
+4. An explicit operator grant matching the exact host user, connection, native knowledge base and operation.
+5. Upstream resource authorization using that connection's key. Existing documents must also belong to the selected base before processing mutations.
+
+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.
+
+## HTTP contract, version 1
+
+All routes have this prefix:
+
+```text
+/api/extensions/knowledge-management/connections/{connection}/bases/{base}
+```
+
+`connection` is the operator-defined connection ID; `base` is the native Provider base ID. Browser clients use DeerFlow's authenticated API and normal CSRF mechanism.
+
+| Method and suffix | Request | Result |
+| --- | --- | --- |
+| `GET /capabilities` | None | Allowed operations, processing options, provider-managed stages |
+| `POST /documents` | URL or manual JSON below | `202` with document processing snapshot |
+| `POST /files?filename=guide.txt` | Raw `application/octet-stream`, at most 16 MiB | `202` with snapshot |
+| `GET /documents/{document}` | None | Current upstream document processing snapshot |
+| `POST /documents/{document}/reprocess` | Processing options object, or `{}` | `202` with upstream snapshot |
+| `POST /documents/{document}/cancel` | `{}` | `202` with upstream snapshot |
+
+Manual ingestion explicitly publishes the source, so it enters the upstream processing pipeline instead of remaining a draft:
+
+```json
+{"kind":"manual","title":"Guide","content":"Cedar evidence","processing":{"chunk_size":512,"chunk_overlap":64}}
+```
+
+URL ingestion delegates fetching and SSRF enforcement to WeKnora's validated URL-ingestion endpoint; DeerFlow does not fetch the URL:
+
+```json
+{"kind":"url","url":"https://example.com/guide","title":"Guide"}
+```
+
+File ingestion optionally accepts `X-Processing-Options` containing the same JSON processing options. JSON requests are limited to 2 MiB; manual text is limited to 1,000,000 characters. Filenames cannot contain path separators or header control characters. No client filesystem path is accepted.
+
+The snapshot contains `version`, `provider`, `connection_id`, `dataset_id`, `document_id`, `state`, `provider_state`, and `searchable`. The WeKnora `dataset_id` uses exactly the existing retrieval adapter's endpoint/tenant/base identity. A completed document can therefore be selected in the existing chat scope selector when the separately configured retrieval connection authorizes it.
+
+States are `queued`, `running`, `completed`, `failed`, `cancelled`, or `unknown`. New upstream statuses stay `unknown`; they are never interpreted as successful. `finalizing` is `running` and may already be searchable, matching the existing retrieval contract. Cancellation is not deletion or rollback of already written chunks. `202` confirms a provider response, not completed processing. A document's terminal parse status does **not** establish that every Wiki/graph enrichment succeeded.
+
+Error semantics: `403` denied, `404` missing resource, `409` conflict, `413` size limit, `415` upload media type, `422` invalid/unsupported input, `502` upstream failure, `503` missing/invalid local management credential (no upstream request made). Raw upstream bodies and credentials are never returned. No writes are automatically retried, including on timeout, cancellation, malformed response or conflict. If delivery is uncertain, reconcile using the upstream catalog before resubmitting; this version does not promise exactly-once ingestion or persist admission records.
+
+## Processing scope
+
+| Pipeline part | Adapter behavior |
+| --- | --- |
+| Parse and chunk | Performed by upstream ingestion/reparse; bounded chunk size/overlap overrides |
+| Vector indexing | Uses the upstream base's model and indexing configuration; capability describes availability |
+| Summary | Optional `summary_enabled` override |
+| Knowledge entry generation/update | Upstream Wiki pipeline of a Wiki-enabled base during ingestion/reparse; no standalone Wiki editing endpoint |
+| Entity extraction, relations and graph indexing | Upstream graph pipeline; `graph_enabled: true` requires that the base already supports graph extraction |
+| Retry/cancel | Explicit reprocess/cancel operations; no automatic retry or local worker |
+
+Missing optional stages are reported as `unsupported`. Processing options are base-aware: graph controls require graph support, and chunk overrides are advertised only for a valid non-parent-child chunking configuration. The extension does not simulate them or accept arbitrary provider options. Parser engine credentials, model selection, Wiki settings and graph extraction schema remain operator-controlled in WeKnora; this API cannot change the base-wide policy. Chunk overrides preserve the upstream base’s other chunking settings. Bases using parent-child mode reject chunk size/overlap overrides with `422` before mutation because upstream derives their effective sizes separately; requests without chunk overrides retain that mode. WeKnora v0.8.3 cannot override a nonzero overlap with zero; this request is rejected instead of silently retaining the old value. Empty reprocess options preserve the stored upstream processing policy. The API does not claim independent execution or progress for each stage.
+
+Platform connections/spaces, resource-sharing UI, a knowledge center, Trace persistence, Wiki page lifecycle editing and multi-hop graph algorithms remain separate work. Processing snapshots contain no raw Trace, source content or model prompts.
+
+## Provider integration and tests
+
+The router only consumes `contracts.ManagementProvider`. An operator can set `provider_factory` to another trusted `module:factory` implementing the optional management contract. This executable setting is not API-writable. Providers must enforce their own resource grants and upstream authorization in every operation and explicitly reject unsupported operations. See [CONTRACT.md](CONTRACT.md) for the handoff boundary.
+
+Run offline tests from the repository root:
+
+```sh
+uv run --no-sync --project backend pytest backend/tests/test_knowledge_management_extension.py -q
+uv run --no-sync --project backend pytest backend/tests/blocking_io/test_knowledge_management_extension.py -q
+```
+
+Tests use synthetic `httpx.MockTransport` responses. They cover ingestion → status → existing scope/retrieval → citation replay, denied identities/resources, upstream errors without credential fallback, request size limits, strict options, optional installation and a second test Provider. Blocking-I/O coverage exercises actual HTTPX/SSL initialization off-loop. These tests do **not** establish real upstream extraction/indexing quality or browser acceptance.
+
+For live acceptance, use a dedicated test base and explicitly granted administrator: upload a unique test document, poll its document ID, confirm searchable state, select it in the existing chat scope selector, ask a question whose answer is unique to that source, and verify the citation after refreshing. Repeat with URL/manual ingestion and with Wiki/graph-enabled bases, checking upstream processing outcomes rather than inferring enrichment success from parse completion. Record upstream version and configuration; never commit credentials or document content. The offline suite never performs live writes. Real OrbStack/GLM-5.3/BGE acceptance, environment repairs and observed limitations are recorded in [VALIDATION.md](VALIDATION.md).
+
+
+### Opt-in live regression
+
+Use a dedicated synthetic test base and an authenticated, explicitly granted
+administrator. Export these variables without committing their values:
+
+- `DEER_FLOW_RUN_LIVE_TESTS=1` and `KNOWLEDGE_MANAGEMENT_ALLOW_WRITES=1`
+- `KNOWLEDGE_GATEWAY_URL`: browser/Gateway origin, including scheme and port
+- `KNOWLEDGE_SESSION_COOKIE`: value of the authenticated `access_token` cookie
+- `KNOWLEDGE_CSRF_TOKEN`: matching CSRF cookie/header value
+- `KNOWLEDGE_CONNECTION_ID`: configured management connection
+- `WEKNORA_BASE_URL`, `WEKNORA_TENANT_ID`, `WEKNORA_KNOWLEDGE_BASE_ID`
+- `WEKNORA_API_KEY`: separately scoped retrieval credential for that test base
+
+```sh
+uv run --no-sync --project backend pytest backend/tests/test_knowledge_management_extension_live.py -m live -q -s --override-ini=addopts=
+```
+
+Each run creates one uniquely marked synthetic manual document, prints its native
+ID for reconciliation, polls for completion, and verifies document-scoped retrieval
+and serialized citation evidence. It neither retries writes nor deletes fixtures.
+If submission is uncertain, inspect the upstream catalog before rerunning. Without
+both opt-ins the test skips; missing credentials after opt-in fail explicitly.
+This API regression does not replace browser or upstream Wiki/graph inspection.
diff --git a/examples/deerflow-extension-knowledge-management/VALIDATION.md b/examples/deerflow-extension-knowledge-management/VALIDATION.md
new file mode 100644
index 0000000000..db3678bb11
--- /dev/null
+++ b/examples/deerflow-extension-knowledge-management/VALIDATION.md
@@ -0,0 +1,116 @@
+# Stage 2 acceptance record
+
+Date: 2026-10-22. Base revision: `29ba806fc`, plus this uncommitted
+knowledge-management extension and tests. All ingested documents were synthetic.
+This records the first Stage 2 management slice, not completion of the whole
+knowledge-platform roadmap.
+
+## Real environment
+
+- macOS with OrbStack; authenticated DeerFlow Gateway, Next.js and nginx.
+- WeKnora `v0.8.3` (adapter source pin
+  `3e22822c1d2397be730c458c7539d0eb8c113a24`), ParadeDB/PostgreSQL 17, Redis.
+- Official WeKnora DocReader `v0.8.3`; Neo4j `2025.10.1` with APOC.
+- GLM-5.3 using the operator's `.env.local` endpoint/credentials; DeerFlow
+  acceptance model used `chat_template_kwargs.enable_thinking=false`.
+- Local Ollama BGE-M3, 1024-dimensional vectors (actual embedding request checked).
+- LightRAG `v1.5.7` with an existing synthetic graph for retrieval regression.
+- Separate restricted management and retrieval keys; exact administrator/base
+  grants. Authentication and CSRF remained enabled.
+
+Private configuration, credentials, scripts, complete logs and screenshots remain
+outside the repository in the local `knowledge-phase2-local` evidence directory.
+No credential values or private model URLs are part of this report.
+
+## Observed outcomes
+
+| Check | Actual outcome |
+| --- | --- |
+| Manual text ingestion | `202`, then completed/searchable; unique marker retrieved from the selected document |
+| Markdown file ingestion | `202`, then completed/searchable |
+| URL ingestion | Initial DocReader failure; explicit reprocessing completed after the environment repair below |
+| Capability discovery | Plain document and Wiki/graph-enabled bases advertised different optional capabilities |
+| Cancel and explicit recovery | Cancellation returned `202`; cancelled/not searchable observed; explicit reprocess returned `202`, then completed with graph and Wiki spans done |
+| Graph extraction/indexing | Graph span done; Neo4j contained two actual relations for the synthetic project; entity node chunk IDs resolved to the original source text |
+| Wiki generation | Published entity/project/organization/summary pages and Index observed; exact keyword retrieval returned attributable evidence |
+| Browser GLM question | One base/one document selected; correct allowance, deadline and synthetic marker, with clickable source text |
+| Citation persistence | Source opened after refresh and after Gateway restart; prior answer and source excerpt persisted |
+| API live regression | Management ingestion → selected retrieval → serialized citation replay, plus selected retrieval and Wiki regressions: 3 passed |
+| Three-tool GLM regression | Explicit Wiki keyword: 3 passed in 15.64 seconds, including real document/Wiki/LightRAG evidence and citation serialization/forwarding/pruning |
+
+Nine real authorization/input checks passed: anonymous `401`, missing CSRF `403`,
+ungranted base `403`, cross-base document `403`, forged identity field `422`,
+unsupported graph option `422`, ordinary user `403`, retrieval-only upstream key
+write `403`, and upstream outside-base access `403`.
+
+The browser synthetic answer was 735 CNY/night, submission within 12 calendar
+days, marker `PHASE2-YEW-735-12`. Graph evidence contained project → leader and
+organization → project relationships. Node provenance was checked; the upstream
+edges had no direct chunk properties. This is not proof of edge-specific
+provenance, arbitrary graph quality or multi-hop reasoning.
+
+## Failures and environment repairs
+
+The existing local stack lacked DocReader and Neo4j. URL ingestion initially
+reported `DOCREADER_UNAVAILABLE`. Official DocReader and Neo4j were added only
+to the private Compose configuration, without publishing their ports. Reparse
+then exposed missing APOC functions; installing the official APOC plugin allowed
+explicit reconciliation/reprocessing to succeed. The local DNS returned Fake-IP
+addresses, requiring exact upstream SSRF allowlist entries for the model host
+and the synthetic URL host. No broad SSRF bypass was added.
+
+Free-form three-tool model runs failed the Wiki evidence assertion (two checks
+passed, one failed). Diagnostic tool calls showed the query
+`紫杉研发项目负责人 资助组织`. WeKnora's Wiki endpoint is lexical; neither that
+phrase nor its whitespace-separated terms matched the published page titles.
+A subsequent prompt explicitly requesting the keyword `紫杉` passed all three
+checks. The fixed-keyword result establishes integration, not unrestricted
+natural-language Wiki recall. No retrieval assertion was removed or weakened.
+
+One pipeline document remained `finalizing`/searchable during an additional
+upstream Wiki update after publishing pages. Final inspection confirmed
+`completed`, with chunking, embedding, graph and both Wiki update spans done.
+The separate cancellation recovery case also reached `completed` with all
+relevant spans done. The intermediate searchable state was not treated as
+terminal completion.
+
+## Offline regression and reproduction
+
+Before live acceptance, the full backend suite passed 35,467 tests across four
+shards (796 skipped); strict blocking-I/O coverage passed 276 tests (one skipped).
+The extension's focused contract and blocking-I/O tests passed 46 checks.
+The final focused run including agent-guidance checks passed 60 tests, with the
+non-opted-in live test skipped. Ruff lint/format and package wheel building passed. Live acceptance added a
+separately gated test; it does not alter normal offline test behavior.
+
+See [README.md](README.md#opt-in-live-regression) for the live environment variables
+and command. That test deliberately writes one unique synthetic document per run,
+retains its ID and does not automatically retry or delete it. Browser acceptance
+requires selecting the resulting document in the existing chat scope selector,
+asking a question unique to the fixture, opening its source, refreshing, then
+restarting Gateway and opening the persisted source again. Wiki/graph acceptance
+must additionally inspect the upstream spans, published pages and actual graph
+records; document parse completion alone is insufficient.
+
+Out of scope: a management frontend, platform spaces/sharing, arbitrary parser or
+model configuration, Trace persistence, standalone Wiki editing, upstream user
+impersonation, multi-Gateway ingestion admission and exactly-once writes.
+
+## PR review follow-up
+
+Review regressions now reject parent-child chunk overrides on manual, URL, file
+and reprocess paths before any upstream mutation; empty and non-chunk overrides
+remain supported. All four rejection cases failed before the fix and passed
+afterward. The oversized-upload case has a compact explicit pytest ID, avoiding
+Windows environment-variable limits (native Windows was not rerun locally).
+Frontend discovery includes the package's actual namespace and localized copy.
+Focused backend/guidance checks: 64 passed, 1 skipped. Frontend catalog: 9 passed;
+frontend lint/typecheck and backend formatting passed.
+
+Further review: capabilities now omit unsupported per-base controls, and invalid
+local credentials return a distinct pre-request 503. Seven new cases failed
+before the fix; focused backend/guidance checks now pass 77 tests (1 live skip).
+The three failing CI catalog E2E cases used the obsolete fixed count 11; they
+now use the catalog length plus explicit localized management-entry assertions.
+All three pass locally against an isolated Next.js development server. Catalog
+unit tests (9), frontend lint/typecheck and backend formatting also pass.
diff --git a/examples/deerflow-extension-knowledge-management/deerflow_extension_knowledge_management/__init__.py b/examples/deerflow-extension-knowledge-management/deerflow_extension_knowledge_management/__init__.py
new file mode 100644
index 0000000000..e1cc361222
--- /dev/null
+++ b/examples/deerflow-extension-knowledge-management/deerflow_extension_knowledge_management/__init__.py
@@ -0,0 +1,36 @@
+"""Optional management backend. No UI or retrieval installation dependency."""
+
+from importlib import import_module
+from typing import Any
+
+from deerflow_extension_api import extension
+from deerflow_extension_api.plugins import BackendAction, PluginContribution
+from pydantic import Field
+
+from .contracts import Model
+from .router import NAMESPACE, build_router
+
+
+class Settings(Model):
+    enabled: bool = Field(default=False, strict=True)
+    provider_factory: str = "deerflow_extension_knowledge_management.weknora:WeKnoraManagement"
+    provider_config: dict[str, Any] = Field(default_factory=dict)
+
+
+async def describe_contract(payload, context):
+    if not context.principal.is_admin or context.principal.is_internal:
+        raise PermissionError("Management requires an administrator")
+    return {"version": 1, "authorization": "explicit-admin-resource-grants", "execution": "provider-owned"}
+
+
+@extension(api="0.2.7", name="knowledge-management")
+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),))
diff --git a/examples/deerflow-extension-knowledge-management/deerflow_extension_knowledge_management/contracts.py b/examples/deerflow-extension-knowledge-management/deerflow_extension_knowledge_management/contracts.py
new file mode 100644
index 0000000000..6b5df3c8b6
--- /dev/null
+++ b/examples/deerflow-extension-knowledge-management/deerflow_extension_knowledge_management/contracts.py
@@ -0,0 +1,98 @@
+"""Version 1 management boundary, owned by the optional extension.
+
+No client identity, credentials, arbitrary upstream JSON, or local task store.
+Providers own execution and resource authorization; submission is not completion.
+"""
+
+from dataclasses import dataclass
+from typing import Literal, Protocol
+
+from deerflow_extension_api.auth import ExtensionPrincipal
+from pydantic import BaseModel, ConfigDict, Field, HttpUrl, model_validator
+
+Operation = Literal["capabilities", "status", "upload", "url", "manual", "reprocess", "cancel"]
+
+
+class ManagementCredentialUnavailable(RuntimeError):
+    """Local credential validation failed before an upstream request."""
+
+
+class Model(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+
+@dataclass(frozen=True)
+class ManagementContext:
+    principal: ExtensionPrincipal
+    # HTTP management does not claim an executing Agent or a platform space.
+    # Future delegates must be supplied by a trusted host, never request JSON.
+
+
+class ProcessingOptions(Model):
+    chunk_size: int | None = Field(default=None, ge=64, le=8192, strict=True)
+    chunk_overlap: int | None = Field(default=None, ge=0, le=4096, strict=True)
+    summary_enabled: bool | None = Field(default=None, strict=True)
+    graph_enabled: bool | None = Field(default=None, strict=True)
+
+    @model_validator(mode="after")
+    def chunks(self):
+        if (self.chunk_size is None) != (self.chunk_overlap is None):
+            raise ValueError("chunk_size and chunk_overlap must be supplied together")
+        if self.chunk_size is not None and self.chunk_overlap >= self.chunk_size:
+            raise ValueError("chunk_overlap must be smaller than chunk_size")
+        return self
+
+
+class IngestRequest(Model):
+    kind: Literal["url", "manual"]
+    title: str = Field(default="", max_length=500)
+    content: str | None = Field(default=None, min_length=1, max_length=1_000_000)
+    url: HttpUrl | None = None
+    processing: ProcessingOptions = Field(default_factory=ProcessingOptions)
+
+    @model_validator(mode="after")
+    def payload(self):
+        if self.kind == "manual":
+            if not self.title.strip() or self.content is None or not self.content.strip() or self.url is not None:
+                raise ValueError("manual ingestion requires title/content and no URL")
+        elif self.url is None or self.content is not None:
+            raise ValueError("URL ingestion requires URL and no content")
+        if self.url is not None and (self.url.username or self.url.password or self.url.fragment):
+            raise ValueError("URL must not contain credentials or a fragment")
+        return self
+
+
+class ProcessingResult(Model):
+    version: Literal[1] = 1
+    provider: str
+    connection_id: str
+    dataset_id: str
+    document_id: str
+    state: Literal["queued", "running", "completed", "failed", "cancelled", "unknown"]
+    provider_state: str
+    searchable: bool
+    # Deliberately no made-up per-stage completion, percent, or synthetic task ID.
+
+
+class Capabilities(Model):
+    version: Literal[1] = 1
+    operations: frozenset[Operation]
+    processing_options: frozenset[str]
+    pipeline: dict[str, Literal["provider_managed", "unsupported"]]
+
+
+class ManagementProvider(Protocol):
+    async def capabilities(self, context: ManagementContext, connection: str, base: str) -> Capabilities:
+        raise NotImplementedError
+
+    async def ingest(self, context: ManagementContext, connection: str, base: str, body: IngestRequest) -> ProcessingResult:
+        raise NotImplementedError
+
+    async def upload(self, context: ManagementContext, connection: str, base: str, filename: str, content: bytes, options: ProcessingOptions) -> ProcessingResult:
+        raise NotImplementedError
+
+    async def status(self, context: ManagementContext, connection: str, base: str, document: str) -> ProcessingResult:
+        raise NotImplementedError
+
+    async def process(self, context: ManagementContext, connection: str, base: str, document: str, action: Literal["reprocess", "cancel"], options: ProcessingOptions) -> ProcessingResult:
+        raise NotImplementedError
diff --git a/examples/deerflow-extension-knowledge-management/deerflow_extension_knowledge_management/router.py b/examples/deerflow-extension-knowledge-management/deerflow_extension_knowledge_management/router.py
new file mode 100644
index 0000000000..dc3105d2df
--- /dev/null
+++ b/examples/deerflow-extension-knowledge-management/deerflow_extension_knowledge_management/router.py
@@ -0,0 +1,99 @@
+"""Session-authenticated, bounded HTTP surface independent of provider APIs."""
+
+import logging
+from typing import Literal
+
+from deerflow_extension_api.auth import arequire_plugin_management
+from fastapi import APIRouter, Depends, HTTPException, Request
+from pydantic import ValidationError
+
+from .contracts import IngestRequest, ManagementContext, ManagementCredentialUnavailable, ManagementProvider, ProcessingOptions
+
+logger = logging.getLogger(__name__)
+
+NAMESPACE = "knowledge-management"
+MAX_UPLOAD_BYTES = 16 * 1024 * 1024
+
+
+async def _context(request: Request) -> ManagementContext:
+    try:
+        principal = await arequire_plugin_management(request, NAMESPACE, scope="read" if request.method == "GET" else "write")
+        if not principal.is_admin or principal.is_internal:
+            raise PermissionError
+    except PermissionError:
+        raise HTTPException(403, "Knowledge management access denied") from None
+    return ManagementContext(principal=principal)
+
+
+async def _invoke(awaitable):
+    try:
+        return await awaitable
+    except PermissionError:
+        raise HTTPException(403, "Knowledge resource access denied") from None
+    except FileNotFoundError:
+        raise HTTPException(404, "Knowledge resource unavailable") from None
+    except FileExistsError:
+        raise HTTPException(409, "Knowledge resource conflict; reconcile before retrying") from None
+    except NotImplementedError:
+        raise HTTPException(422, "Unsupported knowledge capability") from None
+    except ValueError:
+        raise HTTPException(422, "Invalid management request") from None
+    except ManagementCredentialUnavailable as exc:
+        logger.warning("Knowledge management local configuration failed: %s", type(exc).__name__)
+        raise HTTPException(503, "Management credential is not configured") from None
+    except RuntimeError as exc:
+        # Exception text/chains can contain provider credentials or source data.
+        logger.warning("Knowledge management provider invocation failed: %s", type(exc).__name__)
+        raise HTTPException(502, "Provider request failed; reconcile state before retrying a write") from None
+
+
+async def _json(request: Request, model):
+    content = bytearray()
+    async for chunk in request.stream():
+        if len(content) + len(chunk) > 2 * 1024 * 1024:
+            raise HTTPException(413, "Management JSON exceeds 2 MiB")
+        content.extend(chunk)
+    try:
+        return model.model_validate_json(content)
+    except ValidationError:
+        raise HTTPException(422, "Invalid management request") from None
+
+
+def build_router(provider: ManagementProvider) -> APIRouter:
+    router = APIRouter(prefix="/api/extensions/knowledge-management", tags=["knowledge-management"])
+    prefix = "/connections/{connection}/bases/{base}"
+
+    @router.get(prefix + "/capabilities")
+    async def capabilities(connection: str, base: str, context: ManagementContext = Depends(_context)):
+        return await _invoke(provider.capabilities(context, connection, base))
+
+    @router.post(prefix + "/documents", status_code=202)
+    async def ingest(request: Request, connection: str, base: str, context: ManagementContext = Depends(_context)):
+        body = await _json(request, IngestRequest)
+        return await _invoke(provider.ingest(context, connection, base, body))
+
+    @router.post(prefix + "/files", status_code=202)
+    async def upload(request: Request, connection: str, base: str, filename: str, context: ManagementContext = Depends(_context)):
+        if request.headers.get("content-type", "").split(";", 1)[0] != "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
+        content = bytearray()
+        async for chunk in request.stream():
+            if len(content) + len(chunk) > MAX_UPLOAD_BYTES:
+                raise HTTPException(413, "Upload exceeds 16 MiB")
+            content.extend(chunk)
+        return await _invoke(provider.upload(context, connection, base, filename, bytes(content), options))
+
+    @router.get(prefix + "/documents/{document}")
+    async def status(connection: str, base: str, document: str, context: ManagementContext = Depends(_context)):
+        return await _invoke(provider.status(context, connection, base, document))
+
+    @router.post(prefix + "/documents/{document}/{action}", status_code=202)
+    async def process(request: Request, connection: str, base: str, document: str, action: Literal["reprocess", "cancel"], context: ManagementContext = Depends(_context)):
+        body = await _json(request, ProcessingOptions)
+        return await _invoke(provider.process(context, connection, base, document, action, body))
+
+    return router
diff --git a/examples/deerflow-extension-knowledge-management/deerflow_extension_knowledge_management/weknora.py b/examples/deerflow-extension-knowledge-management/deerflow_extension_knowledge_management/weknora.py
new file mode 100644
index 0000000000..14e2e2507b
--- /dev/null
+++ b/examples/deerflow-extension-knowledge-management/deerflow_extension_knowledge_management/weknora.py
@@ -0,0 +1,233 @@
+"""WeKnora v0.8.3 ingestion adapter. API mapping informed by superbuzzy's
+community/weknora/documents.py; authorization is explicit and request-local.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import json
+import logging
+import os
+import re
+from typing import Literal
+
+import httpx
+from pydantic import AnyHttpUrl, Field, field_validator, model_validator
+
+from .contracts import Capabilities, IngestRequest, ManagementContext, ManagementCredentialUnavailable, Model, Operation, ProcessingOptions, ProcessingResult
+
+logger = logging.getLogger(__name__)
+
+MAX_UPLOAD_BYTES = 16 * 1024 * 1024
+
+
+def resource_id(value: object) -> str:
+    if not isinstance(value, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,256}", value):
+        raise ValueError("Invalid resource ID")
+    return value
+
+
+class Grant(Model):
+    user_id: str = Field(min_length=1)
+    knowledge_base_ids: tuple[str, ...] = Field(min_length=1)
+    operations: frozenset[Operation] = Field(min_length=1)
+
+    @model_validator(mode="after")
+    def exact(self):
+        if self.user_id != self.user_id.strip():
+            raise ValueError("Use exact host user identities")
+        for base in self.knowledge_base_ids:
+            resource_id(base)
+        return self
+
+
+class Connection(Model):
+    id: str = Field(pattern=r"^[a-zA-Z0-9_-]{1,64}$")
+    base_url: AnyHttpUrl
+    tenant_id: str = Field(pattern=r"^[1-9][0-9]{0,19}$")
+    api_key_env: str = Field(pattern=r"^[A-Za-z_][A-Za-z0-9_]*$")
+    grants: tuple[Grant, ...] = Field(min_length=1)
+
+    @field_validator("base_url")
+    @classmethod
+    def safe_url(cls, value):
+        if value.username is not None or value.password is not None or value.query is not None or value.fragment is not None:
+            raise ValueError("Connection URL must not contain credentials, query or fragment")
+        return value
+
+
+class Settings(Model):
+    connections: tuple[Connection, ...] = Field(min_length=1, max_length=20)
+    timeout: float = Field(default=30, gt=0, le=120, allow_inf_nan=False)
+
+    @model_validator(mode="after")
+    def unique(self):
+        if len({c.id for c in self.connections}) != len(self.connections):
+            raise ValueError("Duplicate connection ID")
+        return self
+
+
+class WeKnoraManagement:
+    def __init__(self, config: dict, *, transport: httpx.AsyncBaseTransport | None = None):
+        self.settings = Settings.model_validate(config)
+        self.transport = transport
+
+    def _authorize(self, context: ManagementContext, connection: str, base: str, operation: Operation) -> Connection:
+        principal = context.principal
+        if not principal.is_admin or principal.is_internal:
+            raise PermissionError("Management requires an explicitly bound administrator")
+        for conn in self.settings.connections:
+            if conn.id == connection:
+                if any(g.user_id == principal.user_id and base in g.knowledge_base_ids and operation in g.operations for g in conn.grants):
+                    resource_id(base)
+                    return conn
+        raise PermissionError("Management operation is not granted for this resource")
+
+    async def _request(self, conn: Connection, method: str, path: str, **kwargs):
+        key = os.environ.get(conn.api_key_env, "").strip()
+        if not key or not key.isascii() or any(ord(c) < 33 or ord(c) == 127 for c in key):
+            raise ManagementCredentialUnavailable("Management credential is unavailable")
+        try:
+            async with await asyncio.to_thread(
+                httpx.AsyncClient,
+                base_url=str(conn.base_url).rstrip("/"),
+                headers={"X-API-Key": key, "X-Tenant-ID": conn.tenant_id},
+                follow_redirects=False,
+                timeout=self.settings.timeout,
+                transport=self.transport,
+            ) as client:
+                async with client.stream(method, path, **kwargs) as response:
+                    if response.status_code in {401, 403}:
+                        raise PermissionError("Provider denied management access")
+                    if response.status_code == 404:
+                        raise FileNotFoundError("Provider resource is unavailable")
+                    if response.status_code == 409:
+                        raise FileExistsError("Provider reported a resource conflict")
+                    if response.status_code not in {200, 201, 202}:
+                        logger.warning("Knowledge management provider request failed: status=%d", response.status_code)
+                        raise RuntimeError("Provider request failed; reconcile state before retrying a write")
+                    raw = bytearray()
+                    async for part in response.aiter_bytes():
+                        if len(raw) + len(part) > 4 * 1024 * 1024:
+                            logger.warning("Knowledge management provider request failed: response_too_large")
+                            raise RuntimeError("Provider response exceeds limit")
+                        raw.extend(part)
+                    payload = json.loads(raw)
+        except (httpx.HTTPError, ValueError) as exc:
+            # Never log exception text, request URLs, response bodies or chains.
+            logger.warning("Knowledge management provider request failed: %s", type(exc).__name__)
+            raise RuntimeError("Provider response unavailable; reconcile state before retrying a write") from None
+        if not isinstance(payload, dict) or payload.get("success") is not True or not isinstance(payload.get("data"), dict):
+            logger.warning("Knowledge management provider request failed: invalid_envelope")
+            raise RuntimeError("Invalid provider response")
+        return payload["data"]
+
+    async def _base(self, conn: Connection, base: str) -> dict:
+        data = await self._request(conn, "GET", f"/api/v1/knowledge-bases/{base}")
+        if data.get("id") != base:
+            raise PermissionError("Provider returned a different knowledge base")
+        return data
+
+    def _result(self, conn: Connection, base: str, data: dict, document: str | None = None) -> ProcessingResult:
+        if data.get("knowledge_base_id") != base or (document is not None and data.get("id") != document):
+            raise PermissionError("Provider returned a document outside the requested resource")
+        try:
+            doc_id = resource_id(data.get("id"))
+        except ValueError:
+            raise RuntimeError("Invalid provider document ID") from None
+        raw = data.get("parse_status")
+        if not isinstance(raw, str) or len(raw) > 64 or not re.fullmatch(r"[a-z_]+", raw):
+            raise RuntimeError("Invalid provider processing state")
+        key = os.environ.get(conn.api_key_env, "").strip()
+        if key and (key in doc_id or key in raw):
+            raise RuntimeError("Invalid provider document metadata")
+        states = {"pending": "queued", "unprocessed": "queued", "processing": "running", "parsing": "running", "finalizing": "running", "completed": "completed", "failed": "failed", "cancelled": "cancelled"}
+        identity = [str(conn.base_url).rstrip("/"), conn.tenant_id, base]
+        dataset = "wk1_" + hashlib.sha256(json.dumps(identity, ensure_ascii=False).encode()).hexdigest()
+        return ProcessingResult(
+            provider="weknora", connection_id=conn.id, dataset_id=dataset, document_id=doc_id, state=states.get(raw, "unknown"), provider_state=raw, searchable=raw in {"completed", "finalizing"} and data.get("enable_status") == "enabled"
+        )
+
+    async def _document(self, conn: Connection, base: str, document: str) -> ProcessingResult:
+        data = await self._request(conn, "GET", f"/api/v1/knowledge/{resource_id(document)}")
+        return self._result(conn, base, data, document)
+
+    @staticmethod
+    def _options(options: ProcessingOptions, base: dict) -> dict:
+        capabilities = base.get("capabilities", {})
+        if options.graph_enabled is True and (not isinstance(capabilities, dict) or capabilities.get("graph") is not True):
+            raise NotImplementedError("Graph extraction is not enabled for this knowledge base")
+        result = options.model_dump(exclude_none=True, exclude={"chunk_size", "chunk_overlap"})
+        if options.chunk_size is not None:
+            existing = base.get("chunking_config")
+            if not isinstance(existing, dict):
+                raise RuntimeError("Invalid provider chunking configuration")
+            if existing.get("enable_parent_child") is True:
+                raise NotImplementedError("Chunk overrides are unsupported in parent-child mode")
+            # v0.8.3 merges nonzero integers but treats enable_parent_child as
+            # authoritative even when omitted. Preserve its full base snapshot.
+            if options.chunk_overlap == 0 and existing.get("chunk_overlap") != 0:
+                raise NotImplementedError("This provider cannot override nonzero overlap with zero")
+            result["chunking_config"] = {**existing, "chunk_size": options.chunk_size, "chunk_overlap": options.chunk_overlap}
+        return result
+
+    async def capabilities(self, context: ManagementContext, connection: str, base: str) -> Capabilities:
+        conn = self._authorize(context, connection, base, "capabilities")
+        data = await self._base(conn, base)
+        flags = data.get("capabilities")
+        if not isinstance(flags, dict):
+            raise RuntimeError("Invalid provider capabilities")
+        operations = frozenset(op for g in conn.grants if g.user_id == context.principal.user_id and base in g.knowledge_base_ids for op in g.operations)
+        pipeline = {
+            "parse": "provider_managed",
+            "chunk": "provider_managed",
+            "vector": "provider_managed" if flags.get("vector") is True else "unsupported",
+            "entries": "provider_managed" if flags.get("wiki") is True else "unsupported",
+            "entity_relations": "provider_managed" if flags.get("graph") is True else "unsupported",
+        }
+        options = {"summary_enabled"}
+        if flags.get("graph") is True:
+            options.add("graph_enabled")
+        chunking = data.get("chunking_config")
+        if isinstance(chunking, dict) and chunking.get("enable_parent_child") is not True:
+            options.update({"chunk_size", "chunk_overlap"})
+        return Capabilities(operations=operations, processing_options=frozenset(options), pipeline=pipeline)
+
+    async def ingest(self, context: ManagementContext, connection: str, base: str, body: IngestRequest) -> ProcessingResult:
+        conn = self._authorize(context, connection, base, body.kind)
+        config = self._options(body.processing, await self._base(conn, base))
+        payload = {"title": body.title, "process_config": config}
+        if body.kind == "manual":
+            payload.update(content=body.content, status="publish")
+        else:
+            payload["url"] = str(body.url)
+        data = await self._request(conn, "POST", f"/api/v1/knowledge-bases/{base}/knowledge/{body.kind}", json=payload)
+        return self._result(conn, base, data)
+
+    async def upload(self, context: ManagementContext, connection: str, base: str, filename: str, content: bytes, options: ProcessingOptions) -> ProcessingResult:
+        conn = self._authorize(context, connection, base, "upload")
+        if not content or len(content) > MAX_UPLOAD_BYTES or not filename or len(filename) > 255 or any(c in '/\\"' or ord(c) < 32 or ord(c) == 127 for c in filename) or filename in {".", ".."}:
+            raise ValueError("Invalid file name or upload size")
+        config = self._options(options, await self._base(conn, base))
+        data = await self._request(conn, "POST", f"/api/v1/knowledge-bases/{base}/knowledge/file", files={"file": (filename, content, "application/octet-stream")}, data={"process_config": json.dumps(config)})
+        return self._result(conn, base, data)
+
+    async def status(self, context: ManagementContext, connection: str, base: str, document: str) -> ProcessingResult:
+        conn = self._authorize(context, connection, base, "status")
+        await self._base(conn, base)
+        return await self._document(conn, base, document)
+
+    async def process(self, context: ManagementContext, connection: str, base: str, document: str, action: Literal["reprocess", "cancel"], options: ProcessingOptions) -> ProcessingResult:
+        if action not in {"reprocess", "cancel"}:
+            raise NotImplementedError("Unsupported processing operation")
+        conn = self._authorize(context, connection, base, action)
+        if action == "cancel" and options.model_fields_set:
+            raise ValueError("Cancel does not accept processing options")
+        config = self._options(options, await self._base(conn, base))
+        await self._document(conn, base, document)
+        suffix = "reparse" if action == "reprocess" else "cancel-parse"
+        # Omit empty overrides: upstream retains the original processing policy.
+        kwargs = {"json": {"process_config": config}} if config else {}
+        data = await self._request(conn, "POST", f"/api/v1/knowledge/{document}/{suffix}", **kwargs)
+        return self._result(conn, base, data, document)
diff --git a/examples/deerflow-extension-knowledge-management/pyproject.toml b/examples/deerflow-extension-knowledge-management/pyproject.toml
new file mode 100644
index 0000000000..6e0144ec1d
--- /dev/null
+++ b/examples/deerflow-extension-knowledge-management/pyproject.toml
@@ -0,0 +1,25 @@
+[project]
+name = "deerflow-extension-knowledge-management"
+version = "0.1.0"
+description = "Optional provider-owned knowledge ingestion and processing for DeerFlow"
+readme = "README.md"
+requires-python = ">=3.12"
+license = "MIT"
+dependencies = ["deerflow-extension-api>=0.2.7,<0.3", "fastapi>=0.115,<1", "httpx>=0.27,<1", "pydantic>=2,<3"]
+
+[project.entry-points."deerflow.extensions"]
+knowledge-management = "deerflow_extension_knowledge_management:install"
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["deerflow_extension_knowledge_management"]
+
+[tool.ruff]
+line-length = 240
+target-version = "py312"
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "UP"]
diff --git a/frontend/src/core/extensions/catalog.ts b/frontend/src/core/extensions/catalog.ts
index 5630c03b0a..fa1d3c8056 100644
--- a/frontend/src/core/extensions/catalog.ts
+++ b/frontend/src/core/extensions/catalog.ts
@@ -45,6 +45,11 @@ export const extensionCatalog = [
     package: "deerflow-extension-knowledge",
     namespace: null,
   },
+  {
+    key: "knowledgeManagement",
+    package: "deerflow-extension-knowledge-management",
+    namespace: "knowledge-management",
+  },
   { key: "lightrag", package: "deerflow-extension-lightrag", namespace: null },
   { key: "weknora", package: "deerflow-extension-weknora", namespace: null },
   { key: "example", package: "deerflow-extension-example", namespace: null },
diff --git a/frontend/src/core/i18n/locales/en-US.ts b/frontend/src/core/i18n/locales/en-US.ts
index d4e2099dab..562d5f06f3 100644
--- a/frontend/src/core/i18n/locales/en-US.ts
+++ b/frontend/src/core/i18n/locales/en-US.ts
@@ -85,6 +85,11 @@ export const enUS: Translations = {
         description:
           "Retrieve documents and entity relations from LightRAG with shared source citations.",
       },
+      knowledgeManagement: {
+        title: "Knowledge ingestion and processing",
+        description:
+          "Ingest files, URLs and text through WeKnora and manage processing with explicit administrator grants. Requires installation and configuration; no management page is included.",
+      },
       weknora: {
         title: "WeKnora knowledge retrieval",
         description:
diff --git a/frontend/src/core/i18n/locales/types.ts b/frontend/src/core/i18n/locales/types.ts
index 63777579f6..8c93d73e1f 100644
--- a/frontend/src/core/i18n/locales/types.ts
+++ b/frontend/src/core/i18n/locales/types.ts
@@ -35,6 +35,7 @@ export interface Translations {
       | "screening"
       | "retrieval"
       | "knowledge"
+      | "knowledgeManagement"
       | "weknora"
       | "lightrag"
       | "example",
diff --git a/frontend/src/core/i18n/locales/zh-CN.ts b/frontend/src/core/i18n/locales/zh-CN.ts
index b160c4424c..5d44688b7e 100644
--- a/frontend/src/core/i18n/locales/zh-CN.ts
+++ b/frontend/src/core/i18n/locales/zh-CN.ts
@@ -80,6 +80,11 @@ export const zhCN: Translations = {
         title: "LightRAG",
         description: "从 LightRAG 检索文档与实体关系，并使用统一的来源引用。",
       },
+      knowledgeManagement: {
+        title: "知识入库与处理",
+        description:
+          "通过 WeKnora 导入文件、网址和文本，并在管理员明确授权后管理处理任务。需要安装和配置，不包含管理页面。",
+      },
       weknora: {
         title: "WeKnora 知识检索",
         description:
diff --git a/frontend/tests/e2e/AGENTS.md b/frontend/tests/e2e/AGENTS.md
index b62b8169fd..f2addd25d1 100644
--- a/frontend/tests/e2e/AGENTS.md
+++ b/frontend/tests/e2e/AGENTS.md
@@ -8,3 +8,7 @@ For message ordering across compaction/refetch, poll an atomic DOM comparison
 until both nodes settle. Missing or reversed nodes must still fail; keep history,
 refresh and hidden-summary assertions. Avoid fixed sleeps and whole-test retries
 as substitutes for synchronization.
+
+Extension catalog counts follow extensionCatalog.length; the unit completeness
+check independently enumerates packages. Keep localized management discovery
+assertions so adding a package cannot silently remove its visible entry.
diff --git a/frontend/tests/e2e/bookmark-plugin.spec.ts b/frontend/tests/e2e/bookmark-plugin.spec.ts
index 226c2640ca..ba1522de38 100644
--- a/frontend/tests/e2e/bookmark-plugin.spec.ts
+++ b/frontend/tests/e2e/bookmark-plugin.spec.ts
@@ -6,6 +6,8 @@ import path from "node:path";
 
 import { expect, test } from "@playwright/test";
 
+import { extensionCatalog } from "../../src/core/extensions/catalog";
+
 import { mockLangGraphAPI, MOCK_THREAD_ID } from "./utils/mock-api";
 
 let gateway: ChildProcess;
@@ -448,7 +450,18 @@ for (const locale of ["en-US", "zh-CN"]) {
         exact: true,
       }),
     ).toBeVisible();
-    await expect(page.getByRole("article")).toHaveCount(11);
+    await expect(page.getByRole("article")).toHaveCount(
+      extensionCatalog.length,
+    );
+    const management = page.getByRole("article").filter({
+      hasText: zh ? "知识入库与处理" : "Knowledge ingestion and processing",
+    });
+    await expect(management).toHaveCount(1);
+    await expect(management).toContainText(
+      zh
+        ? "需要安装和配置，不包含管理页面。"
+        : "Requires installation and configuration; no management page is included.",
+    );
     const lightrag = page.getByRole("article").filter({ hasText: "LightRAG" });
     await expect(lightrag).toHaveCount(1);
     await expect(lightrag).toContainText(
@@ -672,10 +685,10 @@ test("catalog remains discoverable when runtime discovery fails", async ({
   await expect(
     page.getByRole("alert").filter({ hasText: "Extensions unavailable." }),
   ).toBeVisible();
-  await expect(page.getByRole("article")).toHaveCount(11);
+  await expect(page.getByRole("article")).toHaveCount(extensionCatalog.length);
   await expect(
     page.getByRole("article").filter({ hasText: "Catalog extension" }),
-  ).toHaveCount(11);
+  ).toHaveCount(extensionCatalog.length);
   await expect(
     page.getByRole("article").filter({ hasText: "Script workflows" }),
   ).toHaveCount(1);
diff --git a/frontend/tests/unit/core/extensions/catalog.test.ts b/frontend/tests/unit/core/extensions/catalog.test.ts
index 8941312ee2..5f47c74a51 100644
--- a/frontend/tests/unit/core/extensions/catalog.test.ts
+++ b/frontend/tests/unit/core/extensions/catalog.test.ts
@@ -175,3 +175,23 @@ test("LightRAG discovery is localized without claiming a runtime registration",
     expect(entry?.loaded).toBeUndefined();
   }
 });
+
+test("knowledge management discovery merges its namespace without implying activation", () => {
+  const management = { ...contribution, namespace: "knowledge-management" };
+  for (const [copy, title] of [
+    [enUS.extensions.catalog, "Knowledge ingestion and processing"],
+    [zhCN.extensions.catalog, "知识入库与处理"],
+  ] as const) {
+    const entry = extensionDirectory([], copy).find(
+      (item) => item.package === "deerflow-extension-knowledge-management",
+    );
+    expect(entry?.id).toBe(management.namespace);
+    expect(entry?.title).toBe(title);
+    expect(entry?.loaded).toBeUndefined();
+    const matches = extensionDirectory([management], copy).filter(
+      (item) => item.id === management.namespace,
+    );
+    expect(matches).toHaveLength(1);
+    expect(matches[0]?.loaded).toBe(management);
+  }
+});