| 1 | package sessioncatalog |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "fmt" |
| 6 | "os" |
| 7 | "path/filepath" |
| 8 | "sync/atomic" |
| 9 | "testing" |
| 10 | "time" |
| 11 | |
| 12 | "reasonix/internal/agent" |
| 13 | ) |
| 14 | |
| 15 | func TestRepairBackoffPersistsAndSourceChangesResetIt(t *testing.T) { |
| 16 | ctx := context.Background() |
| 17 | dir := t.TempDir() |
| 18 | path := filepath.Join(dir, "legacy.jsonl") |
| 19 | saveLineageSession(t, path, "question", "answer") |
| 20 | if err := agent.UpdateBranchMeta(path, false, func(meta *agent.BranchMeta) error { |
| 21 | meta.Revision++ |
| 22 | return nil |
| 23 | }); err != nil { |
| 24 | t.Fatal(err) |
| 25 | } |
| 26 | dbPath := filepath.Join(t.TempDir(), "catalog.sqlite") |
| 27 | target := DirectoryTarget{Path: dir, Scope: "global"} |
| 28 | catalog, err := Open(ctx, Options{Path: dbPath, DisableRepair: true}) |
| 29 | if err != nil { |
| 30 | t.Fatal(err) |
| 31 | } |
| 32 | if err := catalog.ReconcileDirectory(ctx, target); err != nil { |
| 33 | t.Fatal(err) |
| 34 | } |
| 35 | retryAt := time.Now().Add(17 * time.Minute).UnixMilli() |
| 36 | if _, err := catalog.db.ExecContext(ctx, `UPDATE catalog_sessions SET repair_state='deferred',repair_attempts=4, |
| 37 | repair_retry_at=?,repair_error_kind='busy',repair_engine_version=? WHERE path_key=?`, retryAt, repairEngineVersion, catalog.pathKey(path)); err != nil { |
| 38 | t.Fatal(err) |
| 39 | } |
| 40 | if err := catalog.Close(ctx); err != nil { |
| 41 | t.Fatal(err) |
| 42 | } |
| 43 | |
| 44 | catalog, err = Open(ctx, Options{Path: dbPath, DisableRepair: true}) |
| 45 | if err != nil { |
| 46 | t.Fatal(err) |
| 47 | } |
| 48 | t.Cleanup(func() { _ = catalog.Close(context.Background()) }) |
| 49 | assertRepairSchedule := func(wantState string, wantAttempts int, wantRetry int64) { |
| 50 | t.Helper() |
| 51 | var state string |
| 52 | var attempts int |
| 53 | var retry int64 |
| 54 | if err := catalog.db.QueryRowContext(ctx, `SELECT repair_state,repair_attempts,repair_retry_at |
| 55 | FROM catalog_sessions WHERE path_key=?`, catalog.pathKey(path)).Scan(&state, &attempts, &retry); err != nil { |
| 56 | t.Fatal(err) |
| 57 | } |
| 58 | if state != wantState || attempts != wantAttempts || retry != wantRetry { |
| 59 | t.Fatalf("repair schedule = %s/%d/%d, want %s/%d/%d", state, attempts, retry, wantState, wantAttempts, wantRetry) |
| 60 | } |
| 61 | } |
| 62 | assertRepairSchedule("deferred", 4, retryAt) |
| 63 | |
| 64 | if err := os.WriteFile(agent.SessionEventLogPath(path), []byte("foreign event artifact\n"), 0o600); err != nil { |
| 65 | t.Fatal(err) |
| 66 | } |
| 67 | if err := catalog.IndexSessionPath(ctx, target, path); err != nil { |
| 68 | t.Fatal(err) |
| 69 | } |
| 70 | assertRepairSchedule("pending", 0, 0) |
| 71 | |
| 72 | if _, err := catalog.db.ExecContext(ctx, `UPDATE catalog_sessions SET repair_state='deferred',repair_attempts=2, |
| 73 | repair_retry_at=?,repair_error_kind='io',repair_source_fingerprint=content_fingerprint||char(0)||meta_fingerprint |
| 74 | WHERE path_key=?`, retryAt, catalog.pathKey(path)); err != nil { |
| 75 | t.Fatal(err) |
| 76 | } |
| 77 | if err := agent.UpdateBranchMeta(path, false, func(meta *agent.BranchMeta) error { |
| 78 | meta.CustomTitle = "changed meta generation" |
| 79 | return nil |
| 80 | }); err != nil { |
| 81 | t.Fatal(err) |
| 82 | } |
| 83 | if err := catalog.IndexSessionPath(ctx, target, path); err != nil { |
| 84 | t.Fatal(err) |
| 85 | } |
| 86 | assertRepairSchedule("pending", 0, 0) |
| 87 | } |
| 88 | |
| 89 | func TestDeferredRepairIsNotReopenedBeforeRetryAt(t *testing.T) { |
| 90 | ctx := context.Background() |
| 91 | now := time.Date(2026, 9, 1, 8, 0, 0, 0, time.UTC) |
| 92 | dir := t.TempDir() |
| 93 | catalog, err := Open(ctx, Options{ |
| 94 | Path: filepath.Join(t.TempDir(), "catalog.sqlite"), DisableRepair: true, |
| 95 | Now: func() time.Time { return now }, |
| 96 | }) |
| 97 | if err != nil { |
| 98 | t.Fatal(err) |
| 99 | } |
| 100 | t.Cleanup(func() { _ = catalog.Close(context.Background()) }) |
| 101 | record := SessionRecord{ |
| 102 | Path: filepath.Join(dir, "large.jsonl"), Directory: dir, Scope: "global", |
| 103 | TurnsState: TurnsUnknown, Health: HealthOK, |
| 104 | } |
| 105 | if _, err := catalog.upsertSessionsWithNotification(ctx, []SessionRecord{record}, nil, "seed", false, upsertDirectoryProjection); err != nil { |
| 106 | t.Fatal(err) |
| 107 | } |
| 108 | var calls int |
| 109 | catalog.testRepairSessionHook = func(context.Context, string) (agent.SessionListingRepairResult, error) { |
| 110 | calls++ |
| 111 | return agent.SessionListingRepairResult{}, agent.ErrSessionListingRepairBusy |
| 112 | } |
| 113 | catalog.runRepairWave(ctx) |
| 114 | catalog.runRepairWave(ctx) |
| 115 | if calls != 1 { |
| 116 | t.Fatalf("repair opened %d times before retry_at, want 1", calls) |
| 117 | } |
| 118 | now = now.Add(30 * time.Second) |
| 119 | catalog.runRepairWave(ctx) |
| 120 | if calls != 2 { |
| 121 | t.Fatalf("repair calls after retry_at = %d, want 2", calls) |
| 122 | } |
| 123 | } |
| 124 | |
| 125 | func TestRepairWaveBatches1056RowsAndDoesNotStarveAfterBlockedRow(t *testing.T) { |
| 126 | ctx := context.Background() |
| 127 | dir := t.TempDir() |
| 128 | catalog, err := Open(ctx, Options{Path: filepath.Join(t.TempDir(), "catalog.sqlite"), DisableRepair: true}) |
| 129 | if err != nil { |
| 130 | t.Fatal(err) |
| 131 | } |
| 132 | t.Cleanup(func() { _ = catalog.Close(context.Background()) }) |
| 133 | records := make([]SessionRecord, 1056) |
| 134 | for i := range records { |
| 135 | records[i] = SessionRecord{ |
| 136 | Path: filepath.Join(dir, fmt.Sprintf("%04d.jsonl", i)), Directory: dir, Scope: "global", |
| 137 | TurnsState: TurnsUnknown, Health: HealthOK, LastActivityAt: int64(1056 - i), |
| 138 | } |
| 139 | } |
| 140 | if _, err := catalog.upsertSessionsWithNotification(ctx, records, nil, "seed", false, upsertDirectoryProjection); err != nil { |
| 141 | t.Fatal(err) |
| 142 | } |
| 143 | failingPath := records[0].Path |
| 144 | catalog.testRepairSessionHook = func(_ context.Context, path string) (agent.SessionListingRepairResult, error) { |
| 145 | if path == failingPath { |
| 146 | return agent.SessionListingRepairResult{Status: agent.SessionListingRepairUnsupported}, nil |
| 147 | } |
| 148 | return agent.SessionListingRepairResult{Status: agent.SessionListingRepairApplied, Preview: "ok", Turns: 1}, nil |
| 149 | } |
| 150 | var reconciles atomic.Int64 |
| 151 | catalog.testReconcileStartHook = func(DirectoryTarget) { reconciles.Add(1) } |
| 152 | lock := catalog.directoryLock(dir) |
| 153 | lock.Lock() |
| 154 | unlocked := false |
| 155 | defer func() { |
| 156 | if !unlocked { |
| 157 | lock.Unlock() |
| 158 | } |
| 159 | }() |
| 160 | |
| 161 | catalog.runRepairWave(ctx) |
| 162 | var valid, blocked int |
| 163 | if err := catalog.db.QueryRowContext(ctx, `SELECT |
| 164 | SUM(CASE WHEN turns_state='valid' THEN 1 ELSE 0 END), |
| 165 | SUM(CASE WHEN turns_state='unknown' AND repair_state='blocked' THEN 1 ELSE 0 END) |
| 166 | FROM catalog_sessions`).Scan(&valid, &blocked); err != nil { |
| 167 | t.Fatal(err) |
| 168 | } |
| 169 | if valid != 1055 || blocked != 1 { |
| 170 | t.Fatalf("repair wave valid/blocked = %d/%d, want 1055/1", valid, blocked) |
| 171 | } |
| 172 | deadline := time.Now().Add(2 * time.Second) |
| 173 | for reconciles.Load() == 0 && time.Now().Before(deadline) { |
| 174 | time.Sleep(10 * time.Millisecond) |
| 175 | } |
| 176 | if got := reconciles.Load(); got != 1 { |
| 177 | t.Fatalf("queued reconciles = %d, want exactly 1", got) |
| 178 | } |
| 179 | time.Sleep(300 * time.Millisecond) |
| 180 | if got := reconciles.Load(); got != 1 { |
| 181 | t.Fatalf("duplicate reconcile started during one wave: %d", got) |
| 182 | } |
| 183 | lock.Unlock() |
| 184 | unlocked = true |
| 185 | } |
| 186 | |
| 187 | func TestRepairResultPreservesDirectoryProjectionUntilReconcile(t *testing.T) { |
| 188 | ctx := context.Background() |
| 189 | catalog, target, _, leaf := openLegacyRecoveryCatalog(t, ctx) |
| 190 | |
| 191 | before, ok, err := catalog.GetSession(ctx, leaf) |
| 192 | if err != nil || !ok { |
| 193 | t.Fatalf("GetSession before repair: ok=%v err=%v", ok, err) |
| 194 | } |
| 195 | beforePage, err := catalog.ListTopics(ctx, TopicPageRequest{Scope: "global", Limit: 50}) |
| 196 | if err != nil || len(beforePage.Items) != 1 { |
| 197 | t.Fatalf("ListTopics before repair: items=%+v err=%v", beforePage.Items, err) |
| 198 | } |
| 199 | |
| 200 | catalog.repairSession(ctx, leaf) |
| 201 | mid, ok, err := catalog.GetSession(ctx, leaf) |
| 202 | if err != nil || !ok { |
| 203 | t.Fatalf("GetSession during repair: ok=%v err=%v", ok, err) |
| 204 | } |
| 205 | midPage, err := catalog.ListTopics(ctx, TopicPageRequest{Scope: "global", Limit: 50}) |
| 206 | if err != nil { |
| 207 | t.Fatal(err) |
| 208 | } |
| 209 | assertDirectoryProjectionEqual(t, mid, before) |
| 210 | if len(midPage.Items) != 1 || midPage.Items[0].TopicID != beforePage.Items[0].TopicID || |
| 211 | midPage.Items[0].RepresentativePath != beforePage.Items[0].RepresentativePath { |
| 212 | t.Fatalf("repair changed topic projection: before=%+v during=%+v", beforePage.Items, midPage.Items) |
| 213 | } |
| 214 | |
| 215 | if err := catalog.ReconcileDirectory(ctx, target); err != nil { |
| 216 | t.Fatal(err) |
| 217 | } |
| 218 | after, ok, err := catalog.GetSession(ctx, leaf) |
| 219 | if err != nil || !ok { |
| 220 | t.Fatalf("GetSession after reconcile: ok=%v err=%v", ok, err) |
| 221 | } |
| 222 | if after.TurnsState != TurnsValid || after.Turns != 2 || after.Preview != "question" { |
| 223 | t.Fatalf("repaired source state was not retained: %+v", after) |
| 224 | } |
| 225 | if after.RecoveryDigest == "" || after.RecoveryRole != RecoveryRoleAdopted || !after.RecoveryCanonical { |
| 226 | t.Fatalf("reconcile did not publish repaired recovery lineage: %+v", after) |
| 227 | } |
| 228 | signature, err := directorySignature(target.Path) |
| 229 | if err != nil { |
| 230 | t.Fatal(err) |
| 231 | } |
| 232 | if skip, err := catalog.directoryScanCanSkip(ctx, target, signature); err != nil || !skip { |
| 233 | t.Fatalf("stable repaired projection cannot skip: skip=%v err=%v", skip, err) |
| 234 | } |
| 235 | } |
| 236 | |
| 237 | func TestRepairDoesNotWaitForDirectoryProjectionLock(t *testing.T) { |
| 238 | ctx := context.Background() |
| 239 | catalog, target, _, leaf := openLegacyRecoveryCatalog(t, ctx) |
| 240 | lock := catalog.directoryLock(target.Path) |
| 241 | lock.Lock() |
| 242 | defer lock.Unlock() |
| 243 | repairDone := make(chan struct{}) |
| 244 | go func() { |
| 245 | catalog.repairSession(ctx, leaf) |
| 246 | close(repairDone) |
| 247 | }() |
| 248 | select { |
| 249 | case <-repairDone: |
| 250 | case <-time.After(2 * time.Second): |
| 251 | t.Fatal("repair waited for the directory projection lock") |
| 252 | } |
| 253 | } |
| 254 | |
| 255 | func openLegacyRecoveryCatalog(t *testing.T, ctx context.Context) (*Catalog, DirectoryTarget, string, string) { |
| 256 | t.Helper() |
| 257 | dir := t.TempDir() |
| 258 | root := filepath.Join(dir, "root.jsonl") |
| 259 | leaf := filepath.Join(dir, "leaf.jsonl") |
| 260 | saveLineageSession(t, root, "question", "answer") |
| 261 | saveLineageSession(t, leaf, "question", "answer", "follow up", "done") |
| 262 | if err := agent.SaveBranchMetaPreserveUpdated(root, agent.BranchMeta{ |
| 263 | ID: "root", Scope: "global", TopicID: "root-topic", TopicTitle: "Root", |
| 264 | SchemaVersion: agent.BranchMetaCountsVersion, Turns: 1, |
| 265 | }); err != nil { |
| 266 | t.Fatal(err) |
| 267 | } |
| 268 | if err := agent.SaveBranchMetaPreserveUpdated(leaf, agent.BranchMeta{ |
| 269 | ID: "leaf", Scope: "global", TopicID: "leaf-topic", TopicTitle: "Leaf", |
| 270 | Recovered: true, ParentID: "root", RecoveryDepth: 1, SchemaVersion: 1, |
| 271 | }); err != nil { |
| 272 | t.Fatal(err) |
| 273 | } |
| 274 | catalog, err := Open(ctx, Options{Path: filepath.Join(t.TempDir(), "catalog.sqlite"), DisableRepair: true}) |
| 275 | if err != nil { |
| 276 | t.Fatal(err) |
| 277 | } |
| 278 | t.Cleanup(func() { _ = catalog.Close(context.Background()) }) |
| 279 | target := DirectoryTarget{Path: dir, Scope: "global"} |
| 280 | if err := catalog.ReconcileDirectory(ctx, target); err != nil { |
| 281 | t.Fatal(err) |
| 282 | } |
| 283 | return catalog, target, root, leaf |
| 284 | } |
| 285 | |
| 286 | func assertDirectoryProjectionEqual(t *testing.T, got, want SessionRecord) { |
| 287 | t.Helper() |
| 288 | if got.TopicID != want.TopicID || got.TopicTitle != want.TopicTitle || |
| 289 | got.RecoveryCopy != want.RecoveryCopy || got.RecoveryGroupID != want.RecoveryGroupID || |
| 290 | got.RecoveryRole != want.RecoveryRole || got.RecoveryCanonical != want.RecoveryCanonical || |
| 291 | got.LogicalTopicID != want.LogicalTopicID || got.OrdinaryVisible != want.OrdinaryVisible { |
| 292 | t.Fatalf("directory projection changed: got=%+v want=%+v", got, want) |
| 293 | } |
| 294 | } |
| 295 |