From 0d760565f3ba42586956c615a0b9112da8a2c5d5 Mon Sep 17 00:00:00 2001 From: JeremyFunk Date: Mon, 17 Aug 2026 13:36:49 +0200 Subject: [PATCH 1/9] feat(query-engine): mirror the AI classification columns on the traces DSL table MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The five Ai* columns landed in the warehouse schema (migration 0016) but the query DSL's traces mirror never learned them, so no read query could reference them. The mirror test only catches unknown or mistyped columns, not missing ones — this is the read path's first prerequisite, not a test failure. Co-Authored-By: Claude Fable 5 --- packages/query-engine/src/ch/tables.ts | 5 +++++ 1 file changed, 5 insertions(+) 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", { From 244dc5af3697f5daf6fbe56091c27a633e88f870 Mon Sep 17 00:00:00 2001 From: JeremyFunk Date: Mon, 17 Aug 2026 14:03:40 +0200 Subject: [PATCH 2/9] feat(query-engine): agent-sessions list, AI-traces list, and shared facets queries MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The first read queries over the AI classification columns. Sessions are state-6 rows grouped by AiSessionKeyHash; the traces tab is every classified span grouped by TraceId. Vendor/service filters are containment predicates in HAVING — per-span classification means one trace or session routinely mixes vendors, and a WHERE would drop the non-matching spans from the row's own aggregates. Facet branches arrayJoin one grouped subquery so their counts share the list's grouping by construction. Swept through the real ClickHouse analyzer (sql-catalog e2e, 150 checks). Co-Authored-By: Claude Fable 5 --- .../src/__sql_baseline__/catalog.sql | 219 ++++++++++++++ .../query-engine/src/ch/builder-fixtures.ts | 55 ++++ packages/query-engine/src/ch/index.ts | 14 + .../src/ch/queries/agent-sessions.test.ts | 131 ++++++++ .../src/ch/queries/agent-sessions.ts | 280 ++++++++++++++++++ packages/query-engine/src/sql-catalog.test.ts | 2 + 6 files changed, 701 insertions(+) create mode 100644 packages/query-engine/src/ch/queries/agent-sessions.test.ts create mode 100644 packages/query-engine/src/ch/queries/agent-sessions.ts diff --git a/packages/query-engine/src/__sql_baseline__/catalog.sql b/packages/query-engine/src/__sql_baseline__/catalog.sql index 1390307a8..15e35861c 100644 --- a/packages/query-engine/src/__sql_baseline__/catalog.sql +++ b/packages/query-engine/src/__sql_baseline__/catalog.sql @@ -22,6 +22,225 @@ 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: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..06ff69fbc 100644 --- a/packages/query-engine/src/ch/builder-fixtures.ts +++ b/packages/query-engine/src/ch/builder-fixtures.ts @@ -632,6 +632,61 @@ 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, + ), + }, + // ----- 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..ae31bbba3 100644 --- a/packages/query-engine/src/ch/index.ts +++ b/packages/query-engine/src/ch/index.ts @@ -151,6 +151,20 @@ export { type SessionActivityOutput, } from "./queries/session-events" +// Queries — Agent Sessions (AI-classified spans: sessions + raw AI traces) +export { + agentSessionsListQuery, + agentTracesListQuery, + agentSessionsFacetsQuery, + type AgentSessionsFilterOpts, + type AgentSessionsListOpts, + type AgentSessionsListOutput, + type AgentTracesListOpts, + type AgentTracesListOutput, + type AgentSessionsFacetsOpts, + type AgentSessionsFacetsOutput, +} 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..f1dcfcfec --- /dev/null +++ b/packages/query-engine/src/ch/queries/agent-sessions.test.ts @@ -0,0 +1,131 @@ +import { describe, expect, it } from "@effect/vitest" +import { compileCH, compileUnion } from "@maple-dev/clickhouse-builder" +import { + 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 != ''") + }) +}) 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..d15f4b446 --- /dev/null +++ b/packages/query-engine/src/ch/queries/agent-sessions.ts @@ -0,0 +1,280 @@ +// 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") +} 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, From 11aed015fb02b655603a6e78496a1770d081c403 Mon Sep 17 00:00:00 2001 From: JeremyFunk Date: Mon, 17 Aug 2026 15:15:13 +0200 Subject: [PATCH 3/9] =?UTF-8?q?feat(api):=20agent-sessions=20read=20interf?= =?UTF-8?q?ace=20=E2=80=94=20observability=20functions=20+=20internal=20ro?= =?UTF-8?q?utes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit listAgentSessions / listAgentTraces / agentSessionsFacets live in query-engine/observability over WarehouseExecutor: one read interface, so the internal HTTP routes here are thin adapters and the future MCP tools call the same functions. The domain contract is internal-tier on purpose — the feature is a product scratchpad and these shapes follow the UI. Co-Authored-By: Claude Fable 5 --- .../routes/internal/agent-sessions.http.ts | 122 +++++++++++++++ apps/api/src/runtime/http-graph.ts | 5 +- packages/domain/src/http/agent-sessions.ts | 148 ++++++++++++++++++ packages/domain/src/http/index.ts | 1 + packages/domain/src/http/internal-api.ts | 2 + .../src/observability/agent-sessions.ts | 116 ++++++++++++++ .../query-engine/src/observability/index.ts | 12 ++ 7 files changed, 405 insertions(+), 1 deletion(-) create mode 100644 apps/api/src/routes/internal/agent-sessions.http.ts create mode 100644 packages/domain/src/http/agent-sessions.ts create mode 100644 packages/query-engine/src/observability/agent-sessions.ts 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..3e374813c --- /dev/null +++ b/apps/api/src/routes/internal/agent-sessions.http.ts @@ -0,0 +1,122 @@ +import { HttpApiBuilder } from "effect/unstable/httpapi" +import { + AgentSessionsFacetsResponse, + AgentSessionsListResponse, + AgentTracesListResponse, + CurrentTenant, + MapleInternalApi, + TraceId, +} from "@maple/domain/http" +import { Effect, Schema } from "effect" +import { + agentSessionsFacets, + 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("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/packages/domain/src/http/agent-sessions.ts b/packages/domain/src/http/agent-sessions.ts new file mode 100644 index 000000000..89008393c --- /dev/null +++ b/packages/domain/src/http/agent-sessions.ts @@ -0,0 +1,148 @@ +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, +}) {} + +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("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/observability/agent-sessions.ts b/packages/query-engine/src/observability/agent-sessions.ts new file mode 100644 index 000000000..65e668757 --- /dev/null +++ b/packages/query-engine/src/observability/agent-sessions.ts @@ -0,0 +1,116 @@ +import { Effect } from "effect" +import * as CH from "../ch" +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", + }) +}) diff --git a/packages/query-engine/src/observability/index.ts b/packages/query-engine/src/observability/index.ts index d7df4dfe7..76dd8b7d0 100644 --- a/packages/query-engine/src/observability/index.ts +++ b/packages/query-engine/src/observability/index.ts @@ -45,3 +45,15 @@ export { type SessionReplayDetailOutput, type SessionTraceSummaryOutput, } from "./session-replays" +export { + listAgentSessions, + listAgentTraces, + agentSessionsFacets, + type AgentSessionsFilterInput, + type ListAgentSessionsInput, + type ListAgentTracesInput, + type AgentSessionsFacetsInput, + type AgentSessionsListOutput, + type AgentTracesListOutput, + type AgentSessionsFacetsOutput, +} from "./agent-sessions" From f9b201103dcba3977ac8db1c0161cda3e22ed8ce Mon Sep 17 00:00:00 2001 From: JeremyFunk Date: Mon, 17 Aug 2026 15:15:25 +0200 Subject: [PATCH 4/9] feat(web): throwaway Agent Sessions dashboard scratchpad MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Owner-flagged throwaway to try out the product shape — one route file, no abstractions, expect a full rebuild once it settles. Two tabs (sessions / raw AI traces) over one filter payload, vendor/service/error quickfilters from the facets endpoint. Co-Authored-By: Claude Fable 5 --- apps/web/src/api/warehouse/agent-sessions.ts | 116 ++++++ .../web/src/components/dashboard/nav-items.ts | 4 + .../services/atoms/warehouse-query-atoms.ts | 17 + apps/web/src/routeTree.gen.ts | 21 + apps/web/src/routes/agent-sessions/index.tsx | 382 ++++++++++++++++++ 5 files changed, 540 insertions(+) create mode 100644 apps/web/src/api/warehouse/agent-sessions.ts create mode 100644 apps/web/src/routes/agent-sessions/index.tsx 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..1807fe4a5 --- /dev/null +++ b/apps/web/src/routes/agent-sessions/index.tsx @@ -0,0 +1,382 @@ +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 } 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) => { + const { startTime, endTime } = resolveEffectiveTimeRange( + search.startTime, + search.endTime, + search.timePreset ?? "24h", + ) + return { + startTime, + endTime, + vendors: search.vendors, + serviceNames: search.services, + hasErrors: search.hasErrors, + } +} + +const PAGE_SIZE = 50 + +export const Route = createFileRoute("/agent-sessions/")({ + component: AgentSessionsPage, + 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)} + + ))} + + ) +} + +function AgentSessionsPage() { + const search = Route.useSearch() + const navigate = useNavigate({ from: Route.fullPath }) + const tab = search.tab ?? "sessions" + + const inputs = useMemo( + () => filterInputs(search), + [ + search.startTime, + search.endTime, + search.timePreset, + search.vendors, + search.services, + search.hasErrors, + ], + ) + + 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} + + + + + + ) +} From 7452a57d9d3a7b254a355d8083c2d20d38cd3aae Mon Sep 17 00:00:00 2001 From: JeremyFunk Date: Mon, 17 Aug 2026 18:16:45 +0200 Subject: [PATCH 5/9] fix(web): let Reload bypass the cache-grid window snap on agent sessions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 24h preset snaps its window end to a 5-minute grid for stable atom keys, so rows ingested moments ago sit outside the window — and the route ignored refreshVersion, so Reload re-fetched the same stale window. Wire the PageRefreshProvider above the page and drop the snap for the reload mount, mirroring useEffectiveTimeRange. Co-Authored-By: Claude Fable 5 --- apps/web/src/routes/agent-sessions/index.tsx | 32 +++++++++++++++----- 1 file changed, 24 insertions(+), 8 deletions(-) diff --git a/apps/web/src/routes/agent-sessions/index.tsx b/apps/web/src/routes/agent-sessions/index.tsx index 1807fe4a5..ca17d3bd2 100644 --- a/apps/web/src/routes/agent-sessions/index.tsx +++ b/apps/web/src/routes/agent-sessions/index.tsx @@ -14,7 +14,10 @@ import { } 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 } from "@/components/time-range-picker/page-refresh-context" +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" @@ -44,11 +47,12 @@ const agentSessionsSearchSchema = Schema.Struct({ type AgentSessionsSearch = typeof agentSessionsSearchSchema.Type -const filterInputs = (search: AgentSessionsSearch) => { +const filterInputs = (search: AgentSessionsSearch, options?: { snap?: boolean }) => { const { startTime, endTime } = resolveEffectiveTimeRange( search.startTime, search.endTime, search.timePreset ?? "24h", + options, ) return { startTime, @@ -62,7 +66,7 @@ const filterInputs = (search: AgentSessionsSearch) => { const PAGE_SIZE = 50 export const Route = createFileRoute("/agent-sessions/")({ - component: AgentSessionsPage, + component: AgentSessionsRoute, validateSearch: Schema.toStandardSchemaV1(agentSessionsSearchSchema), loaderDeps: ({ search }) => search, loader: ({ context, deps }) => { @@ -110,13 +114,26 @@ function VendorChips({ vendors }: { vendors: ReadonlyArray }) { ) } +// 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), + () => filterInputs(search, { snap: refreshVersion === 0 }), [ search.startTime, search.endTime, @@ -124,6 +141,7 @@ function AgentSessionsPage() { search.vendors, search.services, search.hasErrors, + refreshVersion, ], ) @@ -335,8 +353,7 @@ function AgentSessionsPage() { const { startTime, endTime } = inputs return ( - - + {sidebar} @@ -376,7 +393,6 @@ function AgentSessionsPage() { - - + ) } From 9a58ff6d50802ae95cf60b3760e79b590346d997 Mon Sep 17 00:00:00 2001 From: JeremyFunk Date: Mon, 17 Aug 2026 20:49:41 +0200 Subject: [PATCH 6/9] chore(scripts): add tinybird:truncate for wiping the local workspace Lists the workspace's datasources and truncates them (optionally filtered by name); refuses non-localhost hosts without --force since .env.local sometimes carries cloud tokens. Co-Authored-By: Claude Fable 5 --- package.json | 1 + scripts/truncate-tinybird-local.ts | 65 ++++++++++++++++++++++++++++++ 2 files changed, 66 insertions(+) create mode 100644 scripts/truncate-tinybird-local.ts 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/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) From d191da8cf418a756b73e909e69b06c2fb803a50f Mon Sep 17 00:00:00 2001 From: JeremyFunk Date: Tue, 18 Aug 2026 12:26:49 +0200 Subject: [PATCH 7/9] feat(domain): per-span vendor integration layer for the AI read path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit normalizeAiSpan derives facts (role, model, tokens, cost, session display key, conversational content) from a classified span's attributes. A base normalizer speaks standard gen_ai.* semconv; vendor integrations override only what their wire format spells differently, verified against captured data: mastra, eve, vercel_ai_sdk (both dialects), claude_agent_sdk. Vendors without an integration and the unknown:* buckets run the base. Dispatch is strictly per span on the stamped AiVendor slug — a single trace routinely mixes vendors (CrewAI orchestration parenting openinference-openai LLM spans), so per-trace dispatch would misread the proven-normal case. Wrapper spans that repeat their children's aggregated usage (invoke_agent, the legacy AI SDK umbrellas) stay out of the llm role so llm-tier token totals never double-count. Co-Authored-By: Claude Fable 5 --- packages/domain/src/ai/index.ts | 7 + packages/domain/src/ai/integrations.test.ts | 579 ++++++++++++++++++++ packages/domain/src/ai/integrations.ts | 391 +++++++++++++ 3 files changed, 977 insertions(+) create mode 100644 packages/domain/src/ai/integrations.test.ts create mode 100644 packages/domain/src/ai/integrations.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 +} From f4efcb36f8f65352e4bee17a2569a79867d12e64 Mon Sep 17 00:00:00 2001 From: JeremyFunk Date: Tue, 18 Aug 2026 12:27:05 +0200 Subject: [PATCH 8/9] feat(query-engine): agent-session detail read MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two-phase detail: the session key hash resolves to TraceIds (only AiSessionKeyState = 6 rows carry the hash), then every AI span of those traces is fetched — the session's substance usually lives on spans that do not carry the key — and run through the vendor integration layer. getAgentSessionDetail in observability/ is the one durable interface: the internal HTTP route adapts it and the future MCP tool calls it too. Phase 2 takes buffered phase-1 time bounds as a partition hint (TraceId never seeks on the traces sort key) and ships the full SpanAttributes map — the integration layer owns which keys matter per vendor, so a projected key list here would couple the query to every integration. Fetch caps (200 traces / 2000 spans) surface as a truncated flag; extracted conversational text is capped server-side per span. Totals sum the llm role tier only: agent-tier wrappers repeat their children's usage in both AI SDK dialects and would double-count. Co-Authored-By: Claude Fable 5 --- .../src/__sql_baseline__/catalog.sql | 41 +++ .../query-engine/src/ch/builder-fixtures.ts | 22 ++ packages/query-engine/src/ch/index.ts | 7 + .../src/ch/queries/agent-sessions.test.ts | 47 ++++ .../src/ch/queries/agent-sessions.ts | 104 ++++++++ .../src/observability/agent-sessions.ts | 238 ++++++++++++++++++ .../query-engine/src/observability/index.ts | 5 + 7 files changed, 464 insertions(+) diff --git a/packages/query-engine/src/__sql_baseline__/catalog.sql b/packages/query-engine/src/__sql_baseline__/catalog.sql index 15e35861c..244f39cb6 100644 --- a/packages/query-engine/src/__sql_baseline__/catalog.sql +++ b/packages/query-engine/src/__sql_baseline__/catalog.sql @@ -191,6 +191,47 @@ SELECT 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, diff --git a/packages/query-engine/src/ch/builder-fixtures.ts b/packages/query-engine/src/ch/builder-fixtures.ts index 06ff69fbc..1aa5bca30 100644 --- a/packages/query-engine/src/ch/builder-fixtures.ts +++ b/packages/query-engine/src/ch/builder-fixtures.ts @@ -686,6 +686,28 @@ export const builderFixtures: ReadonlyArray = [ 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 diff --git a/packages/query-engine/src/ch/index.ts b/packages/query-engine/src/ch/index.ts index ae31bbba3..ea0cc57bb 100644 --- a/packages/query-engine/src/ch/index.ts +++ b/packages/query-engine/src/ch/index.ts @@ -156,6 +156,10 @@ export { agentSessionsListQuery, agentTracesListQuery, agentSessionsFacetsQuery, + agentSessionTraceIdsQuery, + agentSessionSpansQuery, + AGENT_SESSION_MAX_TRACES, + AGENT_SESSION_MAX_SPANS, type AgentSessionsFilterOpts, type AgentSessionsListOpts, type AgentSessionsListOutput, @@ -163,6 +167,9 @@ export { 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) diff --git a/packages/query-engine/src/ch/queries/agent-sessions.test.ts b/packages/query-engine/src/ch/queries/agent-sessions.test.ts index f1dcfcfec..f8c3517e2 100644 --- a/packages/query-engine/src/ch/queries/agent-sessions.test.ts +++ b/packages/query-engine/src/ch/queries/agent-sessions.test.ts @@ -1,6 +1,10 @@ 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, @@ -129,3 +133,46 @@ describe("agentSessionsFacetsQuery", () => { 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 index d15f4b446..150b0df91 100644 --- a/packages/query-engine/src/ch/queries/agent-sessions.ts +++ b/packages/query-engine/src/ch/queries/agent-sessions.ts @@ -278,3 +278,107 @@ export function agentSessionsFacetsQuery( 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/observability/agent-sessions.ts b/packages/query-engine/src/observability/agent-sessions.ts index 65e668757..4fcb7843e 100644 --- a/packages/query-engine/src/observability/agent-sessions.ts +++ b/packages/query-engine/src/observability/agent-sessions.ts @@ -1,5 +1,11 @@ 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 { @@ -114,3 +120,235 @@ export const agentSessionsFacets = Effect.fn("Observability.agentSessionsFacets" 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 76dd8b7d0..5b650f1c2 100644 --- a/packages/query-engine/src/observability/index.ts +++ b/packages/query-engine/src/observability/index.ts @@ -49,6 +49,7 @@ export { listAgentSessions, listAgentTraces, agentSessionsFacets, + getAgentSessionDetail, type AgentSessionsFilterInput, type ListAgentSessionsInput, type ListAgentTracesInput, @@ -56,4 +57,8 @@ export { type AgentSessionsListOutput, type AgentTracesListOutput, type AgentSessionsFacetsOutput, + type AgentSessionDetailInput, + type AgentSessionDetailOutput, + type AgentSessionTrace, + type NormalizedAiSpan, } from "./agent-sessions" From cffd8653d087c5e16acfae436ba6741c87c78ad6 Mon Sep 17 00:00:00 2001 From: JeremyFunk Date: Tue, 18 Aug 2026 12:27:13 +0200 Subject: [PATCH 9/9] feat(api): agent-session detail endpoint MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit POST /internal/agent-sessions/detail — a thin adapter over getAgentSessionDetail. The response ships normalized AI spans only: the raw attribute maps stay behind the read interface, conversational content travels as the integration layer's extracted (and truncated) inputText/outputText, and null session means the hash matched nothing in the window. Co-Authored-By: Claude Fable 5 --- .../routes/internal/agent-sessions.http.ts | 33 +++++++ packages/domain/src/http/agent-sessions.ts | 89 +++++++++++++++++++ 2 files changed, 122 insertions(+) diff --git a/apps/api/src/routes/internal/agent-sessions.http.ts b/apps/api/src/routes/internal/agent-sessions.http.ts index 3e374813c..4929cf4c8 100644 --- a/apps/api/src/routes/internal/agent-sessions.http.ts +++ b/apps/api/src/routes/internal/agent-sessions.http.ts @@ -1,5 +1,6 @@ import { HttpApiBuilder } from "effect/unstable/httpapi" import { + AgentSessionDetailResponse, AgentSessionsFacetsResponse, AgentSessionsListResponse, AgentTracesListResponse, @@ -10,6 +11,7 @@ import { import { Effect, Schema } from "effect" import { agentSessionsFacets, + getAgentSessionDetail, listAgentSessions, listAgentTraces, type AgentSessionsFilterInput, @@ -97,6 +99,37 @@ export const HttpAgentSessionsInternalLive = HttpApiBuilder.group( }) }), ) + .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 diff --git a/packages/domain/src/http/agent-sessions.ts b/packages/domain/src/http/agent-sessions.ts index 89008393c..dfa381a8b 100644 --- a/packages/domain/src/http/agent-sessions.ts +++ b/packages/domain/src/http/agent-sessions.ts @@ -116,6 +116,88 @@ export class AgentSessionsFacetsResponse 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, @@ -137,6 +219,13 @@ export class AgentSessionsInternalApiGroup extends HttpApiGroup.make("agentSessi error: agentSessionsEndpointErrors, }), ) + .add( + HttpApiEndpoint.post("detail", "/detail", { + payload: AgentSessionDetailRequest, + success: AgentSessionDetailResponse, + error: agentSessionsEndpointErrors, + }), + ) .add( HttpApiEndpoint.post("facets", "/facets", { payload: AgentSessionsFacetsRequest,