2 files changed, 350 insertions, 0 deletions
diff --git a/scripts/newos4-feedback-apply.py b/scripts/newos4-feedback-apply.py
new file mode 100755
index 0000000..38188fb
--- /dev/null
+++ b/scripts/newos4-feedback-apply.py
@@ -0,0 +1,203 @@
+#!/usr/bin/env python3
+"""Apply queued operator decisions from data/feedback/decisions.jsonl.
+
+Runs as darkbot (cron). Mutates exactly two things, both assert-guarded:
+  approve + plan    -> flip the exact '<!-- autonomy: review -->' line to '<!-- autonomy: safe -->'
+  reject  + plan    -> flip it back (an approval that was reconsidered before the night run)
+  approve + report  -> append a high-priority directive via scripts/newos4-steer.py add
+Every decision is recorded in data/feedback/applied.json (darkbot-owned, 0600) so nothing is
+applied twice and php never needs write access to the sources.
+
+Ordering rules (2026-10-26 review):
+  * the LAST decision per item wins — approve followed by reject must not ship the plan;
+  * a decision that is superseded in the same run is recorded, not executed;
+  * one failing item never aborts the run, and applied.json is written after every mutation.
+"""
+import json, os, re, subprocess, sys
+from datetime import datetime, timezone
+
+ROOT = '/var/www/webx11.duckdns.org'
+FB = os.path.join(ROOT, 'spaces/newos4/data/feedback')
+DEC = os.path.join(FB, 'decisions.jsonl')
+APPLIED = os.path.join(FB, 'applied.json')
+STEER = os.path.join(ROOT, 'spaces/newos4/scripts/newos4-steer.py')
+PLANS = os.path.join(ROOT, '.hermes/plans')
+REPORTS = os.path.join(ROOT, 'spaces/newos4/data/reports')
+ROTATE_BYTES = 1024 * 1024          # move decisions.jsonl aside once it passes 1 MB
+
+
+def load_applied():
+    try:
+        with open(APPLIED, encoding='utf-8') as f:
+            d = json.load(f)
+        return d if isinstance(d, dict) else {}
+    except Exception:
+        return {}
+
+
+def save_applied(d):
+    os.makedirs(FB, exist_ok=True)
+    tmp = APPLIED + '.tmp'
+    with open(tmp, 'w', encoding='utf-8') as f:
+        json.dump(d, f, indent=1, ensure_ascii=False)
+    os.chmod(tmp, 0o600)            # www-data must not be able to wipe the ledger
+    os.replace(tmp, APPLIED)
+
+
+def read_decisions():
+    rows = []
+    if not os.path.exists(DEC):
+        return rows
+    with open(DEC, encoding='utf-8', errors='ignore') as f:
+        for i, line in enumerate(f):
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                d = json.loads(line)
+            except Exception:
+                continue                 # torn write: skip, never crash the run
+            d['_row'] = i
+            rows.append(d)
+    return rows
+
+
+def _marker_count(src):
+    return sum(1 for l in src.splitlines() if l.strip() == '<!-- autonomy: review -->')
+
+
+def _flip_marker(path, src, old, new, note):
+    """Flip one standalone marker LINE (not a substring) and keep the file mode."""
+    lines = src.splitlines(keepends=True)
+    hits = [i for i, l in enumerate(lines) if l.strip() == old]
+    if len(hits) != 1:
+        return None, 'error: expected 1 standalone %s line, found %d' % (old, len(hits))
+    i = hits[0]
+    indent = lines[i][:len(lines[i]) - len(lines[i].lstrip())]
+    eol = '\n' if lines[i].endswith('\n') else ''
+    lines[i] = '%s%s%s' % (indent, new, eol)
+    stamp = datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%MZ')
+    body = ''.join(lines)
+    if not body.endswith('\n'):
+        body += '\n'
+    body += '\n<!-- %s by operator via /admin/feedback.html %s%s -->\n' % (
+        'approved' if new.endswith('safe -->') else 'reverted',
+        stamp, (' — ' + note.replace('-->', '')) if note else '')
+    return body, None
+
+
+def apply_plan(name, decision, note):
+    plans_dir = os.path.realpath(PLANS)
+    path = os.path.realpath(os.path.join(plans_dir, name))
+    if os.path.dirname(path) != plans_dir:
+        return 'error: refused, outside plans dir'
+    if not re.fullmatch(r'[A-Za-z0-9._-]+\.md', os.path.basename(path)):
+        return 'error: refused, bad plan name'
+    try:
+        src = open(path, encoding='utf-8').read()
+        mode = os.stat(path).st_mode & 0o777
+    except Exception as e:
+        return 'error: unreadable (%s)' % e
+
+    if decision == 'approve':
+        if _marker_count(src) != 1:
+            return 'error: expected 1 review marker, found %d' % _marker_count(src)
+        body, err = _flip_marker(path, src, '<!-- autonomy: review -->', '<!-- autonomy: safe -->', note)
+    elif decision == 'reject':
+        # only meaningful if a previous run already approved it
+        if src.count('<!-- autonomy: safe -->') == 0:
+            return 'recorded (no action: was never approved)'
+        body, err = _flip_marker(path, src, '<!-- autonomy: safe -->', '<!-- autonomy: review -->', note)
+    else:
+        return 'recorded (no action)'
+    if err or body is None:
+        return err or 'error: no output produced'
+    tmp = path + '.tmp'
+    with open(tmp, 'w', encoding='utf-8') as f:
+        f.write(body)
+    os.chmod(tmp, mode)          # os.replace would otherwise apply the umask (600 -> 644)
+    os.replace(tmp, path)
+    return 'plan %s (%s, mode %o)' % ('approved' if decision == 'approve' else 'reverted', decision, mode)
+
+
+def report_title(rid):
+    """The bare id is useless in a directive once the report file rotates away."""
+    try:
+        with open(os.path.join(REPORTS, rid + '.json'), encoding='utf-8') as f:
+            d = json.load(f)
+        return str(d.get('title') or d.get('message') or rid)[:160]
+    except Exception:
+        return rid
+
+
+def apply_report(name, decision, note):
+    if decision != 'approve':
+        return 'recorded (no action)'
+    text = '%s — %s' % (report_title(name), name)
+    if note:
+        text = '%s — %s' % (text, note)
+    r = subprocess.run([sys.executable, STEER, 'add', text, '--priority', 'high'],
+                       capture_output=True, text=True, timeout=60)
+    if r.returncode != 0:
+        return 'error: steer add exit %d (%s)' % (r.returncode, (r.stderr or '')[-120:])
+    return 'steering directive added'
+
+
+def main():
+    applied = load_applied()
+    rows = read_decisions()
+    if not rows:
+        return 0
+
+    # last decision per item wins; everything earlier in the file is superseded
+    last_row = {}
+    for d in rows:
+        last_row[d.get('item_id', '')] = d['_row']
+
+    done = 0
+    for d in rows:
+        item = d.get('item_id', '')
+        dec = d.get('decision', '')
+        note = d.get('note', '') or ''
+        key = '%s|%s|%s' % (d.get('ts', ''), item, dec)
+        if key in applied:
+            continue
+        kind, _, name = item.partition(':')
+
+        if last_row.get(item) != d['_row']:
+            applied[key] = {'item_id': item, 'decision': dec, 'result': 'superseded by a later decision',
+                            'applied_at': datetime.now(timezone.utc).isoformat(timespec='seconds')}
+            save_applied(applied)
+            continue
+
+        try:
+            if kind == 'plan':
+                result = apply_plan(name, dec, note)
+            elif kind == 'report':
+                result = apply_report(name, dec, note)
+            else:
+                result = 'error: unknown item kind'
+        except subprocess.TimeoutExpired:
+            result = 'error: steer add timed out'
+        except Exception as e:                       # one bad item must not replay the batch
+            result = 'error: %s: %s' % (type(e).__name__, e)
+
+        applied[key] = {'item_id': item, 'decision': dec, 'result': result,
+                        'applied_at': datetime.now(timezone.utc).isoformat(timespec='seconds')}
+        save_applied(applied)
+        done += 1
+        print(('APPLY FAILED %s: %s' if result.startswith('error') else 'applied %s -> %s (%s)')
+              % ((item, result) if result.startswith('error') else (item, dec, result)))
+
+    # keep the queue file bounded; applied.json stays as the permanent ledger
+    try:
+        if os.path.getsize(DEC) > ROTATE_BYTES:
+            os.replace(DEC, DEC + '.1')
+            print('rotated decisions.jsonl -> decisions.jsonl.1 (%d bytes)' % ROTATE_BYTES)
+    except Exception:
+        pass
+    return 0
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())
diff --git a/scripts/newos4-feedback-collect.py b/scripts/newos4-feedback-collect.py
new file mode 100755
index 0000000..66f1653
--- /dev/null
+++ b/scripts/newos4-feedback-collect.py
@@ -0,0 +1,147 @@
+#!/usr/bin/env python3
+"""Snapshot agent-side review items for /admin/feedback.html.
+
+Runs as darkbot (Hermes cron): .hermes/plans/*.md are 600 darkbot, unreadable to php-fpm,
+so the snapshot is the only way the page can see them. Writes atomically, 644 www-data.
+Read-only with respect to every source file.
+"""
+import glob, json, os, re, time
+from datetime import datetime, timezone
+
+ROOT = '/var/www/webx11.duckdns.org'
+OUT_DIR = os.path.join(ROOT, 'admin/docs/feedback')
+OUT = os.path.join(OUT_DIR, 'agent-items.json')
+PLANS = os.path.join(ROOT, '.hermes/plans')
+TRIAGE = os.path.join(ROOT, 'admin/docs/newos4-triage-latest.json')
+STEER = os.path.join(ROOT, 'spaces/newos4/data/maintenance/steering.json')
+DECISIONS = os.path.join(ROOT, 'spaces/newos4/data/feedback/decisions.jsonl')
+
+
+def load(path, default):
+    try:
+        with open(path, encoding='utf-8') as f:
+            return json.load(f)
+    except Exception:
+        return default
+
+
+def decisions():
+    """Latest decision per item id, from the append-only log written by php."""
+    out = {}
+    if not os.path.exists(DECISIONS):
+        return out
+    try:
+        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          # torn line: ignore, never a crash
+                out[d.get('item_id', '')] = d
+    except Exception:
+        pass
+    return out
+
+
+def plans_needing_decision(dec):
+    items = []
+    for p in sorted(glob.glob(os.path.join(PLANS, '*.md'))):
+        try:
+            with open(p, encoding='utf-8', errors='ignore') as fh:
+                head = ''.join(fh.readline() for _ in range(40))
+        except Exception:
+            continue
+        # Exact-line match: the analyser contract puts the marker on the line after the
+        # title. A substring test also matches plans that merely *document* the marker
+        # (measured 2026-10-26: this plan's own audit table inflated the count to 4).
+        if not any(l.strip() == '<!-- autonomy: review -->' for l in head.splitlines()):
+            continue
+        name = os.path.basename(p)
+        title = next((l.lstrip('# ').strip() for l in head.splitlines() if l.startswith('# ')), name)
+        d = dec.get('plan:' + name, {})
+        items.append({
+            'id': 'plan:' + name, 'kind': 'plan', 'title': title, 'path': p,
+            'size': os.path.getsize(p), 'mtime': datetime.fromtimestamp(
+                os.path.getmtime(p), timezone.utc).isoformat(timespec='seconds'),
+            'decision': d.get('decision'), 'decided_at': d.get('decided_at') or d.get('ts'),
+            'note': d.get('note', ''),
+        })
+    return items
+
+
+def reports_needing_decision(dec):
+    triage = load(TRIAGE, {}) or {}
+    new_ids = {r.get('id') for r in (triage.get('new_reports') or [])}
+    items = []
+    for p in sorted(glob.glob(os.path.join(ROOT, 'spaces/newos4/data/reports/*.json'))):
+        try:
+            r = json.load(open(p, encoding='utf-8'))
+        except Exception:
+            continue
+        rid = r.get('id') or os.path.basename(p)[:-5]
+        d = dec.get('report:' + rid, {})
+        items.append({
+            'id': 'report:' + rid, 'kind': 'report', 'title': (r.get('title') or rid)[:180],
+            'created_at': r.get('created_at'), 'auto': bool((r.get('diagnostics') or {}).get('auto')),
+            'new': rid in new_ids, 'decision': d.get('decision'),
+            'decided_at': d.get('decided_at') or d.get('ts'), 'note': d.get('note', ''),
+        })
+    items.sort(key=lambda x: (x.get('created_at') or ''), reverse=True)
+    return items
+
+
+def recent_apply(n):
+    """Last n decision outcomes, newest first — failures included, so the dashboard can show them."""
+    try:
+        with open(APPLIED, encoding='utf-8') as f:
+            d = json.load(f)
+        rows = [v for v in d.values() if isinstance(v, dict)]
+        rows.sort(key=lambda r: str(r.get('applied_at') or ''), reverse=True)
+        return rows[:n]
+    except Exception:
+        return []
+
+
+def main():
+    dec = decisions()
+    steer_raw = load(STEER, [])
+    steer_ok = isinstance(steer_raw, list)
+    steer = [s for s in (steer_raw if steer_ok else []) if isinstance(s, dict) and not s.get('done')]
+    triage = load(TRIAGE, {}) or {}
+    plans = plans_needing_decision(dec)
+    reports = reports_needing_decision(dec)
+    awaiting = [x for x in plans + reports if not x.get('decision')]
+    doc = {
+        'generated_at': datetime.now(timezone.utc).isoformat(timespec='seconds'),
+        'counts': {
+            'plans_review': len(plans), 'reports_total': len(reports),
+            'awaiting_decision': len(awaiting),
+            'steering_pending': len(steer) if steer_ok else -1,   # -1 = steering.json unreadable/wrong shape
+            'triage_reports_new': (triage.get('counts') or {}).get('reports_new', 0),
+        },
+        'plans': plans, 'reports': reports, 'steering': steer[:20],
+        'recent_apply': recent_apply(8),
+        'triage_checks': (triage.get('checks') or {}),
+        'triage_generated_at': triage.get('generated_at'),
+    }
+    os.makedirs(OUT_DIR, exist_ok=True)
+    tmp = OUT + '.tmp'
+    with open(tmp, 'w', encoding='utf-8') as f:
+        json.dump(doc, f, ensure_ascii=False, indent=1)
+    os.replace(tmp, OUT)
+    os.chmod(OUT, 0o644)
+    try:
+        import grp
+        os.chown(OUT, -1, grp.getgrnam('www-data').gr_gid)
+    except Exception:
+        pass
+    print(f"feedback snapshot: plans={len(plans)} reports={len(reports)} "
+          f"awaiting={len(awaiting)} steering={len(steer)} -> {OUT}")
+    return 0
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())