mirror of
https://github.com/lukaszraczylo/llm-testbench.git
synced 2026-10-12 00:45:19 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
dfb9fcc370 | ||
|
|
1310c5bebe | ||
|
|
7b791a78e7 | ||
|
|
6934d4df84 |
@@ -0,0 +1,20 @@
|
||||
name: Autoupdate go.mod and go.sum
|
||||
|
||||
on:
|
||||
workflow_dispatch:
|
||||
schedule:
|
||||
- cron: "0 3 * * *"
|
||||
|
||||
permissions:
|
||||
contents: write
|
||||
actions: write
|
||||
pull-requests: write
|
||||
security-events: write
|
||||
|
||||
jobs:
|
||||
autoupdate:
|
||||
uses: lukaszraczylo/shared-actions/.github/workflows/go-autoupdate.yaml@main
|
||||
with:
|
||||
go-version: ">=1.24"
|
||||
release-workflow: "release.yml"
|
||||
secrets: inherit
|
||||
@@ -0,0 +1,22 @@
|
||||
name: Pull Request
|
||||
|
||||
on:
|
||||
pull_request:
|
||||
branches:
|
||||
- main
|
||||
push:
|
||||
branches:
|
||||
- "**"
|
||||
- "!main"
|
||||
|
||||
permissions:
|
||||
contents: write
|
||||
actions: write
|
||||
pull-requests: write
|
||||
security-events: write
|
||||
|
||||
jobs:
|
||||
pr-checks:
|
||||
uses: lukaszraczylo/shared-actions/.github/workflows/go-pr.yaml@main
|
||||
with:
|
||||
go-version: ">=1.24"
|
||||
@@ -2,86 +2,26 @@ name: Release
|
||||
|
||||
on:
|
||||
push:
|
||||
branches: [main]
|
||||
branches:
|
||||
- main
|
||||
paths:
|
||||
- "**.go"
|
||||
- "go.mod"
|
||||
- "go.sum"
|
||||
- "Dockerfile"
|
||||
- ".goreleaser.yaml"
|
||||
workflow_dispatch:
|
||||
|
||||
permissions:
|
||||
contents: read
|
||||
|
||||
concurrency:
|
||||
group: release-${{ github.ref }}
|
||||
cancel-in-progress: false
|
||||
contents: write
|
||||
packages: write
|
||||
id-token: write
|
||||
|
||||
jobs:
|
||||
release:
|
||||
name: Version, tag, and release
|
||||
runs-on: ubuntu-latest
|
||||
timeout-minutes: 30
|
||||
permissions:
|
||||
contents: write
|
||||
packages: write
|
||||
steps:
|
||||
- name: Checkout repository
|
||||
uses: actions/checkout@v7
|
||||
with:
|
||||
fetch-depth: 0
|
||||
|
||||
- name: Calculate semantic version
|
||||
id: semver
|
||||
uses: lukaszraczylo/semver-generator@b5eb8ae15a62d76e81dc4661a49209a8502cf1d3 # v1.17.18
|
||||
with:
|
||||
config_file: semver.yaml
|
||||
repository_local: true
|
||||
existing: true
|
||||
|
||||
- name: Create tag if new
|
||||
id: tag
|
||||
run: |
|
||||
VERSION="v${{ steps.semver.outputs.semantic_version }}"
|
||||
echo "version=${VERSION}" >> "$GITHUB_OUTPUT"
|
||||
if git rev-parse "${VERSION}" >/dev/null 2>&1; then
|
||||
echo "Tag ${VERSION} already exists; skipping release."
|
||||
echo "new=false" >> "$GITHUB_OUTPUT"
|
||||
else
|
||||
git config user.name "github-actions[bot]"
|
||||
git config user.email "github-actions[bot]@users.noreply.github.com"
|
||||
git tag -a "${VERSION}" -m "Release ${VERSION}"
|
||||
git push origin "${VERSION}"
|
||||
echo "new=true" >> "$GITHUB_OUTPUT"
|
||||
fi
|
||||
|
||||
- name: Set up Go
|
||||
if: steps.tag.outputs.new == 'true'
|
||||
uses: actions/setup-go@v7
|
||||
with:
|
||||
go-version-file: go.mod
|
||||
|
||||
- name: Run tests
|
||||
if: steps.tag.outputs.new == 'true'
|
||||
run: go test -race ./...
|
||||
|
||||
- name: Set up QEMU
|
||||
if: steps.tag.outputs.new == 'true'
|
||||
uses: docker/setup-qemu-action@96fe6ef7f33517b61c61be40b68a1882f3264fb8 # v4.2.0
|
||||
|
||||
- name: Set up Docker Buildx
|
||||
if: steps.tag.outputs.new == 'true'
|
||||
uses: docker/setup-buildx-action@bb05f3f5519dd87d3ba754cc423b652a5edd6d2c # v4.2.0
|
||||
|
||||
- name: Log in to GitHub Container Registry
|
||||
if: steps.tag.outputs.new == 'true'
|
||||
uses: docker/login-action@dbcb813823bdd20940b903addbd779551569679f # v4.6.0
|
||||
with:
|
||||
registry: ghcr.io
|
||||
username: ${{ github.actor }}
|
||||
password: ${{ secrets.GITHUB_TOKEN }}
|
||||
|
||||
- name: Run goreleaser
|
||||
if: steps.tag.outputs.new == 'true'
|
||||
uses: goreleaser/goreleaser-action@f06c13b6b1a9625abc9e6e439d9c05a8f2190e94 # v7.2.3
|
||||
with:
|
||||
distribution: goreleaser
|
||||
version: "~> v2"
|
||||
args: release --clean
|
||||
env:
|
||||
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
|
||||
uses: lukaszraczylo/shared-actions/.github/workflows/go-release.yaml@main
|
||||
with:
|
||||
go-version: ">=1.24"
|
||||
docker-enabled: true
|
||||
docker-registry: ghcr.io
|
||||
secrets: inherit
|
||||
|
||||
@@ -75,6 +75,8 @@ Both `list` and `run` accept:
|
||||
`60s`).
|
||||
- `--out FILE` - write the run as a JSON artifact (raw per-attempt results
|
||||
plus the per-test discrimination rollup) for later `llmtest compare`.
|
||||
- `--seed N` - send `seed: N` with every request (overrides the config's
|
||||
`seed`). Whether the backend honors it is backend-specific.
|
||||
- `--quiet` - suppress progress output on stderr. Progress is on by
|
||||
default: a run can issue hundreds of requests at 30-50s each, and a
|
||||
silent CLI reads as a hang.
|
||||
|
||||
+71
-5
@@ -4,11 +4,15 @@ package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"flag"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/signal"
|
||||
"sort"
|
||||
"strings"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/lukaszraczylo/llm-testbench/internal/config"
|
||||
"github.com/lukaszraczylo/llm-testbench/internal/llm"
|
||||
@@ -121,6 +125,7 @@ func runCommand(args []string) error {
|
||||
repeat := fs.Int("repeat", 1, "samples per (model, test); >1 exposes response instability at temperature 0")
|
||||
out := fs.String("out", "", "write the run as a JSON artifact to this file (for llmtest compare)")
|
||||
quiet := fs.Bool("quiet", false, "suppress progress output on stderr")
|
||||
seedFlag := fs.Int("seed", 0, "send this seed with every request (overrides config's seed)")
|
||||
if err := fs.Parse(args); err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -132,7 +137,7 @@ func runCommand(args []string) error {
|
||||
|
||||
models := cfg.Models
|
||||
if *modelsCSV != "" {
|
||||
models = splitCSV(*modelsCSV)
|
||||
models = dedupe(splitCSV(*modelsCSV))
|
||||
}
|
||||
|
||||
reqTimeout := cfg.RequestTimeout
|
||||
@@ -145,7 +150,18 @@ func runCommand(args []string) error {
|
||||
conc = *concurrency
|
||||
}
|
||||
|
||||
client := llm.NewOpenAIClient(cfg.Endpoint, cfg.APIKey, reqTimeout)
|
||||
seed := cfg.Seed
|
||||
fs.Visit(func(f *flag.Flag) {
|
||||
if f.Name == "seed" {
|
||||
seed = seedFlag
|
||||
}
|
||||
})
|
||||
|
||||
client := llm.NewOpenAIClient(cfg.Endpoint, cfg.APIKey, reqTimeout).
|
||||
WithRetries(cfg.MaxRetries).
|
||||
WithRetryTimeouts(cfg.RetryTimeouts).
|
||||
WithModelTimeouts(cfg.ModelTimeouts).
|
||||
WithStreaming(cfg.Stream, cfg.IdleTimeout)
|
||||
|
||||
registry := tests.All()
|
||||
selected, err := selectTests(registry, *testsCSV, shared.category, shared.subcategory)
|
||||
@@ -164,10 +180,23 @@ func runCommand(args []string) error {
|
||||
MaxTokensDefault: cfg.MaxTokensDefault,
|
||||
Reporter: reporter,
|
||||
Repeat: *repeat,
|
||||
Seed: seed,
|
||||
ModelConcurrency: cfg.ModelConcurrency,
|
||||
})
|
||||
|
||||
ctx := context.Background()
|
||||
// First Ctrl-C stops dispatching and lets in-flight calls drain so the
|
||||
// partial run can still be saved; stop() restores default handling, so
|
||||
// a second Ctrl-C kills the process.
|
||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
||||
defer stop()
|
||||
startedAt := time.Now().UTC()
|
||||
results := r.Run(ctx, models, selected)
|
||||
interrupted := ctx.Err() != nil
|
||||
if interrupted {
|
||||
stop()
|
||||
results = dropCancelled(results)
|
||||
fmt.Fprintf(os.Stderr, "llmtest: interrupted; reporting %d completed results\n", len(results))
|
||||
}
|
||||
|
||||
f, err := validateFormat(*format)
|
||||
if err != nil {
|
||||
@@ -175,7 +204,17 @@ func runCommand(args []string) error {
|
||||
}
|
||||
|
||||
if *out != "" {
|
||||
if err := report.WriteArtifact(*out, selected, models, results); err != nil {
|
||||
meta := report.RunMeta{
|
||||
Version: version,
|
||||
StartedAt: startedAt.Format(time.RFC3339),
|
||||
Endpoint: cfg.Endpoint,
|
||||
Seed: seed,
|
||||
Temperature: requestTemperature,
|
||||
MaxTokensDefault: cfg.MaxTokensDefault,
|
||||
Repeat: max(*repeat, 1),
|
||||
Interrupted: interrupted,
|
||||
}
|
||||
if err := report.WriteArtifact(*out, meta, selected, models, results); err != nil {
|
||||
return fmt.Errorf("write --out: %w", err)
|
||||
}
|
||||
}
|
||||
@@ -183,6 +222,19 @@ func runCommand(args []string) error {
|
||||
return report.Render(os.Stdout, f, selected, models, results)
|
||||
}
|
||||
|
||||
// dropCancelled removes results that never ran because the run was
|
||||
// interrupted, so they are not saved as failures.
|
||||
func dropCancelled(results []runner.Result) []runner.Result {
|
||||
kept := results[:0:0]
|
||||
for _, r := range results {
|
||||
if r.Err != nil && errors.Is(r.Err, context.Canceled) {
|
||||
continue
|
||||
}
|
||||
kept = append(kept, r)
|
||||
}
|
||||
return kept
|
||||
}
|
||||
|
||||
// selectTests resolves the test subset: --tests (exact IDs, error on any
|
||||
// unknown ID) overrides --category/--subcategory filtering.
|
||||
func selectTests(registry *testkit.Registry, testsCSV, category, subcategory string) ([]testkit.Test, error) {
|
||||
@@ -234,7 +286,7 @@ func compareCommand(args []string) error {
|
||||
return err
|
||||
}
|
||||
|
||||
return report.RenderCompare(os.Stdout, report.CompareArtifacts(baseline, current))
|
||||
return report.RenderCompare(os.Stdout, report.CompareArtifacts(baseline, current), report.CompareWarnings(baseline, current)...)
|
||||
}
|
||||
|
||||
// healthCommand audits one or more saved artifacts for suite health:
|
||||
@@ -271,6 +323,20 @@ func validateFormat(format string) (report.Format, error) {
|
||||
}
|
||||
}
|
||||
|
||||
// dedupe drops repeated entries, keeping first-seen order; a model listed
|
||||
// twice would otherwise run, and be averaged, twice.
|
||||
func dedupe(in []string) []string {
|
||||
seen := make(map[string]bool, len(in))
|
||||
out := make([]string, 0, len(in))
|
||||
for _, v := range in {
|
||||
if !seen[v] {
|
||||
seen[v] = true
|
||||
out = append(out, v)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func splitCSV(s string) []string {
|
||||
parts := strings.Split(s, ",")
|
||||
out := make([]string, 0, len(parts))
|
||||
|
||||
@@ -1,11 +1,15 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/lukaszraczylo/llm-testbench/internal/eval"
|
||||
"github.com/lukaszraczylo/llm-testbench/internal/report"
|
||||
"github.com/lukaszraczylo/llm-testbench/internal/runner"
|
||||
"github.com/lukaszraczylo/llm-testbench/internal/testkit"
|
||||
)
|
||||
|
||||
@@ -101,3 +105,22 @@ func TestHealthCommand_RequiresArtifacts(t *testing.T) {
|
||||
t.Errorf("healthCommand(missing file) error = %v, want load error naming the file", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDedupe(t *testing.T) {
|
||||
got := dedupe([]string{"a", "b", "a", "c", "b"})
|
||||
if strings.Join(got, ",") != "a,b,c" {
|
||||
t.Errorf("dedupe = %v, want [a b c]", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDropCancelled(t *testing.T) {
|
||||
in := []runner.Result{
|
||||
{TestID: "ok"},
|
||||
{TestID: "cancelled", Err: fmt.Errorf("wrapped: %w", context.Canceled)},
|
||||
{TestID: "real-error", Err: errors.New("boom")},
|
||||
}
|
||||
got := dropCancelled(in)
|
||||
if len(got) != 2 || got[0].TestID != "ok" || got[1].TestID != "real-error" {
|
||||
t.Errorf("dropCancelled = %+v, want ok and real-error kept", got)
|
||||
}
|
||||
}
|
||||
|
||||
+91
-13
@@ -4,7 +4,9 @@ import (
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/lukaszraczylo/llm-testbench/internal/runner"
|
||||
)
|
||||
@@ -13,24 +15,40 @@ import (
|
||||
// line. A piped log stays readable without one line per request.
|
||||
const pipedProgressEvery = 25
|
||||
|
||||
// pipedProgressInterval also forces a piped line after this much time, so
|
||||
// a run of slow models still logs progress between the count-based lines.
|
||||
const pipedProgressInterval = time.Minute
|
||||
|
||||
// barWidth is the number of cells in the terminal progress bar.
|
||||
const barWidth = 30
|
||||
|
||||
const (
|
||||
barFilled = "\u2588"
|
||||
barEmpty = "\u2591"
|
||||
)
|
||||
|
||||
// stderrProgressReporter implements runner.ProgressReporter, writing a
|
||||
// self-updating "test N of M" counter to w (os.Stderr in production).
|
||||
// progress bar with elapsed time and ETA to w (os.Stderr in production).
|
||||
// Run's stdout stays pipeable (table/markdown/json only): all progress
|
||||
// goes here instead.
|
||||
//
|
||||
// When w is a terminal the counter rewrites itself in place with a
|
||||
// carriage return. When w is piped (a log file), rewriting would produce
|
||||
// one unreadable blob, so it prints a plain line every pipedProgressEvery
|
||||
// completions plus the final one.
|
||||
// When w is a terminal the bar rewrites itself in place with a carriage
|
||||
// return. When w is piped (a log file), rewriting would produce one
|
||||
// unreadable blob, so it prints a plain line every pipedProgressEvery
|
||||
// completions or pipedProgressInterval, plus the final one.
|
||||
type stderrProgressReporter struct {
|
||||
w io.Writer
|
||||
mu sync.Mutex
|
||||
isTTY bool
|
||||
w io.Writer
|
||||
now func() time.Time
|
||||
start time.Time
|
||||
lastPrint time.Time
|
||||
mu sync.Mutex
|
||||
errors int
|
||||
isTTY bool
|
||||
}
|
||||
|
||||
// newStderrProgressReporter builds a stderrProgressReporter writing to w.
|
||||
func newStderrProgressReporter(w io.Writer) *stderrProgressReporter {
|
||||
return &stderrProgressReporter{w: w, isTTY: isTerminal(w)}
|
||||
return &stderrProgressReporter{w: w, isTTY: isTerminal(w), now: time.Now}
|
||||
}
|
||||
|
||||
// isTerminal reports whether w is a character device (an interactive
|
||||
@@ -51,6 +69,8 @@ func isTerminal(w io.Writer) bool {
|
||||
func (r *stderrProgressReporter) ReportStart(totalTests, totalModels, concurrency int) {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
r.start = r.now()
|
||||
r.lastPrint = r.start
|
||||
// A write failure on this side-channel progress stream (a closed pipe,
|
||||
// a full/broken stderr) is not actionable and must not abort the run
|
||||
// that produces the actual report on stdout.
|
||||
@@ -61,22 +81,80 @@ func (r *stderrProgressReporter) ReportStart(totalTests, totalModels, concurrenc
|
||||
// ReportDone implements runner.ProgressReporter. Safe for concurrent calls
|
||||
// from up to Config.Concurrency goroutines at once: the mutex serializes
|
||||
// writes so concurrent completions cannot interleave mid-line.
|
||||
func (r *stderrProgressReporter) ReportDone(done, total int, _ runner.Result) {
|
||||
func (r *stderrProgressReporter) ReportDone(done, total int, res runner.Result) {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
|
||||
if res.Err != nil {
|
||||
r.errors++
|
||||
}
|
||||
now := r.now()
|
||||
elapsed := now.Sub(r.start)
|
||||
|
||||
if r.isTTY {
|
||||
// \r returns to the line start, \x1b[K clears any longer previous
|
||||
// text; the final completion ends the line so later output starts
|
||||
// clean.
|
||||
_, _ = fmt.Fprintf(r.w, "\rtest %d of %d\x1b[K", done, total)
|
||||
_, _ = fmt.Fprintf(r.w, "\r%s\x1b[K", r.statusLine(done, total, elapsed, true))
|
||||
if done == total {
|
||||
_, _ = fmt.Fprintln(r.w)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
if done%pipedProgressEvery == 0 || done == total {
|
||||
_, _ = fmt.Fprintf(r.w, "test %d of %d\n", done, total)
|
||||
if done%pipedProgressEvery == 0 || done == total || now.Sub(r.lastPrint) >= pipedProgressInterval {
|
||||
r.lastPrint = now
|
||||
_, _ = fmt.Fprintln(r.w, r.statusLine(done, total, elapsed, false))
|
||||
}
|
||||
}
|
||||
|
||||
// statusLine renders the progress text; withBar adds the bar for terminals.
|
||||
func (r *stderrProgressReporter) statusLine(done, total int, elapsed time.Duration, withBar bool) string {
|
||||
pct := 0
|
||||
if total > 0 {
|
||||
pct = done * 100 / total
|
||||
}
|
||||
var b strings.Builder
|
||||
if withBar {
|
||||
filled := 0
|
||||
if total > 0 {
|
||||
filled = done * barWidth / total
|
||||
}
|
||||
b.WriteString(strings.Repeat(barFilled, filled) + strings.Repeat(barEmpty, barWidth-filled) + " ")
|
||||
}
|
||||
fmt.Fprintf(&b, "%3d%% %d/%d elapsed %s", pct, done, total, formatDuration(elapsed))
|
||||
if done < total {
|
||||
fmt.Fprintf(&b, " ETA %s", formatETA(done, total, elapsed))
|
||||
}
|
||||
if r.errors > 0 {
|
||||
fmt.Fprintf(&b, " errors %d", r.errors)
|
||||
}
|
||||
return b.String()
|
||||
}
|
||||
|
||||
// formatETA extrapolates the remaining time from the average completion
|
||||
// rate so far; "~" marks it as an estimate and "--" means no data yet.
|
||||
// Jobs vary widely in duration, so it is rough, especially early on.
|
||||
func formatETA(done, total int, elapsed time.Duration) string {
|
||||
if done <= 0 {
|
||||
return "--"
|
||||
}
|
||||
remaining := time.Duration(float64(elapsed) / float64(done) * float64(total-done))
|
||||
return "~" + formatDuration(remaining)
|
||||
}
|
||||
|
||||
// formatDuration renders d rounded to seconds as 45s, 12m04s or 1h02m.
|
||||
func formatDuration(d time.Duration) string {
|
||||
d = d.Round(time.Second)
|
||||
h := int(d / time.Hour)
|
||||
m := int(d % time.Hour / time.Minute)
|
||||
sec := int(d % time.Minute / time.Second)
|
||||
switch {
|
||||
case h > 0:
|
||||
return fmt.Sprintf("%dh%02dm", h, m)
|
||||
case m > 0:
|
||||
return fmt.Sprintf("%dm%02ds", m, sec)
|
||||
default:
|
||||
return fmt.Sprintf("%ds", sec)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,10 +2,12 @@ package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"errors"
|
||||
"regexp"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/lukaszraczylo/llm-testbench/internal/eval"
|
||||
"github.com/lukaszraczylo/llm-testbench/internal/runner"
|
||||
@@ -24,9 +26,26 @@ func TestStderrProgressReporter_ReportStart(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// fakeClock advances only when told, so elapsed/ETA are exact.
|
||||
type fakeClock struct{ t time.Time }
|
||||
|
||||
func (c *fakeClock) now() time.Time { return c.t }
|
||||
|
||||
func newTestReporter(buf *bytes.Buffer, tty bool) (*stderrProgressReporter, *fakeClock) {
|
||||
clk := &fakeClock{t: time.Unix(1_000_000, 0)}
|
||||
r := newStderrProgressReporter(buf)
|
||||
r.isTTY = tty
|
||||
r.now = clk.now
|
||||
r.ReportStart(10, 1, 1)
|
||||
buf.Reset()
|
||||
return r, clk
|
||||
}
|
||||
|
||||
var okResult = runner.Result{Model: "m", TestID: "t", Score: eval.Score{Value: 1}}
|
||||
|
||||
// A bytes.Buffer is not a character device, so newStderrProgressReporter
|
||||
// selects piped mode: a plain line every pipedProgressEvery completions
|
||||
// plus the final one, nothing else.
|
||||
// (or pipedProgressInterval), plus the final one.
|
||||
func TestStderrProgressReporter_ReportDone_Piped(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
@@ -35,17 +54,16 @@ func TestStderrProgressReporter_ReportDone_Piped(t *testing.T) {
|
||||
total int
|
||||
}{
|
||||
{name: "intermediate completion stays silent", done: 3, total: 10, want: ""},
|
||||
{name: "final completion always prints", done: 10, total: 10, want: "test 10 of 10\n"},
|
||||
{name: "every pipedProgressEvery-th completion prints", done: pipedProgressEvery, total: 100, want: "test 25 of 100\n"},
|
||||
{name: "multiple of pipedProgressEvery prints", done: 2 * pipedProgressEvery, total: 100, want: "test 50 of 100\n"},
|
||||
{name: "final completion always prints", done: 10, total: 10, want: "100% 10/10 elapsed 10s\n"},
|
||||
{name: "every pipedProgressEvery-th completion prints", done: pipedProgressEvery, total: 100, want: " 25% 25/100 elapsed 10s ETA ~30s\n"},
|
||||
{name: "off-interval completion stays silent", done: 2*pipedProgressEvery + 1, total: 100, want: ""},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
var buf bytes.Buffer
|
||||
r := newStderrProgressReporter(&buf)
|
||||
r.ReportDone(tt.done, tt.total, runner.Result{Model: "m", TestID: "t", Score: eval.Score{Value: 1}})
|
||||
|
||||
r, clk := newTestReporter(&buf, false)
|
||||
clk.t = clk.t.Add(10 * time.Second)
|
||||
r.ReportDone(tt.done, tt.total, okResult)
|
||||
if got := buf.String(); got != tt.want {
|
||||
t.Errorf("ReportDone(%d, %d) output = %q, want %q", tt.done, tt.total, got, tt.want)
|
||||
}
|
||||
@@ -53,39 +71,77 @@ func TestStderrProgressReporter_ReportDone_Piped(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TTY mode rewrites one counter line in place; the test forces isTTY since
|
||||
// a bytes.Buffer can never be a terminal.
|
||||
func TestStderrProgressReporter_ReportDone_PipedTimeBasedLine(t *testing.T) {
|
||||
var buf bytes.Buffer
|
||||
r, clk := newTestReporter(&buf, false)
|
||||
clk.t = clk.t.Add(pipedProgressInterval - time.Second)
|
||||
r.ReportDone(1, 10, okResult)
|
||||
if buf.Len() != 0 {
|
||||
t.Fatalf("printed before the interval: %q", buf.String())
|
||||
}
|
||||
clk.t = clk.t.Add(2 * time.Second)
|
||||
r.ReportDone(2, 10, okResult)
|
||||
if !strings.Contains(buf.String(), "2/10") {
|
||||
t.Errorf("no time-based line after the interval, got %q", buf.String())
|
||||
}
|
||||
}
|
||||
|
||||
// TTY mode rewrites one bar line in place; the test forces isTTY since a
|
||||
// bytes.Buffer can never be a terminal.
|
||||
func TestStderrProgressReporter_ReportDone_TTY(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
want string
|
||||
done int
|
||||
total int
|
||||
}{
|
||||
{name: "intermediate rewrites in place without a newline", done: 3, total: 10, want: "\rtest 3 of 10\x1b[K"},
|
||||
{name: "final completion ends the line", done: 10, total: 10, want: "\rtest 10 of 10\x1b[K\n"},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
var buf bytes.Buffer
|
||||
r := newStderrProgressReporter(&buf)
|
||||
r.isTTY = true
|
||||
r.ReportDone(tt.done, tt.total, runner.Result{Model: "m", TestID: "t", Score: eval.Score{Value: 1}})
|
||||
var buf bytes.Buffer
|
||||
r, clk := newTestReporter(&buf, true)
|
||||
clk.t = clk.t.Add(30 * time.Second)
|
||||
|
||||
if got := buf.String(); got != tt.want {
|
||||
t.Errorf("ReportDone(%d, %d) output = %q, want %q", tt.done, tt.total, got, tt.want)
|
||||
}
|
||||
})
|
||||
r.ReportDone(5, 10, okResult)
|
||||
want := "\r" + strings.Repeat(barFilled, 15) + strings.Repeat(barEmpty, 15) + " 50% 5/10 elapsed 30s ETA ~30s\x1b[K"
|
||||
if got := buf.String(); got != want {
|
||||
t.Errorf("mid-run output = %q, want %q", got, want)
|
||||
}
|
||||
|
||||
buf.Reset()
|
||||
r.ReportDone(10, 10, okResult)
|
||||
want = "\r" + strings.Repeat(barFilled, barWidth) + " 100% 10/10 elapsed 30s\x1b[K\n"
|
||||
if got := buf.String(); got != want {
|
||||
t.Errorf("final output = %q, want %q (full bar, no ETA, line ended)", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
// TestStderrProgressReporter_ReportDone_ConcurrencySafe writes many
|
||||
// concurrent ReportDone calls (as Runner.Run's goroutines would) and
|
||||
// verifies no write interleaved mid-line - the reporter's own mutex must
|
||||
// serialize writes, not just the counter.
|
||||
func TestStderrProgressReporter_CountsErrors(t *testing.T) {
|
||||
var buf bytes.Buffer
|
||||
r, _ := newTestReporter(&buf, false)
|
||||
r.ReportDone(1, 2, runner.Result{Err: errors.New("boom")})
|
||||
r.ReportDone(2, 2, okResult)
|
||||
if got := buf.String(); !strings.Contains(got, "errors 1") {
|
||||
t.Errorf("output = %q, want an errors count of 1", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFormatDuration(t *testing.T) {
|
||||
tests := map[time.Duration]string{
|
||||
0: "0s",
|
||||
44*time.Second + 600*time.Millisecond: "45s",
|
||||
12*time.Minute + 4*time.Second: "12m04s",
|
||||
time.Hour + 2*time.Minute + 30*time.Second: "1h02m",
|
||||
}
|
||||
for in, want := range tests {
|
||||
if got := formatDuration(in); got != want {
|
||||
t.Errorf("formatDuration(%v) = %q, want %q", in, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestFormatETA_NoDataYet(t *testing.T) {
|
||||
if got := formatETA(0, 10, time.Minute); got != "--" {
|
||||
t.Errorf("formatETA with nothing done = %q, want --", got)
|
||||
}
|
||||
}
|
||||
|
||||
// Concurrent ReportDone calls (as Runner.Run's goroutines would) must not
|
||||
// interleave mid-line: the reporter's own mutex serializes the writes.
|
||||
func TestStderrProgressReporter_ReportDone_ConcurrencySafe(t *testing.T) {
|
||||
var buf bytes.Buffer
|
||||
r := newStderrProgressReporter(&buf)
|
||||
r, _ := newTestReporter(&buf, false)
|
||||
|
||||
const n = 2 * pipedProgressEvery
|
||||
var wg sync.WaitGroup
|
||||
@@ -93,18 +149,18 @@ func TestStderrProgressReporter_ReportDone_ConcurrencySafe(t *testing.T) {
|
||||
wg.Add(1)
|
||||
go func(i int) {
|
||||
defer wg.Done()
|
||||
r.ReportDone(i, n, runner.Result{Model: "m", TestID: "t", Score: eval.Score{Value: 1}})
|
||||
r.ReportDone(i, n, okResult)
|
||||
}(i)
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
// In piped mode exactly the 25th and 50th completions print.
|
||||
// Completion order is arbitrary, so the 25th and 50th calls to arrive
|
||||
// print: exactly two lines.
|
||||
lines := strings.Split(strings.TrimRight(buf.String(), "\n"), "\n")
|
||||
if len(lines) != 2 {
|
||||
t.Fatalf("got %d lines (%q), want 2 (a corrupted/interleaved write would change the count)", len(lines), buf.String())
|
||||
}
|
||||
|
||||
linePattern := regexp.MustCompile(`^test \d+ of \d+$`)
|
||||
linePattern := regexp.MustCompile(`^\s*\d+% \d+/\d+ elapsed \d+s( ETA ~\d+s)?$`)
|
||||
for _, line := range lines {
|
||||
if !linePattern.MatchString(line) {
|
||||
t.Errorf("malformed or interleaved line: %q", line)
|
||||
|
||||
+14
-2
@@ -5,7 +5,19 @@ models:
|
||||
- uni/qwen3.6-27b
|
||||
- uni/btl-4
|
||||
concurrency: 8 # default when omitted; matches internal/config's default
|
||||
request_timeout: 900s # default when omitted
|
||||
max_tokens_default: 12000 # default when omitted; a reasoning model can spend
|
||||
request_timeout: 900s # default when omitted is 300s
|
||||
max_tokens_default: 40000 # default when omitted; a reasoning model can spend
|
||||
# thousands of tokens on reasoning_content before
|
||||
# writing an answer, so this must stay generous
|
||||
max_retries: 2 # retries per request after the first try (default 2; 0 disables).
|
||||
# Raise for flaky gateways: backoff is exponential and
|
||||
# honors Retry-After.
|
||||
# seed: 1234 # optional; sent with every request for backends that honor it
|
||||
# Slow local models: give them their own timeout and queue depth.
|
||||
# retry_timeouts: false # default true; false stops a timed-out call re-running a full slow generation
|
||||
# model_timeouts: # overrides request_timeout per model
|
||||
# local/big-model: 45m
|
||||
# model_concurrency: # caps in-flight calls per model (global concurrency still applies)
|
||||
# local/big-model: 1
|
||||
# stream: false # true = SSE; detects a hung server after idle_timeout instead of the full request timeout
|
||||
# idle_timeout: 5m # max silence in a stream (must exceed prompt-processing time); default 5m
|
||||
|
||||
+78
-12
@@ -3,6 +3,7 @@ package config
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net/url"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
@@ -27,27 +28,42 @@ const (
|
||||
defaultConcurrency = 8
|
||||
defaultRequestTimeout = 300 * time.Second
|
||||
defaultMaxTokens = 40000
|
||||
defaultMaxRetries = 2
|
||||
)
|
||||
|
||||
// Config is the top-level configuration loaded from config.yaml.
|
||||
type Config struct {
|
||||
Endpoint string `yaml:"endpoint"`
|
||||
APIKey string `yaml:"api_key"`
|
||||
Models []string `yaml:"models"`
|
||||
Concurrency int `yaml:"concurrency"`
|
||||
RequestTimeout time.Duration `yaml:"request_timeout"`
|
||||
MaxTokensDefault int `yaml:"max_tokens_default"`
|
||||
Seed *int `yaml:"seed"`
|
||||
ModelTimeouts map[string]time.Duration `yaml:"-"`
|
||||
ModelConcurrency map[string]int `yaml:"model_concurrency"`
|
||||
Endpoint string `yaml:"endpoint"`
|
||||
APIKey string `yaml:"api_key"`
|
||||
Models []string `yaml:"models"`
|
||||
Concurrency int `yaml:"concurrency"`
|
||||
RequestTimeout time.Duration `yaml:"request_timeout"`
|
||||
MaxTokensDefault int `yaml:"max_tokens_default"`
|
||||
MaxRetries int `yaml:"max_retries"`
|
||||
IdleTimeout time.Duration `yaml:"-"`
|
||||
RetryTimeouts bool `yaml:"retry_timeouts"`
|
||||
Stream bool `yaml:"stream"`
|
||||
}
|
||||
|
||||
// rawConfig mirrors Config but with RequestTimeout as a string, since
|
||||
// yaml.v3 does not natively decode Go duration strings.
|
||||
type rawConfig struct {
|
||||
Endpoint string `yaml:"endpoint"`
|
||||
APIKey string `yaml:"api_key"`
|
||||
RequestTimeout string `yaml:"request_timeout"`
|
||||
Models []string `yaml:"models"`
|
||||
Concurrency int `yaml:"concurrency"`
|
||||
MaxTokensDefault int `yaml:"max_tokens_default"`
|
||||
MaxRetries *int `yaml:"max_retries"`
|
||||
Seed *int `yaml:"seed"`
|
||||
RetryTimeouts *bool `yaml:"retry_timeouts"`
|
||||
ModelTimeouts map[string]string `yaml:"model_timeouts"`
|
||||
ModelConcurrency map[string]int `yaml:"model_concurrency"`
|
||||
APIKey string `yaml:"api_key"`
|
||||
Endpoint string `yaml:"endpoint"`
|
||||
RequestTimeout string `yaml:"request_timeout"`
|
||||
IdleTimeout string `yaml:"idle_timeout"`
|
||||
Models []string `yaml:"models"`
|
||||
Concurrency int `yaml:"concurrency"`
|
||||
MaxTokensDefault int `yaml:"max_tokens_default"`
|
||||
Stream bool `yaml:"stream"`
|
||||
}
|
||||
|
||||
// Load reads and parses the YAML config at path, applies the LLMTB_API_KEY
|
||||
@@ -78,6 +94,31 @@ func parse(data []byte) (Config, error) {
|
||||
Models: raw.Models,
|
||||
Concurrency: raw.Concurrency,
|
||||
MaxTokensDefault: raw.MaxTokensDefault,
|
||||
MaxRetries: defaultMaxRetries,
|
||||
Seed: raw.Seed,
|
||||
}
|
||||
if raw.MaxRetries != nil {
|
||||
cfg.MaxRetries = *raw.MaxRetries
|
||||
}
|
||||
cfg.RetryTimeouts = raw.RetryTimeouts == nil || *raw.RetryTimeouts
|
||||
cfg.ModelConcurrency = raw.ModelConcurrency
|
||||
cfg.Stream = raw.Stream
|
||||
if raw.IdleTimeout != "" {
|
||||
d, err := time.ParseDuration(raw.IdleTimeout)
|
||||
if err != nil {
|
||||
return Config{}, fmt.Errorf("config: parse idle_timeout %q: %w", raw.IdleTimeout, err)
|
||||
}
|
||||
cfg.IdleTimeout = d
|
||||
}
|
||||
if len(raw.ModelTimeouts) > 0 {
|
||||
cfg.ModelTimeouts = make(map[string]time.Duration, len(raw.ModelTimeouts))
|
||||
for m, v := range raw.ModelTimeouts {
|
||||
d, err := time.ParseDuration(v)
|
||||
if err != nil {
|
||||
return Config{}, fmt.Errorf("config: parse model_timeouts[%q] %q: %w", m, v, err)
|
||||
}
|
||||
cfg.ModelTimeouts[m] = d
|
||||
}
|
||||
}
|
||||
|
||||
if raw.RequestTimeout != "" {
|
||||
@@ -123,10 +164,35 @@ func (c Config) Validate() error {
|
||||
if len(c.Models) == 0 {
|
||||
return fmt.Errorf("config: models must list at least one model")
|
||||
}
|
||||
u, err := url.Parse(c.Endpoint)
|
||||
if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" {
|
||||
return fmt.Errorf("config: endpoint %q must be an http(s) URL", c.Endpoint)
|
||||
}
|
||||
seen := make(map[string]bool, len(c.Models))
|
||||
for i, m := range c.Models {
|
||||
if m == "" {
|
||||
return fmt.Errorf("config: models[%d] must not be empty", i)
|
||||
}
|
||||
if seen[m] {
|
||||
return fmt.Errorf("config: models[%d] duplicates %q", i, m)
|
||||
}
|
||||
seen[m] = true
|
||||
}
|
||||
for m, d := range c.ModelTimeouts {
|
||||
if d <= 0 {
|
||||
return fmt.Errorf("config: model_timeouts[%q] must be > 0, got %s", m, d)
|
||||
}
|
||||
}
|
||||
for m, n := range c.ModelConcurrency {
|
||||
if n <= 0 {
|
||||
return fmt.Errorf("config: model_concurrency[%q] must be > 0, got %d", m, n)
|
||||
}
|
||||
}
|
||||
if c.IdleTimeout < 0 {
|
||||
return fmt.Errorf("config: idle_timeout must be >= 0, got %s", c.IdleTimeout)
|
||||
}
|
||||
if c.MaxRetries < 0 {
|
||||
return fmt.Errorf("config: max_retries must be >= 0, got %d", c.MaxRetries)
|
||||
}
|
||||
if c.Concurrency <= 0 {
|
||||
return fmt.Errorf("config: concurrency must be > 0, got %d", c.Concurrency)
|
||||
|
||||
@@ -160,6 +160,9 @@ func TestValidate(t *testing.T) {
|
||||
{name: "zero concurrency", mutate: func(c Config) Config { c.Concurrency = 0; return c }},
|
||||
{name: "negative concurrency", mutate: func(c Config) Config { c.Concurrency = -1; return c }},
|
||||
{name: "zero timeout", mutate: func(c Config) Config { c.RequestTimeout = 0; return c }},
|
||||
{name: "non-http endpoint", mutate: func(c Config) Config { c.Endpoint = "llm-gateway.example.com/v1"; return c }},
|
||||
{name: "duplicate model", mutate: func(c Config) Config { c.Models = []string{"m", "m"}; return c }},
|
||||
{name: "negative retries", mutate: func(c Config) Config { c.MaxRetries = -1; return c }},
|
||||
{name: "zero max tokens", mutate: func(c Config) Config { c.MaxTokensDefault = 0; return c }},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
@@ -194,3 +197,65 @@ func TestLoad_MissingFile(t *testing.T) {
|
||||
t.Fatal("Load() error = nil, want error for missing file")
|
||||
}
|
||||
}
|
||||
|
||||
func TestParse_RetriesAndSeed(t *testing.T) {
|
||||
base := "endpoint: https://x\nmodels: [m]\n"
|
||||
|
||||
cfg, err := parse([]byte(base))
|
||||
if err != nil {
|
||||
t.Fatalf("parse() error = %v", err)
|
||||
}
|
||||
if cfg.MaxRetries != defaultMaxRetries || cfg.Seed != nil {
|
||||
t.Errorf("defaults = retries %d seed %v, want %d nil", cfg.MaxRetries, cfg.Seed, defaultMaxRetries)
|
||||
}
|
||||
|
||||
cfg, err = parse([]byte(base + "max_retries: 0\nseed: 7\n"))
|
||||
if err != nil {
|
||||
t.Fatalf("parse() error = %v", err)
|
||||
}
|
||||
if cfg.MaxRetries != 0 {
|
||||
t.Errorf("explicit max_retries 0 = %d, want 0 (retrying disabled)", cfg.MaxRetries)
|
||||
}
|
||||
if cfg.Seed == nil || *cfg.Seed != 7 {
|
||||
t.Errorf("Seed = %v, want 7", cfg.Seed)
|
||||
}
|
||||
}
|
||||
|
||||
func TestParse_ModelOverrides(t *testing.T) {
|
||||
cfg, err := parse([]byte(`
|
||||
endpoint: https://x
|
||||
models: [a, b]
|
||||
retry_timeouts: false
|
||||
model_timeouts: {a: 45m}
|
||||
model_concurrency: {a: 1}
|
||||
`))
|
||||
if err != nil {
|
||||
t.Fatalf("parse() error = %v", err)
|
||||
}
|
||||
if cfg.RetryTimeouts || cfg.ModelTimeouts["a"] != 45*time.Minute || cfg.ModelConcurrency["a"] != 1 {
|
||||
t.Errorf("got %+v", cfg)
|
||||
}
|
||||
def, _ := parse([]byte("endpoint: https://x\nmodels: [a]\n"))
|
||||
if !def.RetryTimeouts {
|
||||
t.Error("retry_timeouts default = false, want true (legacy behavior)")
|
||||
}
|
||||
for _, bad := range []string{"model_timeouts: {a: nope}", "model_timeouts: {a: 0s}", "model_concurrency: {a: 0}"} {
|
||||
if _, err := parse([]byte("endpoint: https://x\nmodels: [a]\n" + bad)); err == nil {
|
||||
t.Errorf("parse(%q) error = nil, want error", bad)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestParse_Streaming(t *testing.T) {
|
||||
cfg, err := parse([]byte("endpoint: https://x\nmodels: [a]\nstream: true\nidle_timeout: 10m\n"))
|
||||
if err != nil || !cfg.Stream || cfg.IdleTimeout != 10*time.Minute {
|
||||
t.Fatalf("cfg = %+v, err = %v", cfg, err)
|
||||
}
|
||||
def, _ := parse([]byte("endpoint: https://x\nmodels: [a]\n"))
|
||||
if def.Stream {
|
||||
t.Error("stream default = true, want false")
|
||||
}
|
||||
if _, err := parse([]byte("endpoint: https://x\nmodels: [a]\nidle_timeout: nope\n")); err == nil {
|
||||
t.Error("bad idle_timeout accepted")
|
||||
}
|
||||
}
|
||||
|
||||
+91
-53
@@ -8,6 +8,7 @@ import (
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"runtime"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -39,8 +40,71 @@ func stripLeadingPackageClause(code string) string {
|
||||
return code
|
||||
}
|
||||
|
||||
// execTimeout bounds every compile/run step of the exec evaluators.
|
||||
const execTimeout = 30 * time.Second
|
||||
// execTimeout bounds every run step (and the C compile) of the exec
|
||||
// evaluators; goBuildTimeout is the longer allowance for a cold `go build`.
|
||||
const (
|
||||
execTimeout = 30 * time.Second
|
||||
goBuildTimeout = 120 * time.Second
|
||||
)
|
||||
|
||||
// execWaitDelay bounds how long Wait lingers for output pipes after the
|
||||
// process is killed; without it a grandchild holding stdout open hangs the
|
||||
// whole run past execTimeout.
|
||||
const execWaitDelay = 2 * time.Second
|
||||
|
||||
// maxExecOutput caps captured stdout/stderr per call, so a runaway print
|
||||
// loop cannot exhaust memory before the timeout fires.
|
||||
const maxExecOutput = 1 << 20
|
||||
|
||||
// execSlots limits concurrent compile/run subprocesses. The runner issues up
|
||||
// to Concurrency scoring calls at once; unbounded, CPU starvation pushes
|
||||
// correct solutions past execTimeout and turns load into false zeros.
|
||||
var execSlots = make(chan struct{}, max(2, runtime.NumCPU()/2))
|
||||
|
||||
// cappedBuffer keeps the first maxExecOutput bytes written and discards the
|
||||
// rest, still reporting full writes so the child never sees a broken pipe.
|
||||
type cappedBuffer struct{ buf bytes.Buffer }
|
||||
|
||||
func (c *cappedBuffer) Write(p []byte) (int, error) {
|
||||
if room := maxExecOutput - c.buf.Len(); room > 0 {
|
||||
c.buf.Write(p[:min(len(p), room)])
|
||||
}
|
||||
return len(p), nil
|
||||
}
|
||||
|
||||
func (c *cappedBuffer) String() string { return c.buf.String() }
|
||||
|
||||
// runBounded runs name with args in dir under env, holding an exec slot,
|
||||
// with a hard timeout and capped output. A timeout is reported as such
|
||||
// rather than as a bare "signal: killed".
|
||||
func runBounded(ctx context.Context, timeout time.Duration, dir string, env []string, name string, args ...string) (stdout, stderr string, err error) {
|
||||
select {
|
||||
case execSlots <- struct{}{}:
|
||||
defer func() { <-execSlots }()
|
||||
case <-ctx.Done():
|
||||
return "", "", ctx.Err()
|
||||
}
|
||||
|
||||
runCtx, cancel := context.WithTimeout(ctx, timeout)
|
||||
defer cancel()
|
||||
|
||||
// #nosec G204 -- running model-generated code under an isolated temp dir
|
||||
// and a minimal env allowlist is this package's purpose; name is always
|
||||
// a fixed toolchain binary or one this package just compiled. See the
|
||||
// README "Requirements" warning (S13).
|
||||
cmd := exec.CommandContext(runCtx, name, args...) //nolint:gosec // see comment above
|
||||
cmd.Dir = dir
|
||||
cmd.Env = env
|
||||
cmd.WaitDelay = execWaitDelay
|
||||
var out, errOut cappedBuffer
|
||||
cmd.Stdout = &out
|
||||
cmd.Stderr = &errOut
|
||||
err = cmd.Run()
|
||||
if err != nil && runCtx.Err() == context.DeadlineExceeded {
|
||||
err = fmt.Errorf("timed out after %s: %w", timeout, err)
|
||||
}
|
||||
return out.String(), errOut.String(), err
|
||||
}
|
||||
|
||||
// CodePlaceholder marks where the extracted response code is substituted
|
||||
// into a harness template.
|
||||
@@ -93,10 +157,19 @@ func minimalExecEnv(homeDir string) []string {
|
||||
goModCacheDir = goEnvValue("GOMODCACHE")
|
||||
})
|
||||
|
||||
// Pinned so output does not vary with the operator's shell: timezone,
|
||||
// hash seeding and locale-dependent encoding. GOTOOLCHAIN=local stops
|
||||
// `go` fetching a different toolchain over the network mid-run.
|
||||
env := []string{
|
||||
"PATH=" + os.Getenv("PATH"),
|
||||
"HOME=" + homeDir,
|
||||
"TMPDIR=" + homeDir,
|
||||
"TZ=UTC",
|
||||
"PYTHONHASHSEED=0",
|
||||
"PYTHONUTF8=1",
|
||||
"PYTHONDONTWRITEBYTECODE=1",
|
||||
"GOTOOLCHAIN=local",
|
||||
"GOFLAGS=-buildvcs=false",
|
||||
}
|
||||
if goCacheDir != "" {
|
||||
env = append(env, "GOCACHE="+goCacheDir)
|
||||
@@ -199,18 +272,16 @@ func (g goRunEval) Evaluate(ctx context.Context, response string) Score {
|
||||
return Score{Value: 0, Detail: fmt.Sprintf("write go.mod: %v", modErr)}
|
||||
}
|
||||
|
||||
runCtx, cancel := context.WithTimeout(ctx, execTimeout)
|
||||
defer cancel()
|
||||
// Build and run as separate steps so a cold compile under load is not
|
||||
// charged against the run timeout.
|
||||
env := minimalExecEnv(root)
|
||||
binPath := filepath.Join(work, "prog")
|
||||
if _, buildStderr, buildErr := runBounded(ctx, goBuildTimeout, work, env, "go", "build", "-o", binPath, "."); buildErr != nil {
|
||||
return Score{Value: 0, Detail: fmt.Sprintf("go build failed: %v; stderr: %s", buildErr, strings.TrimSpace(buildStderr))}
|
||||
}
|
||||
stdout, stderr, err := runBounded(ctx, execTimeout, work, env, binPath)
|
||||
|
||||
cmd := exec.CommandContext(runCtx, "go", "run", ".")
|
||||
cmd.Dir = work
|
||||
cmd.Env = minimalExecEnv(root)
|
||||
var stdout, stderr bytes.Buffer
|
||||
cmd.Stdout = &stdout
|
||||
cmd.Stderr = &stderr
|
||||
err = cmd.Run()
|
||||
|
||||
return compareStdout(stdout.String(), g.want, err, stderr.String())
|
||||
return compareStdout(stdout, g.want, err, stderr)
|
||||
}
|
||||
|
||||
// pyRunEval runs a Python program built from a harness template plus the
|
||||
@@ -247,22 +318,9 @@ func (p pyRunEval) Evaluate(ctx context.Context, response string) Score {
|
||||
return Score{Value: 0, Detail: fmt.Sprintf("write script.py: %v", writeErr)}
|
||||
}
|
||||
|
||||
runCtx, cancel := context.WithTimeout(ctx, execTimeout)
|
||||
defer cancel()
|
||||
stdout, stderr, err := runBounded(ctx, execTimeout, work, minimalExecEnv(root), "python3", scriptPath)
|
||||
|
||||
// #nosec G204 -- this is the eval package's whole purpose: compile/run
|
||||
// model-generated code under an isolated temp dir and a minimal env
|
||||
// allowlist (minimalExecEnv). See README's "Requirements" section for
|
||||
// the corresponding operator-facing warning (S13).
|
||||
cmd := exec.CommandContext(runCtx, "python3", scriptPath) //nolint:gosec // see comment above
|
||||
cmd.Dir = work
|
||||
cmd.Env = minimalExecEnv(root)
|
||||
var stdout, stderr bytes.Buffer
|
||||
cmd.Stdout = &stdout
|
||||
cmd.Stderr = &stderr
|
||||
err = cmd.Run()
|
||||
|
||||
return compareStdout(stdout.String(), p.want, err, stderr.String())
|
||||
return compareStdout(stdout, p.want, err, stderr)
|
||||
}
|
||||
|
||||
// cRunEval compiles/runs a C program built from a harness template plus the
|
||||
@@ -300,32 +358,12 @@ func (c cRunEval) Evaluate(ctx context.Context, response string) Score {
|
||||
}
|
||||
binPath := filepath.Join(work, "prog")
|
||||
|
||||
buildCtx, buildCancel := context.WithTimeout(ctx, execTimeout)
|
||||
defer buildCancel()
|
||||
// #nosec G204 -- compiling model-generated C source under an isolated
|
||||
// temp dir and a minimal env allowlist is this evaluator's job; see the
|
||||
// PyRun comment above and README's "Requirements" warning (S13).
|
||||
buildCmd := exec.CommandContext(buildCtx, "cc", "-o", binPath, srcPath) //nolint:gosec // see comment above
|
||||
buildCmd.Dir = work
|
||||
buildCmd.Env = minimalExecEnv(root)
|
||||
var buildStderr bytes.Buffer
|
||||
buildCmd.Stderr = &buildStderr
|
||||
if buildErr := buildCmd.Run(); buildErr != nil {
|
||||
return Score{Value: 0, Detail: fmt.Sprintf("cc failed: %v; stderr: %s", buildErr, strings.TrimSpace(buildStderr.String()))}
|
||||
env := minimalExecEnv(root)
|
||||
if _, buildStderr, buildErr := runBounded(ctx, execTimeout, work, env, "cc", "-o", binPath, srcPath); buildErr != nil {
|
||||
return Score{Value: 0, Detail: fmt.Sprintf("cc failed: %v; stderr: %s", buildErr, strings.TrimSpace(buildStderr))}
|
||||
}
|
||||
|
||||
runCtx, runCancel := context.WithTimeout(ctx, execTimeout)
|
||||
defer runCancel()
|
||||
// #nosec G204 -- binPath is the binary this same call just compiled,
|
||||
// into a fresh, isolated temp dir this evaluator created; not an
|
||||
// externally supplied executable path.
|
||||
runCmd := exec.CommandContext(runCtx, binPath) //nolint:gosec // see comment above
|
||||
runCmd.Dir = work
|
||||
runCmd.Env = minimalExecEnv(root)
|
||||
var stdout, stderr bytes.Buffer
|
||||
runCmd.Stdout = &stdout
|
||||
runCmd.Stderr = &stderr
|
||||
err = runCmd.Run()
|
||||
stdout, stderr, err := runBounded(ctx, execTimeout, work, env, binPath)
|
||||
|
||||
return compareStdout(stdout.String(), c.want, err, stderr.String())
|
||||
return compareStdout(stdout, c.want, err, stderr)
|
||||
}
|
||||
|
||||
@@ -3,7 +3,9 @@ package eval
|
||||
import (
|
||||
"context"
|
||||
"os/exec"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func requireToolchain(t *testing.T, name string) {
|
||||
@@ -184,3 +186,46 @@ func TestExecEvaluators_ToolchainMissingSkipsNotFails(t *testing.T) {
|
||||
t.Errorf("CRun with empty PATH: Skipped = false, want true (%+v)", cScore)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunBounded_TimeoutIsReported(t *testing.T) {
|
||||
if _, err := exec.LookPath("sleep"); err != nil {
|
||||
t.Skip("sleep not available")
|
||||
}
|
||||
_, _, err := runBounded(context.Background(), 100*time.Millisecond, t.TempDir(), nil, "sleep", "5")
|
||||
if err == nil || !strings.Contains(err.Error(), "timed out after") {
|
||||
t.Errorf("err = %v, want a timed out error", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunBounded_OutputIsCapped(t *testing.T) {
|
||||
if _, err := exec.LookPath("sh"); err != nil {
|
||||
t.Skip("sh not available")
|
||||
}
|
||||
stdout, _, err := runBounded(context.Background(), 10*time.Second, t.TempDir(), nil, "sh", "-c", "yes | head -c 5000000")
|
||||
if err != nil {
|
||||
t.Fatalf("err = %v", err)
|
||||
}
|
||||
if len(stdout) != maxExecOutput {
|
||||
t.Errorf("captured %d bytes, want cap %d", len(stdout), maxExecOutput)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunBounded_GrandchildHoldingPipeDoesNotHang(t *testing.T) {
|
||||
if _, err := exec.LookPath("sh"); err != nil {
|
||||
t.Skip("sh not available")
|
||||
}
|
||||
start := time.Now()
|
||||
_, _, _ = runBounded(context.Background(), 200*time.Millisecond, t.TempDir(), nil, "sh", "-c", "sleep 30 & sleep 30")
|
||||
if elapsed := time.Since(start); elapsed > 10*time.Second {
|
||||
t.Errorf("took %v, want return shortly after timeout + WaitDelay", elapsed)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMinimalExecEnv_PinsDeterminismVars(t *testing.T) {
|
||||
env := strings.Join(minimalExecEnv(t.TempDir()), "\n")
|
||||
for _, want := range []string{"TZ=UTC", "PYTHONHASHSEED=0", "GOTOOLCHAIN=local"} {
|
||||
if !strings.Contains(env, want) {
|
||||
t.Errorf("env missing %s", want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -37,6 +37,7 @@ type ToolCall struct {
|
||||
// non-empty the client advertises them with tool_choice "auto", so the
|
||||
// model decides whether and which to call.
|
||||
type Request struct {
|
||||
Seed *int
|
||||
Model string
|
||||
Messages []Message
|
||||
Tools []Tool
|
||||
@@ -48,6 +49,8 @@ type Request struct {
|
||||
type Response struct {
|
||||
Text string
|
||||
FinishReason string
|
||||
Fingerprint string
|
||||
ServedModel string
|
||||
ToolCalls []ToolCall
|
||||
PromptTokens int
|
||||
CompletionTokens int
|
||||
|
||||
+187
-36
@@ -7,22 +7,46 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"math/rand/v2"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// maxRetries is the number of retry attempts after the initial request on
|
||||
// transport errors or 5xx responses, per PLAN.md ("retry w/ backoff x2").
|
||||
// maxRetries is the default number of retry attempts after the initial
|
||||
// request on transient failures, per PLAN.md ("retry w/ backoff x2").
|
||||
const maxRetries = 2
|
||||
|
||||
// retryBaseDelay is the base backoff delay; attempt N waits N*retryBaseDelay.
|
||||
// retryBaseDelay is the base backoff delay; attempt N waits
|
||||
// retryBaseDelay*2^(N-1), capped at maxRetryDelay, plus jitter.
|
||||
const retryBaseDelay = 250 * time.Millisecond
|
||||
|
||||
// maxRetryDelay caps both the exponential backoff and a server-sent
|
||||
// Retry-After so one bad gateway cannot stall a worker for minutes.
|
||||
const maxRetryDelay = 30 * time.Second
|
||||
|
||||
// maxResponseBytes bounds how much of a response body is read; a runaway
|
||||
// or hostile upstream must not exhaust memory.
|
||||
const maxResponseBytes = 32 << 20
|
||||
|
||||
// jitterFraction is the share of the backoff delay added as random jitter,
|
||||
// so concurrent workers retrying one saturated gateway do not stampede it
|
||||
// in lockstep.
|
||||
const jitterFraction = 0.25
|
||||
|
||||
// OpenAIClient talks to an OpenAI-compatible /v1/chat/completions endpoint.
|
||||
type OpenAIClient struct {
|
||||
httpClient *http.Client
|
||||
endpoint string
|
||||
apiKey string
|
||||
httpClient *http.Client
|
||||
modelTimeouts map[string]time.Duration
|
||||
endpoint string
|
||||
apiKey string
|
||||
retries int
|
||||
baseDelay time.Duration
|
||||
timeout time.Duration
|
||||
noRetryTimeouts bool
|
||||
stream bool
|
||||
idleTimeout time.Duration
|
||||
}
|
||||
|
||||
// NewOpenAIClient builds a client against endpoint (e.g.
|
||||
@@ -30,21 +54,83 @@ type OpenAIClient struct {
|
||||
// per-request timeout.
|
||||
func NewOpenAIClient(endpoint, apiKey string, timeout time.Duration) *OpenAIClient {
|
||||
return &OpenAIClient{
|
||||
endpoint: endpoint,
|
||||
apiKey: apiKey,
|
||||
httpClient: &http.Client{
|
||||
Timeout: timeout,
|
||||
},
|
||||
endpoint: strings.TrimRight(endpoint, "/"),
|
||||
apiKey: apiKey,
|
||||
retries: maxRetries,
|
||||
baseDelay: retryBaseDelay,
|
||||
timeout: timeout,
|
||||
// The per-attempt deadline is applied through the context so it can
|
||||
// differ per model; the client itself has no overall timeout.
|
||||
httpClient: &http.Client{},
|
||||
}
|
||||
}
|
||||
|
||||
// WithModelTimeouts sets per-model request timeouts that override the
|
||||
// default; a slow local model can be given far longer than a hosted one.
|
||||
func (c *OpenAIClient) WithModelTimeouts(m map[string]time.Duration) *OpenAIClient {
|
||||
c.modelTimeouts = m
|
||||
return c
|
||||
}
|
||||
|
||||
// WithRetryTimeouts controls whether an attempt that hit its timeout is
|
||||
// retried. Retrying re-runs a full slow generation, so for slow models
|
||||
// disabling it avoids multiplying the wait by the attempt count.
|
||||
func (c *OpenAIClient) WithRetryTimeouts(retry bool) *OpenAIClient {
|
||||
c.noRetryTimeouts = !retry
|
||||
return c
|
||||
}
|
||||
|
||||
// WithStreaming makes the client request SSE streams. The per-request
|
||||
// timeout stays a hard ceiling; idle (default 5m when zero) is how long the
|
||||
// stream may go silent before the attempt is abandoned, so a hung server is
|
||||
// detected early while a slow-but-progressing generation is left to finish.
|
||||
func (c *OpenAIClient) WithStreaming(enabled bool, idle time.Duration) *OpenAIClient {
|
||||
c.stream = enabled
|
||||
c.idleTimeout = idle
|
||||
if enabled && idle <= 0 {
|
||||
c.idleTimeout = defaultIdleTimeout
|
||||
}
|
||||
return c
|
||||
}
|
||||
|
||||
func (c *OpenAIClient) timeoutFor(model string) time.Duration {
|
||||
if d, ok := c.modelTimeouts[model]; ok && d > 0 {
|
||||
return d
|
||||
}
|
||||
return c.timeout
|
||||
}
|
||||
|
||||
// WithRetries sets the number of retry attempts after the initial request
|
||||
// (0 disables retrying) and returns the client.
|
||||
func (c *OpenAIClient) WithRetries(n int) *OpenAIClient {
|
||||
c.retries = max(n, 0)
|
||||
return c
|
||||
}
|
||||
|
||||
// backoff returns the wait before retry number attempt (1-based): the
|
||||
// server's Retry-After when it sent one, else exponential with jitter.
|
||||
func (c *OpenAIClient) backoff(attempt int, retryAfter time.Duration) time.Duration {
|
||||
if retryAfter > 0 {
|
||||
return min(retryAfter, maxRetryDelay)
|
||||
}
|
||||
d := min(c.baseDelay<<(attempt-1), maxRetryDelay)
|
||||
return d + time.Duration(rand.Float64()*jitterFraction*float64(d)) //nolint:gosec // jitter needs no cryptographic randomness
|
||||
}
|
||||
|
||||
type chatCompletionRequest struct {
|
||||
Model string `json:"model"`
|
||||
ToolChoice string `json:"tool_choice,omitempty"`
|
||||
Messages []chatMsg `json:"messages"`
|
||||
Tools []apiTool `json:"tools,omitempty"`
|
||||
MaxTokens int `json:"max_tokens,omitempty"`
|
||||
Temperature float64 `json:"temperature"`
|
||||
Seed *int `json:"seed,omitempty"`
|
||||
StreamOptions *streamOptions `json:"stream_options,omitempty"`
|
||||
Model string `json:"model"`
|
||||
ToolChoice string `json:"tool_choice,omitempty"`
|
||||
Messages []chatMsg `json:"messages"`
|
||||
Tools []apiTool `json:"tools,omitempty"`
|
||||
MaxTokens int `json:"max_tokens,omitempty"`
|
||||
Temperature float64 `json:"temperature"`
|
||||
Stream bool `json:"stream,omitempty"`
|
||||
}
|
||||
|
||||
type streamOptions struct {
|
||||
IncludeUsage bool `json:"include_usage"`
|
||||
}
|
||||
|
||||
type chatMsg struct {
|
||||
@@ -102,7 +188,9 @@ type chatCompletionResponse struct {
|
||||
Error *struct {
|
||||
Message string `json:"message"`
|
||||
} `json:"error"`
|
||||
Choices []struct {
|
||||
Model string `json:"model"`
|
||||
SystemFingerprint string `json:"system_fingerprint"`
|
||||
Choices []struct {
|
||||
FinishReason string `json:"finish_reason"`
|
||||
Message responseMsg `json:"message"`
|
||||
} `json:"choices"`
|
||||
@@ -145,6 +233,7 @@ func (c *OpenAIClient) Complete(ctx context.Context, req Request) (Response, err
|
||||
Model: req.Model,
|
||||
MaxTokens: req.MaxTokens,
|
||||
Temperature: req.Temperature,
|
||||
Seed: req.Seed,
|
||||
}
|
||||
for _, m := range req.Messages {
|
||||
body.Messages = append(body.Messages, chatMsg(m))
|
||||
@@ -161,15 +250,21 @@ func (c *OpenAIClient) Complete(ctx context.Context, req Request) (Response, err
|
||||
}
|
||||
}
|
||||
|
||||
if c.stream {
|
||||
body.Stream = true
|
||||
body.StreamOptions = &streamOptions{IncludeUsage: true}
|
||||
}
|
||||
|
||||
payload, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return Response{}, fmt.Errorf("llm: marshal request: %w", err)
|
||||
}
|
||||
|
||||
var lastErr error
|
||||
for attempt := 0; attempt <= maxRetries; attempt++ {
|
||||
var retryAfter time.Duration
|
||||
for attempt := 0; attempt <= c.retries; attempt++ {
|
||||
if attempt > 0 {
|
||||
delay := time.Duration(attempt) * retryBaseDelay
|
||||
delay := c.backoff(attempt, retryAfter)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return Response{}, ctx.Err()
|
||||
@@ -181,16 +276,32 @@ func (c *OpenAIClient) Complete(ctx context.Context, req Request) (Response, err
|
||||
// reflects the successful attempt's own generation time, not the
|
||||
// cumulative wait across earlier failed/retried attempts (S8).
|
||||
attemptStart := time.Now()
|
||||
resp, err := c.doRequest(ctx, payload)
|
||||
attemptCtx, cancel := context.WithTimeout(ctx, c.timeoutFor(req.Model))
|
||||
resp, err := c.doRequest(attemptCtx, payload)
|
||||
timedOut := (attemptCtx.Err() == context.DeadlineExceeded && ctx.Err() == nil) || errors.Is(err, errIdleTimeout)
|
||||
cancel()
|
||||
retryAfter = 0
|
||||
if err != nil {
|
||||
lastErr = err
|
||||
if isRetryable(err) {
|
||||
if timedOut && c.noRetryTimeouts {
|
||||
return Response{}, fmt.Errorf("llm: timed out after %s (not retried): %w", c.timeoutFor(req.Model), err)
|
||||
}
|
||||
var re *retryableError
|
||||
if errors.As(err, &re) {
|
||||
retryAfter = re.retryAfter
|
||||
continue
|
||||
}
|
||||
return Response{}, err
|
||||
}
|
||||
|
||||
latency := time.Since(attemptStart)
|
||||
if resp.Error == nil && len(resp.Choices) == 0 {
|
||||
// A 2xx with neither choices nor an error object is a gateway
|
||||
// glitch, not an answer; scoring it would record a spurious
|
||||
// wrong response.
|
||||
lastErr = errors.New("llm: response contained no choices")
|
||||
continue
|
||||
}
|
||||
var text, finishReason string
|
||||
var toolCalls []ToolCall
|
||||
if len(resp.Choices) > 0 {
|
||||
@@ -213,14 +324,17 @@ func (c *OpenAIClient) Complete(ctx context.Context, req Request) (Response, err
|
||||
PromptTokens: resp.Usage.PromptTokens,
|
||||
CompletionTokens: resp.Usage.CompletionTokens,
|
||||
Latency: latency,
|
||||
Fingerprint: resp.SystemFingerprint,
|
||||
ServedModel: resp.Model,
|
||||
}, nil
|
||||
}
|
||||
return Response{}, fmt.Errorf("llm: request failed after %d attempts: %w", maxRetries+1, lastErr)
|
||||
return Response{}, fmt.Errorf("llm: request failed after %d attempts: %w", c.retries+1, lastErr)
|
||||
}
|
||||
|
||||
type retryableError struct {
|
||||
err error
|
||||
status int
|
||||
err error
|
||||
status int
|
||||
retryAfter time.Duration
|
||||
}
|
||||
|
||||
func (e *retryableError) Error() string {
|
||||
@@ -232,18 +346,42 @@ func (e *retryableError) Error() string {
|
||||
|
||||
func (e *retryableError) Unwrap() error { return e.err }
|
||||
|
||||
func isRetryable(err error) bool {
|
||||
var re *retryableError
|
||||
return errors.As(err, &re)
|
||||
// parseRetryAfter reads a delta-seconds Retry-After header; HTTP-date form
|
||||
// and malformed values yield 0 (fall back to computed backoff).
|
||||
func parseRetryAfter(h string) time.Duration {
|
||||
secs, err := strconv.Atoi(strings.TrimSpace(h))
|
||||
if err != nil || secs <= 0 {
|
||||
return 0
|
||||
}
|
||||
return time.Duration(secs) * time.Second
|
||||
}
|
||||
|
||||
func (c *OpenAIClient) doRequest(ctx context.Context, payload []byte) (*chatCompletionResponse, error) {
|
||||
var watchdog *idleWatchdog
|
||||
if c.stream {
|
||||
var cancel context.CancelCauseFunc
|
||||
ctx, cancel = context.WithCancelCause(ctx)
|
||||
defer cancel(nil)
|
||||
watchdog = startIdleWatchdog(cancel, c.idleTimeout)
|
||||
defer watchdog.Stop()
|
||||
}
|
||||
out, err := c.send(ctx, payload, watchdog)
|
||||
if err != nil && c.stream && errors.Is(context.Cause(ctx), errIdleTimeout) {
|
||||
return nil, &retryableError{err: fmt.Errorf("no data for %s: %w", c.idleTimeout, errIdleTimeout)}
|
||||
}
|
||||
return out, err
|
||||
}
|
||||
|
||||
func (c *OpenAIClient) send(ctx context.Context, payload []byte, watchdog *idleWatchdog) (*chatCompletionResponse, error) {
|
||||
url := c.endpoint + "/chat/completions"
|
||||
httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(payload))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("llm: build request: %w", err)
|
||||
}
|
||||
httpReq.Header.Set("Content-Type", "application/json")
|
||||
if c.stream {
|
||||
httpReq.Header.Set("Accept", "text/event-stream")
|
||||
}
|
||||
if c.apiKey != "" {
|
||||
httpReq.Header.Set("Authorization", "Bearer "+c.apiKey)
|
||||
}
|
||||
@@ -253,16 +391,29 @@ func (c *OpenAIClient) doRequest(ctx context.Context, payload []byte) (*chatComp
|
||||
return nil, &retryableError{err: err}
|
||||
}
|
||||
defer func() { _ = httpResp.Body.Close() }() // best-effort; body already fully read below
|
||||
|
||||
respBody, err := io.ReadAll(httpResp.Body)
|
||||
if err != nil {
|
||||
return nil, &retryableError{err: fmt.Errorf("read body: %w", err)}
|
||||
if watchdog != nil {
|
||||
watchdog.Touch()
|
||||
}
|
||||
|
||||
// 429 is as transient as a 5xx for a self-hosted gateway (rate limit or
|
||||
// a briefly saturated upstream); retry it with the same backoff budget.
|
||||
if httpResp.StatusCode >= 500 || httpResp.StatusCode == http.StatusTooManyRequests {
|
||||
return nil, &retryableError{status: httpResp.StatusCode}
|
||||
// 429 and 408 are as transient as a 5xx for a self-hosted gateway (rate
|
||||
// limit, briefly saturated upstream, slow proxy); retry them with the
|
||||
// same backoff budget.
|
||||
if httpResp.StatusCode >= 500 || httpResp.StatusCode == http.StatusTooManyRequests || httpResp.StatusCode == http.StatusRequestTimeout {
|
||||
return nil, &retryableError{
|
||||
status: httpResp.StatusCode,
|
||||
retryAfter: parseRetryAfter(httpResp.Header.Get("Retry-After")),
|
||||
}
|
||||
}
|
||||
|
||||
// A gateway may answer a streaming request with a plain JSON body (an
|
||||
// error, or no streaming support); only parse SSE when it sent SSE.
|
||||
if c.stream && httpResp.StatusCode < 400 && strings.HasPrefix(httpResp.Header.Get("Content-Type"), "text/event-stream") {
|
||||
return readStream(httpResp.Body, watchdog.Touch)
|
||||
}
|
||||
|
||||
respBody, err := io.ReadAll(io.LimitReader(httpResp.Body, maxResponseBytes))
|
||||
if err != nil {
|
||||
return nil, &retryableError{err: fmt.Errorf("read body: %w", err)}
|
||||
}
|
||||
if httpResp.StatusCode >= 400 {
|
||||
return nil, fmt.Errorf("llm: server returned status %d: %s", httpResp.StatusCode, string(respBody))
|
||||
|
||||
+213
-2
@@ -21,6 +21,12 @@ func newTestServer(t *testing.T, handler http.HandlerFunc) *httptest.Server {
|
||||
return srv
|
||||
}
|
||||
|
||||
func writeOK(w http.ResponseWriter, content string) {
|
||||
resp := chatCompletionResponse{}
|
||||
resp.Choices = append(resp.Choices, testChoice(content, "stop"))
|
||||
_ = json.NewEncoder(w).Encode(resp)
|
||||
}
|
||||
|
||||
// testChoice builds one response choice with a non-null content string, the
|
||||
// common case. Its return type is exactly chatCompletionResponse.Choices's
|
||||
// element type, so callers can append() it without repeating the anonymous
|
||||
@@ -137,7 +143,7 @@ func TestOpenAIClient_Complete_SendsTemperatureZero(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatalf("read request body: %v", err)
|
||||
}
|
||||
_ = json.NewEncoder(w).Encode(chatCompletionResponse{})
|
||||
writeOK(w, "ok")
|
||||
})
|
||||
|
||||
client := NewOpenAIClient(srv.URL, "", time.Second)
|
||||
@@ -320,7 +326,7 @@ func TestOpenAIClient_Complete_SendsAPIKey(t *testing.T) {
|
||||
if got := r.Header.Get("Authorization"); got != "Bearer secret-key" {
|
||||
t.Errorf("Authorization = %q, want Bearer secret-key", got)
|
||||
}
|
||||
_ = json.NewEncoder(w).Encode(chatCompletionResponse{})
|
||||
writeOK(w, "ok")
|
||||
})
|
||||
|
||||
client := NewOpenAIClient(srv.URL, "secret-key", time.Second)
|
||||
@@ -394,3 +400,208 @@ func TestOpenAIClient_Complete_MalformedToolArgs(t *testing.T) {
|
||||
t.Errorf("raw args = %q", resp.ToolCalls[0].RawArguments)
|
||||
}
|
||||
}
|
||||
|
||||
func fastClient(url string) *OpenAIClient {
|
||||
c := NewOpenAIClient(url, "", time.Second)
|
||||
c.baseDelay = time.Millisecond
|
||||
return c
|
||||
}
|
||||
|
||||
var oneMsg = Request{Model: "m", Messages: []Message{{Role: "user", Content: "x"}}}
|
||||
|
||||
func TestOpenAIClient_Complete_SendsSeedAndEchoesFingerprint(t *testing.T) {
|
||||
var decoded map[string]any
|
||||
srv := newTestServer(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
b, _ := io.ReadAll(r.Body)
|
||||
_ = json.Unmarshal(b, &decoded)
|
||||
_, _ = w.Write([]byte(`{"model":"served-1","system_fingerprint":"fp-9","choices":[{"finish_reason":"stop","message":{"content":"hi"}}]}`))
|
||||
})
|
||||
seed := 42
|
||||
req := oneMsg
|
||||
req.Seed = &seed
|
||||
resp, err := fastClient(srv.URL).Complete(context.Background(), req)
|
||||
if err != nil {
|
||||
t.Fatalf("Complete() error = %v", err)
|
||||
}
|
||||
if decoded["seed"] != float64(42) {
|
||||
t.Errorf("seed = %v, want 42", decoded["seed"])
|
||||
}
|
||||
if resp.Fingerprint != "fp-9" || resp.ServedModel != "served-1" {
|
||||
t.Errorf("fingerprint/model = %q/%q, want fp-9/served-1", resp.Fingerprint, resp.ServedModel)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOpenAIClient_Complete_OmitsSeedWhenUnset(t *testing.T) {
|
||||
var decoded map[string]any
|
||||
srv := newTestServer(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
b, _ := io.ReadAll(r.Body)
|
||||
_ = json.Unmarshal(b, &decoded)
|
||||
writeOK(w, "ok")
|
||||
})
|
||||
if _, err := fastClient(srv.URL).Complete(context.Background(), oneMsg); err != nil {
|
||||
t.Fatalf("Complete() error = %v", err)
|
||||
}
|
||||
if _, present := decoded["seed"]; present {
|
||||
t.Errorf("seed present in body without Request.Seed: %v", decoded)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOpenAIClient_Complete_RetriesEmptyChoicesThenSucceeds(t *testing.T) {
|
||||
var calls int32
|
||||
srv := newTestServer(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
if atomic.AddInt32(&calls, 1) == 1 {
|
||||
_, _ = w.Write([]byte(`{"choices":[]}`))
|
||||
return
|
||||
}
|
||||
writeOK(w, "ok")
|
||||
})
|
||||
resp, err := fastClient(srv.URL).Complete(context.Background(), oneMsg)
|
||||
if err != nil || resp.Text != "ok" {
|
||||
t.Fatalf("Complete() = %q, %v; want ok, nil", resp.Text, err)
|
||||
}
|
||||
if got := atomic.LoadInt32(&calls); got != 2 {
|
||||
t.Errorf("calls = %d, want 2", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOpenAIClient_Complete_RetriesOn408(t *testing.T) {
|
||||
var calls int32
|
||||
srv := newTestServer(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
if atomic.AddInt32(&calls, 1) == 1 {
|
||||
w.WriteHeader(http.StatusRequestTimeout)
|
||||
return
|
||||
}
|
||||
writeOK(w, "ok")
|
||||
})
|
||||
if _, err := fastClient(srv.URL).Complete(context.Background(), oneMsg); err != nil {
|
||||
t.Fatalf("Complete() error = %v", err)
|
||||
}
|
||||
if got := atomic.LoadInt32(&calls); got != 2 {
|
||||
t.Errorf("calls = %d, want 2", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOpenAIClient_WithRetries(t *testing.T) {
|
||||
var calls int32
|
||||
srv := newTestServer(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
atomic.AddInt32(&calls, 1)
|
||||
w.WriteHeader(http.StatusBadGateway)
|
||||
})
|
||||
if _, err := fastClient(srv.URL).WithRetries(5).Complete(context.Background(), oneMsg); err == nil {
|
||||
t.Fatal("Complete() error = nil, want error")
|
||||
}
|
||||
if got := atomic.LoadInt32(&calls); got != 6 {
|
||||
t.Errorf("calls = %d, want 6 (1 + 5 retries)", got)
|
||||
}
|
||||
|
||||
atomic.StoreInt32(&calls, 0)
|
||||
if _, err := fastClient(srv.URL).WithRetries(0).Complete(context.Background(), oneMsg); err == nil {
|
||||
t.Fatal("Complete() error = nil, want error")
|
||||
}
|
||||
if got := atomic.LoadInt32(&calls); got != 1 {
|
||||
t.Errorf("calls = %d, want 1 (retries disabled)", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOpenAIClient_Complete_HonorsRetryAfter(t *testing.T) {
|
||||
var calls int32
|
||||
srv := newTestServer(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
if atomic.AddInt32(&calls, 1) == 1 {
|
||||
w.Header().Set("Retry-After", "1")
|
||||
w.WriteHeader(http.StatusTooManyRequests)
|
||||
return
|
||||
}
|
||||
writeOK(w, "ok")
|
||||
})
|
||||
start := time.Now()
|
||||
if _, err := fastClient(srv.URL).Complete(context.Background(), oneMsg); err != nil {
|
||||
t.Fatalf("Complete() error = %v", err)
|
||||
}
|
||||
if elapsed := time.Since(start); elapsed < 900*time.Millisecond {
|
||||
t.Errorf("elapsed = %v, want >= ~1s from Retry-After", elapsed)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOpenAIClient_backoff(t *testing.T) {
|
||||
c := NewOpenAIClient("http://x", "", time.Second)
|
||||
for attempt, base := range map[int]time.Duration{1: 250 * time.Millisecond, 2: 500 * time.Millisecond, 3: time.Second} {
|
||||
got := c.backoff(attempt, 0)
|
||||
if got < base || got > base+base/3 {
|
||||
t.Errorf("backoff(%d) = %v, want in [%v, %v]", attempt, got, base, base+base/3)
|
||||
}
|
||||
}
|
||||
if got := c.backoff(30, 0); got > maxRetryDelay*5/4 {
|
||||
t.Errorf("backoff(30) = %v, want capped near %v", got, maxRetryDelay)
|
||||
}
|
||||
if got := c.backoff(1, time.Hour); got != maxRetryDelay {
|
||||
t.Errorf("backoff with huge Retry-After = %v, want cap %v", got, maxRetryDelay)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewOpenAIClient_TrimsTrailingSlash(t *testing.T) {
|
||||
var path string
|
||||
srv := newTestServer(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
path = r.URL.Path
|
||||
writeOK(w, "ok")
|
||||
})
|
||||
if _, err := fastClient(srv.URL+"/v1/").Complete(context.Background(), oneMsg); err != nil {
|
||||
t.Fatalf("Complete() error = %v", err)
|
||||
}
|
||||
if path != "/v1/chat/completions" {
|
||||
t.Errorf("path = %q, want /v1/chat/completions", path)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOpenAIClient_Complete_TimeoutNotRetriedWhenDisabled(t *testing.T) {
|
||||
var calls int32
|
||||
srv := newTestServer(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
atomic.AddInt32(&calls, 1)
|
||||
time.Sleep(300 * time.Millisecond)
|
||||
writeOK(w, "late")
|
||||
})
|
||||
c := NewOpenAIClient(srv.URL, "", 50*time.Millisecond).WithRetryTimeouts(false)
|
||||
c.baseDelay = time.Millisecond
|
||||
_, err := c.Complete(context.Background(), oneMsg)
|
||||
if err == nil || !strings.Contains(err.Error(), "not retried") {
|
||||
t.Fatalf("err = %v, want a timeout not retried", err)
|
||||
}
|
||||
if got := atomic.LoadInt32(&calls); got != 1 {
|
||||
t.Errorf("calls = %d, want 1", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOpenAIClient_Complete_TimeoutRetriedByDefault(t *testing.T) {
|
||||
var calls int32
|
||||
srv := newTestServer(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
if atomic.AddInt32(&calls, 1) == 1 {
|
||||
time.Sleep(300 * time.Millisecond)
|
||||
}
|
||||
writeOK(w, "ok")
|
||||
})
|
||||
c := NewOpenAIClient(srv.URL, "", 100*time.Millisecond)
|
||||
c.baseDelay = time.Millisecond
|
||||
if _, err := c.Complete(context.Background(), oneMsg); err != nil {
|
||||
t.Fatalf("Complete() error = %v", err)
|
||||
}
|
||||
if got := atomic.LoadInt32(&calls); got != 2 {
|
||||
t.Errorf("calls = %d, want 2", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOpenAIClient_Complete_PerModelTimeout(t *testing.T) {
|
||||
srv := newTestServer(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
writeOK(w, "slow but fine")
|
||||
})
|
||||
c := NewOpenAIClient(srv.URL, "", 50*time.Millisecond).
|
||||
WithRetryTimeouts(false).
|
||||
WithModelTimeouts(map[string]time.Duration{"slow-local": 2 * time.Second})
|
||||
req := oneMsg
|
||||
req.Model = "slow-local"
|
||||
if resp, err := c.Complete(context.Background(), req); err != nil || resp.Text != "slow but fine" {
|
||||
t.Fatalf("slow-local = %q, %v; want success under its own timeout", resp.Text, err)
|
||||
}
|
||||
if _, err := c.Complete(context.Background(), oneMsg); err == nil {
|
||||
t.Error("default-timeout model succeeded, want timeout")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,174 @@
|
||||
package llm
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// defaultIdleTimeout is how long a stream may stay silent (no bytes at all,
|
||||
// including before the first token) before the attempt is abandoned. It
|
||||
// must exceed a slow local model's prompt-processing time.
|
||||
const defaultIdleTimeout = 5 * time.Minute
|
||||
|
||||
// maxStreamLine bounds one SSE line.
|
||||
const maxStreamLine = 1 << 20
|
||||
|
||||
const (
|
||||
sseDataPrefix = "data:"
|
||||
sseDone = "[DONE]"
|
||||
)
|
||||
|
||||
// errIdleTimeout marks an attempt abandoned because the stream went silent.
|
||||
var errIdleTimeout = errors.New("llm: stream idle timeout")
|
||||
|
||||
// errStreamTruncated marks a stream that ended without [DONE] or a
|
||||
// finish_reason, so the answer may be incomplete.
|
||||
var errStreamTruncated = errors.New("llm: stream ended before completion")
|
||||
|
||||
type streamToolCall struct {
|
||||
Function struct {
|
||||
Name string `json:"name"`
|
||||
Arguments string `json:"arguments"`
|
||||
} `json:"function"`
|
||||
Index int `json:"index"`
|
||||
}
|
||||
|
||||
type streamChunk struct {
|
||||
Error *struct {
|
||||
Message string `json:"message"`
|
||||
} `json:"error"`
|
||||
Usage *struct {
|
||||
PromptTokens int `json:"prompt_tokens"`
|
||||
CompletionTokens int `json:"completion_tokens"`
|
||||
} `json:"usage"`
|
||||
Model string `json:"model"`
|
||||
SystemFingerprint string `json:"system_fingerprint"`
|
||||
Choices []struct {
|
||||
FinishReason *string `json:"finish_reason"`
|
||||
Delta struct {
|
||||
Content *string `json:"content"`
|
||||
ToolCalls []streamToolCall `json:"tool_calls"`
|
||||
} `json:"delta"`
|
||||
} `json:"choices"`
|
||||
}
|
||||
|
||||
// readStream folds an OpenAI-style SSE stream into the same response shape
|
||||
// the non-streaming path decodes. onActivity is called for every line read,
|
||||
// including keep-alive comments, so the caller can reset its idle timer.
|
||||
// reasoning_content deltas count as activity but are never accumulated,
|
||||
// matching the non-streaming path.
|
||||
func readStream(r io.Reader, onActivity func()) (*chatCompletionResponse, error) {
|
||||
sc := bufio.NewScanner(r)
|
||||
sc.Buffer(make([]byte, 0, 64<<10), maxStreamLine)
|
||||
|
||||
var (
|
||||
out chatCompletionResponse
|
||||
text strings.Builder
|
||||
sawText bool
|
||||
finish string
|
||||
done bool
|
||||
order []int
|
||||
calls = map[int]*apiToolCall{}
|
||||
tooLarge bool
|
||||
)
|
||||
|
||||
for sc.Scan() {
|
||||
onActivity()
|
||||
line := strings.TrimSpace(sc.Text())
|
||||
if !strings.HasPrefix(line, sseDataPrefix) {
|
||||
continue // blank separator, comment or other SSE field
|
||||
}
|
||||
data := strings.TrimSpace(strings.TrimPrefix(line, sseDataPrefix))
|
||||
if data == sseDone {
|
||||
done = true
|
||||
break
|
||||
}
|
||||
var ch streamChunk
|
||||
if err := json.Unmarshal([]byte(data), &ch); err != nil {
|
||||
return nil, &retryableError{err: fmt.Errorf("decode stream chunk: %w", err)}
|
||||
}
|
||||
if ch.Error != nil {
|
||||
out.Error = ch.Error
|
||||
return &out, nil
|
||||
}
|
||||
if ch.Model != "" {
|
||||
out.Model = ch.Model
|
||||
}
|
||||
if ch.SystemFingerprint != "" {
|
||||
out.SystemFingerprint = ch.SystemFingerprint
|
||||
}
|
||||
if ch.Usage != nil {
|
||||
out.Usage.PromptTokens = ch.Usage.PromptTokens
|
||||
out.Usage.CompletionTokens = ch.Usage.CompletionTokens
|
||||
}
|
||||
if len(ch.Choices) == 0 {
|
||||
continue
|
||||
}
|
||||
c := ch.Choices[0]
|
||||
if c.Delta.Content != nil {
|
||||
sawText = true
|
||||
if text.Len()+len(*c.Delta.Content) > maxResponseBytes {
|
||||
tooLarge = true
|
||||
} else {
|
||||
text.WriteString(*c.Delta.Content)
|
||||
}
|
||||
}
|
||||
for _, tc := range c.Delta.ToolCalls {
|
||||
acc, ok := calls[tc.Index]
|
||||
if !ok {
|
||||
acc = &apiToolCall{Type: "function"}
|
||||
calls[tc.Index] = acc
|
||||
order = append(order, tc.Index)
|
||||
}
|
||||
if tc.Function.Name != "" {
|
||||
acc.Function.Name = tc.Function.Name
|
||||
}
|
||||
acc.Function.Arguments += tc.Function.Arguments
|
||||
}
|
||||
if c.FinishReason != nil && *c.FinishReason != "" {
|
||||
finish = *c.FinishReason
|
||||
}
|
||||
}
|
||||
if err := sc.Err(); err != nil {
|
||||
return nil, &retryableError{err: fmt.Errorf("read stream: %w", err)}
|
||||
}
|
||||
if tooLarge {
|
||||
return nil, fmt.Errorf("llm: streamed response exceeded %d bytes", maxResponseBytes)
|
||||
}
|
||||
if !done && finish == "" {
|
||||
return nil, &retryableError{err: errStreamTruncated}
|
||||
}
|
||||
|
||||
msg := responseMsg{Role: "assistant"}
|
||||
if sawText {
|
||||
s := text.String()
|
||||
msg.Content = &s
|
||||
}
|
||||
for _, idx := range order {
|
||||
msg.ToolCalls = append(msg.ToolCalls, *calls[idx])
|
||||
}
|
||||
out.Choices = append(out.Choices, struct {
|
||||
FinishReason string `json:"finish_reason"`
|
||||
Message responseMsg `json:"message"`
|
||||
}{FinishReason: finish, Message: msg})
|
||||
return &out, nil
|
||||
}
|
||||
|
||||
// idleWatchdog cancels an attempt when no activity is seen for idle.
|
||||
type idleWatchdog struct {
|
||||
timer *time.Timer
|
||||
idle time.Duration
|
||||
}
|
||||
|
||||
func startIdleWatchdog(cancel context.CancelCauseFunc, idle time.Duration) *idleWatchdog {
|
||||
return &idleWatchdog{idle: idle, timer: time.AfterFunc(idle, func() { cancel(errIdleTimeout) })}
|
||||
}
|
||||
|
||||
func (w *idleWatchdog) Touch() { w.timer.Reset(w.idle) }
|
||||
func (w *idleWatchdog) Stop() { w.timer.Stop() }
|
||||
@@ -0,0 +1,170 @@
|
||||
package llm
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func sseServer(t *testing.T, frames func(w http.ResponseWriter, fl http.Flusher)) string {
|
||||
t.Helper()
|
||||
return newTestServer(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "text/event-stream")
|
||||
fl := w.(http.Flusher)
|
||||
frames(w, fl)
|
||||
}).URL
|
||||
}
|
||||
|
||||
func streamClient(url string, idle time.Duration) *OpenAIClient {
|
||||
c := fastClient(url).WithStreaming(true, idle)
|
||||
return c
|
||||
}
|
||||
|
||||
func TestStream_AggregatesContentUsageAndFingerprint(t *testing.T) {
|
||||
var body map[string]any
|
||||
url := newTestServer(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
b, _ := io.ReadAll(r.Body)
|
||||
_ = json.Unmarshal(b, &body)
|
||||
w.Header().Set("Content-Type", "text/event-stream")
|
||||
send(w, ": keepalive\n\n")
|
||||
send(w, `data: {"model":"srv","system_fingerprint":"fp","choices":[{"delta":{"reasoning_content":"thinking"}}]}`+"\n\n")
|
||||
send(w, `data: {"choices":[{"delta":{"content":"Hel"}}]}`+"\n\n")
|
||||
send(w, `data: {"choices":[{"delta":{"content":"lo"},"finish_reason":"stop"}]}`+"\n\n")
|
||||
send(w, `data: {"choices":[],"usage":{"prompt_tokens":5,"completion_tokens":7}}`+"\n\n")
|
||||
send(w, "data: [DONE]\n\n")
|
||||
}).URL
|
||||
resp, err := streamClient(url, time.Second).Complete(context.Background(), oneMsg)
|
||||
if err != nil {
|
||||
t.Fatalf("Complete() error = %v", err)
|
||||
}
|
||||
if resp.Text != "Hello" || resp.FinishReason != "stop" || resp.PromptTokens != 5 || resp.CompletionTokens != 7 {
|
||||
t.Errorf("resp = %+v", resp)
|
||||
}
|
||||
if resp.Fingerprint != "fp" || resp.ServedModel != "srv" {
|
||||
t.Errorf("fingerprint/model = %q/%q", resp.Fingerprint, resp.ServedModel)
|
||||
}
|
||||
if body["stream"] != true {
|
||||
t.Errorf("stream = %v, want true", body["stream"])
|
||||
}
|
||||
if opts, _ := body["stream_options"].(map[string]any); opts["include_usage"] != true {
|
||||
t.Errorf("stream_options = %v, want include_usage", body["stream_options"])
|
||||
}
|
||||
}
|
||||
|
||||
func TestStream_AssemblesToolCallFragments(t *testing.T) {
|
||||
url := sseServer(t, func(w http.ResponseWriter, fl http.Flusher) {
|
||||
send(w, `data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"name":"get_w","arguments":"{\"ci"}}]}}]}`+"\n\n")
|
||||
send(w, `data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"ty\":\"X\"}"}},{"index":1,"function":{"name":"b","arguments":"{}"}}]},"finish_reason":"tool_calls"}]}`+"\n\n")
|
||||
send(w, "data: [DONE]\n\n")
|
||||
})
|
||||
resp, err := streamClient(url, time.Second).Complete(context.Background(), oneMsg)
|
||||
if err != nil {
|
||||
t.Fatalf("Complete() error = %v", err)
|
||||
}
|
||||
if len(resp.ToolCalls) != 2 || resp.ToolCalls[0].Name != "get_w" || resp.ToolCalls[0].Arguments["city"] != "X" || !resp.ToolCalls[1].Decoded {
|
||||
t.Errorf("ToolCalls = %+v", resp.ToolCalls)
|
||||
}
|
||||
if resp.Text != "" {
|
||||
t.Errorf("Text = %q, want empty", resp.Text)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStream_ReasoningOnlyTruncationYieldsEmptyText(t *testing.T) {
|
||||
url := sseServer(t, func(w http.ResponseWriter, fl http.Flusher) {
|
||||
send(w, `data: {"choices":[{"delta":{"reasoning_content":"..."},"finish_reason":"length"}]}`+"\n\n")
|
||||
send(w, "data: [DONE]\n\n")
|
||||
})
|
||||
resp, err := streamClient(url, time.Second).Complete(context.Background(), oneMsg)
|
||||
if err != nil || resp.Text != "" || resp.FinishReason != FinishReasonLength {
|
||||
t.Fatalf("resp = %+v, err = %v; want empty text, length", resp, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStream_TruncatedStreamIsRetried(t *testing.T) {
|
||||
var calls int32
|
||||
url := sseServer(t, func(w http.ResponseWriter, fl http.Flusher) {
|
||||
if atomic.AddInt32(&calls, 1) == 1 {
|
||||
send(w, `data: {"choices":[{"delta":{"content":"par"}}]}`+"\n\n")
|
||||
return // connection ends with no [DONE] or finish_reason
|
||||
}
|
||||
send(w, `data: {"choices":[{"delta":{"content":"whole"},"finish_reason":"stop"}]}`+"\n\n")
|
||||
send(w, "data: [DONE]\n\n")
|
||||
})
|
||||
resp, err := streamClient(url, time.Second).Complete(context.Background(), oneMsg)
|
||||
if err != nil || resp.Text != "whole" {
|
||||
t.Fatalf("resp = %q, err = %v; want whole", resp.Text, err)
|
||||
}
|
||||
if got := atomic.LoadInt32(&calls); got != 2 {
|
||||
t.Errorf("calls = %d, want 2", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStream_IdleTimeoutAbandonsHungServer(t *testing.T) {
|
||||
var calls int32
|
||||
url := sseServer(t, func(w http.ResponseWriter, fl http.Flusher) {
|
||||
atomic.AddInt32(&calls, 1)
|
||||
send(w, `data: {"choices":[{"delta":{"content":"x"}}]}`+"\n\n")
|
||||
fl.Flush()
|
||||
time.Sleep(2 * time.Second) // goes silent
|
||||
})
|
||||
c := streamClient(url, 150*time.Millisecond).WithRetryTimeouts(false)
|
||||
start := time.Now()
|
||||
_, err := c.Complete(context.Background(), oneMsg)
|
||||
if err == nil || !strings.Contains(err.Error(), "not retried") {
|
||||
t.Fatalf("err = %v, want idle timeout not retried", err)
|
||||
}
|
||||
if elapsed := time.Since(start); elapsed > time.Second {
|
||||
t.Errorf("took %v, want abandoned near the 150ms idle timeout", elapsed)
|
||||
}
|
||||
if got := atomic.LoadInt32(&calls); got != 1 {
|
||||
t.Errorf("calls = %d, want 1", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStream_SlowButProgressingSurvivesPastIdleTimeout(t *testing.T) {
|
||||
url := sseServer(t, func(w http.ResponseWriter, fl http.Flusher) {
|
||||
for i := 0; i < 6; i++ {
|
||||
send(w, `data: {"choices":[{"delta":{"content":"a"}}]}`+"\n\n")
|
||||
fl.Flush()
|
||||
time.Sleep(80 * time.Millisecond)
|
||||
}
|
||||
send(w, `data: {"choices":[{"delta":{},"finish_reason":"stop"}]}`+"\n\ndata: [DONE]\n\n")
|
||||
})
|
||||
// total ~480ms > idle 200ms, but never silent for 200ms.
|
||||
resp, err := streamClient(url, 200*time.Millisecond).Complete(context.Background(), oneMsg)
|
||||
if err != nil || resp.Text != "aaaaaa" {
|
||||
t.Fatalf("resp = %q, err = %v; want aaaaaa", resp.Text, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStream_ErrorChunkIsRetried(t *testing.T) {
|
||||
var calls int32
|
||||
url := sseServer(t, func(w http.ResponseWriter, fl http.Flusher) {
|
||||
if atomic.AddInt32(&calls, 1) == 1 {
|
||||
send(w, `data: {"error":{"message":"replica warming"}}`+"\n\n")
|
||||
return
|
||||
}
|
||||
send(w, `data: {"choices":[{"delta":{"content":"ok"},"finish_reason":"stop"}]}`+"\n\ndata: [DONE]\n\n")
|
||||
})
|
||||
resp, err := streamClient(url, time.Second).Complete(context.Background(), oneMsg)
|
||||
if err != nil || resp.Text != "ok" {
|
||||
t.Fatalf("resp = %q, err = %v", resp.Text, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStream_PlainJSONFallbackStillParsed(t *testing.T) {
|
||||
srv := newTestServer(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
writeOK(w, "plain")
|
||||
})
|
||||
resp, err := streamClient(srv.URL, time.Second).Complete(context.Background(), oneMsg)
|
||||
if err != nil || resp.Text != "plain" {
|
||||
t.Fatalf("resp = %q, err = %v; want plain JSON body accepted", resp.Text, err)
|
||||
}
|
||||
}
|
||||
|
||||
func send(w io.Writer, s string) { _, _ = io.WriteString(w, s) }
|
||||
@@ -170,8 +170,11 @@ func CompareArtifacts(baseline, current Artifact) []Comparison {
|
||||
|
||||
// RenderCompare writes a comparison report: headline counts, then only
|
||||
// changed/missing rows (stable rows are summarized, not listed).
|
||||
func RenderCompare(w io.Writer, comparisons []Comparison) error {
|
||||
func RenderCompare(w io.Writer, comparisons []Comparison, warnings ...string) error {
|
||||
ew := &errWriter{w: w}
|
||||
for _, warn := range warnings {
|
||||
ew.println("WARNING: " + warn)
|
||||
}
|
||||
|
||||
var reg, imp, stable, added, removed, errored int
|
||||
for _, c := range comparisons {
|
||||
@@ -219,3 +222,9 @@ func deltaText(v float64) string {
|
||||
}
|
||||
return fmt.Sprintf("%+.2f", v)
|
||||
}
|
||||
|
||||
// CompareWarnings reports reasons the two artifacts may not be comparable
|
||||
// like-for-like (catalog, sampling or backend-build drift).
|
||||
func CompareWarnings(baseline, current Artifact) []string {
|
||||
return compareWarnings(baseline.Meta, current.Meta)
|
||||
}
|
||||
|
||||
+34
-5
@@ -6,6 +6,7 @@ import (
|
||||
"io"
|
||||
"math"
|
||||
"os"
|
||||
"path/filepath"
|
||||
|
||||
"github.com/lukaszraczylo/llm-testbench/internal/runner"
|
||||
"github.com/lukaszraczylo/llm-testbench/internal/testkit"
|
||||
@@ -22,6 +23,7 @@ type jsonResult struct {
|
||||
Error string `json:"error,omitempty"`
|
||||
ResponseText string `json:"response_text"`
|
||||
FinishReason string `json:"finish_reason,omitempty"`
|
||||
Fingerprint string `json:"fingerprint,omitempty"`
|
||||
Score float64 `json:"score"`
|
||||
LatencyMS int64 `json:"latency_ms"`
|
||||
PromptTokens int `json:"prompt_tokens"`
|
||||
@@ -36,6 +38,7 @@ type jsonResult struct {
|
||||
// plus the per-test discrimination rollup, so downstream tooling
|
||||
// (compare) can re-derive stability without re-running the suite.
|
||||
type jsonArtifact struct {
|
||||
Meta *RunMeta `json:"meta,omitempty"`
|
||||
Results []jsonResult `json:"results"`
|
||||
Stats []jsonStatRow `json:"stats,omitempty"`
|
||||
}
|
||||
@@ -77,6 +80,7 @@ func buildArtifact(tests []testkit.Test, models []string, results []runner.Resul
|
||||
Detail: r.Score.Detail,
|
||||
ResponseText: r.ResponseText,
|
||||
FinishReason: r.FinishReason,
|
||||
Fingerprint: r.Fingerprint,
|
||||
Skipped: r.Score.Skipped,
|
||||
Truncated: r.Truncated(),
|
||||
LatencyMS: r.Latency.Milliseconds(),
|
||||
@@ -150,12 +154,37 @@ func LoadArtifact(path string) (jsonArtifact, error) {
|
||||
return jsonArtifact{Results: legacy}, nil
|
||||
}
|
||||
|
||||
// WriteArtifact saves the run as a JSON artifact at path, for later
|
||||
// `llmtest compare` runs, regardless of the stdout report format.
|
||||
func WriteArtifact(path string, tests []testkit.Test, models []string, results []runner.Result) error {
|
||||
b, err := json.MarshalIndent(buildArtifact(tests, models, results), "", " ")
|
||||
// WriteArtifact saves the run, stamped with meta, as a JSON artifact at
|
||||
// path for later `llmtest compare` runs, regardless of the stdout report
|
||||
// format. The write is atomic (temp file + rename), so an interrupted
|
||||
// process never leaves a truncated artifact that replaced a good one.
|
||||
func WriteArtifact(path string, meta RunMeta, tests []testkit.Test, models []string, results []runner.Result) error {
|
||||
art := buildArtifact(tests, models, results)
|
||||
meta.CatalogHash = CatalogHash(tests)
|
||||
meta.TestCount = len(tests)
|
||||
meta.Fingerprints = servedFingerprints(results)
|
||||
art.Meta = &meta
|
||||
|
||||
b, err := json.MarshalIndent(art, "", " ")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return os.WriteFile(path, append(b, '\n'), 0o644) //nolint:gosec // 0o644: the artifact is a report meant to be shared/committed, not a secret file
|
||||
|
||||
tmp, err := os.CreateTemp(filepath.Dir(path), ".llmtest-artifact-*")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() { _ = os.Remove(tmp.Name()) }() // no-op after a successful rename
|
||||
if _, err := tmp.Write(append(b, '\n')); err != nil {
|
||||
_ = tmp.Close()
|
||||
return err
|
||||
}
|
||||
if err := tmp.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
// 0o644: the artifact is a report meant to be shared/committed, not a secret file.
|
||||
if err := os.Chmod(tmp.Name(), 0o644); err != nil { //nolint:gosec // see comment above
|
||||
return err
|
||||
}
|
||||
return os.Rename(tmp.Name(), path)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,109 @@
|
||||
package report
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/lukaszraczylo/llm-testbench/internal/runner"
|
||||
"github.com/lukaszraczylo/llm-testbench/internal/testkit"
|
||||
)
|
||||
|
||||
// RunMeta records the settings and environment a run used, so two
|
||||
// artifacts can be judged comparable and a result reproduced.
|
||||
type RunMeta struct {
|
||||
Fingerprints map[string][]string `json:"fingerprints,omitempty"`
|
||||
Seed *int `json:"seed,omitempty"`
|
||||
Version string `json:"version"`
|
||||
StartedAt string `json:"started_at"`
|
||||
Endpoint string `json:"endpoint,omitempty"`
|
||||
CatalogHash string `json:"catalog_hash"`
|
||||
Temperature float64 `json:"temperature"`
|
||||
MaxTokensDefault int `json:"max_tokens_default"`
|
||||
Repeat int `json:"repeat"`
|
||||
TestCount int `json:"test_count"`
|
||||
Interrupted bool `json:"interrupted,omitempty"`
|
||||
}
|
||||
|
||||
// CatalogHash returns a stable digest of the test inputs that determine a
|
||||
// model's answer (id, system prompt, prompt, token floor, tool names).
|
||||
func CatalogHash(tests []testkit.Test) string {
|
||||
sorted := append([]testkit.Test(nil), tests...)
|
||||
sort.Slice(sorted, func(i, j int) bool { return sorted[i].ID < sorted[j].ID })
|
||||
h := sha256.New()
|
||||
for _, t := range sorted {
|
||||
names := make([]string, 0, len(t.Tools))
|
||||
for _, tl := range t.Tools {
|
||||
names = append(names, tl.Name)
|
||||
}
|
||||
_, _ = fmt.Fprintf(h, "%s\x00%s\x00%s\x00%d\x00%s\x01", t.ID, t.System, t.Prompt, t.MaxTokens, strings.Join(names, ","))
|
||||
}
|
||||
return hex.EncodeToString(h.Sum(nil))[:16]
|
||||
}
|
||||
|
||||
// servedFingerprints collects the distinct backend fingerprints each model
|
||||
// answered with; more than one means the serving build changed mid-run.
|
||||
func servedFingerprints(results []runner.Result) map[string][]string {
|
||||
seen := make(map[string]map[string]bool)
|
||||
for _, r := range results {
|
||||
if r.Fingerprint == "" {
|
||||
continue
|
||||
}
|
||||
if seen[r.Model] == nil {
|
||||
seen[r.Model] = make(map[string]bool)
|
||||
}
|
||||
seen[r.Model][r.Fingerprint] = true
|
||||
}
|
||||
if len(seen) == 0 {
|
||||
return nil
|
||||
}
|
||||
out := make(map[string][]string, len(seen))
|
||||
for m, fps := range seen {
|
||||
for fp := range fps {
|
||||
out[m] = append(out[m], fp)
|
||||
}
|
||||
sort.Strings(out[m])
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// compareWarnings lists reasons two artifacts may not be like-for-like.
|
||||
// Artifacts without metadata (older runs) produce no warnings.
|
||||
func compareWarnings(base, cur *RunMeta) []string {
|
||||
if base == nil || cur == nil {
|
||||
return nil
|
||||
}
|
||||
var w []string
|
||||
if base.CatalogHash != cur.CatalogHash {
|
||||
w = append(w, fmt.Sprintf("catalog differs (%s vs %s): prompts changed between runs", base.CatalogHash, cur.CatalogHash))
|
||||
}
|
||||
if base.Temperature != cur.Temperature {
|
||||
w = append(w, fmt.Sprintf("temperature differs (%g vs %g)", base.Temperature, cur.Temperature))
|
||||
}
|
||||
if base.MaxTokensDefault != cur.MaxTokensDefault {
|
||||
w = append(w, fmt.Sprintf("max_tokens_default differs (%d vs %d)", base.MaxTokensDefault, cur.MaxTokensDefault))
|
||||
}
|
||||
if fmtSeed(base.Seed) != fmtSeed(cur.Seed) {
|
||||
w = append(w, fmt.Sprintf("seed differs (%s vs %s)", fmtSeed(base.Seed), fmtSeed(cur.Seed)))
|
||||
}
|
||||
if base.Interrupted || cur.Interrupted {
|
||||
w = append(w, "an artifact is from an interrupted run: missing results are not regressions")
|
||||
}
|
||||
for m, cfps := range cur.Fingerprints {
|
||||
if bfps, ok := base.Fingerprints[m]; ok && strings.Join(bfps, ",") != strings.Join(cfps, ",") {
|
||||
w = append(w, fmt.Sprintf("%s served by a different backend build (%s vs %s)", m, strings.Join(bfps, "+"), strings.Join(cfps, "+")))
|
||||
}
|
||||
}
|
||||
sort.Strings(w)
|
||||
return w
|
||||
}
|
||||
|
||||
func fmtSeed(s *int) string {
|
||||
if s == nil {
|
||||
return "unset"
|
||||
}
|
||||
return strconv.Itoa(*s)
|
||||
}
|
||||
@@ -0,0 +1,71 @@
|
||||
package report
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/lukaszraczylo/llm-testbench/internal/runner"
|
||||
"github.com/lukaszraczylo/llm-testbench/internal/testkit"
|
||||
)
|
||||
|
||||
func TestCatalogHash_StableAndSensitive(t *testing.T) {
|
||||
a := testkit.Test{ID: "a", Prompt: "p1"}
|
||||
b := testkit.Test{ID: "b", Prompt: "p2"}
|
||||
if CatalogHash([]testkit.Test{a, b}) != CatalogHash([]testkit.Test{b, a}) {
|
||||
t.Error("hash depends on input order, want stable")
|
||||
}
|
||||
b2 := b
|
||||
b2.Prompt = "p2 changed"
|
||||
if CatalogHash([]testkit.Test{a, b}) == CatalogHash([]testkit.Test{a, b2}) {
|
||||
t.Error("hash ignores a prompt change")
|
||||
}
|
||||
}
|
||||
|
||||
func TestCompareWarnings(t *testing.T) {
|
||||
seed := 1
|
||||
base := &RunMeta{CatalogHash: "x", Temperature: 0, MaxTokensDefault: 100, Fingerprints: map[string][]string{"m": {"fp1"}}}
|
||||
if w := compareWarnings(base, base); len(w) != 0 {
|
||||
t.Errorf("identical meta warnings = %v, want none", w)
|
||||
}
|
||||
if w := compareWarnings(nil, base); len(w) != 0 {
|
||||
t.Errorf("legacy artifact warnings = %v, want none", w)
|
||||
}
|
||||
cur := &RunMeta{CatalogHash: "y", Temperature: 0, MaxTokensDefault: 200, Seed: &seed, Interrupted: true,
|
||||
Fingerprints: map[string][]string{"m": {"fp2"}}}
|
||||
got := strings.Join(compareWarnings(base, cur), "\n")
|
||||
for _, want := range []string{"catalog differs", "max_tokens_default differs", "seed differs", "interrupted", "different backend build"} {
|
||||
if !strings.Contains(got, want) {
|
||||
t.Errorf("warnings missing %q:\n%s", want, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestWriteArtifact_AtomicWithMetaRoundTrip(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
path := filepath.Join(dir, "run.json")
|
||||
if err := os.WriteFile(path, []byte("previous"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
tests := []testkit.Test{{ID: "t1", Category: "c", Prompt: "p"}}
|
||||
results := []runner.Result{{Model: "m", TestID: "t1", Fingerprint: "fp"}}
|
||||
|
||||
if err := WriteArtifact(path, RunMeta{Version: "v1", Repeat: 1}, tests, []string{"m"}, results); err != nil {
|
||||
t.Fatalf("WriteArtifact() error = %v", err)
|
||||
}
|
||||
art, err := LoadArtifact(path)
|
||||
if err != nil {
|
||||
t.Fatalf("LoadArtifact() error = %v", err)
|
||||
}
|
||||
if art.Meta == nil || art.Meta.Version != "v1" || art.Meta.TestCount != 1 || art.Meta.CatalogHash == "" {
|
||||
t.Errorf("Meta = %+v, want version, test count and catalog hash set", art.Meta)
|
||||
}
|
||||
if got := art.Meta.Fingerprints["m"]; len(got) != 1 || got[0] != "fp" {
|
||||
t.Errorf("Fingerprints = %v, want m:[fp]", art.Meta.Fingerprints)
|
||||
}
|
||||
entries, _ := os.ReadDir(dir)
|
||||
if len(entries) != 1 {
|
||||
t.Errorf("dir has %d entries after write, want only the artifact (no temp leftovers)", len(entries))
|
||||
}
|
||||
}
|
||||
@@ -5,6 +5,7 @@ package runner
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
@@ -59,6 +60,8 @@ type Result struct {
|
||||
TestID string
|
||||
ResponseText string
|
||||
FinishReason string
|
||||
Fingerprint string
|
||||
ServedModel string
|
||||
Score eval.Score
|
||||
Latency time.Duration
|
||||
PromptTokens int
|
||||
@@ -83,7 +86,13 @@ func (r Result) TotalTokens() int {
|
||||
|
||||
// Config controls how the Runner fans work out.
|
||||
type Config struct {
|
||||
Reporter ProgressReporter
|
||||
Reporter ProgressReporter
|
||||
// Seed, when non-nil, is sent with every request for backends that
|
||||
// honor it.
|
||||
Seed *int
|
||||
// ModelConcurrency caps in-flight calls per model (on top of the global
|
||||
// Concurrency), so one slow local model is not flooded into timeouts.
|
||||
ModelConcurrency map[string]int
|
||||
Concurrency int
|
||||
Temperature float64
|
||||
MaxTokensDefault int
|
||||
@@ -138,15 +147,32 @@ func (r *Runner) Run(ctx context.Context, models []string, tests []testkit.Test)
|
||||
if concurrency <= 0 {
|
||||
concurrency = 1
|
||||
}
|
||||
g.SetLimit(concurrency)
|
||||
// Jobs wait for their per-model slot first, then a global slot, so a
|
||||
// capped slow model's queue never occupies global slots other models
|
||||
// need.
|
||||
global := make(chan struct{}, concurrency)
|
||||
|
||||
r.cfg.Reporter.ReportStart(len(tests), len(models), concurrency)
|
||||
|
||||
var completed int32
|
||||
// go.mod's go directive is 1.22+, so loop variables are already
|
||||
// per-iteration scoped; no manual i, j := i, j shadowing needed (N4).
|
||||
sems := make(map[string]chan struct{}, len(r.cfg.ModelConcurrency))
|
||||
for m, n := range r.cfg.ModelConcurrency {
|
||||
if n > 0 {
|
||||
sems[m] = make(chan struct{}, n)
|
||||
}
|
||||
}
|
||||
for i, j := range jobs {
|
||||
g.Go(func() error {
|
||||
if sem, ok := sems[j.model]; ok {
|
||||
if acquire(gctx, sem) {
|
||||
defer func() { <-sem }()
|
||||
}
|
||||
}
|
||||
if acquire(gctx, global) {
|
||||
defer func() { <-global }()
|
||||
}
|
||||
res := r.runOne(gctx, j.model, j.test)
|
||||
res.Attempt = j.attempt
|
||||
results[i] = res
|
||||
@@ -167,8 +193,32 @@ func (r *Runner) Run(ctx context.Context, models []string, tests []testkit.Test)
|
||||
return results
|
||||
}
|
||||
|
||||
// runOne executes a single (model, test) combination.
|
||||
func (r *Runner) runOne(ctx context.Context, model string, test testkit.Test) Result {
|
||||
// acquire takes a slot from sem, reporting false if ctx ended first.
|
||||
func acquire(ctx context.Context, sem chan struct{}) bool {
|
||||
select {
|
||||
case sem <- struct{}{}:
|
||||
return true
|
||||
case <-ctx.Done():
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// runOne executes a single (model, test) combination. A cancelled context
|
||||
// short-circuits to an error result, and a panic in an evaluator becomes an
|
||||
// error result instead of killing a multi-hour run.
|
||||
func (r *Runner) runOne(ctx context.Context, model string, test testkit.Test) (res Result) {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return Result{Model: model, TestID: test.ID, Err: err}
|
||||
}
|
||||
defer func() {
|
||||
if p := recover(); p != nil {
|
||||
res = Result{Model: model, TestID: test.ID, Err: fmt.Errorf("runner: panic scoring test: %v", p)}
|
||||
}
|
||||
}()
|
||||
return r.execute(ctx, model, test)
|
||||
}
|
||||
|
||||
func (r *Runner) execute(ctx context.Context, model string, test testkit.Test) Result {
|
||||
// cfg.MaxTokensDefault is a floor, not a fallback used only when the
|
||||
// test leaves MaxTokens unset: a per-test value only ever raises the
|
||||
// budget, it never lowers it below the default. A reasoning model can
|
||||
@@ -190,6 +240,7 @@ func (r *Runner) runOne(ctx context.Context, model string, test testkit.Test) Re
|
||||
Tools: test.Tools,
|
||||
MaxTokens: maxTokens,
|
||||
Temperature: r.cfg.Temperature,
|
||||
Seed: r.cfg.Seed,
|
||||
}
|
||||
|
||||
// Wall-clock across all attempts, recorded even on error: an error row
|
||||
@@ -223,6 +274,8 @@ func (r *Runner) runOne(ctx context.Context, model string, test testkit.Test) Re
|
||||
Score: score,
|
||||
ResponseText: scored,
|
||||
FinishReason: resp.FinishReason,
|
||||
Fingerprint: resp.Fingerprint,
|
||||
ServedModel: resp.ServedModel,
|
||||
Latency: resp.Latency,
|
||||
PromptTokens: resp.PromptTokens,
|
||||
CompletionTokens: resp.CompletionTokens,
|
||||
|
||||
@@ -410,3 +410,89 @@ func TestRunner_Run_RepeatSamplesEachCombination(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunner_Run_PassesSeedAndRecordsFingerprint(t *testing.T) {
|
||||
var gotSeed *int
|
||||
client := &mockClient{respond: func(req llm.Request) (llm.Response, error) {
|
||||
gotSeed = req.Seed
|
||||
return llm.Response{Text: "OK", Fingerprint: "fp-1", ServedModel: "served"}, nil
|
||||
}}
|
||||
seed := 7
|
||||
r := New(client, Config{Concurrency: 1, MaxTokensDefault: 100, Seed: &seed})
|
||||
res := r.Run(context.Background(), []string{"m"}, []testkit.Test{{ID: "t", Category: "c", Prompt: "p", Eval: echoEval()}})
|
||||
if gotSeed == nil || *gotSeed != 7 {
|
||||
t.Errorf("request seed = %v, want 7", gotSeed)
|
||||
}
|
||||
if res[0].Fingerprint != "fp-1" || res[0].ServedModel != "served" {
|
||||
t.Errorf("fingerprint/model = %q/%q, want fp-1/served", res[0].Fingerprint, res[0].ServedModel)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunner_Run_EvaluatorPanicBecomesErrorResult(t *testing.T) {
|
||||
boom := eval.EvaluatorFunc(func(context.Context, string) eval.Score { panic("boom") })
|
||||
r := New(&mockClient{}, Config{Concurrency: 2, MaxTokensDefault: 100})
|
||||
res := r.Run(context.Background(), []string{"m"}, []testkit.Test{
|
||||
{ID: "bad", Category: "c", Prompt: "p", Eval: boom},
|
||||
{ID: "good", Category: "c", Prompt: "p", Eval: echoEval()},
|
||||
})
|
||||
if res[0].Err == nil || !strings.Contains(res[0].Err.Error(), "boom") {
|
||||
t.Errorf("panicking test Err = %v, want panic error", res[0].Err)
|
||||
}
|
||||
if res[1].Err != nil || res[1].Score.Value != 1 {
|
||||
t.Errorf("sibling test = %+v, want unaffected pass", res[1])
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunner_Run_CancelledContextSkipsCalls(t *testing.T) {
|
||||
var calls int32
|
||||
client := &mockClient{respond: func(llm.Request) (llm.Response, error) {
|
||||
atomic.AddInt32(&calls, 1)
|
||||
return llm.Response{Text: "OK"}, nil
|
||||
}}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
r := New(client, Config{Concurrency: 2, MaxTokensDefault: 100})
|
||||
res := r.Run(ctx, []string{"m"}, []testkit.Test{{ID: "t", Category: "c", Prompt: "p", Eval: echoEval()}})
|
||||
if atomic.LoadInt32(&calls) != 0 {
|
||||
t.Errorf("client called %d times after cancel, want 0", calls)
|
||||
}
|
||||
if !errors.Is(res[0].Err, context.Canceled) {
|
||||
t.Errorf("Err = %v, want context.Canceled", res[0].Err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunner_Run_ModelConcurrencyCapDoesNotStarveOthers(t *testing.T) {
|
||||
var slowIn, slowMax int32
|
||||
fastDone := make(chan struct{}, 8)
|
||||
client := &mockClient{respond: func(req llm.Request) (llm.Response, error) {
|
||||
if req.Model == "slow" {
|
||||
cur := atomic.AddInt32(&slowIn, 1)
|
||||
for {
|
||||
m := atomic.LoadInt32(&slowMax)
|
||||
if cur <= m || atomic.CompareAndSwapInt32(&slowMax, m, cur) {
|
||||
break
|
||||
}
|
||||
}
|
||||
time.Sleep(30 * time.Millisecond)
|
||||
atomic.AddInt32(&slowIn, -1)
|
||||
} else {
|
||||
fastDone <- struct{}{}
|
||||
}
|
||||
return llm.Response{Text: "OK"}, nil
|
||||
}}
|
||||
r := New(client, Config{Concurrency: 2, MaxTokensDefault: 100, ModelConcurrency: map[string]int{"slow": 1}})
|
||||
tests := make([]testkit.Test, 6)
|
||||
for i := range tests {
|
||||
tests[i] = testkit.Test{ID: string(rune('a' + i)), Category: "c", Prompt: "p", Eval: echoEval()}
|
||||
}
|
||||
res := r.Run(context.Background(), []string{"slow", "fast"}, tests)
|
||||
if len(res) != 12 {
|
||||
t.Fatalf("results = %d, want 12", len(res))
|
||||
}
|
||||
if got := atomic.LoadInt32(&slowMax); got != 1 {
|
||||
t.Errorf("slow model max in-flight = %d, want 1", got)
|
||||
}
|
||||
if len(fastDone) != 6 {
|
||||
t.Errorf("fast calls = %d, want 6", len(fastDone))
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user