From 741ac689f83cf529a8c37eab58236677d14945d1 Mon Sep 17 00:00:00 2001 From: Ollie Agent Date: Sat, 15 Aug 2026 20:12:21 +0200 Subject: [PATCH] Persist and expose aggregate usage metrics --- cmd/olliesrv/internal/agent/turn.go | 4 + cmd/olliesrv/internal/fs/spec.go | 79 ++++++++++- cmd/olliesrv/internal/metrics/metrics.go | 172 +++++++++++++++++++++++ 3 files changed, 254 insertions(+), 1 deletion(-) create mode 100644 cmd/olliesrv/internal/metrics/metrics.go diff --git a/cmd/olliesrv/internal/agent/turn.go b/cmd/olliesrv/internal/agent/turn.go index bcdbadb..011daf8 100644 --- a/cmd/olliesrv/internal/agent/turn.go +++ b/cmd/olliesrv/internal/agent/turn.go @@ -9,6 +9,7 @@ import ( "strings" "ollie/cmd/olliesrv/internal/backend" + "ollie/cmd/olliesrv/internal/metrics" "ollie/toolsrv" ) @@ -173,6 +174,9 @@ func (ag *Agent) executeTurn(ctx context.Context, input string) string { CostUSD: costUSD, } ag.history.addUsage(usage, est != 0) + if err := metrics.RecordUsage(ag.sessionID, ag.id, ag.BackendName(), ag.ModelName(), usage, est != 0); err != nil { + ag.log.Error("record metrics: %v", err) + } ag.log.Debug("usage: model=%s input=%d cached=%d cache_create=%d output=%d hit_ratio=%.3f cost=$%.6f", ag.runtime.Backend.Model(), in, cached, creation, out, ag.history.cacheHitRatio(), costUSD) ag.notifyChange() } diff --git a/cmd/olliesrv/internal/fs/spec.go b/cmd/olliesrv/internal/fs/spec.go index af30360..c595c6f 100644 --- a/cmd/olliesrv/internal/fs/spec.go +++ b/cmd/olliesrv/internal/fs/spec.go @@ -15,6 +15,7 @@ import ( "ollie/cmd/olliesrv/internal/agent" "ollie/cmd/olliesrv/internal/backend" + "ollie/cmd/olliesrv/internal/metrics" "ollie/cmd/olliesrv/internal/session" "ollie/paths" "ollie/virtfs" @@ -152,6 +153,33 @@ func buildTreeSpec(cfg *Config) virtfs.FsNodeDecl { return []byte(sb.String()), nil }), ), + virtfs.FileNode("stats", 0444, + virtfs.Read(func() ([]byte, error) { + a, err := metrics.AggregateAll() + if err != nil { + return nil, err + } + return []byte(metrics.Format(a)), nil + }), + ), + virtfs.FileNode("metrics", 0444, + virtfs.Read(func() ([]byte, error) { + a, err := metrics.AggregateAll() + if err != nil { + return nil, err + } + return []byte(metrics.Format(a)), nil + }), + ), + virtfs.FileNode("metrics.by-backend-model", 0444, + virtfs.Read(func() ([]byte, error) { + groups, err := metrics.GroupedAggregate("", "") + if err != nil { + return nil, err + } + return []byte(metrics.FormatGroups(groups)), nil + }), + ), // session/ virtfs.DirNode("session", @@ -245,6 +273,33 @@ func buildSessionChildren( return []byte("false\n"), nil }), ), + virtfs.FileNode("stats", 0444, + virtfs.Read(func() ([]byte, error) { + a, err := metrics.AggregateSession(s.ID) + if err != nil { + return nil, err + } + return []byte(metrics.Format(a)), nil + }), + ), + virtfs.FileNode("metrics", 0444, + virtfs.Read(func() ([]byte, error) { + a, err := metrics.AggregateSession(s.ID) + if err != nil { + return nil, err + } + return []byte(metrics.Format(a)), nil + }), + ), + virtfs.FileNode("metrics.by-backend-model", 0444, + virtfs.Read(func() ([]byte, error) { + groups, err := metrics.GroupedAggregate(s.ID, "") + if err != nil { + return nil, err + } + return []byte(metrics.FormatGroups(groups)), nil + }), + ), virtfs.FileNode("ctl", 0666, virtfs.Rdwr(func(_ context.Context, data []byte) ([]byte, error) { return dispatch(map[string]func([]string) ([]byte, error){ @@ -778,7 +833,29 @@ func buildAgentChildren(a *agent.Agent, s *session.Session) []virtfs.FsNodeDecl ), virtfs.FileNode("stats", 0444, virtfs.Read(func() ([]byte, error) { - return stats(nil) + a, err := metrics.AggregateAgent(s.ID, a.ID()) + if err != nil { + return nil, err + } + return []byte(metrics.Format(a)), nil + }), + ), + virtfs.FileNode("metrics", 0444, + virtfs.Read(func() ([]byte, error) { + a, err := metrics.AggregateAgent(s.ID, a.ID()) + if err != nil { + return nil, err + } + return []byte(metrics.Format(a)), nil + }), + ), + virtfs.FileNode("metrics.by-backend-model", 0444, + virtfs.Read(func() ([]byte, error) { + groups, err := metrics.GroupedAggregate(s.ID, a.ID()) + if err != nil { + return nil, err + } + return []byte(metrics.FormatGroups(groups)), nil }), ), virtfs.FileNode("name", 0666, diff --git a/cmd/olliesrv/internal/metrics/metrics.go b/cmd/olliesrv/internal/metrics/metrics.go new file mode 100644 index 0000000..932121f --- /dev/null +++ b/cmd/olliesrv/internal/metrics/metrics.go @@ -0,0 +1,172 @@ +package metrics + +import ( + "bufio" + "encoding/json" + "fmt" + "os" + "path/filepath" + "sort" + "strings" + "sync" + "time" + + "ollie/cmd/olliesrv/internal/backend" + "ollie/paths" +) + +// Record is one backend usage event. Records are append-only and independently +// queryable, so aggregates can be rebuilt without session state. +type Record struct { + Time time.Time `json:"time"` + SessionID string `json:"session"` + AgentID string `json:"agent"` + Backend string `json:"backend,omitempty"` + Model string `json:"model,omitempty"` + InputTokens int `json:"inputTokens"` + CachedInputTokens int `json:"cachedInputTokens"` + CacheCreationTokens int `json:"cacheCreationTokens"` + OutputTokens int `json:"outputTokens"` + CostUSD float64 `json:"costUSD"` + Estimated bool `json:"estimated,omitempty"` +} + +// Aggregate contains cumulative token, cache, request, and cost metrics. +type Aggregate struct { + InputTokens int + CachedInputTokens int + CacheCreationTokens int + OutputTokens int + Requests int + CostUSD float64 + Estimated bool +} + +func (a Aggregate) CacheHitRatio() float64 { + total := a.InputTokens + a.CachedInputTokens + if total == 0 { + return 0 + } + return float64(a.CachedInputTokens) / float64(total) +} + +func (a *Aggregate) add(r Record) { + a.InputTokens += r.InputTokens + a.CachedInputTokens += r.CachedInputTokens + a.CacheCreationTokens += r.CacheCreationTokens + a.OutputTokens += r.OutputTokens + a.Requests++ + a.CostUSD += r.CostUSD + a.Estimated = a.Estimated || r.Estimated +} + +var appendMu sync.Mutex + +// EventsPath returns the append-only metrics event log path. +func EventsPath() string { return filepath.Join(paths.DataDir(), "metrics", "events.jsonl") } + +// RecordUsage appends one usage event to the global metrics log. +func RecordUsage(sessionID, agentID, backendName, model string, u backend.Usage, estimated bool) error { + r := Record{Time: time.Now().UTC(), SessionID: sessionID, AgentID: agentID, Backend: backendName, Model: model, InputTokens: u.InputTokens, CachedInputTokens: u.CachedInputTokens, CacheCreationTokens: u.CacheCreationTokens, OutputTokens: u.OutputTokens, CostUSD: u.CostUSD, Estimated: estimated} + data, err := json.Marshal(r) + if err != nil { + return err + } + appendMu.Lock() + defer appendMu.Unlock() + path := EventsPath() + if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil { + return err + } + f, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0600) + if err != nil { + return err + } + defer f.Close() + _, err = fmt.Fprintf(f, "%s\n", data) + return err +} + +func AggregateAll() (Aggregate, error) { return aggregate(func(Record) bool { return true }) } +func AggregateSession(sessionID string) (Aggregate, error) { + return aggregate(func(r Record) bool { return r.SessionID == sessionID }) +} +func AggregateAgent(sessionID, agentID string) (Aggregate, error) { + return aggregate(func(r Record) bool { return r.SessionID == sessionID && r.AgentID == agentID }) +} + +func aggregate(match func(Record) bool) (Aggregate, error) { + f, err := os.Open(EventsPath()) + if os.IsNotExist(err) { + return Aggregate{}, nil + } + if err != nil { + return Aggregate{}, err + } + defer f.Close() + var out Aggregate + scanner := bufio.NewScanner(f) + for scanner.Scan() { + var r Record + if json.Unmarshal(scanner.Bytes(), &r) == nil && match(r) { + out.add(r) + } + } + if err := scanner.Err(); err != nil { + return Aggregate{}, err + } + return out, nil +} + +// GroupedAggregate returns aggregates grouped by backend and model. +func GroupedAggregate(sessionID, agentID string) (map[string]Aggregate, error) { + f, err := os.Open(EventsPath()) + if os.IsNotExist(err) { + return map[string]Aggregate{}, nil + } + if err != nil { + return nil, err + } + defer f.Close() + groups := make(map[string]Aggregate) + scanner := bufio.NewScanner(f) + for scanner.Scan() { + var r Record + if json.Unmarshal(scanner.Bytes(), &r) != nil { + continue + } + if (sessionID != "" && r.SessionID != sessionID) || (agentID != "" && r.AgentID != agentID) { + continue + } + key := r.Backend + "\t" + r.Model + a := groups[key] + a.add(r) + groups[key] = a + } + if err := scanner.Err(); err != nil { + return nil, err + } + return groups, nil +} + +// Format renders an aggregate as stable key=value lines for the 9P API. +func Format(a Aggregate) string { + return fmt.Sprintf("inputTokens=%d\ncachedInputTokens=%d\ncacheCreationTokens=%d\noutputTokens=%d\nrequests=%d\ncostUSD=%.6f\ncacheHitRatio=%g\nestimated=%t\n", a.InputTokens, a.CachedInputTokens, a.CacheCreationTokens, a.OutputTokens, a.Requests, a.CostUSD, a.CacheHitRatio(), a.Estimated) +} + +// FormatGroups renders backend/model aggregates as tab-separated records. +func FormatGroups(groups map[string]Aggregate) string { + keys := make([]string, 0, len(groups)) + for key := range groups { + keys = append(keys, key) + } + sort.Strings(keys) + var sb strings.Builder + sb.WriteString("backend\tmodel\tinputTokens\tcachedInputTokens\tcacheCreationTokens\toutputTokens\trequests\tcostUSD\tcacheHitRatio\testimated\n") + for _, key := range keys { + parts := strings.SplitN(key, "\t", 2) + a := groups[key] + fmt.Fprintf(&sb, "%s\t%s\t%d\t%d\t%d\t%d\t%d\t%.6f\t%g\t%t\n", parts[0], parts[1], a.InputTokens, a.CachedInputTokens, a.CacheCreationTokens, a.OutputTokens, a.Requests, a.CostUSD, a.CacheHitRatio(), a.Estimated) + } + return sb.String() +}