Probe
*******
-``onechat probe LOG COUNT`` measures the compactor on a log. It appends
-the first COUNT messages of LOG, one JSON object ``{"kind", "text"}`` on
-each line, to a new chat in ``probe/`` in the data directory, and builds
-its tree with real model calls. The calls use the quota of the account.
-The run stops at the first failed call, or when the used percent of a
-quota window grows by more than 15 points. At the end it reports the
-calls per message, the cached share of the input tokens, the retries,
-the line sizes, and the quota at the first and the last call.
+``onechat probe LOG COUNT [BUILDS]`` measures the compactor on a log. It
+appends the first COUNT messages of LOG, one JSON object ``{"kind",
+"text"}`` on each line, to a new chat in ``probe/`` in the data
+directory, and builds its tree with real model calls. The calls use the
+quota of the account. The run stops at the first failed call, after 660
+calls, or when the used percent of a quota window grows by more than 15
+points. At the end it reports the calls per message, the cached share of
+the input tokens, the retries, the line sizes, and the quota at the
+first and the last call. BUILDS, from 1 to 8, limits the model calls at
+once in the run.
A test writes a generated log of readable made-up messages:
probeLineBytes = 1 << 20
)
-// probe runs "onechat probe LOG COUNT". It appends the first COUNT messages of
-// LOG, one JSON object {kind, text} per line, to a new chat in probe/ in the
-// data directory, and builds its tree with the real compactor. Before each
-// user message it waits for the builds, as a turn does. It stops at the first
-// failed call, after probeCallLimit calls, or when the used percent of a quota
-// window grows by more than probeQuotaLimit points. At the end it reports the
-// calls, the time per call, the calls at once, the cached share, the quota,
-// the retries, and the line sizes.
+// probe runs "onechat probe LOG COUNT [BUILDS]". It appends the first COUNT
+// messages of LOG, one JSON object {kind, text} per line, to a new chat in
+// probe/ in the data directory, and builds its tree with the real compactor.
+// Before each user message it waits for the builds, as a turn does. It stops at
+// the first failed call, after probeCallLimit calls, or when the used percent
+// of a quota window grows by more than probeQuotaLimit points. At the end it
+// reports the calls, the time per call, the calls at once, the cached share,
+// the quota, the retries, and the line sizes. BUILDS, from 1 to maxBuilds,
+// limits the builds that call the model at once in this run only. The core
+// still queues up to maxBuilds builds.
func probe(ctx context.Context, args []string, out io.Writer) error {
- if len(args) != 2 {
- return errors.New("usage: onechat probe LOG COUNT")
+ if len(args) != 2 && len(args) != 3 {
+ return errors.New("usage: onechat probe LOG COUNT [BUILDS]")
}
count, err := strconv.Atoi(args[1])
return fmt.Errorf("COUNT %q is not a positive number", args[1])
}
+ builds := maxBuilds
+ if len(args) == 3 {
+ builds, err = strconv.Atoi(args[2])
+ if err != nil || builds < 1 || builds > maxBuilds {
+ return fmt.Errorf("BUILDS %q is not a number from 1 to %d", args[2], maxBuilds)
+ }
+ }
+
entries, err := readProbeLog(args[0], count)
if err != nil {
return err
usagePath := filepath.Join(root, "usage.jsonl")
client := newClient(address, auth, "onechat-"+probeName, usagePath)
client.LimitCalls(probeCallLimit)
- memory, err := openMemory(ctx, root, time.Now, newCompactor(client, probeName, config.Compactor))
+ summarizer := slotSummarizer{
+ inner: newCompactor(client, probeName, config.Compactor),
+ slots: make(chan struct{}, builds),
+ }
+
+ memory, err := openMemory(ctx, root, time.Now, summarizer)
if err != nil {
return err
}
- _, err = fmt.Fprintf(out, "probe: %d messages into %s with %s at %s effort\n", len(entries), root, config.Compactor.Model, config.Compactor.Effort)
+ _, err = fmt.Fprintf(out, "probe: %d messages into %s with %s at %s effort, %d builds at once\n", len(entries), root, config.Compactor.Model, config.Compactor.Effort, builds)
if err != nil {
return errors.Join(err, memory.Close())
}
return probeStop(records)
}
+// slotSummarizer lets at most cap(slots) builds call inner at once. A build
+// past the limit waits for a slot.
+type slotSummarizer struct {
+ inner Summarizer
+ slots chan struct{}
+}
+
+// Summarize waits for a slot and builds the node with inner.
+func (s slotSummarizer) Summarize(ctx context.Context, task Task) (string, error) {
+ select {
+ case s.slots <- struct{}{}:
+ case <-ctx.Done():
+ return "", ctx.Err()
+ }
+
+ defer func() {
+ <-s.slots
+ }()
+
+ return s.inner.Summarize(ctx, task)
+}
+
// probeEntry is one message of a probe log.
type probeEntry struct {
Kind string `json:"kind"`
return !free, nil
}
-// readUsageFile reads the records of usage.jsonl at path.
+// readUsageFile 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 readUsageFile(path string) (records []usageRecord, err error) {
file, err := os.Open(path)
if errors.Is(err, os.ErrNotExist) {
return nil, fmt.Errorf("read usage file: %w", err)
}
+ slices.SortStableFunc(records, func(a, b usageRecord) int {
+ return callStart(a).Compare(callStart(b))
+ })
+
return records, nil
}
+// callStart returns the start of the call of record, or the zero time when its
+// date is not valid.
+func callStart(record usageRecord) time.Time {
+ start, err := time.Parse(time.RFC3339Nano, record.Date)
+ if err != nil {
+ return time.Time{}
+ }
+
+ return start
+}
+
// usedPercents returns the used percent of each quota window of record, by
// header name.
func usedPercents(record usageRecord) map[string]float64 {
wall := events[len(events)-1].at.Sub(events[0].at).Seconds()
slices.Sort(seconds)
- return fmt.Sprintf("time: %.0f s in all, %.1f s per call (median %.1f, max %.1f), calls at once: at most %d, %.1f on average", wall, total/float64(len(seconds)), seconds[len(seconds)/2], seconds[len(seconds)-1], most, total/max(wall, 1e-9))
+ 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))
}
// quotaText returns the quota headers of record, sorted by name.
require.NoError(t, err)
require.NoError(t, run.memory.Close())
- require.Contains(t, stop, "x-codex-primary-used-percent grew from 10 to")
+ require.Regexp(t, `^x-codex-primary-used-percent grew from \d+ to \d+$`, stop)
require.Less(t, appended, probeMessages)
require.NotEmpty(t, sizes)
require.Contains(t, report.String(), "cached (80.0%)")
require.Contains(t, report.String(), "line bytes: min 400, median 400, mean 400, max 400, 0 over 512")
require.Contains(t, report.String(), "calls at once: at most ")
- require.Contains(t, report.String(), "quota at the first call: primary-used-percent=10")
+ require.Contains(t, report.String(), "quota at the first call: primary-used-percent=")
}
func TestProbeRunStopsAtTheCallLimit(t *testing.T) {
require.Equal(t, 3, server.Calls())
}
+func TestSlotSummarizerLimitsTheBuildsAtOnce(t *testing.T) {
+ inner := newFakeSummarizer()
+ inner.gate = make(chan struct{})
+ summarizer := slotSummarizer{
+ inner: inner,
+ slots: make(chan struct{}, 2),
+ }
+
+ var group sync.WaitGroup
+ for n := range 6 {
+ group.Go(func() {
+ _, err := summarizer.Summarize(t.Context(), Task{
+ Level: 0,
+ Index: n,
+ Input: []string{"user: " + strings.Repeat("x", 600)},
+ })
+ require.NoError(t, err)
+ })
+ }
+
+ require.Eventually(t, func() bool {
+ return inner.Active() == 2
+ }, 10*time.Second, time.Millisecond)
+
+ close(inner.gate)
+ group.Wait()
+ require.Equal(t, 2, inner.peak)
+}
+
+func TestReadUsageFileSortsByCallStart(t *testing.T) {
+ path := filepath.Join(t.TempDir(), "usage.jsonl")
+ usage := usageLog{
+ path: path,
+ }
+
+ for _, date := range []string{"2026-01-01T12:00:02Z", "2026-01-01T12:00:00.5Z", "2026-01-01T12:00:01.25Z"} {
+ require.NoError(t, usage.Write(usageRecord{
+ Date: date,
+ }))
+ }
+
+ records, err := readUsageFile(path)
+ require.NoError(t, err)
+ require.Equal(t, []string{"2026-01-01T12:00:00.5Z", "2026-01-01T12:00:01.25Z", "2026-01-01T12:00:02Z"}, []string{records[0].Date, records[1].Date, records[2].Date})
+}
+
func TestTimeTextCountsTheCallsAtOnce(t *testing.T) {
start := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
record := func(offset, seconds float64) usageRecord {
}
text := timeText([]usageRecord{record(0, 10), record(2, 4), record(3, 2), record(10, 10)})
- require.Equal(t, "time: 20 s in all, 6.5 s per call (median 10.0, max 10.0), calls at once: at most 3, 1.3 on average", text)
+ require.Equal(t, "time: 20 s in all, 12.0 calls per minute, 6.5 s per call (median 10.0, max 10.0), calls at once: at most 3, 1.3 on average", text)
}
// probeLog returns count messages of made-up work on a small bakery order
projects / onechat / commitdiff