ratatoskr-go/internal/opencode/live.go

package opencode

import (
	"context"
	"encoding/json"
	"strings"
	"sync"
	"time"
)

// parseLiveStep пытается распарсить одну NDJSON-строку stdout opencode как
// событие (text/tool/agent). Возвращает nil, если строка не является событием.
func parseLiveStep(line string) *LiveStep {
	line = strings.TrimSpace(line)
	if !strings.HasPrefix(line, "{") {
		return nil
	}
	var obj struct {
		Type string `json:"type"`
		Part struct {
			Text string `json:"text"`
		} `json:"part"`
	}
	if err := json.Unmarshal([]byte(line), &obj); err != nil {
		return nil
	}
	if obj.Type == "" {
		return nil
	}
	return &LiveStep{Type: obj.Type, Text: obj.Part.Text, At: time.Now()}
}

// LiveStep — один наблюдаемый шаг агента из NDJSON-потока opencode run.
// Собирается из live-строк stdout, не из БД.
type LiveStep struct {
	Type string // "text" | "tool" | "agent" | ...
	Text string // содержимое text-парта (для других типов может быть пустым)
	At   time.Time
}

// LiveSession — накопленное состояние живой сессии одной задачи.
type LiveSession struct {
	Agent  string
	Start  time.Time
	Last   time.Time // время последнего LIVE-шага
	Steps  []LiveStep
	MaxLen int
}

// Snapshot возвращает копию live-состояния (безопасно без гонок).
type LiveSnap struct {
	Agent string
	Last  time.Time
	Steps []LiveStep
}

func (s *LiveSession) snapshot() LiveSnap {
	steps := append([]LiveStep(nil), s.Steps...)
	return LiveSnap{Agent: s.Agent, Last: s.Last, Steps: steps}
}

// LiveRegistry — thread-safe журнал живых сессий по taskID.
// Runner пишет шаги в контексте Run, /status N читает снимок.
type LiveRegistry struct {
	mu       sync.Mutex
	sessions map[int64]*LiveSession
}

func NewLiveRegistry() *LiveRegistry {
	return &LiveRegistry{sessions: make(map[int64]*LiveSession)}
}

// Start регистрирует начало сессии задачи. agent — метка субагента (dev/reviewer).
func (r *LiveRegistry) Start(taskID int64, agent string) {
	if r == nil {
		return
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	now := time.Now()
	r.sessions[taskID] = &LiveSession{Agent: agent, Start: now, Last: now, MaxLen: 50}
}

// Observe добавляет live-шаг к сессии задачи. Пропускается, если сессия не начата.
func (r *LiveRegistry) Observe(taskID int64, step LiveStep) {
	if r == nil {
		return
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	s, ok := r.sessions[taskID]
	if !ok {
		return
	}
	s.Last = time.Now()
	s.Steps = append(s.Steps, step)
	if s.MaxLen > 0 && len(s.Steps) > s.MaxLen {
		s.Steps = s.Steps[len(s.Steps)-s.MaxLen:]
	}
}

// Finish удаляет сессию задачи (задача завершилась).
func (r *LiveRegistry) Finish(taskID int64) {
	if r == nil {
		return
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	delete(r.sessions, taskID)
}

// Snap возвращает текущий снимок сессии задачи. ok=false если сессия не активна.
func (r *LiveRegistry) Snap(taskID int64) (LiveSnap, bool) {
	if r == nil {
		return LiveSnap{}, false
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	s, ok := r.sessions[taskID]
	if !ok {
		return LiveSnap{}, false
	}
	return s.snapshot(), true
}

// --- контекст ---

type liveCtxKey struct{}
type liveCtxVal struct {
	reg    *LiveRegistry
	taskID int64
}

// WithLive возвращает контекст, в котором Runner при запуске наблюдает живую
// сессию задачи taskID и пишет шаги в reg. reg==nil означает «не наблюдать».
func WithLive(ctx context.Context, reg *LiveRegistry, taskID int64) context.Context {
	return context.WithValue(ctx, liveCtxKey{}, liveCtxVal{reg: reg, taskID: taskID})
}

// liveFromContext достаёт (reg, taskID). reg может быть nil — тогда не наблюдаем.
func liveFromContext(ctx context.Context) (*LiveRegistry, int64) {
	v, ok := ctx.Value(liveCtxKey{}).(liveCtxVal)
	if !ok {
		return nil, 0
	}
	return v.reg, v.taskID
}