| 1 | package session |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "database/sql" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "path/filepath" |
| 10 | "slices" |
| 11 | "strings" |
| 12 | |
| 13 | "reasonix/internal/projectiondb" |
| 14 | "reasonix/internal/provider" |
| 15 | "reasonix/internal/sessioncontent" |
| 16 | ) |
| 17 | |
| 18 | const ( |
| 19 | historyIndexVersion = 10 |
| 20 | HistoryPageDefaultLimit = 100 |
| 21 | HistoryPageMaxLimit = 500 |
| 22 | HistoryPageMaxBytes = 2 << 20 |
| 23 | historyIndexTxnEvents = 512 |
| 24 | historyIndexTxnBytes = 8 << 20 |
| 25 | ) |
| 26 | |
| 27 | // PersistentMessage is the storage/query representation of a message. It is |
| 28 | // deliberately separate from provider.Message: provider DTOs are materialized |
| 29 | // only at model or compatibility boundaries. |
| 30 | type PersistentMessage struct { |
| 31 | ToolObservations map[string]ToolObservation `json:"toolObservations,omitempty"` |
| 32 | SubmissionID string `json:"submissionId,omitempty"` |
| 33 | SamplingCount *int `json:"samplingCount,omitempty"` |
| 34 | ToolCount *int `json:"toolCount,omitempty"` |
| 35 | TurnFinal bool `json:"turnFinal,omitempty"` |
| 36 | TurnDurationMs int64 `json:"turnDurationMs,omitempty"` |
| 37 | MessageID string `json:"messageId"` |
| 38 | Position int64 `json:"position"` |
| 39 | Version int `json:"version"` |
| 40 | Role string `json:"role"` |
| 41 | Preview string `json:"preview,omitempty"` |
| 42 | EventSequence uint64 `json:"eventSequence"` |
| 43 | VisibleTurn int `json:"visibleTurn"` |
| 44 | Inline json.RawMessage `json:"inline,omitempty"` |
| 45 | ContentRef *sessioncontent.Ref `json:"contentRef,omitempty"` |
| 46 | } |
| 47 | |
| 48 | type MessageHistoryPage struct { |
| 49 | Messages []PersistentMessage `json:"messages"` |
| 50 | SnapshotSequence uint64 `json:"snapshotSequence"` |
| 51 | CoverageSequence uint64 `json:"coverageSequence"` |
| 52 | Status string `json:"status"` |
| 53 | TotalTurns int `json:"totalTurns"` |
| 54 | Generation string `json:"generation"` |
| 55 | NextCursor string `json:"nextCursor,omitempty"` |
| 56 | HasMore bool `json:"hasMore"` |
| 57 | } |
| 58 | |
| 59 | // MessageLocation converts a stable message identity into a cursor for a |
| 60 | // fixed locator snapshot. It contains no message body. |
| 61 | type MessageLocation struct { |
| 62 | Status string `json:"status"` |
| 63 | MessageID string `json:"messageId,omitempty"` |
| 64 | SnapshotSequence uint64 `json:"snapshotSequence"` |
| 65 | CoverageSequence uint64 `json:"coverageSequence"` |
| 66 | Generation string `json:"generation,omitempty"` |
| 67 | Position int64 `json:"position,omitempty"` |
| 68 | VisibleTurn int `json:"visibleTurn,omitempty"` |
| 69 | Cursor string `json:"cursor,omitempty"` |
| 70 | } |
| 71 | |
| 72 | // HistoryPosition is the bounded display metadata for one message in a fixed |
| 73 | // durable snapshot. It deliberately excludes message bodies so callers can |
| 74 | // plan a window without pulling the transcript into memory. |
| 75 | type HistoryPosition struct { |
| 76 | Position int64 `json:"position"` |
| 77 | VisibleTurn int `json:"visibleTurn"` |
| 78 | Role provider.Role `json:"role"` |
| 79 | } |
| 80 | |
| 81 | // HistoryShape describes the complete ordering of a fixed durable snapshot |
| 82 | // using only small per-message metadata. Message bodies are fetched later via |
| 83 | // HistoryWindow. |
| 84 | type HistoryShape struct { |
| 85 | SnapshotSequence uint64 `json:"snapshotSequence"` |
| 86 | Positions []HistoryPosition `json:"positions"` |
| 87 | TotalTurns int `json:"totalTurns"` |
| 88 | } |
| 89 | |
| 90 | type historyCursor struct { |
| 91 | SessionID string `json:"sessionId"` |
| 92 | StorageRevision int `json:"storageRevision"` |
| 93 | SnapshotSequence uint64 `json:"snapshotSequence"` |
| 94 | BeforePosition int64 `json:"beforePosition"` |
| 95 | Projection int `json:"projection"` |
| 96 | Generation string `json:"generation"` |
| 97 | } |
| 98 | |
| 99 | type historyBuildState struct { |
| 100 | commitTurn string |
| 101 | commitTime int64 |
| 102 | nextPosition int64 |
| 103 | visibleTurn int |
| 104 | positions map[string]int64 |
| 105 | turns map[string]int |
| 106 | versions map[string]int |
| 107 | tx *sql.Tx |
| 108 | statements *historyBuildStatements |
| 109 | transactions [][]any |
| 110 | events [][]any |
| 111 | contentRefs [][]any |
| 112 | messages [][]any |
| 113 | } |
| 114 | |
| 115 | type historyPreparation struct { |
| 116 | done chan struct{} |
| 117 | err error |
| 118 | } |
| 119 | |
| 120 | type historyBuildStatements struct { |
| 121 | clear *sql.Stmt |
| 122 | expire *sql.Stmt |
| 123 | } |
| 124 | |
| 125 | func prepareHistoryBuildStatements(ctx context.Context, tx *sql.Tx) (*historyBuildStatements, error) { |
| 126 | statements := &historyBuildStatements{} |
| 127 | queries := []struct { |
| 128 | target **sql.Stmt |
| 129 | query string |
| 130 | }{ |
| 131 | {&statements.clear, `UPDATE messages SET current=0,valid_to=? WHERE current=1`}, |
| 132 | {&statements.expire, `UPDATE messages SET current=0,valid_to=? WHERE message_id=? AND current=1`}, |
| 133 | } |
| 134 | for _, candidate := range queries { |
| 135 | prepared, err := tx.PrepareContext(ctx, candidate.query) |
| 136 | if err != nil { |
| 137 | statements.close() |
| 138 | return nil, err |
| 139 | } |
| 140 | *candidate.target = prepared |
| 141 | } |
| 142 | return statements, nil |
| 143 | } |
| 144 | |
| 145 | func (s *historyBuildStatements) close() { |
| 146 | if s == nil { |
| 147 | return |
| 148 | } |
| 149 | for _, statement := range []*sql.Stmt{s.clear, s.expire} { |
| 150 | if statement != nil { |
| 151 | _ = statement.Close() |
| 152 | } |
| 153 | } |
| 154 | } |
| 155 | |
| 156 | func historyIndexPath(root, sessionID string) string { |
| 157 | return filepath.Join(root, ".query-cache", filepath.Base(sessionID), "history-locator-v2.sqlite") |
| 158 | } |
| 159 | |
| 160 | var historyMigrations = []projectiondb.Migration{{Version: 1, Apply: func(ctx context.Context, tx *sql.Tx) error { |
| 161 | for _, statement := range []string{ |
| 162 | `CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)`, |
| 163 | `CREATE TABLE transactions (commit_id TEXT PRIMARY KEY, first_sequence INTEGER NOT NULL, last_sequence INTEGER NOT NULL, operation_id TEXT NOT NULL UNIQUE, operation_hash TEXT NOT NULL, turn_id TEXT NOT NULL, created_at TEXT NOT NULL)`, |
| 164 | `CREATE TABLE events (sequence INTEGER PRIMARY KEY, commit_id TEXT NOT NULL, event_id TEXT NOT NULL UNIQUE, kind TEXT NOT NULL, payload_digest TEXT NOT NULL DEFAULT '', payload_bytes INTEGER NOT NULL DEFAULT 0)`, |
| 165 | `CREATE TABLE messages (message_id TEXT NOT NULL, version INTEGER NOT NULL, position INTEGER NOT NULL, event_sequence INTEGER NOT NULL, valid_to INTEGER NOT NULL DEFAULT 0, role TEXT NOT NULL, preview TEXT NOT NULL, inline BLOB, content_digest TEXT NOT NULL DEFAULT '', content_bytes INTEGER NOT NULL DEFAULT 0, content_index_digest TEXT NOT NULL DEFAULT '', current INTEGER NOT NULL, PRIMARY KEY(message_id, version))`, |
| 166 | `CREATE UNIQUE INDEX messages_current_position ON messages(position) WHERE current=1`, |
| 167 | `CREATE INDEX messages_current_id ON messages(message_id) WHERE current=1`, |
| 168 | `CREATE TABLE content_refs (digest TEXT NOT NULL, bytes INTEGER NOT NULL, index_digest TEXT NOT NULL DEFAULT '', PRIMARY KEY(digest, bytes, index_digest))`, |
| 169 | } { |
| 170 | if _, err := tx.ExecContext(ctx, statement); err != nil { |
| 171 | return err |
| 172 | } |
| 173 | } |
| 174 | return nil |
| 175 | }}, {Version: 2, Apply: func(ctx context.Context, tx *sql.Tx) error { |
| 176 | _, err := tx.ExecContext(ctx, `ALTER TABLE messages ADD COLUMN search_text TEXT NOT NULL DEFAULT ''`) |
| 177 | return err |
| 178 | }}, {Version: 3, Apply: func(ctx context.Context, tx *sql.Tx) error { |
| 179 | // Fixed-snapshot pages walk positions newest-to-oldest. Without this index, |
| 180 | // SQLite scans and sorts the full message-body table for every page; on a |
| 181 | // GiB history that turns a bounded result into seconds of disk traffic. |
| 182 | _, err := tx.ExecContext(ctx, `CREATE INDEX messages_snapshot_position ON messages(position DESC, event_sequence, valid_to)`) |
| 183 | return err |
| 184 | }}, {Version: 4, Apply: func(ctx context.Context, tx *sql.Tx) error { |
| 185 | _, err := tx.ExecContext(ctx, `ALTER TABLE messages ADD COLUMN visible_turn INTEGER NOT NULL DEFAULT 0`) |
| 186 | return err |
| 187 | }}, {Version: 5, Apply: func(ctx context.Context, tx *sql.Tx) error { |
| 188 | // Revision 5 stops duplicating inline message bodies into search_text. The |
| 189 | // rebuild metadata version forces old indexes through an atomic rebuild. |
| 190 | _, err := tx.ExecContext(ctx, `SELECT 1`) |
| 191 | return err |
| 192 | }}, {Version: 6, Apply: func(ctx context.Context, tx *sql.Tx) error { |
| 193 | _, err := tx.ExecContext(ctx, `ALTER TABLE messages ADD COLUMN visible_user INTEGER NOT NULL DEFAULT 0`) |
| 194 | return err |
| 195 | }}, {Version: 7, Apply: func(ctx context.Context, tx *sql.Tx) error { |
| 196 | _, err := tx.ExecContext(ctx, `CREATE TABLE turn_summaries (turn_id TEXT PRIMARY KEY, start_sequence INTEGER NOT NULL DEFAULT 0, end_sequence INTEGER NOT NULL DEFAULT 0, started_at INTEGER NOT NULL DEFAULT 0, ended_at INTEGER NOT NULL DEFAULT 0, final_message_id TEXT NOT NULL DEFAULT '')`) |
| 197 | if err != nil { |
| 198 | return err |
| 199 | } |
| 200 | _, err = tx.ExecContext(ctx, `CREATE INDEX turn_summaries_final ON turn_summaries(final_message_id)`) |
| 201 | return err |
| 202 | }}, {Version: 8, Apply: func(ctx context.Context, tx *sql.Tx) error { |
| 203 | _, err := tx.ExecContext(ctx, `CREATE TABLE turn_counts (turn_id TEXT NOT NULL, kind TEXT NOT NULL, id TEXT NOT NULL, sequence INTEGER NOT NULL, PRIMARY KEY(turn_id,kind,id))`) |
| 204 | return err |
| 205 | }}, {Version: 9, Apply: func(ctx context.Context, tx *sql.Tx) error { |
| 206 | _, err := tx.ExecContext(ctx, `CREATE TABLE submissions (session_id TEXT NOT NULL, submission_id TEXT NOT NULL, message_id TEXT NOT NULL, sequence INTEGER NOT NULL, PRIMARY KEY(session_id,submission_id)); CREATE INDEX submissions_message ON submissions(message_id)`) |
| 207 | return err |
| 208 | }}, {Version: 10, Apply: func(ctx context.Context, tx *sql.Tx) error { |
| 209 | _, err := tx.ExecContext(ctx, `CREATE TABLE tool_links (message_id TEXT NOT NULL,digest TEXT NOT NULL,call_id TEXT NOT NULL,is_result INTEGER NOT NULL,state TEXT NOT NULL,PRIMARY KEY(message_id,digest,call_id,is_result)); CREATE INDEX tool_links_call ON tool_links(call_id,is_result); CREATE TABLE tool_states (call_id TEXT NOT NULL,sequence INTEGER NOT NULL,state TEXT NOT NULL,PRIMARY KEY(call_id,sequence))`) |
| 210 | return err |
| 211 | }}} |
| 212 | |
| 213 | type SearchHistoryHit struct { |
| 214 | MessageID string `json:"messageId"` |
| 215 | Position int64 `json:"position"` |
| 216 | Role string `json:"role"` |
| 217 | Preview string `json:"preview"` |
| 218 | EventSequence uint64 `json:"eventSequence"` |
| 219 | } |
| 220 | |
| 221 | type SearchHistoryPage struct { |
| 222 | Hits []SearchHistoryHit `json:"hits"` |
| 223 | SnapshotSequence uint64 `json:"snapshotSequence"` |
| 224 | CoverageSequence uint64 `json:"coverageSequence"` |
| 225 | Status string `json:"status"` |
| 226 | NextCursor string `json:"nextCursor,omitempty"` |
| 227 | HasMore bool `json:"hasMore"` |
| 228 | } |
| 229 | |
| 230 | type searchHistoryCursor struct { |
| 231 | SessionID string `json:"sessionId"` |
| 232 | StorageRevision int `json:"storageRevision"` |
| 233 | SnapshotSequence uint64 `json:"snapshotSequence"` |
| 234 | BeforePosition int64 `json:"beforePosition"` |
| 235 | Projection int `json:"projection"` |
| 236 | QueryDigest string `json:"queryDigest"` |
| 237 | Generation string `json:"generation"` |
| 238 | } |
| 239 | |
| 240 | func (q *Query) HistoryPage(ctx context.Context, ref SessionRef, cursor string, limit int) (MessageHistoryPage, error) { |
| 241 | if q == nil { |
| 242 | return MessageHistoryPage{}, errors.New("session: nil query") |
| 243 | } |
| 244 | if err := ref.validate(q.hostID); err != nil { |
| 245 | return MessageHistoryPage{}, err |
| 246 | } |
| 247 | filesystem, ok := q.persistence.(*FilesystemPersistence) |
| 248 | if !ok { |
| 249 | return MessageHistoryPage{}, errors.New("session: history index requires filesystem persistence") |
| 250 | } |
| 251 | if limit <= 0 { |
| 252 | limit = HistoryPageDefaultLimit |
| 253 | } |
| 254 | limit = min(limit, HistoryPageMaxLimit) |
| 255 | path := historyIndexPath(filesystem.Root, ref.SessionID) |
| 256 | ready, err := q.historyLocatorReady(ctx, filesystem, ref.SessionID, path) |
| 257 | if err != nil { |
| 258 | return MessageHistoryPage{}, err |
| 259 | } |
| 260 | if !ready { |
| 261 | return MessageHistoryPage{Messages: []PersistentMessage{}, Status: "preparing"}, nil |
| 262 | } |
| 263 | handle, err := projectiondb.Open(ctx, projectiondb.OpenOptions{Path: path, Migrations: historyMigrations, RequireDisk: true, MaxOpenConns: 1}) |
| 264 | if err != nil { |
| 265 | return MessageHistoryPage{}, err |
| 266 | } |
| 267 | defer handle.DB.Close() |
| 268 | metadata, err := readHistoryIndexMetadata(ctx, handle.DB) |
| 269 | if err != nil { |
| 270 | return MessageHistoryPage{}, err |
| 271 | } |
| 272 | snapshot := metadata.durableSequence |
| 273 | // Empty cursor means the newest page. Subsequent cursors move toward older |
| 274 | // positions while the snapshot sequence remains fixed. |
| 275 | before := int64(^uint64(0) >> 1) |
| 276 | if cursor != "" { |
| 277 | parsed, err := decodeHistoryCursor(cursor) |
| 278 | if err != nil { |
| 279 | return MessageHistoryPage{}, err |
| 280 | } |
| 281 | if parsed.SessionID != ref.SessionID || parsed.StorageRevision != StorageRevision || parsed.Projection != historyIndexVersion || parsed.SnapshotSequence > snapshot || parsed.Generation != metadata.generation { |
| 282 | return MessageHistoryPage{Messages: []PersistentMessage{}, Status: "stale_cursor", CoverageSequence: metadata.durableSequence, Generation: metadata.generation}, nil |
| 283 | } |
| 284 | snapshot = parsed.SnapshotSequence |
| 285 | if parsed.BeforePosition <= 0 { |
| 286 | return MessageHistoryPage{}, errors.New("session: invalid history cursor position") |
| 287 | } |
| 288 | before = parsed.BeforePosition |
| 289 | } |
| 290 | return q.readMessageHistoryPage(ctx, handle.DB, filesystem, ref, metadata, snapshot, before, limit) |
| 291 | } |
| 292 | |
| 293 | func (q *Query) readMessageHistoryPage(ctx context.Context, db *sql.DB, filesystem *FilesystemPersistence, ref SessionRef, metadata historyIndexMetadata, snapshot uint64, before int64, limit int) (MessageHistoryPage, error) { |
| 294 | page := MessageHistoryPage{Messages: []PersistentMessage{}, SnapshotSequence: snapshot, CoverageSequence: metadata.durableSequence, Status: "ready", Generation: metadata.generation} |
| 295 | if err := db.QueryRowContext(ctx, `SELECT COALESCE(MAX(visible_turn),0) FROM messages WHERE event_sequence<=? AND (valid_to=0 OR valid_to>?)`, snapshot, snapshot).Scan(&page.TotalTurns); err != nil { |
| 296 | return MessageHistoryPage{}, err |
| 297 | } |
| 298 | rows, err := db.QueryContext(ctx, `SELECT message_id,position,version,role,preview,event_sequence,visible_turn,inline,content_digest,content_bytes,content_index_digest,COALESCE((SELECT submission_id FROM submissions WHERE submissions.message_id=messages.message_id AND submissions.sequence<=messages.event_sequence AND submissions.session_id=(SELECT value FROM metadata WHERE key='session_id') LIMIT 1),'') FROM messages WHERE position<? AND event_sequence<=? AND (valid_to=0 OR valid_to>?) ORDER BY position DESC LIMIT ?`, before, snapshot, snapshot, limit+1) |
| 299 | if err != nil { |
| 300 | return MessageHistoryPage{}, err |
| 301 | } |
| 302 | defer rows.Close() |
| 303 | encodedBytes := 0 |
| 304 | storageGeneration := q.storageGeneration(ref.SessionID) |
| 305 | for rows.Next() { |
| 306 | var message PersistentMessage |
| 307 | var inline []byte |
| 308 | var digest, indexDigest string |
| 309 | var contentBytes int64 |
| 310 | if err := rows.Scan(&message.MessageID, &message.Position, &message.Version, &message.Role, &message.Preview, &message.EventSequence, &message.VisibleTurn, &inline, &digest, &contentBytes, &indexDigest, &message.SubmissionID); err != nil { |
| 311 | return MessageHistoryPage{}, err |
| 312 | } |
| 313 | if len(page.Messages) == limit { |
| 314 | page.HasMore = true |
| 315 | break |
| 316 | } |
| 317 | message.Inline = append(json.RawMessage(nil), inline...) |
| 318 | if digest != "" { |
| 319 | contentRef := sessioncontent.Ref{Digest: digest, Bytes: contentBytes, IndexDigest: indexDigest, IntegrityBlock: sessioncontent.IntegrityBlockBytes, MediaType: "application/json"} |
| 320 | if contentBytes <= recentInlineBytes && encodedBytes+int(contentBytes) <= HistoryPageMaxBytes { |
| 321 | body, readErr := contentStoreForSessionDir(filepath.Join(filesystem.Root, ref.SessionID)).ReadRange(ctx, contentRef, 0, contentBytes) |
| 322 | if readErr != nil { |
| 323 | return MessageHistoryPage{}, readErr |
| 324 | } |
| 325 | message.Inline = json.RawMessage(body) |
| 326 | } else { |
| 327 | message.ContentRef = &contentRef |
| 328 | q.authorizeContentForGeneration(ref.SessionID, storageGeneration, digest, contentBytes, indexDigest) |
| 329 | } |
| 330 | } |
| 331 | encoded, _ := json.Marshal(message) |
| 332 | if len(page.Messages) > 0 && encodedBytes+len(encoded) > HistoryPageMaxBytes { |
| 333 | page.HasMore = true |
| 334 | break |
| 335 | } |
| 336 | encodedBytes += len(encoded) |
| 337 | page.Messages = append(page.Messages, message) |
| 338 | } |
| 339 | if err := rows.Err(); err != nil { |
| 340 | return MessageHistoryPage{}, err |
| 341 | } |
| 342 | if page.HasMore && len(page.Messages) > 0 { |
| 343 | oldest := page.Messages[len(page.Messages)-1] |
| 344 | page.NextCursor, err = encodeHistoryCursor(historyCursor{SessionID: ref.SessionID, StorageRevision: StorageRevision, SnapshotSequence: snapshot, BeforePosition: oldest.Position, Projection: historyIndexVersion, Generation: metadata.generation}) |
| 345 | if err != nil { |
| 346 | return MessageHistoryPage{}, err |
| 347 | } |
| 348 | } |
| 349 | slices.Reverse(page.Messages) |
| 350 | return page, nil |
| 351 | } |
| 352 | |
| 353 | // LocateMessage resolves a search hit or durable message id without scanning |
| 354 | // message bodies. A zero snapshot selects the locator's latest covered cut. |
| 355 | func (q *Query) LocateMessage(ctx context.Context, ref SessionRef, messageID string, snapshot uint64) (MessageLocation, error) { |
| 356 | if q == nil { |
| 357 | return MessageLocation{}, errors.New("session: nil query") |
| 358 | } |
| 359 | if err := ref.validate(q.hostID); err != nil { |
| 360 | return MessageLocation{}, err |
| 361 | } |
| 362 | messageID = strings.TrimSpace(messageID) |
| 363 | if messageID == "" { |
| 364 | return MessageLocation{}, errors.New("session: message id is required") |
| 365 | } |
| 366 | filesystem, ok := q.persistence.(*FilesystemPersistence) |
| 367 | if !ok { |
| 368 | return MessageLocation{}, errors.New("session: history locator requires filesystem persistence") |
| 369 | } |
| 370 | path := historyIndexPath(filesystem.Root, ref.SessionID) |
| 371 | ready, err := q.historyLocatorReady(ctx, filesystem, ref.SessionID, path) |
| 372 | if err != nil { |
| 373 | return MessageLocation{}, err |
| 374 | } |
| 375 | if !ready { |
| 376 | return MessageLocation{Status: "preparing", MessageID: messageID}, nil |
| 377 | } |
| 378 | handle, err := projectiondb.Open(ctx, projectiondb.OpenOptions{Path: path, Migrations: historyMigrations, RequireDisk: true, MaxOpenConns: 1}) |
| 379 | if err != nil { |
| 380 | return MessageLocation{}, err |
| 381 | } |
| 382 | defer handle.DB.Close() |
| 383 | metadata, err := readHistoryIndexMetadata(ctx, handle.DB) |
| 384 | if err != nil { |
| 385 | return MessageLocation{}, err |
| 386 | } |
| 387 | if snapshot == 0 { |
| 388 | snapshot = metadata.durableSequence |
| 389 | } |
| 390 | location := MessageLocation{Status: "ready", MessageID: messageID, SnapshotSequence: snapshot, CoverageSequence: metadata.durableSequence, Generation: metadata.generation} |
| 391 | if snapshot > metadata.durableSequence { |
| 392 | location.Status = "preparing" |
| 393 | return location, nil |
| 394 | } |
| 395 | err = handle.DB.QueryRowContext(ctx, `SELECT position,visible_turn FROM messages WHERE message_id=? AND event_sequence<=? AND (valid_to=0 OR valid_to>?) ORDER BY version DESC LIMIT 1`, messageID, snapshot, snapshot).Scan(&location.Position, &location.VisibleTurn) |
| 396 | if errors.Is(err, sql.ErrNoRows) { |
| 397 | location.Status = "not_found" |
| 398 | return location, nil |
| 399 | } |
| 400 | if err != nil { |
| 401 | return MessageLocation{}, err |
| 402 | } |
| 403 | location.Cursor, err = encodeHistoryCursor(historyCursor{SessionID: ref.SessionID, StorageRevision: StorageRevision, SnapshotSequence: snapshot, BeforePosition: location.Position + 1, Projection: historyIndexVersion, Generation: metadata.generation}) |
| 404 | return location, err |
| 405 | } |
| 406 | |
| 407 | func (q *Query) prepareHistoryLocator(filesystem *FilesystemPersistence, sessionID, path string) *historyPreparation { |
| 408 | q.historyMu.Lock() |
| 409 | if current := q.historyBuilds[sessionID]; current != nil { |
| 410 | select { |
| 411 | case <-current.done: |
| 412 | if current.err != nil { |
| 413 | q.historyMu.Unlock() |
| 414 | return current |
| 415 | } |
| 416 | // A completed build may have been invalidated by a newer append. |
| 417 | default: |
| 418 | q.historyMu.Unlock() |
| 419 | return current |
| 420 | } |
| 421 | } |
| 422 | preparation := &historyPreparation{done: make(chan struct{})} |
| 423 | q.historyBuilds[sessionID] = preparation |
| 424 | q.historyMu.Unlock() |
| 425 | q.rebuildMu.Lock() |
| 426 | if q.closed { |
| 427 | q.rebuildMu.Unlock() |
| 428 | preparation.err = context.Canceled |
| 429 | close(preparation.done) |
| 430 | return preparation |
| 431 | } |
| 432 | q.rebuildWG.Add(1) |
| 433 | q.rebuildMu.Unlock() |
| 434 | go func() { |
| 435 | defer q.rebuildWG.Done() |
| 436 | // A user-requested history page or locate: highest slot priority. |
| 437 | if err := q.slots.acquire(q.rebuildCtx, rebuildPriorityUser); err != nil { |
| 438 | preparation.err = err |
| 439 | close(preparation.done) |
| 440 | return |
| 441 | } |
| 442 | defer q.slots.release() |
| 443 | lock := q.projectionLock("history", sessionID) |
| 444 | lock.Lock() |
| 445 | preparation.err = ensureHistoryIndex(q.rebuildCtx, filesystem, sessionID, path) |
| 446 | lock.Unlock() |
| 447 | close(preparation.done) |
| 448 | }() |
| 449 | return preparation |
| 450 | } |
| 451 | |
| 452 | // HistoryShape returns the ordering and visible-turn boundaries for the |
| 453 | // current durable snapshot without materializing any message body. |
| 454 | func (q *Query) HistoryShape(ctx context.Context, ref SessionRef) (HistoryShape, error) { |
| 455 | filesystem, path, err := q.prepareHistoryIndex(ctx, ref) |
| 456 | if err != nil { |
| 457 | return HistoryShape{}, err |
| 458 | } |
| 459 | _ = filesystem |
| 460 | handle, err := projectiondb.Open(ctx, projectiondb.OpenOptions{Path: path, Migrations: historyMigrations, RequireDisk: true, MaxOpenConns: 1}) |
| 461 | if err != nil { |
| 462 | return HistoryShape{}, err |
| 463 | } |
| 464 | defer handle.DB.Close() |
| 465 | var snapshot uint64 |
| 466 | if err := scanMetadataUint(handle.DB.QueryRowContext(ctx, `SELECT value FROM metadata WHERE key='durable_sequence'`), &snapshot); err != nil { |
| 467 | return HistoryShape{}, err |
| 468 | } |
| 469 | rows, err := handle.DB.QueryContext(ctx, `SELECT position,visible_turn,role FROM messages WHERE event_sequence<=? AND (valid_to=0 OR valid_to>?) ORDER BY position`, snapshot, snapshot) |
| 470 | if err != nil { |
| 471 | return HistoryShape{}, err |
| 472 | } |
| 473 | defer rows.Close() |
| 474 | shape := HistoryShape{SnapshotSequence: snapshot, Positions: []HistoryPosition{}} |
| 475 | for rows.Next() { |
| 476 | var position HistoryPosition |
| 477 | if err := rows.Scan(&position.Position, &position.VisibleTurn, &position.Role); err != nil { |
| 478 | return HistoryShape{}, err |
| 479 | } |
| 480 | shape.Positions = append(shape.Positions, position) |
| 481 | shape.TotalTurns = max(shape.TotalTurns, position.VisibleTurn) |
| 482 | } |
| 483 | if err := rows.Err(); err != nil { |
| 484 | return HistoryShape{}, err |
| 485 | } |
| 486 | return shape, nil |
| 487 | } |
| 488 | |
| 489 | // HistoryWindow materializes exactly [start,end) from a previously obtained |
| 490 | // durable snapshot. The snapshot must still be representable by the current |
| 491 | // projection; an append is allowed because version intervals retain the old |
| 492 | // view, while an index rebuild remains transparent. |
| 493 | func (q *Query) HistoryWindow(ctx context.Context, ref SessionRef, snapshot uint64, start, end int) ([]provider.Message, error) { |
| 494 | filesystem, path, err := q.prepareHistoryIndex(ctx, ref) |
| 495 | if err != nil { |
| 496 | return nil, err |
| 497 | } |
| 498 | if start < 0 || end < start { |
| 499 | return nil, errors.New("session: invalid history window") |
| 500 | } |
| 501 | handle, err := projectiondb.Open(ctx, projectiondb.OpenOptions{Path: path, Migrations: historyMigrations, RequireDisk: true, MaxOpenConns: 1}) |
| 502 | if err != nil { |
| 503 | return nil, err |
| 504 | } |
| 505 | defer handle.DB.Close() |
| 506 | var current uint64 |
| 507 | if err := scanMetadataUint(handle.DB.QueryRowContext(ctx, `SELECT value FROM metadata WHERE key='durable_sequence'`), ¤t); err != nil { |
| 508 | return nil, err |
| 509 | } |
| 510 | if snapshot > current { |
| 511 | return nil, errors.New("session: history snapshot is newer than durable state") |
| 512 | } |
| 513 | if start == end { |
| 514 | return []provider.Message{}, nil |
| 515 | } |
| 516 | rows, err := handle.DB.QueryContext(ctx, `SELECT inline,content_digest,content_bytes,content_index_digest FROM messages WHERE position>? AND position<=? AND event_sequence<=? AND (valid_to=0 OR valid_to>?) ORDER BY position`, start, end, snapshot, snapshot) |
| 517 | if err != nil { |
| 518 | return nil, err |
| 519 | } |
| 520 | defer rows.Close() |
| 521 | content := contentStoreForSessionDir(filepath.Join(filesystem.Root, ref.SessionID)) |
| 522 | messages := make([]provider.Message, 0, end-start) |
| 523 | for rows.Next() { |
| 524 | var inline []byte |
| 525 | var digest, indexDigest string |
| 526 | var contentBytes int64 |
| 527 | if err := rows.Scan(&inline, &digest, &contentBytes, &indexDigest); err != nil { |
| 528 | return nil, err |
| 529 | } |
| 530 | body := json.RawMessage(inline) |
| 531 | if digest != "" { |
| 532 | body, err = resolveContentPayload(ctx, content, sessioncontent.Ref{Digest: digest, Bytes: contentBytes, IndexDigest: indexDigest, IntegrityBlock: sessioncontent.IntegrityBlockBytes, MediaType: "application/json"}) |
| 533 | if err != nil { |
| 534 | return nil, err |
| 535 | } |
| 536 | } |
| 537 | var message provider.Message |
| 538 | if err := json.Unmarshal(body, &message); err != nil { |
| 539 | return nil, fmt.Errorf("session: decode indexed message: %w", err) |
| 540 | } |
| 541 | messages = append(messages, message) |
| 542 | } |
| 543 | if err := rows.Err(); err != nil { |
| 544 | return nil, err |
| 545 | } |
| 546 | if len(messages) != end-start { |
| 547 | return nil, fmt.Errorf("session: history window length %d, want %d", len(messages), end-start) |
| 548 | } |
| 549 | return messages, nil |
| 550 | } |
| 551 |