diff --git a/apps/api/src/routes/internal/ai-sessions.http.test.ts b/apps/api/src/routes/internal/ai-sessions.http.test.ts index c87233e8a..362df25a9 100644 --- a/apps/api/src/routes/internal/ai-sessions.http.test.ts +++ b/apps/api/src/routes/internal/ai-sessions.http.test.ts @@ -195,6 +195,91 @@ describe("POST /internal/ai-sessions/spans", () => { } }) + // A `trace:` id is Maple's own: the vendor exposed no session key, so the + // trace IS the session. Both reads must key on the trace id — the session + // attribute would match nothing, and the page would report an empty session + // for a trace that is right there. + it("routes a trace-scoped id to the trace-keyed window and span reads", async () => { + const resolved = { + startTime: "2026-08-18 09:00:00.000000000", + endTime: "2026-08-20 11:00:00.000000000", + } + let windowSql: string | undefined + let spansSql: string | undefined + const harness = makeHarness({ + compiledQuery: (_tenant, compiled) => { + windowSql = compiledQueryOf(compiled).sql + return compiledQueryOf(compiled) + .decodeRows([{ ...resolved, spanCount: "9" }]) + .pipe(Effect.orDie) + }, + compiledQueryBounded: (_tenant, compiled) => { + spansSql = compiledQueryOf(compiled).sql + return compiledQueryOf(compiled) + .decodeRows([spanRow(0), spanRow(1)]) + .pipe(Effect.orDie) + }, + }) + + try { + const response = await harness.post("/internal/ai-sessions/spans", { + sessionId: `trace:${TRACE_ID}`, + }) + expect(response.status).toBe(200) + expect(response.body.data).toHaveLength(2) + expect(windowSql).toContain(`TraceId = '${TRACE_ID}'`) + expect(windowSql).not.toContain("maple_ai.session.id") + expect(spansSql).toContain(`TraceId = '${TRACE_ID}'`) + expect(spansSql).not.toContain("maple_ai.session.id") + // The bounds the window read handed back still prune the span read. + expect(spansSql).toContain(`Timestamp >= '${resolved.startTime}'`) + expect(spansSql).not.toContain("__PARAM_") + } finally { + await harness.dispose() + } + }) + + // The prefix is not proof: the value behind it reaches a warehouse param, so + // anything that is not a trace id must not get there. It falls through to the + // session read, where nothing carries it — the empty answer any unknown id gets. + it("does not hand a malformed trace-scoped id to the trace-keyed read", async () => { + let windowSql: string | undefined + let spansRead = false + const harness = makeHarness({ + compiledQuery: (_tenant, compiled) => { + windowSql = compiledQueryOf(compiled).sql + return compiledQueryOf(compiled) + .decodeRows([ + { + startTime: "1970-01-01 00:00:00.000000000", + endTime: "1970-01-02 00:00:00.000000000", + spanCount: "0", + }, + ]) + .pipe(Effect.orDie) + }, + compiledQueryBounded: (_tenant, compiled) => { + spansRead = true + return compiledQueryOf(compiled) + .decodeRows([spanRow(0)]) + .pipe(Effect.orDie) + }, + }) + + try { + const response = await harness.post("/internal/ai-sessions/spans", { + sessionId: "trace:not-a-trace-id' OR 1=1", + }) + expect(response.status).toBe(200) + expect(response.body).toMatchObject({ data: [], truncated: false }) + expect(windowSql).toContain("SpanAttributes['maple_ai.session.id'] =") + expect(windowSql).not.toContain("TraceId =") + expect(spansRead).toBe(false) + } finally { + await harness.dispose() + } + }) + it("answers an id nothing in retention carries without reading spans", async () => { let spansRead = false const harness = makeHarness({ diff --git a/apps/api/src/routes/internal/ai-sessions.http.ts b/apps/api/src/routes/internal/ai-sessions.http.ts index 99f3e1bf8..553906c5d 100644 --- a/apps/api/src/routes/internal/ai-sessions.http.ts +++ b/apps/api/src/routes/internal/ai-sessions.http.ts @@ -9,6 +9,7 @@ import { MapleInternalApi, MAX_AI_SESSION_SPANS_RESPONSE_BYTES, } from "@maple/domain/http" +import { traceSessionTraceId } from "@maple/domain/gen-ai" import { Effect } from "effect" import { CH } from "@maple/query-engine" import * as Integrations from "@maple/query-engine-integrations" @@ -87,12 +88,21 @@ export const HttpAiSessionsInternalLive = HttpApiBuilder.group( payload.startTime !== undefined && payload.endTime !== undefined ? { startTime: payload.startTime, endTime: payload.endTime } : undefined + // A `trace:` id is Maple's own: the vendor exposed no + // session key, so the trace IS the session and both reads key on + // the trace id instead of the session attribute. The helper + // returns `undefined` for a vendor id AND for a prefixed one that + // is not 32 hex characters, so a forged value never reaches the + // trace-keyed param — it takes the session path, where nothing + // carries it and the caller gets the empty-session answer below. + const traceId = traceSessionTraceId(payload.sessionId) // Annotated before the read: a 413 never reaches the code below. // `window_source` is how often the extra resolve round-trip runs // gets watched — it should stay the exception. yield* Effect.annotateCurrentSpan({ orgId: tenant.orgId, "maple.ai.session.id": payload.sessionId, + "maple.ai.session.kind": traceId === undefined ? "vendor" : "trace", "maple.ai.window_source": hint === undefined ? "resolved" : "client", }) // The spans read has to be partition-pruned on both levels, so a @@ -100,16 +110,25 @@ export const HttpAiSessionsInternalLive = HttpApiBuilder.group( // fan-out — see `aiSessionSpansQuery`. One extra round trip, and // only on the deep-link path. const resolved = - hint === undefined - ? yield* warehouse.compiledQuery( - tenant, - CH.compile(Integrations.aiSessionWindowQuery(), { - orgId: tenant.orgId, - sessionId: payload.sessionId, - }), - { profile: "list", context: "aiSessionWindow" }, - ) - : undefined + hint !== undefined + ? undefined + : traceId === undefined + ? yield* warehouse.compiledQuery( + tenant, + CH.compile(Integrations.aiSessionWindowQuery(), { + orgId: tenant.orgId, + sessionId: payload.sessionId, + }), + { profile: "list", context: "aiSessionWindow" }, + ) + : yield* warehouse.compiledQuery( + tenant, + CH.compile(Integrations.aiTraceWindowQuery(), { + orgId: tenant.orgId, + traceId, + }), + { profile: "list", context: "aiTraceWindow" }, + ) // `min`/`max` over no rows return the epoch rather than nothing, so // the count is what distinguishes an unknown session id. const bounds = resolved?.[0] @@ -123,15 +142,26 @@ export const HttpAiSessionsInternalLive = HttpApiBuilder.group( } // One row past the cap: the extra row is what distinguishes a // session that exactly fills the cap from one whose tail was cut. - const compiled = CH.compile( - Integrations.aiSessionSpansQuery({ limit: AI_SESSION_SPANS_MAX_SPANS + 1 }), - { orgId: tenant.orgId, sessionId: payload.sessionId, ...window }, - { rowSchema: Integrations.aiSessionSpansRowSchema }, - ) + const compiled = + traceId === undefined + ? CH.compile( + Integrations.aiSessionSpansQuery({ + limit: AI_SESSION_SPANS_MAX_SPANS + 1, + }), + { orgId: tenant.orgId, sessionId: payload.sessionId, ...window }, + { rowSchema: Integrations.aiSessionSpansRowSchema }, + ) + : CH.compile( + Integrations.aiTraceSpansQuery({ + limit: AI_SESSION_SPANS_MAX_SPANS + 1, + }), + { orgId: tenant.orgId, traceId, ...window }, + { rowSchema: Integrations.aiSessionSpansRowSchema }, + ) const rows = yield* warehouse .compiledQueryBounded(tenant, compiled, { profile: "list", - context: "aiSessionSpans", + context: traceId === undefined ? "aiSessionSpans" : "aiTraceSpans", responseLimits: { maxRows: AI_SESSION_SPANS_MAX_SPANS + 1, maxBytes: MAX_AI_SESSION_SPANS_RESPONSE_BYTES, diff --git a/apps/web/src/components/agent-sessions/agent-sessions-list.tsx b/apps/web/src/components/agent-sessions/agent-sessions-list.tsx index 77d96dac5..e191897de 100644 --- a/apps/web/src/components/agent-sessions/agent-sessions-list.tsx +++ b/apps/web/src/components/agent-sessions/agent-sessions-list.tsx @@ -5,6 +5,7 @@ import { formatRelativeTimeOrDate, toEpochMs } from "@maple/ui/lib/time-format" import { formatSessionDuration } from "@maple/ui/lib/replay-format" import { ChatBubbleSparkleIcon } from "@/components/icons" import { vendorIcon } from "@/lib/agent-sessions/vendor-icon" +import { sessionRowId } from "@/lib/agent-sessions/session-window" import { vendorLabel } from "@/lib/agent-sessions/vendor-label" /** The wire row from `listAiSessions` — one AI agent session, newest first. */ @@ -44,11 +45,11 @@ export function AgentSessionsList({ sessions, limit }: AgentSessionsListProps) { Trace your AI agents with a supported framework, or emit OpenTelemetry{" "} gen_ai{" "} - spans with a{" "} + spans, and their sessions will show up here. A framework that groups its turns with a{" "} maple_ai.session.id {" "} - attribute, and their sessions will show up here. + attribute gets one session across every trace; anything else gets one per trace. @@ -85,8 +86,11 @@ export function AgentSessionsList({ sessions, limit }: AgentSessionsListProps) { {/* Identity lane: session id, framework underneath */}
- - {session.sessionId} + + {sessionRowId(session.sessionId)} {/* On phones the right-hand lanes are gone, so the timestamp anchors the top-right corner of the stacked row. */} diff --git a/apps/web/src/lib/agent-sessions/session-window.test.ts b/apps/web/src/lib/agent-sessions/session-window.test.ts index 21368c746..8bc7b71e2 100644 --- a/apps/web/src/lib/agent-sessions/session-window.test.ts +++ b/apps/web/src/lib/agent-sessions/session-window.test.ts @@ -4,6 +4,7 @@ import { breadcrumbSessionId, buildBackToSessionsHref, resolveWindow, + sessionRowId, } from "@/lib/agent-sessions/session-window" describe("resolveWindow", () => { @@ -48,6 +49,24 @@ describe("breadcrumbSessionId", () => { }) }) +describe("sessionRowId", () => { + // A framework's own id is the reader's vocabulary — it says what the session + // is, whatever its length. + it("leaves a vendor's own id whole", () => { + expect(sessionRowId("wrun_01KZTEBCDEFGHIJKLMNOPQRSTUV")).toBe("wrun_01KZTEBCDEFGHIJKLMNOPQRSTUV") + }) + + it("cuts a synthesized id to the head of its trace id", () => { + expect(sessionRowId("trace:7f3a4b5c6d7e8f901234567890abcdef")).toBe("trace:7f3a4b5c6d7e…") + }) + + // The prefix alone does not make an id Maple's, and a row must not claim a + // trace that isn't one. + it("leaves a prefixed id that is not a trace id alone", () => { + expect(sessionRowId("trace:not-a-trace-id")).toBe("trace:not-a-trace-id") + }) +}) + describe("buildBackToSessionsHref", () => { it("keeps the list's own search and drops the detail page's params", () => { const href = buildBackToSessionsHref("?vendorIds=eve&t=2026-08-19&end=2026-08-19&trace=abc&span=def") diff --git a/apps/web/src/lib/agent-sessions/session-window.ts b/apps/web/src/lib/agent-sessions/session-window.ts index 81f4c0964..5d4616f20 100644 --- a/apps/web/src/lib/agent-sessions/session-window.ts +++ b/apps/web/src/lib/agent-sessions/session-window.ts @@ -1,4 +1,5 @@ import { formatWarehouseDateTime } from "@maple/query-engine" +import { MAPLE_AI_TRACE_SESSION_PREFIX, traceSessionTraceId } from "@maple/domain/gen-ai" import { toEpochMs } from "@maple/ui/lib/time-format" // Slack, not compensation: the list now reports each session's own bounds @@ -65,3 +66,21 @@ export function breadcrumbSessionId(sessionId: string): string { if (sessionId.length <= BREADCRUMB_ID_MAX_CHARS) return sessionId return `${sessionId.slice(0, 9)}…${sessionId.slice(-4)}` } + +/** How much of a synthesized id's trace id a row shows — enough to match against + * a trace id the reader has in hand, short enough not to read as noise. */ +const TRACE_SESSION_ID_CHARS = 12 + +/** + * A session id as the list should show it. + * + * A vendor's own id is the reader's vocabulary and stays whole. A synthesized + * `trace:<32 hex>` id is Maple's, and the tail of it identifies nothing — the + * row shows the head, and the full id stays on the element's title and in the + * URL. + */ +export function sessionRowId(sessionId: string): string { + const traceId = traceSessionTraceId(sessionId) + if (traceId === undefined) return sessionId + return `${MAPLE_AI_TRACE_SESSION_PREFIX}${traceId.slice(0, TRACE_SESSION_ID_CHARS)}…` +} diff --git a/packages/domain/src/gen-ai.ts b/packages/domain/src/gen-ai.ts index 257d93f81..a4efc47db 100644 --- a/packages/domain/src/gen-ai.ts +++ b/packages/domain/src/gen-ai.ts @@ -22,6 +22,39 @@ export const MAPLE_AI_VENDOR_VERSION_ATTR = "maple_ai.vendor.version" /** The vendor's own session id, verbatim. */ export const MAPLE_AI_SESSION_ID_ATTR = "maple_ai.session.id" +/** + * Prefix of the session id Maple synthesizes for a GenAI trace that carries no + * {@link MAPLE_AI_SESSION_ID_ATTR}. + * + * The gateway stamps the session id only where the vendor exposes a session key + * — haystack, litellm, llamaindex, semantic_kernel and effect_ai never do, and + * the `unknown:*` buckets never do — so those traces have no session to belong + * to. Each one IS its own session: `trace:`, with the single trace as + * the whole context. The prefix is what keeps the two id spaces apart, and it + * is a colon-bearing shape no framework's own key is: read the id back with + * {@link traceSessionTraceId} rather than testing the prefix by hand. + */ +export const MAPLE_AI_TRACE_SESSION_PREFIX = "trace:" + +/** A W3C trace id as the warehouse stores it — 32 lowercase hex characters. */ +const TRACE_ID_PATTERN = /^[0-9a-f]{32}$/ + +/** + * The trace id a synthesized session id names, or `undefined` when the id is a + * vendor's own. + * + * A prefixed id that is not shaped like a trace id is `undefined` too. This + * value reaches a warehouse `param.*`, so "looks like a trace id" is the + * boundary check that keeps a forged one out of the trace-keyed read — it falls + * through to the session-attribute read instead, where nothing carries it and + * the caller gets the empty-session answer. + */ +export const traceSessionTraceId = (sessionId: string): string | undefined => { + if (!sessionId.startsWith(MAPLE_AI_TRACE_SESSION_PREFIX)) return undefined + const traceId = sessionId.slice(MAPLE_AI_TRACE_SESSION_PREFIX.length) + return TRACE_ID_PATTERN.test(traceId) ? traceId : undefined +} + // Maple's native convention — the one dialect an app opts into deliberately // rather than inheriting from a framework. These are ordinary span attributes // an emitter writes itself, and Maple's own agents (`apps/api` chat + diff --git a/packages/domain/src/http/ai-sessions.ts b/packages/domain/src/http/ai-sessions.ts index 101b806d8..798f6671e 100644 --- a/packages/domain/src/http/ai-sessions.ts +++ b/packages/domain/src/http/ai-sessions.ts @@ -29,6 +29,8 @@ export class ListAiSessionsRequest extends Schema.Class(" }) {} export const AiSessionListItem = Schema.Struct({ + /** The vendor's own session id, or `trace:` for an agent trace whose + * vendor exposes no session key — see `MAPLE_AI_TRACE_SESSION_PREFIX`. */ sessionId: Schema.String, /** Vendor of the earliest session-bearing span, e.g. `eve`, `vercel_ai_sdk`. */ vendorId: Schema.String, @@ -75,7 +77,13 @@ export class ListAiSessionsFacetsResponse extends Schema.Class( "GetAiSessionSpansRequest", )({ - /** The framework's own session id, verbatim — `maple_ai.session.id`. */ + /** + * The framework's own session id, verbatim — `maple_ai.session.id` — or the + * `trace:` id Maple synthesizes for a GenAI trace that carries none + * (`MAPLE_AI_TRACE_SESSION_PREFIX`). The handler routes on the prefix and + * validates the trace id behind it; a prefixed id that is not one reads as a + * session nothing carries, which answers empty like any unknown id. + */ sessionId: Schema.String.check(Schema.isMinLength(1)), // Optional, and the two halves are read as a pair — supply both or neither. // diff --git a/packages/query-engine-integrations/src/__sql_baseline__/integrations.sql b/packages/query-engine-integrations/src/__sql_baseline__/integrations.sql index f2c5ffce5..db7d89fb9 100644 --- a/packages/query-engine-integrations/src/__sql_baseline__/integrations.sql +++ b/packages/query-engine-integrations/src/__sql_baseline__/integrations.sql @@ -1,28 +1,38 @@ -- builder:ai-sessions:aiSessionFacetsQuery:default SELECT - SpanAttributes['maple_ai.vendor.id'] AS name, - uniqExact(SpanAttributes['maple_ai.session.id']) AS count, + arrayJoin(names) AS name, + uniqExact(if(rawSessionId = '', concat('trace:', traceId), rawSessionId)) AS count, 'vendor' AS facetType + FROM (SELECT + TraceId AS traceId, + max(SpanAttributes['maple_ai.session.id']) AS rawSessionId, + groupUniqArray(SpanAttributes['maple_ai.vendor.id']) AS names FROM traces WHERE OrgId = 'org_sql_catalog' AND Timestamp >= '2026-01-01 10:30:00' AND Timestamp <= '2026-01-03 14:15:00' - AND (mapContains(SpanAttributes, 'maple_ai.session.id') AND SpanAttributes['maple_ai.session.id'] != '') + AND (mapContains(SpanAttributes, 'maple_ai.vendor.id') AND SpanAttributes['maple_ai.vendor.id'] != '') AND SpanAttributes['maple_ai.vendor.id'] != '' + GROUP BY traceId) AS facet_traces GROUP BY name ORDER BY count DESC LIMIT 50 UNION ALL SELECT - ServiceName AS name, - uniqExact(SpanAttributes['maple_ai.session.id']) AS count, + arrayJoin(names) AS name, + uniqExact(if(rawSessionId = '', concat('trace:', traceId), rawSessionId)) AS count, 'service' AS facetType + FROM (SELECT + TraceId AS traceId, + max(SpanAttributes['maple_ai.session.id']) AS rawSessionId, + groupUniqArray(ServiceName) AS names FROM traces WHERE OrgId = 'org_sql_catalog' AND Timestamp >= '2026-01-01 10:30:00' AND Timestamp <= '2026-01-03 14:15:00' - AND (mapContains(SpanAttributes, 'maple_ai.session.id') AND SpanAttributes['maple_ai.session.id'] != '') + AND (mapContains(SpanAttributes, 'maple_ai.vendor.id') AND SpanAttributes['maple_ai.vendor.id'] != '') AND ServiceName != '' + GROUP BY traceId) AS facet_traces GROUP BY name ORDER BY count DESC LIMIT 50 @@ -30,7 +40,7 @@ FORMAT JSON -- builder:ai-sessions:aiSessionListQuery:default SELECT - sessionId AS sessionId, + if(rawSessionId = '', concat('trace:', traceId), rawSessionId) AS sessionId, argMin(vendorId, sessionStart) AS vendorId, argMin(vendorVersion, sessionStart) AS vendorVersion, count() AS traceCount, @@ -42,9 +52,9 @@ SELECT intDiv(max(traceEndNanos) - toUnixTimestamp64Nano(min(traceStart)), 1000000) AS durationMs FROM (SELECT TraceId AS traceId, - max(SpanAttributes['maple_ai.session.id']) AS sessionId, - argMin(SpanAttributes['maple_ai.vendor.id'], if((mapContains(SpanAttributes, 'maple_ai.session.id') AND SpanAttributes['maple_ai.session.id'] != ''), Timestamp, toDateTime('2106-01-01 00:00:00'))) AS vendorId, - argMin(SpanAttributes['maple_ai.vendor.version'], if((mapContains(SpanAttributes, 'maple_ai.session.id') AND SpanAttributes['maple_ai.session.id'] != ''), Timestamp, toDateTime('2106-01-01 00:00:00'))) AS vendorVersion, + max(SpanAttributes['maple_ai.session.id']) AS rawSessionId, + argMin(SpanAttributes['maple_ai.vendor.id'], tuple(multiIf((mapContains(SpanAttributes, 'maple_ai.session.id') AND SpanAttributes['maple_ai.session.id'] != ''), 0, SpanAttributes['maple_ai.vendor.id'] != '', 1, 2), Timestamp)) AS vendorId, + argMin(SpanAttributes['maple_ai.vendor.version'], tuple(multiIf((mapContains(SpanAttributes, 'maple_ai.session.id') AND SpanAttributes['maple_ai.session.id'] != ''), 0, SpanAttributes['maple_ai.vendor.id'] != '', 1, 2), Timestamp)) AS vendorVersion, min(if((mapContains(SpanAttributes, 'maple_ai.session.id') AND SpanAttributes['maple_ai.session.id'] != ''), Timestamp, toDateTime('2106-01-01 00:00:00'))) AS sessionStart, count() AS spanCount, countIf((StatusCode = 'Error' OR (SpanAttributes['maple_ai.vendor.id'] != '' AND (SpanAttributes['error.type'] != '' OR SpanAttributes['gen_ai.response.status'] IN ('failed', 'error'))))) AS errorSpanCount, @@ -61,9 +71,8 @@ SELECT WHERE OrgId = 'org_sql_catalog' AND Timestamp >= '2026-01-01 10:30:00' AND Timestamp <= '2026-01-03 14:15:00' - AND (mapContains(SpanAttributes, 'maple_ai.session.id') AND SpanAttributes['maple_ai.session.id'] != '')) + AND (mapContains(SpanAttributes, 'maple_ai.vendor.id') AND SpanAttributes['maple_ai.vendor.id'] != '')) GROUP BY traceId) AS session_traces - WHERE sessionId != '' GROUP BY sessionId ORDER BY startTime DESC LIMIT 50 @@ -71,7 +80,7 @@ SELECT -- builder:ai-sessions:aiSessionListQuery:filtered SELECT - sessionId AS sessionId, + if(rawSessionId = '', concat('trace:', traceId), rawSessionId) AS sessionId, argMin(vendorId, sessionStart) AS vendorId, argMin(vendorVersion, sessionStart) AS vendorVersion, count() AS traceCount, @@ -83,9 +92,9 @@ SELECT intDiv(max(traceEndNanos) - toUnixTimestamp64Nano(min(traceStart)), 1000000) AS durationMs FROM (SELECT TraceId AS traceId, - max(SpanAttributes['maple_ai.session.id']) AS sessionId, - argMin(SpanAttributes['maple_ai.vendor.id'], if((mapContains(SpanAttributes, 'maple_ai.session.id') AND SpanAttributes['maple_ai.session.id'] != ''), Timestamp, toDateTime('2106-01-01 00:00:00'))) AS vendorId, - argMin(SpanAttributes['maple_ai.vendor.version'], if((mapContains(SpanAttributes, 'maple_ai.session.id') AND SpanAttributes['maple_ai.session.id'] != ''), Timestamp, toDateTime('2106-01-01 00:00:00'))) AS vendorVersion, + max(SpanAttributes['maple_ai.session.id']) AS rawSessionId, + argMin(SpanAttributes['maple_ai.vendor.id'], tuple(multiIf((mapContains(SpanAttributes, 'maple_ai.session.id') AND SpanAttributes['maple_ai.session.id'] != ''), 0, SpanAttributes['maple_ai.vendor.id'] != '', 1, 2), Timestamp)) AS vendorId, + argMin(SpanAttributes['maple_ai.vendor.version'], tuple(multiIf((mapContains(SpanAttributes, 'maple_ai.session.id') AND SpanAttributes['maple_ai.session.id'] != ''), 0, SpanAttributes['maple_ai.vendor.id'] != '', 1, 2), Timestamp)) AS vendorVersion, min(if((mapContains(SpanAttributes, 'maple_ai.session.id') AND SpanAttributes['maple_ai.session.id'] != ''), Timestamp, toDateTime('2106-01-01 00:00:00'))) AS sessionStart, count() AS spanCount, countIf((StatusCode = 'Error' OR (SpanAttributes['maple_ai.vendor.id'] != '' AND (SpanAttributes['error.type'] != '' OR SpanAttributes['gen_ai.response.status'] IN ('failed', 'error'))))) AS errorSpanCount, @@ -102,11 +111,10 @@ SELECT WHERE OrgId = 'org_sql_catalog' AND Timestamp >= '2026-01-01 10:30:00' AND Timestamp <= '2026-01-03 14:15:00' - AND (mapContains(SpanAttributes, 'maple_ai.session.id') AND SpanAttributes['maple_ai.session.id'] != '') + AND (mapContains(SpanAttributes, 'maple_ai.vendor.id') AND SpanAttributes['maple_ai.vendor.id'] != '') AND SpanAttributes['maple_ai.vendor.id'] IN ('eve') AND ServiceName IN ('maple-slack-agent')) GROUP BY traceId) AS session_traces - WHERE sessionId != '' GROUP BY sessionId ORDER BY startTime DESC LIMIT 25 @@ -153,6 +161,39 @@ SELECT AND SpanAttributes['maple_ai.session.id'] = 'wrun_sql_catalog' FORMAT JSON +-- builder:ai-sessions:aiTraceSpansQuery:default +SELECT + TraceId AS traceId, + SpanId AS spanId, + ParentSpanId AS parentSpanId, + SpanName AS spanName, + SpanKind AS spanKind, + ServiceName AS serviceName, + Duration / 1000000 AS durationMs, + StatusCode AS statusCode, + StatusMessage AS statusMessage, + toString(Timestamp) AS timestamp, + SpanAttributes AS spanAttributes, + ResourceAttributes AS resourceAttributes + FROM trace_detail_spans + WHERE OrgId = 'org_sql_catalog' + AND Timestamp >= '2026-01-01 10:30:00' + AND Timestamp <= '2026-01-03 14:15:00' + AND TraceId = '7f3a4b5c6d7e8f901234567890abcdef' + ORDER BY timestamp ASC, spanId ASC + LIMIT 2000 + FORMAT JSON + +-- builder:ai-sessions:aiTraceWindowQuery:default +SELECT + toString(min(Timestamp) - INTERVAL 86400 SECOND) AS startTime, + toString(max(Timestamp) + INTERVAL 86400 SECOND) AS endTime, + count() AS spanCount + FROM traces + WHERE OrgId = 'org_sql_catalog' + AND TraceId = '7f3a4b5c6d7e8f901234567890abcdef' + FORMAT JSON + -- builder:billing-usage:dailyProductEventCountQuery:default SELECT toStartOfInterval(Timestamp, INTERVAL 86400 SECOND) AS day, diff --git a/packages/query-engine-integrations/src/ai/ai-sessions.test.ts b/packages/query-engine-integrations/src/ai/ai-sessions.test.ts index 550436337..5b67f56d2 100644 --- a/packages/query-engine-integrations/src/ai/ai-sessions.test.ts +++ b/packages/query-engine-integrations/src/ai/ai-sessions.test.ts @@ -7,6 +7,8 @@ import { aiSessionSpansQuery, aiSessionSpansRowSchema, aiSessionWindowQuery, + aiTraceSpansQuery, + aiTraceWindowQuery, } from "./ai-sessions" const params = { @@ -17,6 +19,17 @@ const params = { const spanParams = { ...params, sessionId: "wrun_01M0CSAEW96BH2W9185XZPRPKH" } +const TRACE_ID = "7f3a4b5c6d7e8f901234567890abcdef" +const traceParams = { ...params, traceId: TRACE_ID } + +/** The detection predicate every read now keys on — a GenAI marker of any kind, + * which is what admits a trace whose vendor exposes no session key. */ +const VENDOR_GUARD = + "(mapContains(SpanAttributes, 'maple_ai.vendor.id') AND SpanAttributes['maple_ai.vendor.id'] != '')" + +/** The trace's session id, or the synthesized one — the grouping key. */ +const SESSION_KEY = "if(rawSessionId = '', concat('trace:', traceId), rawSessionId)" + const decodeRows = (compiled: CompiledQuery, rows: ReadonlyArray>) => Effect.runSync(compiled.decodeRows(rows)) @@ -48,11 +61,37 @@ describe("aiSessionListQuery", () => { expect(compileUnsafe(aiSessionListQuery(), params).tenantScope).toBe("single-tenant") }) + it("detects on the vendor stamp, not the session id", () => { + const { sql } = compileUnsafe(aiSessionListQuery(), params) + const [, detection] = sql.split("TraceId IN (SELECT") + + // The session id is sparse by vendor — several frameworks never emit one — + // so keying detection on it hid those traces entirely. The vendor stamp is + // on every span the gateway classified, and `mapContains` is what the + // mapKeys bloom index prunes on. + expect(detection).toContain(VENDOR_GUARD) + expect(detection).not.toContain("mapContains(SpanAttributes, 'maple_ai.session.id')") + }) + + it("keys a trace with no session id on the trace itself", () => { + const { sql } = compileUnsafe(aiSessionListQuery(), params) + + // One session per sessionless trace, and the per-trace derived table is the + // only level that can say so: a span of a session-bearing trace carries no + // session id of its own either. + expect(sql).toContain(`${SESSION_KEY} AS sessionId`) + expect(sql).toContain("max(SpanAttributes['maple_ai.session.id']) AS rawSessionId") + expect(sql).toContain("GROUP BY sessionId") + // The guard that used to drop them. The key is never empty now, and a + // blank one would have swallowed every such trace into one session. + expect(sql).not.toContain("WHERE sessionId != ''") + }) + it("tests session-id presence with mapContains AND a non-empty value", () => { const { sql } = compileUnsafe(aiSessionListQuery(), params) // ClickHouse yields '' for a missing Map key, so mapContains alone would - // admit spans carrying an empty session id. + // rank spans carrying an empty session id as session-bearing. expect(sql).toContain( "(mapContains(SpanAttributes, 'maple_ai.session.id') AND SpanAttributes['maple_ai.session.id'] != '')", ) @@ -64,14 +103,26 @@ describe("aiSessionListQuery", () => { // max(vendorId) picked `vercel_ai_sdk` alphabetically over the `eve` that // actually ran the turn — see the builder's doc comment. expect(sql).not.toContain("max(SpanAttributes['maple_ai.vendor.id'])") - expect(sql).toContain("argMin(SpanAttributes['maple_ai.vendor.id'], if(") - expect(sql).toContain("argMin(SpanAttributes['maple_ai.vendor.version'], if(") + expect(sql).toContain("argMin(SpanAttributes['maple_ai.vendor.id'], tuple(multiIf(") + expect(sql).toContain("argMin(SpanAttributes['maple_ai.vendor.version'], tuple(multiIf(") expect(sql).toContain("argMin(vendorId, sessionStart) AS vendorId") expect(sql).toContain("argMin(vendorVersion, sessionStart) AS vendorVersion") // The sentinel must stay inside DateTime's range or toDateTime won't parse. expect(sql).toContain("toDateTime('2106-01-01 00:00:00')") }) + it("ranks a sessionless trace's spans so a vendor-stamped one wins", () => { + const { sql } = compileUnsafe(aiSessionListQuery(), params) + + // Every span of a sessionless trace ties at the sentinel under the session + // ordering alone, and argMin over ties is non-deterministic — it handed + // back whichever span was read first, blank vendor included. Rank first, + // then time, compared as a tuple. + expect(sql).toContain( + `tuple(multiIf((mapContains(SpanAttributes, 'maple_ai.session.id') AND SpanAttributes['maple_ai.session.id'] != ''), 0, SpanAttributes['maple_ai.vendor.id'] != '', 1, 2), Timestamp)`, + ) + }) + it("escapes an org id carrying a quote", () => { const { sql } = compileUnsafe(aiSessionListQuery(), { ...params, orgId: "org'evil" }) @@ -165,15 +216,27 @@ describe("aiSessionFacetsQuery", () => { it("counts distinct sessions per vendor and per service", () => { const { sql } = compileUnionUnsafe(aiSessionFacetsQuery(), params) - expect(sql).toContain("SpanAttributes['maple_ai.vendor.id'] AS name") - expect(sql).toContain("ServiceName AS name") + expect(sql).toContain("groupUniqArray(SpanAttributes['maple_ai.vendor.id']) AS names") + expect(sql).toContain("groupUniqArray(ServiceName) AS names") + expect(sql.split("arrayJoin(names) AS name").length - 1).toBe(2) expect(sql).toContain("'vendor' AS facetType") expect(sql).toContain("'service' AS facetType") - expect(sql.split("uniqExact(SpanAttributes['maple_ai.session.id']) AS count").length - 1).toBe(2) expect(sql.split("GROUP BY name").length - 1).toBe(2) expect(sql.split("ORDER BY count DESC").length - 1).toBe(2) }) + it("counts the trace's session key, resolved one level below the count", () => { + const { sql } = compileUnionUnsafe(aiSessionFacetsQuery(), params) + + // Keyed per span, a facet would count every agent span of a session-bearing + // trace that lacks the id — most of them — as its own sessionless trace, + // and roughly double every number in the sidebar. So the key is resolved + // per trace and only then counted. + expect(sql.split(`uniqExact(${SESSION_KEY}) AS count`).length - 1).toBe(2) + expect(sql.split("GROUP BY traceId").length - 1).toBe(2) + expect(sql).not.toContain("uniqExact(SpanAttributes['maple_ai.session.id'])") + }) + it("repeats the org and window predicates on every union branch", () => { const { sql } = compileUnionUnsafe(aiSessionFacetsQuery(), params) @@ -186,14 +249,13 @@ describe("aiSessionFacetsQuery", () => { expect(compileUnionUnsafe(aiSessionFacetsQuery(), params).tenantScope).toBe("single-tenant") }) - it("counts only session-bearing spans, and drops the blank option", () => { + it("counts over the same population the list detects, and drops the blank option", () => { const { sql } = compileUnionUnsafe(aiSessionFacetsQuery(), params) - expect( - sql.split( - "(mapContains(SpanAttributes, 'maple_ai.session.id') AND SpanAttributes['maple_ai.session.id'] != '')", - ).length - 1, - ).toBe(2) + // Same guard as `aiSessionListQuery`'s detection, so the population a facet + // describes is exactly the population its filter selects. + expect(sql.split(VENDOR_GUARD).length - 1).toBe(2) + expect(sql).not.toContain("mapContains(SpanAttributes, 'maple_ai.session.id')") expect(sql).toContain("SpanAttributes['maple_ai.vendor.id'] != ''") expect(sql).toContain("ServiceName != ''") }) @@ -362,3 +424,142 @@ describe("aiSessionWindowQuery", () => { ]) }) }) + +// The `trace:` half of the pair — a session whose vendor exposes no session key, +// so its id names the trace and neither read touches `maple_ai.session.id`. + +describe("aiTraceWindowQuery", () => { + const traceWindowParams = { orgId: params.orgId, traceId: TRACE_ID } + + it("resolves the bounds from the trace id, without a time predicate", () => { + const { sql } = compileUnsafe(aiTraceWindowQuery(), traceWindowParams) + + // `idx_trace_id` on `traces` is what bounds this, exactly as the mapValues + // bloom index bounds the session-id form. + expect(sql).toContain("FROM traces") + expect(sql).toContain(`TraceId = '${TRACE_ID}'`) + expect(sql).not.toContain("SpanAttributes") + expect(sql).not.toContain("Timestamp >=") + expect(sql).not.toContain("Timestamp <=") + }) + + it("reports bounds padded exactly like the session form", () => { + const { sql } = compileUnsafe(aiTraceWindowQuery(), traceWindowParams) + + expect(sql).toContain("toString(min(Timestamp) - INTERVAL 86400 SECOND) AS startTime") + expect(sql).toContain("toString(max(Timestamp) + INTERVAL 86400 SECOND) AS endTime") + }) + + it("is org-scoped, and escapes the trace id", () => { + const compiled = compileUnsafe(aiTraceWindowQuery(), traceWindowParams) + expect(compiled.tenantScope).toBe("single-tenant") + expect(orgPredicateCount(compiled.sql)).toBe(1) + + // The route only ever passes 32 hex characters, but the compiled SQL is + // where that stops being the only thing between a forged id and the query. + const escaped = compileUnsafe(aiTraceWindowQuery(), { + ...traceWindowParams, + traceId: "trace'evil", + }) + expect(escaped.sql).toContain("TraceId = 'trace\\'evil'") + }) + + it("leaves no unresolved param placeholder", () => { + expect(compileUnsafe(aiTraceWindowQuery(), traceWindowParams).sql).not.toContain("__PARAM_") + }) + + it("decodes the quoted 64-bit count", () => { + const compiled = compileUnsafe(aiTraceWindowQuery(), traceWindowParams) + + expect( + decodeRows(compiled, [ + { + startTime: "2026-08-18 10:33:25.825000000", + endTime: "2026-08-20 10:33:36.242000000", + spanCount: "17", + }, + ]), + ).toEqual([ + { + startTime: "2026-08-18 10:33:25.825000000", + endTime: "2026-08-20 10:33:36.242000000", + spanCount: 17, + }, + ]) + }) +}) + +describe("aiTraceSpansQuery", () => { + it("reads one trace's spans directly, with no detection subquery", () => { + const { sql } = compileUnsafe(aiTraceSpansQuery(), traceParams) + + // `TraceId` is a sort-key prefix of `trace_detail_spans`, so the id alone + // is a seek — there is nothing left for a detection level to resolve. + expect(sql).toContain("FROM trace_detail_spans") + expect(sql).toContain(`TraceId = '${TRACE_ID}'`) + expect(sql).not.toContain("TraceId IN (SELECT") + expect(sql).not.toContain("FROM traces") + expect(sql).not.toContain("maple_ai.session.id") + }) + + it("keeps the projection and the order of the session form", () => { + const { sql } = compileUnsafe(aiTraceSpansQuery(), traceParams) + + // One shape whichever kind of session the detail page opened. + expect(sql).toContain("Duration / 1000000 AS durationMs") + expect(sql).toContain("SpanAttributes AS spanAttributes") + expect(sql).toContain("ResourceAttributes AS resourceAttributes") + expect(sql).toContain("ORDER BY timestamp ASC, spanId ASC") + expect(sql).toContain("LIMIT 2000") + expect(compileUnsafe(aiTraceSpansQuery({ limit: 100 }), traceParams).sql).toContain("LIMIT 100") + }) + + it("still bounds the read by the window", () => { + const { sql } = compileUnsafe(aiTraceSpansQuery(), traceParams) + + // The sort key prunes granules; only the `Timestamp` predicate prunes + // partitions, and this table is PARTITION BY toDate(Timestamp). + expect(sql).toContain(`Timestamp >= '${params.startTime}'`) + expect(sql).toContain(`Timestamp <= '${params.endTime}'`) + }) + + it("is org-scoped on the one level it has, and escapes the trace id", () => { + const compiled = compileUnsafe(aiTraceSpansQuery(), traceParams) + expect(compiled.tenantScope).toBe("single-tenant") + expect(orgPredicateCount(compiled.sql)).toBe(1) + + const escaped = compileUnsafe(aiTraceSpansQuery(), { ...traceParams, traceId: "trace'evil" }) + expect(escaped.sql).toContain("TraceId = 'trace\\'evil'") + }) + + it("leaves no unresolved param placeholder", () => { + expect(compileUnsafe(aiTraceSpansQuery(), traceParams).sql).not.toContain("__PARAM_") + }) + + it("decodes through the same row schema as the session form", () => { + const compiled = compileUnsafe(aiTraceSpansQuery(), traceParams, { + rowSchema: aiSessionSpansRowSchema, + }) + + const [row] = decodeRows(compiled, [ + { + traceId: TRACE_ID, + spanId: "aa11", + parentSpanId: "", + spanName: "chat", + spanKind: "Client", + serviceName: "rag-service", + durationMs: "250", + statusCode: "Ok", + statusMessage: "", + timestamp: "2026-08-19 10:33:25.825000000", + // A sessionless vendor: the stamp is there, the session key is not. + spanAttributes: { "maple_ai.vendor.id": "llamaindex" }, + resourceAttributes: { "service.name": "rag-service" }, + }, + ]) + + expect(row?.durationMs).toBe(250) + expect(row?.spanAttributes).toEqual({ "maple_ai.vendor.id": "llamaindex" }) + }) +}) diff --git a/packages/query-engine-integrations/src/ai/ai-sessions.ts b/packages/query-engine-integrations/src/ai/ai-sessions.ts index dd84fa438..40345029e 100644 --- a/packages/query-engine-integrations/src/ai/ai-sessions.ts +++ b/packages/query-engine-integrations/src/ai/ai-sessions.ts @@ -13,9 +13,19 @@ // deliberate; the dashboard shows the full agent context, not just the spans // the framework happened to label. // +// A trace can carry no session id at all and still be an agent run: several +// vendors (haystack, litellm, llamaindex, semantic_kernel, effect_ai) expose no +// session key, and the `unknown:*` buckets never do. Those traces used to be +// invisible here. They are now sessions of one trace, keyed +// `trace:` (`MAPLE_AI_TRACE_SESSION_PREFIX`) — the same page, with the +// single trace as the whole context. That is why detection keys on the VENDOR +// stamp rather than the session one: the vendor id is on every span the gateway +// classified as GenAI, so it is the marker that finds both populations, and the +// session id becomes a grouping key rather than an admission test. +// // Both queries are that fan-out, in two stages against two different tables: // -// detect — `traces`, filtered on the presence of `maple_ai.session.id`. This +// detect — `traces`, filtered on the presence of `maple_ai.vendor.id`. This // is the only level that can use the `mapKeys(SpanAttributes)` bloom skip // index, and with it the scan stays cheap over a week. It yields the // qualifying trace-id set and nothing else. @@ -52,6 +62,12 @@ // fan-out unpruned. That query is the detection scan alone, which the // `mapValues(SpanAttributes)` bloom index and the table's 30-day TTL do bound. // +// A `trace:` id needs neither the attribute detection nor the fan-out: it names +// the trace outright, so `aiTraceWindowQuery`/`aiTraceSpansQuery` are the same +// two reads with `TraceId = {traceId}` in place of the detection subquery — +// `idx_trace_id` on `traces` for the bounds, the `(OrgId, TraceId, SpanId)` +// sort key on `trace_detail_spans` for the spans. +// // Tenant scoping: a subquery contributes nothing to the outer query's scope, so // every level that reads a table repeats `OrgId = {orgId}` itself. The outermost // level of `aiSessionListQuery` reads a derived table rather than a table, and @@ -73,6 +89,7 @@ import { import { TraceDetailSpans, Traces } from "@maple/query-engine/ch/tables" import { CHNumber } from "@maple/query-engine/ch/schema" import { AI_SESSION_SPANS_MAX_SPANS } from "@maple/domain/http" +import { MAPLE_AI_TRACE_SESSION_PREFIX } from "@maple/domain/gen-ai" const SESSION_ID_ATTR = "maple_ai.session.id" const VENDOR_ID_ATTR = "maple_ai.vendor.id" @@ -108,10 +125,38 @@ const FAN_OUT_PAD_SECONDS = 86_400 const hasSessionId = (attrs: CH.Expr>, get: CH.Expr) => CH.mapContains(attrs, SESSION_ID_ATTR).and(get.neq("")) +/** + * The detection predicate: every span the gateway classified as GenAI carries a + * vendor id, session key or not, so this is the marker that admits a sessionless + * trace without admitting anything that is not an agent span. Same two halves as + * {@link hasSessionId} and for the same reason, and `mapContains` is what the + * `mapKeys(SpanAttributes)` bloom index prunes on. + */ +const hasVendorId = (attrs: CH.Expr>, get: CH.Expr) => + CH.mapContains(attrs, VENDOR_ID_ATTR).and(get.neq("")) + /** Not in the builder's function set; same local helper `tracesDetailQuery` uses. */ const fromUnixTimestamp64Nano = (nanos: CH.Expr): CH.Expr => compileFnCall("fromUnixTimestamp64Nano", nanos) +/** Lexicographic ordering key — ClickHouse compares tuples element by element, + * which is how one `argMin` expresses "lowest rank, then earliest". Not in the + * builder's function set, and never selected: it only ever orders an argMin. */ +const orderTuple = (...parts: ReadonlyArray): CH.Expr => + compileFnCall("tuple", ...parts) + +/** + * The session id a trace is filed under: the vendor's own where it has one, + * else `trace:` — a session of exactly this one trace. + * + * Reads the per-trace derived table rather than raw spans, because + * sessionless-ness is a property of the TRACE and not of the span: most spans of + * a session-bearing trace carry no session id themselves, and keying on that + * would file each of them as its own sessionless trace. + */ +const sessionKey = (rawSessionId: CH.Expr, traceId: CH.Expr): CH.Expr => + CH.if_(rawSessionId.eq(""), CH.concat(MAPLE_AI_TRACE_SESSION_PREFIX, traceId), rawSessionId) + export interface AiSessionListOpts { /** Sessions returned, most recently started first. */ readonly limit?: number @@ -120,6 +165,8 @@ export interface AiSessionListOpts { } export interface AiSessionListOutput { + /** The vendor's own session id, or `trace:` for a trace that has + * none — see `MAPLE_AI_TRACE_SESSION_PREFIX`. */ readonly sessionId: string /** Vendor of the earliest session-bearing span — see `aiSessionListQuery`. */ readonly vendorId: string @@ -137,22 +184,30 @@ export interface AiSessionListOutput { /** * One row per AI agent session in the window. * + * Detection admits any trace with a GenAI span, and the session id then groups + * rather than admits: a trace that carries one is filed under it — with the + * session's other traces — and a trace that carries none becomes a session of + * its own, keyed `trace:`. Sessionless is the normal state for whole + * vendors, not an edge case; see this file's header. + * * `vendorId` is the vendor of the EARLIEST span that carries a session id, not * `max(vendorId)`. A single trace legitimately carries several vendors — an eve * agent calls through the Vercel AI SDK — and `max` picked `vercel_ai_sdk` * alphabetically over `eve` when `eve` was the framework actually running the * turn. The root-most session-bearing span is the one that names the framework, - * so the two `argMin`s (per trace, then across traces) resolve to it. + * so the two `argMin`s (per trace, then across traces) resolve to it. A trace + * with no session-bearing span falls to the next rank of the same ordering — + * its earliest vendor-stamped span, which is that trace's root-most agent span. * * The vendor filter goes on the detection subquery: it is the level the bloom * index serves, and it is the only place `maple_ai.vendor.id` is unambiguous — * a trace's other spans carry other vendors, or none. * - * The service filter goes there too, which means "the session-bearing spans came + * The service filter goes there too, which means "the trace's agent spans came * from this service" rather than "the trace touched this service". A trace spans * services by definition, so the alternative — filtering the fan-out — would - * silently drop spans and under-count `spanCount`. The session-bearing spans come - * from the agent's own service, which is the one a user filtering by service means. + * silently drop spans and under-count `spanCount`. The agent spans come from the + * agent's own service, which is the one a user filtering by service means. * * The time window bounds DETECTION exactly and the fan-out loosely. Once a trace * qualifies it is aggregated across the padded window rather than the caller's, @@ -176,7 +231,7 @@ export function aiSessionListQuery(opts: AiSessionListOpts = {}) { $.OrgId.eq(param.string("orgId")), $.Timestamp.gte(param.dateTimeString("startTime")), $.Timestamp.lte(param.dateTimeString("endTime")), - hasSessionId($.SpanAttributes, $.SpanAttributes.get(SESSION_ID_ATTR)), + hasVendorId($.SpanAttributes, $.SpanAttributes.get(VENDOR_ID_ATTR)), opts.vendorIds?.length ? CH.inList($.SpanAttributes.get(VENDOR_ID_ATTR), opts.vendorIds) : undefined, @@ -193,13 +248,30 @@ export function aiSessionListQuery(opts: AiSessionListOpts = {}) { $.Timestamp, CH.toDateTime(CH.lit(SESSION_ORDER_SENTINEL)), ) + // Ranks a trace's spans for the vendor `argMin`s: session-bearing first, + // then merely vendor-stamped, then the rest, and inside each rank the + // earliest. `sessionOrder` alone ties every span of a SESSIONLESS trace + // at the sentinel, and argMin over ties is non-deterministic — it would + // hand back whichever span ClickHouse read first, blank vendor included. + const vendorOrder = orderTuple( + CH.multiIf( + [ + [hasSessionId($.SpanAttributes, $.SpanAttributes.get(SESSION_ID_ATTR)), CH.lit(0)], + [$.SpanAttributes.get(VENDOR_ID_ATTR).neq(""), CH.lit(1)], + ], + CH.lit(2), + ), + $.Timestamp, + ) return { traceId: $.TraceId, // A trace belongs to one session; the non-bearing spans read `''`, - // which `max` discards. - sessionId: CH.max_($.SpanAttributes.get(SESSION_ID_ATTR)), - vendorId: CH.argMin($.SpanAttributes.get(VENDOR_ID_ATTR), sessionOrder), - vendorVersion: CH.argMin($.SpanAttributes.get(VENDOR_VERSION_ATTR), sessionOrder), + // which `max` discards. Named apart from the outer `sessionId`: that + // one is this value or a synthesized `trace:` id, and an alias that + // referred to itself would be a cyclic alias rather than a fallback. + rawSessionId: CH.max_($.SpanAttributes.get(SESSION_ID_ATTR)), + vendorId: CH.argMin($.SpanAttributes.get(VENDOR_ID_ATTR), vendorOrder), + vendorVersion: CH.argMin($.SpanAttributes.get(VENDOR_VERSION_ATTR), vendorOrder), // Carried so the outer level can order traces by their first // session-bearing span rather than by their first span of any kind. sessionStart: CH.min_(sessionOrder), @@ -251,7 +323,11 @@ export function aiSessionListQuery(opts: AiSessionListOpts = {}) { return ( fromQuery(perTrace, "session_traces") .select(($) => ({ - sessionId: $.sessionId, + // The grouping key, and the only level that can compute it: the + // derived table is one row per trace, so a trace with no session id + // of its own becomes a session of one trace here rather than joining + // every other sessionless trace under `''`. + sessionId: sessionKey($.rawSessionId, $.traceId), vendorId: CH.argMin($.vendorId, $.sessionStart), vendorVersion: CH.argMin($.vendorVersion, $.sessionStart), // `count()`, not `uniq()`: the derived table already emits exactly one @@ -271,12 +347,11 @@ export function aiSessionListQuery(opts: AiSessionListOpts = {}) { 1_000_000, ), })) - // A trace that qualified on `traces` normally cannot roll up empty now - // that the fan-out sees all of its spans — but `trace_detail_spans` is a - // materialized view, so a trace whose session-bearing span has landed in - // one table and not yet the other would otherwise group every such trace - // together under an empty session id. - .where(($) => [$.sessionId.neq("")]) + // No `sessionId != ''` guard any more, and none needed: the key is never + // empty. It used to keep two unrelated populations apart — a trace whose + // session-bearing span has landed in `traces` but not yet in the + // `trace_detail_spans` MV read back empty, and every such trace grouped + // together under one blank session. Both now key on their own trace id. .groupBy("sessionId") .orderBy(["startTime", "desc"]) .limit(limit) @@ -300,22 +375,32 @@ export interface AiSessionFacetsOutput { * the list's filters are applied at that level, so the population a facet * describes is exactly the population its filter selects. * - * That makes the counts ANY-span counts, matching the filter: a session belongs - * to every vendor and every service that ANY of its session-bearing spans - * carries, so a session whose turn spans came from two vendors is counted under - * both and the facet counts sum to more than the number of sessions. Picking one - * value returns exactly the count shown. + * What it cannot do is count per span. A session id is a fact about the TRACE, + * so a facet keyed on the span's own value would count every agent span of a + * session-bearing trace that lacks the id — most of them — as a separate + * sessionless trace, and roughly double every number in the sidebar. Hence the + * per-trace level: one row per trace carrying its key, with the facet's values + * collected alongside and unnested by `arrayJoin` at the counting level. + * + * The counts stay ANY-span counts, matching the filter: a session belongs to + * every vendor and every service that ANY of its agent spans carries, so a + * session whose spans came from two vendors is counted under both and the facet + * counts sum to more than the number of sessions. Picking one value returns + * exactly the count shown. * * `uniqExact` rather than `uniq`: session counts are small enough that the exact * aggregate costs nothing, and the number has to agree with the list beside it. */ export function aiSessionFacetsQuery(): CHUnionQuery { - const facet = (facetType: string, name: ($: ColumnAccessor) => CH.Expr) => - from(Traces) + const facet = ( + facetType: string, + name: ($: ColumnAccessor) => CH.Expr, + ) => { + const perTrace = from(Traces) .select(($) => ({ - name: name($), - count: CH.uniqExact($.SpanAttributes.get(SESSION_ID_ATTR)), - facetType: CH.lit(facetType), + traceId: $.TraceId, + rawSessionId: CH.max_($.SpanAttributes.get(SESSION_ID_ATTR)), + names: CH.groupUniqArray(name($)), })) .where(($) => [ // Every UNION ALL branch reads a table, so every branch carries the org @@ -323,14 +408,22 @@ export function aiSessionFacetsQuery(): CHUnionQuery { $.OrgId.eq(param.string("orgId")), $.Timestamp.gte(param.dateTimeString("startTime")), $.Timestamp.lte(param.dateTimeString("endTime")), - hasSessionId($.SpanAttributes, $.SpanAttributes.get(SESSION_ID_ATTR)), - // A span can be session-bearing without a vendor stamp; a blank option - // filters nothing and is not offered. + hasVendorId($.SpanAttributes, $.SpanAttributes.get(VENDOR_ID_ATTR)), + // A blank option filters nothing and is not offered. name($).neq(""), ]) + .groupBy("traceId") + + return fromQuery(perTrace, "facet_traces") + .select(($) => ({ + name: CH.arrayJoin($.names), + count: CH.uniqExact(sessionKey($.rawSessionId, $.traceId)), + facetType: CH.lit(facetType), + })) .groupBy("name") .orderBy(["count", "desc"]) .limit(50) + } return unionAll( facet("vendor", ($) => $.SpanAttributes.get(VENDOR_ID_ATTR)), @@ -383,6 +476,30 @@ export function aiSessionWindowQuery() { .format("JSON") } +/** + * The same bounds for a `trace:` session — one whose id names a trace outright, + * because the vendor exposed no session key (`MAPLE_AI_TRACE_SESSION_PREFIX`). + * + * No attribute predicate at all: the id IS the trace id, so `idx_trace_id` on + * `traces` prunes what `mapValues(SpanAttributes)` prunes for a vendor session, + * and no presence guard is needed because a trace id cannot be read off a + * missing Map key. The caller extracts and validates the trace id before it + * reaches this param — a forged one must never arrive here as a bare string. + * + * Padded and `spanCount`-terminated exactly like {@link aiSessionWindowQuery}; + * the two are interchangeable to a caller holding only an id. + */ +export function aiTraceWindowQuery() { + return from(Traces) + .select(($) => ({ + startTime: CH.toString_(CH.intervalSub(CH.min_($.Timestamp), FAN_OUT_PAD_SECONDS)), + endTime: CH.toString_(CH.intervalAdd(CH.max_($.Timestamp), FAN_OUT_PAD_SECONDS)), + spanCount: CH.count(), + })) + .where(($) => [$.OrgId.eq(param.string("orgId")), $.TraceId.eq(param.string("traceId"))]) + .format("JSON") +} + export interface AiSessionSpansOpts { readonly limit?: number } @@ -420,6 +537,23 @@ export const aiSessionSpansRowSchema: CompiledQueryRowSchema) => ({ + traceId: $.TraceId, + spanId: $.SpanId, + parentSpanId: $.ParentSpanId, + spanName: $.SpanName, + spanKind: $.SpanKind, + serviceName: $.ServiceName, + durationMs: $.Duration.div(1_000_000), + statusCode: $.StatusCode, + statusMessage: $.StatusMessage, + timestamp: CH.toString_($.Timestamp), + spanAttributes: $.SpanAttributes, + resourceAttributes: $.ResourceAttributes, +}) + /** * Every span of every trace belonging to one session, oldest first. * @@ -473,20 +607,7 @@ export function aiSessionSpansQuery(opts: AiSessionSpansOpts = {}) { return ( from(TraceDetailSpans) - .select(($) => ({ - traceId: $.TraceId, - spanId: $.SpanId, - parentSpanId: $.ParentSpanId, - spanName: $.SpanName, - spanKind: $.SpanKind, - serviceName: $.ServiceName, - durationMs: $.Duration.div(1_000_000), - statusCode: $.StatusCode, - statusMessage: $.StatusMessage, - timestamp: CH.toString_($.Timestamp), - spanAttributes: $.SpanAttributes, - resourceAttributes: $.ResourceAttributes, - })) + .select(spanProjection) .where(($) => [ $.OrgId.eq(param.string("orgId")), $.Timestamp.gte(param.dateTimeString("startTime")), @@ -502,3 +623,32 @@ export function aiSessionSpansQuery(opts: AiSessionSpansOpts = {}) { .format("JSON") ) } + +/** + * Every span of ONE trace, oldest first — the spans of a `trace:` session. + * + * {@link aiSessionSpansQuery} without its detection half: the id already names + * the trace, so there is nothing to resolve and `TraceId` is a sort-key prefix + * of `trace_detail_spans`. Everything else is identical, deliberately — same + * projection, same row schema, same tie-broken order, same truncation contract — + * so the detail page reads one shape whichever kind of session it opened. + * + * The window is still required and still bounds the read: `TraceId` prunes the + * sort key, the `Timestamp` predicate prunes partitions, and only both together + * keep this off every partition the table retains. + */ +export function aiTraceSpansQuery(opts: AiSessionSpansOpts = {}) { + const limit = opts.limit ?? AI_SESSION_SPANS_MAX_SPANS + + return from(TraceDetailSpans) + .select(spanProjection) + .where(($) => [ + $.OrgId.eq(param.string("orgId")), + $.Timestamp.gte(param.dateTimeString("startTime")), + $.Timestamp.lte(param.dateTimeString("endTime")), + $.TraceId.eq(param.string("traceId")), + ]) + .orderBy(["timestamp", "asc"], ["spanId", "asc"]) + .limit(limit) + .format("JSON") +} diff --git a/packages/query-engine-integrations/src/ai/index.ts b/packages/query-engine-integrations/src/ai/index.ts index 519934a18..9db68d007 100644 --- a/packages/query-engine-integrations/src/ai/index.ts +++ b/packages/query-engine-integrations/src/ai/index.ts @@ -13,6 +13,8 @@ export { aiSessionSpansQuery, aiSessionSpansRowSchema, aiSessionWindowQuery, + aiTraceSpansQuery, + aiTraceWindowQuery, type AiSessionFacetsOutput, type AiSessionListOpts, type AiSessionListOutput, diff --git a/packages/query-engine-integrations/src/catalog.ts b/packages/query-engine-integrations/src/catalog.ts index c79dba8c0..6b4ac1932 100644 --- a/packages/query-engine-integrations/src/catalog.ts +++ b/packages/query-engine-integrations/src/catalog.ts @@ -31,6 +31,9 @@ const window = { orgId: ORG_ID, startTime: START_TIME, endTime: END_TIME } /** The window plus the bucket every timeseries builder resolves a param from. */ const bucketed = { ...window, bucketSeconds: 300 } +/** A `trace:` session's trace id, in the 32-hex shape the route validates. */ +const AI_TRACE_ID = "7f3a4b5c6d7e8f901234567890abcdef" + /** One zone's spans, as the /infra/cloudflare pages scope them. */ const cfZone = { ...window, serviceName: "cloudflare-zone-example-com" } const cfZoneBucketed = { ...cfZone, bucketSeconds: 300 } @@ -106,6 +109,25 @@ export const integrationFixtures: ReadonlyArray = [ compile: () => compileUnsafe(CH.aiSessionWindowQuery(), { orgId: ORG_ID, sessionId: "wrun_sql_catalog" }), }, + { + // The same two reads for a `trace:` session — one whose vendor exposes no + // session key, so the id names the trace and the detection half is gone. + module: "ai-sessions", + name: "aiTraceWindowQuery", + label: "default", + compile: () => compileUnsafe(CH.aiTraceWindowQuery(), { orgId: ORG_ID, traceId: AI_TRACE_ID }), + }, + { + module: "ai-sessions", + name: "aiTraceSpansQuery", + label: "default", + compile: () => + compileUnsafe( + CH.aiTraceSpansQuery(), + { ...window, traceId: AI_TRACE_ID }, + { rowSchema: CH.aiSessionSpansRowSchema }, + ), + }, { module: "cloudflare-infra", name: "cloudflareZoneLatencySQL",