package pipeline import ( "errors" "fmt" "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 BILINGUAL editor stage (D30.1 supersede of D1): editor.md feeds it BOTH the // source and the draft, plus the approved glossary's CONFIRMED dst forms as target-consistency // constraints. (The stale "monolingual, never sees the source" framing is D1, fully superseded.) 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 // rateGuards is the per-model wave-concurrency guard set (WS1 §1б), built once in W0 and // read-only in the waves — a transport axis, never snapshot-folded. nil until buildRateGuards. rateGuards map[string]*rateGuard // 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 { pair := r.Book.LangPair() for _, st := range r.Pipeline.Stages { // Resolve the stage's prompt from the pair-keyed pack by the book's language pair (D39 слой 2): // a legacy single `prompt` is pair-agnostic; a `prompts` pack fails LOUD on a missing pair here // (a ja book against a zh-ru-only pack stops instead of silently running Chinese conventions). promptPath, err := st.PromptPathFor(pair) if err != nil { return err } tpl, err := LoadPromptTemplate(promptPath) 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 } // segBudget resolves the pipeline's config.Segmentation into the pipeline-package SegBudget the // chunker consumes (WS2). Kept in the runner (not chunker.go) so the segmenter stays config-free. func (r *Runner) segBudget() SegBudget { seg := r.Pipeline.Segmentation return SegBudget{ DraftBudgetOut: float64(seg.DraftBudgetOut), EditCeilingOut: float64(seg.EditCeilingOut), FertCJK: seg.Fertility.CJK, FertOther: seg.Fertility.Other, } } // reachableModels enumerates every model a run can call: the union of the stage models and their // single-hop escalate_to fallbacks (stagerun.go calls r.client ONLY with model∈{st.Model, EscalateTo}). // This set is complete and static for a book, which is what lets W0 pre-build every client (a third // model axis — channel B / annotator — must extend this to stay race-free). Deterministic order. func (r *Runner) reachableModels() []string { seen := map[string]bool{} var out []string add := func(m string) { if m != "" && !seen[m] { seen[m] = true out = append(out, m) } } for _, st := range r.Pipeline.Stages { add(st.Model) add(st.EscalateTo) } return out } // buildClients EAGER-constructs every reachable LLM client BEFORE any wave goroutine starts (W0). // After this the clients map is READ-ONLY in the waves, so r.client is lock-free and a miss is a // loud error, not a lazy build under a data race — closing the runner.go lazy-init race (D12) and // tripwiring a future un-enumerated model axis. Idempotent; BuildClient needs only the provider // config (not the API key), so eager-build is safe even on a key-less resume. func (r *Runner) buildClients() error { for _, m := range r.reachableModels() { if _, ok := r.clients[m]; ok { continue } c, err := BuildClient(r.Models, m, r.Log) if err != nil { return fmt.Errorf("pipeline: eager-build client for model %q (W0): %w", m, err) } r.clients[m] = c } return nil } // client returns the PRE-BUILT client for a model. Read-only (no lazy build, no lock): buildClients // constructed every reachable client in W0, so the map is only read in the waves; a miss means an // un-enumerated model reached the wire and is a loud error, never a silent lazy build under a race. func (r *Runner) client(model string) (llm.LLMClient, error) { if c, ok := r.clients[model]; ok { return c, nil } return nil, fmt.Errorf("pipeline: no pre-built client for model %q — W0 buildClients enumerates stage models ∪ escalate_to; a model outside that set reached the wire, add it to the eager set to keep the waves race-free", model) }