Cyrene_v2/internal/runtime/memory/embed_incremental_test.go

package memory

import (
	"context"
	"testing"
)

// 本文件断言两件事的**请求数**（M61，2026-10-30）：
//   - 增量：索引里已有向量的行不再重嵌 —— 同一分区第二轮整理轮/审查是 **0** 个请求；
//   - 韧性：一条坏分片只损失它自己，整体性故障最多 3 个请求就收手。
//
// 计数来自假服务端（embed_fake_server_test.go），被测对象是生产路径
// （embedding.NewAPI + vector.New + Store/Tidier/PreInsertReviewer），不是替身。

// batchRequests 是「按批上限发出去要几个请求」的算式（⌈n/batch⌉）。
func batchRequests(n, batch int) int {
	if n <= 0 || batch <= 0 {
		return 0
	}
	return (n + batch - 1) / batch
}

// TestEmbedIncremental_SecondRoundMakesNoRequests 是「增量后不重嵌」的钉死：
// 第一轮把 24 条存量嵌入并**回填索引**（3 个请求 = ⌈24/8⌉），第二轮**一次都不发**。
//
// 删掉复用实现（或把 embedIncremental 改回「每次全量嵌」）这一条立刻打红：
// 第二轮会重新发出同样的 3 个请求。
func TestEmbedIncremental_SecondRoundMakesNoRequests(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	store := newTestStore(t)
	items := reviewItems(t, store, 24)

	srv := newFakeEmbedServer(t)
	emb := srv.embedder()
	// vector.New 会探测一次 embedder 维度（1 个请求），因此基线在它之后取。
	store.attachVectorIndex(fakeIndex(t, emb))

	base := srv.count()
	first, stats1, err := store.embedIncremental(ctx, emb, items, tidyEmbedBatch)
	if err != nil {
		t.Fatalf("embedIncremental(first): %v", err)
	}
	firstReqs := srv.count() - base
	if want := batchRequests(len(items), tidyEmbedBatch); firstReqs != want {
		t.Fatalf("第一轮请求数 = %d, want %d（⌈%d/%d⌉）", firstReqs, want, len(items), tidyEmbedBatch)
	}
	if stats1.Reused != 0 || stats1.Sent != len(items) || stats1.Filled != len(items) || stats1.Dropped != 0 {
		t.Fatalf("第一轮 stats = %+v, want reused=0 sent=%d filled=%d dropped=0", stats1, len(items), len(items))
	}

	base = srv.count()
	second, stats2, err := store.embedIncremental(ctx, emb, items, tidyEmbedBatch)
	if err != nil {
		t.Fatalf("embedIncremental(second): %v", err)
	}
	secondReqs := srv.count() - base
	if secondReqs != 0 {
		t.Fatalf("第二轮请求数 = %d, want 0（已有向量的行不该重嵌）", secondReqs)
	}
	if stats2.Reused != len(items) || stats2.Sent != 0 || stats2.Filled != 0 {
		t.Fatalf("第二轮 stats = %+v, want reused=%d sent=0 filled=0", stats2, len(items))
	}
	for i := range items {
		if len(second[i]) == 0 {
			t.Fatalf("第 %d 条复用失败：向量为空", i)
		}
		if len(second[i]) != len(first[i]) {
			t.Fatalf("第 %d 条维度变了：%d -> %d", i, len(first[i]), len(second[i]))
		}
		for k := range first[i] {
			if second[i][k] != first[i][k] {
				t.Fatalf("第 %d 条复用到了别的向量（分量 %d）", i, k)
			}
		}
	}
	t.Logf("整理/审查同量记忆的请求数：修复前每轮 %d → 修复后第一轮 %d、第二轮 %d（%d 条，批上限 %d）",
		batchRequests(len(items), tidyEmbedBatch), firstReqs, secondReqs, len(items), tidyEmbedBatch)
}

// TestTidyRecall_SecondRoundZeroRequests 是整理轮的生产路径断言：同一分区连续两轮
// recallPairs，第一轮按批上限发请求，第二轮**零请求**（候选召回因此不再每轮重嵌）。
func TestTidyRecall_SecondRoundZeroRequests(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	store := newTestStore(t)
	reviewItems(t, store, 12)

	srv := newFakeEmbedServer(t)
	emb := srv.embedder()
	store.attachVectorIndex(fakeIndex(t, emb))

	tidier, err := NewTidier(TidyConfig{}, TidyDeps{Store: store, Embedder: emb, Log: testLogger()})
	if err != nil {
		t.Fatalf("NewTidier: %v", err)
	}
	partition := tidyPartition{personaID: "persona.a", scope: ScopePersona}
	items, err := store.listTidyPartitionItems(ctx, partition, 500)
	if err != nil {
		t.Fatalf("listTidyPartitionItems: %v", err)
	}
	if len(items) != 12 {
		t.Fatalf("分区条目数 = %d, want 12", len(items))
	}

	base := srv.count()
	ledger := &recallLedger{}
	if _, err := tidier.recallPairs(ctx, partition, items, &TidyStats{}, tidyBudget{}, ledger); err != nil {
		t.Fatalf("recallPairs(round 1): %v", err)
	}
	firstReqs := srv.count() - base
	want := batchRequests(len(items), tidyEmbedBatch)
	if firstReqs != want {
		t.Fatalf("第一轮整理召回请求数 = %d, want %d", firstReqs, want)
	}
	if ledger.embedding != 1 || ledger.lexical != 0 {
		t.Fatalf("ledger = %+v, want 一次 embedding 召回、零次降级", ledger)
	}

	base = srv.count()
	if _, err := tidier.recallPairs(ctx, partition, items, &TidyStats{}, tidyBudget{}, &recallLedger{}); err != nil {
		t.Fatalf("recallPairs(round 2): %v", err)
	}
	secondReqs := srv.count() - base
	if secondReqs != 0 {
		t.Fatalf("第二轮整理召回请求数 = %d, want 0（增量向量化没生效）", secondReqs)
	}
	t.Logf("整理轮 recallPairs 请求数：第一轮 %d → 第二轮 %d（%d 条）", firstReqs, secondReqs, len(items))
}

// TestReviewRecall_SecondReviewZeroRequests 是入库审查的生产路径断言：新条由写入
// 侧落库时已建好向量，因此**第二次审查同一分区是零请求**（候选都从索引取）。
//
// 这把 M15 那条「每条审查 26 个请求」压到「第一条审查 ⌈K/批⌉、之后 0」。
func TestReviewRecall_SecondReviewZeroRequests(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	store := newTestStore(t)
	reviewItems(t, store, 20)

	srv := newFakeEmbedServer(t)
	emb := srv.embedder()
	store.attachVectorIndex(fakeIndex(t, emb))
	reviewer := newTestReviewer(t, store, emb, 0.80)

	first := addItem(t, store, MemoryItem{
		PersonaID: "persona.a", Scope: ScopePersona, Kind: KindRelationship,
		Content: "新写入记忆 第一条",
	})
	base := srv.count()
	if _, degraded, err := reviewer.recallCandidates(ctx, first); err != nil || degraded {
		t.Fatalf("recallCandidates(first) = (degraded=%v, err=%v), want (false, nil)", degraded, err)
	}
	firstReqs := srv.count() - base
	if want := batchRequests(20, tidyEmbedBatch); firstReqs != want {
		t.Fatalf("第一条审查的请求数 = %d, want %d（只有 20 条既有候选缺向量）", firstReqs, want)
	}

	second := addItem(t, store, MemoryItem{
		PersonaID: "persona.a", Scope: ScopePersona, Kind: KindRelationship,
		Content: "新写入记忆 第二条",
	})
	base = srv.count()
	if _, degraded, err := reviewer.recallCandidates(ctx, second); err != nil || degraded {
		t.Fatalf("recallCandidates(second) = (degraded=%v, err=%v), want (false, nil)", degraded, err)
	}
	secondReqs := srv.count() - base
	if secondReqs != 0 {
		t.Fatalf("同分区第二次审查的请求数 = %d, want 0（增量为空 ⇒ 一个请求都不该发）", secondReqs)
	}
	t.Logf("入库审查请求数：修复前 26（201÷8）→ 第一条 %d、第二条 %d", firstReqs, secondReqs)
}