worker: добавлен пакет worker (W1-W5, poll-цикл, runTask, промпт, тесты) + SetMaxOpenConns(1) для SQLITE_BUSY

2026-08-15 09:14:34 +05:00
parent dfe374dd55
commit 4641090398
6 changed files with 627 additions and 0 deletions
--- a/internal/worker/worker.go
+++ b/internal/worker/worker.go
@@ -0,0 +1,218 @@
+package worker
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"path/filepath"
+	"time"
+
+	"github.com/kamelion/ratatoskr-go/internal/opencode"
+	"github.com/kamelion/ratatoskr-go/internal/storage"
+)
+
+// OpenCodeRunner — интерфейс для opencode (подменяемый в тестах).
+type OpenCodeRunner interface {
+	Run(ctx context.Context, prompt, cwd, agent, sessionID string) (*opencode.Result, error)
+}
+
+// PollTaskFunc — callback для обработки готовой задачи (подменяемый в тестах).
+type PollTaskFunc func(ctx context.Context) error
+
+// Worker — планировщик, запускающий готовые задачи (status=ready → running → success/failed/timeout).
+type Worker struct {
+	Store    *storage.Storage
+	Runner   OpenCodeRunner
+	Worktree string        // базовый путь, task.Repo — относительно него
+	Agent    string        // default "dev"
+	Interval time.Duration // интервал опроса БД
+	MaxJobs  int           // макс. параллельных задач
+
+	sem  chan struct{} // семафор
+	cancel context.CancelFunc
+
+	// подменяемый poll для тестов
+	pollFn PollTaskFunc
+}
+
+// Start запускает цикл опроса в фоновой горутине.
+func (w *Worker) Start(ctx context.Context) {
+	if w.Agent == "" {
+		w.Agent = "dev"
+	}
+	if w.Interval <= 0 {
+		w.Interval = 5 * time.Second
+	}
+	if w.MaxJobs <= 0 {
+		w.MaxJobs = 2
+	}
+	w.sem = make(chan struct{}, w.MaxJobs)
+	// заполняем семафор токенами
+	for i := 0; i < w.MaxJobs; i++ {
+		w.sem <- struct{}{}
+	}
+
+	ctx, w.cancel = context.WithCancel(ctx)
+
+	pollFn := w.pollFn
+	if pollFn == nil {
+		pollFn = w.pollAndDispatch
+	}
+
+	go func() {
+		// первый poll сразу
+		_ = pollFn(ctx)
+
+		ticker := time.NewTicker(w.Interval)
+		defer ticker.Stop()
+
+		for {
+			select {
+			case <-ctx.Done():
+				log.Print("worker: stopped")
+				return
+			case <-ticker.C:
+				_ = pollFn(ctx)
+			}
+		}
+	}()
+}
+
+// Stop останавливает воркер (отменяет контекст → убивает активные задачи).
+func (w *Worker) Stop() {
+	if w.cancel != nil {
+		w.cancel()
+	}
+}
+
+// pollAndDispatch ищет готовые задачи, запускает их в пределах свободных слотов.
+func (w *Worker) pollAndDispatch(ctx context.Context) error {
+	slots := len(w.sem)
+	if slots == 0 {
+		return nil
+	}
+
+	tasks, err := w.Store.ListTasks(ctx, storage.TaskFilter{
+		Status: storage.StatusReady,
+		Limit:  slots,
+	})
+	if err != nil {
+		return fmt.Errorf("%w: %v", ErrPoll, err)
+	}
+
+	for _, t := range tasks {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case slot := <-w.sem:
+			task := t
+			go func() {
+				defer func() { w.sem <- slot }()
+				if err := w.runTask(ctx, task); err != nil {
+					log.Printf("worker: task %d: %v", task.ID, err)
+				}
+			}()
+		}
+	}
+	return nil
+}
+
+// runTask выполняет одну задачу: dev-агент через opencode.
+func (w *Worker) runTask(ctx context.Context, task *storage.Task) (err error) {
+	// 1. проверяем статус
+	if task.Status != storage.StatusReady {
+		return fmt.Errorf("%w: task %d status=%q", ErrLaunch, task.ID, task.Status)
+	}
+
+	// 2. ставим running
+	task.Status = storage.StatusRunning
+	if err := w.Store.UpdateTask(ctx, task); err != nil {
+		return fmt.Errorf("%w: set running: %v", ErrUpdate, err)
+	}
+
+	// 3. рендерим промпт
+	prompt, err := RenderDevPrompt(DevPromptData{
+		Title: task.Title,
+		Goal:  task.Goal,
+		Repo:  task.Repo,
+		Why:   task.Why,
+		AC:    task.AC,
+	})
+	if err != nil {
+		return fmt.Errorf("%w: render prompt: %v", ErrTrace, err)
+	}
+
+	// 4. создаём трассу
+	trace := &storage.Trace{
+		TaskID: task.ID,
+		Agent:  w.Agent,
+		Prompt: prompt,
+	}
+	traceID, err := w.Store.AppendTrace(ctx, trace)
+	if err != nil {
+		return fmt.Errorf("%w: create: %v", ErrTrace, err)
+	}
+
+	// 5. вычисляем cwd
+	cwd := w.resolveCwd(task.Repo)
+
+	// 6. запускаем dev-агент
+	res, resErr := w.Runner.Run(ctx, prompt, cwd, w.Agent, "")
+	if resErr != nil {
+		// O1 ErrSpawn — не смог запустить бинарь
+		task.Status = storage.StatusFailed
+		if e := w.Store.UpdateTask(ctx, task); e != nil {
+			err = fmt.Errorf("%w: set failed: %v", ErrUpdate, e)
+			return
+		}
+		w.finalizeTrace(ctx, traceID, storage.TraceFailed, resErr.Error())
+		err = fmt.Errorf("%w: spawn: %v", ErrLaunch, resErr)
+		return
+	}
+
+	// 6b. сохраняем session_id из результата
+	if res.SessionID != "" {
+		_ = w.Store.UpdateTraceSessionID(ctx, traceID, res.SessionID)
+	}
+
+	// 7. определяем результат по RC
+	output := res.Stdout
+	var traceStatus storage.TraceStatus
+
+	switch {
+	case res.RC == 0:
+		task.Status = storage.StatusSuccess
+		traceStatus = storage.TraceSuccess
+	case res.RC == -1:
+		task.Status = storage.StatusTimeout
+		traceStatus = storage.TraceTimeout
+	default:
+		task.Status = storage.StatusFailed
+		traceStatus = storage.TraceFailed
+	}
+
+	// 8. сохраняем результат
+	if e := w.Store.UpdateTask(ctx, task); e != nil {
+		err = fmt.Errorf("%w: set %s: %v", ErrUpdate, task.Status, e)
+		return
+	}
+	w.finalizeTrace(ctx, traceID, traceStatus, output)
+	return nil
+}
+
+// finalizeTrace обновляет output и статус трассы.
+func (w *Worker) finalizeTrace(ctx context.Context, traceID int64, status storage.TraceStatus, output string) {
+	if e := w.Store.UpdateTraceOutput(ctx, traceID, output); e != nil {
+		log.Printf("worker: update trace output %d: %v", traceID, e)
+	}
+	if e := w.Store.UpdateTraceStatus(ctx, traceID, status); e != nil {
+		log.Printf("worker: update trace status %d: %v", traceID, e)
+	}
+}
+
+func (w *Worker) resolveCwd(repo string) string {
+	if repo == "" {
+		return w.Worktree
+	}
+	return filepath.Join(w.Worktree, repo)
+}