| 1 | package session |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "fmt" |
| 7 | "os" |
| 8 | "path/filepath" |
| 9 | "testing" |
| 10 | |
| 11 | "reasonix/internal/provider" |
| 12 | ) |
| 13 | |
| 14 | func TestHistoryReplacementPreservesMessageVersions(t *testing.T) { |
| 15 | for _, incremental := range []bool{false, true} { |
| 16 | name := "rebuild" |
| 17 | if incremental { |
| 18 | name = "incremental" |
| 19 | } |
| 20 | t.Run(name, func(t *testing.T) { |
| 21 | root := filepath.Join(t.TempDir(), "sessions") |
| 22 | service, err := NewService("local", NewFilesystemPersistence(root)) |
| 23 | if err != nil { |
| 24 | t.Fatal(err) |
| 25 | } |
| 26 | t.Cleanup(func() { _ = service.CloseAll(context.Background()) }) |
| 27 | runtime, err := service.Create(t.Context(), CreateOptions{SessionID: "replace"}) |
| 28 | if err != nil { |
| 29 | t.Fatal(err) |
| 30 | } |
| 31 | appendWindowMessages(t, runtime, "retained", "removed") |
| 32 | query, ref := service.Query(), runtime.Ref() |
| 33 | read := func() HistoryWindowPage { |
| 34 | t.Helper() |
| 35 | // Prepare synchronously: the test orders commits and index reads, |
| 36 | // without depending on a background preparation worker or sleeps. |
| 37 | if _, _, err := query.prepareHistoryIndex(t.Context(), ref); err != nil { |
| 38 | t.Fatal(err) |
| 39 | } |
| 40 | page, err := query.ReadHistoryWindow(t.Context(), ref, HistoryWindowRequest{Anchor: "newest"}) |
| 41 | if err != nil || page.Status != "ready" { |
| 42 | t.Fatalf("read history: %+v, %v", page, err) |
| 43 | } |
| 44 | return page |
| 45 | } |
| 46 | if incremental { |
| 47 | read() |
| 48 | searchHistoryReady(t, query, ref, "body", "", 10) |
| 49 | } |
| 50 | for i, ids := range [][]string{{"retained"}, {"retained", "removed"}, {"retained", "removed"}} { |
| 51 | messages := make([]provider.Message, 0, len(ids)) |
| 52 | for _, id := range ids { |
| 53 | messages = append(messages, provider.Message{ID: id, Role: provider.RoleUser, Content: "rewritten " + id}) |
| 54 | } |
| 55 | payload, err := json.Marshal(map[string]any{"messages": messages, "reason": "cancel-or-recovery-rewrite", "sourceSequences": []uint64{1}}) |
| 56 | if err != nil { |
| 57 | t.Fatal(err) |
| 58 | } |
| 59 | if _, err := runtime.Session().Append(t.Context(), Batch{OperationID: fmt.Sprintf("replace-%d", i), Events: []Event{{Kind: "history/replace", Payload: payload}}}); err != nil { |
| 60 | t.Fatal(err) |
| 61 | } |
| 62 | if _, err := runtime.Session().Flush(t.Context()); err != nil { |
| 63 | t.Fatal(err) |
| 64 | } |
| 65 | if incremental || i == 2 { |
| 66 | page := read() |
| 67 | if !idsEqual(windowIDs(t, page), ids) { |
| 68 | t.Fatalf("replacement %d ids: %v", i, windowIDs(t, page)) |
| 69 | } |
| 70 | if page.Messages[0].Version != i+2 { |
| 71 | t.Fatalf("retained version = %d, want %d", page.Messages[0].Version, i+2) |
| 72 | } |
| 73 | if page := searchHistoryReady(t, query, ref, "rewritten", "", 10); len(page.Hits) != len(ids) { |
| 74 | t.Fatalf("search replacement %d: %+v", i, page) |
| 75 | } |
| 76 | } |
| 77 | } |
| 78 | // The same derived index serves the older page API and search. |
| 79 | if page := historyPageReady(t, query, ref, "", 10); len(page.Messages) != 2 { |
| 80 | t.Fatalf("history page: %+v", page) |
| 81 | } |
| 82 | if page := searchHistoryReady(t, query, ref, "rewritten", "", 10); len(page.Hits) != 2 { |
| 83 | t.Fatalf("search page: %+v", page) |
| 84 | } |
| 85 | if err := service.CloseAll(t.Context()); err != nil { |
| 86 | t.Fatal(err) |
| 87 | } |
| 88 | // Cached and rebuilt readers must both recover the existing durable |
| 89 | // transcript after restart; no rewrite of user data is needed. |
| 90 | for _, removeCache := range []bool{false, true} { |
| 91 | if removeCache { |
| 92 | if err := os.RemoveAll(filepath.Join(root, ".query-cache")); err != nil { |
| 93 | t.Fatal(err) |
| 94 | } |
| 95 | } |
| 96 | reopened, err := NewService("local", NewFilesystemPersistence(root)) |
| 97 | if err != nil { |
| 98 | t.Fatal(err) |
| 99 | } |
| 100 | query = reopened.Query() |
| 101 | if page := read(); len(page.Messages) != 2 || page.Messages[0].Version != 4 { |
| 102 | t.Fatalf("restart (cache removed=%v): %+v", removeCache, page) |
| 103 | } |
| 104 | if err := reopened.CloseAll(t.Context()); err != nil { |
| 105 | t.Fatal(err) |
| 106 | } |
| 107 | } |
| 108 | }) |
| 109 | } |
| 110 | } |
| 111 |