textmachine/backend/internal/pipeline/chunkrun.go

package pipeline

import (
	"context"
	"encoding/json"
	"fmt"
	"maps"
	"slices"
	"strings"
	"textmachine/backend/internal/store"
)

// chunkrun.go: disposition-петля уровня ЧАНКА (D2) — стадии по порядку, первый флаг
// останавливает чанк (дальше skipped, платной редактуры мусора нет), плюс горячий
// путь памяти v2 (Select→инъекция per-role→post-check E1) и дешёвые стиль-флаггеры.
// Новые роли-потребители инъекции Ф2 (annotator, voice-слой) добавляются здесь.

// classifyOutput resolves a completion's disposition: FIRST the always-on intrinsic
// classifier (echo/empty/refusal/length — Веха 2), then, only if that is ok AND the
// configurable coverage gate is enabled, the excision gate (шаг 6 / D12 Q3). Both are
// pure and deterministic over (source, output, finish), so a resumed checkpoint
// reproduces the identical verdict for free — the intrinsic part unconditionally, the
// gate part under the same coverage config (folded into the snapshot, so a gate change
// re-pins loudly). The coverage gate runs ONLY on the TRANSLATOR role's output (vs the
// original source): a monolingual editor legitimately restructures sentences, so gating
// it against the source would false-flag a correct edit (see the role check below).
// classifyOutput returns the disposition AND, for a cosmetic sanitizer strip, the resolved EXPORT
// text (the strip remainder) computed ONCE here (T3.4 / L8-stripcosmetic-recomputed-twice): the
// second return travels back to runStage as the authoritative export, so no site re-derives
// stripCosmetic and the "non-empty & clean" invariant is asserted over the very bytes that ship.
// The string is "" for every disposition except FlagSanitizerStripped.
func (r *Runner) classifyOutput(role, source, output, finish string, isFinal bool) (classification, string) {
	cls := classify(classifyInput{Source: source, Output: output, Finish: finish, TargetLang: r.Book.TargetLang})
	if !cls.ok() {
		return cls, ""
	}
	// Output-sanitizer gate (D30.3): a deterministic verdict-axis check for "instant unreadability"
	// defects no other gate catches — a leaked service preamble, a trailing note/edit block, a markdown
	// ### header, a Latin-script insertion, a contentless CJK-leak run. Opt-in (Gates.Sanitizer). Runs
	// ONLY on the FINAL stage's output (isFinal) — the text that SHIPS: an intermediate translator draft
	// is legitimately rough and the EDITOR's job is to clean it, so sanitizing the draft would wrongly
	// skip the recovering editor and drop the chunk to a placeholder (адверсариальное ревью). It is a
	// READABILITY gate whose patterns bake in Cyrillic/Russian conventions, so it is gated behind a
	// Russian target (D39 слой 7, L8-readability-gates-target-blind) — a no-op for any other pair, which
	// would false-flag (prod zh→ru is unaffected). Runs on the completion text like the intrinsic
	// classifier, so a resumed checkpoint re-derives the identical verdict; its rule version folds into
	// the snapshot only when enabled (sanitizerSnapshot).
	if isFinal && r.Pipeline.Gates.Sanitizer.Enabled && isRuTarget(r.Book.TargetLang) {
		if san := sanitizeOutput(output); san.total() > 0 {
			// Cosmetic-only leak (leading markdown header and/or contentless CJK-leak) → strip it and
			// export the remainder flagged, not drop the whole chunk (D35.4a). The guard is
			// load-bearing: we tag sanitizer_stripped ONLY when the strip yields a NON-EMPTY output that
			// is now clean, and we RETURN that stripped text so runStage/resume ship the exact bytes the
			// guard validated. Anything else falls through to the substantive skip. Pure over `output`,
			// so a resumed checkpoint re-derives the identical verdict.
			if san.cosmeticOnly() {
				if stripped := stripCosmetic(output); strings.TrimSpace(stripped) != "" && sanitizeOutput(stripped).total() == 0 {
					return classification{FlagSanitizerStripped, san.summary()}, stripped
				}
			}
			return classification{FlagSanitizerDefect, san.summary()}, ""
		}
	}
	if !r.Pipeline.Gates.Coverage.Enabled {
		return cls, ""
	}
	// The coverage gate compares the output against the ORIGINAL source, which is a
	// source→target translation only for the TRANSLATOR role. A monolingual editor
	// legitimately restructures/merges sentences, so gating ITS output against the
	// source with the translation corridor false-flags a correct edit as excision
	// (self-review finding). Editor/other-role fidelity is a Phase-2 concern (a
	// bilingual judge over the draft, §04 mode 2), not this excision gate.
	if role != roleTranslator {
		return cls, ""
	}
	if cov := coverageCheck(r.Pipeline.Gates.Coverage, source, output, r.Book.SourceLang, r.Book.TargetLang); !cov.cls.ok() {
		return cov.cls, ""
	}
	return cls, ""
}

// translateChunk drives ONE chunk through the stage list. Stages run in order,
// each feeding the next; the FIRST flagged stage stops the chunk — later stages
// are recorded `skipped` (no paid edit over a garbage draft, D2). It also runs the
// memory bank v2 hot path: it Selects the glossary injection once ($0, deterministic),
// injects it into the TRANSLATOR stage, post-checks the translator output (E1), and
// persists the per-chunk retrieval-state (observability). Returns the chunk outcome, the
// exact-matched entity ids (the next chunk's sticky_prev, A5), and an error only on an
// infra failure. stickyPrev is the prior chunks' exact matches in this chapter.
func (r *Runner) translateChunk(ctx context.Context, snapID string, ch Chunk, memSel memorySelection) (*ChunkOutcome, error) {
	out := &ChunkOutcome{Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx, Disposition: DispOK}
	prev := ""
	flagged := false
	var flagReason FlagReason
	// recovered carries a cosmetic sanitizer strip's cleaned export forward from the flagging
	// stage (D35.4a): non-empty only when the FINAL stage flagged FlagSanitizerStripped, "" for a
	// dropped substantive flag. Captured from the FIRST flagging stage (the sanitizer flags only
	// the final stage, so an upstream substantive flag leaves it "" and the chunk exports empty).
	recovered := ""
	// draftText is the FIRST (translator) stage's output — the pre-reflow baseline for the opt-in
	// regression guard's draft→final comparison (== final in a single-stage pipeline).
	draftText := ""
	// bankTelem carries the translator draft's banknote telemetry (WS4 point 10) into the per-chunk
	// retrieval-state. Zero for every channel-off / non-banknote chunk.
	var bankTelem bankFlags

	// Hot path: the glossary selection for this chunk is PRE-COMPUTED in W0 (precomputeSticky, WS1 §1б)
	// and passed in — the sticky chain is a cross-chunk dependency the parallel waves cannot recompute.
	// It is a pure deterministic function of (chunk, sticky_prev, frozen bank), so a resumed chunk
	// reproduces the identical injected bytes. Serialize the per-role injection blocks via the
	// role→renderer registry (D39 слой 7).
	injectionByRole := map[string]string{}
	if r.memory != nil {
		for role, render := range roleInjectionRenderers {
			injectionByRole[role] = render(memSel.injected) // pure → per-role result is map-order-independent
		}
		// The disposition-gated suppressor refused to let a lower-trust longer key eat a higher-trust
		// nested one (D39 слой 4): the approved term survives, but surface the seed-hygiene collision
		// loudly so an operator can reconcile the draft-nesting-over-approved seed (research/13 §7).
		if len(memSel.trustGated) > 0 {
			r.Log.WarnContext(ctx, "memory: lower-trust longer key refused from suppressing a higher-trust nested key (approved term preserved; reconcile the seed)",
				"chapter", ch.Chapter, "chunk", ch.ChunkIdx, "trust_gated", len(memSel.trustGated),
				"first", memSel.trustGated[0].Suppressor+"⊃"+memSel.trustGated[0].Protected)
		}
	}

	for stageIdx, st := range r.Pipeline.Stages {
		if flagged {
			// An earlier stage flagged → this stage is not attempted or billed.
			detail := fmt.Sprintf("skipped: an upstream stage was flagged (%s)", flagReason)
			if err := r.Store.UpsertChunkStatus(store.ChunkStatus{
				BookID: r.Book.BookID, Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx, Stage: st.Name,
				SnapshotID: snapID, Disposition: string(DispSkipped), FlagReason: string(flagReason),
				Detail: detail,
			}); err != nil {
				return out, fmt.Errorf("pipeline: record skipped chunk_status: %w", err)
			}
			out.Stages = append(out.Stages, StageResult{
				Stage: st.Name, Role: st.Role, Model: st.Model,
				Disposition: DispSkipped, FlagReason: flagReason, Detail: detail,
			})
			continue
		}

		// Resolve the per-role memory injection from the registry (D39 слой 7): the TRANSLATOR gets the
		// src→dst glossary block (keys matched against the source chunk); the BILINGUAL editor (D30.1)
		// gets the CONFIRMED dst forms as target-consistency constraints; every other role gets none
		// (empty string → the plain 2-message layout). The injection is a message, so it enters this
		// stage's request_hash automatically (a resumed chunk reproduces it deterministically).
		injection := injectionByRole[st.Role]
		sr, err := r.runStage(ctx, st, stageIdx, snapID, ch, prev, injection)
		if err != nil {
			return out, err
		}
		out.Stages = append(out.Stages, *sr)
		out.CostUSD += sr.CostUSD
		if st.Role == roleTranslator {
			bankTelem = sr.BankFlags // translator draft's banknote telemetry (WS4 point 10)
			// FL-2: the resume fast-path (resumeFromChunkStatus) serves final_hash — the banknote-CLEANED
			// derived checkpoint — and never re-parses the raw block, so it leaves BankFlags at the zero
			// value. Restore the telemetry the fresh run persisted instead of letting persistRetrievalState
			// zero the three banknote columns on every resume of a ready chunk (the comment below promises
			// an identical row). Guarded on an EMPTY BankFlags so the attempt-loop path — which re-derives
			// the telemetry correctly from the raw checkpoint — is never overwritten. Channel-off ⇒ always
			// zero anyway, so the store read is skipped.
			if bankTelem == (bankFlags{}) && sr.FromResume && r.Pipeline.Gates.Banknote.Enabled {
				if prev, err := r.Store.GetRetrievalState(r.Book.BookID, ch.Chapter, ch.ChunkIdx); err == nil && prev != nil {
					bankTelem = bankFlags{
						NLines:    prev.NBanknoteLines,
						ParseFail: prev.BanknoteParseFail != 0,
						Truncated: prev.BanknoteTruncated != 0,
					}
				}
			}
		}
		if sr.Disposition == DispFlagged {
			flagged = true
			flagReason = sr.FlagReason
			recovered = sr.RecoveredText
			continue
		}
		prev = sr.Text
		if stageIdx == 0 {
			draftText = sr.Text // the translator draft, before any reflow (regression-guard baseline)
		}
	}

	// Post-check the FINAL, exported output (E1): the reader sees the last stage's text
	// (the editor's), not the translator draft — the monolingual editor can drift an
	// approved term the translator got right, so checking the draft alone would miss it.
	// Runs on the fresh OR fully-resumed chunk (both carry the final text through `prev`),
	// so it is resume-reproducible. In the default FLAGGER mode a miss is recorded only in
	// the retrieval-state (observability); with the opt-in gate a miss flags the chunk
	// (glossary_miss). Skipped when a stage already flagged (no usable output to check).
	var postMisses []postcheckMiss
	outputChecked := false
	if r.memory != nil && !flagged && prev != "" {
		outputChecked = true
		postMisses = r.memory.postcheck(memSel.injected, prev)
		// The gate flips ONLY on CONFIRMED (approved) misses — an AMBIGUOUS miss is an
		// unverified candidate the model may legitimately reject, so it must not discard a
		// correct translation (external-review major #1).
		if r.Pipeline.Gates.Glossary.PostcheckGate && countConfirmedMisses(postMisses) > 0 {
			flagged = true
			flagReason = FlagGlossaryMiss
			r.Log.WarnContext(ctx, "glossary post-check gate flagged the chunk",
				"chapter", ch.Chapter, "chunk", ch.ChunkIdx, "confirmed_misses", countConfirmedMisses(postMisses))
		}
	}

	// Cheap deterministic style/number flaggers on the FINAL text (cheapgates.go): observability
	// only, never a disposition. Runs on the same fresh-or-resumed output as the post-check, so it
	// is resume-reproducible; skipped when a stage flagged (no usable output). Source is ch.Text
	// (the 万/億 magnitude gate compares source↔output). These are READABILITY flaggers whose rules
	// bake in Russian conventions (em-dash dialogue, ёфикатор, ru magnitude words), so they are gated
	// behind a Russian target (D39 слой 7, L8-readability-gates-target-blind) — a no-op for any other
	// pair, which would false-flag. Prod zh→ru is unaffected (target=ru).
	var cheap cheapGateResult
	if !flagged && prev != "" && isRuTarget(r.Book.TargetLang) {
		cheap = runCheapGates(ch.Text, draftText, prev, r.cheapGateConfig())
		if cheap.total() > 0 {
			r.Log.InfoContext(ctx, "cheap style gates flagged the chunk (observability, not a gate)",
				"chapter", ch.Chapter, "chunk", ch.ChunkIdx, "style_flags", cheap.total(),
				"dialogue_dash", cheap.DialogueDash, "yo", cheap.YoInconsistent,
				"translit_interj", cheap.TranslitInterj, "number_magnitude", cheap.NumberMagnitude)
		}
	}

	// Persist the per-chunk retrieval-state (registry gate #4: convert silent memory
	// degradation into a loud, visible record) plus the cheap style-flag counts. Deterministic +
	// idempotent, so a resume reproduces the identical row — the memory/style counts re-derive from
	// the deterministic selection + post-check, and the banknote columns are carried forward on resume
	// (FL-2 above) since the raw block is not re-parsed on the fast-path. Only when a glossary is
	// materialized (always true in a normal run — seedGlossary sets an at-least-empty bank).
	if r.memory != nil {
		if err := r.persistRetrievalState(snapID, ch, memSel, postMisses, outputChecked, cheap, bankTelem); err != nil {
			return out, err
		}
	}

	// The export contract (D39 слой 6): NORMALISE every final text before it ships — clean OR
	// flagged-stripped — so a cosmetic Unicode artifact (fullwidth glyph, U+3000 indent, stray CJK
	// punctuation, a leaked combining stress mark) is fixed uniformly, not only on a flagged chunk
	// (L5-normalization-fused-to-flag-path). A cosmetic sanitizer strip exports its cleaned remainder
	// (recovered != ""); every other flag exports "" (the contaminated output never reaches TM/export,
	// D2 / D35.4a), and exportNormalize("") == "". Pure ephemeral projection — no wire/checkpoint touch.
	if flagged {
		out.Disposition = DispFlagged
		out.FlagReason = flagReason
		out.FinalText = exportNormalize(recovered)
	} else {
		out.FinalText = exportNormalize(prev)
	}
	return out, nil
}

// injectionRenderer serializes a chunk's selected memory records into a role's injection message.
type injectionRenderer func(injected []pickedEntry) string

// roleInjectionRenderers maps a stage ROLE to the memory-injection renderer it consumes — a registry
// instead of a hand-edited switch (D39 слой 7, L8-role-injection-hardcoded-switch), so a new Ф2
// consumer role (annotator, voice-layer — README) is DATA + one renderer, not surgery on a switch.
// A role absent from the map gets no injection. Renderers are pure, so the per-role result is
// deterministic regardless of map iteration order (no map-order in OUTPUT — invariant #6).
var roleInjectionRenderers = map[string]injectionRenderer{
	roleTranslator: renderGlossaryBlock,         // src→dst glossary block
	roleEditor:     renderEditorConstraintBlock, // CONFIRMED dst forms (target-consistency constraints)
}

// persistRetrievalState writes the per-chunk observability record from the deterministic
// selection + the post-check result. n_exact_hits/n_sticky/n_ambiguous count the INJECTED
// records (what the model saw); spoiler/eviction are the dropped-and-logged totals;
// post-check misses are recorded only when the translator actually produced text.
func (r *Runner) persistRetrievalState(snapID string, ch Chunk, sel memorySelection, misses []postcheckMiss, outputChecked bool, cheap cheapGateResult, bank bankFlags) error {
	rs := store.RetrievalState{
		BookID: r.Book.BookID, Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx, SnapshotID: snapID,
		NStyleFlags:       cheap.total(),
		NBanknoteLines:    bank.NLines,
		BanknoteParseFail: boolToStoreInt(bank.ParseFail),
		BanknoteTruncated: boolToStoreInt(bank.Truncated),
	}
	if cheap.total() > 0 {
		if b, err := json.Marshal(cheap); err == nil {
			rs.StyleDetail = string(b)
		}
	}
	for _, p := range sel.injected {
		if p.via == "sticky" {
			rs.NSticky++
		} else {
			rs.NExactHits++
		}
		if p.disp == memAmbiguous {
			rs.NAmbiguousFlagged++
		}
	}
	rs.NSpoilerBlocked = len(sel.rejected)
	rs.NEvicted = len(sel.evicted)
	// Loud record of the disposition-gated suppressor (D39 слой 4): a longer lower-trust key refused
	// from eating a nested higher-trust one — the term-drift code root, now visible instead of a
	// silent drop (research/13 §7). Detail carries the refused suppressor→protected pairs for a human.
	rs.NTrustGatedSuppress = len(sel.trustGated)
	if len(sel.trustGated) > 0 {
		if b, err := json.Marshal(sel.trustGated); err == nil {
			rs.TrustGateDetail = string(b)
		}
	}
	if outputChecked {
		// n_postcheck_miss is the CONFIRMED-miss count (the actionable consistency-failure
		// signal, external-review #1); the detail carries ALL misses (confirmed AND the
		// ambiguous forced-post-check ones, each disp-tagged) for the human.
		rs.NPostcheckMiss = countConfirmedMisses(misses)
		if len(misses) > 0 {
			if b, err := json.Marshal(misses); err == nil {
				rs.PostcheckDetail = string(b)
			}
		}
	}
	ids := slices.Sorted(maps.Keys(sel.activeIDs))
	if ids == nil {
		// slices.Sorted даёт NIL на пустой карте → json.Marshal рендерил бы "null",
		// а исторический формат персистентной колонки — "[]" (чанк без точных
		// матчей — обычный кейс); держим байты стабильными (находка селфревью №4).
		ids = []string{}
	}
	if b, err := json.Marshal(ids); err == nil {
		rs.InjectedIDs = string(b)
	}
	return r.Store.UpsertRetrievalState(rs)
}

// boolToStoreInt maps a bool telemetry flag to the store's 0/1 integer column form.
func boolToStoreInt(b bool) int {
	if b {
		return 1
	}
	return 0
}

// cheapGateConfig builds the cheap-gate knobs from the book brief: the ё-policy and the
// lower-cased per-project interjection allowlist. Pure, no store access.
func (r *Runner) cheapGateConfig() cheapGateConfig {
	allow := make(map[string]bool, len(r.Book.StyleAllowlist))
	for _, s := range r.Book.StyleAllowlist {
		allow[s] = true
	}
	return cheapGateConfig{
		yoPolicy:          r.Book.YoPolicy,
		allowlist:         allow,
		regressionEnabled: r.Pipeline.Gates.RegressionGuard.Enabled,
	}
}