3 Commits

Author SHA1 Message Date
ki.sagidullin
be4d749c45 fix(opencode): считать реальный прогресс стрима для idle-детекции
All checks were successful
CI / test (pull_request) Successful in 1m9s
CI / build-and-package (amd64, linux) (pull_request) Successful in 53s
CI / build-and-package (amd64, windows) (pull_request) Successful in 52s
Растущий в один text-парт стрим (text-delta) и reasoning больше не
выглядят как зависшая нейронка: idle-таймер сбрасывается по росту
числа партов и суммарной длины text/reasoning.
2026-08-19 19:11:26 +05:00
ki.sagidullin
b5fb583c90 test: починить тесты на Windows
All checks were successful
CI / test (pull_request) Successful in 41s
CI / build-and-package (amd64, linux) (pull_request) Successful in 37s
CI / build-and-package (amd64, windows) (pull_request) Successful in 34s
- E2E (app): e2eFakeAPI переведён на v2 HTTP API opencode (/api/*) с
  определением агента по тексту промпта; Router получает processed-счётчик
  и WaitProcessed, e2eChannel.deliver ждёт асинхронную обработку — убирает
  гонку «запрос сразу после deliver» и коллатеральный 'database is closed'.
- app_test: одинарные YAML-кавычки для путей Windows (backslash-escape) +
  закрытие Store в TestNew/TestNew_RunCtxCancel/TestNew_UpdateWiring.
- config_test: абсолютный путь строится с корнем тома (C:\...) и одинарными
  кавычками YAML.
- opencode/server_test: fakeServeBin на Windows — .cmd с ping (#!/bin/sh
  не исполняется).
- worker_test: TestWorkerSemaphore поллит до целевого статуса вместо
  фиксированных sleep (git на Windows медленнее).
2026-08-19 10:47:55 +05:00
3c8528dbd9 Merge pull request 'feat(opencode): переход на v2 HTTP API opencode (хардпин модели, поллинг вердикта)' (#5) from feat/ad025c1 into main
Some checks failed
CI / test (push) Failing after 30s
CI / build-and-package (amd64, linux) (push) Successful in 49s
CI / build-and-package (amd64, windows) (push) Successful in 52s
Reviewed-on: http://gitea.hal9000.home/kamelion/ratatoskr-go/pulls/5
2026-08-19 08:55:16 +05:00
11 changed files with 344 additions and 104 deletions

View File

@@ -1,39 +1,6 @@
# the name by which the project can be referenced within Serena # the name by which the project can be referenced within Serena/when chatting with the LLM.
project_name: "ratatoskr-go" project_name: "ratatoskr-go"
# list of languages for which language servers are started; choose from:
# al angular ansible bash clojure
# cpp cpp_ccls crystal csharp csharp_omnisharp
# dart elixir elm erlang fortran
# fsharp go groovy haskell haxe
# hlsl html java json julia
# kotlin lean4 lua luau markdown
# matlab msl nix ocaml pascal
# perl php php_phpactor powershell python
# python_jedi python_ty r rego ruby
# ruby_solargraph rust scala scss solidity
# svelte swift systemverilog terraform toml
# typescript typescript_vts vue yaml zig
# (This list may be outdated. For the current list, see values of Language enum here:
# https://github.com/oraios/serena/blob/main/src/solidlsp/ls_config.py
# For some languages, there are alternative language servers, e.g. csharp_omnisharp, ruby_solargraph.)
# Note:
# - For C, use cpp
# - For JavaScript, use typescript
# - For Angular projects, use angular (subsumes typescript+html; requires `npm install` in the project root)
# - For Svelte projects, use svelte (subsumes typescript/javascript for .svelte projects; requires npm)
# - For SCSS / Sass / plain CSS, use scss (some-sass-language-server handles all three)
# - For Free Pascal/Lazarus, use pascal
# Special requirements:
# Some languages require additional setup/installations.
# See here for details: https://oraios.github.io/serena/01-about/020_programming-languages.html#language-servers
# When using multiple languages, the first language server that supports a given file will be used for that file.
# The first language is the default language and the respective language server will be used as a fallback.
# Note that when using the JetBrains backend, language servers are not used and this list is correspondingly ignored.
languages:
- go
# the encoding used by text files in the project # the encoding used by text files in the project
# For a list of possible encodings, see https://docs.python.org/3.11/library/codecs.html#standard-encodings # For a list of possible encodings, see https://docs.python.org/3.11/library/codecs.html#standard-encodings
encoding: "utf-8" encoding: "utf-8"
@@ -55,23 +22,19 @@ ignore_all_files_in_gitignore: true
# advanced configuration option allowing to configure language server-specific options. # advanced configuration option allowing to configure language server-specific options.
# Maps the language key to the options. # Maps the language key to the options.
# Have a look at the docstring of the constructors of the LS implementations within solidlsp (e.g., for C# or PHP) to see which options are available. # The settings are considered only if the project is trusted (see global configuration to define trusted projects).
# No documentation on options means no options are available. # See https://oraios.github.io/serena/02-usage/050_configuration.html#language-server-specific-settings
ls_specific_settings: {} ls_specific_settings: {}
# list of additional workspace folder paths for cross-package reference support (e.g. in monorepos).
# Paths can be absolute or relative to the project root.
# Each folder is registered as an LSP workspace folder, enabling language servers to discover
# symbols and references across package boundaries.
# Currently supported for: TypeScript.
# Example:
# additional_workspace_folders:
# - ../sibling-package
# - ../shared-lib
additional_workspace_folders: []
# list of additional paths to ignore in this project. # list of additional paths to ignore in this project.
# Same syntax as gitignore, so you can use * and **. # Same syntax as gitignore, so you can use * and **.
# Important: quote patterns that start with `*`, otherwise YAML treats them as aliases.
# Example:
# ignored_paths:
# - "examples/**"
# - ".worktrees/**"
# - "**/bin/**"
# - "**/obj/**"
# Note: global ignored_paths from serena_config.yml are also applied additively. # Note: global ignored_paths from serena_config.yml are also applied additively.
ignored_paths: [] ignored_paths: []
@@ -131,3 +94,76 @@ read_only_memory_patterns: []
# Extends the list from the global configuration, merging the two lists. # Extends the list from the global configuration, merging the two lists.
# Example: ["_archive/.*", "_episodes/.*"] # Example: ["_archive/.*", "_episodes/.*"]
ignored_memory_patterns: [] ignored_memory_patterns: []
# list of additional workspace folder paths for cross-package reference support.
# Paths can be absolute or relative to the project root.
# Each folder is registered as an LSP workspace folder, enabling language servers to discover
# symbols and references across package boundaries, but these folders are not indexed by Serena,
# i.e. the respective symbols will not be found using Serena's symbol search tools.
# Example:
# additional_workspace_folders:
# - ../sibling-package
# - ../shared-lib
ls_additional_workspace_folders: []
# list of language servers to start when using the LSP backend; choose from:
# ada al angular ansible bash
# bsl clojure cpp cpp_ccls crystal
# csharp csharp_omnisharp cue dart deno
# elixir elm erlang fortran fsharp
# gdscript gleam go groovy haskell
# haxe hlsl html java json
# julia kotlin latex lean4 lua
# luau markdown matlab msl nextflow
# nix ocaml pascal perl php
# php_phpactor php_phpantom powershell python python_basedpyright
# python_jedi python_pyrefly python_ty qml r
# rego ruby ruby_solargraph rust scala
# scss solidity svelte swift systemverilog
# terraform toml typescript typescript_vts vue
# wolfram yaml zig
# (This list may be outdated; generated with scripts/print_language_list.py;
# For the current list, see values of the LanguageServerId enum here:
# https://github.com/oraios/serena/blob/main/src/solidlsp/ls_config.py)
# For some languages, there are several alternative language servers, e.g. csharp_omnisharp, ruby_solargraph.)
# Note:
# - For C, use cpp
# - For JavaScript, use typescript
# - For Angular projects, use angular (subsumes typescript+html; requires `npm install` in the project root)
# - For Svelte projects, use svelte (subsumes typescript/javascript for .svelte projects; requires npm)
# - For Deno projects, use deno (serves the same .ts/.js files as typescript; requires the deno CLI on PATH)
# - For SCSS / Sass / plain CSS, use scss (some-sass-language-server handles all three)
# - For Free Pascal/Lazarus, use pascal
# Special requirements:
# Some language servers require additional setup/installations.
# See here for details: https://oraios.github.io/serena/01-about/020_programming-languages.html#language-servers
# When using multiple language servers, the first language server that supports a given file will be used for that file.
# The first language server is the default language and the respective language server will be used as a fallback.
# Note that when using the JetBrains backend, language servers are not used and this list is correspondingly ignored.
language_servers:
- go
# list of workspace folder paths (LSP backend only).
# These folders will be used to build up Serena's symbol index.
# Paths must be within the project root and should thus be relative to the project root.
# Furthermore, the paths should not be filtered by ignore settings.
# Default setting: The entire project root folder (".") is considered.
# In (large) monorepos, this can be used to index only subfolders of the project root, e.g.
# ls_workspace_folders:
# - "./subproject1"
# - "./subproject2"
ls_workspace_folders:
- .
# optional shell command to run before the language backend (LSP or JetBrains) is initialised.
# the command runs in the project root directory and is only executed if the project is trusted
# (see trusted_project_path_patterns in the global configuration).
# serena waits for the command to exit: a non-zero exit code is logged as an error but does not
# abort activation. a per-project timeout (activation_command_timeout, default 180s) is the safety
# backstop for non-terminating commands; on expiry the process is killed and activation continues.
# example: activation_command: "npx nx run-many -t build"
activation_command:
# maximum time in seconds to wait for activation_command to complete before killing it (default 180s).
# must be a positive number.
activation_command_timeout: 180.0

View File

@@ -25,6 +25,7 @@ func TestNew(t *testing.T) {
if err != nil { if err != nil {
t.Fatalf("New() err = %v", err) t.Fatalf("New() err = %v", err)
} }
defer a.Store.Close()
if a.Store == nil { if a.Store == nil {
t.Fatal("Store не создан") t.Fatal("Store не создан")
} }
@@ -87,6 +88,7 @@ func TestNew_RunCtxCancel(t *testing.T) {
if err != nil { if err != nil {
t.Fatalf("New() err = %v", err) t.Fatalf("New() err = %v", err)
} }
defer a.Store.Close()
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(context.Background())
cancel() // сразу отменяем cancel() // сразу отменяем
@@ -120,8 +122,8 @@ func TestNew_WorktreeCreated(t *testing.T) {
" token: \"test:token\"", " token: \"test:token\"",
" chat_id: \"12345\"", " chat_id: \"12345\"",
"paths:", "paths:",
" db: \"" + filepath.Join(tmp, "test.db") + "\"", " db: '" + filepath.Join(tmp, "test.db") + "'",
" worktree: \"" + wt + "\"", " worktree: '" + wt + "'",
"", "",
}, "\n") }, "\n")
if err := os.WriteFile(configPath, []byte(content), 0o600); err != nil { if err := os.WriteFile(configPath, []byte(content), 0o600); err != nil {
@@ -165,7 +167,7 @@ func TestNew_UpdateWiring(t *testing.T) {
" base_url: \"https://hub.example.com\"", " base_url: \"https://hub.example.com\"",
" token: \"cfg-update-token\"", " token: \"cfg-update-token\"",
"paths:", "paths:",
" db: \"" + dbPath + "\"", " db: '" + dbPath + "'",
"", // пустая строка в конце "", // пустая строка в конце
}, "\n") }, "\n")
if err := os.WriteFile(configPath, []byte(content), 0o600); err != nil { if err := os.WriteFile(configPath, []byte(content), 0o600); err != nil {
@@ -177,6 +179,7 @@ func TestNew_UpdateWiring(t *testing.T) {
if err != nil { if err != nil {
t.Fatalf("New() err = %v", err) t.Fatalf("New() err = %v", err)
} }
defer a.Store.Close()
if a.Updater == nil { if a.Updater == nil {
t.Fatal("Updater не создан") t.Fatal("Updater не создан")
} }
@@ -197,6 +200,7 @@ func TestNew_UpdateWiring(t *testing.T) {
if err != nil { if err != nil {
t.Fatalf("New() err = %v", err) t.Fatalf("New() err = %v", err)
} }
defer a2.Store.Close()
if a2.Updater.Token != "embedded-update-token" { if a2.Updater.Token != "embedded-update-token" {
t.Errorf("Token = %q, want embedded-update-token (вшитый приоритетнее)", a2.Updater.Token) t.Errorf("Token = %q, want embedded-update-token (вшитый приоритетнее)", a2.Updater.Token)
} }

View File

@@ -36,22 +36,35 @@ import (
) )
// вердикты фейкового агента по имени. // вердикты фейкового агента по имени.
var ( var e2eAgentVerdicts = map[string]string{
e2eAgentVerdicts = map[string]string{
"analyst": `{"phase":"propose","title":"Калькулятор","goal":"Сделать веб-калькулятор","repo":"calc","why":"Нужен для учёта","ac":"Работает + - * /","chat_reply":"Черновик готов."}`, "analyst": `{"phase":"propose","title":"Калькулятор","goal":"Сделать веб-калькулятор","repo":"calc","why":"Нужен для учёта","ac":"Работает + - * /","chat_reply":"Черновик готов."}`,
"dev": `done`, "dev": `done`,
"reviewer": `{"passed":true,"comments":[]}`, "reviewer": `{"passed":true,"comments":[]}`,
} }
)
// e2eFakeAPI поднимает фейковый opencode serve experimental HTTP API (пути // e2eFakeAPI поднимает фейковый opencode serve, эмулирующий v2 HTTP API
// БЕЗ /api) и возвращает URL. По title сессии (ratatoskr-<agent>) определяет // (пути с префиксом /api/*, см. Client в internal/opencode). Агент
// агента и возвращает его вердикт как text-часть ответа на POST /message. // (analyst/dev/reviewer) определяется по тексту промпта на POST
// /api/session/{id}/prompt; вердикт возвращается как text-часть завершённого
// assistant-сообщения, которое отдаёт GET /api/session/{id}/message.
func e2eFakeAPI(t *testing.T) string { func e2eFakeAPI(t *testing.T) string {
t.Helper() t.Helper()
var mu sync.Mutex var mu sync.Mutex
sessions := map[string]string{} // id → agent sessions := map[string]string{} // id → agent
agentOf := func(prompt string) string {
switch {
case strings.Contains(prompt, "Ты — аналитик"):
return "analyst"
case strings.Contains(prompt, "Ты — dev-агент"):
return "dev"
case strings.Contains(prompt, "Ты — ревьюер"):
return "reviewer"
default:
return "unknown"
}
}
verdictFor := func(agent string) string { verdictFor := func(agent string) string {
if v, ok := e2eAgentVerdicts[agent]; ok { if v, ok := e2eAgentVerdicts[agent]; ok {
return v return v
@@ -59,39 +72,59 @@ func e2eFakeAPI(t *testing.T) string {
return "unknown agent" return "unknown agent"
} }
sessionID := func(path, suffix string) string {
return strings.TrimSuffix(strings.TrimPrefix(path, "/api/session/"), suffix)
}
assistantMsg := func(id, agent string) map[string]any {
ts := time.Now().UnixMilli()
return map[string]any{
"id": "m-" + id,
"type": "assistant",
"content": []map[string]any{{"type": "text", "text": verdictFor(agent)}},
"model": map[string]any{"providerID": "test", "id": "m"},
"time": map[string]any{"created": ts, "completed": ts},
}
}
h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch { switch {
case r.Method == http.MethodPost && r.URL.Path == "/session": case r.Method == http.MethodPost && r.URL.Path == "/api/session":
// v2 create: {model:{...}} → {data:{id}}
id := fmt.Sprintf("e2e-%d", len(sessions)+1)
mu.Lock()
sessions[id] = ""
mu.Unlock()
writeJSON(w, map[string]any{"data": map[string]any{"id": id}})
case r.Method == http.MethodPost && strings.HasSuffix(r.URL.Path, "/prompt"):
// v2 durable admit: {prompt:{text}} → {data:{id,timeCreated}}
id := sessionID(r.URL.Path, "/prompt")
var req struct { var req struct {
Title string `json:"title"` Prompt struct {
Text string `json:"text"`
} `json:"prompt"`
} }
_ = json.NewDecoder(r.Body).Decode(&req) _ = json.NewDecoder(r.Body).Decode(&req)
agent := strings.TrimPrefix(req.Title, "ratatoskr-")
mu.Lock() mu.Lock()
id := fmt.Sprintf("e2e-%d", len(sessions)+1) sessions[id] = agentOf(req.Prompt.Text)
sessions[id] = agent
mu.Unlock() mu.Unlock()
// experimental: голая Session, id напрямую. writeJSON(w, map[string]any{"data": map[string]any{"id": "p-" + id, "timeCreated": time.Now().UnixMilli()}})
writeJSON(w, map[string]any{"id": id, "agent": agent, "model": map[string]any{"id": "m"}})
case r.Method == http.MethodPost && strings.HasSuffix(r.URL.Path, "/message"):
// блокирующий ответ: вердикт как text-часть.
id := strings.TrimSuffix(strings.TrimPrefix(r.URL.Path, "/session/"), "/message")
mu.Lock()
agent := sessions[id]
mu.Unlock()
writeJSON(w, map[string]any{"info": map[string]any{"role": "assistant"}, "parts": []map[string]any{{"type": "text", "text": verdictFor(agent)}}})
case r.Method == http.MethodGet && strings.HasSuffix(r.URL.Path, "/message"): case r.Method == http.MethodGet && strings.HasSuffix(r.URL.Path, "/message"):
// поллинг прогресса: голый массив [{info, parts}]. // v2 поллинг: {data:[Session.Message]} (новейшие первыми).
id := strings.TrimSuffix(strings.TrimPrefix(r.URL.Path, "/session/"), "/message") id := sessionID(r.URL.Path, "/message")
mu.Lock() mu.Lock()
agent := sessions[id] agent := sessions[id]
mu.Unlock() mu.Unlock()
writeJSON(w, []map[string]any{{"info": map[string]any{"role": "assistant"}, "parts": []map[string]any{{"type": "text", "text": verdictFor(agent)}}}}) writeJSON(w, map[string]any{"data": []map[string]any{assistantMsg(id, agent)}})
case r.Method == http.MethodPost && strings.HasSuffix(r.URL.Path, "/abort"): case r.Method == http.MethodGet && r.URL.Path == "/api/session/active":
writeJSON(w, map[string]any{"ok": true}) // сессий в активных дренажах нет → ответ завершён.
writeJSON(w, map[string]any{"data": map[string]any{}})
case r.Method == http.MethodPost && strings.HasSuffix(r.URL.Path, "/interrupt"):
writeJSON(w, map[string]any{"data": map[string]any{"ok": true}})
default: default:
http.NotFound(w, r) http.NotFound(w, r)
@@ -143,6 +176,7 @@ func e2eAssemble(t *testing.T) (*App, string, *e2eChannel) {
CoreCtx: coreCtx, CoreCtx: coreCtx,
} }
a.Router = chat.NewRouter(a.handleIncoming) a.Router = chat.NewRouter(a.handleIncoming)
fake.router = a.Router
if err := a.Router.Attach(fake); err != nil { if err := a.Router.Attach(fake); err != nil {
t.Fatalf("Attach fake channel: %v", err) t.Fatalf("Attach fake channel: %v", err)
} }
@@ -168,6 +202,7 @@ func e2eAssemble(t *testing.T) (*App, string, *e2eChannel) {
type e2eChannel struct { type e2eChannel struct {
onMsg chat.Handler onMsg chat.Handler
sent []chat.Message sent []chat.Message
router *chat.Router
} }
func (c *e2eChannel) Run(_ context.Context) error { return nil } func (c *e2eChannel) Run(_ context.Context) error { return nil }
@@ -184,10 +219,21 @@ func (c *e2eChannel) Ask(_ context.Context, _ chat.Address, m chat.Message) erro
func (c *e2eChannel) Close() error { return nil } func (c *e2eChannel) Close() error { return nil }
// deliver отправляет входящее сообщение через роутер: ставит маршрут // deliver отправляет входящее сообщение через роутер: ставит маршрут
// пользователя и вызывает app.handleIncoming (как в проде). // пользователя и вызывает app.handleIncoming (как в проде). Так как роутер
// обрабатывает входящие асинхронно (процессор-горутина), deliver ждёт, пока
// обработка события завершится, — иначе тесты (сразу читающие состояние БД)
// гоняются с обработчиком.
func (c *e2eChannel) deliver(uid chat.UserID, text string) { func (c *e2eChannel) deliver(uid chat.UserID, text string) {
if c.onMsg != nil { if c.onMsg == nil {
return
}
target := int64(0)
if c.router != nil {
target = c.router.Processed() + 1
}
c.onMsg(chat.Incoming{UserID: uid, Address: chat.Address("u://" + string(uid)), Msg: chat.Message{Text: text}, Channel: c}) c.onMsg(chat.Incoming{UserID: uid, Address: chat.Address("u://" + string(uid)), Msg: chat.Message{Text: text}, Channel: c})
if c.router != nil && !c.router.WaitProcessed(target) {
panic("e2e: роутер не обработал входящее за 30s")
} }
} }

View File

@@ -4,6 +4,8 @@ import (
"context" "context"
"fmt" "fmt"
"sync" "sync"
"sync/atomic"
"time"
) )
// Router — единый диспетчер входящих из всех каналов и маршрутизатор исходящих. // Router — единый диспетчер входящих из всех каналов и маршрутизатор исходящих.
@@ -24,6 +26,10 @@ type Router struct {
// long-poll цикл канала (Telegram) не блокируется на время долгого // long-poll цикл канала (Telegram) не блокируется на время долгого
// вызова аналитика и продолжает принимать новые сообщения. // вызова аналитика и продолжает принимать новые сообщения.
incoming chan Incoming incoming chan Incoming
// processed — число обработанных воркером событий (для синхронизации
// тестов с асинхронной очередью: WaitProcessed ждёт обработку события).
processed atomic.Int64
} }
// NewRouter создаёт роутер. onUserMsg — колбэк обработки входящего. // NewRouter создаёт роутер. onUserMsg — колбэк обработки входящего.
@@ -46,9 +52,26 @@ func NewRouter(onUserMsg func(Incoming)) *Router {
func (r *Router) processLoop() { func (r *Router) processLoop() {
for inc := range r.incoming { for inc := range r.incoming {
r.onUserMsg(inc) r.onUserMsg(inc)
r.processed.Add(1)
} }
} }
// Processed возвращает число обработанных воркером входящих событий.
func (r *Router) Processed() int64 { return r.processed.Load() }
// WaitProcessed ждёт, пока воркер обработает не меньше target событий
// (для синхронизации с асинхронной очередью в тестах).
func (r *Router) WaitProcessed(target int64) bool {
deadline := time.Now().Add(30 * time.Second)
for r.processed.Load() < target {
if time.Now().After(deadline) {
return false
}
time.Sleep(2 * time.Millisecond)
}
return true
}
// Attach регистрирует канал и подключает его к обработчику входящих. // Attach регистрирует канал и подключает его к обработчику входящих.
// Возвращает ошибку только при пустом канале (nil). // Возвращает ошибку только при пустом канале (nil).
func (r *Router) Attach(ch Channel) error { func (r *Router) Attach(ch Channel) error {

View File

@@ -251,14 +251,17 @@ func TestResolveExePaths_AbsoluteKept(t *testing.T) {
t.Setenv("TG_TOKEN", "tok") t.Setenv("TG_TOKEN", "tok")
t.Setenv("TG_CHAT_ID", "42") t.Setenv("TG_CHAT_ID", "42")
absDB := filepath.Join(string(filepath.Separator), "data", "ratatoskr.db") // абсолютный для текущей ОС // абсолютные пути «для текущей ОС»: на Windows слэш-относительный путь
absWt := filepath.Join(string(filepath.Separator), "worktrees") // (\data\...) НЕ является абсолютным — нужен корень тома (C:\data\...).
root := filepath.VolumeName(os.TempDir()) + string(filepath.Separator)
absDB := filepath.Join(root, "data", "ratatoskr.db")
absWt := filepath.Join(root, "worktrees")
yaml := `telegram: yaml := `telegram:
username: "${TG_TOKEN}" username: "${TG_TOKEN}"
chat_id: "${TG_CHAT_ID}" chat_id: "${TG_CHAT_ID}"
paths: paths:
db: "` + absDB + `" db: '` + absDB + `'
worktree: "` + absWt + `" worktree: '` + absWt + `'
` `
cfg, err := Load(writeCfg(t, yaml)) cfg, err := Load(writeCfg(t, yaml))
if err != nil { if err != nil {

View File

@@ -292,6 +292,22 @@ func newestAssistant(msgs []v2Message, since int64) (*v2Message, int) {
return newest, count return newest, count
} }
// progressOf — «живой» прогресс новых assistant-сообщений: число контент-партов
// (text/reasoning/tool) + суммарная длина их текста. Растёт во время стриминга,
// когда один и тот же парт увеличивается (и при reasoning), — это и есть
// сигнал, что LLM работает, а не висит.
func progressOf(msgs []v2Message, since int64) (parts, textLen int) {
for _, m := range assistantSince(msgs, since) {
for _, p := range m.Content {
parts++
if p.Type == "text" || p.Type == "reasoning" {
textLen += len(p.Text)
}
}
}
return
}
// assistantText объединяет text-парты новых assistant-сообщений в хронологическом // assistantText объединяет text-парты новых assistant-сообщений в хронологическом
// порядке (сообщения приходят новейшими первыми → идём с конца). // порядке (сообщения приходят новейшими первыми → идём с конца).
func assistantText(msgs []v2Message, since int64) []string { func assistantText(msgs []v2Message, since int64) []string {

View File

@@ -6,6 +6,7 @@ import (
"errors" "errors"
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"strings"
"testing" "testing"
"time" "time"
) )
@@ -16,7 +17,10 @@ import (
// - нормальный: Prompt ставит active=false и в messages кладётся финальное // - нормальный: Prompt ставит active=false и в messages кладётся финальное
// assistant-сообщение (verdictText) → Runner собирает вердикт; // assistant-сообщение (verdictText) → Runner собирает вердикт;
// - blockPrompt: «агент завис» — active=true всегда, сообщений нет → idle abort; // - blockPrompt: «агент завис» — active=true всегда, сообщений нет → idle abort;
// - failCreate / failMessages — имитация ошибок. // - failCreate / failMessages — имитация ошибок;
// - growStream: стрим одного растущего парта — текст/reasoning растёт с
// каждым опросом GET /message (streamPolls раз), active=true, затем
// active=false + финальное завершённое сообщение.
type fakeAPIServer struct { type fakeAPIServer struct {
sessionID string sessionID string
created bool created bool
@@ -28,6 +32,14 @@ type fakeAPIServer struct {
failMessages bool failMessages bool
createdModel *ModelRef // модель, полученная на POST /api/session createdModel *ModelRef // модель, полученная на POST /api/session
promptCalls int promptCalls int
// streamGrow: стрим одного растущего парта — текст/reasoning растёт с
// каждым опросом GET /message, active=true, пока messageCalls не дойдёт до
// streamPolls; затем active=false + финальное завершённое сообщение.
streamGrow bool
streamReasoning bool // растущий парт — reasoning вместо text
streamPolls int // сколько опросов длится «стрим» до завершения
messageCalls int
} }
func (f *fakeAPIServer) handler() http.Handler { func (f *fakeAPIServer) handler() http.Handler {
@@ -96,6 +108,27 @@ func (f *fakeAPIServer) handler() http.Handler {
http.Error(w, "db error", http.StatusInternalServerError) http.Error(w, "db error", http.StatusInternalServerError)
return return
} }
if f.streamGrow {
f.messageCalls++
now := time.Now().UnixMilli()
done := f.messageCalls >= f.streamPolls
msg := v2Message{ID: "msg_stream", Type: "assistant", Time: v2Time{Created: &now}}
switch {
case done:
msg.Content = []v2Part{{Type: "text", Text: "done-stream"}}
msg.Finish = "end_turn"
msg.Time.Completed = &now
f.active = false
case f.streamReasoning:
msg.Content = []v2Part{{Type: "reasoning", Text: strings.Repeat("r", f.messageCalls)}}
f.active = true
default:
msg.Content = []v2Part{{Type: "text", Text: strings.Repeat("x", f.messageCalls)}}
f.active = true
}
writeJSON(w, map[string]any{"data": []v2Message{msg}})
return
}
msgs := f.messages msgs := f.messages
if msgs == nil && f.verdictText != "" && !f.blockPrompt { if msgs == nil && f.verdictText != "" && !f.blockPrompt {
msgs = []v2Message{f.assistantMsg(f.verdictText)} msgs = []v2Message{f.assistantMsg(f.verdictText)}

View File

@@ -120,9 +120,10 @@ func (r *Runner) awaitVerdict(ctx context.Context, c *Client, model *ModelRef, s
admittedAt = adm.TimeCreated admittedAt = adm.TimeCreated
} }
// Прогресс = число text-партов в новых assistant-сообщениях. Рост сбрасывает // Прогресс = число контент-партов + суммарная длина их текста в новых
// idle-таймер (LLM стримит = жив). // assistant-сообщениях (progressOf). Рост сбрасывает idle-таймер: LLM
lastCount := -1 // стримит (даже в один растущий text-парт) или думает (reasoning) = жив.
lastParts, lastTextLen := -1, -1
lastProgress := time.Now() lastProgress := time.Now()
launch := time.Now() launch := time.Now()
@@ -164,10 +165,11 @@ func (r *Runner) awaitVerdict(ctx context.Context, c *Client, model *ModelRef, s
return nil, err return nil, err
} }
cur, count := newestAssistant(msgs, admittedAt) cur, _ := newestAssistant(msgs, admittedAt)
if count != lastCount { parts, textLen := progressOf(msgs, admittedAt)
if parts != lastParts || textLen != lastTextLen {
lastProgress = time.Now() lastProgress = time.Now()
lastCount = count lastParts, lastTextLen = parts, textLen
} }
now := time.Now() now := time.Now()
if now.Sub(lastProgress) > r.IdleTimeout { if now.Sub(lastProgress) > r.IdleTimeout {

View File

@@ -70,6 +70,46 @@ func TestRun_IdleTimeout(t *testing.T) {
} }
} }
func TestRun_StreamingGrowth(t *testing.T) {
t.Setenv("XDG_CONFIG_HOME", t.TempDir())
dir := t.TempDir()
// стрим: один text-парт растёт с каждым опросом (дольше, чем idle timeout),
// но модель жива → idle НЕ должен сработать.
f := &fakeAPIServer{streamGrow: true, streamPolls: 30}
p, _ := fakePool(t, f, dir)
r := &Runner{Pool: p, IdleTimeout: 60 * time.Millisecond,
PollInterval: 5 * time.Millisecond, Stdout: io.Discard}
res, err := r.Run(context.Background(), "task", dir, "dev", "")
if err != nil {
t.Fatalf("Run err: %v", err)
}
if res.RC != 0 {
t.Errorf("RC = %d, want 0 (растущий стрим не должен считаться hung)", res.RC)
}
if !contains(res.Stdout, "done-stream") {
t.Errorf("Stdout = %q, want contain done-stream", res.Stdout)
}
}
func TestRun_ReasoningGrowth(t *testing.T) {
t.Setenv("XDG_CONFIG_HOME", t.TempDir())
dir := t.TempDir()
// стрим: растёт только reasoning-парт (текста нет) — тоже живая активность.
f := &fakeAPIServer{streamGrow: true, streamReasoning: true, streamPolls: 30}
p, _ := fakePool(t, f, dir)
r := &Runner{Pool: p, IdleTimeout: 60 * time.Millisecond,
PollInterval: 5 * time.Millisecond, Stdout: io.Discard}
res, err := r.Run(context.Background(), "task", dir, "dev", "")
if err != nil {
t.Fatalf("Run err: %v", err)
}
if res.RC != 0 {
t.Errorf("RC = %d, want 0 (растущий reasoning не должен считаться hung)", res.RC)
}
}
func TestRun_ContextCancel(t *testing.T) { func TestRun_ContextCancel(t *testing.T) {
t.Setenv("XDG_CONFIG_HOME", t.TempDir()) t.Setenv("XDG_CONFIG_HOME", t.TempDir())
dir := t.TempDir() dir := t.TempDir()

View File

@@ -8,15 +8,25 @@ import (
"os" "os"
"os/exec" "os/exec"
"path/filepath" "path/filepath"
"runtime"
"strings" "strings"
"testing" "testing"
"time" "time"
) )
// fakeServeBin создаёт скрипт, имитирующий opencode serve: просто держит // fakeServeBin создаёт скрипт, имитирующий opencode serve: просто держит
// процесс живым (sleep), чтобы супервайзер мог им владеть и убивать его. // процесс живым (sleep/ping), чтобы супервайзер мог им владеть и убивать его.
// На Windows используется .cmd (с #!/bin/sh нельзя — он не исполняется).
func fakeServeBin(t *testing.T, workdir string) string { func fakeServeBin(t *testing.T, workdir string) string {
t.Helper() t.Helper()
if runtime.GOOS == "windows" {
bin := filepath.Join(workdir, "opencode-serve.cmd")
script := "@echo off\r\necho fake serve started\r\nping -n 300 127.0.0.1 >nul\r\n"
if err := os.WriteFile(bin, []byte(script), 0o755); err != nil {
t.Fatalf("write fake serve bin: %v", err)
}
return bin
}
bin := filepath.Join(workdir, "opencode-serve") bin := filepath.Join(workdir, "opencode-serve")
script := `#!/bin/sh script := `#!/bin/sh
echo "fake serve started" echo "fake serve started"

View File

@@ -4,7 +4,6 @@ import (
"context" "context"
"encoding/json" "encoding/json"
"errors" "errors"
"fmt"
"os" "os"
"os/exec" "os/exec"
"path/filepath" "path/filepath"
@@ -809,13 +808,41 @@ func TestWorkerStartStop(t *testing.T) {
w.Stop() w.Stop()
} }
// waitTaskStatus ждёт, пока задача достигнет статуса want. На Windows git-операции
// воркера заметно медленнее, чем на Linux, поэтому проверки в тестах не могут
// полагаться на фиксированные sleep'ы — только на polling до целевого статуса.
func waitTaskStatus(t *testing.T, ctx context.Context, s *storage.Storage, id int64, want storage.Status) {
t.Helper()
deadline := time.Now().Add(30 * time.Second)
for {
task, err := s.GetTask(ctx, id)
if err != nil {
t.Fatalf("get task %d: %v", id, err)
}
if task.Status == want {
return
}
switch task.Status {
case storage.StatusFailed, storage.StatusTimeout:
t.Fatalf("task %d: status %q, want %q", id, task.Status, want)
}
if time.Now().After(deadline) {
t.Fatalf("task %d: таймаут ожидания %q, последний статус %q", id, want, task.Status)
}
select {
case <-ctx.Done():
t.Fatalf("task %d: ctx done: %v", id, ctx.Err())
case <-time.After(50 * time.Millisecond):
}
}
}
func TestWorkerSemaphore(t *testing.T) { func TestWorkerSemaphore(t *testing.T) {
s := setupWorkerDB(t) s := setupWorkerDB(t)
// создаём 2 ready-задачи // создаём 2 ready-задачи
for i := 0; i < 2; i++ { task1 := createReadyTask(t, s, "task-0")
createReadyTask(t, s, fmt.Sprintf("task-%d", i)) task2 := createReadyTask(t, s, "task-1")
}
w := &Worker{ w := &Worker{
Store: s, Store: s,
@@ -829,12 +856,12 @@ func TestWorkerSemaphore(t *testing.T) {
w.sem = make(chan struct{}, 1) w.sem = make(chan struct{}, 1)
w.sem <- struct{}{} w.sem <- struct{}{}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel() defer cancel()
// первый poll — запустит 1 задачу (макс. 1) // первый poll — запустит 1 задачу (макс. 1)
w.pollAndDispatch(ctx) w.pollAndDispatch(ctx)
time.Sleep(200 * time.Millisecond) waitTaskStatus(t, ctx, s, task1.ID, storage.StatusSuccess)
// 1 должна быть success, 1 — всё ещё approved // 1 должна быть success, 1 — всё ещё approved
success, _ := s.ListTasks(ctx, storage.TaskFilter{Status: storage.StatusSuccess}) success, _ := s.ListTasks(ctx, storage.TaskFilter{Status: storage.StatusSuccess})
@@ -848,7 +875,7 @@ func TestWorkerSemaphore(t *testing.T) {
// первая завершилась и вернула токен в сем — можем диспатчить вторую // первая завершилась и вернула токен в сем — можем диспатчить вторую
w.pollAndDispatch(ctx) w.pollAndDispatch(ctx)
time.Sleep(200 * time.Millisecond) waitTaskStatus(t, ctx, s, task2.ID, storage.StatusSuccess)
success, _ = s.ListTasks(ctx, storage.TaskFilter{Status: storage.StatusSuccess}) success, _ = s.ListTasks(ctx, storage.TaskFilter{Status: storage.StatusSuccess})
if len(success) != 2 { if len(success) != 2 {