| 1 | package agent |
| 2 | |
| 3 | import ( |
| 4 | "fmt" |
| 5 | "math/rand" |
| 6 | "os" |
| 7 | "path/filepath" |
| 8 | "strings" |
| 9 | "testing" |
| 10 | |
| 11 | "reasonix/internal/fileutil" |
| 12 | "reasonix/internal/provider" |
| 13 | "reasonix/internal/store" |
| 14 | ) |
| 15 | |
| 16 | // Crash-consistency model suite: a crash is injected at every durable |
| 17 | // boundary and recovery must never lose a durable descendant, pick sides |
| 18 | // silently, fabricate a chimera, or change on a second recovery. |
| 19 | |
| 20 | type crashSentinel struct{ n int } |
| 21 | |
| 22 | type durabilityRun struct { |
| 23 | t *testing.T |
| 24 | dir string |
| 25 | path string |
| 26 | } |
| 27 | |
| 28 | func newDurabilityRun(t *testing.T) *durabilityRun { |
| 29 | t.Helper() |
| 30 | dir := t.TempDir() |
| 31 | return &durabilityRun{t: t, dir: dir, path: filepath.Join(dir, "session.jsonl")} |
| 32 | } |
| 33 | |
| 34 | func (d *durabilityRun) turn(i int) []provider.Message { |
| 35 | return []provider.Message{ |
| 36 | {Role: provider.RoleUser, Content: fmt.Sprintf("ask %d", i)}, |
| 37 | {Role: provider.RoleAssistant, Content: fmt.Sprintf("answer %d", i)}, |
| 38 | } |
| 39 | } |
| 40 | |
| 41 | // countBoundaries dry-runs fn with a counting hook and returns the ordered op |
| 42 | // names of every durable boundary it crossed. |
| 43 | func (d *durabilityRun) countBoundaries(fn func()) []string { |
| 44 | var ops []string |
| 45 | fileutil.CrashPoint = func(op, path string) { |
| 46 | if strings.HasPrefix(path, d.dir) { |
| 47 | ops = append(ops, op) |
| 48 | } |
| 49 | } |
| 50 | defer func() { fileutil.CrashPoint = nil }() |
| 51 | fn() |
| 52 | return ops |
| 53 | } |
| 54 | |
| 55 | // crashAt injects a panic at the nth durable boundary under the run's dir and |
| 56 | // reports whether fn actually crashed there. |
| 57 | func (d *durabilityRun) crashAt(n int, fn func()) (crashed bool) { |
| 58 | count := 0 |
| 59 | fileutil.CrashPoint = func(op, path string) { |
| 60 | if !strings.HasPrefix(path, d.dir) { |
| 61 | return |
| 62 | } |
| 63 | count++ |
| 64 | if count == n { |
| 65 | panic(crashSentinel{n}) |
| 66 | } |
| 67 | } |
| 68 | defer func() { fileutil.CrashPoint = nil }() |
| 69 | defer func() { |
| 70 | if r := recover(); r != nil { |
| 71 | if _, ok := r.(crashSentinel); !ok { |
| 72 | panic(r) |
| 73 | } |
| 74 | crashed = true |
| 75 | } |
| 76 | }() |
| 77 | fn() |
| 78 | return false |
| 79 | } |
| 80 | |
| 81 | func (d *durabilityRun) countRecoveryFiles() int { |
| 82 | entries, _ := os.ReadDir(d.dir) |
| 83 | n := 0 |
| 84 | for _, e := range entries { |
| 85 | if strings.Contains(e.Name(), "recovery") && strings.HasSuffix(e.Name(), ".jsonl") && |
| 86 | !strings.HasSuffix(e.Name(), ".events.jsonl") { |
| 87 | n++ |
| 88 | } |
| 89 | } |
| 90 | return n |
| 91 | } |
| 92 | |
| 93 | func mustDigest(t *testing.T, msgs []provider.Message) string { |
| 94 | t.Helper() |
| 95 | digest, err := digestSessionMessages(msgs) |
| 96 | if err != nil { |
| 97 | t.Fatalf("digest: %v", err) |
| 98 | } |
| 99 | return digestString(digest) |
| 100 | } |
| 101 | |
| 102 | // recoverAndCheck loads the session twice (I4) and verifies the recovered |
| 103 | // transcript sits between lastSaved and pending in prefix order for appends |
| 104 | // (I1: no rollback below the durable floor; I3: never a chimera beyond |
| 105 | // pending), or equals one of the two endpoints for rewrites (I2/I3). |
| 106 | func (d *durabilityRun) recoverAndCheck(lastSaved, pending []provider.Message, rewrite bool, label string) []provider.Message { |
| 107 | d.t.Helper() |
| 108 | branchesBefore := d.countRecoveryFiles() |
| 109 | s1, err := LoadSession(d.path) |
| 110 | if err != nil { |
| 111 | // A crash before anything ever became durable legitimately leaves no |
| 112 | // session file; the empty floor lost nothing. |
| 113 | if len(lastSaved) == 0 && os.IsNotExist(err) { |
| 114 | return nil |
| 115 | } |
| 116 | d.t.Fatalf("%s: recovery load failed: %v", label, err) |
| 117 | } |
| 118 | s2, err := LoadSession(d.path) |
| 119 | if err != nil { |
| 120 | d.t.Fatalf("%s: second recovery load failed: %v", label, err) |
| 121 | } |
| 122 | if mustDigest(d.t, s1.Messages) != mustDigest(d.t, s2.Messages) { |
| 123 | d.t.Fatalf("%s: recovery not idempotent — two loads disagree", label) |
| 124 | } |
| 125 | if after := d.countRecoveryFiles(); after != branchesBefore { |
| 126 | d.t.Fatalf("%s: pure loads changed recovery-branch count %d→%d", label, branchesBefore, after) |
| 127 | } |
| 128 | got := s1.Messages |
| 129 | if rewrite { |
| 130 | if !messagesEqualForStorageList(got, lastSaved) && !messagesEqualForStorageList(got, pending) { |
| 131 | d.t.Fatalf("%s: rewrite recovery produced a state that is neither endpoint (got %d msgs, endpoints %d/%d)", |
| 132 | label, len(got), len(lastSaved), len(pending)) |
| 133 | } |
| 134 | return got |
| 135 | } |
| 136 | if !messagesHavePrefixWithCompatibleSystem(got, lastSaved) { |
| 137 | d.t.Fatalf("%s: recovery rolled back below the durable floor (got %d msgs, floor %d) — invariant 1 violated", |
| 138 | label, len(got), len(lastSaved)) |
| 139 | } |
| 140 | if !messagesHavePrefixWithCompatibleSystem(pending, got) { |
| 141 | d.t.Fatalf("%s: recovery produced a chimera beyond the pending save (got %d msgs, pending %d) — invariant 3 violated", |
| 142 | label, len(got), len(pending)) |
| 143 | } |
| 144 | return got |
| 145 | } |
| 146 | |
| 147 | // buildSaved replays i committed turns into a fresh session file and returns |
| 148 | // the live session plus its durable transcript. |
| 149 | func (d *durabilityRun) buildSaved(turns int) (*Session, []provider.Message) { |
| 150 | d.t.Helper() |
| 151 | s := NewSession("system prompt") |
| 152 | for i := 1; i <= turns; i++ { |
| 153 | for _, m := range d.turn(i) { |
| 154 | s.Add(m) |
| 155 | } |
| 156 | if err := s.SaveSnapshot(d.path); err != nil { |
| 157 | d.t.Fatalf("seed save %d: %v", i, err) |
| 158 | } |
| 159 | } |
| 160 | return s, append([]provider.Message(nil), s.Messages...) |
| 161 | } |
| 162 | |
| 163 | func TestDurabilityCrashSweepAppendSave(t *testing.T) { |
| 164 | probe := newDurabilityRun(t) |
| 165 | s, _ := probe.buildSaved(1) |
| 166 | for _, m := range probe.turn(2) { |
| 167 | s.Add(m) |
| 168 | } |
| 169 | ops := probe.countBoundaries(func() { |
| 170 | if err := s.SaveSnapshot(probe.path); err != nil { |
| 171 | t.Fatalf("probe save: %v", err) |
| 172 | } |
| 173 | }) |
| 174 | if len(ops) == 0 { |
| 175 | t.Fatal("save crossed no durable boundaries — seam broken") |
| 176 | } |
| 177 | walIdx := -1 |
| 178 | for i, op := range ops { |
| 179 | if op == "wal-append" { |
| 180 | walIdx = i |
| 181 | } |
| 182 | } |
| 183 | t.Logf("append-save boundaries: %v (wal at %d)", ops, walIdx) |
| 184 | |
| 185 | for n := 1; n <= len(ops); n++ { |
| 186 | d := newDurabilityRun(t) |
| 187 | live, saved := d.buildSaved(1) |
| 188 | for _, m := range d.turn(2) { |
| 189 | live.Add(m) |
| 190 | } |
| 191 | pending := append([]provider.Message(nil), live.Messages...) |
| 192 | if !d.crashAt(n, func() { _ = live.SaveSnapshot(d.path) }) { |
| 193 | t.Fatalf("boundary %d: crash did not fire", n) |
| 194 | } |
| 195 | got := d.recoverAndCheck(saved, pending, false, fmt.Sprintf("boundary %d/%d (%s)", n, len(ops), ops[n-1])) |
| 196 | // The WAL is authoritative: once the append event is durable, recovery |
| 197 | // must yield the pending transcript even if the checkpoint never landed. |
| 198 | if walIdx >= 0 && n > walIdx+1 && !messagesEqualForStorageList(got, pending) { |
| 199 | t.Fatalf("boundary %d (%s): WAL was durable but recovery returned %d msgs instead of pending %d", |
| 200 | n, ops[n-1], len(got), len(pending)) |
| 201 | } |
| 202 | } |
| 203 | } |
| 204 | |
| 205 | func TestDurabilityCheckpointWithoutLedgerHeals(t *testing.T) { |
| 206 | probe := newDurabilityRun(t) |
| 207 | s, _ := probe.buildSaved(1) |
| 208 | for _, m := range probe.turn(2) { |
| 209 | s.Add(m) |
| 210 | } |
| 211 | ops := probe.countBoundaries(func() { _ = s.SaveSnapshot(probe.path) }) |
| 212 | // Crash on the LAST boundary: everything before it (WAL + checkpoint) is |
| 213 | // durable, the trailing ledger/index write is not. |
| 214 | n := len(ops) |
| 215 | d := newDurabilityRun(t) |
| 216 | live, _ := d.buildSaved(1) |
| 217 | for _, m := range d.turn(2) { |
| 218 | live.Add(m) |
| 219 | } |
| 220 | pending := append([]provider.Message(nil), live.Messages...) |
| 221 | if !d.crashAt(n, func() { _ = live.SaveSnapshot(d.path) }) { |
| 222 | t.Fatalf("crash at final boundary did not fire (ops=%v)", ops) |
| 223 | } |
| 224 | branches := d.countRecoveryFiles() |
| 225 | loaded, err := LoadSession(d.path) |
| 226 | if err != nil { |
| 227 | t.Fatalf("recovery load: %v", err) |
| 228 | } |
| 229 | if !messagesEqualForStorageList(loaded.Messages, pending) { |
| 230 | t.Fatalf("recovery after ledger-less checkpoint returned %d msgs, want pending %d", len(loaded.Messages), len(pending)) |
| 231 | } |
| 232 | // Healing save: continue on the recovered session without forking a branch. |
| 233 | for _, m := range d.turn(3) { |
| 234 | loaded.Add(m) |
| 235 | } |
| 236 | if err := loaded.SaveSnapshot(d.path); err != nil { |
| 237 | t.Fatalf("post-recovery save must heal, got: %v", err) |
| 238 | } |
| 239 | if got := d.countRecoveryFiles(); got != branches { |
| 240 | t.Fatalf("post-recovery save forked a recovery branch (%d→%d) instead of healing", branches, got) |
| 241 | } |
| 242 | } |
| 243 | |
| 244 | func TestDurabilityTornWALTailReplaysToLastCommit(t *testing.T) { |
| 245 | d := newDurabilityRun(t) |
| 246 | _, saved := d.buildSaved(2) |
| 247 | wal := d.path[:len(d.path)-len(".jsonl")] + ".events.jsonl" |
| 248 | if _, err := os.Stat(wal); err != nil { |
| 249 | // Resolve the actual event-log path via the store layout if it differs. |
| 250 | matches, _ := filepath.Glob(filepath.Join(d.dir, "*.events.jsonl")) |
| 251 | if len(matches) != 1 { |
| 252 | t.Fatalf("cannot locate WAL (stat %v, glob %v)", err, matches) |
| 253 | } |
| 254 | wal = matches[0] |
| 255 | } |
| 256 | f, err := os.OpenFile(wal, os.O_WRONLY|os.O_APPEND, 0o600) |
| 257 | if err != nil { |
| 258 | t.Fatalf("open WAL: %v", err) |
| 259 | } |
| 260 | if _, err := f.WriteString(`{"schema_version":1,"type":"append","messages":[{"role":"u`); err != nil { |
| 261 | t.Fatalf("tear WAL: %v", err) |
| 262 | } |
| 263 | f.Close() |
| 264 | got := d.recoverAndCheck(saved, saved, false, "torn WAL tail") |
| 265 | if !messagesEqualForStorageList(got, saved) { |
| 266 | t.Fatalf("torn tail recovery returned %d msgs, want last clean commit %d", len(got), len(saved)) |
| 267 | } |
| 268 | } |
| 269 | |
| 270 | func TestDurabilityStaleWriterCannotClobber(t *testing.T) { |
| 271 | useSchemaOneLog(t) |
| 272 | d := newDurabilityRun(t) |
| 273 | _, _ = d.buildSaved(1) |
| 274 | |
| 275 | a, err := LoadSession(d.path) |
| 276 | if err != nil { |
| 277 | t.Fatalf("load A: %v", err) |
| 278 | } |
| 279 | b, err := LoadSession(d.path) |
| 280 | if err != nil { |
| 281 | t.Fatalf("load B: %v", err) |
| 282 | } |
| 283 | for _, m := range d.turn(2) { |
| 284 | b.Add(m) |
| 285 | } |
| 286 | if err := b.SaveSnapshot(d.path); err != nil { |
| 287 | t.Fatalf("B save: %v", err) |
| 288 | } |
| 289 | winner := append([]provider.Message(nil), b.Messages...) |
| 290 | |
| 291 | a.Add(provider.Message{Role: provider.RoleUser, Content: "diverged ask"}) |
| 292 | a.Add(provider.Message{Role: provider.RoleAssistant, Content: "diverged answer"}) |
| 293 | saveErr := a.SaveSnapshot(d.path) |
| 294 | |
| 295 | loaded, err := LoadSession(d.path) |
| 296 | if err != nil { |
| 297 | t.Fatalf("reload: %v", err) |
| 298 | } |
| 299 | if saveErr == nil { |
| 300 | // A stale diverged writer may be redirected, never silently accepted |
| 301 | // over B: the main path must still be B's descendant. |
| 302 | if !messagesHavePrefixWithCompatibleSystem(loaded.Messages, winner) { |
| 303 | t.Fatalf("stale writer clobbered the newer transcript: main path %d msgs no longer extends winner %d", |
| 304 | len(loaded.Messages), len(winner)) |
| 305 | } |
| 306 | return |
| 307 | } |
| 308 | if _, ok := SnapshotConflictKind(saveErr); !ok { |
| 309 | t.Fatalf("stale save failed with a non-conflict error: %v", saveErr) |
| 310 | } |
| 311 | if !messagesEqualForStorageList(loaded.Messages, winner) { |
| 312 | t.Fatalf("conflict was reported but main path changed anyway (%d msgs, want %d)", len(loaded.Messages), len(winner)) |
| 313 | } |
| 314 | } |
| 315 | |
| 316 | func TestDurabilityBareSaveBootstrapsWAL(t *testing.T) { |
| 317 | useSchemaOneLog(t) |
| 318 | d := newDurabilityRun(t) |
| 319 | s := NewSession("system prompt") |
| 320 | s.Add(provider.Message{Role: provider.RoleUser, Content: "bare save"}) |
| 321 | if err := s.Save(d.path); err != nil { |
| 322 | t.Fatalf("bare Save: %v", err) |
| 323 | } |
| 324 | probe, err := probeSessionEventLog(d.path) |
| 325 | if err != nil { |
| 326 | t.Fatalf("probe WAL: %v", err) |
| 327 | } |
| 328 | if !probe.native || probe.size == 0 { |
| 329 | t.Fatalf("bare Save did not bootstrap a native WAL: %+v", probe) |
| 330 | } |
| 331 | loaded, err := LoadSession(d.path) |
| 332 | if err != nil { |
| 333 | t.Fatalf("reload bare Save: %v", err) |
| 334 | } |
| 335 | if !messagesEqualForStorageList(loaded.Messages, s.Messages) { |
| 336 | t.Fatalf("bare Save round trip changed transcript: got %d want %d messages", len(loaded.Messages), len(s.Messages)) |
| 337 | } |
| 338 | if _, err := os.Stat(store.SessionEventLog(d.path)); err != nil { |
| 339 | t.Fatalf("bare Save WAL missing: %v", err) |
| 340 | } |
| 341 | } |
| 342 | |
| 343 | func TestDurabilityCrossWriterIDCannotClobber(t *testing.T) { |
| 344 | useSchemaOneLog(t) |
| 345 | originalWriterID := sessionWriterID |
| 346 | t.Cleanup(func() { sessionWriterID = originalWriterID }) |
| 347 | |
| 348 | d := newDurabilityRun(t) |
| 349 | sessionWriterID = "writer-a" |
| 350 | a := NewSession("system prompt") |
| 351 | a.Add(provider.Message{Role: provider.RoleUser, Content: "base"}) |
| 352 | if err := a.SaveSnapshot(d.path); err != nil { |
| 353 | t.Fatalf("writer A seed save: %v", err) |
| 354 | } |
| 355 | a, err := LoadSession(d.path) |
| 356 | if err != nil { |
| 357 | t.Fatalf("writer A load: %v", err) |
| 358 | } |
| 359 | |
| 360 | sessionWriterID = "writer-b" |
| 361 | b, err := LoadSession(d.path) |
| 362 | if err != nil { |
| 363 | t.Fatalf("writer B load: %v", err) |
| 364 | } |
| 365 | b.Add(provider.Message{Role: provider.RoleAssistant, Content: "newer writer B"}) |
| 366 | if err := b.SaveSnapshot(d.path); err != nil { |
| 367 | t.Fatalf("writer B save: %v", err) |
| 368 | } |
| 369 | winner := b.Snapshot() |
| 370 | |
| 371 | sessionWriterID = "writer-a" |
| 372 | a.Add(provider.Message{Role: provider.RoleAssistant, Content: "stale writer A"}) |
| 373 | err = a.SaveSnapshot(d.path) |
| 374 | if err == nil { |
| 375 | t.Fatal("cross-writer stale save unexpectedly succeeded") |
| 376 | } |
| 377 | if _, ok := SnapshotConflictKind(err); !ok { |
| 378 | t.Fatalf("cross-writer stale save error = %v, want snapshot conflict", err) |
| 379 | } |
| 380 | loaded, err := LoadSession(d.path) |
| 381 | if err != nil { |
| 382 | t.Fatalf("reload cross-writer winner: %v", err) |
| 383 | } |
| 384 | if !messagesEqualForStorageList(loaded.Messages, winner) { |
| 385 | t.Fatalf("cross-writer stale save clobbered winner: got %d want %d messages", len(loaded.Messages), len(winner)) |
| 386 | } |
| 387 | } |
| 388 | |
| 389 | func TestDurabilityStaleCompactRewriteCannotClobber(t *testing.T) { |
| 390 | d := newDurabilityRun(t) |
| 391 | _, _ = d.buildSaved(1) |
| 392 | |
| 393 | stale, err := LoadSession(d.path) |
| 394 | if err != nil { |
| 395 | t.Fatalf("load stale session: %v", err) |
| 396 | } |
| 397 | newer, err := LoadSession(d.path) |
| 398 | if err != nil { |
| 399 | t.Fatalf("load newer session: %v", err) |
| 400 | } |
| 401 | newer.Add(provider.Message{Role: provider.RoleUser, Content: "newer durable turn"}) |
| 402 | if err := newer.SaveSnapshot(d.path); err != nil { |
| 403 | t.Fatalf("newer save: %v", err) |
| 404 | } |
| 405 | winner := append([]provider.Message(nil), newer.Messages...) |
| 406 | |
| 407 | stale.Replace(append([]provider.Message(nil), stale.Messages...)) |
| 408 | err = stale.SaveRewriteCompact(d.path) |
| 409 | if err == nil { |
| 410 | t.Fatal("stale compact rewrite unexpectedly succeeded") |
| 411 | } |
| 412 | if _, ok := SnapshotConflictKind(err); !ok { |
| 413 | t.Fatalf("stale compact rewrite error = %v, want snapshot conflict", err) |
| 414 | } |
| 415 | loaded, err := LoadSession(d.path) |
| 416 | if err != nil { |
| 417 | t.Fatalf("reload winner: %v", err) |
| 418 | } |
| 419 | if !messagesEqualForStorageList(loaded.Messages, winner) { |
| 420 | t.Fatalf("stale compact rewrite clobbered winner: got %d want %d messages", len(loaded.Messages), len(winner)) |
| 421 | } |
| 422 | } |
| 423 | |
| 424 | func TestDurabilityRewindSuffixDoesNotResurrect(t *testing.T) { |
| 425 | useSchemaOneLog(t) |
| 426 | d := newDurabilityRun(t) |
| 427 | _, _ = d.buildSaved(3) |
| 428 | |
| 429 | a, err := LoadSession(d.path) |
| 430 | if err != nil { |
| 431 | t.Fatalf("load A: %v", err) |
| 432 | } |
| 433 | b, err := LoadSession(d.path) |
| 434 | if err != nil { |
| 435 | t.Fatalf("load B: %v", err) |
| 436 | } |
| 437 | // B performs an intentional rewind to one turn and commits it. |
| 438 | short := append([]provider.Message(nil), b.Messages[:3]...) // system + turn 1 |
| 439 | b.Rewrite(short, "rewind") |
| 440 | if err := b.SaveRewrite(d.path); err != nil { |
| 441 | t.Fatalf("B rewind save: %v", err) |
| 442 | } |
| 443 | // A, still holding the long pre-rewind transcript, keeps appending. |
| 444 | a.Add(provider.Message{Role: provider.RoleUser, Content: "stale continuation"}) |
| 445 | _ = a.SaveSnapshot(d.path) |
| 446 | |
| 447 | loaded, err := LoadSession(d.path) |
| 448 | if err != nil { |
| 449 | t.Fatalf("reload: %v", err) |
| 450 | } |
| 451 | if messagesHavePrefixWithCompatibleSystem(loaded.Messages, a.Messages) && len(loaded.Messages) >= len(a.Messages) { |
| 452 | t.Fatalf("rewound suffix resurrected on the main path (%d msgs)", len(loaded.Messages)) |
| 453 | } |
| 454 | } |
| 455 | |
| 456 | func TestDurabilityStaleInFlightCompareAndClear(t *testing.T) { |
| 457 | d := newDurabilityRun(t) |
| 458 | _, _ = d.buildSaved(1) |
| 459 | old, err := BeginSessionInFlightTurn(d.path, 1, false) |
| 460 | if err != nil { |
| 461 | t.Fatalf("begin old turn: %v", err) |
| 462 | } |
| 463 | fresh, err := BeginSessionInFlightTurn(d.path, 3, false) |
| 464 | if err != nil { |
| 465 | t.Fatalf("begin fresh turn: %v", err) |
| 466 | } |
| 467 | cleared, err := ClearSessionInFlightTurnIfMatch(d.path, old) |
| 468 | if err != nil { |
| 469 | t.Fatalf("compare-and-clear: %v", err) |
| 470 | } |
| 471 | if cleared { |
| 472 | t.Fatal("stale turn cleared the fresh turn's marker — compare-and-clear broken") |
| 473 | } |
| 474 | cleared, err = ClearSessionInFlightTurnIfMatch(d.path, fresh) |
| 475 | if err != nil || !cleared { |
| 476 | t.Fatalf("owner clear failed: cleared=%v err=%v", cleared, err) |
| 477 | } |
| 478 | } |
| 479 | |
| 480 | func TestDurabilityFuzzCrashConsistency(t *testing.T) { |
| 481 | if testing.Short() { |
| 482 | t.Skip("fuzz sweep skipped in -short") |
| 483 | } |
| 484 | for seed := int64(1); seed <= 20; seed++ { |
| 485 | t.Run(fmt.Sprintf("seed%02d", seed), func(t *testing.T) { |
| 486 | rng := rand.New(rand.NewSource(seed)) |
| 487 | steps := 2 + rng.Intn(5) |
| 488 | crashStep := 1 + rng.Intn(steps) |
| 489 | |
| 490 | type stepKind int |
| 491 | const ( |
| 492 | kindAppend stepKind = iota |
| 493 | kindRewrite |
| 494 | ) |
| 495 | kinds := make([]stepKind, steps) |
| 496 | for i := range kinds { |
| 497 | if rng.Intn(10) < 8 || i == 0 { |
| 498 | kinds[i] = kindAppend |
| 499 | } else { |
| 500 | kinds[i] = kindRewrite |
| 501 | } |
| 502 | } |
| 503 | |
| 504 | apply := func(s *Session, i int) { |
| 505 | switch kinds[i] { |
| 506 | case kindAppend: |
| 507 | s.Add(provider.Message{Role: provider.RoleUser, Content: fmt.Sprintf("s%d ask %d", seed, i)}) |
| 508 | s.Add(provider.Message{Role: provider.RoleAssistant, Content: fmt.Sprintf("s%d answer %d", seed, i)}) |
| 509 | case kindRewrite: |
| 510 | keep := 1 + len(s.Messages)/2 |
| 511 | s.Rewrite(append([]provider.Message(nil), s.Messages[:keep]...), "compact") |
| 512 | } |
| 513 | } |
| 514 | save := func(s *Session, i int, path string) error { |
| 515 | if kinds[i] == kindRewrite { |
| 516 | return s.SaveRewrite(path) |
| 517 | } |
| 518 | return s.SaveSnapshot(path) |
| 519 | } |
| 520 | |
| 521 | // Dry run to count the crash step's boundaries. |
| 522 | probe := newDurabilityRun(t) |
| 523 | ps := NewSession("system prompt") |
| 524 | for i := range crashStep - 1 { |
| 525 | apply(ps, i) |
| 526 | if err := save(ps, i, probe.path); err != nil { |
| 527 | t.Fatalf("probe step %d: %v", i, err) |
| 528 | } |
| 529 | } |
| 530 | apply(ps, crashStep-1) |
| 531 | ops := probe.countBoundaries(func() { _ = save(ps, crashStep-1, probe.path) }) |
| 532 | if len(ops) == 0 { |
| 533 | t.Skip("crash step crossed no boundaries") |
| 534 | } |
| 535 | boundary := 1 + rng.Intn(len(ops)) |
| 536 | |
| 537 | d := newDurabilityRun(t) |
| 538 | s := NewSession("system prompt") |
| 539 | for i := range crashStep - 1 { |
| 540 | apply(s, i) |
| 541 | if err := save(s, i, d.path); err != nil { |
| 542 | t.Fatalf("step %d: %v", i, err) |
| 543 | } |
| 544 | } |
| 545 | var lastSaved []provider.Message |
| 546 | if crashStep > 1 { |
| 547 | lastSaved = append(lastSaved, s.Messages...) |
| 548 | } |
| 549 | apply(s, crashStep-1) |
| 550 | pending := append([]provider.Message(nil), s.Messages...) |
| 551 | if !d.crashAt(boundary, func() { _ = save(s, crashStep-1, d.path) }) { |
| 552 | t.Fatalf("crash at boundary %d/%d did not fire", boundary, len(ops)) |
| 553 | } |
| 554 | d.recoverAndCheck(lastSaved, pending, kinds[crashStep-1] == kindRewrite, |
| 555 | fmt.Sprintf("seed %d step %d boundary %d/%d (%s)", seed, crashStep, boundary, len(ops), ops[boundary-1])) |
| 556 | }) |
| 557 | } |
| 558 | } |
| 559 |