git.giacomolli.uk - onechat/blob - build_run.go

   1 package main
   2
   3 import (
   4         "bufio"
   5         "cmp"
   6         "context"
   7         "encoding/json"
   8         "errors"
   9         "fmt"
  10         "io"
  11         "maps"
  12         "os"
  13         "slices"
  14         "strconv"
  15         "strings"
  16         "time"
  17 )
  18
  19 // buildLineBytes is the most bytes of one line in a build input or usage log.
  20 const buildLineBytes = 1 << 20
  21
  22 // buildWaitTimeout is the longest wait for pending memory builds.
  23 const buildWaitTimeout = time.Second
  24
  25 // buildQuotaLimit is the largest increase in used quota points of a run.
  26 const buildQuotaLimit = 15.0
  27
  28 // BuildRun appends messages, builds their tree, and stops the run by the rules
  29 // of buildStop.
  30 type BuildRun struct {
  31         memory    *Memory
  32         client    *ModelClient
  33         usagePath string
  34 }
  35
  36 // DatedMessage is one message with an optional RFC 3339 date.
  37 type DatedMessage struct {
  38         Kind string `json:"kind"`
  39         Text string `json:"text"`
  40         Date string `json:"date,omitempty"`
  41 }
  42
  43 // Run appends the entries. It waits for the builds only before a user message,
  44 // as a turn does, so up to buildMax builds run at once in between, and it waits
  45 // for all builds at the end. It returns the number of messages appended and why
  46 // the run stopped early, or "" when it did not. After a stop it waits for the
  47 // calls in flight, so no sent call is cancelled.
  48 func (r *BuildRun) Run(ctx context.Context, entries []DatedMessage) (int, string) {
  49         appended, reason := r.append(ctx, entries)
  50         if reason == "" {
  51                 return appended, ""
  52         }
  53
  54         err := r.memory.Wait(ctx)
  55         if err != nil {
  56                 reason += "; " + err.Error()
  57         }
  58
  59         return appended, reason
  60 }
  61
  62 // append appends the entries and stops at the first stop rule.
  63 func (r *BuildRun) append(ctx context.Context, entries []DatedMessage) (int, string) {
  64         for n, entry := range entries {
  65                 if entry.Kind == "user" {
  66                         reason := r.wait(ctx)
  67                         if reason != "" {
  68                                 return n, reason
  69                         }
  70                 }
  71
  72                 if ctx.Err() != nil {
  73                         return n, ctx.Err().Error()
  74                 }
  75
  76                 _, err := r.memory.AppendDated(entry.Kind, entry.Text, entry.Date)
  77                 if err != nil {
  78                         return n, err.Error()
  79                 }
  80
  81                 reason := r.check()
  82                 if reason != "" {
  83                         return n + 1, reason
  84                 }
  85         }
  86
  87         return len(entries), r.wait(ctx)
  88 }
  89
  90 // wait returns when no build runs, or when a stop rule ends the run. It checks
  91 // the rules each second, so a run stops soon also during a long wait.
  92 func (r *BuildRun) wait(ctx context.Context) string {
  93         for {
  94                 reason := r.check()
  95                 if reason != "" {
  96                         return reason
  97                 }
  98
  99                 err := r.waitBuilds(ctx)
 100                 if err == nil {
 101                         return r.check()
 102                 }
 103
 104                 if ctx.Err() != nil {
 105                         return ctx.Err().Error()
 106                 }
 107         }
 108 }
 109
 110 // check returns why the run must stop, or "" when it can go on.
 111 func (r *BuildRun) check() string {
 112         if r.client.LimitReached() {
 113                 return fmt.Sprintf("call limit of %d reached", r.client.CallLimit())
 114         }
 115
 116         records, err := buildReadUsage(r.usagePath)
 117         if err != nil {
 118                 return err.Error()
 119         }
 120
 121         return buildStop(records)
 122 }
 123
 124 // waitBuilds waits up to one second without leaving a child context open.
 125 func (r *BuildRun) waitBuilds(ctx context.Context) error {
 126         ctx, cancel := context.WithTimeout(ctx, buildWaitTimeout)
 127         defer cancel()
 128
 129         return r.memory.Wait(ctx)
 130 }
 131
 132 // buildReadUsage reads the records of usage.jsonl at path, in the order of the
 133 // start of their calls. The file has them in the order of the end of their
 134 // calls, which differs when calls run at once.
 135 func buildReadUsage(path string) (records []ModelUsageRecord, err error) {
 136         file, err := os.Open(path)
 137         if errors.Is(err, os.ErrNotExist) {
 138                 return nil, nil
 139         }
 140
 141         if err != nil {
 142                 return nil, fmt.Errorf("open usage file: %w", err)
 143         }
 144
 145         defer func() {
 146                 closeErr := file.Close()
 147                 if closeErr != nil {
 148                         err = errors.Join(err, fmt.Errorf("close build usage: %w", closeErr))
 149                 }
 150         }()
 151
 152         scanner := bufio.NewScanner(file)
 153         scanner.Buffer(make([]byte, 64*1024), buildLineBytes)
 154
 155         for scanner.Scan() {
 156                 var record ModelUsageRecord
 157
 158                 err = json.Unmarshal(scanner.Bytes(), &record)
 159                 if err != nil {
 160                         return nil, fmt.Errorf("decode usage: %w", err)
 161                 }
 162
 163                 records = append(records, record)
 164         }
 165
 166         err = scanner.Err()
 167         if err != nil {
 168                 return nil, fmt.Errorf("read usage file: %w", err)
 169         }
 170
 171         slices.SortStableFunc(records, func(a, b ModelUsageRecord) int {
 172                 return buildCallStart(a).Compare(buildCallStart(b))
 173         })
 174
 175         return records, nil
 176 }
 177
 178 // buildStop returns why a build run must stop: a failed call, or a quota window
 179 // whose used percent grew by more than buildQuotaLimit points since the first
 180 // call. It returns "" when the run can go on.
 181 func buildStop(records []ModelUsageRecord) string {
 182         for _, record := range records {
 183                 if record.Error != "" {
 184                         return "call failed: " + record.Error
 185                 }
 186         }
 187
 188         if len(records) == 0 {
 189                 return ""
 190         }
 191
 192         first := buildUsedPercents(records[0])
 193         for name, used := range buildUsedPercents(records[len(records)-1]) {
 194                 start, found := first[name]
 195                 if found && used-start > buildQuotaLimit {
 196                         return fmt.Sprintf("%s grew from %g to %g", name, start, used)
 197                 }
 198         }
 199
 200         return ""
 201 }
 202
 203 // buildCallStart returns the start of the call of record, or the zero time when
 204 // its date is not valid.
 205 func buildCallStart(record ModelUsageRecord) time.Time {
 206         start, err := time.Parse(time.RFC3339Nano, record.Date)
 207         if err != nil {
 208                 return time.Time{}
 209         }
 210
 211         return start
 212 }
 213
 214 // buildUsedPercents returns the used percent of each quota window of record, by
 215 // header name.
 216 func buildUsedPercents(record ModelUsageRecord) map[string]float64 {
 217         used := make(map[string]float64)
 218         for name, value := range record.Quota {
 219                 if !strings.HasSuffix(name, "-used-percent") {
 220                         continue
 221                 }
 222
 223                 percent, err := strconv.ParseFloat(value, 64)
 224                 if err == nil {
 225                         used[name] = percent
 226                 }
 227         }
 228
 229         return used
 230 }
 231
 232 // buildWriteReport writes the result of a build run.
 233 func buildWriteReport(out io.Writer, appended int, stop string, records []ModelUsageRecord, sizes []int) error {
 234         var input, cached, output, failed int
 235
 236         for _, record := range records {
 237                 input += record.InputTokens
 238                 cached += record.CachedTokens
 239                 output += record.OutputTokens
 240                 if record.Error != "" {
 241                         failed++
 242                 }
 243         }
 244
 245         var lines []string
 246
 247         if stop != "" {
 248                 lines = append(lines, "stopped early: "+stop)
 249         }
 250
 251         lines = append(lines,
 252                 fmt.Sprintf("messages: %d", appended),
 253                 fmt.Sprintf("calls: %d (%d failed), %.2f per message", len(records), failed, buildRatio(len(records), appended)),
 254                 fmt.Sprintf("model lines: %d, retries: %d, %.2f calls per model line", len(sizes), len(records)-failed-len(sizes), buildRatio(len(records), len(sizes))),
 255                 fmt.Sprintf("tokens: %d input, %d cached (%.1f%%), %d output", input, cached, 100*buildRatio(cached, input), output),
 256         )
 257
 258         if len(sizes) > 0 {
 259                 lines = append(lines, buildSizeText(sizes))
 260         }
 261
 262         if len(records) > 0 {
 263                 lines = append(lines, buildTimeText(records))
 264         }
 265
 266         if len(records) > 0 {
 267                 lines = append(lines,
 268                         "quota at the first call: "+buildQuotaText(records[0]),
 269                         "quota at the last call: "+buildQuotaText(records[len(records)-1]),
 270                 )
 271         }
 272
 273         _, err := io.WriteString(out, strings.Join(lines, "\n")+"\n")
 274         if err != nil {
 275                 return fmt.Errorf("write build report: %w", err)
 276         }
 277
 278         return nil
 279 }
 280
 281 // buildRatio returns a / b, or 0 when b is 0.
 282 func buildRatio(a, b int) float64 {
 283         if b == 0 {
 284                 return 0
 285         }
 286
 287         return float64(a) / float64(b)
 288 }
 289
 290 // buildSizeText describes the sizes of the model lines.
 291 func buildSizeText(sizes []int) string {
 292         sorted := slices.Sorted(slices.Values(sizes))
 293         total, over := 0, 0
 294         for _, size := range sorted {
 295                 total += size
 296                 if size > nodeBytes {
 297                         over++
 298                 }
 299         }
 300
 301         return fmt.Sprintf("line bytes: min %d, median %d, mean %.0f, max %d, %d over %d", sorted[0], sorted[len(sorted)/2], buildRatio(total, len(sorted)), sorted[len(sorted)-1], over, nodeBytes)
 302 }
 303
 304 // buildTimeText describes the time per call and how many calls ran at once. A
 305 // call runs from its date for its seconds.
 306 func buildTimeText(records []ModelUsageRecord) string {
 307         type event struct {
 308                 at    time.Time
 309                 delta int
 310         }
 311
 312         var events []event
 313
 314         seconds := make([]float64, 0, len(records))
 315         total := 0.0
 316         for _, record := range records {
 317                 start, err := time.Parse(time.RFC3339Nano, record.Date)
 318                 if err != nil {
 319                         continue
 320                 }
 321
 322                 length := time.Duration(record.Seconds * float64(time.Second))
 323                 events = append(events, event{
 324                         at:    start,
 325                         delta: 1,
 326                 }, event{
 327                         at:    start.Add(length),
 328                         delta: -1,
 329                 })
 330
 331                 seconds = append(seconds, record.Seconds)
 332                 total += record.Seconds
 333         }
 334
 335         if len(events) == 0 {
 336                 return "time: no dated calls"
 337         }
 338
 339         slices.SortFunc(events, func(a, b event) int {
 340                 return cmp.Or(a.at.Compare(b.at), cmp.Compare(a.delta, b.delta))
 341         })
 342
 343         running, most := 0, 0
 344         for _, e := range events {
 345                 running += e.delta
 346                 most = max(most, running)
 347         }
 348
 349         wall := events[len(events)-1].at.Sub(events[0].at).Seconds()
 350
 351         slices.Sort(seconds)
 352
 353         return fmt.Sprintf("time: %.0f s in all, %.1f calls per minute, %.1f s per call (median %.1f, max %.1f), calls at once: at most %d, %.1f on average", wall, 60*float64(len(seconds))/max(wall, 1e-9), total/float64(len(seconds)), seconds[len(seconds)/2], seconds[len(seconds)-1], most, total/max(wall, 1e-9))
 354 }
 355
 356 // buildQuotaText returns the quota headers of record, sorted by name.
 357 func buildQuotaText(record ModelUsageRecord) string {
 358         names := slices.Sorted(maps.Keys(record.Quota))
 359         parts := make([]string, len(names))
 360         for n, name := range names {
 361                 parts[n] = strings.TrimPrefix(name, "x-codex-") + "=" + record.Quota[name]
 362         }
 363
 364         return strings.Join(parts, " ")
 365 }