perf(clickhouse): prepare task event attribute inserts (#4860)

## Summary

Prepare task event writes for a later storage schema change without
changing the table itself. Event attributes continue to be stored and
read exactly as before.

## Design

The event writer now names every insert column explicitly, ensuring
`attributes` remains part of the input when the column becomes
input-only. The remaining native attribute-path queries read the
existing `attributes_text` representation through JSON extraction
functions.

This PR does not alter the `task_events_v2` schema.

Author: carderne

apps/webapp/app/services/admin/missingLlmModels.server.ts

@@ -26,8 +26,14 @@ export async function getMissingLlmModels(
     name: "missingLlmModels",
     table: "trigger_dev.task_events_v2",
     columns: [
-      { name: "model", expression: "attributes.gen_ai.response.model.:String" },
-      { name: "system", expression: "attributes.gen_ai.system.:String" },
+      {
+        name: "model",
+        expression: "JSONExtractString(attributes_text, 'gen_ai', 'response', 'model')",
+      },
+      {
+        name: "system",
+        expression: "JSONExtractString(attributes_text, 'gen_ai', 'system')",
+      },
       { name: "cnt", expression: "count()" },
     ],
   });
@@ -39,10 +45,15 @@ export async function getMissingLlmModels(
   });
 
   // Only spans that have a model set
-  qb.where("attributes.gen_ai.response.model.:String != {empty: String}", { empty: "" });
+  qb.where("JSONExtractString(attributes_text, 'gen_ai', 'response', 'model') != {empty: String}", {
+    empty: "",
+  });
 
   // Only spans that were NOT cost-enriched (trigger.llm.total_cost is NULL)
-  qb.where("attributes.trigger.llm.total_cost.:Float64 IS NULL", {});
+  qb.where(
+    "JSONExtract(attributes_text, 'trigger', 'llm', 'total_cost', 'Nullable(Float64)') IS NULL",
+    {}
+  );
 
   // Only completed spans
   qb.where("kind = {kind: String}", { kind: "SPAN" });
@@ -107,8 +118,13 @@ export async function getMissingModelSamples(opts: {
   const qb = createBuilder();
 
   qb.where("inserted_at >= {since: DateTime64(3)}", { since: formatDateTime(since) });
-  qb.where("attributes.gen_ai.response.model.:String = {model: String}", { model: opts.model });
-  qb.where("attributes.trigger.llm.total_cost.:Float64 IS NULL", {});
+  qb.where("JSONExtractString(attributes_text, 'gen_ai', 'response', 'model') = {model: String}", {
+    model: opts.model,
+  });
+  qb.where(
+    "JSONExtract(attributes_text, 'trigger', 'llm', 'total_cost', 'Nullable(Float64)') IS NULL",
+    {}
+  );
   qb.where("kind = {kind: String}", { kind: "SPAN" });
   qb.where("status = {status: String}", { status: "OK" });
   qb.orderBy("start_time DESC");

apps/webapp/test/clickhouseEventRepositoryJsonRecovery.test.ts

@@ -89,7 +89,7 @@ describe("ClickhouseEventRepository JSON parse recovery", () => {
         const queryEvents = clickhouse.reader.query({
           name: "event-recovery-check",
           query:
-            "SELECT span_id, toJSONString(attributes) AS attributes_json FROM trigger_dev.task_events_v2 WHERE environment_id = {env_id:String}",
+            "SELECT span_id, attributes_text AS attributes_json FROM trigger_dev.task_events_v2 WHERE environment_id = {env_id:String}",
           schema: z.object({ span_id: z.string(), attributes_json: z.string() }),
           params: z.object({ env_id: z.string() }),
         });

internal-packages/clickhouse/src/client/client.ts

@@ -5,6 +5,7 @@ import {
   type ClickHouseSettings,
   createClient,
   type BaseQueryParams,
+  type InsertParams,
   type InsertResult,
 } from "@clickhouse/client";
 import type { Counter, Histogram, Meter, Span, Tracer, UpDownCounter } from "@internal/tracing";
@@ -1078,6 +1079,7 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter {
   public insertUnsafe<TRecord extends Record<string, any>>(req: {
     name: string;
     table: string;
+    columns?: InsertParams["columns"];
     settings?: ClickHouseSettings;
   }): ClickhouseInsertFunction<TRecord> {
     return async (events, options) => {
@@ -1109,6 +1111,7 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter {
         const [clickhouseError, result] = await tryCatch(
           this.client.insert({
             table: req.table,
+            columns: req.columns,
             format: "JSONEachRow",
             values: eventsArray,
             query_id: queryId,

internal-packages/clickhouse/src/client/types.ts

@@ -5,6 +5,7 @@ import {
   type ClickHouseSettings,
   type BaseQueryParams,
   type CommandResult,
+  type InsertParams,
   type InsertResult,
 } from "@clickhouse/client";
 import type { ClickhouseQueryBuilder, ClickhouseQueryFastBuilder } from "./queryBuilder.js";
@@ -272,6 +273,7 @@ export interface ClickhouseWriter {
   insertUnsafe<TRecord extends Record<string, any>>(req: {
     name: string;
     table: string;
+    columns?: InsertParams["columns"];
     settings?: ClickHouseSettings;
   }): ClickhouseInsertFunction<TRecord>;
 

internal-packages/clickhouse/src/taskEvents.test.ts

@@ -0,0 +1,67 @@
+import { clickhouseTest } from "@internal/testcontainers";
+import { z } from "zod";
+import { ClickHouse } from "./index.js";
+
+function clickhouseDate(value: Date) {
+  return value.toISOString().replace("T", " ").replace("Z", "");
+}
+
+describe("task events v2", () => {
+  clickhouseTest(
+    "stores materialized attributes with explicit insert columns",
+    async ({ clickhouseContainer }) => {
+      const ch = new ClickHouse({ url: clickhouseContainer.getConnectionUrl(), name: "test" });
+      const startTime = new Date("2026-09-01T10:00:00.000Z");
+      const expiresAt = new Date(Date.now() + 90 * 24 * 60 * 60 * 1000);
+      const spanId = "span_ephemeral_attributes";
+
+      const [insertError] = await ch.taskEventsV2.insert([
+        {
+          environment_id: "env_ephemeral_attributes",
+          organization_id: "org_ephemeral_attributes",
+          project_id: "project_ephemeral_attributes",
+          task_identifier: "ephemeral-attributes",
+          run_id: "run_ephemeral_attributes",
+          start_time: clickhouseDate(startTime),
+          duration: "1000000",
+          trace_id: "trace_ephemeral_attributes",
+          span_id: spanId,
+          parent_span_id: "",
+          message: "Ephemeral attributes",
+          kind: "SPAN",
+          status: "OK",
+          attributes: {
+            z: 1,
+            a: "hello",
+            nested: { enabled: true },
+          },
+          metadata: "{}",
+          expires_at: clickhouseDate(expiresAt),
+        },
+      ]);
+      expect(insertError).toBeNull();
+
+      const readAttributes = ch.reader.query({
+        name: "read-ephemeral-task-event-attributes",
+        query: `SELECT attributes_text,
+          toUInt8(inserted_at > toDateTime64('2020-01-01 00:00:00', 3)) AS has_inserted_at
+        FROM trigger_dev.task_events_v2
+        WHERE environment_id = {environmentId: String}
+          AND span_id = {spanId: String}`,
+        params: z.object({ environmentId: z.string(), spanId: z.string() }),
+        schema: z.object({ attributes_text: z.string(), has_inserted_at: z.number() }),
+      });
+      const [readError, rows] = await readAttributes({
+        environmentId: "env_ephemeral_attributes",
+        spanId,
+      });
+      expect(readError).toBeNull();
+      expect(rows).toEqual([
+        {
+          attributes_text: '{"a":"hello","nested":{"enabled":true},"z":1}',
+          has_inserted_at: 1,
+        },
+      ]);
+    }
+  );
+});

internal-packages/clickhouse/src/taskEvents.ts

@@ -176,6 +176,27 @@ export function getSpanDetailsQueryBuilder(ch: ClickhouseReader, settings?: Clic
 // V2 Table Functions (partitioned by inserted_at instead of start_time)
 // ============================================================================
 
+const TASK_EVENT_V2_INSERT_COLUMNS = [
+  "environment_id",
+  "organization_id",
+  "project_id",
+  "task_identifier",
+  "run_id",
+  "start_time",
+  "duration",
+  "trace_id",
+  "span_id",
+  "parent_span_id",
+  "message",
+  "kind",
+  "status",
+  "attributes",
+  "metadata",
+  "expires_at",
+  "machine_id",
+  "inserted_at",
+] satisfies [string, ...string[]];
+
 export const TaskEventV2Input = z.object({
   environment_id: z.string(),
   organization_id: z.string(),
@@ -204,6 +225,7 @@ export function insertTaskEventsV2(ch: ClickhouseWriter, settings?: ClickHouseSe
   return ch.insertUnsafe<TaskEventV2Input>({
     name: "insertTaskEventsV2",
     table: "trigger_dev.task_events_v2",
+    columns: TASK_EVENT_V2_INSERT_COLUMNS,
     settings: {
       enable_json_type: 1,
       type_json_skip_duplicated_paths: 1,