CUST-WP-0071-T05: in-cluster weekly allocation runner (source only)

Stdlib-only CronJob entry point that reads the Kubernetes API (get/list on
nodes, pods, PVCs, PVs, storageclasses, metrics.k8s.io) and Prometheus, reuses
build_observation / reconcile.join / build_report / retain unchanged, and
retains ISO-week reports, per-attempt inputs and an append-only receipt log.
Telemetry window ends Monday 06:00 UTC so weeks stay comparable across DST.
Reconcile now imports PyYAML only in its CLI. Not deployed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

config/weekly-allocation-queries.json

@@ -0,0 +1,16 @@
+{
+  "schema": "custodian.weekly-allocation-queries.v1",
+  "note": "CUST-WP-0071-T05. Namespace-level aggregates proven on 2026-09-28 against telemetry/telemetry-prometheus. They populate namespace_context only; workload rows stay 'investigate' until workload-labelled queries exist (kube_pod_labels is not exported by the current kube-state-metrics). Changing any query text changes the report query_signature and breaks week-on-week comparability for one week.",
+  "window": "7d",
+  "step": "5m",
+  "queries": {
+    "cpu_p95_m": "quantile_over_time(0.95, (sum by(namespace) (rate(container_cpu_usage_seconds_total{container!=\"\",container!=\"POD\",namespace!=\"\"}[5m])) * 1000)[7d:5m])",
+    "cpu_peak_m": "max_over_time((sum by(namespace) (rate(container_cpu_usage_seconds_total{container!=\"\",container!=\"POD\",namespace!=\"\"}[5m])) * 1000)[7d:5m])",
+    "cpu_sample_count": "count_over_time((sum by(namespace) (rate(container_cpu_usage_seconds_total{container!=\"\",container!=\"POD\",namespace!=\"\"}[5m])) * 1000)[7d:5m])",
+    "memory_p95_bytes": "quantile_over_time(0.95, (sum by(namespace) (container_memory_working_set_bytes{container!=\"\",container!=\"POD\",namespace!=\"\"}))[7d:5m])",
+    "memory_peak_bytes": "max_over_time((sum by(namespace) (container_memory_working_set_bytes{container!=\"\",container!=\"POD\",namespace!=\"\"}))[7d:5m])",
+    "memory_sample_count": "count_over_time((sum by(namespace) (container_memory_working_set_bytes{container!=\"\",container!=\"POD\",namespace!=\"\"}))[7d:5m])",
+    "cpu_throttle_ratio": "sum by(namespace) (increase(container_cpu_cfs_throttled_periods_total{container!=\"\",container!=\"POD\",namespace!=\"\"}[7d])) / sum by(namespace) (increase(container_cpu_cfs_periods_total{container!=\"\",container!=\"POD\",namespace!=\"\"}[7d]))",
+    "restarts": "sum by(namespace) (increase(kube_pod_container_status_restarts_total[7d]))"
+  }
+}

docs/weekly-allocation-report.md

@@ -112,3 +112,42 @@ The tests cover historical namespace-only input, missing/zero/NaN, invalid
 aggregation, query failure, compatible/incompatible periods, stable identities
 and owner-task links, stale/future evidence, owner review, resource/capture
 mismatch and retained retry/collision behavior.
+
+## In-cluster runner (CUST-WP-0071-T05, founder decision 2026-10-21)
+
+`docs/decisions/2026-10-21-cust-wp-0071-delivery-interview.md` selects an
+in-cluster Kubernetes CronJob for T05 and supersedes the Activity Core
+placement above. `scripts/weekly_allocation_incluster.py` is the job entry
+point. It is standard-library only and reuses the existing owner code
+unchanged: railiance-cluster `build_observation`, `reconcile.join` (PyYAML is
+now imported only by the reconcile CLI) and this adapter's `build_report` and
+`retain`.
+
+- Reads: the Kubernetes API through the pod ServiceAccount, with get/list on
+  nodes, pods, persistentvolumeclaims, persistentvolumes, storageclasses and
+  metrics.k8s.io nodes/pods only. Prometheus is reached through
+  `PROMETHEUS_URL` with `config/weekly-allocation-queries.json`.
+- Week identity: the ISO week of (run time - 6h). The telemetry window ends at
+  Monday 06:00 UTC, so consecutive windows are exactly seven days apart in CET
+  and CEST. Prometheus keeps seven days of data, so the run must happen on the Monday. A
+  2026-10-21 (Wednesday) read-only dry run against railiance01 retained the
+  report, but per-namespace CPU sample counts reached at most 1512 of 2016,
+  because the window start had already expired.
+- Retention: `<data>/reports/<YYYY>-W<ww>.json` (create-without-replacement),
+  per-attempt inputs under `<data>/runs/<week>/<attempt>/`, and an append-only
+  `<data>/receipts.jsonl` with status `retained`, `already-retained` or
+  `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
+  are namespace-level. kube-state-metrics does not export `kube_pod_labels`, so
+  no workload-labelled series exist yet.
+
+Executor manifests (suspended, not registered with ArgoCD) live in
+railiance-platform `argocd/platform-addons/custodian-allocation-review/`, with
+the draft Application in `argocd/railiance01/drafts/`.
+
+```sh
+python3 -m pytest tests/test_weekly_allocation_incluster.py tests/test_weekly_allocation_report.py -q
+```

scripts/reconcile_railiance_allocation.py

@@ -10,8 +10,6 @@ from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
 
-import yaml
-
 ROOT = Path(__file__).resolve().parents[1]
 CLUSTER_OBS = Path.home() / "railiance-cluster" / "tools" / "observe_cluster_resources.py"
 HEADROOM = Path.home() / "state-hub" / "scripts" / "release_headroom_preflight.py"
@@ -131,6 +129,8 @@ def main() -> int:
         observation = json.loads(args.observation.read_text())
     else:
         observation = cluster.collect(args.kubectl)
+    import yaml  # main only: join() stays stdlib-only for the in-cluster runner
+
     owners = yaml.safe_load(OWNERS.read_text())
     report = join(observation, owners)
     preflight = None

scripts/weekly_allocation_incluster.py

@@ -0,0 +1,272 @@
+#!/usr/bin/env python3
+"""CUST-WP-0071-T05 in-cluster weekly runner: read-only collection, join, retain.
+
+Standard library only, so it runs on a pinned python:3.12-slim image with the
+scripts mounted from a ConfigMap. It reads the Kubernetes API with the pod's
+ServiceAccount (get/list only), queries Prometheus over plain HTTP, reuses the
+existing owner code unchanged (railiance-cluster build_observation, custodian
+reconcile join, weekly_allocation_report build_report/retain) and writes only
+under the data directory. It never resizes, creates tasks or delivers messages.
+
+Week identity is the ISO week of (run time - 6h); the telemetry window always
+ends at Monday 06:00 UTC of that week, so consecutive reports are exactly
+604800 s apart regardless of DST or scheduler jitter. 06:00 UTC is at or before
+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.
+"""
+from __future__ import annotations
+
+import argparse
+from datetime import datetime, timedelta, timezone
+import hashlib
+import importlib.util
+import json
+import os
+from pathlib import Path
+import re
+import ssl
+import sys
+import urllib.error
+import urllib.parse
+import urllib.request
+
+HERE = Path(__file__).resolve().parent
+SA_DIR = Path("/var/run/secrets/kubernetes.io/serviceaccount")
+WEEK = re.compile(r"^(\d{4})-W(\d{2})\.json$")
+# Exactly the reads the collector performs; mirrored by the ClusterRole.
+API_PATHS = {
+    "nodes": "/api/v1/nodes",
+    "pods": "/api/v1/pods",
+    "pvcs": "/api/v1/persistentvolumeclaims",
+    "pvs": "/api/v1/persistentvolumes",
+    "storageclasses": "/apis/storage.k8s.io/v1/storageclasses",
+}
+METRICS_PATHS = {
+    "node_top": "/apis/metrics.k8s.io/v1beta1/nodes",
+    "pod_top": "/apis/metrics.k8s.io/v1beta1/pods",
+}
+
+
+def load(path: Path, name: str):
+    spec = importlib.util.spec_from_file_location(name, path)
+    if spec is None or spec.loader is None:
+        raise RuntimeError(f"cannot load {path}")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def sha256(path: Path) -> str:
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def iso(value: datetime) -> str:
+    return value.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
+
+
+WINDOW_END_HOUR = 6
+
+
+def week_of(now: datetime) -> tuple[str, datetime]:
+    """ISO week label and its Monday 06:00 UTC (the telemetry window end)."""
+    shifted = now.astimezone(timezone.utc) - timedelta(hours=WINDOW_END_HOUR)
+    year, week, weekday = shifted.isocalendar()
+    monday = (shifted - timedelta(days=weekday - 1)).replace(hour=0, minute=0, second=0, microsecond=0)
+    return f"{year}-W{week:02d}", monday + timedelta(hours=WINDOW_END_HOUR)
+
+
+def missed_weeks(previous: str | None, current: str) -> list[str]:
+    if previous is None:
+        return []
+    def monday(label):
+        year, week = int(label[:4]), int(label[6:8])
+        return datetime.fromisocalendar(year, week, 1).replace(tzinfo=timezone.utc, hour=WINDOW_END_HOUR)
+    out, cursor = [], monday(previous) + timedelta(days=7)
+    while cursor < monday(current):
+        out.append(week_of(cursor)[0])
+        cursor += timedelta(days=7)
+    return out
+
+
+def previous_report(reports: Path, label: str) -> Path | None:
+    earlier = sorted(p for p in reports.glob("*.json") if WEEK.match(p.name) and p.stem < label)
+    return earlier[-1] if earlier else None
+
+
+class Kube:
+    """Minimal in-cluster API reader; the token is never logged or persisted."""
+
+    def __init__(self, base: str, token_path: Path, ca_path: Path):
+        self.base = base.rstrip("/")
+        self.token_path = token_path
+        self.context = ssl.create_default_context(cafile=str(ca_path))
+
+    def get(self, path: str):
+        request = urllib.request.Request(self.base + path, headers={
+            "Authorization": "Bearer " + self.token_path.read_text().strip(),
+            "Accept": "application/json"})
+        with urllib.request.urlopen(request, context=self.context, timeout=60) as response:
+            return json.load(response)
+
+
+def top_lines(metrics: dict, *, pods: bool, observer) -> str:
+    """Render metrics.k8s.io lists in the `kubectl top` shape parse_top reads."""
+    lines = []
+    for item in metrics.get("items") or []:
+        meta = item.get("metadata") or {}
+        if pods:
+            cpu = sum(observer.cpu_millicores((c.get("usage") or {}).get("cpu")) for c in item.get("containers") or [])
+            mem = sum(observer.bytes_value((c.get("usage") or {}).get("memory")) for c in item.get("containers") or [])
+            lines.append(f"{meta.get('namespace')} {meta.get('name')} {cpu}m {mem}")
+        else:
+            usage = item.get("usage") or {}
+            cpu = observer.cpu_millicores(usage.get("cpu"))
+            mem = observer.bytes_value(usage.get("memory"))
+            lines.append(f"{meta.get('name')} {cpu}m 0% {mem} 0%")
+    return "\n".join(lines) + ("\n" if lines else "")
+
+
+def collect_snapshots(get, observer) -> dict:
+    snapshots = {name: get(path) for name, path in API_PATHS.items()}
+    for name, path in METRICS_PATHS.items():
+        try:
+            snapshots[name] = top_lines(get(path), pods=name == "pod_top", observer=observer)
+        except (OSError, ValueError, urllib.error.URLError):
+            snapshots[name] = None  # recorded by build_observation as a measurement gap
+    return snapshots
+
+
+def prometheus_query(base: str, promql: str, at: datetime, opener=urllib.request.urlopen) -> dict:
+    """Return Prometheus' JSON response, including error bodies, never raise on 4xx/5xx."""
+    url = base.rstrip("/") + "/api/v1/query?" + urllib.parse.urlencode(
+        {"query": promql, "time": f"{at.timestamp():.3f}"})
+    try:
+        with opener(url, timeout=120) as response:
+            return json.load(response)
+    except urllib.error.HTTPError as exc:
+        try:
+            return json.loads(exc.read())
+        except ValueError:
+            return {"status": "error", "errorType": "http", "error": f"HTTP {exc.code}"}
+    except (OSError, ValueError) as exc:
+        return {"status": "error", "errorType": "transport", "error": type(exc).__name__}
+
+
+def write_json(path: Path, value) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_name(path.name + ".tmp")
+    temporary.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n")
+    os.replace(temporary, path)
+
+
+def append_receipt(data: Path, receipt: dict) -> None:
+    data.mkdir(parents=True, exist_ok=True)
+    with (data / "receipts.jsonl").open("a") as stream:
+        stream.write(json.dumps(receipt, sort_keys=True) + "\n")
+        stream.flush()
+        os.fsync(stream.fileno())
+    print(json.dumps(receipt, sort_keys=True))
+
+
+def run(args, *, now: datetime, get, query) -> dict:
+    sources = {
+        "weekly_allocation_report": args.source_dir / "weekly_allocation_report.py",
+        "reconcile_railiance_allocation": args.source_dir / "reconcile_railiance_allocation.py",
+        "observe_cluster_resources": args.observer,
+        "weekly_allocation_incluster": Path(__file__).resolve(),
+        "queries": args.queries, "ownership": args.ownership, "inventory": args.inventory,
+    }
+    weekly = load(sources["weekly_allocation_report"], "weekly_allocation_report")
+    reconcile = load(sources["reconcile_railiance_allocation"], "reconcile_railiance_allocation")
+    observer = load(sources["observe_cluster_resources"], "observe_cluster_resources")
+    label, window_end = week_of(now)
+    reports = args.data_dir / "reports"
+    target = reports / f"{label}.json"
+    receipt = {"schema": "custodian.weekly-allocation-receipt.v1", "week": label,
+               "run_at": iso(now), "report": str(target), "delivery": "not emitted",
+               "acknowledgment": None}
+    if target.exists():
+        return {**receipt, "status": "already-retained", "report_sha256": sha256(target)}
+    previous_path = previous_report(reports, label)
+    previous_label = previous_path.stem if previous_path else None
+
+    observation = observer.build_observation(collect_snapshots(get, observer), iso(now))
+    owners = json.loads(args.ownership.read_text())
+    allocation = reconcile.join(observation, owners)
+    spec = json.loads(args.queries.read_text())
+    if spec.get("schema") != "custodian.weekly-allocation-queries.v1":
+        raise ValueError("unsupported query specification")
+    telemetry = {"captured_at": window_end.isoformat(), "window": spec["window"], "step": spec["step"],
+                 "source": args.prometheus_url,
+                 "queries": {name: {"promql": promql, "response": query(promql, window_end)}
+                             for name, promql in sorted(spec["queries"].items())}}
+    provenance = {
+        "schema": "custodian.allocation-capture-provenance.v1", "workplan": "CUST-WP-0071-T05",
+        "source_observation": observation["captured_at"], "read_only": True,
+        "secret_surfaces_read": False, "executor": args.executor,
+        "kubernetes_reads": sorted(API_PATHS.values()) + sorted(METRICS_PATHS.values()),
+        "prometheus": args.prometheus_url, "telemetry_window_end": window_end.isoformat(),
+        "inputs": [{"name": n, "path": str(p), "sha256": sha256(p)} for n, p in sorted(sources.items())],
+    }
+    inventory = json.loads(args.inventory.read_text())
+    inputs = {"cluster": observation, "allocation": allocation, "telemetry": telemetry,
+              "inventory": inventory, "provenance": provenance}
+    attempt = args.data_dir / "runs" / label / iso(now).replace(":", "")
+    for name, value in inputs.items():
+        write_json(attempt / f"{name}.json", value)
+    report = weekly.build_report(
+        **inputs, as_of=iso(now), max_age_hours=args.max_age_hours,
+        previous=json.loads(previous_path.read_text()) if previous_path else None,
+        reviews=json.loads(args.reviews.read_text()) if args.reviews and args.reviews.exists() else None)
+    weekly.retain(target, report)
+    failed = sorted(n for n, q in telemetry["queries"].items() if q["response"].get("status") != "success")
+    return {**receipt, "status": "retained", "report_sha256": sha256(target),
+            "inputs_dir": str(attempt), "previous_week": previous_label,
+            "missed_weeks": missed_weeks(previous_label, label),
+            "comparison_compatible": report["comparison"]["compatible"],
+            "workloads": len(report["workloads"]), "failed_queries": failed,
+            "measurement_gaps": len(report["measurement_gaps"])}
+
+
+def parser() -> argparse.ArgumentParser:
+    env = os.environ.get
+    p = argparse.ArgumentParser(description=__doc__)
+    p.add_argument("--data-dir", type=Path, default=Path(env("ASSESSMENT_DATA_DIR", "/data")))
+    p.add_argument("--source-dir", type=Path, default=Path(env("ASSESSMENT_SOURCE_DIR", str(HERE))))
+    p.add_argument("--observer", type=Path, default=Path(env("ASSESSMENT_OBSERVER",
+                   str(Path(env("ASSESSMENT_SOURCE_DIR", str(HERE))) / "observe_cluster_resources.py"))))
+    p.add_argument("--queries", type=Path, default=Path(env("ASSESSMENT_QUERIES", str(HERE.parent / "config/weekly-allocation-queries.json"))))
+    p.add_argument("--ownership", type=Path, default=env("ASSESSMENT_OWNERSHIP") and Path(env("ASSESSMENT_OWNERSHIP")))
+    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("--executor", default=env("ASSESSMENT_EXECUTOR", "unspecified"))
+    p.add_argument("--max-age-hours", type=float, default=float(env("ASSESSMENT_MAX_AGE_HOURS", "48")))
+    return p
+
+
+def main() -> int:
+    args = parser().parse_args()
+    now = datetime.now(timezone.utc)
+    label = week_of(now)[0]
+    try:
+        if not args.ownership or not args.inventory:
+            raise ValueError("ownership and inventory inputs must be bound")
+        host = os.environ["KUBERNETES_SERVICE_HOST"]
+        port = os.environ.get("KUBERNETES_SERVICE_PORT", "443")
+        kube = Kube(f"https://{host}:{port}", SA_DIR / "token", SA_DIR / "ca.crt")
+        receipt = run(args, now=now, get=kube.get,
+                      query=lambda promql, at: prometheus_query(args.prometheus_url, promql, at))
+    except Exception as exc:  # noqa: BLE001 - every failure must leave a receipt
+        append_receipt(args.data_dir, {"schema": "custodian.weekly-allocation-receipt.v1",
+                       "week": label, "run_at": iso(now), "status": "failed",
+                       "error": f"{type(exc).__name__}: {exc}"[:500]})
+        return 1
+    append_receipt(args.data_dir, receipt)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/test_weekly_allocation_incluster.py

@@ -0,0 +1,168 @@
+import argparse
+from datetime import datetime, timezone
+import importlib.util
+import json
+from pathlib import Path
+import tempfile
+import unittest
+
+ROOT = Path(__file__).resolve().parents[1]
+OBSERVER = Path.home() / "railiance-cluster" / "tools" / "observe_cluster_resources.py"
+spec = importlib.util.spec_from_file_location("incluster", ROOT / "scripts/weekly_allocation_incluster.py")
+incluster = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(incluster)
+
+RESOURCE = "resource:railiance:reef-railiance:k3s"
+FAKE_OBSERVER = '''
+def cpu_millicores(v):
+    v = str(v or "0")
+    return int(float(v[:-1]) / 1e6) if v.endswith("n") else int(v[:-1]) if v.endswith("m") else int(float(v) * 1000)
+def bytes_value(v):
+    v = str(v or "0")
+    return int(v[:-2]) * 1024 if v.endswith("Ki") else int(v)
+def build_observation(snapshots, captured_at):
+    rows = [{"namespace": p["metadata"]["namespace"], "workload": p["metadata"]["labels"]["app"], "pods": 1,
+             "requests": {"cpu_millicores": 10, "memory_bytes": 1}, "limits": {}} for p in snapshots["pods"]["items"]]
+    return {"schema_version": "railiance.cluster-resource-observation.v1", "resource_id": "%s",
+            "captured_at": captured_at, "utilization": {"workloads": rows},
+            "top": {"node": snapshots["node_top"], "pod": snapshots["pod_top"]},
+            "measurement_gaps": [] if snapshots["pod_top"] is not None else ["pod metrics unavailable"]}
+''' % RESOURCE
+
+
+def pod(ns, name, app):
+    return {"metadata": {"namespace": ns, "name": name, "labels": {"app": app}},
+            "spec": {"nodeName": "railiance01", "containers": [{"name": "c", "resources": {"requests": {"cpu": "10m", "memory": "16Mi"}}}]},
+            "status": {"phase": "Running"}}
+
+
+class InClusterRunnerTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        base = Path(self.tmp.name)
+        self.data = base / "data"
+        (base / "ownership.json").write_text(json.dumps({
+            "host": {"cluster_resource_id": RESOURCE},
+            "namespaces": {"pilot": {"owner": "railiance-apps", "service": "pilot", "tenant": "unknown"}}}))
+        (base / "inventory.json").write_text(json.dumps({"id": RESOURCE}))
+        (base / "observer.py").write_text(FAKE_OBSERVER)
+        self.reads, self.queries = [], []
+        self.metrics_fail = False
+        self.args = argparse.Namespace(
+            data_dir=self.data, source_dir=ROOT / "scripts", observer=base / "observer.py",
+            queries=ROOT / "config/weekly-allocation-queries.json", ownership=base / "ownership.json",
+            inventory=base / "inventory.json", reviews=None, prometheus_url="http://prom.test:9090",
+            executor="test", max_age_hours=48.0)
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def get(self, path):
+        self.reads.append(path)
+        if path.startswith("/apis/metrics.k8s.io"):
+            if self.metrics_fail:
+                raise OSError("metrics unavailable")
+            if path.endswith("/nodes"):
+                return {"items": [{"metadata": {"name": "railiance01"}, "usage": {"cpu": "1500000000n", "memory": "2048Ki"}}]}
+            return {"items": [{"metadata": {"namespace": "pilot", "name": "app-1"},
+                               "containers": [{"usage": {"cpu": "3m", "memory": "1Ki"}}, {"usage": {"cpu": "2m", "memory": "1Ki"}}]}]}
+        if path == "/api/v1/pods":
+            return {"items": [pod("pilot", "app-1", "app")]}
+        return {"items": []}
+
+    def query(self, promql, at):
+        self.queries.append((promql, at))
+        return {"status": "success", "data": {"resultType": "vector", "result": [
+            {"metric": {"namespace": "pilot"}, "value": [at.timestamp(), "1.5"]}]}}
+
+    def run_at(self, when):
+        return incluster.run(self.args, now=when, get=self.get, query=self.query)
+
+    def test_week_identity_and_window_end_are_dst_independent(self):
+        summer = datetime(2026, 10, 19, 6, 0, 30, tzinfo=timezone.utc)   # 08:00 CEST
+        winter = datetime(2026, 11, 2, 7, 0, 30, tzinfo=timezone.utc)    # 08:00 CET
+        self.assertEqual(incluster.week_of(summer), ("2026-W43", datetime(2026, 10, 19, 6, tzinfo=timezone.utc)))
+        self.assertEqual(incluster.week_of(winter), ("2026-W45", datetime(2026, 11, 2, 6, tzinfo=timezone.utc)))
+        early = datetime(2026, 10, 19, 5, 59, tzinfo=timezone.utc)
+        self.assertEqual(incluster.week_of(early)[0], "2026-W42")
+        self.assertEqual(incluster.missed_weeks("2026-W41", "2026-W44"), ["2026-W42", "2026-W43"])
+        self.assertEqual(incluster.missed_weeks("2026-W52", "2027-W01"), ["2026-W53"])  # 2026 has 53 ISO weeks
+        self.assertEqual(incluster.missed_weeks("2026-W53", "2027-W01"), [])
+
+    def test_reads_only_collector_paths_and_retains_report(self):
+        receipt = self.run_at(datetime(2026, 10, 19, 6, 0, 30, tzinfo=timezone.utc))
+        self.assertEqual(receipt["status"], "retained")
+        self.assertEqual(set(self.reads), set(incluster.API_PATHS.values()) | set(incluster.METRICS_PATHS.values()))
+        self.assertFalse(any("secret" in r or "configmap" in r for r in self.reads))
+        report = json.loads((self.data / "reports/2026-W43.json").read_text())
+        self.assertEqual(report["period"]["capture_end"], "2026-10-19T06:00:00+00:00")
+        self.assertEqual(report["workloads"][0]["owner"], "railiance-apps")
+        self.assertIn("pilot", report["namespace_context"])
+        self.assertTrue(all(at == datetime(2026, 10, 19, 6, tzinfo=timezone.utc) for _, at in self.queries))
+        cluster = json.loads((Path(receipt["inputs_dir"]) / "cluster.json").read_text())
+        self.assertEqual(cluster["top"]["pod"], "pilot app-1 5m 2048\n")
+        self.assertEqual(cluster["top"]["node"], "railiance01 1500m 0% 2097152 0%\n")
+        provenance = json.loads((Path(receipt["inputs_dir"]) / "provenance.json").read_text())
+        self.assertFalse(provenance["secret_surfaces_read"])
+        self.assertEqual(provenance["source_observation"], cluster["captured_at"])
+
+    def test_retry_in_same_week_is_noop_and_next_week_compares(self):
+        first = datetime(2026, 10, 19, 6, 0, 30, tzinfo=timezone.utc)
+        self.run_at(first)
+        before = len(self.reads)
+        again = self.run_at(datetime(2026, 10, 20, 9, 0, tzinfo=timezone.utc))
+        self.assertEqual(again["status"], "already-retained")
+        self.assertEqual(len(self.reads), before)
+        nxt = self.run_at(datetime(2026, 10, 26, 7, 0, 30, tzinfo=timezone.utc))
+        self.assertEqual((nxt["previous_week"], nxt["missed_weeks"]), ("2026-W43", []))
+        self.assertTrue(nxt["comparison_compatible"])
+        later = self.run_at(datetime(2026, 11, 16, 7, 0, 30, tzinfo=timezone.utc))
+        self.assertEqual(later["missed_weeks"], ["2026-W45", "2026-W46"])
+        self.assertFalse(later["comparison_compatible"])
+
+    def test_metrics_api_outage_is_a_gap_and_query_errors_are_reported(self):
+        self.metrics_fail = True
+        self.query = lambda promql, at: {"status": "error", "errorType": "bad_data", "error": "x"}
+        receipt = self.run_at(datetime(2026, 10, 19, 6, 0, 30, tzinfo=timezone.utc))
+        self.assertEqual(receipt["status"], "retained")
+        self.assertIn("cpu_p95_m", receipt["failed_queries"])
+        self.assertGreaterEqual(receipt["measurement_gaps"], 1)
+
+    def test_failure_leaves_failed_receipt_and_no_report(self):
+        self.args.ownership = None
+        with unittest.mock.patch.object(incluster, "parser") as fake_parser, \
+                unittest.mock.patch.object(incluster, "datetime") as fake_dt:
+            fake_parser.return_value.parse_args.return_value = self.args
+            fake_dt.now.return_value = datetime(2026, 10, 19, 6, 0, 30, tzinfo=timezone.utc)
+            self.assertEqual(incluster.main(), 1)
+        receipts = [json.loads(l) for l in (self.data / "receipts.jsonl").read_text().splitlines()]
+        self.assertEqual(receipts[-1]["status"], "failed")
+        self.assertFalse((self.data / "reports").exists())
+
+    def test_prometheus_http_error_body_is_retained(self):
+        import io
+        import urllib.error
+
+        def opener(url, timeout):
+            raise urllib.error.HTTPError(url, 422, "x", {}, io.BytesIO(b'{"status":"error","error":"bad"}'))
+        self.assertEqual(incluster.prometheus_query("http://p", "up", datetime.now(timezone.utc), opener)["error"], "bad")
+
+    @unittest.skipUnless(OBSERVER.exists(), "railiance-cluster checkout not present")
+    def test_real_collector_accepts_rendered_top_lines(self):
+        self.args.observer = OBSERVER
+        self.reads.clear()
+        observer = incluster.load(OBSERVER, "observer_real")
+        snapshots = incluster.collect_snapshots(lambda p: {"items": [{"metadata": {"name": "railiance01"},
+            "status": {"allocatable": {"cpu": "4", "memory": "8Gi", "pods": "110"}}}]} if p == "/api/v1/nodes"
+            else self.get(p), observer)
+        observation = observer.build_observation(snapshots, "2026-10-19T06:00:30Z")
+        self.assertEqual(observation["resource_id"], RESOURCE)
+        self.assertNotIn("node metrics unavailable", observation["measurement_gaps"])
+        self.assertNotIn("pod metrics unavailable", observation["measurement_gaps"])
+        self.assertIn('"cpu_millicores": 1500', json.dumps(observation))
+
+
+import unittest.mock  # noqa: E402
+
+if __name__ == "__main__":
+    unittest.main()