diff --git a/cmd/homed/bootstrap.go b/cmd/homed/bootstrap.go index 9b4c000..7a3944f 100644 --- a/cmd/homed/bootstrap.go +++ b/cmd/homed/bootstrap.go @@ -843,3 +843,27 @@ func parseFlags() options { flag.Parse() return options{dataDir: *dataDir, httpAddr: *httpAddr, cliSocket: *cliSocket, role: *role, boot: *boot} } + +// compactConfigDB 在空闲页够多时压缩配置库;失败只告警(不影响启动)。 +// +// 触发条件(见 internal/config.MaybeCompact):空闲页 >= 1MB 且占页数 >= 25%。 +// 放在插件加载之后调用——迁移/清理大值发生在插件 Start 里,之前调用没有意义。 +func compactConfigDB(cfgReg *internalConfig.ConfigRegistry) { + before := int64(-1) + if st, err := os.Stat(cfgReg.DBPath()); err == nil { + before = st.Size() + } + done, err := cfgReg.MaybeCompact(1<<20, 0.25) + if err != nil { + log.Printf("[homed] warning: 配置库压缩失败: %v", err) + return + } + if !done { + return + } + after := before + if st, err := os.Stat(cfgReg.DBPath()); err == nil { + after = st.Size() + } + log.Printf("[homed] 配置库已压缩: %d -> %d 字节", before, after) +} diff --git a/cmd/homed/main.go b/cmd/homed/main.go index a5db61b..d5888cd 100644 --- a/cmd/homed/main.go +++ b/cmd/homed/main.go @@ -162,6 +162,10 @@ func main() { loadPlugins(cfg, cfgReg, pluginReg, stageHost, opt.boot, opt.dataDir) + // 插件加载完成后再回收空闲页:大值(如老版聊天记录)可能在这一步被搬走/删除, + // 而 SQLite 的 DELETE 不会缩小文件。 + compactConfigDB(cfgReg) + stopRuntime := startAgentRuntime(cfgReg, pluginReg, agent, logDir, ctx) defer stopRuntime() diff --git a/internal/config/registry.go b/internal/config/registry.go index 81dbd64..fe98223 100644 --- a/internal/config/registry.go +++ b/internal/config/registry.go @@ -494,6 +494,49 @@ func (r *ConfigRegistry) Close() error { return r.db.Close() } +// MaybeCompact 在「空闲页占比高且绝对量够大」时做一次 VACUUM,把文件真正缩回去。 +// +// 为什么需要:SQLite 的 DELETE 只把页标成空闲(进 freelist),文件体积不变。 +// 典型场景是聊天记录那类大值被搬走/删除后,config.db 仍占着几 MB。 +// 只在空闲页超过 minFreeBytes 且占比 >= minRatio 时动手,避免每次启动都重写整个库。 +// 返回是否真的执行了 VACUUM。 +func (r *ConfigRegistry) MaybeCompact(minFreeBytes int64, minRatio float64) (bool, error) { + if r.dbPath == "" || r.dbPath == ":memory:" { + return false, nil + } + r.mu.Lock() + defer r.mu.Unlock() + var pageCount, freeCount, pageSize int64 + if err := r.db.QueryRow("PRAGMA page_count").Scan(&pageCount); err != nil { + return false, err + } + if err := r.db.QueryRow("PRAGMA freelist_count").Scan(&freeCount); err != nil { + return false, err + } + if err := r.db.QueryRow("PRAGMA page_size").Scan(&pageSize); err != nil { + return false, err + } + if pageCount == 0 { + return false, nil + } + freeBytes := freeCount * pageSize + if freeBytes < minFreeBytes || float64(freeCount)/float64(pageCount) < minRatio { + return false, nil + } + if _, err := r.db.Exec("VACUUM"); err != nil { + return false, err + } + // 库是 WAL 模式:VACUUM 的结果先落进 -wal,必须再 checkpoint(TRUNCATE) + // 才会真正回写主库文件并缩小它(否则文件大小看着没变)。 + if _, err := r.db.Exec("PRAGMA wal_checkpoint(TRUNCATE)"); err != nil { + return true, err + } + return true, nil +} + +// DBPath 返回配置库路径(空/内存库时为空串)。 +func (r *ConfigRegistry) DBPath() string { return r.dbPath } + var defaultSources = map[string]map[string]string{ "deepseek": {"base_url": "https://api.deepseek.com", "model": "deepseek-v4-flash", "api_key": "", "thinking_enabled": "false", "adapter": "deepseek", "adapter_path": "adapters/deepseek.lua"}, } diff --git a/internal/config/registry_test.go b/internal/config/registry_test.go index be393d0..d5ae878 100644 --- a/internal/config/registry_test.go +++ b/internal/config/registry_test.go @@ -1,6 +1,7 @@ package config import ( + "os" "path/filepath" "strings" "testing" @@ -452,3 +453,51 @@ func keysOf(defs []*ConfigDef) []string { } return out } + +// TestMaybeCompactReclaimsFreePages 钉住「删除大值后文件要真的缩回去」。 +// SQLite 的 DELETE 只把页标空闲,文件体积不变;v1.3.12 之前那条 5MB 的 +// plugin.webui.chathistory 就是这样让 config.db 一直占着几 MB。 +func TestMaybeCompactReclaimsFreePages(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "config.db") + r := NewConfigRegistry(path) + r.RegisterDef(ConfigDef{Key: "plugin.webui.chathistory", Default: ""}) + + big := strings.Repeat("x", 512*1024) + r.Set("plugin.webui.chathistory", big) + if err := r.Flush(); err != nil { + t.Fatalf("Flush: %v", err) + } + before, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + // 阈值远比这次写入小 → 搬走大值后应触发压缩 + if err := r.Delete("plugin.webui.chathistory"); err != nil { + t.Fatalf("Remove: %v", err) + } + done, err := r.MaybeCompact(1, 0.1) + if err != nil { + t.Fatalf("MaybeCompact: %v", err) + } + if !done { + t.Fatal("空闲页远大于阈值时应执行 VACUUM") + } + after, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + if after.Size() >= before.Size() { + t.Fatalf("压缩后文件应变小:%d -> %d", before.Size(), after.Size()) + } + // 数据仍在(压缩不能破坏内容) + r.Set("core.llm.model", "m1") + if v, err := r.Get("core.llm.model"); err != nil || v != "m1" { + t.Fatalf("压缩后读写异常: %v %v", v, err) + } + // 没有空闲页时不应白做功 + if done2, err := r.MaybeCompact(1<<30, 0.9); err != nil || done2 { + t.Fatalf("阈值很高时不应压缩,得到 done=%v err=%v", done2, err) + } + r.Close() +} diff --git a/internal/plugins/webui/handler.go b/internal/plugins/webui/handler.go index b91ce70..3abfb2b 100644 --- a/internal/plugins/webui/handler.go +++ b/internal/plugins/webui/handler.go @@ -92,6 +92,16 @@ type Handler struct { // 插件设置 history_file 可改)。 history *historyStore + // 聊天记录写盘节流:chatDirty 由 persistChatLocked 置位, + // chatPersistLoop 合并连续变更后落盘(chatSaveThrottle 去抖 + + // chatSaveMaxDelay 兜底),Close 前强制落最后一次。 + chatDirty bool + chatDirtySince time.Time + chatSaveWake chan struct{} + chatStop chan struct{} + chatLoopDone chan struct{} + chatCloseOnce sync.Once + chatMsgMu sync.Mutex chatMsgCache map[string]*chatMsgEntry // client_msg_id -> 首次处理结果 chatMsgOrder []string // FIFO 淘汰序 @@ -146,7 +156,12 @@ func NewHandler(s *sdk.PluginSDK) *Handler { // 聊天记录独立存储:默认 /webui_chat_history.json, // 插件设置 history_file 可改(相对路径按 data 目录解析)。 history: newHistoryStore(resolveHistoryFile(settingString(se, "history_file"), webDataDir)), + + chatSaveWake: make(chan struct{}, 1), + chatStop: make(chan struct{}), + chatLoopDone: make(chan struct{}), } + go h.chatPersistLoop() h.loadChatHistory() if s != nil { go h.trackToolEvents() diff --git a/internal/plugins/webui/handler_chat.go b/internal/plugins/webui/handler_chat.go index 2c0bed8..66b606d 100644 --- a/internal/plugins/webui/handler_chat.go +++ b/internal/plugins/webui/handler_chat.go @@ -276,6 +276,98 @@ func (h *Handler) subscribeChatEvents() { // pendingAssistantLocked 返回 chatHistory 中当前进行中的 assistant 消息(已持有 chatMu)。 // 仅当最后一条是 assistant 且尚未产出最终内容时视为进行中,避免跨轮次误合并。 +// chatSaveThrottle 控制写盘频率:变更后延迟这么久落盘,合并连续更新。 +const chatSaveThrottle = 3 * time.Second + +// chatSaveMaxDelay 是连续写入时的强制落盘上限:聊天再密也不超过这么久。 +const chatSaveMaxDelay = 10 * time.Second + +// persistChatLocked 标记聊天记录待写盘(调用方已持 chatMu)。 +// +// 真正的写盘在 chatPersistLoop 里做,并带节流:原先这里是**每条消息都整段 +// 重写一次记录文件**,而一轮对话会触发多次(用户消息、每个工具事件、收尾消息)—— +// 200 条上限下文件可达数 MB,于是单轮就放大出几十 MB 写。 +func (h *Handler) persistChatLocked() { + if h.history == nil { + return + } + if !h.chatDirty { + h.chatDirtySince = time.Now() + } + h.chatDirty = true + select { + case h.chatSaveWake <- struct{}{}: + default: // 已有待处理信号,合并即可 + } +} + +// chatPersistLoop 把聊天记录按节流节奏落盘,直到 Close。 +func (h *Handler) chatPersistLoop() { + defer close(h.chatLoopDone) + timer := time.NewTimer(chatSaveThrottle) + timer.Stop() + defer timer.Stop() + for { + select { + case <-h.chatStop: + h.flushChat() // 关停前把最后一次变更写下去 + return + case <-h.chatSaveWake: + delay := chatSaveThrottle + h.chatMu.Lock() + if !h.chatDirtySince.IsZero() { + if left := chatSaveMaxDelay - time.Since(h.chatDirtySince); left < delay { + if left < 0 { + left = 0 + } + delay = left + } + } + h.chatMu.Unlock() + timer.Reset(delay) + case <-timer.C: + h.flushChat() + } + } +} + +// flushChat 把当前聊天记录快照写盘。文件 IO 不持 chatMu(快照拷出来再写), +// 写失败则重新标脏,等下一轮重试。 +func (h *Handler) flushChat() { + if h.history == nil { + return + } + h.chatMu.Lock() + if !h.chatDirty { + h.chatMu.Unlock() + return + } + h.chatDirty = false + h.chatDirtySince = time.Time{} + msgs := make([]ChatMsg, len(h.chatHistory)) + copy(msgs, h.chatHistory) + h.chatMu.Unlock() + + if err := h.history.Save(msgs); err != nil { + log.Printf("[webui] 写聊天记录 %s 失败(稍后重试): %v", h.history.Path(), err) + h.chatMu.Lock() + h.chatDirty = true + if h.chatDirtySince.IsZero() { + h.chatDirtySince = time.Now() + } + h.chatMu.Unlock() + } +} + +// Close 停掉写盘协程并把最后一次变更落盘(幂等)。 +// 由插件 Stop 调用;不这样做会丢掉最后一轮对话。 +func (h *Handler) Close() { + h.chatCloseOnce.Do(func() { + close(h.chatStop) + <-h.chatLoopDone + }) +} + func (h *Handler) pendingAssistantLocked() *ChatMsg { if h.pendingIdx < 0 || h.pendingIdx >= len(h.chatHistory) { return nil @@ -287,16 +379,6 @@ func (h *Handler) pendingAssistantLocked() *ChatMsg { return msg } -func (h *Handler) persistChatLocked() { - if h.history == nil { - return - } - // 写独立文件(原子替换)。失败只告警:聊天记录不该影响对话主流程。 - if err := h.history.Save(h.chatHistory); err != nil { - log.Printf("[webui] 写聊天记录 %s 失败: %v", h.history.Path(), err) - } -} - func (h *Handler) handleToolEvent(ev *sdk.Event) { payload := ev.Payload tool, _ := payload["tool"].(string) @@ -371,9 +453,6 @@ func (h *Handler) handleToolEvent(ev *sdk.Event) { } } -// chatSaveThrottle 控制写盘频率:最多每 3 秒写一次 -const chatSaveThrottle = 3 * time.Second - func (h *Handler) addChatMsg(msg ChatMsg) { h.chatMu.Lock() h.chatHistory = append(h.chatHistory, msg) diff --git a/internal/plugins/webui/handler_test.go b/internal/plugins/webui/handler_test.go index 4dd1483..13322ea 100644 --- a/internal/plugins/webui/handler_test.go +++ b/internal/plugins/webui/handler_test.go @@ -1246,3 +1246,49 @@ func TestHistoryStoreSaveIsAtomicAndRoundTrips(t *testing.T) { t.Fatalf("回读不一致:%+v", got) } } + +// TestChatPersistenceIsThrottled 钉住聊天记录写盘节流: +// 连续变更不得每条都整段重写文件,但 Close 前必须把最后一次落盘(否则丢对话)。 +func TestChatPersistenceIsThrottled(t *testing.T) { + dir := t.TempDir() + file := filepath.Join(dir, "chat.json") + + cfgReg := internalConfig.NewConfigRegistry("") + webuiCfg := cfgReg.PluginConfig("webui") + webuiCfg.RegisterDef(internalConfig.ConfigDef{Key: "history_file", Default: ""}) + webuiCfg.Set("history_file", file) + + s := testSDK(sdk.SDKConfig{ + Settings: sdk.NewSettings("webui", cfgReg), + Config: sdk.NewConfig(&types.Config{}), + EventBus: events.NewBus(), + }) + h := NewHandler(s) + if h.history.Path() != file { + t.Fatalf("history 路径应为 %s,实际 %s", file, h.history.Path()) + } + + // 模拟一轮对话里的连续变更(用户消息 + 多个工具事件 + 收尾) + base := time.Now() + for i := 0; i < 20; i++ { + h.chatMu.Lock() + h.chatHistory = append(h.chatHistory, ChatMsg{ + Role: "assistant", Content: "消息", Time: base.Add(time.Duration(i) * time.Second).Format(time.RFC3339), + }) + h.persistChatLocked() + h.chatMu.Unlock() + } + // 节流窗口内不应落盘 + if _, err := os.Stat(file); err == nil { + t.Fatal("节流窗口(3s)内不应已经写盘") + } + + // Close 必须把最后一次变更落下去 + h.Close() + msgs := newHistoryStore(file).Load() + if len(msgs) != 20 { + t.Fatalf("Close 后应有 20 条记录,实际 %d", len(msgs)) + } + // Close 幂等 + h.Close() +} diff --git a/internal/plugins/webui/plugin.go b/internal/plugins/webui/plugin.go index 9cd87f2..28441d8 100644 --- a/internal/plugins/webui/plugin.go +++ b/internal/plugins/webui/plugin.go @@ -294,6 +294,11 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error { } func (p *Plugin) Stop() error { + // 先停聊天记录写盘协程并落最后一次,再关服务器: + // 写盘是节流的(chatSaveThrottle),不显式关会丢掉最后一轮对话。 + if p.handler != nil { + p.handler.Close() + } if p.server != nil { return p.server.Close() }