Cyrene_v2/internal/runtime/memory/expire.go

package memory

import (
	"context"
	"fmt"
	"strings"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory/vector"
)

// 本文件是**临时期限到期处置的第 ② 步与第 ③ 步**：后台到期清扫（物理 DELETE）
// 与「删前留审计」（design/23 §19，主人 Q1 裁定「临时记忆到期删除」）。
//
// 三步是刻意分开的（§19.1）：
//
//	① 隐藏 —— 读侧时效闸（timegate.go）：每次读比一次列值，**零写入**、即时；
//	② 删除 —— 本文件：显式、有界、有审计、可停的后台路径；
//	③ 审计 —— 本文件：**每行删除前**先写一条 `memory.tidy.expire`。
//
// 把删除混进读路径是这条链上最容易犯的错：读路径会因此变成「有副作用且不可逆」，
// 一次带 bug 的查询就能删记忆。因此删除只有这里一个出口，只由后台件调用。
//
// 🔴 **与「写路径唯一」红线怎么共存**：`design/23 §15.1 D3-1` 那条约束的是
// **INSERT（入库）**，`Store.Add` 是唯一 INSERT 路径。到期删除是一条 **DELETE
// （清除）**，不是第二条入库路径；既有先例是 `Store.Delete` / `DeleteScoped` 与
// `preinsert_gc.go` 的硬删留痕，两者都只删行、不造行。本清扫与之同类：**自带审计**，
// 不碰 `Store.Add`，也不复用任何「想写记忆」的调用点。
//
// 🔴 **删除范围写死**（`design/23 §19.2`，逐条可测）：
//
//	expires_at <> '' ∧ expires_at <= now ∧ status = 'active' ∧ 来源非受限
//
//   - **空 `expires_at` = 永不过期，绝不删**。长期 / 日常记忆的 `expires_at` 是
//     空串（`store_schema.go` 的 DDL 默认值），因此它们**结构性**进不了扫描集合 ——
//     这不是靠注释约定，而是靠 SQL 里那条 `expires_at <> ''` 谓词。
//   - **受限来源（群聊 / 陌生人）不参与**：判据复用整理/审查同一处实现
//     `nonOwnerSource`（`tidy_rules.go`），因此三处口径不可能各说各话。
//   - **已软删 / 已归档的行不参与**（`status='active'` 谓词）：它们不是「到期的
//     临时记忆」，是整理软动作留下的行；该不该清是另一件事。

const (
	// DefaultExpireSweepInterval 是到期清扫的默认周期：每小时一次
	// （`design/23 §19.1` 表格「到期后 ≤ 一个清扫周期（默认每小时扫一次）」）。
	// 与 `DefaultIngestReviewGCInterval` 同值同由：清扫只是删几行，频率再高无收益。
	DefaultExpireSweepInterval = time.Hour

	// expireSweepBatchSize 是单批取行 / 删除的行数上限，maxExpireSweepBatches 是
	// 一轮清扫的批数上限。两者一起把「一轮清扫」的时长、单次内存占用与**连接池
	// 占用**（本项目的 SQLite 池是单连接）都变得有界 —— 照 `preinsert_gc.go` 里
	// `ingestReviewSweepBatchSize` / `maxIngestReviewSweepBatches` 那一对常量的
	// 批删范式（`design/23 §19.3-3`「一批删除有条数 / 批数上限」）。
	// 上限之积 = 一轮最多删 1 万行；积压时剩下的留给下一轮，不一次 DELETE 全表。
	expireSweepBatchSize  = 500
	maxExpireSweepBatches = 20

	// expireAuditSummaryRunes 是审计摘要的**按 rune 固定上限**（`design/23 §19.4`）。
	// 48 个 rune 足够一眼看出「删的是哪条」，同时把审计表增长压住 —— 到期审计是
	// **逐条高基数**行（一行一个 id），不截断等于把正文搬进审计表。取 48 而不是
	// 更宽的值：摘要只需可辨认，不需要可复原（正文早已不可逆地删掉了）。
	expireAuditSummaryRunes = 48

	// expireAuditVerb 是到期删除的审计动作名，与 `memory.tidy.run/.apply/.reject`
	// 同族（`design/23 §19.4`）。**只此一处**允许审计携带内容片段，且
	// 只在删除这一动作上（本包其余审计一律只带计数）。
	expireAuditVerb = "memory.tidy.expire"
)

// ExpirySweepPolicy 是一轮清扫的口径。零值即默认（由 normalized 填批次上限），
// 因此调用方可以只写 `ExpirySweepPolicy{}`。
//
// 它**没有「关掉清扫」的开关** —— 这是刻意的：临时记忆的到期删除是「声明了生命
// 周期就兑现定义」，留一个能关掉它的开关等于让库里的临时层无界留存（读侧闸只会
// 让它们「看不见」，不会让它们消失）。真要不产生临时记忆，该关的是第五态
// `defer` 的判据，而不是清扫。
type ExpirySweepPolicy struct {
	// BatchSize 是单批上限；<=0 取 expireSweepBatchSize。
	BatchSize int
	// MaxBatches 是一轮的批数上限；<=0 取 maxExpireSweepBatches。
	MaxBatches int
}

// normalized 填默认值。负值与非正值一律取默认（配置异常不该让清扫停摆）。
func (p ExpirySweepPolicy) normalized() ExpirySweepPolicy {
	if p.BatchSize <= 0 {
		p.BatchSize = expireSweepBatchSize
	}
	if p.MaxBatches <= 0 {
		p.MaxBatches = maxExpireSweepBatches
	}
	return p
}

// ExpirySweepStats 是一轮清扫的汇总（**只记计数，绝不记任何行内容**）。
// 它进调用方的一条汇总日志，不进审计 —— 审计是逐行的那一条（ExpireFunc）。
type ExpirySweepStats struct {
	// Deleted 是本轮物理删掉的行数。
	Deleted int64
	// Restricted 是本轮**跳过**的受限来源行数（群聊 / 陌生人）。它非零只说明库里
	// 存在这类历史遗留行，是运维信号，不是错误。
	Restricted int
	// Batches 是本轮实际执行的取行批数。
	Batches int
}

// ExpiredMemory 是一条**即将被物理删除**的到期临时记忆，只带审计需要的字段。
// 它刻意不带 tags / confidence 之类的列：审计口径就是 `design/23 §19.4` 那一行
// `id + persona_id + scope + expires_at + 截断摘要`，多带一列就多一分泄漏面。
type ExpiredMemory struct {
	ID        string
	PersonaID string
	Scope     string
	ExpiresAt string
	Content   string
}

// ExpireFunc 在**每行被删除之前**调用一次，是本清扫的审计出口
// （`design/23 §19.3-3`：每行删除前先写审计）。它没有返回值：审计出口失败与否
// 都不该改变「这一行删不删」—— 审计是留痕，不是闸。为 nil 时不写审计
// （只有不关心留痕的单元测试会这么用；生产的后台件恒传一个）。
type ExpireFunc func(row ExpiredMemory)

// expireCursor 是清扫的**键集游标**（`(expires_at, rowid)`），让被跳过的受限行
// 只被读到一次：每批处理完就前移到该批最后一行，下一批从它之后取。
//
// 为什么不用 `OFFSET`：受限行会被跳过但不删，用 OFFSET 分页时它们会**反复占住
// 每一批的头部**，极端情况下（头部 500 行都是受限行）真该删的行永远轮不到。
// 键集游标让进度与「删了多少」解耦 —— 只要取到了行，扫描位置就一定前移。
type expireCursor struct {
	expiresAt string
	rowid     int64
}

// expiredMemoryRow 是扫描回来的一行：审计字段 + 受限判定要用的溯源 + 游标。
type expiredMemoryRow struct {
	ExpiredMemory
	sourceConversation string
	cursor             expireCursor
}

// selectExpiredMemoriesSQL 按 `design/23 §19.2` 的扫描集合取一批到期行。
//
// 三条谓词各自对应一条边界：`expires_at <> ”` 排除「永不过期」的行（空串是
// 长期 / 日常记忆的默认值，见 store_schema.go 的 DDL），`expires_at <= ?` 取到期
// 的行，`status = ?` 只取在册行。时间戳用定宽 UTC 文本比较：字典序等价于时间序，
// 与写入侧 `formatTime`（timefmt.go）同源 —— 换成 SQLite 的 datetime('now') 会
// 静默判错（timegate.go 有这条反证的实证）。
//
// 游标谓词用 `(expires_at, rowid)` **元组比较**：expires_at 是文本列，同一批
// 落库的行可能同值，只比 expires_at 会让扫描卡在原地。
const selectExpiredMemoriesSQL = `SELECT id, persona_id, scope, expires_at, content,
	source_conversation, rowid
	FROM memory_items
	WHERE status = ? AND expires_at <> '' AND expires_at <= ?
	  AND (expires_at > ? OR (expires_at = ? AND rowid > ?))
	ORDER BY expires_at, rowid
	LIMIT ?`

// deleteExpiredMemorySQL 按主键删一行。它**只**由上一条语句选出来的 id 驱动，
// 因此不存在「按用户输入删行」的面。
const deleteExpiredMemorySQL = `DELETE FROM memory_items WHERE id = ?`

// SweepExpiredMemories 跑一轮到期清扫，返回汇总。now 由调用方给出（便于测试注入
// 固定时钟；本包不在 Store 里藏墙钟，与 SweepIngestReviews 同口径）。
//
// 边界（`design/23 §19.3-3`）：
//
//   - **有界**：一轮最多 maxExpireSweepBatches × expireSweepBatchSize 行，
//     每批一次取行 + 逐行删除，连接池占用与时长都有上界；
//   - **有审计**：每行删除前调 onExpire（§19.4 的 `memory.tidy.expire`）；
//   - **可停**：ctx 取消即刻返回；循环体在每批开跑前复核 ctx.Err()。
//
// 失败语义：取行 / 删行失败时**返回错误并带上已完成的统计**（调用方记 WARN 后等
// 下一轮）。已删的行保持删除、未删的行绝不误删 —— 一轮失败只是「这一轮没扫干净」。
func (s *Store) SweepExpiredMemories(
	ctx context.Context, now time.Time, policy ExpirySweepPolicy, onExpire ExpireFunc,
) (ExpirySweepStats, error) {
	var stats ExpirySweepStats
	if s == nil {
		return stats, ErrNilStore
	}
	p := policy.normalized()
	// 与写入侧同源：列里存的是 formatTime 产出的定宽 UTC 文本，右边也必须由
	// 同一个函数产出（见 timegate.go 的「同源」口径）。
	cutoff := formatTime(now.UTC())
	cursor := expireCursor{rowid: -1}

	for batch := 0; batch < p.MaxBatches; batch++ {
		if err := ctx.Err(); err != nil {
			return stats, fmt.Errorf("memory: sweep expired memories: %w", err)
		}
		rows, err := s.listExpiredMemories(ctx, cutoff, cursor, p.BatchSize)
		if err != nil {
			return stats, err
		}
		if len(rows) == 0 {
			break
		}
		stats.Batches++
		// 游标前移与「删没删」无关：受限行被跳过时也必须往前走，否则下一批会
		// 取回同一批行（见 expireCursor 的注释）。
		cursor = rows[len(rows)-1].cursor

		for _, row := range rows {
			if nonOwnerSource(row.sourceConversation) {
				stats.Restricted++
				continue
			}
			if onExpire != nil {
				onExpire(row.ExpiredMemory)
			}
			if err := s.purgeExpiredMemory(ctx, row.ID); err != nil {
				return stats, err
			}
			stats.Deleted++
		}
		if len(rows) < p.BatchSize {
			break
		}
	}
	return stats, nil
}

// listExpiredMemories 取一批到期行（含游标与溯源列）。limit 非正时返回空切片，
// 因此调用方不必自己防零。
//
// 它把整批**读进内存后才返回**（游标在返回前关闭），所以调用方的删除不会与一个
// 还开着的 SELECT 游标打架 —— 本项目的 SQLite 池是单连接，同一连接上「边读边删」
// 是自寻死路（写要等读，读永远读不完）。
func (s *Store) listExpiredMemories(
	ctx context.Context, cutoff string, cursor expireCursor, limit int,
) ([]expiredMemoryRow, error) {
	if limit <= 0 {
		return nil, nil
	}
	rows, err := s.db.QueryContext(ctx, selectExpiredMemoriesSQL,
		string(tidyStatusActive), cutoff, cursor.expiresAt, cursor.expiresAt, cursor.rowid, limit)
	if err != nil {
		return nil, fmt.Errorf("memory: list expired items: %w", err)
	}
	defer func() { _ = rows.Close() }()

	out := make([]expiredMemoryRow, 0, limit)
	for rows.Next() {
		var r expiredMemoryRow
		if err := rows.Scan(&r.ID, &r.PersonaID, &r.Scope, &r.ExpiresAt, &r.Content,
			&r.sourceConversation, &r.cursor.rowid); err != nil {
			return nil, fmt.Errorf("memory: scan expired item: %w", err)
		}
		r.cursor.expiresAt = r.ExpiresAt
		out = append(out, r)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("memory: iterate expired items: %w", err)
	}
	return out, nil
}

// purgeExpiredMemory 删掉一行到期临时记忆，并把它的向量从持久索引里同步清掉。
//
// 两步的顺序不是随手写的：**先删行、再清向量**。反过来（先清向量）的话，删行
// 失败会留下一行「有行、无向量」的记忆 —— 那一行不会被返回（读侧闸挡着），
// 但检索的语义路径会有一个查不到向量的 id；而先删行时最坏情况只是留一个孤儿
// 向量，那是本 finding 描述的既有状态（可下一轮再清），不会更坏。
func (s *Store) purgeExpiredMemory(ctx context.Context, id string) error {
	if _, err := s.db.ExecContext(ctx, deleteExpiredMemorySQL, id); err != nil {
		return fmt.Errorf("memory: delete expired item %s: %w", id, err)
	}
	s.purgeExpiredVector(ctx, id)
	return nil
}

// purgeExpiredVector 把刚被删除的行从持久语义索引里同步清掉（切片 M1 的 reviewer
// 交上来的 finding）：过期行的向量若留在索引里，会占掉 `idx.Search` 的 top-K
// 预算，只降召回质量、不泄漏（该行已被读侧时效闸挡住，不可能被返回）。
//
// 未接索引、或后端不支持删除（只读的桩）时是空操作；失败只记 WARN ——
// **向量清不掉不该让「行已删」这件事失败**，也不该让下一行不删。
func (s *Store) purgeExpiredVector(ctx context.Context, id string) {
	idx := s.vectorIndex()
	if idx == nil {
		return
	}
	deleter, ok := idx.(vector.Deleter)
	if !ok {
		s.log.Debug("memory: vector index does not support delete, expired vector left behind", "id", id)
		return
	}
	if err := deleter.Delete(ctx, id); err != nil {
		s.log.Warn("memory: expired vector purge failed", "id", id, "error", err)
	}
}

// expireAuditResult 渲染 `design/23 §19.4` 要求的定长结构化串：
//
//	id=<id> persona_id=<..> scope=<..> expires_at=<..> summary=<截断摘要>
//
// 摘要是**本仓唯一**允许进入审计的内容片段，边界限这一个动作 + 固定截断
// （主会话已裁定允许）。它先做空白归一（正文里的换行 / 制表符会把审计行拆成
// 多行，那既破坏「一行一条」的可解析性，也把事情变成「把正文刷进日志」），
// 再按 rune 截断。
func expireAuditResult(row ExpiredMemory) string {
	return fmt.Sprintf("id=%s persona_id=%s scope=%s expires_at=%s summary=%s",
		row.ID, row.PersonaID, row.Scope, row.ExpiresAt, clipExpireSummary(row.Content))
}

// clipExpireSummary 把正文压成一行至多 expireAuditSummaryRunes 个 rune 的摘要。
// 空白（含换行）一律归一成单个空格；截断时追加省略号，让「这条被截过」在审计
// 里可见 —— 省略号本身也占一个 rune，总长仍恒为 ≤ 上限（上限-1 + 省略号）。
//
// 空正文（理论上不该存在，正文是 NOT NULL 且入库时校验非空）返回空串，
// 让审计行仍可解析。
func clipExpireSummary(content string) string {
	flat := strings.Join(strings.Fields(content), " ")
	runes := []rune(flat)
	if len(runes) <= expireAuditSummaryRunes {
		return flat
	}
	return string(runes[:expireAuditSummaryRunes-1]) + "…"
}