package memory import ( "context" "fmt" "log/slog" "sort" "strings" "sync" "time" ) // 本文件是记忆整理的**后台任务形态**(design/23 §4 方案 A/C):定期跑(安静时段 // 降频)、独立运行、零上下文启动、只产建议、绝不推送。 // // 与主人的硬约束对齐(design/23 §4.1): // - 零上下文:本服务的依赖里**没有任何会话/历史读数**,输入只有记忆条目 // 与本次候选数据,因此「不读会话」是结构性成立的,不靠自觉; // - 独立运行:跑在专属隔离会话 TidySessionID(不是 history.MainSessionID), // 因此不改主人的心情、不进主会话泳道(cmd/cyrene 的 isConversationLane // 对它恒为 false); // - 它不是任何角色:系统任务,不带人格提示词,也不把写入限到某个角色的域; // - 不写会话:唯一的写入是候选表与运行表;agent 侧对 chat_messages 零写入。 // // 心跳范式参照 internal/services/backthink(同一低峰/安静时段语义),但在 // 「启动即跑」这一点上与本包旧版**有意不同**(切片 TIDY-1 修掉的正是旧口径): // // - 旧口径:进程启动时只算「下一个安静时段起点」,且「上次 run」只存在进程 // 内存里。后端两天重启十几次 ⇒ 24h 周期永远走不完、安静窗起点永远被下一次 // 重启覆盖,于是 memory_tidy_runs 只有 1 行、77 条候选全 pending。 // - 新口径(到点即跑 + 持久事实):启动时先读**持久事实** // (memory_tidy_runs 里最近一轮的开始时刻),满足任一条件就**立刻补跑一次**: // ① 距上次 run 已 ≥ 当时间隔;② 已在安静窗内且窗口起点晚于上次 run // (= 这个窗口还没跑过)。此后心跳按「上一次 run + **当时时段**的周期」醒。 // // 2026-10-07(主人「频率必须调……或者动态分配,白天间隔短一点,晚上到了静默时期 // 稍微长一点」)追加**分档周期**:非安静时段用 cfg.Interval,安静时段用 // cfg.QuietInterval(唯一判据 = TidyConfig.intervalAt)。安静时段因此从「唯一可跑 // 时段」变成「降频时段」,心跳不再有「窗口外一律跳过」这一档;而「到点即跑 + // catch-up」那三条判据一条未动。 // // 这样「重启不饿死」是可证的:重启只影响内存,判据只依赖持久事实与墙上时钟。 // minTidyTickDelay 是心跳循环的最小等待,防止配置异常(周期/时钟跳变)导致 // 空转忙轮询。 const minTidyTickDelay = time.Minute // TidyService 是记忆整理服务。用 NewTidyService 构建,用 Start 启动心跳、 // Stop 停止它;RunOnce / ManualRun 是「立即整理」入口(design/23 §4 的手动可触发)。 type TidyService struct { tidier *Tidier cfg TidyConfig log *slog.Logger mu sync.Mutex running bool cancel context.CancelFunc lastRun time.Time wg sync.WaitGroup // now 是调度用的时钟(默认取整理器的注入时钟)。测试注入它以模拟 // 「启动 → 跨过安静窗起点」而不用真等一整夜。 now func() time.Time // wait 是心跳的等待原语:返回 false 表示 ctx 已取消。生产是 time.Timer; // 测试注入假等待(推进假时钟后立刻返回),因此补跑判据可以确定性验证。 // // 🔴 它的返回值语义**没有变**(`false` 仍表示「这一觉没睡成就结束了」): // 面板调优的重排**不是**靠它自报,而是靠下面那个可被取消的子 ctx —— 于是注入的 // 假等待(不认识 wake 的那些)在测试里语义逐字不变(见 loop 的等待段)。 wait func(ctx context.Context, d time.Duration) bool // wake 是「面板调优改了配置 ⇒ 把这一觉重排一次」的一次性标记:**有缓冲(cap 1) // + 非阻塞发**,见 tidy_service_tuning.go 的 notifyReschedule。写端跑在面板的 // HTTP 保存路径上,因此**绝不允许阻塞**;满了丢弃是**正确**语义(通道里那一个 // 标记表达的就是「有一次待处理的重排」,重算是幂等的)。 wake chan struct{} // wakeMu 保护 wakeCancel(= 取消「当前这一次等待」的手段;两次等待之间是 nil)。 // 它是 wake 的**中断**那一半:只有标记没有中断的话,新周期要等这一觉(可能是一天) // 睡完才生效 —— 那正是主人把 24h 改成 3h 时看到的「改了跟没改一样」。 wakeMu sync.Mutex wakeCancel context.CancelFunc // runMu 保证同一时刻只有一轮整理(design/23 §9 单写者):心跳与手动触发 // 若撞在一起,后者排队而不是并发改写候选表。 runMu sync.Mutex // runs 是**异步试算**结果的进程内注册表(契约 v0.76 §4.12.1;见 // tidy_dryrun_async.go):不落库、不新开端口、重启即空。 runs *tidyRunRegistry } // NewTidyService 由 cfg 与 deps 构建整理服务。Store 为 nil 时返回 ErrNilStore。 func NewTidyService(cfg TidyConfig, deps TidyDeps) (*TidyService, error) { tidier, err := NewTidier(cfg, deps) if err != nil { return nil, err } return &TidyService{ tidier: tidier, cfg: tidier.Config(), log: tidier.log, now: tidier.now, wait: waitTidyTimer, wake: make(chan struct{}, 1), runs: newTidyRunRegistry(defaultTidyRunTTL), }, nil } // waitTidyTimer 是生产等待:真定时器 + ctx 取消感知。 func waitTidyTimer(ctx context.Context, d time.Duration) bool { timer := time.NewTimer(d) defer timer.Stop() select { case <-ctx.Done(): return false case <-timer.C: return true } } // SessionID 返回整理运行的专属隔离会话标识(design/23 §4.1 B)。 func (s *TidyService) SessionID() string { return TidySessionID } // Enabled 报告整理是否被配置开启(design/23 §4 的 MEMORY_TIDY_ENABLED)。 // // 它走 `enabled()` 而不是直接读字段:`Enabled` 是 `s.cfg` 里的一个字段,而面板调优 // 能在**运行期**整体覆写 `s.cfg`(见 tidy_service_tuning.go 的 ApplyPanelTuning), // 因此该字段的**每一个**读点都必须与写方同步(`s.mu`)。 func (s *TidyService) Enabled() bool { return s.enabled() } // enabled 在读锁下报告整理开关。它是 `s.cfg.Enabled` 的**唯一**安全读法 // (口径见 scheduleSnapshot 与上面 Enabled 的注释)。 func (s *TidyService) enabled() bool { s.mu.Lock() defer s.mu.Unlock() return s.cfg.Enabled } // cfgSnapshot 在读锁下取一份配置快照:凡是不需要 lastRun 的读点用它。 func (s *TidyService) cfgSnapshot() TidyConfig { s.mu.Lock() defer s.mu.Unlock() return s.cfg } // Start 启动心跳。它是幂等的:第二次调用是空操作。配置关闭时只记一条日志 // 并立即返回(关闭是显式选择,绝不静默)。ctx(可为 nil)是所有内部工作的 // 父级:取消它会像 Stop 一样停止服务。 // // 它**同时**起临时期限的到期清扫(design/23 §19 的第 ② / ③ 步,见 // expire_loop.go):清扫与心跳共用同一个 cctx 与 wg,因此它跟整理同生共死 —— // 整理被配置关掉时清扫也不跑(与 `preinsert_gc.go` 的清扫随审查服务启停同口径), // `Stop` 会一并取消并 join 它。 func (s *TidyService) Start(ctx context.Context) { if !s.enabled() { s.log.Info("memory tidy disabled by configuration", "env", "MEMORY_TIDY_ENABLED") return } if ctx == nil { ctx = context.Background() } s.mu.Lock() if s.running { s.mu.Unlock() return } s.running = true cctx, cancel := context.WithCancel(ctx) s.cancel = cancel s.wg.Add(1) // 到期清扫(design/23 §19 的第 ② / ③ 步)与心跳**同生共死**:同一个 cctx、 // 同一个 wg,因此 Stop 会一并取消并 join 它(§19.3-3 的「可停」)。 // // 🔴 为什么这笔 Add 也在锁内(与上面那一笔同处):让 `Stop` 的 `running` 判定 // 与这两笔 Add **原子化** —— 于是不可能出现「Stop 已看到 `running=false` 早退」 // 或「Wait 漏掉一笔尚未 Add 的 goroutine(那会当场泄漏一个清扫 goroutine)」 // 这两种窗口。 // // ⚠️ 这把锁**不**覆盖 `Stop` 的 `wg.Wait`:在同一个 `func (s *TidyService) Stop` // 里,`s.mu.Unlock()` 在前、`s.wg.Wait()`(另一个 goroutine 里)在后,两者之间 // 没有锁。因此 WaitGroup 的复用纪律(「上一轮 Wait 返回前不得再 Add」)**不是** // 由这把锁给的,而是靠「`Start` / `Stop` 在整个进程生命周期各只调用一次」维持: // `Start` 只在组合根启动期调一次(`cmd/cyrene/main.go` 的 // `a.memoryTidy.Start(ctx)`)、`Stop` 只在关停链里调一次(`cmd/cyrene/wire_services.go` // 的 `_ = a.memoryTidy.Stop(ctx)`)。 // 交叉调用(`Start` 与在飞的 `Stop` 并发)在 `-race` 下会 `panic: sync: // WaitGroup is reused before previous Wait has returned` —— 实测该 panic 在 // M3 之前的基线上同样可复现(既有性质,非本切片引入),只是当前调用点**不可达**。 s.wg.Add(1) s.mu.Unlock() // 持久事实先读进来:判定「要不要补跑」只看它,不看进程内状态。 s.loadPersistedLastRun(ctx) // 两个后台件在这里一起起:整理心跳,以及临时期限的到期清扫 // (expire_loop.go)。清扫跑在**自己的** goroutine 与**自己的**小时级 // ticker 上 —— 整理的周期是 24h 档(intervalSec),不能拿来当清扫周期 // (§19.1「到期后 ≤ 一个清扫周期,默认每小时」)。 go s.loop(cctx) go s.expirySweepLoop(cctx) // 启动日志逐项取一次快照(面板调优能在运行期覆写 cfg,因此这些字段也不能裸读)。 cfg := s.cfgSnapshot() s.log.Info("memory tidy started", "interval", cfg.Interval.String(), "quiet_interval", cfg.QuietInterval.String(), "quiet_hours", fmt.Sprintf("%02d:00-%02d:00", cfg.QuietStartHour, cfg.QuietEndHour), "recall_threshold", cfg.RecallThreshold, "scan_limit", cfg.ScanLimit, "compare_limit", cfg.CompareLimit, "candidate_limit", cfg.CandidateLimit, "auto_apply", cfg.autoApplyEnabled(), "auto_apply_per_round", cfg.AutoApplyPerRound, "rule_version", TidyRuleVersion, "session_id", TidySessionID, "last_run_at", s.lastRunLabel()) } // loadPersistedLastRun 把「上次 run 的时刻」从 memory_tidy_runs 读进内存 // (切片 TIDY-1 的持久事实)。读不到时只记 WARN 并保持零值:那是「没有历史」 // 这一正常状态,调度会退到「安静窗内且未跑过 ⇒ 补跑」那一档。 // // 零值会**保守地**让补跑更容易发生(窗口外无历史时不跑、窗口内直接跑一次), // 这正是「宁可多跑一轮,不可永远不跑」的取向 —— 与旧口径相反,因为旧口径的 // 「省一次扫描」代价是两天一条没应用。 func (s *TidyService) loadPersistedLastRun(ctx context.Context) { last, err := s.tidier.deps.Store.LastTidyRunStartedAt(ctx) if err != nil { s.log.Warn("memory tidy last run unreadable, catch-up will rely on the quiet window", "error", err) return } if last.IsZero() { return } s.mu.Lock() s.lastRun = last s.mu.Unlock() } // lastRunLabel 渲染上次 run 的日志标签(零值 = 从未跑过)。 func (s *TidyService) lastRunLabel() string { s.mu.Lock() last := s.lastRun s.mu.Unlock() if last.IsZero() { return "never" } return last.UTC().Format(time.RFC3339) } // Stop 停止心跳并等待进行中的一轮跑完。它是幂等的;若 ctx 在排空完成前被 // 取消则返回包装后的 context 错误,而不是泄漏 goroutine。 func (s *TidyService) Stop(ctx context.Context) error { if ctx == nil { ctx = context.Background() } s.mu.Lock() if !s.running { s.mu.Unlock() return nil } s.running = false cancel := s.cancel s.mu.Unlock() cancel() done := make(chan struct{}) go func() { s.wg.Wait() close(done) }() select { case <-done: s.log.Info("memory tidy stopped") return nil case <-ctx.Done(): return fmt.Errorf("memory/tidy: stop: %w", ctx.Err()) } } // RunOnce 立即跑一轮整理(「立即整理」入口,design/23 §4 的手动可触发)。 // 它不改变心跳节奏,也不依赖上一次的结果 —— 幂等由 rule_version + 数据指纹 // 保证(design/23 §4.1 A)。 func (s *TidyService) RunOnce(ctx context.Context) (TidyStats, error) { return s.runRound(ctx, "manual") } // ManualRun 是**手动真跑**入口(切片 TIDY-1;契约的 // POST /admin/memories/tidy/run)。它与心跳跑的是同一条规则路径、同一份配置, // 差别只有两处: // // 1. **尊重配置开关**:关闭时如实返回 ErrTidyDisabled,不退化成空结果; // 2. **触发标签是 manual**:审计里能区分「人点的」与「心跳/补跑的」。 // // 它写候选与运行留痕(与试算 DryRun 的零副作用相反),并会跑**轮末自动档**: // 规则层判为高置信的候选(字面重复 / 事实等价 ∧ ≥0.95 / 一次性事件且不像习惯) // 由 tidy_auto.go 直接落地,其余仍等人点(候选 apply / 按 run 批量 apply)。 // 关掉自动档只需 `$DATA_DIR/config/memory-tidy.json` 的 `{"autoApply": false}`。 func (s *TidyService) ManualRun(ctx context.Context) (TidyStats, error) { if !s.enabled() { return TidyStats{}, ErrTidyDisabled } return s.runRound(ctx, "manual") } // loop 是心跳 goroutine:先做**启动补跑判定**(持久事实 + 墙上时钟),再进入 // 「等到下一次到点 → 跑一轮 → 按**当时时段**重算下一次」的循环。ctx 被取消时退出。 // // 🔴 2026-10-07(动态频率)**去掉了「安静窗外的唤醒一律跳过」这一档**:安静时段从 // 「整理唯一可跑的时段」变成「整理降频的时段」(design/12 §N A1 给后台思考心跳定下 // 的同一口径),因为主人的频率口径要求白天也跑、且白天跑得比夜里勤(「白天间隔短 // 一点,晚上到了静默时期稍微长一点」)。窗口与补跑**都没有消失**:窗口还决定档位 // (intervalAt)与两条补跑判据(dueNow 的 1/3),「到点即跑 + catch-up」 // (design/23 §4.0)原样保留 —— 每次唤醒仍然要过 dueNow 才跑,不是「醒了就扫」。 func (s *TidyService) loop(ctx context.Context) { defer s.wg.Done() s.catchUp(ctx) for { // 上一轮攒下的「调优唤醒」标记在这里清掉:它全部的含义就是「把下一拍重算 // 一遍」,因此必须在重算**之前**清空 —— 否则重算完发现标记还在,会自旋。 s.drainWake() switch s.waitForTick(ctx) { case tidyTickStopped: // 父 ctx 被取消(Stop / 进程退出),或注入的假等待自报「停」:既有语义。 return case tidyTickWokenByTuning: // 被**面板调优**唤醒:这一觉是按**旧**周期算出来的,现在作废 ⇒ 只重算 // 下一拍,**绝不**因此跑一轮(唤醒不是「到点了」,判据仍只有 dueNow)。 s.log.Debug("memory tidy heartbeat rescheduled by panel tuning", "now_hour", s.now().Hour()) continue } // 到点即跑:间隔用的是**此刻**的时段(nextDelay 里现算的),因此跨过 // 安静时段边界后,下一轮自动换档。判据再次复核是因为等待期间可能有别的 // 一轮跑过(手动触发):那时 lastRun 已经前移,这一拍就不该再跑。 if due, reason := s.dueNow(); !due { s.log.Debug("memory tidy heartbeat skipped: not due", "reason", reason, "now_hour", s.now().Hour()) continue } if _, err := s.runRound(ctx, "heartbeat"); err != nil { s.log.Warn("memory tidy heartbeat round failed", "error", err) } } } // catchUp 做启动补跑(切片 TIDY-1 的调度修复):到点即跑,绝不让「进程重启」 // 把 24h 周期或安静窗永远推掉。 func (s *TidyService) catchUp(ctx context.Context) { due, reason := s.dueNow() if !due { return } s.log.Info("memory tidy catch-up run starting", "reason", reason, "session_id", TidySessionID, "last_run_at", s.lastRunLabel()) if _, err := s.runRound(ctx, "catchup"); err != nil { s.log.Warn("memory tidy catch-up round failed", "reason", reason, "error", err) } } // dueNow 是**到点即跑**的判据,只依赖两样东西:持久事实(s.lastRun,由 // memory_tidy_runs 载入)与墙上时钟。满足任一条即应立刻跑: // // 1. 从未跑过 **且** 已在安静窗内 —— 「这个窗口还没跑过」的最简情形(全新库); // 2. 距上次 run 已 ≥ **当时时段**的周期 —— 周期到点,窗口不再是必要条件(补跑); // 3. 已在安静窗内 **且** 窗口起点晚于上次 run —— 这个窗口还没跑过。 // // 三条都**不**看「进程活了多久」,因此重启多少次都不会饿死整理。反过来,窗口 // 内跑过一次之后再重启,条件 2/3 都不成立(lastRun 在窗口起点之后),因此也 // 不会每次重启都重跑一轮扫描。 // // 判据 2 用 intervalAt(now) 而不是某一个固定周期:周期现在是**分档**的(安静时段 // quietIntervalSec / 其余 intervalSec),用哪一档取决于**现在几点**——这与 // nextDelay 用同一个函数,因此「下一拍的时间」与「到没到点」不可能各说各话。 func (s *TidyService) dueNow() (bool, string) { now := s.now() cfg, last := s.scheduleSnapshot() if last.IsZero() { if cfg.inQuietHours(now) { return true, "no_history_in_quiet_window" } return false, "" } if now.Sub(last) >= cfg.intervalAt(now) { return true, "interval_elapsed" } if cfg.inQuietHours(now) { if start := cfg.currentQuietStart(now); last.Before(start) { return true, "quiet_window_not_run" } } return false, "" } // nextDelay 返回距下一次整理开跑的等待时长。 // // 口径(2026-10-07 动态频率修订): // // - **有历史**:下一次 = 「上一次 run + **当时时段**的周期」(intervalAt); // - **无历史**:锚到下一个安静时段起点(design/23 §4:首次整理落在低峰),窗口 // 被关闭时退化为「现在 + 当时间隔」——与本次改动之前逐字一致; // - 等待恒 ≥ minTidyTickDelay,避免配置异常(周期 ≤ 1 分钟)时空转。 // // 🔴 为什么在这里现算,而不是启动时算一次: // 周期是**分档**的(非安静时段 = intervalSec,安静时段 = quietIntervalSec),而 // 每一轮跑完之后「现在落在哪一档」可能已经变了(例:06:50 跑完那轮落在安静档 ⇒ // 下一拍被推到 10:50;而 10:50 之后应当换回白天的档)。把 intervalAt(now) 放在 // **每一次「等」之前**求值,跨过时段边界后的第一拍就用新档,不必等重启。 // 把它挪回启动时算一次 ⇒ 白天/晚上的切换不会生效(本切片有测试钉住这条)。 func (s *TidyService) nextDelay() time.Duration { now := s.now() cfg, last := s.scheduleSnapshot() // lastRun 与 cfg 必须在**同一把锁**下取到同一版本(见 scheduleSnapshot), // 否则一次面板保存可能让这里拿到「新周期 + 旧历史」的混合体。 var next time.Time if last.IsZero() { next = cfg.nextQuietStart(now) if cfg.quietWindowDisabled() { next = now.Add(cfg.intervalAt(now)) } } else { next = last.Add(cfg.intervalAt(now)) } delay := next.Sub(now) if delay < minTidyTickDelay { return minTidyTickDelay } return delay } // scheduleSnapshot 在锁下同时取「调度配置」与「上次运行时刻」。 // // 🔴 为什么要有它(切片 MEMPANEL / M4 立):`cfg` 在这条特性之前是启动期冻结的, // 因此调度读(`nextDelay` / `dueNow`)直接读字段本来安全;面板调优开了「运行期改 // cfg」这条路之后,调度读**必须**与写方(tidy_service_tuning.go 的 ApplyPanelTuning) // 同步,否则就是数据竞争(`-race` 可复现)。两样东西一起取,是为了让一次判定看到 // 的是**同一个版本**(分开取可能落在两次不同的写入之间)。 func (s *TidyService) scheduleSnapshot() (TidyConfig, time.Time) { s.mu.Lock() defer s.mu.Unlock() return s.cfg, s.lastRun } // runRound 跑一轮并把结果与审计留痕。单写者:runMu 串行化心跳与手动触发。 func (s *TidyService) runRound(ctx context.Context, trigger string) (TidyStats, error) { s.runMu.Lock() defer s.runMu.Unlock() s.log.Info("memory tidy round starting", "trigger", trigger, "session_id", TidySessionID) stats, err := s.tidier.RunOnce(ctx) s.mu.Lock() s.lastRun = s.now() s.mu.Unlock() s.audit(stats, trigger) if err != nil { s.log.Warn("memory tidy round failed", "trigger", trigger, "error", err) } return stats, err } // audit 写一条审计(design/23 §8 的 memory.tidy.* 口径)。审计出口由组合根 // 接到网关的 AuditEvent;未接线时只记日志(绝不静默)。审计只带计数, // **不带任何记忆正文**。 func (s *TidyService) audit(stats TidyStats, trigger string) { hook := s.tidier.deps.Audit result := fmt.Sprintf( "trigger=%s run_id=%s scanned=%d candidates=%d duplicates=%d vetoed=%d kept=%d duration_ms=%d by_action=%s rejected_reasons=%s auto_applied=%d auto_merge=%d auto_archive=%d auto_reclassify=%d auto_eligible=%d auto_truncated=%t", trigger, stats.RunID, stats.Scanned, stats.Candidates, stats.Duplicates, stats.Vetoed, stats.Kept, stats.DurationMS, tidyActionLabel(stats.ByAction), tidyReasonsLabel(stats.VetoReasons), stats.AutoApplied, stats.AutoMerged, stats.AutoArchived, stats.AutoReclassified, stats.AutoEligible, stats.AutoTruncated) if hook == nil { s.log.Info("memory tidy audit (no sink configured)", "what", "memory.tidy.run", "result", result) return } hook(tidyAuditWho, "memory.tidy.run", result) } // tidyActionLabel 把「新建候选按动作的分布」渲染成稳定字符串(按动作名字典序)。 // 与 tidyReasonsLabel 同一手法,供审计与日志使用。 func tidyActionLabel(byAction map[string]int) string { if len(byAction) == 0 { return "none" } keys := make([]string, 0, len(byAction)) for k := range byAction { keys = append(keys, k) } sort.Strings(keys) parts := make([]string, 0, len(keys)) for _, k := range keys { parts = append(parts, fmt.Sprintf("%s=%d", k, byAction[k])) } return strings.Join(parts, ",") } // tidyAuditWho 是整理审计的调用方标签:它是**系统任务**,不是任何角色 // (design/23 §4.1 C)。 const tidyAuditWho = "system/memorytidy"