Reimplement query cache with Iceberg

Author: philcunliffe

CHANGELOG.md

@@ -7,6 +7,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed (breaking)
+
+- Local `ctvs query` cache storage now uses local Iceberg tables under
+  `.collectivus-query/cache` instead of Parquet sidecars under
+  `.collectivus-query/parquet`. Config is now `query.cache.enabled` /
+  `query.cache.dir`, and the CLI override is `--cache-dir`.
+- Refreshes keep per-source JSONL cursors and append from the last processed
+  byte where possible. Source truncation/rewrite or collection schema drift
+  starts a new source epoch instead of recreating every cache partition.
+- Glob-backed collection sources that disappear remain queryable as cache-only
+  partitions instead of being pruned on refresh.
+
 ## [2.2.0] — 2026-05-14
 
 ### Added
@@ -76,7 +88,7 @@ new schema reference, working SQL patterns, and dataset semantics.
 
 ### Changed (breaking)
 
-- **`ctvs query` allows stale Parquet cache by default.** Stale partitions
+- **`ctvs query` allows stale query cache by default.** Stale partitions
   (Parquet exists but may be outdated) now query successfully and emit a
   `warning: querying stale data; …` line to stderr instead of exiting non-zero.
   Stdout output (table, json, jsonl, markdown) is unchanged.

README.md

@@ -135,7 +135,7 @@ Pass a JSON config with `--config <path>` (a local path or url). The schema:
     "redact_headers": ["authorization", "x-api-key", "anthropic-api-key", "cookie", "set-cookie"]
   },
   "sink": { "type": "file", "dir": "./collectivus-data" },
-  "query": { "parquet": { "enabled": true } }
+  "query": { "cache": { "enabled": true } }
 }
 ```
 
@@ -147,7 +147,7 @@ Pass a JSON config with `--config <path>` (a local path or url). The schema:
 | `sink`    | Root directory for Standalone JSONL recordings. Proxy rows land under `<sink.dir>/<gateway_id>/proxy/`; OTLP rows land under `<sink.dir>/<gateway_id>/<signal>/`. Required when `otel` or `proxy` is set in Standalone mode. Accepted but unused in Gateway mode. |
 | `central_server` | Gateway-mode Central server URL, identity settings, config poll interval, and optional `outbox_dir`. Gateway rows are first fsynced to this durable local outbox, then shipped to Central ingest. |
 | `upload`  | Optional. Enables the daily S3 parquet drain. See [S3 upload](#s3-upload). |
-| `query`   | Optional. Configures the local `ctvs query` Parquet cache. `query.parquet.enabled` defaults to `true`; `query.parquet.dir` defaults to `<recording-root>/.collectivus-query/parquet`. |
+| `query`   | Optional. Configures the local `ctvs query` query cache. `query.cache.enabled` defaults to `true`; `query.cache.dir` defaults to `<recording-root>/.collectivus-query/cache`. |
 
 Add an `upload` block to drain JSONL to S3 once a day:
 
@@ -526,7 +526,7 @@ and [provider fields](https://developers.openai.com/codex/config-reference#model
 ## Local query
 
 `ctvs query` reads local recordings only. It never contacts S3 and it does not
-auto-refresh its Parquet cache unless you ask for that explicitly.
+auto-refresh its query cache unless you ask for that explicitly.
 
 ```bash
 ctvs query refresh /path/to/gw1/logs/2026-05-11.jsonl --config collectivus.json
@@ -540,17 +540,19 @@ ctvs collect random-log.jsonl --name random-log --config collectivus.json
 ctvs query sql "select * from random_log" --config collectivus.json
 ```
 
-Cache files are written under
-`<recording-root>/.collectivus-query/parquet/<dataset>/gateway_id=<id>/date=<YYYY-MM-DD>/data.parquet`
-with `data.parquet.meta.json` sidecars.
+Cache cursors are written under
+`<recording-root>/.collectivus-query/cache/datasets/<dataset>/gateway_id=<id>/date=<YYYY-MM-DD>/cursor.json`.
+Rows live in local Iceberg tables under the same partition directory. Refreshes
+append from the last recorded JSONL cursor when possible; truncation, rewrite,
+or schema drift starts a new source epoch.
 
 Freshness is treated asymmetrically (since v1.7.0):
 
 | Partition state | Behavior |
 | --- | --- |
 | `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 file-targeted `ctvs query refresh …` command to run when the source file is known. |
+| `stale` (cache exists, may be outdated) | Query proceeds; a `warning: querying stale data; …` line is written to stderr. Stdout is unchanged. |
+| `missing` (no cache table/cursor) | Query exits with the exact file-targeted `ctvs query refresh …` command to run when the source file is known. |
 
 Use `ctvs query refresh <file.jsonl>` to refresh selected source files, or
 `ctvs query refresh --all [dataset]` when you explicitly want the broader
@@ -565,7 +567,7 @@ outdated data).
 > unchanged; the new warning is written only to stderr. `missing`
 > partitions still error.
 
-Logical datasets are `logs`, `traces`, `metrics`, `proxy_messages`, and `gascity_messages`. `ctvs collect <file.jsonl> --name <name>` registers an external JSONL file as a dynamic table; names are normalized for SQL, so `--name random-log` becomes table `random_log`. Collection tables include `_ctvs_source_path`, `_ctvs_line_number`, `_ctvs_raw`, and inferred top-level JSON fields. `ctvs query schema <dataset>` prints the schema, and `ctvs query catalog` shows which datasets have source and cached rows.
+Logical datasets are `logs`, `traces`, `metrics`, `proxy_messages`, and `gascity_messages`. `ctvs collect <file.jsonl> --name <name>` registers an external JSONL file as a dynamic table; `ctvs collect --glob '<pattern>' --name <name>` backs one table with many source files. Names are normalized for SQL, so `--name random-log` becomes table `random_log`. Collection tables include `_ctvs_source_path`, `_ctvs_line_number`, `_ctvs_raw`, and inferred top-level JSON fields. Deleted glob sources remain queryable from their cache-only partitions until the collection is removed. `ctvs query schema <dataset>` prints the schema, and `ctvs query catalog` shows which datasets have source and cached rows.
 
 ### Conversation log model
 
@@ -753,7 +755,7 @@ the binary into a per-invocation cache that is not stable across runs.
 | `ctvs detach [--client claude\|codex\|all]` | Revert Claude Code and/or Codex without uninstalling the daemon |
 | `ctvs status` | Print daemon (loaded / PID) and Claude Code (attached) state |
 | `ctvs export --config <path> [...]` | Convert recorded JSONL to local Parquet without invoking the upload scheduler |
-| `ctvs query <command> [...]` | Query local recordings through the explicit Parquet cache |
+| `ctvs query <command> [...]` | Query local recordings through the explicit query cache |
 | `ctvs collect <file.jsonl> --name <name>` | Register external JSONL as a dynamic query table |
 | `ctvs skills install [--client claude\|codex\|all]` | Install the bundled Collectivus query LLM skill |
 

package.json

@@ -52,6 +52,7 @@
   "dependencies": {
     "hyparquet": "1.25.8",
     "hyparquet-compressors": "1.1.1",
+    "icebird": "0.7.0",
     "squirreling": "0.12.19"
   }
 }

skills/collectivus-query/SKILL.md

@@ -5,7 +5,7 @@ description: Inspect local Collectivus recordings with the ctvs query CLI. Use w
 
 # Collectivus Query
 
-Use `ctvs query` to inspect local Collectivus recordings. It reads local JSONL recordings and an explicit local Parquet cache; it does not query S3.
+Use `ctvs query` to inspect local Collectivus recordings. It reads local JSONL recordings and an explicit local query cache; it does not query S3.
 
 ## Workflow
 
@@ -39,7 +39,7 @@ ctvs collect <file.jsonl> --name <name>
 ctvs collect --glob '<pattern>' --name <name>
 ```
 
-`ctvs collect` registers an external local JSONL file as a dynamic SQL table and immediately refreshes its Parquet cache. Collection names are normalized for SQL (`random-log` -> `random_log`). Query them with `ctvs query sql "select * from random_log"`. Pass `--glob '<pattern>'` instead of a single path to back one logical table with many files; each matched file becomes its own cache partition and `_ctvs_source_path` tells you which file a row came from.
+`ctvs collect` registers an external local JSONL file as a dynamic SQL table and immediately refreshes its query cache. Collection names are normalized for SQL (`random-log` -> `random_log`). Query them with `ctvs query sql "select * from random_log"`. Pass `--glob '<pattern>'` instead of a single path to back one logical table with many files; each matched file becomes its own cache partition and `_ctvs_source_path` tells you which file a row came from.
 
 ## Proxy conversation log model
 
@@ -107,7 +107,7 @@ Use `JSON_VALUE(<col>, '$.path')` to extract scalars from the `attributes` / `st
 ## Guardrails
 
 - Do not assume the cache auto-refreshes. Query commands default to `--refresh never`, and stale partitions return data with a stderr warning rather than refreshing themselves.
-- Always read stderr. A successful exit code does not mean the data is fresh — a `warning: querying stale data; …` line on stderr means stdout reflects outdated Parquet, and the user should be told before drawing conclusions.
+- Always read stderr. A successful exit code does not mean the data is fresh — a `warning: querying stale data; …` line on stderr means stdout reflects outdated cache rows, and the user should be told before drawing conclusions.
 - Do not paste `--config` into every command by habit. Use it when discovery shows the service is not using `~/.hyp/collectivus.json`.
 - Do not read arbitrary Parquet files directly for `ctvs query sql`; the CLI only allows logical tables.
 - Keep SQL read-only and use only logical datasets: `logs`, `traces`, `metrics`, `proxy_messages`, `gascity_messages`, and registered collection tables from `ctvs query catalog`.

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

@@ -1,9 +1,9 @@
 # Collectivus Query CLI Reference
 
-`ctvs query` reads local Collectivus recordings. It materializes JSONL source files into a local Parquet cache under:
+`ctvs query` reads local Collectivus recordings. It materializes JSONL source files into a local query cache under:
 
 ```text
-<recording-root>/.collectivus-query/parquet/<dataset>/gateway_id=<id>/date=<YYYY-MM-DD>/data.parquet
+<recording-root>/.collectivus-query/cache/datasets/<dataset>/gateway_id=<id>/date=<YYYY-MM-DD>/cursor.json
 ```
 
 External JSONL collections registered with `ctvs collect <file.jsonl> --name <name>` are recorded in:
@@ -12,15 +12,15 @@ External JSONL collections registered with `ctvs collect <file.jsonl> --name <na
 <recording-root>/.collectivus-query/collections.json
 ```
 
-Their Parquet cache lives under `<recording-root>/.collectivus-query/parquet/collections/<table>/data.parquet`. Names are normalized for SQL, so `--name random-log` exposes table `random_log`.
+Their query cache lives under `<recording-root>/.collectivus-query/cache/collections/<table>/source=<hash>/cursor.json`, with rows stored in local Iceberg tables below each source partition. Names are normalized for SQL, so `--name random-log` exposes table `random_log`.
 
 The cache is explicit. Query commands do not refresh it unless `--refresh always` is passed.
 
 Freshness is asymmetric (since v1.7.0):
 
 - `fresh` — query proceeds silently.
-- `stale` (Parquet exists but may be outdated) — query proceeds and writes a `warning: querying stale data; N partition(s) outdated [...] — run '...' to update` line to stderr. Stdout is unchanged.
-- `missing` (no Parquet at all) — query exits with the exact `ctvs query refresh ...` command to run.
+- `stale` (cache exists but may be outdated) — query proceeds and writes a `warning: querying stale data; N partition(s) outdated [...] — run '...' to update` line to stderr. Stdout is unchanged.
+- `missing` (no cache table/cursor) — query exits with the exact `ctvs query refresh ...` command to run.
 
 Pass `--strict-freshness` to restore the pre-1.7 behavior where stale partitions are a hard error.
 
@@ -29,7 +29,7 @@ Commands default to `~/.hyp/collectivus.json`. If the running gateway or OTEL co
 ## Shared Options
 
 - `--config <path|url>`: Collectivus config. Defaults to `~/.hyp/collectivus.json`.
-- `--parquet-dir <dir>`: Override the query cache directory.
+- `--cache-dir <dir>`: Override the query cache directory.
 - `--from <timestamp>` / `--to <timestamp>`: Inclusive timestamp bounds.
 - `--since <duration>`: Relative lower bound such as `15m`, `2h`, or `7d`.
 - `--date <YYYY-MM-DD>`: Restrict to one UTC date partition.
@@ -48,8 +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 <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 refresh <file.jsonl>... [--force]`: Materialize selected JSONL source files into the query cache.
+- `ctvs query refresh --all [dataset] [--force]`: Materialize all matching JSONL source files into the query cache.
 - `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.
@@ -71,9 +71,9 @@ 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 `ctvs query refresh <file.jsonl>` refreshes selected files. Use `--refresh always` to refresh before running the query.
+`ctvs collect` stores the absolute source path (or glob) and immediately refreshes the query 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.
+With `--glob`, one logical table is backed by many source files: each matched file becomes its own cache partition under `.collectivus-query/cache/collections/<table>/source=<hash>/cursor.json`, and refresh appends from each file's recorded cursor when possible. Files that no longer match the glob remain queryable as cache-only partitions. Inside SQL, use `_ctvs_source_path` to see which file a row came from.
 
 Collection tables always include `_ctvs_source_path`, `_ctvs_line_number`, and `_ctvs_raw`, plus inferred top-level JSON fields. Use `--timestamp-column <field>` when registering a file if `--from`, `--to`, `--since`, or `--date` should use a specific field.
 
@@ -82,7 +82,7 @@ Collection tables always include `_ctvs_source_path`, `_ctvs_line_number`, and `
 - `logs`: OTLP log records. Common columns include `gateway_id`, `date`, `timestamp`, `observedTimestamp`, `severityNumber`, `severityText`, `serviceName`, `body`, `traceId`, `spanId`, `resource`, `scope`, and `attributes`.
 - `traces`: OTLP spans. Common columns include `gateway_id`, `date`, `traceId`, `spanId`, `parentSpanId`, `name`, `kind`, `startTimestamp`, `endTimestamp`, `durationMs`, `status`, `serviceName`, `resource`, `scope`, and `attributes`.
 - `metrics`: OTLP metric points. Common columns include `gateway_id`, `date`, `metricName`, `metricType`, `timestamp`, `startTimestamp`, `serviceName`, `value`, `valueInt`, `count`, `sum`, `unit`, `resource`, `scope`, and `attributes`.
-- `proxy_messages`: One row per LLM proxy content part (text block, tool call, tool result, etc.), globally deduped by content-derived `message_id`. See **proxy_messages columns** below for the full 26-column schema; `gateway_id` and `date` are added as partition columns on the on-disk Parquet.
+- `proxy_messages`: One row per LLM proxy content part (text block, tool call, tool result, etc.), globally deduped by content-derived `message_id`. See **proxy_messages columns** below for the full 26-column schema; `gateway_id` and `date` are added as partition columns in the query cache.
 - `gascity_messages`: One row per content block from gascity-captured agent sessions (text, thinking, tool_use, tool_result, attachment). Captured by the `ctvs gascity` supervisor source — agent-attributed (`gascity_template` / `gascity_rig` / `gascity_alias`) and includes per-frame token usage with cache breakdown. Always fresh: the daemon writes Parquet directly to `~/.collectivus/sink/gascity_messages/date=<YYYY-MM-DD>/city=<name>/` (no JSONL stage, no `.meta.json` sidecar). The constant `gateway_id = 'gascity-scribe'` tags the source for cross-source UNIONs with `proxy_messages`. Run `ctvs query schema gascity_messages --format markdown` for the full 47-column schema.
 
 Run `ctvs query schema <dataset> --format json` for the exact columns in the installed version.
@@ -120,7 +120,7 @@ Grain is one row per content part. Rows are deduplicated by `message_id` so a si
 | 25 | `status` | JSON | yes | Sparse: `tool_status` on tool_result, `finish_reason` on the last assistant part, `error_code` / `error_message` on error blocks | `null` when no key applies. |
 | 26 | `attributes` | JSON | yes | `request` settings (`max_tokens`, `thinking`, `output_config`, `context_management`, `stream`), `usage` (assistant only — `input_tokens`, `output_tokens`, `cache_read_tokens`, `cache_write_tokens`), `timing.latency_ms`, `client.claude_version`, `provider_raw.metadata` | `null` when no key applies. |
 
-The on-disk Parquet additionally carries `gateway_id` (STRING, not null) as the leading column and `date` (STRING, not null) at the end. Both are partition keys; `date` is derived from `message_created_at` in UTC.
+The query-cache table additionally carries `gateway_id` (STRING, not null) as the leading column and `date` (STRING, not null) at the end. Both are partition keys; `date` is derived from `message_created_at` in UTC.
 
 ## Example SQL
 

src/cli.js

@@ -57,7 +57,7 @@ Commands:
   ctvs status                                  Report daemon, config, recordings, attach state
   ctvs gascity <subcommand> [...]              Manage gascity supervisor capture sources
   ctvs export --config <path|url> [...]        Convert recorded JSONL to Parquet
-  ctvs query <command> [...]                   Query local recordings through Parquet cache
+  ctvs query <command> [...]                   Query local recordings through query cache
   ctvs collect <file.jsonl> --name <name>      Add external JSONL as a query table
   ctvs skills install [...]                    Install the Collectivus query LLM skill
   ctvs ignore <add|remove|list> [path]         Suppress Claude recording for a folder