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

// 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-v1-merge+kwic+c2-3"

// 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
	// 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
	CostUSD        float64 // what THIS run paid
	CumUSD         float64 // what the 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
}

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

// terminologyOpts resolves the sizing knobs (config value, else the engine default) in ONE place, so the
// batching and the request 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 batchRunes <= 0 {
		batchRunes = terminologyDefaultBatchRunes
	}
	if kwicPer <= 0 {
		kwicPer = terminologyDefaultKWICPer
	}
	if kwicWidth <= 0 {
		kwicWidth = terminologyDefaultKWICWidth
	}
	return batchRunes, kwicPer, kwicWidth
}

// 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,
		})
	}
	cands := terminology.Merge(ms, observed)
	_, kwicPer, kwicWidth := r.terminologyOpts()
	cands = terminology.AttachKWIC(cands, chunks, kwicPer, kwicWidth)

	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)
		}
	}
	for i := range cands {
		terminology.ScoreVariants(&cands[i], opts)
	}
	return cands
}

// 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, terminologyResult, error) {
	var res terminologyResult
	if !r.Pipeline.Gates.Terminology.Enabled || r.terminologyTemplate == nil || len(cands) == 0 {
		return nil, res, nil
	}
	res.Candidates = len(cands)
	for _, c := range cands {
		if c.Origin == terminology.OriginBanknote {
			res.Reverse++
		}
	}
	batchRunes, _, _ := r.terminologyOpts()
	batches := terminology.Batch(cands, batchRunes)
	res.Batches = len(batches)

	// The ESTIMATE comes before any money moves — the directive's «смета ДО живого вызова». It is the same
	// arithmetic the reservation uses, summed over the batches, so the number the operator reads and the
	// number the ledger reserves cannot drift apart.
	// The signed bank is read ONCE for the whole role, not per batch: it is the same law for every batch and
	// a store read per batch would be a query multiplied by a number the operator does not control.
	canon := r.approvedNeighbours()
	msgsPer := make([][]llm.Message, len(batches))
	for i, b := range batches {
		m, err := r.terminologyMessages(b, canon)
		if err != nil {
			return nil, res, err
		}
		msgsPer[i] = m
		res.EstimateUSD += r.terminologyEstimateUSD(m)
	}
	r.Log.InfoContext(ctx, "terminology: consolidating the bank over the whole book",
		"book", r.Book.BookID, "candidates", res.Candidates, "reverse_section", res.Reverse,
		"batches", res.Batches, "model", r.Pipeline.Gates.Terminology.Model,
		"estimate_usd", fmt.Sprintf("%.6f", res.EstimateUSD), "budget_usd", r.Pipeline.Gates.Terminology.BudgetUSD,
		"version", terminologyVersion)

	spent, err := r.Store.RoleSpentUSD(r.Book.BookID, roleTerminologist)
	if err != nil {
		return nil, res, fmt.Errorf("pipeline: terminology budget read: %w", err)
	}
	out := map[string]string{}
	st := config.Stage{Name: terminologyStageName, Role: roleTerminologist, Model: r.Pipeline.Gates.Terminology.Model}
	job, err := r.Store.EnsureJob(r.Book.BookID, 0, terminologyStageName, snapID)
	if err != nil {
		return nil, res, fmt.Errorf("pipeline: terminology job: %w", err)
	}
	for i, b := 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.terminologyCheckpointExists(st, snapID, ch, msgsPer[i])
		if perr != nil {
			return nil, res, perr
		}
		// The budget is checked BEFORE the call and against what the call would COST, not after the fact
		// against what has been spent: a check on spend alone lets the last permitted call push the total
		// past the ceiling by its own size, so the number in the config is not the bound it looks like. The
		// projection is the reservation's own upper bound (it sizes by max_tokens), so this refuses on the
		// conservative side — it can decline a call that would have fitted, never permit one that does not.
		if want := r.terminologyEstimateUSD(msgsPer[i]); !paid && spent+want > r.Pipeline.Gates.Terminology.BudgetUSD {
			r.Log.WarnContext(ctx, "terminology budget would be exceeded by the next batch; the remaining terms stay unconsolidated (status:auto)",
				"book", r.Book.BookID, "spent_usd", fmt.Sprintf("%.6f", spent), "next_batch_usd", fmt.Sprintf("%.6f", want),
				"budget_usd", r.Pipeline.Gates.Terminology.BudgetUSD, "batches_left", len(batches)-i)
			break
		}
		att, aerr := r.runTerminologyAttempt(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 consolidation" exactly as the repair sub-step degrades.
			if errors.Is(aerr, errReserveCeiling) {
				r.Log.WarnContext(ctx, "terminology call denied by a USD ceiling; the bank stays unconsolidated", "book", r.Book.BookID)
				break
			}
			return nil, res, aerr
		}
		res.CostUSD += att.runCost
		res.CumUSD += att.cumCost
		res.Fresh = res.Fresh || att.freshCall
		spent += att.runCost
		keys := make([]string, 0, len(b))
		for _, c := range b {
			keys = append(keys, c.Key)
		}
		got, bad := terminology.ParseReply(att.text, keys, text.NormalizeSourceKey)
		res.BadLines += bad
		// A batch that came back with NOTHING usable is a paid call that bought no terminology: an empty
		// completion, a truncation, a refusal, a reply in prose. Silence here would spend the money, mark
		// every term of the batch "the role declined" (§C2-7's auto mode, which is a DECISION) and exit 0.
		// Say it out loud instead — 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", bad,
				"reply_chars", len(att.text), "cost_usd", fmt.Sprintf("%.6f", att.runCost))
		}
		for k, v := range got {
			out[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, "; "))
	}
	r.Log.InfoContext(ctx, "terminology finished", "book", r.Book.BookID,
		"consolidated", res.Consolidated, "declined", res.Declined, "unanswered", res.Unanswered,
		"bad_lines", res.BadLines, "canon_conflicts", res.CanonConflicts, "cost_usd", fmt.Sprintf("%.6f", res.CostUSD))
	return out, 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

// terminologyCallBudget resolves the model and max_tokens of a terminology call — ONE definition, so the
// checkpoint probe and the call itself can never address different request hashes (the repair precedent).
func (r *Runner) terminologyCallBudget(msgs []llm.Message) (string, int) {
	model := r.Pipeline.Gates.Terminology.Model
	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

// terminologyEstimateUSD 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) terminologyEstimateUSD(msgs []llm.Message) float64 {
	model, maxTokens := r.terminologyCallBudget(msgs)
	promptEst := 0
	for _, m := range msgs {
		promptEst += EstimateTokens(m.Content)
	}
	return ledger.EstimateUSD(r.Pricer.PriceFor(model), promptEst, maxTokens,
		r.Models.AdditiveReasoningTokens(model, "", 0))
}

// terminologyCheckpointExists reports whether THIS batch was already paid for in an earlier run.
func (r *Runner) terminologyCheckpointExists(st config.Stage, snapID string, ch chunk.Chunk, msgs []llm.Message) (bool, error) {
	model, maxTokens := r.terminologyCallBudget(msgs)
	cp, err := r.Store.GetCheckpoint(RequestHash(Request{
		BookID: r.Book.BookID, Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx, Attempt: 0,
		Stage: st.Name, Role: roleTerminologist, Model: model, MaxTokens: maxTokens, SnapshotID: snapID, Messages: msgs,
	}))
	return cp != nil, err
}

// runTerminologyAttempt performs ONE terminology 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) runTerminologyAttempt(ctx context.Context, st config.Stage, snapID string, ch chunk.Chunk,
	job *store.Job, msgs []llm.Message) (stageAttempt, error) {

	model, maxTokens := r.terminologyCallBudget(msgs)
	tst := config.Stage{Name: st.Name, Role: roleTerminologist, Model: model}
	// The log axis, exactly as runStage sets it for a chunk stage: without it every terminology 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, which is what ch already addresses.
	ri, _ := obs.ReqInfoFromContext(ctx)
	ri.Book, ri.Chapter, ri.Chunk, ri.Stage, ri.Role = r.Book.BookID, ch.Chapter, ch.ChunkIdx, tst.Name, roleTerminologist
	ctx = obs.WithReqInfo(ctx, ri)
	return r.runAttempt(ctx, tst, model, snapID, ch, job, 0, maxTokens, msgs, false, false)
}

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