chore(telemetry): collector accepts runtime schema 6 and keeps schemas 1 to 5

Schema 6 (notice 8) carries the exact key sets the client slice will send: model_usage counters gain requests, reasoning_tokens and generation_secs, and model_usage gains a context object of four input-size request counts; the session_start event gains install_age; exec sources gain fleet_worker, agent and scripted; exit_class gains unclean. Each is accepted on v6 only, so a v1 to v5 batch is validated exactly as before and a v4 batch carrying any v6 key or value is still refused. The preUsage shortcut becomes one explicit boolean per version. Rows append: model_usage double12..18 (0 on v4 rows, blob19 tells them apart) and install_age in blob12 of session_start rows. generation_secs is a whole-second session sum rather than milliseconds, so a one-request session does not reveal an exact request duration. PostHog forwarding is untouched, so v6 stays first-party only. report-funnel reads v4 and v6 model usage and excludes fleet_worker, agent and scripted starts as it does ci. A hand-authored crates/telemetry/tests/golden/v6.json is the contract fixture. The doc-welded constants (SCHEMA_VERSION, EVENT_FIELDS, SESSION_SOURCES, EXIT_CLASSES) stay at their v4 values until the client slice updates docs/TELEMETRY.md; the v6 sets are separate constants. Checks: telemetry-ingest npm test 225 passed 0 failed (8 files), tsc --noEmit clean, wrangler dev --local answered 204 to v4.json and v6.json and 400 to a v6 batch with a free-text install_age, wrangler deploy --dry-run ok, cargo fmt --all -- --check ok, scripts/preflight.sh --check OK. Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
2026-11-12 18:18:10 -10:00 · 2026-11-12 17:38:53 -08:00
8 changed files with 403 additions and 40 deletions
@@ -0,0 +1,127 @@
+{
+  "schema_version": 6,
+  "notice_version": 8,
+  "sent_at": "2026-08-03T18:04:11Z",
+  "install_id": "3f2a9c1e-0000-4000-8000-000000000001",
+  "app_version": "0.9.4",
+  "git_sha": "abcdef012345",
+  "surface": "control-plane",
+  "os": "macos",
+  "arch": "aarch64",
+  "libc": "none",
+  "tty": true,
+  "events": [
+    {
+      "event": "install_or_upgrade",
+      "kind": "upgrade",
+      "previous_version": "0.9.3-rc.1"
+    },
+    {
+      "event": "session_start",
+      "source": "fleet_worker",
+      "install_age": "under_week"
+    },
+    {
+      "event": "session_end",
+      "duration_bucket": "1m_10m",
+      "exit_class": "unclean",
+      "cold_start_bucket": "250_1000",
+      "providers": [
+        "custom",
+        "deepseek"
+      ],
+      "counters": {
+        "turns": 14,
+        "tool_calls": 61,
+        "fleet_dispatch": 0,
+        "workflow_run": 0,
+        "subagent_spawn": 2,
+        "mcp_server_connected": 0,
+        "memory_search": 0,
+        "approval_modal_shown": 0,
+        "approval_auto_allowed": 0,
+        "command_palette_open": 3
+      },
+      "errors": {
+        "auth_preflight_failed": 0,
+        "provider_http_4xx": 0,
+        "provider_http_5xx": 1,
+        "tool_denied_by_policy": 0,
+        "tool_timeout": 0,
+        "network_error": 0
+      },
+      "turn_wall": {
+        "lt_5s": 9,
+        "5_30s": 4,
+        "30_120s": 1,
+        "gte_120s": 0
+      }
+    },
+    {
+      "event": "panic",
+      "site": "crates/tui/src/tui/ui.rs:8801:17"
+    },
+    {
+      "event": "product_usage",
+      "counters": {
+        "page_view": 1,
+        "docs_view": 0,
+        "install_copy": 0,
+        "download": 0,
+        "signup": 0,
+        "login": 0,
+        "session_create": 0,
+        "session_resume": 0,
+        "turn_submit": 0,
+        "turn_complete": 0,
+        "settings_open": 0,
+        "integration_connect": 0,
+        "error_shown": 0
+      }
+    },
+    {
+      "event": "operations_summary",
+      "requests": 10,
+      "errors": 1,
+      "duration_ms_total": 1500,
+      "duration_ms_max": 300,
+      "probes": 2,
+      "probes_failed": 0
+    },
+    {
+      "event": "model_usage",
+      "provider": "deepseek",
+      "model": "other",
+      "counters": {
+        "turns_ok": 12,
+        "turns_failed": 1,
+        "turns_interrupted": 0,
+        "input_tokens": 418230,
+        "output_tokens": 9120,
+        "cache_read_tokens": 380001,
+        "cache_write_tokens": 0,
+        "requests": 27,
+        "reasoning_tokens": 2310,
+        "generation_secs": 412
+      },
+      "turn_wall": {
+        "lt_5s": 3,
+        "5_30s": 8,
+        "30_120s": 2,
+        "gte_120s": 0
+      },
+      "context": {
+        "lt_32k": 9,
+        "32k_128k": 11,
+        "128k_256k": 5,
+        "gte_256k": 2
+      }
+    },
+    {
+      "event": "tool_usage",
+      "tool": "read_file",
+      "calls": 41,
+      "errors": 2
+    }
+  ]
+}
@@ -15,9 +15,9 @@ hostname is the only way in.

 The current Engine source defaults usage analytics on under notice version 6, with a
 clear opt-out. Prior explicit declines remain off; a missing acceptance record
-is not treated as consent. Schema 4 carries `notice_version: 6`; schema 3
+is not treated as consent. Schema 6 carries `notice_version: 8` and schema 4 carries `notice_version: 6`; schema 3
 uses notice 5, and legacy schema 2 retains its original explicit
-`consent_version: 4` contract. Schema 1 and schema 4 batches are first-party
+`consent_version: 4` contract. Schema 1, 4 and 6 batches are first-party
 only in this source. Local validation is not proof this updated Worker has been deployed. A user who wants
 to contact nobody sets `telemetry_endpoint = ""`, which writes batches to
 `$CODEWHALE_HOME/telemetry/dryrun.jsonl` instead.
@@ -38,7 +38,7 @@ require separate operator approval; none is established by the local tests.

 Validated schema 3 batches carrying `notice_version: 5` and legacy schema 2
 batches carrying explicit `consent_version: 4` can reach the processor. Schema 1
-and schema 4 batches remain first-party only. Capture uses
+and schema 4 and 6 batches remain first-party only. Capture uses
 `/batch/`, fixed anonymous/no-geo controls, no incoming headers, no cookies,
 no redirects, no response parsing, no retries, and a 1.5-second timeout.
 PostHog failure is isolated from the successful Analytics Engine write.
@@ -143,20 +143,21 @@ take the next free slot.
 | `blob8` | `tty` — `'true'` \| `'false'` |
 | `blob9` | `install_kind` (`install_or_upgrade` only) |
 | `blob10` | `previous_version` (`install_or_upgrade` only) |
-| `blob11` | `session_source` (`session_start` only) |
-| `blob12` | `duration_bucket` (`session_end` only) |
-| `blob13` | `exit_class` (`session_end` only) |
+| `blob11` | `session_source` (`session_start` only; schema 6 adds `fleet_worker`, `agent`, `scripted`) |
+| `blob12` | `duration_bucket` (`session_end`); `install_age` (`session_start`, schema 6 only: `new_today`, `under_week`, `older`, `''` for null or older schemas) |
+| `blob13` | `exit_class` (`session_end` only; schema 6 adds `unclean`) |
 | `blob14` | `cold_start_bucket` (`session_end` only; `''` on surfaces that do not measure it) |
 | `blob15` | `providers`, comma-joined, already sorted and deduplicated; the one `provider` on `model_usage` |
 | `blob16` | `panic_site` (`panic`) — a `crates/…` path or the literal `<dep>`; `model` (`model_usage`); `tool` (`tool_usage`) |
 | `blob17` | `sent_at` — the *batch* timestamp. Events carry none. |
 | `blob18` | `aggregate_counters` — closed JSON counts, for `product_usage` or `operations_summary` |
 | `blob19` | `schema_version` |
-| `blob20` | `privacy_version`: notice6 for v4, notice5 for v3, consent4 for v2, empty for v1; `blob19` keeps their meanings distinct |
+| `blob20` | `privacy_version`: notice8 for v6, notice7 for the v5 website funnel, notice6 for v4, notice5 for v3, consent4 for v2, empty for v1; `blob19` keeps their meanings distinct |
 | `double1..10` | `counters`: `turns`, `tool_calls`, `fleet_dispatch`, `workflow_run`, `subagent_spawn`, `mcp_server_connected`, `memory_search`, `approval_modal_shown`, `approval_auto_allowed`, `command_palette_open` |
 | `double11..16` | `errors`: `auth_preflight_failed`, `provider_http_4xx`, `provider_http_5xx`, `tool_denied_by_policy`, `tool_timeout`, `network_error` |
 | `double17..20` | `turn_wall`: `lt_5s`, `5_30s`, `30_120s`, `gte_120s` |
 | `double1..11` on `model_usage` | `turns_ok`, `turns_failed`, `turns_interrupted`, `input_tokens`, `output_tokens`, `cache_read_tokens`, `cache_write_tokens`, then `turn_wall` `lt_5s`, `5_30s`, `30_120s`, `gte_120s` |
+| `double12..18` on `model_usage` | schema 6 only, `0` on v4 rows (`blob19` tells them apart): `requests`, `reasoning_tokens`, `generation_secs` (whole seconds, session sum), then input-size request counts `lt_32k`, `32k_128k`, `128k_256k`, `gte_256k` |
 | `double1..2` on `tool_usage` | `calls`, `errors` |

 Columns not relevant to an event are `''` / `0`. `tty` is a blob because the 20
@@ -13,7 +13,7 @@ export const CAVEATS = [
  "Runtime install means first initialized enabled telemetry state, not all installations or new people. IDs rotate every 90 days and are removed on opt-out.",
  "Native success is observed model turns_ok, not a first-ever primary or useful task. Missing usage, same-process child activity and late cancellation can affect it.",
  "Dates are received batch dates rather than exact event times. Offline, opted-out and dropped deliveries are absent; ingest has no replay deduplication guarantee.",
-  "Native IDs with an observed CI start are excluded in full because the aggregate schema cannot assign model counts to individual sessions.",
+  "Native IDs with an observed CI, fleet worker, agent or scripted start are excluded in full because the aggregate schema cannot assign model counts to individual sessions.",
 ];

 function instant(value, field) {
@@ -72,7 +72,7 @@ export function buildFunnelReport(input, { now = new Date() } = {}) {
  const installs = new Map(); const successes = new Map(); const started = new Set(); const ci = new Set();
  const uuid = /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/;
  for (const row of input.native_rows) {
-    if (!uuid.test(row.install_id) || !NATIVE_SURFACES.includes(row.surface) || ![1, 2, 3, 4].includes(count(row.schema_version, "schema_version"))) throw new Error("invalid native identity, surface or version");
+    if (!uuid.test(row.install_id) || !NATIVE_SURFACES.includes(row.surface) || ![1, 2, 3, 4, 6].includes(count(row.schema_version, "schema_version"))) throw new Error("invalid native identity, surface or version");
    const at = instant(fresh(row.received_at), "received_at");
    if (!within(at, from, to)) throw new Error("native row outside coverage");
    if (newest === null || at > newest) newest = at;
@@ -82,11 +82,11 @@ export function buildFunnelReport(input, { now = new Date() } = {}) {
      if (!["install", "upgrade", "downgrade"].includes(row.install_kind)) throw new Error("invalid install kind");
      if (row.install_kind === "install" && (!installs.has(row.install_id) || at < installs.get(row.install_id))) installs.set(row.install_id, at);
    } else if (row.event === "session_start") {
-      if (!["interactive", "resume", "fork", "api", "unknown", "ci"].includes(row.session_source)) throw new Error("invalid session source");
-      if (row.session_source === "ci") ci.add(row.install_id);
+      if (!["interactive", "resume", "fork", "api", "unknown", "ci", "fleet_worker", "agent", "scripted"].includes(row.session_source)) throw new Error("invalid session source");
+      if (["ci", "fleet_worker", "agent", "scripted"].includes(row.session_source)) ci.add(row.install_id);
      started.add(row.install_id);
    } else if (row.event === "model_usage") {
-      if (count(row.schema_version, "schema_version") !== 4) throw new Error("model usage requires v4");
+      if (![4, 6].includes(count(row.schema_version, "schema_version"))) throw new Error("model usage requires v4 or v6");
      if (count(row.turns_ok, "turns_ok") > 0) {
        const observations = successes.get(row.install_id) ?? []; observations.push(at); successes.set(row.install_id, observations);
      }
@@ -24,7 +24,14 @@
 */

 import type { Batch, Event } from "./schema";
-import { COUNTER_FIELDS, ERROR_FIELDS, MODEL_COUNTER_FIELDS, TURN_WALL_FIELDS } from "./schema";
+import {
+  CONTEXT_FIELDS,
+  COUNTER_FIELDS,
+  ERROR_FIELDS,
+  MODEL_COUNTER_FIELDS,
+  MODEL_COUNTER_FIELDS_V6_EXTRA,
+  TURN_WALL_FIELDS,
+} from "./schema";

 /** The subset of `AnalyticsEngineDataset` this Worker uses. */
 export interface DataPointSink {
@@ -58,7 +65,7 @@ export const BLOB_COLUMNS = [
  "install_kind", // blob9
  "previous_version", // blob10
  "session_source", // blob11
-  "duration_bucket", // blob12
+  "duration_bucket", // blob12  session_end; install_age on session_start (schema 6)
  "exit_class", // blob13
  "cold_start_bucket", // blob14
  "providers", // blob15 comma-joined, already sorted and deduplicated
@@ -80,6 +87,19 @@ export const DOUBLE_COLUMNS = [
  ...TURN_WALL_FIELDS,
 ] as const;

+/**
+ * `model_usage` doubles, in `double1..double20` order: the seven v4 counters,
+ * the four turn-wall buckets, then (schema 6, appended so v4 queries keep their
+ * meaning) `requests`, `reasoning_tokens`, `generation_secs` and the four
+ * input-size buckets. v4 rows read 0 in `double12..double18`.
+ */
+export const MODEL_USAGE_DOUBLE_COLUMNS = [
+  ...MODEL_COUNTER_FIELDS,
+  ...TURN_WALL_FIELDS,
+  ...MODEL_COUNTER_FIELDS_V6_EXTRA,
+  ...CONTEXT_FIELDS,
+] as const;
+
 const EMPTY_DOUBLES: number[] = DOUBLE_COLUMNS.map(() => 0);

 /** Build the rows for one validated batch. */
@@ -111,6 +131,7 @@ function toDataPoint(batch: Batch, event: Event): DataPoint {
      break;
    case "session_start":
      blobs[10] = event.source;
+      blobs[11] = event.install_age ?? "";
      break;
    case "session_end":
      blobs[11] = event.duration_bucket;
@@ -141,8 +162,10 @@ function toDataPoint(batch: Batch, event: Event): DataPoint {
      doubles = [
        ...MODEL_COUNTER_FIELDS.map((field) => event.counters[field]),
        ...TURN_WALL_FIELDS.map((field) => event.turn_wall[field]),
-        ...EMPTY_DOUBLES.slice(MODEL_COUNTER_FIELDS.length + TURN_WALL_FIELDS.length),
+        ...MODEL_COUNTER_FIELDS_V6_EXTRA.map((field) => event.counters[field] ?? 0),
+        ...CONTEXT_FIELDS.map((field) => event.context?.[field] ?? 0),
      ];
+      doubles = [...doubles, ...EMPTY_DOUBLES.slice(doubles.length)];
      break;
    case "tool_usage":
      blobs[15] = event.tool;
@@ -22,6 +22,9 @@ export const SCHEMA_VERSION = 4;
 /** Notice-v5 batches without `model_usage`, `tool_usage` or the `ci` source. */
 export const V3_SCHEMA_VERSION = 3;

+/** Runtime schema 6: request aggregates, input-size buckets and exec labels. */
+export const V6_SCHEMA_VERSION = 6;
+
 export const FUNNEL_SCHEMA_VERSION = 5;
 export const FUNNEL_NOTICE_VERSION = 7;
 export const FUNNEL_COUNTER_FIELDS = [
@@ -34,6 +37,9 @@ export const FUNNEL_COUNTER_FIELDS = [
 /** Disclosed default-on policy, not a record of human acceptance. */
 export const NOTICE_VERSION = 6;

+/** The notice version schema v6 batches carry. */
+export const V6_NOTICE_VERSION = 8;
+
 /** The notice version schema v3 batches carry. */
 export const V3_NOTICE_VERSION = 5;

@@ -146,6 +152,13 @@ export const SESSION_SOURCES = [
  "ci",
 ] as const;

+/** Exec start labels and the install-age bucket that only schema 6 carries. */
+export const V6_ONLY_SESSION_SOURCES = ["fleet_worker", "agent", "scripted"] as const;
+export const V6_SESSION_SOURCES = [...SESSION_SOURCES, ...V6_ONLY_SESSION_SOURCES] as const;
+
+/** `InstallAge` wire spellings; the field is also `null` when state is unreadable. */
+export const INSTALL_AGES = ["new_today", "under_week", "older"] as const;
+
 /** `DurationBucket` wire spellings. */
 export const DURATION_BUCKETS = [
  "lt_1m",
@@ -157,6 +170,9 @@ export const DURATION_BUCKETS = [
 /** `ExitClass::as_str`. */
 export const EXIT_CLASSES = ["clean", "signal", "panic", "error"] as const;

+/** Schema 6 adds the class for a session recovered from a local checkpoint. */
+export const V6_EXIT_CLASSES = [...EXIT_CLASSES, "unclean"] as const;
+
 /** `ColdStartBucket` wire spellings. */
 export const COLD_START_BUCKETS = [
  "lt_250",
@@ -211,6 +227,21 @@ export const MODEL_COUNTER_FIELDS = [
  "cache_write_tokens",
 ] as const;

+/** Schema 6 appends these to `MODEL_COUNTER_FIELDS`; `generation_secs` is a whole-second session sum. */
+export const MODEL_COUNTER_FIELDS_V6_EXTRA = [
+  "requests",
+  "reasoning_tokens",
+  "generation_secs",
+] as const;
+
+export const V6_MODEL_COUNTER_FIELDS = [
+  ...MODEL_COUNTER_FIELDS,
+  ...MODEL_COUNTER_FIELDS_V6_EXTRA,
+] as const;
+
+/** `ContextSize` wire spellings: request counts by input size, half-open at 32K, 128K and 256K. */
+export const CONTEXT_FIELDS = ["lt_32k", "32k_128k", "128k_256k", "gte_256k"] as const;
+
 /** `event::TOOL_NAMES` plus the `mcp` and `other` buckets. */
 export const TOOL_NAMES = [
  "agent", "apply_patch", "authenticate", "bash", "code_execution", "create_goal",
@@ -286,6 +317,13 @@ export const EVENT_FIELDS: Readonly<Record<string, readonly string[]>> = {
  funnel_usage: ["event", "counters"],
 };

+/** The schema 6 key sets that differ from `EVENT_FIELDS`. */
+export const V6_EVENT_FIELDS: Readonly<Record<string, readonly string[]>> = {
+  ...EVENT_FIELDS,
+  session_start: ["event", "source", "install_age"],
+  model_usage: ["event", "provider", "model", "counters", "turn_wall", "context"],
+};
+
 /** Every event discriminant, for the doc-match test. */
 export const EVENT_NAMES = Object.keys(EVENT_FIELDS);

@@ -430,7 +468,7 @@ export type FunnelCounters = Record<(typeof FUNNEL_COUNTER_FIELDS)[number], numb

 export type Event =
  | { event: "install_or_upgrade"; kind: string; previous_version: string | null }
-  | { event: "session_start"; source: string }
+  | { event: "session_start"; source: string; install_age?: string | null }
  | {
      event: "session_end";
      duration_bucket: string;
@@ -448,8 +486,10 @@ export type Event =
      event: "model_usage";
      provider: string;
      model: string;
-      counters: Record<(typeof MODEL_COUNTER_FIELDS)[number], number>;
+      counters: Record<(typeof MODEL_COUNTER_FIELDS)[number], number> &
+        Partial<Record<(typeof MODEL_COUNTER_FIELDS_V6_EXTRA)[number], number>>;
      turn_wall: TurnWall;
+      context?: Record<(typeof CONTEXT_FIELDS)[number], number>;
    }
  | { event: "tool_usage"; tool: string; calls: number; errors: number }
  | ({ event: "operations_summary" } & Record<(typeof OPERATIONS_FIELDS)[number], number>);
@@ -546,13 +586,14 @@ function u32Map(
  return null;
 }

-function validateEvent(value: unknown, where: string): string | null {
+function validateEvent(value: unknown, where: string, v6 = false): string | null {
  if (!isPlainObject(value)) return `${where}: not an object`;
  const name = value.event;
-  if (typeof name !== "string" || !Object.hasOwn(EVENT_FIELDS, name)) {
+  const eventFields = v6 ? V6_EVENT_FIELDS : EVENT_FIELDS;
+  if (typeof name !== "string" || !Object.hasOwn(eventFields, name)) {
    return `${where}: unknown event discriminant`;
  }
-  const keyError = keysExactly(value, EVENT_FIELDS[name], `${where}(${name})`);
+  const keyError = keysExactly(value, eventFields[name], `${where}(${name})`);
  if (keyError) return keyError;

  switch (name) {
@@ -575,8 +616,15 @@ function validateEvent(value: unknown, where: string): string | null {
      }
      return null;
    }
-    case "session_start":
-      return enumString(value.source, SESSION_SOURCES, `${where}.source`);
+    case "session_start": {
+      const sourceError = enumString(
+        value.source,
+        v6 ? V6_SESSION_SOURCES : SESSION_SOURCES,
+        `${where}.source`,
+      );
+      if (sourceError || !v6 || value.install_age === null) return sourceError;
+      return enumString(value.install_age, INSTALL_AGES, `${where}.install_age`);
+    }
    case "session_end": {
      const bucketError = enumString(
        value.duration_bucket,
@@ -586,7 +634,7 @@ function validateEvent(value: unknown, where: string): string | null {
      if (bucketError) return bucketError;
      const exitError = enumString(
        value.exit_class,
-        EXIT_CLASSES,
+        v6 ? V6_EXIT_CLASSES : EXIT_CLASSES,
        `${where}.exit_class`,
      );
      if (exitError) return exitError;
@@ -638,9 +686,18 @@ function validateEvent(value: unknown, where: string): string | null {
      if (typeof value.model !== "string" || !MODEL_RE.test(value.model)) {
        return `${where}.model: not a catalogued model id`;
      }
+      const counterError = v6
+        ? countMap(value.counters, V6_MODEL_COUNTER_FIELDS, `${where}.counters`, Number.MAX_SAFE_INTEGER) ??
+          u32Map(
+            { requests: (value.counters as Record<string, unknown>).requests },
+            ["requests"],
+            `${where}.counters`,
+          )
+        : countMap(value.counters, MODEL_COUNTER_FIELDS, `${where}.counters`, Number.MAX_SAFE_INTEGER);
      return (
-        countMap(value.counters, MODEL_COUNTER_FIELDS, `${where}.counters`, Number.MAX_SAFE_INTEGER) ??
-        u32Map(value.turn_wall, TURN_WALL_FIELDS, `${where}.turn_wall`)
+        counterError ??
+        u32Map(value.turn_wall, TURN_WALL_FIELDS, `${where}.turn_wall`) ??
+        (v6 ? u32Map(value.context, CONTEXT_FIELDS, `${where}.context`) : null)
      );
    }
    case "tool_usage":
@@ -675,15 +732,12 @@ export function validateBatch(value: unknown): Accepted | Rejection {

  const legacy = value.schema_version === 1;
  const explicitConsent = value.schema_version === 2;
+  const v3 = value.schema_version === V3_SCHEMA_VERSION;
+  const v4 = value.schema_version === SCHEMA_VERSION;
  const funnel = value.schema_version === FUNNEL_SCHEMA_VERSION;
-  const preUsage = value.schema_version !== SCHEMA_VERSION && !funnel;
-  if (
-    !legacy &&
-    !explicitConsent &&
-    value.schema_version !== SCHEMA_VERSION &&
-    !funnel &&
-    value.schema_version !== V3_SCHEMA_VERSION
-  ) {
+  const v6 = value.schema_version === V6_SCHEMA_VERSION;
+  const preUsage = legacy || explicitConsent || v3;
+  if (!legacy && !explicitConsent && !v3 && !v4 && !funnel && !v6) {
    return { ok: false, reason: "batch.schema_version: unsupported" };
  }
  const fields = legacy ? LEGACY_ENVELOPE_FIELDS : explicitConsent ? V2_ENVELOPE_FIELDS : ENVELOPE_FIELDS;
@@ -695,7 +749,7 @@ export function validateBatch(value: unknown): Accepted | Rejection {
  if (
    !legacy &&
    !explicitConsent &&
-    value.notice_version !== (funnel ? FUNNEL_NOTICE_VERSION : preUsage ? V3_NOTICE_VERSION : NOTICE_VERSION)
+    value.notice_version !== (funnel ? FUNNEL_NOTICE_VERSION : v6 ? V6_NOTICE_VERSION : preUsage ? V3_NOTICE_VERSION : NOTICE_VERSION)
  ) {
    return { ok: false, reason: "batch.notice_version: current policy required" };
  }
@@ -762,7 +816,7 @@ export function validateBatch(value: unknown): Accepted | Rejection {
        return { ok: false, reason: "events: operations_summary requires control-plane surface" };
      }
    }
-    const eventError = validateEvent(event, `events[${index}]`);
+    const eventError = validateEvent(event, `events[${index}]`, v6);
    if (eventError) return { ok: false, reason: eventError };
  }

@@ -97,12 +97,12 @@ describe("additive website funnel contract", () => {
    }
  });

-  it("keeps all four historical goldens accepted and runtime v4 first-party only", async () => {
+  it("keeps the historical goldens and v6 accepted, with runtime v4 and v6 first-party only", async () => {
    const fetch = vi.fn(); vi.stubGlobal("fetch", fetch);
-    for (const version of [1, 2, 3, 4]) {
+    for (const version of [1, 2, 3, 4, 6]) {
      const batch = JSON.parse(readFileSync(new URL(`../../crates/telemetry/tests/golden/v${version}.json`, import.meta.url), "utf8"));
      expect(validateBatch(batch).ok, `v${version}`).toBe(true);
-      if (version !== 4) continue;
+      if (version !== 4 && version !== 6) continue;
      const { env, written } = harness();
      expect((await worker.fetch(postJson(batch), { ...env, POSTHOG_HOST: "https://us.i.posthog.com", POSTHOG_PROJECT_TOKEN: "phc_local_test_fixture", POSTHOG_IP_SAFE_EGRESS_VERIFIED: "true" })).status).toBe(204);
      expect(written).toHaveLength(batch.events.length);
@@ -4,8 +4,8 @@ import { describe, expect, it } from "vitest";

 import worker from "../src/index";
 import { INGEST_PATH } from "../src/route";
-import { BLOB_COLUMNS, DOUBLE_COLUMNS } from "../src/datapoint";
-import { MAX_BODY_BYTES, SURFACES } from "../src/schema";
+import { BLOB_COLUMNS, DOUBLE_COLUMNS, MODEL_USAGE_DOUBLE_COLUMNS } from "../src/datapoint";
+import { MAX_BODY_BYTES, SURFACES, validateBatch } from "../src/schema";
 import { goldenBatch, harness, post, postJson } from "./support";

 describe("method and route", () => {
@@ -473,3 +473,148 @@ describe("failing closed and quiet", () => {
    }
  });
 });
+
+describe("schema 6", () => {
+  const golden = (version: number): Record<string, any> =>
+    JSON.parse(
+      readFileSync(
+        new URL(`../../crates/telemetry/tests/golden/v${version}.json`, import.meta.url),
+        "utf8",
+      ),
+    );
+  const event = (batch: Record<string, any>, name: string) =>
+    batch.events.find((item: Record<string, any>) => item.event === name);
+
+  it("accepts every historical golden and the v6 golden", () => {
+    for (const version of [1, 2, 3, 4, 6]) {
+      expect(validateBatch(golden(version)).ok, `v${version}`).toBe(true);
+    }
+  });
+
+  it("answers 204 to the v4 and v6 goldens", async () => {
+    for (const version of [4, 6]) {
+      const { env, written } = harness();
+      expect((await worker.fetch(postJson(golden(version)), env)).status).toBe(204);
+      expect(written).toHaveLength(golden(version).events.length);
+    }
+  });
+
+  it("requires notice 8 on v6 and refuses the v4 notice", () => {
+    const batch = golden(6);
+    batch.notice_version = 6;
+    expect(validateBatch(batch).ok).toBe(false);
+    expect(validateBatch({ ...golden(4), notice_version: 8 }).ok).toBe(false);
+  });
+
+  it("refuses a v4 batch carrying any v6 key or value", () => {
+    const mutations: Array<(batch: Record<string, any>) => void> = [
+      (batch) => { event(batch, "model_usage").context = event(golden(6), "model_usage").context; },
+      (batch) => { event(batch, "model_usage").counters.requests = 1; },
+      (batch) => { event(batch, "model_usage").counters.reasoning_tokens = 1; },
+      (batch) => { event(batch, "model_usage").counters.generation_secs = 1; },
+      (batch) => { event(batch, "session_start").install_age = "older"; },
+      (batch) => { event(batch, "session_start").source = "fleet_worker"; },
+      (batch) => { event(batch, "session_start").source = "agent"; },
+      (batch) => { event(batch, "session_start").source = "scripted"; },
+      (batch) => { event(batch, "session_end").exit_class = "unclean"; },
+    ];
+    for (const [index, mutate] of mutations.entries()) {
+      const batch = golden(4);
+      mutate(batch);
+      expect(validateBatch(batch).ok, `mutation ${index}`).toBe(false);
+    }
+  });
+
+  it("refuses a v6 batch missing or adding a key", () => {
+    const mutations: Array<(batch: Record<string, any>) => void> = [
+      (batch) => { delete event(batch, "model_usage").context; },
+      (batch) => { delete event(batch, "model_usage").context["32k_128k"]; },
+      (batch) => { event(batch, "model_usage").context.gte_512k = 0; },
+      (batch) => { delete event(batch, "model_usage").counters.requests; },
+      (batch) => { delete event(batch, "model_usage").counters.reasoning_tokens; },
+      (batch) => { delete event(batch, "model_usage").counters.generation_secs; },
+      (batch) => { event(batch, "model_usage").counters.generation_ms = 1; },
+      (batch) => { event(batch, "model_usage").model_family = "x"; },
+      (batch) => { delete event(batch, "session_start").install_age; },
+      (batch) => { event(batch, "session_start").extra = "x"; },
+      (batch) => { event(batch, "session_end").counters.requests = 1; },
+    ];
+    for (const [index, mutate] of mutations.entries()) {
+      const batch = golden(6);
+      mutate(batch);
+      expect(validateBatch(batch).ok, `mutation ${index}`).toBe(false);
+    }
+  });
+
+  it("refuses open values where v6 has a closed set or a count", () => {
+    const mutations: Array<(batch: Record<string, any>) => void> = [
+      (batch) => { event(batch, "session_start").install_age = "2026-11-01"; },
+      (batch) => { event(batch, "session_start").install_age = "OLDER"; },
+      (batch) => { event(batch, "session_start").source = "github-actions"; },
+      (batch) => { event(batch, "session_end").exit_class = "killed"; },
+      (batch) => { event(batch, "model_usage").counters.requests = 4294967296; },
+      (batch) => { event(batch, "model_usage").counters.requests = 1.5; },
+      (batch) => { event(batch, "model_usage").counters.generation_secs = -1; },
+      (batch) => { event(batch, "model_usage").counters.reasoning_tokens = "7"; },
+      (batch) => { event(batch, "model_usage").context.lt_32k = 4294967296; },
+      (batch) => { event(batch, "model_usage").context.lt_32k = "1"; },
+      (batch) => { event(batch, "model_usage").model = "/Users/someone/model"; },
+    ];
+    for (const [index, mutate] of mutations.entries()) {
+      const batch = golden(6);
+      mutate(batch);
+      expect(validateBatch(batch).ok, `mutation ${index}`).toBe(false);
+    }
+  });
+
+  it("accepts a null install_age and every v6 source and exit class", () => {
+    for (const source of ["interactive", "resume", "fork", "api", "unknown", "ci", "fleet_worker", "agent", "scripted"]) {
+      for (const age of ["new_today", "under_week", "older", null]) {
+        const batch = golden(6);
+        event(batch, "session_start").source = source;
+        event(batch, "session_start").install_age = age;
+        expect(validateBatch(batch).ok, `${source}/${age}`).toBe(true);
+      }
+    }
+    for (const exit of ["clean", "signal", "panic", "error", "unclean"]) {
+      const batch = golden(6);
+      event(batch, "session_end").exit_class = exit;
+      expect(validateBatch(batch).ok, exit).toBe(true);
+    }
+  });
+
+  it("writes v6 model_usage after the unchanged eleven doubles, and leaves v4 rows zero there", async () => {
+    expect(MODEL_USAGE_DOUBLE_COLUMNS).toHaveLength(18);
+    const v6 = harness();
+    await worker.fetch(postJson(golden(6)), v6.env);
+    const row = v6.written.find((point) => point.blobs[0] === "model_usage")!;
+    expect(row.doubles).toEqual([
+      12, 1, 0, 418230, 9120, 380001, 0,
+      3, 8, 2, 0,
+      27, 2310, 412,
+      9, 11, 5, 2,
+      0, 0,
+    ]);
+    expect(row.blobs[BLOB_COLUMNS.indexOf("schema_version")]).toBe("6");
+
+    const v4 = harness();
+    await worker.fetch(postJson(golden(4)), v4.env);
+    const old = v4.written.find((point) => point.blobs[0] === "model_usage")!;
+    expect(old.doubles).toEqual([
+      12, 1, 0, 418230, 9120, 380001, 0,
+      3, 8, 2, 0,
+      0, 0, 0, 0, 0, 0, 0, 0, 0,
+    ]);
+  });
+
+  it("writes install_age to blob12 of session_start rows only", async () => {
+    const { env, written } = harness();
+    await worker.fetch(postJson(golden(6)), env);
+    const start = written.find((point) => point.blobs[0] === "session_start")!;
+    expect(start.blobs[BLOB_COLUMNS.indexOf("session_source")]).toBe("fleet_worker");
+    expect(start.blobs[BLOB_COLUMNS.indexOf("duration_bucket")]).toBe("under_week");
+    const end = written.find((point) => point.blobs[0] === "session_end")!;
+    expect(end.blobs[BLOB_COLUMNS.indexOf("exit_class")]).toBe("unclean");
+    expect(end.blobs[BLOB_COLUMNS.indexOf("duration_bucket")]).toBe("1m_10m");
+  });
+});
@@ -49,6 +49,19 @@ describe("existing-pipeline owner funnel report", () => {
    expect(buildFunnelReport(input, { now: NOW }).native.seven_day_initialization_cohort.unavailable_reason).toContain("incomplete");
  });

+  it("excludes fleet worker, agent and scripted starts like CI, and accepts v6 model usage", () => {
+    for (const source of ["fleet_worker", "agent", "scripted"]) {
+      const input = receipt();
+      input.native_rows.push({ ...native("session_start", "2026-10-04T00:00:00Z"), session_source: source });
+      input.native_rows.push({ ...native("model_usage", "2026-10-04T00:00:00Z"), schema_version: 6 });
+      const report = buildFunnelReport(input, { now: NOW });
+      expect(report.native.excluded_ci_ids, source).toBe(1);
+      expect(report.native.observed_completed_ids, source).toBe(0);
+    }
+    const input = receipt(); input.native_rows.push({ ...native("model_usage", "2026-10-04T00:00:00Z"), schema_version: 5 });
+    expect(() => buildFunnelReport(input, { now: NOW })).toThrow(/version/);
+  });
+
  it("does not compute a cohort fraction from sampled native records", () => {
    const input = receipt(); input.native_rows[0].weight = 4;
    expect(buildFunnelReport(input, { now: NOW }).native.seven_day_initialization_cohort).toMatchObject({ fraction: null, sampled: true });