textmachine/backend/internal/pipeline/chunkrun.go

package pipeline

import (
	"context"
	"encoding/json"
	"fmt"
	"maps"
	"slices"
	"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).
func (r *Runner) classifyOutput(role, source, output, finish string, isFinal bool) classification {
	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 broken word form.
	// 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 (адверсариальное ревью). Runs on the completion text like the
	// intrinsic classifier, so a resumed checkpoint re-derives the identical verdict for free;
	// its rule version folds into the snapshot only when enabled (sanitizerSnapshot).
	if isFinal && r.Pipeline.Gates.Sanitizer.Enabled {
		if san := sanitizeOutput(output); san.total() > 0 {
			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, stickyPrev map[string]injectionDisposition) (*ChunkOutcome, map[string]injectionDisposition, error) {
	out := &ChunkOutcome{Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx, Disposition: DispOK}
	prev := ""
	flagged := false
	var flagReason FlagReason

	// Hot path: select the glossary records for this chunk ONCE (deterministic, $0), and
	// serialize the translator injection block. Recomputed on every run (resumed chunks
	// included) so the sticky window and the injected bytes are resume-stable.
	var memSel memorySelection
	var translatorInjection, editorInjection string
	activeIDs := map[string]injectionDisposition{}
	if r.memory != nil {
		memSel = r.memory.Select(ch.Text, ch.Chapter, stickyPrev, r.Pipeline.Context.GlossaryTokenBudget)
		translatorInjection = renderGlossaryBlock(memSel.injected)
		editorInjection = renderEditorConstraintBlock(memSel.injected)
		activeIDs = memSel.activeIDs
	}

	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, activeIDs, 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
		}

		// Inject the glossary per role. The TRANSLATOR gets the src→dst block (its keys are
		// matched against the source chunk); the monolingual EDITOR gets the CONFIRMED dst
		// forms as target-consistency constraints (no source — decisions-log D1); every other
		// stage gets none. Empty injection → 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 from the frozen bank).
		injection := ""
		switch st.Role {
		case roleTranslator:
			injection = translatorInjection
		case roleEditor:
			injection = editorInjection
		}
		sr, err := r.runStage(ctx, st, stageIdx, snapID, ch, prev, injection)
		if err != nil {
			return out, activeIDs, err
		}
		out.Stages = append(out.Stages, *sr)
		out.CostUSD += sr.CostUSD
		if sr.Disposition == DispFlagged {
			flagged = true
			flagReason = sr.FlagReason
			continue
		}
		prev = sr.Text
	}

	// 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).
	var cheap cheapGateResult
	if !flagged && prev != "" {
		cheap = runCheapGates(ch.Text, 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 re-derives the identical row. 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); err != nil {
			return out, activeIDs, err
		}
	}

	if flagged {
		out.Disposition = DispFlagged
		out.FlagReason = flagReason
		out.FinalText = "" // garbage/refusal/glossary-miss never propagates to the next chunk or export
	} else {
		out.FinalText = prev
	}
	return out, activeIDs, nil
}

// 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) error {
	rs := store.RetrievalState{
		BookID: r.Book.BookID, Chapter: ch.Chapter, ChunkIdx: ch.ChunkIdx, SnapshotID: snapID,
		NStyleFlags: cheap.total(),
	}
	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)
	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)
}

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