textmachine/backend/internal/pipeline/disposition.go

package pipeline

import (
	"fmt"
	"regexp"
	"strings"
	"unicode"
	"unicode/utf8"

	"textmachine/backend/internal/llm"
)

// disposition.go: the per-chunk×stage verdict machinery of the Веха-2 runner
// (D2). It answers three questions about a completion — is it usable, and if not
// WHY, and may we re-attack it — WITHOUT ever touching the wire (classify is a
// pure function of the completion text + finish reason + a little context, so a
// resume reproduces the same verdict from the same checkpoint, no re-billing).
//
// This is deliberately SEPARATE from the configurable coverage QA-gate
// (Gates.Coverage, шаг 6): that gate is opt-in and threshold-driven; the
// classifier here is intrinsic runner robustness that is always on, so a bad
// chunk is flagged-and-skipped instead of poisoning the pipeline (empty draft →
// empty edit → empty export) or wedging the whole book on a single call.

// Disposition is the resolved state of a chunk×stage over its checkpoints.
type Disposition string

const (
	// DispOK — a usable completion; its text feeds the next stage and is the
	// authoritative checkpoint.
	DispOK Disposition = "ok"
	// DispFlagged — the completion is unusable (refusal / echo / truncation loop
	// / empty / decode error) and retries (if any) are exhausted; the chunk is
	// flagged for a human and later stages are skipped. Money is still accounted.
	DispFlagged Disposition = "flagged"
	// DispSkipped — this stage was NOT attempted because an earlier stage of the
	// same chunk was flagged (no garbage draft → no paid edit over garbage).
	DispSkipped Disposition = "skipped"
)

// FlagReason is the tagged cause of a flag — a contract enum like the Finish*
// constants (D2.2: tagging drives retry policy). The runner re-attacks ONLY the
// retryable subset; the rest are deterministic per model/channel, so a same-model
// retry would just re-refuse and re-bill (routing to channel B / escalation is
// Фаза 2). Some constants are DEFINED here for contract completeness but are not
// yet emitted by Веха 2 — see the notes — they belong to later steps.
type FlagReason string

const (
	reasonOK FlagReason = "" // sentinel: not a flag

	// Emitted by classify() — the retryable pair (D2: bigger max_tokens on the
	// attempt axis, up to the regenerate cap, then flag).
	FlagLength FlagReason = "length" // truncated at max_tokens (a genuine cut, not a loop)
	FlagEmpty  FlagReason = "empty"  // 2xx with no usable text (thinking likely ate the whole budget)

	// Emitted by classify() — deterministic, NOT retryable on the same model.
	FlagLoopDegenerate FlagReason = "loop_degenerate" // repetition loop at a length cut (bigger budget just buys more loop, D2.3)
	FlagHardRefusal    FlagReason = "hard_refusal"    // provider finish_reason=refusal
	FlagSoftRefusal    FlagReason = "soft_refusal"    // refusal-blacklist match on a short output (the real insurance, D2.4)
	FlagContentFilter  FlagReason = "content_filter"  // provider finish_reason=content_filter (best-effort — emission unverified across DS/GLM/Kimi/grok, D2.4)
	FlagCJKArtifact    FlagReason = "cjk_artifact"    // output echoed the CJK source instead of translating (D3.4; фолбэк-черновик — шаг 7, пока просто флаг)

	// Emitted by the runner's billed-decode path.
	FlagDecodeError FlagReason = "decode_error" // 2xx with an unreadable body — billed, conservatively settled, flagged

	// Reserved for later steps — DEFINED for contract stability, NOT emitted by
	// Веха 2. coverage_fail / excision_suspect are verdicts of the configurable
	// coverage gate (шаг 6); hard_block / upstream_not_ok are for HTTP-level
	// content blocks handled when escalation lands (шаг 7).
	FlagCoverageFail    FlagReason = "coverage_fail"
	FlagExcisionSuspect FlagReason = "excision_suspect"
	FlagHardBlock       FlagReason = "hard_block"
	FlagUpstreamNotOK   FlagReason = "upstream_not_ok"

	// FlagGlossaryMiss is the memory-bank post-check verdict (E1, шаг 4): an approved
	// term's src fired in the chunk but no accepted dst form appears in the output — the
	// model ignored the glossary, or we injected the wrong dst and it obeyed. Emitted
	// ONLY when the opt-in glossary post-check gate is enabled (config); in the default
	// flagger mode a miss is recorded in the retrieval-state, not a disposition. NOT
	// retryable (a same-model retry re-produces the same rendering) and NOT auto-
	// escalatable in v1 (the L3 targeted re-ask is a Phase-2 remedy, research/14 §9).
	FlagGlossaryMiss FlagReason = "glossary_miss"

	// FlagSanitizerDefect is the output-sanitizer verdict for a SUBSTANTIVE defect (D30.3,
	// sanitizer.go): the stage output carries an "instant unreadability" defect with no reliable
	// removable boundary — a leaked service preamble, a trailing note/edit block, a Latin-script
	// insertion, or a broken word form. Emitted ONLY when the opt-in Gates.Sanitizer is enabled;
	// the contaminated output is flagged AND skipped (D2 flag+skip) so it never commits to
	// TM/export (FinalText stays ""). NOT retryable (a same-model retry re-produces the artifact)
	// and NOT escalatable in v1 (the editor is pinned; a redrive surfaces the defect to a human).
	FlagSanitizerDefect FlagReason = "sanitizer_defect"

	// FlagSanitizerStripped is the output-sanitizer verdict for a COSMETIC-ONLY defect that was
	// STRIPPED and the remainder EXPORTED (D38 infra-pack, D35.4a): a leading markdown «### Глава»
	// header and/or a stray CJK-leak run, removed deterministically by stripCosmetic. Unlike
	// FlagSanitizerDefect the chunk is NOT lost — the cleaned text is committed as the export
	// (final_hash points at a derived sanitized checkpoint, so the standard
	// final_hash→checkpoint.response_text export contract yields the cleaned text) — but it STAYS
	// flagged so a human verifies the auto-clean ("не терять чанк ... флаг для человека"). NOT
	// retryable / NOT escalatable (deterministic; a redrive would re-produce the same leak).
	FlagSanitizerStripped FlagReason = "sanitizer_stripped"
)

// classifierVersion versions the INTRINSIC classify() verdict logic — the refusal
// blacklist, the CJK-echo threshold (cjkEchoThreshold), the degeneration detector and
// the order in which they run. classify() resolves the ok↔flagged disposition and is
// re-run on resume over legacy / in-flight-crash checkpoints, so a change to any
// threshold (e.g. cjkEchoThreshold 0.15→0.20, a new refusal pattern) would otherwise
// keep the SAME snapshot id and silently re-verdict a resumed chunk — flagged→ok
// re-runs a skipped stage (fresh spend), ok→flagged burns a fresh escalation hop.
// Folded into the snapshot exactly like coverageGateVersion, so such a change is a
// loud --resnapshot, not a silent divergence (external-review finding).
const classifierVersion = "classify-v1-refusal+echo015+loop"

// decodeErrorFinish is the finish_reason the runner stores on a billed-but-
// unreadable 2xx (BilledDecodeError). classify() recognises it so a RESUMED
// decode checkpoint re-resolves to the same FlagDecodeError verdict the live
// path assigned — live and resume must agree (determinism of the resolve).
const decodeErrorFinish = "decode_error"

// retryable reports whether a flag may be re-attacked on the SAME model along the
// attempt axis. Only length/empty: both are budget symptoms a bigger max_tokens
// can cure. Everything else is deterministic (refusal/filter/echo/loop/decode) —
// re-attacking burns money on a guaranteed repeat (D2.2/D2.3).
func (r FlagReason) retryable() bool {
	return r == FlagLength || r == FlagEmpty
}

// escalatable reports whether a flag is a DETERMINISTIC content-failure that a
// DIFFERENT model might fix — the single-hop escalation trigger (D12, the
// deterministic-content-failure class). These arrive as HTTP 200 (the provider
// billed a "translation" that is echo / excision / a refusal), so a same-model
// retry just re-produces them (which is why they are non-retryable) — escalation
// routes them to another model ONCE. length/empty are excluded (retryable on the
// same model with a bigger budget); decode_error is transport-shaped (a billed
// unreadable body), not a content verdict another model would reliably avoid.
func (r FlagReason) escalatable() bool {
	switch r {
	case FlagCJKArtifact, FlagExcisionSuspect, FlagLoopDegenerate,
		FlagHardRefusal, FlagSoftRefusal, FlagContentFilter:
		return true
	}
	return false
}

// disposition maps a reason to the resolved chunk×stage state.
func (r FlagReason) disposition() Disposition {
	if r == reasonOK {
		return DispOK
	}
	return DispFlagged
}

// classifyInput is everything classify needs. It is more than the "(text,
// finish)" shorthand because a faithful 1:1 port of eval/refusal_bench.py
// (03-implementation-notes §3.7) needs the source length (soft-refusal is a
// refusal pattern on an ANOMALOUSLY SHORT output — a long translation that
// merely quotes "I'm sorry, but…" as dialogue must NOT flag) and the target
// language (CJK-echo detection is meaningless when translating INTO a CJK
// language).
type classifyInput struct {
	Source     string
	Output     string
	Finish     string
	TargetLang string
}

// classification is classify's verdict.
type classification struct {
	Reason FlagReason
	Detail string
}

func (c classification) ok() bool { return c.Reason == reasonOK }

// classify is the single verdict function. ORDER MATTERS (D2.1/D2.4): the
// deterministic hard signals (decode / content_filter / refusal finish) and the
// text refusal-blacklist / CJK-echo are checked BEFORE length/empty, so a
// truncated refusal is flagged as a refusal (deterministic, no paid re-attack)
// rather than as a length cut (retryable). Pure: no time, no randomness, no map
// iteration — resume reproduces the identical verdict from the identical
// checkpoint.
func classify(in classifyInput) classification {
	// Strip a <think>…</think> block the model may have leaked into content
	// (mirrors refusal_bench) before judging emptiness/length.
	out := strings.TrimSpace(stripThink(in.Output))
	finish := in.Finish

	// 1) Deterministic finish-reason signals — never retried.
	switch finish {
	case decodeErrorFinish:
		return classification{FlagDecodeError, "billed 2xx with an unreadable body"}
	case llm.FinishContentFilter:
		return classification{FlagContentFilter, "provider finish_reason=content_filter"}
	case llm.FinishRefusal:
		return classification{FlagHardRefusal, "provider finish_reason=refusal"}
	}

	// 2) Text refusal-blacklist / CJK-echo — ДО length/empty (only meaningful on
	//    non-empty text; a truncated refusal must not become a paid length retry).
	if out != "" {
		if isRefusal(out, in.Source) {
			return classification{FlagSoftRefusal, "refusal-blacklist match on a short output"}
		}
		if !isCJKTarget(in.TargetLang) {
			if share := cjkShare(out); share > cjkEchoThreshold {
				return classification{FlagCJKArtifact, fmt.Sprintf("CJK share %.0f%% in output (untranslated echo)", share*100)}
			}
		}
	}

	// 3) Empty completion — retryable (thinking usually ate the entire budget).
	if out == "" {
		return classification{FlagEmpty, "empty completion (finish=" + finish + ")"}
	}

	// 4) Length cut — retryable, UNLESS the output is a degenerate repetition
	//    loop, in which case a bigger budget only buys more loop (D2.3): flag now.
	if finish == llm.FinishLength {
		if degenerateLoop(out) {
			return classification{FlagLoopDegenerate, "repetition loop at a length cut"}
		}
		return classification{FlagLength, "truncated at max_tokens"}
	}

	// 5) A usable completion.
	return classification{reasonOK, ""}
}

// --- refusal blacklist (en/ru/zh/ja) — ported 1:1 from eval/refusal_bench.py ---

var refusalPatterns = []string{
	`I (?:can(?:'|no)t|cannot|won'?t|am (?:not able|unable)) (?:to )?(?:assist|help|translate|continue|provide|fulfill)`,
	`I'?m sorry,? but`,
	`against (?:my|our) (?:content )?polic`,
	`content polic(?:y|ies)`,
	`не могу (?:помочь|перевести|продолжить|выполнить)`,
	`я не буду переводить`,
	`противоречит (?:политике|правилам)`,
	`としてお答えできません`,
	`申し訳ありませんが`,
	`我(?:不能|无法)(?:协助|翻译|继续)`,
	`抱歉[，,]`,
	`^\s*\[?(?:REFUSED|ОТКАЗ)\]?\s*$`,
}

var refusalRE = regexp.MustCompile("(?im)" + strings.Join(refusalPatterns, "|"))

var thinkRE = regexp.MustCompile(`(?s)<think>.*?</think>\s*`)

func stripThink(s string) string { return thinkRE.ReplaceAllString(s, "") }

// isRefusal ports refusal_bench: a refusal-blacklist match AND an anomalously
// short output (shorter than max(400 runes, half the source), so a long faithful
// translation that merely contains a refusal-like phrase does not flag).
func isRefusal(out, src string) bool {
	if !refusalRE.MatchString(out) {
		return false
	}
	threshold := utf8.RuneCountInString(src) / 2
	if threshold < 400 {
		threshold = 400
	}
	return utf8.RuneCountInString(out) < threshold
}

// --- CJK-echo (untranslated source returned instead of a translation, D3.4) ---

// cjkEchoThreshold: >15% Han+kana in a non-CJK-target output means the model
// echoed the source (refusal_bench cjk_share > 0.15).
const cjkEchoThreshold = 0.15

func cjkShare(s string) float64 {
	if s == "" {
		return 0
	}
	cjk, total := 0, 0
	for _, r := range s {
		total++
		if unicode.In(r, unicode.Han, unicode.Hiragana, unicode.Katakana) {
			cjk++
		}
	}
	if total == 0 {
		return 0
	}
	return float64(cjk) / float64(total)
}

// isCJKTarget reports whether the translation TARGET is a CJK language, in which
// case CJK in the output is expected and the echo check is skipped. Phase-1
// scope is →ru, so this is normally false.
func isCJKTarget(lang string) bool {
	switch strings.ToLower(strings.TrimSpace(lang)) {
	case "zh", "ja", "ko", "zh-cn", "zh-tw":
		return true
	}
	return false
}

// isRuTarget reports whether the translation TARGET is Russian — the guard for the READABILITY gate
// set (output-sanitizer + cheap style flaggers), whose rules bake in Cyrillic/Russian conventions
// (D39 слой 7, L8-readability-gates-target-blind). They are a no-op for any other target, which would
// otherwise false-flag (a →en final reads as a "Latin insertion"). Mirrors the pair-aware
// coverage/classify seam. Prod zh→ru is unaffected (target=ru).
func isRuTarget(lang string) bool {
	return strings.EqualFold(strings.TrimSpace(lang), "ru")
}

// --- degeneration detector (n-gram loop) — runs BEFORE the length retry (D2.3) ---

// degenerateLoop reports whether the text is a repetition loop: the word-level
// trigram distinctness collapses when a phrase/sentence repeats to fill the
// budget. Pure and deterministic (only len(distinct) is used, never map order).
// Below a floor of words a length cut of genuinely short text is not judged a
// loop. CJK output (no spaces → few "words") is left to the echo check.
func degenerateLoop(text string) bool {
	const n = 3
	const minWords = 30
	const distinctRatio = 0.25 // <25% distinct trigrams ⇒ ≥75% are repeats ⇒ loop
	words := strings.Fields(text)
	if len(words) < minWords {
		return false
	}
	total := len(words) - n + 1
	seen := make(map[string]struct{}, total)
	for i := 0; i+n <= len(words); i++ {
		seen[strings.Join(words[i:i+n], "\x00")] = struct{}{}
	}
	return float64(len(seen))/float64(total) < distinctRatio
}

// --- max_tokens on the attempt axis (D2.3) ---

// maxTokensForAttempt is the PURE output-token budget for a retry attempt:
// attempt 0 = base, each regeneration DOUBLES it (D2.3 remedy for a length cut —
// the previous budget was too small). Purity is load-bearing: the value enters
// request_hash, so resume must reproduce the identical per-attempt budget. Its
// FORMULA is versioned into the snapshot (maxTokensPolicyVersion) so a change is
// a loud --resnapshot, not a silent checkpoint miss on retried chunks (the same
// discipline as estimatorVersion).
func maxTokensForAttempt(base, attempt int) int {
	if attempt <= 0 {
		return base
	}
	if attempt > 20 { // defensive: never shift by a runaway amount (overflow guard)
		attempt = 20
	}
	return base << uint(attempt)
}