textmachine/backend/internal/pipeline/runner.go

466 lines
20 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

package pipeline
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"log/slog"
"os"
"strings"
"time"
"textmachine/backend/internal/config"
"textmachine/backend/internal/ledger"
"textmachine/backend/internal/llm"
"textmachine/backend/internal/obs"
"textmachine/backend/internal/store"
)
// runner.go: the Phase-0 mini-runner — one chunk through the configured stage
// list (C1: draft→edit) with full money discipline: reserve → call → settle+
// checkpoint (одна транзакция) → request_log. Resume: найденный чекпоинт =
// вызов не повторяется и не оплачивается. Гейты/эскалация/циклы по главам —
// Фаза 1, сюда же (код раннера).
// Runner executes a book's pipeline.
type Runner struct {
Book *config.Book
Models *config.Models
Pipeline *config.Pipeline
Store *store.Store
Pricer *ledger.Pricer
Log *slog.Logger
// Resnapshot re-pins existing jobs to the CURRENT snapshot when configs/
// prompts changed since the job started (tmctl --resnapshot). Без флага
// расхождение — громкая ошибка: смена контекста инвалидирует чекпоинты и
// пере-оплачивает вызовы, это делается только осознанно (Р6).
Resnapshot bool
clients map[string]llm.LLMClient
templates map[string]*PromptTemplate
}
// NewRunner loads all three configs rooted at book.yaml and opens the project
// store. Caller owns Close.
func NewRunner(bookPath string, logger *slog.Logger) (*Runner, error) {
book, err := config.LoadBook(bookPath)
if err != nil {
return nil, err
}
models, err := config.LoadModels(book.ModelsFile)
if err != nil {
return nil, err
}
pipe, err := config.LoadPipeline(book.Pipeline, models)
if err != nil {
return nil, err
}
// Fail-fast ДО открытия store (взятия flock и side-effect'ов): исполнима
// ли механика конфига и заданы ли ключи используемых моделей.
if err := pipe.CheckRunnable(); err != nil {
return nil, err
}
if err := models.CheckKeys(pipe); err != nil {
return nil, err
}
pricer, err := models.Prices()
if err != nil {
return nil, err
}
st, err := store.Open(book.ProjectDB)
if err != nil {
return nil, err
}
r := &Runner{
Book: book,
Models: models,
Pipeline: pipe,
Store: st,
Pricer: pricer,
Log: logger,
clients: map[string]llm.LLMClient{},
templates: map[string]*PromptTemplate{},
}
if err := r.loadTemplates(); err != nil {
st.Close()
return nil, err
}
return r, nil
}
func (r *Runner) Close() error { return r.Store.Close() }
func (r *Runner) loadTemplates() error {
for _, st := range r.Pipeline.Stages {
tpl, err := LoadPromptTemplate(st.Prompt)
if err != nil {
return err
}
r.templates[st.Name] = tpl
}
return nil
}
func (r *Runner) client(model string) (llm.LLMClient, error) {
if c, ok := r.clients[model]; ok {
return c, nil
}
c, err := BuildClient(r.Models, model, r.Log)
if err != nil {
return nil, err
}
r.clients[model] = c
return c, nil
}
// snapshotID materializes the job-context snapshot (§3.2): brief_hash, chunker
// version and the full stage plan (model, prompt version + CONTENT hash,
// sampling). Payload is rendered with a fixed field order — the id is a
// content hash, so identical context re-upserts idempotently.
func (r *Runner) snapshotID() (id, payload string, err error) {
type stageSnap struct {
Name string `json:"name"`
Role string `json:"role"`
Model string `json:"model"`
PromptVersion string `json:"prompt_version"`
PromptSHA256 string `json:"prompt_sha256"`
Temperature float64 `json:"temperature"`
Reasoning string `json:"reasoning"`
// ExtraBody модели меняет ТЕЛО запроса (GLM thinking-off и т.п.) —
// без него правка ручки в models.yaml молча не инвалидировала бы
// чекпоинты (находка ревью). json.Marshal карты сортирует ключи →
// детерминированный рендер.
ModelExtra json.RawMessage `json:"model_extra,omitempty"`
}
snap := struct {
BriefHash string `json:"brief_hash"`
ChunkerVersion string `json:"chunker_version"`
EstimatorVersion string `json:"estimator_version"`
PipelineCore string `json:"pipeline_core"`
// Defaults влияют на maxTokens, а тот входит в request-hash: без них
// правка max_output_ratio молча инвалидировала бы все чекпоинты в
// обход snapshot-гейта (находка ревью).
MaxOutputRatio float64 `json:"max_output_ratio"`
MinMaxTokens int `json:"min_max_tokens"`
Stages []stageSnap `json:"stages"`
}{
BriefHash: r.Book.BriefHash(),
ChunkerVersion: chunkerVersion,
EstimatorVersion: estimatorVersion,
PipelineCore: r.Pipeline.Core,
MaxOutputRatio: r.Pipeline.Defaults.MaxOutputRatio,
MinMaxTokens: r.Pipeline.Defaults.MinMaxTokens,
}
for _, st := range r.Pipeline.Stages {
ss := stageSnap{
Name: st.Name, Role: st.Role, Model: st.Model,
PromptVersion: st.PromptVersion, PromptSHA256: r.templates[st.Name].SHA256,
Temperature: st.Temperature, Reasoning: st.Reasoning,
}
if extra := r.Models.Models[st.Model].ExtraBody; len(extra) > 0 {
raw, merr := json.Marshal(extra)
if merr != nil {
return "", "", merr
}
ss.ModelExtra = raw
}
snap.Stages = append(snap.Stages, ss)
}
data, err := json.Marshal(snap)
if err != nil {
return "", "", err
}
sum := sha256.Sum256(data)
return hex.EncodeToString(sum[:]), string(data), nil
}
// StageResult reports one executed stage.
type StageResult struct {
Stage string
Role string
Model string // фактически ответившая модель
FromResume bool // served from a checkpoint, no call made
Usage llm.Usage
CostUSD float64
LatencyMS int
FinishReason string
Text string
}
// RunResult is the whole chunk's outcome.
type RunResult struct {
BookID string
Chapter int
ChunkIdx int
Stages []StageResult
FinalText string
TotalUSD float64
}
// TranslateOneChunk runs the configured stages over the book's source file as
// a single chunk (Фаза 0). Chapter/chunk are fixed at 1/0 until the chunker
// lands in Phase 1.
func (r *Runner) TranslateOneChunk(ctx context.Context) (*RunResult, error) {
srcRaw, err := os.ReadFile(r.Book.SourceFile)
if err != nil {
return nil, fmt.Errorf("pipeline: read source: %w", err)
}
// Нормализуем (BOM/CRLF/NFC) — request-hash стабилен между ОС/редакторами.
source := NormalizeSource(string(srcRaw))
if source == "" {
return nil, fmt.Errorf("pipeline: source file %s is empty", r.Book.SourceFile)
}
snapID, snapPayload, err := r.snapshotID()
if err != nil {
return nil, err
}
if err := r.Store.UpsertSnapshot(snapID, r.Book.BriefHash(), snapPayload); err != nil {
return nil, err
}
const chapter, chunkIdx = 1, 0
res := &RunResult{BookID: r.Book.BookID, Chapter: chapter, ChunkIdx: chunkIdx}
prev := ""
for _, st := range r.Pipeline.Stages {
sr, err := r.runStage(ctx, st, snapID, chapter, chunkIdx, source, prev)
if err != nil {
return res, err
}
res.Stages = append(res.Stages, *sr)
res.TotalUSD += sr.CostUSD
prev = sr.Text
}
res.FinalText = prev
return res, nil
}
func (r *Runner) runStage(ctx context.Context, st config.Stage, snapID string, chapter, chunkIdx int, source, prev string) (*StageResult, error) {
job, err := r.Store.EnsureJob(r.Book.BookID, chapter, st.Name, snapID)
if err != nil {
return nil, err
}
// Snapshot pinning (Р6): джоба заморожена на снапшоте своего старта. Если
// конфиги/промпты с тех пор изменились — это инвалидация чекпоинтов и
// пере-оплата; выполняем её только по явной команде.
if job.SnapshotID != snapID {
if !r.Resnapshot {
return nil, fmt.Errorf("pipeline: job %s/ch%d/%s was started under snapshot %.12s, current config renders snapshot %.12s — конфиг/промпты изменились; уже оплаченные чекпоинты станут недействительны и вызовы будут пере-оплачены; повторить с --resnapshot, чтобы принять это явно",
r.Book.BookID, chapter, st.Name, job.SnapshotID, snapID)
}
if err := r.Store.UpdateJobSnapshot(job.ID, snapID); err != nil {
return nil, err
}
r.Log.WarnContext(ctx, "job re-pinned to new snapshot (--resnapshot)", "stage", st.Name, "old", job.SnapshotID[:12], "new", snapID[:12])
}
// Статус 'running' выставляется НЕ здесь, а прямо перед вызовом провайдера:
// иначе отказ потолка/ошибка сборки клиента/промах гейта оставляли бы
// джобу навсегда в 'running' (находка ревью). Отказ потолка — retriable,
// джоба остаётся 'pending'; настоящая ошибка — 'failed'.
msgs, err := Messages(r.templates[st.Name], RenderVars{Book: r.Book, Text: source, Draft: prev})
if err != nil {
return nil, err
}
// max_tokens: калибровка полигона — русский выход ≈1.9× входа в токенах;
// ratio из конфига, floor защищает короткие чанки. Входит в request-hash
// (меняет вызов), поэтому считается детерминированно от исходника.
inputEst := EstimateTokens(source)
maxTokens := int(float64(inputEst) * r.Pipeline.Defaults.MaxOutputRatio)
if maxTokens < r.Pipeline.Defaults.MinMaxTokens {
maxTokens = r.Pipeline.Defaults.MinMaxTokens
}
// attempt 0: измерение регенераций подключит гейт-цикл Фазы 1.
const attempt = 0
reqHash := RequestHash(r.Book.BookID, chapter, chunkIdx, attempt, st.Name, st.Role, st.Model,
st.Temperature, st.Reasoning, false, maxTokens, snapID, msgs)
// Дополняем ReqInfo, СОХРАНЯЯ решения допуска (LogBodies) — перезапись с
// нуля отрезала бы задокументированный debug-канал (находка ревью).
ri, _ := obs.ReqInfoFromContext(ctx)
ri.Book, ri.Chapter, ri.Chunk, ri.Stage, ri.Role = r.Book.BookID, chapter, chunkIdx, st.Name, st.Role
ctx = obs.WithReqInfo(ctx, ri)
// Resume path: a checkpoint means the call already happened and was paid —
// serve it, log the hit, never re-bill (приёмка: kill -9 теряет ≤1 вызов).
// Пустой чекпоинт (оплаченный отказ/обрезка/нечитаемое тело) НЕ подаётся
// дальше молча: черновик="" отравил бы редактуру и экспорт (находка ревью).
if cp, err := r.Store.GetCheckpoint(reqHash); err != nil {
return nil, err
} else if cp != nil {
if strings.TrimSpace(cp.ResponseText) == "" {
return nil, fmt.Errorf("pipeline: stage %s has a BILLED but empty checkpoint (finish=%s, hash %.12s) — вызов был оплачен, но пригодного текста нет; до регенераций Фазы 1: удалить строку checkpoints вручную или дождаться механизма attempt", st.Name, cp.FinishReason, reqHash)
}
var usage llm.Usage
_ = json.Unmarshal([]byte(cp.UsageJSON), &usage)
r.Store.LogRequest(ctx, r.Log, store.RequestLog{
BookID: r.Book.BookID, Chapter: chapter, ChunkIdx: chunkIdx,
Stage: st.Name, Role: st.Role, ModelRequested: st.Model, ModelActual: cp.ModelActual,
RequestHash: reqHash, TMHit: true, OK: true, FinishReason: cp.FinishReason,
})
if err := r.Store.SetJobStatus(job.ID, "done"); err != nil {
return nil, err
}
r.Log.InfoContext(ctx, "stage served from checkpoint", "stage", st.Name, "hash", reqHash[:12])
return &StageResult{Stage: st.Name, Role: st.Role, Model: cp.ModelActual,
FromResume: true, Usage: usage, CostUSD: 0, FinishReason: cp.FinishReason, Text: cp.ResponseText}, nil
}
price := r.Pricer.PriceFor(st.Model)
promptEst := 0
for _, m := range msgs {
promptEst += EstimateTokens(m.Content)
}
estimate := ledger.EstimateUSD(price, promptEst, maxTokens)
resv, verdict, err := r.Store.Reserve(r.Book.BookID, estimate, store.Ceilings{
BookUSD: r.Book.Ceilings.BookUSD, DayUSD: r.Book.Ceilings.DayUSD,
})
if err != nil {
_ = r.Store.SetJobStatus(job.ID, "failed")
return nil, err
}
switch verdict {
case store.ReserveDeniedBook:
// Джоба остаётся 'pending' (не 'failed'): поднимут потолок — продолжит.
return nil, fmt.Errorf("pipeline: book USD ceiling reached (%.2f$) — raise ceilings.book_usd or stop", r.Book.Ceilings.BookUSD)
case store.ReserveDeniedDay:
return nil, fmt.Errorf("pipeline: daily USD ceiling reached (%.2f$)", r.Book.Ceilings.DayUSD)
}
client, err := r.client(st.Model)
if err != nil {
_ = r.Store.Release(resv)
_ = r.Store.SetJobStatus(job.ID, "failed")
return nil, err
}
if err := r.Store.SetJobStatus(job.ID, "running"); err != nil {
_ = r.Store.Release(resv)
return nil, err
}
start := time.Now()
resp, err := client.Complete(ctx, llm.LLMRequest{
Model: st.Model,
Messages: msgs,
MaxTokens: maxTokens,
Temperature: st.Temperature,
ReasoningEffort: st.Reasoning,
})
latency := int(time.Since(start).Milliseconds())
if err != nil {
// 2xx с нечитаемым телом: провайдер УЖЕ списал деньги — резерв не
// возвращаем, консервативно селтлим ОЦЕНКУ с пустым чекпоинтом
// (переучёт ≤1 оценки безопаснее слепого потолка) и падаем громко.
var bde *llm.BilledDecodeError
if errors.As(err, &bde) {
if serr := r.Store.SettleWithCheckpoint(resv, estimate, store.Checkpoint{
RequestHash: reqHash, JobID: job.ID, ChunkIdx: chunkIdx, Attempt: attempt,
Stage: st.Name, Role: st.Role, ModelRequested: st.Model, ModelActual: st.Model,
ResponseText: "", UsageJSON: "{}", CostUSD: estimate, FinishReason: "decode_error",
}); serr != nil {
r.Log.ErrorContext(ctx, "settle after billed decode failure also failed", "err", serr)
}
r.Store.LogRequest(ctx, r.Log, store.RequestLog{
BookID: r.Book.BookID, Chapter: chapter, ChunkIdx: chunkIdx,
Stage: st.Name, Role: st.Role, ModelRequested: st.Model,
RequestHash: reqHash, CostUSD: estimate, LatencyMS: latency,
Degraded: "billed_2xx_decode_failed", Err: err.Error(), OK: false,
})
_ = r.Store.SetJobStatus(job.ID, "failed")
return nil, fmt.Errorf("pipeline: stage %s: %w (оценка $%.6f заселтлена консервативно)", st.Name, err, estimate)
}
// No 2xx ever arrived: nothing was billed by the provider — release the
// reservation and surface the failure with a telemetry row.
if rerr := r.Store.Release(resv); rerr != nil {
r.Log.ErrorContext(ctx, "release after failed call also failed", "err", rerr)
}
r.Store.LogRequest(ctx, r.Log, store.RequestLog{
BookID: r.Book.BookID, Chapter: chapter, ChunkIdx: chunkIdx,
Stage: st.Name, Role: st.Role, ModelRequested: st.Model,
RequestHash: reqHash, LatencyMS: latency, Err: err.Error(), OK: false,
})
_ = r.Store.SetJobStatus(job.ID, "failed")
return nil, fmt.Errorf("pipeline: stage %s call: %w", st.Name, err)
}
modelActual := resp.Model
if modelActual == "" {
modelActual = st.Model
}
// Цена по фактически ответившей модели, с fallback на ЗАПРОШЕННУЮ (не на
// дешёвый глобальный якорь), если провайдер вернул канонизированный слаг.
price = r.Pricer.PriceForResponse(st.Model, modelActual)
cost := ledger.CostUSD(price, resp.Usage)
// Платная модель (InputPerM>0) вернула 2xx с нулевым usage — баг учёта
// провайдера/необычный ответ: $0 ослепил бы потолок, поэтому берём
// консервативную оценку и предупреждаем (находка ревью). Для local ($0
// цена) нулевой usage штатен — остаётся $0.
if cost == 0 && price.InputPerM > 0 && resp.Usage.PromptTokens == 0 && resp.Usage.CompletionTokens == 0 {
r.Log.WarnContext(ctx, "paid 2xx with zero usage; settling the reservation estimate to keep the ceiling honest",
"stage", st.Name, "model", modelActual, "estimate_usd", fmt.Sprintf("%.6f", estimate))
cost = estimate
}
usageJSON, err := json.Marshal(resp.Usage)
if err != nil {
return nil, err
}
// Деньги: settle + сырой ответ — одна транзакция (§3.3).
if err := r.Store.SettleWithCheckpoint(resv, cost, store.Checkpoint{
RequestHash: reqHash, JobID: job.ID, ChunkIdx: chunkIdx, Attempt: attempt,
Stage: st.Name, Role: st.Role,
ModelRequested: st.Model, ModelActual: modelActual, ResponseText: resp.Text,
UsageJSON: string(usageJSON), CostUSD: cost, FinishReason: resp.FinishReason,
ProviderRequestID: resp.ProviderRequestID,
}); err != nil {
// The provider HAS billed this 2xx; failing to persist means the money
// state is behind reality — fail loud, never continue on top.
return nil, fmt.Errorf("pipeline: settle stage %s: %w", st.Name, err)
}
// Пустой оплаченный ответ (finish=length: thinking съел бюджет; фильтр):
// деньги заселтлены, чекпоинт есть — но дальше по конвейеру его НЕ подаём:
// черновик="" молча отравил бы редактуру. Громкая остановка (регенерация
// по attempt — Фаза 1).
if strings.TrimSpace(resp.Text) == "" {
r.Store.LogRequest(ctx, r.Log, store.RequestLog{
BookID: r.Book.BookID, Chapter: chapter, ChunkIdx: chunkIdx,
Stage: st.Name, Role: st.Role, ModelRequested: st.Model, ModelActual: modelActual,
RequestHash: reqHash,
PromptTokens: resp.Usage.PromptTokens, CachedTokens: resp.Usage.CachedTokens,
CacheCreationTokens: resp.Usage.CacheCreationTokens,
CompletionTokens: resp.Usage.CompletionTokens, ReasoningTokens: resp.Usage.ReasoningTokens,
CostUSD: cost, LatencyMS: latency, FinishReason: resp.FinishReason,
Degraded: "empty_completion", OK: false,
})
_ = r.Store.SetJobStatus(job.ID, "failed")
return nil, fmt.Errorf("pipeline: stage %s returned an EMPTY completion (finish=%s, billed $%.6f, checkpointed) — вероятно thinking/фильтр съел бюджет; поднимите max_tokens/смените модель и удалите строку checkpoints %.12s", st.Name, resp.FinishReason, cost, reqHash)
}
r.Store.LogRequest(ctx, r.Log, store.RequestLog{
BookID: r.Book.BookID, Chapter: chapter, ChunkIdx: chunkIdx,
Stage: st.Name, Role: st.Role, ModelRequested: st.Model, ModelActual: modelActual,
RequestHash: reqHash,
PromptTokens: resp.Usage.PromptTokens, CachedTokens: resp.Usage.CachedTokens,
CacheCreationTokens: resp.Usage.CacheCreationTokens,
CompletionTokens: resp.Usage.CompletionTokens, ReasoningTokens: resp.Usage.ReasoningTokens,
CostUSD: cost, LatencyMS: latency, FinishReason: resp.FinishReason, OK: true,
})
if err := r.Store.SetJobStatus(job.ID, "done"); err != nil {
return nil, err
}
r.Log.InfoContext(ctx, "stage completed", "stage", st.Name, "model", modelActual,
"cost_usd", fmt.Sprintf("%.6f", cost), "latency_ms", latency,
"prompt_tokens", resp.Usage.PromptTokens, "completion_tokens", resp.Usage.CompletionTokens)
return &StageResult{Stage: st.Name, Role: st.Role, Model: modelActual,
Usage: resp.Usage, CostUSD: cost, LatencyMS: latency,
FinishReason: resp.FinishReason, Text: resp.Text}, nil
}