Skip to content

Commit 007557a

Browse files
philcunliffeclaude
andauthored
ctvs gascity source: capture supervisor sessions as a third Collectivus data source (#111)
* feat(gascity): daemon source skeleton (config, supervisor, workers, dispatch) (#104) Bead 1/6 of co-276c. Adds the gascity capture source as a new worker class inside the existing ctvs daemon — config schema, lifecycle SSE subscriber, per-session frame workers, and a pluggable normalizer dispatcher with stubs that beads 2 and 4 will fill in. The factory wires into buildConfigListeners under section name `gascity` and is enabled in standalone and server modes when the config carries a `gascity` array. Each city gets its own supervisor SSE; spawned per-session workers stream from the supervisor's `format=raw` endpoint and persist atomic cursors so reconnect resumes without data loss. SSE id-extraction was added to SseParser to support `Last-Event-ID` resume. Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * feat(gascity): Claude provider normalizer (co-276c.2) (#105) * feat(gascity): claude provider normalizer (co-276c.2) Implements bead 2 of co-276c — the production Claude normalizer plus its wire contract with the parquet writer (bead 3). Swaps in for the bead-1 `claudeStub` registered by the dispatcher. The normalizer is a pure function (frame, ctx) -> NormalizedRow[]: - All 9 outer frame types covered (assistant, user, attachment, last-prompt, permission-mode, ai-title, file-history-snapshot, queue-operation, system) plus a passthrough row for unknown types so raw_frame is never silently dropped. - All 4 content-block types covered (text, thinking, tool_use, tool_result). Multi-block assistant messages yield N rows that share (message_id, provider_uuid) with distinct part_index. - Outer-frame hoist: cwd, gitBranch, permissionMode, isSidechain, entrypoint, version, promptId, requestId, parentUuid, sourceToolAssistantUUID, timestamp, sessionId. - Assistant message hoist: model, stop_reason/details, full usage block (tokens, cache_*, ephemeral_*h, service_tier, inference_geo, speed); iterations / server_tool_use overflow into attributes.usage_overflow. - tool_result content flattening: list-shaped content (text + tool_reference + ...) flattens into a single content_text string. Wire shape: - NormalizerFn signature changes from `=> void` to `=> NormalizedRow[]`. Dispatcher returns the rows; session_worker captures them and (until bead 3 lands the writer) discards after a debug log so cursor advance semantics are unaffected. - New `src/gascity/normalizers/{claude,index,types.d.ts}` module. `registerProductionNormalizers(dispatcher)` is called from `startGascitySource` so the daemon picks up `claude` automatically. - SessionContext gains an optional `conversationStartedAt`; bead 3 will populate it, the normalizer reads it as-is and stores `null` otherwise. Tests: - 33 new tests in test/gascity/normalizers/claude.test.js driven by 62 sanitized real frames captured from ~/.claude/projects/-Users-phil- hyptown--gc-*/ (UUIDs / msg_/req_/toolu_ ids / sessionIds / absolute paths replaced with deterministic tokens; structural content preserved). - Fixture corpus exercises every outer type + content-block variant, plus tool-call linkage, attachment_type variants (hook_success, skill_listing, hook_non_blocking_error, deferred_tools_delta, task_reminder, edited_text_file), system subtypes (stop_hook_summary, turn_duration, api_error, scheduled_task_fire, away_summary). - Round-trip identity test covers every documented column. - Failure-mode tests: non-object input, unknown frame type, malformed assistant frame, numeric usage strings, missing sessionId, missing conversationStartedAt. - 3 new dispatcher tests cover the return-value contract. Coverage: 81 test files / 1359 tests pass. Typecheck clean. Lint matches baseline (0 errors). Out of scope (per epic): parquet writer + cursor flush (bead 3), codex normalizer (bead 4), CLI surface (bead 5), catalog registration (bead 6). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * fix(gascity): preserve non-string attachment.content in attributes (co-276c.2) Only strip `content` from attachment attributes when it was rendered into content_text (i.e. when it was a non-empty string). Otherwise structured payloads like task_reminder's `content: []` or future array shapes would survive only in `raw_frame` rather than the queryable `attributes` column. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * feat(gascity): parquet writer + per-session cursor management (co-276c.3) (#107) Implements bead 3 of the gascity source: per-session buffers, atomic parquet flushes with snappy compression, write-time uuid dedup, and backfill on startup. Replaces the bead-1 passthrough stub with one that emits a `raw_frame` row so unknown providers still land in `gascity_messages`. The writer takes cursor ownership from the session worker: cursors now only advance after a successful parquet rename, so a daemon killed mid-flush replays the last flushed uuid via the supervisor's `?after=<uuid>` query — backfill catches up, dedup collapses any overlap, and there are no torn writes on disk. - src/gascity/schema.js — `gascity_messages` columns + cell coercion - src/gascity/dedup.js — bounded LRU dedup keyed by (city, session, uuid) - src/gascity/parquet_writer.js — buffer/flush/cursor lifecycle - src/gascity/passthrough.js — raw_frame normalizer for unknown providers - src/gascity/backfill.js — one-shot `/transcript?after=` resume per cursor - src/gascity/{index,normalizer_dispatcher,session_worker,supervisor_subscriber,paths,types}.js — wiring Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * feat(gascity): codex provider normalizer + provider dispatch (co-276c.4) (#108) Adds codexNormalize to the gascity normalizer dispatcher, registered alongside claudeNormalize via registerProductionNormalizers. Mirrors bead 2's contract: pure function, no I/O, returns NormalizedRow[] from a raw Codex CLI session frame; everything we don't understand round-trips through raw_frame + attributes. Codex JSONL session logs live at ~/.codex/sessions/YYYY/MM/DD/rollout-*.jsonl with five outer types (session_meta, turn_context, response_item, event_msg, compacted). The normalizer handles all five and maps every nested response_item / event_msg subtype I found in a real session. Key mapping decisions (documented inline + in types.d.ts): - response_item.message.content[] -> N rows of part_type=text, role + phase in attributes (mirrors Claude's multi-block grain). - response_item.function_call -> tool_use; arguments JSON-string parsed into tool_args, call_id -> tool_call_id. - response_item.custom_tool_call -> tool_use; free-form input (apply_patch diff) preserved verbatim in tool_args. - response_item.*_output -> tool_result with tool_result_for linking back to the originating call_id. exec_command outputs are unwrapped to detect exit_code != 0 -> is_error=true. - response_item.reasoning -> thinking; encrypted_content + summary stay in attributes (encrypted by default; readable text drops into content_text when present). - event_msg.token_count.last_token_usage -> input_tokens / output_tokens / cache_read_input_tokens hoisted onto Claude-shaped slots; cached_input_tokens maps to cache_read_input_tokens; reasoning + total counters live under attributes.info. - event_msg.patch_apply_end -> patch_apply_end row with is_error = !success, tool_result_for = call_id; unified diff under attributes.changes. - session_meta hoist -> cwd, git_branch, entrypoint (source), client_version (cli_version); base instructions + git metadata under attributes. - turn_context hoist -> model, cwd; permission_mode left NULL per spec ("don't invent placeholders"); approval_policy / sandbox_policy stay in attributes. Provider_uuid synthesis: Codex frames carry no per-frame uuid, but the schema requires one (writer dedup key). We sha-1 the JSON-serialised frame to produce a stable 40-hex uuid; re-reading the same JSONL line yields the same hash, so a restart + backfill collapses cleanly through the writer's existing dedup. Multi-block frames mirror Claude: N rows share provider_uuid with distinct part_index. The writer's current dedup is keyed only on provider_uuid (parquet_writer.js:200-211), so sibling rows beyond the first are dropped on disk today -- a pre-existing limitation that affects Claude equally and should be fixed in a follow-up bead (dedup should observe `(provider_uuid, part_index)`). Tests: 36 new tests in test/gascity/normalizers/codex.test.js driven by 39 sanitized real frames captured from two real Codex sessions (IDs, paths, encrypted blobs, commit hashes mapped to deterministic tokens; structural shape preserved). Fixtures cover every outer type plus every response_item + event_msg subtype I saw in the wild. Failure-mode tests cover non-object input, unknown frame types, missing timestamps, non-JSON tool arguments. Tool-linkage round-trip test confirms function_call.call_id matches function_call_output .tool_result_for, same for custom_tool_call. Coverage: 87 test files / 1439 tests pass. Typecheck clean. Lint warnings unchanged from baseline (codex.js + codex.test.js carry the same no-extra-parens warnings as claude.js/claude.test.js). End-to-end smoke test (manual, against the real dispatcher): dispatch({provider:"codex", ...function_call_frame}, ctx) -> [{part_type:"tool_use", tool_name:"exec_command", ...}] dispatch({provider:"claude", ...assistant_frame}, ctx) -> [{part_type:"text", model:"claude-opus-4-7", ...}] Confirms mixed-provider routing through the same dispatcher works. Out of scope per bead spec: gemini normalizer (deferred), CLI surface (bead 5), catalog registration (bead 6). Files: - src/gascity/normalizers/codex.js (new) - src/gascity/normalizers/index.js (register codex) - src/gascity/normalizers/types.d.ts (part_type vocab doc) - test/gascity/normalizers/codex.test.js (new) - test/fixtures/gascity/codex/*.jsonl (19 fixture files) Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * feat(gascity): ctvs gascity CLI subcommands + SIGHUP local reload (co-276c.5) (#109) Adds the `ctvs gascity` subcommand surface (attach/detach/list/backfill/status) and wires it into a minimal SIGHUP-driven control channel for the standalone daemon. Per-city diff lets attach/detach apply without retiring unrelated cities. Runtime state file backs `list` and `status` without an IPC round trip. - src/runtime/{paths,pid_file}.js: PID file lifecycle (write at boot, remove on shutdown, kill -0 staleness probe). - src/gascity/runtime_state.js: GascityRuntimeStateWriter (debounced, atomic tmp+rename) and readRuntimeState(); plumbed through subscriber + worker so list/status reflect live capture progress. - src/cli.js: SIGHUP wiring re-reads local config, runs existing applyDiff, plus a new applyGascitySectionDiff that calls listener.applyCityDiff() for in-place per-city changes (no thundering retire-all on attach/detach). - src/gascity/index.js: startGascitySource returns an enriched listener with applyCityDiff(newCities), tracks subscribers by name; hot reload preserves unchanged-city state. - src/gascity/supervisor_subscriber.js: stop({removeFromState}) lets a hot-reload removal drop a city from the runtime snapshot; otherwise daemon shutdown leaves it. - src/cli/gascity.js: subcommands. - attach: edit JSON config, infer name/api_url from city.toml when given a path, send SIGHUP via PID file, optionally block on lifecycle event (10s). - detach: remove entry, SIGHUP, daemon retires sessions and drops from state. - list: read runtime state, render table or --json. - backfill: walk cursors directly via backfillSession; --since (cursor age), --all (include retired), idempotent via writer dedup set. - status: per-city reachability probe + lifecycle/session/frame summary, --json for scripts. Tests: +110 tests across runtime/pid_file, gascity/runtime_state, cli/gascity, cli/local_reload (PID file, SIGHUP-driven gascity-section reload). Full suite: 1476/1476 pass; typecheck clean; lint matches baseline (0 errors). Out-of-scope per bead spec: catalog registration + skill doc updates (bead 6). Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * feat(gascity): register gascity_messages in query catalog + ctvs-gascity skill (co-276c.6) (#110) Wires the daemon-owned `~/.collectivus/sink/gascity_messages/` parquet store into `ctvs query` as the new `gascity_messages` logical dataset and teaches the `ctvs-gascity` skill how to use it alongside `events`, `session_segments`, and the wire-level `proxy_messages`. - `src/query/schema.js`: add `gascity_messages` to QUERY_DATASETS with the GASCITY_MESSAGES_COLUMNS list (47 cols including agent identity + token hoist) and `gascity` source signal. - `src/query/paths.js`: `discoverGascityPartitions` walks the Hive layout `date=YYYY-MM-DD/city=<name>/part-*.parquet` and yields one CachePartition per part-file. Routed into `expectedCachePartitions` ahead of the JSONL flow so gascity bypasses the parquet-cache layer entirely. Partitions are always-fresh (no `.meta.json` sidecar; the daemon owns writes). - `src/query/sql.js`: `parquetDataSource` skips the `numRows` hint for gascity so squirreling falls back to scan-based COUNT(*) instead of reading 0 from the absent meta sidecar. - `src/query/refresh.js`: `refreshQueryCache` reports gascity partitions as already-fresh (no-op) — the source IS the parquet, nothing to materialize. - `src/cli/query.js`: `catalog`/`status`/`doctor`/`ensureCacheReady` know about gascity — `source_signal` derives from sourceSignalForDataset(), status counts gascity part-files as both source and fresh cache, cache-disabled gates only fire for cache-backed datasets. - `src/cli/init_presets/gascity_skill.md`: third table section with schema highlights, gascity-native example queries (mayor tool calls today, token usage by rig with cache breakdown, tool-result errors, cross-source UNION with proxy_messages), and a "when to use which source" guide. - `README.md` + `skills/collectivus-query/`: short gascity source section and dataset listings updated. - `test/cli/query.test.js`: 6 new tests using the real ParquetWriter to produce fixture sinks under HOME-swapped tmp dirs — catalog/schema, count + group-by SQL, cache-disabled gascity-only queries, refresh no-op, cross-source UNION with proxy_messages, --gateway-id semantics. Acceptance verified: - `ctvs query catalog --format markdown` lists gascity_messages with source_signal=gascity, columns=50, source_partitions counted from disk. - `ctvs query schema gascity_messages --format markdown` returns the full v1 schema (47 typed columns). - All bead-3 example queries run cleanly against fixture data; cross-source UNION uses `part_type` instead of `role` since gascity_messages has no role column (deviation from the bead description, documented in the skill doc). Out of scope (deferred per bead): gemini provider, retention/TTL on parquet store, web dashboard integration. Closes co-276c.6 Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1 parent 2210a5d commit 007557a

104 files changed

Lines changed: 11643 additions & 40 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

README.md

Lines changed: 41 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -105,6 +105,16 @@ tail -f "collectivus-data/$USER/proxy/$(date -u +%F).jsonl"
105105

106106
Full step-by-step: [`docs/walkthrough-claude-code.md`](docs/walkthrough-claude-code.md).
107107

108+
To capture agent-attributed transcripts from a gascity supervisor (separate
109+
from the proxy capture above), attach a city to the same daemon:
110+
111+
```bash
112+
ctvs gascity attach hyptown --api-url http://127.0.0.1:8372
113+
ctvs query sql "select gascity_template, count(*) as parts from gascity_messages group by 1 order by parts desc"
114+
```
115+
116+
See [Gascity source (`gascity_messages`)](#gascity-source-gascity_messages) below.
117+
108118
## Configuration
109119

110120
Pass a JSON config with `--config <path>` (a local path or url). The schema:
@@ -552,7 +562,7 @@ outdated data).
552562
> unchanged; the new warning is written only to stderr. `missing`
553563
> partitions still error.
554564
555-
Logical datasets are `logs`, `traces`, `metrics`, and `proxy_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.
565+
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.
556566

557567
### Conversation log model
558568

@@ -566,6 +576,36 @@ JSON columns (`attributes`, `status`, `tools`, `tool_args`) carry sparse structu
566576

567577
For the full per-column derivation table see [skills/collectivus-query/references/query-cli.md](skills/collectivus-query/references/query-cli.md).
568578

579+
### Gascity source (`gascity_messages`)
580+
581+
`ctvs gascity` is a separate listener that subscribes to a gascity supervisor's
582+
REST API, normalizes provider frames (Claude / Codex), and writes one row per
583+
content block (text / thinking / tool_use / tool_result / attachment) directly
584+
to Parquet at `~/.collectivus/sink/gascity_messages/date=<YYYY-MM-DD>/city=<name>/`.
585+
There is no JSONL stage and no `.meta.json` sidecar: the sink IS the queryable
586+
store, so `ctvs query gascity_messages` is always reading what the daemon has
587+
flushed up to the moment of the call.
588+
589+
```bash
590+
ctvs gascity attach hyptown --api-url http://127.0.0.1:8372
591+
ctvs gascity list
592+
ctvs query schema gascity_messages --format markdown
593+
ctvs query sql "select gascity_template, count(*) from gascity_messages group by 1"
594+
```
595+
596+
`gascity_messages` carries agent-identity columns the proxy can't see —
597+
`gascity_template`, `gascity_rig`, `gascity_alias` — plus per-frame token usage
598+
with cache breakdown (`input_tokens`, `cache_read_input_tokens`,
599+
`cache_creation_input_tokens`). Use it when you need agent-attributed cost
600+
analysis or tool-call inspection; use `proxy_messages` for HTTP-level retry
601+
visibility and request timing. They UNION cleanly via `gateway_id` (a constant
602+
`gascity-scribe` on every gascity row tags the source).
603+
604+
The bundled [`ctvs-gascity` skill](src/cli/init_presets/gascity_skill.md) — installed per-workspace by
605+
`ctvs init gascity` — teaches Claude Code and Codex how to query all three
606+
gascity-aware tables (`events`, `session_segments`, `gascity_messages`) and
607+
their cross-source joins with `proxy_messages`.
608+
569609
### LLM skill
570610

571611
Install the bundled `collectivus-query` skill so Claude Code and Codex know how

bin/cli.js

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22

33
import process from 'node:process'
44

5-
const SUBCOMMANDS = new Set(['install', 'uninstall', 'attach', 'detach', 'status', 'config', 'admin', 'invite', 'export', 'query', 'collect', 'rendezvous', 'join', 'skills', 'claude-hook', 'init', 'ignore'])
5+
const SUBCOMMANDS = new Set(['install', 'uninstall', 'attach', 'detach', 'status', 'config', 'admin', 'invite', 'export', 'query', 'collect', 'rendezvous', 'join', 'skills', 'claude-hook', 'init', 'gascity', 'ignore'])
66

77
const argv = process.argv.slice(2)
88
const subcommand = argv[0]
@@ -108,6 +108,10 @@ async function loadSubcommand(name) {
108108
const { runInitSubcommand } = await import('../src/cli/init.js')
109109
return runInitSubcommand
110110
}
111+
case 'gascity': {
112+
const { runGascity } = await import('../src/cli/gascity.js')
113+
return runGascity
114+
}
111115
case 'ignore': {
112116
const { runIgnore } = await import('../src/cli/ignore.js')
113117
return runIgnore

skills/collectivus-query/SKILL.md

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -107,7 +107,7 @@ Use `JSON_VALUE(<col>, '$.path')` to extract scalars from the `attributes` / `st
107107
- 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.
108108
- Do not paste `--config` into every command by habit. Use it when discovery shows the service is not using `~/.hyp/collectivus.json`.
109109
- Do not read arbitrary Parquet files directly for `ctvs query sql`; the CLI only allows logical tables.
110-
- Keep SQL read-only and use only logical datasets: `logs`, `traces`, `metrics`, `proxy_messages`, and registered collection tables from `ctvs query catalog`.
110+
- Keep SQL read-only and use only logical datasets: `logs`, `traces`, `metrics`, `proxy_messages`, `gascity_messages`, and registered collection tables from `ctvs query catalog`.
111111
- Use UTC dates with `--date YYYY-MM-DD`.
112112
- Use `--service`, `--gateway-id`, `--from`, `--to`, or `--since` to narrow broad investigations.
113113

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -80,6 +80,7 @@ Collection tables always include `_ctvs_source_path`, `_ctvs_line_number`, and `
8080
- `traces`: OTLP spans. Common columns include `gateway_id`, `date`, `traceId`, `spanId`, `parentSpanId`, `name`, `kind`, `startTimestamp`, `endTimestamp`, `durationMs`, `status`, `serviceName`, `resource`, `scope`, and `attributes`.
8181
- `metrics`: OTLP metric points. Common columns include `gateway_id`, `date`, `metricName`, `metricType`, `timestamp`, `startTimestamp`, `serviceName`, `value`, `valueInt`, `count`, `sum`, `unit`, `resource`, `scope`, and `attributes`.
8282
- `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.
83+
- `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.
8384

8485
Run `ctvs query schema <dataset> --format json` for the exact columns in the installed version.
8586

0 commit comments

Comments
 (0)