diff --git a/src/core/cache/maintenance.js b/src/core/cache/maintenance.js
index b8cfe0d3..351255cb 100644
--- a/src/core/cache/maintenance.js
+++ b/src/core/cache/maintenance.js
@@ -18,6 +18,8 @@ import { createLocalIcebergIO, tableUrlForDir } from './iceberg/resolver.js'
 import { columnsFromIcebergSchema } from './iceberg/schema.js'
 import { appendRowsToTable, currentPartitionSpec, currentSchema, scanRowsFromTable, sortColumnsFromMetadata, tableExists } from './iceberg/store.js'
 import { openStreamingAppend } from './iceberg/stream_append.js'
+import { buildSidecarsForTable } from '../search/sidecar_build.js'
+import { GREP_DATASET } from '../search/searchable_columns.js'
 import { isPlainObject } from '../util/json_util.js'
 
 /**
@@ -245,6 +247,45 @@ export async function maintainCache(opts) {
       report.errorMessage = err instanceof Error ? err.message : String(err)
       totalFailed++
     }
+    // The grep sidecar build, on the files the rewrite just finalized:
+    // compaction is the moment a file stops changing, so this is the one
+    // point in a file's life where an index can be built once and stay
+    // valid (LLP 0264 #lifecycle). Only the grep dataset carries indexes,
+    // and only a rewrite that committed has new files to index. Isolated
+    // from the partition's own verdict: an index that cannot be built
+    // costs speed, never the tick, and never correctness (the scan tier
+    // serves whatever has no sidecar).
+    // @ref LLP 0264#lifecycle [implements]: sidecars are built at maintenance right after compaction finalizes the generation's files
+    if (!opts.dryRun && report.compacted && !report.failed && part.dataset === GREP_DATASET) {
+      try {
+        const cursorAfter = readCursorSync(part.path)
+        const liveDir = path.join(part.path, generationLayout(cursorAfter).liveDir)
+        await withSpan(
+          'maintenance.grep_index',
+          {
+            [Attr.COMPONENT]: 'cache',
+            [Attr.OPERATION]: 'maintenance.grep_index',
+            [Attr.DATASET]: part.dataset,
+            status: 'ok',
+          },
+          async (span) => {
+            const built = await buildSidecarsForTable({ tableDir: liveDir })
+            report.sidecarsBuilt = built.built
+            report.sidecarsFailed = built.failed + built.quarantined
+            span.setAttribute('sidecars_built', built.built)
+            span.setAttribute('sidecars_present', built.present)
+            span.setAttribute('sidecars_failed', built.failed)
+            span.setAttribute('sidecars_quarantined', built.quarantined)
+          },
+          { component: 'cache' }
+        )
+      } catch (err) {
+        // Index absence is served by the scan tier, so a build-pass throw
+        // is a warning on the report, never a failed partition.
+        report.sidecarsFailed = (report.sidecarsFailed ?? 0) + 1
+        report.sidecarError = err instanceof Error ? err.message : String(err)
+      }
+    }
     reports.push(report)
     if (!report.failed) maintained++
     totalSnapshotsExpired += report.snapshotsExpired
@@ -1468,8 +1509,13 @@ function liveDataFileCount(partitionDir) {
 function countDataFiles(tableDir) {
   const dataDir = path.join(tableDir, 'data')
   try {
+    // Grep sidecars live beside their data files as `<file>.index.parquet`
+    // and MUST stay out of this count: it feeds the compaction heuristics
+    // and the LLP 0199 baseline gate, so counting sidecars would read a
+    // just-compacted-and-indexed partition as "grew since compaction" and
+    // rewrite it every tick forever.
     return fs.readdirSync(dataDir, { withFileTypes: true })
-      .filter((e) => e.isFile() && e.name.endsWith('.parquet'))
+      .filter((e) => e.isFile() && e.name.endsWith('.parquet') && !e.name.endsWith('.index.parquet'))
       .length
   } catch {
     return 0
@@ -1516,22 +1562,38 @@ function measureMetadataDir(tableDir) {
 }
 
 /**
+ * Data bytes only: grep sidecars are excluded for the same reason
+ * `countDataFiles` excludes them; the avg-file-size heuristic divides
+ * these bytes by that count, so the two must see the same file set.
+ *
+ * The test is `includes`, not `endsWith`, because the build's publish
+ * scratch (`<file>.index.parquet.<uuid>.tmp`) is index bytes too, and it
+ * is the half of the pair that survives a crash: `countDataFiles` already
+ * skips it for want of a `.parquet` suffix, so counting its bytes would
+ * break the shared-file-set invariant above in the dangerous direction.
+ * The average would read HIGHER than the partition really is, and
+ * `needsCompaction` compacts when the average is LOW, so a genuinely
+ * fragmented partition would look healthy and go unrewritten until its
+ * generation retires.
+ *
  * @param {string} tableDir
  * @returns {number}
  */
 function measureDataDir(tableDir) {
-  return measureDir(path.join(tableDir, 'data'))
+  return measureDir(path.join(tableDir, 'data'), (name) => !name.includes('.index.parquet'))
 }
 
 /**
  * @param {string} dir
+ * @param {(name: string) => boolean} [include]
  * @returns {number}
  */
-function measureDir(dir) {
+function measureDir(dir, include) {
   let total = 0
   try {
     for (const entry of fs.readdirSync(dir, { withFileTypes: true })) {
       if (!entry.isFile()) continue
+      if (include && !include(entry.name)) continue
       try {
         total += fs.statSync(path.join(dir, entry.name)).size
       } catch { /* skip */ }
diff --git a/src/core/cache/types.d.ts b/src/core/cache/types.d.ts
index 22842b0b..c90bd66f 100644
--- a/src/core/cache/types.d.ts
+++ b/src/core/cache/types.d.ts
@@ -295,6 +295,12 @@ export interface MaintenancePartitionReport {
   dataFilesAfter: number
   /** Bytes the compaction rewrite actually wrote; absent when it did not run. */
   compactedBytesWritten?: number
+  /** Grep sidecars built for the just-compacted generation; absent when the build did not run. */
+  sidecarsBuilt?: number
+  /** Files whose sidecar build failed or is quarantined; the scan tier serves them. */
+  sidecarsFailed?: number
+  /** The build pass's own error, when the pass itself threw (never fails the partition). */
+  sidecarError?: string
   // Compaction of this partition is known not to reduce its data-file
   // count under the writer running now: either this run's rewrite
   // reproduced the count it started from, or a previous one did and the
diff --git a/src/core/search/grep_service.js b/src/core/search/grep_service.js
index cec852d0..3d5b2ce6 100644
--- a/src/core/search/grep_service.js
+++ b/src/core/search/grep_service.js
@@ -19,7 +19,7 @@ import {
   resolveCallerClass,
 } from '../query/visibility.js'
 import { cellText, compileMatcher, makeSnippet, MAX_MATCH_COLUMNS } from './matcher.js'
-import { SCAN_COLUMNS, SEARCHABLE_COLUMNS } from './searchable_columns.js'
+import { GREP_DATASET, SCAN_COLUMNS, SEARCHABLE_COLUMNS, sidecarPathFor } from './searchable_columns.js'
 
 /**
  * The local grep-search service: the client half of LLP 0264, mirroring the
@@ -58,7 +58,7 @@ import { SCAN_COLUMNS, SEARCHABLE_COLUMNS } from './searchable_columns.js'
  * @import { UsagePolicyResolver } from '../../../src/core/usage-policy/types.js'
  */
 
-const DATASET = 'ai_gateway_messages'
+const DATASET = GREP_DATASET
 
 /**
  * Rows between abort checks inside one brute-scanned file. The deadline has
@@ -233,11 +233,19 @@ export async function executeGrepSearch(args) {
        * the limit promises. Trimming is amortized (it runs once the buffer
        * has doubled), so the walk still costs a bounded number of hits
        * rather than one per match in the cache.
+       *
+       * The same trim runs over the indexed tier's per-file buffer below,
+       * for the same reason: a buffer that grew with the file rather than
+       * with the budget would give up the memory bound this walk promises,
+       * and cutting it in walk order would reintroduce the bug.
+       *
+       * @param {GrepSearchHit[]} list
        */
-      const trimHits = () => {
-        sortHits(hits)
-        if (hits.length > budget) hits.length = budget
+      const trimBuffer = (list) => {
+        sortHits(list)
+        if (list.length > budget) list.length = budget
       }
+      const trimHits = () => trimBuffer(hits)
       /** @param {Record<string, unknown>} row */
       const collect = (row) => {
         hits.push(toHit(row, matcher))
@@ -246,10 +254,7 @@ export async function executeGrepSearch(args) {
 
       /**
        * Search one file through its sidecar. Returns false when the index
-       * proved unusable BEFORE it produced a row, which hands the file to
-       * the scan tier instead; a failure after the first row cannot be
-       * retried that way (the hits already collected would be counted a
-       * second time), so it propagates.
+       * proved unusable, which hands that one file to the scan tier below.
        *
        * The existence probe only rules out a missing sidecar. A sidecar
        * that exists but cannot be read (a half-written index from a killed
@@ -261,39 +266,90 @@ export async function executeGrepSearch(args) {
        * correctness input; LLP 0264 #lifecycle says it never is, so a
        * poisoned file is brute-scanned exactly like an unindexed one.
        *
+       * The attempt therefore runs into a local buffer and commits only
+       * once the index tier finished. A sidecar can tear mid-read (an
+       * external writer; the build's own publish is atomic), and rows
+       * already pushed to the shared buffer could not be taken back, so
+       * committing as it went would leave the choice between double-counting
+       * them on the rescan and failing the whole query. Buffering makes
+       * degrading the file a decision this function can still take at any
+       * point in the read.
+       *
+       * An abort is the one failure that commits the buffer instead of
+       * discarding it: it ends the walk rather than degrading the file, so
+       * there is no rescan to double-count against and the rows the index
+       * already produced belong in the partial answer.
+       *
        * @param {{ filePath: string, deletedPositions: Set<bigint> | undefined }} file
        * @param {Awaited<ReturnType<typeof io.reader>>} indexFile
+       * @param {string} sidecarUrl
        * @returns {Promise<boolean>}
        */
-      const searchIndexed = async (file, indexFile) => {
-        // No `limit` is passed down: a purged or withheld row is filtered
-        // AFTER parquetFind accepts it, so a passed-down limit would count
-        // rows this walk then discards and under-return. The generator is
-        // simply not pulled past the budget instead.
-        const rows = parquetFind({
-          query: matcher.hypQuery,
-          url: file.filePath,
-          indexFile,
-          asyncBufferFactory: async ({ url }) => await io.reader(url),
-          rowFilter: accept,
-          signal,
-        })
-        let produced = false
+      const searchIndexed = async (file, indexFile, sidecarUrl) => {
+        /** @type {GrepSearchHit[]} */
+        const found = []
+        let withheldHere = 0
         try {
+          // No `limit` is passed down: a purged or withheld row is filtered
+          // AFTER parquetFind accepts it, so a passed-down limit would count
+          // rows this walk then discards and under-return. The generator is
+          // simply not pulled past the budget instead.
+          const rows = parquetFind({
+            query: matcher.hypQuery,
+            url: file.filePath,
+            indexFile,
+            asyncBufferFactory: async ({ url }) => await io.reader(url),
+            rowFilter: accept,
+            signal,
+          })
           for await (const row of rows) {
-            produced = true
             if (file.deletedPositions?.has(BigInt(/** @type {number} */ (row.__index__)))) continue
             if (withheld?.(row)) {
-              localOnly.withheldRows += 1
+              withheldHere += 1
               continue
             }
-            collect(row)
+            found.push(toHit(row, matcher))
+            if (found.length >= budget * 2) trimBuffer(found)
           }
         } catch (err) {
-          if (produced || isAbort(err, signal)) throw err
+          if (isAbort(err, signal)) {
+            // Commit before the abort propagates. A deadline lands INSIDE a
+            // file, not between files (hypgrep checks the signal at every
+            // coalesced range boundary), and a newest-first walk makes the
+            // interrupted file the newest one the caller most wants, so
+            // discarding the buffer would answer zero for exactly that file
+            // and lose its withheld-row count out of the report. Safe
+            // precisely because an abort ends the walk: this file is never
+            // rescanned, so no row can be counted twice. It still does not
+            // count as indexed, because it was not served whole.
+            for (const hit of found) hits.push(hit)
+            localOnly.withheldRows += withheldHere
+            throw err
+          }
+          // Both files are named, because the read that failed spans both:
+          // `parquetFind` opens the source data file through the same
+          // factory as the sidecar and runs the row filter per row, so a
+          // torn source parquet reaches this line too and then fails the
+          // rescan below. Deleting the sidecar is the usual remedy and this
+          // warning is its only notice (nothing rebuilds one in place), but
+          // the line must not claim to have proved which file is at fault.
+          getLogger('query').warn('grep_search.indexed_read_failed', {
+            [Attr.COMPONENT]: 'query',
+            [Attr.OPERATION]: 'query.grep_search',
+            sidecar_file: urlToPath(sidecarUrl),
+            data_file: urlToPath(file.filePath),
+            error_message: err instanceof Error ? err.message : String(err),
+          })
           return false
         }
         indexedFiles += 1
+        localOnly.withheldRows += withheldHere
+        // Appended, not spread: `limit` is validated as a positive safe
+        // integer but is not bounded above, so one file may fill a buffer of
+        // millions, and a spread of that many arguments is an argument-count
+        // overflow, not a push.
+        for (const hit of found) hits.push(hit)
+        if (hits.length >= budget * 2) trimHits()
         return true
       }
 
@@ -303,7 +359,7 @@ export async function executeGrepSearch(args) {
         // #lifecycle): probe the filesystem, then degrade this one file to
         // the scan tier if the read races a delete. Results stay exact
         // either way; only the wall clock changes.
-        const sidecarUrl = file.filePath.replace(/\.parquet$/i, '.index.parquet')
+        const sidecarUrl = sidecarPathFor(file.filePath)
         /** @type {Awaited<ReturnType<typeof io.reader>> | null} */
         let indexFile = null
         if (fs.existsSync(urlToPath(sidecarUrl))) {
@@ -317,7 +373,7 @@ export async function executeGrepSearch(args) {
             indexFile = null
           }
         }
-        if (indexFile && await searchIndexed(file, indexFile)) return
+        if (indexFile && await searchIndexed(file, indexFile, sidecarUrl)) return
         scannedFiles += 1
         const sourceFile = await io.reader(file.filePath)
         const rows = await parquetReadObjects({ file: sourceFile, columns: SCAN_COLUMNS })
@@ -426,9 +482,9 @@ function sortHits(hits) {
  * Project a matched row to the shared hit shape. Matched columns come from
  * the same allowlist the row predicate tested, in the set's order, so the
  * content column leads and a column the predicate could not have matched is
- * never reported. Cells render through `cellText` first, so the JSON column
- * (`tool_args`) that produced a `rowTest` match also produces the matched
- * column and its snippet here rather than being skipped as a non-string.
+ * never reported. Cells render through `cellText` first, the same coercion
+ * the row predicate applied, so a row that matched always names at least one
+ * matched column here rather than reporting a hit with none.
  *
  * @param {Record<string, unknown>} row
  * @param {GrepSearchMatcher} matcher
diff --git a/src/core/search/index_worker.js b/src/core/search/index_worker.js
new file mode 100644
index 00000000..605e4f74
--- /dev/null
+++ b/src/core/search/index_worker.js
@@ -0,0 +1,161 @@
+// @ts-check
+
+import { Worker } from 'node:worker_threads'
+
+/**
+ * The main-thread handle on the grep sidecar build worker, ported from the
+ * server's `index-worker.js` shape by decision.
+ *
+ * `createIndex` is seconds of straight-line CPU per sidecar and the daemon
+ * is single-threaded, so running it inline stops ingest, queries, and the
+ * health probe for as long as it runs. Yielding between files does not
+ * help: one file is already longer than any request budget. So the build
+ * moves to a worker thread and the main loop keeps only the scheduling,
+ * the file reads, and the publish.
+ *
+ * One worker serves one sidecar-build pass and is closed with it: within a
+ * pass the loop builds one file at a time, so a pool would buy nothing,
+ * while a worker that outlived the pass would hold an index-sized heap
+ * between maintenance ticks for no one.
+ *
+ * @ref LLP 0264#lifecycle [implements]: the build runs in a worker thread so maintenance never blocks the daemon loop
+ */
+
+/**
+ * @param {{
+ *   log?: { info(msg: string, fields?: object): void, warn(msg: string, fields?: object): void },
+ *   threadUrl?: URL,
+ * }} [args] threadUrl swaps the worker module so tests can drive
+ *   uncommanded death and protocol breaches; production callers omit it
+ */
+export function createIndexWorker({ log, threadUrl } = {}) {
+  /** @type {Worker | null} */
+  let worker = null
+  /**
+   * The live worker's in-flight builds. Reassigned per spawn: each
+   * worker's handlers capture their own map, so a crashed worker's late
+   * error or exit event can only reject its own builds, never a
+   * replacement worker's.
+   * @type {Map<number, { resolve: (bytes: Uint8Array) => void, reject: (err: Error) => void }>}
+   */
+  let pending = new Map()
+  let nextId = 1
+  let closed = false
+  /** Re-evaluate the live worker's ref state; rebound per spawn. */
+  let syncRef = () => {}
+
+  function ensureWorker() {
+    if (worker) return worker
+    const started = new Worker(threadUrl ?? new URL('./index_worker_thread.js', import.meta.url))
+    /** @type {typeof pending} */
+    const owned = new Map()
+    pending = owned
+
+    /**
+     * Fail this worker's in-flight builds. A worker that dies (OOM,
+     * terminate, an unparseable module) must surface as a rejected build
+     * so the caller records the failure and moves on; a silently hung
+     * promise would wedge the maintenance pass forever.
+     *
+     * @param {Error} err
+     */
+    function failAll(err) {
+      const inflight = [...owned.values()]
+      owned.clear()
+      for (const entry of inflight) entry.reject(err)
+    }
+
+    /**
+     * Hold the event loop open exactly while a build is in flight. An
+     * always-unref'd worker deadlocks any process whose loop would
+     * otherwise drain (the awaiting caller's promise is resolved only by
+     * a worker message that an empty loop never waits for), while an
+     * always-ref'd one would hold a shutting-down daemon for the seconds
+     * a build takes. Ref-while-pending gives both callers what they mean:
+     * an awaited build completes, an idle worker never keeps the process
+     * up.
+     */
+    function updateRef() {
+      if (owned.size > 0) started.ref()
+      else started.unref()
+    }
+
+    started.on('message', (/** @type {{ id: number, index?: ArrayBuffer, error?: string }} */ message) => {
+      const entry = owned.get(message.id)
+      if (!entry) return
+      owned.delete(message.id)
+      updateRef()
+      // A message with no index bytes and no error is a protocol breach,
+      // and resolving it would publish an empty sidecar that lists as
+      // done. The length test is the whole guard: `new ArrayBuffer(0)` is
+      // truthy, so a presence-only check would let zero bytes through as a
+      // successful build.
+      if (message.index && message.index.byteLength > 0) entry.resolve(new Uint8Array(message.index))
+      else entry.reject(new Error(message.error ?? 'index worker answered with no index bytes and no error'))
+    })
+    started.on('error', (err) => {
+      if (worker === started) worker = null
+      failAll(err instanceof Error ? err : new Error(String(err)))
+    })
+    started.on('exit', (code) => {
+      if (worker === started) worker = null
+      failAll(new Error(`grep index worker exited before answering (code ${code})`))
+    })
+    started.unref()
+    syncRef = updateRef
+    worker = started
+    log?.info('grep_index.worker_started', {})
+    return started
+  }
+
+  return {
+    /**
+     * Build one sidecar's bytes off the event loop. A whole, exclusively
+     * owned source buffer is transferred (detached here on return);
+     * anything else, such as a view into Node's Buffer pool, is copied
+     * first. Either way callers must not touch `sourceBytes` afterwards.
+     *
+     * @param {Uint8Array} sourceBytes
+     * @returns {Promise<Uint8Array>}
+     */
+    build(sourceBytes) {
+      if (closed) return Promise.reject(new Error('grep index worker is closed'))
+      const active = ensureWorker()
+      const id = nextId
+      nextId += 1
+      const source = transferable(sourceBytes)
+      return new Promise((resolve, reject) => {
+        pending.set(id, { resolve, reject })
+        syncRef()
+        active.postMessage({ id, source }, [source])
+      })
+    },
+    /** Terminate the worker; in-flight builds reject through the exit hook. */
+    async close() {
+      closed = true
+      const active = worker
+      worker = null
+      if (active) await active.terminate()
+    },
+  }
+}
+
+/**
+ * An ArrayBuffer that is safe to hand to `postMessage`'s transfer list.
+ * Node's Buffer allocator hands out views into a shared pool for small
+ * reads, and transferring a pooled buffer would detach every unrelated
+ * Buffer sharing it, so anything that is not a whole, exclusively owned
+ * buffer is copied first.
+ *
+ * @param {Uint8Array} bytes
+ * @returns {ArrayBuffer}
+ */
+function transferable(bytes) {
+  const buffer = bytes.buffer
+  if (buffer instanceof ArrayBuffer && bytes.byteOffset === 0 && bytes.byteLength === buffer.byteLength) {
+    return buffer
+  }
+  const copy = new ArrayBuffer(bytes.byteLength)
+  new Uint8Array(copy).set(bytes)
+  return copy
+}
diff --git a/src/core/search/index_worker_thread.js b/src/core/search/index_worker_thread.js
new file mode 100644
index 00000000..79c4315e
--- /dev/null
+++ b/src/core/search/index_worker_thread.js
@@ -0,0 +1,106 @@
+// @ts-check
+
+import { parentPort } from 'node:worker_threads'
+
+import { parquetMetadataAsync, parquetSchema } from 'hyparquet'
+import { ByteWriter } from 'hyparquet-writer'
+import { createIndex } from 'hypgrep'
+
+import { SEARCHABLE_COLUMNS } from './searchable_columns.js'
+
+/**
+ * The worker end of the grep sidecar build, ported from the server's
+ * `index-worker-thread.js`. One message is one sidecar: source bytes in,
+ * index bytes out, both moved by transfer, so neither multi-megabyte
+ * buffer is duplicated to cross the thread boundary. Transfer is not the
+ * same as never copying: `getBuffer()` slices the writer's backing store,
+ * so the index is copied once here before it is handed to the transfer
+ * list, and a source the caller could not transfer outright was copied
+ * once on the way in.
+ *
+ * This thread does no IO. It never opens the cache and never writes a
+ * sidecar: the caller reads the source and performs the single
+ * write-then-rename that publishes the result. That is what keeps sidecar
+ * existence honest as the completion marker (LLP 0264 #lifecycle) across
+ * a thread that can be killed at any instant.
+ *
+ * @import { FileMetaData, SchemaTree } from 'hyparquet'
+ */
+
+if (!parentPort) throw new Error('index_worker_thread must be started as a worker thread')
+const port = parentPort
+
+port.on('message', (/** @type {{ id: number, source: ArrayBuffer }} */ message) => {
+  void handle(message)
+})
+
+/**
+ * Build one index and post it back. Errors travel as a message, not as an
+ * uncaught rejection: a source file the builder cannot parse must fail
+ * that one file, not tear down the thread mid-pass.
+ *
+ * @param {{ id: number, source: ArrayBuffer }} message
+ */
+async function handle({ id, source }) {
+  try {
+    const bytes = new Uint8Array(source)
+    // hypgrep reads through an AsyncBuffer; the whole file is already
+    // resident, so slicing is a copy out of memory.
+    const sourceFile = {
+      byteLength: bytes.byteLength,
+      /**
+       * @param {number} [start]
+       * @param {number} [end]
+       * @returns {ArrayBuffer}
+       */
+      slice(start, end) {
+        const view = bytes.subarray(start ?? 0, end ?? bytes.byteLength)
+        const out = new ArrayBuffer(view.byteLength)
+        new Uint8Array(out).set(view)
+        return out
+      },
+    }
+    const indexFile = new ByteWriter()
+    const metadata = await parquetMetadataAsync(sourceFile)
+    // @ref LLP 0264#shared [implements]: only the searchable columns are indexed; hypgrep's default would n-gram every string column, and the server measured system_text alone at 90.8% of decoded index text
+    const textColumns = searchableStringColumns(metadata)
+    await createIndex({ sourceFile, sourceMetadata: metadata, indexFile, textColumns })
+    const index = indexFile.getBuffer()
+    port.postMessage({ id, index }, [index])
+  } catch (err) {
+    const error = err instanceof Error ? err.message : String(err)
+    port.postMessage({ id, error })
+  }
+}
+
+/**
+ * The source's string leaf columns narrowed to the searchable set, in
+ * schema order. hypgrep's root export offers no schema walk, so the
+ * string-leaf test (UTF8 converted type or STRING logical type on a
+ * childless node) is restated here against the same hyparquet schema tree
+ * hypgrep itself reads. A source none of whose string columns are
+ * searchable fails loudly: building an index that can never match is a
+ * misconfiguration, not a degenerate success. The caller isolates the
+ * throw, counts it against this file, and quarantines it after the
+ * attempt budget; search still serves the file by scanning.
+ *
+ * @param {FileMetaData} metadata
+ * @returns {string[]}
+ */
+function searchableStringColumns(metadata) {
+  /** @type {string[]} */
+  const stringColumns = []
+  /** @param {SchemaTree} node */
+  function walk(node) {
+    const { element, children } = node
+    const isString = element.converted_type === 'UTF8' || element.logical_type?.type === 'STRING'
+    if (isString && children.length === 0) stringColumns.push(element.name)
+    for (const child of children) walk(child)
+  }
+  walk(parquetSchema(metadata))
+  const textColumns = stringColumns.filter((name) => SEARCHABLE_COLUMNS.has(name))
+  if (textColumns.length === 0) {
+    throw new Error(`source has no searchable string columns (string columns: ${stringColumns.join(', ') || 'none'})`)
+  }
+  return textColumns
+}
diff --git a/src/core/search/matcher.js b/src/core/search/matcher.js
index 04b734f2..80cde3d4 100644
--- a/src/core/search/matcher.js
+++ b/src/core/search/matcher.js
@@ -109,14 +109,22 @@ export function compileMatcher(query, regex) {
 }
 
 /**
- * The searchable text of one cell. Most searchable columns hold STRING,
- * but `tool_args` is a JSON column (iceberg `variant`), so it reads back
- * from parquet as an object. Without this coercion a column named in the
- * allowlist could never produce a hit: a `typeof value === 'string'` gate
- * drops the object form, so searching for a file path or a shell command
- * inside a tool call answers zero while the indexed tier, which reads the
- * column's own text, answers otherwise. That split is precisely the drift
- * the shared module exists to prevent.
+ * The searchable text of one cell. Every allowlisted column holds STRING
+ * today, so on the live path this coercion is a no-op. It stays because it
+ * is what makes `rowTest`, `test`, `locate` and `makeSnippet` answer
+ * identically for whatever a cell decodes to, string or not: the consumer
+ * loop row-tests a row and then per-cell tests each allowlisted column, and
+ * a per-cell predicate narrower than the row predicate makes that loop
+ * report a hit with no matched columns, or throw on `value.slice`.
+ *
+ * What it deliberately does NOT do is make a non-string column searchable
+ * end to end. The indexed tier cannot follow it there: an index worker
+ * skips a VARIANT column, so a cell only this coercion can read would match
+ * on the scan tier and answer zero on the indexed one, which is the drift
+ * the shared module exists to prevent. That is why `tool_args` is out of
+ * the allowlist rather than carried by this function (see
+ * `searchable_columns.js`, and hyparam/hypaware#977, which would restore it
+ * on both tiers at once).
  *
  * A decoded object is rendered as its keys and primitive leaves, one per
  * line, rather than as `JSON.stringify` text. Serialized text carries the
@@ -125,16 +133,11 @@ export function compileMatcher(query, regex) {
  * query for a Windows path or for a multi-line command would miss the
  * very cell it names. The leaf rendering searches what the user sees.
  *
- * A cell that arrives already serialized (the paths that carry `tool_args`
- * verbatim, as `parseMaybeJson` handles elsewhere in the contract) is
- * matched as the text it is: nothing here knows the column, so parsing
- * every JSON-looking string would change what a `content_text` holding a
- * JSON document matches. That asymmetry is bounded (it only shows up for
- * a query containing a JSON escape) and is for the scan paths in T4/T5 to
- * settle with the server, which is the only place the column name is in
- * hand.
+ * A cell that arrives already serialized is matched as the text it is:
+ * nothing here knows the column, so parsing every JSON-looking string would
+ * change what a `content_text` holding a JSON document matches.
  *
- * @ref LLP 0264#shared [implements]: every allowlisted column is really searchable on every tier, including the JSON one
+ * @ref LLP 0264#shared [implements]: one cell coercion, so the row predicate and the per-cell predicate cannot disagree about a cell
  * @param {unknown} value
  * @returns {string}
  */
diff --git a/src/core/search/searchable_columns.js b/src/core/search/searchable_columns.js
index d7739f3f..247f830d 100644
--- a/src/core/search/searchable_columns.js
+++ b/src/core/search/searchable_columns.js
@@ -13,10 +13,18 @@
  * Insertion order is meaningful: matched columns are reported in this
  * order, so the content column leads a hit's snippets.
  *
- * All but one of these hold STRING. `tool_args` is a JSON column, so it
- * reads back from parquet as an object rather than text; the matcher's
- * `cellText` renders it before testing, because a column in this set that
- * cannot produce a hit is worse than one that is absent from it.
+ * Every column here holds STRING. `tool_args` is deliberately absent, and
+ * its absence is a gap recorded rather than left to be rediscovered (the
+ * discipline of server LLP 0157 #identifier-columns). It is the dataset's
+ * one VARIANT column (iceberg `variant`, a JSON cell), and no tier in
+ * either repository can produce a hit from it: both index workers filter
+ * VARIANT out before building, and the server's shared row predicate gates
+ * on `typeof value === 'string' && value !== ''`, which an object-valued
+ * cell fails. A column in this set that cannot produce a hit is worse than
+ * one absent from it, because it is decoded on every brute scan for the
+ * cost and named in the tool description for the promise while answering
+ * zero. hyparam/hypaware#977 restores the coverage on every tier at once,
+ * once hypgrep can index VARIANT.
  *
  * The set is a constant, not configuration. Sharing it is what makes
  * "zero hits" mean the same thing locally and remotely, and a per-install
@@ -28,7 +36,6 @@
 export const SEARCHABLE_COLUMNS = constantSet([
   'content_text',
   'tool_name',
-  'tool_args',
   'session_id',
   'conversation_id',
   'agent_id',
@@ -38,6 +45,32 @@ export const SEARCHABLE_COLUMNS = constantSet([
   'git_remote',
 ])
 
+/**
+ * The one dataset grep search covers, on both repositories: the client
+ * greps its own `ai_gateway_messages` cache, the server the same dataset's
+ * cache and archive. Named here beside the columns it scopes so the search
+ * service and the sidecar-build pass cannot disagree about which tables
+ * carry indexes.
+ */
+export const GREP_DATASET = 'ai_gateway_messages'
+
+/**
+ * The sidecar path beside a data file: hypgrep's own default, which is a
+ * contract. Any reader with byte access to the cache can search a file
+ * with the stock hypgrep CLI, no daemon involved. It lives beside the
+ * allowlist for the same reason `GREP_DATASET` does: the build pass that
+ * publishes a sidecar and the search service that probes for one must
+ * spell this path identically, or the build writes an index nobody looks
+ * for and every file silently falls back to the scan tier. Takes a
+ * filesystem path or a `file://` URL; only the extension is rewritten.
+ *
+ * @param {string} dataFile
+ * @returns {string}
+ */
+export function sidecarPathFor(dataFile) {
+  return dataFile.replace(/\.parquet$/i, '.index.parquet')
+}
+
 /**
  * A Set that cannot be added to, deleted from, or cleared. `SCAN_COLUMNS`
  * below is a load-time snapshot of the allowlist, so a caller mutating the
diff --git a/src/core/search/sidecar_build.js b/src/core/search/sidecar_build.js
new file mode 100644
index 00000000..7ea002fe
--- /dev/null
+++ b/src/core/search/sidecar_build.js
@@ -0,0 +1,178 @@
+// @ts-check
+
+import { randomUUID } from 'node:crypto'
+import fs from 'node:fs'
+import fsPromises from 'node:fs/promises'
+
+import { urlToPath } from '../cache/iceberg/resolver.js'
+import { listLiveDataFiles } from '../cache/iceberg/store.js'
+import { Attr, getLogger } from '../observability/index.js'
+import { createIndexWorker } from './index_worker.js'
+import { sidecarPathFor } from './searchable_columns.js'
+
+/**
+ * The sidecar build pass: give every live data file of a just-compacted
+ * table a hypgrep `.index.parquet` beside it, so the grep service's
+ * indexed tier serves the partition's history instead of brute-scanning
+ * it. Runs at maintenance, after the compaction that finalized the files
+ * (LLP 0264 #lifecycle): compaction is the moment a file stops changing,
+ * so its index can never go stale against its own rows (purge is handled
+ * at read time by position, not by rebuild).
+ *
+ * Sidecar existence IS the completion marker: there is no ledger to
+ * drift, a killed daemon leaves nothing half-claimed (the publish is a
+ * write-then-rename, all or nothing), and a pass over a generation
+ * rebuilds whatever that generation is missing. A file that cannot be
+ * indexed is quarantined after a bounded number of attempts and served by
+ * the scan tier: index presence is purely a performance property, never a
+ * correctness one.
+ *
+ * What that does NOT buy, and callers must not assume it does: a retry.
+ * The pass runs only behind a committed compaction, and a compaction
+ * always publishes a fresh generation directory, so the files this pass
+ * skipped or failed on are gone by the time another pass runs. A missing
+ * sidecar is repaired by the next compaction rewriting the rows into a
+ * new file, not by re-attempting the old one.
+ *
+ * A sidecar also freezes the allowlist it was built over. hypgrep records
+ * the indexed columns in the index itself (`hypgrep.text_columns`) and
+ * prunes candidate blocks to them, and nothing on the read side compares
+ * that stamp against today's `SEARCHABLE_COLUMNS`. So ADDING a column to
+ * the allowlist does not reach a file that is already indexed, and under
+ * the no-retry lifecycle above it never will while that generation lives:
+ * the new column answers on the scan tier and zero on the indexed one,
+ * which is the tier disagreement the shared allowlist exists to prevent.
+ * Any change to the set (hyparam/hypaware#977 is the first one queued) has
+ * to invalidate the existing sidecars, not merely start building new ones.
+ * Recorded here rather than left to be rediscovered, in the same spirit as
+ * the column note in `searchable_columns.js`.
+ *
+ * @ref LLP 0264#lifecycle [implements]: sidecar existence is the idempotency marker, no ledger; an unindexed or poisoned file is brute-scanned, so index state is never a correctness input
+ */
+
+/**
+ * How many failed builds one file may cost before the pass stops
+ * attempting it. Three, because the two failure families this counter
+ * separates are cheap to tell apart: a transient one (a worker killed by
+ * shutdown) clears well inside three passes, while a deterministic one
+ * (the file hypgrep cannot index) costs three builds to prove and then
+ * costs nothing. The ledger is in-memory and process-lifetime on purpose:
+ * a persisted poison list would outlive the bug it recorded. Note it only
+ * bites where one path is offered to more than one pass, which under the
+ * compaction gate above means a caller driving this module directly; the
+ * maintenance pass sees fresh paths every time.
+ */
+const MAX_INDEX_ATTEMPTS = 3
+
+/**
+ * @param {{ maxAttempts?: number }} [args]
+ */
+export function createIndexQuarantine({ maxAttempts = MAX_INDEX_ATTEMPTS } = {}) {
+  /** @type {Map<string, number>} */
+  const failures = new Map()
+  return {
+    /** @param {string} key */
+    isQuarantined(key) {
+      return (failures.get(key) ?? 0) >= maxAttempts
+    },
+    /** @param {string} key */
+    recordFailure(key) {
+      const attempts = (failures.get(key) ?? 0) + 1
+      failures.set(key, attempts)
+      return { attempts, quarantined: attempts >= maxAttempts }
+    },
+    /** @param {string} key */
+    clear(key) {
+      failures.delete(key)
+    },
+  }
+}
+
+/** The ledger every pass in this process shares unless a caller injects its own (tests do). */
+const processQuarantine = createIndexQuarantine()
+
+/**
+ * Build the missing sidecars for one Iceberg table directory, one file at
+ * a time (the pass's memory bound is one data file plus its index). Files
+ * whose sidecar already exists are skipped by the existence marker;
+ * quarantined files are skipped without spending a build. A failure is
+ * recorded, logged, and isolated to its file: the rest of the pass runs.
+ *
+ * @param {{
+ *   tableDir: string,
+ *   quarantine?: ReturnType<typeof createIndexQuarantine>,
+ *   worker?: ReturnType<typeof createIndexWorker>,
+ *   log?: { info(msg: string, fields?: object): void, warn(msg: string, fields?: object): void },
+ * }} args
+ * @returns {Promise<{ built: number, present: number, failed: number, quarantined: number }>}
+ */
+export async function buildSidecarsForTable({ tableDir, quarantine = processQuarantine, worker, log }) {
+  const logger = log ?? getLogger('cache')
+  const report = { built: 0, present: 0, failed: 0, quarantined: 0 }
+  const files = await listLiveDataFiles(tableDir)
+  if (files.length === 0) return report
+  const ownWorker = worker ?? createIndexWorker({ log: logger })
+  try {
+    for (const file of files) {
+      const sourcePath = urlToPath(file.filePath)
+      const sidecarPath = sidecarPathFor(sourcePath)
+      if (fs.existsSync(sidecarPath)) {
+        report.present += 1
+        continue
+      }
+      if (quarantine.isQuarantined(sourcePath)) {
+        report.quarantined += 1
+        continue
+      }
+      // Publish atomically: rename is the only step that makes the
+      // sidecar exist, so a crash mid-write can never leave a partial file
+      // that lists as a finished index. The scratch name carries a random
+      // token because a fixed `<sidecar>.tmp` is only safe for one writer:
+      // the daemon's tick and a hand-run `hyp` sharing a cache would
+      // interleave their writes into the same scratch file and then rename
+      // the mixture into place as a finished sidecar. It also ends outside
+      // `.parquet`, so an in-flight or abandoned file joins no data-file
+      // count, and it is removed on the failure path rather than left to
+      // wait for the generation's retirement.
+      const tmpPath = `${sidecarPath}.${randomUUID()}.tmp`
+      try {
+        const bytes = await fsPromises.readFile(sourcePath)
+        const index = await ownWorker.build(bytes)
+        await fsPromises.writeFile(tmpPath, index)
+        await fsPromises.rename(tmpPath, sidecarPath)
+        quarantine.clear(sourcePath)
+        report.built += 1
+      } catch (err) {
+        await fsPromises.rm(tmpPath, { force: true }).catch(() => {})
+        const { attempts, quarantined } = quarantine.recordFailure(sourcePath)
+        report.failed += 1
+        const message = err instanceof Error ? err.message : String(err)
+        // The data file is named on every line: with one warning per failed
+        // build and a per-file attempt budget, an operator who cannot tell
+        // three retries of one poisoned file from three distinct failures
+        // cannot act on either.
+        logger.warn('grep_index.build_failed', {
+          [Attr.COMPONENT]: 'cache',
+          [Attr.OPERATION]: 'maintenance.grep_index',
+          data_file: sourcePath,
+          attempts,
+          quarantined,
+          error_message: message,
+        })
+        if (quarantined) {
+          logger.warn('grep_index.file_quarantined', {
+            [Attr.COMPONENT]: 'cache',
+            [Attr.OPERATION]: 'maintenance.grep_index',
+            data_file: sourcePath,
+            attempts,
+          })
+        }
+      }
+    }
+  } finally {
+    // A caller-provided worker outlives the pass (the caller owns its
+    // lifecycle); one created here is closed with it.
+    if (!worker) await ownWorker.close()
+  }
+  return report
+}
diff --git a/src/core/search/types.d.ts b/src/core/search/types.d.ts
index e33ee6bc..b517b6e9 100644
--- a/src/core/search/types.d.ts
+++ b/src/core/search/types.d.ts
@@ -47,9 +47,9 @@ export interface GrepSearchResult {
  * A compiled grep query: `hypQuery` feeds hypgrep's index pruning,
  * `test`/`locate` run per cell (locate finds the snippet window),
  * `rowTest` is the whole-row predicate the scan paths share. The cell
- * entry points take `unknown` because a searchable cell is not always
- * text: `tool_args` decodes to an object, and every one of them renders
- * it the same way, so a cell `rowTest` accepted cannot then miss here.
+ * entry points take `unknown` because a row cell is not always text, and
+ * every one of them renders it the same way, so a cell `rowTest` accepted
+ * cannot then miss here.
  */
 export interface GrepSearchMatcher {
   hypQuery: string | RegExp
diff --git a/test/core/cache-compaction-effectiveness.test.js b/test/core/cache-compaction-effectiveness.test.js
index 8ac07ea6..ae0b6541 100644
--- a/test/core/cache-compaction-effectiveness.test.js
+++ b/test/core/cache-compaction-effectiveness.test.js
@@ -118,7 +118,9 @@ async function liveDataFiles(dir) {
   const dataDir = path.join(dir, cursor.tableDir ?? 'table', 'data')
   const entries = await fs.readdir(dataDir, { withFileTypes: true })
   return entries
-    .filter((e) => e.isFile() && e.name.endsWith('.parquet'))
+    // Data files only: compaction now leaves a grep sidecar beside each
+    // one, and truncating a sidecar would not make the rewrite fail.
+    .filter((e) => e.isFile() && e.name.endsWith('.parquet') && !e.name.endsWith('.index.parquet'))
     .map((e) => path.join(dataDir, e.name))
 }
 
diff --git a/test/core/search-grep-service.test.js b/test/core/search-grep-service.test.js
index 721e400d..42c54ab8 100644
--- a/test/core/search-grep-service.test.js
+++ b/test/core/search-grep-service.test.js
@@ -238,18 +238,35 @@ test('literal matching is case-insensitive; regex mode is operator-shaped', asyn
   assert.deepEqual(rx.hits.map((h) => h.sessionId), ['s2'])
 })
 
-test('the JSON column matches through cellText, and reports as the matched column', async () => {
+test('a row matching only in tool_args returns zero hits from BOTH tiers', async () => {
+  // The invariant is tier agreement, not coverage. `tool_args` is VARIANT,
+  // the index worker filters it out, so an indexed file can never answer a
+  // match through it; the scan tier must therefore not answer one either.
+  // Dropping the column from the allowlist is what makes the two agree, and
+  // hyparam/hypaware#977 is where they would agree the other way instead.
   const toolRow = mkRow({
     date: '2026-08-11',
     session_id: 's3',
     tool_name: 'Read',
     tool_args: { file_path: '/repo/hidden_needle_path.js' },
   })
-  const { storage } = await makeCache([[toolRow]])
-  const res = await grep(storage, { query: 'hidden_needle_path' })
-  assert.equal(res.hits.length, 1)
-  assert.equal(res.hits[0].matches[0].column, 'tool_args')
-  assert.match(res.hits[0].matches[0].snippet, /hidden_needle_path/)
+  const { storage, tableDir } = await makeCache([[toolRow]])
+
+  const scanned = await grep(storage, { query: 'hidden_needle_path' })
+  assert.equal(scanned.hits.length, 0)
+  assert.equal(scanned.indexedFiles, 0)
+  assert.ok(scanned.scannedFiles >= 1, 'the scan tier really read the file')
+
+  assert.ok(await buildSidecars(tableDir()) >= 1, 'a sidecar was built')
+  const indexed = await grep(storage, { query: 'hidden_needle_path' })
+  assert.equal(indexed.hits.length, 0)
+  assert.equal(indexed.scannedFiles, 0)
+  assert.ok(indexed.indexedFiles >= 1, 'the indexed tier really served the file')
+
+  // The row itself is still reachable, so the zero above is the column
+  // being unsearchable rather than the row being missing.
+  const byName = await grep(storage, { query: 'Read' })
+  assert.deepEqual(byName.hits.map((h) => h.sessionId), ['s3'])
 })
 
 test('local-only rows are withheld from lower-rank callers, and only from them', async () => {
diff --git a/test/core/search-matcher.test.js b/test/core/search-matcher.test.js
index 1b3ba12b..b79fa249 100644
--- a/test/core/search-matcher.test.js
+++ b/test/core/search-matcher.test.js
@@ -131,17 +131,19 @@ test('literal offsets index the original value, not a lowercased copy', () => {
   assert.equal(makeSnippet(value, compileMatcher('needle', false)), value)
 })
 
-test('a JSON cell is searchable, not silently skipped', () => {
-  // tool_args is a JSON column (iceberg variant), so it reads back from
-  // parquet as an object. It is in the allowlist, so it has to be able to
-  // produce a hit; a typeof-string gate would make it dead weight that is
-  // still decoded on every brute scan.
+test('a non-string cell is coerced, not silently skipped', () => {
+  // Every allowlisted column holds STRING today, so the coercion is not
+  // there to cover a JSON column any more (tool_args is out: #977). It is
+  // there so the row predicate is never wider than the per-cell one for
+  // whatever a cell actually decodes to.
   const matcher = compileMatcher('src/core/search', false)
-  assert.equal(matcher.rowTest({ tool_args: { file_path: 'src/core/search/matcher.js' } }), true)
-  assert.equal(matcher.rowTest({ tool_args: '{"file_path":"src/core/search/matcher.js"}' }), true)
-  assert.equal(matcher.rowTest({ tool_args: { file_path: 'elsewhere.js' } }), false)
-  // An excluded column stays excluded whatever shape it holds.
+  assert.equal(matcher.rowTest({ content_text: { file_path: 'src/core/search/matcher.js' } }), true)
+  assert.equal(matcher.rowTest({ content_text: '{"file_path":"src/core/search/matcher.js"}' }), true)
+  assert.equal(matcher.rowTest({ content_text: { file_path: 'elsewhere.js' } }), false)
+  // An excluded column stays excluded whatever shape it holds, tool_args
+  // now included: the coercion never widens the allowlist.
   assert.equal(matcher.rowTest({ attributes: { path: 'src/core/search/matcher.js' } }), false)
+  assert.equal(matcher.rowTest({ tool_args: { file_path: 'src/core/search/matcher.js' } }), false)
 })
 
 test('cellText renders only the shapes a searchable cell can hold', () => {
@@ -161,26 +163,26 @@ test('cellText renders only the shapes a searchable cell can hold', () => {
   assert.equal(cellText(cyclic), 'file_path\na.js\nself')
 })
 
-test('a JSON cell is searched as its decoded text, escapes and all', () => {
+test('an object cell is searched as its decoded text, escapes and all', () => {
   // JSON.stringify would store a Windows path as C:\\Users\\me and a
   // shell command's newline as the two characters \\n, so a literal query
   // for either would miss the very cell it names.
   const args = { path: 'C:\\Users\\me', command: 'cd /repo\nnpm test', quoted: 'say "hi"' }
-  assert.equal(compileMatcher('C:\\Users\\me', false).rowTest({ tool_args: args }), true)
-  assert.equal(compileMatcher('cd /repo\nnpm test', false).rowTest({ tool_args: args }), true)
-  assert.equal(compileMatcher('say "hi"', false).rowTest({ tool_args: args }), true)
-  // The keys are searchable too, so a query naming a tool argument finds it.
-  assert.equal(compileMatcher('command', false).rowTest({ tool_args: args }), true)
+  assert.equal(compileMatcher('C:\\Users\\me', false).rowTest({ content_text: args }), true)
+  assert.equal(compileMatcher('cd /repo\nnpm test', false).rowTest({ content_text: args }), true)
+  assert.equal(compileMatcher('say "hi"', false).rowTest({ content_text: args }), true)
+  // The keys are rendered too, so a query naming one finds the cell.
+  assert.equal(compileMatcher('command', false).rowTest({ content_text: args }), true)
 })
 
-test('test, locate and makeSnippet agree with rowTest on a JSON cell', () => {
+test('test, locate and makeSnippet agree with rowTest on a non-string cell', () => {
   // The consumer loop is: rowTest the row, then test each allowlisted cell
   // and snippet the ones that matched. A per-cell predicate that only takes
   // strings makes that loop report a hit with no matched columns, or throw
   // on value.slice, for exactly the column the row matched through.
   const matcher = compileMatcher('file_path', false)
   const cell = { file_path: 'src/core/search/matcher.js' }
-  assert.equal(matcher.rowTest({ tool_args: cell }), true)
+  assert.equal(matcher.rowTest({ content_text: cell }), true)
   assert.equal(matcher.test(cell), true)
   assert.deepEqual(matcher.locate(cell), { index: 0, length: 9 })
   assert.equal(makeSnippet(cell, matcher), 'file_path\nsrc/core/search/matcher.js')
diff --git a/test/core/search-searchable-columns.test.js b/test/core/search-searchable-columns.test.js
index 15e4d719..ff59e460 100644
--- a/test/core/search-searchable-columns.test.js
+++ b/test/core/search-searchable-columns.test.js
@@ -18,7 +18,6 @@ test('the searchable allowlist is exactly the server-shared set, content column
   assert.deepEqual([...SEARCHABLE_COLUMNS], [
     'content_text',
     'tool_name',
-    'tool_args',
     'session_id',
     'conversation_id',
     'agent_id',
@@ -37,6 +36,17 @@ test('the bulk machinery columns stay out of the allowlist', () => {
   }
 })
 
+test('the VARIANT column stays out of the allowlist, and out of the scan', () => {
+  // `tool_args` is the dataset's one iceberg `variant` column. No tier in
+  // either repository can produce a hit from it: the index workers filter
+  // VARIANT out before building, and the server's row predicate takes
+  // strings only. So it is absent rather than decoded on every brute scan
+  // for nothing. hyparam/hypaware#977 restores it once hypgrep can index
+  // VARIANT; until then its absence is the thing both repos agree on.
+  assert.equal(SEARCHABLE_COLUMNS.has('tool_args'), false)
+  assert.equal(SCAN_COLUMNS.includes('tool_args'), false)
+})
+
 test('every searchable column exists on the ai_gateway_messages schema', () => {
   const schema = new Set(AI_GATEWAY_MESSAGE_COLUMNS.map((column) => column.name))
   for (const column of SEARCHABLE_COLUMNS) {
diff --git a/test/core/search-sidecar-build.test.js b/test/core/search-sidecar-build.test.js
new file mode 100644
index 00000000..3d9047c4
--- /dev/null
+++ b/test/core/search-sidecar-build.test.js
@@ -0,0 +1,232 @@
+// @ts-check
+
+import test from 'node:test'
+import assert from 'node:assert/strict'
+import fsSync from 'node:fs'
+import fs from 'node:fs/promises'
+import os from 'node:os'
+import path from 'node:path'
+
+import { urlToPath } from '../../src/core/cache/iceberg/resolver.js'
+import { listLiveDataFiles } from '../../src/core/cache/iceberg/store.js'
+import { maintainCache } from '../../src/core/cache/maintenance.js'
+import { appendRowsToSourceTable } from '../../src/core/cache/partition.js'
+import { createQueryStorageService, resolveIcebergDir } from '../../src/core/cache/storage.js'
+import { executeGrepSearch } from '../../src/core/search/grep_service.js'
+import { sidecarPathFor } from '../../src/core/search/searchable_columns.js'
+import { buildSidecarsForTable, createIndexQuarantine } from '../../src/core/search/sidecar_build.js'
+import { aiGatewayDatasetRegistration } from '../../hypaware-core/plugins-workspace/ai-gateway/src/dataset.js'
+
+/**
+ * @import { ColumnSpec } from '../../hypaware-plugin-kernel-types.js'
+ */
+
+const DATASET = 'ai_gateway_messages'
+
+/** @type {ColumnSpec[]} */
+const COLUMNS = [
+  { name: 'session_id', type: 'STRING', nullable: false },
+  { name: 'conversation_id', type: 'STRING', nullable: true },
+  { name: 'agent_id', type: 'STRING', nullable: true },
+  { name: 'cwd', type: 'STRING', nullable: true },
+  { name: 'content_text', type: 'STRING', nullable: true },
+  { name: 'date', type: 'STRING', nullable: false },
+  { name: 'part_id', type: 'STRING', nullable: false },
+  { name: 'message_id', type: 'STRING', nullable: false },
+  { name: 'message_created_at', type: 'TIMESTAMP', nullable: false },
+  { name: 'client_name', type: 'STRING', nullable: true },
+]
+
+let rowSeq = 0
+
+/** @param {Record<string, unknown>} [over] */
+function mkRow(over = {}) {
+  rowSeq += 1
+  const date = typeof over.date === 'string' ? over.date : '2026-08-10'
+  return {
+    session_id: 's1',
+    conversation_id: null,
+    agent_id: null,
+    cwd: '/home/open-proj',
+    content_text: null,
+    date,
+    part_id: `m${rowSeq}#0`,
+    message_id: `m${rowSeq}`,
+    message_created_at: new Date(`${date}T00:00:00Z`).getTime() + rowSeq * 1000,
+    client_name: 'test',
+    ...over,
+  }
+}
+
+/**
+ * @param {Record<string, unknown>[][]} batches
+ * @param {string} [dataset]
+ */
+async function makeCache(batches, dataset = DATASET) {
+  const cacheRoot = await fs.mkdtemp(path.join(os.tmpdir(), 'hyp-sidecar-'))
+  const declaration = aiGatewayDatasetRegistration().cachePartitioning
+  for (const batch of batches) {
+    await appendRowsToSourceTable(cacheRoot, dataset, ['source=test'], COLUMNS, batch, { declaration })
+  }
+  const storage = createQueryStorageService({ cacheRoot })
+  const partitionDir = path.join(cacheRoot, 'datasets', dataset, 'source=test')
+  return { cacheRoot, storage, partitionDir, tableDir: () => resolveIcebergDir(partitionDir) }
+}
+
+/** A worker stand-in whose every build fails. */
+function failingWorker() {
+  return {
+    build: () => Promise.reject(new Error('synthetic build failure')),
+    close: async () => {},
+  }
+}
+
+const quietLog = { info() {}, warn() {} }
+
+const OLD = mkRow({ date: '2026-08-10', session_id: 's1', content_text: 'alpha needle one' })
+const NEW = mkRow({ date: '2026-08-12', session_id: 's2', content_text: 'the needle two' })
+
+test('buildSidecarsForTable builds one sidecar per file, idempotently, and grep serves them', async () => {
+  const { storage, tableDir } = await makeCache([[OLD], [NEW]])
+  const first = await buildSidecarsForTable({ tableDir: tableDir(), log: quietLog })
+  assert.equal(first.built >= 2, true)
+  assert.equal(first.present, 0)
+  assert.equal(first.failed, 0)
+  for (const file of await listLiveDataFiles(tableDir())) {
+    assert.ok(fsSync.existsSync(sidecarPathFor(urlToPath(file.filePath))), 'every live file has a sidecar')
+  }
+  const second = await buildSidecarsForTable({ tableDir: tableDir(), log: quietLog })
+  assert.equal(second.built, 0)
+  assert.equal(second.present, first.built, 'existence is the completion marker; nothing rebuilds')
+
+  const res = await executeGrepSearch({ storage, query: 'needle', limit: 10, includeLocalOnly: true })
+  assert.equal(res.hits.length, 2)
+  assert.equal(res.indexedFiles, first.built, 'the search runs on the indexed tier')
+  assert.equal(res.scannedFiles, 0)
+  assert.deepEqual(res.hits.map((h) => h.sessionId), ['s2', 's1'])
+})
+
+test('a failing build quarantines after three attempts and the scan tier still serves the file', async () => {
+  const { storage, tableDir } = await makeCache([[OLD]])
+  const quarantine = createIndexQuarantine()
+  for (let attempt = 1; attempt <= 3; attempt++) {
+    const report = await buildSidecarsForTable({
+      tableDir: tableDir(), quarantine, worker: failingWorker(), log: quietLog,
+    })
+    assert.equal(report.failed, 1, `attempt ${attempt} spends a build and fails`)
+    assert.equal(report.built, 0)
+  }
+  const afterQuarantine = await buildSidecarsForTable({
+    tableDir: tableDir(), quarantine, worker: failingWorker(), log: quietLog,
+  })
+  assert.equal(afterQuarantine.failed, 0, 'a quarantined file costs no further builds')
+  assert.equal(afterQuarantine.quarantined, 1)
+
+  const res = await executeGrepSearch({ storage, query: 'needle', limit: 10, includeLocalOnly: true })
+  assert.equal(res.hits.length, 1, 'the unindexed file is served by the scan tier')
+  assert.equal(res.scannedFiles, 1)
+  assert.equal(res.indexedFiles, 0)
+})
+
+test('a corrupt sidecar degrades that one file to the scan tier instead of failing the search', async () => {
+  const { storage, tableDir } = await makeCache([[OLD]])
+  const [file] = await listLiveDataFiles(tableDir())
+  await fs.writeFile(sidecarPathFor(urlToPath(file.filePath)), 'not a parquet file')
+  const res = await executeGrepSearch({ storage, query: 'needle', limit: 10, includeLocalOnly: true })
+  assert.equal(res.hits.length, 1)
+  assert.equal(res.indexedFiles, 0)
+  assert.equal(res.scannedFiles, 1, 'the unreadable sidecar fell back to the brute scan')
+})
+
+test('maintenance compaction finalizes files and builds their sidecars', async () => {
+  const { cacheRoot, storage, tableDir } = await makeCache([[OLD], [NEW]])
+  const result = await maintainCache({ cacheRoot, force: true })
+  const report = result.partitions.find((p) => p.dataset === DATASET)
+  assert.ok(report)
+  assert.equal(report.compacted, true)
+  assert.ok((report.sidecarsBuilt ?? 0) >= 1, 'the rewrite queued index builds for its files')
+  assert.equal(report.sidecarsFailed ?? 0, 0)
+
+  const files = await listLiveDataFiles(tableDir())
+  assert.ok(files.length >= 1)
+  for (const file of files) {
+    assert.ok(fsSync.existsSync(sidecarPathFor(urlToPath(file.filePath))), 'every finalized file is indexed')
+  }
+  const res = await executeGrepSearch({ storage, query: 'needle', limit: 10, includeLocalOnly: true })
+  assert.equal(res.hits.length, 2)
+  assert.equal(res.scannedFiles, 0)
+})
+
+test('sidecars do not re-trigger compaction: the data-file counters exclude them', async () => {
+  const { cacheRoot } = await makeCache([[OLD], [NEW]])
+  await maintainCache({ cacheRoot, force: true })
+  // No new data flushed since the rewrite; a second unforced tick must see
+  // a converged partition, not one that "grew" by its own index files.
+  const second = await maintainCache({ cacheRoot })
+  const report = second.partitions.find((p) => p.dataset === DATASET)
+  assert.ok(report)
+  assert.equal(report.compacted, false, 'the sidecars did not read as growth')
+})
+
+test('an orphaned publish scratch counts as index bytes, not data bytes', async () => {
+  // A build killed between the write and the rename leaves
+  // `<file>.index.parquet.<uuid>.tmp` in the live data dir, and nothing
+  // reaps it before the generation retires. `countDataFiles` already skips
+  // it (no `.parquet` suffix), so the byte measure has to skip it too: the
+  // avg-file-size heuristic compacts when the average is LOW, so counting a
+  // large orphan makes a fragmented partition read as healthy and go
+  // unrewritten.
+  const { cacheRoot, tableDir } = await makeCache([[OLD], [NEW]])
+  const files = await listLiveDataFiles(tableDir())
+  assert.ok(files.length >= 2)
+  let dataBytes = 0
+  for (const file of files) dataBytes += (await fs.stat(urlToPath(file.filePath))).size
+  const avgBytes = dataBytes / files.length
+  const orphan = `${sidecarPathFor(urlToPath(files[0].filePath))}.orphaned-build.tmp`
+  await fs.writeFile(orphan, Buffer.alloc(dataBytes * 4))
+
+  // Due by a hair on the real data bytes; not due at all if the orphan's
+  // bytes join the average.
+  const result = await maintainCache({
+    cacheRoot,
+    config: { compact_file_count: 1000, compact_avg_file_bytes: Math.ceil(avgBytes) + 1 },
+  })
+  const report = result.partitions.find((p) => p.dataset === DATASET)
+  assert.ok(report)
+  assert.equal(report.compacted, true, 'the orphaned scratch did not inflate the average file size')
+})
+
+test('a non-grep dataset is compacted without sidecars', async () => {
+  const { cacheRoot, tableDir } = await makeCache([[mkRow({ content_text: 'needle' })]], 'other_dataset')
+  const result = await maintainCache({ cacheRoot, force: true })
+  const report = result.partitions.find((p) => p.dataset === 'other_dataset')
+  assert.ok(report)
+  assert.equal(report.compacted, true)
+  assert.equal(report.sidecarsBuilt, undefined, 'the build pass never ran')
+  for (const file of await listLiveDataFiles(tableDir())) {
+    assert.equal(fsSync.existsSync(sidecarPathFor(urlToPath(file.filePath))), false)
+  }
+})
+
+test('a retired generation dies whole, sidecars included', async () => {
+  const { cacheRoot, partitionDir } = await makeCache([[OLD], [NEW]])
+  await maintainCache({ cacheRoot, force: true })
+  const compactedDir = resolveIcebergDir(partitionDir)
+  assert.ok((await listLiveDataFiles(compactedDir)).length >= 1)
+
+  // New data, then a second rewrite: the first compacted generation (with
+  // its sidecars inside) is retired.
+  const declaration = aiGatewayDatasetRegistration().cachePartitioning
+  await appendRowsToSourceTable(cacheRoot, DATASET, ['source=test'], COLUMNS,
+    [mkRow({ date: '2026-08-14', session_id: 's3', content_text: 'needle three' })], { declaration })
+  await maintainCache({ cacheRoot, force: true })
+  assert.notEqual(resolveIcebergDir(partitionDir), compactedDir, 'a fresh generation is live')
+  assert.ok(fsSync.existsSync(path.join(compactedDir, '.retired')), 'the old generation is marked retired')
+
+  // Backdate the marker past the grace period; the next tick's sweep
+  // reclaims the directory, and the sidecars go with it because they live
+  // inside it: the no-GC-code guarantee this test exists to pin.
+  await fs.writeFile(path.join(compactedDir, '.retired'), new Date(0).toISOString())
+  await maintainCache({ cacheRoot })
+  assert.equal(fsSync.existsSync(compactedDir), false, 'the retired generation and its sidecars are gone')
+})