From ea0354aca20ac4cc21017d499a5c68fc009355bf Mon Sep 17 00:00:00 2001 From: lyon Date: Sun, 21 Jun 2026 21:55:52 +0800 Subject: [PATCH] instrument workbench runtime otel spans --- internal/cloud/server-workbench-http.ts | 6 +- internal/cloud/workbench-runtime-client.ts | 20 +- internal/workbenchruntime/otel.go | 321 +++++++++++++++++++++ internal/workbenchruntime/service.go | 254 +++++++++------- 4 files changed, 486 insertions(+), 115 deletions(-) create mode 100644 internal/workbenchruntime/otel.go diff --git a/internal/cloud/server-workbench-http.ts b/internal/cloud/server-workbench-http.ts index 525b0f2a..9724829b 100644 --- a/internal/cloud/server-workbench-http.ts +++ b/internal/cloud/server-workbench-http.ts @@ -35,7 +35,7 @@ export async function handleWorkbenchReadModelHttp(request, response, url, optio if (url.pathname === "/v1/workbench/sessions") { if (request.method !== "GET") return methodNotAllowed(response, "GET"); - await (perf ? perf.measure("workbench_session_list", () => handleWorkbenchSessionList(response, url, options, auth.actor)) : handleWorkbenchSessionList(response, url, options, auth.actor)); + await (perf ? perf.measure("workbench_session_list", () => handleWorkbenchSessionList(request, response, url, options, auth.actor)) : handleWorkbenchSessionList(request, response, url, options, auth.actor)); return; } @@ -404,13 +404,13 @@ async function authenticateWorkbenchRead(request, response, options) { return auth; } -async function handleWorkbenchSessionList(response, url, options, actor) { +async function handleWorkbenchSessionList(request, response, url, options, actor) { if (url.searchParams.has("projectId") || url.searchParams.has("workspaceId")) return sendJson(response, 400, workbenchError("workbench_authority_removed", "Workbench session list is keyed by sessionId only.")); const limit = boundedSessionListLimit(url.searchParams.get("limit")); const offset = cursorOffset(url.searchParams.get("cursor") ?? url.searchParams.get("after")); const includeSessionId = safeSessionId(url.searchParams.get("includeSessionId")); const includeRouteId = includeSessionId ?? safeConversationId(url.searchParams.get("includeSessionId")); - const runtime = options.workbenchRuntime ?? createWorkbenchRuntimeClient({ env: options.env ?? process.env, fetch: options.fetch, logger: options.logger }); + const runtime = options.workbenchRuntime ?? createWorkbenchRuntimeClient({ env: options.env ?? process.env, fetch: options.fetch, logger: options.logger, traceparent: request?.hwlabHttpRequestContext?.traceparent }); if (typeof runtime?.listWorkbenchSessions !== "function") { throw runtimeDependencyError("workbench_runtime_unconfigured", "Workbench runtime service is required for session list.", false); } diff --git a/internal/cloud/workbench-runtime-client.ts b/internal/cloud/workbench-runtime-client.ts index 5f9b4f56..f54b13f9 100644 --- a/internal/cloud/workbench-runtime-client.ts +++ b/internal/cloud/workbench-runtime-client.ts @@ -11,16 +11,17 @@ export function createWorkbenchRuntimeClient(options = {}) { if (!baseUrl) return null; const timeoutMs = boundedInteger(env.HWLAB_WORKBENCH_RUNTIME_TIMEOUT_MS, DEFAULT_WORKBENCH_RUNTIME_TIMEOUT_MS, 1000, 60000); const fetchFn = options.fetch ?? globalThis.fetch; + const traceparent = normalizeTraceparent(options.traceparent ?? options.traceParent); return { serviceId: "hwlab-workbench-runtime", - queryWorkbenchFacts: (params = {}) => postRuntime({ fetchFn, baseUrl, timeoutMs, path: "/v1/workbench-runtime/facts", body: params }), - queryAgentTraceEvents: (params = {}) => postRuntime({ fetchFn, baseUrl, timeoutMs, path: "/v1/workbench-runtime/trace-events", body: params }), - listWorkbenchSessions: (params = {}) => postRuntime({ fetchFn, baseUrl, timeoutMs, path: "/v1/workbench-runtime/sessions", body: params }), - readWorkbenchProjectionOutbox: (params = {}) => postRuntime({ fetchFn, baseUrl, timeoutMs, path: "/v1/workbench-runtime/projection-outbox", body: params }).then((result) => Array.isArray(result?.rows) ? result.rows : []) + queryWorkbenchFacts: (params = {}) => postRuntime({ fetchFn, baseUrl, timeoutMs, traceparent, path: "/v1/workbench-runtime/facts", body: params }), + queryAgentTraceEvents: (params = {}) => postRuntime({ fetchFn, baseUrl, timeoutMs, traceparent, path: "/v1/workbench-runtime/trace-events", body: params }), + listWorkbenchSessions: (params = {}) => postRuntime({ fetchFn, baseUrl, timeoutMs, traceparent, path: "/v1/workbench-runtime/sessions", body: params }), + readWorkbenchProjectionOutbox: (params = {}) => postRuntime({ fetchFn, baseUrl, timeoutMs, traceparent, path: "/v1/workbench-runtime/projection-outbox", body: params }).then((result) => Array.isArray(result?.rows) ? result.rows : []) }; } -async function postRuntime({ fetchFn, baseUrl, timeoutMs, path, body }) { +async function postRuntime({ fetchFn, baseUrl, timeoutMs, traceparent, path, body }) { if (typeof fetchFn !== "function") { throw runtimeDependencyError({ code: "workbench_runtime_fetch_unavailable", message: "fetch is unavailable for workbench runtime dependency", retryable: false }); } @@ -28,9 +29,11 @@ async function postRuntime({ fetchFn, baseUrl, timeoutMs, path, body }) { const timer = setTimeout(() => controller.abort(), timeoutMs); let response; try { + const headers = { "content-type": "application/json", accept: "application/json" }; + if (traceparent) headers.traceparent = traceparent; response = await fetchFn(new URL(path, baseUrl), { method: "POST", - headers: { "content-type": "application/json", accept: "application/json" }, + headers, body: JSON.stringify(body ?? {}), signal: controller.signal }); @@ -57,6 +60,11 @@ async function responseJson(response) { try { return await response.json(); } catch { return null; } } +function normalizeTraceparent(value) { + const text = String(value ?? "").trim().toLowerCase(); + return /^00-[0-9a-f]{32}-[0-9a-f]{16}-[0-9a-f]{2}$/u.test(text) ? text : ""; +} + function runtimeDependencyError({ code, message, retryable, status = null, payload = null, cause = null }) { const error = new Error(message || code || "workbench runtime dependency failed"); error.name = "WorkbenchRuntimeDependencyError"; diff --git a/internal/workbenchruntime/otel.go b/internal/workbenchruntime/otel.go new file mode 100644 index 00000000..d056a966 --- /dev/null +++ b/internal/workbenchruntime/otel.go @@ -0,0 +1,321 @@ +package workbenchruntime + +import ( + "bytes" + "context" + crand "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "io" + "net/http" + "os" + "strconv" + "strings" + "time" +) + +const ( + otelTraceIDZero = "00000000000000000000000000000000" + otelSpanIDZero = "0000000000000000" + otelSpanKindInternal = 1 + otelSpanKindServer = 2 + otelExportTimeout = 1500 * time.Millisecond +) + +type otelTraceContext struct { + TraceID string + ParentSpanID string + ServerSpanID string +} + +type otelTraceContextKey struct{} + +type queryStageError struct { + stage string + err error +} + +func (e queryStageError) Error() string { return e.err.Error() } +func (e queryStageError) Unwrap() error { return e.err } + +func wrapQueryStage(stage string, err error) error { + if err == nil || strings.TrimSpace(stage) == "" { + return err + } + if queryErrorStage(err) != "" { + return err + } + return queryStageError{stage: stage, err: err} +} + +func queryErrorStage(err error) string { + var stageErr queryStageError + if errors.As(err, &stageErr) { + return stageErr.stage + } + return "" +} + +func newOtelTraceContext(traceparent string) otelTraceContext { + traceID, parentSpanID, ok := parseOtelTraceparent(traceparent) + if !ok { + traceID = newOtelTraceID() + parentSpanID = "" + } + return otelTraceContext{TraceID: traceID, ParentSpanID: parentSpanID, ServerSpanID: newOtelSpanID()} +} + +func (ctx otelTraceContext) traceparent() string { + if !validOtelHex(ctx.TraceID, 32, otelTraceIDZero) || !validOtelHex(ctx.ServerSpanID, 16, otelSpanIDZero) { + return "" + } + return "00-" + ctx.TraceID + "-" + ctx.ServerSpanID + "-01" +} + +func newOtelTraceID() string { return randomOtelHex(16, otelTraceIDZero) } +func newOtelSpanID() string { return randomOtelHex(8, otelSpanIDZero) } + +func randomOtelHex(byteCount int, zero string) string { + buf := make([]byte, byteCount) + if _, err := crand.Read(buf); err != nil { + return strings.TrimSuffix(zero, "0") + "1" + } + value := hex.EncodeToString(buf) + if value == zero { + return strings.TrimSuffix(zero, "0") + "1" + } + return value +} + +func parseOtelTraceparent(value string) (string, string, bool) { + parts := strings.Split(strings.ToLower(strings.TrimSpace(value)), "-") + if len(parts) != 4 { + return "", "", false + } + traceID := parts[1] + spanID := parts[2] + if !validOtelHex(traceID, 32, otelTraceIDZero) || !validOtelHex(spanID, 16, otelSpanIDZero) { + return "", "", false + } + return traceID, spanID, true +} + +func validOtelHex(value string, length int, zero string) bool { + if len(value) != length || value == zero { + return false + } + for _, ch := range value { + if (ch < '0' || ch > '9') && (ch < 'a' || ch > 'f') { + return false + } + } + return true +} + +func (s *Server) otelHTTPHandler(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + startedAt := time.Now().UTC() + trace := newOtelTraceContext(r.Header.Get("traceparent")) + if traceparent := trace.traceparent(); traceparent != "" { + w.Header().Set("traceparent", traceparent) + w.Header().Set("x-hwlab-otel-trace-id", trace.TraceID) + } + ctx := context.WithValue(r.Context(), otelTraceContextKey{}, trace) + writer := &otelStatusWriter{ResponseWriter: w, statusCode: http.StatusOK} + next.ServeHTTP(writer, r.WithContext(ctx)) + statusCode := writer.statusCode + errorCode := "" + if statusCode >= 500 { + errorCode = "http_" + strconv.Itoa(statusCode) + } + s.emitOtelSpanAsync("workbench-runtime.http "+r.Method+" "+r.URL.Path, trace, trace.ServerSpanID, trace.ParentSpanID, otelSpanKindServer, startedAt, map[string]any{ + "http.request.method": r.Method, + "http.route": r.URL.Path, + "http.response.status_code": int64(statusCode), + "hwlab.runtime.stage": "http.server.request", + "valuesRedacted": true, + }, statusCode, errorCode) + }) +} + +type otelStatusWriter struct { + http.ResponseWriter + statusCode int +} + +func (w *otelStatusWriter) WriteHeader(statusCode int) { + w.statusCode = statusCode + w.ResponseWriter.WriteHeader(statusCode) +} + +func (w *otelStatusWriter) Write(payload []byte) (int, error) { + if w.statusCode == 0 { + w.statusCode = http.StatusOK + } + return w.ResponseWriter.Write(payload) +} + +func otelTraceFromContext(ctx context.Context) (otelTraceContext, bool) { + trace, ok := ctx.Value(otelTraceContextKey{}).(otelTraceContext) + if !ok || !validOtelHex(trace.TraceID, 32, otelTraceIDZero) { + return otelTraceContext{}, false + } + return trace, true +} + +func (s *Server) withOtelInternalSpan(ctx context.Context, name string, attrs map[string]any, fn func(context.Context) error) error { + trace, ok := otelTraceFromContext(ctx) + if !ok { + return fn(ctx) + } + startedAt := time.Now().UTC() + spanID := newOtelSpanID() + err := fn(ctx) + errorCode := "" + if err != nil { + errorCode = otelFirst(queryErrorStage(err), sqlErrorCode(err), "query_failed") + } + attributes := map[string]any{ + "hwlab.runtime.stage": name, + "valuesRedacted": true, + } + for key, value := range attrs { + if value != nil { + attributes[key] = value + } + } + s.emitOtelSpanAsync(name, trace, spanID, trace.ServerSpanID, otelSpanKindInternal, startedAt, attributes, 0, errorCode) + return err +} + +func (s *Server) emitOtelSpanAsync(name string, trace otelTraceContext, spanID string, parentSpanID string, kind int, startedAt time.Time, attrs map[string]any, httpStatus int, errorCode string) { + go s.emitOtelSpan(context.Background(), name, trace, spanID, parentSpanID, kind, startedAt, attrs, httpStatus, errorCode) +} + +func (s *Server) emitOtelSpan(ctx context.Context, name string, trace otelTraceContext, spanID string, parentSpanID string, kind int, startedAt time.Time, attrs map[string]any, httpStatus int, errorCode string) { + endpoint := resolveOtelTracesEndpoint() + if endpoint == "" || !validOtelHex(trace.TraceID, 32, otelTraceIDZero) || !validOtelHex(spanID, 16, otelSpanIDZero) { + return + } + endedAt := time.Now().UTC() + if startedAt.IsZero() { + startedAt = endedAt + } + attributes := map[string]any{"otel.trace_id": trace.TraceID} + for key, value := range attrs { + if value != nil { + attributes[key] = value + } + } + if errorCode != "" { + attributes["error.code"] = errorCode + } + statusCode := 1 + status := map[string]any{"code": statusCode} + if errorCode != "" || httpStatus >= 500 { + statusCode = 2 + status = map[string]any{"code": statusCode, "message": otelFirst(errorCode, "request_failed")} + } + span := map[string]any{ + "traceId": trace.TraceID, + "spanId": spanID, + "name": name, + "kind": kind, + "startTimeUnixNano": strconv.FormatInt(startedAt.UTC().UnixNano(), 10), + "endTimeUnixNano": strconv.FormatInt(endedAt.UnixNano(), 10), + "attributes": otelAttributes(attributes), + "status": status, + } + if validOtelHex(parentSpanID, 16, otelSpanIDZero) { + span["parentSpanId"] = parentSpanID + } + body := map[string]any{ + "resourceSpans": []any{map[string]any{ + "resource": map[string]any{"attributes": otelAttributes(s.otelResourceAttributes())}, + "scopeSpans": []any{map[string]any{ + "scope": map[string]any{"name": "hwlab.workbench-runtime", "version": "1"}, + "spans": []any{span}, + }}, + }}, + } + payload, err := json.Marshal(body) + if err != nil { + return + } + requestCtx, cancel := context.WithTimeout(ctx, otelExportTimeout) + defer cancel() + request, err := http.NewRequestWithContext(requestCtx, http.MethodPost, endpoint, bytes.NewReader(payload)) + if err != nil { + return + } + request.Header.Set("Content-Type", "application/json") + client := http.Client{Timeout: otelExportTimeout} + response, err := client.Do(request) + if err != nil { + return + } + _, _ = io.Copy(io.Discard, response.Body) + _ = response.Body.Close() +} + +func resolveOtelTracesEndpoint() string { + if endpoint := otelFirst(os.Getenv("HWLAB_OTEL_EXPORTER_OTLP_TRACES_ENDPOINT"), os.Getenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT")); endpoint != "" { + return strings.TrimRight(endpoint, "/") + } + if endpoint := otelFirst(os.Getenv("HWLAB_OTEL_EXPORTER_OTLP_ENDPOINT"), os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT")); endpoint != "" { + return strings.TrimRight(endpoint, "/") + "/v1/traces" + } + return "" +} + +func (s *Server) otelResourceAttributes() map[string]any { + return map[string]any{ + "service.name": otelFirst(os.Getenv("OTEL_SERVICE_NAME"), serviceID), + "deployment.environment": otelFirst(os.Getenv("HWLAB_ENVIRONMENT"), os.Getenv("HWLAB_RUNTIME_LANE"), "unknown"), + "hwlab.lane": otelFirst(os.Getenv("HWLAB_RUNTIME_LANE"), os.Getenv("HWLAB_GITOPS_PROFILE"), "unknown"), + "k8s.namespace.name": otelFirst(os.Getenv("POD_NAMESPACE"), os.Getenv("HWLAB_NAMESPACE"), "unknown"), + "git.commit": otelFirst(os.Getenv("HWLAB_COMMIT_ID"), os.Getenv("HWLAB_GITOPS_SOURCE_COMMIT"), os.Getenv("HWLAB_REVISION"), "unknown"), + } +} + +func otelAttributes(values map[string]any) []map[string]any { + attrs := make([]map[string]any, 0, len(values)) + for key, value := range values { + if key == "" || value == nil { + continue + } + attrs = append(attrs, map[string]any{"key": key, "value": otelAnyValue(value)}) + } + return attrs +} + +func otelAnyValue(value any) map[string]any { + switch typed := value.(type) { + case bool: + return map[string]any{"boolValue": typed} + case int: + return map[string]any{"intValue": strconv.FormatInt(int64(typed), 10)} + case int64: + return map[string]any{"intValue": strconv.FormatInt(typed, 10)} + case float64: + return map[string]any{"doubleValue": typed} + case string: + return map[string]any{"stringValue": typed} + default: + payload, err := json.Marshal(typed) + if err != nil { + return map[string]any{"stringValue": "{}"} + } + return map[string]any{"stringValue": string(payload)} + } +} + +func otelFirst(values ...string) string { + for _, value := range values { + if text := strings.TrimSpace(value); text != "" { + return text + } + } + return "" +} diff --git a/internal/workbenchruntime/service.go b/internal/workbenchruntime/service.go index 1c809248..9a17e5f7 100644 --- a/internal/workbenchruntime/service.go +++ b/internal/workbenchruntime/service.go @@ -121,7 +121,7 @@ func Run(ctx context.Context) error { db.SetMaxIdleConns(config.PoolMax) } server := NewServer(config, db) - httpServer := &http.Server{Addr: config.Addr, Handler: server.mux, ReadHeaderTimeout: 5 * time.Second} + httpServer := &http.Server{Addr: config.Addr, Handler: server.otelHTTPHandler(server.mux), ReadHeaderTimeout: 5 * time.Second} errCh := make(chan error, 1) go func() { log.Printf("%s listening addr=%s", serviceID, config.Addr) @@ -440,6 +440,8 @@ func (s *Server) queryFacts(ctx context.Context, query factQuery) (map[string][] func (s *Server) queryFactRows(ctx context.Context, table string, jsonColumn string, query factQuery, fields []fieldColumn) ([]any, error) { clauses, args := factClauses(query, fields) limit := boundedLimit(query.Limit, 0, 1000) + family := factFamilyForTable(table) + stage := "workbench_runtime.db." + family if table == "workbench_trace_events" { if query.AfterProjectedSeq > 0 { args = append(args, query.AfterProjectedSeq) @@ -450,92 +452,115 @@ func (s *Server) queryFactRows(ctx context.Context, table string, jsonColumn str args = append(args, limit) sqlText += fmt.Sprintf(" LIMIT $%d", len(args)) } - return s.queryJSONColumn(ctx, sqlText, args) + return s.queryJSONColumn(ctx, stage, table, sqlText, args) } orderDirection := "ASC" - if factOrder(query, factFamilyForTable(table)) == "updated_desc" { + if factOrder(query, family) == "updated_desc" { orderDirection = "DESC" } if table == "workbench_sessions" && query.SessionProjection == "summary" { + stage = "workbench_runtime.db.sessions_summary" sqlText := fmt.Sprintf("SELECT %s FROM workbench_sessions%s ORDER BY updated_at %s", sessionSummarySelectClause(), whereSQL(clauses), orderDirection) if limit > 0 { args = append(args, limit) sqlText += fmt.Sprintf(" LIMIT $%d", len(args)) } - return s.querySessionSummaries(ctx, sqlText, args) + return s.querySessionSummaries(ctx, stage, sqlText, args) } sqlText := fmt.Sprintf("SELECT %s FROM %s%s ORDER BY updated_at %s", jsonColumn, table, whereSQL(clauses), orderDirection) if limit > 0 { args = append(args, limit) sqlText += fmt.Sprintf(" LIMIT $%d", len(args)) } - return s.queryJSONColumn(ctx, sqlText, args) + return s.queryJSONColumn(ctx, stage, table, sqlText, args) } -func (s *Server) queryJSONColumn(ctx context.Context, sqlText string, args []any) ([]any, error) { - rows, err := s.db.QueryContext(ctx, sqlText, args...) - if err != nil { - return nil, err - } - defer rows.Close() +func (s *Server) queryJSONColumn(ctx context.Context, stage string, table string, sqlText string, args []any) ([]any, error) { result := []any{} - for rows.Next() { - var raw sql.NullString - if err := rows.Scan(&raw); err != nil { - return nil, err + err := s.withOtelInternalSpan(ctx, stage, map[string]any{ + "db.system": "postgresql", + "db.operation.name": "SELECT", + "db.sql.table": table, + "db.query.arg_count": int64(len(args)), + }, func(spanCtx context.Context) error { + rows, err := s.db.QueryContext(spanCtx, sqlText, args...) + if err != nil { + return err } - var item any - if raw.Valid && strings.TrimSpace(raw.String) != "" && json.Unmarshal([]byte(raw.String), &item) == nil && item != nil { + defer rows.Close() + for rows.Next() { + var raw sql.NullString + if err := rows.Scan(&raw); err != nil { + return err + } + var item any + if raw.Valid && strings.TrimSpace(raw.String) != "" && json.Unmarshal([]byte(raw.String), &item) == nil && item != nil { + result = append(result, item) + } + } + return rows.Err() + }) + if err != nil { + return nil, wrapQueryStage(stage, err) + } + return result, nil +} + +func (s *Server) querySessionSummaries(ctx context.Context, stage string, sqlText string, args []any) ([]any, error) { + result := []any{} + err := s.withOtelInternalSpan(ctx, stage, map[string]any{ + "db.system": "postgresql", + "db.operation.name": "SELECT", + "db.sql.table": "workbench_sessions", + "db.query.arg_count": int64(len(args)), + }, func(spanCtx context.Context) error { + rows, err := s.db.QueryContext(spanCtx, sqlText, args...) + if err != nil { + return err + } + defer rows.Close() + for rows.Next() { + var sessionID, ownerUserID, projectID, conversationID, threadID, status, lastTraceID, sourceEventID, providerProfile sql.NullString + var projectedSeq, sourceSeq sql.NullInt64 + var terminal, sealed sql.NullBool + var createdAt, updatedAt sql.NullTime + if err := rows.Scan(&sessionID, &ownerUserID, &projectID, &conversationID, &threadID, &status, &lastTraceID, &projectedSeq, &sourceSeq, &sourceEventID, &terminal, &sealed, &createdAt, &updatedAt, &providerProfile); err != nil { + return err + } + if !sessionID.Valid || strings.TrimSpace(sessionID.String) == "" { + continue + } + item := map[string]any{ + "id": sessionID.String, + "sessionId": sessionID.String, + "ownerUserId": nullableString(ownerUserID), + "projectId": nullableString(projectID), + "conversationId": nullableString(conversationID), + "threadId": nullableString(threadID), + "status": first(nullableStringValue(status), "unknown"), + "lastTraceId": nullableString(lastTraceID), + "projectedSeq": nullableInt(projectedSeq), + "sourceSeq": nullableInt(sourceSeq), + "sourceEventId": nullableString(sourceEventID), + "terminal": nullableBool(terminal), + "sealed": nullableBool(sealed), + "createdAt": nullableTime(createdAt), + "updatedAt": first(nullableTime(updatedAt), nullableTime(createdAt)), + "valuesPrinted": false, + "valuesRedacted": true, + } + if providerProfile.Valid && strings.TrimSpace(providerProfile.String) != "" { + item["providerProfile"] = providerProfile.String + item["sessionJson"] = map[string]any{"providerProfile": providerProfile.String, "valuesRedacted": true, "secretMaterialStored": false} + } result = append(result, item) } - } - return result, rows.Err() -} - -func (s *Server) querySessionSummaries(ctx context.Context, sqlText string, args []any) ([]any, error) { - rows, err := s.db.QueryContext(ctx, sqlText, args...) + return rows.Err() + }) if err != nil { - return nil, err + return nil, wrapQueryStage(stage, err) } - defer rows.Close() - result := []any{} - for rows.Next() { - var sessionID, ownerUserID, projectID, conversationID, threadID, status, lastTraceID, sourceEventID, providerProfile sql.NullString - var projectedSeq, sourceSeq sql.NullInt64 - var terminal, sealed sql.NullBool - var createdAt, updatedAt sql.NullTime - if err := rows.Scan(&sessionID, &ownerUserID, &projectID, &conversationID, &threadID, &status, &lastTraceID, &projectedSeq, &sourceSeq, &sourceEventID, &terminal, &sealed, &createdAt, &updatedAt, &providerProfile); err != nil { - return nil, err - } - if !sessionID.Valid || strings.TrimSpace(sessionID.String) == "" { - continue - } - item := map[string]any{ - "id": sessionID.String, - "sessionId": sessionID.String, - "ownerUserId": nullableString(ownerUserID), - "projectId": nullableString(projectID), - "conversationId": nullableString(conversationID), - "threadId": nullableString(threadID), - "status": first(nullableStringValue(status), "unknown"), - "lastTraceId": nullableString(lastTraceID), - "projectedSeq": nullableInt(projectedSeq), - "sourceSeq": nullableInt(sourceSeq), - "sourceEventId": nullableString(sourceEventID), - "terminal": nullableBool(terminal), - "sealed": nullableBool(sealed), - "createdAt": nullableTime(createdAt), - "updatedAt": first(nullableTime(updatedAt), nullableTime(createdAt)), - "valuesPrinted": false, - "valuesRedacted": true, - } - if providerProfile.Valid && strings.TrimSpace(providerProfile.String) != "" { - item["providerProfile"] = providerProfile.String - item["sessionJson"] = map[string]any{"providerProfile": providerProfile.String, "valuesRedacted": true, "secretMaterialStored": false} - } - result = append(result, item) - } - return result, rows.Err() + return result, nil } func (s *Server) queryAgentTraceEvents(ctx context.Context, query traceEventsQuery) ([]any, error) { @@ -550,7 +575,7 @@ func (s *Server) queryAgentTraceEvents(ctx context.Context, query traceEventsQue args = append(args, limit) sqlText += fmt.Sprintf(" LIMIT $%d", len(args)) } - return s.queryJSONColumn(ctx, sqlText, args) + return s.queryJSONColumn(ctx, "workbench_runtime.db.agent_trace_events", "agent_trace_events", sqlText, args) } func (s *Server) readProjectionOutbox(ctx context.Context, query outboxQuery) ([]any, error) { @@ -561,43 +586,55 @@ func (s *Server) readProjectionOutbox(ctx context.Context, query outboxQuery) ([ addTextClause(&clauses, &args, "session_id", query.SessionID) args = append(args, limit) sqlText := fmt.Sprintf("SELECT outbox_seq, trace_id, session_id, turn_id, message_id, projected_seq, source_seq, source_event_id, commit_type, terminal, sealed, payload_json, created_at FROM workbench_projection_outbox WHERE %s ORDER BY outbox_seq ASC LIMIT $%d", strings.Join(clauses, " AND "), len(args)) - rows, err := s.db.QueryContext(ctx, sqlText, args...) - if err != nil { - return nil, err - } - defer rows.Close() + stage := "workbench_runtime.db.projection_outbox" result := []any{} - for rows.Next() { - var outboxSeq, projectedSeq, sourceSeq sql.NullInt64 - var traceID, sessionID, turnID, messageID, sourceEventID, commitType sql.NullString - var terminal, sealed sql.NullBool - var raw sql.NullString - var createdAt sql.NullTime - if err := rows.Scan(&outboxSeq, &traceID, &sessionID, &turnID, &messageID, &projectedSeq, &sourceSeq, &sourceEventID, &commitType, &terminal, &sealed, &raw, &createdAt); err != nil { - return nil, err + err := s.withOtelInternalSpan(ctx, stage, map[string]any{ + "db.system": "postgresql", + "db.operation.name": "SELECT", + "db.sql.table": "workbench_projection_outbox", + "db.query.arg_count": int64(len(args)), + }, func(spanCtx context.Context) error { + rows, err := s.db.QueryContext(spanCtx, sqlText, args...) + if err != nil { + return err } - var payload any - if raw.Valid && strings.TrimSpace(raw.String) != "" { - _ = json.Unmarshal([]byte(raw.String), &payload) + defer rows.Close() + for rows.Next() { + var outboxSeq, projectedSeq, sourceSeq sql.NullInt64 + var traceID, sessionID, turnID, messageID, sourceEventID, commitType sql.NullString + var terminal, sealed sql.NullBool + var raw sql.NullString + var createdAt sql.NullTime + if err := rows.Scan(&outboxSeq, &traceID, &sessionID, &turnID, &messageID, &projectedSeq, &sourceSeq, &sourceEventID, &commitType, &terminal, &sealed, &raw, &createdAt); err != nil { + return err + } + var payload any + if raw.Valid && strings.TrimSpace(raw.String) != "" { + _ = json.Unmarshal([]byte(raw.String), &payload) + } + result = append(result, map[string]any{ + "outboxSeq": nullableInt(outboxSeq), + "traceId": nullableString(traceID), + "sessionId": nullableString(sessionID), + "turnId": nullableString(turnID), + "messageId": nullableString(messageID), + "projectedSeq": nullableInt(projectedSeq), + "sourceSeq": nullableInt(sourceSeq), + "sourceEventId": nullableString(sourceEventID), + "commitType": nullableString(commitType), + "terminal": nullableBool(terminal), + "sealed": nullableBool(sealed), + "payload": payload, + "createdAt": nullableTime(createdAt), + "valuesPrinted": false, + }) } - result = append(result, map[string]any{ - "outboxSeq": nullableInt(outboxSeq), - "traceId": nullableString(traceID), - "sessionId": nullableString(sessionID), - "turnId": nullableString(turnID), - "messageId": nullableString(messageID), - "projectedSeq": nullableInt(projectedSeq), - "sourceSeq": nullableInt(sourceSeq), - "sourceEventId": nullableString(sourceEventID), - "commitType": nullableString(commitType), - "terminal": nullableBool(terminal), - "sealed": nullableBool(sealed), - "payload": payload, - "createdAt": nullableTime(createdAt), - "valuesPrinted": false, - }) + return rows.Err() + }) + if err != nil { + return nil, wrapQueryStage(stage, err) } - return result, rows.Err() + return result, nil } func factClauses(query factQuery, fields []fieldColumn) ([]string, []any) { @@ -1198,16 +1235,21 @@ func decodeJSON(w http.ResponseWriter, r *http.Request, target any) bool { func writeQueryError(w http.ResponseWriter, err error) { log.Printf("%s query failed: %v", serviceID, err) + errorPayload := map[string]any{ + "code": "workbench_runtime_query_failed", + "message": "workbench runtime query failed", + "retryable": true, + "transient": true, + "causeCode": sqlErrorCode(err), + "valuesRedacted": true, + } + if stage := queryErrorStage(err); stage != "" { + errorPayload["stage"] = stage + errorPayload["causeStage"] = stage + } writeJSON(w, http.StatusServiceUnavailable, map[string]any{ - "ok": false, - "error": map[string]any{ - "code": "workbench_runtime_query_failed", - "message": "workbench runtime query failed", - "retryable": true, - "transient": true, - "causeCode": sqlErrorCode(err), - "valuesRedacted": true, - }, + "ok": false, + "error": errorPayload, "valuesRedacted": true, }) }