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 <noreply@anthropic.com>

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

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.
 

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)
 
 

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())

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__":

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"

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, [])