1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
|
#!/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')
APPLIED = os.path.join(ROOT, 'spaces/newos4/data/feedback/applied.json') # ledger written by the applier
NETFAIL = os.path.join(ROOT, 'spaces/newos4/data/debug/network_failures.json') # ingest-filter counter
AUTOREVIEW = os.path.join(ROOT, 'spaces/newos4/data/feedback/autoreview-latest.json')
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)
try:
body = open(p, encoding='utf-8', errors='ignore').read()
except Exception:
body = ''
lines = body.splitlines()
# A decision turns on the plan's own text: the opening 60 lines carry Goal/Approach and the
# task headings show the shape of the work. The full file is 20-40 KB per plan — too much.
excerpt = '\n'.join(lines[:60])[:2400]
tasks = [l.lstrip('#').strip() for l in lines if re.match(r'^#{2,3}\s+Task\b', l)][:20]
status = [l.strip() for l in lines if l.startswith('## Status') or l.startswith('## Implementation')][:2]
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'),
'excerpt': excerpt, 'tasks': tasks, 'status': status,
'decision': d.get('decision'), 'decided_at': d.get('decided_at') or d.get('ts'),
'note': d.get('note', ''), 'actor': d.get('actor', ''),
})
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]
diag = r.get('diagnostics') or {}
errs = diag.get('errors') or []
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(diag.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', ''),
'actor': d.get('actor', ''),
'message': (r.get('body') or '')[:1500], 'page': diag.get('url') or '',
'version': diag.get('version') or '', 'ua': (diag.get('ua') or '')[:120],
'errors': [str(e)[:300] for e in errs[:3]],
})
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')]
net = load(NETFAIL, {}) or {}
net_top = sorted(((net.get('by_source') or {}) if isinstance(net.get('by_source'), dict) else {}).items(),
key=lambda kv: -int(kv[1]))[:3]
doc = {
'network_failures': {'day': net.get('day'), 'total': int(net.get('total') or 0),
'top': [[str(k), int(v)] for k, v in net_top]},
'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),
'autoreview': {k: (load(AUTOREVIEW, {}) or {}).get(k) for k in ('generated_at', 'mode', 'decided', 'skipped')},
'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())
|
