diff --git a/.env.example b/.env.example index 962ecfb0..47374809 100644 --- a/.env.example +++ b/.env.example @@ -2,6 +2,10 @@ MNEMON_DATA_DIR=~/.mnemon MNEMON_STORE=default +# Retention. Set MNEMON_MAX_INSIGHTS=0 to disable automatic pruning. +MNEMON_MAX_INSIGHTS=1000 +MNEMON_AUTO_PRUNE_MIN_AGE=24h + # Optional embeddings. The defaults below use Ollama. MNEMON_EMBED_ENDPOINT=http://localhost:11434 MNEMON_EMBED_MODEL=nomic-embed-text diff --git a/README.md b/README.md index b4858b89..a24f616c 100644 --- a/README.md +++ b/README.md @@ -446,6 +446,16 @@ Mnemon architecture. | `MNEMON_DATA_DIR` | `~/.mnemon` | Base data directory | | `MNEMON_STORE` | *(active file or `default`)* | Named memory store for data isolation | +**Retention**: + +| Environment Variable | Default | Description | +|---|---|---| +| `MNEMON_MAX_INSIGHTS` | `1000` | Active-insight ceiling; `0` disables automatic pruning | +| `MNEMON_AUTO_PRUNE_MIN_AGE` | `24h` | Grace period before an insight can be auto-pruned; accepts `24h`, `7d`, or `0` | + +Each automatic deletion is soft, appears in the oplog as a `prune` operation, and is +reported by ID in the triggering command's `auto_pruned_ids` field. + **Embedding** (only relevant if using embeddings): | Environment Variable | Default | Description | diff --git a/cmd/memory/import.go b/cmd/memory/import.go index c8f7c10f..a20923fc 100644 --- a/cmd/memory/import.go +++ b/cmd/memory/import.go @@ -227,7 +227,7 @@ exports are documented in docs/IMPORT.md.`, edgesInserted := 0 temporalEdgesRepaired := 0 - pruned := 0 + prunedIDs := []string{} if err := db.InTransaction(func() error { // Insert explicit edges for successfully imported insights. for _, de := range draft.Edges { @@ -277,7 +277,7 @@ exports are documented in docs/IMPORT.md.`, } var pruneErr error - pruned, pruneErr = db.AutoPrune(store.MaxInsightsLimit(), nil) + prunedIDs, pruneErr = db.AutoPruneWithResult(store.MaxInsightsLimit(), nil, "") return pruneErr }); err != nil { return fmt.Errorf("finalize import graph: %w", err) @@ -285,13 +285,14 @@ exports are documented in docs/IMPORT.md.`, _ = temporalEdgesRepaired // computed internally; not surfaced in default output summary := map[string]interface{}{ - "imported": countAction(results, "added"), - "updated": countAction(results, "updated"), - "skipped": countAction(results, "skipped"), - "errors": countErrors(results), - "edges_inserted": edgesInserted, - "auto_pruned": pruned, - "results": results, + "imported": countAction(results, "added"), + "updated": countAction(results, "updated"), + "skipped": countAction(results, "skipped"), + "errors": countErrors(results), + "edges_inserted": edgesInserted, + "auto_pruned": len(prunedIDs), + "auto_pruned_ids": prunedIDs, + "results": results, } enc := json.NewEncoder(os.Stdout) enc.SetIndent("", " ") diff --git a/cmd/memory/import_test.go b/cmd/memory/import_test.go index 5431b81c..73dd24bd 100644 --- a/cmd/memory/import_test.go +++ b/cmd/memory/import_test.go @@ -1,6 +1,7 @@ package memory import ( + "encoding/json" "io" "os" "path/filepath" @@ -11,6 +12,79 @@ import ( "github.com/mnemon-dev/mnemon/internal/memory/store" ) +func TestImportGraceProtectsNewbornBatchAndReportsIDs(t *testing.T) { + t.Setenv("MNEMON_EMBED_ENDPOINT", "http://127.0.0.1:1") + t.Setenv("MNEMON_MAX_INSIGHTS", "1") + t.Setenv("MNEMON_AUTO_PRUNE_MIN_AGE", "") + + oldDataDir, oldStoreName, oldReadOnly := dataDir, storeName, readOnly + oldImportNoDiff, oldImportDryRun := importNoDiff, importDryRun + t.Cleanup(func() { + dataDir, storeName, readOnly = oldDataDir, oldStoreName, oldReadOnly + importNoDiff, importDryRun = oldImportNoDiff, oldImportDryRun + }) + dataDir = t.TempDir() + storeName = "" + readOnly = false + importNoDiff = true + importDryRun = false + + draftPath := filepath.Join(t.TempDir(), "memory_draft.json") + draft := `{ + "schema_version": "1", + "insights": [ + {"content": "first newborn memory", "importance": 1, "created_at": "2024-01-01T00:00:00Z"}, + {"content": "second newborn memory", "importance": 1, "created_at": "2024-01-02T00:00:00Z"} + ] +}` + if err := os.WriteFile(draftPath, []byte(draft), 0o644); err != nil { + t.Fatalf("write draft: %v", err) + } + + var runErr error + out := captureStdout(t, func() { + runErr = importCmd.RunE(importCmd, []string{draftPath}) + }) + if runErr != nil { + t.Fatalf("import: %v", runErr) + } + var summary struct { + AutoPruned int `json:"auto_pruned"` + AutoPrunedIDs []string `json:"auto_pruned_ids"` + } + if err := json.Unmarshal([]byte(out), &summary); err != nil { + t.Fatalf("decode import output: %v\n%s", err, out) + } + if summary.AutoPruned != 0 || summary.AutoPrunedIDs == nil || len(summary.AutoPrunedIDs) != 0 { + t.Fatalf("newborn batch was pruned: count %d ids %v", summary.AutoPruned, summary.AutoPrunedIDs) + } + + db, err := store.Open(store.StoreDir(dataDir, store.DefaultStoreName)) + if err != nil { + t.Fatalf("open imported store: %v", err) + } + defer db.Close() + active, err := db.GetAllActiveInsights() + if err != nil { + t.Fatalf("get active insights: %v", err) + } + if len(active) != 2 { + t.Fatalf("active insights = %d, want 2 newborn memories", len(active)) + } + var storedAt string + if err := db.Conn().QueryRow( + `SELECT stored_at FROM insights WHERE content = 'first newborn memory'`).Scan(&storedAt); err != nil { + t.Fatalf("read stored_at: %v", err) + } + physicalTime, err := time.Parse(time.RFC3339, storedAt) + if err != nil { + t.Fatalf("parse stored_at %q: %v", storedAt, err) + } + if physicalTime.Before(time.Now().UTC().Add(-time.Minute)) { + t.Fatalf("historical import inherited event time as stored_at: %s", physicalTime) + } +} + func TestImportRepairsBackdatedTemporalBackbone(t *testing.T) { t.Setenv("MNEMON_EMBED_ENDPOINT", "http://127.0.0.1:1") diff --git a/cmd/memory/remember.go b/cmd/memory/remember.go index 418dc5b6..5f565757 100644 --- a/cmd/memory/remember.go +++ b/cmd/memory/remember.go @@ -208,7 +208,7 @@ var rememberCmd = &cobra.Command{ var ( edgeStats graph.EdgeStats ei float64 - pruned int + prunedIDs []string embedded bool ) err = db.InTransaction(func() error { @@ -259,9 +259,9 @@ var rememberCmd = &cobra.Command{ // Auto-prune if over capacity (excludeID protects the just-created insight) var pruneErr error - pruned, pruneErr = db.AutoPrune(store.MaxInsightsLimit(), []string{insight.ID}) + prunedIDs, pruneErr = db.AutoPruneWithResult(store.MaxInsightsLimit(), []string{insight.ID}, insight.ID) if pruneErr != nil { - fmt.Fprintf(os.Stderr, "warning: auto-prune: %v\n", pruneErr) + return fmt.Errorf("auto-prune: %w", pruneErr) } db.LogOp("remember", insight.ID, insight.Content) @@ -303,7 +303,8 @@ var rememberCmd = &cobra.Command{ "causal_candidates": causalCandidates, "embedded": embedded, "effective_importance": ei, - "auto_pruned": pruned, + "auto_pruned": len(prunedIDs), + "auto_pruned_ids": prunedIDs, } if replacedID != "" { output["replaced_id"] = replacedID diff --git a/cmd/memory/remember_test.go b/cmd/memory/remember_test.go new file mode 100644 index 00000000..2f5b983a --- /dev/null +++ b/cmd/memory/remember_test.go @@ -0,0 +1,152 @@ +package memory + +import ( + "encoding/json" + "strings" + "testing" + "time" + + "github.com/mnemon-dev/mnemon/internal/memory/model" + "github.com/mnemon-dev/mnemon/internal/memory/store" +) + +func configureRememberTest(t *testing.T) { + t.Helper() + oldDataDir, oldStoreName, oldReadOnly := dataDir, storeName, readOnly + oldCategory, oldImportance := remCategory, remImportance + oldTags, oldSource, oldEntities := remTags, remSource, remEntities + oldEntityMode, oldNoDiff := remEntityMode, remNoDiff + t.Cleanup(func() { + dataDir, storeName, readOnly = oldDataDir, oldStoreName, oldReadOnly + remCategory, remImportance = oldCategory, oldImportance + remTags, remSource, remEntities = oldTags, oldSource, oldEntities + remEntityMode, remNoDiff = oldEntityMode, oldNoDiff + }) + + dataDir = t.TempDir() + storeName = "" + readOnly = false + remCategory = "fact" + remImportance = 3 + remTags = "" + remSource = "user" + remEntities = "" + remEntityMode = "merge" + remNoDiff = true + t.Setenv("MNEMON_EMBED_ENDPOINT", "http://127.0.0.1:1") + t.Setenv("MNEMON_MAX_INSIGHTS", "1") + t.Setenv("MNEMON_AUTO_PRUNE_MIN_AGE", "0") +} + +func seedOldPruneCandidate(t *testing.T, id string) { + t.Helper() + db, err := store.Open(store.StoreDir(dataDir, store.DefaultStoreName)) + if err != nil { + t.Fatalf("open seed store: %v", err) + } + createdAt := time.Now().UTC().Add(-48 * time.Hour) + err = db.InsertInsight(&model.Insight{ + ID: id, + Content: "old retention candidate", + Category: model.CategoryFact, + Importance: 1, + Tags: []string{}, + Entities: []string{}, + Source: "test", + CreatedAt: createdAt, + UpdatedAt: createdAt, + }) + if err != nil { + db.Close() + t.Fatalf("insert seed: %v", err) + } + if err := db.Close(); err != nil { + t.Fatalf("close seed store: %v", err) + } +} + +func TestRememberReportsAutoPrunedIDsAndTrigger(t *testing.T) { + configureRememberTest(t) + seedOldPruneCandidate(t, "old-prune-target") + + var runErr error + out := captureStdout(t, func() { + runErr = rememberCmd.RunE(rememberCmd, []string{"new durable memory"}) + }) + if runErr != nil { + t.Fatalf("remember: %v", runErr) + } + var result struct { + ID string `json:"id"` + AutoPruned int `json:"auto_pruned"` + AutoPrunedIDs []string `json:"auto_pruned_ids"` + } + if err := json.Unmarshal([]byte(out), &result); err != nil { + t.Fatalf("decode remember output: %v\n%s", err, out) + } + if result.AutoPruned != 1 || len(result.AutoPrunedIDs) != 1 || result.AutoPrunedIDs[0] != "old-prune-target" { + t.Fatalf("auto-prune result = count %d ids %v", result.AutoPruned, result.AutoPrunedIDs) + } + + db, err := store.Open(store.StoreDir(dataDir, store.DefaultStoreName)) + if err != nil { + t.Fatalf("reopen store: %v", err) + } + defer db.Close() + entries, err := db.GetOplog(20) + if err != nil { + t.Fatalf("get oplog: %v", err) + } + found := false + for _, entry := range entries { + if entry.Operation == "prune" && entry.InsightID == "old-prune-target" { + found = true + if !strings.Contains(entry.Detail, "trigger="+result.ID) { + t.Errorf("auto-prune detail %q does not name trigger %s", entry.Detail, result.ID) + } + } + } + if !found { + t.Fatal("auto-pruned id is missing from the oplog") + } +} + +func TestRememberRollsBackWhenAutoPruneAuditFails(t *testing.T) { + configureRememberTest(t) + seedOldPruneCandidate(t, "rollback-old") + + db, err := store.Open(store.StoreDir(dataDir, store.DefaultStoreName)) + if err != nil { + t.Fatalf("open store: %v", err) + } + if _, err := db.Conn().Exec(` + CREATE TRIGGER reject_remember_auto_prune_audit + BEFORE INSERT ON oplog + WHEN NEW.operation = 'prune' + BEGIN + SELECT RAISE(ABORT, 'audit unavailable'); + END`); err != nil { + db.Close() + t.Fatalf("create rejecting trigger: %v", err) + } + if err := db.Close(); err != nil { + t.Fatalf("close store: %v", err) + } + + if err := rememberCmd.RunE(rememberCmd, []string{"must roll back"}); err == nil { + t.Fatal("remember succeeded after required auto-prune audit failed") + } + + db, err = store.Open(store.StoreDir(dataDir, store.DefaultStoreName)) + if err != nil { + t.Fatalf("reopen store: %v", err) + } + defer db.Close() + active, err := db.GetAllActiveInsights() + if err != nil { + t.Fatalf("get active insights: %v", err) + } + if len(active) != 1 || active[0].ID != "rollback-old" { + t.Fatalf("active insights after rollback = %v, want only rollback-old", active) + } +} diff --git a/docs/IMPORT.md b/docs/IMPORT.md index 17a36a52..024887c1 100644 --- a/docs/IMPORT.md +++ b/docs/IMPORT.md @@ -149,6 +149,7 @@ mnemon import --store project-alpha memory_draft.json "errors": 0, "edges_inserted": 3, "auto_pruned": 0, + "auto_pruned_ids": [], "results": [ {"index": 0, "id": "a1b2c3d4...", "content": "Chose Qdrant...", "action": "added"}, {"index": 1, "id": "e5f6a7b8...", "content": "The user prefers...", "action": "skipped"} @@ -164,6 +165,7 @@ mnemon import --store project-alpha memory_draft.json | `errors` | Number of failed writes. Import allows partial success; script callers should check this is `0` | | `edges_inserted` | Number of explicit edges inserted | | `auto_pruned` | Number of memories auto-pruned after capacity checks | +| `auto_pruned_ids` | Exact IDs soft-deleted by auto-prune; each has a matching `prune` oplog entry | --- diff --git a/docs/USAGE.md b/docs/USAGE.md index eb930f7d..f369eea2 100644 --- a/docs/USAGE.md +++ b/docs/USAGE.md @@ -254,6 +254,13 @@ Nodes are colored by category (decision, fact, insight, preference, context); ed | `MNEMON_EMBED_API_KEY` | (none) | Bearer token for OpenAI-compatible servers | | `MNEMON_EMBED_DIMENSIONS` | (native) | Embedding dimensions; set to truncate (e.g., `256` for Matryoshka models) | | `MNEMON_MAX_INSIGHTS` | `1000` | Active-insight ceiling before auto-pruning starts; `0` disables auto-pruning | +| `MNEMON_AUTO_PRUNE_MIN_AGE` | `24h` | Minimum age before automatic pruning; accepts durations such as `24h`, integer days such as `7d`, or `0` to disable the grace period | + +Auto-prune may temporarily leave the active count above the ceiling when every +eligible insight is still inside the grace period. Age is measured from local +store insertion, so newly imported historical memories are protected too. Each deletion is soft, +commits atomically with a `prune` oplog entry, and is returned by ID in +`auto_pruned_ids` alongside the existing `auto_pruned` count. --- diff --git a/docs/design/05-pipelines.md b/docs/design/05-pipelines.md index ffc0c72d..996dec9f 100644 --- a/docs/design/05-pipelines.md +++ b/docs/design/05-pipelines.md @@ -49,7 +49,8 @@ BEGIN TRANSACTION ├── CreateCausalEdges → keywords + token overlap → auto causal edges └── CreateSemanticEdges → cos >= 0.80 auto-link ④ RefreshEffectiveImportance → update EI decay values - ⑤ AutoPrune → soft-delete lowest EI when total > 1000 + ⑤ AutoPrune → soft-delete lowest EI older than grace period + + required per-ID oplog record COMMIT ``` @@ -74,7 +75,8 @@ COMMIT ], "embedded": true, "effective_importance": 0.85, - "auto_pruned": 0 + "auto_pruned": 0, + "auto_pruned_ids": [] } ``` diff --git a/docs/zh/IMPORT.md b/docs/zh/IMPORT.md index 3205ac73..91a36d8a 100644 --- a/docs/zh/IMPORT.md +++ b/docs/zh/IMPORT.md @@ -148,6 +148,7 @@ mnemon import --store project-alpha memory_draft.json "errors": 0, "edges_inserted": 3, "auto_pruned": 0, + "auto_pruned_ids": [], "results": [ {"index": 0, "id": "a1b2c3d4...", "content": "选择了 Qdrant...", "action": "added"}, {"index": 1, "id": "e5f6a7b8...", "content": "用户偏好简洁的...", "action": "skipped"} @@ -163,6 +164,7 @@ mnemon import --store project-alpha memory_draft.json | `errors` | 写入失败的数量；导入允许部分成功，脚本调用方应检查此字段是否为 0 | | `edges_inserted` | 成功插入的显式边数量 | | `auto_pruned` | 超出容量限制后自动删除的记忆数量 | +| `auto_pruned_ids` | 被自动软删除的准确 ID；每个 ID 都有对应的 `prune` oplog 记录 | --- diff --git a/docs/zh/README.md b/docs/zh/README.md index 8c671aac..da1b0d67 100644 --- a/docs/zh/README.md +++ b/docs/zh/README.md @@ -396,12 +396,17 @@ Sub-agent 委派是可选执行策略。当 runtime 支持时，主 agent 可以 |---------|-------|------| | `MNEMON_DATA_DIR` | `~/.mnemon` | 基础数据目录 | | `MNEMON_STORE` | *（active 文件或 `default`）* | 命名记忆体，用于数据隔离 | +| `MNEMON_MAX_INSIGHTS` | `1000` | 活跃 insight 上限；设为 `0` 可关闭自动清理 | +| `MNEMON_AUTO_PRUNE_MIN_AGE` | `24h` | 自动清理前的保护期；支持 `24h`、`7d` 或 `0` | | `MNEMON_EMBED_ENDPOINT` | `http://localhost:11434` | 嵌入 API 端点 | | `MNEMON_EMBED_MODEL` | `nomic-embed-text` | 嵌入模型名称 | | `MNEMON_EMBED_PROTOCOL` | *（自动探测）* | `ollama` 或 `openai`；端点以 `/v1` 结尾时自动切换 | | `MNEMON_EMBED_API_KEY` | *（无）* | OpenAI 兼容服务器（oMLX、vLLM 等）的 Bearer 令牌 | | `MNEMON_EMBED_DIMENSIONS` | *（原生维度）* | 可选的 Matryoshka 维度截断 | +每次自动删除均为软删除，以 `prune` 操作记录到 oplog，并通过触发命令的 +`auto_pruned_ids` 字段返回具体 ID。 + 嵌入客户端默认使用 Ollama API；当端点以 `/v1` 结尾（或显式设置 `MNEMON_EMBED_PROTOCOL=openai`）时改用 OpenAI 兼容的 embeddings API。例如， 可通过以下配置对接 [oMLX](https://omlx.dev) 等本地服务器： diff --git a/docs/zh/USAGE.md b/docs/zh/USAGE.md index 9eb1139f..8e08fd41 100644 --- a/docs/zh/USAGE.md +++ b/docs/zh/USAGE.md @@ -256,6 +256,12 @@ open graph.html | `MNEMON_EMBED_API_KEY` | （无） | OpenAI 兼容服务器的 Bearer 令牌 | | `MNEMON_EMBED_DIMENSIONS` | (原生维度) | 嵌入向量维度；可设置截断值（例如 Matryoshka 模型使用 `256`） | | `MNEMON_MAX_INSIGHTS` | `1000` | 触发自动清理的活跃洞察数量上限；设为 `0` 可关闭自动清理 | +| `MNEMON_AUTO_PRUNE_MIN_AGE` | `24h` | 可被自动清理前的最短存活时间；支持 `24h`、整数天 `7d`，设为 `0` 可关闭保护期 | + +如果所有候选 insight 都仍处于保护期内，活跃数量可暂时高于上限。保护期按本地 +实际入库时间计算，因此刚导入的历史记忆也会受到保护。每次删除均为软删除，与 +一条 `prune` oplog 记录在同一事务中提交，并通过触发命令的 +`auto_pruned_ids` 返回具体 ID，同时保留原有的 `auto_pruned` 计数。 --- diff --git a/docs/zh/design/05-pipelines.md b/docs/zh/design/05-pipelines.md index e454adfb..06d591f1 100644 --- a/docs/zh/design/05-pipelines.md +++ b/docs/zh/design/05-pipelines.md @@ -47,7 +47,8 @@ BEGIN TRANSACTION ├── CreateCausalEdges → 关键词 + token 重叠 → 自动因果边 └── CreateSemanticEdges → cos ≥ 0.80 自动链接 ④ RefreshEffectiveImportance → 更新 EI 衰减值 - ⑤ AutoPrune → 总量 > 1000 时软删除最低 EI + ⑤ AutoPrune → 软删除超过保护期且 EI 最低的条目 + + 每个 ID 必须写入 oplog COMMIT ``` @@ -72,7 +73,8 @@ COMMIT ], "embedded": true, "effective_importance": 0.85, - "auto_pruned": 0 + "auto_pruned": 0, + "auto_pruned_ids": [] } ``` diff --git a/internal/memory/store/db.go b/internal/memory/store/db.go index d5fd97d5..78abb69c 100644 --- a/internal/memory/store/db.go +++ b/internal/memory/store/db.go @@ -276,6 +276,7 @@ CREATE TABLE IF NOT EXISTS insights ( entities TEXT DEFAULT '[]', source TEXT DEFAULT 'user', access_count INTEGER DEFAULT 0, + stored_at TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, deleted_at TEXT @@ -323,6 +324,34 @@ CREATE INDEX IF NOT EXISTS idx_oplog_created ON oplog(created_at); return fmt.Errorf("add last_accessed_at: %w", err) } + // Retention grace is based on when a row entered this store, not its + // historical event time. Existing rows predate that distinction, so their + // created_at is the least surprising backfill. The column stays nullable for + // compatibility with external sync/import tools that write legacy rows. + if err := addColumnIfNotExists(db.conn, `ALTER TABLE insights ADD COLUMN stored_at TEXT`); err != nil { + return fmt.Errorf("add stored_at: %w", err) + } + if _, err := db.conn.Exec(`UPDATE insights SET stored_at = created_at WHERE stored_at IS NULL OR stored_at = ''`); err != nil { + return fmt.Errorf("backfill stored_at: %w", err) + } + // Keep legacy/external writers safe when they omit the new column. SQLite + // cannot add a column with a non-constant CURRENT_TIMESTAMP default during + // migration, so a trigger supplies the physical insertion time instead. + if _, err := db.conn.Exec(` + CREATE TRIGGER IF NOT EXISTS set_insight_stored_at + AFTER INSERT ON insights + WHEN NEW.stored_at IS NULL OR NEW.stored_at = '' + BEGIN + UPDATE insights + SET stored_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') + WHERE id = NEW.id; + END`); err != nil { + return fmt.Errorf("create stored_at trigger: %w", err) + } + if _, err := db.conn.Exec(`CREATE INDEX IF NOT EXISTS idx_insights_stored ON insights(stored_at)`); err != nil { + return fmt.Errorf("create stored_at index: %w", err) + } + // Phase 3 migration: add embedding column if err := addColumnIfNotExists(db.conn, `ALTER TABLE insights ADD COLUMN embedding BLOB`); err != nil { return fmt.Errorf("add embedding: %w", err) @@ -341,6 +370,9 @@ CREATE INDEX IF NOT EXISTS idx_oplog_created ON oplog(created_at); if _, err := db.conn.Exec(`CREATE INDEX IF NOT EXISTS idx_prune_candidates ON insights(deleted_at, importance, access_count, effective_importance)`); err != nil { return fmt.Errorf("create prune_candidates index: %w", err) } + if _, err := db.conn.Exec(`CREATE INDEX IF NOT EXISTS idx_prune_age_candidates ON insights(deleted_at, importance, access_count, stored_at, effective_importance)`); err != nil { + return fmt.Errorf("create prune_age_candidates index: %w", err) + } // Migration: remove narrative edge type from existing databases if err := db.migrateRemoveNarrativeEdges(); err != nil { diff --git a/internal/memory/store/node.go b/internal/memory/store/node.go index 0f1d95f6..bc4e1fff 100644 --- a/internal/memory/store/node.go +++ b/internal/memory/store/node.go @@ -25,6 +25,10 @@ const ( // PruneBatchSize is how many excess insights to prune at once. PruneBatchSize = 10 + // DefaultAutoPruneMinAge protects newborn insights from write-burst + // retention decisions before they have had a chance to be recalled. + DefaultAutoPruneMinAge = 24 * time.Hour + // MaxInsightsUnlimited is the ceiling MaxInsightsLimit returns when // auto-pruning is switched off. No store reaches it, so AutoPrune's // capacity check never trips and no other code path needs a special case. @@ -56,13 +60,59 @@ func MaxInsightsLimit() int { return n } +// AutoPruneMinAge returns the grace period applied to automatic retention +// pruning. MNEMON_AUTO_PRUNE_MIN_AGE accepts Go durations such as "24h", an +// integer day suffix such as "7d", or "0" to disable the grace period. Invalid +// and negative values fall back to the safe default. +func AutoPruneMinAge() time.Duration { + raw := strings.TrimSpace(os.Getenv("MNEMON_AUTO_PRUNE_MIN_AGE")) + if raw == "" { + return DefaultAutoPruneMinAge + } + + duration, err := parseAutoPruneMinAge(raw) + if err != nil { + fmt.Fprintf(os.Stderr, "warning: invalid MNEMON_AUTO_PRUNE_MIN_AGE %q: %v (using %s)\n", + raw, err, DefaultAutoPruneMinAge) + return DefaultAutoPruneMinAge + } + return duration +} + +func parseAutoPruneMinAge(raw string) (time.Duration, error) { + normalized := strings.ToLower(strings.TrimSpace(raw)) + if strings.HasSuffix(normalized, "d") { + days, err := strconv.ParseInt(strings.TrimSuffix(normalized, "d"), 10, 64) + if err != nil { + return 0, fmt.Errorf("parse days: %w", err) + } + if days < 0 { + return 0, fmt.Errorf("duration must not be negative") + } + if days > int64(math.MaxInt64)/int64(24*time.Hour) { + return 0, fmt.Errorf("duration overflows") + } + return time.Duration(days) * 24 * time.Hour, nil + } + + duration, err := time.ParseDuration(normalized) + if err != nil { + return 0, err + } + if duration < 0 { + return 0, fmt.Errorf("duration must not be negative") + } + return duration, nil +} + // InsertInsight inserts a new insight into the database. func (db *DB) InsertInsight(i *model.Insight) error { _, err := db.execer().Exec( - `INSERT INTO insights (id, content, category, importance, tags, entities, source, access_count, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + `INSERT INTO insights (id, content, category, importance, tags, entities, source, access_count, stored_at, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, i.ID, i.Content, string(i.Category), i.Importance, i.TagsJSON(), i.EntitiesJSON(), i.Source, i.AccessCount, + time.Now().UTC().Format(time.RFC3339), i.CreatedAt.Format(time.RFC3339), i.UpdatedAt.Format(time.RFC3339), ) return err @@ -431,31 +481,47 @@ func (db *DB) GetRetentionCandidates(threshold float64, limit int) ([]RetentionC } // AutoPrune soft-deletes the lowest effective_importance non-immune insights -// when total active count exceeds maxInsights. excludeIDs are protected from pruning -// (typically the just-created insights). Returns number pruned. -// If already inside a transaction (db.tx != nil), executes inline; otherwise wraps in its own transaction. +// when total active count exceeds maxInsights. excludeIDs are protected from +// pruning (typically the just-created insights). Returns number pruned. +// If already inside a transaction (db.tx != nil), executes inline; otherwise +// wraps the deletion and its required audit records in one transaction. func (db *DB) AutoPrune(maxInsights int, excludeIDs []string) (int, error) { + ids, err := db.AutoPruneWithResult(maxInsights, excludeIDs, "") + if err != nil { + return 0, err + } + return len(ids), nil +} + +// AutoPruneWithResult performs AutoPrune and returns every soft-deleted ID. +// triggerInsightID, when present, is written into each durable audit record so +// an operator can connect the retention side effect to its triggering write. +func (db *DB) AutoPruneWithResult(maxInsights int, excludeIDs []string, triggerInsightID string) ([]string, error) { + minAge := AutoPruneMinAge() if db.tx != nil { - return db.autoPrune(maxInsights, excludeIDs) + return db.autoPrune(maxInsights, excludeIDs, triggerInsightID, minAge) } - var pruned int + var prunedIDs []string err := db.InTransaction(func() error { var innerErr error - pruned, innerErr = db.autoPrune(maxInsights, excludeIDs) + prunedIDs, innerErr = db.autoPrune(maxInsights, excludeIDs, triggerInsightID, minAge) return innerErr }) - return pruned, err + if err != nil { + return nil, err + } + return prunedIDs, nil } -func (db *DB) autoPrune(maxInsights int, excludeIDs []string) (int, error) { +func (db *DB) autoPrune(maxInsights int, excludeIDs []string, triggerInsightID string, minAge time.Duration) ([]string, error) { ex := db.execer() var total int if err := ex.QueryRow(`SELECT COUNT(*) FROM insights WHERE deleted_at IS NULL`).Scan(&total); err != nil { - return 0, fmt.Errorf("count insights: %w", err) + return nil, fmt.Errorf("count insights: %w", err) } if total <= maxInsights { - return 0, nil + return []string{}, nil } excess := total - maxInsights @@ -474,51 +540,58 @@ func (db *DB) autoPrune(maxInsights int, excludeIDs []string) (int, error) { } excludeClause = fmt.Sprintf("AND id NOT IN (%s)", strings.Join(placeholders, ",")) } - args = append(args, excess) + cutoff := time.Now().UTC().Add(-minAge).Format(time.RFC3339) + args = append(args, cutoff, excess) // Collect candidate IDs first (close cursor before writing to avoid single-conn deadlock) rows, err := ex.Query( fmt.Sprintf(`SELECT id FROM insights WHERE deleted_at IS NULL AND importance < 4 AND access_count < 3 %s - ORDER BY effective_importance ASC LIMIT ?`, excludeClause), args...) + AND julianday(COALESCE(stored_at, created_at)) <= julianday(?) + ORDER BY effective_importance ASC, created_at ASC, id ASC LIMIT ?`, excludeClause), args...) if err != nil { - return 0, fmt.Errorf("query prune candidates: %w", err) + return nil, fmt.Errorf("query prune candidates: %w", err) } - var ids []string + ids := make([]string, 0, excess) for rows.Next() { var id string if err := rows.Scan(&id); err != nil { rows.Close() - return 0, fmt.Errorf("scan prune candidate: %w", err) + return nil, fmt.Errorf("scan prune candidate: %w", err) } ids = append(ids, id) } + if err := rows.Err(); err != nil { + rows.Close() + return nil, fmt.Errorf("iterate prune candidates: %w", err) + } rows.Close() now := time.Now().UTC().Format(time.RFC3339) - pruned := 0 + prunedIDs := make([]string, 0, len(ids)) for _, id := range ids { res, err := ex.Exec( `UPDATE insights SET deleted_at = ?, updated_at = ? WHERE id = ? AND deleted_at IS NULL`, now, now, id) if err != nil { - return pruned, fmt.Errorf("prune %s: %w", id, err) + return nil, fmt.Errorf("prune %s: %w", id, err) } if n, _ := res.RowsAffected(); n > 0 { if err := db.DeleteEdgesByNode(id); err != nil { - return pruned, fmt.Errorf("delete edges for pruned %s: %w", id, err) + return nil, fmt.Errorf("delete edges for pruned %s: %w", id, err) + } + detail := fmt.Sprintf("auto-prune: over capacity (active=%d, max=%d, min_age=%s)", total, maxInsights, minAge) + if triggerInsightID != "" { + detail += fmt.Sprintf(" trigger=%s", triggerInsightID) + } + if err := db.RecordOp("prune", id, detail); err != nil { + return nil, fmt.Errorf("record auto-prune for %s: %w", id, err) } - // Auto-prune is the only destructive path that leaves no trace: - // every other write the CLI makes -- remember, forget, link, - // import, embed -- records an oplog entry. Without this, a store - // can silently lose thousands of insights with no way to find out - // which, when, or why. - db.LogOp("prune", id, fmt.Sprintf("auto-prune: over capacity (active=%d, max=%d)", total, maxInsights)) - pruned++ + prunedIDs = append(prunedIDs, id) } } - return pruned, nil + return prunedIDs, nil } // BoostRetention boosts an insight's retention: access_count +3, refreshes last_accessed_at. diff --git a/internal/memory/store/oplog.go b/internal/memory/store/oplog.go index afc07ac9..00074159 100644 --- a/internal/memory/store/oplog.go +++ b/internal/memory/store/oplog.go @@ -16,10 +16,23 @@ func (db *DB) LogOp(operation, insightID, detail string) { if db.readOnly { return } + if err := db.RecordOp(operation, insightID, detail); err != nil { + fmt.Fprintf(os.Stderr, "warning: oplog insert: %v\n", err) + } +} + +// RecordOp durably appends one operation and reports insertion failures. Use it +// for destructive transitions whose state change and audit evidence must commit +// or roll back together. Callers must invoke it inside the same transaction as +// the transition. +func (db *DB) RecordOp(operation, insightID, detail string) error { + if db.readOnly { + return fmt.Errorf("database is read-only") + } if _, err := db.execer().Exec( `INSERT INTO oplog (operation, insight_id, detail, created_at) VALUES (?, ?, ?, ?)`, operation, insightID, detail, time.Now().UTC().Format(time.RFC3339)); err != nil { - fmt.Fprintf(os.Stderr, "warning: oplog insert: %v\n", err) + return err } // Trim old entries: only deletes when count exceeds limit (O(1) in the common case). @@ -28,6 +41,7 @@ func (db *DB) LogOp(operation, insightID, detail string) { MaxOplogEntries); err != nil { fmt.Fprintf(os.Stderr, "warning: oplog trim: %v\n", err) } + return nil } // OplogEntry represents a single operation log entry. diff --git a/internal/memory/store/store_test.go b/internal/memory/store/store_test.go index 5e0c1fae..ba19ecda 100644 --- a/internal/memory/store/store_test.go +++ b/internal/memory/store/store_test.go @@ -2,10 +2,12 @@ package store import ( "bytes" + "database/sql" "encoding/binary" "math" "os" "path/filepath" + "strings" "testing" "time" @@ -713,9 +715,84 @@ func TestMigrateRemoveNarrativeEdges_KeepsRealEdgesNamedLikeTheSentinel(t *testi } } +func TestMigrateStoredAtBackfillsLegacyInsights(t *testing.T) { + dir := t.TempDir() + dbPath := filepath.Join(dir, "mnemon.db") + legacy, err := sql.Open("sqlite", dbPath) + if err != nil { + t.Fatalf("open legacy database: %v", err) + } + if _, err := legacy.Exec(` + CREATE TABLE insights ( + id TEXT PRIMARY KEY, + content TEXT NOT NULL, + category TEXT DEFAULT 'general', + importance INTEGER DEFAULT 3, + tags TEXT DEFAULT '[]', + entities TEXT DEFAULT '[]', + source TEXT DEFAULT 'user', + access_count INTEGER DEFAULT 0, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + deleted_at TEXT + ); + INSERT INTO insights + (id, content, created_at, updated_at) + VALUES + ('legacy-row', 'pre-migration memory', '2025-01-02T03:04:05Z', '2025-01-02T03:04:05Z') + `); err != nil { + legacy.Close() + t.Fatalf("create legacy schema: %v", err) + } + if err := legacy.Close(); err != nil { + t.Fatalf("close legacy database: %v", err) + } + + migrated, err := Open(dir) + if err != nil { + t.Fatalf("migrate legacy database: %v", err) + } + defer migrated.Close() + + var storedAt string + if err := migrated.conn.QueryRow( + `SELECT stored_at FROM insights WHERE id = 'legacy-row'`).Scan(&storedAt); err != nil { + t.Fatalf("read stored_at: %v", err) + } + if storedAt != "2025-01-02T03:04:05Z" { + t.Fatalf("stored_at = %q, want legacy created_at", storedAt) + } + + // A sync tool built against the old schema can continue omitting stored_at; + // the migrated store must still protect that incoming row as newborn. + if _, err := migrated.conn.Exec(` + INSERT INTO insights (id, content, created_at, updated_at) + VALUES ('external-row', 'legacy writer payload', '2020-01-01T00:00:00Z', '2020-01-01T00:00:00Z') + `); err != nil { + t.Fatalf("legacy-style insert after migration: %v", err) + } + if err := migrated.conn.QueryRow( + `SELECT stored_at FROM insights WHERE id = 'external-row'`).Scan(&storedAt); err != nil { + t.Fatalf("read external stored_at: %v", err) + } + physicalTime, err := time.Parse(time.RFC3339, storedAt) + if err != nil { + t.Fatalf("parse external stored_at %q: %v", storedAt, err) + } + if physicalTime.Before(time.Now().UTC().Add(-time.Minute)) { + t.Fatalf("legacy-style insert inherited historical event time: %s", physicalTime) + } +} + // --- AutoPrune --- +func disableAutoPruneGrace(t *testing.T) { + t.Helper() + t.Setenv("MNEMON_AUTO_PRUNE_MIN_AGE", "0") +} + func TestAutoPrune_PrunesLowestEI(t *testing.T) { + disableAutoPruneGrace(t) db := testDB(t) // Insert more than max @@ -743,16 +820,18 @@ func TestAutoPrune_PrunesLowestEI(t *testing.T) { // left no oplog entry — a store could silently shed thousands of insights with // no record of which ones. Every pruned id must be recoverable from the oplog. func TestAutoPrune_RecordsOplogEntryPerPrunedInsight(t *testing.T) { + disableAutoPruneGrace(t) db := testDB(t) for i := range 5 { db.InsertInsight(makeInsight("audit-"+string(rune('a'+i)), "content", 2)) } - pruned, err := db.AutoPrune(3, nil) + prunedIDs, err := db.AutoPruneWithResult(3, nil, "trigger-write") if err != nil { t.Fatalf("auto prune: %v", err) } + pruned := len(prunedIDs) if pruned != 2 { t.Fatalf("want 2 pruned, got %d", pruned) } @@ -768,6 +847,9 @@ func TestAutoPrune_RecordsOplogEntryPerPrunedInsight(t *testing.T) { if e.Detail == "" { t.Errorf("prune entry for %s has empty detail", e.InsightID) } + if !strings.Contains(e.Detail, "trigger=trigger-write") { + t.Errorf("prune entry for %s missing trigger: %q", e.InsightID, e.Detail) + } } } if len(logged) != pruned { @@ -789,9 +871,15 @@ func TestAutoPrune_RecordsOplogEntryPerPrunedInsight(t *testing.T) { t.Errorf("oplog claims %s pruned but it is still active", id) } } + for _, id := range prunedIDs { + if !logged[id] { + t.Errorf("returned pruned id %s has no matching oplog entry", id) + } + } } func TestAutoPrune_RespectsImmune(t *testing.T) { + disableAutoPruneGrace(t) db := testDB(t) // Insert 3 insights: 2 immune (importance=4), 1 not @@ -819,6 +907,7 @@ func TestAutoPrune_RespectsImmune(t *testing.T) { } func TestAutoPrune_RespectsExcludeIDs(t *testing.T) { + disableAutoPruneGrace(t) db := testDB(t) ins1 := makeInsight("ex-1", "content a", 1) ins2 := makeInsight("ex-2", "content b", 1) @@ -861,6 +950,95 @@ func TestMaxInsightsLimit(t *testing.T) { } } +func TestAutoPruneMinAge(t *testing.T) { + tests := []struct { + name string + env string + want time.Duration + }{ + {name: "default", env: "", want: DefaultAutoPruneMinAge}, + {name: "hours", env: "2h", want: 2 * time.Hour}, + {name: "days", env: "7d", want: 7 * 24 * time.Hour}, + {name: "padded", env: " 30m ", want: 30 * time.Minute}, + {name: "zero disables", env: "0", want: 0}, + {name: "negative falls back", env: "-1h", want: DefaultAutoPruneMinAge}, + {name: "invalid falls back", env: "tomorrow", want: DefaultAutoPruneMinAge}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Setenv("MNEMON_AUTO_PRUNE_MIN_AGE", tt.env) + if got := AutoPruneMinAge(); got != tt.want { + t.Errorf("AutoPruneMinAge() with %q = %s, want %s", tt.env, got, tt.want) + } + }) + } +} + +func TestAutoPrune_DefaultGraceProtectsNewbornInsights(t *testing.T) { + t.Setenv("MNEMON_AUTO_PRUNE_MIN_AGE", "") + db := testDB(t) + + old := makeInsight("old", "eligible old memory", 1) + old.CreatedAt = time.Now().UTC().Add(-DefaultAutoPruneMinAge - time.Hour) + old.UpdatedAt = old.CreatedAt + if err := db.InsertInsight(old); err != nil { + t.Fatalf("insert old insight: %v", err) + } + if _, err := db.Conn().Exec(`UPDATE insights SET stored_at = ? WHERE id = ?`, + old.CreatedAt.Format(time.RFC3339), old.ID); err != nil { + t.Fatalf("age stored_at for old insight: %v", err) + } + for _, id := range []string{"newborn-a", "newborn-b"} { + if err := db.InsertInsight(makeInsight(id, "newborn memory", 1)); err != nil { + t.Fatalf("insert %s: %v", id, err) + } + } + + prunedIDs, err := db.AutoPruneWithResult(1, nil, "newborn-b") + if err != nil { + t.Fatalf("auto prune: %v", err) + } + if len(prunedIDs) != 1 || prunedIDs[0] != "old" { + t.Fatalf("pruned ids = %v, want [old]", prunedIDs) + } + for _, id := range []string{"newborn-a", "newborn-b"} { + if _, err := db.GetInsightByID(id); err != nil { + t.Errorf("grace period did not protect %s: %v", id, err) + } + } +} + +func TestAutoPrune_AuditFailureRollsBackDeletion(t *testing.T) { + disableAutoPruneGrace(t) + db := testDB(t) + for _, id := range []string{"rollback-a", "rollback-b"} { + if err := db.InsertInsight(makeInsight(id, "must survive", 1)); err != nil { + t.Fatalf("insert %s: %v", id, err) + } + } + if _, err := db.Conn().Exec(` + CREATE TRIGGER reject_auto_prune_audit + BEFORE INSERT ON oplog + WHEN NEW.operation = 'prune' + BEGIN + SELECT RAISE(ABORT, 'audit unavailable'); + END`); err != nil { + t.Fatalf("create rejecting trigger: %v", err) + } + + if _, err := db.AutoPruneWithResult(1, nil, "rollback-b"); err == nil { + t.Fatal("auto prune succeeded without its required audit record") + } + active, err := db.GetAllActiveInsights() + if err != nil { + t.Fatalf("get active insights: %v", err) + } + if len(active) != 2 { + t.Fatalf("audit failure left %d active insights, want 2", len(active)) + } +} + // Switching auto-prune off has to hold at the capacity check itself, not only // at the call sites, or a future caller reintroduces the reaping. func TestAutoPrune_UnlimitedCeilingPrunesNothing(t *testing.T) { @@ -888,6 +1066,7 @@ func TestAutoPrune_UnlimitedCeilingPrunesNothing(t *testing.T) { // gc reports: the same store that prunes under a lower resolved ceiling is // left whole once MNEMON_MAX_INSIGHTS resolves above its size. func TestAutoPrune_RaisedCeilingChangesEnforcement(t *testing.T) { + disableAutoPruneGrace(t) db := testDB(t) for i := range 5 { if err := db.InsertInsight(makeInsight("raised-"+string(rune('a'+i)), "content", 2)); err != nil { diff --git a/scripts/e2e_test.sh b/scripts/e2e_test.sh index 8854ba1c..bd964b10 100755 --- a/scripts/e2e_test.sh +++ b/scripts/e2e_test.sh @@ -764,14 +764,21 @@ banner "Milestone 10: Auto-Prune Lifecycle" TESTDIR10="$TESTDATA/m10" mkdir -p "$TESTDIR10" -step "auto-prune — insert 5 low-imp + 2 high-imp insights (cap=4 for test)" -# We'll use a small cap to test pruning. The cap is hardcoded at 1000 in production, -# so here we test that the mechanism WORKS by checking auto_pruned=0 under cap. +step "auto-prune — default grace protects a same-second write burst" for i in 1 2 3; do - $M --data-dir "$TESTDIR10" remember --no-diff "Low importance note $i" --cat general --imp 1 > /dev/null + OUT=$(MNEMON_MAX_INSIGHTS=2 $M --data-dir "$TESTDIR10" remember --no-diff "Low importance note $i" --cat general --imp 1) done -OUT=$($M --data-dir "$TESTDIR10" remember --no-diff "High importance decision" --cat decision --imp 5) -assert_jq "auto_pruned is 0 under cap" "$OUT" '.auto_pruned' '0' +assert_jq "newborn memories are not pruned" "$OUT" '.auto_pruned' '0' +assert_jq "newborn prune id list is empty" "$OUT" '.auto_pruned_ids | length' '0' + +step "auto-prune — explicit zero grace returns ids and durable audit rows" +OUT=$(MNEMON_MAX_INSIGHTS=2 MNEMON_AUTO_PRUNE_MIN_AGE=0 $M --data-dir "$TESTDIR10" remember --no-diff "High importance decision" --cat decision --imp 5) +assert_jq "two excess memories are pruned" "$OUT" '.auto_pruned' '2' +assert_jq "pruned ids match count" "$OUT" '.auto_pruned_ids | length' '2' +PRUNED_PREFIX=$(echo "$OUT" | jq -r '.auto_pruned_ids[0][0:8]') +LOG_OUT=$($M --data-dir "$TESTDIR10" log --limit 20) +assert_contains "auto-prune operation is visible" "$LOG_OUT" "auto-prune" +assert_contains "auto-prune log names returned id" "$LOG_OUT" "$PRUNED_PREFIX" step "auto-prune — effective_importance varies by importance level" # imp=5 should have much higher EI than imp=1