package opencode import ( "context" "errors" "fmt" "io" "os" "strings" "time" ) // Result — результат запуска opencode-субагента через HTTP API. rc=-1 означает // «оборван по таймауту/контексту» (idle/hard): вызывающий НЕ должен ронять // задачу, а обязан закоммитить/запушить готовую работу и отправить на ревью // (класс O2 Timeout — результат, не ошибка). type Result struct { RC int Stdout string SessionID string } // Runner — запуск opencode-субагентов через v2 HTTP API serve. // // Runner ходит к opencode serve через Pool→Client (пути /api/*, см. README, // минимальная версия opencode). Промпт отправляется неблокирующе (durable // admit), вердикт собирается поллингом новых assistant-сообщений; завершение // ответа определяется по схеме «сессия больше не в активных дренажах» + финальное // assistant-сообщение. type Runner struct { Pool *Pool // пул serve-серверов (обязательный) IdleTimeout time.Duration HardTimeout time.Duration PollInterval time.Duration Debug bool // отладочные логи API-вызовов (из log.level=debug) // Заменяемый для тестов: Stdout io.Writer // диагностика (лог), по умолчанию os.Stderr } func (r *Runner) defaults() { if r.IdleTimeout == 0 { r.IdleTimeout = 5 * time.Minute } if r.HardTimeout == 0 { r.HardTimeout = 20 * time.Minute } if r.PollInterval == 0 { r.PollInterval = 2 * time.Second } if r.Stdout == nil { r.Stdout = os.Stderr } } func (r *Runner) logf(format string, args ...any) { fmt.Fprintf(r.Stdout, format+"\n", args...) } // Run запускает opencode-субагента через HTTP API: создаёт/продолжает сессию // в сервере пула для каталога cwd, отправляет промпт, ждёт вердикт. // // Возвращает *Result (rc, stdout=вердикт, session_id). Ошибка — только класс // O1 ErrRun (не смог обратиться к серверу/сессии). Таймауты дают rc=-1 в // Result, а не error (класс O2). func (r *Runner) Run(ctx context.Context, prompt, cwd, agent, sessionID string) (*Result, error) { r.defaults() if r.Pool == nil { return nil, fmt.Errorf("opencode: Pool не задан (API-режим обязателен)") } srv, err := r.Pool.Ensure(ctx, cwd) if err != nil { return nil, err } c := &Client{BaseURL: srv.Addr(), Password: srv.Password, Directory: srv.Dir, Debug: r.Debug} // Сессия: заданная (resume) или новая. Модель НЕ выбираем — opencode сам // берёт её по умолчанию из своего глобального конфига. sid := sessionID if sid == "" { sid, err = c.CreateSession(ctx, agent) if err != nil { return nil, fmt.Errorf("opencode: create session: %w", err) } r.logf("opencode(%s) session=%s на %s", agent, sid, srv.Addr()) r.logf("opencode(%s) сообщения сессии: %s", agent, SessionMessagesURL(srv.Addr(), sid)) } return r.awaitVerdict(ctx, c, sid, agent, prompt) } // settlePolls — сколько подряд опросов должно подтвердить завершение ответа, // прежде чем считать вердикт финальным (устойчивость к гонке между удалением // сессии из активных дренажей и финализацией последнего сообщения). const settlePolls = 2 // awaitVerdict отправляет промпт (неблокирующе) и поллит новые assistant-сообщения, // контролируя idle/hard таймауты. Завершение: сессия ушла из активных дренажей // И есть новое завершённое assistant-сообщение, стабильное в течение settlePolls // опросов. Возвращает вердикт (текст text-партов), либо rc=-1 при таймауте. func (r *Runner) awaitVerdict(ctx context.Context, c *Client, sid, agent, prompt string) (*Result, error) { // admit промпта; граница «новых» сообщений — время создания user-сообщения. admittedAt := time.Now().UnixMilli() adm, err := c.Prompt(ctx, sid, prompt) if err != nil { return nil, err } if adm != nil && adm.TimeCreated > 0 { admittedAt = adm.TimeCreated } // Прогресс = число контент-партов + суммарная длина их текста в новых // assistant-сообщениях (progressOf). Рост сбрасывает idle-таймер: LLM // стримит (даже в один растущий text-парт) или думает (reasoning) = жив. lastParts, lastTextLen := -1, -1 lastProgress := time.Now() launch := time.Now() doneSeen, emptySeen := 0, 0 abortAnd := func(rc int, why string) (*Result, error) { if err := c.Interrupt(ctx, sid); err != nil { r.logf("opencode(%s) interrupt %s: %v", agent, why, err) } return &Result{RC: rc, Stdout: "", SessionID: sid}, nil } for { if ctx.Err() != nil { r.logf("opencode(%s) ctx cancelled — обрыв (rc=-1)", agent) return abortAnd(-1, "ctx") } msgs, err := c.Messages(ctx, sid) if err != nil { if ctx.Err() != nil { return abortAnd(-1, "ctx") } var ce *ClientErr if errors.As(err, &ce) && ce.Op == "connect" { return nil, fmt.Errorf("opencode: %w", err) } return nil, err } active, err := c.Active(ctx, sid) if err != nil { if ctx.Err() != nil { return abortAnd(-1, "ctx") } var ce *ClientErr if errors.As(err, &ce) && ce.Op == "connect" { return nil, fmt.Errorf("opencode: %w", err) } return nil, err } cur, _ := newestAssistant(msgs, admittedAt) parts, textLen := progressOf(msgs, admittedAt) if parts != lastParts || textLen != lastTextLen { lastProgress = time.Now() lastParts, lastTextLen = parts, textLen } now := time.Now() if now.Sub(lastProgress) > r.IdleTimeout { r.logf("opencode(%s) idle %.0fs — abort", agent, r.IdleTimeout.Seconds()) return abortAnd(-1, "idle") } if now.Sub(launch) > r.HardTimeout { r.logf("opencode(%s) hard timeout %.0fs — abort", agent, r.HardTimeout.Seconds()) return abortAnd(-1, "hard") } switch { case !active && cur != nil && cur.finished(): // ответ закончен — ждём стабильности, затем собираем вердикт doneSeen++ emptySeen = 0 if doneSeen >= settlePolls { return r.verdict(cur, msgs, admittedAt, sid) } case !active && cur == nil: // сессия завершилась, но нового assistant-сообщения так и нет emptySeen++ if emptySeen >= settlePolls { return nil, &ClientErr{Op: "prompt", Err: errors.New("агент не выдал ответ (сессия пуста)")} } default: doneSeen, emptySeen = 0, 0 } select { case <-time.After(r.PollInterval): case <-ctx.Done(): } } } // verdict собирает финальный результат из новых assistant-сообщений. func (r *Runner) verdict(cur *v2Message, msgs []v2Message, since int64, sid string) (*Result, error) { if cur.Error != nil && cur.Error.Message != "" { return nil, &ClientErr{Op: "prompt", Err: errors.New(cur.Error.Message)} } texts, usedReasoning := assistantVerdict(msgs, since) if len(texts) == 0 { return nil, &ClientErr{Op: "prompt", Err: errors.New("нет text-части в ответе")} } if usedReasoning { r.logf("WARN opencode: в ответе нет text-части — использую reasoning-парты как вердикт") } vd := stripFence(strings.Join(texts, "\n")) r.logf("opencode вердикт готов (%d байт)", len(vd)) return &Result{RC: 0, Stdout: vd, SessionID: sid}, nil }