返回 DeepSeek-Reasonix
message_retract_test.go
根目录 / internal / session / message_retract_test.go
1 package session
2
3 import (
4 "context"
5 "encoding/json"
6 "errors"
7 "os"
8 "path/filepath"
9 "testing"
10
11 "reasonix/internal/provider"
12 )
13
14 func TestMessageRetractStrictSchema(t *testing.T) {
15 for _, payload := range []string{`{}`, `{"messageIds":null}`, `{"messageIds":[]}`, `{"messageIds":[""]}`, `{"messageIds":[" "]}`, `{"messageIds":["a","a"]}`, `{"messageIds":["a"],"typo":true}`, `{"messageIds":"a"}`} {
16 t.Run(payload, func(t *testing.T) {
17 _, err := Project([]Commit{{Events: []Event{{Kind: "message/retract", Payload: json.RawMessage(payload)}}}})
18 if !errors.Is(err, ErrDamagedStore) {
19 t.Fatalf("invalid retract payload accepted: %s, err=%v", payload, err)
20 }
21 })
22 }
23 }
24
25 func TestMessageRetractProjectionAndRestore(t *testing.T) {
26 messageEvent := func(kind, id string) Event {
27 payload, err := json.Marshal(map[string]any{"message": provider.Message{ID: id, Role: provider.RoleUser, Content: id}})
28 if err != nil {
29 t.Fatal(err)
30 }
31 return Event{Kind: kind, Payload: payload}
32 }
33 commits := []Commit{{Events: []Event{messageEvent("message/complete", "retained"), messageEvent("message/complete", "removed")}}, {Events: []Event{{Kind: "message/retract", Payload: json.RawMessage(`{"messageIds":["removed"],"reason":"synthetic-turn-interrupted"}`)}}}}
34 check := func(want []string) {
35 t.Helper()
36 p, err := Project(commits)
37 if err != nil {
38 t.Fatal(err)
39 }
40 for name, messages := range map[string][]provider.Message{"history": p.Messages, "model": p.ModelMessages} {
41 ids := make([]string, len(messages))
42 for i := range messages {
43 ids[i] = messages[i].ID
44 }
45 if !idsEqual(ids, want) {
46 t.Fatalf("%s ids=%v, want=%v", name, ids, want)
47 }
48 }
49 }
50 check([]string{"retained"})
51 commits = append(commits, commits[1])
52 check([]string{"retained"})
53 commits = append(commits, Commit{Events: []Event{messageEvent("message/upsert", "removed")}})
54 check([]string{"retained", "removed"})
55 }
56
57 func TestMessageRetractRestoresOriginalTurnAfterReopen(t *testing.T) {
58 t.Run("checkpoint", func(t *testing.T) { testRetractedTurnRestore(t, false) })
59 t.Run("log-replay", func(t *testing.T) { testRetractedTurnRestore(t, true) })
60 }
61
62 func testRetractedTurnRestore(t *testing.T, removeCache bool) {
63 root := filepath.Join(t.TempDir(), "sessions")
64 service, err := NewService("local", NewFilesystemPersistence(root))
65 if err != nil {
66 t.Fatal(err)
67 }
68 t.Cleanup(func() { _ = service.CloseAll(context.Background()) })
69 runtime, err := service.Create(t.Context(), CreateOptions{})
70 if err != nil {
71 t.Fatal(err)
72 }
73 user := json.RawMessage(`{"message":{"id":"input","role":"user","content":"restored input"}}`)
74 if _, err := runtime.Session().Append(t.Context(), Batch{OperationID: "original", TurnID: "original-turn", Events: []Event{
75 {Kind: "turn/start", Payload: json.RawMessage(`{}`)},
76 {Kind: "message/complete", Payload: user},
77 {Kind: "turn/end", Payload: json.RawMessage(`{"status":"completed"}`)},
78 }}); err != nil {
79 t.Fatal(err)
80 }
81 if _, err := runtime.Session().Append(t.Context(), Batch{OperationID: "retract", Events: []Event{{Kind: "message/retract", Payload: json.RawMessage(`{"messageIds":["input"]}`)}}}); err != nil {
82 t.Fatal(err)
83 }
84 if _, err := runtime.Session().Flush(t.Context()); err != nil {
85 t.Fatal(err)
86 }
87 ref := runtime.Ref()
88 if err := service.Close(t.Context(), ref); err != nil {
89 t.Fatal(err)
90 }
91 if removeCache {
92 if err := os.RemoveAll(recoveryCacheDir(filepath.Join(root, ref.SessionID))); err != nil {
93 t.Fatal(err)
94 }
95 }
96 binding, err := service.EnsureExecution(t.Context(), ref)
97 if err != nil {
98 t.Fatal(err)
99 }
100 defer binding.Release(t.Context())
101 runtime, _ = service.Runtime(ref)
102 // Repair batches intentionally have no live turn identity.
103 if _, err := runtime.Session().Append(t.Context(), Batch{OperationID: "restore", Events: []Event{{Kind: "message/upsert", Payload: user}}}); err != nil {
104 t.Fatal(err)
105 }
106 if _, err := runtime.Session().Flush(t.Context()); err != nil {
107 t.Fatal(err)
108 }
109 p := runtime.Session().Snapshot().Projection
110 if p.HiddenTurns["original-turn"] || len(p.TranscriptInputs) != 1 || p.TranscriptInputs[0].TurnID != "original-turn" {
111 t.Fatalf("restored input lost its original turn: hidden=%v inputs=%+v", p.HiddenTurns, p.TranscriptInputs)
112 }
113 info, err := service.Query().Stat(t.Context(), ref)
114 if err != nil {
115 t.Fatal(err)
116 }
117 if info.Turns != 1 || info.Preview != "restored input" || runtime.Session().RecentSnapshot().TotalTurns != 1 {
118 t.Fatalf("restored metadata disagrees: info=%+v recent=%+v", info, runtime.Session().RecentSnapshot())
119 }
120 }
121
122 func TestRecoveredLocalOnlyMessageClearsOnlyItsClosedReplyAnchor(t *testing.T) {
123 _, _, runtime := newSourceService(t, "recovery-anchor")
124 appendCompletedTurn(t, runtime, "first", "first-reply")
125 appendCompletedTurn(t, runtime, "second", "second-reply")
126 payload, _ := json.Marshal(map[string]any{"message": provider.Message{ID: "second-reply", Role: provider.RoleTool, LocalOnly: true}})
127 if _, err := runtime.Session().Append(t.Context(), Batch{OperationID: "repair", Events: []Event{{Kind: "message/upsert", Payload: payload}}}); err != nil {
128 t.Fatal(err)
129 }
130 p := runtime.Session().Snapshot().Projection
131 if p.Turns[0].MessageID != "first-reply" || p.Turns[1].MessageID != "" {
132 t.Fatalf("invalid recovery anchors: %+v", p.Turns)
133 }
134 }
135
136 func TestMessageRetractHistorySearchAndRestart(t *testing.T) {
137 for _, incremental := range []bool{false, true} {
138 name := "rebuild"
139 if incremental {
140 name = "incremental"
141 }
142 t.Run(name, func(t *testing.T) {
143 root := filepath.Join(t.TempDir(), "sessions")
144 service, err := NewService("local", NewFilesystemPersistence(root))
145 if err != nil {
146 t.Fatal(err)
147 }
148 t.Cleanup(func() { _ = service.CloseAll(context.Background()) })
149 runtime, err := service.Create(t.Context(), CreateOptions{SessionID: "retracted"})
150 if err != nil {
151 t.Fatal(err)
152 }
153 appendWindowMessages(t, runtime, "retained", "removed", "tail")
154 query, ref := service.Query(), runtime.Ref()
155 read := func(req HistoryWindowRequest) HistoryWindowPage {
156 t.Helper()
157 if _, _, err := query.prepareHistoryIndex(t.Context(), ref); err != nil {
158 t.Fatal(err)
159 }
160 page, err := query.ReadHistoryWindow(t.Context(), ref, req)
161 if err != nil || page.Status != "ready" {
162 t.Fatalf("history: %+v, %v", page, err)
163 }
164 return page
165 }
166 var oldCursor string
167 if incremental {
168 oldCursor = read(HistoryWindowRequest{Anchor: "newest", Limit: 1}).OlderCursor
169 if page := searchHistoryReady(t, query, ref, "body", "", 10); len(page.Hits) != 3 {
170 t.Fatalf("initial search: %+v", page)
171 }
172 }
173 appendEvent := func(op, kind, payload string) {
174 t.Helper()
175 if _, err := runtime.Session().Append(t.Context(), Batch{OperationID: op, Events: []Event{{Kind: kind, Payload: json.RawMessage(payload)}}}); err != nil {
176 t.Fatal(err)
177 }
178 if _, err := runtime.Session().Flush(t.Context()); err != nil {
179 t.Fatal(err)
180 }
181 }
182 appendEvent("retract", "message/retract", `{"messageIds":["removed"],"reason":"synthetic-turn-interrupted"}`)
183 page := read(HistoryWindowRequest{Anchor: "newest"})
184 if !idsEqual(windowIDs(t, page), []string{"retained", "tail"}) {
185 t.Fatalf("retracted ids=%v", windowIDs(t, page))
186 }
187 if found := searchHistoryReady(t, query, ref, "removed", "", 10); len(found.Hits) != 0 {
188 t.Fatalf("retracted search: %+v", found)
189 }
190 if incremental {
191 old := read(HistoryWindowRequest{Anchor: "cursor", Cursor: oldCursor, Limit: 10})
192 if !idsEqual(windowIDs(t, old), []string{"retained", "removed"}) {
193 t.Fatalf("fixed snapshot changed: %v", windowIDs(t, old))
194 }
195 }
196 appendEvent("retract-again", "message/retract", `{"messageIds":["removed"]}`)
197 read(HistoryWindowRequest{Anchor: "newest"})
198 appendEvent("restore", "message/upsert", `{"message":{"id":"removed","role":"user","content":"restored body-removed"}}`)
199 checkRestored := func() {
200 t.Helper()
201 page := read(HistoryWindowRequest{Anchor: "newest"})
202 if !idsEqual(windowIDs(t, page), []string{"retained", "tail", "removed"}) {
203 t.Fatalf("restored ids=%v", windowIDs(t, page))
204 }
205 if page.Messages[2].Version != 2 {
206 t.Fatalf("restored version=%d, want=2", page.Messages[2].Version)
207 }
208 if found := searchHistoryReady(t, query, ref, "restored", "", 10); len(found.Hits) != 1 {
209 t.Fatalf("restored search: %+v", found)
210 }
211 }
212 checkRestored()
213 if err := service.CloseAll(t.Context()); err != nil {
214 t.Fatal(err)
215 }
216 for _, removeCache := range []bool{false, true} {
217 if removeCache {
218 if err := os.RemoveAll(filepath.Join(root, ".query-cache")); err != nil {
219 t.Fatal(err)
220 }
221 }
222 reopened, err := NewService("local", NewFilesystemPersistence(root))
223 if err != nil {
224 t.Fatal(err)
225 }
226 query = reopened.Query()
227 checkRestored()
228 if err := reopened.CloseAll(t.Context()); err != nil {
229 t.Fatal(err)
230 }
231 }
232 })
233 }
234 }
235
236 func TestMessageRetractRenumbersTurnsWithoutChangingOldSnapshot(t *testing.T) {
237 for _, removeID := range []string{"u1", "u2"} {
238 for _, withAnswer := range []bool{false, true} {
239 name := removeID
240 if withAnswer {
241 name += "-with-answer"
242 }
243 t.Run(name, func(t *testing.T) {
244 service, err := NewService("local", NewFilesystemPersistence(filepath.Join(t.TempDir(), "sessions")))
245 if err != nil {
246 t.Fatal(err)
247 }
248 t.Cleanup(func() { _ = service.CloseAll(context.Background()) })
249 runtime, err := service.Create(t.Context(), CreateOptions{SessionID: "turn-renumber"})
250 if err != nil {
251 t.Fatal(err)
252 }
253 appendMessage := func(op, kind, id string, role provider.Role) {
254 t.Helper()
255 payload, err := json.Marshal(map[string]any{"message": provider.Message{ID: id, Role: role, Content: "body-" + id}})
256 if err != nil {
257 t.Fatal(err)
258 }
259 if _, err := runtime.Session().Append(t.Context(), Batch{OperationID: op, Events: []Event{{Kind: kind, Payload: payload}}}); err != nil {
260 t.Fatal(err)
261 }
262 }
263 for _, id := range []string{"u1", "a1", "u2", "a2", "u3", "a3"} {
264 role := provider.RoleUser
265 if id[0] == 'a' {
266 role = provider.RoleAssistant
267 }
268 appendMessage(id, "message/complete", id, role)
269 }
270 read := func(req HistoryWindowRequest) HistoryWindowPage {
271 t.Helper()
272 if _, err := runtime.Session().Flush(t.Context()); err != nil {
273 t.Fatal(err)
274 }
275 if _, _, err := service.Query().prepareHistoryIndex(t.Context(), runtime.Ref()); err != nil {
276 t.Fatal(err)
277 }
278 page, err := service.Query().ReadHistoryWindow(t.Context(), runtime.Ref(), req)
279 if err != nil || page.Status != "ready" {
280 t.Fatalf("page=%+v err=%v", page, err)
281 }
282 return page
283 }
284 old := read(HistoryWindowRequest{Anchor: "newest", Limit: 1})
285 if old.TotalTurns != 3 || old.Messages[0].VisibleTurn != 3 {
286 t.Fatalf("initial turns=%+v", old)
287 }
288 ids := []string{removeID}
289 if withAnswer {
290 ids = append(ids, "a"+removeID[1:])
291 }
292 payload, _ := json.Marshal(map[string]any{"messageIds": ids})
293 if _, err := runtime.Session().Append(t.Context(), Batch{OperationID: "retract-user", Events: []Event{{Kind: "message/retract", Payload: payload}}}); err != nil {
294 t.Fatal(err)
295 }
296 page := read(HistoryWindowRequest{Anchor: "newest"})
297 if page.TotalTurns != 2 {
298 t.Fatalf("current total turns=%d want=2", page.TotalTurns)
299 }
300 turn := 0
301 for _, m := range page.Messages {
302 if m.Role == string(provider.RoleUser) {
303 turn++
304 }
305 if m.VisibleTurn != turn {
306 t.Fatalf("message %s turn=%d want=%d", m.MessageID, m.VisibleTurn, turn)
307 }
308 }
309 past := read(HistoryWindowRequest{Anchor: "cursor", Cursor: old.OlderCursor, Limit: 10})
310 if past.TotalTurns != 3 || !idsEqual(windowIDs(t, past), []string{"u1", "a1", "u2", "a2", "u3"}) {
311 t.Fatalf("old snapshot changed: %+v", past)
312 }
313 for i, m := range past.Messages {
314 if m.VisibleTurn != i/2+1 {
315 t.Fatalf("old snapshot %s turn=%d", m.MessageID, m.VisibleTurn)
316 }
317 }
318 appendMessage("restore-user", "message/upsert", removeID, provider.RoleUser)
319 restored := read(HistoryWindowRequest{Anchor: "newest"})
320 last := restored.Messages[len(restored.Messages)-1]
321 if restored.TotalTurns != 3 || last.MessageID != removeID || last.VisibleTurn != 3 || last.Version != 2 {
322 t.Fatalf("restored user=%+v total=%d", last, restored.TotalTurns)
323 }
324 })
325 }
326 }
327 }
328
329 func TestMessageRetractCatalogAndRecentFollowVisibleHistory(t *testing.T) {
330 root := filepath.Join(t.TempDir(), "sessions")
331 service, err := NewService("local", NewFilesystemPersistence(root))
332 if err != nil {
333 t.Fatal(err)
334 }
335 t.Cleanup(func() { _ = service.CloseAll(context.Background()) })
336 runtime, err := service.Create(t.Context(), CreateOptions{SessionID: "recent-retraction"})
337 if err != nil {
338 t.Fatal(err)
339 }
340 for _, id := range []string{"first", "second"} {
341 user, _ := json.Marshal(map[string]any{"message": provider.Message{ID: id, Role: provider.RoleUser, Content: id + " question"}})
342 answer, _ := json.Marshal(map[string]any{"message": provider.Message{ID: id + "-answer", Role: provider.RoleAssistant, Content: id + " answer"}})
343 if _, err := runtime.Session().Append(t.Context(), Batch{OperationID: id, TurnID: id + "-turn", Events: []Event{{Kind: "turn/start", Payload: json.RawMessage(`{}`)}, {Kind: "message/complete", Payload: user}, {Kind: "message/complete", Payload: answer}, {Kind: "turn/end", Payload: json.RawMessage(`{"status":"completed"}`)}}}); err != nil {
344 t.Fatal(err)
345 }
346 }
347 if _, err := runtime.Session().Flush(t.Context()); err != nil {
348 t.Fatal(err)
349 }
350 if got := runtime.Session().RecentSnapshot(); got.TotalTurns != 2 {
351 t.Fatalf("initial recent turns=%d", got.TotalTurns)
352 }
353 if _, err := runtime.Session().Append(t.Context(), Batch{OperationID: "remove-first", Events: []Event{{Kind: "message/retract", Payload: json.RawMessage(`{"messageIds":["first","first-answer"]}`)}}}); err != nil {
354 t.Fatal(err)
355 }
356 if _, err := runtime.Session().Flush(t.Context()); err != nil {
357 t.Fatal(err)
358 }
359 check := func(label string, recent RecentSnapshot, info SessionInfo) {
360 t.Helper()
361 if recent.TotalTurns != 1 {
362 t.Errorf("%s recent total=%d want=1", label, recent.TotalTurns)
363 }
364 if info.Turns != 1 || info.Preview != "second question" {
365 t.Errorf("%s catalog turns=%d preview=%q", label, info.Turns, info.Preview)
366 }
367 ids := make([]string, 0, len(recent.Entries))
368 for _, m := range recent.Entries {
369 ids = append(ids, m.MessageID)
370 if m.VisibleTurn != 1 {
371 t.Errorf("%s recent %s visible turn=%d", label, m.MessageID, m.VisibleTurn)
372 }
373 }
374 if !idsEqual(ids, []string{"second", "second-answer"}) {
375 t.Errorf("%s recent ids=%v", label, ids)
376 }
377 }
378 info, err := service.Query().Stat(t.Context(), runtime.Ref())
379 if err != nil {
380 t.Fatal(err)
381 }
382 check("live", runtime.Session().RecentSnapshot(), info)
383 targets := ForkTargets(runtime.Session().Snapshot().Projection)
384 if len(targets.Targets) != 1 || targets.Targets[0].TurnID != "second-turn" || targets.Targets[0].TurnNumber != 1 || targets.Targets[0].MessageID != "second-answer" {
385 t.Fatalf("withdrawn turn remained forkable or numbering changed: %+v", targets)
386 }
387 if seq, reason, err := ForkSequence(runtime.Session().Snapshot().Projection, "first-turn"); err != nil || seq != 0 || reason != ForkHistoryUnverifiable {
388 t.Fatalf("withdrawn turn fork: seq=%d reason=%s err=%v", seq, reason, err)
389 }
390 if seq, reason, err := ForkSequence(runtime.Session().Snapshot().Projection, "second-turn"); err != nil || seq == 0 || reason != ForkAvailable {
391 t.Fatalf("remaining turn fork: seq=%d reason=%s err=%v", seq, reason, err)
392 }
393 ref := runtime.Ref()
394 if err := service.Close(t.Context(), ref); err != nil {
395 t.Fatal(err)
396 }
397 view, err := service.OpenSession(t.Context(), ref)
398 if err != nil {
399 t.Fatal(err)
400 }
401 info, err = service.Query().Stat(t.Context(), ref)
402 if err != nil {
403 t.Fatal(err)
404 }
405 check("cold", view.Recent, info)
406 // Reopen a writer and remove every remaining input. Neither a cached
407 // first-preview fallback nor old turn boundaries may resurrect the list
408 // preview or leave a phantom visible turn.
409 binding, err := service.EnsureExecution(t.Context(), ref)
410 if err != nil {
411 t.Fatal(err)
412 }
413 live, ok := service.Runtime(ref)
414 if !ok {
415 t.Fatal("reopened runtime missing")
416 }
417 if _, err := live.Session().Append(t.Context(), Batch{OperationID: "remove-all-inputs", Events: []Event{{Kind: "message/retract", Payload: json.RawMessage(`{"messageIds":["second","second-answer"]}`)}}}); err != nil {
418 t.Fatal(err)
419 }
420 if _, err := live.Session().Flush(t.Context()); err != nil {
421 t.Fatal(err)
422 }
423 if err := binding.Release(t.Context()); err != nil {
424 t.Fatal(err)
425 }
426 checkEmpty := func(label string, recent RecentSnapshot, info SessionInfo) {
427 t.Helper()
428 if recent.TotalTurns != 0 || len(recent.Entries) != 0 {
429 t.Errorf("%s recent resurrected empty history: %+v", label, recent)
430 }
431 if info.Turns != 0 || info.Preview != "" {
432 t.Errorf("%s catalog resurrected empty history: turns=%d preview=%q", label, info.Turns, info.Preview)
433 }
434 }
435 info, err = service.Query().Stat(t.Context(), ref)
436 if err != nil {
437 t.Fatal(err)
438 }
439 checkEmpty("empty-live", live.Session().RecentSnapshot(), info)
440 checkNoForks := func(projection Projection) {
441 t.Helper()
442 if targets := ForkTargets(projection); len(targets.Targets) != 0 {
443 t.Fatalf("empty history exposes fork targets: %+v", targets)
444 }
445 for _, id := range []string{"first-turn", "second-turn"} {
446 if seq, reason, err := ForkSequence(projection, id); err != nil || seq != 0 || reason != ForkHistoryUnverifiable {
447 t.Fatalf("empty history fork %s: seq=%d reason=%s err=%v", id, seq, reason, err)
448 }
449 }
450 }
451 checkNoForks(live.Session().Snapshot().Projection)
452 if err := service.Close(t.Context(), ref); err != nil {
453 t.Fatal(err)
454 }
455 for _, removeCache := range []bool{false, true} {
456 label := "empty-cached"
457 if removeCache {
458 label = "empty-rebuilt"
459 for _, dir := range []string{filepath.Join(root, ".query-cache", ref.SessionID), recoveryCacheDir(filepath.Join(root, ref.SessionID))} {
460 if err := os.RemoveAll(dir); err != nil {
461 t.Fatal(err)
462 }
463 }
464 }
465 // EnsureExecution deterministically completes checkpoint rebuild rather
466 // than racing a cold reader's background preparation notification.
467 binding, err := service.EnsureExecution(t.Context(), ref)
468 if err != nil {
469 t.Fatal(err)
470 }
471 reopened, ok := service.Runtime(ref)
472 if !ok {
473 t.Fatal("reopened runtime missing")
474 }
475 info, err := service.Query().Stat(t.Context(), ref)
476 if err != nil {
477 t.Fatal(err)
478 }
479 checkEmpty(label, reopened.Session().RecentSnapshot(), info)
480 checkNoForks(reopened.Session().Snapshot().Projection)
481 if err := binding.Release(t.Context()); err != nil {
482 t.Fatal(err)
483 }
484 if err := service.Close(t.Context(), ref); err != nil {
485 t.Fatal(err)
486 }
487 cold, err := service.OpenSession(t.Context(), ref)
488 if err != nil {
489 t.Fatal(err)
490 }
491 info, err = service.Query().Stat(t.Context(), ref)
492 if err != nil {
493 t.Fatal(err)
494 }
495 checkEmpty(label+"-cold", cold.Recent, info)
496 }
497 }
498
498 lines GO