path: root/scripts/newos4-feedback-autoreview.py

#!/usr/bin/env python3
"""Autonomous first-pass reviewer for the WebX11 feedback queue.

Decides ONLY what it can prove, appends the decision to data/feedback/decisions.jsonl tagged
actor=auto-reviewer, and prints one line per action (silent when it did nothing — the cron
contract). The operator's later click always wins: the applier keeps the last row per item.

Modes in data/feedback/autonomy.json: off | safe (default) | aggressive.
"""
import argparse, glob, json, os, re, subprocess, sys
from datetime import datetime, timezone

ROOT = '/var/www/webx11.duckdns.org'
ROOTDIR = ROOT                       # overridable for fixture runs
REPORTS = f'{ROOT}/spaces/newos4/data/reports'
MODULES = f'{ROOT}/spaces/newos4/modules'
DECISIONS = f'{ROOT}/spaces/newos4/data/feedback/decisions.jsonl'
AUTONOMY = f'{ROOT}/spaces/newos4/data/feedback/autonomy.json'
SUMMARY = f'{ROOT}/spaces/newos4/data/feedback/autoreview-latest.json'
CAPS = {'safe': 2, 'aggressive': 5}

NOISE = re.compile(r'fetch failed|failed to fetch|network ?error|load failed|user aborted'
                   r'|signal is aborted|PROBE|probe-only|boom from|\bone-off test\b|\btest only\b', re.I)
IDENT = re.compile(r'[A-Za-z_$][\w$]{4,}')


def load_json(path, default):
    try:
        with open(path, encoding='utf-8') as f:
            return json.load(f)
    except Exception:
        return default


def mode(cap_override=None):
    m = str((load_json(AUTONOMY, {}) or {}).get('mode') or 'safe').lower()
    if m not in CAPS:
        return 'off', 0
    return m, cap_override if cap_override is not None else CAPS[m]


def decisions():
    out = {}
    if not os.path.exists(DECISIONS):
        return out
    with open(DECISIONS, encoding='utf-8') as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                d = json.loads(line)
            except Exception:
                continue
            out[d.get('item_id', '')] = d
    return out


def approved_today():
    today = datetime.now(timezone.utc).strftime('%Y-%m-%d')
    n = 0
    if not os.path.exists(DECISIONS):
        return 0
    with open(DECISIONS, encoding='utf-8') as f:
        for line in f:
            try:
                d = json.loads(line)
            except Exception:
                continue
            if (str(d.get('ts', '')).startswith(today) and d.get('actor') == 'auto-reviewer'
                    and d.get('decision') == 'approve'):
                n += 1
    return n


def identifiers(msg):
    """Distinctive names in an error message: drop numbers/quoted parts, keep real tokens."""
    cleaned = re.sub(r'["\'].*?["\']', ' ', str(msg))
    cleaned = re.sub(r'0x[0-9a-f]+', ' ', cleaned, flags=re.I)
    skip = {'error', 'failed', 'fetch', 'undefined', 'cannot', 'reading', 'properties', 'uncaught',
            'invalid', 'unexpected', 'token', 'https', 'http', 'com', 'org', 'null', 'true', 'false'}
    return [t for t in IDENT.findall(cleaned) if t.lower() not in skip][:4]


def code_hits(tokens):
    """Which tokens still appear in shipped modules (grep -rlF, no shell)."""
    hits = {}
    for t in tokens:
        try:
            r = subprocess.run(['grep', '-rlF', '--include=*.js', t, MODULES],
                               capture_output=True, text=True, timeout=20)
            files = [ln for ln in r.stdout.splitlines() if '.orig.js' not in ln]
        except Exception:
            files = []
        hits[t] = len(files)
    return hits


def token_history(tokens):
    """Commits that added or removed each identifier under modules/ (git log -S).

    This is the proof an "already fixed" verdict needs: the token is gone from HEAD *and* git
    shows it once existed, so a commit removed it rather than it never being there. Without this,
    a report about a string that never lived in the code reads as fixed (fixture RPT-G caught it).
    """
    out = {}
    for t in tokens:
        try:
            r = subprocess.run(['git', '-C', ROOTDIR, 'log', '--oneline', '-S', t, '--',
                                os.path.relpath(MODULES, ROOTDIR)],
                               capture_output=True, text=True, timeout=30)
            out[t] = [ln for ln in r.stdout.splitlines() if ln.strip()]
        except Exception:
            out[t] = []
    return out


def fingerprints():
    """fingerprint -> id, for reports that already exist (duplicate detection)."""
    out = {}
    for p in glob.glob(f'{REPORTS}/RPT-*.json'):
        d = load_json(p, {})
        fp = d.get('fingerprint')
        if fp:
            out.setdefault(fp, []).append(d.get('id') or os.path.basename(p)[:-5])
    return out


def classify(rep, dec, fps, cap_left):
    """Return (decision, reason, evidence) or (None, why_skipped, evidence)."""
    rid = rep.get('id') or ''
    title = str(rep.get('title') or '')
    diag = rep.get('diagnostics') or {}
    errs = [e for e in (diag.get('errors') or []) if isinstance(e, dict)]
    msg = ' '.join(str(e.get('message') or '') for e in errs) or title
    stack = any(e.get('stack') for e in errs)
    auto = bool(diag.get('auto'))

    # R6 first: a human wrote it, a human answers it.
    if not auto:
        return None, 'user report — left for the operator', {}

    # R1 duplicate: same fingerprint already decided.
    fp = rep.get('fingerprint')
    twins = [x for x in fps.get(fp, []) if x != rid]
    if fp and twins and any(('report:' + t) in dec for t in twins):
        done = [t for t in twins if ('report:' + t) in dec]
        return 'resolve', f'duplicate of already-decided {done[0]}', {'fingerprint': fp, 'twin': done[0]}

    # R3 noise: test artefact / client-side network drop.
    noise_hit = NOISE.search(title) or NOISE.search(msg)
    if noise_hit:
        return 'reject', 'noise: test artefact or client-side network failure', {'match': noise_hit.group(0)}

    tokens = identifiers(msg)
    hits = code_hits(tokens) if tokens else {}
    present = [t for t in tokens if hits.get(t)]
    absent = [t for t in tokens if not hits.get(t)]

    # R2 already-fixed: every distinctive token gone AND git proves it was removed (not never there).
    if tokens and not present:
        hist = token_history(tokens)
        proven = {t: c for t, c in hist.items() if c}
        if proven:
            t0 = next(iter(proven))
            return 'resolve', f'already fixed: "{t0}" was removed from modules/ in {proven[t0][0].split()[0]}', {
                'absent_tokens': absent, 'removed_in': proven[t0][0], 'history_tokens': list(proven)}
        return None, 'identifiers gone but git shows they never existed — needs a human', {
            'absent_tokens': absent}

    # R4 still-present: real stack error whose identifiers still exist in shipped code.
    if stack and present and cap_left > 0:
        return 'approve', 'still present in shipped code (stack + identifiers found)', {
            'present_tokens': present, 'hits': {t: hits[t] for t in present}}
    if stack and present and cap_left <= 0:
        return None, 'daily auto-approve cap reached — left for the operator', {'present_tokens': present}

    return None, 'not provable by the rules — left for the operator', {
        'tokens': tokens, 'present': present, 'stack': stack}


def main():
    global REPORTS, MODULES, DECISIONS, ROOTDIR
    ap = argparse.ArgumentParser()
    ap.add_argument('--root', default=ROOT)
    ap.add_argument('--dry-run', action='store_true')
    ap.add_argument('--reports-dir', default=REPORTS)
    ap.add_argument('--decisions', default=DECISIONS)
    ap.add_argument('--modules', default=MODULES)
    ap.add_argument('--cap', type=int, default=None)
    ap.add_argument('--include-plans', action='store_true', help='never used in production')
    a = ap.parse_args()
    ROOTDIR = a.root
    REPORTS, MODULES, DECISIONS = a.reports_dir, a.modules, a.decisions

    m, cap = mode(a.cap)
    if m == 'off':
        return 0                      # silent: autonomy disabled
    cap_left = max(0, cap - approved_today())
    dec = decisions()
    fps = fingerprints()
    rows, skipped = [], []
    for p in sorted(glob.glob(f'{REPORTS}/RPT-*.json')):
        rep = load_json(p, {})
        rid = 'report:' + (rep.get('id') or os.path.basename(p)[:-5])
        if not rep or rid in dec:
            continue                  # already decided → operator or a previous run
        decision, why, ev = classify(rep, dec, fps, cap_left)
        if decision:
            rows.append({'ts': datetime.now(timezone.utc).isoformat(timespec='seconds'),
                         'item_id': rid, 'decision': decision, 'note': why,
                         'actor': 'auto-reviewer', 'evidence': ev})
            if decision == 'approve':
                cap_left -= 1
        else:
            skipped.append({'item_id': rid, 'why': why, 'evidence': ev})

    summary = {'generated_at': datetime.now(timezone.utc).isoformat(timespec='seconds'),
               'mode': m, 'cap_left': cap_left, 'decided': len(rows),
               'skipped': len(skipped), 'rows': rows, 'skips': skipped[:20]}
    if not a.dry_run:
        if rows:
            with open(DECISIONS, 'a', encoding='utf-8') as f:
                for r in rows:
                    f.write(json.dumps(r, ensure_ascii=False) + '\n')
        tmp = SUMMARY + '.tmp'
        os.makedirs(os.path.dirname(SUMMARY), exist_ok=True)
        with open(tmp, 'w', encoding='utf-8') as f:
            json.dump(summary, f, ensure_ascii=False, indent=1)
        os.replace(tmp, SUMMARY)
    for r in rows:
        print(f"{r['decision']:<8} {r['item_id'][:40]:<42} {r['note']}")
    if not rows:
        print(f'autoreview: nothing provable ({len(skipped)} left for the operator, mode={m})'
              if a.dry_run else '', end='')
    return 0


if __name__ == '__main__':
    sys.exit(main())