package main import ( "bufio" "cmp" "context" "encoding/json" "errors" "fmt" "io" "maps" "os" "slices" "strconv" "strings" "time" ) // buildLineBytes is the most bytes of one line in a build input or usage log. const buildLineBytes = 1 << 20 // buildQuotaLimit is the largest increase in used quota points of a run. const buildQuotaLimit = 15.0 // BuildRun appends messages, builds their tree, and stops the run by the rules // of buildStop. type BuildRun struct { memory *Memory client *ModelClient usagePath string } // Run appends the entries. It waits for the builds only before a user message, // as a turn does, so up to buildMax builds run at once in between, and it waits // for all builds at the end. It returns the number of messages appended and why // the run stopped early, or "" when it did not. After a stop it waits for the // calls in flight, so no sent call is cancelled. func (r *BuildRun) Run(ctx context.Context, entries []DatedMessage) (int, string) { appended, reason := r.append(ctx, entries) if reason == "" { return appended, "" } err := r.memory.Wait(ctx) if err != nil { reason += "; " + err.Error() } return appended, reason } // append appends the entries and stops at the first stop rule. func (r *BuildRun) append(ctx context.Context, entries []DatedMessage) (int, string) { for n, entry := range entries { if entry.Kind == "user" { reason := r.wait(ctx) if reason != "" { return n, reason } } if ctx.Err() != nil { return n, ctx.Err().Error() } _, err := r.memory.AppendDated(entry.Kind, entry.Text, entry.Date) if err != nil { return n, err.Error() } reason := r.check() if reason != "" { return n + 1, reason } } return len(entries), r.wait(ctx) } // wait returns when no build runs, or when a stop rule ends the run. It checks // the rules each second, so a run stops soon also during a long wait. func (r *BuildRun) wait(ctx context.Context) string { for { reason := r.check() if reason != "" { return reason } err := r.waitBuilds(ctx) if err == nil { return r.check() } if ctx.Err() != nil { return ctx.Err().Error() } } } // check returns why the run must stop, or "" when it can go on. func (r *BuildRun) check() string { if r.client.LimitReached() { return fmt.Sprintf("call limit of %d reached", r.client.CallLimit()) } records, err := buildReadUsage(r.usagePath) if err != nil { return err.Error() } return buildStop(records) } // waitBuilds waits up to one second without leaving a child context open. func (r *BuildRun) waitBuilds(ctx context.Context) error { ctx, cancel := context.WithTimeout(ctx, time.Second) defer cancel() return r.memory.Wait(ctx) } // DatedMessage is one message with an optional RFC 3339 date. type DatedMessage struct { Kind string `json:"kind"` Text string `json:"text"` Date string `json:"date,omitempty"` } // buildReadUsage reads the records of usage.jsonl at path, in the order of the // start of their calls. The file has them in the order of the end of their // calls, which differs when calls run at once. func buildReadUsage(path string) (records []ModelUsageRecord, err error) { file, err := os.Open(path) if errors.Is(err, os.ErrNotExist) { return nil, nil } if err != nil { return nil, fmt.Errorf("open usage file: %w", err) } defer func() { closeErr := file.Close() if closeErr != nil { err = errors.Join(err, fmt.Errorf("close build usage: %w", closeErr)) } }() scanner := bufio.NewScanner(file) scanner.Buffer(make([]byte, 64*1024), buildLineBytes) for scanner.Scan() { var record ModelUsageRecord err = json.Unmarshal(scanner.Bytes(), &record) if err != nil { return nil, fmt.Errorf("decode usage: %w", err) } records = append(records, record) } err = scanner.Err() if err != nil { return nil, fmt.Errorf("read usage file: %w", err) } slices.SortStableFunc(records, func(a, b ModelUsageRecord) int { return buildCallStart(a).Compare(buildCallStart(b)) }) return records, nil } // buildWriteReport writes the result of a build run. func buildWriteReport(out io.Writer, appended int, stop string, records []ModelUsageRecord, sizes []int) error { var input, cached, output, failed int for _, record := range records { input += record.InputTokens cached += record.CachedTokens output += record.OutputTokens if record.Error != "" { failed++ } } var lines []string if stop != "" { lines = append(lines, "stopped early: "+stop) } lines = append(lines, fmt.Sprintf("messages: %d", appended), fmt.Sprintf("calls: %d (%d failed), %.2f per message", len(records), failed, buildRatio(len(records), appended)), fmt.Sprintf("model lines: %d, retries: %d, %.2f calls per model line", len(sizes), len(records)-failed-len(sizes), buildRatio(len(records), len(sizes))), fmt.Sprintf("tokens: %d input, %d cached (%.1f%%), %d output", input, cached, 100*buildRatio(cached, input), output), ) if len(sizes) > 0 { lines = append(lines, buildSizeText(sizes)) } if len(records) > 0 { lines = append(lines, buildTimeText(records)) } if len(records) > 0 { lines = append(lines, "quota at the first call: "+buildQuotaText(records[0]), "quota at the last call: "+buildQuotaText(records[len(records)-1]), ) } _, err := io.WriteString(out, strings.Join(lines, "\n")+"\n") if err != nil { return fmt.Errorf("write build report: %w", err) } return nil } // buildStop returns why a build run must stop: a failed call, or a quota window // whose used percent grew by more than buildQuotaLimit points since the first // call. It returns "" when the run can go on. func buildStop(records []ModelUsageRecord) string { for _, record := range records { if record.Error != "" { return "call failed: " + record.Error } } if len(records) == 0 { return "" } first := buildUsedPercents(records[0]) for name, used := range buildUsedPercents(records[len(records)-1]) { start, found := first[name] if found && used-start > buildQuotaLimit { return fmt.Sprintf("%s grew from %g to %g", name, start, used) } } return "" } // buildCallStart returns the start of the call of record, or the zero time when // its date is not valid. func buildCallStart(record ModelUsageRecord) time.Time { start, err := time.Parse(time.RFC3339Nano, record.Date) if err != nil { return time.Time{} } return start } // buildRatio returns a / b, or 0 when b is 0. func buildRatio(a, b int) float64 { if b == 0 { return 0 } return float64(a) / float64(b) } // buildSizeText describes the sizes of the model lines. func buildSizeText(sizes []int) string { sorted := slices.Sorted(slices.Values(sizes)) total, over := 0, 0 for _, size := range sorted { total += size if size > nodeBytes { over++ } } 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) } // buildTimeText describes the time per call and how many calls ran at once. A // call runs from its date for its seconds. func buildTimeText(records []ModelUsageRecord) string { type event struct { at time.Time delta int } var events []event seconds := make([]float64, 0, len(records)) total := 0.0 for _, record := range records { start, err := time.Parse(time.RFC3339Nano, record.Date) if err != nil { continue } length := time.Duration(record.Seconds * float64(time.Second)) events = append(events, event{ at: start, delta: 1, }, event{ at: start.Add(length), delta: -1, }) seconds = append(seconds, record.Seconds) total += record.Seconds } if len(events) == 0 { return "time: no dated calls" } slices.SortFunc(events, func(a, b event) int { return cmp.Or(a.at.Compare(b.at), cmp.Compare(a.delta, b.delta)) }) running, most := 0, 0 for _, e := range events { running += e.delta most = max(most, running) } wall := events[len(events)-1].at.Sub(events[0].at).Seconds() slices.Sort(seconds) 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)) } // buildQuotaText returns the quota headers of record, sorted by name. func buildQuotaText(record ModelUsageRecord) string { names := slices.Sorted(maps.Keys(record.Quota)) parts := make([]string, len(names)) for n, name := range names { parts[n] = strings.TrimPrefix(name, "x-codex-") + "=" + record.Quota[name] } return strings.Join(parts, " ") } // buildUsedPercents returns the used percent of each quota window of record, by // header name. func buildUsedPercents(record ModelUsageRecord) map[string]float64 { used := make(map[string]float64) for name, value := range record.Quota { if !strings.HasSuffix(name, "-used-percent") { continue } percent, err := strconv.ParseFloat(value, 64) if err == nil { used[name] = percent } } return used }