From 0d61354ba1b991b55e40704f7586abd39ba780bd Mon Sep 17 00:00:00 2001 From: xiaoxixi Date: Tue, 21 Jul 2026 12:59:46 +0800 Subject: [PATCH] feat: add gateway configuration hot reload --- AGENTS.md | 2 + Cargo.toml | 2 +- README.md | 13 +- docker-compose.test.yml | 2 +- docs/ARCHITECTURE.md | 8 +- docs/CONFIG_HOT_RELOAD_DESIGN.md | 389 ++++++++++++++ docs/CONFIG_HOT_RELOAD_REVIEW.md | 369 +++++++++++++ .../skills/about-picobot/references/config.md | 4 +- src/agent/sub_agent.rs | 36 +- src/client/mod.rs | 24 + src/config/mod.rs | 42 +- src/gateway/http.rs | 46 +- src/gateway/mod.rs | 332 +++++++++--- src/gateway/reload.rs | 486 ++++++++++++++++++ src/gateway/router.rs | 88 +++- src/main.rs | 13 + src/scheduler/mod.rs | 29 ++ src/session/session.rs | 75 ++- src/tools/mod.rs | 2 + src/tools/reload_config.rs | 52 ++ tests/test_config_reload.rs | 146 ++++++ webui/package-lock.json | 4 +- webui/package.json | 2 +- 23 files changed, 2061 insertions(+), 105 deletions(-) create mode 100644 docs/CONFIG_HOT_RELOAD_DESIGN.md create mode 100644 docs/CONFIG_HOT_RELOAD_REVIEW.md create mode 100644 src/gateway/reload.rs create mode 100644 src/tools/reload_config.rs create mode 100644 tests/test_config_reload.rs diff --git a/AGENTS.md b/AGENTS.md index c3a4f5d..2af8154 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -8,6 +8,7 @@ This file is the operational contract for coding agents working in this reposito - `cargo run -- gateway` — start gateway server (binds `127.0.0.1:19876` by default) - `cargo run -- chat` — connect to gateway as CLI client (default `ws://127.0.0.1:19876/ws`) - `cargo run -- run "prompt"` — send one prompt through Gateway, print the terminal Turn, and exit; stdin, JSON, verbose progress, and timeout modes are available +- `cargo run -- reload` — validate and gracefully reload a running Gateway's configuration - `docker compose up -d` — start the container with Gateway bound/published on `0.0.0.0:19876`; override `PICOBOT_GATEWAY_HOST`, `PICOBOT_PUBLISH_HOST`, or `PICOBOT_GATEWAY_PORT` as needed - WebUI — start Gateway, then open `http://127.0.0.1:19876/`; no separate frontend build is required - `cd webui && npm ci && npm run check && npm run build` — validate the Svelte WebUI independently (Node.js 20+); its local `dist/` is ignored @@ -91,6 +92,7 @@ Scheduler → SessionManager scheduled execution → AgentLoop → Scheduler del - **Scheduler** supports legacy direct-delivery jobs and managed `task`/`monitor` jobs; managed agents cannot call `send_message`, and `on_alert` suppresses only healthy informational results - **AgentLoop** is stateless across turns; it receives prepared history, calls LLM providers, executes tools, and returns one result - **WebUI management APIs** only expose allowlisted config/profile/log/storage operations; keep response limits, secret redaction, atomic config writes, and profile path allowlists intact +- **Gateway config reload** validates and prepares a complete next runtime generation before retiring the old one; close runtime admission before draining inbound lanes, Sessions, Scheduler jobs, and background sub-agents; MCP connects only during activation; host/port, workspace, and effective storage path remain restart-only, and runtime reload must never mutate process environment variables - **WebUI slash completion** must consume the existing `get_slash_commands` WebSocket response; do not duplicate the backend command list in frontend source - **WebUI chat rendering** sanitizes Markdown before inserting HTML; durable `turn_committed` deltas calibrate normal terminal Turns without a full history reload, and history must preserve structured tool-call metadata so calls and results remain independently collapsible - **WebUI/TUI file transfer** streams bytes over authenticated HTTP and sends only short-lived upload IDs/attachment metadata over WebSocket; messages persist local media paths without guaranteeing later availability, and client responses must never expose those paths diff --git a/Cargo.toml b/Cargo.toml index b32d7e9..9b971f6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "picobot" -version = "1.2.2" +version = "1.3.0" edition = "2024" [dependencies] diff --git a/README.md b/README.md index 7dc48f1..73d8829 100644 --- a/README.md +++ b/README.md @@ -175,7 +175,7 @@ WebUI 随二进制嵌入，不需要 Node.js、npm 或单独部署静态文件 - 本地滚动日志的尾部查看、过滤和自动刷新。 - `config.json`、`~/.picobot/USER.md`、`~/.picobot/AGENTS.md` 编辑。 -配置接口会掩码 API Key、secret、password 和 token；保留 `********` 再保存不会覆盖原密钥。运行配置采用原子写入并在 Gateway 重启后生效，`USER.md` 和 `AGENTS.md` 则会用于后续构建的 Agent 上下文。 +配置接口会掩码 API Key、secret、password 和 token；保留 `********` 再保存不会覆盖原密钥。运行配置采用原子写入，保存后可执行 `picobot reload` 或发送 `/reload` 热重载；`USER.md` 和 `AGENTS.md` 的修改用于后续构建的 Agent 上下文。 WebUI 默认启用设备配对鉴权，管理 API 与 `/ws` 都拒绝未配对客户端；静态配对页、公开健康检查和配对提交接口除外。唯一的 WebSocket 例外是本机 `picobot run`：请求必须同时来自真实回环对端并持有权限为 `0600` 的 `~/.picobot/web_admin_token`，该管理令牌不能绕过任何管理 API 的设备鉴权。配对令牌只以 SHA-256 哈希写入 `~/.picobot/web_auth.json`。鉴权不提供传输加密；如果通过 `--host 0.0.0.0`、反向代理或端口转发暴露 Gateway，仍必须使用 TLS。可通过 `gateway.require_pairing=false` 显式关闭配对，但不建议在非隔离环境使用。 @@ -210,6 +210,14 @@ picobot service stop picobot service uninstall ``` +修改配置后无需重启 systemd service： + +```bash +picobot reload +``` + +该命令连接正在运行的 Gateway，先解析并校验新配置，再停止接收新工作，等待当前交互 Turn、Scheduler job 和后台子 Agent 到达安全边界后切换运行代。也可在聊天中发送 `/reload`，或让根交互 Agent 在用户明确要求时调用 `reload_config` 工具；子 Agent 与定时任务不能触发重载。监听地址、workspace 和数据库路径涉及进程级资源，不能热重载；修改这些字段时命令会保留旧配置并提示使用 `picobot service restart`。重载会主动断开 WebSocket，TUI/WebUI 随后可重新连接并从持久化历史恢复。受认证客户端可通过 `GET /api/config/reload/status` 查询 generation、切换阶段和最近错误。 + unit 位于 `~/.config/systemd/user/picobot.service`，以执行 `service install` 时的当前目录作为初始工作目录。服务异常退出时由 systemd 自动重启；`stop` 和 `restart` 会通过 SIGTERM 触发 Gateway 的有界优雅关停。 TUI 会把一个随机客户端标识保存到 `~/.picobot/tui_client_id`，因此关闭并重新打开客户端后会恢复同一组 dialog 和最近使用的会话。界面支持流式正文、独立思考与工具状态、历史回放、会话列表与归档筛选、命令补全、Unicode/中文编辑、括号粘贴、多行输入和文件传输。 @@ -292,6 +300,7 @@ Session ID 使用三段式： | `/mcp` | 查看 MCP 服务器和工具状态 | | `/stop` | 停止当前任务并清空队列 | | `/todo [done\|cancel]` | 查看、完成或取消当前 session 的任务计划 | +| `/reload` | 校验并重新加载 Gateway 配置 | | `/?`, `/help` | 查看帮助 | ### 记忆 @@ -318,6 +327,7 @@ PicoBot 有两类记忆： | `http_request` / `web_fetch` | HTTP 请求和网页文本抽取 | | `get_skill` | 列出或读取本地 Skill | | `memory_store` / `memory_recall` / `timeline_recall` / `memory_forget` | 长期记忆操作 | +| `reload_config` | 在用户明确要求时校验并重新加载 Gateway 配置 | | `delegate` | 启动 inline、background 或 parallel 子 Agent | | `todo` | 为复杂、多轮任务创建并更新当前 session 的持久化计划 | | `send_message` | 向指定渠道或当前会话发送消息，可附带文件/截图；WebUI/TUI 当前 Turn 的附件并入最终回复 | @@ -472,6 +482,7 @@ docs/ 面向维护者和 Agent 的架构与开发文档 ## 进一步阅读 - [维护者架构文档](docs/ARCHITECTURE.md) +- [配置热重载设计与实现](docs/CONFIG_HOT_RELOAD_DESIGN.md) - [WebUI 与 TUI 文件收发设计](docs/FILE_TRANSFER_DESIGN.md) - [内置 Skill：架构机制](resources/skills/about-picobot/references/architecture.md) - [配置说明](resources/skills/about-picobot/references/config.md) diff --git a/docker-compose.test.yml b/docker-compose.test.yml index 3b6e9df..17cae9c 100644 --- a/docker-compose.test.yml +++ b/docker-compose.test.yml @@ -3,7 +3,7 @@ services: build: context: . dockerfile: Dockerfile - image: picobot:1.2.2 + image: picobot:1.3.0 container_name: picobot-test restart: unless-stopped ports: diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 6048721..071bff4 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -2,7 +2,7 @@ 本文档描述 PicoBot 当前实现的运行时边界、数据流、并发模型和演进约束。它面向维护者和后续参与改进的 Agent，是代码架构的主入口；行为细节仍以代码和测试为最终依据。 -流式模型输出、reasoning 展示、活动 Turn 快照和 Channel 实时投递的详细设计与取舍见 [STREAMING_TURN_DESIGN.md](STREAMING_TURN_DESIGN.md)。用户输入路由、Session 执行拆分、终态投递确认和历史增量校准的重构方案见 [MESSAGE_FLOW_REFACTOR_DESIGN.md](MESSAGE_FLOW_REFACTOR_DESIGN.md)。 +流式模型输出、reasoning 展示、活动 Turn 快照和 Channel 实时投递的详细设计与取舍见 [STREAMING_TURN_DESIGN.md](STREAMING_TURN_DESIGN.md)。用户输入路由、Session 执行拆分、终态投递确认和历史增量校准的重构方案见 [MESSAGE_FLOW_REFACTOR_DESIGN.md](MESSAGE_FLOW_REFACTOR_DESIGN.md)。配置运行代、重载边界和失败语义见 [CONFIG_HOT_RELOAD_DESIGN.md](CONFIG_HOT_RELOAD_DESIGN.md)。 ## 1. 设计目标 @@ -28,13 +28,15 @@ PicoBot 只有一个二进制，提供三种运行模式： Linux 上可通过 `picobot service install/start/stop/status/restart/uninstall` 管理 systemd 用户服务。unit 固定为 `picobot.service`，其主进程仍是普通 Gateway 模式，不引入额外 daemon/fork 层；异常退出由 systemd 按 `Restart=on-failure` 拉起。 +`picobot reload`、`/reload` 和根交互 Agent 的 `reload_config` 工具共享 Gateway 内部的有界重载控制通道。重载先用启动前捕获的进程环境重新解析配置和 `.env`，构造完整的下一运行代；候选构造失败时旧运行代不变。成功后关闭旧代 admission，等待已进入的消息、交互 Turn、Scheduler job 和后台子 Agent 到达持久化/投递边界，再停止旧渠道和受监督任务、主动关闭旧 WebSocket，并在保留的监听 socket 上启动新运行代。MCP 只在新代激活时连接。每次重载有 generation ID，可通过 `GET /api/config/reload/status` 查询相位。监听 host/port、workspace 和 SQLite 有效路径属于进程级不变量，变更时拒绝热重载并要求完整重启。 + 原生 Gateway 默认绑定 `127.0.0.1:19876`，`gateway.host`/`gateway.port` 可由命令行 `--host`/`--port` 覆盖。Docker Compose 为保证端口映射可达，默认向容器传入 `0.0.0.0:19876`；`PICOBOT_GATEWAY_HOST` 控制容器内监听地址，`PICOBOT_PUBLISH_HOST` 控制宿主机发布地址，`PICOBOT_GATEWAY_PORT` 同时控制监听与映射端口。 CLI TUI 在 `~/.picobot/tui_client_id` 保存非敏感客户端标识，并通过 WebSocket 查询参数 `client_id` 传给 Gateway。`cli_chat` 以该标识作为稳定 chat scope；重连时恢复内存中的当前 dialog，Gateway 重启后则恢复该 scope 最近活跃的未归档 dialog。无效或缺失的标识会退化为连接级随机 scope。 One-shot client 不绕过 Gateway 直接调用 Provider。每次 `run` 生成独立的 `run-` scope，通过相同的 `cli_chat`、MessageBus、SessionManager、AgentLoop 和 Turn delivery 路径执行；它不复用 TUI scope，因而不会替换同一 scope 的活动 WebSocket。默认 stdout 只投影终态 Assistant blocks，进度写到 stderr；超时或 Ctrl-C 会先在当前 scope 发送 `/stop`。 -Gateway 启动时先从配置目录 `.env`、workspace `.env` 和既有进程环境合并启动变量，再初始化日志并切换进程工作目录到 `workspace_dir`。优先级为进程环境 > workspace `.env` > 配置目录 `.env`；配置目录层先用于定位 workspace，workspace 层不得重定向自身位置。环境文件只在单线程启动阶段写入进程环境，不能移到后台任务启动之后。切换完成后所有相对路径都应按 workspace 解释，不能假设仍位于源码仓库。 +Gateway 启动时先从配置目录 `.env`、workspace `.env` 和既有进程环境合并启动变量，再初始化日志并切换进程工作目录到 `workspace_dir`。优先级为进程环境 > workspace `.env` > 配置目录 `.env`；配置目录层先用于定位 workspace，workspace 层不得重定向自身位置。环境文件只在单线程启动阶段写入进程环境，不能移到后台任务启动之后。重载使用启动前保存的进程环境快照解析各层，但不再修改进程环境，避免多线程运行期调用 `set_var`；新 Provider、MCP 和渠道使用解析后配置中的值。切换完成后所有相对路径都应按 workspace 解释，不能假设仍位于源码仓库。 ## 3. 组件关系 @@ -253,7 +255,7 @@ WebUI/TUI 文件字节通过受鉴权的 HTTP 接口流式传输，WebSocket 只 同源 `/api/*` 管理接口只提供显式白名单能力： -- 配置读取/原子写入；响应中的密钥字段统一掩码，原样提交掩码会恢复现有值。运行配置只在重启后生效，不热替换运行中组件。 +- 配置读取/原子写入；响应中的密钥字段统一掩码，原样提交掩码会恢复现有值。`POST /api/config/reload` 通过同一重载控制器校验并切换 Gateway 运行代，`GET /api/config/reload/status` 查询 generation、相位与最近错误。 - `USER.md`、`AGENTS.md` 只允许固定文件名，不接受任意路径。 - 日志、记忆、任务和运行记录均限制单次返回数量；日志目录固定为 `~/.picobot/logs`。 - 任务与记忆读取复用 Storage API，不允许 WebUI 直接持有 SQLite 连接或拼接任意查询。 diff --git a/docs/CONFIG_HOT_RELOAD_DESIGN.md b/docs/CONFIG_HOT_RELOAD_DESIGN.md new file mode 100644 index 0000000..31526f8 --- /dev/null +++ b/docs/CONFIG_HOT_RELOAD_DESIGN.md @@ -0,0 +1,389 @@ +# PicoBot 配置热重载设计与实现 + +本文档描述 PicoBot 1.3.0 配置热重载功能的设计目标、运行时模型、实现边界、失败语义和维护要求。代码与测试是最终事实来源；本文用于解释为什么采用当前方案，以及后续修改必须保持哪些不变量。 + +## 1. 背景 + +PicoBot 的配置并非只在一个全局对象中读取。Gateway 启动时会把配置拆分并复制到多个长生命周期组件： + +- `SessionManager`、现有 `Session`、子 Agent 和 Scheduler 持有 Provider/模型配置。 +- `ChannelManager` 按配置创建并启动飞书、CLI Chat 等 Channel。 +- MCP 配置在启动时用于连接 Server，并把发现的工具注册到 `ToolRegistry`。 +- Browser、文件上传、鉴权、后台任务并发度等配置在各自组件构造时固化。 +- Gateway 的监听 socket、进程 cwd 和 SQLite 连接具有进程级生命周期。 + +因此，简单地重新读取 `config.json` 或替换一个 `Config` 指针并不能可靠生效。这样会造成请求处理组件混用新旧配置，例如新会话使用新模型、旧 Session 仍使用旧 Provider，或者配置显示飞书已禁用但旧连接仍在接收消息。 + +当前实现采用“运行代（runtime generation）切换”：先在旧运行代仍然服务时解析、校验并构造完整候选运行代；候选可用后排空当前交互工作，再回收旧运行代并激活新运行代。 + +## 2. 目标与非目标 + +### 2.1 目标 + +- 提供统一的 `picobot reload`、`/reload` 和 `reload_config` 工具入口。 +- 在停止旧运行代前完成候选配置解析、关键字段校验和依赖构造。 +- 配置错误或候选构造失败时继续使用旧运行代，不中断服务。 +- 尽量让正在执行及已经排队的交互 Turn 完成，避免热重载直接截断发起重载的 Turn。 +- 重新创建所有启动期固化配置的组件，使 Provider、Channel、MCP、Scheduler、Browser、鉴权和上传策略一致地切换。 +- 保持监听 socket，不释放端口，避免切换期间被其他进程抢占。 +- 保持运行期环境变量操作线程安全：热重载不得调用 `std::env::set_var`。 +- 为所有入口提供相同的校验、排队和错误语义。 + +### 2.2 非目标 + +- 不支持热变更监听地址、workspace 或 SQLite 路径。 +- 不承诺 WebSocket 连接无感迁移；切换会主动关闭旧连接，客户端需要重连。 +- 不实现 nginx 式新旧 worker 长时间并行处理连接。PicoBot 是单进程、单 Gateway 运行代模型，采用保留 socket 的顺序切换。 +- 不动态修改已经启动进程的环境变量；`.env` 新值只用于重新解析配置占位符和显式组件配置。 +- 不把“请求已接受”解释为“切换已经完成”。触发方在候选运行代构造成功后收到响应，实际切换在排空阶段之后发生。 +- 不赋予本地 admin token 调用管理 API 的新权限；重载 HTTP API 遵循现有设备鉴权边界。 + +## 3. 核心设计：Gateway 运行代 + +### 3.1 生命周期结构 + +Gateway 进程拥有两层生命周期： + +```text +进程生命周期 +├── 固定配置路径 +├── 启动前进程环境快照 +├── 启动 cwd +├── 原始监听 socket +├── ReloadController +└── 当前 Gateway 运行代（可替换） + ├── GatewayState + ├── SessionManager / Session workers + ├── MessageBus / routers / outbound dispatcher + ├── ChannelManager / Channel connections + ├── Scheduler / MCP / tools + ├── AuthManager / UploadRegistry + ├── Axum Router / WebSocket connections + └── TaskSupervisor +``` + +进程级资源在 `gateway::run()` 外层只创建一次；运行代资源由 `GatewayState::from_config()` 重新构造。 + +### 3.2 为什么保留监听 socket + +`gateway::run()` 首次启动时创建一个非阻塞 `std::net::TcpListener`，并在整个进程生命周期内持有它。每个运行代通过 `try_clone()` 获得一个 Tokio listener 交给 Axum。 + +切换时旧 Axum serve future 停止接受连接并退出，但原始 listener 仍然占有地址。旧运行代清理完成后，新运行代再克隆同一个 listener 开始接受连接。这样可以： + +- 避免重新 bind 失败或端口被其他进程抢占。 +- 保留内核 listen backlog；短暂切换窗口中的新 TCP 连接可能排队等待新运行代接收。 +- 允许 Axum Router、鉴权 middleware 和 WebSocket state 随运行代完整替换。 + +这不是零停顿切换。旧运行代停止和受监督任务回收期间没有 Axum accept loop；当前回收宽限期上限为 10 秒，通常会更短。 + +## 4. 重载控制通道 + +重载协调类型位于 `src/gateway/reload.rs`： + +- `ReloadHandle`：可克隆的请求端，注入 SessionManager、工具和 Gateway HTTP state。 +- `ReloadController`：由 `gateway::run()` 独占，持有请求 receiver、启动环境快照和启动 cwd。 +- `ReloadRequest`：包含 generation ID 与 oneshot response，用于把候选校验/构造结果返回触发方。 +- `ReloadStatus`：记录 `preparing`、`draining`、`activating`、`active`、`failed` 相位、时间与最近错误。 + +控制通道使用容量为 8 的 Tokio MPSC 队列，并用原子 pending 标记保证同一时间最多只有一次重载。`ReloadHandle::request()` 使用 `try_send`： + +- 已有重载尚未进入 `active` 或 `failed` 终态时立即返回 `another configuration reload is already pending`。 +- Gateway 正在退出、receiver 已关闭时立即返回 `gateway is shutting down`。 +- 请求成功入队后等待对应 oneshot 结果。 + +有界队列避免错误调用或模型重复调用形成无界重载积压；并发请求不会排队形成连续运行代切换，而是收到明确冲突错误。 + +## 5. 三种触发入口 + +三个入口只负责鉴权、参数适配和结果展示，最终都调用同一个 `ReloadHandle::request()`。 + +| 入口 | 实现 | 行为 | +|------|------|------| +| `picobot reload` | `src/main.rs`、`client::reload_gateway()` | 把 WebSocket/HTTP Gateway URL 转为 HTTP base URL，使用已保存的 TUI bearer token 调用 `POST /api/config/reload` | +| `/reload` | `SessionManager::execute_slash_command()` | 通过普通 slash command 路由执行，不进入 Agent 队列；结果作为 command output 返回当前 Channel | +| `reload_config` | `ReloadConfigTool` | 仅注册到根交互 Agent；无参数、独占执行，描述要求仅在用户明确要求重载时调用。子 Agent、Cron 和 managed scheduled Agent 无权获得该工具 | + +`POST /api/config/reload` 返回 accepted generation；`GET /api/config/reload/status` 返回当前重载相位和最近错误。并发 `POST` 返回 409，Gateway 退出或候选准备失败返回 503，配置或不可变字段错误返回 400。 + +HTTP 路由属于现有 protected Router，因此： + +- pairing 关闭时沿用 `PairingDisabled` 身份。 +- pairing 开启时需要已配对的 bearer/cookie 凭据。 +- 本地 `web_admin_token` 仍只允许用于既有的 loopback WebSocket 特例，不能绕过管理 API 鉴权。 + +pairing 开启但本机尚未保存 TUI token 时，`picobot reload` 会收到 HTTP 401；应先完成现有配对流程，或从一个已认证的聊天/WebUI 连接触发 `/reload`。 + +`GatewayState::new()` 只构造独立 state，没有运行代主循环，因此其中的 reload handle 明确不可用。正式 Gateway 必须通过 `gateway::run()` 启动，才能执行热重载。 + +## 6. 端到端时序 + +```mermaid +sequenceDiagram + participant Caller as CLI / Slash / Tool + participant RC as ReloadController + participant Old as Old GatewayState + participant New as Candidate GatewayState + participant HTTP as Axum / Listener + + Caller->>RC: ReloadHandle::request() + RC->>RC: 重新读取 config + .env + RC->>RC: 校验 default agent、飞书凭据、不可变字段 + RC->>New: GatewayState::from_config(candidate) + alt 解析、校验或构造失败 + RC-->>Caller: Error + Note over Old: 旧运行代继续服务 + else 候选构造成功 + RC->>Old: 关闭 admission，拒绝新工作 + RC-->>Caller: Accepted + generation / 等待当前任务后切换 + RC->>Old: 等待 inbound、Session、Scheduler 与后台 Agent 排空（最多 60s） + RC->>Old: cancel WebSocket connections + RC->>HTTP: graceful shutdown 当前 serve future + RC->>Old: stop_all channels + RC->>Old: TaskSupervisor shutdown(10s) + RC->>New: start_all channels + message processing + RC->>HTTP: 从保留 listener 创建新 serve future + end +``` + +### 6.1 准备阶段 + +准备阶段在旧运行代继续提供服务时执行： + +1. `load_candidate()` 重新读取当前 Gateway 启动时确定的配置文件。 +2. 使用启动环境快照和启动 cwd 解析 `.env`、占位符与相对 workspace。 +3. 校验 default agent 能解析为完整 `LLMProviderConfig`。 +4. 若飞书启用，校验 `app_id` 和 `app_secret` 非空。 +5. 比较不可热变更字段。 +6. 调用 `GatewayState::from_config()` 构造候选运行代。 + +候选构造会重新创建 Storage handle、MemoryManager、MessageBus、ChannelManager、SessionManager、工具集、MCP 配置 wrapper、AuthManager 和 UploadRegistry。外部 Channel、MCP 连接、消息 routers、outbound dispatcher 和 Scheduler 在候选成为当前运行代前不会激活。`from_config(..., false)` 也不会再次修改进程 cwd 或释放默认配置文件。 + +MCP 的真实连接位于激活阶段，避免准备候选时启动双份 stdio 子进程或提前覆盖进程级 `MCP_SERVER_STATUS`。单个 MCP Server 连接失败沿用启动语义：记录错误并跳过其工具，不会让整个运行代激活失败。 + +SessionManager 构造期注册的通知消费者和周期清理任务已经归候选 `TaskSupervisor` 所有，但在激活前没有候选消息入口；周期清理也会跳过首次 interval tick。 + +### 6.2 接受响应 + +候选运行代构造成功并关闭旧代 admission 后，Controller 通过 oneshot 返回 generation 与消息： + +```text +配置校验通过；Gateway 将在当前任务结束后切换到新配置。 +``` + +此响应表示候选配置已经通过准备阶段，不表示切换完成。先返回响应有两个原因： + +- `/reload` 的 command output 需要通过旧运行代发送给用户。 +- `reload_config` 工具需要返回 tool result，让发起它的 Agent Turn 正常完成。 + +### 6.3 排空阶段 + +每个运行代有一个 `RuntimeAdmission`。Inbound 在进入会话 lane 前获取 activity guard，因此已经进入 lane 的消息也计入排空；关闭 admission 后新消息不再进入会话处理，并尽量收到“正在重新加载”提示。`/reload` 的 command output 使用 outbound delivery acknowledgement，guard 只有在回复实际投递成功或明确失败后才释放，不再依赖固定 sleep。 + +`SessionManager::wait_until_idle()` 同时检查所有已加载 Session： + +- `current_cancel.is_some()` 表示当前有 Agent Turn 正在执行。 +- Session MPSC sender 的剩余容量小于最大容量，表示仍有排队任务。 +- 必须连续空闲 100ms 才认为稳定，避免 worker 刚取出任务、尚未设置 `current_cancel` 的竞态窗口。 + +Scheduler 在领取和执行任务前检查 admission，已执行任务持有 guard 到结果提交与投递完成；后台子 Agent 同样持有 guard。最长统一等待 60 秒。超时不会撤销已经接受的重载，而是记录 warning 并继续回收旧运行代；未完成工作随后会被取消。 + +Axum 自身会优雅等待已进入 handler 的 HTTP 请求；其 graceful shutdown 另有 10 秒硬上限，超过后 abort serve task。未纳入 admission 的维护型后台任务由 `TaskSupervisor` 的 10 秒有界关停负责。 + +### 6.4 切换与回收阶段 + +切换顺序是： + +1. 取消旧 `connection_shutdown`，使 WebSocket handler 主动退出。 +2. 取消当前 Axum generation shutdown token，停止接受新请求并等待已进入的请求完成。 +3. 调用旧 `ChannelManager::stop_all()`，停止外部消息入口。 +4. 取消旧 `TaskSupervisor`，最多等待 10 秒，超时任务被 abort。 +5. 把候选 state 设为当前 state。 +6. 启动候选 Channel、MCP、消息处理循环、outbound dispatcher 和 Scheduler。 +7. 从原始 listener 克隆新 Tokio listener，构建并运行新 Axum Router。 + +旧 WebSocket 不跨运行代迁移。TUI/WebUI 重连后按原有 client scope 从 SQLite 恢复 dialog 和历史；运行中的内存 Session 不直接搬迁到新 SessionManager。 + +## 7. 配置与环境变量语义 + +### 7.1 启动加载 + +正常启动使用以下优先级解析配置： + +```text +进程启动环境 > workspace/.env > config目录/.env +``` + +启动时合并的 `.env` 值会在单线程阶段写入进程环境，供后续 MCP 和工具子进程继承。 + +### 7.2 热重载加载 + +Gateway 在首次调用 `Config::load_from()` 前保存： + +- 原始进程环境 `startup_process_env`。 +- 切换 workspace 前的 `startup_cwd`。 + +热重载调用 `Config::load_for_reload()`： + +- 重新读取 config 目录 `.env` 和 workspace `.env`。 +- 继续以原始启动环境作为最高优先级，避免启动时注入进程环境的旧 `.env` 值错误覆盖新文件。 +- 相对 `workspace_dir` 始终相对于启动 cwd 解析，不受 Gateway 已经 `chdir(workspace)` 影响。 +- 只解析得到候选配置，不调用 `env::set_var`，避免多线程进程中修改全局环境。 + +因此，`.env` 的修改会影响配置中的 `` 占位符和由配置显式传入的新组件。它不会改变现有进程环境；仅依赖继承环境、但没有通过配置显式传值的 Shell/子进程仍会看到启动时环境。需要变更这类继承环境时应完整重启 Gateway。 + +## 8. 热重载边界 + +### 8.1 可通过新运行代生效的配置 + +以下配置消费者会随 `GatewayState` 重建： + +| 配置区域 | 新运行代中的效果 | +|----------|------------------| +| `providers`、`models`、`agents` | 新 SessionManager、Session、主 Agent、子 Agent 和 Scheduler Agent 使用新 Provider/模型参数 | +| `channels` | ChannelManager 重新创建并启动已启用 Channel，allowlist、凭据、媒体和实时投递策略更新 | +| `mcp` | 重新连接 MCP Server，并重新生成工具注册表 | +| `browser` | 根据新配置注册或移除 Browser 工具 | +| `memory` | 重建 MemoryManager；维护任务使用新的 retention 配置 | +| `gateway.scheduler` | 启用、关闭或按新并发/轮询/超时参数创建 Scheduler | +| `gateway.max_concurrent_background_tasks` | 新 SubAgentManager 使用新的并发上限 | +| `gateway.file_transfer` | 新 UploadRegistry 和 WebSocket capability 使用新限制 | +| `gateway.require_pairing` | 新 Router/AuthManager 使用新鉴权要求；已有 WebSocket 在切换时断开 | + +配置中尚未被运行时代码消费的字段，在热重载后仍然不会产生功能效果。例如当前 `session_ttl_hours` 和 `cleanup_interval_minutes` 只完成了解析，尚未接入 Session 清理逻辑。 + +`client.gateway_url` 是 CLI 侧配置：`picobot reload` 在发请求前读取它来确定目标 Gateway，但它不是 Gateway 运行代设置。 + +### 8.2 必须完整重启的配置 + +| 字段 | 原因 | +|------|------| +| `gateway.host`、`gateway.port` | 原始监听 socket 在进程生命周期内固定；热重载不会重新 bind | +| `workspace_dir` | Gateway 已修改进程 cwd，工具路径、媒体路径和相对文件语义均依赖它 | +| `gateway.session_db_path` | Storage、SessionManager、Scheduler 和历史恢复必须共享同一个数据库身份 | +| 未显式进入配置组件的继承环境变量 | 热重载禁止运行期修改进程全局环境 | + +候选值与当前值不一致时，`load_candidate()` 返回错误，并明确提示重启 Gateway。`workspace_dir` 在比较前按启动 cwd 解析并 canonicalize；通过校验后会被归一化为当前绝对 workspace 路径，避免候选构造时受当前 cwd 影响。 + +## 9. 原子性与失败语义 + +这里的“原子”是指请求处理组件的配置可见性：旧运行代不会被逐项改造成半套新配置。它不是数据库事务，也不是两个 worker 的瞬时指针交换；进程级 MCP status 的提前更新是下文记录的已知例外。 + +| 失败阶段 | 行为 | +|----------|------| +| 已有 pending 重载/控制器关闭 | 分别返回 409/503，不读取配置 | +| JSON、`.env`、占位符或默认 Agent 校验失败 | 返回错误，旧运行代保持不变 | +| 不可热变更字段发生变化 | 返回 restart-required 错误，旧运行代保持不变 | +| `GatewayState::from_config()` 构造失败 | 返回错误；候选被丢弃，其 TaskSupervisor 随对象释放取消；旧请求处理运行代保持不变 | +| 单个 MCP Server 连接或工具发现失败 | 记录 MCP 失败状态，候选继续构造且不注册该 Server 的工具；这不视为整体 reload 失败 | +| 等待交互空闲超过 60 秒 | 记录 warning，继续切换，旧运行中的剩余工作可能被取消 | +| 旧 Channel 停止失败 | 记录 error，继续回收其他组件和切换 | +| 旧受监督任务 10 秒内未退出 | TaskSupervisor abort 剩余任务，继续切换 | +| 候选激活阶段 `start_all()` 失败 | `gateway::run()` 返回错误；若由 systemd 管理，则按 service restart policy 重启 | + +准备阶段成功后才向调用者返回 accepted。激活阶段仍可能遇到运行时错误，因此调用者不应把 accepted 当作健康检查；可使用 `GET /api/config/reload/status` 等待相同 generation 进入 `active`，并结合 `/health` 与客户端重连确认。 + +WebUI `PUT /api/config` 只负责原子写文件、恢复被掩码的 secret 并返回 `restart_required: true`；它不会隐式触发重载。显式的 `POST /api/config/reload` 将文件写入与运行代切换解耦，使用户可以批量编辑后主动决定生效时机。 + +## 10. 并发与生命周期不变量 + +维护或扩展热重载时必须保持以下约束： + +1. 只有 `gateway::run()` 拥有 reload receiver 和当前运行代，其他组件只能持有 `ReloadHandle`。 +2. 候选构造不得修改旧 `GatewayState`，也不得提前连接 MCP 或替换旧 MessageBus、Channel、ToolRegistry、MCP status 或 Router。 +3. 不得在热重载路径调用 `env::set_var`；环境文件只允许在单线程首次启动时安装到进程环境。 +4. 不得释放原始 listener 后再尝试 bind 同一地址。 +5. 旧 WebSocket 必须观察 `connection_shutdown`，不能在新鉴权/配置运行代启动后继续无限存活。 +6. 旧长生命周期任务必须由旧 `TaskSupervisor` 回收；候选任务必须由候选 Supervisor 所有。 +7. 排空检查不得长时间持有 `SessionManagerInner` 或 Session mutex；当前实现先复制 Session Arc，再逐一短暂检查。 +8. reload tool 必须保持 exclusive 且只对根交互 Agent 可见，避免后台或子 Agent 触发进程级切换。 +9. admission 必须在会话 lane 入队前获取；Slash 回复、Scheduler job 和后台子 Agent 必须持有 guard 到其持久化/投递边界。 +10. 不可热变更字段的比较必须按有效路径归一化，并发生在候选运行代构造之前。 +11. 新增启动期配置消费者时，应确认它是否由 `GatewayState::from_config()` 重建，并更新本文的热重载边界表。 + +## 11. 关键实现位置 + +| 文件/符号 | 职责 | +|-----------|------| +| `src/gateway/reload.rs` | 重载请求通道、候选配置加载、不可变字段校验 | +| `src/gateway/mod.rs::run` | 持有 listener、当前运行代和切换主循环 | +| `src/gateway/mod.rs::GatewayState::from_config` | 构造一套完整运行代依赖 | +| `src/config/mod.rs::Config::load_for_reload` | 使用启动环境/cwd 安全重新解析配置，不修改进程环境 | +| `src/session/session.rs::wait_until_idle` | 检查活动 Turn、Session 队列和稳定空闲窗口 | +| `src/session/session.rs::execute_slash_command` | `/reload` 入口 | +| `src/tools/reload_config.rs` | Agent 可调用的独占重载工具 | +| `src/gateway/http.rs::reload_config` | 受保护的 `POST /api/config/reload` | +| `src/client/mod.rs::reload_gateway` | CLI HTTP 客户端与 bearer token 注入 | +| `src/main.rs::Command::Reload` | `picobot reload` CLI 定义 | + +## 12. 测试策略 + +当前回归测试覆盖： + +- 候选配置允许 Provider/模型等运行时字段变化。 +- 相对 `workspace_dir` 按启动 cwd 正确解析。 +- workspace 变化被拒绝且返回明确错误。 +- `None` 与 `./picobot.db` 指向同一有效数据库路径时允许重载。 +- admission 关闭后拒绝新工作，并等待现有 activity guard 释放。 +- command output 在 dispatcher 明确确认投递前不会释放处理任务。 +- 真实子进程 Gateway 可完成 generation 2 切换；无效候选返回 400 且旧代 `/health` 继续可用。 +- `/reload` alias 能解析到规范命令。 +- 全量 Rust 单元测试验证 Gateway、Session、Channel、鉴权和 TaskSupervisor 既有行为。 +- 离线协议集成测试验证 slash/WebSocket 相关协议没有退化。 +- Clippy、Cargo build 和 WebUI check/build 验证完整构建链。 + +后续适合增加的集成测试： + +1. 使用可控假 Provider 验证新 model 被下一 Turn 实际使用。 +2. 在长 Turn 中调用 `reload_config`，验证 tool result 和最终消息投递后才断开。 +3. Session 队列有积压时验证重载等待队列排空。 +4. 排空超时、Channel stop 超时和候选激活失败的故障注入。 +5. 重载前后鉴权策略变化以及旧 WebSocket 失效。 + +## 13. 运维使用 + +修改配置并保存后执行： + +```bash +picobot reload +``` + +连接非默认 Gateway： + +```bash +picobot reload --gateway-url https://gateway.example.com +``` + +也可以在支持 slash command 的聊天中发送： + +```text +/reload +``` + +若配置修改涉及监听地址、workspace、数据库路径或必须进入进程继承环境的变量，应执行完整重启： + +```bash +picobot service restart +``` + +建议的运维流程是： + +1. 原子保存配置文件。 +2. 执行 reload 并检查返回是否为候选已接受。 +3. 使用已认证请求轮询 `GET /api/config/reload/status`，确认返回的 generation 进入 `active`。 +4. 等待客户端重连，检查 `/health` 和关键 Channel。 +5. 若激活失败，由 systemd 重启或人工恢复配置后再次启动。 + +## 14. 已知限制与演进方向 + +- 旧运行代回收与新运行代激活是顺序执行，存在短暂 accept/Channel intake 空窗。 +- WebSocket 需要客户端自行重连，服务端没有连接迁移协议。 +- 候选激活失败没有自动回滚到已经回收的旧运行代，依赖 systemd restart 或人工恢复。 +- 同一时间只允许一个 pending 重载；并发请求返回 409，不做合并或排队。 +- `.env` 不能热修改进程继承环境。 +- admission 覆盖消息入口、Scheduler 与后台子 Agent；其他维护型任务仍依赖 TaskSupervisor 的有界关停。 + +可能的后续演进包括： + +- 将 Channel 激活前检查拆成显式 `prepare()`，把更多运行时失败提前到旧运行代仍可回退的阶段。 +- 在不破坏 Channel/Session 边界的前提下，引入动态 Router service，实现新旧 HTTP generation 短期重叠。 +- 为 TUI/WebUI 增加 reload 完成通知和自动重连状态提示。 diff --git a/docs/CONFIG_HOT_RELOAD_REVIEW.md b/docs/CONFIG_HOT_RELOAD_REVIEW.md new file mode 100644 index 0000000..a221d61 --- /dev/null +++ b/docs/CONFIG_HOT_RELOAD_REVIEW.md @@ -0,0 +1,369 @@ +# 配置热重载功能审核报告 + +> 状态：审查完成；主要意见已于 2026-07-21 落地。本文前六节保留首次审查快照，行号、测试数和“当前实现”描述可能已过时；以下处置表与代码为最新结论。 +> +> 本文是对 `docs/CONFIG_HOT_RELOAD_DESIGN.md` 描述的配置热重载功能及其当前未提交实现的综合审核。审核覆盖架构合理性、实现一致性、关键不变量和改进建议；具体行为以代码和测试为最终依据。相关实现位于 `src/gateway/reload.rs`、`src/gateway/mod.rs::run`、`src/config/mod.rs::load_for_reload`、`src/session/session.rs::wait_until_idle`、`src/tools/reload_config.rs` 与 `src/gateway/http.rs::reload_config`。 + +## 0. Review 意见处置与实现结果 + +| 意见 | 处置 | 结论与实现 | +|------|------|------------| +| A1 / E1 MCP 准备期副作用 | 接收 | `connect_all()` 移到运行代激活阶段，候选构造不再启动双份 MCP 或提前覆盖全局 status。 | +| A2 / E2 generation 与状态查询 | 接收 | 每次请求分配 generation；新增 `GET /api/config/reload/status`，状态为 `preparing/draining/activating/active/failed`。 | +| A3 激活失败回滚 | 部分接收 | 接收“需要明确失败状态”，但驳回先启动新 Channel 再停旧 Channel。飞书长连接双开会重复消费事件，风险高于短暂切换空窗；完整 prepare/activate 或可恢复旧代留作后续。 | +| A4 / L4 Agent 工具风险 | 部分接收 | 保留用户要求的 `reload_config`，但只允许根交互 Agent 使用；子 Agent、Cron、managed scheduled Agent 均剔除该工具，并保持 exclusive。驳回直接删除工具。 | +| A5 / E5 后台任务排空 | 接收 | 引入运行代 admission/activity guard；Scheduler 已执行 job 与后台子 Agent 持有 guard，新任务在 drain 后不再进入。 | +| I1 / E4 Slash 回复可能丢失 | 接收 | `/reload` 所在 inbound 在 lane 入队前持有 guard；command output 改为显式等待 outbound delivery acknowledgement，删除固定 500ms sleep。 | +| I2 / F2 DB 路径误判 | 接收 | 比较归一化后的有效路径，`None` 与 `./picobot.db` 可判为同一数据库。 | +| I3 零散预校验 | 部分接收 | 保留 default agent 与 Feishu 的快速错误提示；其余组件统一由候选 `from_config()` 构造验证，不继续扩张 ad-hoc 字段校验。长期采用显式 prepare contract。 | +| I4 HTTP 错误码 | 接收 | pending 返回 409，退出/准备故障返回 503，配置与不可变字段错误返回 400。错误改为 `ReloadError` 类型。 | +| I5 候选双 Storage | 驳回为正确性缺陷 | 同库多连接是 sqlx/SQLite 的正常模式，候选无消息入口；路径身份仍被强制保持一致。该点保留为测试与锁竞争观察项，而非阻止上线。 | +| F1 集成测试 | 部分接收 | 新增真实 Gateway 子进程测试，覆盖成功切换到 generation 2、状态查询、无效候选不影响旧代健康。可控假 Provider、长 Turn 和故障注入仍待补充。 | +| L1 prepare/activate 分离 | 方向接收 | 本轮已把 MCP activation 与候选构造分离；完整组件级接口留作后续架构演进。 | +| L2 动态 Router | 暂缓 | 当前保留 listener 并为 Axum graceful shutdown 增加 10 秒硬上限；不为本功能引入动态 service 复杂度。 | +| L3 客户端完成通知 | 暂缓 | 后端 generation/status 已具备；TUI/WebUI 展示可在后续独立实现。 | + +首次 Review 未指出、但本轮一并修复的两个关键问题：候选构造原先直接在 `select!` 分支内 await，会暂停轮询旧 Axum serve 与进程信号；现在 serve 独立受监督运行，准备和排空阶段都继续响应服务退出。其次，原实现排空前没有关闭入口，持续新消息可能让排空永不稳定；现在 admission 先关闭再 drain。 + +## 1. 审查范围与依据 + +### 1.1 审查对象 + +- 设计文档：`docs/CONFIG_HOT_RELOAD_DESIGN.md` +- 实现：当前未提交的 17 个文件变更与 2 个新增文件（`src/gateway/reload.rs`、`src/tools/reload_config.rs`），共 ~304 行净增 +- 关联变更：`Cargo.toml`、`webui/package.json` 版本号 1.2.2 → 1.3.0；`README.md`、`AGENTS.md`、`docs/ARCHITECTURE.md`、`resources/skills/about-picobot/references/config.md` 文档同步 + +### 1.2 审查依据 + +- 仓库既有架构边界与并发不变量（见 `docs/ARCHITECTURE.md`） +- 现有相似机制（Channel 生命周期、TaskSupervisor、TurnController、OutboundDispatcher） +- 验证命令：`cargo build`、`cargo clippy --all-targets --all-features -- -D warnings`、`cargo test --lib`（330 passed）、`webui && npm run check && npm run build` + +### 1.3 验证结果 + +| 命令 | 结果 | +|------|------| +| `cargo build` | 通过 | +| `cargo clippy --all-targets --all-features -- -D warnings` | 通过 | +| `cargo test --lib` | 330 passed / 0 failed | +| `cd webui && npm run check && npm run build` | 0 errors / 0 warnings | + +## 2. 架构审查 + +### 2.1 整体架构评估 + +**结论：架构方向正确，运行代（runtime generation）切换模型是 PicoBot 配置散落现状下的唯一可靠方案。** + +PicoBot Gateway 启动时把配置拆分复制到 `SessionManager`、`ChannelManager`、MCP、Scheduler、Browser、Auth、Upload 等长生命周期组件；`gateway.host/port`、进程 cwd、SQLite 连接具有进程级生命周期。在这种结构下，任何"原地替换 Config 指针"的方案都会导致请求处理组件混用新旧配置（新会话用新模型、旧 Session 仍用旧 Provider；或飞书配置显示禁用但旧连接仍在接收消息）。 + +运行代切换通过"先在旧代仍服务时构造完整候选代；候选可用后排空当前交互工作，再回收旧代并激活新代"避免了半套配置暴露。这一选择与 PicoBot 单进程、单 Gateway 模型契合，不引入 daemon/fork 层。 + +### 2.2 运行代模型合理性 + +运行代模型的关键设计点均合理： + +| 设计点 | 评估 | +|--------|------| +| 保留原始 `std::net::TcpListener`，每代 `try_clone()` | ✅ 避免 bind 失败与端口抢占，保留内核 backlog | +| 候选构造期间不修改旧 `GatewayState` | ✅ 避免半套配置暴露 | +| `Config::load_for_reload` 使用启动环境快照、不调 `set_var` | ✅ 多线程运行期修改进程环境的危险被正确规避 | +| 不可变字段（host/port/workspace/db_path）边界清晰 | ✅ 边界划分正确 | +| 三个入口（CLI、`/reload`、`reload_config` 工具）共享控制通道 | ✅ 单一校验路径，语义一致 | +| 拒绝 nginx 式新旧 worker 长期并行 | ✅ 单进程规模不值得这份复杂度 | +| `ReloadHandle` 用有界 MPSC + `try_send` | ✅ 无界积压被正确拒绝 | + +### 2.3 边界划分评估 + +热重载边界表（设计文档第 8 节）覆盖完整： + +- 可热重载：`providers`/`models`/`agents`、`channels`、`mcp`、`browser`、`memory`、`gateway.scheduler`、`gateway.max_concurrent_background_tasks`、`gateway.file_transfer`、`gateway.require_pairing` +- 必须重启：`gateway.host`/`port`、`workspace_dir`、`gateway.session_db_path`、未进入配置组件的继承环境变量 + +边界划分与实现中 `load_candidate()` 的校验项一一对应。`workspace_dir` 在比较前按启动 cwd 解析并 canonicalize（`reload.rs:80-89`），与 `from_config` 中 `ensure_workspace_dir` 的 canonicalize 行为一致，比较基准正确。 + +### 2.4 并发不变量评估 + +设计文档第 10 节列出的 10 条不变量在实现中均得到遵守： + +| 不变量 | 实现位置 | 遵守情况 | +|--------|----------|----------| +| 只有 `run()` 拥有 receiver 与当前运行代 | `mod.rs:331` ReloadController 在 `run()` 内创建 | ✅ | +| 候选构造不修改旧 GatewayState | `mod.rs:399` `from_config` 创建独立 state | ✅ | +| 不在热重载路径调 `env::set_var` | `config/mod.rs:611` `apply_to_process=false` | ✅ | +| 不释放原始 listener 后 rebind | `mod.rs:345` listener 在 `run()` 内持有 | ✅ | +| 旧 WebSocket 观察 `connection_shutdown` | `mod.rs:421` 切换前 cancel | ✅ | +| 旧任务由旧 TaskSupervisor 回收 | `mod.rs:432-436` 旧 supervisor shutdown | ✅ | +| 排空检查不长时间持有 Session mutex | `session.rs:2113-2124` 先克隆 Arc 再短锁 | ✅ | +| reload tool 保持 exclusive | `reload_config.rs:49` `exclusive: true` | ✅ | +| 不可变字段比较在候选构造之前 | `reload.rs:80-100` | ✅ | +| 新增启动期配置消费者需更新边界表 | 文档约束 | ⚠️ 维护性约束，无机制强制 | + +## 3. 实现审查 + +### 3.1 与设计文档的一致性 + +实现与设计文档的关键路径高度一致： + +| 设计文档章节 | 实现位置 | 一致性 | +|--------------|----------|--------| +| §6.1 准备阶段：load_candidate → from_config | `mod.rs:386-408` | ✅ | +| §6.2 接受响应：候选构造成功后通过 oneshot 返回 | `mod.rs:410-411` | ✅ | +| §6.3 排空阶段：wait_until_idle 60s + 500ms 投递窗口 | `mod.rs:412-419` | ✅ | +| §6.4 切换顺序：connection_shutdown → generation_shutdown → channel stop → task_supervisor → 新代启动 | `mod.rs:421-422, 429-436, 350-351` | ✅ | +| §4 控制通道：容量 8、try_send、队列满立即返回错误 | `reload.rs:8, 47-54` | ✅ | +| §7.2 环境变量语义：不修改进程环境 | `config/mod.rs:541-553` | ✅ | +| §9 失败语义：候选构造失败时旧代不变 | `mod.rs:394-408` continue 不切换 | ✅ | + +### 3.2 关键路径分析 + +#### 3.2.1 切换时序 + +`gateway::run` 的主循环（`mod.rs:349-441`）正确实现了运行代切换： + +``` +外层 loop { + start_all / start_message_processing // 新代激活 + 内层 loop { select! { serve | process_signal | reload_request } } + serve.await // 等待旧 axum graceful shutdown + channel_manager.stop_all // 停止旧 channel intake + task_supervisor.cancel + shutdown(10s)// 回收旧受监督任务 + state = next_state // 切换 +} +``` + +`wait_for_shutdown_signal()` 在外层循环内创建（`mod.rs:365`），每代新建 future，不存在重复 poll 已完成 future 的 UB。 + +#### 3.2.2 候选构造与旧代隔离 + +`from_config`（`mod.rs:51-236`）为候选创建独立的 Storage、MessageBus、ChannelManager、SessionManager、ToolRegistry、AuthManager、UploadRegistry。候选的 `channel_manager.init()` 仅构造 Channel 对象，不调用 `start()`，因此不会与旧 channel 并发连接飞书 API。 + +#### 3.2.3 监听 socket 保留 + +`std::net::TcpListener`（`mod.rs:345`）在进程生命周期内持有；每代通过 `try_clone()`（`mod.rs:353`）获得新 fd。旧代 `serve` future 退出时仅释放克隆 fd，原始 socket 不释放，新代可重新克隆并 accept。内核 backlog 在切换窗口中暂存新 TCP 连接。 + +### 3.3 测试覆盖评估 + +**结论：单元测试覆盖不足，集成测试完全缺失。** + +当前测试： + +| 测试 | 位置 | 覆盖范围 | +|------|------|----------| +| `candidate_accepts_runtime_changes_and_rejects_workspace_changes` | `reload.rs:127` | load_candidate 的 model_id 变更接受与 workspace 拒绝 | +| `resolve_slash_command("reload")` | `session.rs:3485` | slash 命令解析 | + +缺失但设计文档第 12 节明确列出的测试： + +1. 启动真实 Gateway、修改 model/channel 配置、HTTP 触发重载、验证新代生效 +2. 配置无效时验证旧 WebSocket 与旧 Provider 仍可工作 +3. 长 Turn 中调用 `reload_config`、验证 tool result 与最终消息投递后才断开 +4. Session 队列积压时验证重载等待排空 +5. 排空超时、Channel stop 超时、候选激活失败的故障注入 +6. 重载前后鉴权策略变化与旧 WebSocket 失效 + +当前测试仅覆盖纯函数路径（`load_candidate`、slash 解析），未验证任何运行时切换行为。这是上线前的主要风险点。 + +### 3.4 代码质量 + +- **Clippy**：`-D warnings` 通过 +- **类型安全**：`ReloadHandle::unavailable()`（`reload.rs:39`）通过 drop receiver 使 `try_send` 返回 `Closed`，正确表达"Gateway 未由 `run()` 启动"的语义 +- **错误处理**：候选构造失败时 `request.response.send(Err(...))` 后 `continue`，不切换；MCP 单 server 失败沿用启动语义（记录错误、跳过工具），不阻断候选构造 +- **资源管理**：旧 TaskSupervisor 的 10s 有界 shutdown + abort 保证回收有硬时间边界 + +## 4. 发现的问题 + +### 4.1 架构层面问题 + +#### A1. MCP 在准备阶段连接，制造进程级副作用【中】 + +`from_config`（`mod.rs:171`）调用 `mcp::connect_all()`，立即建立 MCP 客户端连接或启动 stdio 子进程，并更新进程级 `MCP_SERVER_STATUS`（`mcp/mod.rs:42`）。这违反了设计文档第 10 节不变量 #2"候选构造不得修改旧 GatewayState"的精神——MCP status 虽非请求处理状态，但仍是旧代可见的进程级状态。 + +**具体影响：** + +- 候选构造期间，旧代的 `/mcp` 命令显示候选的连接状态而非旧代状态 +- stdio MCP 子进程双份运行（旧代 + 候选）可能竞争资源或 stdin/stdout +- 候选在 `connect_all` 之后失败（如 `ensure_default_maintenance_job` 失败，`mod.rs:194`），MCP 连接被 drop 但 `MCP_SERVER_STATUS` 仍显示 `connected: true`，旧代 `/mcp` 显示陈旧数据 + +设计文档第 6.1 节与第 14 节将此列为"已知例外"。但该例外的收益仅为"提前发现 MCP 失败"——而 MCP 单 server 失败本就被当非致命跳过（`mcp/mod.rs:171`），不需要提前连接来验证。 + +#### A2. 无 generation ID 与状态查询【中】 + +调用方只能得到准备阶段结果（"配置校验通过；Gateway 将在当前任务结束后切换到新配置"），无法查询重载最终是否完成。运维需要翻日志或重连客户端确认切换状态。设计文档第 14 节将此列为"演进方向"，但 generation ID（一个 atomic 计数器）+ `/api/config/reload/status` 端点成本极低，收益显著，应在 v1 内置。 + +#### A3. 候选激活失败无回滚【中-高】 + +切换顺序为：停旧 channel → 取消旧 TaskSupervisor → 切换 state → 启动新 channel（`mod.rs:429-351`）。若新代 `start_all()` 失败，`run()` 返回错误，依赖 systemd 拉起。但 channel 启动失败常为瞬时问题（飞书 5xx、端口冲突），丢掉本来正常服务的旧代去重启是可用性损失。旧代此时已被回收，无法回滚。 + +#### A4. `reload_config` Agent 工具的软约束【低-中】 + +`ReloadConfigTool`（`reload_config.rs`）注册到默认工具集，依赖 description"仅在用户明确要求重新加载配置时调用"约束 LLM。LLM compliance 是软约束，非可靠边界。`exclusive: true` 仅保证不与其他副作用工具并行，不保证调用时机正确。 + +#### A5. 排空仅覆盖交互 Session，不覆盖 Scheduler/后台 Agent【低】 + +`wait_until_idle` 仅检查内存中 Session 的 `current_cancel` 与 `agent_tx` 队列。Scheduler job、独立后台子 Agent、HTTP handler 不在排空范围内。Scheduler job 可能正在写 DB 或发送消息，被 TaskSupervisor 10s abort 截断可能留下不一致状态。设计文档第 6.3 节已承认此范围。 + +### 4.2 实现层面问题 + +#### I1. `/reload` slash command 绕过 `wait_until_idle`【中】 + +`session.rs:2631` 显示 slash command 在 `handle_message` 内联处理，返回 `HandleResult::CommandOutput`，**不进入 session worker 队列**。因此 `wait_until_idle` 检查 `current_cancel`/`agent_tx.capacity()` 时看不到 `/reload` Turn 的活动状态，立即返回（仅 100ms 稳定 + 500ms sleep）。 + +slash command output 经 `process_inbound` → `publish_command_output` → outbound dispatcher → channel API 投递，整条链路必须在 ~600ms（+ serve graceful shutdown + 10s TaskSupervisor shutdown）内完成。对 CLI channel 足够，但对 Feishu 等远端 channel 较紧。设计文档第 6.3 节"为 slash command output 和终态投递留出发送窗口"承认了 500ms 窗口，但 500ms 是固定值，无背压保证。 + +实际窗口因 TaskSupervisor 的 10s shutdown 较宽，不会丢消息——但若 channel `stop_all()` 关闭了连接，in-flight 的 `send_message` 可能失败。 + +#### I2. `session_db_path` 比较为原始字符串【低】 + +`reload.rs:91` 直接比较 `current.gateway.session_db_path != candidate.gateway.session_db_path`。若用户把 `null` 改为 `"./picobot.db"`（解析后同一路径），会被误拒。保守是对的，但产生假阳性。`workspace_dir` 已做 canonicalize 比较，`session_db_path` 应保持一致。 + +#### I3. `load_candidate` 仅校验 Feishu 凭据【低】 + +`reload.rs:73-78` 仅校验飞书 `app_id`/`app_secret` 非空。其他 channel 配置（若有）、MCP 配置、Browser 配置等在 `from_config` 期间才验证，可能在那里失败。这与设计文档第 6.1 节一致（"若飞书启用，校验 app_id 和 app_secret 非空"），但将失败发现延后到了候选构造阶段。 + +#### I4. `reload_config` HTTP handler 错误码语义【低】 + +`http.rs:355` 对所有错误用 `ApiError::bad_request`（400）。"gateway is shutting down"（队列关闭）更适合 503 Service Unavailable，"another configuration reload is already pending"更适合 409 Conflict。 + +#### I5. 候选 Storage 与旧 Storage 共享同一 SQLite 文件【低】 + +`from_config` 为候选创建新 Storage 连接到同一 `picobot.db`。候选的 background notification consumer 与 cleanup task 已归候选 TaskSupervisor，cleanup 跳过首次 tick，无入口触发 notification，故实际不写。理论上有并发写锁竞争可能，实际风险低。设计文档未显式说明此点。 + +### 4.3 严重度分级 + +| 问题 | 严重度 | 影响 | +|------|--------|------| +| A3 候选激活失败无回滚 | 中-高 | 瞬时 channel 故障导致整个 Gateway 重启 | +| A1 MCP 准备阶段连接 | 中 | 进程级 status 污染、子进程双份、失败后状态陈旧 | +| A2 无 generation ID | 中 | 运维无法确认切换完成状态 | +| I1 `/reload` 绕过排空 | 中 | 远端 channel output 投递窗口紧 | +| A4 reload_config 软约束 | 低-中 | LLM 误调用风险 | +| A5 排空不覆盖 Scheduler | 低 | 后台 job 被硬取消可能留下不一致 | +| I2 session_db_path 假阳性 | 低 | 用户需重启而非重载 | +| I3 仅校验 Feishu | 低 | 失败发现延后 | +| I4 HTTP 错误码 | 低 | 语义不准确 | +| I5 双 Storage 连接 | 低 | 理论并发风险 | + +## 5. 改进建议 + +### 5.1 必须修复（上线前） + +#### F1. 补充端到端集成测试 + +至少覆盖设计文档第 12 节列出的前 3 项： + +1. 启动真实 Gateway、修改 model_id、HTTP 触发重载、验证新 model 在新 Turn 中生效 +2. 配置无效（如 default agent 解析失败）时验证旧 WebSocket 与旧 Provider 仍可工作 +3. 长 Turn 中调用 `reload_config`、验证 tool result 与最终消息投递后才断开 + +这些测试无法用单元测试替代，需启动真实 Gateway 进程。 + +#### F2. `session_db_path` 比较归一化 + +`reload.rs:91` 应将 `session_db_path` 相对于 workspace 解析并 canonicalize 后比较，与 `workspace_dir` 处理方式一致。`None` 与 `"picobot.db"`（默认值）应视为等价。 + +### 5.2 建议增强（近期演进） + +#### E1. MCP 连接移出 `from_config`，消除"已知例外" + +将 `mcp::connect_all()` 从 `from_config`（`mod.rs:171`）移到 `start_all()` 阶段，与 channel 启动同相位。收益： + +- 消除进程级 `MCP_SERVER_STATUS` 在候选构造期间被污染 +- 消除 stdio 子进程双份运行 +- 候选失败时无 MCP 连接泄漏 +- 不再需要设计文档第 10 节不变量 #2 的"已知例外"声明 + +代价：MCP 连接失败从"候选构造失败"延后到"激活失败"。但 MCP 单 server 失败本就非致命（跳过该 server 工具），整体激活失败语义不变。 + +#### E2. 引入 generation ID 与 status 查询 + +在 `ReloadController` 增加 `Arc` generation 计数器与 `ReloadState` enum（`Idle`/`Preparing`/`Draining`/`Activating`/`Active`/`Failed`）。提供 `GET /api/config/reload/status` 端点。调用方在收到 "accepted" 后可轮询确认切换完成。成本极低（一个 atomic + 一个路由 + 一个 enum），运维收益显著。 + +#### E3. 旧代保留至新代 `start_all()` 成功 + +调整切换顺序为：先启动新 channel（新代 channel_manager.start_all），成功后再停止旧 channel。短暂双 channel 并存对飞书 webhook 幂等消息可容忍。代价是需处理两代 channel 并存的资源冲突（如媒体目录、飞书事件去重），但避免瞬时 channel 故障导致 Gateway 整体重启。 + +若实现成本过高，至少应在 `start_all()` 失败时尝试重启旧代 channel（旧 TaskSupervisor 已 cancel，可能无法恢复；需评估可行性）。 + +#### E4. `/reload` slash command 排空路径 + +两种方案： + +- **方案 A**：让 reload controller 记录触发源（channel, chat_id），等待该 inbound lane 的当前消息处理完成后再切换，而非泛化等待所有 session idle +- **方案 B**：为 outbound dispatcher 增加显式 drain contract，在 `stop_all` 前等待 outbound 队列排空或超时 + +方案 A 更精确，方案 B 更通用。两者都比固定 500ms sleep 可靠。 + +#### E5. Scheduler 与后台 job 协作式 drain + +在 Scheduler job 的协作取消边界检查 reload token，允许 job 在写 DB 前/后选择继续完成或退出。避免 TaskSupervisor 10s abort 截断 DB 写一半的 job。设计文档第 14 节已列出此项。 + +### 5.3 演进方向（中长期） + +#### L1. `prepare()` / `activate()` 显式分离 + +将 `from_config` 拆为： + +- `construct()`：纯内存，无 I/O，可反复调用 +- `prepare()`：可失败的 I/O（channel health check、MCP 连接、Storage ping），旧代仍服务 +- `activate()`：开始接收消息 + +使更多失败前移到旧代仍可回退的阶段，而非等到 activate 才暴露。设计文档第 14 节已列出此项。 + +#### L2. 动态 Router service + +引入新旧 HTTP generation 短期重叠，消除 accept/Channel intake 空窗。需不破坏 Channel/Session 边界。设计文档第 14 节已列出此项。 + +#### L3. WebUI/TUI reload 完成通知 + +客户端在 WebSocket 重连后显示 reload 完成状态与自动重连提示。依赖 E2 的 generation ID。 + +#### L4. 重新评估 `reload_config` Agent 工具 + +考虑移除该工具，仅保留 CLI 与 `/reload`。Agent 触发进程级状态切换的风险（A4）可能不抵边际收益。若保留，应在 SessionManager 层做调用方校验（如要求参数带确认 token），而非靠 description。 + +## 6. 总体结论 + +### 6.1 设计评估 + +设计文档质量高，边界清晰，不变量明确，失败语义完整。运行代切换模型是 PicoBot 当前架构下的正确选择。设计文档诚实地列出了已知限制（第 14 节），未掩饰缺陷。 + +主要设计层面的不足是把几件本应 v1 内置的能力（generation ID、MCP 相位对齐、旧代保留至新代激活成功）推迟到"演进方向"，导致可用性与可观测性打了折扣。MCP 的"已知例外"（A1）是设计妥协被文档化的典型，留着会让后续维护者也认为"再来一个例外无所谓"，应尽早消除而非长期承担。 + +### 6.2 实现评估 + +实现忠实遵循设计，关键不变量均得到遵守。代码通过 `cargo build`、`cargo clippy -D warnings`、`cargo test --lib`（330 passed）与 WebUI `check/build`。版本号、文档、AGENTS.md 同步更新。 + +主要实现层面的不足是测试覆盖：仅 `load_candidate` 与 slash 解析有单元测试，无任何运行时切换行为的集成验证（I1-I5 中多数问题需要集成测试才能暴露）。设计文档第 12 节明确列出但未实现的 6 项集成测试是上线前的主要风险。 + +### 6.3 上线建议 + +| 项 | 判定 | +|----|------| +| 架构方向 | ✅ 可接受 | +| 实现一致性 | ✅ 可接受 | +| 代码质量 | ✅ 可接受（Clippy/tests/build 全通过） | +| 测试覆盖 | ⚠️ 不足，需补集成测试（F1） | +| 已知限制 | ⚠️ 文档已承认，但 A1/A3 应优先修复 | + +**建议：在完成 F1（集成测试）与 F2（session_db_path 归一化）后可提交。A1（MCP 相位）、A3（无回滚）应列为后续优先修复项，不应长期承担。** + +## 附录：关键文件与符号索引 + +| 文件/符号 | 职责 | 行号 | +|-----------|------|------| +| `src/gateway/reload.rs::ReloadController` | 重载控制通道、候选配置加载、不可变字段校验 | 19-36 | +| `src/gateway/reload.rs::ReloadHandle` | 可克隆的请求端，注入 SessionManager/工具/HTTP state | 14-59 | +| `src/gateway/reload.rs::load_candidate` | 候选配置解析与不可变字段校验 | 61-102 | +| `src/gateway/mod.rs::run` | 持有 listener、当前运行代与切换主循环 | 318-441 | +| `src/gateway/mod.rs::GatewayState::from_config` | 构造一套完整运行代依赖 | 51-236 | +| `src/gateway/mod.rs::build_router` | 为每代构建 Axum Router | 448-489 | +| `src/config/mod.rs::Config::load_for_reload` | 使用启动环境/cwd 安全重新解析配置 | 541-553 | +| `src/config/mod.rs::Config::load_from_with_process_env` | 共享的配置加载实现，支持不写入进程环境 | 547-625 | +| `src/session/session.rs::wait_until_idle` | 检查活动 Turn、Session 队列与稳定空闲窗口 | 2109-2144 | +| `src/session/session.rs::execute_slash_command` | `/reload` slash 入口 | 2093-2099 | +| `src/tools/reload_config.rs::ReloadConfigTool` | Agent 可调用的独占重载工具 | 6-52 | +| `src/gateway/http.rs::reload_config` | 受保护的 `POST /api/config/reload` | 347-360 | +| `src/client/mod.rs::reload_gateway` | CLI HTTP 客户端与 bearer token 注入 | 101-124 | +| `src/main.rs::Command::Reload` | `picobot reload` CLI 定义 | 59-64, 132-138 | +| `src/mcp/mod.rs::connect_all` | MCP 连接（当前在 from_config 期间调用，见 A1） | 126-181 | +| `src/mcp/mod.rs::MCP_SERVER_STATUS` | 进程级 MCP 状态（A1 的副作用源） | 36-45 | diff --git a/resources/skills/about-picobot/references/config.md b/resources/skills/about-picobot/references/config.md index cbe7919..f01f29b 100644 --- a/resources/skills/about-picobot/references/config.md +++ b/resources/skills/about-picobot/references/config.md @@ -3,7 +3,7 @@ 配置文件加载顺序：`~/.picobot/config.json` → 当前目录 `./config.json`。 占位符 `` 从启动环境替换。PicoBot 依次加载 `config.json` 同目录的 `.env`、`workspace_dir/.env`，最后保留启动进程已有环境变量作为最高优先级；workspace 层覆盖配置目录层。合并值也会进入进程环境，供 MCP 和工具子进程继承。workspace `.env` 不能修改用于定位自身的 `workspace_dir`。 -Gateway WebUI 的“配置”页可以编辑实际加载的配置文件。读取时 API Key、secret、password 和 token 会显示为 `********`，保持掩码不变再保存会保留原值；写入采用同目录临时文件替换。运行配置保存后需要重启 Gateway，`USER.md` 与 `AGENTS.md` 的修改用于后续构建的 Agent 上下文。 +Gateway WebUI 的“配置”页可以编辑实际加载的配置文件。读取时 API Key、secret、password 和 token 会显示为 `********`，保持掩码不变再保存会保留原值；写入采用同目录临时文件替换。运行配置保存后可执行 `picobot reload`、发送 `/reload`，或由根交互 Agent 在用户明确要求时调用 `reload_config` 工具。Gateway 会先校验候选配置，停止接收新工作并等待交互 Turn、Scheduler job 和后台子 Agent 到达安全边界后切换；失败时继续使用旧配置。`GET /api/config/reload/status` 可查询 generation、相位与最近错误。`gateway.host`、`gateway.port`、`workspace_dir` 和 `gateway.session_db_path` 的有效路径必须通过完整重启变更。`USER.md` 与 `AGENTS.md` 的修改用于后续构建的 Agent 上下文。 ## config.json 结构 @@ -105,7 +105,7 @@ Gateway WebUI 的“配置”页可以编辑实际加载的配置文件。读取 | `media_dir_max_bytes` | int | 536870912 | 飞书媒体目录容量上限；达到上限后拒绝新下载，不自动删除旧文件 | | `request_timeout_secs` | int | 30 | 单次飞书 HTTP 请求及响应体读取的硬超时，运行时限制在 5–120 秒 | -飞书属于外部渠道：无论是否开启实时卡片，都不会接收模型 reasoning；工具只显示紧凑状态。配置修改需重启 Gateway 生效。 +飞书属于外部渠道：无论是否开启实时卡片，都不会接收模型 reasoning；工具只显示紧凑状态。渠道配置可通过 Gateway 配置重载生效。 ## mcp 字段 diff --git a/src/agent/sub_agent.rs b/src/agent/sub_agent.rs index 086d4b2..c568744 100644 --- a/src/agent/sub_agent.rs +++ b/src/agent/sub_agent.rs @@ -126,6 +126,7 @@ pub struct SubAgentManager { skills_loader: Option>, work_manager: Option>, task_supervisor: crate::task_supervisor::TaskSupervisor, + admission: crate::gateway::reload::RuntimeAdmission, } impl SubAgentManager { @@ -149,9 +150,18 @@ impl SubAgentManager { skills_loader, work_manager: None, task_supervisor, + admission: crate::gateway::reload::RuntimeAdmission::open(), } } + pub(crate) fn with_admission( + mut self, + admission: crate::gateway::reload::RuntimeAdmission, + ) -> Self { + self.admission = admission; + self + } + pub fn with_work_manager(mut self, work_manager: Arc) -> Self { self.work_manager = Some(work_manager); self @@ -164,7 +174,11 @@ impl SubAgentManager { }; let filtered = ToolRegistry::new(); for (name, tool) in self.full_tools.iter() { - if allowed_set.contains(name.as_str()) && name != "delegate" && name != "todo" { + if allowed_set.contains(name.as_str()) + && name != "delegate" + && name != "todo" + && name != "reload_config" + { filtered.register_raw(name, tool); } } @@ -324,6 +338,12 @@ impl SubAgentManager { config: SubAgentConfig, ctx: DelegateContext, ) -> Result { + let activity = self.admission.try_enter().ok_or_else(|| { + SubAgentError::Other( + "gateway is draining for configuration reload and cannot accept background tasks" + .to_string(), + ) + })?; let permit = self .background_permits .clone() @@ -423,6 +443,7 @@ impl SubAgentManager { let spawned = self.task_supervisor.spawn_graceful( format!("sub-agent:{task_id}"), async move { + let _activity = activity; let _permit = permit; let started_at = chrono::Utc::now().timestamp_millis(); @@ -851,4 +872,17 @@ mod tests { assert!(matches!(error, SubAgentError::TooManyTasks(1))); } + + #[test] + fn reload_tool_is_never_delegated_to_sub_agents() { + let manager = manager(1); + manager + .full_tools + .register(crate::tools::ReloadConfigTool::new( + crate::gateway::reload::ReloadHandle::unavailable(), + )); + + let filtered = manager.filter_tools(&Some(vec!["reload_config".to_string()])); + assert!(filtered.get("reload_config").is_none()); + } } diff --git a/src/client/mod.rs b/src/client/mod.rs index db8cd98..8966059 100644 --- a/src/client/mod.rs +++ b/src/client/mod.rs @@ -98,6 +98,30 @@ fn gateway_http_base_url(gateway_url: &str) -> Result Result> { + let base = gateway_http_base_url(gateway_url)?; + let mut request = reqwest::Client::new().post(format!("{base}/api/config/reload")); + if let Some(token) = load_auth_token() { + request = request.bearer_auth(token); + } + let response = request.send().await?; + let status = response.status(); + let body: serde_json::Value = response.json().await?; + if !status.is_success() { + return Err(body + .get("error") + .and_then(serde_json::Value::as_str) + .unwrap_or("configuration reload failed") + .to_string() + .into()); + } + Ok(body + .get("message") + .and_then(serde_json::Value::as_str) + .unwrap_or("configuration reload scheduled") + .to_string()) +} + async fn exchange_pairing_code( gateway_url: &str, code: &str, diff --git a/src/config/mod.rs b/src/config/mod.rs index e8fea34..913ac18 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -538,6 +538,32 @@ impl Config { } pub(crate) fn load_from(path: &Path) -> Result> { + let process_env = collect_process_env(); + Self::load_from_with_process_env(path, &process_env, true, None) + } + + /// Reload configuration without mutating the process environment. The + /// supplied environment must be the process environment captured before + /// startup `.env` layers were installed, preserving the documented + /// precedence while keeping runtime reload thread-safe. + pub(crate) fn load_for_reload( + path: &Path, + startup_process_env: &HashMap, + startup_cwd: &Path, + ) -> Result> { + Self::load_from_with_process_env(path, startup_process_env, false, Some(startup_cwd)) + } + + pub(crate) fn startup_process_env() -> HashMap { + collect_process_env() + } + + fn load_from_with_process_env( + path: &Path, + process_env: &HashMap, + apply_to_process: bool, + workspace_base: Option<&Path>, + ) -> Result> { let config_path = if path.exists() { path.to_path_buf() } else { @@ -554,7 +580,6 @@ impl Config { }; let content = fs::read_to_string(&config_path)?; - let process_env = collect_process_env(); let config_env_path = config_path .parent() .unwrap_or_else(|| Path::new(".")) @@ -563,14 +588,19 @@ impl Config { // The config-directory layer selects the workspace. Loading the workspace // layer first would be circular because its location comes from config.json. - let initial_env = merge_env_layers(&config_env, &HashMap::new(), &process_env); + let initial_env = merge_env_layers(&config_env, &HashMap::new(), process_env); let initial_content = resolve_env_placeholders(&content, &initial_env); let initial_config: Config = serde_json::from_str(&initial_content)?; - let workspace_path = expand_path(&initial_config.workspace_dir); + let mut workspace_path = expand_path(&initial_config.workspace_dir); + if workspace_path.is_relative() + && let Some(base) = workspace_base + { + workspace_path = base.join(workspace_path); + } let workspace_env_path = workspace_path.join(".env"); let workspace_env = read_env_file(&workspace_env_path)?; - let effective_env = merge_env_layers(&config_env, &workspace_env, &process_env); + let effective_env = merge_env_layers(&config_env, &workspace_env, process_env); let resolved_content = resolve_env_placeholders(&content, &effective_env); let config: Config = serde_json::from_str(&resolved_content)?; if config.workspace_dir != initial_config.workspace_dir { @@ -581,7 +611,9 @@ impl Config { .into()); } - apply_env_layers(&config_env, &workspace_env, &process_env); + if apply_to_process { + apply_env_layers(&config_env, &workspace_env, process_env); + } tracing::info!( path = %config_path.display(), config_env = %config_env_path.display(), diff --git a/src/gateway/http.rs b/src/gateway/http.rs index 9cf230f..34a6727 100644 --- a/src/gateway/http.rs +++ b/src/gateway/http.rs @@ -106,6 +106,20 @@ impl ApiError { } } + fn conflict(message: impl Into) -> Self { + Self { + status: StatusCode::CONFLICT, + message: message.into(), + } + } + + fn service_unavailable(message: impl Into) -> Self { + Self { + status: StatusCode::SERVICE_UNAVAILABLE, + message: message.into(), + } + } + fn internal(error: impl std::fmt::Display) -> Self { tracing::error!(error = %error, "WebUI API request failed"); Self { @@ -344,6 +358,36 @@ pub struct ConfigResponse { restart_required: bool, } +#[derive(Serialize)] +pub struct ReloadResponse { + generation: u64, + message: String, +} + +pub async fn reload_config( + State(state): State>, +) -> Result, ApiError> { + let accepted = state.reload.request().await.map_err(|error| match error { + super::reload::ReloadError::AlreadyPending => ApiError::conflict(error.to_string()), + super::reload::ReloadError::ShuttingDown + | super::reload::ReloadError::PreparationFailed(_) => { + ApiError::service_unavailable(error.to_string()) + } + super::reload::ReloadError::InvalidConfig(_) + | super::reload::ReloadError::ImmutableField(_) => ApiError::bad_request(error.to_string()), + })?; + Ok(Json(ReloadResponse { + generation: accepted.generation, + message: accepted.message, + })) +} + +pub async fn reload_status( + State(state): State>, +) -> Json { + Json(state.reload.status()) +} + pub async fn get_config( State(state): State>, ) -> Result, ApiError> { @@ -389,7 +433,7 @@ pub async fn put_config( let pretty = serde_json::to_string_pretty(&incoming).map_err(ApiError::internal)? + "\n"; atomic_write(&state.config_path, pretty.as_bytes()).await?; - tracing::info!(path = %state.config_path.display(), "Configuration updated from WebUI; restart required"); + tracing::info!(path = %state.config_path.display(), "Configuration updated from WebUI; reload or restart required"); let mut response = incoming; redact_secrets(&mut response); diff --git a/src/gateway/mod.rs b/src/gateway/mod.rs index 5fabe1f..71bb193 100644 --- a/src/gateway/mod.rs +++ b/src/gateway/mod.rs @@ -1,5 +1,6 @@ pub mod auth; pub mod http; +pub(crate) mod reload; mod router; pub mod uploads; pub mod ws; @@ -32,20 +33,33 @@ pub struct GatewayState { pub connection_shutdown: tokio_util::sync::CancellationToken, pub auth: auth::AuthManager, pub uploads: uploads::UploadRegistry, + pub(crate) reload: reload::ReloadHandle, + pub(crate) admission: reload::RuntimeAdmission, } impl GatewayState { + /// Construct a standalone state. Configuration reload is available only + /// when the state is owned by [`run`], which owns the generation loop. pub async fn new() -> Result> { let config_path = crate::config::resolve_default_config_path(); let config = Config::load_from(&config_path)?; - Self::from_config(config, config_path).await + Self::from_config( + config, + config_path, + reload::ReloadHandle::unavailable(), + true, + ) + .await } async fn from_config( config: Config, config_path: std::path::PathBuf, + reload: reload::ReloadHandle, + initialize_process: bool, ) -> Result> { let task_supervisor = TaskSupervisor::new(); + let admission = reload::RuntimeAdmission::open(); let delivery_coordinator = DeliveryCoordinator::new(ConversationWriteLocks::default()); let connection_shutdown = tokio_util::sync::CancellationToken::new(); let auth = auth::AuthManager::load( @@ -59,19 +73,24 @@ impl GatewayState { let workspace_path = expand_path(&config.workspace_dir); let workspace_path = ensure_workspace_dir(&workspace_path)?; - // Switch current working directory to workspace - std::env::set_current_dir(&workspace_path).map_err(|e| { - format!( - "Failed to switch to workspace directory {}: {}", - workspace_path.display(), - e - ) - })?; + if initialize_process { + // Startup is single-threaded. Reload candidates reuse the already + // selected workspace and must not mutate process-global cwd. + std::env::set_current_dir(&workspace_path).map_err(|e| { + format!( + "Failed to switch to workspace directory {}: {}", + workspace_path.display(), + e + ) + })?; + } tracing::info!("Using workspace directory: {}", workspace_path.display()); // Release default AGENTS.md and USER.md to ~/.picobot/ if not exist - ensure_default_config_files(); + if initialize_process { + ensure_default_config_files(); + } // Get provider config for SessionManager let mut provider_config = config.get_provider_config("default")?; @@ -141,7 +160,9 @@ impl GatewayState { memory_manager, task_supervisor.clone(), turn_delivery, - ), + reload.clone(), + ) + .with_admission(admission.clone()), browser_config, config.gateway.max_concurrent_background_tasks, )?; @@ -160,21 +181,6 @@ impl GatewayState { valid_channels.clone(), )); - // Initialize MCP servers — connect and register discovered tools - if !config.mcp.servers.is_empty() { - let mcp_tools = mcp::connect_all(&config.mcp).await; - for tool_info in mcp_tools { - let wrapper = mcp::McpToolWrapper::new( - &tool_info.server_name, - tool_info.tool_name, - tool_info.description, - tool_info.schema, - tool_info.connection, - ); - session_manager.tools().register(wrapper); - } - } - // Initialize scheduler if enabled in config let scheduler_config = config.gateway.scheduler.clone().unwrap_or_default(); if scheduler_config.enabled { @@ -225,6 +231,8 @@ impl GatewayState { connection_shutdown, auth, uploads, + reload, + admission, }) } @@ -240,6 +248,21 @@ impl GatewayState { /// Start the message processing loops pub async fn start_message_processing(&self) { + // MCP connections have external/process-wide side effects. Activate + // them only after this generation becomes current, never while it is + // merely a reload candidate. + let mcp_tools = mcp::connect_all(&self.config.mcp).await; + for tool_info in mcp_tools { + let wrapper = mcp::McpToolWrapper::new( + &tool_info.server_name, + tool_info.tool_name, + tool_info.description, + tool_info.schema, + tool_info.connection, + ); + self.session_manager.tools().register(wrapper); + } + let bus = self.bus(); let bus_for_outbound = bus.clone(); let session_manager = self.session_manager.clone(); @@ -276,7 +299,12 @@ impl GatewayState { } }); - router::spawn_message_routers(bus.clone(), session_manager, self.task_supervisor.clone()); + router::spawn_message_routers( + bus.clone(), + session_manager, + self.task_supervisor.clone(), + self.admission.clone(), + ); // Spawn outbound dispatcher let dispatcher = OutboundDispatcher::new( @@ -295,10 +323,11 @@ impl GatewayState { // Spawn scheduler background task if enabled let scheduler_config = self.config.gateway.scheduler.clone().unwrap_or_default(); if scheduler_config.enabled { - let sched = Arc::new(Scheduler::new( + let sched = Arc::new(Scheduler::with_admission( self.storage.clone(), self.session_manager.clone(), scheduler_config, + self.admission.clone(), )); self.task_supervisor.spawn("scheduler", async move { sched.run().await; @@ -313,30 +342,230 @@ pub async fn run( port: Option, ) -> Result<(), Box> { let config_path = crate::config::resolve_default_config_path(); + let startup_process_env = Config::startup_process_env(); + let startup_cwd = std::env::current_dir()?; let config = Config::load_from(&config_path)?; // Initialize logging logging::init_logging(); tracing::info!(config_path = %config_path.display(), "Starting PicoBot Gateway"); - let state = Arc::new(GatewayState::from_config(config, config_path).await?); - - // Start all channels (init already done while constructing GatewayState) - state.channel_manager.start_all().await?; - - // Start message processing (inbound processor + control processor + outbound dispatcher) - state.start_message_processing().await; + let mut reload_controller = reload::ReloadController::new(startup_process_env, startup_cwd); + let mut state = Arc::new( + GatewayState::from_config( + config, + config_path.clone(), + reload_controller.handle.clone(), + true, + ) + .await?, + ); // CLI args override config file values let bind_host = host.unwrap_or_else(|| state.config.gateway.host.clone()); let bind_port = port.unwrap_or(state.config.gateway.port); + let addr = format!("{}:{}", bind_host, bind_port); + let listener = std::net::TcpListener::bind(&addr)?; + listener.set_nonblocking(true)?; + tracing::info!(address = %addr, "Gateway listening"); + let process_signal = wait_for_shutdown_signal(); + tokio::pin!(process_signal); + let mut current_generation = 1_u64; + loop { + if let Err(error) = state.channel_manager.start_all().await { + reload_controller.set_failed(current_generation, error.to_string()); + return Err(error.into()); + } + state.start_message_processing().await; + reload_controller.set_phase(current_generation, reload::ReloadPhase::Active); + let app = build_router(state.clone()); + let generation_listener = TcpListener::from_std(listener.try_clone()?)?; + let generation_shutdown = tokio_util::sync::CancellationToken::new(); + let shutdown_wait = generation_shutdown.clone(); + let mut serve_task = tokio::spawn(async move { + axum::serve( + generation_listener, + app.into_make_service_with_connect_info::(), + ) + .with_graceful_shutdown(async move { shutdown_wait.cancelled().await }) + .await + }); + + let mut next_state = None; + let mut serve_result = None; + 'generation: loop { + tokio::select! { + result = &mut serve_task => { + serve_result = Some(result); + state.connection_shutdown.cancel(); + generation_shutdown.cancel(); + break 'generation; + } + _ = &mut process_signal => { + tracing::info!("Shutdown signal received"); + state.admission.close(); + state.connection_shutdown.cancel(); + generation_shutdown.cancel(); + break 'generation; + } + request = reload_controller.receiver.recv() => { + let Some(request) = request else { + state.admission.close(); + state.connection_shutdown.cancel(); + generation_shutdown.cancel(); + break 'generation; + }; + let requested_generation = request.generation; + reload_controller.set_phase(requested_generation, reload::ReloadPhase::Preparing); + let candidate = match reload::load_candidate( + &config_path, + &reload_controller.startup_process_env, + &reload_controller.startup_cwd, + &state.config, + &state.workspace_dir, + ) { + Ok(candidate) => candidate, + Err(error) => { + reload_controller.set_failed(requested_generation, error.to_string()); + let _ = request.response.send(Err(error)); + continue; + } + }; + let preparation = GatewayState::from_config( + candidate, + config_path.clone(), + reload_controller.handle.clone(), + false, + ); + tokio::pin!(preparation); + let prepared = match tokio::select! { + result = &mut preparation => Some(result), + result = &mut serve_task => { + serve_result = Some(result); + None + } + _ = &mut process_signal => None, + } { + None => { + let error = reload::ReloadError::ShuttingDown; + reload_controller.set_failed(requested_generation, error.to_string()); + let _ = request.response.send(Err(error)); + state.admission.close(); + state.connection_shutdown.cancel(); + generation_shutdown.cancel(); + break 'generation; + } + Some(result) => match result { + Ok(prepared) => Arc::new(prepared), + Err(error) => { + let error = reload::ReloadError::PreparationFailed(format!( + "configuration reload failed: {error}" + )); + reload_controller.set_failed(requested_generation, error.to_string()); + let _ = request.response.send(Err(error)); + continue; + } + }}; + state.admission.close(); + reload_controller.set_phase(requested_generation, reload::ReloadPhase::Draining); + let message = "配置校验通过；Gateway 将在当前任务结束后切换到新配置。".to_string(); + let _ = request.response.send(Ok(reload::ReloadAccepted { + generation: requested_generation, + message, + })); + let drain = async { + tokio::join!( + state.admission.wait_for_idle(), + state.session_manager.wait_until_idle(std::time::Duration::from_secs(60)), + ) + }; + let drain_result = tokio::select! { + result = tokio::time::timeout(std::time::Duration::from_secs(60), drain) => { + Some(matches!(result, Ok(((), true)))) + } + result = &mut serve_task => { + serve_result = Some(result); + None + } + _ = &mut process_signal => None, + }; + let Some(drained) = drain_result else { + reload_controller.set_failed( + requested_generation, + "gateway stopped while draining configuration reload", + ); + state.connection_shutdown.cancel(); + generation_shutdown.cancel(); + break 'generation; + }; + if !drained { + tracing::warn!("Reload drain period ended before all work became idle"); + } + tracing::info!(path = %config_path.display(), "Switching to reloaded configuration"); + reload_controller.set_phase(requested_generation, reload::ReloadPhase::Activating); + state.connection_shutdown.cancel(); + generation_shutdown.cancel(); + next_state = Some((prepared, requested_generation)); + break 'generation; + } + } + } + + if serve_result.is_none() { + serve_result = + match tokio::time::timeout(std::time::Duration::from_secs(10), &mut serve_task) + .await + { + Ok(result) => Some(result), + Err(_) => { + tracing::warn!("Aborting Axum generation after shutdown timeout"); + serve_task.abort(); + Some(serve_task.await) + } + }; + } + if let Err(error) = state.channel_manager.stop_all().await { + tracing::error!(error = %error, "Failed to stop channels cleanly"); + } + state.task_supervisor.cancel(); + state + .task_supervisor + .shutdown(std::time::Duration::from_secs(10)) + .await; + if let Some(result) = serve_result { + match result { + Ok(Ok(())) => {} + Ok(Err(error)) => return Err(error.into()), + Err(error) if error.is_cancelled() => {} + Err(error) => return Err(format!("Gateway server task failed: {error}").into()), + } + } + + match next_state.take() { + Some((prepared, generation)) => { + state = prepared; + current_generation = generation; + } + None => break, + } + } + + Ok(()) +} + +fn build_router(state: Arc) -> Router { let protected = Router::new() .route("/api/health", routing::get(http::health)) .route( "/api/config", routing::get(http::get_config).put(http::put_config), ) + .route("/api/config/reload", routing::post(http::reload_config)) + .route( + "/api/config/reload/status", + routing::get(http::reload_status), + ) .route( "/api/profiles/{name}", routing::get(http::get_profile).put(http::put_profile), @@ -360,7 +589,7 @@ pub async fn run( auth::require_auth, )); - let app = Router::new() + Router::new() .route("/", routing::get(http::webui_index)) .route("/app.js", routing::get(http::webui_script)) .route("/styles.css", routing::get(http::webui_styles)) @@ -369,36 +598,7 @@ pub async fn run( .route("/api/auth/pair", routing::post(auth::pair)) .route("/api/auth/code", routing::post(auth::issue_code)) .merge(protected) - .with_state(state.clone()); - - let addr = format!("{}:{}", bind_host, bind_port); - let listener = TcpListener::bind(&addr).await?; - tracing::info!(address = %addr, "Gateway listening"); - - let connection_shutdown = state.connection_shutdown.clone(); - let serve_result = axum::serve( - listener, - app.into_make_service_with_connect_info::(), - ) - .with_graceful_shutdown(async move { - wait_for_shutdown_signal().await; - tracing::info!("Shutdown signal received"); - connection_shutdown.cancel(); - }) - .await; - - // Stop external intake before waiting for internal work to finish. - if let Err(error) = state.channel_manager.stop_all().await { - tracing::error!(error = %error, "Failed to stop channels cleanly"); - } - state.task_supervisor.cancel(); - state - .task_supervisor - .shutdown(std::time::Duration::from_secs(10)) - .await; - serve_result?; - - Ok(()) + .with_state(state) } async fn wait_for_shutdown_signal() { diff --git a/src/gateway/reload.rs b/src/gateway/reload.rs new file mode 100644 index 0000000..b6c5a44 --- /dev/null +++ b/src/gateway/reload.rs @@ -0,0 +1,486 @@ +use std::collections::HashMap; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; +use std::sync::{Arc, RwLock}; + +use serde::Serialize; +use tokio::sync::{Notify, mpsc, oneshot}; + +use crate::config::Config; + +const RELOAD_QUEUE_CAPACITY: usize = 8; + +#[derive(Clone)] +pub(crate) struct RuntimeAdmission { + inner: Arc, +} + +struct AdmissionInner { + accepting: AtomicBool, + active: AtomicUsize, + idle: Notify, +} + +pub(crate) struct ActivityGuard { + admission: RuntimeAdmission, +} + +impl RuntimeAdmission { + pub fn open() -> Self { + Self { + inner: Arc::new(AdmissionInner { + accepting: AtomicBool::new(true), + active: AtomicUsize::new(0), + idle: Notify::new(), + }), + } + } + + pub fn try_enter(&self) -> Option { + if !self.inner.accepting.load(Ordering::Acquire) { + return None; + } + self.inner.active.fetch_add(1, Ordering::AcqRel); + if !self.inner.accepting.load(Ordering::Acquire) { + self.leave(); + return None; + } + Some(ActivityGuard { + admission: self.clone(), + }) + } + + pub fn close(&self) { + self.inner.accepting.store(false, Ordering::Release); + if self.inner.active.load(Ordering::Acquire) == 0 { + self.inner.idle.notify_waiters(); + } + } + + pub fn is_accepting(&self) -> bool { + self.inner.accepting.load(Ordering::Acquire) + } + + pub async fn wait_for_idle(&self) { + loop { + let notified = self.inner.idle.notified(); + if self.inner.active.load(Ordering::Acquire) == 0 { + return; + } + notified.await; + } + } + + fn leave(&self) { + if self.inner.active.fetch_sub(1, Ordering::AcqRel) == 1 { + self.inner.idle.notify_waiters(); + } + } +} + +impl Drop for ActivityGuard { + fn drop(&mut self) { + self.admission.leave(); + } +} + +pub(crate) struct ReloadRequest { + pub generation: u64, + pub response: oneshot::Sender>, +} + +#[derive(Clone)] +pub struct ReloadHandle { + sender: mpsc::Sender, + next_generation: Arc, + pending: Arc, + status: Arc>, +} + +pub(crate) struct ReloadController { + pub handle: ReloadHandle, + pub receiver: mpsc::Receiver, + pub startup_process_env: HashMap, + pub startup_cwd: PathBuf, + status: Arc>, +} + +#[derive(Debug, Clone, Serialize)] +pub struct ReloadAccepted { + pub generation: u64, + pub message: String, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ReloadPhase { + Active, + Preparing, + Draining, + Activating, + Failed, +} + +#[derive(Debug, Clone, Serialize)] +pub struct ReloadStatus { + pub generation: u64, + pub phase: ReloadPhase, + pub requested_at: Option, + pub activated_at: Option, + pub last_error: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ReloadError { + AlreadyPending, + ShuttingDown, + InvalidConfig(String), + ImmutableField(String), + PreparationFailed(String), +} + +impl std::fmt::Display for ReloadError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::AlreadyPending => { + write!(formatter, "another configuration reload is already pending") + } + Self::ShuttingDown => write!(formatter, "gateway is shutting down"), + Self::InvalidConfig(error) + | Self::ImmutableField(error) + | Self::PreparationFailed(error) => formatter.write_str(error), + } + } +} + +impl std::error::Error for ReloadError {} + +impl ReloadController { + pub fn new(startup_process_env: HashMap, startup_cwd: PathBuf) -> Self { + let (sender, receiver) = mpsc::channel(RELOAD_QUEUE_CAPACITY); + let status = Arc::new(RwLock::new(ReloadStatus { + generation: 1, + phase: ReloadPhase::Active, + requested_at: None, + activated_at: Some(chrono::Utc::now().timestamp_millis()), + last_error: None, + })); + Self { + handle: ReloadHandle { + sender, + next_generation: Arc::new(AtomicU64::new(2)), + pending: Arc::new(AtomicBool::new(false)), + status: status.clone(), + }, + receiver, + startup_process_env, + startup_cwd, + status, + } + } + + pub fn set_phase(&self, generation: u64, phase: ReloadPhase) { + let mut status = self + .status + .write() + .unwrap_or_else(|error| error.into_inner()); + status.generation = generation; + status.phase = phase; + if phase == ReloadPhase::Preparing { + status.requested_at = Some(chrono::Utc::now().timestamp_millis()); + status.activated_at = None; + status.last_error = None; + } + if phase == ReloadPhase::Active { + status.activated_at = Some(chrono::Utc::now().timestamp_millis()); + self.handle.pending.store(false, Ordering::Release); + } + } + + pub fn set_failed(&self, generation: u64, error: impl Into) { + let mut status = self + .status + .write() + .unwrap_or_else(|error| error.into_inner()); + status.generation = generation; + status.phase = ReloadPhase::Failed; + status.last_error = Some(error.into()); + self.handle.pending.store(false, Ordering::Release); + } +} + +impl ReloadHandle { + pub(crate) fn unavailable() -> Self { + let (sender, receiver) = mpsc::channel(1); + drop(receiver); + Self { + sender, + next_generation: Arc::new(AtomicU64::new(1)), + pending: Arc::new(AtomicBool::new(false)), + status: Arc::new(RwLock::new(ReloadStatus { + generation: 0, + phase: ReloadPhase::Failed, + requested_at: None, + activated_at: None, + last_error: Some("reload controller is unavailable".to_string()), + })), + } + } + + pub async fn request(&self) -> Result { + self.pending + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .map_err(|_| ReloadError::AlreadyPending)?; + let generation = self.next_generation.fetch_add(1, Ordering::Relaxed); + let (response, receiver) = oneshot::channel(); + if let Err(error) = self.sender.try_send(ReloadRequest { + generation, + response, + }) { + self.pending.store(false, Ordering::Release); + return Err(match error { + mpsc::error::TrySendError::Full(_) => ReloadError::AlreadyPending, + mpsc::error::TrySendError::Closed(_) => ReloadError::ShuttingDown, + }); + } + match receiver.await { + Ok(result) => result, + Err(_) => { + self.pending.store(false, Ordering::Release); + Err(ReloadError::ShuttingDown) + } + } + } + + pub fn status(&self) -> ReloadStatus { + self.status + .read() + .unwrap_or_else(|error| error.into_inner()) + .clone() + } +} + +pub(crate) fn load_candidate( + config_path: &Path, + startup_process_env: &HashMap, + startup_cwd: &std::path::Path, + current: &Config, + current_workspace: &std::path::Path, +) -> Result { + let mut candidate = Config::load_for_reload(config_path, startup_process_env, startup_cwd) + .map_err(|error| { + ReloadError::InvalidConfig(format!("configuration reload failed: {error}")) + })?; + candidate + .get_provider_config("default") + .map_err(|error| ReloadError::InvalidConfig(format!("invalid default agent: {error}")))?; + if let Some(feishu) = candidate.channels.get("feishu") + && feishu.enabled + && (feishu.app_id.trim().is_empty() || feishu.app_secret.trim().is_empty()) + { + return Err(ReloadError::InvalidConfig( + "enabled channels.feishu requires non-empty app_id and app_secret".to_string(), + )); + } + + let mut candidate_workspace = crate::config::expand_path(&candidate.workspace_dir); + if candidate_workspace.is_relative() { + candidate_workspace = startup_cwd.join(candidate_workspace); + } + let candidate_workspace = candidate_workspace + .canonicalize() + .unwrap_or(candidate_workspace); + if current_workspace != candidate_workspace { + return Err(ReloadError::ImmutableField( + "workspace_dir cannot be reloaded; restart the gateway".to_string(), + )); + } + candidate.workspace_dir = current_workspace.to_string_lossy().to_string(); + if effective_db_path(current, current_workspace) + != effective_db_path(&candidate, current_workspace) + { + return Err(ReloadError::ImmutableField( + "gateway.session_db_path cannot be reloaded; restart the gateway".to_string(), + )); + } + if current.gateway.host != candidate.gateway.host + || current.gateway.port != candidate.gateway.port + { + return Err(ReloadError::ImmutableField( + "gateway.host and gateway.port cannot be reloaded; restart the gateway".to_string(), + )); + } + Ok(candidate) +} + +fn effective_db_path(config: &Config, workspace: &Path) -> PathBuf { + let path = config + .gateway + .session_db_path + .as_deref() + .map(crate::config::expand_path) + .unwrap_or_else(|| workspace.join("picobot.db")); + let path = if path.is_relative() { + workspace.join(path) + } else { + path + }; + path.canonicalize().unwrap_or(path) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn config_json(workspace: &std::path::Path, model_id: &str) -> String { + serde_json::json!({ + "providers": { + "provider": { + "type": "openai", + "base_url": "https://example.invalid/v1", + "api_key": "test" + } + }, + "models": { "model": { "model_id": model_id } }, + "agents": { + "default": { "provider": "provider", "model": "model" } + }, + "workspace_dir": workspace + }) + .to_string() + } + + #[test] + fn candidate_accepts_runtime_changes_and_rejects_workspace_changes() { + let temp = tempfile::tempdir().unwrap(); + let workspace = temp.path().join("workspace"); + std::fs::create_dir_all(&workspace).unwrap(); + let config_path = temp.path().join("config.json"); + let current: Config = serde_json::from_str(&config_json(&workspace, "old-model")).unwrap(); + std::fs::write( + &config_path, + config_json(std::path::Path::new("workspace"), "new-model"), + ) + .unwrap(); + + let candidate = load_candidate( + &config_path, + &HashMap::new(), + temp.path(), + ¤t, + &workspace, + ) + .unwrap(); + assert_eq!( + candidate.get_provider_config("default").unwrap().model_id, + "new-model" + ); + + let other_workspace = temp.path().join("other"); + std::fs::create_dir_all(&other_workspace).unwrap(); + std::fs::write(&config_path, config_json(&other_workspace, "new-model")).unwrap(); + let error = load_candidate( + &config_path, + &HashMap::new(), + temp.path(), + ¤t, + &workspace, + ) + .unwrap_err(); + assert!( + error + .to_string() + .contains("workspace_dir cannot be reloaded") + ); + } + + #[test] + fn equivalent_default_database_paths_are_reloadable() { + let temp = tempfile::tempdir().unwrap(); + let workspace = temp.path().join("workspace"); + std::fs::create_dir_all(&workspace).unwrap(); + std::fs::write(workspace.join("picobot.db"), []).unwrap(); + let config_path = temp.path().join("config.json"); + let current: Config = serde_json::from_str(&config_json(&workspace, "old-model")).unwrap(); + let mut candidate: serde_json::Value = + serde_json::from_str(&config_json(&workspace, "new-model")).unwrap(); + candidate["gateway"] = serde_json::json!({ "session_db_path": "./picobot.db" }); + std::fs::write(&config_path, serde_json::to_vec(&candidate).unwrap()).unwrap(); + + load_candidate( + &config_path, + &HashMap::new(), + temp.path(), + ¤t, + &workspace, + ) + .unwrap(); + } + + #[tokio::test] + async fn admission_closes_and_waits_for_existing_activity() { + let admission = RuntimeAdmission::open(); + let activity = admission.try_enter().unwrap(); + admission.close(); + assert!(admission.try_enter().is_none()); + + let waiting = admission.wait_for_idle(); + tokio::pin!(waiting); + assert!( + tokio::time::timeout(std::time::Duration::from_millis(10), &mut waiting) + .await + .is_err() + ); + drop(activity); + tokio::time::timeout(std::time::Duration::from_secs(1), waiting) + .await + .unwrap(); + } + + #[tokio::test] + async fn one_reload_remains_pending_until_its_generation_is_terminal() { + let mut controller = ReloadController::new(HashMap::new(), PathBuf::from(".")); + let handle = controller.handle.clone(); + let first = tokio::spawn({ + let handle = handle.clone(); + async move { handle.request().await } + }); + let request = controller.receiver.recv().await.unwrap(); + assert_eq!(request.generation, 2); + request + .response + .send(Ok(ReloadAccepted { + generation: 2, + message: "accepted".to_string(), + })) + .unwrap(); + first.await.unwrap().unwrap(); + + assert_eq!( + handle.request().await.unwrap_err(), + ReloadError::AlreadyPending + ); + controller.set_phase(2, ReloadPhase::Active); + + let next = tokio::spawn({ + let handle = handle.clone(); + async move { handle.request().await } + }); + let request = controller.receiver.recv().await.unwrap(); + assert_eq!(request.generation, 3); + controller.set_failed(3, "invalid candidate"); + request + .response + .send(Err(ReloadError::InvalidConfig( + "invalid candidate".to_string(), + ))) + .unwrap(); + assert!(matches!( + next.await.unwrap(), + Err(ReloadError::InvalidConfig(_)) + )); + let status = handle.status(); + assert_eq!(status.generation, 3); + assert_eq!(status.phase, ReloadPhase::Failed); + } +} diff --git a/src/gateway/router.rs b/src/gateway/router.rs index d7868ac..e9e4295 100644 --- a/src/gateway/router.rs +++ b/src/gateway/router.rs @@ -8,6 +8,7 @@ use tokio::sync::{Semaphore, mpsc}; use crate::bus::{ControlMessage, InboundMessage, MessageBus, OutboundMessage}; use crate::channels::ChannelError; use crate::channels::parse_slash_command; +use crate::gateway::reload::{ActivityGuard, RuntimeAdmission}; use crate::session::{SessionCommand, SessionEvent, SessionManager}; use crate::task_supervisor::TaskSupervisor; @@ -19,8 +20,14 @@ pub(super) fn spawn_message_routers( bus: Arc, session_manager: Arc, supervisor: TaskSupervisor, + admission: RuntimeAdmission, ) { - spawn_inbound_router(bus.clone(), session_manager.clone(), supervisor.clone()); + spawn_inbound_router( + bus.clone(), + session_manager.clone(), + supervisor.clone(), + admission, + ); spawn_control_router(bus, session_manager, supervisor); } @@ -28,11 +35,12 @@ fn spawn_inbound_router( bus: Arc, session_manager: Arc, supervisor: TaskSupervisor, + admission: RuntimeAdmission, ) { let lane_supervisor = supervisor.clone(); supervisor.spawn("inbound-router", async move { tracing::info!(lane_capacity = INBOUND_LANE_CAPACITY, "Inbound router started"); - let mut lanes: HashMap> = HashMap::new(); + let mut lanes: HashMap> = HashMap::new(); let mut messages_seen = 0_u64; while let Some(inbound) = bus.consume_inbound().await { @@ -41,12 +49,26 @@ fn spawn_inbound_router( lanes.retain(|_, sender| !sender.is_closed()); } + let Some(activity) = admission.try_enter() else { + publish_command_output( + &bus, + inbound, + "Gateway 正在重新加载配置，请稍后重试。".to_string(), + ) + .await; + continue; + }; + let inbound = AdmittedInbound { inbound, activity }; + // Stop must be able to invalidate a running worker even when an // earlier slow slash command occupies this conversation's lane. - if is_priority_stop(&inbound.content) { + if is_priority_stop(&inbound.inbound.content) { let request_bus = bus.clone(); let request_manager = session_manager.clone(); - let task_name = format!("inbound-stop:{}:{}", inbound.channel, inbound.chat_id); + let task_name = format!( + "inbound-stop:{}:{}", + inbound.inbound.channel, inbound.inbound.chat_id + ); if !lane_supervisor.spawn(task_name, async move { process_inbound(request_bus, request_manager, inbound).await; }) { @@ -55,7 +77,7 @@ fn spawn_inbound_router( continue; } - let key = conversation_key(&inbound.channel, &inbound.chat_id); + let key = conversation_key(&inbound.inbound.channel, &inbound.inbound.chat_id); let mut sender = lanes.get(&key).cloned(); if sender.as_ref().is_none_or(mpsc::Sender::is_closed) { let (new_sender, receiver) = mpsc::channel(INBOUND_LANE_CAPACITY); @@ -63,8 +85,8 @@ fn spawn_inbound_router( &lane_supervisor, bus.clone(), session_manager.clone(), - inbound.channel.clone(), - inbound.chat_id.clone(), + inbound.inbound.channel.clone(), + inbound.inbound.chat_id.clone(), receiver, ) { tracing::warn!("Inbound router is stopping"); @@ -81,10 +103,10 @@ fn spawn_inbound_router( match sender.try_send(inbound) { Ok(()) => {} Err(mpsc::error::TrySendError::Full(inbound)) => { - tracing::warn!(channel = %inbound.channel, chat_id = %inbound.chat_id, "Inbound conversation lane is full"); + tracing::warn!(channel = %inbound.inbound.channel, chat_id = %inbound.inbound.chat_id, "Inbound conversation lane is full"); publish_command_output( &bus, - inbound, + inbound.inbound, "当前对话入口队列已满，请稍后重试。".to_string(), ) .await; @@ -98,15 +120,15 @@ fn spawn_inbound_router( &lane_supervisor, bus.clone(), session_manager.clone(), - inbound.channel.clone(), - inbound.chat_id.clone(), + inbound.inbound.channel.clone(), + inbound.inbound.chat_id.clone(), receiver, ) { break; } lanes.insert(key, new_sender.clone()); - if let Err(error) = new_sender.try_send(inbound) { - tracing::error!(error = %error, "Failed to enqueue input into replacement lane"); + if new_sender.try_send(inbound).is_err() { + tracing::error!("Failed to enqueue input into replacement lane"); } } } @@ -121,7 +143,7 @@ fn spawn_inbound_lane( session_manager: Arc, channel: String, chat_id: String, - receiver: mpsc::Receiver, + receiver: mpsc::Receiver, ) -> bool { supervisor.spawn(format!("inbound-lane:{channel}:{chat_id}"), async move { run_ordered_lane(receiver, INBOUND_LANE_IDLE_TIMEOUT, move |inbound| { @@ -152,8 +174,12 @@ async fn run_ordered_lane( async fn process_inbound( bus: Arc, session_manager: Arc, - inbound: InboundMessage, + admitted: AdmittedInbound, ) { + let AdmittedInbound { + inbound, + activity: _activity, + } = admitted; let result = session_manager.handle_message(&inbound).await; match result { @@ -172,14 +198,20 @@ async fn process_inbound( } async fn publish_assistant_output(bus: &MessageBus, inbound: InboundMessage, content: String) { - publish_output(bus, inbound, content, false).await; + publish_output(bus, inbound, content, false, false).await; } async fn publish_command_output(bus: &MessageBus, inbound: InboundMessage, content: String) { - publish_output(bus, inbound, content, true).await; + publish_output(bus, inbound, content, true, true).await; } -async fn publish_output(bus: &MessageBus, inbound: InboundMessage, content: String, command: bool) { +async fn publish_output( + bus: &MessageBus, + inbound: InboundMessage, + content: String, + command: bool, + confirmed: bool, +) { let mut metadata = inbound.channel_context.private; if command { metadata.insert("_type".to_string(), "command".to_string()); @@ -193,11 +225,21 @@ async fn publish_output(bus: &MessageBus, inbound: InboundMessage, content: Stri metadata, delivery: None, }; - if let Err(error) = bus.publish_outbound(outbound).await { + let result = if confirmed { + bus.deliver_outbound(outbound).await + } else { + bus.publish_outbound(outbound).await + }; + if let Err(error) = result { tracing::error!(error = %error, "Failed to publish routed outbound message"); } } +struct AdmittedInbound { + inbound: InboundMessage, + activity: ActivityGuard, +} + fn spawn_control_router( bus: Arc, session_manager: Arc, @@ -382,7 +424,10 @@ mod tests { }, }; - publish_command_output(&bus, inbound, "done".to_string()).await; + let publish_task = tokio::spawn({ + let bus = bus.clone(); + async move { publish_command_output(&bus, inbound, "done".to_string()).await } + }); let output = bus.consume_outbound().await.unwrap(); assert_eq!(output.reply_to.as_deref(), Some("parent")); @@ -394,6 +439,9 @@ mod tests { output.metadata.get("_type").map(String::as_str), Some("command") ); + assert!(!publish_task.is_finished()); + output.complete_delivery(Ok(())); + publish_task.await.unwrap(); } #[tokio::test] diff --git a/src/main.rs b/src/main.rs index 0fd1da5..79d478f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -56,6 +56,12 @@ enum Command { #[arg(long)] port: Option, }, + /// Reload a running gateway's configuration + Reload { + /// Gateway WebSocket or HTTP URL + #[arg(long)] + gateway_url: Option, + }, /// Generate a one-time browser pairing code from the local gateway Pair { /// Gateway WebSocket or HTTP URL @@ -123,6 +129,13 @@ async fn main() -> Result<(), Box> { Command::Gateway { host, port } => { picobot::gateway::run(host, port).await?; } + Command::Reload { gateway_url } => { + let config = picobot::config::Config::load_default().ok(); + let url = gateway_url + .or_else(|| config.as_ref().map(|c| c.client.gateway_url.clone())) + .unwrap_or_else(|| "ws://127.0.0.1:19876/ws".to_string()); + println!("{}", picobot::client::reload_gateway(&url).await?); + } Command::Pair { gateway_url, revoke_all, diff --git a/src/scheduler/mod.rs b/src/scheduler/mod.rs index 22f76df..cffccd9 100644 --- a/src/scheduler/mod.rs +++ b/src/scheduler/mod.rs @@ -97,6 +97,7 @@ pub struct Scheduler { session_manager: Arc, config: SchedulerConfig, owner: String, + admission: crate::gateway::reload::RuntimeAdmission, } impl Scheduler { @@ -104,12 +105,27 @@ impl Scheduler { storage: Arc, session_manager: Arc, config: SchedulerConfig, + ) -> Self { + Self::with_admission( + storage, + session_manager, + config, + crate::gateway::reload::RuntimeAdmission::open(), + ) + } + + pub(crate) fn with_admission( + storage: Arc, + session_manager: Arc, + config: SchedulerConfig, + admission: crate::gateway::reload::RuntimeAdmission, ) -> Self { Self { storage, session_manager, config, owner: uuid::Uuid::new_v4().to_string(), + admission, } } @@ -132,6 +148,9 @@ impl Scheduler { loop { interval.tick().await; + if !self.admission.is_accepting() { + continue; + } let now = now_ms(); let lease_ms = self .config @@ -167,6 +186,16 @@ impl Scheduler { } async fn execute_claimed_job(self: Arc, job: ScheduledJob) { + let Some(_activity) = self.admission.try_enter() else { + if let Err(error) = self + .storage + .release_scheduled_job_lease(&job.id, &self.owner) + .await + { + tracing::error!(job_id = %job.id, error = %error, "scheduler: failed to release job claimed during reload drain"); + } + return; + }; let start = Instant::now(); let started_at = now_ms(); tracing::info!(job_id = %job.id, job_name = %job.name, "scheduler: executing claimed job"); diff --git a/src/session/session.rs b/src/session/session.rs index 31a6d81..bcd08e9 100644 --- a/src/session/session.rs +++ b/src/session/session.rs @@ -1428,6 +1428,7 @@ pub struct SessionManager { sub_agent_manager: Arc, task_supervisor: crate::task_supervisor::TaskSupervisor, turn_delivery: TurnDeliveryService, + reload: crate::gateway::reload::ReloadHandle, } /// Gateway-owned runtime services shared by all Session workers. @@ -1436,6 +1437,8 @@ pub struct SessionManagerServices { memory_manager: Arc, task_supervisor: crate::task_supervisor::TaskSupervisor, turn_delivery: TurnDeliveryService, + reload: crate::gateway::reload::ReloadHandle, + admission: crate::gateway::reload::RuntimeAdmission, } impl SessionManagerServices { @@ -1444,14 +1447,25 @@ impl SessionManagerServices { memory_manager: Arc, task_supervisor: crate::task_supervisor::TaskSupervisor, turn_delivery: TurnDeliveryService, + reload: crate::gateway::reload::ReloadHandle, ) -> Self { Self { bus, memory_manager, task_supervisor, turn_delivery, + reload, + admission: crate::gateway::reload::RuntimeAdmission::open(), } } + + pub(crate) fn with_admission( + mut self, + admission: crate::gateway::reload::RuntimeAdmission, + ) -> Self { + self.admission = admission; + self + } } struct SessionManagerInner { @@ -1544,6 +1558,11 @@ pub static SLASH_COMMANDS: &[SlashCommand] = &[ description: "查看、完成或取消当前任务计划", aliases: &["/todo"], }, + SlashCommand { + name: "reload", + description: "重新加载配置", + aliases: &["/reload"], + }, ]; fn resolve_slash_command(command: &str) -> Option<&'static SlashCommand> { @@ -1581,6 +1600,8 @@ impl SessionManager { memory_manager, task_supervisor, turn_delivery, + reload, + admission, } = services; let mut skills_loader = SkillsLoader::new(); skills_loader.load_skills(); @@ -1608,9 +1629,11 @@ impl SessionManager { Some(skills_loader.clone()), task_supervisor.clone(), ) + .with_admission(admission) .with_work_manager(work_manager.clone()), ); tools.register(crate::tools::DelegateTool::new(sub_agent_manager.clone())); + tools.register(crate::tools::ReloadConfigTool::new(reload.clone())); // Start background task notification consumer let sm_bus = bus.clone(); @@ -1669,6 +1692,7 @@ impl SessionManager { sub_agent_manager, task_supervisor, turn_delivery, + reload, }) } @@ -1689,11 +1713,12 @@ impl SessionManager { /// 为定时任务创建一个无 session 绑定的 AgentLoop pub fn create_cron_agent(&self) -> Result { + let tools = self.tools.without(&["reload_config"]); let provider = create_provider(self.provider_config.clone()) .map_err(|e| AgentError::Other(format!("failed to create cron provider: {}", e)))?; Ok(AgentLoop::with_provider_and_tools( Arc::from(provider), - self.tools.clone(), + tools, self.provider_config.max_tool_iterations, self.provider_config.model_id.clone(), self.provider_config.workspace_dir.clone(), @@ -1710,6 +1735,7 @@ impl SessionManager { "cron_remove", "cron_enable", "cron_disable", + "reload_config", ]); let provider = create_provider(self.provider_config.clone()) .map_err(|e| AgentError::Other(format!("failed to create scheduled provider: {e}")))?; @@ -2081,6 +2107,12 @@ impl SessionManager { }, } } + "reload" => self + .reload + .request() + .await + .map(|accepted| (None, accepted.message)) + .map_err(|error| AgentError::Other(error.to_string())), _ => Err(AgentError::Other(format!( "未知命令：/{}。输入 /? 获取帮助。", cmd.name @@ -2088,6 +2120,43 @@ impl SessionManager { } } + /// Wait until all interactive session Turns have reached a terminal state. + /// A reload uses this to avoid cancelling the Turn that requested it. + pub async fn wait_until_idle(&self, timeout: std::time::Duration) -> bool { + let deadline = tokio::time::Instant::now() + timeout; + let mut idle_since = None; + loop { + let sessions: Vec<_> = { + let inner = self.inner.lock().await; + inner.sessions.values().cloned().collect() + }; + let mut busy = false; + for session in sessions { + let session = session.lock().await; + let queued = session + .agent_tx + .as_ref() + .is_some_and(|sender| sender.capacity() < sender.max_capacity()); + if session.current_cancel.is_some() || queued { + busy = true; + break; + } + } + if busy { + idle_since = None; + } else { + let since = idle_since.get_or_insert_with(tokio::time::Instant::now); + if since.elapsed() >= std::time::Duration::from_millis(100) { + return true; + } + } + if tokio::time::Instant::now() >= deadline { + return false; + } + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + } + pub async fn create_session( &self, channel: &str, @@ -3427,6 +3496,10 @@ mod slash_command_tests { resolve_slash_command("/help").map(|command| command.name), Some("?") ); + assert_eq!( + resolve_slash_command("reload").map(|command| command.name), + Some("reload") + ); assert!(resolve_slash_command("unknown").is_none()); } } diff --git a/src/tools/mod.rs b/src/tools/mod.rs index 0e789c4..074cba5 100644 --- a/src/tools/mod.rs +++ b/src/tools/mod.rs @@ -17,6 +17,7 @@ pub mod memory; pub mod path_utils; pub mod pty; pub mod registry; +pub mod reload_config; pub mod schema; pub mod send_message; pub mod todo; @@ -39,6 +40,7 @@ pub use maintenance::RoutineMaintenanceTool; pub use memory::{MemoryForgetTool, MemoryRecallTool, MemoryStoreTool, TimelineRecallTool}; pub use pty::{PtyManager, PtyTool}; pub use registry::ToolRegistry; +pub use reload_config::ReloadConfigTool; pub use send_message::SendMessageTool; pub use todo::TodoTool; pub use traits::{OutboundDelivery, OutboundMessenger, Tool, ToolResult, ToolResultWithMedia}; diff --git a/src/tools/reload_config.rs b/src/tools/reload_config.rs new file mode 100644 index 0000000..8db0582 --- /dev/null +++ b/src/tools/reload_config.rs @@ -0,0 +1,52 @@ +use async_trait::async_trait; + +use super::{Tool, ToolResult}; +use crate::gateway::reload::ReloadHandle; + +pub struct ReloadConfigTool { + reload: ReloadHandle, +} + +impl ReloadConfigTool { + pub fn new(reload: ReloadHandle) -> Self { + Self { reload } + } +} + +#[async_trait] +impl Tool for ReloadConfigTool { + fn name(&self) -> &str { + "reload_config" + } + + fn description(&self) -> &str { + "重新读取并校验 PicoBot 配置，然后让 Gateway 优雅切换到新配置。仅在用户明确要求重新加载配置时调用。" + } + + fn parameters_schema(&self) -> serde_json::Value { + serde_json::json!({ + "type": "object", + "properties": {}, + "additionalProperties": false + }) + } + + async fn execute(&self, _args: serde_json::Value) -> anyhow::Result { + match self.reload.request().await { + Ok(accepted) => Ok(ToolResult { + success: true, + output: accepted.message, + error: None, + }), + Err(error) => Ok(ToolResult { + success: false, + output: String::new(), + error: Some(error.to_string()), + }), + } + } + + fn exclusive(&self) -> bool { + true + } +} diff --git a/tests/test_config_reload.rs b/tests/test_config_reload.rs new file mode 100644 index 0000000..045e3fd --- /dev/null +++ b/tests/test_config_reload.rs @@ -0,0 +1,146 @@ +use std::net::TcpListener; +use std::path::Path; +use std::process::Stdio; +use std::time::Duration; + +use serde_json::{Value, json}; +use tokio::process::{Child, Command}; + +struct GatewayProcess(Child); + +impl Drop for GatewayProcess { + fn drop(&mut self) { + let _ = self.0.start_kill(); + } +} + +fn available_port() -> u16 { + TcpListener::bind("127.0.0.1:0") + .unwrap() + .local_addr() + .unwrap() + .port() +} + +fn config(workspace: &Path, model_id: &str) -> Value { + json!({ + "providers": { + "provider": { + "type": "openai", + "base_url": "https://example.invalid/v1", + "api_key": "test" + } + }, + "models": { "model": { "model_id": model_id } }, + "agents": { "default": { "provider": "provider", "model": "model" } }, + "gateway": { "require_pairing": false }, + "workspace_dir": workspace + }) +} + +async fn wait_for_health(client: &reqwest::Client, base: &str) { + for _ in 0..100 { + if client + .get(format!("{base}/health")) + .send() + .await + .is_ok_and(|response| response.status().is_success()) + { + return; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + panic!("gateway did not become healthy"); +} + +#[tokio::test] +async fn gateway_reloads_valid_config_and_keeps_serving_after_invalid_config() { + let temp = tempfile::tempdir().unwrap(); + let home = temp.path().join("home"); + let config_dir = home.join(".picobot"); + let workspace = temp.path().join("workspace"); + std::fs::create_dir_all(&config_dir).unwrap(); + std::fs::create_dir_all(&workspace).unwrap(); + let config_path = config_dir.join("config.json"); + std::fs::write( + &config_path, + serde_json::to_vec_pretty(&config(&workspace, "old-model")).unwrap(), + ) + .unwrap(); + + let port = available_port(); + let child = Command::new(env!("CARGO_BIN_EXE_picobot")) + .args([ + "gateway", + "--host", + "127.0.0.1", + "--port", + &port.to_string(), + ]) + .env("HOME", &home) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .kill_on_drop(true) + .spawn() + .unwrap(); + let mut gateway = GatewayProcess(child); + let client = reqwest::Client::new(); + let base = format!("http://127.0.0.1:{port}"); + wait_for_health(&client, &base).await; + + std::fs::write( + &config_path, + serde_json::to_vec_pretty(&config(&workspace, "new-model")).unwrap(), + ) + .unwrap(); + let accepted: Value = client + .post(format!("{base}/api/config/reload")) + .send() + .await + .unwrap() + .error_for_status() + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(accepted["generation"], 2); + + let mut active = false; + for _ in 0..100 { + let response = client + .get(format!("{base}/api/config/reload/status")) + .send() + .await; + if let Ok(response) = response + && let Ok(status) = response.json::().await + && status["generation"] == 2 + && status["phase"] == "active" + { + active = true; + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + assert!(active, "reloaded generation did not become active"); + + std::fs::write(&config_path, b"{").unwrap(); + let invalid = client + .post(format!("{base}/api/config/reload")) + .send() + .await + .unwrap(); + assert_eq!(invalid.status(), reqwest::StatusCode::BAD_REQUEST); + assert!( + client + .get(format!("{base}/health")) + .send() + .await + .unwrap() + .status() + .is_success(), + "invalid candidate stopped the active generation" + ); + + gateway.0.start_kill().unwrap(); + gateway.0.wait().await.unwrap(); +} diff --git a/webui/package-lock.json b/webui/package-lock.json index 14f452b..e8dd7ea 100644 --- a/webui/package-lock.json +++ b/webui/package-lock.json @@ -1,12 +1,12 @@ { "name": "picobot-webui", - "version": "1.2.2", + "version": "1.3.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "picobot-webui", - "version": "1.2.2", + "version": "1.3.0", "dependencies": { "bits-ui": "^2.0.0", "dompurify": "^3.4.12", diff --git a/webui/package.json b/webui/package.json index 33badae..7616b81 100644 --- a/webui/package.json +++ b/webui/package.json @@ -1,7 +1,7 @@ { "name": "picobot-webui", "private": true, - "version": "1.2.2", + "version": "1.3.0", "type": "module", "engines": { "node": ">=20"