Merge pull request #13594 from SivanCola/codex/history-catalog-recovery-main-v2

Fix legacy recovery identity and catalog reconciliation lifetime / 修复历史恢复身份与目录校验生命周期

desktop/session_catalog.go

@@ -252,24 +252,26 @@ func (a *App) stopSessionCatalog(timeout time.Duration) bool {
 	a.catalogInitialReconcileDone = nil
 	a.catalogMetadataRequests = nil
 	a.catalogClosing = closing
-	a.catalogLifecycleMu.Unlock()
 	if cancel != nil {
 		cancel()
 	}
 	catalog := a.sessionCatalog.Swap(nil)
-	// Pair the nil publication with the request-side locked recheck. Once this
-	// barrier passes, the snapshot contains every reconcile that can use catalog
-	// and no new one can be added.
+	// Pair nil publication with the request-side locked recheck. Keep the
+	// lifecycle lock through detachment so replacement publication cannot
+	// enter this retired-job snapshot.
 	a.catalogReconcileMu.Lock()
 	reconcileDone := make([]<-chan struct{}, 0, len(a.catalogReconcileJobs))
 	for _, job := range a.catalogReconcileJobs {
+		job.cancel()
 		reconcileDone = append(reconcileDone, job.done)
 	}
+	clear(a.catalogReconcileJobs)
 	a.catalogReconcileMu.Unlock()
+	a.catalogLifecycleMu.Unlock()
 	go func() {
 		defer close(closing.done)
-		// A reconcile may wait on discovery that stopping has paused, so it
-		// only defers the close until the deadline, never past it.
+		// Cancellation releases queued discovery waits. Bound any synchronous
+		// admission still in flight by the original close deadline.
 		for _, reconciled := range reconcileDone {
 			if !waitChannelBefore(reconciled, deadline) {
 				break
@@ -417,9 +419,12 @@ func (a *App) emitProjectTreeChangedV2(revision uint64, roots []string, reason s
 }
 
 type desktopCatalogReconcileJob struct {
-	target sessioncatalog.DirectoryTarget
-	dirty  bool
-	done   chan struct{}
+	target  sessioncatalog.DirectoryTarget
+	dirty   bool
+	catalog *sessioncatalog.Catalog
+	ctx     context.Context
+	cancel  context.CancelFunc
+	done    chan struct{}
 }
 
 func (a *App) requestSessionCatalogReconcile(dir string) bool {
@@ -444,62 +449,69 @@ func (a *App) requestSessionCatalogReconcile(dir string) bool {
 	if a.catalogReconcileJobs == nil {
 		a.catalogReconcileJobs = map[string]*desktopCatalogReconcileJob{}
 	}
-	if job := a.catalogReconcileJobs[key]; job != nil {
+	if job := a.catalogReconcileJobs[key]; job != nil && job.catalog == catalog && job.ctx.Err() == nil {
 		job.target = target
 		job.dirty = true
 		a.catalogReconcileMu.Unlock()
 		return true
 	}
-	done := make(chan struct{})
-	a.catalogReconcileJobs[key] = &desktopCatalogReconcileJob{target: target, done: done}
+	ctx, cancel := context.WithCancel(a.bootContext())
+	job := &desktopCatalogReconcileJob{target: target, catalog: catalog, ctx: ctx, cancel: cancel, done: make(chan struct{})}
+	a.catalogReconcileJobs[key] = job
 	a.catalogReconcileMu.Unlock()
-	go a.runSessionCatalogReconcile(key, done)
+	go a.runSessionCatalogReconcile(key, job)
 	return true
 }
 
-func (a *App) runSessionCatalogReconcile(key string, done chan struct{}) {
-	defer close(done)
+func (a *App) runSessionCatalogReconcile(key string, job *desktopCatalogReconcileJob) {
+	defer func() {
+		job.cancel()
+		a.catalogReconcileMu.Lock()
+		if a.catalogReconcileJobs[key] == job {
+			delete(a.catalogReconcileJobs, key)
+		}
+		a.catalogReconcileMu.Unlock()
+		close(job.done)
+	}()
 	for {
 		a.catalogReconcileMu.Lock()
-		job := a.catalogReconcileJobs[key]
-		if job == nil {
+		if a.catalogReconcileJobs[key] != job || job.ctx.Err() != nil || a.shuttingDown.Load() {
 			a.catalogReconcileMu.Unlock()
 			return
 		}
 		target := job.target
 		job.dirty = false
 		a.catalogReconcileMu.Unlock()
-		catalog := a.sessionCatalog.Load()
-		if catalog == nil || a.shuttingDown.Load() {
-			a.catalogReconcileMu.Lock()
-			delete(a.catalogReconcileJobs, key)
-			a.catalogReconcileMu.Unlock()
-			return
-		}
-
 		if a.catalogReconcileHook != nil {
 			a.catalogReconcileHook(target)
 		}
-		// Discovery projects metadata without rewriting organization sidecars or
-		// proving recovery ancestry through transcript reads.
-		if settled, accepted := catalog.ScheduleReconcile(target); accepted {
+		// Discovery projects metadata only. Retired or invalidated owners cannot
+		// settle paused tokens, so bind scheduling and completion to this job.
+		if job.ctx.Err() == nil && a.sessionCatalog.Load() == job.catalog {
 			select {
-			case <-settled:
-			case <-a.bootContext().Done():
+			case <-job.catalog.Invalidated():
+				job.cancel()
+			default:
+				if settled, accepted := job.catalog.ScheduleReconcile(target); accepted {
+					select {
+					case <-settled:
+					case <-job.ctx.Done():
+					case <-job.catalog.Invalidated():
+						job.cancel()
+					}
+				}
 			}
 		}
 
 		a.catalogReconcileMu.Lock()
-		job = a.catalogReconcileJobs[key]
-		if job == nil {
-			a.catalogReconcileMu.Unlock()
-			return
-		}
-		if job.dirty && !a.shuttingDown.Load() {
+		if a.catalogReconcileJobs[key] == job && job.dirty && job.ctx.Err() == nil &&
+			a.sessionCatalog.Load() == job.catalog && !a.shuttingDown.Load() {
 			a.catalogReconcileMu.Unlock()
 			continue
 		}
-		delete(a.catalogReconcileJobs, key)
+		if a.catalogReconcileJobs[key] == job {
+			delete(a.catalogReconcileJobs, key)
+		}
 		a.catalogReconcileMu.Unlock()
 		if a.catalogReconcileDoneHook != nil {
 			a.catalogReconcileDoneHook(target)

desktop/session_catalog_reconcile_lifetime_test.go

@@ -0,0 +1,226 @@
+package main
+
+import (
+	"context"
+	"database/sql"
+	"path/filepath"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"reasonix/internal/sessioncatalog"
+)
+
+func TestStopSessionCatalogJoinsReconcileQueuedBeforeDiscoveryStarts(t *testing.T) {
+	isolateDesktopUserDirs(t)
+	ctx, cancel := context.WithCancel(t.Context())
+	app := NewApp()
+	app.ctx = ctx
+	requested := make(chan struct{}, 1)
+	catalog, err := sessioncatalog.Open(ctx, sessioncatalog.Options{
+		InMemory: true, MetadataOnly: true, StartPaused: true,
+		OnDiscovery: func(event sessioncatalog.DiscoveryEvent) {
+			if event.Phase == "requested" {
+				requested <- struct{}{}
+			}
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	app.sessionCatalog.Store(catalog)
+	var jobDone <-chan struct{}
+	t.Cleanup(func() {
+		cancel()
+		if !app.stopSessionCatalog(sessionCatalogTestDeadline) {
+			t.Error("catalog fixture did not close cleanly")
+		}
+		if jobDone != nil && !waitChannelBefore(jobDone, time.Now().Add(sessionCatalogTestDeadline)) {
+			t.Error("explicit reconcile outlived its fixture")
+		}
+	})
+
+	dir := t.TempDir()
+	if !app.requestSessionCatalogReconcile(dir) {
+		t.Fatal("explicit reconcile was not admitted")
+	}
+	app.catalogReconcileMu.Lock()
+	jobDone = app.catalogReconcileJobs[projectRootKey(dir)].done
+	app.catalogReconcileMu.Unlock()
+	// The worker is inside ScheduleReconcile, but paused discovery cannot settle
+	// its token. Stop must cancel and join the desktop job independently.
+	<-requested
+	if !app.stopSessionCatalog(time.Second) {
+		t.Fatal("catalog stop did not join its paused discovery waiter")
+	}
+	select {
+	case <-jobDone:
+	default:
+		t.Fatal("catalog stop returned before its reconcile goroutine exited")
+	}
+}
+
+func TestRetiredSessionCatalogReconcileCannotJoinReplacement(t *testing.T) {
+	isolateDesktopUserDirs(t)
+	ctx, cancel := context.WithCancel(t.Context())
+	app := NewApp()
+	app.ctx = ctx
+	var oldRequests atomic.Int32
+	openPaused := func(onDiscovery func(sessioncatalog.DiscoveryEvent)) *sessioncatalog.Catalog {
+		catalog, err := sessioncatalog.Open(ctx, sessioncatalog.Options{
+			InMemory: true, MetadataOnly: true, StartPaused: true, OnDiscovery: onDiscovery,
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+		return catalog
+	}
+	oldCatalog := openPaused(func(event sessioncatalog.DiscoveryEvent) {
+		if event.Phase == "requested" {
+			oldRequests.Add(1)
+		}
+	})
+	app.sessionCatalog.Store(oldCatalog)
+	entered := make(chan struct{}, 2)
+	oldRelease, newRelease := make(chan struct{}), make(chan struct{})
+	var oldOnce, newOnce sync.Once
+	var calls atomic.Int32
+	app.catalogReconcileHook = func(sessioncatalog.DirectoryTarget) {
+		call := calls.Add(1)
+		entered <- struct{}{}
+		if call == 1 {
+			<-oldRelease
+		} else {
+			<-newRelease
+		}
+	}
+	var jobs []<-chan struct{}
+	t.Cleanup(func() {
+		oldOnce.Do(func() { close(oldRelease) })
+		newOnce.Do(func() { close(newRelease) })
+		cancel()
+		if !app.stopSessionCatalog(sessionCatalogTestDeadline) {
+			t.Error("replacement catalog fixture did not close cleanly")
+		}
+		for _, done := range jobs {
+			if !waitChannelBefore(done, time.Now().Add(sessionCatalogTestDeadline)) {
+				t.Error("retired reconcile outlived its fixture")
+			}
+		}
+	})
+	dir := t.TempDir()
+	if !app.requestSessionCatalogReconcile(dir) {
+		t.Fatal("old reconcile was not admitted")
+	}
+	key := projectRootKey(dir)
+	app.catalogReconcileMu.Lock()
+	oldJob := app.catalogReconcileJobs[key]
+	app.catalogReconcileMu.Unlock()
+	jobs = append(jobs, oldJob.done)
+	<-entered
+	if app.stopSessionCatalog(25 * time.Millisecond) {
+		t.Fatal("stop joined the old job before its hook released")
+	}
+
+	// A timed-out rebuild may publish the next catalog while the old hook is
+	// still unwinding. The same directory must receive a distinct job owner.
+	app.sessionCatalog.Store(openPaused(nil))
+	if !app.requestSessionCatalogReconcile(dir) {
+		t.Fatal("replacement reconcile was not admitted")
+	}
+	app.catalogReconcileMu.Lock()
+	newJob := app.catalogReconcileJobs[key]
+	app.catalogReconcileMu.Unlock()
+	jobs = append(jobs, newJob.done)
+	if oldJob == newJob {
+		t.Fatal("replacement catalog inherited a retired reconcile job")
+	}
+	<-entered
+	oldOnce.Do(func() { close(oldRelease) })
+	if !waitChannelBefore(oldJob.done, time.Now().Add(time.Second)) {
+		t.Fatal("retired reconcile did not exit after its hook released")
+	}
+	app.catalogReconcileMu.Lock()
+	currentJob := app.catalogReconcileJobs[key]
+	app.catalogReconcileMu.Unlock()
+	if currentJob != newJob {
+		t.Fatal("retired reconcile removed the replacement job")
+	}
+	if oldRequests.Load() != 0 {
+		t.Fatal("retired reconcile scheduled discovery after cancellation")
+	}
+	newOnce.Do(func() { close(newRelease) })
+	if !app.stopSessionCatalog(time.Second) {
+		t.Fatal("replacement reconcile did not stop cleanly")
+	}
+}
+
+func TestInvalidatedSessionCatalogReleasesQueuedReconcileWithoutAppShutdown(t *testing.T) {
+	isolateDesktopUserDirs(t)
+	ctx, cancel := context.WithCancel(t.Context())
+	app := NewApp()
+	app.ctx = ctx
+	requested := make(chan struct{}, 1)
+	path := filepath.Join(t.TempDir(), "catalog.sqlite")
+	catalog, err := sessioncatalog.Open(ctx, sessioncatalog.Options{
+		Path: path, MetadataOnly: true, StartPaused: true,
+		OnDiscovery: func(event sessioncatalog.DiscoveryEvent) {
+			if event.Phase == "requested" {
+				requested <- struct{}{}
+			}
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	app.sessionCatalog.Store(catalog)
+	var jobDone <-chan struct{}
+	t.Cleanup(func() {
+		cancel()
+		if !app.stopSessionCatalog(sessionCatalogTestDeadline) {
+			t.Error("invalidated catalog fixture did not close cleanly")
+		}
+		if jobDone != nil && !waitChannelBefore(jobDone, time.Now().Add(sessionCatalogTestDeadline)) {
+			t.Error("invalidated reconcile outlived its fixture")
+		}
+	})
+	dir := t.TempDir()
+	if !app.requestSessionCatalogReconcile(dir) {
+		t.Fatal("explicit reconcile was not admitted")
+	}
+	app.catalogReconcileMu.Lock()
+	jobDone = app.catalogReconcileJobs[projectRootKey(dir)].done
+	app.catalogReconcileMu.Unlock()
+	<-requested
+
+	// Corrupt this disposable generation's root page. A fresh public read lease
+	// must report corruption and revoke it while discovery is still paused.
+	db, err := sql.Open("sqlite", path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, err = db.Exec(`PRAGMA writable_schema=ON; UPDATE sqlite_master SET rootpage=2147483647 WHERE type='table' AND name='catalog_state'; PRAGMA schema_version=999`)
+	closeErr := db.Close()
+	if err != nil || closeErr != nil {
+		t.Fatal(err, closeErr)
+	}
+	lease, err := catalog.OpenReadLease(ctx)
+	if lease != nil {
+		lease.Close()
+	}
+	if err == nil {
+		t.Fatal("corrupt root page remained readable")
+	}
+	select {
+	case <-catalog.Invalidated():
+	default:
+		t.Fatalf("public read did not invalidate the corrupt generation: %v", err)
+	}
+	if !waitChannelBefore(jobDone, time.Now().Add(time.Second)) {
+		t.Fatal("invalidated discovery kept the desktop job waiting for app shutdown")
+	}
+	if ctx.Err() != nil {
+		t.Fatal("reconcile completion depended on app shutdown")
+	}
+}

desktop/tabs_topic_test.go

@@ -734,28 +734,6 @@ func TestAmbiguousLegacyRecoverySessionsMigrateIntoTopics(t *testing.T) {
 	}
 }
 
-func TestUnmodifiedRecoveryCopyDoesNotMigrateIntoTopics(t *testing.T) {
-	isolateDesktopUserDirs(t)
-
-	dir := config.SessionDir()
-	if err := os.MkdirAll(dir, 0o755); err != nil {
-		t.Fatalf("mkdir sessions: %v", err)
-	}
-	parent, recovery, branchMsgs := forkDesktopRecoveryBranch(t, dir, "normal")
-	coverDesktopRecoveryParent(t, parent, branchMsgs)
-
-	app := NewApp()
-	nodes := waitForCatalogTopic(t, app, "global", "", legacySessionTopicID(parent))
-	if len(nodes) != 1 || len(nodes[0].Children) != 2 {
-		t.Fatalf("project tree = %#v, want both sources before content proof", nodes)
-	}
-	if meta, ok, err := agent.LoadBranchMeta(recovery); err != nil || !ok {
-		t.Fatalf("load recovery meta: ok=%v err=%v", ok, err)
-	} else if strings.TrimSpace(meta.TopicID) != "" {
-		t.Fatalf("parent-covered recovery copy was migrated into topic %q", meta.TopicID)
-	}
-}
-
 func TestCoveredRecoveryCopyBecomesVisibleAfterMigratedParentDeletion(t *testing.T) {
 	isolateDesktopUserDirs(t)
 

desktop/topic_activation_canonical_test.go

@@ -15,23 +15,30 @@ import (
 
 func canonicalTopicHistoryReady(t *testing.T, app *App, tabID string) session.HistoryWindowPage {
 	t.Helper()
-	deadline := time.Now().Add(5 * time.Second)
-	for {
-		page, err := app.SessionHistoryWindowForTab(tabID, session.HistoryWindowRequest{Anchor: "newest", Limit: 32})
+	req := session.HistoryWindowRequest{Anchor: "newest", Limit: 32}
+	page, err := app.SessionHistoryWindowForTab(tabID, req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if page.Status == "preparing" {
+		query, ref, err := app.canonicalSessionQuery(tabID)
 		if err != nil {
 			t.Fatal(err)
 		}
-		if page.Status != "preparing" {
-			if page.Status != "ready" || len(page.Messages) == 0 {
-				t.Fatalf("history is not readable: status=%s messages=%d", page.Status, len(page.Messages))
-			}
-			return page
+		// TitleMessages joins this query's locator preparation, so completion
+		// supplies the readiness boundary without polling filesystem speed.
+		if _, err := query.TitleMessages(t.Context(), ref, 1); err != nil {
+			t.Fatal(err)
 		}
-		if time.Now().After(deadline) {
-			t.Fatal("independent history preparation did not finish")
+		page, err = app.SessionHistoryWindowForTab(tabID, req)
+		if err != nil {
+			t.Fatal(err)
 		}
-		time.Sleep(time.Millisecond)
 	}
+	if page.Status != "ready" || len(page.Messages) == 0 {
+		t.Fatalf("history is not readable: status=%s messages=%d", page.Status, len(page.Messages))
+	}
+	return page
 }
 
 func TestCanonicalSidebarActivationReplacesRemoteSurface(t *testing.T) {

desktop/topic_history_preparation_owner_test.go

@@ -0,0 +1,80 @@
+package main
+
+import (
+	"context"
+	"sync"
+	"testing"
+
+	"reasonix/internal/session"
+)
+
+func TestCanonicalInactiveHistoryPreparationWaitsForItsOwnerWithoutRuntime(t *testing.T) {
+	app, source, target, root, _ := canonicalWorkspaceOpenFixture(t)
+	app.readyHook = func() {}
+	installNoopRuntimeEvents(app, source.sink)
+	gate := newTabBuildGate(app)
+	t.Cleanup(func() { app.shutdown(context.Background()) })
+	t.Cleanup(gate.releaseAll)
+	release, err := app.historyMaintenance.Foreground(t.Context())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(release)
+	if err := app.workspaceRegistry().EnsureSessionTopic(t.Context(), target.Ref().SessionID, "owner-gated-topic", "History owner"); err != nil {
+		t.Fatal(err)
+	}
+	meta, err := app.openProjectTabInactive(root, "owner-gated-topic")
+	if err != nil {
+		t.Fatal(err)
+	}
+	gate.waitEntered(t, meta.ID)
+	request := session.HistoryWindowRequest{Anchor: "newest", Limit: 32}
+	page, err := app.SessionHistoryWindowForTab(meta.ID, request)
+	if err != nil || page.Status != "preparing" {
+		t.Fatalf("history before admission: %+v, %v", page, err)
+	}
+	query, ref, err := app.canonicalSessionQuery(meta.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	entered := make(chan struct{})
+	wait := &topicHistoryOwnerWaitContext{Context: t.Context(), entered: entered}
+	done := make(chan struct{})
+	var prepareErr error
+	go func() {
+		_, prepareErr = query.TitleMessages(wait, ref, 1)
+		close(done)
+	}()
+	t.Cleanup(func() { release(); <-done })
+	<-entered
+	select {
+	case <-done:
+		t.Fatalf("history barrier completed while admission was held: %v", prepareErr)
+	default:
+	}
+	release()
+	<-done
+	if prepareErr != nil {
+		t.Fatal(prepareErr)
+	}
+	page, err = app.SessionHistoryWindowForTab(meta.ID, request)
+	if err != nil || page.Status != "ready" || len(page.Messages) == 0 || page.Messages[len(page.Messages)-1].MessageID != "session-B-user" {
+		t.Fatalf("history after preparation: %+v, %v", page, err)
+	}
+	app.mu.RLock()
+	defer app.mu.RUnlock()
+	if app.activeTabID != source.ID || app.tabs[meta.ID].Ctrl != nil || app.tabs[meta.ID].Ready {
+		t.Fatal("history preparation started the gated runtime or changed the visible session")
+	}
+}
+
+type topicHistoryOwnerWaitContext struct {
+	context.Context
+	entered chan struct{}
+	once    sync.Once
+}
+
+func (c *topicHistoryOwnerWaitContext) Done() <-chan struct{} {
+	c.once.Do(func() { close(c.entered) })
+	return c.Context.Done()
+}

desktop/topic_recovery_projection_test.go

@@ -0,0 +1,58 @@
+package main
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"reasonix/internal/agent"
+	"reasonix/internal/config"
+)
+
+func TestUnmodifiedRecoveryCopyDoesNotMigrateIntoTopics(t *testing.T) {
+	isolateDesktopUserDirs(t)
+	dir := config.SessionDir()
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	parent, recovery, branchMsgs := forkDesktopRecoveryBranch(t, dir, "normal")
+	coverDesktopRecoveryParent(t, parent, branchMsgs)
+	metaPath := agent.BranchMetaPath(recovery)
+	before, err := os.ReadFile(metaPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	app := NewApp()
+	// Metadata discovery may proceed, but content proof must wait until this
+	// test has observed both sources. Wall-clock timing cannot fence that phase.
+	release, err := app.historyMaintenance.Foreground(t.Context())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer release()
+	app.startSessionCatalog()
+	catalog := waitForSessionCatalogForTest(t, app, nil)
+	catalog.PrioritizeWorkspace("global", "")
+	nodes := waitForCatalogTopic(t, app, "global", "", legacySessionTopicID(parent))
+	if len(nodes) != 1 || len(nodes[0].Children) != 2 {
+		t.Fatalf("project tree = %#v, want both sources before content proof", nodes)
+	}
+	release()
+	if err := catalog.ProveRecoveryLineage(t.Context()); err != nil {
+		t.Fatal(err)
+	}
+	nodes = mustListProjectTree(t, app)
+	if len(nodes) != 1 || len(nodes[0].Children) != 1 || filepath.Clean(nodes[0].Children[0].SessionPath) != filepath.Clean(parent) {
+		t.Fatalf("project tree = %#v, want the covering parent after content proof", nodes)
+	}
+	if meta, ok, err := agent.LoadBranchMeta(recovery); err != nil || !ok {
+		t.Fatalf("load recovery meta: ok=%v err=%v", ok, err)
+	} else if strings.TrimSpace(meta.TopicID) != "" {
+		t.Fatalf("parent-covered recovery copy was migrated into topic %q", meta.TopicID)
+	}
+	after, err := os.ReadFile(metaPath)
+	if err != nil || string(after) != string(before) {
+		t.Fatalf("content proof changed the recovery sidecar: %v", err)
+	}
+}

internal/sessioncatalog/legacy_lineage_identity_test.go

@@ -0,0 +1,152 @@
+package sessioncatalog
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"reasonix/internal/agent"
+)
+
+func TestLegacyRecoveryIdentitySurvivesContentProofAndParentDeletion(t *testing.T) {
+	for _, metadataOnly := range []bool{false, true} {
+		t.Run(map[bool]string{false: "full", true: "metadata"}[metadataOnly], func(t *testing.T) {
+			ctx := t.Context()
+			dir := t.TempDir()
+			parent, recovery, _ := forkCatalogRecoveryBranch(t, dir, "legacy-parent")
+			coverCatalogRecoveryParent(t, parent, recovery)
+			before := map[string]string{}
+			for _, path := range []string{parent, recovery, agent.BranchMetaPath(parent), agent.BranchMetaPath(recovery)} {
+				data, err := os.ReadFile(path)
+				if err != nil {
+					t.Fatal(err)
+				}
+				before[path] = string(data)
+			}
+			opts := Options{Path: filepath.Join(t.TempDir(), "catalog.sqlite"), MetadataOnly: metadataOnly, DisableRepair: true, StartPaused: true}
+			catalog, err := Open(ctx, opts)
+			if err != nil {
+				t.Fatal(err)
+			}
+			t.Cleanup(func() {
+				if err := catalog.Close(context.Background()); err != nil {
+					t.Error(err)
+				}
+			})
+			reopen := func() {
+				t.Helper()
+				if err := catalog.Close(ctx); err != nil {
+					t.Fatal(err)
+				}
+				catalog, err = Open(ctx, opts)
+				if err != nil {
+					t.Fatal(err)
+				}
+			}
+			target := DirectoryTarget{Path: dir, Scope: "global"}
+			check := func(path, topic string, visible bool) {
+				t.Helper()
+				record, ok, err := catalog.GetSession(ctx, path)
+				if err != nil || !ok || record.TopicID != topic || record.OrdinaryVisible != visible {
+					t.Fatalf("record = %+v, ok=%v err=%v; want topic=%q visible=%v", record, ok, err, topic, visible)
+				}
+			}
+			reconcile := func() {
+				t.Helper()
+				if err := catalog.ReconcileDirectory(ctx, target); err != nil {
+					t.Fatal(err)
+				}
+				if err := catalog.ProveRecoveryLineage(ctx); err != nil {
+					t.Fatal(err)
+				}
+			}
+			reconcile()
+			check(parent, agent.LegacySessionTopicID(parent), true)
+			check(recovery, agent.LegacySessionTopicID(parent), false)
+			// An older executable may have certified this unchanged directory
+			// with the incompatible fallback. Reopen must repair that cache too.
+			oldTopic := "recovery:" + agent.BranchID(parent)
+			for _, query := range []string{
+				"UPDATE catalog_sessions SET topic_id=?,logical_topic_id=?",
+				"UPDATE catalog_topics SET topic_id=? WHERE topic_id<>?",
+			} {
+				if _, err := catalog.db.ExecContext(ctx, query, oldTopic, oldTopic); err != nil {
+					t.Fatal(err)
+				}
+			}
+			reopen()
+			reconcile()
+			check(parent, agent.LegacySessionTopicID(parent), true)
+			check(recovery, agent.LegacySessionTopicID(parent), false)
+			if err := os.Remove(parent); err != nil {
+				t.Fatal(err)
+			}
+			if err := os.Remove(agent.SessionEventLogPath(parent)); err != nil {
+				t.Fatal(err)
+			}
+			reconcile()
+			check(recovery, agent.LegacySessionTopicID(recovery), true)
+			reopen()
+			reconcile()
+			check(recovery, agent.LegacySessionTopicID(recovery), true)
+			// Reopening a rebuilt projection must converge without migrating the
+			// authoritative transcript or either legacy branch sidecar.
+			for _, path := range []string{recovery, agent.BranchMetaPath(parent), agent.BranchMetaPath(recovery)} {
+				data, err := os.ReadFile(filepath.Clean(path))
+				if err != nil || string(data) != before[path] {
+					t.Fatalf("catalog changed authority %s: %v", filepath.Base(path), err)
+				}
+			}
+		})
+	}
+}
+
+func TestLegacyRootlessRecoveryGroupUsesOneRepresentativeIdentity(t *testing.T) {
+	leaf := SessionRecord{Path: filepath.Join(t.TempDir(), "leaf.jsonl"), Recovered: true, ParentID: "gone", RecoveryGroupID: "gone", RecoveryCanonical: true}
+	peer := SessionRecord{Path: filepath.Join(filepath.Dir(leaf.Path), "peer.jsonl"), Recovered: true, ParentID: "gone", RecoveryGroupID: "gone"}
+	for _, input := range [][]SessionRecord{{leaf, peer}, {peer, leaf}} {
+		visible := 0
+		for _, got := range projectLogicalSessions(input) {
+			if got.TopicID != agent.LegacySessionTopicID(leaf.Path) || got.LogicalTopicID != got.TopicID {
+				t.Fatalf("rootless group identity = %+v", got)
+			}
+			if got.OrdinaryVisible {
+				visible++
+				if got.Path != leaf.Path {
+					t.Fatalf("visible recovery = %q, want canonical leaf", got.Path)
+				}
+			}
+		}
+		if visible != 1 {
+			t.Fatalf("visible recovery rows = %d, want 1", visible)
+		}
+	}
+}
+
+func TestExplicitRecoveryPrefixedTopicIdentityIsPreserved(t *testing.T) {
+	rootPath := filepath.Join(t.TempDir(), "root.jsonl")
+	group := agent.BranchID(rootPath)
+	oldTopic := "recovery:" + group
+	root := SessionRecord{Path: rootPath}
+	leaf := SessionRecord{Path: filepath.Join(filepath.Dir(rootPath), "leaf.jsonl"), Recovered: true, ParentID: group, RecoveryGroupID: group, RecoveryCanonical: true}
+	for _, placement := range []string{"root", "member", "rootless"} {
+		t.Run(placement, func(t *testing.T) {
+			parent, recovery := root, leaf
+			if placement == "root" {
+				parent.TopicID, parent.TopicTitle = oldTopic, "Persisted title"
+			} else {
+				recovery.TopicID, recovery.TopicTitle = oldTopic, "Persisted title"
+			}
+			records := []SessionRecord{parent, recovery}
+			if placement == "rootless" {
+				records = records[1:]
+			}
+			for _, got := range projectLogicalSessions(records) {
+				if got.TopicID != oldTopic || got.LogicalTopicID != oldTopic || got.TopicTitle != "Persisted title" {
+					t.Fatalf("explicit recovery-prefixed authority changed: %+v", got)
+				}
+			}
+		})
+	}
+}

internal/sessioncatalog/lineage.go

@@ -274,64 +274,7 @@ func projectLogicalSessions(records []SessionRecord) []SessionRecord {
 		}
 	}
 
-	type groupAnchor struct {
-		topicID    string
-		topicTitle string
-		createdAt  int64
-		rootIndex  int
-	}
-	anchors := map[string]*groupAnchor{}
-	for i, rec := range records {
-		if !rec.Recovered {
-			id := agent.BranchID(rec.Path)
-			anchors[id] = &groupAnchor{
-				topicID: rec.TopicID, topicTitle: rec.TopicTitle,
-				createdAt: rec.CreatedAt, rootIndex: i,
-			}
-		}
-	}
-	// Fill missing logical topics from recovered members (root topic empty or
-	// root missing). Prefer earliest non-empty member topic_id.
-	for _, rec := range records {
-		if !rec.Recovered {
-			continue
-		}
-		groupID := strings.TrimSpace(rec.RecoveryGroupID)
-		if groupID == "" {
-			continue
-		}
-		anchor := anchors[groupID]
-		if anchor == nil {
-			anchor = &groupAnchor{rootIndex: -1, createdAt: rec.CreatedAt}
-			anchors[groupID] = anchor
-		}
-		topic := strings.TrimSpace(rec.TopicID)
-		if topic == "" {
-			continue
-		}
-		if strings.TrimSpace(anchor.topicID) == "" ||
-			(rec.CreatedAt > 0 && (anchor.createdAt == 0 || rec.CreatedAt < anchor.createdAt) &&
-				(anchor.rootIndex < 0 || strings.TrimSpace(records[anchor.rootIndex].TopicID) == "")) {
-			// Keep a non-empty root topic when present; only override empty roots.
-			if anchor.rootIndex >= 0 && strings.TrimSpace(records[anchor.rootIndex].TopicID) != "" {
-				continue
-			}
-			anchor.topicID = topic
-			anchor.topicTitle = rec.TopicTitle
-			if rec.CreatedAt > 0 {
-				anchor.createdAt = rec.CreatedAt
-			}
-		}
-	}
-	for groupID, anchor := range anchors {
-		if strings.TrimSpace(anchor.topicID) == "" {
-			// Stable internal topic so rootless recovery storms still collapse.
-			anchor.topicID = "recovery:" + groupID
-			if anchor.topicTitle == "" {
-				anchor.topicTitle = groupID
-			}
-		}
-	}
+	anchors := logicalSessionAnchors(records)
 	for i := range records {
 		groupID := ""
 		if records[i].Recovered {

internal/sessioncatalog/lineage_identity.go

@@ -0,0 +1,81 @@
+package sessioncatalog
+
+import (
+	"path/filepath"
+	"strings"
+
+	"reasonix/internal/agent"
+)
+
+type groupAnchor struct {
+	topicID       string
+	topicTitle    string
+	createdAt     int64
+	rootIndex     int
+	fallbackIndex int
+}
+
+func logicalSessionAnchors(records []SessionRecord) map[string]*groupAnchor {
+	anchors := map[string]*groupAnchor{}
+	for i, rec := range records {
+		if !rec.Recovered {
+			id := agent.BranchID(rec.Path)
+			anchors[id] = &groupAnchor{
+				topicID: rec.TopicID, topicTitle: rec.TopicTitle,
+				createdAt: rec.CreatedAt, rootIndex: i, fallbackIndex: i,
+			}
+		}
+	}
+	// Fill missing logical topics from recovered members (root topic empty or
+	// root missing). Prefer earliest non-empty member topic_id.
+	for i, rec := range records {
+		if !rec.Recovered {
+			continue
+		}
+		groupID := strings.TrimSpace(rec.RecoveryGroupID)
+		if groupID == "" {
+			continue
+		}
+		anchor := anchors[groupID]
+		if anchor == nil {
+			anchor = &groupAnchor{rootIndex: -1, createdAt: rec.CreatedAt, fallbackIndex: i}
+			anchors[groupID] = anchor
+		}
+		if anchor.rootIndex < 0 {
+			preferred := pickPreferredRecovery([]SessionRecord{records[anchor.fallbackIndex], rec})
+			if preferred.Path == rec.Path {
+				anchor.fallbackIndex = i
+			}
+		}
+		topic := strings.TrimSpace(rec.TopicID)
+		if topic == "" {
+			continue
+		}
+		if strings.TrimSpace(anchor.topicID) == "" ||
+			(rec.CreatedAt > 0 && (anchor.createdAt == 0 || rec.CreatedAt < anchor.createdAt) &&
+				(anchor.rootIndex < 0 || strings.TrimSpace(records[anchor.rootIndex].TopicID) == "")) {
+			// Keep a non-empty root topic when present; only override empty roots.
+			if anchor.rootIndex >= 0 && strings.TrimSpace(records[anchor.rootIndex].TopicID) != "" {
+				continue
+			}
+			anchor.topicID = topic
+			anchor.topicTitle = rec.TopicTitle
+			if rec.CreatedAt > 0 {
+				anchor.createdAt = rec.CreatedAt
+			}
+		}
+	}
+	for _, anchor := range anchors {
+		if strings.TrimSpace(anchor.topicID) == "" {
+			// Content proof must retain metadata discovery's legacy identity.
+			// With no surviving root, use the ordinary recovery representative;
+			// every sibling still folds onto that one topic without sidecar writes.
+			path := records[anchor.fallbackIndex].Path
+			anchor.topicID = agent.LegacySessionTopicID(path)
+			if anchor.topicTitle == "" {
+				anchor.topicTitle = strings.TrimSuffix(filepath.Base(path), ".jsonl")
+			}
+		}
+	}
+	return anchors
+}