feat(observability): agent sessions read path — queries, API, scratchpad UI

apps/api/src/routes/internal/agent-sessions.http.ts

@@ -0,0 +1,155 @@
+import { HttpApiBuilder } from "effect/unstable/httpapi"
+import {
+	AgentSessionDetailResponse,
+	AgentSessionsFacetsResponse,
+	AgentSessionsListResponse,
+	AgentTracesListResponse,
+	CurrentTenant,
+	MapleInternalApi,
+	TraceId,
+} from "@maple/domain/http"
+import { Effect, Schema } from "effect"
+import {
+	agentSessionsFacets,
+	getAgentSessionDetail,
+	listAgentSessions,
+	listAgentTraces,
+	type AgentSessionsFilterInput,
+} from "@maple/query-engine/observability"
+import { provideWarehouseExecutorFromTenant } from "@/services/warehouse/WarehouseQueryService"
+
+const decodeTraceId = Schema.decodeSync(TraceId)
+
+/**
+ * Agent Sessions dashboard routes — thin adapters over the
+ * `@maple/query-engine/observability` read functions, which are the durable
+ * interface (the future MCP tools call those functions, not these routes).
+ * Internal tier: the response shapes follow the UI.
+ */
+export const HttpAgentSessionsInternalLive = HttpApiBuilder.group(
+	MapleInternalApi,
+	"agentSessionsInternal",
+	(handlers) =>
+		Effect.gen(function* () {
+			const filterInput = (payload: {
+				startTime: string
+				endTime: string
+				vendors?: readonly string[] | undefined
+				serviceNames?: readonly string[] | undefined
+				hasErrors?: boolean | undefined
+			}): AgentSessionsFilterInput => ({
+				startTime: payload.startTime,
+				endTime: payload.endTime,
+				vendors: payload.vendors,
+				serviceNames: payload.serviceNames,
+				hasErrors: payload.hasErrors,
+			})
+
+			return handlers
+				.handle("list", ({ payload }) =>
+					Effect.gen(function* () {
+						const tenant = yield* CurrentTenant.Context
+						yield* Effect.annotateCurrentSpan({ orgId: tenant.orgId })
+						const rows = yield* listAgentSessions({
+							...filterInput(payload),
+							limit: payload.limit,
+							offset: payload.offset,
+						}).pipe(provideWarehouseExecutorFromTenant(tenant))
+						// Integer aggregates can arrive as JSON strings (quoted 64-bit wire on
+						// gateway/readonly clusters); these queries declare no rowSchema, so
+						// coerce before the Schema.Number response fields validate.
+						return new AgentSessionsListResponse({
+							data: rows.map((row) => ({
+								sessionKeyHash: row.sessionKeyHash,
+								startTime: row.startTime,
+								endTime: row.endTime,
+								durationMs: Number(row.durationMs),
+								traceCount: Number(row.traceCount),
+								keyedSpanCount: Number(row.keyedSpanCount),
+								errorCount: Number(row.errorCount),
+								vendors: row.vendors,
+								serviceNames: row.serviceNames,
+							})),
+						})
+					}),
+				)
+				.handle("traces", ({ payload }) =>
+					Effect.gen(function* () {
+						const tenant = yield* CurrentTenant.Context
+						yield* Effect.annotateCurrentSpan({ orgId: tenant.orgId })
+						const rows = yield* listAgentTraces({
+							...filterInput(payload),
+							limit: payload.limit,
+							offset: payload.offset,
+						}).pipe(provideWarehouseExecutorFromTenant(tenant))
+						return new AgentTracesListResponse({
+							data: rows.map((row) => ({
+								traceId: decodeTraceId(row.traceId),
+								startTime: row.startTime,
+								endTime: row.endTime,
+								durationMs: Number(row.durationMs),
+								aiSpanCount: Number(row.aiSpanCount),
+								errorCount: Number(row.errorCount),
+								vendors: row.vendors,
+								serviceNames: row.serviceNames,
+								firstSpanName: row.firstSpanName,
+								bestSessionKeyState: Number(row.bestSessionKeyState),
+								sessionKeyHash: row.sessionKeyHash,
+							})),
+						})
+					}),
+				)
+				.handle("detail", ({ payload }) =>
+					Effect.gen(function* () {
+						const tenant = yield* CurrentTenant.Context
+						yield* Effect.annotateCurrentSpan({ orgId: tenant.orgId })
+						const detail = yield* getAgentSessionDetail({
+							sessionKeyHash: payload.sessionKeyHash,
+							startTime: payload.startTime,
+							endTime: payload.endTime,
+						}).pipe(provideWarehouseExecutorFromTenant(tenant))
+						// The read function already normalizes every numeric through the
+						// integration layer (real JS numbers, not wire strings); only the
+						// TraceId brand needs decoding at this boundary.
+						return new AgentSessionDetailResponse({
+							session:
+								detail === null
+									? null
+									: {
+											...detail,
+											traces: detail.traces.map((trace) => ({
+												traceId: decodeTraceId(trace.traceId),
+												startTime: trace.startTime,
+												durationMs: trace.durationMs,
+												errorCount: trace.errorCount,
+												spans: trace.spans.map(
+													({ traceId: _traceId, ...span }) => span,
+												),
+											})),
+										},
+						})
+					}),
+				)
+				.handle("facets", ({ payload }) =>
+					Effect.gen(function* () {
+						const tenant = yield* CurrentTenant.Context
+						yield* Effect.annotateCurrentSpan({ orgId: tenant.orgId })
+						const rows = yield* agentSessionsFacets({
+							...filterInput(payload),
+							tab: payload.tab,
+						}).pipe(provideWarehouseExecutorFromTenant(tenant))
+						const pick = (facetType: string) =>
+							rows
+								.filter((row) => row.facetType === facetType)
+								.map((row) => ({ name: row.name, count: Number(row.count) }))
+						return new AgentSessionsFacetsResponse({
+							vendors: pick("vendor"),
+							services: pick("service"),
+							// The error branch emits a single row, and none at all when the
+							// filtered population has no errors.
+							errorCount: Number(rows.find((row) => row.facetType === "error")?.count ?? 0),
+						})
+					}),
+				)
+		}),
+)

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),
 	),

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<typeof ListAgentSessionsInput>
+
+const AgentSessionsFacetsInput = Schema.Struct({
+	...AgentSessionsFilterInput.fields,
+	tab: Schema.Literals(["sessions", "traces"]),
+})
+export type AgentSessionsFacetsInput = Schema.Schema.Type<typeof AgentSessionsFacetsInput>
+
+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 }
+})

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 },
 	],
 }
 

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.