Cyrene_v2/internal/runtime/memory/tidy_service.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 已 ≥ interval；② 已在安静窗内且窗口起点晚于上次 run
//     （= 这个窗口还没跑过）。此后心跳照旧只在安静窗内跑。
//
// 这样「重启不饿死」是可证的：重启只影响内存，判据只依赖持久事实与墙上时钟。

// 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；
	// 测试注入假等待（推进假时钟后立刻返回），因此补跑判据可以确定性验证。
	wait func(ctx context.Context, d time.Duration) bool

	// 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,
		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）。
func (s *TidyService) Enabled() bool { return s.cfg.Enabled }

// Start 启动心跳。它是幂等的：第二次调用是空操作。配置关闭时只记一条日志
// 并立即返回（关闭是显式选择，绝不静默）。ctx（可为 nil）是所有内部工作的
// 父级：取消它会像 Stop 一样停止服务。
func (s *TidyService) Start(ctx context.Context) {
	if !s.cfg.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)
	s.mu.Unlock()

	// 持久事实先读进来：判定「要不要补跑」只看它，不看进程内状态。
	s.loadPersistedLastRun(ctx)

	go s.loop(cctx)

	s.log.Info("memory tidy started",
		"interval", s.cfg.Interval.String(),
		"quiet_hours", fmt.Sprintf("%02d:00-%02d:00", s.cfg.QuietStartHour, s.cfg.QuietEndHour),
		"recall_threshold", s.cfg.RecallThreshold,
		"scan_limit", s.cfg.ScanLimit,
		"compare_limit", s.cfg.CompareLimit,
		"candidate_limit", s.cfg.CandidateLimit,
		"auto_apply", s.cfg.autoApplyEnabled(),
		"auto_apply_per_round", s.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.cfg.Enabled {
		return TidyStats{}, ErrTidyDisabled
	}
	return s.runRound(ctx, "manual")
}

// loop 是心跳 goroutine：先做**启动补跑判定**（持久事实 + 墙上时钟），再进入
// 「等到下一个低峰起点 → 在窗口内时跑一轮 → 重算」的循环。ctx 被取消时退出。
func (s *TidyService) loop(ctx context.Context) {
	defer s.wg.Done()
	s.catchUp(ctx)
	for {
		if !s.wait(ctx, s.nextDelay()) {
			return
		}
		if ctx.Err() != nil {
			return
		}
		if !s.cfg.inQuietHours(s.now()) {
			// 低峰窗口外不跑：整理不是随时可做的动作（design/23 §4/§9）。
			s.log.Debug("memory tidy skipped: outside quiet hours",
				"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 已 ≥ interval —— 周期到点，窗口不再是必要条件（补跑）；
//  3. 已在安静窗内 **且** 窗口起点晚于上次 run —— 这个窗口还没跑过。
//
// 三条都**不**看「进程活了多久」，因此重启多少次都不会饿死整理。反过来，窗口
// 内跑过一次之后再重启，条件 2/3 都不成立（lastRun 在窗口起点之后），因此也
// 不会每次重启都重跑一轮扫描。
func (s *TidyService) dueNow() (bool, string) {
	now := s.now()
	s.mu.Lock()
	last := s.lastRun
	s.mu.Unlock()

	if last.IsZero() {
		if s.cfg.inQuietHours(now) {
			return true, "no_history_in_quiet_window"
		}
		return false, ""
	}
	if now.Sub(last) >= s.cfg.Interval {
		return true, "interval_elapsed"
	}
	if s.cfg.inQuietHours(now) {
		if start := s.cfg.currentQuietStart(now); last.Before(start) {
			return true, "quiet_window_not_run"
		}
	}
	return false, ""
}

// nextDelay 返回距下一次整理开跑的等待时长。
//
// 口径：下一次 = 下一个低峰窗口起点（窗口关闭时 = 现在 + 周期），并且不早于
// 「上一次运行 + 周期」—— 让配置的周期成为真实下限，而不是让安静时段每天
// 一次把「每周一次」冲掉。等待恒 ≥ minTidyTickDelay，避免配置异常时空转。
func (s *TidyService) nextDelay() time.Duration {
	now := s.now()
	next := s.cfg.nextQuietStart(now)
	if s.cfg.quietWindowDisabled() {
		next = now.Add(s.cfg.Interval)
	}
	s.mu.Lock()
	last := s.lastRun
	s.mu.Unlock()
	if !last.IsZero() {
		if floor := last.Add(s.cfg.Interval); next.Before(floor) {
			next = floor
		}
	}
	delay := next.Sub(now)
	if delay < minTidyTickDelay {
		return minTidyTickDelay
	}
	return delay
}

// 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_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.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"