133 lines
5.1 KiB
Python
133 lines
5.1 KiB
Python
#!/usr/bin/env python3
|
|
"""CHECK: re-derive the Sol cron-fleet triage from the FROZEN job snapshot and
|
|
compare against the classification markdown.
|
|
|
|
Source of truth: `cron-fleet-triage-2026-10-20-refresh.json` — a frozen copy of
|
|
the cron job state captured at generation time (`cron_gen.py`). The live cron
|
|
config is rewritten by the fleet scheduler, so classification drift would make a
|
|
live re-read meaningless. This verifier recomputes from the frozen snapshot at
|
|
the artifact's own clock; the expected values are DERIVED from the snapshot, not
|
|
hard-coded.
|
|
|
|
Exit code 0 = artifact consistent with its frozen source. Nonzero = wrong/missing.
|
|
|
|
Single bare command:
|
|
python3 wizard-work/sol/cron-fleet-triage-refresh-verify.py
|
|
"""
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import datetime
|
|
|
|
OUT_MD = '/root/.hermes/wizard-outputs/sol/cron-fleet-triage-2026-10-20-refresh.md'
|
|
OUT_JSON = '/root/.hermes/wizard-outputs/sol/cron-fleet-triage-2026-10-20-refresh.json'
|
|
|
|
|
|
def parse_iso(s):
|
|
try:
|
|
return datetime.datetime.fromisoformat(s.strip())
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def main():
|
|
for p in (OUT_MD, OUT_JSON):
|
|
if not os.path.exists(p):
|
|
print('FAIL: missing %s' % p)
|
|
sys.exit(1)
|
|
|
|
snap = json.load(open(OUT_JSON))
|
|
s = open(OUT_MD, encoding='utf-8').read()
|
|
t0 = parse_iso(
|
|
[l for l in s.splitlines() if l.startswith('generated:')][0]
|
|
.split('generated: ', 1)[1].strip()
|
|
)
|
|
if t0 is None:
|
|
print('FAIL: could not parse generated timestamp')
|
|
sys.exit(1)
|
|
|
|
rows = []
|
|
for j in snap['jobs']:
|
|
rows.append({
|
|
'id': j['id'], 'name': j['name'], 'enabled': j['enabled'],
|
|
'paused': j['paused'], 'next': parse_iso(j.get('next_run_at')),
|
|
'last': parse_iso(j.get('last_run_at')),
|
|
'last_status': j.get('last_status'),
|
|
'last_error': j.get('last_error') or '', 'schedule': j.get('schedule'),
|
|
'deliver': j.get('deliver'), 'fire_claim': j.get('fire_claim'),
|
|
})
|
|
|
|
total = len(rows)
|
|
enabled = sum(1 for r in rows if r['enabled'])
|
|
paused = sum(1 for r in rows if r['paused'])
|
|
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]
|
|
never = [r for r in rows if not r['next']]
|
|
never = [r for r in never if not r['paused']]
|
|
|
|
fails = []
|
|
def check(label, actual, expected):
|
|
if actual != expected:
|
|
fails.append('%s: got %r expected %r' % (label, actual, expected))
|
|
|
|
# Internal partition of the fleet (from the frozen snapshot).
|
|
live_ids = sorted(r['id'] for r in live)
|
|
overdue_ids = sorted(r['id'] for r in overdue)
|
|
stale_ids = sorted(r['id'] for r in rows if r['paused'])
|
|
never_ids = sorted(r['id'] for r in never)
|
|
all_ids = sorted(r['id'] for r in rows)
|
|
check('partition.total', total, len(all_ids))
|
|
check('partition.enabled', enabled, len(live_ids) + len(overdue_ids))
|
|
check('partition.no-duplicates', len(all_ids), len(set(all_ids)))
|
|
check('partition.sum', len(live_ids) + len(overdue_ids) +
|
|
len(stale_ids) + len(never_ids), total)
|
|
|
|
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 '
|
|
r'settle or drop the draft \((\d+) paused\) and never-scheduled '
|
|
r'\((\d+)\) backlogs', s)
|
|
if not m:
|
|
fails.append('verdict line not found')
|
|
else:
|
|
check('verdict.live', int(m.group(1)), len(live_ids))
|
|
check('verdict.stale', int(m.group(2)), len(stale_ids))
|
|
check('verdict.never', int(m.group(3)), len(never_ids))
|
|
|
|
a_block = s.split('## A - Enabled & live')[1].split('## B - Draft-stalled')[0]
|
|
listed_a = sorted(re.findall(r'\|\s*([0-9a-f]{12,})\s*\|', a_block))
|
|
check('A.id.count', len(listed_a), len(live_ids))
|
|
check('A.ids', listed_a, live_ids)
|
|
|
|
b_block = s.split('## B - Draft-stalled')[1].split('## C - Never-scheduled')[0]
|
|
listed_b = sorted(re.findall(r'\|\s*([0-9a-f]{12,})\s*\|', b_block))
|
|
check('B.id.count', len(listed_b), len(stale_ids))
|
|
check('B.ids', listed_b, stale_ids)
|
|
|
|
c_block = s.split('## C - Never-scheduled')[1].split('## Fire-control lock cross-check')[0]
|
|
listed_c = sorted(re.findall(r'\|\s*([0-9a-f]{12,})\s*\|', c_block))
|
|
check('C.id.count', len(listed_c), len(never_ids))
|
|
check('C.ids', listed_c, never_ids)
|
|
|
|
if fails:
|
|
print('FAIL: cron-fleet triage refresh artifact')
|
|
for f in fails:
|
|
print(' - ' + f)
|
|
sys.exit(1)
|
|
|
|
print('PASS: snapshot verified at generated=%s' % t0)
|
|
print(' total=%d enabled=%d paused=%d live=%d overdue=%d never=%d' % (
|
|
total, enabled, paused, len(live_ids), len(overdue_ids), len(never_ids)))
|
|
print(' A=%d B=%d C=%d' % (len(live_ids), len(stale_ids), len(never_ids)))
|
|
print(' source jobs frozen in %s' % os.path.basename(OUT_JSON))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|