feat: множественные репозитории (Repos) и клонирование в воркере
All checks were successful
CI / test (push) Successful in 46s
CI / build-and-package (amd64, darwin) (push) Successful in 36s
CI / build-and-package (amd64, linux) (push) Successful in 37s
CI / build-and-package (amd64, windows) (push) Successful in 44s
CI / build-and-package (arm64, darwin) (push) Successful in 35s
CI / build-and-package (arm64, linux) (push) Successful in 36s

- Task.Repos []string (XML-колонка repos, обратная совместимость с repo)
- config: блок git {base_url, token}
- аналитик: ответ repos[], шаблон показывает список
- core: propose без repos → возврат в сбор (E1)
- worker вариант A: один dev из общего cwd, prepareRepos клонирует
  недостающие репо (git clone), validateRepoName (E3), ErrRepoNotGit (E4)
- ошибки E1-E4 в worker/errors.go
This commit is contained in:
Hermes
2026-08-16 09:18:15 +05:00
parent 7be75fe14f
commit bd3d825738
14 changed files with 346 additions and 71 deletions

View File

@@ -14,6 +14,10 @@ telegram:
# hard_timeout: "20m" # hard_timeout: "20m"
# idle_timeout: "2m" # idle_timeout: "2m"
# git (источник репозиториев для git clone)
# base_url: "http://gitea.hal9000.home" # базовый URL git-хоста (обязательно для клонирования)
# token: "${GIT_TOKEN}" # токен для приватных репозиториев
# chat: # chat:
# poll_interval: "30s" # poll_interval: "30s"

View File

@@ -31,7 +31,8 @@ type AnalystResponse struct {
Phase string `json:"phase"` Phase string `json:"phase"`
Title string `json:"title"` Title string `json:"title"`
Goal string `json:"goal"` Goal string `json:"goal"`
Repo string `json:"repo"` Repo string `json:"repo"` // одиночный репо (обратная совместимость)
Repos []string `json:"repos"` // список репо (основной)
Why string `json:"why"` Why string `json:"why"`
AC string `json:"ac"` AC string `json:"ac"`
Questions []string `json:"questions"` Questions []string `json:"questions"`
@@ -57,7 +58,7 @@ func (a *Analyst) Decide(ctx context.Context, history []core.Message, draft stor
td := TemplateData{ td := TemplateData{
Title: draft.Title, Title: draft.Title,
Goal: draft.Goal, Goal: draft.Goal,
Repo: draft.Repo, Repos: draft.EffectiveRepos(),
Why: draft.Why, Why: draft.Why,
AC: draft.AC, AC: draft.AC,
History: hist, History: hist,
@@ -114,6 +115,11 @@ func (a *Analyst) Decide(ctx context.Context, history []core.Message, draft stor
if ar.Repo != "" { if ar.Repo != "" {
dec.Draft.Repo = ar.Repo dec.Draft.Repo = ar.Repo
} }
if len(ar.Repos) > 0 {
dec.Draft.Repos = ar.Repos
// Синхронизируем одиночный repo для старых потребителей.
dec.Draft.Repo = strings.Join(ar.Repos, ",")
}
if ar.Why != "" { if ar.Why != "" {
dec.Draft.Why = ar.Why dec.Draft.Why = ar.Why
} }
@@ -148,7 +154,7 @@ func validateResponse(ar *AnalystResponse) error {
return fmt.Errorf("phase=ask, но нет ни chat_reply, ни questions") return fmt.Errorf("phase=ask, но нет ни chat_reply, ни questions")
} }
case "propose": case "propose":
if ar.Title == "" && ar.Goal == "" && ar.Repo == "" && ar.Why == "" && ar.AC == "" { if ar.Title == "" && ar.Goal == "" && len(ar.Repos) == 0 && ar.Why == "" && ar.AC == "" {
return fmt.Errorf("phase=propose, но нет ни одного изменённого поля") return fmt.Errorf("phase=propose, но нет ни одного изменённого поля")
} }
case "abort": case "abort":

View File

@@ -20,7 +20,7 @@ var promptTemplate = template.Must(template.New("analyst").Parse(`Ты — ан
|------|----------| |------|----------|
| title | краткое название (14 слова) | | title | краткое название (14 слова) |
| goal | цель задачи: что именно нужно сделать | | goal | цель задачи: что именно нужно сделать |
| repo | путь к репозиторию (относительно /opt/data/src/) | | repos | список репозиториев (имена на git-хосте; для связанных — все сразу) |
| why | зачем это нужно, контекст | | why | зачем это нужно, контекст |
| ac | acceptance criteria — конкретный результат, что считается готовым | | ac | acceptance criteria — конкретный результат, что считается готовым |
@@ -35,7 +35,12 @@ var promptTemplate = template.Must(template.New("analyst").Parse(`Ты — ан
**Текущий черновик:** **Текущий черновик:**
{{if .Title}} title: {{.Title}}{{else}} title: (не задано){{end}} {{if .Title}} title: {{.Title}}{{else}} title: (не задано){{end}}
{{if .Goal}} goal: {{.Goal}}{{else}} goal: (не задано){{end}} {{if .Goal}} goal: {{.Goal}}{{else}} goal: (не задано){{end}}
{{if .Repo}} repo: {{.Repo}}{{else}} repo: (не задано){{end}} {{if .Repos}}
repos:
{{- range .Repos}}
- {{.}}
{{- end}}
{{- else}} repos: (не задано){{end}}
{{if .Why}} why: {{.Why}}{{else}} why: (не задано){{end}} {{if .Why}} why: {{.Why}}{{else}} why: (не задано){{end}}
{{if .AC}} ac: {{.AC}}{{else}} ac: (не задано){{end}} {{if .AC}} ac: {{.AC}}{{else}} ac: (не задано){{end}}
@@ -44,7 +49,7 @@ var promptTemplate = template.Must(template.New("analyst").Parse(`Ты — ан
"phase": "ask|propose|abort", "phase": "ask|propose|abort",
"title": "название (только если меняешь)", "title": "название (только если меняешь)",
"goal": "цель (только если меняешь)", "goal": "цель (только если меняешь)",
"repo": "путь к репозиторию (только если меняешь)", "repos": ["имя_репо_1", "имя_репо_2"],
"why": "зачем (только если меняешь)", "why": "зачем (только если меняешь)",
"ac": "критерии (только если меняешь)", "ac": "критерии (только если меняешь)",
"questions": ["вопрос 1", "вопрос 2"], "questions": ["вопрос 1", "вопрос 2"],
@@ -57,7 +62,7 @@ var promptTemplate = template.Must(template.New("analyst").Parse(`Ты — ан
type TemplateData struct { type TemplateData struct {
Title string Title string
Goal string Goal string
Repo string Repos []string
Why string Why string
AC string AC string
History string // отформатированная переписка History string // отформатированная переписка

View File

@@ -105,6 +105,8 @@ func New(configPath string) (*App, error) {
Agent: "dev", Agent: "dev",
Interval: 5 * time.Second, Interval: 5 * time.Second,
MaxJobs: 2, MaxJobs: 2,
GitBaseURL: cfg.Git.BaseURL,
GitToken: cfg.Git.Token,
} }
a.Router = router a.Router = router
a.Worker = w a.Worker = w

View File

@@ -36,10 +36,17 @@ func (d *Duration) UnmarshalYAML(unmarshal func(any) error) error {
type Config struct { type Config struct {
Telegram TelegramCfg `yaml:"telegram"` Telegram TelegramCfg `yaml:"telegram"`
OpenCode OpenCodeCfg `yaml:"opencode"` OpenCode OpenCodeCfg `yaml:"opencode"`
Git GitCfg `yaml:"git"`
Chat ChatCfg `yaml:"chat"` Chat ChatCfg `yaml:"chat"`
Paths PathsCfg `yaml:"paths"` Paths PathsCfg `yaml:"paths"`
} }
// GitCfg — источник репозиториев (для git clone).
type GitCfg struct {
BaseURL string `yaml:"base_url" env:"GIT_BASE_URL"`
Token string `yaml:"token" env:"GIT_TOKEN"`
}
type TelegramCfg struct { type TelegramCfg struct {
Token string `yaml:"token" env:"TG_TOKEN"` Token string `yaml:"token" env:"TG_TOKEN"`
ChatID string `yaml:"chat_id" env:"TG_CHAT_ID"` ChatID string `yaml:"chat_id" env:"TG_CHAT_ID"`

View File

@@ -318,8 +318,17 @@ func (c *Core) runDecide(ctx context.Context, task *storage.Task, force bool) (R
return Result{Reply: reply, Action: "drop", TaskID: task.ID, Status: task.Status}, nil return Result{Reply: reply, Action: "drop", TaskID: task.ID, Status: task.Status}, nil
case "propose": case "propose":
// применяем черновик и переходим в ready // применяем черновик
applyDraft(task, decision.Draft) applyDraft(task, decision.Draft)
// E1: propose без репозиториев → остаёмся в сборе, просим уточнить.
if len(task.EffectiveRepos()) == 0 {
task.Status = storage.StatusCollecting
if err := c.Store.UpdateTask(ctx, task); err != nil {
return Result{}, err
}
reply := decChatReply(decision, "Укажи, в каком репозитории(ях) вести работу.")
return Result{Reply: reply, Action: "send", TaskID: task.ID, Status: task.Status}, nil
}
task.Status = storage.StatusReady task.Status = storage.StatusReady
if err := c.Store.UpdateTask(ctx, task); err != nil { if err := c.Store.UpdateTask(ctx, task); err != nil {
return Result{}, err return Result{}, err
@@ -353,6 +362,10 @@ func applyDraft(task *storage.Task, dec storage.Task) {
if dec.Repo != "" { if dec.Repo != "" {
task.Repo = dec.Repo task.Repo = dec.Repo
} }
if len(dec.Repos) > 0 {
task.Repos = dec.Repos
task.Repo = strings.Join(dec.Repos, ",")
}
if dec.Why != "" { if dec.Why != "" {
task.Why = dec.Why task.Why = dec.Why
} }
@@ -361,6 +374,14 @@ func applyDraft(task *storage.Task, dec storage.Task) {
} }
} }
// decChatReply возвращает ChatReply из вердикта или переданный дефолт.
func decChatReply(dec Decision, fallback string) string {
if dec.ChatReply != "" {
return dec.ChatReply
}
return fallback
}
// buildAskReply формирует ответ с вопросами. // buildAskReply формирует ответ с вопросами.
func buildAskReply(dec Decision, max int) string { func buildAskReply(dec Decision, max int) string {
reply := dec.ChatReply reply := dec.ChatReply
@@ -391,8 +412,8 @@ func formatSummary(t storage.Task) string {
if t.Title != "" { if t.Title != "" {
b.WriteString("**Название:** " + t.Title + "\n") b.WriteString("**Название:** " + t.Title + "\n")
} }
if t.Repo != "" { if len(t.EffectiveRepos()) > 0 {
b.WriteString("**Репозиторий:** " + t.Repo + "\n") b.WriteString("**Репозитории:** " + strings.Join(t.EffectiveRepos(), ", ") + "\n")
} }
if t.Goal != "" { if t.Goal != "" {
b.WriteString("**Цель:** " + t.Goal + "\n") b.WriteString("**Цель:** " + t.Goal + "\n")

View File

@@ -148,7 +148,7 @@ func TestAbortReturnsDrop(t *testing.T) {
func TestConsentInReady(t *testing.T) { func TestConsentInReady(t *testing.T) {
c, ctx, store := setupCore(t, func(ctx context.Context, history []Message, draft storage.Task, force bool) (Decision, error) { c, ctx, store := setupCore(t, func(ctx context.Context, history []Message, draft storage.Task, force bool) (Decision, error) {
return Decision{Phase: "propose", Draft: storage.Task{Title: "X"}}, nil return Decision{Phase: "propose", Draft: storage.Task{Title: "X", Repos: []string{"repo-x"}}}, nil
}) })
id := mkTask(t, store, ctx, "u1") id := mkTask(t, store, ctx, "u1")
@@ -166,7 +166,7 @@ func TestEditInReadyGoesCollecting(t *testing.T) {
var calls int var calls int
c, ctx, store := setupCore(t, func(ctx context.Context, history []Message, draft storage.Task, force bool) (Decision, error) { c, ctx, store := setupCore(t, func(ctx context.Context, history []Message, draft storage.Task, force bool) (Decision, error) {
calls++ calls++
return Decision{Phase: "propose", Draft: storage.Task{Title: "X"}}, nil return Decision{Phase: "propose", Draft: storage.Task{Title: "X", Repos: []string{"repo-x"}}}, nil
}) })
id := mkTask(t, store, ctx, "u1") id := mkTask(t, store, ctx, "u1")

View File

@@ -1,5 +1,7 @@
package storage package storage
import "encoding/json"
// Status — статус задачи (state machine). // Status — статус задачи (state machine).
type Status string type Status string
@@ -63,7 +65,8 @@ type Task struct {
ChatID string `json:"chat_id"` // tg://<id> ChatID string `json:"chat_id"` // tg://<id>
Title string `json:"title"` Title string `json:"title"`
Goal string `json:"goal"` Goal string `json:"goal"`
Repo string `json:"repo"` Repo string `json:"repo"` // обратная совместимость: одиночный репозиторий
Repos []string `json:"repos"` // список репозиториев (основной)
Why string `json:"why"` Why string `json:"why"`
AC string `json:"ac"` // acceptance criteria AC string `json:"ac"` // acceptance criteria
TaskTag string `json:"task_tag"` // UUID, стабильный на всю жизнь TaskTag string `json:"task_tag"` // UUID, стабильный на всю жизнь
@@ -72,6 +75,38 @@ type Task struct {
UpdatedAt SQLiteTime `json:"updated_at"` UpdatedAt SQLiteTime `json:"updated_at"`
} }
// ReposJoined возвращает repos как одну строку (JSON-массив) для хранения в БД.
// Пустой список → пустая строка.
func (t *Task) ReposJoined() string {
if len(t.Repos) == 0 {
return ""
}
b, _ := json.Marshal(t.Repos)
return string(b)
}
// EffectiveRepos возвращает сам репо: если Repos пуст, но Repo задан —
// подтягивает одиночный (обратная совместимость).
func (t *Task) EffectiveRepos() []string {
if len(t.Repos) > 0 {
return t.Repos
}
if t.Repo != "" {
return []string{t.Repo}
}
return nil
}
// SetReposFromDB заполняет Repos из сохранённой строки (JSON), либо
// из одиночного repo (обратная совместимость).
func (t *Task) SetReposFromDB(repos string) {
if repos == "" {
t.Repos = nil
return
}
_ = json.Unmarshal([]byte(repos), &t.Repos)
}
// Trace — запись трассировки выполнения. // Trace — запись трассировки выполнения.
type TraceStatus string type TraceStatus string

View File

@@ -5,6 +5,7 @@ import (
"database/sql" "database/sql"
"database/sql/driver" "database/sql/driver"
"fmt" "fmt"
"strings"
"time" "time"
_ "modernc.org/sqlite" _ "modernc.org/sqlite"
@@ -127,6 +128,7 @@ func (s *Storage) migrate(ctx context.Context) error {
why TEXT NOT NULL DEFAULT '', why TEXT NOT NULL DEFAULT '',
ac TEXT NOT NULL DEFAULT '', ac TEXT NOT NULL DEFAULT '',
task_tag TEXT NOT NULL DEFAULT '', task_tag TEXT NOT NULL DEFAULT '',
repos TEXT NOT NULL DEFAULT '',
status TEXT NOT NULL DEFAULT 'draft', status TEXT NOT NULL DEFAULT 'draft',
created_at INTEGER NOT NULL DEFAULT (unixepoch()), created_at INTEGER NOT NULL DEFAULT (unixepoch()),
updated_at INTEGER NOT NULL DEFAULT (unixepoch()) updated_at INTEGER NOT NULL DEFAULT (unixepoch())
@@ -162,9 +164,24 @@ func (s *Storage) migrate(ctx context.Context) error {
if _, err := s.db.ExecContext(ctx, schema); err != nil { if _, err := s.db.ExecContext(ctx, schema); err != nil {
return fmt.Errorf("%w: migrate: %w", ErrDB, err) return fmt.Errorf("%w: migrate: %w", ErrDB, err)
} }
// Доп. колонка repos (множественные репозитории). Idempotent: если колонка
// уже есть, ALTER вернёт ошибку duplicate column — её игнорируем.
if _, err := s.db.ExecContext(ctx,
`ALTER TABLE tasks ADD COLUMN repos TEXT NOT NULL DEFAULT ''`); err != nil {
// SQLite 3.35+ выдаёт "duplicate column name"; ранние версии — "duplicate column".
if !isDuplicateColumn(err) {
return fmt.Errorf("%w: migrate add repos: %w", ErrDB, err)
}
}
return nil return nil
} }
// isDuplicateColumn распознаёт ошибку SQLite «duplicate column name».
func isDuplicateColumn(err error) bool {
msg := strings.ToLower(err.Error())
return strings.Contains(msg, "duplicate column")
}
// Now возвращает текущее время UTC как SQLiteTime. // Now возвращает текущее время UTC как SQLiteTime.
func Now() SQLiteTime { func Now() SQLiteTime {
return SQLiteTime(time.Now().UTC()) return SQLiteTime(time.Now().UTC())

View File

@@ -11,9 +11,9 @@ import (
func (s *Storage) CreateTask(ctx context.Context, t *Task) (int64, error) { func (s *Storage) CreateTask(ctx context.Context, t *Task) (int64, error) {
now := Now() now := Now()
res, err := s.db.ExecContext(ctx, ` res, err := s.db.ExecContext(ctx, `
INSERT INTO tasks (chat_id, title, goal, repo, why, ac, task_tag, status, created_at, updated_at) INSERT INTO tasks (chat_id, title, goal, repo, repos, why, ac, task_tag, status, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
t.ChatID, t.Title, t.Goal, t.Repo, t.Why, t.AC, t.ChatID, t.Title, t.Goal, t.Repo, t.ReposJoined(), t.Why, t.AC,
t.TaskTag, StatusDraft, now, now, t.TaskTag, StatusDraft, now, now,
) )
if err != nil { if err != nil {
@@ -33,10 +33,11 @@ func (s *Storage) CreateTask(ctx context.Context, t *Task) (int64, error) {
// GetTask возвращает задачу по ID. // GetTask возвращает задачу по ID.
func (s *Storage) GetTask(ctx context.Context, id int64) (*Task, error) { func (s *Storage) GetTask(ctx context.Context, id int64) (*Task, error) {
t := &Task{} t := &Task{}
var reposStr string
err := s.db.QueryRowContext(ctx, ` err := s.db.QueryRowContext(ctx, `
SELECT id, chat_id, title, goal, repo, why, ac, task_tag, status, created_at, updated_at SELECT id, chat_id, title, goal, repo, repos, why, ac, task_tag, status, created_at, updated_at
FROM tasks WHERE id = ?`, id).Scan( FROM tasks WHERE id = ?`, id).Scan(
&t.ID, &t.ChatID, &t.Title, &t.Goal, &t.Repo, &t.ID, &t.ChatID, &t.Title, &t.Goal, &t.Repo, &reposStr,
&t.Why, &t.AC, &t.TaskTag, &t.Status, &t.CreatedAt, &t.UpdatedAt, &t.Why, &t.AC, &t.TaskTag, &t.Status, &t.CreatedAt, &t.UpdatedAt,
) )
if err == sql.ErrNoRows { if err == sql.ErrNoRows {
@@ -45,6 +46,7 @@ func (s *Storage) GetTask(ctx context.Context, id int64) (*Task, error) {
if err != nil { if err != nil {
return nil, fmt.Errorf("%w: get task %d: %w", ErrDB, id, err) return nil, fmt.Errorf("%w: get task %d: %w", ErrDB, id, err)
} }
t.SetReposFromDB(reposStr)
return t, nil return t, nil
} }
@@ -72,9 +74,9 @@ func (s *Storage) UpdateTask(ctx context.Context, t *Task) error {
now := Now() now := Now()
res, err := s.db.ExecContext(ctx, ` res, err := s.db.ExecContext(ctx, `
UPDATE tasks UPDATE tasks
SET title=?, goal=?, repo=?, why=?, ac=?, status=?, updated_at=? SET title=?, goal=?, repo=?, repos=?, why=?, ac=?, status=?, updated_at=?
WHERE id=?`, WHERE id=?`,
t.Title, t.Goal, t.Repo, t.Why, t.AC, t.Status, now, t.ID, t.Title, t.Goal, t.Repo, t.ReposJoined(), t.Why, t.AC, t.Status, now, t.ID,
) )
if err != nil { if err != nil {
return fmt.Errorf("%w: update task %d: %w", ErrDB, t.ID, err) return fmt.Errorf("%w: update task %d: %w", ErrDB, t.ID, err)
@@ -91,12 +93,13 @@ func (s *Storage) UpdateTask(ctx context.Context, t *Task) error {
// Терминальные статусы: success, cancelled, aborted, closed. // Терминальные статусы: success, cancelled, aborted, closed.
func (s *Storage) GetActiveTaskByChatID(ctx context.Context, chatID string) (*Task, error) { func (s *Storage) GetActiveTaskByChatID(ctx context.Context, chatID string) (*Task, error) {
t := &Task{} t := &Task{}
var reposStr string
err := s.db.QueryRowContext(ctx, ` err := s.db.QueryRowContext(ctx, `
SELECT id, chat_id, title, goal, repo, why, ac, task_tag, status, created_at, updated_at SELECT id, chat_id, title, goal, repo, repos, why, ac, task_tag, status, created_at, updated_at
FROM tasks FROM tasks
WHERE chat_id = ? AND status NOT IN ('success','cancelled','aborted','closed') WHERE chat_id = ? AND status NOT IN ('success','cancelled','aborted','closed')
ORDER BY updated_at DESC LIMIT 1`, chatID).Scan( ORDER BY updated_at DESC LIMIT 1`, chatID).Scan(
&t.ID, &t.ChatID, &t.Title, &t.Goal, &t.Repo, &t.ID, &t.ChatID, &t.Title, &t.Goal, &t.Repo, &reposStr,
&t.Why, &t.AC, &t.TaskTag, &t.Status, &t.CreatedAt, &t.UpdatedAt) &t.Why, &t.AC, &t.TaskTag, &t.Status, &t.CreatedAt, &t.UpdatedAt)
if err == sql.ErrNoRows { if err == sql.ErrNoRows {
return nil, fmt.Errorf("%w: no active task for chat %s", ErrNotFound, chatID) return nil, fmt.Errorf("%w: no active task for chat %s", ErrNotFound, chatID)
@@ -104,6 +107,7 @@ func (s *Storage) GetActiveTaskByChatID(ctx context.Context, chatID string) (*Ta
if err != nil { if err != nil {
return nil, fmt.Errorf("%w: get active task %s: %w", ErrDB, chatID, err) return nil, fmt.Errorf("%w: get active task %s: %w", ErrDB, chatID, err)
} }
t.SetReposFromDB(reposStr)
return t, nil return t, nil
} }
@@ -125,7 +129,7 @@ func (s *Storage) ListTasks(ctx context.Context, filter TaskFilter) ([]*Task, er
args = append(args, filter.Limit, filter.Offset) args = append(args, filter.Limit, filter.Offset)
rows, err := s.db.QueryContext(ctx, ` rows, err := s.db.QueryContext(ctx, `
SELECT id, chat_id, title, goal, repo, why, ac, task_tag, status, created_at, updated_at SELECT id, chat_id, title, goal, repo, repos, why, ac, task_tag, status, created_at, updated_at
FROM tasks WHERE `+where+` ORDER BY updated_at DESC LIMIT ? OFFSET ?`, args...) FROM tasks WHERE `+where+` ORDER BY updated_at DESC LIMIT ? OFFSET ?`, args...)
if err != nil { if err != nil {
return nil, fmt.Errorf("%w: list tasks: %w", ErrDB, err) return nil, fmt.Errorf("%w: list tasks: %w", ErrDB, err)
@@ -135,10 +139,12 @@ func (s *Storage) ListTasks(ctx context.Context, filter TaskFilter) ([]*Task, er
var tasks []*Task var tasks []*Task
for rows.Next() { for rows.Next() {
t := &Task{} t := &Task{}
if err := rows.Scan(&t.ID, &t.ChatID, &t.Title, &t.Goal, &t.Repo, var reposStr string
if err := rows.Scan(&t.ID, &t.ChatID, &t.Title, &t.Goal, &t.Repo, &reposStr,
&t.Why, &t.AC, &t.TaskTag, &t.Status, &t.CreatedAt, &t.UpdatedAt); err != nil { &t.Why, &t.AC, &t.TaskTag, &t.Status, &t.CreatedAt, &t.UpdatedAt); err != nil {
return nil, fmt.Errorf("%w: scan task: %w", ErrDB, err) return nil, fmt.Errorf("%w: scan task: %w", ErrDB, err)
} }
t.SetReposFromDB(reposStr)
tasks = append(tasks, t) tasks = append(tasks, t)
} }
return tasks, rows.Err() return tasks, rows.Err()

View File

@@ -2,7 +2,7 @@ package worker
import "errors" import "errors"
// Классы ошибок W1W5. // Классы ошибок W1W5 (планировщик) и E1E4 (репозитории).
var ( var (
// W1 — ошибка опроса БД. // W1 — ошибка опроса БД.
ErrPoll = errors.New("W1: poll error") ErrPoll = errors.New("W1: poll error")
@@ -18,4 +18,16 @@ var (
// W5 — превышена параллельность, задача пропущена. // W5 — превышена параллельность, задача пропущена.
ErrConcurrencyLimit = errors.New("W5: concurrency limit") ErrConcurrencyLimit = errors.New("W5: concurrency limit")
// E1 — задача без репозиториев (нечего клонировать/править).
ErrNoRepos = errors.New("E1: no repositories")
// E2 — git clone упал.
ErrClone = errors.New("E2: clone failed")
// E3 — имя репозитория содержит путь-эскейп (../, / и т.п.).
ErrRepoPathHint = errors.New("E3: repo path escape")
// E4 — папка существует, но не является git-репозиторием.
ErrRepoNotGit = errors.New("E4: existing dir is not a git repo")
) )

View File

@@ -6,28 +6,35 @@ import (
) )
// devPromptTemplate — промпт для dev-агента при запуске задачи. // devPromptTemplate — промпт для dev-агента при запуске задачи.
var devPromptTemplate = template.Must(template.New("dev").Parse(`Ты — dev-агент, реализуешь задачу в репозитории. var devPromptTemplate = template.Must(template.New("dev").Parse(`Ты — dev-агент, реализуешь задачу в репозитории(ях).
**Задача:** **Задача:**
{{if .Title}}Название: {{.Title}}{{end}} {{if .Title}}Название: {{.Title}}{{end}}
{{if .Goal}}Цель: {{.Goal}}{{end}} {{if .Goal}}Цель: {{.Goal}}{{end}}
{{if .Repos}}
Репозитории (доступны как подпапки текущего каталога):
{{- range .Repos}}
- {{.}}
{{- end}}
{{end}}
{{if .Why}}Зачем: {{.Why}}{{end}} {{if .Why}}Зачем: {{.Why}}{{end}}
{{if .Repo}}Репозиторий: {{.Repo}}{{end}}
{{if .AC}}Критерии готовности: {{if .AC}}Критерии готовности:
{{.AC}}{{end}} {{.AC}}{{end}}
**Инструкции:** **Инструкции:**
1. Напиши код, реализующий задачу. 1. Рабочий каталог — общий корень, в котором лежат все репозитории по именам.
2. Убедись, что все acceptance criteria выполнены. Правь файлы внутри нужного репозитория (./имя_репо/...). Связанные репозитории меняй согласованно.
3. В процессе работы пользуйся встроенными инструментами opencode (чтение файлов, поиск, редактирование). 2. Напиши код, реализующий задачу.
4. По окончании работы верни краткий отчёт о том, что сделано. 3. Убедись, что все acceptance criteria выполнены.
4. Пользуйся встроенными инструментами opencode (чтение файлов, поиск, редактирование).
5. По окончании верни краткий отчёт о том, что сделано.
`)) `))
// DevPromptData — данные для рендера dev-промпта. // DevPromptData — данные для рендера dev-промпта.
type DevPromptData struct { type DevPromptData struct {
Title string Title string
Goal string Goal string
Repo string Repos []string
Why string Why string
AC string AC string
} }

View File

@@ -4,7 +4,10 @@ import (
"context" "context"
"fmt" "fmt"
"log" "log"
"os"
"os/exec"
"path/filepath" "path/filepath"
"strings"
"time" "time"
"github.com/kamelion/ratatoskr-go/internal/opencode" "github.com/kamelion/ratatoskr-go/internal/opencode"
@@ -23,11 +26,15 @@ type PollTaskFunc func(ctx context.Context) error
type Worker struct { type Worker struct {
Store *storage.Storage Store *storage.Storage
Runner OpenCodeRunner Runner OpenCodeRunner
Worktree string // базовый путь, task.Repo — относительно него Worktree string // общий каталог, репозитории вкладываются в него по имени
Agent string // default "dev" Agent string // default "dev"
Interval time.Duration // интервал опроса БД Interval time.Duration // интервал опроса БД
MaxJobs int // макс. параллельных задач MaxJobs int // макс. параллельных задач
// Git — источник репозиториев для клонирования.
GitBaseURL string
GitToken string
sem chan struct{} // семафор sem chan struct{} // семафор
cancel context.CancelFunc cancel context.CancelFunc
@@ -117,24 +124,42 @@ func (w *Worker) pollAndDispatch(ctx context.Context) error {
return nil return nil
} }
// runTask выполняет одну задачу: dev-агент через opencode. // runTask выполняет одну задачу: готовит репозитории, затем dev-агент через opencode.
func (w *Worker) runTask(ctx context.Context, task *storage.Task) (err error) { func (w *Worker) runTask(ctx context.Context, task *storage.Task) (err error) {
// 1. проверяем статус // 1. проверяем статус
if task.Status != storage.StatusReady { if task.Status != storage.StatusReady {
return fmt.Errorf("%w: task %d status=%q", ErrLaunch, task.ID, task.Status) return fmt.Errorf("%w: task %d status=%q", ErrLaunch, task.ID, task.Status)
} }
// 2. ставим running repos := task.EffectiveRepos()
if len(repos) == 0 {
return fmt.Errorf("%w: task %d: %v", ErrLaunch, task.ID, ErrNoRepos)
}
// 1b. проверяем имена репо (E3): не допускаем путь-escape.
for _, r := range repos {
if err := validateRepoName(r); err != nil {
return fmt.Errorf("%w: task %d repo %q: %w", ErrLaunch, task.ID, r, err)
}
}
// 2. ставим running (после валидации — чтобы плохие имена не жгли состояние)
task.Status = storage.StatusRunning task.Status = storage.StatusRunning
if err := w.Store.UpdateTask(ctx, task); err != nil { if err := w.Store.UpdateTask(ctx, task); err != nil {
return fmt.Errorf("%w: set running: %v", ErrUpdate, err) return fmt.Errorf("%w: set running: %v", ErrUpdate, err)
} }
// 2b. клонируем недостающие репозитории в общий каталог.
if err := w.prepareRepos(ctx, repos); err != nil {
w.failTask(ctx, task)
return fmt.Errorf("%w: %v", ErrClone, err)
}
// 3. рендерим промпт // 3. рендерим промпт
prompt, err := RenderDevPrompt(DevPromptData{ prompt, err := RenderDevPrompt(DevPromptData{
Title: task.Title, Title: task.Title,
Goal: task.Goal, Goal: task.Goal,
Repo: task.Repo, Repos: repos,
Why: task.Why, Why: task.Why,
AC: task.AC, AC: task.AC,
}) })
@@ -153,21 +178,16 @@ func (w *Worker) runTask(ctx context.Context, task *storage.Task) (err error) {
return fmt.Errorf("%w: create: %v", ErrTrace, err) return fmt.Errorf("%w: create: %v", ErrTrace, err)
} }
// 5. вычисляем cwd // 5. cwd — общий каталог (вариант A: один dev видит все репозитории).
cwd := w.resolveCwd(task.Repo) cwd := w.Worktree
// 6. запускаем dev-агент // 6. запускаем dev-агент
res, resErr := w.Runner.Run(ctx, prompt, cwd, w.Agent, "") res, resErr := w.Runner.Run(ctx, prompt, cwd, w.Agent, "")
if resErr != nil { if resErr != nil {
// O1 ErrSpawn — не смог запустить бинарь // O1 ErrSpawn — не смог запустить бинарь
task.Status = storage.StatusFailed w.failTask(ctx, task)
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()) w.finalizeTrace(ctx, traceID, storage.TraceFailed, resErr.Error())
err = fmt.Errorf("%w: spawn: %v", ErrLaunch, resErr) return fmt.Errorf("%w: spawn: %v", ErrLaunch, resErr)
return
} }
// 6b. сохраняем session_id из результата // 6b. сохраняем session_id из результата
@@ -193,13 +213,20 @@ func (w *Worker) runTask(ctx context.Context, task *storage.Task) (err error) {
// 8. сохраняем результат // 8. сохраняем результат
if e := w.Store.UpdateTask(ctx, task); e != nil { if e := w.Store.UpdateTask(ctx, task); e != nil {
err = fmt.Errorf("%w: set %s: %v", ErrUpdate, task.Status, e) return fmt.Errorf("%w: set %s: %v", ErrUpdate, task.Status, e)
return
} }
w.finalizeTrace(ctx, traceID, traceStatus, output) w.finalizeTrace(ctx, traceID, traceStatus, output)
return nil return nil
} }
// failTask помечает задачу failed.
func (w *Worker) failTask(ctx context.Context, task *storage.Task) {
task.Status = storage.StatusFailed
if e := w.Store.UpdateTask(ctx, task); e != nil {
log.Printf("worker: task %d: set failed: %v", task.ID, e)
}
}
// finalizeTrace обновляет output и статус трассы. // finalizeTrace обновляет output и статус трассы.
func (w *Worker) finalizeTrace(ctx context.Context, traceID int64, status storage.TraceStatus, output string) { func (w *Worker) finalizeTrace(ctx context.Context, traceID int64, status storage.TraceStatus, output string) {
if e := w.Store.UpdateTraceOutput(ctx, traceID, output); e != nil { if e := w.Store.UpdateTraceOutput(ctx, traceID, output); e != nil {
@@ -210,9 +237,63 @@ func (w *Worker) finalizeTrace(ctx context.Context, traceID int64, status storag
} }
} }
func (w *Worker) resolveCwd(repo string) string { // prepareRepos гарантирует наличие всех репозиториев в общем каталоге:
if repo == "" { // папка есть → используем как есть; нет → git clone.
return w.Worktree func (w *Worker) prepareRepos(ctx context.Context, repos []string) error {
for _, r := range repos {
dst := filepath.Join(w.Worktree, r)
if _, err := os.Stat(dst); err == nil {
gitdir := filepath.Join(dst, ".git")
if _, e := os.Stat(gitdir); e != nil {
// E4: папка есть, но не git-репо — клонировать поверх нельзя.
return fmt.Errorf("%w: %s существует, но не git-репозиторий", ErrRepoNotGit, r)
} }
return filepath.Join(w.Worktree, repo) continue // уже готово
}
if err := w.clone(ctx, r); err != nil {
return fmt.Errorf("%w: %v", ErrClone, err)
}
}
return nil
}
// clone клонирует репозиторий r в ./worktrees/<r>.
func (w *Worker) clone(ctx context.Context, repo string) error {
base := strings.TrimRight(w.GitBaseURL, "/")
if base == "" {
return fmt.Errorf("git.base_url не задан в конфиге")
}
url := buildCloneURL(base, repo)
dst := filepath.Join(w.Worktree, repo)
args := []string{"clone"}
if w.GitToken != "" {
// https-базовый URL: кладём токен внутрь URL (для приватных репозиториев).
args = append(args, "--config", "http.extraHeader=Authorization: Bearer "+w.GitToken)
}
args = append(args, url, dst)
cmd := exec.CommandContext(ctx, "git", args...)
out, err := cmd.CombinedOutput()
if err != nil {
return fmt.Errorf("%w: git clone %s: %s", ErrClone, repo, strings.TrimSpace(string(out)))
}
return nil
}
// buildCloneURL собирает URL клона из base_url и имени репозитория.
func buildCloneURL(base, repo string) string {
return strings.TrimRight(base, "/") + "/" + repo + ".git"
}
// validateRepoName отклоняет имена с пути-эскейпом (E3).
func validateRepoName(repo string) error {
if repo == "" {
return fmt.Errorf("%w: пустое имя", ErrRepoPathHint)
}
if strings.Contains(repo, "/") || strings.Contains(repo, "..") {
return fmt.Errorf("%w: %q", ErrRepoPathHint, repo)
}
return nil
} }

View File

@@ -4,6 +4,7 @@ import (
"context" "context"
"errors" "errors"
"fmt" "fmt"
"os"
"path/filepath" "path/filepath"
"strings" "strings"
"testing" "testing"
@@ -41,7 +42,7 @@ func createReadyTask(t *testing.T, s *storage.Storage, title string) *storage.Ta
ChatID: "tg://worker-test", ChatID: "tg://worker-test",
Title: title, Title: title,
Goal: "сделать " + title, Goal: "сделать " + title,
Repo: "test/" + title, Repos: []string{title},
Why: "для теста", Why: "для теста",
AC: "работает", AC: "работает",
TaskTag: "test-" + title, TaskTag: "test-" + title,
@@ -63,6 +64,16 @@ func createReadyTask(t *testing.T, s *storage.Storage, title string) *storage.Ta
return task return task
} }
// seedFakeRepo создаёт в worktree/<repo>/ папку с .git, чтобы prepareRepos
// прошёл без реального git clone.
func seedFakeRepo(t *testing.T, worktree, repo string) {
t.Helper()
dir := filepath.Join(worktree, repo)
if err := os.MkdirAll(filepath.Join(dir, ".git"), 0o755); err != nil {
t.Fatalf("seed repo %s: %v", repo, err)
}
}
func TestWorkerHappyPath(t *testing.T) { func TestWorkerHappyPath(t *testing.T) {
s := setupWorkerDB(t) s := setupWorkerDB(t)
task := createReadyTask(t, s, "calc") task := createReadyTask(t, s, "calc")
@@ -73,6 +84,7 @@ func TestWorkerHappyPath(t *testing.T) {
Worktree: t.TempDir(), Worktree: t.TempDir(),
Agent: "dev", Agent: "dev",
} }
seedFakeRepo(t, w.Worktree, "calc")
ctx := context.Background() ctx := context.Background()
if err := w.runTask(ctx, task); err != nil { if err := w.runTask(ctx, task); err != nil {
@@ -114,6 +126,7 @@ func TestWorkerTimeout(t *testing.T) {
Runner: &mockRunnerWorker{result: &opencode.Result{RC: -1, Stdout: ""}}, Runner: &mockRunnerWorker{result: &opencode.Result{RC: -1, Stdout: ""}},
Worktree: t.TempDir(), Worktree: t.TempDir(),
} }
seedFakeRepo(t, w.Worktree, "slow")
ctx := context.Background() ctx := context.Background()
_ = w.runTask(ctx, task) _ = w.runTask(ctx, task)
@@ -147,6 +160,7 @@ func TestWorkerSpawnError(t *testing.T) {
Runner: &mockRunnerWorker{err: errors.New("opencode not found")}, Runner: &mockRunnerWorker{err: errors.New("opencode not found")},
Worktree: t.TempDir(), Worktree: t.TempDir(),
} }
seedFakeRepo(t, w.Worktree, "spawn-fail")
ctx := context.Background() ctx := context.Background()
_ = w.runTask(ctx, task) _ = w.runTask(ctx, task)
@@ -180,6 +194,7 @@ func TestWorkerNonZeroExit(t *testing.T) {
Runner: &mockRunnerWorker{result: &opencode.Result{RC: 7, Stdout: "error"}}, Runner: &mockRunnerWorker{result: &opencode.Result{RC: 7, Stdout: "error"}},
Worktree: t.TempDir(), Worktree: t.TempDir(),
} }
seedFakeRepo(t, w.Worktree, "fail")
ctx := context.Background() ctx := context.Background()
_ = w.runTask(ctx, task) _ = w.runTask(ctx, task)
@@ -236,6 +251,7 @@ func TestWorkerPromptRendered(t *testing.T) {
Runner: &mockRunnerWorker{result: &opencode.Result{RC: 0, Stdout: "ok", SessionID: "s"}}, Runner: &mockRunnerWorker{result: &opencode.Result{RC: 0, Stdout: "ok", SessionID: "s"}},
Worktree: t.TempDir(), Worktree: t.TempDir(),
} }
seedFakeRepo(t, w.Worktree, "prompt-test")
ctx := context.Background() ctx := context.Background()
_ = w.runTask(ctx, task) _ = w.runTask(ctx, task)
@@ -254,20 +270,74 @@ func TestWorkerPromptRendered(t *testing.T) {
if !strings.Contains(tr.Prompt, "prompt-test") { if !strings.Contains(tr.Prompt, "prompt-test") {
t.Error("prompt не содержит название задачи") t.Error("prompt не содержит название задачи")
} }
if !strings.Contains(tr.Prompt, "test/prompt-test") { if !strings.Contains(tr.Prompt, " - prompt-test") {
t.Error("prompt не содержит repo") t.Error("prompt не содержит репозиторий prompt-test")
} }
} }
func TestWorkerResolveCwd(t *testing.T) { func TestValidateRepoName(t *testing.T) {
base := "/opt/data/src" valid := []string{"calc", "proj-a", "my.repo", "node_2"}
w := &Worker{Worktree: base} for _, r := range valid {
if err := validateRepoName(r); err != nil {
if got := w.resolveCwd(""); got != base { t.Errorf("validateRepoName(%q) = %v, want nil", r, err)
t.Errorf("empty repo: got %q, want %q", got, base)
} }
if got := w.resolveCwd("tools/calc"); got != filepath.Join(base, "tools/calc") { }
t.Errorf("repo: got %q, want %q", got, filepath.Join(base, "tools/calc"))
invalid := []string{"", "../etc", "a/b", "a/../b", ".."}
for _, r := range invalid {
if err := validateRepoName(r); err == nil {
t.Errorf("validateRepoName(%q) = nil, want E3", r)
} else if !errors.Is(err, ErrRepoPathHint) {
t.Errorf("validateRepoName(%q) err = %v, want E3", r, err)
}
}
}
func TestBuildCloneURL(t *testing.T) {
if got := buildCloneURL("http://gitea.hal9000.home", "proj-a"); got != "http://gitea.hal9000.home/proj-a.git" {
t.Errorf("buildCloneURL = %q", got)
}
if got := buildCloneURL("http://gitea.hal9000.home/", "proj-b"); got != "http://gitea.hal9000.home/proj-b.git" {
t.Errorf("buildCloneURL trailing slash = %q", got)
}
}
func TestPrepareRepos(t *testing.T) {
s := setupWorkerDB(t)
_ = s
wt := t.TempDir()
w := &Worker{Worktree: wt, GitBaseURL: "http://gitea.hal9000.home"}
// seedFakeRepo уже создал .git — prepareRepos должен пройти без клона.
seedFakeRepo(t, wt, "proj-a")
if err := w.prepareRepos(context.Background(), []string{"proj-a"}); err != nil {
t.Fatalf("prepareRepos existing: %v", err)
}
// отсутствующий репо без git в PATH → E2 (clone упал), но не паника.
err := w.prepareRepos(context.Background(), []string{"missing"})
if err == nil {
t.Fatal("prepareRepos missing: expected error")
}
if !errors.Is(err, ErrClone) {
t.Errorf("prepareRepos missing err = %v, want E2", err)
}
}
func TestPrepareReposNonGitDir(t *testing.T) {
wt := t.TempDir()
w := &Worker{Worktree: wt}
// папка есть, но без .git → E4.
if err := os.MkdirAll(filepath.Join(wt, "plain"), 0o755); err != nil {
t.Fatal(err)
}
err := w.prepareRepos(context.Background(), []string{"plain"})
if err == nil {
t.Fatal("expected E4 error")
}
if !errors.Is(err, ErrRepoNotGit) {
t.Errorf("err = %v, want E4", err)
} }
} }
@@ -303,6 +373,8 @@ func TestWorkerSemaphore(t *testing.T) {
MaxJobs: 1, MaxJobs: 1,
Interval: 50 * time.Millisecond, Interval: 50 * time.Millisecond,
} }
seedFakeRepo(t, w.Worktree, "task-0")
seedFakeRepo(t, w.Worktree, "task-1")
w.sem = make(chan struct{}, 1) w.sem = make(chan struct{}, 1)
w.sem <- struct{}{} w.sem <- struct{}{}