Files
Cyrene_v2/internal/runtime/memory/store.go
T

444 lines
14 KiB
Go

// Package memory implements the Cyrene v2 memory subsystem described in
// DevDocs/design/03-记忆.md.
//
// It exposes three layers:
//
// - Store: a SQLite-backed long-term memory store. It persists memory items
// with persona/role isolation (design §0 layer ②, §4) and offers
// Add/Get/ListByPersona/Delete plus a manual "记住" path (design §1).
// - WorkingMemory: a per-session sliding window of messages plus a summary
// (design §0 layer ①).
// - Retriever: relevance + recency retrieval and context injection
// (design §2).
//
// Extraction (design §1) is exposed through the Extractor interface; the
// concrete M2 implementation is the rule-based ManualExtractor, which records
// what the user explicitly asks to remember. LLM-driven automatic extraction is
// deferred to milestone M4 and is not implemented here.
//
// Milestone M4 additions that this package deliberately leaves out but keeps
// extension points for: persistent vector storage, decay scheduling, and the
// project-index hierarchy (design §3). The schema already carries decay_weight
// and the retriever already plugs an optional embedding.Embedder, so those
// features land without a schema migration.
package memory
import (
"context"
"crypto/rand"
"database/sql"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"io"
"log/slog"
"strings"
"time"
)
// Scope classifies who a long-term memory belongs to. It is orthogonal to Kind
// and both are stored on every item (design §3 note).
type Scope string
const (
// ScopeUser is a memory about a user's habits, shared across personas.
ScopeUser Scope = "user"
// ScopePersona is a persona's everyday memory; other personas are read-only.
ScopePersona Scope = "persona"
// ScopeProject is a project-scoped memory, readable and writable by all.
ScopeProject Scope = "project"
)
// valid reports whether s is one of the defined scopes.
func (s Scope) valid() bool {
switch s {
case ScopeUser, ScopePersona, ScopeProject:
return true
default:
return false
}
}
// Kind classifies the content category of a long-term memory item.
type Kind string
const (
// KindRelationship is a memory about "you and me": preferences, habits,
// agreements (design §0 "关系记忆").
KindRelationship Kind = "relationship"
// KindWorldEvent is a memory about objective facts and events in the world
// (design §0 "世界/事件记忆").
KindWorldEvent Kind = "world_event"
)
// valid reports whether k is one of the defined kinds.
func (k Kind) valid() bool {
switch k {
case KindRelationship, KindWorldEvent:
return true
default:
return false
}
}
// MemoryItem is a single long-term memory record. The zero value is not a valid
// item: PersonaID and Content are required, Scope and Kind are normalized to
// their defaults by Add.
type MemoryItem struct {
ID string
PersonaID string
RoleID string
Scope Scope
Kind Kind
Content string
Tags []string
Confidence float64
SourceConversation string
CreatedAt time.Time
UpdatedAt time.Time
DecayWeight float64
}
// Sentinels distinguishing memory package failures. Each is wrapped with %w at
// the call site so callers can use errors.Is.
var (
// ErrMissingPersona is returned when an item has no PersonaID.
ErrMissingPersona = errors.New("memory: persona_id is required")
// ErrMissingContent is returned when an item has no content.
ErrMissingContent = errors.New("memory: content is required")
// ErrInvalidScope is returned when an item's Scope is not a defined scope.
ErrInvalidScope = errors.New("memory: invalid scope")
// ErrInvalidKind is returned when an item's Kind is not a defined kind.
ErrInvalidKind = errors.New("memory: invalid kind")
// ErrNotFound is returned when an item id does not exist.
ErrNotFound = errors.New("memory: item not found")
// ErrNilDB is returned when NewStore receives a nil *sql.DB.
ErrNilDB = errors.New("memory: nil db")
)
// Schema defaults and time-window constants.
const (
defaultConfidence = 1.0
defaultDecayWeight = 1.0
)
// Store is a long-term memory store backed by a *sql.DB.
//
// The store owns its schema: NewStore creates the memory_items table and its
// indexes idempotently. Concurrency is delegated to *sql.DB, which is safe for
// concurrent use by contract, so Store holds no further lock. Every query is
// parameterized (?) and lists columns explicitly (no SELECT *), and reads take
// a context.Cancel.
type Store struct {
db *sql.DB
log *slog.Logger
}
// rowScanner is satisfied by both *sql.Row and *sql.Rows, so a single scanItem
// helper serves Get and ListByPersona.
type rowScanner interface {
Scan(dest ...any) error
}
// NewStore builds a Store on db, ensuring the memory_items schema exists.
//
// db must be a live, migrated connection (the migration framework itself is
// owned by internal/db). A nil db returns ErrNilDB. log may be nil, in which
// case a discard logger is used. ctx controls schema creation and must be
// honored by the caller.
func NewStore(ctx context.Context, db *sql.DB, log *slog.Logger) (*Store, error) {
if db == nil {
return nil, ErrNilDB
}
log = discardLogger(log)
for _, ddl := range []string{
createMemoryItemsTable,
createMemoryItemsPersonaIndex,
createMemoryItemsPersonaCreatedIndex,
} {
if _, err := db.ExecContext(ctx, ddl); err != nil {
return nil, fmt.Errorf("memory: apply schema: %w", err)
}
}
return &Store{db: db, log: log}, nil
}
// Add persists item, assigning an id and missing timestamps, and normalizing
// Scope/Kind/Confidence/DecayWeight to their defaults. It returns the stored
// item with the id and timestamps filled in.
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)
}
_, err = s.db.ExecContext(ctx, `INSERT INTO memory_items
(id, persona_id, role_id, scope, kind, content, tags, confidence,
source_conversation, created_at, updated_at, decay_weight)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
item.ID, item.PersonaID, item.RoleID, string(item.Scope), string(item.Kind),
item.Content, string(tagsJSON), item.Confidence, item.SourceConversation,
formatTime(item.CreatedAt), formatTime(item.UpdatedAt), item.DecayWeight)
if err != nil {
return MemoryItem{}, fmt.Errorf("memory: insert item: %w", err)
}
return item, nil
}
// Get returns the item with the given id. It returns ErrNotFound when the id is
// empty or does not exist.
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, 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 returns every item for personaID, newest first. It is the
// persona-isolation boundary: items belonging to other personas are never
// returned, so personas never share memory (design §4).
func (s *Store) ListByPersona(ctx context.Context, personaID string) ([]MemoryItem, error) {
if strings.TrimSpace(personaID) == "" {
return nil, ErrMissingPersona
}
rows, err := s.db.QueryContext(ctx, `SELECT id, persona_id, role_id, scope, kind,
content, tags, confidence, source_conversation, created_at, updated_at,
decay_weight FROM memory_items WHERE persona_id = ?
ORDER BY created_at DESC`, personaID)
if err != nil {
return nil, fmt.Errorf("memory: list persona items: %w", err)
}
defer func() { _ = rows.Close() }()
items := make([]MemoryItem, 0)
for rows.Next() {
item, err := scanItem(rows)
if err != nil {
return nil, fmt.Errorf("memory: scan persona item: %w", err)
}
items = append(items, item)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("memory: iterate persona items: %w", err)
}
return items, nil
}
// Delete removes the item with the given id and returns ErrNotFound when the id
// is empty or no row was removed. Callers should only pass ids they obtained
// through a persona-scoped read (ListByPersona/Get) so deletion stays within
// the caller's persona boundary.
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 writes a single long-term memory item for content text — the
// "记住 X" path from design §1. The item is scoped to the persona by default and
// recorded as a relationship memory at full confidence. Returns the stored item.
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 fills default Scope/Kind/Confidence/DecayWeight then validates.
// It is called by Add so callers may omit values while still catching
// explicitly invalid ones.
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 checks required and enumerated fields. It is pure: it does not apply
// defaults, so it can also be used to reject a caller-provided value.
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 assigns an id and missing CreatedAt/UpdatedAt values. It
// returns an error only when a fresh id cannot be generated.
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
}
// scanItem reads one memory_items row. tags is stored as a JSON array string;
// timestamps are stored as RFC3339 strings.
func scanItem(sc rowScanner) (MemoryItem, error) {
var (
it MemoryItem
scope, kind string
tagsJSON string
createdAt, updatedAt string
)
if err := sc.Scan(&it.ID, &it.PersonaID, &it.RoleID, &scope, &kind,
&it.Content, &tagsJSON, &it.Confidence, &it.SourceConversation,
&createdAt, &updatedAt, &it.DecayWeight); err != nil {
return MemoryItem{}, err
}
it.Scope = Scope(scope)
it.Kind = Kind(kind)
if err := json.Unmarshal([]byte(tagsJSON), &it.Tags); err != nil {
return MemoryItem{}, fmt.Errorf("memory: decode tags: %w", err)
}
if it.Tags == nil {
it.Tags = []string{}
}
var err error
if it.CreatedAt, err = parseTime(createdAt); err != nil {
return MemoryItem{}, err
}
if it.UpdatedAt, err = parseTime(updatedAt); err != nil {
return MemoryItem{}, err
}
return it, nil
}
// newID returns a 128-bit cryptographically random id, hex-encoded. It avoids a
// UUID dependency so the package stays self-contained. rand.Read fails only if
// the platform entropy source is broken, so the error is propagated.
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
}
// formatTime encodes t in UTC as an RFC3339 string. The value is always written
// as UTC so ordering and round-trips are stable across processes.
func formatTime(t time.Time) string {
return t.UTC().Format(time.RFC3339Nano)
}
// parseTime decodes a stored RFC3339 string. Blank input yields a zero time.
func parseTime(s string) (time.Time, error) {
if strings.TrimSpace(s) == "" {
return time.Time{}, nil
}
t, err := time.Parse(time.RFC3339Nano, s)
if err != nil {
return time.Time{}, fmt.Errorf("memory: parse time %q: %w", s, err)
}
return t, nil
}
// discardLogger returns log if non-nil, otherwise a logger that writes nowhere.
func discardLogger(log *slog.Logger) *slog.Logger {
if log != nil {
return log
}
return slog.New(slog.NewTextHandler(io.Discard, nil))
}
// Schema DDL. The migration framework in internal/db is the authoritative
// mechanism for domain tables, but it is intentionally not used here: the db
// package's migration tests assert an exact migration count, and the memory
// store must be self-contained and usable from any *sql.DB. The table and
// indexes are therefore created idempotently by NewStore.
const createMemoryItemsTable = `
CREATE TABLE IF NOT EXISTS memory_items (
id TEXT PRIMARY KEY,
persona_id TEXT NOT NULL,
role_id TEXT NOT NULL DEFAULT '',
scope TEXT NOT NULL,
kind TEXT NOT NULL,
content TEXT NOT NULL,
tags TEXT NOT NULL DEFAULT '[]',
confidence REAL NOT NULL DEFAULT 1.0,
source_conversation TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
decay_weight REAL NOT NULL DEFAULT 1.0
)`
const createMemoryItemsPersonaIndex = `
CREATE INDEX IF NOT EXISTS idx_memory_items_persona_id
ON memory_items (persona_id)`
const createMemoryItemsPersonaCreatedIndex = `
CREATE INDEX IF NOT EXISTS idx_memory_items_persona_id_created_at
ON memory_items (persona_id, created_at)`