fix(session-index): bound FTS rebuild memory

CHANGELOG.md

@@ -6,6 +6,17 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
 
+### Fixed
+- **Full-text session rebuilds now have a bounded working set.** Schema v5
+  still indexes every eligible session and the same title, user, assistant,
+  system, tool, and file-operation excerpts, but commits them in transactions
+  capped at 4 MiB or 128 sessions instead of one FTS5 segment per session.
+  Each batch releases SQLite caches and applies malloc-zone pressure relief,
+  preventing a large first-run rebuild from retaining a multi-gigabyte heap.
+  SQLite temporary storage is file-backed and its page cache is capped at
+  32 MiB; search semantics, snippets, roles, and the existing 512 KiB
+  per-session body policy are unchanged.
+
 ## [0.4.1] - 2026-08-20
 
 Release validation no longer races an FSEvents stream rearm when a watched

Sources/AgentSessionKit/Sessions/SessionIndexService.swift

@@ -18,6 +18,12 @@ public actor SessionIndexService {
     private let registry: SessionProviderRegistry
     private let bodyIndexing: @Sendable () -> Bool
 
+    /// Body bytes committed in one SQLite/FTS5 transaction. The cap is on
+    /// source excerpt bytes (not session count), so a batch remains bounded
+    /// even when several sessions hit the existing 512 KiB per-session cap.
+    static let indexBatchExcerptByteLimit = 4 * 1024 * 1024
+    static let indexBatchEntryLimit = 128
+
     public init(
         homeDirectory: String = RealHomeDirectory.path,
         store: SessionIndexStore,
@@ -60,13 +66,32 @@ public actor SessionIndexService {
         var seen: Set<String> = []
         let total = files.count
         var done = 0
+        var batch: [SessionIndexStore.IndexBatchEntry] = []
+        var batchExcerptBytes = 0
         for file in files {
             let hash = Self.pathHash(file.url.path)
             seen.insert(hash)
-            await index(file: file.url, adapter: file.adapter, pathHash: hash, bodies: indexBodies)
+            if let entry = await prepareIndexEntry(
+                file: file.url,
+                adapter: file.adapter,
+                pathHash: hash,
+                bodies: indexBodies
+            ) {
+                let entryBytes = entry.excerpts?.reduce(0) { $0 + $1.excerpt.utf8.count } ?? 0
+                if !batch.isEmpty,
+                   batchExcerptBytes + entryBytes > Self.indexBatchExcerptByteLimit
+                    || batch.count >= Self.indexBatchEntryLimit {
+                    await commit(batch)
+                    batch.removeAll(keepingCapacity: true)
+                    batchExcerptBytes = 0
+                }
+                batch.append(entry)
+                batchExcerptBytes += entryBytes
+            }
             done += 1
             progress?(done, total)
         }
+        await commit(batch)
 
         do {
             try await store.pruneMissing(existingPathHashes: seen)
@@ -147,46 +172,60 @@ public actor SessionIndexService {
         )
     }
 
-    // MARK: - One file
+    // MARK: - Bounded rebuild batches
 
-    private func index(
+    private func prepareIndexEntry(
         file url: URL,
         adapter: any SessionProviderAdapter,
         pathHash: String,
         bodies: Bool
-    ) async {
-        guard let fingerprint = Self.fingerprint(url) else { return }
+    ) async -> SessionIndexStore.IndexBatchEntry? {
+        guard let fingerprint = Self.fingerprint(url) else { return nil }
         do {
             if let cursor = try await store.fileCursor(pathHash: pathHash),
                cursor.mtimeNanos == fingerprint.mtimeNanos,
                cursor.size == fingerprint.size {
-                return
+                return nil
             }
-            let summary = try adapter.extractMetadata(fileURL: url)
-            let row = try await store.upsertSession(summary)
-            if bodies {
-                let document = try adapter.parseTranscript(fileURL: url, range: nil)
-                try await store.replaceMessages(
-                    sessionRow: row,
-                    excerpts: Self.excerpts(from: document, provider: summary.provider)
+            return try autoreleasepool {
+                let summary = try adapter.extractMetadata(fileURL: url)
+                let excerpts: [SessionIndexStore.MessageExcerpt]?
+                if bodies {
+                    let document = try adapter.parseTranscript(fileURL: url, range: nil)
+                    excerpts = Self.excerpts(from: document, provider: summary.provider)
+                } else {
+                    excerpts = nil
+                }
+                return SessionIndexStore.IndexBatchEntry(
+                    summary: summary,
+                    pathHash: pathHash,
+                    path: url.path,
+                    provider: adapter.provider,
+                    mtimeNanos: fingerprint.mtimeNanos,
+                    size: fingerprint.size,
+                    excerpts: excerpts
                 )
             }
-            try await store.saveFileCursor(
-                pathHash: pathHash,
-                path: url.path,
-                provider: adapter.provider,
-                mtimeNanos: fingerprint.mtimeNanos,
-                size: fingerprint.size,
-                sessionRow: row
-            )
         } catch {
             // The path is the user's own filesystem; log the file name
             // only, and let the next refresh retry.
             KitLog.warn(
                 "Session index: skipped \(adapter.provider.rawValue) file "
                     + "\(KitLog.sanitize(url.lastPathComponent))."
             )
+            return nil
+        }
+    }
+
+    private func commit(_ entries: [SessionIndexStore.IndexBatchEntry]) async {
+        guard !entries.isEmpty else { return }
+        do {
+            try await store.applyIndexBatch(entries)
+        } catch {
+            KitLog.warn("Session index: committing a bounded batch failed.")
         }
+        await store.releaseTransientMemory()
+        _ = malloc_zone_pressure_relief(nil, 0)
     }
 
     static func pathHash(_ path: String) -> String {

Sources/AgentSessionKit/Sessions/SessionIndexStore.swift

@@ -65,7 +65,12 @@ public actor SessionIndexStore {
     ///
     /// v4 also changes no columns; it backfills system and tool excerpts now
     /// that callers can choose the message roles a search is allowed to hit.
-    static let schemaVersion = 4
+    ///
+    /// v5 rebuilds the same logical index through bounded bulk transactions.
+    /// The body/search result set is unchanged; only the construction path is
+    /// different, so an interrupted v4 rebuild cannot keep its pathological
+    /// one-session-per-FTS-transaction layout.
+    static let schemaVersion = 5
 
     /// Open (or create) the index at `url`.
     ///
@@ -114,6 +119,8 @@ public actor SessionIndexStore {
             PRAGMA journal_mode=WAL;
             PRAGMA synchronous=NORMAL;
             PRAGMA foreign_keys=ON;
+            PRAGMA temp_store=FILE;
+            PRAGMA cache_size=-32768;
             """
         guard sqlite3_exec(database, preamble, nil, nil, nil) == SQLITE_OK else {
             throw SessionIndexError.open
@@ -285,39 +292,118 @@ public actor SessionIndexStore {
         }
     }
 
+    public struct IndexBatchEntry: Sendable {
+        public let summary: SessionSummary
+        public let pathHash: String
+        public let path: String
+        public let provider: SessionProvider
+        public let mtimeNanos: Int64
+        public let size: Int64
+        /// `nil` means metadata-only indexing. An empty array deliberately
+        /// clears a session whose readable transcript became empty.
+        public let excerpts: [MessageExcerpt]?
+
+        public init(
+            summary: SessionSummary,
+            pathHash: String,
+            path: String,
+            provider: SessionProvider,
+            mtimeNanos: Int64,
+            size: Int64,
+            excerpts: [MessageExcerpt]?
+        ) {
+            self.summary = summary
+            self.pathHash = pathHash
+            self.path = path
+            self.provider = provider
+            self.mtimeNanos = mtimeNanos
+            self.size = size
+            self.excerpts = excerpts
+        }
+    }
+
+    /// Commits a bounded group of sessions as one FTS transaction. Building a
+    /// large index one session per transaction creates thousands of tiny FTS5
+    /// segments and repeatedly merges them; batching keeps both segment count
+    /// and merge working sets bounded without dropping a single excerpt.
+    public func applyIndexBatch(_ entries: [IndexBatchEntry]) throws {
+        guard !entries.isEmpty else { return }
+        try execute("BEGIN IMMEDIATE")
+        do {
+            for entry in entries {
+                let row = try upsertSession(entry.summary)
+                if let excerpts = entry.excerpts {
+                    try replaceMessagesUncommitted(sessionRow: row, excerpts: excerpts)
+                }
+                try saveFileCursor(
+                    pathHash: entry.pathHash,
+                    path: entry.path,
+                    provider: entry.provider,
+                    mtimeNanos: entry.mtimeNanos,
+                    size: entry.size,
+                    sessionRow: row
+                )
+            }
+            try execute("COMMIT")
+        } catch {
+            try? execute("ROLLBACK")
+            throw error
+        }
+    }
+
     /// Replace every indexed excerpt for one session. The delete trigger
     /// clears the old FTS rows, so this is also how a shrinking
     /// transcript stops matching its removed text.
     public func replaceMessages(sessionRow: Int64, excerpts: [MessageExcerpt]) throws {
         try execute("BEGIN IMMEDIATE")
         do {
-            try run("DELETE FROM session_messages WHERE session_row = ?", [.integer(sessionRow)])
-            if !excerpts.isEmpty {
-                let statement = try prepare(
-                    "INSERT INTO session_messages(session_row, seq, role, excerpt) VALUES(?, ?, ?, ?)"
-                )
-                defer { sqlite3_finalize(statement) }
-                for excerpt in excerpts {
-                    sqlite3_reset(statement)
-                    sqlite3_clear_bindings(statement)
-                    bindAll([
-                        .integer(sessionRow),
-                        .integer(Int64(excerpt.seq)),
-                        .text(excerpt.role.rawValue),
-                        .text(excerpt.excerpt)
-                    ], to: statement)
-                    guard sqlite3_step(statement) == SQLITE_DONE else {
-                        throw SessionIndexError.statement
-                    }
-                }
-            }
+            try replaceMessagesUncommitted(sessionRow: sessionRow, excerpts: excerpts)
             try execute("COMMIT")
         } catch {
             try? execute("ROLLBACK")
             throw error
         }
     }
 
+    private func replaceMessagesUncommitted(
+        sessionRow: Int64,
+        excerpts: [MessageExcerpt]
+    ) throws {
+        try run("DELETE FROM session_messages WHERE session_row = ?", [.integer(sessionRow)])
+        guard !excerpts.isEmpty else { return }
+        let statement = try prepare(
+            "INSERT INTO session_messages(session_row, seq, role, excerpt) VALUES(?, ?, ?, ?)"
+        )
+        defer { sqlite3_finalize(statement) }
+        for excerpt in excerpts {
+            sqlite3_reset(statement)
+            sqlite3_clear_bindings(statement)
+            bindAll([
+                .integer(sessionRow),
+                .integer(Int64(excerpt.seq)),
+                .text(excerpt.role.rawValue),
+                .text(excerpt.excerpt)
+            ], to: statement)
+            guard sqlite3_step(statement) == SQLITE_DONE else {
+                throw SessionIndexError.statement
+            }
+        }
+    }
+
+    /// Asks SQLite to release connection-local caches after each bulk batch.
+    /// The host additionally applies malloc-zone pressure relief, so neither
+    /// layer retains a rebuild-sized high-water mark.
+    public func releaseTransientMemory() {
+        guard let database else { return }
+        _ = sqlite3_db_release_memory(database)
+        var logFrames: Int32 = 0
+        var checkpointedFrames: Int32 = 0
+        _ = sqlite3_wal_checkpoint_v2(
+            database, nil, SQLITE_CHECKPOINT_PASSIVE,
+            &logFrames, &checkpointedFrames
+        )
+    }
+
     /// Fingerprint of a file as of its last successful index pass.
     public struct FileCursor: Hashable, Sendable {
         public let mtimeNanos: Int64

Tests/AgentSessionKitTests/SessionIndexStoreTests.swift

@@ -97,6 +97,45 @@ final class SessionIndexStoreTests: XCTestCase {
         XCTAssertEqual(count, 2)
     }
 
+    func testBoundedIndexBatchPreservesSessionsBodiesAndCursors() async throws {
+        let store = try makeStore()
+        let first = summary(id: "first", path: "/first.jsonl")
+        let second = summary(id: "second", path: "/second.jsonl")
+        try await store.applyIndexBatch([
+            .init(
+                summary: first,
+                pathHash: "hash-first",
+                path: first.sourcePath,
+                provider: first.provider,
+                mtimeNanos: 11,
+                size: 22,
+                excerpts: [.init(seq: 0, role: .user, excerpt: "first complete body")]
+            ),
+            .init(
+                summary: second,
+                pathHash: "hash-second",
+                path: second.sourcePath,
+                provider: second.provider,
+                mtimeNanos: 33,
+                size: 44,
+                excerpts: [.init(seq: 0, role: .tool, excerpt: "second file operation")]
+            )
+        ])
+
+        let sessionCount = try await store.sessionCount()
+        let messageCount = try await store.messageCount()
+        let firstHits = try await store.search(text: "complete body", scopes: [.user])
+        let secondHits = try await store.search(text: "file operation", scopes: [.tool])
+        XCTAssertEqual(sessionCount, 2)
+        XCTAssertEqual(messageCount, 2)
+        XCTAssertEqual(firstHits.map(\.summary.sessionID), ["first"])
+        XCTAssertEqual(secondHits.map(\.summary.sessionID), ["second"])
+        let cursor = try await store.fileCursor(pathHash: "hash-second")
+        XCTAssertEqual(cursor?.mtimeNanos, 33)
+        XCTAssertEqual(cursor?.size, 44)
+        XCTAssertNotNil(cursor?.sessionRow)
+    }
+
     func testSummariesComeBackMostRecentlyActiveFirst() async throws {
         let store = try makeStore()
         try await store.upsertSession(summary(
@@ -618,9 +657,9 @@ final class SessionIndexStoreTests: XCTestCase {
 
         let rebuilt = try makeStore()
         let remaining = try await rebuilt.sessionCount()
-        XCTAssertEqual(SessionIndexStore.schemaVersion, 4)
+        XCTAssertEqual(SessionIndexStore.schemaVersion, 5)
         XCTAssertEqual(remaining, 0)
-        XCTAssertEqual(try userVersion(), 4)
+        XCTAssertEqual(try userVersion(), 5)
 
         try await rebuilt.upsertSession(summary(harness: .claudeCode, model: "claude-fable-5"))
         let model = try await rebuilt.allSummaries().first?.model