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