`refresh.py`

103 lines
#!/usr/bin/env python3
"""Incremental refresh — keep the viewer current with new Claude rejections.
Runs scan (Stage 1), then classifies ONLY candidates not already in the existing
data/rejections.json (reusing prior heretic verdicts by (file, ts) key so the slow
27B model is paid only for genuinely NEW rejections), then rebuilds data/rejections.json.
Designed to run unattended on a Mac2 schedule (launchd). $0 local."""
import datetime
import glob
import json
import os
import re
import subprocess
import sys
import time
import urllib.request

HERE = os.path.dirname(os.path.abspath(__file__))
CANDS = '/tmp/refusal_cands.json'
DATA = os.path.join(HERE, 'data', 'rejections.json')
PROJROOT = os.path.expanduser('~/.claude/projects')
OLLAMA_URL = 'http://localhost:11434/api/chat'
MODEL = 'qwen3.8-27b-heretic:latest'
SYS = ("You audit an AI coding assistant's replies. Given the USER request and the ASSISTANT reply, decide "
"if the assistant REFUSED or REFUSED-IN-PART on POLICY/LEGAL grounds (copyright, trademark, "
"intellectual-property/impersonation, or website terms-of-service / scraping / anti-bot / unauthorized-access). "
"A mere capability limit is NOT a policy refusal. Agreeing to proceed is NOT a refusal. Respond ONLY with compact JSON: "
'{"refusal":true|false,"category":"copyright|trademark|ip-impersonation|website-tos|other","reason":"<=12 words"}')


def classify(user: str, refusal: str) -> dict:
    body = json.dumps({"model": MODEL, "stream": False, "think": False,
        "messages": [{"role": "system", "content": SYS},
                     {"role": "user", "content": f"USER REQUEST:\n{user}\n\nASSISTANT REPLY:\n{refusal}\n\nJSON:"}],
        "options": {"temperature": 0, "num_predict": 80}}).encode()  # 80 tokens sufficient for compact JSON verdict
    r = urllib.request.urlopen(urllib.request.Request(OLLAMA_URL, body, {"Content-Type": "application/json"}), timeout=180)
    txt = re.sub(r"<think>.*?</think>", "", json.loads(r.read())["message"]["content"], flags=re.DOTALL).strip()
    mm = re.search(r"\{.*\}", txt, re.DOTALL)
    return json.loads(mm.group(0)) if mm else {"refusal": False, "category": "other", "reason": "parse-fail"}


def main() -> None:
    # Stage 1 — scan
    subprocess.run([sys.executable, os.path.join(HERE, 'scan.py')], check=True)
    with open(CANDS) as fh:
        cands = json.load(fh)
    if not isinstance(cands, list):
        sys.exit(f"[refresh] unexpected scan output in {CANDS} (expected list, got {type(cands).__name__})")

    # verdict cache from the existing data file (key = (file, ts))
    cache: dict[tuple, dict] = {}
    if os.path.exists(DATA):
        with open(DATA) as fh:
            for cached_item in json.load(fh).get('items', []):
                cache[(cached_item.get('file'), cached_item.get('ts'))] = cached_item

    # project dir basename -> readable project name
    basename_project: dict[str, str] = {}
    for proj_dir in glob.glob(os.path.join(PROJROOT, '*')):
        proj_base = os.path.basename(proj_dir)
        readable = proj_base.split('-Projects-')[-1] if '-Projects-' in proj_base else proj_base.lstrip('-')
        for f in glob.glob(os.path.join(proj_dir, '**', '*.jsonl'), recursive=True):
            basename_project[os.path.basename(f)] = readable

    items: list[dict] = []
    new_count = 0
    t0 = time.time()
    for c in cands:
        prev = cache.get((c.get('file'), c.get('ts')))
        if prev and not prev.get('error'):  # reuse only GOOD prior verdicts; error verdicts retry next run
            conf, cat, rsn, err = prev.get('confirmed', False), prev.get('category', 'other'), prev.get('reason', ''), False
        else:
            err = False
            try:
                v = classify(c['user'], c['refusal'])
            except Exception as e:  # noqa: BLE001 — intentional resilience; KeyboardInterrupt/SystemExit still propagate
                print(f"[warn] classify failed for {c.get('file')} ts={c.get('ts')}: {e}", file=sys.stderr)
                v = {"refusal": False, "category": "other", "reason": "(unclassified — model unavailable)"}
                err = True  # flag so this item is NOT cached-reused — it re-classifies next run until the model answers
            conf, cat, rsn = bool(v.get('refusal')), v.get('category', 'other'), v.get('reason', '')
            new_count += 1
        items.append({'user': c['user'], 'refusal': c['refusal'], 'confirmed': conf,
                      'category': cat, 'reason': rsn, 'net': c.get('net', 'A'),  # 'A' = Net A (keyword net); 'B' = tone-only net
                      'error': err, 'ts': c['ts'], 'file': c['file'],
                      'project': basename_project.get(c.get('file', ''), '—')})
    items.sort(key=lambda x: (x['ts'] or ''), reverse=True)

    out = {'generated': datetime.datetime.now().astimezone().isoformat(),
           'scanned_files': len(glob.glob(os.path.join(PROJROOT, '**', '*.jsonl'), recursive=True)),
           'total_candidates': len(items),
           'confirmed_count': sum(1 for i in items if i['confirmed']),
           'items': items}
    os.makedirs(os.path.dirname(DATA), exist_ok=True)
    # atomic write: a launchd SIGTERM/timeout mid-dump must never truncate the viewer's only data file
    tmp = DATA + '.tmp'
    with open(tmp, 'w') as fh:
        json.dump(out, fh, indent=1)
    os.replace(tmp, DATA)
    print(f"[{out['generated']}] refresh: {len(items)} items ({out['confirmed_count']} confirmed), {new_count} newly classified, {time.time() - t0:.0f}s")


if __name__ == '__main__':
    main()