textmachine/backend/internal/pipeline/runner.go

package pipeline

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"sort"
	"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 Веха-2 book runner — every (chapter, chunk) from the chunker
// through the configured stage list (C1: draft→edit) with full money discipline:
// reserve → call → settle+checkpoint (одна транзакция) → request_log. A bad
// chunk is FLAGGED (disposition, chunk_status) and the loop continues (D2), not
// a run-crash; only an infra failure aborts. Resume reads chunk_status BEFORE
// rendering, so a done or terminally-flagged chunk is never re-attacked or
// re-billed. Retries walk the `attempt` axis (bigger max_tokens for the
// retryable length/empty subset, up to the regenerate cap, then flag).
// Escalation and the real linguistic chunker are later steps of Фаза 1.

// 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"

// errReserveCeiling is wrapped into the error runAttempt returns when a book/day USD
// ceiling denies a reservation. The PRIMARY path propagates it (the book durably
// pauses and resumes once the ceiling is raised, D4); the OPTIONAL escalation hop
// catches it and degrades to keeping the primary flag instead of aborting the whole
// book on every run (self-review finding).
var errReserveCeiling = errors.New("reserve ceiling reached")

// 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.
func NewReadOnlyRunner(bookPath string, logger *slog.Logger) (*Runner, error) {
	return openRunner(bookPath, logger, false)
}

func openRunner(bookPath string, logger *slog.Logger, checkKeys 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 checkKeys {
		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
}

// contextSnap freezes the context-assembly knobs (§3.8) inside the snapshot.
// Fixed field order — it is part of a content hash.
type contextSnap struct {
	GlossaryInjection   string `json:"glossary_injection"`
	GlossaryTokenBudget int    `json:"glossary_token_budget"`
	STMDepth            int    `json:"stm_depth"`
	OverlapTokens       int    `json:"overlap_tokens"`
	CacheTTL            string `json:"cache_ttl"`
}

// coverageSnap freezes the excision coverage-gate config inside the snapshot, so
// enabling the gate OR tuning its thresholds is a loud --resnapshot, never a silent
// mismatch between an old checkpoint's stored disposition and a changed gate. HONEST
// COST (D15): the snapshotID is folded into every call's RequestHash (render.go), so a
// --resnapshot changes EVERY request_hash → every checkpoint misses → the whole book is
// RE-BILLED, even for byte-identical wire requests whose verdict merely needs
// re-classifying. There is no free "re-classify from the checkpoint" here: the
// content-addressed reuse that COULD make an unchanged wire request free (msgsContentHash,
// render.go) is gated on an UNCHANGED snapshot (runner.go), so it never fires across a
// resnapshot. This all-or-nothing re-pay is ACCEPTED for the static acceptance book (the
// gate is loud, divergences do not occur) but is exactly why content-addressed checkpoint
// reuse must be designed before the ongoing/append-chapters mode (D15.2). Only folded when
// ENABLED: tweaking a disabled gate's bounds must not force a re-pin (the gate produces no
// verdicts while off). json.Marshal sorts the LenRatio map keys → deterministic render.
// Mirrors the estimator/max_tokens-policy discipline (render.go) for a NON-wire input that
// still determines a checkpoint's resolved verdict.
type coverageSnap struct {
	Enabled       bool                 `json:"enabled"`
	Version       string               `json:"version,omitempty"`
	LenRatio      map[string][]float64 `json:"len_ratio,omitempty"`
	SentCovMin    float64              `json:"sent_cov_min,omitempty"`
	MinChunkChars int                  `json:"min_chunk_chars,omitempty"`
}

// coverageSnapshot renders the gate's snapshot component: {enabled:false} when off
// (so toggling on is a visible change), the full config + algorithm version when on.
func (r *Runner) coverageSnapshot() coverageSnap {
	cov := coverageSnap{Enabled: r.Pipeline.Gates.Coverage.Enabled}
	if cov.Enabled {
		cov.Version = coverageGateVersion
		cov.LenRatio = r.Pipeline.Gates.Coverage.LenRatio
		cov.SentCovMin = r.Pipeline.Gates.Coverage.SentCovMin
		cov.MinChunkChars = r.Pipeline.Gates.Coverage.MinChunkChars
	}
	return cov
}

// classifyOutput resolves a completion's disposition: FIRST the always-on intrinsic
// classifier (echo/empty/refusal/length — Веха 2), then, only if that is ok AND the
// configurable coverage gate is enabled, the excision gate (шаг 6 / D12 Q3). Both are
// pure and deterministic over (source, output, finish), so a resumed checkpoint
// reproduces the identical verdict for free — the intrinsic part unconditionally, the
// gate part under the same coverage config (folded into the snapshot, so a gate change
// re-pins loudly). The coverage gate runs ONLY on the TRANSLATOR role's output (vs the
// original source): a monolingual editor legitimately restructures sentences, so gating
// it against the source would false-flag a correct edit (see the role check below).
func (r *Runner) classifyOutput(role, source, output, finish string) classification {
	cls := classify(classifyInput{Source: source, Output: output, Finish: finish, TargetLang: r.Book.TargetLang})
	if !cls.ok() || !r.Pipeline.Gates.Coverage.Enabled {
		return cls
	}
	// The coverage gate compares the output against the ORIGINAL source, which is a
	// source→target translation only for the TRANSLATOR role. A monolingual editor
	// legitimately restructures/merges sentences, so gating ITS output against the
	// source with the translation corridor false-flags a correct edit as excision
	// (self-review finding). Editor/other-role fidelity is a Phase-2 concern (a
	// bilingual judge over the draft, §04 mode 2), not this excision gate.
	if role != roleTranslator {
		return cls
	}
	if cov := coverageCheck(r.Pipeline.Gates.Coverage, source, output, r.Book.SourceLang, r.Book.TargetLang); !cov.cls.ok() {
		return cov.cls
	}
	return cls
}

// escalationBudgetRemains reports whether the book may still spend on a single-hop
// fallback draft: escalation is OPT-IN via escalation.budget_usd (0 = disabled, the
// boevoy default — a stage's escalate_to is inert until a premium budget is set), and
// capped at that budget summed over the book's escalation checkpoints (money-path
// durable, resume-safe). A PRE-HOP soft cap: the gate admits a hop while spent <
// budget and does NOT pre-estimate the hop's own cost, so a single hop may overshoot
// the budget by up to its full cost; the NEXT chunk's escalation is then denied. Size
// the budget with that worst case in mind — it bounds TOTAL escalation, not per-hop.
func (r *Runner) escalationBudgetRemains() (bool, error) {
	budget := r.Pipeline.Escal.BudgetUSD
	if budget <= 0 {
		return false, nil
	}
	spent, err := r.Store.EscalationSpentUSD(r.Book.BookID)
	if err != nil {
		return false, err
	}
	return spent < budget, nil
}

// memoryVersion is the content-hash of the deterministically materialized injected
// memory (the frozen APPROVED glossary rows + the normalization/matcher algorithm
// versions), the memory component of the snapshot (D5.2/D8, F1 CLOSED). It is the
// Version() of the bank materialized once before the loop, so a change to the approved
// glossary — or to the deterministic machinery (trad→simp table, matcher) — fires the
// resnapshot gate LOUDLY instead of a stale checkpoint re-paying a diverged translation.
// nil bank (report path / a book with no glossary) → the empty-materialization hash, a
// stable constant. STM is excluded (rebuilt from checkpoints, §3.2). Auto/draft rows are
// excluded per D8 (their injected-content changes are caught at the per-chunk
// content_hash level; a future autopopulation milestone revisits this).
func (r *Runner) memoryVersion() string {
	if r.memory != nil {
		return r.memory.Version()
	}
	return computeMemoryVersion(nil, r.Pipeline.Gates.Glossary.PostcheckGate)
}

// snapshotID materializes the job-context snapshot (§3.2): brief_hash, chunker
// version, context-assembly knobs, the memory version and the full stage plan
// (model, prompt version + CONTENT hash, sampling, resolved capability).
// 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"`
		// Провайдер-уровневые оверрайды (local kind переопределяет wire
		// temperature/max_tokens ПОСЛЕ вычисления request-hash) — тоже влияют
		// на фактический запрос; без них правка providers.local.max_tokens
		// давала бы тот же snapshot и ложный checkpoint-hit на стендовом
		// local-пути (находка внешнего ревью F2).
		ProviderTemp   float64 `json:"provider_temp,omitempty"`
		ProviderMaxTok int     `json:"provider_max_tok,omitempty"`
		// ProviderModel — the local-kind backend tag the provider swaps onto the wire
		// AFTER the request-hash (the actual model that answers). The MOST impactful
		// local override, yet it was missing here while its weaker temp/max_tok siblings
		// were folded: a local swap 8b→14b mid-book keeps the same snapID and resume
		// serves the old model (external-review). Folded so it is a loud --resnapshot.
		ProviderModel string `json:"provider_model,omitempty"`
		// Capability — резолвнутая wire-форма модели (D3.1): budget-ключ,
		// temperature-режим, reasoning-контроль. Меняет ТЕЛО запроса (max_tokens
		// vs max_completion_tokens, отправлять ли temperature, thinking-выключа-
		// тель), поэтому обязана входить в снапшот — иначе правка каппы молча
		// false-хитит чекпоинты (тот же класс D5.2, что и payload ниже).
		Capability json.RawMessage `json:"capability,omitempty"`
		// EscalateTo + EscalateCapability — the single-hop fallback model and its
		// resolved wire shape (Веха 2.5). The fallback's request_hash uses ITS model,
		// and its wire body uses ITS capability — both must be in the snapshot, or
		// changing the escalation model / its caps would silently false-hit an
		// escalated chunk's checkpoint (closes the D5.2 escalation-capability techdebt:
		// stages were folded, escalation models were not, until the cycle landed).
		EscalateTo         string          `json:"escalate_to,omitempty"`
		EscalateCapability json.RawMessage `json:"escalate_capability,omitempty"`
		// The fallback model's OTHER wire-affecting inputs, mirroring the primary
		// fold above: its top-level extra_body (merged into the escalation wire body)
		// and its provider-level temperature/max_tokens overrides (local kind). Without
		// these, editing the fallback's extra_body / provider knobs would change the
		// escalation wire but leave the snapshot identical → a silent false-hit on a
		// resumed escalated chunk (self-review: the primary path guards this, escalation
		// did not).
		EscalateExtra          json.RawMessage `json:"escalate_extra,omitempty"`
		EscalateProviderTemp   float64         `json:"escalate_provider_temp,omitempty"`
		EscalateProviderMaxTok int             `json:"escalate_provider_max_tok,omitempty"`
		EscalateProviderModel  string          `json:"escalate_provider_model,omitempty"`
	}
	snap := struct {
		BriefHash        string `json:"brief_hash"`
		ChunkerVersion   string `json:"chunker_version"`
		EstimatorVersion string `json:"estimator_version"`
		// MaxTokensPolicy versions the attempt→max_tokens scaling (Веха 2): a
		// change to the retry doubling shifts attempt≥1 request_hashes, so it
		// belongs in the snapshot as a loud invalidation (same class as
		// estimator_version, applied to the regeneration axis).
		MaxTokensPolicy string `json:"max_tokens_policy"`
		// ClassifierVersion versions the intrinsic classify() verdict logic (thresholds
		// + order), so a re-verdict on a resumed checkpoint is a loud --resnapshot, not
		// a silent flagged↔ok divergence (external-review; symmetric to coverage).
		ClassifierVersion string `json:"classifier_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"`
		// ContextAssembly — ручки сборки контекста (глоссарий/STM/overlap/TTL,
		// §3.8). Их правка меняет ВХОД модели, когда память войдёт в msgs; сворачи-
		// ваем ДО инъекции, чтобы смена budget'а инъекции не промахнулась мимо
		// resnapshot-гейта (D5.2). Фикс-порядок полей — это content-hash.
		ContextAssembly contextSnap `json:"context_assembly"`
		// MemoryVersion — content-hash детерминированно материализованной инъекти-
		// руемой памяти (approved-глоссарий+резюме+series-bible). Не bump-счётчик
		// (D8): забыть пересчитать нельзя, а забытый bump переоткрыл бы класс тихой
		// переоплаты D5.2. До миграции банка памяти (v2) материализация пуста —
		// хэш стабилен; когда память войдёт в msgs, поле сменится и resnapshot-
		// гейт сработает громко. STM сюда НЕ входит (пересобирается из чекпоинтов).
		MemoryVersion string `json:"memory_version"`
		// PostcheckGate — the memory-bank post-check gate (E1). When true a post-check
		// miss flips the chunk to flagged; folding its on/off state makes toggling it a
		// loud --resnapshot (it changes a checkpoint's RESOLVED disposition), not a silent
		// re-verdict on resume (same class as coverage/classifier). The post-check
		// ALGORITHM version is already inside MemoryVersion (memoryMatchVersion).
		PostcheckGate bool `json:"postcheck_gate"`
		// Coverage — excision QA-gate config (D12 Q3). It does not touch the wire,
		// but it determines a checkpoint's RESOLVED verdict, so a gate change must be
		// a loud --resnapshot, not a silent re-verdict on resume (coverageSnapshot).
		Coverage coverageSnap `json:"coverage"`
		// StyleCheckVersion versions the cheap deterministic post-check rules (dialogue-dash,
		// yofikator, translit-interjection blocklist, 万/億 magnitude gate — cheapgates.go). They
		// are observability, not wire, but editing a rule shifts the recorded style-flag counts, so
		// a bump is a loud --resnapshot (same verdict class as ClassifierVersion). The ё-policy is
		// part of BriefHash (a book field), so a policy change already re-pins via brief_hash.
		StyleCheckVersion string      `json:"style_check_version"`
		Stages            []stageSnap `json:"stages"`
	}{
		BriefHash:         r.Book.BriefHash(),
		ChunkerVersion:    chunkerVersion,
		EstimatorVersion:  estimatorVersion,
		MaxTokensPolicy:   maxTokensPolicyVersion,
		ClassifierVersion: classifierVersion,
		PipelineCore:      r.Pipeline.Core,
		MaxOutputRatio:    r.Pipeline.Defaults.MaxOutputRatio,
		MinMaxTokens:      r.Pipeline.Defaults.MinMaxTokens,
		ContextAssembly: contextSnap{
			GlossaryInjection:   r.Pipeline.Context.GlossaryInjection,
			GlossaryTokenBudget: r.Pipeline.Context.GlossaryTokenBudget,
			STMDepth:            r.Pipeline.Context.STMDepth,
			OverlapTokens:       r.Pipeline.Context.OverlapTokens,
			CacheTTL:            r.Pipeline.Context.CacheTTL,
		},
		MemoryVersion:     r.memoryVersion(),
		PostcheckGate:     r.Pipeline.Gates.Glossary.PostcheckGate,
		Coverage:          r.coverageSnapshot(),
		StyleCheckVersion: cheapGateVersion,
	}
	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 prov, ok := r.Models.Providers[r.Models.Models[st.Model].Provider]; ok {
			ss.ProviderTemp, ss.ProviderMaxTok, ss.ProviderModel = prov.Temperature, prov.MaxTokens, prov.Model
		}
		if extra := r.Models.Models[st.Model].ExtraBody; len(extra) > 0 {
			raw, merr := json.Marshal(extra)
			if merr != nil {
				return "", "", merr
			}
			ss.ModelExtra = raw
		}
		// Резолвнутая каппа — тем же ResolveCapability, что и у клиента, поэтому
		// хэшируемое всегда совпадает с отправляемым. json.Marshal сортирует
		// ключи карт (off_extra_body) → детерминированный рендер.
		capRaw, cerr := json.Marshal(r.Models.ResolveCapability(st.Model))
		if cerr != nil {
			return "", "", cerr
		}
		ss.Capability = capRaw
		// Fold the single-hop fallback model + its resolved capability (Веха 2.5):
		// both are wire-affecting inputs of an escalated call, so changing them is a
		// loud --resnapshot, not a silent false-hit on an escalated checkpoint.
		if st.EscalateTo != "" {
			ss.EscalateTo = st.EscalateTo
			escCap, eerr := json.Marshal(r.Models.ResolveCapability(st.EscalateTo))
			if eerr != nil {
				return "", "", eerr
			}
			ss.EscalateCapability = escCap
			if prov, ok := r.Models.Providers[r.Models.Models[st.EscalateTo].Provider]; ok {
				ss.EscalateProviderTemp, ss.EscalateProviderMaxTok, ss.EscalateProviderModel = prov.Temperature, prov.MaxTokens, prov.Model
			}
			if extra := r.Models.Models[st.EscalateTo].ExtraBody; len(extra) > 0 {
				raw, merr := json.Marshal(extra)
				if merr != nil {
					return "", "", merr
				}
				ss.EscalateExtra = 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 of a chunk.
type StageResult struct {
	Stage        string
	Role         string
	Model        string // фактически ответившая модель
	FromResume   bool   // no provider call was made THIS run (fully served from checkpoints)
	Usage        llm.Usage
	CostUSD      float64 // THIS run's spend (0 when fully served from checkpoints)
	CumCostUSD   float64 // sum across ALL attempts of this chunk×stage (F3-honest, incl. retries)
	LatencyMS    int
	FinishReason string
	Text         string // the usable output (only on an ok disposition)
	Disposition  Disposition
	FlagReason   FlagReason // "" when ok
	Detail       string
	Attempts     int
	// Escalated — a single-hop fallback draft was tried this stage (D12); when the
	// fallback passed the re-gate, Model above is the fallback (it answered).
	Escalated       bool
	EscalationModel string // the fallback model when Escalated ("" otherwise)
}

// ChunkOutcome is one chunk's result across the stage list.
type ChunkOutcome struct {
	Chapter     int
	ChunkIdx    int
	Stages      []StageResult
	FinalText   string      // the last stage's output; "" when the chunk is flagged
	Disposition Disposition // ok | flagged (a chunk has no "skipped" — that is a per-later-stage state)
	FlagReason  FlagReason  // the flagging stage's reason ("" when ok)
	CostUSD     float64     // THIS run's spend on this chunk
}

// BookResult aggregates a whole run over every chapter×chunk.
type BookResult struct {
	BookID   string
	Chunks   []ChunkOutcome
	TotalUSD float64 // THIS run's spend across all chunks
	Flagged  int     // number of flagged chunks (acceptance допускает N)
}

// ExitCode is the run's shell disposition: 0 clean, 2 completed-with-flags
// (acceptance допускает N флагов — приёмка это разрешает). An infra failure is
// an error from TranslateBook, which the CLI maps to 1.
func (b *BookResult) ExitCode() int {
	if b.Flagged > 0 {
		return 2
	}
	return 0
}

// CompletedWithFlags is the typed sentinel the CLI maps to exit code 2: the book
// finished end-to-end but N chunks were flagged for a human. It is NOT an infra
// failure (a plain error → exit 1); it is a "clean run, attention needed" signal
// carried up through `func run() error` in the idiomatic Go way.
type CompletedWithFlags struct {
	Flagged int
	Total   int
}

func (e *CompletedWithFlags) Error() string {
	return fmt.Sprintf("completed with %d/%d chunk(s) flagged for review", e.Flagged, e.Total)
}

// TranslateBook runs the whole book: split the normalized source into
// chapter/chunk units (chunker.go) and drive each chunk through the stage list.
// A bad chunk is flagged and the loop CONTINUES (D2); only an infra failure
// (ceiling, config change without --resnapshot, unbilled call failure, store
// error) aborts with an error, from which resume continues.
func (r *Runner) TranslateBook(ctx context.Context) (*BookResult, error) {
	// Ingest reads + normalizes the source (txt/epub) into ordered per-chapter text
	// and captures ruby readings (ingest.go), decoding the txt source per book.encoding
	// (auto/utf8/gb18030). Offline and deterministic ($0, no LLM).
	doc, err := IngestEncoded(r.Book.SourceFile, r.Book.Encoding, r.Book.SourceLang)
	if err != nil {
		return nil, err
	}

	// Persist captured ruby readings (idempotent; consumed by seedGlossary below into
	// auto glossary candidates — шаг 4). Never injected into a prompt here (§7d); a
	// resume re-persists the same rows at $0.
	if err := r.persistRuby(doc.Ruby); err != nil {
		return nil, fmt.Errorf("pipeline: persist ruby readings: %w", err)
	}

	// Seed + materialize the glossary BEFORE the snapshot: the frozen approved rows feed
	// memoryVersion() → the snapshot (F1), so the snapshot must be computed with the bank
	// already materialized. Deterministic and $0 (seed file + classified ruby, no LLM).
	if err := r.seedGlossary(ctx); err != nil {
		return nil, err
	}

	// Snapshot is book-level (brief + stage plan + memory), computed once and
	// upserted before the loop; every job pins to it. memoryVersion() now reflects the
	// materialized bank, so a changed approved glossary re-pins loudly (F1).
	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
	}

	chunks := SplitChunks(doc.Chapters)
	if len(chunks) == 0 {
		return nil, fmt.Errorf("pipeline: source file %s produced no chunks after normalization", r.Book.SourceFile)
	}

	res := &BookResult{BookID: r.Book.BookID}
	// Sticky scene-inertia state (A5): the exact-matched ids of the recent chunks in
	// the CURRENT chapter, reset at a chapter boundary (a new chapter is a scene change).
	// Rebuilt deterministically each run because translateChunk recomputes the ($0)
	// selection for EVERY chunk, resumed ones included — so sticky is resume-stable.
	var stickyWin []map[string]injectionDisposition
	prevChapter := 0
	for _, ch := range chunks {
		if ch.Chapter != prevChapter {
			stickyWin = nil
			prevChapter = ch.Chapter
		}
		outcome, activeIDs, err := r.translateChunk(ctx, snapID, ch, unionSticky(stickyWin))
		if err != nil {
			// Infra failure: abort. Chunks already committed are checkpointed;
			// resume continues from here at $0 for the done work.
			return res, err
		}
		res.Chunks = append(res.Chunks, *outcome)
		res.TotalUSD += outcome.CostUSD
		if outcome.Disposition == DispFlagged {
			res.Flagged++
		}
		// Advance the sticky window (keep the last stickyDepth chunks' exact matches).
		stickyWin = append(stickyWin, activeIDs)
		if len(stickyWin) > stickyDepth {
			stickyWin = stickyWin[len(stickyWin)-stickyDepth:]
		}
	}
	return res, nil
}

// unionSticky merges the recent chunks' exact-matched ids into one sticky_prev, carrying
// each id's DISPOSITION (D16.2). When an id fired in several recent chunks, the MOST RECENT
// firing wins (win is ordered oldest→newest), so a re-established CONFIRMED match overrides
// an older collision-prone AMBIGUOUS one — and vice-versa, never silently upgrading.
func unionSticky(win []map[string]injectionDisposition) map[string]injectionDisposition {
	if len(win) == 0 {
		return nil
	}
	out := map[string]injectionDisposition{}
	for _, s := range win {
		for id, disp := range s {
			out[id] = disp
		}
	}
	return out
}

// seedGlossary REPLACES the book's glossary from its deterministic inputs — the manual
// seed file (curated approved/draft terms) and the captured ruby readings (classified
// into auto candidates) — then MATERIALIZES the frozen bank for this job (r.memory). Run
// once before snapshotID: the frozen APPROVED rows are hashed into memoryVersion (F1), so
// editing the seed is a loud --resnapshot. Idempotent (full replace), $0, no LLM. B2
// approved dst-collisions are logged (not fatal — some collisions are legitimate).
func (r *Runner) seedGlossary(ctx context.Context) error {
	var entries []store.GlossaryEntry
	if r.Book.GlossarySeed != "" {
		seed, err := loadGlossarySeed(r.Book.GlossarySeed)
		if err != nil {
			return err
		}
		entries = append(entries, seed...)
	}
	manualSrcs := map[string]bool{}
	for _, e := range entries {
		manualSrcs[e.Src] = true
	}
	ruby, err := r.Store.RubyReadingsForBook(r.Book.BookID)
	if err != nil {
		return err
	}
	// D16.4: attach a manual term's kana ruby-reading as an alias (kana spelling matchable)
	// BEFORE appending the auto-candidates, so it only touches the curated manual entries. A
	// reading that would collide with a different seeded term (homophone) is skipped+logged,
	// not attached (which would fail the book loud on an alias the operator cannot edit out).
	if skipped := attachRubyAliasesToManual(entries, ruby); len(skipped) > 0 {
		r.Log.WarnContext(ctx, "ruby kana-alias skipped as a homophone collision (kana form left unmatchable; disambiguate in the seed if needed)",
			"skipped", strings.Join(skipped, "; "))
	}
	entries = append(entries, rubyToCandidates(ruby, manualSrcs)...)
	for i := range entries {
		entries[i].BookID = r.Book.BookID
	}
	// Task-6 re-audit: fail loud on a firing key shared by two DIFFERENT approved terms with
	// different dst + overlapping windows (the alias generalization of the D16.1 polysemy
	// livelock) — checked over the FULL entry set (incl. ruby-attached aliases) before persisting.
	if cols := approvedSharedKeyCollisions(entries); len(cols) > 0 {
		return fmt.Errorf("pipeline: glossary shared-key collisions (A2 / D16.1 livelock class):\n  - %s", strings.Join(cols, "\n  - "))
	}
	if err := r.Store.ReplaceGlossary(r.Book.BookID, entries); err != nil {
		return err
	}
	rows, err := r.Store.GlossaryForBook(r.Book.BookID)
	if err != nil {
		return err
	}
	r.memory = materializeMemory(rows, r.Pipeline.Gates.Glossary.PostcheckGate)
	if cols := injectivityCollisions(rows); len(cols) > 0 {
		r.Log.WarnContext(ctx, "glossary approved dst-collisions (B2: two source terms share one Russian surface — the reader cannot tell them apart)",
			"collisions", strings.Join(cols, "; "))
	}
	r.Log.InfoContext(ctx, "glossary materialized", "book", r.Book.BookID,
		"entries", len(rows), "memory_version", r.memory.Version()[:12])
	return nil
}

// persistRuby aggregates the ingested ruby occurrences into one row per
// (base, reading) — first_chapter = MIN, occurrences = full-book count — and
// REPLACES the book's whole ruby set (store.ReplaceRubyReadings). Idempotent: the
// aggregation is recomputed identically on every ingest, so a resume re-writes the
// same rows; a source edit converges every column — a pair removed by the edit
// disappears instead of lingering as a phantom (external-review #6). Called
// unconditionally (even for zero readings — a txt book or a source that dropped its
// furigana) so the full-replace clears any stale rows. The write order is sorted for
// deterministic, test-stable behavior. Memory v2 (шаг 4) consumes ruby_readings into
// a glossary name-lock (D9); nothing here injects into a prompt (§7d).
func (r *Runner) persistRuby(readings []RubyReading) error {
	type agg struct {
		first int
		count int
	}
	seen := map[[2]string]*agg{}
	order := make([][2]string, 0, len(readings))
	for _, rr := range readings {
		key := [2]string{rr.Base, rr.Reading}
		a, ok := seen[key]
		if !ok {
			a = &agg{first: rr.Chapter}
			seen[key] = a
			order = append(order, key)
		}
		if rr.Chapter < a.first {
			a.first = rr.Chapter
		}
		a.count++
	}
	sort.Slice(order, func(i, j int) bool {
		if order[i][0] != order[j][0] {
			return order[i][0] < order[j][0]
		}
		return order[i][1] < order[j][1]
	})
	rows := make([]store.RubyReading, 0, len(order))
	for _, key := range order {
		a := seen[key]
		rows = append(rows, store.RubyReading{
			BookID: r.Book.BookID, Base: key[0], Reading: key[1],
			FirstChapter: a.first, Occurrences: a.count,
		})
	}
	return r.Store.ReplaceRubyReadings(r.Book.BookID, rows)
}

// translateChunk drives ONE chunk through the stage list. Stages run in order,
// each feeding the next; the FIRST flagged stage stops the chunk — later stages
// are recorded `skipped` (no paid edit over a garbage draft, D2). It also runs the
// memory bank v2 hot path: it Selects the glossary injection once ($0, deterministic),
// injects it into the TRANSLATOR stage, post-checks the translator output (E1), and
// persists the per-chunk retrieval-state (observability). Returns the chunk outcome, the
// exact-matched entity ids (the next chunk's sticky_prev, A5), and an error only on an
// infra failure. stickyPrev is the prior chunks' exact matches in this chapter.
func (r *Runner) translateChunk(ctx context.Context, snapID string, ch Chunk, stickyPrev map[string]injectionDisposition) (*ChunkOutcome, map[string]injectionDisposition, error) {
	out := &ChunkOutcome{Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx, Disposition: DispOK}
	prev := ""
	flagged := false
	var flagReason FlagReason

	// Hot path: select the glossary records for this chunk ONCE (deterministic, $0), and
	// serialize the translator injection block. Recomputed on every run (resumed chunks
	// included) so the sticky window and the injected bytes are resume-stable.
	var memSel memorySelection
	var translatorInjection, editorInjection string
	activeIDs := map[string]injectionDisposition{}
	if r.memory != nil {
		memSel = r.memory.Select(ch.Text, ch.Chapter, stickyPrev, r.Pipeline.Context.GlossaryTokenBudget)
		translatorInjection = renderGlossaryBlock(memSel.injected)
		editorInjection = renderEditorConstraintBlock(memSel.injected)
		activeIDs = memSel.activeIDs
	}

	for stageIdx, st := range r.Pipeline.Stages {
		if flagged {
			// An earlier stage flagged → this stage is not attempted or billed.
			detail := fmt.Sprintf("skipped: an upstream stage was flagged (%s)", flagReason)
			if err := r.Store.UpsertChunkStatus(store.ChunkStatus{
				BookID: r.Book.BookID, Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx, Stage: st.Name,
				SnapshotID: snapID, Disposition: string(DispSkipped), FlagReason: string(flagReason),
				Detail: detail,
			}); err != nil {
				return out, activeIDs, fmt.Errorf("pipeline: record skipped chunk_status: %w", err)
			}
			out.Stages = append(out.Stages, StageResult{
				Stage: st.Name, Role: st.Role, Model: st.Model,
				Disposition: DispSkipped, FlagReason: flagReason, Detail: detail,
			})
			continue
		}

		// Inject the glossary per role. The TRANSLATOR gets the src→dst block (its keys are
		// matched against the source chunk); the monolingual EDITOR gets the CONFIRMED dst
		// forms as target-consistency constraints (no source — decisions-log D1); every other
		// stage gets none. Empty injection → the plain 2-message layout. The injection is a
		// message, so it enters this stage's request_hash automatically (a resumed chunk
		// reproduces it deterministically from the frozen bank).
		injection := ""
		switch st.Role {
		case roleTranslator:
			injection = translatorInjection
		case roleEditor:
			injection = editorInjection
		}
		sr, err := r.runStage(ctx, st, stageIdx, snapID, ch, prev, injection)
		if err != nil {
			return out, activeIDs, err
		}
		out.Stages = append(out.Stages, *sr)
		out.CostUSD += sr.CostUSD
		if sr.Disposition == DispFlagged {
			flagged = true
			flagReason = sr.FlagReason
			continue
		}
		prev = sr.Text
	}

	// Post-check the FINAL, exported output (E1): the reader sees the last stage's text
	// (the editor's), not the translator draft — the monolingual editor can drift an
	// approved term the translator got right, so checking the draft alone would miss it.
	// Runs on the fresh OR fully-resumed chunk (both carry the final text through `prev`),
	// so it is resume-reproducible. In the default FLAGGER mode a miss is recorded only in
	// the retrieval-state (observability); with the opt-in gate a miss flags the chunk
	// (glossary_miss). Skipped when a stage already flagged (no usable output to check).
	var postMisses []postcheckMiss
	outputChecked := false
	if r.memory != nil && !flagged && prev != "" {
		outputChecked = true
		postMisses = r.memory.postcheck(memSel.injected, prev)
		// The gate flips ONLY on CONFIRMED (approved) misses — an AMBIGUOUS miss is an
		// unverified candidate the model may legitimately reject, so it must not discard a
		// correct translation (external-review major #1).
		if r.Pipeline.Gates.Glossary.PostcheckGate && countConfirmedMisses(postMisses) > 0 {
			flagged = true
			flagReason = FlagGlossaryMiss
			r.Log.WarnContext(ctx, "glossary post-check gate flagged the chunk",
				"chapter", ch.Chapter, "chunk", ch.ChunkIdx, "confirmed_misses", countConfirmedMisses(postMisses))
		}
	}

	// Cheap deterministic style/number flaggers on the FINAL text (cheapgates.go): observability
	// only, never a disposition. Runs on the same fresh-or-resumed output as the post-check, so it
	// is resume-reproducible; skipped when a stage flagged (no usable output). Source is ch.Text
	// (the 万/億 magnitude gate compares source↔output).
	var cheap cheapGateResult
	if !flagged && prev != "" {
		cheap = runCheapGates(ch.Text, prev, r.cheapGateConfig())
		if cheap.total() > 0 {
			r.Log.InfoContext(ctx, "cheap style gates flagged the chunk (observability, not a gate)",
				"chapter", ch.Chapter, "chunk", ch.ChunkIdx, "style_flags", cheap.total(),
				"dialogue_dash", cheap.DialogueDash, "yo", cheap.YoInconsistent,
				"translit_interj", cheap.TranslitInterj, "number_magnitude", cheap.NumberMagnitude)
		}
	}

	// Persist the per-chunk retrieval-state (registry gate #4: convert silent memory
	// degradation into a loud, visible record) plus the cheap style-flag counts. Deterministic +
	// idempotent, so a resume re-derives the identical row. Only when a glossary is materialized
	// (always true in a normal run — seedGlossary sets an at-least-empty bank).
	if r.memory != nil {
		if err := r.persistRetrievalState(snapID, ch, memSel, postMisses, outputChecked, cheap); err != nil {
			return out, activeIDs, err
		}
	}

	if flagged {
		out.Disposition = DispFlagged
		out.FlagReason = flagReason
		out.FinalText = "" // garbage/refusal/glossary-miss never propagates to the next chunk or export
	} else {
		out.FinalText = prev
	}
	return out, activeIDs, nil
}

// persistRetrievalState writes the per-chunk observability record from the deterministic
// selection + the post-check result. n_exact_hits/n_sticky/n_ambiguous count the INJECTED
// records (what the model saw); spoiler/eviction are the dropped-and-logged totals;
// post-check misses are recorded only when the translator actually produced text.
func (r *Runner) persistRetrievalState(snapID string, ch Chunk, sel memorySelection, misses []postcheckMiss, outputChecked bool, cheap cheapGateResult) error {
	rs := store.RetrievalState{
		BookID: r.Book.BookID, Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx, SnapshotID: snapID,
		NStyleFlags: cheap.total(),
	}
	if cheap.total() > 0 {
		if b, err := json.Marshal(cheap); err == nil {
			rs.StyleDetail = string(b)
		}
	}
	for _, p := range sel.injected {
		if p.via == "sticky" {
			rs.NSticky++
		} else {
			rs.NExactHits++
		}
		if p.disp == memAmbiguous {
			rs.NAmbiguousFlagged++
		}
	}
	rs.NSpoilerBlocked = len(sel.rejected)
	rs.NEvicted = len(sel.evicted)
	if outputChecked {
		// n_postcheck_miss is the CONFIRMED-miss count (the actionable consistency-failure
		// signal, external-review #1); the detail carries ALL misses (confirmed AND the
		// ambiguous forced-post-check ones, each disp-tagged) for the human.
		rs.NPostcheckMiss = countConfirmedMisses(misses)
		if len(misses) > 0 {
			if b, err := json.Marshal(misses); err == nil {
				rs.PostcheckDetail = string(b)
			}
		}
	}
	ids := make([]string, 0, len(sel.activeIDs))
	for id := range sel.activeIDs {
		ids = append(ids, id)
	}
	sort.Strings(ids)
	if b, err := json.Marshal(ids); err == nil {
		rs.InjectedIDs = string(b)
	}
	return r.Store.UpsertRetrievalState(rs)
}

// cheapGateConfig builds the cheap-gate knobs from the book brief: the ё-policy and the
// lower-cased per-project interjection allowlist. Pure, no store access.
func (r *Runner) cheapGateConfig() cheapGateConfig {
	allow := make(map[string]bool, len(r.Book.StyleAllowlist))
	for _, s := range r.Book.StyleAllowlist {
		allow[s] = true
	}
	return cheapGateConfig{yoPolicy: r.Book.YoPolicy, allowlist: allow}
}

// runStage runs ONE stage of ONE chunk to a terminal disposition. It first
// resumes a resolved verdict (chunk_status read BEFORE any render — anti-wedge
// #1), otherwise walks the attempt axis: render → per-attempt checkpoint resume
// or a fresh reserve/call/settle → classify → retry the retryable {length,empty}
// subset with a doubled budget up to the regenerate cap, then flag. Returns an
// error ONLY on an infra failure; a bad completion is a disposition, never an
// error.
func (r *Runner) runStage(ctx context.Context, st config.Stage, stageIdx int, snapID string, ch Chunk, prev, injection string) (*StageResult, error) {
	job, err := r.Store.EnsureJob(r.Book.BookID, ch.Chapter, st.Name, snapID)
	if err != nil {
		return nil, err
	}
	// Snapshot pinning (Р6): a job frozen on a stale snapshot is a loud stop
	// unless --resnapshot explicitly accepts the re-translation.
	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, ch.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])
	}

	// Render the wire messages up front. This is cheap (string substitution — the
	// expensive part is the LLM call, still gated below) and it lets the resume
	// fast-path be CONTENT-VERIFIED: the source bytes are not in the snapshot, so
	// a positional chunk_status row must be checked against the current rendered
	// content, else an edited source would serve a stale, divergent translation.
	msgs, err := MessagesWithInjection(r.templates[st.Name], RenderVars{Book: r.Book, Text: ch.Text, Draft: prev}, injection)
	if err != nil {
		return nil, err
	}
	contentHash := msgsContentHash(msgs)

	// Anti-wedge #1 + content-guard (self-review): resume resolves from
	// chunk_status BEFORE any LLM call. A flagged chunk is not in TM (garbage
	// never commits), so absence-in-TM ≠ "not done" — the disposition row is the
	// resume authority, and a terminally-flagged chunk is NOT re-attacked (no
	// infinite paid loop, even if the regenerate budget was raised). The row is
	// trusted ONLY when snapshot AND content both match: a stale-snapshot row
	// (post --resnapshot) or a stale-content row (source edited, not in the
	// snapshot) is ignored → the attempt loop below re-derives it content-safely
	// (a source edit becomes a silent per-chunk re-translate, §3.4, not a stale
	// serve). A `skipped` row is ignored too: translateChunk re-derives the skip.
	if cs, err := r.Store.GetChunkStatus(r.Book.BookID, ch.Chapter, ch.ChunkIdx, st.Name); err != nil {
		return nil, err
	} else if cs != nil && cs.SnapshotID == snapID && cs.ContentHash == contentHash && cs.Disposition != string(DispSkipped) {
		return r.resumeFromChunkStatus(ctx, st, ch, cs)
	}

	// max_tokens base is sized from the text THIS stage processes: the source for
	// the translator, the prior draft for later stages (D2.5 — the monolingual
	// editor works over the Russian draft ≈1.9× the CJK source, so sizing edit
	// from source under-budgets and false-triggers a length retry).
	sizingText := ch.Text
	if stageIdx > 0 {
		sizingText = prev
	}
	baseMaxTokens := int(float64(EstimateTokens(sizingText)) * r.Pipeline.Defaults.MaxOutputRatio)
	if baseMaxTokens < r.Pipeline.Defaults.MinMaxTokens {
		baseMaxTokens = r.Pipeline.Defaults.MinMaxTokens
	}

	// Дополняем ReqInfo, СОХРАНЯЯ решения допуска (LogBodies) — перезапись с нуля
	// отрезала бы задокументированный debug-канал (находка ревью).
	ri, _ := obs.ReqInfoFromContext(ctx)
	ri.Book, ri.Chapter, ri.Chunk, ri.Stage, ri.Role = r.Book.BookID, ch.Chapter, ch.ChunkIdx, st.Name, st.Role
	ctx = obs.WithReqInfo(ctx, ri)

	maxRegen := r.Pipeline.Retries.RegenerateBeforeEscalate
	if maxRegen < 0 {
		maxRegen = 0
	}

	var cumCost, runCost float64
	var last stageAttempt
	anyFresh := false
	attemptsMade := 0
	for attempt := 0; ; attempt++ {
		maxTokens := maxTokensForAttempt(baseMaxTokens, attempt)
		att, err := r.runAttempt(ctx, st, st.Model, snapID, ch, job, attempt, maxTokens, msgs, false)
		if err != nil {
			return nil, err // infra failure
		}
		attemptsMade = attempt + 1
		cumCost += att.cumCost
		runCost += att.runCost
		anyFresh = anyFresh || att.freshCall
		last = att
		if att.cls.ok() {
			break
		}
		// Flagged: re-attack only the retryable subset, only while regenerations
		// remain (a bigger budget on the attempt axis, D2.3). Everything else is
		// deterministic — a same-model retry would re-refuse and re-bill (D2.2).
		if att.cls.Reason.retryable() && attempt < maxRegen {
			r.Log.WarnContext(ctx, "stage flagged, regenerating with a larger budget",
				"stage", st.Name, "chapter", ch.Chapter, "chunk", ch.ChunkIdx,
				"attempt", attempt, "reason", string(att.cls.Reason), "next_max_tokens", maxTokensForAttempt(baseMaxTokens, attempt+1))
			continue
		}
		break
	}

	// Single-hop escalation (D12 deterministic-content-failure class): a flag that a
	// DIFFERENT model might fix (echo / excision / refusal) is routed ONCE to the
	// stage's named fallback, under the book's escalation.budget_usd, and RE-GATED
	// (classifyOutput runs on the fallback output too — §3.8). The editor declares no
	// escalate_to → pinned per book (its style must not drift to a foreign model,
	// D12/2605.13368). The fallback uses its OWN model (the request_hash axis), so its
	// checkpoint never collides with the failed primary's, and it is exactly ONE call
	// (not a fresh retry budget): total calls/chunk = primary attempts + 1 hop, never
	// reset on the fallback (the LiteLLM #19985 retry×fallback blow-up guard).
	escalated, escModel := false, ""
	if last.cls.Reason.escalatable() && st.EscalateTo != "" {
		// Idempotency (self-review): if the fallback hop already happened its
		// checkpoint exists and was already paid — REPLAY it for free regardless of the
		// budget, so a crash AFTER the hop settled but BEFORE chunk_status was written
		// re-serves it on resume rather than discarding a paid, successful translation
		// and flipping the verdict OK→flagged. Only a FRESH hop is budget-gated. The
		// hash mirrors runAttempt's for (model=EscalateTo, attempt=0, maxTokens=base).
		fbHash := RequestHash(r.Book.BookID, ch.Chapter, ch.ChunkIdx, 0, st.Name, st.Role, st.EscalateTo,
			st.Temperature, st.Reasoning, false, baseMaxTokens, snapID, msgs)
		fbExists, err := r.Store.GetCheckpoint(fbHash)
		if err != nil {
			return nil, err
		}
		mayHop := fbExists != nil
		if !mayHop {
			if mayHop, err = r.escalationBudgetRemains(); err != nil {
				return nil, err
			}
		}
		if mayHop {
			fb, err := r.runAttempt(ctx, st, st.EscalateTo, snapID, ch, job, 0, baseMaxTokens, msgs, true)
			if err != nil {
				// An OPTIONAL hop that trips a USD ceiling must NOT abort the whole book
				// (and re-abort on every resume): the chunk is already flagged, so keep
				// the primary flag and continue. Any other infra error still aborts.
				if !errors.Is(err, errReserveCeiling) {
					return nil, err
				}
				r.Log.WarnContext(ctx, "escalation hop denied by a USD ceiling; keeping the primary flag",
					"stage", st.Name, "chapter", ch.Chapter, "chunk", ch.ChunkIdx, "reason", string(last.cls.Reason))
			} else {
				cumCost += fb.cumCost
				runCost += fb.runCost
				anyFresh = anyFresh || fb.freshCall
				escalated, escModel = true, st.EscalateTo
				r.Log.WarnContext(ctx, "stage escalated to a fallback model",
					"stage", st.Name, "chapter", ch.Chapter, "chunk", ch.ChunkIdx,
					"primary_reason", string(last.cls.Reason), "fallback", st.EscalateTo,
					"fallback_disposition", string(fb.cls.Reason.disposition()))
				if fb.cls.ok() {
					last = fb // the fallback draft passed the re-gate → it is authoritative
				}
				// else: the fallback also failed → keep the primary flag (last unchanged);
				// the fallback call is billed and counted, the chunk stays flagged (1 hop).
			}
		}
	}

	disposition := last.cls.Reason.disposition()
	finalHash := ""
	if disposition == DispOK {
		finalHash = last.reqHash
	}
	// chunk_status is a RESOLVE over the checkpoints: cost_usd sums every attempt
	// (F3-honest — retries are counted), final_hash points at the authoritative
	// checkpoint the ok path serves on resume.
	if err := r.Store.UpsertChunkStatus(store.ChunkStatus{
		BookID: r.Book.BookID, Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx, Stage: st.Name,
		SnapshotID: snapID, ContentHash: contentHash, Disposition: string(disposition), FlagReason: string(last.cls.Reason),
		Attempts: attemptsMade, FinalHash: finalHash, CostUSD: cumCost, Detail: last.cls.Detail,
		Escalated: escalated, EscalationModel: escModel,
	}); err != nil {
		return nil, fmt.Errorf("pipeline: record chunk_status: %w", err)
	}

	sr := &StageResult{
		Stage: st.Name, Role: st.Role, Model: last.modelActual,
		FromResume:      !anyFresh,
		Usage:           last.usage,
		CostUSD:         runCost,
		CumCostUSD:      cumCost,
		LatencyMS:       last.latency,
		FinishReason:    last.finish,
		Disposition:     disposition,
		FlagReason:      last.cls.Reason,
		Detail:          last.cls.Detail,
		Attempts:        attemptsMade,
		Escalated:       escalated,
		EscalationModel: escModel,
	}
	if disposition == DispOK {
		sr.Text = last.text
	} else {
		r.Log.WarnContext(ctx, "stage flagged", "stage", st.Name, "chapter", ch.Chapter,
			"chunk", ch.ChunkIdx, "reason", string(last.cls.Reason), "attempts", attemptsMade,
			"cum_cost_usd", fmt.Sprintf("%.6f", cumCost))
	}
	return sr, nil
}

// stageAttempt is the result of one attempt (a checkpoint hit or a fresh call).
type stageAttempt struct {
	reqHash     string
	cls         classification
	text        string
	usage       llm.Usage
	finish      string
	modelActual string
	latency     int
	runCost     float64 // billed THIS run (0 on a checkpoint hit)
	cumCost     float64 // this attempt's cost (checkpoint cost on a hit, fresh cost on a call)
	freshCall   bool    // a provider call was made this run
}

// runAttempt executes exactly one attempt on the request-hash axis: a checkpoint
// hit is classified for free (self-heal, incl. legacy Фаза-0 empty/decode
// checkpoints — no re-billing), otherwise a fresh reserve → call → settle+
// checkpoint. It returns an error only on an infra failure; a bad completion
// comes back as a classification on the attempt.
func (r *Runner) runAttempt(ctx context.Context, st config.Stage, model, snapID string, ch Chunk, job *store.Job, attempt, maxTokens int, msgs []llm.Message, escalation bool) (stageAttempt, error) {
	reqHash := RequestHash(r.Book.BookID, ch.Chapter, ch.ChunkIdx, attempt, st.Name, st.Role, model,
		st.Temperature, st.Reasoning, false, maxTokens, snapID, msgs)
	att := stageAttempt{reqHash: reqHash, modelActual: model}

	// Resume on the attempt axis: a checkpoint means THIS attempt already happened
	// and was billed — classify its text and never re-bill (kill -9 loses ≤1 call;
	// legacy empty/decode checkpoints self-heal by classification, not a crash).
	if cp, err := r.Store.GetCheckpoint(reqHash); err != nil {
		return att, err
	} else if cp != nil {
		var usage llm.Usage
		_ = json.Unmarshal([]byte(cp.UsageJSON), &usage)
		att.usage = usage
		att.text = cp.ResponseText
		att.finish = cp.FinishReason
		att.modelActual = cp.ModelActual
		att.cumCost = cp.CostUSD
		att.cls = r.classifyOutput(st.Role, ch.Text, cp.ResponseText, cp.FinishReason)
		r.Store.LogRequest(ctx, r.Log, store.RequestLog{
			BookID: r.Book.BookID, Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx,
			Stage: st.Name, Role: st.Role, ModelRequested: model, ModelActual: cp.ModelActual,
			RequestHash: reqHash, TMHit: true, OK: att.cls.ok(), FinishReason: cp.FinishReason,
			Degraded: degradedTag(att.cls),
		})
		_ = r.Store.SetJobStatus(job.ID, "done")
		r.Log.InfoContext(ctx, "attempt served from checkpoint", "stage", st.Name,
			"attempt", attempt, "hash", reqHash[:12], "disposition", string(att.cls.Reason.disposition()))
		return att, nil
	}

	// Fresh call: reserve → call → settle+checkpoint (atomic, §3.3).
	price := r.Pricer.PriceFor(model)
	promptEst := 0
	for _, m := range msgs {
		promptEst += EstimateTokens(m.Content)
	}
	// D13.6: reserve the additive reasoning buffer for the ACTUAL model being called (the
	// escalate_to fallback may differ from st.Model and have its own provider). 0 for
	// subset providers / reasoning-off — the ceiling then sees only completion, as before.
	reasoningBudget := r.Models.AdditiveReasoningTokens(model, st.Reasoning, st.ReasoningMaxTokens)
	estimate := ledger.EstimateUSD(price, promptEst, maxTokens, reasoningBudget)

	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 att, err
	}
	switch verdict {
	case store.ReserveDeniedBook:
		// Ceiling is a hard, book-wide stop (not a per-chunk flag): the job stays
		// 'pending' and resume continues once the ceiling is raised. Wrapped so an
		// optional escalation hop can degrade instead of aborting the book.
		return att, fmt.Errorf("pipeline: book USD ceiling reached (%.2f$) — raise ceilings.book_usd or stop: %w", r.Book.Ceilings.BookUSD, errReserveCeiling)
	case store.ReserveDeniedDay:
		return att, fmt.Errorf("pipeline: daily USD ceiling reached (%.2f$): %w", r.Book.Ceilings.DayUSD, errReserveCeiling)
	}

	client, err := r.client(model)
	if err != nil {
		_ = r.Store.Release(resv)
		_ = r.Store.SetJobStatus(job.ID, "failed")
		return att, err
	}
	if err := r.Store.SetJobStatus(job.ID, "running"); err != nil {
		_ = r.Store.Release(resv)
		return att, err
	}
	att.freshCall = true

	start := time.Now()
	resp, err := client.Complete(ctx, llm.LLMRequest{
		Model:           model,
		Messages:        msgs,
		MaxTokens:       maxTokens,
		Temperature:     st.Temperature,
		ReasoningEffort: st.Reasoning,
	})
	att.latency = int(time.Since(start).Milliseconds())
	if err != nil {
		var bde *llm.BilledDecodeError
		if errors.As(err, &bde) {
			// 2xx with an unreadable body: the provider ALREADY billed. Do not
			// release; conservatively settle the ESTIMATE with an empty decode
			// checkpoint, then FLAG (anti-wedge #2 — the loop continues, this is
			// not an infra crash). A settle failure here is a real infra fault.
			if serr := r.Store.SettleWithCheckpoint(resv, estimate, store.Checkpoint{
				RequestHash: reqHash, JobID: job.ID, ChunkIdx: ch.ChunkIdx, Attempt: attempt,
				Stage: st.Name, Role: st.Role, ModelRequested: model, ModelActual: model,
				ResponseText: "", UsageJSON: "{}", CostUSD: estimate, FinishReason: decodeErrorFinish,
				Escalation: escalation,
			}); serr != nil {
				return att, fmt.Errorf("pipeline: settle after billed decode failure: %w", serr)
			}
			att.finish = decodeErrorFinish
			att.cumCost, att.runCost = estimate, estimate
			att.cls = classification{FlagDecodeError, "billed 2xx with an unreadable body: " + err.Error()}
			r.Store.LogRequest(ctx, r.Log, store.RequestLog{
				BookID: r.Book.BookID, Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx,
				Stage: st.Name, Role: st.Role, ModelRequested: model,
				RequestHash: reqHash, CostUSD: estimate, LatencyMS: att.latency,
				Degraded: "billed_2xx_decode_failed", Err: err.Error(), OK: false, FinishReason: decodeErrorFinish,
			})
			_ = r.Store.SetJobStatus(job.ID, "done")
			return att, nil
		}
		// No 2xx ever arrived: nothing was billed. Release and surface as an INFRA
		// failure — a long book pauses/resumes on an outage or terminal 4xx (D4),
		// rather than flag-storming every remaining chunk on a dead provider.
		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: ch.Chapter, ChunkIdx: ch.ChunkIdx,
			Stage: st.Name, Role: st.Role, ModelRequested: model,
			RequestHash: reqHash, LatencyMS: att.latency, Err: err.Error(), OK: false,
		})
		_ = r.Store.SetJobStatus(job.ID, "failed")
		return att, fmt.Errorf("pipeline: stage %s call: %w", st.Name, err)
	}

	modelActual := resp.Model
	if modelActual == "" {
		modelActual = model
	}
	att.modelActual = modelActual
	att.text = resp.Text
	att.finish = resp.FinishReason
	att.usage = resp.Usage

	// Цена по фактически ответившей модели, с fallback на ЗАПРОШЕННУЮ (не на
	// дешёвый глобальный якорь), если провайдер вернул канонизированный слаг.
	price = r.Pricer.PriceForResponse(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 att, err
	}
	att.cumCost, att.runCost = cost, cost

	// Деньги: settle + сырой ответ — одна транзакция (§3.3). ВСЕГДА, даже для
	// пустого/усечённого/отказного ответа: он оплачен провайдером, чекпоинт
	// хранит его до цента, а годность решает classify ПОСЛЕ (деньги ≠ вердикт).
	if err := r.Store.SettleWithCheckpoint(resv, cost, store.Checkpoint{
		RequestHash: reqHash, JobID: job.ID, ChunkIdx: ch.ChunkIdx, Attempt: attempt,
		Stage: st.Name, Role: st.Role,
		ModelRequested: model, ModelActual: modelActual, ResponseText: resp.Text,
		UsageJSON: string(usageJSON), CostUSD: cost, FinishReason: resp.FinishReason,
		ProviderRequestID: resp.ProviderRequestID, Escalation: escalation,
	}); 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 att, fmt.Errorf("pipeline: settle stage %s: %w", st.Name, err)
	}

	// Classify AFTER the money is durably settled+checkpointed. F4 lives here: a
	// non-empty truncated length draft is now classified (flagged/retried), never
	// silently passed downstream as OK; an empty completion is flagged too, not a
	// run-crash.
	att.cls = r.classifyOutput(st.Role, ch.Text, resp.Text, resp.FinishReason)

	r.Store.LogRequest(ctx, r.Log, store.RequestLog{
		BookID: r.Book.BookID, Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx,
		Stage: st.Name, Role: st.Role, ModelRequested: 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: att.latency, FinishReason: resp.FinishReason,
		OK: att.cls.ok(), Degraded: degradedTag(att.cls),
	})
	_ = r.Store.SetJobStatus(job.ID, "done")
	r.Log.InfoContext(ctx, "attempt completed", "stage", st.Name, "attempt", attempt, "model", modelActual,
		"cost_usd", fmt.Sprintf("%.6f", cost), "latency_ms", att.latency,
		"disposition", string(att.cls.Reason.disposition()), "reason", string(att.cls.Reason))
	return att, nil
}

// resumeFromChunkStatus serves a chunk×stage whose disposition is already
// resolved (read before any render — no re-billing). An ok stage returns its
// authoritative checkpoint text to feed the next stage; a flagged stage returns
// the flag WITHOUT re-attacking (a terminal flag never re-enters the paid loop).
func (r *Runner) resumeFromChunkStatus(ctx context.Context, st config.Stage, ch Chunk, cs *store.ChunkStatus) (*StageResult, error) {
	sr := &StageResult{
		Stage: st.Name, Role: st.Role, Model: st.Model, FromResume: true,
		CostUSD: 0, CumCostUSD: cs.CostUSD,
		Disposition: Disposition(cs.Disposition), FlagReason: FlagReason(cs.FlagReason),
		Detail: cs.Detail, Attempts: cs.Attempts,
		// Escalation attribution now travels on chunk_status (D15.3), so it is preserved
		// across resume for BOTH an ok escalation AND a flagged-then-escalated chunk (the
		// status read-model reads it without re-opening the checkpoint).
		Escalated: cs.Escalated, EscalationModel: cs.EscalationModel,
	}
	if cs.Disposition == string(DispOK) {
		cp, err := r.Store.GetCheckpoint(cs.FinalHash)
		if err != nil {
			return nil, err
		}
		if cp == nil || strings.TrimSpace(cp.ResponseText) == "" {
			// chunk_status ok is only written AFTER its checkpoint is durably
			// settled, so this is a torn/tampered store — fail loud rather than
			// feed the next stage an empty draft.
			return nil, fmt.Errorf("pipeline: chunk_status ok for %s/ch%d/chunk%d/%s references checkpoint %.12s which is missing/empty — inconsistent store",
				r.Book.BookID, ch.Chapter, ch.ChunkIdx, st.Name, cs.FinalHash)
		}
		var usage llm.Usage
		_ = json.Unmarshal([]byte(cp.UsageJSON), &usage)
		sr.Usage = usage
		sr.Model = cp.ModelActual
		sr.FinishReason = cp.FinishReason
		sr.Text = cp.ResponseText
	}
	r.Store.LogRequest(ctx, r.Log, store.RequestLog{
		BookID: r.Book.BookID, Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx,
		Stage: st.Name, Role: st.Role, ModelRequested: st.Model, ModelActual: sr.Model,
		RequestHash: cs.FinalHash, TMHit: true, OK: cs.Disposition == string(DispOK),
		FinishReason: sr.FinishReason, Degraded: nonOKTag(cs.Disposition, cs.FlagReason),
	})
	r.Log.InfoContext(ctx, "stage resolved from chunk_status", "stage", st.Name,
		"chapter", ch.Chapter, "chunk", ch.ChunkIdx, "disposition", cs.Disposition, "reason", cs.FlagReason)
	return sr, nil
}

// degradedTag surfaces a flag reason into the request_log `degraded` column.
func degradedTag(cls classification) string {
	if cls.ok() {
		return ""
	}
	return string(cls.Reason)
}

// nonOKTag is the request_log `degraded` value for a resumed disposition row.
func nonOKTag(disposition, reason string) string {
	if disposition == string(DispOK) {
		return ""
	}
	return reason
}