AskaEth
4df8d6f53a
review P1：applyTidyConfig 在运行期**整体覆写** s.cfg，而我只把调度读收进了持锁 访问器 —— 同一字段还有几处无锁读，其中两处是**活端点**： - TidyService.DryRun 的开关判定（在取 runMu **之前**）⇒ POST …/tidy/dry-run 与 ?async=1 那条会起后台 goroutine 的版本（真机 90s 级）； - TidyService.ManualRun 的开关判定（同样在 runMu 之前）⇒ POST …/tidy/run； - （原本也漏的）Enabled() 与 Start 的启动日志块。 按**字段**重扫了一遍（不按文件）：现在生产代码里对 s.cfg 的读**只剩**三个持锁访问器 内部（enabled / cfgSnapshot / scheduleSnapshot），写点只有 applyTidyConfig 一处； s.tidier.cfg 的读点全在持 runMu 的轮次 / 试算路径上（含异步版经 DryRun）。 新增第三条并发回归哨兵：真并发跑「试算 + 手动真跑」两个活端点 × 面板保存。 修前红/修后绿：撤回 DryRun 的同步 ⇒ 报「写 applyTidyConfig × 读 DryRun」；撤回 ManualRun 的同步 ⇒ 报「写 applyTidyConfig × 读 ManualRun」；两次变异都逐字还原 （diff 0 差异）。 顺手两条（塔指定）： - 那条并发哨兵的 doc 与实现相反（还写着「正确顺序是 Stop → 写 → Start」），已改成 唤醒机制下的说法 —— 与「注释与实现不符」同类，不留旧说法。 - 非法值的回显文案原按**字节**截断（s[:40]），中文会被切成半截（非法 UTF-8，经 JSON 编码变成 U+FFFD 替换符）。改成按 **rune** 截断，直接复用同包既有的 clipRunes；用例取值刻意让截断点落在多字节字符中间（JSON 引号 + 一个 ASCII 前缀 错开对齐），并做过变异复验：改回按字节 ⇒ 用例红（切出半个汉字）。
503 lines
22 KiB
Go
503 lines
22 KiB
Go
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"
|