Restore legacy Desktop history without duplicate recovery copies / 完善旧版对话迁移并排除重复恢复副本 (#10395)

* Restore legacy Desktop history with durable migration receipts

Problem: Desktop upgrades can omit older conversations, duplicate converted
generations, or expose automatically generated conflict recovery copies.

Root cause: discovery depended on disposable metadata, legacy migration only
adopted selected heads, and repeated imports compared against mutable targets
without accounting for conversion provenance or recovery identity.

Fix: discover durable v1-v4 sources, stage all live legacy heads and preview
formats, reconcile proven conversion lineages, and persist per-source adoption
receipts with revision checkpoints. Preserve source bytes and independent
continuations while excluding automatic recovery artifacts and their conversions.

Verification: Desktop full module coverage passed in three measured partitions
and its cmd/internal packages. Migration race regressions passed, including
restart, deletion, interruption, source repair, recovery artifacts, and submission
identity. Root suite had one goal-driver deadline failure; clean baseline and
working-tree isolated reproductions passed without code or timeout changes.
Generated inventory, repository lint tests, formatting, and diff checks passed.

* Separate migration discovery and lineage reduction responsibilities

Problem: CI rejected oversized migration functions and an outdated loop form.
Root cause: source discovery, paired-store admission, and lineage reduction
were embedded in orchestration functions beyond repository complexity limits.
Fix: extract the existing decisions into focused helpers and use an integer
range while retaining restart diagnostics. Keep all CI gates and thresholds.
Verification: repository lint, Desktop golangci-lint, generated inventory
freshness, diff checks, and the focused Desktop migration race suite passed.
This commit is contained in:
SivanCola authored and GitHub committed 2026-09-16 15:51:15 +08:00
1 parent 8b426dc870
commit a4b83056a9
19 files changed
+3222 -88

No files matched your search

+244 -78
View File
@@ -4,9 +4,7 @@ import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"log/slog"
"os"
"path/filepath"
@@ -17,15 +15,24 @@ import (
"reasonix/internal/config"
"reasonix/internal/fileutil"
"reasonix/internal/session"
"reasonix/internal/store"
)
type desktopMigrationRecord struct {
SourceKey string `json:"sourceKey"`
TargetSessionID string `json:"targetSessionId"`
ContentDigest string `json:"contentDigest,omitempty"`
Status string `json:"status"`
ErrorCode string `json:"errorCode,omitempty"`
Attempts int `json:"attempts"`
SourceKey string `json:"sourceKey"`
TargetSessionID string `json:"targetSessionId"`
ContentDigest string `json:"contentDigest,omitempty"`
SourceRevision string `json:"sourceRevision,omitempty"`
Status string `json:"status"`
ErrorCode string `json:"errorCode,omitempty"`
Attempts int `json:"attempts"`
PreviousCompletion *desktopMigrationReceipt `json:"previousCompletion,omitempty"`
LegacyHeads []string `json:"legacyHeads,omitempty"`
LegacySelectedHead string `json:"legacySelectedHead,omitempty"`
LegacyPrimaryHead string `json:"legacyPrimaryHead,omitempty"`
LegacyHeadsRevision string `json:"legacyHeadsRevision,omitempty"`
LegacyAdoption *desktopMigrationReceipt `json:"legacyAdoption,omitempty"`
LegacyConversions []desktopMigrationConversion `json:"legacyConversions,omitempty"`
}
type desktopMigrationLedger struct {
@@ -92,10 +99,19 @@ func slogWarnDesktopMigration(err error) {
}
type desktopMigrationSource struct {
root string
scope string
workspaceRoot string
exact map[string]bool
root string
scope string
workspaceRoot string
exact map[string]bool
records map[string]desktopMigrationRecord
pairedRoot string
pairedIDs map[string]bool
legacyAdoption *desktopMigrationReceipt
pairedAdoption *desktopMigrationReceipt
conversions map[string][]desktopMigrationConversion
headConversions []desktopMigrationConversion
handledStores map[string]bool
conversionAdoptions []*desktopMigrationReceipt
}
func (a *App) migrateDesktopSessionsV5(ctx context.Context) error {
@@ -115,10 +131,18 @@ func (a *App) migrateDesktopSessionsV5(ctx context.Context) error {
sources[key] = source
return source
}
add("global", "", config.SessionStoreDir())
add("global", "", config.ProjectSessionStoreDir(globalWorkspaceRoot()))
addStores := func(scope, workspaceRoot, root string) {
if strings.TrimSpace(root) == "" {
return
}
for _, candidate := range desktopLegacyStoreRoots(root) {
add(scope, workspaceRoot, candidate)
}
}
addStores("global", "", config.SessionStoreDir())
addStores("global", "", config.ProjectSessionStoreDir(globalWorkspaceRoot()))
for _, project := range projects.Projects {
add("project", project.Root, config.ProjectSessionStoreDir(project.Root))
addStores("project", project.Root, config.ProjectSessionStoreDir(project.Root))
}
for _, tab := range tabs.Tabs {
if strings.TrimSpace(tab.SessionID) == "" {
@@ -131,13 +155,9 @@ func (a *App) migrateDesktopSessionsV5(ctx context.Context) error {
if source := add(tab.Scope, tab.WorkspaceRoot, root); source != nil {
source.exact[tab.SessionID] = true
}
addStores(tab.Scope, tab.WorkspaceRoot, root)
}
var joined error
for _, source := range sources {
if err := a.migrateCanonicalStore(ctx, *source); err != nil {
joined = errors.Join(joined, err)
}
}
legacySources := map[string]desktopMigrationSource{}
addLegacy := func(scope, workspaceRoot, dir string) {
dir = filepath.Clean(strings.TrimSpace(dir))
@@ -146,7 +166,7 @@ func (a *App) migrateDesktopSessionsV5(ctx context.Context) error {
}
key := canonicalRuntimeRoot(dir)
if _, ok := legacySources[key]; !ok {
legacySources[key] = desktopMigrationSource{root: dir, scope: scope, workspaceRoot: workspaceRoot, exact: map[string]bool{}}
legacySources[key] = desktopMigrationSource{root: dir, scope: scope, workspaceRoot: workspaceRoot, exact: map[string]bool{}, pairedRoot: filepath.Join(filepath.Dir(dir), "sessions-v4")}
}
}
addLegacy("global", "", config.SessionDir())
@@ -163,16 +183,63 @@ func (a *App) migrateDesktopSessionsV5(ctx context.Context) error {
key := canonicalRuntimeRoot(dir)
source, ok := legacySources[key]
if !ok {
source = desktopMigrationSource{root: dir, scope: tab.Scope, workspaceRoot: tab.WorkspaceRoot, exact: map[string]bool{}}
source = desktopMigrationSource{root: dir, scope: tab.Scope, workspaceRoot: tab.WorkspaceRoot, exact: map[string]bool{}, pairedRoot: filepath.Join(filepath.Dir(dir), "sessions-v4")}
}
source.exact[path] = true
legacySources[key] = source
}
// Include retired roots discovered only through saved legacy tabs before
// indexing conversion provenance.
for _, source := range legacySources {
addStores(source.scope, source.workspaceRoot, source.pairedRoot)
}
conversions, handledStores, conversionErr := discoverDesktopMigrationConversions(ctx, sources)
joined = errors.Join(joined, conversionErr)
for _, source := range legacySources {
source.conversions, source.handledStores = conversions, handledStores
joined = errors.Join(joined, markDesktopMigrationPairedStores(source, sources))
if err := a.migrateLegacyDirectory(ctx, source); err != nil {
joined = errors.Join(joined, err)
}
}
joined = errors.Join(joined, a.migrateStoredConversionLineages(ctx, conversions, sources, handledStores))
for _, source := range sources {
source.handledStores = handledStores
if err := a.migrateCanonicalStore(ctx, *source); err != nil {
joined = errors.Join(joined, err)
}
}
return joined
}
// A paired checkpoint and event store are one migration decision. Never
// publish the sidecar independently after a conflict or source failure.
func markDesktopMigrationPairedStores(source desktopMigrationSource, sources map[string]*desktopMigrationSource) error {
entries, err := os.ReadDir(source.root)
if os.IsNotExist(err) {
return nil
}
if err != nil {
return err
}
var joined error
for _, entry := range entries {
if entry.IsDir() || !store.IsSessionTranscriptName(entry.Name()) || strings.HasPrefix(entry.Name(), ".") {
continue
}
path := filepath.Join(source.root, entry.Name())
pairedRoot, err := desktopLegacyPairedRoot(path, source.pairedRoot)
if err != nil {
joined = errors.Join(joined, err)
continue
}
if paired := sources[canonicalRuntimeRoot(pairedRoot)]; paired != nil {
if paired.pairedIDs == nil {
paired.pairedIDs = map[string]bool{}
}
paired.pairedIDs[agent.BranchID(path)] = true
}
}
return joined
}
@@ -182,42 +249,74 @@ func (a *App) migrateCanonicalStore(ctx context.Context, source desktopMigration
} else if err != nil {
return err
}
old, err := session.NewService("migration-source", session.NewFilesystemPersistence(source.root))
persistence := session.NewFilesystemPersistence(source.root)
old, err := session.NewService("migration-source", persistence)
if err != nil {
return err
}
defer func() { retErr = errors.Join(retErr, old.Shutdown(context.Background())) }()
infos, err := listAllCanonicalSessionInfo(ctx, old.Query())
if err != nil {
return err
}
workspaceID, err := a.ensureDesktopWorkspace(ctx, source.scope, source.workspaceRoot)
ledger, err := readDesktopMigrationLedger()
if err != nil {
return err
}
source.records = ledger.Records
var joined error
for _, info := range infos {
if ctx.Err() != nil {
return ctx.Err()
var cursor string
for {
// A disposable catalog cache cannot decide whether durable history
// exists. Enumerate every identity, including entries with failed or
// missing metadata, without scheduling writes to the source cache.
page, err := persistence.List(ctx, cursor, 100)
if err != nil {
return errors.Join(joined, err)
}
exact := source.exact[info.SessionID]
if !exact && info.Turns == 0 && strings.TrimSpace(info.Title) == "" && strings.TrimSpace(info.Preview) == "" {
continue
for _, info := range page.Sessions {
if ctx.Err() != nil {
return errors.Join(joined, ctx.Err())
}
if source.pairedIDs[info.SessionID] || source.handledStores[canonicalRuntimeRoot(filepath.Join(source.root, info.SessionID))] {
continue
}
if info.Codec == session.PrototypeCodec || info.Codec == session.LegacyLinearCodec || info.Codec == session.FinalV31Codec {
joined = errors.Join(joined, a.migratePreviewSession(ctx, source, info.SessionID))
continue
}
if err := a.migrateCanonicalSession(ctx, old, source, "", info.SessionID); err != nil {
joined = errors.Join(joined, err)
}
}
if err := a.migrateCanonicalSession(ctx, old, source, workspaceID, info.SessionID); err != nil {
joined = errors.Join(joined, err)
if page.NextCursor == "" {
return joined
}
if page.NextCursor <= cursor {
return errors.Join(joined, errors.New("desktop migration source cursor did not advance"))
}
cursor = page.NextCursor
}
return joined
}
func (a *App) migrateCanonicalSession(ctx context.Context, old *session.Service, source desktopMigrationSource, workspaceID, sessionID string) error {
digest := sha256.Sum256([]byte(canonicalRuntimeRoot(source.root) + "\x00" + sessionID))
key := hex.EncodeToString(digest[:])
key := desktopCanonicalMigrationKey(source.root, sessionID)
checkpoint, err := newDesktopMigrationCheckpoint(source, key, canonicalMigrationSourceFiles(source.root, sessionID))
if err != nil {
return err
}
if checkpoint.unchanged() {
return checkpoint.skip()
}
oldRef := session.SessionRef{HostID: "migration-source", SessionID: sessionID}
contentDigest, err := canonicalMigrationDigest(ctx, old.Query(), oldRef)
if err != nil {
return err
return errors.Join(err, updateDesktopMigrationLedger(key, sessionID, "failed", "source_read"))
}
if checkpoint.matchesCompletedContent(contentDigest) {
return checkpoint.complete(checkpoint.record.TargetSessionID, contentDigest)
}
if workspaceID == "" {
workspaceID, err = a.ensureDesktopMigrationWorkspace(ctx, source)
if err != nil {
return err
}
}
target := a.desktopSessionService("")
targetID, needsImport, err := resolveMigrationTarget(ctx, target.Query(), sessionID, key, contentDigest)
@@ -250,7 +349,7 @@ func (a *App) migrateCanonicalSession(ctx context.Context, old *session.Service,
_ = updateDesktopMigrationLedger(key, targetID, "failed", "registry", contentDigest)
return err
}
return updateDesktopMigrationLedger(key, targetID, "completed", "", contentDigest)
return checkpoint.complete(targetID, contentDigest)
}
func (a *App) migrateLegacyDirectory(ctx context.Context, source desktopMigrationSource) error {
@@ -261,28 +360,30 @@ func (a *App) migrateLegacyDirectory(ctx context.Context, source desktopMigratio
if err != nil {
return err
}
workspaceID, err := a.ensureDesktopWorkspace(ctx, source.scope, source.workspaceRoot)
ledger, err := readDesktopMigrationLedger()
if err != nil {
return err
}
source.records = ledger.Records
var joined error
for _, entry := range entries {
if ctx.Err() != nil {
return ctx.Err()
}
if entry.IsDir() || filepath.Ext(entry.Name()) != ".jsonl" || strings.HasPrefix(entry.Name(), ".") {
if entry.IsDir() || !store.IsSessionTranscriptName(entry.Name()) || strings.HasPrefix(entry.Name(), ".") {
continue
}
path := filepath.Join(source.root, entry.Name())
exact := source.exact[path]
meta, hasMeta, metaErr := agent.LoadBranchMeta(path)
if metaErr != nil && !exact {
if desktopMigrationAutomaticRecovery(path) {
continue
}
if !exact && (!hasMeta || (meta.Turns == 0 && strings.TrimSpace(meta.Name) == "" && strings.TrimSpace(meta.TopicTitle) == "")) {
perFile := source
perFile.pairedRoot, err = desktopLegacyPairedRoot(path, source.pairedRoot)
if err != nil {
joined = errors.Join(joined, err, updateDesktopMigrationLedger(desktopLegacyMigrationKey(path), "", "failed", "source_stat"))
continue
}
if err := a.migrateLegacySession(ctx, path, source, workspaceID); err != nil {
if err := a.migrateLegacyHeads(ctx, path, perFile); err != nil {
joined = errors.Join(joined, err)
}
}
@@ -290,11 +391,34 @@ func (a *App) migrateLegacyDirectory(ctx context.Context, source desktopMigratio
}
func (a *App) migrateLegacySession(ctx context.Context, path string, source desktopMigrationSource, workspaceID string) (retErr error) {
digest := sha256.Sum256([]byte(canonicalRuntimeRoot(path)))
key := hex.EncodeToString(digest[:])
if err := updateDesktopMigrationLedger(key, "", "pending", ""); err != nil {
return a.migrateLegacyHead(ctx, path, source, workspaceID, "", desktopLegacyMigrationKey(path), true)
}
func (a *App) migrateLegacyHead(ctx context.Context, path string, source desktopMigrationSource, workspaceID, headID, key string, paired bool) (retErr error) {
checkpoint, err := newDesktopMigrationCheckpoint(source, key, desktopLegacyMigrationFiles(path, source))
if err != nil {
return err
}
if checkpoint.unchanged() {
return checkpoint.skip()
}
if len(source.headConversions) > 0 {
if err := a.migrateConversionLineage(ctx, path, headID, source, &checkpoint, workspaceID); err != nil {
return errors.Join(err, updateDesktopMigrationLedger(key, checkpoint.record.TargetSessionID, "failed", "conversion_import"))
}
return nil
}
if paired && source.pairedRoot != "" {
// Previous v5 builds may already have adopted the paired canonical
// store before discovering its metadata-less legacy checkpoint.
pairedKey := desktopCanonicalMigrationKey(source.pairedRoot, agent.BranchID(path))
record := source.records[pairedKey]
if record.Status == "completed" {
source.pairedAdoption = &desktopMigrationReceipt{TargetSessionID: record.TargetSessionID, ContentDigest: record.ContentDigest}
} else {
source.pairedAdoption = record.PreviousCompletion
}
}
stageRoot, err := os.MkdirTemp("", "reasonix-legacy-import-")
if err != nil {
return err
@@ -305,35 +429,72 @@ func (a *App) migrateLegacySession(ctx context.Context, path string, source desk
return err
}
defer func() { retErr = errors.Join(retErr, stage.Shutdown(context.Background())) }()
runtime, result, err := stage.ContinueImported(ctx, path, "")
var runtime *session.Runtime
if paired && source.pairedRoot != "" {
runtime, _, err = stage.ContinueImportedFrom(ctx, path, source.pairedRoot, headID)
} else {
runtime, _, err = stage.ContinueImported(ctx, path, headID)
}
if err != nil {
_ = updateDesktopMigrationLedger(key, "", "failed", "legacy_import")
return err
}
bundle := filepath.Join(stageRoot, "bundle")
if err := stage.Export(ctx, runtime.Ref(), bundle); err != nil {
_ = updateDesktopMigrationLedger(key, result.TargetID, "failed", "export")
return err
}
if err := stage.Close(ctx, runtime.Ref()); err != nil {
return err
}
return a.publishStagedMigration(ctx, source, checkpoint, stage, runtime.Ref(), workspaceID, session.SessionOriginLegacyImport)
}
func (a *App) publishStagedMigration(ctx context.Context, source desktopMigrationSource, checkpoint desktopMigrationCheckpoint, stage *session.Service, ref session.SessionRef, workspaceID string, origin session.SessionOrigin) error {
key := checkpoint.key
target := a.desktopSessionService("")
contentDigest, err := canonicalMigrationDigest(ctx, stage.Query(), runtime.Ref())
contentDigest, err := canonicalMigrationDigest(ctx, stage.Query(), ref)
if err != nil {
return err
}
targetID, needsImport, err := resolveMigrationTarget(ctx, target.Query(), result.TargetID, key, contentDigest)
if checkpoint.matchesCompletedContent(contentDigest) {
return checkpoint.complete(checkpoint.record.TargetSessionID, contentDigest)
}
// An older migrator adopted only the selected DAG head under the path key.
// Recognize that receipt when adding explicit head identities, even if the
// selected head has since changed and the v5 target has been continued.
if source.legacyAdoption != nil && source.legacyAdoption.ContentDigest == contentDigest {
return checkpoint.complete(source.legacyAdoption.TargetSessionID, contentDigest)
}
if source.pairedAdoption != nil && source.pairedAdoption.ContentDigest == contentDigest {
return checkpoint.complete(source.pairedAdoption.TargetSessionID, contentDigest)
}
for _, receipt := range source.conversionAdoptions {
if receipt.ContentDigest == contentDigest {
return checkpoint.complete(receipt.TargetSessionID, contentDigest)
}
}
if workspaceID == "" {
workspaceID, err = a.ensureDesktopMigrationWorkspace(ctx, source)
if err != nil {
return err
}
}
targetID, needsImport, err := resolveMigrationTarget(ctx, target.Query(), ref.SessionID, key, contentDigest)
if err != nil {
_ = updateDesktopMigrationLedger(key, result.TargetID, "failed", "target_conflict", contentDigest)
_ = updateDesktopMigrationLedger(key, ref.SessionID, "failed", "target_conflict", contentDigest)
return err
}
if err := updateDesktopMigrationLedger(key, targetID, "pending", "", contentDigest); err != nil {
return err
}
if needsImport {
tmp, err := os.MkdirTemp("", "reasonix-v5-bundle-")
if err != nil {
return err
}
defer os.RemoveAll(tmp)
bundle := filepath.Join(tmp, "bundle")
if err := stage.Export(ctx, ref, bundle); err != nil {
return errors.Join(err, updateDesktopMigrationLedger(key, targetID, "failed", "export", contentDigest))
}
if _, err := target.ImportWithHeader(ctx, bundle, session.CreateOptions{
SessionID: targetID, CWD: desktopWorkspaceRoot(source.scope, source.workspaceRoot), Origin: session.SessionOriginLegacyImport,
SessionID: targetID, CWD: desktopWorkspaceRoot(source.scope, source.workspaceRoot), Origin: origin,
}); err != nil {
_ = updateDesktopMigrationLedger(key, targetID, "failed", "import", contentDigest)
return err
@@ -343,7 +504,7 @@ func (a *App) migrateLegacySession(ctx context.Context, path string, source desk
_ = updateDesktopMigrationLedger(key, targetID, "failed", "registry", contentDigest)
return err
}
return updateDesktopMigrationLedger(key, targetID, "completed", "", contentDigest)
return checkpoint.complete(targetID, contentDigest)
}
func canonicalMigrationDigest(ctx context.Context, query *session.Query, ref session.SessionRef) (string, error) {
@@ -388,34 +549,39 @@ func resolveMigrationTarget(ctx context.Context, query *session.Query, preferred
return "", false, errors.New("migration target identity collision")
}
func updateDesktopMigrationLedger(sourceKey, targetID, status, errorCode string, digest ...string) error {
// sourceState optionally supplies the content digest followed by a file revision.
func updateDesktopMigrationLedger(sourceKey, targetID, status, errorCode string, sourceState ...string) error {
desktopMigrationMu.Lock()
defer desktopMigrationMu.Unlock()
path := desktopMigrationLedgerPath()
ledger := desktopMigrationLedger{Version: 1, Records: map[string]desktopMigrationRecord{}}
if body, err := os.ReadFile(path); err == nil {
if unmarshalErr := json.Unmarshal(body, &ledger); unmarshalErr != nil {
return unmarshalErr
}
} else if !os.IsNotExist(err) {
ledger, original, err := readDesktopMigrationLedgerFile()
if err != nil {
return err
}
if ledger.Version > 1 {
return fmt.Errorf("desktop migration ledger version %d is unsupported", ledger.Version)
}
if ledger.Records == nil {
ledger.Records = map[string]desktopMigrationRecord{}
}
record := ledger.Records[sourceKey]
// A failed scan or interrupted update must not erase proof that an older
// source revision was already adopted (and may have been continued).
if record.Status == "completed" && status != "completed" {
record.PreviousCompletion = &desktopMigrationReceipt{
TargetSessionID: record.TargetSessionID, ContentDigest: record.ContentDigest, SourceRevision: record.SourceRevision,
}
}
record.SourceKey, record.TargetSessionID, record.Status, record.ErrorCode = sourceKey, targetID, status, errorCode
if len(digest) > 0 {
record.ContentDigest = digest[0]
if len(sourceState) > 0 {
record.ContentDigest = sourceState[0]
}
record.SourceRevision = ""
if status == "completed" && len(sourceState) > 1 {
record.SourceRevision = sourceState[1]
}
if status == "completed" {
record.PreviousCompletion = nil
}
if status == "pending" {
record.Attempts++
}
ledger.Records[sourceKey] = record
body, err := json.MarshalIndent(ledger, "", " ")
body, err := marshalDesktopMigrationRecord(original, ledger, sourceKey)
if err != nil {
return err
}
+220
View File
@@ -0,0 +1,220 @@
package main
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"os"
"path/filepath"
"sort"
"reasonix/internal/store"
)
// Migration may add members to an existing workspace, but it does not own the
// user's title or visibility. Avoid rewriting presentation for each new source.
func (a *App) ensureDesktopMigrationWorkspace(ctx context.Context, source desktopMigrationSource) (string, error) {
id := desktopWorkspaceID(source.scope, source.workspaceRoot)
state, err := a.workspaceRegistry().Load(ctx)
if err != nil {
return "", err
}
if _, exists := state.Workspaces[id]; exists {
return id, nil
}
return a.ensureDesktopWorkspace(ctx, source.scope, source.workspaceRoot)
}
// A completed ledger entry records adoption of a source, not equality with
// today's target. The target can be continued, archived or deliberately deleted.
// None of those actions authorize importing the old conversation again.
type desktopMigrationCheckpoint struct {
key string
files []string
revision string
record desktopMigrationRecord
refresh bool
}
type desktopMigrationReceipt struct {
TargetSessionID string `json:"targetSessionId"`
ContentDigest string `json:"contentDigest,omitempty"`
SourceRevision string `json:"sourceRevision,omitempty"`
}
func newDesktopMigrationCheckpoint(source desktopMigrationSource, key string, files []string) (desktopMigrationCheckpoint, error) {
records := source.records
if records == nil {
ledger, err := readDesktopMigrationLedger()
if err != nil {
return desktopMigrationCheckpoint{}, err
}
records = ledger.Records
}
revision, err := desktopMigrationSourceRevision(files)
if err != nil {
return desktopMigrationCheckpoint{}, errors.Join(err, updateDesktopMigrationLedger(key, records[key].TargetSessionID, "failed", "source_stat"))
}
record := records[key]
refresh := record.Status != "completed" && record.PreviousCompletion != nil
if refresh {
previous := record.PreviousCompletion
record.Status, record.TargetSessionID = "completed", previous.TargetSessionID
record.ContentDigest, record.SourceRevision = previous.ContentDigest, previous.SourceRevision
}
return desktopMigrationCheckpoint{key: key, files: files, revision: revision, record: record, refresh: refresh}, nil
}
func (c desktopMigrationCheckpoint) completed() bool {
return c.record.Status == "completed" && c.record.TargetSessionID != ""
}
func (c desktopMigrationCheckpoint) unchanged() bool {
return c.completed() && c.record.SourceRevision == c.revision
}
func (c desktopMigrationCheckpoint) skip() error {
if c.refresh {
return c.complete(c.record.TargetSessionID, c.record.ContentDigest)
}
return nil
}
// Old ledgers have no source revision. Compare their recorded source digest
// once, without comparing to a target that may already contain newer work.
func (c desktopMigrationCheckpoint) matchesCompletedContent(digest string) bool {
return c.completed() && digest != "" && c.record.ContentDigest == digest
}
func (c desktopMigrationCheckpoint) complete(targetID, digest string) error {
revision, err := desktopMigrationSourceRevision(c.files)
if err != nil || revision != c.revision {
return errors.Join(errors.New("desktop migration source changed during import"), err,
updateDesktopMigrationLedger(c.key, targetID, "failed", "source_changed", digest))
}
return updateDesktopMigrationLedger(c.key, targetID, "completed", "", digest, c.revision)
}
func canonicalMigrationSourceFiles(root, sessionID string) []string {
dir := filepath.Join(root, sessionID)
// Content blobs are immutable and referenced by the event log. Disposable
// indexes, checkpoints and lock files do not change the migration input.
return []string{filepath.Join(dir, "manifest.json"), filepath.Join(dir, "events.frames"),
filepath.Join(dir, "events.jsonl"), filepath.Join(dir, "header.json")}
}
func desktopCanonicalMigrationKey(root, sessionID string) string {
digest := sha256.Sum256([]byte(canonicalRuntimeRoot(root) + "\x00" + sessionID))
return hex.EncodeToString(digest[:])
}
func legacyMigrationSourceFiles(path string) []string {
// Use the same artifact set as the legacy importer's freeze operation,
// including DAG/event sidecars whose changes need not rewrite the JSONL.
return append([]string{path}, store.SessionSidecarFiles(path)...)
}
func desktopLegacyMigrationKey(path string) string {
digest := sha256.Sum256([]byte(canonicalRuntimeRoot(path)))
return hex.EncodeToString(digest[:])
}
// This is a cheap filesystem revision, not a content integrity checksum.
// Normal source writes change size or mtime; no history bodies are read here.
// Include absent files so creation/removal of a sidecar invalidates the stamp.
func desktopMigrationSourceRevision(files []string) (string, error) {
paths := append([]string(nil), files...)
sort.Strings(paths)
hash := sha256.New()
for _, path := range paths {
info, err := os.Stat(path)
if os.IsNotExist(err) {
fmt.Fprintf(hash, "%q:missing\n", path)
continue
}
if err != nil {
return "", err
}
fmt.Fprintf(hash, "%q:%d:%d:%d\n", path, info.Mode(), info.Size(), info.ModTime().UnixNano())
}
return "stat-v1-" + hex.EncodeToString(hash.Sum(nil)), nil
}
func readDesktopMigrationLedger() (desktopMigrationLedger, error) {
desktopMigrationMu.Lock()
defer desktopMigrationMu.Unlock()
ledger, _, err := readDesktopMigrationLedgerFile()
return ledger, err
}
// Callers serialize reads and atomic replacement with desktopMigrationMu.
func readDesktopMigrationLedgerFile() (desktopMigrationLedger, []byte, error) {
ledger := desktopMigrationLedger{Version: 1, Records: map[string]desktopMigrationRecord{}}
body, err := os.ReadFile(desktopMigrationLedgerPath())
if os.IsNotExist(err) {
return ledger, nil, nil
}
if err != nil {
return ledger, nil, err
}
if err := json.Unmarshal(body, &ledger); err != nil {
return ledger, nil, err
}
if ledger.Version > 1 {
return ledger, nil, fmt.Errorf("desktop migration ledger version %d is unsupported", ledger.Version)
}
if ledger.Records == nil {
ledger.Records = map[string]desktopMigrationRecord{}
}
return ledger, body, nil
}
// Preserve unknown root/record fields when adding an optional revision to a v1
// ledger. Older writers can drop the revision; the digest fallback remains safe.
func marshalDesktopMigrationRecord(original []byte, ledger desktopMigrationLedger, key string) ([]byte, error) {
root := map[string]json.RawMessage{}
if len(original) > 0 {
if err := json.Unmarshal(original, &root); err != nil {
return nil, err
}
}
if root == nil {
root = map[string]json.RawMessage{}
}
records := map[string]map[string]json.RawMessage{}
if body := root["records"]; len(body) > 0 {
if err := json.Unmarshal(body, &records); err != nil {
return nil, err
}
}
if records == nil {
records = map[string]map[string]json.RawMessage{}
}
fields := records[key]
if fields == nil {
fields = map[string]json.RawMessage{}
}
for _, name := range []string{"sourceKey", "targetSessionId", "contentDigest", "sourceRevision", "previousCompletion", "status", "errorCode", "attempts", "legacyHeads", "legacySelectedHead", "legacyPrimaryHead", "legacyHeadsRevision", "legacyAdoption", "legacyConversions"} {
delete(fields, name)
}
body, err := json.Marshal(ledger.Records[key])
if err != nil {
return nil, err
}
if err := json.Unmarshal(body, &fields); err != nil {
return nil, err
}
records[key] = fields
root["records"], err = json.Marshal(records)
if err != nil {
return nil, err
}
root["version"], err = json.Marshal(ledger.Version)
if err != nil {
return nil, err
}
return json.MarshalIndent(root, "", " ")
}
@@ -0,0 +1,309 @@
package main
import (
"bytes"
"encoding/json"
"os"
"path/filepath"
"testing"
"time"
"reasonix/desktop/internal/workspacestate"
"reasonix/internal/agent"
"reasonix/internal/config"
"reasonix/internal/provider"
"reasonix/internal/session"
"reasonix/internal/store"
)
func appendMigrationTestMessage(t *testing.T, service *session.Service, ref session.SessionRef, id string) {
t.Helper()
binding, err := service.Open(t.Context(), ref)
if err != nil {
t.Fatal(err)
}
payload, _ := json.Marshal(map[string]any{"message": provider.Message{ID: id, Role: provider.RoleUser, Content: id}})
if _, err := binding.Runtime().Session().AppendBatch(t.Context(), id, []session.Event{{Kind: "message/complete", Payload: payload}}); err != nil {
t.Fatal(err)
}
if err := binding.Release(t.Context()); err != nil {
t.Fatal(err)
}
if err := service.Close(t.Context(), ref); err != nil {
t.Fatal(err)
}
}
func onlyMigrationRecord(t *testing.T) desktopMigrationRecord {
t.Helper()
ledger, err := readDesktopMigrationLedger()
if err != nil || len(ledger.Records) != 1 {
t.Fatalf("ledger = %#v, %v", ledger, err)
}
for _, record := range ledger.Records {
return record
}
panic("unreachable")
}
func TestIncrementalMigrationPreservesContinuedTargetsAndImportsSourceChanges(t *testing.T) {
for _, kind := range []string{"v4", "legacy"} {
for _, oldLedger := range []bool{false, true} {
name := kind + "/current-ledger"
if oldLedger {
name = kind + "/old-ledger"
}
t.Run(name, func(t *testing.T) {
isolateDesktopUserDirs(t)
var migrate func(*App) error
var growSource func()
var sourceFile string
if kind == "v4" {
root := config.SessionStoreDir()
old := coldV4MigrationFixture(t, root, "native")
source := desktopMigrationSource{root: root, scope: "global"}
migrate = func(app *App) error { return app.migrateCanonicalStore(t.Context(), source) }
growSource = func() {
appendMigrationTestMessage(t, old, session.SessionRef{HostID: "migration-source", SessionID: "native"}, "new-source-work")
}
sourceFile = filepath.Join(root, "native", "events.frames")
} else {
root := config.SessionDir()
sourceFile = filepath.Join(root, "old.jsonl")
legacy := agent.NewSession("system")
legacy.Add(provider.Message{ID: "old", Role: provider.RoleUser, Content: "original work"})
if err := legacy.Save(sourceFile); err != nil {
t.Fatal(err)
}
source := desktopMigrationSource{root: root, scope: "global", exact: map[string]bool{sourceFile: true}}
migrate = func(app *App) error {
err := app.migrateLegacyDirectory(t.Context(), source)
// Subsequent startups no longer have an open legacy tab.
source.exact = nil
return err
}
growSource = func() {
legacy.Add(provider.Message{ID: "new-source-work", Role: provider.RoleUser, Content: "new-source-work"})
if err := legacy.Save(sourceFile); err != nil {
t.Fatal(err)
}
}
}
app := NewApp()
t.Cleanup(app.closeSessionServices)
if err := migrate(app); err != nil {
t.Fatal(err)
}
first := onlyMigrationRecord(t)
if first.SourceRevision == "" || first.Status != "completed" {
t.Fatalf("missing completed source revision: %#v", first)
}
if kind == "legacy" {
if err := os.Remove(store.SessionMeta(sourceFile)); err != nil && !os.IsNotExist(err) {
t.Fatal(err)
}
}
if oldLedger {
if err := updateDesktopMigrationLedger(first.SourceKey, first.TargetSessionID, "completed", "", first.ContentDigest); err != nil {
t.Fatal(err)
}
}
ref := session.SessionRef{HostID: localDesktopHostID, SessionID: first.TargetSessionID}
appendMigrationTestMessage(t, app.desktopSessionService(""), ref, "new-v5-work")
if err := app.workspaceRegistry().RenameWorkspace(t.Context(), workspacestate.GlobalWorkspaceID, "user workspace name"); err != nil {
t.Fatal(err)
}
if err := app.workspaceRegistry().SetWorkspaceVisible(t.Context(), workspacestate.GlobalWorkspaceID, false); err != nil {
t.Fatal(err)
}
workspaceBefore, err := os.ReadFile(config.DesktopWorkspaceStatePath())
if err != nil {
t.Fatal(err)
}
app.closeSessionServices()
app = NewApp()
t.Cleanup(app.closeSessionServices)
for range 2 {
if err := migrate(app); err != nil {
t.Fatal(err)
}
current := onlyMigrationRecord(t)
if current.TargetSessionID != first.TargetSessionID || current.Attempts != first.Attempts || current.SourceRevision == "" {
t.Fatalf("continued target was remigrated: first=%#v current=%#v", first, current)
}
}
// A metadata-only source change must refresh its checkpoint using
// the recorded source digest, not compare against the continued target.
stamp := time.Now().Add(-time.Hour)
if err := os.Chtimes(sourceFile, stamp, stamp); err != nil {
t.Fatal(err)
}
if err := migrate(app); err != nil {
t.Fatal(err)
}
if got := onlyMigrationRecord(t); got.Attempts != first.Attempts || got.TargetSessionID != first.TargetSessionID {
t.Fatalf("metadata change remigrated content: %#v", got)
}
// A failed attempt must retain the previous adoption receipt.
// Restoring the source after a temporary error cannot resurrect
// an old copy beside the target that the user has continued.
if err := updateDesktopMigrationLedger(first.SourceKey, "incomplete-target", "failed", "source_changed", "incomplete-digest"); err != nil {
t.Fatal(err)
}
if err := migrate(app); err != nil {
t.Fatal(err)
}
if got := onlyMigrationRecord(t); got.Status != "completed" || got.TargetSessionID != first.TargetSessionID || got.PreviousCompletion != nil {
t.Fatalf("previous adoption receipt was lost: %#v", got)
}
workspaceAfter, err := os.ReadFile(config.DesktopWorkspaceStatePath())
if err != nil || !bytes.Equal(workspaceBefore, workspaceAfter) {
t.Fatalf("completed migration rewrote workspace state: %v", err)
}
growSource()
if err := migrate(app); err != nil {
t.Fatal(err)
}
changed := onlyMigrationRecord(t)
if changed.TargetSessionID == first.TargetSessionID || changed.Attempts != first.Attempts+1 || changed.ContentDigest == first.ContentDigest {
t.Fatalf("source update not imported separately: %#v", changed)
}
state, err := app.workspaceRegistry().Load(t.Context())
if err != nil || state.Workspaces[workspacestate.GlobalWorkspaceID].Title != "user workspace name" || state.Workspaces[workspacestate.GlobalWorkspaceID].Visible {
t.Fatalf("new source reset workspace presentation: %#v, %v", state, err)
}
for range 2 {
if err := migrate(app); err != nil {
t.Fatal(err)
}
}
infos, err := listAllCanonicalSessionInfo(t.Context(), app.desktopSessionService("").Query())
if err != nil || len(infos) != 2 {
t.Fatalf("expected original continued target and one source update: %#v, %v", infos, err)
}
history, err := app.desktopSessionService("").Query().History(t.Context(), ref)
if err != nil || len(history) == 0 || history[len(history)-1].Content != "new-v5-work" {
t.Fatalf("continued work changed: %#v, %v", history, err)
}
newRef := session.SessionRef{HostID: localDesktopHostID, SessionID: changed.TargetSessionID}
history, err = app.desktopSessionService("").Query().History(t.Context(), newRef)
if err != nil || len(history) == 0 || history[len(history)-1].Content != "new-source-work" {
t.Fatalf("source update lost: %#v, %v", history, err)
}
before, err := os.ReadFile(desktopMigrationLedgerPath())
if err != nil {
t.Fatal(err)
}
if err := app.desktopSessionService("").Delete(t.Context(), newRef); err != nil {
t.Fatal(err)
}
// No temporary import is permitted for a completed revision.
t.Setenv("TMPDIR", filepath.Join(t.TempDir(), "missing"))
if err := migrate(app); err != nil {
t.Fatal(err)
}
after, err := os.ReadFile(desktopMigrationLedgerPath())
if err != nil || !bytes.Equal(before, after) {
t.Fatalf("unchanged migration rewrote ledger: %v", err)
}
if _, err := os.Stat(filepath.Join(app.desktopSessions.root, changed.TargetSessionID)); !os.IsNotExist(err) {
t.Fatalf("deleted target was resurrected: %v", err)
}
})
}
}
}
func TestMigrationCheckpointRejectsSourceChangeDuringImport(t *testing.T) {
isolateDesktopUserDirs(t)
path := filepath.Join(t.TempDir(), "source.jsonl")
if err := os.WriteFile(path, []byte("before"), 0o600); err != nil {
t.Fatal(err)
}
checkpoint, err := newDesktopMigrationCheckpoint(desktopMigrationSource{}, "source", legacyMigrationSourceFiles(path))
if err != nil {
t.Fatal(err)
}
// An event sidecar can advance while the JSONL checkpoint stays untouched.
if err := os.WriteFile(store.SessionEventLog(path), []byte("new durable events"), 0o600); err != nil {
t.Fatal(err)
}
if err := checkpoint.complete("target", "digest"); err == nil {
t.Fatal("must not certify a revision that was not the frozen import input")
}
record := onlyMigrationRecord(t)
if record.Status != "failed" || record.ErrorCode != "source_changed" || record.SourceRevision != "" {
t.Fatalf("source race must remain retryable: %#v", record)
}
}
func TestMigrationLedgerOptionalRevisionPreservesUnknownFields(t *testing.T) {
isolateDesktopUserDirs(t)
path := desktopMigrationLedgerPath()
if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
t.Fatal(err)
}
original := []byte(`{"version":1,"futureRoot":{"keep":true},"records":{"source":{"sourceKey":"source","targetSessionId":"target","status":"completed","attempts":1,"contentDigest":"digest","futureRecord":"keep"}}}`)
if err := os.WriteFile(path, original, 0o600); err != nil {
t.Fatal(err)
}
if err := updateDesktopMigrationLedger("source", "target", "completed", "", "digest", "stat-v1-revision"); err != nil {
t.Fatal(err)
}
body, err := os.ReadFile(path)
if err != nil {
t.Fatal(err)
}
var preserved struct {
FutureRoot map[string]bool `json:"futureRoot"`
Records map[string]struct {
FutureRecord string `json:"futureRecord"`
SourceRevision string `json:"sourceRevision"`
} `json:"records"`
}
if err := json.Unmarshal(body, &preserved); err != nil || !preserved.FutureRoot["keep"] || preserved.Records["source"].FutureRecord != "keep" || preserved.Records["source"].SourceRevision != "stat-v1-revision" {
t.Fatalf("optional field update lost data: %s, %v", body, err)
}
// The previous reader ignores the optional field while retaining its
// existing diagnostics/status contract.
var previous struct {
Version int `json:"version"`
Records map[string]struct {
Status string `json:"status"`
ContentDigest string `json:"contentDigest"`
} `json:"records"`
}
if err := json.Unmarshal(body, &previous); err != nil || previous.Version != 1 || previous.Records["source"].Status != "completed" || previous.Records["source"].ContentDigest != "digest" {
t.Fatalf("old reader contract changed: %#v, %v", previous, err)
}
}
func TestMigrationRefusesUnreadableOrFutureLedger(t *testing.T) {
for _, body := range []string{`{"version":2,"records":{}}`, `{broken`} {
t.Run(body, func(t *testing.T) {
isolateDesktopUserDirs(t)
root := config.SessionStoreDir()
coldV4MigrationFixture(t, root, "native")
path := desktopMigrationLedgerPath()
if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(path, []byte(body), 0o600); err != nil {
t.Fatal(err)
}
app := NewApp()
t.Cleanup(app.closeSessionServices)
if err := app.migrateCanonicalStore(t.Context(), desktopMigrationSource{root: root, scope: "global"}); err == nil {
t.Fatal("unknown adoption state must not trigger another import")
}
after, err := os.ReadFile(path)
if err != nil || string(after) != body {
t.Fatalf("ledger was overwritten: %s, %v", after, err)
}
infos, err := listAllCanonicalSessionInfo(t.Context(), app.desktopSessionService("").Query())
if err != nil || len(infos) != 0 {
t.Fatalf("import ran despite unreadable ledger: %#v, %v", infos, err)
}
})
}
}
+219
View File
@@ -0,0 +1,219 @@
package main
import (
"context"
"crypto/sha256"
"encoding/hex"
"errors"
"os"
"path/filepath"
"slices"
"reasonix/internal/agent"
"reasonix/internal/fileutil"
"reasonix/internal/session"
)
func desktopLegacySourceFiles(path, pairedRoot string) []string {
files := legacyMigrationSourceFiles(path)
for _, root := range desktopLegacyStoreRoots(pairedRoot) {
files = append(files, canonicalMigrationSourceFiles(root, agent.BranchID(path))...)
}
return files
}
// Before the identity cutover, SessionDirV3 / ProjectSessionDirV3 used this
// sibling directory. Its presence is independent of the current v4 root.
func desktopLegacyStoreRoots(root string) []string {
if root == "" {
return nil
}
parent := filepath.Dir(root)
return []string{filepath.Join(parent, "sessions-v4"), filepath.Join(parent, "sessions-v3")}
}
func desktopLegacyPairedRoot(path, root string) (string, error) {
for _, candidate := range desktopLegacyStoreRoots(root) {
if _, err := os.Stat(filepath.Join(candidate, agent.BranchID(path))); err == nil {
return candidate, nil
} else if !os.IsNotExist(err) {
return "", err
}
}
return root, nil
}
func desktopLegacyHeadKey(path, head string) string {
digest := sha256.Sum256([]byte(canonicalRuntimeRoot(path) + "\x00head\x00" + head))
return hex.EncodeToString(digest[:])
}
// The cached head list is certified by the same source revision as imports.
// Completed DAGs therefore need only stat calls on later startups. The first
// head keeps the historical path key even when the old app changes selection.
func (a *App) migrateLegacyHeads(ctx context.Context, path string, source desktopMigrationSource) error {
key := desktopLegacyMigrationKey(path)
files := desktopLegacyMigrationFiles(path, source)
cp, err := newDesktopMigrationCheckpoint(source, key, files)
if err != nil {
return err
}
record := source.records[key]
if len(record.LegacyHeads) == 0 || record.LegacyHeadsRevision != cp.revision {
heads, err := session.LegacyMigrationHeads(ctx, path)
if err != nil {
return errors.Join(err, updateDesktopMigrationLedger(key, record.TargetSessionID, "failed", "source_heads"))
}
ids := []string{}
selected := ""
for _, head := range heads {
if head.Retired {
continue
}
ids = append(ids, head.ID)
if head.Selected {
selected = head.ID
}
}
if len(heads) == 0 {
ids = []string{""}
}
if len(ids) == 0 {
return errors.New("legacy DAG has no live heads")
}
primary := selected
if len(record.LegacyHeads) > 0 && slices.Contains(ids, record.LegacyHeads[0]) {
primary = record.LegacyHeads[0]
}
if index := slices.Index(ids, primary); index > 0 {
ids[0], ids[index] = ids[index], ids[0]
}
conversions, err := resolveDesktopConversionHeads(ctx, path, source.conversions[canonicalRuntimeRoot(path)])
if err != nil {
return errors.Join(err, updateDesktopMigrationLedger(key, record.TargetSessionID, "failed", "conversion_heads"))
}
if revision, err := desktopMigrationSourceRevision(files); err != nil || revision != cp.revision {
return errors.Join(errors.New("legacy source changed while listing heads"), err,
updateDesktopMigrationLedger(key, record.TargetSessionID, "failed", "source_changed"))
}
record, err = saveDesktopMigrationHeads(key, ids, selected, cp.revision, conversions)
if err != nil {
return err
}
}
source.legacyAdoption = record.LegacyAdoption
var joined error
for _, head := range record.LegacyHeads {
if err := ctx.Err(); err != nil {
return errors.Join(joined, err)
}
headKey := key
if head != record.LegacyPrimaryHead {
headKey = desktopLegacyHeadKey(path, head)
}
perHead := source
for _, converted := range record.LegacyConversions {
if converted.HeadID != head {
continue
}
perHead.headConversions = append(perHead.headConversions, converted)
if source.handledStores != nil {
source.handledStores[canonicalRuntimeRoot(filepath.Join(converted.Root, converted.SessionID))] = true
}
}
if len(perHead.headConversions) > 0 && head == record.LegacySelectedHead && source.pairedRoot != "" {
if err := perHead.includePairedConversion(path, head, headKey); err != nil {
joined = errors.Join(joined, err)
continue
}
}
joined = errors.Join(joined, a.migrateLegacyHead(ctx, path, perHead, "", head, headKey, head == record.LegacySelectedHead))
}
return joined
}
// The identity cutover can leave a current paired store without a Source field
// alongside provenance-linked conversions. Include it before suppressing its scan.
func (source *desktopMigrationSource) includePairedConversion(path, head, key string) error {
id := agent.BranchID(path)
dir := filepath.Join(source.pairedRoot, id)
manifest, err := readDesktopMigrationManifest(dir)
if os.IsNotExist(err) {
return nil
}
if err != nil {
return errors.Join(err, updateDesktopMigrationLedger(key, "", "failed", "paired_read"))
}
if manifest.SessionID != id {
return errors.Join(session.ErrDamagedStore, updateDesktopMigrationLedger(key, "", "failed", "paired_identity"))
}
for _, candidate := range source.headConversions {
if sameDesktopPath(filepath.Join(candidate.Root, candidate.SessionID), dir) {
return nil
}
}
source.headConversions = append([]desktopMigrationConversion{{Root: source.pairedRoot, SessionID: id, HeadID: head, Codec: manifest.Codec}}, source.headConversions...)
return nil
}
func saveDesktopMigrationHeads(key string, heads []string, selected, revision string, conversions []desktopMigrationConversion) (desktopMigrationRecord, error) {
desktopMigrationMu.Lock()
defer desktopMigrationMu.Unlock()
ledger, original, err := readDesktopMigrationLedgerFile()
if err != nil {
return desktopMigrationRecord{}, err
}
record := ledger.Records[key]
if record.LegacyPrimaryHead == "" && len(heads) > 0 {
record.LegacyPrimaryHead = heads[0]
}
if len(record.LegacyHeads) == 0 && record.LegacyAdoption == nil {
if record.Status == "completed" {
record.LegacyAdoption = &desktopMigrationReceipt{TargetSessionID: record.TargetSessionID, ContentDigest: record.ContentDigest, SourceRevision: record.SourceRevision}
} else if record.PreviousCompletion != nil {
receipt := *record.PreviousCompletion
record.LegacyAdoption = &receipt
}
}
record.SourceKey = key
record.LegacyHeads, record.LegacySelectedHead, record.LegacyHeadsRevision = heads, selected, revision
record.LegacyConversions = conversions
ledger.Records[key] = record
body, err := marshalDesktopMigrationRecord(original, ledger, key)
if err != nil {
return record, err
}
if err := os.MkdirAll(filepath.Dir(desktopMigrationLedgerPath()), 0o700); err != nil {
return record, err
}
return record, fileutil.AtomicWriteFileStrict(desktopMigrationLedgerPath(), append(body, '\n'), 0o600)
}
func (a *App) migratePreviewSession(ctx context.Context, source desktopMigrationSource, id string) (retErr error) {
key := desktopCanonicalMigrationKey(source.root, id)
cp, err := newDesktopMigrationCheckpoint(source, key, canonicalMigrationSourceFiles(source.root, id))
if err != nil {
return err
}
if cp.unchanged() {
return cp.skip()
}
tmp, err := os.MkdirTemp("", "reasonix-v3-import-")
if err != nil {
return err
}
defer os.RemoveAll(tmp)
stage, err := session.NewService("migration-stage", session.NewFilesystemPersistence(filepath.Join(tmp, "sessions-v4")))
if err != nil {
return err
}
defer func() { retErr = errors.Join(retErr, stage.Shutdown(context.Background())) }()
runtime, _, err := stage.ContinuePrototype(ctx, filepath.Join(source.root, id))
if err != nil {
return errors.Join(err, updateDesktopMigrationLedger(key, id, "failed", "preview_import"))
}
if err := stage.Close(ctx, runtime.Ref()); err != nil {
return err
}
return a.publishStagedMigration(ctx, source, cp, stage, runtime.Ref(), "", session.SessionOriginCanonicalImport)
}
+415
View File
@@ -0,0 +1,415 @@
package main
import (
"context"
"encoding/json"
"errors"
"os"
"path/filepath"
"sort"
"strings"
"reasonix/internal/agent"
"reasonix/internal/provider"
"reasonix/internal/session"
"reasonix/internal/store"
)
// A conversion is linked by persisted provenance, never by title or a coincident
// ID. LegacyDir identifies the immutable original snapshot of an unnamed head.
type desktopMigrationConversion struct {
Root string `json:"root"`
SessionID string `json:"sessionId"`
HeadID string `json:"headId,omitempty"`
LegacyDir string `json:"legacyDir,omitempty"`
Codec string `json:"codec"`
Depth int `json:"depth"`
}
type desktopMigrationStoredSource struct {
dir string
manifest session.Manifest
}
func discoverDesktopMigrationConversions(ctx context.Context, sources map[string]*desktopMigrationSource) (map[string][]desktopMigrationConversion, map[string]bool, error) {
nodes := map[string]desktopMigrationStoredSource{}
var joined error
for _, source := range sources {
entries, err := os.ReadDir(source.root)
if os.IsNotExist(err) {
continue
}
if err != nil {
joined = errors.Join(joined, err)
continue
}
for _, entry := range entries {
if err := ctx.Err(); err != nil {
return nil, nil, err
}
if !entry.IsDir() || strings.HasPrefix(entry.Name(), ".") {
continue
}
dir := filepath.Join(source.root, entry.Name())
manifest, err := readDesktopMigrationManifest(dir)
// Normal source migration reports damaged/unsupported stores. A bad
// manifest cannot authorize suppressing another source.
if err != nil || manifest.SessionID != entry.Name() {
continue
}
nodes[canonicalRuntimeRoot(dir)] = desktopMigrationStoredSource{dir: dir, manifest: manifest}
}
}
var origin func(desktopMigrationStoredSource, map[string]bool) (string, string, string, int, bool)
origin = func(node desktopMigrationStoredSource, seen map[string]bool) (string, string, string, int, bool) {
key := canonicalRuntimeRoot(node.dir)
if seen[key] || len(seen) >= 32 {
return "", "", "", 0, false
}
if node.manifest.Source == nil {
return key, "", "", 0, true
}
seen[key] = true
provenance := node.manifest.Source
if store.IsSessionTranscriptName(filepath.Base(provenance.Path)) && (provenance.Version == "" || provenance.Version == "legacy") {
return canonicalRuntimeRoot(provenance.Path), provenance.LegacyHeadID, filepath.Join(node.dir, "legacy"), 1, true
}
parent, found := nodes[canonicalRuntimeRoot(provenance.Path)]
if !found {
// Preview conversion preserves the old manifest inside its archive,
// so ancestry remains available after removal of an intermediate store.
dir := filepath.Join(node.dir, "legacy", "prototype")
manifest, err := readDesktopMigrationManifest(dir)
if err != nil {
return "", "", "", 0, false
}
parent = desktopMigrationStoredSource{dir: dir, manifest: manifest}
}
path, head, legacyDir, depth, ok := origin(parent, seen)
if ok && parent.manifest.Source == nil {
// Separate conversions archive the same native source in different
// directories. Keep its persisted identity after the original is
// removed; the archive location is not a new conversation origin.
path = canonicalRuntimeRoot(provenance.Path)
}
return path, head, legacyDir, depth + 1, ok
}
hidden := desktopMigrationRecoveryStores(nodes)
result := map[string][]desktopMigrationConversion{}
for key, node := range nodes {
if hidden[key] {
continue
}
path, head, legacyDir, depth, ok := origin(node, map[string]bool{})
if !ok {
continue
}
result[path] = append(result[path], desktopMigrationConversion{Root: filepath.Dir(node.dir), SessionID: filepath.Base(node.dir), HeadID: head, LegacyDir: legacyDir, Codec: node.manifest.Codec, Depth: depth})
}
for path := range result {
sort.Slice(result[path], func(i, j int) bool {
a, b := result[path][i], result[path][j]
if a.Depth != b.Depth {
return a.Depth > b.Depth
}
return filepath.Join(a.Root, a.SessionID) < filepath.Join(b.Root, b.SessionID)
})
}
return result, hidden, joined
}
func readDesktopMigrationManifest(dir string) (session.Manifest, error) {
body, err := os.ReadFile(filepath.Join(dir, "manifest.json"))
if err != nil {
return session.Manifest{}, err
}
var manifest session.Manifest
if err := json.Unmarshal(body, &manifest); err != nil {
return manifest, err
}
if manifest.Codec != session.Codec && manifest.Codec != session.PrototypeCodec && manifest.Codec != session.LegacyLinearCodec && manifest.Codec != session.FinalV31Codec {
return manifest, session.ErrUnsupportedVersion
}
return manifest, nil
}
func desktopLegacyMigrationFiles(path string, source desktopMigrationSource) []string {
files := desktopLegacySourceFiles(path, source.pairedRoot)
for _, converted := range source.conversions[canonicalRuntimeRoot(path)] {
files = append(files, canonicalMigrationSourceFiles(converted.Root, converted.SessionID)...)
files = append(files, legacyMigrationSourceFiles(filepath.Join(converted.LegacyDir, filepath.Base(path)))...)
}
return files
}
func resolveDesktopConversionHeads(ctx context.Context, path string, candidates []desktopMigrationConversion) ([]desktopMigrationConversion, error) {
resolved := append([]desktopMigrationConversion(nil), candidates...)
for index := range resolved {
converted := &resolved[index]
if converted.HeadID != "" || converted.LegacyDir == "" {
continue
}
// An omitted head means the selected head at conversion time, not the
// source's current selection. Read the preserved snapshot to recover it.
frozenPath := filepath.Join(converted.LegacyDir, filepath.Base(path))
heads, err := session.LegacyMigrationHeads(ctx, frozenPath)
if err != nil {
return nil, err
}
for _, head := range heads {
if head.Selected && !head.Retired {
converted.HeadID = head.ID
break
}
}
}
return resolved, nil
}
type desktopMigrationStagedNode struct {
service *session.Service
checkpoint desktopMigrationCheckpoint
ref session.SessionRef
messages []provider.Message
digest string
targetID string
coveredBy int
origin session.SessionOrigin
}
// All generations of one legacy head are staged before publication. Maximal
// histories survive; equal/prefix ancestors get receipts for the same target.
// Incomparable continuations remain separate conversations.
func (a *App) migrateConversionLineage(ctx context.Context, path, headID string, source desktopMigrationSource, cp *desktopMigrationCheckpoint, workspaceID string) (retErr error) {
tmp, err := os.MkdirTemp("", "reasonix-lineage-")
if err != nil {
return err
}
defer os.RemoveAll(tmp)
stages := []*session.Service{}
defer func() {
for _, stage := range stages {
retErr = errors.Join(retErr, stage.Shutdown(context.Background()))
}
}()
newStage := func() (*session.Service, error) {
// A continued descendant can have the same ID/source stamp as the
// import of its ancestor. Isolate inputs so reuse cannot substitute a
// descendant's newer history while computing the ancestor's digest.
dir, err := os.MkdirTemp(tmp, "source-")
if err != nil {
return nil, err
}
stage, err := session.NewService("migration-stage", session.NewFilesystemPersistence(filepath.Join(dir, "sessions-v4")))
if err == nil {
stages = append(stages, stage)
}
return stage, err
}
nodes := []desktopMigrationStagedNode{}
add := func(stage *session.Service, ref session.SessionRef, checkpoint desktopMigrationCheckpoint, origin session.SessionOrigin) error {
if err := stage.Close(ctx, ref); err != nil {
return err
}
messages, err := stage.Query().History(ctx, ref)
if err != nil {
return err
}
digest, err := agent.ContentDigestForMessages(messages)
if err != nil {
return err
}
nodes = append(nodes, desktopMigrationStagedNode{service: stage, checkpoint: checkpoint, ref: ref, messages: messages, digest: digest, coveredBy: -1, origin: origin})
return nil
}
for _, converted := range source.headConversions {
key := desktopCanonicalMigrationKey(converted.Root, converted.SessionID)
checkpoint, err := newDesktopMigrationCheckpoint(source, key, canonicalMigrationSourceFiles(converted.Root, converted.SessionID))
if err != nil {
return err
}
stage, err := newStage()
if err != nil {
return err
}
var runtime *session.Runtime
if converted.Codec == session.Codec {
runtime, _, err = stage.ContinueImportedSource(ctx, path, filepath.Join(converted.Root, converted.SessionID), headID)
} else {
runtime, _, err = stage.ContinuePrototype(ctx, filepath.Join(converted.Root, converted.SessionID))
}
if err != nil {
return errors.Join(err, updateDesktopMigrationLedger(key, "", "failed", "conversion_read"))
}
if err := add(stage, runtime.Ref(), checkpoint, session.SessionOriginCanonicalImport); err != nil {
return err
}
}
if cp != nil {
stage, err := newStage()
if err != nil {
return err
}
runtime, _, err := stage.ContinueImported(ctx, path, headID)
if err != nil {
return err
}
if err := add(stage, runtime.Ref(), *cp, session.SessionOriginLegacyImport); err != nil {
return err
}
}
for _, node := range nodes {
if node.checkpoint.completed() {
source.conversionAdoptions = append(source.conversionAdoptions, &desktopMigrationReceipt{TargetSessionID: node.checkpoint.record.TargetSessionID, ContentDigest: node.checkpoint.record.ContentDigest})
}
}
reduceDesktopMigrationLineage(nodes)
for i := range nodes {
if nodes[i].coveredBy >= 0 {
continue
}
if err := a.publishStagedMigration(ctx, source, nodes[i].checkpoint, nodes[i].service, nodes[i].ref, workspaceID, nodes[i].origin); err != nil {
return err
}
ledger, err := readDesktopMigrationLedger()
if err != nil {
return err
}
nodes[i].targetID = ledger.Records[nodes[i].checkpoint.key].TargetSessionID
}
for i := range nodes {
if nodes[i].coveredBy < 0 {
continue
}
winner := nodes[i].coveredBy
for nodes[winner].coveredBy >= 0 {
winner = nodes[winner].coveredBy
}
// Preserve an older receipt for a target that may already have been
// continued/deleted. Do not detach or resurrect that user's target.
targetID := nodes[winner].targetID
if nodes[i].checkpoint.matchesCompletedContent(nodes[i].digest) {
targetID = nodes[i].checkpoint.record.TargetSessionID
}
if err := nodes[i].checkpoint.complete(targetID, nodes[i].digest); err != nil {
return err
}
}
return nil
}
// Select maximal histories within an already-proven lineage. Equal histories
// prefer the earlier (more recently converted) node; prefix edges cannot cycle.
func reduceDesktopMigrationLineage(nodes []desktopMigrationStagedNode) {
for i := range nodes {
for j := range nodes {
if i == j || !session.MigrationHistoryContains(nodes[j].messages, nodes[i].messages) {
continue
}
if session.MigrationHistoryContains(nodes[i].messages, nodes[j].messages) && j > i {
continue
}
nodes[i].coveredBy = j
break
}
}
}
// Stored-only chains (for example native v3 -> v4 with both directories left
// behind) use the same reduction even when no JSONL source remains on disk.
func (a *App) migrateStoredConversionLineages(ctx context.Context, conversions map[string][]desktopMigrationConversion, sources map[string]*desktopMigrationSource, handled map[string]bool) error {
var joined error
for path, candidates := range conversions {
remaining := []desktopMigrationConversion{}
for _, candidate := range candidates {
if handled[canonicalRuntimeRoot(filepath.Join(candidate.Root, candidate.SessionID))] {
continue
}
owner := sources[canonicalRuntimeRoot(candidate.Root)]
if owner != nil && owner.pairedIDs[candidate.SessionID] {
continue
}
remaining = append(remaining, candidate)
}
if len(remaining) < 2 {
continue
}
// Head recovery can require replaying an archived DAG. Completed
// stored-only chains must bypass that work just like live legacy files.
ledger, err := readDesktopMigrationLedger()
if err != nil {
return errors.Join(joined, err)
}
allUnchanged := true
for _, candidate := range remaining {
cp, err := newDesktopMigrationCheckpoint(desktopMigrationSource{records: ledger.Records}, desktopCanonicalMigrationKey(candidate.Root, candidate.SessionID), canonicalMigrationSourceFiles(candidate.Root, candidate.SessionID))
if err != nil || !cp.unchanged() {
allUnchanged = false
break
}
if err := cp.skip(); err != nil {
joined = errors.Join(joined, err)
allUnchanged = false
break
}
}
if allUnchanged {
for _, candidate := range remaining {
handled[canonicalRuntimeRoot(filepath.Join(candidate.Root, candidate.SessionID))] = true
}
continue
}
resolved, err := resolveDesktopConversionHeads(ctx, path, remaining)
if err != nil {
joined = errors.Join(joined, err)
for _, candidate := range remaining {
handled[canonicalRuntimeRoot(filepath.Join(candidate.Root, candidate.SessionID))] = true
joined = errors.Join(joined, updateDesktopMigrationLedger(desktopCanonicalMigrationKey(candidate.Root, candidate.SessionID), "", "failed", "conversion_heads"))
}
continue
}
groups := map[string][]desktopMigrationConversion{}
for _, candidate := range resolved {
groups[candidate.HeadID] = append(groups[candidate.HeadID], candidate)
}
for _, group := range groups {
if len(group) < 2 {
continue
}
owner := sources[canonicalRuntimeRoot(group[0].Root)]
if owner == nil {
continue
}
source := *owner
source.headConversions = group
ledger, err := readDesktopMigrationLedger()
if err != nil {
return errors.Join(joined, err)
}
source.records = ledger.Records
unchanged := true
for _, candidate := range group {
handled[canonicalRuntimeRoot(filepath.Join(candidate.Root, candidate.SessionID))] = true
cp, err := newDesktopMigrationCheckpoint(source, desktopCanonicalMigrationKey(candidate.Root, candidate.SessionID), canonicalMigrationSourceFiles(candidate.Root, candidate.SessionID))
if err != nil {
joined = errors.Join(joined, err)
unchanged = false
continue
}
if !cp.unchanged() {
unchanged = false
} else if err := cp.skip(); err != nil {
joined = errors.Join(joined, err)
unchanged = false
}
}
if unchanged {
continue
}
if err := a.migrateConversionLineage(ctx, "", "", source, nil, ""); err != nil {
joined = errors.Join(joined, err)
}
}
}
return joined
}
@@ -0,0 +1,450 @@
package main
import (
"bytes"
"encoding/json"
"fmt"
"os"
"path/filepath"
"testing"
"reasonix/internal/agent"
"reasonix/internal/config"
"reasonix/internal/provider"
"reasonix/internal/session"
)
// Re-encode a real conversion in the retired JSONL codec, retaining Source and
// the immutable legacy snapshot exactly as old v2 -> v3 migration did.
func encodeMigrationFixtureAsV3(t *testing.T, dir string) {
t.Helper()
commits, err := session.Replay(dir, nil)
if err != nil {
t.Fatal(err)
}
manifest, err := readDesktopMigrationManifest(dir)
if err != nil {
t.Fatal(err)
}
manifest.SchemaVersion, manifest.Codec, manifest.StorageRevision, manifest.ContentRoot = 3, session.FinalV31Codec, 0, ""
writeMigrationJSON(t, filepath.Join(dir, "manifest.json"), manifest)
var log bytes.Buffer
for _, commit := range commits {
commit.SchemaVersion, commit.Codec = 3, session.FinalV31Codec
if err := json.NewEncoder(&log).Encode(commit); err != nil {
t.Fatal(err)
}
}
if err := os.WriteFile(filepath.Join(dir, "events.jsonl"), log.Bytes(), 0o600); err != nil {
t.Fatal(err)
}
if err := os.Remove(filepath.Join(dir, "events.frames")); err != nil {
t.Fatal(err)
}
}
func convertedV3Fixture(t *testing.T, path, root, head string, extra ...string) string {
t.Helper()
converted, err := session.MigrateLegacyHead(t.Context(), path, root, head)
if err != nil {
t.Fatal(err)
}
service, err := session.NewService("fixture", session.NewFilesystemPersistence(root))
if err != nil {
t.Fatal(err)
}
for _, id := range extra {
appendMigrationTestMessage(t, service, session.SessionRef{HostID: "fixture", SessionID: converted.TargetID}, id)
}
if err := service.Shutdown(t.Context()); err != nil {
t.Fatal(err)
}
encodeMigrationFixtureAsV3(t, converted.TargetDir)
return converted.TargetID
}
func migrationSingleDAGFixture(t *testing.T) (string, *agent.Session, string) {
t.Helper()
path := filepath.Join(config.SessionDir(), "original-v2.jsonl")
legacy := agent.NewSession("system")
legacy.Add(provider.Message{ID: "question", Role: provider.RoleUser, Content: "original question"})
if err := legacy.Save(path); err != nil {
t.Fatal(err)
}
heads, err := agent.ListSessionHeads(path)
if err != nil || len(heads) != 1 {
t.Fatalf("DAG fixture: %v %v", heads, err)
}
return path, legacy, heads[0].ID
}
func assertLineageRestart(t *testing.T, app *App, count int) {
t.Helper()
histories := v5MigrationHistories(t, app)
if len(histories) != count {
t.Fatalf("expected %d independent histories: %v", count, histories)
}
for id := range histories {
appendMigrationTestMessage(t, app.desktopSessionService(""), session.SessionRef{HostID: localDesktopHostID, SessionID: id}, "continued-in-v5")
}
app.closeSessionServices()
ledgerBefore, err := os.ReadFile(desktopMigrationLedgerPath())
if err != nil {
t.Fatal(err)
}
restarted := NewApp()
t.Cleanup(restarted.closeSessionServices)
t.Setenv("TMPDIR", filepath.Join(t.TempDir(), "missing"))
for range 2 {
if err := restarted.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
}
ledgerAfter, err := os.ReadFile(desktopMigrationLedgerPath())
if err != nil || !bytes.Equal(ledgerBefore, ledgerAfter) {
t.Fatalf("unchanged lineage rewrote ledger: %v", err)
}
after := v5MigrationHistories(t, restarted)
if len(after) != count {
t.Fatalf("restart duplicated histories: %v", after)
}
for id := range histories {
if got := after[id]; len(got) == 0 || got[len(got)-1] != "continued-in-v5" {
t.Fatalf("continued target lost: %s %v", id, got)
}
}
// Completed aliases must not resurrect a deliberately deleted v5 target.
for id := range histories {
if err := restarted.desktopSessionService("").Delete(t.Context(), session.SessionRef{HostID: localDesktopHostID, SessionID: id}); err != nil {
t.Fatal(err)
}
if err := restarted.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
if _, err := os.Stat(filepath.Join(restarted.desktopSessions.root, id)); !os.IsNotExist(err) {
t.Fatalf("deleted target resurrected: %v", err)
}
break
}
}
func TestDesktopV5LineageV2ConvertedToV3(t *testing.T) {
for _, explicitHead := range []bool{false, true} {
for _, scenario := range []string{"equal", "v3-newer", "v2-newer", "diverged", "v2-already-in-v5", "v3-already-in-v5"} {
t.Run(scenario+map[bool]string{false: "/selected", true: "/explicit"}[explicitHead], func(t *testing.T) {
isolateDesktopUserDirs(t)
path, legacy, head := migrationSingleDAGFixture(t)
selected := ""
if explicitHead {
selected = head
}
root := filepath.Join(filepath.Dir(config.SessionStoreDir()), "sessions-v3")
var extra []string
if scenario == "v3-newer" || scenario == "diverged" || scenario == "v3-already-in-v5" {
extra = []string{"continued-in-v3"}
}
id := convertedV3Fixture(t, path, root, selected, extra...)
if scenario == "v2-newer" || scenario == "diverged" {
legacy.Add(provider.Message{ID: "v2-added", Role: provider.RoleUser, Content: "continued-in-v2"})
if err := legacy.Save(path); err != nil {
t.Fatal(err)
}
}
before := migrationSourceSnapshot(t, append(legacyMigrationSourceFiles(path), canonicalMigrationSourceFiles(root, id)...))
app := NewApp()
t.Cleanup(app.closeSessionServices)
if scenario == "v2-already-in-v5" {
if err := app.migrateLegacySession(t.Context(), path, desktopMigrationSource{root: filepath.Dir(path), scope: "global"}, ""); err != nil {
t.Fatal(err)
}
old := onlyMigrationRecord(t)
appendMigrationTestMessage(t, app.desktopSessionService(""), session.SessionRef{HostID: localDesktopHostID, SessionID: old.TargetSessionID}, "pre-existing-v5-work")
}
if scenario == "v3-already-in-v5" {
if err := app.migratePreviewSession(t.Context(), desktopMigrationSource{root: root, scope: "global"}, id); err != nil {
t.Fatal(err)
}
old := onlyMigrationRecord(t)
appendMigrationTestMessage(t, app.desktopSessionService(""), session.SessionRef{HostID: localDesktopHostID, SessionID: old.TargetSessionID}, "pre-existing-v5-work")
}
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
count := 1
if scenario == "diverged" {
count = 2
}
histories := v5MigrationHistories(t, app)
if len(histories) != count {
t.Fatalf("converted source duplicated: %v", histories)
}
want := "original question"
if scenario == "v3-newer" {
want = "continued-in-v3"
}
if scenario == "v2-newer" {
want = "continued-in-v2"
}
if scenario == "v3-already-in-v5" || scenario == "v2-already-in-v5" {
want = "pre-existing-v5-work"
}
if scenario != "diverged" {
for _, history := range histories {
if history[len(history)-1] != want {
t.Fatalf("newer history lost: %v", history)
}
}
}
assertLineageRestart(t, app, count)
assertMigrationSourceSnapshot(t, before)
})
}
}
}
func TestDesktopV5LineageResumesAfterTargetPublication(t *testing.T) {
isolateDesktopUserDirs(t)
path, _, head := migrationSingleDAGFixture(t)
root := filepath.Join(filepath.Dir(config.SessionStoreDir()), "sessions-v3")
id := convertedV3Fixture(t, path, root, head, "converted continuation")
converted := desktopMigrationConversion{Root: root, SessionID: id, HeadID: head, Codec: session.FinalV31Codec, Depth: 1, LegacyDir: filepath.Join(root, id, "legacy")}
source := desktopMigrationSource{root: filepath.Dir(path), scope: "global", headConversions: []desktopMigrationConversion{converted}, conversions: map[string][]desktopMigrationConversion{canonicalRuntimeRoot(path): {converted}}}
cp, err := newDesktopMigrationCheckpoint(source, desktopLegacyMigrationKey(path), desktopLegacyMigrationFiles(path, source))
if err != nil {
t.Fatal(err)
}
app := NewApp()
t.Cleanup(app.closeSessionServices)
if err := app.migrateConversionLineage(t.Context(), path, head, source, &cp, "missing-workspace"); err == nil {
t.Fatal("expected interrupted workspace publication")
}
infos, err := listAllCanonicalSessionInfo(t.Context(), app.desktopSessionService("").Query())
if err != nil || len(infos) != 1 {
t.Fatalf("expected one durable published target: %v %v", infos, err)
}
var first string
for id := range infos {
first = id
}
app.closeSessionServices()
app = NewApp()
t.Cleanup(app.closeSessionServices)
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
got := v5MigrationHistories(t, app)
if len(got) != 1 || len(got[first]) == 0 {
t.Fatalf("retry changed or duplicated target: %v", got)
}
assertLineageRestart(t, app, 1)
}
func TestDesktopV5LineageConvertedHeadDoesNotHideOtherHeads(t *testing.T) {
isolateDesktopUserDirs(t)
path, legacy, originalHead := migrationSingleDAGFixture(t)
child, err := legacy.ForkHead(path, legacy.Snapshot()[1].ID, agent.HeadKindFork, "other")
if err != nil {
t.Fatal(err)
}
legacy.Add(provider.Message{ID: "child", Role: provider.RoleUser, Content: "child history"})
if err := legacy.Save(path); err != nil {
t.Fatal(err)
}
root := filepath.Join(filepath.Dir(config.SessionStoreDir()), "sessions-v3")
convertedV3Fixture(t, path, root, "", "v3 child continuation")
// The manifest's omitted head refers to child at conversion time.
if err := legacy.SwitchHead(path, originalHead); err != nil {
t.Fatal(err)
}
app := NewApp()
t.Cleanup(app.closeSessionServices)
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
histories := v5MigrationHistories(t, app)
if len(histories) != 2 {
t.Fatalf("converted %s hid another head or duplicated itself: %v", child, histories)
}
assertLineageRestart(t, app, 2)
}
func TestDesktopV5LineageMultipleConversions(t *testing.T) {
for _, diverged := range []bool{false, true} {
t.Run(map[bool]string{false: "same-history", true: "independent-continuations"}[diverged], func(t *testing.T) {
isolateDesktopUserDirs(t)
path, _, head := migrationSingleDAGFixture(t)
v3root := filepath.Join(filepath.Dir(config.SessionStoreDir()), "sessions-v3")
convertedV3Fixture(t, path, v3root, head, "first conversion work")
extra := "first conversion work"
if diverged {
extra = "independent second conversion"
}
convertedV3Fixture(t, path, config.SessionStoreDir(), head, extra)
app := NewApp()
t.Cleanup(app.closeSessionServices)
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
want := 1
if diverged {
want = 2
}
assertLineageRestart(t, app, want)
})
}
}
func TestDesktopV5LineageMultipleGenerations(t *testing.T) {
for _, legacyPresent := range []bool{false, true} {
t.Run(map[bool]string{false: "stored-only", true: "v2-v3-v4"}[legacyPresent], func(t *testing.T) {
isolateDesktopUserDirs(t)
path, _, _ := migrationSingleDAGFixture(t)
v3root := filepath.Join(filepath.Dir(config.SessionStoreDir()), "sessions-v3")
id := convertedV3Fixture(t, path, v3root, "", "v3 continuation")
v4, err := session.NewService("fixture", session.NewFilesystemPersistence(config.SessionStoreDir()))
if err != nil {
t.Fatal(err)
}
runtime, _, err := v4.ContinuePrototype(t.Context(), filepath.Join(v3root, id))
if err != nil {
t.Fatal(err)
}
appendMigrationTestMessage(t, v4, runtime.Ref(), "v4 continuation")
if err := v4.Shutdown(t.Context()); err != nil {
t.Fatal(err)
}
if !legacyPresent {
if err := os.Remove(path); err != nil {
t.Fatal(err)
}
}
app := NewApp()
t.Cleanup(app.closeSessionServices)
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
for _, history := range v5MigrationHistories(t, app) {
if history[len(history)-1] != "v4 continuation" {
t.Fatalf("latest generation lost: %v", history)
}
}
ledger, err := readDesktopMigrationLedger()
if err != nil {
t.Fatal(err)
}
ancestor := ledger.Records[desktopCanonicalMigrationKey(v3root, id)]
descendant := ledger.Records[desktopCanonicalMigrationKey(config.SessionStoreDir(), runtime.Ref().SessionID)]
if ancestor.ContentDigest == "" || ancestor.ContentDigest == descendant.ContentDigest {
t.Fatal("ancestor digest was replaced by the continued descendant during staging")
}
assertLineageRestart(t, app, 1)
})
}
}
func TestDesktopV5LineageNativeV3AndUnrelatedEqualSessions(t *testing.T) {
isolateDesktopUserDirs(t)
root := filepath.Join(filepath.Dir(config.SessionStoreDir()), "sessions-v3")
messages := []provider.Message{{ID: "same", Role: provider.RoleUser, Content: "same content"}}
writeV3MigrationFixture(t, root, "native", session.FinalV31Codec, messages)
writeV3MigrationFixture(t, root, "unrelated", session.FinalV31Codec, messages)
v4, err := session.NewService("fixture", session.NewFilesystemPersistence(config.SessionStoreDir()))
if err != nil {
t.Fatal(err)
}
runtime, _, err := v4.ContinuePrototype(t.Context(), filepath.Join(root, "native"))
if err != nil {
t.Fatal(err)
}
appendMigrationTestMessage(t, v4, runtime.Ref(), "native v4 work")
if err := v4.Shutdown(t.Context()); err != nil {
t.Fatal(err)
}
app := NewApp()
t.Cleanup(app.closeSessionServices)
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
assertLineageRestart(t, app, 2)
}
func TestDesktopV5LineageDeletedNativeV3WithMultipleConversions(t *testing.T) {
for _, scenario := range []string{"equal", "prefix", "diverged"} {
t.Run(scenario, func(t *testing.T) {
isolateDesktopUserDirs(t)
root := filepath.Join(filepath.Dir(config.SessionStoreDir()), "sessions-v3")
writeV3MigrationFixture(t, root, "native", session.FinalV31Codec, []provider.Message{{ID: "original", Role: provider.RoleUser, Content: "original"}})
for index, targetRoot := range []string{config.SessionStoreDir(), config.ProjectSessionStoreDir(globalWorkspaceRoot())} {
v4, err := session.NewService("fixture", session.NewFilesystemPersistence(targetRoot))
if err != nil {
t.Fatal(err)
}
runtime, _, err := v4.ContinuePrototype(t.Context(), filepath.Join(root, "native"))
if err != nil {
t.Fatal(err)
}
work := "shared continuation"
if index == 1 && scenario == "diverged" {
work = "independent continuation"
}
appendMigrationTestMessage(t, v4, runtime.Ref(), work)
if index == 1 && scenario == "prefix" {
appendMigrationTestMessage(t, v4, runtime.Ref(), "later work")
}
if err := v4.Shutdown(t.Context()); err != nil {
t.Fatal(err)
}
}
if err := os.RemoveAll(filepath.Join(root, "native")); err != nil {
t.Fatal(err)
}
app := NewApp()
t.Cleanup(app.closeSessionServices)
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
want := 1
if scenario == "diverged" {
want = 2
}
assertLineageRestart(t, app, want)
})
}
}
func TestDesktopV5LineageIncludesUnstampedPairedCanonical(t *testing.T) {
isolateDesktopUserDirs(t)
path, legacy, head := migrationSingleDAGFixture(t)
root := config.SessionStoreDir()
convertedV3Fixture(t, path, filepath.Join(filepath.Dir(root), "sessions-v3"), head, "converted work")
canonical, err := session.NewService("fixture", session.NewFilesystemPersistence(root))
if err != nil {
t.Fatal(err)
}
runtime, err := canonical.Create(t.Context(), session.CreateOptions{SessionID: agent.BranchID(path)})
if err != nil {
t.Fatal(err)
}
for index, message := range legacy.Snapshot() {
payload, _ := json.Marshal(map[string]any{"message": message})
if _, err := runtime.Session().AppendBatch(t.Context(), fmt.Sprintf("seed-%d", index), []session.Event{{Kind: "message/complete", Payload: payload}}); err != nil {
t.Fatal(err)
}
}
appendMigrationTestMessage(t, canonical, runtime.Ref(), "converted work")
appendMigrationTestMessage(t, canonical, runtime.Ref(), "paired canonical work")
if err := canonical.Shutdown(t.Context()); err != nil {
t.Fatal(err)
}
app := NewApp()
t.Cleanup(app.closeSessionServices)
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
for _, history := range v5MigrationHistories(t, app) {
if history[len(history)-1] != "paired canonical work" {
t.Fatalf("paired canonical history omitted: %v", history)
}
}
assertLineageRestart(t, app, 1)
}
+60
View File
@@ -0,0 +1,60 @@
package main
import (
"path/filepath"
"strings"
"reasonix/internal/agent"
"reasonix/internal/store"
)
// Recovery copies are durable safety artifacts, not ordinary conversations.
// Classification uses the historical writer's identity markers, never titles,
// text equality or the mere presence of a parent (which also marks user forks).
// Missing display metadata must still allow ordinary history to migrate.
func desktopMigrationAutomaticRecovery(path string) bool {
if agent.LooksLikeRecoveryFilename(path) {
return true
}
meta, found, err := agent.LoadBranchMeta(path)
return err == nil && found && (meta.Recovered || meta.EffectiveVersionKind() == agent.VersionRecovery || strings.TrimSpace(meta.RecoveryDigest) != "")
}
// Carry the exclusion through paired stores and conversion ancestry. Original
// files can be gone, so consult the immutable archived metadata as well. This
// only controls discovery; it never deletes or rewrites any recovery artifact.
func desktopMigrationRecoveryStores(nodes map[string]desktopMigrationStoredSource) map[string]bool {
hidden := map[string]bool{}
var recovery func(desktopMigrationStoredSource, map[string]bool) bool
recovery = func(node desktopMigrationStoredSource, seen map[string]bool) bool {
key := canonicalRuntimeRoot(node.dir)
if seen[key] || len(seen) >= 32 {
return false
}
seen[key] = true
if agent.LooksLikeRecoveryFilename(node.manifest.SessionID+".jsonl") ||
desktopMigrationAutomaticRecovery(filepath.Join(filepath.Dir(filepath.Dir(node.dir)), "sessions", node.manifest.SessionID+".jsonl")) {
return true
}
source := node.manifest.Source
if source == nil {
return false
}
if store.IsSessionTranscriptName(filepath.Base(source.Path)) && (source.Version == "" || source.Version == "legacy") {
return desktopMigrationAutomaticRecovery(source.Path) ||
desktopMigrationAutomaticRecovery(filepath.Join(node.dir, "legacy", filepath.Base(source.Path)))
}
if parent, found := nodes[canonicalRuntimeRoot(source.Path)]; found {
return recovery(parent, seen)
}
dir := filepath.Join(node.dir, "legacy", "prototype")
manifest, err := readDesktopMigrationManifest(dir)
return err == nil && recovery(desktopMigrationStoredSource{dir: dir, manifest: manifest}, seen)
}
for key, node := range nodes {
if recovery(node, map[string]bool{}) {
hidden[key] = true
}
}
return hidden
}
@@ -0,0 +1,161 @@
package main
import (
"os"
"path/filepath"
"testing"
"reasonix/internal/agent"
"reasonix/internal/config"
"reasonix/internal/provider"
"reasonix/internal/session"
)
func TestDesktopV5UpgradeExcludesAutomaticRecoveryCopies(t *testing.T) {
isolateDesktopUserDirs(t)
dir := config.SessionDir()
for _, fixture := range []struct {
name string
meta agent.BranchMeta
want bool
}{
{"ordinary", agent.BranchMeta{}, true},
{"user-fork", agent.BranchMeta{ParentID: "ordinary"}, true},
{"my-recovery-notes", agent.BranchMeta{}, true},
{"marked-copy", agent.BranchMeta{Recovered: true, ParentID: "ordinary"}, false},
{"version-copy", agent.BranchMeta{VersionKind: agent.VersionRecovery}, false},
{"digest-copy", agent.BranchMeta{RecoveryDigest: "historical-marker"}, false},
{"ordinary-recovery-0123456789abcdef", agent.BranchMeta{}, false},
{"ordinary-recovery-0123456789abcdef-recovery-fedcba9876543210", agent.BranchMeta{}, false},
} {
path := filepath.Join(dir, fixture.name+".jsonl")
legacy := agent.NewSession("system")
legacy.Add(provider.Message{ID: fixture.name, Role: provider.RoleUser, Content: fixture.name})
if err := legacy.Save(path); err != nil {
t.Fatal(err)
}
writeMigrationJSON(t, agent.BranchMetaPath(path), fixture.meta)
if fixture.name == "ordinary-recovery-0123456789abcdef" {
if err := os.Remove(agent.BranchMetaPath(path)); err != nil {
t.Fatal(err)
}
}
if got := desktopMigrationAutomaticRecovery(path); got == fixture.want {
t.Fatalf("unexpected recovery classification: %s = %v", fixture.name, got)
}
}
app := NewApp()
t.Cleanup(app.closeSessionServices)
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
for _, history := range v5MigrationHistories(t, app) {
last := history[len(history)-1]
if last != "ordinary" && last != "user-fork" && last != "my-recovery-notes" {
t.Fatalf("recovery artifact appeared as a conversation: %v", history)
}
}
assertLineageRestart(t, app, 3)
}
func TestDesktopV5UpgradeExcludesConvertedRecoveryCopies(t *testing.T) {
for _, removed := range []bool{false, true} {
name := "original-present"
if removed {
name = "original-removed"
}
t.Run(name, func(t *testing.T) {
isolateDesktopUserDirs(t)
path, _, _ := migrationSingleDAGFixture(t)
copyPath := filepath.Join(filepath.Dir(path), "original-v2-recovery-0123456789abcdef.jsonl")
copySession := agent.NewSession("system")
copySession.Add(provider.Message{ID: "copy-only", Role: provider.RoleUser, Content: "automatic conflict data"})
if err := copySession.Save(copyPath); err != nil {
t.Fatal(err)
}
v3root := filepath.Join(filepath.Dir(config.SessionStoreDir()), "sessions-v3")
id := convertedV3Fixture(t, copyPath, v3root, "")
v4, err := session.NewService("fixture", session.NewFilesystemPersistence(config.SessionStoreDir()))
if err != nil {
t.Fatal(err)
}
if _, _, err := v4.ContinuePrototype(t.Context(), filepath.Join(v3root, id)); err != nil {
t.Fatal(err)
}
if err := v4.Shutdown(t.Context()); err != nil {
t.Fatal(err)
}
if removed {
for _, file := range legacyMigrationSourceFiles(copyPath) {
if err := os.Remove(file); err != nil && !os.IsNotExist(err) {
t.Fatal(err)
}
}
// Only the descendant's archived manifest still knows the origin.
if err := os.RemoveAll(filepath.Join(v3root, id)); err != nil {
t.Fatal(err)
}
}
app := NewApp()
t.Cleanup(app.closeSessionServices)
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
assertLineageRestart(t, app, 1)
})
}
}
func TestDesktopV5UpgradeExcludesRecoveryStoreWithoutTranscript(t *testing.T) {
isolateDesktopUserDirs(t)
root := config.SessionStoreDir()
for _, id := range []string{"ordinary", "only-meta-copy", "ordinary-recovery-0123456789abcdef"} {
service, err := session.NewService("fixture", session.NewFilesystemPersistence(root))
if err != nil {
t.Fatal(err)
}
runtime, err := service.Create(t.Context(), session.CreateOptions{SessionID: id})
if err != nil {
t.Fatal(err)
}
appendMigrationTestMessage(t, service, runtime.Ref(), id)
if err := service.Shutdown(t.Context()); err != nil {
t.Fatal(err)
}
}
writeMigrationJSON(t, agent.BranchMetaPath(filepath.Join(config.SessionDir(), "only-meta-copy.jsonl")), agent.BranchMeta{Recovered: true})
app := NewApp()
t.Cleanup(app.closeSessionServices)
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
assertLineageRestart(t, app, 1)
}
func TestDesktopV5UpgradeExcludesArchivedRecoveryMetadata(t *testing.T) {
isolateDesktopUserDirs(t)
migrationSingleDAGFixture(t)
path := filepath.Join(config.SessionDir(), "metadata-marked-copy.jsonl")
legacy := agent.NewSession("system")
legacy.Add(provider.Message{ID: "automatic", Role: provider.RoleUser, Content: "conflict copy"})
if err := legacy.Save(path); err != nil {
t.Fatal(err)
}
writeMigrationJSON(t, agent.BranchMetaPath(path), agent.BranchMeta{Recovered: true})
root := filepath.Join(filepath.Dir(config.SessionStoreDir()), "sessions-v3")
id := convertedV3Fixture(t, path, root, "")
for _, file := range legacyMigrationSourceFiles(path) {
if err := os.Remove(file); err != nil && !os.IsNotExist(err) {
t.Fatal(err)
}
}
archived := filepath.Join(root, id, "legacy", filepath.Base(path))
before := migrationSourceSnapshot(t, append(canonicalMigrationSourceFiles(root, id), legacyMigrationSourceFiles(archived)...))
app := NewApp()
t.Cleanup(app.closeSessionServices)
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
assertLineageRestart(t, app, 1)
assertMigrationSourceSnapshot(t, before)
}
@@ -0,0 +1,102 @@
package main
import (
"bytes"
"encoding/json"
"path/filepath"
"testing"
"reasonix/internal/config"
"reasonix/internal/provider"
"reasonix/internal/session"
)
// Submission receipts introduced by #10392 are host metadata. Migration must
// preserve the event bytes, keep receipt scope tied to the session identity,
// and keep both title/history consumers and model messages compatible.
func TestDesktopV5UpgradeSubmissionIdentity(t *testing.T) {
for _, collision := range []bool{false, true} {
name := "same-identity"
if collision {
name = "independent-conflict-identity"
}
t.Run(name, func(t *testing.T) {
isolateDesktopUserDirs(t)
root := config.SessionStoreDir()
id := "accepted-source"
writer, err := session.CreateStore(filepath.Join(root, id), id)
if err != nil {
t.Fatal(err)
}
receipt := session.SubmissionReceipt{SessionID: id, SubmissionID: "send", Fingerprint: "request-fingerprint", TurnID: "turn", MessageID: "user"}
receiptBody, _ := json.Marshal(receipt)
message := provider.Message{ID: "user", Role: provider.RoleUser, Content: "question for the title"}
messageBody, _ := json.Marshal(map[string]any{"message": message})
if _, err := writer.Append(t.Context(), session.Batch{OperationID: "accept", TurnID: "turn", Events: []session.Event{
{Kind: "turn/start"},
{Kind: "submission/accepted", Optional: true, Payload: receiptBody},
{Kind: "message/complete", Payload: messageBody},
{Kind: "turn/end", Payload: json.RawMessage(`{"status":"completed"}`)},
}}); err != nil {
t.Fatal(err)
}
if err := writer.Close(t.Context()); err != nil {
t.Fatal(err)
}
before := migrationSourceSnapshot(t, canonicalMigrationSourceFiles(root, id))
app := NewApp()
t.Cleanup(app.closeSessionServices)
if collision {
other, err := app.desktopSessionService("").Create(t.Context(), session.CreateOptions{SessionID: id})
if err != nil {
t.Fatal(err)
}
appendMigrationTestMessage(t, app.desktopSessionService(""), other.Ref(), "unrelated existing work")
}
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
ledger, err := readDesktopMigrationLedger()
if err != nil {
t.Fatal(err)
}
targetID := ledger.Records[desktopCanonicalMigrationKey(root, id)].TargetSessionID
if (targetID != id) != collision {
t.Fatalf("unexpected identity: %q", targetID)
}
ref := session.SessionRef{HostID: localDesktopHostID, SessionID: targetID}
query := app.desktopSessionService("").Query()
// This uses #10389's new canonical title reader and waits for the
// same history index that #10392 annotates with submission IDs.
titles, err := query.TitleMessages(t.Context(), ref, 3)
if err != nil || len(titles) != 1 || titles[0].Content != message.Content {
t.Fatalf("migrated title input: %+v, %v", titles, err)
}
page, err := query.ReadHistoryWindow(t.Context(), ref, session.HistoryWindowRequest{Anchor: "newest", Limit: 10})
if err != nil || page.Status != "ready" || len(page.Messages) != 1 {
t.Fatalf("migrated history: %+v, %v", page, err)
}
wantSubmission := "send"
if collision {
wantSubmission = ""
}
if page.Messages[0].SubmissionID != wantSubmission {
t.Fatalf("submission scope: %+v", page.Messages[0])
}
snapshot, err := query.Snapshot(t.Context(), ref)
if err != nil {
t.Fatal(err)
}
receipts, _ := json.Marshal(snapshot.Projection.Submissions)
if !bytes.Contains(receipts, receiptBody) {
t.Fatalf("source receipt lost: %s", receipts)
}
model, _ := json.Marshal(snapshot.Projection.Messages)
if bytes.Contains(model, []byte("submissionId")) || bytes.Contains(model, []byte("fingerprint")) {
t.Fatal("submission metadata entered model messages")
}
assertMigrationSourceSnapshot(t, before)
assertLineageRestart(t, app, 1)
})
}
}
+164
View File
@@ -1,16 +1,180 @@
package main
import (
"bytes"
"encoding/json"
"errors"
"os"
"path/filepath"
"testing"
"reasonix/desktop/internal/workspacestate"
"reasonix/internal/agent"
"reasonix/internal/config"
"reasonix/internal/provider"
"reasonix/internal/session"
)
// These fixtures have no Desktop header or open tab, just like v4-only
// conversations left behind after a downgrade. Their catalog cache is absent.
func coldV4MigrationFixture(t *testing.T, root, id string) *session.Service {
t.Helper()
service, err := session.NewService("migration-source", session.NewFilesystemPersistence(root))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = service.Shutdown(t.Context()) })
runtime, err := service.Create(t.Context(), session.CreateOptions{SessionID: id})
if err != nil {
t.Fatal(err)
}
payload, _ := json.Marshal(map[string]any{"message": provider.Message{ID: "user", Role: provider.RoleUser, Content: "恢复完整对话"}})
if _, err := runtime.Session().AppendBatch(t.Context(), "message", []session.Event{{Kind: "message/complete", Payload: payload}}); err != nil {
t.Fatal(err)
}
if err := service.Close(t.Context(), runtime.Ref()); err != nil {
t.Fatal(err)
}
if err := os.RemoveAll(filepath.Join(root, ".query-cache")); err != nil {
t.Fatal(err)
}
info, err := session.NewFilesystemPersistence(root).Stat(t.Context(), id)
if err != nil || info.MetadataStatus != session.MetadataPending || info.Turns != 0 || info.Title != "" || info.Preview != "" {
t.Fatalf("fixture must reproduce cold, empty display metadata: %#v, %v", info, err)
}
return service
}
func TestDesktopV5StartupMigratesColdV4WithoutLegacyOrOpenTab(t *testing.T) {
for _, scope := range []string{"project", "global"} {
t.Run(scope, func(t *testing.T) {
isolateDesktopUserDirs(t)
workspace := filepath.Join(t.TempDir(), "中文项目")
if err := os.MkdirAll(workspace, 0o700); err != nil {
t.Fatal(err)
}
sourceRoot := config.SessionStoreDir()
if scope == "project" {
sourceRoot = config.ProjectSessionStoreDir(workspace)
if err := saveProjectsFile(desktopProjectFile{Projects: []desktopProject{{Root: workspace}}}); err != nil {
t.Fatal(err)
}
}
const id = "v4-native"
coldV4MigrationFixture(t, sourceRoot, id)
original := map[string][]byte{}
for _, name := range []string{"manifest.json", "events.frames"} {
body, err := os.ReadFile(filepath.Join(sourceRoot, id, name))
if err != nil {
t.Fatal(err)
}
original[name] = body
}
for attempt := range 2 {
app := NewApp()
t.Cleanup(app.closeSessionServices)
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
workspaceID := workspacestate.GlobalWorkspaceID
if scope == "project" {
workspaceID = desktopWorkspaceID(scope, workspace)
}
state, err := app.workspaceRegistry().Load(t.Context())
if err != nil {
t.Fatal(err)
}
if ids := state.Workspaces[workspaceID].SessionIDs; len(ids) != 1 || ids[0] != id {
t.Fatalf("attempt %d: cold v4 membership = %v", attempt, ids)
}
page, err := app.ReadSessionHistory(session.SessionRef{HostID: localDesktopHostID, SessionID: id}, "", 10)
if err != nil || len(page.Messages) != 1 || page.Messages[0].Content != "恢复完整对话" {
t.Fatalf("restored history = %#v, %v", page, err)
}
diagnostics, err := app.GetSessionArchitectureDiagnostics()
if err != nil || diagnostics.MigrationCompleted != 1 || diagnostics.SessionHeadersTotal != 1 {
t.Fatalf("migration diagnostics = %#v, %v", diagnostics, err)
}
app.closeSessionServices()
}
for name, before := range original {
after, err := os.ReadFile(filepath.Join(sourceRoot, id, name))
if err != nil || !bytes.Equal(before, after) {
t.Fatalf("source %s changed: %v", name, err)
}
}
info, err := session.NewFilesystemPersistence(sourceRoot).Stat(t.Context(), id)
if err != nil || info.MetadataStatus != session.MetadataPending {
t.Fatalf("migration must not rebuild source display metadata: %#v, %v", info, err)
}
})
}
}
func TestCanonicalV4MigrationReportsSourceFailureAndRetriesAfterRepair(t *testing.T) {
isolateDesktopUserDirs(t)
root := config.SessionStoreDir()
coldV4MigrationFixture(t, root, "healthy")
coldV4MigrationFixture(t, root, "damaged")
manifestPath := filepath.Join(root, "damaged", "manifest.json")
original, err := os.ReadFile(manifestPath)
if err != nil {
t.Fatal(err)
}
if err := os.WriteFile(manifestPath, []byte("broken"), 0o600); err != nil {
t.Fatal(err)
}
app := NewApp()
t.Cleanup(app.closeSessionServices)
if err := app.migrateDesktopSessionsV5(t.Context()); err == nil {
t.Fatal("damaged source must not silently disappear")
}
diagnostics, err := app.GetSessionArchitectureDiagnostics()
if err != nil || diagnostics.MigrationFailed != 1 || diagnostics.MigrationCompleted != 1 || diagnostics.WorkspaceMembersTotal != 1 {
t.Fatalf("failure must be reported while healthy source migrates: %#v, %v", diagnostics, err)
}
if err := os.WriteFile(manifestPath, original, 0o600); err != nil {
t.Fatal(err)
}
app.closeSessionServices()
app = NewApp()
t.Cleanup(app.closeSessionServices)
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
diagnostics, err = app.GetSessionArchitectureDiagnostics()
if err != nil || diagnostics.MigrationFailed != 0 || diagnostics.MigrationCompleted != 2 || diagnostics.WorkspaceMembersTotal != 2 {
t.Fatalf("repaired source must retry: %#v, %v", diagnostics, err)
}
}
func TestCanonicalV4MigrationResumesAfterPublicationBeforeWorkspaceAttach(t *testing.T) {
isolateDesktopUserDirs(t)
root := config.SessionStoreDir()
old := coldV4MigrationFixture(t, root, "published")
app := NewApp()
t.Cleanup(app.closeSessionServices)
source := desktopMigrationSource{root: root, scope: "global"}
err := app.migrateCanonicalSession(t.Context(), old, source, "missing-workspace", "published")
if !errors.Is(err, workspacestate.ErrWorkspaceNotFound) {
t.Fatalf("expected interruption at registry publication: %v", err)
}
ref := session.SessionRef{HostID: localDesktopHostID, SessionID: "published"}
if _, err := app.desktopSessionService("").Query().Snapshot(t.Context(), ref); err != nil {
t.Fatalf("target must already be durable: %v", err)
}
app.closeSessionServices()
app = NewApp()
t.Cleanup(app.closeSessionServices)
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
diagnostics, err := app.GetSessionArchitectureDiagnostics()
if err != nil || diagnostics.MigrationFailed != 0 || diagnostics.MigrationCompleted != 1 || diagnostics.SessionHeadersTotal != 1 || diagnostics.WorkspaceMembersTotal != 1 {
t.Fatalf("restart must attach existing target exactly once: %#v, %v", diagnostics, err)
}
}
func TestCanonicalV4MigrationPublishesHeaderThenWorkspaceMembershipIdempotently(t *testing.T) {
isolateDesktopUserDirs(t)
root := t.TempDir()
@@ -0,0 +1,544 @@
package main
import (
"bytes"
"encoding/json"
"errors"
"os"
"path/filepath"
"reflect"
"sort"
"strings"
"testing"
"time"
"reasonix/desktop/internal/workspacestate"
"reasonix/internal/agent"
"reasonix/internal/config"
"reasonix/internal/provider"
"reasonix/internal/session"
"reasonix/internal/store"
)
func writeMigrationJSON(t *testing.T, path string, value any) {
t.Helper()
body, err := json.Marshal(value)
if err != nil {
t.Fatal(err)
}
if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(path, append(body, '\n'), 0o600); err != nil {
t.Fatal(err)
}
}
func migrationSourceSnapshot(t *testing.T, paths []string) map[string][]byte {
t.Helper()
result := map[string][]byte{}
for _, path := range paths {
body, err := os.ReadFile(path)
if os.IsNotExist(err) {
continue
}
if err != nil {
t.Fatal(err)
}
result[path] = body
}
return result
}
func assertMigrationSourceSnapshot(t *testing.T, before map[string][]byte) {
t.Helper()
for path, body := range before {
after, err := os.ReadFile(path)
if err != nil || !bytes.Equal(body, after) {
t.Fatalf("source changed: %s: %v", path, err)
}
}
}
func v5MigrationHistories(t *testing.T, app *App) map[string][]string {
t.Helper()
state, err := app.workspaceRegistry().Load(t.Context())
if err != nil {
t.Fatal(err)
}
result := map[string][]string{}
for _, workspace := range state.Workspaces {
for _, id := range workspace.SessionIDs {
page, err := app.ReadSessionHistory(session.SessionRef{HostID: localDesktopHostID, SessionID: id}, "", 100)
if err != nil {
t.Fatal(err)
}
var content []string
for _, message := range page.Messages {
content = append(content, message.Content)
}
result[id] = content
}
}
return result
}
// Make a real, validated commit with the public writer, then encode it using
// the historical wire envelope. The retired format uses JSONL, not frames.
func writeV3MigrationFixture(t *testing.T, root, id, codec string, messages []provider.Message) {
t.Helper()
tmp := filepath.Join(t.TempDir(), id)
writer, err := session.CreateStore(tmp, id)
if err != nil {
t.Fatal(err)
}
var events []session.Event
for _, message := range messages {
payload, _ := json.Marshal(map[string]any{"message": message})
events = append(events, session.Event{Kind: "message/complete", Payload: payload})
}
if _, err := writer.Append(t.Context(), session.Batch{OperationID: "fixture", Events: events}); err != nil {
t.Fatal(err)
}
if err := writer.Close(t.Context()); err != nil {
t.Fatal(err)
}
commits, err := session.Replay(tmp, nil)
if err != nil {
t.Fatal(err)
}
dir := filepath.Join(root, id)
writeMigrationJSON(t, filepath.Join(dir, "manifest.json"), session.Manifest{SchemaVersion: 3, Codec: codec, SessionID: id, WriterGeneration: 1, CreatedAt: time.Now().UTC()})
var log bytes.Buffer
for _, commit := range commits {
commit.SchemaVersion, commit.Codec = 3, codec
if err := json.NewEncoder(&log).Encode(commit); err != nil {
t.Fatal(err)
}
}
if err := os.WriteFile(filepath.Join(dir, "events.jsonl"), log.Bytes(), 0o600); err != nil {
t.Fatal(err)
}
}
func TestDesktopV5UpgradeV1WithoutMetadata(t *testing.T) {
for _, scope := range []string{"global", "project"} {
for _, events := range []bool{false, true} {
t.Run(scope+"/events="+map[bool]string{false: "no", true: "yes"}[events], func(t *testing.T) {
isolateDesktopUserDirs(t)
root := config.SessionDir()
if scope == "project" {
workspace := filepath.Join(t.TempDir(), "中文旧项目")
root = desktopSessionDir(workspace)
if err := saveProjectsFile(desktopProjectFile{Projects: []desktopProject{{Root: workspace}}}); err != nil {
t.Fatal(err)
}
}
path := filepath.Join(root, "old.jsonl")
writeMigrationJSON(t, path, provider.Message{Role: provider.RoleUser, Content: "checkpoint"})
want := "checkpoint"
if events {
want = "newer-v1-event"
writeMigrationJSON(t, store.SessionEventLog(path), struct {
SchemaVersion int `json:"schema_version"`
Type string `json:"type"`
Messages []provider.Message `json:"messages"`
}{1, "replace", []provider.Message{{Role: provider.RoleUser, Content: want}}})
}
// These are not independent transcripts even though they end in JSONL.
for _, suffix := range []string{".turns.jsonl", ".guardian.jsonl", ".conflicts.jsonl"} {
if err := os.WriteFile(filepath.Join(root, "old"+suffix), []byte("not a conversation\n"), 0o600); err != nil {
t.Fatal(err)
}
}
before := migrationSourceSnapshot(t, legacyMigrationSourceFiles(path))
app := NewApp()
t.Cleanup(app.closeSessionServices)
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
histories := v5MigrationHistories(t, app)
if len(histories) != 1 {
t.Fatalf("histories=%v", histories)
}
for id, history := range histories {
if !reflect.DeepEqual(history, []string{want}) {
t.Fatalf("history=%v", history)
}
appendMigrationTestMessage(t, app.desktopSessionService(""), session.SessionRef{HostID: localDesktopHostID, SessionID: id}, "continued")
}
app.closeSessionServices()
app = NewApp()
t.Cleanup(app.closeSessionServices)
// A completed source must work without staging or history replay.
t.Setenv("TMPDIR", filepath.Join(t.TempDir(), "missing"))
for range 2 {
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
}
after := v5MigrationHistories(t, app)
if len(after) != 1 {
t.Fatalf("duplicate after continuation: %v", after)
}
for _, history := range after {
if history[len(history)-1] != "continued" {
t.Fatalf("lost continued history: %v", history)
}
}
assertMigrationSourceSnapshot(t, before)
})
}
}
}
func TestDesktopV5UpgradeAllV2HeadsAndOldReceipt(t *testing.T) {
for _, previousMigration := range []bool{false, true} {
t.Run(map[bool]string{false: "fresh", true: "previous-migrator"}[previousMigration], func(t *testing.T) {
isolateDesktopUserDirs(t)
path := filepath.Join(config.SessionDir(), "branches.jsonl")
legacy := agent.NewSession("system")
legacy.Add(provider.Message{ID: "question", Role: provider.RoleUser, Content: "root question"})
legacy.Add(provider.Message{ID: "answer", Role: provider.RoleAssistant, Content: "root answer"})
if err := legacy.Save(path); err != nil {
t.Fatal(err)
}
forkAt := legacy.Snapshot()[2].ID
child, err := legacy.ForkHead(path, forkAt, agent.HeadKindFork, "child")
if err != nil {
t.Fatal(err)
}
legacy.Add(provider.Message{ID: "child-only", Role: provider.RoleUser, Content: "child only"})
if err := legacy.Save(path); err != nil {
t.Fatal(err)
}
heads, err := agent.ListSessionHeads(path)
if err != nil {
t.Fatal(err)
}
var rootHead string
for _, head := range heads {
if head.ID != child {
rootHead = head.ID
}
}
retired, err := legacy.ForkHead(path, forkAt, agent.HeadKindFork, "deleted")
if err != nil {
t.Fatal(err)
}
legacy.Add(provider.Message{ID: "deleted-work", Role: provider.RoleUser, Content: "retired branch"})
if err := legacy.Save(path); err != nil {
t.Fatal(err)
}
if err := legacy.SwitchHead(path, child); err != nil {
t.Fatal(err)
}
if err := agent.RetireSessionHead(path, retired); err != nil {
t.Fatal(err)
}
app := NewApp()
t.Cleanup(app.closeSessionServices)
var oldTarget string
if previousMigration {
if err := app.migrateLegacySession(t.Context(), path, desktopMigrationSource{root: filepath.Dir(path), scope: "global"}, ""); err != nil {
t.Fatal(err)
}
oldTarget = onlyMigrationRecord(t).TargetSessionID
appendMigrationTestMessage(t, app.desktopSessionService(""), session.SessionRef{HostID: localDesktopHostID, SessionID: oldTarget}, "continued-before-upgrade")
// Upgrade must recognize the receipt even if the old app selected
// another head after that migration.
if err := legacy.SwitchHead(path, rootHead); err != nil {
t.Fatal(err)
}
}
before := migrationSourceSnapshot(t, legacyMigrationSourceFiles(path))
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
histories := v5MigrationHistories(t, app)
if len(histories) != 2 {
t.Fatalf("expected both branches: %v", histories)
}
if previousMigration && histories[oldTarget][len(histories[oldTarget])-1] != "continued-before-upgrade" {
t.Fatalf("old adopted target lost: %v", histories)
}
for id := range histories {
appendMigrationTestMessage(t, app.desktopSessionService(""), session.SessionRef{HostID: localDesktopHostID, SessionID: id}, "v5-continued")
}
app.closeSessionServices()
app = NewApp()
t.Cleanup(app.closeSessionServices)
tmp := t.TempDir()
t.Setenv("TMPDIR", filepath.Join(tmp, "missing"))
for range 2 {
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
}
if got := v5MigrationHistories(t, app); len(got) != 2 {
t.Fatalf("duplicates: %v", got)
}
assertMigrationSourceSnapshot(t, before)
t.Setenv("TMPDIR", tmp)
if err := legacy.SwitchHead(path, child); err != nil {
t.Fatal(err)
}
third, err := legacy.ForkHead(path, forkAt, agent.HeadKindFork, "third")
if err != nil || third == "" {
t.Fatal(err)
}
legacy.Add(provider.Message{ID: "third-work", Role: provider.RoleUser, Content: "third branch"})
if err := legacy.Save(path); err != nil {
t.Fatal(err)
}
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
if got := v5MigrationHistories(t, app); len(got) != 3 {
t.Fatalf("new head missing or original duplicated: %v", got)
}
ledger, err := readDesktopMigrationLedger()
if err != nil {
t.Fatal(err)
}
primary := ledger.Records[desktopLegacyMigrationKey(path)].LegacyPrimaryHead
if err := agent.RetireSessionHead(path, primary); err != nil {
t.Fatal(err)
}
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
if got := v5MigrationHistories(t, app); len(got) != 3 {
t.Fatalf("retiring primary changed other head identities: %v", got)
}
})
}
}
func TestDesktopV5UpgradeV3Formats(t *testing.T) {
for _, codec := range []string{session.PrototypeCodec, session.LegacyLinearCodec, session.FinalV31Codec} {
t.Run(codec, func(t *testing.T) {
isolateDesktopUserDirs(t)
root, id := filepath.Join(filepath.Dir(config.SessionStoreDir()), "sessions-v3"), "old-v3"
writeV3MigrationFixture(t, root, id, codec, []provider.Message{{ID: "user", Role: provider.RoleUser, Content: "v3 history"}})
before := migrationSourceSnapshot(t, canonicalMigrationSourceFiles(root, id))
app := NewApp()
t.Cleanup(app.closeSessionServices)
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
first := onlyMigrationRecord(t)
histories := v5MigrationHistories(t, app)
if !reflect.DeepEqual(histories[first.TargetSessionID], []string{"v3 history"}) {
t.Fatalf("history=%v", histories)
}
appendMigrationTestMessage(t, app.desktopSessionService(""), session.SessionRef{HostID: localDesktopHostID, SessionID: first.TargetSessionID}, "v5 continued")
app.closeSessionServices()
app = NewApp()
t.Cleanup(app.closeSessionServices)
t.Setenv("TMPDIR", filepath.Join(t.TempDir(), "missing"))
for range 2 {
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
}
if record := onlyMigrationRecord(t); record.TargetSessionID != first.TargetSessionID || record.Attempts != first.Attempts {
t.Fatalf("remigrated: %#v", record)
}
assertMigrationSourceSnapshot(t, before)
})
}
}
func TestDesktopV5UpgradePairedV3AndLegacy(t *testing.T) {
for _, relation := range []string{"events-newer", "transcript-newer", "divergent"} {
t.Run(relation, func(t *testing.T) {
isolateDesktopUserDirs(t)
path := filepath.Join(config.SessionDir(), "paired.jsonl")
legacy := agent.NewSession("system")
legacy.Add(provider.Message{ID: "user", Role: provider.RoleUser, Content: "common"})
if err := legacy.Save(path); err != nil {
t.Fatal(err)
}
messages := legacy.Snapshot()
switch relation {
case "events-newer":
messages = append(messages, provider.Message{ID: "newer", Role: provider.RoleAssistant, Content: "event reply"})
case "transcript-newer":
legacy.Add(provider.Message{ID: "newer", Role: provider.RoleAssistant, Content: "transcript reply"})
if err := legacy.Save(path); err != nil {
t.Fatal(err)
}
case "divergent":
messages[len(messages)-1].Content = "conflict"
}
writeV3MigrationFixture(t, config.SessionStoreDir(), agent.BranchID(path), session.LegacyLinearCodec, messages)
before := migrationSourceSnapshot(t, desktopLegacySourceFiles(path, config.SessionStoreDir()))
app := NewApp()
t.Cleanup(app.closeSessionServices)
err := app.migrateDesktopSessionsV5(t.Context())
if relation == "divergent" {
if err == nil {
t.Fatal("divergent history silently selected")
}
if got := v5MigrationHistories(t, app); len(got) != 0 {
t.Fatalf("published conflicting source: %v", got)
}
} else {
if err != nil {
t.Fatal(err)
}
histories := v5MigrationHistories(t, app)
if len(histories) != 1 {
t.Fatalf("paired history duplicated: %v", histories)
}
for _, history := range histories {
if !strings.HasSuffix(history[len(history)-1], "reply") {
t.Fatalf("new work lost: %v", history)
}
}
}
assertMigrationSourceSnapshot(t, before)
})
}
}
func TestDesktopV5UpgradeLegacyFailureRetryAndSibling(t *testing.T) {
isolateDesktopUserDirs(t)
root := config.SessionDir()
writeMigrationJSON(t, filepath.Join(root, "healthy.jsonl"), provider.Message{Role: provider.RoleUser, Content: "healthy"})
bad := filepath.Join(root, "broken.jsonl")
if err := os.WriteFile(bad, []byte("{broken\n"), 0o600); err != nil {
t.Fatal(err)
}
app := NewApp()
t.Cleanup(app.closeSessionServices)
if err := app.migrateDesktopSessionsV5(t.Context()); err == nil {
t.Fatal("broken source silently ignored")
}
if got := v5MigrationHistories(t, app); len(got) != 1 {
t.Fatalf("healthy sibling missing: %v", got)
}
writeMigrationJSON(t, bad, provider.Message{Role: provider.RoleUser, Content: "repaired"})
app.closeSessionServices()
app = NewApp()
t.Cleanup(app.closeSessionServices)
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
state, err := app.workspaceRegistry().Load(t.Context())
if err != nil {
t.Fatal(err)
}
ids := state.Workspaces[workspacestate.GlobalWorkspaceID].SessionIDs
sort.Strings(ids)
if len(ids) != 2 {
t.Fatalf("retry failed: %v", ids)
}
}
func TestDesktopV5UpgradePairedCanonicalKeepsPriorAdoption(t *testing.T) {
for _, previousMigration := range []bool{false, true} {
t.Run(map[bool]string{false: "fresh", true: "already-continued"}[previousMigration], func(t *testing.T) {
isolateDesktopUserDirs(t)
path := filepath.Join(config.SessionDir(), "paired-canonical.jsonl")
writeMigrationJSON(t, path, provider.Message{Role: provider.RoleUser, Content: "obsolete checkpoint"})
root, id := config.SessionStoreDir(), agent.BranchID(path)
coldV4MigrationFixture(t, root, id)
app := NewApp()
t.Cleanup(app.closeSessionServices)
if previousMigration {
if err := app.migrateCanonicalStore(t.Context(), desktopMigrationSource{root: root, scope: "global"}); err != nil {
t.Fatal(err)
}
appendMigrationTestMessage(t, app.desktopSessionService(""), session.SessionRef{HostID: localDesktopHostID, SessionID: id}, "v5 continued")
}
before := migrationSourceSnapshot(t, desktopLegacySourceFiles(path, root))
for range 2 {
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
}
history := v5MigrationHistories(t, app)
if len(history) != 1 || history[id][0] != "恢复完整对话" {
t.Fatalf("canonical authority lost or duplicated: %v", history)
}
if previousMigration && history[id][len(history[id])-1] != "v5 continued" {
t.Fatalf("continued work lost: %v", history)
}
assertMigrationSourceSnapshot(t, before)
})
}
}
func TestDesktopV5UpgradeLegacyResumesWorkspacePublication(t *testing.T) {
isolateDesktopUserDirs(t)
path := filepath.Join(config.SessionDir(), "interrupted.jsonl")
writeMigrationJSON(t, path, provider.Message{Role: provider.RoleUser, Content: "durable history"})
app := NewApp()
t.Cleanup(app.closeSessionServices)
err := app.migrateLegacySession(t.Context(), path, desktopMigrationSource{root: filepath.Dir(path), scope: "global"}, "missing-workspace")
if !errors.Is(err, workspacestate.ErrWorkspaceNotFound) {
t.Fatalf("expected attach failure: %v", err)
}
first := onlyMigrationRecord(t)
if first.Status != "failed" || first.TargetSessionID == "" {
t.Fatalf("missing retry receipt: %#v", first)
}
app.closeSessionServices()
app = NewApp()
t.Cleanup(app.closeSessionServices)
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
current := onlyMigrationRecord(t)
if current.Status != "completed" || current.TargetSessionID != first.TargetSessionID {
t.Fatalf("retry changed identity: %#v", current)
}
if got := v5MigrationHistories(t, app); len(got) != 1 {
t.Fatalf("interrupted import duplicated: %v", got)
}
}
func TestDesktopV5UpgradeProjectV3FailureRetry(t *testing.T) {
isolateDesktopUserDirs(t)
workspace := filepath.Join(t.TempDir(), "历史项目")
if err := saveProjectsFile(desktopProjectFile{Projects: []desktopProject{{Root: workspace}}}); err != nil {
t.Fatal(err)
}
root := filepath.Join(filepath.Dir(config.ProjectSessionStoreDir(workspace)), "sessions-v3")
writeV3MigrationFixture(t, root, "healthy", session.FinalV31Codec, []provider.Message{{ID: "user", Role: provider.RoleUser, Content: "healthy"}})
writeV3MigrationFixture(t, root, "broken", session.PrototypeCodec, []provider.Message{{ID: "user", Role: provider.RoleUser, Content: "repaired"}})
logPath := filepath.Join(root, "broken", "events.jsonl")
original, err := os.ReadFile(logPath)
if err != nil {
t.Fatal(err)
}
if err := os.WriteFile(logPath, []byte("{damaged}\n"), 0o600); err != nil {
t.Fatal(err)
}
app := NewApp()
t.Cleanup(app.closeSessionServices)
if err := app.migrateDesktopSessionsV5(t.Context()); err == nil {
t.Fatal("damaged preview accepted")
}
if got := v5MigrationHistories(t, app); len(got) != 1 {
t.Fatalf("healthy sibling missing: %v", got)
}
if err := os.WriteFile(logPath, original, 0o600); err != nil {
t.Fatal(err)
}
app.closeSessionServices()
app = NewApp()
t.Cleanup(app.closeSessionServices)
if err := app.migrateDesktopSessionsV5(t.Context()); err != nil {
t.Fatal(err)
}
state, err := app.workspaceRegistry().Load(t.Context())
if err != nil {
t.Fatal(err)
}
if got := state.Workspaces[desktopWorkspaceID("project", workspace)].SessionIDs; len(got) != 2 {
t.Fatalf("project v3 recovery=%v", got)
}
}
+6 -3
View File
@@ -12,11 +12,11 @@ Every entry carries one class: `keep-business` keeps its Go implementation and c
| frontend-native | 0 | 0 | 0 | 0 |
| frontend-event | 10 | 0 | 0 | 10 |
| css-marker | 10 | 0 | 0 | 10 |
| persistence | 13 | 0 | 0 | 13 |
| persistence | 16 | 0 | 0 | 16 |
| shell-file | 1 | 39 | 4 | 44 |
| artifact | 5 | 0 | 0 | 5 |
| ci-job | 26 | 0 | 0 | 26 |
| **all** | | | | **784** |
| **all** | | | | **787** |
## Desktop commands (Go `App` methods bound to the UI)
@@ -753,14 +753,17 @@ Every entry carries one class: `keep-business` keeps its Go implementation and c
| `desktop-workspaces.json` | | desktop/workspace.go:35 | keep-business (保留业务实现) | format unchanged; read by both shells |
| `desktop-zoom.json` | | desktop/zoom_factor.go:20 | keep-business (保留业务实现) | format unchanged; read by both shells |
| `downloaded.json` | | desktop/updater.go:541 | keep-business (保留业务实现) | format unchanged; read by both shells |
| `events.jsonl` | | desktop/session_v5_migration_checkpoint.go:106 | keep-business (保留业务实现) | format unchanged; read by both shells |
| `fork-operations.json` | | desktop/fork_operations.go:37 | keep-business (保留业务实现) | format unchanged; read by both shells |
| `header.json` | | desktop/session_v5_migration_checkpoint.go:106 | keep-business (保留业务实现) | format unchanged; read by both shells |
| `heartbeat-tasks.json` | | desktop/heartbeat.go:160 | keep-business (保留业务实现) | format unchanged; read by both shells |
| `heartbeat-tasks.runs.json` | | desktop/heartbeat_sidecar.go:34 | keep-business (保留业务实现) | format unchanged; read by both shells |
| `manifest.json` | | desktop/session_v5_migration_checkpoint.go:105 | keep-business (保留业务实现) | format unchanged; read by both shells |
| `mcp-registry-v0.1.json` | | desktop/app.go:7014 | keep-business (保留业务实现) | format unchanged; read by both shells |
| `operations-v1.json` | | desktop/browser_executor.go:141 | keep-business (保留业务实现) | format unchanged; read by both shells |
| `primary.lock` | | desktop/lifecycle_diagnostics.go:106 | keep-business (保留业务实现) | format unchanged; read by both shells |
| `remote-provider-trust.json` | | desktop/remote_legacy_cleanup.go:31 | keep-business (保留业务实现) | format unchanged; read by both shells |
| `session-migration-v5.json` | | desktop/session_v5_migration.go:39 | keep-business (保留业务实现) | format unchanged; read by both shells |
| `session-migration-v5.json` | | desktop/session_v5_migration.go:46 | keep-business (保留业务实现) | format unchanged; read by both shells |
## Shell-specific Go files
+22 -1
View File
@@ -5568,6 +5568,13 @@
"class": "keep-business",
"owner": "format unchanged; read by both shells"
},
{
"kind": "persistence",
"name": "events.jsonl",
"location": "desktop/session_v5_migration_checkpoint.go:106",
"class": "keep-business",
"owner": "format unchanged; read by both shells"
},
{
"kind": "persistence",
"name": "fork-operations.json",
@@ -5575,6 +5582,13 @@
"class": "keep-business",
"owner": "format unchanged; read by both shells"
},
{
"kind": "persistence",
"name": "header.json",
"location": "desktop/session_v5_migration_checkpoint.go:106",
"class": "keep-business",
"owner": "format unchanged; read by both shells"
},
{
"kind": "persistence",
"name": "heartbeat-tasks.json",
@@ -5589,6 +5603,13 @@
"class": "keep-business",
"owner": "format unchanged; read by both shells"
},
{
"kind": "persistence",
"name": "manifest.json",
"location": "desktop/session_v5_migration_checkpoint.go:105",
"class": "keep-business",
"owner": "format unchanged; read by both shells"
},
{
"kind": "persistence",
"name": "mcp-registry-v0.1.json",
@@ -5620,7 +5641,7 @@
{
"kind": "persistence",
"name": "session-migration-v5.json",
"location": "desktop/session_v5_migration.go:39",
"location": "desktop/session_v5_migration.go:46",
"class": "keep-business",
"owner": "format unchanged; read by both shells"
},
+157
View File
@@ -47,6 +47,96 @@ before their ledger record becomes complete. Same-ID/same-content imports are
idempotent; same-ID/different-content imports receive a stable `migr-<hash>` ID.
Failures remain retryable and do not stop other sources from migrating.
Canonical v4 migration enumerates durable session identities independently of
display metadata. A missing or stale catalog cache must never make a conversation
look empty and exclude it from migration; no open tab or legacy JSONL equivalent
is required. Source read failures are recorded in the migration ledger and exposed
by session architecture diagnostics. Startup retries them and finishes workspace
attachment for targets published before an interruption, without duplicating them.
Older session inputs in global, known-project and saved-tab directories are
discovered during the same startup pass, including historical `sessions-v3`
directories alongside `sessions-v4`. Schema-1
transcripts are imported even when their display sidecars are absent. Schema-2
DAG transcripts enumerate every live head and import each head independently;
the first adopted head retains its source key across selection changes and
retirement. Older migration receipts are reconciled against source content so
upgrading from an older migrator does not duplicate continued targets. Retired
heads remain excluded. Prototype
v3 and linear v3/v3.1 stores use their explicit read-only adapters and are staged
into the v5 codec after validation. Paired transcript/event sources are compared
before publication; divergent histories remain readable but are refused rather
than silently choosing one.
Automatic conflict-recovery artifacts are excluded from the ordinary v5 list.
Discovery recognizes `recovered`, `version_kind=recovery`, a recovery digest,
and the historical `-recovery-<16 hex>` filename convention. The exclusion
follows paired v3/v4 stores and conversion provenance, including archived
metadata when originals are absent. Original files remain untouched. An
ordinary user fork, a shared title, or an arbitrary filename containing
"recovery" does not qualify for exclusion.
Persisted `manifest.source` provenance also links already-converted generations,
including v2 -> v3, v2 -> v3 -> v4, and native v3 -> v4. All inputs of a linked
head are staged separately so an ancestor's source digest cannot accidentally
read a continued descendant that has the same identity. Equal histories and
strict prefixes share one v5 target containing the complete history. Independent
continuations become separate targets; unrelated sessions are never coalesced
just because their text or title matches. An omitted legacy head is resolved from
the conversion's immutable original snapshot, not the old app's current selection.
Stored-only chains are resolved even after the original JSONL is removed.
If a native v3 source directory is removed after multiple conversions, its
persisted source path still identifies the shared origin across archived copies.
Each source receives its own adoption receipt, including covered ancestors.
Already-completed receipts and v5 continuations remain authoritative; migration
does not delete existing v5 conversations or rewrite their content to force a
merge. Source changes with additional independent work are imported separately.
Each startup still enumerates source directories, but completed migrations use a
lightweight file revision (file presence, mode, size and nanosecond modification
time). Unchanged sources do not replay history, construct temporary imports,
compare target history, reattach membership or rewrite the ledger. Legacy checks
include the same sidecar set as the importer's freeze operation. This revision is
a change detector for normal writes, not an integrity checksum for edits that
deliberately preserve both file size and modification time.
The ledger records that a source was adopted. Continuing, archiving or deleting
the v5 target does not make that source eligible again. Missing revision markers
in older completed records are filled after a one-time comparison with the
recorded **source** content digest, never with the target's current history.
Changed source content is imported separately without overwriting continued v5
work. Failed attempts retain the previous completion receipt; a source changed
during import is not certified as complete. Pending and failed imports remain
retryable on the next startup.
| Ledger field | Older data | New reader/writer | Previous reader/writer |
| --- | --- | --- | --- |
| `sourceRevision` (optional) | Absent | Reconcile source digest once, then skip unchanged revisions | Ignores the field; may drop it on write, requiring reconciliation again |
| `previousCompletion` (optional) | Absent | Retains prior adoption proof across failed updates | Ignores the field; old releases retain their old migration behavior |
| `legacyHeads`, `legacyHeadsRevision` (optional) | Absent | Enumerates frozen heads once; skips replay while the source revision is unchanged | Ignores these fields; may drop them on write |
| `legacyPrimaryHead`, `legacySelectedHead` (optional) | Absent | Keeps per-head identity stable while routing the selected head to its paired store | Ignores these fields; old migration behavior remains |
| `legacyAdoption` (optional) | Absent | Preserves a pre-head-migration receipt when selection changes or migration is interrupted | Ignores this field; old migration behavior remains |
| `legacyConversions` (optional) | Absent | Caches resolved conversion ancestry/head IDs under the source revision | Ignores this field; discovery is repeated if it is dropped |
| Source format | Old data | New reader | Previous reader | Compatibility |
| --- | --- | --- | --- | --- |
| JSONL / schema-1 | Preserved byte-for-byte | Imported without display metadata; event history takes precedence | Still reads original source | Read-only upgrade |
| schema-2 DAG | Preserved byte-for-byte | Every live head becomes a resumable v5 session; retired heads stay excluded | Original selection and branches remain intact | Read-only upgrade |
| Prototype v3 / linear v3 / v3.1 | Preserved byte-for-byte | Explicit adapter validates and stages into v5 | Still reads original source | Read-only upgrade |
| v5 output | Separate directory | Native v5 reader/writer | Not visible to older versions | Existing isolation boundary |
The ledger stays at version 1. Writes preserve unknown root and record fields;
unreadable or future-version ledgers are refused rather than replaced. Session
formats and provider-visible message bytes are unchanged.
Optional `submission/accepted` events added by #10392 survive canonical export
and migration unchanged. Receipts retain their original session scope: a
same-identity migration exposes them in history, while a conflict mapped to an
independent identity does not adopt the old session's admission keys. This
matches fork isolation and keeps host metadata out of provider messages. The
canonical title reader introduced by #10389 reads migrated history directly.
Protocol 9 is a hard Desktop shell/host boundary. Older releases retain their
original data but do not see sessions created only in v5. Remote and Serve
session protocols are unchanged.
@@ -76,3 +166,70 @@ v5 迁移器只读保留 canonical v4 与 legacy JSONL。每个导入都先在
校验和原子发布,再写入 Workspace;同 ID 同内容幂等复用,同 ID 不同内容稳定
映射为 `migr-<hash>`。Protocol 9 是 Desktop shell/host 的硬边界;远端与 Serve
协议保持不变,旧版只能看到升级前保留的数据。
canonical v4 迁移按持久化会话身份枚举,不依赖标题、轮次、预览等显示缓存。
缓存缺失或过期不能成为跳过会话的依据,也不要求会话有打开的 tab 或对应的
legacy JSONL。源数据读取失败会记入迁移台账并通过会话架构诊断接口报告;
下次启动会重试,并为中断前已发布的目标补齐 Workspace 归属,不重复创建。
启动扫描同时覆盖全局、已登记项目和保存的 tab 所指向的旧目录,包括历史
`sessions-v3` 与现有 `sessions-v4`。v1 JSONL / schema-1 日志不再依赖显示
元数据;通过统一文件分类排除事件、运行记录等辅助 JSONL,避免误导入为对话。
v2 schema-2 DAG 从冻结副本枚举所有未删除分支,分别恢复为可续聊的 v5 会话;
在旧版本中切换或删除主分支不会改变其他分支的迁移身份。已删除分支不会复活。
v3 原型、线性 v3、v3.1 均使用显式只读转换器。旧快照与配对事件存储先比较
历史,选择包含完整较新历史的一方;存在真正分歧时保留源文件并报告失败。
自动生成的冲突恢复副本不迁入 v5 普通会话列表。通过 `recovered`、
`version_kind=recovery`、恢复摘要以及历史 `-recovery-<16位十六进制>`
文件名识别,排除规则沿配对的 v3/v4 存储和转换来源链继承;原文件缺失时也检查
保留的归档元数据。原始文件保留不动。普通用户分支、相同标题,以及仅包含
“recovery”文字的普通文件名不受影响。
迁移还会根据 `manifest.source` 追溯已经转换的来源关系,覆盖 v2→v3、
v2→v3→v4、原生 v3→v4 等多代目录同时残留的情况。每代源在独立临时目录中
校验,避免同 ID 的较新副本污染较早源的内容摘要。相同历史或只包含较早前缀
的源共用一份完整的 v5 会话;同源但各自续聊后产生不同内容的记录分别保留。
没有来源关系的对话,即使文本或标题相同,也不会被合并。
旧转换记录未填写 head 时,从它保留的原始快照确认当时选中的分支,不拿旧软件
当前选择代替。即使原始 JSONL 已移除,保留的 v3/v4 转换链仍会去重。
原生 v3 目录在多次转换后被移除时,也按保存的原始来源路径识别各归档副本的关系。
被覆盖的较早源也记录各自的迁移完成证明,重启、v5 续聊或删除目标不会使它再次导入。
已完成的旧迁移记录和 v5 中的新工作仍受保护;不会为强行合并而删除已有会话或
覆盖其内容。旧源后来出现额外的独立工作时,按新分支保留。
| 来源格式 | 旧数据 | 新版读取行为 | 旧版读取行为 | 兼容结论 |
| --- | --- | --- | --- | --- |
| JSONL / schema-1 | 原字节保留 | 无元数据也迁移,事件历史优先 | 仍读取原文件 | 只读升级 |
| schema-2 DAG | 原字节保留 | 恢复全部未删除分支 | 原分支与默认选择不变 | 只读升级 |
| v3 原型 / 线性 v3 / v3.1 | 原字节保留 | 校验后转换并导入 v5 | 仍读取原文件 | 只读升级 |
| v5 输出 | 独立目录 | 原生读写 | 旧版不可见 | 沿用隔离边界 |
每次启动仍枚举旧目录,但已完成项只检查文件是否存在、类型、大小与修改时间。
源文件未变化时,不重读历史、不生成临时导入、不比较目标内容、不重新登记归属,
也不重写台账。JSONL 检查包含导入器冻结的全部 sidecar。此标记用于识别正常文件
写入,不是用于检测刻意保持大小和修改时间不变的改写的完整性校验。
台账确认的是旧源已经被接收;在 v5 续聊、归档或删除目标,都不会触发重新导入。
旧台账缺少 `sourceRevision` 时,会与台账保存的源内容摘要核对一次后补齐标记,
不会拿旧源与已续聊的目标比较。旧源内容真正增加或变化时单独导入,保留 v5 中
的新工作。失败更新通过可选字段 `previousCompletion` 保留上次成功记录;迁移中
源数据发生变化则不标为完成,下次启动继续重试。
新增可选字段 `legacyHeads`、`legacyHeadsRevision` 缓存已校验的分支列表,
旧源不变时不重放 DAG。`legacyPrimaryHead` 固定首次迁移的主分支身份,
`legacySelectedHead` 标识应与配对存储比较的当前分支;`legacyAdoption`
保留旧迁移器的成功记录,按源内容识别已经在 v5 续聊的目标。旧台账没有这些
字段时自动补齐,原有成功记录不会因升级或重试而丢失。
`legacyConversions` 缓存已经核对的转换来源及分支身份,参与同一源版本校验,
源数据未变化时无需重新回放原始分支快照。
台账仍为 version 1,旧读取器忽略新增可选字段;旧写入器可能丢弃这些字段,
重新升级后需要再次核对。旧版本本身仍按旧迁移策略运行。新写入保留未知根字段
和记录字段,拒绝覆盖损坏或未来版本的台账;会话格式与模型可见消息内容不变。
#10392 新增的可选 `submission/accepted` 事件通过 canonical 导出与迁移原样保留。
回执继续按原会话 ID 隔离:同 ID 迁移后的历史保留提交标识;因冲突映射为独立 ID
的会话不会接收旧会话的提交幂等键,与 fork 隔离规则一致。回执不进入模型消息。
#10389 新增的 canonical 标题读取接口可直接读取迁移后的历史。
+15 -2
View File
@@ -191,17 +191,30 @@ func (st *sessionDAGState) reachable() map[string]struct{} {
// ListSessionHeads replays a schema-2 log and returns its heads. A schema-1
// session has no heads and returns nil, nil.
func ListSessionHeads(path string) ([]SessionHead, error) {
probe, err := probeSessionEventLog(path)
return listSessionHeads(context.Background(), path, defaultSessionReplayLimits, false)
}
// ListSessionHeadsForMigration enumerates a frozen source without the cumulative
// interactive replay budget. The migration owner must freeze the source first.
func ListSessionHeadsForMigration(ctx context.Context, path string) ([]SessionHead, error) {
return listSessionHeads(ctx, path, migrationSessionReplayLimits(), true)
}
func listSessionHeads(ctx context.Context, path string, limits sessionReplayLimits, strict bool) ([]SessionHead, error) {
probe, err := probeSessionEventLogWithLimits(path, limits)
if err != nil {
return nil, err
}
if !probe.dag {
return nil, nil
}
st, err := replaySessionDAG(context.Background(), store.SessionEventLog(path), defaultSessionReplayLimits)
st, err := replaySessionDAG(ctx, store.SessionEventLog(path), limits)
if err != nil {
return nil, fmt.Errorf("list session heads: %w", err)
}
if strict && st.damaged {
return nil, fmt.Errorf("list session heads: incomplete legacy DAG")
}
heads := st.headList()
slices.SortStableFunc(heads, func(a, b SessionHead) int {
return a.CreatedAt.Compare(b.CreatedAt)
+27 -4
View File
@@ -24,19 +24,42 @@ type ImportResult struct {
Kind string
}
// MigrationHistoryContains compares durable transcript meaning using the same
// normalization as paired imports. Callers must establish provenance first;
// matching text alone is not proof that two sessions share an identity.
func MigrationHistoryContains(history, prefix []provider.Message) bool {
history, prefix = comparableImportMessages(history), comparableImportMessages(prefix)
if len(prefix) > len(history) {
return false
}
for index := range prefix {
if !reflect.DeepEqual(history[index], prefix[index]) {
return false
}
}
return true
}
func importSourceForLegacy(ctx context.Context, sourcePath, targetRoot, headID string) (ImportResult, error) {
return importSourceForLegacyWithHeader(ctx, sourcePath, targetRoot, headID, CreateOptions{})
}
func importSourceForLegacyWithHeader(ctx context.Context, sourcePath, targetRoot, headID string, options CreateOptions) (ImportResult, error) {
return importSourceForLegacyFrom(ctx, sourcePath, targetRoot, targetRoot, headID, options)
}
func importSourceForLegacyFrom(ctx context.Context, sourcePath, sourceRoot, targetRoot, headID string, options CreateOptions) (ImportResult, error) {
return importSourceForLegacyAt(ctx, sourcePath, filepath.Join(sourceRoot, agent.BranchID(sourcePath)), targetRoot, headID, options)
}
func importSourceForLegacyAt(ctx context.Context, sourcePath, previewDir, targetRoot, headID string, options CreateOptions) (ImportResult, error) {
// The identity cutover deliberately reuses BranchID(sourcePath) for the
// canonical runtime. Once a final v4 store exists at that identity it is
// authoritative: treating it as a retired "paired preview" both rejects a
// valid codec and can remigrate an older checkpoint over newer v4 work.
previewDir := filepath.Join(targetRoot, agent.BranchID(sourcePath))
if final, finalErr := readManifest(filepath.Join(previewDir, "manifest.json")); finalErr == nil {
if final.SessionID != agent.BranchID(sourcePath) {
return ImportResult{}, fmt.Errorf("session: canonical store identity %q does not match legacy identity %q", final.SessionID, agent.BranchID(sourcePath))
if final.SessionID != filepath.Base(previewDir) {
return ImportResult{}, fmt.Errorf("session: canonical store identity %q does not match directory identity %q", final.SessionID, filepath.Base(previewDir))
}
source := Source{Path: sourcePath, Version: Codec}
if final.Source != nil {
@@ -135,7 +158,7 @@ func compareLegacySpool(path string, preview []provider.Message) (importMessageR
if len(comparable) == 0 {
continue
}
if firstDifference < 0 && (legacyCount >= len(preview) || !reflect.DeepEqual(comparable[0], preview[legacyCount])) {
if firstDifference < 0 && legacyCount < len(preview) && !reflect.DeepEqual(comparable[0], preview[legacyCount]) {
firstDifference = legacyCount
}
legacyCount++
@@ -0,0 +1,46 @@
package session
import (
"encoding/json"
"os"
"path/filepath"
"testing"
"reasonix/internal/provider"
)
func TestCompareLegacySpoolPreservesBothPrefixDirections(t *testing.T) {
message := func(id string) provider.Message {
return provider.Message{ID: id, Role: provider.RoleUser, Content: id}
}
for _, test := range []struct {
name string
legacy, preview []provider.Message
want importMessageRelation
}{
{"equal", []provider.Message{message("a")}, []provider.Message{message("a")}, importMessagesEqual},
{"newer-events", []provider.Message{message("a")}, []provider.Message{message("a"), message("b")}, importLegacyPrefix},
{"newer-transcript", []provider.Message{message("a"), message("b")}, []provider.Message{message("a")}, importPreviewPrefix},
{"divergent-longer-transcript", []provider.Message{message("a"), message("c"), message("d")}, []provider.Message{message("a"), message("b")}, importMessagesConflict},
} {
t.Run(test.name, func(t *testing.T) {
path := filepath.Join(t.TempDir(), "messages.jsons")
file, err := os.Create(path)
if err != nil {
t.Fatal(err)
}
for _, message := range test.legacy {
if err := json.NewEncoder(file).Encode(message); err != nil {
t.Fatal(err)
}
}
if err := file.Close(); err != nil {
t.Fatal(err)
}
got, count, _, err := compareLegacySpool(path, test.preview)
if err != nil || got != test.want || count != len(test.legacy) {
t.Fatalf("relation=%v count=%d err=%v", got, count, err)
}
})
}
}
+23
View File
@@ -102,6 +102,29 @@ type frozenLegacyHead struct {
freezeDir string
}
// LegacyMigrationHeads lists all heads from an immutable copy, without changing
// source selection, caches, or logs. Retired heads are returned for the caller
// to distinguish deliberate deletion from a missing branch.
func LegacyMigrationHeads(ctx context.Context, sourcePath string) ([]agent.SessionHead, error) {
sourcePath = agent.CanonicalSessionPath(sourcePath)
lease, err := agent.TryAcquireSessionLease(sourcePath)
if err != nil {
return nil, err
}
artifacts, _, dir, err := freezeLegacyArtifacts(ctx, sourcePath)
lease.Release()
if err != nil {
return nil, err
}
defer os.RemoveAll(dir)
for _, artifact := range artifacts {
if artifact.path == sourcePath {
return agent.ListSessionHeadsForMigration(ctx, artifact.frozenPath)
}
}
return nil, os.ErrNotExist
}
// freezeLegacyHead acquires the source lease, copies every durable artifact
// byte-for-byte, then parses only the frozen copy. The lease is released before
// parsing, which is safe precisely because the parse never reads the original.
+38
View File
@@ -4,11 +4,13 @@ import (
"context"
"errors"
"fmt"
"os"
"path/filepath"
"sync"
"sync/atomic"
"time"
"reasonix/internal/agent"
"reasonix/internal/event"
"reasonix/internal/transcript"
)
@@ -358,6 +360,42 @@ func (s *Service) ContinuePrototype(ctx context.Context, sourceDir string) (*Run
return runtime, result, err
}
// ContinueImportedFrom resolves legacy history against its original paired
// store while publishing only into this service's separate staging root.
func (s *Service) ContinueImportedFrom(ctx context.Context, sourcePath, sourceRoot, headID string) (*Runtime, ImportResult, error) {
return s.ContinueImportedSource(ctx, sourcePath, filepath.Join(sourceRoot, agent.BranchID(sourcePath)), headID)
}
// ContinueImportedSource accepts a provenance-linked directory whose identity
// may have changed when a legacy head was previously converted.
func (s *Service) ContinueImportedSource(ctx context.Context, sourcePath, sourceDir, headID string) (*Runtime, ImportResult, error) {
filesystem, ok := s.persistence.(*FilesystemPersistence)
if !ok {
return nil, ImportResult{}, errors.New("session: persistence does not support imported sessions")
}
result, err := importSourceForLegacyAt(ctx, sourcePath, sourceDir, filesystem.Root, headID, CreateOptions{})
if err != nil {
return nil, result, err
}
sourceRoot := filepath.Dir(sourceDir)
if result.Kind == "final" && filepath.Clean(sourceRoot) != filepath.Clean(filesystem.Root) {
tmp, err := os.MkdirTemp("", "reasonix-canonical-stage-")
if err != nil {
return nil, result, err
}
defer os.RemoveAll(tmp)
bundle := filepath.Join(tmp, "bundle")
if err := NewFilesystemPersistence(sourceRoot).exportCold(ctx, result.TargetID, bundle); err != nil {
return nil, result, err
}
if _, err := s.Import(ctx, bundle); err != nil {
return nil, result, err
}
}
runtime, err := s.openRuntime(ctx, SessionRef{HostID: s.hostID, SessionID: result.TargetID})
return runtime, result, err
}
// ContinueStoredPreview upgrades a pre-ownership linear store selected by its
// former session id. The old directory remains read-only; execution resumes on
// the deterministic final-codec identity returned here.