| 1 | package agent |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "crypto/sha256" |
| 6 | "fmt" |
| 7 | "log/slog" |
| 8 | "os" |
| 9 | "path/filepath" |
| 10 | |
| 11 | "reasonix/internal/provider" |
| 12 | ) |
| 13 | |
| 14 | // writeRecoveryBranchAtPath persists one bounded recovery lane. collision is |
| 15 | // true only when path already contains content this Session did not persist; |
| 16 | // callers rotate lanes instead of overwriting that independent transcript. |
| 17 | func (s *Session) writeRecoveryBranchAtPath( |
| 18 | path string, |
| 19 | opts RecoveryBranchOptions, |
| 20 | msgs []provider.Message, |
| 21 | digest [sha256.Size]byte, |
| 22 | version uint64, |
| 23 | rewriteVersion int, |
| 24 | preview string, |
| 25 | turns int, |
| 26 | digestText string, |
| 27 | recoveryDepth int, |
| 28 | shutdown bool, |
| 29 | ) (RecoveryBranchInfo, bool, error) { |
| 30 | unlockPath := lockSessionSavePath(path) |
| 31 | defer unlockPath() |
| 32 | unlockFile, err := lockSessionFile(path) |
| 33 | if err != nil { |
| 34 | return RecoveryBranchInfo{}, false, fmt.Errorf("lock recovery session file: %w", err) |
| 35 | } |
| 36 | defer unlockFile() |
| 37 | if loaded, loadErr := loadSessionUnlocked(path); loadErr == nil && loaded != nil { |
| 38 | existingDigest, digestErr := digestSessionMessages(loaded.Snapshot()) |
| 39 | if digestErr != nil { |
| 40 | return RecoveryBranchInfo{}, false, digestErr |
| 41 | } |
| 42 | if bytes.Equal(existingDigest[:], digest[:]) { |
| 43 | unlockMeta, lockErr := LockSessionMetaPath(path) |
| 44 | if lockErr != nil { |
| 45 | return RecoveryBranchInfo{}, false, lockErr |
| 46 | } |
| 47 | defer unlockMeta() |
| 48 | if _, err := copyValidContextProjection(opts.OriginalPath, path, msgs); err != nil { |
| 49 | slog.Warn("session: recovery branch did not inherit context projection", "path", path, "err", err) |
| 50 | } |
| 51 | meta, err := s.saveRecoveryBranchMetaLocked(path, opts, preview, turns, digestText, recoveryDepth, true) |
| 52 | if err != nil { |
| 53 | return RecoveryBranchInfo{}, false, err |
| 54 | } |
| 55 | if err := writeSessionEventIndex(path, msgs, digest, meta.Revision); err != nil { |
| 56 | slog.Warn("session: keeping recovery branch after event index write failure", "path", path, "err", err) |
| 57 | } |
| 58 | if err := refreshSessionDisplayIndex(path, msgs, digest, meta.Revision, -1); err != nil { |
| 59 | slog.Warn("session: keeping recovery branch after display index write failure", "path", path, "err", err) |
| 60 | } |
| 61 | s.markPersisted(path, digest, version, meta.Revision, rewriteVersion) |
| 62 | return RecoveryBranchInfo{Path: path, Digest: digestText, Existing: true, Meta: meta, Preview: preview, Turns: turns}, false, nil |
| 63 | } |
| 64 | state := s.persistState(path) |
| 65 | if !state.ok || !bytes.Equal(state.digest[:], existingDigest[:]) { |
| 66 | return RecoveryBranchInfo{}, true, nil |
| 67 | } |
| 68 | } else if loadErr != nil && !os.IsNotExist(loadErr) { |
| 69 | return RecoveryBranchInfo{}, false, loadErr |
| 70 | } |
| 71 | |
| 72 | if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { |
| 73 | return RecoveryBranchInfo{}, false, fmt.Errorf("create recovery session dir: %w", err) |
| 74 | } |
| 75 | unlockMeta, err := LockSessionMetaPath(path) |
| 76 | if err != nil { |
| 77 | return RecoveryBranchInfo{}, false, err |
| 78 | } |
| 79 | defer unlockMeta() |
| 80 | meta, err := s.prepareRecoveryBranchMetaLocked(path, opts, preview, turns, digestText, recoveryDepth, false) |
| 81 | if err != nil { |
| 82 | return RecoveryBranchInfo{}, false, err |
| 83 | } |
| 84 | probe, err := probeSessionEventLog(path) |
| 85 | if err != nil { |
| 86 | return RecoveryBranchInfo{}, false, err |
| 87 | } |
| 88 | if probe.native { |
| 89 | if err := writeRecoveryEventLog(path, msgs, digest, meta.Revision, shutdown); err != nil { |
| 90 | return RecoveryBranchInfo{}, false, err |
| 91 | } |
| 92 | } |
| 93 | if err := writeSessionMessages(path, msgs); err != nil { |
| 94 | return RecoveryBranchInfo{}, false, err |
| 95 | } |
| 96 | if _, err := copyValidContextProjection(opts.OriginalPath, path, msgs); err != nil { |
| 97 | slog.Warn("session: recovery branch did not inherit context projection", "path", path, "err", err) |
| 98 | } |
| 99 | if err := saveBranchMeta(path, meta, true); err != nil { |
| 100 | return RecoveryBranchInfo{}, false, err |
| 101 | } |
| 102 | if err := writeSessionEventIndex(path, msgs, digest, meta.Revision); err != nil { |
| 103 | slog.Warn("session: keeping recovery branch after event index write failure", "path", path, "err", err) |
| 104 | } |
| 105 | if err := refreshSessionDisplayIndex(path, msgs, digest, meta.Revision, -1); err != nil { |
| 106 | slog.Warn("session: keeping recovery branch after display index write failure", "path", path, "err", err) |
| 107 | } |
| 108 | s.markPersisted(path, digest, version, meta.Revision, rewriteVersion) |
| 109 | return RecoveryBranchInfo{Path: path, Digest: digestText, Meta: meta, Preview: preview, Turns: turns}, false, nil |
| 110 | } |
| 111 | |
| 112 | // CopyValidContextProjection copies a parent's projection to this session's |
| 113 | // path only when the covered canonical prefix still matches. Forks use it to |
| 114 | // preserve compacted context without borrowing stale parent history. |
| 115 | func (s *Session) CopyValidContextProjection(originalPath, targetPath string) (bool, error) { |
| 116 | if s == nil { |
| 117 | return false, nil |
| 118 | } |
| 119 | return copyValidContextProjection(originalPath, targetPath, s.Snapshot()) |
| 120 | } |
| 121 | |
| 122 | func copyValidContextProjection(originalPath, targetPath string, msgs []provider.Message) (bool, error) { |
| 123 | if _, ok, err := LoadCompactionState(targetPath); err != nil { |
| 124 | return false, err |
| 125 | } else if ok { |
| 126 | return false, nil |
| 127 | } |
| 128 | st, ok, err := LoadCompactionState(originalPath) |
| 129 | if err != nil { |
| 130 | return false, err |
| 131 | } |
| 132 | if !ok { |
| 133 | return false, nil |
| 134 | } |
| 135 | migratePromotedCoveredPrefixHash(&st, msgs) |
| 136 | n := st.Projection.CoveredCount |
| 137 | if len(st.Projection.Messages) == 0 || n <= 0 || n > len(msgs) || |
| 138 | st.Projection.CoveredPrefixHash == "" || |
| 139 | coveredPrefixHash(msgs, n) != st.Projection.CoveredPrefixHash { |
| 140 | return false, nil |
| 141 | } |
| 142 | if err := SaveCompactionState(targetPath, st); err != nil { |
| 143 | return false, err |
| 144 | } |
| 145 | return true, nil |
| 146 | } |
| 147 | |
| 148 | // LoadValidContextProjectionForMigration reconstructs the exact model-visible |
| 149 | // view from a frozen legacy sidecar when it still authenticates the supplied |
| 150 | // canonical transcript. Migration deliberately validates content independently |
| 151 | // of the old model/workspace lineage because the canonical target persists that |
| 152 | // selection separately. |
| 153 | func LoadValidContextProjectionForMigration(sessionPath string, canonical []provider.Message) ([]provider.Message, bool, error) { |
| 154 | st, ok, err := LoadCompactionState(sessionPath) |
| 155 | if err != nil || !ok { |
| 156 | return nil, false, err |
| 157 | } |
| 158 | migratePromotedCoveredPrefixHash(&st, canonical) |
| 159 | if !projectionContentValid(st, canonical) { |
| 160 | return nil, false, nil |
| 161 | } |
| 162 | visible := modelVisibleFromProjection(st.Projection, canonical) |
| 163 | if len(visible) == 0 { |
| 164 | return nil, false, nil |
| 165 | } |
| 166 | return visible, true, nil |
| 167 | } |
| 168 | |
| 169 | // healEmptyCheckpointFromWAL rebuilds a missing or 0-byte checkpoint from its |
| 170 | // own valid event log. Healthy checkpoints return before probing the WAL. |
| 171 | func healEmptyCheckpointFromWAL(path string) error { |
| 172 | info, statErr := os.Stat(path) |
| 173 | needHeal := statErr != nil && os.IsNotExist(statErr) |
| 174 | if statErr == nil { |
| 175 | needHeal = info.Size() == 0 |
| 176 | } |
| 177 | if !needHeal { |
| 178 | return nil |
| 179 | } |
| 180 | probe, err := probeSessionEventLog(path) |
| 181 | if err != nil { |
| 182 | return err |
| 183 | } |
| 184 | if !probe.native || probe.size <= 0 || probe.futureSchema { |
| 185 | return nil |
| 186 | } |
| 187 | msgs, fromEvents, damaged, err := loadSessionMessages(path) |
| 188 | if err != nil || !fromEvents || damaged || len(msgs) == 0 { |
| 189 | return nil |
| 190 | } |
| 191 | if err := writeSessionMessages(path, msgs); err != nil { |
| 192 | return fmt.Errorf("rebuild checkpoint from WAL: %w", err) |
| 193 | } |
| 194 | return nil |
| 195 | } |
| 196 |