#!/usr/bin/env python3 """newOS4 triage engine — turns telemetry + reports into one actionable digest. Read-only by default. --apply performs safe hygiene only (archive old reports). Never edits app code, nginx, services or user content. """ import argparse, json, os, re, subprocess, sys, time from datetime import datetime, timezone ROOT = '/var/www/webx11.duckdns.org/spaces/newos4' DEBUG = os.path.join(ROOT, 'data', 'debug') REPORTS = os.path.join(ROOT, 'data', 'reports') ARCHIVE = os.path.join(REPORTS, 'archive') OUTDIR = '/var/www/webx11.duckdns.org/admin/docs' STATE_DIR = '/var/lib/webx11/newos4-maintenance' STATE = os.path.join(STATE_DIR, 'state.json') NOISE = re.compile(r'(user aborted|aborted without reason|signal timed out|AbortError|TimeoutError)', re.I) SELF = re.compile(r'/api/debug/(ingest|report)\.php', re.I) def load(path, default): try: with open(path, encoding='utf-8') as f: return json.load(f) except Exception: return default def save(path, obj): """Write atomically, under flock, and leave the file readable by nginx.""" import fcntl os.makedirs(os.path.dirname(path), exist_ok=True) lock = open(path + '.lock', 'a+') fcntl.flock(lock, fcntl.LOCK_EX) try: tmp = path + '.tmp' with open(tmp, 'w', encoding='utf-8') as f: json.dump(obj, f, indent=2) f.flush() os.fsync(f.fileno()) os.replace(tmp, path) os.chmod(path, 0o644) # a 600 file 403s behind the admin gate finally: fcntl.flock(lock, fcntl.LOCK_UN) lock.close() def ts(s): if not s: return None try: d = datetime.fromisoformat(str(s).replace('Z', '+00:00')) return d if d.tzinfo else d.replace(tzinfo=timezone.utc) except Exception: return None def age_hours(s): d = ts(s) return None if d is None else round((datetime.now(timezone.utc) - d).total_seconds() / 3600, 1) def _ledger_module(): """Load scripts/newos4-ledger.py by path (the filename has a hyphen, so no plain import).""" try: import importlib.util path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'newos4-ledger.py') spec = importlib.util.spec_from_file_location('newos4_ledger', path) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod except Exception: return None def check_runner(): r = subprocess.run([sys.executable, 'scripts/newos4-maintain.py'], cwd=ROOT, capture_output=True, text=True, timeout=300) lines = [l for l in (r.stdout or '').splitlines() if l.startswith(('FAIL', 'WARN'))] return {'exit': r.returncode, 'findings': lines} def safe(s, limit=200): """Make untrusted telemetry safe to paste into an agent prompt: one line, no control chars, bounded. Everything the agent sees from here is DATA, never an instruction.""" s = re.sub(r'[\x00-\x1f\x7f]+', ' ', str(s or '')) s = re.sub(r'\s+', ' ', s).strip() return s[:limit] def classify(err): msg = str(err.get('message', '')) if SELF.search(msg): return 'self-referential' if NOISE.search(msg): return 'noise' return 'real' def build(apply_fix=False): errors = load(os.path.join(DEBUG, 'errors.json'), []) reports = [] if os.path.isdir(REPORTS): for name in sorted(os.listdir(REPORTS)): if name.startswith('RPT-') and name.endswith('.json'): r = load(os.path.join(REPORTS, name), None) if isinstance(r, dict): r['_file'] = os.path.join(REPORTS, name) reports.append(r) state = load(STATE, {'seen_reports': [], 'last_run': None}) seen = set(state.get('seen_reports', [])) new_reports = [r for r in reports if r.get('id') not in seen] buckets = {'real': [], 'noise': [], 'self-referential': []} for e in errors: buckets[classify(e)].append(e) # A signature's lifetime count is not a priority: the four biggest entries in this file are one # 8-day-old Ollama outage. Rank by recency + normalised clusters, and let the ledger say whether a # shipped fix actually held. ledger = _ledger_module() groups = ledger.rows_by_norm(errors) if ledger else {} fixes = ledger.verify(errors) if ledger else [] fix_summary = ledger.summarise(fixes) if ledger else {'regressed': [], 'watching': [], 'closed': []} active = [e for e in errors if (age_hours(e.get('last_seen')) or 1e9) < 24] stale = [e for e in errors if (age_hours(e.get('last_seen')) or 0) > 168] clusters = sorted( ({'norm': g['norm'], 'count': g['count'], 'signatures': g['signatures'], 'last_hours': age_hours(g['last_seen']), 'frame': g['frame'], 'message': safe((g['messages'] or [''])[0], 120)} for g in groups.values()), key=lambda x: (x['last_hours'] if x['last_hours'] is not None else 1e9, -x['count']) )[:8] digest = { 'generated_at': datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ'), 'counts': { 'records': len(errors), 'occurrences': sum(int(e.get('count') or 1) for e in errors), 'real_records': len(buckets['real']), 'noise_records': len(buckets['noise']), 'self_referential_records': len(buckets['self-referential']), 'reports_total': len(reports), 'reports_new': len(new_reports), 'active_records': len(active), 'stale_records': len(stale), 'clusters': len(groups), 'fixes_regressed': len(fix_summary['regressed']), 'fixes_watching': len(fix_summary['watching']), 'fixes_closed': len(fix_summary['closed']), }, 'clusters': clusters, 'fixes': fix_summary, 'top_real': sorted( ({'message': safe(e.get('message'), 200), 'type': e.get('type'), 'count': int(e.get('count') or 1), 'age_hours': age_hours(e.get('last_seen')), 'frame': safe(e.get('frame'), 160), 'session': safe(e.get('session_id'), 40)} for e in buckets['real']), key=lambda x: -x['count'])[:10], 'new_reports': [{'id': r.get('id'), 'title': safe(r.get('title'), 160), 'created_at': r.get('created_at'), 'age_hours': age_hours(r.get('created_at')), 'body': safe(r.get('body'), 400)} for r in new_reports[:10]], 'self_referential': [str(e.get('message'))[:120] for e in buckets['self-referential']], 'checks': check_runner(), } if apply_fix: os.makedirs(ARCHIVE, exist_ok=True) moved = 0 cutoff = time.time() - 90 * 86400 for r in reports: t = ts(r.get('created_at')) if t and t.timestamp() < cutoff: os.replace(r['_file'], os.path.join(ARCHIVE, os.path.basename(r['_file']))) moved += 1 digest['counts']['reports_archived'] = moved save(STATE, {'seen_reports': sorted(seen | {r.get('id') for r in new_reports if r.get('id')}), 'last_run': digest['generated_at']}) save(os.path.join(OUTDIR, 'newos4-triage-latest.json'), digest) c = digest['counts'] print(f"newOS4 triage {digest['generated_at']}: records={c['records']} " f"occurrences={c['occurrences']} real={c['real_records']} noise={c['noise_records']} " f"self={c['self_referential_records']} reports={c['reports_total']} new={c['reports_new']}") print(f" ACTIVE| {c['active_records']} records seen in last 24h | stale>7d {c['stale_records']} " f"| clusters {c['clusters']} of {c['records']} signatures") for cl in digest['clusters'][:5]: fr = f" @ {cl['frame']}" if cl.get('frame') else '' print(f" CLUSTER| {cl['count']}x ({cl['signatures']} sig) {safe(cl['message'], 95)}{fr}") for fx in digest['fixes']['regressed']: print(f" FIX| REGRESSED {fx['commit']} {fx['count']}x {safe(fx['message'], 80)} ({fx['detail']})") for fx in digest['fixes']['watching']: print(f" FIX| watching {fx['commit']} {safe(fx['message'], 80)} ({fx['detail']})") for f in digest['checks']['findings']: print('check:', f) for e in digest['top_real'][:5]: print(f" DATA| top: {e['count']}x {e['type']} ({e['age_hours']}h) {safe(e['message'], 110)}") if e.get('frame'): print(f" FRAME| {e['frame']}") for r in digest['new_reports'][:5]: print(f" DATA| report: {r['id']} {safe(r['title'], 110)}") return digest if __name__ == '__main__': ap = argparse.ArgumentParser() ap.add_argument('--apply', action='store_true') ap.add_argument('--json', action='store_true') a = ap.parse_args() d = build(apply_fix=a.apply) if a.json: print(json.dumps(d, indent=2))