| 1 | package agent |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "crypto/rand" |
| 7 | "crypto/sha256" |
| 8 | "encoding/hex" |
| 9 | "encoding/json" |
| 10 | "errors" |
| 11 | "fmt" |
| 12 | "hash" |
| 13 | "log/slog" |
| 14 | "math" |
| 15 | "os" |
| 16 | "path/filepath" |
| 17 | "sort" |
| 18 | "strings" |
| 19 | "sync" |
| 20 | "time" |
| 21 | "unicode/utf8" |
| 22 | |
| 23 | "reasonix/internal/fileutil" |
| 24 | fileencoding "reasonix/internal/fileutil/encoding" |
| 25 | "reasonix/internal/provider" |
| 26 | "reasonix/internal/store" |
| 27 | ) |
| 28 | |
| 29 | const ( |
| 30 | cleanupPendingExt = ".cleanup-pending.json" |
| 31 | maxRecoveryParentStemBytes = 80 |
| 32 | sessionLockSidecarSuffix = ".jsonl.lock" |
| 33 | sessionLeaseLockSidecarSuffix = ".jsonl.lease.lock" |
| 34 | sessionLeaseInfoSidecarSuffix = ".jsonl.lease.json" |
| 35 | guardianSidecarSuffix = ".guardian.jsonl" |
| 36 | // nameMaxBytes is the single-component filename limit shared by the |
| 37 | // filesystems Reasonix targets (APFS, ext4, NTFS all cap at 255). |
| 38 | nameMaxBytes = 255 |
| 39 | // maxSessionBasenameBytes bounds transcript basenames that reconciliation |
| 40 | // leaves in place. Sidecars append up to ~16 bytes to the transcript name |
| 41 | // or its stem (".lease.lock", ".cleanup-pending.json", ".guardian.jsonl"), |
| 42 | // so 224 keeps every sidecar comfortably under nameMaxBytes with headroom |
| 43 | // for future suffixes. Names past this bound come from the pre-bounded |
| 44 | // recovery cascade and get renamed by reconcileOverlongSessionFilenames. |
| 45 | maxSessionBasenameBytes = 224 |
| 46 | ) |
| 47 | |
| 48 | var ( |
| 49 | sessionSaveLocks sync.Map |
| 50 | // sessionFileLockWait bounds cross-process save-lock acquisition. Session |
| 51 | // leases normally prevent competing writers, but CLI/legacy writers and a |
| 52 | // stalled process can still hold the compatibility .lock file. Navigation |
| 53 | // and desktop shutdown snapshot synchronously; waiting forever here wedges |
| 54 | // the UI and keeps the session lease (and WebView) alive indefinitely. |
| 55 | // Package vars let focused tests shorten the wait without slowing the suite. |
| 56 | sessionFileLockWait = 5 * time.Second |
| 57 | sessionFileLockPollInterval = 25 * time.Millisecond |
| 58 | sessionMetaLockWait = 5 * time.Second |
| 59 | ErrSessionSnapshotConflict = errors.New("session snapshot conflicts with newer transcript") |
| 60 | // ErrSessionExternallyRemoved means a live Session still has a verified |
| 61 | // baseline for path, but every authoritative transcript artifact disappeared. |
| 62 | // Treating it as a first save would silently recreate a file the user or an |
| 63 | // external cleanup tool deliberately removed. |
| 64 | ErrSessionExternallyRemoved = errors.New("session was removed while still open") |
| 65 | ErrSessionRecoveryNotNeeded = errors.New("session recovery not needed") |
| 66 | // ErrSessionFileLockHeld reports that another process kept the |
| 67 | // compatibility save lock for the full bounded acquisition window. Callers |
| 68 | // that are about to terminate can use this sentinel to persist a recovery |
| 69 | // branch without waiting on the same stalled file again. |
| 70 | ErrSessionFileLockHeld = errors.New("session file lock held") |
| 71 | // ErrSessionRecoveryDepthExceeded is retained for older callers. New |
| 72 | // recovery writes update one stable branch and no longer return it. |
| 73 | ErrSessionRecoveryDepthExceeded = errors.New("session recovery chain depth exceeded") |
| 74 | sessionWriterID = newSessionWriterID() |
| 75 | ) |
| 76 | |
| 77 | // SessionRecoveryMaxDepth is the historical nested-fork cap. New writes stamp |
| 78 | // RecoveryDepth=1 and update one stable path instead of deepening a chain. |
| 79 | const SessionRecoveryMaxDepth = 3 |
| 80 | |
| 81 | type sessionPersistState struct { |
| 82 | projectionPending bool |
| 83 | path string |
| 84 | digest [sha256.Size]byte |
| 85 | version uint64 |
| 86 | revision int64 |
| 87 | // revisionKnown marks revision as a real ledger value. It is false when |
| 88 | // the baseline was established while the meta sidecar was unreadable |
| 89 | // (torn or corrupt): the session must still open, but revision 0 must not |
| 90 | // pose as a baseline or every honest on-disk revision would read as a |
| 91 | // stale-runtime conflict. CAS checks fall back to digest+version until a |
| 92 | // successful save re-learns the revision. |
| 93 | revisionKnown bool |
| 94 | // saveVerified marks a baseline established by a completed save in this |
| 95 | // process, whose write path verified transcript and ledger agree. A |
| 96 | // baseline adopted at load time pairs the disk transcript with whatever |
| 97 | // the meta sidecar said — which can lag the transcript after an |
| 98 | // interrupted save — so only save-verified baselines may arm the |
| 99 | // snapshot no-op fast path; the first save after a load must run in full |
| 100 | // and heal a stale ledger. |
| 101 | saveVerified bool |
| 102 | ok bool |
| 103 | } |
| 104 | |
| 105 | type sessionSaveMode int |
| 106 | |
| 107 | const ( |
| 108 | sessionSaveSnapshot sessionSaveMode = iota |
| 109 | sessionSaveRewrite |
| 110 | sessionSaveRewriteCompact |
| 111 | sessionSaveToolCheckpoint |
| 112 | ) |
| 113 | |
| 114 | type snapshotWriteDecision struct { |
| 115 | revision int64 |
| 116 | reservedRevision int64 |
| 117 | upToDate bool |
| 118 | appendFrom int |
| 119 | appendOnly bool |
| 120 | // repairLog is set when the on-disk event log was damaged (torn tail with |
| 121 | // a lost suffix, or nothing decodable): the safe write shape is a full |
| 122 | // rewrite that also compacts the log back to a healthy single event. |
| 123 | repairLog bool |
| 124 | // ledgerStale is set when the on-disk transcript already matches the |
| 125 | // snapshot but the meta ledger still describes older content — the |
| 126 | // aftermath of a save whose bytes landed and whose revision record then |
| 127 | // failed. The up-to-date path must heal the ledger instead of skipping it. |
| 128 | ledgerStale bool |
| 129 | } |
| 130 | |
| 131 | type SessionSnapshotConflictKind string |
| 132 | |
| 133 | const ( |
| 134 | SessionSnapshotConflictStalePrefix SessionSnapshotConflictKind = "stale_prefix" |
| 135 | SessionSnapshotConflictDiverged SessionSnapshotConflictKind = "diverged" |
| 136 | ) |
| 137 | |
| 138 | type SessionSnapshotConflictError struct { |
| 139 | Path string |
| 140 | Kind SessionSnapshotConflictKind |
| 141 | ExistingMessages int |
| 142 | SnapshotMessages int |
| 143 | BaseRevision int64 |
| 144 | DiskRevision int64 |
| 145 | } |
| 146 | |
| 147 | func (e *SessionSnapshotConflictError) Error() string { |
| 148 | if e == nil { |
| 149 | return ErrSessionSnapshotConflict.Error() |
| 150 | } |
| 151 | switch e.Kind { |
| 152 | case SessionSnapshotConflictStalePrefix: |
| 153 | return fmt.Sprintf("%s: %s has %d messages at revision %d; stale snapshot has %d messages from revision %d", |
| 154 | ErrSessionSnapshotConflict, e.Path, e.ExistingMessages, e.DiskRevision, e.SnapshotMessages, e.BaseRevision) |
| 155 | default: |
| 156 | return fmt.Sprintf("%s: %s diverged on disk (%d messages, revision %d) from snapshot (%d messages, revision %d)", |
| 157 | ErrSessionSnapshotConflict, e.Path, e.ExistingMessages, e.DiskRevision, e.SnapshotMessages, e.BaseRevision) |
| 158 | } |
| 159 | } |
| 160 | |
| 161 | func (e *SessionSnapshotConflictError) Unwrap() error { |
| 162 | return ErrSessionSnapshotConflict |
| 163 | } |
| 164 | |
| 165 | func SnapshotConflictKind(err error) (SessionSnapshotConflictKind, bool) { |
| 166 | var conflict *SessionSnapshotConflictError |
| 167 | if errors.As(err, &conflict) && conflict != nil { |
| 168 | return conflict.Kind, true |
| 169 | } |
| 170 | return "", false |
| 171 | } |
| 172 | |
| 173 | const RecoveryBranchDefaultName = "Recovered unsaved changes from stale runtime" |
| 174 | |
| 175 | type RecoveryBranchOptions struct { |
| 176 | OriginalPath string |
| 177 | Name string |
| 178 | Reason string |
| 179 | BranchMeta BranchMeta |
| 180 | BaseRevision int64 |
| 181 | DiskRevision int64 |
| 182 | } |
| 183 | |
| 184 | type RecoveryBranchInfo struct { |
| 185 | Path string |
| 186 | Digest string |
| 187 | Existing bool |
| 188 | Meta BranchMeta |
| 189 | Preview string |
| 190 | Turns int |
| 191 | } |
| 192 | |
| 193 | // Save persists the session using the normal CAS-protected snapshot protocol. |
| 194 | // It is kept as the convenient default for callers that do not need to spell |
| 195 | // out rewrite intent; it must never provide a force-overwrite escape hatch. |
| 196 | // The .jsonl file remains as a compatibility checkpoint and discovery anchor; |
| 197 | // the append-only event log is authoritative once present, while the .jsonl |
| 198 | // checkpoint also serves as the random-read model for history paging. |
| 199 | func (s *Session) Save(path string) error { |
| 200 | return s.saveObserved(path, sessionSaveSnapshot) |
| 201 | } |
| 202 | |
| 203 | // SaveSnapshot writes a normal autosave/snapshot only when doing so cannot hide |
| 204 | // a newer transcript already on disk. Explicit history rewrites such as rewind, |
| 205 | // compaction, and cancel recovery should call SaveRewrite instead. |
| 206 | func (s *Session) SaveSnapshot(path string) error { |
| 207 | return s.saveObserved(path, sessionSaveSnapshot) |
| 208 | } |
| 209 | |
| 210 | // SaveRewrite writes an intentional non-append history rewrite only while this |
| 211 | // Session still owns the current on-disk transcript baseline. It prevents a |
| 212 | // stale controller from force-rewinding a newer transcript written elsewhere. |
| 213 | func (s *Session) SaveRewrite(path string) error { |
| 214 | return s.saveObserved(path, sessionSaveRewrite) |
| 215 | } |
| 216 | |
| 217 | // SaveRewriteCompact performs a CAS-protected rewrite and folds the event log |
| 218 | // to one replace record. It is for destructive maintenance such as redaction: |
| 219 | // retaining old WAL records would keep the removed bytes recoverable on disk. |
| 220 | func (s *Session) SaveRewriteCompact(path string) error { |
| 221 | return s.saveObserved(path, sessionSaveRewriteCompact) |
| 222 | } |
| 223 | |
| 224 | func (s *Session) save(path string, mode sessionSaveMode) error { |
| 225 | if path == "" { |
| 226 | return fmt.Errorf("empty session path") |
| 227 | } |
| 228 | return s.withSessionSaveLocks(path, func() error { |
| 229 | return s.saveLocked(path, mode) |
| 230 | }) |
| 231 | } |
| 232 | |
| 233 | func (s *Session) withSessionSaveLocks(path string, fn func() error) error { |
| 234 | if strings.TrimSpace(path) == "" { |
| 235 | return fmt.Errorf("empty session path") |
| 236 | } |
| 237 | unlock := lockSessionSavePath(path) |
| 238 | defer unlock() |
| 239 | if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { |
| 240 | return fmt.Errorf("create session dir: %w", err) |
| 241 | } |
| 242 | unlockFile, err := lockSessionFile(path) |
| 243 | if err != nil { |
| 244 | return fmt.Errorf("lock session file: %w", err) |
| 245 | } |
| 246 | defer unlockFile() |
| 247 | return fn() |
| 248 | } |
| 249 | |
| 250 | func sessionArtifactExists(path string) bool { |
| 251 | if _, err := os.Lstat(path); err == nil { |
| 252 | return true |
| 253 | } |
| 254 | for _, artifact := range store.SessionSidecarFiles(path) { |
| 255 | // A legacy v0 event transcript can share the native |
| 256 | // `<id>.events.jsonl` name with the destination being imported. It is |
| 257 | // intentionally left in place, and must not make SaveIfAbsent believe |
| 258 | // that the v1 `<id>.jsonl` destination already exists. Native event logs |
| 259 | // remain owned artifacts and still protect the destination from a second |
| 260 | // writer. |
| 261 | if artifact == store.SessionEventLog(path) { |
| 262 | probe, probeErr := probeSessionEventLog(path) |
| 263 | if probeErr != nil { |
| 264 | return true |
| 265 | } |
| 266 | if !probe.native { |
| 267 | continue |
| 268 | } |
| 269 | } |
| 270 | if _, err := os.Lstat(artifact); err == nil { |
| 271 | return true |
| 272 | } |
| 273 | } |
| 274 | return false |
| 275 | } |
| 276 | |
| 277 | func (s *Session) saveLocked(path string, mode sessionSaveMode) error { |
| 278 | baseRevision := int64(0) |
| 279 | releaseAuth, err := s.requireWriteAuthorityForSave(path) |
| 280 | if err != nil { |
| 281 | return err |
| 282 | } |
| 283 | defer releaseAuth() |
| 284 | observeUnleasedSessionWrite(path, mode) |
| 285 | // Heal an empty/missing checkpoint from a valid WAL before classification |
| 286 | // so a 0-byte .jsonl never forces a false diverged recovery. |
| 287 | if err := healEmptyCheckpointFromWAL(path); err != nil { |
| 288 | return err |
| 289 | } |
| 290 | if mode == sessionSaveSnapshot && s.snapshotUpToDate(path) { |
| 291 | // Nothing changed since the last successful save to this exact path: |
| 292 | // skip the rest of the save — including the full transcript serialize |
| 293 | // + digest + disk probe the up-to-date decision below would still |
| 294 | // pay. Desktop switch/close/prune paths snapshot defensively on every |
| 295 | // navigation, and on large sessions that per-save cost is the |
| 296 | // user-visible seconds of UI freeze in #6607. Version bookkeeping |
| 297 | // makes this exact: any Add/Replace/preview update bumps version, any |
| 298 | // rewrite bumps rewriteVersion, and load-time repairs or log damage |
| 299 | // disarm the fast path until a real save persists them. The check |
| 300 | // runs under the save locks, not before them, so a saver that waited |
| 301 | // on a concurrent writer still re-evaluates against the state it must |
| 302 | // persist when it finally enters the critical section. |
| 303 | return nil |
| 304 | } |
| 305 | // Capture the snapshot only while holding the save locks. Concurrent |
| 306 | // in-process savers (turn-end snapshot, periodic autosave, shutdown |
| 307 | // snapshot) that captured before locking could land out of order: the |
| 308 | // stalest capture written last would then read the newer transcript it |
| 309 | // lost the race to as a bogus stale-prefix conflict. |
| 310 | msgs, version, rewriteVersion := s.snapshotWithVersion() |
| 311 | digest, contentBytes, err := digestAndSizeSessionMessages(msgs) |
| 312 | if err != nil { |
| 313 | return err |
| 314 | } |
| 315 | probe, err := s.probeNativeLogForSave(path) |
| 316 | if err != nil { |
| 317 | return err |
| 318 | } |
| 319 | if route := s.dagSaveRoute(path, probe); route != dagRouteSchemaOne { |
| 320 | return s.saveDAGLocked(path, mode, route, msgs, version, rewriteVersion, digest) |
| 321 | } |
| 322 | repairLog := false |
| 323 | deferProjection := mode.defersProjection() |
| 324 | ownedRewrite := mode.allowsOwnedRewrite() |
| 325 | decision, err := s.classifySnapshotWriteForCommit(path, msgs, digest, version, ownedRewrite, mode) |
| 326 | if err != nil { |
| 327 | return err |
| 328 | } |
| 329 | if decision.upToDate && mode != sessionSaveRewriteCompact { |
| 330 | // Disk already holds exactly this transcript. Rewriting it would only |
| 331 | // bump the revision, invalidating the persistence baseline of every |
| 332 | // other runtime resumed on this file and turning their next |
| 333 | // legitimate save into a stale-runtime conflict. Skip the write and |
| 334 | // adopt the current on-disk revision as this session's baseline. |
| 335 | if decision.ledgerStale { |
| 336 | // ...unless the ledger never learned about this transcript: a |
| 337 | // prior save landed its bytes and then failed to record the |
| 338 | // revision. Same-content retries are exactly the "later save" |
| 339 | // that failure deferred to, and skipping here would strand the |
| 340 | // ledger on the old digest forever. Record now, reproducing |
| 341 | // the state the interrupted save would have left. |
| 342 | revision, err := recordSessionContentRevision(path, digest, decision.revision, decision.reservedRevision) |
| 343 | if err != nil { |
| 344 | return err |
| 345 | } |
| 346 | displayModelCurrent := true |
| 347 | if err := writeSessionMessages(path, msgs); err != nil { |
| 348 | displayModelCurrent = false |
| 349 | slog.Warn("session: keeping save after display read-model repair failure", "path", path, "err", err) |
| 350 | } |
| 351 | if probe.native { |
| 352 | if err := writeSessionEventIndex(path, msgs, digest, revision); err != nil { |
| 353 | // See the append path below: index loss must not fail a |
| 354 | // save whose transcript and revision already landed. |
| 355 | slog.Warn("session: keeping save after event index write failure", "path", path, "err", err) |
| 356 | } |
| 357 | } |
| 358 | if displayModelCurrent { |
| 359 | if err := refreshCheckpointDisplayIndex(path, msgs, digest, revision, -1, deferProjection); err != nil { |
| 360 | // The display index is a derived sidecar; transcript durability |
| 361 | // must not depend on rebuilding it successfully. |
| 362 | slog.Warn("session: keeping save after display index write failure", "path", path, "err", err) |
| 363 | } |
| 364 | } |
| 365 | s.markCheckpointPersisted(path, digest, version, revision, rewriteVersion, msgs, deferProjection) |
| 366 | return nil |
| 367 | } |
| 368 | s.refreshPendingCheckpointProjection(path, msgs, digest, decision.revision, deferProjection) |
| 369 | s.markCheckpointPersisted(path, digest, version, decision.revision, rewriteVersion, msgs, deferProjection) |
| 370 | return nil |
| 371 | } |
| 372 | if decision.appendOnly && probe.native && mode != sessionSaveRewriteCompact { |
| 373 | logSize := sessionEventLogSize(path) |
| 374 | displayModelCurrent := false |
| 375 | switch { |
| 376 | case logSize == 0: |
| 377 | if err := appendSessionReplaceEvent(path, msgs, digest, decision.revision, "snapshot"); err != nil { |
| 378 | return err |
| 379 | } |
| 380 | displayModelCurrent, err = appendSessionDisplayReadModel(path, msgs, decision.appendFrom, decision.revision) |
| 381 | case sessionEventLogOversized(logSize, contentBytes): |
| 382 | // Fold history into one replace event and refresh the random-read |
| 383 | // model atomically. Normal appends keep it current below too. |
| 384 | if err := compactSessionEventLog(path, msgs, digest, decision.revision, "compact"); err != nil { |
| 385 | return err |
| 386 | } |
| 387 | if err := writeSessionMessages(path, msgs); err != nil { |
| 388 | return err |
| 389 | } |
| 390 | displayModelCurrent = true |
| 391 | default: |
| 392 | if err := appendSessionAppendEvent(path, decision.appendFrom, msgs[decision.appendFrom:], digest, decision.revision); err != nil { |
| 393 | return err |
| 394 | } |
| 395 | displayModelCurrent, err = appendSessionDisplayReadModel(path, msgs, decision.appendFrom, decision.revision) |
| 396 | } |
| 397 | if err != nil { |
| 398 | slog.Warn("session: keeping save after display read-model append failure", "path", path, "err", err) |
| 399 | } |
| 400 | revision, err := recordSessionContentRevision(path, digest, decision.revision, decision.reservedRevision) |
| 401 | if err != nil { |
| 402 | return err |
| 403 | } |
| 404 | if err := writeSessionEventIndex(path, msgs, digest, revision); err != nil { |
| 405 | // The event index is only a listing accelerator; the transcript |
| 406 | // and its revision are already durable above. Failing the save |
| 407 | // here would skip markPersisted and leave the in-memory baseline |
| 408 | // behind the disk state it just wrote, misreading the next save |
| 409 | // as a stale-runtime conflict. |
| 410 | slog.Warn("session: keeping save after event index write failure", "path", path, "err", err) |
| 411 | } |
| 412 | if displayModelCurrent { |
| 413 | if err := refreshCheckpointDisplayIndex(path, msgs, digest, revision, decision.appendFrom, deferProjection); err != nil { |
| 414 | // The append boundary lets the refresh extend the previous index |
| 415 | // instead of re-encoding the whole transcript. |
| 416 | slog.Warn("session: keeping save after display index write failure", "path", path, "err", err) |
| 417 | } |
| 418 | } |
| 419 | s.markCheckpointPersisted(path, digest, version, revision, rewriteVersion, msgs, deferProjection) |
| 420 | return nil |
| 421 | } |
| 422 | baseRevision = decision.revision |
| 423 | repairLog = decision.repairLog |
| 424 | // Full-rewrite path: new snapshots, intentional history rewrites, and |
| 425 | // damage repairs. The event log mutates first so a crash between the two |
| 426 | // writes leaves the newer transcript authoritative; the anchor rewrite |
| 427 | // keeps the compatibility .jsonl fresh for direct readers. |
| 428 | reason := mode.eventReason() |
| 429 | if repairLog { |
| 430 | reason = "repair" |
| 431 | } |
| 432 | logSize := sessionEventLogSize(path) |
| 433 | switch { |
| 434 | case !probe.native: |
| 435 | // A foreign file (legacy import leftover) squats the native log path. |
| 436 | // Never write into or over it — the session stays checkpoint-only. |
| 437 | case mode == sessionSaveRewriteCompact: |
| 438 | // Maintenance rewrites compact even a short log so redacted or otherwise |
| 439 | // removed bytes do not remain recoverable in historical WAL records. |
| 440 | if err := compactSessionEventLog(path, msgs, digest, baseRevision, reason); err != nil { |
| 441 | return err |
| 442 | } |
| 443 | case repairLog, sessionEventLogOversized(logSize, contentBytes): |
| 444 | if err := compactSessionEventLog(path, msgs, digest, baseRevision, reason); err != nil { |
| 445 | return err |
| 446 | } |
| 447 | default: |
| 448 | if err := appendSessionReplaceEvent(path, msgs, digest, baseRevision, reason); err != nil { |
| 449 | return err |
| 450 | } |
| 451 | } |
| 452 | if err := writeSessionMessages(path, msgs); err != nil { |
| 453 | return err |
| 454 | } |
| 455 | revision, err := recordSessionContentRevision(path, digest, baseRevision, decision.reservedRevision) |
| 456 | if err != nil { |
| 457 | return err |
| 458 | } |
| 459 | if probe.native { |
| 460 | if err := writeSessionEventIndex(path, msgs, digest, revision); err != nil { |
| 461 | // See the append path above: index loss must not fail a save whose |
| 462 | // transcript and revision already landed. |
| 463 | slog.Warn("session: keeping save after event index write failure", "path", path, "err", err) |
| 464 | } |
| 465 | } |
| 466 | if err := refreshCheckpointDisplayIndex(path, msgs, digest, revision, -1, deferProjection); err != nil { |
| 467 | // Warn-only like the event index above: the display index is a pure |
| 468 | // derived sidecar and must never fail a save. |
| 469 | slog.Warn("session: keeping save after display index write failure", "path", path, "err", err) |
| 470 | } |
| 471 | s.markCheckpointPersisted(path, digest, version, revision, rewriteVersion, msgs, deferProjection) |
| 472 | return nil |
| 473 | } |
| 474 | |
| 475 | // checkSnapshotWrite decides whether this session may write msgs over path, and |
| 476 | // whether the safe write shape is a no-op, append-only suffix, or full rewrite. |
| 477 | func (s *Session) checkSnapshotWrite(path string, next []provider.Message, nextDigest [sha256.Size]byte, nextVersion uint64, allowOwnedRewrite bool) (snapshotWriteDecision, error) { |
| 478 | baseState := s.persistState(path) |
| 479 | current, err := loadSessionUnlocked(path) |
| 480 | if err != nil { |
| 481 | if os.IsNotExist(err) { |
| 482 | if baseState.ok { |
| 483 | return snapshotWriteDecision{}, ErrSessionExternallyRemoved |
| 484 | } |
| 485 | return snapshotWriteDecision{}, nil |
| 486 | } |
| 487 | return snapshotWriteDecision{}, err |
| 488 | } |
| 489 | currentRevision, currentLedgerDigest, err := sessionContentRevision(path) |
| 490 | if err != nil { |
| 491 | return snapshotWriteDecision{}, err |
| 492 | } |
| 493 | existing := current.Snapshot() |
| 494 | existingDigest, err := digestSessionMessages(existing) |
| 495 | if err != nil { |
| 496 | return snapshotWriteDecision{}, err |
| 497 | } |
| 498 | // raw is the transcript as stored, before load-time normalization repaired |
| 499 | // it; it equals existing when no repair ran. The prefix checks below must |
| 500 | // be able to fall back to it: a mid-turn snapshot legitimately cuts an |
| 501 | // assistant tool call from its still-running result, normalization then |
| 502 | // fabricates a placeholder answer on load, and the live session's real |
| 503 | // result collides with that placeholder — misreading a pure append as |
| 504 | // divergence (and forking a bogus recovery branch). |
| 505 | raw, rawDigest := existing, existingDigest |
| 506 | rawDiffers := current.normalizedDirty && len(current.rawMessages) > 0 |
| 507 | if rawDiffers { |
| 508 | raw = current.rawMessages |
| 509 | if rawDigest, err = digestSessionMessages(raw); err != nil { |
| 510 | return snapshotWriteDecision{}, err |
| 511 | } |
| 512 | } |
| 513 | contentUnchanged := bytes.Equal(existingDigest[:], nextDigest[:]) |
| 514 | exactAppend := messagesHavePrefix(next, existing) |
| 515 | appendShaped := contentUnchanged || exactAppend || messagesHavePrefixWithCompatibleSystem(next, existing) |
| 516 | repairPending := current.normalizedDirty |
| 517 | if !appendShaped && rawDiffers { |
| 518 | rawUnchanged := bytes.Equal(rawDigest[:], nextDigest[:]) |
| 519 | rawAppend := messagesHavePrefix(next, raw) |
| 520 | if rawUnchanged || rawAppend || messagesHavePrefixWithCompatibleSystem(next, raw) { |
| 521 | existing = raw |
| 522 | contentUnchanged = rawUnchanged |
| 523 | exactAppend = rawAppend |
| 524 | appendShaped = true |
| 525 | // The snapshot supersedes the repaired view — appending it lands |
| 526 | // the real tool results where the placeholders were fabricated — |
| 527 | // so no load-time repair is left to force a rewrite. |
| 528 | repairPending = false |
| 529 | } |
| 530 | } |
| 531 | if !appendShaped && baseState.ok && baseState.revisionKnown && |
| 532 | baseState.revision == currentRevision && !contentUnchanged { |
| 533 | // Revision equality alone is not ownership proof. Require digest |
| 534 | // ancestry or a live generation-bound write authority covering path. |
| 535 | if s.ownsWritableBaseline(path, existingDigest, rawDigest, rawDiffers, currentRevision, currentLedgerDigest, nextVersion) { |
| 536 | appendShaped = true |
| 537 | } |
| 538 | } |
| 539 | if appendShaped { |
| 540 | // An unknown-revision baseline (meta sidecar unreadable at load) cannot |
| 541 | // vouch for revision equality; the digest/prefix checks above already |
| 542 | // vouch for the content, so only a known baseline arms the CAS check. |
| 543 | // Under an append-shaped write (at most a compatible leading-system |
| 544 | // swap) a stale revision is ledger drift — a reset sidecar, a |
| 545 | // same-content heal, or another runtime recording messages this |
| 546 | // snapshot already contains — unless the transcript was rewound. |
| 547 | // Locating the persisted baseline among the snapshot's prefixes and |
| 548 | // requiring the disk transcript to still reach it tells the two apart: |
| 549 | // drift keeps the baseline reachable, while a rewind cut below it and |
| 550 | // appending would resurrect the suffix another runtime removed. |
| 551 | if baseState.ok && baseState.revisionKnown && currentRevision != baseState.revision && !contentUnchanged && |
| 552 | !appendCoversPersistedBaseline(next, existing, baseState.digest) { |
| 553 | return snapshotWriteDecision{}, snapshotConflict(path, existing, next, baseState.revision, currentRevision) |
| 554 | } |
| 555 | // A normalized-dirty load means LoadSession repaired the history on the |
| 556 | // way in: the digests match but the raw bytes on disk do not, so the |
| 557 | // repair still needs a real write to persist. A damaged event log |
| 558 | // likewise needs a real write (rewrite + compact) even when the |
| 559 | // replayable prefix already matches this snapshot. |
| 560 | decision := snapshotWriteDecision{ |
| 561 | revision: currentRevision, |
| 562 | upToDate: contentUnchanged && !repairPending && !current.eventLogDamaged, |
| 563 | repairLog: current.eventLogDamaged, |
| 564 | } |
| 565 | // A ledger digest that describes different content than the transcript |
| 566 | // on disk is the aftermath of a save whose bytes landed and whose |
| 567 | // revision record then failed (crash or fail-closed record between the |
| 568 | // two writes). Only a non-empty mismatch counts: a missing sidecar or |
| 569 | // a legacy one without a digest is a legitimate state, and stamping it |
| 570 | // here would bump revisions other runtimes still hold as baselines. |
| 571 | if decision.upToDate && currentLedgerDigest != "" && currentLedgerDigest != digestString(nextDigest) { |
| 572 | decision.ledgerStale = true |
| 573 | } |
| 574 | // An append is only chain-safe when existing measures the transcript |
| 575 | // the event log actually replays. Under a pending load-time repair the |
| 576 | // normalized view differs from the raw log, so an append event indexed |
| 577 | // against it breaks the replay chain and orphans the appended suffix; |
| 578 | // fall through to the full rewrite, which also persists the repair. |
| 579 | if exactAppend && !contentUnchanged && len(existing) < len(next) && !current.eventLogDamaged && !repairPending { |
| 580 | decision.appendOnly = true |
| 581 | decision.appendFrom = len(existing) |
| 582 | } |
| 583 | return decision, nil |
| 584 | } |
| 585 | if allowOwnedRewrite { |
| 586 | if s.ownsWritableBaseline(path, existingDigest, rawDigest, rawDiffers, currentRevision, currentLedgerDigest, nextVersion) { |
| 587 | return snapshotWriteDecision{revision: currentRevision, repairLog: current.eventLogDamaged}, nil |
| 588 | } |
| 589 | } |
| 590 | // Bound controllers: missing/stale authority must not fork recovery. |
| 591 | if err := s.authorityErrorForPath(path); err != nil { |
| 592 | return snapshotWriteDecision{}, err |
| 593 | } |
| 594 | if messagesHavePrefix(existing, next) || messagesHavePrefixWithCompatibleSystem(existing, next) || |
| 595 | (rawDiffers && (messagesHavePrefix(raw, next) || messagesHavePrefixWithCompatibleSystem(raw, next))) { |
| 596 | return snapshotWriteDecision{}, &SessionSnapshotConflictError{ |
| 597 | Path: path, |
| 598 | Kind: SessionSnapshotConflictStalePrefix, |
| 599 | ExistingMessages: len(existing), |
| 600 | SnapshotMessages: len(next), |
| 601 | BaseRevision: baseState.revision, |
| 602 | DiskRevision: currentRevision, |
| 603 | } |
| 604 | } |
| 605 | return snapshotWriteDecision{}, &SessionSnapshotConflictError{ |
| 606 | Path: path, |
| 607 | Kind: SessionSnapshotConflictDiverged, |
| 608 | ExistingMessages: len(existing), |
| 609 | SnapshotMessages: len(next), |
| 610 | BaseRevision: baseState.revision, |
| 611 | DiskRevision: currentRevision, |
| 612 | } |
| 613 | } |
| 614 | |
| 615 | func snapshotConflict(path string, existing, next []provider.Message, baseRevision, diskRevision int64) error { |
| 616 | kind := SessionSnapshotConflictDiverged |
| 617 | if messagesHavePrefix(existing, next) || messagesHavePrefixWithCompatibleSystem(existing, next) { |
| 618 | kind = SessionSnapshotConflictStalePrefix |
| 619 | } |
| 620 | return &SessionSnapshotConflictError{ |
| 621 | Path: path, |
| 622 | Kind: kind, |
| 623 | ExistingMessages: len(existing), |
| 624 | SnapshotMessages: len(next), |
| 625 | BaseRevision: baseRevision, |
| 626 | DiskRevision: diskRevision, |
| 627 | } |
| 628 | } |
| 629 | |
| 630 | func (s *Session) SaveRecoveryBranch(opts RecoveryBranchOptions) (RecoveryBranchInfo, error) { |
| 631 | return s.saveRecoveryBranch(opts, false) |
| 632 | } |
| 633 | |
| 634 | // SaveShutdownRecoveryBranch persists the current transcript to a distinct |
| 635 | // recovery branch after the normal shutdown snapshot failed with |
| 636 | // ErrSessionFileLockHeld. It deliberately does not re-lock or inspect the |
| 637 | // original session file: doing so would repeat the same bounded timeout and |
| 638 | // let process teardown discard the only remaining in-memory copy. |
| 639 | // |
| 640 | // The recovery filename includes this live Session's isolated lane, so another |
| 641 | // controller in the same process cannot replace the emergency copy. |
| 642 | // The result still uses the normal session, event-log, and branch-meta formats |
| 643 | // and is therefore discoverable and resumable through existing flows. |
| 644 | func (s *Session) SaveShutdownRecoveryBranch(opts RecoveryBranchOptions) (RecoveryBranchInfo, error) { |
| 645 | return s.saveRecoveryBranch(opts, true) |
| 646 | } |
| 647 | |
| 648 | // SaveConflictRecoveryBranch writes the depth-cap isolated copy (one path per |
| 649 | // live Session). Subsequent conflicts from that Session rewrite it in place. |
| 650 | func (s *Session) SaveConflictRecoveryBranch(opts RecoveryBranchOptions) (RecoveryBranchInfo, error) { |
| 651 | return s.saveRecoveryBranch(opts, true) |
| 652 | } |
| 653 | |
| 654 | func (s *Session) saveRecoveryBranch(opts RecoveryBranchOptions, shutdown bool) (RecoveryBranchInfo, error) { |
| 655 | originalPath := strings.TrimSpace(opts.OriginalPath) |
| 656 | if originalPath == "" { |
| 657 | return RecoveryBranchInfo{}, fmt.Errorf("empty original session path") |
| 658 | } |
| 659 | opts.OriginalPath = originalPath |
| 660 | msgs, version, rewriteVersion := s.snapshotWithVersion() |
| 661 | preview, turns := SessionPreviewFromMessages(msgs) |
| 662 | if turns == 0 { |
| 663 | return RecoveryBranchInfo{}, ErrSessionRecoveryNotNeeded |
| 664 | } |
| 665 | digest, err := digestSessionMessages(msgs) |
| 666 | if err != nil { |
| 667 | return RecoveryBranchInfo{}, err |
| 668 | } |
| 669 | digestText := digestString(digest) |
| 670 | |
| 671 | if !shutdown { |
| 672 | unlockOriginal := lockSessionSavePath(originalPath) |
| 673 | unlockOriginalFile, lockErr := lockSessionFile(originalPath) |
| 674 | if lockErr != nil { |
| 675 | unlockOriginal() |
| 676 | return RecoveryBranchInfo{}, fmt.Errorf("lock original session file: %w", lockErr) |
| 677 | } |
| 678 | current, loadErr := loadSessionUnlocked(originalPath) |
| 679 | unlockOriginalFile() |
| 680 | unlockOriginal() |
| 681 | if loadErr != nil && !os.IsNotExist(loadErr) { |
| 682 | return RecoveryBranchInfo{}, loadErr |
| 683 | } |
| 684 | if loadErr == nil && current != nil { |
| 685 | existing := current.Snapshot() |
| 686 | existingDigest, digestErr := digestSessionMessages(existing) |
| 687 | if digestErr != nil { |
| 688 | return RecoveryBranchInfo{}, digestErr |
| 689 | } |
| 690 | covered := bytes.Equal(existingDigest[:], digest[:]) || |
| 691 | messagesHavePrefix(existing, msgs) || |
| 692 | messagesHavePrefixWithCompatibleSystem(existing, msgs) |
| 693 | if !covered && current.normalizedDirty && len(current.rawMessages) > 0 { |
| 694 | // Judge coverage against the pre-repair transcript too, for the |
| 695 | // same reason as checkSnapshotWrite: load-time normalization can |
| 696 | // reshape what is actually stored, and a recovery fork is only |
| 697 | // warranted when the stored bytes themselves fail to cover this |
| 698 | // snapshot. |
| 699 | raw := current.rawMessages |
| 700 | rawDigest, rawErr := digestSessionMessages(raw) |
| 701 | if rawErr != nil { |
| 702 | return RecoveryBranchInfo{}, rawErr |
| 703 | } |
| 704 | covered = bytes.Equal(rawDigest[:], digest[:]) || |
| 705 | messagesHavePrefix(raw, msgs) || |
| 706 | messagesHavePrefixWithCompatibleSystem(raw, msgs) |
| 707 | } |
| 708 | if covered { |
| 709 | return RecoveryBranchInfo{}, ErrSessionRecoveryNotNeeded |
| 710 | } |
| 711 | } |
| 712 | } |
| 713 | |
| 714 | // One stable recovery file per (root branch, writer generation). Nested |
| 715 | // -recovery- names peel back to the root so conflicts update in place. |
| 716 | for range 8 { |
| 717 | recoveryPath, lane := s.isolatedRecoverySessionPath(originalPath) |
| 718 | info, collision, err := s.writeRecoveryBranchAtPath(recoveryPath, opts, msgs, digest, |
| 719 | version, rewriteVersion, preview, turns, digestText, 1, shutdown) |
| 720 | if err != nil { |
| 721 | return RecoveryBranchInfo{}, err |
| 722 | } |
| 723 | if !collision { |
| 724 | return info, nil |
| 725 | } |
| 726 | s.rotateRecoveryLane(lane) |
| 727 | } |
| 728 | return RecoveryBranchInfo{}, fmt.Errorf("allocate isolated recovery lane: too many existing collisions") |
| 729 | } |
| 730 | |
| 731 | func recoveryParentStem(parent string) string { |
| 732 | parent = strings.TrimSpace(parent) |
| 733 | if parent == "" { |
| 734 | return "session" |
| 735 | } |
| 736 | sum := sha256.Sum256([]byte(parent)) |
| 737 | if before, _, ok := strings.Cut(parent, "-recovery-"); ok { |
| 738 | base := strings.Trim(before, "-_. ") |
| 739 | if base == "" { |
| 740 | base = "session" |
| 741 | } |
| 742 | base = strings.Trim(truncateUTF8Bytes(base, maxRecoveryParentStemBytes), "-_. ") |
| 743 | if base == "" { |
| 744 | base = "session" |
| 745 | } |
| 746 | return fmt.Sprintf("%s-%x", base, sum[:6]) |
| 747 | } |
| 748 | if len(parent) <= maxRecoveryParentStemBytes { |
| 749 | return parent |
| 750 | } |
| 751 | prefix := strings.Trim(truncateUTF8Bytes(parent, maxRecoveryParentStemBytes), "-_. ") |
| 752 | if prefix == "" { |
| 753 | prefix = "session" |
| 754 | } |
| 755 | return fmt.Sprintf("%s-%x", prefix, sum[:6]) |
| 756 | } |
| 757 | |
| 758 | func truncateUTF8Bytes(s string, max int) string { |
| 759 | if max <= 0 { |
| 760 | return "" |
| 761 | } |
| 762 | if len(s) <= max { |
| 763 | return s |
| 764 | } |
| 765 | used := 0 |
| 766 | for i, r := range s { |
| 767 | size := utf8.RuneLen(r) |
| 768 | if size < 0 { |
| 769 | size = 1 |
| 770 | } |
| 771 | if used+size > max { |
| 772 | return s[:i] |
| 773 | } |
| 774 | used += size |
| 775 | } |
| 776 | return s |
| 777 | } |
| 778 | |
| 779 | func firstNonEmpty(values ...string) string { |
| 780 | for _, value := range values { |
| 781 | if strings.TrimSpace(value) != "" { |
| 782 | return value |
| 783 | } |
| 784 | } |
| 785 | return "" |
| 786 | } |
| 787 | |
| 788 | func (s *Session) ownsPersistedState(path string, existingDigest [sha256.Size]byte, existingRevision int64, existingLedgerDigest string, nextVersion uint64) bool { |
| 789 | state := s.persistState(path) |
| 790 | if !state.ok || state.version > nextVersion || !bytes.Equal(existingDigest[:], state.digest[:]) { |
| 791 | return false |
| 792 | } |
| 793 | // An unknown-revision baseline still owns the transcript it loaded — the |
| 794 | // digest+version match proves it. Requiring revision equality here would |
| 795 | // make every rewrite from such a baseline a permanent conflict, because |
| 796 | // the revision can only be re-learned by a successful save. |
| 797 | // A disk ledger with no recorded revision is the mirror case: recorded |
| 798 | // revisions start at 1, so revision 0 means the sidecar was deleted or |
| 799 | // rebuilt by a listing-only writer after this session's save. An absent |
| 800 | // claim cannot revoke the ownership the digest+version match proves. |
| 801 | if !state.revisionKnown || existingRevision == 0 || state.revision == existingRevision { |
| 802 | return true |
| 803 | } |
| 804 | // A foreign revision stamp whose recorded digest still describes these |
| 805 | // exact bytes (a same-content heal or no-op record by another runtime) |
| 806 | // vouches for no content of its own: the transcript is byte-for-byte what |
| 807 | // this session last persisted, so rewriting it destroys nothing of |
| 808 | // theirs — at worst the conflict moves to the stamper's next divergent |
| 809 | // save, where its in-memory history forks a recovery branch as usual. |
| 810 | // A stamp that disagrees with the on-disk transcript (or a legacy stamp |
| 811 | // with no digest) keeps revoking ownership: that is the aftermath of a |
| 812 | // save whose bytes and record split, the bytes cannot be attributed, and |
| 813 | // only the conservative conflict path preserves both sides. |
| 814 | return existingLedgerDigest == digestString(existingDigest) |
| 815 | } |
| 816 | |
| 817 | // snapshotUpToDate reports whether a snapshot save to path is a provable |
| 818 | // no-op from in-memory bookkeeping alone: the last successful save went to |
| 819 | // this same path with a known ledger revision, the transcript version and |
| 820 | // rewrite version have not moved since, and no load-time repair or event-log |
| 821 | // damage is waiting to be persisted. Every one of these flags fails open — |
| 822 | // when any is unset or stale the caller falls through to the full save path, |
| 823 | // which re-derives the truth from disk. |
| 824 | func (s *Session) snapshotUpToDate(path string) bool { |
| 825 | key := canonicalSessionSavePath(path) |
| 826 | s.mu.RLock() |
| 827 | defer s.mu.RUnlock() |
| 828 | return s.persisted.ok && |
| 829 | s.persisted.saveVerified && |
| 830 | !s.persisted.projectionPending && |
| 831 | s.persisted.path == key && |
| 832 | s.persisted.version == s.version && |
| 833 | s.persisted.revisionKnown && |
| 834 | s.rewriteVersion == s.persistedRewriteVersion && |
| 835 | !s.normalizedDirty && |
| 836 | !s.eventLogDamaged |
| 837 | } |
| 838 | |
| 839 | func (s *Session) persistState(path string) sessionPersistState { |
| 840 | key := canonicalSessionSavePath(path) |
| 841 | s.mu.RLock() |
| 842 | defer s.mu.RUnlock() |
| 843 | if s.persisted.ok && s.persisted.path == key { |
| 844 | return s.persisted |
| 845 | } |
| 846 | return sessionPersistState{} |
| 847 | } |
| 848 | |
| 849 | // PersistedState is a read-only view of the baseline the session last |
| 850 | // persisted to (or loaded from) its transcript path. History paging uses it to |
| 851 | // validate a display-index sidecar against the live session without touching |
| 852 | // disk: an index built at the same revision with the same content digest |
| 853 | // describes exactly the persisted prefix of the in-memory log. |
| 854 | type PersistedState struct { |
| 855 | // Digest is the content digest of the persisted transcript. |
| 856 | Digest [sha256.Size]byte |
| 857 | // DigestHex is Digest in the hex form sidecars store. |
| 858 | DigestHex string |
| 859 | // Revision is the CAS ledger revision of the persisted transcript. It is |
| 860 | // meaningful only when RevisionKnown is true. |
| 861 | Revision int64 |
| 862 | RevisionKnown bool |
| 863 | // RewriteEpoch is the highest rewriteVersion that has reached disk. It |
| 864 | // changes only when a content rewrite (compaction, rewind, …) is saved, so |
| 865 | // it doubles as a stable epoch token for history entry IDs: append-only |
| 866 | // saves keep it, rewrites bump it. |
| 867 | RewriteEpoch int |
| 868 | // AppendOnlyTail reports that no rewrite landed after the baseline, so the |
| 869 | // persisted transcript is still a prefix of the in-memory log (the tail, |
| 870 | // if any, is unsaved appends). |
| 871 | AppendOnlyTail bool |
| 872 | // UnchangedSincePersisted reports that the in-memory log is exactly the |
| 873 | // persisted transcript (no appends, no rewrites since the baseline). |
| 874 | UnchangedSincePersisted bool |
| 875 | } |
| 876 | |
| 877 | // PersistedState returns the session's persistence baseline for path, or |
| 878 | // false when the session has never persisted to (or loaded from) that path. |
| 879 | func (s *Session) PersistedState(path string) (PersistedState, bool) { |
| 880 | key := canonicalSessionSavePath(path) |
| 881 | s.mu.RLock() |
| 882 | defer s.mu.RUnlock() |
| 883 | if !s.persisted.ok || s.persisted.path != key { |
| 884 | return PersistedState{}, false |
| 885 | } |
| 886 | return PersistedState{ |
| 887 | Digest: s.persisted.digest, |
| 888 | DigestHex: digestString(s.persisted.digest), |
| 889 | Revision: s.persisted.revision, |
| 890 | RevisionKnown: s.persisted.revisionKnown, |
| 891 | RewriteEpoch: s.persistedRewriteVersion, |
| 892 | AppendOnlyTail: s.rewriteVersion == s.persistedRewriteVersion, |
| 893 | UnchangedSincePersisted: s.persisted.version == s.version, |
| 894 | }, true |
| 895 | } |
| 896 | |
| 897 | // SessionContentIdentity returns the ledger identity of the authoritative |
| 898 | // persisted transcript without loading its message bodies. It is intended for |
| 899 | // derived sidecars such as the desktop display index: a matching checkpoint |
| 900 | // size alone cannot prove that offsets still describe the event-log-backed |
| 901 | // transcript. The bool is false for legacy sessions that have no digest in |
| 902 | // their branch metadata; callers must then validate against the transcript |
| 903 | // bytes directly. |
| 904 | func SessionContentIdentity(path string) (PersistedState, bool, error) { |
| 905 | revision, digestHex, err := sessionContentRevision(path) |
| 906 | if err != nil { |
| 907 | return PersistedState{}, false, err |
| 908 | } |
| 909 | if digestHex == "" { |
| 910 | return PersistedState{}, false, nil |
| 911 | } |
| 912 | var digest [sha256.Size]byte |
| 913 | decoded, err := hex.DecodeString(digestHex) |
| 914 | if err != nil || len(decoded) != len(digest) { |
| 915 | return PersistedState{}, false, fmt.Errorf("invalid session content digest") |
| 916 | } |
| 917 | copy(digest[:], decoded) |
| 918 | return PersistedState{ |
| 919 | Digest: digest, |
| 920 | DigestHex: digestString(digest), |
| 921 | Revision: revision, |
| 922 | RevisionKnown: true, |
| 923 | }, true, nil |
| 924 | } |
| 925 | |
| 926 | // sessionContentRevision reads the CAS ledger (revision + content digest) from |
| 927 | // the branch-meta sidecar. A missing sidecar is revision 0 — a session that |
| 928 | // has never recorded one. An unreadable sidecar is an error: reporting it as |
| 929 | // revision 0 would desync every runtime baseline from the ledger and turn the |
| 930 | // next honest save into a bogus conflict (and a recovery branch). |
| 931 | func sessionContentRevision(path string) (int64, string, error) { |
| 932 | meta, ok, err := loadBranchMetaRetry(path) |
| 933 | if err != nil { |
| 934 | return 0, "", err |
| 935 | } |
| 936 | if !ok { |
| 937 | return 0, "", nil |
| 938 | } |
| 939 | return meta.Revision, strings.TrimSpace(meta.ContentDigest), nil |
| 940 | } |
| 941 | |
| 942 | func recordSessionContentRevision(path string, digest [sha256.Size]byte, baseRevision, reservedRevision int64) (int64, error) { |
| 943 | // Revision allocation is a read-modify-write transaction. Holding only the |
| 944 | // final SaveBranchMeta lock would still let another writer replace the |
| 945 | // sidecar between our read and write, so keep the ledger lock across the |
| 946 | // increment and read-back as well. |
| 947 | unlock, err := LockSessionMetaPath(path) |
| 948 | if err != nil { |
| 949 | return 0, err |
| 950 | } |
| 951 | defer unlock() |
| 952 | |
| 953 | meta, ok, err := loadBranchMetaRetry(path) |
| 954 | if err != nil { |
| 955 | // Fail the save instead of rebuilding the ledger from a bad read: the |
| 956 | // transcript bytes already landed, so a later save can record the |
| 957 | // revision once the sidecar reads cleanly again — a content-bearing |
| 958 | // save lands here again, and a same-content retry heals through the |
| 959 | // up-to-date ledgerStale path in save(). |
| 960 | return 0, err |
| 961 | } |
| 962 | if !ok { |
| 963 | meta = BranchMeta{ID: BranchID(path)} |
| 964 | } |
| 965 | if reservedRevision > 0 { |
| 966 | if reservedRevision != baseRevision+1 || meta.Revision != reservedRevision || meta.SchemaVersion != 0 { |
| 967 | return 0, fmt.Errorf("session revision reservation changed: base=%d reserved=%d current=%d schema=%d", baseRevision, reservedRevision, meta.Revision, meta.SchemaVersion) |
| 968 | } |
| 969 | } else { |
| 970 | if meta.Revision < baseRevision { |
| 971 | meta.Revision = baseRevision |
| 972 | } |
| 973 | if meta.Revision == math.MaxInt64 { |
| 974 | return 0, fmt.Errorf("session revision exhausted") |
| 975 | } |
| 976 | meta.Revision++ |
| 977 | } |
| 978 | meta.ContentDigest = digestString(digest) |
| 979 | meta.WriterID = SessionWriterID() |
| 980 | if err := saveBranchMeta(path, meta, false); err != nil { |
| 981 | return 0, err |
| 982 | } |
| 983 | stored, ok, err := loadBranchMetaRetry(path) |
| 984 | if err != nil { |
| 985 | return 0, err |
| 986 | } |
| 987 | if ok && stored.Revision > 0 { |
| 988 | return stored.Revision, nil |
| 989 | } |
| 990 | return meta.Revision, nil |
| 991 | } |
| 992 | |
| 993 | func digestString(digest [sha256.Size]byte) string { |
| 994 | return fmt.Sprintf("%x", digest[:]) |
| 995 | } |
| 996 | |
| 997 | func SessionWriterID() string { |
| 998 | return sessionWriterID |
| 999 | } |
| 1000 | |
| 1001 | func newSessionWriterID() string { |
| 1002 | host, _ := os.Hostname() |
| 1003 | host = strings.TrimSpace(host) |
| 1004 | if host == "" { |
| 1005 | host = "unknown-host" |
| 1006 | } |
| 1007 | host = strings.Map(func(r rune) rune { |
| 1008 | switch { |
| 1009 | case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9', r == '-', r == '_', r == '.': |
| 1010 | return r |
| 1011 | default: |
| 1012 | return '-' |
| 1013 | } |
| 1014 | }, host) |
| 1015 | var nonce [8]byte |
| 1016 | if _, err := rand.Read(nonce[:]); err != nil { |
| 1017 | return fmt.Sprintf("%s-%d-%d", host, os.Getpid(), time.Now().UnixNano()) |
| 1018 | } |
| 1019 | return fmt.Sprintf("%s-%d-%x", host, os.Getpid(), nonce[:]) |
| 1020 | } |
| 1021 | |
| 1022 | func digestSessionMessages(msgs []provider.Message) ([sha256.Size]byte, error) { |
| 1023 | digest, _, err := digestAndSizeSessionMessages(msgs) |
| 1024 | return digest, err |
| 1025 | } |
| 1026 | |
| 1027 | func messageForSessionIdentity(m provider.Message) provider.Message { |
| 1028 | // CreatedAt and ID are local metadata. Keep them out of transcript identity |
| 1029 | // so older builds that ignore the optional fields can share the same event- |
| 1030 | // log revision and append without false conflicts. |
| 1031 | m.CreatedAt = 0 |
| 1032 | m.ID = "" |
| 1033 | return m |
| 1034 | } |
| 1035 | |
| 1036 | // digestAndSizeSessionMessages also reports the encoded transcript size, which |
| 1037 | // the save path uses to bound the event log relative to the live content. |
| 1038 | func digestAndSizeSessionMessages(msgs []provider.Message) ([sha256.Size]byte, int64, error) { |
| 1039 | h := sha256.New() |
| 1040 | size := int64(0) |
| 1041 | for _, m := range msgs { |
| 1042 | m = messageForSessionIdentity(m) |
| 1043 | b, err := json.Marshal(m) |
| 1044 | if err != nil { |
| 1045 | return [sha256.Size]byte{}, 0, err |
| 1046 | } |
| 1047 | if _, err := h.Write(b); err != nil { |
| 1048 | return [sha256.Size]byte{}, 0, err |
| 1049 | } |
| 1050 | if _, err := h.Write([]byte{'\n'}); err != nil { |
| 1051 | return [sha256.Size]byte{}, 0, err |
| 1052 | } |
| 1053 | size += int64(len(b)) + 1 |
| 1054 | } |
| 1055 | var out [sha256.Size]byte |
| 1056 | copy(out[:], h.Sum(nil)) |
| 1057 | return out, size, nil |
| 1058 | } |
| 1059 | |
| 1060 | // sessionTranscriptHasher accumulates the transcript digest one message at a |
| 1061 | // time, exactly like digestAndSizeSessionMessages, so load paths hash during |
| 1062 | // decode instead of re-serializing. A nil receiver disables hashing. |
| 1063 | type sessionTranscriptHasher struct { |
| 1064 | h hash.Hash |
| 1065 | err error |
| 1066 | } |
| 1067 | |
| 1068 | func newSessionTranscriptHasher() *sessionTranscriptHasher { |
| 1069 | return &sessionTranscriptHasher{h: sha256.New()} |
| 1070 | } |
| 1071 | |
| 1072 | // rehash restarts accumulation over msgs (event-log replace records). |
| 1073 | func (s *sessionTranscriptHasher) rehash(msgs []provider.Message) { |
| 1074 | if s == nil { |
| 1075 | return |
| 1076 | } |
| 1077 | s.h.Reset() |
| 1078 | s.err = nil |
| 1079 | s.addAll(msgs) |
| 1080 | } |
| 1081 | |
| 1082 | func (s *sessionTranscriptHasher) addAll(msgs []provider.Message) { |
| 1083 | for _, m := range msgs { |
| 1084 | s.add(m) |
| 1085 | } |
| 1086 | } |
| 1087 | |
| 1088 | // add hashes one message and returns it for append chaining. |
| 1089 | func (s *sessionTranscriptHasher) add(m provider.Message) provider.Message { |
| 1090 | if s == nil || s.err != nil { |
| 1091 | return m |
| 1092 | } |
| 1093 | b, err := json.Marshal(messageForSessionIdentity(m)) |
| 1094 | if err != nil { |
| 1095 | s.err = err |
| 1096 | return m |
| 1097 | } |
| 1098 | _, _ = s.h.Write(b) |
| 1099 | _, _ = s.h.Write([]byte{'\n'}) |
| 1100 | return m |
| 1101 | } |
| 1102 | |
| 1103 | // sum reports the accumulated digest; ok is false when hashing was disabled |
| 1104 | // or a message failed to encode, and callers fall back to digestSessionMessages. |
| 1105 | func (s *sessionTranscriptHasher) sum() (digest [sha256.Size]byte, ok bool) { |
| 1106 | if s == nil || s.err != nil { |
| 1107 | return digest, false |
| 1108 | } |
| 1109 | copy(digest[:], s.h.Sum(nil)) |
| 1110 | return digest, true |
| 1111 | } |
| 1112 | |
| 1113 | func messagesHavePrefix(full, prefix []provider.Message) bool { |
| 1114 | if len(prefix) > len(full) { |
| 1115 | return false |
| 1116 | } |
| 1117 | for i := range prefix { |
| 1118 | if !messagesEqualForStorage(full[i], prefix[i]) { |
| 1119 | return false |
| 1120 | } |
| 1121 | } |
| 1122 | return true |
| 1123 | } |
| 1124 | |
| 1125 | // messagesPrefixDigestDepth returns the number of leading messages of msgs |
| 1126 | // whose storage digest equals target, or -1 when no prefix matches. The |
| 1127 | // digest accumulates exactly like digestAndSizeSessionMessages, so a match at |
| 1128 | // depth k means msgs[:k] has the same transcript identity as target. |
| 1129 | func messagesPrefixDigestDepth(msgs []provider.Message, target [sha256.Size]byte) int { |
| 1130 | h := sha256.New() |
| 1131 | sum := make([]byte, 0, sha256.Size) |
| 1132 | for i, m := range msgs { |
| 1133 | m = messageForSessionIdentity(m) |
| 1134 | b, err := json.Marshal(m) |
| 1135 | if err != nil { |
| 1136 | return -1 |
| 1137 | } |
| 1138 | h.Write(b) |
| 1139 | h.Write([]byte{'\n'}) |
| 1140 | sum = h.Sum(sum[:0]) |
| 1141 | if bytes.Equal(sum, target[:]) { |
| 1142 | return i + 1 |
| 1143 | } |
| 1144 | } |
| 1145 | return -1 |
| 1146 | } |
| 1147 | |
| 1148 | // appendCoversPersistedBaseline reports whether an append-shaped write (disk |
| 1149 | // transcript a prefix of next, modulo a compatible leading-system swap) still |
| 1150 | // covers everything this session ever persisted: the baseline digest must be |
| 1151 | // reachable as a prefix of the pending snapshot, and the disk transcript must |
| 1152 | // still extend at least to that depth. A shorter disk transcript means some |
| 1153 | // other runtime deliberately rewound below the baseline — appending over it |
| 1154 | // would resurrect the removed suffix, so the caller must conflict instead. |
| 1155 | func appendCoversPersistedBaseline(next, existing []provider.Message, baseDigest [sha256.Size]byte) bool { |
| 1156 | depth := messagesPrefixDigestDepth(next, baseDigest) |
| 1157 | if depth < 0 && len(next) > 0 && len(existing) > 0 && |
| 1158 | next[0].Role == provider.RoleSystem && existing[0].Role == provider.RoleSystem && |
| 1159 | !messagesEqualForStorage(next[0], existing[0]) { |
| 1160 | // A resume that swapped the system prompt persisted its baseline with |
| 1161 | // the previous system message — the one still on disk. Re-anchor the |
| 1162 | // search on that message so the swap alone doesn't hide the baseline. |
| 1163 | variant := append([]provider.Message{existing[0]}, next[1:]...) |
| 1164 | depth = messagesPrefixDigestDepth(variant, baseDigest) |
| 1165 | } |
| 1166 | return depth >= 0 && len(existing) >= depth |
| 1167 | } |
| 1168 | |
| 1169 | func messagesHavePrefixWithCompatibleSystem(full, prefix []provider.Message) bool { |
| 1170 | full = messagesWithoutLeadingSystem(full) |
| 1171 | prefix = messagesWithoutLeadingSystem(prefix) |
| 1172 | return messagesHavePrefix(full, prefix) |
| 1173 | } |
| 1174 | |
| 1175 | func messagesWithoutLeadingSystem(msgs []provider.Message) []provider.Message { |
| 1176 | if len(msgs) > 0 && msgs[0].Role == provider.RoleSystem { |
| 1177 | return msgs[1:] |
| 1178 | } |
| 1179 | return msgs |
| 1180 | } |
| 1181 | |
| 1182 | func messagesEqualForStorage(a, b provider.Message) bool { |
| 1183 | a = messageForSessionIdentity(a) |
| 1184 | b = messageForSessionIdentity(b) |
| 1185 | ab, err := json.Marshal(a) |
| 1186 | if err != nil { |
| 1187 | return false |
| 1188 | } |
| 1189 | bb, err := json.Marshal(b) |
| 1190 | if err != nil { |
| 1191 | return false |
| 1192 | } |
| 1193 | return bytes.Equal(ab, bb) |
| 1194 | } |
| 1195 | |
| 1196 | func messagesEqualForStorageList(a, b []provider.Message) bool { |
| 1197 | if len(a) != len(b) { |
| 1198 | return false |
| 1199 | } |
| 1200 | for i := range a { |
| 1201 | if !messagesEqualForStorage(a[i], b[i]) { |
| 1202 | return false |
| 1203 | } |
| 1204 | } |
| 1205 | return true |
| 1206 | } |
| 1207 | |
| 1208 | func messagesCompatibleForStorageBaseline(a, b []provider.Message) bool { |
| 1209 | if messagesEqualForStorageList(a, b) { |
| 1210 | return true |
| 1211 | } |
| 1212 | return messagesEqualForStorageList(messagesWithoutLeadingSystem(a), messagesWithoutLeadingSystem(b)) |
| 1213 | } |
| 1214 | |
| 1215 | func lockSessionSavePath(path string) func() { |
| 1216 | key := canonicalSessionSavePath(path) |
| 1217 | v, _ := sessionSaveLocks.LoadOrStore(key, &sync.Mutex{}) |
| 1218 | mu := v.(*sync.Mutex) |
| 1219 | mu.Lock() |
| 1220 | return mu.Unlock |
| 1221 | } |
| 1222 | |
| 1223 | // tryLockSessionSavePath lets low-priority maintenance yield immediately to a |
| 1224 | // foreground save. The returned unlock is non-nil only when acquired. |
| 1225 | func tryLockSessionSavePath(path string) (func(), bool) { |
| 1226 | key := canonicalSessionSavePath(path) |
| 1227 | v, _ := sessionSaveLocks.LoadOrStore(key, &sync.Mutex{}) |
| 1228 | mu := v.(*sync.Mutex) |
| 1229 | if !mu.TryLock() { |
| 1230 | return nil, false |
| 1231 | } |
| 1232 | return mu.Unlock, true |
| 1233 | } |
| 1234 | |
| 1235 | // lockSessionFile waits briefly for the cross-process compatibility save lock. |
| 1236 | // A short overlap with a legitimate writer is allowed to settle, but an |
| 1237 | // stalled or indefinitely held lock fails the save instead of freezing tab |
| 1238 | // switching or application shutdown. The caller keeps its in-memory transcript |
| 1239 | // and can retry through the existing autosave/recovery paths. |
| 1240 | func lockSessionFile(path string) (func(), error) { |
| 1241 | wait := sessionFileLockWait |
| 1242 | poll := sessionFileLockPollInterval |
| 1243 | if poll <= 0 { |
| 1244 | poll = time.Millisecond |
| 1245 | } |
| 1246 | deadline := time.Now().Add(wait) |
| 1247 | for { |
| 1248 | unlock, err := tryLockSessionFile(path) |
| 1249 | if err == nil { |
| 1250 | return unlock, nil |
| 1251 | } |
| 1252 | if !errors.Is(err, ErrSessionFileLockHeld) { |
| 1253 | return nil, err |
| 1254 | } |
| 1255 | remaining := time.Until(deadline) |
| 1256 | if wait <= 0 || remaining <= 0 { |
| 1257 | return nil, ErrSessionFileLockHeld |
| 1258 | } |
| 1259 | if poll > remaining { |
| 1260 | poll = remaining |
| 1261 | } |
| 1262 | time.Sleep(poll) |
| 1263 | } |
| 1264 | } |
| 1265 | |
| 1266 | // UpdateBranchMeta is the owner-level metadata API. The callback runs while |
| 1267 | // the cross-process metadata lock is held, so callers cannot accidentally |
| 1268 | // load a stale sidecar and overwrite fields written by another runtime. |
| 1269 | func UpdateBranchMeta(path string, touchUpdated bool, update func(*BranchMeta) error) error { |
| 1270 | unlock, err := LockSessionMetaPath(path) |
| 1271 | if err != nil { |
| 1272 | return err |
| 1273 | } |
| 1274 | defer unlock() |
| 1275 | m, err := ensureBranchMetaUnlocked(path) |
| 1276 | if err != nil { |
| 1277 | return err |
| 1278 | } |
| 1279 | if update != nil { |
| 1280 | if err := update(&m); err != nil { |
| 1281 | return err |
| 1282 | } |
| 1283 | } |
| 1284 | // The callback mutated the latest record while the cross-process lock was |
| 1285 | // held, so write it verbatim. Re-merging title fields here would undo an |
| 1286 | // intentional title mutation (including same-value saves that advance the |
| 1287 | // opaque title revision). Whole-record transcript/listing writers use the |
| 1288 | // preserving saveBranchMeta path instead. |
| 1289 | return saveBranchMetaContextMode(context.Background(), path, m, touchUpdated, false) |
| 1290 | } |
| 1291 | |
| 1292 | // LoadSession reads a saved session into a fresh Session value. New sessions |
| 1293 | // replay the append-only event log; legacy sessions without an event log fall |
| 1294 | // back to the compatibility .jsonl checkpoint. A damaged log is replayed to its |
| 1295 | // last clean record (or the checkpoint when nothing decodes) and flagged so the |
| 1296 | // next save heals it with a rewrite-and-compact. |
| 1297 | // In-process loads share the save path mutex so they cannot observe a local |
| 1298 | // SaveSnapshot between appending an event-log record and refreshing the index. |
| 1299 | // Missing files surface as os.IsNotExist so callers can fall through to a |
| 1300 | // new session. |
| 1301 | func LoadSession(path string) (*Session, error) { |
| 1302 | unlock := lockSessionSavePath(path) |
| 1303 | defer unlock() |
| 1304 | return loadSessionUnlocked(path) |
| 1305 | } |
| 1306 | |
| 1307 | // LoadSessionContext restores an interactive runtime without accepting a |
| 1308 | // damaged prefix as its complete execution context. Both save-lock admission |
| 1309 | // and replay are cancellable; ordinary replay budgets remain in force. |
| 1310 | func LoadSessionContext(ctx context.Context, path string) (*Session, error) { |
| 1311 | for { |
| 1312 | if err := ctx.Err(); err != nil { |
| 1313 | return nil, err |
| 1314 | } |
| 1315 | if unlock, ok := tryLockSessionSavePath(path); ok { |
| 1316 | defer unlock() |
| 1317 | return loadSessionUnlockedWithContextMode(ctx, path, defaultSessionReplayLimits, true) |
| 1318 | } |
| 1319 | timer := time.NewTimer(5 * time.Millisecond) |
| 1320 | select { |
| 1321 | case <-ctx.Done(): |
| 1322 | timer.Stop() |
| 1323 | return nil, ctx.Err() |
| 1324 | case <-timer.C: |
| 1325 | } |
| 1326 | } |
| 1327 | } |
| 1328 | |
| 1329 | func loadSessionUnlocked(path string) (*Session, error) { |
| 1330 | return loadSessionUnlockedWithContext(context.Background(), path, defaultSessionReplayLimits) |
| 1331 | } |
| 1332 | |
| 1333 | func loadSessionUnlockedWithLimits(path string, limits sessionReplayLimits) (*Session, error) { |
| 1334 | return loadSessionUnlockedWithContext(context.Background(), path, limits) |
| 1335 | } |
| 1336 | |
| 1337 | func loadSessionUnlockedWithContext(ctx context.Context, path string, limits sessionReplayLimits) (*Session, error) { |
| 1338 | return loadSessionUnlockedWithContextMode(ctx, path, limits, false) |
| 1339 | } |
| 1340 | |
| 1341 | func loadSessionUnlockedWithContextMode(ctx context.Context, path string, limits sessionReplayLimits, rejectDamage bool) (*Session, error) { |
| 1342 | hasher := newSessionTranscriptHasher() |
| 1343 | res, err := loadSessionTranscript(ctx, path, limits, hasher) |
| 1344 | if err != nil { |
| 1345 | return nil, err |
| 1346 | } |
| 1347 | if rejectDamage && res.damaged { |
| 1348 | return nil, fmt.Errorf("%w: authoritative event log has no complete recoverable prefix", ErrSessionHistoryDamaged) |
| 1349 | } |
| 1350 | msgs := res.msgs |
| 1351 | persistFormat := sessionPersistLegacy |
| 1352 | if res.dag { |
| 1353 | persistFormat = sessionPersistDAG |
| 1354 | } |
| 1355 | s := &Session{Messages: msgs, eventLogDamaged: res.damaged, persistFormat: persistFormat, head: sessionHeadState{ref: res.head, dag: res.dag, headCount: res.headCount, state: res.state, openTurn: res.openTurn, events: res.events}} |
| 1356 | // Repair persisted-history-safe issues before anything reads the session. |
| 1357 | // Old sessions (pre adde2d3e) and interrupted turns can carry empty tool-call |
| 1358 | // names, dangling tool_calls, or half-streamed argument JSON that DeepSeek |
| 1359 | // rejects with a 400 on replay. Wire-only cleanup, such as dropping orphan |
| 1360 | // tool messages, stays in the provider send path so Save/LoadSession keeps |
| 1361 | // its round-trip contract. The fast path returns the input slice unchanged |
| 1362 | // for a well-formed history, so we detect an actual repair by comparing |
| 1363 | // slice headers: when NormalizeSession allocated a new backing array, the |
| 1364 | // session is marked dirty so the next Save persists the fix. |
| 1365 | normalized := NormalizeSession(s.Messages) |
| 1366 | normalized = migrateLegacyProviderContent(normalized) |
| 1367 | if len(normalized) != len(s.Messages) || (len(s.Messages) > 0 && &normalized[0] != &s.Messages[0]) { |
| 1368 | s.normalizedDirty = true |
| 1369 | // Keep the pre-repair transcript: checkSnapshotWrite must be able to |
| 1370 | // recognize a snapshot that extends the bytes actually on disk, which |
| 1371 | // the repaired view no longer represents (an interrupted tool turn |
| 1372 | // gets a placeholder result fabricated here that the live session |
| 1373 | // answered for real). |
| 1374 | s.rawMessages = msgs |
| 1375 | } |
| 1376 | s.Messages = normalized |
| 1377 | assignLegacyMessageIDs(path, s.Messages) |
| 1378 | // Decode already hashed the transcript; when the repairs above returned it |
| 1379 | // unchanged (the common case) reuse that digest, else re-hash the repair. |
| 1380 | digest, digestOK := hasher.sum() |
| 1381 | if s.normalizedDirty || !digestOK { |
| 1382 | if d, derr := digestSessionMessages(s.Messages); derr == nil { |
| 1383 | digest, digestOK = d, true |
| 1384 | } else { |
| 1385 | digestOK = false |
| 1386 | } |
| 1387 | } |
| 1388 | if digestOK { |
| 1389 | // Pair the raw pre-repair transcript when normalization changed it. |
| 1390 | diskView := s.Messages |
| 1391 | if s.normalizedDirty { |
| 1392 | diskView = s.rawMessages |
| 1393 | } |
| 1394 | if meta, ok, metaErr := loadBranchMetaRetry(path); metaErr != nil { |
| 1395 | // The sidecar exists but is unreadable even after retries (torn or |
| 1396 | // corrupt). The session must still open, but revision 0 must not |
| 1397 | // pose as a real baseline: the next save would misread the honest |
| 1398 | // on-disk revision as another runtime's write and fork a recovery |
| 1399 | // branch. Anchor the baseline on digest+version only until a |
| 1400 | // successful save re-learns the revision. |
| 1401 | s.markPersistedRevisionUnknown(path, digest, s.version, s.rewriteVersion, diskView) |
| 1402 | } else { |
| 1403 | revision := int64(0) |
| 1404 | if ok { |
| 1405 | revision = meta.Revision |
| 1406 | } |
| 1407 | s.markPersistedFromLoad(path, digest, s.version, revision, s.rewriteVersion, diskView) |
| 1408 | } |
| 1409 | } |
| 1410 | return s, nil |
| 1411 | } |
| 1412 | |
| 1413 | // LoadSessionForMigration reads a frozen legacy transcript without applying |
| 1414 | // the interactive 128 MiB/record-count replay budgets. Callers must first stop |
| 1415 | // writers and freeze the source bytes; ordinary UI and runtime opens must keep |
| 1416 | // using LoadSession so an untrusted or corrupt live file cannot exhaust memory. |
| 1417 | func LoadSessionForMigration(ctx context.Context, path string) (*Session, error) { |
| 1418 | if ctx == nil { |
| 1419 | ctx = context.Background() |
| 1420 | } |
| 1421 | unlock := lockSessionSavePath(path) |
| 1422 | defer unlock() |
| 1423 | return loadSessionUnlockedWithContextMode(ctx, path, migrationSessionReplayLimits(), true) |
| 1424 | } |
| 1425 | |
| 1426 | // SessionInfo summarises a saved session for the --resume picker: where it is on |
| 1427 | // disk, when it was created/last active, the first user message as a preview, and |
| 1428 | // a rough turn count. |
| 1429 | type SessionInfo struct { |
| 1430 | Path string |
| 1431 | CreatedAt time.Time |
| 1432 | LastActivityAt time.Time |
| 1433 | ModTime time.Time // compatibility alias for LastActivityAt |
| 1434 | Preview string |
| 1435 | Turns int |
| 1436 | CountsKnown bool |
| 1437 | Scope string |
| 1438 | WorkspaceRoot string |
| 1439 | TopicID string |
| 1440 | TopicTitle string |
| 1441 | CustomTitle string |
| 1442 | Recovered bool |
| 1443 | RecoveryReason string |
| 1444 | RecoveryDigest string |
| 1445 | ParentID string |
| 1446 | VersionKind SessionVersionKind |
| 1447 | VersionState SessionVersionState |
| 1448 | ParentConversationID string |
| 1449 | ParentVersionID string |
| 1450 | } |
| 1451 | |
| 1452 | // CleanupPendingMeta records that a session was logically removed but still has |
| 1453 | // artifacts waiting for a background job to unwind before physical cleanup. |
| 1454 | type CleanupPendingMeta struct { |
| 1455 | Operation string `json:"operation"` |
| 1456 | CreatedAt int64 `json:"createdAt"` |
| 1457 | } |
| 1458 | |
| 1459 | // CleanupPendingInfo describes one durable delayed-cleanup marker and the |
| 1460 | // session transcript it belongs to. |
| 1461 | type CleanupPendingInfo struct { |
| 1462 | SessionPath string |
| 1463 | MarkerPath string |
| 1464 | Meta CleanupPendingMeta |
| 1465 | } |
| 1466 | |
| 1467 | // CleanupPendingPath returns the durable marker path for a session transcript. |
| 1468 | func CleanupPendingPath(sessionPath string) string { |
| 1469 | return store.SessionCleanupPending(sessionPath) |
| 1470 | } |
| 1471 | |
| 1472 | // MarkCleanupPending hides a logically removed session from resume/list surfaces |
| 1473 | // until delayed physical cleanup has finished. |
| 1474 | func MarkCleanupPending(sessionPath, operation string) error { |
| 1475 | path := CleanupPendingPath(sessionPath) |
| 1476 | if path == "" { |
| 1477 | return nil |
| 1478 | } |
| 1479 | if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { |
| 1480 | return err |
| 1481 | } |
| 1482 | meta := CleanupPendingMeta{Operation: strings.TrimSpace(operation), CreatedAt: time.Now().UnixMilli()} |
| 1483 | b, err := json.MarshalIndent(meta, "", " ") |
| 1484 | if err != nil { |
| 1485 | return err |
| 1486 | } |
| 1487 | // The marker controls session visibility during delayed cleanup. Publish it |
| 1488 | // atomically so a crash cannot leave malformed JSON that hides the session |
| 1489 | // and blocks reconciliation on the next startup. |
| 1490 | return fileutil.AtomicWriteFile(path, b, 0o644) |
| 1491 | } |
| 1492 | |
| 1493 | // ClearCleanupPending removes a delayed-cleanup marker after physical cleanup. |
| 1494 | func ClearCleanupPending(sessionPath string) error { |
| 1495 | path := CleanupPendingPath(sessionPath) |
| 1496 | if path == "" { |
| 1497 | return nil |
| 1498 | } |
| 1499 | if err := os.Remove(path); err != nil && !os.IsNotExist(err) { |
| 1500 | return err |
| 1501 | } |
| 1502 | return nil |
| 1503 | } |
| 1504 | |
| 1505 | // IsCleanupPending reports whether a session is hidden pending delayed cleanup. |
| 1506 | func IsCleanupPending(sessionPath string) bool { |
| 1507 | path := CleanupPendingPath(sessionPath) |
| 1508 | if path == "" { |
| 1509 | return false |
| 1510 | } |
| 1511 | _, err := os.Stat(path) |
| 1512 | return err == nil |
| 1513 | } |
| 1514 | |
| 1515 | // IsVisibleSession reports whether a persisted session should appear on normal |
| 1516 | // user/agent-facing list, restore, and retrieval surfaces. |
| 1517 | func IsVisibleSession(sessionPath string) bool { |
| 1518 | return strings.TrimSpace(sessionPath) != "" && !IsCleanupPending(sessionPath) |
| 1519 | } |
| 1520 | |
| 1521 | // ListCleanupPending returns delayed-cleanup markers left in dir. A missing |
| 1522 | // directory is not an error. |
| 1523 | func ListCleanupPending(dir string) ([]CleanupPendingInfo, error) { |
| 1524 | dir = strings.TrimSpace(dir) |
| 1525 | if dir == "" { |
| 1526 | return nil, nil |
| 1527 | } |
| 1528 | entries, err := os.ReadDir(dir) |
| 1529 | if err != nil { |
| 1530 | if os.IsNotExist(err) { |
| 1531 | return nil, nil |
| 1532 | } |
| 1533 | return nil, err |
| 1534 | } |
| 1535 | var out []CleanupPendingInfo |
| 1536 | for _, e := range entries { |
| 1537 | if e.IsDir() || !strings.HasSuffix(e.Name(), cleanupPendingExt) { |
| 1538 | continue |
| 1539 | } |
| 1540 | markerPath := filepath.Join(dir, e.Name()) |
| 1541 | var meta CleanupPendingMeta |
| 1542 | b, err := fileencoding.ReadFileUTF8(markerPath) |
| 1543 | if err != nil { |
| 1544 | if os.IsNotExist(err) { |
| 1545 | continue |
| 1546 | } |
| 1547 | return nil, err |
| 1548 | } |
| 1549 | if strings.TrimSpace(string(b)) != "" { |
| 1550 | if err := json.Unmarshal(b, &meta); err != nil { |
| 1551 | return nil, fmt.Errorf("read cleanup-pending marker %s: %w", markerPath, err) |
| 1552 | } |
| 1553 | } |
| 1554 | name := strings.TrimSuffix(e.Name(), cleanupPendingExt) + ".jsonl" |
| 1555 | out = append(out, CleanupPendingInfo{ |
| 1556 | SessionPath: filepath.Join(dir, name), |
| 1557 | MarkerPath: markerPath, |
| 1558 | Meta: meta, |
| 1559 | }) |
| 1560 | } |
| 1561 | sort.Slice(out, func(i, j int) bool { |
| 1562 | return out[i].SessionPath < out[j].SessionPath |
| 1563 | }) |
| 1564 | return out, nil |
| 1565 | } |
| 1566 | |
| 1567 | // ReconcileCleanupPending retries physical cleanup for leftover delayed-cleanup |
| 1568 | // markers and stale lock/lease sidecars. It keeps going after individual |
| 1569 | // cleanup errors and returns them joined. |
| 1570 | func ReconcileCleanupPending(dir string, cleanup func(CleanupPendingInfo) error) error { |
| 1571 | var errs []error |
| 1572 | if err := ReconcileSessionSidecars(dir); err != nil { |
| 1573 | errs = append(errs, err) |
| 1574 | } |
| 1575 | if err := reconcileRecoveryTrashStages(dir); err != nil { |
| 1576 | errs = append(errs, err) |
| 1577 | } |
| 1578 | pending, err := ListCleanupPending(dir) |
| 1579 | if err != nil { |
| 1580 | errs = append(errs, err) |
| 1581 | return errors.Join(errs...) |
| 1582 | } |
| 1583 | for _, item := range pending { |
| 1584 | handled, err := reconcileRecoveryTrashPending(item) |
| 1585 | if !handled { |
| 1586 | if cleanup == nil { |
| 1587 | continue |
| 1588 | } |
| 1589 | err = cleanup(item) |
| 1590 | } |
| 1591 | if err != nil { |
| 1592 | errs = append(errs, fmt.Errorf("%s: %w", item.SessionPath, err)) |
| 1593 | } |
| 1594 | } |
| 1595 | return errors.Join(errs...) |
| 1596 | } |
| 1597 | |
| 1598 | // ReconcileSessionSidecars renames transcripts whose filenames outgrew their |
| 1599 | // sidecars and removes stale lock and lease files left beside sessions by |
| 1600 | // older runtimes. It never removes .jsonl transcripts; recovered conversations |
| 1601 | // may contain useful user history even when their names are ugly. |
| 1602 | func ReconcileSessionSidecars(dir string) error { |
| 1603 | dir = strings.TrimSpace(dir) |
| 1604 | if dir == "" { |
| 1605 | return nil |
| 1606 | } |
| 1607 | var errs []error |
| 1608 | if err := reconcileOverlongSessionFilenames(dir); err != nil { |
| 1609 | errs = append(errs, err) |
| 1610 | } |
| 1611 | // Re-list after the rename pass: it retires old names and their sidecars. |
| 1612 | entries, err := os.ReadDir(dir) |
| 1613 | if err != nil { |
| 1614 | if os.IsNotExist(err) { |
| 1615 | return errors.Join(errs...) |
| 1616 | } |
| 1617 | errs = append(errs, err) |
| 1618 | return errors.Join(errs...) |
| 1619 | } |
| 1620 | for _, e := range entries { |
| 1621 | if e.IsDir() { |
| 1622 | continue |
| 1623 | } |
| 1624 | name := e.Name() |
| 1625 | sidecarPath := filepath.Join(dir, name) |
| 1626 | switch { |
| 1627 | case strings.HasSuffix(name, sessionLeaseInfoSidecarSuffix): |
| 1628 | base := filepath.Join(dir, strings.TrimSuffix(name, ".lease.json")) |
| 1629 | if err := removeStaleSessionLeaseInfoSidecar(base, sidecarPath); err != nil { |
| 1630 | errs = append(errs, fmt.Errorf("%s: %w", sidecarPath, err)) |
| 1631 | } |
| 1632 | case strings.HasSuffix(name, sessionLeaseLockSidecarSuffix): |
| 1633 | base := filepath.Join(dir, strings.TrimSuffix(name, ".lease.lock")) |
| 1634 | if err := removeStaleSessionLeaseLockSidecar(base, sidecarPath); err != nil { |
| 1635 | errs = append(errs, fmt.Errorf("%s: %w", sidecarPath, err)) |
| 1636 | } |
| 1637 | case strings.HasSuffix(name, sessionLockSidecarSuffix): |
| 1638 | base := filepath.Join(dir, strings.TrimSuffix(name, ".lock")) |
| 1639 | if err := removeStaleSessionLockSidecar(base, sidecarPath); err != nil { |
| 1640 | errs = append(errs, fmt.Errorf("%s: %w", sidecarPath, err)) |
| 1641 | } |
| 1642 | } |
| 1643 | } |
| 1644 | return errors.Join(errs...) |
| 1645 | } |
| 1646 | |
| 1647 | func removeStaleSessionLockSidecar(basePath, sidecarPath string) error { |
| 1648 | basePath = canonicalSessionSavePath(basePath) |
| 1649 | if sessionLeaseHeldLocally(basePath) || SessionLeaseHeldByOtherRuntime(basePath) { |
| 1650 | return nil |
| 1651 | } |
| 1652 | lock, err := tryTakeSessionLockFile(sidecarPath) |
| 1653 | if err != nil { |
| 1654 | if errors.Is(err, ErrSessionFileLockHeld) { |
| 1655 | return nil |
| 1656 | } |
| 1657 | return err |
| 1658 | } |
| 1659 | // The removal is atomic with the release (unlink-under-flock on Unix, |
| 1660 | // delete-disposition on the held handle on Windows), so a concurrent |
| 1661 | // saver can never acquire a lock file that is being deleted under it. |
| 1662 | return lock.RemoveAndUnlock() |
| 1663 | } |
| 1664 | |
| 1665 | // removeStaleSessionLeaseLockSidecar retires a leftover .lease.lock. The file |
| 1666 | // is the lease lock itself, so taking it non-blocking proves no runtime holds |
| 1667 | // the lease, and RemoveAndUnlock deletes it atomically with the release. |
| 1668 | func removeStaleSessionLeaseLockSidecar(basePath, _ string) error { |
| 1669 | basePath = canonicalSessionSavePath(basePath) |
| 1670 | if sessionLeaseHeldLocally(basePath) { |
| 1671 | return nil |
| 1672 | } |
| 1673 | lock, err := tryTakeSessionLeaseLock(basePath) |
| 1674 | if err != nil { |
| 1675 | if errors.Is(err, ErrSessionLeaseHeld) { |
| 1676 | return nil |
| 1677 | } |
| 1678 | return err |
| 1679 | } |
| 1680 | return lock.RemoveAndUnlock() |
| 1681 | } |
| 1682 | |
| 1683 | // removeStaleSessionLeaseInfoSidecar retires a leftover .lease.json while |
| 1684 | // holding the lease lock, so no runtime can adopt the info file mid-removal. |
| 1685 | // The info file itself is never held open by anyone, so a plain remove under |
| 1686 | // the lock is safe on every platform. |
| 1687 | func removeStaleSessionLeaseInfoSidecar(basePath, sidecarPath string) error { |
| 1688 | basePath = canonicalSessionSavePath(basePath) |
| 1689 | if sessionLeaseHeldLocally(basePath) { |
| 1690 | return nil |
| 1691 | } |
| 1692 | lockPath := basePath + ".lease.lock" |
| 1693 | if _, err := os.Stat(lockPath); err == nil { |
| 1694 | unlock, err := tryLockSessionLeaseFile(basePath) |
| 1695 | if err != nil { |
| 1696 | if errors.Is(err, ErrSessionLeaseHeld) { |
| 1697 | return nil |
| 1698 | } |
| 1699 | return err |
| 1700 | } |
| 1701 | removeErr := os.Remove(sidecarPath) |
| 1702 | if unlock != nil { |
| 1703 | unlock() |
| 1704 | } |
| 1705 | if removeErr != nil && !os.IsNotExist(removeErr) { |
| 1706 | return removeErr |
| 1707 | } |
| 1708 | return nil |
| 1709 | } else if !os.IsNotExist(err) { |
| 1710 | return err |
| 1711 | } |
| 1712 | // No lease lock file: holders keep it present (and locked) for their whole |
| 1713 | // lifetime, so the leftover info sidecar has no owner to race with. |
| 1714 | if err := os.Remove(sidecarPath); err != nil && !os.IsNotExist(err) { |
| 1715 | return err |
| 1716 | } |
| 1717 | return nil |
| 1718 | } |
| 1719 | |
| 1720 | func sessionLeaseHeldLocally(path string) bool { |
| 1721 | _, ok := sessionLeaseOwners.Load(CanonicalSessionPath(path)) |
| 1722 | return ok |
| 1723 | } |
| 1724 | |
| 1725 | // sessionLockSidecarFits reports whether basePath's .lock sidecar name stays |
| 1726 | // within the filesystem's per-component limit; past it, no process can hold |
| 1727 | // (or ever have held) the file lock, because the lock file cannot be created. |
| 1728 | func sessionLockSidecarFits(basePath string) bool { |
| 1729 | return len(filepath.Base(basePath))+len(".lock") <= nameMaxBytes |
| 1730 | } |
| 1731 | |
| 1732 | // sessionLeaseSidecarFits is the lease-file analogue of sessionLockSidecarFits. |
| 1733 | func sessionLeaseSidecarFits(basePath string) bool { |
| 1734 | return len(filepath.Base(basePath))+len(".lease.lock") <= nameMaxBytes |
| 1735 | } |
| 1736 | |
| 1737 | // reconcileOverlongSessionFilenames renames transcripts whose basenames grew |
| 1738 | // past maxSessionBasenameBytes — the leftover shape of the pre-bounded |
| 1739 | // recovery cascade (#5923), where lock and lease sidecars could no longer be |
| 1740 | // created and the session became unsaveable. The conversation bytes are kept |
| 1741 | // verbatim under a bounded name derived the same way new recovery branches |
| 1742 | // are named; branch meta moves along with its ID rewritten, and sessions |
| 1743 | // pointing at the old ID are re-parented so lineage survives the rename. |
| 1744 | func reconcileOverlongSessionFilenames(dir string) error { |
| 1745 | entries, err := os.ReadDir(dir) |
| 1746 | if err != nil { |
| 1747 | if os.IsNotExist(err) { |
| 1748 | return nil |
| 1749 | } |
| 1750 | return err |
| 1751 | } |
| 1752 | var errs []error |
| 1753 | renamed := map[string]string{} // old branch ID -> new branch ID |
| 1754 | for _, e := range entries { |
| 1755 | name := e.Name() |
| 1756 | if e.IsDir() || !store.IsSessionTranscriptName(name) { |
| 1757 | continue |
| 1758 | } |
| 1759 | if len(name) <= maxSessionBasenameBytes { |
| 1760 | continue |
| 1761 | } |
| 1762 | oldPath := filepath.Join(dir, name) |
| 1763 | if IsCleanupPending(oldPath) { |
| 1764 | // Being deleted; renaming would orphan the cleanup marker. |
| 1765 | continue |
| 1766 | } |
| 1767 | newID, err := renameOverlongSession(oldPath) |
| 1768 | if err != nil { |
| 1769 | errs = append(errs, fmt.Errorf("%s: %w", oldPath, err)) |
| 1770 | } |
| 1771 | // A non-empty newID means the transcript rename landed even if some |
| 1772 | // sidecar migration failed; record it so children still re-parent — |
| 1773 | // this run is the only one that knows the old-to-new mapping. |
| 1774 | if newID != "" { |
| 1775 | renamed[BranchID(oldPath)] = newID |
| 1776 | } |
| 1777 | } |
| 1778 | if len(renamed) > 0 { |
| 1779 | if err := reparentSessionBranches(dir, renamed); err != nil { |
| 1780 | errs = append(errs, err) |
| 1781 | } |
| 1782 | } |
| 1783 | return errors.Join(errs...) |
| 1784 | } |
| 1785 | |
| 1786 | // renameOverlongSession moves one overlong transcript to its bounded name and |
| 1787 | // migrates the sidecars that carry user state. It returns the new branch ID, |
| 1788 | // or "" when the session was skipped because a runtime may still own it. |
| 1789 | func renameOverlongSession(oldPath string) (string, error) { |
| 1790 | oldID := BranchID(oldPath) |
| 1791 | newID := recoveryParentStem(oldID) |
| 1792 | if newID == oldID { |
| 1793 | return "", nil |
| 1794 | } |
| 1795 | newPath := filepath.Join(filepath.Dir(oldPath), newID+".jsonl") |
| 1796 | if _, err := os.Stat(newPath); err == nil { |
| 1797 | return "", fmt.Errorf("rename target %s already exists", filepath.Base(newPath)) |
| 1798 | } else if !os.IsNotExist(err) { |
| 1799 | return "", err |
| 1800 | } |
| 1801 | unlockOld := lockSessionSavePath(oldPath) |
| 1802 | defer unlockOld() |
| 1803 | unlockNew := lockSessionSavePath(newPath) |
| 1804 | defer unlockNew() |
| 1805 | if sessionLeaseHeldLocally(oldPath) { |
| 1806 | return "", nil |
| 1807 | } |
| 1808 | // Names past the sidecar limit cannot have lease or lock holders in any |
| 1809 | // process — the holder files themselves are uncreatable — so probing them |
| 1810 | // would only manufacture ENAMETOOLONG errors and wrongly skip the exact |
| 1811 | // sessions this pass exists to repair. |
| 1812 | if sessionLeaseSidecarFits(oldPath) && SessionLeaseHeldByOtherRuntime(oldPath) { |
| 1813 | return "", nil |
| 1814 | } |
| 1815 | var lockFile *sessionLockFile |
| 1816 | if sessionLockSidecarFits(oldPath) { |
| 1817 | lock, err := tryTakeSessionLockFile(oldPath + ".lock") |
| 1818 | if err != nil { |
| 1819 | if errors.Is(err, ErrSessionFileLockHeld) { |
| 1820 | return "", nil |
| 1821 | } |
| 1822 | return "", err |
| 1823 | } |
| 1824 | lockFile = lock |
| 1825 | } |
| 1826 | if err := os.Rename(oldPath, newPath); err != nil { |
| 1827 | // Nothing moved: the old transcript is intact and the next |
| 1828 | // reconciliation can retry, so its lock file stays in place too. |
| 1829 | if lockFile != nil { |
| 1830 | lockFile.Unlock() |
| 1831 | } |
| 1832 | return "", err |
| 1833 | } |
| 1834 | // The transcript is committed under its new name from here on. Sidecar |
| 1835 | // migration and lock cleanup failures are reported, but the new ID is |
| 1836 | // still returned so the caller re-parents children: the old name is gone, |
| 1837 | // and a later run would have no way to reconstruct this mapping. |
| 1838 | var errs []error |
| 1839 | if err := migrateSessionSidecars(oldPath, newPath, newID); err != nil { |
| 1840 | errs = append(errs, err) |
| 1841 | } |
| 1842 | // Retire the old disposable lease sidecars: any holder was ruled out |
| 1843 | // above, and nothing keeps these files open, so a plain remove is safe. |
| 1844 | if sessionLeaseSidecarFits(oldPath) { |
| 1845 | for _, stale := range []string{oldPath + ".lease.lock", oldPath + ".lease.json"} { |
| 1846 | if err := os.Remove(stale); err != nil && !os.IsNotExist(err) { |
| 1847 | errs = append(errs, err) |
| 1848 | } |
| 1849 | } |
| 1850 | } |
| 1851 | // The old .lock goes atomically with the release of the lock we hold on it. |
| 1852 | if lockFile != nil { |
| 1853 | if err := lockFile.RemoveAndUnlock(); err != nil { |
| 1854 | errs = append(errs, err) |
| 1855 | } |
| 1856 | } |
| 1857 | return newID, errors.Join(errs...) |
| 1858 | } |
| 1859 | |
| 1860 | // migrateSessionSidecars moves the user-state sidecars of a renamed session: |
| 1861 | // branch meta (with its ID rewritten to match the new filename), goal state, |
| 1862 | // and the checkpoint/job directories. Lock and lease files are disposable and |
| 1863 | // are removed by the caller instead. |
| 1864 | func migrateSessionSidecars(oldPath, newPath, newID string) error { |
| 1865 | errs := []error{migratePinnedContextSidecar(oldPath, newPath, newID)} |
| 1866 | if len(filepath.Base(oldPath))+len(".meta") <= nameMaxBytes { |
| 1867 | if meta, ok, err := LoadBranchMeta(oldPath); err != nil { |
| 1868 | errs = append(errs, err) |
| 1869 | } else if ok { |
| 1870 | meta.ID = newID |
| 1871 | if err := SaveBranchMetaPreserveUpdated(newPath, meta); err != nil { |
| 1872 | errs = append(errs, err) |
| 1873 | } else if err := os.Remove(BranchMetaPath(oldPath)); err != nil && !os.IsNotExist(err) { |
| 1874 | errs = append(errs, err) |
| 1875 | } |
| 1876 | } |
| 1877 | } |
| 1878 | for _, pair := range [][2]string{ |
| 1879 | {store.SessionGoalState(oldPath), store.SessionGoalState(newPath)}, |
| 1880 | {store.SessionEventLog(oldPath), store.SessionEventLog(newPath)}, |
| 1881 | {store.SessionEventLogDamaged(oldPath), store.SessionEventLogDamaged(newPath)}, |
| 1882 | {store.SessionTurnEventLog(oldPath), store.SessionTurnEventLog(newPath)}, |
| 1883 | {store.SessionTurnEventLogDamaged(oldPath), store.SessionTurnEventLogDamaged(newPath)}, |
| 1884 | {store.SessionEventIndex(oldPath), store.SessionEventIndex(newPath)}, |
| 1885 | {store.SessionConflictLog(oldPath), store.SessionConflictLog(newPath)}, |
| 1886 | {store.SessionRecoveryState(oldPath), store.SessionRecoveryState(newPath)}, |
| 1887 | {store.SessionCheckpointDir(oldPath), store.SessionCheckpointDir(newPath)}, |
| 1888 | {store.SessionJobsDir(oldPath), store.SessionJobsDir(newPath)}, |
| 1889 | {store.SessionInboxDir(oldPath), store.SessionInboxDir(newPath)}, |
| 1890 | } { |
| 1891 | // A source name past the filesystem limit cannot exist; renaming it |
| 1892 | // would just manufacture ENAMETOOLONG instead of a clean not-exist. |
| 1893 | if len(filepath.Base(pair[0])) > nameMaxBytes { |
| 1894 | continue |
| 1895 | } |
| 1896 | if err := os.Rename(pair[0], pair[1]); err != nil && !os.IsNotExist(err) { |
| 1897 | errs = append(errs, err) |
| 1898 | } |
| 1899 | } |
| 1900 | return errors.Join(errs...) |
| 1901 | } |
| 1902 | |
| 1903 | // reparentSessionBranches rewrites ParentID references from renamed branch IDs |
| 1904 | // to their bounded replacements so the branch tree stays connected. |
| 1905 | func reparentSessionBranches(dir string, renamed map[string]string) error { |
| 1906 | entries, err := os.ReadDir(dir) |
| 1907 | if err != nil { |
| 1908 | return err |
| 1909 | } |
| 1910 | var errs []error |
| 1911 | for _, e := range entries { |
| 1912 | name := e.Name() |
| 1913 | if e.IsDir() || !store.IsSessionTranscriptName(name) { |
| 1914 | continue |
| 1915 | } |
| 1916 | if len(name)+len(".meta") > nameMaxBytes { |
| 1917 | continue |
| 1918 | } |
| 1919 | path := filepath.Join(dir, name) |
| 1920 | unlock := lockSessionSavePath(path) |
| 1921 | meta, ok, err := LoadBranchMeta(path) |
| 1922 | if err == nil && ok { |
| 1923 | if newParent, hit := renamed[meta.ParentID]; hit && newParent != meta.ParentID { |
| 1924 | meta.ParentID = newParent |
| 1925 | err = SaveBranchMetaPreserveUpdated(path, meta) |
| 1926 | } |
| 1927 | } |
| 1928 | unlock() |
| 1929 | if err != nil { |
| 1930 | errs = append(errs, fmt.Errorf("%s: %w", path, err)) |
| 1931 | } |
| 1932 | } |
| 1933 | return errors.Join(errs...) |
| 1934 | } |
| 1935 | |
| 1936 | // ListSessionOrder returns every *.jsonl session under dir in the same |
| 1937 | // most-recently-active order used by ListSessions, using only file metadata and |
| 1938 | // branch sidecars. A missing directory is not an error. |
| 1939 | func ListSessionOrder(dir string) ([]SessionOrderInfo, error) { |
| 1940 | return ListSessionOrderWithRecoveryPreferenceResolver(dir, RecoveryPreferenceCurrent) |
| 1941 | } |
| 1942 | |
| 1943 | // ListSessionOrderWithRecoveryPreferenceResolver lets catalog reconciliation |
| 1944 | // reuse its wave-local transcript snapshot when validating explicit choices. |
| 1945 | // Other callers keep the ordinary ListSessionOrder behavior above. |
| 1946 | func ListSessionOrderWithRecoveryPreferenceResolver(dir string, resolve RecoveryPreferenceResolver) ([]SessionOrderInfo, error) { |
| 1947 | if resolve == nil { |
| 1948 | resolve = RecoveryPreferenceCurrent |
| 1949 | } |
| 1950 | entries, err := os.ReadDir(dir) |
| 1951 | if err != nil { |
| 1952 | if os.IsNotExist(err) { |
| 1953 | return nil, nil |
| 1954 | } |
| 1955 | return nil, err |
| 1956 | } |
| 1957 | var out []SessionOrderInfo |
| 1958 | for _, e := range entries { |
| 1959 | if e.IsDir() || !store.IsSessionTranscriptName(e.Name()) { |
| 1960 | continue |
| 1961 | } |
| 1962 | info, err := e.Info() |
| 1963 | if err != nil { |
| 1964 | continue |
| 1965 | } |
| 1966 | full := filepath.Join(dir, e.Name()) |
| 1967 | if !IsVisibleSession(full) { |
| 1968 | continue |
| 1969 | } |
| 1970 | contentMod := SessionContentModTime(full) |
| 1971 | if contentMod.IsZero() { |
| 1972 | contentMod = info.ModTime() |
| 1973 | } |
| 1974 | createdAt := info.ModTime() |
| 1975 | lastActivityAt := contentMod |
| 1976 | scope := "global" |
| 1977 | workspaceRoot := "" |
| 1978 | topicID := "" |
| 1979 | topicTitle := "" |
| 1980 | customTitle := "" |
| 1981 | recovered := false |
| 1982 | recoveryReason := "" |
| 1983 | recoveryDigest := "" |
| 1984 | parentID := "" |
| 1985 | versionKind := VersionNormal |
| 1986 | versionState := VersionActive |
| 1987 | parentConversationID := "" |
| 1988 | var headMirror BranchMeta |
| 1989 | parentVersionID := "" |
| 1990 | recoveryPreferred := false |
| 1991 | turns := 0 |
| 1992 | preview := "" |
| 1993 | schemaVersion := 0 |
| 1994 | revision := int64(0) |
| 1995 | contentDigest := "" |
| 1996 | listingRevision := int64(0) |
| 1997 | listingContentDigest := "" |
| 1998 | if meta, ok, err := LoadBranchMeta(full); err == nil && ok { |
| 1999 | if !meta.CreatedAt.IsZero() { |
| 2000 | createdAt = meta.CreatedAt |
| 2001 | } |
| 2002 | if !meta.UpdatedAt.IsZero() { |
| 2003 | lastActivityAt = meta.UpdatedAt |
| 2004 | } |
| 2005 | scope = meta.DefaultScope() |
| 2006 | workspaceRoot = meta.WorkspaceRoot |
| 2007 | topicID = meta.TopicID |
| 2008 | topicTitle = meta.TopicTitle |
| 2009 | customTitle = meta.CustomTitle |
| 2010 | recovered = meta.Recovered |
| 2011 | recoveryReason = meta.RecoveryReason |
| 2012 | recoveryDigest = meta.RecoveryDigest |
| 2013 | parentID = meta.ParentID |
| 2014 | versionKind = meta.EffectiveVersionKind() |
| 2015 | versionState = meta.EffectiveVersionState() |
| 2016 | parentConversationID = meta.ParentConversationID |
| 2017 | parentVersionID = meta.ParentVersionID |
| 2018 | recoveryPreferred = resolve(full, meta) |
| 2019 | turns = meta.Turns |
| 2020 | preview = meta.Preview |
| 2021 | schemaVersion = meta.SchemaVersion |
| 2022 | revision = meta.Revision |
| 2023 | contentDigest = meta.ContentDigest |
| 2024 | listingRevision = meta.ListingRevision |
| 2025 | listingContentDigest = meta.ListingContentDigest |
| 2026 | headMirror = meta |
| 2027 | } |
| 2028 | // Old recovery files may lack Recovered meta; filename still proves |
| 2029 | // automatic recovery lineage for catalog folding. |
| 2030 | if !recovered && LooksLikeRecoveryFilename(full) { |
| 2031 | recovered = true |
| 2032 | versionKind = VersionRecovery |
| 2033 | if parentID == "" { |
| 2034 | if parent, ok := RecoveryFilenameParentID(full); ok { |
| 2035 | parentID = parent |
| 2036 | } |
| 2037 | } |
| 2038 | } |
| 2039 | out = append(out, SessionOrderInfo{ |
| 2040 | Path: full, |
| 2041 | CreatedAt: createdAt, |
| 2042 | LastActivityAt: lastActivityAt, |
| 2043 | ModTime: lastActivityAt, |
| 2044 | Scope: scope, |
| 2045 | WorkspaceRoot: workspaceRoot, |
| 2046 | TopicID: topicID, |
| 2047 | TopicTitle: topicTitle, |
| 2048 | CustomTitle: customTitle, |
| 2049 | Recovered: recovered, |
| 2050 | RecoveryReason: recoveryReason, |
| 2051 | RecoveryDigest: recoveryDigest, |
| 2052 | ParentID: parentID, |
| 2053 | VersionKind: versionKind, |
| 2054 | VersionState: versionState, |
| 2055 | ParentConversationID: parentConversationID, |
| 2056 | ParentVersionID: parentVersionID, |
| 2057 | RecoveryPreferred: recoveryPreferred, |
| 2058 | Turns: turns, |
| 2059 | Preview: preview, |
| 2060 | SchemaVersion: schemaVersion, |
| 2061 | Revision: revision, |
| 2062 | ContentDigest: contentDigest, |
| 2063 | ListingRevision: listingRevision, |
| 2064 | ListingContentDigest: listingContentDigest, |
| 2065 | HeadID: headMirror.HeadID, |
| 2066 | HeadCount: headMirror.HeadCount, |
| 2067 | LogSchema: headMirror.LogSchema, |
| 2068 | }) |
| 2069 | } |
| 2070 | sort.Slice(out, func(i, j int) bool { |
| 2071 | if out[i].LastActivityAt.Equal(out[j].LastActivityAt) { |
| 2072 | return out[i].Path < out[j].Path |
| 2073 | } |
| 2074 | return out[i].LastActivityAt.After(out[j].LastActivityAt) |
| 2075 | }) |
| 2076 | return out, nil |
| 2077 | } |
| 2078 | |
| 2079 | // ListSessions returns every non-empty *.jsonl session under dir, |
| 2080 | // most-recently-active first, each with a preview line so the picker can show |
| 2081 | // something the user recognises. It never decodes a transcript: legacy counts |
| 2082 | // remain explicitly unknown until the session catalog's single repair worker |
| 2083 | // validates them. A missing directory is not an error. |
| 2084 | func ListSessions(dir string) ([]SessionInfo, error) { |
| 2085 | ordered, err := ListSessionOrder(dir) |
| 2086 | if err != nil { |
| 2087 | return nil, err |
| 2088 | } |
| 2089 | var out []SessionInfo |
| 2090 | for _, session := range ordered { |
| 2091 | preview, turns := session.Preview, session.Turns |
| 2092 | if !sessionListingProjectionFresh(session.SchemaVersion, turns, session.Revision, session.ListingRevision, session.ContentDigest, session.ListingContentDigest) { |
| 2093 | if !sessionArtifactsHaveContent(session.Path) { |
| 2094 | continue |
| 2095 | } |
| 2096 | if strings.TrimSpace(preview) == "" { |
| 2097 | preview = "History is being indexed — " + filepath.Base(session.Path) |
| 2098 | } |
| 2099 | out = append(out, sessionInfoFromOrder(session, preview, turns, false)) |
| 2100 | continue |
| 2101 | } |
| 2102 | if turns == 0 { |
| 2103 | // Never had user interaction — an empty conversation that should not |
| 2104 | // appear in the history panel or the resume picker. |
| 2105 | continue |
| 2106 | } |
| 2107 | out = append(out, sessionInfoFromOrder(session, preview, turns, true)) |
| 2108 | } |
| 2109 | return out, nil |
| 2110 | } |
| 2111 | |
| 2112 | // SessionPreview returns the same preview and user-turn count used by |
| 2113 | // ListSessions for one session file. |
| 2114 | func SessionPreview(path string) (string, int) { |
| 2115 | return previewSession(path) |
| 2116 | } |
| 2117 | |
| 2118 | // SessionPreviewWithError returns the same preview and user-turn count as |
| 2119 | // SessionPreview, but preserves read and decode failures for callers that must |
| 2120 | // not persist a fallback derived from an unreadable transcript. |
| 2121 | func SessionPreviewWithError(path string) (string, int, error) { |
| 2122 | return previewSessionWithError(path) |
| 2123 | } |
| 2124 | |
| 2125 | // SessionPreviewFromMessages computes the same preview line and user-turn count |
| 2126 | // as previewSession, but from an in-memory message slice. Session.Save writes |
| 2127 | // exactly these messages to the .jsonl, so this is byte-for-byte equivalent to |
| 2128 | // decoding the file — letting the autosave path persist the counts into the |
| 2129 | // sidecar without a disk read. |
| 2130 | func SessionPreviewFromMessages(msgs []provider.Message) (string, int) { |
| 2131 | first := "" |
| 2132 | turns := 0 |
| 2133 | for _, m := range msgs { |
| 2134 | if IsUserAuthoredTurnMessage(m) { |
| 2135 | turns++ |
| 2136 | if first == "" { |
| 2137 | first = truncatePreview(previewProse(UserMessageText(m))) |
| 2138 | } |
| 2139 | } |
| 2140 | } |
| 2141 | return first, turns |
| 2142 | } |
| 2143 | |
| 2144 | // previewSession returns the first user message (truncated) and the number of |
| 2145 | // user-role messages so the picker can show "5 turns · 'help me debug the…'". |
| 2146 | // Errors are swallowed — a malformed file just shows up with an empty preview. |
| 2147 | func previewSession(path string) (string, int) { |
| 2148 | preview, turns, _ := previewSessionWithError(path) |
| 2149 | return preview, turns |
| 2150 | } |
| 2151 | |
| 2152 | // previewProse drops the leading @file references a prompt opens with so the |
| 2153 | // preview shows what was asked rather than a row of paths. A prompt that is |
| 2154 | // nothing but references keeps them — there is nothing else to show. |
| 2155 | func previewProse(s string) string { |
| 2156 | rest := strings.TrimLeft(s, " \t") |
| 2157 | for strings.HasPrefix(rest, "@") { |
| 2158 | end := strings.IndexAny(rest, " \t\r\n") |
| 2159 | if end < 0 { |
| 2160 | return s |
| 2161 | } |
| 2162 | next := strings.TrimLeft(rest[end:], " \t") |
| 2163 | if strings.TrimSpace(next) == "" { |
| 2164 | return s |
| 2165 | } |
| 2166 | rest = next |
| 2167 | } |
| 2168 | if rest == "" { |
| 2169 | return s |
| 2170 | } |
| 2171 | return rest |
| 2172 | } |
| 2173 | |
| 2174 | // truncatePreview clamps a preview line to 80 runes with an ellipsis, matching |
| 2175 | // what the pickers render. |
| 2176 | func truncatePreview(s string) string { |
| 2177 | if r := []rune(s); len(r) > 80 { |
| 2178 | return string(r[:77]) + "…" |
| 2179 | } |
| 2180 | return s |
| 2181 | } |
| 2182 |