diff --git a/internal/events/bus.go b/internal/events/bus.go index ddc744a7..1a7f254b 100644 --- a/internal/events/bus.go +++ b/internal/events/bus.go @@ -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 = "*" ) diff --git a/internal/plugins/pluginmgr/kernel_update.go b/internal/plugins/pluginmgr/kernel_update.go index dde21e8b..57976e13 100644 --- a/internal/plugins/pluginmgr/kernel_update.go +++ b/internal/plugins/pluginmgr/kernel_update.go @@ -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) } diff --git a/internal/plugins/webui/handler_chat.go b/internal/plugins/webui/handler_chat.go index fc6fcf93..958b7a1f 100644 --- a/internal/plugins/webui/handler_chat.go +++ b/internal/plugins/webui/handler_chat.go @@ -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 与聚合内容重复追加。 diff --git a/internal/plugins/webui/kernel_update.go b/internal/plugins/webui/kernel_update.go index d1ddd6b3..09ff464a 100644 --- a/internal/plugins/webui/kernel_update.go +++ b/internal/plugins/webui/kernel_update.go @@ -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) } diff --git a/internal/sdk/plugin.go b/internal/sdk/plugin.go index 96b884a1..9990db84 100644 --- a/internal/sdk/plugin.go +++ b/internal/sdk/plugin.go @@ -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) diff --git a/internal/update/events.go b/internal/update/events.go new file mode 100644 index 00000000..568d03e4 --- /dev/null +++ b/internal/update/events.go @@ -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), + }, + }) +} diff --git a/internal/update/kernel.go b/internal/update/kernel.go index 4d9f6f61..637f10a9 100644 --- a/internal/update/kernel.go +++ b/internal/update/kernel.go @@ -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 diff --git a/internal/update/rollback.go b/internal/update/rollback.go index 9a63487b..f4efebc4 100644 --- a/internal/update/rollback.go +++ b/internal/update/rollback.go @@ -31,9 +31,9 @@ package update // 是崩溃，不是回退版本。这里只认一个事实：**上次升级留下的版本没能启动成功**。 import ( - "log" "encoding/json" "fmt" + "log" "os" "path/filepath" "time"