diff --git a/internal/chat/router.go b/internal/chat/router.go
index b62a9ce..95e355b 100644
--- a/internal/chat/router.go
+++ b/internal/chat/router.go
@@ -24,7 +24,6 @@ type Router struct {
// long-poll цикл канала (Telegram) не блокируется на время долгого
// вызова аналитика и продолжает принимать новые сообщения.
incoming chan Incoming
- wg sync.WaitGroup
}
// NewRouter создаёт роутер. onUserMsg — колбэк обработки входящего.
@@ -47,7 +46,6 @@ func NewRouter(onUserMsg func(Incoming)) *Router {
func (r *Router) processLoop() {
for inc := range r.incoming {
r.onUserMsg(inc)
- r.wg.Done()
}
}
@@ -80,7 +78,6 @@ func (r *Router) handleIncoming(inc Incoming) {
// Асинхронная обработка: кладём событие в очередь воркера и сразу
// возвращаемся, не блокируя вызывающий long-poll цикл канала.
- r.wg.Add(1)
r.incoming <- inc
}
diff --git a/internal/chat/telegram/telegram.go b/internal/chat/telegram/telegram.go
index 072f907..759b31b 100644
--- a/internal/chat/telegram/telegram.go
+++ b/internal/chat/telegram/telegram.go
@@ -16,6 +16,7 @@ import (
"strconv"
"strings"
"time"
+ "unicode/utf8"
"github.com/kamelion/ratatoskr-go/internal/chat"
)
@@ -109,7 +110,7 @@ func (ch *Channel) handleUpdate(ctx context.Context, upd update) {
func (ch *Channel) sendMsg(chatID, text string) error {
body, _ := json.Marshal(map[string]string{
"chat_id": chatID,
- "text": text[:min(len(text), 4000)],
+ "text": truncateUTF8(text, 4000),
"parse_mode": "HTML",
})
url := fmt.Sprintf(ch.apiURL+"sendMessage", ch.token)
@@ -155,10 +156,10 @@ func (ch *Channel) getUpdates(ctx context.Context, offset int64, timeout int) ([
// formatOutgoing собирает Message в HTML-строку: текст + нумерованные Options.
func formatOutgoing(m chat.Message) string {
if len(m.Options) == 0 {
- return m.Text
+ return escapeHTML(m.Text)
}
var buf bytes.Buffer
- buf.WriteString(m.Text)
+ buf.WriteString(escapeHTML(m.Text))
buf.WriteString("\n\n")
for i, opt := range m.Options {
buf.WriteString(fmt.Sprintf("%d. %s\n", i+1, escapeHTML(opt.Label)))
@@ -183,6 +184,18 @@ func escapeHTML(s string) string {
return buf.String()
}
+// truncateUTF8 обрезает s до max байт, не разрывая UTF-8 последовательности.
+func truncateUTF8(s string, max int) string {
+ if len(s) <= max {
+ return s
+ }
+ s = s[:max]
+ for len(s) > 0 && !utf8.ValidString(s) {
+ s = s[:len(s)-1]
+ }
+ return s
+}
+
// ---- Telegram API types ----
type tgResponse struct {
@@ -202,4 +215,4 @@ type message struct {
Chat struct {
ID int64 `json:"id"`
} `json:"chat"`
-}
\ No newline at end of file
+}
diff --git a/internal/chat/telegram/telegram_test.go b/internal/chat/telegram/telegram_test.go
index dd814e1..3552f33 100644
--- a/internal/chat/telegram/telegram_test.go
+++ b/internal/chat/telegram/telegram_test.go
@@ -9,6 +9,7 @@ import (
"sync"
"testing"
"time"
+ "unicode/utf8"
"github.com/kamelion/ratatoskr-go/internal/chat"
)
@@ -111,6 +112,36 @@ func TestSendWithOptions(t *testing.T) {
}
}
+func TestFormatOutgoingEscapesText(t *testing.T) {
+ if got := formatOutgoing(chat.Message{Text: "2 < 3 & 4 > 1"}); got != "2 < 3 & 4 > 1" {
+ t.Errorf("text escape = %q", got)
+ }
+ got := formatOutgoing(chat.Message{
+ Text: "a 0 {
+ userTurns := 0
+ for _, h := range history {
+ if h.Role == "user" {
+ userTurns++
+ }
+ }
+ if userTurns > c.MaxTurns {
+ return Result{
+ Reply: "Превышен лимит ходов сбора (" + itoa(int64(c.MaxTurns)) + "). Используйте /skip чтобы сформулировать черновик, или /start для новой задачи.",
+ TaskID: task.ID,
+ Status: task.Status,
+ }, nil
+ }
+ }
+
msgs := make([]Message, 0, len(history))
for _, h := range history {
msgs = append(msgs, Message{Role: h.Role, Content: h.Content})
@@ -353,7 +356,7 @@ func (c *Core) runDecide(ctx context.Context, task *storage.Task, force bool) (R
if reply == "" {
reply = "Недостаточно данных. Начните заново (/start)."
}
- return Result{Reply: reply, Action: "drop", TaskID: task.ID, Status: task.Status}, nil
+ return Result{Reply: reply, TaskID: task.ID, Status: task.Status}, nil
case "propose", "ready":
// применяем черновик (для ready — текущий, без изменений)
@@ -365,7 +368,7 @@ func (c *Core) runDecide(ctx context.Context, task *storage.Task, force bool) (R
return Result{}, err
}
reply := decChatReply(decision, "Укажи, в каком репозитории(ях) вести работу.")
- return Result{Reply: reply, Action: "send", TaskID: task.ID, Status: task.Status}, nil
+ return Result{Reply: reply, TaskID: task.ID, Status: task.Status}, nil
}
task.Status = storage.StatusReady
if err := c.Store.UpdateTask(ctx, task); err != nil {
@@ -373,7 +376,6 @@ func (c *Core) runDecide(ctx context.Context, task *storage.Task, force bool) (R
}
return Result{
Reply: formatSummary(*task),
- Action: "summary",
TaskID: task.ID,
Status: task.Status,
}, nil
@@ -385,7 +387,7 @@ func (c *Core) runDecide(ctx context.Context, task *storage.Task, force bool) (R
return Result{}, err
}
reply := buildAskReply(decision, c.MaxQuestionsPerTurn)
- return Result{Reply: reply, Action: "send", TaskID: task.ID, Status: task.Status}, nil
+ return Result{Reply: reply, TaskID: task.ID, Status: task.Status}, nil
}
}
diff --git a/internal/core/core_test.go b/internal/core/core_test.go
index 14dd3b9..a475a92 100644
--- a/internal/core/core_test.go
+++ b/internal/core/core_test.go
@@ -2,6 +2,7 @@ package core
import (
"context"
+ "strings"
"testing"
"github.com/kamelion/ratatoskr-go/internal/storage"
@@ -44,13 +45,11 @@ func TestStartCreatesCollecting(t *testing.T) {
c, ctx, store := setupCore(t, nil)
id := mkTask(t, store, ctx, "u1")
- res, err := c.ProcessTurn(ctx, id, "/start")
+ _, err := c.ProcessTurn(ctx, id, "/start")
if err != nil {
t.Fatalf("ProcessTurn /start: %v", err)
}
- if res.Action != "greeting" {
- t.Fatalf("action = %q, want greeting", res.Action)
- }
+
task, _ := store.GetTask(ctx, id)
if task.Status != storage.StatusCollecting {
t.Fatalf("status = %s, want collecting", task.Status)
@@ -61,13 +60,11 @@ func TestCancelSetsCancelled(t *testing.T) {
c, ctx, store := setupCore(t, nil)
id := mkTask(t, store, ctx, "u1")
- res, err := c.ProcessTurn(ctx, id, "/cancel")
+ _, err := c.ProcessTurn(ctx, id, "/cancel")
if err != nil {
t.Fatalf("ProcessTurn /cancel: %v", err)
}
- if res.Action != "drop" {
- t.Fatalf("action = %q, want drop", res.Action)
- }
+
task, _ := store.GetTask(ctx, id)
if task.Status != storage.StatusCancelled {
t.Fatalf("status = %s, want cancelled", task.Status)
@@ -84,13 +81,11 @@ func TestSingleTurnPropose(t *testing.T) {
})
id := mkTask(t, store, ctx, "u1")
- res, err := c.ProcessTurn(ctx, id, "Сделай калькулятор")
+ _, err := c.ProcessTurn(ctx, id, "Сделай калькулятор")
if err != nil {
t.Fatalf("ProcessTurn: %v", err)
}
- if res.Action != "summary" {
- t.Fatalf("action = %q, want summary", res.Action)
- }
+
task, _ := store.GetTask(ctx, id)
if task.Status != storage.StatusReady {
t.Fatalf("status = %s, want ready", task.Status)
@@ -114,9 +109,7 @@ func TestAskReturnsQuestions(t *testing.T) {
if err != nil {
t.Fatalf("ProcessTurn: %v", err)
}
- if res.Action != "send" {
- t.Fatalf("action = %q, want send", res.Action)
- }
+
if res.Reply != "Уточню\n1. Какой язык?\n2. Какой срок?" {
t.Fatalf("reply = %q", res.Reply)
}
@@ -133,13 +126,11 @@ func TestAbortReturnsDrop(t *testing.T) {
})
id := mkTask(t, store, ctx, "u1")
- res, err := c.ProcessTurn(ctx, id, "привет")
+ _, err := c.ProcessTurn(ctx, id, "привет")
if err != nil {
t.Fatalf("ProcessTurn: %v", err)
}
- if res.Action != "drop" {
- t.Fatalf("action = %q, want drop", res.Action)
- }
+
task, _ := store.GetTask(ctx, id)
if task.Status != storage.StatusAborted {
t.Fatalf("status = %s, want aborted", task.Status)
@@ -153,13 +144,11 @@ func TestConsentInReady(t *testing.T) {
id := mkTask(t, store, ctx, "u1")
_, _ = c.ProcessTurn(ctx, id, "сделай задачу")
- res, err := c.ProcessTurn(ctx, id, "создавай")
+ _, err := c.ProcessTurn(ctx, id, "создавай")
if err != nil {
t.Fatalf("ProcessTurn создавай: %v", err)
}
- if res.Action != "created:"+itoa(id) {
- t.Fatalf("action = %q, want created:%d", res.Action, id)
- }
+
task, _ := store.GetTask(ctx, id)
if task.Status != storage.StatusApproved {
t.Fatalf("status после создавай = %s, want approved", task.Status)
@@ -176,13 +165,11 @@ func TestEditInReadyGoesCollecting(t *testing.T) {
_, _ = c.ProcessTurn(ctx, id, "сделай X")
// в ready пишем правку, не согласие
- res, err := c.ProcessTurn(ctx, id, "нет, лучше Y")
+ _, err := c.ProcessTurn(ctx, id, "нет, лучше Y")
if err != nil {
t.Fatalf("ProcessTurn edit: %v", err)
}
- if res.Action != "summary" {
- t.Fatalf("action = %q, want summary", res.Action)
- }
+
if calls != 2 {
t.Fatalf("decide calls = %d, want 2", calls)
}
@@ -194,25 +181,59 @@ func TestEditInReadyGoesCollecting(t *testing.T) {
func TestRetryNotFound(t *testing.T) {
c, ctx, _ := setupCore(t, nil)
- res, err := c.ProcessTurn(ctx, 999, "/retry 999")
+ _, err := c.ProcessTurn(ctx, 999, "/retry 999")
if err != nil {
t.Fatalf("ProcessTurn /retry: %v", err)
}
- if res.Action != "send" {
- t.Fatalf("action = %q, want send", res.Action)
- }
+
}
func TestUnknownCommand(t *testing.T) {
c, ctx, store := setupCore(t, nil)
id := mkTask(t, store, ctx, "u1")
- res, err := c.ProcessTurn(ctx, id, "/bogus")
+ _, err := c.ProcessTurn(ctx, id, "/bogus")
if err != nil {
t.Fatalf("ProcessTurn: %v", err)
}
- if res.Action != "send" {
- t.Fatalf("action = %q, want send", res.Action)
+
+}
+
+func TestMaxTurnsBlocksExcessCollection(t *testing.T) {
+ var calls int
+ c, ctx, store := setupCore(t, func(ctx context.Context, history []Message, draft storage.Task, force bool) (Decision, error) {
+ calls++
+ return Decision{Phase: "ask", ChatReply: "Ещё вопрос"}, nil
+ })
+ c.MaxTurns = 2
+ id := mkTask(t, store, ctx, "u1")
+
+ _, err := c.ProcessTurn(ctx, id, "первый факт")
+ if err != nil {
+ t.Fatalf("1-й ход: %v", err)
+ }
+ _, err = c.ProcessTurn(ctx, id, "второй факт")
+ if err != nil {
+ t.Fatalf("2-й ход: %v", err)
+ }
+ res, err := c.ProcessTurn(ctx, id, "третий факт")
+ if err != nil {
+ t.Fatalf("3-й ход: %v", err)
+ }
+ if calls != 2 {
+ t.Fatalf("decide calls = %d, want 2", calls)
+ }
+ if !strings.Contains(res.Reply, "лимит") {
+ t.Fatalf("reply = %q, want упоминание лимита", res.Reply)
+ }
+
+ // /skip — принудительный вызов, лимит не мешает
+ _, err = c.ProcessTurn(ctx, id, "/skip")
+ if err != nil {
+ t.Fatalf("/skip: %v", err)
+ }
+ if calls != 3 {
+ t.Fatalf("decide calls after /skip = %d, want 3", calls)
}
}
@@ -229,4 +250,4 @@ func TestParseTaskID(t *testing.T) {
if _, ok := parseTaskID("abc"); ok {
t.Fatal("parseTaskID(abc) should be invalid")
}
-}
\ No newline at end of file
+}
diff --git a/internal/core/decider.go b/internal/core/decider.go
index 235e096..58770e5 100644
--- a/internal/core/decider.go
+++ b/internal/core/decider.go
@@ -15,10 +15,10 @@ type Message struct {
// Decision — вердикт аналитика.
type Decision struct {
- Phase string // "ask" | "propose" | "ready" | "abort"
- Draft storage.Task // обновлённые поля черновика
- Questions []string // вопросы для phase=ask
- ChatReply string // ответ пользователю
+ Phase string // "ask" | "propose" | "ready" | "abort"
+ Draft storage.Task // обновлённые поля черновика
+ Questions []string // вопросы для phase=ask
+ ChatReply string // ответ пользователю
}
// Decider — интерфейс для вызова аналитика (opencode).
@@ -29,7 +29,6 @@ type Decider interface {
// Result — результат одного хода.
type Result struct {
Reply string
- Action string // send | summary | created:N | abort | drop | greeting
TaskID int64
Status storage.Status
}
@@ -46,4 +45,4 @@ func isConsent(text string) bool {
}
}
return false
-}
\ No newline at end of file
+}
diff --git a/internal/opencode/extract.go b/internal/opencode/extract.go
index 721e0a9..d13f7ea 100644
--- a/internal/opencode/extract.go
+++ b/internal/opencode/extract.go
@@ -4,7 +4,6 @@
// Контракты перенесены 1-в-1 из Python-версии (extract.py / opencode.py):
// - ExtractVerdict: последний text-парт из NDJSON-потока opencode run --format json
// - ExtractJSON: fenced ```json``` → первый {...}
-// - Run/ResumeDev: запуск процесса с idle/hard timeout по opencode.db
package opencode
import (
@@ -14,9 +13,8 @@ import (
)
var (
- fenceRe = regexp.MustCompile("```(?:json)?\\s*([\\s\\S]*?)```")
- jsonBlockRe = regexp.MustCompile("\\{[\\s\\S]*\\}")
- sessionRe = regexp.MustCompile(`"session_id"\s*:\s*"([^"]+)"`)
+ fenceRe = regexp.MustCompile("```(?:json)?\\s*([\\s\\S]*?)```")
+ jsonBlockRe = regexp.MustCompile("\\{[\\s\\S]*\\}")
)
// ExtractVerdict возвращает текст вердикта из NDJSON-потока opencode run --format json.
@@ -76,15 +74,6 @@ func ExtractJSON(text string) (map[string]json.RawMessage, bool) {
return nil, false
}
-// SessionIDFromOutput извлекает session_id из текстового вывода opencode.
-func SessionIDFromOutput(out string) (string, bool) {
- m := sessionRe.FindStringSubmatch(out)
- if len(m) > 1 && m[1] != "" {
- return m[1], true
- }
- return "", false
-}
-
// stripFence обрезает внешние ```json``` (или ```) ограждения вокруг фрагмента.
// Используется для вердиктов, которые модель может вернуть в markdown-фенсе.
func stripFence(s string) string {
diff --git a/internal/opencode/extract_test.go b/internal/opencode/extract_test.go
index 5a206ee..ca051db 100644
--- a/internal/opencode/extract_test.go
+++ b/internal/opencode/extract_test.go
@@ -104,12 +104,3 @@ func TestExtractJSON(t *testing.T) {
})
}
}
-
-func TestSessionIDFromOutput(t *testing.T) {
- if s, ok := SessionIDFromOutput(`{"session_id":"abc123"}`); !ok || s != "abc123" {
- t.Fatalf("got %q %v", s, ok)
- }
- if _, ok := SessionIDFromOutput("no session here"); ok {
- t.Fatal("expected no match")
- }
-}
diff --git a/internal/opencode/live.go b/internal/opencode/live.go
index d83b768..ce0dbcd 100644
--- a/internal/opencode/live.go
+++ b/internal/opencode/live.go
@@ -2,36 +2,11 @@ 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, не из БД.
+// LiveStep — один наблюдаемый шаг агента.
type LiveStep struct {
Type string // "text" | "tool" | "agent" | ...
Text string // содержимое text-парта (для других типов может быть пустым)
diff --git a/internal/opencode/pgid_linux.go b/internal/opencode/pgid_linux.go
deleted file mode 100644
index bbc14f6..0000000
--- a/internal/opencode/pgid_linux.go
+++ /dev/null
@@ -1,23 +0,0 @@
-//go:build linux
-
-package opencode
-
-import (
- "os/exec"
- "syscall"
-)
-
-func sysProcAttr(proc *exec.Cmd) {
- proc.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
-}
-
-// killProcGroup убивает всю process-group по лидеру pid (SIGKILL дочерним и
-// SIGTERM лидеру). Игнорирует ошибки: weakest-effort teardown.
-func killProcGroup(pid int) {
- pgid, err := syscall.Getpgid(pid)
- if err != nil {
- return
- }
- _ = syscall.Kill(-pgid, syscall.SIGKILL)
- _ = syscall.Kill(pid, syscall.SIGKILL)
-}
diff --git a/internal/opencode/pgid_other.go b/internal/opencode/pgid_other.go
deleted file mode 100644
index d087d29..0000000
--- a/internal/opencode/pgid_other.go
+++ /dev/null
@@ -1,9 +0,0 @@
-//go:build !linux
-
-package opencode
-
-import "os/exec"
-
-func sysProcAttr(_ *exec.Cmd) {}
-
-func killProcGroup(pid int) {}
diff --git a/internal/opencode/runner.go b/internal/opencode/runner.go
index 37d9d20..c8a99f6 100644
--- a/internal/opencode/runner.go
+++ b/internal/opencode/runner.go
@@ -164,22 +164,4 @@ func (r *Runner) awaitVerdict(ctx context.Context, c *Client, sid, agent, prompt
case <-ctx.Done():
}
}
-}
-
-// ResumeDev — запуск dev-агента с resume-fallback. Если resume (sessionID)
-// падает с rc!=0 — повторяем ОДИН раз свежей сессией в том же каталоге.
-// rc=-1 (обрыв по таймауту) НЕ триггерит 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 {
- return res, false
- }
- if res.RC != 0 && res.RC != -1 && sessionID != "" {
- r.logf("dev resume rc=%d — запускаю заново свежей сессией (каталог сохраняю)", res.RC)
- res, _ = r.Run(ctx, prompt+resumeFallbackNote, cwd, "dev", "")
- }
- return res, res.RC == -1
-}
-
-const resumeFallbackNote = "\n\n(Возобновление сессии не удалось; продолжи с учётом уже сделанных изменений в worktree.)"
\ No newline at end of file
+}
\ No newline at end of file
diff --git a/internal/opencode/runner_test.go b/internal/opencode/runner_test.go
index d890f62..f02a172 100644
--- a/internal/opencode/runner_test.go
+++ b/internal/opencode/runner_test.go
@@ -93,26 +93,6 @@ func TestRun_ContextCancel(t *testing.T) {
}
}
-// TestResumeDev_Fallback: resume (sessionID) "падает" rc!=0 только когда самого
-// сервера нет; в фейке такого нет, поэтому проверяем, что при успехе
-// fallback не срабатывает и таймаут не выставляется.
-func TestResumeDev_NoFallbackOnSuccess(t *testing.T) {
- dir := t.TempDir()
- f := &fakeAPIServer{
- verdictParts: []part{{Type: "text", Text: "ok"}},
- }
- p, _ := fakePool(t, f, dir)
-
- r := &Runner{Pool: p, PollInterval: 5 * time.Millisecond}
- res, timedOut := r.ResumeDev(context.Background(), "task", dir, "lost-session")
- if timedOut {
- t.Error("timedOut = true, want false (успех не должен считаться таймаутом)")
- }
- if res.RC != 0 {
- t.Errorf("RC = %d, want 0", res.RC)
- }
-}
-
func contains(s, sub string) bool {
return len(s) >= len(sub) && (s == sub || len(s) > 0 && indexOf(s, sub) >= 0)
}