返回 DeepSeek-Reasonix
review_regression_test.go
根目录 / internal / session / review_regression_test.go
1 package session
2
3 import (
4 "bytes"
5 "context"
6 "encoding/json"
7 "errors"
8 "io"
9 "os"
10 "path/filepath"
11 "runtime"
12 "testing"
13 "time"
14
15 "reasonix/internal/filelock"
16 "reasonix/internal/provider"
17 )
18
19 // owner grants host authority for an exact instance. Tests use it wherever a
20 // host would retire a runtime it published itself.
21 func owner(t *testing.T, service *Service, runtime *Runtime) *RuntimeOwner {
22 t.Helper()
23 grant, err := service.Owner(runtime)
24 if err != nil {
25 t.Fatalf("owner grant: %v", err)
26 }
27 return grant
28 }
29
30 func reviewRuntime(t *testing.T) (*Service, *Runtime) {
31 t.Helper()
32 service, err := NewService("local", NewFilesystemPersistence(filepath.Join(t.TempDir(), "sessions-v4")))
33 if err != nil {
34 t.Fatal(err)
35 }
36 t.Cleanup(func() { _ = service.CloseAll(context.Background()) })
37 runtime, err := service.Create(t.Context(), CreateOptions{SessionID: "review"})
38 if err != nil {
39 t.Fatal(err)
40 }
41 t.Cleanup(func() { _ = runtime.close(context.Background()) })
42 return service, runtime
43 }
44
45 func TestStateSnapshotOmitsHistory(t *testing.T) {
46 _, runtime := reviewRuntime(t)
47 payload, err := json.Marshal(map[string]any{"message": provider.Message{ID: "visible", Role: provider.RoleUser, Content: "hello"}})
48 if err != nil {
49 t.Fatal(err)
50 }
51 if _, err := runtime.Session().AppendBatch(t.Context(), "message", []Event{{Kind: "message/complete", Payload: payload}}); err != nil {
52 t.Fatal(err)
53 }
54 state := runtime.StateSnapshot().Session
55 if state.EventSequence != 1 || len(state.Projection.Messages) != 0 || len(state.Projection.ModelMessages) != 0 {
56 t.Fatalf("activity snapshot includes history or loses its sequence: %+v", state)
57 }
58 if len(runtime.Snapshot().Session.Projection.Messages) != 1 {
59 t.Fatal("state snapshot changed the stored history")
60 }
61 }
62
63 func TestSessionIdentityRejectsPathsWithoutCreatingFiles(t *testing.T) {
64 persistence := NewFilesystemPersistence(t.TempDir())
65 for _, id := range []string{"..", "../escape", "a/b", `a\b`, "/absolute", ".hidden", "CON", "com1.log", "bad:name", ".", ""} {
66 if _, err := persistence.Open(id, ReadWrite); err == nil {
67 t.Fatalf("accepted path as identity: %q", id)
68 }
69 }
70 entries, err := os.ReadDir(persistence.Root)
71 if err != nil || len(entries) != 0 {
72 t.Fatalf("invalid opens changed the root: %v, %v", entries, err)
73 }
74 }
75
76 func TestSessionIdentityRejectsSymlinkOutsideRoot(t *testing.T) {
77 base := t.TempDir()
78 outside := filepath.Join(base, "outside")
79 store, err := CreateStore(outside, "escape")
80 if err != nil {
81 t.Fatal(err)
82 }
83 if err := store.Close(t.Context()); err != nil {
84 t.Fatal(err)
85 }
86 root := filepath.Join(base, "sessions-v4")
87 if err := os.MkdirAll(root, 0o700); err != nil {
88 t.Fatal(err)
89 }
90 if err := os.Symlink(outside, filepath.Join(root, "escape")); err != nil {
91 t.Skipf("symlink unavailable: %v", err)
92 }
93 persistence := NewFilesystemPersistence(root)
94 if _, err := persistence.Open("escape", ReadOnly); err == nil {
95 t.Fatal("read-only open followed a session symlink outside the store root")
96 }
97 if err := persistence.Delete(t.Context(), "escape"); err == nil {
98 t.Fatal("delete followed a session symlink outside the store root")
99 }
100 }
101
102 func TestDirectoryOwnershipExcludesWriterDuringRename(t *testing.T) {
103 _, runtime := reviewRuntime(t)
104 store := runtime.Session().Handle().(*Store)
105 if err := runtime.close(t.Context()); err != nil {
106 t.Fatal(err)
107 }
108 release, err := filelock.Acquire(t.Context(), directoryOwnershipPath(store.dir))
109 if err != nil {
110 t.Fatal(err)
111 }
112 defer release()
113 if reopened, err := Open(store.dir, store.SessionID()); !errors.Is(err, ErrWriterOwned) {
114 if reopened != nil {
115 _ = reopened.Close(t.Context())
116 }
117 t.Fatalf("writer entered the directory transition: %v", err)
118 }
119 // The ownership file stays outside the moved tree, including on Windows.
120 if err := os.Rename(store.dir, store.dir+"-removed"); err != nil {
121 t.Fatalf("directory ownership prevents rename: %v", err)
122 }
123 }
124
125 func TestCancelReceiptDoesNotWaitForSessionProjection(t *testing.T) {
126 service, runtime := reviewRuntime(t)
127 ctx, _ := bindTestExecution(t, runtime, "model")
128 store := runtime.Session().Handle().(*Store)
129 store.mu.Lock()
130 defer store.mu.Unlock()
131 done := make(chan error, 1)
132 go func() { _, err := service.CancelSession(runtime.Ref()); done <- err }()
133 select {
134 case err := <-done:
135 if err != nil || !errors.Is(ctx.Err(), context.Canceled) {
136 t.Fatalf("cancel = %v, context = %v", err, ctx.Err())
137 }
138 case <-time.After(5 * time.Second):
139 t.Fatal("cancel receipt waits for the projection lock")
140 }
141 }
142
143 func TestCancelSignalsWithoutRuntimeMutex(t *testing.T) {
144 service, runtime := reviewRuntime(t)
145 ctx, _ := bindTestExecution(t, runtime, "model")
146
147 runtime.mu.Lock()
148 done := make(chan CancelReceipt, 1)
149 go func() {
150 receipt, cancelErr := service.CancelSession(runtime.Ref())
151 if cancelErr != nil {
152 t.Errorf("cancel: %v", cancelErr)
153 }
154 done <- receipt
155 }()
156 select {
157 case receipt := <-done:
158 if !receipt.Accepted || receipt.Phase != RuntimeCancelling {
159 t.Fatalf("receipt = %+v", receipt)
160 }
161 if !errors.Is(ctx.Err(), context.Canceled) {
162 t.Fatalf("activity context = %v", ctx.Err())
163 }
164 case <-time.After(5 * time.Second):
165 runtime.mu.Unlock()
166 t.Fatal("cancel waited for the runtime commit mutex")
167 }
168 runtime.mu.Unlock()
169 }
170
171 func TestUnknownRequiredPrefixDoesNotTruncateTail(t *testing.T) {
172 _, runtime := reviewRuntime(t)
173 store := runtime.Session().Handle().(*Store)
174 if _, err := runtime.Session().Append(t.Context(), Batch{OperationID: "known", Events: []Event{{Kind: "diagnostic"}}}); err != nil {
175 t.Fatal(err)
176 }
177 if err := runtime.close(t.Context()); err != nil {
178 t.Fatal(err)
179 }
180 path := filepath.Join(store.dir, currentLogName)
181 data, err := os.ReadFile(path)
182 if err != nil {
183 t.Fatal(err)
184 }
185 unknown := Commit{
186 SchemaVersion: SchemaVersion, Codec: Codec, RecordType: "commit", ID: "future-commit",
187 OperationID: "future-operation", OperationHash: "future-hash", FirstSequence: 2,
188 EventCount: 1, WriterGeneration: store.Manifest().WriterGeneration, CreatedAt: time.Now().UTC(),
189 Events: []Event{{ID: "future-event", Sequence: 2, Kind: "future/required"}},
190 }
191 var encoded bytes.Buffer
192 if _, err := encodeV4Commits(t.Context(), &encoded, contentStoreForSessionDir(store.dir), []Commit{unknown}); err != nil {
193 t.Fatal(err)
194 }
195 data = append(data, encoded.Bytes()...)
196 data = append(data, []byte(`{"torn":`)...)
197 if err := os.WriteFile(path, data, 0600); err != nil {
198 t.Fatal(err)
199 }
200 if _, err := Open(store.dir, store.SessionID()); !errors.Is(err, ErrUnsupportedVersion) {
201 t.Fatalf("open = %v", err)
202 }
203 after, err := os.ReadFile(path)
204 if err != nil || !bytes.Equal(after, data) {
205 t.Fatalf("unsupported log was changed: %v", err)
206 }
207 }
208
209 func TestSnapshotCannotMutateAcceptedMessageMetadata(t *testing.T) {
210 _, runtime := reviewRuntime(t)
211 payload, err := json.Marshal(map[string]any{"message": provider.Message{ID: "m1", Role: provider.RoleUser, Images: []string{"original"}}})
212 if err != nil {
213 t.Fatal(err)
214 }
215 if _, err := runtime.Session().AppendBatch(t.Context(), "input", []Event{{Kind: "message/complete", Payload: payload}}); err != nil {
216 t.Fatal(err)
217 }
218 snapshot := runtime.Session().Snapshot()
219 snapshot.Projection.Messages[0].Images[0] = "changed"
220 snapshot.Projection.ModelMessages[0].Images[0] = "changed-again"
221 if got := runtime.Session().Snapshot().Projection.Messages[0].Images[0]; got != "original" {
222 t.Fatalf("observer mutated accepted message: %q", got)
223 }
224 }
225
226 func TestOldRuntimeDisposerCannotCloseSuccessor(t *testing.T) {
227 service, old := reviewRuntime(t)
228 oldOwner := owner(t, service, old)
229 if err := oldOwner.Close(t.Context()); err != nil {
230 t.Fatal(err)
231 }
232 binding, err := service.Open(t.Context(), old.Ref())
233 if err != nil {
234 t.Fatal(err)
235 }
236 next := binding.Runtime()
237 t.Cleanup(func() { _ = binding.Release(context.Background()) })
238 // The delayed disposer still holds its own grant for the retired instance.
239 if err := oldOwner.Close(t.Context()); err != nil {
240 t.Fatal(err)
241 }
242 if got, ok := service.Runtime(next.Ref()); !ok || got != next {
243 t.Fatal("old disposer removed successor")
244 }
245 if _, err := next.Session().AppendBatch(t.Context(), "still-open", []Event{{Kind: "diagnostic", Optional: true}}); err != nil {
246 t.Fatal(err)
247 }
248 }
249
250 func TestClientBindingOwnsDetachButNotRuntimeClose(t *testing.T) {
251 service, runtime := reviewRuntime(t)
252 service.idleTTL = 20 * time.Millisecond
253 first, err := service.Bind(runtime)
254 if err != nil {
255 t.Fatal(err)
256 }
257 second, err := service.Bind(runtime)
258 if err != nil {
259 t.Fatal(err)
260 }
261 if err := owner(t, service, runtime).Close(t.Context()); !errors.Is(err, ErrRuntimeBound) {
262 t.Fatalf("close bound runtime = %v", err)
263 }
264 if err := first.Release(t.Context()); err != nil {
265 t.Fatal(err)
266 }
267 if got, ok := service.Runtime(runtime.Ref()); !ok || got != runtime {
268 t.Fatal("one client detached the runtime used by another client")
269 }
270 if err := second.Release(t.Context()); err != nil {
271 t.Fatal(err)
272 }
273 if got, ok := service.Runtime(runtime.Ref()); !ok || got != runtime {
274 t.Fatal("idle runtime was not retained for quick rebinding")
275 }
276 deadline := time.Now().Add(time.Second)
277 for time.Now().Before(deadline) {
278 if _, ok := service.Runtime(runtime.Ref()); !ok {
279 return
280 }
281 time.Sleep(time.Millisecond)
282 }
283 t.Fatal("idle runtime remained published after its retention period")
284 }
285
286 func TestIdleRuntimeCacheBudgetRetiresLeastRecentlyUsedRuntime(t *testing.T) {
287 service, first := reviewRuntime(t)
288 service.idleTTL = time.Hour
289 service.idleBudget = 64 << 10
290 firstBinding, err := service.Bind(first)
291 if err != nil {
292 t.Fatal(err)
293 }
294 second, err := service.Create(t.Context(), CreateOptions{SessionID: "budget-second"})
295 if err != nil {
296 t.Fatal(err)
297 }
298 t.Cleanup(func() {
299 for _, ref := range []SessionRef{first.Ref(), second.Ref()} {
300 if _, live := service.Runtime(ref); live {
301 _ = service.Close(context.Background(), ref)
302 }
303 }
304 })
305 secondBinding, err := service.Bind(second)
306 if err != nil {
307 t.Fatal(err)
308 }
309 if err := firstBinding.Release(t.Context()); err != nil {
310 t.Fatal(err)
311 }
312 if err := secondBinding.Release(t.Context()); err != nil {
313 t.Fatal(err)
314 }
315 deadline := time.Now().Add(time.Second)
316 for time.Now().Before(deadline) {
317 _, firstLive := service.Runtime(first.Ref())
318 _, secondLive := service.Runtime(second.Ref())
319 if !firstLive && secondLive {
320 return
321 }
322 time.Sleep(time.Millisecond)
323 }
324 t.Fatal("idle cache budget did not retire the least recently used runtime")
325 }
326
327 func TestLastClientDetachDoesNotCancelActiveRuntime(t *testing.T) {
328 service, runtime := reviewRuntime(t)
329 service.idleTTL = 20 * time.Millisecond
330 binding, err := service.Bind(runtime)
331 if err != nil {
332 t.Fatal(err)
333 }
334 ctx, exec := bindTestExecution(t, runtime, "model")
335 if err := binding.Release(t.Context()); err != nil {
336 t.Fatal(err)
337 }
338 if ctx.Err() != nil {
339 t.Fatalf("client detach cancelled host-owned activity: %v", ctx.Err())
340 }
341 if got, ok := service.Runtime(runtime.Ref()); !ok || got != runtime {
342 t.Fatal("active runtime retired when its last client detached")
343 }
344 exec.Finish()
345 if _, ok := service.Runtime(runtime.Ref()); !ok {
346 t.Fatal("completed runtime was not retained for quick rebinding")
347 }
348 deadline := time.Now().Add(time.Second)
349 for time.Now().Before(deadline) {
350 if _, ok := service.Runtime(runtime.Ref()); !ok {
351 return
352 }
353 time.Sleep(time.Millisecond)
354 }
355 t.Fatal("unbound runtime did not retire after its retention period")
356 }
357
358 func TestOwnerCloseFencesConcurrentClientBinding(t *testing.T) {
359 service, sessionRuntime := reviewRuntime(t)
360 sessionRuntime.mu.Lock()
361 closed := make(chan error, 1)
362 grant := owner(t, service, sessionRuntime)
363 go func() { closed <- grant.Close(t.Context()) }()
364 deadline := time.After(5 * time.Second)
365 for {
366 service.mu.Lock()
367 retiring := service.retiring[sessionRuntime] != nil
368 service.mu.Unlock()
369 if retiring {
370 break
371 }
372 select {
373 case <-deadline:
374 sessionRuntime.mu.Unlock()
375 t.Fatal("owner close did not publish its retirement fence")
376 default:
377 runtime.Gosched()
378 }
379 }
380 if _, err := service.Bind(sessionRuntime); !errors.Is(err, ErrRuntimeRetiring) {
381 sessionRuntime.mu.Unlock()
382 t.Fatalf("bind during owner close = %v", err)
383 }
384 sessionRuntime.mu.Unlock()
385 if err := <-closed; err != nil {
386 t.Fatal(err)
387 }
388 }
389
390 func TestTitleCanReturnToEarlierValue(t *testing.T) {
391 service, runtime := reviewRuntime(t)
392 for _, title := range []string{"A", "B", "A"} {
393 if err := service.SetTitle(t.Context(), runtime.Ref(), title); err != nil {
394 t.Fatal(err)
395 }
396 if got := runtime.Session().Snapshot().Projection.Title; got != title {
397 t.Fatalf("title = %q, want %q", got, title)
398 }
399 }
400 }
401
402 func TestCloseFailureUnregistersReleasedWriter(t *testing.T) {
403 service, runtime := reviewRuntime(t)
404 store := runtime.Session().Handle().(*Store)
405 failure := errors.New("disk unavailable")
406 store.writeFn = func(context.Context, io.Writer, []byte) error { return failure }
407 if _, err := runtime.Session().AppendBatch(t.Context(), "pending", []Event{{Kind: "diagnostic", Optional: true}}); err != nil {
408 t.Fatal(err)
409 }
410 if err := service.Close(t.Context(), runtime.Ref()); !errors.Is(err, failure) {
411 t.Fatalf("close = %v", err)
412 }
413 if _, ok := service.Runtime(runtime.Ref()); ok {
414 t.Fatal("closed writer remains attachable")
415 }
416 if err := service.Close(t.Context(), runtime.Ref()); !errors.Is(err, failure) {
417 t.Fatalf("repeat close = %v", err)
418 }
419 binding, err := service.Open(t.Context(), runtime.Ref())
420 if err != nil {
421 t.Fatal(err)
422 }
423 t.Cleanup(func() { _ = binding.Release(context.Background()) })
424 if binding.Runtime() == runtime {
425 t.Fatal("open returned released writer")
426 }
427 }
428
429 func TestRewindBeforeFirstTurnPreservesInitialization(t *testing.T) {
430 service, runtime := reviewRuntime(t)
431 if _, err := runtime.Session().AppendBatch(t.Context(), "config", []Event{{Kind: "session/config", Payload: []byte(`{"modelRef":"test/model"}`)}}); err != nil {
432 t.Fatal(err)
433 }
434 if _, err := runtime.Session().Append(t.Context(), Batch{OperationID: "input", TurnID: "first", Events: []Event{{Kind: "turn/start"}, {Kind: "turn/end", Payload: []byte(`{"status":"completed"}`)}}}); err != nil {
435 t.Fatal(err)
436 }
437 child, err := service.Rewind(t.Context(), runtime.Ref(), "first", "rewound")
438 if err != nil {
439 t.Fatal(err)
440 }
441 t.Cleanup(func() { _ = service.Close(context.Background(), child.Ref()) })
442 if got := child.Session().Snapshot().Projection.ModelRef; got != "test/model" {
443 t.Fatalf("model = %q", got)
444 }
445 }
446
447 func TestFinishedActivityCancelsItsContext(t *testing.T) {
448 _, runtime := reviewRuntime(t)
449 ctx, exec := bindTestExecution(t, runtime, "model")
450 exec.cancel()
451 exec.Finish()
452 if !errors.Is(ctx.Err(), context.Canceled) {
453 t.Fatal("finished activity retains live cancellation context")
454 }
455 }
456
457 func TestForkCopiesOwnedAttachments(t *testing.T) {
458 service, runtime := reviewRuntime(t)
459 dir := runtime.Session().Handle().(*Store).dir
460 asset := filepath.Join(dir, "attachments", "input.txt")
461 if err := os.MkdirAll(filepath.Dir(asset), 0700); err != nil {
462 t.Fatal(err)
463 }
464 if err := os.WriteFile(asset, []byte("owned context"), 0600); err != nil {
465 t.Fatal(err)
466 }
467 if _, err := runtime.Session().Append(t.Context(), Batch{OperationID: "input", TurnID: "first", Events: []Event{{Kind: "turn/start"}, {Kind: "turn/end", Payload: []byte(`{"status":"completed"}`)}}}); err != nil {
468 t.Fatal(err)
469 }
470 child, err := service.Fork(t.Context(), runtime.Ref(), "first", "child")
471 if err != nil {
472 t.Fatal(err)
473 }
474 t.Cleanup(func() { _ = service.Close(context.Background(), child.Ref()) })
475 if err := service.Delete(t.Context(), runtime.Ref()); err != nil {
476 t.Fatal(err)
477 }
478 data, err := os.ReadFile(filepath.Join(child.Session().Handle().(*Store).dir, "attachments", "input.txt"))
479 if err != nil || string(data) != "owned context" {
480 t.Fatalf("child attachment = %q, %v", data, err)
481 }
482 }
483
483 lines GO