From 2de12a6ced6e40b492cf0613b033edd671bbde6c Mon Sep 17 00:00:00 2001 From: siddiqui irshad Date: Tue, 8 Sep 2026 20:34:06 +0530 Subject: [PATCH 1/2] fix(sessions): reconstruct Copilot transcripts from GenAI spans Copilot lists sessions from session-start logs, but those events do not map to transcript roles, so opening a listed session 404ed. Build the conversation from matching GenAI spans instead. Co-authored-by: Cursor --- CHANGELOG.md | 6 + backend/internal/handlers/query_test.go | 59 +++ backend/internal/otlp/logs.go | 4 + backend/internal/otlp/logs_test.go | 47 +++ backend/internal/storage/logs_store.go | 187 +++------ backend/internal/storage/logs_store_test.go | 4 + .../internal/storage/session_transcript.go | 368 ++++++++++++++++++ .../storage/session_transcript_test.go | 352 +++++++++++++++++ 8 files changed, 904 insertions(+), 123 deletions(-) create mode 100644 backend/internal/storage/session_transcript.go create mode 100644 backend/internal/storage/session_transcript_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index fd3951e..c9788c8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +### Fixed + +- Fixed GitHub Copilot local session transcripts returning "Failed to load session transcript". Copilot lists sessions from `copilot_chat.session.start` logs, but conversation content is on GenAI spans; transcripts now reconstruct user, assistant, and tool messages from those spans instead of 404ing when no log events map to roles. + ## [0.5.0] - 2026-06-18 ### Added diff --git a/backend/internal/handlers/query_test.go b/backend/internal/handlers/query_test.go index ddcf2fe..533cada 100644 --- a/backend/internal/handlers/query_test.go +++ b/backend/internal/handlers/query_test.go @@ -643,3 +643,62 @@ func TestGetBreakdownValues(t *testing.T) { }) } } + +func TestGetSessionTranscript_CopilotListedSessionReturnsTranscript(t *testing.T) { + h, cleanup := setupTestHandlers(t) + defer cleanup() + + ctx := context.Background() + now := time.Now().UTC().Truncate(time.Millisecond) + sessionID := "copilot-conv-1" + + if err := h.store.InsertLogs(ctx, []api.LogRecord{{ + Timestamp: now, + ServiceName: "copilot-chat", + Body: "copilot_chat.session.start", + LogAttributes: map[string]string{ + "event.name": "copilot_chat.session.start", + "session.id": sessionID, + }, + }}); err != nil { + t.Fatalf("failed to insert copilot session log: %v", err) + } + if err := h.store.InsertSpans(ctx, []api.Span{{ + Timestamp: now.Add(time.Second), + TraceID: "trace-copilot", + SpanID: "span-agent", + ServiceName: "copilot-chat", + SpanName: "invoke_agent copilot", + StatusCode: "OK", + SpanAttributes: map[string]string{ + "gen_ai.operation.name": "invoke_agent", + "gen_ai.conversation.id": sessionID, + "gen_ai.input.messages": `[{"role":"user","parts":[{"type":"text","content":"hello from copilot"}]}]`, + }, + }}); err != nil { + t.Fatalf("failed to insert copilot spans: %v", err) + } + + req := httptest.NewRequest(http.MethodGet, "/api/sessions/"+sessionID+"/transcript", nil) + rctx := chi.NewRouteContext() + rctx.URLParams.Add("sessionId", sessionID) + req = req.WithContext(context.WithValue(req.Context(), chi.RouteCtxKey, rctx)) + + rec := httptest.NewRecorder() + h.GetSessionTranscript(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("expected status 200, got %d: %s", rec.Code, rec.Body.String()) + } + + var resp api.TranscriptResponse + if err := json.NewDecoder(rec.Body).Decode(&resp); err != nil { + t.Fatalf("failed to decode response: %v", err) + } + if resp.SessionID != sessionID { + t.Fatalf("expected session %q, got %q", sessionID, resp.SessionID) + } + if len(resp.Messages) == 0 || resp.Messages[0].Role != "user" { + t.Fatalf("expected reconstructed user message, got %+v", resp.Messages) + } +} diff --git a/backend/internal/otlp/logs.go b/backend/internal/otlp/logs.go index 7fd04b9..ff257ea 100644 --- a/backend/internal/otlp/logs.go +++ b/backend/internal/otlp/logs.go @@ -3,6 +3,7 @@ package otlp import ( "encoding/json" "strconv" + "strings" "time" "github.com/tobilg/ai-observer/internal/api" @@ -35,6 +36,9 @@ func ConvertLogs(req *collogspb.ExportLogsServiceRequest) LogConversionResult { for _, lr := range sl.GetLogRecords() { logAttrs := convertAttributes(lr.GetAttributes()) + if eventName := strings.TrimSpace(lr.GetEventName()); eventName != "" && logAttrs["event.name"] == "" { + logAttrs["event.name"] = eventName + } // Resolve timestamp with fallbacks before creating log record timestamp := nanosToTime(lr.GetTimeUnixNano()) diff --git a/backend/internal/otlp/logs_test.go b/backend/internal/otlp/logs_test.go index c195b5b..8a1d85d 100644 --- a/backend/internal/otlp/logs_test.go +++ b/backend/internal/otlp/logs_test.go @@ -384,3 +384,50 @@ func TestConvertLogs_CodexSSEEvent_NonResponseCompleted_NoMetrics(t *testing.T) t.Errorf("Expected 0 derived metrics for chunk event, got %d", len(result.DerivedMetrics)) } } + +func TestConvertLogs_PreservesProtoEventName(t *testing.T) { + payload := `{ + "resourceLogs": [{ + "resource": { + "attributes": [ + {"key": "service.name", "value": {"stringValue": "copilot-chat"}} + ] + }, + "scopeLogs": [{ + "scope": {"name": "copilot-chat"}, + "logRecords": [ + { + "timeUnixNano": "1703500000000000000", + "severityNumber": 9, + "eventName": "copilot_chat.session.start", + "body": {"stringValue": "copilot_chat.session.start"}, + "attributes": [ + {"key": "session.id", "value": {"stringValue": "conv-123"}} + ] + } + ] + }] + }] + }` + + decoder, err := GetDecoder("application/json") + if err != nil { + t.Fatalf("Failed to get decoder: %v", err) + } + + req, err := decoder.DecodeLogs(strings.NewReader(payload)) + if err != nil { + t.Fatalf("Failed to decode logs: %v", err) + } + + result := ConvertLogs(req) + if len(result.Logs) != 1 { + t.Fatalf("Expected 1 log record, got %d", len(result.Logs)) + } + if got := result.Logs[0].LogAttributes["event.name"]; got != "copilot_chat.session.start" { + t.Fatalf("expected event.name from proto EventName, got %q", got) + } + if got := result.Logs[0].LogAttributes["session.id"]; got != "conv-123" { + t.Fatalf("expected session.id conv-123, got %q", got) + } +} diff --git a/backend/internal/storage/logs_store.go b/backend/internal/storage/logs_store.go index c9c00cf..336d9f8 100644 --- a/backend/internal/storage/logs_store.go +++ b/backend/internal/storage/logs_store.go @@ -225,7 +225,8 @@ func (s *DuckDBStore) GetLogLevels(ctx context.Context) (map[string]int64, error } // QuerySessions returns sessions with transcript messages from all services -// Supports: Claude Code (transcript.message), Gemini CLI (session.id), Codex CLI (conversation.id), Copilot (gen_ai.conversation.id), OpenCode (session.id) +// Supports: Claude Code (transcript.message), Gemini CLI (session.id), Codex CLI (conversation.id), +// Copilot (session.id / gen_ai.conversation.id / copilot_chat.session_id), OpenCode (session.id) func (s *DuckDBStore) QuerySessions(ctx context.Context, service string, from, to time.Time, limit, offset int) (*api.SessionsResponse, error) { s.mu.RLock() defer s.mu.RUnlock() @@ -240,13 +241,9 @@ func (s *DuckDBStore) QuerySessions(ctx context.Context, service string, from, t // - Gemini CLI: session.id in logAttributes // - Codex CLI: conversation.id in logAttributes // Note: Keys contain dots, use JSONPath with escaped quotes: $."key.name" - query := ` + query := fmt.Sprintf(` SELECT - COALESCE( - json_extract_string(LogAttributes, '$."session.id"'), - json_extract_string(LogAttributes, '$."conversation.id"'), - json_extract_string(LogAttributes, '$."gen_ai.conversation.id"') - ) as session_id, + %s as session_id, ServiceName, MIN(Timestamp) as start_time, MAX(Timestamp) as last_time, @@ -260,12 +257,8 @@ func (s *DuckDBStore) QuerySessions(ctx context.Context, service string, from, t FROM otel_logs WHERE Timestamp >= ?::TIMESTAMP AND Timestamp <= ?::TIMESTAMP AND json_valid(LogAttributes) - AND ( - json_extract_string(LogAttributes, '$."session.id"') IS NOT NULL - OR json_extract_string(LogAttributes, '$."conversation.id"') IS NOT NULL - OR json_extract_string(LogAttributes, '$."gen_ai.conversation.id"') IS NOT NULL - ) - ` + AND %s + `, sessionIDCoalesceSQL("LogAttributes"), sessionIDPresentSQL("LogAttributes")) args := []interface{}{fromStr, toStr} if service != "" { @@ -280,21 +273,13 @@ func (s *DuckDBStore) QuerySessions(ctx context.Context, service string, from, t ` // Get total count first - countQuery := ` - SELECT COUNT(DISTINCT COALESCE( - json_extract_string(LogAttributes, '$."session.id"'), - json_extract_string(LogAttributes, '$."conversation.id"'), - json_extract_string(LogAttributes, '$."gen_ai.conversation.id"') - )) + countQuery := fmt.Sprintf(` + SELECT COUNT(DISTINCT %s) FROM otel_logs WHERE Timestamp >= ?::TIMESTAMP AND Timestamp <= ?::TIMESTAMP AND json_valid(LogAttributes) - AND ( - json_extract_string(LogAttributes, '$."session.id"') IS NOT NULL - OR json_extract_string(LogAttributes, '$."conversation.id"') IS NOT NULL - OR json_extract_string(LogAttributes, '$."gen_ai.conversation.id"') IS NOT NULL - ) - ` + AND %s + `, sessionIDCoalesceSQL("LogAttributes"), sessionIDPresentSQL("LogAttributes")) countArgs := []interface{}{fromStr, toStr} if service != "" { countQuery += " AND ServiceName = ?" @@ -346,114 +331,31 @@ func (s *DuckDBStore) QuerySessions(ctx context.Context, service string, from, t }, nil } -// GetSessionTranscript returns all logs for a session, mapping events to transcript roles -// Supports: Claude Code, Gemini CLI, Codex CLI, GitHub Copilot +// GetSessionTranscript returns transcript messages for a session. +// Copilot local sessions are listed from session-start logs but the conversation +// content lives on GenAI spans, so matching spans are merged into the transcript. func (s *DuckDBStore) GetSessionTranscript(ctx context.Context, sessionID string) (*api.TranscriptResponse, error) { s.mu.RLock() defer s.mu.RUnlock() - // Query for logs matching either session.id or conversation.id - // Note: Keys contain dots, use JSONPath with escaped quotes: $."key.name" - // Use json_valid() guard to skip rows with malformed JSON in LogAttributes - query := ` - SELECT - Timestamp, - ServiceName, - Body, - LogAttributes - FROM otel_logs - WHERE json_valid(LogAttributes) - AND ( - json_extract_string(LogAttributes, '$."session.id"') = ? - OR json_extract_string(LogAttributes, '$."conversation.id"') = ? - OR json_extract_string(LogAttributes, '$."gen_ai.conversation.id"') = ? - ) - ORDER BY Timestamp ASC - ` - - rows, err := s.db.QueryContext(ctx, query, sessionID, sessionID, sessionID) + logRows, err := s.querySessionLogsLocked(ctx, sessionID) if err != nil { - return nil, fmt.Errorf("querying transcript: %w", err) + return nil, err } - defer rows.Close() - - var messages []api.TranscriptMessage - var serviceName string - var startTime, lastTime time.Time - isFirst := true - index := 0 - - for rows.Next() { - var timestamp time.Time - var svc string - var body sql.NullString - var logAttrs interface{} - - if err := rows.Scan(×tamp, &svc, &body, &logAttrs); err != nil { - return nil, fmt.Errorf("scanning transcript message: %w", err) - } - - attrs := scanJSONToMap(logAttrs) - - if isFirst { - serviceName = svc - startTime = timestamp - isFirst = false - } - lastTime = timestamp - - // Map event types to roles based on service - eventName := attrs["event.name"] - role := mapEventToRole(eventName, svc) - - // For transcript.message events, use the message.role attribute - if eventName == "transcript.message" { - role = attrs["message.role"] - } - - // Skip events that don't map to transcript roles - if role == "" { - continue - } - - // Get index from attributes if available (Claude Code imported) - if idxStr, ok := attrs["message.index"]; ok { - fmt.Sscanf(idxStr, "%d", &index) - } - - // Extract actual content based on event type - content := extractMessageContent(eventName, attrs, body.String) - - msg := api.TranscriptMessage{ - Timestamp: timestamp, - Role: role, - Content: content, - Index: index, - Model: getModelName(attrs), - ToolName: getToolName(attrs, eventName), - ToolInput: getToolInput(attrs), - ToolOutput: getToolOutput(attrs), - InputTokens: parseIntAttr(attrs, "input_tokens", "inputTokens", "gen_ai.usage.input_tokens", "llm.token_count.prompt"), - OutputTokens: parseIntAttr(attrs, "output_tokens", "outputTokens", "gen_ai.usage.output_tokens", "llm.token_count.completion"), - CacheRead: parseIntAttr(attrs, "cache_read_input_tokens", "cacheRead", "gen_ai.usage.cache_read.input_tokens"), - CacheWrite: parseIntAttr(attrs, "cache_creation_input_tokens", "cacheWrite", "gen_ai.usage.cache_creation.input_tokens"), - CostUSD: parseFloatAttr(attrs, "cost_usd", "costUsd", "llm.cost"), - DurationMs: parseIntAttr(attrs, "duration_ms", "durationMs"), - Success: parseBoolAttr(attrs, "success", "tool_success"), - OutputSize: parseIntAttr(attrs, "tool_result_size_bytes", "outputSize"), - } - - messages = append(messages, msg) - index++ - } - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("iterating transcript: %w", err) + spans, err := s.querySessionSpansLocked(ctx, sessionID) + if err != nil { + return nil, err } - - if len(messages) == 0 { + if len(logRows) == 0 && len(spans) == 0 { return nil, fmt.Errorf("session not found: %s", sessionID) } + messages := mergeTranscriptMessages( + transcriptMessagesFromLogs(logRows), + transcriptMessagesFromSpans(spans), + ) + + serviceName, startTime, lastTime := sessionTranscriptBounds(logRows, spans) return &api.TranscriptResponse{ SessionID: sessionID, ServiceName: serviceName, @@ -463,6 +365,39 @@ func (s *DuckDBStore) GetSessionTranscript(ctx context.Context, sessionID string }, nil } +func sessionTranscriptBounds(logRows []sessionLogRow, spans []api.Span) (string, time.Time, time.Time) { + var serviceName string + var startTime, lastTime time.Time + + for _, row := range logRows { + if serviceName == "" { + serviceName = row.ServiceName + } + if startTime.IsZero() || row.Timestamp.Before(startTime) { + startTime = row.Timestamp + } + if lastTime.IsZero() || row.Timestamp.After(lastTime) { + lastTime = row.Timestamp + } + } + for _, span := range spans { + if serviceName == "" { + serviceName = span.ServiceName + } + if startTime.IsZero() || span.Timestamp.Before(startTime) { + startTime = span.Timestamp + } + end := span.Timestamp + if span.Duration > 0 { + end = span.Timestamp.Add(time.Duration(span.Duration)) + } + if lastTime.IsZero() || end.After(lastTime) { + lastTime = end + } + } + return serviceName, startTime, lastTime +} + // mapEventToRole converts event names to transcript roles func mapEventToRole(eventName, serviceName string) string { switch eventName { @@ -648,9 +583,15 @@ func extractMessageContent(eventName string, attrs map[string]string, body strin return body case "gen_ai.client.inference.operation.details", "copilot_chat.agent.turn": + if output := extractGenAIMessageText(attrs["gen_ai.output.messages"], "assistant"); output != "" { + return output + } if output, ok := attrs["gen_ai.output.messages"]; ok && output != "" { return output } + if input := extractGenAIMessageText(attrs["gen_ai.input.messages"], "user"); input != "" { + return input + } return body case "copilot_chat.tool.call": diff --git a/backend/internal/storage/logs_store_test.go b/backend/internal/storage/logs_store_test.go index 3c60ee7..a75292c 100644 --- a/backend/internal/storage/logs_store_test.go +++ b/backend/internal/storage/logs_store_test.go @@ -16,6 +16,10 @@ func TestOpenCodeTranscriptHelpers(t *testing.T) { {"session.idle", "system"}, {"session.error", "system"}, {"commit", "system"}, + {"gen_ai.client.inference.operation.details", "assistant"}, + {"copilot_chat.agent.turn", "assistant"}, + {"copilot_chat.tool.call", "tool_use"}, + {"copilot_chat.session.start", ""}, } for _, tt := range roleTests { diff --git a/backend/internal/storage/session_transcript.go b/backend/internal/storage/session_transcript.go new file mode 100644 index 0000000..f5fa5e0 --- /dev/null +++ b/backend/internal/storage/session_transcript.go @@ -0,0 +1,368 @@ +package storage + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "sort" + "strings" + "time" + + "github.com/tobilg/ai-observer/internal/api" +) + +var sessionIDAttributeKeys = []string{ + "session.id", + "conversation.id", + "gen_ai.conversation.id", + "copilot_chat.session_id", + "copilot_chat.chat_session_id", +} + +func sessionIDJSONPath(key string) string { + return `$."` + key + `"` +} + +func sessionIDCoalesceSQL(column string) string { + parts := make([]string, 0, len(sessionIDAttributeKeys)) + for _, key := range sessionIDAttributeKeys { + parts = append(parts, fmt.Sprintf("json_extract_string(%s, '%s')", column, sessionIDJSONPath(key))) + } + return "COALESCE(" + strings.Join(parts, ", ") + ")" +} + +func sessionIDPresentSQL(column string) string { + parts := make([]string, 0, len(sessionIDAttributeKeys)) + for _, key := range sessionIDAttributeKeys { + parts = append(parts, fmt.Sprintf("json_extract_string(%s, '%s') IS NOT NULL", column, sessionIDJSONPath(key))) + } + return "(" + strings.Join(parts, " OR ") + ")" +} + +func sessionIDMatchSQL(column string) string { + parts := make([]string, 0, len(sessionIDAttributeKeys)) + for _, key := range sessionIDAttributeKeys { + parts = append(parts, fmt.Sprintf("json_extract_string(%s, '%s') = ?", column, sessionIDJSONPath(key))) + } + return "(" + strings.Join(parts, " OR ") + ")" +} + +func sessionIDMatchArgs(sessionID string) []interface{} { + args := make([]interface{}, len(sessionIDAttributeKeys)) + for i := range sessionIDAttributeKeys { + args[i] = sessionID + } + return args +} + +type sessionLogRow struct { + Timestamp time.Time + ServiceName string + Body string + Attrs map[string]string +} + +func (s *DuckDBStore) querySessionLogsLocked(ctx context.Context, sessionID string) ([]sessionLogRow, error) { + query := fmt.Sprintf(` + SELECT + Timestamp, + ServiceName, + Body, + LogAttributes + FROM otel_logs + WHERE json_valid(LogAttributes) + AND %s + ORDER BY Timestamp ASC + `, sessionIDMatchSQL("LogAttributes")) + + rows, err := s.db.QueryContext(ctx, query, sessionIDMatchArgs(sessionID)...) + if err != nil { + return nil, fmt.Errorf("querying transcript: %w", err) + } + defer rows.Close() + + var result []sessionLogRow + for rows.Next() { + var row sessionLogRow + var body sql.NullString + var logAttrs interface{} + if err := rows.Scan(&row.Timestamp, &row.ServiceName, &body, &logAttrs); err != nil { + return nil, fmt.Errorf("scanning transcript message: %w", err) + } + row.Body = body.String + row.Attrs = scanJSONToMap(logAttrs) + result = append(result, row) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterating transcript: %w", err) + } + return result, nil +} + +func (s *DuckDBStore) querySessionSpansLocked(ctx context.Context, sessionID string) ([]api.Span, error) { + query := fmt.Sprintf(` + SELECT + Timestamp, TraceId, SpanId, ParentSpanId, TraceState, + SpanName, SpanKind, ServiceName, ResourceAttributes, + ScopeName, ScopeVersion, SpanAttributes, Duration, + StatusCode, StatusMessage + FROM otel_traces + WHERE json_valid(SpanAttributes) + AND %s + ORDER BY Timestamp ASC + `, sessionIDMatchSQL("SpanAttributes")) + + return s.scanSpans(ctx, query, sessionIDMatchArgs(sessionID)...) +} + +func transcriptMessagesFromLogs(rows []sessionLogRow) []api.TranscriptMessage { + var messages []api.TranscriptMessage + index := 0 + for _, row := range rows { + eventName := row.Attrs["event.name"] + if eventName == "" && strings.Contains(row.Body, ".") && !strings.Contains(row.Body, " ") { + eventName = row.Body + } + + role := mapEventToRole(eventName, row.ServiceName) + if eventName == "transcript.message" { + role = row.Attrs["message.role"] + } + if role == "" { + continue + } + + if idxStr, ok := row.Attrs["message.index"]; ok { + fmt.Sscanf(idxStr, "%d", &index) + } + + messages = append(messages, api.TranscriptMessage{ + Timestamp: row.Timestamp, + Role: role, + Content: extractMessageContent(eventName, row.Attrs, row.Body), + Index: index, + Model: getModelName(row.Attrs), + ToolName: getToolName(row.Attrs, eventName), + ToolInput: getToolInput(row.Attrs), + ToolOutput: getToolOutput(row.Attrs), + InputTokens: parseIntAttr(row.Attrs, "input_tokens", "inputTokens", "gen_ai.usage.input_tokens", "llm.token_count.prompt"), + OutputTokens: parseIntAttr(row.Attrs, "output_tokens", "outputTokens", "gen_ai.usage.output_tokens", "llm.token_count.completion"), + CacheRead: parseIntAttr(row.Attrs, "cache_read_input_tokens", "cacheRead", "gen_ai.usage.cache_read.input_tokens"), + CacheWrite: parseIntAttr(row.Attrs, "cache_creation_input_tokens", "cacheWrite", "gen_ai.usage.cache_creation.input_tokens"), + CostUSD: parseFloatAttr(row.Attrs, "cost_usd", "costUsd", "llm.cost"), + DurationMs: parseIntAttr(row.Attrs, "duration_ms", "durationMs"), + Success: parseBoolAttr(row.Attrs, "success", "tool_success"), + OutputSize: parseIntAttr(row.Attrs, "tool_result_size_bytes", "outputSize"), + }) + index++ + } + return messages +} + +func transcriptMessagesFromSpans(spans []api.Span) []api.TranscriptMessage { + var messages []api.TranscriptMessage + for _, span := range spans { + messages = append(messages, transcriptMessagesFromSpan(span)...) + } + return messages +} + +func transcriptMessagesFromSpan(span api.Span) []api.TranscriptMessage { + attrs := span.SpanAttributes + if attrs == nil { + return nil + } + + op := strings.ToLower(strings.TrimSpace(attrs["gen_ai.operation.name"])) + model := getModelName(attrs) + durationMs := int(span.Duration / int64(time.Millisecond)) + if durationMs < 0 { + durationMs = 0 + } + + switch op { + case "invoke_agent": + content := firstNonEmpty( + extractGenAIMessageText(attrs["gen_ai.input.messages"], "user"), + attrs["copilot_chat.user_request"], + ) + if content == "" { + return nil + } + return []api.TranscriptMessage{{ + Timestamp: span.Timestamp, + Role: "user", + Content: content, + Model: model, + DurationMs: durationMs, + }} + + case "chat": + content := extractGenAIMessageText(attrs["gen_ai.output.messages"], "assistant") + msg := api.TranscriptMessage{ + Timestamp: span.Timestamp, + Role: "assistant", + Content: content, + Model: model, + InputTokens: parseIntAttr(attrs, "gen_ai.usage.input_tokens", "input_tokens"), + OutputTokens: parseIntAttr(attrs, "gen_ai.usage.output_tokens", "output_tokens"), + CacheRead: parseIntAttr(attrs, "gen_ai.usage.cache_read.input_tokens"), + CacheWrite: parseIntAttr(attrs, "gen_ai.usage.cache_creation.input_tokens"), + DurationMs: durationMs, + } + if content == "" && msg.InputTokens == 0 && msg.OutputTokens == 0 && msg.CacheRead == 0 && msg.CacheWrite == 0 { + return nil + } + return []api.TranscriptMessage{msg} + + case "execute_tool": + toolName := getToolName(attrs, op) + if toolName == "" { + toolName = span.SpanName + } + toolInput := getToolInput(attrs) + toolOutput := getToolOutput(attrs) + var success *bool + switch strings.ToUpper(span.StatusCode) { + case "OK", "STATUS_CODE_OK": + ok := true + success = &ok + case "ERROR", "STATUS_CODE_ERROR": + ok := false + success = &ok + default: + success = parseBoolAttr(attrs, "success") + } + + messages := []api.TranscriptMessage{{ + Timestamp: span.Timestamp, + Role: "tool_use", + Content: toolInput, + Model: model, + ToolName: toolName, + ToolInput: toolInput, + DurationMs: durationMs, + Success: success, + }} + if toolOutput != "" { + messages = append(messages, api.TranscriptMessage{ + Timestamp: span.Timestamp.Add(time.Millisecond), + Role: "tool_result", + Content: toolOutput, + Model: model, + ToolName: toolName, + ToolOutput: toolOutput, + Success: success, + OutputSize: len(toolOutput), + }) + } + return messages + } + + return nil +} + +func mergeTranscriptMessages(groups ...[]api.TranscriptMessage) []api.TranscriptMessage { + var merged []api.TranscriptMessage + seen := make(map[string]struct{}) + for _, group := range groups { + for _, msg := range group { + key := transcriptDedupKey(msg) + if _, exists := seen[key]; exists { + continue + } + seen[key] = struct{}{} + merged = append(merged, msg) + } + } + + sort.SliceStable(merged, func(i, j int) bool { + return merged[i].Timestamp.Before(merged[j].Timestamp) + }) + for i := range merged { + merged[i].Index = i + } + if merged == nil { + return []api.TranscriptMessage{} + } + return merged +} + +func transcriptDedupKey(msg api.TranscriptMessage) string { + return strings.Join([]string{ + msg.Timestamp.UTC().Format(time.RFC3339Nano), + msg.Role, + msg.Content, + msg.ToolName, + msg.ToolInput, + msg.ToolOutput, + }, "\x1f") +} + +func extractGenAIMessageText(raw, role string) string { + raw = strings.TrimSpace(raw) + if raw == "" { + return "" + } + + var messages []map[string]interface{} + if err := json.Unmarshal([]byte(raw), &messages); err != nil { + var single map[string]interface{} + if err := json.Unmarshal([]byte(raw), &single); err != nil { + return "" + } + messages = []map[string]interface{}{single} + } + + var parts []string + for _, msg := range messages { + msgRole, _ := msg["role"].(string) + if role != "" && msgRole != "" && !strings.EqualFold(msgRole, role) { + continue + } + if text := genAIContentText(msg["content"]); text != "" { + parts = append(parts, text) + continue + } + if text := genAIContentText(msg["parts"]); text != "" { + parts = append(parts, text) + } + } + return strings.Join(parts, "\n") +} + +func genAIContentText(value interface{}) string { + switch v := value.(type) { + case string: + return strings.TrimSpace(v) + case []interface{}: + var parts []string + for _, item := range v { + if text := genAIContentText(item); text != "" { + parts = append(parts, text) + } + } + return strings.Join(parts, "\n") + case map[string]interface{}: + if text, ok := v["content"].(string); ok && strings.TrimSpace(text) != "" { + return strings.TrimSpace(text) + } + if text, ok := v["text"].(string); ok && strings.TrimSpace(text) != "" { + return strings.TrimSpace(text) + } + if inner, ok := v["parts"]; ok { + return genAIContentText(inner) + } + } + return "" +} + +func firstNonEmpty(values ...string) string { + for _, value := range values { + if strings.TrimSpace(value) != "" { + return value + } + } + return "" +} diff --git a/backend/internal/storage/session_transcript_test.go b/backend/internal/storage/session_transcript_test.go new file mode 100644 index 0000000..5cf075c --- /dev/null +++ b/backend/internal/storage/session_transcript_test.go @@ -0,0 +1,352 @@ +package storage + +import ( + "context" + "strings" + "testing" + "time" + + "github.com/tobilg/ai-observer/internal/api" +) + +func TestExtractGenAIMessageText(t *testing.T) { + tests := []struct { + name string + raw string + role string + expected string + }{ + { + name: "copilot parts content", + raw: `[{"role":"user","parts":[{"type":"text","content":"Refactor this function"}]}]`, + role: "user", + expected: "Refactor this function", + }, + { + name: "parts text field", + raw: `[{"role":"assistant","parts":[{"type":"text","text":"Done."}]}]`, + role: "assistant", + expected: "Done.", + }, + { + name: "plain content string", + raw: `[{"role":"user","content":"Hello"}]`, + role: "user", + expected: "Hello", + }, + { + name: "filters other roles", + raw: `[{"role":"system","content":"ignore"},{"role":"user","content":"keep"}]`, + role: "user", + expected: "keep", + }, + { + name: "empty", + raw: "", + role: "user", + expected: "", + }, + { + name: "invalid json", + raw: "not-json", + role: "user", + expected: "", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := extractGenAIMessageText(tt.raw, tt.role) + if got != tt.expected { + t.Fatalf("expected %q, got %q", tt.expected, got) + } + }) + } +} + +func TestTranscriptMessagesFromSpan(t *testing.T) { + now := time.Date(2026, 6, 24, 12, 0, 0, 0, time.UTC) + + t.Run("invoke_agent user message", func(t *testing.T) { + msgs := transcriptMessagesFromSpan(api.Span{ + Timestamp: now, + ServiceName: "copilot-chat", + SpanAttributes: map[string]string{ + "gen_ai.operation.name": "invoke_agent", + "gen_ai.input.messages": `[{"role":"user","parts":[{"type":"text","content":"What is 2+2?"}]}]`, + "gen_ai.request.model": "gpt-4o", + }, + }) + if len(msgs) != 1 { + t.Fatalf("expected 1 message, got %d", len(msgs)) + } + if msgs[0].Role != "user" || msgs[0].Content != "What is 2+2?" { + t.Fatalf("unexpected message: %+v", msgs[0]) + } + }) + + t.Run("chat assistant with tokens", func(t *testing.T) { + msgs := transcriptMessagesFromSpan(api.Span{ + Timestamp: now, + Duration: int64(1500 * time.Millisecond), + ServiceName: "copilot-chat", + SpanAttributes: map[string]string{ + "gen_ai.operation.name": "chat", + "gen_ai.output.messages": `[{"role":"assistant","parts":[{"type":"text","content":"4"}]}]`, + "gen_ai.response.model": "gpt-4o-2024-08-06", + "gen_ai.usage.input_tokens": "12", + "gen_ai.usage.output_tokens": "3", + }, + }) + if len(msgs) != 1 { + t.Fatalf("expected 1 message, got %d", len(msgs)) + } + if msgs[0].Role != "assistant" || msgs[0].Content != "4" || msgs[0].InputTokens != 12 || msgs[0].DurationMs != 1500 { + t.Fatalf("unexpected message: %+v", msgs[0]) + } + }) + + t.Run("execute_tool use and result", func(t *testing.T) { + ok := true + msgs := transcriptMessagesFromSpan(api.Span{ + Timestamp: now, + ServiceName: "copilot-chat", + SpanName: "execute_tool readFile", + StatusCode: "OK", + SpanAttributes: map[string]string{ + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "readFile", + "gen_ai.tool.call.arguments": `{"filePath":"README.md"}`, + "gen_ai.tool.call.result": "hello", + }, + }) + if len(msgs) != 2 { + t.Fatalf("expected 2 messages, got %d", len(msgs)) + } + if msgs[0].Role != "tool_use" || msgs[0].ToolName != "readFile" || msgs[0].ToolInput != `{"filePath":"README.md"}` { + t.Fatalf("unexpected tool_use: %+v", msgs[0]) + } + if msgs[1].Role != "tool_result" || msgs[1].ToolOutput != "hello" { + t.Fatalf("unexpected tool_result: %+v", msgs[1]) + } + if msgs[0].Success == nil || *msgs[0].Success != ok { + t.Fatalf("expected success=true, got %+v", msgs[0].Success) + } + }) +} + +func TestGetSessionTranscript_CopilotSessionStartWithoutMappedEvents(t *testing.T) { + store, cleanup := setupTestStore(t) + defer cleanup() + + ctx := context.Background() + now := time.Date(2026, 6, 24, 12, 0, 0, 0, time.UTC) + sessionID := "a1b2c3d4-e5f6-7890-abcd-ef1234567890" + + if err := store.InsertLogs(ctx, []api.LogRecord{{ + Timestamp: now, + ServiceName: "copilot-chat", + Body: "copilot_chat.session.start", + LogAttributes: map[string]string{ + "event.name": "copilot_chat.session.start", + "session.id": sessionID, + "gen_ai.request.model": "gpt-4o", + "gen_ai.agent.name": "copilot", + }, + }}); err != nil { + t.Fatalf("InsertLogs failed: %v", err) + } + + listed, err := store.QuerySessions(ctx, "", now.Add(-time.Hour), now.Add(time.Hour), 10, 0) + if err != nil { + t.Fatalf("QuerySessions failed: %v", err) + } + if listed.Total != 1 || len(listed.Sessions) != 1 || listed.Sessions[0].SessionID != sessionID { + t.Fatalf("expected listed copilot session %q, got %+v", sessionID, listed) + } + + resp, err := store.GetSessionTranscript(ctx, sessionID) + if err != nil { + t.Fatalf("GetSessionTranscript should not 404 for a listed copilot session: %v", err) + } + if resp.ServiceName != "copilot-chat" { + t.Fatalf("expected service copilot-chat, got %q", resp.ServiceName) + } + if len(resp.Messages) != 0 { + t.Fatalf("expected no mapped log messages, got %+v", resp.Messages) + } +} + +func TestGetSessionTranscript_CopilotReconstructsFromSpans(t *testing.T) { + store, cleanup := setupTestStore(t) + defer cleanup() + + ctx := context.Background() + now := time.Date(2026, 6, 24, 12, 0, 0, 0, time.UTC) + sessionID := "conv-copilot-42" + + if err := store.InsertLogs(ctx, []api.LogRecord{{ + Timestamp: now, + ServiceName: "copilot-chat", + Body: "copilot_chat.session.start", + LogAttributes: map[string]string{ + "event.name": "copilot_chat.session.start", + "session.id": sessionID, + }, + }}); err != nil { + t.Fatalf("InsertLogs failed: %v", err) + } + + if err := store.InsertSpans(ctx, []api.Span{ + { + Timestamp: now.Add(time.Second), + TraceID: "trace-1", + SpanID: "span-agent", + ServiceName: "copilot-chat", + SpanName: "invoke_agent copilot", + Duration: int64(5 * time.Second), + StatusCode: "OK", + SpanAttributes: map[string]string{ + "gen_ai.operation.name": "invoke_agent", + "gen_ai.conversation.id": sessionID, + "copilot_chat.session_id": sessionID, + "gen_ai.input.messages": `[{"role":"user","parts":[{"type":"text","content":"Fix the flaky test"}]}]`, + "gen_ai.request.model": "gpt-4o", + }, + }, + { + Timestamp: now.Add(2 * time.Second), + TraceID: "trace-1", + SpanID: "span-tool", + ParentSpanID: "span-agent", + ServiceName: "copilot-chat", + SpanName: "execute_tool readFile", + Duration: int64(50 * time.Millisecond), + StatusCode: "OK", + SpanAttributes: map[string]string{ + "gen_ai.operation.name": "execute_tool", + "gen_ai.conversation.id": sessionID, + "gen_ai.tool.name": "readFile", + "gen_ai.tool.call.arguments": `{"filePath":"test.go"}`, + "gen_ai.tool.call.result": "package test", + }, + }, + { + Timestamp: now.Add(3 * time.Second), + TraceID: "trace-1", + SpanID: "span-chat", + ParentSpanID: "span-agent", + ServiceName: "copilot-chat", + SpanName: "chat gpt-4o", + Duration: int64(1200 * time.Millisecond), + StatusCode: "OK", + SpanAttributes: map[string]string{ + "gen_ai.operation.name": "chat", + "gen_ai.conversation.id": sessionID, + "gen_ai.response.model": "gpt-4o-2024-08-06", + "gen_ai.output.messages": `[{"role":"assistant","parts":[{"type":"text","content":"The test is racy because of shared state."}]}]`, + "gen_ai.usage.input_tokens": "80", + "gen_ai.usage.output_tokens": "20", + }, + }, + }); err != nil { + t.Fatalf("InsertSpans failed: %v", err) + } + + resp, err := store.GetSessionTranscript(ctx, sessionID) + if err != nil { + t.Fatalf("GetSessionTranscript failed: %v", err) + } + if resp.ServiceName != "copilot-chat" { + t.Fatalf("expected service copilot-chat, got %q", resp.ServiceName) + } + if len(resp.Messages) != 4 { + t.Fatalf("expected user + tool_use + tool_result + assistant, got %d: %+v", len(resp.Messages), rolesOf(resp.Messages)) + } + if resp.Messages[0].Role != "user" || !strings.Contains(resp.Messages[0].Content, "Fix the flaky test") { + t.Fatalf("expected user prompt, got %+v", resp.Messages[0]) + } + if resp.Messages[1].Role != "tool_use" || resp.Messages[1].ToolName != "readFile" { + t.Fatalf("expected tool_use, got %+v", resp.Messages[1]) + } + if resp.Messages[2].Role != "tool_result" || resp.Messages[2].ToolOutput != "package test" { + t.Fatalf("expected tool_result, got %+v", resp.Messages[2]) + } + if resp.Messages[3].Role != "assistant" || !strings.Contains(resp.Messages[3].Content, "racy") { + t.Fatalf("expected assistant reply, got %+v", resp.Messages[3]) + } + if resp.Messages[3].InputTokens != 80 || resp.Messages[3].OutputTokens != 20 { + t.Fatalf("expected token counts on assistant message, got %+v", resp.Messages[3]) + } +} + +func TestGetSessionTranscript_ClaudeImportedStillWorks(t *testing.T) { + store, cleanup := setupTestStore(t) + defer cleanup() + + ctx := context.Background() + now := time.Date(2026, 1, 2, 15, 4, 5, 0, time.UTC) + + if err := store.InsertLogs(ctx, []api.LogRecord{ + { + Timestamp: now, + ServiceName: "claude-code", + Body: "Hello Claude", + LogAttributes: map[string]string{ + "event.name": "transcript.message", + "session.id": "claude-sess-1", + "message.role": "user", + "message.index": "0", + }, + }, + { + Timestamp: now.Add(time.Second), + ServiceName: "claude-code", + Body: "Hi there", + LogAttributes: map[string]string{ + "event.name": "transcript.message", + "session.id": "claude-sess-1", + "message.role": "assistant", + "message.index": "1", + "model": "claude-sonnet-4", + }, + }, + }); err != nil { + t.Fatalf("InsertLogs failed: %v", err) + } + + resp, err := store.GetSessionTranscript(ctx, "claude-sess-1") + if err != nil { + t.Fatalf("GetSessionTranscript failed: %v", err) + } + if len(resp.Messages) != 2 { + t.Fatalf("expected 2 messages, got %d", len(resp.Messages)) + } + if resp.Messages[0].Role != "user" || resp.Messages[0].Content != "Hello Claude" { + t.Fatalf("unexpected user message: %+v", resp.Messages[0]) + } + if resp.Messages[1].Role != "assistant" || resp.Messages[1].Content != "Hi there" { + t.Fatalf("unexpected assistant message: %+v", resp.Messages[1]) + } +} + +func TestGetSessionTranscript_NotFound(t *testing.T) { + store, cleanup := setupTestStore(t) + defer cleanup() + + _, err := store.GetSessionTranscript(context.Background(), "missing") + if err == nil { + t.Fatal("expected error for missing session") + } + if !strings.Contains(err.Error(), "session not found") { + t.Fatalf("expected session not found error, got %v", err) + } +} + +func rolesOf(messages []api.TranscriptMessage) []string { + roles := make([]string, len(messages)) + for i, msg := range messages { + roles[i] = msg.Role + } + return roles +} From 9abe1ae8b74cf749f7e7773631daeb9429c684e5 Mon Sep 17 00:00:00 2001 From: tobilg Date: Tue, 15 Sep 2026 09:35:13 +0200 Subject: [PATCH 2/2] fix(sessions): preserve imported messages and complete span transcripts Resolve Copilot child spans through session aliases and parent relationships. Preserve imported records and correlate live messages by span identity. Recover parent assistant output without duplicating child replies or usage, and timestamp tool results at completion. Add regression coverage for imported tool results, session boundaries, child identifiers, parent output, repeated replies, log/span correlation, and overlapping tools. --- backend/internal/storage/logs_store.go | 28 +- .../internal/storage/session_transcript.go | 296 ++++++++++++++--- .../session_transcript_regression_test.go | 309 ++++++++++++++++++ 3 files changed, 577 insertions(+), 56 deletions(-) create mode 100644 backend/internal/storage/session_transcript_regression_test.go diff --git a/backend/internal/storage/logs_store.go b/backend/internal/storage/logs_store.go index 336d9f8..2e86e8c 100644 --- a/backend/internal/storage/logs_store.go +++ b/backend/internal/storage/logs_store.go @@ -342,18 +342,27 @@ func (s *DuckDBStore) GetSessionTranscript(ctx context.Context, sessionID string if err != nil { return nil, err } - spans, err := s.querySessionSpansLocked(ctx, sessionID) - if err != nil { - return nil, err + // Imported transcripts already contain the conversation. Preserve their + // records, indices and bounds even when live telemetry also exists. + imported := false + for _, row := range logRows { + if row.Attrs["event.name"] == "transcript.message" { + imported = true + break + } + } + var spans []api.Span + if !imported { + spans, err = s.querySessionSpansLocked(ctx, sessionID) + if err != nil { + return nil, err + } } if len(logRows) == 0 && len(spans) == 0 { return nil, fmt.Errorf("session not found: %s", sessionID) } - messages := mergeTranscriptMessages( - transcriptMessagesFromLogs(logRows), - transcriptMessagesFromSpans(spans), - ) + messages := mergeTranscriptMessages(logRows, transcriptMessagesFromSpans(spans, logRows)) serviceName, startTime, lastTime := sessionTranscriptBounds(logRows, spans) return &api.TranscriptResponse{ @@ -387,10 +396,7 @@ func sessionTranscriptBounds(logRows []sessionLogRow, spans []api.Span) (string, if startTime.IsZero() || span.Timestamp.Before(startTime) { startTime = span.Timestamp } - end := span.Timestamp - if span.Duration > 0 { - end = span.Timestamp.Add(time.Duration(span.Duration)) - } + end := spanEndTime(span) if lastTime.IsZero() || end.After(lastTime) { lastTime = end } diff --git a/backend/internal/storage/session_transcript.go b/backend/internal/storage/session_transcript.go index f5fa5e0..e1971e5 100644 --- a/backend/internal/storage/session_transcript.go +++ b/backend/internal/storage/session_transcript.go @@ -59,6 +59,8 @@ func sessionIDMatchArgs(sessionID string) []interface{} { type sessionLogRow struct { Timestamp time.Time ServiceName string + TraceID string + SpanID string Body string Attrs map[string]string } @@ -68,6 +70,8 @@ func (s *DuckDBStore) querySessionLogsLocked(ctx context.Context, sessionID stri SELECT Timestamp, ServiceName, + TraceId, + SpanId, Body, LogAttributes FROM otel_logs @@ -85,12 +89,14 @@ func (s *DuckDBStore) querySessionLogsLocked(ctx context.Context, sessionID stri var result []sessionLogRow for rows.Next() { var row sessionLogRow - var body sql.NullString + var body, traceID, spanID sql.NullString var logAttrs interface{} - if err := rows.Scan(&row.Timestamp, &row.ServiceName, &body, &logAttrs); err != nil { + if err := rows.Scan(&row.Timestamp, &row.ServiceName, &traceID, &spanID, &body, &logAttrs); err != nil { return nil, fmt.Errorf("scanning transcript message: %w", err) } row.Body = body.String + row.TraceID = traceID.String + row.SpanID = spanID.String row.Attrs = scanJSONToMap(logAttrs) result = append(result, row) } @@ -102,18 +108,96 @@ func (s *DuckDBStore) querySessionLogsLocked(ctx context.Context, sessionID stri func (s *DuckDBStore) querySessionSpansLocked(ctx context.Context, sessionID string) ([]api.Span, error) { query := fmt.Sprintf(` + WITH matching_traces AS ( + SELECT DISTINCT ServiceName, TraceId + FROM otel_traces + WHERE json_valid(SpanAttributes) AND %s + ) SELECT - Timestamp, TraceId, SpanId, ParentSpanId, TraceState, - SpanName, SpanKind, ServiceName, ResourceAttributes, - ScopeName, ScopeVersion, SpanAttributes, Duration, - StatusCode, StatusMessage - FROM otel_traces - WHERE json_valid(SpanAttributes) - AND %s - ORDER BY Timestamp ASC + t.Timestamp, t.TraceId, t.SpanId, t.ParentSpanId, t.TraceState, + t.SpanName, t.SpanKind, t.ServiceName, t.ResourceAttributes, + t.ScopeName, t.ScopeVersion, t.SpanAttributes, t.Duration, + t.StatusCode, t.StatusMessage + FROM otel_traces t + JOIN matching_traces m ON t.ServiceName = m.ServiceName AND t.TraceId = m.TraceId + ORDER BY t.Timestamp ASC `, sessionIDMatchSQL("SpanAttributes")) - return s.scanSpans(ctx, query, sessionIDMatchArgs(sessionID)...) + spans, err := s.scanSpans(ctx, query, sessionIDMatchArgs(sessionID)...) + if err != nil { + return nil, err + } + return selectSessionSpans(spans, sessionID), nil +} + +type sessionSpanKey struct { + ServiceName string + TraceID string + SpanID string +} + +func spanKey(span api.Span) sessionSpanKey { + return sessionSpanKey{span.ServiceName, span.TraceID, span.SpanID} +} + +func parentSpanKey(span api.Span) sessionSpanKey { + return sessionSpanKey{span.ServiceName, span.TraceID, span.ParentSpanID} +} + +// A Copilot agent can bridge an internal conversation ID and a VS Code chat ID. +// Recover descendants without IDs, but do not pull in unrelated siblings or +// descendants that explicitly identify a different session. +func selectSessionSpans(spans []api.Span, sessionID string) []api.Span { + aliases := make(map[string]map[string]bool) + selected := make(map[sessionSpanKey]bool) + children := make(map[sessionSpanKey][]api.Span) + var queue []api.Span + for _, span := range spans { + children[parentSpanKey(span)] = append(children[parentSpanKey(span)], span) + for _, key := range sessionIDAttributeKeys { + if id := span.SpanAttributes[key]; id == "" || id != sessionID { + continue + } + if aliases[span.ServiceName] == nil { + aliases[span.ServiceName] = make(map[string]bool) + } + for _, aliasKey := range sessionIDAttributeKeys { + if id := span.SpanAttributes[aliasKey]; id != "" { + aliases[span.ServiceName][id] = true + } + } + selected[spanKey(span)] = true + queue = append(queue, span) + break + } + } + for i := 0; i < len(queue); i++ { + for _, child := range children[spanKey(queue[i])] { + if selected[spanKey(child)] { + continue + } + hasID, matches := false, false + for _, key := range sessionIDAttributeKeys { + if id := child.SpanAttributes[key]; id != "" { + hasID = true + matches = matches || aliases[child.ServiceName][id] + } + } + if hasID && !matches { + continue + } + selected[spanKey(child)] = true + queue = append(queue, child) + } + } + var result []api.Span + for _, span := range spans { + if selected[spanKey(span)] { + result = append(result, span) + delete(selected, spanKey(span)) // Deduplicate telemetry by span identity. + } + } + return result } func transcriptMessagesFromLogs(rows []sessionLogRow) []api.TranscriptMessage { @@ -160,21 +244,91 @@ func transcriptMessagesFromLogs(rows []sessionLogRow) []api.TranscriptMessage { return messages } -func transcriptMessagesFromSpans(spans []api.Span) []api.TranscriptMessage { - var messages []api.TranscriptMessage +type spanTranscriptMessage struct { + Source sessionSpanKey + Message api.TranscriptMessage +} + +func transcriptMessagesFromSpans(spans []api.Span, logs []sessionLogRow) []spanTranscriptMessage { + byID := make(map[sessionSpanKey]api.Span, len(spans)) + generated := make(map[sessionSpanKey][]api.TranscriptMessage, len(spans)) + for _, span := range spans { + byID[spanKey(span)] = span + generated[spanKey(span)] = transcriptMessagesFromSpan(span) + } + logMessages := make(map[sessionSpanKey][]api.TranscriptMessage) + for _, row := range logs { + if row.TraceID != "" && row.SpanID != "" { + key := sessionSpanKey{row.ServiceName, row.TraceID, row.SpanID} + logMessages[key] = append(logMessages[key], transcriptMessagesFromLogs([]sessionLogRow{row})...) + } + } + // Agent usage aggregates its child calls. Track child replies and usage so + // parent output is a fallback, without duplicating replies or token counts. + descendantOutputs := make(map[sessionSpanKey]map[string]bool) + descendantUsage := make(map[sessionSpanKey]bool) + for _, span := range spans { + coverage := append([]api.TranscriptMessage(nil), generated[spanKey(span)]...) + coverage = append(coverage, logMessages[spanKey(span)]...) + for _, msg := range coverage { + if msg.Role != "assistant" { + continue + } + seen := map[sessionSpanKey]bool{spanKey(span): true} + for parent := parentSpanKey(span); parent.SpanID != "" && !seen[parent]; { + seen[parent] = true + if descendantOutputs[parent] == nil { + descendantOutputs[parent] = make(map[string]bool) + } + if msg.Content != "" { + descendantOutputs[parent][msg.Content] = true + } + if msg.InputTokens != 0 || msg.OutputTokens != 0 || msg.CacheRead != 0 || msg.CacheWrite != 0 { + descendantUsage[parent] = true + } + ancestor, ok := byID[parent] + if !ok { + break + } + parent = parentSpanKey(ancestor) + } + } + } + var messages []spanTranscriptMessage for _, span := range spans { - messages = append(messages, transcriptMessagesFromSpan(span)...) + for _, msg := range generated[spanKey(span)] { + if spanOperation(span) == "invoke_agent" && msg.Role == "assistant" { + if descendantOutputs[spanKey(span)][msg.Content] { + continue + } + if descendantUsage[spanKey(span)] { + msg.InputTokens, msg.OutputTokens, msg.CacheRead, msg.CacheWrite = 0, 0, 0, 0 + } + } + messages = append(messages, spanTranscriptMessage{Source: spanKey(span), Message: msg}) + } } return messages } +func spanOperation(span api.Span) string { + return strings.ToLower(strings.TrimSpace(span.SpanAttributes["gen_ai.operation.name"])) +} + +func spanEndTime(span api.Span) time.Time { + if span.Duration > 0 { + return span.Timestamp.Add(time.Duration(span.Duration)) + } + return span.Timestamp +} + func transcriptMessagesFromSpan(span api.Span) []api.TranscriptMessage { attrs := span.SpanAttributes if attrs == nil { return nil } - op := strings.ToLower(strings.TrimSpace(attrs["gen_ai.operation.name"])) + op := spanOperation(span) model := getModelName(attrs) durationMs := int(span.Duration / int64(time.Millisecond)) if durationMs < 0 { @@ -187,16 +341,24 @@ func transcriptMessagesFromSpan(span api.Span) []api.TranscriptMessage { extractGenAIMessageText(attrs["gen_ai.input.messages"], "user"), attrs["copilot_chat.user_request"], ) - if content == "" { - return nil + var messages []api.TranscriptMessage + if content != "" { + messages = append(messages, api.TranscriptMessage{ + Timestamp: span.Timestamp, Role: "user", Content: content, + Model: model, DurationMs: durationMs, + }) } - return []api.TranscriptMessage{{ - Timestamp: span.Timestamp, - Role: "user", - Content: content, - Model: model, - DurationMs: durationMs, - }} + if output := extractGenAIMessageText(attrs["gen_ai.output.messages"], "assistant"); output != "" { + messages = append(messages, api.TranscriptMessage{ + Timestamp: spanEndTime(span), Role: "assistant", Content: output, + Model: model, DurationMs: durationMs, + InputTokens: parseIntAttr(attrs, "gen_ai.usage.input_tokens", "input_tokens"), + OutputTokens: parseIntAttr(attrs, "gen_ai.usage.output_tokens", "output_tokens"), + CacheRead: parseIntAttr(attrs, "gen_ai.usage.cache_read.input_tokens"), + CacheWrite: parseIntAttr(attrs, "gen_ai.usage.cache_creation.input_tokens"), + }) + } + return messages case "chat": content := extractGenAIMessageText(attrs["gen_ai.output.messages"], "assistant") @@ -247,7 +409,7 @@ func transcriptMessagesFromSpan(span api.Span) []api.TranscriptMessage { }} if toolOutput != "" { messages = append(messages, api.TranscriptMessage{ - Timestamp: span.Timestamp.Add(time.Millisecond), + Timestamp: spanEndTime(span), Role: "tool_result", Content: toolOutput, Model: model, @@ -263,18 +425,42 @@ func transcriptMessagesFromSpan(span api.Span) []api.TranscriptMessage { return nil } -func mergeTranscriptMessages(groups ...[]api.TranscriptMessage) []api.TranscriptMessage { - var merged []api.TranscriptMessage - seen := make(map[string]struct{}) - for _, group := range groups { - for _, msg := range group { - key := transcriptDedupKey(msg) - if _, exists := seen[key]; exists { - continue - } - seen[key] = struct{}{} - merged = append(merged, msg) +// Only correlate messages from different signals when they identify the same +// span and content. Equal text in distinct log entries is not a duplicate. +func mergeTranscriptMessages(rows []sessionLogRow, spans []spanTranscriptMessage) []api.TranscriptMessage { + merged := transcriptMessagesFromLogs(rows) + if len(spans) == 0 { + if merged == nil { + return []api.TranscriptMessage{} } + return merged // Preserve imported message indices. + } + type messageKey struct { + Source sessionSpanKey + Role, Content, ToolName, ToolInput, ToolOutput string + } + keyFor := func(source sessionSpanKey, msg api.TranscriptMessage) messageKey { + return messageKey{source, msg.Role, msg.Content, msg.ToolName, msg.ToolInput, msg.ToolOutput} + } + logMessages := make(map[messageKey]int) + index := 0 + for _, row := range rows { + msgs := transcriptMessagesFromLogs([]sessionLogRow{row}) + if len(msgs) == 0 { + continue + } + if row.TraceID != "" && row.SpanID != "" && row.Attrs["event.name"] != "transcript.message" { + source := sessionSpanKey{row.ServiceName, row.TraceID, row.SpanID} + logMessages[keyFor(source, msgs[0])] = index + } + index++ + } + for _, span := range spans { + if i, ok := logMessages[keyFor(span.Source, span.Message)]; ok { + merged[i] = enrichTranscriptMessage(merged[i], span.Message) + continue + } + merged = append(merged, span.Message) } sort.SliceStable(merged, func(i, j int) bool { @@ -289,15 +475,35 @@ func mergeTranscriptMessages(groups ...[]api.TranscriptMessage) []api.Transcript return merged } -func transcriptDedupKey(msg api.TranscriptMessage) string { - return strings.Join([]string{ - msg.Timestamp.UTC().Format(time.RFC3339Nano), - msg.Role, - msg.Content, - msg.ToolName, - msg.ToolInput, - msg.ToolOutput, - }, "\x1f") +func enrichTranscriptMessage(log, span api.TranscriptMessage) api.TranscriptMessage { + if log.Model == "" { + log.Model = span.Model + } + if log.InputTokens == 0 { + log.InputTokens = span.InputTokens + } + if log.OutputTokens == 0 { + log.OutputTokens = span.OutputTokens + } + if log.CacheRead == 0 { + log.CacheRead = span.CacheRead + } + if log.CacheWrite == 0 { + log.CacheWrite = span.CacheWrite + } + if log.CostUSD == 0 { + log.CostUSD = span.CostUSD + } + if log.DurationMs == 0 { + log.DurationMs = span.DurationMs + } + if log.Success == nil { + log.Success = span.Success + } + if log.OutputSize == 0 { + log.OutputSize = span.OutputSize + } + return log } func extractGenAIMessageText(raw, role string) string { diff --git a/backend/internal/storage/session_transcript_regression_test.go b/backend/internal/storage/session_transcript_regression_test.go new file mode 100644 index 0000000..acc78ba --- /dev/null +++ b/backend/internal/storage/session_transcript_regression_test.go @@ -0,0 +1,309 @@ +package storage_test + +import ( + "context" + "path/filepath" + "testing" + "time" + + "github.com/tobilg/ai-observer/internal/api" + "github.com/tobilg/ai-observer/internal/importer" + "github.com/tobilg/ai-observer/internal/storage" +) + +func transcriptRegressionStore(t *testing.T) *storage.DuckDBStore { + t.Helper() + s, err := storage.NewDuckDBStore(filepath.Join(t.TempDir(), "review.duckdb")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { s.Close() }) + return s +} + +func TestSessionTranscript_PreserveDistinctImportedToolResults(t *testing.T) { + s := transcriptRegressionStore(t) + now := time.Date(2026, 6, 24, 12, 0, 0, 0, time.UTC) + index := 12 + p := &importer.ClaudeParser{} + logs := p.CreateTranscriptLogs(importer.ClaudeJSONLEntry{ + Type: "user", + Message: &importer.ClaudeMessage{Content: []importer.ClaudeContent{ + {Type: "tool_result", ToolUseID: "call-a", Content: "success"}, + {Type: "tool_result", ToolUseID: "call-b", Content: "success"}, + }}, + }, now, "claude-session", &index) + if len(logs) != 2 { + t.Fatalf("parser produced %d logs", len(logs)) + } + if err := s.InsertLogs(context.Background(), logs); err != nil { + t.Fatal(err) + } + if err := s.InsertSpans(context.Background(), []api.Span{{ + Timestamp: now.Add(time.Second), ServiceName: "claude-code", TraceID: "live", SpanID: "chat", + SpanAttributes: map[string]string{ + "gen_ai.operation.name": "chat", "session.id": "claude-session", + "gen_ai.output.messages": `[{"role":"assistant","content":"live response"}]`, + }, + }}); err != nil { + t.Fatal(err) + } + got, err := s.GetSessionTranscript(context.Background(), "claude-session") + if err != nil { + t.Fatal(err) + } + if len(got.Messages) != 2 { + t.Fatalf("two distinct imported tool results (call-a and call-b) must survive; got %d messages: %+v", len(got.Messages), got.Messages) + } + if got.Messages[0].Index != 12 || got.Messages[1].Index != 13 || !got.LastTime.Equal(now) { + t.Fatalf("imported indices and bounds changed: %+v", got) + } +} + +func TestSessionTranscript_CopilotChildSpanCorrelation(t *testing.T) { + for _, hasChatSessionID := range []bool{true, false} { + name := "no_session_attribute" + if hasChatSessionID { + name = "alternate_chat_session_id" + } + t.Run(name, func(t *testing.T) { + s := transcriptRegressionStore(t) + now := time.Date(2026, 6, 24, 12, 0, 0, 0, time.UTC) + ctx := context.Background() + if err := s.InsertLogs(ctx, []api.LogRecord{{ + Timestamp: now, ServiceName: "copilot-chat", Body: "copilot_chat.session.start", + LogAttributes: map[string]string{"event.name": "copilot_chat.session.start", "session.id": "conversation-1"}, + }}); err != nil { + t.Fatal(err) + } + toolAttrs := map[string]string{ + "gen_ai.operation.name": "execute_tool", "gen_ai.tool.name": "readFile", + "gen_ai.tool.call.arguments": `{"filePath":"README.md"}`, "gen_ai.tool.call.result": "file contents", + } + if hasChatSessionID { + toolAttrs["copilot_chat.chat_session_id"] = "vscode-chat-1" + } + if err := s.InsertSpans(ctx, []api.Span{ + {Timestamp: now, ServiceName: "copilot-chat", TraceID: "trace-1", SpanID: "agent", Duration: int64(3 * time.Second), + SpanAttributes: map[string]string{ + "gen_ai.operation.name": "invoke_agent", "gen_ai.conversation.id": "conversation-1", + "copilot_chat.session_id": "conversation-1", "copilot_chat.chat_session_id": "vscode-chat-1", + "gen_ai.input.messages": `[{"role":"user","parts":[{"type":"text","content":"Read README"}]}]`, + }}, + {Timestamp: now.Add(time.Second), ServiceName: "copilot-chat", TraceID: "trace-1", SpanID: "tool", ParentSpanID: "agent", Duration: int64(time.Second), SpanAttributes: toolAttrs}, + // These share a trace or parent ID but are not part of this session. + {Timestamp: now.Add(time.Second), ServiceName: "copilot-chat", TraceID: "trace-1", SpanID: "sibling", SpanAttributes: map[string]string{"gen_ai.operation.name": "execute_tool"}}, + {Timestamp: now.Add(time.Second), ServiceName: "other-service", TraceID: "trace-1", SpanID: "other-service-tool", ParentSpanID: "agent", SpanAttributes: map[string]string{"gen_ai.operation.name": "execute_tool"}}, + {Timestamp: now.Add(time.Second), ServiceName: "copilot-chat", TraceID: "trace-1", SpanID: "other-session-tool", ParentSpanID: "agent", SpanAttributes: map[string]string{"gen_ai.operation.name": "execute_tool", "gen_ai.conversation.id": "other-session"}}, + }); err != nil { + t.Fatal(err) + } + listed, err := s.QuerySessions(ctx, "", now.Add(-time.Hour), now.Add(time.Hour), 10, 0) + if err != nil { + t.Fatal(err) + } + if len(listed.Sessions) != 1 { + t.Fatalf("expected one listed session: %+v", listed) + } + got, err := s.GetSessionTranscript(ctx, listed.Sessions[0].SessionID) + if err != nil { + t.Fatal(err) + } + if len(got.Messages) != 3 { + t.Fatalf("expected user, tool_use, tool_result for linked child span; got %d messages: %+v", len(got.Messages), got.Messages) + } + }) + } +} + +func TestSessionTranscript_ToolResultUsesCompletionTime(t *testing.T) { + s := transcriptRegressionStore(t) + now := time.Date(2026, 6, 24, 12, 0, 0, 0, time.UTC) + if err := s.InsertSpans(context.Background(), []api.Span{{ + Timestamp: now, ServiceName: "copilot-chat", TraceID: "trace-1", SpanID: "tool", Duration: int64(5 * time.Second), + SpanAttributes: map[string]string{"gen_ai.operation.name": "execute_tool", "gen_ai.conversation.id": "conversation-1", "gen_ai.tool.name": "runCommand", "gen_ai.tool.call.result": "done"}, + }}); err != nil { + t.Fatal(err) + } + got, err := s.GetSessionTranscript(context.Background(), "conversation-1") + if err != nil { + t.Fatal(err) + } + if len(got.Messages) != 2 { + t.Fatalf("expected two messages: %+v", got.Messages) + } + if want := now.Add(5 * time.Second); !got.Messages[1].Timestamp.Equal(want) { + t.Fatalf("tool result must occur at completion %s; got %s", want.Format(time.RFC3339Nano), got.Messages[1].Timestamp.Format(time.RFC3339Nano)) + } +} + +func TestSessionTranscript_AgentOutputFallback(t *testing.T) { + s := transcriptRegressionStore(t) + now := time.Date(2026, 6, 24, 12, 0, 0, 0, time.UTC) + if err := s.InsertSpans(context.Background(), []api.Span{{ + Timestamp: now, ServiceName: "copilot-chat", TraceID: "trace-1", SpanID: "agent", Duration: int64(time.Second), + SpanAttributes: map[string]string{ + "gen_ai.operation.name": "invoke_agent", "gen_ai.conversation.id": "conversation-1", + "gen_ai.input.messages": `[{"role":"user","parts":[{"type":"text","content":"Hello"}]}]`, + "gen_ai.output.messages": `[{"role":"assistant","parts":[{"type":"text","content":"Hi there"}]}]`, + "gen_ai.usage.input_tokens": "11", "gen_ai.usage.output_tokens": "3", + }, + }}); err != nil { + t.Fatal(err) + } + got, err := s.GetSessionTranscript(context.Background(), "conversation-1") + if err != nil { + t.Fatal(err) + } + for _, message := range got.Messages { + if message.Role == "assistant" && message.Content == "Hi there" { + if !message.Timestamp.Equal(now.Add(time.Second)) || message.InputTokens != 11 || message.OutputTokens != 3 { + t.Fatalf("unexpected fallback timestamp or usage: %+v", message) + } + return + } + } + t.Fatalf("captured assistant output on agent span was discarded: %+v", got.Messages) +} + +func TestSessionTranscript_AgentFallbackDoesNotDuplicateChildOutputOrUsage(t *testing.T) { + for _, output := range []string{"done", "intermediate", ""} { + t.Run("child_output="+output, func(t *testing.T) { + s := transcriptRegressionStore(t) + now := time.Date(2026, 6, 24, 12, 0, 0, 0, time.UTC) + chatAttrs := map[string]string{"gen_ai.operation.name": "chat", "gen_ai.usage.input_tokens": "11", "gen_ai.usage.output_tokens": "3"} + if output != "" { + chatAttrs["gen_ai.output.messages"] = `[{"role":"assistant","content":"` + output + `"}]` + } + if err := s.InsertSpans(context.Background(), []api.Span{ + {Timestamp: now, ServiceName: "copilot-chat", TraceID: "trace", SpanID: "agent", Duration: int64(5 * time.Second), SpanAttributes: map[string]string{ + "gen_ai.operation.name": "invoke_agent", "gen_ai.conversation.id": "session", + "gen_ai.output.messages": `[{"role":"assistant","content":"done"}]`, + "gen_ai.usage.input_tokens": "11", "gen_ai.usage.output_tokens": "3", + }}, + {Timestamp: now.Add(time.Second), ServiceName: "copilot-chat", TraceID: "trace", SpanID: "wrapper", ParentSpanID: "agent"}, + {Timestamp: now.Add(2 * time.Second), ServiceName: "copilot-chat", TraceID: "trace", SpanID: "chat", ParentSpanID: "wrapper", SpanAttributes: chatAttrs}, + }); err != nil { + t.Fatal(err) + } + got, err := s.GetSessionTranscript(context.Background(), "session") + if err != nil { + t.Fatal(err) + } + var finalReplies, inputTokens, outputTokens int + for _, msg := range got.Messages { + if msg.Content == "done" { + finalReplies++ + } + inputTokens += msg.InputTokens + outputTokens += msg.OutputTokens + } + if finalReplies != 1 || inputTokens != 11 || outputTokens != 3 { + t.Fatalf("duplicated or missing reply/usage: %+v", got.Messages) + } + }) + } +} + +func TestSessionTranscript_IdenticalRepliesInDifferentTurnsSurvive(t *testing.T) { + s := transcriptRegressionStore(t) + now := time.Date(2026, 6, 24, 12, 0, 0, 0, time.UTC) + var spans []api.Span + for _, id := range []string{"first", "second"} { + spans = append(spans, api.Span{Timestamp: now, ServiceName: "copilot-chat", TraceID: "trace", SpanID: id, SpanAttributes: map[string]string{ + "gen_ai.operation.name": "chat", "gen_ai.conversation.id": "session", "gen_ai.output.messages": `[{"role":"assistant","content":"done"}]`, + }}) + } + if err := s.InsertSpans(context.Background(), spans); err != nil { + t.Fatal(err) + } + got, err := s.GetSessionTranscript(context.Background(), "session") + if err != nil { + t.Fatal(err) + } + if len(got.Messages) != 2 { + t.Fatalf("distinct replies were removed: %+v", got.Messages) + } +} + +func TestSessionTranscript_LogSpanCorrelationPreservesDistinctEventsAndUsage(t *testing.T) { + s := transcriptRegressionStore(t) + now := time.Date(2026, 6, 24, 12, 0, 0, 0, time.UTC) + var logs []api.LogRecord + var spans []api.Span + for _, id := range []string{"first", "second"} { + logs = append(logs, api.LogRecord{Timestamp: now, ServiceName: "copilot-chat", TraceID: "trace", SpanID: id, LogAttributes: map[string]string{ + "event.name": "gen_ai.client.inference.operation.details", "session.id": "session", "gen_ai.output.messages": `[{"role":"assistant","content":"done"}]`, + }}) + spans = append(spans, api.Span{Timestamp: now, ServiceName: "copilot-chat", TraceID: "trace", SpanID: id, SpanAttributes: map[string]string{ + "gen_ai.operation.name": "chat", "gen_ai.conversation.id": "session", "gen_ai.output.messages": `[{"role":"assistant","content":"done"}]`, "gen_ai.usage.input_tokens": "11", + }}) + } + if err := s.InsertLogs(context.Background(), logs); err != nil { + t.Fatal(err) + } + if err := s.InsertSpans(context.Background(), spans); err != nil { + t.Fatal(err) + } + got, err := s.GetSessionTranscript(context.Background(), "session") + if err != nil { + t.Fatal(err) + } + if len(got.Messages) != 2 || got.Messages[0].InputTokens != 11 || got.Messages[1].InputTokens != 11 { + t.Fatalf("expected two distinct replies enriched with span usage: %+v", got.Messages) + } +} + +func TestSessionTranscript_OverlappingToolCompletionOrder(t *testing.T) { + s := transcriptRegressionStore(t) + now := time.Date(2026, 6, 24, 12, 0, 0, 0, time.UTC) + var spans []api.Span + for i, name := range []string{"slow", "fast"} { + duration := 5 * time.Second + if name == "fast" { + duration = time.Second + } + spans = append(spans, api.Span{Timestamp: now.Add(time.Duration(i) * time.Second), ServiceName: "copilot-chat", TraceID: "trace", SpanID: name, Duration: int64(duration), SpanAttributes: map[string]string{ + "gen_ai.operation.name": "execute_tool", "gen_ai.conversation.id": "session", "gen_ai.tool.name": name, "gen_ai.tool.call.result": name + " result", + }}) + } + if err := s.InsertSpans(context.Background(), spans); err != nil { + t.Fatal(err) + } + got, err := s.GetSessionTranscript(context.Background(), "session") + if err != nil { + t.Fatal(err) + } + if len(got.Messages) != 4 || got.Messages[1].Role != "tool_use" || got.Messages[2].ToolName != "fast" || got.Messages[3].ToolName != "slow" { + t.Fatalf("unexpected overlapping tool order: %+v", got.Messages) + } +} + +func TestSessionTranscript_AgentFallbackRecognizesChildLogOutput(t *testing.T) { + s := transcriptRegressionStore(t) + now := time.Date(2026, 6, 24, 12, 0, 0, 0, time.UTC) + if err := s.InsertSpans(context.Background(), []api.Span{ + {Timestamp: now, ServiceName: "copilot-chat", TraceID: "trace", SpanID: "agent", SpanAttributes: map[string]string{ + "gen_ai.operation.name": "invoke_agent", "gen_ai.conversation.id": "session", + "gen_ai.output.messages": `[{"role":"assistant","content":"done"}]`, "gen_ai.usage.input_tokens": "11", + }}, + {Timestamp: now.Add(time.Second), ServiceName: "copilot-chat", TraceID: "trace", SpanID: "chat", ParentSpanID: "agent", SpanAttributes: map[string]string{"gen_ai.operation.name": "chat"}}, + }); err != nil { + t.Fatal(err) + } + if err := s.InsertLogs(context.Background(), []api.LogRecord{{ + Timestamp: now.Add(2 * time.Second), ServiceName: "copilot-chat", TraceID: "trace", SpanID: "chat", LogAttributes: map[string]string{ + "event.name": "gen_ai.client.inference.operation.details", "session.id": "session", + "gen_ai.output.messages": `[{"role":"assistant","content":"done"}]`, "gen_ai.usage.input_tokens": "11", + }, + }}); err != nil { + t.Fatal(err) + } + got, err := s.GetSessionTranscript(context.Background(), "session") + if err != nil { + t.Fatal(err) + } + if len(got.Messages) != 1 || got.Messages[0].Content != "done" || got.Messages[0].InputTokens != 11 { + t.Fatalf("parent reply duplicates correlated child log: %+v", got.Messages) + } +}