Cyrene_v2/internal/runtime/memory/preinsert_async_test.go

package memory

import (
	"context"
	"errors"
	"strings"
	"testing"
	"time"

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

// 本文件是 D3「入库前审查」的**隔离与异步**测试（design/23 §15 D3-3/4/5/7/8）：
// 群聊/陌生人来源绝不触碰、跨分区绝不合并、写入方绝不等待审查、队列有界诚实、
// 开关关闭时行为不变、留痕如实登记降级。

// TestReviewRestrictedSourceIsNeverTouched 是 design/23 §6/§9 红线的钉死：
// 群聊/陌生人来源的条目**既不被审查、也不被合并**（只读不写）。
func TestReviewRestrictedSourceIsNeverTouched(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	store := newTestStore(t)
	base := tidyTestTime(t)
	groupItem := reviewAt(fixtureHabitShort, base, 0.9)
	groupItem.SourceConversation = FormatSourceConversation("sess-qq-group-1", "m-1")
	group := seedTidyItem(t, store, groupItem)
	owner := seedTidyItem(t, store, reviewAt(fixtureHabitShort, base.Add(time.Minute), 0.9))

	r := newTestReviewer(t, store, stubEmbedder{vecs: map[string]embedding.Vector{
		fixtureHabitShort: {1, 0},
	}}, 0.80)

	// ① 隔离来源的**新条目**：概不审查。
	if _, err := r.Review(ctx, group); !errors.Is(err, ErrReviewRestrictedSource) {
		t.Fatalf("隔离来源审查 err = %v, want %v", err, ErrReviewRestrictedSource)
	}
	if reviews, err := store.ListIngestReviews(ctx, 10); err != nil || len(reviews) != 0 {
		t.Fatalf("隔离来源不该留任何留痕：reviews=%+v err=%v", reviews, err)
	}
	if got := readReviewRow(t, store, group.ID); got.status != tidyStatusActive {
		t.Fatalf("隔离来源被改写了：%+v", got)
	}

	// ② 隔离来源作为**候选**：不得与主人条目合并（群体来源硬门）。
	rev, err := r.Review(ctx, owner)
	if err != nil {
		t.Fatalf("Review: %v", err)
	}
	if rev.Decision != IngestCreate || rev.Veto != VetoSource {
		t.Fatalf("verdict = %+v, want create + source 否决", rev)
	}
	if got := readReviewRow(t, store, owner.ID); got.status != tidyStatusActive {
		t.Fatalf("主人条目被跨来源合并了：%+v", got)
	}
	if n := countActiveRows(t, store, "persona.a"); n != 2 {
		t.Fatalf("活跃行数 = %d, want 2", n)
	}
}

// TestReviewNeverCrossesPersonaOrScope 是 design/23 §15 D3-9⑤ 与 §6 硬约束：
// 跨角色 / 跨作用域**绝不合并** —— 运行期连比较都不发生，判定层再兜一次否决。
func TestReviewNeverCrossesPersonaOrScope(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	store := newTestStore(t)
	base := tidyTestTime(t)
	other := habitItem(fixtureHabitShort, base, 0.9)
	other.PersonaID = "persona.b"
	otherItem := seedTidyItem(t, store, other)
	otherScope := MemoryItem{
		PersonaID: "persona.a", Scope: ScopeUser, Kind: KindRelationship,
		Content: fixtureHabitShort, Confidence: 0.9, CreatedAt: base,
		SourceConversation: FormatSourceConversation("web", "msg-1"),
	}
	seedTidyItem(t, store, otherScope)
	mine := seedTidyItem(t, store, reviewAt(fixtureHabitShort, base.Add(time.Minute), 0.9))

	r := newTestReviewer(t, store, stubEmbedder{vecs: map[string]embedding.Vector{
		fixtureHabitShort: {1, 0},
	}}, 0.80)
	rev, err := r.Review(ctx, mine)
	if err != nil {
		t.Fatalf("Review: %v", err)
	}
	if rev.Decision != IngestCreate {
		t.Fatalf("跨分区被判成 %q，分区之间永不比较（§6）", rev.Decision)
	}
	if got := readReviewRow(t, store, otherItem.ID); got.status != tidyStatusActive {
		t.Fatalf("别的角色的记忆被改写了：%+v", got)
	}

	// 判定层的纵深防御：直接喂一条跨角色候选，必须被硬门否决。
	verdict := judgeIngest(mine, []ingestCandidate{{item: otherItem, score: 0.99, mode: tidyRecallEmbedding}})
	if verdict.Decision != IngestCreate || verdict.Veto != VetoPersona {
		t.Fatalf("跨角色判定 = %+v, want create + persona 否决", verdict)
	}
}

// TestReviewIsNonBlockingAndAsyncPathWorks 是 design/23 §15 D3-3/D3-4：
// 写入方**绝不等待**异步审查；审查在后台（工作协程）按队列跑完。
//
// ⚠️ 2026-10-07 起 D3 是**两层**：写路径上有同步入库闸（preinsert_gate.go，
// 重复当场落 merged），闸没拦下的才进这条异步队列。因此本用例**显式关闸**
// （DisableIngestGate）来单独验异步那一层 —— 否则它测的会是闸而不是队列。
func TestReviewIsNonBlockingAndAsyncPathWorks(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	store := newTestStore(t)
	base := tidyTestTime(t)
	existing := seedTidyItem(t, store, reviewAt(fixtureHabitShort, base, 0.9))

	asyncOnly, err := NewPreInsertReviewer(ReviewConfig{
		Enabled: true, QueueSize: 8, ScanLimit: 50, Timeout: 5 * time.Second,
		Threshold: 0.80, DisableIngestGate: true,
	}, ReviewDeps{
		Store: store,
		Embedder: stubEmbedder{vecs: map[string]embedding.Vector{
			fixtureHabitShort: {1, 0},
		}},
		Now: func() time.Time { return tidyTestTime(t) },
		Log: testLogger(),
	})
	if err != nil {
		t.Fatalf("NewPreInsertReviewer: %v", err)
	}
	r := asyncOnly
	store.AttachIngestReviewer(r)
	// 未 Start：写入方投递后**立即返回**，审查尚未发生（异步性证据）。
	start := time.Now()
	dup := seedTidyItem(t, store, reviewAt(fixtureHabitShort, base.Add(time.Minute), 0.9))
	if elapsed := time.Since(start); elapsed > 2*time.Second {
		t.Fatalf("写入等了 %s，审查绝不能阻塞写入方", elapsed)
	}
	if got := readReviewRow(t, store, dup.ID); got.status != tidyStatusActive {
		t.Fatalf("关闸后审查仍在写入方返回前生效了：%+v", got)
	}

	r.Start(ctx)
	t.Cleanup(func() {
		stopCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = r.Stop(stopCtx)
	})
	waitFor(t, func() bool {
		return readReviewRow(t, store, dup.ID).status == ingestStatusMerged
	}, "后台审查把重复行折回既有行")
	if got := readReviewRow(t, store, existing.ID); got.mergeCount != 2 {
		t.Fatalf("survivor = %+v, want merge_count=2", got)
	}
}

// waitFor 轮询等待条件成立（后台工作协程的完成信号），超时即失败。
func waitFor(t *testing.T, cond func() bool, what string) {
	t.Helper()
	deadline := time.Now().Add(3 * time.Second)
	for time.Now().Before(deadline) {
		if cond() {
			return
		}
		time.Sleep(5 * time.Millisecond)
	}
	t.Fatalf("等待超时：%s", what)
}

// TestReviewQueueIsBoundedAndDropsHonestly 是 design/23 §15 D3-4：审查有界，
// 队列满时**有界丢弃**（绝不阻塞写入方），且次数被如实计数。
func TestReviewQueueIsBoundedAndDropsHonestly(t *testing.T) {
	t.Parallel()
	store := newTestStore(t)
	r, err := NewPreInsertReviewer(ReviewConfig{
		Enabled: true, QueueSize: 2, ScanLimit: 10, Timeout: time.Second, Threshold: 0.8,
	}, ReviewDeps{Store: store, Log: testLogger()})
	if err != nil {
		t.Fatalf("NewPreInsertReviewer: %v", err)
	}
	item := reviewAt(fixtureHabitShort, tidyTestTime(t), 0.9)
	item.ID = "item-1"
	accepted := 0
	for i := 0; i < 5; i++ {
		if r.Enqueue(item) {
			accepted++
		}
	}
	if accepted != 2 {
		t.Fatalf("入队 = %d, want 2（队列上限）", accepted)
	}
	if got := r.Stats().Dropped; got != 3 {
		t.Fatalf("丢弃计数 = %d, want 3（绝不静默）", got)
	}
}

// TestReviewIsIdempotentPerItem 是幂等口径：同一条记忆被审第二次时**不再判定**
// （它已不在册），因此存活行的 merge_count 不会被同一条事实数两次。
func TestReviewIsIdempotentPerItem(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	store := newTestStore(t)
	base := tidyTestTime(t)
	existing := seedTidyItem(t, store, reviewAt(fixtureHabitShort, base, 0.9))
	dup := seedTidyItem(t, store, reviewAt(fixtureHabitShort, base.Add(time.Minute), 0.9))

	r := newTestReviewer(t, store, stubEmbedder{vecs: map[string]embedding.Vector{
		fixtureHabitShort: {1, 0},
	}}, 0.80)
	if _, err := r.Review(ctx, dup); err != nil {
		t.Fatalf("第一次 Review: %v", err)
	}
	if _, err := r.Review(ctx, dup); !errors.Is(err, ErrReviewItemNotActive) {
		t.Fatalf("第二次 Review err = %v, want %v", err, ErrReviewItemNotActive)
	}
	if got := readReviewRow(t, store, existing.ID); got.mergeCount != 2 {
		t.Fatalf("存活行 merge_count = %d, want 2（同一条事实只该数一次）", got.mergeCount)
	}
	if reviews, err := store.ListIngestReviews(ctx, 10); err != nil || len(reviews) != 1 {
		t.Fatalf("审查留痕 = %+v（err=%v）, want 恰为一条", reviews, err)
	}
}

// TestReviewDisabledChangesNothing 是 design/23 §15 D3-7 的开关口径：
// 关闭时审查一次性拒绝，且**不接线**的 store 写入路径逐字回到今天。
func TestReviewDisabledChangesNothing(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	store := newTestStore(t)
	base := tidyTestTime(t)
	seedTidyItem(t, store, reviewAt(fixtureHabitShort, base, 0.9))

	off, err := NewPreInsertReviewer(ReviewConfig{Enabled: false}, ReviewDeps{Store: store, Log: testLogger()})
	if err != nil {
		t.Fatalf("NewPreInsertReviewer: %v", err)
	}
	off.Start(ctx) // 关闭时只记一条日志并立即返回，不起任何协程
	store.AttachIngestReviewer(off)
	if off.Enqueue(reviewAt(fixtureHabitShort, base, 0.9)) {
		t.Fatal("关闭时不该接受审查任务")
	}
	dup := seedTidyItem(t, store, reviewAt(fixtureHabitShort, base.Add(time.Minute), 0.9))
	if _, err := off.Review(ctx, dup); !errors.Is(err, ErrReviewDisabled) {
		t.Fatalf("关闭时 Review err = %v, want %v", err, ErrReviewDisabled)
	}
	if got := off.Stats(); got.Dropped != 0 {
		t.Fatalf("关闭时不该把条目记成「被丢弃」：%+v", got)
	}
	// 未接线的 store：连审查留痕都不该有，两条记忆都照旧在册。
	if reviews, err := store.ListIngestReviews(ctx, 10); err != nil || len(reviews) != 0 {
		t.Fatalf("关闭时不该有留痕：%+v", reviews)
	}
	if n := countActiveRows(t, store, "persona.a"); n != 2 {
		t.Fatalf("关闭时行为被改动了：活跃行数 = %d, want 2", n)
	}
}

// TestReviewReusesTidySessionAndStaysZeroLLM 是 design/23 §15 D3 的装配口径：
// 复用既有 sess-memorytidy 装配（不造第二形态会话），且 T1 默认关 + 本切片
// 未实现 ⇒ 灰区（相似但不包含）**不调任何模型**，只按 T0 回落 create。
func TestReviewReusesTidySessionAndStaysZeroLLM(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	store := newTestStore(t)
	base := tidyTestTime(t)
	seedTidyItem(t, store, reviewAt(fixtureEnergyRedBull, base, 0.95))
	gray := seedTidyItem(t, store, reviewAt(fixtureEnergyIced, base.Add(time.Minute), 0.95))

	r := newTestReviewer(t, store, stubEmbedder{vecs: map[string]embedding.Vector{
		fixtureEnergyRedBull: {1, 0},
		fixtureEnergyIced:    simVector(0.88),
	}}, 0.80)
	if r.SessionID() != TidySessionID || !strings.HasPrefix(r.SessionID(), "sess-") {
		t.Fatalf("审查会话 = %q, want 复用 %q", r.SessionID(), TidySessionID)
	}
	rev, err := r.Review(ctx, gray)
	if err != nil {
		t.Fatalf("Review: %v", err)
	}
	if rev.Decision != IngestCreate {
		t.Fatalf("灰区被判成 %q，T1 关闭时必须保守回落 create", rev.Decision)
	}
	if rev.RecallMode != tidyRecallEmbedding {
		t.Fatalf("召回来源 = %q, want %q", rev.RecallMode, tidyRecallEmbedding)
	}
}

// TestReviewDegradedRecallIsRecordedHonestly 是 design/23 §15 D3-8 的「是否降级」：
// embedder 报错时降级到词项余弦，判定照旧，但留痕必须如实写 degraded。
func TestReviewDegradedRecallIsRecordedHonestly(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	store := newTestStore(t)
	base := tidyTestTime(t)
	seedTidyItem(t, store, reviewAt(fixtureHabitShort, base, 0.9))
	fresh := seedTidyItem(t, store, reviewAt(fixtureHabitLong, base.Add(time.Minute), 0.9))

	r := newTestReviewer(t, store, failingReviewEmbedder{}, 0.80)
	rev, err := r.Review(ctx, fresh)
	if err != nil {
		t.Fatalf("Review: %v", err)
	}
	if !rev.Degraded || rev.RecallMode != tidyRecallLexical {
		t.Fatalf("降级未被如实登记：%+v", rev)
	}
	if got := readReviewRow(t, store, fresh.ID); got.status != tidyStatusActive {
		t.Fatalf("降级不该丢掉新条目：%+v", got)
	}
}

// failingReviewEmbedder 模拟「embedder 挂」：Embed 恒报错（design §15 D3-5 的情形之一）。
type failingReviewEmbedder struct{}

func (failingReviewEmbedder) Embed(context.Context, []string) ([]embedding.Vector, error) {
	return nil, errors.New("embedder down")
}

func (failingReviewEmbedder) Dim() int { return 2 }