返回 DeepSeek-Reasonix
record.go
根目录 / internal / stats / record.go
1 // Package stats records per-call token usage as append-only daily JSONL files
2 // under the user state root (config.StatsDir), and aggregates them for the
3 // desktop "usage statistics" panel.
4 //
5 // Design notes:
6 // - Only provider usage (including request-only failures) and turn
7 // completions (event.TurnDone) are recorded here. Turn markers power the
8 // panel's "completed turns" metric;
9 // they are deliberately not presented as distinct conversation sessions.
10 // Token usage was never persisted before this feature, so token numbers
11 // accumulate from the day the feature ships.
12 // - Files are append-only: each record is one JSON line appended with
13 // O_APPEND. A crash mid-line leaves at most one torn trailing line, which
14 // decodeRecords tolerates and skips.
15 package stats
16
17 import (
18 "bufio"
19 "context"
20 "crypto/sha256"
21 "encoding/json"
22 "errors"
23 "io"
24 "os"
25 "path/filepath"
26 "strings"
27 "time"
28
29 filelock "reasonix/internal/identitylock"
30 "reasonix/internal/usagecatalog"
31 )
32
33 // dayLayout names one stats file per UTC-free local day, e.g. 2026-08-02.jsonl.
34 const dayLayout = "2006-01-02"
35
36 const appendLockTimeout = 2 * time.Second
37
38 // record is one line in a daily stats file. TurnDone marks a completed turn so
39 // per-day turn counts are available without touching session files.
40 type record struct {
41 Timestamp time.Time `json:"ts"`
42 ModelRef string `json:"model,omitempty"` // canonical "provider/model"
43 Source string `json:"source,omitempty"` // desktop | cli | serve | bot | remote
44 Prompt int `json:"prompt,omitempty"`
45 Completion int `json:"completion,omitempty"`
46 Reasoning int `json:"reasoning,omitempty"`
47 CacheHit int `json:"cache_hit,omitempty"`
48 CacheMiss int `json:"cache_miss,omitempty"`
49 Total int `json:"total,omitempty"`
50 Requests int `json:"requests,omitempty"` // provider requests represented by this row
51 Turn bool `json:"turn,omitempty"` // true for TurnDone marker rows
52 // Cost quote fields (additive; older readers ignore them).
53 UsageSource string `json:"usage_source,omitempty"`
54 CostAmount string `json:"cost_amount,omitempty"` // original amount decimal
55 CostCurrency string `json:"cost_currency,omitempty"` // original ISO
56 SelectedAmount string `json:"selected_amount,omitempty"` // display valuation
57 SelectedCurrency string `json:"selected_currency,omitempty"`
58 CostComplete *bool `json:"cost_complete,omitempty"`
59 DisplayComplete *bool `json:"display_complete,omitempty"`
60 DisplayStatus string `json:"display_status,omitempty"`
61 AggregateMode string `json:"aggregate_mode,omitempty"`
62 OriginalTotals []string `json:"original_totals,omitempty"`
63 CostEstimated bool `json:"cost_estimated,omitempty"`
64 LegacyEstimate bool `json:"legacy_estimate,omitempty"`
65 PricingFingerprint string `json:"pricing_fingerprint,omitempty"`
66 RateDate string `json:"rate_date,omitempty"`
67 RateBand string `json:"rate_band,omitempty"`
68 RatedAt string `json:"rated_at,omitempty"`
69 IncompleteReason string `json:"incomplete_reason,omitempty"`
70 BillingMode string `json:"billing_mode,omitempty"`
71 // ValuationCNY/USD amounts when present (occurrence-time).
72 ValuationCNY string `json:"valuation_cny,omitempty"`
73 ValuationUSD string `json:"valuation_usd,omitempty"`
74 // SelectedCost is a float compatibility mirror of SelectedAmount.
75 SelectedCost float64 `json:"selected_cost,omitempty"`
76 }
77
78 // Writer appends records to the daily stats file for a given stats dir.
79 type Writer struct {
80 dir string
81 usage *usageManager
82 }
83
84 // NewWriter returns a Writer rooted at dir. An empty dir disables recording
85 // (query-only usage).
86 func NewWriter(dir string) *Writer {
87 dir = strings.TrimSpace(dir)
88 return &Writer{dir: dir}
89 }
90
91 // Append writes one record, appending to the daily file (O_APPEND) so records
92 // from concurrent turns never overwrite each other. Each record is a single
93 // JSON line; a crash mid-line leaves at most one torn trailing line, which
94 // decodeRecords tolerates.
95 func (w *Writer) Append(r record) error {
96 if w == nil || w.dir == "" {
97 return nil
98 }
99 day := r.Timestamp.Format(dayLayout)
100 path := filepath.Join(w.dir, day+".jsonl")
101 b, err := json.Marshal(r)
102 if err != nil {
103 return err
104 }
105 if err := os.MkdirAll(w.dir, 0o700); err != nil {
106 return err
107 }
108 ctx, cancel := context.WithTimeout(context.Background(), appendLockTimeout)
109 defer cancel()
110 release, err := filelock.Acquire(ctx, filepath.Join(w.dir, ".append.lock"))
111 if err != nil {
112 return err
113 }
114 released := false
115 defer func() {
116 if !released {
117 release()
118 }
119 }()
120 f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR|os.O_APPEND, 0o600)
121 if err != nil {
122 return err
123 }
124 if err := ensureRecordBoundary(f); err != nil {
125 _ = f.Close()
126 return err
127 }
128 info, err := f.Stat()
129 if err != nil {
130 _ = f.Close()
131 return err
132 }
133 offset := info.Size()
134 line := append(b, '\n')
135 if _, err = f.Write(line); err != nil {
136 _ = f.Close()
137 return err
138 }
139 if err := f.Close(); err != nil {
140 return err
141 }
142 release()
143 released = true
144 if w.usage != nil {
145 if catalog := w.usage.catalog.Load(); catalog != nil {
146 hash := sha256.Sum256(b)
147 catalog.Enqueue(usagecatalog.AppendReceipt{Path: path, Day: day, Offset: offset, Length: len(line), LineHash: fmtHash(hash[:])}, usageEntry(day, r))
148 }
149 }
150 return nil
151 }
152
153 func fmtHash(hash []byte) string {
154 const digits = "0123456789abcdef"
155 out := make([]byte, len(hash)*2)
156 for i, value := range hash {
157 out[i*2] = digits[value>>4]
158 out[i*2+1] = digits[value&15]
159 }
160 return string(out)
161 }
162
163 // ensureRecordBoundary separates a torn trailing JSON object from the next
164 // append. The caller holds the cross-process append lock, so checking the last
165 // byte and repairing it cannot race another Reasonix writer.
166 func ensureRecordBoundary(f *os.File) error {
167 st, err := f.Stat()
168 if err != nil || st.Size() == 0 {
169 return err
170 }
171 var tail [1]byte
172 if _, err := f.ReadAt(tail[:], st.Size()-1); err != nil {
173 return err
174 }
175 if tail[0] == '\n' {
176 return nil
177 }
178 _, err = f.Write([]byte{'\n'})
179 return err
180 }
181
182 // readDaily loads one daily file into records. Missing files yield nil, nil.
183 func readDaily(dir, day string) ([]record, error) {
184 if strings.TrimSpace(dir) == "" {
185 return nil, nil
186 }
187 f, err := os.Open(filepath.Join(dir, day+".jsonl"))
188 if errors.Is(err, os.ErrNotExist) {
189 return nil, nil
190 }
191 if err != nil {
192 return nil, err
193 }
194 defer f.Close()
195 return decodeRecords(f)
196 }
197
198 // readDailyRange snapshots the available daily files with one directory scan,
199 // then reads only dates requested by the query. Long custom ranges are often
200 // mostly empty; avoiding one failed os.Open per absent day keeps their cost
201 // proportional to the data that actually exists.
202 func readDailyRange(dir string, days []string) (map[string][]record, error) {
203 out := make(map[string][]record)
204 if strings.TrimSpace(dir) == "" || len(days) == 0 {
205 return out, nil
206 }
207 wanted := make(map[string]struct{}, len(days))
208 for _, day := range days {
209 wanted[day] = struct{}{}
210 }
211 entries, err := os.ReadDir(dir)
212 if errors.Is(err, os.ErrNotExist) {
213 return out, nil
214 }
215 if err != nil {
216 return nil, err
217 }
218 for _, entry := range entries {
219 if entry.IsDir() {
220 continue
221 }
222 name := entry.Name()
223 if !strings.HasSuffix(name, ".jsonl") {
224 continue
225 }
226 day := strings.TrimSuffix(name, ".jsonl")
227 if _, ok := wanted[day]; !ok {
228 continue
229 }
230 records, err := readDaily(dir, day)
231 if err != nil {
232 return nil, err
233 }
234 out[day] = records
235 }
236 return out, nil
237 }
238
239 func decodeRecords(r io.Reader) ([]record, error) {
240 var out []record
241 sc := bufio.NewScanner(r)
242 sc.Buffer(make([]byte, 0, 64*1024), 1024*1024)
243 for sc.Scan() {
244 line := strings.TrimSpace(sc.Text())
245 if line == "" {
246 continue
247 }
248 var rec record
249 if err := json.Unmarshal([]byte(line), &rec); err != nil {
250 // Malformed lines (a crash mid-write or a manual edit) are skipped
251 // rather than failing the whole day's aggregation. This tolerates
252 // any number of bad lines; a fully corrupt file reads as an empty
253 // day, which is preferable to the panel erroring out.
254 continue
255 }
256 out = append(out, rec)
257 }
258 return out, sc.Err()
259 }
260
261 // daysInRange lists the daily file names (without extension) whose timestamps
262 // intersect [from, to], inclusive.
263 func daysInRange(from, to time.Time) []string {
264 from = dayStart(from)
265 to = dayStart(to)
266 if to.Before(from) {
267 return nil
268 }
269 var days []string
270 for d := from; !d.After(to); d = d.AddDate(0, 0, 1) {
271 days = append(days, d.Format(dayLayout))
272 }
273 return days
274 }
275
276 func dayStart(t time.Time) time.Time {
277 y, m, d := t.Date()
278 return time.Date(y, m, d, 0, 0, 0, 0, t.Location())
279 }
280
280 lines GO