diff --git a/src/core/cache/iceberg/store.js b/src/core/cache/iceberg/store.js index 5ee9a2e8..23d45a30 100644 --- a/src/core/cache/iceberg/store.js +++ b/src/core/cache/iceberg/store.js @@ -448,6 +448,52 @@ 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. 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 }[]>} + */ +export async function listLiveDataFiles(tablePath) { + if (!tableExists(tablePath)) return [] + const { resolver, lister } = await getLocalIO() + const url = tableUrlForDir(tablePath) + 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 [] + 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..cec852d0 --- /dev/null +++ b/src/core/search/grep_service.js @@ -0,0 +1,507 @@ +// @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 + // `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) + 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). + // 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[]} */ + const hits = [] + let exhausted = true + 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() + } + + /** + * 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 + // #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) { + // 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 + } + } + 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 }) + for (let i = 0; i < rows.length; i++) { + if (i % ABORT_CHECK_ROWS === 0) signal?.throwIfAborted() + if (file.deletedPositions?.has(BigInt(i))) continue + const row = rows[i] + if (!accept(row)) continue + if (withheld?.(row)) { + localOnly.withheldRows += 1 + continue + } + collect(row) + } + } + + try { + for (const file of files) { + signal?.throwIfAborted() + if (hits.length >= budget) { + 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))) { + 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, signal)) throw err + exhausted = false + } + + trimHits() + const truncated = hits.length > limit + if (truncated) hits.length = limit + + 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 + 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 + }) +} + +/** + * 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 && chainId === undefined) return () => true + return (row) => { + 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). + 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 +} + +/** + * 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/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..721e400d --- /dev/null +++ b/test/core/search-grep-service.test.js @@ -0,0 +1,415 @@ +// @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('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' }) + 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('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' }) + 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/) +}) + +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') +})