172 lines
6.9 KiB
Go
172 lines
6.9 KiB
Go
package pipeline
|
||
|
||
import (
|
||
"errors"
|
||
"log/slog"
|
||
"os"
|
||
|
||
"textmachine/backend/internal/config"
|
||
"textmachine/backend/internal/ledger"
|
||
"textmachine/backend/internal/llm"
|
||
"textmachine/backend/internal/store"
|
||
)
|
||
|
||
// runner.go: сетап раннера — Runner и его wiring (конфиг-стек, store в нужном
|
||
// режиме владения, кэши клиентов/шаблонов). Сама машинерия книги разнесена по
|
||
// связным файлам (пакет №4): снапшот — snapshot.go, сид/ruby — seeding.go, цикл
|
||
// книги — bookrun.go, disposition-петля чанка — chunkrun.go, исполнение стадии
|
||
// с деньгами (reserve → call → settle+checkpoint) — stagerun.go, single-hop
|
||
// эскалация с ре-гейтом — escalation.go, $0-resume решённого — resume.go.
|
||
|
||
// roleTranslator is the stage role whose output is a direct source→target
|
||
// translation — the only role the excision coverage-gate is meaningful on.
|
||
const roleTranslator = "translator"
|
||
|
||
// roleEditor is the monolingual editor stage (D1): it receives the draft plus the approved
|
||
// glossary's dst forms as target-consistency constraints, never the source.
|
||
const roleEditor = "editor"
|
||
|
||
// 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
|
||
|
||
// memory is the book's glossary FROZEN for this job (materialized once in
|
||
// TranslateBook, after seeding, before snapshotID). Its Version() is the F1
|
||
// content-hash folded into the snapshot; its Select drives per-chunk injection.
|
||
// nil until materialized (report path / no glossary) → memoryVersion() falls back
|
||
// to the empty-materialization hash, a stable constant.
|
||
memory *MemoryBank
|
||
}
|
||
|
||
// 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) {
|
||
return openRunner(bookPath, logger, true)
|
||
}
|
||
|
||
// NewReadOnlyRunner builds a runner for the $0 read-only commands (`tmctl report`/`status`): it
|
||
// SKIPS the API-key preflight so an auditor WITHOUT provider keys can read the store (D20.4). Those
|
||
// commands make zero LLM calls (they only project the persisted rows / re-chunk the source), so a
|
||
// missing key is irrelevant — the config-mechanics and adult-channel lints still run. translate and
|
||
// redrive (which DO call providers) keep NewRunner.
|
||
//
|
||
// The store opens READ-ONLY and WITHOUT the exclusive flock (store.OpenReadOnly), so
|
||
// status/report work DURING a live translate — раньше оператор был заперт от них на всё
|
||
// время прогона (боль smoke-прогона пакета №4). Первое касание проекта (файла БД ещё
|
||
// нет) падает обратно на полный Open: создать+мигрировать пустую базу — прежнее
|
||
// поведение «status до первого прогона показывает 0/N pending», и живого писателя,
|
||
// с которым можно столкнуться, в этот момент не существует.
|
||
func NewReadOnlyRunner(bookPath string, logger *slog.Logger) (*Runner, error) {
|
||
return openRunner(bookPath, logger, false)
|
||
}
|
||
|
||
// openRunner loads the config stack and opens the store. forWrite selects the call
|
||
// path: translate/redrive (провайдерские ключи обязательны, store во владение —
|
||
// flock+миграции+recovery) против status/report (без ключей, store read-only).
|
||
func openRunner(bookPath string, logger *slog.Logger, forWrite bool) (*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 := pipe.CheckAdultChannel(book.Adult); err != nil {
|
||
return nil, err
|
||
}
|
||
// Key preflight only for the call paths that actually reach a provider — read-only report/status
|
||
// are $0 (D20.4). CheckRunnable/CheckAdultChannel above stay unconditional (they cost nothing and
|
||
// catch a broken config even on a read).
|
||
if forWrite {
|
||
if err := models.CheckKeys(pipe); err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
pricer, err := models.Prices()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
var st *store.Store
|
||
if forWrite {
|
||
// Write path (translate/redrive): exclusive owner — flock, migrations,
|
||
// reservation recovery.
|
||
st, err = store.Open(book.ProjectDB)
|
||
} else {
|
||
// Read path (status/report): no flock — работает при живом прогоне; на
|
||
// первом касании проекта (базы ещё нет) — полный Open (создать+мигрировать).
|
||
st, err = store.OpenReadOnly(book.ProjectDB)
|
||
if err != nil && errors.Is(err, os.ErrNotExist) {
|
||
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
|
||
}
|
||
// Resolve the effective system once per stage: fold the few-shot block into
|
||
// System unless the stage switched it off (D38.4). FewShot is retained so the
|
||
// snapshot can fold the on/off state only for stages that actually have a block.
|
||
tpl.System = tpl.SystemFor(fewShotEnabled(st))
|
||
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
|
||
}
|