ZelligeAI/tessera-compressor

1"""2segmenting.py — Passage segmentation, classification, and fidelity gates for the3tessera-compressor harness.4 5Extracted from the harness the compressor was accepted under (same functions the6teacher mint used). Pure text processing: no network, no credentials.7 8Flow: segment -> group_steps -> classify_passage per passage -> model call ->9gate -> rules fallback on failure. A failed passage costs a few dozen tokens of10savings, never content.11"""12import re13 14CJK = re.compile(r'[一-鿿㐀-䶿]')15NUM = re.compile(r'\d+(?:\.\d+)?')16IDENT = re.compile(r'`[^`\n]+`|\b[A-Za-z]+(?:_[A-Za-z0-9]+)+\b|\b[a-z]+[A-Z][A-Za-z0-9]*\b')17FENCE = re.compile(r'```.*?```', re.DOTALL)18SENT_SPLIT = re.compile(r'(?<=[.!?;])\s+')19_LIST_MARKER = re.compile(r'(?:^|[\n\s(])(\d{1,2})[.)]\s')20_OPS = set('+-*/=<>≤≥≠∈∀∃¬→⇒%^{}[]')21 22 23def segment(text):24    """Split a reasoning block into ordered segments; code fences are atomic and marked."""25    segs = []  # (kind, text) kind ∈ {'code','prose'}26    pos = 027    for m in FENCE.finditer(text):28        before = text[pos:m.start()]29        segs.extend(('prose', s) for s in _split_prose(before))30        segs.append(('code', m.group(0)))31        pos = m.end()32    segs.extend(('prose', s) for s in _split_prose(text[pos:]))33    return [(k, s) for k, s in segs if s.strip()]34 35 36def _split_prose(text):37    out = []38    for line in text.split('\n'):39        line = line.strip()40        if not line:41            continue42        out.extend(s.strip() for s in SENT_SPLIT.split(line) if s.strip())43    return out44 45 46def group_steps(segs, max_words=160, max_sents=10):47    """Merge consecutive prose sentences into step-sized passages; code stays atomic."""48    out, buf, words = [], [], 049 50    def flush():51        nonlocal buf, words52        if buf:53            out.append(('prose', ' '.join(buf)))54            buf, words = [], 055 56    for kind, s in segs:57        if kind == 'code':58            flush()59            out.append((kind, s))60            continue61        buf.append(s)62        words += len(s.split())63        if words >= max_words or len(buf) >= max_sents:64            flush()65    flush()66    return out67 68 69def facts(s):70    """Numbers + identifiers that must survive compression.71    List-enumeration markers ("1. Load...") are structure, not facts."""72    nums = set(NUM.findall(s)) - set(_LIST_MARKER.findall(s))73    idents = set(i.strip('`') for i in IDENT.findall(s))74    return nums | idents75 76 77def facts_preserved(src, out):78    """Substring presence — regex \\b breaks against adjacent CJK chars.79    Returns the list of MISSING facts (empty list = all preserved)."""80    out_n = out.replace(',', '')81    return [f for f in facts(src) if f.replace(',', '') not in out_n]82 83 84def classify_passage(seg, seen_facts, ntok):85    """'load' = fact-dense or novel-fact-bearing (step-faithful treatment);86       'narr' = search/narrative (stub treatment).87    ntok is a callable: text -> token count under your target tokenizer."""88    f = facts(seg)89    novel = f - seen_facts90    toks = max(ntok(seg), 1)91    dens = (len(NUM.findall(seg)) + len(IDENT.findall(seg))92            + sum(seg.count(o) for o in _OPS)) / toks93    if novel and (dens >= 0.08 or len(novel) >= 3):94        return 'load'95    if dens >= 0.15:96        return 'load'97    return 'narr'98 99 100def gate(src_seg, rules_seg, out, ntok, novel=None):101    """Deterministic per-passage fidelity gate.102    Returns None if the model output is admissible, else a short fail-reason103    string; on failure the caller uses rules_seg instead.104 105    novel: the passage's facts that are NOT already in the accumulated chain.106    The prompt tells the model never to restate chain content, so only novel107    facts are required to survive (matching the acceptance harness). Pass None108    to require every fact of the passage (stricter, for chainless use)."""109    if not out or not out.strip():110        return "empty"111    if '```' in out:112        return "fence"113    if len(out) > 2 * len(src_seg) + 40:           # explanation/blow-up guard114        return "blowup"115    required = facts(src_seg) if novel is None else novel116    out_n = out.replace(',', '')117    if any(f.replace(',', '') not in out_n for f in required):118        return "facts"119    if ntok(out) > ntok(rules_seg):                # must not exceed the rules-only version120        return "tokens"121    return None122