Cyrene_v2/internal/runtime/memory/store.go

// memory 包实现 Cyrene v2 记忆子系统，规格见
// DevDocs/design/03-记忆.md。
//
// 它对外暴露三层：
//
//   - Store：基于 SQLite 的长期记忆存储。它按角色隔离持久化记忆条目
//     （design §0 layer ②、§4），经 AddScoped/ListScope/ListProject/
//     DeleteScoped 施加三作用域访问规则（design §3），并提供
//     Add/Get/ListByPersona/Delete，外加一条手动的
//     「记住」路径（design §1）。每条条目可携带 project_id，链到
//     项目索引层级。
//   - WorkingMemory：每会话的消息滑动窗口加一份摘要
//     （design §0 layer ①）。
//   - Retriever：相关性（关键词 + 可选语义）× 衰减检索，带
//     溯源与上下文注入（design §2）。它可选接收
//     vector.Index 作为持久化语义路径，以及 projectindex.Store，使
//     项目作用域的命中带出其链接的项目条目。
//
// 提取（design §1）经 Extractor 接口暴露，有两个
// 实现：ManualExtractor（规则式，记录一条显式的「记住 X」
// 指令）与 LLMExtractor（驱动 model.Adapter 对
// 一轮对话去噪去重，经 Extract 同步运行或经 ExtractAsync 异步运行）。
//
// M4 里程碑在此实现：schema 携带 decay_weight 与
// project_id，检索器接入可选的 embedding.Embedder / vector.Index
// 与项目索引存储，存储层施加 user/persona/project
// 作用域规则。除既有包外不需要任何外部依赖。
package memory

import (
	"context"
	"crypto/rand"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"strings"
	"sync"
	"time"

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

// Scope 分类一条长期记忆归属于谁。它与 Kind 正交，
// 两者都存于每条条目上（design §3 注）。
type Scope string

const (
	// ScopeUser 是关于用户习惯的记忆，跨角色共享。
	ScopeUser Scope = "user"
	// ScopePersona 是角色的日常记忆；其他角色只读。
	ScopePersona Scope = "persona"
	// ScopeProject 是项目作用域记忆，人人可读可写。
	ScopeProject Scope = "project"
)

// valid 报告 s 是否为已定义的作用域之一。
func (s Scope) valid() bool {
	switch s {
	case ScopeUser, ScopePersona, ScopeProject:
		return true
	default:
		return false
	}
}

// Kind 分类一条长期记忆条目的内容类别。
type Kind string

const (
	// KindRelationship 是关于「你和我」的记忆：偏好、习惯、
	// 约定（design §0「关系记忆」）。
	KindRelationship Kind = "relationship"
	// KindWorldEvent 是关于客观事实与世界事件的记忆
	// （design §0「世界/事件记忆」）。
	KindWorldEvent Kind = "world_event"
)

// valid 报告 k 是否为已定义的 kind 之一。
func (k Kind) valid() bool {
	switch k {
	case KindRelationship, KindWorldEvent:
		return true
	default:
		return false
	}
}

// MemoryItem 是一条长期记忆记录。零值不是合法
// 条目：PersonaID 与 Content 必填，Scope 与 Kind 由 Add 规范化到
// 其默认值。
type MemoryItem struct {
	ID                 string
	PersonaID          string
	RoleID             string
	ProjectID          string
	Scope              Scope
	Kind               Kind
	Content            string
	Tags               []string
	Confidence         float64
	SourceConversation string
	CreatedAt          time.Time
	UpdatedAt          time.Time
	DecayWeight        float64

	// ExpiresAt 是**临时记忆**的有效期末（design/03 §9.6、design/23 §18.1）。
	// 零值 = 不设过期，落库为空串，四条列表读路径恒可见；非零值落库为定宽 UTC
	// 文本，到期后该行即从读路径消失，直到后台清扫物理删除（design/23 §19）。
	//
	// ⚠️ **本字段目前只写不读**：Add 会把它写进 expires_at 列，但读路径一律不
	// SELECT 该列（闸是纯 SQL，见 timegate.go），故 Get/List* 回来的条目
	// ExpiresAt 恒为零值。要读回时效需同步 13 列 SELECT 与 scanItem，并连带
	// tidy/preinsert 的内部召回 SQL —— 那是改扫描口径，不在本切片授权范围。
	ExpiresAt time.Time
}

// 区分 memory 包各类失败的哨兵错误。每个都在
// 调用点用 %w 包裹，调用方可用 errors.Is 判断。
var (
	// ErrMissingPersona 在条目没有 PersonaID 时返回。
	ErrMissingPersona = errors.New("memory: persona_id is required")

	// ErrMissingContent 在条目没有内容时返回。
	ErrMissingContent = errors.New("memory: content is required")

	// ErrInvalidScope 在条目的 Scope 不是已定义作用域时返回。
	ErrInvalidScope = errors.New("memory: invalid scope")

	// ErrInvalidKind 在条目的 Kind 不是已定义 kind 时返回。
	ErrInvalidKind = errors.New("memory: invalid kind")

	// ErrNotFound 在条目 id 不存在时返回。
	ErrNotFound = errors.New("memory: item not found")

	// ErrNilDB 在 NewStore 收到 nil *sql.DB 时返回。
	ErrNilDB = errors.New("memory: nil db")
)

// schema 默认值与时间窗口常量。
const (
	defaultConfidence  = 1.0
	defaultDecayWeight = 1.0
)

// Store 是基于 *sql.DB 的长期记忆存储。
//
// 存储自己拥有 schema：NewStore 幂等地创建 memory_items 表及其
// 索引。并发交由 *sql.DB，按约定它可安全
// 并发使用，因此除语义索引的接入外，Store 不再持锁。每个查询都
// 参数化（?）并显式列出列（无 SELECT *），读取带
// context.Cancel。
type Store struct {
	db  *sql.DB
	log *slog.Logger

	// vecMu 保护 vec，使持久语义索引能在装配后并发接入，而写入
	// 路径每写一条记忆就 embed + add 到它（见 indexItem）。vec 为
	// nil 表示没有配置持久索引，检索据此回退到内存余弦。
	vecMu sync.RWMutex
	vec   vector.Index

	// projMu 保护 proj，使项目索引存储能在装配后并发接入，而写入路径
	// 每写一条项目作用域记忆就确保它链接的项目索引条目存在（见
	// ensureProjectIndex）。proj 为 nil 表示未接入项目索引，此时写入
	// 不触碰项目索引（design §3 项目记忆链接项目索引）。
	projMu sync.RWMutex
	proj   *projectindex.Store

	// reviewMu 保护 review，使入库前审查（design/23 §15.1 D3）能在装配后
	// 并发接入：接入后每条经 Add 落库的条目都会**先过同步入库闸**（确定性判据
	// 命中即落 merged，见 preinsert_gate.go），闸没拦下的再投入异步审查队列，
	// 由独立的记忆管理会话判定 merge/modify/create/skip。review 为 nil 表示
	// 未启用或未接线，此时写入路径行为逐字与接入前一致。
	reviewMu sync.RWMutex
	review   *PreInsertReviewer
}

// NewStore 在 db 上构建 Store，确保 memory_items schema 存在。
//
// db 必须是存活且已迁移的连接（迁移框架本身由
// internal/db 拥有）。nil db 返回 ErrNilDB。log 可为 nil，此时
// 使用丢弃型 logger。ctx 控制 schema 创建，调用方必须
// 遵守。任何旧的变宽时间戳都就地
// 规范化（见 normalizeMemoryTimestamps），使文本顺序与时间顺序一致。
func NewStore(ctx context.Context, db *sql.DB, log *slog.Logger) (*Store, error) {
	if db == nil {
		return nil, ErrNilDB
	}
	log = logger.WithSource(discardLogger(log), "memory")
	for _, ddl := range []string{
		createMemoryItemsTable,
		createMemoryItemsPersonaIndex,
		createMemoryItemsPersonaCreatedIndex,
	} {
		if _, err := db.ExecContext(ctx, ddl); err != nil {
			return nil, fmt.Errorf("memory: apply schema: %w", err)
		}
	}
	if err := ensureProjectIDColumn(ctx, db); err != nil {
		return nil, err
	}
	// 记忆整理（design/23 §10）的列与两张新表：同样在这一处幂等前滚，
	// 使「在任意 *sql.DB 上都能整理」与 memory_items 自身口径一致。
	if err := ensureTidySchema(ctx, db); err != nil {
		return nil, err
	}
	// 入库前审查（design/23 §15.1 D3）的留痕表：同一口径 —— 表在这里幂等
	// 创建，因此审查能力在任何 *sql.DB 上都可用；未接线审查时它只是一张空表。
	if err := ensureIngestReviewSchema(ctx, db); err != nil {
		return nil, err
	}
	if err := normalizeMemoryTimestamps(ctx, db); err != nil {
		return nil, err
	}
	return &Store{db: db, log: log}, nil
}

// Add 持久化 item，分配 id 与缺失的时间戳，并把
// Scope/Kind/Confidence/DecayWeight 规范化到默认值。它返回已落库的
// 条目，id 与时间戳均已填充。
//
// 🔴 **同步入库闸**（主人 2026-10-07 裁定）：本函数是全仓**唯一**写路径，因此
// 闸就摆在这里 —— 落库**之前**先在同分区内跑一次确定性判定（复用
// judgeIngest → tidy_rules.go / tidy_containment.go），判出「真重复」时这一行
// 直接落 status='merged' + merged_into=<存活行>（**照常插入、一个字不丢、可
// revert、从不进 active 集合**），于是 active 集合当场就是干净的。闸的总墙钟
// 预算显式且有界（见 defaultReviewGateBudget），超预算/候选源不可用一律
// fail-open（照常插入 + 留痕，**绝不静默丢记忆**）。闸未接线或被开关关掉时，
// 这段逻辑是零成本的 passthrough，写入路径逐字回到改动之前。
//
// 接入持久语义索引（见 attachVectorIndex）时，落库后同一条目也会
// 被向量化并写入索引，使索引与 memory_items 同步 —— 语义检索路径因此
// 真的有向量可查，而不是对一个恒空索引搜索（design/03 §3「真正启用
// 向量检索」、design/04 §7）。索引写入失败只记日志：条目已经在
// 关系表里，语义增强的失败绝不使记忆写入失败，检索侧也会回退到
// 内存余弦（见 Retriever.semanticScores）。被闸判为重复的行**不建索引**：
// 它不进召回、也不该占掉语义检索 top-K 的一个位置。
//
// `item.ExpiresAt`（临时记忆的时效，design/03 §9.6）同在这一处落库为
// `expires_at` 列：Add 是全仓唯一 INSERT 路径（design/23 §15.1 D3-1），
// 因此时效列的真生产者**只能**在这里 —— 读侧时效闸（timegate.go）比对的
// 就是这一列，若写入侧不写出它，闸就退化成恒真的空壳。
func (s *Store) Add(ctx context.Context, item MemoryItem) (MemoryItem, error) {
	if err := item.normalize(); err != nil {
		return MemoryItem{}, err
	}
	var err error
	if item, err = applyTimestamps(item); err != nil {
		return MemoryItem{}, err
	}
	tagsJSON, err := json.Marshal(item.Tags)
	if err != nil {
		return MemoryItem{}, fmt.Errorf("memory: encode tags: %w", err)
	}

	plan := s.planIngest(ctx, item)
	_, err = s.db.ExecContext(ctx, `INSERT INTO memory_items
		(id, persona_id, role_id, project_id, scope, kind, content, tags, confidence,
		 source_conversation, created_at, updated_at, decay_weight, status, merged_into,
		 expires_at)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		item.ID, item.PersonaID, item.RoleID, item.ProjectID, string(item.Scope), string(item.Kind),
		item.Content, string(tagsJSON), item.Confidence, item.SourceConversation,
		formatTime(item.CreatedAt), formatTime(item.UpdatedAt), item.DecayWeight,
		plan.insertStatus(), plan.absorbedID(), expiryText(item.ExpiresAt))
	if err != nil {
		return MemoryItem{}, fmt.Errorf("memory: insert item: %w", err)
	}
	s.finishIngest(ctx, item, plan)
	if plan.Action != ingestPlanAbsorbNew {
		s.indexItem(ctx, item)
		s.ensureProjectIndex(ctx, item)
	}
	if r := s.ingestReviewer(); r != nil {
		r.recordGate(ctx, item, plan, time.Now().UTC())
	}
	// 入库前审查（design/23 §15.1 D3）的**异步**那一层：闸没拦下的（真新 /
	// 拿不准 / 闸关闭）照旧投入审查队列，由后台用语义召回再判一次；闸挡下的
	// 那条**绝不再投**（同一件事不该有两条路径）。投递是非阻塞的。未接线审查
	// 时是空操作。
	if plan.Async {
		s.enqueueReview(item)
	}
	return item, nil
}

// Get 返回给定 id 的条目。id 为空或
// 不存在时返回 ErrNotFound。
//
// 🔴 **刻意不过任何读侧闸**（status 与时效都不过；design/03 §11.2 第 3 条）：
// 撤销软合并必须先读得到被吸收的行，人工核查 / 清算也要先读得到已过期与
// 已失效的行 —— 加闸等于把审计与撤销的抓手一并抹掉。要「面向用户的可见集」
// 请走 ListByPersona / ListScope / ListProject / List 或 Retriever（那些都过闸）。
func (s *Store) Get(ctx context.Context, id string) (MemoryItem, error) {
	if strings.TrimSpace(id) == "" {
		return MemoryItem{}, ErrNotFound
	}
	row := s.db.QueryRowContext(ctx, `SELECT id, persona_id, role_id, project_id, scope, kind,
		content, tags, confidence, source_conversation, created_at, updated_at,
		decay_weight FROM memory_items WHERE id = ?`, id)
	item, err := scanItem(row)
	if errors.Is(err, sql.ErrNoRows) {
		return MemoryItem{}, ErrNotFound
	}
	if err != nil {
		return MemoryItem{}, fmt.Errorf("memory: get item: %w", err)
	}
	return item, nil
}

// ListByPersona 返回 personaID 的全部条目，最新的在前。rowid 打破
// created_at 并列，使同一时钟刻内写入的条目保持稳定的
// 最新优先顺序。它是角色隔离边界：属于
// 其他角色的条目绝不返回，因此角色之间从不共享各自的角色日常
// 命名空间（design §4）。它只是检索候选集的*自有*
// 那一半 —— 检索器在排名前把它与共享的用户/项目作用域
// 取并集（design §3/§4，见 Retriever.visibleCandidates）。
//
// 只返回**在册**（status='active'）条目：整理与入库前审查都只做软动作，
// 被合并/被归档的行内容原样留在表里，因此「能看到什么记忆」的每条读路径
// 都必须过 status 闸，否则清理过的行照样会被检索到、照样占满面板。
//
// 同时过**时效闸**（timeGateClause）：已过期（expires_at ≤ now）与已失效
// （valid_to ≤ now）的行一并不再返回 —— 判据细节与「为什么必须下沉到这里」
// 见 timegate.go 与 design/03 §9.6/§11.2。
func (s *Store) ListByPersona(ctx context.Context, personaID string) ([]MemoryItem, error) {
	if strings.TrimSpace(personaID) == "" {
		return nil, ErrMissingPersona
	}
	// 闸的左右操作数同源：两边都由 formatTime 产出定宽 UTC 文本。
	now := timeGateNow()
	rows, err := s.db.QueryContext(ctx, `SELECT id, persona_id, role_id, project_id, scope, kind,
		content, tags, confidence, source_conversation, created_at, updated_at,
		decay_weight FROM memory_items WHERE persona_id = ? AND status = ?`+timeGateClause+`
		ORDER BY created_at DESC, rowid DESC`,
		personaID, string(tidyStatusActive), now, now)
	if err != nil {
		return nil, fmt.Errorf("memory: list persona items: %w", err)
	}
	return scanItems(rows, "persona item")
}

// List 返回跨所有角色的每条**在册**记忆条目，最新的在前。rowid
// 打破 created_at 并列，使同一时钟刻内写入的条目保持
// 稳定的最新优先顺序。它是聚合/管理视图，不受角色隔离边界限制；必须
// 遵守隔离的调用方应改用 ListByPersona 或 ListScope。返回的
// 切片非 nil。
//
// 「在册」= status='active'：被软合并（'merged'）与被归档（'archived'）
// 的行**内容与行都还在**，但不再参与检索、不再进上下文、也不再出现在
// 管理面板的记忆列表里（design/23 §8/§10 的软动作口径）。要看它们走
// 审查留痕 `GET /admin/memories/reviews` 或按 id 用 Get —— 后者刻意
// 不过 status 闸，因为撤销软合并必须先读得到被吸收的行。
//
// 同样过**时效闸**：临时记忆到期（expires_at ≤ now）与事实失效
// （valid_to ≤ now）的行一并从这里消失（timegate.go、design/03 §9.6）。
func (s *Store) List(ctx context.Context) ([]MemoryItem, error) {
	now := timeGateNow()
	rows, err := s.db.QueryContext(ctx, `SELECT id, persona_id, role_id, project_id, scope, kind,
		content, tags, confidence, source_conversation, created_at, updated_at,
		decay_weight FROM memory_items WHERE status = ?`+timeGateClause+`
		ORDER BY created_at DESC, rowid DESC`, string(tidyStatusActive), now, now)
	if err != nil {
		return nil, fmt.Errorf("memory: list items: %w", err)
	}
	return scanItems(rows, "item")
}

// Delete 删除给定 id 的条目，id 为空或
// 没有删除任何行时返回 ErrNotFound。调用方只应传入经
// 角色作用域读取（ListByPersona/Get）得到的 id，使删除始终留在
// 调用方的角色边界内。
func (s *Store) Delete(ctx context.Context, id string) error {
	if strings.TrimSpace(id) == "" {
		return ErrNotFound
	}
	res, err := s.db.ExecContext(ctx, `DELETE FROM memory_items WHERE id = ?`, id)
	if err != nil {
		return fmt.Errorf("memory: delete item: %w", err)
	}
	n, err := res.RowsAffected()
	if err != nil {
		return fmt.Errorf("memory: delete row count: %w", err)
	}
	if n == 0 {
		return ErrNotFound
	}
	return nil
}

// RememberManual 为 content 文本写入一条长期记忆条目 —— design §1 中的
// 「记住 X」路径。条目默认取角色作用域，
// 并以满置信度记为关系记忆。返回已落库的条目。
func (s *Store) RememberManual(ctx context.Context, personaID, roleID, text string) (MemoryItem, error) {
	item := MemoryItem{
		PersonaID: personaID,
		RoleID:    roleID,
		Scope:     ScopePersona,
		Kind:      KindRelationship,
		Content:   strings.TrimSpace(text),
	}
	if item.Content == "" {
		return MemoryItem{}, ErrMissingContent
	}
	return s.Add(ctx, item)
}

// normalize 填充默认的 Scope/Kind/Confidence/DecayWeight，然后校验。
// 它由 Add 调用，因此调用方可以省略取值，同时仍能
// 捕获显式非法值。
func (it *MemoryItem) normalize() error {
	if it.Scope == "" {
		it.Scope = ScopePersona
	}
	if it.Kind == "" {
		it.Kind = KindRelationship
	}
	if it.Confidence == 0 {
		it.Confidence = defaultConfidence
	}
	if it.DecayWeight == 0 {
		it.DecayWeight = defaultDecayWeight
	}
	return it.validate()
}

// validate 检查必填与枚举字段。它是纯函数：不施加
// 默认值，因此也可用于拒绝调用方提供的值。
func (it *MemoryItem) validate() error {
	if strings.TrimSpace(it.PersonaID) == "" {
		return ErrMissingPersona
	}
	if strings.TrimSpace(it.Content) == "" {
		return ErrMissingContent
	}
	if !it.Scope.valid() {
		return ErrInvalidScope
	}
	if !it.Kind.valid() {
		return ErrInvalidKind
	}
	return nil
}

// applyTimestamps 分配 id 与缺失的 CreatedAt/UpdatedAt 值。
// 仅在无法生成新 id 时返回错误。
func applyTimestamps(item MemoryItem) (MemoryItem, error) {
	now := time.Now().UTC()
	if item.ID == "" {
		id, err := newID()
		if err != nil {
			return MemoryItem{}, err
		}
		item.ID = id
	}
	if item.CreatedAt.IsZero() {
		item.CreatedAt = now
	}
	if item.UpdatedAt.IsZero() {
		item.UpdatedAt = now
	}
	return item, nil
}

// newID 返回一个 128 位密码学随机 id，十六进制编码。它避开
// UUID 依赖，使本包保持自包含。rand.Read 只在
// 平台熵源损坏时失败，因此错误向上传播。
func newID() (string, error) {
	b := make([]byte, 16)
	if _, err := rand.Read(b); err != nil {
		return "", fmt.Errorf("memory: generate id: %w", err)
	}
	return hex.EncodeToString(b), nil
}

// discardLogger 在 log 非 nil 时返回 log，否则返回一个不写任何地方的 logger。
func discardLogger(log *slog.Logger) *slog.Logger {
	if log != nil {
		return log
	}
	return slog.New(slog.NewTextHandler(io.Discard, nil))
}