From 7747943832737a7e6ea168c4de118b862611dfab Mon Sep 17 00:00:00 2001 From: lyon Date: Sat, 27 Jun 2026 05:05:22 +0800 Subject: [PATCH] obs(workbench): add session read spans for #1078 --- internal/cloud/server-workbench-http.ts | 137 +++++++++++++++++++++++- 1 file changed, 136 insertions(+), 1 deletion(-) diff --git a/internal/cloud/server-workbench-http.ts b/internal/cloud/server-workbench-http.ts index 3ce62a16..703c46ba 100644 --- a/internal/cloud/server-workbench-http.ts +++ b/internal/cloud/server-workbench-http.ts @@ -722,6 +722,15 @@ async function handleWorkbenchSessionList(request, response, url, options, actor const readModel = workbenchRuntimeFactsReadModel(request, options, actor); const payload = await handleWorkbenchSessionListFromFacts(readModel, url, options, actor, { limit, offset, includeRouteId }); recordWorkbenchCacheMetric(options, url, payload, startedAt); + emitWorkbenchSessionListReadOtel(request, options, { + startTimeMs: startedAt, + statusCode: 200, + source: "facts", + limit, + offset, + includeSessionId: includeRouteId, + payload + }); return sendJson(response, 200, payload); } const payload = await runtime.listWorkbenchSessions({ @@ -732,6 +741,15 @@ async function handleWorkbenchSessionList(request, response, url, options, actor actor: { id: actor.id, role: actor.role ?? "user", valuesRedacted: true } }); recordWorkbenchCacheMetric(options, url, payload, startedAt); + emitWorkbenchSessionListReadOtel(request, options, { + startTimeMs: startedAt, + statusCode: 200, + source: "runtime", + limit, + offset, + includeSessionId: includeRouteId, + payload + }); return sendJson(response, 200, payload); } @@ -1632,6 +1650,7 @@ async function handleWorkbenchSessionDetail(request, response, options, actor, s } async function handleWorkbenchMessagePage(request, response, url, options, actor, sessionId) { + const startedAt = nowMs(); const readModel = workbenchRuntimeFactsReadModel(request, options, actor); const limit = boundedLimit(url.searchParams.get("limit")); const cursorParam = url.searchParams.get("cursor") ?? url.searchParams.get("after"); @@ -1652,7 +1671,7 @@ async function handleWorkbenchMessagePage(request, response, url, options, actor const page = messages.slice(offset, offset + limit); const nextOffset = offset + page.length; const resolvedSessionId = factSessionId(session); - sendJson(response, 200, { + const body = { ok: true, status: "succeeded", contractVersion: "workbench-message-page-v1", @@ -1665,7 +1684,18 @@ async function handleWorkbenchMessagePage(request, response, url, options, actor hasMore: nextOffset < messages.length, valuesRedacted: true, secretMaterialStored: false + }; + emitWorkbenchSessionMessagesReadOtel(request, options, { + startTimeMs: startedAt, + statusCode: 200, + sessionId: resolvedSessionId, + limit, + offset, + messages: page, + total: messages.length, + hasMore: body.hasMore }); + sendJson(response, 200, body); } async function handleWorkbenchTurnSnapshot(request, response, url, options, actor, rawTurnId) { @@ -1991,6 +2021,111 @@ function emitWorkbenchTraceEventsReadOtel(traceId, options = {}, fields = {}, er }); } +function emitWorkbenchSessionListReadOtel(request, options = {}, fields = {}, error = null) { + const payload = fields.payload && typeof fields.payload === "object" ? fields.payload : {}; + const sessions = Array.isArray(payload.sessions) ? payload.sessions : []; + const fallbackTitleCount = sessions.filter(sessionSummaryNeedsTitleFallback).length; + const traceIds = uniqueText(sessions.map((session) => session?.lastTraceId)).slice(0, 8); + const businessTraceId = (safeTraceId(traceIds[0]) ?? textValue(fields.includeSessionId)) || "workbench_session_list"; + void emitCodeAgentOtelSpan("session_list_read", businessTraceId, options.env ?? process.env, { + startTimeMs: fields.startTimeMs, + status: error ? "error" : "ok", + error, + attributes: { + "http.method": "GET", + "http.route": "/v1/workbench/sessions", + "http.status_code": Number.isInteger(Number(fields.statusCode)) ? Number(fields.statusCode) : null, + source: fields.source ?? null, + limit: nonNegativeInteger(fields.limit), + offset: nonNegativeInteger(fields.offset), + includeSessionId: fields.includeSessionId ?? null, + sessionCount: sessions.length, + fallbackTitleCount, + fallbackTitleRatio: sessions.length > 0 ? Number((fallbackTitleCount / sessions.length).toFixed(4)) : 0, + emptyPreviewCount: sessions.filter((session) => !sessionPreviewText(session)).length, + sessionIds: uniqueText(sessions.map((session) => session?.sessionId)).slice(0, 8).join(","), + traceIds: traceIds.join(","), + hasMore: payload.hasMore === true, + valuesRedacted: true + } + }); +} + +function emitWorkbenchSessionMessagesReadOtel(request, options = {}, fields = {}, error = null) { + const messages = Array.isArray(fields.messages) ? fields.messages : []; + const traceIds = uniqueText(messages.map((message) => message?.traceId)).slice(0, 8); + const businessTraceId = (safeTraceId(traceIds[0]) ?? textValue(fields.sessionId)) || "workbench_session_messages"; + const roleSequence = messages.map((message) => roleSequenceSymbol(message?.role)).join(""); + void emitCodeAgentOtelSpan("session_messages_read", businessTraceId, options.env ?? process.env, { + startTimeMs: fields.startTimeMs, + status: error ? "error" : "ok", + error, + attributes: { + "http.method": "GET", + "http.route": "/v1/workbench/sessions/:id/messages", + "http.status_code": Number.isInteger(Number(fields.statusCode)) ? Number(fields.statusCode) : null, + sessionId: fields.sessionId ?? null, + limit: nonNegativeInteger(fields.limit), + offset: nonNegativeInteger(fields.offset), + returnedMessages: messages.length, + totalMessages: nonNegativeInteger(fields.total), + hasMore: fields.hasMore === true, + roleSequencePrefix: roleSequence.slice(0, 32), + consecutiveUserPrefix: consecutiveRolePrefix(messages, "user"), + adjacentSameRoleCount: adjacentSameRoleCount(messages), + userCount: messages.filter((message) => normalizeRole(message?.role) === "user").length, + agentCount: messages.filter((message) => isAssistantLikeRole(message?.role)).length, + traceIds: traceIds.join(","), + valuesRedacted: true + } + }); +} + +function sessionSummaryNeedsTitleFallback(session) { + if (!session || typeof session !== "object") return true; + return !sessionPreviewText(session); +} + +function sessionPreviewText(session) { + return textValue( + session?.firstUserMessagePreview + ?? session?.snapshot?.firstUserMessagePreview + ?? session?.userPreview + ?? session?.title + ?? session?.name + ); +} + +function roleSequenceSymbol(role) { + const normalized = normalizeRole(role); + if (normalized === "user") return "U"; + if (normalized === "agent" || normalized === "assistant") return "A"; + if (normalized === "system") return "S"; + return "?"; +} + +function normalizeRole(role) { + return textValue(role).toLowerCase(); +} + +function consecutiveRolePrefix(messages = [], role) { + let count = 0; + for (const message of messages) { + if (normalizeRole(message?.role) !== role) break; + count += 1; + } + return count; +} + +function adjacentSameRoleCount(messages = []) { + let count = 0; + for (let index = 1; index < messages.length; index += 1) { + const previous = normalizeRole(messages[index - 1]?.role); + if (previous && previous === normalizeRole(messages[index]?.role)) count += 1; + } + return count; +} + function traceEventPageFromFacts(sourceEvents, options, metadata = {}) { const rows = factArray(sourceEvents).filter((event) => factProjectedSeq(event)).sort(compareFactTraceEventsAsc); const collision = projectedSeqCollision(rows);