fix(opencode): убрать смешение слоёв API — перейти целиком на experimental (/session)
Корень проблемы «не получаем результаты»: клиент смешивал два слоя opencode
serve. CreateSession ходил на /api/session (v2, ждал {data.id}), Verdict — на
/api/session/{id}/message?order=desc и ждал {data:[{type,content}]}, где поле
content[].type/text физически отсутствует, поэтому вердикт никогда не находился
и поллинг уходил в вечный таймаут. Abort и вовсе звал несуществующий /interrupt.
Теперь весь код на experimental-слое, как сверено с sst/opencode (ветка dev):
- CreateSession: POST /session → голая Session, id в .id.
- Send: блокирующий POST /session/{id}/message, тело {parts:[{type:text,text}]},
вердикт из частей parts[].type=="text" ответа. Это и есть результат — метод
Verdict и отдельный GET удалены.
- textCount (прогресс): GET /session/{id}/message → голый массив [{info, parts}].
- Abort: POST /session/{id}/abort.
Runner: блокирующий Send запускается в горутине (канал вердикта/ошибки),
параллельно поллим textCount (рост text-частей сбрасывает idle-таймер). При
idle/hard-таймауте или отмене контекста — Abort + cancel() Send-горутины → rc=-1.
Send ходит через отдельный http.Client без жёсткого Timeout (управляется ctx),
чтобы длинная генерация не обрывалась на 30s. Тесты/fakeAPIServer переведены на
экспериментальный формат. Версия → 0.2.2.
This commit is contained in:
@@ -2,7 +2,6 @@ package opencode
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
@@ -84,17 +83,27 @@ func (r *Runner) Run(ctx context.Context, prompt, cwd, agent, sessionID string)
|
||||
r.logf("opencode(%s) session=%s на %s", agent, sid, srv.Addr())
|
||||
}
|
||||
|
||||
// Отправляем промпт (неблокирующий — сервер начинает выполнение).
|
||||
if err := c.Send(ctx, sid, prompt); err != nil {
|
||||
return nil, fmt.Errorf("opencode: prompt: %w", err)
|
||||
}
|
||||
|
||||
return r.awaitVerdict(ctx, c, sid, agent)
|
||||
// Отправляем промпт (блокирующий Send в горутине; вердикт придёт из него),
|
||||
// параллельно поллим прогресс и контролируем idle/hard таймауты.
|
||||
return r.awaitVerdict(ctx, c, sid, agent, prompt)
|
||||
}
|
||||
|
||||
// awaitVerdict поллит сообщения сессии, пока не появится готовый text-вердикт
|
||||
// от assistant, либо не истечёт idle/hard таймаут (тогда Abort + rc=-1).
|
||||
func (r *Runner) awaitVerdict(ctx context.Context, c *Client, sid, agent string) (*Result, error) {
|
||||
// awaitVerdict запускает блокирующий Send и параллельно поллит прогресс
|
||||
// (рост числа text-частей = агент жив, сбрасывает idle). Возвращается вердикт
|
||||
// из ответа Send, либо rc=-1 при idle/hard таймауте (тогда Abort + отмена ctx).
|
||||
func (r *Runner) awaitVerdict(ctx context.Context, c *Client, sid, agent, prompt string) (*Result, error) {
|
||||
sendCtx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
type sendOut struct {
|
||||
vd string
|
||||
err error
|
||||
}
|
||||
sendCh := make(chan sendOut, 1)
|
||||
go func() {
|
||||
vd, err := c.Send(sendCtx, sid, prompt)
|
||||
sendCh <- sendOut{vd: vd, err: err}
|
||||
}()
|
||||
|
||||
// Прогресс = сумма text-частей во всех assistant-сообщениях сессии. Рост
|
||||
// сбрасывает idle-таймер (LLM стримит = жив).
|
||||
var mu sync.Mutex
|
||||
@@ -102,13 +111,18 @@ func (r *Runner) awaitVerdict(ctx context.Context, c *Client, sid, agent string)
|
||||
lastProgress := time.Now()
|
||||
launch := time.Now()
|
||||
|
||||
abortAnd := func(rc int, why string) (*Result, error) {
|
||||
if err := c.Abort(ctx, sid); err != nil {
|
||||
r.logf("opencode(%s) abort %s: %v", agent, why, err)
|
||||
}
|
||||
cancel()
|
||||
return &Result{RC: rc, Stdout: "", SessionID: sid}, nil
|
||||
}
|
||||
|
||||
for {
|
||||
if ctx.Err() != nil {
|
||||
if err := c.Abort(ctx, sid); err != nil {
|
||||
r.logf("opencode(%s) abort (ctx): %v", agent, err)
|
||||
}
|
||||
r.logf("opencode(%s) ctx cancelled — обрыв (rc=-1)", agent)
|
||||
return &Result{RC: -1, Stdout: "", SessionID: sid}, nil
|
||||
return abortAnd(-1, "ctx")
|
||||
}
|
||||
|
||||
count, _ := c.textCount(ctx, sid)
|
||||
@@ -119,80 +133,39 @@ func (r *Runner) awaitVerdict(ctx context.Context, c *Client, sid, agent string)
|
||||
}
|
||||
mu.Unlock()
|
||||
|
||||
// Пробуем вердикт (дешёвый GET). Если готов — выходим.
|
||||
vd, vErr := c.Verdict(ctx, sid)
|
||||
if vErr == nil && vd != "" {
|
||||
r.logf("opencode(%s) вердикт готов (%d байт)", agent, len(vd))
|
||||
return &Result{RC: 0, Stdout: vd, SessionID: sid}, nil
|
||||
}
|
||||
// Сервер недоступен — фатально (не таймаут). Но если это следствие
|
||||
// отмены контекста (cancel прилетел прямо во время запроса) — это обрыв
|
||||
// по контексту, а не недоступность сервера: вернём RC=-1 на следующей
|
||||
// итерации (проверка ctx.Err() вверху) и не подменяем класс ошибки.
|
||||
var ce *ClientErr
|
||||
if errors.As(vErr, &ce) && ce.Op == "connect" && ctx.Err() == nil {
|
||||
return nil, fmt.Errorf("opencode: %w", vErr)
|
||||
}
|
||||
|
||||
now := time.Now()
|
||||
if now.Sub(lastProgress) > r.IdleTimeout {
|
||||
r.logf("opencode(%s) idle %.0fs — abort", agent, r.IdleTimeout.Seconds())
|
||||
if err := c.Abort(ctx, sid); err != nil {
|
||||
r.logf("opencode(%s) abort idle: %v", agent, err)
|
||||
}
|
||||
return &Result{RC: -1, Stdout: "", SessionID: sid}, nil
|
||||
return abortAnd(-1, "idle")
|
||||
}
|
||||
// hard — общий бюджет от старта запуска.
|
||||
if now.Sub(launch) > r.HardTimeout {
|
||||
r.logf("opencode(%s) hard timeout %.0fs — abort", agent, r.HardTimeout.Seconds())
|
||||
if err := c.Abort(ctx, sid); err != nil {
|
||||
r.logf("opencode(%s) abort hard: %v", agent, err)
|
||||
}
|
||||
return &Result{RC: -1, Stdout: "", SessionID: sid}, nil
|
||||
return abortAnd(-1, "hard")
|
||||
}
|
||||
|
||||
select {
|
||||
case out := <-sendCh:
|
||||
// Send завершился. Ошибка — connect (сервер недоступен) и ctx жив →
|
||||
// фатально, не таймаут. Если ctx уже отменён — это обрыв, а не ошибка.
|
||||
if out.err != nil {
|
||||
var ce *ClientErr
|
||||
if errors.As(out.err, &ce) && ce.Op == "connect" && ctx.Err() == nil {
|
||||
return nil, fmt.Errorf("opencode: %w", out.err)
|
||||
}
|
||||
if ctx.Err() != nil {
|
||||
return abortAnd(-1, "ctx")
|
||||
}
|
||||
return nil, out.err
|
||||
}
|
||||
r.logf("opencode(%s) вердикт готов (%d байт)", agent, len(out.vd))
|
||||
return &Result{RC: 0, Stdout: out.vd, SessionID: sid}, nil
|
||||
case <-time.After(r.PollInterval):
|
||||
case <-ctx.Done():
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// textCount считает число text-частей в assistant-сообщениях (для progress).
|
||||
func (c *Client) textCount(ctx context.Context, sessionID string) (int, error) {
|
||||
msgs, err := c.messages(ctx, sessionID)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
n := 0
|
||||
for _, m := range msgs {
|
||||
if m.Type != "assistant" {
|
||||
continue
|
||||
}
|
||||
for _, p := range m.Content {
|
||||
if p.Type == "text" && p.Text != "" {
|
||||
n++
|
||||
}
|
||||
}
|
||||
}
|
||||
return n, nil
|
||||
}
|
||||
|
||||
// messages возвращает сырые сообщения сессии (для поллинга прогресса).
|
||||
func (c *Client) messages(ctx context.Context, sessionID string) ([]sessionMessage, error) {
|
||||
raw, err := c.do(ctx, "GET", "/api/session/"+sessionID+"/message?order=asc&limit=200", "messages", nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var out struct {
|
||||
Data []sessionMessage `json:"data"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &out); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return out.Data, nil
|
||||
}
|
||||
|
||||
// ResumeDev — запуск dev-агента с resume-fallback. Если resume (sessionID)
|
||||
// падает с rc!=0 — повторяем ОДИН раз свежей сессией в том же каталоге.
|
||||
// rc=-1 (обрыв по таймауту) НЕ триггерит fallback.
|
||||
|
||||
Reference in New Issue
Block a user