Cyrene_v2/internal/runtime/memory/tidy.go

package memory

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"sort"
	"strings"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding"
)

// 本文件是记忆整理（design/23-记忆整理与去重）的整理器与编排：逐分区扫描 →
// 候选召回 → 包含判定 → 规则否决 → 候选落库 + 运行留痕。
//
// 三条不可走样的口径（design/23 §0/§5.1）：
//  1. **相似度只用来召回候选对**，永不直接当合并判据；
//  2. **合并的唯一条件是「事实集合的包含关系」**（A ⊆ B 才并入 B）；
//     相似但并列的事实只建链接，绝不合并；
//  3. **规则否决一票否决**（跨作用域/跨角色/跨项目/群聊来源/时效污染/
//     数值不一致/否定冲突/低置信），命中即踢出候选。
//
// 第一版**只产建议**：RunOnce 唯一的写入是候选表与运行表，
// 绝不改写、绝不删除任何一条记忆（design/23 §3 非目标 + §7）。

// TidyAction 是整理建议的动作类别（design/23 §10 的 action 枚举）。
type TidyAction string

const (
	// TidyMerge 表示 A 的事实集合是 B 的子集，建议把 A 并入 B（保留 B）。
	// 它仍需人工确认（design/23 §7），第一版绝不自动应用。
	TidyMerge TidyAction = "merge"
	// TidyLink 表示两条相似但**互不包含**，只建链接/分组，绝不合并
	// （design/23 §1.3 的 [1]/[3]「红牛 vs 冰美式」）。
	TidyLink TidyAction = "link"
	// TidyArchive 表示含绝对日期/时刻的一次性事件，建议归档而非合并
	// （design/23 §5.1 T0-5）。
	TidyArchive TidyAction = "archive"
	// TidyReclassify 表示 kind 误分类（design/23 §5.1 T0-6），只产候选不自动改。
	TidyReclassify TidyAction = "reclassify"
	// TidyKeep 是「拿不准 = 什么都不做」（design/23 §5.2-10）。它是合法动作，
	// 但第一版不为它落库（只进否决计数与日志），以免候选队列被噪声填满。
	TidyKeep TidyAction = "keep"
)

// TidyVetoReason 是规则否决的原因标签：进日志、进运行表的原因分布
// （design/23 §9「每轮记否决原因分布」），也是测试断言的抓手。
type TidyVetoReason string

const (
	// VetoNone 表示未命中任何否决规则。
	VetoNone TidyVetoReason = ""
	// VetoScope 是跨作用域硬门（design/23 §5.2-7）：user/persona/project 之间永不合。
	VetoScope TidyVetoReason = "scope"
	// VetoPersona 是跨角色硬门（design/23 §5.2-8）。
	VetoPersona TidyVetoReason = "persona"
	// VetoProject 是跨项目硬门（design/23 §5.2-8）。
	VetoProject TidyVetoReason = "project"
	// VetoSource 是群聊/陌生人来源（design/23 §5.2-9、§6）：只读不写。
	VetoSource TidyVetoReason = "source"
	// VetoTemporal 是时效污染（design/23 §5.2-1）：一次性事件/有期限状态
	// 不等于长期事实（[8]「9月8日晚」≠ 熬夜习惯）。
	VetoTemporal TidyVetoReason = "temporal"
	// VetoNumeric 是数值/配置不一致（design/23 §5.2-2）：数值不同即不同事实。
	VetoNumeric TidyVetoReason = "numeric"
	// VetoConflict 是否定/冲突（design/23 §5.2-3）：X vs 非X 是覆盖不是重复。
	VetoConflict TidyVetoReason = "conflict"
	// VetoLowConfidence 是低置信否决门（design/23 §5.2-6）。
	VetoLowConfidence TidyVetoReason = "low_confidence"
)

// ErrNilStore 在整理器没有收到记忆存储时返回。
var ErrNilStore = errors.New("memory: tidy needs a store")

// ErrTidyEmbedCount 在 embedder 返回的向量条数与请求条数不一致时返回：
// 缺失向量会让相似度与条目错位，宁可整体回退到词项余弦。
var ErrTidyEmbedCount = errors.New("memory/tidy: embedder returned a mismatched vector count")

// TidyRun 是一轮整理的留痕（design/23 §10 memory_tidy_runs）。
type TidyRun struct {
	ID          string
	StartedAt   time.Time
	FinishedAt  time.Time
	Scanned     int
	Candidates  int
	Applied     int
	Skipped     int
	Errors      int
	Vetoed      int
	VetoReasons map[string]int
	RuleVersion string
}

// TidyStats 是一轮整理的观测口径：**扫描 N / 候选 M / 建议 K / 否决原因分布**
// （design/23 §9）。它同时写进日志与运行表，绝不静默。
type TidyStats struct {
	// RunID 是本轮在 memory_tidy_runs 里的留痕 id（切片 TIDY-1）：候选按它归属，
	// 「按 run 批量应用」因此有确定的抓手。留痕写失败时它仍是本轮已用的 id，
	// 调用方不该据此假装留痕存在。
	RunID       string
	Scanned     int
	Partitions  int
	Compared    int
	Pairs       int
	Candidates  int
	Duplicates  int
	Skipped     int
	Vetoed      int
	VetoReasons map[string]int
	// ByAction 是本轮**新建**候选按动作的分布（幂等跳过的 duplicates 不计，
	// 它们不是这一轮的产出）。它是「新增/合并/软删各多少」这条回执的领域来源。
	ByAction map[string]int
	// Errors 是本轮的失败计数（分区失败 + 候选写入失败），进运行表。
	Errors int
	// Truncated 是被上限截断的分区标签（扫描或比对上限），供运维放大配额。
	Truncated []string
	// DurationMS 是本轮的墙钟耗时（毫秒），与日志里的 duration_ms 同一算法。
	DurationMS int64
}

// TidyDeps 是注入给整理器的依赖。Store 必需；其余可空并优雅降级。
type TidyDeps struct {
	// Store 是记忆存储。必需 —— 整理只读写它，绝不接触会话历史。
	Store *Store
	// Embedder 是候选召回的相似度来源。为 nil 时回退到零依赖的
	// 字符二元组余弦（离线可用、测试确定性），见 tidy_recall.go。
	Embedder embedding.Embedder
	// Audit 是审计出口（design/23 §8 的 memory.tidy.* 动作）。可选：
	// 组合根把它接到网关的 AuditEvent；为 nil 时只记日志。
	Audit func(who, what, result string)
	// Now 是运行时间戳的墙钟来源，便于测试注入。
	Now func() time.Time
	// Log 是基础日志器，来源标记为 "memory/tidy"。
	Log *slog.Logger
}

// Tidier 是记忆整理器。它并发安全**仅限**「一轮之内不重入」：
// 服务层用一把互斥锁保证同一时刻只有一轮（design/23 §9 单写者）。
type Tidier struct {
	cfg  TidyConfig
	deps TidyDeps
	log  *slog.Logger
	now  func() time.Time
}

// NewTidier 由 cfg 与 deps 构建整理器。Store 为 nil 时返回 ErrNilStore。
func NewTidier(cfg TidyConfig, deps TidyDeps) (*Tidier, error) {
	if deps.Store == nil {
		return nil, ErrNilStore
	}
	cfg = normalizeTidyConfig(cfg)
	now := deps.Now
	if now == nil {
		now = time.Now
	}
	return &Tidier{
		cfg:  cfg,
		deps: deps,
		log:  logger.WithSource(baseTidyLog(deps.Log), "memory/tidy"),
		now:  now,
	}, nil
}

// Config 返回生效的整理配置（已套默认值）。
func (t *Tidier) Config() TidyConfig { return t.cfg }

// tidyCandidateSink 是**本轮的候选出口**：一个分区比完就把候选交给它。
// 生产轮是落库（persistCandidates），试算轮只收集（tidy_dryrun.go）——
// 规则引擎本身与出口无关，这正是「禁止写第二套实现」的落点。
type tidyCandidateSink func(ctx context.Context, cands []TidyCandidate, stats *TidyStats)

// tidyRecall 是**本轮的召回策略**：给定已截断的参与比对条目，产出候选对。
// 它把「相似度来源不可用时怎么办」的差异收在一处：生产轮降级到零依赖词项
// 余弦，试算轮如实上报错误。budget 是本轮的时间预算闸（见 tidy_budget.go），
// 由调用方一路传下来，使两条策略共享**同一个**预算（而不是各自从 ctx 再派生）。
type tidyRecall func(ctx context.Context, p tidyPartition, items []MemoryItem, stats *TidyStats, budget tidyBudget) ([]tidyPair, error)

// tidyRoundPolicy 是本轮整理的出口策略：候选去哪 + 召回来源不可用时怎么办。
// 「生产轮」与「试算轮」的**全部**差异都在这两个字段上，规则引擎其余部分
// 逐字共用。
type tidyRoundPolicy struct {
	sink   tidyCandidateSink
	recall tidyRecall
}

// RunOnce 跑一轮整理，并写一条运行留痕。它返回本轮的观测计数。
//
// 失败隔离（design/23 §9）：单个分区失败不影响其它分区，失败计数进运行表与
// 日志；只有「所有分区都失败」或运行留痕本身写不进去时才回错误。
func (t *Tidier) RunOnce(ctx context.Context) (TidyStats, error) {
	started := t.now().UTC()
	// 运行 id 与轮次开始**同时**定下来：候选要按它归属（切片 TIDY-1 的按轮
	// 批量应用），运行留痕也要用它，因此不能等到扫完再生成。
	runID := mustNewTidyID()
	stats, partitions, err := t.beginRound(ctx)
	if err != nil {
		return stats, err
	}
	stats.RunID = runID

	failures := t.scanPartitions(ctx, partitions, &stats, tidyRoundPolicy{
		sink: func(ctx context.Context, cands []TidyCandidate, stats *TidyStats) {
			t.persistCandidates(ctx, runID, cands, stats)
		},
		recall: t.fallbackRecall,
	})

	run := TidyRun{
		ID:          runID,
		StartedAt:   started,
		FinishedAt:  t.now().UTC(),
		Scanned:     stats.Scanned,
		Candidates:  stats.Candidates,
		Applied:     0, // 第一版绝不自动应用（design/23 §7）
		Skipped:     stats.Skipped + stats.Duplicates,
		Errors:      stats.Errors,
		Vetoed:      stats.Vetoed,
		VetoReasons: stats.VetoReasons,
		RuleVersion: TidyRuleVersion,
	}
	stats.DurationMS = run.FinishedAt.Sub(run.StartedAt).Milliseconds()
	if err := t.deps.Store.persistTidyRun(ctx, run); err != nil {
		failures = append(failures, err)
	}

	t.logRun(stats, run)

	switch {
	case len(failures) == 0:
		return stats, nil
	case len(failures) == len(partitions) && len(partitions) > 0:
		return stats, fmt.Errorf("memory/tidy: every partition failed: %w", errors.Join(failures...))
	default:
		return stats, fmt.Errorf("memory/tidy: run finished with errors: %w", errors.Join(failures...))
	}
}

// logRun 输出本轮的可观测口径：扫描 N / 候选 M / 建议 K / 否决原因分布。
// 它带 source（memory/tidy），因此这条汇总永远可回溯到整理子系统。
func (t *Tidier) logRun(stats TidyStats, run TidyRun) {
	fields := []any{
		"run_id", run.ID,
		"rule_version", run.RuleVersion,
		"partitions", stats.Partitions,
		"scanned", run.Scanned,
		"compared", stats.Compared,
		"pairs", stats.Pairs,
		"candidates", run.Candidates,
		"duplicates", stats.Duplicates,
		"skipped", stats.Skipped,
		"vetoed", run.Vetoed,
		"veto_reasons", tidyReasonsLabel(run.VetoReasons),
		"errors", run.Errors,
		"duration_ms", run.FinishedAt.Sub(run.StartedAt).Milliseconds(),
		"recall_mode", tidyRecallMode(t.deps.Embedder != nil),
	}
	if len(stats.Truncated) > 0 {
		fields = append(fields, "truncated", strings.Join(stats.Truncated, ","))
	}
	t.log.Info("memory tidy run finished", fields...)
}

// beginRound 开一轮：准备口径容器并列出待整理分区。分区列表拿不到时整轮失败，
// 且**不写运行留痕** —— 那一轮根本没开始（口径与 design §9 一致）。
func (t *Tidier) beginRound(ctx context.Context) (TidyStats, []tidyPartition, error) {
	stats := TidyStats{
		VetoReasons: make(map[string]int),
		ByAction:    make(map[string]int),
	}
	partitions, err := t.deps.Store.listTidyPartitions(ctx)
	if err != nil {
		return stats, nil, fmt.Errorf("memory/tidy: list partitions: %w", err)
	}
	stats.Partitions = len(partitions)
	return stats, partitions, nil
}

// scanPartitions 逐个分区跑规则并把候选交给 policy.sink。它返回失败列表
// （单个分区失败不阻塞其它分区，design §9）；成功时返回 nil。
//
// 时间预算（tidy_budget.go）在**轮**级派生一次：ctx 有墙钟上限（同步/异步试算）
// 时，预算用尽后的分区不再开跑，而是如实登记为 (budget) 截断。
func (t *Tidier) scanPartitions(ctx context.Context, partitions []tidyPartition, stats *TidyStats, policy tidyRoundPolicy) []error {
	budget := t.newTidyBudget(ctx)
	var failures []error
	for _, p := range partitions {
		if err := t.runPartition(ctx, p, stats, policy, budget); err != nil {
			stats.Errors++
			failures = append(failures, err)
			t.log.Warn("memory tidy partition failed", "partition", p.label(), "error", err)
		}
	}
	return failures
}

// fallbackRecall 是**生产轮**的召回：相似度来源不可用时降级到零依赖的词项
// 余弦（离线/无 embedding 配置时可跑，宁可比语义路径更保守），并记一条 WARN。
// 两个来源**不可混用阈值**：阈值只有一个（design §13 开放点 2 的 0.80）。
func (t *Tidier) fallbackRecall(ctx context.Context, p tidyPartition, items []MemoryItem, stats *TidyStats, budget tidyBudget) ([]tidyPair, error) {
	pairs, err := t.recallPairs(ctx, p, items, stats, budget)
	if err == nil {
		return pairs, nil
	}
	t.log.Warn("memory tidy embedding recall unavailable, falling back to lexical",
		"partition", p.label(), "error", err)
	return t.collectPairs(p, items, t.cfg.RecallThreshold,
		lexicalSimFunc(items), tidyRecallLexical, stats, budget), nil
}

// runPartition 整理一个分区（persona_id + scope + project_id）。
// 分区之间**永不比较**（design/23 §6 硬约束）。候选去处与召回降级策略都由
// policy 决定，因此生产轮与试算轮走的是**同一条**扫描→召回→判定→构造候选
// 的路径。
func (t *Tidier) runPartition(ctx context.Context, p tidyPartition, stats *TidyStats, policy tidyRoundPolicy, budget tidyBudget) error {
	// 预算用尽 ⇒ 这个分区本轮根本不看。**如实登记**（响应 truncated + WARN）：
	// 「没扫到」与「扫了没有候选」必须能区分，否则面板会把预算截断读成
	// 「这个分区没有建议」。
	if budget.exhausted() {
		stats.Truncated = append(stats.Truncated, budgetTruncationLabel(p))
		t.log.Warn("memory tidy partition skipped by time budget", "partition", p.label())
		return nil
	}

	items, err := t.deps.Store.listTidyPartitionItems(ctx, p, t.cfg.ScanLimit)
	if err != nil {
		return fmt.Errorf("memory/tidy: list partition %s: %w", p.label(), err)
	}
	stats.Scanned += len(items)
	if len(items) == t.cfg.ScanLimit {
		// 扫描上限命中：本分区更大的部分这一轮不看（design §9「超限分片，下一轮
		// 继续」）。本实现只做「最新优先」的截断、不推进分片游标，因此这里必须
		// 留下痕迹，否则「配置了 5000 却永远只看最新 5000」会无人知晓。
		// 日志带**真总数**（LIMIT 之后的 len(items) 只是下限）：另查一次 COUNT，
		// 否则 WARN 会把「至少 5000」说成「正好 5000」。
		total, totalErr := t.deps.Store.countTidyPartitionItems(ctx, p)
		stats.Truncated = append(stats.Truncated, p.label()+"(scan)")
		if totalErr != nil {
			t.log.Warn("memory tidy partition hit the scan limit (total unknown)",
				"partition", p.label(), "limit", t.cfg.ScanLimit, "error", totalErr)
		} else {
			t.log.Warn("memory tidy partition hit the scan limit",
				"partition", p.label(), "limit", t.cfg.ScanLimit, "total", total)
		}
	}

	comparable, archives := t.classifyItems(items, stats)

	// 截断与计数只做一次：召回策略拿到的是**已截断**的条目，回退重算不重复计数。
	compared := t.limitCompareItems(p, comparable, stats)
	pairs, err := policy.recall(ctx, p, compared, stats, budget)
	if err != nil {
		return fmt.Errorf("memory/tidy: recall partition %s: %w", p.label(), err)
	}

	candidates := archives
	for _, pr := range pairs {
		verdict := judgePair(pr.a, pr.b, pr.score)
		if verdict.Veto != VetoNone {
			stats.Vetoed++
			stats.VetoReasons[string(verdict.Veto)]++
			continue
		}
		candidates = append(candidates, buildCandidates(pr, verdict, t.now().UTC())...)
	}
	// 每分区候选上限（design §9）对**全部候选**生效（含归档建议），
	// 超出的只登记不入队，并记进 stats.Truncated（绝不静默）。
	if len(candidates) > t.cfg.CandidateLimit {
		// WARN 带总数/上限/分区名：候选被上限截断意味着面板看到的**少于**规则
		// 判出的，必须能从这里读出被丢掉了多少。
		t.log.Warn("memory tidy candidates truncated by per-partition limit",
			"partition", p.label(), "candidates", len(candidates), "limit", t.cfg.CandidateLimit)
		stats.Skipped += len(candidates) - t.cfg.CandidateLimit
		stats.Truncated = append(stats.Truncated, p.label()+"(candidates)")
		candidates = candidates[:t.cfg.CandidateLimit]
	}
	policy.sink(ctx, candidates, stats)
	return nil
}

// classifyItems 把分区内条目分流成**可比对**与**归档候选**（跳过的不产出）：
//   - 群聊/陌生人来源 → 排除在扫描范围外（design/23 §6，只读不写）；
//   - 含绝对日期/时刻 → 一次性事件，走归档而非合并（[8]「9月8日晚22:38」），
//     且**不参与任何 merge**（design §5.1-5）；
//   - 过短内容 → 没有可判定的事实要素，跳过（不产噪声候选）。
func (t *Tidier) classifyItems(items []MemoryItem, stats *TidyStats) (comparable []MemoryItem, archives []TidyCandidate) {
	comparable = make([]MemoryItem, 0, len(items))
	for _, it := range items {
		switch t.classifyForTidy(it, stats) {
		case tidySkip:
			continue
		case tidyArchive:
			archives = append(archives, archiveCandidate(it, t.now().UTC()))
			continue
		default:
			comparable = append(comparable, it)
		}
	}
	return comparable, archives
}

// tidyDisposition 是扫描时一条记忆的处置分流。
type tidyDisposition int

const (
	tidyCompare tidyDisposition = iota
	tidyArchive
	tidySkip
)

// classifyForTidy 决定一条记忆进入比对、进归档候选，还是被排除：
//   - 群聊/陌生人来源 → 排除在扫描范围外（design/23 §6，只读不写）；
//   - 含绝对日期/时刻 → 一次性事件，走归档而非合并（[8]「9月8日晚22:38」）；
//   - 过短内容 → 没有可判定的事实要素，跳过（不产噪声候选）。
func (t *Tidier) classifyForTidy(it MemoryItem, stats *TidyStats) tidyDisposition {
	if nonOwnerSource(it.SourceConversation) {
		stats.Vetoed++
		stats.VetoReasons[string(VetoSource)]++
		return tidySkip
	}
	if hasAbsoluteTimestamp(it.Content) {
		return tidyArchive
	}
	if len([]rune(it.Content)) < minTidyContentRunes {
		return tidySkip
	}
	return tidyCompare
}

// persistCandidates 落库一批候选：幂等键已存在时只计入 duplicates
// （design/23 §9「重复跑不重复产候选」），新建的计入 Candidates 并按动作进
// ByAction（回执的领域来源）。runID 是本轮的留痕 id：「按 run 批量应用」据此
// 找到这一轮产出的候选；幂等跳过的行保留它**首次**被产出的轮次，不被改写。
func (t *Tidier) persistCandidates(ctx context.Context, runID string, cands []TidyCandidate, stats *TidyStats) {
	for _, c := range cands {
		c.RunID = runID
		created, err := t.deps.Store.insertTidyCandidate(ctx, c)
		if err != nil {
			stats.Errors++
			t.log.Warn("memory tidy candidate insert failed", "action", string(c.Action), "error", err)
			continue
		}
		if !created {
			stats.Duplicates++
			continue
		}
		stats.Candidates++
		if stats.ByAction != nil {
			stats.ByAction[string(c.Action)]++
		}
	}
}

// tidyReasonsLabel 把否决原因分布渲染成稳定的日志字符串（按原因排序）。
func tidyReasonsLabel(reasons map[string]int) string {
	if len(reasons) == 0 {
		return "none"
	}
	keys := make([]string, 0, len(reasons))
	for k := range reasons {
		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, reasons[k]))
	}
	return strings.Join(parts, ",")
}

// baseTidyLog 在 log 非 nil 时返回它，否则返回丢弃式日志器。
func baseTidyLog(log *slog.Logger) *slog.Logger {
	return discardLogger(log)
}