textmachine/backend/internal/pipeline/terminologist.go

package pipeline

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"strings"

	"textmachine/backend/internal/chunk"
	"textmachine/backend/internal/config"
	"textmachine/backend/internal/lang"
	"textmachine/backend/internal/llm"
	"textmachine/backend/internal/membank"
	"textmachine/backend/internal/miner"
	"textmachine/backend/internal/obs"
	"textmachine/backend/internal/store"
	"textmachine/backend/internal/terminology"
	"textmachine/backend/internal/text"
)

// terminologist.go: the TERMINOLOGIST role (pack-20, ratified D39.42 п.1) — the step that was missing
// between "which surfaces belong in the bank" and "what does this book call them".
//
// WHERE IT SITS. At the bank-mining stop, after the draft wave has produced the whole book and before the
// stop/auto decision. That position is not a convenience: it is the only moment at which BOTH the complete
// source and a complete draft of every chapter exist, which is exactly what translating a term correctly
// needs. Owner, 26.07: «намайнить в банк можем весь контекст, потому что есть черновые переводы всей книги
// сразу … дальше отрабатывает модель, которая видит весь контекст и переводит весь банк памяти сразу».
//
// WHAT IT MAY DO. Propose ONE consolidated rendering per candidate. It cannot approve anything: the term it
// touches emits `draft` (the §C2-7 mode that carries a rendering), never `approved` — the miner has no
// code path to that word and this role adds none. A term it declines emits `auto`, inert. So the worst a
// bad terminologist call can do is put a marked, unverified proposal in front of the editor and the owner;
// it can never make the book use a word nobody signed.
//
// WHAT IT COSTS. A handful of batched calls per BOOK (not per chunk), on a cheap model, over a block whose
// size is bounded by config. The estimate is logged BEFORE the first call, the calls run on the shared
// reserve→call→settle+checkpoint path so a resume replays them for $0, and the whole class is capped by
// gates.terminology.budget_usd.

// roleTerminologist is the synthetic stage role every terminologist call is addressed under. Like the
// repair role it keeps the call on its own request-hash axis and gives its checkpoints, request_log rows
// and ledger entries their own queryable class — the cost marker, so "what did terminology spend" is
// answerable with no migration.
const roleTerminologist = "terminologist"

// roleClassifier is the §2 type-classifier phase's cost axis. Its own role marker keeps its checkpoints,
// request_log rows and ledger entries on a queryable class of their own and on their own request-hash axis,
// separate from the render phase — "what did classification spend" is answerable with no migration.
const roleClassifier = "classifier"

// terminologyStageName is the synthetic stage name the calls carry. It is NOT a pipeline stage (a stage
// would join a wave, take a snapshot axis and a chunk_status row per unit); it is an addressing label.
const terminologyStageName = "terminology"

// terminologyVersion versions the ASSEMBLY algorithm (merge → KWIC → §C2-3 ranking → batching → parse). It
// is deliberately NOT snapshot-folded — see config.TerminologyGate — but it is logged with the run so a
// signature map can be attributed to the algorithm that produced it.
const terminologyVersion = "terminology-v3-merge+kwic+c2-3+series+families+formfold"

// Engine defaults for the block sizing. They bound ONE call's input; the whole book is covered by batching.
const (
	terminologyDefaultBatchRunes = 6000
	terminologyDefaultKWICPer    = 3
	terminologyDefaultKWICWidth  = 40
)

// terminologyResult is the run's outcome for the report and the logs. Derived from what happened; never
// stored as a row of its own.
type terminologyResult struct {
	Candidates int // merged candidates handed to the role
	Reverse    int // of those, banknote-only (the coverage the miner structurally cannot see)
	Batches    int // calls attempted
	// Gendered is how many candidates the CLASSIFIER gave a gender datum (backlog row 210). Reported
	// because a zero here and a zero in the bank are the same number for two entirely different reasons —
	// nobody asked (the classifier is off), or nobody answered — and the axis spent a whole era dead with
	// every counter beside it reading clean.
	Gendered int
	// BatchesDropped / ClassifyBatchesDropped are how many batches the BUDGET left unbought in each pass —
	// the difference between the pass a role planned and the pass it ran. Without them «consolidated=42
	// unanswered=16» reads as a verdict about the TERMS when it is partly a verdict about the MONEY.
	//
	// ⚠ TWO FIELDS AND NOT ONE, because the two passes have SEPARATE budgets (`budget_usd` and
	// `classify_budget_usd`) and the cold run's incident was on the CLASSIFIER: its $0.02 was exhausted
	// while the terminologist's $0.05 was not, and three of four classify batches were never bought. The
	// first version of this carrier reported only the render pass, so the very run that motivated it would
	// still have shown zero. Found by acceptance.
	BatchesDropped         int
	ClassifyBatchesDropped int
	Consolidated           int // terms that came back with a rendering → status:draft
	Declined               int // terms the role explicitly could not render → status:auto
	Unanswered             int // terms no reply line covered (also status:auto — silence is not a decision)
	BadLines               int // reply lines the parser refused
	// OffLanguage counts refused lines whose rendering was not in the target's script — separate from
	// BadLines because it means the model answered in another language, not that it broke the format.
	OffLanguage int
	// CanonConflicts counts consolidated renderings that contradict a row the owner already signed. It is
	// OBSERVABILITY, never a gate: the rows stay unverified either way, and this is what tells the owner
	// which of them to look at first.
	CanonConflicts int
	// SelfConflicts counts consolidations that contradict ANOTHER consolidation of the same run (§G2) — the
	// majority class on a live bank, and the one nothing looked at before this pack. Contradictions carries
	// the same finding per key, so the stop table can mark the rows instead of printing a bare total.
	SelfConflicts  int
	Contradictions map[string][]string
	// BankConflicts counts consolidated renderings that disagree with a row the bank already holds for the
	// same firing surface — the shape neither check above can see (membank.ConsolidationKeyConflicts).
	// BankHoldRows carries the findings themselves, ungrouped: bankStopRows matches them to its rows, so
	// the key they are looked up by is decided in one place instead of once on each side of a map. Kept
	// apart from Contradictions because the two are different decisions for the owner.
	BankConflicts int
	BankHoldRows  []membank.BankKeyConflict
	// Conf is the role's own stated confidence per key. It sorts the review list «least sure first» and does
	// nothing else — never a weight, a threshold or a cross-model comparison (D39.102).
	Conf        map[string]int
	CostUSD     float64 // what THIS run's RENDER phase paid
	CumUSD      float64 // what the render calls cost in total (a replayed checkpoint is $0 now, not then)
	Fresh       bool    // at least one call actually reached the provider this run
	EstimateUSD float64 // the pre-call projection, logged before any money moves
	// Reclassified is how many candidate types the §2 classifier phase actually changed; ClassifyCostUSD is
	// what that phase paid this run. Both zero when classify_types is off.
	Reclassified    int
	ClassifyCostUSD float64
	// Families is how many family GROUPS §G1 detected; FamiliesRefused how many of their merges the member
	// cap turned down, and FamiliesHeld how many a series held part of back. Both of the latter mean the same
	// thing to the owner — a family met split across two calls — and both are zero when the source declares
	// no family data.
	Families        int
	FamiliesRefused int
	FamiliesHeld    int
}

// loadTerminologyTemplate loads the pair's terminologist prompt when the gate is on. Mirrors
// loadRepairTemplates: a gate that cannot fire is refused at LOAD time, before any billing.
func (r *Runner) loadTerminologyTemplate() error {
	if !r.Pipeline.Gates.Terminology.Enabled {
		return nil
	}
	tpl, err := LoadPromptTemplate(r.Pipeline.Gates.Terminology.PromptPath)
	if err != nil {
		return err
	}
	r.terminologyTemplate = tpl
	if err := r.loadFamilyParams(); err != nil {
		return err
	}
	return r.loadClassifierTemplate()
}

// loadFamilyParams resolves the §G1 family channel from the SOURCE language's declared morphology, once, at
// LOAD time — before any billing, like every other gate precondition. A source with no family data yields a
// disabled channel and the batcher behaves exactly as it did before the channel existed.
//
// It is also where a data typo dies: the file names engine TYPES, and lang cannot check them against the
// engine's closed vocabulary without depending on this layer. A rule for a type nothing emits would parse
// fine and leave the channel quietly half-off — the same silent-empty-table class the pack loader refuses.
func (r *Runner) loadFamilyParams() error {
	fm := lang.FamilyMorphology(r.Book.SourceLang)
	if !fm.Enabled() {
		r.familyParams = terminology.FamilyParams{}
		return nil
	}
	sEnabled, headFinal := lang.SeriesMorphology(r.Book.SourceLang)
	p := terminology.FamilyParams{
		Enabled: sEnabled, HeadFinal: headFinal,
		Affix:      make(map[string]terminology.FamilyAffix, len(fm.Affix)),
		MinMembers: fm.MinMembers, MaxMembers: fm.MaxMembers, ContainmentRunes: fm.ContainmentRunes,
	}
	for typ, a := range fm.Affix {
		if !terminology.CandidateTypes[typ] {
			return fmt.Errorf("pipeline: the family morphology of source %q names type %q, which no candidate can carry (accepted: %s) — a typo here would parse fine and leave the family channel silently half-off",
				r.Book.SourceLang, typ, strings.Join(terminology.TypeNames(terminology.CandidateTypes), "|"))
		}
		p.Affix[typ] = terminology.FamilyAffix{Suffix: a.Suffix, MinRunes: a.MinRunes}
	}
	r.familyParams = p
	return nil
}

// loadClassifierTemplate loads the pair's §2 classifier prompt when classify_types is on. Off → nil, and the
// classifier phase is inert. Called from loadTerminologyTemplate: the classifier only exists as a phase of
// the terminology gate.
func (r *Runner) loadClassifierTemplate() error {
	if !r.Pipeline.Gates.Terminology.ClassifyTypes {
		return nil
	}
	tpl, err := LoadPromptTemplate(r.Pipeline.Gates.Terminology.ClassifyPromptPath)
	if err != nil {
		return err
	}
	r.classifierTemplate = tpl
	return nil
}

// loadTargetScript resolves the answer-language screen's script once. config refuses an unknown name for
// an enabled gate, so an unresolved script here means the book simply declared none — inert, and said out
// loud when a channel that would have used it is on.
func (r *Runner) loadTargetScript() {
	if name := r.Pipeline.Gates.Terminology.TargetScript; name != "" {
		r.targetScript, _ = terminology.ScriptByName(name)
	}
	if r.targetScript == nil && r.Pipeline.Gates.Banknote.Enabled {
		r.Log.Warn("the banknote channel is on but no gates.terminology.target_script is declared: draft-side proposals are NOT screened for the answer language, so a rendering in another script can enter the signature map and the auto-bank",
			"book", r.Book.BookID)
	}
}

// terminologyOpts resolves the sizing knobs in ONE place — explicit config, else the pair's own data,
// else the engine default — so batching and rendering can never disagree about what a batch is.
func (r *Runner) terminologyOpts() (batchRunes, kwicPer, kwicWidth int) {
	g := r.Pipeline.Gates.Terminology
	batchRunes, kwicPer, kwicWidth = g.BatchRunes, g.KWICPerTerm, g.KWICWidth
	if pair := r.packTerminology(); pair != nil {
		if kwicPer <= 0 {
			kwicPer = pair.KWICPerTerm
		}
		if kwicWidth <= 0 {
			kwicWidth = pair.KWICWidth
		}
	}
	if batchRunes <= 0 {
		batchRunes = terminologyDefaultBatchRunes
	}
	if kwicPer <= 0 {
		kwicPer = terminologyDefaultKWICPer
	}
	if kwicWidth <= 0 {
		kwicWidth = terminologyDefaultKWICWidth
	}
	return batchRunes, kwicPer, kwicWidth
}

func (r *Runner) packTerminology() *lang.TerminologySizing {
	if r.pack == nil {
		return nil
	}
	return r.pack.Terminology
}

// buildBankCandidates is the $0 half of the role: the two-way miner∪banknote merge, the source contexts,
// and the §C2-3 ranking of whatever the drafts already produced. It runs whether or not the gate is on —
// with the gate off nothing consumes the ranking, but building it costs nothing and it is what the stop's
// table is rendered from.
func (r *Runner) buildBankCandidates(mined []miner.Term, observed []terminology.Observed, chunks []terminology.Chunk) []terminology.Candidate {
	ms := make([]terminology.Mined, 0, len(mined))
	for _, m := range mined {
		ms = append(ms, terminology.Mined{
			Key: text.NormalizeSourceKey(m.Src), Src: m.Src, Type: m.Type,
			Freq: m.Freq, SinceCh: m.SinceCh, Aliases: m.Aliases, Evidence: m.Evidence,
		})
	}
	// The vote is counted per target-form CONVENTION, not per byte string (§G5): «Море истинной ци» and
	// «море истинной ци» are one decision, and splitting their evidence hands the §C2-3 frequency factor to
	// whichever spelling a chunk happened to repeat. The normalizer is the SAME one the post-check matches
	// against, so the fold cannot disagree with the check that reads the result.
	cands := terminology.Merge(ms, observed, text.NormalizeTargetForm)
	_, kwicPer, kwicWidth := r.terminologyOpts()
	cands = terminology.AttachKWIC(cands, chunks, kwicPer, kwicWidth)

	opts := r.scoreOpts()
	for i := range cands {
		terminology.ScoreVariants(&cands[i], opts)
	}
	return cands
}

// scoreOpts builds the §C2-3 scoring options — the approved-neighbour anchor and the pair's transliteration
// conformance, which routes on TYPE (only name/place answer to the transliteration convention). Extracted so
// the initial candidate build and the §2 post-classify re-score share ONE definition and can never disagree
// about how a variant is scored.
func (r *Runner) scoreOpts() terminology.ScoreOpts {
	opts := terminology.ScoreOpts{Neighbours: approvedNeighbours(r.glossaryRows())}
	if r.pack != nil {
		pack := r.pack
		opts.Conformance = func(dst, typ string) float64 {
			if typ != "name" && typ != "place" {
				return 0 // the convention only speaks about transliterated entities
			}
			return miner.PalladiusConformance(dst, pack)
		}
	}
	return opts
}

// glossaryRows is the book's stored bank, every row with its status and window. The anchor below narrows
// it to the signed rows; ConsolidationKeyConflicts needs it whole, since an unsigned row owes one
// rendering too. A read failure degrades rather than aborts — everything built on it improves a
// consolidation, none of it is a precondition — but it is LOGGED: a degraded read and a clean bank
// produce the same zero in every counter downstream.
func (r *Runner) glossaryRows() []store.GlossaryEntry {
	rows, err := r.Store.GlossaryForBook(r.Book.BookID)
	if err != nil {
		// Called from the scoring pass as well as the role, so it can repeat within one run; a repeated
		// line is cheaper than a zero that reads as a clean bank.
		r.Log.Warn("terminology: could not read the bank — whatever this call feeds goes silent (the canon anchor, the bank conflict check, or both), and its zero then means «not asked» rather than «nothing found»",
			"book", r.Book.BookID, "err", err)
		return nil
	}
	return rows
}

// approvedNeighbours is the already-signed bank, as the §C2-3 "agreement with approved siblings" anchor.
// Only APPROVED rows qualify: an unverified row agreeing with an unverified row is not evidence.
func approvedNeighbours(rows []store.GlossaryEntry) []terminology.Neighbour {
	out := make([]terminology.Neighbour, 0, len(rows))
	for _, e := range rows {
		if e.Status == "approved" && e.Dst != "" {
			out = append(out, terminology.Neighbour{Src: text.NormalizeSourceKey(e.Src), Dst: e.Dst})
		}
	}
	sort.Slice(out, func(i, j int) bool { return out[i].Src < out[j].Src })
	return out
}

// runTerminologist calls the role over the candidate list and returns key → consolidated rendering. With
// the gate off it is a no-op returning an empty map and a zero result, so every existing book takes a
// byte-identical path and pays nothing.
//
// A term absent from the returned map keeps NO dst — the auto branch of §C2-7. That is the deliberate
// reading of silence: a role that did not answer has not decided, and an undecided term must not enter the
// wire carrying a guess.
func (r *Runner) runTerminologist(ctx context.Context, snapID string, cands []terminology.Candidate) (consolidated, classified, gendered map[string]string, res terminologyResult, err error) {
	if !r.Pipeline.Gates.Terminology.Enabled || r.terminologyTemplate == nil || len(cands) == 0 {
		return nil, nil, nil, res, nil
	}
	res.Candidates = len(cands)
	for _, c := range cands {
		if c.Origin == terminology.OriginBanknote {
			res.Reverse++
		}
	}
	batchRunes, _, _ := r.terminologyOpts()

	// §2 type re-derivation, BEFORE the render: a focused pass re-classifies every candidate, so a realia
	// surface mistyped as a name (元石) is no longer FORCED to a transliteration. The corrected type routes
	// conformance (re-scored here), primes the wire block, and is returned so the caller stamps it as the
	// banked type. Off (or unanswered) → the heuristic draft type stands.
	classified, gendered, crun, cerr := r.runClassifier(ctx, snapID, cands)
	if cerr != nil {
		return nil, nil, nil, res, cerr
	}
	res.Gendered = len(gendered)
	res.ClassifyCostUSD = crun.costUSD
	// The classify pass has its OWN budget (classify_budget_usd), so it has its own cut — and the cold
	// run's incident was on THIS pass, not the render one.
	res.ClassifyBatchesDropped = crun.dropped
	if len(classified) > 0 {
		res.Reclassified = applyTypes(cands, classified)
		opts := r.scoreOpts()
		for i := range cands {
			terminology.ScoreVariants(&cands[i], opts)
		}
	}

	// §1 + §G1: co-batch each grade/rank SERIES and each term FAMILY so the role picks ONE generic head, and
	// ONE shared root, for the whole set. The pair-data — whether the source forms rune-morpheme series,
	// where the head sits, and which side of a surface carries a family's root — comes from the language
	// layer, so the batcher stays pair-agnostic and a source with neither takes a byte-identical path.
	sEnabled, headFinal := lang.SeriesMorphology(r.Book.SourceLang)
	seriesID := terminology.DetectSeries(cands, terminology.SeriesParams{Enabled: sEnabled, HeadFinal: headFinal})
	fp := r.familyParams
	fams := terminology.DetectFamilies(cands, fp)
	unitID, ustats := terminology.MergeUnits(seriesID, fams, fp)
	batches := terminology.Batch(cands, batchRunes, unitID)
	res.Batches = len(batches)
	res.Families, res.FamiliesRefused, res.FamiliesHeld = len(fams), ustats.Refused, ustats.Held
	if ustats.Refused > 0 || ustats.Held > 0 {
		// Either guard leaves a family split across calls — the very defect this channel exists to close — so
		// both are named rather than left to be inferred from a bank that disagrees with itself.
		r.Log.WarnContext(ctx, "terminology: some families were NOT co-batched whole — the member cap refused the merge, or a series with an unrelated root kept its members; those families can still disagree with themselves across calls",
			"book", r.Book.BookID, "refused_by_cap", ustats.Refused, "held_by_series", ustats.Held, "max_members", fp.MaxMembers)
	}
	// §1/§G1 keep a unit (or a lone candidate) WHOLE even past the budget — splitting one would defeat the
	// co-batching it exists for — so a co-batched grade set or an evidence-heavy single term can render
	// over the cap. That is deliberate but NOT silent: an over-cap unit strains the model's output ceiling (the
	// cap-8000 mine), so name it while the run can still be watched.
	for i, b := range batches {
		if br := terminology.BatchRunes(b); br > batchRunes {
			r.Log.WarnContext(ctx, "terminology: a batch renders OVER the size budget and is sent WHOLE (a series/family is co-batched by design, §1/§G1; a lone candidate cannot be split) — watch the model's output cap on this call",
				"book", r.Book.BookID, "batch", i, "terms", len(b), "batch_runes", br, "budget_runes", batchRunes)
		}
	}

	// The bank is read ONCE for the whole role, not per batch: it is the same law for every batch. The
	// signed rows are the anchor the model is shown; the whole set is what the conflict checks below read.
	bank := r.glossaryRows()
	canon := approvedNeighbours(bank)
	plan := bankRolePlan{
		role: roleTerminologist, budgetUSD: r.Pipeline.Gates.Terminology.BudgetUSD,
		messages: func(b []terminology.Candidate) ([]llm.Message, error) { return r.terminologyMessages(b, canon) },
	}
	run, rerr := r.runBankRoleBatches(ctx, snapID, plan, batches, "render")
	// The CUT is a property of the pass, and it has to reach the report: a bank that is partially
	// consolidated because the money ran out is a different object from one the role fully considered.
	res.BatchesDropped = run.dropped
	if rerr != nil {
		return nil, nil, nil, res, rerr
	}
	res.EstimateUSD, res.CostUSD, res.CumUSD, res.Fresh = run.estimateUSD, run.costUSD, run.cumUSD, run.fresh

	out := map[string]string{}
	res.Conf = map[string]int{}
	for i, b := range batches {
		if i >= run.attempted {
			break // the budget or a ceiling stopped the pass here; runBankRoleBatches already said so
		}
		if run.texts[i] == "" {
			// An EMPTY completion on a call that was actually made. It used to `continue` in silence, which is
			// how a paid batch that returned nothing became indistinguishable from «the role declined these
			// terms» — a DECISION in §C2-7 — with the run exiting 0.
			r.Log.WarnContext(ctx, "terminology: a paid batch came back with an EMPTY completion; its terms stay unconsolidated",
				"book", r.Book.BookID, "batch", i, "asked", len(b), "answered", 0)
			continue
		}
		got, conf, st := terminology.ParseReply(run.texts[i], candKeys(b), text.NormalizeSourceKey, r.targetScript)
		res.BadLines += st.Bad
		res.OffLanguage += st.OffLanguage
		// Asked-vs-answered per batch: the one number that separates «the model skipped half the block» from
		// «the parser refused half the lines», and neither is visible in a total.
		r.Log.InfoContext(ctx, "terminology render batch", "book", r.Book.BookID,
			"batch", i, "asked", len(b), "answered", len(got), "bad_lines", st.Bad, "reply_chars", len(run.texts[i]))
		// A batch answered largely in another language is the measured cold-start failure, not noise: say it
		// while the run is happening, naming the lines, because the terms themselves just stay auto.
		if st.OffLanguage > 0 {
			r.Log.WarnContext(ctx, "terminology: the model answered in another script; those lines are REFUSED (the terms stay unconsolidated) — a book with few signed rows gives the model no target-language anchor",
				"book", r.Book.BookID, "batch", i, "terms", len(b), "off_language", st.OffLanguage,
				"target_script", r.Pipeline.Gates.Terminology.TargetScript, "lines", strings.Join(st.OffLanguageSamples, "; "))
		}
		// A batch that came back with NOTHING usable is a paid call that bought no terminology. Silence here
		// would spend the money, mark every term "the role declined" and exit 0. Say it out loud; the run still
		// continues, because an unconsolidated bank is the state this step started from, not a broken one.
		if len(got) == 0 {
			r.Log.WarnContext(ctx, "terminology: a paid batch returned nothing the parser could use; its terms stay unconsolidated",
				"book", r.Book.BookID, "batch", i, "terms", len(b), "bad_lines", st.Bad, "reply_chars", len(run.texts[i]))
		}
		for k, v := range got {
			out[k] = v
		}
		for k, v := range conf {
			res.Conf[k] = v
		}
	}
	for _, c := range cands {
		v, answered := out[c.Key]
		switch {
		case !answered:
			res.Unanswered++
		case v == "":
			res.Declined++
		default:
			res.Consolidated++
		}
	}
	// The canon check is $0 and runs on EVERY consolidation, signed anchor or not: the anchor is a nudge to
	// a model, and a nudge that is not verified is a hope. Named terms, not a bare count — a conflict is
	// actionable only if the operator is told which rendering fights which signature.
	if conflicts := terminology.CanonConflicts(cands, out, canon); len(conflicts) > 0 {
		res.CanonConflicts = len(conflicts)
		named := make([]string, 0, len(conflicts))
		for _, cf := range conflicts {
			named = append(named, fmt.Sprintf("%s→%q contradicts the signed %s→%q", cf.Src, cf.Dst, cf.CanonSrc, cf.CanonDst))
		}
		r.Log.WarnContext(ctx, "terminology: consolidated renderings contradict the SIGNED bank; they stay UNSIGNED — which no longer holds them back from the wire (D39.104) — and are the first rows to review at the stop",
			"book", r.Book.BookID, "conflicts", len(conflicts), "terms", strings.Join(named, "; "))
	}
	// §G2: the same check with the run's OWN consolidations on the right-hand side. On a live bank this is
	// the MAJORITY of the contradictions (18 of 149, research/24 §A4) and until now nobody looked: the canon
	// check can only see rows the owner already signed, and a book being consolidated for the first time has
	// almost none. $0, evidence-side, never a gate.
	if self := terminology.ConsolidationConflicts(cands, out); len(self) > 0 {
		res.SelfConflicts = len(self)
		res.Contradictions = map[string][]string{}
		named := make([]string, 0, len(self))
		for _, cf := range self {
			named = append(named, fmt.Sprintf("%s→%q drops %s→%q", cf.Src, cf.Dst, cf.PartSrc, cf.PartDst))
			res.Contradictions[cf.Src] = append(res.Contradictions[cf.Src], fmt.Sprintf("%s→%q", cf.PartSrc, cf.PartDst))
		}
		r.Log.WarnContext(ctx, "terminology: consolidations of THIS run contradict each other — a compound's rendering drops the rendering the same reply gave its own part; they stay unverified and are review rows at the stop",
			"book", r.Book.BookID, "conflicts", len(self), "terms", strings.Join(named, "; "))
	}
	// The BANK side: a rendering this run consolidated for a surface the bank already renders differently,
	// which neither check above reaches. Neither side is asked for a signature — one does not change what
	// the model is shown (D39.104 п.2). $0, evidence-side, never a gate.
	if cols := membank.ConsolidationKeyConflicts(consolidatedRows(cands, out), bank); len(cols) > 0 {
		res.BankConflicts = len(cols)
		res.BankHoldRows = cols
		r.Log.WarnContext(ctx, "terminology: a consolidated rendering DISAGREES with a row the bank already holds for the same firing surface — the book calls this term something else, and the sheet at the stop is where that is decided (whether both renderings also reach a wire depends on the emission and the glossary UNIQUE key; this does not claim they will)",
			"book", r.Book.BookID, "conflicts", len(cols), "terms", strings.Join(membank.ConflictMessages(cols), "; "))
	}
	// The $0 label screen (§2 warm-run hygiene): a name/place row whose rendering was clearly TRANSLATED is a
	// label/rendering disagreement worth a human's eye. It is a review FLAG, never a gate, and explicitly not
	// a safety net for the transliteration harm — the classifier phase is what prevents that (see
	// terminology.TypeLabelMismatches). Named, not a bare count, so the operator knows which rows to open.
	if flags := terminology.TypeLabelMismatches(labelRows(cands, out)); len(flags) > 0 {
		named := make([]string, 0, len(flags))
		for _, f := range flags {
			named = append(named, fmt.Sprintf("%s [%s]→%q", f.Src, f.Type, f.Dst))
		}
		r.Log.WarnContext(ctx, "terminology: name/place rows carry a translated rendering — a label/rendering mismatch to review (hygiene flag, not a gate)",
			"book", r.Book.BookID, "rows", len(flags), "terms", strings.Join(named, "; "))
	}
	if res.BatchesDropped > 0 || res.ClassifyBatchesDropped > 0 {
		// The pass ended SHORT, and the counters below cannot say so on their own: a term left
		// unconsolidated because the budget ran out is not a term the role declined to render. Both budgets
		// are named because they are different budgets and an operator raises one of them, not «the» one.
		r.Log.WarnContext(ctx, "terminology: this bank is PARTIALLY consolidated — a budget cut a pass, so some terms were never offered to the role at all; `unanswered` below counts them together with terms the role saw and did not answer",
			"book", r.Book.BookID, "render_batches_dropped", res.BatchesDropped,
			"classify_batches_dropped", res.ClassifyBatchesDropped)
	}
	r.Log.InfoContext(ctx, "terminology finished", "book", r.Book.BookID,
		"bank_conflicts", res.BankConflicts,
		"batches_dropped", res.BatchesDropped, "classify_batches_dropped", res.ClassifyBatchesDropped,
		"consolidated", res.Consolidated, "declined", res.Declined, "unanswered", res.Unanswered,
		"reclassified", res.Reclassified, "bad_lines", res.BadLines, "off_language", res.OffLanguage,
		"canon_conflicts", res.CanonConflicts, "self_conflicts", res.SelfConflicts,
		"families", res.Families, "families_refused", res.FamiliesRefused, "families_held", res.FamiliesHeld,
		"cost_usd", fmt.Sprintf("%.6f", res.CostUSD),
		"classify_cost_usd", fmt.Sprintf("%.6f", res.ClassifyCostUSD))
	return out, classified, gendered, res, nil
}

// terminologyMessages renders ONE batch into the wire messages: the pair's authored role prompt (system),
// the canon anchor as a code-assembled injection message, and the candidate block as the user turn. The
// block goes through {{text}} — the closed placeholder set stays closed, exactly as the memory injection
// did rather than growing a new placeholder.
//
// The anchor carries the book's own SIGNED rows (CANON — what the owner already decided, which a
// consolidation may not contradict) and nothing else: the pair/genre block that shipped beside it was
// removed by D39.47, because a market-wide default is exactly the thing this project has no authority to
// assert. It is DATA; every word explaining it lives in the pair's prompt, so a new pair needs no Go edit.
func (r *Runner) terminologyMessages(batch []terminology.Candidate, canon []terminology.Neighbour) ([]llm.Message, error) {
	return MessagesWithInjection(r.terminologyTemplate,
		RenderVars{Book: r.Book, Text: terminology.RenderBatch(batch)},
		terminology.RenderCanonAnchor(terminology.CanonFor(batch, canon, terminologyCanonCap)))
}

// terminologyCanonCap bounds the signed rows one batch carries. The anchor is a REMINDER of the law that
// touches these terms, not a copy of the bank: an unbounded block would grow with the book and eventually
// cost more than the candidates it accompanies.
const terminologyCanonCap = 40

// runClassifier is the §2 type-classifier phase: a focused pass on the same candidate batches that returns
// key → corrected type AND key → gender. Off (or no template) → nil maps and a zero run, so the
// terminologist keeps the draft heuristic type and pays nothing. The classifier needs no series co-batching
// (it decides a class per term) and no canon anchor (a class is a property of the source, not of the
// signed bank).
//
// ⚠ WHY THE GENDER RIDES HERE (backlog row 210, the pack's choice among the three the brief named). The
// engine has PROMISED a character's gender on the wire since D39.21 and never produced one: the
// terminologist's prompt returns three fields and none of them is gender, and the miner reads `gender`
// out of the hand-written seed only — so on any book nobody seeded by hand the axis is dead by
// construction, and «плывущий род» is not a rendering defect but a missing input. Of the ways to produce
// it, this pass is the only one that costs NOTHING: it already runs over exactly these candidates with
// exactly these contexts, so the answer grows by one short word per line instead of by a whole pass of
// paid calls. It is also the right ROLE. Gender is a property of the entity, read off the source contexts
// — the same act as deciding `name` against `term` — while the terminologist's act is choosing a rendering;
// and the terminologist's third column is the CONFIDENCE, around which a delicate mis-split rule is written
// (replyColumns) after a live defect glued «95%» onto a rendering. This parser is positional over a closed
// vocabulary and cannot glue.
//
// ⚠ AND IT IS GATED ON classify_types. A book whose config leaves the classifier off produces no gender
// at all, by exactly the same mechanism that leaves its types on the draft heuristic.
func (r *Runner) runClassifier(ctx context.Context, snapID string, cands []terminology.Candidate) (types, genders map[string]string, run bankRoleRun, err error) {
	g := r.Pipeline.Gates.Terminology
	if !g.ClassifyTypes || r.classifierTemplate == nil || len(cands) == 0 {
		return nil, nil, bankRoleRun{}, nil
	}
	batchRunes, _, _ := r.terminologyOpts()
	batches := terminology.Batch(cands, batchRunes, nil)
	plan := bankRolePlan{role: roleClassifier, budgetUSD: g.ClassifyBudgetUSD, messages: r.classifierMessages}
	run, err = r.runBankRoleBatches(ctx, snapID, plan, batches, "classify")
	if err != nil {
		return nil, nil, run, err
	}
	types, genders = map[string]string{}, map[string]string{}
	bad, badGender, noGender := 0, 0, 0
	for i, b := range batches {
		if i >= run.attempted {
			break // the budget stopped the pass here, and it already said so
		}
		if run.texts[i] == "" {
			// The same silence the render phase carried: a paid classify batch that returned NOTHING left every
			// one of its terms on the draft heuristic type — the mistyping this phase exists to remove — and the
			// only trace was that `bad` stayed 0, which reads as a clean pass.
			r.Log.WarnContext(ctx, "terminology classify: a paid batch came back with an EMPTY completion; its terms keep the draft heuristic type and produce no gender",
				"book", r.Book.BookID, "batch", i, "asked", len(b), "answered", 0)
			continue
		}
		got, gots, st := terminology.ParseTypes(run.texts[i], candKeys(b), text.NormalizeSourceKey)
		bad, badGender, noGender = bad+st.Bad, badGender+st.BadGender, noGender+st.NoGender
		r.Log.InfoContext(ctx, "terminology classify batch", "book", r.Book.BookID,
			"batch", i, "asked", len(b), "answered", len(got), "gendered", len(gots),
			"bad_lines", st.Bad, "bad_gender", st.BadGender, "no_gender_column", st.NoGender)
		for k, v := range got {
			types[k] = v
		}
		for k, v := range gots {
			genders[k] = v
		}
	}
	if bad > 0 {
		r.Log.WarnContext(ctx, "terminology classify: some reply lines were off-vocabulary or malformed; those terms keep their draft type",
			"book", r.Book.BookID, "bad_lines", bad)
	}
	// The gender column has its own warning because it has its own failure: a model can classify every term
	// correctly and answer the gender question in words nobody asked for, or stop answering it at all — and
	// row 210's whole subject is an axis that was dead while every counter beside it read clean.
	if badGender > 0 || noGender > 0 {
		r.Log.WarnContext(ctx, "terminology classify: the GENDER column did not land on every line — those terms carry no gender datum, and a bank with no gender renders no gender directive (backlog row 210)",
			"book", r.Book.BookID, "off_vocabulary", badGender, "column_absent", noGender,
			"vocabulary", strings.Join(terminology.TypeNames(terminology.Genders), "|"))
	}
	return types, genders, run, nil
}

// classifierMessages renders ONE batch into the classifier's wire messages: the pair's authored classifier
// prompt (system) and the candidate block as the user turn, with no canon anchor. Engine-neutral, like the
// terminologist block — the class DEFINITIONS live in the pair's prompt, so a new pair needs no Go edit.
func (r *Runner) classifierMessages(batch []terminology.Candidate) ([]llm.Message, error) {
	return MessagesWithInjection(r.classifierTemplate,
		RenderVars{Book: r.Book, Text: terminology.RenderBatch(batch)}, "")
}

// applyTypes stamps the classifier's corrected types onto the candidates in place and returns how many it
// actually changed. A term the classifier did not answer keeps its draft type.
func applyTypes(cands []terminology.Candidate, classified map[string]string) int {
	n := 0
	for i := range cands {
		if t, ok := classified[cands[i].Key]; ok && t != "" && t != cands[i].Type {
			cands[i].Type = t
			n++
		}
	}
	return n
}

// candKeys is the batch's candidate keys, in order — what the reply parsers screen a reply against.
func candKeys(cands []terminology.Candidate) []string {
	keys := make([]string, len(cands))
	for i, c := range cands {
		keys[i] = c.Key
	}
	return keys
}

// labelRows pairs each candidate that received a consolidated rendering with its (corrected) type, for the
// $0 label screen.
func labelRows(cands []terminology.Candidate, consolidated map[string]string) []terminology.LabelRow {
	var rows []terminology.LabelRow
	for _, c := range cands {
		if dst := consolidated[c.Key]; dst != "" {
			rows = append(rows, terminology.LabelRow{Src: c.Src, Type: c.Type, Dst: dst})
		}
	}
	return rows
}

// bankCallBudget resolves the model and max_tokens of a bank-role call (terminologist OR classifier) — ONE
// definition, so the checkpoint probe and the call itself can never address different request hashes (the
// repair precedent).
func (r *Runner) bankCallBudget(model string, msgs []llm.Message) (string, int) {
	est := 0
	for _, m := range msgs {
		est += EstimateTokens(m.Content)
	}
	// The reply is one short line per term, so it is a FRACTION of the input, not a multiple of it: sizing
	// it by MaxOutputRatio (the translation ratio) would reserve — and on a ceiling, deny — many times what
	// the call can possibly emit.
	maxTokens := est/2 + terminologyReplyFloor
	if maxTokens < r.Pipeline.Defaults.MinMaxTokens {
		maxTokens = r.Pipeline.Defaults.MinMaxTokens
	}
	return model, r.applyModelFloor(maxTokens, model)
}

// terminologyReplyFloor is the headroom one batch's reply needs beyond the proportional estimate.
const terminologyReplyFloor = 256

// bankCallEstimateUSD projects ONE call's cost with the same price/estimate arithmetic the reservation uses,
// so the pre-call number and the reserved number are the same number.
func (r *Runner) bankCallEstimateUSD(st config.Stage, msgs []llm.Message) float64 {
	_, maxTokens := r.bankCallBudget(st.Model, msgs)
	// The buffer is read off the SAME stage the call will use rather than passed as a literal, so this line
	// cannot drift from the attempt's. ⚠ It is structurally 0 on every path today and that is NOT because of
	// the effort: AdditiveReasoningTokens ignores its effort argument entirely (D39.26 добор B) and returns 0
	// whenever the declared buffer is 0, which InternalCall pins. The gate's additive-provider refusal
	// (config.LoadPipeline) is what makes that safe rather than blind.
	return r.callEstimateUSD(st, st.Model, msgs, maxTokens)
}

// bankCheckpointExists reports whether THIS batch was already paid for in an earlier run, on the role's own
// request-hash axis — the SAME identity runBankAttempt will address (attemptRequest), never a hand-rebuilt
// copy of it: this probe gates the role sub-budget, so an identity that drifts from the attempt's turns the
// gate off (see attemptRequest).
func (r *Runner) bankCheckpointExists(st config.Stage, snapID string, ch chunk.Chunk, msgs []llm.Message) (bool, error) {
	_, maxTokens := r.bankCallBudget(st.Model, msgs)
	cp, err := r.Store.GetCheckpoint(RequestHash(r.attemptRequest(st, st.Model, snapID, ch, 0, maxTokens, msgs)))
	return cp != nil, err
}

// runBankAttempt performs ONE bank-role call on the shared money path: reserve → call → settle+checkpoint,
// with a checkpoint hit replayed for free. It reuses runAttempt rather than re-implementing the money
// sequence — a second copy of that sequence is the drift this codebase already paid to remove once.
// isFinal=false: the reply is a term table, not shipping prose, so the output sanitizer must not judge it;
// the intrinsic classifier still runs and is a real guard (a refusal reply would otherwise be parsed as
// terminology).
func (r *Runner) runBankAttempt(ctx context.Context, st config.Stage, snapID string, ch chunk.Chunk,
	job *store.Job, msgs []llm.Message) (stageAttempt, error) {

	_, maxTokens := r.bankCallBudget(st.Model, msgs)
	// The log axis, exactly as runStage sets it for a chunk stage: without it every bank-role line in a paid
	// run reads "calling model deepseek-v4-flash" with no book, no role and no batch — unattributable in a
	// multi-book log and unmatchable to the batch that produced a bad reply. Chunk carries the batch ordinal.
	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)
	// NOT mandatory: the terminology pass degrades on a ceiling, and it runs between the waves with
	// nothing in flight — there would be nothing to wait for even if it were allowed to.
	return r.runAttempt(ctx, st, st.Model, snapID, ch, job, 0, maxTokens, msgs, false, false, false)
}

// bankRoleSettings resolves what a bank ROLE is called with. The classifier phase may run its OWN model
// (classification is cheaper than a render), but BOTH phases share ONE effort knob — deliberately, and on
// evidence: paired samples of the ratified 6/6 acceptance set put the classifier at 4/5 on `low` against
// 5/5 at `high`, which n=5 cannot separate, while `high` costs 2.3× per call. The render phase, by
// contrast, is FORCED to a low effort (at the vendor default it burns the whole budget thinking and
// returns an empty body). One level satisfies both, so a second key would be a split nothing measured.
func (r *Runner) bankRoleSettings(role string) (model, reasoning string) {
	g := r.Pipeline.Gates.Terminology
	switch role {
	case roleClassifier:
		return g.ClassifierModel(), g.Reasoning
	case roleTerminologist:
		return g.Model, g.Reasoning
	}
	// A future bank role (the annotator, the Ф2 judge this seam anticipates) must not silently inherit the
	// terminologist's model and effort: that is the same "a default is indistinguishable from a decision"
	// shape this pack exists to remove. Panic is right here — the role set is a compile-time constant of the
	// engine, so reaching this is a programming error, not a config one, and it can only happen before any
	// money moves.
	panic("pipeline: bankRoleSettings has no settings for bank role " + role + " — add its resolution to the gate")
}

// isBankRole reports whether a role's completion is one of the engine's TERM TABLES rather than prose in
// the target language. It is the single place that question is answered, so the two verdict rules that
// depend on it — the source-echo share and the target-language screen — cannot come to disagree about
// which roles are exempt, which is exactly how the classifier ended up flagged on every healthy batch
// while the terminologist beside it was exempt (13-tech-debt-anchors §Б-105).
//
// It is keyed on the closed role set rather than sniffed from the reply, because the FORM is something the
// engine specified when it built the request; deriving it back out of the answer would let a malformed
// reply talk its way out of the screen. A future bank role joins by one line here, next to the line it
// already needs in bankRoleSettings.
func isBankRole(role string) bool {
	return role == roleTerminologist || role == roleClassifier
}

// bankStage is the ONE place a bank-role call's stage is derived — for BOTH roles and for the live rig, so
// "the classifier probe measured the production shape" is structural rather than two literals kept in step
// by hand. The settings come from the gate that owns the call class (config.InternalCall).
func (r *Runner) bankStage(role string) config.Stage {
	model, reasoning := r.bankRoleSettings(role)
	return config.InternalCall{Name: terminologyStageName, Role: role, Model: model, Reasoning: reasoning}.Stage()
}

// bankRolePlan is one bank-level role's plan over a batch list: its cost axis (role/model), its own book-wide
// ceiling, and how it builds one batch's wire messages. The money sequence — estimate, budget-gate,
// checkpoint-or-call — is identical for the classifier and the terminologist; only the messages and the reply
// PARSING differ, and parsing is the caller's job on the returned texts.
// ⚠ NO `model` field: the model is derived from the role by bankRoleSettings, inside bankStage. It used to
// be carried here too and read only by a log line, which made the log a SECOND source of truth about what
// was billed — the exact shape this pack removed everywhere else.
type bankRolePlan struct {
	role      string
	budgetUSD float64
	messages  func(batch []terminology.Candidate) ([]llm.Message, error)
}

// bankRoleRun is what one role's pass produced: each batch's reply text (in batch order, "" for a batch the
// budget cut or that failed soft), plus the cost accounting for the report.
type bankRoleRun struct {
	texts []string
	// attempted is how many batches the pass actually reached before a budget ceiling or a soft denial stopped
	// it. Without it "" is ambiguous — an EMPTY completion the run paid for and a batch never called look the
	// same — and the caller cannot warn about the first without crying wolf about the second.
	attempted   int
	estimateUSD float64
	costUSD     float64
	cumUSD      float64
	fresh       bool
	// planned / dropped record the CUT: how many batches the pass was built from, and how many the budget
	// left unbought. `dropped > 0` is «this bank is partially consolidated», which is a fact about the
	// book's terminology that nothing else in the run's output carries — and reporting a partial pass as a
	// pass is the shape this whole edit exists to stop.
	planned int
	dropped int
}

// runBankRoleBatches runs plan over batches on the shared money path. The estimate is logged BEFORE any call;
// a budget ceiling or a soft denial stops the pass and leaves the remaining terms untouched — the run never
// aborts on this optional step. logKind names the phase (render|classify) in the logs.
func (r *Runner) runBankRoleBatches(ctx context.Context, snapID string, plan bankRolePlan, batches [][]terminology.Candidate, logKind string) (bankRoleRun, error) {
	run := bankRoleRun{texts: make([]string, len(batches))}
	// ONE stage for the whole pass — the estimate, the checkpoint probe and the attempt all read it, so the
	// three can never be sized against different knobs (the estimate is the reservation's own upper bound).
	st := r.bankStage(plan.role)
	msgsPer := make([][]llm.Message, len(batches))
	for i, b := range batches {
		m, err := plan.messages(b)
		if err != nil {
			return run, err
		}
		msgsPer[i] = m
		run.estimateUSD += r.bankCallEstimateUSD(st, m)
	}
	r.Log.InfoContext(ctx, "terminology "+logKind+": estimate before any call",
		"book", r.Book.BookID, "role", plan.role, "batches", len(batches), "model", st.Model, "reasoning", st.Reasoning,
		"estimate_usd", fmt.Sprintf("%.6f", run.estimateUSD), "budget_usd", plan.budgetUSD,
		"version", terminologyVersion, "bank_data", lang.BankDataVersion())

	spent, err := r.Store.RoleSpentUSD(r.Book.BookID, plan.role)
	if err != nil {
		return run, fmt.Errorf("pipeline: %s budget read: %w", plan.role, err)
	}
	job, err := r.Store.EnsureJob(r.Book.BookID, 0, terminologyStageName, snapID)
	if err != nil {
		return run, fmt.Errorf("pipeline: %s job: %w", plan.role, err)
	}
	// ⛔ THE PLAN IS CUT TO WHAT FITS **BEFORE THE FIRST CALL**, not discovered halfway through it.
	//
	// This pass used to print its estimate beside its budget, see that it did not fit, and start anyway —
	// the very next statement after the log line was an error check on READING the budget, and no
	// comparison stood between them. It then stopped mid-way on the per-batch gate below, leaving a
	// half-consolidated bank and paid work whose report said nothing about being cut short. On the cold run
	// that happened twice, and three batches of four were never bought.
	//
	// ⛔ AND IT IS A TRUNCATION, NOT A REFUSAL, deliberately. Refusing with a Refusal class would put this
	// in the 10–19 band, and that band promises the platform «nothing reached a provider, nothing was
	// spent». This pass runs INSIDE the bank-mining stop — after a paid draft wave — so the promise would
	// be false and the engine would be lying about money to the one consumer that acts on the band
	// destructively. The estimate is also a deliberate UPPER bound, so «estimate > budget ⇒ refuse» would
	// deny work that fits: the cold run has a counterexample of exactly that shape, a pass whose estimate
	// ($0.024239) exceeded its budget ($0.02) and which completed for a fraction of it.
	//
	// So: decide up front, say what was cut, and run only that. Already-paid batches cost nothing and are
	// admitted regardless — holding them back would save no money and lose their result.
	fits, plannedUSD := 0, 0.0
	probe := spent
	paidBatch := make([]bool, len(batches))
	for i := range batches {
		ch := chunk.Chunk{Chapter: 0, ChunkIdx: i}
		paid, perr := r.bankCheckpointExists(st, snapID, ch, msgsPer[i])
		if perr != nil {
			return run, perr
		}
		paidBatch[i] = paid
		if paid {
			fits++
			continue
		}
		// The bound is what the call would COST (the reservation's own upper bound), so the cut lands on
		// the conservative side and the config number is the bound it looks like.
		want := r.bankCallEstimateUSD(st, msgsPer[i])
		if probe+want > plan.budgetUSD {
			break
		}
		probe += want
		plannedUSD += want
		fits++
	}
	if fits < len(batches) {
		// Said BEFORE the first call and naming every term of the decision, because the operator's action
		// (raise the budget and re-run, or accept a partial bank) depends on all of them.
		r.Log.WarnContext(ctx, "terminology "+logKind+": the plan does NOT fit the budget and is CUT TO WHAT DOES — the remaining terms are left unchanged, and this is decided BEFORE the first call rather than discovered part-way through",
			"book", r.Book.BookID, "role", plan.role, "batches_planned", len(batches), "batches_running", fits,
			"batches_dropped", len(batches)-fits, "spent_usd", fmt.Sprintf("%.6f", spent),
			"this_pass_usd", fmt.Sprintf("%.6f", plannedUSD), "budget_usd", plan.budgetUSD,
			"estimate_usd", fmt.Sprintf("%.6f", run.estimateUSD))
	}
	run.planned, run.dropped = len(batches), len(batches)-fits
	for i := 0; i < fits; i++ {
		// The synthetic chunk addresses the batch: chapter 0 is the BOOK level (no real chapter is 0), and the
		// batch ordinal is the chunk index, so two batches can never collide on one checkpoint.
		ch := chunk.Chunk{Chapter: 0, ChunkIdx: i}
		att, aerr := r.runBankAttempt(ctx, st, snapID, ch, job, msgsPer[i])
		if aerr != nil {
			// A ceiling denial must not abort the book: this step is optional and the draft wave is already
			// paid for. Degrade to "no change" exactly as the repair sub-step degrades.
			if errors.Is(aerr, errReserveCeiling) {
				r.Log.WarnContext(ctx, "terminology "+logKind+": call denied by a USD ceiling; remaining terms left unchanged", "book", r.Book.BookID, "role", plan.role)
				break
			}
			return run, aerr
		}
		run.costUSD += att.runCost
		run.cumUSD += att.cumCost
		run.fresh = run.fresh || att.freshCall
		spent += att.runCost
		run.texts[i] = att.text
		run.attempted = i + 1
	}
	return run, nil
}

// consolidatedRows turns this run's consolidation into the rows those renderings would become, which is
// what ConsolidationKeyConflicts reads: the candidate's surface, its identity-cluster aliases and the
// window a mined row carries (first chapter of appearance, no end). An unanswered candidate carries no
// rendering, becomes no row, and contradicts nothing.
//
// Both extra fields are load-bearing for the check: the window decides whether an existing row is a
// replacement rather than a disagreement, and the aliases are the other surfaces a landed row fires on.
// Built for every answered candidate, not only those the emission keeps — a dropped candidate still
// reaches the review sheet (not the signature map, a different document), so its disagreement is read.
func consolidatedRows(cands []terminology.Candidate, consolidated map[string]string) []store.GlossaryEntry {
	if len(consolidated) == 0 {
		return nil
	}
	out := make([]store.GlossaryEntry, 0, len(consolidated))
	for _, c := range cands {
		dst := consolidated[c.Key]
		if dst == "" {
			continue
		}
		// c.Key, not c.Src, and it is the landing source on BOTH emission paths — the check compares the
		// store's raw UNIQUE key, so it must be given the string the row will actually hold. A banknote
		// candidate carries the surface as the draft wrote it and lands as the key
		// (mining.reverseSectionTerms). A mined candidate lands with the MINER's surface, which is a
		// substring of already-normalized chunk text and therefore equals its own key — that equality is
		// text.NormalizeSourceKey's idempotence, pinned by TestNormalizeSourceKeyIsIdempotent, because
		// without it this line would compare a landing row against a source it does not have.
		e := store.GlossaryEntry{Src: c.Key, Dst: dst, SinceCh: c.SinceCh}
		for _, a := range c.Aliases {
			e.Aliases = append(e.Aliases, store.GlossaryAlias{Alias: a})
		}
		out = append(out, e)
	}
	return out
}

// attachConsolidatedDst stamps the terminologist's renderings onto the mined terms, which is what selects
// the emission MODE in miner.DeltaYAML (§C2-7). Terms the role did not answer are left untouched.
func attachConsolidatedDst(mined []miner.Term, consolidated map[string]string) []miner.Term {
	if len(consolidated) == 0 {
		return mined
	}
	out := make([]miner.Term, len(mined))
	copy(out, mined)
	for i := range out {
		if dst := consolidated[text.NormalizeSourceKey(out[i].Src)]; dst != "" {
			out[i].Dst = dst
		}
	}
	return out
}

// attachClassifiedGender stamps the §2 classifier's gender datum onto the mined terms — the input backlog
// row 210 exists for, and the ONLY automatic producer of the field. It never changes which terms are in the
// delta and never overwrites with an empty value: a term the classifier answered `none` for, or did not
// answer at all, carries no gender, exactly as before.
//
// ⚠ THE SEED STILL WINS, and not by an ordering trick here. These rows go into the auto-bank, which the
// loader keeps OUT of any key a signed row already holds (loadAutoBank, mining.go): a term the owner
// entered by hand — with or without a gender — is the row the bank uses, and the engine's proposal for the
// same key is dropped, named, in the log. So «the model decided my character is male» is not a state the
// owner can be surprised by on a term they signed.
func attachClassifiedGender(mined []miner.Term, gendered map[string]string) []miner.Term {
	if len(gendered) == 0 {
		return mined
	}
	out := make([]miner.Term, len(mined))
	copy(out, mined)
	for i := range out {
		if g := gendered[text.NormalizeSourceKey(out[i].Src)]; g != "" {
			out[i].Gender = g
		}
	}
	return out
}

// attachClassifiedType stamps the §2 classifier's corrected types onto the mined terms, so the BANKED type
// is the re-derived one, not the draft heuristic. It never changes which terms are in the delta — emission
// eligibility was already decided upstream by the miner — only the type recorded on the rows already there.
func attachClassifiedType(mined []miner.Term, classified map[string]string) []miner.Term {
	if len(classified) == 0 {
		return mined
	}
	out := make([]miner.Term, len(mined))
	copy(out, mined)
	for i := range out {
		if t := classified[text.NormalizeSourceKey(out[i].Src)]; t != "" {
			out[i].Type = t
		}
	}
	return out
}