diff --git a/wwwroot/api/kb_diag.dspy b/wwwroot/api/kb_diag.dspy
new file mode 100644
index 0000000..ff19f5f
--- /dev/null
+++ b/wwwroot/api/kb_diag.dspy
@@ -0,0 +1,34 @@
+# kb_diag.dspy - 临时诊断端点：真实请求上下文验证 rag_client 全链路
+# 用后即删（不留生产痕迹）。验证点：owner解析→dapi发key→HTTP检索
+import json as _j
+
+uid = await get_user()
+if not uid:
+    return _j.dumps({"ok": False, "error": "未登录"}, ensure_ascii=False)
+
+project_id = (params_kw or {}).get('project_id', '').strip()
+query = (params_kw or {}).get('query', '').strip() or '测试入库闭环'
+if not project_id:
+    return _j.dumps({"ok": False, "error": "缺少 project_id"}, ensure_ascii=False)
+
+out = {"steps": []}
+try:
+    from pipeline_service.rag_client import (resolve_project_owner, get_owner_apikey,
+                                             tool_rag_kb_list, tool_rag_search)
+    owner, err = await resolve_project_owner(project_id)
+    out["steps"].append({"step": "resolve_owner", "owner": owner, "err": err})
+    if err:
+        return _j.dumps({"ok": False, "out": out}, ensure_ascii=False, default=str)
+    key, err = await get_owner_apikey(owner)
+    out["steps"].append({"step": "get_key", "key_prefix": (key or "")[:8],
+                         "key_len": len(key or ""), "err": err})
+    if err:
+        return _j.dumps({"ok": False, "out": out}, ensure_ascii=False, default=str)
+    out["kb_list"] = await tool_rag_kb_list(project_id)
+    out["search"] = (await tool_rag_search(project_id, query))[:600]
+    out["ok"] = True
+except Exception as e:
+    import traceback
+    out["ok"] = False
+    out["traceback"] = traceback.format_exc()[-800:]
+return _j.dumps(out, ensure_ascii=False, default=str)