From 9edd3151ac5c93e5319a9581c0df145f8d93c68f Mon Sep 17 00:00:00 2001 From: codex Date: Wed, 21 Oct 2026 08:40:12 +0200 Subject: [PATCH] CUST-WP-0071-T05: hub delivery per weekly report and daily missed/failed-run check 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 --- Makefile | 2 +- docs/weekly-allocation-report.md | 21 ++++- scripts/security_flow_daily.py | 33 +++++++- scripts/weekly_allocation_check.py | 99 +++++++++++++++++++++++ scripts/weekly_allocation_incluster.py | 65 ++++++++++++++- tests/test_weekly_allocation_check.py | 74 +++++++++++++++++ tests/test_weekly_allocation_incluster.py | 49 +++++++++++ 7 files changed, 337 insertions(+), 6 deletions(-) create mode 100644 scripts/weekly_allocation_check.py create mode 100644 tests/test_weekly_allocation_check.py diff --git a/Makefile b/Makefile index d0474b3..bcbdab2 100644 --- a/Makefile +++ b/Makefile @@ -5,7 +5,7 @@ security-flow-daily: ## Sync resolved decisions and notify changed actionable wo python3 scripts/security_flow_daily.py --apply security-flow-check: ## Run the deterministic security flow and generated commit tests - python3 -m pytest tests/test_security_flow_daily.py tests/test_generated_commit_guard.py -q + python3 -m pytest tests/test_security_flow_daily.py tests/test_generated_commit_guard.py tests/test_weekly_allocation_check.py -q generated-commit-check: ## Reject staged generated-only changes python3 scripts/check_generated_only_commit.py --staged diff --git a/docs/weekly-allocation-report.md b/docs/weekly-allocation-report.md index a17c3bc..06bafe9 100644 --- a/docs/weekly-allocation-report.md +++ b/docs/weekly-allocation-report.md @@ -139,8 +139,25 @@ now imported only by the reconcile CLI) and this adapter's `build_report` and `failed`, previous week, `missed_weeks`, comparability and failed queries. - Idempotency: a retained week is a no-op success. A failed attempt leaves no report, so the Job retry or a manual run collects again. -- Not done: delivery, owner acknowledgment and owner-task creation - (`delivery: not emitted`). Workload rows stay `investigate` because the queries +- Delivery (founder decision 2026-10-21, "Hub message + repo tasks"): with + `STATE_HUB_URL` set, each newly retained week produces one State Hub message + from `custodian-allocation-review` to `the-custodian`. The message carries a + summary, the report path and its sha256, and its id is recorded in + `receipts.jsonl`. A week that already has a message id is never posted again. + If a post fails, the receipt says `delivery: failed`, the Job exits 1, and + its retry posts for the already-retained week. The next Custodian session + reviews the report and records recommendations as tasks in the owner + repositories (ADR-001). The job itself creates no tasks. +- Missed/failed-run check: `scripts/weekly_allocation_check.py`, which + `scripts/security_flow_daily.py` calls. It flags `failed` when the latest + receipt failed, and `missing` when the due ISO week has no retained report + (the current week falls due on Monday 12:00 Europe/Berlin). The agent is + notified once per distinct flag set. It is inactive until + `CUSTODIAN_ALLOCATION_RECEIPTS` (or `--allocation-receipts`) is set to `hub` + (delivery messages, where a failure shows as missing) or to a path holding + a copy of `receipts.jsonl`. Set it in `infra/security-flow/` only when the + CronJob is activated. +- Not done: owner acknowledgment. Workload rows stay `investigate` because the queries are namespace-level. kube-state-metrics does not export `kube_pod_labels`, so no workload-labelled series exist yet. diff --git a/scripts/security_flow_daily.py b/scripts/security_flow_daily.py index 1d18054..c33512c 100644 --- a/scripts/security_flow_daily.py +++ b/scripts/security_flow_daily.py @@ -6,12 +6,14 @@ Dry-run by default. --apply is the scheduled mode. No credentials are stored. import argparse import fcntl import json +import os from pathlib import Path import subprocess import sys from urllib.request import Request, urlopen from security_flow_report import DEFAULT_REPOS +import weekly_allocation_check SCRIPTS = Path(__file__).resolve().parent @@ -27,15 +29,26 @@ def notifications(report, state, agent, founder): return pending -def post(hub, recipient, payload): +def post(hub, recipient, payload, subject="Daily security flow"): body = {"from_agent": "the-custodian", "to_agent": recipient, - "subject": "Daily security flow", "body": json.dumps(payload, sort_keys=True)} + "subject": subject, "body": json.dumps(payload, sort_keys=True)} request = Request(hub.rstrip("/") + "/messages/", data=json.dumps(body).encode(), headers={"Content-Type": "application/json"}, method="POST") with urlopen(request, timeout=30) as response: return json.load(response)["id"] +def allocation_notice(result, state): + """CUST-WP-0071-T05: notify once per distinct flag set; clear when healthy.""" + if result.get("status") != "flagged": + if result.get("status") == "ok": + state.pop("allocation", None) + return None + if result["flags"] == state.get("allocation"): + return None + return {"WEEKLY_ALLOCATION": result} + + def save(path, data): temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text(json.dumps(data, indent=2) + "\n") @@ -50,6 +63,8 @@ def main(): parser.add_argument("--agent", default="the-custodian") parser.add_argument("--founder", default="human") parser.add_argument("--state-dir", type=Path, default=Path.home() / ".local/state/custodian-security-flow") + parser.add_argument("--allocation-receipts", default=os.environ.get("CUSTODIAN_ALLOCATION_RECEIPTS"), + help="weekly allocation receipts.jsonl path or 'hub'; unset keeps the check inactive") parser.add_argument("--apply", action="store_true") args = parser.parse_args() args.state_dir.mkdir(parents=True, exist_ok=True) @@ -72,6 +87,20 @@ def main(): state[key] = payload state[key + "_message_id"] = message_id save(state_path, state) # A later recipient failure does not replay earlier sends. + try: + allocation = weekly_allocation_check.check(args.allocation_receipts, args.hub) + except (OSError, ValueError) as exc: # never break the security flow + allocation = {"status": "error", "error": type(exc).__name__} + notice = allocation_notice(allocation, state) + if notice and not args.apply: + print(json.dumps({"dry_run": True, "to_agent": args.agent, "report": notice})) + elif notice: + state["allocation_message_id"] = post(args.hub, args.agent, notice, "Weekly allocation assessment check") + state["allocation"] = allocation["flags"] + if args.apply: + save(state_path, state) + if allocation["status"] != "inactive": + print(json.dumps({"weekly_allocation": allocation}, sort_keys=True)) print(raw) diff --git a/scripts/weekly_allocation_check.py b/scripts/weekly_allocation_check.py new file mode 100644 index 0000000..154ec8c --- /dev/null +++ b/scripts/weekly_allocation_check.py @@ -0,0 +1,99 @@ +#!/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()) diff --git a/scripts/weekly_allocation_incluster.py b/scripts/weekly_allocation_incluster.py index ddfea2c..be72ea4 100644 --- a/scripts/weekly_allocation_incluster.py +++ b/scripts/weekly_allocation_incluster.py @@ -15,6 +15,12 @@ the Monday 08:00 Europe/Berlin schedule in both CET and CEST and keeps the window start as close as possible to the seven-day Prometheus retention edge. A week whose report is already retained is a no-op success (idempotent retry); a failed attempt leaves no report, so the Job's retry or the next manual run collects again. + +Delivery (founder decision 2026-10-21, "Hub message + repo tasks"): after a +week is retained, one State Hub message goes from custodian-allocation-review +to the-custodian with a short summary, report path and sha256. The message id +is recorded in receipts.jsonl; a week that already has a message id is never +posted again, and a retained-but-undelivered week is posted by the next run. """ from __future__ import annotations @@ -170,6 +176,61 @@ def append_receipt(data: Path, receipt: dict) -> None: print(json.dumps(receipt, sort_keys=True)) +def delivered_message(data: Path, week: str) -> str | None: + path = data / "receipts.jsonl" + if not path.exists(): + return None + for line in reversed(path.read_text().splitlines()): + try: + entry = json.loads(line) + except ValueError: + continue + if entry.get("week") == week and entry.get("message_id"): + return entry["message_id"] + return None + + +def summary(receipt: dict) -> dict: + report = json.loads(Path(receipt["report"]).read_text()) + decisions: dict[str, int] = {} + for item in report["recommendations"]: + decisions[item["decision"]] = decisions.get(item["decision"], 0) + 1 + return {"week": receipt["week"], "report": receipt["report"], "report_sha256": receipt["report_sha256"], + "capture_end": report["period"]["capture_end"], "workloads": len(report["workloads"]), + "namespaces_with_metrics": len(report["namespace_context"]), + "recommendations": dict(sorted(decisions.items())), + "comparison_compatible": report["comparison"]["compatible"], + "measurement_gaps": len(report["measurement_gaps"]), + "failed_queries": sorted(n for n, q in report["queries"].items() if q.get("response_status") != "success"), + "next_step": "Custodian session reviews and records owner-repo tasks (ADR-001); no automatic resizing."} + + +def hub_post(hub: str, body: dict) -> str: + request = urllib.request.Request(hub.rstrip("/") + "/messages/", data=json.dumps(body).encode(), + headers={"Content-Type": "application/json"}, method="POST") + with urllib.request.urlopen(request, timeout=30) as response: + return str(json.load(response)["id"]) + + +def deliver(data: Path, receipt: dict, hub: str | None, post=hub_post) -> tuple[dict, bool]: + """Post one hub message per retained week; returns (receipt, ok).""" + if receipt.get("status") not in {"retained", "already-retained"}: + return receipt, True + existing = delivered_message(data, receipt["week"]) + if existing: + return {**receipt, "delivery": "already-posted", "message_id": existing}, True + if not hub: + return {**receipt, "delivery": "not configured"}, True + content = summary(receipt) + body = {"from_agent": "custodian-allocation-review", "to_agent": "the-custodian", + "subject": f"Weekly allocation assessment {receipt['week']}", + "body": json.dumps(content, sort_keys=True)} + try: + return {**receipt, "delivery": "posted", "message_id": post(hub, body)}, True + except (OSError, ValueError, KeyError) as exc: + return {**receipt, "delivery": f"failed: {type(exc).__name__}"}, False + + def run(args, *, now: datetime, get, query) -> dict: sources = { "weekly_allocation_report": args.source_dir / "weekly_allocation_report.py", @@ -242,6 +303,7 @@ def parser() -> argparse.ArgumentParser: p.add_argument("--inventory", type=Path, default=env("ASSESSMENT_INVENTORY") and Path(env("ASSESSMENT_INVENTORY"))) p.add_argument("--reviews", type=Path, default=env("ASSESSMENT_REVIEWS") and Path(env("ASSESSMENT_REVIEWS"))) p.add_argument("--prometheus-url", default=env("PROMETHEUS_URL", "http://telemetry-prometheus.telemetry.svc:9090")) + p.add_argument("--hub-url", default=env("STATE_HUB_URL"), help="State Hub API; unset disables delivery") p.add_argument("--executor", default=env("ASSESSMENT_EXECUTOR", "unspecified")) p.add_argument("--max-age-hours", type=float, default=float(env("ASSESSMENT_MAX_AGE_HOURS", "48"))) return p @@ -264,8 +326,9 @@ def main() -> int: "week": label, "run_at": iso(now), "status": "failed", "error": f"{type(exc).__name__}: {exc}"[:500]}) return 1 + receipt, delivered = deliver(args.data_dir, receipt, args.hub_url) append_receipt(args.data_dir, receipt) - return 0 + return 0 if delivered else 1 # undelivered: Job retry posts it (report already retained) if __name__ == "__main__": diff --git a/tests/test_weekly_allocation_check.py b/tests/test_weekly_allocation_check.py new file mode 100644 index 0000000..44afb79 --- /dev/null +++ b/tests/test_weekly_allocation_check.py @@ -0,0 +1,74 @@ +from datetime import datetime, timezone +import io +import json +from pathlib import Path +import sys + +SCRIPTS = Path(__file__).resolve().parents[1] / "scripts" +sys.path.insert(0, str(SCRIPTS)) +import security_flow_daily as daily # noqa: E402 +import weekly_allocation_check as check # noqa: E402 + +UTC = timezone.utc + + +def test_due_week_turns_over_monday_noon_berlin(): + assert check.due_week(datetime(2026, 10, 19, 9, 59, tzinfo=UTC)) == "2026-W42" # 11:59 CEST + assert check.due_week(datetime(2026, 10, 19, 10, 0, tzinfo=UTC)) == "2026-W43" # 12:00 CEST + assert check.due_week(datetime(2026, 11, 2, 10, 59, tzinfo=UTC)) == "2026-W44" # 11:59 CET + assert check.due_week(datetime(2026, 11, 2, 11, 0, tzinfo=UTC)) == "2026-W45" + assert check.due_week(datetime(2026, 10, 25, 20, 0, tzinfo=UTC)) == "2026-W43" # Sunday + + +def test_missing_failed_and_ok(): + now = datetime(2026, 10, 19, 12, 0, tzinfo=UTC) + assert check.evaluate([], now)["flags"][0]["flag"] == "missing" + ok = [{"week": "2026-W43", "status": "retained", "run_at": "2026-10-19T06:00:30Z"}] + assert check.evaluate(ok, now)["status"] == "ok" + failed = ok + [{"week": "2026-W43", "status": "failed", "run_at": "2026-10-19T07:00:00Z", "error": "x"}] + assert [f["flag"] for f in check.evaluate(failed, now)["flags"]] == ["failed"] + only_failed = [failed[1]] + assert [f["flag"] for f in check.evaluate(only_failed, now)["flags"]] == ["failed", "missing"] + + +def test_inactive_without_source_and_file_and_hub_sources(tmp_path): + assert check.check(None, "http://hub")["status"] == "inactive" + receipts = tmp_path / "receipts.jsonl" + receipts.write_text(json.dumps({"week": "2026-W43", "status": "already-retained", "run_at": "a"}) + "\npartial{\n") + now = datetime(2026, 10, 20, tzinfo=UTC) + assert check.check(str(receipts), "http://hub", now)["status"] == "ok" + urls = [] + + def opener(url, timeout): + urls.append(url) + return io.BytesIO(json.dumps([{"body": json.dumps({"week": "2026-W42"}), "created_at": "z"}]).encode()) + result = check.check("hub", "http://hub", now, opener) + assert "from_agent=custodian-allocation-review" in urls[0] + assert result["flags"][0] == {"flag": "missing", "week": "2026-W43", + "detail": "no retained weekly allocation report for the due ISO week"} + + +def test_daily_notice_is_deduplicated_and_cleared(): + flagged = {"status": "flagged", "flags": [{"flag": "missing", "week": "2026-W43"}]} + state = {} + assert daily.allocation_notice(flagged, state) == {"WEEKLY_ALLOCATION": flagged} + state["allocation"] = flagged["flags"] + assert daily.allocation_notice(flagged, state) is None + assert daily.allocation_notice({"status": "ok"}, state) is None and "allocation" not in state + assert daily.allocation_notice({"status": "inactive"}, {}) is None + + +def test_daily_flow_posts_allocation_flag_once(tmp_path, monkeypatch): + report = {"READY": [], "UNBLOCKED": [], "ATTENDED": []} + receipts = tmp_path / "receipts.jsonl" + receipts.write_text(json.dumps({"week": "2020-W01", "status": "failed", "run_at": "x"}) + "\n") + monkeypatch.setattr(sys, "argv", ["d", "--apply", "--state-dir", str(tmp_path), "--allocation-receipts", str(receipts)]) + monkeypatch.setattr(daily.subprocess, "run", lambda *a, **kw: None) + monkeypatch.setattr(daily.subprocess, "check_output", lambda *a, **kw: json.dumps(report)) + sent = [] + monkeypatch.setattr(daily, "post", lambda hub, to, payload, subject="Daily security flow": sent.append((to, subject, payload)) or "m1") + daily.main() + daily.main() + allocation = [s for s in sent if s[1] == "Weekly allocation assessment check"] + assert len(allocation) == 1 and allocation[0][0] == "the-custodian" + assert json.loads((tmp_path / "delivery.json").read_text())["allocation_message_id"] == "m1" diff --git a/tests/test_weekly_allocation_incluster.py b/tests/test_weekly_allocation_incluster.py index e8af114..a711651 100644 --- a/tests/test_weekly_allocation_incluster.py +++ b/tests/test_weekly_allocation_incluster.py @@ -166,3 +166,52 @@ import unittest.mock # noqa: E402 if __name__ == "__main__": unittest.main() + + +class DeliveryTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.data = Path(self.tmp.name) + report = {"recommendations": [{"decision": "investigate"}, {"decision": "investigate"}], + "period": {"capture_end": "2026-10-19T06:00:00+00:00"}, "workloads": [{}, {}], + "namespace_context": {"a": {}}, "comparison": {"compatible": False}, + "measurement_gaps": [], "queries": {"cpu_p95_m": {"response_status": "success"}}} + (self.data / "r.json").write_text(json.dumps(report)) + self.receipt = {"week": "2026-W43", "status": "retained", "report": str(self.data / "r.json"), + "report_sha256": "abc"} + self.posts = [] + + def tearDown(self): + self.tmp.cleanup() + + def post(self, hub, body): + self.posts.append((hub, body)) + return "msg-1" + + def test_posts_once_per_week_and_records_id(self): + receipt, ok = incluster.deliver(self.data, self.receipt, "http://hub", self.post) + self.assertTrue(ok) + self.assertEqual((receipt["delivery"], receipt["message_id"]), ("posted", "msg-1")) + hub, body = self.posts[0] + self.assertEqual((body["from_agent"], body["to_agent"]), ("custodian-allocation-review", "the-custodian")) + content = json.loads(body["body"]) + self.assertEqual((content["report_sha256"], content["recommendations"]), ("abc", {"investigate": 2})) + incluster.append_receipt(self.data, receipt) + again, ok = incluster.deliver(self.data, {**self.receipt, "status": "already-retained"}, "http://hub", self.post) + self.assertEqual((again["delivery"], again["message_id"], len(self.posts)), ("already-posted", "msg-1", 1)) + + def test_failed_post_is_retried_by_next_run(self): + def broken(hub, body): + raise OSError("down") + receipt, ok = incluster.deliver(self.data, self.receipt, "http://hub", broken) + self.assertFalse(ok) + self.assertEqual(receipt["delivery"], "failed: OSError") + incluster.append_receipt(self.data, receipt) + retry, ok = incluster.deliver(self.data, {**self.receipt, "status": "already-retained"}, "http://hub", self.post) + self.assertTrue(ok) + self.assertEqual(retry["delivery"], "posted") + + def test_unconfigured_hub_and_failed_runs_do_not_post(self): + self.assertEqual(incluster.deliver(self.data, self.receipt, None, self.post)[0]["delivery"], "not configured") + self.assertEqual(incluster.deliver(self.data, {"status": "failed", "week": "x"}, "http://hub", self.post)[1], True) + self.assertEqual(self.posts, [])