diff --git a/cmd/lk/analytics.go b/cmd/lk/analytics.go index 1adc5b99..54952e01 100644 --- a/cmd/lk/analytics.go +++ b/cmd/lk/analytics.go @@ -47,6 +47,12 @@ const ( // records or agent log records a page reads by default. const defaultPageLimit = 50 +// defaultTraceLimit caps how many spans `session traces` reads: ten of the +// server's largest pages, enough for a long agent session's tree. The tree +// needs a span's parent to place it, so the command reads every page up to +// the limit rather than printing one page. +const defaultTraceLimit = 1000 + var ( AnalyticsCommands = []*cli.Command{ { @@ -111,6 +117,13 @@ var ( Action: sessionRead(logOptions, fetchSessionLogs), Flags: append([]cli.Flag{jsonFlag}, analyticsLogFlags()...), }, + { + Name: "traces", + Usage: "Print a session's trace spans as a tree (requires --experimental-auth)", + ArgsUsage: "SESSION_ID", + Action: sessionRead(traceOptions, fetchSessionTraces), + Flags: append([]cli.Flag{jsonFlag}, analyticsTraceFlags()...), + }, }, }, }, @@ -290,6 +303,28 @@ func analyticsLogFlags() []cli.Flag { }, pageFlags(defaultPageLimit, "records")...) } +// analyticsTraceFlags returns fresh instances of the trace spans' own flags +// (the shared jsonFlag is added by the command), for the same reason as +// analyticsSessionListFlags. They aren't pageFlags: the command reads every +// page up to --limit, and --cursor resumes a read that stopped there. +func analyticsTraceFlags() []cli.Flag { + return []cli.Flag{ + &cli.IntFlag{ + Name: "limit", + Usage: "Maximum number of spans to read; the command reads every page up to it", + Value: defaultTraceLimit, + }, + // Hidden like session list's: pass the cursor a read that stopped at its + // limit printed to read the spans after it. Their parents were read + // before, so they print as roots. + &cli.StringFlag{ + Name: "cursor", + Usage: "Read from `CURSOR`, where a prior read stopped", + Hidden: true, + }, + } +} + // analyticsListModeFlags: --page (offset) exists only on the API-key analytics // endpoint, and the Public API session list is cursor-paginated. --start/--end // work in both modes. The filter flags are sent only to the Public API for now. @@ -697,12 +732,12 @@ func getUserAnalyticsSession(ctx context.Context, cmd *cli.Command) error { } // sessionRead builds the action of a command that reads one thing about a -// session — its participants, recordings, transcript or agent logs — which -// only the Public API serves. The action refuses to run without -// --experimental-auth before checking anything else, then reads the -// SESSION_ID argument and the command's options, so a bad flag fails before -// the project lookup, and hands fetch a client signed in as the user and the -// selected project. +// session — its participants, recordings, transcript, agent logs or trace +// spans — which only the Public API serves. The +// action refuses to run without --experimental-auth before checking anything +// else, then reads the SESSION_ID argument and the command's options, so a +// bad flag fails before the project lookup, and hands fetch a client signed +// in as the user and the selected project. func sessionRead[O any]( readOptions func(*cli.Command) (O, error), fetch func(ctx context.Context, client *public.Client, projectID, sessionID string, opts O, asJSON bool) error, @@ -932,9 +967,10 @@ func emptyTranscriptReason(ctx context.Context, client *public.Client, projectID "the recording is redacted); a session without an agent has none", sessionID) } -// sessionReadError explains why a session has no participants, transcript or -// agent logs (what) to print, and otherwise annotates the error like the -// other Public API commands. access is what the read requires. +// sessionReadError explains why a session has no participants, transcript, +// agent logs or trace spans (what) to print, and otherwise annotates the +// error like the other Public API commands. access is what the read +// requires. func sessionReadError(err error, projectID, sessionID, what string, access sessionReadAccess) error { if dashboardURL, ok := public.ObservabilityDisabled(err); ok { return observabilityDisabledError(sessionID, what, dashboardURL) @@ -952,7 +988,7 @@ const ( // projectReadAccess reads: a session's participants. projectReadAccess sessionReadAccess = iota // projectAdminAccess reads, which can hold user data: a session's - // recordings, transcript and agent logs. + // recordings, transcript, agent logs and trace spans. projectAdminAccess ) @@ -1016,3 +1052,61 @@ func emptyLogsReason(sessionID string, opts public.LogOptions) string { return fmt.Sprintf("Session %s has no agent logs: a session without an agent has none, "+ "and a running agent's records appear as it exports them", sessionID) } + +// traceReadOptions bounds `session traces`' read: unlike the API's pages, +// Limit counts spans across every page read. +type traceReadOptions struct { + // Limit is the most spans to read. + Limit int + // Cursor starts the read where a prior one stopped; empty starts from the + // session's first span. + Cursor string +} + +// traceOptions reads the trace flags. A bad limit fails here, before the +// project lookup. +func traceOptions(cmd *cli.Command) (traceReadOptions, error) { + page, err := pageOptions(cmd) + if err != nil { + return traceReadOptions{}, err + } + return traceReadOptions{Limit: int(page.Limit), Cursor: page.Cursor}, nil +} + +// fetchSessionTraces reads a session's spans page by page, in full pages but +// no more than opts.Limit in all, and prints them as one tree, saying why +// when there are none. A read that stops at the limit keeps the cursor for +// the rest. +func fetchSessionTraces(ctx context.Context, client *public.Client, projectID, sessionID string, opts traceReadOptions, asJSON bool) error { + read := public.TracePage{NextCursor: opts.Cursor} + for { + page, err := client.GetSessionTraces(ctx, projectID, sessionID, public.PageOptions{ + Limit: int32(min(public.MaxTracePageSize, opts.Limit-len(read.Spans))), + Cursor: read.NextCursor, + }) + if err != nil { + return sessionReadError(err, projectID, sessionID, "trace spans", projectAdminAccess) + } + read.Spans = append(read.Spans, page.Spans...) + read.NextCursor = page.NextCursor + if read.NextCursor == "" || len(page.Spans) == 0 || len(read.Spans) >= opts.Limit { + break + } + } + var empty string + if len(read.Spans) == 0 { + empty = emptyTracesReason(sessionID, opts.Cursor != "") + } + return render.SessionTraces(out, asJSON, read, empty) +} + +// emptyTracesReason explains a read with no spans. With user data recording +// off a first page with none is an error instead, so an empty one here means +// the agents exported nothing. +func emptyTracesReason(sessionID string, laterPage bool) string { + if laterPage { + return "No more spans" + } + return fmt.Sprintf("Session %s has no trace spans: a session without an agent has none, "+ + "and a running agent's spans appear as it exports them", sessionID) +} diff --git a/cmd/lk/analytics_test.go b/cmd/lk/analytics_test.go index 580df211..1648fd5e 100644 --- a/cmd/lk/analytics_test.go +++ b/cmd/lk/analytics_test.go @@ -1315,6 +1315,266 @@ func TestFetchSessionLogsErrors(t *testing.T) { } } +func TestSessionTracesCommand(t *testing.T) { + analyticsCmd := findCommandByName(AnalyticsCommands, "analytics") + require.NotNil(t, analyticsCmd) + sessionCmd := findCommandByName(analyticsCmd.Commands, "session") + require.NotNil(t, sessionCmd) + tracesCmd := findCommandByName(sessionCmd.Commands, "traces") + require.NotNil(t, tracesCmd, "'analytics session traces' command must exist") + require.NotNil(t, tracesCmd.Action) + for _, name := range []string{"limit", "cursor", "json"} { + assert.NotNil(t, findFlagByName(tracesCmd.Flags, name), "--%s", name) + } +} + +// TestSessionTracesRequiresExperimentalAuth checks the trace read, which has +// no API-key endpoint, refuses to run without --experimental-auth before +// reading its arguments or any config. +func TestSessionTracesRequiresExperimentalAuth(t *testing.T) { + for _, args := range [][]string{ + {"--experimental", "session", "traces", "RM_1"}, + {"--experimental", "session", "traces", "RM_1", "--limit", "0"}, + {"--experimental", "session", "traces"}, + } { + err := runAnalytics(args...) + require.ErrorContains(t, err, "only available under --experimental-auth") + } +} + +// traceCmdOptions runs traceOptions with the given arguments on a command +// built from fresh analyticsTraceFlags. +func traceCmdOptions(t *testing.T, args ...string) (traceReadOptions, error) { + t.Helper() + var opts traceReadOptions + var optsErr error + cmd := &cli.Command{ + Name: "traces", + Flags: analyticsTraceFlags(), + Action: func(_ context.Context, cmd *cli.Command) error { + opts, optsErr = traceOptions(cmd) + return nil + }, + } + require.NoError(t, cmd.Run(context.Background(), append([]string{"traces"}, args...))) + return opts, optsErr +} + +func TestTraceOptions(t *testing.T) { + tests := []struct { + name string + args []string + want traceReadOptions + wantErr string + }{ + {name: "defaults", want: traceReadOptions{Limit: defaultTraceLimit}}, + {name: "limit and cursor", args: []string{"--limit", "250", "--cursor", "abc"}, want: traceReadOptions{Limit: 250, Cursor: "abc"}}, + {name: "non-positive limit", args: []string{"--limit", "0"}, wantErr: "limit must be greater than 0"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + opts, err := traceCmdOptions(t, tt.args...) + if tt.wantErr != "" { + require.ErrorContains(t, err, tt.wantErr) + return + } + require.NoError(t, err) + assert.Equal(t, tt.want, opts) + }) + } +} + +// tracesAPI starts a stand-in Public API that answers the trace read with +// status and, for each cursor it is sent ("" for the first page), that page's +// body. It records the query of every request. +func tracesAPI(t *testing.T, status int, pages map[string]string) (*public.Client, *[]url.Values) { + t.Helper() + var queries []url.Values + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != "Bearer sekret" || r.URL.Path != "/v1/projects/p1/sessions/RM_1/traces" { + http.NotFound(w, r) + return + } + queries = append(queries, r.URL.Query()) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _, _ = w.Write([]byte(pages[r.URL.Query().Get("page.cursor")])) + })) + t.Cleanup(srv.Close) + client, err := public.New(srv.URL, "sekret") + require.NoError(t, err) + return client, &queries +} + +// tracePages is a session's trace across two pages, the second starting with +// the children of a span on the first. +var tracePages = map[string]string{ + "": `{"spans":[` + + `{"spanId":"a1","name":"agent_session"},` + + `{"spanId":"b2","parentSpanId":"a1","name":"agent_turn"}],` + + `"pageInfo":{"nextCursor":"c2","hasMore":true}}`, + "c2": `{"spans":[` + + `{"spanId":"c3","parentSpanId":"b2","name":"llm_request"},` + + `{"spanId":"d4","parentSpanId":"b2","name":"tts_request"}]}`, +} + +// TestFetchSessionTraces checks the command reads every page up to its limit, +// asking for full pages but no more spans than it has left, and prints one +// tree across the pages; a read that stops at its limit says how to read the +// rest. +func TestFetchSessionTraces(t *testing.T) { + tests := []struct { + name string + opts traceReadOptions + wantQueries []url.Values + wantOut []string + wantStatus []string + }{ + { + name: "every page", + opts: traceReadOptions{Limit: defaultTraceLimit}, + wantQueries: []url.Values{ + {"page.pageSize": {"100"}}, + {"page.pageSize": {"100"}, "page.cursor": {"c2"}}, + }, + wantOut: []string{"agent_session", "└─ agent_turn", " ├─ llm_request", " └─ tts_request"}, + }, + { + name: "stops at the limit", + opts: traceReadOptions{Limit: 2}, + wantQueries: []url.Values{{"page.pageSize": {"2"}}}, + wantOut: []string{"agent_session", "└─ agent_turn"}, + wantStatus: []string{"Printed 2 spans; more remain", "re-run with --cursor c2"}, + }, + { + name: "last page asks for what is left", + opts: traceReadOptions{Limit: 3}, + wantQueries: []url.Values{ + {"page.pageSize": {"3"}}, + {"page.pageSize": {"1"}, "page.cursor": {"c2"}}, + }, + }, + { + name: "from a cursor", + opts: traceReadOptions{Limit: defaultTraceLimit, Cursor: "c2"}, + wantQueries: []url.Values{{"page.pageSize": {"100"}, "page.cursor": {"c2"}}}, + wantOut: []string{"llm_request", "tts_request"}, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + client, queries := tracesAPI(t, http.StatusOK, tracePages) + stdout, stderr := captureOut(t) + + require.NoError(t, fetchSessionTraces(context.Background(), client, "p1", "RM_1", tt.opts, false)) + + assert.Equal(t, tt.wantQueries, *queries) + for _, want := range tt.wantOut { + assert.Contains(t, stdout.String(), want) + } + for _, want := range tt.wantStatus { + assert.Contains(t, stderr.String(), want) + } + if len(tt.wantStatus) == 0 { + assert.Empty(t, stderr.String()) + } + }) + } +} + +// TestFetchSessionTracesEmpty checks an empty read says why: the session has +// no spans, or a cursor's read has no more. +func TestFetchSessionTracesEmpty(t *testing.T) { + for _, tt := range []struct { + cursor string + wantStatus string + }{ + {wantStatus: "Session RM_1 has no trace spans"}, + {cursor: "c9", wantStatus: "No more spans"}, + } { + client, _ := tracesAPI(t, http.StatusOK, map[string]string{"": `{}`, "c9": `{"spans":[]}`}) + stdout, stderr := captureOut(t) + + require.NoError(t, fetchSessionTraces(context.Background(), client, "p1", "RM_1", traceReadOptions{Limit: 10, Cursor: tt.cursor}, false)) + assert.Empty(t, stdout.String()) + assert.Contains(t, stderr.String(), tt.wantStatus) + } +} + +// TestFetchSessionTracesJSON checks --json prints every span read, across +// pages, and still explains an empty read on stderr. +func TestFetchSessionTracesJSON(t *testing.T) { + client, _ := tracesAPI(t, http.StatusOK, tracePages) + stdout, _ := captureOut(t) + require.NoError(t, fetchSessionTraces(context.Background(), client, "p1", "RM_1", traceReadOptions{Limit: 10}, true)) + var got struct { + Items []map[string]any `json:"items"` + } + require.NoError(t, json.Unmarshal(stdout.Bytes(), &got)) + require.Len(t, got.Items, 4) + assert.Equal(t, "b2", got.Items[3]["parentSpanId"]) + + client, _ = tracesAPI(t, http.StatusOK, map[string]string{"": `{}`}) + stdout, stderr := captureOut(t) + require.NoError(t, fetchSessionTraces(context.Background(), client, "p1", "RM_1", traceReadOptions{Limit: 10}, true)) + assert.JSONEq(t, `{"items":[]}`, stdout.String()) + assert.Contains(t, stderr.String(), "has no trace spans") +} + +// TestFetchSessionTracesErrors checks why a session has no spans to print: it +// doesn't exist, or user data recording is off. +func TestFetchSessionTracesErrors(t *testing.T) { + tests := []struct { + name string + status int + body string + wantErr []string + }{ + { + name: "unknown session", + status: http.StatusNotFound, + body: `{"code":5,"message":"session not found"}`, + wantErr: []string{"no session RM_1 in project p1", "session not found"}, + }, + { + name: "recording off", + status: http.StatusBadRequest, + body: `{"code":9,"message":"user data recording is off for this project, so nothing was captured to read","details":[` + + `{"@type":"type.googleapis.com/livekit.publicapi.observability.v1.ObservabilityDisabled","dashboardUrl":"https://cloud.example/projects/p1/settings/observability"}]}`, + wantErr: []string{ + "session RM_1 has no trace spans", + "user data recording is off for this project", + "https://cloud.example/projects/p1/settings/observability", + }, + }, + { + name: "signed out", + status: http.StatusUnauthorized, + body: `{"code":16,"message":"authentication required"}`, + wantErr: []string{"authentication required", "lk cloud auth"}, + }, + { + name: "permission denied", + status: http.StatusForbidden, + body: `{"code":7,"message":"permission denied"}`, + wantErr: []string{"permission denied", "reading a session's trace spans requires being a project admin"}, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + client, _ := tracesAPI(t, tt.status, map[string]string{"": tt.body}) + stdout, _ := captureOut(t) + + err := fetchSessionTraces(context.Background(), client, "p1", "RM_1", traceReadOptions{Limit: 10}, false) + require.Error(t, err) + for _, want := range tt.wantErr { + assert.Contains(t, err.Error(), want) + } + assert.Empty(t, stdout.String()) + }) + } +} + // TestSessionAPIError checks a permission denial on a Public-API-only session // read says what the read requires, never to use API-key credentials, which // these reads can't use, while other errors keep cloudAPIError's hints. diff --git a/pkg/public/analytics.go b/pkg/public/analytics.go index efd3b7df..60066b8e 100644 --- a/pkg/public/analytics.go +++ b/pkg/public/analytics.go @@ -142,8 +142,8 @@ func (c *Client) ListProjectSessions(ctx context.Context, projectID string, opts } // PageOptions pages a read of one of a session's lists: its participants, -// transcript or agent logs. Zero values start from the first item with the -// server's page size. +// transcript, agent logs or trace spans. Zero values +// start from the first item with the server's page size. type PageOptions struct { // Limit caps how many items a page holds; 0 lets the server choose (50). // The server caps pages at 100, and a page can hold fewer items than its diff --git a/pkg/public/oapi/oapi.gen.go b/pkg/public/oapi/oapi.gen.go index e1270607..ecced3c2 100644 --- a/pkg/public/oapi/oapi.gen.go +++ b/pkg/public/oapi/oapi.gen.go @@ -656,6 +656,57 @@ func (e LivekitPublicapiObservabilityV1Role) Valid() bool { } } +// Defines values for LivekitPublicapiObservabilityV1SpanKind. +const ( + SPANKINDCLIENT LivekitPublicapiObservabilityV1SpanKind = "SPAN_KIND_CLIENT" + SPANKINDCONSUMER LivekitPublicapiObservabilityV1SpanKind = "SPAN_KIND_CONSUMER" + SPANKINDINTERNAL LivekitPublicapiObservabilityV1SpanKind = "SPAN_KIND_INTERNAL" + SPANKINDPRODUCER LivekitPublicapiObservabilityV1SpanKind = "SPAN_KIND_PRODUCER" + SPANKINDSERVER LivekitPublicapiObservabilityV1SpanKind = "SPAN_KIND_SERVER" + SPANKINDUNSPECIFIED LivekitPublicapiObservabilityV1SpanKind = "SPAN_KIND_UNSPECIFIED" +) + +// Valid indicates whether the value is a known member of the LivekitPublicapiObservabilityV1SpanKind enum. +func (e LivekitPublicapiObservabilityV1SpanKind) Valid() bool { + switch e { + case SPANKINDCLIENT: + return true + case SPANKINDCONSUMER: + return true + case SPANKINDINTERNAL: + return true + case SPANKINDPRODUCER: + return true + case SPANKINDSERVER: + return true + case SPANKINDUNSPECIFIED: + return true + default: + return false + } +} + +// Defines values for LivekitPublicapiObservabilityV1SpanStatus. +const ( + SPANSTATUSERROR LivekitPublicapiObservabilityV1SpanStatus = "SPAN_STATUS_ERROR" + SPANSTATUSOK LivekitPublicapiObservabilityV1SpanStatus = "SPAN_STATUS_OK" + SPANSTATUSUNSPECIFIED LivekitPublicapiObservabilityV1SpanStatus = "SPAN_STATUS_UNSPECIFIED" +) + +// Valid indicates whether the value is a known member of the LivekitPublicapiObservabilityV1SpanStatus enum. +func (e LivekitPublicapiObservabilityV1SpanStatus) Valid() bool { + switch e { + case SPANSTATUSERROR: + return true + case SPANSTATUSOK: + return true + case SPANSTATUSUNSPECIFIED: + return true + default: + return false + } +} + // CloudProtocolPIIRedactionCategory PIIRedactionCategory is the canonical taxonomy of PII entity types a project // // may select for redaction. It is the shared contract across Cloud, backend, @@ -4386,9 +4437,15 @@ type LivekitPublicapiObservabilityV1RecordingFileType string // LivekitPublicapiObservabilityV1Role Role is who a transcript message is from. type LivekitPublicapiObservabilityV1Role string -// LivekitPublicapiObservabilityV1Span Span is one unit of work in the session's trace. +// LivekitPublicapiObservabilityV1Span Span is one unit of work in the session's trace, such as an agent turn or +// +// an LLM request. type LivekitPublicapiObservabilityV1Span struct { - Attributes *map[string]string `json:"attributes,omitempty"` + // Attributes The span's attributes, typed and nested as the agent recorded them, with + // its resource's and scope's attributes merged in, such as lk.agent_name; + // the span's own value wins. An integer beyond ±2^53 is its decimal string, + // and bytes are base64. + Attributes *map[string]*GoogleProtobufValue `json:"attributes,omitempty"` // EndTime A Timestamp represents a point in time independent of any time zone or local // calendar, encoded as a count of seconds and fractions of seconds at @@ -4484,11 +4541,16 @@ type LivekitPublicapiObservabilityV1Span struct { EndTime *GoogleProtobufTimestamp `json:"endTime,omitempty"` Events *[]LivekitPublicapiObservabilityV1SpanEvent `json:"events,omitempty"` - // Kind OTEL span kind, projected as a string (e.g. "internal", "client"). - Kind *string `json:"kind,omitempty"` - Name *string `json:"name,omitempty"` + // Kind SpanKind is a span's OpenTelemetry kind. + Kind *LivekitPublicapiObservabilityV1SpanKind `json:"kind,omitempty"` + Name *string `json:"name,omitempty"` + + // ParentSpanId Empty for a root span. A span whose parent isn't among the session's + // spans is a root too, as the dashboard shows it. ParentSpanId *string `json:"parentSpanId,omitempty"` - SpanId *string `json:"spanId,omitempty"` + + // SpanId The span's ids, as lowercase hex. + SpanId *string `json:"spanId,omitempty"` // StartTime A Timestamp represents a point in time independent of any time zone or local // calendar, encoded as a count of seconds and fractions of seconds at @@ -4583,16 +4645,19 @@ type LivekitPublicapiObservabilityV1Span struct { // Examples: 2023-01-15T01:30:15.01Z, 2024-12-25T12:00:00Z StartTime *GoogleProtobufTimestamp `json:"startTime,omitempty"` - // Status "ok" | "error" | "" (unset). - Status *string `json:"status,omitempty"` - StatusMessage *string `json:"statusMessage,omitempty"` - TraceId *string `json:"traceId,omitempty"` + // Status SpanStatus is a span's OpenTelemetry status code. + Status *LivekitPublicapiObservabilityV1SpanStatus `json:"status,omitempty"` + StatusMessage *string `json:"statusMessage,omitempty"` + TraceId *string `json:"traceId,omitempty"` } -// LivekitPublicapiObservabilityV1SpanEvent SpanEvent is a timestamped event recorded within a Span. +// LivekitPublicapiObservabilityV1SpanEvent SpanEvent is a timestamped event recorded within a Span, such as an +// +// exception. type LivekitPublicapiObservabilityV1SpanEvent struct { - Attributes *map[string]string `json:"attributes,omitempty"` - Name *string `json:"name,omitempty"` + // Attributes The event's own attributes, typed and nested as the agent recorded them. + Attributes *map[string]*GoogleProtobufValue `json:"attributes,omitempty"` + Name *string `json:"name,omitempty"` // Timestamp A Timestamp represents a point in time independent of any time zone or local // calendar, encoded as a count of seconds and fractions of seconds at @@ -4688,11 +4753,26 @@ type LivekitPublicapiObservabilityV1SpanEvent struct { Timestamp *GoogleProtobufTimestamp `json:"timestamp,omitempty"` } +// LivekitPublicapiObservabilityV1SpanKind SpanKind is a span's OpenTelemetry kind. +type LivekitPublicapiObservabilityV1SpanKind string + +// LivekitPublicapiObservabilityV1SpanStatus SpanStatus is a span's OpenTelemetry status code. +type LivekitPublicapiObservabilityV1SpanStatus string + // LivekitPublicapiObservabilityV1TracesGetResponse defines model for livekit.publicapi.observability.v1.Traces.Get.Response. type LivekitPublicapiObservabilityV1TracesGetResponse struct { // PageInfo PageInfo is the cursor-pagination metadata shared by every list response. - PageInfo *LivekitPublicapiCommonV1PageInfo `json:"pageInfo,omitempty"` - Spans *[]LivekitPublicapiObservabilityV1Span `json:"spans,omitempty"` + PageInfo *LivekitPublicapiCommonV1PageInfo `json:"pageInfo,omitempty"` + + // Spans By start time, oldest first. Spans that start together are ordered by + // span id as the aggregator stores it: the id's 8 bytes read as a + // little-endian unsigned integer, which isn't the order of its hex + // string. The pages read the spans the first page found, so a span + // exported later waits for a new read. On an ACTIVE session, a span + // exported mid-read can still shift the later pages, which are read by + // offset, so a span can repeat or be skipped: dedupe by span_id, and + // read again from the first page to pick up a skipped span. + Spans *[]LivekitPublicapiObservabilityV1Span `json:"spans,omitempty"` } // LivekitPublicapiObservabilityV1TranscriptGetResponse defines model for livekit.publicapi.observability.v1.Transcript.Get.Response. @@ -10248,10 +10328,17 @@ type ClientInterface interface { // Corresponds with GET /v1/projects/{projectId}/sessions/{sessionId}/recording-url (the `ObservabilityServiceGetSessionRecordingURL` operationId). ObservabilityServiceGetSessionRecordingURL(ctx context.Context, projectId string, sessionId string, params *ObservabilityServiceGetSessionRecordingURLParams, reqEditors ...RequestEditorFn) (*http.Response, error) - // ObservabilityServiceGetSessionTraces GetSessionTraces + // ObservabilityServiceGetSessionTraces GetSessionTraces returns the spans the session's agents exported, by + // start time: what the dashboard's trace view shows. Each span names its + // parent, so the pages rebuild the span tree, and its attributes carry the + // turn and latency fields the dashboard's turn view reads, such as + // lk.generation_id and lk.response.ttft. // - // GetSessionTraces returns the session's spans. Sufficient on its own to - // answer latency questions. + // An ACTIVE session returns what its agents have exported so far; a span + // exported after the first page waits for a new read. The later pages are + // read by offset, so a span exported mid-read can still shift them: a span + // can repeat, or be skipped. Dedupe by span_id, and read again from the + // first page to pick up a skipped span. An unknown session is NotFound. // // Corresponds with GET /v1/projects/{projectId}/sessions/{sessionId}/traces (the `ObservabilityServiceGetSessionTraces` operationId). ObservabilityServiceGetSessionTraces(ctx context.Context, projectId string, sessionId string, params *ObservabilityServiceGetSessionTracesParams, reqEditors ...RequestEditorFn) (*http.Response, error) @@ -11298,11 +11385,19 @@ func (c *Client) ObservabilityServiceGetSessionRecordingURL(ctx context.Context, return c.Client.Do(req) } -// ObservabilityServiceGetSessionTraces GetSessionTraces +// ObservabilityServiceGetSessionTraces GetSessionTraces returns the spans the session's agents exported, by // -// GetSessionTraces returns the session's spans. Sufficient on its own to +// start time: what the dashboard's trace view shows. Each span names its +// parent, so the pages rebuild the span tree, and its attributes carry the +// turn and latency fields the dashboard's turn view reads, such as +// lk.generation_id and lk.response.ttft. // -// answer latency questions. +// An ACTIVE session returns what its agents have exported so far; a span +// +// exported after the first page waits for a new read. The later pages are +// read by offset, so a span exported mid-read can still shift them: a span +// can repeat, or be skipped. Dedupe by span_id, and read again from the +// first page to pick up a skipped span. An unknown session is NotFound. // // Corresponds with GET /v1/projects/{projectId}/sessions/{sessionId}/traces (the `ObservabilityServiceGetSessionTraces` operationId). func (c *Client) ObservabilityServiceGetSessionTraces(ctx context.Context, projectId string, sessionId string, params *ObservabilityServiceGetSessionTracesParams, reqEditors ...RequestEditorFn) (*http.Response, error) { @@ -16026,10 +16121,17 @@ type ClientWithResponsesInterface interface { // Corresponds with GET /v1/projects/{projectId}/sessions/{sessionId}/recording-url (the `ObservabilityServiceGetSessionRecordingURL` operationId). ObservabilityServiceGetSessionRecordingURLWithResponse(ctx context.Context, projectId string, sessionId string, params *ObservabilityServiceGetSessionRecordingURLParams, reqEditors ...RequestEditorFn) (*ObservabilityServiceGetSessionRecordingURLResponse, error) - // ObservabilityServiceGetSessionTracesWithResponse GetSessionTraces + // ObservabilityServiceGetSessionTracesWithResponse GetSessionTraces returns the spans the session's agents exported, by + // start time: what the dashboard's trace view shows. Each span names its + // parent, so the pages rebuild the span tree, and its attributes carry the + // turn and latency fields the dashboard's turn view reads, such as + // lk.generation_id and lk.response.ttft. // - // GetSessionTraces returns the session's spans. Sufficient on its own to - // answer latency questions. + // An ACTIVE session returns what its agents have exported so far; a span + // exported after the first page waits for a new read. The later pages are + // read by offset, so a span exported mid-read can still shift them: a span + // can repeat, or be skipped. Dedupe by span_id, and read again from the + // first page to pick up a skipped span. An unknown session is NotFound. // // Returns a wrapper object for the known response body format(s). // @@ -19626,11 +19728,19 @@ func (c *ClientWithResponses) ObservabilityServiceGetSessionRecordingURLWithResp return ParseObservabilityServiceGetSessionRecordingURLResponse(rsp) } -// ObservabilityServiceGetSessionTracesWithResponse GetSessionTraces +// ObservabilityServiceGetSessionTracesWithResponse GetSessionTraces returns the spans the session's agents exported, by +// +// start time: what the dashboard's trace view shows. Each span names its +// parent, so the pages rebuild the span tree, and its attributes carry the +// turn and latency fields the dashboard's turn view reads, such as +// lk.generation_id and lk.response.ttft. // -// GetSessionTraces returns the session's spans. Sufficient on its own to +// An ACTIVE session returns what its agents have exported so far; a span // -// answer latency questions. +// exported after the first page waits for a new read. The later pages are +// read by offset, so a span exported mid-read can still shift them: a span +// can repeat, or be skipped. Dedupe by span_id, and read again from the +// first page to pick up a skipped span. An unknown session is NotFound. // // Returns a wrapper object for the known response body format(s). // diff --git a/pkg/public/observability.go b/pkg/public/observability.go index a66b0df0..d7a9e5a8 100644 --- a/pkg/public/observability.go +++ b/pkg/public/observability.go @@ -270,6 +270,44 @@ func (c *Client) GetSessionLogs(ctx context.Context, projectID, sessionID string return &LogPage{Records: items(resp.JSON200.Records), NextCursor: pageCursor(resp.JSON200.PageInfo)}, nil } +// MaxTracePageSize is the most spans the server returns in one page of +// GetSessionTraces; a larger PageOptions.Limit is clamped to it. +const MaxTracePageSize = 100 + +// TracePage is one page of a session's trace spans. +type TracePage struct { + // Spans are by start time, oldest first, ties broken by span id. Each names + // its parent, so the pages together rebuild the span tree. + Spans []oapi.LivekitPublicapiObservabilityV1Span + // NextCursor is non-empty when more pages remain (pass it back as + // PageOptions.Cursor). + NextCursor string +} + +// GetSessionTraces returns one page of the spans a session's agents exported: +// what the dashboard's trace view shows. A span's attributes are typed, with +// its resource's and scope's merged in. The pages read the spans the first +// page found, so a span exported later needs a new read; an active session +// returns what its agents have exported so far. An unknown session is NotFound +// (see IsNotFound); an empty first page while the project's user data +// recording is off is a FailedPrecondition that ObservabilityDisabled +// recognizes. +func (c *Client) GetSessionTraces(ctx context.Context, projectID, sessionID string, opts PageOptions) (*TracePage, error) { + if err := opts.Validate(); err != nil { + return nil, err + } + params := &oapi.ObservabilityServiceGetSessionTracesParams{} + params.PagePageSize, params.PageCursor = opts.params() + resp, err := c.gen.ObservabilityServiceGetSessionTracesWithResponse(ctx, projectID, sessionID, params) + if err != nil { + return nil, err + } + if resp.JSON200 == nil { + return nil, responseError(resp.StatusCode(), resp.Body) + } + return &TracePage{Spans: items(resp.JSON200.Spans), NextCursor: pageCursor(resp.JSON200.PageInfo)}, nil +} + // gzipMagic opens every gzip stream (RFC 1952). var gzipMagic = []byte{0x1f, 0x8b} diff --git a/pkg/public/observability_test.go b/pkg/public/observability_test.go index f8616474..13f4c1c5 100644 --- a/pkg/public/observability_test.go +++ b/pkg/public/observability_test.go @@ -653,3 +653,165 @@ func TestGetSessionLogsErrors(t *testing.T) { }) } } + +// tracesPage is a GetSessionTraces response as the server's REST transcoder +// writes it: a root span with typed attributes, and a child that failed with +// an exception event. +const tracesPage = `{ + "spans": [ + { + "traceId": "0af7651916cd43dd8448eb211c80319c", + "spanId": "a1a1a1a1a1a1a1a1", + "name": "agent_session", + "kind": "SPAN_KIND_INTERNAL", + "startTime": "2026-10-07T11:00:00Z", + "endTime": "2026-10-07T11:05:00Z", + "status": "SPAN_STATUS_OK", + "attributes": {"lk.agent_name": "triage", "lk.room_name": "r1", "retries": 2, "nested": {"ok": true}} + }, + { + "traceId": "0af7651916cd43dd8448eb211c80319c", + "spanId": "b2b2b2b2b2b2b2b2", + "parentSpanId": "a1a1a1a1a1a1a1a1", + "name": "llm_request", + "kind": "SPAN_KIND_CLIENT", + "startTime": "2026-10-07T11:00:01Z", + "endTime": "2026-10-07T11:00:01.820Z", + "status": "SPAN_STATUS_ERROR", + "statusMessage": "rate limited", + "attributes": {"lk.response.ttft": 0.31}, + "events": [{"name": "exception", "timestamp": "2026-10-07T11:00:01.800Z", "attributes": {"exception.type": "RateLimitError"}}] + } + ], + "pageInfo": {"nextCursor": "next", "hasMore": true} +}` + +// TestGetSessionTraces checks the request GetSessionTraces sends for each +// option, and that spans come back typed with the page's cursor. +func TestGetSessionTraces(t *testing.T) { + tests := []struct { + name string + opts PageOptions + want url.Values + }{ + {name: "zero options send nothing", want: url.Values{}}, + { + name: "paging", + opts: PageOptions{Limit: 100, Cursor: "abc"}, + want: url.Values{"page.pageSize": {"100"}, "page.cursor": {"abc"}}, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var gotAuth, gotPath string + var gotQuery url.Values + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotAuth, gotPath, gotQuery = r.Header.Get("Authorization"), r.URL.Path, r.URL.Query() + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(tracesPage)) + })) + t.Cleanup(srv.Close) + + c, err := New(srv.URL, "sekret") + require.NoError(t, err) + + page, err := c.GetSessionTraces(context.Background(), "p1", "RM_1", tt.opts) + require.NoError(t, err) + + assert.Equal(t, "Bearer sekret", gotAuth) + assert.Equal(t, "/v1/projects/p1/sessions/RM_1/traces", gotPath) + assert.Equal(t, tt.want, gotQuery) + + assert.Equal(t, "next", page.NextCursor) + require.Len(t, page.Spans, 2) + root, child := page.Spans[0], page.Spans[1] + assert.Equal(t, "a1a1a1a1a1a1a1a1", *root.SpanId) + assert.Nil(t, root.ParentSpanId) + assert.Equal(t, oapi.SPANKINDINTERNAL, *root.Kind) + assert.Equal(t, oapi.SPANSTATUSOK, *root.Status) + assert.True(t, time.Date(2026, 10, 7, 11, 5, 0, 0, time.UTC).Equal(*root.EndTime)) + + assert.Equal(t, "a1a1a1a1a1a1a1a1", *child.ParentSpanId) + assert.Equal(t, "llm_request", *child.Name) + assert.Equal(t, oapi.SPANSTATUSERROR, *child.Status) + assert.Equal(t, "rate limited", *child.StatusMessage) + require.Len(t, *child.Events, 1) + assert.Equal(t, "exception", *(*child.Events)[0].Name) + }) + } +} + +// TestSpanJSON checks a span marshals back to the API's own shape, typed +// attributes and events included, so --json prints what the server sent. +func TestSpanJSON(t *testing.T) { + const span = `{"spanId":"b2","parentSpanId":"a1","name":"llm_request","kind":"SPAN_KIND_CLIENT",` + + `"startTime":"2026-10-07T11:00:01Z","status":"SPAN_STATUS_ERROR",` + + `"attributes":{"lk.response.ttft":0.31,"tags":["a","b"],"nested":{"ok":true},"none":null},` + + `"events":[{"name":"exception","attributes":{"exception.type":"RateLimitError"}}]}` + var s oapi.LivekitPublicapiObservabilityV1Span + require.NoError(t, json.Unmarshal([]byte(span), &s)) + got, err := json.Marshal(s) + require.NoError(t, err) + assert.JSONEq(t, span, string(got)) +} + +// TestGetSessionTracesRejectsBadLimit confirms a negative limit fails +// Validate, and fails GetSessionTraces before any request is sent. +func TestGetSessionTracesRejectsBadLimit(t *testing.T) { + opts := PageOptions{Limit: -1} + require.EqualError(t, opts.Validate(), "limit must not be negative") + + called := false + srv := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { called = true })) + t.Cleanup(srv.Close) + c, err := New(srv.URL, "sekret") + require.NoError(t, err) + + _, err = c.GetSessionTraces(context.Background(), "p1", "RM_1", opts) + require.EqualError(t, err, "limit must not be negative") + assert.False(t, called, "no request should be sent") +} + +// TestGetSessionTracesErrors checks an unknown session is NotFound and an +// empty first page with user data recording off carries ObservabilityDisabled. +func TestGetSessionTracesErrors(t *testing.T) { + tests := []struct { + name string + status int + body string + wantNotFound bool + wantDisabled bool + }{ + { + name: "unknown session", + status: http.StatusNotFound, + body: `{"code":5,"message":"session not found"}`, + wantNotFound: true, + }, + { + name: "recording off", + status: http.StatusBadRequest, + body: `{"code":9,"message":"user data recording is off","details":[` + + `{"@type":"type.googleapis.com/livekit.publicapi.observability.v1.ObservabilityDisabled","dashboardUrl":"https://cloud.example/p1"}]}`, + wantDisabled: true, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(tt.status) + _, _ = w.Write([]byte(tt.body)) + })) + t.Cleanup(srv.Close) + c, err := New(srv.URL, "sekret") + require.NoError(t, err) + + _, err = c.GetSessionTraces(context.Background(), "p1", "RM_1", PageOptions{}) + require.Error(t, err) + assert.Equal(t, tt.wantNotFound, IsNotFound(err)) + _, disabled := ObservabilityDisabled(err) + assert.Equal(t, tt.wantDisabled, disabled) + }) + } +} diff --git a/pkg/public/render/analytics.go b/pkg/public/render/analytics.go index 65ec94b2..7a60f75d 100644 --- a/pkg/public/render/analytics.go +++ b/pkg/public/render/analytics.go @@ -169,9 +169,9 @@ func SessionDetail(p *util.Printer, asJSON bool, s oapi.LivekitPublicapiAnalytic return nil } -// jsonPage is the --json shape of a page of a session's transcript or agent -// logs: {items, nextCursor}, like util.RenderPage's, with each item as the -// API sent it. +// jsonPage is the --json shape of a page of a session's transcript, agent +// logs or trace spans: {items, nextCursor}, like +// util.RenderPage's, with each item as the API sent it. type jsonPage[T any] struct { Items []T `json:"items"` NextCursor string `json:"nextCursor,omitempty"` @@ -468,6 +468,124 @@ func logLevelName(r oapi.LivekitPublicapiObservabilityV1LogRecord) string { return util.Dash(strings.ToUpper(oneLine(util.Deref(r.SeverityText)))) } +// SessionTraces prints a session's spans as a tree built from their parent +// ids, one line per span with its start time, duration and name, and a failed +// span's status message; --json prints the spans as the API sent them. A span +// whose parent isn't among them prints as a root, as the dashboard shows it. +// empty says why there are no spans, on stderr in both modes so --json output +// stays parseable. page.NextCursor is where the read stopped short of the +// session's last span. +func SessionTraces(p *util.Printer, asJSON bool, page public.TracePage, empty string) error { + jp := jsonPage[oapi.LivekitPublicapiObservabilityV1Span]{Items: page.Spans, NextCursor: page.NextCursor} + if err := renderPage(p, asJSON, jp, empty, spanLines); err != nil || asJSON || page.NextCursor == "" { + return err + } + p.Statusf("Printed %d spans; more remain — raise --limit to read them into this tree, or re-run with %s for the next ones", + len(page.Spans), util.Accented("--cursor "+page.NextCursor)) + return nil +} + +// spanLines renders spans as a tree, one line per span. +func spanLines(spans []oapi.LivekitPublicapiObservabilityV1Span) []string { + tree := spanTree(spans) + lines := make([]string, 0, len(tree)) + for _, l := range tree { + lines = append(lines, spanLine(l.span, l.guide)) + } + return lines +} + +// spanTreeLine is a span in tree order with the guide that draws its place. +type spanTreeLine struct { + span oapi.LivekitPublicapiObservabilityV1Span + guide string +} + +// spanTree orders spans depth-first from their roots, keeping the API's start +// time order among siblings. A root is a span with no parent or whose parent +// isn't among spans. Spans whose parents only name each other, which an agent +// shouldn't export, print from the first of them, so every span prints once. +func spanTree(spans []oapi.LivekitPublicapiObservabilityV1Span) []spanTreeLine { + ids := make(map[string]bool, len(spans)) + for _, s := range spans { + ids[util.Deref(s.SpanId)] = true + } + children := make(map[string][]int) + var roots []int + for i, s := range spans { + if parent := util.Deref(s.ParentSpanId); parent != "" && ids[parent] { + children[parent] = append(children[parent], i) + } else { + roots = append(roots, i) + } + } + + lines := make([]spanTreeLine, 0, len(spans)) + visited := make([]bool, len(spans)) + var walk func(i int, guide, indent string) + walk = func(i int, guide, indent string) { + visited[i] = true + lines = append(lines, spanTreeLine{span: spans[i], guide: guide}) + var next []int + for _, c := range children[util.Deref(spans[i].SpanId)] { + if !visited[c] { + next = append(next, c) + } + } + for n, c := range next { + if n == len(next)-1 { + walk(c, indent+"└─ ", indent+" ") + } else { + walk(c, indent+"├─ ", indent+"│ ") + } + } + } + for _, i := range roots { + walk(i, "", "") + } + for i := range spans { + if !visited[i] { + walk(i, "", "") + } + } + return lines +} + +// spanLine renders one span as a single line: its start time, its duration, +// its name after the tree guide, and a failed span's status message. +func spanLine(s oapi.LivekitPublicapiObservabilityV1Span, guide string) string { + at := "-" + if s.StartTime != nil && !s.StartTime.IsZero() { + at = s.StartTime.Local().Format("15:04:05.000") + } + line := fmt.Sprintf("%-12s %8s %s%s", at, spanDuration(s), guide, dashText(s.Name)) + if s.Status != nil && *s.Status == oapi.SPANSTATUSERROR { + if msg := oneLine(util.Deref(s.StatusMessage)); msg != "" { + line += " [error: " + msg + "]" + } else { + line += " [error]" + } + } + return line +} + +// spanDuration renders how long a span took, or a dash for one that hasn't +// ended. +func spanDuration(s oapi.LivekitPublicapiObservabilityV1Span) string { + if s.StartTime == nil || s.EndTime == nil || s.EndTime.Before(*s.StartTime) { + return "-" + } + d := s.EndTime.Sub(*s.StartTime) + switch { + case d < time.Second: + return formatMs(float64(d) / float64(time.Millisecond)) + case d < time.Minute: + return fmt.Sprintf("%.2fs", d.Seconds()) + default: + return d.Round(time.Second).String() + } +} + // RecordingLabel names a recording ("audio" or "chat-history") for a sentence. func RecordingLabel(recording string) string { if recording == public.RecordingChatHistory { diff --git a/pkg/public/render/analytics_test.go b/pkg/public/render/analytics_test.go index 344d6493..e0f1a1be 100644 --- a/pkg/public/render/analytics_test.go +++ b/pkg/public/render/analytics_test.go @@ -513,3 +513,132 @@ func TestSessionLogsJSON(t *testing.T) { assert.Equal(t, "b7ad6b7169203331", got.Items[1]["spanId"]) assert.Equal(t, "2026-10-07T11:00:01.25Z", got.Items[0]["timestamp"]) } + +// traceSpans is a session's spans as the API sends them, by start time: a +// root with two children, one of which has a failed child and one still +// running, and a span whose parent wasn't exported, which the tree shows as a +// root. +const traceSpans = `[ + {"spanId": "a1", "name": "agent_session", "kind": "SPAN_KIND_INTERNAL", "startTime": "2026-10-07T11:00:00Z", "endTime": "2026-10-07T11:05:00Z", "status": "SPAN_STATUS_OK", "attributes": {"lk.agent_name": "triage"}}, + {"spanId": "b2", "parentSpanId": "a1", "name": "user_turn", "startTime": "2026-10-07T11:00:01Z", "endTime": "2026-10-07T11:00:01.320Z"}, + {"spanId": "c3", "parentSpanId": "a1", "name": "agent_turn", "startTime": "2026-10-07T11:00:01.400Z", "endTime": "2026-10-07T11:00:03.150Z"}, + {"spanId": "d4", "parentSpanId": "c3", "name": "llm_request", "startTime": "2026-10-07T11:00:01.500Z", "endTime": "2026-10-07T11:00:02.320Z", "status": "SPAN_STATUS_ERROR", "statusMessage": "rate\n limited", "events": [{"name": "exception"}]}, + {"spanId": "e5", "parentSpanId": "c3", "name": "tts_request", "startTime": "2026-10-07T11:00:02.400Z"}, + {"spanId": "f6", "parentSpanId": "ffff", "name": "on_enter", "startTime": "2026-10-07T11:00:04Z", "endTime": "2026-10-07T11:00:04.0004Z", "status": "SPAN_STATUS_ERROR"} +]` + +func decodeSpans(t *testing.T, body string) []oapi.LivekitPublicapiObservabilityV1Span { + t.Helper() + var spans []oapi.LivekitPublicapiObservabilityV1Span + require.NoError(t, json.Unmarshal([]byte(body), &spans)) + return spans +} + +// TestSessionTracesText checks the spans print as a tree built from their +// parent ids, each line with its start time and duration, a failed span with +// its status message, and a span whose parent is missing as a root. +func TestSessionTracesText(t *testing.T) { + prevLocal := time.Local + time.Local = time.UTC + t.Cleanup(func() { time.Local = prevLocal }) + + page := public.TracePage{Spans: decodeSpans(t, traceSpans)} + var stdout, stderr bytes.Buffer + require.NoError(t, SessionTraces(util.NewPrinter(&stdout, &stderr, false), false, page, "unused")) + + assert.Equal(t, strings.Join([]string{ + "11:00:00.000 5m0s agent_session", + "11:00:01.000 320ms ├─ user_turn", + "11:00:01.400 1.75s └─ agent_turn", + "11:00:01.500 820ms ├─ llm_request [error: rate limited]", + "11:00:02.400 - └─ tts_request", + "11:00:04.000 0.4ms on_enter [error]", + "", + }, "\n"), stdout.String()) + assert.Empty(t, stderr.String()) +} + +// TestSessionTracesCycle checks spans whose parents name each other, which an +// agent shouldn't export, still print once each instead of looping. +// TestSessionTracesStripEscapes checks a span's name and status message, +// which the agent's code and the libraries it calls chose, reach a terminal +// with their escape sequences stripped. +func TestSessionTracesStripEscapes(t *testing.T) { + text := "hi" + escapes + status := oapi.SPANSTATUSERROR + spans := []oapi.LivekitPublicapiObservabilityV1Span{ + {SpanId: ptr("a"), Name: &text, Status: &status, StatusMessage: &text}, + {SpanId: ptr("b"), ParentSpanId: ptr("a"), Name: &text}, + } + + var stdout, stderr bytes.Buffer + require.NoError(t, SessionTraces(terminalPrinter(&stdout, &stderr), false, public.TracePage{Spans: spans}, "")) + assertNoEscapes(t, stdout.String()) + assert.Equal(t, strings.Join([]string{ + "- - hi]0;pwned[2J [error: hi]0;pwned[2J]", + "- - └─ hi]0;pwned[2J", + "", + }, "\n"), stdout.String()) +} + +func TestSessionTracesCycle(t *testing.T) { + spans := decodeSpans(t, `[ + {"spanId": "x", "parentSpanId": "y", "name": "x"}, + {"spanId": "y", "parentSpanId": "x", "name": "y"}, + {"spanId": "z", "parentSpanId": "z", "name": "z"} + ]`) + var stdout bytes.Buffer + require.NoError(t, SessionTraces(util.NewPrinter(&stdout, nil, false), false, public.TracePage{Spans: spans}, "")) + assert.Equal(t, strings.Join([]string{ + "- - x", + "- - └─ y", + "- - z", + "", + }, "\n"), stdout.String()) +} + +// TestSessionTracesMore checks a read that stopped at its limit says how to +// read the rest. +func TestSessionTracesMore(t *testing.T) { + page := public.TracePage{Spans: decodeSpans(t, traceSpans), NextCursor: "c2"} + var stdout, stderr bytes.Buffer + require.NoError(t, SessionTraces(util.NewPrinter(&stdout, &stderr, false), false, page, "")) + assert.Contains(t, stderr.String(), "Printed 6 spans; more remain") + assert.Contains(t, stderr.String(), "--limit") + assert.Contains(t, stderr.String(), "or re-run with --cursor c2 for the next ones") +} + +// TestSessionTracesEmpty prints the caller's reason for no spans, on stderr in +// both modes. +func TestSessionTracesEmpty(t *testing.T) { + var stdout, stderr bytes.Buffer + require.NoError(t, SessionTraces(util.NewPrinter(&stdout, &stderr, false), false, public.TracePage{}, "No spans")) + assert.Empty(t, stdout.String()) + assert.Equal(t, "No spans\n", stderr.String()) + + stdout.Reset() + stderr.Reset() + require.NoError(t, SessionTraces(util.NewPrinter(&stdout, &stderr, false), true, public.TracePage{}, "No spans")) + assert.JSONEq(t, `{"items":[]}`, stdout.String()) + assert.Equal(t, "No spans\n", stderr.String()) +} + +// TestSessionTracesJSON checks --json prints the spans as the API sent them, +// in its order, with the cursor where the read stopped. +func TestSessionTracesJSON(t *testing.T) { + page := public.TracePage{Spans: decodeSpans(t, traceSpans), NextCursor: "c2"} + var stdout bytes.Buffer + require.NoError(t, SessionTraces(util.NewPrinter(&stdout, nil, true), true, page, "")) + + var got struct { + Items []map[string]any `json:"items"` + NextCursor string `json:"nextCursor"` + } + require.NoError(t, json.Unmarshal(stdout.Bytes(), &got)) + assert.Equal(t, "c2", got.NextCursor) + require.Len(t, got.Items, 6) + assert.Equal(t, "a1", got.Items[0]["spanId"]) + assert.Equal(t, map[string]any{"lk.agent_name": "triage"}, got.Items[0]["attributes"]) + assert.Equal(t, "SPAN_STATUS_ERROR", got.Items[3]["status"]) + assert.Equal(t, "c3", got.Items[3]["parentSpanId"]) +}