ratatoskr-go/internal/opencode/runner.go

package opencode

import (
	"bufio"
	"context"
	"database/sql"
	"fmt"
	"io"
	"os"
	"os/exec"
	"strings"
	"sync"
	"time"

	_ "modernc.org/sqlite" // чисто-Go драйвер, без CGO → один статический бинарь
)

// Result — результат запуска opencode run. rc=-1 означает «убит по таймауту»
// (idle/hard): вызывающий НЕ должен ронять задачу, а обязан закоммитить/запушить
// готовую работу и отправить на ревью (класс O2 Timeout — результат, не ошибка).
type Result struct {
	RC        int
	Stdout    string
	SessionID string
}

// Runner — конфигурация запуска opencode-субагентов.
type Runner struct {
	Bin      string // путь к opencode (по умолчанию "opencode")
	DBPath   string // путь к opencode.db (idle-детекция активности)
	Config   string // путь к opencode.json (OPENCODE_CONFIG)
	ConfigDir string // путь к каталогу с агентами (OPENCODE_CONFIG_DIR)
	IdleTimeout time.Duration // нет новых сообщений в БД → зависание
	HardTimeout time.Duration // общий лимит на запуск
	PollInterval time.Duration

	// Заменяемые для тестов:
	Stdout io.Writer // диагностика (лог), по умолчанию os.Stderr
}

func (r *Runner) defaults() {
	if r.Bin == "" {
		r.Bin = "opencode"
	}
	if r.IdleTimeout == 0 {
		r.IdleTimeout = 2 * time.Minute
	}
	if r.HardTimeout == 0 {
		r.HardTimeout = 20 * time.Minute
	}
	if r.PollInterval == 0 {
		r.PollInterval = 2 * time.Second
	}
	if r.Stdout == nil {
		r.Stdout = os.Stderr
	}
}

func (r *Runner) logf(format string, args ...any) {
	fmt.Fprintf(r.Stdout, format+"\n", args...)
}

// maxDirMsgTS — максимальный time_updated (мс) по всем сообщениям сессий этого
// worktree: сигнал «модель/субагенты ещё активны». nil-nil если БД нет/пуста.
func (r *Runner) maxDirMsgTS(ctx context.Context, worktree string) (int64, bool) {
	if r.DBPath == "" {
		return 0, false
	}
	db, err := sql.Open("sqlite", "file:"+r.DBPath+"?mode=ro")
	if err != nil {
		return 0, false
	}
	defer db.Close()
	var ts sql.NullInt64
	err = db.QueryRowContext(ctx,
		"SELECT MAX(m.time_updated) FROM message m JOIN session s ON s.id = m.session_id WHERE s.directory = ?",
		worktree).Scan(&ts)
	if err != nil || !ts.Valid {
		return 0, false
	}
	return ts.Int64, true
}

func (r *Runner) latestSession(ctx context.Context, worktree, agent string) (string, bool) {
	if r.DBPath == "" {
		return "", false
	}
	db, err := sql.Open("sqlite", "file:"+r.DBPath+"?mode=ro")
	if err != nil {
		return "", false
	}
	defer db.Close()
	q := "SELECT id FROM session WHERE directory = ?"
	args := []any{worktree}
	if agent != "" {
		q += " AND agent = ?"
		args = append(args, agent)
	}
	q += " ORDER BY time_created DESC LIMIT 1"
	var id string
	if err := db.QueryRowContext(ctx, q, args...).Scan(&id); err != nil {
		return "", false
	}
	return id, id != ""
}

// Run запускает opencode run. Возвращает *Result (rc, stdout, session_id).
// Ошибка — только класс O1 ErrSpawn (не смог запустить бинарь). Таймауты
// дают rc=-1 в Result, а не error (класс O2).
func (r *Runner) Run(ctx context.Context, prompt, cwd, agent, sessionID string) (*Result, error) {
	r.defaults()
	cmd := []string{r.Bin, "run", "--agent", agent, "--format", "json", "--dir", cwd}
	if sessionID != "" {
		cmd = append(cmd, "--session", sessionID)
	}
	cmd = append(cmd, prompt)

	env := append(os.Environ(),
		"OPENCODE_DISABLE_AUTOUPDATE=1",
		"OPENCODE_DISABLE_MODELS_FETCH=1")
	if r.Config != "" {
		env = append(env, "OPENCODE_CONFIG="+r.Config)
	}
	if r.ConfigDir != "" {
		env = append(env, "OPENCODE_CONFIG_DIR="+r.ConfigDir)
	}

	proc := exec.CommandContext(ctx, cmd[0], cmd[1:]...)
	proc.Env = env
	proc.Dir = cwd
	// Убиваем всю process-group, чтобы дочерние процессы (sleep и т.п.) тоже
	// умерли и закрыли унаследованные stdout-fd (иначе <-done виснет).
	setpgid(proc)
	stdout, err := proc.StdoutPipe()
	if err != nil {
		return nil, fmt.Errorf("opencode: stdout pipe: %w", err)
	}
	proc.Stderr = proc.Stdout
	if err := proc.Start(); err != nil {
		return nil, fmt.Errorf("opencode: start %v: %w", cmd[0], err)
	}

	var buf []string
	var mu sync.Mutex
	done := make(chan struct{})
	go func() {
		defer close(done)
		sc := bufio.NewScanner(stdout)
		for sc.Scan() {
			mu.Lock()
			buf = append(buf, sc.Text())
			mu.Unlock()
		}
	}()

	baseline, _ := r.maxDirMsgTS(ctx, cwd)
	lastProgress := time.Now()
	launch := time.Now()
	killed := false

pollLoop:
	for {
		select {
		case <-done:
			// процесс завершился (pipe EOF) — выходим, берём exit code
			break pollLoop
		case <-ctx.Done():
			killGroup(proc)
			killed = true
			break pollLoop
		default:
		}
		if proc.ProcessState != nil && proc.ProcessState.Exited() {
			break pollLoop
		}
		now := time.Now()
		ts, ok := r.maxDirMsgTS(ctx, cwd)
		if ok && ts > baseline {
			lastProgress = now
		}
		if now.Sub(lastProgress) > r.IdleTimeout {
			r.logf("opencode(%s) idle %.0fs (нет новых сообщений) — kill", agent, r.IdleTimeout.Seconds())
			killGroup(proc)
			killed = true
			break pollLoop
		}
		if now.Sub(launch) > r.HardTimeout {
			r.logf("opencode(%s) hard timeout %.0fs — kill", agent, r.HardTimeout.Seconds())
			killGroup(proc)
			killed = true
			break pollLoop
		}
		time.Sleep(r.PollInterval)
	}

	<-done
	procErr := proc.Wait()
	rc := proc.ProcessState.ExitCode()
	if rc < 0 {
		rc = 1
	}
	if killed {
		rc = -1
	}
	_ = procErr

	mu.Lock()
	out := strings.Join(buf, "\n")
	mu.Unlock()

	sid := sessionID
	if s, ok := SessionIDFromOutput(out); ok {
		sid = s
	}
	if rc == -1 && sid == "" {
		if s, ok := r.latestSession(ctx, cwd, agent); ok {
			sid = s
		}
	}
	r.logf("opencode(%s) rc=%d", agent, rc)
	return &Result{RC: rc, Stdout: out, SessionID: sid}, nil
}

// ResumeDev — запуск dev-агента с resume-fallback. Если resume (sessionID)
// падает с rc!=0 (напр. сессия потеряна) — повторяем ОДИН раз свежей сессией
// в том же worktree. rc=-1 (kill по таймауту) НЕ триггерит fallback.
// Возвращает (result, timedOut).
func (r *Runner) ResumeDev(ctx context.Context, prompt, cwd, sessionID string) (*Result, bool) {
	res, err := r.Run(ctx, prompt, cwd, "dev", sessionID)
	if err != nil {
		// spawn-ошибку не ретраим fallback'ом — она повторится
		return res, false
	}
	if res.RC != 0 && res.RC != -1 && sessionID != "" {
		r.logf("dev resume rc=%d — запускаю заново без --session (worktree сохраняю)", res.RC)
		res, _ = r.Run(ctx, prompt+resumeFallbackNote, cwd, "dev", "")
	}
	return res, res.RC == -1
}

const resumeFallbackNote = "\n\n(Возобновление сессии не удалось; продолжи с учётом уже сделанных изменений в worktree.)"

// --- process-group helpers (Linux) ---
// Ставим процесс в собственную process-group, чтобы killGroup мог убить и
// дочерние процессы (иначе они держат унаследованные stdout-fd и <-done виснет).

func setpgid(proc *exec.Cmd) {
	sysProcAttr(proc)
}

func killGroup(proc *exec.Cmd) {
	if proc.Process != nil {
		killProcGroup(proc.Process.Pid)
	}
	_ = proc.Process.Kill()
}