feat(ingest): classify AI spans on the write path and store the columns

apps/api/src/services/warehouse/WarehouseQueryService.clickhouse.e2e.test.ts

@@ -22,6 +22,9 @@ import {
 const enabled = clickhouseE2eEnabled
 const database = uniqueDatabase("maple_raw_sql_e2e")
 const orgId = "org_raw_sql_e2e"
+/** Isolates the migration-0015 default-readback probe row from the row-level
+ * fixtures the query tests assert on. */
+const aiProbeOrgId = "org_raw_sql_e2e_ai_probe"
 
 const assertSearchSchemaApplied = async (): Promise<void> => {
 	const migrationRevision = (
@@ -140,6 +143,88 @@ SETTINGS enable_full_text_index = 1`,
 	assert.include(explain, "idx_lower_body_text")
 }
 
+/**
+ * Migration 0015 is storage-only — nothing writes these columns yet — so the
+ * only thing that can prove it applied is the physical schema. Asserted against
+ * a real server because `set(0)` and `tokenbf_v1` are the kind of DDL a
+ * SQL-text test happily accepts and ClickHouse rejects.
+ */
+const assertAiClassificationSchemaApplied = async (): Promise<void> => {
+	const migrationRevision = (
+		await clickhouseExec(
+			"SELECT count() FROM _maple_schema_migrations WHERE version = 15 FORMAT TabSeparated",
+			database,
+		)
+	).trim()
+	assert.strictEqual(migrationRevision, "1", "AI classification migration 15 was not recorded")
+
+	const columns = (
+		await clickhouseExec(
+			`SELECT name, type, default_kind
+FROM system.columns
+WHERE database = currentDatabase()
+  AND table = 'traces'
+  AND name IN ('AiVendor', 'AiSessionKeyState', 'AiSessionKeyHash', 'AiRulesVersion', 'AiRollupHour')
+ORDER BY name
+FORMAT TabSeparated`,
+			database,
+		)
+	)
+		.trim()
+		// TabSeparated escapes single quotes, so `DateTime('UTC')` arrives as
+		// `DateTime(\'UTC\')`. Unescape rather than encode the escaping into the
+		// expectation, which would read as a typo.
+		.replaceAll("\\'", "'")
+	assert.strictEqual(
+		columns,
+		[
+			"AiRollupHour\tDateTime('UTC')\tDEFAULT",
+			"AiRulesVersion\tUInt32\tDEFAULT",
+			"AiSessionKeyHash\tUInt64\tDEFAULT",
+			"AiSessionKeyState\tUInt8\tDEFAULT",
+			"AiVendor\tLowCardinality(String)\tDEFAULT",
+		].join("\n"),
+		"AI classification columns are missing, mistyped, or not default-computed",
+	)
+
+	const indexes = (
+		await clickhouseExec(
+			`SELECT name, type_full, granularity
+FROM system.data_skipping_indices
+WHERE database = currentDatabase()
+  AND table = 'traces'
+  AND name IN ('idx_ai_vendor', 'idx_scope_name')
+ORDER BY name
+FORMAT TabSeparated`,
+			database,
+		)
+	).trim()
+	assert.strictEqual(
+		indexes,
+		["idx_ai_vendor\tset(0)\t4", "idx_scope_name\ttokenbf_v1(4096, 3, 0)\t4"].join("\n"),
+		"AI classification skip indexes are missing or declared with the wrong type",
+	)
+
+	// A writer that names none of the new columns — which is every writer until
+	// the classifier ships — must still land a readable row. Written under its own
+	// org so it stays out of the row-level fixtures the tests below assert on.
+	await clickhouseExec(
+		`INSERT INTO traces (OrgId, Timestamp, TraceId, SpanId, SpanName, SpanKind, ServiceName, Duration, StatusCode)
+		 VALUES ('${aiProbeOrgId}', now64(9), 'trace-ai-default', 'span-ai-default', 'GET /ai', 'Server', 'api', 1, 'Ok')`,
+		database,
+	)
+	const defaults = (
+		await clickhouseExec(
+			`SELECT AiVendor = '', AiSessionKeyState, AiSessionKeyHash, AiRulesVersion, toUnixTimestamp(AiRollupHour)
+FROM traces
+WHERE OrgId = '${aiProbeOrgId}' AND TraceId = 'trace-ai-default'
+FORMAT TabSeparated`,
+			database,
+		)
+	).trim()
+	assert.strictEqual(defaults, "1\t0\t0\t0\t0", "AI classification defaults do not read back")
+}
+
 const trackedDbs: TestDb[] = []
 const asOrgId = Schema.decodeUnknownSync(OrgId)
 const asUserId = Schema.decodeUnknownSync(UserId)
@@ -196,6 +281,7 @@ describe.skipIf(!enabled)("WarehouseQueryService ClickHouse raw-SQL E2E", () =>
 		await clickhouseExec(`CREATE DATABASE ${database}`)
 		await applyRealMigrations(database)
 		await assertSearchSchemaApplied()
+		await assertAiClassificationSchemaApplied()
 		await clickhouseExec(
 			`INSERT INTO traces
 			 (OrgId, Timestamp, TraceId, SpanId, SpanName, SpanKind, ServiceName, Duration, StatusCode)

apps/cli/src/server/local-schema-history.ts

@@ -51,4 +51,11 @@ export const LOCAL_SCHEMA_HISTORY: ReadonlyArray<LocalSchemaHistoryEntry> = Obje
 		manifestDigest: "826f9363db5dd7722debd0c87a5b74a5b66387f4752abc219d4cc0ce76358a9e",
 		projectRevision: "27015e7036e9cacaa5156bcc10a3aead96cb4fa2fcb7c615c272c691f2cbf54a",
 	}),
+	Object.freeze({
+		version: 5,
+		fingerprint: "b3059dd34e85858f",
+		digest: "b3059dd34e85858f8893cd7fc88d9c28f489992c39fb2a334f8caf1747a69c21",
+		manifestDigest: "e0b0e0a9af30cc7aca51cec02c566dab9f4cbfda1374c177a7caee9a46a31783",
+		projectRevision: "09513d18e8cdea657efa56dbe764defebe66a28e5397411dc03fadb7f19f1c58",
+	}),
 ] as const)

apps/cli/src/server/local-schema-version.ts

@@ -1,4 +1,4 @@
 // Increment this value for every structural change to the generated local
 // schema. The compatibility manifest and migration registry must be updated in
 // the same change before a new value can ship.
-export const LOCAL_SCHEMA_VERSION = 4 as const
+export const LOCAL_SCHEMA_VERSION = 5 as const

apps/cli/src/server/local-store-migrations.ts

@@ -36,6 +36,7 @@ import { legacyToCurrentModule } from "./local-store-migrations/legacy-to-curren
 import { v1ToV2ErrorRollupModule } from "./local-store-migrations/v1-to-v2-error-rollup"
 import { v2ToV3ServiceMapIngestBridgeModule } from "./local-store-migrations/v2-to-v3-service-map-ingest-bridge"
 import { v3ToV4WebEventsModule } from "./local-store-migrations/v3-to-v4-web-events"
+import { v4ToV5AiClassificationColumnsModule } from "./local-store-migrations/v4-to-v5-ai-classification-columns"
 import type {
 	AnyLocalStoreMigrationModule,
 	LocalStoreMigration,
@@ -58,6 +59,7 @@ export { legacyToCurrentModule } from "./local-store-migrations/legacy-to-curren
 export { v1ToV2ErrorRollupModule } from "./local-store-migrations/v1-to-v2-error-rollup"
 export { v2ToV3ServiceMapIngestBridgeModule } from "./local-store-migrations/v2-to-v3-service-map-ingest-bridge"
 export { v3ToV4WebEventsModule } from "./local-store-migrations/v3-to-v4-web-events"
+export { v4ToV5AiClassificationColumnsModule } from "./local-store-migrations/v4-to-v5-ai-classification-columns"
 
 const NONTERMINAL_PHASES = new Set<MigrationPhase>([
 	"planned",
@@ -122,6 +124,7 @@ export const localStoreMigrations: ReadonlyArray<AnyLocalStoreMigrationModule> =
 	v1ToV2ErrorRollupModule,
 	v2ToV3ServiceMapIngestBridgeModule,
 	v3ToV4WebEventsModule,
+	v4ToV5AiClassificationColumnsModule,
 ]
 
 export const validateMigrationRegistry = (