From 70c9fba52972b3fe4a2b01b89455f8f6e602ac47 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Tue, 8 Sep 2026 23:20:12 +0800 Subject: [PATCH] =?UTF-8?q?feat(plugin):=20=E6=8F=92=E4=BB=B6=E7=83=AD?= =?UTF-8?q?=E5=AE=89=E8=A3=85(MCP=20server)=20=E2=80=94=20POST=20/admin/pl?= =?UTF-8?q?ugins/install=20=E8=BF=9E=E6=8E=A5+=E5=8F=91=E7=8E=B0+=E6=B3=A8?= =?UTF-8?q?=E5=86=8C=E5=B7=A5=E5=85=B7=E8=BF=9Bregistry+=E8=90=BD=E7=9B=98?= =?UTF-8?q?plugins.json?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - capability/plugin/manager.go: Manager(Install/Uninstall/List/Restore/Close); Install=连接+发现+mcp:name注册工具,失败不落盘; holder持有*mcp.Registry; boot Restore重装 - config/runtimeconfig/plugins.go: PluginConfig(Normalize/Validate), LoadPlugins/SavePlugins(0600+tmp+rename)/Upsert/Delete/Masked(apiKey+敏感env掩码) - mcp: MCPConfig加APIKey/Env, http/sse Bearer应用, stdio子进程env合并; registry.ToolNames() - gateway: adminPluginInstall/Uninstall/ListInstalledPlugins; Deps.Plugins注入 - cmd/cyrene/plugins.go: wirePlugins(boot Restore+close) - 契约: body{kind:mcp,name,config{transport, endpoint?/command?,args?,apiKey?,env?}} → {installed,name,kind,transport,toolCount}; 失败502不落盘/非法400/未注入501; 卸载连带unregister+删配置 - 测试(manager/plugins/admin/mcp_auth/组合根), build/vet/test绿+3平台交叉编译 - Docs/api-client v0.28 + api-reference install=501改真实; .gitignore加plugins.json - 前端api.ts仍发旧body(无config)会400, 面板安装表单待后续 --- .gitignore | 1 + Docs/api-client.md | 8 +- Docs/api-reference.md | 4 +- cmd/cyrene/plugins.go | 20 + cmd/cyrene/plugins_test.go | 97 +++++ cmd/cyrene/wire.go | 7 + internal/capability/mcp/http.go | 10 + internal/capability/mcp/mcp.go | 6 + internal/capability/mcp/mcp_auth_test.go | 47 +++ internal/capability/mcp/registry.go | 8 + internal/capability/mcp/sse.go | 11 + internal/capability/mcp/stdio.go | 23 ++ internal/capability/plugin/manager.go | 318 ++++++++++++++++ internal/capability/plugin/manager_test.go | 341 ++++++++++++++++++ internal/config/runtimeconfig/plugins.go | 255 +++++++++++++ internal/config/runtimeconfig/plugins_test.go | 181 ++++++++++ internal/services/gateway/admin_plugins.go | 238 ++++++++++++ .../services/gateway/admin_plugins_test.go | 301 ++++++++++++++++ .../admin_write_plugins_memory_sessions.go | 55 +-- internal/services/gateway/server.go | 11 + 20 files changed, 1885 insertions(+), 57 deletions(-) create mode 100644 cmd/cyrene/plugins.go create mode 100644 cmd/cyrene/plugins_test.go create mode 100644 internal/capability/mcp/mcp_auth_test.go create mode 100644 internal/capability/plugin/manager.go create mode 100644 internal/capability/plugin/manager_test.go create mode 100644 internal/config/runtimeconfig/plugins.go create mode 100644 internal/config/runtimeconfig/plugins_test.go create mode 100644 internal/services/gateway/admin_plugins.go create mode 100644 internal/services/gateway/admin_plugins_test.go diff --git a/.gitignore b/.gitignore index c394843..4b974ba 100644 --- a/.gitignore +++ b/.gitignore @@ -5,6 +5,7 @@ DevDocs/ .env models.json channels.json +plugins.json *.local *.pem *.key diff --git a/Docs/api-client.md b/Docs/api-client.md index 9934152..ae0b5b9 100644 --- a/Docs/api-client.md +++ b/Docs/api-client.md @@ -54,6 +54,7 @@ | 3 | `/api/v1/admin/memories` | GET | ✅ | 记忆列表（手动添加/检索已实现；编辑/删除/衰减随 M10 扩展） | | 4 | `/api/v1/admin/plugins` | GET | ✅ | 插件/能力列表（含 MCP/渠道） | | 5 | `/api/v1/admin/plugins/policy` | POST | ✅ | 写工具策略（owner-only，审计） | +| 5a | `/api/v1/admin/plugins/installed` | GET | ✅ | **已装运行时插件列表**（owner/admin 读）：`{data:[{id,name,kind,transport,endpoint?,command?,args?,apiKey?,env?,createdAt?,live,toolCount}]}`，`apiKey`/敏感 `env` 值**掩码**（如 `****1234`），`live` = 是否已连接注册，`toolCount` = 该 server 注册工具数 | | 6 | `/api/v1/admin/channels` | GET | ✅ | 渠道列表/健康 + 每平台已配账号（脱敏 token 掩码，design/04 §3.1）；已连接渠道的账号附 **bot 身份**（`botId`/`botUsername`/`botNickname`，如 KOOK `GET /api/v3/user/me` 自动获取） | | 7 | `/api/v1/admin/iot` | GET | ✅ | IoT 设备列表 | | 8 | `/api/v1/admin/users` | GET/POST | ✅(GET) | 用户/角色/API key（GET 已实现；POST 创建用户/签发 key 随 M10 落地） | @@ -73,7 +74,7 @@ | 22 | `/api/v1/admin/system/backup` | POST | ✅(写) | 备份数据（owner-only，审计）；用 SQLite `VACUUM INTO` 快照（含已提交 WAL 内容）到 `/backup-.db` → `{path,size}` | | 23 | `/api/v1/admin/system/restore` | POST | ✅(写) | 恢复数据（owner-only，审计）；body `{path}` → 校验路径在 dataDir 内→**先备份当前**→替换 DB 文件→`{restored:true,backup,note:"需重启后生效"}`（运行进程持有 DB 句柄，热替换需重启可见） | -> **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/observability/metrics` · `GET /api/v1/admin/observability/trace` · `GET /api/v1/admin/observability/audit` 已实现，返回真实数据；`GET /api/v1/admin/logs` 仍🕐未实现（前端按契约封装、后端 404 时显示"待实现"壳，不报错）。**写操作已补齐**（见下方「管理面板写操作契约」）：人格/API key/设备控制/自动化/LLM 测试/渠道连接与识别测试/记忆检索/会话导出/插件卸载**真实持久化或真实生效**；**系统设置安全与存储类**（吊销 API key / 轮换 JWT 密钥 / 备份 / 恢复）也已**真实实现**（见下方「系统设置契约」）；仍为后置 M10：**新增用户**、**IoT 场景与规则**、**插件安装（需源码清单）**、**主人档案保存**、**编辑环境变量/重启服务/检查更新**、**知识库删除**——后端如实返回 `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/rag` · `POST /api/v1/admin/rag/query` · `GET /api/v1/admin/observability/metrics` · `GET /api/v1/admin/observability/trace` · `GET /api/v1/admin/observability/audit` 已实现，返回真实数据；`GET /api/v1/admin/logs` 仍🕐未实现（前端按契约封装、后端 404 时显示"待实现"壳，不报错）。**写操作已补齐**（见下方「管理面板写操作契约」）：人格/API key/设备控制/自动化/LLM 测试/渠道连接与识别测试/记忆检索/会话导出/插件卸载**真实持久化或真实生效**；**系统设置安全与存储类**（吊销 API key / 轮换 JWT 密钥 / 备份 / 恢复）也已**真实实现**（见下方「系统设置契约」）；仍为后置 M10：**新增用户**、**IoT 场景与规则**、**主人档案保存**、**编辑环境变量/重启服务/检查更新**、**知识库删除**——后端如实返回 `501 not_implemented`，不伪造成功。（**插件安装已实现**：v1 为 MCP server 热安装，见下方写操作契约。） **管理面板写操作契约**（`POST/PUT/DELETE /api/v1/admin/*`，均 **owner-only**，读为 owner/admin；写操作经 audit 落 `audit_logs`；成功 `{data}` / 失败 `{error:{code,message}}`）： - `POST /admin/personas` body `{id,name,role,systemPrompt?,characterRef?,modes?/mode?,memoryNamespace?}` → `201 {data:{id,name,character,isDefault,mode,memoryNamespace}}`（注册进 persona registry + 持久化到 `configs/personas/.yaml`）；id 已存在回 `409 duplicate`。 @@ -100,8 +101,8 @@ - `PUT /admin/channels/{platform}/config`（owner-only，审计）body `{id?,token?,baseURL?,remoteURL?,listenAddr?,mode?,accessToken?,selfID?,config?,allowGroups?,denyGroups?,allowUsers?,denyUsers?,groupMsgPerMin?,msgPerMin?,maxMessages?}` → `201 {data:{persisted:true,path,platform,accountId}}`。**真实持久化 + 热更新**（design/04 §3.1，无需重启）：读当前有效配置（channels.json→env 兜底）→ `UpsertChannel` 该平台账号 → 落盘 `channels.json`（本地、gitignore，0600）→ `bridge.Reload(platform)` 热替换该平台 adapter（进行中消息用旧、新消息用新）；token 写入文件但**永不在响应/日志回显**（GET 列表为掩码）。`platform` 必填；空配置回 `400`；未注入渠道配置服务回 `500 channels_unavailable`、写盘失败 `500 channel_persist_failed`、热重载失败 `500 channel_reload_failed`。`.env`（`CHANNELS_JSON` / `CHANNEL_*_TOKEN`）为兜底。 - `DELETE /admin/channels/{platform}/config?id={accountId}`（owner-only，审计）→ `200 {data:{persisted:true,path,platform,accountId}}`。删除该平台指定账号 → 落盘 `channels.json` → `bridge.Reload(platform)` 热更新（该账号 adapter 优雅关停）。`id` 必填；未注入服务回 `500 channels_unavailable`。 - `GET /admin/channels/{platform}/messages?limit=&before=`（owner/admin 读）→ `200 {data:[{id,sessionId,role,content,mode?,mood?,image?,platform?,deviceId?,senderName?,createdAt}]}`。**IM 适配器日志流**（design/16 §4.6 §4）：该平台**所有非主人私聊/群聊消息**（隔离会话 `sess-xxx`，非主会话 `web`）以消息流显示，`senderName` 标发送者昵称；未注入 `HistoryStore` 回 `501`。 -- `POST /admin/plugins/install` → `501 not_implemented`（需插件源码清单）。 -- `POST /admin/plugins/uninstall` body `{name}` → `200 {data:{name,uninstalled:true}}`（真实从 registry 移除）；`404`。 +- `POST /admin/plugins/install`（**owner-only + 审计 `plugin.install`**）body `{kind:"mcp",name,config:{transport:"stdio|http|sse",endpoint?|command?,args?,apiKey?,env?}}` → `200 {data:{installed:true,name,kind:"mcp",transport,toolCount}}`。**v1 只装 MCP server**（单二进制无法热装 Go 插件）：真实 connect+discover，把发现的工具**热注册**进 capability registry（下次模型请求即注入，无需重启）→ 落盘 `plugins.json`（本地、gitignore、0600，`apiKey` 写入但**永不回显/不落日志**）→ 持有 live server 供后续刷新/卸载。**同名重复安装 = 刷新**（先卸旧再装新，失败回滚旧）。校验：`name` 必填、`kind` 非 `mcp` 回 `400`、transport 非法/`stdio` 缺 `command`/`http|sse` 缺 `endpoint` 回 `400`；**连接/发现/注册失败回 `502 plugin_install_failed` 且不落盘**（诚实，不静默装一半）。未注入 plugin manager 回 `501 not_implemented`。 +- `POST /admin/plugins/uninstall`（**owner-only + 审计 `plugin.uninstall`**）body `{name}` → `200 {data:{name,uninstalled:true,kind:"mcp"}}`（已装 MCP：`Unregister` 移除其工具 + close + 删 `plugins.json` 条目；**非 MCP/普通能力**走原 `registry.Remove`，返回 `{data:{name,uninstalled:true}}`）；未命中回 `404`。 - `POST /admin/memories` body `{personaId?,content,scope?}` → `201 {data:{id,scope,text}}`。**手动添加记忆**（owner 私聊直写）；`scope` ∈ `user|persona|project`，缺省 `persona`（与「记住…」同路径）；无效非空 `scope` 回 `400`；未注入 `memory.Store` 回 `501`。 - `POST /admin/memories/search` body `{query,personaId?,topK?}` → `200 {data:{query,personaId,semantic,hits:[{id,scope,text,score}]}}`。真实检索（keyword×衰减）；**语义/向量路径**需主会话注入 vector index（`semantic` 当前恒 `false`，诚实反映未接向量器）。 - `GET /admin/sessions/{id}/export` → `200 {data:{session:{platform,channelId,sessionId,actorRole,owner,isGroup,memWritable}}}`；`404`。当前导出会话身份元数据（完整消息转录由 M10 提供）。 @@ -215,6 +216,7 @@ - 语音：上行 `voice_input`（base64 音频或已转写 text），下行 `voice_output`（合成音频 base64 + text）。 ## 5. 变更记录 +- `v0.28`（**MCP 插件热安装/卸载**）：`POST /api/v1/admin/plugins/install`（**由 501 改为真实实现**，owner-only + 审计）body `{kind:"mcp",name,config:{transport,endpoint?|command?,args?,apiKey?,env?}}` → `200 {data:{installed,name,kind,transport,toolCount}}`——真实连接 MCP server、发现并**热注册工具**进 capability registry（下一轮模型请求即注入），配置落盘 `plugins.json`（gitignore、0600）；连接失败回 `502` 且**不落盘**；同名重复 = 刷新。`POST /api/v1/admin/plugins/uninstall` 扩展：MCP 插件走 `Unregister`（移除工具 + close + 删配置），普通能力保持原 `registry.Remove`。新增 `GET /api/v1/admin/plugins/installed`（owner/admin 读）列出已装插件（`apiKey`/敏感 `env` 掩码 + `live`/`toolCount`）。`mcp.MCPConfig` 新增 `APIKey`（http/sse 以 `Authorization: Bearer` 发送）与 `Env`（stdio 子进程环境）；`plugins.json` 于启动时 `Restore` 重新安装。**契约版本 `v0.28`**。 - `v0.27`（**LLM token 预算可配置化**）：`models.json` 模型组新增可选 **`maxTokens`**（`int`，缺省/`0` = 未设），作为该组主/备链的 token 预算；`GET /api/v1/admin/llm` 的 `AdminLlmProvider` 与 `POST /api/v1/admin/llm` body/回显同步该字段（`omitempty`，未设为 0 时不出现）。`.env` 新增 **`LLM_FALLBACK_MAX_TOKENS`**（`int`，默认 `8192`，`0` = 关闭兜底），只作用于**确有回退模型**时链末端的回退步。**优先级**：显式 `Request.MaxTokens`（审查/判定类小预算）> 生效组 `maxTokens` > `LLM_FALLBACK_MAX_TOKENS`（仅链末端、且组未设时）> 适配器默认（anthropic 8192）。单步链（无 fallback）不套用 env 兜底。`maxTokens` 为负回 `400 bad_request`。**契约版本 `v0.27`**。 - `v0.26`（**A5 消息正向增量**）：`GET /api/v1/sessions/{id}/messages` 新增 **`after`** 查询参数（``，升序返回 `created_at > after` 的消息，默认 `limit=50`/上限 200）——解决「WS 重连只回放主会话最近 50 条、断线超 50 条永久丢」的缺口；`before` 现行为不变；**`after` 与 `before` 互斥**（同传 `400`）；负数/非数字 `400`。`HistoryStore` 新增 `ListAfter`（走 `idx_chat_msgs_session_created`，`rowid` 兜底同毫秒并列）。**游标契约**：客户端用「本页最后一条 `createdAt`」作下一页 `after`；同毫秒消息 > limit 时纯 `createdAt` 游标会跨页重读（超罕见，客户端按 `id` 去重兜底，与现有绕行一致）。 - `v0.25`（**B2 refresh token**）：`POST /api/v1/auth/refresh`（**新增**，`{refreshToken}` → 新 access + 新 refresh，默认单次使用旋转，并发重放 fail-closed `invalid_grant`）+ `POST /api/v1/auth/logout`（**新增**，撤单个或全设备）。`POST /api/v1/auth/login` 在 `{token,expiresAt,role}` 之外**新增 `refreshToken`**（不透明 256-bit 随机串，**仅 SHA-256 哈希落库**，`AUTH_REFRESH_TTL` 默认 `720h`，`AUTH_REFRESH_ROTATE_DISABLED` 关旋转）。新表 `refresh_tokens`（迁移 0008，id/user_id/role/token_hash UNIQUE/expires_at/revoked_at；含过期/轮换原子事务）；`rotate-jwt` 成功即删除全部 refresh（返回 `refreshTokensRevoked`），`revoke-keys` 不动 refresh。**密码/secret 记录见 `internal/auth`；refresh 全文绝不落日志/审计。** diff --git a/Docs/api-reference.md b/Docs/api-reference.md index 23432dc..6b24cbe 100644 --- a/Docs/api-reference.md +++ b/Docs/api-reference.md @@ -116,8 +116,8 @@ | 50 | POST | `/api/v1/admin/channels/{platform}/message/test` | **owner** | 200 | KOOK 发送测试消息 | M17 | | 51 | PUT | `/api/v1/admin/channels/{platform}/config` | **owner** | 201 | 保存渠道账号配置（落盘 + 热更新） | M16 | | 52 | DELETE | `/api/v1/admin/channels/{platform}/config` | **owner** | 200 | 删除渠道账号 | M16 | -| 53 | POST | `/api/v1/admin/plugins/install` | **owner** | 501 | 安装插件（需源码清单） | M10 | -| 54 | POST | `/api/v1/admin/plugins/uninstall` | **owner** | 200 | 移除插件 | M8 | +| 53 | POST | `/api/v1/admin/plugins/install` | **owner** | 200 | **热安装 MCP server 插件**（body `{kind:"mcp",name,config:{transport,endpoint?/command?,args?,apiKey?,env?}}` → 连接+发现+`RegisterTool` 进 registry → 落盘 `plugins.json` → `200 {data:{installed,name,kind,transport,toolCount}}`）；连接/发现失败 `502 plugin_install_failed` 且**不落盘**；非法 `400`；manager 未注入 `501` | 后 M12 | +| 54 | POST | `/api/v1/admin/plugins/uninstall` | **owner** | 200 | 移除插件（MCP → `Unregister`+删 `plugins.json` 配置；普通能力 → `reg.Remove`） | M8 | | 55 | POST | `/api/v1/admin/memories` | **owner** | 201 | 手动添加记忆 | M8 | | 56 | POST | `/api/v1/admin/memories/search` | **owner** | 200 | 记忆检索 | M8 | | 57 | POST | `/api/v1/admin/rag` | **owner** | 201 | 导入知识文档 | M10 | diff --git a/cmd/cyrene/plugins.go b/cmd/cyrene/plugins.go new file mode 100644 index 0000000..bde25c2 --- /dev/null +++ b/cmd/cyrene/plugins.go @@ -0,0 +1,20 @@ +package main + +import ( + "context" + + pluginmgr "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/plugin" + "git.yeij.top/AskaEth/Cyrene_v2/internal/config/runtimeconfig" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" +) + +// wirePlugins builds the runtime plugin manager over the capability registry and +// the gitignored plugins.json, then restores every persisted MCP server so an +// installed plugin survives a restart. A server that is unreachable at boot is +// logged and left in plugins.json (the owner can retry or uninstall it); boot +// never fails because an external server is down. +func (a *app) wirePlugins(ctx context.Context) { + path := runtimeconfig.DefaultPluginsPath(resolveDataDir(a.cfg)) + a.plugins = pluginmgr.NewManager(a.caps, path, logger.WithSource(a.log, "capability/plugin")) + a.plugins.Restore(ctx) +} diff --git a/cmd/cyrene/plugins_test.go b/cmd/cyrene/plugins_test.go new file mode 100644 index 0000000..20dc469 --- /dev/null +++ b/cmd/cyrene/plugins_test.go @@ -0,0 +1,97 @@ +package main + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry" + "git.yeij.top/AskaEth/Cyrene_v2/internal/config" + "git.yeij.top/AskaEth/Cyrene_v2/internal/config/runtimeconfig" +) + +// pluginFakeMCP is a minimal Streamable HTTP MCP server for the composition-root +// test: it answers initialize and tools/list over POST. +type pluginFakeMCP struct{ tools []string } + +func (f pluginFakeMCP) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodGet { + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + if fl, ok := w.(http.Flusher); ok { + fl.Flush() + } + <-r.Context().Done() + return + } + body, _ := io.ReadAll(r.Body) + var req struct { + ID json.RawMessage `json:"id"` + Method string `json:"method"` + } + _ = json.Unmarshal(body, &req) + if len(req.ID) == 0 || string(req.ID) == "null" { + w.WriteHeader(http.StatusAccepted) + return + } + var result any + switch req.Method { + case "initialize": + result = map[string]any{"protocolVersion": "2025-11-25", "capabilities": map[string]any{}, "serverInfo": map[string]any{"name": "fake", "version": "1.0"}} + case "tools/list": + list := make([]map[string]any, 0, len(f.tools)) + for _, name := range f.tools { + list = append(list, map[string]any{"name": name, "description": name, "inputSchema": map[string]any{"type": "object"}}) + } + result = map[string]any{"tools": list} + default: + result = map[string]any{} + } + resp, _ := json.Marshal(map[string]any{"jsonrpc": "2.0", "id": json.RawMessage(req.ID), "result": result}) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write(resp) +} + +// TestWirePlugins_RestoresPersistedServer proves the composition root builds the +// plugin manager over the resolved data dir (plugins.json) and reinstalls a +// persisted MCP server at boot, so a panel install survives a restart. +func TestWirePlugins_RestoresPersistedServer(t *testing.T) { + dir := t.TempDir() + mcpSrv := httptest.NewServer(pluginFakeMCP{tools: []string{"add"}}) + defer mcpSrv.Close() + plugins := []runtimeconfig.PluginConfig{{ + ID: "p1", + Name: "calc", + Kind: runtimeconfig.PluginKindMCP, + Transport: runtimeconfig.TransportHTTP, + Endpoint: mcpSrv.URL, + CreatedAt: "2026-09-08T00:00:00Z", + }} + raw, err := json.Marshal(plugins) + if err != nil { + t.Fatalf("marshal: %v", err) + } + if err := os.WriteFile(filepath.Join(dir, "plugins.json"), raw, 0o600); err != nil { + t.Fatalf("write plugins.json: %v", err) + } + + caps := registry.New() + a := &app{caps: caps, cfg: &config.Config{DataDir: dir}, log: testLog()} + a.wirePlugins(context.Background()) + if a.plugins == nil { + t.Fatal("plugins manager not wired") + } + defer a.plugins.Close(context.Background()) + + if _, err := caps.Get("add"); err != nil { + t.Fatalf("restored tool add not registered: %v", err) + } + if _, err := caps.Get("mcp:calc"); err != nil { + t.Fatalf("mcp:calc capability not registered: %v", err) + } +} diff --git a/cmd/cyrene/wire.go b/cmd/cyrene/wire.go index 5a3402b..b91d3bc 100644 --- a/cmd/cyrene/wire.go +++ b/cmd/cyrene/wire.go @@ -14,6 +14,7 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/mcp" + pluginmgr "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/plugin" "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry" "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session" @@ -93,6 +94,7 @@ type app struct { iot *iot.Manager voice *voice.Chain mcpRegs []*mcp.Registry + plugins *pluginmgr.Manager sessions *session.SessionManager automation *automations.Service backthink *backthink.Backthink @@ -214,6 +216,7 @@ func newApp(ctx context.Context, cfg *config.Config, log *slog.Logger) (*app, er if err := a.wireAuxiliary(); err != nil { return a, err } + a.wirePlugins(ctx) if err := a.wireChannels(); err != nil { return a, err } @@ -690,6 +693,7 @@ func (a *app) wireFrontend() error { RAG: ragStore, LLM: a.llm, Channels: a.channels, + Plugins: a.plugins, OwnerRegistrar: &ownerRegistrar{path: defaultOwnerPath, owner: a.owner, log: a.log}, AuditStore: auditStore, Streaming: true, @@ -784,6 +788,9 @@ func (a *app) close(ctx context.Context) { for _, reg := range a.mcpRegs { _ = reg.Unregister(ctx) } + if a.plugins != nil { + a.plugins.Close(ctx) + } if a.tasks != nil { _ = a.tasks.Close() } diff --git a/internal/capability/mcp/http.go b/internal/capability/mcp/http.go index 4b3f177..bcfd055 100644 --- a/internal/capability/mcp/http.go +++ b/internal/capability/mcp/http.go @@ -145,6 +145,7 @@ func (s *streamableHTTPTransport) post(ctx context.Context, body []byte) (*http. func (s *streamableHTTPTransport) applyHeaders(req *http.Request) { req.Header.Set("Content-Type", "application/json") req.Header.Set("Accept", "application/json, text/event-stream") + s.applyAuth(req) s.mu.Lock() session := s.session proto := "" @@ -160,6 +161,14 @@ func (s *streamableHTTPTransport) applyHeaders(req *http.Request) { } } +// applyAuth sets the bearer credential an authenticated server requires. It is +// a no-op when the config carries no API key. +func (s *streamableHTTPTransport) applyAuth(req *http.Request) { + if s.cfg.APIKey != "" { + req.Header.Set("Authorization", "Bearer "+s.cfg.APIKey) + } +} + // captureSession stores a session id returned by the server. func (s *streamableHTTPTransport) captureSession(resp *http.Response) { if sid := resp.Header.Get(mcpSessionIDHeader); sid != "" { @@ -321,6 +330,7 @@ func (s *streamableHTTPTransport) readStandalone(ctx context.Context) { return } req.Header.Set("Accept", "text/event-stream") + s.applyAuth(req) s.mu.Lock() session := s.session proto := "" diff --git a/internal/capability/mcp/mcp.go b/internal/capability/mcp/mcp.go index 1b3dcfc..7609691 100644 --- a/internal/capability/mcp/mcp.go +++ b/internal/capability/mcp/mcp.go @@ -149,6 +149,12 @@ type MCPConfig struct { // TransportType selects the transport explicitly: "stdio", "sse" or // "http" (Streamable HTTP). Empty means infer from Command/URL. TransportType string `env:"MCP_TRANSPORT"` + // APIKey is the bearer credential sent to an http/sse server as an + // Authorization header. Empty means the server needs no auth. + APIKey string `env:"MCP_API_KEY"` + // Env is extra environment for a stdio subprocess, merged over the parent + // process environment (extra keys win). It is ignored by the HTTP transports. + Env map[string]string `env:"MCP_ENV"` // Enabled reports whether this server should be activated at startup. Enabled bool `env:"MCP_ENABLED"` } diff --git a/internal/capability/mcp/mcp_auth_test.go b/internal/capability/mcp/mcp_auth_test.go new file mode 100644 index 0000000..bb2366c --- /dev/null +++ b/internal/capability/mcp/mcp_auth_test.go @@ -0,0 +1,47 @@ +package mcp + +import ( + "context" + "net/http" + "testing" +) + +func TestEnvWith_OverridesAndPreserves(t *testing.T) { + t.Parallel() + base := []string{"PATH=/usr/bin", "HOME=/root"} + got := envWith(base, map[string]string{"HOME": "/tmp", "MCP_TOKEN": "abc"}) + if len(got) != 3 { + t.Fatalf("len = %d, want 3 (%v)", len(got), got) + } + seen := map[string]bool{} + for _, kv := range got { + seen[kv] = true + } + if !seen["PATH=/usr/bin"] || !seen["HOME=/tmp"] || !seen["MCP_TOKEN=abc"] { + t.Fatalf("merged env = %v", got) + } + if len(envWith(base, nil)) != len(base) { + t.Fatal("empty extra changed the base environment") + } +} + +func TestHTTPTransport_SendsAPIKey(t *testing.T) { + t.Parallel() + got := make(chan string, 1) + srv := httpTestServer(t, func(w http.ResponseWriter, r *http.Request) { + got <- r.Header.Get("Authorization") + httpJSONResponse(w, r) + }) + defer srv.Close() + tr := newHTTPTransport(MCPConfig{Name: "s", URL: srv.URL, APIKey: "sk-1234"}, defaultLogger()) + defer tr.Close() + if err := tr.Start(context.Background()); err != nil { + t.Fatalf("Start: %v", err) + } + if err := tr.Call(context.Background(), "tools/list", nil, &listToolsResult{}); err != nil { + t.Fatalf("Call: %v", err) + } + if auth := <-got; auth != "Bearer sk-1234" { + t.Fatalf("Authorization = %q, want bearer key", auth) + } +} diff --git a/internal/capability/mcp/registry.go b/internal/capability/mcp/registry.go index 8b1e92a..bb58bbd 100644 --- a/internal/capability/mcp/registry.go +++ b/internal/capability/mcp/registry.go @@ -127,6 +127,14 @@ func (r *Registry) refresh() { r.log.Info("mcp tools refreshed", "name", r.mcp.cfg.Name, "count", len(r.names)) } +// ToolNames returns a copy of the tool names this registry registered, in +// registration order. It is empty before Register succeeds. +func (r *Registry) ToolNames() []string { + r.mu.Lock() + defer r.mu.Unlock() + return append([]string(nil), r.names...) +} + // Unregister removes every tool this registry previously registered and closes // the underlying client. It is safe to call multiple times. func (r *Registry) Unregister(ctx context.Context) error { diff --git a/internal/capability/mcp/sse.go b/internal/capability/mcp/sse.go index 83f6384..9b483f1 100644 --- a/internal/capability/mcp/sse.go +++ b/internal/capability/mcp/sse.go @@ -71,6 +71,7 @@ func (s *sseTransport) Start(ctx context.Context) error { return fmt.Errorf("mcp sse get: %w", err) } req.Header.Set("Accept", "text/event-stream") + s.applyAuth(req) resp, err := http.DefaultClient.Do(req) if err != nil { return fmt.Errorf("mcp sse connect %s: %w", s.cfg.URL, err) @@ -187,6 +188,7 @@ func (s *sseTransport) Call(ctx context.Context, method string, params any, resu } httpReq.Header.Set("Content-Type", "application/json") httpReq.Header.Set("Accept", "application/json") + s.applyAuth(httpReq) resp, err := http.DefaultClient.Do(httpReq) if err != nil { return fmt.Errorf("mcp sse %s post: %w", method, err) @@ -245,6 +247,7 @@ func (s *sseTransport) notify(ctx context.Context, method string, params any) er return fmt.Errorf("mcp notify %s: %w", method, err) } httpReq.Header.Set("Content-Type", "application/json") + s.applyAuth(httpReq) resp, err := http.DefaultClient.Do(httpReq) if err != nil { return fmt.Errorf("mcp notify %s: %w", method, err) @@ -254,6 +257,14 @@ func (s *sseTransport) notify(ctx context.Context, method string, params any) er return nil } +// applyAuth sets the bearer credential an authenticated server requires. It is +// a no-op when the config carries no API key. +func (s *sseTransport) applyAuth(req *http.Request) { + if s.cfg.APIKey != "" { + req.Header.Set("Authorization", "Bearer "+s.cfg.APIKey) + } +} + // Close closes the SSE stream and any pending responses. func (s *sseTransport) Close() error { s.closeOnce.Do(func() { diff --git a/internal/capability/mcp/stdio.go b/internal/capability/mcp/stdio.go index aca18f1..f79163b 100644 --- a/internal/capability/mcp/stdio.go +++ b/internal/capability/mcp/stdio.go @@ -8,6 +8,7 @@ import ( "fmt" "io" "log/slog" + "os" "os/exec" "strconv" "strings" @@ -51,6 +52,7 @@ func (s *stdioTransport) Start(ctx context.Context) error { return fmt.Errorf("mcp stdio: %w", ErrNoTransport) } cmd := exec.Command(s.cfg.Command, s.cfg.Args...) + cmd.Env = envWith(os.Environ(), s.cfg.Env) stdin, err := cmd.StdinPipe() if err != nil { return fmt.Errorf("mcp stdio stdin: %w", err) @@ -200,3 +202,24 @@ func (l *lineRPC) notify(ctx context.Context, method string, params any) error { } return nil } + +// envWith returns base with extra merged over it (an extra key overrides a +// same-named base entry). A nil/empty extra returns base unchanged. +func envWith(base []string, extra map[string]string) []string { + if len(extra) == 0 { + return base + } + out := make([]string, 0, len(base)+len(extra)) + for _, kv := range base { + if k, _, ok := strings.Cut(kv, "="); ok { + if _, dup := extra[k]; dup { + continue + } + } + out = append(out, kv) + } + for k, v := range extra { + out = append(out, k+"="+v) + } + return out +} diff --git a/internal/capability/plugin/manager.go b/internal/capability/plugin/manager.go new file mode 100644 index 0000000..b15c398 --- /dev/null +++ b/internal/capability/plugin/manager.go @@ -0,0 +1,318 @@ +// Package plugin manages runtime-installed external tool providers. In v1 the +// only installable kind is an MCP server (DevDocs/design/09 插件生态 §1): a +// compiled Go plugin cannot be hot-loaded into a single static binary, so a +// runtime "install" is an external MCP server whose config is persisted and +// whose discovered tools are hot-registered into the shared capability registry. +// +// The manager owns the live *mcp.Registry of every installed server so the +// server can be refreshed (reinstall under the same name) or uninstalled, and +// the config can be restored on the next boot. A connect/discover failure never +// persists a half-installed plugin. +package plugin + +import ( + "context" + "errors" + "fmt" + "log/slog" + "strings" + "sync" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/mcp" + "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry" + "git.yeij.top/AskaEth/Cyrene_v2/internal/config/runtimeconfig" +) + +const ( + // connectTimeout bounds one server's connect + discover handshake so an + // unreachable server fails the install instead of hanging the request. + connectTimeout = 15 * time.Second + // mcpCapabilityPrefix names the KindMCP capability a server surfaces as, so + // the admin plugin list enumerates the server itself as well as its tools. + mcpCapabilityPrefix = "mcp:" +) + +// Installed is one plugin as reported to the admin API: its persisted config, +// whether a live server currently backs it, and how many tools it registered. +type Installed struct { + // Config is the persisted plugin config (the caller masks it before wire). + Config runtimeconfig.PluginConfig + // Live reports whether the MCP server is connected and its tools registered. + Live bool + // ToolCount is the number of tools the live server registered (0 when not + // live). + ToolCount int +} + +// server is a live installed MCP server. +type server struct { + cfg runtimeconfig.PluginConfig + reg *mcp.Registry + toolCount int +} + +// Manager installs, refreshes, lists and uninstalls runtime plugins over one +// capability registry and one plugins.json path. It is safe for concurrent use. +type Manager struct { + reg *registry.Registry + path string + log *slog.Logger + + mu sync.Mutex + servers map[string]*server +} + +// NewManager returns a Manager backed by reg and the plugins.json at path. +func NewManager(reg *registry.Registry, path string, log *slog.Logger) *Manager { + if log == nil { + log = slog.New(slog.DiscardHandler) + } + return &Manager{reg: reg, path: path, log: log, servers: make(map[string]*server)} +} + +// Install connects cfg's MCP server, registers its tools, persists the config +// and holds the live registry. Installing a name that already exists refreshes +// it: the previous server is unregistered first, and if the new one fails to +// connect the previous one is restored (the on-disk config is never changed on a +// failure). +func (m *Manager) Install(ctx context.Context, cfg runtimeconfig.PluginConfig) (Installed, error) { + return m.install(ctx, cfg, true) +} + +// install is Install with the persist step optional (Restore reuses it without +// rewriting the file it just read). +func (m *Manager) install(ctx context.Context, cfg runtimeconfig.PluginConfig, persist bool) (Installed, error) { + cfg = cfg.Normalize() + if err := cfg.Validate(); err != nil { + return Installed{}, err + } + prev := m.take(cfg.Name) + cfg = inheritIdentity(cfg, prev) + if strings.TrimSpace(cfg.CreatedAt) == "" { + cfg.CreatedAt = time.Now().UTC().Format(time.RFC3339) + } + if prev != nil { + m.stop(ctx, prev) + } + srv, err := m.start(ctx, cfg) + if err != nil { + m.restore(prev) + return Installed{}, err + } + if persist { + if err := m.persistUpsert(cfg); err != nil { + m.stop(ctx, srv) + m.restore(prev) + return Installed{}, err + } + } + m.put(srv) + m.log.Info("plugin installed", "name", cfg.Name, "kind", cfg.Kind, "transport", cfg.Transport, "tools", srv.toolCount) + return Installed{Config: cfg, Live: true, ToolCount: srv.toolCount}, nil +} + +// Uninstall removes the plugin matching name: it unregisters the live server's +// tools and closes it, then deletes the persisted config. It reports whether +// anything matched, so the caller can fall back to a plain capability removal. +func (m *Manager) Uninstall(ctx context.Context, name string) (bool, error) { + srv := m.take(name) + deleted, err := m.persistDelete(name) + if srv != nil { + m.stop(ctx, srv) + } + if err != nil { + return srv != nil, fmt.Errorf("plugin %q: %w", name, err) + } + removed := srv != nil || deleted + if removed { + m.log.Info("plugin uninstalled", "name", name) + } + return removed, nil +} + +// List returns every persisted plugin (in file order) followed by any live +// server not in the file, each annotated with its live state and tool count. +func (m *Manager) List() []Installed { + cfgs, err := runtimeconfig.LoadPlugins(m.path) + if err != nil { + m.log.Warn("plugins.json load skipped", "path", m.path, "error", err) + cfgs = nil + } + out := make([]Installed, 0, len(cfgs)) + seen := make(map[string]bool, len(cfgs)) + m.mu.Lock() + defer m.mu.Unlock() + for _, cfg := range cfgs { + item := Installed{Config: cfg} + if srv, ok := m.servers[cfg.Name]; ok { + item.Live = true + item.ToolCount = srv.toolCount + } + out = append(out, item) + seen[cfg.Name] = true + } + for name, srv := range m.servers { + if seen[name] { + continue + } + out = append(out, Installed{Config: srv.cfg, Live: true, ToolCount: srv.toolCount}) + } + return out +} + +// Restore installs every persisted plugin. A server that is unreachable at boot +// is logged and left in plugins.json (so the owner can retry or uninstall it); +// boot never fails because an external server is down. +func (m *Manager) Restore(ctx context.Context) { + cfgs, err := runtimeconfig.LoadPlugins(m.path) + if err != nil { + m.log.Warn("plugins.json load skipped", "path", m.path, "error", err) + return + } + for _, cfg := range cfgs { + if _, err := m.install(ctx, cfg, false); err != nil { + m.log.Warn("plugin restore skipped", "name", cfg.Name, "error", err) + } + } +} + +// Close unregisters every live server without touching plugins.json. +func (m *Manager) Close(ctx context.Context) { + m.mu.Lock() + live := make([]*server, 0, len(m.servers)) + for _, srv := range m.servers { + live = append(live, srv) + } + m.servers = make(map[string]*server) + m.mu.Unlock() + for _, srv := range live { + m.stop(ctx, srv) + } +} + +// start connects cfg, registers its tools and surfaces its KindMCP capability. +// On a registration failure it cleans up and returns the error. +func (m *Manager) start(ctx context.Context, cfg runtimeconfig.PluginConfig) (*server, error) { + mreg := mcp.NewRegistry(m.reg, mcpConfigOf(cfg), mcp.WithLogger(m.log)) + connectCtx, cancel := context.WithTimeout(ctx, connectTimeout) + defer cancel() + if err := mreg.Register(connectCtx); err != nil { + _ = mreg.Unregister(context.Background()) + return nil, fmt.Errorf("plugin %q: %w", cfg.Name, err) + } + names := mreg.ToolNames() + err := m.reg.Register(registry.Capability{ + Name: mcpCapabilityPrefix + cfg.Name, + Kind: registry.KindMCP, + Description: "MCP server " + cfg.Name, + Value: mreg, + }) + if err != nil && !errors.Is(err, registry.ErrDuplicate) { + m.log.Warn("plugin capability register skipped", "name", cfg.Name, "error", err) + } + return &server{cfg: cfg, reg: mreg, toolCount: len(names)}, nil +} + +// stop unregisters a live server's tools and closes it, then drops its KindMCP +// capability. Errors are logged, never fatal: uninstall must still remove the +// config. +func (m *Manager) stop(ctx context.Context, srv *server) { + if srv == nil { + return + } + if err := srv.reg.Unregister(ctx); err != nil { + m.log.Warn("plugin unregister", "name", srv.cfg.Name, "error", err) + } + if err := m.reg.Remove(mcpCapabilityPrefix + srv.cfg.Name); err != nil && !errors.Is(err, registry.ErrNotFound) { + m.log.Warn("plugin capability remove", "name", srv.cfg.Name, "error", err) + } +} + +// restore re-installs a server that a failed refresh displaced, so a failed +// refresh never leaves the registry without a previously working server. +func (m *Manager) restore(prev *server) { + if prev == nil { + return + } + srv, err := m.start(context.Background(), prev.cfg) + if err != nil { + m.log.Warn("plugin rollback failed", "name", prev.cfg.Name, "error", err) + return + } + m.put(srv) +} + +// take removes and returns the live server named name, or nil. +func (m *Manager) take(name string) *server { + m.mu.Lock() + defer m.mu.Unlock() + srv := m.servers[name] + delete(m.servers, name) + return srv +} + +// put records srv as the live server for its name. +func (m *Manager) put(srv *server) { + m.mu.Lock() + defer m.mu.Unlock() + m.servers[srv.cfg.Name] = srv +} + +// persistUpsert writes cfg into plugins.json (upsert by name). +func (m *Manager) persistUpsert(cfg runtimeconfig.PluginConfig) error { + cur, err := runtimeconfig.LoadPlugins(m.path) + if err != nil { + return fmt.Errorf("load plugins.json: %w", err) + } + if err := runtimeconfig.SavePlugins(m.path, runtimeconfig.UpsertPlugin(cur, cfg)); err != nil { + return fmt.Errorf("save plugins.json: %w", err) + } + return nil +} + +// persistDelete removes name (by name or id) from plugins.json, reporting +// whether an entry was removed. A missing file is not an error. +func (m *Manager) persistDelete(name string) (bool, error) { + cur, err := runtimeconfig.LoadPlugins(m.path) + if err != nil { + return false, fmt.Errorf("load plugins.json: %w", err) + } + out, removed := runtimeconfig.DeletePlugin(cur, name) + if !removed { + return false, nil + } + if err := runtimeconfig.SavePlugins(m.path, out); err != nil { + return false, fmt.Errorf("save plugins.json: %w", err) + } + return true, nil +} + +// mcpConfigOf maps a plugin config to the MCP client config. +func mcpConfigOf(cfg runtimeconfig.PluginConfig) mcp.MCPConfig { + return mcp.MCPConfig{ + Name: cfg.Name, + Command: cfg.Command, + Args: cfg.Args, + URL: cfg.Endpoint, + TransportType: cfg.Transport, + APIKey: cfg.APIKey, + Env: cfg.Env, + Enabled: true, + } +} + +// inheritIdentity keeps a refresh's id and install time when the request omits +// them, so a reinstall updates a plugin rather than minting a new identity. +func inheritIdentity(cfg runtimeconfig.PluginConfig, prev *server) runtimeconfig.PluginConfig { + if prev == nil { + return cfg + } + if strings.TrimSpace(cfg.ID) == "" { + cfg.ID = prev.cfg.ID + } + if strings.TrimSpace(cfg.CreatedAt) == "" { + cfg.CreatedAt = prev.cfg.CreatedAt + } + return cfg +} diff --git a/internal/capability/plugin/manager_test.go b/internal/capability/plugin/manager_test.go new file mode 100644 index 0000000..5e0059a --- /dev/null +++ b/internal/capability/plugin/manager_test.go @@ -0,0 +1,341 @@ +package plugin + +import ( + "bufio" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "sync" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry" + "git.yeij.top/AskaEth/Cyrene_v2/internal/config/runtimeconfig" +) + +// TestMCPStdioHelper is re-executed as the child process of a stdio install +// test (Command=os.Args[0], -test.run=TestMCPStdioHelper). It speaks the MCP +// JSON-RPC handshake on stdin/stdout and requires MCP_HELPER_TOKEN=ok, which +// proves the plugin config's Env reached the subprocess. Without the marker it +// returns immediately, so a normal test run ignores it. +func TestMCPStdioHelper(t *testing.T) { + if os.Getenv("GO_WANT_MCP_STDIO_HELPER") != "1" { + return + } + if os.Getenv("MCP_HELPER_TOKEN") != "ok" { + os.Exit(3) + } + sc := bufio.NewScanner(os.Stdin) + for sc.Scan() { + line := sc.Bytes() + var req struct { + ID json.RawMessage `json:"id"` + Method string `json:"method"` + } + if json.Unmarshal(line, &req) != nil || len(req.ID) == 0 { + continue + } + var result any + switch req.Method { + case "initialize": + result = map[string]any{"protocolVersion": "2025-11-25", "capabilities": map[string]any{}, "serverInfo": map[string]any{"name": "helper", "version": "1.0"}} + case "tools/list": + result = map[string]any{"tools": []map[string]any{{"name": "ping", "description": "ping", "inputSchema": map[string]any{"type": "object"}}}} + default: + result = map[string]any{} + } + resp, _ := json.Marshal(map[string]any{"jsonrpc": "2.0", "id": json.RawMessage(req.ID), "result": result}) + _, _ = fmt.Fprintln(os.Stdout, string(resp)) + } +} + +// fakeMCP is a minimal Streamable HTTP MCP server: it answers initialize and +// tools/list over POST and keeps a standalone GET SSE stream open until the +// request context ends. wantAuth, when set, requires the bearer header. +type fakeMCP struct { + wantAuth string + tools []string + + mu sync.Mutex + auth string +} + +func (f *fakeMCP) setTools(tools ...string) { + f.mu.Lock() + defer f.mu.Unlock() + f.tools = tools +} + +func (f *fakeMCP) authHeader() string { + f.mu.Lock() + defer f.mu.Unlock() + return f.auth +} + +func (f *fakeMCP) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodGet { + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + if fl, ok := w.(http.Flusher); ok { + fl.Flush() + } + <-r.Context().Done() + return + } + f.mu.Lock() + f.auth = r.Header.Get("Authorization") + tools := append([]string(nil), f.tools...) + f.mu.Unlock() + if f.wantAuth != "" && r.Header.Get("Authorization") != "Bearer "+f.wantAuth { + w.WriteHeader(http.StatusUnauthorized) + return + } + body, _ := io.ReadAll(r.Body) + var req struct { + ID json.RawMessage `json:"id"` + Method string `json:"method"` + } + _ = json.Unmarshal(body, &req) + if len(req.ID) == 0 || string(req.ID) == "null" { + w.WriteHeader(http.StatusAccepted) + return + } + var result any + switch req.Method { + case "initialize": + result = map[string]any{ + "protocolVersion": "2025-11-25", + "capabilities": map[string]any{}, + "serverInfo": map[string]any{"name": "fake", "version": "1.0"}, + } + case "tools/list": + list := make([]map[string]any, 0, len(tools)) + for _, name := range tools { + list = append(list, map[string]any{ + "name": name, + "description": name, + "inputSchema": map[string]any{"type": "object"}, + }) + } + result = map[string]any{"tools": list} + default: + result = map[string]any{} + } + resp, _ := json.Marshal(map[string]any{"jsonrpc": "2.0", "id": json.RawMessage(req.ID), "result": result}) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write(resp) +} + +func newFakeMCP(t *testing.T, wantAuth string, tools ...string) *httptest.Server { + t.Helper() + srv := httptest.NewServer(&fakeMCP{wantAuth: wantAuth, tools: tools}) + t.Cleanup(srv.Close) + return srv +} + +func toolNames(reg *registry.Registry) map[string]bool { + out := make(map[string]bool) + for _, c := range reg.List(registry.KindTool) { + out[c.Name] = true + } + return out +} + +func TestManagerInstall_RegistersToolsAndPersists(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "plugins.json") + reg := registry.New() + srv := newFakeMCP(t, "sk-1234", "add", "mul") + mgr := NewManager(reg, path, nil) + defer mgr.Close(context.Background()) + + got, err := mgr.Install(context.Background(), runtimeconfig.PluginConfig{ + Name: "calc", + Transport: runtimeconfig.TransportHTTP, + Endpoint: srv.URL, + APIKey: "sk-1234", + }) + if err != nil { + t.Fatalf("Install: %v", err) + } + if !got.Live || got.ToolCount != 2 { + t.Fatalf("installed = %+v, want live with 2 tools", got) + } + names := toolNames(reg) + if !names["add"] || !names["mul"] { + t.Fatalf("registered tools = %v, want add+mul", names) + } + if _, err := reg.Get("mcp:calc"); err != nil { + t.Fatalf("mcp:calc capability: %v", err) + } + + persisted, err := runtimeconfig.LoadPlugins(path) + if err != nil { + t.Fatalf("LoadPlugins: %v", err) + } + if len(persisted) != 1 || persisted[0].Name != "calc" || persisted[0].APIKey != "sk-1234" { + t.Fatalf("persisted = %+v", persisted) + } + if persisted[0].CreatedAt == "" || persisted[0].ID == "" { + t.Fatalf("persisted identity incomplete: %+v", persisted[0]) + } + if masked := runtimeconfig.MaskedPlugin(persisted[0]); masked.APIKey == "sk-1234" || masked.APIKey != "****1234" { + t.Fatalf("masked key = %q, want ****1234", masked.APIKey) + } + if auth := srv.Config.Handler.(*fakeMCP).authHeader(); auth != "Bearer sk-1234" { + t.Fatalf("server saw auth %q, want bearer key", auth) + } + if list := mgr.List(); len(list) != 1 || !list[0].Live || list[0].ToolCount != 2 { + t.Fatalf("List = %+v", list) + } +} + +func TestManagerUninstall_RemovesToolsAndConfig(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "plugins.json") + reg := registry.New() + srv := newFakeMCP(t, "", "add") + mgr := NewManager(reg, path, nil) + defer mgr.Close(context.Background()) + if _, err := mgr.Install(context.Background(), runtimeconfig.PluginConfig{Name: "calc", Endpoint: srv.URL}); err != nil { + t.Fatalf("Install: %v", err) + } + + removed, err := mgr.Uninstall(context.Background(), "calc") + if err != nil { + t.Fatalf("Uninstall: %v", err) + } + if !removed { + t.Fatal("removed = false, want true") + } + if names := toolNames(reg); len(names) != 0 { + t.Fatalf("tools after uninstall = %v, want none", names) + } + if _, err := reg.Get("mcp:calc"); err == nil { + t.Fatal("mcp:calc capability still registered") + } + persisted, err := runtimeconfig.LoadPlugins(path) + if err != nil { + t.Fatalf("LoadPlugins: %v", err) + } + if len(persisted) != 0 { + t.Fatalf("persisted = %+v, want empty", persisted) + } + if removed, err := mgr.Uninstall(context.Background(), "calc"); err != nil || removed { + t.Fatalf("second uninstall = %v, %v; want false, nil", removed, err) + } +} + +func TestManagerInstall_ConnectFailureDoesNotPersist(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "plugins.json") + reg := registry.New() + srv := httptest.NewServer(http.NotFoundHandler()) + url := srv.URL + srv.Close() + mgr := NewManager(reg, path, nil) + defer mgr.Close(context.Background()) + + if _, err := mgr.Install(context.Background(), runtimeconfig.PluginConfig{Name: "down", Endpoint: url}); err == nil { + t.Fatal("Install succeeded against a closed server") + } + if names := toolNames(reg); len(names) != 0 { + t.Fatalf("tools = %v, want none", names) + } + persisted, err := runtimeconfig.LoadPlugins(path) + if err != nil { + t.Fatalf("LoadPlugins: %v", err) + } + if len(persisted) != 0 { + t.Fatalf("persisted = %+v, want empty", persisted) + } + if list := mgr.List(); len(list) != 0 { + t.Fatalf("List = %+v, want empty", list) + } +} + +func TestManagerInstall_RefreshReplacesTools(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "plugins.json") + reg := registry.New() + srv := newFakeMCP(t, "", "add") + mgr := NewManager(reg, path, nil) + defer mgr.Close(context.Background()) + ctx := context.Background() + first, err := mgr.Install(ctx, runtimeconfig.PluginConfig{Name: "calc", Endpoint: srv.URL}) + if err != nil { + t.Fatalf("Install: %v", err) + } + srv.Config.Handler.(*fakeMCP).setTools("div") + + second, err := mgr.Install(ctx, runtimeconfig.PluginConfig{Name: "calc", Endpoint: srv.URL}) + if err != nil { + t.Fatalf("reinstall: %v", err) + } + if second.ToolCount != 1 { + t.Fatalf("toolCount = %d, want 1", second.ToolCount) + } + if second.Config.ID != first.Config.ID { + t.Fatalf("id changed on refresh: %q -> %q", first.Config.ID, second.Config.ID) + } + names := toolNames(reg) + if names["add"] { + t.Fatalf("stale tool add still registered: %v", names) + } + if !names["div"] { + t.Fatalf("new tool div missing: %v", names) + } + persisted, err := runtimeconfig.LoadPlugins(path) + if err != nil { + t.Fatalf("LoadPlugins: %v", err) + } + if len(persisted) != 1 { + t.Fatalf("persisted = %+v, want one entry", persisted) + } +} + +func TestManagerInstall_StdioWithEnv(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "plugins.json") + reg := registry.New() + mgr := NewManager(reg, path, nil) + defer mgr.Close(context.Background()) + + got, err := mgr.Install(context.Background(), runtimeconfig.PluginConfig{ + Name: "stdio", + Transport: runtimeconfig.TransportStdio, + Command: os.Args[0], + Args: []string{"-test.run=TestMCPStdioHelper"}, + Env: map[string]string{"GO_WANT_MCP_STDIO_HELPER": "1", "MCP_HELPER_TOKEN": "ok"}, + }) + if err != nil { + t.Fatalf("Install: %v", err) + } + if got.ToolCount != 1 || !toolNames(reg)["ping"] { + t.Fatalf("installed = %+v, tools = %v; want one ping tool", got, toolNames(reg)) + } +} + +func TestManagerRestore_ReinstallsPersistedPlugins(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "plugins.json") + srv := newFakeMCP(t, "", "add") + reg := registry.New() + mgr := NewManager(reg, path, nil) + if _, err := mgr.Install(context.Background(), runtimeconfig.PluginConfig{Name: "calc", Endpoint: srv.URL}); err != nil { + t.Fatalf("Install: %v", err) + } + mgr.Close(context.Background()) + + reg2 := registry.New() + mgr2 := NewManager(reg2, path, nil) + defer mgr2.Close(context.Background()) + mgr2.Restore(context.Background()) + if names := toolNames(reg2); !names["add"] { + t.Fatalf("restored tools = %v, want add", names) + } +} diff --git a/internal/config/runtimeconfig/plugins.go b/internal/config/runtimeconfig/plugins.go new file mode 100644 index 0000000..76b731a --- /dev/null +++ b/internal/config/runtimeconfig/plugins.go @@ -0,0 +1,255 @@ +// plugins.go loads and persists the runtime-installed external tool providers: +// the MCP server list in plugins.json. Unlike models.json/channels.json there is +// no env fallback — a plugin only exists because the owner installed it through +// the admin API, so an absent file simply means "none installed". +// +// plugins.json is gitignored (local runtime config) and may hold a server API +// key, so this package never logs a key and MaskedPlugin redacts every secret +// before a caller returns it to a client. +package runtimeconfig + +import ( + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "strings" +) + +// pluginsFileName is the plugins.json basename stored under the data dir. +const pluginsFileName = "plugins.json" + +// PluginKindMCP is the only installable plugin kind in v1: an external MCP +// server whose tools are hot-registered into the capability registry. A compiled +// Go plugin cannot be hot-installed in a single static binary (the project red +// line), so an MCP server is the runtime-installable unit. +const PluginKindMCP = "mcp" + +// Transport names accepted in a PluginConfig.Transport. +const ( + TransportStdio = "stdio" + TransportHTTP = "http" + TransportSSE = "sse" +) + +// ErrInvalidPlugin is returned when a plugins.json entry fails validation. +var ErrInvalidPlugin = errors.New("runtimeconfig: invalid plugin config") + +// PluginConfig describes one runtime-installed external provider. It is the +// on-disk plugins.json shape and the admin install body's config block. +type PluginConfig struct { + // ID is a stable identifier, generated when the panel omits one. + ID string `json:"id"` + // Name is the unique plugin label; it names the MCP server and prefixes its + // KindMCP capability ("mcp:"). + Name string `json:"name"` + // Kind classifies the provider. Only "mcp" is installable in v1. + Kind string `json:"kind"` + // Transport selects the wire transport: "stdio", "http" (Streamable HTTP) or + // "sse". Empty is inferred from Command/Endpoint. + Transport string `json:"transport,omitempty"` + // Endpoint is the server URL for http/sse transports. + Endpoint string `json:"endpoint,omitempty"` + // Command is the executable path for a stdio transport. + Command string `json:"command,omitempty"` + // Args are the stdio command arguments. + Args []string `json:"args,omitempty"` + // APIKey is the bearer credential sent to an http/sse server. It is + // persisted but never returned or logged verbatim (MaskedPlugin). + APIKey string `json:"apiKey,omitempty"` + // Env is extra environment for a stdio subprocess. Values under a sensitive + // key (token/secret/password/key...) are redacted by MaskedPlugin. + Env map[string]string `json:"env,omitempty"` + // CreatedAt is the RFC3339 install time. + CreatedAt string `json:"createdAt,omitempty"` +} + +// DefaultPluginsPath returns the plugins.json path under dataDir. +func DefaultPluginsPath(dataDir string) string { + return filepath.Join(dataDir, pluginsFileName) +} + +// Normalize fills the defaults an install request may omit: kind=mcp and the +// transport inferred from command/endpoint (stdio when a command is set, +// otherwise http when an endpoint is set). +func (p PluginConfig) Normalize() PluginConfig { + if strings.TrimSpace(p.Kind) == "" { + p.Kind = PluginKindMCP + } + if strings.TrimSpace(p.Transport) == "" { + switch { + case strings.TrimSpace(p.Command) != "": + p.Transport = TransportStdio + case strings.TrimSpace(p.Endpoint) != "": + p.Transport = TransportHTTP + } + } + return p +} + +// Validate checks the plugin's required fields. It validates the normalized +// form so a caller that only fills endpoint or command still gets a coherent +// transport. It never includes a secret in the returned error. +func (p PluginConfig) Validate() error { + p = p.Normalize() + if strings.TrimSpace(p.Name) == "" { + return fmt.Errorf("%w: name is required", ErrInvalidPlugin) + } + if p.Kind != PluginKindMCP { + return fmt.Errorf("%w: kind %q must be %s", ErrInvalidPlugin, p.Kind, PluginKindMCP) + } + switch p.Transport { + case TransportStdio: + if strings.TrimSpace(p.Command) == "" { + return fmt.Errorf("%w: command is required for %s transport", ErrInvalidPlugin, TransportStdio) + } + case TransportHTTP, "streamable", TransportSSE: + if strings.TrimSpace(p.Endpoint) == "" { + return fmt.Errorf("%w: endpoint is required for %s transport", ErrInvalidPlugin, p.Transport) + } + default: + return fmt.Errorf("%w: transport %q must be stdio, http, streamable or sse", ErrInvalidPlugin, p.Transport) + } + return nil +} + +// LoadPlugins returns the persisted plugin list. A missing file is the normal +// "nothing installed" case and yields an empty list with no error. A +// present-but-invalid file returns a non-nil error so the caller can warn +// instead of silently dropping installed plugins. +func LoadPlugins(path string) ([]PluginConfig, error) { + raw, err := os.ReadFile(path) + if err != nil { + if os.IsNotExist(err) { + return []PluginConfig{}, nil + } + return nil, fmt.Errorf("runtimeconfig: read %s: %w", path, err) + } + if len(strings.TrimSpace(string(raw))) == 0 { + return nil, fmt.Errorf("%w: empty file", ErrInvalidPlugin) + } + var plugins []PluginConfig + if err := json.Unmarshal(raw, &plugins); err != nil { + return nil, fmt.Errorf("%w: decode: %w", ErrInvalidPlugin, err) + } + for i, p := range plugins { + if err := p.Validate(); err != nil { + return nil, fmt.Errorf("%w: entry %d: %w", ErrInvalidPlugin, i, err) + } + } + if plugins == nil { + plugins = []PluginConfig{} + } + return plugins, nil +} + +// SavePlugins writes plugins as a pretty-printed JSON array to path with mode +// 0600 (the file may hold an API key). It writes to a temp file then renames so +// a concurrent reader never sees a partial file. Every entry is validated first. +func SavePlugins(path string, plugins []PluginConfig) error { + dir := filepath.Dir(path) + if dir != "" && dir != "." { + if err := os.MkdirAll(dir, 0o755); err != nil { + return fmt.Errorf("runtimeconfig: mkdir: %w", err) + } + } + out := make([]PluginConfig, 0, len(plugins)) + for i, p := range plugins { + p = p.Normalize() + if err := p.Validate(); err != nil { + return fmt.Errorf("runtimeconfig: save: entry %d: %w", i, err) + } + out = append(out, p) + } + raw, err := json.MarshalIndent(out, "", " ") + if err != nil { + return fmt.Errorf("runtimeconfig: marshal: %w", err) + } + tmp := path + ".tmp" + if err := os.WriteFile(tmp, raw, 0o600); err != nil { + return fmt.Errorf("runtimeconfig: write %s: %w", tmp, err) + } + if err := os.Rename(tmp, path); err != nil { + return fmt.Errorf("runtimeconfig: rename: %w", err) + } + return nil +} + +// UpsertPlugin returns plugins with p inserted or replaced in place by name +// (falling back to id), preserving order. A blank id is assigned a generated one +// so a panel save without an explicit id still produces a stable, addressable +// entry. +func UpsertPlugin(plugins []PluginConfig, p PluginConfig) []PluginConfig { + p = p.Normalize() + if strings.TrimSpace(p.ID) == "" { + p.ID = NewID() + } + out := make([]PluginConfig, 0, len(plugins)+1) + replaced := false + for _, cur := range plugins { + if samePlugin(cur, p) { + out = append(out, p) + replaced = true + continue + } + out = append(out, cur) + } + if !replaced { + out = append(out, p) + } + return out +} + +// DeletePlugin returns plugins with the entry matching name (by name or id) +// removed. The bool reports whether an entry was actually removed. +func DeletePlugin(plugins []PluginConfig, name string) ([]PluginConfig, bool) { + out := make([]PluginConfig, 0, len(plugins)) + removed := false + for _, cur := range plugins { + if cur.Name == name || cur.ID == name { + removed = true + continue + } + out = append(out, cur) + } + return out, removed +} + +// MaskedPlugin returns a copy of p whose API key and sensitive env values are +// redacted, for a wire response or a log line. The on-disk plugins.json is +// untouched. +func MaskedPlugin(p PluginConfig) PluginConfig { + out := p + if strings.TrimSpace(p.APIKey) != "" { + out.APIKey = maskToken(strings.TrimSpace(p.APIKey)) + } + out.Env = maskEnv(p.Env) + return out +} + +// maskEnv redacts the values of sensitive env keys, passing non-sensitive keys, +// empty values and non-string values through unchanged. +func maskEnv(env map[string]string) map[string]string { + if env == nil { + return nil + } + out := make(map[string]string, len(env)) + for k, v := range env { + if isSensitiveKey(k) && strings.TrimSpace(v) != "" { + out[k] = maskToken(v) + continue + } + out[k] = v + } + return out +} + +// samePlugin reports whether cur is the entry p replaces: same name when p has +// one, otherwise same id. +func samePlugin(cur, p PluginConfig) bool { + if p.Name != "" && cur.Name == p.Name { + return true + } + return p.ID != "" && cur.ID == p.ID +} diff --git a/internal/config/runtimeconfig/plugins_test.go b/internal/config/runtimeconfig/plugins_test.go new file mode 100644 index 0000000..039ca13 --- /dev/null +++ b/internal/config/runtimeconfig/plugins_test.go @@ -0,0 +1,181 @@ +package runtimeconfig + +import ( + "encoding/json" + "errors" + "os" + "path/filepath" + "strings" + "testing" +) + +func TestLoadPlugins_MissingFileIsEmpty(t *testing.T) { + t.Parallel() + got, err := LoadPlugins(filepath.Join(t.TempDir(), "plugins.json")) + if err != nil { + t.Fatalf("LoadPlugins: %v", err) + } + if len(got) != 0 { + t.Fatalf("len = %d, want 0", len(got)) + } +} + +func TestSaveLoadPlugins_RoundTrip(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "nested", "plugins.json") + in := []PluginConfig{{ + ID: "p1", + Name: "calc", + Kind: PluginKindMCP, + Transport: TransportHTTP, + Endpoint: "http://127.0.0.1:3001/mcp", + APIKey: "sk-secret-1234", + CreatedAt: "2026-09-08T00:00:00Z", + }} + if err := SavePlugins(path, in); err != nil { + t.Fatalf("SavePlugins: %v", err) + } + info, err := os.Stat(path) + if err != nil { + t.Fatalf("stat: %v", err) + } + if perm := info.Mode().Perm(); perm != 0o600 { + t.Fatalf("mode = %o, want 600", perm) + } + got, err := LoadPlugins(path) + if err != nil { + t.Fatalf("LoadPlugins: %v", err) + } + if len(got) != 1 || got[0].Name != "calc" || got[0].APIKey != "sk-secret-1234" { + t.Fatalf("got = %+v", got) + } +} + +func TestLoadPlugins_EmptyArrayIsValid(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "plugins.json") + if err := os.WriteFile(path, []byte("[]"), 0o600); err != nil { + t.Fatalf("write: %v", err) + } + got, err := LoadPlugins(path) + if err != nil { + t.Fatalf("LoadPlugins: %v", err) + } + if len(got) != 0 { + t.Fatalf("len = %d, want 0", len(got)) + } +} + +func TestLoadPlugins_InvalidReturnsError(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "plugins.json") + if err := os.WriteFile(path, []byte(`[{"name":""}]`), 0o600); err != nil { + t.Fatalf("write: %v", err) + } + if _, err := LoadPlugins(path); !errors.Is(err, ErrInvalidPlugin) { + t.Fatalf("err = %v, want ErrInvalidPlugin", err) + } +} + +func TestValidatePlugin(t *testing.T) { + t.Parallel() + cases := []struct { + name string + cfg PluginConfig + wantErr bool + }{ + {name: "stdio ok", cfg: PluginConfig{Name: "a", Command: "mcp-a"}}, + {name: "http ok", cfg: PluginConfig{Name: "b", Endpoint: "http://x/mcp"}}, + {name: "sse ok", cfg: PluginConfig{Name: "c", Transport: TransportSSE, Endpoint: "http://x/sse"}}, + {name: "blank name", cfg: PluginConfig{Command: "x"}, wantErr: true}, + {name: "bad kind", cfg: PluginConfig{Name: "d", Kind: "tool", Command: "x"}, wantErr: true}, + {name: "stdio no command", cfg: PluginConfig{Name: "e", Transport: TransportStdio}, wantErr: true}, + {name: "http no endpoint", cfg: PluginConfig{Name: "f", Transport: TransportHTTP}, wantErr: true}, + {name: "bad transport", cfg: PluginConfig{Name: "g", Transport: "carrier-pigeon", Endpoint: "http://x"}, wantErr: true}, + {name: "no transport no fields", cfg: PluginConfig{Name: "h"}, wantErr: true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + err := tc.cfg.Validate() + if tc.wantErr && err == nil { + t.Fatal("want error, got nil") + } + if !tc.wantErr && err != nil { + t.Fatalf("unexpected error: %v", err) + } + }) + } +} + +func TestUpsertPlugin_ByNameReplacesAndPreservesOrder(t *testing.T) { + t.Parallel() + base := []PluginConfig{ + {ID: "1", Name: "a", Command: "a"}, + {ID: "2", Name: "b", Command: "b"}, + } + got := UpsertPlugin(base, PluginConfig{Name: "b", Command: "b2"}) + if len(got) != 2 { + t.Fatalf("len = %d, want 2", len(got)) + } + if got[1].Name != "b" || got[1].Command != "b2" { + t.Fatalf("got[1] = %+v", got[1]) + } + got = UpsertPlugin(got, PluginConfig{Name: "c", Command: "c"}) + if len(got) != 3 || got[2].Name != "c" || got[2].ID == "" { + t.Fatalf("append = %+v", got) + } +} + +func TestDeletePlugin(t *testing.T) { + t.Parallel() + base := []PluginConfig{{ID: "1", Name: "a", Command: "a"}, {ID: "2", Name: "b", Command: "b"}} + got, removed := DeletePlugin(base, "a") + if !removed || len(got) != 1 || got[0].Name != "b" { + t.Fatalf("removed=%v got=%+v", removed, got) + } + got, removed = DeletePlugin(base, "2") + if !removed || len(got) != 1 || got[0].Name != "a" { + t.Fatalf("by id: removed=%v got=%+v", removed, got) + } + if _, removed := DeletePlugin(base, "nope"); removed { + t.Fatal("removed = true for unknown name") + } +} + +func TestMaskedPlugin_EmptySecretsStayEmpty(t *testing.T) { + t.Parallel() + got := MaskedPlugin(PluginConfig{Name: "calc", Env: map[string]string{"API_KEY": ""}}) + if got.APIKey != "" { + t.Fatalf("apiKey = %q, want empty", got.APIKey) + } + if got.Env["API_KEY"] != "" { + t.Fatalf("empty env secret = %q, want empty", got.Env["API_KEY"]) + } +} + +func TestMaskedPlugin_RedactsKeyAndEnv(t *testing.T) { + t.Parallel() + p := PluginConfig{ + Name: "calc", + APIKey: "sk-secret-1234", + Env: map[string]string{"PATH": "/usr/bin", "GITHUB_TOKEN": "ghp-abcdefgh"}, + } + got := MaskedPlugin(p) + if got.APIKey != "****1234" { + t.Fatalf("apiKey = %q, want ****1234", got.APIKey) + } + if got.Env["PATH"] != "/usr/bin" { + t.Fatalf("PATH = %q, want unchanged", got.Env["PATH"]) + } + if got.Env["GITHUB_TOKEN"] != "****efgh" { + t.Fatalf("GITHUB_TOKEN = %q, want masked", got.Env["GITHUB_TOKEN"]) + } + if p.APIKey != "sk-secret-1234" { + t.Fatal("original mutated") + } + raw, _ := json.Marshal(got) + if strings.Contains(string(raw), "sk-secret-1234") || strings.Contains(string(raw), "ghp-abcdefgh") { + t.Fatalf("masked JSON leaks a secret: %s", raw) + } +} diff --git a/internal/services/gateway/admin_plugins.go b/internal/services/gateway/admin_plugins.go new file mode 100644 index 0000000..185c098 --- /dev/null +++ b/internal/services/gateway/admin_plugins.go @@ -0,0 +1,238 @@ +package gateway + +// Admin plugin install/uninstall and the installed-plugin listing. A runtime +// "plugin" in v1 is an external MCP server: install connects it, hot-registers +// its tools into the capability registry (so they are injected into the next +// model request without a restart), persists its config to the gitignored +// plugins.json, and holds the live server for a later refresh/uninstall +// (design/09 插件生态 §1). Installing a compiled Go plugin is impossible in a +// single static binary, so it is not offered. Writes are owner-only and audited; +// an API key is persisted but never returned or logged verbatim. + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "strings" + "time" + + pluginmgr "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/plugin" + capreg "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry" + "git.yeij.top/AskaEth/Cyrene_v2/internal/config/runtimeconfig" + apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors" +) + +// pluginInstallTimeout bounds the connect+discover handshake the install request +// waits for. The plugin manager applies its own per-server timeout inside it. +const pluginInstallTimeout = 20 * time.Second + +// pluginActionRequest is the POST /api/v1/admin/plugins/uninstall body. +type pluginActionRequest struct { + Name string `json:"name"` + Kind string `json:"kind"` +} + +// pluginInstallRequest is the POST /api/v1/admin/plugins/install body. Config +// carries the MCP server connection fields; the top-level name/kind are kept +// separate so the request reads like the other admin writes. +type pluginInstallRequest struct { + Kind string `json:"kind"` + Name string `json:"name"` + Config pluginInstallConfig `json:"config"` +} + +// pluginInstallConfig is the MCP server connection block. Endpoint is the URL +// for http/sse; url is accepted as an alias. APIKey/Env are persisted but never +// echoed back. +type pluginInstallConfig struct { + Transport string `json:"transport"` + Endpoint string `json:"endpoint"` + URL string `json:"url"` + Command string `json:"command"` + Args []string `json:"args"` + APIKey string `json:"apiKey"` + Env map[string]string `json:"env"` +} + +// AdminInstalledPlugin is the wire shape of one installed plugin in the admin +// list. The API key and any sensitive env values are masked. +type AdminInstalledPlugin struct { + ID string `json:"id"` + Name string `json:"name"` + Kind string `json:"kind"` + Transport string `json:"transport"` + Endpoint string `json:"endpoint,omitempty"` + Command string `json:"command,omitempty"` + Args []string `json:"args,omitempty"` + APIKey string `json:"apiKey,omitempty"` + Env map[string]string `json:"env,omitempty"` + CreatedAt string `json:"createdAt,omitempty"` + // Live reports whether the MCP server is connected and its tools registered. + Live bool `json:"live"` + // ToolCount is the number of tools the live server registered. + ToolCount int `json:"toolCount"` +} + +// adminPluginInstall installs (or refreshes) an MCP server plugin: it validates +// the request, connects and discovers the server's tools, then persists the +// config. A connect/discover/register failure returns 502 and persists nothing, +// so a plugin is never half-installed. +func (s *Server) adminPluginInstall(w http.ResponseWriter, r *http.Request) { + var req pluginInstallRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + s.auditClaims(r, "plugin.install", "error") + writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "invalid request body")) + return + } + name := strings.TrimSpace(req.Name) + if name == "" { + s.auditClaims(r, "plugin.install", "error") + writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "name is required")) + return + } + kind := strings.TrimSpace(req.Kind) + if kind == "" { + kind = runtimeconfig.PluginKindMCP + } + if kind != runtimeconfig.PluginKindMCP { + s.auditClaims(r, "plugin.install", "error") + writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "kind %q is not installable; only %s", kind, runtimeconfig.PluginKindMCP)) + return + } + if s.plugins == nil { + s.auditClaims(r, "plugin.install", "error") + writeError(w, &apperr.AppError{Code: "not_implemented", Message: "插件服务未注入（未注入 plugin manager）", HTTPCode: http.StatusNotImplemented}) + return + } + cfg := runtimeconfig.PluginConfig{ + Name: name, + Kind: kind, + Transport: strings.TrimSpace(req.Config.Transport), + Endpoint: firstNonEmpty(req.Config.Endpoint, req.Config.URL), + Command: strings.TrimSpace(req.Config.Command), + Args: req.Config.Args, + APIKey: req.Config.APIKey, + Env: req.Config.Env, + }.Normalize() + if err := cfg.Validate(); err != nil { + s.auditClaims(r, "plugin.install", "error") + writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "%v", err)) + return + } + ctx, cancel := context.WithTimeout(r.Context(), pluginInstallTimeout) + defer cancel() + res, err := s.plugins.Install(ctx, cfg) + if err != nil { + s.auditClaims(r, "plugin.install", "error") + s.log.Error("plugin install failed", "name", name, "transport", cfg.Transport, "error", err) + writeError(w, &apperr.AppError{ + Code: "plugin_install_failed", + Message: fmt.Sprintf("连接/发现 MCP server %q 失败，未落盘", name), + Cause: err, + HTTPCode: http.StatusBadGateway, + }) + return + } + s.auditClaims(r, "plugin.install", "ok") + writeJSON(w, http.StatusOK, map[string]any{ + "installed": true, + "name": res.Config.Name, + "kind": res.Config.Kind, + "transport": res.Config.Transport, + "toolCount": res.ToolCount, + }) +} + +// adminPluginUninstall uninstalls a plugin. An installed MCP server is closed and +// its tools removed through the plugin manager, then its config is deleted from +// plugins.json; any other name falls back to removing the capability from the +// registry, preserving the pre-existing behavior. +func (s *Server) adminPluginUninstall(w http.ResponseWriter, r *http.Request) { + var req pluginActionRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "invalid request body")) + return + } + name := strings.TrimSpace(req.Name) + if name == "" { + s.auditClaims(r, "plugin.uninstall", "error") + writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "name is required")) + return + } + if s.plugins != nil { + removed, err := s.plugins.Uninstall(r.Context(), name) + if err != nil { + s.auditClaims(r, "plugin.uninstall", "error") + s.log.Error("plugin uninstall failed", "name", name, "error", err) + writeError(w, apperr.Wrap(apperr.ErrInternal, "plugin_remove_failed", "failed to uninstall plugin")) + return + } + if removed { + s.auditClaims(r, "plugin.uninstall", "ok") + writeJSON(w, http.StatusOK, map[string]any{"name": name, "uninstalled": true, "kind": runtimeconfig.PluginKindMCP}) + return + } + } + reg, ok := s.caps.(*capreg.Registry) + if !ok { + s.auditClaims(r, "plugin.uninstall", "error") + writeError(w, apperr.Wrap(apperr.ErrInternal, "plugins_unavailable", "capability registry is not configured")) + return + } + if err := reg.Remove(name); err != nil { + s.auditClaims(r, "plugin.uninstall", "error") + if errors.Is(err, capreg.ErrNotFound) { + writeError(w, apperr.Wrap(apperr.ErrNotFound, "not_found", "plugin %q not found", name)) + return + } + writeError(w, apperr.Wrap(apperr.ErrInternal, "plugin_remove_failed", "failed to uninstall plugin")) + return + } + s.auditClaims(r, "plugin.uninstall", "ok") + writeJSON(w, http.StatusOK, map[string]any{"name": name, "uninstalled": true}) +} + +// adminListInstalledPlugins lists the runtime-installed plugins with their +// masked config and per-server tool count. It is a read endpoint (owner/admin) +// and never returns an API key or a sensitive env value verbatim. +func (s *Server) adminListInstalledPlugins(w http.ResponseWriter, r *http.Request) { + out := make([]AdminInstalledPlugin, 0) + if s.plugins != nil { + for _, it := range s.plugins.List() { + out = append(out, adminInstalledPluginOf(it)) + } + } + writeJSON(w, http.StatusOK, out) +} + +// adminInstalledPluginOf maps an installed plugin to its masked wire shape. +func adminInstalledPluginOf(it pluginmgr.Installed) AdminInstalledPlugin { + cfg := runtimeconfig.MaskedPlugin(it.Config) + return AdminInstalledPlugin{ + ID: cfg.ID, + Name: cfg.Name, + Kind: cfg.Kind, + Transport: cfg.Transport, + Endpoint: cfg.Endpoint, + Command: cfg.Command, + Args: cfg.Args, + APIKey: cfg.APIKey, + Env: cfg.Env, + CreatedAt: cfg.CreatedAt, + Live: it.Live, + ToolCount: it.ToolCount, + } +} + +// firstNonEmpty returns the first non-blank value, so an install body may send +// either endpoint or url for the server address. +func firstNonEmpty(values ...string) string { + for _, v := range values { + if strings.TrimSpace(v) != "" { + return strings.TrimSpace(v) + } + } + return "" +} diff --git a/internal/services/gateway/admin_plugins_test.go b/internal/services/gateway/admin_plugins_test.go new file mode 100644 index 0000000..adf2d55 --- /dev/null +++ b/internal/services/gateway/admin_plugins_test.go @@ -0,0 +1,301 @@ +package gateway + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "path/filepath" + "strings" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" + "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" + pluginmgr "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/plugin" + capreg "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry" + "git.yeij.top/AskaEth/Cyrene_v2/internal/config" + "git.yeij.top/AskaEth/Cyrene_v2/internal/config/runtimeconfig" +) + +// fakeMCPHandler is a minimal Streamable HTTP MCP server for the plugin tests. +// wantAuth, when set, requires the bearer header; tools is mutated atomically so +// a refresh test can change the tool list. +type fakeMCPHandler struct { + wantAuth string + tools []string +} + +func (f *fakeMCPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodGet { + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + if fl, ok := w.(http.Flusher); ok { + fl.Flush() + } + <-r.Context().Done() + return + } + if f.wantAuth != "" && r.Header.Get("Authorization") != "Bearer "+f.wantAuth { + w.WriteHeader(http.StatusUnauthorized) + return + } + body, _ := io.ReadAll(r.Body) + var req struct { + ID json.RawMessage `json:"id"` + Method string `json:"method"` + } + _ = json.Unmarshal(body, &req) + if len(req.ID) == 0 || string(req.ID) == "null" { + w.WriteHeader(http.StatusAccepted) + return + } + var result any + switch req.Method { + case "initialize": + result = map[string]any{"protocolVersion": "2025-11-25", "capabilities": map[string]any{}, "serverInfo": map[string]any{"name": "fake", "version": "1.0"}} + case "tools/list": + list := make([]map[string]any, 0, len(f.tools)) + for _, name := range f.tools { + list = append(list, map[string]any{"name": name, "description": name, "inputSchema": map[string]any{"type": "object"}}) + } + result = map[string]any{"tools": list} + default: + result = map[string]any{} + } + resp, _ := json.Marshal(map[string]any{"jsonrpc": "2.0", "id": json.RawMessage(req.ID), "result": result}) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write(resp) +} + +// pluginTestEnv is a gateway wired with a real plugin manager over a temp +// plugins.json. +type pluginTestEnv struct { + ts *httptest.Server + caps *capreg.Registry + path string + token string + mcpURL string +} + +func newPluginTestEnv(t *testing.T, mcp *fakeMCPHandler) *pluginTestEnv { + t.Helper() + mcpURL := "" + if mcp != nil { + ts := httptest.NewServer(mcp) + t.Cleanup(ts.Close) + mcpURL = ts.URL + } + jwt := auth.New(config.AuthConfig{JWTSecret: testSecret}) + caps := capreg.New() + path := filepath.Join(t.TempDir(), "plugins.json") + mgr := pluginmgr.NewManager(caps, path, nil) + cfg := &config.Config{Auth: config.AuthConfig{AdminUser: "admin", AdminPass: "secret", JWTSecret: testSecret}} + srv, err := New(cfg, Deps{Auth: jwt, Bus: bus.New(), Registry: caps, Plugins: mgr}) + if err != nil { + t.Fatalf("new server: %v", err) + } + ts := httptest.NewServer(srv.Handler()) + t.Cleanup(func() { + ts.Close() + mgr.Close(context.Background()) + }) + tok, err := jwt.Sign("acct", auth.RoleOwner, time.Hour) + if err != nil { + t.Fatalf("sign: %v", err) + } + return &pluginTestEnv{ts: ts, caps: caps, path: path, token: tok, mcpURL: mcpURL} +} + +// newFakeMCPHandler returns a fake server plus a setter for its tool list. +func newFakeMCPHandler(wantAuth string, tools ...string) *fakeMCPHandler { + return &fakeMCPHandler{wantAuth: wantAuth, tools: tools} +} + +func installedTools(reg *capreg.Registry) map[string]bool { + out := make(map[string]bool) + for _, c := range reg.List(capreg.KindTool) { + out[c.Name] = true + } + return out +} + +func TestAdminPluginInstall_RegistersToolsPersistsAndMasks(t *testing.T) { + t.Parallel() + mcp := newFakeMCPHandler("sk-1234", "add", "mul") + env := newPluginTestEnv(t, mcp) + resp, raw := doReq(t, http.MethodPost, env.ts.URL+"/api/v1/admin/plugins/install", map[string]any{ + "kind": "mcp", + "name": "calc", + "config": map[string]any{ + "transport": "http", + "endpoint": env.mcpURL, + "apiKey": "sk-1234", + }, + }, env.token) + if resp.StatusCode != http.StatusOK { + t.Fatalf("install status = %d, body = %s", resp.StatusCode, raw) + } + var envelope struct { + Data struct { + Installed bool `json:"installed"` + Name string `json:"name"` + Transport string `json:"transport"` + ToolCount int `json:"toolCount"` + } `json:"data"` + } + if err := json.Unmarshal(raw, &envelope); err != nil { + t.Fatalf("decode: %v (%s)", err, raw) + } + if !envelope.Data.Installed || envelope.Data.ToolCount != 2 || envelope.Data.Transport != "http" { + t.Fatalf("install data = %+v", envelope.Data) + } + names := installedTools(env.caps) + if !names["add"] || !names["mul"] { + t.Fatalf("registered tools = %v", names) + } + if _, err := env.caps.Get("mcp:calc"); err != nil { + t.Fatalf("mcp:calc capability: %v", err) + } + persisted, err := runtimeconfig.LoadPlugins(env.path) + if err != nil || len(persisted) != 1 || persisted[0].APIKey != "sk-1234" { + t.Fatalf("persisted = %+v, err = %v", persisted, err) + } + + resp, raw = doReq(t, http.MethodGet, env.ts.URL+"/api/v1/admin/plugins/installed", nil, env.token) + if resp.StatusCode != http.StatusOK { + t.Fatalf("list status = %d, body = %s", resp.StatusCode, raw) + } + if strings.Contains(string(raw), "sk-1234") { + t.Fatalf("installed list leaks the API key: %s", raw) + } + var listEnv struct { + Data []map[string]any `json:"data"` + } + if err := json.Unmarshal(raw, &listEnv); err != nil { + t.Fatalf("decode list: %v (%s)", err, raw) + } + if len(listEnv.Data) != 1 { + t.Fatalf("list = %+v, want one entry", listEnv.Data) + } + if got := listEnv.Data[0]["apiKey"]; got != "****1234" { + t.Fatalf("masked apiKey = %v, want ****1234", got) + } + if got := listEnv.Data[0]["live"]; got != true { + t.Fatalf("live = %v, want true", got) + } +} + +func TestAdminPluginUninstall_RemovesToolsAndConfig(t *testing.T) { + t.Parallel() + mcp := newFakeMCPHandler("", "add") + env := newPluginTestEnv(t, mcp) + resp, raw := doReq(t, http.MethodPost, env.ts.URL+"/api/v1/admin/plugins/install", map[string]any{ + "name": "calc", + "config": map[string]any{"endpoint": env.mcpURL}, + }, env.token) + if resp.StatusCode != http.StatusOK { + t.Fatalf("install status = %d, body = %s", resp.StatusCode, raw) + } + + resp, raw = doReq(t, http.MethodPost, env.ts.URL+"/api/v1/admin/plugins/uninstall", map[string]any{"name": "calc"}, env.token) + if resp.StatusCode != http.StatusOK { + t.Fatalf("uninstall status = %d, body = %s", resp.StatusCode, raw) + } + var envelope struct { + Data struct { + Uninstalled bool `json:"uninstalled"` + Kind string `json:"kind"` + } `json:"data"` + } + if err := json.Unmarshal(raw, &envelope); err != nil { + t.Fatalf("decode: %v (%s)", err, raw) + } + if !envelope.Data.Uninstalled || envelope.Data.Kind != runtimeconfig.PluginKindMCP { + t.Fatalf("uninstall data = %+v", envelope.Data) + } + if names := installedTools(env.caps); len(names) != 0 { + t.Fatalf("tools after uninstall = %v", names) + } + if _, err := env.caps.Get("mcp:calc"); err == nil { + t.Fatal("mcp:calc capability still registered") + } + persisted, err := runtimeconfig.LoadPlugins(env.path) + if err != nil || len(persisted) != 0 { + t.Fatalf("persisted = %+v, err = %v", persisted, err) + } +} + +func TestAdminPluginInstall_ConnectFailureReturns502NoPersist(t *testing.T) { + t.Parallel() + dead := httptest.NewServer(http.NotFoundHandler()) + url := dead.URL + dead.Close() + env := newPluginTestEnv(t, nil) + resp, raw := doReq(t, http.MethodPost, env.ts.URL+"/api/v1/admin/plugins/install", map[string]any{ + "name": "down", + "config": map[string]any{"endpoint": url}, + }, env.token) + if resp.StatusCode != http.StatusBadGateway { + t.Fatalf("status = %d, want 502, body = %s", resp.StatusCode, raw) + } + if names := installedTools(env.caps); len(names) != 0 { + t.Fatalf("tools = %v, want none", names) + } + persisted, err := runtimeconfig.LoadPlugins(env.path) + if err != nil || len(persisted) != 0 { + t.Fatalf("persisted = %+v, err = %v", persisted, err) + } +} + +func TestAdminPluginUninstall_FallsBackToCapabilityRemoval(t *testing.T) { + t.Parallel() + env := newPluginTestEnv(t, nil) + if err := env.caps.RegisterTool(stubTool{}); err != nil { + t.Fatalf("register tool: %v", err) + } + resp, raw := doReq(t, http.MethodPost, env.ts.URL+"/api/v1/admin/plugins/uninstall", map[string]any{"name": "echo"}, env.token) + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, body = %s", resp.StatusCode, raw) + } + if _, err := env.caps.Get("echo"); err == nil { + t.Fatal("echo capability still registered") + } + resp, _ = doReq(t, http.MethodPost, env.ts.URL+"/api/v1/admin/plugins/uninstall", map[string]any{"name": "nope"}, env.token) + if resp.StatusCode != http.StatusNotFound { + t.Fatalf("unknown status = %d, want 404", resp.StatusCode) + } +} + +func TestAdminPluginInstall_InvalidRequestsReturn400(t *testing.T) { + t.Parallel() + env := newPluginTestEnv(t, newFakeMCPHandler("")) + cases := []struct { + name string + body any + raw string + }{ + {name: "no name", body: map[string]any{"config": map[string]any{"command": "x"}}}, + {name: "bad kind", body: map[string]any{"kind": "tool", "name": "x", "config": map[string]any{"command": "x"}}}, + {name: "no transport source", body: map[string]any{"name": "x", "config": map[string]any{}}}, + {name: "stdio no command", body: map[string]any{"name": "x", "config": map[string]any{"transport": "stdio"}}}, + {name: "bad transport", body: map[string]any{"name": "x", "config": map[string]any{"transport": "pigeon", "endpoint": "http://x"}}}, + {name: "malformed json", raw: "{"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + var resp *http.Response + var raw []byte + if tc.raw != "" { + resp, raw = doRaw(t, http.MethodPost, env.ts.URL+"/api/v1/admin/plugins/install", tc.raw, env.token) + } else { + resp, raw = doReq(t, http.MethodPost, env.ts.URL+"/api/v1/admin/plugins/install", tc.body, env.token) + } + if resp.StatusCode != http.StatusBadRequest { + t.Fatalf("status = %d, want 400, body = %s", resp.StatusCode, raw) + } + }) + } +} diff --git a/internal/services/gateway/admin_write_plugins_memory_sessions.go b/internal/services/gateway/admin_write_plugins_memory_sessions.go index b8bc997..db16a65 100644 --- a/internal/services/gateway/admin_write_plugins_memory_sessions.go +++ b/internal/services/gateway/admin_write_plugins_memory_sessions.go @@ -3,66 +3,17 @@ package gateway import ( "context" "encoding/json" - "errors" "net/http" "strings" - capreg "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry" "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session" apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" ) -// Plugin uninstall, memory retrieval and session export admin writes -// (owner-only, audited). Uninstall removes a capability from the registry; memory -// search runs a real retrieval; session export returns the session metadata. - -// pluginActionRequest is the POST /api/v1/admin/plugins/install|uninstall body. -type pluginActionRequest struct { - Name string `json:"name"` - Kind string `json:"kind"` - Description string `json:"description"` -} - -// adminPluginInstall answers honestly: installing a plugin needs a real plugin -// source (compiled binary or skill YAML), which the admin API cannot supply. It -// returns a clear 501 rather than fabricating a registered plugin. -func (s *Server) adminPluginInstall(w http.ResponseWriter, r *http.Request) { - s.auditClaims(r, "plugin.install", "error") - writeError(w, &apperr.AppError{Code: "not_implemented", Message: "插件安装需插件源码清单（编译二进制/YAML）为 M10 提供；卸载可直接调用 registry", HTTPCode: http.StatusNotImplemented}) -} - -// adminPluginUninstall removes a registered capability from the registry — a real -// effect that stops the plugin being enumerated or dispatched. -func (s *Server) adminPluginUninstall(w http.ResponseWriter, r *http.Request) { - var req pluginActionRequest - 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.Name) == "" { - s.auditClaims(r, "plugin.uninstall", "error") - writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "name is required")) - return - } - reg, ok := s.caps.(*capreg.Registry) - if !ok { - s.auditClaims(r, "plugin.uninstall", "error") - writeError(w, apperr.Wrap(apperr.ErrInternal, "plugins_unavailable", "capability registry is not configured")) - return - } - if err := reg.Remove(req.Name); err != nil { - s.auditClaims(r, "plugin.uninstall", "error") - if errors.Is(err, capreg.ErrNotFound) { - writeError(w, apperr.Wrap(apperr.ErrNotFound, "not_found", "plugin %q not found", req.Name)) - return - } - writeError(w, apperr.Wrap(apperr.ErrInternal, "plugin_remove_failed", "failed to uninstall plugin")) - return - } - s.auditClaims(r, "plugin.uninstall", "ok") - writeJSON(w, http.StatusOK, map[string]any{"name": req.Name, "uninstalled": true}) -} +// Memory retrieval and session export admin writes (owner-only, audited). Memory +// search runs a real retrieval; session export returns the session metadata. The +// plugin install/uninstall writes live in admin_plugins.go. // memorySearchRequest is the POST /api/v1/admin/memories/search body. type memorySearchRequest struct { diff --git a/internal/services/gateway/server.go b/internal/services/gateway/server.go index 85db8e0..6500dbf 100644 --- a/internal/services/gateway/server.go +++ b/internal/services/gateway/server.go @@ -30,6 +30,7 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" + pluginmgr "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/plugin" capreg "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry" "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session" "git.yeij.top/AskaEth/Cyrene_v2/internal/config" @@ -293,6 +294,11 @@ type Deps struct { // Optional: when nil the admin /channels/config endpoints return a clear // not-configured error, matching how the LLM stack behaves when unset. Channels ChannelConfigStore + // Plugins is the runtime-installable external tool provider manager + // (plugins.json + live MCP servers). Optional: when nil install answers a + // clear not-implemented error and uninstall falls back to a plain capability + // removal, so pre-existing gateway tests keep working. + Plugins *pluginmgr.Manager // OwnerRegistrar stamps a scan-login's bound owner id into the owner // whitelist (configs/users/owner.yaml) and hot-updates the live profile. // Optional: when nil the weixin confirm flow still persists channels.json but @@ -364,6 +370,9 @@ type Server struct { // that do not wire it keep working. llm *model.RuntimeLLM channels ChannelConfigStore + // plugins is the runtime-installable external tool provider manager. See + // Deps.Plugins. + plugins *pluginmgr.Manager // ownerRegistrar stamps a scan-login bound owner id into the owner whitelist // (best-effort). See Deps.OwnerRegistrar. ownerRegistrar OwnerRegistrar @@ -405,6 +414,7 @@ func New(cfg *config.Config, deps Deps) (*Server, error) { frontend: deps.Frontend, llm: deps.LLM, channels: deps.Channels, + plugins: deps.Plugins, ownerRegistrar: deps.OwnerRegistrar, webhooks: deps.Webhooks, dbPath: deps.DatabasePath, @@ -472,6 +482,7 @@ func (s *Server) Handler() http.Handler { mux.Handle("GET /api/v1/admin/sessions", adminAuth(http.HandlerFunc(s.adminListSessions))) mux.Handle("GET /api/v1/admin/memories", adminAuth(http.HandlerFunc(s.adminListMemories))) mux.Handle("GET /api/v1/admin/plugins", adminAuth(http.HandlerFunc(s.adminListPlugins))) + mux.Handle("GET /api/v1/admin/plugins/installed", adminAuth(http.HandlerFunc(s.adminListInstalledPlugins))) mux.Handle("GET /api/v1/admin/channels", adminAuth(http.HandlerFunc(s.adminListChannels))) mux.Handle("GET /api/v1/admin/channels/messages", adminAuth(http.HandlerFunc(s.handleChannelMessages))) mux.Handle("GET /api/v1/admin/channels/{platform}/messages", adminAuth(http.HandlerFunc(s.handleChannelPlatformLog)))