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
	// 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
	// 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)
	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)
		}
		// 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.
	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
	// ⚠ 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 {
		r.Log.WarnContext(ctx, "terminology: this book had already paid for the bank role BEFORE this run, and this run bought the pass again — those earlier calls were made under a batch composition the already-banked filter does not reproduce, so their checkpoints could not be found. 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 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
		}
	}
	// 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)
	}
	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
}

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