Cyrene_v2/internal/runtime/memory/preinsert.go

package memory

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"sync"
	"time"

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

// 本文件是 D3「入库前审查」（design/23-记忆整理与去重 §15.1 D3）的服务形态：
// 新记忆落库**之后**，由**独立的记忆管理会话**（复用既有的 sess-memorytidy
// 装配，见 TidySessionID）异步判定它与同分区既有记忆的关系，四态结论见
// IngestDecision。
//
// 与 design §15 D3-1/3/5 的取舍（逐条，因为设计稿这一处同时出现「写入之前」
// 与「首选异步 + 结论回写」两种表述，实现必须把口径写死）：
//
//  1. **触发点在写入之后**（异步）：回合收尾的写入路径上多出一次 embedding
//     网络调用就会把主人的回合拖长（§15 D3-3「不许因为审查把对话回合拖长」），
//     而 §15 D3-3 要求的正是「审查结论**回写**到 memory_items」—— 只有行
//     已存在才谈得上回写。因此本实现是「先落库、再由审查把重复条目软合并
//     回去」，在观测上与「不新建」等价（活跃行数不变），且**不阻塞任何回合**；
//  2. **失败即安全**：条目先落库 ⇒ 审查失败/超时/embedder 不可用时「记忆绝不
//     丢」是结构性成立的，不靠回滚（§15 D3-5）。失败路径**只记 WARN、绝不
//     改写任何一行**；
//  3. **规则优先、模型兜底**：T0 是纯规则（design §15 D3-2 T0），**恒为零模型
//     调用**；T1（灰区调一次轻量模型，§15.1 D3-2）默认**开**，但只在「T0 判不出
//     结论且存在灰区候选」时才发出**至多一次**调用（preinsert_t1.go），并且
//     要有模型出口才成立 —— 两者任一缺席即是结构性零模型调用。
//     **同步入库闸（preinsert_gate.go）不碰 T1**，写入路径因此仍是零模型调用。
//
// 代价（如实登记）：极短窗口内库里会多一行待裁定的重复条目，随后被软合并
// （status='merged' + merged_into，可 revert，绝不物理删除，design §8）。

// IngestDecision 是入库前审查的四态结论（design/23 §15.1 D3）。
type IngestDecision string

const (
	// IngestSkip 表示已有**归一化后完全等价**的条目：新行折回既有行，
	// 不产生第二条活跃记忆（§15 D3-2 T0-1）。
	IngestSkip IngestDecision = "skip"
	// IngestMerge 表示新条目是既有条目事实集合的**子集**：并入既有，
	// 既有行存活（§15 D3-2 T0-2）。
	IngestMerge IngestDecision = "merge"
	// IngestModify 表示既有条目是新条目事实集合的子集：**新行承载更全的
	// 内容**，旧行置为被合并（§15 D3-6「改写必须是新行 + 旧行置为被合并」）。
	IngestModify IngestDecision = "modify"
	// IngestCreate 表示确属新条目（互不包含 / 被规则否决 / 没有召回）：
	// 正常新建，不做任何改写（§15 D3-2 T0-3）。
	IngestCreate IngestDecision = "create"
)

// IngestRuleVersion 是入库前审查规则集的版本号，进每条审查留痕（§15 D3-8）。
// **改判定规则必须同时升它**，否则新结论与旧留痕无法区分。
//
// 版本史：
//   - `d3.1` 四态规则层（T0）+ 同步入库闸；
//   - `d3.2` 加入 **T1 灰区模型判定**（preinsert_t1.go，design/23 §15.1 D3-2）：
//     灰区（相似、互不包含、未被规则否决）在模型出口接上时会被改判成
//     merge/modify，因此 `reason` 上出现了 `t1_*` 族。
const IngestRuleVersion = "d3.2"

// 调优旋钮（ReviewConfig）与它们的默认值常量在 preinsert_config.go。

var (
	// ErrReviewDisabled 在审查被配置显式关闭却又被直接调用时返回。
	ErrReviewDisabled = errors.New("memory/review: disabled by configuration")

	// ErrReviewRestrictedSource 在待审条目来自群聊/陌生人这类**只读**上下文时
	// 返回。它不是失败：这类条目凭什么都不该被审查（既不合并也不改写，
	// design §6 / §15 D3-7 + AGENTS §9 的作用域隔离红线）。
	ErrReviewRestrictedSource = errors.New("memory/review: restricted source is read-only")

	// ErrReviewItemNotActive 在该条目已经不在册（被前一次审查合并掉）却又被
	// 要求审查时返回。它是幂等保护：再判一次会把存活行的 merge_count 重复加一。
	ErrReviewItemNotActive = errors.New("memory/review: item is no longer active")
)

// ReviewDeps 是注入给入库前审查的依赖。Store 必需；其余可空并优雅降级。
type ReviewDeps struct {
	// Store 是记忆存储。必需。
	Store *Store
	// Embedder 是候选召回的相似度来源。为 nil 时退回零依赖的字符二元组
	// 余弦（与整理生产轮的回退同一口径，见 tidy_recall.go）。
	Embedder embedding.Embedder
	// Audit 是审计出口（可空；为 nil 时只记日志）。
	Audit func(who, what, result string)
	// Now 是墙钟来源，便于测试注入。
	Now func() time.Time
	// Log 是基础日志器，来源标记为 "memory/review"。
	Log *slog.Logger
	// GateCandidates 是**同步入库闸**的候选读取出口。为 nil 时用 store 的
	// 生产口径（Store.listReviewCandidates，分区过滤写在 SQL 里）。测试用它
	// 注入慢/坏的候选源，验证「预算有界」与「超时 fail-open」这两条。
	GateCandidates func(ctx context.Context, p tidyPartition, excludeID string, limit int) ([]MemoryItem, error)
	// Judge 是 **T1 灰区判定**的一次性模型调用口（design/23 §15.1 D3-2）。
	//
	// 为 nil 时 T1 **结构性关闭**（零模型调用）—— 这正是「T1 关闭 ⇒ 零模型调用」
	// 从承诺变成事实的那条线。它只取模型适配器的 Chat 一面（T1 不流式、不重试、
	// 不并发），注入的是**既有** model.Adapter 的调用口（组合根装配的就是模型栈
	// 本身，含按角色路由与主→备→本地降级），不新造任何 LLM 客户端。
	//
	// ⚠️ 它**只被异步审查层使用**（PreInsertReviewer.Review）；同步入库闸
	// （preinsert_gate.go）的判定链里根本没有它，写入路径因此零模型调用。
	Judge T1Judge
}

// IngestReview 是一条入库前审查留痕（design/23 §15 D3-8）。它**不含记忆正文**：
// 命中的既有 id 足以让 owner-only 的读侧回表取原文。
type IngestReview struct {
	ID        string
	ItemID    string
	PersonaID string
	Scope     Scope
	ProjectID string
	Decision  IngestDecision
	Reason    string
	// Veto 是命中的否决规则（`merge`/`modify`/`skip` 时为空）：它是结构化依据，
	// 面板据此聚合「为什么这一条没被合并」，而不是去解析 reason 字符串。
	Veto       TidyVetoReason
	TargetID   string
	MergedIDs  []string
	Score      float64
	RecallMode string
	Degraded   bool
	RuleVer    string
	SessionID  string
	CreatedAt  time.Time
}

// PreInsertReviewer 是入库前审查服务。用 NewPreInsertReviewer 构建，用 Start
// 拉起工作协程、Stop 停它；Review 是同步入口（测试与手动核查用）。
type PreInsertReviewer struct {
	cfg   ReviewConfig
	deps  ReviewDeps
	log   *slog.Logger
	now   func() time.Time
	queue chan MemoryItem

	mu      sync.Mutex
	running bool
	cancel  context.CancelFunc
	wg      sync.WaitGroup

	statsMu sync.Mutex
	stats   ReviewStats
}

// NewPreInsertReviewer 由 cfg 与 deps 构建审查服务。Store 为 nil 时返回
// ErrNilStore（与整理器同一口径：没有记忆存储就没有审查）。
func NewPreInsertReviewer(cfg ReviewConfig, deps ReviewDeps) (*PreInsertReviewer, error) {
	if deps.Store == nil {
		return nil, ErrNilStore
	}
	cfg = normalizeReviewConfig(cfg)
	now := deps.Now
	if now == nil {
		now = time.Now
	}
	return &PreInsertReviewer{
		cfg:   cfg,
		deps:  deps,
		log:   logger.WithSource(baseTidyLog(deps.Log), "memory/review"),
		now:   now,
		queue: make(chan MemoryItem, cfg.QueueSize),
	}, nil
}

// gateEnabled 报告**同步入库闸**是否生效（任务书规格 3 的开关）。
// 总开关关闭或闸被显式关掉时它都为 false，此时写入路径逐字回到「只插不审」。
func (r *PreInsertReviewer) gateEnabled() bool {
	return r != nil && r.cfg.Enabled && !r.cfg.DisableIngestGate
}

// gateScanLimit 返回同步闸实际使用的候选上限：配置值与其默认值的较小者 ——
// 写入路径上的扫描必须比异步队列更紧（任务书规格 1）。
func (r *PreInsertReviewer) gateScanLimit() int {
	limit := r.cfg.GateScanLimit
	if limit <= 0 {
		limit = defaultReviewGateScanLimit
	}
	if r.cfg.ScanLimit > 0 && r.cfg.ScanLimit < limit {
		return r.cfg.ScanLimit
	}
	return limit
}

// SessionID 返回审查运行的会话标识：**复用既有的 sess-memorytidy 装配**
// （design/23 §4.1 B；§15 D3 要求「走记忆管理会话」，不许造第二形态会话）。
func (r *PreInsertReviewer) SessionID() string { return TidySessionID }

// Enabled 报告审查是否被配置开启。
func (r *PreInsertReviewer) Enabled() bool { return r.cfg.Enabled }

// Start 拉起工作协程（审查队列 + 留痕保留期清扫）。幂等；配置关闭时只记一条日志
// 并立即返回（关闭是显式选择，绝不静默）。
func (r *PreInsertReviewer) Start(ctx context.Context) {
	if !r.cfg.Enabled {
		r.log.Info("memory ingest review disabled by configuration", "env", "MEMORY_REVIEW_ENABLED")
		return
	}
	if ctx == nil {
		ctx = context.Background()
	}
	r.mu.Lock()
	if r.running {
		r.mu.Unlock()
		return
	}
	r.running = true
	cctx, cancel := context.WithCancel(ctx)
	r.cancel = cancel
	// 两个协程（审查工作协程 + 留痕清扫循环）**在释放锁之前**一次性登记，
	// 避免 Stop 的 wg.Wait 与随后的 Add 并发（WaitGroup 的「计数归零后不许再 Add」
	// 是硬约束，不只是风格）。
	r.wg.Add(2)
	r.mu.Unlock()

	go r.worker(cctx)
	// 留痕清扫（design/23 §15 D3-8 的 memory_ingest_reviews）与审查工作协程同生共死：
	// 启动期扫一次、此后按间隔定时扫（口径见 preinsert_gc.go）。它删的是**已经过期
	// 的观测留痕**，与审查本身的判定/记忆行无关，因此失败只记 WARN、绝不退出。
	go r.sweepLoop(cctx)

	r.log.Info("memory ingest review started",
		"session_id", TidySessionID,
		"queue", r.cfg.QueueSize,
		"scan_limit", r.cfg.ScanLimit,
		"timeout", r.cfg.Timeout.String(),
		"recall_threshold", r.cfg.Threshold,
		"rule_version", IngestRuleVersion,
		"sync_gate", r.gateEnabled(),
		"gate_budget", r.cfg.GateBudget.String(),
		"gate_scan_limit", r.gateScanLimit(),
		// T1 是否有模型出口：`t1_judge=false` 有两种成因（配置关掉 / 没接模型），
		// 分开写让「以为开着其实没接」在启动日志里就看得见。
		"t1_judge", r.t1Enabled(),
		"t1_off_by_config", r.cfg.DisableT1,
		"t1_timeout", r.t1Timeout().String(),
		"t1_max_gray", defaultT1MaxGray,
		"sweep_interval", DefaultIngestReviewGCInterval.String(),
		"retention", DefaultIngestReviewRetention.String(),
		"max_reviews", DefaultIngestReviewMaxRows)
}

// Stop 停止工作协程并等待正在审的那一条跑完。幂等；ctx 在排空前被取消时返回
// 包装后的 context 错误，而不是泄漏 goroutine。
func (r *PreInsertReviewer) Stop(ctx context.Context) error {
	if ctx == nil {
		ctx = context.Background()
	}
	r.mu.Lock()
	if !r.running {
		r.mu.Unlock()
		return nil
	}
	r.running = false
	cancel := r.cancel
	r.mu.Unlock()

	cancel()
	done := make(chan struct{})
	go func() {
		r.wg.Wait()
		close(done)
	}()
	select {
	case <-done:
		r.log.Info("memory ingest review stopped", "stats", reviewStatsLabel(r.Stats()))
		return nil
	case <-ctx.Done():
		return fmt.Errorf("memory/review: stop: %w", ctx.Err())
	}
}

// Enqueue 把一条刚落库的记忆投入审查队列。它是**非阻塞**的：队列满时返回
// false（该条不做审查，记忆本身早已落库，绝不因此丢记忆）并记一条 WARN。
func (r *PreInsertReviewer) Enqueue(item MemoryItem) bool {
	if !r.cfg.Enabled {
		return false
	}
	select {
	case r.queue <- item:
		return true
	default:
		r.bump(func(s *ReviewStats) { s.Dropped++ })
		r.log.Warn("memory ingest review queue full, item left unreviewed",
			"item_id", item.ID, "scope", string(item.Scope), "queue", r.cfg.QueueSize)
		return false
	}
}

// Stats 返回累计计数（快照）。
func (r *PreInsertReviewer) Stats() ReviewStats {
	r.statsMu.Lock()
	defer r.statsMu.Unlock()
	return r.stats
}

// worker 是单读者循环：一条一条审（单写者的自然形态，design §9 的单写者口径）。
func (r *PreInsertReviewer) worker(ctx context.Context) {
	defer r.wg.Done()
	for {
		select {
		case <-ctx.Done():
			return
		case item := <-r.queue:
			r.process(ctx, item)
		}
	}
}

// process 审一条：单条超时由 r.cfg.Timeout 把守。失败**绝不改写任何一行**，
// 只如实记 WARN（design §15 D3-5）。
func (r *PreInsertReviewer) process(ctx context.Context, item MemoryItem) {
	rctx, cancel := context.WithTimeout(ctx, r.cfg.Timeout)
	defer cancel()
	if _, err := r.Review(rctx, item); err != nil {
		if errors.Is(err, ErrReviewRestrictedSource) || errors.Is(err, ErrReviewItemNotActive) {
			// 只读来源 / 已不在册：都不是失败，也都零写入。
			r.log.Debug("memory ingest review skipped", "item_id", item.ID, "reason", err)
			return
		}
		r.bump(func(s *ReviewStats) { s.Failed++ })
		r.log.Warn("memory ingest review failed, memory left untouched",
			"item_id", item.ID, "scope", string(item.Scope), "error", err)
	}
}

// Review 同步审一条并返回留痕。它在以下情形**不做任何改写**并返回错误：
// 召回不可用（超时/取消/存储错误）、改写失败。调用方据此记 WARN，
// 而记忆本身早已落库，因此「不丢记忆」是结构性成立的。
func (r *PreInsertReviewer) Review(ctx context.Context, item MemoryItem) (IngestReview, error) {
	if !r.cfg.Enabled {
		return IngestReview{}, ErrReviewDisabled
	}
	if nonOwnerSource(item.SourceConversation) {
		return IngestReview{}, ErrReviewRestrictedSource
	}
	// 幂等：已经不在册的行不再判（否则重复判会把存活行的 merge_count 数两次）。
	status, err := r.deps.Store.reviewItemStatus(ctx, item.ID)
	if err != nil {
		return IngestReview{}, fmt.Errorf("memory/review: item status: %w", err)
	}
	if status != string(tidyStatusActive) {
		return IngestReview{}, ErrReviewItemNotActive
	}
	now := r.now().UTC()
	cands, degraded, err := r.recallCandidates(ctx, item)
	if err != nil {
		return IngestReview{}, fmt.Errorf("memory/review: recall: %w", err)
	}
	if err := ctx.Err(); err != nil {
		// 超时/取消发生在召回之后：同样不改写（fail-safe，§15 D3-5）。
		return IngestReview{}, fmt.Errorf("memory/review: %w", err)
	}
	verdict := judgeIngest(item, cands)
	// T1（灰区模型判定，design/23 §15.1 D3-2）**只挂在异步审查层**：规则层判不出
	// 结论且有灰区候选时，用至多一次模型调用改判。未启用 / 无出口 / 失败一律
	// 原样回落 create —— 它没有「改写记忆」的代码路径，因此失败绝不可能丢记忆。
	verdict = r.maybeJudgeT1(ctx, item, verdict)
	merged, err := r.applyVerdict(ctx, verdict, item, now)
	if err != nil {
		return IngestReview{}, fmt.Errorf("memory/review: apply: %w", err)
	}
	rev := r.buildRecord(item, verdict, merged, degraded, now)
	if err := r.deps.Store.insertIngestReview(ctx, rev); err != nil {
		// 留痕失败**不改判决定**：改写已经生效，这里只如实记 WARN，绝不回滚
		// 一次已经生效的合并（回滚会把审查看成比结论更权威，那更危险）。
		r.log.Warn("memory ingest review record insert failed", "item_id", item.ID, "error", err)
	}
	r.auditWith(reviewAuditVerb, rev)
	r.recordDecision(rev)
	return rev, nil
}

// applyVerdict 落定判定：create 什么都不做；其余把「被吸收的那一行」软合并到
// 存活行（design §8：软删除 + merged_into + merged_from + merge_count，
// 绝不物理删除，绝不改任何一行的 content）。
func (r *PreInsertReviewer) applyVerdict(ctx context.Context, v ingestVerdict, item MemoryItem, now time.Time) ([]string, error) {
	if v.Decision == IngestCreate {
		return nil, nil
	}
	if v.Absorbed.ID == "" || v.Existing.ID == "" {
		return nil, fmt.Errorf("memory/review: %s without a matched existing item", v.Decision)
	}
	survivor := item
	if v.Absorbed.ID == item.ID {
		// merge / skip：被吸收的是新行，存活方是既有行。
		survivor = v.Existing
	}
	if err := r.deps.Store.absorbIngestItem(ctx, v.Absorbed, survivor, now); err != nil {
		return nil, err
	}
	return []string{v.Absorbed.ID}, nil
}

// buildRecord 组装留痕（design §15 D3-8：决定 / 依据 / 命中的既有 id /
// 是否降级 + 规则版本与会话标识）。
func (r *PreInsertReviewer) buildRecord(item MemoryItem, v ingestVerdict, merged []string, degraded bool, now time.Time) IngestReview {
	rev := IngestReview{
		ID:         mustNewTidyID(),
		ItemID:     item.ID,
		PersonaID:  item.PersonaID,
		Scope:      item.Scope,
		ProjectID:  item.ProjectID,
		Decision:   v.Decision,
		Reason:     v.Reason,
		Veto:       v.Veto,
		TargetID:   v.Existing.ID,
		MergedIDs:  merged,
		Score:      v.Score,
		RecallMode: v.Mode,
		Degraded:   degraded,
		RuleVer:    IngestRuleVersion,
		SessionID:  TidySessionID,
		CreatedAt:  now,
	}
	if rev.MergedIDs == nil {
		rev.MergedIDs = []string{}
	}
	return rev
}