initial

2026-05-28 19:49:29 +00:00
commit 0bfe0c314d
7 changed files with 766 additions and 0 deletions
@@ -0,0 +1,126 @@
+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
+}