Run script/setup in new worktrees and give each task a port block
This commit is contained in:
@@ -8,6 +8,7 @@ import (
|
||||
|
||||
"g.fc.run/misc/byre/internal/api"
|
||||
"g.fc.run/misc/byre/internal/claude"
|
||||
"g.fc.run/misc/byre/internal/store"
|
||||
"g.fc.run/misc/byre/internal/todo"
|
||||
)
|
||||
|
||||
@@ -60,7 +61,8 @@ func (m *Manager) bounceProseQuestion(l *live) claude.HookFunc {
|
||||
}
|
||||
|
||||
func (m *Manager) taskGuidance(ctx context.Context, task api.Task) (string, error) {
|
||||
g := guidance
|
||||
g := guidance + fmt.Sprintf("\n- This task's ports are %d to %d; $PORT holds the first. Run dev servers and anything else that listens on these, so parallel tasks don't collide.",
|
||||
task.PortBase, task.PortBase+store.PortBlock-1)
|
||||
if task.TodoItem != nil {
|
||||
g += fmt.Sprintf("\n- This task was started from the item %q in %s at the repo root. When the work is done, remove that item from %s in this worktree, so the removal lands in the same merge as the work.",
|
||||
*task.TodoItem, todo.File, todo.File)
|
||||
|
||||
+26
-14
@@ -62,6 +62,10 @@ type Manager struct {
|
||||
snapMu sync.Mutex
|
||||
snaps map[string]*snapState
|
||||
|
||||
setupMu sync.Mutex
|
||||
setups map[string]bool
|
||||
setupFailures map[string]string
|
||||
|
||||
bad *badLines
|
||||
watchStop chan struct{}
|
||||
}
|
||||
@@ -95,18 +99,20 @@ type live struct {
|
||||
|
||||
func New(st *store.Store, limits Limits, worktreeRoot, leadRoot string) *Manager {
|
||||
m := &Manager{
|
||||
store: st,
|
||||
limits: limits,
|
||||
worktreeRoot: worktreeRoot,
|
||||
leadRoot: leadRoot,
|
||||
live: map[string]*live{},
|
||||
rateLimits: map[api.AgentSessionVendor]map[string]any{},
|
||||
leadAnswered: map[string]bool{},
|
||||
paneFocused: map[string]bool{},
|
||||
signals: make(chan api.Event, 1024),
|
||||
snaps: map[string]*snapState{},
|
||||
bad: newBadLines(),
|
||||
watchStop: make(chan struct{}),
|
||||
store: st,
|
||||
limits: limits,
|
||||
worktreeRoot: worktreeRoot,
|
||||
leadRoot: leadRoot,
|
||||
live: map[string]*live{},
|
||||
rateLimits: map[api.AgentSessionVendor]map[string]any{},
|
||||
leadAnswered: map[string]bool{},
|
||||
paneFocused: map[string]bool{},
|
||||
signals: make(chan api.Event, 1024),
|
||||
snaps: map[string]*snapState{},
|
||||
setups: map[string]bool{},
|
||||
setupFailures: map[string]string{},
|
||||
bad: newBadLines(),
|
||||
watchStop: make(chan struct{}),
|
||||
}
|
||||
go m.signalLoop()
|
||||
return m
|
||||
@@ -406,7 +412,7 @@ func (m *Manager) turn(ctx context.Context, sess api.AgentSession, task api.Task
|
||||
return api.AgentSession{}, err
|
||||
}
|
||||
|
||||
if room {
|
||||
if room && !m.settingUp(task.Id) {
|
||||
return m.start(ctx, sess, task, actor, text, images, api.AgentSessionStateIdle)
|
||||
}
|
||||
|
||||
@@ -500,6 +506,9 @@ func (m *Manager) drain(ctx context.Context) error {
|
||||
}
|
||||
|
||||
for _, q := range queued {
|
||||
if m.settingUp(q.TaskID) {
|
||||
continue
|
||||
}
|
||||
room, err := m.hasRoom(ctx, q.Vendor)
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -652,7 +661,7 @@ func (m *Manager) ensureLive(ctx context.Context, sess api.AgentSession, task ap
|
||||
}
|
||||
tools := []mcp.Tool{m.askTool(l), m.cancelQuestionTool(l), m.createTaskTool(l), m.setMergeTitleTool(l)}
|
||||
hooks := map[string][]claude.HookMatcher{
|
||||
"UserPromptSubmit": {{Func: remindGuidance(g)}},
|
||||
"UserPromptSubmit": {{Func: m.remindTaskGuidance(l, g)}},
|
||||
"Stop": {{Func: m.bounceProseQuestion(l)}},
|
||||
"PostToolUse": {{Func: m.afterTool(l)}},
|
||||
}
|
||||
@@ -677,6 +686,9 @@ func (m *Manager) ensureLive(ctx context.Context, sess api.AgentSession, task ap
|
||||
Hooks: hooks,
|
||||
OnBadLine: func(line []byte, reason string) { m.recordBadLine(l, line, reason) },
|
||||
}
|
||||
if l.owner == "" {
|
||||
opts.Env = portEnv(task)
|
||||
}
|
||||
started, err := m.store.SessionStarted(ctx, sess.Id)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
||||
@@ -0,0 +1,113 @@
|
||||
package agents
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io/fs"
|
||||
"log/slog"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"g.fc.run/misc/byre/internal/api"
|
||||
"g.fc.run/misc/byre/internal/claude"
|
||||
)
|
||||
|
||||
const (
|
||||
setupScript = "script/setup"
|
||||
setupTimeout = 15 * time.Minute
|
||||
setupOutputBytes = 64 << 10
|
||||
)
|
||||
|
||||
func portEnv(t api.Task) []string {
|
||||
return []string{"PORT=" + strconv.Itoa(t.PortBase)}
|
||||
}
|
||||
|
||||
func (m *Manager) startSetup(t api.Task) {
|
||||
path := filepath.Join(t.WorktreePath, setupScript)
|
||||
_, err := os.Stat(path)
|
||||
if errors.Is(err, fs.ErrNotExist) {
|
||||
return
|
||||
}
|
||||
|
||||
m.setupMu.Lock()
|
||||
m.setups[t.Id] = true
|
||||
m.setupMu.Unlock()
|
||||
|
||||
go m.runSetup(t, path)
|
||||
}
|
||||
|
||||
func (m *Manager) settingUp(taskID string) bool {
|
||||
m.setupMu.Lock()
|
||||
defer m.setupMu.Unlock()
|
||||
return m.setups[taskID]
|
||||
}
|
||||
|
||||
func (m *Manager) takeSetupFailure(taskID string) string {
|
||||
m.setupMu.Lock()
|
||||
defer m.setupMu.Unlock()
|
||||
msg := m.setupFailures[taskID]
|
||||
delete(m.setupFailures, taskID)
|
||||
return msg
|
||||
}
|
||||
|
||||
func (m *Manager) runSetup(t api.Task, path string) {
|
||||
ctx := context.Background()
|
||||
m.emitTask(ctx, t, "setup.started", map[string]any{"script": setupScript, "port_base": t.PortBase})
|
||||
|
||||
start := time.Now()
|
||||
cctx, cancel := context.WithTimeout(ctx, setupTimeout)
|
||||
defer cancel()
|
||||
cmd := exec.CommandContext(cctx, path)
|
||||
cmd.Dir = t.WorktreePath
|
||||
cmd.Env = append(os.Environ(), portEnv(t)...)
|
||||
out, err := cmd.CombinedOutput()
|
||||
if cctx.Err() != nil {
|
||||
err = fmt.Errorf("timed out after %s", setupTimeout)
|
||||
}
|
||||
|
||||
output := string(out)
|
||||
if len(out) > setupOutputBytes {
|
||||
output = "…" + string(out[len(out)-setupOutputBytes:])
|
||||
}
|
||||
payload := map[string]any{"script": setupScript, "output": output, "duration_ms": time.Since(start).Milliseconds()}
|
||||
if err != nil {
|
||||
payload["error"] = err.Error()
|
||||
}
|
||||
slog.Info("setup ended", "task", t.Id, "err", err, "took", time.Since(start))
|
||||
m.emitTask(ctx, t, "setup.ended", payload)
|
||||
|
||||
m.setupMu.Lock()
|
||||
delete(m.setups, t.Id)
|
||||
if err != nil {
|
||||
m.setupFailures[t.Id] = fmt.Sprintf("byre ran %s in this worktree when the task was created, and it failed (%s). Its output:\n\n```\n%s\n```\n\nThe worktree may be missing dependencies or env files the setup would have provided.", setupScript, err, output)
|
||||
}
|
||||
m.setupMu.Unlock()
|
||||
|
||||
err = m.drain(ctx)
|
||||
if err != nil {
|
||||
slog.Error("start queued turns", "err", err)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) emitTask(ctx context.Context, t api.Task, kind string, payload map[string]any) {
|
||||
err := m.store.AppendEvent(ctx, api.Event{Kind: kind, ProjectId: &t.ProjectId, TaskId: &t.Id, Actor: "byre", Payload: payload})
|
||||
if err != nil {
|
||||
slog.Error("record setup event", "task", t.Id, "kind", kind, "err", err)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) remindTaskGuidance(l *live, text string) claude.HookFunc {
|
||||
return func(ctx context.Context, input json.RawMessage) (map[string]any, error) {
|
||||
failure := m.takeSetupFailure(l.taskID)
|
||||
if failure == "" {
|
||||
return remindGuidance(text)(ctx, input)
|
||||
}
|
||||
m.fileError(l, failure)
|
||||
return remindGuidance(text+"\n\n"+failure)(ctx, input)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,91 @@
|
||||
package agents_test
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"g.fc.run/misc/byre/internal/api"
|
||||
)
|
||||
|
||||
func (e *env) commitSetup(script string) {
|
||||
e.t.Helper()
|
||||
err := os.MkdirAll(filepath.Join(e.repo, "script"), 0o755)
|
||||
if err != nil {
|
||||
e.t.Fatal(err)
|
||||
}
|
||||
err = os.WriteFile(filepath.Join(e.repo, "script", "setup"), []byte(script), 0o755)
|
||||
if err != nil {
|
||||
e.t.Fatal(err)
|
||||
}
|
||||
gitOut(e.t, e.repo, "add", "script/setup")
|
||||
gitOut(e.t, e.repo, "commit", "--quiet", "--message", "setup")
|
||||
}
|
||||
|
||||
func (e *env) waitSetup(taskID string) api.Event {
|
||||
e.t.Helper()
|
||||
seen := e.waitFor("setup.ended", func(ev api.Event) bool {
|
||||
return ev.Kind == "setup.ended" && ev.TaskId != nil && *ev.TaskId == taskID
|
||||
})
|
||||
return seen[len(seen)-1]
|
||||
}
|
||||
|
||||
func TestSetupScriptAndPorts(t *testing.T) {
|
||||
e := newEnv(t)
|
||||
e.commitSetup("#!/bin/sh\nset -e\ncp \"$(git rev-parse --path-format=absolute --git-common-dir)/../.env\" .env\necho \"port $PORT\"\n")
|
||||
err := os.WriteFile(filepath.Join(e.repo, ".env"), []byte("SECRET=1\n"), 0o644)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
m := e.manager()
|
||||
|
||||
task, err := m.CreateTask(e.ctx, e.project, "with setup", login, "main", login)
|
||||
if err != nil {
|
||||
t.Fatalf("create task: %v", err)
|
||||
}
|
||||
if task.PortBase != 41010 {
|
||||
t.Fatalf("port_base = %d, want 41010 (41000 is taken by the first task)", task.PortBase)
|
||||
}
|
||||
|
||||
ev := e.waitSetup(task.Id)
|
||||
if ev.Payload["error"] != nil {
|
||||
t.Fatalf("setup failed: %v", ev.Payload)
|
||||
}
|
||||
want := "port 41010"
|
||||
if out, _ := ev.Payload["output"].(string); !strings.Contains(out, want) {
|
||||
t.Fatalf("output %q lacks %q", out, want)
|
||||
}
|
||||
env, err := os.ReadFile(filepath.Join(task.WorktreePath, ".env"))
|
||||
if err != nil || string(env) != "SECRET=1\n" {
|
||||
t.Fatalf("worktree .env = %q, %v", env, err)
|
||||
}
|
||||
|
||||
_, err = e.store.DiscardTask(e.ctx, e.task.Id, login)
|
||||
if err != nil {
|
||||
t.Fatalf("discard: %v", err)
|
||||
}
|
||||
reused, err := m.CreateTask(e.ctx, e.project, "reuses ports", login, "main", login)
|
||||
if err != nil {
|
||||
t.Fatalf("create task: %v", err)
|
||||
}
|
||||
if reused.PortBase != 41000 {
|
||||
t.Fatalf("port_base = %d, want the freed 41000", reused.PortBase)
|
||||
}
|
||||
e.waitSetup(reused.Id)
|
||||
}
|
||||
|
||||
func TestSetupFailure(t *testing.T) {
|
||||
e := newEnv(t)
|
||||
e.commitSetup("#!/bin/sh\necho installing\nexit 3\n")
|
||||
m := e.manager()
|
||||
|
||||
task, err := m.CreateTask(e.ctx, e.project, "failing setup", login, "main", login)
|
||||
if err != nil {
|
||||
t.Fatalf("create task: %v", err)
|
||||
}
|
||||
ev := e.waitSetup(task.Id)
|
||||
if ev.Payload["error"] != "exit status 3" || ev.Payload["output"] != "installing\n" {
|
||||
t.Fatalf("setup.ended payload = %v", ev.Payload)
|
||||
}
|
||||
}
|
||||
@@ -85,6 +85,11 @@ func (m *Manager) createTask(ctx context.Context, project api.Project, t api.Tas
|
||||
if err != nil {
|
||||
return api.Task{}, err
|
||||
}
|
||||
t, err = m.store.GetTask(ctx, t.Id)
|
||||
if err != nil {
|
||||
return api.Task{}, err
|
||||
}
|
||||
m.startSetup(t)
|
||||
return t, nil
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user