fathom/internal/web/traces.go

package web

import (
	"net/http"
	"sort"
	"strings"

	"git.z.gd/zzg/fathom/internal/model"
	"git.z.gd/zzg/fathom/internal/store"
)

// TraceSummary is one row of the trace list. The fields beyond identity and
// timing are what make the list scannable without opening anything: an engineer
// looking for the run that went wrong wants to see cost, tool count, and error
// state at a glance.
type TraceSummary struct {
	TraceID    string  `json:"trace_id"`
	Name       string  `json:"name"`
	Service    string  `json:"service"`
	StartTS    int64   `json:"start_ts"`
	EndTS      int64   `json:"end_ts"`
	DurMicros  int64   `json:"duration_micros"`
	SpanCount  int     `json:"span_count"`
	Status     string  `json:"status"`
	ErrorCount int     `json:"error_count"`
	Tokens     int64   `json:"tokens"`
	Cached     int64   `json:"cached_tokens"`
	CostUSD    float64 `json:"cost_usd"`
	ToolCalls  int     `json:"tool_calls"`
	LLMCalls   int     `json:"llm_calls"`
	Model      string  `json:"model"`
	Preview    string  `json:"preview"`

	// Partial marks a trace whose root span has not arrived. Claude Code emits
	// the root interaction span when the interaction ends, so a run still in
	// flight has every child span and no root. Saying so beats labelling the
	// run after whichever child happened to land first, which reads as a
	// finished trace with a strange name.
	Partial bool `json:"partial"`
}

// tokenAttrs and costAttrs are the attribute names emitters use for the numbers
// the list surfaces. Several are checked because there is no single convention:
// Claude Code uses input_tokens, the OTel GenAI conventions use
// gen_ai.usage.input_tokens, and OpenLLMetry uses llm.usage.*.
var tokenAttrs = []string{
	"input_tokens", "output_tokens",
	"gen_ai.usage.input_tokens", "gen_ai.usage.output_tokens",
	"llm.usage.total_tokens", "gen_ai.usage.total_tokens",
}

// Cache reads are counted separately, not folded into the token total.
//
// A real run measured 16 input and 1,301 output tokens against 280,054 cache
// reads. Summing them would report "281k tokens" and make every cached run look
// enormous; omitting them entirely reports "1.3k" and hides where the context
// actually came from. Both numbers are interesting, so both are kept.
var cacheAttrs = []string{
	"cache_read_tokens", "cache_creation_tokens",
	"gen_ai.usage.cache_read_input_tokens",
}

var costAttrs = []string{"cost_usd", "gen_ai.usage.cost"}

var modelAttrs = []string{"gen_ai.request.model", "model", "llm.request.model"}

// handleTraceList aggregates spans into per-trace summaries.
//
// This is a full scan of the requested range, by design. There is no trace
// index, because building one would mean a second source of truth to keep
// consistent with the shards — the failure mode being far worse than the scan
// it saves. The response reports rows_scanned so that claim stays checkable
// rather than aspirational.
func (s *Server) handleTraceList(w http.ResponseWriter, r *http.Request) {
	minTS, maxTS := timeRange(r)
	limit := intParam(r, "limit", 200)
	statusFilter := r.URL.Query().Get("status")
	search := strings.ToLower(strings.TrimSpace(r.URL.Query().Get("q")))

	summaries := map[string]*TraceSummary{}
	// rootSeen tracks whether a trace's true root span has been found, so a
	// later non-root span does not overwrite the name with a child's.
	rootSeen := map[string]bool{}

	scanned, err := s.store.ScanSpans(minTS, maxTS,
		func(sp model.Span, sh *store.Shard, c *store.Chunk, i int) bool {
			t := summaries[sp.TraceID]
			if t == nil {
				t = &TraceSummary{
					TraceID: sp.TraceID,
					Service: sp.Service,
					StartTS: sp.StartTS,
					EndTS:   sp.EndTS,
					Status:  "ok",
				}
				summaries[sp.TraceID] = t
			}
			mergeSpan(t, &sp, rootSeen)
			return true
		})
	if err != nil {
		httpError(w, http.StatusInternalServerError, "scanning spans: %v", err)
		return
	}

	// Cost lives on log records, not spans: Claude Code reports it on the
	// api_request event. A spans-only scan therefore shows nothing in the cost
	// column, which is how this was caught. Scanning logs for the handful of
	// cost-bearing events is cheap next to the span scan already done.
	logScan, err := s.store.ScanLogs(minTS, maxTS,
		func(rec model.LogRecord, sh *store.Shard, c *store.Chunk, i int) bool {
			t := summaries[rec.TraceID]
			if t == nil {
				return true
			}
			for _, k := range costAttrs {
				if v, ok := rec.Attrs[k]; ok {
					if f, ok := v.Float(); ok {
						t.CostUSD += f
					}
				}
			}
			if rec.EventName == "user_prompt" && t.Preview == "" {
				if v, ok := rec.Attrs["prompt"]; ok && v.Str != "" {
					t.Preview = truncateRunes(v.Str, 160)
				}
			}
			return true
		})
	if err != nil {
		httpError(w, http.StatusInternalServerError, "scanning logs: %v", err)
		return
	}

	out := make([]TraceSummary, 0, len(summaries))
	for id, t := range summaries {
		t.Partial = !rootSeen[id]
		t.DurMicros = t.EndTS - t.StartTS
		if t.DurMicros < 0 {
			t.DurMicros = 0
		}
		if statusFilter != "" && t.Status != statusFilter {
			continue
		}
		if search != "" && !matchesSearch(t, search) {
			continue
		}
		out = append(out, *t)
	}

	// Newest first: the trace you want is almost always the one that just ran.
	sort.Slice(out, func(i, j int) bool { return out[i].StartTS > out[j].StartTS })
	total := len(out)
	if limit > 0 && len(out) > limit {
		out = out[:limit]
	}

	writeJSON(w, map[string]any{
		"traces": out,
		"total":  total,
		"stats": map[string]any{
			"shards_scanned": scanned.Shards + logScan.Shards,
			"chunks_scanned": scanned.Chunks + logScan.Chunks,
			"rows_scanned":   scanned.Rows + logScan.Rows,
		},
	})
}

func matchesSearch(t *TraceSummary, q string) bool {
	hay := strings.ToLower(t.Name + " " + t.TraceID + " " + t.Model + " " + t.Preview)
	return strings.Contains(hay, q)
}

// mergeSpan folds one span into its trace's summary.
func mergeSpan(t *TraceSummary, sp *model.Span, rootSeen map[string]bool) {
	t.SpanCount++
	if sp.StartTS < t.StartTS {
		t.StartTS = sp.StartTS
	}
	if sp.EndTS > t.EndTS {
		t.EndTS = sp.EndTS
	}
	if sp.Status == model.StatusError {
		t.ErrorCount++
		t.Status = "error"
	}

	// The root span names the trace.
	if sp.ParentSpanID == "" && (!rootSeen[sp.TraceID] || sp.StartTS <= t.StartTS) {
		t.Name = sp.Name
		rootSeen[sp.TraceID] = true
	}

	switch sp.Kind {
	case model.KindToolCall:
		// Claude Code emits a `tool` span with `tool.execution` and
		// `tool.blocked_on_user` children. Counting all three would treat one
		// tool call as three, so only the outermost counts.
		if !strings.Contains(sp.Name, ".execution") && !strings.Contains(sp.Name, ".blocked_on_user") {
			t.ToolCalls++
		}
	case model.KindLLMCall:
		t.LLMCalls++
	}

	for _, k := range tokenAttrs {
		if v, ok := sp.Attrs[k]; ok {
			if f, ok := v.Float(); ok {
				t.Tokens += int64(f)
			}
		}
	}
	for _, k := range cacheAttrs {
		if v, ok := sp.Attrs[k]; ok {
			if f, ok := v.Float(); ok {
				t.Cached += int64(f)
			}
		}
	}
	for _, k := range costAttrs {
		if v, ok := sp.Attrs[k]; ok {
			if f, ok := v.Float(); ok {
				t.CostUSD += f
			}
		}
	}
	if t.Model == "" {
		for _, k := range modelAttrs {
			if v, ok := sp.Attrs[k]; ok && v.Str != "" {
				t.Model = v.Str
				break
			}
		}
	}
	// A short excerpt of the prompt makes the list scannable by what the run
	// was about rather than by opaque trace ids.
	if t.Preview == "" {
		if v, ok := sp.Attrs["user_prompt"]; ok && v.Str != "" {
			t.Preview = truncateRunes(v.Str, 140)
		}
	}
}

func truncateRunes(s string, n int) string {
	s = strings.Join(strings.Fields(s), " ")
	r := []rune(s)
	if len(r) <= n {
		return s
	}
	return string(r[:n]) + "…"
}

// handleTraceDetail rebuilds one trace: its span tree, its correlated log
// records, and the blob text for both.
//
// Blobs are resolved here rather than lazily per span because a trace is opened
// once and read whole; a round trip per span would make the UI feel worse than
// the extra bytes cost.
func (s *Server) handleTraceDetail(w http.ResponseWriter, r *http.Request) {
	id := r.PathValue("id")
	if id == "" {
		httpError(w, http.StatusBadRequest, "no trace id given")
		return
	}
	minTS, maxTS := traceLookupRange(r)

	var spans []model.Span
	spanScan, err := s.store.ScanSpans(minTS, maxTS,
		func(sp model.Span, sh *store.Shard, c *store.Chunk, i int) bool {
			if sp.TraceID != id {
				return true
			}
			if in, err := sh.Blob().Read(store.InputRef(c, i)); err == nil {
				sp.Input = in
			}
			if out, err := sh.Blob().Read(store.OutputRef(c, i)); err == nil {
				sp.Output = out
			}
			spans = append(spans, sp)
			return true
		})
	if err != nil {
		httpError(w, http.StatusInternalServerError, "scanning spans: %v", err)
		return
	}
	if len(spans) == 0 {
		httpError(w, http.StatusNotFound, "no trace %q in the range searched", id)
		return
	}

	var logs []model.LogRecord
	logScan, err := s.store.ScanLogs(minTS, maxTS,
		func(rec model.LogRecord, sh *store.Shard, c *store.Chunk, i int) bool {
			if rec.TraceID != id {
				return true
			}
			if body, err := sh.Blob().Read(store.BodyRef(c, i)); err == nil {
				rec.Body = body
			}
			logs = append(logs, rec)
			return true
		})
	if err != nil {
		httpError(w, http.StatusInternalServerError, "scanning logs: %v", err)
		return
	}

	tr := s.corr.Build(id, spans, logs)

	writeJSON(w, map[string]any{
		"trace":     tr,
		"flat":      tr.Flatten(),
		"log_count": len(logs),
		"stats": map[string]any{
			"shards_scanned": spanScan.Shards + logScan.Shards,
			"rows_scanned":   spanScan.Rows + logScan.Rows,
		},
	})
}

// traceLookupRange widens the default window for a detail lookup.
//
// A trace id carries no timestamp, so finding one means scanning. Defaulting to
// seven days rather than the list's one day means a link to a trace from
// yesterday still resolves, while still bounding the scan — which is the whole
// reason the time partition is the only index.
func traceLookupRange(r *http.Request) (int64, int64) {
	if r.URL.Query().Get("from") != "" || r.URL.Query().Get("to") != "" {
		return timeRange(r)
	}
	now := nowMicros()
	return now - 7*24*3600*1_000_000, now
}