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..2e86e8c 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,40 @@ 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) - } - 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) + 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 } - - // 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) + var spans []api.Span + if !imported { + 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(logRows, transcriptMessagesFromSpans(spans, logRows)) + + serviceName, startTime, lastTime := sessionTranscriptBounds(logRows, spans) return &api.TranscriptResponse{ SessionID: sessionID, ServiceName: serviceName, @@ -463,6 +374,36 @@ 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 := spanEndTime(span) + 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 +589,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..e1971e5 --- /dev/null +++ b/backend/internal/storage/session_transcript.go @@ -0,0 +1,574 @@ +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 + TraceID string + SpanID string + Body string + Attrs map[string]string +} + +func (s *DuckDBStore) querySessionLogsLocked(ctx context.Context, sessionID string) ([]sessionLogRow, error) { + query := fmt.Sprintf(` + SELECT + Timestamp, + ServiceName, + TraceId, + SpanId, + 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, traceID, spanID sql.NullString + var logAttrs interface{} + 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) + } + 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(` + WITH matching_traces AS ( + SELECT DISTINCT ServiceName, TraceId + FROM otel_traces + WHERE json_valid(SpanAttributes) AND %s + ) + SELECT + 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")) + + 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 { + 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 +} + +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 { + 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 := spanOperation(span) + 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"], + ) + var messages []api.TranscriptMessage + if content != "" { + messages = append(messages, 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") + 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: spanEndTime(span), + Role: "tool_result", + Content: toolOutput, + Model: model, + ToolName: toolName, + ToolOutput: toolOutput, + Success: success, + OutputSize: len(toolOutput), + }) + } + return messages + } + + return nil +} + +// 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 { + 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 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 { + 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_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) + } +} 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 +}