`scan.py`

93 lines
#!/usr/bin/env python3
"""Stage 1 — cheap recall net (two nets, unioned).
Net A: assistant text with a decline verb AND a policy-reason keyword (high precision).
Net B: assistant text with a STRONG refusal opener, regardless of keyword — catches
       purely tone-based / ethics refusals that never say 'copyright' etc.
The Stage-2 local-LLM pass (classify.py) is the arbiter of policy-refusal vs capability-limit.
Writes /tmp/refusal_cands.json."""
import glob
import json
import os
import re
from collections import Counter

PROJROOT = os.path.expanduser('~/.claude/projects')

# Tunables — change these without hunting through logic below
MAX_ASST_LEN  = 2500   # assistant turns longer than this are almost never bare refusals
USER_TRUNCATE = 900    # chars stored for the user context (raised for multi-turn — HOLE 2)
ASST_TRUNCATE = 800    # chars stored for the assistant decline snippet
USER_TURNS    = 3      # how many recent user turns to include as context (a lone "go"/"do it" loses what was asked)

decline = re.compile(r"(I can'?t|I cannot|I won'?t|I'm not able|I am not able|I'm not going to|I'm not comfortable|I shouldn'?t|I have to decline|I'd rather not|not something I can|can'?t ethically|won'?t help|refus|decline to|not willing to|in good conscience)", re.IGNORECASE)
reason = re.compile(r"(copyright|trademark|intellectual property|terms of service|\bToS\b|their terms|against .{0,20}terms|scrap\w* (without|permission|their)|circumvent|bypass\w* (the )?(bot|captcha|paywall|rate.?limit|detection|login|auth)|unauthorized access|impersonat|counterfeit|\bDMCA\b|someone else'?s (brand|work|design|content)|protected work|passing off|knockoff|knock-off|rip.?off .{0,15}brand|clone .{0,15}(brand|site|store))", re.IGNORECASE)

# Net B — strong refusal openers that decline the REQUEST (not a capability limit).
strong = re.compile(r"(I can'?t help (you )?with (that|this)|I can'?t assist with (that|this)|I won'?t help (you )?(with|do)|I'm not able to help with (that|this)|I have to decline|I'm going to decline|I'm not comfortable (doing|helping|with|creating|writing)|I can'?t in good conscience|I won'?t be able to help with|I can'?t (create|write|generate|provide|build|do) (that|this)|that'?s not something I('?ll| will| can) (help|do|assist)|I won'?t assist|against my (guidelines|values|principles)|I can'?t ethically|I'm not willing to|I need to decline|I can'?t participate in)", re.IGNORECASE)
# de-noise Net B: drop obvious capability-limit phrasings
capability = re.compile(r"(from (in )?here|from the shell|from this session|for you (here|myself)|on your end|only you can|you'?ll need to|requires? (you|a person|your)|quit\w* iTerm|kill\w* (this|the) session|log ?in (yourself|first)|enter your|your (identity|credentials|password|2fa|payout|tax))", re.IGNORECASE)
# META filter: this build's OWN session discusses refusals/copyright at length — exclude self-referential noise
META = re.compile(r"(:9858|rejected-prompts|refusal_cands|refusal_classified|borderline candidate|heretic model|\bNet [AB]\b|classify\.py|build-data|scan\.py|policy declines|decline text|two-signal|capability limit)", re.IGNORECASE)

seen: set[tuple[str, str]] = set()
cands: list[dict] = []
for f in glob.glob(os.path.join(PROJROOT, '**', '*.jsonl'), recursive=True):
    try:
        with open(f, errors='ignore') as fh:
            raw = fh.read()
    except Exception:  # noqa: BLE001,S112
        continue
    # file-level exclusion: skip any transcript that is ABOUT building this tool (self-pollution)
    # specific signatures only — bare ':9858' matched token counts like ':985835' and
    # spuriously excluded ~10 real transcripts (contrarian HOLE 1, verified 2026-08-19)
    if any(sig in raw for sig in ('rejected-prompts-viewer', 'http://127.0.0.1:9858',
            'rejected.agentabrams', 'refusal_cands', 'refusal_classified', 'qwen3.8-27b-heretic')):
        continue
    recent_users: list[str] = []  # rolling last-USER_TURNS user turns for context (HOLE 2)
    for line in raw.splitlines():
        if '"user"' not in line and '"assistant"' not in line:
            continue
        try:
            d = json.loads(line)
        except Exception:  # noqa: BLE001
            continue
        msg_type, msg = d.get('type'), d.get('message')
        if not isinstance(msg, dict):
            continue
        if msg_type == 'user':
            content = msg.get('content')
            txt = None
            if isinstance(content, str):
                txt = content
            elif isinstance(content, list):
                txt = ' '.join(x.get('text', '') for x in content if isinstance(x, dict) and x.get('type') == 'text')
            if txt and txt.strip():
                recent_users.append(txt.strip())
                recent_users = recent_users[-USER_TURNS:]  # keep only the last USER_TURNS
        elif msg_type == 'assistant':
            for block in (msg.get('content') or []):
                if isinstance(block, dict) and block.get('type') == 'text':
                    tx = block.get('text', '')
                    if len(tx) >= MAX_ASST_LEN:
                        continue
                    netA = decline.search(tx) and reason.search(tx)
                    netB = strong.search(tx) and not capability.search(tx)
                    if not (netA or netB):
                        continue
                    # multi-turn context, oldest->newest; keep the most-recent tail if over budget
                    user_ctx = '  ⏎  '.join(recent_users)[-USER_TRUNCATE:]
                    if META.search(tx) or META.search(user_ctx):  # skip this build's own meta-chatter
                        continue
                    key = (os.path.basename(f), tx[:200])
                    if key in seen:
                        continue
                    seen.add(key)
                    cands.append({'file': os.path.basename(f), 'ts': d.get('timestamp'),
                                  'user': user_ctx, 'refusal': tx[:ASST_TRUNCATE],
                                  'net': 'A' if netA else 'B'})
                    break

with open('/tmp/refusal_cands.json', 'w') as fh:
    json.dump(cands, fh, indent=1)
print(f"CANDIDATES: {len(cands)}  by-net={dict(Counter(c['net'] for c in cands))} -> /tmp/refusal_cands.json")