127 lines
3.3 KiB
Go
127 lines
3.3 KiB
Go
package main
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"fmt"
|
|
"log"
|
|
"os"
|
|
"os/exec"
|
|
"path/filepath"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
type Job struct {
|
|
EventType string
|
|
Event *GiteaEvent
|
|
Trigger *Trigger
|
|
}
|
|
|
|
// Worker runs jobs serially from an in-memory FIFO queue. One Claude session
|
|
// at a time — keeps logs readable, avoids racing on git workdirs, and bounds
|
|
// API spend.
|
|
type Worker struct {
|
|
cfg Config
|
|
queue chan Job
|
|
}
|
|
|
|
func NewWorker(cfg Config) *Worker {
|
|
w := &Worker{cfg: cfg, queue: make(chan Job, 128)}
|
|
go w.run()
|
|
return w
|
|
}
|
|
|
|
func (w *Worker) Enqueue(j Job) bool {
|
|
select {
|
|
case w.queue <- j:
|
|
return true
|
|
default:
|
|
log.Printf("[queue] FULL — dropping %s on %s", j.EventType, j.Event.Repository.FullName)
|
|
return false
|
|
}
|
|
}
|
|
|
|
func (w *Worker) run() {
|
|
for job := range w.queue {
|
|
start := time.Now()
|
|
tag := fmt.Sprintf("%s#%d", job.Event.Repository.FullName, job.Trigger.Number)
|
|
log.Printf("[job %s] starting (event=%s reason=%s)", tag, job.EventType, job.Trigger.Reason)
|
|
if err := w.handle(job); err != nil {
|
|
log.Printf("[job %s] FAILED after %s: %v", tag, time.Since(start), err)
|
|
} else {
|
|
log.Printf("[job %s] done in %s", tag, time.Since(start))
|
|
}
|
|
}
|
|
}
|
|
|
|
func (w *Worker) handle(job Job) error {
|
|
dir, err := os.MkdirTemp(w.cfg.WorkspaceRoot, "klaus-*")
|
|
if err != nil {
|
|
return fmt.Errorf("mkdtemp: %w", err)
|
|
}
|
|
preserve := false
|
|
defer func() {
|
|
if preserve {
|
|
log.Printf("[workspace] preserved at %s", dir)
|
|
return
|
|
}
|
|
_ = os.RemoveAll(dir)
|
|
}()
|
|
|
|
repoDir := filepath.Join(dir, "repo")
|
|
sshURL := job.Event.Repository.SSHURL
|
|
if sshURL == "" {
|
|
return fmt.Errorf("repository has no ssh_url; cannot clone")
|
|
}
|
|
if err := runShell(dir, "git", "clone", sshURL, repoDir); err != nil {
|
|
return fmt.Errorf("clone: %w", err)
|
|
}
|
|
|
|
// For PR-related triggers, check out the PR head so Klaus sees the right code.
|
|
if job.Trigger.IsPR && job.Trigger.BranchHint != "" {
|
|
if err := runShell(repoDir, "git", "fetch", "origin", job.Trigger.BranchHint); err != nil {
|
|
log.Printf("[job] warn: fetch %s: %v", job.Trigger.BranchHint, err)
|
|
} else if err := runShell(repoDir, "git", "checkout", job.Trigger.BranchHint); err != nil {
|
|
log.Printf("[job] warn: checkout %s: %v", job.Trigger.BranchHint, err)
|
|
}
|
|
}
|
|
|
|
prompt := buildPrompt(job, repoDir)
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), w.cfg.JobTimeout)
|
|
defer cancel()
|
|
|
|
cmd := exec.CommandContext(ctx, "claude",
|
|
"-p", prompt,
|
|
"--permission-mode", "bypassPermissions",
|
|
)
|
|
cmd.Dir = repoDir
|
|
cmd.Env = append(os.Environ(),
|
|
// Don't let Klaus's per-job sessions pollute the klaus-bot project memory.
|
|
"CLAUDE_CODE_AUTO_MEMORY=0",
|
|
)
|
|
cmd.Stdout = os.Stdout
|
|
cmd.Stderr = os.Stderr
|
|
if err := cmd.Run(); err != nil {
|
|
preserve = true
|
|
return fmt.Errorf("claude: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// runShell runs a command with logged invocation; on failure the combined
|
|
// output is returned in the error so the operator can see what went wrong.
|
|
func runShell(dir, name string, args ...string) error {
|
|
log.Printf("$ (cwd=%s) %s %s", dir, name, strings.Join(args, " "))
|
|
cmd := exec.Command(name, args...)
|
|
cmd.Dir = dir
|
|
var out bytes.Buffer
|
|
cmd.Stdout = &out
|
|
cmd.Stderr = &out
|
|
if err := cmd.Run(); err != nil {
|
|
return fmt.Errorf("%s %s: %w\n%s", name, strings.Join(args, " "), err, out.String())
|
|
}
|
|
return nil
|
|
}
|