X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/baf7a940be8927278e3b4b87e7366cebfc16532f..0da50dc187c8b4aa23ac1ddd0418c0da779389d9:/probe.go?ds=sidebyside

diff --git a/probe.go b/probe.go
index ba656ff..384d0f8 100644
--- a/probe.go
+++ b/probe.go
@@ -2,18 +2,15 @@ package main
 
 import (
 	"bufio"
-	"cmp"
 	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
-	"maps"
+	"log/slog"
 	"os"
 	"path/filepath"
-	"slices"
 	"strconv"
-	"strings"
 	"time"
 )
 
@@ -22,15 +19,8 @@ const (
 	// it.
 	probeName = "Probe"
 
-	// probeQuotaLimit is how many points the used percent of a quota window
-	// can grow in a probe run before the run stops.
-	probeQuotaLimit = 15.0
-
 	// probeCallLimit is the most model calls of a probe run.
 	probeCallLimit = 660
-
-	// probeLineBytes is the most bytes of one line of a probe log.
-	probeLineBytes = 1 << 20
 )
 
 // probe runs "onechat probe LOG COUNT [BUILDS]". It appends the first COUNT
@@ -38,7 +28,7 @@ const (
 // 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
+// of a quota window grows by more than buildQuotaLimit 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 buildMax,
 // limits the builds that call the model at once in this run only. The core
@@ -50,45 +40,56 @@ func probe(ctx context.Context, args []string, out io.Writer) (err error) {
 	}
 
 	count, err := strconv.Atoi(args[1])
-	if err != nil || count <= 0 {
+	if err != nil {
+		return fmt.Errorf("parse probe count: %w", err)
+	}
+
+	if count <= 0 {
 		return fmt.Errorf("COUNT %q is not a positive number", args[1])
 	}
 
 	builds := buildMax
 	if len(args) == 3 {
 		builds, err = strconv.Atoi(args[2])
-		if err != nil || builds < 1 || builds > buildMax {
+		if err != nil {
+			return fmt.Errorf("parse probe builds: %w", err)
+		}
+
+		if builds < 1 || builds > buildMax {
 			return fmt.Errorf("BUILDS %q is not a number from 1 to %d", args[2], buildMax)
 		}
 	}
 
 	entries, err := probeReadLog(args[0], count)
 	if err != nil {
-		return err
+		return fmt.Errorf("read probe input: %w", err)
 	}
 
 	dir, err := dataDir()
 	if err != nil {
-		return err
+		return fmt.Errorf("find probe data directory: %w", err)
 	}
 
 	config, err := loadConfig(dir)
 	if err != nil {
-		return err
+		return fmt.Errorf("load probe configuration: %w", err)
 	}
 
 	tokens, err := lockTokens(dir)
 	if err != nil {
-		return err
+		return fmt.Errorf("lock probe tokens: %w", err)
 	}
 
 	defer func() {
-		err = errors.Join(err, tokens.Close())
+		closeErr := tokens.Close()
+		if closeErr != nil {
+			err = errors.Join(err, fmt.Errorf("close probe token lock: %w", closeErr))
+		}
 	}()
 
 	address, auth, err := modelEndpoint(config, dir)
 	if err != nil {
-		return err
+		return fmt.Errorf("select probe endpoint: %w", err)
 	}
 
 	root := filepath.Join(dir, "probe", time.Now().Format("20060102-150405"))
@@ -100,40 +101,48 @@ func probe(ctx context.Context, args []string, out io.Writer) (err error) {
 	usagePath := filepath.Join(root, "usage.jsonl")
 	client := newModelClient(address, auth, "onechat-"+probeName, usagePath)
 	client.LimitCalls(probeCallLimit)
+
 	summarizer := ProbeSlotSummarizer{
-		inner: newCompactor(client, probePrompt, config.Compactor),
+		inner: newCompactor(client, probePrompt, config.Compactor, promptTools(), &ImageStore{dir: filepath.Join(dir, imageDirName)}),
 		slots: make(chan struct{}, builds),
 	}
 
-	memory, err := openMemory(ctx, root, time.Now, summarizer)
+	// The report of the probe counts the failed calls, so the lines of the
+	// failed builds are not needed.
+	memory, err := openMemory(ctx, root, time.Now, summarizer, slog.New(slog.DiscardHandler))
 	if err != nil {
-		return err
+		return fmt.Errorf("open probe memory: %w", err)
 	}
 
 	_, 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 fmt.Errorf("write probe header: %w", errors.Join(err, memory.Close()))
 	}
 
-	run := ProbeRun{
+	run := BuildRun{
 		memory:    memory,
 		client:    client,
 		usagePath: usagePath,
 	}
 
 	appended, stop := run.Run(ctx, entries)
-	sizes, err := probeModelLines(memory)
+	sizes, err := memory.buildModelLines()
 	err = errors.Join(err, memory.Close())
 	if err != nil {
-		return err
+		return fmt.Errorf("finish probe memory: %w", err)
 	}
 
-	records, err := probeReadUsage(usagePath)
+	records, err := buildReadUsage(usagePath)
 	if err != nil {
-		return err
+		return fmt.Errorf("read probe usage: %w", err)
 	}
 
-	return probeWriteReport(out, appended, stop, records, sizes)
+	err = buildWriteReport(out, appended, stop, records, sizes)
+	if err != nil {
+		return fmt.Errorf("write probe report: %w", err)
+	}
+
+	return nil
 }
 
 // ProbeSlotSummarizer lets at most cap(slots) builds call inner at once. A
@@ -158,119 +167,33 @@ func (s ProbeSlotSummarizer) Summarize(ctx context.Context, task Task) (string,
 	return s.inner.Summarize(ctx, task)
 }
 
-// ProbeRun appends the messages of a probe run to its chat and stops the run by
-// the rules of probeStop.
-type ProbeRun 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 *ProbeRun) Run(ctx context.Context, entries []ProbeEntry) (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 *ProbeRun) append(ctx context.Context, entries []ProbeEntry) (int, string) {
-	for n, entry := range entries {
-		if entry.Kind == "user" {
-			reason := r.wait(ctx)
-			if reason != "" {
-				return n, reason
-			}
-		}
-
-		_, err := r.memory.Append(entry.Kind, entry.Text)
-		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 *ProbeRun) wait(ctx context.Context) string {
-	for {
-		reason := r.check()
-		if reason != "" {
-			return reason
-		}
-
-		waitCtx, cancel := context.WithTimeout(ctx, time.Second)
-		err := r.memory.Wait(waitCtx)
-		cancel()
-		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 *ProbeRun) check() string {
-	if r.client.LimitReached() {
-		return fmt.Sprintf("call limit of %d reached", r.client.limit)
-	}
-
-	records, err := probeReadUsage(r.usagePath)
-	if err != nil {
-		return err.Error()
-	}
-
-	return probeStop(records)
-}
-
-// ProbeEntry is one message of a probe log.
-type ProbeEntry struct {
-	Kind string `json:"kind"`
-	Text string `json:"text"`
-}
-
-// probeReadLog reads the first count messages of the probe log at path.
-func probeReadLog(path string, count int) (entries []ProbeEntry, err error) {
+// probeReadLog reads the first count messages of the probe log at path. Source
+// dates are ignored because a probe uses its own clock.
+func probeReadLog(path string, count int) (entries []DatedMessage, err error) {
 	file, err := os.Open(path)
 	if err != nil {
 		return nil, fmt.Errorf("open probe log: %w", err)
 	}
 
 	defer func() {
-		err = errors.Join(err, file.Close())
+		closeErr := file.Close()
+		if closeErr != nil {
+			err = errors.Join(err, fmt.Errorf("close probe log: %w", closeErr))
+		}
 	}()
 
 	scanner := bufio.NewScanner(file)
-	scanner.Buffer(make([]byte, 64*1024), probeLineBytes)
+	scanner.Buffer(make([]byte, 64*1024), buildLineBytes)
+
 	for len(entries) < count && scanner.Scan() {
-		var entry ProbeEntry
+		var entry DatedMessage
+
 		err = json.Unmarshal(scanner.Bytes(), &entry)
 		if err != nil {
 			return nil, fmt.Errorf("probe log line %d: %w", len(entries)+1, err)
 		}
 
+		entry.Date = ""
 		entries = append(entries, entry)
 	}
 
@@ -287,284 +210,7 @@ func probeReadLog(path string, count int) (entries []ProbeEntry, err error) {
 }
 
 // probePrompt returns the system prompt of a probe run: the prompt of the agent
-// probeName, without instructions.
+// probeName, without instructions and with a fixed instructions file path.
 func probePrompt() (string, error) {
-	return prompt(probeName, ""), nil
-}
-
-// probeModelLines returns the sizes of the nodes that the model wrote, in tree
-// order. A node that fits its source with no model call is not one of them.
-func probeModelLines(memory *Memory) ([]int, error) {
-	memory.mutex.Lock()
-	defer memory.mutex.Unlock()
-
-	keys, err := memory.tree.Keys()
-	if err != nil {
-		return nil, err
-	}
-
-	var sizes []int
-	for _, key := range keys {
-		node, _, err := memory.tree.Node(key.level, key.index)
-		if err != nil {
-			return nil, err
-		}
-
-		model, err := probeBuiltByModel(memory, key)
-		if err != nil {
-			return nil, err
-		}
-
-		if model {
-			sizes = append(sizes, node.Size)
-		}
-	}
-
-	return sizes, nil
-}
-
-// probeReadUsage 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 probeReadUsage(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() {
-		err = errors.Join(err, file.Close())
-	}()
-
-	scanner := bufio.NewScanner(file)
-	scanner.Buffer(make([]byte, 64*1024), probeLineBytes)
-	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 probeCallStart(a).Compare(probeCallStart(b))
-	})
-
-	return records, nil
-}
-
-// probeWriteReport writes the result of a probe run.
-func probeWriteReport(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, probeRatio(len(records), appended)),
-		fmt.Sprintf("model lines: %d, retries: %d, %.2f calls per model line", len(sizes), len(records)-failed-len(sizes), probeRatio(len(records), len(sizes))),
-		fmt.Sprintf("tokens: %d input, %d cached (%.1f%%), %d output", input, cached, 100*probeRatio(cached, input), output),
-	)
-
-	if len(sizes) > 0 {
-		lines = append(lines, probeSizeText(sizes))
-	}
-
-	if len(records) > 0 {
-		lines = append(lines, probeTimeText(records))
-	}
-
-	if len(records) > 0 {
-		lines = append(lines,
-			"quota at the first call: "+probeQuotaText(records[0]),
-			"quota at the last call: "+probeQuotaText(records[len(records)-1]),
-		)
-	}
-
-	_, err := io.WriteString(out, strings.Join(lines, "\n")+"\n")
-
-	return err
-}
-
-// probeStop returns why a probe run must stop: a failed call, or a quota window
-// whose used percent grew by more than probeQuotaLimit points since the first
-// call. It returns "" when the run can go on.
-func probeStop(records []ModelUsageRecord) string {
-	for _, record := range records {
-		if record.Error != "" {
-			return "call failed: " + record.Error
-		}
-	}
-
-	if len(records) == 0 {
-		return ""
-	}
-
-	first := probeUsedPercents(records[0])
-	for name, used := range probeUsedPercents(records[len(records)-1]) {
-		start, found := first[name]
-		if found && used-start > probeQuotaLimit {
-			return fmt.Sprintf("%s grew from %g to %g", name, start, used)
-		}
-	}
-
-	return ""
-}
-
-// probeBuiltByModel tells if the node key needed a model call: a leaf whose
-// source is over nodeBytes, or a merge whose children do not fit in one node.
-func probeBuiltByModel(memory *Memory, key NodeKey) (bool, error) {
-	if key.level == 0 {
-		message, err := memory.log.Message(key.index)
-		if err != nil {
-			return false, err
-		}
-
-		_, free := freeLeaf(message)
-
-		return !free, nil
-	}
-
-	left, right, err := memory.children(key)
-	if err != nil {
-		return false, err
-	}
-
-	_, free := freeMerge(left, right)
-
-	return !free, nil
-}
-
-// probeCallStart returns the start of the call of record, or the zero time when
-// its date is not valid.
-func probeCallStart(record ModelUsageRecord) time.Time {
-	start, err := time.Parse(time.RFC3339Nano, record.Date)
-	if err != nil {
-		return time.Time{}
-	}
-
-	return start
-}
-
-// probeRatio returns a / b, or 0 when b is 0.
-func probeRatio(a, b int) float64 {
-	if b == 0 {
-		return 0
-	}
-
-	return float64(a) / float64(b)
-}
-
-// probeSizeText describes the sizes of the model lines.
-func probeSizeText(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], probeRatio(total, len(sorted)), sorted[len(sorted)-1], over, nodeBytes)
-}
-
-// probeTimeText describes the time per call and how many calls ran at once. A
-// call runs from its date for its seconds.
-func probeTimeText(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))
-}
-
-// probeQuotaText returns the quota headers of record, sorted by name.
-func probeQuotaText(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, " ")
-}
-
-// probeUsedPercents returns the used percent of each quota window of record, by
-// header name.
-func probeUsedPercents(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
+	return prompt(probeName, "/PROBE/instructions.md", ""), nil
 }