返回 DeepSeek-Reasonix
sessions_catalog.go
根目录 / internal / cli / sessions_catalog.go
1 package cli
2
3 import (
4 "context"
5 "encoding/json"
6 "errors"
7 "flag"
8 "fmt"
9 "os"
10 "path/filepath"
11 "strings"
12
13 "reasonix/internal/agent"
14 "reasonix/internal/config"
15 "reasonix/internal/sessioncatalog"
16 )
17
18 func sessionOrSessionsCommand(command string, args []string) int {
19 if command == "sessions" {
20 return sessionsCommand(args)
21 }
22 return sessionCommand(args)
23 }
24
25 func sessionsCommand(args []string) int {
26 if len(args) == 0 {
27 fmt.Fprintln(os.Stderr, "usage: reasonix sessions <reindex|diagnose|cleanup> [--dir PATH] [--json]")
28 return 2
29 }
30 switch args[0] {
31 case "diagnose":
32 return sessionsRecoveryCommand(args[1:], false)
33 case "cleanup":
34 return sessionsRecoveryCommand(args[1:], true)
35 case "reindex":
36 default:
37 fmt.Fprintln(os.Stderr, "usage: reasonix sessions <reindex|diagnose|cleanup> [--dir PATH] [--json]")
38 return 2
39 }
40 fs := flag.NewFlagSet("sessions reindex", flag.ContinueOnError)
41 var dirs stringListFlag
42 jsonOut := fs.Bool("json", false, "print the rebuilt catalog status as JSON")
43 fs.Var(&dirs, "dir", "session directory to index; repeat for multiple directories")
44 if code, ok := parseCommandFlags(fs, args[1:]); !ok {
45 return code
46 }
47 if fs.NArg() != 0 {
48 fmt.Fprintln(os.Stderr, "usage: reasonix sessions reindex [--dir PATH] [--json]")
49 return 2
50 }
51 if len(dirs) == 0 {
52 status, err := sessioncatalog.Rebuild(context.Background(), sessioncatalog.DefaultPath(), defaultSessionCatalogTargets())
53 if err != nil {
54 fmt.Fprintln(os.Stderr, "error:", err)
55 return 1
56 }
57 return printSessionCatalogRebuild(status, *jsonOut)
58 }
59 targets := make([]sessioncatalog.DirectoryTarget, 0, len(dirs))
60 for _, dir := range dirs {
61 targets = append(targets, sessioncatalog.DirectoryTarget{Path: dir, Scope: "global"})
62 }
63 targets = sessioncatalog.UniqueDirectoryTargets(targets)
64 status, err := sessioncatalog.Rebuild(context.Background(), sessioncatalog.DefaultPath(), targets)
65 if err != nil {
66 fmt.Fprintln(os.Stderr, "error:", err)
67 return 1
68 }
69 return printSessionCatalogRebuild(status, *jsonOut)
70 }
71
72 type sessionRecoveryReport struct {
73 Directories int `json:"directories"`
74 SourceSessions int `json:"sourceSessions"`
75 IndexedSessions int `json:"indexedSessions"`
76 UnindexedSessions int `json:"unindexedSessions"`
77 StaleDirectories int `json:"staleDirectories"`
78 Groups int `json:"groups"`
79 Branches int `json:"branches"`
80 AdoptedGroups int `json:"adoptedGroups"`
81 DivergedGroups int `json:"divergedGroups"`
82 CleanupEligible int `json:"cleanupEligible"`
83 MovedToTrash int `json:"movedToTrash"`
84 Busy int `json:"busy"`
85 SessionLogs int `json:"sessionLogs"` // schema-2 logs; their versions are heads, never copies
86 Heads int `json:"heads"`
87 CoveredHeads int `json:"coveredHeads"`
88 RetiredHeads int `json:"retiredHeads"`
89 Errors []string `json:"errors"`
90 DryRun bool `json:"dryRun"`
91 }
92
93 func sessionsRecoveryCommand(args []string, cleanup bool) int {
94 name := "sessions diagnose"
95 if cleanup {
96 name = "sessions cleanup"
97 }
98 fs := flag.NewFlagSet(name, flag.ContinueOnError)
99 var dirs stringListFlag
100 apply := fs.Bool("apply", false, "move safe covered recovery branches to recoverable trash")
101 jsonOut := fs.Bool("json", false, "print the recovery report as JSON")
102 fs.Var(&dirs, "dir", "session directory to inspect; repeat for multiple directories")
103 if code, ok := parseCommandFlags(fs, args); !ok {
104 return code
105 }
106 if fs.NArg() != 0 || (!cleanup && *apply) {
107 fmt.Fprintf(os.Stderr, "usage: reasonix %s [--dir PATH] [--json]", name)
108 if cleanup {
109 fmt.Fprint(os.Stderr, " [--apply]")
110 }
111 fmt.Fprintln(os.Stderr)
112 return 2
113 }
114 if len(dirs) == 0 {
115 for _, target := range defaultSessionCatalogTargets() {
116 dirs = append(dirs, target.Path)
117 }
118 }
119 targets := make([]sessioncatalog.DirectoryTarget, 0, len(dirs))
120 for _, dir := range dirs {
121 targets = append(targets, sessioncatalog.DirectoryTarget{Path: dir, Scope: "global"})
122 }
123 targets = sessioncatalog.UniqueDirectoryTargets(targets)
124 report := sessionRecoveryReport{Errors: []string{}, DryRun: !cleanup || !*apply}
125 persisted, persistedErr := sessioncatalog.Open(context.Background(), sessioncatalog.Options{
126 Path: sessioncatalog.DefaultPath(), DisableRepair: true,
127 })
128 if persistedErr != nil {
129 report.Errors = append(report.Errors, "open persisted session catalog: "+persistedErr.Error())
130 } else {
131 defer persisted.Close(context.Background())
132 }
133 for _, target := range targets {
134 dir := target.Path
135 report.Directories++
136 inspectSessionRecoveryDirectory(context.Background(), dir, persisted, &report)
137 }
138 printSessionRecoveryReport(report, *jsonOut, cleanup)
139 for _, message := range report.Errors {
140 fmt.Fprintln(os.Stderr, "warning:", message)
141 }
142 if len(report.Errors) > 0 {
143 return 1
144 }
145 return 0
146 }
147
148 func printSessionRecoveryReport(report sessionRecoveryReport, jsonOut, cleanup bool) {
149 if jsonOut {
150 enc := json.NewEncoder(os.Stdout)
151 enc.SetIndent("", " ")
152 _ = enc.Encode(report)
153 return
154 }
155 fmt.Printf("source sessions: %d; indexed sessions: %d; unindexed: %d; stale directories: %d\n", report.SourceSessions, report.IndexedSessions, report.UnindexedSessions, report.StaleDirectories)
156 fmt.Printf("recovery groups: %d (%d adopted, %d diverged)\n", report.Groups, report.AdoptedGroups, report.DivergedGroups)
157 fmt.Printf("recovery branches: %d; safe cleanup: %d; moved: %d; busy: %d\n", report.Branches, report.CleanupEligible, report.MovedToTrash, report.Busy)
158 fmt.Printf("session logs: %d; heads: %d (%d covered, %d retired)\n", report.SessionLogs, report.Heads, report.CoveredHeads, report.RetiredHeads)
159 if cleanup && report.CoveredHeads > 0 {
160 fmt.Println("covered heads live inside their session log; retire them from the app's session versions dialog")
161 }
162 if report.DryRun && cleanup {
163 fmt.Println("dry run; pass --apply to move safe branches to recoverable trash")
164 }
165 }
166
167 func inspectSessionRecoveryDirectory(ctx context.Context, dir string, persisted *sessioncatalog.Catalog, report *sessionRecoveryReport) {
168 updateSessionRecoveryCounts(ctx, dir, persisted, report)
169 catalog, err := sessioncatalog.Open(ctx, sessioncatalog.Options{InMemory: true, DisableRepair: true})
170 if err != nil {
171 report.Errors = append(report.Errors, err.Error())
172 return
173 }
174 defer catalog.Close(ctx)
175 if err := catalog.ReconcileDirectory(ctx, sessioncatalog.DirectoryTarget{Path: dir, Scope: "global"}); err != nil {
176 report.Errors = append(report.Errors, err.Error())
177 return
178 }
179 groups, err := catalog.ListRecoveryGroups(ctx, dir)
180 if err != nil {
181 report.Errors = append(report.Errors, err.Error())
182 return
183 }
184 for _, group := range groups {
185 inspectSessionRecoveryGroup(dir, group, report)
186 }
187 }
188
189 func updateSessionRecoveryCounts(ctx context.Context, dir string, persisted *sessioncatalog.Catalog, report *sessionRecoveryReport) {
190 source, err := agent.ListSessionOrder(dir)
191 if err != nil {
192 report.Errors = append(report.Errors, err.Error())
193 return
194 }
195 report.SourceSessions += len(source)
196 countSessionLogHeads(source, report)
197 if persisted == nil {
198 return
199 }
200 indexed, err := persisted.CountDirectorySessions(ctx, dir)
201 if err != nil {
202 report.Errors = append(report.Errors, err.Error())
203 return
204 }
205 report.IndexedSessions += int(indexed)
206 if indexed == int64(len(source)) {
207 return
208 }
209 report.StaleDirectories++
210 if len(source) > int(indexed) {
211 report.UnindexedSessions += len(source) - int(indexed)
212 }
213 }
214
215 func inspectSessionRecoveryGroup(dir string, group sessioncatalog.RecoveryGroup, report *sessionRecoveryReport) {
216 report.Groups++
217 report.Branches += len(group.Members)
218 canonical, diverged := recoveryGroupState(group.Members)
219 if canonical == "" {
220 if diverged {
221 report.DivergedGroups++
222 }
223 return
224 }
225 report.AdoptedGroups++
226 candidates := coveredRecoveryCandidates(group.Members, canonical)
227 report.CleanupEligible += len(candidates)
228 if report.DryRun || len(candidates) == 0 {
229 return
230 }
231 if err := agent.ReparentRecoveryCanonical(canonical, group.ID, dir); err != nil {
232 if errors.Is(err, agent.ErrSessionLeaseHeld) {
233 report.Busy += len(candidates)
234 } else {
235 report.Errors = append(report.Errors, err.Error())
236 }
237 return
238 }
239 for _, candidate := range candidates {
240 if err := agent.TrashRecoveryBranchCoveredBy(candidate, canonical, dir); err != nil {
241 if errors.Is(err, agent.ErrSessionLeaseHeld) {
242 report.Busy++
243 } else {
244 report.Errors = append(report.Errors, err.Error())
245 }
246 continue
247 }
248 report.MovedToTrash++
249 }
250 }
251
252 func recoveryGroupState(members []sessioncatalog.SessionRecord) (canonical string, diverged bool) {
253 for _, member := range members {
254 if member.RecoveryCanonical && (member.RecoveryRole == sessioncatalog.RecoveryRoleAdopted || member.RecoveryRole == sessioncatalog.RecoveryRolePreferred) {
255 canonical = member.Path
256 }
257 if member.RecoveryRole == sessioncatalog.RecoveryRoleDiverged {
258 diverged = true
259 }
260 }
261 return canonical, diverged
262 }
263
264 func coveredRecoveryCandidates(members []sessioncatalog.SessionRecord, canonical string) []string {
265 candidates := make([]string, 0, len(members))
266 for _, member := range members {
267 if member.Path != canonical && member.RecoveryRole == sessioncatalog.RecoveryRoleCoveredCopy {
268 candidates = append(candidates, member.Path)
269 }
270 }
271 return candidates
272 }
273
274 func printSessionCatalogRebuild(status sessioncatalog.Status, jsonOut bool) int {
275 if jsonOut {
276 enc := json.NewEncoder(os.Stdout)
277 enc.SetIndent("", " ")
278 if err := enc.Encode(status); err != nil {
279 fmt.Fprintln(os.Stderr, err)
280 return 1
281 }
282 return 0
283 }
284 fmt.Printf("rebuilt session catalog: %d sessions, revision %d\n", status.Indexed, status.Revision)
285 return 0
286 }
287
288 func defaultSessionCatalogTargets() []sessioncatalog.DirectoryTarget {
289 type project struct {
290 Root string `json:"root"`
291 }
292 type projectFile struct {
293 Projects []project `json:"projects"`
294 }
295 home := config.ReasonixHomeDir()
296 var saved projectFile
297 if data, err := os.ReadFile(filepath.Join(home, "desktop-projects.json")); err == nil {
298 _ = json.Unmarshal(data, &saved)
299 }
300 targets := make([]sessioncatalog.DirectoryTarget, 0, len(saved.Projects)+2)
301 add := func(target sessioncatalog.DirectoryTarget) {
302 targets = append(targets, target)
303 }
304 add(sessioncatalog.DirectoryTarget{Path: config.SessionDir(), Scope: "global"})
305 add(sessioncatalog.DirectoryTarget{
306 Path: config.ProjectSessionDir(filepath.Join(home, "global-workspace")),
307 Scope: "global",
308 })
309 for _, savedProject := range saved.Projects {
310 root := strings.TrimSpace(savedProject.Root)
311 if root == "" {
312 continue
313 }
314 add(sessioncatalog.DirectoryTarget{
315 Path: config.ProjectSessionDir(root), Scope: "project", WorkspaceRoot: root,
316 })
317 }
318 return sessioncatalog.UniqueDirectoryTargets(targets)
319 }
320
321 // countSessionLogHeads reports schema-2 logs by their heads. Such logs never
322 // join recovery groups, so cleanup has nothing to move for them.
323 func countSessionLogHeads(source []agent.SessionOrderInfo, report *sessionRecoveryReport) {
324 for _, info := range source {
325 heads, err := agent.ListSessionHeads(info.Path)
326 if err != nil || len(heads) == 0 {
327 continue
328 }
329 report.SessionLogs++
330 for _, head := range heads {
331 report.Heads++
332 switch {
333 case head.Retired:
334 report.RetiredHeads++
335 case head.Covered:
336 report.CoveredHeads++
337 }
338 }
339 }
340 }
341
341 lines GO