返回 DeepSeek-Reasonix
prototype_import_test.go
根目录 / internal / session / prototype_import_test.go
1 package session
2
3 import (
4 "bytes"
5 "context"
6 "encoding/json"
7 "errors"
8 "os"
9 "path/filepath"
10 "strings"
11 "testing"
12 "time"
13
14 "reasonix/internal/agent"
15 "reasonix/internal/filelock"
16 "reasonix/internal/fileutil"
17 "reasonix/internal/provider"
18 )
19
20 func writePrototypeStore(t *testing.T, dir string, events []Event, torn string) {
21 t.Helper()
22 if err := os.MkdirAll(dir, 0o700); err != nil {
23 t.Fatal(err)
24 }
25 manifest := Manifest{SchemaVersion: 3, Codec: PrototypeCodec, SessionID: "prototype", CreatedAt: time.Now().UTC(), WriterGeneration: 1}
26 manifestBytes, _ := json.Marshal(manifest)
27 if err := fileutil.AtomicWriteFileStrict(filepath.Join(dir, "manifest.json"), append(manifestBytes, '\n'), 0o600); err != nil {
28 t.Fatal(err)
29 }
30 var log []byte
31 if len(events) > 0 {
32 for i := range events {
33 events[i].ID = "event-" + string(rune('a'+i))
34 events[i].Sequence = uint64(i + 1)
35 }
36 hash, err := hashOperation("prototype", "", events)
37 if err != nil {
38 t.Fatal(err)
39 }
40 commit := Commit{SchemaVersion: 3, Codec: PrototypeCodec, RecordType: "commit", ID: "prototype-commit", OperationID: "prototype-operation", OperationHash: hash, FirstSequence: 1, EventCount: len(events), WriterGeneration: 1, CreatedAt: time.Now().UTC(), Events: events}
41 line, _ := json.Marshal(commit)
42 log = append(line, '\n')
43 }
44 log = append(log, torn...)
45 if err := fileutil.AtomicWriteFileStrict(filepath.Join(dir, "events.jsonl"), log, 0o600); err != nil {
46 t.Fatal(err)
47 }
48 }
49
50 func TestContinueImportedResolvesPairedHistoryStructurally(t *testing.T) {
51 root := t.TempDir()
52 legacyDir := filepath.Join(root, "sessions")
53 legacyPath := filepath.Join(legacyDir, "paired.jsonl")
54 if err := os.MkdirAll(legacyDir, 0o700); err != nil {
55 t.Fatal(err)
56 }
57 legacy := agent.NewSession("system")
58 legacy.Add(provider.Message{ID: "user-1", Role: provider.RoleUser, Content: "hello"})
59 if err := legacy.Save(legacyPath); err != nil {
60 t.Fatal(err)
61 }
62 messages := legacy.Snapshot()
63 payload, err := json.Marshal(map[string]any{"messages": messages})
64 if err != nil {
65 t.Fatal(err)
66 }
67 targetRoot := filepath.Join(root, "sessions-v4")
68 previewDir := filepath.Join(targetRoot, agent.BranchID(legacyPath))
69 writePrototypeStore(t, previewDir, []Event{{Kind: "context/replace", Payload: payload}}, "")
70
71 result, err := importSourceForLegacyWithHeader(t.Context(), legacyPath, targetRoot, "", CreateOptions{
72 CWD: "/workspace", Origin: SessionOriginLegacyImport,
73 })
74 if err != nil {
75 t.Fatal(err)
76 }
77 if result.Kind != "events" || result.TargetID == agent.BranchID(legacyPath) {
78 t.Fatalf("resolved import = %+v", result)
79 }
80 info, err := NewFilesystemPersistence(targetRoot).Stat(t.Context(), result.TargetID)
81 // Headers record CWD in OS-native form; clean the expectation the same way.
82 if err != nil || info.CWD != filepath.Clean("/workspace") || info.Origin != SessionOriginLegacyImport {
83 t.Fatalf("imported header = %+v, %v", info, err)
84 }
85 }
86
87 func TestContinueImportedReusesFinalCanonicalStore(t *testing.T) {
88 root := t.TempDir()
89 legacyPath := filepath.Join(root, "legacy.jsonl")
90 legacy := agent.NewSession("system")
91 legacy.Add(provider.Message{Role: provider.RoleUser, Content: "old"})
92 if err := legacy.Save(legacyPath); err != nil {
93 t.Fatal(err)
94 }
95 targetRoot := filepath.Join(root, "sessions-v4")
96 id := agent.BranchID(legacyPath)
97 canonical, err := Open(filepath.Join(targetRoot, id), id)
98 if err != nil {
99 t.Fatal(err)
100 }
101 payload, _ := json.Marshal(struct {
102 Message provider.Message `json:"message"`
103 }{Message: provider.Message{ID: "new", Role: provider.RoleAssistant, Content: "newer canonical work"}})
104 if _, err := canonical.Append(t.Context(), Batch{OperationID: "canonical", Events: []Event{{Kind: "message/complete", Payload: payload}}}); err != nil {
105 t.Fatal(err)
106 }
107 if _, err := canonical.Flush(t.Context()); err != nil {
108 t.Fatal(err)
109 }
110 if err := canonical.Close(t.Context()); err != nil {
111 t.Fatal(err)
112 }
113
114 result, err := importSourceForLegacy(t.Context(), legacyPath, targetRoot, "")
115 if err != nil {
116 t.Fatal(err)
117 }
118 if result.TargetID != id || !result.Reused || result.Kind != "final" {
119 t.Fatalf("result = %+v", result)
120 }
121 }
122
123 func TestContinueImportedRefusesDivergentPairedHistory(t *testing.T) {
124 root := t.TempDir()
125 legacyDir := filepath.Join(root, "sessions")
126 legacyPath := filepath.Join(legacyDir, "conflict.jsonl")
127 if err := os.MkdirAll(legacyDir, 0o700); err != nil {
128 t.Fatal(err)
129 }
130 legacy := agent.NewSession("system")
131 legacy.Add(provider.Message{ID: "legacy-user", Role: provider.RoleUser, Content: "legacy"})
132 if err := legacy.Save(legacyPath); err != nil {
133 t.Fatal(err)
134 }
135 payload, err := json.Marshal(map[string]any{"messages": []provider.Message{{ID: "event-user", Role: provider.RoleUser, Content: "events"}}})
136 if err != nil {
137 t.Fatal(err)
138 }
139 targetRoot := filepath.Join(root, "sessions-v4")
140 writePrototypeStore(t, filepath.Join(targetRoot, agent.BranchID(legacyPath)), []Event{{Kind: "context/replace", Payload: payload}}, "")
141 if _, err := importSourceForLegacy(t.Context(), legacyPath, targetRoot, ""); !errors.Is(err, ErrImportConflict) {
142 t.Fatalf("conflicting import = %v", err)
143 }
144 // Classification happens before publication: a refused import must not have
145 // adopted the transcript as an executable target. The event sidecar source
146 // directory is the only pre-existing entry under the target root.
147 assertNoMigrationTarget(t, targetRoot, legacyPath, legacyDir)
148 }
149
150 // assertNoMigrationTarget proves the frozen legacy head was never published.
151 // The paired sidecar source is an input, not a target, so it is expected.
152 func assertNoMigrationTarget(t *testing.T, targetRoot, legacyPath, legacyDir string) {
153 t.Helper()
154 sourceSHA, err := sourceDigestForTest(legacyPath, legacyDir)
155 if err != nil {
156 t.Fatal(err)
157 }
158 targetID := migrationTargetID(legacyPath, sourceSHA, "")
159 if _, statErr := os.Stat(filepath.Join(targetRoot, targetID)); statErr == nil {
160 t.Fatalf("legacy target %q was published before classification", targetID)
161 }
162 }
163
164 // sourceDigestForTest recomputes the frozen source digest the same way the
165 // migration path does.
166 func sourceDigestForTest(sourcePath, sourceDir string) (string, error) {
167 artifacts, source, freezeDir, err := freezeLegacyArtifacts(context.Background(), sourcePath)
168 if err != nil {
169 return "", err
170 }
171 defer os.RemoveAll(freezeDir)
172 _ = artifacts
173 _ = sourceDir
174 return source.SHA256, nil
175 }
176
177 func TestContinueImportedPublishesOnlyTheSelectedSource(t *testing.T) {
178 root := t.TempDir()
179 legacyDir := filepath.Join(root, "sessions")
180 legacyPath := filepath.Join(legacyDir, "paired.jsonl")
181 if err := os.MkdirAll(legacyDir, 0o700); err != nil {
182 t.Fatal(err)
183 }
184 legacy := agent.NewSession("system")
185 legacy.Add(provider.Message{ID: "shared", Role: provider.RoleUser, Content: "first"})
186 if err := legacy.Save(legacyPath); err != nil {
187 t.Fatal(err)
188 }
189 // The sidecar holds the transcript's own prefix plus one newer message, so
190 // it is the only source that can be resumed without losing work.
191 previewMessages := append(legacy.Snapshot(), provider.Message{ID: "newer", Role: provider.RoleAssistant, Content: "second"})
192 payload, err := json.Marshal(map[string]any{"messages": previewMessages})
193 if err != nil {
194 t.Fatal(err)
195 }
196 targetRoot := filepath.Join(root, "sessions-v4")
197 writePrototypeStore(t, filepath.Join(targetRoot, agent.BranchID(legacyPath)), []Event{{Kind: "context/replace", Payload: payload}}, "")
198 result, err := importSourceForLegacy(t.Context(), legacyPath, targetRoot, "")
199 if err != nil {
200 t.Fatal(err)
201 }
202 if result.Kind != "events" {
203 t.Fatalf("selected kind = %q, want events", result.Kind)
204 }
205 // The transcript target must not exist: only the selected source is built.
206 assertNoMigrationTarget(t, targetRoot, legacyPath, legacyDir)
207 messages, err := projectedMessagesForTest(t, targetRoot, result.TargetID)
208 if err != nil {
209 t.Fatal(err)
210 }
211 if len(messages) != len(previewMessages) || messages[len(messages)-1].ID != "newer" {
212 t.Fatalf("selected projection has %d messages, want the newer sidecar history", len(messages))
213 }
214 }
215
216 func projectedMessagesForTest(t *testing.T, targetRoot, sessionID string) ([]provider.Message, error) {
217 t.Helper()
218 handle, err := NewFilesystemPersistence(targetRoot).Open(sessionID, ReadOnly)
219 if err != nil {
220 return nil, err
221 }
222 defer handle.Close(context.Background())
223 projection := Projection{}
224 var cursor uint64
225 for {
226 page, readErr := handle.Read(t.Context(), cursor, 1000)
227 if readErr != nil {
228 return nil, readErr
229 }
230 for _, commit := range page.Commits {
231 if applyErr := applyProjectionCommit(&projection, commit); applyErr != nil {
232 return nil, applyErr
233 }
234 }
235 if !page.Truncated {
236 return projection.Messages, nil
237 }
238 cursor = page.Next
239 }
240 }
241
242 func TestPreviewFreezeRefusesExactWriterOwnership(t *testing.T) {
243 dir := filepath.Join(t.TempDir(), "preview")
244 writePrototypeStore(t, dir, nil, "")
245 release, err := filelock.Acquire(t.Context(), filepath.Join(dir, "writer.lock"))
246 if err != nil {
247 t.Fatal(err)
248 }
249 defer release()
250 if _, err := freezePreview(t.Context(), dir); !errors.Is(err, ErrWriterOwned) {
251 t.Fatalf("freeze error = %v, want ErrWriterOwned", err)
252 }
253 }
254
255 func TestContinueStoredPreviewUpgradesLinearV3ToFinalCodec(t *testing.T) {
256 root := filepath.Join(t.TempDir(), "sessions-v4")
257 oldDir := filepath.Join(root, "old-linear")
258 store, err := CreateStore(oldDir, "old-linear")
259 if err != nil {
260 t.Fatal(err)
261 }
262 payload, _ := json.Marshal(map[string]any{"message": provider.Message{ID: "user", Role: provider.RoleUser, Content: "hello"}})
263 if _, err := store.Append(t.Context(), Batch{OperationID: "message", Events: []Event{{Kind: "message/complete", Payload: payload}}}); err != nil {
264 t.Fatal(err)
265 }
266 if err := store.Close(t.Context()); err != nil {
267 t.Fatal(err)
268 }
269 commits, err := Replay(oldDir, nil)
270 if err != nil {
271 t.Fatal(err)
272 }
273 manifest := Manifest{SchemaVersion: 3, Codec: LegacyLinearCodec, SessionID: "old-linear", CreatedAt: time.Now().UTC(), WriterGeneration: 1}
274 manifestBytes, _ := json.Marshal(manifest)
275 if err := os.WriteFile(filepath.Join(oldDir, "manifest.json"), append(manifestBytes, '\n'), 0o600); err != nil {
276 t.Fatal(err)
277 }
278 var log bytes.Buffer
279 for _, commit := range commits {
280 commit.SchemaVersion = 3
281 commit.Codec = LegacyLinearCodec
282 line, _ := json.Marshal(commit)
283 log.Write(line)
284 log.WriteByte('\n')
285 }
286 if err := os.WriteFile(filepath.Join(oldDir, "events.jsonl"), log.Bytes(), 0o600); err != nil {
287 t.Fatal(err)
288 }
289
290 service, err := NewService("local", NewFilesystemPersistence(root))
291 if err != nil {
292 t.Fatal(err)
293 }
294 t.Cleanup(func() { _ = service.CloseAll(context.Background()) })
295 runtime, result, err := service.ContinueStoredPreview(t.Context(), "old-linear")
296 if err != nil {
297 t.Fatal(err)
298 }
299 t.Cleanup(func() { _ = service.Close(context.Background(), runtime.Ref()) })
300 if result.Source.Version != LegacyLinearCodec || runtime.Ref().SessionID == "old-linear" {
301 t.Fatalf("upgrade = %+v, ref = %+v", result, runtime.Ref())
302 }
303 if got := runtime.Session().Snapshot().Projection.ModelMessages; len(got) != 1 || got[0].ID != "user" {
304 t.Fatalf("upgraded messages = %+v", got)
305 }
306 }
307
308 func TestContinueStoredPreviewUpgradesUnpublishedV4Draft(t *testing.T) {
309 root := filepath.Join(t.TempDir(), "sessions-v4")
310 draftDir := filepath.Join(root, "draft-v4")
311 store, err := CreateStore(draftDir, "draft-v4")
312 if err != nil {
313 t.Fatal(err)
314 }
315 payload, _ := json.Marshal(map[string]any{"message": provider.Message{ID: "user", Role: provider.RoleUser, Content: "hello"}})
316 if _, err := store.Append(t.Context(), Batch{OperationID: "message", Events: []Event{{Kind: "message/complete", Payload: payload}}}); err != nil {
317 t.Fatal(err)
318 }
319 if err := store.Close(t.Context()); err != nil {
320 t.Fatal(err)
321 }
322 manifest, err := readManifest(filepath.Join(draftDir, "manifest.json"))
323 if err != nil {
324 t.Fatal(err)
325 }
326 manifest.StorageRevision = 0
327 if err := writeManifestFile(filepath.Join(draftDir, "manifest.json"), manifest); err != nil {
328 t.Fatal(err)
329 }
330 if _, err := Open(draftDir, "draft-v4"); !errors.Is(err, ErrUnsupportedVersion) {
331 t.Fatalf("direct draft Open error = %v", err)
332 }
333
334 service, err := NewService("local", NewFilesystemPersistence(root))
335 if err != nil {
336 t.Fatal(err)
337 }
338 t.Cleanup(func() { _ = service.CloseAll(context.Background()) })
339 runtime, result, err := service.ContinueStoredPreview(t.Context(), "draft-v4")
340 if err != nil {
341 t.Fatal(err)
342 }
343 t.Cleanup(func() { _ = service.Close(context.Background(), runtime.Ref()) })
344 if result.Source.Version != Codec || runtime.Ref().SessionID == "draft-v4" {
345 t.Fatalf("upgrade = %+v, ref = %+v", result, runtime.Ref())
346 }
347 upgraded, err := readManifest(filepath.Join(root, runtime.Ref().SessionID, "manifest.json"))
348 if err != nil {
349 t.Fatal(err)
350 }
351 if upgraded.StorageRevision != StorageRevision {
352 t.Fatalf("storage revision = %d, want %d", upgraded.StorageRevision, StorageRevision)
353 }
354 if got := runtime.Session().Snapshot().Projection.ModelMessages; len(got) != 1 || got[0].ID != "user" {
355 t.Fatalf("upgraded messages = %+v", got)
356 }
357 var original Manifest
358 originalBytes, err := os.ReadFile(filepath.Join(draftDir, "manifest.json"))
359 if err == nil {
360 err = json.Unmarshal(originalBytes, &original)
361 }
362 if err != nil || original.StorageRevision != 0 {
363 t.Fatalf("draft source changed: %+v, %v", original, err)
364 }
365 }
366
367 func TestPrototypeRequiresExplicitImportAndPreservesTornTail(t *testing.T) {
368 root := t.TempDir()
369 source := filepath.Join(root, "prototype")
370 payload, _ := json.Marshal(map[string]any{"messages": []provider.Message{{ID: "message-1", Role: provider.RoleUser, Content: "hello"}}, "reason": "prototype"})
371 writePrototypeStore(t, source, []Event{{Kind: "context/replace", Payload: payload}}, `{"torn":`)
372 if _, err := Open(source, "prototype"); !errors.Is(err, ErrUnsupportedVersion) {
373 t.Fatalf("direct prototype Open error = %v", err)
374 }
375
376 result, err := ImportPrototype(t.Context(), source, filepath.Join(root, "final"))
377 if err != nil {
378 t.Fatal(err)
379 }
380 if result.ImportedEvents != 1 {
381 t.Fatalf("imported events = %d", result.ImportedEvents)
382 }
383 store, err := Open(result.TargetDir, result.TargetID)
384 if err != nil {
385 t.Fatal(err)
386 }
387 projection := store.Snapshot().Projection
388 if len(projection.Messages) != 1 || projection.Messages[0].ID != "message-1" || len(projection.ModelMessages) != 1 {
389 t.Fatalf("projection = %+v", projection)
390 }
391 if err := store.Close(context.Background()); err != nil {
392 t.Fatal(err)
393 }
394 frozen, err := os.ReadFile(filepath.Join(result.TargetDir, "legacy", "prototype", "events.jsonl"))
395 if err != nil {
396 t.Fatal(err)
397 }
398 if string(frozen[len(frozen)-len(`{"torn":`):]) != `{"torn":` {
399 t.Fatalf("prototype tail was not preserved: %q", frozen)
400 }
401
402 reused, err := ImportPrototype(t.Context(), source, filepath.Join(root, "final"))
403 if err != nil || !reused.Reused || reused.TargetID != result.TargetID {
404 t.Fatalf("reused import = %+v, %v", reused, err)
405 }
406 }
407
408 func TestPrototypeImportAcceptsLegacyRecordLargerThan64MiB(t *testing.T) {
409 if testing.Short() {
410 t.Skip("capacity regression")
411 }
412 root := t.TempDir()
413 source := filepath.Join(root, "prototype-large-record")
414 large := strings.Repeat("x", (64<<20)+(1<<20))
415 payload, err := json.Marshal(map[string]any{"message": provider.Message{ID: "large-message", Role: provider.RoleAssistant, Content: large}})
416 if err != nil {
417 t.Fatal(err)
418 }
419 writePrototypeStore(t, source, []Event{{Kind: "message/complete", Payload: payload}}, "")
420 result, err := ImportPrototype(t.Context(), source, filepath.Join(root, "final"))
421 if err != nil {
422 t.Fatal(err)
423 }
424 store, err := Open(result.TargetDir, result.TargetID)
425 if err != nil {
426 t.Fatal(err)
427 }
428 defer store.Close(context.Background())
429 messages := store.Snapshot().Projection.Messages
430 if len(messages) != 1 || messages[0].Content != large {
431 t.Fatalf("large legacy record round trip = %d messages / %d bytes", len(messages), func() int {
432 if len(messages) == 0 {
433 return 0
434 }
435 return len(messages[0].Content)
436 }())
437 }
438 }
439
440 func TestPrototypeImportRejectsUnknownRequiredEvent(t *testing.T) {
441 root := t.TempDir()
442 source := filepath.Join(root, "prototype")
443 writePrototypeStore(t, source, []Event{{Kind: "future/required"}}, "")
444 if _, err := ImportPrototype(t.Context(), source, filepath.Join(root, "final")); !errors.Is(err, ErrUnsupportedVersion) {
445 t.Fatalf("unknown required import error = %v", err)
446 }
447 entries, err := os.ReadDir(filepath.Join(root, "final"))
448 if err != nil && !os.IsNotExist(err) {
449 t.Fatal(err)
450 }
451 for _, entry := range entries {
452 if !entry.IsDir() || entry.Name()[0] != '.' {
453 t.Fatalf("failed import published %q", entry.Name())
454 }
455 }
456 }
457
458 func TestImportMessageComparisonIgnoresNonAuthoritativeToolRecovery(t *testing.T) {
459 base := []provider.Message{{
460 Role: provider.RoleAssistant,
461 ID: "assistant-tool",
462 ToolCalls: []provider.ToolCall{{
463 ID: "call-1", Name: "read_file", Arguments: `{"path":"input.txt"}`,
464 }},
465 }}
466 legacy := append([]provider.Message(nil), base...)
467 legacy[0].ToolCalls = append([]provider.ToolCall(nil), base[0].ToolCalls...)
468 legacy[0].ToolCalls[0].Recovery = &provider.ToolCallRecord{}
469 legacy[0].WorkDurationMs = 41
470
471 if !messagesEqual(legacy, base) || !messagesPrefix(legacy, base) {
472 t.Fatal("process-local recovery metadata created a false import conflict")
473 }
474 if !messagesPrefix([]provider.Message{}, base) {
475 t.Fatal("an empty legacy transcript was not recognized as an event-history prefix")
476 }
477 }
478
478 lines GO