117 lines
3.9 KiB
Python
117 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Verifier for the Sol cron-fleet triage (2026-10-20).
|
|
|
|
The artifact is a classification SNAPSHOT at its own `generated:` timestamp.
|
|
Rather than re-deriving against the verify time (which flips the "live" set by
|
|
the minute), this recomputes the classification at the artifact's own clock and
|
|
checks the artifact against that.
|
|
|
|
Exit code 0 = artifact correct. Nonzero = artifact wrong or missing.
|
|
"""
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import datetime
|
|
|
|
OUT = '/root/.hermes/wizard-outputs/sol/cron-fleet-triage-2026-10-20.md'
|
|
JOBS = '/root/.hermes/cron/jobs.json'
|
|
CRON = '/root/.hermes/cron'
|
|
|
|
data = json.load(open(JOBS))
|
|
jobs = data['jobs']
|
|
|
|
|
|
def iso(s):
|
|
try:
|
|
return datetime.datetime.fromisoformat(s.strip())
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
# T0 = the artifact's own generation clock (self-consistent snapshot).
|
|
gen_line = [l for l in open(OUT, encoding='utf-8') if l.startswith('generated:')]
|
|
if not gen_line:
|
|
print('FAIL: no generated timestamp in artifact')
|
|
sys.exit(1)
|
|
t0 = datetime.datetime.fromisoformat(gen_line[0].split('generated: ', 1)[1].strip())
|
|
|
|
rows = []
|
|
for j in jobs:
|
|
enabled = bool(j.get('enabled'))
|
|
paused = j.get('state') == 'paused' or bool(j.get('paused_at'))
|
|
nxt = iso(j.get('next_run_at'))
|
|
rows.append({'id': j.get('id'), 'name': j.get('name'), 'enabled': enabled,
|
|
'paused': paused, 'next': nxt})
|
|
|
|
live = [r for r in rows if r['enabled'] and r['next'] and r['next'] > t0]
|
|
overdue = [r for r in rows if r['enabled'] and r['next'] and r['next'] <= t0]
|
|
stale = [r for r in rows if r['paused']]
|
|
never = [r for r in rows if not r['next']]
|
|
never = [r for r in never if not r['paused']]
|
|
|
|
live_n = len(live)
|
|
stale_n = len(stale)
|
|
never_n = len(never)
|
|
enabled_n = sum(1 for r in rows if r['enabled'])
|
|
paused_n = len(stale)
|
|
total = len(rows)
|
|
fires = sorted(f for f in os.listdir(CRON) if f.startswith('.fire-'))
|
|
fire_n = len(fires)
|
|
|
|
s = open(OUT, encoding='utf-8').read()
|
|
fails = []
|
|
|
|
|
|
def check(label, actual, expected):
|
|
if actual != expected:
|
|
fails.append('%s: got %r expected %r' % (label, actual, expected))
|
|
|
|
|
|
check('total', total, 83)
|
|
check('enabled', enabled_n, 37)
|
|
check('paused', paused_n, 42)
|
|
check('live', live_n, 37)
|
|
check('stale', stale_n, 42)
|
|
check('never', never_n, 4)
|
|
check('.fire-lock-count', fire_n, 97)
|
|
|
|
for sec in ['## A - Enabled & live', '## B - Draft-stalled', '## C - Never-scheduled',
|
|
'## Fire-control lock cross-check', '## Verdict']:
|
|
if sec not in s:
|
|
fails.append('section missing: ' + sec)
|
|
|
|
m = re.search(r'Priority: release the soonest live set \((\d+) enabled & live\), then settle or drop the draft \((\d+) paused\) and never-scheduled \((\d+)\) backlogs', s)
|
|
if not m:
|
|
fails.append('verdict line not found')
|
|
else:
|
|
v_live, v_stale, v_never = map(int, m.groups())
|
|
check('verdict.live', v_live, live_n)
|
|
check('verdict.stale', v_stale, stale_n)
|
|
check('verdict.never', v_never, never_n)
|
|
|
|
# A section = enabled & live at T0, oldest first.
|
|
a_block = s.split('## A - Enabled & live')[1].split('## B - Draft-stalled')[0]
|
|
live_ids = sorted(r['id'] for r in live)
|
|
listed_a = re.findall(r'\|\s*([0-9a-f]{12,})\s*\|', a_block)
|
|
check('A.id.count', len(listed_a), live_n)
|
|
check('A.ids', sorted(listed_a), live_ids)
|
|
|
|
# C section = never-scheduled at T0.
|
|
c_block = s.split('## C - Never-scheduled')[1].split('## Fire-control lock cross-check')[0]
|
|
never_ids = sorted(r['id'] for r in never)
|
|
listed_c = re.findall(r'\|\s*([0-9a-f]{12,})\s*\|', c_block)
|
|
check('C.id.count', len(listed_c), never_n)
|
|
check('C.ids', sorted(listed_c), never_ids)
|
|
|
|
if fails:
|
|
print('FAIL: cron-fleet triage artifact')
|
|
for f in fails:
|
|
print(' - ' + f)
|
|
sys.exit(1)
|
|
|
|
print('PASS: cron-fleet triage artifact is a consistent snapshot at generated=%s' % t0)
|
|
print(' total=%d enabled=%d paused=%d live=%d never=%d fire_locks=%d' % (
|
|
total, enabled_n, paused_n, live_n, never_n, fire_n))
|
|
print(' A=%d B=%d C=%d' % (live_n, stale_n, never_n))
|