| 1 | package sessioncatalog |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "os" |
| 6 | "path/filepath" |
| 7 | "strings" |
| 8 | "testing" |
| 9 | "time" |
| 10 | |
| 11 | "reasonix/internal/agent" |
| 12 | "reasonix/internal/provider" |
| 13 | ) |
| 14 | |
| 15 | func TestDefaultPathUsesV8CacheFile(t *testing.T) { |
| 16 | t.Parallel() |
| 17 | path := DefaultPath() |
| 18 | if path == "" { |
| 19 | // CacheDir unavailable in this environment; empty is still valid. |
| 20 | return |
| 21 | } |
| 22 | if !strings.HasSuffix(filepath.ToSlash(path), "session-catalog/v9.sqlite") { |
| 23 | t.Fatalf("DefaultPath = %q, want .../session-catalog/v9.sqlite", path) |
| 24 | } |
| 25 | if strings.Contains(path, "v1.sqlite") { |
| 26 | t.Fatalf("DefaultPath must not reuse the 1.24.0 v1 cache: %q", path) |
| 27 | } |
| 28 | } |
| 29 | |
| 30 | func forkCatalogRecoveryBranch(t *testing.T, dir, name string) (parentPath, branchPath string, branchMsgs int) { |
| 31 | t.Helper() |
| 32 | parentPath = filepath.Join(dir, name+".jsonl") |
| 33 | parent := agent.NewSession("sys") |
| 34 | parent.Add(agentMessage("user", "first")) |
| 35 | parent.Add(agentMessage("assistant", "one")) |
| 36 | parent.Add(agentMessage("user", "disk "+name)) |
| 37 | if err := parent.Save(parentPath); err != nil { |
| 38 | t.Fatalf("Save parent: %v", err) |
| 39 | } |
| 40 | stale := agent.NewSession("sys") |
| 41 | stale.Add(agentMessage("user", "first")) |
| 42 | stale.Add(agentMessage("assistant", "one")) |
| 43 | stale.Add(agentMessage("user", "local "+name)) |
| 44 | info, err := stale.SaveRecoveryBranch(agent.RecoveryBranchOptions{OriginalPath: parentPath}) |
| 45 | if err != nil { |
| 46 | t.Fatalf("SaveRecoveryBranch: %v", err) |
| 47 | } |
| 48 | return parentPath, info.Path, len(stale.Snapshot()) |
| 49 | } |
| 50 | |
| 51 | func coverCatalogRecoveryParent(t *testing.T, parentPath, branchPath string) { |
| 52 | t.Helper() |
| 53 | branch, err := agent.LoadSession(branchPath) |
| 54 | if err != nil { |
| 55 | t.Fatalf("Load branch: %v", err) |
| 56 | } |
| 57 | parent, err := agent.LoadSession(parentPath) |
| 58 | if err != nil { |
| 59 | t.Fatalf("Load parent: %v", err) |
| 60 | } |
| 61 | parent.Replace(append([]provider.Message(nil), branch.Snapshot()...)) |
| 62 | parent.Add(agentMessage("assistant", "parent kept the recovery content")) |
| 63 | if err := parent.SaveRewrite(parentPath); err != nil { |
| 64 | t.Fatalf("Save covering parent: %v", err) |
| 65 | } |
| 66 | } |
| 67 | |
| 68 | func agentMessage(role, content string) provider.Message { |
| 69 | switch role { |
| 70 | case "assistant": |
| 71 | return provider.Message{Role: provider.RoleAssistant, Content: content} |
| 72 | default: |
| 73 | return provider.Message{Role: provider.RoleUser, Content: content} |
| 74 | } |
| 75 | } |
| 76 | |
| 77 | func assignCatalogTopic(t *testing.T, path, topicID string) { |
| 78 | t.Helper() |
| 79 | if err := agent.UpdateBranchMeta(path, false, func(meta *agent.BranchMeta) error { |
| 80 | meta.Scope = "global" |
| 81 | meta.TopicID = topicID |
| 82 | meta.TopicTitle = topicID |
| 83 | meta.SchemaVersion = agent.BranchMetaCountsVersion |
| 84 | return nil |
| 85 | }); err != nil { |
| 86 | t.Fatalf("UpdateBranchMeta %s: %v", path, err) |
| 87 | } |
| 88 | } |
| 89 | |
| 90 | func upsertProjectedSessionForTest(t *testing.T, catalog *Catalog, record SessionRecord) { |
| 91 | t.Helper() |
| 92 | if _, err := catalog.upsertSessionsWithNotification(context.Background(), []SessionRecord{record}, nil, "test_projection", true, upsertDirectoryProjection); err != nil { |
| 93 | t.Fatal(err) |
| 94 | } |
| 95 | } |
| 96 | |
| 97 | func TestReconcileMarksCoveredRecoveryCopyWithoutMutatingSessions(t *testing.T) { |
| 98 | t.Parallel() |
| 99 | ctx := context.Background() |
| 100 | dir := t.TempDir() |
| 101 | parentPath, coveredPath, _ := forkCatalogRecoveryBranch(t, dir, "covered") |
| 102 | coverCatalogRecoveryParent(t, parentPath, coveredPath) |
| 103 | _, divergedPath, _ := forkCatalogRecoveryBranch(t, dir, "diverged") |
| 104 | assignCatalogTopic(t, parentPath, "shared") |
| 105 | assignCatalogTopic(t, coveredPath, "shared") |
| 106 | assignCatalogTopic(t, divergedPath, "adopted") |
| 107 | |
| 108 | // Capture pre-index fingerprints so a catalog open never rewrites authority. |
| 109 | hashFile := func(path string) string { |
| 110 | t.Helper() |
| 111 | data, err := os.ReadFile(path) |
| 112 | if err != nil { |
| 113 | t.Fatal(err) |
| 114 | } |
| 115 | return string(data) |
| 116 | } |
| 117 | before := map[string]string{ |
| 118 | parentPath: hashFile(parentPath), |
| 119 | coveredPath: hashFile(coveredPath), |
| 120 | divergedPath: hashFile(divergedPath), |
| 121 | } |
| 122 | for _, path := range []string{parentPath, coveredPath, divergedPath} { |
| 123 | before[agent.BranchMetaPath(path)] = hashFile(agent.BranchMetaPath(path)) |
| 124 | } |
| 125 | |
| 126 | catalog, err := Open(ctx, Options{Path: filepath.Join(t.TempDir(), "catalog.sqlite"), DisableRepair: true}) |
| 127 | if err != nil { |
| 128 | t.Fatal(err) |
| 129 | } |
| 130 | t.Cleanup(func() { _ = catalog.Close(context.Background()) }) |
| 131 | if err := catalog.ReconcileDirectory(ctx, DirectoryTarget{Path: dir, Scope: "global"}); err != nil { |
| 132 | t.Fatal(err) |
| 133 | } |
| 134 | |
| 135 | covered, ok, err := catalog.GetSession(ctx, coveredPath) |
| 136 | if err != nil || !ok { |
| 137 | t.Fatalf("GetSession covered: ok=%v err=%v", ok, err) |
| 138 | } |
| 139 | if !covered.Recovered || !covered.RecoveryCopy { |
| 140 | t.Fatalf("covered record = %+v, want recovered recoveryCopy", covered) |
| 141 | } |
| 142 | diverged, ok, err := catalog.GetSession(ctx, divergedPath) |
| 143 | if err != nil || !ok { |
| 144 | t.Fatalf("GetSession diverged: ok=%v err=%v", ok, err) |
| 145 | } |
| 146 | if !diverged.Recovered || diverged.RecoveryCopy { |
| 147 | t.Fatalf("diverged record = %+v, want recovered without recoveryCopy", diverged) |
| 148 | } |
| 149 | parent, ok, err := catalog.GetSession(ctx, parentPath) |
| 150 | if err != nil || !ok || parent.RecoveryCopy { |
| 151 | t.Fatalf("parent record = %+v ok=%v err=%v", parent, ok, err) |
| 152 | } |
| 153 | |
| 154 | shared, ok, err := catalog.GetTopic(ctx, TopicKey{Scope: "global", TopicID: "shared"}) |
| 155 | if err != nil || !ok { |
| 156 | t.Fatalf("GetTopic shared: ok=%v err=%v", ok, err) |
| 157 | } |
| 158 | if shared.Turns != parent.Turns { |
| 159 | t.Fatalf("shared turns = %d, want parent turns %d without copy inflation", shared.Turns, parent.Turns) |
| 160 | } |
| 161 | if shared.RecoveryState == "recovery_only" { |
| 162 | t.Fatalf("shared topic should not be recovery_only: %+v", shared) |
| 163 | } |
| 164 | adopted, ok, err := catalog.GetTopic(ctx, TopicKey{Scope: "global", TopicID: "adopted"}) |
| 165 | if err != nil || !ok { |
| 166 | t.Fatalf("GetTopic adopted: ok=%v err=%v", ok, err) |
| 167 | } |
| 168 | if adopted.RecoveryState == "recovery_only" || adopted.Turns <= 0 { |
| 169 | t.Fatalf("adopted topic = %+v, want visible continued recovery with turns", adopted) |
| 170 | } |
| 171 | |
| 172 | for path, want := range before { |
| 173 | if got := hashFile(path); got != want { |
| 174 | t.Fatalf("authoritative file changed during catalog reconcile: %s", path) |
| 175 | } |
| 176 | } |
| 177 | } |
| 178 | |
| 179 | func TestTopicTurnsIgnoreRecoveryCopyButKeepActivity(t *testing.T) { |
| 180 | t.Parallel() |
| 181 | ctx := context.Background() |
| 182 | catalog, err := Open(ctx, Options{Path: filepath.Join(t.TempDir(), "catalog.sqlite"), DisableRepair: true}) |
| 183 | if err != nil { |
| 184 | t.Fatal(err) |
| 185 | } |
| 186 | t.Cleanup(func() { _ = catalog.Close(context.Background()) }) |
| 187 | base := time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC).UnixMilli() |
| 188 | upsertProjectedSessionForTest(t, catalog, SessionRecord{ |
| 189 | Path: "/s/parent.jsonl", Directory: "/s", Scope: "global", TopicID: "t", |
| 190 | Turns: 3, TurnsState: TurnsValid, Health: HealthOK, LastActivityAt: base, |
| 191 | }) |
| 192 | upsertProjectedSessionForTest(t, catalog, SessionRecord{ |
| 193 | Path: "/s/copy.jsonl", Directory: "/s", Scope: "global", TopicID: "t", |
| 194 | Turns: 9, TurnsState: TurnsValid, Health: HealthOK, Recovered: true, RecoveryCopy: true, |
| 195 | LastActivityAt: base + 60_000, |
| 196 | }) |
| 197 | topic, ok, err := catalog.GetTopic(ctx, TopicKey{Scope: "global", TopicID: "t"}) |
| 198 | if err != nil || !ok { |
| 199 | t.Fatalf("GetTopic: ok=%v err=%v", ok, err) |
| 200 | } |
| 201 | if topic.Turns != 3 { |
| 202 | t.Fatalf("turns = %d, want 3", topic.Turns) |
| 203 | } |
| 204 | if topic.LastActivityAt != base+60_000 { |
| 205 | t.Fatalf("lastActivityAt = %d, want recovery activity", topic.LastActivityAt) |
| 206 | } |
| 207 | if topic.RecoveryState != "recovery_only" && topic.RecoveryState != "" { |
| 208 | t.Fatalf("recovery_state = %q, want ordinary/covered-only state", topic.RecoveryState) |
| 209 | } |
| 210 | } |
| 211 | |
| 212 | func TestTopicTurnsUseMaxOfNormalLineageAndAdoptedRecovery(t *testing.T) { |
| 213 | t.Parallel() |
| 214 | ctx := context.Background() |
| 215 | catalog, err := Open(ctx, Options{Path: filepath.Join(t.TempDir(), "catalog.sqlite"), DisableRepair: true}) |
| 216 | if err != nil { |
| 217 | t.Fatal(err) |
| 218 | } |
| 219 | t.Cleanup(func() { _ = catalog.Close(context.Background()) }) |
| 220 | upsert := func(path string, turns int, recovered, recoveryCopy bool) { |
| 221 | t.Helper() |
| 222 | upsertProjectedSessionForTest(t, catalog, SessionRecord{ |
| 223 | Path: path, Directory: "/s", Scope: "global", TopicID: "t", |
| 224 | Turns: turns, TurnsState: TurnsValid, Health: HealthOK, |
| 225 | Recovered: recovered, RecoveryCopy: recoveryCopy, |
| 226 | }) |
| 227 | } |
| 228 | upsert("/s/parent.jsonl", 3, false, false) |
| 229 | upsert("/s/adopted.jsonl", 5, true, false) |
| 230 | upsert("/s/copy.jsonl", 99, true, true) |
| 231 | |
| 232 | topic, ok, err := catalog.GetTopic(ctx, TopicKey{Scope: "global", TopicID: "t"}) |
| 233 | if err != nil || !ok { |
| 234 | t.Fatalf("GetTopic: ok=%v err=%v", ok, err) |
| 235 | } |
| 236 | if topic.Turns != 5 { |
| 237 | t.Fatalf("turns = %d, want max(normal=3, adopted=5); recovery copy must not inflate", topic.Turns) |
| 238 | } |
| 239 | |
| 240 | upsert("/s/second-normal.jsonl", 4, false, false) |
| 241 | topic, ok, err = catalog.GetTopic(ctx, TopicKey{Scope: "global", TopicID: "t"}) |
| 242 | if err != nil || !ok { |
| 243 | t.Fatalf("GetTopic after second normal: ok=%v err=%v", ok, err) |
| 244 | } |
| 245 | if topic.Turns != 7 { |
| 246 | t.Fatalf("turns = %d, want max(normal=7, adopted=5)", topic.Turns) |
| 247 | } |
| 248 | } |
| 249 | |
| 250 | func TestRecoveryOnlyTopicState(t *testing.T) { |
| 251 | t.Parallel() |
| 252 | ctx := context.Background() |
| 253 | catalog, err := Open(ctx, Options{Path: filepath.Join(t.TempDir(), "catalog.sqlite"), DisableRepair: true}) |
| 254 | if err != nil { |
| 255 | t.Fatal(err) |
| 256 | } |
| 257 | t.Cleanup(func() { _ = catalog.Close(context.Background()) }) |
| 258 | upsertProjectedSessionForTest(t, catalog, SessionRecord{ |
| 259 | Path: "/s/only-copy.jsonl", Directory: "/s", Scope: "global", TopicID: "copy-only", |
| 260 | Turns: 4, TurnsState: TurnsValid, Health: HealthOK, Recovered: true, RecoveryCopy: true, |
| 261 | LastActivityAt: 100, |
| 262 | }) |
| 263 | topic, ok, err := catalog.GetTopic(ctx, TopicKey{Scope: "global", TopicID: "copy-only"}) |
| 264 | if err != nil || !ok { |
| 265 | t.Fatalf("GetTopic: ok=%v err=%v", ok, err) |
| 266 | } |
| 267 | if topic.RecoveryState != "recovery_only" { |
| 268 | t.Fatalf("recovery_state = %q, want recovery_only", topic.RecoveryState) |
| 269 | } |
| 270 | if topic.Turns != 0 { |
| 271 | t.Fatalf("turns = %d, want 0 for recovery-only topic", topic.Turns) |
| 272 | } |
| 273 | } |
| 274 |