| 1 | package agent |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "os" |
| 8 | "path/filepath" |
| 9 | "strings" |
| 10 | "testing" |
| 11 | "time" |
| 12 | |
| 13 | "reasonix/internal/provider" |
| 14 | "reasonix/internal/store" |
| 15 | ) |
| 16 | |
| 17 | func dagTestSession(t *testing.T) string { |
| 18 | t.Helper() |
| 19 | return filepath.Join(t.TempDir(), "dag.jsonl") |
| 20 | } |
| 21 | |
| 22 | func dagMsg(role provider.Role, content, id string) provider.Message { |
| 23 | return provider.Message{Role: role, Content: content, ID: id} |
| 24 | } |
| 25 | |
| 26 | func dagMessageEntry(t *testing.T, head, parent, turn string, m provider.Message, at time.Time) sessionDAGEntry { |
| 27 | t.Helper() |
| 28 | e, err := newSessionDAGMessageEntry(head, parent, "", turn, m, at) |
| 29 | if err != nil { |
| 30 | t.Fatalf("message entry: %v", err) |
| 31 | } |
| 32 | return e |
| 33 | } |
| 34 | |
| 35 | func dagAppend(t *testing.T, sessionPath string, entries ...sessionDAGEntry) int64 { |
| 36 | t.Helper() |
| 37 | size, err := appendSessionDAGEntries(sessionPath, entries, false) |
| 38 | if err != nil { |
| 39 | t.Fatalf("append: %v", err) |
| 40 | } |
| 41 | return size |
| 42 | } |
| 43 | |
| 44 | func dagReplay(t *testing.T, sessionPath string) *sessionDAGState { |
| 45 | t.Helper() |
| 46 | st, err := replaySessionDAG(context.Background(), store.SessionEventLog(sessionPath), defaultSessionReplayLimits) |
| 47 | if err != nil { |
| 48 | t.Fatalf("replay: %v", err) |
| 49 | } |
| 50 | return st |
| 51 | } |
| 52 | |
| 53 | func dagContents(msgs []provider.Message) []string { |
| 54 | out := make([]string, 0, len(msgs)) |
| 55 | for _, m := range msgs { |
| 56 | out = append(out, m.Content) |
| 57 | } |
| 58 | return out |
| 59 | } |
| 60 | |
| 61 | // dagLinearLog writes log header + system + user + assistant + user on main. |
| 62 | func dagLinearLog(t *testing.T, sessionPath string) (ids []string, base time.Time) { |
| 63 | t.Helper() |
| 64 | base = time.Date(2026, 1, 8, 10, 0, 0, 0, time.UTC) |
| 65 | msgs := []provider.Message{ |
| 66 | dagMsg(provider.RoleSystem, "sys", "S0"), |
| 67 | dagMsg(provider.RoleUser, "q1", "U1"), |
| 68 | dagMsg(provider.RoleAssistant, "a1", "A1"), |
| 69 | dagMsg(provider.RoleUser, "q2", "U2"), |
| 70 | } |
| 71 | entries := []sessionDAGEntry{{Type: sessionDAGTypeLog, Generation: 1, At: base}} |
| 72 | parent := "" |
| 73 | for i, m := range msgs { |
| 74 | entries = append(entries, dagMessageEntry(t, SessionMainHead, parent, "t1", m, base.Add(time.Duration(i)*time.Second))) |
| 75 | parent = m.ID |
| 76 | ids = append(ids, m.ID) |
| 77 | } |
| 78 | dagAppend(t, sessionPath, entries...) |
| 79 | return ids, base |
| 80 | } |
| 81 | |
| 82 | func TestDAGReplayLinearChainMaterializes(t *testing.T) { |
| 83 | path := dagTestSession(t) |
| 84 | ids, base := dagLinearLog(t, path) |
| 85 | st := dagReplay(t, path) |
| 86 | if st.damaged || st.generation != 1 || st.records != 5 { |
| 87 | t.Fatalf("state damaged=%v generation=%d records=%d", st.damaged, st.generation, st.records) |
| 88 | } |
| 89 | if got := st.selectedHead(); got != SessionMainHead { |
| 90 | t.Fatalf("selectedHead = %q", got) |
| 91 | } |
| 92 | msgs, times := st.materialize(SessionMainHead) |
| 93 | if got := dagContents(msgs); strings.Join(got, ",") != "sys,q1,a1,q2" { |
| 94 | t.Fatalf("materialized %v", got) |
| 95 | } |
| 96 | for i, m := range msgs { |
| 97 | if m.ID != ids[i] { |
| 98 | t.Fatalf("message %d id %q, want %q", i, m.ID, ids[i]) |
| 99 | } |
| 100 | if !times[i].Equal(base.Add(time.Duration(i) * time.Second)) { |
| 101 | t.Fatalf("message %d time %v", i, times[i]) |
| 102 | } |
| 103 | } |
| 104 | if st.heads[SessionMainHead].leaf != "U2" { |
| 105 | t.Fatalf("main leaf = %q", st.heads[SessionMainHead].leaf) |
| 106 | } |
| 107 | } |
| 108 | |
| 109 | func TestDAGForkRewindSelectRetireSemantics(t *testing.T) { |
| 110 | path := dagTestSession(t) |
| 111 | _, base := dagLinearLog(t, path) |
| 112 | // Fork F from A1 and continue it; rewind main back to U1. |
| 113 | dagAppend(t, path, |
| 114 | sessionDAGEntry{Type: sessionDAGTypeFork, Head: SessionMainHead, NewHead: "F", From: "A1", Kind: HeadKindFork, Name: "alt", At: base.Add(10 * time.Second)}, |
| 115 | dagMessageEntry(t, "F", "A1", "t2", dagMsg(provider.RoleUser, "q2-alt", "U2b"), base.Add(11*time.Second)), |
| 116 | sessionDAGEntry{Type: sessionDAGTypeRewind, Head: SessionMainHead, To: "U1", Cause: "test", At: base.Add(12 * time.Second)}, |
| 117 | ) |
| 118 | st := dagReplay(t, path) |
| 119 | if got := dagChain(st, SessionMainHead); strings.Join(got, ",") != "sys,q1" { |
| 120 | t.Fatalf("main after rewind %v", got) |
| 121 | } |
| 122 | if got := dagChain(st, "F"); strings.Join(got, ",") != "sys,q1,a1,q2-alt" { |
| 123 | t.Fatalf("fork chain %v", got) |
| 124 | } |
| 125 | // Newest activity wins: the rewind marker on main is the latest entry. |
| 126 | if got := st.selectedHead(); got != SessionMainHead { |
| 127 | t.Fatalf("selectedHead = %q, want main (newest activity)", got) |
| 128 | } |
| 129 | dagAppend(t, path, sessionDAGEntry{Type: sessionDAGTypeSelect, Head: "F", At: base.Add(13 * time.Second)}) |
| 130 | if got := dagReplay(t, path).selectedHead(); got != "F" { |
| 131 | t.Fatalf("selectedHead after select = %q", got) |
| 132 | } |
| 133 | dagAppend(t, path, sessionDAGEntry{Type: sessionDAGTypeRetire, Head: "F", At: base.Add(14 * time.Second)}) |
| 134 | st = dagReplay(t, path) |
| 135 | if got := st.selectedHead(); got != SessionMainHead { |
| 136 | t.Fatalf("selectedHead after retire = %q", got) |
| 137 | } |
| 138 | heads, err := ListSessionHeads(path) |
| 139 | if err != nil { |
| 140 | t.Fatal(err) |
| 141 | } |
| 142 | if len(heads) != 2 || heads[0].ID != SessionMainHead || heads[1].ID != "F" { |
| 143 | t.Fatalf("heads = %+v", heads) |
| 144 | } |
| 145 | if !heads[1].Retired || heads[1].Name != "alt" || heads[1].ForkFrom != "A1" || heads[1].ParentHead != SessionMainHead || heads[1].MessageCount != 4 { |
| 146 | t.Fatalf("fork head record = %+v", heads[1]) |
| 147 | } |
| 148 | if !heads[0].Selected || heads[0].MessageCount != 2 || heads[0].Kind != HeadKindMain { |
| 149 | t.Fatalf("main head record = %+v", heads[0]) |
| 150 | } |
| 151 | } |
| 152 | |
| 153 | func TestDAGPatchSystemAndRedactOverlays(t *testing.T) { |
| 154 | path := dagTestSession(t) |
| 155 | _, base := dagLinearLog(t, path) |
| 156 | sys, err := encodeSessionDAGMessage(dagMsg(provider.RoleSystem, "sys-v2", "")) |
| 157 | if err != nil { |
| 158 | t.Fatal(err) |
| 159 | } |
| 160 | replacement, err := encodeSessionDAGMessage(dagMsg(provider.RoleAssistant, "[redacted]", "")) |
| 161 | if err != nil { |
| 162 | t.Fatal(err) |
| 163 | } |
| 164 | patched, err := encodeSessionDAGMessage(provider.Message{Role: provider.RoleUser, Content: "q1", Edited: true, WorkDurationMs: 7}) |
| 165 | if err != nil { |
| 166 | t.Fatal(err) |
| 167 | } |
| 168 | dagAppend(t, path, |
| 169 | sessionDAGEntry{Type: sessionDAGTypePatch, Head: SessionMainHead, Target: "U1", Msgs: patched, At: base.Add(20 * time.Second)}, |
| 170 | sessionDAGEntry{Type: sessionDAGTypeSystem, Head: SessionMainHead, Msgs: sys, At: base.Add(21 * time.Second)}, |
| 171 | sessionDAGEntry{Type: sessionDAGTypeRedact, Head: SessionMainHead, Targets: map[string]json.RawMessage{"A1": replacement}, Reason: "secret", At: base.Add(22 * time.Second)}, |
| 172 | ) |
| 173 | st := dagReplay(t, path) |
| 174 | msgs, _ := st.materialize(SessionMainHead) |
| 175 | if msgs[0].Content != "sys-v2" || msgs[0].ID != "S0" { |
| 176 | t.Fatalf("system override = %+v", msgs[0]) |
| 177 | } |
| 178 | if !msgs[1].Edited || msgs[1].WorkDurationMs != 7 || msgs[1].Content != "q1" { |
| 179 | t.Fatalf("patched message = %+v", msgs[1]) |
| 180 | } |
| 181 | if msgs[2].Content != "[redacted]" || msgs[2].ID != "A1" { |
| 182 | t.Fatalf("redacted message = %+v", msgs[2]) |
| 183 | } |
| 184 | // The head's leaf and the chain are unaffected by overlays. |
| 185 | if st.heads[SessionMainHead].leaf != "U2" || len(msgs) != 4 { |
| 186 | t.Fatalf("leaf %q len %d", st.heads[SessionMainHead].leaf, len(msgs)) |
| 187 | } |
| 188 | } |
| 189 | |
| 190 | func TestDAGTornTailStopsAtLastGoodEntryAndRepairsWhenQuiet(t *testing.T) { |
| 191 | path := dagTestSession(t) |
| 192 | dagLinearLog(t, path) |
| 193 | logPath := store.SessionEventLog(path) |
| 194 | good, err := os.Stat(logPath) |
| 195 | if err != nil { |
| 196 | t.Fatal(err) |
| 197 | } |
| 198 | f, err := os.OpenFile(logPath, os.O_WRONLY|os.O_APPEND, 0o600) |
| 199 | if err != nil { |
| 200 | t.Fatal(err) |
| 201 | } |
| 202 | if _, err := f.WriteString(`{"schema_version":2,"type":"message","id":"X","head":"main","msgs":[{"role":"user","con`); err != nil { |
| 203 | t.Fatal(err) |
| 204 | } |
| 205 | f.Close() |
| 206 | st := dagReplay(t, path) |
| 207 | if !st.damaged || st.lastGoodEnd != good.Size()-1 || st.records != 5 { |
| 208 | t.Fatalf("damaged=%v lastGoodEnd=%d good=%d records=%d", st.damaged, st.lastGoodEnd, good.Size(), st.records) |
| 209 | } |
| 210 | if got := dagChain(st, SessionMainHead); strings.Join(got, ",") != "sys,q1,a1,q2" { |
| 211 | t.Fatalf("prefix %v", got) |
| 212 | } |
| 213 | // A young tail may still be another writer's in-progress append. |
| 214 | info, _ := os.Stat(logPath) |
| 215 | if repaired, err := repairSessionDAGTail(path, st, info.ModTime()); err != nil || repaired { |
| 216 | t.Fatalf("young tail repaired=%v err=%v", repaired, err) |
| 217 | } |
| 218 | repaired, err := repairSessionDAGTail(path, st, info.ModTime().Add(sessionDAGTailRepairMinAge+time.Second)) |
| 219 | if err != nil || !repaired { |
| 220 | t.Fatalf("quiet tail repaired=%v err=%v", repaired, err) |
| 221 | } |
| 222 | if _, err := os.Stat(store.SessionEventLogDamaged(path)); err != nil { |
| 223 | t.Fatalf("damaged sidecar: %v", err) |
| 224 | } |
| 225 | after := dagReplay(t, path) |
| 226 | if after.damaged || after.records != 5 { |
| 227 | t.Fatalf("after repair damaged=%v records=%d", after.damaged, after.records) |
| 228 | } |
| 229 | b, _ := os.ReadFile(logPath) |
| 230 | if !strings.HasSuffix(string(b), "\n") || strings.Contains(string(b), `"id":"X"`) { |
| 231 | t.Fatalf("repaired log tail: %q", string(b[len(b)-40:])) |
| 232 | } |
| 233 | } |
| 234 | |
| 235 | func TestDAGDanglingParentBecomesOrphanRoot(t *testing.T) { |
| 236 | path := dagTestSession(t) |
| 237 | base := time.Date(2026, 9, 8, 10, 0, 0, 0, time.UTC) |
| 238 | dagAppend(t, path, |
| 239 | sessionDAGEntry{Type: sessionDAGTypeLog, Generation: 3, RotatedFrom: 2, At: base}, |
| 240 | dagMessageEntry(t, SessionMainHead, "GONE", "", dagMsg(provider.RoleUser, "orphan", "O1"), base), |
| 241 | dagMessageEntry(t, SessionMainHead, "O1", "", dagMsg(provider.RoleAssistant, "child", "O2"), base.Add(time.Second)), |
| 242 | ) |
| 243 | st := dagReplay(t, path) |
| 244 | if st.damaged || len(st.orphans) != 1 || st.orphans[0] != "O1" { |
| 245 | t.Fatalf("damaged=%v orphans=%v", st.damaged, st.orphans) |
| 246 | } |
| 247 | if got := dagChain(st, SessionMainHead); strings.Join(got, ",") != "orphan,child" { |
| 248 | t.Fatalf("chain %v", got) |
| 249 | } |
| 250 | } |
| 251 | |
| 252 | func TestDAGUnknownTypeAndFutureSchemaAreHardErrors(t *testing.T) { |
| 253 | path := dagTestSession(t) |
| 254 | dagAppend(t, path, sessionDAGEntry{Type: sessionDAGTypeLog, Generation: 1}) |
| 255 | logPath := store.SessionEventLog(path) |
| 256 | appendRaw := func(line string) { |
| 257 | t.Helper() |
| 258 | f, err := os.OpenFile(logPath, os.O_WRONLY|os.O_APPEND, 0o600) |
| 259 | if err != nil { |
| 260 | t.Fatal(err) |
| 261 | } |
| 262 | if _, err := f.WriteString(line + "\n"); err != nil { |
| 263 | t.Fatal(err) |
| 264 | } |
| 265 | f.Close() |
| 266 | } |
| 267 | appendRaw(`{"schema_version":2,"type":"teleport","head":"main","at":"2026-09-08T10:00:00Z"}`) |
| 268 | if _, err := replaySessionDAG(context.Background(), logPath, defaultSessionReplayLimits); err == nil || !strings.Contains(err.Error(), `unsupported entry type "teleport"`) { |
| 269 | t.Fatalf("unknown type err = %v", err) |
| 270 | } |
| 271 | if _, err := LoadSession(path); err == nil { |
| 272 | t.Fatal("LoadSession must refuse a log with an unknown entry type") |
| 273 | } |
| 274 | if err := os.WriteFile(logPath, []byte(`{"schema_version":3,"type":"log","at":"2026-09-08T10:00:00Z"}`+"\n"), 0o600); err != nil { |
| 275 | t.Fatal(err) |
| 276 | } |
| 277 | probe, err := probeSessionEventLog(path) |
| 278 | if err != nil || !probe.futureSchema || probe.dag || probe.native { |
| 279 | t.Fatalf("schema 3 probe = %+v err=%v", probe, err) |
| 280 | } |
| 281 | if _, err := LoadSession(path); err == nil || !strings.Contains(err.Error(), "schema 3") { |
| 282 | t.Fatalf("schema 3 load err = %v", err) |
| 283 | } |
| 284 | } |
| 285 | |
| 286 | func TestDAGReplayHonorsRecordBudget(t *testing.T) { |
| 287 | path := dagTestSession(t) |
| 288 | dagLinearLog(t, path) |
| 289 | limits := defaultSessionReplayLimits |
| 290 | limits.maxRecords = 3 |
| 291 | _, err := replaySessionDAG(context.Background(), store.SessionEventLog(path), limits) |
| 292 | if !errors.Is(err, ErrSessionReplayLimitExceeded) { |
| 293 | t.Fatalf("err = %v, want replay limit", err) |
| 294 | } |
| 295 | } |
| 296 | |
| 297 | func TestLoadSessionReadsSelectedHeadOfDAGLog(t *testing.T) { |
| 298 | path := dagTestSession(t) |
| 299 | ids, base := dagLinearLog(t, path) |
| 300 | dagAppend(t, path, |
| 301 | sessionDAGEntry{Type: sessionDAGTypeFork, Head: SessionMainHead, NewHead: "F", From: "A1", Kind: HeadKindConcurrent, At: base.Add(30 * time.Second)}, |
| 302 | dagMessageEntry(t, "F", "A1", "", dagMsg(provider.RoleUser, "from-other-writer", "U9"), base.Add(31*time.Second)), |
| 303 | ) |
| 304 | probe, err := probeSessionEventLog(path) |
| 305 | if err != nil || !probe.dag || probe.native || probe.futureSchema { |
| 306 | t.Fatalf("probe = %+v err=%v", probe, err) |
| 307 | } |
| 308 | s, err := LoadSession(path) |
| 309 | if err != nil { |
| 310 | t.Fatalf("LoadSession: %v", err) |
| 311 | } |
| 312 | if got := dagContents(s.Messages); strings.Join(got, ",") != "sys,q1,a1,from-other-writer" { |
| 313 | t.Fatalf("loaded newest head %v", got) |
| 314 | } |
| 315 | ref, ok := s.Head() |
| 316 | if !ok || ref.HeadID != "F" || ref.LeafID != "U9" || ref.LogGeneration != 1 || ref.LogOffset <= 0 { |
| 317 | t.Fatalf("head ref = %+v ok=%v", ref, ok) |
| 318 | } |
| 319 | if s.Messages[1].ID != ids[1] || s.LeafID() != "U9" { |
| 320 | t.Fatalf("ids not preserved: %q leaf %q", s.Messages[1].ID, s.LeafID()) |
| 321 | } |
| 322 | users, err := LoadSessionUserMessages(path) |
| 323 | if err != nil || len(users) != 2 || users[1].Message.Content != "from-other-writer" || !users[1].At.Equal(base.Add(31*time.Second)) { |
| 324 | t.Fatalf("user messages = %+v err=%v", users, err) |
| 325 | } |
| 326 | // A save extends the loaded head in place: the log grows by the new entry, |
| 327 | // the selected-head cache follows, and no transcript copy appears. |
| 328 | before, _ := os.ReadFile(store.SessionEventLog(path)) |
| 329 | s.Add(dagMsg(provider.RoleAssistant, "reply", "")) |
| 330 | if err := s.Save(path); err != nil { |
| 331 | t.Fatalf("Save: %v", err) |
| 332 | } |
| 333 | after, _ := os.ReadFile(store.SessionEventLog(path)) |
| 334 | if !strings.HasPrefix(string(after), string(before)) || len(after) == len(before) { |
| 335 | t.Fatal("save must append to the schema-2 log without rewriting it") |
| 336 | } |
| 337 | if b, err := os.ReadFile(path); err != nil || !strings.Contains(string(b), `"reply"`) { |
| 338 | t.Fatalf("checkpoint not written: %v", err) |
| 339 | } |
| 340 | reloaded, err := LoadSession(path) |
| 341 | if err != nil || reloaded.LeafID() != s.LeafID() || len(reloaded.Messages) != 5 { |
| 342 | t.Fatalf("reload after save: err=%v leaf %q vs %q len %d", err, reloaded.LeafID(), s.LeafID(), len(reloaded.Messages)) |
| 343 | } |
| 344 | entries, _ := os.ReadDir(filepath.Dir(path)) |
| 345 | for _, entry := range entries { |
| 346 | if store.IsSessionTranscriptName(entry.Name()) && entry.Name() != filepath.Base(path) { |
| 347 | t.Fatalf("save created a transcript copy: %s", entry.Name()) |
| 348 | } |
| 349 | } |
| 350 | } |
| 351 | |
| 352 | func TestSchemaOneReaderRefusesDAGLogWithoutTruncating(t *testing.T) { |
| 353 | path := dagTestSession(t) |
| 354 | dagLinearLog(t, path) |
| 355 | logPath := store.SessionEventLog(path) |
| 356 | before, err := os.ReadFile(logPath) |
| 357 | if err != nil { |
| 358 | t.Fatal(err) |
| 359 | } |
| 360 | if _, err := replaySessionEventLog(logPath); err == nil || !strings.Contains(err.Error(), "unsupported schema version 2") { |
| 361 | t.Fatalf("schema-1 replay err = %v", err) |
| 362 | } |
| 363 | if err := repairSessionEventLogTail(path); err == nil { |
| 364 | t.Fatal("schema-1 tail repair must fail closed on a schema-2 log") |
| 365 | } |
| 366 | after, err := os.ReadFile(logPath) |
| 367 | if err != nil { |
| 368 | t.Fatal(err) |
| 369 | } |
| 370 | if string(before) != string(after) { |
| 371 | t.Fatal("schema-1 tail repair modified a schema-2 log") |
| 372 | } |
| 373 | } |
| 374 | |
| 375 | func dagChain(st *sessionDAGState, head string) []string { |
| 376 | msgs, _ := st.materialize(head) |
| 377 | return dagContents(msgs) |
| 378 | } |
| 379 | |
| 380 | // useSchemaOneLog pins a test to the schema-1 writer: it exercises mechanics |
| 381 | // (replace records, revision CAS, recovery copies) that only that path has. |
| 382 | func useSchemaOneLog(t *testing.T) { |
| 383 | t.Helper() |
| 384 | t.Setenv(SessionLogSchemaEnv, "v1") |
| 385 | } |
| 386 | |
| 387 | func schemaOneTempDir(t *testing.T) string { |
| 388 | t.Helper() |
| 389 | useSchemaOneLog(t) |
| 390 | return t.TempDir() |
| 391 | } |
| 392 | |
| 393 | func schemaOneSessionPath(t *testing.T, name string) string { |
| 394 | t.Helper() |
| 395 | return filepath.Join(schemaOneTempDir(t), name) |
| 396 | } |
| 397 |