Add file-targeted query refresh

Author: philcunliffe

README.md

@@ -529,7 +529,8 @@ and [provider fields](https://developers.openai.com/codex/config-reference#model
 auto-refresh its Parquet cache unless you ask for that explicitly.
 
 ```bash
-ctvs query refresh --config collectivus.json
+ctvs query refresh /path/to/gw1/logs/2026-05-11.jsonl --config collectivus.json
+ctvs query refresh --all logs --config collectivus.json
 ctvs query logs --config collectivus.json --since 1h
 ctvs query traces slow --config collectivus.json --limit 20
 ctvs query metrics series latency.ms --config collectivus.json
@@ -549,9 +550,11 @@ Freshness is treated asymmetrically (since v1.7.0):
 | --- | --- |
 | `fresh` | Query proceeds silently. |
 | `stale` (Parquet exists, may be outdated) | Query proceeds; a `warning: querying stale data; …` line is written to stderr. Stdout is unchanged. |
-| `missing` (no Parquet at all) | Query exits with the exact `ctvs query refresh …` command to run. |
+| `missing` (no Parquet at all) | Query exits with the exact file-targeted `ctvs query refresh …` command to run when the source file is known. |
 
-Use `--refresh always` to force a refresh before the query runs. Use
+Use `ctvs query refresh <file.jsonl>` to refresh selected source files, or
+`ctvs query refresh --all [dataset]` when you explicitly want the broader
+walk. Use `--refresh always` to force a refresh before the query runs. Use
 `--strict-freshness` to restore the pre-1.7 behavior where stale partitions
 are a hard error (useful in CI / scheduled jobs that must never read
 outdated data).

skills/collectivus-query/references/query-cli.md

@@ -38,6 +38,8 @@ Commands default to `~/.hyp/collectivus.json`. If the running gateway or OTEL co
 - `--limit <n>`: Maximum rows to render. Default `100`, maximum `1000`.
 - `--format <fmt>`: `table`, `json`, `jsonl`, or `markdown`.
 - `--refresh <mode>`: `never` or `always`. Default `never`.
+- `--all`: Refresh all matching source files for `ctvs query refresh`.
+- `--force`: Rebuild fresh cache partitions too for `ctvs query refresh`.
 - `--strict-freshness`: Treat stale partitions as a hard error (pre-1.7 behavior). Off by default.
 
 ## Commands
@@ -46,7 +48,8 @@ Commands default to `~/.hyp/collectivus.json`. If the running gateway or OTEL co
 - `ctvs query status`: Inspect source partitions and cache freshness.
 - `ctvs query catalog`: List logical datasets, columns, source partitions, and cached row counts.
 - `ctvs query schema <dataset>`: Print static schema for a logical dataset.
-- `ctvs query refresh [dataset] [--force]`: Materialize local JSONL into query-cache Parquet.
+- `ctvs query refresh <file.jsonl>... [--force]`: Materialize selected JSONL source files into query-cache Parquet.
+- `ctvs query refresh --all [dataset] [--force]`: Materialize all matching JSONL source files into query-cache Parquet.
 - `ctvs query sample <dataset>`: Show sample rows.
 - `ctvs query sql <select-sql>`: Run read-only SQL over logical datasets.
 - `ctvs query logs [count|tail]`: List logs, count logs, or tail live JSONL without requiring cache.
@@ -68,7 +71,7 @@ ctvs collect --glob '/path/to/segments/**/*.jsonl' --name segments
 ctvs query sql "select * from random_log" --format json
 ```
 
-`ctvs collect` stores the absolute source path (or glob) and immediately refreshes the Parquet cache. If the source file changes later, normal query freshness rules apply: stale cached data is queryable with a stderr warning, `--strict-freshness` turns that into an error, and `--refresh always` refreshes before running the query.
+`ctvs collect` stores the absolute source path (or glob) and immediately refreshes the Parquet cache. If the source file changes later, normal query freshness rules apply: stale cached data is queryable with a stderr warning, `--strict-freshness` turns that into an error, and `ctvs query refresh <file.jsonl>` refreshes selected files. Use `--refresh always` to refresh before running the query.
 
 With `--glob`, one logical table is backed by many source files: each matched file becomes its own cache partition under `.collectivus-query/parquet/collections/<table>/source=<hash>/data.parquet`, and only files whose mtime/size changed re-materialize on refresh. Files that no longer match the glob are pruned from the cache on the next refresh. Inside SQL, use `_ctvs_source_path` to see which file a row came from.
 

src/cli/init_presets/gascity_skill.md

@@ -161,11 +161,11 @@ ORDER BY source, idx;
 
 ## Freshness
 
-`session_segments` is glob-backed: new segment files only appear in the cache after a refresh. Run `ctvs query refresh session_segments` to pick up new segments, or use `--refresh always` on any query. Deleted segment files are pruned from the cache on the next refresh.
+`session_segments` is glob-backed: new segment files only appear in the cache after a refresh. Run `ctvs query refresh <segment-file.jsonl>` for selected segment files, `ctvs query refresh --all session_segments` to pick up every matching segment, or use `--refresh always` on any query. Deleted segment files are pruned from the cache on the next `--all` refresh.
 
 `events` is append-only single-file; mtime/size changes trigger re-materialization on refresh.
 
-`gascity_messages` is **always fresh** — the daemon writes Parquet directly into the sink (no JSONL stage, no `.meta.json` sidecar), so query-time discovery picks up every part-file the writer has flushed. `ctvs query refresh gascity_messages` is a documented no-op (it lists existing partitions as already-fresh). To pull in newly-flushed rows simply rerun the query.
+`gascity_messages` is **always fresh** — the daemon writes Parquet directly into the sink (no JSONL stage, no `.meta.json` sidecar), so query-time discovery picks up every part-file the writer has flushed. `ctvs query refresh --all gascity_messages` is a documented no-op (it lists existing partitions as already-fresh). To pull in newly-flushed rows simply rerun the query.
 
 Full schemas: `ctvs query schema events --format markdown`, `ctvs query schema session_segments --format markdown`, `ctvs query schema gascity_messages --format markdown`. Catalog: `ctvs query catalog --format markdown`.
 
@@ -176,5 +176,5 @@ Refreshing isn't free. `events.jsonl` and the `session_segments/**/*.jsonl` file
 Recommended workflow:
 
 1. Run `ctvs query status` first. The summary shows which date ranges are already cached and which are stale; cheap queries against a covered range never need a refresh.
-2. Only invoke `--refresh always` or `ctvs query refresh <dataset>` when a needed date range is missing or `status` reports staleness in the window you care about.
+2. Only invoke `--refresh always`, `ctvs query refresh <file.jsonl>`, or `ctvs query refresh --all <dataset>` when a needed date range is missing or `status` reports staleness in the window you care about.
 3. Do not reflexively pass `--refresh always` "just in case". Stale-data queries print a warning to stderr (the default behavior); reading that warning is cheaper than re-materializing the cache. Treat refresh as a deliberate step, not a default.

src/cli/query.js

@@ -1,4 +1,5 @@
 import fs from 'node:fs'
+import path from 'node:path'
 import process from 'node:process'
 import { ConfigError, loadConfigAsync as defaultLoadConfig } from '../config.js'
 import { defaultConfigPath } from './common.js'
@@ -47,7 +48,8 @@ Commands:
   status                         Inspect JSONL sources and query-cache freshness
   catalog                        List logical datasets and cached row counts
   schema <dataset>               Print the static logical schema
-  refresh [dataset] [--force]    Materialize local JSONL into query-cache Parquet
+  refresh <file.jsonl>...        Materialize selected JSONL source files into Parquet
+  refresh --all [dataset]        Materialize all matching JSONL into query-cache Parquet
   sql <select-sql>               Run read-only SELECT SQL over logical datasets
   sample <dataset>               Show sample rows
   doctor                         Check query prerequisites
@@ -75,6 +77,8 @@ Shared options:
   --refresh <mode>               never or always (default: never).
                                  Stale partitions query with a stderr warning;
                                  missing partitions always error.
+  --all                          Refresh all matching source files (refresh command only)
+  --force                        Rebuild fresh cache partitions too (refresh command only)
   --strict-freshness             Treat stale partitions as errors (pre-1.7 behavior)
   --help, -h                     Show this help`
 
@@ -191,6 +195,7 @@ export async function runQuery(argv, hooks = {}) {
  *   limit: number,
  *   format: QueryFormat,
  *   refresh: QueryRefreshMode,
+ *   all: boolean,
  *   force: boolean,
  *   strictFreshness: boolean,
  *   error?: string,
@@ -205,12 +210,14 @@ export function parseQueryArgs(argv) {
     limit: DEFAULT_LIMIT,
     format: 'table',
     refresh: 'never',
+    all: false,
     force: false,
     strictFreshness: false,
   }
   for (let i = 0; i < argv.length; i++) {
     const arg = argv[i]
     if (arg === '--help' || arg === '-h') { out.help = true; return out }
+    if (arg === '--all') { out.all = true; continue }
     if (arg === '--force') { out.force = true; continue }
     if (arg === '--strict-freshness') { out.strictFreshness = true; continue }
     /**
@@ -393,7 +400,7 @@ function handleSchema(paths, parsed, stdout, stderr) {
   }
   const meta = readAnyCollectionMeta(paths.parquetDir, collection)
   if (!meta) {
-    stderr.write(`error: query cache is missing for ${collection.table}. Run: ${refreshCommand(parsed)}\n`)
+    stderr.write(`error: query cache is missing for ${collection.table}. Run: ${refreshCommand(parsed, undefined, { ...baseScope(parsed), datasets: [collection.table] })}\n`)
     return 1
   }
   const rows = meta.columns.map((column) => ({
@@ -414,11 +421,31 @@ function handleSchema(paths, parsed, stdout, stderr) {
  * @returns {Promise<number>}
  */
 async function handleRefresh(paths, parsed, stdout, stderr) {
-  const scope = scopeWithOptionalDataset(paths, parsed, parsed.positionals[1])
   if (!paths.parquetEnabled || !paths.parquetDir) {
     stderr.write('error: query parquet cache is disabled; pass --parquet-dir to refresh explicitly\n')
     return 1
   }
+  const targets = parsed.positionals.slice(1)
+  /** @type {QueryScope} */
+  let scope
+  if (parsed.all) {
+    if (targets.length > 1) {
+      stderr.write('error: refresh --all accepts at most one dataset\n')
+      return 2
+    }
+    scope = scopeWithOptionalDataset(paths, parsed, targets[0])
+  } else {
+    if (targets.length === 0) {
+      stderr.write('error: refresh requires one or more JSONL files; pass --all to refresh all matching sources\n')
+      return 2
+    }
+    const datasetTarget = targets.length === 1 ? resolveQueryTable(paths, targets[0]) : undefined
+    if (datasetTarget) {
+      stderr.write(`error: refresh ${targets[0]} targets a dataset; pass --all to refresh all ${datasetTarget} sources\n`)
+      return 2
+    }
+    scope = refreshScopeForSourceFiles(paths, parsed, targets)
+  }
   const result = await refreshAllCaches({ paths, scope, force: parsed.force, stdout })
   if (result.written === 0 && result.skipped === 0 && result.failures === 0) {
     stdout.write(`No JSONL files matched in ${paths.recordingRoot}.\n`)
@@ -793,7 +820,7 @@ async function ensureCacheReady(paths, scope, parsed) {
     const detail = `${partitionLabel(first)}: missing${first.reason ? ` (${first.reason})` : ''}`
     return {
       ok: false,
-      message: `error: query cache is missing for ${detail}. Run: ${refreshCommand(parsed)}`,
+      message: `error: query cache is missing for ${detail}. Run: ${refreshCommand(parsed, missing, scope)}`,
     }
   }
 
@@ -803,14 +830,14 @@ async function ensureCacheReady(paths, scope, parsed) {
       const detail = `${partitionLabel(first)}: stale${first.reason ? ` (${first.reason})` : ''}`
       return {
         ok: false,
-        message: `error: query cache is stale for ${detail} (--strict-freshness set). Run: ${refreshCommand(parsed)}`,
+        message: `error: query cache is stale for ${detail} (--strict-freshness set). Run: ${refreshCommand(parsed, stale, scope)}`,
       }
     }
     const summary = stale.slice(0, 3).map((state) => `${partitionLabel(state)}${state.reason ? ` (${state.reason})` : ''}`).join(', ')
     const more = stale.length > 3 ? `, +${stale.length - 3} more` : ''
     return {
       ok: true,
-      warnings: [`warning: querying stale data; ${stale.length} partition(s) outdated [${summary}${more}] — run '${refreshCommand(parsed)}' to update`],
+      warnings: [`warning: querying stale data; ${stale.length} partition(s) outdated [${summary}${more}] — run '${refreshCommand(parsed, stale, scope)}' to update`],
     }
   }
 
@@ -819,17 +846,50 @@ async function ensureCacheReady(paths, scope, parsed) {
 
 /**
  * @param {ReturnType<typeof parseQueryArgs>} parsed
+ * @param {Array<{ partition: unknown }>} [states]
+ * @param {QueryScope} [scope]
  * @returns {string}
  */
-function refreshCommand(parsed) {
+function refreshCommand(parsed, states, scope) {
   const parts = ['ctvs', 'query', 'refresh']
+  const sourcePaths = states ? refreshableSourcePaths(states) : []
+  if (sourcePaths.length > 0 && sourcePaths.length <= 5) {
+    parts.push(...sourcePaths.map(shellQuote))
+  } else {
+    parts.push('--all')
+    const datasets = scope?.datasets ?? (scope?.dataset ? [scope.dataset] : undefined)
+    if (datasets?.length === 1) parts.push(datasets[0])
+  }
   if (parsed.configPath) parts.push('--config', shellQuote(parsed.configPath))
   if (parsed.parquetDir) parts.push('--parquet-dir', shellQuote(parsed.parquetDir))
   if (parsed.gatewayId) parts.push('--gateway-id', shellQuote(parsed.gatewayId))
   if (parsed.date) parts.push('--date', parsed.date)
   return parts.join(' ')
 }
 
+/**
+ * @param {Array<{ partition: unknown }>} states
+ * @returns {string[]}
+ */
+function refreshableSourcePaths(states) {
+  const seen = new Set()
+  /** @type {string[]} */
+  const out = []
+  for (const state of states) {
+    const { partition: rawPartition } = state
+    const partition = /** @type {Record<string, unknown>} */ (rawPartition)
+    const sourcePath = partition.jsonlPath
+    if (typeof sourcePath !== 'string') continue
+    if (!sourcePath.endsWith('.jsonl')) continue
+    if (!fs.existsSync(sourcePath)) continue
+    const abs = path.resolve(sourcePath)
+    if (seen.has(abs)) continue
+    seen.add(abs)
+    out.push(abs)
+  }
+  return out
+}
+
 /**
  * @param {QueryPaths} paths
  * @param {ReturnType<typeof parseQueryArgs>} parsed
@@ -1097,6 +1157,53 @@ function scopeWithOptionalDataset(paths, parsed, rawDataset) {
   return scope
 }
 
+/**
+ * @param {QueryPaths} paths
+ * @param {ReturnType<typeof parseQueryArgs>} parsed
+ * @param {string[]} targets
+ * @returns {QueryScope}
+ */
+function refreshScopeForSourceFiles(paths, parsed, targets) {
+  /** @type {string[]} */
+  const sourcePaths = []
+  const seenTargets = new Set()
+  for (const target of targets) {
+    const abs = path.resolve(target)
+    if (seenTargets.has(abs)) continue
+    seenTargets.add(abs)
+    sourcePaths.push(abs)
+  }
+
+  const scope = { ...baseScope(parsed), sourcePaths }
+  const matched = new Set()
+  const datasets = new Set()
+  for (const source of discoverSourceFiles(paths.recordingRoot, scope)) {
+    matched.add(path.resolve(source.jsonlPath))
+    datasets.add(datasetForSourceSignal(source.signal))
+  }
+  for (const partition of expectedCollectionPartitions(paths, scope)) {
+    matched.add(path.resolve(partition.jsonlPath))
+    datasets.add(partition.table)
+  }
+
+  const unknown = sourcePaths.filter((sourcePath) => !matched.has(sourcePath))
+  if (unknown.length > 0) {
+    throw new Error(`refresh source file is not a known recording or collection source: ${unknown[0]}`)
+  }
+  if (datasets.size === 0) {
+    throw new Error('no refreshable source files matched')
+  }
+  return { ...scope, datasets: [...datasets] }
+}
+
+/**
+ * @param {import('../query/types.js').SourceFile['signal']} signal
+ * @returns {string}
+ */
+function datasetForSourceSignal(signal) {
+  return signal === 'proxy' ? 'proxy_messages' : signal
+}
+
 /**
  * @param {QueryPaths} paths
  * @param {string} raw

src/query/collections.js

@@ -277,8 +277,11 @@ export function expectedCollectionPartitions(paths, scope) {
   if (!paths.parquetDir) return []
   const manifest = readCollectionsManifest(paths.recordingRoot)
   const wanted = collectionTablesForScope(manifest, scope)
-  const parquetDir = paths.parquetDir
-  return wanted.flatMap((collection) => collectionPartitionsFor(parquetDir, collection))
+  const { parquetDir } = paths
+  const sourcePaths = sourcePathFilter(scope)
+  const partitions = wanted.flatMap((collection) => collectionPartitionsFor(parquetDir, collection))
+  if (!sourcePaths) return partitions
+  return partitions.filter((partition) => sourcePaths.has(path.resolve(partition.jsonlPath)))
 }
 
 /**
@@ -540,6 +543,7 @@ export function readCollectionCacheMeta(metaPath) {
  */
 function pruneOrphanCollectionPartitions(paths, scope, stdout) {
   if (!paths.parquetDir) return
+  if (scope.sourcePaths && scope.sourcePaths.length > 0) return
   const manifest = readCollectionsManifest(paths.recordingRoot)
   const wanted = collectionTablesForScope(manifest, scope)
   for (const collection of wanted) {
@@ -894,6 +898,15 @@ function collectionTablesForScope(manifest, scope) {
   return all.filter((collection) => wanted.has(collection.table))
 }
 
+/**
+ * @param {QueryScope} scope
+ * @returns {Set<string> | undefined}
+ */
+function sourcePathFilter(scope) {
+  if (!scope.sourcePaths) return undefined
+  return new Set(scope.sourcePaths.map((sourcePath) => path.resolve(sourcePath)))
+}
+
 /**
  * @returns {CollectionsManifest}
  */

src/query/paths.js

@@ -150,6 +150,7 @@ export function discoverSourceFiles(root, scope) {
   const wantedSignals = datasets
     ? new Set(datasets.map((dataset) => sourceSignalForDataset(dataset)))
     : new Set(['logs', 'traces', 'metrics', 'proxy'])
+  const sourcePaths = sourcePathFilter(scope)
   /** @type {SourceFile[]} */
   const files = []
   const gatewayIds = scope.gatewayId ? [scope.gatewayId] : safeReadDir(root)
@@ -166,6 +167,7 @@ export function discoverSourceFiles(root, scope) {
         const date = match[1]
         if (scope.date && date !== scope.date) continue
         const jsonlPath = path.join(signalDir, entry)
+        if (sourcePaths && !sourcePaths.has(path.resolve(jsonlPath))) continue
         const stat = safeStat(jsonlPath)
         if (!stat || !stat.isFile()) continue
         files.push({
@@ -430,6 +432,7 @@ function staleReason(partition, meta) {
 export function listCacheMetas(parquetDir, scope) {
   const requestedDatasets = scope.datasets ?? (scope.dataset ? [scope.dataset] : undefined)
   const datasets = requestedDatasets ? requestedDatasets.filter(isQueryDataset) : QUERY_DATASETS
+  const sourcePaths = sourcePathFilter(scope)
   /** @type {CacheMeta[]} */
   const out = []
   for (const dataset of datasets) {
@@ -446,6 +449,7 @@ export function listCacheMetas(parquetDir, scope) {
         const date = dateMatch[1]
         if (scope.date && date !== scope.date) continue
         const meta = readCacheMeta(path.join(gatewayDir, dateEntry, 'data.parquet.meta.json'))
+        if (meta && sourcePaths && !sourcePaths.has(path.resolve(meta.source_path))) continue
         if (meta) out.push(meta)
       }
     }
@@ -513,6 +517,15 @@ function isFile(p) {
   return safeStat(p)?.isFile() ?? false
 }
 
+/**
+ * @param {QueryScope} scope
+ * @returns {Set<string> | undefined}
+ */
+function sourcePathFilter(scope) {
+  if (!scope.sourcePaths) return undefined
+  return new Set(scope.sourcePaths.map((sourcePath) => path.resolve(sourcePath)))
+}
+
 /**
  * @param {SourceFile} a
  * @param {SourceFile} b

src/query/types.d.ts

@@ -14,6 +14,8 @@ export type QueryRefreshMode = 'never' | 'always'
 export interface QueryScope {
   dataset?: string
   datasets?: string[]
+  /** Absolute source JSONL paths to include when refreshing or inspecting a narrow cache scope. */
+  sourcePaths?: string[]
   gatewayId?: string
   date?: string
   from?: string