feat: наблюдение за живой opencode-сессией (/status N показывает что делает агент)

LiveRegistry собирает live-шаги из stdout процесса opencode по taskID
(через контекст, без изменения интерфейса Runner). /status N для running-задачи
прикладывает живое состояние: активный субагент, давность последнего шага,
последние text-шаги агента. Ничего не перезапускается и не убивается —
только наблюдение по запросу.

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

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
+}

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)
+	}
+}

internal/core/core.go

@@ -10,11 +10,20 @@ import (
 
 // Core — ядро Ratatoskr: машина состояний поверх storage.
 type Core struct {
-	Store             *storage.Storage
+	// Store — хранилище задач и трасс.
-	Decide            Decider
+	Store               *storage.Storage
-	MaxTurns          int // D3
+	Decide              Decider
-	MaxConfirmCycles  int // D4
+	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:])
-}
+}

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
+}

internal/opencode/runner.go

@@ -26,12 +26,12 @@ type Result struct {
 
 // Runner — конфигурация запуска opencode-субагентов.
 type Runner struct {
-	Bin      string // путь к opencode (по умолчанию "opencode")
+	Bin          string        // путь к opencode (по умолчанию "opencode")
-	DBPath   string // путь к opencode.db (idle-детекция активности)
+	DBPath       string        // путь к opencode.db (idle-детекция активности)
-	Config   string // путь к opencode.json (OPENCODE_CONFIG)
+	Config       string        // путь к opencode.json (OPENCODE_CONFIG)
-	ConfigDir string // путь к каталогу с агентами (OPENCODE_CONFIG_DIR)
+	ConfigDir    string        // путь к каталогу с агентами (OPENCODE_CONFIG_DIR)
-	IdleTimeout time.Duration // нет новых сообщений в БД → зависание
+	IdleTimeout  time.Duration // нет новых сообщений в БД → зависание
-	HardTimeout 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)
+				}
+			}
 		}
 	}()
 

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=ok  -> мгновенный успех, печатает NDJSON c session_id
-//   $FAKE_MODE=fail-> exit 7 (resume-fallback)
+//	$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)

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
-}
+}

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 сессия в текущей ветке)
+		// 5. запускаем dev-агента (fresh сессия в текущей ветке)
-		res, resErr := w.Runner.Run(ctx, prompt, cwd, w.Agent, "")
+		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
-}
+}