返回 DeepSeek-Reasonix
topic_archive.go
根目录 / desktop / topic_archive.go
1 package main
2
3 import (
4 "errors"
5 "fmt"
6 "log/slog"
7 "os"
8 "path/filepath"
9 "strings"
10 "time"
11
12 "reasonix/internal/agent"
13 "reasonix/internal/control"
14 "reasonix/internal/store"
15 )
16
17 var (
18 errTopicHasActiveWork = errors.New("wait for the session to finish, answer pending prompts, and stop background jobs before archiving this topic")
19 errTopicArchiveBusy = errors.New("Reasonix is finishing another session change — wait a moment and retry archiving")
20 )
21
22 var topicArchiveCleanupHookForTest func() error
23
24 type topicArchiveTrace struct {
25 phase string
26 targetCount int
27 runtimeCount int
28 }
29
30 func (a *App) TrashTopic(topicID string) error {
31 return friendlySessionFileError(a.archiveCompatibleTopic(topicID))
32 }
33
34 func (a *App) topicHasActiveRuntimeWork(topicID string) bool {
35 a.mu.RLock()
36 defer a.mu.RUnlock()
37 for _, tabs := range []map[string]*WorkspaceTab{a.tabs, a.detachedSessions} {
38 for _, tab := range tabs {
39 if tab != nil && tab.TopicID == topicID && tab.hasActiveRuntimeWork() {
40 return true
41 }
42 }
43 }
44 return false
45 }
46
47 func (a *App) trashTopic(topicID string) (retErr error) {
48 topicID = strings.TrimSpace(topicID)
49 if topicID == "" {
50 return fmt.Errorf("topicID is required")
51 }
52 started := time.Now()
53 trace := topicArchiveTrace{phase: "start"}
54 defer func() {
55 outcome := "ok"
56 if retErr != nil {
57 outcome = "failed"
58 }
59 slog.Debug("desktop: topic archive timing", "outcome", outcome, "phase", trace.phase,
60 "total_ms", time.Since(started).Milliseconds(), "target_count", trace.targetCount, "runtime_count", trace.runtimeCount)
61 }()
62 fallback, changedDirs, err := a.commitTopicArchive(topicID, &trace)
63 if err != nil {
64 return err
65 }
66 if fallback.workspaceRoot != "" {
67 changedDirs = append(changedDirs, desktopSessionDir(fallback.workspaceRoot))
68 } else if fallback.needs {
69 changedDirs = append(changedDirs, desktopSessionDir(globalWorkspaceRoot()))
70 }
71 // The last visible topic leaves no replacement runtime (the frontend lands
72 // on the workspace draft). Abandoned transient blanks still go, or
73 // reconciliation could promote a default-titled blank into the registry.
74 keepPath := ""
75 a.mu.RLock()
76 if tab := a.tabs[a.activeTabID]; tab != nil {
77 keepPath = tab.SessionPath
78 }
79 a.mu.RUnlock()
80 trace.phase = "discard_unused_blanks"
81 a.discardUnusedTransientBlankSessions(changedDirs, keepPath)
82 trace.phase = "notify"
83 if len(changedDirs) > 0 {
84 a.emitProjectTreeChangedForSessionDirs(changedDirs...)
85 } else {
86 a.emitProjectTreeMetadataChanged()
87 }
88 return nil
89 }
90
91 func (a *App) commitTopicArchive(topicID string, trace *topicArchiveTrace) (fallbackRuntimeTarget, []string, error) {
92 trace.phase = "runtime_lock"
93 releaseRuntime, ok := a.tryLockRuntimeMutation("trash-topic")
94 if !ok {
95 return fallbackRuntimeTarget{}, nil, errTopicArchiveBusy
96 }
97 defer releaseRuntime()
98 trace.phase = "removal_lock"
99 if !a.sessionRemovalMu.TryLock() {
100 return fallbackRuntimeTarget{}, nil, errTopicArchiveBusy
101 }
102 defer a.sessionRemovalMu.Unlock()
103 trace.phase = "active_work_check"
104 if a.topicHasActiveRuntimeWork(topicID) {
105 return fallbackRuntimeTarget{}, nil, errTopicHasActiveWork
106 }
107 trace.phase = "target_scan"
108 targets, err := a.topicTrashTargets(topicID)
109 if err != nil {
110 return fallbackRuntimeTarget{}, nil, err
111 }
112 trace.targetCount = len(targets)
113 changedDirs := make([]string, 0, len(targets))
114 for _, target := range targets {
115 changedDirs = append(changedDirs, target.dir)
116 }
117 trace.phase = "snapshot"
118 removed := a.captureTopicRuntimeBindings(topicID)
119 trace.runtimeCount = len(removed)
120 if err := a.snapshotTopicRuntimeBindings(removed); err != nil {
121 return fallbackRuntimeTarget{}, nil, err
122 }
123 trace.phase = "acquire_removal_ownership"
124 ownership, err := acquireTopicArchiveOwnership(targets, removed)
125 if err != nil {
126 return fallbackRuntimeTarget{}, nil, err
127 }
128 defer ownership.release()
129 trace.phase = "mark_cleanup_pending"
130 rollbackMarkers, err := markTopicArchiveCleanupPending(topicID, targets)
131 if err != nil {
132 ownership.rollback()
133 return fallbackRuntimeTarget{}, nil, err
134 }
135 trace.phase = "detach_runtimes"
136 fallback, unchanged := a.removeTopicRuntimeBindingsIfUnchanged(topicID, removed)
137 if !unchanged {
138 rollbackMarkers()
139 ownership.rollback()
140 return fallbackRuntimeTarget{}, nil, errTopicArchiveBusy
141 }
142 a.finalizeRemovedTopicRuntimes(removed)
143 destroyBegun := false
144 closedRemoved := map[control.SessionAPI]bool{}
145 defer func() {
146 if destroyBegun {
147 a.closeRemainingRemovedSessionRuntimesAfterDestroyAdmissionHeld(removed, closedRemoved)
148 } else {
149 a.closeRemainingRemovedSessionRuntimesAdmissionHeld(removed, closedRemoved)
150 }
151 }()
152 trace.phase = "teardown"
153 destroyBatches := make([][]control.SessionDestroyHandle, len(targets))
154 for i, target := range targets {
155 destroys := a.destroyHandlesForSession(target.dir, target.sessionPath, removed)
156 destroyBatches[i] = destroys
157 destroyBegun = destroyBegun || len(destroys) > 0
158 }
159 timedOutTargets := waitDestroyHandleBatches(destroyBatches)
160 for i, target := range targets {
161 destroys := destroyBatches[i]
162 a.closeRemovedSessionRuntimesForSessionAfterDestroyAdmissionHeld(removed, target.dir, target.sessionPath, closedRemoved)
163 a.removeSessionCatalogPath(target.sessionPath, "topic_archived")
164 if timedOutTargets[i] {
165 guard := ownership.take(target.sessionPath)
166 go delayedDesktopTopicTrash(target.dir, target.sessionPath, target.key, guard, destroys)
167 continue
168 }
169 trace.phase = "move_artifacts"
170 var err error
171 if hook := topicArchiveCleanupHookForTest; hook != nil {
172 err = hook()
173 }
174 if err == nil {
175 err = trashSessionArtifactsWithGuard(target.dir, target.sessionPath, target.key, ownership.take(target.sessionPath))
176 }
177 finishDestroyHandles(destroys)
178 if err != nil {
179 // Cleanup-pending is the durable commit point. Once bindings have
180 // detached, report the archive as accepted and let startup
181 // reconciliation finish any filesystem operation that could not.
182 slog.Warn("desktop: topic archive cleanup remains pending")
183 }
184 }
185 trace.phase = "delete_topic_metadata"
186 if err := a.deleteTopic(topicID); err != nil {
187 slog.Warn("desktop: topic archive metadata cleanup remains pending")
188 } else if err := clearTopicArchiveMetadataPending(topicID); err != nil {
189 slog.Warn("desktop: topic archive metadata marker cleanup remains pending")
190 }
191 return fallback, changedDirs, nil
192 }
193
194 func markTopicArchiveCleanupPending(topicID string, targets []topicTrashTarget) (func(), error) {
195 if err := markTopicArchiveMetadataPending(topicID, targets); err != nil {
196 return nil, err
197 }
198 marked := make([]string, 0, len(targets))
199 rollback := func() {
200 for _, path := range marked {
201 if err := agent.ClearCleanupPending(path); err != nil {
202 slog.Warn("desktop: rollback topic archive marker failed")
203 }
204 }
205 if err := clearTopicArchiveMetadataPending(topicID); err != nil {
206 slog.Warn("desktop: rollback topic archive metadata marker failed")
207 }
208 }
209 for _, target := range targets {
210 if err := agent.MarkCleanupPending(target.sessionPath, "delete"); err != nil {
211 rollback()
212 return rollback, err
213 }
214 marked = append(marked, target.sessionPath)
215 }
216 return rollback, nil
217 }
218
219 type topicTrashTarget struct {
220 dir string
221 sessionPath string
222 key string
223 }
224
225 func (a *App) topicTrashTargets(topicID string) ([]topicTrashTarget, error) {
226 topicID = strings.TrimSpace(topicID)
227 var targets []topicTrashTarget
228 seen := map[string]bool{}
229 addTarget := func(dir, path string) error {
230 sessionPath, key, err := validateSessionPath(dir, path)
231 if err != nil {
232 return err
233 }
234 id := dir + "\x00" + sessionPath
235 if seen[id] {
236 return nil
237 }
238 seen[id] = true
239 if err := validateSessionTrashTarget(dir, sessionPath, key); err != nil {
240 return err
241 }
242 targets = append(targets, topicTrashTarget{dir: dir, sessionPath: sessionPath, key: key})
243 return nil
244 }
245 for _, dir := range a.knownSessionDirs() {
246 index, err := topicSessionIndexForDir(dir)
247 if err != nil {
248 return nil, err
249 }
250 for _, match := range index.byTopic[topicID] {
251 if !agent.IsCleanupPending(match.path) {
252 if err := addTarget(dir, match.path); err != nil {
253 return nil, err
254 }
255 }
256 }
257 }
258 a.mu.RLock()
259 var runtimeTargets []struct{ dir, path string }
260 for _, tab := range a.runtimeTabsLocked() {
261 if tab == nil || tab.TopicID != topicID {
262 continue
263 }
264 if path := canonicalTabSessionPath(tab.currentSessionPath()); path != "" {
265 dir := tabSessionDir(tab)
266 if filepath.IsAbs(path) {
267 dir = filepath.Dir(path)
268 }
269 runtimeTargets = append(runtimeTargets, struct{ dir, path string }{dir: dir, path: path})
270 }
271 }
272 a.mu.RUnlock()
273 for _, target := range runtimeTargets {
274 if err := addTarget(target.dir, target.path); err != nil {
275 return nil, err
276 }
277 }
278 return targets, nil
279 }
280
281 func topicIndexedInRegistry(scope, workspaceRoot, topicID string) bool {
282 topicID = strings.TrimSpace(topicID)
283 if topicID == "" {
284 return false
285 }
286 if strings.TrimSpace(loadTopicTitles(topicTitleRoot(scope, workspaceRoot))[topicID]) != "" {
287 return true
288 }
289 f := loadProjectsFile()
290 if scope != "project" {
291 return containsDesktopString(f.GlobalTopics, topicID)
292 }
293 if i := projectIndexByRoot(f.Projects, workspaceRoot); i >= 0 {
294 return containsDesktopString(f.Projects[i].Topics, topicID)
295 }
296 return false
297 }
298
299 func (a *App) discardUnusedTransientBlankSessions(dirs []string, keepPath string) {
300 // Forced legacy repair promotes indexed sidecars under this same lock. Hold
301 // it through classification, deletion, and scoped registry cleanup so a
302 // concurrent reconcile cannot reinsert a zero-byte ghost after we sweep it.
303 legacyMigrationMu.Lock()
304 defer legacyMigrationMu.Unlock()
305
306 keepPath = canonicalTabSessionPath(strings.TrimSpace(keepPath))
307 kept := map[string]bool{}
308 if keepPath != "" {
309 kept[keepPath] = true
310 }
311 if a != nil {
312 a.mu.RLock()
313 for _, tabs := range []map[string]*WorkspaceTab{a.tabs, a.detachedSessions} {
314 for _, tab := range tabs {
315 if tab == nil {
316 continue
317 }
318 if path := canonicalTabSessionPath(strings.TrimSpace(tab.SessionPath)); path != "" {
319 kept[path] = true
320 }
321 }
322 }
323 a.mu.RUnlock()
324 }
325 siblingDirs := append([]string(nil), dirs...)
326 if a != nil {
327 siblingDirs = append(siblingDirs, a.knownSessionDirs()...)
328 }
329 seen := map[string]bool{}
330 for _, dir := range dirs {
331 dir = strings.TrimSpace(dir)
332 if dir == "" || seen[dir] {
333 continue
334 }
335 seen[dir] = true
336 entries, err := os.ReadDir(dir)
337 if err != nil {
338 continue
339 }
340 for _, entry := range entries {
341 name := entry.Name()
342 if entry.IsDir() || !store.IsSessionTranscriptName(name) {
343 continue
344 }
345 path := filepath.Join(dir, name)
346 if kept[canonicalTabSessionPath(path)] {
347 continue
348 }
349 if !unusedTransientBlankSession(dir, path) {
350 continue
351 }
352 meta, hasMeta, _ := agent.LoadBranchMeta(path)
353 removed := discardTransientBlankSessionArtifacts(path)
354 if removed && hasMeta && !transientTopicHasSibling(siblingDirs, path, meta) {
355 cleanupTransientBlankTopicRegistration(meta)
356 }
357 if removed && a != nil {
358 a.removeSessionCatalogPath(path, "transient_blank_discarded")
359 }
360 }
361 }
362 }
363
364 func unusedTransientBlankSession(dir, path string) bool {
365 resolved, ok := pinnedTabSessionPath(dir, path)
366 if !ok {
367 return false
368 }
369 info, err := os.Stat(resolved)
370 if err != nil || info.IsDir() || info.Size() != 0 {
371 return false
372 }
373 meta, ok, err := agent.LoadBranchMeta(resolved)
374 if err != nil || !ok {
375 return true
376 }
377 topicID := strings.TrimSpace(meta.TopicID)
378 if topicID == "" {
379 return true
380 }
381 if !isDefaultTopicTitle(meta.TopicTitle) && strings.TrimSpace(meta.TopicTitle) != "" {
382 return false
383 }
384 // A zero-byte default sidecar stays transient after registry projection.
385 // The caller's keep set protects every visible or detached runtime; registry
386 // presence alone cannot prove user content.
387 return true
388 }
389
390 func transientTopicHasSibling(dirs []string, excludedPath string, target agent.BranchMeta) bool {
391 seen := make(map[string]bool, len(dirs))
392 for _, dir := range dirs {
393 dir = strings.TrimSpace(dir)
394 key := projectRootKey(dir)
395 if dir == "" || seen[key] {
396 continue
397 }
398 seen[key] = true
399 entries, err := os.ReadDir(dir)
400 if err != nil {
401 continue
402 }
403 for _, entry := range entries {
404 if entry.IsDir() || !store.IsSessionTranscriptName(entry.Name()) {
405 continue
406 }
407 path := filepath.Join(dir, entry.Name())
408 if sameDesktopPath(path, excludedPath) {
409 continue
410 }
411 meta, ok, err := agent.LoadBranchMeta(path)
412 if err != nil || !ok || strings.TrimSpace(meta.TopicID) != strings.TrimSpace(target.TopicID) {
413 continue
414 }
415 sameRoot := meta.DefaultScope() != "project" || sameProjectRoot(meta.WorkspaceRoot, target.WorkspaceRoot)
416 if meta.DefaultScope() == target.DefaultScope() && sameRoot {
417 return true
418 }
419 }
420 }
421 return false
422 }
423
424 func cleanupTransientBlankTopicRegistration(meta agent.BranchMeta) {
425 topicID := strings.TrimSpace(meta.TopicID)
426 if topicID == "" {
427 return
428 }
429 scope, root := meta.DefaultScope(), normalizeProjectRoot(meta.WorkspaceRoot)
430 _ = updateProjectsFile(func(f *desktopProjectFile) (bool, error) {
431 changed := false
432 if scope != "project" {
433 if next := removeString(f.GlobalTopics, topicID); !sameStringList(next, f.GlobalTopics) {
434 f.GlobalTopics, changed = next, true
435 }
436 if next := removeString(f.GlobalPinnedTopics, topicID); !sameStringList(next, f.GlobalPinnedTopics) {
437 f.GlobalPinnedTopics, changed = next, true
438 }
439 if next, removed := groupsWithoutTopic(f.GlobalGroups, topicID); removed {
440 f.GlobalGroups, f.GlobalGroupsRevision, changed = next, f.GlobalGroupsRevision+1, true
441 }
442 return changed, nil
443 }
444 if index := projectIndexByRoot(f.Projects, root); index >= 0 {
445 project := &f.Projects[index]
446 if next := removeString(project.Topics, topicID); !sameStringList(next, project.Topics) {
447 project.Topics, changed = next, true
448 }
449 if next := removeString(project.PinnedTopics, topicID); !sameStringList(next, project.PinnedTopics) {
450 project.PinnedTopics, changed = next, true
451 }
452 if next, removed := groupsWithoutTopic(project.Groups, topicID); removed {
453 project.Groups, project.GroupsRevision, changed = next, project.GroupsRevision+1, true
454 }
455 }
456 return changed, nil
457 })
458 titleRoot := topicTitleRoot(scope, root)
459 _ = deleteTopicState(titleRoot, topicID)
460 }
461
461 lines GO