textmachine/backend/internal/pipeline/runner.go

package pipeline

import (
	"errors"
	"fmt"
	"log/slog"
	"os"
	"path/filepath"
	"sync"

	"textmachine/backend/internal/config"
	"textmachine/backend/internal/lang"
	"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 the precompute pass 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

	// baseMemory is the DRAFT-wave glossary bank: the BASE rows only (Source∈{seed,ruby,auto}, EXCLUDING
	// Source:mined). The draft wave's injection MUST be selected over this — not the enriched `memory` —
	// so it is BYTE-IDENTICAL across a bank-mining enrichment, matching the draft-wave snapshot which folds
	// baseMemoryVersion (mined-excluded). Otherwise signing a mined term would change the draft wire (the
	// injection is a message folded into request_hash) and silently re-bill the draft wave on the owner
	// re-run — even though the base snapshot is unchanged (the review-confirmed «переоплата ОДНА» hole).
	// When the book has NO mined rows (every $0 test / the golden), it is the SAME object as `memory`
	// (identical content) — no double materialization, the injection is unchanged. The editor keeps the
	// enriched `memory`. nil ⇔ memory is nil (materialized together in seedGlossary).
	baseMemory *MemoryBank

	// pack is the book's language-data pack (internal/lang), loaded once in openRunner from
	// book.LangpackRoot (D39.15/16). The the bank-mining stop bank-miner reads its tables; pack.Version() is folded into
	// the snapshot (a pack edit is a loud --resnapshot). nil when the book declares no langpack_root, or
	// its pair has no catalog dir — then the miner is inert (the bank-mining stop auto-continues) and the fold is omitted.
	pack *lang.Pack

	// escMu serializes the single-hop escalation budget admission across the PARALLEL draft workers
	// (R1). escalationBudgetRemains is a non-atomic read-then-act over EscalationSpentUSD, so N concurrent
	// draft chunks could each read spent<budget and all be admitted → the premium soft-cap overshoots by
	// up to N-1 hops. Holding escMu across the budget check AND the fresh hop's settle makes the cap exact
	// (a later chunk reads the updated spend). Escalations are the rare content-failure exception, so the
	// contention is negligible; the $0 checkpoint-replay of an already-paid hop stays lock-free.
	escMu sync.Mutex

	// lastMinedCount is the size of the most recent the bank-mining stop mined delta (set by runBankMiningStop) — carried into the
	// WaveSignatureStop the driver returns so the CLI can report "N terms await signature". Single-writer
	// (runBankMiningStop runs between the parallel waves, not inside one), so no synchronization is needed.
	lastMinedCount int
}

// 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
	}
	if err := r.loadLangPack(); err != nil {
		st.Close()
		return nil, err
	}
	return r, nil
}

// loadLangPack resolves the book's language-data pack (D39.15/16), loaded on BOTH the write path and the
// read-only path so status/export/report reproduce the identical snapshot (pack.Version() is folded). The
// contract mirrors the owner directive R1: a pair WITH a catalog directory (configs/langpacks/<src>-<tgt>/)
// loads FAIL-LOUD (a missing/corrupt file stops the run, never a silently-empty miner); a pair WITHOUT a
// catalog — or a book with no langpack_root — runs with a nil pack (the miner is inert, the bank-mining stop auto-continues,
// and the snapshot fold is omitted). Presence of the pair directory is the "catalog exists" signal.
func (r *Runner) loadLangPack() error {
	if r.Book.LangpackRoot == "" {
		return nil // no langpack declared → nil pack, miner inert
	}
	pairDir := filepath.Join(r.Book.LangpackRoot, r.Book.LangPair())
	if fi, err := os.Stat(pairDir); err != nil || !fi.IsDir() {
		// No catalog for this pair → nil-and-run (a ja book against a zh-only root just runs без майнинга).
		return nil
	}
	pack, err := lang.Load(r.Book.LangpackRoot, r.Book.SourceLang, r.Book.TargetLang)
	if err != nil {
		return fmt.Errorf("pipeline: load langpack for %s: %w", r.Book.LangPair(), err)
	}
	r.pack = pack
	return nil
}

// packVersion is the snapshot-folded language-pack version: pack.Version() when a pack is loaded, "" when
// not (omitted from the snapshot so a no-pack book is byte-stable and never re-billed for a feature it does
// not use). A pack DATA edit changes Version() → the snapshot moves → a loud --resnapshot (R1, drift-proof).
func (r *Runner) packVersion() string {
	if r.pack != nil {
		return r.pack.Version()
	}
	return ""
}

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 the precompute pass 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 (the precompute pass).
// 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 (the precompute pass): %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 the precompute pass, 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 — the precompute pass 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)
}