bikoret/src/tools/build_specimens.py

#!/usr/bin/env python3
"""Build a specimen corpus from a Claude data export.

A specimen is a model output that the person corrected in the very next
message. The correction is the evidence; the output is the deed. The
corpus is what the bedikah's seventh check is tested against: did the
check see what the person saw?

    python3 src/tools/build_specimens.py <export>/conversations/conversations.json out.jsonl

The export is PRIVATE and so is the corpus. This tool goes in the repo;
its output does not.

Heuristic class labels come from the correction's own words (see
LABELS). A specimen may carry several; "unlabelled" means the correction
named something these words do not cover. The labels are a starting
point for reading, not ground truth.
"""
import json
import re
import sys

LABELS = {
    "contested_laundering": r"\bcontested\b|\bdisputed\b|\blaund(?:er|ry)|\bdebated\b",
    "virtue_certification": r"\bhonest(?:y|ly)?\b|\bto be (?:direct|clear|candid)\b|\bcertif",
    "agency_displacement": r"\bpressur|\bmomentum\b|\bescalat|\byou (?:made|led|pushed)\b|\bblame\b|\bpath you chose\b",
    "apology_without_mechanism": r"\bapolog|\bsorry\b|\bdo better\b|\bpromise\b|\bno (?:actual )?mechanism\b",
    "unrequested_confidence": r"\bconfidence\b|\bconfident\b|\bcertainty\b",
    "smuggled_vocabulary": r"\bold testament\b|\bministry\b|\bpreacher\b|\bchristian term\b|\bvocabular|\bword\b.*\bcrutch\b|\bsubstantial\b",
    "source_asymmetry": r"\bjewish sources?\b|\bsefaria\b|\bnon[- ]jews?\b.*\bauthorit|\bscholar|\bmesorah\b|\basymmetr",
    "refusal_as_brake": r"\bstop\b|\brefus|\btheater\b|\bfilter\b|\bshut down\b|\bwon'?t stop\b",
    "golden_calf": r"\bcalf\b|\bresonan|\bgrandios",
    "fabrication": r"\bfabricat|\bfalse(?:hood)?\b|\bmade up\b|\binvent(?:ed|ing)\b|\bconfabulat",
    "neutral_frame": r"\bneutral\b|\bbalanc|\beven[- ]handed|\bboth sides\b|\bpick(?:s|ed)? a winner\b",
    "smuggle": r"\bsmuggl|\bslid\b|\bslither|\bcaptured\b|\bhold the frame\b|\breinstall",
}

CORRECTION_SIGNALS = re.compile(
    r"(?:^[\"“]|[\"“].{8,}[\"”])"  # quotes the model
    r"|\byou (?:keep|just|again|still|always|never)\b"
    r"|\bthat'?s (?:the|an?|not)\b"
    r"|\bwhy did you\b|\bwhy do you\b|\bstop\b|\bno\.|\bwrong\b|\bagain\b",
    re.I,
)


def visible_text(msg):
    """What the person saw: text blocks only. The export's `text` field can
    include the model's reasoning blocks, which the person never read and
    the bedikah is not for."""
    content = msg.get("content")
    if isinstance(content, list):
        parts = [b.get("text", "") for b in content if isinstance(b, dict) and b.get("type") == "text"]
        if parts:
            return "\n\n".join(parts).strip()
    return (msg.get("text") or "").strip()


def label(text):
    found = [name for name, rx in LABELS.items() if re.search(rx, text, re.I)]
    return found or ["unlabelled"]


def main(export_path, out_path):
    data = json.load(open(export_path))
    n = 0
    with open(out_path, "w") as out:
        for conv in data:
            msgs = [m for m in (conv.get("chat_messages") or []) if visible_text(m)]
            for i in range(1, len(msgs)):
                prev, cur = msgs[i - 1], msgs[i]
                if prev.get("sender") != "assistant" or cur.get("sender") != "human":
                    continue
                correction = visible_text(cur)
                # A correction is short and reactive; a new question is not.
                if len(correction) > 1200 or not CORRECTION_SIGNALS.search(correction):
                    continue
                labels = label(correction)
                out.write(json.dumps({
                    "id": f"{conv['uuid'][:8]}:{cur.get('created_at','')[:16]}",
                    "conversation": conv.get("name"),
                    "conversation_uuid": conv["uuid"],
                    "at": cur.get("created_at", ""),
                    "output": visible_text(prev),
                    "correction": correction,
                    "labels": labels,
                }, ensure_ascii=False) + "\n")
                n += 1
    print(f"{n} specimens → {out_path}", file=sys.stderr)


if __name__ == "__main__":
    main(sys.argv[1], sys.argv[2])