fathom/internal/insights/analyzer.go

package insights

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"sync"
	"time"

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

// Analyzer runs the detectors in the background and publishes the latest result.
//
// It is deliberately a reader. It scans the store through the same public API
// the UI uses, holds its findings in memory, and writes nothing to the data
// directory except the reader's own dismissals. Nothing else in Fathom depends
// on it, and switching it off removes a tab and changes nothing else.
//
// Results are recomputed rather than persisted because recomputing is
// milliseconds and a cached analysis is a second source of truth that can
// disagree with the traces it claims to describe.
type Analyzer struct {
	store     *store.Store
	detectors []Detector

	// WindowDays bounds how far back an analysis looks. Habits worth capturing
	// are recent by definition; a pattern from six weeks ago is history.
	WindowDays int

	// Interval is how often the analysis is refreshed when traces are arriving.
	Interval time.Duration

	now func() time.Time

	mu      sync.RWMutex
	latest  Result
	hasRun  bool
	running bool

	dismissPath string
	dismissed   map[string]int64 // insight id -> when dismissed

	// local is a model server found on this machine, if any. Nil is the
	// ordinary case and every feature that would use it degrades silently.
	local    Labeler
	embedder Embedder
	judge    Judge
	status   LocalStatus

	surfacedPath string
	surfaced     *surfacedState

	crystallize *Crystallize

	trigger chan struct{}
	stop    chan struct{}
	stopped sync.Once
	wg      sync.WaitGroup
}

// Options configure an Analyzer.
type Options struct {
	Store      *store.Store
	DataDir    string
	WindowDays int
	Interval   time.Duration
	Now        func() time.Time

	// Local is an optional model server on this machine.
	Local    Labeler
	Embedder Embedder
	Judge    Judge
	Status   LocalStatus
}

// Labeler names a cluster from examples.
type Labeler interface {
	Label(ctx context.Context, prompts []string) (string, error)
}

// LocalStatus describes what was found on this machine, for the UI.
//
// Reported rather than hidden: a finding that came from semantic clustering can
// differ from one produced on a machine with no model, and a reader should be
// able to see why rather than wonder.
type LocalStatus struct {
	Available  bool   `json:"available"`
	Enabled    bool   `json:"enabled"`
	Endpoint   string `json:"endpoint,omitempty"`
	Kind       string `json:"kind,omitempty"`
	EmbedModel string `json:"embed_model,omitempty"`
	ChatModel  string `json:"chat_model,omitempty"`

	// UsedFor says what the last analysis actually used it for, which is not
	// the same as what is available: a model can be present and still fail.
	UsedFor []string `json:"used_for,omitempty"`
}

func New(opts Options) *Analyzer {
	if opts.WindowDays <= 0 {
		opts.WindowDays = 7
	}
	if opts.Interval <= 0 {
		opts.Interval = 2 * time.Minute
	}
	if opts.Now == nil {
		opts.Now = time.Now
	}
	crystallize := NewCrystallize()

	a := &Analyzer{
		store:      opts.Store,
		WindowDays: opts.WindowDays,
		Interval:   opts.Interval,
		now:        opts.Now,
		local:      opts.Local,
		embedder:   opts.Embedder,
		judge:      opts.Judge,
		status:     opts.Status,
		detectors: []Detector{
			crystallize,
			NewContextWeight(),
			NewWaste(),
		},
		crystallize: crystallize,
		dismissed:   map[string]int64{},
		trigger:     make(chan struct{}, 1),
		stop:        make(chan struct{}),
	}
	a.status.Enabled = opts.Status.Available
	a.applyLocal()
	if opts.DataDir != "" {
		a.dismissPath = filepath.Join(opts.DataDir, "insights-dismissed.json")
		a.surfacedPath = filepath.Join(opts.DataDir, "insights-surfaced.json")
		a.loadDismissed()
	}
	return a
}

// Start begins background analysis.
func (a *Analyzer) Start() {
	a.wg.Add(1)
	go a.loop()
}

func (a *Analyzer) Close() {
	a.stopped.Do(func() { close(a.stop) })
	a.wg.Wait()
}

// Notify asks for a refresh soon. Wired to the store's flush hook so an
// analysis follows new traces rather than a clock.
//
// Non-blocking and coalescing: a burst of flushes schedules one run, and a run
// already in flight is not queued behind another.
func (a *Analyzer) Notify() {
	select {
	case a.trigger <- struct{}{}:
	default:
	}
}

func (a *Analyzer) loop() {
	defer a.wg.Done()

	// An initial pass so the tab has something the first time it is opened.
	a.Refresh()

	// The ticker is a floor, not the primary trigger: it catches the case where
	// traces stopped arriving but the window has moved on and old findings have
	// aged out.
	tick := time.NewTicker(a.Interval)
	defer tick.Stop()

	// Debounce flush notifications. Ingest flushes about once a second while an
	// agent is working, and re-analysing on each one would burn CPU to produce
	// the same answer.
	var debounce <-chan time.Time
	for {
		select {
		case <-a.stop:
			return
		case <-a.trigger:
			if debounce == nil {
				debounce = time.After(10 * time.Second)
			}
		case <-debounce:
			debounce = nil
			a.Refresh()
		case <-tick.C:
			a.Refresh()
		}
	}
}

// Refresh runs the detectors once, synchronously, over the analyzer's default
// window, and publishes the result for Latest.
func (a *Analyzer) Refresh() {
	a.mu.Lock()
	if a.running {
		a.mu.Unlock()
		return
	}
	a.running = true
	a.mu.Unlock()

	defer func() {
		a.mu.Lock()
		a.running = false
		a.mu.Unlock()
	}()

	now := a.now()
	to := now.UnixMicro()
	from := now.AddDate(0, 0, -a.WindowDays).UnixMicro()

	res, err := a.analyze(from, to)
	if err != nil {
		return // leave the previous result in place rather than blanking the tab
	}

	a.mu.Lock()
	a.latest = res
	a.hasRun = true
	a.mu.Unlock()
}

// AnalyzeRange runs the detectors once over an explicit window and returns the
// result directly, without touching the cached background result that Latest
// and the surfacing/hook path read.
//
// This is what lets the UI's time-range picker apply to the Insights tab: the
// background Analyzer keeps its own fixed window (recent habits, checked on a
// schedule), while a request for a specific range gets its own on-demand
// analysis. Recomputing is milliseconds, so there is no need to cache it.
func (a *Analyzer) AnalyzeRange(from, to int64) (Result, error) {
	return a.analyze(from, to)
}

// analyze is the shared core behind Refresh and AnalyzeRange: extract facts,
// run every detector, drop what has been dismissed and not recurred, and group
// the rest for display.
func (a *Analyzer) analyze(from, to int64) (Result, error) {
	traces, err := a.extract(from, to)
	if err != nil {
		return Result{}, err
	}

	var all []Insight
	for _, d := range a.detectors {
		all = append(all, d.Detect(traces)...)
	}

	a.mu.RLock()
	dismissed := make(map[string]int64, len(a.dismissed))
	for k, v := range a.dismissed {
		dismissed[k] = v
	}
	a.mu.RUnlock()

	kept := all[:0]
	for _, in := range all {
		// A dismissal applies until the pattern recurs after it. Suppressing
		// forever would mean a habit you consciously accepted last month stays
		// invisible when it comes back at ten times the cost.
		if when, ok := dismissed[in.ID]; ok && in.LastSeen <= when {
			continue
		}
		kept = append(kept, in)
	}

	res := group(kept, Window{From: from, To: to}, traces, to)
	res.MinOccurrences = a.crystallize.MinRuns
	res.Dismissed = len(all) - len(kept)

	used := []string{}
	if a.crystallize.LastMethod() == MethodEmbedding {
		used = append(used, "clustering")
	}
	if a.enabled() && a.local != nil {
		if n := a.labelFindings(&res); n > 0 {
			used = append(used, "naming")
		}
	}

	a.mu.RLock()
	status := a.status
	status.UsedFor = used
	a.mu.RUnlock()
	res.Local = status

	return res, nil
}

// labelFindings asks the local model to name clustered findings.
//
// Strictly additive: the model sees prompts that are already grouped and writes
// a phrase. It is never given a number, never asked to decide what is a
// finding, and a failure leaves the heuristic title in place. Returns how many
// labels were applied.
func (a *Analyzer) labelFindings(res *Result) int {
	ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second)
	defer cancel()

	applied := 0
	for ti := range res.Themes {
		for ii := range res.Themes[ti].Insights {
			in := &res.Themes[ti].Insights[ii]
			if !strings.HasPrefix(in.ID, "crystallize/ask/") || len(in.Evidence) < 2 {
				continue
			}
			var prompts []string
			for _, e := range in.Evidence {
				if e.Prompt != "" {
					prompts = append(prompts, e.Prompt)
				}
			}
			if len(prompts) < 2 {
				continue
			}
			label, err := a.local.Label(ctx, prompts)
			if err != nil || label == "" {
				continue // keep the heuristic title
			}
			in.Label = label
			in.Title = fmt.Sprintf("%s — asked %d times", label, in.Occurrences)
			applied++
		}
	}
	return applied
}

// applyLocal wires the embedder into the detector that can use it.
func (a *Analyzer) applyLocal() {
	if a.status.Enabled {
		a.crystallize.Embedder = a.embedder
		return
	}
	a.crystallize.Embedder = nil
}

func (a *Analyzer) enabled() bool {
	a.mu.RLock()
	defer a.mu.RUnlock()
	return a.status.Enabled
}

// SetLocalEnabled turns local-model use on or off and re-analyses.
func (a *Analyzer) SetLocalEnabled(on bool) {
	a.mu.Lock()
	a.status.Enabled = on && a.status.Available
	a.mu.Unlock()
	a.applyLocal()
	a.Refresh()
}

// Status returns what was found on this machine.
func (a *Analyzer) Status() LocalStatus {
	a.mu.RLock()
	defer a.mu.RUnlock()
	return a.status
}

// Latest returns the most recent analysis and whether one has completed.
func (a *Analyzer) Latest() (Result, bool) {
	a.mu.RLock()
	defer a.mu.RUnlock()
	return a.latest, a.hasRun
}

// Dismiss hides an insight until the underlying pattern recurs.
func (a *Analyzer) Dismiss(id string) {
	a.mu.Lock()
	a.dismissed[id] = a.now().UnixMicro()
	a.mu.Unlock()
	a.saveDismissed()
	a.Refresh()
}

// Restore un-dismisses everything.
func (a *Analyzer) Restore() {
	a.mu.Lock()
	a.dismissed = map[string]int64{}
	a.mu.Unlock()
	a.saveDismissed()
	a.Refresh()
}

func (a *Analyzer) loadDismissed() {
	b, err := os.ReadFile(a.dismissPath)
	if err != nil {
		return
	}
	var m map[string]int64
	if err := json.Unmarshal(b, &m); err != nil {
		return // a corrupt dismissal file costs the dismissals, nothing else
	}
	a.dismissed = m
}

func (a *Analyzer) saveDismissed() {
	if a.dismissPath == "" {
		return
	}
	a.mu.RLock()
	b, err := json.MarshalIndent(a.dismissed, "", "  ")
	a.mu.RUnlock()
	if err != nil {
		return
	}
	// Write and rename, so an interrupted save cannot leave a half-written file
	// that fails to parse on the next start.
	tmp := a.dismissPath + ".tmp"
	if err := os.WriteFile(tmp, b, 0o644); err != nil {
		return
	}
	os.Rename(tmp, a.dismissPath)
}

// extract turns stored spans and logs into per-trace facts.
//
// One pass over each table, then correlation per trace, so every detector reads
// the same prepared view and none of them touch the store.
func (a *Analyzer) extract(from, to int64) ([]TraceFacts, error) {
	spansByTrace := map[string][]model.Span{}
	if _, err := a.store.ScanSpans(from, to, func(sp model.Span, sh *store.Shard, c *store.Chunk, i int) bool {
		spansByTrace[sp.TraceID] = append(spansByTrace[sp.TraceID], sp)
		return true
	}); err != nil {
		return nil, err
	}

	logsByTrace := map[string][]model.LogRecord{}
	if _, err := a.store.ScanLogs(from, to, func(rec model.LogRecord, sh *store.Shard, c *store.Chunk, i int) bool {
		logsByTrace[rec.TraceID] = append(logsByTrace[rec.TraceID], rec)
		return true
	}); err != nil {
		return nil, err
	}

	corr := tracetree.NewCorrelator()
	out := make([]TraceFacts, 0, len(spansByTrace))
	for id, spans := range spansByTrace {
		out = append(out, buildFacts(corr, id, spans, logsByTrace[id]))
	}
	sort.SliceStable(out, func(i, j int) bool { return out[i].StartTS < out[j].StartTS })
	return out, nil
}

// buildFacts flattens one correlated trace into the detectors' input.
func buildFacts(corr *tracetree.Correlator, id string, spans []model.Span, logs []model.LogRecord) TraceFacts {
	tr := corr.Build(id, spans, logs)

	f := TraceFacts{
		TraceID:   id,
		StartTS:   tr.StartTS,
		EndTS:     tr.EndTS,
		SpanCount: tr.SpanCount,
	}

	for _, n := range tr.Flatten() {
		sp := n.Span

		if f.Prompt == "" {
			f.Prompt = firstStr(sp.Attrs, promptKeys)
		}
		if f.Model == "" {
			f.Model = firstStr(sp.Attrs, modelKeys)
		}

		switch sp.Kind {
		case model.KindLLMCall:
			f.LLMCalls = append(f.LLMCalls, llmCallFrom(sp, n.Logs))

		case model.KindToolCall:
			// Only the outermost span of a tool call. Claude Code emits
			// `tool`, `tool.execution`, and `tool.blocked_on_user` for one
			// logical call, and counting all three would treble every sequence.
			if isWrapperSpan(sp.Name) {
				continue
			}
			f.Calls = append(f.Calls, toolCallFrom(sp, n.Logs))
		}

		f.CostUSD += costOf(sp, n.Logs)

		for _, l := range n.Logs {
			if f.Prompt == "" && isPromptEvent(l.EventName) {
				f.Prompt = firstStr(l.Attrs, promptKeys)
			}
		}
		if f.Prompt == "" {
			for _, l := range n.Logs {
				if isPromptEvent(l.EventName) && l.Body != "" {
					f.Prompt = l.Body
					break
				}
			}
		}

		f.Tokens += firstNum(sp.Attrs, inputTokenKeys) + firstNum(sp.Attrs, outputTokenKeys)
		f.Cached += firstNum(sp.Attrs, cacheReadKeys) + firstNum(sp.Attrs, cacheWriteKeys)
	}

	sort.SliceStable(f.LLMCalls, func(i, j int) bool { return i < j })
	return f
}

func isWrapperSpan(name string) bool {
	return len(name) > 0 &&
		(hasSuffix(name, ".execution") || hasSuffix(name, ".blocked_on_user"))
}

func hasSuffix(s, suffix string) bool {
	return len(s) >= len(suffix) && s[len(s)-len(suffix):] == suffix
}

func toolCallFrom(sp model.Span, logs []model.LogRecord) ToolCall {
	c := ToolCall{
		Name:      "tool",
		Succeeded: sp.Status != model.StatusError,
		DurMicros: sp.DurationMicros(),
	}
	if n := firstStr(sp.Attrs, toolNameKeys); n != "" {
		c.Name = n
	}

	// The arguments live on the correlated tool_result record, not the span —
	// which is the same reason the trace view needs log correlation at all.
	for _, l := range logs {
		if l.EventName != "tool_result" && l.EventName != "tool_decision" {
			continue
		}
		if n := firstStr(l.Attrs, toolNameKeys); n != "" {
			c.Name = n
		}
		for _, k := range []string{"tool_input", "tool_parameters"} {
			if v, ok := l.Attrs[k]; ok && v.Str != "" && c.Input == "" {
				c.Input = v.Str
			}
		}
		if v, ok := l.Attrs["tool_result_size_bytes"]; ok {
			if n, ok := v.Float(); ok && int64(n) > c.OutputSize {
				c.OutputSize = int64(n)
			}
		}
		if v, ok := l.Attrs["success"]; ok && v.Str == "false" {
			c.Succeeded = false
		}
	}
	if c.OutputSize == 0 {
		for _, l := range logs {
			if l.EventName == "tool_result" && l.Body != "" {
				c.OutputSize = int64(len(l.Body))
			}
		}
	}

	c.Signature, c.Target = normalizeCall(c.Name, c.Input)
	return c
}

func llmCallFrom(sp model.Span, logs []model.LogRecord) LLMCall {
	c := LLMCall{DurMicros: sp.DurationMicros()}
	c.InputTokens = firstNum(sp.Attrs, inputTokenKeys)
	c.OutputToken = firstNum(sp.Attrs, outputTokenKeys)
	c.CacheRead = firstNum(sp.Attrs, cacheReadKeys)
	c.Model = firstStr(sp.Attrs, modelKeys)
	c.StopReason = firstStr(sp.Attrs, stopReasonKeys)
	c.CostUSD = costOf(sp, logs)
	return c
}

// DebugExtract exposes fact extraction for diagnostics. It exists so the
// extraction step can be inspected against a real store without standing up an
// analyzer, and is not used by the server.
func DebugExtract(s *store.Store, windowDays int) ([]TraceFacts, error) {
	a := New(Options{Store: s, WindowDays: windowDays})
	now := a.now()
	return a.extract(now.AddDate(0, 0, -windowDays).UnixMicro(), now.UnixMicro())
}