| 1 | package historycatalog |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "path/filepath" |
| 6 | "testing" |
| 7 | |
| 8 | "reasonix/internal/agent" |
| 9 | "reasonix/internal/provider" |
| 10 | ) |
| 11 | |
| 12 | func saveMessages(t *testing.T, path string, messages ...provider.Message) { |
| 13 | t.Helper() |
| 14 | session := agent.NewSession("") |
| 15 | for _, message := range messages { |
| 16 | session.Add(message) |
| 17 | } |
| 18 | if err := session.Save(path); err != nil { |
| 19 | t.Fatal(err) |
| 20 | } |
| 21 | } |
| 22 | |
| 23 | func TestReconcileAndSearchFTSWithoutStoredBody(t *testing.T) { |
| 24 | t.Parallel() |
| 25 | ctx := context.Background() |
| 26 | root := t.TempDir() |
| 27 | path := filepath.Join(root, "decision.jsonl") |
| 28 | saveMessages(t, path, |
| 29 | provider.Message{Role: provider.RoleUser, Content: "Should we use vector embeddings?"}, |
| 30 | provider.Message{Role: provider.RoleAssistant, Content: "Keep lightweight BM25 retrieval for history."}) |
| 31 | catalog, err := Open(ctx, Options{Path: filepath.Join(t.TempDir(), "history.sqlite")}) |
| 32 | if err != nil { |
| 33 | t.Fatal(err) |
| 34 | } |
| 35 | t.Cleanup(func() { _ = catalog.Close(context.Background()) }) |
| 36 | registered := Root{Path: root, Source: "project", Scope: "project", WorkspaceRoot: "/workspace"} |
| 37 | if err := catalog.ReconcileRoot(ctx, registered); err != nil { |
| 38 | t.Fatal(err) |
| 39 | } |
| 40 | result, err := catalog.Search(ctx, SearchRequest{Query: "lightweight BM25", Scope: "project", WorkspaceRoot: "/workspace", Kinds: []string{"assistant_text"}, Roots: []string{root}, Limit: 5}) |
| 41 | if err != nil { |
| 42 | t.Fatal(err) |
| 43 | } |
| 44 | if len(result.Items) != 1 || result.Items[0].SessionPath != path || result.Items[0].MessageIndex != 1 { |
| 45 | t.Fatalf("result=%#v", result) |
| 46 | } |
| 47 | var storedTerms string |
| 48 | if err := catalog.db.QueryRow(`SELECT terms FROM history_fts WHERE rowid=?`, result.Items[0].RowID).Scan(&storedTerms); err == nil { |
| 49 | t.Fatalf("contentless FTS unexpectedly returned stored terms %q", storedTerms) |
| 50 | } |
| 51 | } |
| 52 | |
| 53 | func TestDrainPendingIncludesRegisteredRoots(t *testing.T) { |
| 54 | ctx := context.Background() |
| 55 | root := t.TempDir() |
| 56 | path := filepath.Join(root, "pending-root.jsonl") |
| 57 | saveMessages(t, path, provider.Message{Role: provider.RoleUser, Content: "registered root flush marker"}) |
| 58 | catalog, err := Open(ctx, Options{InMemory: true}) |
| 59 | if err != nil { |
| 60 | t.Fatal(err) |
| 61 | } |
| 62 | // Stop the background consumer so this test exercises drainPending's own |
| 63 | // contract deterministically instead of racing the worker's select loop. |
| 64 | catalog.cancel() |
| 65 | catalog.wg.Wait() |
| 66 | t.Cleanup(func() { _ = catalog.Close(context.Background()) }) |
| 67 | if !catalog.RegisterRoot(Root{Path: root, Scope: "global"}) { |
| 68 | t.Fatal("registered root was not queued") |
| 69 | } |
| 70 | catalog.drainPending(ctx) |
| 71 | result, err := catalog.Search(ctx, SearchRequest{Query: "marker", Roots: []string{root}}) |
| 72 | if err != nil || len(result.Items) != 1 || result.Items[0].SessionPath != path { |
| 73 | t.Fatalf("drained root result=%#v err=%v", result, err) |
| 74 | } |
| 75 | } |
| 76 | |
| 77 | func TestRewriteRemovesOldTerms(t *testing.T) { |
| 78 | t.Parallel() |
| 79 | ctx := context.Background() |
| 80 | root := t.TempDir() |
| 81 | path := filepath.Join(root, "rewrite.jsonl") |
| 82 | saveMessages(t, path, provider.Message{Role: provider.RoleUser, Content: "obsolete unicorn marker"}) |
| 83 | catalog, err := Open(ctx, Options{Path: filepath.Join(t.TempDir(), "history.sqlite")}) |
| 84 | if err != nil { |
| 85 | t.Fatal(err) |
| 86 | } |
| 87 | t.Cleanup(func() { _ = catalog.Close(context.Background()) }) |
| 88 | target := Root{Path: root, Source: "project", Scope: "project", WorkspaceRoot: root} |
| 89 | if err := catalog.ReconcileRoot(ctx, target); err != nil { |
| 90 | t.Fatal(err) |
| 91 | } |
| 92 | session, err := agent.LoadSession(path) |
| 93 | if err != nil { |
| 94 | t.Fatal(err) |
| 95 | } |
| 96 | session.Rewrite([]provider.Message{{Role: provider.RoleUser, Content: "replacement phoenix marker"}}, "test") |
| 97 | if err := session.SaveRewrite(path); err != nil { |
| 98 | t.Fatal(err) |
| 99 | } |
| 100 | if err := catalog.ReconcileRoot(ctx, target); err != nil { |
| 101 | t.Fatal(err) |
| 102 | } |
| 103 | oldResult, err := catalog.Search(ctx, SearchRequest{Query: "unicorn", Kinds: []string{"user_text"}, Roots: []string{root}}) |
| 104 | if err != nil { |
| 105 | t.Fatal(err) |
| 106 | } |
| 107 | if len(oldResult.Items) != 0 { |
| 108 | t.Fatalf("stale terms survived rewrite: %#v", oldResult.Items) |
| 109 | } |
| 110 | newResult, err := catalog.Search(ctx, SearchRequest{Query: "phoenix", Kinds: []string{"user_text"}, Roots: []string{root}}) |
| 111 | if err != nil || len(newResult.Items) != 1 { |
| 112 | t.Fatalf("newResult=%#v err=%v", newResult, err) |
| 113 | } |
| 114 | } |
| 115 | |
| 116 | func TestAppendIndexesOnlyDisplayTail(t *testing.T) { |
| 117 | t.Parallel() |
| 118 | ctx := context.Background() |
| 119 | root := t.TempDir() |
| 120 | path := filepath.Join(root, "append.jsonl") |
| 121 | saveMessages(t, path, provider.Message{Role: provider.RoleUser, Content: "stable prefix marker"}) |
| 122 | catalog, err := Open(ctx, Options{Path: filepath.Join(t.TempDir(), "history.sqlite")}) |
| 123 | if err != nil { |
| 124 | t.Fatal(err) |
| 125 | } |
| 126 | t.Cleanup(func() { _ = catalog.Close(context.Background()) }) |
| 127 | target := Root{Path: root, Source: "project", Scope: "project", WorkspaceRoot: root} |
| 128 | if err := catalog.ReconcileRoot(ctx, target); err != nil { |
| 129 | t.Fatal(err) |
| 130 | } |
| 131 | var prefixRowID int64 |
| 132 | if err := catalog.db.QueryRow(`SELECT id FROM history_documents WHERE source_path=? AND message_index=0`, path).Scan(&prefixRowID); err != nil { |
| 133 | t.Fatal(err) |
| 134 | } |
| 135 | session, err := agent.LoadSession(path) |
| 136 | if err != nil { |
| 137 | t.Fatal(err) |
| 138 | } |
| 139 | session.Add(provider.Message{Role: provider.RoleAssistant, Content: "new append-only phoenix"}) |
| 140 | if err := session.SaveSnapshot(path); err != nil { |
| 141 | t.Fatal(err) |
| 142 | } |
| 143 | if err := catalog.indexPath(ctx, target, path, 0, 1); err != nil { |
| 144 | t.Fatal(err) |
| 145 | } |
| 146 | var unchangedRowID int64 |
| 147 | if err := catalog.db.QueryRow(`SELECT id FROM history_documents WHERE source_path=? AND message_index=0`, path).Scan(&unchangedRowID); err != nil { |
| 148 | t.Fatal(err) |
| 149 | } |
| 150 | if unchangedRowID != prefixRowID { |
| 151 | t.Fatalf("prefix row was rebuilt: before=%d after=%d", prefixRowID, unchangedRowID) |
| 152 | } |
| 153 | result, err := catalog.Search(ctx, SearchRequest{Query: "phoenix", Roots: []string{root}}) |
| 154 | if err != nil || len(result.Items) != 1 || result.Items[0].MessageIndex != 1 { |
| 155 | t.Fatalf("appended result=%#v err=%v", result, err) |
| 156 | } |
| 157 | } |
| 158 | |
| 159 | func TestSearchKeysetContinuesPastCatalogLimit(t *testing.T) { |
| 160 | t.Parallel() |
| 161 | ctx := context.Background() |
| 162 | root := t.TempDir() |
| 163 | path := filepath.Join(root, "many.jsonl") |
| 164 | messages := make([]provider.Message, 0, 5) |
| 165 | for range 5 { |
| 166 | messages = append(messages, provider.Message{Role: provider.RoleUser, Content: "shared pagination marker"}) |
| 167 | } |
| 168 | saveMessages(t, path, messages...) |
| 169 | catalog, err := Open(ctx, Options{Path: filepath.Join(t.TempDir(), "history.sqlite")}) |
| 170 | if err != nil { |
| 171 | t.Fatal(err) |
| 172 | } |
| 173 | t.Cleanup(func() { _ = catalog.Close(context.Background()) }) |
| 174 | if err := catalog.ReconcileRoot(ctx, Root{Path: root, Source: "project", Scope: "project", WorkspaceRoot: root}); err != nil { |
| 175 | t.Fatal(err) |
| 176 | } |
| 177 | request := SearchRequest{Query: "pagination", Kinds: []string{"user_text"}, Roots: []string{root}, Limit: 2} |
| 178 | seen := map[int]bool{} |
| 179 | for { |
| 180 | result, err := catalog.Search(ctx, request) |
| 181 | if err != nil { |
| 182 | t.Fatal(err) |
| 183 | } |
| 184 | if len(result.Items) == 0 { |
| 185 | break |
| 186 | } |
| 187 | for _, item := range result.Items { |
| 188 | if seen[item.MessageIndex] { |
| 189 | t.Fatalf("message %d repeated across keyset pages", item.MessageIndex) |
| 190 | } |
| 191 | seen[item.MessageIndex] = true |
| 192 | } |
| 193 | last := result.Items[len(result.Items)-1] |
| 194 | request.After = &SearchCursor{Rank: last.Rank, SessionPath: last.SessionPath, MessageIndex: last.MessageIndex, |
| 195 | PartIndex: last.PartIndex, RowID: last.RowID} |
| 196 | } |
| 197 | if len(seen) != len(messages) { |
| 198 | t.Fatalf("indexed messages=%d, want %d", len(seen), len(messages)) |
| 199 | } |
| 200 | } |
| 201 | |
| 202 | func TestToolOutputIsIndexedAndSearchableByExplicitKind(t *testing.T) { |
| 203 | t.Parallel() |
| 204 | ctx := context.Background() |
| 205 | root := t.TempDir() |
| 206 | path := filepath.Join(root, "tools.jsonl") |
| 207 | saveMessages(t, path, |
| 208 | provider.Message{Role: provider.RoleAssistant, ToolCalls: []provider.ToolCall{{ID: "1", Name: "bash", Arguments: `{"cmd":"echo hello"}`}}}, |
| 209 | provider.Message{Role: provider.RoleTool, ToolCallID: "1", Name: "bash", Content: "zephyroutputtokenxyz hello"}, |
| 210 | provider.Message{Role: provider.RoleTool, ToolCallID: "2", Name: "bash", Content: "error: permission denied on quasarerrortokenabc"}, |
| 211 | ) |
| 212 | catalog, err := Open(ctx, Options{Path: filepath.Join(t.TempDir(), "history.sqlite")}) |
| 213 | if err != nil { |
| 214 | t.Fatal(err) |
| 215 | } |
| 216 | t.Cleanup(func() { _ = catalog.Close(context.Background()) }) |
| 217 | if err := catalog.ReconcileRoot(ctx, Root{Path: root, Source: "project", Scope: "project", WorkspaceRoot: root}); err != nil { |
| 218 | t.Fatal(err) |
| 219 | } |
| 220 | |
| 221 | defaultKinds := []string{"user_text", "assistant_text", "tool_input", "tool_error"} |
| 222 | defaultResult, err := catalog.Search(ctx, SearchRequest{Query: "zephyroutputtokenxyz", Kinds: defaultKinds, Roots: []string{root}, Limit: 5}) |
| 223 | if err != nil { |
| 224 | t.Fatal(err) |
| 225 | } |
| 226 | if len(defaultResult.Items) != 0 { |
| 227 | t.Fatalf("default kinds unexpectedly returned tool_output hits: %#v", defaultResult.Items) |
| 228 | } |
| 229 | |
| 230 | outputResult, err := catalog.Search(ctx, SearchRequest{Query: "zephyroutputtokenxyz", Kinds: []string{"tool_output"}, Roots: []string{root}, Limit: 5}) |
| 231 | if err != nil { |
| 232 | t.Fatal(err) |
| 233 | } |
| 234 | if len(outputResult.Items) != 1 || outputResult.Items[0].Kind != "tool_output" { |
| 235 | t.Fatalf("tool_output result=%#v", outputResult.Items) |
| 236 | } |
| 237 | |
| 238 | errorResult, err := catalog.Search(ctx, SearchRequest{Query: "quasarerrortokenabc", Kinds: []string{"tool_error"}, Roots: []string{root}, Limit: 5}) |
| 239 | if err != nil { |
| 240 | t.Fatal(err) |
| 241 | } |
| 242 | if len(errorResult.Items) != 1 || errorResult.Items[0].Kind != "tool_error" { |
| 243 | t.Fatalf("tool_error result=%#v", errorResult.Items) |
| 244 | } |
| 245 | } |
| 246 | |
| 247 | func TestTokenizerVersionMismatchClearsMixedProjection(t *testing.T) { |
| 248 | t.Parallel() |
| 249 | ctx := context.Background() |
| 250 | root := t.TempDir() |
| 251 | path := filepath.Join(root, "old-tokenizer.jsonl") |
| 252 | databasePath := filepath.Join(t.TempDir(), "history.sqlite") |
| 253 | saveMessages(t, path, provider.Message{Role: provider.RoleUser, Content: "tokenizer migration marker"}) |
| 254 | catalog, err := Open(ctx, Options{Path: databasePath}) |
| 255 | if err != nil { |
| 256 | t.Fatal(err) |
| 257 | } |
| 258 | if err := catalog.ReconcileRoot(ctx, Root{Path: root, Source: "project", Scope: "project", WorkspaceRoot: root}); err != nil { |
| 259 | t.Fatal(err) |
| 260 | } |
| 261 | if _, err := catalog.db.Exec(`UPDATE history_state SET tokenizer_version=?`, TokenizerVersion+1); err != nil { |
| 262 | t.Fatal(err) |
| 263 | } |
| 264 | if err := catalog.Close(ctx); err != nil { |
| 265 | t.Fatal(err) |
| 266 | } |
| 267 | reopened, err := Open(ctx, Options{Path: databasePath}) |
| 268 | if err != nil { |
| 269 | t.Fatal(err) |
| 270 | } |
| 271 | t.Cleanup(func() { _ = reopened.Close(context.Background()) }) |
| 272 | result, err := reopened.Search(ctx, SearchRequest{Query: "migration", Roots: []string{root}}) |
| 273 | if err != nil || len(result.Items) != 0 { |
| 274 | t.Fatalf("mixed tokenizer rows survived: result=%#v err=%v", result, err) |
| 275 | } |
| 276 | } |
| 277 |