| 1 | package session |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "crypto/sha256" |
| 7 | "encoding/json" |
| 8 | "errors" |
| 9 | "fmt" |
| 10 | "io" |
| 11 | "os" |
| 12 | "path/filepath" |
| 13 | "sort" |
| 14 | "strings" |
| 15 | "time" |
| 16 | |
| 17 | "github.com/klauspost/compress/zstd" |
| 18 | bolt "go.etcd.io/bbolt" |
| 19 | |
| 20 | "reasonix/internal/agent" |
| 21 | "reasonix/internal/fileutil" |
| 22 | "reasonix/internal/provider" |
| 23 | "reasonix/internal/sessioncontent" |
| 24 | ) |
| 25 | |
| 26 | // Version 6 retains optional submission receipts in recovery checkpoints. |
| 27 | // Older projections are disposable and rebuild from the unchanged durable log. |
| 28 | const recoveryProjectionVersion = 6 |
| 29 | |
| 30 | const ( |
| 31 | recoveryFormatVersion = 1 |
| 32 | recoveryDBName = "recovery-v1.bolt" |
| 33 | storageIdentityName = "storage.identity.json" |
| 34 | RecentMessageLimit = 100 |
| 35 | recentInlineBytes = 32 << 10 |
| 36 | recentResponseBytes = 512 << 10 |
| 37 | recentSnapshotName = "recent-v1.json" |
| 38 | ) |
| 39 | |
| 40 | var ( |
| 41 | recoveryMetaBucket = []byte("meta") |
| 42 | recoveryCheckpointBucket = []byte("checkpoints") |
| 43 | recoveryOperationBucket = []byte("operations") |
| 44 | recoveryCurrentKey = []byte("current") |
| 45 | recoveryPreviousKey = []byte("previous") |
| 46 | ) |
| 47 | |
| 48 | // RecoveryOpenStats reports work performed by the normal open path. It is |
| 49 | // intentionally small and stable enough for capacity tests and host telemetry. |
| 50 | type RecoveryOpenStats struct { |
| 51 | UsedCheckpoint bool `json:"usedCheckpoint"` |
| 52 | LogBytesRead int64 `json:"logBytesRead"` |
| 53 | LogBytesTotal int64 `json:"logBytesTotal"` |
| 54 | TailCommits int `json:"tailCommits"` |
| 55 | } |
| 56 | |
| 57 | // RecentSnapshot is the bounded, read-only baseline used before history or |
| 58 | // search projections are available. |
| 59 | type RecentSnapshot struct { |
| 60 | Version int `json:"version"` |
| 61 | SessionID string `json:"sessionId"` |
| 62 | StorageGeneration string `json:"storageGeneration"` |
| 63 | DurableSequence uint64 `json:"durableSequence"` |
| 64 | TotalTurns int `json:"totalTurns"` |
| 65 | Entries []PersistentMessage `json:"entries"` |
| 66 | Title string `json:"title,omitempty"` |
| 67 | ModelRef string `json:"modelRef,omitempty"` |
| 68 | ModelIdentity string `json:"modelIdentity,omitempty"` |
| 69 | } |
| 70 | |
| 71 | type storageIdentity struct { |
| 72 | Version int `json:"version"` |
| 73 | SessionID string `json:"sessionId"` |
| 74 | Generation string `json:"generation"` |
| 75 | LogPrefixDigest string `json:"logPrefixDigest,omitempty"` |
| 76 | CreatedAt time.Time `json:"createdAt"` |
| 77 | } |
| 78 | |
| 79 | type recoveryCheckpoint struct { |
| 80 | Version int `json:"version"` |
| 81 | SessionID string `json:"sessionId"` |
| 82 | StorageGeneration string `json:"storageGeneration"` |
| 83 | StorageRevision int `json:"storageRevision"` |
| 84 | DurableSequence uint64 `json:"durableSequence"` |
| 85 | LogOffset int64 `json:"logOffset"` |
| 86 | AnchorOffset int64 `json:"anchorOffset,omitempty"` |
| 87 | AnchorFirst uint64 `json:"anchorFirst,omitempty"` |
| 88 | AnchorCommitID string `json:"anchorCommitId,omitempty"` |
| 89 | AnchorHash string `json:"anchorHash,omitempty"` |
| 90 | ProjectionVersion int `json:"projectionVersion"` |
| 91 | Projection Projection `json:"projection"` |
| 92 | RecentMessages []provider.Message `json:"recentMessages,omitempty"` |
| 93 | CatalogPreview string `json:"catalogPreview,omitempty"` |
| 94 | CreatedAt time.Time `json:"createdAt"` |
| 95 | } |
| 96 | |
| 97 | type recoveryOperation struct { |
| 98 | Hash string `json:"hash"` |
| 99 | CommitID string `json:"commitId"` |
| 100 | FirstSequence uint64 `json:"firstSequence"` |
| 101 | EventCount int `json:"eventCount"` |
| 102 | TurnID string `json:"turnId,omitempty"` |
| 103 | OperationID string `json:"operationId"` |
| 104 | OperationHash string `json:"operationHash"` |
| 105 | WriterGeneration uint64 `json:"writerGeneration"` |
| 106 | CreatedAt time.Time `json:"createdAt"` |
| 107 | } |
| 108 | |
| 109 | func operationForRecovery(record operationRecord) recoveryOperation { |
| 110 | commit := record.commit |
| 111 | return recoveryOperation{ |
| 112 | Hash: record.hash, CommitID: commit.ID, FirstSequence: commit.FirstSequence, |
| 113 | EventCount: commit.EventCount, TurnID: commit.TurnID, OperationID: commit.OperationID, |
| 114 | OperationHash: commit.OperationHash, WriterGeneration: commit.WriterGeneration, |
| 115 | CreatedAt: commit.CreatedAt, |
| 116 | } |
| 117 | } |
| 118 | |
| 119 | func (o recoveryOperation) record() operationRecord { |
| 120 | return operationRecord{hash: o.Hash, commit: Commit{ |
| 121 | SchemaVersion: SchemaVersion, Codec: Codec, RecordType: "commit", ID: o.CommitID, |
| 122 | OperationID: o.OperationID, OperationHash: o.OperationHash, |
| 123 | FirstSequence: o.FirstSequence, EventCount: o.EventCount, TurnID: o.TurnID, |
| 124 | WriterGeneration: o.WriterGeneration, CreatedAt: o.CreatedAt, |
| 125 | }} |
| 126 | } |
| 127 | |
| 128 | type recoveryStore struct { |
| 129 | db *bolt.DB |
| 130 | path string |
| 131 | recent string |
| 132 | sessionDir string |
| 133 | identity storageIdentity |
| 134 | } |
| 135 | |
| 136 | func recoveryCacheDir(sessionDir string) string { |
| 137 | return filepath.Join(filepath.Dir(sessionDir), ".recovery-cache", filepath.Base(sessionDir)) |
| 138 | } |
| 139 | |
| 140 | func ensureStorageIdentity(sessionDir string, manifest Manifest) (storageIdentity, error) { |
| 141 | path := filepath.Join(sessionDir, storageIdentityName) |
| 142 | prefix, prefixErr := storageLogPrefix(sessionDir, manifest) |
| 143 | if prefixErr != nil && !os.IsNotExist(prefixErr) { |
| 144 | return storageIdentity{}, prefixErr |
| 145 | } |
| 146 | if data, err := os.ReadFile(path); err == nil { |
| 147 | var identity storageIdentity |
| 148 | if json.Unmarshal(data, &identity) == nil && identity.Version == recoveryFormatVersion && identity.SessionID == manifest.SessionID && strings.TrimSpace(identity.Generation) != "" { |
| 149 | if identity.LogPrefixDigest == "" && prefix != "" { |
| 150 | identity.LogPrefixDigest = prefix |
| 151 | encoded, marshalErr := json.Marshal(identity) |
| 152 | if marshalErr != nil { |
| 153 | return storageIdentity{}, marshalErr |
| 154 | } |
| 155 | if writeErr := fileutil.AtomicWriteFileStrict(path, append(encoded, '\n'), 0o600); writeErr != nil { |
| 156 | return storageIdentity{}, writeErr |
| 157 | } |
| 158 | return identity, nil |
| 159 | } |
| 160 | if prefix == "" || identity.LogPrefixDigest == prefix { |
| 161 | return identity, nil |
| 162 | } |
| 163 | } |
| 164 | } |
| 165 | identity := storageIdentity{Version: recoveryFormatVersion, SessionID: manifest.SessionID, Generation: randomID(), LogPrefixDigest: prefix, CreatedAt: time.Now().UTC()} |
| 166 | data, err := json.Marshal(identity) |
| 167 | if err != nil { |
| 168 | return storageIdentity{}, err |
| 169 | } |
| 170 | if err := fileutil.AtomicWriteFileStrict(path, append(data, '\n'), 0o600); err != nil { |
| 171 | return storageIdentity{}, err |
| 172 | } |
| 173 | return identity, nil |
| 174 | } |
| 175 | |
| 176 | func storageLogPrefix(sessionDir string, manifest Manifest) (string, error) { |
| 177 | file, err := os.Open(logPathForManifest(sessionDir, manifest)) |
| 178 | if err != nil { |
| 179 | return "", err |
| 180 | } |
| 181 | defer file.Close() |
| 182 | // The framed transaction header makes the first 64 bytes immutable after |
| 183 | // the first commit. Hashing a larger short-file prefix would change merely |
| 184 | // because a normal append extended a log shorter than that prefix. |
| 185 | buffer := make([]byte, 64) |
| 186 | n, err := file.Read(buffer) |
| 187 | if err != nil && !errors.Is(err, io.EOF) { |
| 188 | return "", err |
| 189 | } |
| 190 | if n == 0 { |
| 191 | return "", nil |
| 192 | } |
| 193 | digest := sha256.Sum256(buffer[:n]) |
| 194 | return fmt.Sprintf("%x", digest[:]), nil |
| 195 | } |
| 196 | |
| 197 | func readStorageIdentity(sessionDir string, manifest Manifest) (storageIdentity, error) { |
| 198 | data, err := os.ReadFile(filepath.Join(sessionDir, storageIdentityName)) |
| 199 | if err != nil { |
| 200 | return storageIdentity{}, err |
| 201 | } |
| 202 | var identity storageIdentity |
| 203 | if err := json.Unmarshal(data, &identity); err != nil { |
| 204 | return storageIdentity{}, err |
| 205 | } |
| 206 | if identity.Version != recoveryFormatVersion || identity.SessionID != manifest.SessionID || strings.TrimSpace(identity.Generation) == "" { |
| 207 | return storageIdentity{}, ErrStaleGeneration |
| 208 | } |
| 209 | prefix, err := storageLogPrefix(sessionDir, manifest) |
| 210 | if err != nil && !os.IsNotExist(err) { |
| 211 | return storageIdentity{}, err |
| 212 | } |
| 213 | if identity.LogPrefixDigest != "" && prefix != identity.LogPrefixDigest { |
| 214 | return storageIdentity{}, ErrStaleGeneration |
| 215 | } |
| 216 | return identity, nil |
| 217 | } |
| 218 | |
| 219 | func openRecoveryStore(sessionDir string, identity storageIdentity) (*recoveryStore, error) { |
| 220 | dir := recoveryCacheDir(sessionDir) |
| 221 | if err := os.MkdirAll(dir, 0o700); err != nil { |
| 222 | return nil, err |
| 223 | } |
| 224 | path := filepath.Join(dir, recoveryDBName) |
| 225 | db, err := bolt.Open(path, 0o600, &bolt.Options{Timeout: 250 * time.Millisecond, NoFreelistSync: true}) |
| 226 | if err != nil { |
| 227 | return nil, err |
| 228 | } |
| 229 | store := &recoveryStore{db: db, path: path, recent: filepath.Join(dir, recentSnapshotName), sessionDir: sessionDir, identity: identity} |
| 230 | err = db.Update(func(tx *bolt.Tx) error { |
| 231 | meta, err := tx.CreateBucketIfNotExists(recoveryMetaBucket) |
| 232 | if err != nil { |
| 233 | return err |
| 234 | } |
| 235 | checkpoints, err := tx.CreateBucketIfNotExists(recoveryCheckpointBucket) |
| 236 | if err != nil { |
| 237 | return err |
| 238 | } |
| 239 | operations, err := tx.CreateBucketIfNotExists(recoveryOperationBucket) |
| 240 | if err != nil { |
| 241 | return err |
| 242 | } |
| 243 | _ = checkpoints |
| 244 | _ = operations |
| 245 | stored := string(meta.Get([]byte("storage_generation"))) |
| 246 | if stored != "" && stored != identity.Generation { |
| 247 | if err := tx.DeleteBucket(recoveryCheckpointBucket); err != nil { |
| 248 | return err |
| 249 | } |
| 250 | if err := tx.DeleteBucket(recoveryOperationBucket); err != nil { |
| 251 | return err |
| 252 | } |
| 253 | if _, err := tx.CreateBucket(recoveryCheckpointBucket); err != nil { |
| 254 | return err |
| 255 | } |
| 256 | if _, err := tx.CreateBucket(recoveryOperationBucket); err != nil { |
| 257 | return err |
| 258 | } |
| 259 | } |
| 260 | return meta.Put([]byte("storage_generation"), []byte(identity.Generation)) |
| 261 | }) |
| 262 | if err != nil { |
| 263 | _ = db.Close() |
| 264 | return nil, err |
| 265 | } |
| 266 | return store, nil |
| 267 | } |
| 268 | |
| 269 | func openRecoveryStoreRepair(sessionDir string, identity storageIdentity) (*recoveryStore, error) { |
| 270 | store, err := openRecoveryStore(sessionDir, identity) |
| 271 | if err == nil { |
| 272 | return store, nil |
| 273 | } |
| 274 | path := filepath.Join(recoveryCacheDir(sessionDir), recoveryDBName) |
| 275 | if _, statErr := os.Stat(path); statErr == nil { |
| 276 | _ = os.Rename(path, path+fmt.Sprintf(".corrupt-%d", time.Now().UTC().UnixNano())) |
| 277 | } |
| 278 | return openRecoveryStore(sessionDir, identity) |
| 279 | } |
| 280 | |
| 281 | func (s *recoveryStore) close() error { |
| 282 | if s == nil || s.db == nil { |
| 283 | return nil |
| 284 | } |
| 285 | return s.db.Close() |
| 286 | } |
| 287 | |
| 288 | func encodeRecoveryValue(value any) ([]byte, error) { |
| 289 | raw, err := json.Marshal(value) |
| 290 | if err != nil { |
| 291 | return nil, err |
| 292 | } |
| 293 | encoder, err := zstd.NewWriter(nil, zstd.WithEncoderConcurrency(1), zstd.WithEncoderLevel(zstd.SpeedFastest)) |
| 294 | if err != nil { |
| 295 | return nil, err |
| 296 | } |
| 297 | defer encoder.Close() |
| 298 | return encoder.EncodeAll(raw, nil), nil |
| 299 | } |
| 300 | |
| 301 | func decodeRecoveryValue(data []byte, value any) error { |
| 302 | decoder, err := zstd.NewReader(nil, zstd.WithDecoderConcurrency(1), zstd.WithDecoderMaxMemory(64<<20)) |
| 303 | if err != nil { |
| 304 | return err |
| 305 | } |
| 306 | defer decoder.Close() |
| 307 | raw, err := decoder.DecodeAll(data, nil) |
| 308 | if err != nil { |
| 309 | return err |
| 310 | } |
| 311 | return json.Unmarshal(raw, value) |
| 312 | } |
| 313 | |
| 314 | func (s *recoveryStore) loadCheckpoints() ([]recoveryCheckpoint, error) { |
| 315 | if s == nil || s.db == nil { |
| 316 | return nil, os.ErrNotExist |
| 317 | } |
| 318 | var encoded [][]byte |
| 319 | err := s.db.View(func(tx *bolt.Tx) error { |
| 320 | bucket := tx.Bucket(recoveryCheckpointBucket) |
| 321 | if bucket == nil { |
| 322 | return os.ErrNotExist |
| 323 | } |
| 324 | for _, key := range [][]byte{recoveryCurrentKey, recoveryPreviousKey} { |
| 325 | if value := bucket.Get(key); value != nil { |
| 326 | encoded = append(encoded, append([]byte(nil), value...)) |
| 327 | } |
| 328 | } |
| 329 | if len(encoded) == 0 { |
| 330 | return os.ErrNotExist |
| 331 | } |
| 332 | return nil |
| 333 | }) |
| 334 | if err != nil { |
| 335 | return nil, err |
| 336 | } |
| 337 | checkpoints := make([]recoveryCheckpoint, 0, len(encoded)) |
| 338 | for _, value := range encoded { |
| 339 | var checkpoint recoveryCheckpoint |
| 340 | if decodeRecoveryValue(value, &checkpoint) == nil { |
| 341 | checkpoints = append(checkpoints, checkpoint) |
| 342 | } |
| 343 | } |
| 344 | if len(checkpoints) == 0 { |
| 345 | return nil, ErrDamagedStore |
| 346 | } |
| 347 | return checkpoints, nil |
| 348 | } |
| 349 | |
| 350 | func (s *recoveryStore) lookupOperation(operationID string) (operationRecord, bool, error) { |
| 351 | if s == nil || s.db == nil { |
| 352 | return operationRecord{}, false, nil |
| 353 | } |
| 354 | var encoded []byte |
| 355 | err := s.db.View(func(tx *bolt.Tx) error { |
| 356 | bucket := tx.Bucket(recoveryOperationBucket) |
| 357 | if bucket == nil { |
| 358 | return nil |
| 359 | } |
| 360 | encoded = append(encoded, bucket.Get([]byte(operationID))...) |
| 361 | return nil |
| 362 | }) |
| 363 | if err != nil || len(encoded) == 0 { |
| 364 | return operationRecord{}, false, err |
| 365 | } |
| 366 | var operation recoveryOperation |
| 367 | if err := json.Unmarshal(encoded, &operation); err != nil { |
| 368 | return operationRecord{}, false, err |
| 369 | } |
| 370 | return operation.record(), true, nil |
| 371 | } |
| 372 | |
| 373 | func (s *recoveryStore) publish(ctx context.Context, checkpoint recoveryCheckpoint, operations map[string]operationRecord) error { |
| 374 | if s == nil || s.db == nil { |
| 375 | return errors.New("session: recovery store unavailable") |
| 376 | } |
| 377 | if err := ctx.Err(); err != nil { |
| 378 | return err |
| 379 | } |
| 380 | if s.identity.LogPrefixDigest == "" { |
| 381 | manifest, err := readStoredManifest(filepath.Join(s.sessionDir, "manifest.json")) |
| 382 | if err != nil { |
| 383 | return err |
| 384 | } |
| 385 | prefix, err := storageLogPrefix(s.sessionDir, manifest) |
| 386 | if err != nil { |
| 387 | return err |
| 388 | } |
| 389 | if prefix != "" { |
| 390 | s.identity.LogPrefixDigest = prefix |
| 391 | encodedIdentity, err := json.Marshal(s.identity) |
| 392 | if err != nil { |
| 393 | return err |
| 394 | } |
| 395 | if err := fileutil.AtomicWriteFileStrict(filepath.Join(s.sessionDir, storageIdentityName), append(encodedIdentity, '\n'), 0o600); err != nil { |
| 396 | return err |
| 397 | } |
| 398 | } |
| 399 | } |
| 400 | checkpoint.Version = recoveryFormatVersion |
| 401 | checkpoint.StorageGeneration = s.identity.Generation |
| 402 | checkpoint.CreatedAt = time.Now().UTC() |
| 403 | encoded, err := encodeRecoveryValue(checkpoint) |
| 404 | if err != nil { |
| 405 | return err |
| 406 | } |
| 407 | keys := make([]string, 0, len(operations)) |
| 408 | for key := range operations { |
| 409 | keys = append(keys, key) |
| 410 | } |
| 411 | sort.Strings(keys) |
| 412 | err = s.db.Update(func(tx *bolt.Tx) error { |
| 413 | checkpoints := tx.Bucket(recoveryCheckpointBucket) |
| 414 | operationBucket := tx.Bucket(recoveryOperationBucket) |
| 415 | if checkpoints == nil || operationBucket == nil { |
| 416 | return errors.New("session: recovery buckets unavailable") |
| 417 | } |
| 418 | if current := checkpoints.Get(recoveryCurrentKey); current != nil { |
| 419 | if err := checkpoints.Put(recoveryPreviousKey, current); err != nil { |
| 420 | return err |
| 421 | } |
| 422 | } |
| 423 | for _, key := range keys { |
| 424 | operation := operationForRecovery(operations[key]) |
| 425 | value, err := json.Marshal(operation) |
| 426 | if err != nil { |
| 427 | return err |
| 428 | } |
| 429 | if err := operationBucket.Put([]byte(key), value); err != nil { |
| 430 | return err |
| 431 | } |
| 432 | } |
| 433 | if err := checkpoints.Put(recoveryCurrentKey, encoded); err != nil { |
| 434 | return err |
| 435 | } |
| 436 | return tx.Bucket(recoveryMetaBucket).Put([]byte("coverage_sequence"), fmt.Append(nil, checkpoint.DurableSequence)) |
| 437 | }) |
| 438 | if err != nil { |
| 439 | return err |
| 440 | } |
| 441 | recent := RecentSnapshot{ |
| 442 | Version: recoveryFormatVersion, SessionID: checkpoint.SessionID, |
| 443 | StorageGeneration: checkpoint.StorageGeneration, DurableSequence: checkpoint.DurableSequence, |
| 444 | Title: checkpoint.Projection.Title, |
| 445 | ModelRef: checkpoint.Projection.ModelRef, ModelIdentity: checkpoint.Projection.ModelIdentity, |
| 446 | TotalTurns: visibleBoundaryCount(checkpoint.Projection, false), |
| 447 | } |
| 448 | recent.Entries, err = buildRecentEntries(ctx, s.sessionDir, checkpoint.RecentMessages, checkpoint.DurableSequence, recent.TotalTurns) |
| 449 | attachSubmissionEntries(checkpoint.Projection.Submissions, checkpoint.SessionID, recent.Entries) |
| 450 | if err != nil { |
| 451 | return err |
| 452 | } |
| 453 | data, err := json.Marshal(recent) |
| 454 | if err != nil { |
| 455 | return err |
| 456 | } |
| 457 | return fileutil.AtomicWriteFileStrict(s.recent, append(data, '\n'), 0o600) |
| 458 | } |
| 459 | |
| 460 | func readRecentSnapshot(sessionDir string, identity storageIdentity) (RecentSnapshot, error) { |
| 461 | data, err := os.ReadFile(filepath.Join(recoveryCacheDir(sessionDir), recentSnapshotName)) |
| 462 | if err != nil { |
| 463 | return RecentSnapshot{}, err |
| 464 | } |
| 465 | var snapshot RecentSnapshot |
| 466 | if err := json.Unmarshal(data, &snapshot); err != nil { |
| 467 | return RecentSnapshot{}, err |
| 468 | } |
| 469 | if snapshot.Version != recoveryFormatVersion || snapshot.SessionID != identity.SessionID || snapshot.StorageGeneration != identity.Generation { |
| 470 | return RecentSnapshot{}, ErrStaleGeneration |
| 471 | } |
| 472 | if len(snapshot.Entries) > RecentMessageLimit { |
| 473 | return RecentSnapshot{}, ErrDamagedStore |
| 474 | } |
| 475 | return snapshot, nil |
| 476 | } |
| 477 | |
| 478 | // buildRecentEntries produces the bounded public baseline. Large canonical |
| 479 | // messages are stored once in ContentStore and represented by a preview plus a |
| 480 | // range-readable reference, so recent-v1.json cannot grow with tool output. |
| 481 | func buildRecentEntries(ctx context.Context, sessionDir string, messages []provider.Message, sequence uint64, totalTurns int) ([]PersistentMessage, error) { |
| 482 | if len(messages) > RecentMessageLimit { |
| 483 | messages = messages[len(messages)-RecentMessageLimit:] |
| 484 | } |
| 485 | entries := make([]PersistentMessage, 0, len(messages)) |
| 486 | content := contentStoreForSessionDir(sessionDir) |
| 487 | inlineBytes := 0 |
| 488 | visibleTurn := totalTurns |
| 489 | for _, message := range messages { |
| 490 | if agent.IsUserAuthoredTurnMessage(message) { |
| 491 | visibleTurn-- |
| 492 | } |
| 493 | } |
| 494 | visibleTurn = max(visibleTurn, 0) |
| 495 | for position, message := range messages { |
| 496 | if err := ctx.Err(); err != nil { |
| 497 | return nil, err |
| 498 | } |
| 499 | if agent.IsUserAuthoredTurnMessage(message) { |
| 500 | visibleTurn++ |
| 501 | } |
| 502 | body, err := json.Marshal(message) |
| 503 | if err != nil { |
| 504 | return nil, err |
| 505 | } |
| 506 | entry := PersistentMessage{ |
| 507 | MessageID: message.ID, Position: int64(position + 1), Version: 1, |
| 508 | Role: string(message.Role), Preview: messagePreview(message), |
| 509 | EventSequence: sequence, VisibleTurn: visibleTurn, |
| 510 | } |
| 511 | if len(body) <= recentInlineBytes && inlineBytes+len(body) <= recentResponseBytes { |
| 512 | entry.Inline = body |
| 513 | inlineBytes += len(body) |
| 514 | } else { |
| 515 | ref, err := content.Put(ctx, bytes.NewReader(body), sessioncontent.Metadata{MediaType: "application/json"}) |
| 516 | if err != nil { |
| 517 | return nil, err |
| 518 | } |
| 519 | entry.ContentRef = &ref |
| 520 | previewBody, err := recentDisplayMessage(message) |
| 521 | if err != nil { |
| 522 | return nil, err |
| 523 | } |
| 524 | if inlineBytes+len(previewBody) <= recentResponseBytes { |
| 525 | entry.Inline = previewBody |
| 526 | inlineBytes += len(previewBody) |
| 527 | } |
| 528 | } |
| 529 | entries = append(entries, entry) |
| 530 | } |
| 531 | return entries, nil |
| 532 | } |
| 533 | |
| 534 | func recentDisplayMessage(message provider.Message) (json.RawMessage, error) { |
| 535 | preview := detachMessages([]provider.Message{message})[0] |
| 536 | preview.Content = messagePreview(message) |
| 537 | preview.RawContent = "" |
| 538 | preview.ProviderContent = "" |
| 539 | preview.Images = nil |
| 540 | preview.ImageInputs = nil |
| 541 | preview.ResponsesItems = nil |
| 542 | preview.ThinkingBlocks = nil |
| 543 | if runes := []rune(preview.ReasoningContent); len(runes) > 4096 { |
| 544 | preview.ReasoningContent = string(runes[:4096]) |
| 545 | } |
| 546 | for i := range preview.ToolCalls { |
| 547 | if runes := []rune(preview.ToolCalls[i].Arguments); len(runes) > 2048 { |
| 548 | preview.ToolCalls[i].Arguments = string(runes[:2048]) |
| 549 | } |
| 550 | } |
| 551 | body, err := json.Marshal(preview) |
| 552 | if err != nil { |
| 553 | return nil, err |
| 554 | } |
| 555 | if len(body) <= recentInlineBytes { |
| 556 | return body, nil |
| 557 | } |
| 558 | // Preserve the fields required to place the row even when optional display |
| 559 | // metadata alone exceeds the per-message preview budget. |
| 560 | return json.Marshal(provider.Message{ |
| 561 | ID: preview.ID, Role: preview.Role, Content: preview.Content, |
| 562 | ToolCallID: preview.ToolCallID, Name: preview.Name, |
| 563 | CreatedAt: preview.CreatedAt, WorkDurationMs: preview.WorkDurationMs, |
| 564 | }) |
| 565 | } |
| 566 | |
| 567 | func checkpointFromStartup(manifest Manifest, identity storageIdentity, state *startupSessionState) recoveryCheckpoint { |
| 568 | projection, _ := Project(nil) |
| 569 | if state != nil { |
| 570 | projection = cloneProjection(state.projection) |
| 571 | projection.Messages = nil |
| 572 | } |
| 573 | checkpoint := recoveryCheckpoint{ |
| 574 | Version: recoveryFormatVersion, SessionID: manifest.SessionID, |
| 575 | StorageGeneration: identity.Generation, StorageRevision: StorageRevision, |
| 576 | ProjectionVersion: recoveryProjectionVersion, Projection: projection, |
| 577 | } |
| 578 | if state != nil { |
| 579 | checkpoint.DurableSequence = state.durable |
| 580 | checkpoint.LogOffset = state.tip.LogOffset |
| 581 | checkpoint.AnchorOffset = state.tip.AnchorOffset |
| 582 | checkpoint.AnchorFirst = state.tip.AnchorFirst |
| 583 | checkpoint.AnchorCommitID = state.tip.AnchorCommitID |
| 584 | checkpoint.AnchorHash = state.tip.AnchorHash |
| 585 | checkpoint.RecentMessages = detachMessages(state.recentMessages) |
| 586 | checkpoint.CatalogPreview = state.catalogPreview |
| 587 | } |
| 588 | return checkpoint |
| 589 | } |
| 590 | |
| 591 | func loadRecoveryStartupState(ctx context.Context, dir string, file *os.File, info os.FileInfo, recovery *recoveryStore, identity storageIdentity) (*startupSessionState, int64, bool, RecoveryOpenStats, bool) { |
| 592 | stats := RecoveryOpenStats{LogBytesTotal: info.Size()} |
| 593 | checkpoints, err := recovery.loadCheckpoints() |
| 594 | if err != nil { |
| 595 | return nil, 0, false, stats, false |
| 596 | } |
| 597 | for _, checkpoint := range checkpoints { |
| 598 | if state, end, torn, attempt, ok := tryRecoveryCheckpoint(ctx, dir, file, info, identity, checkpoint); ok { |
| 599 | return state, end, torn, attempt, true |
| 600 | } |
| 601 | } |
| 602 | return nil, 0, false, stats, false |
| 603 | } |
| 604 | |
| 605 | func tryRecoveryCheckpoint(ctx context.Context, dir string, file *os.File, info os.FileInfo, identity storageIdentity, checkpoint recoveryCheckpoint) (*startupSessionState, int64, bool, RecoveryOpenStats, bool) { |
| 606 | stats := RecoveryOpenStats{LogBytesTotal: info.Size()} |
| 607 | if checkpoint.Version != recoveryFormatVersion || checkpoint.ProjectionVersion != recoveryProjectionVersion || |
| 608 | checkpoint.SessionID != identity.SessionID || checkpoint.StorageGeneration != identity.Generation || |
| 609 | checkpoint.StorageRevision != StorageRevision || checkpoint.LogOffset < 0 || checkpoint.LogOffset > info.Size() || |
| 610 | checkpoint.Projection.CommittedSequence != checkpoint.DurableSequence { |
| 611 | return nil, 0, false, stats, false |
| 612 | } |
| 613 | if checkpoint.DurableSequence == 0 { |
| 614 | if checkpoint.LogOffset != 0 { |
| 615 | return nil, 0, false, stats, false |
| 616 | } |
| 617 | } else { |
| 618 | if checkpoint.AnchorOffset < 0 || checkpoint.AnchorOffset >= checkpoint.LogOffset || checkpoint.AnchorFirst == 0 || checkpoint.AnchorCommitID == "" { |
| 619 | return nil, 0, false, stats, false |
| 620 | } |
| 621 | var anchor Commit |
| 622 | var anchorEnd int64 |
| 623 | err := scanV4CommitFileRefs(ctx, file, checkpoint.AnchorOffset, checkpoint.AnchorFirst, contentStoreForSessionDir(dir), nil, func(_ int64, commit Commit) bool { |
| 624 | anchor = commit |
| 625 | anchorEnd, _ = file.Seek(0, 1) |
| 626 | return false |
| 627 | }) |
| 628 | if err != nil || anchor.ID != checkpoint.AnchorCommitID || anchor.OperationHash != checkpoint.AnchorHash || |
| 629 | anchor.LastSequence() != checkpoint.DurableSequence || anchorEnd != checkpoint.LogOffset { |
| 630 | return nil, 0, false, stats, false |
| 631 | } |
| 632 | stats.LogBytesRead += anchorEnd - checkpoint.AnchorOffset |
| 633 | } |
| 634 | |
| 635 | state := &startupSessionState{ |
| 636 | projection: cloneProjection(checkpoint.Projection), operations: map[string]operationRecord{}, |
| 637 | durable: checkpoint.DurableSequence, catalogPreview: checkpoint.CatalogPreview, |
| 638 | recentMessages: detachMessages(checkpoint.RecentMessages), |
| 639 | tip: durableTip{LogOffset: checkpoint.LogOffset, AnchorOffset: checkpoint.AnchorOffset, |
| 640 | AnchorFirst: checkpoint.AnchorFirst, AnchorCommitID: checkpoint.AnchorCommitID, AnchorHash: checkpoint.AnchorHash}, |
| 641 | } |
| 642 | var projectionErr error |
| 643 | content := contentStoreForSessionDir(dir) |
| 644 | err := scanV4CommitFile(ctx, file, checkpoint.LogOffset, checkpoint.DurableSequence+1, content, nil, func(offset int64, commit Commit) bool { |
| 645 | if err := applyProjectionCommit(&state.projection, commit); err != nil { |
| 646 | projectionErr = err |
| 647 | return false |
| 648 | } |
| 649 | if err := applyRecentCommit(&state.recentMessages, commit); err != nil { |
| 650 | projectionErr = err |
| 651 | return false |
| 652 | } |
| 653 | state.operations[commit.OperationID] = compactOperationRecord(commit) |
| 654 | state.durable = commit.LastSequence() |
| 655 | state.tip.AnchorOffset = offset |
| 656 | state.tip.AnchorFirst = commit.FirstSequence |
| 657 | state.tip.AnchorCommitID = commit.ID |
| 658 | state.tip.AnchorHash = commit.OperationHash |
| 659 | state.tip.LogOffset, _ = file.Seek(0, 1) |
| 660 | stats.TailCommits++ |
| 661 | return true |
| 662 | }) |
| 663 | if err != nil || projectionErr != nil { |
| 664 | return nil, 0, false, stats, false |
| 665 | } |
| 666 | state.projection.Messages = nil |
| 667 | state.projection.CommittedSequence = state.durable |
| 668 | stats.UsedCheckpoint = true |
| 669 | stats.LogBytesRead += max(state.tip.LogOffset-checkpoint.LogOffset, 0) |
| 670 | return state, state.tip.LogOffset, state.tip.LogOffset < info.Size(), stats, true |
| 671 | } |
| 672 | |
| 673 | func applyRecentCommit(messages *[]provider.Message, commit Commit) error { |
| 674 | projection, _ := Project(nil) |
| 675 | projection.Messages = detachMessages(*messages) |
| 676 | recent := commit |
| 677 | recent.Events = nil |
| 678 | for _, event := range commit.Events { |
| 679 | switch event.Kind { |
| 680 | case "message/complete", "message/upsert", "message/retract", "history/replace", "legacy/import": |
| 681 | recent.Events = append(recent.Events, event) |
| 682 | } |
| 683 | } |
| 684 | if len(recent.Events) == 0 { |
| 685 | return nil |
| 686 | } |
| 687 | if err := applyProjectionCommit(&projection, recent); err != nil { |
| 688 | return err |
| 689 | } |
| 690 | if len(projection.Messages) > RecentMessageLimit { |
| 691 | projection.Messages = append([]provider.Message(nil), projection.Messages[len(projection.Messages)-RecentMessageLimit:]...) |
| 692 | } |
| 693 | *messages = detachMessages(projection.Messages) |
| 694 | return nil |
| 695 | } |
| 696 |