+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
+
+// buildCallLimit is the most model calls of a run.
+const buildCallLimit = 660
+
+// 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
+}
projects / onechat / commitdiff