9edd3151ac
The in-cluster runner posts one State Hub message (custodian-allocation-review -> the-custodian) per newly retained week, records the message id in receipts.jsonl, never reposts, and retries an undelivered week. New weekly_allocation_check.py flags a failed latest receipt or a missing report for the due ISO week (Monday 12:00 Europe/Berlin); security_flow_daily.py notifies once per distinct flag set and stays inactive until a receipts source is configured. Founder decision 2026-10-21 (T05 follow-up answers). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
99 lines
4 KiB
Python
99 lines
4 KiB
Python
#!/usr/bin/env python3
|
|
"""CUST-WP-0071-T05 missed/failed-run check for the weekly allocation assessment.
|
|
|
|
Flags `failed` when the latest receipt failed, and `missing` when no retained
|
|
report exists for the due ISO week. The current week falls due on Monday 12:00
|
|
Europe/Berlin; before then the previous week is checked. Week labels follow
|
|
weekly_allocation_incluster.week_of (ISO week of UTC time minus 6h).
|
|
|
|
Receipts source (inactive when unset, so harmless before the CronJob runs):
|
|
- a path to a copy of the job's receipts.jsonl (failed and missing), or
|
|
- `hub`: the job's delivery messages (from custodian-allocation-review to
|
|
the-custodian). Only retained weeks are posted, so a failed run shows as missing.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from datetime import datetime, timedelta, timezone
|
|
import json
|
|
from pathlib import Path
|
|
from urllib.parse import urlencode
|
|
from urllib.request import urlopen
|
|
from zoneinfo import ZoneInfo
|
|
|
|
BERLIN = ZoneInfo("Europe/Berlin")
|
|
OK = {"retained", "already-retained"}
|
|
WINDOW_END_HOUR = 6
|
|
|
|
|
|
def week_label(moment: datetime) -> str:
|
|
year, week, _ = (moment.astimezone(timezone.utc) - timedelta(hours=WINDOW_END_HOUR)).isocalendar()
|
|
return f"{year}-W{week:02d}"
|
|
|
|
|
|
def due_week(now: datetime) -> str:
|
|
local = now.astimezone(BERLIN)
|
|
monday = (local - timedelta(days=local.weekday())).replace(hour=12, minute=0, second=0, microsecond=0)
|
|
return week_label(monday if local >= monday else monday - timedelta(days=7))
|
|
|
|
|
|
def evaluate(receipts: list[dict], now: datetime) -> dict:
|
|
week = due_week(now)
|
|
ordered = sorted(receipts, key=lambda r: r.get("run_at") or "")
|
|
flags = []
|
|
if ordered and ordered[-1].get("status") == "failed":
|
|
flags.append({"flag": "failed", "week": ordered[-1].get("week"), "run_at": ordered[-1].get("run_at"),
|
|
"error": ordered[-1].get("error")})
|
|
if not any(r.get("week") == week and r.get("status") in OK for r in receipts):
|
|
flags.append({"flag": "missing", "week": week,
|
|
"detail": "no retained weekly allocation report for the due ISO week"})
|
|
return {"status": "flagged" if flags else "ok", "due_week": week, "flags": flags,
|
|
"recovery": "inspect CronJob custodian-assessment/custodian-allocation-review; rerun with "
|
|
"kubectl create job --from=cronjob/custodian-allocation-review (same-week rerun is idempotent)"
|
|
if flags else None}
|
|
|
|
|
|
def from_file(path: Path) -> list[dict]:
|
|
out = []
|
|
for line in path.read_text().splitlines():
|
|
try:
|
|
out.append(json.loads(line))
|
|
except ValueError:
|
|
continue
|
|
return out
|
|
|
|
|
|
def from_hub(hub: str, opener=urlopen) -> list[dict]:
|
|
query = urlencode({"to_agent": "the-custodian", "from_agent": "custodian-allocation-review", "limit": 50})
|
|
with opener(hub.rstrip("/") + "/messages/?" + query, timeout=30) as response:
|
|
messages = json.load(response)
|
|
out = []
|
|
for message in messages:
|
|
try:
|
|
body = json.loads(message.get("body") or "{}")
|
|
except ValueError:
|
|
continue
|
|
if body.get("week"):
|
|
out.append({"week": body["week"], "status": "retained", "run_at": message.get("created_at")})
|
|
return out
|
|
|
|
|
|
def check(source: str | None, hub: str, now: datetime | None = None, opener=urlopen) -> dict:
|
|
if not source:
|
|
return {"status": "inactive", "reason": "no weekly allocation receipts source configured"}
|
|
receipts = from_hub(hub, opener) if source == "hub" else from_file(Path(source))
|
|
return evaluate(receipts, now or datetime.now(timezone.utc))
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
parser.add_argument("--receipts", help="receipts.jsonl path or 'hub'")
|
|
parser.add_argument("--hub", default="http://127.0.0.1:8000")
|
|
args = parser.parse_args()
|
|
result = check(args.receipts, args.hub)
|
|
print(json.dumps(result, indent=2))
|
|
return 1 if result["status"] == "flagged" else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|