Upload folder using huggingface_hub

app/PROVENANCE.json

@@ -0,0 +1,40 @@
+{
+  "service": "lain-parser",
+  "version": "1.1.0",
+  "frozen": "2026-10-02",
+  "artifact": {
+    "name": "e5-s2",
+    "source": "artifacts/e5-s2 (trained cell runs/parser-t4/e5-s2)",
+    "model_sha256": "d928668e5ed3c8ea4ac69e42f8d11c90f82120e3307dffaf3a02eb801c077e54",
+    "config_sha256": "fa76e5b7d0091d266e337b57e6bb74f703581136f8d740b9c361488417e5cbd0",
+    "evaluation": "docs/E-SERIES.md — real-final-v1 e5-s2 fp32 138/196; this int8 artifact 137/196 (69.9%) on CPU; arm +6.5 pp over the R38 baseline"
+  },
+  "snapshot": {
+    "model": "AniFileBERT DAPT-512-long",
+    "source": "artifacts/anifilebert-dapt512-long",
+    "weights_sha256": "509125db5cc4bee9f3949b44e1c17f86b89406c8efee6a7376c39de8b3db7a0d"
+  },
+  "vendored_app": {
+    "app/fp/parser2.py": {
+      "source": "fp/parser2.py",
+      "sha256": "89abf39364c34d1e159d56553f1044cce06040d23898c6b8c5fa849b4dae7efb"
+    },
+    "app/fp/corpus2.py": {
+      "source": "fp/corpus2.py",
+      "sha256": "a1953d116fa81f0a3103d25d5a20ba7216adc7405aab337888165f7b0a636893"
+    },
+    "app/fp/parser3_model.py": {
+      "source": "fp/parser3_model.py",
+      "sha256": "be0ef67cf753a730b7fa86538d36fe50cfddea53c71b70ac7a29dee1cbaf39ee"
+    },
+    "app/fp/normalize.py": {
+      "source": "fp/normalize.py",
+      "sha256": "7e5d096f9f36f2d3d289bf3d75b6d680bbfa17339431c154a00dc7fe8268ee01"
+    },
+    "app/fp/charclass.py": {
+      "source": "fp/charclass.py",
+      "sha256": "f1bae0df50d83d44bf4df4b2df013a604aa704ae81db70f49c88f758653e7644"
+    }
+  },
+  "claim": "development evidence; assistant-produced labels; see docs/E-SERIES.md"
+}

app/fp/__init__.py

@@ -0,0 +1,6 @@
+"""Vendored subset of the filename-parser package (see ../PROVENANCE.json).
+
+Only the inference path ships here: character encoder, BIO/interval head,
+contract resolver and their shared corpus constants. The training modules are
+not vendored on purpose, so the service cannot drift into training behaviour.
+"""

app/fp/charclass.py

@@ -0,0 +1,92 @@
+"""Unicode class features for the character encoder (E2).
+
+A character's identity embedding says nothing about a code point the model has
+never seen; its Unicode general category and script do. Every character gets a
+(category, script) pair, and the parser adds one learned embedding for each to
+the identity embedding, so an unseen separator or bracket still arrives as
+"opening punctuation" or "dash".
+
+Class 0 is reserved for positions that are not characters ([CLS], [SEP],
+padding). The table is built from this interpreter's `unicodedata` and exported
+into the Go artifact as ranges, so both runtimes use the same assignment even
+where Unicode versions differ.
+"""
+
+import bisect
+import unicodedata
+
+import numpy as np
+
+CATEGORIES = ("Lu", "Ll", "Lt", "Lm", "Lo", "Mn", "Mc", "Me", "Nd", "Nl", "No",
+              "Pc", "Pd", "Ps", "Pe", "Pi", "Pf", "Po", "Sm", "Sc", "Sk", "So",
+              "Zs", "Zl", "Zp", "Cc", "Cf", "Cs", "Co", "Cn")
+# Block ranges, not the Unicode Script property: `unicodedata` does not expose
+# scripts, and blocks are what separate the signals a filename cares about
+# (fullwidth vs ASCII punctuation, kana vs Han). Unlisted code points are "other".
+SCRIPT_BLOCKS = (
+    ("latin", ((0x0000, 0x024F), (0x1E00, 0x1EFF), (0x2C60, 0x2C7F), (0xA720, 0xA7FF),
+               (0xAB30, 0xAB6F))),
+    ("ipa-modifier", ((0x0250, 0x036F),)),
+    ("greek", ((0x0370, 0x03FF), (0x1F00, 0x1FFF))),
+    ("cyrillic", ((0x0400, 0x052F), (0x1C80, 0x1C8F), (0x2DE0, 0x2DFF), (0xA640, 0xA69F))),
+    ("armenian-hebrew", ((0x0530, 0x05FF),)),
+    ("arabic", ((0x0600, 0x06FF), (0x0750, 0x077F), (0x08A0, 0x08FF), (0xFB50, 0xFDFF),
+                (0xFE70, 0xFEFF))),
+    ("indic", ((0x0900, 0x0DFF),)),
+    ("southeast-asian", ((0x0E00, 0x0FFF), (0x1000, 0x109F), (0x1780, 0x17FF))),
+    ("hangul", ((0x1100, 0x11FF), (0x3130, 0x318F), (0xA960, 0xA97F), (0xAC00, 0xD7FF))),
+    ("general-punctuation", ((0x2000, 0x206F),)),
+    ("symbols", ((0x2070, 0x2BFF),)),
+    ("cjk-symbols", ((0x3000, 0x303F), (0x3200, 0x33FF), (0xFE10, 0xFE1F), (0xFE30, 0xFE4F))),
+    ("hiragana", ((0x3040, 0x309F),)),
+    ("katakana", ((0x30A0, 0x30FF), (0x31F0, 0x31FF), (0xFF66, 0xFF9F))),
+    ("han", ((0x2E80, 0x2FDF), (0x3400, 0x4DBF), (0x4E00, 0x9FFF), (0xF900, 0xFAFF),
+             (0x20000, 0x3FFFF))),
+    ("fullwidth", ((0xFF00, 0xFF65), (0xFFE0, 0xFFEF))),
+    ("emoji", ((0x1F000, 0x1FAFF),)),
+)
+SCRIPTS = tuple(name for name, _ in SCRIPT_BLOCKS) + ("other",)
+N_CATEGORIES, N_SCRIPTS = len(CATEGORIES) + 1, len(SCRIPTS) + 1
+LIMIT = 0x110000
+# Categories that carry structure rather than content; identity dropout hits
+# them harder (see Parser.character_embeddings).
+STRUCTURAL = frozenset(i + 1 for i, name in enumerate(CATEGORIES) if name[0] in "PSZ")
+
+_TABLES = None
+
+
+def tables():
+    """(category ids, script ids) for every code point, as uint8 arrays."""
+    global _TABLES
+    if _TABLES is None:
+        category = np.empty(LIMIT, dtype=np.uint8)
+        index = {name: i + 1 for i, name in enumerate(CATEGORIES)}
+        for cp in range(LIMIT):
+            category[cp] = index[unicodedata.category(chr(cp))]
+        script = np.full(LIMIT, SCRIPTS.index("other") + 1, dtype=np.uint8)
+        for name, ranges in SCRIPT_BLOCKS:
+            for a, b in ranges:
+                script[a:b + 1] = SCRIPTS.index(name) + 1
+        _TABLES = category, script
+    return _TABLES
+
+
+def classes(char):
+    category, script = tables()
+    return int(category[ord(char)]), int(script[ord(char)])
+
+
+def ranges():
+    """Run-length [start, end, category, script] rows covering 0..0x10FFFF."""
+    category, script = tables()
+    combined = category.astype(np.uint16) << 8 | script
+    change = np.flatnonzero(np.diff(combined)) + 1
+    starts = np.concatenate(([0], change))
+    ends = np.concatenate((change - 1, [LIMIT - 1]))
+    return [[int(a), int(b), int(category[a]), int(script[a])] for a, b in zip(starts, ends)]
+
+
+def lookup(table, cp):
+    """Reference implementation of the Go range lookup (for parity tests)."""
+    position = bisect.bisect_right([row[0] for row in table], cp) - 1
+    return table[position][2], table[position][3]

app/fp/corpus2.py

@@ -0,0 +1,777 @@
+"""Contract v2.1 synthetic corpus where *layout* is a locked experimental axis.
+
+The pilot (v1) trained on three layouts and one per harder split, so a held-out
+layout scored zero and the result could not be attributed. This corpus makes
+layout families first-class, named and disjoint across splits: arm A trains on
+the narrow set, arm B on the broad set, and both are scored on identical rows
+whose layouts neither arm ever saw. That is the difference between measuring
+memorization and measuring generalization over formats.
+
+Three things make the gold here trustworthy rather than convenient:
+
+*  Every record is derived by `fp.normalize.record`, the same contract code that
+   scores predictions. The generator never writes an output value by hand, so a
+   case the contract cannot represent fails here instead of becoming a silent
+   disagreement between gold and prediction.
+*  Season values come from span *text* through `normalize.season_value`. There is
+   no value head and no annotator number, so "the model learned the numbers"
+   cannot be confused with "the code parsed the numbers".
+*  Every row is strict: each atom carries a kind, including DISCARD for junk, so
+   unexplained debris is impossible by construction and `check_debris` proves it.
+
+Layout parts are explicit so a typo cannot become silent text: a `str` is a
+literal that must contain no word characters, `("a", key)` is an atom (a span),
+and `("j", text)` is junk claimed as DISCARD.
+"""
+
+import argparse
+import hashlib
+import json
+import random
+import re
+from collections import Counter
+from pathlib import Path
+
+from fp.normalize import record as contract_record
+
+ROOT = Path("data/parser-v2")
+SEED = 20260915
+MAX_CHARS = 120  # the character encoders cap at 128 positions; nothing truncates
+# Tolerated exactly as `fp.gold` tolerates them: a real outcome to be reproduced,
+# not a generator bug. Anything else means the row cannot be represented.
+TOLERATED = {"season-unspecified", "dimensions-no-class"}
+
+# The span kinds this experiment's head may emit, and therefore the kinds the
+# corpus must demonstrate. Kept here, not in the trainer, so a kind that exists
+# in one place only is a test failure rather than an unmeasured blind spot.
+KINDS = ("TITLE", "EPISODE_TITLE", "GROUP", "SEASON", "EPISODE", "RANGE", "LIST", "YEAR",
+         "RESOLUTION", "DIMENSIONS", "DISCARD")
+
+LANGS = ("en", "pt", "es", "ru", "ja", "zh")
+WORDS = {
+    "en": ["Quiet", "Harbor", "Copper", "Moon", "Garden", "River", "Paper", "Forest",
+           "Station", "Lantern", "Winter", "Signal"],
+    "pt": ["Porto", "Lua", "Jardim", "Rio", "Papel", "Bosque", "Pedra", "Silencio",
+           "Estacao", "Farol", "Inverno", "Sinal"],
+    "es": ["Puerto", "Luna", "Jardin", "Rio", "Papel", "Bosque", "Piedra", "Silencio",
+           "Estacion", "Faro", "Invierno", "Senal"],
+    "ru": ["Тихий", "Порт", "Луна", "Сад", "Река", "Бумага", "Лес", "Камень",
+           "Вокзал", "Маяк", "Зима", "Сигнал"],
+    "ja": ["静かな", "港", "月", "庭", "川", "紙", "森", "石",
+           "駅", "灯台", "冬", "信号"],
+    "zh": ["安静", "港湾", "月亮", "花园", "河流", "纸张", "森林", "石头",
+           "车站", "灯塔", "冬天", "信号"],
+}
+# Authored second-script forms, not real transliterations: a controlled bilingual
+# vocabulary for testing alias handling. The spans are the point, not the
+# linguistics, and nobody should read these rows as language data.
+ALIAS = {
+    "ru": ["Tihiy", "Port", "Luna", "Sad", "Reka", "Bumaga", "Les", "Kamen",
+           "Vokzal", "Mayak", "Zima", "Signal"],
+    "ja": ["Shizuka", "Minato", "Tsuki", "Niwa", "Kawa", "Kami", "Mori", "Ishi",
+           "Eki", "Toudai", "Fuyu", "Shingou"],
+    "zh": ["Anjing", "Gangwan", "Yueliang", "Huayuan", "Heliu", "Zhiwang", "Senlin",
+           "Shitou", "Chezhan", "Dengta", "Dongtian", "Xinhao"],
+}
+SUBTITLE = {
+    "en": ["Rain Memory", "The Last Promise", "A Quiet Harbor", "Paper Lanterns",
+           "North Wind", "Signals at Dusk"],
+    "pt": ["Memoria da Chuva", "A Ultima Promessa", "Noite Serena", "Vento Norte",
+           "Farol Apagado", "Cidade Velha"],
+    "es": ["Recuerdo de Lluvia", "La Ultima Promesa", "Noche Serena", "Viento Norte",
+           "Farol Apagado", "Ciudad Vieja"],
+    "ru": ["Память дождя", "Последнее обещание", "Тихая ночь", "Северный ветер",
+           "Погасший маяк", "Старый город"],
+    "ja": ["雨の記憶", "最後の約束", "静かな夜", "北の風", "消えた灯台", "古い街"],
+    "zh": ["雨的记忆", "最后的约定", "安静的夜", "北风", "消失的灯塔", "旧城"],
+}
+SEASON_MARKS = {
+    "en": ["Season {n}", "{ord} Season", "Season {n}", "s{n}"],
+    "pt": ["{ord} Temporada", "Temporada {n}", "temporada {n}"],
+    "es": ["{ord} Temporada", "Temporada {n}", "temp {n}"],
+    "ru": ["{ord} Сезон", "Сезон {n}", "сезон {n}"],
+    "ja": ["第{n}期", "シーズン{n}", "{ord}シーズン"],
+    "zh": ["第{ord}季", "第{n}季"],
+}
+ORDINALS = {
+    "en": ["First", "Second", "Third", "Fourth"],
+    "pt": ["Primeira", "Segunda", "Terceira", "Quarta"],
+    "es": ["Primera", "Segunda", "Tercera", "Cuarta"],
+    "ru": ["Первый", "Второй", "Третий", "Четвёртый"],
+    "ja": ["一", "二", "三", "四"],
+    "zh": ["一", "二", "三", "四"],
+}
+FINAL_MARK = {"en": "Final Season", "pt": "Temporada Final", "es": "Temporada Final",
+              "ru": "Последний Сезон", "ja": "最終シーズン", "zh": "最终季"}
+GROUPS = ["Nebula-raws", "Lantern Works", "Aoi-Subs", "Meridian", "Kite-Team",
+          "Harbor-Encode", "Sable-raws"]
+JUNK = ["REMASTERED", "BluRay", "x264", "MULTI-AUDIO", "DVDRIP", "Hi10P", "WEBRip",
+        "AC3", "PROPER", "字幕组", "双语", "Batch", "Complete", "CD1", "Vol02"]
+EXTENSIONS = [".mkv", ".mp4", ".avi", ".webm"]
+RESOLUTIONS = ["480p", "720p", "1080p", "2160p", "720P", "1080P"]
+DIMENSIONS = ["1280x720", "1920x1080", "640x480", "720x576", "3840x2160"]
+
+
+def digest(path):
+    return hashlib.sha256(Path(path).read_bytes()).hexdigest()
+
+
+def layout_inventory():
+    """(name, tag, parts). `narrow` mirrors the pilot's three shapes; `broad` is
+    the format-coverage hypothesis; `heldout-*` trains nothing in this experiment.
+    """
+    return [
+        ("n1", "narrow", ["[", ("a", "g"), "] ", ("a", "t"), " - ", ("a", "e"),
+                          " [", ("a", "R"), "]"]),
+        ("n2", "narrow", [("a", "t"), " [", ("a", "s"), "] ", ("a", "e"), " ",
+                          ("a", "R"), "-", ("a", "d1")]),
+        ("n3", "narrow", [("a", "t"), "_", ("a", "s"), "_", ("a", "e"), "_", ("a", "R"),
+                          "_[", ("a", "d1"), "]"]),
+        ("b1", "broad", [("a", "g"), " ", ("a", "t"), " - ", ("a", "e"), " [", ("a", "R"),
+                         "][", ("a", "d1"), "]"]),
+        ("b2", "broad", [("a", "t"), " (", ("a", "y"), ") - ", ("a", "e"), " [", ("a", "R"),
+                         "] [", ("a", "d1"), "]"]),
+        ("b3", "broad", [("a", "t"), ".", ("a", "y"), ".", ("a", "R"), ".", ("a", "d1"),
+                         "-", ("a", "d2"), "-", ("a", "g")]),
+        ("b4", "broad", [("a", "t"), " ", ("a", "x"), " [", ("a", "R"), "] [", ("a", "g"), "]"]),
+        ("b5", "broad", [("a", "g"), " - ", ("a", "t"), " - ", ("a", "e"), " - ", ("a", "x")]),
+        ("b6", "broad", [("a", "t"), " ", ("a", "a"), " [", ("a", "s"), "] [", ("a", "R"), "]"]),
+        ("b7", "broad", [("a", "t"), " ", ("a", "a"), " ", ("a", "s"), " ", ("a", "e")]),
+        ("b8", "broad", [("a", "t"), " ", ("a", "sc"), ("a", "ec"), " [", ("a", "R"), "]"]),
+        ("b9", "broad", ["[", ("a", "d1"), "] ", ("a", "t"), " [", ("a", "x"), "] ",
+                         ("a", "e"), " (", ("a", "R"), ")"]),
+        ("b10", "broad", [("a", "t"), " - ", ("a", "e"), " ", ("a", "d1"), " ", ("a", "R"),
+                          " [", ("a", "g"), "]"]),
+        ("b11", "broad", ["[", ("a", "g"), "] ", ("a", "t"), "(", ("a", "y"), ") ",
+                          ("a", "s"), " ", ("a", "l")]),
+        ("b12", "broad", [("a", "t"), " ", ("a", "s"), " ", ("a", "r"), " ", ("a", "R"),
+                          " ", ("a", "d1")]),
+        ("b13", "broad", [("a", "t"), " [", ("a", "D"), "] ", ("a", "e"), " [", ("a", "d1"),
+                          "] [", ("a", "g"), "]"]),
+        ("b14", "broad", [("a", "t"), " ", ("a", "fm"), " ", ("a", "e"), " [", ("a", "R"), "]"]),
+        ("b15", "broad", [("a", "g"), "/", ("a", "t"), "/", ("a", "td"), " ", ("a", "e"),
+                          " [", ("a", "R"), "]"]),
+        ("b16", "broad", [("a", "t"), " ", ("a", "s"), " ", ("a", "ej"), " [", ("a", "R"), "]"]),
+        # held-out format families
+        ("h1", "heldout-layout-a", [("j", "TV"), "/", ("a", "t"), "/", ("a", "s"), "/", ("a", "td"), " ",
+                                    ("a", "d1"), " ", ("a", "e")]),
+        ("h2", "heldout-layout-a", [("a", "t"), "(", ("a", "y"), ")", ("a", "sc"),
+                                     "(", ("a", "R"), ")(", ("a", "d1"), ")"]),
+        ("h3", "heldout-layout-a", [("a", "t"), " - ", ("a", "e3"), ("a", "vt"), " - ",
+                                    ("a", "x"), " [", ("a", "R"), "]"]),
+        ("h4", "heldout-layout-a", ["[", ("a", "g"), "] ", ("a", "t"), " ", ("a", "ej"),
+                                    " [", ("a", "D"), "]"]),
+        ("h5", "heldout-layout-b", ["【", ("a", "g"), "】", ("a", "t"), "【", ("a", "ej"),
+                                    "】【", ("a", "R"), "】"]),
+        ("h6", "heldout-layout-b", [("a", "t"), " | ", ("a", "a"), " | ", ("a", "e"), " | ",
+                                    ("a", "d1")]),
+        ("h7", "heldout-layout-b", ["(", ("a", "d1"), ") ", ("a", "t"), " ", ("a", "ej"),
+                                    " ", ("a", "x"), " ", ("a", "R")]),
+        ("h8", "heldout-layout-b", [("a", "t"), "_", ("a", "s"), "_", ("a", "e"), "-",
+                                    ("a", "x"), "_", ("a", "d1"), "_[", ("a", "g"), "]"]),
+        ("h9", "heldout-layout-b", [("j", "Series"), "/", ("a", "t"), "/", ("a", "epp"), "-", ("a", "td"),
+                                    " ", ("a", "d1")]),
+        ("h10", "heldout-layout-b", [("a", "t"), " ", ("a", "a"), " ", ("a", "tc"), " ",
+                                      ("a", "l"), " [", ("a", "R"), "]"]),
+        # held-out lexical markers in familiar shapes
+        ("k1", "heldout-lexical", [("a", "g"), " ", ("a", "t"), " ", ("a", "epw"), " [",
+                                   ("a", "R"), "]"]),
+        ("k2", "heldout-lexical", [("a", "t"), " ", ("a", "chp"), ("j", " of 24"), " [", ("a", "R"), "]"]),
+        ("k3", "heldout-lexical", [("a", "t"), " ", ("a", "sw"), " ", ("a", "epw")]),
+        # held-out values: 4-digit episodes, wide ranges, big season numbers
+        ("v1", "heldout-value", [("a", "t"), " - ", ("a", "e4"), " [", ("a", "R"), "] [",
+                                 ("a", "d1"), "]"]),
+        ("v2", "heldout-value", [("a", "t"), " ", ("a", "sbig"), " ", ("a", "rwide"), " [",
+                                 ("a", "R"), "]"]),
+        ("v3", "heldout-value", [("a", "t"), " ", ("a", "sc"), ("a", "e4"), " [", ("a", "R"),
+                                 "] [", ("a", "g"), "]"]),
+        # no work title: the row must abstain
+        ("z1", "narrow-neg", [("a", "d1"), ".", ("a", "d2"), ".", ("a", "R")]),
+        ("z2", "broad-neg", ["[", ("a", "d1"), "] ", ("a", "tk"), " ", ("a", "d2")]),
+        ("z3", "broad-neg", [("a", "tk"), "/", ("a", "s"), "/", ("a", "d1")]),
+        ("z4", "heldout-neg", [("a", "d1"), "-", ("a", "d2"), "-", ("a", "D"), " [",
+                               ("a", "g"), "]"]),
+    ]
+
+
+LAYOUTS = {name: (tag, parts) for name, tag, parts in layout_inventory()}
+# Which layouts an arm may see, and which corpus splits those layouts produced.
+ARM_TAGS = {"A-narrow": ["narrow", "narrow-neg"],
+            "B-broad": ["narrow", "broad", "narrow-neg", "broad-neg"]}
+ARM_SPLITS = {"A-narrow": ["train-narrow"], "B-broad": ["train-narrow", "train-extra"]}
+NEEDS_ALIAS = {"b6", "b7", "h6", "h10"}
+LITERAL_OK = re.compile(r"^[\W_]*$", re.UNICODE)
+
+
+def make_atoms(rng, lang, title, alias):
+    """The row's typed fragments. Probes reuse this dict and change exactly one key."""
+    season_number = rng.randrange(1, 5)
+    episode = rng.randrange(1, 60)
+    return {
+        "t": ("TITLE", title),
+        "a": ("TITLE", alias or title),
+        "td": ("DISCARD", title),  # a path component repeating the title: no new evidence
+        "x": ("EPISODE_TITLE", rng.choice(SUBTITLE[lang])),
+        "s": ("SEASON", rng.choice(SEASON_MARKS[lang]).format(
+            n=season_number, p=rng.randrange(1, 3), ord=rng.choice(ORDINALS[lang]))),
+        "fm": ("SEASON", FINAL_MARK[lang]),
+        "sc": ("SEASON", f"S{season_number:02d}"),
+        "tc": ("SEASON", f"T{season_number:02d}"),
+        "sbig": ("SEASON", f"Season {rng.randrange(10, 24)}"),
+        "sw": ("SEASON", rng.choice(["Season number {n}", "Season {n}"]).format(n=season_number)),
+        "n2": ("EPISODE", f"{episode:02d}"),
+        "n3": ("EPISODE", f"{episode}"),
+        "epp": ("EPISODE", f"EP{episode:02d}"),
+        "e": ("EPISODE", f"{episode:02d}"),
+        "ec": ("EPISODE", f"E{episode:02d}"),
+        "e3": ("EPISODE", f"{episode:03d}"),
+        "e4": ("EPISODE", f"{episode * 100:04d}"),
+        "ej": ("EPISODE", f"第{episode:02d}话"),
+        "epw": ("EPISODE", f"Episode {episode}"),
+        "chp": ("EPISODE", f"Chapter {episode:02d}"),
+        "y": ("YEAR", str(rng.randrange(1985, 2026))),
+        "l": ("LIST", ",".join(f"{x:02d}" for x in sorted(rng.sample(range(1, 24), 3)))),
+        "r": ("RANGE", f"{rng.randrange(1, 12):02d}-{rng.randrange(13, 24):02d}"),
+        "rwide": ("RANGE", f"{rng.randrange(1, 5):02d} ~ {rng.randrange(20, 26):02d}"),
+        "R": ("RESOLUTION", rng.choice(RESOLUTIONS)),
+        "D": ("DIMENSIONS", rng.choice(DIMENSIONS)),
+        "g": ("GROUP", rng.choice(GROUPS)),
+        "d1": ("DISCARD", rng.choice(JUNK)),
+        "d2": ("DISCARD", rng.choice(JUNK)),
+        "tk": ("DISCARD", f"{rng.randrange(16 ** 7, 16 ** 8):x}"),
+        "vt": ("DISCARD", f"v{rng.randrange(2, 4)}"),
+    }
+
+
+def render(atoms, parts, extension):
+    query, spans = "", []
+    for part in parts:
+        if isinstance(part, str):
+            if not LITERAL_OK.match(part):
+                raise AssertionError(f"literal carries word characters: {part!r}")
+            query += part
+            continue
+        kind, text = atoms[part[1]] if part[0] == "a" else ("DISCARD", part[1])
+        spans.append({"kind": kind, "text": text, "start": len(query),
+                      "end": len(query) + len(text)})
+        query += text
+    return query + extension, spans
+
+
+def finalize(query, spans, layout, lang, family, used_queries, quarantine):
+    """Contract-validate one rendered row, or quarantine it with the reason."""
+    if len(query) > MAX_CHARS or query in used_queries:
+        quarantine["collision-or-too-long"] += 1
+        return None
+    derived, issues = contract_record(query, spans, check_debris=True)
+    blocking = [i for i in issues if i not in TOLERATED]
+    if blocking or (derived is None and not layout.startswith("z")):
+        quarantine[",".join(blocking) or "no-record"] += 1
+        return None
+    used_queries.add(query)
+    return {"query": query, "spans": spans, "record": derived,
+            "tolerated": sorted(set(issues)), "layout": layout, "language": lang,
+            "family": family, "script": "cjk" if lang in ("ja", "zh") else "latin-cyrillic"}
+
+
+JOINERS = (" ", " - ", "_", " | ", "-", "(", ")", " ")
+PATH_ROOTS = ("Archive", "Collection", "Library", "Shelf", "Videos")
+# A gap the boundary view may rewrite: separators only. Brackets are excluded
+# because a span's `[` is structure, not spacing - removing it manufactures text
+# like `Title1080p]`, which no real filename contains and which the contract has no
+# way to reject, since unclaimed punctuation is debris-free by definition.
+SEPARATOR_GAP = re.compile(r"[\s._-]+")
+TIGHT_JOINERS = ("", ".", "|", "-", "_")
+# Round 5's marker-role vocabulary. Deliberately disjoint from every marker word
+# that appears in the held-out lexical split: `Episode`, `Chapter` and
+# `Season number` must stay unseen, or the lexical axis would measure memory of
+# the probe's own words instead of transfer. `test_marker_vocabulary_is_disjoint
+# _from_the_heldout_lexical_split` enforces that, because a comment cannot.
+EPISODE_MARKERS = ("Ep", "Eps", "Episodio", "Эпизод", "Bölüm")
+EPISODE_FORMS = ("{m} {n}", "{m}{n}", "{n} {m}")
+
+
+def _single_number(text):
+    """The span's number, only if the text contains exactly one digit group.
+
+    The contract's EPISODE rule requires one number, and a two-number fragment
+    ("05-07") is a RANGE the augmentation must not rewrite into a fake episode.
+    """
+    numbers = re.findall(r"\d+", text)
+    return numbers[0] if len(numbers) == 1 else None
+
+
+def augment_markers(row, rng, attempts=4):
+    """Attach a word marker to numeric evidence, keeping the derived record fixed.
+
+    Training layouts express episodes as bare numbers (`05`, `E05`, `第05话`) and
+    seasons through `SEASON_MARKS`, so nothing in the corpus ever tells the model
+    that a *word* belongs to a number's span. That missing supervision is the
+    round-4 explanation for the lexical collapse: faced with `Season 4 Episode 21`
+    the model has no reason to prefer EPISODE over EPISODE_TITLE.
+
+    Like the value and path views, gold is re-derived by the contract and the view
+    is refused unless the record is byte-identical: the marker must be extra text
+    inside an already-typed span, never a re-labelled value.
+    """
+    if row["record"] is None:
+        return None
+    ordered = sorted(row["spans"], key=lambda span: span["start"])
+    eligible = []
+    for index, span in enumerate(ordered):
+        if span["kind"] not in ("EPISODE", "SEASON"):
+            continue
+        text = row["query"][span["start"]:span["end"]]
+        number = _single_number(text)
+        if number is None or len(number) > 4:
+            continue
+        eligible.append((index, span, text, number))
+    if not eligible:
+        return None
+    for _ in range(attempts):
+        chosen = [item for item in eligible if rng.random() < 0.7] or [rng.choice(eligible)]
+        replacements = {}
+        for index, span, text, number in chosen:
+            if span["kind"] == "SEASON":
+                # Re-render with another mark from the same language's table: the
+                # number and its value are unchanged, only which word carries it.
+                marks = [mark for mark in SEASON_MARKS[row["language"]]
+                         if mark.format(n=number, ord=number, p=number) != text]
+                if not marks:
+                    continue
+                value = int(number)
+                replacements[index] = rng.choice(marks).format(
+                    n=value, p=min(value, 3), ord=ORDINALS[row["language"]][min(value, 4) - 1])
+                continue
+            marker = rng.choice(EPISODE_MARKERS)
+            replacements[index] = rng.choice(EPISODE_FORMS).format(m=marker, n=number)
+        if not replacements:
+            continue
+        query, spans = _replace_span_text(row, replacements)
+        if len(query) > MAX_CHARS:
+            continue
+        derived, issues = contract_record(query, spans, check_debris=True)
+        if derived != row["record"] or [i for i in issues if i not in TOLERATED]:
+            continue
+        return {**row, "query": query, "spans": spans, "record": derived,
+                "tolerated": sorted(set(issues)), "augmented": True,
+                "augmented_axes": sorted(set(row.get("augmented_axes", [])) | {"marker"})}
+    return None
+
+
+def augment_boundary(row, rng, attempts=6):
+    """Tighten the separator between two adjacent atoms, brackets untouched.
+
+    Every layout in the corpus separates atoms with visible whitespace, so the
+    model can treat "there is always a gap" as a free signal; the held-out layouts
+    use `.`, `|`, `_` and bare adjacency instead. Round 3's error analysis put the
+    residual failure on boundary cuts rather than vocabulary, so this view makes the
+    cut itself the task while keeping the atom types and the record fixed.
+
+    Only gaps made purely of separator characters are eligible, and the replacement
+    is another separator, never the empty-and-bracketless deletion of structure:
+    deleting `" ["` would yield `Title1080p]` - text no real filename contains. The
+    contract cannot catch that (unclaimed punctuation is debris-free by design), so
+    the surface form is constrained here instead, and gold equality is still checked.
+    """
+    if row["record"] is None:
+        return None
+    ordered = sorted(row["spans"], key=lambda span: span["start"])
+    gaps = []
+    for index in range(len(ordered) - 1):
+        gap = row["query"][ordered[index]["end"]:ordered[index + 1]["start"]]
+        if SEPARATOR_GAP.fullmatch(gap):
+            gaps.append((index, gap))
+    if not gaps:
+        return None
+    for _ in range(attempts):
+        index, gap = rng.choice(gaps)
+        left, right = ordered[index], ordered[index + 1]
+        if left["kind"] == right["kind"]:
+            continue
+        replacement = rng.choice(TIGHT_JOINERS)
+        if replacement == gap:
+            continue
+        start = left["end"]
+        query = row["query"][:start] + replacement + row["query"][start + len(gap):]
+        shift = len(replacement) - len(gap)
+        spans = [dict(span) if position <= index else
+                 {**span, "start": span["start"] + shift, "end": span["end"] + shift}
+                 for position, span in enumerate(ordered)]
+        if len(query) > MAX_CHARS:
+            continue
+        derived, issues = contract_record(query, spans, check_debris=True)
+        if derived != row["record"] or [i for i in issues if i not in TOLERATED]:
+            continue
+        return {**row, "query": query, "spans": spans, "record": derived,
+                "tolerated": sorted(set(issues)), "augmented": True,
+                "augmented_axes": sorted(set(row.get("augmented_axes", [])) | {"boundary"})}
+    return None
+
+
+def augment(row, rng, attempts=4):
+    """The same atoms in a different arrangement, gold re-derived by the contract.
+
+    Round 1 measured the opposite of the obvious hypothesis: 19 hand-written
+    layout families fit their own distribution to 83.6 % and scored 0.4 % on
+    unseen arrangements, *worse* than 3 layouts. Adding formats teaches the model
+    to look up a format. Breaking position inside every training row instead makes
+    the lookup impossible, so the only signal left is local: digits, brackets,
+    known group names, resolution tokens.
+
+    Identity fragments keep their relative order. `record` joins title fragments
+    in file order, so a blind shuffle would change the gold title and the
+    augmentation would no longer be the same row in a different costume.
+
+    Returns None when a shuffle lands on text the contract cannot represent -
+    two adjacent numeric spans, say - rather than shipping a row whose gold is a
+    guess.
+    """
+    original = [(span["kind"], span["text"]) for span in
+                sorted(row["spans"], key=lambda s: s["start"])]
+    extension = row["query"][row["query"].rindex("."):]
+    for _ in range(attempts):
+        parts = list(original)
+        rng.shuffle(parts)
+        slots = [i for i, part in enumerate(parts) if part[0] in ("TITLE", "EPISODE_TITLE")]
+        identity = [part for part in original if part[0] in ("TITLE", "EPISODE_TITLE")]
+        for slot, part in zip(slots, identity):
+            parts[slot] = part
+        query, spans = "", []
+        for index, (kind, text) in enumerate(parts):
+            if index:
+                query += rng.choice(JOINERS)
+            spans.append({"kind": kind, "text": text, "start": len(query),
+                          "end": len(query) + len(text)})
+            query += text
+        query += extension
+        if len(query) > MAX_CHARS:
+            continue
+        derived, issues = contract_record(query, spans, check_debris=True)
+        if [i for i in issues if i not in TOLERATED] or derived != row["record"]:
+            continue
+        return {**row, "query": query, "spans": spans, "record": derived,
+                "tolerated": sorted(set(issues)), "augmented": True}
+    return None
+
+
+def _replace_span_text(row, replacements):
+    """Rebuild a row after replacing span text, retaining every literal byte.
+
+    `replacements` is keyed by the span's index in start order. Rebuilding rather
+    than editing offsets in place keeps span alignment auditable and preserves the
+    parent layout's punctuation exactly.
+    """
+    old = sorted(row["spans"], key=lambda span: span["start"])
+    query, spans, cursor = "", [], 0
+    for index, span in enumerate(old):
+        query += row["query"][cursor:span["start"]]
+        text = replacements.get(index, row["query"][span["start"]:span["end"]])
+        start = len(query)
+        query += text
+        spans.append({"kind": span["kind"], "text": text,
+                      "start": start, "end": len(query)})
+        cursor = span["end"]
+    query += row["query"][cursor:]
+    return query, spans
+
+
+def augment_values(row, rng, attempts=4):
+    """Broaden numeric evidence without changing any non-numeric field.
+
+    Positional augmentation removes the layout cue that distinguished a four-digit
+    episode from YEAR/RESOLUTION, but round 2 never supplied four-digit EPISODE
+    evidence. This view changes digits *inside already typed spans*, re-derives the
+    record through the contract, and refuses any collateral field change.
+    """
+    if row["record"] is None:
+        return None
+    ordered = sorted(row["spans"], key=lambda span: span["start"])
+    eligible = [(index, span) for index, span in enumerate(ordered)
+                if span["kind"] in ("EPISODE", "SEASON")
+                and re.search(r"\d+", row["query"][span["start"]:span["end"]])]
+    if not eligible:
+        return None
+    for _ in range(attempts):
+        replacements = {}
+        chosen = [pair for pair in eligible if rng.random() < 0.7]
+        if not chosen:
+            chosen = [rng.choice(eligible)]
+        for index, span in chosen:
+            text = row["query"][span["start"]:span["end"]]
+            if span["kind"] == "EPISODE":
+                # Current held-out 4-digit episodes are multiples of 100. Train on
+                # the width, never their exact values.
+                number = rng.randrange(1001, 10000)
+                while number % 100 == 0:
+                    number = rng.randrange(1001, 10000)
+            else:
+                # heldout-value's big seasons are 10-23.
+                number = rng.randrange(24, 31)
+            replacements[index] = re.sub(r"\d+", str(number), text, count=1)
+        query, spans = _replace_span_text(row, replacements)
+        if len(query) > MAX_CHARS:
+            continue
+        derived, issues = contract_record(query, spans, check_debris=True)
+        if derived is None or [issue for issue in issues if issue not in TOLERATED]:
+            continue
+        changed = {key for key in set(row["record"]) | set(derived)
+                   if row["record"].get(key) != derived.get(key)}
+        if not changed or not changed <= {"season", "episodes"}:
+            continue
+        return {**row, "query": query, "spans": spans, "record": derived,
+                "tolerated": sorted(set(issues)),
+                "augmented": True,
+                "augmented_axes": sorted(set(row.get("augmented_axes", [])) | {"value"})}
+    return None
+
+
+def augment_path(row, rng, attempts=4):
+    """Place the row in randomized path components and add duplicate evidence.
+
+    Roots are generic development vocabulary, deliberately not the held-out `TV`
+    or `Series` literals. A duplicate title is DISCARD, as contract §2 requires;
+    successful views must derive the parent record exactly.
+    """
+    if row["record"] is None:
+        return None
+    extension = row["query"][row["query"].rindex("."):]
+    original = [(span["kind"], row["query"][span["start"]:span["end"]])
+                for span in sorted(row["spans"], key=lambda span: span["start"])]
+    titles = [text for kind, text in original if kind == "TITLE"]
+    if not titles:
+        return None
+    for _ in range(attempts):
+        parts = [("DISCARD", rng.choice(PATH_ROOTS)), *original]
+        # A duplicate fragment is not a second title (§2). Put it in a varying
+        # component so the view teaches identity comparison rather than one layout.
+        parts.insert(rng.randrange(2, len(parts) + 1), ("DISCARD", rng.choice(titles)))
+        boundaries = len(parts) - 1
+        path_boundaries = set(rng.sample(range(boundaries), k=min(3, boundaries)))
+        query, spans = "", []
+        for index, (kind, text) in enumerate(parts):
+            if index:
+                query += rng.choice(("/", "\\")) if index - 1 in path_boundaries else rng.choice(JOINERS)
+            start = len(query)
+            query += text
+            spans.append({"kind": kind, "text": text, "start": start, "end": len(query)})
+        query += extension
+        if len(query) > MAX_CHARS:
+            continue
+        derived, issues = contract_record(query, spans, check_debris=True)
+        if derived != row["record"] or [issue for issue in issues if issue not in TOLERATED]:
+            continue
+        return {**row, "query": query, "spans": spans, "record": derived,
+                "tolerated": sorted(set(issues)),
+                "augmented": True,
+                "augmented_axes": sorted(set(row.get("augmented_axes", [])) | {"path"})}
+    return None
+
+
+def augment_training(row, mode, position_rng, value_rng, path_rng, marker_rng=None,
+                     boundary_rng=None):
+    """Compose the supported axes while keeping each axis's random stream separate.
+
+    Every mode starts from the positional shuffle, so the added axes are always
+    measured *on top of* the layout cue being broken. The extra RNGs exist for the
+    same reason the value and path ones do: sharing a stream with the position
+    shuffle would make the round-2 paired controls depend on which later axis the
+    arm happened to use.
+    """
+    view = augment(row, position_rng)
+    if view is None:
+        return None
+    axes = set(view.get("augmented_axes", [])) | {"position"}
+    view["augmented_axes"] = sorted(axes)
+    if mode in ("shuffle-values", "shuffle-rich"):
+        view = augment_values(view, value_rng)
+        if view is None:
+            return None
+    if mode in ("shuffle-paths", "shuffle-rich"):
+        view = augment_path(view, path_rng)
+        if view is None:
+            return None
+    if mode == "shuffle-markers":
+        if marker_rng is None:
+            raise ValueError("shuffle-markers needs its own rng")
+        view = augment_markers(view, marker_rng)
+        if view is None:
+            return None
+    if mode == "shuffle-boundary":
+        if boundary_rng is None:
+            raise ValueError("shuffle-boundary needs its own rng")
+        view = augment_boundary(view, boundary_rng)
+        if view is None:
+            return None
+    return view
+
+
+def build_split(rng, names, count, families, used_queries, quarantine):
+    picked = []
+    for index in range(count):
+        lang, title, alias = families[index]
+        choices = [n for n in names if n not in NEEDS_ALIAS or alias]
+        if not choices:
+            quarantine["no-layout-for-family"] += 1
+            continue
+        layout = rng.choice(choices)
+        atoms = make_atoms(rng, lang, title, alias)
+        query, spans = render(atoms, LAYOUTS[layout][1], rng.choice(EXTENSIONS))
+        row = finalize(query, spans, layout, lang, title, used_queries, quarantine)
+        if row:
+            picked.append(row)
+    return picked
+
+
+# Paired probes: one controlled change per row, everything else identical to the
+# matching `probe-base` row. The pilot showed a bare held-out number cannot be
+# attributed, so each probe answers exactly one question about which factor broke
+# the model. `probe-*` rows are diagnostics, never claim metrics.
+PROBE_BASE = "n1"
+PROBE_LETTERS = ("ru", "ja", "zh")  # the script probe needs a second script to exist
+
+
+def build_probes(rng, families, used_queries, quarantine):
+    base, variants = [], {f"probe-{name}": [] for name in
+                          ("layout", "script", "value", "episode-title")}
+    heldout = [n for n, (tag, parts) in LAYOUTS.items()
+               if tag.startswith("heldout-layout") and ("a", "a") not in parts]
+    base_parts = LAYOUTS[PROBE_BASE][1]
+    with_title = base_parts[:5] + [("a", "x"), "-"] + base_parts[5:]
+    for lang, title, alias in families:
+        if lang not in PROBE_LETTERS or not alias:
+            continue
+        atoms = make_atoms(rng, lang, title, alias)
+        query, spans = render(atoms, base_parts, ".mkv")
+        parent = finalize(query, spans, PROBE_BASE, lang, title, used_queries, quarantine)
+        if not parent:
+            continue
+        base.append(parent)
+        number = len(base)
+        cases = {"layout": (LAYOUTS[rng.choice(heldout)][1], None),
+                 "script": (base_parts, {"t": ("TITLE", alias)}),
+                 "value": (base_parts, {"e": ("EPISODE", f"{number * 100:04d}")}),
+                 "episode-title": (with_title, None)}
+        for name, (parts, changed) in cases.items():
+            probe_atoms = dict(atoms)
+            probe_atoms.update(changed or {})
+            query, spans = render(probe_atoms, parts, ".mkv")
+            row = finalize(query, spans, f"probe-{name}", lang, title, used_queries, quarantine)
+            if row:
+                row["probe_parent"] = parent["query"]
+                variants[f"probe-{name}"].append(row)
+    return {"probe-base": base, **variants}
+
+
+def generate():
+    if ROOT.exists():
+        raise SystemExit(f"refusing to overwrite the locked corpus at {ROOT}")
+    ROOT.mkdir(parents=True)
+    rng = random.Random(SEED)
+    families = []
+    seen_titles = set()
+    for lang in LANGS:
+        words = WORDS[lang]
+        for a, b, c in [(x, y, z) for x in words for y in words for z in words
+                        if x != y and y != z]:
+            title = f"{a} {b} {c}"
+            if title in seen_titles:
+                continue  # an identical string in two languages would leak families
+            seen_titles.add(title)
+            table = dict(zip(words, ALIAS.get(lang, [])))
+            families.append((lang, title, " ".join(table.get(w, w) for w in title.split(" "))
+                             if lang in ALIAS else None))
+    rng.shuffle(families)
+    spec = [("train-narrow", ["narrow", "narrow-neg"], 1600),
+            ("train-extra", ["broad", "broad-neg"], 3200),
+            ("validation", ["narrow", "broad", "narrow-neg", "broad-neg"], 240),
+            ("easy", ["narrow", "broad", "narrow-neg", "broad-neg"], 240),
+            ("heldout-layout", ["heldout-layout-a", "heldout-layout-b"], 240),
+            ("heldout-lexical", ["heldout-lexical"], 160),
+            ("heldout-value", ["heldout-value"], 160),
+            ("heldout-negative", ["heldout-neg"], 120)]
+    total = sum(count for _, _, count in spec) + 600  # 600 more families for probes
+    if len(families) < total:
+        raise SystemExit(f"only {len(families)} disjoint title families for {total} rows")
+    manifest = {"seed": SEED, "max_chars": MAX_CHARS, "schema": "contract v2.1",
+                "labels_from": "authored layout atoms, derived by fp.normalize.record",
+                "claim": "format generalization on controlled data, not real-world accuracy",
+                "arms": {k: {"layout_tags": v, "train_splits": ARM_SPLITS[k]}
+                             for k, v in ARM_TAGS.items()},
+                "layouts": {name: {"tag": tag, "atoms": sorted({p[1] for p in parts
+                                                                if not isinstance(p, str)})}
+                            for name, (tag, parts) in LAYOUTS.items()},
+                "splits": {}, "quarantine": {}}
+    cursor, used, quarantine = 0, set(), Counter()
+
+    def emit(split, rows):
+        path = ROOT / f"{split}.jsonl"
+        path.write_text("".join(json.dumps(r, ensure_ascii=False) + "\n" for r in rows))
+        manifest["splits"][split] = {
+            "n": len(rows), "sha256": digest(path),
+            "layouts": sorted({r["layout"] for r in rows}),
+            "languages": dict(Counter(r["language"] for r in rows)),
+            "fields": dict(Counter(k for r in rows for k in (r["record"] or {}))),
+            "abstentions": sum(1 for r in rows if r["record"] is None),
+            "max_chars": max(len(r["query"]) for r in rows)}
+        for name in manifest["splits"][split]["layouts"]:
+            if split.startswith("train") and LAYOUTS[name][0].startswith("heldout"):
+                raise AssertionError(f"held-out layout {name} leaked into {split}")
+
+    for split, tags, count in spec:
+        names = [n for n, (tag, _) in LAYOUTS.items() if tag in tags]
+        subset = families[cursor:cursor + count]
+        cursor += count
+        rows = build_split(rng, names, count, subset, used, quarantine)
+        assert rows, f"{split} produced nothing"
+        emit(split, rows)
+    # Probes get their own families, so a probe row never shares a title with a
+    # scored split; the pairing is inside the probe set, not across splits.
+    probe_subset = families[cursor:cursor + 600]
+    for split, rows in build_probes(rng, probe_subset, used, quarantine).items():
+        emit(split, rows)
+    manifest["quarantine"] = dict(quarantine)
+    (ROOT / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n")
+    print(json.dumps({k: v["n"] for k, v in manifest["splits"].items()} |
+                     {"quarantine": dict(quarantine)}, ensure_ascii=False, indent=2))
+
+
+def rows(split):
+    manifest = json.loads((ROOT / "manifest.json").read_text())
+    path = ROOT / f"{split}.jsonl"
+    data = [json.loads(line) for line in path.read_text().splitlines()]
+    assert digest(path) == manifest["splits"][split]["sha256"], f"{split}: hash mismatch"
+    assert len(data) == manifest["splits"][split]["n"], f"{split}: row count changed"
+    return data
+
+
+def verify():
+    manifest = json.loads((ROOT / "manifest.json").read_text())
+    problems, queries, families = [], set(), {}
+    for split in manifest["splits"]:
+        group = "probes" if split.startswith("probe-") else split
+        for row in rows(split):
+            queries.add(row["query"])
+            owner = families.setdefault(row["family"], group)
+            if owner != group:
+                problems.append(f"{row['family']}: family in {group} and {owner}")
+            derived, issues = contract_record(row["query"], row["spans"], check_debris=True)
+            if derived != row["record"] or sorted(set(issues)) != row["tolerated"]:
+                problems.append(f"{row['query'][:40]}: gold not reproducible {issues}")
+            if len(row["query"]) > manifest["max_chars"]:
+                problems.append(f"{row['query'][:40]}: exceeds locked length")
+    print(json.dumps({"rows": len(queries), "families": len(families),
+                      "problem_count": len(problems), "problems": problems[:20]},
+                     ensure_ascii=False, indent=2))
+    return 1 if problems else 0
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--verify", action="store_true")
+    args = parser.parse_args(argv)
+    return verify() if args.verify else (generate(), 0)[1]
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

app/fp/normalize.py

@@ -0,0 +1,544 @@
+"""Contract v2 normalizer: evidence spans -> emitted record, deterministically.
+
+Implements docs/LABELING-CONTRACT.md sections 3-7. This is the only permitted
+path from labels to an emitted record: annotators mark spans, this code derives
+the record. Hand-editing an output string is a contract violation, so the
+derivation has to be inspectable and testable on its own.
+
+Span dicts: `{'kind', 'start', 'end'}` plus optional `text` (must match the
+query bytes), `options` and `value`. Kinds: TITLE, EPISODE_TITLE, GROUP, YEAR, SEASON,
+EPISODE, RANGE, LIST, RESOLUTION, DIMENSIONS, DISCARD. `options` carries
+label-side decisions the contract refuses to guess from bytes: `is_title`,
+`fold_case`, `items`. `value` carries a normalized season number (None means
+an explicit-but-unbounded ordinal, which is omitted, never 0).
+
+`covered` may list extra [start, end) ranges (for example a path component
+carrying no field) so that the debris check is a fact about the label, not a
+guess about punctuation. Debris checking is opt-in because real corpora are
+partially labeled; a full-label row should turn it on.
+"""
+
+import re
+import unicodedata
+
+# A group name is copied verbatim from one bracket, so a collab credit such as
+# `ACB字幕组&天空字幕组` stays one string; two separate group spans conflict.
+MULTI_GROUP = re.compile(r"\]\s*\[")
+EXTENSION = re.compile(r"\.[A-Za-z0-9]{1,8}$")
+
+# These punctuation marks are used as structural delimiters by the controlled
+# format suites. They are not title characters at a predicted span edge; the
+# gold derivation still uses the original spans and never calls this trimmer.
+# The Unicode members were added after the consumed fresh-6 audit exposed
+# full-width/vertical delimiters bleeding into TITLE, SEASON, and EPISODE spans,
+# then extended for the fresh-7 delimiter inventory *before* fresh-7 was
+# generated, so no suite can be accused of picking delimiters after seeing the
+# trimmer. The list is duplicated verbatim in goport/normalize.go; both sides
+# are pinned by tests. ASCII "+" is deliberately absent: it occurs in real
+# titles ("C++") and would corrupt evidence to trim it.
+STRUCTURAL_EDGE_CHARS = frozenset(
+    "#<>{}@|／∥｜⁘❖【】『』⟦⟧‖‣⁓⧸⁄∕⋯»〔〕❪❫｟｠〘〙〖〗«》"
+)
+
+# The roster above is only the manual half of the edge test. Any non-ASCII rune in the
+# dash, bracket or punctuation categories is an edge rune too, even when the roster
+# predates it: that is what stops the set from measuring which delimiters happened to be
+# enumerated before a suite shipped. ASCII runes are never widened this way -- ASCII
+# punctuation legitimately terminates role text ("Title...", "(Recut)", "Title '05'"), so
+# ASCII stays roster-only.
+_EDGE_CATEGORIES = frozenset({"Pd", "Po", "Ps", "Pe", "Pi", "Pf", "Pc"})
+
+
+def is_structural_edge(ch: str) -> bool:
+    """Whether `ch` may be trimmed from a *predicted* span edge. Gold spans never see this."""
+    if ord(ch) < 0x80:
+        return ch in STRUCTURAL_EDGE_CHARS
+    return ch in STRUCTURAL_EDGE_CHARS or unicodedata.category(ch) in _EDGE_CATEGORIES
+
+PATH_HASH_GROUP = re.compile(
+    r"(?i)(?:[- ](?:subs?|raws?|team|works?|encode|unit|signal|depot|delta|plow|"
+    r"cellar|marrow|nest|harbor|lantern|rip|release|group)|"
+    r"(?:subs|raws|team|works|unit|signal|depot|delta|plow|cellar|marrow|nest|"
+    r"harbor|lantern))$"
+)
+
+# Contract §10: sidecars and documents are not media, even with a perfect
+# title. Containers (.rar/.zip/.7z/.torrent) are NOT listed: §10 parses them.
+NON_MEDIA_EXTENSIONS = frozenset({
+    "srt", "ass", "ssa", "vtt", "sub", "idx", "sup", "smi",          # subtitles
+    "jpg", "jpeg", "png", "gif", "webp", "bmp", "tbn",               # images
+    "nfo", "txt", "pdf", "epub", "docx", "doc", "md", "html", "htm", # documents
+    "xml", "json", "log",
+    "md5", "sfv", "sha1", "sha256", "par2",                         # checksums
+    "gz", "tar", "bz2", "xz",                                       # non-media archives
+})
+
+
+def non_media_extension(query):
+    """The final dot-suffix, lower-cased, when it names a non-media sidecar;
+    otherwise None. Position-anchored like `EXTENSION`: only the true suffix
+    counts, a `.srt` in the middle of a path is not a decision here."""
+    match = EXTENSION.search(query)
+    if not match:
+        return None
+    ext = match.group(0)[1:].lower()
+    return ext if ext in NON_MEDIA_EXTENSIONS else None
+
+
+def leading_hash_stem(query):
+    """Whether a query has the titleless CRC-prefixed release-stem shape.
+
+    This is intentionally narrower than ``query.startswith('[')``: the stem
+    must have a hex token, two dash-separated fields, and a closed resolution
+    class at the end. It is a decision-layer rule for rows whose only apparent
+    title is a group-shaped token between a checksum and codec junk; ordinary
+    bracket-prefixed titles and two-field releases remain untouched.
+    """
+    if not re.match(r"^\[[0-9a-fA-F]{7,12}\]\s+", query):
+        return False
+    rest = query[query.find("]") + 1:].lstrip()
+    fields = rest.split(" - ")
+    if len(fields) != 3 or not all(fields[:2]):
+        return False
+    tail = fields[2]
+    return bool(re.fullmatch(r"(?:480p|576p|720p|1080p|1080i|2160p)\.[A-Za-z0-9]{1,8}",
+                             tail, flags=re.IGNORECASE))
+
+
+UNICODE_SPACE = re.compile("[   -   　]")
+DIGITS = re.compile(r"\d+(?:\.\d+)*")
+YEAR_LIKE = re.compile(r"\b\d{4}\b")
+OPEN_BRACKETS = "([｛「『〈《【“｢"
+CLOSE_BRACKETS = ")]｝」』〉》】”｣"
+BRACKETS = OPEN_BRACKETS + CLOSE_BRACKETS
+# Brackets are deliberately NOT edge-stripped: an unmatched bracket is either
+# lexical title content or a label error, and silently deleting one side of a
+# pair corrupts evidence. Only balanced pairs are removed (see normalize_text).
+STRIP_CHARS = " \t\r\n-–—~～·•:：/\\."
+
+
+def _balanced(text):
+    return all(text.count(open_) >= text.count(close)
+               for open_, close in zip(OPEN_BRACKETS, CLOSE_BRACKETS))
+
+
+def _strip_bracket_pairs(text):
+    """Contract 3: brackets are removed only as balanced pairs enclosing the
+    whole fragment. `ポコニャン(叮噹貓)` keeps its parentheses (they are inside
+    the title), `｢title」` and `「title」` lose them (they delimit it)."""
+    while (len(text) >= 2 and text[0] in OPEN_BRACKETS and text[-1] in CLOSE_BRACKETS
+           and _balanced(text[1:-1])):
+        text = _fold(text[1:-1]).strip(STRIP_CHARS)
+    return text
+
+SCHEMA_VALUES = {"480p", "576p", "720p", "1080p", "1080i", "2160p"}
+# Contract 7: a bare height is only HD evidence. SD heights are shared with
+# computer displays (640x480 VGA, 800x480) and non-square-pixel DVD captures,
+# so they never establish a broadcast class.
+HEIGHT_CLASS = {720: "720p", 1080: "1080p", 2160: "2160p"}
+MIN_ASPECT, MAX_ASPECT = 1.7, 1.95  # widescreen video frames only
+MAX_EPISODES = 64
+YEAR_MIN, YEAR_MAX = 1870, 2099
+
+# Contract 5 value resolution. Deliberately a *closed* vocabulary: if a season
+# marker is not in here, the row reports `season-unparsable` instead of guessing.
+# A model that emits the right span never has to know numbers, and this table is
+# the auditable place where text becomes a value.
+DIGITS = re.compile(r"\d+")
+FINAL_WORDS = ("final", "última", "ultimas", "last", "fin", "おしまい", "最終", "最后", "最终",
+               "最後", "끝", "последн")
+ORDINAL_WORDS = {
+    "first": 1, "second": 2, "third": 3, "fourth": 4, "fifth": 5, "sixth": 6,
+    "seventh": 7, "eighth": 8, "ninth": 9, "tenth": 10, "primeira": 1, "segunda": 2,
+    "terceira": 3, "quarta": 4, "quinta": 5, "primera": 1, "tercera": 3,
+    "premiere": 1, "deuxieme": 2, "deuxième": 2, "troisieme": 3, "troisième": 3,
+    "erste": 1, "zweite": 2, "dritte": 3, "primo": 1, "secundo": 2, "seconda": 2,
+    "prima": 1, "terzo": 3, "uma": 1, "dois": 2, "tres": 3, "três": 3, "cuarta": 4,
+    "quinta": 5, "sexta": 6, "fifth": 5,
+    "первый": 1, "второй": 2, "третий": 3, "четвёртый": 4, "пятый": 5, "шестой": 6,
+    "седьмой": 7, "восьмой": 8, "девятый": 9, "десятый": 10,
+}
+CJK_NUMERALS = {"一": 1, "二": 2, "三": 3, "四": 4, "五": 5, "六": 6, "七": 7,
+                "八": 8, "九": 9, "十": 10, "壹": 1, "貳": 2, "叁": 3}
+
+
+def season_value(text):
+    """int, None (=explicit season with no number), or 'unparsable'.
+
+    Digits win over words: `2nd Season` and `第2期` are the same value. A
+    multi-number marker (`Season 2 of 12`, `S01E02` mistagged) is unparsable
+    rather than a pick-one, because either choice is a guess about the work.
+    """
+    folded = normalize_text(text, fold_case=True)
+    if any(word in folded for word in FINAL_WORDS):
+        return None
+    numbers = DIGITS.findall(folded)
+    if len(numbers) == 1:
+        value = int(numbers[0])
+        return value if 0 <= value <= 99 else "unparsable"
+    if len(numbers) > 1:
+        return "unparsable"
+    for word in folded.split(" "):
+        if word in ORDINAL_WORDS:
+            return ORDINAL_WORDS[word]
+    runs = [c for c in folded if c in CJK_NUMERALS]
+    if len(runs) == 1:
+        return CJK_NUMERALS[runs[0]]
+    return "unparsable"
+
+
+def _fold(text):
+    """Contract 3: underscores and exotic whitespace become spaces, runs
+    collapse, nothing is transliterated. Dots are handled separately, because
+    whether one is a delimiter depends on its neighbours (see
+    `fold_dot_delimiters`)."""
+    text = UNICODE_SPACE.sub(" ", text)
+    text = text.replace("_", " ")
+    text = re.sub(r"[-–—]{2,}", "-", text)
+    return re.sub(r" {2,}", " ", text).strip()
+
+
+ALPHA = re.compile(r"[^\W\d_]+\Z", re.UNICODE)
+NUMERIC = re.compile(r"\d+\Z")
+
+
+def _segment_kind(unit):
+    if not unit:
+        return "empty"
+    if ALPHA.match(unit):
+        return "word"
+    if NUMERIC.match(unit):
+        return "number"
+    return "mixed"
+
+
+def fold_dot_delimiters(text):
+    """Contract 3.2, title fragments only, decided dot by dot inside each
+    whitespace token, where a segment is a *word* (letters of any script), a
+    *number* (digits only), *mixed* (letters and digits together: `1080p`,
+    `2a`, `S01`, `x264`) or *empty* (a doubled or trailing dot).
+
+    A dot splits when either neighbour is a word (`CITY.HUNTER` ->
+    `CITY HUNTER`, `Akane.1988` -> `Akane 1988`, `Movie.2.Innocence` ->
+    `Movie 2 Innocence`), when the right side is mixed (`Ghost.1080p` ->
+    `Ghost 1080p`: a technical tag is never title text, and a split that only
+    touches title-adjacent junk cannot lose evidence), or when a side is empty
+    (`Dr.`, `Ghost.` lose the delimiter dot). The one shape that is never cut
+    is `number . number` before any word has appeared in the token:
+    `1917.2019` and `2.43.121` survive as single tokens, because a run that
+    opens with digits reads as an encoded value and splitting it would be a
+    guess about which part is the title. Single letters are words, so
+    `A.K.A.` folds to `A K A`.
+
+    Consequence: a title that genuinely contains a digit-dot-digit sequence
+    must be labeled as two fragments. That is the honest cost of a rule an
+    annotator can apply without knowing the work."""
+    pieces = []
+    for token in text.split(" "):
+        segments = token.split(".")
+        if len(segments) == 1:
+            if token:
+                pieces.append(token)
+            continue
+        kinds = [_segment_kind(s) for s in segments]
+        cut, word_seen = [0], False
+        for index in range(1, len(segments)):
+            previous, current = kinds[index - 1], kinds[index]
+            touches_word = previous == "word" or current == "word"
+            if current in ("mixed", "empty") or touches_word:
+                cut.append(index)  # a word or junk boundary: split here
+            elif previous == "number" and current == "number" and word_seen:
+                cut.append(index)  # `Akane.1988.2020` splits; `1917.2019` does not
+            word_seen = word_seen or touches_word
+        for boundary, edge in zip(cut, cut[1:] + [len(segments)]):
+            unit = ".".join(segments[boundary:edge])
+            if unit:
+                pieces.append(unit)
+    text = " ".join(pieces)
+    return re.sub(r" {2,}", " ", text).strip()
+
+
+def trim_structural_edges(query, spans):
+    """Trim structural delimiter runes from predicted span edges.
+
+    This is deliberately prediction-only. Gold labels are never rewritten:
+    a label span that owns `#`, `<`, `>`, `{`, `}`, `@`, or `|` remains
+    lossless. The trimmer handles a decoder that includes a neighboring
+    delimiter in a role span, and drops a span made only of delimiters.
+    """
+    out = []
+    changed = 0
+    for span in spans:
+        start, end = span["start"], span["end"]
+        first, last = start, end
+        while first < last and is_structural_edge(query[first]):
+            first += 1
+        while last > first and is_structural_edge(query[last - 1]):
+            last -= 1
+        if first != start or last != end:
+            changed += 1
+        if first < last:
+            out.append({**span, "start": first, "end": last})
+    return out, changed
+
+
+def titleless_path_group(query, spans):
+    """Narrow prediction gate for ``(hash)/junk/Group/res`` stems.
+
+    The parenthesized hash, slash path, and group-like title suffix must all be
+    present, with no episode, season, year, or episode-title evidence. A
+    positive row with any additional title evidence is left untouched.
+    """
+    if not re.match(r"^\([0-9a-fA-F]{7,12}\)", query) or "/" not in query:
+        return False
+    titles = [span for span in spans if span["kind"] == "TITLE"]
+    evidence = [span for span in spans if span["kind"] in
+                ("EPISODE", "EPISODE_TITLE", "SEASON", "YEAR")]
+    if len(titles) != 1 or evidence:
+        return False
+    return bool(PATH_HASH_GROUP.search(query[titles[0]["start"]:titles[0]["end"]].strip()))
+
+
+def normalize_text(text, is_title=False, fold_case=False, keep_edges=False):
+    """`keep_edges=True` is for an already-assembled title: an edge bracket or
+    hyphen there belongs to an adjacent fragment, so edge stripping is skipped
+    and only byte-internal rules (separators, whitespace, delimiter dots)
+    apply. Fragments must not use it - that is where delimiters are removed."""
+    text = _fold(text)
+    if not keep_edges:
+        text = text.strip(STRIP_CHARS)
+        text = _strip_bracket_pairs(text)
+        text = text.strip(STRIP_CHARS)
+    if fold_case:
+        text = text.lower()
+    if is_title:
+        text = fold_dot_delimiters(text)
+    return _fold(text)
+
+
+def _range_endpoints(text):
+    """Return (None, (lo, hi), issues). Repeated season prefixes
+    (`S01E01-S01E03`) and embedded technical values (`[01 ~ 24][1080p]`,
+    `1280x720`) are removed first, so the endpoints are the only numbers left.
+    Endpoints are 1-3 digits: a 4-digit number is a year or a checksum, not an
+    episode index."""
+    stripped = re.sub(r"(?i)s\d+", " ", text)
+    stripped = re.sub(r"(?i)\d{3,5}\s*[x×]\s*\d{3,5}|\d{3,4}[pi]\b", " ", stripped)
+    nums = [int(n) for n in re.findall(r"\d{1,3}", stripped)]
+    if not nums:
+        return None, None, ["range-no-digits"]
+    if len(nums) > 2:
+        return None, None, ["range-too-complex"]
+    if len(nums) == 1:
+        return None, None, ["range-single-value"]
+    if not re.search(r"[-~–—至到]", text):
+        return None, None, ["range-no-connector"]
+    return None, (nums[0], nums[-1]), []
+
+
+def expand_range(text, cap=MAX_EPISODES):
+    _, endpoints, issues = _range_endpoints(text)
+    if issues:
+        return [], issues
+    a, b = endpoints
+    if a == b:
+        issues.append("range-single-value")
+    lo, hi = min(a, b), max(a, b)
+    if hi - lo + 1 > cap:
+        return [], ["list-too-long"]
+    if hi - lo + 1 > 200:
+        issues.append("range-suspiciously-long")
+    return list(range(lo, hi + 1)), issues
+
+
+def expand_list(span, text=None):
+    """Item values come from the label side when there is one: re-reading them
+    from bytes would reintroduce exactly the ambiguity the contract says to
+    quarantine. A *predicted* span has no label side, so for those the digits
+    inside the claimed boundary are read - the model's claim is the boundary.
+    """
+    values = span.get("options", {}).get("items")
+    if values is None:
+        values = re.findall(r"\d+", text if text is not None else span["text"])
+    values = sorted({int(v) for v in values})
+    if not values:
+        return [], ["list-empty"]
+    if len(values) > MAX_EPISODES:
+        return [], ["list-too-long"]
+    return values, []
+
+
+def debris(query, claims):
+    """Unclaimed, non-punctuation text: what a full-label row cannot explain.
+
+    Only the container suffix at the very end of the query is excused, and it
+    is excused by *position*, not by appearance: a `.720p` left unclaimed in
+    the middle of a name is a missing decision and must be reported even
+    though it looks like an extension. Stripping dot-suffixes from the
+    leftovers instead silently forgives exactly the technical tags that strict
+    rows exist to catch."""
+    mask = bytearray(len(query))
+    suffix = EXTENSION.search(query)  # anchored at `$`, so this is the final suffix
+    for start, end in list(claims) + ([(suffix.start(), suffix.end())] if suffix else []):
+        for i in range(max(0, min(start, len(query))), min(max(0, end), len(query))):
+            mask[i] = 1
+    text = "".join(ch for i, ch in enumerate(query) if not mask[i])
+    # Contract 3.3: a run of underscores or hyphens that only delimits fields is
+    # debris the pipeline removes, not text that needs a recorded decision. `_`
+    # counted as a word character here, which made every underscored layout look
+    # unexplainable.
+    text = text.replace("_", " ")
+    return [part for part in re.split(r"[^\w]+", text, flags=re.UNICODE) if part]
+
+
+def record(query, spans, *, covered=None, check_debris=False):
+    """Derive the emitted record from labeled spans. Returns (record|None,
+    issues). Issues never silently change a value; conflicting or
+    unrepresentable evidence is reported so the row can be quarantined."""
+    issues = []
+    claims = [(s["start"], s["end"]) for s in spans] + list(covered or [])
+    title, episode_title, episodes = [], [], []
+    group = year = season = resolution = dimensions = None
+    for span in sorted(spans, key=lambda s: (s["start"], s["end"])):
+        kind = span["kind"]
+        text = span.get("text")
+        if text is None:
+            text = query[span["start"]:span["end"]]
+        elif query[span["start"]:span["end"]] != text:
+            issues.append("span-text-mismatch")
+        opts = span.get("options", {})
+        if kind in ("TITLE", "EPISODE_TITLE"):
+            # Which fragment is the work title and which is the episode title is
+            # the annotator's decision, recorded as a span kind. Position cannot
+            # decide it: a second TITLE fragment is usually the same work title
+            # in another script, while a second EPISODE_TITLE fragment is the
+            # next segment of one episode's name.
+            bucket = title if kind == "TITLE" else episode_title
+            value = normalize_text(text, is_title=True,
+                                   fold_case=opts.get("fold_case", False))
+            if value:
+                bucket.append(value)
+            else:
+                issues.append("empty-title-fragment")
+        elif kind == "GROUP":
+            value = normalize_text(text)
+            if group is not None:
+                issues.append("group-multiple-brackets")
+            elif MULTI_GROUP.search(value):
+                issues.append("group-multiple-names")
+            group = group or value
+        elif kind == "YEAR":
+            digits = YEAR_LIKE.findall(text)
+            if len(digits) != 1:
+                issues.append("year-ambiguous")
+                continue
+            value = int(digits[0])
+            if not YEAR_MIN <= value <= YEAR_MAX:
+                issues.append("year-out-of-range")
+            elif year is not None and year != value:
+                issues.append("year-conflict")
+            else:
+                year = value
+        elif kind == "SEASON":
+            if "value" in span:
+                value = span["value"]  # annotator-supplied (contract 5, worksheet)
+            else:
+                value = season_value(text)  # predicted span: text decides, not a head
+                if value == "unparsable":
+                    issues.append("season-unparsable")
+                    continue
+            if value is None:
+                issues.append("season-unspecified")
+            elif season is not None and season != value:
+                issues.append("season-conflict")
+            elif not 0 <= value <= 99:
+                issues.append("season-out-of-range")
+            else:
+                season = value
+        elif kind == "EPISODE":
+            nums = re.findall(r"\d+", text)
+            if len(nums) != 1:
+                issues.append("episode-ambiguous")
+            else:
+                episodes.append(int(nums[0]))
+        elif kind == "RANGE":
+            values, extra = expand_range(text)
+            issues += extra
+            episodes += values
+        elif kind == "LIST":
+            values, extra = expand_list(span, text)
+            issues += extra
+            episodes += values
+        elif kind == "RESOLUTION":
+            value = normalize_text(text, fold_case=True)
+            if value not in SCHEMA_VALUES:
+                issues.append("resolution-unknown-class")
+            elif resolution and resolution != value:
+                issues.append("resolution-conflict")
+            else:
+                resolution = value
+        elif kind == "DIMENSIONS":
+            match = re.search(r"(\d{3,5})\s*[x×]\s*(\d{3,5})", text, re.I)
+            if not match:
+                issues.append("dimensions-unreadable")
+                continue
+            width, height = int(match.group(1)), int(match.group(2))
+            value = HEIGHT_CLASS.get(height) if MIN_ASPECT <= width / height <= MAX_ASPECT else None
+            if value is None:
+                issues.append("dimensions-no-class")
+            elif resolution and resolution != value:
+                issues.append("resolution-conflict")
+            else:
+                resolution, dimensions = value, text
+        elif kind != "DISCARD":
+            issues.append("unhandled-kind:" + kind)
+
+    if check_debris:
+        left = debris(query, claims)
+        if left:
+            issues.append("unexplained-debris:" + "|".join(left[:6]))
+
+    out = {}
+    if title:
+        # join=True: an edge bracket here belongs to an adjacent fragment, so
+        # only byte-internal rules apply to the assembled title.
+        out["title"] = normalize_text(" ".join(title), is_title=True, keep_edges=True)
+    if episode_title:
+        out["episode_title"] = normalize_text(" ".join(episode_title), is_title=True,
+                                             keep_edges=True)
+    if year is not None:
+        out["year"] = year
+    if season is not None:
+        out["season"] = season
+    if episodes:
+        values = sorted(set(episodes))
+        if len(values) > MAX_EPISODES:
+            # Contract 9: never silently truncate; the row is quarantined.
+            issues.append("list-too-long")
+        out["episodes"] = values
+    if group:
+        out["release_group"] = group
+    if resolution:
+        out["resolution"] = resolution
+    # `required: title` means a title-less record is the same output as no call.
+    return (out if out.get("title") else None), issues
+
+
+def gated_record(query, spans, **kwargs):
+    """`record` followed by the decision layer applied at prediction time.
+
+    A non-media suffix or the narrow CRC-prefixed titleless release-stem shape
+    forces abstention and reports why in the issues list.
+    """
+    ext = non_media_extension(query)
+    trimmed, trimmed_count = trim_structural_edges(query, spans)
+    out, issues = record(query, trimmed, **kwargs)
+    if trimmed_count:
+        issues = issues + [f"trimmed-structural-edges:{trimmed_count}"]
+    if out is not None and ext is not None:
+        return None, issues + [f"gated-non-media-extension:{ext}"]
+    if out is not None and leading_hash_stem(query):
+        return None, issues + ["gated-leading-hash-stem"]
+    if out is not None and titleless_path_group(query, trimmed):
+        return None, issues + ["gated-titleless-path-group"]
+    return out, issues

app/fp/parser2.py

@@ -0,0 +1,891 @@
+"""Train and score the production head design on the contract v2.1 corpus.
+
+The architecture question this answers is narrow and specific: can a small
+encoder emit *typed spans* over layouts it never saw, with every value resolved
+by deterministic contract code? There is no value head here on purpose. The
+pilot needed one, and its hard-split season accuracy was the only field that
+survived pretraining; if this run reproduces field-level reliability without it,
+the head was compensating for narrow data rather than doing necessary work.
+
+Predictions are scored through `fp.normalize.record`, the same function that
+derived the gold. A prediction is therefore a parse under the ratified contract,
+not a string diff: abstention, ranges, lists and case folding are all evaluated
+as the contract defines them, which is what makes these numbers comparable with
+the real-data numbers we cannot produce yet.
+
+Usage:
+    python -m fp.parser2 --arm B-broad --mode characters --seed 0 \
+        --out runs/parser-v2/broad-characters-0
+"""
+
+import argparse
+import importlib.metadata
+import json
+import math
+import os
+import random
+import statistics
+import sys
+import time
+from collections import Counter, defaultdict
+from pathlib import Path
+
+# Required by deterministic CUDA matrix multiplication. This has to be present
+# before the first CUDA context is initialized; setting it here keeps remote runs
+# reproducible without relying on one shell's environment.
+os.environ.setdefault("CUBLAS_WORKSPACE_CONFIG", ":4096:8")
+
+import torch
+from torch import nn
+from transformers import BertConfig, BertModel
+
+from fp import corpus2, normalize
+from fp.corpus2 import ARM_SPLITS, ROOT as CORPUS, augment, augment_training, digest, rows
+from fp.corpus2 import KINDS as corpus_KINDS
+from fp.normalize import record as contract_record
+
+SNAP = Path(os.environ.get(
+    "ANIFILEBERT_SNAPSHOT",
+    "/home/lain/.cache/huggingface/hub/models--ModerRAS--AniFileBERT/snapshots/"
+    "d8ddb0b54dad4d65a60ab7eba5acb06a2a1fab02"))
+KINDS = list(corpus_KINDS)
+LABELS = ["O"] + [prefix + kind for kind in KINDS for prefix in ("B-", "I-")]
+# A schedule is a named, frozen list, not the meaning of a flag. Round 4 shipped
+# `balanced` as four views; growing that same name to six views in round 5 would
+# have made `scripts/parser-v2-round4.sh` silently produce a different experiment
+# under a published name, so the older list stays pinned under `balanced4`.
+SCHEDULES = {"balanced4": ("none", "shuffle", "shuffle-values", "shuffle-paths"),
+             "balanced": ("none", "shuffle", "shuffle-values", "shuffle-paths",
+                          "shuffle-markers", "shuffle-boundary"),
+             # Round 5b's dose control. Uniform rotation makes every view's share
+             # 1/|views|, so adding two views silently cut the value view from 3/12 to
+             # 2/12 -- which explains the probe-value drop without any interference
+             # hypothesis. This list restores the value share to 0.25 (C1's) while
+             # keeping the marker share at C4's 2/12, trading only boundary, the view
+             # with the weakest measured effect. A 12-slot list visited once per epoch
+             # makes each row's counts exact, not merely balanced within one.
+             "balanced-dose": ("none", "none", "shuffle", "shuffle",
+                               "shuffle-values", "shuffle-values", "shuffle-values",
+                               "shuffle-paths", "shuffle-paths",
+                               "shuffle-markers", "shuffle-markers",
+                               "shuffle-boundary")}
+BALANCED_VIEWS = SCHEDULES["balanced"]
+SELECTION_VIEWS = ("plain", "shuffle", "markers", "boundary")
+
+
+def balanced_view(row_index, epoch, seed, views=BALANCED_VIEWS):
+    """Exact rotation over a named schedule.
+
+    Unlike random independent gates, this guarantees each row receives each view the
+    same number of times over a full schedule: three times per view in round 4's
+    twelve epochs and four views, twice per view in round 5's six. `seed` rotates the
+    starting cell without changing the balance.
+    """
+    return views[(row_index + epoch + seed) % len(views)]
+
+
+def selection_views(validation_rows, seed, forbidden_queries):
+    """Fixed derived views of the validation split, for checkpoint selection.
+
+    Round 4 selected on seen-layout validation while the acceptance target is about
+    transformed views, so the choice now averages the same transformations it is
+    meant to be good at. Only validation rows are read. A derived view that lands on
+    any scored query is replaced by its parent: an augmented row that happened to
+    reproduce held-out text would leak the evaluation into selection through the same
+    back door that round 3 had to close for training rows.
+    """
+    streams = {"shuffle": (augment, random.Random(seed + 20260919)),
+               "markers": (corpus2.augment_markers, random.Random(seed + 20260920)),
+               "boundary": (corpus2.augment_boundary, random.Random(seed + 20260921))}
+    views = {"plain": list(validation_rows)}
+    counts = {name: {"accepted": 0, "fell_back": 0, "leakage_guarded": 0}
+              for name in streams}
+    for name, (transform, rng) in streams.items():
+        derived = []
+        for row in validation_rows:
+            candidate = transform(row, rng)
+            if candidate is None:
+                counts[name]["fell_back"] += 1
+                derived.append(row)
+                continue
+            if candidate["query"] in forbidden_queries:
+                counts[name]["leakage_guarded"] += 1
+                derived.append(row)
+                continue
+            counts[name]["accepted"] += 1
+            derived.append(candidate)
+        views[name] = derived
+    return views, counts
+SCORE_SPLITS = ("validation", "easy", "heldout-layout", "heldout-lexical", "heldout-value",
+                "heldout-negative", "probe-base", "probe-layout", "probe-script",
+                "probe-value", "probe-episode-title")
+
+
+def save_json(path, value):
+    Path(path).write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n")
+
+
+def resolve_device(requested):
+    if requested == "auto":
+        return torch.device("cuda" if torch.cuda.is_available() else "cpu")
+    if requested == "cuda" and not torch.cuda.is_available():
+        raise SystemExit("--device cuda requested, but torch.cuda.is_available() is false")
+    return torch.device(requested)
+
+
+def on_device(batch, device):
+    return tuple(tensor.to(device, non_blocking=device.type == "cuda") for tensor in batch)
+
+
+def cpu_state_dict(model):
+    """Portable checkpoint: never make a local rescore depend on CUDA existing."""
+    return {name: value.detach().cpu() for name, value in model.state_dict().items()}
+
+
+class Extractor(nn.Module):
+    def __init__(self, mode, n_tags):
+        super().__init__()
+        self.mode, self.tokenizer = mode, None
+        if mode != "bytes":
+            sys.path.insert(0, str(SNAP))
+            from anifilebert.model import load_model
+            from anifilebert.tokenizer import load_tokenizer
+            self.tokenizer = load_tokenizer(str(SNAP))
+            if mode == "pretrained":
+                self.encoder = load_model(str(SNAP)).bert
+            else:
+                config = BertConfig.from_pretrained(str(SNAP), local_files_only=True)
+                self.encoder = BertModel(config, add_pooling_layer=False)
+        else:
+            config = BertConfig(vocab_size=257, hidden_size=256, num_hidden_layers=4,
+                                num_attention_heads=8, intermediate_size=1024,
+                                max_position_embeddings=512, pad_token_id=256)
+            self.encoder = BertModel(config, add_pooling_layer=False)
+        self.tags = nn.Linear(256, n_tags)
+
+    def encode(self, row):
+        """Tokens plus the [start, end) token window of every character.
+
+        Byte mode widens the window for multi-byte characters, so a boundary can
+        only ever fall between complete UTF-8 characters.
+        """
+        query = row["query"]
+        if self.mode == "bytes":
+            ids, starts, ends, pos = list(query.encode()), [], [], 0
+            for char in query:
+                starts.append(pos)
+                pos += len(char.encode())
+                ends.append(pos)
+        else:
+            tokens = self.tokenizer.convert_tokens_to_ids(self.tokenizer.tokenize(query))
+            ids = [self.tokenizer.cls_token_id] + tokens + [self.tokenizer.sep_token_id]
+            starts, ends = list(range(1, len(query) + 1)), list(range(2, len(query) + 2))
+            assert len(tokens) == len(query), (
+                "char-window pooling assumes one token per character; the tokenizer "
+                "changed. Refusing to train on misaligned labels.")
+        assert len(ids) <= self.encoder.config.max_position_embeddings, "No truncation"
+        labels = [0] * len(query)
+        for span in row["spans"]:
+            start, end = span["start"], span["end"]
+            labels[start] = LABELS.index("B-" + span["kind"])
+            for i in range(start + 1, end):
+                labels[i] = LABELS.index("I-" + span["kind"])
+        return ids, starts, ends, labels, row
+
+    def forward(self, batch):
+        ids, attention, starts, ends = batch
+        hidden = self.encoder(input_ids=ids, attention_mask=attention).last_hidden_state
+        cumulative = torch.cat((torch.zeros_like(hidden[:, :1]), hidden.cumsum(1)), 1)
+        a = starts.unsqueeze(-1).expand(-1, -1, hidden.shape[-1])
+        b = ends.unsqueeze(-1).expand_as(a)
+        width = (ends - starts).clamp_min(1).unsqueeze(-1)
+        chars = (cumulative.gather(1, b) - cumulative.gather(1, a)) / width
+        # Flatten before the tag layer: DynamicQuantizedLinear, which is what the
+        # INT8 export installs, only accepts 2-D input, so a rank-3 call would make
+        # the shipped artifact crash while the fp32 prototype scored fine.
+        tagged = self.tags(chars.reshape(-1, chars.shape[-1]))
+        return tagged.reshape(*chars.shape[:2], -1), chars
+
+
+def collate(items, pad):
+    length = max(len(item[0]) for item in items)
+    chars = max(len(item[1]) for item in items)
+    ids = torch.full((len(items), length), pad, dtype=torch.long)
+    attention = torch.zeros_like(ids)
+    starts = torch.zeros((len(items), chars), dtype=torch.long)
+    ends, labels = starts.clone(), torch.full((len(items), chars), -100, dtype=torch.long)
+    for i, (tokens, a, b, y, _) in enumerate(items):
+        ids[i, :len(tokens)] = torch.tensor(tokens)
+        attention[i, :len(tokens)] = 1
+        starts[i, :len(a)], ends[i, :len(b)] = torch.tensor(a), torch.tensor(b)
+        labels[i, :len(y)] = torch.tensor(y)
+    return (ids, attention, starts, ends), labels
+
+
+def legality_matrix():
+    """Zero where a BIO jump is legal, -inf where it is not.
+
+    This is the strict control for the empirical table below: a constraint says
+    which sequences are admissible and lets the emissions decide among them. The
+    empirical matrix additionally supplies a learned ordering prior, so its effect
+    must be measured separately rather than confused with BIO legality.
+    """
+    allow = torch.zeros(len(LABELS), len(LABELS))
+    for source, from_tag in enumerate(LABELS):
+        for target, to_tag in enumerate(LABELS):
+            if to_tag.startswith("I-") and from_tag not in (to_tag, "B-" + to_tag[2:]):
+                allow[source, target] = -torch.inf
+    return allow
+
+
+def transitions(items):
+    """Empirical log-transition matrix over the tag sequence.
+
+    This answers a different question than `legality_matrix`: whether the
+    *distribution* of arrangements the model was trained on supplies a useful tag
+    order. It is not the default because it can trade transfer for fit; `--transition`
+    keeps that factor explicit.
+
+    Greedy per-position argmax is allowed to emit an `I-X` that opens a span, and
+    the legalizer silently throws those away. A transition table makes the model
+    answer for that: it costs one matrix, needs no retraining, and can be applied
+    to an existing checkpoint, which is why this is the cheapest hypothesis here.
+    """
+    counts = torch.zeros(len(LABELS), len(LABELS))
+    for item in items:
+        previous = LABELS.index("O")
+        for label in item[3]:
+            counts[previous, label] += 1
+            previous = label
+        counts[previous, LABELS.index("O")] += 1
+    probabilities = counts / counts.sum(1, keepdim=True).clamp_min(1)
+    return torch.log(probabilities + 1e-12)
+
+
+def viterbi(logits, transition):
+    """Exact best tag sequence under a transition matrix. Legal in the sense the
+    matrix defines it: an impossible jump is -inf, not a silent drop.
+
+    The reduction is over the *source* axis. `scores` is indexed [from, to], so the
+    best predecessor of each target is `max(0)`; an earlier version of this function
+    took `max(1)`, which collapsed targets instead of sources and returned an all-`O`
+    path on every split - silently scoring 0 spans on rows the same checkpoint got
+    311/311 right. The test `test_viterbi_returns_one_state_per_frame_and_repairs_an_orphan`
+    exists because a bug that makes a decoder *abstain* looks like a cautious model
+    in an aggregate table.
+    """
+    best = torch.full((len(LABELS),), -torch.inf)
+    best[LABELS.index("O")] = 0.0
+    back = []
+    for frame in logits:
+        scores = best.unsqueeze(1) + transition  # [from, to]
+        back.append(scores.argmax(0))            # best predecessor for each target
+        best = frame + scores.max(0).values      # score of the best path ending at each target
+    path = [int(best.argmax())]
+    for step in reversed(back):
+        path.append(int(step[path[-1]]))
+    path.reverse()
+    # One state per frame, not n + 1: the recursion's seed state is not a tag. An
+    # earlier version returned it, so every frame was scored with its neighbour's
+    # label - which in a span task shifts every boundary by one character.
+    return path[1:]
+
+
+def viterbi_batch(logits, lengths, transition):
+    """Vectorized equivalent of `viterbi`, with a distinct end for every row."""
+    batch, frames, labels = logits.shape
+    lengths = torch.as_tensor(lengths, device=logits.device)
+    best = torch.full((batch, labels), -torch.inf, device=logits.device,
+                      dtype=logits.dtype)
+    best[:, LABELS.index("O")] = 0.0
+    back = []
+    for index in range(frames):
+        scores = best.unsqueeze(2) + transition.unsqueeze(0)
+        predecessor = scores.argmax(1)
+        candidate = logits[:, index] + scores.max(1).values
+        active = (index < lengths).unsqueeze(1)
+        best = torch.where(active, candidate, best)
+        back.append(predecessor)
+    current = best.argmax(1)
+    paths = torch.zeros((batch, frames), dtype=torch.long, device=logits.device)
+    for index in reversed(range(frames)):
+        active = index < lengths
+        paths[:, index] = torch.where(active, current, paths[:, index])
+        predecessor = back[index].gather(1, current.unsqueeze(1)).squeeze(1)
+        current = torch.where(active, predecessor, current)
+    return [paths[row, :int(length)].tolist() for row, length in enumerate(lengths)]
+
+
+def decode(tag_ids, limit):
+    """BIO to spans. An `I-` that does not continue its own kind ends the span, and
+    an orphan `I-` becomes nothing: the decoder never invents a boundary."""
+    spans, kind, begin = [], None, 0
+    for i, tag in enumerate([LABELS[t] for t in tag_ids[:limit]] + ["O"]):
+        if kind is not None and tag != "I-" + kind:
+            spans.append((kind, begin, i))
+            kind = None
+        if tag.startswith("B-"):
+            kind, begin = tag[2:], i
+    return spans
+
+
+def predict_spans(query, tag_ids):
+    """Decoder output as contract spans, with no text and no value: the record is
+    whatever `fp.normalize` can derive from the boundaries alone."""
+    return [{"kind": kind, "start": start, "end": end}
+            for kind, start, end in decode(tag_ids, len(query))]
+
+
+def tag_paths(logits, lengths, decode_mode="greedy", transition=None):
+    """Decode only real character frames, never another row's batch padding.
+
+    Viterbi backtracking starts at its final frame, so decoding the padded suffix
+    can change the state selected for the final real character. Greedy happened to
+    be insensitive because `predict_spans` truncates, but both decoders should be
+    explicitly batch-composition invariant.
+    """
+    if decode_mode == "greedy":
+        return [frame[:length].argmax(-1).tolist()
+                for frame, length in zip(logits, lengths)]
+    return viterbi_batch(logits, lengths, transition)
+
+
+def label_weights(items):
+    """Inverse frequency, sqrt-damped, mean 1. DISCARD and TITLE dominate the
+    character budget, and without this the rare kinds are unlearnable.
+
+    Keys are label *ids*: returning them indexed by name silently produced all
+    zeros, which made the loss NaN and the exported artifact crash.
+    """
+    counts = Counter(label for item in items for label in item[3] if label >= 0)
+    total = sum(counts.values())
+    raw = {label_id: 1 / math.sqrt(count / total * len(LABELS))
+           for label_id, count in counts.items()}
+    mean = sum(raw[label] * counts[label] for label in raw) / total
+    return [raw.get(index, 0.0) / mean for index in range(len(LABELS))]
+
+
+@torch.inference_mode()
+def evaluate(model, encoded, batch_size=64, timed=False, decode_mode="greedy", transition=None):
+    model.eval()
+    device = next(model.parameters()).device
+    counts, bylang, bylayout, predictions, latencies = Counter(), {}, {}, [], []
+    per_kind = defaultdict(Counter)
+    start = time.perf_counter()
+    for offset in range(0, len(encoded), batch_size):
+        items = encoded[offset:offset + batch_size]
+        batch, _ = collate(items, model.encoder.config.pad_token_id)
+        batch = on_device(batch, device)
+        logits, _ = model(batch)
+        # Decoding is tiny and intentionally stays on CPU, where Viterbi's Python
+        # loop and the deterministic contract resolver already operate.
+        logits = logits.detach().cpu()
+        tags = tag_paths(logits, [len(item[1]) for item in items], decode_mode, transition)
+        for i, item in enumerate(items):
+            row = item[-1]
+            spans = predict_spans(row["query"], tags[i])
+            actual, issues = normalize.gated_record(row["query"], spans)
+            expected = row["record"]
+            exact = actual == expected
+            counts["n"] += 1
+            counts["exact"] += exact
+            counts["abstained"] += actual is None
+            counts["wrong_abstain"] += actual is None and expected is not None
+            counts["false_emit"] += actual is not None and expected is None
+            counts["issue_rows"] += bool(issues)
+            for span in row["spans"]:
+                per_kind[span["kind"]]["gold"] += 1
+            for span in spans:
+                per_kind[span["kind"]]["pred"] += 1
+            gold = {(s["kind"], s["start"], s["end"]) for s in row["spans"]}
+            pred = {(s["kind"], s["start"], s["end"]) for s in spans}
+            counts["span_tp"] += len(gold & pred)
+            counts["span_pred"], counts["span_gold"] = counts["span_pred"] + len(pred), counts["span_gold"] + len(gold)
+            for kind, _, _ in gold & pred:
+                per_kind[kind]["tp"] += 1
+            for key, value in (actual or {}).items():
+                counts["emitted_fields"] += 1
+                counts["wrong_emitted_fields"] += (expected or {}).get(key) != value
+            for key, value in (expected or {}).items():
+                counts["gold_fields"] += 1
+                # Compare prediction to gold. Writing `(expected or {})` on both
+                # sides made every gold field look correct and reported 100%.
+                counts["correct_fields"] += (actual or {}).get(key) == value
+            counts["correct_abstain"] += actual is None and expected is None
+            bylang.setdefault(row["language"], Counter())["n"] += 1
+            bylang[row["language"]]["exact"] += exact
+            slot = bylayout.setdefault(row["layout"], Counter())
+            slot["n"] += 1
+            slot["exact"] += exact
+            predictions.append({"query": row["query"], "layout": row["layout"],
+                                "language": row["language"], "expected": expected,
+                                "actual": actual, "spans": spans,
+                                "issues": issues, "exact": exact})
+            if timed:
+                latencies.append(1000 * single_row_ms(model, item))
+    elapsed = time.perf_counter() - start
+    out = dict(counts)
+    out["span_f1"] = f1(counts["span_tp"], counts["span_pred"], counts["span_gold"])
+    out["kinds"] = {kind: {"gold": c["gold"], "pred": c["pred"], "tp": c["tp"],
+                           "f1": f1(c["tp"], c["pred"], c["gold"])}
+                    for kind, c in sorted(per_kind.items())}
+    out["by_language"] = {k: dict(v) for k, v in sorted(bylang.items())}
+    out["by_layout"] = {k: dict(v) for k, v in sorted(bylayout.items())}
+    out["ms_per_row_batched"] = 1000 * elapsed / max(1, len(encoded))
+    if latencies:
+        out["latency_ms_p50"] = round(statistics.median(latencies), 2)
+        out["latency_ms_p95"] = round(sorted(latencies)[int(0.95 * len(latencies))], 2)
+    return out, predictions
+
+
+def single_row_ms(model, item):
+    began = time.perf_counter()
+    batch, _ = collate([item], model.encoder.config.pad_token_id)
+    device = next(model.parameters()).device
+    batch = on_device(batch, device)
+    logits, _ = model(batch)
+    torch.cuda.synchronize(device) if device.type == "cuda" else None
+    logits.argmax(-1)
+    return time.perf_counter() - began
+
+
+def f1(tp, pred, gold):
+    precision, recall = (tp / pred if pred else 0.0), (tp / gold if gold else 0.0)
+    return round(2 * precision * recall / (precision + recall), 4) if precision + recall else 0.0
+
+
+def train(args):
+    out = Path(args.out)
+    out.mkdir(parents=True, exist_ok=False)
+    torch.set_num_threads(args.threads)
+    random.seed(args.seed)
+    torch.manual_seed(args.seed)
+    torch.cuda.manual_seed_all(args.seed)
+    torch.use_deterministic_algorithms(True)
+    device = resolve_device(args.device)
+    train_splits = ARM_SPLITS[args.arm]
+    train_rows = [row for split in train_splits for row in rows(split)]
+    data = {split: rows(split) for split in SCORE_SPLITS}
+    scored_queries = set()
+    for value in data.values():
+        scored_queries |= {row["query"] for row in value}
+    leaked = scored_queries.intersection(row["query"] for row in train_rows)
+    assert not leaked, f"train rows appear in a scored split: {sorted(leaked)[:3]}"
+    if args.smoke:
+        train_rows, data = train_rows[:160], {k: v[:40] for k, v in data.items()}
+    model = Extractor(args.mode, len(LABELS))
+    encoded = {key: [model.encode(row) for row in value] for key, value in
+               list(data.items()) + [("train", train_rows)]}
+    model.to(device)
+    weights = torch.tensor(label_weights(encoded["train"]), dtype=torch.float, device=device)
+    transition = (legality_matrix() if args.transition == "legal"
+                  else transitions(encoded["train"]))
+    versions = {p: importlib.metadata.version(p) for p in ("torch", "transformers", "numpy")}
+    unknown = {}
+    if model.tokenizer:
+        unk = model.tokenizer.unk_token_id
+        # The audit number that matters is how much of the training text the
+        # vocabulary cannot represent at all, recorded per run.
+        tokens = [token for item in encoded["train"] for token in item[0]]
+        unknown = {"train": round(sum(t == unk for t in tokens) / len(tokens), 4)}
+    # Recorded verbatim so a published run can never be reinterpreted after the
+    # view list changes: `--augment balanced` means whatever this file says it meant.
+    views = SCHEDULES.get(args.augment)
+    selected_views, selection_counts = selection_views(data["validation"], args.seed,
+                                                       scored_queries)
+    selection_encoded = {name: [model.encode(row) for row in value]
+                         for name, value in selected_views.items()}
+    save_json(out / "config.json", {
+        "args": vars(args), "versions": versions, "corpus_sha256": digest(CORPUS / "manifest.json"),
+        "source_sha256s": {"parser2": digest(__file__),
+                           "corpus2": digest(Path(corpus2.__file__)),
+                           "normalize": digest(Path(normalize.__file__))},
+        "corpus_splits": train_splits, "train_rows": len(train_rows),
+        "parameters": sum(p.numel() for p in model.parameters()),
+        "source_sha256": digest(__file__), "encoder": model.encoder.config.to_dict(),
+        "snapshot": str(SNAP) if args.mode != "bytes" else None,
+        "pretraining_overlap": "unknown for upstream weights" if args.mode == "pretrained" else None,
+        "value_head": False, "unknown_token_fraction": unknown,
+        "device": {"requested": args.device, "resolved": str(device),
+                   "cuda": torch.version.cuda,
+                   "name": torch.cuda.get_device_name(device) if device.type == "cuda" else None},
+        "label_weights_min": round(float(weights.min()), 3),
+        "label_weights_max": round(float(weights.max()), 3),
+        "balanced_views": list(views) if views else None,
+        "selection": ("mean exact-record over fixed plain/shuffle/markers/boundary "
+                      "views of the validation split; earliest checkpoint breaks ties")
+        if args.select == "multi-view" else
+        "validation exact-record; earliest checkpoint breaks ties",
+        # Recorded before training so a reviewer can see that selection never read a
+        # held-out row, and can rebuild the same views from the seed.
+        "selection_view_counts": selection_counts,
+        "selection_view_queries": {name: [row["query"] for row in value]
+                                   for name, value in selected_views.items()}})
+    optimizer = torch.optim.AdamW([
+        {"params": model.encoder.parameters(), "lr": 5e-5 if args.mode == "pretrained" else 3e-4},
+        {"params": model.tags.parameters(), "lr": 3e-4}], weight_decay=0.01)
+    best, history, started = -1.0, [], time.perf_counter()
+    rng = random.Random(args.seed)
+    balanced_position_rng = random.Random(args.seed + 20260918)
+    value_rng = random.Random(args.seed + 20260916)
+    path_rng = random.Random(args.seed + 20260917)
+    marker_rng = random.Random(args.seed + 20260922)
+    boundary_rng = random.Random(args.seed + 20260923)
+    for epoch in range(args.epochs):
+        model.train()
+        items, augmentation = [], Counter()
+        for row_index, row in enumerate(train_rows):
+            view = row
+            view_mode = (balanced_view(row_index, epoch, args.seed, views)
+                         if args.augment in SCHEDULES else args.augment)
+            selected = (view_mode != "none" and
+                        (args.augment in SCHEDULES or rng.random() < args.augment_rate))
+            augmentation["scheduled_" + view_mode] += 1
+            if selected:
+                augmentation["selected"] += 1
+                # A different arrangement of the same atoms every epoch: the label
+                # multiset is unchanged unless this arm explicitly adds value
+                # support. Axis-specific RNGs do not perturb which positional view
+                # the paired round-2 control would have seen.
+                position_rng = balanced_position_rng if args.augment in SCHEDULES else rng
+                candidate = augment_training(row, view_mode, position_rng, value_rng, path_rng,
+                                             marker_rng, boundary_rng)
+                # The augmented view is built at train time, so it never passed the
+                # corpus dedupe. A view that lands on a scored query would be test
+                # leakage through the back door, and a shuffled negative row is the
+                # realistic way to hit one.
+                if candidate and candidate["query"] not in scored_queries:
+                    view = candidate
+                    augmentation["accepted"] += 1
+                    for axis in candidate.get("augmented_axes", []):
+                        augmentation[axis] += 1
+                elif candidate:
+                    augmentation["scored_query_collision"] += 1
+                else:
+                    augmentation["rejected"] += 1
+            items.append(model.encode(view))
+        rng.shuffle(items)
+        losses = []
+        for offset in range(0, len(items), args.batch):
+            minibatch = items[offset:offset + args.batch]
+            batch, labels = collate(minibatch, model.encoder.config.pad_token_id)
+            batch, labels = on_device(batch, device), labels.to(device, non_blocking=True)
+            logits, _ = model(batch)
+            loss = nn.functional.cross_entropy(logits.reshape(-1, len(LABELS)),
+                                               labels.reshape(-1), weight=weights)
+            if not torch.isfinite(loss):
+                raise SystemExit(f"non-finite loss at epoch {epoch + 1} step {offset}: "
+                                 f"a zero/nan weight or lr is producing a model that only "
+                                 f"looks trained")
+            optimizer.zero_grad()
+            loss.backward()
+            nn.utils.clip_grad_norm_(model.parameters(), 1.0)
+            optimizer.step()
+            losses.append(float(loss.detach()))
+        val, _ = evaluate(model, encoded["validation"], decode_mode=args.decode,
+                          transition=transition)
+        # The selection objective is the mean over the fixed derived views; the
+        # plain validation number stays recorded so rounds 3 and 4 remain readable.
+        view_scores = {}
+        for name in SELECTION_VIEWS:
+            if name == "plain":
+                view_scores[name] = val["exact"]
+                continue
+            scored_view, _ = evaluate(model, selection_encoded[name],
+                                      decode_mode=args.decode, transition=transition)
+            view_scores[name] = scored_view["exact"]
+        objective = (sum(view_scores.values()) /
+                     (len(SELECTION_VIEWS) * max(1, val["n"])))
+        entry = {"epoch": epoch + 1, "loss": sum(losses) / len(losses),
+                  "validation_exact": val["exact"], "validation_span_f1": val["span_f1"],
+                  "validation_kinds": {k: v["f1"] for k, v in val["kinds"].items()},
+                  "selection_objective": round(objective, 4),
+                  "selection_views": view_scores,
+                  "augmentation": dict(sorted(augmentation.items())),
+                  "elapsed_s": round(time.perf_counter() - started, 1)}
+        history.append(entry)
+        save_json(out / "history.json", history)
+        print(f"{args.arm} {args.mode} {args.seed} ep{epoch + 1} "
+              f"val {val['exact']}/{val['n']} sel {objective:.3f} f1 {val['span_f1']} "
+              f"{entry['elapsed_s']}s", flush=True)
+        chosen = objective if args.select == "multi-view" else val["exact"] / max(1, val["n"])
+        if chosen > best:
+            best = chosen
+            torch.save(cpu_state_dict(model), out / "best.pt")
+    model.load_state_dict(torch.load(out / "best.pt", map_location="cpu", weights_only=True))
+    results = {}
+    for split in SCORE_SPLITS:
+        scored, preds = evaluate(model, encoded[split], timed=split == "easy",
+                                 decode_mode=args.decode, transition=transition)
+        results[split] = scored
+        (out / f"{split}.jsonl").write_text("".join(
+            json.dumps(p, ensure_ascii=False) + "\n" for p in preds))
+    # Purely positional probe, derived at evaluation time from `probe-base`: the
+    # same atoms in a new arrangement, so unlike `probe-layout` it introduces no
+    # unseen marker vocabulary. `heldout-layout` confounds the two, and round 1
+    # showed that a zero there is ambiguous between "format" and "lexicon".
+    # Gold is the parent record, which `augment` proves equal before returning.
+    probe_rng = random.Random(args.seed + 20260915)
+    base = data["probe-base"]
+    views = [(augment(row, probe_rng) or row) for row in base]
+    scored, preds = evaluate(model, [model.encode(row) for row in views])
+    scored["derived_rearranged"] = sum(1 for row, view in zip(base, views) if view is not row)
+    scored["derived_from"] = "probe-base re-arranged by corpus2.augment at eval time"
+    results["probe-shuffle"] = scored
+    (out / "probe-shuffle.jsonl").write_text("".join(
+        json.dumps(p, ensure_ascii=False) + "\n" for p in preds))
+    results["training_seconds"] = round(time.perf_counter() - started, 1)
+    results["checkpoint_sha256"] = digest(out / "best.pt")
+    # `best` is whichever objective was registered: a count for seen-layout
+    # validation, a rate for the multi-view mean. Reporting both explicitly keeps a
+    # round-4 and a round-5 history readable without pretending they are the same
+    # number.
+    results["selected_checkpoint_objective"] = {"mode": args.select,
+                                                "value": round(best, 4)}
+    results["selection_view_counts"] = selection_counts
+    export(model, args, out, encoded, results)
+    save_json(out / "results.json", results)
+    print(json.dumps({split: {"exact": f"{results[split]['exact']}/{results[split]['n']}",
+                               "span_f1": results[split]["span_f1"]}
+                      for split in ("validation", "easy", "heldout-layout", "heldout-lexical",
+                                    "heldout-value", "heldout-negative", "probe-base",
+                                    "probe-layout", "probe-shuffle")}, indent=2), flush=True)
+
+
+def export(model, args, out, encoded, results):
+    """Evaluate the artifact that would actually ship, not the fp32 prototype."""
+    # Dynamic INT8 is a CPU deployment format. The checkpoint is deliberately
+    # portable and all artifact comparisons happen after the same fp32 model is
+    # returned to CPU, regardless of where training ran.
+    model.cpu().eval()
+    fp32 = out / "model.fp32.pt"
+    torch.save(model.state_dict(), fp32)
+    quantized = torch.quantization.quantize_dynamic(model, {nn.Linear}, dtype=torch.qint8)
+    qpath = out / "model.int8.pt"
+    torch.save(quantized.state_dict(), qpath)
+    sizes = {"fp32_mib": round(fp32.stat().st_size / 2 ** 20, 2),
+             "int8_mib": round(qpath.stat().st_size / 2 ** 20, 2)}
+    rows_sample = encoded["easy"][:64]
+    scored, _ = evaluate(quantized, rows_sample, timed=False)
+    # Like-for-like: the quantized number only means anything next to the fp32
+    # number on the same rows, which is the comparison the pilot had to add.
+    baseline, _ = evaluate(model, rows_sample)
+    # Quantization is scored on held-out formats too, not only on in-distribution
+    # rows: a drop that only appears on unseen layouts would be invisible here
+    # otherwise, and the shipped artifact is the one that has to answer for it.
+    unseen, _ = evaluate(quantized, encoded["heldout-layout"])
+    results["export"] = {**sizes, "int8_subset_n": scored["n"],
+                         "int8_subset_exact": scored["exact"],
+                         "fp32_subset_exact": baseline["exact"],
+                         "int8_span_f1_subset": scored["span_f1"],
+                         "fp32_span_f1_subset": baseline["span_f1"],
+                         "int8_heldout_layout_exact": f"{unseen['exact']}/{unseen['n']}",
+                         "fp32_heldout_layout_exact": f"{results['heldout-layout']['exact']}/"
+                                                      f"{results['heldout-layout']['n']}",
+                         "int8_latency_p50": None, "int8_latency_p95": None}
+    with torch.inference_mode():
+        latencies = [1000 * single_row_ms(quantized, item) for item in rows_sample]
+    results["export"]["int8_latency_p50"] = round(statistics.median(latencies), 2)
+    results["export"]["int8_latency_p95"] = round(sorted(latencies)[int(0.95 * len(latencies))], 2)
+
+
+def score_artifact(args):
+    """Score the *shipped* INT8 file on the full battery, under both decoders.
+
+    Training writes `model.int8.pt`, but until now only a 64-row subset and
+    heldout-layout were ever read from that file: the artifact comparisons in the
+    documentation were re-quantizations of the checkpoint on the scoring machine,
+    not the bytes that would ship. This loads the exported file and states explicitly
+    whether it equals a fresh local re-quantization, so a reader knows which object
+    produced each number.
+    """
+    out = Path(args.score_artifact)
+    config = json.loads((out / "config.json").read_text())
+    target = out / f"artifact-int8-{args.transition}.json"
+    if target.exists():
+        raise SystemExit(f"refusing to overwrite {target}")
+    shipped_path = out / "model.int8.pt"
+    if not shipped_path.exists():
+        raise SystemExit(f"{out} has no exported artifact to score")
+    torch.set_num_threads(args.threads)
+    mode = config["args"]["mode"]
+
+    def quantized_model():
+        # Dynamic INT8 has no state_dict-compatible loader for an unquantized module:
+        # quantize first, then load the shipped weights into the same structure.
+        candidate = torch.quantization.quantize_dynamic(Extractor(mode, len(LABELS)),
+                                                       {nn.Linear}, dtype=torch.qint8)
+        candidate.load_state_dict(torch.load(shipped_path, map_location="cpu",
+                                             weights_only=False))
+        return candidate.eval()
+
+    def decode_tags(network, items, decode_mode="greedy", transition=None):
+        """The tag sequences a model emits for `items`, for behaviour comparison."""
+        network.eval()
+        tags = []
+        with torch.no_grad():
+            for offset in range(0, len(items), 64):
+                chunk = items[offset:offset + 64]
+                batch, _ = collate(chunk, network.encoder.config.pad_token_id)
+                logits, _ = network(batch)
+                tags += tag_paths(logits.detach().cpu(), [len(item[1]) for item in chunk],
+                                  decode_mode, transition)
+        return tags
+
+    def checkpoint():
+        fresh = Extractor(mode, len(LABELS))
+        fresh.load_state_dict(torch.load(out / "best.pt", map_location="cpu",
+                                         weights_only=True))
+        return fresh.eval()
+
+    shipped = quantized_model()
+    reference = checkpoint()
+    # A quantized `state_dict` holds packed params (tuples, dtypes), so comparing its
+    # internals compares a serialization format. What a reader needs to know is
+    # whether the shipped bytes *behave* like a fresh re-quantization of this
+    # checkpoint, so the equality claim is made about predictions.
+    sample = [reference.encode(row) for row in rows("easy")[:32]]
+    shipped_tags = decode_tags(shipped, sample)
+    reference_tags = decode_tags(reference, sample)
+    requant_tags = decode_tags(torch.quantization.quantize_dynamic(checkpoint(),
+                                                                   {nn.Linear},
+                                                                   dtype=torch.qint8),
+                              sample)
+    train_items = [reference.encode(row)
+                   for split in ARM_SPLITS[config["args"]["arm"]] for row in rows(split)]
+    transition = (legality_matrix() if args.transition == "legal"
+                  else transitions(train_items))
+    encoded = {split: [reference.encode(row) for row in rows(split)]
+               for split in SCORE_SPLITS}
+    # Same seed formula and row order as the trainer, so `probe-shuffle` here is the
+    # same measurement the fp32 rescores report.
+    probe_rng = random.Random(config["args"]["seed"] + 20260915)
+    base = rows("probe-base")
+    views = [(augment(row, probe_rng) or row) for row in base]
+    encoded["probe-shuffle"] = [reference.encode(row) for row in views]
+    result = {"checkpoint_sha256": digest(out / "best.pt"),
+              "artifact_sha256": digest(shipped_path),
+              "artifact_mib": round(shipped_path.stat().st_size / 2 ** 20, 2),
+              "shipped_matches_requantization": shipped_tags == requant_tags,
+              "shipped_matches_fp32_on_sample": shipped_tags == reference_tags,
+              "sample_rows": len(sample), "transition": args.transition, "decodes": {}}
+    for decode_mode in ("greedy", "viterbi"):
+        result["decodes"][decode_mode] = {
+            split: evaluate(shipped, items, decode_mode=decode_mode,
+                            transition=transition)[0]
+            for split, items in encoded.items()}
+    save_json(target, result)
+    print(json.dumps({"artifact_sha256": result["artifact_sha256"][:12],
+                      "shipped_matches_requantization":
+                          result["shipped_matches_requantization"],
+                      "shipped_matches_fp32_on_sample":
+                          result["shipped_matches_fp32_on_sample"],
+                      "greedy": {split: f"{scores['exact']}/{scores['n']}" for split, scores
+                                 in result["decodes"]["greedy"].items()},
+                      "viterbi": {split: f"{scores['exact']}/{scores['n']}" for split, scores
+                                  in result["decodes"]["viterbi"].items()
+                                  if split in ("validation", "heldout-layout",
+                                               "heldout-lexical", "probe-value")}}, indent=2))
+
+
+def rescore(args):
+    """Re-score an existing checkpoint with a different decoder: no retraining, so
+    the comparison isolates decoding from everything else. Also checks that the
+    greedy path reproduces the run's recorded numbers before trusting the delta."""
+    out = Path(args.rescore)
+    config = json.loads((out / "config.json").read_text())
+    target = out / f"rescore-{args.transition}.json"
+    if target.exists():
+        raise SystemExit(f"refusing to overwrite {target}")
+    torch.set_num_threads(args.threads)
+    device = resolve_device(args.device)
+    model = Extractor(config["args"]["mode"], len(LABELS))
+    model.load_state_dict(torch.load(out / "best.pt", map_location="cpu", weights_only=True))
+    model.to(device)
+    train_items = [model.encode(row)
+                   for split in ARM_SPLITS[config["args"]["arm"]] for row in rows(split)]
+    transition = (legality_matrix() if args.transition == "legal"
+                  else transitions(train_items))
+    encoded = {split: [model.encode(row) for row in rows(split)] for split in SCORE_SPLITS}
+    # The positional probe added after round 1, computed here so a checkpoint
+    # trained before it exists can still be compared on it. Same seed formula and
+    # same row order as the trainer, so the numbers are the same measurement.
+    probe_rng = random.Random(config["args"]["seed"] + 20260915)
+    base = rows("probe-base")
+    views = [(augment(row, probe_rng) or row) for row in base]
+    shuffle_items = [model.encode(row) for row in views]
+    result = {"checkpoint_sha256": digest(out / "best.pt"),
+              "args": config["args"],
+              "transition": args.transition,
+              "probe_shuffle_rearranged": sum(1 for row, view in zip(base, views)
+                                              if view is not row),
+              "decodes": {}}
+    for mode in ("greedy", "viterbi"):
+        result["decodes"][mode] = {
+            split: evaluate(model, encoded[split],
+                            decode_mode=mode, transition=transition)[0]
+            for split in SCORE_SPLITS}
+        result["decodes"][mode]["probe-shuffle"] = evaluate(
+            model, shuffle_items, decode_mode=mode, transition=transition)[0]
+    recorded = json.loads((out / "results.json").read_text())
+    result["greedy_matches_recorded"] = all(
+        result["decodes"]["greedy"][split]["exact"] == recorded[split]["exact"]
+        for split in SCORE_SPLITS)
+    result["delta_viterbi_minus_greedy_exact"] = {
+        split: result["decodes"]["viterbi"][split]["exact"]
+        - result["decodes"]["greedy"][split]["exact"] for split in SCORE_SPLITS}
+    save_json(target, result)
+    print(json.dumps({"greedy_matches_recorded": result["greedy_matches_recorded"],
+                      "probe_shuffle_rearranged": result["probe_shuffle_rearranged"],
+                      "probe_shuffle": {mode: f"{scores['probe-shuffle']['exact']}/"
+                                              f"{scores['probe-shuffle']['n']}"
+                                        for mode, scores in result["decodes"].items()},
+                      "delta": result["delta_viterbi_minus_greedy_exact"]}, indent=2))
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--arm", choices=["A-narrow", "B-broad"], default="B-broad")
+    parser.add_argument("--mode", choices=["bytes", "characters", "pretrained"], default="characters")
+    parser.add_argument("--seed", type=int, default=0)
+    parser.add_argument("--epochs", type=int, default=6)
+    parser.add_argument("--batch", type=int, default=32)
+    parser.add_argument("--threads", type=int, default=3)
+    parser.add_argument("--device", choices=["auto", "cpu", "cuda"], default="auto")
+    parser.add_argument("--decode", choices=["greedy", "viterbi"], default="greedy")
+    parser.add_argument("--transition", choices=["legal", "empirical"], default="legal",
+                        help="which matrix Viterbi uses: BIO admissibility only "
+                             "(default) or empirical tag-transition frequencies")
+    parser.add_argument("--augment", choices=["none", "shuffle", "shuffle-values",
+                                               "shuffle-paths", "shuffle-rich", "balanced4",
+                                               "balanced", "balanced-dose", "shuffle-markers",
+                                               "shuffle-boundary"],
+                        default="none",
+                        help="shuffle atom arrangement per epoch; breaks positional cues. "
+                             "`balanced` is the current six-view schedule; `balanced4` is "
+                             "round 4's frozen four-view list")
+    parser.add_argument("--select", choices=["validation", "multi-view"], default="validation",
+                        help="checkpoint objective: seen-layout validation exact-record "
+                             "(rounds 1-4) or the mean over fixed derived validation views")
+    parser.add_argument("--augment-rate", type=float, default=0.5,
+                        help="share of training rows re-arranged each epoch")
+    parser.add_argument("--rescore", help="re-score a run directory with both decoders")
+    parser.add_argument("--score-artifact",
+                        help="score a run's exported model.int8.pt on every split under "
+                             "both decoders, and report whether it matches a fresh local "
+                             "re-quantization of the checkpoint")
+    parser.add_argument("--smoke", action="store_true",
+                        help="tiny subsets to check plumbing; never a measurement")
+    parser.add_argument("--out")
+    args = parser.parse_args()
+    if args.rescore:
+        return rescore(args)
+    if args.score_artifact:
+        return score_artifact(args)
+    if not args.out:
+        parser.error("--out is required unless --rescore or --score-artifact is given")
+    train(args)
+
+
+if __name__ == "__main__":
+    main()

app/fp/parser3_model.py

@@ -0,0 +1,420 @@
+"""Lossless character inputs and complete-span heads for the v3 controls.
+
+The encoder still sees one position per Unicode scalar, so UTF-8 fallback does
+not consume a different positional budget or split an output character. Existing
+vocabulary IDs and embeddings are preserved. Unknown scalars have a reversible
+integer ID and four ordered UTF-8 byte slots, never the shared UNK ID alone.
+"""
+
+import numpy as np
+import torch
+from torch import nn
+
+from fp import charclass
+from fp.parser2 import Extractor, KINDS, LABELS, predict_spans, tag_paths, legality_matrix
+
+TYPES = ["O", *KINDS]
+
+
+def utf8_slots(codepoints):
+    """UTF-8 bytes + 1, padded with zero to four slots (tensorized, ordered)."""
+    cp = codepoints.clamp_min(0)
+    two, three, four = cp >= 0x80, cp >= 0x800, cp >= 0x10000
+    first = torch.where(four, 0xF0 | (cp >> 18), torch.where(
+        three, 0xE0 | (cp >> 12), torch.where(two, 0xC0 | (cp >> 6), cp)))
+    second = torch.where(four, 0x80 | ((cp >> 12) & 63), torch.where(
+        three, 0x80 | ((cp >> 6) & 63), 0x80 | (cp & 63)))
+    third = torch.where(four, 0x80 | ((cp >> 6) & 63), 0x80 | (cp & 63))
+    fourth = 0x80 | (cp & 63)
+    return torch.stack((first + 1, torch.where(two, second + 1, 0),
+                        torch.where(three, third + 1, 0),
+                        torch.where(four, fourth + 1, 0)), dim=-1)
+
+
+class Parser(Extractor):
+    def __init__(self, representation="unicode", head="bio", *, encoder=None, tokenizer=None,
+                 char_classes=False):
+        if representation not in ("legacy", "unicode", "subword") or head not in ("bio", "span"):
+            raise ValueError("unknown representation or head")
+        if encoder is None:
+            super().__init__("pretrained", len(LABELS))
+        else:
+            # Tiny injected encoders make alignment/export tests independent of downloads.
+            nn.Module.__init__(self)
+            self.encoder, self.tokenizer, self.mode = encoder, tokenizer, "pretrained"
+            self.tags = nn.Linear(encoder.config.hidden_size, len(LABELS))
+        self.representation, self.head = representation, head
+        self.base_vocab_size = self.encoder.config.vocab_size
+        hidden = self.encoder.config.hidden_size
+        if representation == "unicode":
+            if hidden % 4:
+                raise ValueError("hidden size must be divisible by four")
+            self.byte_slots = nn.ModuleList([nn.Embedding(257, hidden // 4, padding_idx=0)
+                                             for _ in range(4)])
+            for embedding in self.byte_slots:
+                nn.init.normal_(embedding.weight, std=0.02)
+                with torch.no_grad():
+                    embedding.weight[0].zero_()
+            self.byte_projection = nn.Linear(hidden, hidden, bias=False)
+            nn.init.eye_(self.byte_projection.weight)
+        if head == "span":
+            self.tags = nn.Linear(hidden, len(TYPES))
+            self.starts = nn.Linear(hidden, len(TYPES))
+            self.ends = nn.Linear(hidden, len(TYPES))
+        self.char_classes = char_classes
+        # (structural, other, fallback) identity-dropout rates; set by the trainer.
+        self.identity_dropout = (0.0, 0.0, 0.0)
+        if char_classes:
+            if representation != "unicode":
+                raise ValueError("character classes need the unicode representation")
+            # Zero-initialised: a model built on an existing encoder starts out
+            # computing exactly what it computed without class features.
+            self.category_embedding = nn.Embedding(charclass.N_CATEGORIES, hidden)
+            self.script_embedding = nn.Embedding(charclass.N_SCRIPTS, hidden)
+            nn.init.zeros_(self.category_embedding.weight)
+            nn.init.zeros_(self.script_embedding.weight)
+            # Stands in for a dropped character identity during training only.
+            self.anonymous = nn.Parameter(torch.randn(hidden) * 0.02)
+            category, script = charclass.tables()
+            id_cp = torch.full((self.base_vocab_size,), -1, dtype=torch.long)
+            for token, index in self.tokenizer.get_vocab().items():
+                if len(token) == 1 and index < self.base_vocab_size:
+                    id_cp[index] = ord(token)
+            # Lookup tables are rebuilt from code, never saved: the exported Go
+            # artifact carries the same assignment as ranges in its JSON sidecar.
+            self.register_buffer("id_codepoint", id_cp, persistent=False)
+            self.register_buffer("category_table", torch.from_numpy(category.astype("int64")),
+                                 persistent=False)
+            self.register_buffer("script_table", torch.from_numpy(script.astype("int64")),
+                                 persistent=False)
+            structural = torch.zeros(charclass.N_CATEGORIES, dtype=torch.bool)
+            structural[list(charclass.STRUCTURAL)] = True
+            self.register_buffer("structural_category", structural, persistent=False)
+
+    def fits(self, row):
+        """Encoder-window check: characters for character encoders, tokens for subword.
+
+        The frozen character encoder sees one position per code point plus the
+        two specials; a subword encoder sees the tokenizer's pieces. No row is
+        ever truncated, so the caller filters with this before encoding.
+        """
+        if self.representation == "subword":
+            tokens = self.tokenizer(row["query"], add_special_tokens=True)["input_ids"]
+            return len(tokens) <= self.encoder.config.max_position_embeddings
+        return len(row["query"]) + 2 <= self.encoder.config.max_position_embeddings
+
+    def encode(self, row):
+        if self.representation == "subword":
+            return self._encode_subword(row)
+        query = row["query"]
+        # encode() rejects surrogate code points as well as invalid UTF-8 text.
+        query.encode("utf-8", errors="strict")
+        tokens = self.tokenizer.tokenize(query)
+        if tokens != list(query):
+            raise ValueError("v3 requires a character tokenizer with exact original offsets")
+        ids = self.tokenizer.convert_tokens_to_ids(tokens)
+        if self.representation == "unicode":
+            ids = [self.base_vocab_size + ord(char) if token == self.tokenizer.unk_token_id
+                   else token for char, token in zip(query, ids)]
+        ids = [self.tokenizer.cls_token_id, *ids, self.tokenizer.sep_token_id]
+        if len(ids) > self.encoder.config.max_position_embeddings:
+            raise ValueError(f"{len(ids)} positions exceed encoder limit; no truncation")
+        labels = self._labels(query, row["spans"], row.get("partial", False))
+        return ids, list(range(1, len(query) + 1)), list(range(2, len(query) + 2)), labels, row
+
+    def _labels(self, query, spans, partial=False):
+        """BIO labels for one row, refusing invalid or misquoted spans.
+
+        Partial rows mark uncovered characters -100 so the loss ignores them:
+        the row asserts only what its spans cover, not that nothing else exists.
+        """
+        labels = [-100 if partial else 0] * len(query)
+        previous = 0
+        for span in spans:
+            a, b, kind = span["start"], span["end"], span["kind"]
+            if not 0 <= previous <= a < b <= len(query):
+                raise ValueError("invalid or overlapping span offsets")
+            previous = b
+            if span.get("text", query[a:b]) != query[a:b]:
+                raise ValueError("span text does not match original offsets")
+            labels[a] = LABELS.index("B-" + kind)
+            labels[a + 1:b] = [LABELS.index("I-" + kind)] * (b - a - 1)
+        return labels
+
+    def _encode_subword(self, row):
+        """Subword tokens with one character window per code point.
+
+        Each character shares the window of the token(s) covering it, so the
+        cumulative-mean pooling in forward() still yields exactly one feature per
+        character and the span head keeps character resolution. A character the
+        tokenizer never covers is an error, never a silent guess.
+        """
+        query = row["query"]
+        query.encode("utf-8", errors="strict")
+        encoding = self.tokenizer(query, add_special_tokens=True,
+                                  return_offsets_mapping=True)
+        ids, offsets = encoding["input_ids"], encoding["offset_mapping"]
+        if len(ids) > self.encoder.config.max_position_embeddings:
+            raise ValueError(f"{len(ids)} tokens exceed encoder limit; no truncation")
+        starts, ends = [], []
+        for index in range(len(query)):
+            first = last = None
+            for position, (start, stop) in enumerate(offsets):
+                if start == stop or not start <= index < stop:
+                    continue
+                if first is None:
+                    first = position
+                last = position
+            if first is not None:
+                starts.append((first, last + 1))
+            else:
+                starts.append(None)
+        if query and all(window is None for window in starts):
+            raise ValueError(f"tokenizer covers nothing in {query!r}")
+        # Word-piece tokenizers normalize some characters away (sentencepiece
+        # collapses repeated whitespace). Those characters inherit the nearest
+        # covered character's window so the character grid stays intact; the
+        # count is recorded on the model and reported in the run config.
+        uncovered = 0
+        for index in range(len(query)):
+            if starts[index] is not None:
+                continue
+            uncovered += 1
+            previous = next((starts[position] for position in range(index - 1, -1, -1)
+                             if starts[position] is not None), None)
+            following = next((starts[position] for position in range(index + 1, len(query))
+                              if starts[position] is not None), None)
+            starts[index] = previous or following
+        self.uncovered_chars = getattr(self, "uncovered_chars", 0) + uncovered
+        ends = [window[1] for window in starts]
+        starts = [window[0] for window in starts]
+        return (ids, starts, ends,
+                self._labels(query, row["spans"], row.get("partial", False)), row)
+
+    def character_embeddings(self, ids):
+        fallback = ids >= self.base_vocab_size
+        known = ids.masked_fill(fallback, self.tokenizer.unk_token_id)
+        embedded = self.encoder.embeddings.word_embeddings(known)
+        if self.representation == "legacy":
+            return embedded
+        slots = utf8_slots(ids - self.base_vocab_size)
+        composed = torch.cat([embedding(slots[..., i])
+                              for i, embedding in enumerate(self.byte_slots)], dim=-1)
+        composed = self.byte_projection(composed.reshape(-1, composed.shape[-1])).reshape_as(embedded)
+        embedded = torch.where(fallback.unsqueeze(-1), composed, embedded)
+        if not self.char_classes:
+            return embedded
+        codepoint = torch.where(fallback, ids - self.base_vocab_size,
+                                self.id_codepoint[ids.clamp(0, self.base_vocab_size - 1)])
+        is_char = codepoint >= 0
+        safe = codepoint.clamp(0, charclass.LIMIT - 1)
+        category = torch.where(is_char, self.category_table[safe], 0)
+        script = torch.where(is_char, self.script_table[safe], 0)
+        structural, other, unseen = self.identity_dropout
+        if self.training and (structural or other or unseen):
+            rate = torch.where(self.structural_category[category], structural, other)
+            rate = torch.where(fallback, torch.clamp(rate, min=unseen), rate)
+            dropped = is_char & (torch.rand(rate.shape, device=rate.device) < rate)
+            embedded = torch.where(dropped.unsqueeze(-1), self.anonymous.to(embedded.dtype),
+                                   embedded)
+        return embedded + self.category_embedding(category) + self.script_embedding(script)
+
+    def forward(self, batch):
+        ids, attention, starts, ends = batch
+        if self.representation in ("legacy", "subword"):
+            hidden = self.encoder(input_ids=ids, attention_mask=attention).last_hidden_state
+        else:
+            hidden = self.encoder(inputs_embeds=self.character_embeddings(ids),
+                                  attention_mask=attention).last_hidden_state
+        cumulative = torch.cat((torch.zeros_like(hidden[:, :1]), hidden.cumsum(1)), 1)
+        a = starts.unsqueeze(-1).expand(-1, -1, hidden.shape[-1])
+        b = ends.unsqueeze(-1).expand_as(a)
+        chars = (cumulative.gather(1, b) - cumulative.gather(1, a)) / (
+            ends - starts).clamp_min(1).unsqueeze(-1)
+        logits = self.tags(chars.reshape(-1, chars.shape[-1]))
+        return logits.reshape(*chars.shape[:2], -1), chars
+
+    def boundary_logits(self, chars):
+        flat = chars.reshape(-1, chars.shape[-1])
+        shape = (*chars.shape[:2], len(TYPES))
+        return self.starts(flat).reshape(shape), self.ends(flat).reshape(shape)
+
+
+def span_targets(labels):
+    """Type/start/end class targets, respecting adjacent same-type gold spans."""
+    valid = labels >= 0
+    types = torch.where(valid, (labels + 1) // 2, -100)
+    starts = torch.where(valid, torch.where(labels % 2 == 1, types, 0), -100)
+    following = torch.cat((labels[:, 1:], torch.zeros_like(labels[:, :1])), dim=1)
+    continues = (following == 2 * types) & (types > 0)
+    ends = torch.where(valid, torch.where((types > 0) & ~continues, types, 0), -100)
+    return types, starts, ends
+
+
+def class_weights(targets, size):
+    """Smoothed inverse-square-root weights; unseen classes never have zero loss."""
+    valid = targets[targets >= 0]
+    counts = torch.bincount(valid, minlength=size).float().add(1)
+    weights = counts.rsqrt()
+    return weights / (weights * counts).sum() * counts.sum()
+
+
+def loss(model, logits, chars, labels, weights, *, semicrf=0.0, lengths=None,
+         teacher=None, kd_weight=0.0, kd_temperature=1.0):
+    """Factorised span/BIO cross-entropy, plus the optional structured and KD terms.
+
+    `weights` entries may be None (uniform classes). `semicrf` adds the globally
+    normalised interval objective for fully labelled rows; `teacher` holds the
+    teacher's (type, start, end) logits for distillation on labelled characters.
+    """
+    if model.head == "bio":
+        return nn.functional.cross_entropy(logits.reshape(-1, len(LABELS)),
+                                           labels.reshape(-1), weight=weights[0])
+    start, end = model.boundary_logits(chars)
+    targets = span_targets(labels)
+    terms = [nn.functional.cross_entropy(pred.reshape(-1, len(TYPES)), target.reshape(-1),
+                                        weight=weight)
+             for pred, target, weight in zip((logits, start, end), targets, weights)]
+    total = sum(terms) / len(terms)
+    if semicrf:
+        total = total + semicrf * semicrf_loss(logits, start, end, labels, lengths)
+    if teacher is not None and kd_weight:
+        total = total + kd_weight * distillation_loss((logits, start, end), teacher, labels,
+                                                      kd_temperature)
+    return total
+
+
+def interval_potentials(type_logits, start_logits, end_logits):
+    """Per-kind log-odds against the null class: the scores `interval_decode` sums."""
+    return [x[..., 1:] - x[..., :1] for x in (type_logits, start_logits, end_logits)]
+
+
+def semicrf_log_partition(body, start, end, lengths):
+    """log sum over every segmentation of exp(score), batched.
+
+    A segmentation is background characters (score 0) and non-overlapping typed
+    spans scoring interior body + start + end, exactly the space `interval_decode`
+    maximises over. O(L^2 K) per row.
+    """
+    batch, length, kinds = body.shape
+    prefix = torch.cat((body.new_zeros(batch, 1, kinds), body.cumsum(1)), dim=1)
+    alpha = [body.new_zeros(batch)]
+    for stop in range(1, length + 1):
+        previous = torch.stack(alpha, dim=1)                       # [B, stop]
+        spans = (previous.unsqueeze(-1) + prefix[:, stop:stop + 1] - prefix[:, :stop]
+                 + start[:, :stop] + end[:, stop - 1:stop])        # [B, stop, K]
+        alpha.append(torch.logaddexp(alpha[-1], spans.reshape(batch, -1).logsumexp(-1)))
+    alpha = torch.stack(alpha, dim=1)
+    return alpha.gather(1, lengths.unsqueeze(1)).squeeze(1)
+
+
+def semicrf_loss(type_logits, start_logits, end_logits, labels, lengths):
+    """Per-character negative log-likelihood of the gold segmentation.
+
+    Rows carrying unlabelled characters (partial rows) have no single gold
+    segmentation and are left to the factorised terms.
+    """
+    positions = torch.arange(labels.shape[1], device=labels.device)
+    inside = positions.unsqueeze(0) < lengths.unsqueeze(1)
+    full = ~((labels < 0) & inside).any(1) & (lengths > 0)
+    if not full.any():
+        return type_logits.sum() * 0
+    body, start, end = interval_potentials(type_logits[full], start_logits[full],
+                                           end_logits[full])
+    labels, lengths = labels[full], lengths[full]
+    types, starts, ends = span_targets(labels)
+
+    def gathered(scores, target):
+        picked = scores.gather(2, (target.clamp_min(1) - 1).unsqueeze(-1)).squeeze(-1)
+        return (picked * (target > 0)).sum(1)
+
+    gold = gathered(body, types) + gathered(start, starts) + gathered(end, ends)
+    partition = semicrf_log_partition(body, start, end, lengths)
+    return ((partition - gold) / lengths.clamp_min(1)).mean()
+
+
+def distillation_loss(student, teacher, labels, temperature):
+    """Temperature-scaled KL to the teacher, on labelled characters only.
+
+    Characters a row leaves unlabelled are excluded, so no teacher prediction ever
+    stands in for a missing gold label.
+    """
+    mask = (labels >= 0).reshape(-1)
+    if not mask.any():
+        return student[0].sum() * 0
+    total = 0
+    for s, t in zip(student, teacher):
+        s = s.reshape(-1, s.shape[-1])[mask] / temperature
+        t = t.reshape(-1, t.shape[-1])[mask] / temperature
+        total = total + nn.functional.kl_div(s.log_softmax(-1), t.log_softmax(-1),
+                                             log_target=True, reduction="batchmean")
+    return total * temperature ** 2 / len(student)
+
+
+def record_confidence(type_logits, start_logits, end_logits, length):
+    """exp(score(best segmentation) - log Z): the semi-CRF probability of the decode.
+
+    A score for ranking and thresholds, not a calibrated probability unless the
+    model was trained with the structured term and checked for calibration.
+    """
+    body, start, end = interval_potentials(type_logits[:length].unsqueeze(0),
+                                           start_logits[:length].unsqueeze(0),
+                                           end_logits[:length].unsqueeze(0))
+    spans = interval_decode(type_logits[:length], start_logits[:length], end_logits[:length])
+    best = 0.0
+    for span in spans:
+        k = KINDS.index(span["kind"])
+        best += float(body[0, span["start"]:span["end"], k].sum()
+                      + start[0, span["start"], k] + end[0, span["end"] - 1, k])
+    partition = semicrf_log_partition(body, start, end,
+                                      torch.tensor([length], device=body.device))
+    return float(torch.exp(torch.tensor(best) - partition.cpu()[0]))
+
+
+def interval_decode(type_logits, start_logits, end_logits):
+    """Best non-overlapping complete spans, with an explicit all-background path.
+
+    A candidate's score is interior type-vs-background log odds plus type-vs-null
+    odds at its start and end. No vocabulary, delimiter, or position rule enters
+    decoding. Adjacent spans are legal, including two of the same type.
+    """
+    arrays = [x.detach().cpu().numpy() if isinstance(x, torch.Tensor) else np.asarray(x)
+              for x in (type_logits, start_logits, end_logits)]
+    if any(a.ndim != 2 or a.shape != arrays[0].shape for a in arrays):
+        raise ValueError("span logits must be aligned matrices")
+    if not all(np.isfinite(a).all() for a in arrays):
+        raise ValueError("non-finite span logits")
+    body, start, end = [a[:, 1:] - a[:, :1] for a in arrays]
+    length, kinds = body.shape
+    prefix = np.concatenate((np.zeros((1, kinds)), body.cumsum(0)), axis=0)
+    best, back = np.zeros(length + 1), [None] * (length + 1)
+    for stop in range(1, length + 1):
+        scores = (best[:stop, None] + prefix[stop] - prefix[:stop]
+                  + start[:stop] + end[stop - 1])
+        index = int(scores.argmax())
+        begin, kind = divmod(index, kinds)
+        if scores[begin, kind] > best[stop - 1]:
+            best[stop], back[stop] = scores[begin, kind], (begin, kind)
+        else:
+            best[stop] = best[stop - 1]
+    spans, stop = [], length
+    while stop:
+        if back[stop] is None:
+            stop -= 1
+        else:
+            begin, kind = back[stop]
+            spans.append({"kind": KINDS[kind], "start": begin, "end": stop})
+            stop = begin
+    return list(reversed(spans))
+
+
+def decode_batch(model, logits, chars, items, decoder):
+    lengths = [len(item[-1]["query"]) for item in items]
+    if model.head == "bio":
+        paths = tag_paths(logits.detach().cpu(), lengths, decoder, legality_matrix())
+        return [predict_spans(item[-1]["query"], path) for item, path in zip(items, paths)]
+    if decoder != "span":
+        raise ValueError("span head requires span decoding")
+    starts, ends = model.boundary_logits(chars)
+    return [interval_decode(logits[i, :n], starts[i, :n], ends[i, :n])
+            for i, n in enumerate(lengths)]

app/serve.py

@@ -0,0 +1,264 @@
+#!/usr/bin/env python3
+"""Local parser service for Lain.
+
+Serves one frozen filename parser over HTTP so a native or containerised Lain
+can identify release names through a model. Read-only: never writes to disk and
+holds exactly one artifact.
+
+    GET  /health                 -> {"status": "ok", ...}
+    POST /parse {"filename": …}  -> {"record": {...} | null, "spans": [...], ...}
+
+The record is derived by the same gated contract resolver used at evaluation
+time (``fp.normalize.gated_record``), so a null record is the model abstaining
+and the caller should fall back to its own providers.
+
+Usage:
+    serve.py --artifact <prefix>/artifact/v3mix-bio-s1 --snapshot <prefix>/anifilebert
+    serve.py --check "Some.Show.S01E02.1080p-GROUP.mkv"    # one-shot, prints JSON
+"""
+
+import argparse
+import hashlib
+import json
+import logging
+import os
+import pathlib
+import socket
+import sys
+import threading
+import time
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+import torch
+from torch import nn
+
+log = logging.getLogger("lain-parser")
+APP = pathlib.Path(__file__).resolve().parent
+MAX_BODY = 64 * 1024
+
+
+class Parser:
+    """One artifact, loaded once, guarded by a lock for the HTTP threads."""
+
+    def __init__(self, artifact, snapshot):
+        self.artifact = pathlib.Path(artifact).resolve()
+        self.snapshot = pathlib.Path(snapshot).resolve()
+        # The fp modules read the snapshot path at import time.
+        os.environ["ANIFILEBERT_SNAPSHOT"] = str(self.snapshot)
+        if str(APP) not in sys.path:
+            sys.path.insert(0, str(APP))
+        from fp import normalize
+        from fp.parser2 import collate
+        from fp.parser3_model import Parser as Model, decode_batch
+
+        self.normalize, self.collate, self.decode_batch = normalize, collate, decode_batch
+        config = json.loads((self.artifact / "config.json").read_text())
+        model = Model(config["representation"], config["head"],
+                      char_classes=config.get("char_classes", False)).eval()
+        model = torch.ao.quantization.quantize_dynamic(model, {nn.Linear}, dtype=torch.qint8)
+        state = torch.load(self.artifact / "model.int8.pt", map_location="cpu",
+                           weights_only=False)
+        model.load_state_dict(state)
+        model.eval()
+        self.model, self.config = model, config
+        self.decoder = "viterbi" if config["head"] == "bio" else "span"
+        self.limit = model.encoder.config.max_position_embeddings - 2
+        self.lock = threading.Lock()
+        self.warmup()
+
+    def digest(self):
+        return hashlib.sha256((self.artifact / "model.int8.pt").read_bytes()).hexdigest()
+
+    @torch.inference_mode()
+    def warmup(self):
+        self.parse("Example.Show.S01E01.1080p-GROUP")
+
+    @torch.inference_mode()
+    def parse(self, query):
+        """Return (record, spans, issues); record is None on abstention."""
+        if len(query) > self.limit:
+            return None, [], [], "over-encoder-limit"
+        row = {"query": query, "record": None, "spans": [], "language": "real",
+               "layout": "real", "view": "real", "family": "real"}
+        with self.lock:
+            item = self.model.encode(row)
+            batch, _ = self.collate([item], self.model.encoder.config.pad_token_id)
+            logits, chars = self.model(batch)
+            spans = self.decode_batch(self.model, logits, chars, [item], self.decoder)[0]
+        record, issues = self.normalize.gated_record(query, spans)
+        return record, spans, issues, None
+
+
+class Handler(BaseHTTPRequestHandler):
+    server_version = "lain-parser/1.0"
+
+    def log_message(self, fmt, *args):
+        log.debug("%s %s", self.address_string(), fmt % args)
+
+    def _send(self, code, payload):
+        body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
+        self.send_response(code)
+        self.send_header("Content-Type", "application/json; charset=utf-8")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def do_GET(self):
+        if self.path != "/health":
+            self._send(404, {"error": "not-found"})
+            return
+        parser = self.server.parser
+        self._send(200, {"status": "ok", "artifact": parser.artifact.name,
+                         "head": parser.config["head"],
+                         "max_chars": parser.limit,
+                         "uptime_s": round(time.time() - self.server.started, 1)})
+
+    def do_POST(self):
+        if self.path != "/parse":
+            self._send(404, {"error": "not-found"})
+            return
+        try:
+            length = int(self.headers.get("Content-Length") or 0)
+        except ValueError:
+            length = 0
+        if length <= 0 or length > MAX_BODY:
+            self._send(400, {"error": "empty-body"})
+            return
+        try:
+            payload = json.loads(self.rfile.read(length))
+        except (ValueError, UnicodeDecodeError):
+            self._send(400, {"error": "invalid-json"})
+            return
+        filename = payload.get("filename") if isinstance(payload, dict) else None
+        if not isinstance(filename, str) or not filename.strip():
+            self._send(400, {"error": "invalid-filename"})
+            return
+        began = time.perf_counter()
+        try:
+            record, spans, issues, limited = self.server.parser.parse(filename)
+        except ValueError as exc:
+            self._send(400, {"error": "invalid-filename", "detail": str(exc)})
+            return
+        except Exception:  # never take the service down over one bad request
+            log.exception("parse failed for %r", filename[:120])
+            self._send(500, {"error": "internal"})
+            return
+        elapsed = round(1000 * (time.perf_counter() - began), 2)
+        log.info("parse %.2fms abstain=%s chars=%d", elapsed, record is None, len(filename))
+        self._send(200, {"artifact": self.server.parser.artifact.name,
+                         "record": record, "spans": spans, "issues": issues,
+                         "abstain_reason": limited, "latency_ms": elapsed})
+
+
+def build_parsers(artifact, snapshot):
+    parser = Parser(artifact, snapshot)
+    log.info("artifact %s head=%s max_chars=%d sha256=%s", parser.artifact.name,
+             parser.config["head"], parser.limit, parser.digest()[:16])
+    return parser
+
+
+class UnixHTTPServer(ThreadingHTTPServer):
+    """HTTP over a Unix domain socket, for containers that cannot reach the host.
+
+    A container with Lain's data directory bind-mounted sees the socket at
+    /data/parser.sock and needs no firewall rule or published port.
+    """
+
+    address_family = socket.AF_UNIX
+    daemon_threads = True
+
+    def server_bind(self):
+        path = self.server_address
+        if os.path.exists(path):
+            os.unlink(path)
+        self.socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        self.socket.bind(path)
+        os.chmod(path, 0o666)  # the container's user is not our user
+        self.server_name = "lain-parser"
+
+    def server_activate(self):
+        self.socket.listen(self.request_queue_size)
+
+    def get_request(self):
+        request, _ = self.socket.accept()
+        return request, ("unix", 0)
+
+    def server_close(self):
+        super().server_close()
+        try:
+            os.unlink(self.server_address)
+        except OSError:
+            pass
+
+
+def serve(hosts, port, parser, unix_path):
+    started = time.time()
+    servers = []
+    for host in hosts:
+        try:
+            server = ThreadingHTTPServer((host, port), Handler)
+        except OSError as exc:
+            log.warning("cannot bind %s:%d: %s", host, port, exc)
+            continue
+        server.parser = parser
+        server.started = started
+        servers.append(server)
+    if unix_path:
+        try:
+            server = UnixHTTPServer(unix_path, Handler)
+        except OSError as exc:
+            log.warning("cannot bind unix socket %s: %s", unix_path, exc)
+        else:
+            server.parser = parser
+            server.started = started
+            servers.append(server)
+    if not servers:
+        raise SystemExit("no bind address worked; is the port already in use?")
+    for server in servers:
+        threading.Thread(target=server.serve_forever, daemon=True).start()
+        if isinstance(server, UnixHTTPServer):
+            log.info("listening on unix://%s", server.server_address)
+        else:
+            log.info("listening on http://%s:%d", *server.server_address[:2])
+    try:
+        while True:
+            time.sleep(3600)
+    except KeyboardInterrupt:
+        log.info("shutting down")
+
+
+def main(argv=None):
+    args = parse_args(argv)
+    logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO,
+                        format="%(asctime)s %(levelname)s %(message)s")
+    parser = build_parsers(args.artifact, args.snapshot)
+    if args.check is not None:
+        record, spans, issues, limited = parser.parse(args.check)
+        print(json.dumps({"filename": args.check, "record": record, "spans": spans,
+                          "issues": issues, "abstain_reason": limited},
+                         ensure_ascii=False, indent=2))
+        return 0
+    serve([h.strip() for h in args.host.split(",") if h.strip()], args.port, parser,
+          args.unix_socket)
+    return 0
+
+
+def parse_args(argv):
+    default_prefix = pathlib.Path(os.environ.get("LAIN_PARSER_PREFIX",
+                                                 pathlib.Path.home() / ".local/share/lain-parser"))
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("--artifact", default=str(default_prefix / "artifact/v3mix-bio-s1"))
+    ap.add_argument("--snapshot", default=str(default_prefix / "anifilebert"))
+    ap.add_argument("--host", default=os.environ.get("LAIN_PARSER_HOST", "127.0.0.1,172.17.0.1"),
+                    help="comma-separated bind addresses")
+    ap.add_argument("--port", type=int, default=int(os.environ.get("LAIN_PARSER_PORT", "9377")))
+    ap.add_argument("--unix-socket", default=os.environ.get("LAIN_PARSER_SOCKET", ""),
+                    help="also listen on this Unix socket path")
+    ap.add_argument("--check", metavar="FILENAME",
+                    help="parse one filename, print JSON and exit")
+    ap.add_argument("--verbose", action="store_true")
+    return ap.parse_args(argv)
+
+
+if __name__ == "__main__":
+    sys.exit(main())