| 1 | package sessioncatalog |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "os" |
| 6 | "path/filepath" |
| 7 | "strings" |
| 8 | "testing" |
| 9 | |
| 10 | "reasonix/internal/agent" |
| 11 | "reasonix/internal/projectiondb" |
| 12 | "reasonix/internal/provider" |
| 13 | "reasonix/internal/store" |
| 14 | ) |
| 15 | |
| 16 | func writeSchemaTwoSession(t *testing.T, path string) (*agent.Session, *agent.Session) { |
| 17 | t.Helper() |
| 18 | a := agent.NewSession("sys") |
| 19 | a.Add(provider.Message{Role: provider.RoleUser, Content: "q1"}) |
| 20 | a.Add(provider.Message{Role: provider.RoleAssistant, Content: "a1"}) |
| 21 | if err := a.Save(path); err != nil { |
| 22 | t.Fatal(err) |
| 23 | } |
| 24 | if err := agent.SaveBranchMeta(path, agent.BranchMeta{Scope: "global", TopicID: "topic-heads", TopicTitle: "Heads"}); err != nil { |
| 25 | t.Fatal(err) |
| 26 | } |
| 27 | if err := a.Save(path); err != nil { |
| 28 | t.Fatal(err) |
| 29 | } |
| 30 | b, err := agent.LoadSession(path) |
| 31 | if err != nil { |
| 32 | t.Fatal(err) |
| 33 | } |
| 34 | a.Add(provider.Message{Role: provider.RoleUser, Content: "q2 from a"}) |
| 35 | if err := a.Save(path); err != nil { |
| 36 | t.Fatal(err) |
| 37 | } |
| 38 | b.Add(provider.Message{Role: provider.RoleUser, Content: "q2 from b"}) |
| 39 | if err := b.Save(path); err != nil { |
| 40 | t.Fatal(err) |
| 41 | } |
| 42 | return a, b |
| 43 | } |
| 44 | |
| 45 | func TestReconcileProjectsSchemaTwoHeadsFromTheIndex(t *testing.T) { |
| 46 | ctx := context.Background() |
| 47 | dir := t.TempDir() |
| 48 | path := filepath.Join(dir, "heads.jsonl") |
| 49 | _, b := writeSchemaTwoSession(t, path) |
| 50 | |
| 51 | catalog, err := Open(ctx, Options{InMemory: true, DisableRepair: true}) |
| 52 | if err != nil { |
| 53 | t.Fatal(err) |
| 54 | } |
| 55 | t.Cleanup(func() { _ = catalog.Close(context.Background()) }) |
| 56 | if err := catalog.ReconcileDirectory(ctx, DirectoryTarget{Path: dir, Scope: "global"}); err != nil { |
| 57 | t.Fatal(err) |
| 58 | } |
| 59 | page, err := catalog.ListSessions(ctx, SessionPageRequest{Scope: "global", Limit: 10}) |
| 60 | if err != nil || len(page.Items) != 1 { |
| 61 | t.Fatalf("sessions = %+v err=%v", page.Items, err) |
| 62 | } |
| 63 | rec := page.Items[0] |
| 64 | refB, _ := b.Head() |
| 65 | if rec.LogFormat != 2 || rec.HeadCount != 2 || rec.SelectedHeadID != refB.HeadID { |
| 66 | t.Fatalf("record = logFormat %d heads %d selected %q, want schema 2 with b's head selected (%q)", rec.LogFormat, rec.HeadCount, rec.SelectedHeadID, refB.HeadID) |
| 67 | } |
| 68 | if !strings.Contains(rec.ContentFingerprint, "|h:"+refB.HeadID+":") { |
| 69 | t.Fatalf("fingerprint %q must bind the selected head", rec.ContentFingerprint) |
| 70 | } |
| 71 | if rec.TurnsState != TurnsValid || rec.Recovered || rec.RecoveryRole != RecoveryRoleNormal || !rec.OrdinaryVisible { |
| 72 | t.Fatalf("schema-2 row must be an ordinary valid session: %+v", rec) |
| 73 | } |
| 74 | heads, err := catalog.ListHeads(ctx, path) |
| 75 | if err != nil || len(heads) != 2 { |
| 76 | t.Fatalf("heads = %+v err=%v", heads, err) |
| 77 | } |
| 78 | if heads[0].ID != agent.SessionMainHead || heads[0].Kind != agent.HeadKindMain || heads[0].Selected { |
| 79 | t.Fatalf("main head = %+v", heads[0]) |
| 80 | } |
| 81 | if heads[1].ID != refB.HeadID || heads[1].Kind != agent.HeadKindConcurrent || !heads[1].Selected || heads[1].Turns != 2 || heads[1].Path != path { |
| 82 | t.Fatalf("concurrent head = %+v", heads[1]) |
| 83 | } |
| 84 | if none, err := catalog.ListHeads(ctx, filepath.Join(dir, "missing.jsonl")); err != nil || none == nil || len(none) != 0 { |
| 85 | t.Fatalf("missing session heads = %#v err=%v, want empty slice", none, err) |
| 86 | } |
| 87 | } |
| 88 | |
| 89 | func TestReconcileMarksStaleHeadIndexUnknownAndKeepsRows(t *testing.T) { |
| 90 | ctx := context.Background() |
| 91 | dir := t.TempDir() |
| 92 | path := filepath.Join(dir, "heads.jsonl") |
| 93 | writeSchemaTwoSession(t, path) |
| 94 | catalog, err := Open(ctx, Options{InMemory: true, DisableRepair: true}) |
| 95 | if err != nil { |
| 96 | t.Fatal(err) |
| 97 | } |
| 98 | t.Cleanup(func() { _ = catalog.Close(context.Background()) }) |
| 99 | if err := catalog.ReconcileDirectory(ctx, DirectoryTarget{Path: dir, Scope: "global"}); err != nil { |
| 100 | t.Fatal(err) |
| 101 | } |
| 102 | // A log that grew without its index (a writer died mid-save) is projected |
| 103 | // from the sidecar mirror and queued for repair; the head rows stay. |
| 104 | if err := os.Remove(store.SessionEventIndex(path)); err != nil { |
| 105 | t.Fatal(err) |
| 106 | } |
| 107 | f, err := os.OpenFile(store.SessionEventLog(path), os.O_WRONLY|os.O_APPEND, 0o600) |
| 108 | if err != nil { |
| 109 | t.Fatal(err) |
| 110 | } |
| 111 | if _, err := f.WriteString("\n"); err != nil { |
| 112 | t.Fatal(err) |
| 113 | } |
| 114 | f.Close() |
| 115 | if err := catalog.ReconcileDirectory(ctx, DirectoryTarget{Path: dir, Scope: "global"}); err != nil { |
| 116 | t.Fatal(err) |
| 117 | } |
| 118 | page, err := catalog.ListSessions(ctx, SessionPageRequest{Scope: "global", Limit: 10}) |
| 119 | if err != nil || len(page.Items) != 1 { |
| 120 | t.Fatalf("sessions = %+v err=%v", page.Items, err) |
| 121 | } |
| 122 | rec := page.Items[0] |
| 123 | if rec.LogFormat != 2 || rec.HeadCount != 2 || rec.SelectedHeadID == "" || rec.TurnsState != TurnsUnknown { |
| 124 | t.Fatalf("stale projection = %+v, want schema-2 mirror with unknown counts", rec) |
| 125 | } |
| 126 | if heads, err := catalog.ListHeads(ctx, path); err != nil || len(heads) != 2 { |
| 127 | t.Fatalf("head rows after stale index = %+v err=%v", heads, err) |
| 128 | } |
| 129 | } |
| 130 | |
| 131 | func TestFreshCatalogOverStaleHeadIndexProjectsEveryHead(t *testing.T) { |
| 132 | ctx := context.Background() |
| 133 | dir := t.TempDir() |
| 134 | path := filepath.Join(dir, "heads.jsonl") |
| 135 | writeSchemaTwoSession(t, path) |
| 136 | // An older build appends to the log without rewriting the head index. |
| 137 | if err := os.Remove(store.SessionEventIndex(path)); err != nil { |
| 138 | t.Fatal(err) |
| 139 | } |
| 140 | catalog, err := Open(ctx, Options{Path: filepath.Join(t.TempDir(), "catalog.sqlite"), DisableRepair: true}) |
| 141 | if err != nil { |
| 142 | t.Fatal(err) |
| 143 | } |
| 144 | t.Cleanup(func() { _ = catalog.Close(context.Background()) }) |
| 145 | target := DirectoryTarget{Path: dir, Scope: "global"} |
| 146 | assertHeadRowsMatchCount := func(stage string) { |
| 147 | t.Helper() |
| 148 | var rows, declared int |
| 149 | if err := catalog.db.QueryRowContext(ctx, `SELECT (SELECT COUNT(*) FROM catalog_heads), |
| 150 | (SELECT COALESCE(SUM(head_count),0) FROM catalog_sessions)`).Scan(&rows, &declared); err != nil { |
| 151 | t.Fatal(err) |
| 152 | } |
| 153 | if rows != declared || rows != 2 { |
| 154 | t.Fatalf("%s: catalog_heads rows=%d, sum(head_count)=%d, want 2 and 2", stage, rows, declared) |
| 155 | } |
| 156 | page, err := catalog.ListSessions(ctx, SessionPageRequest{Scope: "global", Limit: 10}) |
| 157 | if err != nil || len(page.Items) != 1 || page.Items[0].Path != path { |
| 158 | t.Fatalf("%s: sessions = %+v err=%v", stage, page.Items, err) |
| 159 | } |
| 160 | } |
| 161 | if err := catalog.ReconcileDirectory(ctx, target); err != nil { |
| 162 | t.Fatal(err) |
| 163 | } |
| 164 | catalog.runRepairWave(ctx) |
| 165 | assertHeadRowsMatchCount("after repair") |
| 166 | if rec, ok, err := catalog.GetSession(ctx, path); err != nil || !ok || rec.TurnsState != TurnsValid { |
| 167 | t.Fatalf("repaired session = %+v ok=%v err=%v, want valid", rec, ok, err) |
| 168 | } |
| 169 | if err := catalog.ReconcileDirectory(ctx, target); err != nil { |
| 170 | t.Fatal(err) |
| 171 | } |
| 172 | assertHeadRowsMatchCount("after reconcile") |
| 173 | } |
| 174 | |
| 175 | func TestMigrationV12AddsHeadProjectionAndForcesRescan(t *testing.T) { |
| 176 | ctx := context.Background() |
| 177 | path := filepath.Join(t.TempDir(), "catalog.sqlite") |
| 178 | handle, err := projectiondb.Open(ctx, projectiondb.OpenOptions{ |
| 179 | Path: path, MemoryName: "session-catalog-v11-test", Migrations: sessionMigrations()[:11], RequireDisk: true, |
| 180 | }) |
| 181 | if err != nil { |
| 182 | t.Fatal(err) |
| 183 | } |
| 184 | for _, statement := range []string{ |
| 185 | `INSERT INTO catalog_directories(path,path_key,scope) VALUES('/Sessions','/sessions','global')`, |
| 186 | `INSERT INTO catalog_sessions(path,path_key,directory,directory_key,scope,topic_id) VALUES('/Sessions/Old.jsonl','/sessions/old.jsonl','/Sessions','/sessions','global','topic')`, |
| 187 | } { |
| 188 | if _, err := handle.DB.ExecContext(ctx, statement); err != nil { |
| 189 | _ = handle.DB.Close() |
| 190 | t.Fatal(err) |
| 191 | } |
| 192 | } |
| 193 | if err := handle.DB.Close(); err != nil { |
| 194 | t.Fatal(err) |
| 195 | } |
| 196 | migrated, err := projectiondb.Open(ctx, projectiondb.OpenOptions{ |
| 197 | Path: path, MemoryName: "session-catalog-v12-test", Migrations: sessionMigrations()[:12], RequireDisk: true, |
| 198 | }) |
| 199 | if err != nil { |
| 200 | t.Fatal(err) |
| 201 | } |
| 202 | t.Cleanup(func() { _ = migrated.DB.Close() }) |
| 203 | catalog := &Catalog{db: migrated.DB, pathIdentity: PathIdentityKey} |
| 204 | var directories, sessions, logFormat int |
| 205 | if err := catalog.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM catalog_directories`).Scan(&directories); err != nil { |
| 206 | t.Fatal(err) |
| 207 | } |
| 208 | if err := catalog.db.QueryRowContext(ctx, `SELECT COUNT(*), MAX(log_format) FROM catalog_sessions`).Scan(&sessions, &logFormat); err != nil { |
| 209 | t.Fatal(err) |
| 210 | } |
| 211 | if directories != 0 || sessions != 1 || logFormat != 1 { |
| 212 | t.Fatalf("after v12: directories=%d sessions=%d log_format=%d, want rescan forced and rows kept as schema 1", directories, sessions, logFormat) |
| 213 | } |
| 214 | if heads, err := catalog.ListHeads(ctx, "/Sessions/Old.jsonl"); err != nil || len(heads) != 0 { |
| 215 | t.Fatalf("legacy row heads = %+v err=%v", heads, err) |
| 216 | } |
| 217 | } |
| 218 |