mirror of
https://github.com/esengine/DeepSeek-Reasonix.git
synced 2026-11-07 02:36:01 -10:00
Merge pull request #14213 from chuiwenwei/fix/import-completes-the-sidecar
fix(migrate): an imported session completes its sidecar
This commit is contained in:
6 files changed
+551
-30
No files matched your search
@@ -263,6 +263,16 @@ func importJsonlSessions(entries []os.DirEntry, srcDir, globalDest string, hasEv
|
||||
if err := copySubagentArtifacts(srcDir, destDir, base); err != nil {
|
||||
hadArtifactFailure = true
|
||||
}
|
||||
// The transcript lands before the sidecar that carries its
|
||||
// meaning; a stopped run leaves it here, so finish the sidecar
|
||||
// instead of treating the session as done.
|
||||
wrote, err := completeBranchMetaSidecar(jsonlPath, dest)
|
||||
if err != nil {
|
||||
hadArtifactFailure = true
|
||||
} else if wrote {
|
||||
recordImportedTitle(destDir, base, summary)
|
||||
imported++
|
||||
}
|
||||
}
|
||||
continue
|
||||
}
|
||||
@@ -276,7 +286,12 @@ func importJsonlSessions(entries []os.DirEntry, srcDir, globalDest string, hasEv
|
||||
_ = os.Chtimes(dest, srcInfo.ModTime(), srcInfo.ModTime())
|
||||
}
|
||||
if copyBranchMeta {
|
||||
copyBranchMetaSidecar(jsonlPath, dest)
|
||||
if err := copyBranchMetaSidecar(jsonlPath, dest); err != nil {
|
||||
// The transcript is there but the session's meaning is not:
|
||||
// leave the import incomplete so the next run finishes it.
|
||||
hadArtifactFailure = true
|
||||
continue
|
||||
}
|
||||
if err := copySubagentArtifacts(srcDir, destDir, base); err != nil {
|
||||
hadArtifactFailure = true
|
||||
}
|
||||
@@ -716,6 +731,13 @@ func rehomeStrandedSessions(srcDir, globalDest, marker string, projectDir func(s
|
||||
if err := copySubagentArtifacts(srcDir, destDir, base); err != nil {
|
||||
hadCopyFailure = true
|
||||
}
|
||||
// The previous boot may have published the transcript and been
|
||||
// stopped before the sidecar; finish that instead of skipping it.
|
||||
if wrote, err := completeBranchMetaSidecar(srcPath, dest); err != nil {
|
||||
hadCopyFailure = true
|
||||
} else if wrote {
|
||||
recordImportedTitle(destDir, base, summary)
|
||||
}
|
||||
continue // already routed on a previous boot
|
||||
}
|
||||
if err := transformAndCopyJsonl(srcPath, dest); err != nil {
|
||||
@@ -723,7 +745,9 @@ func rehomeStrandedSessions(srcDir, globalDest, marker string, projectDir func(s
|
||||
continue
|
||||
}
|
||||
_ = os.Chtimes(dest, info.ModTime(), info.ModTime()) // preserve resume ordering
|
||||
copyBranchMetaSidecar(srcPath, dest)
|
||||
if err := copyBranchMetaSidecar(srcPath, dest); err != nil {
|
||||
hadCopyFailure = true
|
||||
}
|
||||
if err := copySubagentArtifacts(srcDir, destDir, base); err != nil {
|
||||
hadCopyFailure = true
|
||||
}
|
||||
@@ -769,32 +793,37 @@ func strandedSessionDestDir(srcDir, srcPath, base string, projectDir func(string
|
||||
// desktop sidebar keeps the session's title, topic, and tree position. Best
|
||||
// effort: a missing or unreadable sidecar just means the session shows with a
|
||||
// generated title.
|
||||
func copyBranchMetaSidecar(srcPath, dstPath string) {
|
||||
func copyBranchMetaSidecar(srcPath, dstPath string) error {
|
||||
b, err := os.ReadFile(BranchMetaPath(srcPath))
|
||||
if err != nil {
|
||||
return
|
||||
if errors.Is(err, os.ErrNotExist) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
dstMeta := BranchMetaPath(dstPath)
|
||||
if err := os.MkdirAll(filepath.Dir(dstMeta), 0o755); err != nil {
|
||||
return
|
||||
return err
|
||||
}
|
||||
tmp, err := os.CreateTemp(filepath.Dir(dstMeta), ".branch.*.tmp")
|
||||
if err != nil {
|
||||
return
|
||||
return err
|
||||
}
|
||||
tmpPath := tmp.Name()
|
||||
if _, err := tmp.Write(b); err != nil {
|
||||
tmp.Close()
|
||||
os.Remove(tmpPath)
|
||||
return
|
||||
return err
|
||||
}
|
||||
if err := tmp.Close(); err != nil {
|
||||
os.Remove(tmpPath)
|
||||
return
|
||||
return err
|
||||
}
|
||||
if err := os.Rename(tmpPath, dstMeta); err != nil {
|
||||
os.Remove(tmpPath)
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func copySubagentArtifacts(srcSessionDir, dstSessionDir, parentSession string) error {
|
||||
@@ -862,20 +891,6 @@ func copyFileIfExists(src, dst string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// sameDirPath reports whether two directory paths resolve to the same location.
|
||||
func sameDirPath(a, b string) bool {
|
||||
ca, cb := filepath.Clean(a), filepath.Clean(b)
|
||||
if ca == cb {
|
||||
return true
|
||||
}
|
||||
if aa, err := filepath.Abs(ca); err == nil {
|
||||
if bb, err := filepath.Abs(cb); err == nil {
|
||||
return aa == bb
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// reconstructSession folds the chronological event stream into the provider
|
||||
// message sequence. Tool results inherit their tool name from the assistant turn
|
||||
// that issued the call (the v0.x result event carries only the call id).
|
||||
|
||||
@@ -32,3 +32,25 @@ func linkFileNoReplace(src, dst string) error {
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// completeBranchMetaSidecar finishes a transcript whose sidecar an interrupted
|
||||
// import never published. It reports whether it wrote one. A sidecar that is
|
||||
// already there — readable or not — is never rewritten, so the ordinary retry of
|
||||
// a finished session copies nothing, and a corrupt sidecar is not clobbered by a
|
||||
// source that has one too.
|
||||
func completeBranchMetaSidecar(srcPath, dstPath string) (bool, error) {
|
||||
// LoadBranchMeta is the pre-existing reader: ok means the sidecar is there,
|
||||
// an error means it exists but cannot be trusted, and only ok == false with
|
||||
// no error means there is nothing to preserve.
|
||||
_, ok, err := LoadBranchMeta(dstPath)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
if ok {
|
||||
return false, nil
|
||||
}
|
||||
if err := copyBranchMetaSidecar(srcPath, dstPath); err != nil {
|
||||
return false, err
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
@@ -168,3 +168,17 @@ func readDirUnder(dir, name string) ([]os.DirEntry, error) {
|
||||
defer f.Close()
|
||||
return f.ReadDir(-1)
|
||||
}
|
||||
|
||||
// sameDirPath reports whether two directory paths resolve to the same location.
|
||||
func sameDirPath(a, b string) bool {
|
||||
ca, cb := filepath.Clean(a), filepath.Clean(b)
|
||||
if ca == cb {
|
||||
return true
|
||||
}
|
||||
if aa, err := filepath.Abs(ca); err == nil {
|
||||
if bb, err := filepath.Abs(cb); err == nil {
|
||||
return aa == bb
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -1,6 +1,7 @@
|
||||
package sessionstore
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"os"
|
||||
@@ -1180,3 +1181,198 @@ func TestTransformAndCopyJsonlFlattensNestedToolCalls(t *testing.T) {
|
||||
t.Errorf("final content = %q", msgs[3].Content)
|
||||
}
|
||||
}
|
||||
|
||||
// interruptedFlatImport is the state an import killed between its snapshot save
|
||||
// (the transcript) and copyBranchMetaSidecar leaves: the transcript in its routed
|
||||
// project dir, nothing beside it, and the 1.x sidecar still only in the source.
|
||||
func interruptedFlatImport(t *testing.T, archived bool) (src, global, proj, ws, dest string) {
|
||||
t.Helper()
|
||||
root := testenv.TempDir(t)
|
||||
src = filepath.Join(root, "src")
|
||||
global = filepath.Join(root, "global")
|
||||
ws = filepath.Join(root, "ws")
|
||||
proj = filepath.Join(root, "proj")
|
||||
for _, d := range []string{src, global, ws, proj} {
|
||||
if err := os.MkdirAll(d, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
srcPath := filepath.Join(src, "chat-1.jsonl")
|
||||
if err := os.WriteFile(srcPath, []byte(legacyMessageLog), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := SaveBranchMeta(srcPath, BranchMeta{
|
||||
Scope: "project", WorkspaceRoot: ws, TopicTitle: "1.x topic",
|
||||
Archived: archived, Pinned: true, ParentID: "parent-1",
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
dest = filepath.Join(proj, "chat-1.jsonl")
|
||||
if err := os.WriteFile(dest, []byte(legacyMessageLog), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return src, global, proj, ws, dest
|
||||
}
|
||||
|
||||
// The transcript is published before the sidecar that carries its meaning, so a
|
||||
// kill in that window leaves a transcript with no meta. The next run must finish
|
||||
// the sidecar; skipping the session loses its scope and returns an archived
|
||||
// conversation to the active list.
|
||||
func TestInterruptedFlatImportRetryCompletesTheBranchMeta(t *testing.T) {
|
||||
src, global, proj, ws, dest := interruptedFlatImport(t, true)
|
||||
|
||||
if n, err := MigrateLegacySessions(src, global, func(string) string { return proj }); err != nil || n != 1 {
|
||||
t.Fatalf("retry: n=%d err=%v, want 1 nil", n, err)
|
||||
}
|
||||
got, ok, err := LoadBranchMeta(dest)
|
||||
if err != nil || !ok {
|
||||
t.Fatalf("retry left the transcript without a sidecar: ok=%v err=%v", ok, err)
|
||||
}
|
||||
if got.Scope != "project" || got.WorkspaceRoot != ws || got.TopicTitle != "1.x topic" ||
|
||||
!got.Archived || !got.Pinned || got.ParentID != "parent-1" {
|
||||
t.Fatalf("completed sidecar = %+v, want the 1.x fields", got)
|
||||
}
|
||||
|
||||
// Exact output: the completed sidecar is byte-for-byte what a fresh import
|
||||
// of the same source writes, including the title it records.
|
||||
fresh := filepath.Join(filepath.Dir(src), "fresh")
|
||||
if err := os.MkdirAll(fresh, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n, _, err := ImportLegacySessionsFromExplicitDir(src, fresh, func(string) string { return fresh }); err != nil || n != 1 {
|
||||
t.Fatalf("fresh import: n=%d err=%v", n, err)
|
||||
}
|
||||
wantMeta, err := os.ReadFile(BranchMetaPath(filepath.Join(fresh, "chat-1.jsonl")))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
gotMeta, err := os.ReadFile(BranchMetaPath(dest))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !bytes.Equal(gotMeta, wantMeta) {
|
||||
t.Fatalf("completed sidecar differs from a fresh import:\n got %s\nwant %s", gotMeta, wantMeta)
|
||||
}
|
||||
wantTitles, err := os.ReadFile(filepath.Join(fresh, ".titles.json"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
gotTitles, err := os.ReadFile(filepath.Join(proj, ".titles.json"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !bytes.Equal(gotTitles, wantTitles) {
|
||||
t.Fatalf("completed title differs from a fresh import:\n got %s\nwant %s", gotTitles, wantTitles)
|
||||
}
|
||||
}
|
||||
|
||||
// A finished import is done: the ordinary retry must copy nothing over the
|
||||
// transcript and must leave the sidecar exactly as it was.
|
||||
func TestCompletedFlatImportIsNotReCopied(t *testing.T) {
|
||||
root := testenv.TempDir(t)
|
||||
src := filepath.Join(root, "src")
|
||||
global := filepath.Join(root, "global")
|
||||
ws := filepath.Join(root, "ws")
|
||||
proj := filepath.Join(root, "proj")
|
||||
for _, d := range []string{src, global, ws, proj} {
|
||||
if err := os.MkdirAll(d, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
srcPath := filepath.Join(src, "chat-1.jsonl")
|
||||
if err := os.WriteFile(srcPath, []byte(legacyMessageLog), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := SaveBranchMeta(srcPath, BranchMeta{
|
||||
Scope: "project", WorkspaceRoot: ws, TopicTitle: "1.x topic", Archived: true,
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
router := func(string) string { return proj }
|
||||
if n, _, err := ImportLegacySessionsFromExplicitDir(src, global, router); err != nil || n != 1 {
|
||||
t.Fatalf("first import: n=%d err=%v", n, err)
|
||||
}
|
||||
dest := filepath.Join(proj, "chat-1.jsonl")
|
||||
pinned := time.Now().Add(-2 * time.Hour).Truncate(time.Second)
|
||||
touched := []string{dest, BranchMetaPath(dest), filepath.Join(proj, ".titles.json")}
|
||||
for _, p := range touched {
|
||||
if err := os.Chtimes(p, pinned, pinned); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
before, err := os.ReadFile(dest)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
beforeMeta, err := os.ReadFile(BranchMetaPath(dest))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// An explicit import re-runs every pass, so this reaches the destination
|
||||
// branch that must recognize a completed sidecar.
|
||||
if n, _, err := ImportLegacySessionsFromExplicitDir(src, global, router); err != nil || n != 0 {
|
||||
t.Fatalf("retry: n=%d err=%v, want 0 nil", n, err)
|
||||
}
|
||||
after, err := os.ReadFile(dest)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
afterMeta, err := os.ReadFile(BranchMetaPath(dest))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !bytes.Equal(before, after) || !bytes.Equal(beforeMeta, afterMeta) {
|
||||
t.Fatal("a completed import was re-copied")
|
||||
}
|
||||
for _, p := range touched {
|
||||
info, err := os.Stat(p)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !info.ModTime().Equal(pinned) {
|
||||
t.Fatalf("%s was rewritten on a completed retry: mtime %s want %s", p, info.ModTime(), pinned)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A sidecar that cannot be written must not let the import claim the session:
|
||||
// the marker stays off so the next run retries and completes it.
|
||||
func TestFlatImportWithholdsMarkerWhenTheSidecarCannotBeWritten(t *testing.T) {
|
||||
src, global, proj, ws, dest := interruptedFlatImport(t, true)
|
||||
blocked := BranchMetaPath(dest)
|
||||
if err := os.Mkdir(blocked, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
n, err := MigrateLegacySessions(src, global, func(string) string { return proj })
|
||||
if err != nil {
|
||||
t.Fatalf("migrate: %v", err)
|
||||
}
|
||||
if n != 0 {
|
||||
t.Fatalf("imported %d although the sidecar could not be written", n)
|
||||
}
|
||||
for _, marker := range []string{legacyRoutedHomeImportMarker, legacyJsonlPassMarker} {
|
||||
if importMarkerExists(global, marker) {
|
||||
t.Fatalf("marker %q stamped after a sidecar write failed; the session will never be retried", marker)
|
||||
}
|
||||
}
|
||||
|
||||
// Repair the destination and let the next boot finish the session.
|
||||
if err := os.Remove(blocked); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n, err := MigrateLegacySessions(src, global, func(string) string { return proj }); err != nil || n != 1 {
|
||||
t.Fatalf("retry: n=%d err=%v, want 1 nil", n, err)
|
||||
}
|
||||
meta, ok, err := LoadBranchMeta(dest)
|
||||
if err != nil || !ok {
|
||||
t.Fatalf("retry left no sidecar: ok=%v err=%v", ok, err)
|
||||
}
|
||||
if !meta.Archived || meta.Scope != "project" || meta.WorkspaceRoot != ws {
|
||||
t.Fatalf("completed sidecar = %+v", meta)
|
||||
}
|
||||
if !importMarkerExists(global, legacyRoutedHomeImportMarker) {
|
||||
t.Fatal("marker missing after the sidecar was completed")
|
||||
}
|
||||
}
|
||||
@@ -12,6 +12,7 @@ import (
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
"reasonix/internal/contract/provider"
|
||||
"reasonix/internal/state/sessionv4"
|
||||
)
|
||||
|
||||
@@ -89,7 +90,16 @@ func PrepareSessionPath(path string) error {
|
||||
return importV4(path, s)
|
||||
}
|
||||
meta, ok, err := LoadBranchMeta(path)
|
||||
if err != nil || !ok || meta.ImportedFrom == nil || meta.ImportedFrom.LogSize == s.LogSize {
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
if !ok || meta.ImportedFrom == nil {
|
||||
// The transcript landed but the sidecar that records what the
|
||||
// conversation means did not: finish it before the copy is used.
|
||||
_, err := completeV4Import(path, s, BranchMeta{})
|
||||
return err
|
||||
}
|
||||
if meta.ImportedFrom.LogSize == s.LogSize {
|
||||
return nil
|
||||
}
|
||||
current, err := loadSessionUnlocked(path)
|
||||
@@ -110,7 +120,7 @@ func PrepareSessionPath(path string) error {
|
||||
// This path refreshes a copy that already exists, so its meta already carries
|
||||
// whatever the first import decided — including the archived mark. Passing an
|
||||
// empty start is what keeps that decision rather than resetting it.
|
||||
return saveImportMeta(path, s, t, BranchMeta{})
|
||||
return saveImportMeta(path, s, t, BranchMeta{}, t.Messages)
|
||||
}
|
||||
|
||||
func importV4(path string, s sessionv4.Session) error {
|
||||
@@ -129,7 +139,7 @@ func importV4As(path string, s sessionv4.Session, start BranchMeta) error {
|
||||
if err := sess.SaveIfAbsent(path); err != nil && !os.IsExist(err) {
|
||||
return fmt.Errorf("import 1.x session %s: %w", s.Manifest.SessionID, err)
|
||||
}
|
||||
return saveImportMeta(path, s, t, start)
|
||||
return saveImportMeta(path, s, t, start, t.Messages)
|
||||
}
|
||||
|
||||
// importedTitle is the name 1.x showed for the conversation: the title its log
|
||||
@@ -161,12 +171,14 @@ func healImportedTitle(path string, s sessionv4.Session) {
|
||||
_ = SaveBranchMeta(path, meta)
|
||||
}
|
||||
|
||||
func saveImportMeta(path string, s sessionv4.Session, t sessionv4.Transcript, start BranchMeta) error {
|
||||
func saveImportMeta(path string, s sessionv4.Session, t sessionv4.Transcript, start BranchMeta, content []provider.Message) error {
|
||||
digest, err := DigestSessionMessages(t.Messages)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
preview, turns := SessionPreviewFromMessages(t.Messages)
|
||||
// content is what the copy holds, which is the source's transcript on a
|
||||
// first import and may be a superset of it on a completed interrupted one.
|
||||
preview, turns := SessionPreviewFromMessages(content)
|
||||
meta, _, _ := LoadBranchMeta(path)
|
||||
meta.ID = BranchID(path)
|
||||
meta.CreatedAt = s.Manifest.CreatedAt
|
||||
@@ -185,6 +197,86 @@ func saveImportMeta(path string, s sessionv4.Session, t sessionv4.Transcript, st
|
||||
return SaveBranchMeta(path, meta)
|
||||
}
|
||||
|
||||
// completeV4Import finishes a copy an interrupted run left behind: the transcript
|
||||
// landed, but the sidecar that carries what it cannot say — archive mark, title,
|
||||
// pin, and source — did not. A copy with a sidecar is done and is left alone;
|
||||
// one without is completed here. It reports whether it wrote the sidecar.
|
||||
func completeV4Import(path string, s sessionv4.Session, start BranchMeta) (bool, error) {
|
||||
meta, ok, err := LoadBranchMeta(path)
|
||||
if err != nil {
|
||||
// A sidecar that exists but cannot be read is not one to rewrite on an
|
||||
// import pass; leave it for the recovery path.
|
||||
return false, nil
|
||||
}
|
||||
if ok && meta.ImportedFrom != nil {
|
||||
healImportedTitle(path, s)
|
||||
return false, nil
|
||||
}
|
||||
t, err := s.Transcript()
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("import 1.x session %s: %w", s.Manifest.SessionID, err)
|
||||
}
|
||||
if _, turns := SessionPreviewFromMessages(t.Messages); turns == 0 {
|
||||
// Nothing to bring over; writing counts derived from an empty source
|
||||
// would make a copy that does have turns look empty.
|
||||
return false, nil
|
||||
}
|
||||
content, err := absorbInterruptedCopy(path, t)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("import 1.x session %s: %w", s.Manifest.SessionID, err)
|
||||
}
|
||||
if err := saveImportMeta(path, s, t, start, content); err != nil {
|
||||
return false, fmt.Errorf("import 1.x session %s: %w", s.Manifest.SessionID, err)
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// absorbInterruptedCopy brings an interrupted import's copy up to the source
|
||||
// while it is still only a prefix of it. A copy the person continued in 2.x is
|
||||
// left as it is: its extra turns are not in 1.x, and replaying over it would
|
||||
// lose them. It returns the messages the copy holds afterwards, so the sidecar
|
||||
// describes the conversation that exists rather than the one in the source.
|
||||
func absorbInterruptedCopy(path string, t sessionv4.Transcript) ([]provider.Message, error) {
|
||||
// sessionArtifactExists is the store's reader for this question. It answers
|
||||
// bool, so missing and unreadable look alike; a copy that is there but
|
||||
// cannot be read still fails when it is loaded below.
|
||||
if !sessionArtifactExists(path) {
|
||||
sess := NewSession("")
|
||||
sess.Messages = t.Messages
|
||||
if err := sess.SaveIfAbsent(path); err != nil && !os.IsExist(err) {
|
||||
return nil, err
|
||||
}
|
||||
return t.Messages, nil
|
||||
}
|
||||
current, err := loadSessionUnlocked(path)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
copied := current.Snapshot()
|
||||
if len(copied) >= len(t.Messages) || !isMessagePrefixOf(copied, t.Messages) {
|
||||
return copied, nil
|
||||
}
|
||||
current.Rewrite(t.Messages, "complete an interrupted import")
|
||||
if err := current.SaveRewrite(path); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return t.Messages, nil
|
||||
}
|
||||
|
||||
// isMessagePrefixOf reports whether copied is exactly the first len(copied)
|
||||
// messages of full, using the same digest the import records for its source.
|
||||
func isMessagePrefixOf(copied, full []provider.Message) bool {
|
||||
if len(copied) > len(full) {
|
||||
return false
|
||||
}
|
||||
have, err := DigestSessionMessages(copied)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
want, err := DigestSessionMessages(full[:len(copied)])
|
||||
return err == nil && have == want
|
||||
}
|
||||
|
||||
// withV4Sessions adds the 1.x v4 conversations beside dir that have no import
|
||||
// yet; an imported one is already in the list as its own transcript.
|
||||
func withV4Sessions(dir string, out []SessionInfo) []SessionInfo {
|
||||
@@ -242,8 +334,9 @@ func readV4Cwd(store fs.FS, name string) string {
|
||||
|
||||
// ImportV4From copies the 1.x v4 conversations under store into dir. With a route
|
||||
// each goes where it is placed, and one naming no project is skipped as SkipNoProject;
|
||||
// without one everything goes to dir. Imported and empty ones are skipped silently,
|
||||
// an unreadable one becomes a SkippedSession, and importedAs supplies each copy's meta.
|
||||
// without one everything goes to dir. A finished copy is left alone, an
|
||||
// interrupted one is completed, an empty one is skipped silently, an unreadable
|
||||
// one becomes a SkippedSession, and importedAs supplies each copy's meta.
|
||||
func ImportV4From(store fs.FS, dir string, route func(V4Origin) V4Placement, importedAs func(sessionID string) BranchMeta) (int, []SkippedSession, error) {
|
||||
entries, err := fs.ReadDir(store, ".")
|
||||
if err != nil {
|
||||
@@ -287,7 +380,18 @@ func ImportV4From(store fs.FS, dir string, route func(V4Origin) V4Placement, imp
|
||||
}
|
||||
path := v4ImportPath(target, id)
|
||||
if sessionArtifactExists(path) {
|
||||
healImportedTitle(path, s)
|
||||
start := BranchMeta{}
|
||||
if importedAs != nil {
|
||||
start = importedAs(id)
|
||||
}
|
||||
completed, err := completeV4Import(path, s, start)
|
||||
if err != nil {
|
||||
skip(e.Name(), classifyWriteSkip(err))
|
||||
continue
|
||||
}
|
||||
if completed {
|
||||
imported++
|
||||
}
|
||||
continue
|
||||
}
|
||||
t, err := s.Transcript()
|
||||
|
||||
@@ -333,3 +333,173 @@ func TestPrepareSessionPathKeepsTheArchivedMark(t *testing.T) {
|
||||
t.Fatal("refreshing the copy brought an archived conversation back")
|
||||
}
|
||||
}
|
||||
|
||||
// An interrupted v4 import leaves the transcript and no complete sidecar. The
|
||||
// next import must complete the sidecar — keeping 1.x's archive mark — and bring
|
||||
// the copy up to the source when 1.x went on without it. The sidecar may be
|
||||
// absent outright or be one an earlier import left without its source.
|
||||
func TestInterruptedV4ImportRetryCompletesTheSidecar(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
removeSidecar bool
|
||||
}{
|
||||
{"sidecar missing", true},
|
||||
{"sidecar without its import source", false},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
s, dir, sessions := v4Project(t)
|
||||
opened, err := sessionv4.Open(dir)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
tr, err := opened.Transcript()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
path := v4ImportPath(sessions, v4TestID)
|
||||
partial := NewSession("")
|
||||
partial.Messages = tr.Messages
|
||||
if err := partial.SaveIfAbsent(path); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if tc.removeSidecar {
|
||||
if err := os.Remove(BranchMetaPath(path)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
} else if meta, ok, err := LoadBranchMeta(path); err != nil || !ok || meta.ImportedFrom != nil {
|
||||
t.Fatalf("setup: sidecar ok=%v importedFrom=%v err=%v, want one without its source", ok, meta.ImportedFrom != nil, err)
|
||||
}
|
||||
// 1.x continues the conversation while the copy has no complete
|
||||
// sidecar.
|
||||
s.Batch(dir, v4fixture.Ended,
|
||||
v4fixture.Event{Kind: "message/complete", Payload: v4fixture.Msg("m9", "user", "continued in 1.x")},
|
||||
v4fixture.Event{Kind: "message/complete", Payload: v4fixture.Msg("m10", "assistant", "answer in 1.x")},
|
||||
)
|
||||
|
||||
n, skipped, err := ImportV4From(os.DirFS(s.Root), sessions, nil, func(string) BranchMeta {
|
||||
return BranchMeta{Archived: true}
|
||||
})
|
||||
if err != nil || len(skipped) != 0 {
|
||||
t.Fatalf("rerun: n=%d skipped=%v err=%v", n, skipped, err)
|
||||
}
|
||||
if n != 1 {
|
||||
t.Fatalf("rerun completed %d copies, want 1", n)
|
||||
}
|
||||
meta, ok, err := LoadBranchMeta(path)
|
||||
if err != nil || !ok {
|
||||
t.Fatalf("rerun left no sidecar: ok=%v err=%v", ok, err)
|
||||
}
|
||||
if !meta.Archived || meta.ImportedFrom == nil {
|
||||
t.Fatalf("completed sidecar = %+v, want archived with its import source", meta)
|
||||
}
|
||||
loaded, err := LoadSession(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if msgs := loaded.Snapshot(); len(msgs) != 6 || msgs[5].Content != "answer in 1.x" {
|
||||
t.Fatalf("copy brought up to %d messages, want 6 with 1.x's tail", len(msgs))
|
||||
}
|
||||
if meta, _, _ := LoadBranchMeta(path); !meta.Archived {
|
||||
t.Fatal("following 1.x dropped the archived mark")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// A copy with both its transcript and its sidecar is complete: a rerun must not
|
||||
// re-copy it or touch the sidecar.
|
||||
func TestCompletedV4ImportIsNotReCopied(t *testing.T) {
|
||||
s, _, sessions := v4Project(t)
|
||||
if n, skipped, err := ImportV4From(os.DirFS(s.Root), sessions, nil, nil); n != 1 || len(skipped) != 0 || err != nil {
|
||||
t.Fatalf("first import: n=%d skipped=%v err=%v", n, skipped, err)
|
||||
}
|
||||
path := v4ImportPath(sessions, v4TestID)
|
||||
pinned := time.Now().Add(-2 * time.Hour).Truncate(time.Second)
|
||||
for _, p := range []string{path, BranchMetaPath(path)} {
|
||||
if err := os.Chtimes(p, pinned, pinned); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
before, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
beforeMeta, err := os.ReadFile(BranchMetaPath(path))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if n, skipped, err := ImportV4From(os.DirFS(s.Root), sessions, nil, nil); n != 0 || len(skipped) != 0 || err != nil {
|
||||
t.Fatalf("rerun: n=%d skipped=%v err=%v, want 0 nil", n, skipped, err)
|
||||
}
|
||||
after, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
afterMeta, err := os.ReadFile(BranchMetaPath(path))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if string(before) != string(after) || string(beforeMeta) != string(afterMeta) {
|
||||
t.Fatal("a completed v4 import was re-copied")
|
||||
}
|
||||
for _, p := range []string{path, BranchMetaPath(path)} {
|
||||
info, err := os.Stat(p)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !info.ModTime().Equal(pinned) {
|
||||
t.Fatalf("%s was rewritten on a completed retry: mtime %s want %s", p, info.ModTime(), pinned)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Completing an interrupted import never replays over turns the person added in
|
||||
// 2.x, and the sidecar it writes describes the copy that exists.
|
||||
func TestInterruptedV4ImportKeepsA2xTurn(t *testing.T) {
|
||||
s, dir, sessions := v4Project(t)
|
||||
opened, err := sessionv4.Open(dir)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
tr, err := opened.Transcript()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
path := v4ImportPath(sessions, v4TestID)
|
||||
partial := NewSession("")
|
||||
partial.Messages = tr.Messages
|
||||
if err := partial.SaveIfAbsent(path); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The person continues in 2.x before the import gets its second run.
|
||||
current, err := loadSessionUnlocked(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
current.Add(provider.Message{Role: provider.RoleUser, Content: "from 2.x"})
|
||||
if err := current.SaveSnapshot(path); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if _, skipped, err := ImportV4From(os.DirFS(s.Root), sessions, nil, func(string) BranchMeta {
|
||||
return BranchMeta{Archived: true}
|
||||
}); err != nil || len(skipped) != 0 {
|
||||
t.Fatalf("rerun: skipped=%v err=%v", skipped, err)
|
||||
}
|
||||
loaded, err := LoadSession(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
msgs := loaded.Snapshot()
|
||||
if len(msgs) != 5 || !strings.Contains(msgs[4].Content, "from 2.x") {
|
||||
t.Fatalf("the 2.x turn was lost: %+v", msgs)
|
||||
}
|
||||
meta, ok, err := LoadBranchMeta(path)
|
||||
if err != nil || !ok || !meta.Archived {
|
||||
t.Fatalf("sidecar = %+v ok=%v err=%v, want it completed and archived", meta, ok, err)
|
||||
}
|
||||
if meta.Turns != 2 {
|
||||
t.Fatalf("sidecar lists %d turns, want the copy's 2", meta.Turns)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user