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/ledger"
	"textmachine/backend/internal/llm"
	"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 injected-with-⟨проверить⟩ mode of §C2-7), never `approved` — the miner has no
// code path to that word and this role adds none. A term it declines emits `auto`, inert. So the worst a
// bad terminologist call can do is put a marked, unverified proposal in front of the editor and the owner;
// it can never make the book use a word nobody signed.
//
// WHAT IT COSTS. A handful of batched calls per BOOK (not per chunk), on a cheap model, over a block whose
// size is bounded by config. The estimate is logged BEFORE the first call, the calls run on the shared
// reserve→call→settle+checkpoint path so a resume replays them for $0, and the whole class is capped by
// gates.terminology.budget_usd.

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

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

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

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

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

// terminologyResult is the run's outcome for the report and the logs. Derived from what happened; never
// stored as a row of its own.
type terminologyResult struct {
	Candidates   int // merged candidates handed to the role
	Reverse      int // of those, banknote-only (the coverage the miner structurally cannot see)
	Batches      int // calls attempted
	Consolidated int // terms that came back with a rendering → status:draft
	Declined     int // terms the role explicitly could not render → status:auto
	Unanswered   int // terms no reply line covered (also status:auto — silence is not a decision)
	BadLines     int // reply lines the parser refused
	// OffLanguage counts refused lines whose rendering was not in the target's script — separate from
	// BadLines because it means the model answered in another language, not that it broke the format.
	OffLanguage int
	// CanonConflicts counts consolidated renderings that contradict a row the owner already signed. It is
	// OBSERVABILITY, never a gate: the rows stay unverified either way, and this is what tells the owner
	// which of them to look at first.
	CanonConflicts int
	// SelfConflicts counts consolidations that contradict ANOTHER consolidation of the same run (§G2) — the
	// majority class on a live bank, and the one nothing looked at before this pack. Contradictions carries
	// the same finding per key, so the stop table can mark the rows instead of printing a bare total.
	SelfConflicts  int
	Contradictions map[string][]string
	// 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: r.approvedNeighbours()}
	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
}

// 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 (r *Runner) approvedNeighbours() []terminology.Neighbour {
	rows, err := r.Store.GlossaryForBook(r.Book.BookID)
	if err != nil {
		return nil // an anchor is an improvement, never a precondition — a read failure degrades, not aborts
	}
	out := make([]terminology.Neighbour, 0, len(rows))
	for _, e := range rows {
		if e.Status == "approved" && e.Dst != "" {
			out = append(out, terminology.Neighbour{Src: text.NormalizeSourceKey(e.Src), Dst: e.Dst})
		}
	}
	sort.Slice(out, func(i, j int) bool { return out[i].Src < out[j].Src })
	return out
}

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

	// §2 type re-derivation, BEFORE the render: a focused pass re-classifies every candidate, so a realia
	// surface mistyped as a name (元石) is no longer FORCED to a transliteration. The corrected type routes
	// conformance (re-scored here), primes the wire block, and is returned so the caller stamps it as the
	// banked type. Off (or unanswered) → the heuristic draft type stands.
	classified, crun, cerr := r.runClassifier(ctx, snapID, cands)
	if cerr != nil {
		return nil, nil, res, cerr
	}
	res.ClassifyCostUSD = crun.costUSD
	if len(classified) > 0 {
		res.Reclassified = applyTypes(cands, classified)
		opts := r.scoreOpts()
		for i := range cands {
			terminology.ScoreVariants(&cands[i], opts)
		}
	}

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

	// The signed bank is read ONCE for the whole role, not per batch: it is the same law for every batch.
	canon := r.approvedNeighbours()
	plan := bankRolePlan{
		role: roleTerminologist, budgetUSD: r.Pipeline.Gates.Terminology.BudgetUSD,
		messages: func(b []terminology.Candidate) ([]llm.Message, error) { return r.terminologyMessages(b, canon) },
	}
	run, err := r.runBankRoleBatches(ctx, snapID, plan, batches, "render")
	if err != nil {
		return nil, nil, res, err
	}
	res.EstimateUSD, res.CostUSD, res.CumUSD, res.Fresh = run.estimateUSD, run.costUSD, run.cumUSD, run.fresh

	out := map[string]string{}
	res.Conf = map[string]int{}
	for i, b := range batches {
		if i >= run.attempted {
			break // the budget or a ceiling stopped the pass here; runBankRoleBatches already said so
		}
		if run.texts[i] == "" {
			// An EMPTY completion on a call that was actually made. It used to `continue` in silence, which is
			// how a paid batch that returned nothing became indistinguishable from «the role declined these
			// terms» — a DECISION in §C2-7 — with the run exiting 0.
			r.Log.WarnContext(ctx, "terminology: a paid batch came back with an EMPTY completion; its terms stay unconsolidated",
				"book", r.Book.BookID, "batch", i, "asked", len(b), "answered", 0)
			continue
		}
		got, conf, st := terminology.ParseReply(run.texts[i], candKeys(b), text.NormalizeSourceKey, r.targetScript)
		res.BadLines += st.Bad
		res.OffLanguage += st.OffLanguage
		// Asked-vs-answered per batch: the one number that separates «the model skipped half the block» from
		// «the parser refused half the lines», and neither is visible in a total.
		r.Log.InfoContext(ctx, "terminology render batch", "book", r.Book.BookID,
			"batch", i, "asked", len(b), "answered", len(got), "bad_lines", st.Bad, "reply_chars", len(run.texts[i]))
		// A batch answered largely in another language is the measured cold-start failure, not noise: say it
		// while the run is happening, naming the lines, because the terms themselves just stay auto.
		if st.OffLanguage > 0 {
			r.Log.WarnContext(ctx, "terminology: the model answered in another script; those lines are REFUSED (the terms stay unconsolidated) — a book with few signed rows gives the model no target-language anchor",
				"book", r.Book.BookID, "batch", i, "terms", len(b), "off_language", st.OffLanguage,
				"target_script", r.Pipeline.Gates.Terminology.TargetScript, "lines", strings.Join(st.OffLanguageSamples, "; "))
		}
		// A batch that came back with NOTHING usable is a paid call that bought no terminology. Silence here
		// would spend the money, mark every term "the role declined" and exit 0. Say it out loud; the run still
		// continues, because an unconsolidated bank is the state this step started from, not a broken one.
		if len(got) == 0 {
			r.Log.WarnContext(ctx, "terminology: a paid batch returned nothing the parser could use; its terms stay unconsolidated",
				"book", r.Book.BookID, "batch", i, "terms", len(b), "bad_lines", st.Bad, "reply_chars", len(run.texts[i]))
		}
		for k, v := range got {
			out[k] = v
		}
		for k, v := range conf {
			res.Conf[k] = v
		}
	}
	for _, c := range cands {
		v, answered := out[c.Key]
		switch {
		case !answered:
			res.Unanswered++
		case v == "":
			res.Declined++
		default:
			res.Consolidated++
		}
	}
	// The canon check is $0 and runs on EVERY consolidation, signed anchor or not: the anchor is a nudge to
	// a model, and a nudge that is not verified is a hope. Named terms, not a bare count — a conflict is
	// actionable only if the operator is told which rendering fights which signature.
	if conflicts := terminology.CanonConflicts(cands, out, canon); len(conflicts) > 0 {
		res.CanonConflicts = len(conflicts)
		named := make([]string, 0, len(conflicts))
		for _, cf := range conflicts {
			named = append(named, fmt.Sprintf("%s→%q contradicts the signed %s→%q", cf.Src, cf.Dst, cf.CanonSrc, cf.CanonDst))
		}
		r.Log.WarnContext(ctx, "terminology: consolidated renderings contradict the SIGNED bank; they stay unverified (⟨проверить⟩) and are the first rows to review at the stop",
			"book", r.Book.BookID, "conflicts", len(conflicts), "terms", strings.Join(named, "; "))
	}
	// §G2: the same check with the run's OWN consolidations on the right-hand side. On a live bank this is
	// the MAJORITY of the contradictions (18 of 149, research/24 §A4) and until now nobody looked: the canon
	// check can only see rows the owner already signed, and a book being consolidated for the first time has
	// almost none. $0, evidence-side, never a gate.
	if self := terminology.ConsolidationConflicts(cands, out); len(self) > 0 {
		res.SelfConflicts = len(self)
		res.Contradictions = map[string][]string{}
		named := make([]string, 0, len(self))
		for _, cf := range self {
			named = append(named, fmt.Sprintf("%s→%q drops %s→%q", cf.Src, cf.Dst, cf.PartSrc, cf.PartDst))
			res.Contradictions[cf.Src] = append(res.Contradictions[cf.Src], fmt.Sprintf("%s→%q", cf.PartSrc, cf.PartDst))
		}
		r.Log.WarnContext(ctx, "terminology: consolidations of THIS run contradict each other — a compound's rendering drops the rendering the same reply gave its own part; they stay unverified and are review rows at the stop",
			"book", r.Book.BookID, "conflicts", len(self), "terms", strings.Join(named, "; "))
	}
	// The $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, "; "))
	}
	r.Log.InfoContext(ctx, "terminology finished", "book", r.Book.BookID,
		"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, 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. Off (or no template) → a nil map 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).
func (r *Runner) runClassifier(ctx context.Context, snapID string, cands []terminology.Candidate) (map[string]string, bankRoleRun, error) {
	g := r.Pipeline.Gates.Terminology
	if !g.ClassifyTypes || r.classifierTemplate == nil || len(cands) == 0 {
		return 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, run, err
	}
	out := map[string]string{}
	bad := 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",
				"book", r.Book.BookID, "batch", i, "asked", len(b), "answered", 0)
			continue
		}
		got, st := terminology.ParseTypes(run.texts[i], candKeys(b), text.NormalizeSourceKey)
		bad += st.Bad
		r.Log.InfoContext(ctx, "terminology classify batch", "book", r.Book.BookID,
			"batch", i, "asked", len(b), "answered", len(got), "bad_lines", st.Bad)
		for k, v := range got {
			out[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)
	}
	return out, 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)
	promptEst := 0
	for _, m := range msgs {
		promptEst += EstimateTokens(m.Content)
	}
	// 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 ledger.EstimateUSD(r.Pricer.PriceFor(st.Model), promptEst, maxTokens,
		r.Models.AdditiveReasoningTokens(st.Model, st.Reasoning, st.ReasoningMaxTokens))
}

// 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)
	return r.runAttempt(ctx, st, st.Model, snapID, ch, job, 0, maxTokens, msgs, 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")
}

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

// 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)
	}
	for i := range batches {
		// 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}
		paid, perr := r.bankCheckpointExists(st, snapID, ch, msgsPer[i])
		if perr != nil {
			return run, perr
		}
		// The budget is checked BEFORE the call and against what the call would COST (the reservation's own
		// upper bound), so it refuses on the conservative side and the config number is the bound it looks like.
		if want := r.bankCallEstimateUSD(st, msgsPer[i]); !paid && spent+want > plan.budgetUSD {
			r.Log.WarnContext(ctx, "terminology "+logKind+": budget would be exceeded by the next batch; the remaining terms are left unchanged",
				"book", r.Book.BookID, "role", plan.role, "spent_usd", fmt.Sprintf("%.6f", spent),
				"next_batch_usd", fmt.Sprintf("%.6f", want), "budget_usd", plan.budgetUSD, "batches_left", len(batches)-i)
			break
		}
		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
}

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

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