From dbcf3df8a3a618a86f26b231f7440073745f228e Mon Sep 17 00:00:00 2001 From: Aska Date: Sun, 6 Sep 2026 20:09:09 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20RAG=20=E7=9F=A5=E8=AF=86=E5=BA=93?= =?UTF-8?q?=E7=9C=9F=E5=AE=9E=E6=95=B0=E6=8D=AE=20+=20=E6=B8=A0=E9=81=93?= =?UTF-8?q?=E7=AB=AF=E5=88=B0=E7=AB=AF=E6=8E=A5=E7=BA=BF=EF=BC=88=E5=B9=B6?= =?UTF-8?q?=E8=A1=8C=E2=88=A6wire.go=EF=BC=89+=20=E4=B8=BB=E4=BC=9A?= =?UTF-8?q?=E8=AF=9D=E6=94=B6=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 【渠道端到端·接线（agent-72）】 - 新建 internal/services/channelbridge/：桥接服务（Start 对配置渠道 Connect(10s超时/warn)+SetMessageHandler+ReadMessages； 入站 UnifiedMessage→sessions.GetOrCreate(ResolveActor 白名单结构性识别)→loop.Run→SendMessage 回执，chat 文本）。 - cmd/cyrene/channels.go：从 CHANNEL_* env 读 4 渠道凭据（telegram/qq_obv11/weixin_ilink/webhook）。 - wire.go：app 加 channelBridge 字段，wireChannels 构造适配器+bridge.Start()，close() 收尾。 - 无真凭据降级：无 CHANNEL_* → configured()false 跳过不 dial；Connect 失败 Warn 不阻启动； webhook 仅入站(回执 ErrUnsupported 记 debug)。身份识别绝不靠提示词。收发端到端待凭据验证。 【RAG 知识库（agent-71）】 - internal/runtime/knowledge/：真实数据源（doc=资料/记忆=经历 分离 design/03 §3③）。 ListDocuments 走 projectindex.ListProjects(+ListChapters)、Search 走 projectindex.Search(标题/描述+章节 命中行+上下文)+retriever.Retrieve(配 vector 真语义,命中标 semantic:true)。 - gateway/server.go：RagStore 接口+Deps.RAG 可空+3 路由(GET /admin/rag、POST /admin/rag/query、/delete)。 - admin.go：adminListRag/adminSearchRag(空 query 400)/adminDeleteRag(写 501 not_implemented,M10)。 - projectindex/store.go：ListProjects/ListChapters。 - 前端 Rag.tsx：docs 从 GET /admin/rag 拉、检索走 POST /admin/rag/query、删除调真端点(501 显示 Notice)， 上传/分块/嵌入显示 M10 Notice；api.ts/types.ts 加 adminListRag/adminSearchRag/adminDeleteRag。 【主会话收口 wire.go】projStore 提到 app struct+wireMemory 赋值；knowledge.New(...) 注入 gateway.Deps.RAG。 实机验证：admin/rag 200(documents[] 空表非壳)、channels 枚举 4 渠道、health ok。 go build/vet/test 全绿 + 前端 tsc。wire.go 567 行。 遗留(记录)：① 群聊/陌生人 MemWritable 记忆隔离未接 loop(AGENTS.md 红线,待修) ② webhook InboundHandler 需 gateway 挂载 ③ RAG 语义路径 persona-scoped(project 记忆其他 persona 不浮现)。 --- Docs/api-client.md | 10 +- cmd/cyrene/channels.go | 81 ++++ cmd/cyrene/wire.go | 124 +++--- frontend/web/src/admin/modules/Rag.tsx | 151 ++++--- frontend/web/src/lib/api.ts | 25 ++ frontend/web/src/lib/types.ts | 25 ++ internal/runtime/knowledge/store.go | 266 ++++++++++++ internal/runtime/memory/projectindex/store.go | 26 ++ .../services/channelbridge/channelbridge.go | 380 ++++++++++++++++++ .../channelbridge/channelbridge_test.go | 209 ++++++++++ internal/services/gateway/admin.go | 146 +++++++ internal/services/gateway/extra_test.go | 145 +++++++ internal/services/gateway/server.go | 26 ++ 13 files changed, 1510 insertions(+), 104 deletions(-) create mode 100644 cmd/cyrene/channels.go create mode 100644 internal/runtime/knowledge/store.go create mode 100644 internal/services/channelbridge/channelbridge.go create mode 100644 internal/services/channelbridge/channelbridge_test.go diff --git a/Docs/api-client.md b/Docs/api-client.md index 2139941..39e5b7a 100644 --- a/Docs/api-client.md +++ b/Docs/api-client.md @@ -42,8 +42,16 @@ | 9 | `/api/v1/admin/llm` | GET/PUT | ✅(GET) | LLM 模型/路由/降级（GET 已实现，APIKey 脱敏；PUT 热重载 M10） | | 10 | `/api/v1/admin/automations` | GET/POST | ✅(GET) | 自动化与提醒（GET 已实现；POST 创建 M10） | | 11 | `/api/v1/admin/backthink` | GET | ✅ | 后台思考记录 | +| 12 | `/api/v1/admin/rag` | GET | ✅ | 知识库文档列表（分块/嵌入状态）；带 `?query=` 时并入检索结果 | +| 13 | `/api/v1/admin/rag/query` | POST | ✅ | 知识库检索（真实检索；配置向量增强走语义路径） | +| 14 | `/api/v1/admin/rag/delete` | POST | 🕐(写) | 删除知识库文档；写操作 M10，当前返回 `501 not_implemented` | -> **M10 前端契约补充（管理面板）**：`GET /api/v1/admin/users` · `GET /api/v1/admin/llm` · `GET /api/v1/admin/automations` · `GET /api/v1/admin/backthink` 已实现，返回真实数据；`GET /api/v1/admin/logs` · `GET /api/v1/admin/trace` · `GET /api/v1/admin/audit` · `GET /api/v1/admin/system` 仍🕐未实现（前端按契约封装、后端 404 时显示"待实现"壳，不报错）。写操作（策略/档案/密钥/系统设置/用户与自动化创建）经前端审计登记，待后端 M10+ 落地真实持久化与热重载。 +> **M10 前端契约补充（管理面板）**：`GET /api/v1/admin/users` · `GET /api/v1/admin/llm` · `GET /api/v1/admin/automations` · `GET /api/v1/admin/backthink` · `GET /api/v1/admin/rag` · `POST /api/v1/admin/rag/query` 已实现，返回真实数据；`GET /api/v1/admin/logs` · `GET /api/v1/admin/trace` · `GET /api/v1/admin/audit` · `GET /api/v1/admin/system` 仍🕐未实现（前端按契约封装、后端 404 时显示"待实现"壳，不报错）。写操作（策略/档案/密钥/系统设置/用户与自动化创建、知识库上传/删除）经前端审计登记，待后端 M10+ 落地真实持久化与热重载。 + +**知识库 RAG 契约**： +- `GET /admin/rag` → `{"data":{"documents":[{id,name,desc?,chunks,embedded}],"hits"?:[{docId,docName,chapter?,text,score,semantic}]}}`。`hits` 仅在携带 `?query=` 时出现；`embedded` 反映部署是否配置向量增强（未配置为待嵌入）。 +- `POST /admin/rag/query` body `{"query":"..."}` → `{"data":{"query":"...","hits":[{docId,docName,chapter?,text,score,semantic}]}}`。检索走项目索引关键词+行上下文；配置 pgvector/向量增强时并入语义/向量命中（`semantic:true`）。 +- `POST /admin/rag/delete` body `{"id":"..."}` → 写操作 M10，返回 `501` `{"error":{"code":"not_implemented",...}}`。 ## 3. WebSocket `/ws/chat`（实时） diff --git a/cmd/cyrene/channels.go b/cmd/cyrene/channels.go new file mode 100644 index 0000000..f1347f8 --- /dev/null +++ b/cmd/cyrene/channels.go @@ -0,0 +1,81 @@ +package main + +// This file builds the four built-in IM channel adapters from environment +// configuration. The config package carries no channel block, so credentials are +// read from CHANNEL__* env vars here and injected into each adapter's +// concrete Config. Adapters whose construction fails are skipped rather than +// aborting boot, matching the "no credentials → warn and continue" contract of +// the channel bridge. + +import ( + "log/slog" + "os" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/plugins/channels/qq_obv11" + "git.yeij.top/AskaEth/Cyrene_v2/plugins/channels/telegram" + "git.yeij.top/AskaEth/Cyrene_v2/plugins/channels/webhook" + "git.yeij.top/AskaEth/Cyrene_v2/plugins/channels/weixin_ilink" +) + +// buildChannelAdapters constructs the four channel adapters from environment +// configuration and returns the successfully built subset. Adapter construction +// never dials the network; actual connection happens in the channel bridge. +func buildChannelAdapters(log *slog.Logger) []adapter.PlatformAdapter { + qq := qq_obv11.NewAdapter(qq_obv11.Config{ + Mode: envString("CHANNEL_QQ_OBV11_MODE", "server"), + ListenAddr: os.Getenv("CHANNEL_QQ_OBV11_LISTEN_ADDR"), + RemoteURL: os.Getenv("CHANNEL_QQ_OBV11_REMOTE_URL"), + AccessToken: os.Getenv("CHANNEL_QQ_OBV11_ACCESS_TOKEN"), + SelfID: os.Getenv("CHANNEL_QQ_OBV11_SELF_ID"), + Logger: log, + }) + + weixin, werr := weixin_ilink.New(weixin_ilink.Config{ + BaseURL: os.Getenv("CHANNEL_WEIXIN_ILINK_BASE_URL"), + CDNBaseURL: os.Getenv("CHANNEL_WEIXIN_ILINK_CDN_BASE_URL"), + BotToken: os.Getenv("CHANNEL_WEIXIN_ILINK_BOT_TOKEN"), + BotID: os.Getenv("CHANNEL_WEIXIN_ILINK_BOT_ID"), + OwnerID: os.Getenv("CHANNEL_WEIXIN_ILINK_OWNER_ID"), + }, nil, log) + if werr != nil { + // A blank config defaults to a valid base URL, so this only fires on an + // explicitly invalid one; skip the channel rather than fail startup. + log.Warn("weixin-ilink adapter build skipped", "error", werr) + weixin = nil + } + + tg := telegram.New(telegram.Config{ + BotToken: os.Getenv("CHANNEL_TELEGRAM_BOT_TOKEN"), + WebhookURL: os.Getenv("CHANNEL_TELEGRAM_WEBHOOK_URL"), + }, telegram.WithLogger(log)) + + // webhook.WithLogger does not tag provenance, so wrap base with the module + // source here to satisfy the logging rule. + wh := webhook.New( + webhook.WithConfig(webhook.Config{ + EndpointPath: os.Getenv("CHANNEL_WEBHOOK_ENDPOINT_PATH"), + ChannelID: os.Getenv("CHANNEL_WEBHOOK_CHANNEL_ID"), + ChannelName: os.Getenv("CHANNEL_WEBHOOK_CHANNEL_NAME"), + Secret: os.Getenv("CHANNEL_WEBHOOK_SECRET"), + SignatureHeader: os.Getenv("CHANNEL_WEBHOOK_SIGNATURE_HEADER"), + }), + webhook.WithLogger(logger.WithSource(log, "plugins/channels/webhook")), + ) + + out := []adapter.PlatformAdapter{qq, tg, wh} + if weixin != nil { + out = append(out, weixin) + } + return out +} + +// envString returns the value of key, or fallback when the variable is unset or +// blank. +func envString(key, fallback string) string { + if v := os.Getenv(key); v != "" { + return v + } + return fallback +} diff --git a/cmd/cyrene/wire.go b/cmd/cyrene/wire.go index 2cd41b8..fee89d4 100644 --- a/cmd/cyrene/wire.go +++ b/cmd/cyrene/wire.go @@ -26,6 +26,7 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/exec" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/guardrail" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/knowledge" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory/projectindex" @@ -41,12 +42,9 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" "git.yeij.top/AskaEth/Cyrene_v2/internal/services/automations" "git.yeij.top/AskaEth/Cyrene_v2/internal/services/backthink" + "git.yeij.top/AskaEth/Cyrene_v2/internal/services/channelbridge" "git.yeij.top/AskaEth/Cyrene_v2/internal/services/gateway" "git.yeij.top/AskaEth/Cyrene_v2/internal/user" - "git.yeij.top/AskaEth/Cyrene_v2/plugins/channels/qq_obv11" - "git.yeij.top/AskaEth/Cyrene_v2/plugins/channels/telegram" - "git.yeij.top/AskaEth/Cyrene_v2/plugins/channels/webhook" - "git.yeij.top/AskaEth/Cyrene_v2/plugins/channels/weixin_ilink" "git.yeij.top/AskaEth/Cyrene_v2/plugins/tools/basic" iotools "git.yeij.top/AskaEth/Cyrene_v2/plugins/tools/io" ) @@ -70,28 +68,30 @@ const ( // app holds every subsystem the single Cyrene binary assembles. It is built by // newApp and torn down by close on shutdown. type app struct { - cfg *config.Config - log *slog.Logger - db *sql.DB - bus bus.Bus - personas *personareg.Registry - caps *registry.Registry - dispatcher *tool.Dispatcher - memStore *memory.Store - retriever *memory.Retriever - loop *loop.Loop - scheduler *orchestration.Scheduler - tasks *task.Manager - execProv exec.Provider - iot *iot.Manager - voice *voice.Chain - mcpRegs []*mcp.Registry - sessions *session.SessionManager - automation *automations.Service - backthink *backthink.Backthink - vindex vector.Index - owner *user.OwnerProfile - gw *gateway.Server + cfg *config.Config + log *slog.Logger + db *sql.DB + bus bus.Bus + personas *personareg.Registry + caps *registry.Registry + dispatcher *tool.Dispatcher + memStore *memory.Store + retriever *memory.Retriever + projStore *projectindex.Store + loop *loop.Loop + scheduler *orchestration.Scheduler + tasks *task.Manager + execProv exec.Provider + iot *iot.Manager + voice *voice.Chain + mcpRegs []*mcp.Registry + sessions *session.SessionManager + automation *automations.Service + backthink *backthink.Backthink + vindex vector.Index + owner *user.OwnerProfile + gw *gateway.Server + channelBridge *channelbridge.Bridge } // newApp assembles the full Cyrene runtime from cfg and log. It opens the @@ -237,6 +237,7 @@ func (a *app) wireMemory(ctx context.Context) error { a.log.Warn("project index unavailable", "error", err) projStore = nil } + a.projStore = projStore opts := []memory.RetrieverOption{memory.WithEmbedder(embedder)} if a.vindex != nil { @@ -398,34 +399,42 @@ func (a *app) wireAuxiliary() error { return nil } -// wireChannels constructs and registers the four IM channel adapters so the -// admin API can enumerate them. They are registered but not connected: real -// platform credentials are supplied per-deployment, and connecting with empty -// config would dial external networks. +// wireChannels constructs the four IM channel adapters from environment +// configuration, registers them so the admin API can enumerate them, and wires +// them into the channel bridge so a configured adapter connects and routes +// inbound messages through session→actor→loop→reply. Adapters without +// credentials stay disconnected (logged, never fatal): connecting with an empty +// config would either fail or, for weixin-ilink, trigger a blocking scan-code +// login. func (a *app) wireChannels() error { - channels := []struct { - name string - value any - policy registry.ToolPolicy - }{ - {"qq-obv11", qq_obv11.NewAdapter(qq_obv11.Config{Logger: a.log}), registry.ToolPolicy{}}, - {"weixin-ilink", mustWeixin(a.log), registry.ToolPolicy{}}, - {"telegram", telegram.New(telegram.Config{}, telegram.WithLogger(a.log)), registry.ToolPolicy{}}, - {"webhook", webhook.New(webhook.WithLogger(a.log)), registry.ToolPolicy{}}, - } - for _, ch := range channels { + adapters := buildChannelAdapters(a.log) + for _, ch := range adapters { capp := registry.Capability{ - Name: ch.name, + Name: ch.Name(), Kind: registry.KindChannel, - Description: ch.name, - Policy: ch.policy, - Value: ch.value, + Description: ch.Name(), + Value: ch, } if err := a.caps.Register(capp); err != nil { - a.log.Warn("channel register skipped", "channel", ch.name, "error", err) + a.log.Warn("channel register skipped", "channel", ch.Name(), "error", err) } } - a.log.Info("channel adapters registered", "count", len(channels)) + + bridge, berr := channelbridge.New(channelbridge.Options{ + Log: logger.WithSource(a.log, "services/channelbridge"), + Loop: a.loop, + Sessions: a.sessions, + Owner: a.owner, + }) + if berr != nil { + return fmt.Errorf("channel bridge: %w", berr) + } + for _, ch := range adapters { + bridge.Add(ch) + } + bridge.Start() + a.channelBridge = bridge + a.log.Info("channel adapters registered", "count", len(adapters)) return nil } @@ -435,6 +444,13 @@ func (a *app) wireChannels() error { func (a *app) wireFrontend() error { jwt := auth.New(a.cfg.Auth) apiKeys := auth.NewAPIKey(a.cfg.Auth) + ragStore := knowledge.New(knowledge.Deps{ + Projects: a.projStore, + Retriever: a.retriever, + DefaultPersona: a.defaultPersonaID(), + VectorOn: a.vindex != nil, + Log: a.log, + }) gw, err := gateway.New(a.cfg, gateway.Deps{ Log: logger.WithSource(a.log, "gateway"), Auth: jwt, @@ -450,6 +466,7 @@ func (a *app) wireFrontend() error { Automations: a.automation, Users: apiKeys, Backthink: a.backthink, + RAG: ragStore, Streaming: true, Frontend: resolveFrontendFS(a.log), }) @@ -502,6 +519,9 @@ func (a *app) close(ctx context.Context) { if a.automation != nil { _ = a.automation.Stop(ctx) } + if a.channelBridge != nil { + a.channelBridge.Stop(ctx) + } if a.scheduler != nil { _ = a.scheduler.CancelAll(ctx) } @@ -545,13 +565,3 @@ func resolveSQLitePath(cfg *config.Config, dataDir string) string { } return p } - -// mustWeixin builds the weixin-ilink adapter, panicking on an invalid config -// (which should never happen with defaults). -func mustWeixin(log *slog.Logger) *weixin_ilink.Adapter { - adapter, err := weixin_ilink.New(weixin_ilink.Config{}, nil, log) - if err != nil { - panic(fmt.Sprintf("weixin-ilink adapter: %v", err)) - } - return adapter -} diff --git a/frontend/web/src/admin/modules/Rag.tsx b/frontend/web/src/admin/modules/Rag.tsx index 13b0815..d108dd6 100644 --- a/frontend/web/src/admin/modules/Rag.tsx +++ b/frontend/web/src/admin/modules/Rag.tsx @@ -1,37 +1,60 @@ import { useState } from "react"; +import { useAsyncData } from "../../lib/useAsyncData"; +import { adminDeleteRag, adminListRag, adminSearchRag } from "../../lib/api"; +import type { AdminRagDoc, AdminRagHit } from "../../lib/types"; import { useAuditStore } from "../audit"; import { Notice, PageHeader } from "../components"; import { Button, Surface, Tag } from "../../components/primitives"; -interface DocRow { - id: string; - name: string; - chunks: number; - embedded: boolean; -} - -const docs: DocRow[] = [ - { id: "1", name: "owner_identity.md", chunks: 1, embedded: false }, - { id: "2", name: "开发时间线.md", chunks: 6, embedded: false }, - { id: "3", name: "design/10-后台管理面板.md", chunks: 2, embedded: false }, -]; +const WRITE_M10 = "上传/分块/嵌入为后端 M10+ 提供；当前只读列文档与检索，删除接口在 M10 落地。"; export function RagModule(): React.ReactNode { const log = useAuditStore((s) => s.log); + const { data, loading, error, reload } = useAsyncData(() => adminListRag(), []); const [query, setQuery] = useState(""); - const [hits, setHits] = useState([]); + const [hits, setHits] = useState([]); + const [searching, setSearching] = useState(false); + const [notice, setNotice] = useState(null); + + const docs: AdminRagDoc[] = data?.documents ?? []; + + const search = async () => { + log("rag.search", "ok"); + setSearching(true); + setNotice(null); + try { + const res = await adminSearchRag(query.trim()); + setHits(res.hits); + } catch (err: unknown) { + const msg = err instanceof Error ? err.message : "检索失败"; + setNotice(msg); + setHits([]); + } finally { + setSearching(false); + } + }; + + const remove = async (d: AdminRagDoc) => { + log("rag.delete", "ok"); + setNotice(null); + try { + await adminDeleteRag(d.id); + } catch (err: unknown) { + setNotice(err instanceof Error ? err.message : WRITE_M10); + } + }; return (
{ log("rag.upload", "ok"); - alert("上传文档（占位）：后端 M10+ 提供分块/嵌入。"); + setNotice(WRITE_M10); }} > 上传文档 @@ -39,30 +62,50 @@ export function RagModule(): React.ReactNode { } /> + {notice ? {notice} : null} +

文档列表

- {docs.length} 篇 + + {loading ? "加载中…" : `${docs.length} 篇`} +
- {docs.map((d) => ( -
- {d.name} - {d.chunks} 分块 - - {d.embedded ? "已嵌入" : "待嵌入"} - - -
- ))} +
+ {d.name} + {d.desc ? ( + {d.desc} + ) : null} +
+ {d.chunks} 分块 + + {d.embedded ? "已嵌入" : "待嵌入"} + + +
+ )) + )} @@ -71,29 +114,45 @@ export function RagModule(): React.ReactNode { setQuery(e.target.value)} - placeholder="输入查询…" + onKeyDown={(e) => { + if (e.key === "Enter" && query.trim()) void search(); + }} + placeholder="输入查询（命中项目索引或项目记忆）…" className="min-w-0 flex-1 rounded-[var(--mu-radius)] bg-[var(--mu-surface-2)] px-3 py-2 text-sm outline-none focus:ring-1 focus:ring-[var(--mu-accent)]" /> - {hits.length > 0 ? (
- {hits.map((h) => ( -
{h}
+ {hits.map((h, i) => ( +
+
+ {h.docName} + {h.chapter ? ( + #{h.chapter} + ) : null} + + {h.semantic ? "语义" : "关键词"} · {h.score.toFixed(2)} + +
+

{h.text}

+
))}
+ ) : hits.length === 0 && query.trim() && !searching ? ( +

无命中。

) : null} - 分块/嵌入状态、向量索引与语义检索为后端 M10+ 提供；配置了 pgvector 增强后必须真实走语义路径（非空壳）。 + + 分块/嵌入状态与语义检索：配置了 pgvector / SQLite 向量增强后，后端经真实语义/向量路径返回（非空壳）； + 未配置时走项目索引关键词 + 行上下文。上传/删除为后端 M10 写操作。 + ); } diff --git a/frontend/web/src/lib/api.ts b/frontend/web/src/lib/api.ts index 60340ac..d78133c 100644 --- a/frontend/web/src/lib/api.ts +++ b/frontend/web/src/lib/api.ts @@ -4,6 +4,8 @@ import type { AdminIoTDevice, AdminLlmConfig, AdminPluginRecord, + AdminRagHit, + AdminRagResponse, AdminUserRecord, ApiErrorShape, ChatSyncResult, @@ -244,3 +246,26 @@ export function adminGetLlm(): Promise { export function adminListAutomations(): Promise { return request("/admin/automations"); } +/** GET /api/v1/admin/rag — lists knowledge documents; when query is present the + * backend also returns matching hits. */ +export function adminListRag(query?: string): Promise { + const qs = query ? `?query=${encodeURIComponent(query)}` : ""; + return request(`/admin/rag${qs}`); +} +/** POST /api/v1/admin/rag/query — runs a real knowledge-base query. */ +export function adminSearchRag( + query: string +): Promise<{ query: string; hits: AdminRagHit[] }> { + return request<{ query: string; hits: AdminRagHit[] }>("/admin/rag/query", { + method: "POST", + body: JSON.stringify({ query }), + }); +} +/** POST /api/v1/admin/rag/delete — deletes a document. The backend returns a + * clear 501 until M10 lands the write path (upload/chunk/embed/delete). */ +export function adminDeleteRag(id: string): Promise<{ id: string }> { + return request<{ id: string }>("/admin/rag/delete", { + method: "POST", + body: JSON.stringify({ id }), + }); +} diff --git a/frontend/web/src/lib/types.ts b/frontend/web/src/lib/types.ts index 0c3b99b..c68b8e7 100644 --- a/frontend/web/src/lib/types.ts +++ b/frontend/web/src/lib/types.ts @@ -249,6 +249,31 @@ export interface AdminLogRecord { msg: string; } +/** A knowledge-base document (GET /api/v1/admin/rag). */ +export interface AdminRagDoc { + id: string; + name: string; + desc?: string; + chunks: number; + embedded: boolean; +} + +/** One RAG retrieval hit (GET /api/v1/admin/rag?query= / POST /rag/query). */ +export interface AdminRagHit { + docId: string; + docName: string; + chapter?: string; + text: string; + score: number; + semantic: boolean; +} + +/** The GET /api/v1/admin/rag response shape. */ +export interface AdminRagResponse { + documents: AdminRagDoc[]; + hits?: AdminRagHit[]; +} + /** One audit entry: who / what / when / result. */ export interface AuditRecord { id: string; diff --git a/internal/runtime/knowledge/store.go b/internal/runtime/knowledge/store.go new file mode 100644 index 0000000..8ceedb7 --- /dev/null +++ b/internal/runtime/knowledge/store.go @@ -0,0 +1,266 @@ +// Package knowledge implements the RAG knowledge-base data source described in +// DevDocs/design/10-后台管理面板.md §3.12 (module 8 知识库 RAG) and the +// knowledge/memory separation in DevDocs/design/03-记忆.md §3 (③ 知识库 = +// 参考资料, 记忆 = 经历). +// +// It reads the project index (项目索引 → 项目描述 → 章节内容) as the set of knowledge +// documents and answers searches over it with line-level context. When a semantic +// retriever is wired, it also folds in project-linked long-term memories as +// semantic hits, ranked by the retriever's real keyword × semantic × decay path — +// so a configured vector index (pgvector / SQLite vector) is genuinely used and +// never a shell. +// +// The store is read-only for this milestone: upload/chunk/embed and delete land +// in M10. The write methods return ErrWriteUnsupported so the admin handler can +// answer honestly instead of fabricating a success. +package knowledge + +import ( + "context" + "errors" + "fmt" + "io" + "log/slog" + "strings" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory/projectindex" +) + +// ErrWriteUnsupported is returned by the write operations. Upload/chunk/embed and +// delete are milestone M10; the admin handler maps it to a clear 501 so the +// frontend can explain rather than show a placeholder alert. +var ErrWriteUnsupported = errors.New("knowledge: write operations (upload/chunk/embed/delete) land in M10") + +// defaultTopK bounds the semantic retrieval path. +const defaultTopK = 8 + +// Retriever is the subset of the memory retriever the knowledge store uses for +// semantic ranking. *memory.Retriever implements it; nil means no semantic path. +type Retriever interface { + Retrieve(ctx context.Context, personaID, query string, topK int) ([]memory.Result, error) +} + +// Doc is one knowledge-base document backing a project-index entry. +type Doc struct { + // ID is the project index id. + ID string + // Name is the project title (the document's display name). + Name string + // Desc is the project description (layer 2 of the hierarchy). + Desc string + // Chunks is the number of chapters the document is split into. + Chunks int + // Embedded reports whether a persistent vector index is wired for this + // deployment. This is the deployment-level semantic capability; per-document + // embedding state is not tracked yet (M10). + Embedded bool +} + +// Hit is one retrieval result surfaced to the admin panel. +type Hit struct { + // DocID is the owning project index id. + DocID string + // DocName is the owning project title. + DocName string + // Chapter is the chapter title that matched (empty for a semantic/memory hit). + Chapter string + // Text is the matched content: a chapter's located line(s) with context, or + // the memory content for a semantic hit. + Text string + // Score is the relevance score (higher = more relevant). + Score float64 + // Semantic reports whether the hit was ranked by the semantic/vector path. + Semantic bool +} + +// Deps carries the construction inputs. Every field is nil-tolerant so a store +// without a project index or retriever degrades to read-only empty answers. +type Deps struct { + // Projects is the project index backing the document list and text search. + // Optional; when nil the document list and text search return empty. + Projects *projectindex.Store + // Retriever ranks project-linked memories semantically. Optional; the + // semantic path runs only when both Retriever and DefaultPersona are set. + Retriever Retriever + // DefaultPersona scopes semantic retrieval to one persona. Optional. + DefaultPersona string + // VectorOn reports whether a persistent vector index (pgvector / SQLite + // vector) is wired, so Doc.Embedded is honest and never a shell. + VectorOn bool + // Log is the base logger. Nil uses a discard logger tagged with this package. + Log *slog.Logger +} + +// Store is the RAG data source. It is stateless and safe for concurrent use. +type Store struct { + projects *projectindex.Store + retriever Retriever + defaultPersona string + vectorOn bool + log *slog.Logger +} + +// New builds a Store. Availability degrades gracefully: a nil project index means +// no documents, a nil retriever means keyword/line search only. +func New(d Deps) *Store { + log := d.Log + if log == nil { + log = slog.New(slog.NewTextHandler(io.Discard, nil)) + } + return &Store{ + projects: d.Projects, + retriever: d.Retriever, + defaultPersona: strings.TrimSpace(d.DefaultPersona), + vectorOn: d.VectorOn, + log: logger.WithSource(log, "runtime/knowledge"), + } +} + +// ListDocuments enumerates the knowledge-base documents (project-index entries). +// A nil project index returns an empty, non-nil slice. +func (s *Store) ListDocuments(ctx context.Context) ([]Doc, error) { + out := make([]Doc, 0) + if s.projects == nil { + return out, nil + } + projects, err := s.projects.ListProjects(ctx) + if err != nil { + return nil, fmt.Errorf("knowledge: list projects: %w", err) + } + for i := range projects { + chapters, cerr := s.projects.ListChapters(ctx, projects[i].ID) + if cerr != nil { + s.log.Error("list chapters", "project", projects[i].ID, "err", cerr) + continue + } + out = append(out, Doc{ + ID: projects[i].ID, + Name: projects[i].Title, + Desc: projects[i].Desc, + Chunks: len(chapters), + Embedded: s.vectorOn, + }) + } + return out, nil +} + +// Search runs the knowledge-base search over the project index (line-level, +// hierarchy-aware) and, when a semantic retriever is configured, folds in +// project-linked memories ranked by the real vector path. A blank query returns +// no hits. +func (s *Store) Search(ctx context.Context, query string) ([]Hit, error) { + hits := make([]Hit, 0) + if strings.TrimSpace(query) == "" { + return hits, nil + } + if s.projects != nil { + textHits, err := s.searchProjects(ctx, query) + if err != nil { + return nil, err + } + hits = append(hits, textHits...) + } + if s.retriever != nil && s.defaultPersona != "" { + semHits, err := s.searchSemantic(ctx, query) + if err != nil { + s.log.Error("semantic search", "err", err) + } else { + hits = append(hits, semHits...) + } + } + return hits, nil +} + +// searchProjects walks every project and returns chapter/item hits via +// projectindex.Search (keyword + line context). A project deleted between the +// list and the search is skipped rather than failing the whole search. +func (s *Store) searchProjects(ctx context.Context, query string) ([]Hit, error) { + projects, err := s.projects.ListProjects(ctx) + if err != nil { + return nil, fmt.Errorf("knowledge: list projects: %w", err) + } + hits := make([]Hit, 0) + for i := range projects { + res, err := s.projects.Search(ctx, projects[i].ID, query) + if errors.Is(err, projectindex.ErrProjectNotFound) { + continue + } + if err != nil { + return nil, fmt.Errorf("knowledge: search project %q: %w", projects[i].ID, err) + } + if res.Project != nil { + hits = append(hits, Hit{ + DocID: res.Project.ID, + DocName: res.Project.Title, + Text: describeProject(res.Project), + Score: float64(len(res.Chapters) + 1), + }) + } + for _, ch := range res.Chapters { + hits = append(hits, chapterHit(projects[i], ch)) + } + } + return hits, nil +} + +// searchSemantic ranks memories via the retriever's real retrieval path and maps +// each project-linked result to a Hit (marked Semantic so the panel can show it). +func (s *Store) searchSemantic(ctx context.Context, query string) ([]Hit, error) { + results, err := s.retriever.Retrieve(ctx, s.defaultPersona, query, defaultTopK) + if err != nil { + return nil, fmt.Errorf("knowledge: retrieve: %w", err) + } + hits := make([]Hit, 0, len(results)) + for _, r := range results { + if r.Project == nil { + continue + } + hits = append(hits, Hit{ + DocID: r.Project.ID, + DocName: r.Project.Title, + Text: r.Item.Content, + Score: r.Score, + Semantic: true, + }) + } + return hits, nil +} + +// chapterHit maps a matched chapter to a Hit. Text is the located line plus its +// surrounding context (projectindex.LineMatch.Context), and Score reflects how +// many lines in the chapter matched. +func chapterHit(p projectindex.Project, ch projectindex.ChapterHit) Hit { + text := "" + if len(ch.Matches) > 0 { + text = ch.Matches[0].Context + } + return Hit{ + DocID: p.ID, + DocName: p.Title, + Chapter: ch.Chapter.Title, + Text: text, + Score: float64(len(ch.Matches)), + } +} + +// describeProject summarizes a project-index hit (title/desc match) for the panel. +func describeProject(p *projectindex.Project) string { + if p.Desc != "" { + return p.Desc + } + return p.Title +} + +// IndexDocument is not implemented in this milestone. It returns +// ErrWriteUnsupported so the handler answers honestly (upload/chunk/embed = M10). +func (s *Store) IndexDocument(_ context.Context, _ Doc) error { + return ErrWriteUnsupported +} + +// DeleteDocument is not implemented in this milestone. It returns +// ErrWriteUnsupported so the handler answers honestly (delete = M10). +func (s *Store) DeleteDocument(_ context.Context, _ string) error { + return ErrWriteUnsupported +} diff --git a/internal/runtime/memory/projectindex/store.go b/internal/runtime/memory/projectindex/store.go index 46ab836..b0b660c 100644 --- a/internal/runtime/memory/projectindex/store.go +++ b/internal/runtime/memory/projectindex/store.go @@ -105,6 +105,32 @@ func discardLogger(log *slog.Logger) *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, nil)) } +// ListProjects returns every project index entry in insertion order (created_at +// ASC, id ASC). An empty (non-nil) slice is returned when there are none, so +// callers can range over it unconditionally. A read error is wrapped with %w so +// callers can use errors.Is against the package sentinels. +func (s *Store) ListProjects(ctx context.Context) ([]Project, error) { + rows, err := s.db.QueryContext(ctx, `SELECT id, title, desc, created_at, updated_at + FROM project ORDER BY created_at ASC, id ASC`) + if err != nil { + return nil, fmt.Errorf("projectindex: list projects: %w", err) + } + defer func() { _ = rows.Close() }() + + projects := make([]Project, 0) + for rows.Next() { + p, err := scanProject(rows) + if err != nil { + return nil, fmt.Errorf("projectindex: scan project: %w", err) + } + projects = append(projects, *p) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("projectindex: iterate projects: %w", err) + } + return projects, nil +} + // Schema DDL, applied idempotently by New. The two tables model the three-layer // hierarchy: the project row is the index entry (layer 1) plus its description // (layer 2), and project_chapter holds the chapter content (layer 3). diff --git a/internal/services/channelbridge/channelbridge.go b/internal/services/channelbridge/channelbridge.go new file mode 100644 index 0000000..7b6320d --- /dev/null +++ b/internal/services/channelbridge/channelbridge.go @@ -0,0 +1,380 @@ +// Package channelbridge wires the registered IM channel adapters into the Cyrene +// runtime: it connects each configured adapter, installs the inbound message +// handler, and routes every normalized inbound message through the +// channel→session→actor→loop→reply pipeline (DevDocs/design/08 §0-§2). +// +// The bridge is deliberately generic over adapter.PlatformAdapter. It owns the +// per-platform Connect lifecycle and the inbound→reply path; the concrete +// adapters themselves (and their configuration, sourced from env) are assembled +// in cmd/cyrene/wire.go and handed in via Add. It never reads message text to +// decide identity — that is ResolveActor's job (actor metadata + owner whitelist, +// a project red line). +package channelbridge + +import ( + "context" + "errors" + "fmt" + "log/slog" + "strings" + "sync" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/user" +) + +// Knob constants. Compile-time so no bare magic number appears in the run path +// (project rule §10). +const ( + // connectTimeout bounds a single adapter Connect so a hung dial cannot stall + // boot (a failed connect logs a warning and is skipped, never fatal). + connectTimeout = 10 * time.Second + // maxConcurrent bounds how many inbound messages run the agent loop at once, + // so a burst of channel messages cannot spawn unbounded goroutines. + maxConcurrent = 16 +) + +// ErrNilLoop is returned when the bridge is built without a chat runner. +var ErrNilLoop = errors.New("channelbridge: nil loop runner") + +// ErrNilSessions is returned when the bridge is built without a session store. +var ErrNilSessions = errors.New("channelbridge: nil session store") + +// Chatter runs a single agent turn. *loop.Loop is the production implementation; +// tests inject a stub. +type Chatter interface { + Run(ctx context.Context, sessionID string, messages []model.Message) (*loop.Result, error) +} + +// SessionStore resolves the channel↔session binding for an inbound message. +// *session.SessionManager is the production implementation. +type SessionStore interface { + GetOrCreate(ctx context.Context, msg adapter.UnifiedMessage, owner *user.OwnerProfile) (session.Session, error) +} + +// Options carries the bridge's injected dependencies. Loop and Sessions are +// required; Owner may be nil (then no account is recognized as owner, the safe +// default-deny posture from design/11 §0). +type Options struct { + Log *slog.Logger + Loop Chatter + Sessions SessionStore + Owner *user.OwnerProfile +} + +// Bridge owns the channel adapter lifecycle and the inbound message pipeline. It +// is safe for concurrent use. +type Bridge struct { + log *slog.Logger + run Chatter + sessions SessionStore + owner *user.OwnerProfile + + mu sync.RWMutex + adapters map[string]adapter.PlatformAdapter + connected map[string]bool + closed bool + + ctx context.Context + cancel context.CancelFunc + wg sync.WaitGroup + + sem chan struct{} +} + +// New builds a Bridge from opts. It returns an error when a required dependency +// is missing so the caller fails fast rather than silently dropping messages. +func New(opts Options) (*Bridge, error) { + if opts.Loop == nil { + return nil, ErrNilLoop + } + if opts.Sessions == nil { + return nil, ErrNilSessions + } + log := opts.Log + if log == nil { + log = logger.New(logger.Options{Out: nil}) + } + return &Bridge{ + log: logger.WithSource(log, "services/channelbridge"), + run: opts.Loop, + sessions: opts.Sessions, + owner: opts.Owner, + adapters: make(map[string]adapter.PlatformAdapter), + connected: make(map[string]bool), + cancel: func() {}, + sem: make(chan struct{}, maxConcurrent), + }, nil +} + +// Add registers an adapter by its Name. It is safe to call before Start; adding +// after Start has no effect on already-running adapters. +func (b *Bridge) Add(a adapter.PlatformAdapter) { + if a == nil { + return + } + b.mu.Lock() + defer b.mu.Unlock() + b.adapters[a.Name()] = a +} + +// Start connects every configured adapter, installs the inbound handler, and +// launches the read loops. Adapters without credentials (or whose Connect fails) +// are logged and left disconnected; they never fail startup. Start blocks only +// long enough to connect (bounded by connectTimeout per adapter) and returns. +func (b *Bridge) Start() { + b.mu.Lock() + if b.closed { + b.mu.Unlock() + return + } + b.ctx, b.cancel = context.WithCancel(context.Background()) + b.mu.Unlock() + + for _, a := range b.snapshot() { + if !b.configured(a) { + b.log.Warn("channel skipped: no credentials", "channel", a.Name()) + continue + } + connCtx, cancel := context.WithTimeout(context.Background(), connectTimeout) + err := a.Connect(connCtx) + cancel() + if err != nil { + b.log.Warn("channel connect failed, leaving disconnected", + "channel", a.Name(), "error", err) + continue + } + b.mu.Lock() + b.connected[a.Name()] = true + b.mu.Unlock() + + a.SetMessageHandler(b.onMessage) + b.wg.Add(1) + go b.readLoop(a) + b.log.Info("channel connected", "channel", a.Name()) + } + b.log.Info("channel bridge started", "connected", b.ConnectedCount(), + "registered", len(b.snapshot())) +} + +// Stop cancels the read loops, disconnects every adapter and waits for in-flight +// message processing to finish. It is idempotent. +func (b *Bridge) Stop(ctx context.Context) { + b.mu.Lock() + if b.closed { + b.mu.Unlock() + return + } + b.closed = true + cancel := b.cancel + b.mu.Unlock() + + cancel() + for name, a := range b.snapshot() { + if err := a.Disconnect(ctx); err != nil { + b.log.Warn("channel disconnect", "channel", name, "error", err) + } + } + b.wg.Wait() + b.log.Info("channel bridge stopped") +} + +// IsConnected reports whether the named adapter is currently connected. +func (b *Bridge) IsConnected(name string) bool { + b.mu.RLock() + defer b.mu.RUnlock() + return b.connected[name] +} + +// ConnectedCount reports how many adapters are currently connected. +func (b *Bridge) ConnectedCount() int { + b.mu.RLock() + defer b.mu.RUnlock() + n := 0 + for _, ok := range b.connected { + if ok { + n++ + } + } + return n +} + +// snapshot returns a copy of the registered adapters for a lock-free iteration. +func (b *Bridge) snapshot() []adapter.PlatformAdapter { + b.mu.RLock() + defer b.mu.RUnlock() + out := make([]adapter.PlatformAdapter, 0, len(b.adapters)) + for _, a := range b.adapters { + out = append(out, a) + } + return out +} + +// configured reports whether the adapter carries enough configuration to attempt +// a Connect. It inspects the adapter's own Config() snapshot for the credential +// each platform needs, so a bare (empty) adapter is skipped rather than dialed. +// This is what keeps weixin-ilink from falling into its blocking QR-login flow +// when no bot token is present. +func (b *Bridge) configured(a adapter.PlatformAdapter) bool { + cfg := a.Config() + switch a.Name() { + case "telegram", "weixin-ilink": + return cfgStr(cfg, "bot_token") != "" + case "qq-obv11": + if cfgStr(cfg, "mode") == "client" { + return cfgStr(cfg, "remoteURL") != "" + } + return cfgStr(cfg, "listenAddr") != "" + case "webhook": + // Inbound-only: no external credential is required to hold the session. + return true + default: + return true + } +} + +// readLoop runs an adapter's blocking read path until the bridge stops. The emit +// sink is wired to onMessage so poll-style adapters (Telegram getUpdates, weixin +// long poll) deliver through the same pipeline as push-style adapters. +func (b *Bridge) readLoop(a adapter.PlatformAdapter) { + defer b.wg.Done() + ctx := b.ctx + a.ReadMessages(ctx, func(msg adapter.UnifiedMessage) { + b.onMessage(ctx, msg) + }) +} + +// onMessage is the adapter-facing inbound callback. It runs the pipeline in a +// bounded goroutine so the adapter's read loop never blocks on a slow agent turn. +func (b *Bridge) onMessage(ctx context.Context, msg adapter.UnifiedMessage) { + b.mu.Lock() + if b.closed { + b.mu.Unlock() + return + } + b.mu.Unlock() + + if err := msg.Validate(); err != nil { + b.log.Warn("inbound message invalid, dropped", "platform", msg.Platform, "error", err) + return + } + + select { + case b.sem <- struct{}{}: + case <-ctx.Done(): + return + } + + // Add the worker under the same lock that Stop uses to set closed, so a + // concurrent Stop either sees our Add before it waits, or we observe closed + // and bail without leaking a goroutine. + b.mu.Lock() + if b.closed { + b.mu.Unlock() + <-b.sem + return + } + b.wg.Add(1) + b.mu.Unlock() + + go func() { + defer b.wg.Done() + defer func() { <-b.sem }() + b.process(ctx, msg) + }() +} + +// process runs the full inbound pipeline: resolve the session/actor, drive the +// agent loop with the chat text, and send the loop's reply back through the +// channel. Identity is resolved structurally by the session layer; the bridge +// never reads message content to decide who the sender is. +func (b *Bridge) process(ctx context.Context, msg adapter.UnifiedMessage) { + text := strings.TrimSpace(msg.Content) + if text == "" { + b.log.Debug("inbound message has no chat text, skipped", + "platform", msg.Platform, "channel", msg.ChannelID) + return + } + + sess, err := b.sessions.GetOrCreate(ctx, msg, b.owner) + if err != nil { + b.log.Warn("session resolve failed", "platform", msg.Platform, + "channel", msg.ChannelID, "error", err) + return + } + + b.log.Debug("inbound → session", + "platform", msg.Platform, "channel", msg.ChannelID, + "session", sess.SessionID, "actor_role", sess.ActorRole, + "env", sess.Env(), "mem_writable", sess.MemWritable) + + res, err := b.run.Run(ctx, sess.SessionID, []model.Message{ + model.TextMessage(model.RoleUser, text), + }) + if err != nil { + b.log.Warn("agent run failed", "platform", msg.Platform, + "session", sess.SessionID, "error", err) + return + } + + reply := strings.TrimSpace(res.Text) + if reply == "" { + b.log.Debug("no reply produced", "platform", msg.Platform, "session", sess.SessionID) + return + } + if err := b.sendReply(ctx, msg, reply); err != nil { + if errors.Is(err, adapter.ErrUnsupported) { + b.log.Debug("channel does not support outbound reply, skipped", + "platform", msg.Platform, "session", sess.SessionID) + return + } + b.log.Warn("send reply failed", "platform", msg.Platform, "error", err) + } +} + +// sendReply delivers the loop's text back to the channel's source conversation. +// It mirrors the inbound message's channel targeting fields so the adapter +// resolves the same peer the message came from. +func (b *Bridge) sendReply(ctx context.Context, in adapter.UnifiedMessage, text string) error { + a := b.adapter(in.Platform) + if a == nil { + return fmt.Errorf("channelbridge: no adapter for platform %q", in.Platform) + } + out := adapter.UnifiedMessage{ + Platform: in.Platform, + ChannelType: in.ChannelType, + ChannelID: in.ChannelID, + GroupID: in.GroupID, + GroupName: in.GroupName, + SenderID: in.SenderID, + Sender: in.Sender, + Content: text, + ContentType: adapter.ContentText, + Timestamp: time.Now().UTC(), + } + return a.SendMessage(ctx, adapter.UnifiedResponse{Messages: []adapter.UnifiedMessage{out}}) +} + +// adapter returns the registered adapter for a platform name, or nil. +func (b *Bridge) adapter(platform string) adapter.PlatformAdapter { + b.mu.RLock() + defer b.mu.RUnlock() + return b.adapters[platform] +} + +// cfgStr reads a string field from an adapter config map, returning "" when +// absent. +func cfgStr(m map[string]any, key string) string { + if m == nil { + return "" + } + if s, ok := m[key].(string); ok { + return s + } + return "" +} diff --git a/internal/services/channelbridge/channelbridge_test.go b/internal/services/channelbridge/channelbridge_test.go new file mode 100644 index 0000000..538cd02 --- /dev/null +++ b/internal/services/channelbridge/channelbridge_test.go @@ -0,0 +1,209 @@ +package channelbridge + +import ( + "context" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/user" +) + +// fakeAdapter is a minimal PlatformAdapter that records outbound sends so the +// bridge's reply path can be asserted without network I/O. +type fakeAdapter struct { + name string + connected bool + sends []adapter.UnifiedMessage + handler adapter.MessageHandler + sendErr error +} + +func (f *fakeAdapter) Name() string { return f.name } +func (f *fakeAdapter) Config() map[string]any { + return map[string]any{"bot_token": "configured"} +} +func (f *fakeAdapter) Connect(context.Context) error { f.connected = true; return nil } +func (f *fakeAdapter) Disconnect(context.Context) error { + f.connected = false + return nil +} +func (f *fakeAdapter) IsConnected() bool { return f.connected } +func (f *fakeAdapter) HealthCheck(context.Context) error { return nil } +func (f *fakeAdapter) ReadMessages(ctx context.Context, _ func(adapter.UnifiedMessage)) { + <-ctx.Done() +} +func (f *fakeAdapter) SendMessage(_ context.Context, resp adapter.UnifiedResponse) error { + for _, m := range resp.Messages { + f.sends = append(f.sends, m) + } + return f.sendErr +} +func (f *fakeAdapter) SendProactive(context.Context, adapter.UnifiedMessage) error { return nil } +func (f *fakeAdapter) SetMessageHandler(fn adapter.MessageHandler) { f.handler = fn } +func (f *fakeAdapter) Capabilities() adapter.Capabilities { return adapter.DefaultCapabilities() } +func (f *fakeAdapter) ListChannels(context.Context) ([]adapter.ChannelInfo, error) { return nil, nil } + +// stubChatter records the session id and messages passed to the agent loop. +type stubChatter struct { + sessionID string + messages []model.Message + res *loop.Result + err error +} + +func (s *stubChatter) Run(_ context.Context, sessionID string, msgs []model.Message) (*loop.Result, error) { + s.sessionID = sessionID + s.messages = msgs + return s.res, s.err +} + +// stubSessions records the inbound message and returns a fixed session. +type stubSessions struct { + msg adapter.UnifiedMessage + owner *user.OwnerProfile + sess session.Session + err error +} + +func (s *stubSessions) GetOrCreate(_ context.Context, msg adapter.UnifiedMessage, owner *user.OwnerProfile) (session.Session, error) { + s.msg = msg + s.owner = owner + if s.err != nil { + return session.Session{}, s.err + } + return s.sess, nil +} + +func directMessage() adapter.UnifiedMessage { + return adapter.UnifiedMessage{ + Platform: "telegram", + ChannelID: "100", + ChannelType: adapter.ChannelDirect, + SenderID: "100", + Content: "hello", + ContentType: adapter.ContentText, + } +} + +func newTestBridge(t *testing.T) (*Bridge, *stubChatter, *stubSessions, *fakeAdapter) { + t.Helper() + chat := &stubChatter{res: &loop.Result{Text: "hi back"}} + sess := &stubSessions{sess: session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}} + ad := &fakeAdapter{name: "telegram"} + + b, err := New(Options{Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}}) + if err != nil { + t.Fatalf("New: %v", err) + } + b.Add(ad) + return b, chat, sess, ad +} + +func TestProcessRunsPipelineAndReplies(t *testing.T) { + t.Parallel() + + b, chat, sess, ad := newTestBridge(t) + ctx := context.Background() + + b.process(ctx, directMessage()) + + if sess.msg.SenderID != "100" { + t.Errorf("resolved sender = %q, want 100", sess.msg.SenderID) + } + if chat.sessionID != "sess-1" { + t.Errorf("loop session id = %q, want sess-1", chat.sessionID) + } + if len(chat.messages) != 1 || chat.messages[0].Role != model.RoleUser { + t.Fatalf("loop messages = %+v, want single user message", chat.messages) + } + if got := chat.messages[0].Content[0].Text; got != "hello" { + t.Errorf("loop text = %q, want hello", got) + } + if len(ad.sends) != 1 { + t.Fatalf("send count = %d, want 1", len(ad.sends)) + } + reply := ad.sends[0] + if reply.Content != "hi back" || reply.ContentType != adapter.ContentText { + t.Errorf("reply = %+v, want text 'hi back'", reply) + } + if reply.ChannelID != "100" || reply.Platform != "telegram" { + t.Errorf("reply target = %+v, want platform telegram channel 100", reply) + } +} + +func TestProcessSkipsEmptyText(t *testing.T) { + t.Parallel() + + b, chat, _, ad := newTestBridge(t) + msg := directMessage() + msg.Content = " " + + b.process(context.Background(), msg) + + if chat.sessionID != "" { + t.Errorf("loop ran for empty content, session = %q", chat.sessionID) + } + if len(ad.sends) != 0 { + t.Errorf("send count = %d, want 0 for empty content", len(ad.sends)) + } +} + +func TestProcessLoopErrorDoesNotSend(t *testing.T) { + t.Parallel() + + b, chat, _, ad := newTestBridge(t) + chat.res = nil + chat.err = context.DeadlineExceeded + + b.process(context.Background(), directMessage()) + + if chat.sessionID != "sess-1" { + t.Errorf("loop should still run for valid text, session = %q", chat.sessionID) + } + if len(ad.sends) != 0 { + t.Errorf("send count = %d, want 0 on loop error", len(ad.sends)) + } +} + +func TestProcessGroupMessageKeepsGroupTarget(t *testing.T) { + t.Parallel() + + b, chat, _, ad := newTestBridge(t) + msg := adapter.UnifiedMessage{ + Platform: "telegram", + ChannelID: "-20", + ChannelType: adapter.ChannelGroup, + GroupID: "-20", + GroupName: "Gang", + SenderID: "9", + Content: "hello all", + ContentType: adapter.ContentText, + } + chat.res = &loop.Result{Text: "group ack"} + + b.process(context.Background(), msg) + + if len(ad.sends) != 1 { + t.Fatalf("send count = %d, want 1", len(ad.sends)) + } + reply := ad.sends[0] + if reply.ChannelType != adapter.ChannelGroup || reply.GroupID != "-20" { + t.Errorf("group reply target = %+v, want group -20", reply) + } +} + +func TestProcessEmptyReplySendsNothing(t *testing.T) { + t.Parallel() + + b, chat, _, ad := newTestBridge(t) + chat.res = &loop.Result{Text: " "} + + b.process(context.Background(), directMessage()) + + if len(ad.sends) != 0 { + t.Errorf("send count = %d, want 0 for empty reply", len(ad.sends)) + } +} diff --git a/internal/services/gateway/admin.go b/internal/services/gateway/admin.go index 0476dfa..ce48a55 100644 --- a/internal/services/gateway/admin.go +++ b/internal/services/gateway/admin.go @@ -6,11 +6,13 @@ import ( "errors" "net/http" "strconv" + "strings" "time" "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" capreg "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry" apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/knowledge" "git.yeij.top/AskaEth/Cyrene_v2/pkg/plugins/sdk" ) @@ -251,6 +253,150 @@ func (s *Server) adminListBackthink(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, out) } +// AdminRagDoc is the wire shape of one knowledge-base document (project-index +// entry) in the admin RAG list. +type AdminRagDoc struct { + ID string `json:"id"` + Name string `json:"name"` + Desc string `json:"desc,omitempty"` + Chunks int `json:"chunks"` + Embedded bool `json:"embedded"` +} + +// AdminRagHit is the wire shape of one RAG retrieval hit. +type AdminRagHit struct { + DocID string `json:"docId"` + DocName string `json:"docName"` + Chapter string `json:"chapter,omitempty"` + Text string `json:"text"` + Score float64 `json:"score"` + Semantic bool `json:"semantic"` +} + +// AdminRagResponse is the wire shape of GET /api/v1/admin/rag. Hits is present +// only when a query parameter was supplied, so callers distinguish "no search +// requested" from "searched and found nothing". +type AdminRagResponse struct { + Documents []AdminRagDoc `json:"documents"` + Hits []AdminRagHit `json:"hits,omitempty"` +} + +// ragQueryRequest is the POST /api/v1/admin/rag/query body. +type ragQueryRequest struct { + Query string `json:"query"` +} + +// ragDeleteRequest is the POST /api/v1/admin/rag/delete body. +type ragDeleteRequest struct { + ID string `json:"id"` +} + +// adminListRag lists the knowledge-base documents and, when ?query= is present, +// returns the matching hits too. Without a wired RagStore it degrades to an empty +// list (backward compatible — never a 500). +func (s *Server) adminListRag(w http.ResponseWriter, r *http.Request) { + out := AdminRagResponse{Documents: make([]AdminRagDoc, 0)} + if s.rag == nil { + writeJSON(w, http.StatusOK, out) + return + } + docs, err := s.rag.ListDocuments(r.Context()) + if err != nil { + s.log.Error("list rag", "err", err) + writeError(w, apperr.Wrap(apperr.ErrInternal, "rag_failed", "failed to list knowledge documents")) + return + } + out.Documents = ragDocsFrom(docs) + if query := strings.TrimSpace(r.URL.Query().Get("query")); query != "" { + hits, herr := s.rag.Search(r.Context(), query) + if herr != nil { + s.log.Error("search rag", "err", herr) + writeError(w, apperr.Wrap(apperr.ErrInternal, "rag_search_failed", "failed to search knowledge base")) + return + } + out.Hits = ragHitsFrom(hits) + } + writeJSON(w, http.StatusOK, out) +} + +// adminSearchRag runs a knowledge-base query and returns the matching hits. +func (s *Server) adminSearchRag(w http.ResponseWriter, r *http.Request) { + if s.rag == nil { + s.auditClaims(r, "rag.query", "error") + writeError(w, &apperr.AppError{Code: "not_implemented", Message: "知识库 RAG 服务未配置（未注入 RagStore）", HTTPCode: http.StatusNotImplemented}) + return + } + var req ragQueryRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "invalid request body")) + return + } + if strings.TrimSpace(req.Query) == "" { + writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "query is required")) + return + } + hits, err := s.rag.Search(r.Context(), req.Query) + if err != nil { + s.log.Error("search rag", "err", err) + writeError(w, apperr.Wrap(apperr.ErrInternal, "rag_search_failed", "failed to search knowledge base")) + return + } + s.auditClaims(r, "rag.query", "ok") + writeJSON(w, http.StatusOK, map[string]any{"query": req.Query, "hits": ragHitsFrom(hits)}) +} + +// adminDeleteRag answers the document delete honestly: write operations +// (upload/chunk/embed/delete) are milestone M10, so it returns a clear 501 rather +// than a fabricated success. The route is audited as an admin write. +func (s *Server) adminDeleteRag(w http.ResponseWriter, r *http.Request) { + if s.rag == nil { + s.auditClaims(r, "rag.delete", "error") + writeError(w, &apperr.AppError{Code: "not_implemented", Message: "知识库 RAG 服务未配置（未注入 RagStore）", HTTPCode: http.StatusNotImplemented}) + return + } + var req ragDeleteRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "invalid request body")) + return + } + if strings.TrimSpace(req.ID) == "" { + writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "id is required")) + return + } + if err := s.rag.DeleteDocument(r.Context(), req.ID); err != nil { + s.auditClaims(r, "rag.delete", "error") + writeError(w, &apperr.AppError{Code: "not_implemented", Message: "知识库写操作（上传/分块/嵌入/删除）由 M10 后端提供", HTTPCode: http.StatusNotImplemented}) + return + } + s.auditClaims(r, "rag.delete", "ok") + writeJSON(w, http.StatusOK, map[string]any{"id": req.ID}) +} + +// ragDocsFrom maps knowledge docs to their admin wire shape. +func ragDocsFrom(docs []knowledge.Doc) []AdminRagDoc { + out := make([]AdminRagDoc, 0, len(docs)) + for _, d := range docs { + out = append(out, AdminRagDoc{ID: d.ID, Name: d.Name, Desc: d.Desc, Chunks: d.Chunks, Embedded: d.Embedded}) + } + return out +} + +// ragHitsFrom maps knowledge hits to their admin wire shape. +func ragHitsFrom(hits []knowledge.Hit) []AdminRagHit { + out := make([]AdminRagHit, 0, len(hits)) + for _, h := range hits { + out = append(out, AdminRagHit{ + DocID: h.DocID, + DocName: h.DocName, + Chapter: h.Chapter, + Text: h.Text, + Score: h.Score, + Semantic: h.Semantic, + }) + } + return out +} + // adminSetPolicy upserts the reachability markers for a capability. It is the // audited admin write example (owner-only, enforced by the route registration). func (s *Server) adminSetPolicy(w http.ResponseWriter, r *http.Request) { diff --git a/internal/services/gateway/extra_test.go b/internal/services/gateway/extra_test.go index 0324616..fab3e3b 100644 --- a/internal/services/gateway/extra_test.go +++ b/internal/services/gateway/extra_test.go @@ -22,6 +22,7 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/config" "git.yeij.top/AskaEth/Cyrene_v2/internal/iot" personareg "git.yeij.top/AskaEth/Cyrene_v2/internal/persona/registry" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/knowledge" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" @@ -188,6 +189,21 @@ type stubMemories struct { func (s *stubMemories) List(ctx context.Context) ([]memory.MemoryItem, error) { return s.list, s.err } +// stubRag mimics RagStore, returning domain docs/hits the gateway maps to its +// admin wire shape. +type stubRag struct { + docs []knowledge.Doc + hits []knowledge.Hit + err error +} + +func (s *stubRag) ListDocuments(ctx context.Context) ([]knowledge.Doc, error) { return s.docs, s.err } +func (s *stubRag) Search(ctx context.Context, query string) ([]knowledge.Hit, error) { + return s.hits, s.err +} +func (s *stubRag) IndexDocument(ctx context.Context, doc knowledge.Doc) error { return s.err } +func (s *stubRag) DeleteDocument(ctx context.Context, id string) error { return s.err } + // stubIoT mimics IoTStore. type stubIoT struct { list []iot.Device @@ -1219,3 +1235,132 @@ func reqWithOrigin(origin string) *http.Request { func reqWithQuery(k, v string) *http.Request { return httptest.NewRequest(http.MethodGet, "/ws/chat?"+k+"="+v, nil) } + +// --- RAG knowledge base ---------------------------------------------------- + +func TestAdminRagListsAndSearches(t *testing.T) { + t.Parallel() + docs := []knowledge.Doc{{ID: "p1", Name: "开发时间线.md", Desc: "milestones", Chunks: 3, Embedded: true}} + hits := []knowledge.Hit{{DocID: "p1", DocName: "开发时间线.md", Chapter: "M4", Text: "RAG 检索测试", Score: 1.0}} + srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{RAG: &stubRag{docs: docs, hits: hits}}) + _ = srv + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + // List only. + resp, body := doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/rag", nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("list status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + var list struct { + Documents []AdminRagDoc `json:"documents"` + Hits []AdminRagHit `json:"hits"` + } + if err := json.Unmarshal(decodeBody(t, body)["data"], &list); err != nil { + t.Fatalf("decode list: %v (body=%s)", err, body) + } + if len(list.Documents) != 1 || list.Documents[0].Name != "开发时间线.md" { + t.Fatalf("documents = %+v, want one 开发时间线.md", list.Documents) + } + if list.Hits != nil { + t.Fatalf("hits = %+v, want omitted without query", list.Hits) + } + + // List + search via ?query=. + resp2, body2 := doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/rag?query=rag", nil, tok) + if resp2.StatusCode != http.StatusOK { + t.Fatalf("search status = %d, want 200 (body=%s)", resp2.StatusCode, body2) + } + var search struct { + Documents []AdminRagDoc `json:"documents"` + Hits []AdminRagHit `json:"hits"` + } + if err := json.Unmarshal(decodeBody(t, body2)["data"], &search); err != nil { + t.Fatalf("decode search: %v (body=%s)", err, body2) + } + if len(search.Hits) != 1 || search.Hits[0].DocName != "开发时间线.md" { + t.Fatalf("hits = %+v, want one hit", search.Hits) + } +} + +func TestAdminRagQueryPost(t *testing.T) { + t.Parallel() + hits := []knowledge.Hit{{DocID: "p1", DocName: "开发时间线.md", Text: "RAG 检索测试", Score: 1.0}} + srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{RAG: &stubRag{hits: hits}}) + _ = srv + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + resp, body := doReq(t, http.MethodPost, ts.URL+"/api/v1/admin/rag/query", map[string]any{"query": "rag"}, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + var res struct { + Query string `json:"query"` + Hits []AdminRagHit `json:"hits"` + } + if err := json.Unmarshal(decodeBody(t, body)["data"], &res); err != nil { + t.Fatalf("decode: %v (body=%s)", err, body) + } + if res.Query != "rag" || len(res.Hits) != 1 { + t.Fatalf("response = %+v, want query=rag with 1 hit", res) + } +} + +func TestAdminRagEmptyWhenUnwired(t *testing.T) { + t.Parallel() + srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{}) + _ = srv + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + // List degrades to empty (never a 500). + resp, body := doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/rag", nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("list status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + var list struct { + Documents []AdminRagDoc `json:"documents"` + } + if err := json.Unmarshal(decodeBody(t, body)["data"], &list); err != nil { + t.Fatalf("decode: %v (body=%s)", err, body) + } + if len(list.Documents) != 0 { + t.Fatalf("documents = %+v, want empty", list.Documents) + } + + // Query reports a clear 501 (backend not implemented). + resp2, body2 := doReq(t, http.MethodPost, ts.URL+"/api/v1/admin/rag/query", map[string]any{"query": "rag"}, tok) + if resp2.StatusCode != http.StatusNotImplemented { + t.Fatalf("query status = %d, want 501 (body=%s)", resp2.StatusCode, body2) + } + wantErrorShape(t, body2) + if !strings.Contains(string(body2), "not_implemented") { + t.Fatalf("error body = %s, want not_implemented code", body2) + } +} + +func TestAdminRagDeleteNotImplemented(t *testing.T) { + t.Parallel() + srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{RAG: &stubRag{err: knowledge.ErrWriteUnsupported}}) + _ = srv + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + resp, body := doReq(t, http.MethodPost, ts.URL+"/api/v1/admin/rag/delete", map[string]any{"id": "p1"}, tok) + if resp.StatusCode != http.StatusNotImplemented { + t.Fatalf("status = %d, want 501 (body=%s)", resp.StatusCode, body) + } + wantErrorShape(t, body) + if !strings.Contains(string(body), "M10") { + t.Fatalf("error body = %s, want M10 write message", body) + } +} + +func TestAdminRagDeleteRequiresID(t *testing.T) { + t.Parallel() + srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{RAG: &stubRag{}}) + _ = srv + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + resp, _ := doReq(t, http.MethodPost, ts.URL+"/api/v1/admin/rag/delete", map[string]any{"id": " "}, tok) + if resp.StatusCode != http.StatusBadRequest { + t.Fatalf("status = %d, want 400", resp.StatusCode) + } +} diff --git a/internal/services/gateway/server.go b/internal/services/gateway/server.go index 5f668f5..28d5e0f 100644 --- a/internal/services/gateway/server.go +++ b/internal/services/gateway/server.go @@ -35,6 +35,7 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/iot" "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" personareg "git.yeij.top/AskaEth/Cyrene_v2/internal/persona/registry" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/knowledge" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" @@ -132,6 +133,23 @@ type BackthinkStore interface { List(ctx context.Context, limit int) ([]backthink.Record, error) } +// RagStore backs the admin knowledge-base RAG module (design/10 §3.12 module 8). +// *runtime/knowledge.Store is the production implementation. Every method is +// read-only for this milestone: IndexDocument/DeleteDocument return a clear +// "write lands in M10" error rather than a fabricated success. +type RagStore interface { + // ListDocuments enumerates the knowledge-base documents (project-index + // entries). + ListDocuments(ctx context.Context) ([]knowledge.Doc, error) + // Search answers a knowledge-base query (line-level hierarchy + optional + // semantic/vector ranking). + Search(ctx context.Context, query string) ([]knowledge.Hit, error) + // IndexDocument uploads/chunks/embeds a document. M10; returns an error. + IndexDocument(ctx context.Context, doc knowledge.Doc) error + // DeleteDocument removes a document. M10; returns an error. + DeleteDocument(ctx context.Context, id string) error +} + // Deps carries the gateway's injected dependencies. Every field is nil-tolerant // except Auth, which is required (authentication backs login and every RBAC // decision). Registry/Loop are satisfied by the concrete *capability.Registry and @@ -170,6 +188,9 @@ type Deps struct { Users UserStore // Backthink lists background-think records. Optional. Backthink BackthinkStore + // RAG backs the admin knowledge-base RAG module. Optional; list/search + // return empty when unset, and write endpoints report a clear 501. + RAG RagStore // Streaming marks the injected Loop as streaming-capable (its OnStream is // wired to publish ws.stream_chunk). When set the hub skips its own single // final stream_chunk emission, since deltas already stream through OnStream. @@ -197,6 +218,7 @@ type Server struct { automations AutomationStore users UserStore backthink BackthinkStore + rag RagStore voice VoiceService streaming bool frontend fs.FS @@ -229,6 +251,7 @@ func New(cfg *config.Config, deps Deps) (*Server, error) { automations: deps.Automations, users: deps.Users, backthink: deps.Backthink, + rag: deps.RAG, streaming: deps.Streaming, frontend: deps.Frontend, log: sourcedLog(deps.Log), @@ -269,6 +292,9 @@ func (s *Server) Handler() http.Handler { mux.Handle("GET /api/v1/admin/automations", adminAuth(http.HandlerFunc(s.adminListAutomations))) mux.Handle("GET /api/v1/admin/users", adminAuth(http.HandlerFunc(s.adminListUsers))) mux.Handle("GET /api/v1/admin/backthink", adminAuth(http.HandlerFunc(s.adminListBackthink))) + mux.Handle("GET /api/v1/admin/rag", adminAuth(http.HandlerFunc(s.adminListRag))) + mux.Handle("POST /api/v1/admin/rag/query", adminAuth(http.HandlerFunc(s.adminSearchRag))) + mux.Handle("POST /api/v1/admin/rag/delete", adminAuth(http.HandlerFunc(s.adminDeleteRag))) mux.Handle("POST /api/v1/admin/plugins/policy", s.auth.Authorize(auth.RoleOwner)(http.HandlerFunc(s.adminSetPolicy))) var h http.Handler = mux