AskaEth
454ddb0b70
主人实测：memory_items 1191 行、候选 77 条全 pending、memory_tidy_runs 仅 1 行、两晚定时没跑。 根因两条（都在这片里修掉）： ① **以前根本没有执行面** —— 只有只读试算，候选永远 pending ② **调度被重启饿死** —— interval=24h 的 ticker 从进程启动算起，两天重启十几次 ⇒ 24h 永远走不完 - 应用路径：逐条 / 按 run 批量（默认 100、硬顶 500、超限 truncated）/ revert；**软合并** （status + merged_into/merged_from/merge_count，**绝不改 content、行不删**）+ 审计 （memory.tidy.apply / memory.tidy.revert）+ **幂等**（候选状态条件更新 + 行 status 闸两道） - 调度：改成**到点即跑**（catchUp：无历史+窗口内 / 距上次 run ≥ interval / 窗口起点晚于上次 run）， 「上次 run」以 memory_tidy_runs 为**持久事实**；注入时钟用例证明**重启不饿死** - 手动真跑端点 POST /admin/memories/tidy/run（同档鉴权 + 审计 + 回执：新增/合并/软删/耗时） - 候选表加 run_id 列 + 索引（前滚；顺序=建表→补列→建索引，否则既有库会 no such column） - 3 个变异自检（删 catchUp / 去掉两道 status 闸 / truncated 恒 false） ⚠️ 生产库零动作：1191 行真跑与 77 条候选应用**未执行**（按红线由主人确认后主会话执行） ⚠️ 契约段在报告里（candidates 列表/apply/revert/tidy.run），Docs/ 待登记
351 lines
12 KiB
Go
351 lines
12 KiB
Go
package memory
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"sort"
|
||
"strings"
|
||
"testing"
|
||
"time"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding"
|
||
)
|
||
|
||
// 本文件锁定记忆整理「试算」（契约 v0.75；切片 MT1）的四条判据：
|
||
//
|
||
// ① **零副作用**：试算前后 memory_items / memory_tidy_candidates /
|
||
// memory_tidy_runs **逐行不变**（这是 dry-run 的命门）；
|
||
// ② **真复用同一套规则**：同一批数据上，试算候选与真跑一轮落库的候选
|
||
// 逐字段一致（不是另一套阈值实现）；
|
||
// ③ **尊重既有配置上限**：每分区候选上限（CandidateLimit）在试算里同样生效；
|
||
// ④ **诚实错误**：语义召回不可用 / 整理被配置关闭时如实报错，绝不回空候选
|
||
// 假装「没有建议」。
|
||
|
||
// failingEmbedder 模拟「语义召回不可用」（例如 llama.cpp :11434 不可达）。
|
||
type failingEmbedder struct{ err error }
|
||
|
||
func (f failingEmbedder) Embed(context.Context, []string) ([]embedding.Vector, error) {
|
||
return nil, f.err
|
||
}
|
||
|
||
func (f failingEmbedder) Dim() int { return 8 }
|
||
|
||
// 三张相关表的**显式列清单**（不用 SELECT *）：逐行快照按完整行内容比对，
|
||
// 任何一列被改动都会让快照不等。
|
||
var (
|
||
memoryItemsColumns = []string{
|
||
"id", "persona_id", "role_id", "scope", "kind", "content", "tags", "confidence",
|
||
"source_conversation", "created_at", "updated_at", "decay_weight", "status",
|
||
"merged_into", "merged_from", "merge_count", "occurred_at", "valid_to", "expires_at",
|
||
}
|
||
tidyCandidateColumns = []string{
|
||
"id", "persona_id", "scope", "project_id", "kind", "action", "source_ids", "target_id",
|
||
"proposed_content", "merged_content", "reason", "evidence", "rule_version",
|
||
"score_min", "score_max", "status", "idempotency_key", "run_id", "created_at",
|
||
"decided_at", "decided_by",
|
||
}
|
||
tidyRunColumns = []string{
|
||
"id", "started_at", "finished_at", "scanned", "candidates", "applied", "skipped",
|
||
"errors", "vetoed", "veto_reasons", "rule_version",
|
||
}
|
||
)
|
||
|
||
// tidyTableSnapshot 是三张表在某一时刻的逐行内容。
|
||
type tidyTableSnapshot struct {
|
||
items []string
|
||
candidates []string
|
||
runs []string
|
||
}
|
||
|
||
// dumpTidyTable 把一张表按主键序渲染成字符串行。
|
||
func dumpTidyTable(t *testing.T, store *Store, table string, columns []string) []string {
|
||
t.Helper()
|
||
query := fmt.Sprintf("SELECT %s FROM %s ORDER BY id", strings.Join(columns, ", "), table)
|
||
rows, err := store.db.QueryContext(context.Background(), query)
|
||
if err != nil {
|
||
t.Fatalf("dump %s: %v", table, err)
|
||
}
|
||
defer func() { _ = rows.Close() }()
|
||
|
||
out := make([]string, 0)
|
||
for rows.Next() {
|
||
vals := make([]any, len(columns))
|
||
ptrs := make([]any, len(columns))
|
||
for i := range vals {
|
||
ptrs[i] = &vals[i]
|
||
}
|
||
if err := rows.Scan(ptrs...); err != nil {
|
||
t.Fatalf("scan %s: %v", table, err)
|
||
}
|
||
parts := make([]string, 0, len(columns))
|
||
for i, col := range columns {
|
||
parts = append(parts, col+"="+formatCell(vals[i]))
|
||
}
|
||
out = append(out, strings.Join(parts, "|"))
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
t.Fatalf("iterate %s: %v", table, err)
|
||
}
|
||
return out
|
||
}
|
||
|
||
// formatCell 把 scan 出来的值渲染成稳定文本（TEXT 列以 []byte 出现）。
|
||
func formatCell(v any) string {
|
||
switch got := v.(type) {
|
||
case nil:
|
||
return "<nil>"
|
||
case []byte:
|
||
return string(got)
|
||
default:
|
||
return fmt.Sprintf("%v", got)
|
||
}
|
||
}
|
||
|
||
// snapshotTidyTables 拍一张三表快照。
|
||
func snapshotTidyTables(t *testing.T, store *Store) tidyTableSnapshot {
|
||
t.Helper()
|
||
return tidyTableSnapshot{
|
||
items: dumpTidyTable(t, store, "memory_items", memoryItemsColumns),
|
||
candidates: dumpTidyTable(t, store, "memory_tidy_candidates", tidyCandidateColumns),
|
||
runs: dumpTidyTable(t, store, "memory_tidy_runs", tidyRunColumns),
|
||
}
|
||
}
|
||
|
||
// assertRowsUnchanged 逐行比对快照：任何一行被新增/删除/改列都会失败。
|
||
func assertRowsUnchanged(t *testing.T, what string, before, after []string) {
|
||
t.Helper()
|
||
if len(before) != len(after) {
|
||
t.Fatalf("%s 行数 %d → %d，试算有副作用", what, len(before), len(after))
|
||
}
|
||
for i := range before {
|
||
if before[i] != after[i] {
|
||
t.Fatalf("%s 第 %d 行被改动：\n before=%s\n after =%s", what, i, before[i], after[i])
|
||
}
|
||
}
|
||
}
|
||
|
||
// candidateSignature 把一条候选渲染成与 id/落库时间无关的稳定签名，用于比对
|
||
// 「试算候选」与「真跑一轮的候选」是否逐字段一致。
|
||
func candidateSignature(c TidyCandidate) string {
|
||
return fmt.Sprintf("%s|sources=%s|target=%s|kind=%s|reason=%s|proposed=%q|merged=%q|rule=%s|min=%.6f|max=%.6f",
|
||
c.Action, strings.Join(c.SourceIDs, ","), c.TargetID, c.Kind, c.Reason,
|
||
c.ProposedContent, c.MergedContent, c.RuleVersion, c.ScoreMin, c.ScoreMax)
|
||
}
|
||
|
||
// sortedSignatures 返回按签名排序的候选集合（消除落库读回的排序差异）。
|
||
func sortedSignatures(cands []TidyCandidate) []string {
|
||
out := make([]string, 0, len(cands))
|
||
for _, c := range cands {
|
||
out = append(out, candidateSignature(c))
|
||
}
|
||
sort.Strings(out)
|
||
return out
|
||
}
|
||
|
||
// seedDryRunFixture 落三条记忆：一对必合（短 ⊆ 长）+ 一条一次性事件（走 archive）。
|
||
func seedDryRunFixture(t *testing.T, store *Store) {
|
||
t.Helper()
|
||
base := tidyTestTime(t)
|
||
seedTidyItem(t, store, habitItem(fixtureHabitShort, base, 0.9))
|
||
seedTidyItem(t, store, habitItem(fixtureHabitLong, base.Add(time.Hour), 0.9))
|
||
seedTidyItem(t, store, habitItem(fixtureOneOffEvent, base.Add(2*time.Hour), 0.9))
|
||
}
|
||
|
||
// dryRunFixtureEmbedder 让上述三条恰好产出一 merge + 一 archive。
|
||
func dryRunFixtureEmbedder() stubEmbedder {
|
||
return stubEmbedder{vecs: map[string]embedding.Vector{
|
||
fixtureHabitShort: {1, 0},
|
||
fixtureHabitLong: simVector(0.95),
|
||
fixtureOneOffEvent: simVector(0.9),
|
||
}}
|
||
}
|
||
|
||
// TestTidyDryRun_ZeroSideEffects 是判据 ①：试算跑完之后，memory_items /
|
||
// memory_tidy_candidates / memory_tidy_runs **逐行**与跑之前完全一致。
|
||
func TestTidyDryRun_ZeroSideEffects(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
store := newTestStore(t)
|
||
seedDryRunFixture(t, store)
|
||
before := snapshotTidyTables(t, store)
|
||
|
||
tr := newTestTidier(t, store, tidyTestCfg(0.80), dryRunFixtureEmbedder())
|
||
out, err := tr.DryRun(ctx)
|
||
if err != nil {
|
||
t.Fatalf("DryRun: %v", err)
|
||
}
|
||
if len(out.Candidates) == 0 {
|
||
t.Fatalf("夹具应当产出候选（否则零副作用断言是空转）：%+v", out.Stats)
|
||
}
|
||
|
||
assertRowsUnchanged(t, "memory_items", before.items, snapshotTidyTables(t, store).items)
|
||
assertRowsUnchanged(t, "memory_tidy_candidates", before.candidates, snapshotTidyTables(t, store).candidates)
|
||
assertRowsUnchanged(t, "memory_tidy_runs", before.runs, snapshotTidyTables(t, store).runs)
|
||
}
|
||
|
||
// TestTidyDryRun_MatchesRealRunCandidates 是判据 ②：试算走的必须是**同一套规则**
|
||
// —— 同一批数据上，试算候选与真跑一轮落库的候选逐字段一致。
|
||
func TestTidyDryRun_MatchesRealRunCandidates(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
store := newTestStore(t)
|
||
seedDryRunFixture(t, store)
|
||
tr := newTestTidier(t, store, tidyTestCfg(0.80), dryRunFixtureEmbedder())
|
||
|
||
out, err := tr.DryRun(ctx)
|
||
if err != nil {
|
||
t.Fatalf("DryRun: %v", err)
|
||
}
|
||
if len(out.Candidates) == 0 {
|
||
t.Fatal("夹具应当产出候选")
|
||
}
|
||
if _, err := tr.RunOnce(ctx); err != nil {
|
||
t.Fatalf("RunOnce: %v", err)
|
||
}
|
||
persisted, err := store.ListTidyCandidates(ctx, "", 100)
|
||
if err != nil {
|
||
t.Fatalf("ListTidyCandidates: %v", err)
|
||
}
|
||
|
||
want := sortedSignatures(out.Candidates)
|
||
got := sortedSignatures(persisted)
|
||
if strings.Join(got, "\n") != strings.Join(want, "\n") {
|
||
t.Fatalf("试算候选与真跑一轮的候选不一致：\n dry=%v\n real=%v", want, got)
|
||
}
|
||
}
|
||
|
||
// TestTidyDryRun_RespectsCandidateLimit 是判据 ③：每分区候选上限（既有配置旋钮）
|
||
// 在试算里同样生效，且超出的候选被如实登记（Skipped / Truncated），绝不静默。
|
||
func TestTidyDryRun_RespectsCandidateLimit(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
store := newTestStore(t)
|
||
seedDryRunFixture(t, store)
|
||
|
||
cfg := tidyTestCfg(0.80)
|
||
cfg.CandidateLimit = 1
|
||
tr := newTestTidier(t, store, cfg, dryRunFixtureEmbedder())
|
||
|
||
out, err := tr.DryRun(ctx)
|
||
if err != nil {
|
||
t.Fatalf("DryRun: %v", err)
|
||
}
|
||
if len(out.Candidates) != cfg.CandidateLimit {
|
||
t.Fatalf("候选数 = %d, want %d（必须尊重既有每分区上限）", len(out.Candidates), cfg.CandidateLimit)
|
||
}
|
||
if out.Stats.Skipped != 1 {
|
||
t.Fatalf("超限候选未被如实登记：skipped=%d, want 1", out.Stats.Skipped)
|
||
}
|
||
if len(out.Stats.Truncated) == 0 {
|
||
t.Fatalf("候选上限命中必须留下截断痕迹：%+v", out.Stats)
|
||
}
|
||
}
|
||
|
||
// TestTidyDryRun_EmbeddingUnavailableIsHonest 是判据 ④：语义召回不可用时**如实
|
||
// 报错**（ErrTidyRecallUnavailable），绝不静默降级成「没有候选」；而生产轮在同样
|
||
// 条件下依旧按既有口径回退到词项余弦（试算的严格性只属于试算）。
|
||
func TestTidyDryRun_EmbeddingUnavailableIsHonest(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
store := newTestStore(t)
|
||
seedDryRunFixture(t, store)
|
||
tr := newTestTidier(t, store, tidyTestCfg(0.80),
|
||
failingEmbedder{err: errors.New("dial tcp 127.0.0.1:11434: connectex: 目标计算机积极拒绝")})
|
||
|
||
before := snapshotTidyTables(t, store)
|
||
out, err := tr.DryRun(ctx)
|
||
if !errors.Is(err, ErrTidyRecallUnavailable) {
|
||
t.Fatalf("DryRun err = %v, want ErrTidyRecallUnavailable", err)
|
||
}
|
||
if len(out.Candidates) != 0 {
|
||
t.Fatalf("召回不可用时不得产出候选：%+v", out.Candidates)
|
||
}
|
||
assertRowsUnchanged(t, "memory_tidy_candidates", before.candidates, snapshotTidyTables(t, store).candidates)
|
||
assertRowsUnchanged(t, "memory_tidy_runs", before.runs, snapshotTidyTables(t, store).runs)
|
||
|
||
// 生产轮（RunOnce）不受试算的严格性影响：它照旧降级并跑完一轮。
|
||
if _, err := tr.RunOnce(ctx); err != nil {
|
||
t.Fatalf("生产轮应当回退到词项余弦并可跑完：%v", err)
|
||
}
|
||
runs, err := store.ListTidyRuns(ctx, 10)
|
||
if err != nil {
|
||
t.Fatalf("ListTidyRuns: %v", err)
|
||
}
|
||
if len(runs) != 1 {
|
||
t.Fatalf("生产轮留痕 = %d 条, want 1（试算本身不写留痕）", len(runs))
|
||
}
|
||
}
|
||
|
||
// TestTidyDryRun_RecallModeMatchesEmbedder 钉住试算报出的召回来源与依赖一致。
|
||
func TestTidyDryRun_RecallModeMatchesEmbedder(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
store := newTestStore(t)
|
||
seedDryRunFixture(t, store)
|
||
|
||
withEmbedder := newTestTidier(t, store, tidyTestCfg(0.80), dryRunFixtureEmbedder())
|
||
out, err := withEmbedder.DryRun(ctx)
|
||
if err != nil {
|
||
t.Fatalf("DryRun(embedder): %v", err)
|
||
}
|
||
if out.RecallMode != tidyRecallEmbedding {
|
||
t.Fatalf("recallMode = %q, want %q", out.RecallMode, tidyRecallEmbedding)
|
||
}
|
||
|
||
withoutEmbedder := newTestTidier(t, store, tidyTestCfg(0.80), nil)
|
||
out, err = withoutEmbedder.DryRun(ctx)
|
||
if err != nil {
|
||
t.Fatalf("DryRun(lexical): %v", err)
|
||
}
|
||
if out.RecallMode != tidyRecallLexical {
|
||
t.Fatalf("recallMode = %q, want %q", out.RecallMode, tidyRecallLexical)
|
||
}
|
||
}
|
||
|
||
// TestTidyServiceDryRun_DisabledIsHonest 钉住「配置关闭 ⇒ 诚实错误」：关闭是显式
|
||
// 选择，试算不绕过它，也不回一份空候选假装「没有建议」。
|
||
func TestTidyServiceDryRun_DisabledIsHonest(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
store := newTestStore(t)
|
||
seedDryRunFixture(t, store)
|
||
|
||
cfg := tidyTestCfg(0.80)
|
||
cfg.Enabled = false
|
||
svc := newTidyServiceForTest(t, store, cfg, dryRunFixtureEmbedder(), time.Now, nil)
|
||
if _, err := svc.DryRun(ctx); !errors.Is(err, ErrTidyDisabled) {
|
||
t.Fatalf("DryRun err = %v, want ErrTidyDisabled", err)
|
||
}
|
||
runs, err := store.ListTidyRuns(ctx, 10)
|
||
if err != nil {
|
||
t.Fatalf("ListTidyRuns: %v", err)
|
||
}
|
||
if len(runs) != 0 {
|
||
t.Fatalf("关闭时不得留下任何留痕：%+v", runs)
|
||
}
|
||
}
|
||
|
||
// TestTidyServiceDryRun_EnabledWritesNothing 是服务层的零副作用回归：开启状态下
|
||
// 经服务入口试算同样一行都不写（连运行留痕都不写）。
|
||
func TestTidyServiceDryRun_EnabledWritesNothing(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
store := newTestStore(t)
|
||
seedDryRunFixture(t, store)
|
||
before := snapshotTidyTables(t, store)
|
||
|
||
svc := newTidyServiceForTest(t, store, tidyTestCfg(0.80), dryRunFixtureEmbedder(), time.Now, nil)
|
||
out, err := svc.DryRun(ctx)
|
||
if err != nil {
|
||
t.Fatalf("DryRun: %v", err)
|
||
}
|
||
if len(out.Candidates) == 0 {
|
||
t.Fatal("夹具应当产出候选")
|
||
}
|
||
after := snapshotTidyTables(t, store)
|
||
assertRowsUnchanged(t, "memory_items", before.items, after.items)
|
||
assertRowsUnchanged(t, "memory_tidy_candidates", before.candidates, after.candidates)
|
||
assertRowsUnchanged(t, "memory_tidy_runs", before.runs, after.runs)
|
||
}
|