Cyrene_v2/internal/runtime/memory/expire_loop_test.go

package memory

import (
	"bytes"
	"context"
	"log/slog"
	"strings"
	"sync"
	"testing"
	"time"

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

// 本文件锁定期清扫的**接线与后台形态**（design/23 §19）：挂点（TidyService.Start
// 里的 `go s.expirySweepLoop(cctx)`）、随 Stop 可停、一轮只写一行汇总日志，以及
// 「删行时同步清掉向量」这条 finding 的集成断言。清扫自身的边界（哪些行删、
// 哪些行留、审计长什么样）在 expire_test.go。
//
// TestExpirySweepOnce_一轮只写一行汇总：逐条的那一行进审计，日志里一轮只有一行
// （§19.4「一轮汇总只进日志，绝不逐条把正文刷进日志」）。
func TestExpirySweepOnce_一轮只写一行汇总(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	store := newTestStore(t)
	now := tidyTestTime(t)

	var buf bytes.Buffer
	var mu sync.Mutex
	audit := func(who, what, result string) {
		mu.Lock()
		defer mu.Unlock()
		_ = who
		_ = what
		_ = result
	}
	svc, err := NewTidyService(tidyTestCfg(0.80), TidyDeps{
		Store: store, Log: slog.New(slog.NewTextHandler(&buf, nil)),
		Now: fixedTidyClock(now), Audit: audit,
	})
	if err != nil {
		t.Fatalf("NewTidyService: %v", err)
	}
	for i := 0; i < 3; i++ {
		addItem(t, store, MemoryItem{
			PersonaID: "persona.a", Scope: ScopePersona, Kind: KindRelationship,
			Content: "日志里不该出现的正文片段", ExpiresAt: now.Add(-time.Minute),
		})
	}

	svc.expirySweepOnce(ctx)

	lines := strings.Split(strings.TrimSpace(buf.String()), "\n")
	if len(lines) != 1 {
		t.Fatalf("一轮写入了 %d 行日志，want 1（只允许一行汇总）：\n%s", len(lines), buf.String())
	}
	if !strings.Contains(lines[0], "deleted=3") {
		t.Fatalf("汇总行没有计数：%q", lines[0])
	}
	if strings.Contains(buf.String(), "日志里不该出现的正文片段") {
		t.Fatalf("日志里出现了正文 —— 逐条正文只许进审计的截断摘要：%q", buf.String())
	}
}

// TestTidyServiceStart_清扫随服务起停 钉住**接线**（design/23 §19.3-3 的「可停」）：
// 清扫起在 TidyService.Start 里，与心跳共用同一个 cctx 与 wg，因此
//
//	① Start 一跑它就立刻扫一轮（启动补扫，清掉上次退出后到期的行）；
//	② Stop 会取消它并等它退出 —— Stop 没超时即证明它真的随 ctx 走了。
//
// 它同时是 `cmd/cyrene/wire_memory_tidy_test.go` 那条 seam 断言的运行时对照：把
// Start 里那行 `go s.expirySweepLoop(cctx)` 摘掉，这条用例会红（行永远不会被删，
// 且 Stop 会因为 wg 计数对不上而超时）。
func TestTidyServiceStart_清扫随服务起停(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	store := newTestStore(t)
	// 10:00Z：不在安静窗（23–7）内 ⇒ Start 里的补跑判定为「不该跑」，整理不搅局。
	now := tidyTestTime(t)
	row := addItem(t, store, MemoryItem{
		PersonaID: "persona.a", Scope: ScopePersona, Kind: KindRelationship,
		Content: "随服务启动被扫掉的到期行", ExpiresAt: now.Add(-time.Minute),
	})

	var (
		mu     sync.Mutex
		audits []string
	)
	svc := newTidyServiceForTest(t, store, tidyTestCfg(0.80), nil, fixedTidyClock(now),
		func(_, what, result string) {
			mu.Lock()
			defer mu.Unlock()
			audits = append(audits, what+" "+result)
		})
	// 假等待：把整理心跳停在 ctx 上（真定时器要等 24h 档）。它只影响心跳 ——
	// 清扫是**另一个** goroutine + 自己的小时级 ticker。
	svc.wait = func(ctx context.Context, _ time.Duration) bool {
		<-ctx.Done()
		return false
	}

	svc.Start(ctx)

	waitUntilRowGone(t, store, row.ID)

	stopCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()
	if err := svc.Stop(stopCtx); err != nil {
		t.Fatalf("Stop 未能在预算内合上 —— 清扫没有随 ctx 退出 / 没登记进同一个 wg: %v", err)
	}

	mu.Lock()
	defer mu.Unlock()
	if len(audits) != 1 || !strings.HasPrefix(audits[0], expireAuditVerb+" ") {
		t.Fatalf("审计 = %v, want 一条 what=%s", audits, expireAuditVerb)
	}
}

// TestTidyServiceStart_整理关掉时清扫不跑 钉住那条**行为契约**（不是断言字段那种
// 假牙）：`MEMORY_TIDY_ENABLED=false` ⇒ 到期清扫**不跑**，过期行**一条都不删**。
//
// 为什么值得一条专门的用例：清扫那行 `go s.expirySweepLoop(cctx)` 就排在
// `if !s.cfg.Enabled { … return }` **之后**。将来谁把 Enabled 检查挪到后面、
// 或让 Start 在关掉时也走完整条路，过期行就会在「整理被关掉」时被物理删除 ——
// 那是与「后台件随宿主启停」（`preinsert_gc` 随审查服务启停，design/23 §19.3-2）
// 相反的行为，而且**不会红任何别的用例**（读侧闸照样挡着，表面上一切正常）。
//
// 断言是「没有删除发生 + 没有审计 + running 保持 false」三件一起 —— 只看
// `cfg.Enabled` 字段是假牙：那证明不了 Start 真的按关掉处理。
func TestTidyServiceStart_整理关掉时清扫不跑(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	store := newTestStore(t)
	now := tidyTestTime(t)
	row := addItem(t, store, MemoryItem{
		PersonaID: "persona.a", Scope: ScopePersona, Kind: KindRelationship,
		Content: "整理关掉时不该被删的到期行", ExpiresAt: now.Add(-time.Minute),
	})

	var (
		mu     sync.Mutex
		audits []string
	)
	cfg := tidyTestCfg(0.80)
	cfg.Enabled = false
	svc := newTidyServiceForTest(t, store, cfg, nil, fixedTidyClock(now),
		func(_, what, result string) {
			mu.Lock()
			defer mu.Unlock()
			audits = append(audits, what+" "+result)
		})
	// 假等待：万一清扫/心跳被起起来了，它会停在这里而不是空转（不影响断言）。
	svc.wait = func(ctx context.Context, _ time.Duration) bool {
		<-ctx.Done()
		return false
	}

	svc.Start(ctx)

	// ① 确定性断言：Start 走的是「关掉 ⇒ 直接返回」那条路。
	if svc.running {
		t.Fatal("Enabled=false 时 Start 应当直接返回，running 必须保持 false")
	}
	// ② 让「清扫其实被起起来了」有机会显形：清扫是启动即扫一轮，中间只有
	// 「取行 + 删行」一次 SQLite 往返（同一台机上实测在微秒级，远比这个窗口小）；
	// 对照组 TestTidyServiceStart_清扫随服务起停 用的就是同一个语料与同一个时钟，
	// 它在毫秒级内把行删掉。给同一个数量级的窗口再断言「什么也没发生」。
	time.Sleep(200 * time.Millisecond)
	if !memoryRowExists(t, store, row.ID) {
		t.Fatal("整理被关掉了，到期行却被删了 —— 清扫跑在了 Enabled 闸之外")
	}
	mu.Lock()
	defer mu.Unlock()
	if len(audits) != 0 {
		t.Fatalf("整理被关掉了，却写出了到期审计：%v", audits)
	}
}

// TestSweepExpiredMemories_同步清掉向量 是切片 M1 的 reviewer 交上来的那条 finding
// 的断言：过期行被物理删除时，它的**向量也必须一起清掉** —— 否则向量留在索引里
// 占掉 idx.Search 的 top-K 预算，召回质量静默下降（行本身已被读侧闸挡住，不会泄漏）。
func TestSweepExpiredMemories_同步清掉向量(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	store := newTestStore(t)
	idx := newTestVectorIndex(t, 64)
	store.attachVectorIndex(idx)

	now := tidyTestTime(t)
	expired := addItem(t, store, MemoryItem{
		PersonaID: "persona.a", Scope: ScopePersona, Kind: KindRelationship,
		Content: "到期的向量条目", ExpiresAt: now.Add(-time.Minute),
	})
	kept := addItem(t, store, MemoryItem{
		PersonaID: "persona.a", Scope: ScopePersona, Kind: KindRelationship,
		Content: "长期留着的向量条目",
	})

	if !vectorIndexHas(t, ctx, idx, expired.ID) {
		t.Fatal("前置条件不成立：到期行的向量本来就该在索引里（写入侧没喂进去）")
	}
	if !vectorIndexHas(t, ctx, idx, kept.ID) {
		t.Fatal("前置条件不成立：保留行的向量本来就该在索引里")
	}

	if _, err := store.SweepExpiredMemories(ctx, now, ExpirySweepPolicy{}, nil); err != nil {
		t.Fatalf("SweepExpiredMemories: %v", err)
	}

	if vectorIndexHas(t, ctx, idx, expired.ID) {
		t.Fatal("到期行删了、向量还在索引里 —— 它会占掉 top-K 预算（M1 reviewer 的 finding）")
	}
	if !vectorIndexHas(t, ctx, idx, kept.ID) {
		t.Fatal("保留行的向量被误清了（清算成了通杀）")
	}
}

// vectorIndexHas 用 n-gram 索引自带的 embedder 查一次，报告某 id 是否还在索引里。
func vectorIndexHas(t *testing.T, ctx context.Context, idx vector.Index, id string) bool {
	t.Helper()
	vecs, err := idx.Embed(ctx, []string{"到期的向量条目 长期留着的向量条目"})
	if err != nil {
		t.Fatalf("embed query: %v", err)
	}
	hits, err := idx.Search(ctx, vecs[0], 50)
	if err != nil {
		t.Fatalf("search index: %v", err)
	}
	for _, h := range hits {
		if h.ID == id {
			return true
		}
	}
	return false
}

// TestClipExpireSummary 单钉摘要裁剪这件小事：空白归一（含换行）、按 rune 截断、
// 截断才加省略号。
func TestClipExpireSummary(t *testing.T) {
	t.Parallel()
	tests := []struct {
		name string
		in   string
		want string
	}{
		{name: "短文本原样", in: "谢谢啦", want: "谢谢啦"},
		{name: "空白归一", in: "一\n\t二  三", want: "一 二 三"},
		{name: "空文本", in: "   ", want: ""},
		{
			name: "超长按 rune 截断并省略号",
			in:   strings.Repeat("很", expireAuditSummaryRunes+10),
			want: strings.Repeat("很", expireAuditSummaryRunes-1) + "…",
		},
	}
	for _, tc := range tests {
		tc := tc
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()
			if got := clipExpireSummary(tc.in); got != tc.want {
				t.Fatalf("clipExpireSummary(%q) = %q, want %q", tc.in, got, tc.want)
			}
		})
	}
}

// waitUntilRowGone 轮询等一行被删掉（清扫跑在另一个 goroutine 上）。
func waitUntilRowGone(t *testing.T, store *Store, id string) {
	t.Helper()
	deadline := time.Now().Add(5 * time.Second)
	for time.Now().Before(deadline) {
		if !memoryRowExists(t, store, id) {
			return
		}
		time.Sleep(5 * time.Millisecond)
	}
	t.Fatalf("等不到行 %s 被清扫删除（清扫没跑起来？）", id)
}