From f49824b0d144ab81f5291e700aaf15135e7020f5 Mon Sep 17 00:00:00 2001 From: Brendan McMullen Date: Wed, 19 Aug 2026 10:38:26 -0700 Subject: [PATCH 1/3] The local grep service: one newest-first file walk, sidecar-or-scan per file (LLP 0265 T4) src/core/search/grep_service.js is the client half of LLP 0264: the same two-tier search the server runs, folded onto the client's single cache. executeGrepSearch flushes the dataset's pending spool first (the query seam's own freshness move, now exported from sql.js), walks every live data file newest message-day first, and serves each file through its hypgrep sidecar (parquetFind, index proposes, shared matcher confirms) when one exists beside it, by brute scan under the narrow SCAN_COLUMNS projection when not. Budget, truncated/exhausted, hit shape and sort order mirror the server byte for byte through the shared core/search modules. Two row gates the server does not need: - Purge: a raw file read does not apply Iceberg position deletes, so the walk carries each file's committed delete positions (new listLiveDataFiles export on the iceberg store) and filters both tiers by them; a stale sidecar cannot resurrect a purged row (LLP 0104). - Visibility: every surfaced row passes the LLP 0105 lattice check via cwdWithheldFromCaller, hoisted out of withLocalOnlyVisibility so the SQL path and this walk share one predicate instead of two copies. The check runs after the match, so withheldRows counts hits the caller was not allowed to see and costs no result budget. The shared types gain GrepSearchParams, the wire shape every serving surface accepts. Tests cover both tiers end to end on a real Iceberg cache: locators and snippets, newest-first truncation, from/to and session/chain scoping, the JSON tool_args column matching through cellText, local-only withholding at every caller rank, purge on both tiers, the sub-ngram literal query answering exactly through an index, and spool-captured rows surfacing after the service's own flush. Co-Authored-By: Claude Fable 5 --- src/core/cache/iceberg/store.js | 47 +++ src/core/query/sql.js | 9 +- src/core/query/visibility.js | 27 +- src/core/search/grep_service.js | 405 ++++++++++++++++++++++++++ src/core/search/types.d.ts | 17 ++ test/core/search-grep-service.test.js | 323 ++++++++++++++++++++ 6 files changed, 826 insertions(+), 2 deletions(-) create mode 100644 src/core/search/grep_service.js create mode 100644 test/core/search-grep-service.test.js diff --git a/src/core/cache/iceberg/store.js b/src/core/cache/iceberg/store.js index 5ee9a2e8..9ea90663 100644 --- a/src/core/cache/iceberg/store.js +++ b/src/core/cache/iceberg/store.js @@ -448,6 +448,53 @@ async function loadDeletedPositions(metadata, resolver, dataFileMap) { return out } +/** + * List the table's live data files with their identity-partition values and + * committed position-delete positions, for readers that walk files directly + * rather than scanning the table as one stream. The grep service is the + * consumer: its two tiers are per FILE (a sidecar-indexed file is searched + * through `parquetFind`, an unindexed one is brute-scanned), so it needs the + * file list, each file's partition `date` for the newest-first walk, and the + * deleted positions, because a raw file read does not apply position deletes + * and would otherwise resurrect purged rows (LLP 0104). + * + * Files whose manifest entry is deleted (status 2) or not parquet are + * excluded. An unreadable or snapshot-less table returns `[]`, matching the + * "empty table" degradation of `dataSourceForTable`. + * + * @param {string} tablePath + * @returns {Promise<{ filePath: string, partition: Record, recordCount: number, deletedPositions: Set | undefined }[]>} + */ +export async function listLiveDataFiles(tablePath) { + if (!tableExists(tablePath)) return [] + const { resolver, lister } = await getLocalIO() + const url = tableUrlForDir(tablePath) + /** @type {TableMetadata} */ + let metadata + try { + ({ metadata } = await loadLatestFileCatalogMetadata({ tableUrl: url, resolver, lister })) + } catch { + return [] + } + if (metadata['current-snapshot-id'] === undefined || !metadata.snapshots?.length) return [] + const dataFileMap = await findDataFileEntries(metadata, resolver) + if (dataFileMap.size === 0) return [] + const deleted = await loadDeletedPositions(metadata, resolver, dataFileMap) + /** @type {{ filePath: string, partition: Record, recordCount: number, deletedPositions: Set | undefined }[]} */ + const out = [] + for (const [filePath, { partition, entry }] of dataFileMap) { + const file = entry.data_file + if (String(file.file_format).toLowerCase() !== 'parquet') continue + out.push({ + filePath, + partition, + recordCount: Number(file.record_count ?? 0), + deletedPositions: deleted.get(filePath), + }) + } + return out +} + /** * Streaming counterpart to `readRowsFromTable`. Yields rows one at a * time so callers (in particular `QueryStorageService.readRows`) never diff --git a/src/core/query/sql.js b/src/core/query/sql.js index a1f7d470..96b11bba 100644 --- a/src/core/query/sql.js +++ b/src/core/query/sql.js @@ -483,6 +483,13 @@ export async function executeQuerySql(args) { } /** + * The query seam's freshness move: flush each referenced partition's + * pending spool (debounced under `auto`, forced under `always`, skipped + * under `never`) and report the staleness the debounce accepted. + * Exported so the grep service applies the identical policy; two read + * surfaces with different flush rules would answer differently about + * the same seconds-old row. + * * @param {{ * partitions: Array<{ tablePath?: string }>, * storage: ExtendedQueryStorageService, @@ -490,7 +497,7 @@ export async function executeQuerySql(args) { * messages: string[], * }} args */ -async function settlePendingCacheForQuery(args) { +export async function settlePendingCacheForQuery(args) { const now = Date.now() for (const partition of args.partitions) { if (!partition.tablePath) continue diff --git a/src/core/query/visibility.js b/src/core/query/visibility.js index 41057111..f54d254b 100644 --- a/src/core/query/visibility.js +++ b/src/core/query/visibility.js @@ -68,6 +68,31 @@ export function callerSeesEverything(callerRank) { return callerRank >= MAX_CLASS_RANK } +/** + * The one row-level withholding rule: a row whose `cwd` resolves to a class + * that outranks the caller's on the restrictiveness lattice is withheld. + * `withLocalOnlyVisibility` below applies it on the SQL read path; the grep + * service applies it per scanned row on its file walk (which cannot route + * through an `AsyncDataSource` wrapper). One exported predicate rather than + * two copies, so the two read surfaces cannot drift on what "local-only" + * hides. A cwd-less value returns false: those rows are `full`-class by + * construction on cwd-bearing datasets (see the wrapper's contract below). + * + * A corrupt machine-local list makes `resolve` throw + * (LocalOnlyListUnreadableError); callers let it propagate so the read + * fails loudly rather than silently resolving to "nothing withheld". + * + * @ref LLP 0105 [implements]: caller class >= row class on the lattice, the shared predicate form + * @param {UsagePolicyResolver} resolver + * @param {number} callerRank + * @param {unknown} cwd + * @returns {boolean} + */ +export function cwdWithheldFromCaller(resolver, callerRank, cwd) { + if (typeof cwd !== 'string' || cwd === '') return false + return CLASS_RANK[resolver.resolve(cwd).class] > callerRank +} + /** * Decorate one dataset's data source so every row the engine pulls honors * LLP 0105's invariant: content may only surface in a context at least as @@ -153,7 +178,7 @@ export function withLocalOnlyVisibility(source, opts) { // (LocalOnlyListUnreadableError); let it propagate so the query // fails loudly rather than silently resolving to "nothing // withheld", matching the export seam's fail-safe polarity. - if (CLASS_RANK[resolver.resolve(cwd).class] > callerRank) { + if (cwdWithheldFromCaller(resolver, callerRank, cwd)) { report.withheldRows += 1 continue } diff --git a/src/core/search/grep_service.js b/src/core/search/grep_service.js new file mode 100644 index 00000000..890a4785 --- /dev/null +++ b/src/core/search/grep_service.js @@ -0,0 +1,405 @@ +// @ts-check + +import fs from 'node:fs' + +import { parquetReadObjects } from 'hyparquet' +import { parquetFind } from 'hypgrep' + +import { createLocalIcebergIO, urlToPath } from '../cache/iceberg/resolver.js' +import { listLiveDataFiles } from '../cache/iceberg/store.js' +import { datasetForTablePath } from '../cache/paths.js' +import { discoverSpoolTables } from '../cache/spool.js' +import { resolveIcebergDir } from '../cache/storage.js' +import { Attr, getLogger, withSpan } from '../observability/index.js' +import { settlePendingCacheForQuery } from '../query/sql.js' +import { + callerSeesEverything, + cwdWithheldFromCaller, + defaultQueryVisibilityResolver, + resolveCallerClass, +} from '../query/visibility.js' +import { cellText, compileMatcher, makeSnippet, MAX_MATCH_COLUMNS } from './matcher.js' +import { SCAN_COLUMNS, SEARCHABLE_COLUMNS } from './searchable_columns.js' + +/** + * The local grep-search service: the client half of LLP 0264, mirroring the + * server's `src/search/grep-search.js` tier for tier. One walk over the + * cache's live data files, newest message-day first; a file with a hypgrep + * sidecar is searched through `parquetFind` (the index proposes candidate + * blocks, the shared matcher confirms), a file without one is brute-scanned + * under the narrow `SCAN_COLUMNS` projection. Files are processed strictly + * sequentially, so the request's memory bound is one data file plus its + * index, never a day's worth. No sidecar anywhere (the tree before T6 of + * LLP 0265 runs) means every file takes the scan tier: slower, never wrong. + * + * Unlike the server there is no cross-tier day exclusion: a client row lives + * in exactly one data file, and each file is served by exactly one tier, so + * a row cannot be counted twice by construction. + * + * Two client-side row gates the server does not have: + * + * - **Purge.** A raw file read does not apply Iceberg position deletes, so + * every tier filters rows through the file's committed delete positions + * (`listLiveDataFiles`); a purged row can neither match nor surface, even + * when a stale sidecar still proposes it (LLP 0104). + * - **Visibility.** Every surfaced row passes the LLP 0105 lattice check via + * the same `cwdWithheldFromCaller` predicate the SQL read path applies. + * The check runs AFTER the match predicate, so `localOnly.withheldRows` + * counts hits the caller was not allowed to see - the number the verb + * renders as actionable guidance - and an out-of-rank row consumes no + * result budget. + * + * @ref LLP 0264#decision [implements]: the client mirror of the server's two-tier grep, cache scan beside sidecar-indexed files + * @ref LLP 0264#visibility [implements]: the local scan enforces LLP 0105 with the caller's cwd; --remote inherits the server's own gate instead + * + * @import { ExtendedQueryStorageService } from '../../../src/core/cache/types.js' + * @import { GrepSearchHit, GrepSearchMatcher, GrepSearchParams, GrepSearchResult } from '../../../src/core/search/types.js' + * @import { LocalOnlyVisibilityReport, RefreshMode } from '../../../src/core/query/types.js' + * @import { UsagePolicyResolver } from '../../../src/core/usage-policy/types.js' + */ + +const DATASET = 'ai_gateway_messages' + +/** + * Rows between abort checks inside one brute-scanned file. The deadline has + * to be able to land in the middle of a file, not only between files: a + * compacted file holds many sessions' rows, and the per-row predicate is + * where a large scan actually spends its wall clock. + */ +const ABORT_CHECK_ROWS = 256 + +/** + * A file whose partition day could not be decoded sorts as newest and is + * never day-pruned: pruning must prove a file out of the window before + * skipping it, and walking it early keeps the "newest first" promise + * conservative rather than wrong. + */ +const UNKNOWN_DAY_SORT_KEY = '￿' + +/** + * Run one grep search over the local cache. + * + * `params` is the shared wire shape (`GrepSearchParams`); the rest is the + * client seam: the storage service for discovery and spool freshness, the + * LLP 0105 caller identity, and the abort signal. The result extends the + * shared `GrepSearchResult` with the local-only visibility report, the + * freshness messages the spool debounce produced, and per-tier file counts + * so surfaces (and smokes) can prove which path served the answer. + * + * @param {GrepSearchParams & { + * storage: ExtendedQueryStorageService, + * includeLocalOnly?: boolean, + * callerCwd?: string | null, + * usagePolicyResolver?: UsagePolicyResolver, + * refresh?: RefreshMode, + * signal?: AbortSignal, + * }} args + * @returns {Promise} + */ +export async function executeGrepSearch(args) { + const { storage, signal } = args + const limit = args.limit + // Collect one past the limit: the overflow row is the proof that + // `truncated` is true, and is never returned. + const budget = limit + 1 + const matcher = compileMatcher(args.query, args.regex === true) + const chainPred = compileChainPredicate(args) + const rowFrom = args.from + const rowTo = args.to + /** @param {Record} row */ + const dayPred = (row) => { + const day = typeof row.date === 'string' ? row.date.slice(0, 10) : null + if (day === null) return rowFrom === undefined && rowTo === undefined + if (rowFrom !== undefined && day < rowFrom) return false + if (rowTo !== undefined && day > rowTo) return false + return true + } + /** @param {Record} row */ + const accept = (row) => chainPred(row) && dayPred(row) && matcher.rowTest(row) + + /** @type {LocalOnlyVisibilityReport} */ + const localOnly = { callerClass: 'unknown', filtered: false, withheldRows: 0, suppressedRows: 0 } + /** @type {((row: Record) => boolean) | null} */ + let withheld = null + if (args.includeLocalOnly !== true) { + const resolver = args.usagePolicyResolver ?? defaultQueryVisibilityResolver(storage) + const { callerClass, callerRank } = resolveCallerClass(resolver, args.callerCwd) + localOnly.callerClass = callerClass + if (!callerSeesEverything(callerRank)) { + localOnly.filtered = true + withheld = (row) => cwdWithheldFromCaller(resolver, callerRank, row.cwd) + } + } + + return withSpan( + 'query.grep_search', + { + [Attr.COMPONENT]: 'query', + [Attr.OPERATION]: 'query.grep_search', + [Attr.DATASET]: DATASET, + // The pattern itself is user search text and may name a secret; + // record its shape, never its content. + query_length: args.query.length, + regex_mode: args.regex === true, + status: 'ok', + }, + async (span) => { + // The settle list is spool tables PLUS committed partitions: the + // gateway's live rows spool under a label table (proxy_messages_v5) + // that has no cursor until its first flush, so partition discovery + // alone would never flush - and never find - a row captured seconds + // ago. The SQL seam reaches those tables through the dataset's own + // discoverParts; this service enumerates them from the spool itself + // to the same effect. + /** @type {{ tablePath: string }[]} */ + const settleTargets = [] + try { + for (const tablePath of await discoverSpoolTables(storage.cacheRoot)) { + if (datasetForTablePath(storage.cacheRoot, tablePath) === DATASET) settleTargets.push({ tablePath }) + } + } catch { + // An unreadable spool root means nothing is pending to flush. + } + for (const p of await storage.discoverCachePartitions({ datasets: [DATASET] })) { + settleTargets.push({ tablePath: p.path }) + } + /** @type {string[]} */ + const freshnessMessages = [] + await settlePendingCacheForQuery({ + partitions: settleTargets, + storage, + refresh: args.refresh ?? 'auto', + messages: freshnessMessages, + }) + // Re-discover after the flush: a first flush mints the source + // partition directories the walk below reads (the same re-discovery + // the dataset's createDataSource performs on the SQL path). + const partitions = await storage.discoverCachePartitions({ datasets: [DATASET] }) + + /** @type {{ filePath: string, day: string | null, deletedPositions: Set | undefined }[]} */ + const files = [] + for (const partition of partitions) { + for (const file of await listLiveDataFiles(resolveIcebergDir(partition.path))) { + files.push({ + filePath: file.filePath, + day: toDayString(file.partition.date), + deletedPositions: file.deletedPositions, + }) + } + } + // Newest message-day first, across every source partition at once, so + // a truncated answer keeps the newest matches whichever client wrote + // them (the server's walk order, applied to the client's layout). + files.sort((a, b) => ((a.day ?? UNKNOWN_DAY_SORT_KEY) < (b.day ?? UNKNOWN_DAY_SORT_KEY) ? 1 : -1)) + + const { resolver: io } = await createLocalIcebergIO() + /** @type {GrepSearchHit[]} */ + const hits = [] + let exhausted = true + let indexedFiles = 0 + let scannedFiles = 0 + + /** @param {{ filePath: string, deletedPositions: Set | undefined }} file */ + const searchFile = async (file) => { + // Sidecar existence IS the index marker, no ledger (LLP 0264 + // #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') + /** @type {Awaited> | null} */ + let indexFile = null + if (fs.existsSync(urlToPath(sidecarUrl))) { + try { + indexFile = await io.reader(sidecarUrl) + } catch (err) { + if (/** @type {Error & { code?: string }} */ (err)?.code !== 'ENOENT') throw err + } + } + if (indexFile) { + indexedFiles += 1 + // 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) { + if (file.deletedPositions?.has(BigInt(/** @type {number} */ (row.__index__)))) continue + if (withheld?.(row)) { + localOnly.withheldRows += 1 + continue + } + hits.push(toHit(row, matcher)) + if (hits.length >= budget) return + } + return + } + scannedFiles += 1 + const sourceFile = await io.reader(file.filePath) + const rows = await parquetReadObjects({ file: sourceFile, columns: SCAN_COLUMNS }) + for (let i = 0; i < rows.length; i++) { + if (i % ABORT_CHECK_ROWS === 0) signal?.throwIfAborted() + if (hits.length >= budget) return + if (file.deletedPositions?.has(BigInt(i))) continue + const row = rows[i] + if (!accept(row)) continue + if (withheld?.(row)) { + localOnly.withheldRows += 1 + continue + } + hits.push(toHit(row, matcher)) + } + } + + try { + for (const file of files) { + signal?.throwIfAborted() + if (hits.length >= budget) { + exhausted = false + break + } + if (file.day !== null + && ((rowFrom !== undefined && file.day < rowFrom) || (rowTo !== undefined && file.day > rowTo))) { + continue + } + await searchFile(file) + } + } catch (err) { + // The caller aborting mid-walk keeps what was found: a partial + // answer marked not exhausted, never an error. + if (!isAbort(err)) throw err + exhausted = false + } + + const truncated = hits.length > limit + if (truncated) hits.length = limit + sortHits(hits) + + span.setAttribute('file_count', files.length) + span.setAttribute('indexed_file_count', indexedFiles) + span.setAttribute('scanned_file_count', scannedFiles) + span.setAttribute('hit_count', hits.length) + span.setAttribute('truncated', truncated) + span.setAttribute('caller_usage_class', localOnly.callerClass) + span.setAttribute('local_only_withheld_rows', localOnly.withheldRows) + // Counts only, never content or raw paths, matching the SQL seam's + // `usage_policy.query_withhold` discipline (LLP 0080 #telemetry). + if (localOnly.withheldRows > 0) { + getLogger('query').debug('usage_policy.query_withhold', { + [Attr.COMPONENT]: 'query', + caller_usage_class: localOnly.callerClass, + withheld_row_count: localOnly.withheldRows, + suppressed_row_count: 0, + }) + } + + return { + hits, + truncated, + exhausted: exhausted && !truncated, + localOnly, + freshnessMessages, + indexedFiles, + scannedFiles, + } + }, + { component: 'query' } + ) +} + +/** + * Newest first: intrinsic day, then creation time, then part id for a + * stable order within a message. Identical to the server's ordering, so a + * local answer and a `--remote` answer to the same query read the same. + * + * @param {GrepSearchHit[]} hits + */ +function sortHits(hits) { + hits.sort((a, b) => { + if (a.date !== b.date) return a.date < b.date ? 1 : -1 + const at = a.messageCreatedAt ?? '' + const bt = b.messageCreatedAt ?? '' + if (at !== bt) return at < bt ? 1 : -1 + return (a.partId ?? '') < (b.partId ?? '') ? 1 : -1 + }) +} + +/** + * 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. + * + * @param {Record} row + * @param {GrepSearchMatcher} matcher + * @returns {GrepSearchHit} + */ +function toHit(row, matcher) { + /** @type {{ column: string, snippet: string }[]} */ + const matches = [] + for (const column of SEARCHABLE_COLUMNS) { + const text = cellText(row[column]) + if (text === '' || !matcher.test(text)) continue + matches.push({ column, snippet: makeSnippet(text, matcher) }) + if (matches.length >= MAX_MATCH_COLUMNS) break + } + return { + date: typeof row.date === 'string' ? row.date.slice(0, 10) : '', + sessionId: typeof row.session_id === 'string' ? row.session_id : '', + agentId: stringOrNull(row.agent_id), + conversationId: stringOrNull(row.conversation_id), + partId: stringOrNull(row.part_id), + messageId: stringOrNull(row.message_id), + messageCreatedAt: stringOrNull(row.message_created_at), + matches, + } +} + +/** + * @param {GrepSearchParams} params + * @returns {(row: Record) => boolean} + */ +function compileChainPredicate(params) { + const { sessionId, chainId } = params + if (sessionId === undefined) return () => true + return (row) => { + if (row.session_id !== sessionId) return false + if (chainId === undefined) return true + // A chain id names either side of the pair, the same matching rule the + // server applies (its LLP 0117 locator query). + return row.agent_id === chainId || row.conversation_id === chainId + } +} + +/** @param {unknown} value */ +function stringOrNull(value) { + if (value === null || value === undefined) return null + if (value instanceof Date) return value.toISOString() + return String(value) +} + +/** A partition or timestamp value as a YYYY-MM-DD day, however it materializes. */ +/** @param {unknown} value */ +function toDayString(value) { + if (value instanceof Date) return value.toISOString().slice(0, 10) + if (typeof value === 'string' && value.length >= 10) return value.slice(0, 10) + return null +} + +/** @param {unknown} err */ +function isAbort(err) { + return err instanceof Error && err.name === 'AbortError' +} diff --git a/src/core/search/types.d.ts b/src/core/search/types.d.ts index 0452ba58..e33ee6bc 100644 --- a/src/core/search/types.d.ts +++ b/src/core/search/types.d.ts @@ -3,6 +3,23 @@ * and `--remote` answer in the same shape (LLP 0264 #shared). */ +/** + * The caller-supplied search parameters, identical on every serving + * surface: `hyp query grep` locally, the `grep_search` MCP tool, and the + * same tool spoken to a server through `--remote`. Server-side wrappers + * extend this with their own routing fields (`org`); those never ride + * the wire from a client. + */ +export interface GrepSearchParams { + query: string + regex?: boolean + sessionId?: string + chainId?: string + from?: string + to?: string + limit: number +} + /** One matching message row, projected to locators plus bounded snippets. */ export interface GrepSearchHit { date: string diff --git a/test/core/search-grep-service.test.js b/test/core/search-grep-service.test.js new file mode 100644 index 00000000..5b2bb448 --- /dev/null +++ b/test/core/search-grep-service.test.js @@ -0,0 +1,323 @@ +// @ts-check + +import test from 'node:test' +import assert from 'node:assert/strict' +import fs from 'node:fs/promises' +import os from 'node:os' +import path from 'node:path' + +import { ByteWriter } from 'hyparquet-writer' +import { createIndex } from 'hypgrep' + +import { urlToPath } from '../../src/core/cache/iceberg/resolver.js' +import { deleteMatchingRows, listLiveDataFiles } from '../../src/core/cache/iceberg/store.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 { aiGatewayDatasetRegistration } from '../../hypaware-core/plugins-workspace/ai-gateway/src/dataset.js' + +/** + * @import { ColumnSpec } from '../../hypaware-plugin-kernel-types.js' + * @import { UsagePolicyResolver } from '../../src/core/usage-policy/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: 'model', type: 'STRING', nullable: true }, + { name: 'cwd', type: 'STRING', nullable: true }, + { name: 'git_branch', type: 'STRING', nullable: true }, + { name: 'git_remote', type: 'STRING', nullable: true }, + { name: 'tool_name', type: 'STRING', nullable: true }, + { name: 'tool_args', type: 'JSON', 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 + +/** + * A gateway-shaped row; `message_created_at` increases with insertion order + * within a day so newest-first assertions are deterministic. + * + * @param {Record} [over] + * @returns {Record} + */ +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, + model: 'claude-fable-5', + cwd: '/home/open-proj', + git_branch: null, + git_remote: null, + tool_name: null, + tool_args: null, + 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, + } +} + +/** + * Build a real single-source cache: each batch is one Iceberg append, so + * rows with distinct identity-partition tuples land in distinct data files. + * + * @param {Record[][]} batches + */ +async function makeCache(batches) { + const cacheRoot = await fs.mkdtemp(path.join(os.tmpdir(), 'hyp-grep-service-')) + const declaration = aiGatewayDatasetRegistration().cachePartitioning + for (const batch of batches) { + await appendRowsToSourceTable(cacheRoot, DATASET, ['source=test'], COLUMNS, batch, { declaration }) + } + const storage = createQueryStorageService({ + cacheRoot, + getDeclaration: (dataset) => (dataset === DATASET ? declaration : undefined), + }) + return { cacheRoot, storage, tableDir: () => resolveIcebergDir(path.join(cacheRoot, 'datasets', DATASET, 'source=test')) } +} + +/** + * @param {ReturnType} storage + * @param {Record} [over] + */ +function grep(storage, over = {}) { + return executeGrepSearch(/** @type {any} */ ({ + storage, + query: 'needle', + limit: 10, + includeLocalOnly: true, + ...over, + })) +} + +/** + * Build a hypgrep sidecar beside every live data file of the table, the + * shape T6's maintenance pass will produce. + * + * @param {string} tableDir + * @returns {Promise} how many sidecars were written + */ +async function buildSidecars(tableDir) { + const files = await listLiveDataFiles(tableDir) + for (const file of files) { + const sourcePath = urlToPath(file.filePath) + const bytes = await fs.readFile(sourcePath) + const buffer = bytes.buffer.slice(bytes.byteOffset, bytes.byteOffset + bytes.byteLength) + const sourceFile = { + byteLength: buffer.byteLength, + /** @param {number} start @param {number} [end] */ + slice: (start, end) => buffer.slice(start, end), + } + const writer = new ByteWriter() + await createIndex({ sourceFile, indexFile: writer }) + await fs.writeFile(sourcePath.replace(/\.parquet$/, '.index.parquet'), Buffer.from(writer.getBuffer())) + } + return files.length +} + +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', + conversation_id: 'c2', + agent_id: 'a2', + content_text: 'the needle two', +}) + +test('scan tier: hits carry locators and snippets, newest day first', async () => { + const { storage } = await makeCache([[OLD], [NEW]]) + const res = await grep(storage) + assert.equal(res.hits.length, 2) + assert.equal(res.truncated, false) + assert.equal(res.exhausted, true) + assert.equal(res.indexedFiles, 0) + assert.ok(res.scannedFiles >= 2, 'both files took the scan tier') + const [first, second] = res.hits + assert.equal(first.date, '2026-08-12') + assert.equal(first.sessionId, 's2') + assert.equal(first.conversationId, 'c2') + assert.equal(first.agentId, 'a2') + assert.equal(typeof first.messageId, 'string') + assert.equal(typeof first.partId, 'string') + assert.ok(first.messageCreatedAt, 'creation time surfaces on the hit') + assert.equal(first.matches[0].column, 'content_text') + assert.match(first.matches[0].snippet, /needle/) + assert.equal(second.date, '2026-08-10') +}) + +test('scan tier: the limit truncates to the newest matches', async () => { + const { storage } = await makeCache([[OLD], [NEW]]) + const res = await grep(storage, { limit: 1 }) + assert.equal(res.hits.length, 1) + assert.equal(res.hits[0].date, '2026-08-12', 'the newest match survives truncation') + assert.equal(res.truncated, true) + assert.equal(res.exhausted, false) +}) + +test('scan tier: from/to narrow by day at the file walk', async () => { + const { storage } = await makeCache([[OLD], [NEW]]) + const fromOnly = await grep(storage, { from: '2026-08-11' }) + assert.deepEqual(fromOnly.hits.map((h) => h.date), ['2026-08-12']) + const toOnly = await grep(storage, { to: '2026-08-11' }) + assert.deepEqual(toOnly.hits.map((h) => h.date), ['2026-08-10']) +}) + +test('scan tier: session and chain predicates scope the walk', async () => { + const { storage } = await makeCache([[OLD], [NEW]]) + const bySession = await grep(storage, { sessionId: 's1' }) + assert.deepEqual(bySession.hits.map((h) => h.sessionId), ['s1']) + const byChain = await grep(storage, { sessionId: 's2', chainId: 'a2' }) + assert.deepEqual(byChain.hits.map((h) => h.sessionId), ['s2']) + const wrongChain = await grep(storage, { sessionId: 's2', chainId: 'zz' }) + assert.equal(wrongChain.hits.length, 0) +}) + +test('literal matching is case-insensitive; regex mode is operator-shaped', async () => { + const { storage } = await makeCache([[OLD], [NEW]]) + const upper = await grep(storage, { query: 'NEEDLE' }) + assert.equal(upper.hits.length, 2) + const rx = await grep(storage, { query: 'ne+dle t.o', regex: true }) + assert.deepEqual(rx.hits.map((h) => h.sessionId), ['s2']) +}) + +test('the JSON column matches through cellText, and reports as the matched column', async () => { + 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/) +}) + +test('local-only rows are withheld from lower-rank callers, and only from them', async () => { + const openRow = mkRow({ date: '2026-08-10', session_id: 'open', cwd: '/home/open-proj', content_text: 'needle open' }) + const privateRow = mkRow({ date: '2026-08-12', session_id: 'priv', cwd: '/home/private-proj', content_text: 'needle private' }) + const { storage } = await makeCache([[openRow], [privateRow]]) + /** @type {UsagePolicyResolver} */ + const resolver = { + resolve: (cwd) => ({ + class: cwd.includes('private') ? 'local-only' : 'full', + governedBy: null, + declared: null, + }), + isIgnored: () => false, + } + + const fullCaller = await grep(storage, { + includeLocalOnly: false, callerCwd: '/home/open-proj', usagePolicyResolver: resolver, + }) + assert.deepEqual(fullCaller.hits.map((h) => h.sessionId), ['open']) + assert.equal(fullCaller.localOnly.callerClass, 'full') + assert.equal(fullCaller.localOnly.filtered, true) + assert.equal(fullCaller.localOnly.withheldRows, 1) + + const noCwdCaller = await grep(storage, { includeLocalOnly: false, usagePolicyResolver: resolver }) + assert.deepEqual(noCwdCaller.hits.map((h) => h.sessionId), ['open'], 'no derivable cwd fails closed') + assert.equal(noCwdCaller.localOnly.callerClass, 'unknown') + + const localOnlyCaller = await grep(storage, { + includeLocalOnly: false, callerCwd: '/home/private-proj', usagePolicyResolver: resolver, + }) + assert.equal(localOnlyCaller.hits.length, 2, 'an equal-rank caller sees the local-only row') + assert.equal(localOnlyCaller.localOnly.withheldRows, 0) + + const withOverride = await grep(storage, { + includeLocalOnly: true, callerCwd: '/home/open-proj', usagePolicyResolver: resolver, + }) + assert.equal(withOverride.hits.length, 2, 'the override surfaces every row') + assert.equal(withOverride.localOnly.filtered, false) +}) + +test('a purged row cannot surface from the scan tier', async () => { + const { storage, tableDir } = await makeCache([[OLD], [NEW]]) + const deleted = await deleteMatchingRows( + tableDir(), + (row) => row.session_id === 's2', + { columns: ['session_id'] } + ) + assert.equal(deleted.rowsDeleted, 1) + const res = await grep(storage) + assert.deepEqual(res.hits.map((h) => h.sessionId), ['s1']) +}) + +test('indexed tier: sidecars serve every file with identical hits', async () => { + const { storage, tableDir } = await makeCache([[OLD], [NEW]]) + const before = await grep(storage) + const sidecars = await buildSidecars(tableDir()) + assert.ok(sidecars >= 2) + const res = await grep(storage) + assert.equal(res.indexedFiles, sidecars, 'every file was served through its sidecar') + assert.equal(res.scannedFiles, 0) + assert.deepEqual(res.hits, before.hits, 'the two tiers answer identically') +}) + +test('indexed tier: a query below the ngram length still answers exactly', async () => { + const { storage, tableDir } = await makeCache([[OLD], [NEW]]) + await buildSidecars(tableDir()) + // 'dle' is shorter than hypgrep's default ngram, so the index proposes + // every block and the shared matcher does the real work: slower, never + // wrong (the LLP 0265 T7 "literal cliff" is performance, not truth). + const res = await grep(storage, { query: 'dle' }) + assert.equal(res.hits.length, 2) + assert.equal(res.indexedFiles >= 2, true) +}) + +test('indexed tier: a stale sidecar cannot resurrect a purged row', async () => { + const { storage, tableDir } = await makeCache([[OLD], [NEW]]) + await buildSidecars(tableDir()) + await deleteMatchingRows(tableDir(), (row) => row.session_id === 's2', { columns: ['session_id'] }) + const res = await grep(storage) + assert.ok(res.indexedFiles >= 1, 'the walk still ran through the sidecars') + assert.deepEqual(res.hits.map((h) => h.sessionId), ['s1'], 'the purged row is filtered by position') +}) + +test('rows captured into the spool are found after the freshness flush', async () => { + const { storage } = await makeCache([[OLD]]) + const spooled = mkRow({ date: '2026-08-13', session_id: 'spooled', content_text: 'fresh needle from the spool' }) + const labelTable = storage.cacheTablePath(DATASET, ['proxy_messages_v5']) + await storage.appendRows(labelTable, COLUMNS, [spooled]) + const res = await grep(storage) + assert.ok( + res.hits.some((h) => h.sessionId === 'spooled'), + 'the spool was flushed by the search itself' + ) +}) + +test('an empty cache answers empty and exhausted', async () => { + const cacheRoot = await fs.mkdtemp(path.join(os.tmpdir(), 'hyp-grep-empty-')) + const storage = createQueryStorageService({ cacheRoot }) + const res = await grep(storage) + assert.deepEqual(res.hits, []) + assert.equal(res.truncated, false) + assert.equal(res.exhausted, true) +}) + +test('an empty or oversized query refuses up front', async () => { + const cacheRoot = await fs.mkdtemp(path.join(os.tmpdir(), 'hyp-grep-refuse-')) + const storage = createQueryStorageService({ cacheRoot }) + await assert.rejects(() => grep(storage, { query: '' }), /non-empty/) + await assert.rejects(() => grep(storage, { query: 'x'.repeat(2000) }), /at most/) + await assert.rejects(() => grep(storage, { query: '(', regex: true }), /not a valid regular expression/) +}) From 8b694e5bd89ab2c58559f042b687e0f8c690a679 Mon Sep 17 00:00:00 2001 From: test Date: Thu, 20 Aug 2026 15:30:20 +0000 Subject: [PATCH 2/3] Grep search: truncate in sort order, honor a chainId alone, and stop failing open Review fixes on the T4 grep service (LLP 0265): - The limit returned the OLDEST matches. `hits.length = limit` ran before `sortHits`, so the cut fell in walk order; rows inside one data file run oldest to newest, so `--limit 3` over today's session answered the first three matches, not the newest three. Hits are now trimmed in sort order (amortized, so the buffer stays bounded), and the file walk stops only once every file still ahead is strictly older than the oldest kept hit. - `AbortSignal.timeout` threw out of the service. Its reason is a DOMException named `TimeoutError`, which the `AbortError` name check rejected, so the documented "partial answer, marked not exhausted" became an error for the deadline shape the signal exists to carry. - `chainId` without `sessionId` was silently discarded, answering across every session instead of the chain's. - `limit` is validated beside the query: an absent one made the budget NaN and walked the whole cache, a negative one threw a bare RangeError. - `listLiveDataFiles` swallowed a metadata load failure and answered `[]`, so a corrupt table made grep report zero hits where `hyp query sql` raises. It now propagates, matching `dataSourceForTable`. - `sortHits`' tiebreak returns 0 for equal keys, now that the buffer is sorted repeatedly. Five regression tests, each verified to fail on the pre-fix tree. Co-Authored-By: Claude Opus 5 (1M context) --- src/core/cache/iceberg/store.js | 17 +++--- src/core/search/grep_service.js | 86 ++++++++++++++++++++++----- test/core/search-grep-service.test.js | 59 ++++++++++++++++++ 3 files changed, 139 insertions(+), 23 deletions(-) diff --git a/src/core/cache/iceberg/store.js b/src/core/cache/iceberg/store.js index 9ea90663..23d45a30 100644 --- a/src/core/cache/iceberg/store.js +++ b/src/core/cache/iceberg/store.js @@ -459,8 +459,13 @@ async function loadDeletedPositions(metadata, resolver, dataFileMap) { * and would otherwise resurrect purged rows (LLP 0104). * * Files whose manifest entry is deleted (status 2) or not parquet are - * excluded. An unreadable or snapshot-less table returns `[]`, matching the - * "empty table" degradation of `dataSourceForTable`. + * excluded. A missing or snapshot-less table returns `[]`, matching the + * "empty table" degradation of `dataSourceForTable`. A metadata load that + * FAILS propagates, for the same reason: unreadable table metadata means an + * unknown row set, and a reader that swallows it reports "no matches" over a + * partition it never read. `dataSourceForTable` lets that error out, so + * `hyp query sql` fails loudly; grep search must not answer zero where SQL + * raises. * * @param {string} tablePath * @returns {Promise<{ filePath: string, partition: Record, recordCount: number, deletedPositions: Set | undefined }[]>} @@ -469,13 +474,7 @@ export async function listLiveDataFiles(tablePath) { if (!tableExists(tablePath)) return [] const { resolver, lister } = await getLocalIO() const url = tableUrlForDir(tablePath) - /** @type {TableMetadata} */ - let metadata - try { - ({ metadata } = await loadLatestFileCatalogMetadata({ tableUrl: url, resolver, lister })) - } catch { - return [] - } + const { metadata } = await loadLatestFileCatalogMetadata({ tableUrl: url, resolver, lister }) if (metadata['current-snapshot-id'] === undefined || !metadata.snapshots?.length) return [] const dataFileMap = await findDataFileEntries(metadata, resolver) if (dataFileMap.size === 0) return [] diff --git a/src/core/search/grep_service.js b/src/core/search/grep_service.js index 890a4785..5010a009 100644 --- a/src/core/search/grep_service.js +++ b/src/core/search/grep_service.js @@ -104,7 +104,16 @@ const UNKNOWN_DAY_SORT_KEY = ' export async function executeGrepSearch(args) { const { storage, signal } = args const limit = args.limit - // Collect one past the limit: the overflow row is the proof that + // `limit` is validated here for the same reason the query is: this is the + // wire shape a serving surface hands straight through, so an unchecked + // value fails late and wrong instead of up front. An absent limit makes + // the budget NaN, so the walk never stops and collects every match in the + // cache; a negative one reaches the result trim and throws a bare + // `RangeError: Invalid array length` from deep inside the service. + if (!Number.isSafeInteger(limit) || limit <= 0) { + throw new Error('limit must be a positive integer') + } + // Collect one past the limit: the overflow hit is the proof that // `truncated` is true, and is never returned. const budget = limit + 1 const matcher = compileMatcher(args.query, args.regex === true) @@ -204,6 +213,27 @@ export async function executeGrepSearch(args) { let indexedFiles = 0 let scannedFiles = 0 + /** + * Keep the newest `budget` hits and drop the rest. Truncation has to + * happen in SORT order, never in walk order: rows inside one data file + * are in write order (LLP 0022 clusters a file by session, so a + * session's rows run oldest to newest), and one message-day is many + * files, so cutting the tail of the walk keeps the OLDEST matches of + * whichever file first filled the budget, the exact opposite of what + * 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. + */ + const trimHits = () => { + sortHits(hits) + if (hits.length > budget) hits.length = budget + } + /** @param {Record} row */ + const collect = (row) => { + hits.push(toHit(row, matcher)) + if (hits.length >= budget * 2) trimHits() + } + /** @param {{ filePath: string, deletedPositions: Set | undefined }} file */ const searchFile = async (file) => { // Sidecar existence IS the index marker, no ledger (LLP 0264 @@ -240,8 +270,7 @@ export async function executeGrepSearch(args) { localOnly.withheldRows += 1 continue } - hits.push(toHit(row, matcher)) - if (hits.length >= budget) return + collect(row) } return } @@ -250,7 +279,6 @@ export async function executeGrepSearch(args) { const rows = await parquetReadObjects({ file: sourceFile, columns: SCAN_COLUMNS }) for (let i = 0; i < rows.length; i++) { if (i % ABORT_CHECK_ROWS === 0) signal?.throwIfAborted() - if (hits.length >= budget) return if (file.deletedPositions?.has(BigInt(i))) continue const row = rows[i] if (!accept(row)) continue @@ -258,7 +286,7 @@ export async function executeGrepSearch(args) { localOnly.withheldRows += 1 continue } - hits.push(toHit(row, matcher)) + collect(row) } } @@ -266,8 +294,17 @@ export async function executeGrepSearch(args) { for (const file of files) { signal?.throwIfAborted() if (hits.length >= budget) { - exhausted = false - break + trimHits() + // Files are walked day-descending, so every file still ahead + // holds rows no newer than this one's day. Once the budget is + // full of hits strictly newer than that day, nothing left in the + // walk can displace one and the walk stops. A same-day file (or + // one whose day would not decode) is still read, because its + // rows can outrank a kept hit. + if (file.day !== null && file.day < hits[hits.length - 1].date) { + exhausted = false + break + } } if (file.day !== null && ((rowFrom !== undefined && file.day < rowFrom) || (rowTo !== undefined && file.day > rowTo))) { @@ -278,13 +315,13 @@ export async function executeGrepSearch(args) { } catch (err) { // The caller aborting mid-walk keeps what was found: a partial // answer marked not exhausted, never an error. - if (!isAbort(err)) throw err + if (!isAbort(err, signal)) throw err exhausted = false } + trimHits() const truncated = hits.length > limit if (truncated) hits.length = limit - sortHits(hits) span.setAttribute('file_count', files.length) span.setAttribute('indexed_file_count', indexedFiles) @@ -331,7 +368,13 @@ function sortHits(hits) { const at = a.messageCreatedAt ?? '' const bt = b.messageCreatedAt ?? '' if (at !== bt) return at < bt ? 1 : -1 - return (a.partId ?? '') < (b.partId ?? '') ? 1 : -1 + const ap = a.partId ?? '' + const bp = b.partId ?? '' + // Equal keys compare equal: the walk trims in sort order and so sorts + // the buffer repeatedly, and a comparator that never returns 0 would + // reshuffle indistinguishable hits on every pass. + if (ap === bp) return 0 + return ap < bp ? 1 : -1 }) } @@ -374,9 +417,9 @@ function toHit(row, matcher) { */ function compileChainPredicate(params) { const { sessionId, chainId } = params - if (sessionId === undefined) return () => true + if (sessionId === undefined && chainId === undefined) return () => true return (row) => { - if (row.session_id !== sessionId) return false + if (sessionId !== undefined && row.session_id !== sessionId) return false if (chainId === undefined) return true // A chain id names either side of the pair, the same matching rule the // server applies (its LLP 0117 locator query). @@ -399,7 +442,22 @@ function toDayString(value) { return null } -/** @param {unknown} err */ -function isAbort(err) { +/** + * Did this error come from the caller's deadline rather than from the walk? + * `throwIfAborted` rethrows `signal.reason` verbatim, and the natural + * deadline (`AbortSignal.timeout`) makes that reason a `DOMException` named + * `TimeoutError`, not `AbortError`, so a name check alone turns the + * documented "partial answer, marked not exhausted" into a thrown error for + * the one abort shape the service exists to serve. Identity against the + * signal's own reason accepts every abort shape, a caller's custom + * `abort(reason)` included, without swallowing an unrelated failure that + * happens to race the deadline. + * + * @param {unknown} err + * @param {AbortSignal | undefined} signal + * @returns {boolean} + */ +function isAbort(err, signal) { + if (signal?.aborted === true && err === signal.reason) return true return err instanceof Error && err.name === 'AbortError' } diff --git a/test/core/search-grep-service.test.js b/test/core/search-grep-service.test.js index 5b2bb448..c1c1216e 100644 --- a/test/core/search-grep-service.test.js +++ b/test/core/search-grep-service.test.js @@ -171,6 +171,47 @@ test('scan tier: the limit truncates to the newest matches', async () => { assert.equal(res.exhausted, false) }) +test('the limit keeps the newest matches inside one file, not the first ones walked', async () => { + // One append, so all ten rows share a data file and the newest-day file + // walk cannot order them: only sort-order truncation can. Rows land in + // insertion order, which is oldest first, so a walk-order cut would answer + // m1..m3. + const batch = [] + for (let i = 0; i < 10; i++) batch.push(mkRow({ date: '2026-08-14', content_text: `needle body ${i}` })) + const { storage } = await makeCache([batch]) + const all = await grep(storage, { limit: 100 }) + assert.equal(all.hits.length, 10) + const newest = all.hits.slice(0, 3).map((h) => h.messageId) + const capped = await grep(storage, { limit: 3 }) + assert.deepEqual(capped.hits.map((h) => h.messageId), newest, 'the newest three survive the limit') + assert.equal(capped.truncated, true) + assert.equal(capped.exhausted, false) +}) + +test('a chain id alone scopes the walk, with no session id beside it', async () => { + const { storage } = await makeCache([[OLD], [NEW]]) + const byChain = await grep(storage, { chainId: 'a2' }) + assert.deepEqual(byChain.hits.map((h) => h.sessionId), ['s2']) + const byConversation = await grep(storage, { chainId: 'c2' }) + assert.deepEqual(byConversation.hits.map((h) => h.sessionId), ['s2']) + const unknownChain = await grep(storage, { chainId: 'zz' }) + assert.equal(unknownChain.hits.length, 0, 'an unmatched chain id filters, it does not fall open') +}) + +test('a deadline signal returns the partial answer rather than throwing', async () => { + const { storage } = await makeCache([[OLD], [NEW]]) + // AbortSignal.timeout's reason is a DOMException named TimeoutError, not + // AbortError: the deadline shape the service is built for. + const deadline = AbortSignal.timeout(1) + await new Promise((resolve) => setTimeout(resolve, 10)) + const res = await grep(storage, { signal: deadline }) + assert.equal(res.exhausted, false, 'an aborted walk is not exhausted') + const controller = new AbortController() + controller.abort() + const plain = await grep(storage, { signal: controller.signal }) + assert.equal(plain.exhausted, false) +}) + test('scan tier: from/to narrow by day at the file walk', async () => { const { storage } = await makeCache([[OLD], [NEW]]) const fromOnly = await grep(storage, { from: '2026-08-11' }) @@ -321,3 +362,21 @@ test('an empty or oversized query refuses up front', async () => { await assert.rejects(() => grep(storage, { query: 'x'.repeat(2000) }), /at most/) await assert.rejects(() => grep(storage, { query: '(', regex: true }), /not a valid regular expression/) }) + +test('a missing or non-positive limit refuses up front', async () => { + const cacheRoot = await fs.mkdtemp(path.join(os.tmpdir(), 'hyp-grep-limit-')) + const storage = createQueryStorageService({ cacheRoot }) + await assert.rejects(() => grep(storage, { limit: undefined }), /positive integer/) + await assert.rejects(() => grep(storage, { limit: 0 }), /positive integer/) + await assert.rejects(() => grep(storage, { limit: -1 }), /positive integer/) + await assert.rejects(() => grep(storage, { limit: 2.5 }), /positive integer/) +}) + +test('unreadable table metadata fails the search rather than answering zero', async () => { + const { storage, cacheRoot } = await makeCache([[OLD], [NEW]]) + const metadataDir = path.join(resolveIcebergDir(path.join(cacheRoot, 'datasets', DATASET, 'source=test')), 'metadata') + for (const name of await fs.readdir(metadataDir)) { + if (name.endsWith('.metadata.json')) await fs.writeFile(path.join(metadataDir, name), '{ truncated') + } + await assert.rejects(() => grep(storage), 'a corrupt table raises, matching the SQL read path') +}) From b590b1102ca0a7d78ea452fc11b915cac239bfd6 Mon Sep 17 00:00:00 2001 From: test Date: Thu, 20 Aug 2026 16:35:34 +0000 Subject: [PATCH 3/3] Grep search: a poisoned sidecar degrades one file, not the whole query The sidecar existence probe only ruled out a missing index. A sidecar that exists but cannot be read (a half-written index from a killed build, a truncation from a full disk, a format the installed hypgrep refuses) throws from inside parquetFind, where the footer is parsed and the version checked, and nothing caught it: one poisoned sidecar failed every grep over the whole cache, including the partitions the walk never reached. That makes index state a correctness input, which LLP 0264 #lifecycle says it never is. The indexed read now degrades to the scan tier when the index proves unusable before it produced a row, and the reader catch no longer special-cases ENOENT: any unreadable sidecar is an unindexed file. A failure after the first row still propagates, because retrying it as a scan would count the collected hits twice. Also give the file-walk comparator a 0 for equal days, like sortHits: one day is many files and the early break reads the walk as strictly day-descending, so same-day order should come from the comparator rather than from whatever the engine's sort happens to do. Co-Authored-By: Claude Opus 5 (1M context) --- src/core/search/grep_service.js | 96 +++++++++++++++++++-------- test/core/search-grep-service.test.js | 33 +++++++++ 2 files changed, 103 insertions(+), 26 deletions(-) diff --git a/src/core/search/grep_service.js b/src/core/search/grep_service.js index 5010a009..cec852d0 100644 --- a/src/core/search/grep_service.js +++ b/src/core/search/grep_service.js @@ -204,7 +204,17 @@ export async function executeGrepSearch(args) { // Newest message-day first, across every source partition at once, so // a truncated answer keeps the newest matches whichever client wrote // them (the server's walk order, applied to the client's layout). - files.sort((a, b) => ((a.day ?? UNKNOWN_DAY_SORT_KEY) < (b.day ?? UNKNOWN_DAY_SORT_KEY) ? 1 : -1)) + // Equal days compare equal, like `sortHits` below: one day is many + // files, and the early break below reads the walk as strictly + // day-descending, so a comparator that answered -1 both ways for two + // same-day files would leave that order to whatever the engine's sort + // happens to do rather than to the comparator. + files.sort((a, b) => { + const ad = a.day ?? UNKNOWN_DAY_SORT_KEY + const bd = b.day ?? UNKNOWN_DAY_SORT_KEY + if (ad === bd) return 0 + return ad < bd ? 1 : -1 + }) const { resolver: io } = await createLocalIcebergIO() /** @type {GrepSearchHit[]} */ @@ -234,6 +244,59 @@ export async function executeGrepSearch(args) { if (hits.length >= budget * 2) trimHits() } + /** + * 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. + * + * The existence probe only rules out a missing sidecar. A sidecar + * that exists but cannot be read (a half-written index from a killed + * build, a truncation from a full disk, a format the installed + * hypgrep refuses) throws from inside `parquetFind`, where the footer + * is parsed and the version checked. Left uncaught, one poisoned + * sidecar fails every grep over the whole cache, including the + * partitions the walk never reached, which would make index state a + * correctness input; LLP 0264 #lifecycle says it never is, so a + * poisoned file is brute-scanned exactly like an unindexed one. + * + * @param {{ filePath: string, deletedPositions: Set | undefined }} file + * @param {Awaited>} indexFile + * @returns {Promise} + */ + 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 + try { + for await (const row of rows) { + produced = true + if (file.deletedPositions?.has(BigInt(/** @type {number} */ (row.__index__)))) continue + if (withheld?.(row)) { + localOnly.withheldRows += 1 + continue + } + collect(row) + } + } catch (err) { + if (produced || isAbort(err, signal)) throw err + return false + } + indexedFiles += 1 + return true + } + /** @param {{ filePath: string, deletedPositions: Set | undefined }} file */ const searchFile = async (file) => { // Sidecar existence IS the index marker, no ledger (LLP 0264 @@ -247,33 +310,14 @@ export async function executeGrepSearch(args) { try { indexFile = await io.reader(sidecarUrl) } catch (err) { - if (/** @type {Error & { code?: string }} */ (err)?.code !== 'ENOENT') throw err - } - } - if (indexFile) { - indexedFiles += 1 - // 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) { - if (file.deletedPositions?.has(BigInt(/** @type {number} */ (row.__index__)))) continue - if (withheld?.(row)) { - localOnly.withheldRows += 1 - continue - } - collect(row) + // Every reader failure degrades, not only the delete race: an + // unreadable sidecar is an unindexed file, and an unindexed + // file is the scan tier's, never the caller's error. + if (isAbort(err, signal)) throw err + indexFile = null } - return } + if (indexFile && await searchIndexed(file, indexFile)) return scannedFiles += 1 const sourceFile = await io.reader(file.filePath) const rows = await parquetReadObjects({ file: sourceFile, columns: SCAN_COLUMNS }) diff --git a/test/core/search-grep-service.test.js b/test/core/search-grep-service.test.js index c1c1216e..721e400d 100644 --- a/test/core/search-grep-service.test.js +++ b/test/core/search-grep-service.test.js @@ -334,6 +334,39 @@ test('indexed tier: a stale sidecar cannot resurrect a purged row', async () => assert.deepEqual(res.hits.map((h) => h.sessionId), ['s1'], 'the purged row is filtered by position') }) +test('indexed tier: a poisoned sidecar degrades that file, it does not fail the search', async () => { + const { storage, tableDir } = await makeCache([[OLD], [NEW]]) + const before = await grep(storage) + await buildSidecars(tableDir()) + // A half-written index: the file exists, so the existence probe accepts + // it, and the footer parse inside parquetFind is what fails. LLP 0264 + // #lifecycle makes index state a performance property only, so this one + // file falls back to the scan tier and the answer is unchanged. + const files = await listLiveDataFiles(tableDir()) + const poisoned = urlToPath(files[0].filePath).replace(/\.parquet$/, '.index.parquet') + await fs.writeFile(poisoned, 'PAR1 not really an index') + const res = await grep(storage) + assert.deepEqual(res.hits, before.hits, 'the poisoned file still answers, through the scan tier') + assert.equal(res.scannedFiles, 1, 'exactly the poisoned file degraded') + assert.equal(res.indexedFiles, files.length - 1) +}) + +test('indexed tier: an unreadable sidecar degrades that file rather than throwing', async () => { + const { storage, tableDir } = await makeCache([[NEW]]) + const before = await grep(storage) + await buildSidecars(tableDir()) + const files = await listLiveDataFiles(tableDir()) + const sidecar = urlToPath(files[0].filePath).replace(/\.parquet$/, '.index.parquet') + // A directory where the sidecar should be: the probe sees it, the read + // fails with EISDIR rather than the ENOENT the delete race produces. + await fs.rm(sidecar) + await fs.mkdir(sidecar) + const res = await grep(storage) + assert.deepEqual(res.hits, before.hits) + assert.equal(res.indexedFiles, 0) + assert.equal(res.scannedFiles, files.length) +}) + test('rows captured into the spool are found after the freshness flush', async () => { const { storage } = await makeCache([[OLD]]) const spooled = mkRow({ date: '2026-08-13', session_id: 'spooled', content_text: 'fresh needle from the spool' })