diff --git a/internal/app/app.go b/internal/app/app.go index 27c3c04..d7e49e9 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -103,6 +103,9 @@ func New(configPath, version, updateToken string) (*App, error) { // Core — ядро машины состояний coreCtx := core.New(store, analystCtx) + // Live-журнал живых сессий: воркер пишет шаги агента, /status N их читает. + live := opencode.NewLiveRegistry() + // Router — единый диспетчер входящих из всех каналов a := &App{ Config: cfg, @@ -129,10 +132,14 @@ func New(configPath, version, updateToken string) (*App, error) { MaxJobs: 2, GitBaseURL: cfg.Git.BaseURL, GitToken: cfg.Git.Token, + Live: live, } a.Router = router a.Worker = w + // /status N читает живое состояние сессии через liveProbe-адаптер. + coreCtx.Live = &liveProbe{reg: live} + // Auto-обновление: .new/.old рядом с бинарником (Dir пуст → binDir() от os.Executable). // Токен: вшитый updateToken приоритетнее update.token из конфига. udToken := updateToken diff --git a/internal/app/live.go b/internal/app/live.go new file mode 100644 index 0000000..6c7da4f --- /dev/null +++ b/internal/app/live.go @@ -0,0 +1,46 @@ +package app + +import ( + "fmt" + "strings" + "time" + + "github.com/kamelion/ratatoskr-go/internal/opencode" +) + +// liveProbe адаптирует *opencode.LiveRegistry под интерфейс core.LiveProber. +// Формирует короткое человекочитаемое описание живой сессии задачи. +type liveProbe struct { + reg *opencode.LiveRegistry +} + +func (p *liveProbe) Probe(taskID int64) (string, bool) { + if p == nil || p.reg == nil { + return "", false + } + snap, ok := p.reg.Snap(taskID) + if !ok { + return "", false + } + var b strings.Builder + b.WriteString(fmt.Sprintf("Агент **%s** активен", snap.Agent)) + if idl := time.Since(snap.Last).Round(time.Second); idl > 0 { + b.WriteString(fmt.Sprintf(" (последний шаг %s назад)", idl)) + } + // последние text-шаги — что агент делает прямо сейчас + n := 0 + for i := len(snap.Steps) - 1; i >= 0 && n < 3; i-- { + st := snap.Steps[i] + if st.Type != "text" || strings.TrimSpace(st.Text) == "" { + continue + } + text := strings.TrimSpace(st.Text) + if len(text) > 140 { + text = text[:140] + "…" + } + text = strings.ReplaceAll(text, "\n", " ") + b.WriteString("\n▸ " + text) + n++ + } + return b.String(), true +} diff --git a/internal/app/live_test.go b/internal/app/live_test.go new file mode 100644 index 0000000..91e99a9 --- /dev/null +++ b/internal/app/live_test.go @@ -0,0 +1,39 @@ +package app + +import ( + "strings" + "testing" + "time" + + "github.com/kamelion/ratatoskr-go/internal/opencode" +) + +func TestLiveProbe_Fmt(t *testing.T) { + reg := opencode.NewLiveRegistry() + p := &liveProbe{reg: reg} + + // Нет сессии — Probe должен вернуть ok=false. + if _, ok := p.Probe(1); ok { + t.Error("Probe без сессии вернул ok=true") + } + + reg.Start(7, "dev") + reg.Observe(7, opencode.LiveStep{Type: "text", Text: "анализирую код", At: time.Now()}) + reg.Observe(7, opencode.LiveStep{Type: "tool", At: time.Now()}) + reg.Observe(7, opencode.LiveStep{Type: "text", Text: "пишу фичу по задаче номер триста семь", At: time.Now()}) + + desc, ok := p.Probe(7) + if !ok { + t.Fatal("Probe с сессией вернул ok=false") + } + if !strings.Contains(desc, "dev") { + t.Errorf("описание не содержит агента: %q", desc) + } + if !strings.Contains(desc, "пишу фичу") { + t.Errorf("описание не содержит последний text-шаг: %q", desc) + } + // text-шаги: оба, tools пропущен + if n := strings.Count(desc, "▸"); n != 2 { + t.Errorf("ожидал 2 text-шага (▸), вижу %d: %q", n, desc) + } +} diff --git a/internal/core/core.go b/internal/core/core.go index 96e6871..8f571b8 100644 --- a/internal/core/core.go +++ b/internal/core/core.go @@ -10,11 +10,20 @@ import ( // Core — ядро Ratatoskr: машина состояний поверх storage. type Core struct { - Store *storage.Storage - Decide Decider - MaxTurns int // D3 - MaxConfirmCycles int // D4 + // Store — хранилище задач и трасс. + Store *storage.Storage + Decide Decider + MaxTurns int // D3 + MaxConfirmCycles int // D4 MaxQuestionsPerTurn int + // Live — опциональный просмотр живой сессии задачи (для /status N). + Live LiveProber +} + +// LiveProber — абстракция за журналом живых сессий (реализация — *opencode.LiveRegistry). +// Возвращает текстовое описание текущего состояния сессии задачи и ok=была ли активна. +type LiveProber interface { + Probe(taskID int64) (description string, ok bool) } // New создаёт Core с дефолтными лимитами. @@ -85,7 +94,7 @@ func (c *Core) handleCommand(ctx context.Context, taskID int64, text string) (Re return c.handleContinue(ctx, rest) default: return Result{ - Reply: "Неизвестная команда. Доступно: /start /cancel /skip /retry N /status N", + Reply: "Неизвестная команда. Доступно: /start /cancel /skip /retry N /status N", Action: "send", TaskID: taskID, }, nil @@ -171,7 +180,7 @@ func (c *Core) handleRetry(ctx context.Context, rest string) (Result, error) { id, ok := parseTaskID(rest) if !ok { return Result{ - Reply: "Укажите номер задачи: `/retry 5`.", + Reply: "Укажите номер задачи: `/retry 5`.", Action: "send", }, nil } @@ -199,7 +208,7 @@ func (c *Core) handleStatus(ctx context.Context, rest string) (Result, error) { id, ok := parseTaskID(rest) if !ok { return Result{ - Reply: "Укажите номер задачи: `/status 5`.", + Reply: "Укажите номер задачи: `/status 5`.", Action: "send", }, nil } @@ -207,8 +216,15 @@ func (c *Core) handleStatus(ctx context.Context, rest string) (Result, error) { if err != nil { return c.notFoundReply(ctx, id, err) } + reply := "Задача #" + itoa(id) + ": " + string(task.Status) + // Для выполняемой задачи прикладываем живое состояние сессии (если настроено). + if task.Status == storage.StatusRunning && c.Live != nil { + if desc, ok := c.Live.Probe(id); ok && desc != "" { + reply += "\n" + desc + } + } return Result{ - Reply: "Задача #" + itoa(id) + ": " + string(task.Status), + Reply: reply, Action: "send", TaskID: id, Status: task.Status, @@ -220,7 +236,7 @@ func (c *Core) handleContinue(ctx context.Context, rest string) (Result, error) id, ok := parseTaskID(rest) if !ok { return Result{ - Reply: "Укажите номер задачи: `/continue 5`.", + Reply: "Укажите номер задачи: `/continue 5`.", Action: "send", }, nil } @@ -229,7 +245,7 @@ func (c *Core) handleContinue(ctx context.Context, rest string) (Result, error) return c.notFoundReply(ctx, id, err) } return Result{ - Reply: "Задача #" + itoa(id) + " в статусе " + string(task.Status) + + Reply: "Задача #" + itoa(id) + " в статусе " + string(task.Status) + ". Резюм сессии — пока не реализован.", Action: "send", TaskID: id, @@ -449,4 +465,4 @@ func itoa(n int64) string { buf[i] = '-' } return string(buf[i:]) -} \ No newline at end of file +} diff --git a/internal/opencode/live.go b/internal/opencode/live.go new file mode 100644 index 0000000..d83b768 --- /dev/null +++ b/internal/opencode/live.go @@ -0,0 +1,147 @@ +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 +} diff --git a/internal/opencode/runner.go b/internal/opencode/runner.go index fc999d1..c41a104 100644 --- a/internal/opencode/runner.go +++ b/internal/opencode/runner.go @@ -26,12 +26,12 @@ type Result struct { // 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 // общий лимит на запуск + 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 // Заменяемые для тестов: @@ -143,13 +143,25 @@ func (r *Runner) Run(ctx context.Context, prompt, cwd, agent, sessionID string) var buf []string var mu sync.Mutex done := make(chan struct{}) + // Живое наблюдение сессии (если задано через WithLive в контексте). + liveReg, liveTask := liveFromContext(ctx) + if liveReg != nil && liveTask != 0 { + liveReg.Start(liveTask, agent) + defer liveReg.Finish(liveTask) + } go func() { defer close(done) sc := bufio.NewScanner(stdout) for sc.Scan() { + line := sc.Text() mu.Lock() - buf = append(buf, sc.Text()) + buf = append(buf, line) mu.Unlock() + if liveReg != nil { + if st := parseLiveStep(line); st != nil { + liveReg.Observe(liveTask, *st) + } + } } }() diff --git a/internal/opencode/runner_test.go b/internal/opencode/runner_test.go index 6f8416d..279dccb 100644 --- a/internal/opencode/runner_test.go +++ b/internal/opencode/runner_test.go @@ -9,9 +9,10 @@ import ( ) // fakeOpenCode создаёт shell-скрипт, имитирующий opencode run: -// $FAKE_MODE=ok -> мгновенный успех, печатает NDJSON c session_id -// $FAKE_MODE=slow-> спит долго (для idle/hard timeout) -// $FAKE_MODE=fail-> exit 7 (resume-fallback) +// +// $FAKE_MODE=ok -> мгновенный успех, печатает NDJSON c session_id +// $FAKE_MODE=slow-> спит долго (для idle/hard timeout) +// $FAKE_MODE=fail-> exit 7 (resume-fallback) func fakeOpenCode(t *testing.T, workdir string) string { t.Helper() bin := filepath.Join(workdir, "opencode") @@ -59,6 +60,28 @@ func TestRun_Success(t *testing.T) { } } +func TestRun_LiveRegistry(t *testing.T) { + dir := t.TempDir() + bin := fakeOpenCode(t, dir) + t.Setenv("FAKE_MODE", "ok") + + reg := NewLiveRegistry() + ctx := WithLive(context.Background(), reg, 42) + + r := &Runner{Bin: bin, PollInterval: 20 * time.Millisecond} + res, err := r.Run(ctx, "task", dir, "dev", "") + if err != nil { + t.Fatalf("Run err: %v", err) + } + if res.RC != 0 { + t.Fatalf("RC = %d, want 0", res.RC) + } + // После завершения Finish удаляет сессию → Snap не найден. + if _, ok := reg.Snap(42); ok { + t.Error("сессия не удалена после Finish (должна быть, т.к. задача завершилась)") + } +} + func TestRun_IdleTimeout(t *testing.T) { dir := t.TempDir() bin := fakeOpenCode(t, dir) diff --git a/internal/worker/review.go b/internal/worker/review.go index fc20ce5..7022aef 100644 --- a/internal/worker/review.go +++ b/internal/worker/review.go @@ -68,7 +68,7 @@ func (w *Worker) runReviewer(ctx context.Context, taskID int64, cwd, prompt stri return nil, "", 0, fmt.Errorf("%w: %v", ErrReviewTrace, err) } - res, resErr := w.Runner.Run(ctx, prompt, cwd, "reviewer", "") + res, resErr := w.Runner.Run(w.runCtx(ctx, taskID), prompt, cwd, "reviewer", "") if resErr != nil { w.finalizeTrace(ctx, traceID, storage.TraceFailed, resErr.Error()) return nil, "", traceID, fmt.Errorf("%w: %v", ErrReviewSpawn, resErr) @@ -114,4 +114,4 @@ func parseReviewVerdict(out string) (*reviewVerdict, error) { return nil, err } return &v, nil -} \ No newline at end of file +} diff --git a/internal/worker/worker.go b/internal/worker/worker.go index d945a42..e39f6ab 100644 --- a/internal/worker/worker.go +++ b/internal/worker/worker.go @@ -35,6 +35,10 @@ type Worker struct { GitBaseURL string GitToken string + // Live — опциональный журнал живых сессий (наблюдение /status N). + // Через него Runner пишет live-шаги задачи; nil — наблюдение выключено. + Live *opencode.LiveRegistry + sem chan struct{} // семафор cancel context.CancelFunc @@ -42,6 +46,15 @@ type Worker struct { pollFn PollTaskFunc } +// runCtx оборачивает контекст запуска субагента, привязывая живое наблюдение +// сессии задачи (если Live-журнал включён). Возвращает ctx без изменений при nil Live. +func (w *Worker) runCtx(ctx context.Context, taskID int64) context.Context { + if w.Live == nil { + return ctx + } + return opencode.WithLive(ctx, w.Live, taskID) +} + // Start запускает цикл опроса в фоновой горутине. func (w *Worker) Start(ctx context.Context) { if w.Agent == "" { @@ -195,8 +208,8 @@ func (w *Worker) runTask(ctx context.Context, task *storage.Task) (err error) { return fmt.Errorf("%w: create: %v", ErrTrace, tErr) } - // 5. запускаем dev-агент (fresh сессия в текущей ветке) - res, resErr := w.Runner.Run(ctx, prompt, cwd, w.Agent, "") + // 5. запускаем dev-агента (fresh сессия в текущей ветке) + res, resErr := w.Runner.Run(w.runCtx(ctx, task.ID), prompt, cwd, w.Agent, "") if resErr != nil { // O1 ErrSpawn — не смог запустить бинарь w.failTask(ctx, task) @@ -416,4 +429,4 @@ func validateRepoName(repo string) error { } } return nil -} \ No newline at end of file +}