#!/usr/bin/env python3 """Ф2.5 memory-eval (INDICATIVE run + harness scaffold) — the deciding-experiment design for the memory-bank bet (research/13 §Вердикт, decisions-log D-пилот, D13.5). QUESTION (owner's central fear + go/no-go on the bank bet): is DETERMINISTIC SELECTIVE glossary injection (our bank) worth it vs just putting the WHOLE glossary in a long context, and does a WRONG injection silently degrade the translation? WMT25 three-mode protocol, RENAMED M0–M3 per D13.4 (the old C0–C3 collided with exp09 §3 CORE configurations): M0 no glossary (baseline — model's base behaviour) M1 SELECTIVE bank injection (only records whose keys fire in the chunk) — OUR bank M2 FULL glossary + sliding summary in the prompt — "just long context" M3 RANDOM/WRONG glossary (dst shuffled) — adversarial arm, directly tests silent degradation Decision rule (pre-registered, exp09 §6): M1≈M2 quality but M1 cheaper → bank justified; M2≫M1 → need fuller context; M3 drops vs M0 on FIDELITY → wrong injection harms → post-check/disposition justified (owner's fear confirmed & mitigated, not just asserted). THREE AXES, measured SEPARATELY (exp09 §6): (a) consistency — decl-aware (declmetric.py, pymorphy3 + Go-style stored-decl) approved-dst rendering across the chunks where src fires; (b) FIDELITY/omission — LLM judge of a DIFFERENT family vs the SOURCE (D13.3 family rule: judge NOT from the draft/translator model's family). This is the axis that makes the pre-registered "M3 drops vs M0 on fidelity" rule TESTABLE — without it the central owner-fear measurement is unverifiable (07-review §4.5). CometKiwi is OPTIONAL and only with the caveat "not validated on ru-literature" (research/11-gap-2) — not wired here. (c) cost — input/output tokens per mode. INJECTION = eval-side MIRROR of backend/internal/pipeline/memory.go (THE SOURCE OF TRUTH), synced D13.5 across the 7 divergences the review flagged (07-review §4.5): 1. sticky depth = stickyDepth(2), union window, reset at chapter boundary (runner.go:534-560); 2. until_ch spoiler gate (memory.go:374-382), not only since_ch; 3. span-containment gated on a SPOILER-VALID suppressor (memory.go:591-617); 4. token budget + F2 eviction in priority order (memory.go:262-370); 5. sticky disposition INHERITANCE — a collision-prone AMBIGUOUS carried by sticky stays AMBIGUOUS, not upgraded to CONFIRMED. The ratified D16.2 fix, now LANDED in Go v3 (memoryMatchVersion memmatch-v3 "sticky-inherit-disp"); the mirror matches the v3 contract and was re-verified against the landed Go by the package-2 self-test; 6. FULL vendored trad2simp table (eval/pilot/trad2simp.txt, hash-synced with the Go embed), not a 12-entry stub; 7. default-ignorables = unicode.Cf + variation selectors FE00..FE0F / E0100..E01EF (memnorm.go isIgnorableForMatch), not one hardcoded VS char; 8. source word-boundary for spaced-phonetic keys (D16.3, memory.go suppressUnboundedPhonetic) — a Latin/Cyrillic key fires only as a WHOLE WORD (rose⊄roseanne), a cross-script neighbour is a valid boundary; KANA/HAN keys are NOT boundary-checked (no word spaces; name+particle recall). Package-2 Task 1a — the mirror previously described this carve-out but did not apply it in select(); now applied, parity-covered by TestSourcePhoneticWordBoundary + the kana carve-out; 9. apostrophe-variant fold ‘ ’ ʼ ＇ → ' on the source side (norm_src) AND the target side (declmetric.normalize_target), symmetric with Go isApostropheVariant (package-2 Task 1b) — critical for the en→ru pilot arm (д'Артаньян / O'Brien). Han-script tests cover ALL planes (Ext A–I + non-decomposing compat), and significantLen counts Unicode LETTERS only (mirrors Go's unicode.IsLetter branch; excludes kana middle-dot U+30FB and combining marks) — closes the adversarial-review deltas #1/#2 (supplementary-plane names). KNOWN LATENT GAPS (package-2 parity review, verified NOT reachable for the zh acceptance book / Ah-Q pilot, so left as-is; documented, not silent): (2) the fixed-range kana/hangul/Han anchor tests in spaced_phonetic_script/_is_han approximate Go's full unicode script tables — they miss exotic ranges (Kana Phonetic-Ext U+31F0.., conjoining Jamo, CJK radicals, iteration marks 々〆) and 84 unassigned F900 code points, but none FLIP spaced_phonetic_script's result for a realistic zh/ja/en key (a real key always carries a true anchor); (4) str.isspace() treats C0 separators U+001C–1F as space where Go does not — cosmetic (both are non-letter boundaries). U+0130 İ (the sole str.lower()≠Go divergence) and the glossaryLineTokens kana marks (ー・) ARE fixed above. On any divergence the Go code wins. (eval/memory_hotpath.py is the PRE-cc57c7b prototype — do NOT reuse.) ECHO CONTROL (D13.5 / 07-review §4.5): the model ECHOES the injected glossary preamble ("ГЛОССАРИЙ … 阿Q → А-кью …") before the translation. Scoring the whole output then counts the echoed dst forms as "rendered" → M-modes inflated, and the ADVERSARIAL M3 harm UNDER-counted (the review traced chunk1-M3 inflated 0.667→1.0). We now (a) DETECT the preamble/source echo (reusing refusal_bench's CJK classifier), (b) STRIP it to the translation body before scoring, and (c) FLAG the datapoint echo_contaminated and EXCLUDE it from the clean headline aggregate (optionally --regen-on-echo re-asks once with a hardened prompt) — never a silent strip. Provenance (eval rule #1): FULL raw outputs saved (not out[:160]); every run stamps model id + UTC date + usage; a NEW output file per run (never overwrite — exp02 lost 66 raw calls that way). Usage: eval/.venv/bin/python eval/pilot/memory_eval.py --dry-run # selection mechanism eval/.venv/bin/python eval/pilot/memory_eval.py --self-test # mirror invariants (no API) eval/.venv/bin/python eval/pilot/memory_eval.py --chapter 6 # indicative M0–M3 on Ah-Q """ from __future__ import annotations import argparse, json, re, sys, time, unicodedata from datetime import datetime, timezone from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) import refusal_bench as rb import declmetric as dm ROOT = Path(__file__).resolve().parent SAMPLES = ROOT.parent / "data" / "samples" OUTDIR = ROOT.parent / "data" / "pilot" OUTDIR.mkdir(parents=True, exist_ok=True) TRAD_TABLE = ROOT / "trad2simp.txt" # vendored from backend/internal/pipeline/data/trad2simp.txt # --- normalization (faithful mirror of memnorm.go) -------------------------------- def _load_trad2simp(path: Path) -> dict[str, str]: """Parse the vendored ' ' table (mirror of parseTradTable). A corrupt line is a loud error — the same fail-loud contract as Go, so a truncated table can't silently re-open the A4 orthography hole. The file has 508 content lines → 500 unique trad keys: 8 chars (東決莊語說談關飛) are listed under two radical families with the SAME simp target — benign consistent duplicates (later-wins, exactly like Go). A duplicate with a DIFFERENT target is a typo → fail loud (mirror of Go's parseTradTable panic), package-2 Task 1d.""" m: dict[str, str] = {} if not path.exists(): raise SystemExit(f"memory_eval: vendored trad2simp table missing: {path}\n" f" cp backend/internal/pipeline/data/trad2simp.txt {path}") for i, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1): s = line.strip() if not s or s.startswith("#"): continue fields = s.split() if len(fields) != 2 or len(fields[0]) != 1 or len(fields[1]) != 1: raise SystemExit(f"memory_eval: trad2simp line {i}: want 2 single-rune fields, got {s!r}") tr, si = fields if tr in m and m[tr] != si: # inconsistent duplicate = typo → loud, like Go raise SystemExit(f"memory_eval: trad2simp line {i}: {tr!r} maps to both {m[tr]!r} and {si!r}") m[tr] = si return m TRAD2SIMP = _load_trad2simp(TRAD_TABLE) def _is_ignorable(c: str) -> bool: """Mirror of Go isIgnorableForMatch: unicode.Cf + variation selectors FE00..FE0F / E0100..E01EF.""" o = ord(c) return unicodedata.category(c) == "Cf" or (0xFE00 <= o <= 0xFE0F) or (0xE0100 <= o <= 0xE01EF) # Typographic apostrophe/single-quote variants that fold to ASCII '\'' (mirror of Go # isApostropheVariant): U+2018/2019 curly single quotes, U+02BC modifier-letter apostrophe, # U+FF07 fullwidth apostrophe. Smart-quote editors turn a source key O'Brien's apostrophe into # U+2019, which NFKC does NOT fold to U+0027 — leaving the key silently un-matchable on en names. _APOSTROPHE_VARIANTS = frozenset("‘’ʼ＇") def _is_letter(c: str) -> bool: return unicodedata.category(c)[0] == "L" # unicode.IsLetter equivalent (L*) def _is_han(c: str) -> bool: """Full-plane Han (mirror of Go unicode.Is(unicode.Han)): Ext A, URO, compat, Ext B–I, compat-supplement. Covers supplementary-plane ideographs (rare surnames/fantasy names a webnovel corpus can hit) that BMP-only ranges miss — adversarial-review delta #2.""" o = ord(c) return (0x3400 <= o <= 0x4DBF or 0x4E00 <= o <= 0x9FFF or 0xF900 <= o <= 0xFAFF or 0x20000 <= o <= 0x2A6DF or 0x2A700 <= o <= 0x2EBEF or 0x2F800 <= o <= 0x2FA1F or 0x30000 <= o <= 0x323AF) def norm_src(s: str) -> str: """Mirror of memnorm.go normalizeSourceKey: NFKC → apostrophe-variant fold → drop ignorables → trad→simp (per rune) → katakana→hiragana → Unicode lower. Applied SYMMETRICALLY to keys and chunk text (the apostrophe fold keeps O'Brien matchable across smart-quote/ASCII forms).""" s = unicodedata.normalize("NFKC", s) out = [] for c in s: if c in _APOSTROPHE_VARIANTS: c = "'" # typographic apostrophe (O'Brien) folds to ASCII so a key matches either form if _is_ignorable(c): continue c = TRAD2SIMP.get(c, c) o = ord(c) if 0x30A1 <= o <= 0x30F6: # katakana → hiragana c = chr(o - 0x60) if c == "İ": # U+0130: Go unicode.ToLower→'i'; Python str.lower→'i̇' (i+U+0307). Match Go out.append("i"); continue # the ONLY rune where .lower() diverges from Go (parity-review #1) out.append(c.lower()) return "".join(out) def significant_len(nk: str) -> int: """Mirror of memnorm.go significantLen: count Unicode LETTERS (category L*). Since Han/kana/ hangul letters are all category L across every plane, this subsumes Go's explicit Han/Hiragana/Katakana/Hangul clause AND excludes the non-letters it also excludes (kana middle-dot U+30FB Po, combining marks Mn) — more faithful than a fixed block range.""" return sum(1 for c in nk if _is_letter(c)) def any_han(nk: str) -> bool: return any(_is_han(c) for c in nk) def min_key_len_for(nk: str) -> int: return 2 if any_han(nk) else 3 # minKeyLenHan / minKeyLenPhonetic def collision_prone(nk: str) -> bool: """Short AND purely phonetic (no Han anchor) → likely to fire inside an unrelated word → injected AMBIGUOUS (memory.go collisionProneKey; threshold = minKeyLenPhonetic=3).""" return (not any_han(nk)) and significant_len(nk) <= 3 # --- source word-boundary for spaced-phonetic keys (D16.3, memory.go suppressUnboundedPhonetic) -- def _is_latin_letter(c: str) -> bool: return _is_letter(c) and "LATIN" in unicodedata.name(c, "") def _is_cyrillic_letter(c: str) -> bool: return _is_letter(c) and "CYRILLIC" in unicodedata.name(c, "") def spaced_phonetic_script(norm_key: str) -> str | None: """Mirror of memory.go spacedPhoneticScript: the word-delimited alphabetic script a normalized key belongs to ('latin'|'cyrillic'), or None when the key carries a Han/kana/ hangul anchor (no word segmentation) or MIXES the two spaced scripts. Digits/punctuation don't set the script but don't disqualify (so "o'brien" is still latin).""" script = None for c in norm_key: o = ord(c) if _is_han(c) or (0x3040 <= o <= 0x30FF) or (0xAC00 <= o <= 0xD7A3): return None # ideographic/kana/hangul anchor is never boundary-checked here if _is_latin_letter(c): if script == "cyrillic": return None # mixed spaced scripts — do not boundary-check script = "latin" elif _is_cyrillic_letter(c): if script == "latin": return None script = "cyrillic" return script def _letter_in_script(c: str, script: str) -> bool: """Mirror of memory.go letterInScript: r is a LETTER of the given spaced script — the same-script letter that breaks a word boundary. A digit/punct/space or a letter of another script is a valid boundary and returns False.""" if script == "latin": return _is_latin_letter(c) if script == "cyrillic": return _is_cyrillic_letter(c) return False def unbounded_phonetic(key: str, start: int, end: int, ntext: str) -> bool: """Mirror of memory.go suppressUnboundedPhonetic's per-occurrence test: True → SUPPRESS. A Latin/Cyrillic key that fired INSIDE a longer same-script word ("rose" in "roseanne"/ "roses") is not a whole-entity match. A cross-script neighbour (a Latin name abutting a Han char in unspaced source) is a valid boundary → keep. Kana/Han keys return script None → never suppressed (Japanese has no word spaces; name+particle recall — see kana_precision).""" script = spaced_phonetic_script(key) if script is None: return False before_bad = start > 0 and _letter_in_script(ntext[start - 1], script) after_bad = end < len(ntext) and _letter_in_script(ntext[end], script) return before_bad or after_bad STICKY_DEPTH = 2 # memory.go stickyDepth # --- disposition (mirror of memory.go dispositionFor + injectionDisposition) ------ CONFIRMED, AMBIGUOUS, REJECT = "confirmed", "ambiguous", "reject" def disposition_for(status: str, via: str, allow_short: bool) -> str: """Exact-match disposition. approved → confirmed; auto/draft → ambiguous; an approved entry matched by a COLLISION-PRONE key is downgraded to ambiguous (A2 surface-collision). via=='sticky' NEVER reaches here in the D16.2 mirror — sticky INHERITS its disposition.""" if via != "sticky" and not allow_short and collision_prone(via): return AMBIGUOUS return CONFIRMED if status == "approved" else AMBIGUOUS # --- glossary (Ah-Q) as store-like rows ------------------------------------------- # Fields mirror store.GlossaryEntry columns the hot path reads. One SPOILER entry (革命党, # since_ch=7) demonstrates the since gate; 尼姑 is status=draft to exercise AMBIGUOUS-by-status. GLOSSARY = [ {"src": "阿Q", "dst": "А-кью", "status": "approved", "sense": "", "since_ch": 0, "until_ch": 0, "allow_short": False, "aliases": [], "type": "name", "lemma_keys": ["а-кью"], "hyphen_pat": r"А[-\s]?кью", "decl_forms": ["А-кью"]}, {"src": "未庄", "dst": "Вэйчжуан", "status": "approved", "sense": "", "since_ch": 0, "until_ch": 0, "allow_short": False, "aliases": [], "type": "place", "lemma_keys": ["вэйчжуан"], "decl_forms": ["Вэйчжуане", "Вэйчжуан", "Вэйчжуана"]}, {"src": "赵太爷", "dst": "почтенный Чжао", "status": "approved", "sense": "", "since_ch": 0, "until_ch": 0, "allow_short": False, "aliases": ["赵老太爷"], "type": "name", "lemma_keys": ["чжао"], "decl_forms": ["почтенного Чжао", "почтенный Чжао", "Чжао"]}, {"src": "王胡", "dst": "Бородатый Ван", "status": "approved", "sense": "", "since_ch": 0, "until_ch": 0, "allow_short": False, "aliases": ["王癞胡", "癞胡"], "type": "name", "lemma_keys": ["ван"], "decl_forms": ["Бородатого Вана", "Бородатый Ван", "Бородатому Вану"]}, {"src": "小D", "dst": "Маленький Дэн", "status": "approved", "sense": "", "since_ch": 0, "until_ch": 0, "allow_short": False, "aliases": [], "type": "name", "lemma_keys": ["дэн"], "decl_forms": ["Маленького Дэна", "Маленький Дэн", "Дэна"]}, {"src": "假洋鬼子", "dst": "Поддельный заморский чёрт", "status": "approved", "sense": "", "since_ch": 0, "until_ch": 0, "allow_short": False, "aliases": [], "type": "nickname", "lemma_keys": ["заморский", "чёрт"], "decl_forms": ["Поддельного заморского чёрта", "заморский чёрт"]}, {"src": "吴妈", "dst": "У-ма", "status": "approved", "sense": "", "since_ch": 0, "until_ch": 0, "allow_short": False, "aliases": [], "type": "name", "lemma_keys": ["у-ма"], "hyphen_pat": r"У[-\s]?ма", "decl_forms": ["У-ма", "У-мы"]}, {"src": "秀才", "dst": "сюцай", "status": "approved", "sense": "", "since_ch": 0, "until_ch": 0, "allow_short": False, "aliases": [], "type": "title", "lemma_keys": ["сюцай"], "decl_forms": ["сюцая", "сюцай", "сюцаю"]}, {"src": "尼姑", "dst": "монашка", "status": "draft", "sense": "", "since_ch": 0, "until_ch": 0, "allow_short": False, "aliases": [], "type": "term", "lemma_keys": ["монашка"], "decl_forms": ["монашку", "монашка", "монашки"]}, # SPOILER demo: only valid from chapter 7 (revolution). Injected earlier → hard reject. {"src": "革命党", "dst": "революционеры", "status": "approved", "sense": "", "since_ch": 7, "until_ch": 0, "allow_short": False, "aliases": [], "type": "term", "lemma_keys": ["революционер"], "decl_forms": ["революционеров", "революционеры"]}, ] def eligible_keys(entry: dict) -> list[str]: """Normalized source surfaces (src + aliases) passing the per-lang min-key ban (A3). A record with empty dst is NOT matchable (renders nothing) — mirror memory.go:167-173.""" if not str(entry.get("dst", "")).strip(): return [] out = [] for raw in [entry["src"]] + entry.get("aliases", []): nk = norm_src(raw) if nk and (significant_len(nk) >= min_key_len_for(nk) or entry.get("allow_short")): out.append(nk) return out def spoiler_blocked(entry: dict, chapter: int) -> bool: """Mirror of memory.go spoilerBlocked: since_ch/until_ch window (0 = unbounded).""" if entry.get("since_ch", 0) and chapter < entry["since_ch"]: return True if entry.get("until_ch", 0) and chapter > entry["until_ch"]: return True return False def glossary_line_tokens(entry: dict) -> int: """Mirror of memory.go glossaryLineTokens: cjk + other/3 over 'src → dst' (floor-free). cjk bucket = full-plane Han + kana block + hangul (matches Go unicode.In(Han,Hiragana, Katakana,Hangul); supplementary-plane Han now weighted ×1, not ÷3).""" cjk = other = 0 # Go glossaryLineTokens buckets by unicode.In(Han,Hiragana,Katakana,Hangul), which keys on SCRIPT # (not Script_Extensions). Mirror the marks that BMP Han/kana ranges miss (D19.5(д) token-bucket # sync; each verified against Go's tables via a scratch unicode.In probe): 々 U+3005 and 〇 U+3007 # are Script=Han → CJK (weight 1); halfwidth-katakana LETTERS U+FF66–FF9D are Script=Katakana → CJK. # 〆 U+3006 is NOT Han in Go (→ other) so it is deliberately NOT added. kana_nonletter EXCLUDES the # Script=Common marks Go counts as `other`: ゠・ー (U+30A0/30FB/30FC), combining U+3099/309A, and # the SPACING voiced marks ゛゜ U+309B/309C (previously mis-counted as CJK — the parity gap this fixes). han_marks = {0x3005, 0x3007} kana_nonletter = {0x30A0, 0x30FB, 0x30FC, 0x3099, 0x309A, 0x309B, 0x309C} for c in f"{entry['src']} → {entry.get('dst','')}": o = ord(c) if (_is_han(c) or o in han_marks or (0x3040 <= o <= 0x30FF and o not in kana_nonletter) or (0xFF66 <= o <= 0xFF9D) or (0xAC00 <= o <= 0xD7A3)): cjk += 1 elif c.isspace(): pass else: other += 1 return cjk + other // 3 def _key_occurrences(ntext: str, key: str) -> list[tuple[int, int]]: """All [start,end) occurrences of key in ntext (rune indices == str indices in Python).""" spans, i = [], ntext.find(key) while i != -1: spans.append((i, i + len(key))) i = ntext.find(key, i + 1) return spans def select(chunk: str, chapter: int, sticky_prev: dict[str, str], budget_tokens: int = 0) -> dict: """Faithful mirror of MemoryBank.Select for ONE chunk: span-match → spoiler-gated containment → disposition → sticky (inherited disposition) → priority token budget. sticky_prev: {id -> inherited disposition} from the prior ≤stickyDepth chunks of this chapter. Returns injected/rejected/evicted lists + active {id->disposition} for the next chunk's sticky window.""" ntext = norm_src(chunk) # 1. all key occurrences with owning entry indices occ = [] # (start, end, key, entry_idx) for ei, e in enumerate(GLOSSARY): for k in eligible_keys(e): for (s, t) in _key_occurrences(ntext, k): occ.append((s, t, k, ei)) occ.sort(key=lambda m: (m[0], m[1], m[2])) # 1b. source word-boundary for spaced-phonetic keys (D16.3, memory.go suppressUnboundedPhonetic), # BEFORE containment — a Latin/Cyrillic key fired inside a longer same-script word # ("rose" ⊂ "roseanne") is not a whole-entity match. Kana/Han keys are NOT boundary-checked. occ = [m for m in occ if not unbounded_phonetic(m[2], m[0], m[1], ntext)] # 2. longest-match containment, suppressor gated on spoiler-VALIDITY (memory.go:591-617) def valid_suppressor(m): return not spoiler_blocked(GLOSSARY[m[3]], chapter) kept = [] for i, m in enumerate(occ): contained = False for j, o in enumerate(occ): if i == j: continue if o[0] <= m[0] and o[1] >= m[1] and (o[1]-o[0]) > (m[1]-m[0]) and valid_suppressor(o): contained = True break if not contained: kept.append(m) # 3. surviving occ → entries, longest firing key per entry matched_via: dict[int, str] = {} for (s, t, k, ei) in kept: if ei not in matched_via or len(k) > len(matched_via[ei]): matched_via[ei] = k injected_hits: dict[int, dict] = {} # entry_idx -> picked rejected = [] active: dict[str, str] = {} # id -> disposition (exact matches, for next sticky) for ei in range(len(GLOSSARY)): via = matched_via.get(ei) if via is None: continue e = GLOSSARY[ei] eid = _eid(e) if spoiler_blocked(e, chapter): rejected.append({**e, "_via": via, "_disp": REJECT}) continue disp = disposition_for(e["status"], via, e.get("allow_short", False)) injected_hits[ei] = {**e, "_via": via, "_disp": disp} active[eid] = disp # exact-matched ids feed the next chunk's sticky (NOT sticky-carried) # 4. sticky scene-inertia: carry prior-chunk exact matches not re-matched here, INHERITING # their disposition (D16.2), unless the spoiler window blocks them (spoiler beats sticky). hit_ids = {_eid(p) for p in injected_hits.values()} for ei, e in enumerate(GLOSSARY): eid = _eid(e) if eid not in sticky_prev or eid in hit_ids or ei in injected_hits: continue if spoiler_blocked(e, chapter): continue injected_hits[ei] = {**e, "_via": "sticky", "_disp": sticky_prev[eid]} # 5. priority order + F2 token budget/eviction (memory.go:342-368) picked = list(injected_hits.values()) picked.sort(key=_priority_rank) evicted = [] if budget_tokens > 0: used, cut = 0, len(picked) for i, p in enumerate(picked): used += glossary_line_tokens(p) if used > budget_tokens: cut = i break picked, evicted = picked[:cut], picked[cut:] return {"injected": picked, "rejected": rejected, "evicted": evicted, "active": active} def _eid(e: dict) -> str: return f'{e["src"]}\x1f{e.get("sense","")}\x1f{e.get("since_ch",0)}\x1f{e.get("until_ch",0)}' def _priority_rank(p: dict) -> int: rank = 0 if p["_disp"] != CONFIRMED: rank |= 1 << 2 if p["_via"] == "sticky": rank |= 1 << 1 if p["status"] != "approved": rank |= 1 << 0 return rank # --- prompt assembly per mode ----------------------------------------------------- SYSTEM = ("Ты профессиональный литературный переводчик. Переведи фрагмент китайского " "произведения на русский язык. Сохрани все реплики и детали без пропусков; стиль — " "живой литературный русский. Выведи ТОЛЬКО перевод — НЕ повторяй глоссарий и инструкции, " "начни сразу с текста перевода.") SYSTEM_HARDENED = SYSTEM + (" ВАЖНО: не выводи блок ГЛОССАРИЙ, строки вида «иероглиф → перевод», " "заголовки или пояснения — только сам художественный перевод.") # Byte-synced with Go glossaryBlockHeader (memory.go) — the ⟨проверить⟩ clause was missing from # the mirror (package-2 Task 1d), so the injected M1 prompt now matches what the hot path emits. GHEAD = ("ГЛОССАРИЙ (используй эти утверждённые переводы имён и терминов последовательно; " "строки с пометкой ⟨проверить⟩ — неподтверждённые кандидаты):") def gloss_block(entries: list[dict], full: bool = False) -> str: if not entries: return "" lines = [] for e in entries: line = f"{e['src']} → {e['dst']}" if not full and e.get("_disp") == AMBIGUOUS: line += " ⟨проверить⟩" lines.append(line) return GHEAD + "\n" + "\n".join(lines) def build_prompt(chunk: str, mode: str, sel: dict, summary: str, shuffled: dict) -> str: if mode == "M0": return "ФРАГМЕНТ:\n" + chunk if mode == "M1": # selective bank blk = gloss_block(sel["injected"]) return (blk + "\n\n" if blk else "") + "ФРАГМЕНТ:\n" + chunk if mode == "M2": # full glossary + sliding summary (long context) blk = gloss_block([{**e, "_disp": CONFIRMED} for e in GLOSSARY if not e.get("since_ch") and not e.get("until_ch") and str(e.get("dst", "")).strip()], full=True) pre = (f"КРАТКОЕ СОДЕРЖАНИЕ ПРЕДЫДУЩЕГО:\n{summary}\n\n" if summary else "") return pre + blk + "\n\nФРАГМЕНТ:\n" + chunk if mode == "M3": # random/wrong glossary (adversarial): fired entries with SHUFFLED dst wrong = [{**e, "dst": shuffled[e["src"]], "_disp": CONFIRMED} for e in sel["injected"] if e["_via"] != "sticky"] blk = gloss_block(wrong, full=True) return (blk + "\n\n" if blk else "") + "ФРАГМЕНТ:\n" + chunk raise ValueError(mode) # --- echo control (D13.5 / 07-review §4.5) ---------------------------------------- GLOSS_ECHO_MARKERS = ("ГЛОССАРИЙ", "глоссарий", "КАНОНИЧЕСКИЕ ПЕРЕВОД", "используй эти утверждённые", "использовать последовательно", "используйте эти", "КРАТКОЕ СОДЕРЖАНИЕ", "⟨проверить", "проверить⟩") ARROW_RE = re.compile(r".+\s*(→|->|=>)\s*.+") def _is_gloss_echo_line(s: str) -> bool: """A single output line is a glossary/instruction ECHO if it is either (a) a HEADER echo — the line STARTS with a gloss marker (case-insensitive) — or (b) a 'src → dst' gloss ENTRY echo — an arrow with CJK on the LEFT (the source term), a SHORT line, and NO sentence-ending punctuation. Used for BOTH the leading preamble and the embedded/trailing echo scan (package-2 Task 1c). Both rules are TIGHTENED vs the first cut (parity-review #5): anchoring the marker to the line start no longer strips prose that merely contains «глоссарий» mid-sentence, and the entry rule's period/length guard no longer strips prose like «Путь (道) → его судьба была предрешена.». A prose source-echo (untranslated CJK with NO arrow) is deliberately NOT matched here — cjk_share catches it. KNOWN BOUNDARY (D19.5(д), documented not fixed): line-start anchoring still FALSE-POSITIVES on a body line that GENUINELY begins with a marker word — e.g. prose «Глоссарий рода был утерян в пожаре.» starting its own line is flagged as a header echo and dropped from the scored body. This is a deliberate CONSERVATIVE choice, not a silent one: the datapoint carries a visible glossary_preamble_echo / embedded_glossary_echo reason (and is marked echo_contaminated), so an over-strip is auditable and never inflates the clean headline unseen. A tighter fix (require a trailing ':' or an adjacent arrow-line) is deferred — over-flagging a rare sentence-initial «Глоссарий» is safe; letting a real echo glossary through is not.""" s = s.strip() if not s: return False sl = s.lower() if any(sl.startswith(m.lower()) for m in GLOSS_ECHO_MARKERS): return True m = ARROW_RE.match(s) if m: left = re.split(r"→|->|=>", s, 1)[0] if rb.CJK_RE.search(left) and len(s) <= 80 and not re.search(r"[.!?…]", s): return True return False def detect_echo(raw: str) -> tuple[str, list[str]]: """Strip glossary/instruction-echo to the translation body AND report contamination reasons. Two scopes (package-2 Task 1c — the old detector only caught the LEADING preamble, so a trailing/embedded echo glossary «…текст…\\nГЛОССАРИЙ:\\n阿Q → А-кью» silently inflated consistency): (1) a LEADING preamble region (markers / arrow-gloss / blanks) is skipped; (2) any REMAINING line that is a gloss-echo line (a marker, or a 'src → dst' arrow line with CJK, ANYWHERE in the body — tail or embedded) is dropped from the scored body and flagged. Also flags source-echo (CJK leaked into the body, the DeepSeek-style untranslated echo).""" lines = raw.split("\n") # (1) leading preamble region body_start, saw_pre = 0, False for i, ln in enumerate(lines): s = ln.strip() if not s: if saw_pre: body_start = i + 1 continue if _is_gloss_echo_line(s): saw_pre, body_start = True, i + 1 continue break # first prose line → body begins # (2) embedded/trailing echo lines anywhere in the remaining body → drop from the scored body kept, n_embedded = [], 0 for ln in lines[body_start:]: if _is_gloss_echo_line(ln): n_embedded += 1 continue kept.append(ln) body = "\n".join(kept).strip() reasons = [] if saw_pre: reasons.append("glossary_preamble_echo") if n_embedded: reasons.append(f"embedded_glossary_echo({n_embedded})") if not body: # whole output was preamble/echo → keep raw but flag hard reasons.append("empty_after_strip") body = raw.strip() ces = rb.cjk_share(body) if ces > 0.10: reasons.append(f"source_echo(cjk={ces:.0%})") return body, reasons # --- fidelity judge (cross-family, D13.3) ----------------------------------------- FAMILY = {"deepseek": "deepseek", "grok": "xai", "gemini": "google", "gpt": "openai", "glm": "zhipu", "kimi": "moonshot", "mistral": "mistral", "qwen": "alibaba"} def family_of(model: str) -> str: for k, v in FAMILY.items(): if k in model: return v return model FIDELITY_SYSTEM = ( "Ты — строгий редактор-оценщик ВЕРНОСТИ художественного перевода. Тебе дают ИСХОДНЫЙ " "фрагмент (китайский) и его ПЕРЕВОД на русский. Оцени ТОЛЬКО верность исходнику (не стиль): " "искажения смысла (mistranslation), пропуски (omission), отсебятину (addition), особенно " "НЕВЕРНО переданные имена/термины. Верни СТРОГО JSON без пояснений: " '{"fidelity": <0-100>, "n_mistranslation": , "n_omission": , "n_addition": , ' '"wrong_names": [<строки>], "notes": "<кратко>"}. fidelity=100 — идеально верно; ниже — за ' "каждое искажение/пропуск. Имена, переданные не как в источнике, — это mistranslation.") def fidelity_judge(source: str, translation: str, judge_model: str, translator_model: str, providers: dict) -> dict: """Cross-family fidelity judge vs the SOURCE (D13.3: judge family != translator family).""" if family_of(judge_model) == family_of(translator_model): raise ValueError(f"fidelity judge {judge_model} shares family with translator " f"{translator_model} ({family_of(judge_model)}) — violates D13.3") user = (f"ИСХОДНЫЙ ФРАГМЕНТ (китайский):\n{source}\n\n" f"ПЕРЕВОД НА РУССКИЙ (оцени его верность):\n{translation}\n\nВерни JSON.") p = providers[judge_model] txt, err, usage = _retry_call({"name": judge_model, "model": p["model"], **p}, FIDELITY_SYSTEM, user, timeout=200) if err or not txt: return {"error": err or "empty", "usage": usage or {}} m = re.search(r"\{.*\}", txt, re.S) if not m: return {"error": "no-json", "raw": txt[:300], "usage": usage or {}} try: out = json.loads(m.group(0)) except json.JSONDecodeError: return {"error": "bad-json", "raw": txt[:300], "usage": usage or {}} out["usage"] = usage or {} out["judge_model"] = judge_model return out # --- providers (translator + judge) ----------------------------------------------- PROVIDERS = { "deepseek-v4-flash": {"base_url": "https://api.deepseek.com/v1", "api_key_env": "DEEPSEEK_API_KEY", "model": "deepseek-v4-flash", "max_tokens": 8000}, "grok-4.20-0309-non-reasoning": {"base_url": "https://api.x.ai/v1", "api_key_env": "XAI_API_KEY", "model": "grok-4.20-0309-non-reasoning", "max_tokens": 8000, "temperature": 0.3}, "gemini-2.5-flash": {"base_url": "https://generativelanguage.googleapis.com/v1beta/openai", "api_key_env": "GEMINI_API_KEY", "model": "gemini-2.5-flash", "max_tokens": 8000, "extra_body": {"extra_body": {"google": {"thinking_config": {"thinking_budget": 0}}}}}, "gpt-5-mini": {"base_url": "https://api.openai.com/v1", "api_key_env": "OPENAI_API_KEY", "model": "gpt-5-mini", "reasoning_params": True, "max_tokens": 8000, "extra_body": {"reasoning_effort": "minimal"}}, } def _retry_call(p: dict, system: str, user: str, timeout: int = 200, attempts: int = 3): """rb.call_provider with backoff on TRANSIENT errors (503 overload / 429 rate / 5xx / transport) so a provider spike does not punch a hole in the fidelity axis or a translation. A content refusal / config error is NOT retried (it is a real signal).""" last = (None, "unknown", {}) for i in range(attempts): t, err, usage = rb.call_provider(p, system, user, timeout=timeout) if not err or not re.match(r"(http\|(503|429|500|502|504))|transport\|", err or ""): return t, err, usage last = (t, err, usage) time.sleep(2 * (i + 1)) return last def translate(prompt: str, model: str, hardened: bool = False) -> tuple[str, dict]: p = PROVIDERS[model] sysmsg = SYSTEM_HARDENED if hardened else SYSTEM t, err, usage = _retry_call({"name": model, **p}, sysmsg, prompt, timeout=200) return (t or ""), (usage or {}) # --- chunking (reuse the coverage harness's chunker semantics) -------------------- def chunk_text(text: str, target=1200) -> list[str]: paras = [p.strip() for p in re.split(r"\n\s*\n", text) if p.strip()] chunks, buf = [], [] for p in paras: if buf and len("".join(buf)) + len(p) > target: chunks.append("\n\n".join(buf)); buf = [] buf.append(p) if buf: chunks.append("\n\n".join(buf)) return chunks # --- self-test: mirror invariants without any API call ---------------------------- def self_test() -> int: fails = [] def check(cond, msg): if not cond: fails.append(msg) # since gate: 革命党 rejected at ch6, injected at ch7 s6 = select("革命党来了，未庄震动。", 6, {}) check(any(r["src"] == "革命党" for r in s6["rejected"]), "since-gate: 革命党 not rejected at ch6") s7 = select("革命党来了，未庄震动。", 7, {}) check(any(p["src"] == "革命党" for p in s7["injected"]), "since-gate: 革命党 not injected at ch7") # until gate (synthetic): a term valid only until ch3 must be rejected at ch10 global GLOSSARY saved = GLOSSARY GLOSSARY = saved + [{"src": "旧党", "dst": "старая партия", "status": "approved", "sense": "", "since_ch": 0, "until_ch": 3, "allow_short": False, "aliases": [], "type": "term", "lemma_keys": ["партия"], "decl_forms": ["старой партии", "старая партия"]}] check(spoiler_blocked(GLOSSARY[-1], 10), "until-gate: 旧党 not blocked at ch10") check(not spoiler_blocked(GLOSSARY[-1], 2), "until-gate: 旧党 wrongly blocked at ch2") GLOSSARY = saved # collision downgrade: a short phonetic key (allow_short off) → AMBIGUOUS even if approved check(disposition_for("approved", "ai", False) == AMBIGUOUS, "collision: 'ai' not downgraded") check(disposition_for("approved", "мерос", False) == CONFIRMED, "collision: 5-char key wrongly downgraded") check(disposition_for("approved", "リン", False) == AMBIGUOUS, "collision: 2-kana key not downgraded") # supplementary-plane Han is Han-anchored (floor 2, CONFIRMED), not phonetic — review delta #2 check(any_han("\U00020000") and not collision_prone("\U00020000\U00020001"), "supplementary-plane Han not treated as Han") # significant_len counts letters only (kana middle-dot U+30FB excluded) — review delta #1 check(significant_len("リ・ン") == 2, "significant_len counts kana middle-dot U+30FB") # sticky depth-2 + disposition inheritance: a draft term inherits AMBIGUOUS across a # pronominal chunk, then falls out after depth 2. win = [] def union(w): # mirror unionSticky out = {} for d in w: out.update(d) return out c0 = select("这尼姑是谁。", 6, union(win)); win = (win + [c0["active"]])[-STICKY_DEPTH:] check(any(p["src"] == "尼姑" and p["_disp"] == AMBIGUOUS for p in c0["injected"]), "sticky: 尼姑(draft) not AMBIGUOUS on exact match") c1 = select("他走了。", 6, union(win)); win = (win + [c1["active"]])[-STICKY_DEPTH:] sk = [p for p in c1["injected"] if p["src"] == "尼姑"] check(sk and sk[0]["_via"] == "sticky" and sk[0]["_disp"] == AMBIGUOUS, "sticky: 尼姑 not carried with INHERITED ambiguous disposition (D16.2)") # depth-2 window: a term exact-matched at c0 is carried across c1 AND c2 (the window holds the # last stickyDepth active sets), then falls out at c3 — mirror of runner.go unionSticky/trim. c2 = select("又走了。", 6, union(win)); win = (win + [c2["active"]])[-STICKY_DEPTH:] check(any(p["src"] == "尼姑" and p["_via"] == "sticky" for p in c2["injected"]), "sticky: 尼姑 should still be carried at depth 2 (c2)") c3 = select("再走了。", 6, union(win)); win = (win + [c3["active"]])[-STICKY_DEPTH:] check(not any(p["src"] == "尼姑" for p in c3["injected"]), "sticky: 尼姑 should fall out after the depth-2 window (c3)") # spoiler-gated containment (synthetic): a spoiler-blocked longer key must NOT suppress a # valid shorter nested name. GLOSSARY = saved + [ {"src": "林", "dst": "Линь", "status": "approved", "sense": "", "since_ch": 0, "until_ch": 0, "allow_short": True, "aliases": [], "type": "name", "lemma_keys": ["линь"], "decl_forms": ["Линь"]}, {"src": "林动的父亲", "dst": "отец Линь Дуна", "status": "approved", "sense": "", "since_ch": 0, "until_ch": 3, "allow_short": False, "aliases": [], "type": "name", "lemma_keys": ["отец"], "decl_forms": ["отца Линь Дуна"]}] sc = select("林动的父亲来了。", 10, {}) check(any(p["src"] == "林" for p in sc["injected"]), "containment: 林 suppressed by spoiler-blocked longer key at ch10") GLOSSARY = saved # source phonetic word-boundary (D16.3, mirror of Go TestSourcePhoneticWordBoundary): a # Latin/Cyrillic key fires only as a WHOLE WORD, a cross-script neighbour is a valid boundary. GLOSSARY = saved + [ {"src": "rose", "dst": "Роза", "status": "approved", "sense": "", "since_ch": 0, "until_ch": 0, "allow_short": False, "aliases": [], "type": "name", "lemma_keys": ["роза"], "decl_forms": ["Роза"]}] check(len(select("roseanne walked in.", 1, {})["injected"]) == 0, "src-boundary: latin key fired inside 'roseanne'") check(len(select("the roses bloomed.", 1, {})["injected"]) == 0, "src-boundary: latin key fired inside 'roses'") rz = select("a rose bloomed.", 1, {})["injected"] check(any(p["src"] == "rose" and p["_disp"] == CONFIRMED for p in rz), "src-boundary: standalone latin key must fire CONFIRMED") check(any(p["src"] == "rose" for p in select("rose.", 1, {})["injected"]), "src-boundary: latin key at string edges must fire") check(any(p["src"] == "rose" for p in select("我叫rose。", 1, {})["injected"]), "src-boundary: latin key abutting a cross-script (Han) neighbour must fire") GLOSSARY = saved # kana carve-out (mirror of Go TestKanaNotSourceBoundaryChecked): a long kana key still fires # between kana neighbours (name+particle と…が), NOT source-boundary-suppressed. GLOSSARY = saved + [ {"src": "ながいなまえ", "dst": "Длинное имя", "status": "approved", "sense": "", "since_ch": 0, "until_ch": 0, "allow_short": False, "aliases": [], "type": "name", "lemma_keys": ["длинное"], "decl_forms": ["Длинное имя"]}] check(any(p["src"] == "ながいなまえ" and p["_disp"] == CONFIRMED for p in select("私とながいなまえが会った。", 1, {})["injected"]), "src-boundary: kana key must NOT be boundary-suppressed (name+particle recall)") GLOSSARY = saved # apostrophe-variant fold (package-2 Task 1b, mirror of Go TestApostropheFold): typographic ≡ # ASCII on the source side, and an ASCII-apostrophe key matches a typographic-apostrophe chunk. check(norm_src("O’Brien") == norm_src("O'Brien"), "apostrophe: source fold not applied") GLOSSARY = saved + [ {"src": "o'brien", "dst": "О'Брайен", "status": "approved", "sense": "", "since_ch": 0, "until_ch": 0, "allow_short": False, "aliases": [], "type": "name", "lemma_keys": ["о'брайен"], "decl_forms": ["О'Брайен"]}] check(any(p["src"] == "o'brien" for p in select("mr o’brien arrived.", 1, {})["injected"]), "apostrophe: typographic apostrophe in chunk must match ASCII-apostrophe key") GLOSSARY = saved # token budget eviction: tiny budget keeps only the highest-priority record sb = select("阿Q和未庄和王胡和秀才。", 6, {}, budget_tokens=3) check(len(sb["evicted"]) > 0 and len(sb["injected"]) >= 1, "budget: no eviction under tiny budget") # echo detector (package-2 Task 1c): leading preamble, embedded/trailing glossary, and clean. b1, r1 = detect_echo("ГЛОССАРИЙ (используй эти утверждённые):\n阿Q → А-кью\n\nЖил-был А-кью.") check(b1 == "Жил-был А-кью." and any("preamble" in r for r in r1), "echo: leading preamble not stripped/flagged") b2, r2 = detect_echo("Жил-был А-кью.\n\nГЛОССАРИЙ:\n阿Q → А-кью") check(b2 == "Жил-был А-кью." and any("embedded_glossary_echo" in r for r in r2), "echo: trailing glossary not stripped/flagged") b3, r3 = detect_echo("Начало главы.\n阿Q → А-кью\nКонец главы.") check(b3 == "Начало главы.\nКонец главы." and any("embedded_glossary_echo" in r for r in r3), "echo: embedded arrow-gloss line not stripped/flagged") b4, r4 = detect_echo("Жил-был А-кью в деревне Вэйчжуан.") check(b4 == "Жил-был А-кью в деревне Вэйчжуан." and not r4, "echo: clean translation wrongly flagged") # echo detector must NOT strip legit prose (parity-review #5): a buried marker word, or an # arrow+CJK line that is actually a sentence, are kept. b5, r5 = detect_echo("В конце книги он нашёл глоссарий терминов.") check(b5 == "В конце книги он нашёл глоссарий терминов." and not r5, "echo: FP — buried «глоссарий» wrongly stripped") b6, r6 = detect_echo("Путь (道) → его судьба была предрешена.") check(b6 == "Путь (道) → его судьба была предрешена." and "glossary" not in " ".join(r6), "echo: FP — prose with arrow+CJK wrongly stripped as gloss") # İ→lower Go-parity (parity-review #1): norm_src must fold İ to 'i' (Python .lower() → 'i̇') check(norm_src("İ") == "i" and dm.normalize_target("İ") == "i", "İ: not folded to Go's lowercase 'i'") # trad2simp byte-parity with the Go embed (package-2 Task 1d): the vendored table must stay # byte-identical to backend/internal/pipeline/data/trad2simp.txt (memoryNormVersion hashes its # bytes) — a drift silently diverges the normalizers. Skipped if the Go source isn't checked out. go_embed = ROOT.parent.parent / "backend" / "internal" / "pipeline" / "data" / "trad2simp.txt" if go_embed.exists(): import hashlib h_eval = hashlib.md5(TRAD_TABLE.read_bytes()).hexdigest() h_go = hashlib.md5(go_embed.read_bytes()).hexdigest() check(h_eval == h_go, f"trad2simp: vendored table diverged from Go embed ({h_eval} != {h_go})") print("SELF-TEST:", "OK" if not fails else "FAIL") for f in fails: print(" ✗", f) return 1 if fails else 0 def main(): ap = argparse.ArgumentParser() ap.add_argument("--src", default=str(SAMPLES / "zh" / "luxun-ah-q-ch5-9.txt")) ap.add_argument("--chapter", type=int, default=6, help="chapter number for the spoiler gate " "(single-chapter indicative: no per-chunk chapter-reset; the deciding run " "feeds real multi-chapter books, where the sticky window resets per chapter)") ap.add_argument("--modes", default="M0,M1,M2,M3") ap.add_argument("--model", default="grok-4.20-0309-non-reasoning", help="translator/draft model") ap.add_argument("--judge", default="gemini-2.5-flash", help="cross-family fidelity judge (D13.3)") ap.add_argument("--no-fidelity", action="store_true", help="skip the LLM fidelity axis") ap.add_argument("--budget", type=int, default=0, help="glossary token budget (0 = unbounded)") ap.add_argument("--regen-on-echo", action="store_true", help="re-ask once with a hardened prompt on echo") ap.add_argument("--max-chunks", type=int, default=5) ap.add_argument("--dry-run", action="store_true") ap.add_argument("--self-test", action="store_true") ap.add_argument("--out", default="") args = ap.parse_args() if args.self_test: sys.exit(self_test()) stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") out_path = Path(args.out) if args.out else OUTDIR / f"memory_eval_M_{stamp}.json" raw_dir = OUTDIR / f"raw_{stamp}" text = Path(args.src).read_text(encoding="utf-8") chunks = chunk_text(text)[:args.max_chunks] modes = args.modes.split(",") # deterministic wrong-dst shuffle for M3 (rotate dst among present names/places) names = [e for e in GLOSSARY if e["type"] in ("name", "place")] rot = {names[i]["src"]: names[(i + 1) % len(names)]["dst"] for i in range(len(names))} shuffled = {e["src"]: rot.get(e["src"], "НЕВЕРНО") for e in GLOSSARY} print(f"src={Path(args.src).name} chapter={args.chapter} chunks={len(chunks)} model={args.model} " f"judge={args.judge} modes={modes} budget={args.budget}", file=sys.stderr) if not args.dry_run and not args.no_fidelity: if family_of(args.judge) == family_of(args.model): sys.exit(f"D13.3: judge {args.judge} shares family with translator {args.model}") raw_dir.mkdir(parents=True, exist_ok=True) rows, sticky_win = [], [] # sticky_win: list of {id->disp}, capped at STICKY_DEPTH def union_sticky(): out = {} for d in sticky_win: out.update(d) return out for ci, ch in enumerate(chunks): sel = select(ch, args.chapter, union_sticky(), args.budget) present = {e["src"]: e for e in sel["injected"] if e["_via"] != "sticky"} row = {"chunk": ci, "src_chars": len(ch), "injected": [{"src": e["src"], "disp": e["_disp"], "via": e["_via"]} for e in sel["injected"]], "rejected_spoiler": [{"src": r["src"]} for r in sel["rejected"]], "evicted": [{"src": e["src"]} for e in sel["evicted"]], "present_for_consistency": list(present)} if not args.dry_run: row["by_mode"] = {} for mode in modes: prompt = build_prompt(ch, mode, sel, summary="", shuffled=shuffled) raw, usage = translate(prompt, args.model) body, echo = detect_echo(raw) regen = None if echo and args.regen_on_echo: raw2, usage2 = translate(prompt, args.model, hardened=True) body2, echo2 = detect_echo(raw2) regen = {"raw": raw2, "echo": echo2, "usage": usage2} if not echo2: # clean on retry → use it body, echo, usage, raw = body2, echo2, usage2, raw2 cons = dm.consistency(body, {s: {"dst": e["dst"], "lemma_keys": e.get("lemma_keys", []), "hyphen_pat": e.get("hyphen_pat"), "decl_forms": e.get("decl_forms", [])} for s, e in present.items()}) if present else None fid = None if not args.no_fidelity: fid = fidelity_judge(ch, body, args.judge, args.model, PROVIDERS) # persist FULL raw output for post-hoc audit (never a preview) (raw_dir / f"chunk{ci}_{mode}.txt").write_text( f"# model={args.model} chunk={ci} mode={mode} chapter={args.chapter} echo={echo}\n" f"# usage={usage}\n\n{raw}", encoding="utf-8") row["by_mode"][mode] = { "echo_contaminated": echo or None, "consistency_pymorphy": cons["pymorphy"]["score"] if cons else None, "consistency_stored_decl": cons["stored_decl"]["score"] if cons else None, "missed_pymorphy": cons["pymorphy"]["missed"] if cons else [], "missed_stored_decl": cons["stored_decl"]["missed"] if cons else [], "fidelity": fid, "in_tok": usage.get("prompt_tokens"), "out_tok": usage.get("completion_tokens"), "out_full": raw, "body_scored": body, "regen": regen} fscore = (fid or {}).get("fidelity") if isinstance(fid, dict) else None print(f" chunk{ci} {mode}: cons(pym)={row['by_mode'][mode]['consistency_pymorphy']} " f"cons(decl)={row['by_mode'][mode]['consistency_stored_decl']} " f"fidelity={fscore} echo={echo or '-'} " f"in={usage.get('prompt_tokens')} out={usage.get('completion_tokens')}", file=sys.stderr) sticky_win = (sticky_win + [sel["active"]])[-STICKY_DEPTH:] rows.append(row) meta = {"run_utc": stamp, "src": args.src, "chapter": args.chapter, "modes": modes, "translator_model": args.model, "judge_model": None if args.no_fidelity else args.judge, "budget_tokens": args.budget, "regen_on_echo": args.regen_on_echo, "trad2simp_entries": len(TRAD2SIMP), "sticky_depth": STICKY_DEPTH, "note": "M0-M3 = terminology modes (exp09 §6); mirror synced with memory.go (Go v3 " "memmatch-v3): D16.2 sticky-inherit-disp, D16.3 source phonetic word-boundary, " "apostrophe fold (package-2 Tasks 1a/1b) — all parity-covered by --self-test."} out_path.write_text(json.dumps({"meta": meta, "rows": rows}, ensure_ascii=False, indent=2)) print(f"\nwrote {out_path}", file=sys.stderr) if not args.dry_run: print(f"raw outputs: {raw_dir}", file=sys.stderr) _print_aggregate(rows, modes) if args.dry_run: for r in rows: print(f"chunk{r['chunk']} inj={[(e['src'], e['disp'], e['via']) for e in r['injected']]} " f"spoiler_rej={[e['src'] for e in r['rejected_spoiler']]} evicted={[e['src'] for e in r['evicted']]}") def _print_aggregate(rows, modes): """Headline aggregate with clean (echo-excluded) vs all points, per mode.""" print("\n=== AGGREGATE (mean over chunks; clean = echo-contaminated points excluded) ===", file=sys.stderr) for mode in modes: pts = [r["by_mode"][mode] for r in rows if "by_mode" in r] clean = [p for p in pts if not p["echo_contaminated"]] def mean(seq, key, src=pts): vals = [p[key] for p in src if isinstance(p.get(key), (int, float))] return round(sum(vals) / len(vals), 3) if vals else None def fmean(src): vals = [p["fidelity"]["fidelity"] for p in src if isinstance(p.get("fidelity"), dict) and isinstance(p["fidelity"].get("fidelity"), (int, float))] return round(sum(vals) / len(vals), 1) if vals else None n_echo = sum(1 for p in pts if p["echo_contaminated"]) print(f" {mode}: cons(decl) all={mean(pts,'consistency_stored_decl')} clean={mean(clean,'consistency_stored_decl',clean)} " f"| fidelity all={fmean(pts)} clean={fmean(clean)} | echo_pts={n_echo}/{len(pts)}", file=sys.stderr) if __name__ == "__main__": main()