diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index eadd9166..1fa209ce 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -60,5 +60,6 @@ jobs: python-version: "3.12" - run: python -m pip install -e 'clients/python[dev]' -e 'integrations/nemo-gym[test]' - run: python -m pytest clients/python/tests -q + - run: python -m pytest internal/benchmarking/swebench/trajectories -q - run: ruff check clients/python - run: python -m pytest integrations/nemo-gym/tests -q diff --git a/cmd/benchmarking/swebench-loader/main.go b/cmd/benchmarking/swebench-loader/main.go new file mode 100644 index 00000000..1be90cbd --- /dev/null +++ b/cmd/benchmarking/swebench-loader/main.go @@ -0,0 +1,246 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// swebench-loader drives SWE-bench task images through the ate-env API as a +// simulated RL job and writes one JSONL event per operation. +// +// swebench-loader run -dataset manifest.jsonl -mode eval -images 500 -tasks-per-image 6 -width 500 -out run.jsonl +// swebench-loader run -dataset manifest.jsonl -mode steps -images 500 -rollouts 2 -steps 8 -park suspend -out run.jsonl +// swebench-loader report run.jsonl [-json summary.json] +package main + +import ( + "context" + "encoding/json" + "flag" + "fmt" + "log/slog" + "os" + "os/signal" + "strings" + "syscall" + "time" + + "github.com/agent-substrate/env/internal/benchmarking/swebench" +) + +func main() { + if len(os.Args) < 2 { + usage() + os.Exit(2) + } + var err error + switch os.Args[1] { + case "run": + err = runCmd(os.Args[2:]) + case "report": + err = reportCmd(os.Args[2:]) + case "-h", "--help", "help": + usage() + return + default: + usage() + os.Exit(2) + } + if err != nil { + fmt.Fprintln(os.Stderr, "error:", err) + os.Exit(1) + } +} + +func usage() { + fmt.Fprintln(os.Stderr, "usage: swebench-loader run [flags] | swebench-loader report FILE [-json OUT]") +} + +func runCmd(args []string) (retErr error) { + fs := flag.NewFlagSet("run", flag.ContinueOnError) + var ( + endpoint = fs.String("endpoint", "localhost:7777", "ate-env API address (host:port).") + connections = fs.Int("connections", 4, "gRPC connections to the API, handed out round-robin per environment; a Service balances per connection, so use at least the number of API replicas.") + atespace = fs.String("atespace", "ate-env", "Atespace the environments are created in.") + baseTemplate = fs.String("base-template", "default-template", "Base ActorTemplate the per-image templates derive from (ate-env manifest template).") + dataset = fs.String("dataset", "", "JSONL manifest of SWE-bench instances; a path or an https URL. Required.") + images = fs.Int("images", 0, "Use the first N instances of the dataset; 0 means all.") + instances = fs.String("instances", "", "Comma-separated instance ids to use instead of the first N.") + excludeFile = fs.String("exclude-file", "", "File of instance ids to leave out, one per line; text after the id is a comment.") + trajectory = fs.String("trajectory", swebench.DefaultTrajectory, fmt.Sprintf("Trajectory file or built-in name (%s).", strings.Join(swebench.Names(), ", "))) + mode = fs.String("mode", string(swebench.ModeEval), "eval (images x tasks-per-image, each create/replay/delete) or steps (images x rollouts kept across steps).") + width = fs.Int("width", 50, "Most tasks in flight at once.") + tasksPerImage = fs.Int("tasks-per-image", 1, "eval: replays per image.") + rollouts = fs.Int("rollouts", 2, "steps: environments per image.") + steps = fs.Int("steps", 8, "steps: training steps.") + park = fs.String("park", string(swebench.ParkSuspend), "steps: what happens between steps: suspend, delete or none.") + createTimeout = fs.Duration("create-timeout", 5*time.Minute, "Deadline for one create call.") + stepTimeout = fs.Duration("step-timeout", 10*time.Minute, "Default deadline for a run step without its own timeout.") + wakeBudget = fs.Duration("wake-budget", 3*time.Minute, "How long a task's first command is retried while the environment is still waking or its image is still being baked.") + wakeBackoff = fs.Duration("wake-backoff", swebench.DefaultWakeBackoff, "Initial pause between wake retries; doubles up to five times this.") + runID = fs.String("run-id", "", "Run id, prefix of every environment name: lowercase letters, digits and '-', at most 20 characters; default swe-.") + out = fs.String("out", "", "JSONL events file; default .jsonl in the working directory.") + keep = fs.Bool("keep", false, "Leave the environments in place when the run ends.") + dryRun = fs.Bool("dry-run", false, "Load and render everything, print the plan and exit without touching the API.") + ) + if err := fs.Parse(args); err != nil { + return err + } + if *dataset == "" { + return fmt.Errorf("-dataset is required") + } + if *runID == "" { + *runID = fmt.Sprintf("swe-%d", time.Now().Unix()) + } + if *out == "" { + *out = *runID + ".jsonl" + } + log := slog.New(slog.NewJSONHandler(os.Stderr, nil)) + + all, err := swebench.LoadDataset(*dataset) + if err != nil { + return err + } + var ids []string + if *instances != "" { + ids = strings.Split(*instances, ",") + } + var exclude []string + if *excludeFile != "" { + if exclude, err = swebench.ReadIDList(*excludeFile); err != nil { + return err + } + } + selected, err := swebench.Select(all, ids, exclude, *images) + if err != nil { + return err + } + if *images > 0 && len(selected) < *images && len(ids) == 0 { + fmt.Fprintf(os.Stderr, "note: %d instances requested, %d available after the exclude list\n", *images, len(selected)) + } + traj, err := swebench.LoadTrajectory(*trajectory) + if err != nil { + return err + } + for _, in := range selected { + if _, err := traj.Render(in); err != nil { + return err + } + } + + fanout := *tasksPerImage + if swebench.Mode(*mode) == swebench.ModeSteps { + fanout = *rollouts + } + log.Info("plan", slog.String("run", *runID), slog.String("mode", *mode), slog.Int("images", len(selected)), + slog.Int("environments", len(selected)*fanout), slog.Int("width", *width), slog.Int("steps", *steps), + slog.String("park", *park), slog.String("trajectory", traj.Name), slog.String("out", *out)) + if *dryRun { + return nil + } + + backend, err := swebench.NewEnvBackend(*endpoint, *atespace, *baseTemplate, *connections) + if err != nil { + return err + } + defer backend.Close() + + f, err := os.Create(*out) + if err != nil { + return err + } + rec := swebench.NewRecorder(f) + // Events lost to a write error, or a file that does not flush, must fail + // the run rather than produce a report with holes. + defer func() { + if cerr := f.Close(); cerr != nil && retErr == nil { + retErr = fmt.Errorf("closing %s: %w", *out, cerr) + } + if rerr := rec.Err(); rerr != nil && retErr == nil { + retErr = fmt.Errorf("writing events to %s: %w", *out, rerr) + } + }() + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + sum, err := swebench.Run(ctx, swebench.Config{ + RunID: *runID, + Backend: backend, + Instances: selected, + Trajectory: traj, + Recorder: rec, + Logger: log, + Mode: swebench.Mode(*mode), + Width: *width, + TasksPerImage: *tasksPerImage, + Rollouts: *rollouts, + Steps: *steps, + Park: swebench.Park(*park), + CreateTimeout: *createTimeout, + StepTimeout: *stepTimeout, + WakeBudget: *wakeBudget, + WakeBackoff: *wakeBackoff, + Keep: *keep, + }) + if sum != nil { + log.Info("run finished", slog.Int("tasks", sum.Tasks), slog.Int("failed", sum.Failed), + slog.Int("rewarded", sum.Rewarded), slog.Duration("duration", sum.Duration)) + } + if err != nil { + return err + } + if sum.Failed > 0 { + return fmt.Errorf("%d of %d tasks failed", sum.Failed, sum.Tasks) + } + return nil +} + +func reportCmd(args []string) error { + if len(args) < 1 { + return fmt.Errorf("report needs the JSONL file") + } + path := args[0] + fs := flag.NewFlagSet("report", flag.ContinueOnError) + jsonOut := fs.String("json", "", "Also write the summary as JSON to this path.") + unresolvedOut := fs.String("unresolved", "", "Write the unresolved instances to this path in -exclude-file format.") + unresolvedNote := fs.String("unresolved-note", "Instances graded unresolved", "First comment line of the -unresolved file.") + if err := fs.Parse(args[1:]); err != nil { + return err + } + f, err := os.Open(path) + if err != nil { + return err + } + defer f.Close() + rep, err := swebench.Summarize(f) + if err != nil { + return err + } + rep.WriteText(os.Stdout) + if *unresolvedOut != "" { + uf, err := os.Create(*unresolvedOut) + if err != nil { + return err + } + rep.WriteExclusions(uf, *unresolvedNote) + if err := uf.Close(); err != nil { + return err + } + } + if *jsonOut != "" { + b, err := json.MarshalIndent(rep, "", " ") + if err != nil { + return err + } + return os.WriteFile(*jsonOut, b, 0o644) + } + return nil +} diff --git a/internal/benchmarking/swebench/backend.go b/internal/benchmarking/swebench/backend.go new file mode 100644 index 00000000..9ec190ba --- /dev/null +++ b/internal/benchmarking/swebench/backend.go @@ -0,0 +1,247 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package swebench + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "io/fs" + "sync/atomic" + "time" + + env "github.com/agent-substrate/env/clients/go" + ateenvv1alpha "github.com/agent-substrate/env/proto/ateenv/v1alpha" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/types/known/durationpb" +) + +// Result is what a Run step returned. +type Result struct { + ExitCode int + Stdout string + Stderr string +} + +// Sandbox is the subset of an environment the loader drives. +type Sandbox interface { + Run(ctx context.Context, command string, timeout time.Duration) (Result, error) + WriteFile(ctx context.Context, path string, content []byte, mode fs.FileMode) error + Suspend(ctx context.Context) error + Delete(ctx context.Context) error +} + +// Backend creates sandboxes. The loader is written against this interface so +// tests run it against an in-memory fake and a second SDK or backend can be +// compared with the same workload. +type Backend interface { + Create(ctx context.Context, id, image string) (Sandbox, error) + // Delete removes an environment by name without a Sandbox handle, for + // one whose create call failed on the client side but may have + // succeeded on the server. + Delete(ctx context.Context, id string) error +} + +// EnvBackend is the Backend over the ate-env API through the Go SDK. +// +// It holds several connections and hands them out round-robin per +// environment: a Kubernetes Service balances per connection, so one +// connection would put every task on one API replica, and gRPC caps the +// concurrent streams on a connection. +type EnvBackend struct { + clients []*env.Client + next atomic.Uint64 + atespace string + baseTemplate string +} + +// NewEnvBackend dials the ate-env API over connections connections (at +// least one). Environments are created in atespace from baseTemplate with +// the task image as rootfs; the API derives and bakes one template per +// distinct image on first use. +func NewEnvBackend(endpoint, atespace, baseTemplate string, connections int) (*EnvBackend, error) { + if connections < 1 { + connections = 1 + } + b := &EnvBackend{atespace: atespace, baseTemplate: baseTemplate} + for i := 0; i < connections; i++ { + c, err := env.NewClient(env.ClientOptions{Endpoint: endpoint}) + if err != nil { + b.Close() + return nil, err + } + b.clients = append(b.clients, c) + } + return b, nil +} + +// Close releases the API connections. +func (b *EnvBackend) Close() error { + var first error + for _, c := range b.clients { + if err := c.Close(); err != nil && first == nil { + first = err + } + } + return first +} + +// pick returns the next connection round-robin. +func (b *EnvBackend) pick() *env.Client { + return b.clients[int(b.next.Add(1)-1)%len(b.clients)] +} + +// Delete implements Backend. +func (b *EnvBackend) Delete(ctx context.Context, id string) error { + return b.pick().Delete(ctx, b.atespace, id) +} + +// Create implements Backend. The environment keeps the connection it was +// created on for the rest of its life. +func (b *EnvBackend) Create(ctx context.Context, id, image string) (Sandbox, error) { + e, err := b.pick().Create(ctx, &ateenvv1alpha.CreateEnvironmentRequest{ + Id: id, + Atespace: b.atespace, + Template: &ateenvv1alpha.Template{Name: b.baseTemplate, Atespace: b.atespace}, + RootfsImage: image, + }) + if err != nil { + return nil, fmt.Errorf("create %s: %w", id, err) + } + return envSandbox{e: e}, nil +} + +type envSandbox struct { + e *env.Env +} + +// Polling bounds for a running process: short at first so quick commands +// return quickly, backing off so a long test run does not hammer the API. +// Variables so tests can shorten them. +var ( + pollMin = 50 * time.Millisecond + pollMax = 2 * time.Second +) + +const ( + // pollRetries is how many consecutive transient poll errors are + // tolerated before the step fails; a wake or an API pod restart costs a + // few. + pollRetries = 10 + // killTimeout bounds the kill of a process Run abandons. + killTimeout = 10 * time.Second +) + +// Run starts the command, polls until it exits and then reads its output. +// +// The SDK's streaming Shell holds one request open through the router for +// the whole command, and the router's route timeout (five minutes by +// default) cuts it off; SWE-bench test runs take up to twelve minutes. +// Start, poll and read are each short calls, so no single request can +// outlive it. This belongs in the SDK as an option of Env.Run, since every +// SDK user with long commands meets the same timeout; it lives here until +// that change lands so the benchmark does not wait on it. +// +// If Run returns before the process has exited, it kills the process, so a +// retry of the same command never runs beside the first copy. +func (s envSandbox) Run(ctx context.Context, command string, timeout time.Duration) (res Result, err error) { + start := &ateenvv1alpha.StartProcessRequest{Command: []string{"sh", "-c", command}} + if timeout > 0 { + start.Timeout = durationpb.New(timeout) + } + proc, err := s.e.StartProcess(ctx, start) + if err != nil { + return Result{}, err + } + exited := false + defer func() { + if exited { + return + } + kctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), killTimeout) + defer cancel() + if _, kerr := proc.Kill(kctx); kerr != nil && !errors.Is(kerr, env.ErrProcessExited) && err != nil { + err = fmt.Errorf("%w (and killing the process failed: %v)", err, kerr) + } + }() + + wait := pollMin + failures := 0 + for { + select { + case <-ctx.Done(): + return Result{}, ctx.Err() + case <-time.After(wait): + } + p, perr := s.e.FindProcess(ctx, proc.ID()) + switch { + case perr == nil: + failures = 0 + if p.GetState() == ateenvv1alpha.ProcessState_PROCESS_STATE_EXITED { + exited = true + return s.collect(ctx, proc, int(p.GetExitCode())) + } + case status.Code(perr) == codes.Unavailable || status.Code(perr) == codes.DeadlineExceeded: + failures++ + if failures > pollRetries { + return Result{}, fmt.Errorf("poll %s: %w", proc.ID(), perr) + } + default: + return Result{}, perr + } + if wait < pollMax { + wait = min(wait*2, pollMax) + } + } +} + +// collect reads the exited process's buffered output in one non-following +// call. +func (s envSandbox) collect(ctx context.Context, proc *env.Process, exitCode int) (Result, error) { + stream, err := proc.Output(ctx, &ateenvv1alpha.StreamProcessOutputRequest{}) + if err != nil { + return Result{}, err + } + var stdout, stderr bytes.Buffer + for { + msg, err := stream.Recv() + if errors.Is(err, io.EOF) { + break + } + if err != nil { + return Result{}, err + } + switch out := msg.GetOutput().(type) { + case *ateenvv1alpha.ProcessOutput_Stdout: + stdout.Write(out.Stdout) + case *ateenvv1alpha.ProcessOutput_Stderr: + stderr.Write(out.Stderr) + case *ateenvv1alpha.ProcessOutput_Exit: + exitCode = int(out.Exit.GetExitCode()) + } + } + return Result{ExitCode: exitCode, Stdout: stdout.String(), Stderr: stderr.String()}, nil +} + +func (s envSandbox) WriteFile(ctx context.Context, path string, content []byte, mode fs.FileMode) error { + return s.e.WriteFile(ctx, path, bytes.NewReader(content), mode) +} + +func (s envSandbox) Suspend(ctx context.Context) error { return s.e.Suspend(ctx) } + +func (s envSandbox) Delete(ctx context.Context) error { return s.e.Delete(ctx) } diff --git a/internal/benchmarking/swebench/backend_test.go b/internal/benchmarking/swebench/backend_test.go new file mode 100644 index 00000000..f0cc88d4 --- /dev/null +++ b/internal/benchmarking/swebench/backend_test.go @@ -0,0 +1,270 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package swebench + +import ( + "context" + "fmt" + "net" + "strings" + "sync" + "testing" + "time" + + env "github.com/agent-substrate/env/clients/go" + ateenvv1alpha "github.com/agent-substrate/env/proto/ateenv/v1alpha" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" + "google.golang.org/grpc/test/bufconn" +) + +// fakeAPI is an in-process ate-env API: enough of the environment, process +// and file services for the backend's start, poll and read path. +type fakeAPI struct { + ateenvv1alpha.UnimplementedEnvironmentServiceServer + ateenvv1alpha.UnimplementedProcessServiceServer + ateenvv1alpha.UnimplementedFileSystemServiceServer + + mu sync.Mutex + // pollsUntilExit is how many GetProcess calls a process stays running; + // unavailablePolls is how many GetProcess calls answer Unavailable first. + pollsUntilExit int + unavailablePolls int + exitCode int32 + stdout, stderr string + + nextID int + polls map[string]int + exited map[string]bool + killed []string + created []string + deleted []string + commands []string +} + +func (f *fakeAPI) CreateEnvironment(_ context.Context, req *ateenvv1alpha.CreateEnvironmentRequest) (*ateenvv1alpha.CreateEnvironmentResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.created = append(f.created, req.GetId()) + return &ateenvv1alpha.CreateEnvironmentResponse{Environment: &ateenvv1alpha.Environment{Id: req.GetId(), Atespace: req.GetAtespace()}}, nil +} + +func (f *fakeAPI) DeleteEnvironment(_ context.Context, req *ateenvv1alpha.DeleteEnvironmentRequest) (*ateenvv1alpha.DeleteEnvironmentResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.deleted = append(f.deleted, req.GetId()) + return &ateenvv1alpha.DeleteEnvironmentResponse{}, nil +} + +func (f *fakeAPI) StartProcess(_ context.Context, req *ateenvv1alpha.StartProcessRequest) (*ateenvv1alpha.Process, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.nextID++ + id := fmt.Sprintf("p%d", f.nextID) + f.commands = append(f.commands, strings.Join(req.GetCommand(), " ")) + return &ateenvv1alpha.Process{ProcessId: id, State: ateenvv1alpha.ProcessState_PROCESS_STATE_RUNNING}, nil +} + +func (f *fakeAPI) GetProcess(_ context.Context, req *ateenvv1alpha.GetProcessRequest) (*ateenvv1alpha.Process, error) { + f.mu.Lock() + defer f.mu.Unlock() + if f.unavailablePolls > 0 { + f.unavailablePolls-- + return nil, status.Error(codes.Unavailable, "router: no healthy upstream") + } + f.polls[req.GetProcessId()]++ + p := &ateenvv1alpha.Process{ProcessId: req.GetProcessId(), State: ateenvv1alpha.ProcessState_PROCESS_STATE_RUNNING} + if f.polls[req.GetProcessId()] > f.pollsUntilExit { + f.exited[req.GetProcessId()] = true + p.State = ateenvv1alpha.ProcessState_PROCESS_STATE_EXITED + p.ExitCode = f.exitCode + } + return p, nil +} + +func (f *fakeAPI) StreamProcessOutput(req *ateenvv1alpha.StreamProcessOutputRequest, stream grpc.ServerStreamingServer[ateenvv1alpha.ProcessOutput]) error { + f.mu.Lock() + stdout, stderr, exited, code := f.stdout, f.stderr, f.exited[req.GetProcessId()], f.exitCode + f.mu.Unlock() + if stdout != "" { + if err := stream.Send(&ateenvv1alpha.ProcessOutput{Output: &ateenvv1alpha.ProcessOutput_Stdout{Stdout: []byte(stdout)}}); err != nil { + return err + } + } + if stderr != "" { + if err := stream.Send(&ateenvv1alpha.ProcessOutput{Output: &ateenvv1alpha.ProcessOutput_Stderr{Stderr: []byte(stderr)}}); err != nil { + return err + } + } + if exited { + return stream.Send(&ateenvv1alpha.ProcessOutput{Output: &ateenvv1alpha.ProcessOutput_Exit{Exit: &ateenvv1alpha.Process{ + ProcessId: req.GetProcessId(), State: ateenvv1alpha.ProcessState_PROCESS_STATE_EXITED, ExitCode: code}}}) + } + return nil +} + +func (f *fakeAPI) SignalProcess(_ context.Context, req *ateenvv1alpha.SignalProcessRequest) (*ateenvv1alpha.Process, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.killed = append(f.killed, req.GetProcessId()) + f.exited[req.GetProcessId()] = true + return &ateenvv1alpha.Process{ProcessId: req.GetProcessId(), State: ateenvv1alpha.ProcessState_PROCESS_STATE_EXITED, ExitCode: 137}, nil +} + +// newFakeBackend serves fakeAPI over an in-memory connection and returns an +// EnvBackend on it, the same path the binary uses with a real address. +func newFakeBackend(t *testing.T, api *fakeAPI) *EnvBackend { + t.Helper() + api.polls = map[string]int{} + api.exited = map[string]bool{} + lis := bufconn.Listen(1 << 20) + srv := grpc.NewServer() + ateenvv1alpha.RegisterEnvironmentServiceServer(srv, api) + ateenvv1alpha.RegisterProcessServiceServer(srv, api) + ateenvv1alpha.RegisterFileSystemServiceServer(srv, api) + go func() { _ = srv.Serve(lis) }() + conn, err := grpc.NewClient("passthrough:///bufconn", + grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { return lis.DialContext(ctx) }), + grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + t.Fatal(err) + } + c, err := env.NewClient(env.ClientOptions{GRPCConn: conn}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = c.Close(); _ = conn.Close(); srv.Stop() }) + return &EnvBackend{clients: []*env.Client{c}, atespace: "ate-env", baseTemplate: "default-template"} +} + +func fastPolls(t *testing.T) { + t.Helper() + oldMin, oldMax := pollMin, pollMax + pollMin, pollMax = time.Millisecond, 5*time.Millisecond + t.Cleanup(func() { pollMin, pollMax = oldMin, oldMax }) +} + +func TestEnvBackendRunCollectsExitAndOutput(t *testing.T) { + fastPolls(t) + api := &fakeAPI{pollsUntilExit: 3, exitCode: 4, stdout: "out", stderr: "err"} + b := newFakeBackend(t, api) + sb, err := b.Create(context.Background(), "e1", "reg/img@"+goodDigest) + if err != nil { + t.Fatal(err) + } + res, err := sb.Run(context.Background(), "echo hi", time.Minute) + if err != nil { + t.Fatal(err) + } + if res.ExitCode != 4 || res.Stdout != "out" || res.Stderr != "err" { + t.Fatalf("result %+v", res) + } + if len(api.killed) != 0 { + t.Fatalf("a process that exited must not be killed: %v", api.killed) + } + if len(api.created) != 1 || api.created[0] != "e1" { + t.Fatalf("created %v", api.created) + } + if !strings.HasSuffix(api.commands[0], "echo hi") { + t.Fatalf("command %q", api.commands[0]) + } +} + +func TestEnvBackendRunToleratesTransientPolls(t *testing.T) { + fastPolls(t) + api := &fakeAPI{pollsUntilExit: 1, unavailablePolls: pollRetries} + b := newFakeBackend(t, api) + sb, _ := b.Create(context.Background(), "e1", "reg/img@"+goodDigest) + if _, err := sb.Run(context.Background(), "true", time.Minute); err != nil { + t.Fatalf("%d transient poll errors must be tolerated: %v", pollRetries, err) + } +} + +func TestEnvBackendRunGivesUpAndKills(t *testing.T) { + fastPolls(t) + api := &fakeAPI{pollsUntilExit: 100, unavailablePolls: pollRetries + 1} + b := newFakeBackend(t, api) + sb, _ := b.Create(context.Background(), "e1", "reg/img@"+goodDigest) + _, err := sb.Run(context.Background(), "sleep 100", time.Minute) + if err == nil || status.Code(err) != codes.Unavailable { + t.Fatalf("expected the poll error after the retry limit, got %v", err) + } + if len(api.killed) != 1 { + t.Fatalf("an abandoned process must be killed once, got %v", api.killed) + } +} + +func TestEnvBackendRunKillsOnCancel(t *testing.T) { + fastPolls(t) + api := &fakeAPI{pollsUntilExit: 1 << 30} + b := newFakeBackend(t, api) + sb, _ := b.Create(context.Background(), "e1", "reg/img@"+goodDigest) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond) + defer cancel() + if _, err := sb.Run(ctx, "sleep 100", time.Minute); err == nil { + t.Fatal("a cancelled Run must return an error") + } + if len(api.killed) != 1 { + t.Fatalf("the process must be killed after a cancel, got %v", api.killed) + } +} + +func TestEnvBackendDeleteByName(t *testing.T) { + api := &fakeAPI{} + b := newFakeBackend(t, api) + if err := b.Delete(context.Background(), "gone"); err != nil { + t.Fatal(err) + } + if len(api.deleted) != 1 || api.deleted[0] != "gone" { + t.Fatalf("deleted %v", api.deleted) + } +} + +func TestEnvBackendSpreadsConnections(t *testing.T) { + // grpc.NewClient does not dial until the first call, so a backend over + // an unreachable address is enough to check the pool and the rotation. + b, err := NewEnvBackend("127.0.0.1:1", "ate-env", "default-template", 3) + if err != nil { + t.Fatal(err) + } + defer b.Close() + if len(b.clients) != 3 { + t.Fatalf("%d clients, want 3", len(b.clients)) + } + seen := map[int]int{} + for i := 0; i < 9; i++ { + c := b.pick() + for j, cc := range b.clients { + if cc == c { + seen[j]++ + } + } + } + for j := 0; j < 3; j++ { + if seen[j] != 3 { + t.Fatalf("connection %d picked %d times, want 3: %v", j, seen[j], seen) + } + } + one, err := NewEnvBackend("127.0.0.1:1", "ate-env", "default-template", 0) + if err != nil { + t.Fatal(err) + } + defer one.Close() + if len(one.clients) != 1 { + t.Fatalf("connections 0 must mean 1, got %d", len(one.clients)) + } +} diff --git a/internal/benchmarking/swebench/dataset.go b/internal/benchmarking/swebench/dataset.go new file mode 100644 index 00000000..32016462 --- /dev/null +++ b/internal/benchmarking/swebench/dataset.go @@ -0,0 +1,216 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package swebench + +import ( + "bufio" + "bytes" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "os" + "regexp" + "strings" + "time" +) + +// Instance is one SWE-bench task: the digest-pinned task image and the +// material a trajectory needs to evaluate it. The manifest is produced by +// benchmarking/swebench/make-dataset.py, one JSON object per line. +type Instance struct { + InstanceID string `json:"instance_id"` + Repo string `json:"repo"` + // Image is the task image, pinned by digest (repo@sha256:...). Substrate + // requires pinned images because changing the image invalidates snapshots. + Image string `json:"image"` + // Patch is the gold patch; applying it makes FailToPass pass. + Patch string `json:"patch"` + // TestPatch adds or changes the tests that check the fix. + TestPatch string `json:"test_patch"` + // EvalScript is the SWE-bench harness eval script for this instance: it + // applies TestPatch, runs the repo's test command and restores the test + // files. Exit code 0 means every selected test passed. + EvalScript string `json:"eval_script"` + FailToPass []string `json:"fail_to_pass"` + PassToPass []string `json:"pass_to_pass"` + // LogParser names the SWE-bench harness parser for this repo's test + // output (parse_log_pytest, parse_log_django, ...); EvalType is + // pass_and_fail or fail_only. Both come from the dataset row. + LogParser string `json:"log_parser"` + EvalType string `json:"eval_type"` +} + +var digestPinned = regexp.MustCompile(`@sha256:[0-9a-f]{64}$`) + +// Validate reports the first thing that would make the instance unusable. +func (in Instance) Validate() error { + switch { + case in.InstanceID == "": + return errors.New("instance_id is required") + case in.Image == "": + return fmt.Errorf("%s: image is required", in.InstanceID) + case !digestPinned.MatchString(in.Image): + return fmt.Errorf("%s: image %q is not digest-pinned", in.InstanceID, in.Image) + case in.EvalScript == "": + return fmt.Errorf("%s: eval_script is required", in.InstanceID) + case in.LogParser == "": + return fmt.Errorf("%s: log_parser is required", in.InstanceID) + } + return nil +} + +// datasetFetchTimeout bounds a manifest download; a stalled server must not +// hang the loader before it starts. +const datasetFetchTimeout = 2 * time.Minute + +// LoadDataset reads a JSONL manifest from a local path or an https URL. The +// manifest carries scripts that run inside the sandboxes, so a remote source +// must be https. +func LoadDataset(source string) ([]Instance, error) { + var r io.ReadCloser + if strings.HasPrefix(source, "http://") { + return nil, fmt.Errorf("dataset %s: remote manifests must use https", source) + } + if strings.HasPrefix(source, "https://") { + client := &http.Client{Timeout: datasetFetchTimeout} + resp, err := client.Get(source) + if err != nil { + return nil, fmt.Errorf("fetch dataset: %w", err) + } + if resp.StatusCode != http.StatusOK { + resp.Body.Close() + return nil, fmt.Errorf("fetch dataset: HTTP %d", resp.StatusCode) + } + r = resp.Body + } else { + f, err := os.Open(source) + if err != nil { + return nil, fmt.Errorf("open dataset: %w", err) + } + r = f + } + defer r.Close() + return ParseDataset(r) +} + +// ParseDataset decodes a JSONL manifest. Blank lines are skipped; every +// record must validate and instance ids must be unique. +func ParseDataset(r io.Reader) ([]Instance, error) { + sc := bufio.NewScanner(r) + // Eval scripts and patches run to hundreds of KiB for some repos. + sc.Buffer(make([]byte, 0, 1<<20), 64<<20) + var out []Instance + seen := map[string]bool{} + line := 0 + for sc.Scan() { + line++ + b := bytes.TrimSpace(sc.Bytes()) + if len(b) == 0 { + continue + } + var in Instance + if err := json.Unmarshal(b, &in); err != nil { + return nil, fmt.Errorf("dataset line %d: %w", line, err) + } + if err := in.Validate(); err != nil { + return nil, fmt.Errorf("dataset line %d: %w", line, err) + } + if seen[in.InstanceID] { + return nil, fmt.Errorf("dataset line %d: duplicate instance %s", line, in.InstanceID) + } + seen[in.InstanceID] = true + out = append(out, in) + } + if err := sc.Err(); err != nil { + return nil, fmt.Errorf("read dataset: %w", err) + } + if len(out) == 0 { + return nil, errors.New("dataset is empty") + } + return out, nil +} + +// Select narrows a dataset: exclude drops the named instances first (a +// known-bad list; unknown names are ignored), ids then keeps only the named +// instances (in manifest order) and errors on an unknown or excluded id, and +// limit > 0 keeps the first limit of what is left. +func Select(all []Instance, ids, exclude []string, limit int) ([]Instance, error) { + out := all + drop := map[string]bool{} + if len(exclude) > 0 { + for _, id := range exclude { + drop[id] = true + } + out = nil + for _, in := range all { + if !drop[in.InstanceID] { + out = append(out, in) + } + } + all = out + } + if len(ids) > 0 { + for _, id := range ids { + if drop[id] { + return nil, fmt.Errorf("instance %s is on the exclude list", id) + } + } + want := map[string]bool{} + for _, id := range ids { + want[id] = true + } + out = nil + for _, in := range all { + if want[in.InstanceID] { + out = append(out, in) + delete(want, in.InstanceID) + } + } + if len(want) > 0 { + missing := make([]string, 0, len(want)) + for id := range want { + missing = append(missing, id) + } + return nil, fmt.Errorf("instances not in dataset: %s", strings.Join(missing, ", ")) + } + } + if limit > 0 && limit < len(out) { + out = out[:limit] + } + return out, nil +} + +// ReadIDList reads instance ids from a file, one per line. Blank lines and +// lines starting with '#' are skipped, and anything after the first +// whitespace on a line is a comment, so a list can say why each id is there. +func ReadIDList(path string) ([]string, error) { + f, err := os.Open(path) + if err != nil { + return nil, err + } + defer f.Close() + var ids []string + sc := bufio.NewScanner(f) + for sc.Scan() { + line := strings.TrimSpace(sc.Text()) + if line == "" || strings.HasPrefix(line, "#") { + continue + } + ids = append(ids, strings.Fields(line)[0]) + } + return ids, sc.Err() +} diff --git a/internal/benchmarking/swebench/dataset_test.go b/internal/benchmarking/swebench/dataset_test.go new file mode 100644 index 00000000..f2955dea --- /dev/null +++ b/internal/benchmarking/swebench/dataset_test.go @@ -0,0 +1,110 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package swebench + +import ( + "os" + "path/filepath" + "strings" + "testing" +) + +const goodDigest = "sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + +func manifest(ids ...string) string { + var b strings.Builder + for _, id := range ids { + b.WriteString(`{"instance_id":"` + id + `","repo":"r","image":"reg/img@` + goodDigest + `","patch":"p","eval_script":"e","log_parser":"parse_log_pytest"}` + "\n") + } + return b.String() +} + +func TestParseDataset(t *testing.T) { + got, err := ParseDataset(strings.NewReader("\n" + manifest("a__a-1", "b__b-2") + "\n")) + if err != nil { + t.Fatal(err) + } + if len(got) != 2 || got[0].InstanceID != "a__a-1" || got[1].InstanceID != "b__b-2" { + t.Fatalf("got %+v", got) + } +} + +func TestParseDatasetRejects(t *testing.T) { + cases := map[string]string{ + "empty": "", + "bad json": "{not json}\n", + "no id": `{"image":"reg/img@` + goodDigest + `"}` + "\n", + "no image": `{"instance_id":"x"}` + "\n", + "tag not pin": `{"instance_id":"x","image":"reg/img:latest"}` + "\n", + "duplicate": manifest("a", "a"), + "no eval": `{"instance_id":"x","image":"reg/img@` + goodDigest + `","log_parser":"parse_log_pytest"}` + "\n", + "no parser": `{"instance_id":"x","image":"reg/img@` + goodDigest + `","eval_script":"e"}` + "\n", + } + for name, in := range cases { + if _, err := ParseDataset(strings.NewReader(in)); err == nil { + t.Errorf("%s: expected an error", name) + } + } +} + +func TestSelect(t *testing.T) { + all, err := ParseDataset(strings.NewReader(manifest("a", "b", "c"))) + if err != nil { + t.Fatal(err) + } + got, err := Select(all, nil, nil, 2) + if err != nil || len(got) != 2 || got[1].InstanceID != "b" { + t.Fatalf("limit: got %+v, %v", got, err) + } + got, err = Select(all, []string{"c", "a"}, nil, 0) + if err != nil || len(got) != 2 || got[0].InstanceID != "a" || got[1].InstanceID != "c" { + t.Fatalf("ids keep manifest order: got %+v, %v", got, err) + } + got, err = Select(all, []string{"c", "a"}, nil, 1) + if err != nil || len(got) != 1 || got[0].InstanceID != "a" { + t.Fatalf("ids then limit: got %+v, %v", got, err) + } + if _, err := Select(all, []string{"zz"}, nil, 0); err == nil { + t.Fatal("unknown id must error") + } +} + +func TestSelectExcludeAndIDList(t *testing.T) { + all, err := ParseDataset(strings.NewReader(manifest("a", "b", "c"))) + if err != nil { + t.Fatal(err) + } + got, err := Select(all, nil, []string{"b", "unknown"}, 0) + if err != nil || len(got) != 2 || got[0].InstanceID != "a" || got[1].InstanceID != "c" { + t.Fatalf("exclude: got %+v, %v", got, err) + } + if _, err := Select(all, []string{"b"}, []string{"b"}, 0); err == nil || !strings.Contains(err.Error(), "exclude list") { + t.Fatalf("an excluded id asked for by name must say so: %v", err) + } + path := filepath.Join(t.TempDir(), "ids.txt") + if err := os.WriteFile(path, []byte("# known network-dependent\nb needs pypi\n\n c\n"), 0o644); err != nil { + t.Fatal(err) + } + ids, err := ReadIDList(path) + if err != nil || len(ids) != 2 || ids[0] != "b" || ids[1] != "c" { + t.Fatalf("ReadIDList: %v, %v", ids, err) + } +} + +func TestLoadDatasetRefusesPlainHTTP(t *testing.T) { + if _, err := LoadDataset("http://example.invalid/manifest.jsonl"); err == nil || !strings.Contains(err.Error(), "https") { + t.Fatalf("plain http must be refused: %v", err) + } +} diff --git a/internal/benchmarking/swebench/doc.go b/internal/benchmarking/swebench/doc.go new file mode 100644 index 00000000..ae8ef3ef --- /dev/null +++ b/internal/benchmarking/swebench/doc.go @@ -0,0 +1,35 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package swebench holds the data types of the Fast Substrate RL benchmark on +// SWE-bench Verified, and the runner that drives them through the ate-env API. +// +// - Instance and LoadDataset: the JSONL manifest, one record per SWE-bench +// instance (digest-pinned task image, gold patch, harness eval script, +// the tests that decide the verdict, the repo's log parser). +// - Trajectory: the scripted rollout a task replays (run a command or write +// a file, with think time, timeout, expected exit and one reward step), +// with setup steps that run once per environment and reset steps that +// run before each later step. The built-in swebench-gold trajectory +// applies the gold patch, runs the harness eval script and grades the +// log with swebench-grade.py, a port of the SWE-bench harness's rules. +// - Event and Recorder: one JSON line per operation of a run. +// - Summarize and Build: the report over those events, with percentiles, +// retry counts, a per-step table and the unresolved instances in +// exclude-file format. +// +// The layout mirrors substrate's internal/benchmarking/boomer so a user class +// can be lifted into that harness later; it is not imported because Go does +// not allow importing another module's internal packages. +package swebench diff --git a/internal/benchmarking/swebench/events.go b/internal/benchmarking/swebench/events.go new file mode 100644 index 00000000..78d51c14 --- /dev/null +++ b/internal/benchmarking/swebench/events.go @@ -0,0 +1,111 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package swebench + +import ( + "encoding/json" + "io" + "sync" + "time" +) + +// Event names. Every timed operation of a run is one of these. +const ( + EventCreate = "create" // CreateEnvironment call + EventTTFI = "ttfi" // create start to the first command's return + EventWake = "wake" // first command after a park, including the wake it triggers + EventStep = "step" // one trajectory step (name is the step name) + EventTask = "task" // one trajectory replay end to end + EventSuspend = "suspend" // SuspendEnvironment call + EventDelete = "delete" // DeleteEnvironment call + EventBarrier = "step_barrier" // steps mode: first task start to last task end in a training step + EventPark = "park" // steps mode: parking the whole fleet between steps + EventRun = "run" // whole run +) + +// Event is one JSONL record. Dur is wall time as seen by the loader; it +// includes the network and the API server, which is what a trainer sees. +type Event struct { + Time time.Time `json:"ts"` + RunID string `json:"run"` + Event string `json:"event"` + Name string `json:"name,omitempty"` + Env string `json:"env,omitempty"` + Instance string `json:"instance,omitempty"` + // Step is the training step in steps mode (1-based); 0 in eval mode. + Step int `json:"step,omitempty"` + // Task numbers the replays of one instance (eval: task index; steps: + // rollout index). + Task int `json:"task,omitempty"` + DurMS float64 `json:"dur_ms"` + OK bool `json:"ok"` + Exit *int `json:"exit,omitempty"` + Reward *int `json:"reward,omitempty"` + Err string `json:"err,omitempty"` + // Count is the number of tasks or environments an aggregate event covers. + Count int `json:"count,omitempty"` + // Tail is the last bytes of a failed or reward step's output. + Tail string `json:"tail,omitempty"` + // Attempts is how many tries a ttfi or wake took; above 1 means the + // environment was not serving when first asked. + Attempts int `json:"attempts,omitempty"` +} + +// Recorder writes events as JSON lines. It is safe for concurrent use. +type Recorder struct { + mu sync.Mutex + w io.Writer + enc *json.Encoder + n int + err error +} + +// NewRecorder writes to w; a nil w discards events. +func NewRecorder(w io.Writer) *Recorder { + if w == nil { + w = io.Discard + } + return &Recorder{w: w, enc: json.NewEncoder(w)} +} + +// Record writes one event, stamping the time if unset. +func (r *Recorder) Record(e Event) { + if e.Time.IsZero() { + e.Time = time.Now() + } + r.mu.Lock() + defer r.mu.Unlock() + if err := r.enc.Encode(e); err != nil && r.err == nil { + r.err = err + } + r.n++ +} + +// Len is the number of events written so far. +func (r *Recorder) Len() int { + r.mu.Lock() + defer r.mu.Unlock() + return r.n +} + +// Err is the first write error, if any; a run whose events were lost must +// not report success. +func (r *Recorder) Err() error { + r.mu.Lock() + defer r.mu.Unlock() + return r.err +} + +func intp(v int) *int { return &v } diff --git a/internal/benchmarking/swebench/report.go b/internal/benchmarking/swebench/report.go new file mode 100644 index 00000000..b78b8e13 --- /dev/null +++ b/internal/benchmarking/swebench/report.go @@ -0,0 +1,317 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package swebench + +import ( + "bufio" + "bytes" + "encoding/json" + "fmt" + "io" + "math" + "sort" + "strings" + "text/tabwriter" +) + +// Stat summarizes one kind of event. +type Stat struct { + Event string `json:"event"` + Name string `json:"name,omitempty"` + Count int `json:"count"` + Failed int `json:"failed"` + P50MS float64 `json:"p50_ms"` + P95MS float64 `json:"p95_ms"` + MaxMS float64 `json:"max_ms"` + // Retried counts ttfi and wake events that took more than one attempt: + // the environment was not serving when first asked. + Retried int `json:"retried,omitempty"` +} + +// Unresolved is an instance whose reward step scored 0, with the grader's +// verdict when the step's output ended in one. +type Unresolved struct { + Instance string `json:"instance"` + Reason string `json:"reason"` +} + +// StepStat is one training step in steps mode. +type StepStat struct { + Step int `json:"step"` + Tasks int `json:"tasks"` + Failed int `json:"failed"` + Rewarded int `json:"rewarded"` + // Live is how many environments were still in the fleet at this step; + // steps mode drops an environment after its first failure, so a + // shrinking Live makes later steps' counts incomparable. + Live int `json:"live,omitempty"` + BarrierMS float64 `json:"barrier_ms"` + ParkMS float64 `json:"park_ms"` + WakeP50MS float64 `json:"wake_p50_ms"` + WakeP95MS float64 `json:"wake_p95_ms"` +} + +// Report is the summary of one run's JSONL, stable enough to diff between +// runs. +type Report struct { + RunID string `json:"run"` + Mode string `json:"mode,omitempty"` + RunMS float64 `json:"run_ms,omitempty"` + Tasks int `json:"tasks"` + Failed int `json:"failed"` + Rewarded int `json:"rewarded"` + RewardRate float64 `json:"reward_rate"` + Stats []Stat `json:"stats"` + Steps []StepStat `json:"steps,omitempty"` + // Errors counts distinct error strings, most frequent first. + Errors []ErrorCount `json:"errors,omitempty"` + // Unresolved lists the instances whose reward was 0, once each. + Unresolved []Unresolved `json:"unresolved,omitempty"` +} + +// ErrorCount is one error string and how often it occurred. +type ErrorCount struct { + Err string `json:"err"` + Count int `json:"count"` +} + +// Summarize reads JSONL events and builds the report. +func Summarize(r io.Reader) (*Report, error) { + sc := bufio.NewScanner(r) + sc.Buffer(make([]byte, 0, 1<<20), 16<<20) + var events []Event + for sc.Scan() { + b := bytes.TrimSpace(sc.Bytes()) + if len(b) == 0 { + continue + } + var e Event + if err := json.Unmarshal(b, &e); err != nil { + return nil, fmt.Errorf("parse event: %w", err) + } + events = append(events, e) + } + if err := sc.Err(); err != nil { + return nil, err + } + return Build(events), nil +} + +// Build computes the report from events in memory. +func Build(events []Event) *Report { + rep := &Report{} + type key struct{ event, name string } + durs := map[key][]float64{} + failed := map[key]int{} + retried := map[key]int{} + unresolved := map[string]string{} + var order []key + steps := map[int]*StepStat{} + wakes := map[int][]float64{} + errs := map[string]int{} + + for _, e := range events { + if rep.RunID == "" { + rep.RunID = e.RunID + } + k := key{e.Event, ""} + if e.Event == EventStep || e.Event == EventPark { + k.name = e.Name + } + if _, ok := durs[k]; !ok { + order = append(order, k) + durs[k] = nil + } + if e.Attempts > 1 { + retried[k]++ + } + if e.Event == EventStep && e.Reward != nil && *e.Reward == 0 { + if _, seen := unresolved[e.Instance]; !seen { + unresolved[e.Instance] = verdict(e.Tail) + } + } + if e.OK { + durs[k] = append(durs[k], e.DurMS) + } else { + failed[k]++ + if e.Err != "" { + errs[e.Err]++ + } + } + switch e.Event { + case EventRun: + rep.Mode = e.Name + rep.RunMS = e.DurMS + case EventTask: + rep.Tasks++ + if !e.OK { + rep.Failed++ + } + if e.Reward != nil && *e.Reward == 1 { + rep.Rewarded++ + } + if e.Step > 0 { + s := stepStat(steps, e.Step) + s.Tasks++ + if !e.OK { + s.Failed++ + } + if e.Reward != nil && *e.Reward == 1 { + s.Rewarded++ + } + } + case EventBarrier: + stepStat(steps, e.Step).BarrierMS = e.DurMS + stepStat(steps, e.Step).Live = e.Count + case EventPark: + stepStat(steps, e.Step).ParkMS = e.DurMS + case EventWake: + if e.OK { + wakes[e.Step] = append(wakes[e.Step], e.DurMS) + } + } + } + if rep.Tasks > 0 { + rep.RewardRate = float64(rep.Rewarded) / float64(rep.Tasks) + } + for _, k := range order { + d := durs[k] + sort.Float64s(d) + rep.Stats = append(rep.Stats, Stat{Event: k.event, Name: k.name, Count: len(d) + failed[k], Failed: failed[k], + P50MS: percentile(d, 0.50), P95MS: percentile(d, 0.95), MaxMS: percentile(d, 1), Retried: retried[k]}) + } + for step, s := range steps { + w := wakes[step] + sort.Float64s(w) + s.WakeP50MS = percentile(w, 0.5) + s.WakeP95MS = percentile(w, 0.95) + rep.Steps = append(rep.Steps, *s) + } + sort.Slice(rep.Steps, func(i, j int) bool { return rep.Steps[i].Step < rep.Steps[j].Step }) + for msg, n := range errs { + rep.Errors = append(rep.Errors, ErrorCount{Err: msg, Count: n}) + } + for inst, why := range unresolved { + rep.Unresolved = append(rep.Unresolved, Unresolved{Instance: inst, Reason: why}) + } + sort.Slice(rep.Unresolved, func(i, j int) bool { return rep.Unresolved[i].Instance < rep.Unresolved[j].Instance }) + sort.Slice(rep.Errors, func(i, j int) bool { + if rep.Errors[i].Count != rep.Errors[j].Count { + return rep.Errors[i].Count > rep.Errors[j].Count + } + return rep.Errors[i].Err < rep.Errors[j].Err + }) + return rep +} + +// verdict extracts the grader's verdict from the end of a reward step's +// output: "invalid run", "f2p A/B p2p C/D", or "reward 0" when the output +// does not end in the grader's JSON line. +func verdict(tail string) string { + i := strings.LastIndex(tail, `{"valid"`) + if i < 0 { + return "reward 0" + } + var g struct { + Valid bool `json:"valid"` + F2P string `json:"f2p"` + P2P string `json:"p2p"` + } + if err := json.Unmarshal([]byte(strings.TrimSpace(tail[i:])), &g); err != nil { + return "reward 0" + } + if !g.Valid { + return "invalid run" + } + return fmt.Sprintf("f2p %s p2p %s", g.F2P, g.P2P) +} + +// WriteExclusions writes the unresolved instances in the exclude-file format: +// one id per line followed by the reason as a comment. +func (rep *Report) WriteExclusions(w io.Writer, header string) { + fmt.Fprintf(w, "# %s\n# Generated from run %s: %d of %d tasks unresolved.\n", header, rep.RunID, len(rep.Unresolved), rep.Tasks) + for _, u := range rep.Unresolved { + fmt.Fprintf(w, "%s %s\n", u.Instance, u.Reason) + } +} + +func stepStat(m map[int]*StepStat, step int) *StepStat { + s, ok := m[step] + if !ok { + s = &StepStat{Step: step} + m[step] = s + } + return s +} + +// percentile of a sorted slice by the nearest-rank method, ceil(p*n)-1; 0 +// for an empty slice. +func percentile(sorted []float64, p float64) float64 { + n := len(sorted) + if n == 0 { + return 0 + } + if p >= 1 { + return sorted[n-1] + } + i := int(math.Ceil(p*float64(n))) - 1 + if i < 0 { + i = 0 + } + if i >= n { + i = n - 1 + } + return sorted[i] +} + +// WriteText prints the report as tables. +func (rep *Report) WriteText(w io.Writer) { + fmt.Fprintf(w, "run %s mode=%s tasks=%d failed=%d rewarded=%d reward_rate=%.3f run=%.1fs\n\n", + rep.RunID, rep.Mode, rep.Tasks, rep.Failed, rep.Rewarded, rep.RewardRate, rep.RunMS/1000) + tw := tabwriter.NewWriter(w, 0, 0, 2, ' ', 0) + fmt.Fprintln(tw, "event\tname\tcount\tfailed\tretried\tp50_ms\tp95_ms\tmax_ms") + for _, s := range rep.Stats { + fmt.Fprintf(tw, "%s\t%s\t%d\t%d\t%d\t%.0f\t%.0f\t%.0f\n", s.Event, s.Name, s.Count, s.Failed, s.Retried, s.P50MS, s.P95MS, s.MaxMS) + } + tw.Flush() + if len(rep.Steps) > 0 { + fmt.Fprintln(w) + tw = tabwriter.NewWriter(w, 0, 0, 2, ' ', 0) + fmt.Fprintln(tw, "step\tlive\ttasks\tfailed\trewarded\tbarrier_s\tpark_s\twake_p50_ms\twake_p95_ms") + for _, s := range rep.Steps { + fmt.Fprintf(tw, "%d\t%d\t%d\t%d\t%d\t%.1f\t%.1f\t%.0f\t%.0f\n", s.Step, s.Live, s.Tasks, s.Failed, s.Rewarded, + s.BarrierMS/1000, s.ParkMS/1000, s.WakeP50MS, s.WakeP95MS) + } + tw.Flush() + } + if len(rep.Unresolved) > 0 { + fmt.Fprintf(w, "\nunresolved: %d instances (report -unresolved FILE writes them in exclude-file format)\n", len(rep.Unresolved)) + } + if len(rep.Errors) > 0 { + fmt.Fprintln(w) + fmt.Fprintln(w, "errors:") + for i, e := range rep.Errors { + if i == 10 { + fmt.Fprintf(w, " ... %d more\n", len(rep.Errors)-10) + break + } + msg := strings.ReplaceAll(e.Err, "\n", " ") + if len(msg) > 160 { + msg = msg[:160] + "..." + } + fmt.Fprintf(w, " %5d %s\n", e.Count, msg) + } + } +} diff --git a/internal/benchmarking/swebench/report_test.go b/internal/benchmarking/swebench/report_test.go new file mode 100644 index 00000000..dadf87a4 --- /dev/null +++ b/internal/benchmarking/swebench/report_test.go @@ -0,0 +1,139 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package swebench + +import ( + "bytes" + "errors" + "strings" + "testing" +) + +func TestPercentile(t *testing.T) { + s := []float64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10} + if p := percentile(s, 0.5); p != 5 { + t.Errorf("p50 = %v", p) + } + if p := percentile(s, 0.95); p != 10 { + t.Errorf("p95 = %v", p) + } + if p := percentile(s, 1); p != 10 { + t.Errorf("max = %v", p) + } + if p := percentile(nil, 0.5); p != 0 { + t.Errorf("empty = %v", p) + } + thirteen := make([]float64, 13) + for i := range thirteen { + thirteen[i] = float64(i + 1) + } + if p := percentile(thirteen, 0.95); p != 13 { + t.Errorf("nearest rank p95 of 13 samples = %v, want the 13th", p) + } +} + +func TestSummarizeRoundTrip(t *testing.T) { + var buf bytes.Buffer + rec := NewRecorder(&buf) + rec.Record(Event{RunID: "r", Event: EventCreate, Env: "e1", DurMS: 100, OK: true}) + rec.Record(Event{RunID: "r", Event: EventCreate, Env: "e2", DurMS: 300, OK: false, Err: "boom"}) + rec.Record(Event{RunID: "r", Event: EventStep, Name: "run_tests", DurMS: 50, OK: true, Exit: intp(0)}) + rec.Record(Event{RunID: "r", Event: EventStep, Name: "explore", DurMS: 5, OK: true, Exit: intp(0)}) + rec.Record(Event{RunID: "r", Event: EventTask, Step: 1, DurMS: 500, OK: true, Reward: intp(1)}) + rec.Record(Event{RunID: "r", Event: EventTask, Step: 1, DurMS: 600, OK: false}) + rec.Record(Event{RunID: "r", Event: EventWake, Step: 2, DurMS: 40, OK: true, Attempts: 3}) + rec.Record(Event{RunID: "r", Event: EventStep, Name: "grade", Instance: "repo__x-1", DurMS: 9, OK: true, Reward: intp(0), + Tail: `{"valid": true, "resolved": false, "f2p": "1/1", "p2p": "3/4"}`}) + rec.Record(Event{RunID: "r", Event: EventStep, Name: "grade", Instance: "repo__x-2", DurMS: 9, OK: true, Reward: intp(0), Tail: "no json"}) + rec.Record(Event{RunID: "r", Event: EventStep, Name: "grade", Instance: "repo__x-1", DurMS: 9, OK: true, Reward: intp(0), + Tail: `{"valid": false, "resolved": false}`}) + rec.Record(Event{RunID: "r", Event: EventTask, Step: 2, DurMS: 700, OK: true, Reward: intp(0)}) + rec.Record(Event{RunID: "r", Event: EventBarrier, Step: 1, DurMS: 650, OK: true, Count: 2}) + rec.Record(Event{RunID: "r", Event: EventPark, Name: "suspend", Step: 1, DurMS: 90, OK: true, Count: 1}) + rec.Record(Event{RunID: "r", Event: EventRun, Name: "steps", DurMS: 2000, OK: false, Count: 3}) + if rec.Len() != 14 { + t.Fatalf("recorded %d", rec.Len()) + } + + rep, err := Summarize(strings.NewReader(buf.String())) + if err != nil { + t.Fatal(err) + } + if rep.RunID != "r" || rep.Mode != "steps" || rep.RunMS != 2000 { + t.Fatalf("header: %+v", rep) + } + if rep.Tasks != 3 || rep.Failed != 1 || rep.Rewarded != 1 { + t.Fatalf("tasks: %+v", rep) + } + find := func(event, name string) Stat { + for _, s := range rep.Stats { + if s.Event == event && s.Name == name { + return s + } + } + t.Fatalf("no stat %s/%s in %+v", event, name, rep.Stats) + return Stat{} + } + if c := find(EventCreate, ""); c.Count != 2 || c.Failed != 1 || c.P50MS != 100 || c.MaxMS != 100 { + t.Fatalf("create: %+v", c) + } + if s := find(EventStep, "run_tests"); s.Count != 1 || s.P50MS != 50 { + t.Fatalf("step: %+v", s) + } + if len(rep.Steps) != 2 || rep.Steps[0].BarrierMS != 650 || rep.Steps[0].ParkMS != 90 || rep.Steps[0].Live != 2 || rep.Steps[1].WakeP50MS != 40 { + t.Fatalf("steps: %+v", rep.Steps) + } + if len(rep.Errors) != 1 || rep.Errors[0].Err != "boom" || rep.Errors[0].Count != 1 { + t.Fatalf("errors: %+v", rep.Errors) + } + if w := find(EventWake, ""); w.Retried != 1 { + t.Fatalf("wake retried: %+v", w) + } + if len(rep.Unresolved) != 2 || rep.Unresolved[0].Instance != "repo__x-1" || rep.Unresolved[0].Reason != "f2p 1/1 p2p 3/4" || + rep.Unresolved[1].Reason != "reward 0" { + t.Fatalf("unresolved: %+v", rep.Unresolved) + } + var excl bytes.Buffer + rep.WriteExclusions(&excl, "test list") + if !strings.Contains(excl.String(), "# test list\n") || !strings.Contains(excl.String(), "repo__x-1 f2p 1/1 p2p 3/4\n") { + t.Fatalf("exclusions:\n%s", excl.String()) + } + if v := verdict(`{"valid": false}`); v != "invalid run" { + t.Fatalf("verdict: %q", v) + } + var text bytes.Buffer + rep.WriteText(&text) + for _, want := range []string{"run r", "create", "run_tests", "barrier_s", "boom"} { + if !strings.Contains(text.String(), want) { + t.Errorf("text report lacks %q:\n%s", want, text.String()) + } + } +} + +type failingWriter struct{} + +func (failingWriter) Write([]byte) (int, error) { return 0, errors.New("disk full") } + +func TestRecorderKeepsFirstWriteError(t *testing.T) { + rec := NewRecorder(failingWriter{}) + rec.Record(Event{RunID: "r", Event: EventCreate}) + rec.Record(Event{RunID: "r", Event: EventDelete}) + if err := rec.Err(); err == nil || err.Error() != "disk full" { + t.Fatalf("Err = %v", err) + } + if NewRecorder(nil).Err() != nil { + t.Fatal("a discarding recorder has no error") + } +} diff --git a/internal/benchmarking/swebench/run.go b/internal/benchmarking/swebench/run.go new file mode 100644 index 00000000..ebd713af --- /dev/null +++ b/internal/benchmarking/swebench/run.go @@ -0,0 +1,637 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package swebench + +import ( + "context" + "errors" + "fmt" + "hash/fnv" + "log/slog" + "regexp" + "strings" + "sync" + "time" + + env "github.com/agent-substrate/env/clients/go" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// Mode selects the workload shape. +type Mode string + +const ( + // ModeEval runs every (image, task) once: create, replay, delete. + ModeEval Mode = "eval" + // ModeSteps keeps a fleet across training steps and parks it in between. + ModeSteps Mode = "steps" +) + +// Park is what happens to the fleet between training steps. +type Park string + +const ( + ParkNone Park = "none" // environments stay running + ParkSuspend Park = "suspend" // SuspendEnvironment; the next command wakes it + ParkDelete Park = "delete" // DeleteEnvironment; the next step creates again +) + +// Config is one run. Every field that shapes the load is a parameter so the +// same binary serves the smoke test and the 500-wide run. +type Config struct { + RunID string + Backend Backend + Instances []Instance + Trajectory *Trajectory + Recorder *Recorder + Logger *slog.Logger + + Mode Mode + // Width is the most tasks in flight at once (the "500 wide"). + Width int + // TasksPerImage is the eval mode fan-out per image. + TasksPerImage int + // Rollouts and Steps shape steps mode: Instances x Rollouts environments + // driven through Steps training steps. + Rollouts int + Steps int + Park Park + + // CreateTimeout bounds one create call; StepTimeout is the default for a + // run step without its own timeout. + CreateTimeout time.Duration + StepTimeout time.Duration + // WakeBudget is how long the first command after a create or a park is + // retried on transient errors: a wake that outlasts the router's parking + // budget answers Unavailable, and a first create whose image is still + // being golden-baked answers Unknown until the bake is done. WakeBackoff + // is the initial pause between attempts; it doubles up to five times + // that. + WakeBudget time.Duration + WakeBackoff time.Duration + // Keep leaves the environments in place when the run ends. + Keep bool +} + +// Summary is what a run produced, for the exit status and the log. +type Summary struct { + Tasks int + Failed int + Rewarded int + Duration time.Duration +} + +func (c *Config) validate() error { + switch { + case c.RunID == "": + return errors.New("run id is required") + case !runIDPattern.MatchString(c.RunID): + return fmt.Errorf("run id %q must be lowercase letters, digits and '-'", c.RunID) + case len(c.RunID) > maxRunIDLen: + return fmt.Errorf("run id %q is longer than %d characters; it prefixes every environment name", c.RunID, maxRunIDLen) + case c.Backend == nil: + return errors.New("backend is required") + case len(c.Instances) == 0: + return errors.New("no instances") + case c.Trajectory == nil: + return errors.New("trajectory is required") + case c.Width < 1: + return errors.New("width must be >= 1") + } + switch c.Mode { + case ModeEval: + if c.TasksPerImage < 1 { + return errors.New("tasks per image must be >= 1") + } + case ModeSteps: + if c.Rollouts < 1 || c.Steps < 1 { + return errors.New("rollouts and steps must be >= 1") + } + switch c.Park { + case ParkNone, ParkSuspend, ParkDelete: + default: + return fmt.Errorf("unknown park mode %q", c.Park) + } + default: + return fmt.Errorf("unknown mode %q", c.Mode) + } + return nil +} + +func (c *Config) defaults() { + if c.Recorder == nil { + c.Recorder = NewRecorder(nil) + } + if c.Logger == nil { + c.Logger = slog.Default() + } + if c.CreateTimeout <= 0 { + c.CreateTimeout = 5 * time.Minute + } + if c.StepTimeout <= 0 { + c.StepTimeout = 10 * time.Minute + } + if c.WakeBudget <= 0 { + c.WakeBudget = 3 * time.Minute + } + if c.WakeBackoff <= 0 { + c.WakeBackoff = DefaultWakeBackoff + } +} + +// DefaultWakeBackoff is the initial pause between first-command retries. +const DefaultWakeBackoff = 2 * time.Second + +// runIDPattern is Substrate's name rule (a k8s short name); the run id is a +// prefix of every environment name. maxRunIDLen keeps the longest name the +// loader builds (run id, 36 characters of instance id, a replay index and a +// step suffix) under the 63-character limit. +var runIDPattern = regexp.MustCompile(`^[a-z0-9]([-a-z0-9]*[a-z0-9])?$`) + +const maxRunIDLen = 20 + +// cleanupTimeout bounds the deletes that run after a cancelled or finished +// run; they use a context that survives the cancellation so nothing leaks. +const cleanupTimeout = 10 * time.Minute + +func ms(d time.Duration) float64 { return float64(d) / float64(time.Millisecond) } + +// tail returns at most n trailing bytes of s. +func tail(s string, n int) string { + if len(s) <= n { + return s + } + return s[len(s)-n:] +} + +// environment is one sandbox and its progress. One goroutine owns it at a +// time; phases are separated by barriers, so it needs no lock. +type environment struct { + // id is the current actor name; base is the name of the first one. A + // recreate after a delete park gets a fresh name, since Substrate + // deletes asynchronously and the old name is not free immediately. + id string + base string + inst Instance + task int + traj *Trajectory + + sb Sandbox + createStart time.Time + // failed environments are skipped for the rest of a steps run. + failed bool +} + +type runner struct { + cfg *Config + rec *Recorder + log *slog.Logger + + mu sync.Mutex + tasks int + failed int + rewarded int +} + +// Run executes the configured workload and returns its summary. A non-nil +// error means the run could not be carried out; task failures are counted in +// the Summary and recorded as events, not returned. +func Run(ctx context.Context, cfg Config) (*Summary, error) { + if err := cfg.validate(); err != nil { + return nil, err + } + cfg.defaults() + r := &runner{cfg: &cfg, rec: cfg.Recorder, log: cfg.Logger} + + envs, err := r.plan() + if err != nil { + return nil, err + } + start := time.Now() + switch cfg.Mode { + case ModeEval: + r.runEval(ctx, envs) + case ModeSteps: + r.runSteps(ctx, envs) + } + dur := time.Since(start) + s := &Summary{Tasks: r.tasks, Failed: r.failed, Rewarded: r.rewarded, Duration: dur} + r.rec.Record(Event{RunID: cfg.RunID, Event: EventRun, Name: string(cfg.Mode), DurMS: ms(dur), + OK: s.Failed == 0, Count: s.Tasks}) + return s, ctx.Err() +} + +// plan renders the trajectory per instance and names every environment. +func (r *runner) plan() ([]*environment, error) { + fanout := r.cfg.TasksPerImage + if r.cfg.Mode == ModeSteps { + fanout = r.cfg.Rollouts + } + envs := make([]*environment, 0, len(r.cfg.Instances)*fanout) + for _, in := range r.cfg.Instances { + traj, err := r.cfg.Trajectory.Render(in) + if err != nil { + return nil, err + } + for k := 0; k < fanout; k++ { + id := EnvID(r.cfg.RunID, in.InstanceID, k) + envs = append(envs, &environment{ + id: id, + base: id, + inst: in, + task: k, + traj: traj, + }) + } + } + return envs, nil +} + +// EnvID derives a Substrate-valid name (lowercase alphanumerics and '-', at +// most 63 characters) from the run, the instance and the replay index. A +// truncated instance id gets a hash suffix so distinct instances stay +// distinct. +func EnvID(runID, instanceID string, k int) string { + const maxInstance = 36 + b := strings.Builder{} + for _, c := range strings.ToLower(instanceID) { + switch { + case c >= 'a' && c <= 'z', c >= '0' && c <= '9': + b.WriteRune(c) + default: + b.WriteByte('-') + } + } + inst := strings.Trim(b.String(), "-") + if len(inst) > maxInstance { + h := fnv.New32a() + _, _ = h.Write([]byte(instanceID)) + inst = fmt.Sprintf("%s-%08x", strings.Trim(inst[:maxInstance-9], "-"), h.Sum32()) + } + return fmt.Sprintf("%s-%s-%d", runID, inst, k) +} + +// forEach runs fn over every index with at most width in flight and waits +// for all of them. It stops launching when ctx is done. +func forEach(ctx context.Context, n, width int, fn func(i int)) { + sem := make(chan struct{}, width) + var wg sync.WaitGroup + for i := 0; i < n; i++ { + select { + case <-ctx.Done(): + wg.Wait() + return + case sem <- struct{}{}: + } + wg.Add(1) + go func(i int) { + defer wg.Done() + defer func() { <-sem }() + fn(i) + }(i) + } + wg.Wait() +} + +// runEval is the evaluation shape: every environment is created, replayed +// once and deleted, Width at a time. +func (r *runner) runEval(ctx context.Context, envs []*environment) { + var done int + forEach(ctx, len(envs), r.cfg.Width, func(i int) { + e := envs[i] + if r.create(ctx, e, 0) { + r.replay(ctx, e, 0) + } + if !r.cfg.Keep { + // Delete with a context that survives a cancelled run, or a + // Ctrl-C at width 500 leaves 500 environments behind. + cleanup, cancel := context.WithTimeout(context.WithoutCancel(ctx), cleanupTimeout) + r.remove(cleanup, e, 0) + cancel() + } + r.mu.Lock() + done++ + if done%100 == 0 || done == len(envs) { + r.log.Info("eval progress", slog.Int("done", done), slog.Int("total", len(envs)), + slog.Int("failed", r.failed), slog.Int("rewarded", r.rewarded)) + } + r.mu.Unlock() + }) +} + +// runSteps is the training shape: the fleet is created once, every step +// replays the trajectory on all of it behind a barrier, and the fleet is +// parked between steps. +func (r *runner) runSteps(ctx context.Context, envs []*environment) { + forEach(ctx, len(envs), r.cfg.Width, func(i int) { r.create(ctx, envs[i], 0) }) + r.log.Info("fleet created", slog.Int("environments", len(envs)), slog.Int("failed", r.countFailed(envs))) + + for step := 1; step <= r.cfg.Steps; step++ { + if ctx.Err() != nil { + break + } + if step > 1 && r.cfg.Park == ParkDelete { + forEach(ctx, len(envs), r.cfg.Width, func(i int) { + if !envs[i].failed { + r.create(ctx, envs[i], step) + } + }) + } + barrier := time.Now() + forEach(ctx, len(envs), r.cfg.Width, func(i int) { + if !envs[i].failed { + r.replay(ctx, envs[i], step) + } + }) + live := len(envs) - r.countFailed(envs) + r.rec.Record(Event{RunID: r.cfg.RunID, Event: EventBarrier, Step: step, DurMS: ms(time.Since(barrier)), + OK: ctx.Err() == nil, Count: live}) + r.log.Info("step done", slog.Int("step", step), slog.Duration("barrier", time.Since(barrier)), + slog.Int("live", live), slog.Int("failed", r.failed), slog.Int("rewarded", r.rewarded)) + + if step == r.cfg.Steps || r.cfg.Park == ParkNone || ctx.Err() != nil { + continue + } + parkStart := time.Now() + forEach(ctx, len(envs), r.cfg.Width, func(i int) { + e := envs[i] + if e.failed { + return + } + switch r.cfg.Park { + case ParkSuspend: + r.suspend(ctx, e, step) + case ParkDelete: + r.remove(ctx, e, step) + } + }) + r.rec.Record(Event{RunID: r.cfg.RunID, Event: EventPark, Name: string(r.cfg.Park), Step: step, + DurMS: ms(time.Since(parkStart)), OK: ctx.Err() == nil, Count: live}) + } + + if r.cfg.Keep { + return + } + // Tear down with a context that survives a cancelled run so nothing + // leaks on Ctrl-C. + cleanup, cancel := context.WithTimeout(context.WithoutCancel(ctx), cleanupTimeout) + defer cancel() + forEach(cleanup, len(envs), r.cfg.Width, func(i int) { + if envs[i].sb != nil { + r.remove(cleanup, envs[i], r.cfg.Steps) + } + }) +} + +func (r *runner) countFailed(envs []*environment) int { + n := 0 + for _, e := range envs { + if e.failed { + n++ + } + } + return n +} + +// create makes the environment and records the call. It returns false and +// marks the environment failed when the create did not succeed. +func (r *runner) create(ctx context.Context, e *environment, step int) bool { + if step > 1 { + e.id = fmt.Sprintf("%s-s%d", e.base, step) + } + cctx, cancel := context.WithTimeout(ctx, r.cfg.CreateTimeout) + defer cancel() + e.createStart = time.Now() + sb, err := r.cfg.Backend.Create(cctx, e.id, e.inst.Image) + ev := r.event(EventCreate, e, step, time.Since(e.createStart), err) + r.rec.Record(ev) + if err != nil { + e.failed = true + // A task that never got an environment is still a failed task, in + // the report as well as in the summary. + task := r.event(EventTask, e, step, 0, err) + r.rec.Record(task) + r.countTask(e, step, false, nil) + r.log.Warn("create failed", slog.String("env", e.id), slog.String("err", err.Error())) + // The server may still finish a create whose call failed or timed + // out on our side; delete by name so it cannot leak or collide. + cleanup, cancel := context.WithTimeout(context.WithoutCancel(ctx), cleanupTimeout) + defer cancel() + if derr := r.cfg.Backend.Delete(cleanup, e.id); derr != nil && !isNotFound(derr) { + r.log.Warn("delete after failed create", slog.String("env", e.id), slog.String("err", derr.Error())) + } + return false + } + e.sb = sb + return true +} + +// replay runs the trajectory once on e and records one task event. A fresh +// environment (its first step, or any step after a delete park) runs the +// Setup steps first; one that survived a park runs the Reset steps first. +func (r *runner) replay(ctx context.Context, e *environment, step int) { + fresh := step <= 1 || r.cfg.Park == ParkDelete + prefix := e.traj.Reset + if fresh { + prefix = e.traj.Setup + } + steps := append(append([]Step{}, prefix...), e.traj.Steps...) + // The first instruction of a task either follows a create (eval, or a + // delete park) or a park (suspend); its wall time is the TTFI or the + // wake, and it is the one that may find the environment not yet + // serving. + first := true + firstEvent := EventTTFI + firstStart := e.createStart + if step > 1 && r.cfg.Park != ParkDelete { + firstEvent = EventWake + firstStart = time.Time{} + } + + taskStart := time.Now() + ok := true + var reward *int + for _, s := range steps { + // No think time before the first command: it would land in the + // ttfi or wake measurement, and the "model" has nothing to think + // about before it has seen the environment. + if s.Think > 0 && !first { + select { + case <-ctx.Done(): + ok = false + case <-time.After(time.Duration(s.Think)): + } + if !ok { + break + } + } + stepStart := time.Now() + if firstStart.IsZero() { + firstStart = stepStart + } + res, attempts, err := r.doStep(ctx, e, s, first) + dur := time.Since(stepStart) + if first { + fe := r.event(firstEvent, e, step, time.Since(firstStart), err) + fe.Attempts = attempts + r.rec.Record(fe) + first = false + } + ev := r.event(EventStep, e, step, dur, err) + ev.Name = s.Name + if s.Run != "" && err == nil { + ev.Exit = intp(res.ExitCode) + } + switch { + case err != nil: + ok = false + case s.Reward: + reward = intp(0) + if res.ExitCode == 0 { + *reward = 1 + } + ev.Reward = reward + // The grader's verdict is the last line of a reward step's + // output; keep enough of it for the report to read it back. + ev.Tail = tail(res.Stdout+res.Stderr, 4000) + case s.ExpectExit != nil && res.ExitCode != *s.ExpectExit: + ok = false + ev.OK = false + ev.Err = fmt.Sprintf("exit %d, expected %d", res.ExitCode, *s.ExpectExit) + ev.Tail = tail(res.Stdout+res.Stderr, 2000) + } + r.rec.Record(ev) + if !ok { + r.log.Warn("step failed", slog.String("env", e.id), slog.String("step", s.Name), slog.String("err", ev.Err)) + break + } + } + if ok && reward == nil && e.traj.HasReward() { + ok = false + } + ev := r.event(EventTask, e, step, time.Since(taskStart), nil) + ev.OK = ok + ev.Reward = reward + r.rec.Record(ev) + r.countTask(e, step, ok, reward) + if !ok && r.cfg.Mode == ModeSteps { + e.failed = true + } +} + +// doStep performs one step and reports how many attempts it took. The +// first step of a task retries transient errors within WakeBudget: the +// environment may still be waking, or its image may still be baking. +func (r *runner) doStep(ctx context.Context, e *environment, s Step, first bool) (Result, int, error) { + deadline := time.Now().Add(r.cfg.WakeBudget) + wait := r.cfg.WakeBackoff + var ( + res Result + err error + ) + for attempt := 1; ; attempt++ { + if s.Write != nil { + mode, _ := s.Write.FileMode() + err = e.sb.WriteFile(ctx, s.Write.Path, []byte(s.Write.Content), mode) + } else { + timeout := time.Duration(s.Timeout) + if timeout <= 0 { + timeout = r.cfg.StepTimeout + } + sctx, cancel := context.WithTimeout(ctx, timeout+time.Minute) + res, err = e.sb.Run(sctx, s.Run, timeout) + cancel() + } + if err == nil || !first || !isTransient(err) || ctx.Err() != nil || time.Now().Add(wait).After(deadline) { + return res, attempt, err + } + select { + case <-ctx.Done(): + return res, attempt, err + case <-time.After(wait): + } + wait = min(wait*2, 5*r.cfg.WakeBackoff) + } +} + +// isTransient reports whether a first-command error is worth another try. +// Unavailable is the router giving up on a wake; Unknown, Internal and +// Aborted are what a resume that races the golden bake or a worker restart +// surfaces as. Argument and precondition errors are not retried. +// isNotFound reports whether a delete found nothing to delete. +func isNotFound(err error) bool { + if errors.Is(err, env.ErrNotFound) { + return true + } + st, ok := status.FromError(err) + return ok && st.Code() == codes.NotFound +} + +func isTransient(err error) bool { + st, ok := status.FromError(err) + if !ok { + return false + } + switch st.Code() { + case codes.Unavailable, codes.DeadlineExceeded, codes.Unknown, codes.Internal, codes.Aborted: + return true + } + return false +} + +func (r *runner) suspend(ctx context.Context, e *environment, step int) { + start := time.Now() + err := e.sb.Suspend(ctx) + r.rec.Record(r.event(EventSuspend, e, step, time.Since(start), err)) + if err != nil { + r.log.Warn("suspend failed", slog.String("env", e.id), slog.String("err", err.Error())) + } +} + +func (r *runner) remove(ctx context.Context, e *environment, step int) { + if e.sb == nil { + return + } + start := time.Now() + err := e.sb.Delete(ctx) + r.rec.Record(r.event(EventDelete, e, step, time.Since(start), err)) + if err != nil { + r.log.Warn("delete failed", slog.String("env", e.id), slog.String("err", err.Error())) + return + } + e.sb = nil +} + +func (r *runner) event(kind string, e *environment, step int, d time.Duration, err error) Event { + ev := Event{RunID: r.cfg.RunID, Event: kind, Env: e.id, Instance: e.inst.InstanceID, + Step: step, Task: e.task, DurMS: ms(d), OK: err == nil} + if err != nil { + ev.Err = err.Error() + } + return ev +} + +func (r *runner) countTask(e *environment, step int, ok bool, reward *int) { + r.mu.Lock() + defer r.mu.Unlock() + r.tasks++ + if !ok { + r.failed++ + } + if reward != nil && *reward == 1 { + r.rewarded++ + } +} diff --git a/internal/benchmarking/swebench/run_test.go b/internal/benchmarking/swebench/run_test.go new file mode 100644 index 00000000..908221d7 --- /dev/null +++ b/internal/benchmarking/swebench/run_test.go @@ -0,0 +1,588 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package swebench + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "io" + "io/fs" + "log/slog" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// fakeBackend is an in-memory Backend that records calls and can fail on +// demand. Its sandboxes answer every command with exit code 0 unless the +// command names a nonzero exit (e.g. "exit 1"). +type fakeBackend struct { + mu sync.Mutex + created map[string]int + suspended map[string]int + deleted map[string]int + deletedByID map[string]int + // onCreate, when set, runs after every successful create; tests use it + // to cancel a run at a known point. + onCreate func(id string) + commands []string + files map[string]string + failCreate map[string]bool + unavailable int32 // first-command Unavailable errors still to return + + inflight atomic.Int32 + maxInflight atomic.Int32 +} + +func newFake() *fakeBackend { + return &fakeBackend{created: map[string]int{}, suspended: map[string]int{}, deleted: map[string]int{}, + deletedByID: map[string]int{}, files: map[string]string{}, failCreate: map[string]bool{}} +} + +func (f *fakeBackend) Delete(ctx context.Context, id string) error { + f.mu.Lock() + defer f.mu.Unlock() + f.deletedByID[id]++ + if f.created[id] == 0 { + return status.Error(codes.NotFound, "no such environment") + } + return nil +} + +func (f *fakeBackend) Create(ctx context.Context, id, image string) (Sandbox, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.created[id]++ + if f.failCreate[id] { + return nil, errors.New("create refused") + } + if f.onCreate != nil { + f.onCreate(id) + } + return &fakeSandbox{b: f, id: id}, nil +} + +type fakeSandbox struct { + b *fakeBackend + id string +} + +func (s *fakeSandbox) enter() func() { + n := s.b.inflight.Add(1) + for { + m := s.b.maxInflight.Load() + if n <= m || s.b.maxInflight.CompareAndSwap(m, n) { + break + } + } + return func() { s.b.inflight.Add(-1) } +} + +func (s *fakeSandbox) Run(ctx context.Context, command string, timeout time.Duration) (Result, error) { + defer s.enter()() + if atomic.AddInt32(&s.b.unavailable, -1) >= 0 { + return Result{}, status.Error(codes.Unavailable, "waking") + } + atomic.AddInt32(&s.b.unavailable, 1) + time.Sleep(time.Millisecond) + s.b.mu.Lock() + s.b.commands = append(s.b.commands, s.id+": "+command) + s.b.mu.Unlock() + if strings.HasPrefix(command, "exit ") { + return Result{ExitCode: int(command[5] - '0'), Stderr: "failed"}, nil + } + return Result{Stdout: "ok"}, nil +} + +func (s *fakeSandbox) WriteFile(ctx context.Context, path string, content []byte, mode fs.FileMode) error { + defer s.enter()() + s.b.mu.Lock() + defer s.b.mu.Unlock() + s.b.files[s.id+":"+path] = string(content) + return nil +} + +func (s *fakeSandbox) Suspend(ctx context.Context) error { + s.b.mu.Lock() + defer s.b.mu.Unlock() + s.b.suspended[s.id]++ + return nil +} + +// Delete fails on a done context, as the real API would, so a cleanup that +// runs on a cancelled run context is visible as a leak. +func (s *fakeSandbox) Delete(ctx context.Context) error { + if err := ctx.Err(); err != nil { + return err + } + s.b.mu.Lock() + defer s.b.mu.Unlock() + s.b.deleted[s.id]++ + return nil +} + +func instances(n int) []Instance { + out := make([]Instance, n) + for i := range out { + out[i] = Instance{InstanceID: "repo__repo-" + string(rune('a'+i)), Image: "reg/img@" + goodDigest, + Patch: "PATCH", EvalScript: "EVAL"} + } + return out +} + +func testTrajectory(t *testing.T, rewardCmd string) *Trajectory { + t.Helper() + tr, err := ParseTrajectory([]byte(` +name: test +setup: + - {name: baseline, run: "git commit baseline", expect_exit: 0} +reset: + - {name: reset, run: "git checkout", expect_exit: 0} +steps: + - {name: write, write: {path: /tmp/p, content: "{{.Patch}}"}} + - {name: apply, run: "git apply", expect_exit: 0} + - {name: tests, run: "` + rewardCmd + `", reward: true} +`)) + if err != nil { + t.Fatal(err) + } + return tr +} + +func baseConfig(t *testing.T, b Backend, out *bytes.Buffer) Config { + t.Helper() + var w io.Writer + if out != nil { + w = out + } + return Config{ + RunID: "t", + Backend: b, + Instances: instances(3), + Trajectory: testTrajectory(t, "true"), + Recorder: NewRecorder(w), + Logger: slog.New(slog.NewTextHandler(&bytes.Buffer{}, nil)), + Width: 2, + WakeBackoff: time.Millisecond, + } +} + +func TestEvalCreatesReplaysDeletes(t *testing.T) { + f := newFake() + var out bytes.Buffer + cfg := baseConfig(t, f, &out) + cfg.Mode = ModeEval + cfg.TasksPerImage = 2 + + sum, err := Run(context.Background(), cfg) + if err != nil { + t.Fatal(err) + } + if sum.Tasks != 6 || sum.Failed != 0 || sum.Rewarded != 6 { + t.Fatalf("summary %+v", sum) + } + if len(f.created) != 6 || len(f.deleted) != 6 { + t.Fatalf("created %d deleted %d", len(f.created), len(f.deleted)) + } + for id, n := range f.created { + if n != 1 || f.deleted[id] != 1 { + t.Fatalf("env %s created %d deleted %d", id, n, f.deleted[id]) + } + } + if got := f.maxInflight.Load(); got > 2 { + t.Fatalf("max in flight %d > width 2", got) + } + if f.files["t-repo--repo-a-0:/tmp/p"] != "PATCH" { + t.Fatalf("patch not written: %v", f.files) + } + resets, setups := countCommands(f, "git checkout", "git commit baseline") + if resets != 0 || setups != 6 { + t.Fatalf("eval mode: %d resets (want 0), %d setups (want 6, once per environment)", resets, setups) + } + rep := Build(parseEvents(t, out.String())) + if rep.Tasks != 6 || rep.RewardRate != 1 { + t.Fatalf("report %+v", rep) + } + for _, ev := range []string{EventCreate, EventTTFI, EventTask, EventDelete, EventRun} { + if !hasStat(rep, ev) { + t.Errorf("no %s events", ev) + } + } +} + +func TestEvalCleanupSurvivesCancel(t *testing.T) { + f := newFake() + var out bytes.Buffer + cfg := baseConfig(t, f, &out) + cfg.Mode = ModeEval + cfg.TasksPerImage = 1 + cfg.Instances = instances(6) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + // Cancel the run while its first environments are in flight, so later + // creates never start and the in-flight ones must still be deleted. + var once sync.Once + f.onCreate = func(string) { once.Do(cancel) } + if _, err := Run(ctx, cfg); err == nil { + t.Fatal("a cancelled run must report the cancellation") + } + f.mu.Lock() + defer f.mu.Unlock() + if len(f.created) == 0 { + t.Fatal("nothing was created before the cancel; the test proves nothing") + } + for id := range f.created { + if f.deleted[id] != 1 { + t.Fatalf("environment %s leaked after cancel: deleted %d", id, f.deleted[id]) + } + } +} + +func TestFailedCreateIsDeletedByName(t *testing.T) { + f := newFake() + bad := EnvID("t", "repo__repo-a", 0) + f.failCreate[bad] = true + var out bytes.Buffer + cfg := baseConfig(t, f, &out) + cfg.Mode = ModeEval + cfg.TasksPerImage = 1 + if _, err := Run(context.Background(), cfg); err != nil { + t.Fatal(err) + } + if f.deletedByID[bad] != 1 { + t.Fatalf("the failed create must be deleted by name once, got %d", f.deletedByID[bad]) + } + for id, n := range f.deletedByID { + if id != bad && n != 0 { + t.Fatalf("%s deleted by name although its create succeeded", id) + } + } +} + +func TestNoThinkBeforeFirstCommand(t *testing.T) { + f := newFake() + var out bytes.Buffer + cfg := baseConfig(t, f, &out) + cfg.Mode = ModeEval + cfg.TasksPerImage = 1 + cfg.Instances = instances(1) + tr, err := ParseTrajectory([]byte("name: t\nsteps:\n - {name: a, run: \"true\", think: 5s, reward: true}\n")) + if err != nil { + t.Fatal(err) + } + cfg.Trajectory = tr + start := time.Now() + if _, err := Run(context.Background(), cfg); err != nil { + t.Fatal(err) + } + if time.Since(start) > time.Second { + t.Fatal("think time before the first command was not skipped") + } +} + +func TestEvalCountsFailures(t *testing.T) { + f := newFake() + f.failCreate[EnvID("t", "repo__repo-a", 0)] = true + var out bytes.Buffer + cfg := baseConfig(t, f, &out) + cfg.Mode = ModeEval + cfg.TasksPerImage = 1 + cfg.Trajectory = testTrajectory(t, "exit 1") + + sum, err := Run(context.Background(), cfg) + if err != nil { + t.Fatal(err) + } + // One create failure; the other two run to the reward step and get 0. + if sum.Tasks != 3 || sum.Failed != 1 || sum.Rewarded != 0 { + t.Fatalf("summary %+v", sum) + } + rep := Build(parseEvents(t, out.String())) + if rep.RewardRate != 0 || rep.Failed != 1 { + t.Fatalf("report %+v", rep) + } +} + +func TestExpectExitFailsTask(t *testing.T) { + f := newFake() + var out bytes.Buffer + cfg := baseConfig(t, f, &out) + cfg.Mode = ModeEval + cfg.TasksPerImage = 1 + tr, err := ParseTrajectory([]byte("name: t\nsteps:\n - {name: a, run: \"exit 2\", expect_exit: 0}\n - {name: b, run: \"true\"}\n")) + if err != nil { + t.Fatal(err) + } + cfg.Trajectory = tr + sum, err := Run(context.Background(), cfg) + if err != nil { + t.Fatal(err) + } + if sum.Failed != 3 { + t.Fatalf("summary %+v", sum) + } + for _, c := range f.commands { + if strings.HasSuffix(c, ": true") { + t.Fatal("steps after a failed expectation must not run") + } + } +} + +func TestStepsSuspendPark(t *testing.T) { + f := newFake() + var out bytes.Buffer + cfg := baseConfig(t, f, &out) + cfg.Mode = ModeSteps + cfg.Rollouts = 2 + cfg.Steps = 3 + cfg.Park = ParkSuspend + + sum, err := Run(context.Background(), cfg) + if err != nil { + t.Fatal(err) + } + if sum.Tasks != 18 || sum.Failed != 0 || sum.Rewarded != 18 { + t.Fatalf("summary %+v", sum) + } + if len(f.created) != 6 { + t.Fatalf("created %d envs, want 6 created once", len(f.created)) + } + for id, n := range f.created { + if n != 1 || f.suspended[id] != 2 || f.deleted[id] != 1 { + t.Fatalf("env %s created %d suspended %d deleted %d", id, n, f.suspended[id], f.deleted[id]) + } + } + resets, setups := countCommands(f, "git checkout", "git commit baseline") + if resets != 12 || setups != 6 { + t.Fatalf("suspend park: %d resets (want 6 envs x 2 later steps), %d setups (want 6, once per environment)", resets, setups) + } + rep := Build(parseEvents(t, out.String())) + if len(rep.Steps) != 3 || rep.Steps[0].Tasks != 6 || rep.Steps[0].ParkMS == 0 || rep.Steps[2].ParkMS != 0 { + t.Fatalf("steps %+v", rep.Steps) + } + if s := stat(rep, EventWake); s.Count != 12 { + t.Fatalf("wake events %+v, want 12 (6 envs x steps 2 and 3)", s) + } + if s := stat(rep, EventTTFI); s.Count != 6 { + t.Fatalf("ttfi events %+v, want 6", s) + } +} + +func TestStepsDeleteParkRecreates(t *testing.T) { + f := newFake() + var out bytes.Buffer + cfg := baseConfig(t, f, &out) + cfg.Mode = ModeSteps + cfg.Rollouts = 1 + cfg.Steps = 2 + cfg.Park = ParkDelete + + if _, err := Run(context.Background(), cfg); err != nil { + t.Fatal(err) + } + // Each step creates under a fresh name: 3 environments x 2 steps. + if len(f.created) != 6 { + t.Fatalf("created %d distinct names, want 6: %v", len(f.created), f.created) + } + fresh := 0 + for id, n := range f.created { + if n != 1 || f.deleted[id] != 1 || f.suspended[id] != 0 { + t.Fatalf("env %s created %d deleted %d suspended %d", id, n, f.deleted[id], f.suspended[id]) + } + if strings.HasSuffix(id, "-s2") { + fresh++ + } + } + if fresh != 3 { + t.Fatalf("%d step-2 names, want 3", fresh) + } + rep := Build(parseEvents(t, out.String())) + if s := stat(rep, EventTTFI); s.Count != 6 { + t.Fatalf("ttfi events %+v, want 3 envs x 2 creates", s) + } + if hasStat(rep, EventWake) { + t.Fatal("delete park has no wake events") + } + resets, setups := countCommands(f, "git checkout", "git commit baseline") + if resets != 0 || setups != 6 { + t.Fatalf("delete park: %d resets (want 0, every environment is new), %d setups (want 3 envs x 2 creates)", resets, setups) + } +} + +// countCommands counts the fake's recorded commands containing each needle. +func countCommands(f *fakeBackend, resetNeedle, setupNeedle string) (resets, setups int) { + f.mu.Lock() + defer f.mu.Unlock() + for _, c := range f.commands { + if strings.Contains(c, resetNeedle) { + resets++ + } + if strings.Contains(c, setupNeedle) { + setups++ + } + } + return resets, setups +} + +func TestStepsKeepLeavesFleet(t *testing.T) { + f := newFake() + var out bytes.Buffer + cfg := baseConfig(t, f, &out) + cfg.Mode = ModeSteps + cfg.Rollouts = 1 + cfg.Steps = 1 + cfg.Park = ParkNone + cfg.Keep = true + if _, err := Run(context.Background(), cfg); err != nil { + t.Fatal(err) + } + if len(f.deleted) != 0 || len(f.suspended) != 0 { + t.Fatalf("keep: deleted %v suspended %v", f.deleted, f.suspended) + } +} + +func TestFirstCommandRetriesUnavailable(t *testing.T) { + f := newFake() + f.unavailable = 2 + var out bytes.Buffer + cfg := baseConfig(t, f, &out) + cfg.Mode = ModeEval + cfg.TasksPerImage = 1 + cfg.Instances = instances(1) + cfg.WakeBudget = time.Second + tr, err := ParseTrajectory([]byte("name: t\nsteps:\n - {name: a, run: \"true\", reward: true}\n")) + if err != nil { + t.Fatal(err) + } + cfg.Trajectory = tr + sum, err := Run(context.Background(), cfg) + if err != nil { + t.Fatal(err) + } + if sum.Failed != 0 { + t.Fatalf("summary %+v", sum) + } + + rep := Build(parseEvents(t, out.String())) + if s := stat(rep, EventTTFI); s.Count != 1 { + t.Fatalf("ttfi %+v", s) + } + for _, e := range parseEvents(t, out.String()) { + if e.Event == EventTTFI && e.Attempts != 3 { + t.Fatalf("ttfi attempts = %d, want 3", e.Attempts) + } + } + + f = newFake() + f.unavailable = 1000 + cfg.Backend = f + cfg.WakeBudget = 5 * time.Millisecond + cfg.WakeBackoff = 2 * time.Millisecond + sum, err = Run(context.Background(), cfg) + if err != nil { + t.Fatal(err) + } + if sum.Failed != 1 { + t.Fatalf("budget exhausted must fail the task: %+v", sum) + } +} + +func TestIsTransient(t *testing.T) { + for code, want := range map[codes.Code]bool{ + codes.Unavailable: true, codes.Unknown: true, codes.Internal: true, codes.Aborted: true, codes.DeadlineExceeded: true, + codes.InvalidArgument: false, codes.NotFound: false, codes.FailedPrecondition: false, codes.AlreadyExists: false, + } { + if got := isTransient(status.Error(code, "x")); got != want { + t.Errorf("%v: transient = %v, want %v", code, got, want) + } + } + if isTransient(errors.New("plain")) { + t.Error("a non-status error is not transient") + } +} + +func TestConfigValidation(t *testing.T) { + f := newFake() + good := baseConfig(t, f, nil) + good.Mode = ModeEval + good.TasksPerImage = 1 + bad := []func(c *Config){ + func(c *Config) { c.RunID = "" }, + func(c *Config) { c.RunID = "Swe-1" }, + func(c *Config) { c.RunID = "swe_1" }, + func(c *Config) { c.RunID = strings.Repeat("a", 21) }, + func(c *Config) { c.Backend = nil }, + func(c *Config) { c.Instances = nil }, + func(c *Config) { c.Trajectory = nil }, + func(c *Config) { c.Width = 0 }, + func(c *Config) { c.TasksPerImage = 0 }, + func(c *Config) { c.Mode = "bogus" }, + func(c *Config) { c.Mode = ModeSteps; c.Rollouts = 0; c.Steps = 1 }, + func(c *Config) { c.Mode = ModeSteps; c.Rollouts = 1; c.Steps = 1; c.Park = "bogus" }, + } + for i, mutate := range bad { + c := good + mutate(&c) + if _, err := Run(context.Background(), c); err == nil { + t.Errorf("case %d: expected a validation error", i) + } + } +} + +func TestEnvID(t *testing.T) { + if got := EnvID("swe-1", "astropy__astropy-7336", 2); got != "swe-1-astropy--astropy-7336-2" { + t.Fatalf("got %q", got) + } + long := EnvID("r", strings.Repeat("x", 80)+"__y", 0) + if len(long) > 63 { + t.Fatalf("too long: %q", long) + } + if long == EnvID("r", strings.Repeat("x", 80)+"__z", 0) { + t.Fatal("truncated ids must stay distinct") + } +} + +func parseEvents(t *testing.T, jsonl string) []Event { + t.Helper() + var events []Event + for _, line := range strings.Split(strings.TrimSpace(jsonl), "\n") { + var e Event + if err := json.Unmarshal([]byte(line), &e); err != nil { + t.Fatal(err) + } + events = append(events, e) + } + return events +} + +func stat(rep *Report, event string) Stat { + for _, s := range rep.Stats { + if s.Event == event { + return s + } + } + return Stat{} +} + +func hasStat(rep *Report, event string) bool { + return stat(rep, event).Count > 0 +} diff --git a/internal/benchmarking/swebench/trajectories/swebench-gold.yaml b/internal/benchmarking/swebench/trajectories/swebench-gold.yaml new file mode 100644 index 00000000..c6b6102a --- /dev/null +++ b/internal/benchmarking/swebench/trajectories/swebench-gold.yaml @@ -0,0 +1,86 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# The default trajectory: what a SWE agent does on a SWE-bench task, with the +# gold patch standing in for the model's edit so the reward is known in +# advance. A healthy run rewards every task; a drop in reward rate is a +# correctness regression, not a model result. +# +# /testbed and /opt/miniconda3/envs/testbed are where SWE-bench task images +# keep the repository and its interpreter. Think times are the simulated +# policy latency between commands. +name: swebench-gold +# The image's working tree is the state a rollout starts from, and it is not +# the git HEAD: SWE-bench edits files when it builds the image (sphinx's +# tox.ini gains the per-test output flag the grader reads) and editable +# installs leave metadata in untracked directories. Setup commits that state +# as a local baseline once per environment; reset returns to it. +setup: + - name: baseline + run: >- + cd /testbed && + git -c user.email=bench@localhost -c user.name=bench add -A && + git -c user.email=bench@localhost -c user.name=bench commit -q --allow-empty -m "benchmark baseline" + expect_exit: 0 +reset: + - name: reset_repo + run: cd /testbed && git reset -q --hard && git clean -qfd + expect_exit: 0 +steps: + - name: explore + run: cd /testbed && git status --short | head -n 20; ls + think: 1s + expect_exit: 0 + - name: read_files + run: cd /testbed && git diff --stat HEAD && git log -1 --format=%H + think: 2s + expect_exit: 0 + - name: write_patch + write: + path: /tmp/gold.diff + content: "{{.Patch}}" + - name: apply_patch + run: cd /testbed && git apply -v /tmp/gold.diff + think: 3s + expect_exit: 0 + - name: write_eval + write: + path: /tmp/eval.sh + content: "{{.EvalScript}}" + mode: "0755" + # The harness script's exit code is not the verdict: it runs without -e, + # ends with a git checkout that fails when the test patch added files, and + # the test command itself exits nonzero when unrelated tests in the same + # files fail. The verdict is what the SWE-bench harness computes: parse the + # log with the repo's parser and check the FAIL_TO_PASS and PASS_TO_PASS + # tests. swebench-grade.py does that inside the sandbox from the instance + # record; its exit code is the reward. + - name: run_tests + run: >- + /bin/bash /tmp/eval.sh > /tmp/eval.log 2>&1; + tail -c 1500 /tmp/eval.log; + grep -o '>>>>> Test Exit Code: [0-9]*' /tmp/eval.log | tail -n 1 + think: 1s + timeout: 20m + - name: write_grader + write: + path: /tmp/grade.py + file: swebench-grade.py + - name: write_instance + write: + path: /tmp/instance.json + content: "{{json .}}" + - name: grade + run: /opt/miniconda3/bin/python /tmp/grade.py /tmp/instance.json /tmp/eval.log + reward: true diff --git a/internal/benchmarking/swebench/trajectories/swebench-grade.py b/internal/benchmarking/swebench/trajectories/swebench-grade.py new file mode 100644 index 00000000..e5d7db53 --- /dev/null +++ b/internal/benchmarking/swebench/trajectories/swebench-grade.py @@ -0,0 +1,317 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Grade a SWE-bench eval log the way the SWE-bench harness does. + + python swebench-grade.py INSTANCE.json EVAL.log + +INSTANCE.json is the dataset record (log_parser, eval_type, fail_to_pass, +pass_to_pass). The log is parsed with the repo's parser into a test -> status +map; the instance is resolved when every FAIL_TO_PASS test passed and every +PASS_TO_PASS test passed or was skipped. Exit 0 when resolved, 1 otherwise, +2 when the log shows the run itself was invalid (patch did not apply, tests +errored or timed out). One JSON line with the counts goes to stdout. + +Standard library only: it runs inside the task image with whatever Python +the image carries. +""" + +import json +import re +import sys + +PASSED, FAILED, SKIPPED, ERROR, XFAIL = "PASSED", "FAILED", "SKIPPED", "ERROR", "XFAIL" +STATUSES = (FAILED, PASSED, SKIPPED, ERROR, XFAIL) +PASSING = {PASSED, XFAIL} + +START_OUTPUT = ">>>>> Start Test Output" +END_OUTPUT = ">>>>> End Test Output" +BAD_MARKERS = (">>>>> Patch Apply Failed", ">>>>> Reset Failed", ">>>>> Tests Errored", ">>>>> Tests Timed Out") +EXIT_CODE = re.compile(r">>>>> Test Exit Code:\s*(-?\d+)") +# Evidence that a suite ran at all, for logs whose parser records only +# failures; mirrors the harness's SUITE_RAN. +SUITE_RAN = re.compile( + r"Executed [1-9]\d* of \d+" + r"|TOTAL: [1-9]\d* (?:SUCCESS|FAILED)" + r"|[1-9]\d* passing" + r"|Tests:\s+[1-9]\d*" + r"|Test Suites:\s+(?:\d+ \w+, )*[1-9]\d* total" + r"|^# tests [1-9]\d*" + r"|[1-9]\d* specs?, \d+ failures?" + r"|': ok$", + re.M, +) +SKIP_SUMMARY = re.compile(r"^\[\d+\]$") +CONTROL = str.maketrans("", "", "".join(chr(c) for c in range(1, 32))) + + +def _status_line(line, status_map, rewrite=None): + """pytest -rA short summary: ' [ - reason]'.""" + if not line.startswith(STATUSES): + return + if line.startswith(FAILED): + line = line.replace(" - ", " ") + parts = line.split() + if len(parts) < 2: + return + status, name = parts[0], parts[1] + if status == SKIPPED and SKIP_SUMMARY.match(name): + return + if rewrite: + name = rewrite(name) + status_map[name] = status + + +def parse_pytest(log): + status_map = {} + for line in log.split("\n"): + _status_line(line, status_map) + return status_map + + +_OPTION = re.compile(r"(.*?)\[(.*)\]") + + +def parse_pytest_options(log): + def rewrite(name): + m = _OPTION.search(name) + if not m: + return name + main, option = m.groups() + if option.startswith("/") and not option.startswith("//") and "*" not in option: + option = "/" + option.split("/")[-1] + return f"{main}[{option}]" + + status_map = {} + for line in log.split("\n"): + _status_line(line, status_map, rewrite) + return status_map + + +def parse_pytest_v2(log): + status_map = {} + for line in log.split("\n"): + line = re.sub(r"\[(\d+)m", "", line).translate(CONTROL) + if line.startswith(STATUSES): + if line.startswith(FAILED): + line = line.split(" - ", 1)[0] + parts = line.split() + if len(parts) >= 2 and not (parts[0] == SKIPPED and SKIP_SUMMARY.match(parts[1])): + status_map[" ".join(parts[1:])] = parts[0] + elif line.endswith(STATUSES): + parts = line.split() + if len(parts) >= 2: + status_map[" ".join(parts[:-1])] = parts[-1] + return status_map + + +def parse_matplotlib(log): + log = log.replace("MouseButton.LEFT", "1").replace("MouseButton.RIGHT", "3") + return parse_pytest(log) + + +def parse_seaborn(log): + status_map = {} + for line in log.split("\n"): + parts = line.split() + if len(parts) < 2: + continue + if line.startswith(FAILED): + status_map[parts[1]] = FAILED + elif f" {PASSED} " in line and parts[1] == PASSED: + status_map[parts[0]] = PASSED + elif line.startswith(PASSED): + status_map[parts[1]] = PASSED + return status_map + + +def parse_sympy(log): + status_map = {} + for m in re.findall(r"(_*) (.*)\.py:(.*) (_*)", log): + status_map[f"{m[1]}.py:{m[2]}"] = FAILED + for line in log.split("\n"): + line = line.strip() + if not line.startswith("test_"): + continue + name = line.split()[0] + if line.endswith(" E"): + status_map[name] = ERROR + elif line.endswith(" F"): + status_map[name] = FAILED + elif line.endswith(" ok"): + status_map[name] = PASSED + return status_map + + +def parse_django(log): + status_map = {} + prev_test = None + for raw in log.split("\n"): + line = raw.strip() + if "--version is equivalent to version" in line: + status_map["--version is equivalent to version"] = PASSED + if " ... " in line: + prev_test = line.split(" ... ")[0] + for suffix in (" ... ok", " ... OK", " ... OK"): + if line.endswith(suffix): + if line.startswith("Applying sites.0002_alter_domain_unique...test_no_migrations"): + line = line.split("...", 1)[-1].strip() + status_map[line.rsplit(suffix, 1)[0]] = PASSED + break + if " ... skipped" in line: + status_map[line.split(" ... skipped")[0]] = SKIPPED + if line.endswith(" ... FAIL"): + status_map[line.split(" ... FAIL")[0]] = FAILED + if line.startswith("FAIL:"): + status_map[line.split()[1].strip()] = FAILED + if line.endswith(" ... ERROR"): + status_map[line.split(" ... ERROR")[0]] = ERROR + if line.startswith("ERROR:"): + status_map[line.split()[1].strip()] = ERROR + if line.startswith("ok") and prev_test is not None: + status_map[prev_test] = PASSED + for pattern in ( + r"^(.*?)\s\.\.\.\sTesting\ against\ Django\ installed\ in\ ((?s:.*?))\ silenced\)\.\nok$", + r"^(.*?)\s\.\.\.\sInternal\ Server\ Error:\ \/(.*)\/\nok$", + r"^(.*?)\s\.\.\.\sSystem check identified no issues \(0 silenced\)\nok$", + ): + for m in re.finditer(pattern, log, re.MULTILINE): + status_map[m.group(1)] = PASSED + return status_map + + +PARSERS = { + "parse_log_pytest": parse_pytest, + "parse_log_astroid": parse_pytest, + "parse_log_flask": parse_pytest, + "parse_log_marshmallow": parse_pytest, + "parse_log_pvlib": parse_pytest, + "parse_log_pyvista": parse_pytest, + "parse_log_sqlfluff": parse_pytest, + "parse_log_xarray": parse_pytest, + "parse_log_pytest_options": parse_pytest_options, + "parse_log_pydicom": parse_pytest_options, + "parse_log_requests": parse_pytest_options, + "parse_log_pylint": parse_pytest_options, + "parse_log_pytest_v2": parse_pytest_v2, + "parse_log_astropy": parse_pytest_v2, + "parse_log_scikit": parse_pytest_v2, + "parse_log_sphinx": parse_pytest_v2, + "parse_log_matplotlib": parse_matplotlib, + "parse_log_seaborn": parse_seaborn, + "parse_log_sympy": parse_sympy, + "parse_log_django": parse_django, +} + + +def resolve_case(case, status_map): + """Exact match, or a prefix match for ids truncated mid-parameter when the + candidates agree on the outcome (676 Verified ids are truncated).""" + if case in status_map: + return case + if case.count("[") > case.count("]"): + matches = [k for k in status_map if k.startswith(case)] + if matches and len({status_map[k] in PASSING for k in matches}) == 1: + return matches[0] + return None + + +def passed(case, status_map): + key = resolve_case(case, status_map) + return key is not None and status_map[key] in PASSING + + +def maintained(case, status_map): + key = resolve_case(case, status_map) + return passed(case, status_map) or (key is not None and status_map[key] == SKIPPED) + + +def failed(case, status_map): + """Missing, FAILED, ERROR or SKIPPED, as the harness's test_failed.""" + key = resolve_case(case, status_map) + return key is None or status_map[key] in (FAILED, ERROR, SKIPPED) + + +def status_map_from_log(parser, content): + """The harness's validity rules, then the parser over the marked region.""" + if any(marker in content for marker in BAD_MARKERS): + return None + start, end = content.find(START_OUTPUT), content.find(END_OUTPUT) + if start < 0 or end < 0 or end < start: + return None + sliced = content[start + len(START_OUTPUT):end] + status_map = parser(sliced) or parser(content) + if not status_map and not SUITE_RAN.search(content): + return None + m = EXIT_CODE.search(content) + exit_code = int(m.group(1)) if m else None + if exit_code not in (None, 0) and status_map and not any(s in (FAILED, ERROR) for s in status_map.values()): + return None + return status_map + + +def grade(instance, content): + parser = PARSERS.get(instance.get("log_parser", "")) + if parser is None: + raise SystemExit(f"unknown log_parser {instance.get('log_parser')!r}") + status_map = status_map_from_log(parser, content) + if status_map is None: + return {"valid": False, "resolved": False} + fail_only = instance.get("eval_type", "pass_and_fail").lower() == "fail_only" + f2p = instance.get("fail_to_pass", []) + p2p = instance.get("pass_to_pass", []) + + # As in the harness, every test lands in a success list, a failure list, + # or neither (an unknown status counts for nothing); the instance is + # resolved when both failure lists are empty. + def split(tests, ok, bad): + success = [t for t in tests if ok(t)] + failure = [t for t in tests if not ok(t) and bad(t)] + return success, failure + + if fail_only: + bad = lambda t: status_map.get(t) == FAILED # noqa: E731 + f2p_ok, f2p_bad = split(f2p, lambda t: not bad(t), bad) + p2p_ok, p2p_bad = split(p2p, lambda t: not bad(t), bad) + else: + f2p_ok, f2p_bad = split(f2p, lambda t: passed(t, status_map), lambda t: failed(t, status_map)) + p2p_ok, p2p_bad = split(p2p, lambda t: maintained(t, status_map), lambda t: failed(t, status_map)) + resolved = not f2p_bad and not p2p_bad + return { + "valid": True, + "resolved": resolved, + "f2p": f"{len(f2p_ok)}/{len(f2p)}", + "p2p": f"{len(p2p_ok)}/{len(p2p)}", + "f2p_failed": f2p_bad[:5], + "p2p_failed": p2p_bad[:5], + } + + +def main(argv): + if len(argv) != 3: + print(__doc__, file=sys.stderr) + return 2 + with open(argv[1]) as f: + instance = json.load(f) + with open(argv[2], errors="replace") as f: + content = f.read() + report = grade(instance, content) + print(json.dumps(report)) + if not report["valid"]: + return 2 + return 0 if report["resolved"] else 1 + + +if __name__ == "__main__": + sys.exit(main(sys.argv)) diff --git a/internal/benchmarking/swebench/trajectories/test_swebench_grade.py b/internal/benchmarking/swebench/trajectories/test_swebench_grade.py new file mode 100644 index 00000000..7e906af2 --- /dev/null +++ b/internal/benchmarking/swebench/trajectories/test_swebench_grade.py @@ -0,0 +1,154 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Tests for the in-sandbox grader next to it: + + python3 -m pytest internal/benchmarking/swebench/trajectories -q +""" + +import importlib.util +import pathlib + +GRADER = pathlib.Path(__file__).resolve().with_name("swebench-grade.py") + +spec = importlib.util.spec_from_file_location("grader", GRADER) +grader = importlib.util.module_from_spec(spec) +spec.loader.exec_module(grader) + + +def wrap(body, exit_code=0): + return f"+ : '>>>>> Start Test Output'\n{body}\n+ : '>>>>> End Test Output'\n>>>>> Test Exit Code: {exit_code}\n" + + +PYTEST_LOG = """\ +============================= test session starts ============================== +PASSED astropy/units/tests/test_a.py::test_one +PASSED astropy/units/tests/test_a.py::test_param[log(photon s-1)] +FAILED astropy/units/tests/test_a.py::test_unrelated - AssertionError: boom +XFAIL astropy/units/tests/test_a.py::test_known +SKIPPED [3] astropy/units/tests/test_a.py:12: no network +========================= 1 failed, 2 passed in 0.42s ========================== +""" + + +def test_pytest_v2_resolves_despite_unrelated_failure(): + inst = { + "log_parser": "parse_log_astropy", + "eval_type": "pass_and_fail", + "fail_to_pass": ["astropy/units/tests/test_a.py::test_one"], + "pass_to_pass": ["astropy/units/tests/test_a.py::test_known"], + } + rep = grader.grade(inst, wrap(PYTEST_LOG, exit_code=1)) + assert rep["valid"] and rep["resolved"], rep + + +def test_pytest_truncated_parametrized_id_prefix_matches(): + inst = { + "log_parser": "parse_log_pytest", + "fail_to_pass": ["astropy/units/tests/test_a.py::test_param[log(photon"], + "pass_to_pass": [], + } + assert grader.grade(inst, wrap(PYTEST_LOG))["resolved"] + + +def test_f2p_failed_is_not_resolved(): + inst = { + "log_parser": "parse_log_pytest", + "fail_to_pass": ["astropy/units/tests/test_a.py::test_unrelated"], + "pass_to_pass": [], + } + rep = grader.grade(inst, wrap(PYTEST_LOG, exit_code=1)) + assert rep["valid"] and not rep["resolved"] and rep["f2p"] == "0/1" + + +def test_skipped_f2p_is_not_resolved_but_skipped_p2p_is_maintained(): + log = "SKIPPED tests/test_x.py::test_a\nPASSED tests/test_x.py::test_b\n" + inst = {"log_parser": "parse_log_pytest", "fail_to_pass": ["tests/test_x.py::test_a"], "pass_to_pass": []} + assert not grader.grade(inst, wrap(log))["resolved"] + inst = {"log_parser": "parse_log_pytest", "fail_to_pass": ["tests/test_x.py::test_b"], "pass_to_pass": ["tests/test_x.py::test_a"]} + assert grader.grade(inst, wrap(log))["resolved"] + + +DJANGO_LOG = """\ +test_loader (migrations.test_loader.LoaderTests) ... ok +test_namespace (migrations.test_loader.LoaderTests) +Some docstring here ... ok +test_broken (migrations.test_loader.LoaderTests) ... FAIL +test_skipped (migrations.test_loader.LoaderTests) ... skipped 'no db' +test_err (migrations.test_loader.LoaderTests) ... ERROR +---------------------------------------------------------------------- +Ran 5 tests in 0.1s +FAILED (failures=1, errors=1) +""" + + +def test_django_parser_statuses(): + sm = grader.parse_django(DJANGO_LOG) + assert sm["test_loader (migrations.test_loader.LoaderTests)"] == "PASSED" + # A docstring on the next line is what the harness records; the port keeps + # that behavior so verdicts match the harness, brittle as it is. + assert sm["Some docstring here"] == "PASSED" + assert sm["test_broken (migrations.test_loader.LoaderTests)"] == "FAILED" + assert sm["test_skipped (migrations.test_loader.LoaderTests)"] == "SKIPPED" + assert sm["test_err (migrations.test_loader.LoaderTests)"] == "ERROR" + + +def test_django_resolution(): + inst = { + "log_parser": "parse_log_django", + "fail_to_pass": ["test_loader (migrations.test_loader.LoaderTests)"], + "pass_to_pass": ["test_skipped (migrations.test_loader.LoaderTests)"], + } + assert grader.grade(inst, wrap(DJANGO_LOG, exit_code=1))["resolved"] + + +def test_sympy_parser(): + log = "test_one ok\ntest_two F\ntest_three E\n________ sympy/core/tests/test_x.py:test_four ________\n" + sm = grader.parse_sympy(log) + assert sm == {"test_one": "PASSED", "test_two": "FAILED", "test_three": "ERROR", + "sympy/core/tests/test_x.py:test_four": "FAILED"} + + +def test_pytest_options_normalizes_paths(): + log = "PASSED tests/test_a.py::test_file[/tmp/xyz/data.txt]\n" + sm = grader.parse_pytest_options(log) + assert sm == {"tests/test_a.py::test_file[/data.txt]": "PASSED"} + + +def test_invalid_runs(): + inst = {"log_parser": "parse_log_pytest", "fail_to_pass": ["t"], "pass_to_pass": []} + assert not grader.grade(inst, ">>>>> Patch Apply Failed\n" + wrap("PASSED t"))["valid"] + assert not grader.grade(inst, "PASSED t\n")["valid"] + # Nonzero exit with a log that shows no failure means the log is not the run. + assert not grader.grade(inst, wrap("PASSED t", exit_code=1))["valid"] + # Markers in the wrong order, or no parsed result and no sign the suite ran. + assert not grader.grade(inst, "+ : '>>>>> End Test Output'\nPASSED t\n+ : '>>>>> Start Test Output'\n")["valid"] + assert not grader.grade(inst, wrap("nothing here"))["valid"] + assert grader.grade(inst, wrap("Executed 3 of 3 specs"))["valid"] + + +def test_unknown_status_counts_for_nothing(): + # A recorded status outside the known set is neither a pass nor a failure + # in the harness, so it must not block resolution. + log = "PASSEDX tests/test_x.py::test_a\nPASSED tests/test_x.py::test_b\n" + inst = {"log_parser": "parse_log_pytest", "fail_to_pass": ["tests/test_x.py::test_b"], + "pass_to_pass": ["tests/test_x.py::test_a"]} + rep = grader.grade(inst, wrap(log)) + assert rep["resolved"] and rep["p2p"] == "0/1" and rep["p2p_failed"] == [], rep + + +def test_unknown_parser_exits(): + import pytest + with pytest.raises(SystemExit): + grader.grade({"log_parser": "nope"}, wrap("")) diff --git a/internal/benchmarking/swebench/trajectory.go b/internal/benchmarking/swebench/trajectory.go new file mode 100644 index 00000000..80227089 --- /dev/null +++ b/internal/benchmarking/swebench/trajectory.go @@ -0,0 +1,332 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package swebench + +import ( + "bytes" + "embed" + "encoding/json" + "errors" + "fmt" + "io/fs" + "os" + "path/filepath" + "strconv" + "strings" + "text/template" + "time" + + "sigs.k8s.io/yaml" +) + +//go:embed trajectories/* +var trajectoryFS embed.FS + +// DefaultTrajectory is the built-in trajectory used when none is given. +const DefaultTrajectory = "swebench-gold" + +// Duration is a time.Duration that decodes from a string such as "1s". +type Duration time.Duration + +// UnmarshalJSON accepts a quoted Go duration string. +func (d *Duration) UnmarshalJSON(b []byte) error { + s, err := strconv.Unquote(string(b)) + if err != nil { + return fmt.Errorf("duration must be a quoted string like \"500ms\": %s", string(b)) + } + v, err := time.ParseDuration(s) + if err != nil { + return err + } + *d = Duration(v) + return nil +} + +// MarshalJSON writes the duration back as a string. +func (d Duration) MarshalJSON() ([]byte, error) { + return []byte(strconv.Quote(time.Duration(d).String())), nil +} + +// WriteSpec puts a file into the sandbox. Content is a template over the +// Instance, so a step can write the gold patch or the eval script. +type WriteSpec struct { + Path string `json:"path"` + Content string `json:"content"` + // File names a file next to the trajectory whose contents become + // Content when the trajectory is loaded: a built-in asset for built-in + // trajectories, a path relative to the YAML file otherwise. File + // contents are written as they are; only inline Content is a template. + File string `json:"file,omitempty"` + // Mode is an octal file mode string; empty means 0644. + Mode string `json:"mode,omitempty"` + + // raw marks Content as loaded from File, so Render leaves it alone. + raw bool +} + +// FileMode parses Mode. +func (w WriteSpec) FileMode() (fs.FileMode, error) { + if w.Mode == "" { + return 0o644, nil + } + v, err := strconv.ParseUint(w.Mode, 8, 32) + if err != nil { + return 0, fmt.Errorf("mode %q: %w", w.Mode, err) + } + return fs.FileMode(v), nil +} + +// Step is one action of a trajectory: exactly one of Run or Write. Run and +// Write.Content are text/template strings evaluated against the Instance. +type Step struct { + Name string `json:"name"` + Run string `json:"run,omitempty"` + Write *WriteSpec `json:"write,omitempty"` + // Think is the simulated model latency before the step (an RL rollout + // waits for the policy between commands). + Think Duration `json:"think,omitempty"` + // Timeout bounds a Run step; zero means the runner's default. + Timeout Duration `json:"timeout,omitempty"` + // ExpectExit, when set, fails the task if the exit code differs. A + // Reward step never fails the task on exit code: its exit code is the + // reward. + ExpectExit *int `json:"expect_exit,omitempty"` + // Reward marks the step whose exit code decides the task reward: + // 1 when it exits 0, else 0. At most one step per trajectory. + Reward bool `json:"reward,omitempty"` +} + +// Trajectory is the scripted rollout replayed in every environment. +type Trajectory struct { + Name string `json:"name"` + // Setup runs once per environment, before its first rollout, for + // example to record the image's working state as the baseline that + // Reset returns to. + Setup []Step `json:"setup,omitempty"` + // Reset runs before Steps on every later step an environment survives + // into, so a rollout starts from the image's state again (the gold patch + // cannot be applied twice). + Reset []Step `json:"reset,omitempty"` + Steps []Step `json:"steps"` +} + +// Names lists the built-in trajectories. +func Names() []string { + entries, err := fs.ReadDir(trajectoryFS, "trajectories") + if err != nil { + return nil + } + var names []string + for _, e := range entries { + if n, ok := strings.CutSuffix(e.Name(), ".yaml"); ok { + names = append(names, n) + } + } + return names +} + +// LoadTrajectory returns a built-in trajectory by name or one read from a +// file when source names an existing path. Empty means DefaultTrajectory. +func LoadTrajectory(source string) (*Trajectory, error) { + if source == "" { + source = DefaultTrajectory + } + if data, err := os.ReadFile(source); err == nil { + dir := filepath.Dir(source) + return parseWithAssets(data, func(name string) ([]byte, error) { + return os.ReadFile(filepath.Join(dir, name)) + }) + } + data, err := trajectoryFS.ReadFile("trajectories/" + source + ".yaml") + if err != nil { + return nil, fmt.Errorf("no file or built-in trajectory %q (built-in: %s)", source, strings.Join(Names(), ", ")) + } + return parseWithAssets(data, func(name string) ([]byte, error) { + return trajectoryFS.ReadFile("trajectories/" + name) + }) +} + +// parseWithAssets parses the trajectory and resolves every write.file +// through read, so the loaded trajectory carries its contents. +func parseWithAssets(data []byte, read func(name string) ([]byte, error)) (*Trajectory, error) { + t, err := ParseTrajectory(data) + if err != nil { + return nil, err + } + for _, steps := range [][]Step{t.Setup, t.Reset, t.Steps} { + for i := range steps { + w := steps[i].Write + if w == nil || w.File == "" { + continue + } + if strings.Contains(w.File, "/") || strings.Contains(w.File, "..") { + return nil, fmt.Errorf("trajectory %s step %s: write.file must be a plain file name", t.Name, steps[i].Name) + } + b, err := read(w.File) + if err != nil { + return nil, fmt.Errorf("trajectory %s step %s: %w", t.Name, steps[i].Name, err) + } + w.Content = string(b) + w.File = "" + w.raw = true + } + } + return t, nil +} + +// ParseTrajectory decodes and validates a YAML trajectory. +func ParseTrajectory(data []byte) (*Trajectory, error) { + var t Trajectory + if err := yaml.UnmarshalStrict(data, &t); err != nil { + return nil, fmt.Errorf("decode trajectory: %w", err) + } + if err := t.Validate(); err != nil { + return nil, err + } + return &t, nil +} + +// Validate checks the trajectory's shape without rendering it. +func (t *Trajectory) Validate() error { + if t.Name == "" { + return errors.New("trajectory: name is required") + } + if len(t.Steps) == 0 { + return fmt.Errorf("trajectory %s: at least one step is required", t.Name) + } + rewards := 0 + for _, s := range t.Steps { + if err := s.validate(); err != nil { + return fmt.Errorf("trajectory %s: %w", t.Name, err) + } + if s.Reward { + rewards++ + } + } + if rewards > 1 { + return fmt.Errorf("trajectory %s: more than one reward step", t.Name) + } + for section, steps := range map[string][]Step{"setup": t.Setup, "reset": t.Reset} { + for _, s := range steps { + if err := s.validate(); err != nil { + return fmt.Errorf("trajectory %s %s: %w", t.Name, section, err) + } + if s.Reward { + return fmt.Errorf("trajectory %s %s: step %s cannot be a reward step", t.Name, section, s.Name) + } + } + } + return nil +} + +func (s Step) validate() error { + if s.Name == "" { + return errors.New("step without a name") + } + switch { + case s.Run != "" && s.Write != nil: + return fmt.Errorf("step %s: run and write are exclusive", s.Name) + case s.Run == "" && s.Write == nil: + return fmt.Errorf("step %s: one of run or write is required", s.Name) + case s.Write != nil && s.Write.Path == "": + return fmt.Errorf("step %s: write.path is required", s.Name) + case s.Write != nil && s.Write.File != "" && s.Write.Content != "": + return fmt.Errorf("step %s: write.file and write.content are exclusive", s.Name) + case s.Write != nil && (s.Reward || s.ExpectExit != nil): + return fmt.Errorf("step %s: reward and expect_exit apply to run steps only", s.Name) + } + if s.Write != nil { + if _, err := s.Write.FileMode(); err != nil { + return fmt.Errorf("step %s: %w", s.Name, err) + } + } + if s.Think < 0 || s.Timeout < 0 { + return fmt.Errorf("step %s: think and timeout cannot be negative", s.Name) + } + return nil +} + +// HasReward reports whether a step decides the task reward. +func (t *Trajectory) HasReward() bool { + for _, s := range t.Steps { + if s.Reward { + return true + } + } + return false +} + +// Render evaluates every template in the trajectory against in and returns +// the concrete copy to replay. The original is not modified. +func (t *Trajectory) Render(in Instance) (*Trajectory, error) { + out := &Trajectory{Name: t.Name} + var err error + if out.Setup, err = renderSteps(t.Name, t.Setup, in); err != nil { + return nil, err + } + if out.Reset, err = renderSteps(t.Name, t.Reset, in); err != nil { + return nil, err + } + if out.Steps, err = renderSteps(t.Name, t.Steps, in); err != nil { + return nil, err + } + return out, nil +} + +func renderSteps(tname string, steps []Step, in Instance) ([]Step, error) { + out := make([]Step, len(steps)) + for i, s := range steps { + r := s + var err error + if r.Run, err = renderText(s.Run, in); err != nil { + return nil, fmt.Errorf("trajectory %s step %s run: %w", tname, s.Name, err) + } + if s.Write != nil { + w := *s.Write + if !w.raw { + if w.Content, err = renderText(s.Write.Content, in); err != nil { + return nil, fmt.Errorf("trajectory %s step %s write: %w", tname, s.Name, err) + } + } + r.Write = &w + } + out[i] = r + } + return out, nil +} + +// templateFuncs are available in step templates. json renders a value as +// JSON, for handing an instance's fields to a script in the sandbox. +var templateFuncs = template.FuncMap{ + "json": func(v any) (string, error) { + b, err := json.Marshal(v) + return string(b), err + }, +} + +func renderText(text string, in Instance) (string, error) { + if !strings.Contains(text, "{{") { + return text, nil + } + tmpl, err := template.New("step").Option("missingkey=error").Funcs(templateFuncs).Parse(text) + if err != nil { + return "", err + } + var buf bytes.Buffer + if err := tmpl.Execute(&buf, in); err != nil { + return "", err + } + return buf.String(), nil +} diff --git a/internal/benchmarking/swebench/trajectory_test.go b/internal/benchmarking/swebench/trajectory_test.go new file mode 100644 index 00000000..ab656052 --- /dev/null +++ b/internal/benchmarking/swebench/trajectory_test.go @@ -0,0 +1,185 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package swebench + +import ( + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +func TestBuiltinTrajectoriesLoadAndRender(t *testing.T) { + names := Names() + if len(names) == 0 { + t.Fatal("no built-in trajectories") + } + in := Instance{InstanceID: "x", Image: "reg/img@" + goodDigest, Patch: "PATCH", EvalScript: "EVAL"} + for _, n := range names { + tr, err := LoadTrajectory(n) + if err != nil { + t.Fatalf("%s: %v", n, err) + } + r, err := tr.Render(in) + if err != nil { + t.Fatalf("%s render: %v", n, err) + } + if len(r.Steps) != len(tr.Steps) { + t.Fatalf("%s: rendered %d steps, want %d", n, len(r.Steps), len(tr.Steps)) + } + for _, s := range r.Steps { + if s.Write != nil && (s.Write.Content == "" || s.Write.File != "") { + t.Fatalf("%s step %s: write.file not resolved into content", n, s.Name) + } + } + if len(r.Setup) != len(tr.Setup) || len(r.Reset) != len(tr.Reset) { + t.Fatalf("%s: setup/reset not rendered", n) + } + } + def, err := LoadTrajectory("") + if err != nil || def.Name != DefaultTrajectory { + t.Fatalf("default: %v %v", def, err) + } + if !def.HasReward() { + t.Fatal("default trajectory has no reward step") + } +} + +func TestRenderSubstitutesInstanceFields(t *testing.T) { + tr, err := ParseTrajectory([]byte(` +name: t +steps: + - name: w + write: {path: /tmp/p, content: "{{.Patch}}", mode: "0755"} + - name: r + run: "echo {{.InstanceID}}" + think: 250ms + timeout: 1m + reward: true +`)) + if err != nil { + t.Fatal(err) + } + r, err := tr.Render(Instance{InstanceID: "id-1", Patch: "line1\nline2"}) + if err != nil { + t.Fatal(err) + } + if r.Steps[0].Write.Content != "line1\nline2" || r.Steps[1].Run != "echo id-1" { + t.Fatalf("render: %+v", r.Steps) + } + if tr.Steps[1].Run != "echo {{.InstanceID}}" { + t.Fatal("Render must not modify the original") + } + if time.Duration(r.Steps[1].Think) != 250*time.Millisecond || time.Duration(r.Steps[1].Timeout) != time.Minute { + t.Fatalf("durations: %+v", r.Steps[1]) + } + if m, _ := r.Steps[0].Write.FileMode(); m != 0o755 { + t.Fatalf("mode: %v", m) + } +} + +func TestRenderUnknownFieldFails(t *testing.T) { + tr, err := ParseTrajectory([]byte("name: t\nsteps:\n - name: r\n run: \"{{.Nope}}\"\n")) + if err != nil { + t.Fatal(err) + } + if _, err := tr.Render(Instance{}); err == nil { + t.Fatal("unknown field must fail at render") + } +} + +func TestParseTrajectoryRejects(t *testing.T) { + cases := map[string]string{ + "no name": "steps:\n - name: a\n run: true\n", + "no steps": "name: t\n", + "run and write": "name: t\nsteps:\n - name: a\n run: x\n write: {path: /p}\n", + "neither": "name: t\nsteps:\n - name: a\n", + "two rewards": "name: t\nsteps:\n - {name: a, run: x, reward: true}\n - {name: b, run: y, reward: true}\n", + "reward reset": "name: t\nreset:\n - {name: a, run: x, reward: true}\nsteps:\n - {name: b, run: y}\n", + "reward setup": "name: t\nsetup:\n - {name: a, run: x, reward: true}\nsteps:\n - {name: b, run: y}\n", + "write reward": "name: t\nsteps:\n - name: a\n write: {path: /p}\n reward: true\n", + "bad mode": "name: t\nsteps:\n - name: a\n write: {path: /p, mode: \"zz\"}\n", + "bad think": "name: t\nsteps:\n - name: a\n run: x\n think: soon\n", + "neg think": "name: t\nsteps:\n - name: a\n run: x\n think: -1s\n", + "neg timeout": "name: t\nsteps:\n - name: a\n run: x\n timeout: -1m\n", + "unknown field": "name: t\nsteps:\n - name: a\n run: x\n bogus: 1\n", + } + for name, in := range cases { + if _, err := ParseTrajectory([]byte(in)); err == nil { + t.Errorf("%s: expected an error", name) + } else if strings.Contains(err.Error(), "panic") { + t.Errorf("%s: %v", name, err) + } + } +} + +func TestWriteFileResolvesNextToTrajectory(t *testing.T) { + dir := t.TempDir() + // The asset contains template syntax on purpose: file contents are not + // rendered. + if err := os.WriteFile(filepath.Join(dir, "asset.py"), []byte("print('{{not a template}}')\n"), 0o644); err != nil { + t.Fatal(err) + } + yamlPath := filepath.Join(dir, "t.yaml") + if err := os.WriteFile(yamlPath, []byte("name: t\nsteps:\n - name: w\n write: {path: /tmp/a.py, file: asset.py}\n - name: r\n run: \"true\"\n"), 0o644); err != nil { + t.Fatal(err) + } + tr, err := LoadTrajectory(yamlPath) + if err != nil { + t.Fatal(err) + } + if got := tr.Steps[0].Write.Content; got != "print('{{not a template}}')\n" { + t.Fatalf("content %q", got) + } + r, err := tr.Render(Instance{InstanceID: "x"}) + if err != nil { + t.Fatalf("rendering a trajectory with a file asset: %v", err) + } + if r.Steps[0].Write.Content != tr.Steps[0].Write.Content { + t.Fatal("file asset contents must not be rendered") + } + for _, bad := range []string{"missing.py", "../asset.py", "sub/asset.py"} { + if err := os.WriteFile(yamlPath, []byte("name: t\nsteps:\n - name: w\n write: {path: /tmp/a, file: "+bad+"}\n"), 0o644); err != nil { + t.Fatal(err) + } + if _, err := LoadTrajectory(yamlPath); err == nil { + t.Errorf("write.file %q must fail", bad) + } + } + if _, err := ParseTrajectory([]byte("name: t\nsteps:\n - name: w\n write: {path: /p, file: a, content: b}\n")); err == nil { + t.Error("file and content together must fail") + } +} + +func TestJSONTemplateFunc(t *testing.T) { + tr, err := ParseTrajectory([]byte("name: t\nsteps:\n - name: w\n write: {path: /tmp/i.json, content: \"{{json .}}\"}\n")) + if err != nil { + t.Fatal(err) + } + in := Instance{InstanceID: "x", Image: "reg/img@" + goodDigest, FailToPass: []string{"a::b[1]"}, LogParser: "parse_log_pytest"} + r, err := tr.Render(in) + if err != nil { + t.Fatal(err) + } + var back Instance + if err := json.Unmarshal([]byte(r.Steps[0].Write.Content), &back); err != nil { + t.Fatalf("not JSON: %v: %s", err, r.Steps[0].Write.Content) + } + if back.InstanceID != "x" || back.LogParser != "parse_log_pytest" || len(back.FailToPass) != 1 { + t.Fatalf("round trip: %+v", back) + } +}