| 1 | package historycatalog |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "fmt" |
| 6 | "os" |
| 7 | "path/filepath" |
| 8 | "strings" |
| 9 | "testing" |
| 10 | "time" |
| 11 | |
| 12 | "reasonix/internal/agent" |
| 13 | "reasonix/internal/provider" |
| 14 | ) |
| 15 | |
| 16 | func TestResolveMaxBytes(t *testing.T) { |
| 17 | t.Parallel() |
| 18 | tests := []struct { |
| 19 | name string |
| 20 | option int64 |
| 21 | configuredMB int |
| 22 | want int64 |
| 23 | }{ |
| 24 | {"explicit option wins", 12345, 512, 12345}, |
| 25 | {"config max_mb", 0, 512, 512 << 20}, |
| 26 | {"built-in default", 0, 0, DefaultMaxBytes}, |
| 27 | {"negative config ignored", 0, -10, DefaultMaxBytes}, |
| 28 | } |
| 29 | for _, tt := range tests { |
| 30 | t.Run(tt.name, func(t *testing.T) { |
| 31 | t.Parallel() |
| 32 | if got := resolveMaxBytes(tt.option, tt.configuredMB); got != tt.want { |
| 33 | t.Fatalf("resolveMaxBytes(%d,%d)=%d, want %d", tt.option, tt.configuredMB, got, tt.want) |
| 34 | } |
| 35 | }) |
| 36 | } |
| 37 | if DefaultMaxBytes != 256<<20 { |
| 38 | t.Fatalf("DefaultMaxBytes=%d, want 256MiB", DefaultMaxBytes) |
| 39 | } |
| 40 | } |
| 41 | |
| 42 | // saveToolSession writes a session with exchanges tool calls, each carrying a |
| 43 | // 32KB output, so one session's index footprint is meaningful next to the |
| 44 | // database's fixed page overhead. |
| 45 | func saveToolSession(t *testing.T, path, marker string, exchanges int) { |
| 46 | t.Helper() |
| 47 | messages := []provider.Message{{Role: provider.RoleUser, Content: "question about " + marker}} |
| 48 | for i := range exchanges { |
| 49 | id := fmt.Sprintf("call-%d", i) |
| 50 | content := id + " " + strings.Repeat("payload ", 4096) |
| 51 | if i == 0 { |
| 52 | // Keep the marker in exactly one document so search-hit counts are |
| 53 | // per-session, not per-exchange. |
| 54 | content = id + " " + marker + " " + strings.Repeat("payload ", 4096) |
| 55 | } |
| 56 | messages = append(messages, |
| 57 | provider.Message{Role: provider.RoleAssistant, ToolCalls: []provider.ToolCall{{ID: id, Name: "bash", Arguments: `{"cmd":"echo ` + id + `"}`}}}, |
| 58 | provider.Message{Role: provider.RoleTool, ToolCallID: id, Name: "bash", Content: content}) |
| 59 | } |
| 60 | saveMessages(t, path, messages...) |
| 61 | } |
| 62 | |
| 63 | // setSessionModTime back-dates the session file and its sidecars so |
| 64 | // last_activity_at ordering is deterministic. |
| 65 | func setSessionModTime(t *testing.T, path string, mod time.Time) { |
| 66 | t.Helper() |
| 67 | stem := strings.TrimSuffix(path, filepath.Ext(path)) |
| 68 | matches, err := filepath.Glob(stem + "*") |
| 69 | if err != nil { |
| 70 | t.Fatal(err) |
| 71 | } |
| 72 | for _, match := range matches { |
| 73 | if err := os.Chtimes(match, mod, mod); err != nil { |
| 74 | t.Fatal(err) |
| 75 | } |
| 76 | } |
| 77 | } |
| 78 | |
| 79 | func toolOutputHits(t *testing.T, catalog *Catalog, root, marker string) int { |
| 80 | t.Helper() |
| 81 | result, err := catalog.Search(context.Background(), SearchRequest{Query: marker, Kinds: []string{"tool_output"}, Roots: []string{root}}) |
| 82 | if err != nil { |
| 83 | t.Fatal(err) |
| 84 | } |
| 85 | return len(result.Items) |
| 86 | } |
| 87 | |
| 88 | func sourceHealth(t *testing.T, catalog *Catalog, path string) string { |
| 89 | t.Helper() |
| 90 | var health string |
| 91 | if err := catalog.db.QueryRow(`SELECT health FROM history_sources WHERE path=?`, path).Scan(&health); err != nil { |
| 92 | t.Fatal(err) |
| 93 | } |
| 94 | return health |
| 95 | } |
| 96 | |
| 97 | // checkpoint folds the WAL so file-size-based caps are measured post-fold. |
| 98 | func checkpoint(t *testing.T, catalog *Catalog) { |
| 99 | t.Helper() |
| 100 | if _, err := catalog.db.Exec(`PRAGMA wal_checkpoint(TRUNCATE)`); err != nil { |
| 101 | t.Fatal(err) |
| 102 | } |
| 103 | } |
| 104 | |
| 105 | // appendToSession grows an on-disk session so its content fingerprint changes. |
| 106 | func appendToSession(t *testing.T, path, marker string) { |
| 107 | t.Helper() |
| 108 | session, err := agent.LoadSession(path) |
| 109 | if err != nil { |
| 110 | t.Fatal(err) |
| 111 | } |
| 112 | session.Add(provider.Message{Role: provider.RoleUser, Content: "more about " + marker}) |
| 113 | if err := session.SaveSnapshot(path); err != nil { |
| 114 | t.Fatal(err) |
| 115 | } |
| 116 | } |
| 117 | |
| 118 | func TestGovernSizeUnderCapIsNoOp(t *testing.T) { |
| 119 | t.Parallel() |
| 120 | ctx := context.Background() |
| 121 | root := t.TempDir() |
| 122 | path := filepath.Join(root, "alpha.jsonl") |
| 123 | saveToolSession(t, path, "alphamarker", 4) |
| 124 | catalog, err := Open(ctx, Options{Path: filepath.Join(t.TempDir(), "history.sqlite"), MaxBytes: 1 << 30}) |
| 125 | if err != nil { |
| 126 | t.Fatal(err) |
| 127 | } |
| 128 | t.Cleanup(func() { _ = catalog.Close(context.Background()) }) |
| 129 | if err := catalog.ReconcileRoot(ctx, Root{Path: root, Scope: "global"}); err != nil { |
| 130 | t.Fatal(err) |
| 131 | } |
| 132 | catalog.governSize(ctx) |
| 133 | if hits := toolOutputHits(t, catalog, root, "alphamarker"); hits != 1 { |
| 134 | t.Fatalf("under-cap governSize dropped rows: hits=%d", hits) |
| 135 | } |
| 136 | if health := sourceHealth(t, catalog, path); health != "ok" { |
| 137 | t.Fatalf("health=%q, want ok", health) |
| 138 | } |
| 139 | } |
| 140 | |
| 141 | func TestEvictOldestBatchEvictsLeastRecentlyActiveFirst(t *testing.T) { |
| 142 | t.Parallel() |
| 143 | ctx := context.Background() |
| 144 | root := t.TempDir() |
| 145 | dbPath := filepath.Join(t.TempDir(), "history.sqlite") |
| 146 | now := time.Now() |
| 147 | paths := map[string]string{} |
| 148 | for i, marker := range []string{"alpha", "beta", "gamma"} { |
| 149 | path := filepath.Join(root, marker+".jsonl") |
| 150 | saveToolSession(t, path, marker+"marker", 4) |
| 151 | setSessionModTime(t, path, now.Add(time.Duration(i-3)*time.Hour)) |
| 152 | paths[marker] = path |
| 153 | } |
| 154 | catalog, err := Open(ctx, Options{Path: dbPath, MaxBytes: 1 << 30}) |
| 155 | if err != nil { |
| 156 | t.Fatal(err) |
| 157 | } |
| 158 | t.Cleanup(func() { _ = catalog.Close(context.Background()) }) |
| 159 | if err := catalog.ReconcileRoot(ctx, Root{Path: root, Scope: "global"}); err != nil { |
| 160 | t.Fatal(err) |
| 161 | } |
| 162 | evicted, err := catalog.evictOldestBatch(ctx, historyDBFileSize(dbPath), 1) |
| 163 | if err != nil { |
| 164 | t.Fatal(err) |
| 165 | } |
| 166 | if evicted != 1 { |
| 167 | t.Fatalf("evicted=%d, want exactly 1 for a 1-byte overage", evicted) |
| 168 | } |
| 169 | if hits := toolOutputHits(t, catalog, root, "alphamarker"); hits != 0 { |
| 170 | t.Fatalf("oldest session still searchable: hits=%d", hits) |
| 171 | } |
| 172 | for _, marker := range []string{"beta", "gamma"} { |
| 173 | if hits := toolOutputHits(t, catalog, root, marker+"marker"); hits != 1 { |
| 174 | t.Fatalf("%s wrongly evicted: hits=%d", marker, hits) |
| 175 | } |
| 176 | } |
| 177 | if health := sourceHealth(t, catalog, paths["alpha"]); health != "evicted" { |
| 178 | t.Fatalf("evicted source health=%q, want evicted", health) |
| 179 | } |
| 180 | if _, err := os.Stat(paths["alpha"]); err != nil { |
| 181 | t.Fatalf("eviction touched the source session file: %v", err) |
| 182 | } |
| 183 | } |
| 184 | |
| 185 | func TestGovernSizeEvictsDownToTarget(t *testing.T) { |
| 186 | t.Parallel() |
| 187 | ctx := context.Background() |
| 188 | root := t.TempDir() |
| 189 | dbPath := filepath.Join(t.TempDir(), "history.sqlite") |
| 190 | now := time.Now() |
| 191 | markers := []string{"alpha", "beta", "gamma", "delta"} |
| 192 | for i, marker := range markers { |
| 193 | path := filepath.Join(root, marker+".jsonl") |
| 194 | saveToolSession(t, path, marker+"marker", 24) |
| 195 | setSessionModTime(t, path, now.Add(time.Duration(i-len(markers))*time.Hour)) |
| 196 | } |
| 197 | catalog, err := Open(ctx, Options{Path: dbPath, MaxBytes: 1 << 30}) |
| 198 | if err != nil { |
| 199 | t.Fatal(err) |
| 200 | } |
| 201 | t.Cleanup(func() { _ = catalog.Close(context.Background()) }) |
| 202 | if err := catalog.ReconcileRoot(ctx, Root{Path: root, Scope: "global"}); err != nil { |
| 203 | t.Fatal(err) |
| 204 | } |
| 205 | checkpoint(t, catalog) |
| 206 | size := historyDBFileSize(dbPath) |
| 207 | catalog.opts.MaxBytes = size / 2 // over cap, well under the 2x rebuild threshold |
| 208 | catalog.governSize(ctx) |
| 209 | |
| 210 | target := (size / 2) * evictTargetPercent / 100 |
| 211 | final := historyDBFileSize(dbPath) |
| 212 | evictedCount := 0 |
| 213 | for i, marker := range markers { |
| 214 | path := filepath.Join(root, marker+".jsonl") |
| 215 | if sourceHealth(t, catalog, path) == "evicted" { |
| 216 | // Eviction must be a prefix of the last-activity-ascending order. |
| 217 | for j := range i { |
| 218 | prev := filepath.Join(root, markers[j]+".jsonl") |
| 219 | if sourceHealth(t, catalog, prev) != "evicted" { |
| 220 | t.Fatalf("%s evicted while older %s survived", marker, markers[j]) |
| 221 | } |
| 222 | } |
| 223 | evictedCount++ |
| 224 | if hits := toolOutputHits(t, catalog, root, marker+"marker"); hits != 0 { |
| 225 | t.Fatalf("evicted %s still searchable", marker) |
| 226 | } |
| 227 | if _, err := os.Stat(path); err != nil { |
| 228 | t.Fatalf("eviction touched source file %s: %v", marker, err) |
| 229 | } |
| 230 | } else if hits := toolOutputHits(t, catalog, root, marker+"marker"); hits != 1 { |
| 231 | t.Fatalf("retained %s not searchable: hits=%d", marker, hits) |
| 232 | } |
| 233 | } |
| 234 | if evictedCount == 0 { |
| 235 | t.Fatal("over-cap governSize evicted nothing") |
| 236 | } |
| 237 | if final > target && evictedCount < len(markers) { |
| 238 | t.Fatalf("size=%d still above target=%d with sessions left", final, target) |
| 239 | } |
| 240 | if final >= size { |
| 241 | t.Fatalf("eviction reclaimed nothing: before=%d after=%d", size, final) |
| 242 | } |
| 243 | } |
| 244 | |
| 245 | func TestGovernSizeFarOverCapWipesProjection(t *testing.T) { |
| 246 | t.Parallel() |
| 247 | ctx := context.Background() |
| 248 | root := t.TempDir() |
| 249 | dbPath := filepath.Join(t.TempDir(), "history.sqlite") |
| 250 | alphaPath := filepath.Join(root, "alpha.jsonl") |
| 251 | saveToolSession(t, alphaPath, "alphamarker", 4) |
| 252 | catalog, err := Open(ctx, Options{Path: dbPath, MaxBytes: 1 << 30}) |
| 253 | if err != nil { |
| 254 | t.Fatal(err) |
| 255 | } |
| 256 | t.Cleanup(func() { _ = catalog.Close(context.Background()) }) |
| 257 | if err := catalog.ReconcileRoot(ctx, Root{Path: root, Scope: "global"}); err != nil { |
| 258 | t.Fatal(err) |
| 259 | } |
| 260 | checkpoint(t, catalog) |
| 261 | size := historyDBFileSize(dbPath) |
| 262 | catalog.opts.MaxBytes = size / 4 // beyond the 2x rebuild threshold |
| 263 | catalog.governSize(ctx) |
| 264 | for _, table := range []string{"history_fts", "history_documents", "history_sources", "history_roots"} { |
| 265 | var count int |
| 266 | if err := catalog.db.QueryRow(`SELECT COUNT(*) FROM ` + table).Scan(&count); err != nil { |
| 267 | t.Fatal(err) |
| 268 | } |
| 269 | if count != 0 { |
| 270 | t.Fatalf("%s still has %d rows after oversize wipe", table, count) |
| 271 | } |
| 272 | } |
| 273 | if _, err := os.Stat(alphaPath); err != nil { |
| 274 | t.Fatalf("wipe touched the source session file: %v", err) |
| 275 | } |
| 276 | } |
| 277 | |
| 278 | func TestOpenTriggersBackgroundWipeWhenFarOverCap(t *testing.T) { |
| 279 | t.Parallel() |
| 280 | ctx := context.Background() |
| 281 | root := t.TempDir() |
| 282 | dbPath := filepath.Join(t.TempDir(), "history.sqlite") |
| 283 | saveToolSession(t, filepath.Join(root, "alpha.jsonl"), "alphamarker", 4) |
| 284 | catalog, err := Open(ctx, Options{Path: dbPath, MaxBytes: 1 << 30}) |
| 285 | if err != nil { |
| 286 | t.Fatal(err) |
| 287 | } |
| 288 | if err := catalog.ReconcileRoot(ctx, Root{Path: root, Scope: "global"}); err != nil { |
| 289 | t.Fatal(err) |
| 290 | } |
| 291 | if err := catalog.Close(ctx); err != nil { |
| 292 | t.Fatal(err) |
| 293 | } |
| 294 | size := historyDBFileSize(dbPath) |
| 295 | rebuildDone := make(chan struct{}, 1) |
| 296 | reopened, err := Open(ctx, Options{Path: dbPath, MaxBytes: size / 4, OnRevision: func(_ Status, _ []string, reason string) { |
| 297 | if reason == "rebuild-oversize" { |
| 298 | select { |
| 299 | case rebuildDone <- struct{}{}: |
| 300 | default: |
| 301 | } |
| 302 | } |
| 303 | }}) |
| 304 | if err != nil { |
| 305 | t.Fatal(err) |
| 306 | } |
| 307 | t.Cleanup(func() { _ = reopened.Close(context.Background()) }) |
| 308 | select { |
| 309 | case <-rebuildDone: |
| 310 | case <-time.After(20 * time.Second): |
| 311 | t.Fatal("background wipe did not publish rebuild completion within 20s") |
| 312 | } |
| 313 | // The wipe must be followed by a rescan that rebuilds the index (now with |
| 314 | // truncated tool payloads) without blocking startup. |
| 315 | if !reopened.RegisterRoot(Root{Path: root, Scope: "global"}) { |
| 316 | t.Fatal("register rebuilt root was not queued") |
| 317 | } |
| 318 | flushCtx, cancel := context.WithTimeout(ctx, 20*time.Second) |
| 319 | defer cancel() |
| 320 | if err := reopened.Flush(flushCtx); err != nil { |
| 321 | t.Fatalf("flush rebuilt root: %v", err) |
| 322 | } |
| 323 | result, err := reopened.Search(ctx, SearchRequest{Query: "alphamarker", Kinds: []string{"tool_output"}, Roots: []string{root}}) |
| 324 | if err != nil || len(result.Items) != 1 { |
| 325 | t.Fatalf("rebuild rescan result=%#v err=%v", result, err) |
| 326 | } |
| 327 | } |
| 328 | |
| 329 | func TestEvictedSessionStaysEvictedWhenUnchangedAndReindexesOnChange(t *testing.T) { |
| 330 | t.Parallel() |
| 331 | ctx := context.Background() |
| 332 | root := t.TempDir() |
| 333 | dbPath := filepath.Join(t.TempDir(), "history.sqlite") |
| 334 | now := time.Now() |
| 335 | alphaPath := filepath.Join(root, "alpha.jsonl") |
| 336 | betaPath := filepath.Join(root, "beta.jsonl") |
| 337 | saveToolSession(t, alphaPath, "alphamarker", 4) |
| 338 | saveToolSession(t, betaPath, "betamarker", 4) |
| 339 | setSessionModTime(t, alphaPath, now.Add(-2*time.Hour)) |
| 340 | setSessionModTime(t, betaPath, now.Add(-time.Hour)) |
| 341 | catalog, err := Open(ctx, Options{Path: dbPath, MaxBytes: 1 << 30}) |
| 342 | if err != nil { |
| 343 | t.Fatal(err) |
| 344 | } |
| 345 | t.Cleanup(func() { _ = catalog.Close(context.Background()) }) |
| 346 | registered := Root{Path: root, Scope: "global"} |
| 347 | if err := catalog.ReconcileRoot(ctx, registered); err != nil { |
| 348 | t.Fatal(err) |
| 349 | } |
| 350 | if _, err := catalog.evictOldestBatch(ctx, historyDBFileSize(dbPath), 1); err != nil { |
| 351 | t.Fatal(err) |
| 352 | } |
| 353 | if health := sourceHealth(t, catalog, alphaPath); health != "evicted" { |
| 354 | t.Fatalf("alpha health=%q, want evicted", health) |
| 355 | } |
| 356 | |
| 357 | // Touch beta so the root signature changes and both paths are revisited; |
| 358 | // alpha is unchanged and must stay out of the index. |
| 359 | appendToSession(t, betaPath, "betamarker") |
| 360 | if err := catalog.ReconcileRoot(ctx, registered); err != nil { |
| 361 | t.Fatal(err) |
| 362 | } |
| 363 | if health := sourceHealth(t, catalog, alphaPath); health != "evicted" { |
| 364 | t.Fatalf("unchanged alpha resurrected: health=%q", health) |
| 365 | } |
| 366 | if hits := toolOutputHits(t, catalog, root, "alphamarker"); hits != 0 { |
| 367 | t.Fatalf("unchanged alpha re-indexed: hits=%d", hits) |
| 368 | } |
| 369 | |
| 370 | // A content change fully re-indexes an evicted session. |
| 371 | appendToSession(t, alphaPath, "alphamarker") |
| 372 | if err := catalog.indexPath(ctx, registered, alphaPath, 0, -1); err != nil { |
| 373 | t.Fatal(err) |
| 374 | } |
| 375 | if health := sourceHealth(t, catalog, alphaPath); health != "ok" { |
| 376 | t.Fatalf("changed alpha health=%q, want ok", health) |
| 377 | } |
| 378 | if hits := toolOutputHits(t, catalog, root, "alphamarker"); hits != 1 { |
| 379 | t.Fatalf("changed alpha not re-indexed: hits=%d", hits) |
| 380 | } |
| 381 | } |
| 382 |