返回 DeepSeek-Reasonix
lineage.go
根目录 / internal / sessioncatalog / lineage.go
1 package sessioncatalog
2
3 import (
4 "path/filepath"
5 "sort"
6 "strings"
7
8 "reasonix/internal/agent"
9 )
10
11 // classifyRecoveryLineage assigns recovery_group_id, recovery_role, and
12 // recovery_canonical from real content ancestry. File names alone never decide
13 // ownership; covered copies are those whose messages are a prefix of a still-
14 // present parent/ancestor, adopted is a unique leaf covering the group, and
15 // diverged marks multiple non-covering leaves under one group.
16 func classifyRecoveryLineage(record SessionRecord) SessionRecord {
17 if !record.Recovered {
18 if record.RecoveryRole == "" {
19 record.RecoveryRole = RecoveryRoleNormal
20 }
21 return record
22 }
23 parentPath := recoveryParentPath(record)
24 groupID := firstNonEmpty(record.ParentID, agent.BranchID(record.Path))
25 record.RecoveryGroupID = groupID
26
27 if record.RecoveryCopy || (parentPath != "" && agent.RecoveryBranchCoveredByParent(record.Path, filepath.Dir(record.Path))) {
28 record.RecoveryCopy = true
29 record.RecoveryRole = RecoveryRoleCoveredCopy
30 record.RecoveryCanonical = false
31 return record
32 }
33 // Default for non-covered recovery leaves: diverged until a group pass
34 // promotes a unique covering leaf to adopted/canonical.
35 record.RecoveryRole = RecoveryRoleDiverged
36 record.RecoveryCanonical = false
37 return record
38 }
39
40 type recoveryContentResult struct {
41 snapshot agent.SessionContentSnapshot
42 ok bool
43 }
44
45 type recoveryContentCache struct {
46 entries map[string]recoveryContentResult
47 onLoad func(string)
48 strict bool
49 }
50
51 func newRecoveryContentCache(onLoad func(string)) *recoveryContentCache {
52 return &recoveryContentCache{entries: map[string]recoveryContentResult{}, onLoad: onLoad}
53 }
54
55 func newStrictRecoveryContentCache(onLoad func(string)) *recoveryContentCache {
56 return &recoveryContentCache{entries: map[string]recoveryContentResult{}, onLoad: onLoad, strict: true}
57 }
58
59 func listSessionOrderWithContent(dir string, content *recoveryContentCache) ([]agent.SessionOrderInfo, error) {
60 return agent.ListSessionOrderWithRecoveryPreferenceResolver(dir, func(path string, meta agent.BranchMeta) bool {
61 digest := strings.TrimSpace(meta.RecoveryPreferredDigest)
62 if !meta.RecoveryPreferred || digest == "" {
63 return false
64 }
65 _, ok := content.load(path, digest)
66 return ok
67 })
68 }
69
70 func (c *recoveryContentCache) load(path, digest string) (agent.SessionContentSnapshot, bool) {
71 key := PathIdentityKey(path)
72 result, loaded := c.entries[key]
73 if !loaded {
74 if c.onLoad != nil {
75 c.onLoad(path)
76 }
77 result.snapshot, result.ok = agent.LoadSessionContentSnapshot(path)
78 c.entries[key] = result
79 }
80 if !result.ok || strings.TrimSpace(digest) != "" && !result.snapshot.MatchesDigest(digest) {
81 return agent.SessionContentSnapshot{}, false
82 }
83 return result.snapshot, true
84 }
85
86 func (c *recoveryContentCache) loadRecord(record SessionRecord) (agent.SessionContentSnapshot, bool) {
87 if c.strict && record.Recovered && strings.TrimSpace(record.RecoveryDigest) == "" {
88 return agent.SessionContentSnapshot{}, false
89 }
90 return c.load(record.Path, record.RecoveryDigest)
91 }
92
93 func classifyRecoveryLineageWithContent(record SessionRecord, content *recoveryContentCache) SessionRecord {
94 if !record.Recovered {
95 if record.RecoveryRole == "" {
96 record.RecoveryRole = RecoveryRoleNormal
97 }
98 return record
99 }
100 record.RecoveryCopy = false
101 record.RecoveryGroupID = firstNonEmpty(record.ParentID, agent.BranchID(record.Path))
102 record.RecoveryRole = RecoveryRoleDiverged
103 record.RecoveryCanonical = false
104 parentPath := recoveryParentPath(record)
105 if parentPath == "" {
106 return record
107 }
108 branch, branchOK := content.loadRecord(record)
109 parent, parentOK := content.load(parentPath, "")
110 if branchOK && parentOK && parent.Covers(branch) {
111 record.RecoveryCopy = true
112 record.RecoveryRole = RecoveryRoleCoveredCopy
113 }
114 return record
115 }
116
117 // promoteCanonicalLeaves marks unique non-covered leaves that cover every
118 // ancestor in their group as adopted/canonical. Multiple non-covering leaves
119 // stay diverged. open/running/pinned/leased decisions are left to callers.
120 func promoteCanonicalLeaves(records []SessionRecord) []SessionRecord {
121 return promoteCanonicalLeavesWithContent(records, newRecoveryContentCache(nil))
122 }
123
124 func promoteCanonicalLeavesWithContent(records []SessionRecord, cache *recoveryContentCache) []SessionRecord {
125 byGroup, groupRoot := recoveryLineageGroups(records)
126 for groupID, idxs := range byGroup {
127 for _, i := range idxs {
128 records[i].RecoveryRole = RecoveryRoleDiverged
129 records[i].RecoveryCanonical = false
130 }
131 rootIndex, hasRoot := groupRoot[groupID]
132 preferred := -1
133 preferenceIndexes := append([]int{}, idxs...)
134 if hasRoot {
135 preferenceIndexes = append(preferenceIndexes, rootIndex)
136 }
137 for _, index := range preferenceIndexes {
138 if records[index].RecoveryPreferred {
139 if preferred >= 0 {
140 preferred = -2 // multiple stale preferences fail closed
141 break
142 }
143 preferred = index
144 }
145 }
146 if preferred >= 0 {
147 records[preferred].RecoveryRole = RecoveryRolePreferred
148 records[preferred].RecoveryCanonical = true
149 continue
150 }
151 contentIdxs := append([]int{}, idxs...)
152 if hasRoot {
153 contentIdxs = append(contentIdxs, rootIndex)
154 }
155 content := loadRecoveryGroupContent(records, contentIdxs, cache)
156 candidate, ok := uniqueLongestRecovery(records, idxs, content)
157 if !ok {
158 // No unique covering leaf: still pick one stable representative so
159 // ordinary visibility never expands into a wall of forks.
160 if len(idxs) > 0 {
161 stable := pickPreferredRecovery(indexesToRecords(records, idxs))
162 for _, i := range idxs {
163 if records[i].Path == stable.Path {
164 records[i].RecoveryCanonical = true
165 break
166 }
167 }
168 }
169 continue
170 }
171 if hasRoot && !recoveryCandidateCovers(candidate, rootIndex, idxs, content) {
172 records[candidate].RecoveryCanonical = true
173 continue
174 }
175 if !hasRoot {
176 // Root missing: the unique covering leaf is still the ordinary
177 // representative, but without a parent it cannot prove adoption.
178 records[candidate].RecoveryCanonical = true
179 for _, i := range idxs {
180 if i == candidate {
181 continue
182 }
183 if member, ok := content[i]; ok {
184 if candidateContent, ok := content[candidate]; ok && candidateContent.Covers(member) {
185 records[i].RecoveryCopy = true
186 records[i].RecoveryRole = RecoveryRoleCoveredCopy
187 }
188 }
189 }
190 continue
191 }
192 records[candidate].RecoveryRole = RecoveryRoleAdopted
193 records[candidate].RecoveryCanonical = true
194 // Once one leaf contains the entire group, every other recovery
195 // member is an ancestor/equivalent copy covered by that canonical.
196 // Marking them covered is what lets History and explicit cleanup
197 // converge an old chain instead of leaving hundreds of non-canonical
198 // "diverged" rows that preserve no unique content.
199 for _, i := range idxs {
200 if i == candidate {
201 continue
202 }
203 records[i].RecoveryCopy = true
204 records[i].RecoveryRole = RecoveryRoleCoveredCopy
205 }
206 }
207 return projectLogicalSessions(records)
208 }
209
210 func indexesToRecords(records []SessionRecord, idxs []int) []SessionRecord {
211 out := make([]SessionRecord, 0, len(idxs))
212 for _, i := range idxs {
213 out = append(out, records[i])
214 }
215 return out
216 }
217
218 // projectLogicalSessions re-anchors recovered physical files onto one logical
219 // topic and marks the single ordinary-list representative. Catalog projection
220 // only — authoritative JSONL/meta topic_id values are never rewritten.
221 func projectLogicalSessions(records []SessionRecord) []SessionRecord {
222 byID := make(map[string]int, len(records))
223 for i := range records {
224 byID[agent.BranchID(records[i].Path)] = i
225 if !records[i].Recovered {
226 if records[i].LogicalTopicID == "" {
227 records[i].LogicalTopicID = records[i].TopicID
228 }
229 continue
230 }
231 // Filename fallback when meta ParentID is empty.
232 if strings.TrimSpace(records[i].ParentID) == "" {
233 if parent, ok := agent.RecoveryFilenameParentID(records[i].Path); ok {
234 records[i].ParentID = parent
235 if records[i].RecoveryGroupID == "" {
236 records[i].RecoveryGroupID = parent
237 }
238 }
239 }
240 if records[i].RecoveryGroupID == "" {
241 records[i].RecoveryGroupID = firstNonEmpty(records[i].ParentID, agent.BranchID(records[i].Path))
242 }
243 }
244 // Walk parent chains so intermediate recovery files also share the root id.
245 for i := range records {
246 if !records[i].Recovered {
247 continue
248 }
249 groupID := strings.TrimSpace(records[i].RecoveryGroupID)
250 seen := map[string]struct{}{}
251 parentID := strings.TrimSpace(records[i].ParentID)
252 for parentID != "" {
253 if _, loop := seen[parentID]; loop {
254 break
255 }
256 seen[parentID] = struct{}{}
257 parentIndex, ok := byID[parentID]
258 if !ok {
259 groupID = firstNonEmpty(groupID, parentID)
260 break
261 }
262 parent := records[parentIndex]
263 if !parent.Recovered {
264 groupID = parentID
265 break
266 }
267 parentID = strings.TrimSpace(parent.ParentID)
268 if parent.RecoveryGroupID != "" {
269 groupID = parent.RecoveryGroupID
270 }
271 }
272 if groupID != "" {
273 records[i].RecoveryGroupID = groupID
274 }
275 }
276
277 type groupAnchor struct {
278 topicID string
279 topicTitle string
280 createdAt int64
281 rootIndex int
282 }
283 anchors := map[string]*groupAnchor{}
284 for i, rec := range records {
285 if !rec.Recovered {
286 id := agent.BranchID(rec.Path)
287 anchors[id] = &groupAnchor{
288 topicID: rec.TopicID, topicTitle: rec.TopicTitle,
289 createdAt: rec.CreatedAt, rootIndex: i,
290 }
291 }
292 }
293 // Fill missing logical topics from recovered members (root topic empty or
294 // root missing). Prefer earliest non-empty member topic_id.
295 for _, rec := range records {
296 if !rec.Recovered {
297 continue
298 }
299 groupID := strings.TrimSpace(rec.RecoveryGroupID)
300 if groupID == "" {
301 continue
302 }
303 anchor := anchors[groupID]
304 if anchor == nil {
305 anchor = &groupAnchor{rootIndex: -1, createdAt: rec.CreatedAt}
306 anchors[groupID] = anchor
307 }
308 topic := strings.TrimSpace(rec.TopicID)
309 if topic == "" {
310 continue
311 }
312 if strings.TrimSpace(anchor.topicID) == "" ||
313 (rec.CreatedAt > 0 && (anchor.createdAt == 0 || rec.CreatedAt < anchor.createdAt) &&
314 (anchor.rootIndex < 0 || strings.TrimSpace(records[anchor.rootIndex].TopicID) == "")) {
315 // Keep a non-empty root topic when present; only override empty roots.
316 if anchor.rootIndex >= 0 && strings.TrimSpace(records[anchor.rootIndex].TopicID) != "" {
317 continue
318 }
319 anchor.topicID = topic
320 anchor.topicTitle = rec.TopicTitle
321 if rec.CreatedAt > 0 {
322 anchor.createdAt = rec.CreatedAt
323 }
324 }
325 }
326 for groupID, anchor := range anchors {
327 if strings.TrimSpace(anchor.topicID) == "" {
328 // Stable internal topic so rootless recovery storms still collapse.
329 anchor.topicID = "recovery:" + groupID
330 if anchor.topicTitle == "" {
331 anchor.topicTitle = groupID
332 }
333 }
334 }
335 for i := range records {
336 groupID := ""
337 if records[i].Recovered {
338 groupID = strings.TrimSpace(records[i].RecoveryGroupID)
339 } else {
340 groupID = agent.BranchID(records[i].Path)
341 }
342 anchor := anchors[groupID]
343 if anchor == nil {
344 records[i].LogicalTopicID = firstNonEmpty(records[i].TopicID, "recovery:"+agent.BranchID(records[i].Path))
345 continue
346 }
347 records[i].LogicalTopicID = anchor.topicID
348 // Catalog ListTopics keys off TopicID. Re-anchor recovered physical
349 // rows (and roots that only inherited a member topic) onto one logical
350 // topic so pagination never materializes recovery replica walls.
351 if anchor.topicID != "" && (records[i].Recovered || strings.TrimSpace(records[i].TopicID) == "") {
352 records[i].TopicID = anchor.topicID
353 }
354 if anchor.topicTitle != "" && (records[i].Recovered || strings.TrimSpace(records[i].TopicTitle) == "") {
355 records[i].TopicTitle = anchor.topicTitle
356 }
357 }
358
359 preferred := PreferredOrdinarySessionPaths(records)
360 for i := range records {
361 _, records[i].OrdinaryVisible = preferred[strings.TrimSpace(records[i].Path)]
362 // When a group has a normal root, the root is the ordinary row even if
363 // an adopted leaf is the open target (canonical path).
364 if !records[i].Recovered {
365 records[i].OrdinaryVisible = true
366 }
367 }
368 // Exactly one ordinary-visible recovered leaf when the root is missing.
369 byGroup := map[string][]int{}
370 rootPresent := map[string]bool{}
371 for i, rec := range records {
372 if !rec.Recovered {
373 rootPresent[agent.BranchID(rec.Path)] = true
374 continue
375 }
376 if rec.RecoveryCopy || rec.RecoveryRole == RecoveryRoleCoveredCopy {
377 records[i].OrdinaryVisible = false
378 continue
379 }
380 byGroup[rec.RecoveryGroupID] = append(byGroup[rec.RecoveryGroupID], i)
381 }
382 for groupID, idxs := range byGroup {
383 if rootPresent[groupID] {
384 for _, i := range idxs {
385 records[i].OrdinaryVisible = false
386 }
387 continue
388 }
389 best := -1
390 for _, i := range idxs {
391 if records[i].RecoveryCanonical {
392 best = i
393 break
394 }
395 }
396 if best < 0 {
397 bestIdx := pickPreferredRecovery(indexesToRecords(records, idxs))
398 for _, i := range idxs {
399 if records[i].Path == bestIdx.Path {
400 best = i
401 break
402 }
403 }
404 }
405 for _, i := range idxs {
406 records[i].OrdinaryVisible = i == best
407 }
408 }
409 return records
410 }
411
412 func recoveryLineageGroups(records []SessionRecord) (map[string][]int, map[string]int) {
413 byID := make(map[string]int, len(records))
414 for i := range records {
415 byID[agent.BranchID(records[i].Path)] = i
416 }
417 byGroup := map[string][]int{}
418 groupRoot := map[string]int{}
419 for i := range records {
420 rec := records[i]
421 if !rec.Recovered {
422 continue
423 }
424 parentID := strings.TrimSpace(rec.ParentID)
425 seen := map[string]struct{}{}
426 for parentID != "" {
427 if _, loop := seen[parentID]; loop {
428 parentID = ""
429 break
430 }
431 seen[parentID] = struct{}{}
432 parentIndex, ok := byID[parentID]
433 if !ok {
434 parentID = ""
435 break
436 }
437 parent := records[parentIndex]
438 if !parent.Recovered {
439 groupRoot[parentID] = parentIndex
440 break
441 }
442 parentID = strings.TrimSpace(parent.ParentID)
443 }
444 if parentID != "" {
445 records[i].RecoveryGroupID = parentID
446 }
447 }
448 for i, rec := range records {
449 if rec.RecoveryGroupID == "" || rec.RecoveryRole == RecoveryRoleCoveredCopy || rec.RecoveryRole == RecoveryRoleNormal {
450 continue
451 }
452 byGroup[rec.RecoveryGroupID] = append(byGroup[rec.RecoveryGroupID], i)
453 }
454 return byGroup, groupRoot
455 }
456
457 func loadRecoveryGroupContent(records []SessionRecord, idxs []int, cache *recoveryContentCache) map[int]agent.SessionContentSnapshot {
458 content := make(map[int]agent.SessionContentSnapshot, len(idxs))
459 for _, index := range idxs {
460 if _, loaded := content[index]; loaded {
461 continue
462 }
463 if snapshot, ok := cache.loadRecord(records[index]); ok {
464 content[index] = snapshot
465 }
466 }
467 return content
468 }
469
470 func uniqueLongestRecovery(records []SessionRecord, idxs []int, content map[int]agent.SessionContentSnapshot) (int, bool) {
471 if len(idxs) == 0 {
472 return 0, false
473 }
474 // Turns/preview metadata is repaired asynchronously and must not decide
475 // lineage. Doing so made the first scan permanently label every legacy fork
476 // diverged when its sidecar still had turns_state=unknown. Instead, find the
477 // leaves that actually cover every recovered member. Equivalent leaves are
478 // safe to collapse and use a deterministic activity/path tie-breaker.
479 maxLen := -1
480 for _, index := range idxs {
481 if snapshot, ok := content[index]; ok && snapshot.Len() > maxLen {
482 maxLen = snapshot.Len()
483 }
484 }
485 if maxLen < 0 {
486 return 0, false
487 }
488 candidates := make([]int, 0, 1)
489 for _, candidate := range idxs {
490 candidateContent, ok := content[candidate]
491 if !ok || candidateContent.Len() != maxLen {
492 continue
493 }
494 coversAll := true
495 for _, other := range idxs {
496 otherContent, ok := content[other]
497 if !ok || (candidate != other && !candidateContent.Covers(otherContent)) {
498 coversAll = false
499 break
500 }
501 }
502 if coversAll {
503 candidates = append(candidates, candidate)
504 }
505 }
506 if len(candidates) == 0 {
507 return 0, false
508 }
509 sort.SliceStable(candidates, func(i, j int) bool {
510 a, b := records[candidates[i]], records[candidates[j]]
511 if a.LastActivityAt != b.LastActivityAt {
512 return a.LastActivityAt > b.LastActivityAt
513 }
514 return a.Path < b.Path
515 })
516 return candidates[0], true
517 }
518
519 func recoveryCandidateCovers(candidate, root int, idxs []int, content map[int]agent.SessionContentSnapshot) bool {
520 candidateContent, ok := content[candidate]
521 if !ok {
522 return false
523 }
524 rootContent, ok := content[root]
525 if !ok || !candidateContent.Covers(rootContent) {
526 return false
527 }
528 for _, i := range idxs {
529 memberContent, ok := content[i]
530 if !ok || (i != candidate && !candidateContent.Covers(memberContent)) {
531 return false
532 }
533 }
534 return true
535 }
536
537 func recoveryParentPath(record SessionRecord) string {
538 parentID := strings.TrimSpace(record.ParentID)
539 if parentID == "" {
540 return ""
541 }
542 dir := filepath.Dir(record.Path)
543 if dir == "" || dir == "." {
544 return ""
545 }
546 candidate := filepath.Join(dir, parentID+".jsonl")
547 return candidate
548 }
549
550 func firstNonEmpty(values ...string) string {
551 for _, v := range values {
552 if strings.TrimSpace(v) != "" {
553 return strings.TrimSpace(v)
554 }
555 }
556 return ""
557 }
558
559 // CanonicalSessionPathForTopic returns the path that open/restore should bind
560 // when a unique adopted/canonical leaf exists for the topic's sessions.
561 // Empty means keep the caller's path.
562 func CanonicalSessionPathForTopic(sessions []SessionRecord, current string) string {
563 var canonical string
564 canonicalRole := ""
565 for _, s := range sessions {
566 if s.RecoveryCanonical && (s.RecoveryRole == RecoveryRoleAdopted || s.RecoveryRole == RecoveryRolePreferred) {
567 if canonical != "" && canonical != s.Path {
568 // Ambiguous: do not retarget.
569 return ""
570 }
571 canonical = s.Path
572 canonicalRole = s.RecoveryRole
573 }
574 }
575 if canonicalRole == RecoveryRolePreferred && current != "" {
576 for _, session := range sessions {
577 if session.Path == current && session.Recovered {
578 // An explicit click on another recovery leaf is inspection, not a
579 // request to follow the default choice.
580 return ""
581 }
582 }
583 }
584 if canonical == "" || canonical == current {
585 return ""
586 }
587 return canonical
588 }
589
590 // OrdinaryContinuePath returns a safe continuation when current is the ordinary
591 // parent. Explicit recovery paths stay put so History can inspect them.
592 func OrdinaryContinuePath(sessions []SessionRecord, current string) string {
593 canonical := CanonicalSessionPathForTopic(sessions, current)
594 if canonical == "" {
595 canonical = uniqueLinearRecoveryLeaf(sessions)
596 }
597 current = strings.TrimSpace(current)
598 if canonical == "" || current == canonical {
599 return ""
600 }
601 if current == "" {
602 return canonical
603 }
604 for _, session := range sessions {
605 if session.Path != current {
606 continue
607 }
608 if session.Recovered {
609 return ""
610 }
611 return canonical
612 }
613 return canonical
614 }
615
616 // uniqueLinearRecoveryLeaf selects the only monotonic descendant in a complete
617 // parent chain. It is an open target only; content coverage still exclusively
618 // controls adoption and cleanup.
619 func uniqueLinearRecoveryLeaf(sessions []SessionRecord) string {
620 if len(sessions) < 2 {
621 return ""
622 }
623 byID := make(map[string]int, len(sessions))
624 root := -1
625 recovered := 0
626 for i, session := range sessions {
627 id := agent.BranchID(session.Path)
628 if id == "" {
629 return ""
630 }
631 if _, duplicate := byID[id]; duplicate {
632 return ""
633 }
634 byID[id] = i
635 if session.Recovered {
636 recovered++
637 continue
638 }
639 if root >= 0 {
640 return ""
641 }
642 root = i
643 }
644 if root < 0 || recovered == 0 {
645 return ""
646 }
647 children := make(map[string]int, recovered)
648 for i, session := range sessions {
649 if !session.Recovered {
650 continue
651 }
652 parentID := strings.TrimSpace(session.ParentID)
653 parent, ok := byID[parentID]
654 if parentID == "" || !ok {
655 return ""
656 }
657 if _, forked := children[parentID]; forked {
658 return ""
659 }
660 if session.TurnsState == TurnsValid && sessions[parent].TurnsState == TurnsValid && session.Turns < sessions[parent].Turns {
661 return ""
662 }
663 if session.LastActivityAt > 0 && sessions[parent].LastActivityAt > 0 && session.LastActivityAt < sessions[parent].LastActivityAt {
664 return ""
665 }
666 children[parentID] = i
667 }
668 seen := make(map[int]struct{}, recovered+1)
669 leaf := root
670 for {
671 if _, duplicate := seen[leaf]; duplicate {
672 return ""
673 }
674 seen[leaf] = struct{}{}
675 next, ok := children[agent.BranchID(sessions[leaf].Path)]
676 if !ok {
677 break
678 }
679 leaf = next
680 }
681 if len(seen) != len(sessions) || !sessions[leaf].Recovered {
682 return ""
683 }
684 return strings.TrimSpace(sessions[leaf].Path)
685 }
686
686 lines GO