#!/usr/bin/env python3
"""
Mining "LLM-isms" (2/3/4-grams) from Coding-Harness Transcripts (Claude Code + Albedo SQLite)
=============================================================================================
Implements Monroe et al. (2008, "Fightin' Words") z-scored log-odds ratio with an informative
Dirichlet prior (alpha_w = alpha_0 * p_background(w)) comparing multi-provider assistant prose
against Dawn's human user turns.
Fixes all 8 items in PLAN.md §9 plus all 6 audit findings from `audits/opus_audit_slop_miner_and_1c_v3.md`:
1. Parses Claude Code `.jsonl` rows (`{"type": "assistant"|"user", "message": ...}`), skipping `isSidechain`
and `isMeta` rows, and stripping `>` quote lines from human turns.
2. Reads Albedo's Erlang External Term Format (`\x83`, including zlib-compressed tag 80) directly from
`~/.albedo/albedo.sqlite` (`transcript` for provider-tagged assistant replies excluding `reasoning_text`;
`submission_events` + root-session `transcript` `user` rows for human turns), counting any decode errors.
3. Tags assistant prose by provider AND canonical model family (`claude`, `qwen`, `gemini`, `glm`, `openai`,
`xiaomi`, `deepseek`), requiring `min_families >= 2` so single-family variants cannot pass alone.
4. Requires a non-empty human reference baseline (fails closed if `N_human == 0`).
5. Uses Monroe's exact informative Dirichlet prior `alpha_w = alpha_0 * p_bg(w)` with an 85/15
mixture of per-order normalized independent unigram-product background prior and per-order
uniform prior over the candidate vocabulary (`0.85 * p_indep + 0.15 * p_unif`, zero self-pooled
n-gram term in `alpha_w`, so `c_llm` never cancels at `c_hum == 0`) + 4-term asymptotic variance.
6. Normalizes curly apostrophes (`\u2018/\u2019 -> "'"`) and replaces EVERY stripped span (inline code, URLs,
paths, XML tags, fenced code, hyphenated identifiers, numbers, non-Latin runs) with a hard clause boundary
(`"\n"`) BEFORE n-gramming so phantom n-grams never bridge across removed spans.
7. Maps phrases to Qwen3.8-Flash-Next token tuples across bare/space/newline/quote/paren/bracket, first-word-cap,
and mid-phrase pronoun-`I` variants (`2 <= len(ids) <= 8`), failing closed if the tokenizer cannot load.
8. Drops the bogus 1.5% / 3.6% letter-lipogram cap, applies `PROTECTED_STOP_NGRAMS` to all sub-tuples, and
restricts candidates to conversational/rhetorical signposting rather than build/test status reports.
"""
import argparse
import glob
import json
import math
import os
import re
import sqlite3
import struct
import zlib
from collections import Counter, defaultdict
from pathlib import Path
from typing import Dict, Iterable, List, Optional, Set, Tuple
# Hard-boundary regexes: every match is replaced with "\n" (never " ") so adjacent words never bridge
FENCED_CODE_RE = re.compile(r"```[\s\S]*?```", re.MULTILINE)
INLINE_CODE_RE = re.compile(r"`[^`\n]+`")
THINK_BLOCK_RE = re.compile(r"<think>[\s\S]*?</think>", re.IGNORECASE)
TOOL_BLOCK_RE = re.compile(
r"<(tool_call|tool_result|function_calls|invoke|antml:function_calls|system-note|mail)[\s\S]*?</\1>",
re.IGNORECASE,
)
XML_TAG_RE = re.compile(r"<[^>]+>")
URL_PATH_RE = re.compile(r"https?://\S+|(?:/[\w.\-]+){2,}")
MARKDOWN_TABLE_LINE_RE = re.compile(r"^\s*\|.*\|\s*$", re.MULTILINE)
MD_HEADER_LINE_RE = re.compile(r"^\s*#{1,6}\s+.*$", re.MULTILINE)
QUOTE_LINE_RE = re.compile(r"^\s*>.*$", re.MULTILINE)
BOX_LINE_RE = re.compile(r"^.*[│┌┐└┘├┤┬┴┼─═║].*$", re.MULTILINE)
HYPHEN_ID_RE = re.compile(r"\b[a-z0-9_]+(?:-[a-z0-9_]+)+\b", re.IGNORECASE)
CLAUSE_SPLIT_RE = re.compile(r"[\n\r.!?;:,—–()\[\]{}\"“”/\\|<>+*=#~^`]+")
NON_WORD_SPAN_RE = re.compile(r"[^a-z'\s]+")
WORD_EXACT_RE = re.compile(r"^[a-z]+(?:'[a-z]+)?$")
PRONOUN_I_RE = re.compile(r"\bi(?='|\b)")
FUNCTION_WORDS: Set[str] = {
"the", "a", "an", "of", "in", "to", "for", "with", "on", "at", "from", "by", "as",
"into", "over", "after", "before", "under", "between", "through", "during", "without",
"within", "against", "among", "around", "across", "behind", "beyond", "upon", "about",
"and", "or", "but", "so", "if", "then", "than", "when", "while", "where", "because",
"since", "unless", "until", "though", "although", "whether", "both", "either", "neither",
"is", "are", "was", "were", "be", "been", "being", "am",
"has", "have", "had", "having", "do", "does", "did", "doing", "done",
"will", "would", "shall", "should", "can", "could", "may", "might", "must",
"not", "no", "nor", "only", "also", "very", "too", "just", "still", "even", "now", "already", "yet",
"once", "whenever", "wherever", "however", "therefore", "meanwhile", "far", "away", "right",
"it", "its", "it's", "it'll", "this", "that", "these", "those", "that's", "that'll",
"there", "there's", "there'll", "here", "here's",
"he", "him", "his", "he's", "she", "her", "hers", "she's",
"they", "them", "their", "theirs", "they're", "they'll", "they've", "they'd",
"we", "us", "our", "ours", "we're", "we'll", "we've", "we'd",
"you", "your", "yours", "you're", "you'll", "you've", "you'd",
"i", "me", "my", "mine", "i'm", "i'll", "i've", "i'd",
"who", "whom", "whose", "who's", "which", "what", "what's", "whatever", "whichever",
"how", "how's", "why", "why's", "where's", "when's",
"all", "any", "some", "many", "much", "more", "most", "other", "another", "such", "own",
"same", "new", "old", "first", "second", "third", "last", "next", "one", "two", "three", "four", "five",
"each", "every", "few", "several", "enough", "less", "least", "up", "down", "out", "off",
"don't", "doesn't", "didn't", "won't", "wouldn't", "can't", "couldn't", "shouldn't",
"isn't", "aren't", "wasn't", "weren't", "haven't", "hasn't", "hadn't",
}
RHETORICAL_ALLOWLIST: Set[Tuple[str, ...]] = {
("let", "me"), ("i", "will"), ("i'll", "check"), ("i'll", "inspect"), ("i'll", "look"),
("here", "is"), ("here", "are"), ("here's", "what"), ("here's", "the"), ("rather", "than"),
("in", "order", "to"), ("as", "well", "as"), ("if", "you", "want"), ("if", "you'd", "like"),
("if", "you'd", "rather"), ("would", "you", "like"), ("do", "you", "want"), ("want", "me", "to"),
("let", "me", "know"), ("feel", "free", "to"), ("make", "sure"), ("keep", "in", "mind"),
("worth", "noting"), ("at", "this", "point"), ("by", "default"), ("under", "the", "hood"),
("out", "of", "the", "box"), ("end", "to", "end"), ("right", "away"),
("so", "i'll"), ("and", "i'll"), ("so", "i'm"), ("so", "i'd"),
("two", "things"), ("three", "things"), ("that", "said"), ("with", "that", "said"),
("in", "other", "words"), ("on", "the", "other", "hand"), ("at", "a", "high", "level"),
("for", "what", "it's", "worth"), ("to", "be", "fair"), ("to", "be", "clear"), ("to", "be", "honest"),
("bottom", "line"), ("short", "version"), ("good", "catch"), ("great", "question"),
("makes", "sense"), ("you're", "right"), ("spot", "on"), ("is", "exactly"), ("exactly", "what"),
("let", "me", "check"), ("let", "me", "see"), ("let", "me", "look"),
("let", "me", "read"), ("let", "me", "verify"), ("let", "me", "inspect"), ("now", "let", "me"),
}
CONVERSATIONAL_ANCHORS: Set[str] = {
"i", "i'm", "i'll", "i've", "i'd", "me", "my", "we", "we're", "we'll", "we've", "we'd", "us",
"let", "let's", "you", "you're", "you'll", "you've", "you'd", "your", "yours",
"want", "wants", "wanted", "would", "could", "should", "shall", "can", "may", "might",
"here", "here's", "there", "there's", "that's", "it's", "what", "what's", "which",
"rather", "instead", "however", "therefore", "thus", "hence", "overall", "basically",
"essentially", "actually", "certainly", "absolutely", "definitely", "clearly", "obviously",
"simply", "just", "exact", "exactly", "precisely", "expected", "sense", "catch", "spot",
"happy", "glad", "delighted", "feel", "free", "please", "sorry", "apologies", "apologize",
"worth", "noting", "mind", "keep", "bear", "fair", "honest", "frankly", "short", "bottom",
"hood", "box", "hand", "words", "summary", "meanwhile", "far",
}
REPO_SPECIFIC_WORDS: Set[str] = {
"chernobog", "appview", "appview's", "spindle", "bobbin", "hydrant", "albedo", "tangled",
"knot", "knotserver", "jetstream", "plc", "atproto", "did", "rkey", "nsid", "lexicon",
"gleam", "erlang", "beam", "otp", "lustre", "mist", "wisp", "pog", "squirrel",
"gguf", "qwen", "qwen4exp", "ukisai", "imatrix", "reap", "moce", "whittle", "chitako",
"hotchpotch", "lottolabs", "colab", "strata", "nvfp", "mxfp", "gdn", "qsa", "ple",
"bunker", "caddy", "wireguard", "nftables", "restic", "launchd", "plist", "nix", "nixpkgs",
"hyd", "eu", "auditor", "opus", "haiku", "sonnet",
}
# Build/test/ops status vocabulary excluded unless explicitly in RHETORICAL_ALLOWLIST
TECH_STATUS_WORDS: Set[str] = {
"clean", "dirty", "running", "stopped", "pushed", "committed", "merged", "staged",
"built", "compiled", "passed", "passing", "failed", "failing", "test", "tests", "tested",
"suite", "build", "builds", "job", "jobs", "cell", "cells", "kernel", "daemon",
"server", "client", "file", "files", "line", "lines", "code", "script", "scripts",
"function", "functions", "table", "tables", "column", "columns", "row", "rows",
"token", "tokens", "model", "models", "layer", "layers", "expert", "experts",
"shard", "shards", "tensor", "tensors", "cache", "cached", "memory", "disk",
"gpu", "cpu", "ssd", "ram", "vram", "repo", "repos", "branch", "commit", "commits",
"diff", "patch", "patched", "pr", "issue", "issues", "bug", "bugs", "error", "errors",
"warning", "warnings", "matrix", "feature", "features", "private", "public", "local",
"remote", "session", "sessions", "turn", "turns", "prompt", "prompts", "context",
"output", "outputs", "input", "inputs", "result", "results", "value", "values",
"type", "types", "field", "fields", "key", "keys", "byte", "bytes", "bit", "bits",
"chunk", "chunks", "slice", "sliced", "prune", "pruned", "pruning", "eval", "evals",
"phase", "step", "steps", "task", "tasks", "work", "item", "items", "ledger",
"wakes", "wake", "idle", "background", "thread", "threads", "process", "processes",
"command", "commands", "flag", "flags", "arg", "args", "option", "options",
"config", "configs", "path", "paths", "dir", "directory", "folder", "json", "sqlite",
"database", "schema", "query", "queries", "index", "indices", "array", "list", "dict",
"map", "maps", "set", "sets", "tuple", "tuples", "string", "strings", "int", "float",
"bool", "true", "false", "null", "none", "open", "closed", "going", "changed",
"touch", "touched", "kept", "sent", "asked", "waiting", "waits", "wait", "left",
"ran", "added", "read", "reading", "wrote", "written", "writing", "updated", "updating",
"fixed", "fixing", "removed", "removing", "deleted", "deleting", "created", "creating",
"need", "needed", "found", "tell", "see", "checking",
"mac", "macbook", "checkout", "change", "changes", "mailed", "mail",
"saved", "save", "rerun", "finishes", "finish", "finished", "measured",
}
PROTECTED_STOP_NGRAMS: Set[Tuple[str, ...]] = {
("exit", "code"), ("syntax", "error"), ("stack", "trace"), ("file", "path"),
("line", "number"), ("pull", "request"), ("unit", "test"), ("type", "error"),
("test", "suite"), ("unit", "tests"), ("command", "line"), ("source", "code"),
("kv", "cache"), ("shared", "expert"), ("routed", "experts"), ("linear", "attention"),
("read", "only"), ("full", "suite"), ("tests", "pass"), ("working", "copy"),
("feature", "matrix"), ("private", "repos"), ("break", "check"), ("hyd", "eu"),
# Protect Phase 3 abstention & epistemic-honesty disclosures so Phase 4 anti-slop never penalizes them
("don't", "know"), ("i", "don't", "know"), ("not", "sure"), ("i'm", "not", "sure"),
("can't", "confirm"), ("cannot", "confirm"), ("can't", "verify"), ("cannot", "verify"),
("haven't", "checked"), ("i", "haven't", "checked"), ("didn't", "run"), ("i", "didn't", "run"),
("haven't", "run"), ("i", "haven't", "run"), ("don't", "think"), ("i", "don't", "think"),
}
# Explicit conversational/rhetorical LLM-ism patterns (excludes 1st-person past-tense work logs and generic modal pairs)
SIGNPOST_PREFIXES: Tuple[Tuple[str, ...], ...] = (
("let", "me"), ("now", "let", "me"), ("so", "let", "me"), ("lem", "me"),
("so", "i'll"), ("and", "i'll"), ("so", "i'm"), ("so", "i'd"), ("i'd", "rather"),
("i'll", "check"), ("i'll", "look"), ("i'll", "see"), ("i'll", "verify"), ("i'll", "inspect"),
("here's", "what"), ("here's", "the"), ("here's", "how"), ("here", "is"), ("here", "are"),
("want", "me"), ("do", "you", "want"), ("or", "do", "you"), ("if", "you", "want"),
("if", "you'd"), ("would", "you", "like"), ("you", "want", "me"), ("you", "want", "to"),
("you", "want", "it"), ("say", "if", "you"), ("whenever", "you", "want"),
("feel", "free"), ("let", "me", "know"), ("happy", "to"), ("glad", "to"),
("is", "exactly"), ("which", "is", "exactly"), ("that's", "exactly"), ("exactly", "what"),
("what's", "actually"), ("instead", "of", "guessing"), ("there's", "nothing", "to"),
("it's", "worth"), ("worth", "noting"), ("keep", "in", "mind"), ("bear", "in", "mind"),
("one", "catch"), ("the", "catch", "is"), ("catch", "is", "that"),
)
def classify_remediation_mechanism(ng: Tuple[str, ...]) -> Tuple[str, str]:
"""
Classifies a mined LLM-ism into:
- ("phase3_sentence_surgery", "pre_tool_narration" | "closing_solicitation" | "sycophantic_opener")
for turn-level speech acts where single-token FTPO would just cause synonym whack-a-mole, vs
- ("phase4_lexical_ftpo", "lexical_rhetorical_cliche")
for phrase-level rhetorical filler safe for single-token FTPO in Phase 4.
"""
phrase = " ".join(ng)
# 1. Sycophantic openers
if phrase in ("you're right", "good catch", "great question", "spot on", "makes sense"):
return "phase3_sentence_surgery", "sycophantic_opener"
# 2. Closing permission-solicitation / menu-offering speech acts
if phrase in ("your call", "don't want", "i'd want", "or do you", "or do you want") or (
any(w in ng for w in ("want", "wanted", "rather", "like"))
and any(w in ng for w in ("you", "you'd", "me", "to"))
):
return "phase3_sentence_surgery", "closing_solicitation"
if ng[:2] in (("feel", "free"), ("let", "me")) and "know" in ng:
return "phase3_sentence_surgery", "closing_solicitation"
# 3. Pre-tool narration / action-announcement speech acts
if (
ng[:2] in (
("let", "me"), ("lem", "me"), ("so", "i'll"), ("and", "i'll"),
("so", "i'm"), ("so", "i'd"), ("i'll", "check"), ("i'll", "look"),
("i'll", "see"), ("i'll", "verify"), ("i'll", "inspect"),
)
or ng[:3] in (("now", "let", "me"), ("so", "let", "me"))
or "check" in ng
):
return "phase3_sentence_surgery", "pre_tool_narration"
# 4. Surface lexical / rhetorical filler cliches (safe for Phase 4 single-token FTPO)
return "phase4_lexical_ftpo", "lexical_rhetorical_cliche"
def provider_to_family(prov: str) -> str:
"""Maps a harness:provider tag (`albedo:tangled-clod`, `cc:claude-opus-5`, `albedo:alibaba`) to its model family."""
p = prov.lower()
if any(k in p for k in ("clod", "claude", "anthropic")):
return "claude"
if any(k in p for k in ("alibaba", "qwen")):
return "qwen"
if any(k in p for k in ("antigravity", "gemini", "google")):
return "gemini"
if any(k in p for k in ("charm", "glm", "zhipu")):
return "glm"
if any(k in p for k in ("codex", "openai", "gpt")):
return "openai"
if "xiaomi" in p or "mimo" in p:
return "xiaomi"
if "deepseek" in p:
return "deepseek"
return p
# First-person work-report verbs ("i made", "i moved", "i tried", "i'll write", "i'll run")
# that describe agent task execution rather than rhetorical/conversational signposting
EXCLUDED_WORK_AND_GENERIC_WORDS: Set[str] = {
"think", "far", "expected",
"checked", "made", "put", "used", "moved", "filed", "tried", "took", "went", "gave",
"broke", "looked", "dropped", "started", "told", "said", "write", "read", "run",
"add", "make", "start", "pick", "report", "leaving", "earlier", "cheap", "reach",
"happens", "matches", "needs", "makes", "means", "never", "gives", "meant", "wanted",
"say", "ever", "probably", "fine", "real", "one", "same", "summary",
"working", "terminal", "does", "leave", "keep", "use", "things", "this",
}
def is_stylistic_candidate(ng: Tuple[str, ...]) -> bool:
if any(w in REPO_SPECIFIC_WORDS for w in ng):
return False
# Reject if any contiguous sub-tuple is in PROTECTED_STOP_NGRAMS (prevents super-gram leaks like "the full suite")
for k in (2, 3, 4):
if k <= len(ng):
for i in range(len(ng) - k + 1):
if ng[i : i + k] in PROTECTED_STOP_NGRAMS:
return False
if ng in RHETORICAL_ALLOWLIST:
return True
for pref in SIGNPOST_PREFIXES:
if len(ng) >= len(pref) and ng[: len(pref)] == pref:
if not any(w in TECH_STATUS_WORDS for w in ng):
return True
if all(w in FUNCTION_WORDS for w in ng):
return False
if not any(w in CONVERSATIONAL_ANCHORS for w in ng):
return False
if any(w in TECH_STATUS_WORDS or w in EXCLUDED_WORK_AND_GENERIC_WORDS for w in ng):
return False
# Exclude bare "i <verb>" / "i'm <verb>" / "can <verb>" / "would <verb>" / "which <verb>" / "what <verb>" outside SIGNPOST_PREFIXES
if ng[0] in {"i", "i'm", "me", "my", "can", "would", "could", "should", "which", "what", "that", "that's", "there's", "it's", "to", "keep", "and", "or", "check"}:
return False
bare_edge = {
"the", "a", "an", "in", "on", "of", "for", "with", "from", "by", "at",
"into", "over", "after", "before", "against", "between", "through", "under", "without", "within",
}
if ng[0] in bare_edge or (ng[-1] in bare_edge and ng[:2] != ("let", "me") and ng[:2] != ("i'll", "look")):
return False
if len(ng) == 2 and ng[1] in {
"to", "and", "or", "but", "so", "if", "as", "that", "which", "who", "when", "where",
"while", "because", "since", "not", "no", "also", "still", "already", "just", "only",
"even", "now", "yet", "it", "this", "that",
}:
return False
if len(ng) == 2 and ng[0] in {"and", "or", "but", "is", "are", "was", "were", "be", "been", "has", "have", "had"}:
return False
if not any(w in FUNCTION_WORDS for w in ng):
return False
return True
def clean_prose(raw_text: str, is_human: bool = False) -> str:
"""
Normalizes curly apostrophes/quotes and replaces every non-prose span (code blocks, inline code,
<think>, tool/system XML, URLs/paths, markdown headers/tables, hyphenated identifiers, and
human blockquotes/TUI boxes) with a hard newline boundary ('\\n') so words never bridge across removed spans.
"""
if not raw_text:
return ""
text = (
raw_text.replace("\u2019", "'")
.replace("\u2018", "'")
.replace("\u201c", '"')
.replace("\u201d", '"')
)
if is_human:
text = QUOTE_LINE_RE.sub("\n", text)
text = BOX_LINE_RE.sub("\n", text)
text = THINK_BLOCK_RE.sub("\n", text)
text = TOOL_BLOCK_RE.sub("\n", text)
text = FENCED_CODE_RE.sub("\n", text)
text = MD_HEADER_LINE_RE.sub("\n", text)
text = MARKDOWN_TABLE_LINE_RE.sub("\n", text)
text = INLINE_CODE_RE.sub("\n", text)
text = XML_TAG_RE.sub("\n", text)
text = URL_PATH_RE.sub("\n", text)
text = HYPHEN_ID_RE.sub("\n", text)
return text.strip()
def extract_clauses(text: str, is_human: bool = False) -> List[List[str]]:
"""
Cleans text and splits on both clause punctuation AND any non-[a-z'\\s] span (digits, CJK, symbols)
so only genuinely adjacent English words in the original prose form contiguous word runs.
"""
cleaned = clean_prose(text, is_human=is_human)
if not cleaned:
return []
clauses: List[List[str]] = []
for raw_clause in CLAUSE_SPLIT_RE.split(cleaned.lower()):
for sub_clause in NON_WORD_SPAN_RE.split(raw_clause):
toks = sub_clause.split()
valid_run: List[str] = []
for tok in toks:
tok_clean = tok.strip("'")
if (len(tok_clean) > 1 or tok_clean in ("a", "i")) and WORD_EXACT_RE.match(tok_clean):
valid_run.append(tok_clean)
else:
if len(valid_run) >= 2:
clauses.append(valid_run)
valid_run = []
if len(valid_run) >= 2:
clauses.append(valid_run)
return clauses
def extract_ngrams(
text: str,
orders: Tuple[int, ...] = (2, 3, 4),
is_human: bool = False,
) -> List[Tuple[str, ...]]:
"""Extracts stylistic candidate 2/3/4-grams strictly within hard-bounded clauses."""
out: List[Tuple[str, ...]] = []
for words in extract_clauses(text, is_human=is_human):
for n in orders:
if len(words) < n:
continue
for i in range(len(words) - n + 1):
ng = tuple(words[i : i + n])
if is_stylistic_candidate(ng):
out.append(ng)
return out
def decode_etf_term(buf: bytes, pos: int = 0):
"""Erlang External Term Format (ETF, tag 131) decoder for Albedo SQLite payloads."""
tag = buf[pos]
pos += 1
if tag == 80: # COMPRESSED
uncompressed_size = struct.unpack(">I", buf[pos : pos + 4])[0]
decompressed = zlib.decompress(buf[pos + 4 :])
if len(decompressed) != uncompressed_size:
raise ValueError(f"ETF compressed size mismatch: {len(decompressed)} != {uncompressed_size}")
val, _ = decode_etf_term(decompressed, 0)
return val, len(buf)
if tag == 97:
return buf[pos], pos + 1
if tag == 98:
return struct.unpack(">i", buf[pos : pos + 4])[0], pos + 4
if tag == 99:
return float(buf[pos : pos + 31].rstrip(b"\x00")), pos + 31
if tag == 70:
return struct.unpack(">d", buf[pos : pos + 8])[0], pos + 8
if tag in (100, 118):
n = struct.unpack(">H", buf[pos : pos + 2])[0]
pos += 2
return ("atom", buf[pos : pos + n].decode("utf-8", errors="replace")), pos + n
if tag in (115, 119):
n = buf[pos]
pos += 1
return ("atom", buf[pos : pos + n].decode("utf-8", errors="replace")), pos + n
if tag == 104:
arity = buf[pos]
pos += 1
items = []
for _ in range(arity):
val, pos = decode_etf_term(buf, pos)
items.append(val)
return tuple(items), pos
if tag == 105:
arity = struct.unpack(">I", buf[pos : pos + 4])[0]
pos += 4
items = []
for _ in range(arity):
val, pos = decode_etf_term(buf, pos)
items.append(val)
return tuple(items), pos
if tag == 106:
return [], pos
if tag == 107:
n = struct.unpack(">H", buf[pos : pos + 2])[0]
pos += 2
return buf[pos : pos + n].decode("latin1"), pos + n
if tag == 108:
n = struct.unpack(">I", buf[pos : pos + 4])[0]
pos += 4
items = []
for _ in range(n):
val, pos = decode_etf_term(buf, pos)
items.append(val)
_tail, pos = decode_etf_term(buf, pos)
return items, pos
if tag == 109:
n = struct.unpack(">I", buf[pos : pos + 4])[0]
pos += 4
raw = buf[pos : pos + n]
try:
return raw.decode("utf-8"), pos + n
except UnicodeDecodeError:
return ("bytes", len(raw)), pos + n
if tag == 77: # BIT_BINARY_EXT
n = struct.unpack(">I", buf[pos : pos + 4])[0]
_bits = buf[pos + 4]
pos += 5
return ("bit_binary", n), pos + n
if tag == 110:
n = buf[pos]
sign = buf[pos + 1]
pos += 2
val = int.from_bytes(buf[pos : pos + n], "little")
return (-val if sign else val), pos + n
if tag == 111:
n = struct.unpack(">I", buf[pos : pos + 4])[0]
sign = buf[pos + 4]
pos += 5
val = int.from_bytes(buf[pos : pos + n], "little")
return (-val if sign else val), pos + n
if tag == 116:
arity = struct.unpack(">I", buf[pos : pos + 4])[0]
pos += 4
d = {}
for _ in range(arity):
k, pos = decode_etf_term(buf, pos)
v, pos = decode_etf_term(buf, pos)
d[k] = v
return d, pos
raise ValueError(f"Unhandled ETF tag {tag} at pos {pos - 1}")
def iter_claude_code_turns(paths: Iterable[str]) -> Tuple[List[Tuple[str, str]], List[str]]:
"""
Parses Claude Code `.jsonl` and OpenAI/Anthropic `.json`/`.jsonl` conversation files.
Skips `isMeta` and `isSidechain` rows so subagent prompts are never counted as human turns.
Returns (assistant_turns [(provider_tag, raw_text)], human_turns [raw_text]).
"""
asst_turns: List[Tuple[str, str]] = []
human_turns: List[str] = []
for path in paths:
if path.endswith(".jsonl"):
with open(path, "r", encoding="utf-8", errors="ignore") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
obj = json.loads(line)
except json.JSONDecodeError:
continue
if not isinstance(obj, dict) or obj.get("isMeta") or obj.get("isSidechain"):
continue
msg = obj.get("message")
if isinstance(msg, dict):
role = msg.get("role") or obj.get("type")
model = str(msg.get("model") or obj.get("model") or "claude-code")
if model == "<synthetic>":
continue
content = msg.get("content", "")
if isinstance(content, list):
if role == "user" and any(
isinstance(b, dict) and b.get("type") == "tool_result" for b in content
):
continue
content = "\n".join(
b.get("text", "")
for b in content
if isinstance(b, dict) and b.get("type") == "text"
)
if isinstance(content, str) and content.strip():
raw = content.strip()
if role == "assistant":
asst_turns.append((f"cc:{model}", raw))
elif role == "user":
if any(
t in raw
for t in (
"<command-message>",
"Base directory for this skill:",
"<system-reminder>",
"<local-command-",
)
):
continue
if raw != "assis:":
human_turns.append(raw)
continue
model = str(obj.get("model", obj.get("provider", "jsonl_model")))
messages = obj.get("messages", obj.get("conversation", []))
if isinstance(messages, list):
for m in messages:
if not isinstance(m, dict):
continue
r = m.get("role")
c = m.get("content", "")
if isinstance(c, list):
c = "\n".join(
b.get("text", "")
for b in c
if isinstance(b, dict) and b.get("type") == "text"
)
if isinstance(c, str) and c.strip():
if r == "assistant":
asst_turns.append((model, c.strip()))
elif r == "user":
human_turns.append(c.strip())
elif path.endswith(".json"):
try:
with open(path, "r", encoding="utf-8", errors="ignore") as f:
data = json.load(f)
items = data if isinstance(data, list) else [data]
for obj in items:
if isinstance(obj, dict):
model = str(obj.get("model", "json_model"))
for m in obj.get("messages", []):
if isinstance(m, dict) and isinstance(m.get("content"), str) and m["content"].strip():
if m.get("role") == "assistant":
asst_turns.append((model, m["content"].strip()))
elif m.get("role") == "user":
human_turns.append(m["content"].strip())
except Exception:
continue
return asst_turns, human_turns
def iter_albedo_sqlite_turns(sqlite_path: str) -> Tuple[List[Tuple[str, str]], List[str], int]:
"""
Reads assistant turns (tagged by `albedo:<provider>`, excluding `reasoning_text` items) from `transcript`
and Dawn's human turns from both `submission_events` and root-session `transcript` `user` rows
in `~/.albedo/albedo.sqlite`. Returns (asst_turns, human_turns, etf_decode_errors).
"""
asst_turns: List[Tuple[str, str]] = []
human_turns: List[str] = []
human_seen: Set[str] = set()
etf_decode_errors = 0
con = sqlite3.connect(f"file:{sqlite_path}?mode=ro", uri=True)
try:
child_sessions: Set[str] = set()
try:
for (s_id,) in con.execute(
"SELECT session FROM session_family WHERE depth > 0 OR parent IS NOT NULL"
):
if isinstance(s_id, str):
child_sessions.add(s_id)
except sqlite3.OperationalError:
pass
for (pl,) in con.execute("SELECT payload FROM submission_events"):
if not pl or pl[0] != 131:
continue
try:
term, _ = decode_etf_term(pl, 1)
except Exception:
etf_decode_errors += 1
continue
if isinstance(term, tuple) and len(term) >= 4 and term[0] == ("atom", "display"):
txt, mode, src = term[1], term[2], str(term[3])
if isinstance(txt, str) and mode == "chat" and src.startswith("cli-"):
raw = txt.strip()
if raw and not raw.startswith(("/", "[scheduled", "<mail", "<system")):
if raw not in human_seen:
human_seen.add(raw)
human_turns.append(raw)
for sess_id, pl in con.execute("SELECT session, payload FROM transcript WHERE row_class = 'user'"):
if sess_id in child_sessions or not pl or pl[0] != 131:
continue
try:
term, _ = decode_etf_term(pl, 1)
except Exception:
etf_decode_errors += 1
continue
if (
isinstance(term, tuple)
and len(term) == 2
and isinstance(term[1], tuple)
and len(term[1]) == 2
and term[1][0] == ("atom", "user")
and isinstance(term[1][1], str)
):
raw = term[1][1].strip()
if raw and not raw.startswith(
("/", "[scheduled", "<mail", "<system", "The images below archive")
):
if raw not in human_seen:
human_seen.add(raw)
human_turns.append(raw)
for prov, pl in con.execute("SELECT provider, payload FROM transcript WHERE row_class = 'other'"):
if not pl or pl[0] != 131:
continue
try:
term, _ = decode_etf_term(pl, 1)
except Exception:
etf_decode_errors += 1
continue
if (
isinstance(term, tuple)
and len(term) == 2
and isinstance(term[1], tuple)
and len(term[1]) >= 2
and term[1][0] == ("atom", "replay")
):
rep = term[1][1]
if isinstance(rep, tuple) and len(rep) >= 3 and isinstance(rep[2], dict):
d = rep[2]
if d.get("type") in (
"reasoning",
"reasoning_text",
"function_call",
"function_call_output",
"tool_use",
"tool_result",
):
continue
if d.get("role") not in (None, "assistant"):
continue
c = d.get("content")
txt = ""
if isinstance(c, str):
txt = c.strip()
elif isinstance(c, list):
parts = [
b["text"]
for b in c
if isinstance(b, dict)
and b.get("type")
not in ("reasoning", "reasoning_text", "thinking", "tool_use", "tool_result")
and isinstance(b.get("text"), str)
]
txt = "\n".join(parts).strip()
if txt:
asst_turns.append((f"albedo:{prov or 'unknown'}", txt))
finally:
con.close()
return asst_turns, human_turns, etf_decode_errors
def compute_fightin_words_z_scores(
llm_counts: Counter,
human_counts: Counter,
provider_counts_per_ng: Dict[Tuple[str, ...], Counter],
unigram_counts: Optional[Counter] = None,
min_count: int = 12,
min_providers: int = 2,
min_families: int = 2,
alpha_0: float = 5000.0,
z_threshold: float = 1.65,
min_rate_ratio: float = 3.0,
) -> List[Dict]:
"""
Monroe et al. (2008) z-scored log-odds ratio with an Informative Dirichlet Prior:
p_pooled(w) = (c_llm(w) + c_hum(w)) / (N_llm + N_hum)
p_indep(w) = order_share[k] * prod_{u in w} P_unigram(u) / Z_k (normalized per order k)
p_unif(w) = order_share[k] / |V_k|
p_bg(w) = 0.85 * p_indep(w) + 0.15 * p_unif(w) (sum_{w in V} alpha_w == alpha_0, zero self-pooled term,
and c_llm does NOT cancel out of delta_w when c_hum == 0)
alpha_w = alpha_0 * p_bg(w)
delta_w = log((c_llm + alpha_w) / (N_llm + alpha_0 - c_llm - alpha_w))
- log((c_hum + alpha_w) / (N_hum + alpha_0 - c_hum - alpha_w))
var_w = 1/(c_llm + alpha_w) + 1/(N_llm + alpha_0 - c_llm - alpha_w)
+ 1/(c_hum + alpha_w) + 1/(N_hum + alpha_0 - c_hum - alpha_w)
z_w = delta_w / sqrt(var_w)
"""
n_llm = sum(llm_counts.values())
n_hum = sum(human_counts.values())
if n_llm <= 0:
raise ValueError("LLM assistant corpus has 0 n-grams.")
if n_hum <= 0:
raise ValueError("Human reference corpus has 0 n-grams; a human baseline is required.")
vocab = sorted(set(llm_counts.keys()) | set(human_counts.keys()))
total_bg = float(n_llm + n_hum)
# Build unigram frequency table for the independent background component
if unigram_counts is None:
unigram_counts = Counter()
for ng in vocab:
c_tot = llm_counts.get(ng, 0) + human_counts.get(ng, 0)
for w in ng:
unigram_counts[w] += c_tot
total_uni = float(sum(unigram_counts.values())) + float(len(unigram_counts))
cand_vocab = [
ng for ng in vocab if llm_counts.get(ng, 0) >= min_count and is_stylistic_candidate(ng)
]
raw_indep: Dict[Tuple[str, ...], float] = {}
sum_indep_by_order: Counter = Counter()
order_bg_mass: Counter = Counter()
order_vocab_cnt: Counter = Counter()
for ng in cand_vocab:
k = len(ng)
c_tot = llm_counts.get(ng, 0) + human_counts.get(ng, 0)
order_bg_mass[k] += c_tot
order_vocab_cnt[k] += 1
p = 1.0
for w in ng:
p *= (unigram_counts.get(w, 0) + 1.0) / total_uni
raw_indep[ng] = p
sum_indep_by_order[k] += p
total_cand_bg = float(sum(order_bg_mass.values())) or 1.0
raw_results = []
for ng in cand_vocab:
c_llm = llm_counts.get(ng, 0)
p_dist = provider_counts_per_ng.get(ng, Counter())
num_providers = len(p_dist)
if num_providers < min_providers:
continue
fam_dist: Counter = Counter()
for prov, cnt in p_dist.items():
fam_dist[provider_to_family(prov)] += cnt
num_families = len(fam_dist)
if num_families < min_families:
continue
# Require strong cross-family support (v4 §3.4): >= 2 families each with >= max(5, 5% of c_llm) hits
if min_families >= 2:
min_fam_Floor = max(5, int(0.05 * c_llm))
fams_meeting_floor = sum(1 for cnt in fam_dist.values() if cnt >= min_fam_Floor)
if fams_meeting_floor < 2:
continue
c_hum = human_counts.get(ng, 0)
llm_rate = 10000.0 * c_llm / n_llm
hum_rate = 10000.0 * c_hum / n_hum
if llm_rate < min_rate_ratio * max(hum_rate, 0.05):
continue
k = len(ng)
order_share = (order_bg_mass[k] / total_cand_bg) if total_cand_bg > 0 else (1.0 / 3.0)
p_pooled = (c_llm + c_hum) / total_bg
p_indep = order_share * (raw_indep[ng] / (sum_indep_by_order[k] or 1.0))
p_unif = order_share / max(1, order_vocab_cnt[k])
# Unigram-product + uniform prior (no self-pooled n-gram count in alpha_w): 85% per-order unigram product + 15% per-order uniform
# (zero self-pooled c_llm term in alpha_w, so c_llm never cancels and there is no pooled ceiling!)
p_bg = 0.85 * p_indep + 0.15 * p_unif
alpha_w = alpha_0 * p_bg
num_l = c_llm + alpha_w
den_l = max(n_llm + alpha_0 - num_l, 1e-9)
num_h = c_hum + alpha_w
den_h = max(n_hum + alpha_0 - num_h, 1e-9)
delta = math.log(num_l / den_l) - math.log(num_h / den_h)
variance = (1.0 / num_l) + (1.0 / den_l) + (1.0 / num_h) + (1.0 / den_h)
z_score = delta / math.sqrt(variance)
# Also compute pure-pooled Monroe z/delta for reference
alpha_pool = alpha_0 * p_pooled
nl_p = c_llm + alpha_pool
dl_p = max(n_llm + alpha_0 - nl_p, 1e-9)
nh_p = c_hum + alpha_pool
dh_p = max(n_hum + alpha_0 - nh_p, 1e-9)
delta_pool = math.log(nl_p / dl_p) - math.log(nh_p / dh_p)
var_pool = (1.0 / nl_p) + (1.0 / dl_p) + (1.0 / nh_p) + (1.0 / dh_p)
z_pool = delta_pool / math.sqrt(var_pool)
if z_score >= z_threshold:
mech, cat = classify_remediation_mechanism(ng)
raw_results.append({
"ngram": list(ng),
"phrase": " ".join(ng),
"order": len(ng),
"remediation_mechanism": mech,
"speech_act_category": cat,
"z_score": round(z_score, 3),
"log_odds_delta": round(delta, 3),
"pooled_z_score": round(z_pool, 3),
"pooled_log_odds_delta": round(delta_pool, 3),
"alpha_w": round(alpha_w, 6),
"llm_count": c_llm,
"human_count": c_hum,
"llm_rate_per_10k": round(llm_rate, 3),
"human_rate_per_10k": round(hum_rate, 3),
"num_providers": num_providers,
"num_families": num_families,
"model_families": dict(fam_dist.most_common()),
"top_providers": dict(p_dist.most_common(4)),
})
raw_results.sort(key=lambda x: x["z_score"], reverse=True)
# Deduplicate sub-phrases that are >=85% explained by a longer selected super-phrase
by_tuple = {tuple(x["ngram"]): x for x in raw_results}
deduped = []
for item in raw_results:
ng = tuple(item["ngram"])
subsumed = False
for other_ng, other_item in by_tuple.items():
if len(other_ng) > len(ng) and other_item["llm_count"] >= 0.85 * item["llm_count"]:
for start in range(len(other_ng) - len(ng) + 1):
if other_ng[start : start + len(ng)] == ng:
subsumed = True
break
if subsumed:
break
if not subsumed:
deduped.append(item)
return deduped
def first_word_cap(phrase: str) -> str:
return (phrase[0].upper() + phrase[1:]) if phrase else phrase
def generate_phrase_surface_variants(phrase: str) -> List[str]:
"""
Generates surface variants for Qwen3.8-Flash-Next BPE tokenization:
- lowercase phrase + first-word-capitalized phrase
- mid-phrase pronoun-`I` capitalized form (`so i'll` -> `so I'll`, `So I'll`)
- prefixes: bare, leading space (`" "`), newline (`"\\n"`), quote (`'"'`, `' "'`),
and open-paren/bracket (`"("`, `" ("`, `"["`, `" ["`) which Qwen's pretokenizer merges with the first word.
"""
i_cased = PRONOUN_I_RE.sub("I", phrase)
bases = []
for b in (phrase, first_word_cap(phrase), i_cased, first_word_cap(i_cased)):
if b and b not in bases:
bases.append(b)
variants: List[str] = []
seen: Set[str] = set()
for b in bases:
for prefix in ("", " ", "\n", '"', ' "', "(", " (", "[", " ["):
v = prefix + b
if v not in seen:
seen.add(v)
variants.append(v)
return variants
def map_phrases_to_qwen_token_tuples(
phrases: List[str],
tokenizer_name: str = "Qwen/Qwen3.8-Flash-Next",
allow_no_tokenizer: bool = False,
) -> Tuple[List[List[int]], List[str], Dict[str, List[List[int]]]]:
surface_set: List[str] = []
seen_surf: Set[str] = set()
per_phrase_variants: Dict[str, List[str]] = {}
for phrase in phrases:
p_vars = generate_phrase_surface_variants(phrase)
per_phrase_variants[phrase] = p_vars
for v in p_vars:
if v not in seen_surf:
seen_surf.add(v)
surface_set.append(v)
encode_fn = None
last_err: Optional[Exception] = None
try:
from tokenizers import Tokenizer
fast_tok = Tokenizer.from_pretrained(tokenizer_name)
encode_fn = lambda s: fast_tok.encode(s, add_special_tokens=False).ids
except Exception as e1:
last_err = e1
try:
from transformers import AutoTokenizer
hf_tok = AutoTokenizer.from_pretrained(tokenizer_name, trust_remote_code=True)
encode_fn = lambda s: hf_tok.encode(s, add_special_tokens=False)
except Exception as e2:
last_err = e2
if encode_fn is None:
if allow_no_tokenizer:
return [], surface_set, {}
raise RuntimeError(
f"Failed to load tokenizer '{tokenizer_name}' via `tokenizers` or `transformers` "
f"(run with `uv run --with tokenizers` or pass `--allow-no-tokenizer`): {last_err}"
)
token_tuples: Set[Tuple[int, ...]] = set()
phrase_to_tuples: Dict[str, List[List[int]]] = {}
for phrase, p_vars in per_phrase_variants.items():
p_set: Set[Tuple[int, ...]] = set()
for variant in p_vars:
ids = encode_fn(variant)
if 2 <= len(ids) <= 8:
t = tuple(ids)
token_tuples.add(t)
p_set.add(t)
phrase_to_tuples[phrase] = [list(t) for t in sorted(p_set)]
return [list(t) for t in sorted(token_tuples)], surface_set, phrase_to_tuples
def run_self_test() -> None:
# 1. Clause boundary & NO phantom n-grams across inline code, numbers, URLs, or hyphenated names
ngs = extract_ngrams("Let me check. Rather than guessing!\nFeel free to ask.")
phrases = {" ".join(x) for x in ngs}
assert "let me" in phrases and "let me check" in phrases
assert "rather than" in phrases and "feel free to" in phrases
assert "check rather" not in phrases, f"Crossed period boundary: {phrases}"
assert "guessing feel" not in phrases, f"Crossed newline/! boundary: {phrases}"
phantom_clauses = extract_clauses(
"run `make check` then commit; K320 is fast, 42 tests pass; see https://example.com/a now please; on hyd-eu-1 and break-check"
)
flat_pairs = {
(w[i], w[i + 1]) for w in phantom_clauses for i in range(len(w) - 1)
}
for forbidden in (("run", "then"), ("fast", "tests"), ("see", "now"), ("hyd", "eu"), ("break", "check")):
assert forbidden not in flat_pairs, f"Phantom n-gram leaked: {forbidden} in {phantom_clauses}"
# 2. Curly apostrophe normalization & human quote stripping
curly_clauses = extract_clauses("Wait — don’t worry, I’ll check it.", is_human=False)
assert ["don't", "worry"] in curly_clauses and ["i'll", "check", "it"] in curly_clauses, curly_clauses
hum_quoted = extract_clauses("> let me check that for you\ni want this fixed", is_human=True)
assert all("let" not in c for c in hum_quoted), f"Quoted assistant line leaked into human clauses: {hum_quoted}"
# 3. Sub-tuple protected stop n-gram filtering ("the full suite" rejected because ("full", "suite") is protected)
assert not is_stylistic_candidate(("full", "suite"))
assert not is_stylistic_candidate(("the", "full", "suite"))
assert not is_stylistic_candidate(("is", "clean"))
assert not is_stylistic_candidate(("still", "running"))
assert not is_stylistic_candidate(("i", "don't", "know"))
assert not is_stylistic_candidate(("i'm", "not", "sure"))
assert is_stylistic_candidate(("let", "me", "check"))
assert is_stylistic_candidate(("so", "i'll"))
# 4. ETF round-trip decoding (tags 97, 98, 100, 104, 109, 116, and zlib-compressed tag 80)
inner_etf = (
b"\x68\x02" # SMALL_TUPLE_EXT arity 2
b"\x64\x00\x04user" # ATOM_EXT "user"
b"\x6d\x00\x00\x00\x05hello" # BINARY_EXT "hello"
)
val, pos = decode_etf_term(inner_etf, 0)
assert val == (("atom", "user"), "hello") and pos == len(inner_etf)
comp_etf = b"\x50" + struct.pack(">I", len(inner_etf)) + zlib.compress(inner_etf)
val_c, _ = decode_etf_term(comp_etf, 0)
assert val_c == (("atom", "user"), "hello")
# 5. Fail-closed on empty human baseline, non-constant delta at c_hum=0, and model-family filter
try:
compute_fightin_words_z_scores(
Counter({("let", "me"): 20}),
Counter(),
{("let", "me"): Counter({"albedo:claude": 10, "albedo:alibaba": 10})},
)
raise AssertionError("Should have failed on empty human baseline")
except ValueError:
pass
# Verify Claude-only multi-tag ("albedo:tangled-clod" + "cc:claude-opus-5") is rejected when min_families=2
claude_only = compute_fightin_words_z_scores(
Counter({("let", "me"): 50}),
Counter({("you", "want"): 10}),
{("let", "me"): Counter({"albedo:tangled-clod": 30, "cc:claude-opus-5": 20})},
min_count=10,
min_providers=2,
min_families=2,
)
assert len(claude_only) == 0, f"Claude-only multi-tag should be filtered by min_families=2: {claude_only}"
# Verify c_hum=0 phrases with different counts do not collapse to identical log_odds_delta
multi_res = compute_fightin_words_z_scores(
Counter({("let", "me"): 200, ("so", "i'll"): 40}),
Counter({("you", "want"): 50}),
{
("let", "me"): Counter({"albedo:claude": 100, "albedo:alibaba": 100}),
("so", "i'll"): Counter({"albedo:claude": 20, "albedo:alibaba": 20}),
},
unigram_counts=Counter({"let": 210, "me": 250, "so": 400, "i'll": 150, "you": 100, "want": 100}),
min_count=10,
min_providers=2,
min_families=2,
alpha_0=10.0,
z_threshold=0.1,
)
assert len(multi_res) == 2
assert multi_res[0]["log_odds_delta"] != multi_res[1]["log_odds_delta"], (
f"c_hum=0 phrases collapsed to identical delta: {multi_res}"
)
# 5b. Verify 3-gram c_hum=0 phrases at realistic alpha_0=5000.0 do NOT collapse to identical delta
trigram_res = compute_fightin_words_z_scores(
Counter({("let", "me"): 50000, ("let", "me", "check"): 3000, ("if", "you", "want"): 1200}),
Counter({("you", "want"): 8000}),
{
("let", "me"): Counter({"albedo:claude": 25000, "albedo:alibaba": 25000}),
("let", "me", "check"): Counter({"albedo:claude": 1500, "albedo:alibaba": 1500}),
("if", "you", "want"): Counter({"albedo:claude": 600, "albedo:alibaba": 600}),
},
unigram_counts=Counter({"let": 53000, "me": 53000, "check": 3500, "if": 9000, "you": 15000, "want": 10000}),
min_count=10,
min_providers=2,
min_families=2,
alpha_0=5000.0,
z_threshold=0.5,
)
tri_by_p = {r["phrase"]: r for r in trigram_res}
assert "let me check" in tri_by_p and "if you want" in tri_by_p, tri_by_p
assert abs(tri_by_p["let me check"]["log_odds_delta"] - tri_by_p["if you want"]["log_odds_delta"]) > 0.2, (
f"3-gram c_hum=0 phrases should have distinct deltas under unigram prior: {tri_by_p}"
)
# Numeric z & delta recomputation check on ("let", "me", "check")
row_lmc = tri_by_p["let me check"]
aw = row_lmc["alpha_w"]
nl_tot, nh_tot = 54200.0, 8000.0
num_l, den_l = 3000.0 + aw, nl_tot + 5000.0 - (3000.0 + aw)
num_h, den_h = 0.0 + aw, nh_tot + 5000.0 - aw
expected_delta = math.log(num_l / den_l) - math.log(num_h / den_h)
expected_var = (1.0 / num_l) + (1.0 / den_l) + (1.0 / num_h) + (1.0 / den_h)
expected_z = expected_delta / math.sqrt(expected_var)
assert abs(row_lmc["log_odds_delta"] - round(expected_delta, 3)) <= 1e-3
assert abs(row_lmc["z_score"] - round(expected_z, 3)) <= 1e-3
# 6. CC .jsonl isSidechain & Albedo SQLite user/other row parsing test
import tempfile
with tempfile.TemporaryDirectory() as tmpd:
cc_p = os.path.join(tmpd, "test_cc.jsonl")
with open(cc_p, "w", encoding="utf-8") as f:
f.write(json.dumps({"isSidechain": True, "message": {"role": "user", "content": "subagent prompt let me check"}}) + "\n")
f.write(json.dumps({"isSidechain": False, "message": {"role": "user", "content": "real human turn"}}) + "\n")
f.write(json.dumps({"isSidechain": False, "message": {"role": "assistant", "model": "claude-opus-5", "content": "let me check that"}}) + "\n")
a_cc, h_cc = iter_claude_code_turns([cc_p])
assert len(a_cc) == 1 and a_cc[0] == ("cc:claude-opus-5", "let me check that"), a_cc
assert len(h_cc) == 1 and h_cc[0] == "real human turn", h_cc
db_p = os.path.join(tmpd, "test_albedo.sqlite")
con = sqlite3.connect(db_p)
con.execute("CREATE TABLE session_family (session TEXT, parent TEXT, depth INT)")
con.execute("INSERT INTO session_family VALUES ('child_sess', 'root_sess', 1)")
con.execute("CREATE TABLE submission_events (payload BLOB)")
con.execute("CREATE TABLE transcript (session TEXT, provider TEXT, row_class TEXT, payload BLOB)")
# Root user row vs child user row
root_u_etf = b"\x83\x68\x02\x61\x01\x68\x02\x64\x00\x04user\x6d\x00\x00\x00\x0bhello world"
child_u_etf = b"\x83\x68\x02\x61\x01\x68\x02\x64\x00\x04user\x6d\x00\x00\x00\x0bchild task!"
con.execute("INSERT INTO transcript VALUES ('root_sess', 'claude', 'user', ?)", (root_u_etf,))
con.execute("INSERT INTO transcript VALUES ('child_sess', 'claude', 'user', ?)", (child_u_etf,))
con.commit()
con.close()
a_alb, h_alb, errs = iter_albedo_sqlite_turns(db_p)
assert errs == 0 and h_alb == ["hello world"], (a_alb, h_alb, errs)
# 7. Surface variants include mid-phrase pronoun `I` and paren/bracket prefixes
assert first_word_cap("let me check JSON") == "Let me check JSON"
s_vars = generate_phrase_surface_variants("so i'll")
for expected_v in ("so i'll", "so I'll", "So I'll", " so I'll", "(so i'll", " (so I'll", "[so i'll"):
assert expected_v in s_vars, f"Missing variant {expected_v!r} in {s_vars}"
print("mine_llm_isms_from_transcripts.py self-test: PASS")
def main():
parser = argparse.ArgumentParser(description="Mine LLM-ism 2/3/4-grams against Dawn's human baseline.")
parser.add_argument("--self-test", action="store_true", help="Run unit self-tests and exit")
parser.add_argument(
"--transcripts-dir",
type=str,
default=str(Path.home() / ".claude/projects"),
help="Directory of Claude Code / JSONL transcripts",
)
parser.add_argument(
"--albedo-sqlite",
type=str,
default=str(Path.home() / ".albedo/albedo.sqlite"),
help="Path to Albedo SQLite database",
)
parser.add_argument(
"--human-ref-dir",
type=str,
default=None,
help="Optional extra directory of human reference text",
)
parser.add_argument(
"--exclude-albedo-assistant",
action="store_true",
help="Mine only from unstyled non-Albedo assistant turns",
)
parser.add_argument(
"--allow-no-tokenizer",
action="store_true",
help="Allow running without Qwen tokenizer installed",
)
parser.add_argument("--top-k", type=int, default=350, help="Number of top LLM-ism n-grams to output")
parser.add_argument("--min-count", type=int, default=15, help="Minimum LLM occurrences")
parser.add_argument("--min-providers", type=int, default=2, help="Minimum distinct provider tags")
parser.add_argument("--min-families", type=int, default=2, help="Minimum distinct model families")
parser.add_argument("--alpha-0", type=float, default=5000.0, help="Monroe informative Dirichlet prior strength alpha_0")
parser.add_argument("--z-threshold", type=float, default=1.65, help="Minimum Monroe z-score")
parser.add_argument("--output", type=str, default="results/banned_llm_isms.json", help="Output JSON path")
args = parser.parse_args()
if args.self_test:
run_self_test()
return
asst_turns: List[Tuple[str, str]] = []
human_turns: List[str] = []
etf_decode_errors = 0
if args.transcripts_dir and os.path.exists(args.transcripts_dir):
t_files = []
for ext in ("*.jsonl", "*.json"):
t_files.extend(glob.glob(os.path.join(args.transcripts_dir, "**", ext), recursive=True))
a_cc, h_cc = iter_claude_code_turns(sorted(t_files))
asst_turns.extend(a_cc)
human_turns.extend(h_cc)
if args.albedo_sqlite and os.path.exists(args.albedo_sqlite):
a_alb, h_alb, etf_errs = iter_albedo_sqlite_turns(args.albedo_sqlite)
etf_decode_errors += etf_errs
human_turns.extend(h_alb)
if not args.exclude_albedo_assistant:
asst_turns.extend(a_alb)
if args.human_ref_dir and os.path.exists(args.human_ref_dir):
h_files = []
for ext in ("*.md", "*.txt", "*.jsonl", "*.json"):
h_files.extend(glob.glob(os.path.join(args.human_ref_dir, "**", ext), recursive=True))
for p in sorted(h_files):
with open(p, "r", encoding="utf-8", errors="ignore") as f:
raw = f.read().strip()
if raw:
human_turns.append(raw)
llm_counts: Counter = Counter()
provider_counts_per_ng: Dict[Tuple[str, ...], Counter] = defaultdict(Counter)
turns_by_provider: Counter = Counter()
turns_by_family: Counter = Counter()
ngrams_by_provider: Counter = Counter()
unigram_counts: Counter = Counter()
for prov, raw_prose in asst_turns:
turns_by_provider[prov] += 1
turns_by_family[provider_to_family(prov)] += 1
clauses = extract_clauses(raw_prose, is_human=False)
ng_cnt = 0
for words in clauses:
for w in words:
unigram_counts[w] += 1
for n in (2, 3, 4):
if len(words) < n:
continue
for i in range(len(words) - n + 1):
ng = tuple(words[i : i + n])
if is_stylistic_candidate(ng):
llm_counts[ng] += 1
provider_counts_per_ng[ng][prov] += 1
ng_cnt += 1
ngrams_by_provider[prov] += ng_cnt
human_counts: Counter = Counter()
for raw_prose in human_turns:
clauses = extract_clauses(raw_prose, is_human=True)
for words in clauses:
for w in words:
unigram_counts[w] += 1
for n in (2, 3, 4):
if len(words) < n:
continue
for i in range(len(words) - n + 1):
ng = tuple(words[i : i + n])
if is_stylistic_candidate(ng):
human_counts[ng] += 1
ranked = compute_fightin_words_z_scores(
llm_counts=llm_counts,
human_counts=human_counts,
provider_counts_per_ng=provider_counts_per_ng,
unigram_counts=unigram_counts,
min_count=args.min_count,
min_providers=args.min_providers,
min_families=args.min_families,
alpha_0=args.alpha_0,
z_threshold=args.z_threshold,
)
selected = ranked[: args.top_k]
per_provider_top = {}
for prov, p_turn_cnt in turns_by_provider.most_common():
if p_turn_cnt < 50:
continue
p_counts = Counter({ng: dist[prov] for ng, dist in provider_counts_per_ng.items() if dist[prov] > 0})
p_ranked = compute_fightin_words_z_scores(
llm_counts=p_counts,
human_counts=human_counts,
provider_counts_per_ng=provider_counts_per_ng,
unigram_counts=unigram_counts,
min_count=max(5, args.min_count // 2),
min_providers=1,
min_families=1,
alpha_0=args.alpha_0,
z_threshold=1.5,
)
per_provider_top[prov] = [
{
"phrase": x["phrase"],
"z_score": x["z_score"],
"count": x["llm_count"],
"human_count": x["human_count"],
}
for x in p_ranked[:15]
]
phrases = [x["phrase"] for x in selected]
qwen_token_tuples, surface_variants, phrase_to_tuples = map_phrases_to_qwen_token_tuples(
phrases, allow_no_tokenizer=args.allow_no_tokenizer
)
for item in selected:
item["qwen_token_id_tuples"] = phrase_to_tuples.get(item["phrase"], [])
phase3_speech_acts = [x for x in selected if x["remediation_mechanism"] == "phase3_sentence_surgery"]
phase4_lexical = [x for x in selected if x["remediation_mechanism"] == "phase4_lexical_ftpo"]
ftpo_tuple_set = sorted({tuple(t) for x in phase4_lexical for t in x.get("qwen_token_id_tuples", [])})
ftpo_qwen_tuples = [list(t) for t in ftpo_tuple_set]
os.makedirs(os.path.dirname(os.path.abspath(args.output)), exist_ok=True)
payload = {
"summary": {
"total_assistant_turns": len(asst_turns),
"total_human_turns": len(human_turns),
"etf_decode_errors": etf_decode_errors,
"total_llm_ngrams": sum(llm_counts.values()),
"total_human_ngrams": sum(human_counts.values()),
"total_unigram_tokens": sum(unigram_counts.values()),
"unique_unigram_vocab": len(unigram_counts),
"assistant_turns_by_family": dict(turns_by_family.most_common()),
"assistant_turns_by_provider": dict(turns_by_provider.most_common()),
"assistant_ngrams_by_provider": dict(ngrams_by_provider.most_common()),
"monroe_alpha_0": args.alpha_0,
"prior_type": "0.85_per_order_unigram_plus_0.15_per_order_uniform",
"min_count": args.min_count,
"min_providers": args.min_providers,
"min_families": args.min_families,
"z_threshold": args.z_threshold,
"selected_ngrams_count": len(selected),
"phase3_speech_act_ngrams_count": len(phase3_speech_acts),
"phase4_lexical_ftpo_ngrams_count": len(phase4_lexical),
"surface_variants_count": len(surface_variants),
"qwen_token_tuples_count": len(qwen_token_tuples),
"ftpo_qwen_token_tuples_count": len(ftpo_qwen_tuples),
},
"per_provider_top_15": per_provider_top,
"banned_ngrams": selected,
"phase3_speech_act_ngrams": phase3_speech_acts,
"phase4_lexical_ftpo_ngrams": phase4_lexical,
"surface_variants": surface_variants,
"qwen_token_id_tuples": qwen_token_tuples,
"ftpo_qwen_token_id_tuples": ftpo_qwen_tuples,
}
with open(args.output, "w", encoding="utf-8") as f:
json.dump(payload, f, indent=2, ensure_ascii=False)
print(
f"Mined {len(selected)} LLM-isms from {len(asst_turns)} assistant turns vs {len(human_turns)} human turns "
f"({sum(llm_counts.values()):,} LLM n-grams vs {sum(human_counts.values()):,} human n-grams, "
f"etf_errors={etf_decode_errors}) -> {args.output}"
)
if __name__ == "__main__":
main()