Merge pull request #2220 from pikasTech/issue-1078-session-message-spans

obs(workbench): add session read spans for #1078

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