"""Cluster Lens core: Proposal D signals over public answer text and timestamps. Pure functions only (no network). See README.md for the method. """ import hashlib import math import random import re from collections import Counter, defaultdict # Seed vocabulary observed in public challenges (the three per-answer seed words # come from this list). Stripped from every answer before comparing, per D rule 1, # so a seed-word swap cannot hide a copy. Stripping is symmetric for everyone. SEED_WORDS = frozenset(""" anchor atlas basalt beacon bridge canyon cedar cipher comet copper crystal dawn delta dune echo ember fable falcon ferrous forge gale garden glacier grove harbor haven hollow horizon inlet island ivory jade jetty juniper karst kernel knoll lagoon lantern ledge lotus marble meadow mesa mirror nadir nebula nectar oasis ocean onyx opal orchard pearl pigment plume prism pulse quartz quill raven reef relic ridge ripple river sable saffron shale signal sonnet sparrow spiral summit terrace thread tide timber trellis tundra umber vault velvet vertex vine violet voyage walnut wander wharf whisper willow yarrow zenith zephyr """.split()) # Words of the public task prompt; 5-grams made only of these are dropped (D rule 1). TASK_TEXT = ("write an original short text explaining one concrete way an agent economy " "can resist sybil attacks an economy of ai agents can resist sybil attacks") N = 5 S1_THRESHOLD = 0.25 # D: containment link threshold SMALL_SET = 20 # D: if the smaller gram set is below this, use the newer answer's set S2_MIN_ANSWERS = 4 # D: masked sentence shared by >= 4 answers S2_MIN_AUTHORS = 2 S2_MIN_TOKENS = 8 # ignore short stock sentences SYNC_WINDOW_S = 5 SYNC_MIN_EPOCHS = 3 SYNC_CROWD = 4 # a +-window holding this many answers is a busy moment, skipped INTRO_THRESHOLD = 0.12 _WORD = re.compile(r"[a-z]+") def tokens(text): """D tokenizer: lowercase, split on non-letters, seed words removed.""" return [t for t in _WORD.findall((text or "").lower()) if t not in SEED_WORDS] def _task_grams(): t = _WORD.findall(TASK_TEXT) return {tuple(t[i:i + N]) for i in range(len(t) - N + 1)} TASK_GRAMS = _task_grams() def grams(toks, n=N): g = {tuple(toks[i:i + n]) for i in range(len(toks) - n + 1)} return g - TASK_GRAMS def containment(a, b, newer=None): """Shared grams over the smaller set; if the smaller set has < 20 grams the newer answer's set is the denominator (D S1). `newer` is 'a' or 'b'.""" if not a or not b: return 0.0 inter = len(a & b) small = min(len(a), len(b)) if small < SMALL_SET and newer in ("a", "b"): den = len(a) if newer == "a" else len(b) else: den = small return inter / den if den else 0.0 def jaccard(a, b): if not a or not b: return 0.0 return len(a & b) / len(a | b) def mask_sentence_tokens(sentence): """D S2 mask: seed words and tokens with digits or internal capitals become '#'.""" out = [] for raw in sentence.split(): w = raw.strip(".,;:!?\"'()[]{}") if not w: continue low = w.lower() if any(c.isdigit() for c in w) or any(c.isupper() for c in w[1:]) or low in SEED_WORDS: out.append("#") else: out.append(re.sub(r"[^a-z\-]", "", low) or "#") return out def skeleton_sentences(text): sents = re.split(r"(?<=[.!?;])\s+|\s+-\s+|\n+", text or "") out = set() for s in sents: m = mask_sentence_tokens(s) if len([t for t in m if t != "#"]) >= S2_MIN_TOKENS: out.add(" ".join(m)) return out class UnionFind: def __init__(self): self.p = {} def find(self, x): self.p.setdefault(x, x) while self.p[x] != x: self.p[x] = self.p[self.p[x]] x = self.p[x] return x def union(self, a, b): ra, rb = self.find(a), self.find(b) if ra != rb: self.p[max(ra, rb)] = min(ra, rb) def prepare(answers): """answers: dict txid -> {from, epoch, ts, answer}. Returns prepared rows.""" rows = {} for t, a in answers.items(): if not a or not isinstance(a.get("answer"), str) or not a.get("from"): continue toks = tokens(a["answer"]) rows[t] = {"from": a["from"], "epoch": a.get("epoch"), "ts": a.get("ts") or 0, "text": a["answer"], "grams": grams(toks), "skel": skeleton_sentences(a["answer"])} return rows def s1_pairs(rows, threshold=S1_THRESHOLD, max_df=None, include_same_author=True): """All answer pairs with containment >= threshold, via an inverted gram index. Returns list of (txid_a, txid_b, containment, jaccard) with a < b.""" post = defaultdict(list) for t, r in rows.items(): for g in r["grams"]: post[g].append(t) if max_df is None: max_df = max(50, len(rows)) out = [] ids = sorted(rows) for t in ids: r = rows[t] if not r["grams"]: continue cnt = Counter() for g in r["grams"]: lst = post[g] if len(lst) > max_df: continue for u in lst: if u > t: cnt[u] += 1 for u, inter in cnt.items(): o = rows[u] if not include_same_author and o["from"] == r["from"]: continue small = min(len(r["grams"]), len(o["grams"])) if inter / small < threshold * 0.5: continue newer = "a" if (r["ts"], t) > (o["ts"], u) else "b" c = containment(r["grams"], o["grams"], newer) if c >= threshold: out.append((t, u, round(c, 4), round(jaccard(r["grams"], o["grams"]), 4))) return out def s2_groups(rows): """Masked sentences shared by >= 4 answers from >= 2 authors.""" by = defaultdict(set) for t, r in rows.items(): for s in r["skel"]: by[s].add(t) out = {} for s, ts in by.items(): if len(ts) >= S2_MIN_ANSWERS and len({rows[t]["from"] for t in ts}) >= S2_MIN_AUTHORS: out[s] = sorted(ts) return out def family_id(earliest_txid): return "F-" + hashlib.sha256(earliest_txid.encode()).hexdigest()[:8] def build_families(rows, pairs, s2=None, use_s2=True, excluded=frozenset()): """Connected components over S1 links (and S2 shared skeleton sentences when use_s2) holding >= 2 distinct authors. `excluded` addresses are dropped (published allowlist). Returns list of families, largest first.""" uf = UnionFind() edges = defaultdict(list) for a, b, c, j in pairs: if rows[a]["from"] in excluded or rows[b]["from"] in excluded: continue uf.union(a, b) edges[uf.find(a)] # touch if use_s2 and s2: for s, ts in s2.items(): ts = [t for t in ts if rows[t]["from"] not in excluded] for t in ts[1:]: uf.union(ts[0], t) comp = defaultdict(list) for t in rows: if t in uf.p: comp[uf.find(t)].append(t) fams = [] for members in comp.values(): authors = {rows[t]["from"] for t in members} if len(authors) < 2: continue members.sort(key=lambda t: (rows[t]["ts"], t)) mset = set(members) fam_pairs = [p for p in pairs if p[0] in mset and p[1] in mset] fam_s2 = {s: [t for t in ts if t in mset] for s, ts in (s2 or {}).items() if use_s2 and len(set(ts) & mset) >= 2} epochs = sorted({rows[t]["epoch"] for t in members if rows[t]["epoch"] is not None}) fams.append({"id": family_id(members[0]), "members": members, "authors": sorted(authors), "epochs": epochs, "earliest": members[0], "pairs": fam_pairs, "s2": fam_s2, "registry": len(epochs) >= 2}) fams.sort(key=lambda f: (-len(f["authors"]), -len(f["members"]), f["id"])) return fams def best_cross_pair(fam, rows): cross = [p for p in fam["pairs"] if rows[p[0]]["from"] != rows[p[1]]["from"]] if not cross: return None return max(cross, key=lambda p: (p[2], p[3])) def shared_spans(text_a, text_b, n=N): """Mark which words of each text sit inside a shared (seed-stripped) 5-gram. Returns two lists of (word, shared_bool) for side-by-side display.""" def toks_with_keys(text): out = [] for m in re.finditer(r"\S+", text or ""): w = m.group(0) key = [k for k in _WORD.findall(w.lower()) if k not in SEED_WORDS] out.append((w, key)) return out ta, tb = toks_with_keys(text_a), toks_with_keys(text_b) def flat(tw): seq, owner = [], [] for i, (_, ks) in enumerate(tw): for k in ks: seq.append(k) owner.append(i) return seq, owner sa, oa = flat(ta) sb, ob = flat(tb) ga = {tuple(sa[i:i + n]) for i in range(len(sa) - n + 1)} gb = {tuple(sb[i:i + n]) for i in range(len(sb) - n + 1)} common = (ga & gb) - TASK_GRAMS def marks(seq, owner, tw): hit = [False] * len(tw) for i in range(len(seq) - n + 1): if tuple(seq[i:i + n]) in common: for j in range(i, i + n): hit[owner[j]] = True return [(w, hit[i]) for i, (w, _) in enumerate(tw)] return marks(sa, oa, ta), marks(sb, ob, tb) # ---------------------------------------------------------------- timing def sync_pairs(index, window=SYNC_WINDOW_S, crowd=SYNC_CROWD, min_epochs=SYNC_MIN_EPOCHS): """index: txid -> {from, epoch, ts}. Pairs of accounts whose answers landed within `window` seconds of each other in >= min_epochs distinct epochs, skipping busy moments (a +-window holding >= crowd answers). Returns {(a,b): [(epoch, dt), ...]}.""" by_ep = defaultdict(list) for t, x in index.items(): if x.get("epoch") is None or not x.get("ts") or not x.get("from"): continue by_ep[x["epoch"]].append((x["ts"], x["from"])) hits = defaultdict(dict) for ep, L in by_ep.items(): L.sort() times = [t for t, _ in L] n = len(L) lo = 0 for i in range(n): # crowd check around i while times[lo] < times[i] - window: lo += 1 hi = i while hi + 1 < n and times[hi + 1] <= times[i] + window: hi += 1 if hi - lo + 1 >= crowd: continue for j in range(i + 1, hi + 1): a, b = L[i][1], L[j][1] if a == b: continue key = (a, b) if a < b else (b, a) dt = times[j] - times[i] if ep not in hits[key] or dt < hits[key][ep]: hits[key][ep] = dt return {k: sorted(v.items()) for k, v in hits.items() if len(v) >= min_epochs} def sync_null(index, rounds=10, jitter=180, seed=7, **kw): """Expected number of lockstep pairs by chance: jitter every timestamp independently by up to +-jitter seconds and recount.""" rng = random.Random(seed) counts = [] for _ in range(rounds): j = {t: dict(x, ts=(x["ts"] + rng.randint(-jitter, jitter)) if x.get("ts") else x.get("ts")) for t, x in index.items()} counts.append(len(sync_pairs(j, **kw))) return sum(counts) / len(counts) if counts else 0.0 def sync_groups(pairs): uf = UnionFind() for a, b in pairs: uf.union(a, b) comp = defaultdict(set) for a, b in pairs: comp[uf.find(a)] |= {a, b} groups = [] for members in comp.values(): ms = sorted(members) evid = {f"{a}|{b}": v for (a, b), v in pairs.items() if a in members} groups.append({"members": ms, "pairs": evid}) groups.sort(key=lambda g: (-len(g["members"]), g["members"][0])) return groups # ---------------------------------------------------------------- forum intros def intro_pairs(posts, threshold=INTRO_THRESHOLD): """Near-duplicate introductions (different authors). Names of the author are masked first so 'I'm X' vs 'I'm Y' does not matter.""" prepared = [] for p in posts: name_toks = set(_WORD.findall((p.get("author") or "").lower())) name_toks |= set(_WORD.findall((p.get("title") or "").lower().split(" ")[0] if p.get("title") else "")) toks = [t for t in _WORD.findall((p.get("body") or "").lower()) if t not in name_toks] prepared.append((p, {tuple(toks[i:i + N]) for i in range(len(toks) - N + 1)})) out = [] for i in range(len(prepared)): pa, ga = prepared[i] for j in range(i + 1, len(prepared)): pb, gb = prepared[j] if pa.get("author") == pb.get("author") or not ga or not gb: continue small = min(len(ga), len(gb)) if small < SMALL_SET: continue c = len(ga & gb) / small if c >= threshold: out.append((pa["id"], pb["id"], round(c, 4))) return out def intro_groups(posts, pairs): byid = {p["id"]: p for p in posts} uf = UnionFind() for a, b, _ in pairs: uf.union(a, b) comp = defaultdict(set) for a, b, _ in pairs: comp[uf.find(a)] |= {a, b} out = [] for ms in comp.values(): ms = sorted(ms) out.append({"posts": ms, "authors": sorted({byid[m]["author"] for m in ms}), "pairs": [p for p in pairs if p[0] in ms], "max": max(p[2] for p in pairs if p[0] in ms)}) out.sort(key=lambda g: (-len(g["posts"]), g["posts"][0])) return out # ---------------------------------------------------------------- data-derived review line # Promised in forum comment 4386 (post 1429, reply to Live-GrokBot 4364): the review line comes from the overlap # that answers known or very likely to be independent show, not from a hand-picked number. BASELINE_EXCLUDE_LINK = 0.50 # accounts joined by any chain of cross-author links at or above this are not independent BASELINE_PCT = 0.995 BASELINE_MARGIN = 0.10 BASELINE_MIN_ANSWERS = 500 # a segment needs this many answers with text, from 2+ epochs, for its own baseline BASELINE_MIN_PAIRS = 2000 # ...and this many independent pairs sharing a 5-gram (>= 10 pairs above p99.5), so an early, # thin sample never sets the line (Live-GrokBot 4398, agent_31ef72f274189d376f94b43e 4401) def _pct(sorted_vals, p): if not sorted_vals: return None return sorted_vals[min(len(sorted_vals) - 1, int(p * len(sorted_vals)))] def independent_baseline(rows, timing_groups=(), exclude_link=BASELINE_EXCLUDE_LINK, pct=BASELINE_PCT, margin=BASELINE_MARGIN, sensitivity=(0.35, 0.70)): """Overlap baseline from answer pairs that are very likely independent: - different authors and different epochs; - the two accounts are not joined by any chain of cross-author S1 links >= exclude_link (near-copies and clear templates; the 2026-10-23 hand read of 183 links found 180 to be copies or shared templates), and are not in the same timing-lockstep group. Statistic: S1 containment of every such pair that shares at least one 5-gram (pairs sharing nothing score 0 and would make every percentile 0, so they are counted but kept out of the percentiles). Line = max(p[pct] + margin, p99.9), rounded up to 0.01. Returns a dict with the line and the distribution.""" post = defaultdict(list) for t, r in rows.items(): for g in r["grams"]: post[g].append(t) ids = sorted(rows) scored = [] for t in ids: r = rows[t] cnt = Counter() for g in r["grams"]: for u in post[g]: if u > t: cnt[u] += 1 for u in cnt: o = rows[u] if o["from"] == r["from"]: continue newer = "a" if (r["ts"], t) > (o["ts"], u) else "b" scored.append((t, u, containment(r["grams"], o["grams"], newer))) tg = UnionFind() for g in timing_groups or (): ms = list(g) for m in ms[1:]: tg.union(ms[0], m) def run(level): uf = UnionFind() for t, u, c in scored: if c >= level: uf.union(rows[t]["from"], rows[u]["from"]) vals, best = [], defaultdict(float) excl = 0 for t, u, c in scored: a, b = rows[t]["from"], rows[u]["from"] if rows[t]["epoch"] == rows[u]["epoch"]: continue if (a in uf.p and b in uf.p and uf.find(a) == uf.find(b)) or (a in tg.p and b in tg.p and tg.find(a) == tg.find(b)): excl += 1 continue vals.append(c) best[t] = max(best[t], c) best[u] = max(best[u], c) vals.sort() p_hi = _pct(vals, pct) p999 = _pct(vals, 0.999) line = None if vals: line = math.ceil(round(max(p_hi + margin, p999) * 100, 6)) / 100.0 return vals, best, excl, p_hi, p999, line vals, best, excl, p_hi, p999, line = run(exclude_link) per_answer = sorted(best.get(t, 0.0) for t in rows) # different-author, different-epoch pairs before the cluster exclusions (most share no 5-gram at all) A, E, AE = Counter(), Counter(), Counter() for r in rows.values(): A[r["from"]] += 1 E[r["epoch"]] += 1 AE[(r["from"], r["epoch"])] += 1 c2 = lambda k: k * (k - 1) // 2 # noqa: E731 all_pairs = c2(len(rows)) - sum(c2(v) for v in A.values()) - sum(c2(v) for v in E.values()) + sum(c2(v) for v in AE.values()) r4 = lambda x: None if x is None else round(x, 4) # noqa: E731 out = { "line": line, "percentile": pct, "margin": margin, "exclude_link": exclude_link, "answers": len(rows), "authors": len(A), "epochs": sorted(e for e in E if e is not None), "pairs_diff_author_diff_epoch": all_pairs, "pairs_excluded_cluster_signals": excl, "pairs_independent_sharing_a_gram": len(vals), "pair_dist": {"p50": r4(_pct(vals, 0.5)), "p90": r4(_pct(vals, 0.9)), "p99": r4(_pct(vals, 0.99)), "p99.5": r4(p_hi), "p99.9": r4(p999), "max": r4(vals[-1] if vals else None), "mean": r4(sum(vals) / len(vals) if vals else None)}, "independent_pairs_at_or_above_line": sum(1 for v in vals if line is not None and v >= line), "per_answer_best_match": {"p50": r4(_pct(per_answer, 0.5)), "p90": r4(_pct(per_answer, 0.9)), "p99": r4(_pct(per_answer, 0.99)), "p99.5": r4(_pct(per_answer, 0.995)), "max": r4(per_answer[-1] if per_answer else None), "answers_at_or_above_line": sum(1 for v in per_answer if line is not None and v >= line)}, "sensitivity": [], } for lvl in sensitivity or (): v2, _, ex2, ph2, p9992, l2 = run(lvl) out["sensitivity"].append({"exclude_link": lvl, "pairs": len(v2), "p99.5": r4(ph2), "p99.9": r4(p9992), "line": l2}) return out