Unify migrated session identity and lifecycle / 统一新旧会话身份与生命周期 (#10676)

* fix(desktop): keep purged history out of topic placeholders

Problem: Permanently deleted history could reappear as a metadata-only topic and fail subsequent archive requests with session not found.

Root cause: Sidebar adoption filtering depended on live membership and presentation, both removed by purge, while legacy metadata and source receipts remained.

Fix: Derive an immutable topic adoption index from durable committed evidence, preserve minimal topic identity in purge receipts, and share the index across materialized, lazy, and pinned projections.

Verification: Workspace state tests with race detection, archive/restore/purge/restart integration coverage, shared-topic and independent-source preservation, pinned/manual-order regressions, previous operation codec round-trip, incremental lint, and diff checks passed. Native packaged Windows/macOS UI was not revalidated.

* fix(desktop): unify migrated session identity and lifecycle

Problem: A migrated conversation could appear as both a canonical session and a historical source, highlight twice, and reappear after lifecycle operations.

Root cause: Physical path normalization changed source hashes while adoption receipts and pending imports retained older keys. Lookup and lifecycle callers did not share compatibility aliases.

Fix: Derive immutable source aliases from durable mapping receipts and use them across discovery, navigation, tab restore, import preparation, sidebar projections, and lifecycle guards. Preserve independent head and reviewed-version identities. Persist explicit source-cleanup intent with purge tombstones, clean only unchanged exclusive canonical originals under ownership locks, and retain shared, changed, offline, and multi-head sources.

Verification: Focused identity, archive/restore/purge, shared-source protection, writer contention, and process-restart tests passed. Workspace state race tests, filesystem purge tests, incremental lint, and diff checks passed. The complete Desktop CI partition run is in progress; the initial unpartitioned run hit its six-minute package timeout. Packaged Windows and macOS UI has not been revalidated.

* fix(desktop): separate migration identity and purge planning owners

Problem: The PR failed repository size and complexity checks and lacked the required documentation-impact declaration.

Root cause: Source alias lookup and purge-plan encoding grew existing lifecycle coordinators beyond their file budgets, and historical head resolution remained embedded in the generic source resolver. The historical migration guide still implied that all original files survive explicit permanent deletion.

Fix: Move source mapping lookup and versioned purge-plan construction into focused owners, extract historical head resolution, and document unified display/lifecycle and guarded source cleanup in both language guides. Preserve all behavior and test coverage without raising budgets or disabling checks.

Verification: Repository repolint, the documentation-impact validator, isolated-cache golangci-lint 2.12.2, migration identity/import/head/purge regressions, workspace-state package tests, and diff checks passed.
This commit is contained in:
SivanCola authored and GitHub committed 2026-09-23 15:34:46 +08:00
1 parent 07fd2fb0df
commit fba4febbb6
35 files changed
+1461 -144

No files matched your search

+2 -2
View File
@@ -2686,7 +2686,7 @@ func (a *App) ListTrashedSessions() []SessionMeta {
if !explicitlyDeletedLegacyEntry(path) {
continue
}
if _, adopted := state.SourceMappings[desktopSourceKey(path, "")]; adopted {
if _, adopted, err := state.ResolveSource(desktopSourceKey(path, "")); adopted || err != nil {
continue
}
infos, err := agent.ListSessions(filepath.Dir(path))
@@ -3293,7 +3293,7 @@ func (a *App) purgeTrashedSession(path string, requireRedundantRecovery bool) er
if err != nil {
return err
}
if _, adopted := state.SourceMappings[desktopSourceKey(path, "")]; adopted {
if _, adopted, err := state.ResolveSource(desktopSourceKey(path, "")); adopted || err != nil {
return errors.New("the historical source is preserved for its restored session")
}
if !explicitlyDeletedLegacyEntry(path) {
+9 -7
View File
@@ -277,7 +277,7 @@ func (a *App) historicalCanonicalTopicsFromProjection(scope, root string, state
continue
}
node := entry.node
if _, adopted := historicalMappingForSource(state, node.Source.SourceKey); adopted {
if _, adopted, err := historicalMappingForSource(state, node.Source.SourceKey); adopted || err != nil {
continue
}
node.PreparationStatus = "available"
@@ -307,14 +307,20 @@ func applyHistoricalPresentations(nodes []ProjectNode, saved historicalImportQue
// A shell-only read must not create workspaces or migrate organization state.
// Sources without canonical members still need their persisted pin overlays.
func (a *App) historicalPinnedShellsFromProjection(req ProjectTopicPageRequest, state workspacestate.State, index *workspacestate.WorkspaceIndex, legacy desktopProject) ([]ProjectNode, error) {
workspaceID, _, _ := index.Resolve(req.WorkspaceRoot)
if req.Scope != "project" {
workspaceID = workspacestate.GlobalWorkspaceID
}
adopted := map[string]bool{}
for _, mapping := range state.SourceMappings {
adopted["source\x00local\x00"+mapping.SourceKey] = true
for _, key := range state.SourceKeys(mapping.SourceKey) {
adopted["source\x00local\x00"+key] = true
}
if sourceMappingHasPathAlias(mapping) {
adopted[sessionRuntimeKey(mapping.Path)] = true
}
}
page, err := a.unadoptedLegacyTopics(req, adopted, nil)
page, err := a.unadoptedLegacyTopics(req, adopted, state.AdoptedTopicIDs(workspaceID))
if err != nil {
return nil, err
}
@@ -322,10 +328,6 @@ func (a *App) historicalPinnedShellsFromProjection(req ProjectTopicPageRequest,
if saved, err := readHistoricalSidecar(); err == nil {
applyHistoricalPresentations(nodes, saved)
}
workspaceID, _, _ := index.Resolve(req.WorkspaceRoot)
if req.Scope != "project" {
workspaceID = workspacestate.GlobalWorkspaceID
}
workspace := state.Workspaces[workspaceID]
org := projectedShellOrganization(workspace, state, nodes, legacy)
req.pinnedOnly = true
+11 -24
View File
@@ -227,7 +227,11 @@ func historicalImportView(state workspacestate.State, id string, source historic
view = HistoricalSessionView{ID: id, Title: filepath.Base(source.path), Format: source.format, Status: "available"}
}
view.Source = &SessionSourceRef{HostID: localDesktopHostID, SourceKey: desktopSourceKey(source.path, source.head), Path: source.path, HeadID: source.head}
mapping, ok := historicalMappingForSource(state, id)
mapping, ok, err := historicalMappingForSource(state, id)
if err != nil {
view.Status, view.ErrorCode, view.Session = "failed", "target_changed", nil
return view
}
if !ok {
return view
}
@@ -241,27 +245,6 @@ func historicalImportView(state workspacestate.State, id string, source historic
return view
}
func historicalSourceKeyMatches(mappingKey, sourceID string) bool {
return mappingKey == sourceID || strings.HasPrefix(mappingKey, sourceID+":review:")
}
func historicalMappingForSource(state workspacestate.State, sourceID string) (workspacestate.SourceMapping, bool) {
if mapping, ok := state.SourceMappings[sourceID]; ok {
return mapping, true
}
keys := make([]string, 0, len(state.SourceMappings))
for key := range state.SourceMappings {
if historicalSourceKeyMatches(key, sourceID) {
keys = append(keys, key)
}
}
sort.Strings(keys)
if len(keys) == 0 {
return workspacestate.SourceMapping{}, false
}
return state.SourceMappings[keys[0]], true
}
func (c *historicalImportCoordinator) initialize(ctx context.Context) {
if c.sources != nil {
if !c.stopped && !c.running && len(c.calls) == 0 && c.ctx.Err() != nil {
@@ -506,9 +489,10 @@ func (a *App) importHistoricalSource(ctx context.Context, id string, source hist
if result, handled, err := a.resumeConflictingHistoricalVersion(ctx, state, source, workspace); handled {
return result, err
}
migration := desktopMigrationSource{scope: source.scope, workspaceRoot: source.root, headID: source.head, versionFingerprint: source.version}
migration := desktopMigrationSource{scope: source.scope, workspaceRoot: source.root, headID: source.head, versionFingerprint: source.version, registeredSourceKey: id}
if resume := pendingHistoricalOperation(state, id); resume != nil {
migration.operationID = resume.ID
migration.registeredSourceKey = resume.Mapping.SourceKey
}
err = a.convertHistoricalSource(ctx, source, migration, workspace)
if err != nil {
@@ -518,7 +502,10 @@ func (a *App) importHistoricalSource(ctx context.Context, id string, source hist
if err != nil {
return SessionRestoreResult{}, err
}
mapping, ok := state.SourceMappings[id]
mapping, ok, err := state.ResolveSource(id)
if err != nil {
return SessionRestoreResult{}, err
}
if !ok {
return SessionRestoreResult{}, errors.New("historical import has not committed")
}
+26 -3
View File
@@ -2,8 +2,10 @@ package main
import (
"encoding/json"
"os"
"path/filepath"
"reflect"
"strings"
"sync"
"sync/atomic"
"testing"
@@ -212,6 +214,16 @@ func TestHistoricalQueueRestartsPaused(t *testing.T) {
}
func TestHistoricalSourceUpdateImportsOneStableBranch(t *testing.T) {
for _, oldIdentity := range []bool{false, true} {
name := "current"
if oldIdentity {
name = "old-identity"
}
t.Run(name, func(t *testing.T) { testHistoricalSourceUpdateImportsOneStableBranch(t, oldIdentity) })
}
}
func testHistoricalSourceUpdateImportsOneStableBranch(t *testing.T, oldIdentity bool) {
isolateDesktopUserDirs(t)
old := coldV4MigrationFixture(t, config.SessionStoreDir(), "updated-source")
app := newHistoricalLifecycleApp(t)
@@ -223,6 +235,9 @@ func TestHistoricalSourceUpdateImportsOneStableBranch(t *testing.T) {
if err != nil {
t.Fatal(err)
}
if oldIdentity {
rewriteHistoricalSourceIdentityForTest(t, app, list.Items[0].ID)
}
binding, err := old.Open(t.Context(), session.SessionRef{HostID: "migration-source", SessionID: "updated-source"})
if err != nil {
t.Fatal(err)
@@ -507,14 +522,19 @@ func TestHistoricalImportDoesNotReviveArchivedOrDeletedTarget(t *testing.T) {
if state.SessionStates[result.Session.SessionID].Lifecycle != lifecycle || len(state.Workspaces[workspacestate.GlobalWorkspaceID].SessionIDs) != retainedMemberships || len(state.SourceMappings) != 1 {
t.Fatalf("rescan/import revived target: %+v", state)
}
assertStartupHistorySourceUnchanged(t, root, "retained-source", original)
if lifecycle == workspacestate.Archived {
assertStartupHistorySourceUnchanged(t, root, "retained-source", original)
} else if _, err := os.Stat(filepath.Join(root, "retained-source")); !os.IsNotExist(err) {
t.Fatalf("exclusive original survived purge: %v", err)
}
})
}
}
func TestHistoricalImportResumesPriorDurablePhase(t *testing.T) {
for _, phase := range []string{"prepared", "content_ready", "content_ready_old_metadata"} {
t.Run(phase, func(t *testing.T) {
for _, scenario := range []string{"prepared", "content_ready", "content_ready_old_metadata", "prepared_old_identity", "content_ready_old_identity"} {
t.Run(scenario, func(t *testing.T) {
phase := strings.TrimSuffix(scenario, "_old_identity")
isolateDesktopUserDirs(t)
root := config.SessionStoreDir()
const sessionID = "interrupted-source"
@@ -560,6 +580,9 @@ func TestHistoricalImportResumesPriorDurablePhase(t *testing.T) {
}
}
}
if strings.HasSuffix(scenario, "_old_identity") {
rewriteHistoricalSourceIdentityForTest(t, app, desktopSourceKey(path, ""))
}
before, err := app.workspaceRegistry().Load(t.Context())
if err != nil {
t.Fatal(err)
+13 -3
View File
@@ -33,7 +33,14 @@ func validateDesktopOperationSources(state workspacestate.State, op workspacesta
func pendingHistoricalOperation(state workspacestate.State, id string) *workspacestate.Operation {
var resume *workspacestate.Operation
for _, candidate := range state.PendingOperations {
if candidate.Phase == "committed" || candidate.Mapping == nil || !historicalSourceKeyMatches(candidate.Mapping.SourceKey, id) {
if candidate.Phase == "committed" || candidate.Mapping == nil {
continue
}
matches := false
for _, key := range state.SourceKeys(candidate.Mapping.SourceKey) {
matches = matches || historicalSourceKeyMatches(key, id)
}
if !matches {
continue
}
if candidate.Kind != "import" && candidate.Kind != "restore" {
@@ -51,9 +58,12 @@ func pendingHistoricalOperation(state workspacestate.State, id string) *workspac
// Committed mappings and published content are independent of retained sources.
// Check them before touching a source lock, path, fingerprint, or version.
func (a *App) resumeReadyHistoricalImport(ctx context.Context, state workspacestate.State, id string, source historicalSource) (SessionRestoreResult, bool, error) {
mapping, mapped := historicalMappingForSource(state, id)
mapping, mapped, err := historicalMappingForSource(state, id)
if source.version != "" {
mapping, mapped = state.SourceMappings[id]
mapping, mapped, err = state.ResolveSource(id)
}
if err != nil {
return SessionRestoreResult{}, true, err
}
if mapped {
if state.SessionStates[mapping.SessionID].Lifecycle != workspacestate.Active {
+185
View File
@@ -0,0 +1,185 @@
package main
import (
"crypto/sha256"
"encoding/json"
"fmt"
"os"
"path/filepath"
"reflect"
"slices"
"testing"
"reasonix/desktop/internal/workspacestate"
"reasonix/internal/config"
)
func rewriteHistoricalSourceIdentityForTest(t *testing.T, app *App, key string) string {
t.Helper()
state, err := app.workspaceRegistry().Load(t.Context())
if err != nil {
t.Fatal(err)
}
mapping, committed := state.SourceMappings[key]
if !committed {
for _, op := range state.PendingOperations {
if op.Mapping != nil && op.Mapping.SourceKey == key {
mapping = *op.Mapping
break
}
}
}
if mapping.Path == "" {
t.Fatal("fixture has no source receipt")
}
oldKey := fmt.Sprintf("%x", sha256.Sum256([]byte(mapping.Path+"\x00")))
if oldKey == key { // Case-sensitive CI still exercises a changed persisted identity.
oldKey = fmt.Sprintf("%x", sha256.Sum256([]byte("previous-spelling:"+mapping.Path+"\x00")))
}
if committed {
delete(state.SourceMappings, key)
mapping.SourceKey = oldKey
state.SourceMappings[oldKey] = mapping
}
for id, op := range state.PendingOperations {
if op.Mapping != nil && op.Mapping.SourceKey == key {
copy := *op.Mapping
copy.SourceKey = oldKey
op.Mapping = &copy
state.PendingOperations[id] = op
}
}
body, err := json.Marshal(state)
if err != nil {
t.Fatal(err)
}
registryPath := app.workspaceRegistry().Path()
if err := os.WriteFile(registryPath, body, 0600); err != nil {
t.Fatal(err)
}
return oldKey
}
func TestHistoricalSourceIdentityUpgradeKeepsOneConversation(t *testing.T) {
isolateDesktopUserDirs(t)
const id = "MixedCase-history"
coldV4MigrationFixture(t, config.SessionStoreDir(), id)
app := newHistoricalLifecycleApp(t)
key := historicalLifecycleID(t, app, id)
imported, err := app.ImportHistoricalSession(key)
if err != nil {
t.Fatal(err)
}
app.stopHistoricalImports()
app.closeSessionServices()
oldKey := rewriteHistoricalSourceIdentityForTest(t, app, key)
state, err := app.workspaceRegistry().Load(t.Context())
if err != nil {
t.Fatal(err)
}
mapping := state.SourceMappings[oldKey]
for range 2 {
app = newHistoricalLifecycleApp(t)
listed, err := app.ListHistoricalSessions()
if err != nil || len(listed.Items) != 1 || listed.Items[0].Session == nil || *listed.Items[0].Session != imported.Session {
t.Fatalf("discovery lost adoption: %+v %v", listed, err)
}
page, err := app.ListProjectTopics(ProjectTopicPageRequest{Scope: "global", Limit: 50})
if err != nil || len(page.Items) != 1 || page.Items[0].Session == nil || *page.Items[0].Session != imported.Session {
t.Fatalf("duplicate sidebar rows: %+v %v", page.Items, err)
}
if !slices.Contains(page.Items[0].IdentityAliases, "source\x00local\x00"+key) {
t.Fatal("current source row cannot merge with canonical identity")
}
for _, supplied := range []string{oldKey, key} {
target, err := app.resolveSessionTarget(SessionSelector{Source: &SessionSourceRef{HostID: localDesktopHostID, SourceKey: supplied, Path: mapping.Path}})
if err != nil || target.SessionRef != imported.Session {
t.Fatalf("source %q opened another target: %+v %v", supplied, target, err)
}
preparedKey, _, err := app.historicalSourceForSelector(SessionSelector{Source: &SessionSourceRef{SourceKey: supplied, Path: mapping.Path}})
if err != nil || preparedKey != oldKey {
t.Fatalf("preparation changed durable identity: %s %v", preparedKey, err)
}
}
if _, err := app.resolveSessionTarget(SessionSelector{Source: &SessionSourceRef{SourceKey: oldKey, Path: filepath.Join(mapping.Path, "different")}}); err == nil {
t.Fatal("old source key authorized a different path")
}
ref, adopted, err := app.legacyCanonicalRef(t.Context(), mapping.Path)
if err != nil || !adopted || ref != imported.Session {
t.Fatalf("saved path failed canonical binding: %+v %v", ref, err)
}
view, err := app.workspaceRegistry().Load(t.Context())
if err != nil {
t.Fatal(err)
}
saved := desktopTabEntry{Scope: "global", SessionPath: mapping.Path}
if app.savedTabHistoricalSource(saved, savedTabReconcileEvidence{registry: view}) != nil {
t.Fatal("restored canonical tab became a historical shell")
}
again, err := app.ImportHistoricalSession(key)
if err != nil || again.Session != imported.Session {
t.Fatalf("reimport duplicated identity: %+v %v", again, err)
}
after, err := app.workspaceRegistry().Load(t.Context())
if err != nil || !reflect.DeepEqual(after.SourceMappings, state.SourceMappings) || !reflect.DeepEqual(after.PendingOperations, state.PendingOperations) || !reflect.DeepEqual(after.SessionStates, state.SessionStates) {
t.Fatal("compatibility lookup changed adoption evidence or session lifecycles")
}
if len(after.Workspaces[workspacestate.GlobalWorkspaceID].SessionIDs) != 1 {
t.Fatal("reimport added a second conversation")
}
app.desktopSessions.readSnapshots.close()
app.stopHistoricalImports()
app.closeSessionServices()
}
// One lifecycle owns both source spellings, including after a cold start.
app = newHistoricalLifecycleApp(t)
assertRows := func(want int) {
t.Helper()
if _, err := app.ListHistoricalSessions(); err != nil {
t.Fatal(err)
}
page, err := app.ListProjectTopics(ProjectTopicPageRequest{Scope: "global", Limit: 50})
if err != nil || len(page.Items) != want {
t.Fatalf("lifecycle sidebar: %+v %v; want %d", page.Items, err, want)
}
}
assertRetiredAfterRestart := func(lifecycle string) {
t.Helper()
app.desktopSessions.readSnapshots.close()
app.stopHistoricalImports()
app.closeSessionServices()
app = newHistoricalLifecycleApp(t)
assertRows(0)
if _, err := app.ImportHistoricalSession(key); err == nil {
t.Fatal("retired source was imported again")
}
current, err := app.workspaceRegistry().Load(t.Context())
if err != nil || current.SessionStates[imported.Session.SessionID].Lifecycle != lifecycle || len(current.SourceMappings) != 1 {
t.Fatalf("retired identity changed: %v", err)
}
}
archive, err := app.ArchiveSessionTarget(SessionSelector{Source: &SessionSourceRef{Path: mapping.Path, SourceKey: key}})
if err != nil || !archive.Committed {
t.Fatalf("archive through old source: %+v %v", archive, err)
}
assertRetiredAfterRestart(workspacestate.Archived)
trash, err := app.ListTrashEntries("", "", 50)
if err != nil || len(trash.Items) != 1 || trash.Items[0].Ref == nil || *trash.Items[0].Ref != imported.Session {
t.Fatalf("duplicate archive entries: %+v %v", trash, err)
}
if err := app.RestoreCanonicalSession(imported.Session); err != nil {
t.Fatal(err)
}
assertRows(1)
if err := app.ArchiveCanonicalSession(imported.Session); err != nil {
t.Fatal(err)
}
if err := app.PurgeCanonicalSession(imported.Session); err != nil {
t.Fatal(err)
}
assertRetiredAfterRestart(workspacestate.Deleted)
app.desktopSessions.readSnapshots.close()
if _, err := os.Stat(mapping.Path); !os.IsNotExist(err) {
t.Fatalf("exclusive original survived purge: %v", err)
}
}
+32
View File
@@ -0,0 +1,32 @@
package main
import (
"sort"
"strings"
"reasonix/desktop/internal/workspacestate"
)
func historicalSourceKeyMatches(mappingKey, sourceID string) bool {
return mappingKey == sourceID || strings.HasPrefix(mappingKey, sourceID+":review:")
}
func historicalMappingForSource(state workspacestate.State, sourceID string) (workspacestate.SourceMapping, bool, error) {
if mapping, ok, err := state.ResolveSource(sourceID); ok || err != nil {
return mapping, ok, err
}
keys := make([]string, 0, len(state.SourceMappings))
for key := range state.SourceMappings {
for _, alias := range state.SourceKeys(key) {
if historicalSourceKeyMatches(alias, sourceID) {
keys = append(keys, key)
break
}
}
}
sort.Strings(keys)
if len(keys) == 0 {
return workspacestate.SourceMapping{}, false, nil
}
return state.SourceMappings[keys[0]], true, nil
}
+1 -1
View File
@@ -23,7 +23,7 @@ func (a *App) savedTabHistoricalSource(entry desktopTabEntry, evidence savedTabR
}
path := agent.CanonicalSessionPath(entry.SessionPath)
key := desktopSourceKey(path, entry.SessionHeadID)
if _, adopted := historicalMappingForSource(evidence.registry, key); adopted {
if _, adopted, err := historicalMappingForSource(evidence.registry, key); adopted || err != nil {
return nil
}
// The saved identity is already known. Stat just this source, without
+24 -6
View File
@@ -1,6 +1,7 @@
package workspacestate
import (
"bytes"
"context"
"crypto/sha256"
"encoding/hex"
@@ -442,28 +443,28 @@ func cancelPreparedPurge(state *State, id string) {
func samePurgeIdentity(left, right Operation) bool {
return left.ID == right.ID && left.Kind == "purge" && right.Kind == "purge" && left.Lifecycle == right.Lifecycle &&
left.ExpectedGeneration == right.ExpectedGeneration && slices.Equal(left.SessionIDs, right.SessionIDs)
left.ExpectedGeneration == right.ExpectedGeneration && slices.Equal(left.SessionIDs, right.SessionIDs) && bytes.Equal(left.Request, right.Request)
}
// BeginPurge atomically validates the archived generation, publishes the
// deletion tombstone and records the resumable purge operation.
func (s *Store) BeginPurge(ctx context.Context, id string, expected uint64) error {
return s.beginOrResumePurge(ctx, id, expected, nil)
return s.beginOrResumePurge(ctx, id, expected, nil, false)
}
// ResumePurge continues only the observed operation. It cannot recreate a
// deletion intent after a restore superseded that operation.
func (s *Store) ResumePurge(ctx context.Context, id string, observed Operation) error {
return s.beginOrResumePurge(ctx, id, observed.ExpectedGeneration, &observed)
return s.beginOrResumePurge(ctx, id, observed.ExpectedGeneration, &observed, false)
}
// ResumePurgeForRequest keeps both the request snapshot and the observed
// transaction identity. Neither may be refreshed while waiting for locks.
func (s *Store) ResumePurgeForRequest(ctx context.Context, id string, expected uint64, observed Operation) error {
return s.beginOrResumePurge(ctx, id, expected, &observed)
return s.beginOrResumePurge(ctx, id, expected, &observed, false)
}
func (s *Store) beginOrResumePurge(ctx context.Context, id string, expected uint64, observed *Operation) error {
func (s *Store) beginOrResumePurge(ctx context.Context, id string, expected uint64, observed *Operation, cleanupSources bool) error {
stale := false
err := s.mutate(ctx, func(state *State) error {
key := "purge-" + id
@@ -500,7 +501,15 @@ func (s *Store) beginOrResumePurge(ctx context.Context, id string, expected uint
if !known || status.Lifecycle != Archived || status.Generation > expected {
return ErrMutationConflict
}
state.PendingOperations[key] = Operation{ID: key, Kind: "purge", Phase: "tombstoned", Lifecycle: Deleted, SessionIDs: []string{id}, ExpectedGeneration: status.Generation}
op := Operation{ID: key, Kind: "purge", Phase: "tombstoned", Lifecycle: Deleted, SessionIDs: []string{id}, ExpectedGeneration: status.Generation}
if cleanupSources {
var err error
op.Request, err = purgeSourceCleanupRequest(*state, id)
if err != nil {
return err
}
}
state.PendingOperations[key] = op
setLifecycle(state, id, Deleted)
return nil
default:
@@ -549,6 +558,15 @@ func (s *Store) CompletePurge(ctx context.Context, id string) error {
default:
return ErrMutationConflict
}
// Keep only the consumed topic identity in the existing purge receipt.
// Canonical sessions and runtime-adopted sources may have no import
// journal from which a future display reader could recover that identity.
if topicID := state.Presentation[id].TopicID; topicID != "" {
op.WorkspaceID, _ = sessionOwner(*state, id)
if op.Presentation == nil {
op.Presentation = &Presentation{TopicID: topicID}
}
}
for key, workspace := range state.Workspaces {
workspace.SessionIDs = remove(workspace.SessionIDs, id)
state.Workspaces[key] = workspace
@@ -0,0 +1,33 @@
package workspacestate
import (
"context"
"encoding/json"
"slices"
"strings"
)
// PurgeSourceCleanup records the exact adopted sources eligible for an explicit
// deletion request. Older purge journals without this request retain originals.
// Multi-file legacy DAGs are retained: their independent heads and artifacts
// cannot be deleted as one canonical session directory.
type PurgeSourceCleanup struct {
Version int `json:"sourceCleanupVersion"`
Sources []SourceMapping `json:"sources"`
}
func (s *Store) BeginPurgeWithSources(ctx context.Context, id string, expected uint64) error {
return s.beginOrResumePurge(ctx, id, expected, nil, true)
}
func purgeSourceCleanupRequest(state State, id string) (json.RawMessage, error) {
plan := PurgeSourceCleanup{Version: 1}
for _, mapping := range state.SourceMappings {
if mapping.SessionID == id && mapping.Format == "canonical" && mapping.HeadID == "" {
mapping.RetainedArtifacts = nil
plan.Sources = append(plan.Sources, mapping)
}
}
slices.SortFunc(plan.Sources, func(a, b SourceMapping) int { return strings.Compare(a.SourceKey, b.SourceKey) })
return json.Marshal(plan)
}
@@ -13,12 +13,13 @@ import (
// owns its maps privately; retaining it never observes a later publication.
// Verification is O(file bytes); lookups on a retained snapshot are O(1).
type ReadSnapshot struct {
state State
owners map[string]string
conflicts map[string]bool
activeTopics map[string]int
headSources map[string]bool
versions *ReadVersions
state State
owners map[string]string
conflicts map[string]bool
activeTopics map[string]int
adoptedTopics map[string]map[string]bool
headSources map[string]bool
versions *ReadVersions
}
// SessionMetadata deliberately excludes workspace members and organization:
@@ -40,7 +41,9 @@ type snapshotVerification struct {
}
func (s *Store) publishSnapshotLocked(body []byte, state State) {
state.sourceIdentities = newSourceIdentityIndex(state)
r := &ReadSnapshot{state: state, owners: make(map[string]string), conflicts: make(map[string]bool), activeTopics: make(map[string]int), headSources: make(map[string]bool), versions: NewReadVersions(state)}
r.adoptedTopics = adoptedTopicIndex(state)
for key, workspace := range state.Workspaces {
for _, id := range workspace.SessionIDs {
if _, exists := r.owners[id]; exists {
@@ -144,8 +147,12 @@ func (r *ReadSnapshot) Session(id string) SessionMetadata {
}
func (r *ReadSnapshot) Source(key string) (SourceMapping, bool) {
mapping, ok := r.state.SourceMappings[key]
return cloneSourceMapping(mapping), ok
mapping, ok, _ := r.state.ResolveSource(key)
return mapping, ok
}
func (r *ReadSnapshot) ResolveSource(key string) (SourceMapping, bool, error) {
return r.state.ResolveSource(key)
}
func sourcePathKey(path string) (string, error) {
@@ -0,0 +1,100 @@
package workspacestate
import (
"crypto/sha256"
"fmt"
"slices"
"strings"
)
// Source identities written before physical-path normalization remain durable
// receipts. Add read aliases without rewriting their keys or operation journals.
type sourceIdentityIndex struct {
aliases map[string][]string
owners map[string][]string
}
func newSourceIdentityIndex(state State) *sourceIdentityIndex {
index := &sourceIdentityIndex{aliases: map[string][]string{}, owners: map[string][]string{}}
paths := map[string]string{}
add := func(mapping SourceMapping, committed bool) {
key := mapping.SourceKey
if key == "" {
return
}
keys, known := index.aliases[key]
if !known {
keys = []string{key}
path, resolved := paths[mapping.Path]
if !resolved {
path, _ = sourcePathKey(mapping.Path)
paths[mapping.Path] = path
}
if path != "" {
normalized := fmt.Sprintf("%x", sha256.Sum256([]byte(path+"\x00"+mapping.HeadID)))
if _, version, found := strings.Cut(key, ":review:"); found {
normalized += ":review:" + version
}
if normalized != key {
keys = append(keys, normalized)
}
}
index.aliases[key] = keys
}
if committed {
for _, alias := range keys {
index.owners[alias] = append(index.owners[alias], key)
}
}
}
for _, mapping := range state.SourceMappings {
add(mapping, true)
}
for _, op := range state.PendingOperations {
if op.Mapping != nil {
add(*op.Mapping, false)
}
}
return index
}
func (s State) sourceIdentityIndex() *sourceIdentityIndex {
if s.sourceIdentities != nil {
return s.sourceIdentities
}
return newSourceIdentityIndex(s)
}
// SourceKeys returns independent storage and current lookup identities. Head
// and reviewed-version suffixes are never collapsed into the parent source.
func (s State) SourceKeys(key string) []string {
keys := s.sourceIdentityIndex().aliases[key]
if len(keys) == 0 {
return []string{key}
}
return slices.Clone(keys)
}
// ResolveSource rejects ambiguous normalized ownership instead of choosing a
// destination or admitting another import. Exact durable keys remain usable.
func (s State) ResolveSource(key string) (SourceMapping, bool, error) {
if mapping, ok := s.SourceMappings[key]; ok {
return cloneSourceMapping(mapping), true, nil
}
var result SourceMapping
found := false
for _, owner := range s.sourceIdentityIndex().owners[key] {
mapping, exists := s.SourceMappings[owner]
if !exists {
continue
}
if found && (result.SessionID != mapping.SessionID || result.WorkspaceID != mapping.WorkspaceID || result.Fingerprint != mapping.Fingerprint) {
return SourceMapping{}, false, ErrMutationConflict
}
if !found || mapping.SourceKey < result.SourceKey {
result = mapping
}
found = true
}
return cloneSourceMapping(result), found, nil
}
@@ -0,0 +1,108 @@
package workspacestate
import (
"bytes"
"crypto/sha256"
"encoding/json"
"errors"
"fmt"
"os"
"path/filepath"
"slices"
"testing"
)
func TestSourceIdentityAliasesSurviveReopenWithoutRewritingReceipts(t *testing.T) {
path := filepath.Join(t.TempDir(), "MixedCase", "history")
if err := os.MkdirAll(path, 0700); err != nil {
t.Fatal(err)
}
physical, err := sourcePathKey(path)
if err != nil {
t.Fatal(err)
}
current := func(head string) string { return fmt.Sprintf("%x", sha256.Sum256([]byte(physical+"\x00"+head))) }
state := newState()
// Simulate durable pre-normalization keys on every CI filesystem.
for _, head := range []string{"", "main", "fork"} {
key := "old-" + head
state.SourceMappings[key] = SourceMapping{SourceKey: key, Path: path, HeadID: head, SessionID: "session-" + head, WorkspaceID: GlobalWorkspaceID, Fingerprint: "fingerprint"}
}
version := state.SourceMappings["old-main"]
version.SourceKey, version.SessionID = "old-main:review:version-1", "reviewed"
state.SourceMappings[version.SourceKey] = version
pending := SourceMapping{SourceKey: "old-pending", Path: path, HeadID: "pending"}
state.PendingOperations["pending"] = Operation{ID: "pending", Mapping: &pending, Kind: "import", Phase: "content_ready", Lifecycle: Active}
body, err := json.Marshal(state)
if err != nil {
t.Fatal(err)
}
file := filepath.Join(t.TempDir(), "state.json")
if err := os.WriteFile(file, body, 0600); err != nil {
t.Fatal(err)
}
for range 2 {
store := NewStore(file)
for _, project := range []bool{false, true} {
view, err := store.loadSnapshot(t.Context(), project)
if err != nil {
t.Fatal(err)
}
for _, head := range []string{"", "main", "fork"} {
mapping, found, err := view.ResolveSource(current(head))
if err != nil || !found || mapping.SourceKey != "old-"+head || mapping.SessionID != "session-"+head {
t.Fatalf("head %q resolved to %+v, %v", head, mapping, err)
}
}
mapping, found, err := view.ResolveSource(current("main") + ":review:version-1")
if err != nil || !found || mapping.SessionID != "reviewed" {
t.Fatalf("reviewed version: %+v %v", mapping, err)
}
if _, found, _ := view.ResolveSource(current("unadopted")); found {
t.Fatal("independent head was consumed")
}
if _, found, _ := view.ResolveSource(current("main") + ":review:version-2"); found {
t.Fatal("new version was consumed")
}
if !slices.Contains(view.SourceKeys("old-pending"), current("pending")) {
t.Fatal("pending operation lost its alias")
}
keys := view.SourceKeys("old-main")
keys[0] = "changed"
if view.SourceKeys("old-main")[0] != "old-main" {
t.Fatal("caller changed cached aliases")
}
}
snapshot, err := store.VerifySnapshot(t.Context())
if err != nil {
t.Fatal(err)
}
mapping, found, err := snapshot.ResolveSource(current(""))
if err != nil || !found || mapping.SourceKey != "old-" {
t.Fatalf("execution snapshot: %+v %v", mapping, err)
}
after, err := os.ReadFile(file)
if err != nil || !bytes.Equal(after, body) {
t.Fatal("read aliases rewrote persisted receipts")
}
}
}
func TestSourceIdentityAliasesRejectAmbiguousOwnership(t *testing.T) {
path := t.TempDir()
physical, err := sourcePathKey(path)
if err != nil {
t.Fatal(err)
}
key := fmt.Sprintf("%x", sha256.Sum256([]byte(physical+"\x00")))
state := newState()
for _, id := range []string{"first", "second"} {
state.SourceMappings[id] = SourceMapping{SourceKey: id, Path: path, SessionID: id}
}
if _, found, err := state.ResolveSource(key); found || !errors.Is(err, ErrMutationConflict) {
t.Fatalf("ambiguous source: found=%v err=%v", found, err)
}
if mapping, found, err := state.ResolveSource("first"); err != nil || !found || mapping.SessionID != "first" {
t.Fatal("exact durable identity was lost")
}
}
+7 -2
View File
@@ -67,7 +67,10 @@ type State struct {
RecoveryEntries map[string]RecoveryEntry `json:"recoveryEntries"`
Presentation map[string]Presentation `json:"presentation"`
TopicRemovals map[string]TopicRemoval `json:"topicRemovals,omitempty"`
extra map[string]json.RawMessage
// Immutable derived index for display copies; never serialized.
adoptedTopics map[string]map[string]bool
sourceIdentities *sourceIdentityIndex
extra map[string]json.RawMessage
}
type Store struct {
@@ -129,8 +132,10 @@ func (r *ReadSnapshot) cloneState(projection bool) State {
if projection {
state = State{Version: state.Version, Generation: state.Generation, Initialized: state.Initialized,
WorkspaceIDs: state.WorkspaceIDs, Workspaces: state.Workspaces,
SessionStates: state.SessionStates, SourceMappings: state.SourceMappings, Presentation: state.Presentation}
SessionStates: state.SessionStates, SourceMappings: state.SourceMappings, Presentation: state.Presentation,
adoptedTopics: r.adoptedTopics}
}
state.sourceIdentities = r.state.sourceIdentities
return cloneSnapshot(state)
}
@@ -0,0 +1,54 @@
package workspacestate
import "maps"
// AdoptedTopicIDs includes durable adoption evidence after purge removes live
// membership and presentation. Display projections retain only this small
// derived index, not the operation journals or their request/content payloads.
func (s State) AdoptedTopicIDs(workspaceID string) map[string]bool {
index := s.adoptedTopics
if index == nil {
index = adoptedTopicIndex(s)
}
return maps.Clone(index[workspaceID])
}
func adoptedTopicIndex(state State) map[string]map[string]bool {
index := map[string]map[string]bool{}
add := func(workspaceID, topicID string) {
if workspaceID == "" || topicID == "" {
return
}
if index[workspaceID] == nil {
index[workspaceID] = map[string]bool{}
}
index[workspaceID][topicID] = true
}
for workspaceID, workspace := range state.Workspaces {
for _, id := range workspace.SessionIDs {
add(workspaceID, state.Presentation[id].TopicID)
}
}
for _, op := range state.PendingOperations {
if op.Phase != "committed" || op.Presentation == nil {
continue
}
if op.Kind == "purge" && len(op.SessionIDs) == 1 && ClassifyPurge(state, op.SessionIDs[0]) == PurgeCommitted {
add(op.WorkspaceID, op.Presentation.TopicID)
continue
}
if op.Mapping == nil {
continue
}
mapping, exists := state.SourceMappings[op.Mapping.SourceKey]
if exists && mapping.WorkspaceID == op.WorkspaceID && mapping.SessionID == op.Mapping.SessionID {
add(mapping.WorkspaceID, op.Presentation.TopicID)
}
}
for _, removal := range state.TopicRemovals {
if removal.Disposition == "archive_sessions" && removal.Phase == "committed" {
add(removal.WorkspaceID, removal.TopicID)
}
}
return index
}
@@ -0,0 +1,126 @@
package workspacestate
import (
"encoding/json"
"os"
"path/filepath"
"reflect"
"testing"
previous "reasonix/desktop/internal/workspacestate/testdata/v2previous"
)
func TestTopicAdoptionProjectionSurvivesPurgeAndReopen(t *testing.T) {
state := newState()
state.Workspaces["w"] = Workspace{ID: "w", SessionIDs: []string{"archived"}}
state.Presentation["archived"] = Presentation{TopicID: "archived-topic"}
state.SessionStates["archived"] = SessionState{Lifecycle: Archived}
state.SessionStates["purged"] = SessionState{Lifecycle: Deleted}
mapping := SourceMapping{SourceKey: "source", Path: "retained.jsonl", Format: "legacy", Fingerprint: "fingerprint", SessionID: "purged", WorkspaceID: "w"}
state.SourceMappings[mapping.SourceKey] = mapping
state.PendingOperations["import"] = Operation{ID: "import", Kind: "import", Lifecycle: Active, Phase: "committed", WorkspaceID: "w", Mapping: &mapping, Presentation: &Presentation{TopicID: "purged-topic"}}
state.TopicRemovals["remove"] = TopicRemoval{ID: "remove", WorkspaceID: "w", TopicID: "removed-topic", Disposition: "archive_sessions", Phase: "committed"}
state.TopicRemovals["pending"] = TopicRemoval{ID: "pending", WorkspaceID: "w", TopicID: "pending-topic", Disposition: "archive_sessions", Phase: "prepared"}
state.TopicRemovals["placeholder"] = TopicRemoval{ID: "placeholder", WorkspaceID: "w", TopicID: "placeholder-topic", Disposition: "archive_placeholder", Phase: "restored"}
state.PendingOperations["uncommitted"] = Operation{ID: "uncommitted", Kind: "import", Lifecycle: Active, Phase: "content_ready", WorkspaceID: "w", Mapping: &mapping, Presentation: &Presentation{TopicID: "not-adopted"}}
// Historical registries can retain multiple source/head receipts for one
// topic, old purge receipts with no presentation, and a failed new archive.
// None of that should make the consumed topic into a fresh placeholder.
head := mapping
head.SourceKey, head.HeadID = "source-main", "main"
state.SourceMappings[head.SourceKey] = head
state.PendingOperations["import-main"] = Operation{ID: "import-main", Kind: "import", Lifecycle: Active, Phase: "committed", WorkspaceID: "w", Mapping: &head, Presentation: &Presentation{TopicID: "purged-topic"}}
sibling := mapping
sibling.SourceKey, sibling.SessionID = "source-sibling", "purged-sibling"
state.SourceMappings[sibling.SourceKey] = sibling
state.SessionStates[sibling.SessionID] = SessionState{Lifecycle: Deleted}
state.PendingOperations["import-sibling"] = Operation{ID: "import-sibling", Kind: "import", Lifecycle: Archived, Phase: "committed", WorkspaceID: "w", Mapping: &sibling, Presentation: &Presentation{TopicID: "purged-topic"}}
for _, id := range []string{"purged", "purged-sibling"} {
state.PendingOperations["purge-"+id] = Operation{ID: "purge-" + id, Kind: "purge", Lifecycle: Deleted, Phase: "committed", SessionIDs: []string{id}}
}
state.TopicRemovals["retry-purged"] = TopicRemoval{ID: "retry-purged", WorkspaceID: "w", TopicID: "purged-topic", Disposition: "archive_sessions", Phase: "prepared"}
want := map[string]bool{"archived-topic": true, "purged-topic": true, "removed-topic": true}
body, err := json.Marshal(state)
if err != nil {
t.Fatal(err)
}
path := filepath.Join(t.TempDir(), "state.json")
if err := os.WriteFile(path, body, 0600); err != nil {
t.Fatal(err)
}
for range 2 {
store := NewStore(path)
projection, err := store.LoadProjection(t.Context())
if err != nil {
t.Fatal(err)
}
if got := projection.AdoptedTopicIDs("w"); !reflect.DeepEqual(got, want) {
t.Fatalf("adopted topics = %v; want %v", got, want)
}
if len(projection.AdoptedTopicIDs("other")) != 0 || len(projection.PendingOperations) != 0 || len(projection.TopicRemovals) != 0 {
t.Fatal("projection leaked another workspace or operation payloads")
}
projection.AdoptedTopicIDs("w")["unrelated"] = true
if projection.AdoptedTopicIDs("w")["unrelated"] {
t.Fatal("caller mutated the cached adoption index")
}
full, err := store.Load(t.Context())
if err != nil || !reflect.DeepEqual(full.AdoptedTopicIDs("w"), want) {
t.Fatalf("full snapshot adoption differs: %v", err)
}
after, err := os.ReadFile(path)
if err != nil || string(after) != string(body) {
t.Fatal("projection rewrote registry bytes")
}
}
}
func TestPurgeRetainsTopicAdoptionWithoutImport(t *testing.T) {
store, _ := seedArchivedProcessState(t)
if err := store.EnsureSessionTopic(t.Context(), "victim", "legacy-topic", "Private title"); err != nil {
t.Fatal(err)
}
state, err := store.Load(t.Context())
if err != nil {
t.Fatal(err)
}
if err := store.BeginPurge(t.Context(), "victim", state.Generation); err != nil {
t.Fatal(err)
}
if err := store.AdvancePurge(t.Context(), "victim", "content_removed"); err != nil {
t.Fatal(err)
}
// Resume completion with a new store to cover an interrupted purge.
store = NewStore(store.Path())
if err := store.CompletePurge(t.Context(), "victim"); err != nil {
t.Fatal(err)
}
if err := store.RenameWorkspace(t.Context(), GlobalWorkspaceID, "Unrelated edit"); err != nil {
t.Fatal(err)
}
state, err = NewStore(store.Path()).Load(t.Context())
if err != nil {
t.Fatal(err)
}
if !state.AdoptedTopicIDs(GlobalWorkspaceID)["legacy-topic"] || len(state.Workspaces[GlobalWorkspaceID].SessionIDs) != 0 || len(state.Presentation) != 0 {
t.Fatal("purge lost topic adoption or retained live presentation")
}
receipt := state.PendingOperations["purge-victim"]
if receipt.Presentation == nil || receipt.Presentation.Title != "" || receipt.Mapping != nil {
t.Fatal("purge receipt retained more than topic identity")
}
// These are existing operation fields, including for older codecs. The
// registry schema fence remains unchanged; this checks the receipt shape.
body, err := json.Marshal(receipt)
if err != nil {
t.Fatal(err)
}
var old previous.Operation
if err := json.Unmarshal(body, &old); err != nil {
t.Fatal(err)
}
roundtrip, err := json.Marshal(old)
if err != nil || string(roundtrip) != string(body) {
t.Fatal("previous operation codec changed the purge receipt")
}
}
@@ -63,7 +63,10 @@ func (a *App) resolveLegacyCleanupSourceTarget(item legacycleanup.Candidate) (le
if err != nil {
return legacyCleanupSourceTarget{}, "unknown", "workspace_unavailable", false
}
mapping, ok := state.SourceMappings[desktopSourceKey(item.SourcePath, item.SourceHeadID)]
mapping, ok, resolveErr := state.ResolveSource(desktopSourceKey(item.SourcePath, item.SourceHeadID))
if resolveErr != nil {
return legacyCleanupSourceTarget{}, "protected", "migration_identity_changed", false
}
if !ok {
return legacyCleanupSourceTarget{}, "unknown", "legacy_session_requires_migration", false
}
@@ -114,8 +117,8 @@ func (a *App) archiveLegacyCleanupSource(item legacycleanup.Candidate, target le
}
defer sourceGuard.Release()
verify := func(ctx context.Context, latest workspacestate.State) error {
current, exists := latest.SourceMappings[desktopSourceKey(item.SourcePath, item.SourceHeadID)]
if !exists || current.SessionID != target.mapping.SessionID || current.WorkspaceID != target.mapping.WorkspaceID ||
current, exists, resolveErr := latest.ResolveSource(desktopSourceKey(item.SourcePath, item.SourceHeadID))
if resolveErr != nil || !exists || current.SessionID != target.mapping.SessionID || current.WorkspaceID != target.mapping.WorkspaceID ||
!sameDesktopPath(current.Path, item.SourcePath) || current.HeadID != item.SourceHeadID {
return fmt.Errorf("%w: migration mapping changed", errLegacyCleanupStateChanged)
}
@@ -253,8 +256,8 @@ func (a *App) classifyLegacyCleanupRegistry(ctx context.Context, ref session.Ses
func classifyLegacyCleanupSourceMappings(state workspacestate.State, ref session.SessionRef, frozen legacycleanup.Candidate) (legacyCleanupDecision, bool) {
for _, source := range frozen.Sources {
mapping, exists := state.SourceMappings[desktopSourceKey(source.Path, source.HeadID)]
if !exists || mapping.SessionID != ref.SessionID || mapping.WorkspaceID != frozen.WorkspaceID || mapping.Format != "legacy" ||
mapping, exists, resolveErr := state.ResolveSource(desktopSourceKey(source.Path, source.HeadID))
if resolveErr != nil || !exists || mapping.SessionID != ref.SessionID || mapping.WorkspaceID != frozen.WorkspaceID || mapping.Format != "legacy" ||
!sameDesktopPath(mapping.Path, source.Path) || mapping.HeadID != source.HeadID {
return legacyCleanupDecision{"protected", "legacy_mapping_changed", session.SessionInfo{}, session.Snapshot{}}, false
}
@@ -6,6 +6,7 @@ import (
"testing"
"time"
"reasonix/desktop/internal/workspacestate"
"reasonix/internal/agent"
"reasonix/internal/provider"
"reasonix/internal/session"
@@ -87,6 +88,10 @@ func TestArchivedHistoricalHeadDoesNotHideUnadoptedSibling(t *testing.T) {
if err := legacy.Save(path); err != nil {
t.Fatal(err)
}
before, err := desktopSourceFingerprint(path)
if err != nil {
t.Fatal(err)
}
app := NewApp()
app.ctx = t.Context()
pinDesktopSessionRoot(t, app)
@@ -112,4 +117,30 @@ func TestArchivedHistoricalHeadDoesNotHideUnadoptedSibling(t *testing.T) {
if err != nil || len(page.Items) != 2 {
t.Fatalf("restored heads = %+v, %v", page.Items, err)
}
if result, err := app.ArchiveSessionTarget(selector); err != nil || !result.Committed {
t.Fatalf("rearchive one head = %+v, %v", result, err)
}
state, err := app.workspaceRegistry().Load(t.Context())
if err != nil {
t.Fatal(err)
}
mapping, found, err := state.ResolveSource(desktopSourceKey(path, head))
if err != nil || !found {
t.Fatalf("adopted head missing: %v", err)
}
ref := session.SessionRef{HostID: localDesktopHostID, SessionID: mapping.SessionID}
if err := app.PurgeCanonicalSession(ref); err != nil {
t.Fatal(err)
}
if after, err := desktopSourceFingerprint(path); err != nil || after != before {
t.Fatalf("purge damaged independent legacy head: %v", err)
}
page, err = app.ListProjectTopics(ProjectTopicPageRequest{Scope: "global", Limit: 50})
if err != nil || len(page.Items) != 1 || page.Items[0].Source == nil || page.Items[0].Source.HeadID == head {
t.Fatalf("purge hid sibling or revived deleted head: %+v %v", page.Items, err)
}
state, err = app.workspaceRegistry().Load(t.Context())
if err != nil || state.SessionStates[ref.SessionID].Lifecycle != workspacestate.Deleted {
t.Fatalf("head tombstone lost: %v", err)
}
}
+3 -1
View File
@@ -61,7 +61,9 @@ func (a *App) listSessionsFromDir(dir, active string) []SessionMeta {
}
adopted := map[string]bool{}
for _, mapping := range state.SourceMappings {
adopted["source\x00local\x00"+mapping.SourceKey] = true
for _, key := range state.SourceKeys(mapping.SourceKey) {
adopted["source\x00local\x00"+key] = true
}
if sourceMappingHasPathAlias(mapping) {
adopted[sessionRuntimeKey(mapping.Path)] = true
}
+5 -1
View File
@@ -79,7 +79,11 @@ func (a *App) PurgeTrashedSession(path string) error {
if err != nil {
return err
}
if mapping, adopted := state.SourceMappings[desktopSourceKey(path, "")]; adopted {
mapping, adopted, err := state.ResolveSource(desktopSourceKey(path, ""))
if err != nil {
return err
}
if adopted {
return a.PurgeCanonicalSession(session.SessionRef{HostID: localDesktopHostID, SessionID: mapping.SessionID})
}
return errors.New("historical session has no verified canonical identity")
+12 -3
View File
@@ -48,7 +48,11 @@ func (a *App) completeRegisteredMigration(ctx context.Context, source desktopMig
return err
}
key := source.mappingKey(path)
if mapping, exists := state.SourceMappings[key]; exists && source.operationID == "" {
mapping, exists, err := state.ResolveSource(key)
if err != nil {
return err
}
if exists && source.operationID == "" {
if mapping.SessionID != id {
return workspacestate.ErrMutationConflict
}
@@ -111,7 +115,9 @@ func (a *App) quarantineChangedMigration(ctx context.Context, source desktopMigr
if err != nil {
return true, err
}
if _, exists := state.SourceMappings[desktopSourceKey(path, source.headID)]; !exists {
if _, exists, err := state.ResolveSource(desktopSourceKey(path, source.headID)); err != nil {
return true, err
} else if !exists {
return false, nil
}
}
@@ -140,7 +146,10 @@ func (a *App) checkAdoptedMigrationSource(ctx context.Context, source desktopMig
if state.SessionStates[cp.record.TargetSessionID].Lifecycle == workspacestate.Deleted {
return true, nil
}
mapping, exists := state.SourceMappings[source.mappingKey(path)]
mapping, exists, err := state.ResolveSource(source.mappingKey(path))
if err != nil {
return true, err
}
if !exists {
return false, nil
}
+7 -3
View File
@@ -204,7 +204,9 @@ func workspaceSourceAliases(state workspacestate.State, workspaceID string) map[
if m.WorkspaceID != workspaceID {
continue
}
result[m.SessionID] = append(result[m.SessionID], "source\x00local\x00"+m.SourceKey)
for _, key := range state.SourceKeys(m.SourceKey) {
result[m.SessionID] = append(result[m.SessionID], "source\x00local\x00"+key)
}
if sourceMappingHasPathAlias(m) {
result[m.SessionID] = append(result[m.SessionID], "path\x00"+m.Path)
}
@@ -221,7 +223,9 @@ func sourceAliases(state workspacestate.State, workspaceID, sessionID string) []
if m.WorkspaceID != workspaceID || m.SessionID != sessionID {
continue
}
aliases = append(aliases, "source\x00local\x00"+m.SourceKey)
for _, key := range state.SourceKeys(m.SourceKey) {
aliases = append(aliases, "source\x00local\x00"+key)
}
if sourceMappingHasPathAlias(m) {
aliases = append(aliases, "path\x00"+m.Path)
}
@@ -316,7 +320,7 @@ func applyResolvedOrganizationMutation(state *workspacestate.State, workspaceID
if target.Source == nil || target.Source.SourceKey == "" {
return workspacestate.ErrMutationConflict
}
if _, adopted := state.SourceMappings[target.Source.SourceKey]; adopted {
if _, adopted, err := state.ResolveSource(target.Source.SourceKey); adopted || err != nil {
return workspacestate.ErrMutationConflict
}
continue
+20 -9
View File
@@ -4,6 +4,7 @@ import (
"context"
"errors"
"os"
"slices"
"sort"
"strings"
@@ -58,20 +59,27 @@ func (a *App) historicalSourceForSelector(selector SessionSelector) (string, his
return "", historicalSource{}, newSessionOperationError(sessionOperationTargetNotFound, "The source no longer exists.")
}
id := desktopSourceKey(ref.Path, ref.HeadID)
if ref.SourceKey != "" && ref.SourceKey != id {
state, loadErr := a.workspaceRegistry().Load(a.bootContext())
if loadErr != nil {
return "", historicalSource{}, loadErr
}
if ref.SourceKey != "" && !slices.Contains(state.SourceKeys(ref.SourceKey), id) {
return "", historicalSource{}, newSessionOperationError("target_changed", "The source identity changed.")
}
mapping, adopted, resolveErr := state.ResolveSource(id)
if resolveErr != nil {
return "", historicalSource{}, resolveErr
}
format, scope, root := "legacy", "global", ""
if info, statErr := os.Stat(ref.Path); statErr == nil && info.IsDir() {
format = "canonical"
}
if state, loadErr := a.workspaceRegistry().Load(a.bootContext()); loadErr == nil {
if mapping, ok := state.SourceMappings[id]; ok {
workspace := state.Workspaces[mapping.WorkspaceID]
root = workspace.Root
if mapping.WorkspaceID != "global" {
scope = "project"
}
if adopted {
id = mapping.SourceKey
workspace := state.Workspaces[mapping.WorkspaceID]
root = workspace.Root
if mapping.WorkspaceID != "global" {
scope = "project"
}
}
if root == "" {
@@ -257,7 +265,10 @@ func (a *App) checkHistoricalSourceUpdate(ctx context.Context, id string, source
if err != nil {
return HistoricalSourceUpdateView{SourceKey: id, Status: "failed", ErrorCode: "registry_unavailable", Retryable: true}
}
mapping, ok := state.SourceMappings[id]
mapping, ok, resolveErr := state.ResolveSource(id)
if resolveErr != nil {
return HistoricalSourceUpdateView{SourceKey: id, Status: "failed", ErrorCode: "target_changed"}
}
if !ok {
return HistoricalSourceUpdateView{SourceKey: id, Status: "not_prepared", Retryable: true}
}
+7 -4
View File
@@ -10,9 +10,8 @@ import (
"reasonix/internal/session"
)
// PurgeCanonicalSession permanently removes only archived sessions. Legacy
// migration originals are retained as upgrade evidence, with mappings acting
// as tombstones so background discovery cannot import them again.
// PurgeCanonicalSession permanently removes only archived sessions. Adoption
// receipts survive deletion, including when an exclusive original is removed.
func (a *App) PurgeCanonicalSession(ref session.SessionRef) error {
if err := validateLocalSessionRef(ref); err != nil {
return err
@@ -128,7 +127,7 @@ func (a *App) executeCanonicalPurge(ctx context.Context, ref session.SessionRef,
if observed != nil {
return store.ResumePurgeForRequest(ctx, ref.SessionID, expected, *observed)
}
return store.BeginPurge(ctx, ref.SessionID, expected)
return store.BeginPurgeWithSources(ctx, ref.SessionID, expected)
}
switch workspacestate.ClassifyPurge(state, ref.SessionID) {
case workspacestate.PurgeCommitted:
@@ -166,6 +165,10 @@ func (a *App) executeCanonicalPurge(ctx context.Context, ref session.SessionRef,
return fmt.Errorf("purge cleanup pending: %w", err)
}
a.lifecycleCheckpoint("after-file-cleanup")
if err := a.purgeMigratedSources(ctx, ref.SessionID); err != nil {
return fmt.Errorf("source cleanup pending: %w", err)
}
a.lifecycleCheckpoint("after-source-cleanup")
if err := a.sessionUIStore().PurgeComposer(ctx, composerRecordKey(ref)); err != nil {
return err
}
+136
View File
@@ -0,0 +1,136 @@
package main
import (
"context"
"encoding/json"
"errors"
"os"
"path/filepath"
"strings"
"reasonix/desktop/internal/workspacestate"
"reasonix/internal/session"
)
var errPurgeSourceRetained = errors.New("migrated source must be retained")
// Cleanup is opt-in in the original tombstone transaction. Replaying an older
// purge must not grant new authority to remove upgrade evidence. Canonical
// directories use the same ownership lock and crash-safe staging as native
// deletion; shared content pools and legacy multi-head files remain untouched.
func (a *App) purgeMigratedSources(ctx context.Context, id string) error {
state, err := a.workspaceRegistry().Load(ctx)
if err != nil {
return err
}
op := state.PendingOperations["purge-"+id]
if len(op.Request) == 0 {
return nil
}
var plan workspacestate.PurgeSourceCleanup
if err := json.Unmarshal(op.Request, &plan); err != nil {
return err
}
if plan.Version != 1 {
return workspacestate.ErrUnsupportedVersion
}
for _, mapping := range plan.Sources {
if mapping.SessionID != id || mapping.Format != "canonical" || mapping.HeadID != "" || !filepath.IsAbs(mapping.Path) {
return workspacestate.ErrMutationConflict
}
// Never let a legacy receipt authorize deletion within the live store.
if pathsOverlapForPurge(mapping.Path, a.desktopSessions.root) || !purgeSourceExclusive(state, id, mapping) {
continue
}
// An offline or relocated historical root is not required to delete
// the canonical conversation. The adoption tombstone still hides it
// if it returns. Never follow a replaced directory symlink for cleanup.
rootInfo, rootErr := os.Lstat(filepath.Dir(mapping.Path))
if os.IsNotExist(rootErr) || (rootErr == nil && !rootInfo.IsDir()) {
continue
}
if rootErr != nil {
return rootErr
}
if info, err := os.Lstat(mapping.Path); err == nil && !info.IsDir() {
continue
}
fs := session.NewFilesystemPersistence(filepath.Dir(mapping.Path))
err := fs.PurgeWithTombstone(ctx, filepath.Base(mapping.Path), func() error {
// Both directory ownership and writer ownership are held here.
// Imports hold shared directory ownership through registry commit.
current, err := a.workspaceRegistry().Load(ctx)
if err != nil {
return err
}
if !purgeSourceExclusive(current, id, mapping) {
return errPurgeSourceRetained
}
fingerprint, err := desktopSourceFingerprint(mapping.Path)
if err == nil && fingerprint != mapping.Fingerprint {
return errPurgeSourceRetained
}
// An absent source can be a completed rename from a prior attempt.
// FilesystemPersistence verifies ownership of any staged directory.
if err != nil && !os.IsNotExist(err) {
return errPurgeSourceRetained
}
a.lifecycleCheckpoint("before-source-cleanup")
return nil
})
if err != nil && !errors.Is(err, errPurgeSourceRetained) {
return err
}
}
return nil
}
func purgeSourceExclusive(state workspacestate.State, id string, source workspacestate.SourceMapping) bool {
if state.SessionStates[id].Lifecycle != workspacestate.Deleted || workspacestate.ClassifyPurge(state, id) != workspacestate.PurgeTombstoned {
return false
}
registered, ok := state.SourceMappings[source.SourceKey]
if !ok || registered.Path != source.Path || registered.SessionID != id || registered.Fingerprint != source.Fingerprint {
return false
}
for _, mapping := range state.SourceMappings {
if state.SessionStates[mapping.SessionID].Lifecycle != workspacestate.Deleted && purgeMappingReferences(mapping, source.Path) {
return false
}
}
for _, op := range state.PendingOperations {
if op.Phase != "committed" && op.Mapping != nil && purgeMappingReferences(*op.Mapping, source.Path) {
return false
}
}
for _, recovery := range state.RecoveryEntries {
if recovery.Status != "restored" && pathsOverlapForPurge(recovery.Path, source.Path) {
return false
}
}
return true
}
func purgeMappingReferences(mapping workspacestate.SourceMapping, source string) bool {
if pathsOverlapForPurge(mapping.Path, source) {
return true
}
for _, artifact := range mapping.RetainedArtifacts {
if pathsOverlapForPurge(artifact, source) {
return true
}
}
return false
}
func pathsOverlapForPurge(left, right string) bool {
if left == "" || right == "" {
return false
}
l, errLeft := canonicalRuntimeRootErr(left)
r, errRight := canonicalRuntimeRootErr(right)
if errLeft != nil || errRight != nil {
return true // Uncertain ownership is never deletion authority.
}
return l == r || strings.HasPrefix(l, r+string(filepath.Separator)) || strings.HasPrefix(r, l+string(filepath.Separator))
}
+223
View File
@@ -0,0 +1,223 @@
package main
import (
"context"
"encoding/json"
"errors"
"os"
"os/exec"
"path/filepath"
"testing"
"time"
"reasonix/desktop/internal/workspacestate"
"reasonix/internal/config"
"reasonix/internal/identitylock"
"reasonix/internal/session"
)
func TestPurgeMigratedSourcesProtectsOtherOwnersAndChangedContent(t *testing.T) {
for _, scenario := range []string{"exclusive", "active_owner", "archived_owner", "pending_import", "recovery", "changed", "older_purge", "missing_root"} {
t.Run(scenario, func(t *testing.T) {
isolateDesktopUserDirs(t)
root := config.SessionStoreDir()
coldV4MigrationFixture(t, root, "source")
app := newHistoricalLifecycleApp(t)
key := historicalLifecycleID(t, app, "source")
result, err := app.ImportHistoricalSession(key)
if err != nil {
t.Fatal(err)
}
state, err := app.workspaceRegistry().Load(t.Context())
if err != nil {
t.Fatal(err)
}
mapping := state.SourceMappings[key]
var otherID string
switch scenario {
case "active_owner", "archived_owner":
other := addLifecycleFixtureSession(t, app, "other-owner")
otherID = other.SessionID
shared := mapping
shared.SourceKey += ":review:other"
shared.SessionID = otherID
if err := app.workspaceRegistry().RecordSource(t.Context(), shared, workspacestate.Presentation{}); err != nil {
t.Fatal(err)
}
if scenario == "archived_owner" {
if err := app.ArchiveCanonicalSession(other); err != nil {
t.Fatal(err)
}
}
case "pending_import":
pending := mapping
pending.SourceKey, pending.SessionID = key+":review:pending", "pending-owner"
if err := app.workspaceRegistry().BeginOperation(t.Context(), workspacestate.Operation{ID: "pending-source-import", Kind: "import", Lifecycle: workspacestate.Active, Mapping: &pending}); err != nil {
t.Fatal(err)
}
case "recovery":
if err := app.sourceRecovery(t.Context(), mapping.Path, "canonical", "source_changed_after_adoption", "global", ""); err != nil {
t.Fatal(err)
}
case "changed":
// A prior application continued the old source after migration.
path := filepath.Join(mapping.Path, "events.frames")
body, err := os.ReadFile(path)
if err != nil {
t.Fatal(err)
}
if err := os.WriteFile(path, append(body, []byte("later source bytes")...), 0600); err != nil {
t.Fatal(err)
}
}
if err := app.ArchiveCanonicalSession(result.Session); err != nil {
t.Fatal(err)
}
if scenario == "older_purge" {
state, err = app.workspaceRegistry().Load(t.Context())
if err != nil {
t.Fatal(err)
}
if err := app.workspaceRegistry().BeginPurge(t.Context(), result.Session.SessionID, state.Generation); err != nil {
t.Fatal(err)
}
}
before, err := desktopSourceFingerprint(mapping.Path)
if err != nil {
t.Fatal(err)
}
verifyPath := mapping.Path
if scenario == "missing_root" {
if err := os.Rename(root, root+"-offline"); err != nil {
t.Fatal(err)
}
verifyPath = filepath.Join(root+"-offline", "source")
}
if err := app.PurgeCanonicalSession(result.Session); err != nil {
t.Fatal(err)
}
if scenario == "exclusive" {
if _, err := os.Stat(mapping.Path); !os.IsNotExist(err) {
t.Fatalf("exclusive original retained: %v", err)
}
} else if after, err := desktopSourceFingerprint(verifyPath); err != nil || after != before {
t.Fatalf("protected original changed: %v", err)
}
state, err = app.workspaceRegistry().Load(t.Context())
if err != nil || workspacestate.ClassifyPurge(state, result.Session.SessionID) != workspacestate.PurgeCommitted {
t.Fatalf("purge incomplete: %v", err)
}
if state.SourceMappings[key].SessionID != result.Session.SessionID {
t.Fatal("deletion lost source tombstone")
}
if otherID != "" && state.SessionStates[otherID].Lifecycle == workspacestate.Deleted {
t.Fatal("purge deleted another source owner")
}
if otherID != "" {
other := session.SessionRef{HostID: localDesktopHostID, SessionID: otherID}
if scenario == "active_owner" {
if err := app.ArchiveCanonicalSession(other); err != nil {
t.Fatal(err)
}
}
if err := app.PurgeCanonicalSession(other); err != nil {
t.Fatal(err)
}
if _, err := os.Stat(mapping.Path); !os.IsNotExist(err) {
t.Fatalf("last source owner left the exclusive original: %v", err)
}
}
})
}
}
func TestPurgeMigratedSourcesBusyWriterResumesFromTombstone(t *testing.T) {
isolateDesktopUserDirs(t)
root := config.SessionStoreDir()
coldV4MigrationFixture(t, root, "busy-source")
app := newHistoricalLifecycleApp(t)
key := historicalLifecycleID(t, app, "busy-source")
result, err := app.ImportHistoricalSession(key)
if err != nil {
t.Fatal(err)
}
if err := app.ArchiveCanonicalSession(result.Session); err != nil {
t.Fatal(err)
}
release, err := identitylock.TryAcquire(filepath.Join(root, "busy-source", "writer.lock"))
if err != nil {
t.Fatal(err)
}
defer release()
if err := app.PurgeCanonicalSession(result.Session); !errors.Is(err, identitylock.ErrHeld) {
t.Fatalf("purge ignored source writer: %v", err)
}
state, err := app.workspaceRegistry().Load(t.Context())
if err != nil || workspacestate.ClassifyPurge(state, result.Session.SessionID) != workspacestate.PurgeTombstoned {
t.Fatalf("missing resumable tombstone: %v", err)
}
var plan workspacestate.PurgeSourceCleanup
if err := json.Unmarshal(state.PendingOperations["purge-"+result.Session.SessionID].Request, &plan); err != nil || len(plan.Sources) != 1 {
t.Fatalf("source cleanup intent lost: %+v %v", plan, err)
}
release()
app.closeSessionServices()
app = newHistoricalLifecycleApp(t)
if err := app.recoverDesktopSessionOperations(t.Context()); err != nil {
t.Fatal(err)
}
if _, err := os.Stat(filepath.Join(root, "busy-source")); !os.IsNotExist(err) {
t.Fatalf("resumed purge retained source: %v", err)
}
if _, err := app.ImportHistoricalSession(key); err == nil {
t.Fatal("purged source was imported again")
}
}
func TestPurgeMigratedSourcesCrashRestart(t *testing.T) {
for _, phase := range []string{"after-tombstone", "before-source-cleanup", "after-source-cleanup"} {
t.Run(phase, func(t *testing.T) {
isolateDesktopUserDirs(t)
root := config.SessionStoreDir()
coldV4MigrationFixture(t, root, "crash-source")
app := newHistoricalLifecycleApp(t)
key := historicalLifecycleID(t, app, "crash-source")
result, err := app.ImportHistoricalSession(key)
if err != nil {
t.Fatal(err)
}
if err := app.ArchiveCanonicalSession(result.Session); err != nil {
t.Fatal(err)
}
req := lifecycleRequest(t, app, result.Session, "source-crash", "purge")
body, err := json.Marshal(req)
if err != nil {
t.Fatal(err)
}
app.stopHistoricalImports()
app.closeSessionServices()
for _, checkpoint := range []string{phase, "", ""} {
ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second)
cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestPurgeCommandCrashHelper$")
cmd.Env = append(os.Environ(), "REASONIX_PURGE_APP_ROOT="+app.desktopSessions.root, "REASONIX_PURGE_APP_REGISTRY="+app.workspaceRegistry().Path(), "REASONIX_PURGE_APP_REQUEST="+string(body), "REASONIX_PURGE_APP_POINT="+checkpoint)
output, err := cmd.CombinedOutput()
cancel()
var exit *exec.ExitError
if checkpoint != "" {
if !errors.As(err, &exit) || exit.ExitCode() != 23 {
t.Fatalf("source checkpoint not reached: %v %s", err, output)
}
} else if err != nil {
t.Fatalf("source recovery failed: %v %s", err, output)
}
}
if _, err := os.Stat(filepath.Join(root, "crash-source")); !os.IsNotExist(err) {
t.Fatalf("original survived crash recovery: %v", err)
}
state, err := workspacestate.NewStore(app.workspaceRegistry().Path()).Load(t.Context())
if err != nil || state.SourceMappings[key].SessionID != result.Session.SessionID || workspacestate.ClassifyPurge(state, result.Session.SessionID) != workspacestate.PurgeCommitted {
t.Fatalf("recovery lost deletion evidence: %v", err)
}
})
}
}
+13 -3
View File
@@ -9,6 +9,7 @@ import (
"io"
"os"
"path/filepath"
"slices"
"sort"
"strings"
@@ -29,6 +30,9 @@ func desktopSourceKey(path, head string) string {
}
func (source desktopMigrationSource) mappingKey(path string) string {
if source.registeredSourceKey != "" {
return source.registeredSourceKey
}
key := desktopSourceKey(path, source.headID)
if source.versionFingerprint != "" {
key += ":review:" + source.versionFingerprint
@@ -276,7 +280,7 @@ func (a *App) resolveDesktopImportTarget(ctx context.Context, query *session.Que
for _, op := range state.PendingOperations {
// Explicit source versions reserve their own mapping key. Recover the
// exact reservation even when the imported manifest has no provenance.
if op.Mapping == nil || op.Mapping.SourceKey != mappingKey || op.Mapping.Fingerprint != fingerprint || len(op.SessionIDs) != 1 {
if op.Mapping == nil || !slices.Contains(state.SourceKeys(op.Mapping.SourceKey), mappingKey) || op.Mapping.Fingerprint != fingerprint || len(op.SessionIDs) != 1 {
continue
}
id := op.SessionIDs[0]
@@ -300,7 +304,10 @@ func (a *App) legacyCanonicalRef(ctx context.Context, path string) (session.Sess
if err != nil {
return session.SessionRef{}, false, err
}
mapping, adopted := snapshot.Source(desktopSourceKey(path, ""))
mapping, adopted, err := snapshot.ResolveSource(desktopSourceKey(path, ""))
if err != nil {
return session.SessionRef{}, false, err
}
if !adopted {
// DAG migration records each head separately. A path-only legacy tab
// still refers to the selected head, not a new import of that path.
@@ -315,7 +322,10 @@ func (a *App) legacyCanonicalRef(ctx context.Context, path string) (session.Sess
}
for _, head := range heads {
if head.Selected && !head.Retired {
mapping, adopted = snapshot.Source(desktopSourceKey(path, head.ID))
mapping, adopted, err = snapshot.ResolveSource(desktopSourceKey(path, head.ID))
if err != nil {
return session.SessionRef{}, false, err
}
break
}
}
+48 -40
View File
@@ -5,6 +5,7 @@ import (
"path/filepath"
"reasonix/internal/agent"
"reasonix/internal/session"
"slices"
"strings"
)
@@ -17,14 +18,18 @@ func (a *App) resolveSourceSessionTarget(selector SessionSelector, allowArchived
return SessionTarget{}, newSessionOperationError("target_not_found", "The source no longer exists.")
}
key := desktopSourceKey(source.Path, source.HeadID)
if source.SourceKey != "" && source.SourceKey != key {
return SessionTarget{}, newSessionOperationError("target_changed", "The source identity changed.")
}
state, err := a.workspaceRegistry().Load(a.bootContext())
if err != nil {
return SessionTarget{}, err
}
if mapping, ok := state.SourceMappings[key]; ok {
if source.SourceKey != "" && !slices.Contains(state.SourceKeys(source.SourceKey), key) {
return SessionTarget{}, newSessionOperationError("target_changed", "The source identity changed.")
}
mapping, ok, err := state.ResolveSource(key)
if err != nil {
return SessionTarget{}, err
}
if ok {
return a.resolveCanonicalSessionTargetState(session.SessionRef{HostID: localDesktopHostID, SessionID: mapping.SessionID}, selector.TopicID, allowArchived)
}
if source.HeadID == "" {
@@ -48,42 +53,45 @@ func (a *App) resolveSourceSessionTarget(selector SessionSelector, allowArchived
}
}
if source.HeadID != "" {
dir, validated, err := a.sessionDirForPath(source.Path)
if err != nil {
return SessionTarget{}, err
}
if _, _, err = validateSessionPath(dir, validated); err != nil {
return SessionTarget{}, err
}
heads, err := agent.ListSessionHeads(validated)
if err != nil {
return SessionTarget{}, err
}
found := false
for _, head := range heads {
if head.ID == source.HeadID && !head.Retired {
found = true
}
}
if !found {
return SessionTarget{}, newSessionOperationError("target_not_found", "This historical head is no longer available.")
}
copy := *source
copy.Path, copy.HostID, copy.SourceKey = validated, localDesktopHostID, key
target := SessionTarget{Source: &copy, SessionPath: validated, Scope: "global"}
for _, project := range loadProjectsFile().Projects {
if canonicalRuntimeRoot(dir) == canonicalRuntimeRoot(desktopSessionDir(project.Root)) {
target.Scope, target.WorkspaceRoot = "project", project.Root
break
}
}
if meta, ok, err := agent.LoadBranchMeta(validated); err == nil && ok {
target.TopicID = meta.TopicID
if meta.Scope != "" {
target.Scope, target.WorkspaceRoot = meta.Scope, meta.WorkspaceRoot
}
}
return target, nil
return a.resolveHistoricalHeadTarget(*source, key)
}
return a.resolveLegacySessionTarget(source.Path, selector.TopicID, allowArchived)
}
func (a *App) resolveHistoricalHeadTarget(source SessionSourceRef, key string) (SessionTarget, error) {
dir, validated, err := a.sessionDirForPath(source.Path)
if err != nil {
return SessionTarget{}, err
}
if _, _, err = validateSessionPath(dir, validated); err != nil {
return SessionTarget{}, err
}
heads, err := agent.ListSessionHeads(validated)
if err != nil {
return SessionTarget{}, err
}
found := false
for _, head := range heads {
if head.ID == source.HeadID && !head.Retired {
found = true
}
}
if !found {
return SessionTarget{}, newSessionOperationError("target_not_found", "This historical head is no longer available.")
}
source.Path, source.HostID, source.SourceKey = validated, localDesktopHostID, key
target := SessionTarget{Source: &source, SessionPath: validated, Scope: "global"}
for _, project := range loadProjectsFile().Projects {
if canonicalRuntimeRoot(dir) == canonicalRuntimeRoot(desktopSessionDir(project.Root)) {
target.Scope, target.WorkspaceRoot = "project", project.Root
break
}
}
if meta, ok, err := agent.LoadBranchMeta(validated); err == nil && ok {
target.TopicID = meta.TopicID
if meta.Scope != "" {
target.Scope, target.WorkspaceRoot = meta.Scope, meta.WorkspaceRoot
}
}
return target, nil
}
+1 -4
View File
@@ -54,10 +54,7 @@ func (a *App) lazyProjectTopicSnapshot(req ProjectTopicPageRequest, reader works
if saved, err := readHistoricalSidecar(); err == nil {
applyHistoricalPresentations(sources, saved)
}
adoptedTopics := map[string]bool{}
for _, id := range state.Workspaces[workspaceID].SessionIDs {
adoptedTopics[state.Presentation[id].TopicID] = true
}
adoptedTopics := state.AdoptedTopicIDs(workspaceID)
// Unbacked user-created topics remain visible without inspecting a body.
for _, node := range a.withRemovablePlaceholderTopics(req, state, nil, adoptedTopics) {
found, err := catalog.HasTopicSessions(ctx, req.Scope, req.WorkspaceRoot, node.TopicID)
+1
View File
@@ -162,6 +162,7 @@ func slogWarnDesktopMigration(err error) {
type desktopMigrationSource struct {
operationID string
registeredSourceKey string
headID string
deferArchive bool
versionFingerprint string
+14
View File
@@ -213,6 +213,7 @@ func TestSidebarKeepsCanonicalSessionsWithSharedTopicIndependent(t *testing.T) {
app := NewApp()
t.Cleanup(app.closeSessionServices)
app.ctx = t.Context()
installNoopRuntimeEvents(app)
app.desktopSessions.root = filepath.Join(root, "desktop-sessions-v5", "by-id")
app.desktopSessions.workspaceState = workspacestate.NewStore(filepath.Join(root, "desktop", "workspace-state-v1.json"))
workspaceID, err := app.ensureDesktopWorkspace(t.Context(), "project", root)
@@ -250,6 +251,19 @@ func TestSidebarKeepsCanonicalSessionsWithSharedTopicIndependent(t *testing.T) {
if !seen["branch-a"] || !seen["branch-b"] {
t.Fatalf("lost a durable branch: %#v", seen)
}
// A topic-level adoption marker must never hide another canonical member
// when just one of the sessions sharing that topic is permanently deleted.
victim := session.SessionRef{HostID: localDesktopHostID, SessionID: "branch-a"}
if err := app.ArchiveCanonicalSession(victim); err != nil {
t.Fatal(err)
}
if err := app.PurgeCanonicalSession(victim); err != nil {
t.Fatal(err)
}
page, err = app.unifiedProjectTopics(ProjectTopicPageRequest{Scope: "project", WorkspaceRoot: root, Limit: 1})
if err != nil || len(page.Items) != 1 || page.Items[0].Session == nil || page.Items[0].Session.SessionID != "branch-b" || page.NextCursor != "" {
t.Fatalf("purging one sibling hid or duplicated the live member: %#v, %v", page, err)
}
}
func TestForkSessionTargetRetryReusesDurableChildWithoutOpeningTab(t *testing.T) {
+4 -5
View File
@@ -178,18 +178,17 @@ func (a *App) materializeProjectTopics(req ProjectTopicPageRequest, reader works
workspace.SessionIDs = admittedWorkspaceTopicMembers(req, state, workspace)
infos, _ := listWorkspaceSessionInfo(a.bootContext(), reader, workspace.SessionIDs)
adopted := map[string]bool{}
adoptedTopics := map[string]bool{}
adoptedTopics := state.AdoptedTopicIDs(workspaceID)
for _, m := range state.SourceMappings {
if m.WorkspaceID == workspaceID {
adopted["source\x00local\x00"+m.SourceKey] = true
for _, key := range state.SourceKeys(m.SourceKey) {
adopted["source\x00local\x00"+key] = true
}
if sourceMappingHasPathAlias(m) {
adopted[sessionRuntimeKey(m.Path)] = true
}
}
}
for _, id := range state.Workspaces[workspaceID].SessionIDs {
adoptedTopics[state.Presentation[id].TopicID] = true
}
all := req
all.Cursor = ""
all.Query = ""
+142
View File
@@ -0,0 +1,142 @@
package main
import (
"bytes"
"os"
"path/filepath"
"testing"
"time"
"reasonix/desktop/internal/workspacestate"
"reasonix/internal/sessioncatalog"
)
func TestLegacyTopicArchiveVisibility(t *testing.T) {
for _, scope := range []string{"global", "project"} {
for _, metadataOnly := range []bool{false, true} {
for _, topicRoute := range []bool{true, false} {
name := scope + "/materialized"
if metadataOnly {
name = scope + "/lazy"
}
if topicRoute {
name += "/topic"
} else {
name += "/session"
}
t.Run(name, func(t *testing.T) {
a, topic, root := topicRemovalFixture(t, scope, "Historical conversation")
if err := a.SetTopicPinned(topic.ID, true); err != nil {
t.Fatal(err)
}
canonicalRoot := pinDesktopSessionRoot(t, a)
sourceRoot := root
if scope == "global" {
sourceRoot = globalWorkspaceRoot()
}
dir := desktopSessionDir(sourceRoot)
if err := os.MkdirAll(dir, 0700); err != nil {
t.Fatal(err)
}
sourcePath := writeTopicSessionWithPrompt(t, dir, "archive-visibility.jsonl", topic.ID, topic.Title, root, "preserve this history", time.Now())
original, err := os.ReadFile(sourcePath)
if err != nil {
t.Fatal(err)
}
catalog, err := sessioncatalog.Open(t.Context(), sessioncatalog.Options{Path: filepath.Join(t.TempDir(), "catalog.sqlite"), MetadataOnly: metadataOnly, DisableRepair: true})
if err != nil {
t.Fatal(err)
}
a.sessionCatalog.Store(catalog)
t.Cleanup(func() { a.desktopSessions.readSnapshots.close(); a.stopSessionCatalog(time.Second) })
reconcileSessionCatalogForTest(t, a, dir, scope, root)
assertVisible := func(want int) {
t.Helper()
page, err := a.ListProjectTopics(ProjectTopicPageRequest{Scope: scope, WorkspaceRoot: root, Limit: 50})
if err != nil || len(page.Items) != want {
t.Fatalf("sidebar = %+v, %v; want %d", page.Items, err, want)
}
}
assertVisible(1)
if topicRoute {
result, err := a.RemoveTopic(inspectedTopicRemoval(t, a, topic.ID))
if err != nil || !result.Committed {
t.Fatalf("archive topic = %+v, %v", result, err)
}
} else {
result, err := a.ArchiveSessionTarget(SessionSelector{SessionPath: sourcePath})
if err != nil || !result.Committed {
t.Fatalf("archive session = %+v, %v", result, err)
}
}
assertVisible(0)
trash, err := a.ListTrashEntries("", "", 50)
if err != nil || len(trash.Items) != 1 || trash.Items[0].Ref == nil {
t.Fatalf("trash = %+v, %v", trash, err)
}
if err := a.RestoreCanonicalSession(*trash.Items[0].Ref); err != nil {
t.Fatal(err)
}
assertVisible(1)
if err := a.ArchiveCanonicalSession(*trash.Items[0].Ref); err != nil {
t.Fatal(err)
}
assertVisible(0)
if err := a.PurgeCanonicalSession(*trash.Items[0].Ref); err != nil {
t.Fatal(err)
}
assertVisible(0)
retained, err := os.ReadFile(sourcePath)
if err != nil || !bytes.Equal(retained, original) {
t.Fatal("archive/purge changed the retained legacy source")
}
// A stale metadata-only row would take the native-confirmation
// route and fail exactly like the reported second archive attempt.
stale := inspectedTopicRemoval(t, a, topic.ID)
stale.OperationID += "-stale"
failed, err := a.RemoveTopic(stale)
if err != nil || failed.Committed || failed.ErrorMessage != "session not found" {
t.Fatalf("stale archive = %+v, %v", failed, err)
}
assertVisible(0)
// The retained source and metadata must not recreate a conversation
// after a fresh process loads the registry without a warm catalog.
a.closeSessionServices()
a.desktopSessions.readSnapshots.close()
a.stopSessionCatalog(time.Second)
a = NewApp()
a.ctx = t.Context()
a.desktopSessions.root = canonicalRoot
installNoopRuntimeEvents(a)
t.Cleanup(a.closeSessionServices)
assertVisible(0)
state, err := a.workspaceRegistry().LoadProjection(t.Context())
if err != nil {
t.Fatal(err)
}
pins, err := a.historicalPinnedShellsFromProjection(ProjectTopicPageRequest{Scope: scope, WorkspaceRoot: root}, state, workspacestate.NewWorkspaceIndex(state), desktopProject{Root: root, Topics: []string{topic.ID}, PinnedTopics: []string{topic.ID}})
if err != nil || len(pins) != 0 {
t.Fatalf("purged topic returned in pinned shell: %+v, %v", pins, err)
}
// Independent legacy content can share the consumed topic ID.
// Its real source row must remain visible after rediscovery.
sibling := writeTopicSessionWithPrompt(t, dir, "independent-sibling.jsonl", topic.ID, topic.Title, root, "independent history", time.Now())
catalog, err = sessioncatalog.Open(t.Context(), sessioncatalog.Options{Path: filepath.Join(t.TempDir(), "rebuilt-catalog.sqlite"), MetadataOnly: metadataOnly, DisableRepair: true})
if err != nil {
t.Fatal(err)
}
a.sessionCatalog.Store(catalog)
reconcileSessionCatalogForTest(t, a, dir, scope, root)
page, err := a.ListProjectTopics(ProjectTopicPageRequest{Scope: scope, WorkspaceRoot: root, Limit: 1})
if err != nil || len(page.Items) != 1 || !sameDesktopPath(page.Items[0].SessionPath, sibling) || page.NextCursor != "" {
t.Fatalf("topic adoption hid independent history: %+v, %v", page, err)
}
if _, err := a.CreateTopic(scope, root, "Unrelated empty topic"); err != nil {
t.Fatal(err)
}
assertVisible(2)
})
}
}
}
}
+25 -6
View File
@@ -37,6 +37,15 @@ contains archived/deleted canonical sessions only.
and lifecycle fences without requiring the old source or its ownership lock.
- Archive and purge state remains authoritative. Deleted or archived sessions
are not resurrected by a later catalog scan.
- Once adopted, the source and its canonical target share one conversation
identity. The sidebar displays only the canonical row; archive hides both
aliases and creates one trash entry, while restore returns only that row.
- Permanent deletion keeps adoption and topic-removal evidence. An explicit
purge records source-cleanup intent before removing content and may also
remove an unchanged canonical source directory once no active, archived,
pending-import, or unresolved recovery owner references it. Shared sources,
changed content, legacy multi-file DAG originals, and shared content pools
remain intact. A missing historical root does not block canonical deletion.
- `PrepareSession`, `GetSessionPreparation`, and
`CancelSessionPreparation` expose `queued`, `preparing`, `ready`, `blocked`,
`failed`, and `cancelled` scheduling states without adding lifecycle phases.
@@ -60,8 +69,13 @@ queue revisions are rejected rather than overwriting another process's work.
## Compatibility
Source files remain unchanged. Existing source mappings, recovery entries,
unknown fields, presentations, and retained artifacts are preserved. The
Discovery, import, archive, and restore leave source files unchanged. Explicit
permanent deletion follows the guarded cleanup rules above. Source mappings,
recovery entries, and unknown fields remain durable; deleted sessions retain
minimal adoption evidence after live presentation and membership are removed.
Path normalization aliases resolve older source hashes without rewriting their
receipts or pending operation identities. Independent head and reviewed-version
identities remain separate; ambiguous normalized ownership is rejected. The
path-only legacy route remains an alias for the selected DAG head; valid head
indexes expose alternate heads as separate on-demand sources. A missing or
stale index degrades to one source row and never causes event-log replay in a
@@ -69,15 +83,17 @@ listing RPC.
The queue sidecar is scheduling intent only. The workspace lifecycle registry
remains authoritative for target Session IDs and commit state. Older builds
ignore the sidecar and optional RPC fields; committed sessions and retained
sources remain readable after rollback. Preparation metadata is never added to
ignore the sidecar and optional RPC fields; surviving committed sessions and
retained sources remain readable after rollback. Explicitly purged content is
not restored by rollback. Older purge journals without source-cleanup intent
never gain authority to remove originals on upgrade. Preparation metadata is never added to
model prompts or transcript messages.
| Data | Compatibility behavior |
| --- | --- |
| Lifecycle registry | No new operation types or phases; existing target IDs and mappings remain authoritative. |
| Lifecycle registry | No new operation types or phases; existing target IDs and mappings remain authoritative. Optional versioned cleanup intent uses the purge request payload. Older readers may leave originals behind but cannot restore the deleted canonical identity. |
| Version 1 queue sidecar | Optional `queueRevision` defaults to zero; existing files load paused. Unknown root and presentation fields survive writes. |
| Rollback to pre-sidecar builds | Scheduling is unavailable; source files and committed canonical sessions remain readable. |
| Rollback to pre-sidecar builds | Scheduling is unavailable; retained source files and surviving canonical sessions remain readable. Purged content remains deleted. |
| Concurrent older sidecar writers | Older code does not honor the new worker lease/revision contract; do not run mixed-version batch writers. Upgrade all Desktop instances first. |
## Verification
@@ -90,3 +106,6 @@ open-after-commit, and batch controls. Regression tests additionally cover norma
global/project discovery while a source is occupied, title/pin search and adoption,
shutdown during commit followed by restart, independent sidecar writers, unknown
field preservation, source-independent recovery, and late navigation responses.
Lifecycle regressions cover older source hashes across restart, one trash entry,
independent heads, shared-source protection, source-writer contention, missing
historical roots, and process exit before and after source cleanup.
@@ -29,6 +29,12 @@ Desktop 启动不再转换所有旧版 transcript 或 canonical v4 目录。启
- 已有 `content_ready` 的操作校验持久化目标和生命周期边界后直接提交,不要求
旧来源仍存在或能够获得其所有权锁,也不会再次转换成第二个会话。
- 归档和删除状态具有最终权威性;目录重新扫描不会让会话复活。
- 迁移完成后,旧来源与新版目标共用一个会话身份。侧栏只展示新版;归档同时隐藏
新旧入口,回收站保留一条记录;恢复只返回新版入口。
- 彻底删除保留迁移和主题移除记录,在删除正文之前持久化旧来源清理意图。
只有内容未变化、且没有活跃或已归档会话、待提交导入及未解决恢复记录引用的
canonical 旧目录才会一起清理。共享来源、已变化内容、旧版多文件 DAG 原件和
共享内容池继续保留。旧历史目录缺失时,不阻止删除新版会话。
- 标题和置顶覆盖项在搜索、排序前生效,迁移提交时转移到目标。
- 侧栏和历史共用准备入口;取消响应必须匹配导航意图、操作 ID 和修订号。
新分支准备期间切换页面,其完成结果不能抢回焦点。
@@ -40,15 +46,18 @@ Desktop 启动不再转换所有旧版 transcript 或 canonical v4 目录。启
## 兼容性
源文件保持不变。已有 source mapping、恢复记录、未知字段、展示信息和保留
产物都会保留。旧版按路径打开的入口仍是当前 DAG head 的别名;有效的 head
发现、导入、归档和恢复不修改源文件;明确执行彻底删除时遵循上述清理条件。
source mapping、恢复记录和未知字段继续保留;移除展示信息和成员关系前,会
保存阻止会话复活的最小迁移证据。路径规范化产生的旧、新来源键通过别名关联,
不改写已有记录或待完成操作身份;独立 head 和明确导入的内容版本仍保持独立,
存在歧义的来源归属会被拒绝。旧版按路径打开的入口仍是当前 DAG head 的别名;有效的 head
索引会把其他 head 暴露为独立的按需源,但列目录时不会重放事件日志。
| 数据 | 兼容行为 |
| --- | --- |
| 生命周期账本 | 不新增操作类型和阶段;目标 ID 与提交映射仍由原账本决定。 |
| 生命周期账本 | 不新增操作类型和阶段;目标 ID 与提交映射仍由原账本决定。清理意图作为带版本的可选 purge request 保存。旧版本可能保留原件,但不能恢复已删除的新版会话;不含清理意图的旧删除记录也不会在升级后扩大删除范围。 |
| v1 调度 sidecar | 可选 `queueRevision` 缺失时按零处理;已有队列以暂停状态载入;保留根对象和展示项的未知字段。 |
| 回退至尚无 sidecar 的版本 | 调度暂不可用;原始来源和已提交会话仍可读取。 |
| 回退至尚无 sidecar 的版本 | 调度暂不可用;保留下来的来源和未彻底删除的会话仍可读取。已彻底删除的内容不会因回退恢复。 |
| 与旧 sidecar 写入器同时运行 | 旧实现不遵守新 worker 租约和修订号;应先升级所有 Desktop 实例,避免混版本批量写入。 |
准备状态不进入模型提示或 transcript 消息。
@@ -60,3 +69,5 @@ Desktop 启动不再转换所有旧版 transcript 或 canonical v4 目录。启
新增回归覆盖全局/项目来源在占用时仍可发现、普通列表中的标题和置顶、迁移去重、
提交期间退出并重启、双实例 sidecar 写入、未知字段保留、来源无关恢复,以及
取消和分支完成的迟到导航响应。测试使用一次性夹具,不修改个人历史数据。
生命周期回归还覆盖旧来源键的重启兼容、回收站单条记录、独立分支、共享来源保护、
来源写入锁争用、历史目录缺失,以及清理旧来源前后进程退出的恢复。