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"

// logKeyReconsolidated is the structured key the one-time re-consolidation warning carries, and it exists
// so that the warning and the test that guards it read ONE carrier.
//
// ⚠ WHY A KEY AND NOT THE SENTENCE. That guard has now been written twice against the message's PROSE, and
// the second time the prose had already been reworded: the test grepped a substring that occurs zero times
// in this module, so it could not fire at all and the defect it was written for went green. A key is short,
// is not rewritten when the sentence is improved, and — being a constant both sides share — cannot be
// renamed on one side only: the test would stop compiling rather than stop guarding.
const logKeyReconsolidated = "reconsolidated"

// 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
	// DeclinedByPhrase is the part of Declined the role wrote in WORDS instead of with the engine sentinel,
	// and NoLetters the part of BadLines refused for carrying no letter at all («90»). Both are subsets of
	// a counter beside them, and both are reported because the whole class has a MEASURED population of
	// zero on everything this project has bought (0 of 69 and 0 of 66, research/35 §2.0 row Б-5): a
	// mechanism whose population is zero is one nobody can tell from a mechanism that is not running, and
	// these two counters are the difference.
	DeclinedByPhrase int
	NoLetters        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. SelfConflictRows
	// carries the findings themselves, UNGROUPED, so the stop table can mark the rows instead of printing a
	// bare total — and so the key they are matched by is decided in one place (bankStopRows' findingsFor)
	// instead of once here and once on the far side of a map.
	SelfConflicts    int
	SelfConflictRows []terminology.ConsolidationConflict
	// 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 SelfConflictRows because the two are different decisions for the owner.
	BankConflicts int
	BankHoldRows  []membank.BankKeyConflict
	// BankSettled is how many candidates never reached the paid role because the bank had already settled
	// them — the surface is seeded and every draft proposed the rendering the bank holds. It is the saving,
	// and it is carried rather than only logged so the report can state it with its denominator
	// (Candidates) instead of with a word.
	BankSettled int
	// BankSettledKeys are the surfaces behind that count, carried so the sheet can SAY the row was never
	// asked about instead of printing an empty rendering — which reads as «undecided» beside three other
	// facts that share the glyph.
	BankSettledKeys []string
	// 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
	// ClassifyAsked / ClassifyAnswered are the classifier's own answer share, summed over the batches it
	// actually called.
	//
	// ⛔ THE SHARE EXISTED AND REACHED NOBODY, which is the whole defect here. It was computed per batch and
	// written into a mid-pass INFO line, so the only carrier of «the classifier answered 4 of 22» was a log
	// somebody had to be reading at the time; the run's summary printed `classify_batches_dropped=0`, which
	// counts batches the BUDGET left unbought and reads to a person as «nobody was left unanswered». On the
	// cold run of file B the summary said dropped=0, reclassified counted normally, and 42 of 66 terms had
	// no machine type. These two fields are that per-batch number raised to the total — not a third counter
	// beside the render pass's `Unanswered`, which is a fact about a DIFFERENT pass.
	//
	// ⚠ ASKED IS THE DENOMINATOR AND IT IS NOT `Candidates`: a batch the budget never bought asked nobody
	// anything, and folding it in here would report the model as silent about terms it was never shown.
	ClassifyAsked    int
	ClassifyAnswered int
	// BankRegens / BankUnusable / BankStepsRefused are the LADDER's facts, summed over both phases: extra
	// rungs bought, batches that stayed unusable to the last rung, and rungs a phase sub-budget refused.
	// They are what tells «the model could not answer this» apart from «the phase ran out of money», which
	// before the ladder existed were the same silence.
	BankRegens       int
	BankUnusable     int
	BankStepsRefused int
	// 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.
//
// A NIL result means the pass did not run at all, which is a different fact from a pass that ran and
// consolidated nothing — the counters of the second are a verdict about the bank, the counters of the
// first are a verdict about nothing. The projections downstream publish that difference (bankexport.go).
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, nil, nil
	}
	res = &terminologyResult{Candidates: len(cands)}
	for _, c := range cands {
		if c.Origin == terminology.OriginBanknote {
			res.Reverse++
		}
	}
	// ⚠ `paid` is a SEPARATE slice and `cands` is left whole on purpose. The caller renders the signature
	// sheet and the delta from the candidates it passed in, and the two passes below MUTATE candidates in
	// place — applyTypes stamps the classifier's type, ScoreVariants re-ranks the renderings. Filtering the
	// caller's own slice would have copied the survivors into a new array, so those mutations would have
	// stopped reaching the sheet: the operator would read heuristic types and stale rankings, and nothing
	// would say so. Every fixture in the package runs with the classifier OFF, where neither mutation
	// happens at all, so the tests written for this filter could not have seen it.
	// ⛔ INDICES, not a filtered slice, and that is the fix for a defect this file has now grown three
	// times. The two passes below MUTATE candidates in place (applyTypes stamps the classifier's type,
	// ScoreVariants re-ranks), the caller renders the sheet from `cands`, and the ROLE's request is built
	// from the paid subset — so a subset held as its own array carries the stamps to neither, or to only
	// one of the two. Keeping positions lets the subset be REBUILT from `cands` after the mutations, so the
	// sheet and the wire read the same candidates.
	paidIdx, settled := r.dropBankSettled(ctx, cands)
	res.BankSettled = settled
	if settled > 0 {
		kept := map[int]bool{}
		for _, i := range paidIdx {
			kept[i] = true
		}
		for i, c := range cands {
			if !kept[i] {
				res.BankSettledKeys = append(res.BankSettledKeys, c.Key)
			}
		}
	}
	if len(paidIdx) == 0 {
		return nil, nil, nil, res, nil
	}
	paid := pickCandidates(cands, paidIdx)
	// ⛔ ASKED HERE, BEFORE THE FIRST PAID CALL, and the position is the whole point. The question is
	// whether this book had bank-role checkpoints BEFORE this run — and after the passes below it is
	// unanswerable, because they settle checkpoints of their own: a FRESH book that has just paid once
	// answers "yes, it had paid" and the operator is told he paid twice. That is exactly what the previous
	// version of this warning did, one viton after the version before it told every ordinary resume the
	// same lie. A read failure leaves it false — a warning about money must not be invented from an error.
	paidBefore, perr := r.Store.HasCheckpointForStage(r.Book.BookID, terminologyStageName)
	if perr != nil {
		r.Log.WarnContext(ctx, "terminology: could not read whether this book had paid for the bank role before this run, so a one-time re-consolidation would go unannounced; nothing else is affected",
			"book", r.Book.BookID, "err", perr)
	}
	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, paid)
	// ⛔ THE MONEY IS RECORDED BEFORE THE VERDICT, and the order is the same one runStage keeps for the
	// same reason: what a pass BOUGHT is a fact about what happened, not about whether it succeeded. Read
	// after the error check, these two fields reported ZERO for a pass that had paid for batches and then
	// met a broken provider — a struct lying about money to whoever reads it next.
	//
	// ⚠ WHAT THIS IS AND IS NOT. Today nothing reads it on that path: an error here ends the run, the
	// driver returns no result, and the money is on the LEDGER either way (the batch settles through
	// SettleWithCheckpoint like any call, RoleSpentUSD sums it from the durable checkpoints, and
	// TestKillMinus9LosesAtMostOneCall pins that a settled checkpoint's money survives even a SIGKILL).
	// So this is insurance against the next reader, not the repair of a live loss — backlog row 388 claimed
	// more than that, and the claim was mine.
	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
	// ⛔ THE LADDER'S COUNTS BELONG ON THIS SIDE OF THE CHECK TOO, and they were on the other one. The rule
	// is stated ten lines above for the money — what a pass BOUGHT is a fact about what happened, not about
	// whether it succeeded — and a rung bought is exactly that kind of fact. Read after the error check,
	// these four reported ZERO for a phase that had climbed and then met a broken provider: the render half
	// records the identical fields BEFORE its own check, so the two halves disagreed about when a count
	// becomes true. Found by acceptance, on an asymmetry nothing pins.
	res.ClassifyAsked, res.ClassifyAnswered = crun.asked, crun.answered
	res.BankRegens += crun.regens
	res.BankUnusable += crun.unusable
	res.BankStepsRefused += crun.stepsRefused
	if cerr != nil {
		return nil, nil, nil, res, cerr
	}
	res.Gendered = len(gendered)
	if len(classified) > 0 {
		res.Reclassified = applyTypes(cands, classified)
		opts := r.scoreOpts()
		for i := range cands {
			terminology.ScoreVariants(&cands[i], opts)
		}
		// REBUILT so the batches below carry the corrected type. The classifier is bought FOR that field
		// and it travels to the role for the whole batch; without this the money is spent, the run prints
		// `reclassified=N`, and the wire still says what the heuristic guessed.
		paid = pickCandidates(cands, paidIdx)
	}

	// §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.
	// Over `paid`: a series or a family the role is not asked about cannot be co-batched, and detecting it
	// over the full set would only make the units disagree with the batches built from them.
	sEnabled, headFinal := lang.SeriesMorphology(r.Book.SourceLang)
	seriesID := terminology.DetectSeries(paid, terminology.SeriesParams{Enabled: sEnabled, HeadFinal: headFinal})
	fp := r.familyParams
	fams := terminology.DetectFamilies(paid, fp)
	unitID, ustats := terminology.MergeUnits(seriesID, fams, fp)
	batches := terminology.Batch(paid, 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.
	// The SPEND is recorded here for the same reason and in the same breath as the classifier's above: it
	// is what happened, whatever the verdict turns out to be.
	res.BatchesDropped = run.dropped
	res.BankRegens += run.regens
	res.BankUnusable += run.unusable
	res.BankStepsRefused += run.stepsRefused
	res.EstimateUSD, res.CostUSD, res.CumUSD, res.Fresh = run.estimateUSD, run.costUSD, run.cumUSD, run.fresh
	if rerr != nil {
		return nil, nil, nil, res, rerr
	}
	// ⚠ THE ONE-TIME COST OF THE ALREADY-BANKED FILTER. It needs BOTH halves and each is knowable at only
	// one moment: that the book had paid BEFORE this run (paidBefore, taken above the passes) and that this
	// run paid ANYWAY (run.fresh, knowable only now). Either half alone is a lie that has already been told
	// here twice — on "the book has paid", which every ordinary resume of a filtered book satisfies, and on
	// a probe taken after the passes, which a freshly-paid first run satisfies trivially.
	if res.BankSettled > 0 && run.fresh && paidBefore {
		// ⚠ THE CAUSE IS NAMED AS A PAIR, because the code cannot tell which of the two happened and a
		// message that picks one is wrong half the time. The old wording asserted the composition change
		// alone; a book whose only earlier bank row was BURNED — money settled, no result, unreplayable —
		// gets the same warning with a reason that is simply false for it, and a message lying about its
		// own cause is what D39.93 п.2 forbids. Both branches leave the operator with the same action, so
		// naming both costs nothing and claiming one costs the truth.
		r.Log.WarnContext(ctx, "terminology: this book had already paid for the bank role BEFORE this run, and this run bought the pass again — the earlier calls' checkpoints could not be reused, either because the batch composition changed (the already-banked filter does not reproduce it) or because they recorded money with no result and cannot be replayed. This is a ONE-TIME cost; every later run replays for $0",
			logKeyReconsolidated, true,
			"book", r.Book.BookID, "skipped", res.BankSettled, "of_candidates", res.Candidates, "cost_usd", fmt.Sprintf("%.6f", run.costUSD))
	}

	out := map[string]string{}
	res.Conf = map[string]int{}
	for i, b := range batches {
		if !run.ran[i] {
			continue // this batch was never called; runBankRoleBatches already said why
		}
		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,
			r.declinedByPhrase)
		res.BadLines += st.Bad
		res.OffLanguage += st.OffLanguage
		res.DeclinedByPhrase += st.DeclinedByPhrase
		res.NoLetters += st.NoLetters
		// 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 rendering with no letters at all — «90», «——». (A ONE-rune answer is refused a check earlier, on
		// length, so it never reaches this counter.) Named rather than counted, for the same
		// reason the off-language lines are: the population of this class on bought material has been zero
		// so far, so the first live one is evidence about a model or about a truncated reply, and a bare
		// count cannot tell those apart.
		if st.NoLetters > 0 {
			r.Log.WarnContext(ctx, "terminology: reply lines carried a rendering with NO LETTERS at all; they are REFUSED (the terms read as unanswered, not as declined) — a bank cannot hold «90» as a book's canon",
				"book", r.Book.BookID, "batch", i, "terms", len(b), "no_letters", st.NoLetters,
				"lines", strings.Join(st.NoLettersSamples, "; "))
		}
		// A decline the role wrote in WORDS instead of with the sentinel. The terms land where the sentinel
		// would have put them; what the warning is about is the PROMPT — the pair asked for ⟦TM-NO-DST⟧ and
		// got prose, and nothing else in the run would ever say so.
		if st.DeclinedByPhrase > 0 {
			r.Log.WarnContext(ctx, "terminology: the role declined term(s) in WORDS rather than with the engine sentinel; they are treated as declined (NOT banked), and the pair's prompt is what needs the look",
				"book", r.Book.BookID, "batch", i, "terms", len(b), "declined_by_phrase", st.DeclinedByPhrase)
		}
		// 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
		}
	}
	// Over `paid`, not over every candidate: a candidate nobody was asked about has not gone UNANSWERED —
	// that word means the role was given it and said nothing, which is a fact about the model.
	for _, c := range paid {
		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.SelfConflictRows = self
		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(terminology.ConsolidationConflictMessages(self), "; "))
	}
	// 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, "; "))
	}
	// A 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.
	//
	// ⛔ TWO WARNINGS AND NOT ONE, because they are two different facts on two different budgets and only
	// the first is about the BANK. One warning fired by either counter said «this bank is PARTIALLY
	// consolidated» on a run whose render pass was INTACT and whose classifier alone was cut — measured
	// 08.09: `batches_dropped=0, classify_batches_dropped=1, consolidated=29`. That is a false alarm about
	// the object an owner signs, and it also made the engine disagree with its own read-out, which reports
	// the bank's completeness from the render pass alone (bankexport.go, BankConsolidation).
	if res.BatchesDropped > 0 {
		r.Log.WarnContext(ctx, "terminology: this bank is PARTIALLY consolidated — a budget cut the RENDER 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)
	}
	if res.ClassifyBatchesDropped > 0 {
		r.Log.WarnContext(ctx, "terminology: the TYPE classifier ended short — its OWN budget cut a pass, so some candidates keep the draft heuristic type; the bank's renderings are unaffected and this alone does not make the bank partially consolidated",
			"book", r.Book.BookID, "classify_batches_dropped", res.ClassifyBatchesDropped)
	}
	// ⛔ THE SHARE THE CLASSIFIER ACTUALLY ANSWERED, and it is a DIFFERENT fact from the line above. That
	// one is about money the phase did not spend; this one is about terms the phase paid to ask and did not
	// get an answer for. Read as one, they hid the cold run's whole finding: `classify_batches_dropped=0`
	// with 42 of 66 terms left on the draft heuristic. A term with no answered type keeps that heuristic,
	// and the heuristic is wrong 12–22% of the time on a field that FORCES transliteration — so the silence
	// is not cosmetic, it is a place-name shipped as a person's name in every chapter it occurs in.
	if res.ClassifyAsked > res.ClassifyAnswered {
		r.Log.WarnContext(ctx, "terminology: the TYPE classifier did not answer every term it was PAID to be asked about; those terms keep the draft heuristic type, which is what forces a transliteration on a mistyped surface",
			"book", r.Book.BookID, "asked", res.ClassifyAsked, "answered", res.ClassifyAnswered,
			"unanswered", res.ClassifyAsked-res.ClassifyAnswered)
	}
	r.Log.InfoContext(ctx, "terminology finished", "book", r.Book.BookID,
		"bank_conflicts", res.BankConflicts,
		"batches_dropped", res.BatchesDropped, "classify_batches_dropped", res.ClassifyBatchesDropped,
		// The classifier's answer share, with its own denominator: «dropped» above is money never spent,
		// these are terms bought and unanswered, and a reader given only the first infers the second wrongly.
		"classify_asked", res.ClassifyAsked, "classify_answered", res.ClassifyAnswered,
		// What the ladder did across both phases — extra rungs bought, batches unusable to the last rung,
		// rungs a phase sub-budget refused.
		"bank_regenerations", res.BankRegens, "bank_unusable_batches", res.BankUnusable,
		"bank_steps_refused", res.BankStepsRefused,
		"consolidated", res.Consolidated, "declined", res.Declined, "unanswered", res.Unanswered,
		"reclassified", res.Reclassified, "bad_lines", res.BadLines, "off_language", res.OffLanguage,
		"declined_by_phrase", res.DeclinedByPhrase, "no_letters", res.NoLetters,
		"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
	// asked counts the terms this phase actually PUT to the model — batches it called, whether or not they
	// answered. A batch the budget never bought is not in it (see terminologyResult.ClassifyAsked).
	asked, answered := 0, 0
	for i, b := range batches {
		if !run.ran[i] {
			continue // this batch was never called, and the pass already said why
		}
		asked += len(b)
		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
		answered += len(got)
		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), "|"))
	}
	run.asked, run.answered = asked, answered
	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).
//
// ⛔ A BURNED CHECKPOINT IS NOT A PAID BATCH. One that records money and no result cannot be replayed,
// so runAttempt walks past it and buys the batch again — and a probe answering «already paid» would let
// that purchase escape the role sub-budget entirely, which is the one thing this gate exists to bound.
// The key can only hold such a row because cut calls are now settled there; before that, «a checkpoint
// exists» and «this batch is done» were the same statement.
func (r *Runner) bankCheckpointExists(st config.Stage, snapID string, ch chunk.Chunk, msgs []llm.Message) (bool, error) {
	_, maxTokens := r.bankCallBudget(st.Model, msgs)
	return r.paidAfterBurns(st, st.Model, snapID, ch, 0, maxTokens, msgs)
}

// runBankAttempt walks ONE bank-role batch up the attempt ladder — the SAME walk a chunk stage takes
// (attemptladder.go), on the same money path: reserve → call → settle+checkpoint, with every already-paid
// rung replayed for free.
//
// ⛔ THE LADDER IS THE POINT, and its absence here was the defect. This function used to make exactly one
// call and hand back its TEXT, dropping the classification runAttempt had already computed — so a batch the
// engine itself knew was truncated or empty was paid for, discarded, and counted as terms the role declined
// to render. Reaching the ladder is not a new mechanism; it is reading a verdict that was already there.
//
// 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).
// ⛔ THE PHASE'S POLICY ARRIVES AS ONE OBJECT, and that is deliberate rather than tidy. The re-ask COUNT
// and the money that bounds it are two halves of one decision, and holding them as two arguments made
// «no budget ⇒ no re-ask» an invariant kept by a comment: a later caller passing (nil, 1) would get an
// admission hook of nil, every rung admitted with no paid probe and no ceiling, and a bank batch doubling
// its way outside the phase budget in silence — the very hole §4.2 exists to close. As one object the
// pairing cannot come apart: a nil budget IS a single shot.
//
// ⚠ A nil budget is therefore a real caller, not a defensive branch: the reprobe rig measures what ONE
// classifier call comes back with, and its published numbers («4 of 5 runs reach 6/6 at low») are about
// single calls. Give it the pass's retry policy and it silently starts re-asking a truncated reply while
// still reporting per-call figures — a rig measuring one object and naming another.
func (r *Runner) runBankAttempt(ctx context.Context, st config.Stage, snapID string, ch chunk.Chunk,
	job *store.Job, msgs []llm.Message, budget *roleBudget) (ladderRun, 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)
	var afford func(ladderStep) bool
	maxRegens := 0
	if budget != nil {
		afford, maxRegens = budget.admitStep, budget.regens
	}
	return r.walkAttemptLadder(ctx, ladderCall{
		stage: st, model: st.Model, snapID: snapID, ch: ch, job: job, msgs: msgs,
		baseMaxTokens: maxTokens,
		maxRegens:     maxRegens,
		// echoRegens 0, and not because nobody configured it: the echo re-roll answers cjk_artifact, and
		// that flag cannot be raised for a bank role at all — isBankRole exempts a term table from the
		// source-echo share and the target-language screen, because a table of Chinese surfaces IS mostly
		// source script. A budget for a remedy to an unreachable flag would be a key that reads as a
		// decision and is none.
		//
		// ⚠ SO THE BANK HAS EXACTLY ONE REMEDY ON THIS LADDER, and the reader should not infer two from
		// runStage. The other branch — answering an empty reply with LESS thinking at the same budget — is
		// dead here for a second, independent reason: it needs Retries.LowerEffortOnEmpty (off in every
		// shipping pipeline, pinned) AND a rung below the configured effort, and the shipping bank roles
		// are configured at `low`, which is the bottom of Models.ReducedEffort's ladder. Either of those
		// changing makes the branch live, which is why the admission below is written to be correct under
		// a lowered effort rather than merely unreachable.
		echoRegens: 0,
		// 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.
		mandatory: false,
		isFinal:   false,
		// The phase's own sub-budget is what bounds a rung, and it is the SAME object the pre-flight
		// planned the first rungs against — see roleBudget.
		afford: afford,
	})
}

// roleBudget is ONE bank role's ADMISSION POLICY — the phase ceiling
// (gates.terminology.budget_usd / classify_budget_usd) together with how many rungs a batch may buy under
// it — carried across the whole pass so that every purchase the pass makes is judged against one running
// number, and so that the money rule and the count rule cannot be held apart by a caller.
//
// ⛔ IT IS ONE OBJECT BECAUSE THE PRE-FLIGHT AND THE LADDER MUST BE TALKING ABOUT THE SAME MONEY. The
// pre-flight decides, before the first call, which batches fit; the ladder then buys rungs that pre-flight
// never saw. While the two had separate arithmetic the rungs were bounded by nothing but the BOOK ceiling —
// a phase told to spend at most $1.00 could double its way past it and no counter would say so.
//
// ⚠ IT NEVER GIVES AN EARMARK BACK, and the conservatism is deliberate rather than overlooked: a call is
// admitted at the reservation's own UPPER bound and usually settles far below it, so a long pass carries a
// stale-high figure. That is harmless while the budget is orders of magnitude above the pass — the shipping
// number is $1.00 against a measured $0.0297 — and it stops being harmless the day a budget is set near the
// estimate, where it would refuse a rung the phase could in fact afford. The refusal is loud
// (admitLadderStep), so that day arrives as a log line rather than as a quietly shorter bank.
type roleBudget struct {
	limit     float64
	committed float64
	// regens is how many rungs this phase may buy for ONE batch — the money rule and the count rule of the
	// same policy, held together so that no caller can hold one without the other.
	regens int
}

// admit books an estimate against the phase ceiling, reporting whether it fits.
func (b *roleBudget) admit(estimateUSD float64) bool {
	if b.committed+estimateUSD > b.limit {
		return false
	}
	b.committed += estimateUSD
	return true
}

// admitStep is admit as the ladder asks it. The step carries its own price, from the one definition every
// gate prices a call with, so this cannot drift from what the reservation will book.
func (b *roleBudget) admitStep(step ladderStep) bool { return b.admit(step.estimateUSD) }

// 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
	// ran says, PER BATCH, whether the pass actually reached 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.
	//
	// ⛔ IT CANNOT BE A COUNT. Admission is not a prefix: an already-paid batch costs nothing and is
	// admitted whatever the budget says, so a batch the budget refuses can sit BEFORE batches that are
	// free to serve. A count answers «how many from the start», which silently drops every paid batch
	// behind the first unaffordable one — their replay was $0 and their result was already bought.
	ran         []bool
	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
	// regens is how many EXTRA rungs the ladder bought across the pass, and unusable how many batches ended
	// on a flagged reply anyway. They are separate from `dropped` because they answer a different question:
	// `dropped` is money that was never spent, these two are money that was.
	regens   int
	unusable int
	// stepsRefused counts rungs the PHASE BUDGET turned down. Without it a bank cut short by its own
	// sub-budget is indistinguishable from one the model simply could not answer — and the operator's
	// action differs: raise the budget, or look at the model.
	//
	// ⚠ RUNGS AND BATCHES COINCIDE HERE, and only because the walk STOPS at its first refusal — one walk
	// can contribute at most one. A ladder that went on trying cheaper rungs after a refusal would make
	// this a count of rungs over a smaller number of batches, and the summary key would start meaning
	// something else without changing its name.
	stepsRefused int
	// asked / answered are the PARSE side of the pass, and they are filled by the phase that owns the
	// parser — runBankRoleBatches never touches them, because what counts as an answer is a property of
	// the reply FORMAT, which differs per phase. One field, one writer, as everywhere else here.
	asked    int
	answered 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)), ran: make([]bool, 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.
	// ⛔ ADMISSION IS PER BATCH, NOT A PREFIX. An already-paid batch costs nothing, so refusing one
	// unaffordable batch must not take the paid batches BEHIND it: their replay is $0 and their result is
	// already bought, and dropping them buys nothing while losing a consolidated bank. The loop used to
	// `break` here and hand the consumer a prefix bound, which made the comment above («admitted
	// regardless») false for every paid batch that happened to sit after a refused one.
	// ⛔ ONE RUNNING NUMBER FOR THE WHOLE PASS, and it outlives this loop. The rungs the ladder buys later
	// are purchases this plan never saw, so they are booked against this same object — see roleBudget.
	budget := &roleBudget{limit: plan.budgetUSD, committed: spent, regens: r.Pipeline.Gates.Terminology.Regenerate}
	fits, plannedUSD := 0, 0.0
	admit := 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
		}
		if paid {
			admit[i] = true
			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 !budget.admit(want) {
			continue // this one cannot be afforded; the ones after it may still be free
		}
		admit[i] = true
		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 := range batches {
		if !admit[i] {
			continue
		}
		// 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}
		lr, aerr := r.runBankAttempt(ctx, st, snapID, ch, job, msgsPer[i], budget)
		// ⛔ THE MONEY IS TAKEN BEFORE THE VERDICT, the order this file already keeps for the classifier
		// pass and runStage for its hop: a walk that met a broken provider on its second rung still paid
		// for its first, and reading these fields behind the error check reports a pass that spent as one
		// that spent nothing.
		run.costUSD += lr.runCost
		run.cumUSD += lr.cumCost
		run.fresh = run.fresh || lr.anyFresh
		run.regens += lr.regens
		if lr.refusedStep != nil {
			run.stepsRefused++
		}
		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) {
				// ⛔ AND THE RUNG ALREADY PAID FOR IS KEPT, which the ladder made possible and the single
				// call before it could not. A ceiling refusing rung 0 buys nothing, so «leave the batch
				// untouched» was the whole truth; a ceiling refusing rung 1 arrives with rung 0 BOUGHT,
				// classified and holding whatever lines the model managed — on the cold run that shape was
				// four terms of twenty-two. Dropping it here would settle the money and discard the answer,
				// and the batch would read as one nobody called.
				if lr.judged > 0 {
					run.texts[i], run.ran[i] = lr.last.text, true
					run.unusable++
				}
				r.Log.WarnContext(ctx, "terminology "+logKind+": a USD ceiling refused this call; the terms of every LATER batch are left unchanged, and whatever this batch had already bought is kept",
					"book", r.Book.BookID, "role", plan.role, "batch", i,
					"rungs_paid_for_this_batch", lr.judged, "kept_reply", lr.judged > 0)
				break
			}
			return run, aerr
		}
		// A batch whose LAST rung still came back unusable is not the same thing as one the parser could
		// make nothing of: the engine itself said this reply is truncated or empty, and the ladder had
		// nothing left to try. Counted so the summary can tell the two apart (the parse counters below
		// cannot: an empty reply and a reply full of unparsable lines both answer zero terms).
		if !lr.last.cls.ok() {
			run.unusable++
			r.Log.WarnContext(ctx, "terminology "+logKind+": a batch is STILL unusable after every re-ask this phase could buy; its terms are left unchanged",
				"book", r.Book.BookID, "role", plan.role, "batch", i, "terms", len(batches[i]),
				"reason", string(lr.last.cls.Reason), "attempts", lr.attempts, "regenerations", lr.regens)
		}
		// The terminal rung's text is what the caller parses, INCLUDING an unusable one: a table cut at
		// the ceiling still holds the lines it managed to emit, and those terms are better banked than
		// thrown away. That is the behaviour this path always had — the ladder only adds the chance of a
		// whole reply before it — and `unusable` above is what stops the salvage from reading as success.
		run.texts[i] = lr.last.text
		run.ran[i] = true
	}
	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
}

// dropBankSettled removes, BEFORE any money moves, the candidates the bank has already settled: the role
// and the classifier were being paid for surfaces whose renderings the emission then threw away.
//
// ⛔ WHY THE TEST IS TWO CONDITIONS AND NOT ONE. "The bank already holds this surface" is NOT sufficient,
// and a filter written that way would take back with one hand what the sheet is for. The role is asked
// about banked surfaces ON PURPOSE: its answer is what CanonConflicts and ConsolidationKeyConflicts read,
// and those are the marks that tell the owner, on the sheet he signs by, that the book already calls this
// term something else. Measured on the corpus (13 stop tables, 1365 candidates, against the owner's own
// signed seed): 439 candidates sit on a surface the seed holds, and in 366 of them at least one draft
// disagreed with the seed's rendering. A one-condition filter would have bought a saving by silencing 366
// disputes — which is the defect the neighbouring row of this same pack exists to close.
//
// So a candidate is dropped only when BOTH hold:
//
//   - the bank holds its surface as a SEED SURFACE, decided by the emission's own predicate
//     (unsignedEngineSurfaces → membank.IsEngineUnsigned) rather than by a second rule of this function's
//     own, and it is banknote-ONLY, which is the exact population reverseSectionTerms drops. That pairing
//     is what makes the drop output-equivalent FOR THE DELTA: the term was never going to reach it;
//   - every rendering the drafts actually proposed for it equals the bank's, under the same target-form
//     fold the vote is counted with. One disagreement and it is PAID for, because that disagreement is the
//     finding. A candidate with no observed rendering at all is paid for too — banknote candidates always
//     carry one, so this cannot fire, but "no evidence" must never read as "evidence of agreement".
//
// ⛔ IT IS NOT OUTPUT-EQUIVALENT FOR THE SIGNING SHEET, and saying otherwise would be the same class of
// false description this pack exists to remove. The second condition tests what the DRAFTS proposed, not
// what the ROLE would have answered. Where the drafts agree with the bank and the role would have
// disagreed, the candidate is dropped and the disagreement is never produced, so the sheet loses a
// bankHolds mark it would otherwise carry — the very thing the neighbouring row of this pack widens.
// The population is bounded and measured: on the corpus (13 stop tables, 1365 candidates, against the
// owner's signed seed) 439 candidates sit on a surface the seed holds and 56 of them have every observed
// draft agreeing, so 56 of 1365 is the UPPER bound of what this can silence — the subset of those whose
// role answer would in fact have differed, which is smaller and not measurable without paying for it. The
// remaining 17 of that 439 carry no observed draft at all and are paid for by the rule above.
//
// ⛔ THE RESUME TRAP, AND WHY THE CURE IS "ALWAYS", NOT "SOMETIMES". Batches are packed greedily by rune
// budget, so removing any candidate changes the composition of at least one surviving batch, and a
// bank-role batch is addressed by the HASH of its request (bankCheckpointExists): a run whose composition
// differs from the run that paid finds no checkpoint and buys the pass again. The first cure written here
// was «stand down for a book that has already paid», and running it showed the cure was the bug: run one
// filtered and paid under the FILTERED composition, run two then stood down, rebuilt the FULL composition
// and re-bought everything. A rule that decides differently on the second run is not a resume rule.
//
// So the filter is unconditional, which makes the composition a pure function of the candidates and the
// bank and therefore identical on every resume — checkpoints replay for $0, forever, which is the property
// that was actually wanted. The one book this cannot help is one consolidated BEFORE the filter existed:
// its checkpoints carry the old composition and cannot be found again. That is a single re-consolidation,
// bounded by gates.terminology.budget_usd, and it is announced rather than discovered — see the warning
// below. Making even that unnecessary would take a checkpoint identity that survives a change of batch
// composition, which is a change to the money path's own addressing and belongs in its own decision.
func (r *Runner) dropBankSettled(ctx context.Context, cands []terminology.Candidate) (keptIdx []int, dropped int) {
	all := func() []int {
		out := make([]int, len(cands))
		for i := range cands {
			out[i] = i
		}
		return out
	}
	settled := bankSettledSurfaces(unsignedEngineSurfaces(r.glossaryRows()))
	if len(settled) == 0 {
		return all(), 0
	}
	var names []string
	for i, c := range cands {
		if bankSettles(settled, c) {
			dropped++
			if len(names) < 20 {
				names = append(names, c.Src)
			}
			continue
		}
		keptIdx = append(keptIdx, i)
	}
	if dropped == 0 {
		return all(), 0
	}
	// The saving is a NUMBER with its denominator, not a claim: "12 skipped" is unreadable without "of
	// 300", and the operator is the one who decides whether that is worth a look at his seed.
	r.Log.InfoContext(ctx, "terminology: candidates the bank has already settled are NOT sent to the paid role — the bank holds the surface and every draft proposed the rendering it holds, so the emission would have dropped the answer",
		"book", r.Book.BookID, "skipped", dropped, "of_candidates", len(cands), "still_paid", len(keptIdx),
		"terms", strings.Join(names, ", "), "terms_listed", len(names))
	return keptIdx, dropped
}

// pickCandidates rebuilds the paid subset from the CURRENT candidates. Called again after the passes that
// mutate them, so the request the role sees and the sheet the owner reads are built from one set of rows.
func pickCandidates(cands []terminology.Candidate, idx []int) []terminology.Candidate {
	out := make([]terminology.Candidate, 0, len(idx))
	for _, i := range idx {
		out = append(out, cands[i])
	}
	return out
}

// bankSettledSurfaces indexes the seed surfaces by their firing key, aliases included: a banknote proposal
// can name a term by an alias, and the emission's own guard reads aliases too.
func bankSettledSurfaces(seed []store.GlossaryEntry) map[string]store.GlossaryEntry {
	out := make(map[string]store.GlossaryEntry, len(seed))
	for _, e := range seed {
		if strings.TrimSpace(e.Dst) == "" {
			continue // nothing to agree with
		}
		out[text.NormalizeSourceKey(e.Src)] = e
		for _, a := range e.Aliases {
			out[text.NormalizeSourceKey(a.Alias)] = e
		}
	}
	return out
}

// bankSettles reports whether one candidate is settled by the bank — see dropBankSettled for why both
// halves are required.
func bankSettles(settled map[string]store.GlossaryEntry, c terminology.Candidate) bool {
	if c.Origin != terminology.OriginBanknote || len(c.Variants) == 0 {
		return false
	}
	e, held := settled[c.Key]
	if !held {
		return false
	}
	want := text.NormalizeTargetForm(e.Dst)
	for _, v := range c.Variants {
		if text.NormalizeTargetForm(v.Dst) != want {
			return false // the disagreement IS the finding: pay, and let it reach the sheet
		}
	}
	return true
}