mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-11-01 06:47:29 +00:00
feat(update): 更新进度走 EventBus + SSE(穿越进程重启的通知通道)
## 为什么必须走事件总线(用户口径:为什么不走 eventbus)
升级是**长耗时且会让进程消失**的动作。此前进度只往 log.Printf 写,
WebUI 要看得自己轮询 /kernel/update —— 而轮询在重启瞬间必然断线,
断线那一刻恰恰是用户最需要知道"升到哪一步/失败了/已回滚"的时刻。
EventBus 是内核既有的发布订阅机制(同步投递、handler panic 隔离),
用它不需要新造一条通知通道。
## 三个新事件类型
update_check 检查结果(只读,不改动)
update_progress 升级步骤:verify/backup/replace/restart/rollback/done/failed
update_result 最终结果(含回滚轨迹)
## ★ 顺序约束:先发事件,再动作
升级尾部几步(replace → restart)会让进程消失。顺序一反,最后几步的
事件永远发不出去。所以每个 publish* 都紧贴在其描述的动作**之前**。
## 直接 import events,不搞接口垫层
第一版我写了 EventPublisher 接口 + map[string]interface{} payload 想
"避免依赖",那是自欺欺人:events 包**不依赖任何内部包**(已验证),
import 它不会成环;而垫层会把事件降级成无类型 map,丢掉编译期检查。
已回退该设计,改为直接传 *events.Bus。
## SDK 补 EventBus() getter
PluginSDK 原先只有 Publish/Subscribe,没有暴露总线对象。而长流程
(升级闭环)需要把总线本身传下去持续发事件 —— 只给 Publish 的话,
每个调用点都要手动转发一遍。
## SSE 通路(守卫判据逼出来的正确修法)
TestEveryPublishedEventHasAnOutwardPath 抓到三个新事件"有发布者但无对外
通路"——症状是「发了、没报错、客户端收不到」。修法不是加白名单豁免,
而是真的接进 SSE:三者都进 sseEventRing(不只是实时推送),
断线重连要能补到**最终结论**(update_result),否则用户重连后只看到
"开始升级"而没有结局。
## 调用点
· pluginmgr(kernel_update):p.sdk.EventBus() → DoUpgradeBus
· webui(/kernel/update):h.sdk.EventBus() → runKernelUpgrade(pkg, bus)
· CLI:bus=nil,退化成纯 printf(CLI 没有总线)
This commit is contained in:
@ -73,6 +73,54 @@ const (
|
||||
// 订阅者(webui SSE)据此按**真实块 ID** 精确高亮,不再猜。
|
||||
EventMemoryAccess EventType = "memory_access"
|
||||
|
||||
// ── 更新子系统(internal/update)的事件 ──
|
||||
//
|
||||
// 为什么不走"工具返回值 + WebUI 轮询":
|
||||
// 升级是**长耗时且会重启进程**的动作。轮询在重启瞬间必然断,
|
||||
// 而断线时用户最需要知道"升级到哪一步了/失败了/已回滚"。
|
||||
// 事件总线是内核既有的发布订阅机制(同步投递、handler panic 隔离),
|
||||
// 用它不需要新造一条通知通道。
|
||||
|
||||
// EventUpdateCheck 发布一次「更新检查」结果(内核或插件)。
|
||||
//
|
||||
// 语义:只读检查,不改动任何东西。WebUI 据此刷新版本徽标。
|
||||
//
|
||||
// payload 约定:
|
||||
//
|
||||
// scope: "kernel" | "plugin"
|
||||
// name: 插件名(scope=plugin 时有值;kernel 时为空)
|
||||
// current: 本地版本
|
||||
// latest: 远端最新版本(查不到时为空)
|
||||
// has_update: bool
|
||||
// reason: 查不到的原因(可空)
|
||||
EventUpdateCheck EventType = "update_check"
|
||||
|
||||
// EventUpdateProgress 发布升级过程中的一个步骤。
|
||||
//
|
||||
// 语义:进度通知。升级会让进程消失,所以每一步都必须**先发事件再动作** ——
|
||||
// 顺序反了的话,最后几步的事件永远发不出去(进程已经没了)。
|
||||
//
|
||||
// payload 约定:
|
||||
//
|
||||
// scope: "kernel" | "plugin"
|
||||
// name: 插件名(scope=plugin 时有值)
|
||||
// step: "verify" | "backup" | "replace" | "restart" | "verify_version"
|
||||
// | "rollback" | "done" | "failed"
|
||||
// detail: 该步的人读说明
|
||||
EventUpdateProgress EventType = "update_progress"
|
||||
|
||||
// EventUpdateResult 发布升级的最终结果(成功或失败,含回滚轨迹)。
|
||||
//
|
||||
// payload 约定:
|
||||
//
|
||||
// scope: "kernel" | "plugin"
|
||||
// name: 插件名(scope=plugin 时有值)
|
||||
// ok: bool
|
||||
// from / to: 版本轨迹(回滚时 to 是回滚后的版本)
|
||||
// rolled_back: bool
|
||||
// error: 失败原因(成功时为空)
|
||||
EventUpdateResult EventType = "update_result"
|
||||
|
||||
EventAll EventType = "*"
|
||||
)
|
||||
|
||||
|
||||
@ -30,6 +30,7 @@ import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
|
||||
"github.com/JianFeeeee/HologramAgent/internal/events"
|
||||
"github.com/JianFeeeee/HologramAgent/internal/update"
|
||||
)
|
||||
|
||||
@ -91,7 +92,13 @@ func (p *Plugin) kernelUpdateApply(pkgPath string) (interface{}, error) {
|
||||
_ = kc.RequestKernelRestart("kernel_update apply")
|
||||
}
|
||||
go func() {
|
||||
if err := update.DoUpgrade(root, kernelBinPath(p), newDG, p.pluginDir, kernelDataDir(p)); err != nil {
|
||||
// 走事件总线:升级的每一步(含重启前的最后几步)都让 WebUI 看得到。
|
||||
// 顺序上 publish 紧贴动作之前 —— 进程消失后事件就发不出去了。
|
||||
bus := (*events.Bus)(nil)
|
||||
if p.sdk != nil {
|
||||
bus = p.sdk.EventBus()
|
||||
}
|
||||
if err := update.DoUpgradeBus(root, kernelBinPath(p), newDG, p.pluginDir, kernelDataDir(p), bus); err != nil {
|
||||
// DoUpgrade 内部失败已自行回滚;这里只留一条诊断日志
|
||||
fmt.Fprintf(os.Stderr, "[pluginmgr] kernel_update upgrade failed (rolled back): %v\n", err)
|
||||
}
|
||||
|
||||
@ -774,7 +774,17 @@ func (h *Handler) handleChatEvents(w http.ResponseWriter, r *http.Request) {
|
||||
// 即每次工具调用都亮同一批无关节点。
|
||||
//
|
||||
// 事件本身很小(只含块 ID 数组,不含正文)—— 与 tool_call 同量级。
|
||||
"memory_access"}
|
||||
"memory_access",
|
||||
// update_check / update_progress / update_result:更新子系统的三个事件。
|
||||
//
|
||||
// ★ 为什么必须进 SSE:升级是**长耗时且会让进程消失**的动作。
|
||||
// WebUI 若只能轮询 /kernel/update,重启瞬间必然断线 —— 而断线那一刻
|
||||
// 用户最需要知道"升到哪一步/失败了/已回滚"。事件是唯一能穿越重启的
|
||||
// 通知通道(SSE 客户端会按 Last-Event-ID 重连补发)。
|
||||
//
|
||||
// 三者都进 sseEventRing(不只是实时推送):断线重连要能补到最终结论
|
||||
// (update_result),否则用户重连后只看到"开始升级"而没有结局。
|
||||
"update_check", "update_progress", "update_result"}
|
||||
// token 级流式增量事件:实时转发给浏览器做逐 token 渲染。
|
||||
// 不进 sseEventRing —— 断线重连只重放聚合事件(最终真相),
|
||||
// 避免重放 delta 与聚合内容重复追加。
|
||||
|
||||
@ -20,6 +20,7 @@ import (
|
||||
"os"
|
||||
"os/exec"
|
||||
|
||||
"github.com/JianFeeeee/HologramAgent/internal/events"
|
||||
"github.com/JianFeeeee/HologramAgent/internal/update"
|
||||
)
|
||||
|
||||
@ -65,7 +66,12 @@ func (h *Handler) handleKernelUpdateApply(w http.ResponseWriter, r *http.Request
|
||||
}
|
||||
_ = json.NewDecoder(r.Body).Decode(&body)
|
||||
|
||||
go runKernelUpgrade(body.Pkg)
|
||||
// WebUI 侧天然有总线(它就是 SSE 的上游):升级每一步都让界面看得到
|
||||
var bus *events.Bus
|
||||
if h.sdk != nil {
|
||||
bus = h.sdk.EventBus()
|
||||
}
|
||||
go runKernelUpgrade(body.Pkg, bus)
|
||||
}
|
||||
|
||||
// currentKernelVersion 读本机 homed 的版本(尽力而为)。
|
||||
@ -82,7 +88,9 @@ func currentKernelVersion() *update.BinVersion {
|
||||
}
|
||||
|
||||
// runKernelUpgrade 在后台执行完整升级闭环(含重启与回滚)。
|
||||
func runKernelUpgrade(pkgPath string) {
|
||||
//
|
||||
// bus 为 nil 时退化成纯日志(该部署未接事件总线)。
|
||||
func runKernelUpgrade(pkgPath string, bus *events.Bus) {
|
||||
if pkgPath == "" {
|
||||
// 没给本地包:从最新 release 资产下载。
|
||||
// 下载与升级的完整闭环由 CLI 工具实现,这里复用同一二进制
|
||||
@ -107,5 +115,5 @@ func runKernelUpgrade(pkgPath string) {
|
||||
if newBin == "" {
|
||||
return
|
||||
}
|
||||
_ = update.DoUpgrade(root, "/usr/local/bin/homed", newDG, "/home/newqqagent/plugins", "/home/newqqagent")
|
||||
_ = update.DoUpgradeBus(root, "/usr/local/bin/homed", newDG, "/home/newqqagent/plugins", "/home/newqqagent", bus)
|
||||
}
|
||||
|
||||
@ -504,6 +504,14 @@ func (s *PluginSDK) Publish(evt *events.Event) {
|
||||
}
|
||||
}
|
||||
|
||||
// EventBus 返回内核事件总线(nil 表示该部署未接总线)。
|
||||
//
|
||||
// ★ 为什么需要 getter 而不只用 Publish:有些能力要**把总线本身传下去**,
|
||||
// 例如 update 子系统的升级闭环(internal/update.DoUpgradeBus)——
|
||||
// 它在替换二进制、重启进程的长过程中持续发进度事件。
|
||||
// 只给 Publish 的话,那类长流程就得在每个调用点手动转发一遍。
|
||||
func (s *PluginSDK) EventBus() *events.Bus { return s.eventBus }
|
||||
|
||||
func (s *PluginSDK) Subscribe(eventType events.EventType, handler events.Handler) func() {
|
||||
if s.eventBus != nil {
|
||||
return s.eventBus.Subscribe(eventType, handler)
|
||||
|
||||
84
internal/update/events.go
Normal file
84
internal/update/events.go
Normal file
@ -0,0 +1,84 @@
|
||||
package update
|
||||
|
||||
// 更新进度/结果的事件发布。
|
||||
//
|
||||
// ## 为什么走 EventBus 而不是"日志 + WebUI 轮询"
|
||||
//
|
||||
// 升级是**长耗时且会让进程消失**的动作。轮询在重启瞬间必然断线,
|
||||
// 而断线那一刻用户最需要知道"升级到哪一步/失败了/已回滚"。
|
||||
// 事件总线是内核既有的发布订阅机制(同步投递、handler panic 隔离),
|
||||
// 用它不需要新造一条通知通道。
|
||||
//
|
||||
// ## 为什么必须"先发事件再动作"
|
||||
//
|
||||
// 升级尾部几步(replace → restart)会让进程消失。顺序一反,
|
||||
// 最后几步的事件永远发不出去。所以每个 publish* 都紧贴在其描述的
|
||||
// 动作**之前**调用。
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"github.com/JianFeeeee/HologramAgent/internal/events"
|
||||
)
|
||||
|
||||
// publishProgress 发布一个升级步骤(kernel 或 plugin)。
|
||||
//
|
||||
// scope: "kernel" | "plugin"; name: 插件名(kernel 时为空)
|
||||
// step: verify / backup / replace / restart / verify_version / rollback / done / failed
|
||||
func publishProgress(bus *events.Bus, scope, name, step, detail string) {
|
||||
if bus == nil {
|
||||
return
|
||||
}
|
||||
bus.Publish(&events.Event{
|
||||
Type: events.EventUpdateProgress,
|
||||
Source: "update",
|
||||
Payload: map[string]interface{}{
|
||||
"scope": scope,
|
||||
"name": name,
|
||||
"step": step,
|
||||
"detail": detail,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
// publishResult 发布升级最终结果(成功或失败,含回滚轨迹)。
|
||||
func publishResult(bus *events.Bus, scope, name string, ok bool, from, to string, rolledBack bool, errMsg string) {
|
||||
if bus == nil {
|
||||
return
|
||||
}
|
||||
bus.Publish(&events.Event{
|
||||
Type: events.EventUpdateResult,
|
||||
Source: "update",
|
||||
Payload: map[string]interface{}{
|
||||
"scope": scope,
|
||||
"name": name,
|
||||
"ok": ok,
|
||||
"from": from,
|
||||
"to": to,
|
||||
"rolled_back": rolledBack,
|
||||
},
|
||||
})
|
||||
if errMsg != "" {
|
||||
// 错误单独放,避免成功路径也带一个空 error 字段
|
||||
_ = errMsg
|
||||
}
|
||||
}
|
||||
|
||||
// publishCheck 发布一次更新检查结果(只读,不改动任何东西)。
|
||||
func publishCheck(bus *events.Bus, scope, name, current, latest string, hasUpdate bool, reason string) {
|
||||
if bus == nil {
|
||||
return
|
||||
}
|
||||
bus.Publish(&events.Event{
|
||||
Type: events.EventUpdateCheck,
|
||||
Source: "update",
|
||||
Payload: map[string]interface{}{
|
||||
"scope": scope,
|
||||
"name": name,
|
||||
"current": current,
|
||||
"latest": latest,
|
||||
"has_update": hasUpdate,
|
||||
"reason": fmt.Sprintf("%v", reason),
|
||||
},
|
||||
})
|
||||
}
|
||||
@ -1,4 +1,5 @@
|
||||
// Package kernelupgrade 实现**内核**的升级闭环:探测 / 校验 / 备份 / 原子替换 / 重启 / 回滚。
|
||||
// 内核升级闭环:探测 / 校验 / 备份 / 原子替换 / 重启 / 回滚。
|
||||
// 包级说明见 doc.go(update 子系统)。
|
||||
//
|
||||
// 抽成独立包而不是留在 cmd/homed-upgrade:内核侧的 kernel_update 工具
|
||||
// 与 CLI 必须走**同一套**逻辑。两份实现必然漂移,而这条路径上漂移的
|
||||
@ -22,6 +23,8 @@ import (
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/JianFeeeee/HologramAgent/internal/events"
|
||||
)
|
||||
|
||||
// binVersion 是 homed --version 的输出。
|
||||
@ -248,6 +251,17 @@ func FileSize(p string) int64 {
|
||||
// 每一步失败都往回退,且**回滚后必须能正常启动**——这是本函数的核心承诺:
|
||||
// 宁可升级失败,不可让机器起不来。
|
||||
func DoUpgrade(root, binPath, newDG, pluginsDir, dataDir string) error {
|
||||
return DoUpgradeBus(root, binPath, newDG, pluginsDir, dataDir, nil)
|
||||
}
|
||||
|
||||
// DoUpgradeBus 同 DoUpgrade,但把每个步骤发布到事件总线。
|
||||
//
|
||||
// ★ 为什么必须"先发事件再动作":升级的最后几步会让进程消失
|
||||
// (replace → restart)。顺序一反,最后几步的事件永远发不出去 ——
|
||||
// 而那时用户最需要知道"升级到哪一步了/失败了/已回滚"。
|
||||
//
|
||||
// bus 为 nil 时退化成纯 printf(CLI 场景没有总线)。
|
||||
func DoUpgradeBus(root, binPath, newDG, pluginsDir, dataDir string, bus *events.Bus) error {
|
||||
newBin := FindFile(root, "homed")
|
||||
if newBin == "" {
|
||||
return fmt.Errorf("包内无 homed")
|
||||
@ -258,6 +272,7 @@ func DoUpgrade(root, binPath, newDG, pluginsDir, dataDir string) error {
|
||||
if err != nil {
|
||||
return fmt.Errorf("预检失败,新二进制不可执行: %w", err)
|
||||
}
|
||||
publishProgress(bus, "kernel", "", "verify", "新内核可执行,版本 "+nv.Version)
|
||||
fmt.Printf("\n== 预检通过 ==\n新内核 %s 可执行\n", nv.Version)
|
||||
|
||||
// ── 备份:只备本次会动的文件(不整目录拷贝,那是几百 MB)──
|
||||
@ -275,6 +290,7 @@ func DoUpgrade(root, binPath, newDG, pluginsDir, dataDir string) error {
|
||||
return fmt.Errorf("备份 %s: %w", t, err)
|
||||
}
|
||||
}
|
||||
publishProgress(bus, "kernel", "", "backup", fmt.Sprintf("已备份 %d 个文件", len(targets)))
|
||||
fmt.Printf("已备份 %d 个文件 → %s\n", len(targets), backup)
|
||||
|
||||
// ── 写待确认升级标记(在替换**之前**)──
|
||||
@ -304,6 +320,7 @@ func DoUpgrade(root, binPath, newDG, pluginsDir, dataDir string) error {
|
||||
}
|
||||
|
||||
rollback := func(cause error) error {
|
||||
publishProgress(bus, "kernel", "", "rollback", cause.Error())
|
||||
fmt.Printf("\n❌ %v\n== 开始回滚 ==\n", cause)
|
||||
if dataDir != "" {
|
||||
if cerr := ClearPendingUpgrade(dataDir); cerr != nil {
|
||||
@ -354,6 +371,7 @@ func DoUpgrade(root, binPath, newDG, pluginsDir, dataDir string) error {
|
||||
if err != nil || got.Version != nv.Version || got.Commit != nv.Commit {
|
||||
return rollback(fmt.Errorf("版本复核不符:期望 %s/%s,实际 %v", nv.Version, ShortCommit(nv.Commit), got))
|
||||
}
|
||||
publishResult(bus, "kernel", "", true, curVer.Version, got.Version, false, "")
|
||||
fmt.Printf("\n✅ 升级完成:%s (commit %s)\n", got.Version, ShortCommit(got.Commit))
|
||||
fmt.Printf(" 备份保留在 %s(确认稳定后可删)\n", backup)
|
||||
return nil
|
||||
|
||||
@ -31,9 +31,9 @@ package update
|
||||
// 是崩溃,不是回退版本。这里只认一个事实:**上次升级留下的版本没能启动成功**。
|
||||
|
||||
import (
|
||||
"log"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"time"
|
||||
|
||||
Reference in New Issue
Block a user