diff --git a/apps/api/src/routes/internal/agent-sessions.http.ts b/apps/api/src/routes/internal/agent-sessions.http.ts new file mode 100644 index 000000000..4929cf4c8 --- /dev/null +++ b/apps/api/src/routes/internal/agent-sessions.http.ts @@ -0,0 +1,155 @@ +import { HttpApiBuilder } from "effect/unstable/httpapi" +import { + AgentSessionDetailResponse, + AgentSessionsFacetsResponse, + AgentSessionsListResponse, + AgentTracesListResponse, + CurrentTenant, + MapleInternalApi, + TraceId, +} from "@maple/domain/http" +import { Effect, Schema } from "effect" +import { + agentSessionsFacets, + getAgentSessionDetail, + listAgentSessions, + listAgentTraces, + type AgentSessionsFilterInput, +} from "@maple/query-engine/observability" +import { provideWarehouseExecutorFromTenant } from "@/services/warehouse/WarehouseQueryService" + +const decodeTraceId = Schema.decodeSync(TraceId) + +/** + * Agent Sessions dashboard routes — thin adapters over the + * `@maple/query-engine/observability` read functions, which are the durable + * interface (the future MCP tools call those functions, not these routes). + * Internal tier: the response shapes follow the UI. + */ +export const HttpAgentSessionsInternalLive = HttpApiBuilder.group( + MapleInternalApi, + "agentSessionsInternal", + (handlers) => + Effect.gen(function* () { + const filterInput = (payload: { + startTime: string + endTime: string + vendors?: readonly string[] | undefined + serviceNames?: readonly string[] | undefined + hasErrors?: boolean | undefined + }): AgentSessionsFilterInput => ({ + startTime: payload.startTime, + endTime: payload.endTime, + vendors: payload.vendors, + serviceNames: payload.serviceNames, + hasErrors: payload.hasErrors, + }) + + return handlers + .handle("list", ({ payload }) => + Effect.gen(function* () { + const tenant = yield* CurrentTenant.Context + yield* Effect.annotateCurrentSpan({ orgId: tenant.orgId }) + const rows = yield* listAgentSessions({ + ...filterInput(payload), + limit: payload.limit, + offset: payload.offset, + }).pipe(provideWarehouseExecutorFromTenant(tenant)) + // Integer aggregates can arrive as JSON strings (quoted 64-bit wire on + // gateway/readonly clusters); these queries declare no rowSchema, so + // coerce before the Schema.Number response fields validate. + return new AgentSessionsListResponse({ + data: rows.map((row) => ({ + sessionKeyHash: row.sessionKeyHash, + startTime: row.startTime, + endTime: row.endTime, + durationMs: Number(row.durationMs), + traceCount: Number(row.traceCount), + keyedSpanCount: Number(row.keyedSpanCount), + errorCount: Number(row.errorCount), + vendors: row.vendors, + serviceNames: row.serviceNames, + })), + }) + }), + ) + .handle("traces", ({ payload }) => + Effect.gen(function* () { + const tenant = yield* CurrentTenant.Context + yield* Effect.annotateCurrentSpan({ orgId: tenant.orgId }) + const rows = yield* listAgentTraces({ + ...filterInput(payload), + limit: payload.limit, + offset: payload.offset, + }).pipe(provideWarehouseExecutorFromTenant(tenant)) + return new AgentTracesListResponse({ + data: rows.map((row) => ({ + traceId: decodeTraceId(row.traceId), + startTime: row.startTime, + endTime: row.endTime, + durationMs: Number(row.durationMs), + aiSpanCount: Number(row.aiSpanCount), + errorCount: Number(row.errorCount), + vendors: row.vendors, + serviceNames: row.serviceNames, + firstSpanName: row.firstSpanName, + bestSessionKeyState: Number(row.bestSessionKeyState), + sessionKeyHash: row.sessionKeyHash, + })), + }) + }), + ) + .handle("detail", ({ payload }) => + Effect.gen(function* () { + const tenant = yield* CurrentTenant.Context + yield* Effect.annotateCurrentSpan({ orgId: tenant.orgId }) + const detail = yield* getAgentSessionDetail({ + sessionKeyHash: payload.sessionKeyHash, + startTime: payload.startTime, + endTime: payload.endTime, + }).pipe(provideWarehouseExecutorFromTenant(tenant)) + // The read function already normalizes every numeric through the + // integration layer (real JS numbers, not wire strings); only the + // TraceId brand needs decoding at this boundary. + return new AgentSessionDetailResponse({ + session: + detail === null + ? null + : { + ...detail, + traces: detail.traces.map((trace) => ({ + traceId: decodeTraceId(trace.traceId), + startTime: trace.startTime, + durationMs: trace.durationMs, + errorCount: trace.errorCount, + spans: trace.spans.map( + ({ traceId: _traceId, ...span }) => span, + ), + })), + }, + }) + }), + ) + .handle("facets", ({ payload }) => + Effect.gen(function* () { + const tenant = yield* CurrentTenant.Context + yield* Effect.annotateCurrentSpan({ orgId: tenant.orgId }) + const rows = yield* agentSessionsFacets({ + ...filterInput(payload), + tab: payload.tab, + }).pipe(provideWarehouseExecutorFromTenant(tenant)) + const pick = (facetType: string) => + rows + .filter((row) => row.facetType === facetType) + .map((row) => ({ name: row.name, count: Number(row.count) })) + return new AgentSessionsFacetsResponse({ + vendors: pick("vendor"), + services: pick("service"), + // The error branch emits a single row, and none at all when the + // filtered population has no errors. + errorCount: Number(rows.find((row) => row.facetType === "error")?.count ?? 0), + }) + }), + ) + }), +) diff --git a/apps/api/src/runtime/http-graph.ts b/apps/api/src/runtime/http-graph.ts index bce96b60e..5efb2edc5 100644 --- a/apps/api/src/runtime/http-graph.ts +++ b/apps/api/src/runtime/http-graph.ts @@ -6,6 +6,7 @@ import { HttpApiBuilder, HttpApiScalar } from "effect/unstable/httpapi" import { API_CORS_OPTIONS } from "@/http/api-cors" import { McpLive } from "@/mcp/app" import { Env } from "@/platform/Env" +import { HttpAgentSessionsInternalLive } from "@/routes/internal/agent-sessions.http" import { HttpAiTriageLive } from "@/routes/internal/ai-triage.http" import { HttpAuthLive, HttpAuthPublicLive } from "@/routes/v1/auth.http" import { HttpBillingLive } from "@/routes/internal/billing.http" @@ -96,7 +97,9 @@ const ApiRoutes = HttpApiBuilder.layer(MapleApi).pipe( * which is generated from `MapleApi`. */ const ApiInternalRoutes = HttpApiBuilder.layer(MapleInternalApi).pipe( - Layer.provide(Layer.mergeAll(HttpQueryEngineLive, HttpSessionReplaysInternalLive)), + Layer.provide( + Layer.mergeAll(HttpQueryEngineLive, HttpSessionReplaysInternalLive, HttpAgentSessionsInternalLive), + ), Layer.provide( Layer.mergeAll(HttpAiTriageLive, HttpBillingLive, HttpChatLive, HttpDemoLive, HttpDigestLive), ), diff --git a/apps/web/src/api/warehouse/agent-sessions.ts b/apps/web/src/api/warehouse/agent-sessions.ts new file mode 100644 index 000000000..6a268c162 --- /dev/null +++ b/apps/web/src/api/warehouse/agent-sessions.ts @@ -0,0 +1,116 @@ +import { Clock, Effect, Schema } from "effect" +import { + AgentSessionsFacetsRequest, + AgentSessionsListRequest, + AgentTracesListRequest, +} from "@maple/domain/http" +import { MapleInternalAtomClient } from "@/lib/services/common/internal-atom-client" +import { WarehouseDateTimeString, decodeInput, runWarehouseQuery } from "@/api/warehouse/effect-utils" +import { formatWarehouseDateTime } from "@maple/query-engine" + +// Agent Sessions (AI-classified spans) — throwaway product-scratchpad feature. +// Thin pass-throughs to /internal/agent-sessions; the durable read interface is +// @maple/query-engine/observability, which those routes adapt. + +const AgentSessionsFilterInput = Schema.Struct({ + startTime: Schema.optional(WarehouseDateTimeString), + endTime: Schema.optional(WarehouseDateTimeString), + vendors: Schema.optional(Schema.Array(Schema.String)), + serviceNames: Schema.optional(Schema.Array(Schema.String)), + hasErrors: Schema.optional(Schema.Boolean), +}) + +const ListAgentSessionsInput = Schema.Struct({ + ...AgentSessionsFilterInput.fields, + limit: Schema.optional(Schema.Number), + offset: Schema.optional(Schema.Number), +}) +export type ListAgentSessionsInput = Schema.Schema.Type + +const AgentSessionsFacetsInput = Schema.Struct({ + ...AgentSessionsFilterInput.fields, + tab: Schema.Literals(["sessions", "traces"]), +}) +export type AgentSessionsFacetsInput = Schema.Schema.Type + +const defaultTimeRange = (nowMs: number) => ({ + startTime: formatWarehouseDateTime(nowMs - 24 * 60 * 60 * 1000), + endTime: formatWarehouseDateTime(nowMs), +}) + +export const listAgentSessions = Effect.fn("AgentSessions.list")(function* ({ + data, +}: { + data: ListAgentSessionsInput +}) { + const input = yield* decodeInput(ListAgentSessionsInput, data ?? {}, "listAgentSessions") + const fallback = defaultTimeRange(yield* Clock.currentTimeMillis) + const result = yield* runWarehouseQuery("listAgentSessions", () => + Effect.gen(function* () { + const client = yield* MapleInternalAtomClient + return yield* client.agentSessionsInternal.list({ + payload: new AgentSessionsListRequest({ + startTime: input.startTime ?? fallback.startTime, + endTime: input.endTime ?? fallback.endTime, + vendors: input.vendors, + serviceNames: input.serviceNames, + hasErrors: input.hasErrors, + limit: input.limit ?? 50, + offset: input.offset ?? 0, + }), + }) + }), + ) + return { data: result.data } +}) + +export const listAgentTraces = Effect.fn("AgentSessions.traces")(function* ({ + data, +}: { + data: ListAgentSessionsInput +}) { + const input = yield* decodeInput(ListAgentSessionsInput, data ?? {}, "listAgentTraces") + const fallback = defaultTimeRange(yield* Clock.currentTimeMillis) + const result = yield* runWarehouseQuery("listAgentTraces", () => + Effect.gen(function* () { + const client = yield* MapleInternalAtomClient + return yield* client.agentSessionsInternal.traces({ + payload: new AgentTracesListRequest({ + startTime: input.startTime ?? fallback.startTime, + endTime: input.endTime ?? fallback.endTime, + vendors: input.vendors, + serviceNames: input.serviceNames, + hasErrors: input.hasErrors, + limit: input.limit ?? 50, + offset: input.offset ?? 0, + }), + }) + }), + ) + return { data: result.data } +}) + +export const getAgentSessionsFacets = Effect.fn("AgentSessions.facets")(function* ({ + data, +}: { + data: AgentSessionsFacetsInput +}) { + const input = yield* decodeInput(AgentSessionsFacetsInput, data ?? {}, "agentSessionsFacets") + const fallback = defaultTimeRange(yield* Clock.currentTimeMillis) + const result = yield* runWarehouseQuery("agentSessionsFacets", () => + Effect.gen(function* () { + const client = yield* MapleInternalAtomClient + return yield* client.agentSessionsInternal.facets({ + payload: new AgentSessionsFacetsRequest({ + startTime: input.startTime ?? fallback.startTime, + endTime: input.endTime ?? fallback.endTime, + vendors: input.vendors, + serviceNames: input.serviceNames, + hasErrors: input.hasErrors, + tab: input.tab, + }), + }) + }), + ) + return { vendors: result.vendors, services: result.services, errorCount: result.errorCount } +}) diff --git a/apps/web/src/components/dashboard/nav-items.ts b/apps/web/src/components/dashboard/nav-items.ts index eba92f62c..e8f2004b9 100644 --- a/apps/web/src/components/dashboard/nav-items.ts +++ b/apps/web/src/components/dashboard/nav-items.ts @@ -2,6 +2,7 @@ import { BellIcon, ChartBarHorizontalIcon, ChartLineIcon, + ChatBubbleSparkleIcon, CircleWarningIcon, CloudflareIcon, ComputerIcon, @@ -110,6 +111,9 @@ const exploreItem: NavItem = { { title: "Logs", href: "/logs", icon: FileIcon }, { title: "Metrics", href: "/metrics", icon: ChartLineIcon }, { title: "Replays", href: "/replays", icon: PlayRotateClockwiseIcon }, + // Throwaway scratchpad feature (see routes/agent-sessions) — nav slot is + // provisional along with the rest of it. + { title: "Agents", href: "/agent-sessions", icon: ChatBubbleSparkleIcon }, ], } diff --git a/apps/web/src/lib/services/atoms/warehouse-query-atoms.ts b/apps/web/src/lib/services/atoms/warehouse-query-atoms.ts index a811130f4..6ff159118 100644 --- a/apps/web/src/lib/services/atoms/warehouse-query-atoms.ts +++ b/apps/web/src/lib/services/atoms/warehouse-query-atoms.ts @@ -110,6 +110,11 @@ import { getSessionTraceSummaries, listReplays, } from "@/api/warehouse/replays" +import { + getAgentSessionsFacets, + listAgentSessions, + listAgentTraces, +} from "@/api/warehouse/agent-sessions" import { getWebAnalyticsBreakdowns, getWebAnalyticsPages, @@ -250,6 +255,18 @@ export const replaysFacetsResultAtom = makeQueryAtomFamily(getReplaysFacets, { staleTime: 30_000, }) +export const listAgentSessionsResultAtom = makeQueryAtomFamily(listAgentSessions, { + staleTime: 30_000, +}) + +export const listAgentTracesResultAtom = makeQueryAtomFamily(listAgentTraces, { + staleTime: 30_000, +}) + +export const agentSessionsFacetsResultAtom = makeQueryAtomFamily(getAgentSessionsFacets, { + staleTime: 30_000, +}) + // Web analytics — one page, five atoms, all 30s. Traffic numbers are watched // during a launch, so a longer TTL reads as a stalled page; a shorter one just // re-runs the same 30-day-TTL scans. diff --git a/apps/web/src/routeTree.gen.ts b/apps/web/src/routeTree.gen.ts index f304c9bda..c64015f7a 100644 --- a/apps/web/src/routeTree.gen.ts +++ b/apps/web/src/routeTree.gen.ts @@ -26,6 +26,7 @@ import { Route as ServiceMapRouteImport } from './routes/service-map' import { Route as SettingsRouteImport } from './routes/settings' import { Route as SignInRouteImport } from './routes/sign-in' import { Route as SignUpRouteImport } from './routes/sign-up' +import { Route as AgentSessionsIndexRouteImport } from './routes/agent-sessions/index' import { Route as AlertsIndexRouteImport } from './routes/alerts/index' import { Route as AlertsRuleIdRouteImport } from './routes/alerts/$ruleId' import { Route as AlertsCreateRouteImport } from './routes/alerts/create' @@ -164,6 +165,11 @@ const SignUpRoute = SignUpRouteImport.update({ path: '/sign-up', getParentRoute: () => rootRouteImport, } as any) +const AgentSessionsIndexRoute = AgentSessionsIndexRouteImport.update({ + id: '/agent-sessions/', + path: '/agent-sessions/', + getParentRoute: () => rootRouteImport, +} as any) const AlertsIndexRoute = AlertsIndexRouteImport.update({ id: '/alerts/', path: '/alerts/', @@ -472,6 +478,7 @@ export interface FileRoutesByFullPath { '/services/$serviceName': typeof ServicesServiceNameRoute '/share/$token': typeof ShareTokenRoute '/traces/$traceId': typeof TracesTraceIdRoute + '/agent-sessions/': typeof AgentSessionsIndexRoute '/alerts/': typeof AlertsIndexRoute '/analytics/': typeof AnalyticsIndexRoute '/anomalies/': typeof AnomaliesIndexRoute @@ -542,6 +549,7 @@ export interface FileRoutesByTo { '/services/$serviceName': typeof ServicesServiceNameRoute '/share/$token': typeof ShareTokenRoute '/traces/$traceId': typeof TracesTraceIdRoute + '/agent-sessions': typeof AgentSessionsIndexRoute '/alerts': typeof AlertsIndexRoute '/analytics': typeof AnalyticsIndexRoute '/anomalies': typeof AnomaliesIndexRoute @@ -614,6 +622,7 @@ export interface FileRoutesById { '/services/$serviceName': typeof ServicesServiceNameRoute '/share/$token': typeof ShareTokenRoute '/traces/$traceId': typeof TracesTraceIdRoute + '/agent-sessions/': typeof AgentSessionsIndexRoute '/alerts/': typeof AlertsIndexRoute '/analytics/': typeof AnalyticsIndexRoute '/anomalies/': typeof AnomaliesIndexRoute @@ -687,6 +696,7 @@ export interface FileRouteTypes { | '/services/$serviceName' | '/share/$token' | '/traces/$traceId' + | '/agent-sessions/' | '/alerts/' | '/analytics/' | '/anomalies/' @@ -757,6 +767,7 @@ export interface FileRouteTypes { | '/services/$serviceName' | '/share/$token' | '/traces/$traceId' + | '/agent-sessions' | '/alerts' | '/analytics' | '/anomalies' @@ -828,6 +839,7 @@ export interface FileRouteTypes { | '/services/$serviceName' | '/share/$token' | '/traces/$traceId' + | '/agent-sessions/' | '/alerts/' | '/analytics/' | '/anomalies/' @@ -895,6 +907,7 @@ export interface RootRouteChildren { ServicesServiceNameRoute: typeof ServicesServiceNameRoute ShareTokenRoute: typeof ShareTokenRoute TracesTraceIdRoute: typeof TracesTraceIdRoute + AgentSessionsIndexRoute: typeof AgentSessionsIndexRoute AlertsIndexRoute: typeof AlertsIndexRoute AnalyticsIndexRoute: typeof AnalyticsIndexRoute AnomaliesIndexRoute: typeof AnomaliesIndexRoute @@ -1044,6 +1057,13 @@ declare module '@tanstack/react-router' { preLoaderRoute: typeof SignUpRouteImport parentRoute: typeof rootRouteImport } + '/agent-sessions/': { + id: '/agent-sessions/' + path: '/agent-sessions' + fullPath: '/agent-sessions/' + preLoaderRoute: typeof AgentSessionsIndexRouteImport + parentRoute: typeof rootRouteImport + } '/alerts/': { id: '/alerts/' path: '/alerts' @@ -1476,6 +1496,7 @@ const rootRouteChildren: RootRouteChildren = { ServicesServiceNameRoute: ServicesServiceNameRoute, ShareTokenRoute: ShareTokenRoute, TracesTraceIdRoute: TracesTraceIdRoute, + AgentSessionsIndexRoute: AgentSessionsIndexRoute, AlertsIndexRoute: AlertsIndexRoute, AnalyticsIndexRoute: AnalyticsIndexRoute, AnomaliesIndexRoute: AnomaliesIndexRoute, diff --git a/apps/web/src/routes/agent-sessions/index.tsx b/apps/web/src/routes/agent-sessions/index.tsx new file mode 100644 index 000000000..ca17d3bd2 --- /dev/null +++ b/apps/web/src/routes/agent-sessions/index.tsx @@ -0,0 +1,398 @@ +import { warmAtoms } from "@effect-router/core" +import { useMemo } from "react" +import { createFileRoute, Link, useNavigate } from "@tanstack/react-router" +import { Schema } from "effect" +import { AI_VENDOR_LABELS } from "@maple/domain/ai" + +import { DashboardLayout } from "@/components/layout/dashboard-layout" +import { BooleanFromStringParam, OptionalStringArrayParam } from "@/lib/search-params" +import { Result, useAtomValue } from "@/lib/effect-atom" +import { + agentSessionsFacetsResultAtom, + listAgentSessionsResultAtom, + listAgentTracesResultAtom, +} from "@/lib/services/atoms/warehouse-query-atoms" +import { TimeRangeSearchFields, applyTimeRangeSearch } from "@/components/time-range-picker/search" +import { TimeRangeHeaderControls } from "@/components/time-range-picker/time-range-header-controls" +import { + PageRefreshProvider, + useOptionalPageRefreshContext, +} from "@/components/time-range-picker/page-refresh-context" +import type { TimeRange } from "@/components/time-range-picker/types" +import { resolveEffectiveTimeRange } from "@/hooks/use-effective-time-range" +import { QueryErrorState } from "@/components/common/query-error-state" +import { FilterSection, SingleCheckboxFilter } from "@/components/filters/filter-section" +import { + FilterSidebarBody, + FilterSidebarError, + FilterSidebarFrame, + FilterSidebarHeader, + FilterSidebarLoading, +} from "@/components/filters/filter-sidebar" +import { Skeleton } from "@maple/ui/components/ui/skeleton" +import { Badge } from "@maple/ui/components/ui/badge" + +// THROWAWAY product scratchpad (owner-flagged): first UI over the AI +// classification read path, to try out what "Agent Sessions" should even be. +// Expect a full rebuild once the product shape settles — don't extract +// abstractions from this file, and don't emulate it. + +const agentSessionsSearchSchema = Schema.Struct({ + tab: Schema.optional(Schema.Literals(["sessions", "traces"])), + vendors: OptionalStringArrayParam, + services: OptionalStringArrayParam, + hasErrors: Schema.optional(Schema.Union([Schema.Boolean, BooleanFromStringParam])), + ...TimeRangeSearchFields, +}) + +type AgentSessionsSearch = typeof agentSessionsSearchSchema.Type + +const filterInputs = (search: AgentSessionsSearch, options?: { snap?: boolean }) => { + const { startTime, endTime } = resolveEffectiveTimeRange( + search.startTime, + search.endTime, + search.timePreset ?? "24h", + options, + ) + return { + startTime, + endTime, + vendors: search.vendors, + serviceNames: search.services, + hasErrors: search.hasErrors, + } +} + +const PAGE_SIZE = 50 + +export const Route = createFileRoute("/agent-sessions/")({ + component: AgentSessionsRoute, + validateSearch: Schema.toStandardSchemaV1(agentSessionsSearchSchema), + loaderDeps: ({ search }) => search, + loader: ({ context, deps }) => { + const inputs = filterInputs(deps) + const tab = deps.tab ?? "sessions" + warmAtoms(context.effectRegistry, [ + tab === "sessions" + ? listAgentSessionsResultAtom({ data: { ...inputs, limit: PAGE_SIZE, offset: 0 } }) + : listAgentTracesResultAtom({ data: { ...inputs, limit: PAGE_SIZE, offset: 0 } }), + agentSessionsFacetsResultAtom({ data: { ...inputs, tab } }), + ]) + }, +}) + +const vendorLabel = (slug: string) => (AI_VENDOR_LABELS as Record)[slug] ?? slug + +/** Write-side AiSessionKeyState enum (ai_classifier.rs), for the traces tab. */ +const KEY_STATE_LABELS: Record = { + 1: "no session rules", + 2: "not authoritative", + 3: "key absent", + 4: "key invalid", + 5: "sub-session", + 6: "in session", +} + +const formatMs = (ms: number) => { + if (!Number.isFinite(ms) || ms < 0) return "–" + if (ms < 1000) return `${Math.round(ms)}ms` + if (ms < 60_000) return `${(ms / 1000).toFixed(1)}s` + return `${Math.floor(ms / 60_000)}m ${Math.round((ms % 60_000) / 1000)}s` +} + +const formatTime = (ts: string) => ts.slice(0, 19) + +function VendorChips({ vendors }: { vendors: ReadonlyArray }) { + return ( + + {vendors.map((v) => ( + + {vendorLabel(v)} + + ))} + + ) +} + +// Provider above the page so the page can read refreshVersion — the Reload +// button must bypass the cache-grid window snap or fresh rows stay invisible +// for up to a grid interval (5m on the 24h preset). +function AgentSessionsRoute() { + const search = Route.useSearch() + return ( + + + + ) +} + +function AgentSessionsPage() { + const search = Route.useSearch() + const navigate = useNavigate({ from: Route.fullPath }) + const tab = search.tab ?? "sessions" + const refreshVersion = useOptionalPageRefreshContext()?.refreshVersion ?? 0 + + const inputs = useMemo( + () => filterInputs(search, { snap: refreshVersion === 0 }), + [ + search.startTime, + search.endTime, + search.timePreset, + search.vendors, + search.services, + search.hasErrors, + refreshVersion, + ], + ) + + const sessionsResult = useAtomValue( + listAgentSessionsResultAtom({ data: { ...inputs, limit: PAGE_SIZE, offset: 0 } }), + ) + const tracesResult = useAtomValue( + listAgentTracesResultAtom({ data: { ...inputs, limit: PAGE_SIZE, offset: 0 } }), + ) + const facetsResult = useAtomValue(agentSessionsFacetsResultAtom({ data: { ...inputs, tab } })) + + const handleTimeChange = (range: TimeRange, options?: { replace?: boolean }) => { + navigate({ replace: options?.replace, search: (prev) => applyTimeRangeSearch(prev, range) }) + } + + const hasActiveFilters = + (search.vendors?.length ?? 0) > 0 || (search.services?.length ?? 0) > 0 || search.hasErrors === true + + const sidebar = Result.builder(facetsResult) + .onInitial(() => ) + .onError((error) => ) + .onSuccess((facets, result) => ( + + + navigate({ + search: (prev) => ({ + ...prev, + vendors: undefined, + services: undefined, + hasErrors: undefined, + }), + }) + } + /> + + ({ name: f.name, count: f.count }))} + selected={search.vendors ?? []} + onChange={(selected) => + navigate({ + search: (prev) => ({ + ...prev, + vendors: selected.length > 0 ? selected : undefined, + }), + }) + } + getOptionLabel={vendorLabel} + /> + ({ name: f.name, count: f.count }))} + selected={search.services ?? []} + onChange={(selected) => + navigate({ + search: (prev) => ({ + ...prev, + services: selected.length > 0 ? selected : undefined, + }), + }) + } + /> + + navigate({ search: (prev) => ({ ...prev, hasErrors: checked ? true : undefined }) }) + } + /> + + + )) + .render() + + const listSkeleton = ( +
+ {Array.from({ length: 8 }).map((_, i) => ( +
+
+ + +
+ +
+ ))} +
+ ) + + const sessionsTable = Result.builder(sessionsResult) + .onInitial(() => listSkeleton) + .onError((error) => ) + .onSuccess(({ data }) => + data.length === 0 ? ( +

+ No agent sessions in this window. Sessions appear when AI spans carry a + session-granularity key. +

+ ) : ( + + + + + + + + + + + + + + + {data.map((row) => ( + + + + + + + + + + + ))} + +
SessionVendorsServicesTracesAI spansErrorsDurationLast activity
{row.sessionKeyHash.slice(0, 12)}… + + {row.serviceNames.join(", ")}{row.traceCount}{row.keyedSpanCount} + {row.errorCount > 0 ? ( + {row.errorCount} + ) : ( + 0 + )} + {formatMs(row.durationMs)} + {formatTime(row.endTime)} +
+ ), + ) + .render() + + const tracesTable = Result.builder(tracesResult) + .onInitial(() => listSkeleton) + .onError((error) => ) + .onSuccess(({ data }) => + data.length === 0 ? ( +

+ No AI-classified traces in this window. +

+ ) : ( + + + + + + + + + + + + + + + {data.map((row) => ( + + + + + + + + + + + ))} + +
TraceVendorsServicesSessionAI spansErrorsAI windowStart
+ + {row.firstSpanName || row.traceId.slice(0, 16)} + + + + {row.serviceNames.join(", ")} + {row.sessionKeyHash !== "" ? ( + {row.sessionKeyHash.slice(0, 12)}… + ) : ( + (KEY_STATE_LABELS[row.bestSessionKeyState] ?? "not examined") + )} + {row.aiSpanCount} + {row.errorCount > 0 ? ( + {row.errorCount} + ) : ( + 0 + )} + {formatMs(row.durationMs)} + {formatTime(row.startTime)} +
+ ), + ) + .render() + + const { startTime, endTime } = inputs + + return ( + + + + {sidebar} + + + + + +
+ {(["sessions", "traces"] as const).map((t) => ( + + ))} +
+ + + {tab === "sessions" ? sessionsTable : tracesTable} + + + + + ) +} diff --git a/package.json b/package.json index 22b63fd94..209f27847 100644 --- a/package.json +++ b/package.json @@ -38,6 +38,7 @@ "test": "turbo test", "tinybird:manifest": "bun run ./scripts/generate-tinybird-project-manifest.ts", "tinybird:manifest:check": "bun run ./scripts/generate-tinybird-project-manifest.ts --check", + "tinybird:truncate": "bun run ./scripts/truncate-tinybird-local.ts", "clickhouse:schema": "bun run ./scripts/generate-clickhouse-schema.ts && bun run ./scripts/generate-clickhouse-schema-sql.ts && bun run ./scripts/generate-clickhouse-insert-mappings.ts && bun run ./scripts/lint-clickhouse-schema.ts", "clickhouse:schema:check": "bun run ./scripts/generate-clickhouse-schema.ts --check && bun run ./scripts/generate-clickhouse-schema-sql.ts --check && bun run ./scripts/generate-clickhouse-insert-mappings.ts --check && bun run ./scripts/lint-clickhouse-schema.ts && bun run ./scripts/check-local-schema-manifest.ts", "clickhouse:schema:lint": "bun run ./scripts/lint-clickhouse-schema.ts", diff --git a/packages/domain/src/ai/index.ts b/packages/domain/src/ai/index.ts index 7b6b3a7b9..ea8ad2ec9 100644 --- a/packages/domain/src/ai/index.ts +++ b/packages/domain/src/ai/index.ts @@ -9,4 +9,11 @@ export { AI_VENDORS, AI_VENDOR_LABELS, type AiVendor } from "./vendors" +export { + normalizeAiSpan, + type AiSpanFacts, + type AiSpanInput, + type AiSpanRole, +} from "./integrations" + export { AI_VENDORS_ROLLUP_ENABLEMENT_HOUR_ENV, AI_VENDORS_ROLLUP_TABLE } from "./rollup-enablement" diff --git a/packages/domain/src/ai/integrations.test.ts b/packages/domain/src/ai/integrations.test.ts new file mode 100644 index 000000000..81f165fec --- /dev/null +++ b/packages/domain/src/ai/integrations.test.ts @@ -0,0 +1,579 @@ +import { describe, expect, it } from "vitest" +import { normalizeAiSpan } from "./integrations" + +// Attribute fixtures are lifted from the trace-capture corpus (the same +// captures the Rust classifier's rules were derived from), trimmed to the keys +// the normalizer reads plus decoys proving it ignores payload attributes. + +describe("normalizeAiSpan", () => { + it("base: semconv chat span (unknown:genai bucket)", () => { + const facts = normalizeAiSpan({ + vendor: "unknown:genai", + spanName: "chat openai/gpt-5.6-luna", + attributes: { + "gen_ai.operation.name": "chat", + "gen_ai.request.model": "openai/gpt-5.6-luna", + "gen_ai.usage.input_tokens": "4927", + "gen_ai.usage.output_tokens": "64", + "gen_ai.usage.cache_read.input_tokens": "0", + "gen_ai.usage.cache_creation.input_tokens": "4924", + }, + }) + expect(facts).toEqual({ + role: "llm", + operation: "chat", + providerName: null, + model: "openai/gpt-5.6-luna", + inputTokens: 4927, + outputTokens: 64, + cacheReadTokens: 0, + cacheCreationTokens: 4924, + reasoningTokens: null, + costUsd: null, + sessionKey: null, + compacted: null, + previousResponseId: null, + agentName: null, + agentId: null, + agentDescription: null, + agentVersion: null, + workflowName: null, + toolName: null, + toolCallId: null, + toolType: null, + toolDescription: null, + toolDefinitions: null, + responseId: null, + finishReasons: null, + responseStatus: null, + timeToFirstChunk: null, + errorType: null, + systemInstructions: null, + promptName: null, + promptVersion: null, + promptVariables: null, + inputText: null, + outputText: null, + }) + }) + + it("base: identity, agent/workflow/tool names and response metadata", () => { + const tool = normalizeAiSpan({ + vendor: "unknown:genai", + spanName: "execute_tool get_weather", + attributes: { + "gen_ai.operation.name": "execute_tool", + "gen_ai.provider.name": "anthropic", + "gen_ai.agent.name": "Math Tutor", + "gen_ai.agent.id": "asst_01H9", + "gen_ai.workflow.name": "research_crew", + "gen_ai.tool.name": "get_weather", + "gen_ai.tool.call.id": "call_9f2c", + "gen_ai.tool.type": "function", + }, + }) + expect(tool.role).toBe("tool") + expect(tool.operation).toBe("execute_tool") + expect(tool.providerName).toBe("anthropic") + expect(tool.agentName).toBe("Math Tutor") + expect(tool.agentId).toBe("asst_01H9") + expect(tool.workflowName).toBe("research_crew") + expect(tool.toolName).toBe("get_weather") + expect(tool.toolCallId).toBe("call_9f2c") + expect(tool.toolType).toBe("function") + + const chat = normalizeAiSpan({ + vendor: "unknown:genai", + spanName: "chat gpt-4o", + attributes: { + "gen_ai.operation.name": "chat", + "gen_ai.response.id": "chatcmpl-123", + "gen_ai.usage.output_tokens": "512", + // A subset of output_tokens — surfaced, never summed into a total. + "gen_ai.usage.reasoning.output_tokens": "448", + }, + }) + expect(chat.responseId).toBe("chatcmpl-123") + expect(chat.outputTokens).toBe(512) + expect(chat.reasoningTokens).toBe(448) + }) + + it("base: newer standard operations get their tier, the rest stay honest", () => { + const workflow = normalizeAiSpan({ + vendor: "unknown:genai", + spanName: "invoke_workflow research", + attributes: { "gen_ai.operation.name": "invoke_workflow" }, + }) + expect(workflow.role).toBe("workflow") + + const plan = normalizeAiSpan({ + vendor: "unknown:genai", + spanName: "plan Math Tutor", + attributes: { "gen_ai.operation.name": "plan" }, + }) + expect(plan.role).toBe("agent") + + // Retrieval and the memory family have no tier of ours; `operation` is + // what keeps them identifiable. + const retrieval = normalizeAiSpan({ + vendor: "unknown:genai", + spanName: "retrieval docs", + attributes: { "gen_ai.operation.name": "retrieval" }, + }) + expect(retrieval.role).toBe("other") + expect(retrieval.operation).toBe("retrieval") + }) + + it("base: finish reasons survive either wire encoding", () => { + const attrs = (value: string) => ({ + vendor: "unknown:genai", + spanName: "chat", + attributes: { "gen_ai.operation.name": "chat", "gen_ai.response.finish_reasons": value }, + }) + // The collector's JSON encoding of a string[] attribute. + expect(normalizeAiSpan(attrs('["stop","length"]')).finishReasons).toEqual(["stop", "length"]) + // Exporters that flatten a single-element list to the bare value. + expect(normalizeAiSpan(attrs("stop")).finishReasons).toEqual(["stop"]) + // Neither JSON nor a list of strings: keep the raw value rather than drop it. + expect(normalizeAiSpan(attrs('["stop"')).finishReasons).toEqual(['["stop"']) + expect(normalizeAiSpan(attrs("[1,2]")).finishReasons).toEqual(["[1,2]"]) + expect(normalizeAiSpan(attrs("")).finishReasons).toBeNull() + expect( + normalizeAiSpan({ vendor: "unknown:genai", spanName: "chat", attributes: {} }) + .finishReasons, + ).toBeNull() + }) + + it("base: lifecycle, error, streaming and prompt-template facts", () => { + const facts = normalizeAiSpan({ + vendor: "unknown:genai", + spanName: "chat gpt-4o", + attributes: { + "gen_ai.operation.name": "chat", + "gen_ai.request.previous_response.id": "resp_01H9", + "gen_ai.response.status": "in_progress", + "gen_ai.response.time_to_first_chunk": "0.482", + // The borrowed Stable attribute, outside the gen_ai.* family. + "error.type": "rate_limit_exceeded", + "gen_ai.agent.description": "Answers algebra questions", + "gen_ai.agent.version": "3.1.0", + "gen_ai.tool.description": "Looks up the current forecast", + "gen_ai.tool.definitions": '[{"type":"function","name":"get_weather"}]', + "gen_ai.system_instructions": "You are a helpful assistant.", + "gen_ai.prompt.name": "weather_briefing", + "gen_ai.prompt.version": "7", + }, + }) + expect(facts.previousResponseId).toBe("resp_01H9") + expect(facts.responseStatus).toBe("in_progress") + expect(facts.timeToFirstChunk).toBe(0.482) + expect(facts.errorType).toBe("rate_limit_exceeded") + expect(facts.agentDescription).toBe("Answers algebra questions") + expect(facts.agentVersion).toBe("3.1.0") + expect(facts.toolDescription).toBe("Looks up the current forecast") + // Raw JSON on purpose — rendering the tool list is a UI concern. + expect(facts.toolDefinitions).toBe('[{"type":"function","name":"get_weather"}]') + expect(facts.systemInstructions).toBe("You are a helpful assistant.") + expect(facts.promptName).toBe("weather_briefing") + expect(facts.promptVersion).toBe("7") + }) + + it("base: compaction is a string boolean on the wire", () => { + const compacted = (attributes: Record) => + normalizeAiSpan({ vendor: "unknown:genai", spanName: "chat", attributes }).compacted + expect(compacted({ "gen_ai.conversation.compacted": "true" })).toBe(true) + // Spec says instrumentations never emit false, but wire data is not a promise. + expect(compacted({ "gen_ai.conversation.compacted": "false" })).toBe(false) + // Anything else is not a boolean the exporter meant — absent, not falsey. + expect(compacted({ "gen_ai.conversation.compacted": "TRUE" })).toBeNull() + expect(compacted({ "gen_ai.conversation.compacted": "1" })).toBeNull() + expect(compacted({ "gen_ai.conversation.compacted": "" })).toBeNull() + expect(compacted({})).toBeNull() + }) + + it("base: prompt variables are collected off the key prefix", () => { + const facts = normalizeAiSpan({ + vendor: "unknown:genai", + spanName: "chat", + attributes: { + "gen_ai.operation.name": "chat", + "gen_ai.prompt.name": "weather_briefing", + "gen_ai.prompt.variable.city": "Amsterdam", + "gen_ai.prompt.variable.tone": "terse", + // Neighbours under the prompt namespace are not variables. + "gen_ai.prompt.version": "7", + }, + }) + expect(facts.promptVariables).toEqual({ city: "Amsterdam", tone: "terse" }) + + const none = normalizeAiSpan({ + vendor: "unknown:genai", + spanName: "chat", + attributes: { "gen_ai.operation.name": "chat" }, + }) + expect(none.promptVariables).toBeNull() + }) + + it("base: response model wins over request model", () => { + const facts = normalizeAiSpan({ + vendor: "unknown:genai", + spanName: "chat", + attributes: { + "gen_ai.request.model": "gpt-4o", + "gen_ai.response.model": "gpt-4o-2024-08-06", + }, + }) + expect(facts.model).toBe("gpt-4o-2024-08-06") + }) + + it("base: a vendor without an integration falls through untouched", () => { + // A crewai orchestration span: no gen_ai.* payload at all — every fact + // honestly unknown, tokens come from its openinference-openai children. + const facts = normalizeAiSpan({ + vendor: "crewai", + spanName: "research_crew.kickoff", + attributes: { crew_key: "8b6f4a", "session.id": "run-oi-17" }, + }) + expect(facts.role).toBe("other") + expect(facts.inputTokens).toBeNull() + // `session.id` is crewai's keyed attribute, but without an integration the + // base must not guess a display key from a non-semconv spelling. + expect(facts.sessionKey).toBeNull() + }) + + it("mastra: span-type roles override, semconv tokens stay", () => { + const inference = normalizeAiSpan({ + vendor: "mastra", + spanName: "model_inference weather_worker", + attributes: { + "mastra.span.type": "model_inference", + "gen_ai.operation.name": "model_inference", + "mastra.metadata.runId": "cc714c79-e4c9-4877-baf4-0b3789761ed9", + }, + }) + expect(inference.role).toBe("llm") + + const step = normalizeAiSpan({ + vendor: "mastra", + spanName: "workflow_step amsterdam_briefing", + attributes: { "mastra.span.type": "workflow_step" }, + }) + expect(step.role).toBe("workflow") + + // The token-bearing `chat` span is pure semconv — base answers everything, + // including the session display key from `gen_ai.conversation.id`. + const chat = normalizeAiSpan({ + vendor: "mastra", + spanName: "chat openai/gpt-4o-mini", + attributes: { + "mastra.span.type": "model_generation", + "gen_ai.operation.name": "chat", + "gen_ai.request.model": "openai/gpt-4o-mini", + "gen_ai.response.model": "openai/gpt-4o-mini", + "gen_ai.usage.input_tokens": "89", + "gen_ai.usage.output_tokens": "88", + "gen_ai.conversation.id": "support-thread-42", + }, + }) + expect(chat.role).toBe("llm") + expect(chat.inputTokens).toBe(89) + expect(chat.sessionKey).toBe("support-thread-42") + }) + + it("claude_agent_sdk: bare-key dialect", () => { + const llm = normalizeAiSpan({ + vendor: "claude_agent_sdk", + spanName: "claude_code.llm_request", + attributes: { + "span.type": "llm_request", + model: "anthropic/claude-haiku-4.5", + input_tokens: "3550", + output_tokens: "408", + cache_read_tokens: "4248", + cache_creation_tokens: "2025", + "session.id": "f0f992b0-c9f7-4d8c-8c93-beaea0904dda", + "gen_ai.system": "anthropic", + }, + }) + expect(llm).toEqual({ + role: "llm", + operation: "llm_request", + providerName: null, + model: "anthropic/claude-haiku-4.5", + inputTokens: 3550, + outputTokens: 408, + cacheReadTokens: 4248, + cacheCreationTokens: 2025, + reasoningTokens: null, + costUsd: null, + sessionKey: "f0f992b0-c9f7-4d8c-8c93-beaea0904dda", + compacted: null, + previousResponseId: null, + agentName: null, + agentId: null, + agentDescription: null, + agentVersion: null, + workflowName: null, + toolName: null, + toolCallId: null, + toolType: null, + toolDescription: null, + toolDefinitions: null, + responseId: null, + finishReasons: null, + responseStatus: null, + timeToFirstChunk: null, + errorType: null, + systemInstructions: null, + promptName: null, + promptVersion: null, + promptVariables: null, + inputText: null, + outputText: null, + }) + + const gate = normalizeAiSpan({ + vendor: "claude_agent_sdk", + spanName: "claude_code.tool.blocked_on_user", + attributes: { + "span.type": "tool.blocked_on_user", + "session.id": "f0f992b0-c9f7-4d8c-8c93-beaea0904dda", + }, + }) + expect(gate.role).toBe("tool") + expect(gate.inputTokens).toBeNull() + + const turn = normalizeAiSpan({ + vendor: "claude_agent_sdk", + spanName: "claude_code.interaction", + attributes: { "span.type": "interaction", "session.id": "s" }, + }) + expect(turn.role).toBe("agent") + }) + + it("vercel_ai_sdk: GenAI dialect is base + agent_step", () => { + const step = normalizeAiSpan({ + vendor: "vercel_ai_sdk", + spanName: "step 1", + attributes: { + "gen_ai.operation.name": "agent_step", + "ai.settings.context.eve.session.id": "wrun_01KZAAFFZRHHRYC8MY9MDANASQ", + }, + }) + expect(step.role).toBe("agent") + expect(step.sessionKey).toBe("wrun_01KZAAFFZRHHRYC8MY9MDANASQ") + + const chat = normalizeAiSpan({ + vendor: "vercel_ai_sdk", + spanName: "chat openai/gpt-4o-mini", + attributes: { + "gen_ai.operation.name": "chat", + "gen_ai.request.model": "openai/gpt-4o-mini", + "gen_ai.usage.input_tokens": "258", + "gen_ai.usage.output_tokens": "83", + // Detail spellings ride along in the GenAI dialect — must not shadow. + "ai.usage.inputTokenDetails.noCacheTokens": "258", + }, + }) + expect(chat.role).toBe("llm") + expect(chat.inputTokens).toBe(258) + expect(chat.model).toBe("openai/gpt-4o-mini") + }) + + it("vercel_ai_sdk: legacy dialect spells everything under ai.*", () => { + // An umbrella span: ONLY `ai.*` spellings — the base finds nothing. Its + // usage repeats the child doGenerate aggregates, so the role is the agent + // tier (llm-tier totals would double-count it). + const umbrella = normalizeAiSpan({ + vendor: "vercel_ai_sdk", + spanName: "ai.generateText", + attributes: { + "ai.operationId": "ai.generateText", + "ai.model.id": "openai/gpt-4o-mini", + "ai.usage.inputTokens": "209", + "ai.usage.outputTokens": "44", + "ai.usage.cachedInputTokens": "0", + }, + }) + expect(umbrella).toEqual({ + role: "agent", + operation: "ai.generateText", + providerName: null, + model: "openai/gpt-4o-mini", + inputTokens: 209, + outputTokens: 44, + cacheReadTokens: 0, + cacheCreationTokens: null, + reasoningTokens: null, + costUsd: null, + sessionKey: null, + compacted: null, + previousResponseId: null, + agentName: null, + agentId: null, + agentDescription: null, + agentVersion: null, + workflowName: null, + toolName: null, + toolCallId: null, + toolType: null, + toolDescription: null, + toolDefinitions: null, + responseId: null, + finishReasons: null, + responseStatus: null, + timeToFirstChunk: null, + errorType: null, + systemInstructions: null, + promptName: null, + promptVersion: null, + promptVariables: null, + inputText: null, + outputText: null, + }) + + // A doGenerate span carries BOTH spellings; the semconv one must win so + // the two dialects can never double-report. + const doGenerate = normalizeAiSpan({ + vendor: "vercel_ai_sdk", + spanName: "ai.generateText.doGenerate", + attributes: { + "ai.operationId": "ai.generateText.doGenerate", + "ai.model.id": "openai/gpt-4o-mini", + "ai.usage.inputTokens": "209", + "gen_ai.usage.input_tokens": "209", + "gen_ai.usage.output_tokens": "44", + "gen_ai.request.model": "openai/gpt-4o-mini", + }, + }) + expect(doGenerate.role).toBe("llm") + expect(doGenerate.inputTokens).toBe(209) + expect(doGenerate.outputTokens).toBe(44) + + const toolCall = normalizeAiSpan({ + vendor: "vercel_ai_sdk", + spanName: "ai.toolCall", + attributes: { "ai.operationId": "ai.toolCall" }, + }) + expect(toolCall.role).toBe("tool") + }) + + it("content: each vendor's conversational spellings resolve to input/output", () => { + // Semconv messages (GenAI dialect / mastra chat spans) — base. + const semconv = normalizeAiSpan({ + vendor: "vercel_ai_sdk", + spanName: "chat openai/gpt-4o-mini", + attributes: { + "gen_ai.operation.name": "chat", + "gen_ai.input.messages": '[{"role":"user","parts":[{"type":"text","content":"hi"}]}]', + "gen_ai.output.messages": + '[{"role":"assistant","parts":[{"type":"text","content":"hello"}]}]', + }, + }) + expect(semconv.inputText).toContain('"hi"') + expect(semconv.outputText).toContain('"hello"') + + // Semconv tool call/result — base. + const tool = normalizeAiSpan({ + vendor: "vercel_ai_sdk", + spanName: "execute_tool get_weather", + attributes: { + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.call.arguments": '{"city":"Amsterdam"}', + "gen_ai.tool.call.result": '{"temperature_c":21}', + }, + }) + expect(tool.inputText).toBe('{"city":"Amsterdam"}') + expect(tool.outputText).toBe('{"temperature_c":21}') + + // Legacy AI SDK spellings. + const legacy = normalizeAiSpan({ + vendor: "vercel_ai_sdk", + spanName: "ai.generateText", + attributes: { + "ai.operationId": "ai.generateText", + "ai.prompt.messages": '[{"role":"user","content":"hi"}]', + "ai.response.text": "hello there", + }, + }) + expect(legacy.inputText).toBe('[{"role":"user","content":"hi"}]') + expect(legacy.outputText).toBe("hello there") + + // Mastra's per-span-type keys. + const mastra = normalizeAiSpan({ + vendor: "mastra", + spanName: "agent_run orchestrator", + attributes: { + "mastra.span.type": "agent_run", + "mastra.agent_run.input": "Produce a briefing", + "mastra.agent_run.output": '{"text":"**Target City: Amsterdam**"}', + }, + }) + expect(mastra.inputText).toBe("Produce a briefing") + expect(mastra.outputText).toBe('{"text":"**Target City: Amsterdam**"}') + + // Claude's turn-root user_prompt. + const claude = normalizeAiSpan({ + vendor: "claude_agent_sdk", + spanName: "claude_code.interaction", + attributes: { "span.type": "interaction", user_prompt: "Produce a briefing" }, + }) + expect(claude.inputText).toBe("Produce a briefing") + }) + + it("operation: every vendor's own span-type spelling reaches the fact", () => { + // Mastra's span type, which is also what its `gen_ai.operation.name` says. + const mastra = normalizeAiSpan({ + vendor: "mastra", + spanName: "workflow_step amsterdam_briefing", + attributes: { "mastra.span.type": "workflow_step" }, + }) + expect(mastra.operation).toBe("workflow_step") + + const claude = normalizeAiSpan({ + vendor: "claude_agent_sdk", + spanName: "claude_code.tool.execution", + attributes: { "span.type": "tool.execution", "session.id": "s" }, + }) + expect(claude.operation).toBe("tool.execution") + + // Legacy AI SDK dialect: `ai.operationId` is the operation name. + const legacy = normalizeAiSpan({ + vendor: "vercel_ai_sdk", + spanName: "ai.generateText.doGenerate", + attributes: { "ai.operationId": "ai.generateText.doGenerate" }, + }) + expect(legacy.operation).toBe("ai.generateText.doGenerate") + + // GenAI dialect: no `ai.operationId`, so the base's semconv reading stands. + const genai = normalizeAiSpan({ + vendor: "vercel_ai_sdk", + spanName: "chat openai/gpt-4o-mini", + attributes: { "gen_ai.operation.name": "chat" }, + }) + expect(genai.operation).toBe("chat") + + // eve's turn root has no operation key at all — null, not a guess. + const eve = normalizeAiSpan({ + vendor: "eve", + spanName: "ai.eve.turn", + attributes: { "eve.session.id": "wrun_01KZ" }, + }) + expect(eve.operation).toBeNull() + }) + + it("eve: the turn root is the agent tier and carries the session key", () => { + const turn = normalizeAiSpan({ + vendor: "eve", + spanName: "ai.eve.turn", + attributes: { + "eve.session.id": "wrun_01KZAAFFZRHHRYC8MY9MDANASQ", + "eve.turn.id": "turn_1", + "ai.telemetry.functionId": "slack-agent", + }, + }) + expect(turn.role).toBe("agent") + expect(turn.sessionKey).toBe("wrun_01KZAAFFZRHHRYC8MY9MDANASQ") + expect(turn.inputTokens).toBeNull() + }) +}) diff --git a/packages/domain/src/ai/integrations.ts b/packages/domain/src/ai/integrations.ts new file mode 100644 index 000000000..1b4164572 --- /dev/null +++ b/packages/domain/src/ai/integrations.ts @@ -0,0 +1,391 @@ +// Per-span vendor integrations for the AI read path. +// +// A normalized AI span is the raw span merged with the *facts* this module +// derives from its attributes. Dispatch is strictly per span on the stamped +// `AiVendor` slug — never per trace or per session: a single trace routinely +// mixes vendors (CrewAI orchestration spans parenting openinference-openai LLM +// spans is the proven normal case), and in that shape the token counts live on +// the child spans of a *different* vendor than the session-keyed ones. +// +// The base normalizer speaks standard `gen_ai.*` semconv. A vendor integration +// overrides only what its vendor spells differently; every field it does not +// return keeps the base's answer. Vendors without an integration — including +// the `unknown:*` buckets — run the base alone. Attribute spellings are +// verified against the trace-capture corpus (`trace-capture` repo), not docs: +// several vendors' docs disagree with their wire format. + +export type AiSpanRole = "llm" | "tool" | "agent" | "workflow" | "other" + +export interface AiSpanInput { + /** The stamped `AiVendor` slug. Typed `string`, not `AiVendor`: a slug the + * Rust classifier ships before this package redeploys must degrade to the + * base normalizer, not throw. */ + readonly vendor: string + readonly spanName: string + readonly attributes: Readonly> +} + +/** What the integration layer can know about an AI span. `null` = the vendor + * does not put this fact on this span (tokens live only on LLM-call spans, + * session keys only on session-authoritative ones). */ +export interface AiSpanFacts { + /** The five tiers a session rollup counts in: `llm` (a model call), `tool`, + * `agent`, `workflow`, and `other` for everything with no tier of ours. */ + readonly role: AiSpanRole + /** The operation name as the vendor spelled it, unmapped. `role` coarsens + * this into five tiers; this keeps the distinction the tiers throw away + * (`chat` vs `embeddings`, `workflow_step` vs `workflow_run`) and is the + * only fact left for operations with no tier of ours (`retrieval`, the + * memory family). */ + readonly operation: string | null + /** Semconv's own provider discriminator (`anthropic`, `openai`, …). Not the + * same question as our stamped `AiVendor`, which names the instrumentation: + * a mastra span reports provider `openai`. */ + readonly providerName: string | null + /** The model as the provider named it, response spelling preferred over the + * request's: a request for `gpt-4o` is answered by `gpt-4o-2024-08-06`, and + * the resolved one is what a cost or drift question needs. */ + readonly model: string | null + /** Billed prompt tokens. INCLUDES the two cache counts below, so a total is + * `input + output` and never adds them again. */ + readonly inputTokens: number | null + /** Billed completion tokens; includes `reasoningTokens`. */ + readonly outputTokens: number | null + /** Prompt tokens served from the provider's cache. A subset of `inputTokens`. */ + readonly cacheReadTokens: number | null + /** Prompt tokens written to the provider's cache. A subset of `inputTokens`. */ + readonly cacheCreationTokens: number | null + /** Thinking tokens. A SUBSET of `outputTokens` per spec, never an addend: + * adding it into any total double-counts. Display only. */ + readonly reasoningTokens: number | null + /** The provider's own cost figure in USD. `gen_ai.usage.cost` is a de-facto + * vendor extension, not semconv — most exporters omit it and we do not + * price tokens ourselves, so null is the common case. */ + readonly costUsd: number | null + /** The plaintext session key, for display — the hash column is opaque. Lives + * in span attributes under a vendor-specific key on the spans that carry it. */ + readonly sessionKey: string | null + /** True when the agent compacted its context before this call. Spec says + * instrumentations set it only when compaction was reliably detected and + * never set it to false, so `false` is rare-but-possible wire data rather + * than a promise that no compaction happened; null = nothing was exported. */ + readonly compacted: boolean | null + /** The chained-context id this request continued from (OpenAI Responses + * `previous_response_id`, Google Interactions `previous_interaction_id`) — + * enough to stitch a session together when no conversation id exists. */ + readonly previousResponseId: string | null + /** Names for the tiers `role` identifies. `agentId` is a *hosted* agent's + * stable resource id (assistant id, Bedrock ARN) — semconv explicitly keeps + * transient in-process instances out of it, so it is null on most spans. */ + readonly agentName: string | null + readonly agentId: string | null + /** Free-form agent description and the agent definition's version string. */ + readonly agentDescription: string | null + readonly agentVersion: string | null + readonly workflowName: string | null + /** Tool-span identity. `toolCallId` is the provider's call id, which links + * this execution back to its request part in the parent LLM span's output + * messages; `toolType` is `function` | `extension` | `datastore`. */ + readonly toolName: string | null + readonly toolCallId: string | null + readonly toolType: string | null + /** The tool's own description, as advertised to the model. */ + readonly toolDescription: string | null + /** The tool list available to the model/agent on this call, as a raw JSON + * string (same rationale as `inputText`: rendering is a UI concern). + * Opt-in and potentially large. */ + readonly toolDefinitions: string | null + /** The provider's completion id (`chatcmpl-…`) — the handle for quoting a + * generation back to the provider. */ + readonly responseId: string | null + /** Why each generation stopped (`["stop"]`, `["length"]`), one entry per + * generation — truncation vs stop vs filter without parsing message JSON. */ + readonly finishReasons: readonly string[] | null + /** Lifecycle of a possibly background/long-running generation. Well-known + * values `queued`, `in_progress`, `completed`, `incomplete`, `failed`, + * `cancelled`; custom values are allowed. Distinct from `finishReasons`, + * which says why the model stopped once it produced output. */ + readonly responseStatus: string | null + /** Seconds (double) from issuing the request to the first streamed chunk. + * Only on streaming requests. */ + readonly timeToFirstChunk: number | null + /** Provider error code, exception class name, or another low-cardinality + * identifier, with `_OTHER` as the fallback. Set only when the operation + * failed. Borrowed from the Stable `error.type`, not the `gen_ai.*` family. */ + readonly errorType: string | null + /** The system prompt, raw, for APIs that take it separately from the chat + * history. Opt-in and sensitive; instructions embedded in the history land + * in `inputText` instead. */ + readonly systemInstructions: string | null + /** Prompt-template identity: the registered template's name and version. */ + readonly promptName: string | null + readonly promptVersion: string | null + /** The template's variables, keyed by variable name with the serialized + * value — collected from every `gen_ai.prompt.variable.*` attribute. + * Opt-in; null when the span carries no such key. */ + readonly promptVariables: Readonly> | null + /** Conversational content, as the vendor recorded it — the input side (chat + * messages JSON, a plain prompt, tool-call arguments) and the output side + * (messages JSON, response text, tool result). Raw strings on purpose: the + * formats differ per vendor and rendering them is a UI concern; this layer + * only knows WHICH attribute holds them. `null` = not exported. */ + readonly inputText: string | null + readonly outputText: string | null +} + +/** A vendor's overrides: mutable while being built, partial because every + * omitted field keeps the base's answer. */ +type AiSpanFactOverrides = { -readonly [K in keyof AiSpanFacts]?: AiSpanFacts[K] } + +type AiVendorIntegration = (span: AiSpanInput) => AiSpanFactOverrides + +const num = (attrs: AiSpanInput["attributes"], key: string): number | null => { + const raw = attrs[key] + if (raw === undefined || raw === "") return null + const value = Number(raw) + return Number.isFinite(value) ? value : null +} + +const str = (attrs: AiSpanInput["attributes"], key: string): string | null => { + const raw = attrs[key] + return raw === undefined || raw === "" ? null : raw +} + +// Booleans reach us as the string map's `"true"` / `"false"`; anything else is +// not a boolean the exporter meant, so it reads as absent rather than falsey. +const bool = (attrs: AiSpanInput["attributes"], key: string): boolean | null => { + const raw = attrs[key] + return raw === "true" ? true : raw === "false" ? false : null +} + +// Array-valued attributes reach us through a string map, so the collector's +// JSON encoding (`["stop","length"]`) is what lands. Not every exporter encodes +// though — a one-element list often arrives as the bare value — so anything +// that is not a JSON array of strings reads as a single element rather than +// being dropped. +const strArray = (attrs: AiSpanInput["attributes"], key: string): readonly string[] | null => { + const raw = attrs[key] + if (raw === undefined || raw === "") return null + if (raw.startsWith("[")) { + try { + const parsed: unknown = JSON.parse(raw) + if (Array.isArray(parsed) && parsed.every((item): item is string => typeof item === "string")) + return parsed + } catch { + // Not JSON after all; the raw string is still a value. + } + } + return [raw] +} + +// Template variables are one attribute per variable, so the only way to read +// them is a key scan: the suffix after the prefix is the variable name. +const PROMPT_VARIABLE_PREFIX = "gen_ai.prompt.variable." + +const promptVariables = ( + attrs: AiSpanInput["attributes"], +): Readonly> | null => { + let collected: Record | null = null + for (const [key, value] of Object.entries(attrs)) { + if (!key.startsWith(PROMPT_VARIABLE_PREFIX)) continue + collected ??= {} + collected[key.slice(PROMPT_VARIABLE_PREFIX.length)] = value + } + return collected +} + +// Semconv `gen_ai.operation.name` → role. Anything unlisted is "other" — an +// honest bucket, not a guess: `retrieval`, `fetch_response` and the memory +// family are real semconv operations with no tier of ours to land in, and +// forcing them into one would misreport every session rollup. `operation` +// keeps their exact name either way. +const SEMCONV_OPERATION_ROLES: Readonly> = { + chat: "llm", + text_completion: "llm", + generate_content: "llm", + embeddings: "llm", + execute_tool: "tool", + invoke_agent: "agent", + create_agent: "agent", + plan: "agent", + invoke_workflow: "workflow", +} + +const baseNormalize = ({ attributes }: AiSpanInput): AiSpanFacts => ({ + role: SEMCONV_OPERATION_ROLES[attributes["gen_ai.operation.name"] ?? ""] ?? "other", + operation: str(attributes, "gen_ai.operation.name"), + providerName: str(attributes, "gen_ai.provider.name"), + model: str(attributes, "gen_ai.response.model") ?? str(attributes, "gen_ai.request.model"), + inputTokens: num(attributes, "gen_ai.usage.input_tokens"), + outputTokens: num(attributes, "gen_ai.usage.output_tokens"), + cacheReadTokens: num(attributes, "gen_ai.usage.cache_read.input_tokens"), + cacheCreationTokens: num(attributes, "gen_ai.usage.cache_creation.input_tokens"), + reasoningTokens: num(attributes, "gen_ai.usage.reasoning.output_tokens"), + costUsd: num(attributes, "gen_ai.usage.cost"), + sessionKey: str(attributes, "gen_ai.conversation.id"), + compacted: bool(attributes, "gen_ai.conversation.compacted"), + previousResponseId: str(attributes, "gen_ai.request.previous_response.id"), + agentName: str(attributes, "gen_ai.agent.name"), + agentId: str(attributes, "gen_ai.agent.id"), + agentDescription: str(attributes, "gen_ai.agent.description"), + agentVersion: str(attributes, "gen_ai.agent.version"), + workflowName: str(attributes, "gen_ai.workflow.name"), + toolName: str(attributes, "gen_ai.tool.name"), + toolCallId: str(attributes, "gen_ai.tool.call.id"), + toolType: str(attributes, "gen_ai.tool.type"), + toolDescription: str(attributes, "gen_ai.tool.description"), + toolDefinitions: str(attributes, "gen_ai.tool.definitions"), + responseId: str(attributes, "gen_ai.response.id"), + finishReasons: strArray(attributes, "gen_ai.response.finish_reasons"), + responseStatus: str(attributes, "gen_ai.response.status"), + timeToFirstChunk: num(attributes, "gen_ai.response.time_to_first_chunk"), + errorType: str(attributes, "error.type"), + systemInstructions: str(attributes, "gen_ai.system_instructions"), + promptName: str(attributes, "gen_ai.prompt.name"), + promptVersion: str(attributes, "gen_ai.prompt.version"), + promptVariables: promptVariables(attributes), + inputText: + str(attributes, "gen_ai.input.messages") ?? str(attributes, "gen_ai.tool.call.arguments"), + outputText: + str(attributes, "gen_ai.output.messages") ?? str(attributes, "gen_ai.tool.call.result"), +}) + +// mastra — tokens/model/session key are clean semconv; the one divergence is +// that `gen_ai.operation.name` doubles as Mastra's span-type field, so most of +// its values are not semconv operation names. `mastra.span.type` carries the +// same value and survives attribute-family degradation, so it is the input. +const mastraIntegration: AiVendorIntegration = ({ attributes }) => { + const spanType = attributes["mastra.span.type"] ?? attributes["gen_ai.operation.name"] ?? "" + const facts: AiSpanFactOverrides = {} + // The span type IS mastra's operation name, and it is the spelling that + // survives when the `gen_ai.*` family degrades away. + if (spanType !== "") facts.operation = spanType + if (spanType.startsWith("model_")) facts.role = "llm" + else if (spanType.startsWith("workflow_") || spanType === "invoke_workflow") + facts.role = "workflow" + else if (spanType === "agent_run") facts.role = "agent" + // Content rides under `mastra..input` / `.output` on every span + // type (agent_run, model_step, workflow_*); `chat` spans carry the semconv + // messages instead, which the base already reads. + const input = str(attributes, `mastra.${spanType}.input`) + if (input !== null) facts.inputText = input + const output = str(attributes, `mastra.${spanType}.output`) + if (output !== null) facts.outputText = output + return facts +} + +// claude_agent_sdk — the whole dialect is bare, unnamespaced keys: `span.type`, +// `model`, `input_tokens`, `session.id`. (`gen_ai.request.model` also rides +// along, so the base covers model too; tokens and the session key do not.) +const CLAUDE_SPAN_TYPE_ROLES: Readonly> = { + llm_request: "llm", + tool: "tool", + "tool.execution": "tool", + "tool.blocked_on_user": "tool", + // The per-turn root. `subagent.spawn` is unreachable in the shipped CLI but + // named in the dialect; both are the agent tier. + interaction: "agent", + "subagent.spawn": "agent", +} + +const claudeAgentSdkIntegration: AiVendorIntegration = ({ attributes }) => { + const facts: AiSpanFactOverrides = { + role: CLAUDE_SPAN_TYPE_ROLES[attributes["span.type"] ?? ""] ?? "other", + operation: str(attributes, "span.type"), + sessionKey: str(attributes, "session.id"), + } + const model = str(attributes, "model") + if (model !== null) facts.model = model + const inputTokens = num(attributes, "input_tokens") + if (inputTokens !== null) { + facts.inputTokens = inputTokens + facts.outputTokens = num(attributes, "output_tokens") + facts.cacheReadTokens = num(attributes, "cache_read_tokens") + facts.cacheCreationTokens = num(attributes, "cache_creation_tokens") + } + // The CLI exports no message content; the turn root's `user_prompt` (behind + // OTEL_LOG_USER_PROMPTS) is the only conversational text in the dialect. + const userPrompt = str(attributes, "user_prompt") + if (userPrompt !== null) facts.inputText = userPrompt + return facts +} + +// vercel_ai_sdk — one vendor, two mutually exclusive wire dialects. The GenAI +// dialect is semconv and the base handles it (plus `agent_step`, an +// SDK-invented operation name). The legacy dialect spells everything under +// `ai.*` (`ai.usage.inputTokens`, `ai.model.id`, `ai.operationId`) and its +// umbrella spans (`ai.generateText`) carry ONLY the `ai.*` spellings. The +// session key is the eve-hosted case: the framework splices its session id +// into the runtime context as `ai.settings.context.eve.session.id`; a plain +// AI SDK span has no session-key convention at all. +const vercelAiSdkIntegration: AiVendorIntegration = ({ attributes }) => { + const facts: AiSpanFactOverrides = {} + const operationId = attributes["ai.operationId"] + if (operationId === "ai.toolCall") facts.role = "tool" + else if (operationId !== undefined) { + // Legacy dialect: the `.doGenerate`/`.doStream`/`.doEmbed` leaf is the model + // call; the umbrella (`ai.generateText`, …) REPEATS its leaves' aggregated + // usage, so it must stay out of the llm tier or session totals double-count. + facts.role = /\.do[A-Z]/.test(operationId) ? "llm" : "agent" + } else if (attributes["gen_ai.operation.name"] === "agent_step") facts.role = "agent" + // The legacy dialect's operation name; the GenAI dialect carries the semconv + // one the base already read. + if (operationId !== undefined && operationId !== "") facts.operation = operationId + const model = str(attributes, "ai.model.id") + const baseHasModel = + attributes["gen_ai.response.model"] !== undefined || + attributes["gen_ai.request.model"] !== undefined + if (model !== null && !baseHasModel) facts.model = model + const inputTokens = num(attributes, "ai.usage.inputTokens") + if (inputTokens !== null && attributes["gen_ai.usage.input_tokens"] === undefined) { + facts.inputTokens = inputTokens + facts.outputTokens = num(attributes, "ai.usage.outputTokens") + facts.cacheReadTokens = num(attributes, "ai.usage.cachedInputTokens") + } + const sessionKey = str(attributes, "ai.settings.context.eve.session.id") + if (sessionKey !== null) facts.sessionKey = sessionKey + // Legacy-dialect content spellings; the GenAI dialect uses the semconv + // message keys the base reads. + const input = + str(attributes, "ai.prompt.messages") ?? + str(attributes, "ai.prompt") ?? + str(attributes, "ai.toolCall.args") + if (input !== null && facts.inputText === undefined && attributes["gen_ai.input.messages"] === undefined) + facts.inputText = input + const output = str(attributes, "ai.response.text") ?? str(attributes, "ai.toolCall.result") + if (output !== null && attributes["gen_ai.output.messages"] === undefined) + facts.outputText = output + return facts +} + +// eve — claims exactly one span shape: the `ai.eve.turn` turn root (scope +// `eve`). The bare `eve.*` keys exist only there; the turn's model/tool spans +// are vercel_ai_sdk's. A turn root is the agent tier and carries no tokens. +const eveIntegration: AiVendorIntegration = ({ attributes }) => ({ + role: "agent", + sessionKey: str(attributes, "eve.session.id"), +}) + +// Only vendors whose wire format diverges from semconv in a way we have +// verified against captured data get an entry. Absence means "the base is +// right", not "unsupported". +const INTEGRATIONS: Readonly> = { + mastra: mastraIntegration, + claude_agent_sdk: claudeAgentSdkIntegration, + vercel_ai_sdk: vercelAiSdkIntegration, + eve: eveIntegration, +} + +/** Base semconv facts merged with the vendor's overrides — the one entry point. + * A field the integration returns as `undefined` keeps the base's answer. */ +export const normalizeAiSpan = (span: AiSpanInput): AiSpanFacts => { + const facts = baseNormalize(span) + const integration = INTEGRATIONS[span.vendor] + if (integration === undefined) return facts + const overrides = integration(span) + const merged: Record = { ...facts } + for (const [key, value] of Object.entries(overrides)) { + if (value !== undefined) merged[key] = value + } + return merged as unknown as AiSpanFacts +} diff --git a/packages/domain/src/http/agent-sessions.ts b/packages/domain/src/http/agent-sessions.ts new file mode 100644 index 000000000..dfa381a8b --- /dev/null +++ b/packages/domain/src/http/agent-sessions.ts @@ -0,0 +1,237 @@ +import { HttpApiEndpoint, HttpApiGroup } from "effect/unstable/httpapi" +import { Schema } from "effect" +import { TraceId } from "../primitives" +import { TinybirdDateTime } from "../query-engine" +import { SessionAuthorization } from "./current-tenant" +import { QueryEngineExecutionError, QueryEngineTimeoutError } from "./query-engine" +import { warehouseHttpErrors } from "./warehouse" + +// Agent Sessions endpoint schemas +// +// The dashboard read path over the AI classification columns on `traces` +// (AiVendor / AiSessionKeyState / AiSessionKeyHash). Internal tier on purpose: +// the feature is a product scratchpad and these shapes follow the UI. The +// durable read interface is `@maple/query-engine/observability` — the handlers +// here are thin adapters over it, and the future MCP tools call the same +// functions rather than these routes. + +// One filter payload for both tabs and the facets, so the sidebar counts can't +// mean something different from the rows. `vendors`/`serviceNames` are +// containment filters ("has at least one matching AI span") — classification +// is per-span and multi-vendor sessions/traces are the norm. All optional +// filters are JS-constructed by the web client → `Schema.optional`, not +// `optionalKey` (see CLAUDE.md). +const agentSessionsFilterFields = { + startTime: TinybirdDateTime, + endTime: TinybirdDateTime, + /** Vendor slugs (`AI_VENDORS` in `@maple/domain`, including `unknown:*`). */ + vendors: Schema.optional(Schema.Array(Schema.String)), + serviceNames: Schema.optional(Schema.Array(Schema.String)), + hasErrors: Schema.optional(Schema.Boolean), +} as const + +export class AgentSessionsListRequest extends Schema.Class( + "AgentSessionsListRequest", +)({ + ...agentSessionsFilterFields, + limit: Schema.optional(Schema.Number), + offset: Schema.optional(Schema.Number), +}) {} + +export const AgentSessionListItem = Schema.Struct({ + /** Opaque session id: `toString(cityHash64(key))`. The plaintext key lives in + * span attributes under a vendor-specific key and is resolved by the detail + * read, not the list. */ + sessionKeyHash: Schema.String, + startTime: Schema.String, + endTime: Schema.String, + durationMs: Schema.Number, + traceCount: Schema.Number, + /** Only the session-authoritative key-carrying spans — a session's traces + * hold more spans than this. Same population for `errorCount`. */ + keyedSpanCount: Schema.Number, + errorCount: Schema.Number, + vendors: Schema.Array(Schema.String), + serviceNames: Schema.Array(Schema.String), +}) + +export class AgentSessionsListResponse extends Schema.Class( + "AgentSessionsListResponse", +)({ + data: Schema.Array(AgentSessionListItem), +}) {} + +export class AgentTracesListRequest extends Schema.Class( + "AgentTracesListRequest", +)({ + ...agentSessionsFilterFields, + limit: Schema.optional(Schema.Number), + offset: Schema.optional(Schema.Number), +}) {} + +export const AgentTraceListItem = Schema.Struct({ + traceId: TraceId, + startTime: Schema.String, + endTime: Schema.String, + /** Window of the trace's AI spans only — the full trace can be wider. */ + durationMs: Schema.Number, + aiSpanCount: Schema.Number, + errorCount: Schema.Number, + vendors: Schema.Array(Schema.String), + serviceNames: Schema.Array(Schema.String), + firstSpanName: Schema.String, + /** max(AiSessionKeyState): 6 = belongs to a session, lower explains why not + * (write-side enum in `ai_classifier.rs`). */ + bestSessionKeyState: Schema.Number, + /** Session key hash as a string, `''` when the trace carries none. */ + sessionKeyHash: Schema.String, +}) + +export class AgentTracesListResponse extends Schema.Class( + "AgentTracesListResponse", +)({ + data: Schema.Array(AgentTraceListItem), +}) {} + +export class AgentSessionsFacetsRequest extends Schema.Class( + "AgentSessionsFacetsRequest", +)({ + ...agentSessionsFilterFields, + /** Counting unit for every facet — match the open tab. */ + tab: Schema.Literals(["sessions", "traces"]), +}) {} + +export const AgentFacetItem = Schema.Struct({ + name: Schema.String, + count: Schema.Number, +}) + +export class AgentSessionsFacetsResponse extends Schema.Class( + "AgentSessionsFacetsResponse", +)({ + vendors: Schema.Array(AgentFacetItem), + services: Schema.Array(AgentFacetItem), + /** Sessions/traces (per `tab`) with at least one errored AI span, within the + * current filter. */ + errorCount: Schema.Number, +}) {} + +export class AgentSessionDetailRequest extends Schema.Class( + "AgentSessionDetailRequest", +)({ + /** Opaque session id from the list rows: `toString(AiSessionKeyHash)`. */ + sessionKeyHash: Schema.String, + startTime: TinybirdDateTime, + endTime: TinybirdDateTime, +}) {} + +/** One AI span with the vendor-integration facts merged in. The raw attribute + * maps stay out of this shape — conversational content travels only as the + * integration layer's extracted (and truncated) `inputText`/`outputText`. */ +export const NormalizedAiSpanItem = Schema.Struct({ + spanId: Schema.String, + parentSpanId: Schema.String, + startTime: Schema.String, + durationMs: Schema.Number, + spanName: Schema.String, + spanKind: Schema.String, + serviceName: Schema.String, + statusCode: Schema.String, + statusMessage: Schema.String, + vendor: Schema.String, + sessionKeyState: Schema.Number, + role: Schema.Literals(["llm", "tool", "agent", "workflow", "other"]), + model: Schema.NullOr(Schema.String), + inputTokens: Schema.NullOr(Schema.Number), + outputTokens: Schema.NullOr(Schema.Number), + cacheReadTokens: Schema.NullOr(Schema.Number), + cacheCreationTokens: Schema.NullOr(Schema.Number), + costUsd: Schema.NullOr(Schema.Number), + sessionKey: Schema.NullOr(Schema.String), + /** Vendor-recorded conversational content (messages JSON / prompt / tool + * args and result), truncated server-side. Raw strings — format varies per + * vendor and rendering is the client's job. */ + inputText: Schema.NullOr(Schema.String), + outputText: Schema.NullOr(Schema.String), +}) + +export const AgentSessionTraceItem = Schema.Struct({ + traceId: TraceId, + startTime: Schema.String, + durationMs: Schema.Number, + errorCount: Schema.Number, + spans: Schema.Array(NormalizedAiSpanItem), +}) + +export const AgentSessionDetail = Schema.Struct({ + sessionKeyHash: Schema.String, + /** Plaintext session key when a vendor integration resolved one; the UI + * falls back to the hash. */ + sessionKey: Schema.NullOr(Schema.String), + startTime: Schema.String, + endTime: Schema.String, + durationMs: Schema.Number, + /** The session blew a fetch cap and every count below undercounts. */ + truncated: Schema.Boolean, + totals: Schema.Struct({ + spanCount: Schema.Number, + llmCallCount: Schema.Number, + toolCallCount: Schema.Number, + errorCount: Schema.Number, + inputTokens: Schema.Number, + outputTokens: Schema.Number, + cacheReadTokens: Schema.Number, + cacheCreationTokens: Schema.Number, + /** `null` when no span priced itself — unknown, never free. */ + costUsd: Schema.NullOr(Schema.Number), + }), + vendors: Schema.Array(Schema.String), + serviceNames: Schema.Array(Schema.String), + models: Schema.Array(Schema.String), + traces: Schema.Array(AgentSessionTraceItem), +}) + +export class AgentSessionDetailResponse extends Schema.Class( + "AgentSessionDetailResponse", +)({ + /** `null`: the hash matched nothing in the window (expired, or a foreign id). */ + session: Schema.NullOr(AgentSessionDetail), +}) {} + +const agentSessionsEndpointErrors = [ + QueryEngineExecutionError, + QueryEngineTimeoutError, + ...warehouseHttpErrors, +] as const + +export class AgentSessionsInternalApiGroup extends HttpApiGroup.make("agentSessionsInternal") + .add( + HttpApiEndpoint.post("list", "/list", { + payload: AgentSessionsListRequest, + success: AgentSessionsListResponse, + error: agentSessionsEndpointErrors, + }), + ) + .add( + HttpApiEndpoint.post("traces", "/traces", { + payload: AgentTracesListRequest, + success: AgentTracesListResponse, + error: agentSessionsEndpointErrors, + }), + ) + .add( + HttpApiEndpoint.post("detail", "/detail", { + payload: AgentSessionDetailRequest, + success: AgentSessionDetailResponse, + error: agentSessionsEndpointErrors, + }), + ) + .add( + HttpApiEndpoint.post("facets", "/facets", { + payload: AgentSessionsFacetsRequest, + success: AgentSessionsFacetsResponse, + error: agentSessionsEndpointErrors, + }), + ) + .prefix("/internal/agent-sessions") + .middleware(SessionAuthorization) {} diff --git a/packages/domain/src/http/index.ts b/packages/domain/src/http/index.ts index 3d65b46dc..20c031b79 100644 --- a/packages/domain/src/http/index.ts +++ b/packages/domain/src/http/index.ts @@ -26,6 +26,7 @@ export * from "./query-engine" export * from "./recommendation-issues" export * from "./scrape-targets" export * from "./scraper-internal" +export * from "./agent-sessions" export * from "./session-replay" export * from "./share" export * from "./slack-internal" diff --git a/packages/domain/src/http/internal-api.ts b/packages/domain/src/http/internal-api.ts index 8e7064bd6..433f7e78b 100644 --- a/packages/domain/src/http/internal-api.ts +++ b/packages/domain/src/http/internal-api.ts @@ -1,4 +1,5 @@ import { HttpApi, OpenApi } from "effect/unstable/httpapi" +import { AgentSessionsInternalApiGroup } from "./agent-sessions" import { AiTriageApiGroup } from "./ai-triage" import { BillingApiGroup } from "./billing" import { ChatApiGroup } from "./chat" @@ -35,6 +36,7 @@ import { V1SchemaErrors, V1UnexpectedErrors } from "./v1-boundary" * split costs the frontend nothing. */ export class MapleInternalApi extends HttpApi.make("MapleInternalApi") + .add(AgentSessionsInternalApiGroup) .add(AiTriageApiGroup) .add(BillingApiGroup) .add(ChatApiGroup) diff --git a/packages/query-engine/src/__sql_baseline__/catalog.sql b/packages/query-engine/src/__sql_baseline__/catalog.sql index 1390307a8..244f39cb6 100644 --- a/packages/query-engine/src/__sql_baseline__/catalog.sql +++ b/packages/query-engine/src/__sql_baseline__/catalog.sql @@ -22,6 +22,266 @@ SELECT GROUP BY orgId FORMAT JSON +-- builder:agent-sessions:agentSessionsFacetsQuery:sessions-tab [26d2d920] +SELECT + arrayJoin(vendors) AS name, + count() AS count, + 'vendor' AS facetType + FROM (SELECT + toString(AiSessionKeyHash) AS groupKey, + groupUniqArray(AiVendor) AS vendors, + groupUniqArray(ServiceName) AS serviceNames, + countIf(StatusCode = 'Error') AS errorCount + FROM traces + WHERE OrgId = 'org_sql_catalog' + AND Timestamp >= '2026-01-01 10:30:00' + AND Timestamp <= '2026-01-03 14:15:00' + AND AiSessionKeyState = 6 + GROUP BY groupKey) AS g + GROUP BY name + HAVING name != '' + ORDER BY count DESC + LIMIT 50 +UNION ALL +SELECT + arrayJoin(serviceNames) AS name, + count() AS count, + 'service' AS facetType + FROM (SELECT + toString(AiSessionKeyHash) AS groupKey, + groupUniqArray(AiVendor) AS vendors, + groupUniqArray(ServiceName) AS serviceNames, + countIf(StatusCode = 'Error') AS errorCount + FROM traces + WHERE OrgId = 'org_sql_catalog' + AND Timestamp >= '2026-01-01 10:30:00' + AND Timestamp <= '2026-01-03 14:15:00' + AND AiSessionKeyState = 6 + GROUP BY groupKey) AS g + GROUP BY name + HAVING name != '' + ORDER BY count DESC + LIMIT 50 +UNION ALL +SELECT + 'error' AS name, + count() AS count, + 'error' AS facetType + FROM (SELECT + toString(AiSessionKeyHash) AS groupKey, + groupUniqArray(AiVendor) AS vendors, + groupUniqArray(ServiceName) AS serviceNames, + countIf(StatusCode = 'Error') AS errorCount + FROM traces + WHERE OrgId = 'org_sql_catalog' + AND Timestamp >= '2026-01-01 10:30:00' + AND Timestamp <= '2026-01-03 14:15:00' + AND AiSessionKeyState = 6 + GROUP BY groupKey) AS g + WHERE errorCount > 0 +FORMAT JSON + +-- builder:agent-sessions:agentSessionsFacetsQuery:traces-tab-filtered [671db227] +SELECT + arrayJoin(vendors) AS name, + count() AS count, + 'vendor' AS facetType + FROM (SELECT + TraceId AS groupKey, + groupUniqArray(AiVendor) AS vendors, + groupUniqArray(ServiceName) AS serviceNames, + countIf(StatusCode = 'Error') AS errorCount + FROM traces + WHERE OrgId = 'org_sql_catalog' + AND Timestamp >= '2026-01-01 10:30:00' + AND Timestamp <= '2026-01-03 14:15:00' + AND AiVendor != '' + GROUP BY groupKey) AS g + WHERE errorCount > 0 + GROUP BY name + HAVING name != '' + ORDER BY count DESC + LIMIT 50 +UNION ALL +SELECT + arrayJoin(serviceNames) AS name, + count() AS count, + 'service' AS facetType + FROM (SELECT + TraceId AS groupKey, + groupUniqArray(AiVendor) AS vendors, + groupUniqArray(ServiceName) AS serviceNames, + countIf(StatusCode = 'Error') AS errorCount + FROM traces + WHERE OrgId = 'org_sql_catalog' + AND Timestamp >= '2026-01-01 10:30:00' + AND Timestamp <= '2026-01-03 14:15:00' + AND AiVendor != '' + GROUP BY groupKey) AS g + WHERE has(vendors, 'crewai') + AND errorCount > 0 + GROUP BY name + HAVING name != '' + ORDER BY count DESC + LIMIT 50 +UNION ALL +SELECT + 'error' AS name, + count() AS count, + 'error' AS facetType + FROM (SELECT + TraceId AS groupKey, + groupUniqArray(AiVendor) AS vendors, + groupUniqArray(ServiceName) AS serviceNames, + countIf(StatusCode = 'Error') AS errorCount + FROM traces + WHERE OrgId = 'org_sql_catalog' + AND Timestamp >= '2026-01-01 10:30:00' + AND Timestamp <= '2026-01-03 14:15:00' + AND AiVendor != '' + GROUP BY groupKey) AS g + WHERE has(vendors, 'crewai') + AND errorCount > 0 +FORMAT JSON + +-- builder:agent-sessions:agentSessionsListQuery:default [f361c39f] +SELECT + toString(AiSessionKeyHash) AS sessionKeyHash, + min(Timestamp) AS startTime, + max(Timestamp) AS endTime, + max(toUnixTimestamp64Nano(Timestamp) + toInt64(Duration)) / 1000000 - min(toUnixTimestamp64Nano(Timestamp)) / 1000000 AS durationMs, + uniq(TraceId) AS traceCount, + count() AS keyedSpanCount, + countIf(StatusCode = 'Error') AS errorCount, + groupUniqArray(AiVendor) AS vendors, + groupUniqArray(ServiceName) AS serviceNames + FROM traces + WHERE OrgId = 'org_sql_catalog' + AND Timestamp >= '2026-01-01 10:30:00' + AND Timestamp <= '2026-01-03 14:15:00' + AND AiSessionKeyState = 6 + GROUP BY sessionKeyHash + ORDER BY endTime DESC + LIMIT 50 + OFFSET 0 + FORMAT JSON + +-- builder:agent-sessions:agentSessionsListQuery:filtered [1afdfb41] +SELECT + toString(AiSessionKeyHash) AS sessionKeyHash, + min(Timestamp) AS startTime, + max(Timestamp) AS endTime, + max(toUnixTimestamp64Nano(Timestamp) + toInt64(Duration)) / 1000000 - min(toUnixTimestamp64Nano(Timestamp)) / 1000000 AS durationMs, + uniq(TraceId) AS traceCount, + count() AS keyedSpanCount, + countIf(StatusCode = 'Error') AS errorCount, + groupUniqArray(AiVendor) AS vendors, + groupUniqArray(ServiceName) AS serviceNames + FROM traces + WHERE OrgId = 'org_sql_catalog' + AND Timestamp >= '2026-01-01 10:30:00' + AND Timestamp <= '2026-01-03 14:15:00' + AND AiSessionKeyState = 6 + GROUP BY sessionKeyHash + HAVING (has(groupUniqArray(AiVendor), 'crewai') OR has(groupUniqArray(AiVendor), 'vercel_ai_sdk')) + AND has(groupUniqArray(ServiceName), 'checkout') + AND countIf(StatusCode = 'Error') > 0 + ORDER BY endTime DESC + LIMIT 50 + OFFSET 0 + FORMAT JSON + +-- builder:agent-sessions:agentSessionSpansQuery:default [3dba1d73] +SELECT + TraceId AS traceId, + SpanId AS spanId, + ParentSpanId AS parentSpanId, + Timestamp AS timestamp, + Duration / 1000000 AS durationMs, + SpanName AS spanName, + SpanKind AS spanKind, + ServiceName AS serviceName, + StatusCode AS statusCode, + StatusMessage AS statusMessage, + AiVendor AS vendor, + AiSessionKeyState AS sessionKeyState, + SpanAttributes AS spanAttributes + FROM traces + WHERE OrgId = 'org_sql_catalog' + AND Timestamp >= '2026-01-01 10:30:00' + AND Timestamp <= '2026-01-03 14:15:00' + AND TraceId IN ('0af7651916cd43dd8448eb211c80319c', '4bf92f3577b34da6a3ce929d0e0e4736') + AND AiVendor != '' + ORDER BY timestamp ASC, spanId ASC + LIMIT 2000 + FORMAT JSON + +-- builder:agent-sessions:agentSessionTraceIdsQuery:default [6f04e848] +SELECT + TraceId AS traceId, + min(Timestamp) AS startTime, + max(Timestamp) AS endTime + FROM traces + WHERE OrgId = 'org_sql_catalog' + AND Timestamp >= '2026-01-01 10:30:00' + AND Timestamp <= '2026-01-03 14:15:00' + AND AiSessionKeyState = 6 + AND toString(AiSessionKeyHash) = '13464164225153980885' + GROUP BY traceId + ORDER BY startTime ASC + LIMIT 200 + FORMAT JSON + +-- builder:agent-sessions:agentTracesListQuery:default [dc798139] +SELECT + TraceId AS traceId, + min(Timestamp) AS startTime, + max(Timestamp) AS endTime, + max(toUnixTimestamp64Nano(Timestamp) + toInt64(Duration)) / 1000000 - min(toUnixTimestamp64Nano(Timestamp)) / 1000000 AS durationMs, + count() AS aiSpanCount, + countIf(StatusCode = 'Error') AS errorCount, + groupUniqArray(AiVendor) AS vendors, + groupUniqArray(ServiceName) AS serviceNames, + argMin(SpanName, Timestamp) AS firstSpanName, + max(AiSessionKeyState) AS bestSessionKeyState, + if(maxIf(AiSessionKeyHash, AiSessionKeyState = 6) > 0, toString(maxIf(AiSessionKeyHash, AiSessionKeyState = 6)), '') AS sessionKeyHash + FROM traces + WHERE OrgId = 'org_sql_catalog' + AND Timestamp >= '2026-01-01 10:30:00' + AND Timestamp <= '2026-01-03 14:15:00' + AND AiVendor != '' + GROUP BY traceId + ORDER BY startTime DESC + LIMIT 50 + OFFSET 0 + FORMAT JSON + +-- builder:agent-sessions:agentTracesListQuery:filtered [211143ca] +SELECT + TraceId AS traceId, + min(Timestamp) AS startTime, + max(Timestamp) AS endTime, + max(toUnixTimestamp64Nano(Timestamp) + toInt64(Duration)) / 1000000 - min(toUnixTimestamp64Nano(Timestamp)) / 1000000 AS durationMs, + count() AS aiSpanCount, + countIf(StatusCode = 'Error') AS errorCount, + groupUniqArray(AiVendor) AS vendors, + groupUniqArray(ServiceName) AS serviceNames, + argMin(SpanName, Timestamp) AS firstSpanName, + max(AiSessionKeyState) AS bestSessionKeyState, + if(maxIf(AiSessionKeyHash, AiSessionKeyState = 6) > 0, toString(maxIf(AiSessionKeyHash, AiSessionKeyState = 6)), '') AS sessionKeyHash + FROM traces + WHERE OrgId = 'org_sql_catalog' + AND Timestamp >= '2026-01-01 10:30:00' + AND Timestamp <= '2026-01-03 14:15:00' + AND AiVendor != '' + GROUP BY traceId + HAVING has(groupUniqArray(AiVendor), 'claude_agent_sdk') + AND countIf(StatusCode = 'Error') > 0 + ORDER BY startTime DESC + LIMIT 50 + OFFSET 0 + FORMAT JSON + -- builder:errors:errorFingerprintsQuery:envFiltered [f1269c9f] SELECT toString(FingerprintHash) AS fingerprintHash diff --git a/packages/query-engine/src/ch/builder-fixtures.ts b/packages/query-engine/src/ch/builder-fixtures.ts index 68cf4455f..1aa5bca30 100644 --- a/packages/query-engine/src/ch/builder-fixtures.ts +++ b/packages/query-engine/src/ch/builder-fixtures.ts @@ -632,6 +632,83 @@ export const builderFixtures: ReadonlyArray = [ ), }, + // ----- agent-sessions: the AI classification read path (no app call sites + // ----- yet — the observability/HTTP layer lands next; parameters mirror its + // ----- planned filter payload). Filtered variants force the HAVING + // ----- containment branch, and each facet tab is its own grouped-subquery + // ----- shape, so both compile here. + { + module: "agent-sessions", + name: "agentSessionsListQuery", + label: "default", + compile: () => CH.compile(CH.agentSessionsListQuery({}), window), + }, + { + module: "agent-sessions", + name: "agentSessionsListQuery", + label: "filtered", + compile: () => + CH.compile( + CH.agentSessionsListQuery({ + vendors: ["crewai", "vercel_ai_sdk"], + serviceNames: ["checkout"], + hasErrors: true, + }), + window, + ), + }, + { + module: "agent-sessions", + name: "agentTracesListQuery", + label: "default", + compile: () => CH.compile(CH.agentTracesListQuery({}), window), + }, + { + module: "agent-sessions", + name: "agentTracesListQuery", + label: "filtered", + compile: () => + CH.compile(CH.agentTracesListQuery({ vendors: ["claude_agent_sdk"], hasErrors: true }), window), + }, + { + module: "agent-sessions", + name: "agentSessionsFacetsQuery", + label: "sessions-tab", + compile: () => CH.compileUnion(CH.agentSessionsFacetsQuery({ tab: "sessions" }), window), + }, + { + module: "agent-sessions", + name: "agentSessionsFacetsQuery", + label: "traces-tab-filtered", + compile: () => + CH.compileUnion( + CH.agentSessionsFacetsQuery({ tab: "traces", vendors: ["crewai"], hasErrors: true }), + window, + ), + }, + { + module: "agent-sessions", + name: "agentSessionTraceIdsQuery", + label: "default", + compile: () => + CH.compile(CH.agentSessionTraceIdsQuery(), { + ...window, + sessionKeyHash: "13464164225153980885", + }), + }, + { + module: "agent-sessions", + name: "agentSessionSpansQuery", + label: "default", + compile: () => + CH.compile( + CH.agentSessionSpansQuery({ + traceIds: [TRACE_ID, "4bf92f3577b34da6a3ce929d0e0e4736"], + }), + window, + ), + }, + // ----- activity: the only deliberately cross-org builders in the product. // ----- Fixtured so the catalog's tenant-scope test actually exercises the // ----- cross-org branch, rather than asserting a rule nothing exemplifies. diff --git a/packages/query-engine/src/ch/index.ts b/packages/query-engine/src/ch/index.ts index 8d74dde75..ea0cc57bb 100644 --- a/packages/query-engine/src/ch/index.ts +++ b/packages/query-engine/src/ch/index.ts @@ -151,6 +151,27 @@ export { type SessionActivityOutput, } from "./queries/session-events" +// Queries — Agent Sessions (AI-classified spans: sessions + raw AI traces) +export { + agentSessionsListQuery, + agentTracesListQuery, + agentSessionsFacetsQuery, + agentSessionTraceIdsQuery, + agentSessionSpansQuery, + AGENT_SESSION_MAX_TRACES, + AGENT_SESSION_MAX_SPANS, + type AgentSessionsFilterOpts, + type AgentSessionsListOpts, + type AgentSessionsListOutput, + type AgentTracesListOpts, + type AgentTracesListOutput, + type AgentSessionsFacetsOpts, + type AgentSessionsFacetsOutput, + type AgentSessionTraceIdsOutput, + type AgentSessionSpansOpts, + type AgentSessionSpansOutput, +} from "./queries/agent-sessions" + // Queries — Web Analytics (product analytics over the browser SDK's session data) export { webAnalyticsSummaryQuery, diff --git a/packages/query-engine/src/ch/queries/agent-sessions.test.ts b/packages/query-engine/src/ch/queries/agent-sessions.test.ts new file mode 100644 index 000000000..f8c3517e2 --- /dev/null +++ b/packages/query-engine/src/ch/queries/agent-sessions.test.ts @@ -0,0 +1,178 @@ +import { describe, expect, it } from "@effect/vitest" +import { compileCH, compileUnion } from "@maple-dev/clickhouse-builder" +import { + AGENT_SESSION_MAX_SPANS, + AGENT_SESSION_MAX_TRACES, + agentSessionSpansQuery, + agentSessionTraceIdsQuery, + agentSessionsFacetsQuery, + agentSessionsListQuery, + agentTracesListQuery, +} from "./agent-sessions" + +const WINDOW = { orgId: "org_1", startTime: "2026-06-24 04:00:00", endTime: "2026-06-25 06:00:00" } + +describe("agentSessionsListQuery", () => { + it("groups session-granularity spans by the key hash, org-scoped", () => { + const { sql, tenantScope } = compileCH(agentSessionsListQuery({}), WINDOW) + expect(tenantScope).toBe("org") + expect(sql).toContain("FROM traces") + expect(sql).toContain("AiSessionKeyState = 6") + expect(sql).toContain("GROUP BY sessionKeyHash") + expect(sql).toContain("FORMAT JSON") + }) + + // UInt64 above 2^53 corrupts as a JS number — the hash must ride the wire + // as a string (the sql-catalog identity sweep enforces the same rule). + it("toString-wraps the UInt64 key hash", () => { + const { sql } = compileCH(agentSessionsListQuery({}), WINDOW) + expect(sql).toContain("toString(AiSessionKeyHash) AS sessionKeyHash") + }) + + // The DSL's infix operators don't parenthesize, so a - b/n is one edit away. + // Each aggregate must divide to ms BEFORE the subtraction, and Duration must + // cast to Int64 (no UInt64/Int64 supertype to add the nano timestamp to). + it("computes the span window with per-term division and an Int64 duration", () => { + const { sql } = compileCH(agentSessionsListQuery({}), WINDOW) + expect(sql).toContain( + "max(toUnixTimestamp64Nano(Timestamp) + toInt64(Duration)) / 1000000 - min(toUnixTimestamp64Nano(Timestamp)) / 1000000 AS durationMs", + ) + }) + + // Vendor/service filters mean containment ("has at least one matching + // span"). In WHERE they would silently drop the other vendors' spans from + // the row's own vendors/counts — the multi-vendor case (CrewAI wrapping + // openinference-openai) is the norm, not the edge. + it("applies vendor containment in HAVING, not WHERE", () => { + const { sql } = compileCH( + agentSessionsListQuery({ vendors: ["crewai", "vercel_ai_sdk"] }), + WINDOW, + ) + const whereClause = sql.slice(sql.indexOf("WHERE"), sql.indexOf("GROUP BY")) + expect(whereClause).not.toContain("AiVendor") + expect(sql).toContain( + "HAVING (has(groupUniqArray(AiVendor), 'crewai') OR has(groupUniqArray(AiVendor), 'vercel_ai_sdk'))", + ) + }) + + it("filters errored sessions post-aggregation", () => { + const { sql } = compileCH(agentSessionsListQuery({ hasErrors: true }), WINDOW) + expect(sql).toContain("HAVING countIf(StatusCode = 'Error') > 0") + }) +}) + +describe("agentTracesListQuery", () => { + it("groups every AI-classified span by trace at any key state", () => { + const { sql, tenantScope } = compileCH(agentTracesListQuery({}), WINDOW) + expect(tenantScope).toBe("org") + expect(sql).toContain("GROUP BY traceId") + const whereClause = sql.slice(sql.indexOf("WHERE"), sql.indexOf("GROUP BY")) + expect(whereClause).toContain("AiVendor != ''") + // No key-state gate: unkeyed AI spans are exactly what this tab surfaces. + expect(whereClause).not.toContain("AiSessionKeyState") + }) + + it("surfaces the session linkage: best state and ''-when-absent key hash", () => { + const { sql } = compileCH(agentTracesListQuery({}), WINDOW) + expect(sql).toContain("max(AiSessionKeyState) AS bestSessionKeyState") + expect(sql).toContain( + "if(maxIf(AiSessionKeyHash, AiSessionKeyState = 6) > 0, toString(maxIf(AiSessionKeyHash, AiSessionKeyState = 6)), '') AS sessionKeyHash", + ) + }) + + it("applies service containment in HAVING like the sessions list", () => { + const { sql } = compileCH(agentTracesListQuery({ serviceNames: ["checkout"] }), WINDOW) + expect(sql).toContain("HAVING has(groupUniqArray(ServiceName), 'checkout')") + }) +}) + +describe("agentSessionsFacetsQuery", () => { + it("counts sessions on the sessions tab and traces on the traces tab", () => { + const sessions = compileUnion(agentSessionsFacetsQuery({ tab: "sessions" }), WINDOW) + expect(sessions.sql).toContain("toString(AiSessionKeyHash) AS groupKey") + expect(sessions.sql).toContain("AiSessionKeyState = 6") + const traces = compileUnion(agentSessionsFacetsQuery({ tab: "traces" }), WINDOW) + expect(traces.sql).toContain("TraceId AS groupKey") + expect(traces.sql).toContain("AiVendor != ''") + }) + + // Tenant scope must survive the fromQuery wrapping — the branches carry no + // OrgId predicate of their own; it lives in the grouped subquery. + it("stays org-scoped through the grouped subquery", () => { + const { tenantScope } = compileUnion(agentSessionsFacetsQuery({ tab: "sessions" }), WINDOW) + expect(tenantScope).toBe("org") + }) + + // A grouped row expands via arrayJoin, so a mixed-vendor session counts + // once under each vendor it contains — matching the list's containment + // filters by construction. + it("expands the aggregated arrays with arrayJoin per branch", () => { + const { sql } = compileUnion(agentSessionsFacetsQuery({ tab: "sessions" }), WINDOW) + expect(sql).toContain("arrayJoin(vendors) AS name") + expect(sql).toContain("arrayJoin(serviceNames) AS name") + }) + + it("excludes each dimension's own filter from its branch", () => { + const { sql } = compileUnion( + agentSessionsFacetsQuery({ tab: "sessions", vendors: ["crewai"], hasErrors: true }), + WINDOW, + ) + const [vendorBranch, serviceBranch, errorBranch] = sql.split("UNION ALL") + expect(vendorBranch).not.toContain("has(vendors,") + expect(vendorBranch).toContain("errorCount > 0") + expect(serviceBranch).toContain("has(vendors, 'crewai')") + expect(serviceBranch).toContain("errorCount > 0") + // The error branch keeps the vendor filter but not its own toggle — the + // count answers "how many WOULD match if you enabled it". + expect(errorBranch).toContain("has(vendors, 'crewai')") + expect(errorBranch).toContain("errorCount > 0") + }) + + it("drops empty facet values after the arrayJoin", () => { + const { sql } = compileUnion(agentSessionsFacetsQuery({ tab: "sessions" }), WINDOW) + expect(sql).toContain("HAVING name != ''") + }) +}) + +describe("agentSessionTraceIdsQuery", () => { + const window = { ...WINDOW, sessionKeyHash: "13464164225153980885" } + + it("resolves the hash to TraceIds over session-granularity rows only", () => { + const { sql, tenantScope } = compileCH(agentSessionTraceIdsQuery(), window) + expect(tenantScope).toBe("org") + expect(sql).toContain("AiSessionKeyState = 6") + expect(sql).toContain("GROUP BY traceId") + expect(sql).toContain(`LIMIT ${AGENT_SESSION_MAX_TRACES}`) + }) + + // The hash is a UInt64 identity: it arrives as a string and must never round + // through a JS number, so the comparison happens on the string side. + it("compares the key hash as a string", () => { + const { sql } = compileCH(agentSessionTraceIdsQuery(), window) + expect(sql).toContain("toString(AiSessionKeyHash) = '13464164225153980885'") + }) +}) + +describe("agentSessionSpansQuery", () => { + const traceIds = ["0af7651916cd43dd8448eb211c80319c", "4bf92f3577b34da6a3ce929d0e0e4736"] + + it("fetches every AI span of the session's traces, org-scoped", () => { + const { sql, tenantScope } = compileCH(agentSessionSpansQuery({ traceIds }), WINDOW) + expect(tenantScope).toBe("org") + expect(sql).toContain("AiVendor != ''") + expect(sql).toContain( + "TraceId IN ('0af7651916cd43dd8448eb211c80319c', '4bf92f3577b34da6a3ce929d0e0e4736')", + ) + expect(sql).toContain(`LIMIT ${AGENT_SESSION_MAX_SPANS}`) + }) + + // The integration layer dispatches on attribute spellings per vendor, so the + // whole map travels — a projected key list here would couple this query to + // every integration. + it("selects the full SpanAttributes map and the classification columns", () => { + const { sql } = compileCH(agentSessionSpansQuery({ traceIds }), WINDOW) + expect(sql).toContain("SpanAttributes AS spanAttributes") + expect(sql).toContain("AiVendor AS vendor") + expect(sql).toContain("AiSessionKeyState AS sessionKeyState") + }) +}) diff --git a/packages/query-engine/src/ch/queries/agent-sessions.ts b/packages/query-engine/src/ch/queries/agent-sessions.ts new file mode 100644 index 000000000..150b0df91 --- /dev/null +++ b/packages/query-engine/src/ch/queries/agent-sessions.ts @@ -0,0 +1,384 @@ +// Typed Agent Sessions Queries +// +// Read path over the AI classification columns the ingest classifier stamps on +// `traces` (AiVendor / AiSessionKeyState / AiSessionKeyHash — migration 0016). +// One filter payload feeds two tabs: +// +// - sessions: rows whose key resolved at session granularity +// (`AiSessionKeyState = 6`), grouped by `AiSessionKeyHash`. +// - traces: every AI-classified row (`AiVendor != ''`) at any key state, +// grouped by `TraceId`. +// +// Classification is strictly per span, and a single trace routinely mixes +// vendors (a CrewAI orchestration span parenting openinference-instrumented +// OpenAI calls), so `vendors` is an array on every output row and the +// vendor/service filters mean *containment*: "has at least one matching span". +// Containment is a post-aggregation predicate, so those filters live in HAVING +// over the same aggregates the row reports — never in WHERE, which would drop +// the non-matching spans from the row's own counts. +// +// The facet branches read one grouped subquery and `arrayJoin` its arrays, so +// facet counts and list rows share a single grouping by construction (the +// traces list/facets pair maintains that invariant across two hand-written +// extractors, which is how its exclusion filters drifted). +// +// The session key itself is not selectable here: only the hash is a column, +// and the plaintext lives in `SpanAttributes` under a vendor-specific key. +// Resolving it for display is the vendor-integration layer's job on the +// detail read. + +import * as CH from "@maple-dev/clickhouse-builder/expr" +import { param } from "@maple-dev/clickhouse-builder" +import { from, fromQuery, type ColumnAccessor } from "@maple-dev/clickhouse-builder" +import { unionAll, type CHUnionQuery } from "@maple-dev/clickhouse-builder" +import { Traces } from "../tables" +import type { FacetOutput } from "./query-helpers" + +/** `AiSessionKeyState` value for "key resolved at session granularity" — the + * only state whose hash identifies a customer-facing session. Frozen on the + * write side (`session_state::SESSION` in `ai_classifier.rs`); the rollup MV + * persists comparisons over it, so it can never renumber. */ +const SESSION_GRANULARITY = 6 + +/** "Contains at least one span matching ANY of `values`" over an aggregated + * array column. Returns undefined (condition dropped) when no filter is set. */ +const containsAny = ( + arr: CH.Expr>, + values: readonly string[] | undefined, +): CH.Condition | undefined => + values && values.length > 0 + ? values.map((v) => CH.has(arr, v)).reduce((a, b) => a.or(b)) + : undefined + +/** Wall-clock ms from the first span start to the last span END — `Duration` + * is nanoseconds, so the last span's own runtime counts. A plain max(start) − + * min(start) would report 0 for the common single-span-per-key case. + * + * Each aggregate divides before the subtraction (the DSL's infix operators + * don't parenthesize, so `a.sub(b).div(n)` binds as `a - b/n`), and Duration + * casts to Int64 because ClickHouse has no UInt64/Int64 supertype to add the + * Int64 nanosecond timestamp to. */ +const spanWindowMs = ($: ColumnAccessor): CH.Expr => + CH.max_(CH.toUnixTimestamp64Nano($.Timestamp).add(CH.toInt64($.Duration))) + .div(1_000_000) + .sub(CH.min_(CH.toUnixTimestamp64Nano($.Timestamp)).div(1_000_000)) + +// Shared filters + +export interface AgentSessionsFilterOpts { + /** Only sessions/traces containing a span from ANY of these vendor slugs. */ + vendors?: readonly string[] + /** Only sessions/traces containing an AI span from ANY of these services. */ + serviceNames?: readonly string[] + /** Only sessions/traces containing at least one Error-status AI span. */ + hasErrors?: boolean +} + +// Sessions list + +export interface AgentSessionsListOpts extends AgentSessionsFilterOpts { + limit?: number + offset?: number +} + +export interface AgentSessionsListOutput { + /** `toString(AiSessionKeyHash)` — a UInt64 rides the JSON wire as a string + * or corrupts above 2^53. Opaque id; the detail read resolves the display + * key from span attributes. */ + readonly sessionKeyHash: string + readonly startTime: string + /** Start of the latest key-carrying span (its runtime is folded into + * `durationMs`, not this timestamp). */ + readonly endTime: string + readonly durationMs: number + readonly traceCount: number + /** Counts only the session-authoritative spans that carry the key — the + * session's traces hold more spans, which the detail read fetches. Same + * caveat for `errorCount`. */ + readonly keyedSpanCount: number + readonly errorCount: number + readonly vendors: ReadonlyArray + readonly serviceNames: ReadonlyArray +} + +export function agentSessionsListQuery(opts: AgentSessionsListOpts) { + return from(Traces) + .select(($) => ({ + sessionKeyHash: CH.toString_($.AiSessionKeyHash), + startTime: CH.min_($.Timestamp), + endTime: CH.max_($.Timestamp), + durationMs: spanWindowMs($), + traceCount: CH.uniq($.TraceId), + keyedSpanCount: CH.count(), + errorCount: CH.countIf($.StatusCode.eq("Error")), + vendors: CH.groupUniqArray($.AiVendor), + serviceNames: CH.groupUniqArray($.ServiceName), + })) + .where(($) => [ + $.OrgId.eq(param.string("orgId")), + $.Timestamp.gte(param.dateTime("startTime")), + $.Timestamp.lte(param.dateTime("endTime")), + $.AiSessionKeyState.eq(SESSION_GRANULARITY), + ]) + .groupBy("sessionKeyHash") + .having(($) => [ + containsAny(CH.groupUniqArray($.AiVendor), opts.vendors), + containsAny(CH.groupUniqArray($.ServiceName), opts.serviceNames), + CH.whenTrue(opts.hasErrors, () => CH.countIf($.StatusCode.eq("Error")).gt(0)), + ]) + .orderBy(["endTime", "desc"]) + .limit(opts.limit ?? 50) + .offset(opts.offset ?? 0) + .format("JSON") +} + +// AI traces list + +export interface AgentTracesListOpts extends AgentSessionsFilterOpts { + limit?: number + offset?: number +} + +export interface AgentTracesListOutput { + readonly traceId: string + readonly startTime: string + readonly endTime: string + /** Window of the trace's AI spans only — a trace's full wall time can be + * wider (non-AI root, queue time). The trace detail view owns that number. */ + readonly durationMs: number + readonly aiSpanCount: number + readonly errorCount: number + readonly vendors: ReadonlyArray + readonly serviceNames: ReadonlyArray + /** Earliest AI span's name, as a row label. */ + readonly firstSpanName: string + /** max(AiSessionKeyState) across the trace's AI spans: 6 = belongs to a + * session, anything lower explains why it doesn't (write-side enum). */ + readonly bestSessionKeyState: number + /** Session-granularity key hash as a string, '' when the trace carries none. + * A trace spanning several sessions surfaces one arbitrarily (max). */ + readonly sessionKeyHash: string +} + +export function agentTracesListQuery(opts: AgentTracesListOpts) { + return from(Traces) + .select(($) => { + const sessionHash = CH.maxIf($.AiSessionKeyHash, $.AiSessionKeyState.eq(SESSION_GRANULARITY)) + return { + traceId: $.TraceId, + startTime: CH.min_($.Timestamp), + endTime: CH.max_($.Timestamp), + durationMs: spanWindowMs($), + aiSpanCount: CH.count(), + errorCount: CH.countIf($.StatusCode.eq("Error")), + vendors: CH.groupUniqArray($.AiVendor), + serviceNames: CH.groupUniqArray($.ServiceName), + firstSpanName: CH.argMin($.SpanName, $.Timestamp), + bestSessionKeyState: CH.max_($.AiSessionKeyState), + sessionKeyHash: CH.if_(sessionHash.gt(0), CH.toString_(sessionHash), CH.lit("")), + } + }) + .where(($) => [ + $.OrgId.eq(param.string("orgId")), + $.Timestamp.gte(param.dateTime("startTime")), + $.Timestamp.lte(param.dateTime("endTime")), + $.AiVendor.neq(""), + ]) + .groupBy("traceId") + .having(($) => [ + containsAny(CH.groupUniqArray($.AiVendor), opts.vendors), + containsAny(CH.groupUniqArray($.ServiceName), opts.serviceNames), + CH.whenTrue(opts.hasErrors, () => CH.countIf($.StatusCode.eq("Error")).gt(0)), + ]) + .orderBy(["startTime", "desc"]) + .limit(opts.limit ?? 50) + .offset(opts.offset ?? 0) + .format("JSON") +} + +// Facets (UNION ALL — vendor / service / error count) +// +// Counts are per session or per trace depending on the tab, so every branch +// reads the same grouped subquery the list uses and applies the *other* +// dimensions' filters to it — a selected vendor doesn't collapse the vendor +// facet to one option, but does narrow the service counts, and vice versa. + +export interface AgentSessionsFacetsOpts extends AgentSessionsFilterOpts { + /** Which tab's counting unit to use: distinct sessions or distinct traces. */ + tab: "sessions" | "traces" +} + +export type AgentSessionsFacetsOutput = FacetOutput + +type AgentFacetKey = "vendor" | "service" | "error" + +export function agentSessionsFacetsQuery( + opts: AgentSessionsFacetsOpts, +): CHUnionQuery { + const grouped = from(Traces) + .select(($) => ({ + groupKey: opts.tab === "sessions" ? CH.toString_($.AiSessionKeyHash) : $.TraceId, + vendors: CH.groupUniqArray($.AiVendor), + serviceNames: CH.groupUniqArray($.ServiceName), + errorCount: CH.countIf($.StatusCode.eq("Error")), + })) + .where(($) => [ + $.OrgId.eq(param.string("orgId")), + $.Timestamp.gte(param.dateTime("startTime")), + $.Timestamp.lte(param.dateTime("endTime")), + opts.tab === "sessions" ? $.AiSessionKeyState.eq(SESSION_GRANULARITY) : $.AiVendor.neq(""), + ]) + .groupBy("groupKey") + + // The fromQuery accessor's refs are untyped (`any`), so branch callbacks + // funnel through this typed view of the grouped row. + interface GroupedRefs { + readonly vendors: CH.Expr> + readonly serviceNames: CH.Expr> + readonly errorCount: CH.Expr + } + + const otherFilters = ( + $: GroupedRefs, + exclude: AgentFacetKey, + ): Array => [ + exclude === "vendor" ? undefined : containsAny($.vendors, opts.vendors), + exclude === "service" ? undefined : containsAny($.serviceNames, opts.serviceNames), + exclude === "error" ? undefined : CH.whenTrue(opts.hasErrors, () => $.errorCount.gt(0)), + ] + + // Each grouped row expands to one row per distinct array element, so a + // mixed-vendor session counts once under every vendor it contains. + const arrayFacet = (facetType: "vendor" | "service", column: "vendors" | "serviceNames") => + fromQuery(grouped, "g") + .select(($: any) => ({ + name: CH.arrayJoin($[column] as CH.Expr>), + count: CH.count(), + facetType: CH.lit(facetType), + })) + .where(($: any) => otherFilters($, facetType)) + // ServiceName can legitimately be '' — dropped like every facet sidebar + // does. Post-arrayJoin the alias only resolves in HAVING. + .having(() => [CH.dynamicColumn("name").neq("")]) + .groupBy("name") + .orderBy(["count", "desc"]) + .limit(50) + + const errorFacet = fromQuery(grouped, "g") + .select(() => ({ + name: CH.lit("error"), + count: CH.count(), + facetType: CH.lit("error"), + })) + .where(($: any) => [...otherFilters($, "error"), ($.errorCount as CH.Expr).gt(0)]) + + return unionAll( + arrayFacet("vendor", "vendors"), + arrayFacet("service", "serviceNames"), + errorFacet, + ).format("JSON") +} + +// Session detail (two-phase) +// +// Phase 1 resolves the session key hash to the session's TraceIds — only +// `AiSessionKeyState = 6` rows carry the hash. Phase 2 fetches ALL AI spans +// (`AiVendor != ''`) of those traces, because the session's substance usually +// lives on spans that do NOT carry the key: in the CrewAI shape the token +// counts sit on child openinference-openai LLM spans, and only the +// orchestration spans are keyed. Every fetched span runs through the vendor +// integration layer (`@maple/domain/ai`) on the read side; SpanAttributes +// travels whole because that layer owns which keys matter per vendor — a +// projected-key list here would couple the query to every integration's +// spellings. + +/** Caps, not pagination: a session past either bound is degenerate (a leaked + * process-wide key) and the detail view is the wrong lens for it. The read + * layer reports truncation rather than silently pretending completeness. */ +export const AGENT_SESSION_MAX_TRACES = 200 +export const AGENT_SESSION_MAX_SPANS = 2000 + +export interface AgentSessionTraceIdsOutput { + readonly traceId: string + /** Window of the trace's KEY-CARRYING spans only — phase 2 re-derives real + * bounds from all AI spans. These exist to give phase 2 a buffered + * partition hint. */ + readonly startTime: string + readonly endTime: string +} + +export function agentSessionTraceIdsQuery() { + return from(Traces) + .select(($) => ({ + traceId: $.TraceId, + startTime: CH.min_($.Timestamp), + endTime: CH.max_($.Timestamp), + })) + .where(($) => [ + $.OrgId.eq(param.string("orgId")), + $.Timestamp.gte(param.dateTime("startTime")), + $.Timestamp.lte(param.dateTime("endTime")), + $.AiSessionKeyState.eq(SESSION_GRANULARITY), + // String-side comparison: the hash is a UInt64 identity and 2^53-unsafe, + // so it crosses every boundary as a string (house rule). + CH.toString_($.AiSessionKeyHash).eq(param.string("sessionKeyHash")), + ]) + .groupBy("traceId") + .orderBy(["startTime", "asc"]) + .limit(AGENT_SESSION_MAX_TRACES) + .format("JSON") +} + +export interface AgentSessionSpansOpts { + /** Phase-1 TraceIds. Values from our own warehouse, not user input. */ + traceIds: readonly string[] +} + +export interface AgentSessionSpansOutput { + readonly traceId: string + readonly spanId: string + readonly parentSpanId: string + readonly timestamp: string + readonly durationMs: number + readonly spanName: string + readonly spanKind: string + readonly serviceName: string + readonly statusCode: string + readonly statusMessage: string + readonly vendor: string + readonly sessionKeyState: number + readonly spanAttributes: Record +} + +/** The `startTime`/`endTime` params are buffered phase-1 bounds — a partition + * hint (the sort key is (OrgId, ServiceName, SpanName, Timestamp), so TraceId + * never seeks), padded by the caller so AI spans adjacent to the keyed window + * aren't clipped. */ +export function agentSessionSpansQuery(opts: AgentSessionSpansOpts) { + return from(Traces) + .select(($) => ({ + traceId: $.TraceId, + spanId: $.SpanId, + parentSpanId: $.ParentSpanId, + timestamp: $.Timestamp, + durationMs: $.Duration.div(1_000_000), + spanName: $.SpanName, + spanKind: $.SpanKind, + serviceName: $.ServiceName, + statusCode: $.StatusCode, + statusMessage: $.StatusMessage, + vendor: $.AiVendor, + sessionKeyState: $.AiSessionKeyState, + spanAttributes: $.SpanAttributes, + })) + .where(($) => [ + $.OrgId.eq(param.string("orgId")), + $.Timestamp.gte(param.dateTime("startTime")), + $.Timestamp.lte(param.dateTime("endTime")), + $.TraceId.in_(...opts.traceIds), + $.AiVendor.neq(""), + ]) + .orderBy(["timestamp", "asc"], ["spanId", "asc"]) + .limit(AGENT_SESSION_MAX_SPANS) + .format("JSON") +} diff --git a/packages/query-engine/src/ch/tables.ts b/packages/query-engine/src/ch/tables.ts index 4d8bdfc82..8ea68dc72 100644 --- a/packages/query-engine/src/ch/tables.ts +++ b/packages/query-engine/src/ch/tables.ts @@ -38,6 +38,11 @@ export const Traces = table("traces", { ResourceAttributeItems: T.array(T.string), ScopeAttributeItems: T.array(T.string), SpanAttributeItems: T.array(T.string), + AiVendor: T.string, + AiSessionKeyState: T.uint8, + AiSessionKeyHash: T.uint64, + AiRulesVersion: T.uint32, + AiRollupHour: T.dateTime, }) export const TraceDetailSpans = table("trace_detail_spans", { diff --git a/packages/query-engine/src/observability/agent-sessions.ts b/packages/query-engine/src/observability/agent-sessions.ts new file mode 100644 index 000000000..4fcb7843e --- /dev/null +++ b/packages/query-engine/src/observability/agent-sessions.ts @@ -0,0 +1,354 @@ +import { Effect } from "effect" +import { normalizeAiSpan, type AiSpanFacts } from "@maple/domain/ai" +import * as CH from "../ch" +import { + formatWarehouseDateTime, + formatWarehouseDateTimeMs, + parseWarehouseDateTime, +} from "../datetime" +import { WarehouseExecutor } from "./WarehouseExecutor" + +export type { + AgentSessionsFacetsOutput, + AgentSessionsListOutput, + AgentTracesListOutput, +} from "../ch/queries/agent-sessions" + +/** + * The shared filter payload for the Agent Sessions feature — one shape feeds + * the sessions list, the raw AI-traces list, and both tabs' facets, so the + * sidebar counts can never mean something different from the rows. + * + * `vendors`/`serviceNames` are containment filters: "has at least one matching + * AI span". Classification is per-span and multi-vendor traces are the norm + * (a CrewAI orchestration span parenting openinference-instrumented OpenAI + * calls), so exact-match semantics would be a lie. + */ +export interface AgentSessionsFilterInput { + readonly startTime: string + readonly endTime: string + /** Vendor slugs from `@maple/domain` `AI_VENDORS`, including `unknown:*`. */ + readonly vendors?: readonly string[] + readonly serviceNames?: readonly string[] + readonly hasErrors?: boolean +} + +export interface ListAgentSessionsInput extends AgentSessionsFilterInput { + readonly limit?: number + readonly offset?: number +} + +/** + * List AI agent sessions: spans whose classified session key resolved at + * session granularity, grouped by key hash and ordered by latest activity. + * Counts cover only the key-carrying spans — resolving a session's full traces + * and display key is the detail read's job. + */ +export const listAgentSessions = Effect.fn("Observability.listAgentSessions")(function* ( + input: ListAgentSessionsInput, +) { + const executor = yield* WarehouseExecutor + yield* Effect.annotateCurrentSpan("orgId", executor.orgId) + const compiled = CH.compile( + CH.agentSessionsListQuery({ + vendors: input.vendors, + serviceNames: input.serviceNames, + hasErrors: input.hasErrors, + limit: input.limit, + offset: input.offset, + }), + { orgId: executor.orgId, startTime: input.startTime, endTime: input.endTime }, + ) + return yield* executor.compiledQuery(compiled, { profile: "list", context: "listAgentSessions" }) +}) + +export interface ListAgentTracesInput extends AgentSessionsFilterInput { + readonly limit?: number + readonly offset?: number +} + +/** + * List raw AI traces: every AI-classified span at any session-key state, + * grouped by trace. This is the companion tab to `listAgentSessions` — spans + * that never resolved a session key (`bestSessionKeyState < 6`) only surface + * here, with the state explaining why. + */ +export const listAgentTraces = Effect.fn("Observability.listAgentTraces")(function* ( + input: ListAgentTracesInput, +) { + const executor = yield* WarehouseExecutor + yield* Effect.annotateCurrentSpan("orgId", executor.orgId) + const compiled = CH.compile( + CH.agentTracesListQuery({ + vendors: input.vendors, + serviceNames: input.serviceNames, + hasErrors: input.hasErrors, + limit: input.limit, + offset: input.offset, + }), + { orgId: executor.orgId, startTime: input.startTime, endTime: input.endTime }, + ) + return yield* executor.compiledQuery(compiled, { profile: "list", context: "listAgentTraces" }) +}) + +export interface AgentSessionsFacetsInput extends AgentSessionsFilterInput { + /** Counting unit: distinct sessions or distinct traces — match the open tab. */ + readonly tab: "sessions" | "traces" +} + +/** + * Facet counts (vendor / service / has-errors) for the quickfilter sidebar, + * counted per session or per trace to match the open tab. Each dimension's own + * selection is excluded from its branch so it doesn't collapse to one option. + */ +export const agentSessionsFacets = Effect.fn("Observability.agentSessionsFacets")(function* ( + input: AgentSessionsFacetsInput, +) { + const executor = yield* WarehouseExecutor + yield* Effect.annotateCurrentSpan("orgId", executor.orgId) + const compiled = CH.compileUnion( + CH.agentSessionsFacetsQuery({ + tab: input.tab, + vendors: input.vendors, + serviceNames: input.serviceNames, + hasErrors: input.hasErrors, + }), + { orgId: executor.orgId, startTime: input.startTime, endTime: input.endTime }, + ) + return yield* executor.compiledQuery(compiled, { + profile: "list", + context: "agentSessionsFacets", + }) +}) + +// Session detail + +export interface AgentSessionDetailInput { + /** Opaque session id from the list read: `toString(AiSessionKeyHash)`. */ + readonly sessionKeyHash: string + readonly startTime: string + readonly endTime: string +} + +/** An AI span with the vendor-integration facts merged in — raw span identity + * and timing plus what `@maple/domain/ai` could normalize out of its + * attributes. The attribute maps themselves stay behind this boundary; only + * the extracted conversational text crosses it, truncated. */ +export interface NormalizedAiSpan extends AiSpanFacts { + readonly traceId: string + readonly spanId: string + readonly parentSpanId: string + readonly startTime: string + readonly durationMs: number + readonly spanName: string + readonly spanKind: string + readonly serviceName: string + readonly statusCode: string + readonly statusMessage: string + readonly vendor: string + readonly sessionKeyState: number +} + +export interface AgentSessionTrace { + readonly traceId: string + readonly startTime: string + readonly durationMs: number + readonly errorCount: number + readonly spans: ReadonlyArray +} + +export interface AgentSessionDetailOutput { + readonly sessionKeyHash: string + /** The plaintext session key, resolved from span attributes by the vendor + * integration of a key-carrying span. `null` when no integration knows the + * vendor's spelling — the UI falls back to the hash. */ + readonly sessionKey: string | null + readonly startTime: string + readonly endTime: string + readonly durationMs: number + /** True when the session blew a fetch cap (`AGENT_SESSION_MAX_TRACES` / + * `AGENT_SESSION_MAX_SPANS`) and the numbers below undercount. */ + readonly truncated: boolean + readonly totals: { + readonly spanCount: number + readonly llmCallCount: number + readonly toolCallCount: number + readonly errorCount: number + readonly inputTokens: number + readonly outputTokens: number + readonly cacheReadTokens: number + readonly cacheCreationTokens: number + /** `null` when no span priced itself — "unknown", never "free". */ + readonly costUsd: number | null + } + readonly vendors: ReadonlyArray + readonly serviceNames: ReadonlyArray + readonly models: ReadonlyArray + readonly traces: ReadonlyArray +} + +/** Ceiling for each span's extracted conversational text on this interface — + * a single agent prompt can be hundreds of KB and a session holds up to 2000 + * spans, so uncapped content makes the payload unusable long before the UI + * could render it. */ +const MAX_CONTENT_CHARS = 10_000 + +const truncateContent = (text: string | null): string | null => + text !== null && text.length > MAX_CONTENT_CHARS + ? `${text.slice(0, MAX_CONTENT_CHARS)}… [truncated]` + : text + +/** Phase-2 pad around the keyed spans' window. The key-carrying spans bound + * the session's *anchors*, not its traces: sibling AI spans (the token-bearing + * LLM children) start before and end after them. An hour each side is far + * beyond any sane trace's skew and still prunes partitions. */ +const SPAN_WINDOW_PAD_MS = 60 * 60 * 1000 + +/** + * The session detail read, and the one durable interface over it — the + * internal HTTP route adapts this function and the future MCP tool calls it + * too. Two phases: the key hash resolves to the session's TraceIds (only + * `AiSessionKeyState = 6` rows carry the hash), then every AI span of those + * traces is fetched and run through the per-span vendor integration layer. + * Totals sum over ALL normalized spans regardless of vendor: in the mixed + * CrewAI shape the tokens live on child openinference-openai spans, not on the + * session-keyed ones. + * + * Returns `null` when the hash matches nothing in the window. + */ +export const getAgentSessionDetail = Effect.fn("Observability.getAgentSessionDetail")(function* ( + input: AgentSessionDetailInput, +) { + const executor = yield* WarehouseExecutor + yield* Effect.annotateCurrentSpan("orgId", executor.orgId) + + const traceRows = yield* executor.compiledQuery( + CH.compile(CH.agentSessionTraceIdsQuery(), { + orgId: executor.orgId, + startTime: input.startTime, + endTime: input.endTime, + sessionKeyHash: input.sessionKeyHash, + }), + { profile: "list", context: "agentSessionTraceIds" }, + ) + if (traceRows.length === 0) return null + + const keyedStartMs = Math.min(...traceRows.map((row) => parseWarehouseDateTime(row.startTime))) + const keyedEndMs = Math.max(...traceRows.map((row) => parseWarehouseDateTime(row.endTime))) + const spanRows = yield* executor.compiledQuery( + CH.compile(CH.agentSessionSpansQuery({ traceIds: traceRows.map((row) => row.traceId) }), { + orgId: executor.orgId, + startTime: formatWarehouseDateTime(keyedStartMs - SPAN_WINDOW_PAD_MS), + endTime: formatWarehouseDateTime(keyedEndMs + SPAN_WINDOW_PAD_MS), + }), + { profile: "list", context: "agentSessionSpans" }, + ) + + const spans: NormalizedAiSpan[] = spanRows.map((row) => { + const facts = normalizeAiSpan({ + vendor: row.vendor, + spanName: row.spanName, + attributes: row.spanAttributes, + }) + return { + traceId: row.traceId, + spanId: row.spanId, + parentSpanId: row.parentSpanId, + startTime: row.timestamp, + durationMs: Number(row.durationMs), + spanName: row.spanName, + spanKind: row.spanKind, + serviceName: row.serviceName, + statusCode: row.statusCode, + statusMessage: row.statusMessage, + vendor: row.vendor, + sessionKeyState: Number(row.sessionKeyState), + ...facts, + inputText: truncateContent(facts.inputText), + outputText: truncateContent(facts.outputText), + } + }) + + const totals = { + spanCount: spans.length, + llmCallCount: 0, + toolCallCount: 0, + errorCount: 0, + inputTokens: 0, + outputTokens: 0, + cacheReadTokens: 0, + cacheCreationTokens: 0, + costUsd: null as number | null, + } + let sessionKey: string | null = null + let startMs = Number.POSITIVE_INFINITY + let endMs = Number.NEGATIVE_INFINITY + for (const span of spans) { + if (span.role === "tool") totals.toolCallCount += 1 + if (span.statusCode === "Error") totals.errorCount += 1 + // Token/cost totals sum the llm tier ONLY: agent-tier wrappers repeat + // their children's aggregated usage in both AI SDK dialects + // (`invoke_agent` mirrors its `chat`, `ai.generateText` its + // `.doGenerate`s), so summing every token-carrying span double-counts. + // The integrations keep those spans out of the llm role for exactly this + // reason. `llmCallCount` counts the calls that reported usage — a + // framework's extra llm-tier wrapper spans (mastra's `model_step`) carry + // none and would inflate a plain role count. + if (span.role === "llm") { + totals.inputTokens += span.inputTokens ?? 0 + totals.outputTokens += span.outputTokens ?? 0 + totals.cacheReadTokens += span.cacheReadTokens ?? 0 + totals.cacheCreationTokens += span.cacheCreationTokens ?? 0 + if (span.costUsd !== null) totals.costUsd = (totals.costUsd ?? 0) + span.costUsd + if (span.inputTokens !== null || span.outputTokens !== null) totals.llmCallCount += 1 + } + // The display key must be the value the stamped hash was computed FROM, so + // only a session-granularity span's extraction qualifies — any other + // span's sessionKey fact may be a different (sub-session) identifier. + if (sessionKey === null && span.sessionKey !== null && span.sessionKeyState === 6) { + sessionKey = span.sessionKey + } + const spanStart = parseWarehouseDateTime(span.startTime) + if (spanStart < startMs) startMs = spanStart + if (spanStart + span.durationMs > endMs) endMs = spanStart + span.durationMs + } + + const byTrace = new Map() + for (const span of spans) { + const bucket = byTrace.get(span.traceId) + if (bucket === undefined) byTrace.set(span.traceId, [span]) + else bucket.push(span) + } + const traces: AgentSessionTrace[] = [...byTrace.entries()].map(([traceId, traceSpans]) => { + const start = parseWarehouseDateTime(traceSpans[0]!.startTime) + const end = Math.max( + ...traceSpans.map((span) => parseWarehouseDateTime(span.startTime) + span.durationMs), + ) + return { + traceId, + startTime: traceSpans[0]!.startTime, + durationMs: end - start, + errorCount: traceSpans.filter((span) => span.statusCode === "Error").length, + spans: traceSpans, + } + }) + + const distinct = (values: ReadonlyArray) => + [...new Set(values.filter((value): value is string => value !== null && value !== ""))].sort() + + return { + sessionKeyHash: input.sessionKeyHash, + sessionKey, + startTime: formatWarehouseDateTimeMs(startMs), + endTime: formatWarehouseDateTimeMs(endMs), + durationMs: endMs - startMs, + truncated: + traceRows.length >= CH.AGENT_SESSION_MAX_TRACES || + spans.length >= CH.AGENT_SESSION_MAX_SPANS, + totals, + vendors: distinct(spans.map((span) => span.vendor)), + serviceNames: distinct(spans.map((span) => span.serviceName)), + models: distinct(spans.map((span) => span.model)), + traces, + } satisfies AgentSessionDetailOutput +}) diff --git a/packages/query-engine/src/observability/index.ts b/packages/query-engine/src/observability/index.ts index d7df4dfe7..5b650f1c2 100644 --- a/packages/query-engine/src/observability/index.ts +++ b/packages/query-engine/src/observability/index.ts @@ -45,3 +45,20 @@ export { type SessionReplayDetailOutput, type SessionTraceSummaryOutput, } from "./session-replays" +export { + listAgentSessions, + listAgentTraces, + agentSessionsFacets, + getAgentSessionDetail, + type AgentSessionsFilterInput, + type ListAgentSessionsInput, + type ListAgentTracesInput, + type AgentSessionsFacetsInput, + type AgentSessionsListOutput, + type AgentTracesListOutput, + type AgentSessionsFacetsOutput, + type AgentSessionDetailInput, + type AgentSessionDetailOutput, + type AgentSessionTrace, + type NormalizedAiSpan, +} from "./agent-sessions" diff --git a/packages/query-engine/src/sql-catalog.test.ts b/packages/query-engine/src/sql-catalog.test.ts index 4f8cbc50c..af18bd153 100644 --- a/packages/query-engine/src/sql-catalog.test.ts +++ b/packages/query-engine/src/sql-catalog.test.ts @@ -15,6 +15,7 @@ import { } from "./sql-catalog" import { builderFixtures } from "./ch/builder-fixtures" import * as activityQueries from "./ch/queries/activity" +import * as agentSessionQueries from "./ch/queries/agent-sessions" import * as alertCheckQueries from "./ch/queries/alert-checks" import * as anomalyQueries from "./ch/queries/anomaly" import * as attributeKeyQueries from "./ch/queries/attribute-keys" @@ -177,6 +178,7 @@ describe("sql catalog", () => { const QUERY_MODULES: Record> = { activity: activityQueries, + "agent-sessions": agentSessionQueries, "alert-checks": alertCheckQueries, anomaly: anomalyQueries, "attribute-keys": attributeKeyQueries, diff --git a/scripts/truncate-tinybird-local.ts b/scripts/truncate-tinybird-local.ts new file mode 100644 index 000000000..210ad7135 --- /dev/null +++ b/scripts/truncate-tinybird-local.ts @@ -0,0 +1,65 @@ +#!/usr/bin/env bun +/** + * Truncate every datasource in the local Tinybird workspace — a clean slate for + * re-seeding during UI development. + * + * bun run tinybird:truncate # all datasources + * bun run tinybird:truncate traces # just these, by name + * + * Reads TINYBIRD_HOST / TINYBIRD_TOKEN from the environment (.env.local at the + * repo root — bun loads it automatically). The datasource list comes from the + * workspace itself, so new datasources never need adding here. + * + * Refuses non-localhost hosts: .env.local sometimes points at a cloud + * workspace (PR-branch and staging token swaps), and this command must never + * be one stale env file away from emptying it. --force overrides. + */ + +const args = process.argv.slice(2) +const force = args.includes("--force") +const only = new Set(args.filter((a) => !a.startsWith("--"))) + +const host = (process.env.TINYBIRD_HOST ?? "http://localhost:7181").replace(/\/+$/, "") +const token = process.env.TINYBIRD_TOKEN +if (!token) { + console.error("TINYBIRD_TOKEN is not set — is .env.local present at the repo root?") + process.exit(1) +} + +const hostname = new URL(host).hostname +const isLocal = hostname === "localhost" || hostname === "127.0.0.1" || hostname.endsWith(".localhost") +if (!isLocal && !force) { + console.error(`refusing to truncate non-local Tinybird at ${host} (pass --force if you really mean it)`) + process.exit(1) +} + +const authed = { headers: { Authorization: `Bearer ${token}` } } +const list = await fetch(`${host}/v0/datasources`, authed) +if (!list.ok) { + console.error(`GET /v0/datasources failed: ${list.status} ${await list.text()}`) + process.exit(1) +} +const { datasources } = (await list.json()) as { datasources: Array<{ name: string }> } + +const targets = datasources.map((d) => d.name).filter((name) => only.size === 0 || only.has(name)) +const unknown = [...only].filter((name) => !targets.includes(name)) +if (unknown.length > 0) { + console.error(`no such datasource: ${unknown.join(", ")}`) + process.exit(1) +} +if (targets.length === 0) { + console.log("no datasources to truncate") + process.exit(0) +} + +let failed = false +for (const name of targets) { + const res = await fetch(`${host}/v0/datasources/${name}/truncate`, { method: "POST", ...authed }) + if (res.ok) { + console.log(`truncated ${name}`) + } else { + failed = true + console.error(`truncate ${name} failed: ${res.status} ${await res.text()}`) + } +} +process.exit(failed ? 1 : 0)