feat: prepare pinned Custodian worker bundle and required input admission

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a15374-f59f-7a31-86d2-d23b5d2ed181

Dockerfile

@@ -28,8 +28,9 @@ COPY Dockerfile ./Dockerfile
 COPY .forgejo/workflows/image.yaml ./.forgejo/workflows/image.yaml
 COPY schemas/ ./schemas/
 COPY k8s/ ./k8s/
-COPY scripts/render_gitops.py scripts/check_gitops_promotion.py ./scripts/
+COPY scripts/render_gitops.py scripts/check_gitops_promotion.py scripts/verify_custodian_bundle.py ./scripts/
-RUN uv sync --frozen --extra dev && uv run --no-sync python scripts/render_gitops.py --check --verify-frontend --verify-platform && uv run --no-sync pytest -q -p no:cacheprovider tests/test_frontend_patterns.py tests/test_frontend_monthly_review.py tests/test_pqrst_monthly.py tests/test_repository_grant.py tests/test_instruction_evaluation.py tests/test_admin_sync_api.py tests/test_gitops_release.py tests/test_release_broker.py tests/test_release_transport.py tests/test_release_dispatch.py
+COPY activity-definitions/ ./activity-definitions/
+RUN uv sync --frozen --extra dev && uv run --no-sync python scripts/render_gitops.py --check --verify-frontend --verify-platform && uv run --no-sync pytest -q -p no:cacheprovider tests/test_frontend_patterns.py tests/test_frontend_monthly_review.py tests/test_pqrst_monthly.py tests/test_repository_grant.py tests/test_instruction_evaluation.py tests/test_admin_sync_api.py tests/test_gitops_release.py tests/test_release_broker.py tests/test_release_transport.py tests/test_release_dispatch.py tests/test_custodian_bundle.py tests/test_custodian_inputs.py tests/test_custodian_operations.py
 
 # Stage 2 — runtime image
 FROM python:3.12-slim@sha256:78387bc3881b8273120a12ebe6c1ab22b018ccc2c9adf565ae1ac9b536e184ea AS runtime
@@ -40,6 +41,10 @@ COPY --from=builder /opt/repo-manager/.venv /opt/repo-manager/.venv
 COPY alembic.ini ./
 COPY migrations/ ./migrations/
 COPY scripts/ ./scripts/
+# Pinned owner commands; local hash verification performs no network or execution.
+COPY k8s/custodian-release/commands/ /opt/custodian-commands/
+COPY k8s/custodian-release/input-policy.json /opt/custodian-input-policy.json
+RUN python /app/scripts/verify_custodian_bundle.py --root /opt/custodian-commands
 COPY activity-definitions/ ./activity-definitions/
 COPY event-types/ ./event-types/
 COPY tasks/ ./tasks/

SCOPE.md

@@ -381,3 +381,12 @@ placement, accepted mounts/network, delivery and first unattended proof remain
 unverified; see docs/custodian-bounded-operations.md. This adds no scheduler or
 generic shell. The generated .custodian-brief.md is now an ignored local
 orientation snapshot; its existing bytes remain local.
+
+
+Custodian delivery preparation now includes an image-baked, hash-verified four-file
+owner bundle, tunable input policy, and a required daily pre-execution gate validating
+completion markers, exact snapshot hashes and independent per-repo Forge
+freshness receipts. The reviewed candidate in `k8s/custodian-release/` is outside
+active kustomization; placeholders prevent claiming an accepted image/snapshot.
+Both definitions stay disabled. No runtime publication, retention deletion or
+schedule activation is implemented by these source artifacts.

docs/custodian-bounded-operations.md

@@ -74,3 +74,11 @@ dispatch. It never invokes the real security command, Hub or schedule activation
 
 Concrete delivery options and unresolved owner choices are recorded in
 [custodian-runtime-binding-proposal.md](custodian-runtime-binding-proposal.md).
+
+Daily execution requires `ACTIVITY_CORE_CUSTODIAN_INPUT_POLICY` and `ACTIVITY_CORE_CUSTODIAN_FRESHNESS_RECEIPTS`, a protected JSON array of
+existing Repo Manager receipts. Admission verifies exact complete snapshot
+contents and fresh per-repo Forge observations before starting subprocesses.
+The existing-worker candidate configures this gate; missing policy or receipts
+refuse execution before subprocess. Native freshness noops rely on protected
+Repo Manager issuer custody: the issuer checks HEAD against its last successful
+commit before emitting `forge_head_matches_last_success`.

docs/custodian-runtime-binding-proposal.md

@@ -1,5 +1,11 @@
 # Custodian runtime binding proposal
 
+Update: user selected the existing worker. The concrete pinned image package,
+private state/fleet mount candidate and tunable policy now live in
+[`k8s/custodian-release/README.md`](../k8s/custodian-release/README.md).
+That packet supersedes the placement alternatives below. It remains outside
+active kustomization; schedules are disabled.
+
 Review artifact, 2026-10-19. Not a deployment or activation approval. Both native
 ActivityDefinitions remain `enabled: false`. No manifests, images or schedules
 were changed for this proposal. Read with `custodian-bounded-operations.md`.

k8s/custodian-release/README.md

@@ -0,0 +1,90 @@
+# Existing-worker Custodian delivery candidate
+
+User selected the existing Activity Core worker. This directory prepares that
+choice; it is not referenced by active `k8s/gitops/kustomization.yaml`. Neither
+schedule is enabled. The patch adds no CPU/memory request or separate process.
+The source Dockerfile now embeds the four fixed owner scripts and checks their
+actual bytes during build. Source pin is in `commands/manifest.json`; do not
+infer deployed bytes from Git HEAD. A new image digest is required.
+
+`worker-patch.yaml` is a strategic-merge review artifact preserving existing env,
+volumes and resources. It reuses `actcore-ops-artefacts` with two narrowly scoped
+subPaths: private state and a selected immutable snapshot. Explicit image and
+snapshot placeholders must be replaced with real accepted evidence. Do not apply
+this file directly. Review its merged diff and render through owner GitOps after
+source/image acceptance. No Kubernetes resources were read or changed here.
+
+`input-policy.json` holds tunable defaults: 2-hour Forge observation age, 5-minute
+clock tolerance, exact 15-repo set, 64MiB snapshot bound, recommended 35-day daily
+and 13-bundle weekly retention. Retention settings are policy for owner review,
+not an installed pruning job. Referenced evidence must survive pruning. Consumer
+admission enforces observation age, clock tolerance, scope, hashes and size;
+it ignores the exporter's unverified observation claim and instead requires
+independent existing Repo Manager receipts for each pinned commit. A recent
+`forge_head_matches_last_success` noop is valid re-observation; an old commit can
+remain current. Applied/reconciled rows must bind expected/derived/projected
+commits. Receipt files are trusted protected owner evidence, not signed proofs.
+
+## Concrete publisher contract
+
+Repo Manager commit `3e2726f` adds offline `python -m repo_manager.fleet_snapshot`;
+its `docs/custodian-fleet-input-contract.md` defines input request, publication and
+recommended ownership. It does not fetch objects or establish remote freshness.
+Export selected verified Forge commits from the existing approved repository
+object stores, retain the complete matching receipt rows as a JSON array, and
+copy the completed immutable tree to `custodian/snapshots/<id>` on the claim.
+Only workplans/decisions are admitted. Do not fetch credentials into the worker.
+
+The candidate uses a fixed snapshot mount deliberately. It can prove one bounded
+run after acceptance; it is not a recurring publication handoff. Before enabling
+the daily schedule, add the owned pre-run publication/handoff within native
+Activity Core orchestration, or a reviewed serialized mount update with consumers
+quiescent. A static stale snapshot fails the freshness gate; never refresh its
+timestamp. The worker image's packaged Repo Manager version is still the earlier
+freshness CLI pin; the exporter is a publisher-side prerequisite, not falsely
+claimed installed there.
+
+## Required preflight and verification
+
+1. Review/build the new Activity Core image and verify actual installed package
+   and scripts in it. Run `python /app/scripts/verify_custodian_bundle.py --root
+   /opt/custodian-commands --expected-revision
+   60d893af1bf77f457504b64ff1c9ffb5c6ae6579`. Expected manifest SHA-256:
+   `8fddc3e90fe90992de2369cb4354ee43d9d25352e958ee371a702dc974f1df1f`.
+   Inspect the installed `activity_core.custodian_inputs` and operations module
+   hashes against this reviewed build too; old editable installs are insufficient.
+2. Verify current shared-claim capacity, backups and ownership before populating
+   any input/state. Precreate `custodian/state` and required children owned by
+   UID1000, mode0700; never change the entire claim. Measure actual snapshot
+   footprint against retention before accepting this existing claim. Insufficient
+   capacity is a source/storage prerequisite; do not silently increase spending.
+3. Freeze a complete snapshot plus its independent fresh receipt JSON and the
+   weekly owner request/real collector inputs. Mount the snapshot read-only;
+   protect state inputs from concurrent writers. Verify every manifest hash and
+   owner receipt at runtime. Owner-approved source publication expands readable
+   work-record scope to the explicit repos only; it grants no secret/credential
+   access and should be reviewed as that exact data-access delta.
+4. Confirm existing direct Hub admission for decision GET/resolve, message POST
+   and progress sink. Source URL alone proves neither authorization nor delivery.
+   Keep configured recipients and owner accountability. No edge-relay queued
+   result can substitute for a central message ID.
+5. Review resulting Deployment diff, current CPU/headroom and rollout policy.
+   No added requests are proposed. Existing rolling surge still needs current
+   headroom; verify available capacity rather than assuming this source's empty
+   requests justify extra capacity. Keep schedules disabled/paused throughout.
+6. Offline fixture tests precede any separately reviewed bounded real run. Verify
+   missing/stale snapshot rejection before subprocess and collect operation,
+   report, sink and owner acknowledgment evidence. First unattended proof remains
+   an acceptance task, not a source-test result.
+
+Existing standing authorization can cover non-destructive owner deployment when
+the reviewed diff, dry-run, rollback, abort owner and receipt are concrete. This
+packet does not invent a blanket new founder gate. Changes to access scope,
+credentials, public publication or spending limits retain their own authority
+requirements; root reviews this exact candidate before deployment.
+
+Rollback restores prior image and mount/env source through GitOps, keeps schedules
+paused, and preserves state/checkpoints/old bundles and snapshots. Daily Hub writes
+cannot be undone by rollback. Inspect delivery checkpoints before retries. The
+Activity Core runtime owner is the proposed abort/recovery owner; identify the
+actual operator in the deployment receipt before applying.

k8s/custodian-release/commands/manifest.json

@@ -0,0 +1,10 @@
+{
+  "schema": "activity-core.custodian-command-bundle.v1",
+  "source_revision": "60d893af1bf77f457504b64ff1c9ffb5c6ae6579",
+  "files": {
+    "scripts/security_flow_daily.py": "4a4e84702b3fb38d086695bb6d601430d8914bd79286b4f683794206f110c353",
+    "scripts/security_flow_report.py": "eac9fdcc56d7f3ff613e6848c7523fb9a421d82a3b564a51c36f8c2b842eef02",
+    "scripts/sync_resolved_decisions.py": "182433ca7cd31e2b2773246ec1be02569ba6f089ac009add6543e5e48777b5ec",
+    "scripts/weekly_allocation_report.py": "93f546243331308d7e60f8db36e988d9847537a2fe46cce49ee2fffba1f7f247"
+  }
+}

k8s/custodian-release/commands/scripts/security_flow_daily.py

@@ -0,0 +1,79 @@
+#!/usr/bin/env python3
+"""Daily decision sync and security report, with per-recipient delivery checkpoints.
+
+Dry-run by default. --apply is the scheduled mode. No credentials are stored.
+"""
+import argparse
+import fcntl
+import json
+from pathlib import Path
+import subprocess
+import sys
+from urllib.request import Request, urlopen
+
+from security_flow_report import DEFAULT_REPOS
+
+SCRIPTS = Path(__file__).resolve().parent
+
+
+def notifications(report, state, agent, founder):
+    pending = []
+    ready = {key: report[key] for key in ("READY", "UNBLOCKED")}
+    if ready != state.get("agent"):
+        pending.append(("agent", agent, ready))
+    # ATTENDED is the daily operator reminder; empty lists never message the founder.
+    if report["ATTENDED"]:
+        pending.append(("founder", founder, {"ATTENDED": report["ATTENDED"]}))
+    return pending
+
+
+def post(hub, recipient, payload):
+    body = {"from_agent": "the-custodian", "to_agent": recipient,
+            "subject": "Daily security flow", "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 save(path, data):
+    temporary = path.with_suffix(path.suffix + ".tmp")
+    temporary.write_text(json.dumps(data, indent=2) + "\n")
+    temporary.replace(path)
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--root", default=str(Path.home()))
+    parser.add_argument("--repos", default=",".join(dict.fromkeys(["the-custodian", *DEFAULT_REPOS])))
+    parser.add_argument("--hub", default="http://127.0.0.1:8000")
+    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("--apply", action="store_true")
+    args = parser.parse_args()
+    args.state_dir.mkdir(parents=True, exist_ok=True)
+    with (args.state_dir / "run.lock").open("w") as lock:
+        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        command = [sys.executable, str(SCRIPTS / "sync_resolved_decisions.py"),
+                   "--root", args.root, "--hub", args.hub]
+        subprocess.run(command + (["--apply"] if args.apply else []), check=True)
+        raw = subprocess.check_output([sys.executable, str(SCRIPTS / "security_flow_report.py"),
+            "--root", args.root, "--repos", args.repos, "--json"], text=True)
+        report = json.loads(raw)
+        state_path = args.state_dir / "delivery.json"
+        state = json.loads(state_path.read_text()) if state_path.exists() else {}
+        save(args.state_dir / "report.json", report)
+        for key, recipient, payload in notifications(report, state, args.agent, args.founder):
+            if not args.apply:
+                print(json.dumps({"dry_run": True, "to_agent": recipient, "report": payload}))
+                continue
+            message_id = post(args.hub, recipient, payload)
+            state[key] = payload
+            state[key + "_message_id"] = message_id
+            save(state_path, state)  # A later recipient failure does not replay earlier sends.
+        print(raw)
+
+
+if __name__ == "__main__":
+    main()

k8s/custodian-release/commands/scripts/security_flow_report.py

@@ -0,0 +1,116 @@
+#!/usr/bin/env python3
+"""Deterministic flow report for the NetKingdom security estate.
+
+Replaces agent sessions that poll wait tasks. Reads workplan files only (no hub,
+no network) and prints five lists:
+
+  READY       todo/progress tasks with no unresolved explicit dependency or owner/human gate
+  UNBLOCKED   wait tasks whose every depends_on target is done/cancel: requalify to todo
+  ATTENDED    needs_human wait tasks: the attended-window checklist for the operator/founder
+  STALLED     wait tasks qualified only by blocked_on message-from (owner reply), oldest first by file
+  DEPENDENT   todo/progress tasks that still carry an unresolved dependency or owner gate
+
+Usage: security_flow_report.py [--repos a,b,c] [--json]
+"""
+import argparse, glob, json, os, re
+
+DEFAULT_REPOS = ("the-custodian net-kingdom railiance-platform activity-core access-engine secrets-engine ops-mason "
+                 "ops-warden key-cape user-engine tenant-engine gate-house kings-guard "
+                 "policy-nexus rapp-policy-nexus").split()
+TERMINAL = {"done", "cancel"}
+
+
+def field(block, key):
+    m = re.search(rf"^{key}:\s*(.*)$", block, re.M)
+    return m.group(1).strip().strip('"') if m else ""
+
+
+def dependencies(block):
+    # Accept the convention's inline lists and YAML's indented multiline lists.
+    m = re.search(r"^depends_on:[ \t]*(.*(?:\n[ \t]+[^\n]*)*)", block, re.M)
+    return re.findall(r"[A-Z][A-Z0-9-]+-T\d+|[A-Z][A-Z0-9-]+-WP-\d+", m[1]) if m else []
+
+
+def load(root, repos):
+    tasks, wps = {}, {}
+    for r in repos:
+        for f in glob.glob(os.path.join(root, r, "workplans", "**", "*.md"), recursive=True):
+            text = open(f, encoding="utf-8", errors="ignore").read()
+            fm = text.split("---")[1] if text.startswith("---") else ""
+            wp = {"repo": r, "file": os.path.relpath(f, root),
+                  "status": field(fm, "status"), "flavor": field(fm, "flavor"),
+                  "title": field(fm, "title")}
+            for b in re.findall(r"```task\n(.*?)```", text, re.S):
+                tid = field(b, "id")
+                if not tid:
+                    continue
+                deps = list(dict.fromkeys(dependencies(fm) + dependencies(b)))
+                tasks[tid] = {"id": tid, "status": field(b, "status"), "deps": deps,
+                              "human": field(b, "needs_human") == "true",
+                              "blocked_on": field(b, "blocked_on"),
+                              "reason": field(b, "blocking_reason") or field(b, "intervention_note"),
+                              "priority": field(b, "priority"), "wp": wp}
+    return tasks
+
+
+def classify(tasks):
+    def dep_done(d):
+        if d in tasks:
+            return tasks[d]["status"] in TERMINAL
+        # workplan-level dependency: every known task of that workplan is terminal
+        own = [t for t in tasks.values() if t["id"].startswith(d + "-T")]
+        return bool(own) and all(t["status"] in TERMINAL for t in own)
+
+    out = {"READY": [], "UNBLOCKED": [], "ATTENDED": [], "STALLED": [], "DEPENDENT": []}
+    for t in tasks.values():
+        wp = t["wp"]
+        if wp["status"] in ("finished", "archived") or wp["flavor"] == "residual":
+            continue
+        missing = [d for d in t["deps"] if not dep_done(d)]
+        if t["status"] in ("todo", "progress"):
+            if t["human"]:
+                out["ATTENDED"].append(t | {"pending_dependencies": missing})
+            elif missing or t["blocked_on"]:
+                out["DEPENDENT"].append(t | {"pending_dependencies": missing})
+            else:
+                out["READY"].append(t)
+        elif t["status"] == "wait":
+            if t["human"]:
+                out["ATTENDED"].append(t | {"pending_dependencies": missing})
+            elif t["deps"] and all(dep_done(d) for d in t["deps"]):
+                out["UNBLOCKED"].append(t)
+            elif not t["deps"] and t["blocked_on"].startswith("message-from"):
+                out["STALLED"].append(t)
+    prio = {"high": 0, "medium": 1, "low": 2}
+    for k in out:
+        out[k].sort(key=lambda t: (prio.get(t["priority"], 3), t["id"]))
+    return out
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--root", default=os.path.expanduser("~"))
+    ap.add_argument("--repos", default=",".join(DEFAULT_REPOS))
+    ap.add_argument("--json", action="store_true")
+    a = ap.parse_args()
+    res = classify(load(a.root, a.repos.split(",")))
+    if a.json:
+        print(json.dumps({k: [{x: t[x] for x in ("id", "status", "priority", "reason", "blocked_on")}
+                              | {"repo": t["wp"]["repo"]}
+                              | ({"pending_dependencies": t["pending_dependencies"]} if "pending_dependencies" in t else {})
+                              for t in v] for k, v in res.items()}, indent=1))
+        return
+    for k, v in res.items():
+        print(f"\n## {k} ({len(v)})")
+        for t in v:
+            extra = t["blocked_on"] if k == "STALLED" else t["reason"]
+            if k == "DEPENDENT":
+                extra = ", ".join(t["pending_dependencies"]) or t["blocked_on"]
+            elif k == "ATTENDED" and t["pending_dependencies"]:
+                extra = "After " + ", ".join(t["pending_dependencies"]) + ": " + extra
+            print(f"- {t['id']} [{t['status']}/{t['priority'] or '-'}] {t['wp']['repo']}"
+                  + (f" — {extra[:140]}" if extra else ""))
+
+
+if __name__ == "__main__":
+    main()

k8s/custodian-release/commands/scripts/sync_resolved_decisions.py

@@ -0,0 +1,58 @@
+#!/usr/bin/env python3
+"""Resolve open hub decisions that a repo decision log already records as resolved.
+
+Repo files are authoritative (ADR-001); the hub is a read model. This scans
+~/*/decisions/*.md for fenced yaml blocks with `id`, `title`, `status: resolved`
+and resolves the open hub decision whose title matches exactly (whitespace-
+normalized, case-insensitive). Uses the sanctioned POST /decisions/{id}/resolve.
+
+Usage: sync_resolved_decisions.py [--apply] [--hub http://127.0.0.1:8000]
+"""
+import argparse, glob, json, os, re, urllib.request
+
+
+def norm(s):
+    return re.sub(r"\s+", " ", s.strip().strip("'\"")).lower()
+
+
+def repo_resolved(root):
+    out = {}
+    for f in glob.glob(os.path.join(root, "*", "decisions", "*.md")):
+        text = open(f, encoding="utf-8", errors="ignore").read()
+        for b in re.findall(r"```yaml\n(.*?)```", text, re.S):
+            g = lambda k: (re.search(rf"^{k}:\s*(.*)$", b, re.M) or [None, ""])[1]
+            if g("status").strip() in ("resolved", "accepted") and g("title"):
+                out[norm(g("title"))] = (g("id").strip(), os.path.relpath(f, root))
+    return out
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--hub", default="http://127.0.0.1:8000")
+    ap.add_argument("--root", default=os.path.expanduser("~"))
+    ap.add_argument("--apply", action="store_true")
+    a = ap.parse_args()
+    known = repo_resolved(a.root)
+    data = json.load(urllib.request.urlopen(f"{a.hub}/decisions/?status=open&limit=500"))
+    data = data.get("items", data) if isinstance(data, dict) else data
+    hits = 0
+    for d in data:
+        t = norm(d["title"])
+        m = known.get(t) or next((v for k, v in known.items()
+                                  if min(len(k), len(t)) >= 40 and (k.startswith(t) or t.startswith(k))), None)
+        if not m:
+            continue
+        hits += 1
+        rid, path = m
+        print(f"{d['id'][:8]}  {rid}  {path}  | {d['title'][:70]}")
+        if a.apply:
+            body = json.dumps({"rationale": f"Resolved in repo decision log {path} ({rid}); hub synced to repo state.",
+                               "decided_by": "bernd"}).encode()
+            req = urllib.request.Request(f"{a.hub}/decisions/{d['id']}/resolve", data=body,
+                                         headers={"Content-Type": "application/json"}, method="POST")
+            urllib.request.urlopen(req).read()
+    print(f"{hits} open hub decisions match resolved repo records" + ("" if a.apply else " (dry run)"))
+
+
+if __name__ == "__main__":
+    main()

k8s/custodian-release/commands/scripts/weekly_allocation_report.py

@@ -0,0 +1,321 @@
+#!/usr/bin/env python3
+"""Retain CUST-WP-0071 sizing evidence; no network, resizing or task emission."""
+from __future__ import annotations
+
+import argparse
+from datetime import datetime, timedelta, timezone
+import hashlib
+import json
+import math
+import os
+from pathlib import Path
+import re
+import tempfile
+
+SCHEMA = "custodian.weekly-allocation.v1"
+METRICS = (
+    "cpu_p95_m", "cpu_peak_m", "memory_p95_bytes", "memory_peak_bytes",
+    "cpu_throttle_ratio", "latency_p95_seconds", "errors", "request_count",
+    "restarts", "oom_events", "cpu_sample_count", "memory_sample_count",
+)
+REQUIRED = tuple(m for m in METRICS if not m.endswith("sample_count"))
+
+
+def timestamp(value):
+    result = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    if result.tzinfo is None:
+        raise ValueError("timestamps must include timezone")
+    return result.astimezone(timezone.utc)
+
+
+def duration(value):
+    match = re.fullmatch(r"([1-9][0-9]*)([smhdw])", value)
+    if not match:
+        raise ValueError(f"unsupported duration: {value}")
+    return int(match[1]) * {"s": 1, "m": 60, "h": 3600, "d": 86400, "w": 604800}[match[2]]
+
+
+def digest(value):
+    return hashlib.sha256(json.dumps(value, sort_keys=True, allow_nan=False).encode()).hexdigest()
+
+
+def finite(value):
+    number = float(value)
+    if not math.isfinite(number) or number < 0:
+        raise ValueError("metric must be finite and nonnegative")
+    return number
+
+
+def metric_series(telemetry):
+    """Do not distribute namespace aggregates over workloads, or sum percentiles."""
+    series, queries = {}, {}
+    seconds = duration(telemetry["window"])
+    for name, query in sorted(telemetry.get("queries", {}).items()):
+        response = query.get("response", {})
+        data = response.get("data", {})
+        queries[name] = {
+            "promql": query.get("promql"), "response_status": response.get("status"),
+            "result_type": data.get("resultType"), "warnings": response.get("warnings", []),
+            "error": response.get("error"), "response_sha256": digest(response),
+            "evaluation_times": [],
+        }
+        if name not in METRICS or response.get("status") != "success" or data.get("resultType") != "vector":
+            continue
+        for sample in data.get("result", []):
+            labels = sample.get("metric", {})
+            namespace, workload = labels.get("namespace"), labels.get("workload")
+            # Additional dimensions are ambiguous: do not silently collapse pod/job series.
+            if not namespace or set(labels) - {"namespace", "workload", "__name__"}:
+                queries[name].setdefault("rejected", []).append("unaggregated or absent identity labels")
+                continue
+            key = (namespace, workload)
+            values = series.setdefault(key, {})
+            if name in values:
+                values[name] = {"value": None, "reason": "ambiguous duplicate series"}
+                continue
+            try:
+                epoch, raw = sample["value"]
+                end = datetime.fromtimestamp(finite(epoch), timezone.utc)
+                number = finite(raw)
+            except (KeyError, ValueError, TypeError, OverflowError):
+                values[name] = {"value": None, "reason": "invalid sample"}
+                continue
+            queries[name]["evaluation_times"].append(end.isoformat())
+            values[name] = {
+                "value": number, "window_start": (end - timedelta(seconds=seconds)).isoformat(),
+                "window_end": end.isoformat(), "query": name,
+                "capture_alignment_seconds": (end - timestamp(telemetry["captured_at"])).total_seconds(),
+            }
+    return series, queries
+
+
+def recommendation_id(resource, namespace, workload):
+    return "allocation-" + digest([resource, namespace, workload, "sizing-review"])[:24]
+
+
+def build_report(cluster, allocation, telemetry, inventory, provenance, *, as_of,
+                 previous=None, reviews=None, max_age_hours=48):
+    now = timestamp(as_of)
+    if cluster.get("schema_version") != "railiance.cluster-resource-observation.v1":
+        raise ValueError("unsupported cluster schema")
+    if allocation.get("schema") != "custodian.allocation-reconcile.v1":
+        raise ValueError("unsupported allocation schema")
+    resource = cluster["resource_id"]
+    if inventory.get("id") != resource or allocation.get("host", {}).get("cluster_resource_id") != resource:
+        raise ValueError("resource identity mismatch")
+    if allocation.get("count_host_and_cluster_cpus_once") is not True:
+        raise ValueError("allocation must deduplicate host and cluster CPUs")
+    if cluster["captured_at"] != allocation["captured_at"]:
+        raise ValueError("cluster and reconcile snapshots differ")
+    if provenance.get("source_observation") != cluster["captured_at"]:
+        raise ValueError("provenance does not identify cluster snapshot")
+    seconds = duration(telemetry["window"])
+    if seconds != 604800:
+        raise ValueError("weekly report requires a seven-day telemetry window")
+    step = duration(telemetry["step"])
+    series, queries = metric_series(telemetry)
+    captured = {"cluster": cluster["captured_at"], "telemetry": telemetry["captured_at"]}
+    freshness = {}
+    for name, value in captured.items():
+        age = (now - timestamp(value)).total_seconds()
+        freshness[name] = {"age_seconds": age, "status": "future" if age < 0 else "stale" if age > max_age_hours * 3600 else "fresh"}
+    query_signature = digest({name: q["promql"] for name, q in queries.items()})
+    end = timestamp(telemetry["captured_at"])
+    comparison_reason = "no previous retained report"
+    compatible = False
+    if previous:
+        prev_end = timestamp(previous["period"]["capture_end"])
+        compatible = (
+            previous.get("schema") == SCHEMA and previous.get("resource_id") == resource
+            and previous["period"].get("seconds") == seconds
+            and previous["period"].get("step_seconds") == step
+            and previous.get("query_signature") == query_signature
+            and abs((end - prev_end).total_seconds() - seconds) <= step
+        )
+        comparison_reason = "comparable consecutive capture windows" if compatible else "nonconsecutive windows or changed query/resource/step contract"
+    previous_rows = {(r["namespace"], r["workload"]): r for r in (previous or {}).get("workloads", [])}
+    owners = {(r["namespace"], r["workload"]): r for r in allocation.get("workloads", [])}
+    if len(owners) != len(allocation.get("workloads", [])):
+        raise ValueError("duplicate allocation workload identity")
+    reviews = reviews or {}
+    rows, recommendations = [], []
+    seen = set()
+    for workload in sorted(cluster.get("utilization", {}).get("workloads", []), key=lambda x: (x["namespace"], x["workload"])):
+        key = workload["namespace"], workload["workload"]
+        if key in seen:
+            raise ValueError("duplicate cluster workload identity")
+        seen.add(key)
+        owner = owners.get(key, {})
+        measurements = {m: series.get(key, {}).get(m, {"value": None, "reason": "no workload series"}) for m in METRICS}
+        gaps = [m for m in REQUIRED if measurements[m]["value"] is None]
+        metric_stale = [m for m, v in measurements.items() if v["value"] is not None and not 0 <= (now - timestamp(v["window_end"])).total_seconds() <= max_age_hours * 3600]
+        metric_unaligned = [m for m, v in measurements.items() if v["value"] is not None and abs(v["capture_alignment_seconds"]) > step]
+        rid = recommendation_id(resource, *key)
+        reasons = []
+        if gaps:
+            reasons.append("missing workload metrics: " + ", ".join(gaps))
+        if metric_stale or any(v["status"] != "fresh" for v in freshness.values()):
+            reasons.append("stale or future observation")
+        if metric_unaligned:
+            reasons.append("metric evaluation windows differ from capture by more than one step")
+        if owner.get("owner", "unknown") == "unknown":
+            reasons.append("unknown owner")
+        if workload.get("requests", {}).get("cpu_millicores") == 0:
+            reasons.append("zero CPU request requires owner review")
+        review = reviews.get(rid)
+        decision, confidence, evidence = "investigate", "low", []
+        if review:
+            if review.get("decision") not in {"keep", "increase", "decrease", "investigate"}:
+                raise ValueError("invalid owner review decision")
+            required = ("owner", "evidence_refs", "receiving_task", "configuration", "rollback_ref", "reviewed_at", "period_end", "coverage_ref", "service_acceptance_ref", "margin_ref")
+            if any(not review.get(k) for k in required):
+                raise ValueError("owner review missing evidence, receiving task or configuration/rollback")
+            if review["owner"] != owner.get("owner") or review["period_end"] != telemetry["captured_at"]:
+                reasons.append("owner review does not match this owner/window")
+            elif timestamp(review["reviewed_at"]) > now:
+                reasons.append("owner review is future-dated")
+            elif not reasons:
+                decision, confidence, evidence = review["decision"], "owner-reviewed", review["evidence_refs"]
+            else:
+                reasons.append("owner proposal retained; incomplete evidence prevents accepted sizing recommendation")
+        else:
+            reasons.append("no owner-reviewed sizing decision")
+        row = {
+            "namespace": key[0], "workload": key[1], "owner": owner.get("owner", "unknown"),
+            "service": owner.get("service", "unknown"), "tenant": owner.get("tenant", "unknown"),
+            "pods": workload.get("pods"), "requests": workload.get("requests"),
+            "limits": workload.get("limits"), "instantaneous_observed": workload.get("observed"),
+            "metrics": measurements, "missing_metrics": gaps,
+            "namespace_context_ref": key[0] if (key[0], None) in series else None,
+            "raw_scrape_coverage": "unknown; evaluation counts do not prove source coverage",
+            "recommendation_id": rid, "week_on_week": {}, "comparison_reasons": {},
+        }
+        prior = previous_rows.get(key)
+        for metric in METRICS:
+            current = measurements[metric]["value"]
+            old_measurement = (prior or {}).get("metrics", {}).get(metric, {})
+            old = old_measurement.get("value")
+            metric_compatible = False
+            reason = comparison_reason
+            if compatible and current is not None and old is not None:
+                try:
+                    old_end, old_start = timestamp(old_measurement["window_end"]), timestamp(old_measurement["window_start"])
+                    new_end, new_start = timestamp(measurements[metric]["window_end"]), timestamp(measurements[metric]["window_start"])
+                    metric_compatible = (
+                        (old_end - old_start).total_seconds() == seconds
+                        and (new_end - new_start).total_seconds() == seconds
+                        and abs((new_end - old_end).total_seconds() - seconds) <= step
+                        and abs((new_start - old_end).total_seconds()) <= step
+                        and abs(measurements[metric]["capture_alignment_seconds"]) <= step
+                        and abs((old_end - timestamp(previous["period"]["capture_end"])).total_seconds()) <= step
+                    )
+                    reason = "comparable consecutive metric windows" if metric_compatible else "metric evaluation windows are not consecutive or capture-aligned"
+                except (KeyError, ValueError, TypeError):
+                    reason = "missing or invalid previous metric window"
+            elif compatible:
+                reason = "missing current or previous metric"
+            row["week_on_week"][metric] = current - old if metric_compatible else None
+            row["comparison_reasons"][metric] = reason
+        rows.append(row)
+        prior_recommendation = next((r for r in (previous or {}).get("recommendations", []) if r["id"] == rid), {})
+        recommendations.append({
+            "id": rid, "decision": decision, "confidence": confidence, "owner": row["owner"],
+            "reasons": reasons, "evidence_refs": evidence, "owner_proposal": review,
+            "receiving_task": review.get("receiving_task") if review else prior_recommendation.get("receiving_task"),
+            "previous_decision": prior_recommendation.get("decision"),
+            "owner_outcome": review.get("outcome") if review else None,
+            "delivery": "not emitted", "acknowledgment": None,
+        })
+    if seen != set(owners):
+        raise ValueError("cluster and allocation workload inventories differ")
+    return {
+        "schema": SCHEMA, "workplan": "CUST-WP-0071-T05", "as_of": as_of,
+        "adapter_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
+        "resource_id": resource, "period": {"capture_end": telemetry["captured_at"], "seconds": seconds, "step_seconds": step,
+            "note": "Actual per-query windows are retained on samples; capture time is not a common query evaluation time."},
+        "freshness": freshness, "max_age_hours": max_age_hours, "query_signature": query_signature,
+        "queries": queries, "collector_provenance": provenance,
+        "source_sha256": {"cluster": digest(cluster), "allocation": digest(allocation), "telemetry": digest(telemetry), "inventory": digest(inventory)},
+        "node": {"capacity": cluster.get("capacity"), "allocation": cluster.get("allocation"),
+            "scheduled_request_cpu_m": allocation.get("scheduled_request_cpu_m"),
+            "pending_unscheduled_cpu_m": allocation.get("pending_unscheduled_cpu_m"),
+            "reservation_residual_cpu_m": allocation.get("residual_cpu_m"),
+            "release_preflight": allocation.get("state_hub_preflight_observation"),
+            "operating_margin": None, "release_margin_acceptance": "not established by reservation residual",
+            "host_cluster_counted_once": True},
+        "portfolio": {k: inventory.get(k) for k in ("id", "schema_version", "ownership", "capacity", "relationships", "evidence")},
+        "workloads": rows,
+        "namespace_context": {ns: values for (ns, workload), values in sorted(series.items(), key=lambda x: str(x[0])) if workload is None},
+        "comparison": {"compatible": compatible, "reason": comparison_reason,
+            "previous_report_sha256": digest(previous) if previous else None,
+            "disappeared_workloads": [list(key) for key in sorted(set(previous_rows) - seen)]},
+        "measurement_gaps": cluster.get("measurement_gaps", []) + allocation.get("measurement_gaps", []),
+        "recommendations": recommendations,
+        "limitations": ["No representative pilot acceptance inferred from telemetry.",
+            "Namespace percentiles and peaks are not workload metrics or additive node peaks.",
+            "Portfolio capacity is dated evidence, not additional host capacity.",
+            "No procurement, sizing changes, owner-task creation, delivery or acknowledgment performed.",
+            "Installed executor, scoped reads, independent missed-run handling and first unattended receipt remain T05 gates."],
+    }
+
+
+def retain(path, report):
+    """Never overwrite a retained week; identical retries succeed."""
+    encoded = json.dumps(report, indent=2, sort_keys=True, allow_nan=False) + "\n"
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = None
+    try:
+        with tempfile.NamedTemporaryFile(mode="w", dir=path.parent, delete=False) as stream:
+            temporary = Path(stream.name)
+            stream.write(encoded)
+            stream.flush()
+            os.fsync(stream.fileno())
+        try:
+            os.link(temporary, path)
+        except FileExistsError:
+            if path.is_symlink() or path.read_text() != encoded:
+                raise ValueError("retained report differs; choose a new explicit revision path")
+    finally:
+        if temporary is not None:
+            temporary.unlink(missing_ok=True)
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--request", type=Path, help="Executor-owned JSON request; paths relative to its directory")
+    for name in ("cluster", "allocation", "telemetry", "inventory", "provenance"):
+        parser.add_argument("--" + name, type=Path)
+    parser.add_argument("--previous", type=Path)
+    parser.add_argument("--reviews", type=Path)
+    parser.add_argument("--as-of")
+    parser.add_argument("--max-age-hours", type=float, default=48)
+    parser.add_argument("--output", type=Path)
+    args = parser.parse_args()
+    try:
+        if args.request:
+            request = json.loads(args.request.read_text())
+            if request.get("schema") != "custodian.weekly-allocation-request.v1":
+                raise ValueError("unsupported executor request schema")
+            if any(getattr(args, key) for key in ("cluster", "allocation", "telemetry", "inventory", "provenance", "previous", "reviews", "as_of", "output")):
+                raise ValueError("request cannot be mixed with input/output overrides")
+            for key in ("cluster", "allocation", "telemetry", "inventory", "provenance", "previous", "reviews", "output"):
+                value = request.get(key)
+                setattr(args, key, args.request.parent / value if value else None)
+            args.as_of = request.get("as_of")
+        if any(not getattr(args, key) for key in ("cluster", "allocation", "telemetry", "inventory", "provenance", "as_of", "output")):
+            raise ValueError("executor input paths, as_of and retention output must be explicitly bound")
+        if not math.isfinite(args.max_age_hours) or args.max_age_hours <= 0:
+            raise ValueError("max-age-hours must be positive and finite")
+        inputs = {name: json.loads(getattr(args, name).read_text()) for name in ("cluster", "allocation", "telemetry", "inventory", "provenance")}
+        report = build_report(**inputs, as_of=args.as_of, max_age_hours=args.max_age_hours,
+            previous=json.loads(args.previous.read_text()) if args.previous else None,
+            reviews=json.loads(args.reviews.read_text()) if args.reviews else None)
+        retain(args.output, report)
+    except (OSError, ValueError, KeyError, TypeError) as exc:
+        parser.exit(2, f"weekly allocation report refused: {exc}\n")
+    print(json.dumps({"status": "retained", "path": str(args.output), "sha256": digest(report),
+                      "workloads": len(report["workloads"]), "delivery": "not emitted"}))
+
+
+if __name__ == "__main__":
+    main()

k8s/custodian-release/input-policy.json

@@ -0,0 +1,29 @@
+{
+  "schema": "activity-core.custodian-input-policy.v1",
+  "publisher_owner": "repo-manager",
+  "consumer_owner": "activity-core",
+  "max_observation_age_seconds": 7200,
+  "max_future_skew_seconds": 300,
+  "daily_retention_days": 35,
+  "weekly_retention_count": 13,
+  "preserve_referenced_evidence": true,
+  "max_snapshot_bytes": 67108864,
+  "repositories": [
+    "the-custodian",
+    "net-kingdom",
+    "railiance-platform",
+    "activity-core",
+    "access-engine",
+    "secrets-engine",
+    "ops-mason",
+    "ops-warden",
+    "key-cape",
+    "user-engine",
+    "tenant-engine",
+    "gate-house",
+    "kings-guard",
+    "policy-nexus",
+    "rapp-policy-nexus"
+  ],
+  "expected_hub_instance": "railiance01"
+}

k8s/custodian-release/worker-patch.yaml

@@ -0,0 +1,40 @@
+# REVIEW ONLY: not referenced by k8s/gitops/kustomization.yaml.
+# Replace both explicit placeholders only after image/input evidence is accepted.
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: actcore-worker
+  namespace: activity-core
+spec:
+  template:
+    spec:
+      containers:
+        - name: worker
+          image: forgejo.coulomb.social/coulomb/activity-core@sha256:REQUIRES_REVIEWED_IMAGE_DIGEST
+          env:
+            - name: ACTIVITY_CORE_CUSTODIAN_ROOT
+              value: /opt/custodian-commands
+            - name: ACTIVITY_CORE_CUSTODIAN_FLEET_ROOT
+              value: /var/lib/custodian-inputs/fleet
+            - name: ACTIVITY_CORE_CUSTODIAN_STATE_ROOT
+              value: /var/lib/activity-core/custodian
+            - name: ACTIVITY_CORE_CUSTODIAN_REPOS
+              value: the-custodian,net-kingdom,railiance-platform,activity-core,access-engine,secrets-engine,ops-mason,ops-warden,key-cape,user-engine,tenant-engine,gate-house,kings-guard,policy-nexus,rapp-policy-nexus
+            - name: ACTIVITY_CORE_CUSTODIAN_HUB
+              value: http://state-hub.state-hub.svc.cluster.local:8000
+            - name: ACTIVITY_CORE_CUSTODIAN_INPUT_POLICY
+              value: /opt/custodian-input-policy.json
+            - name: ACTIVITY_CORE_CUSTODIAN_FRESHNESS_RECEIPTS
+              value: /var/lib/activity-core/custodian/inputs/freshness-receipts.json
+            - name: ACTIVITY_CORE_CUSTODIAN_ALLOCATION_REQUEST
+              value: /var/lib/activity-core/custodian/requests/current.json
+          volumeMounts:
+            - name: ops-artefacts
+              mountPath: /var/lib/activity-core/custodian
+              subPath: custodian/state
+            - name: ops-artefacts
+              mountPath: /var/lib/custodian-inputs/fleet
+              subPath: custodian/snapshots/REQUIRES_ACCEPTED_SNAPSHOT_ID
+              readOnly: true
+# Existing ops-artefacts volume/claim, UID1000 and resources are preserved.
+# Owner precreates only these private subdirectories; no whole-claim chmod/chown.

scripts/verify_custodian_bundle.py

@@ -0,0 +1,34 @@
+#!/usr/bin/env python3
+"""Verify actual installed bytes of the fixed Custodian command bundle; no execution."""
+import argparse
+import hashlib
+import json
+from pathlib import Path
+
+FILES = {"scripts/security_flow_daily.py", "scripts/security_flow_report.py",
+         "scripts/sync_resolved_decisions.py", "scripts/weekly_allocation_report.py"}
+
+
+def verify(root, expected_revision=None):
+    root = Path(root)
+    manifest = json.loads((root / "manifest.json").read_text())
+    if manifest.get("schema") != "activity-core.custodian-command-bundle.v1" or set(manifest.get("files", {})) != FILES:
+        raise ValueError("unexpected bundle contract")
+    if expected_revision and manifest.get("source_revision") != expected_revision:
+        raise ValueError("unexpected source revision")
+    for relative, expected in manifest["files"].items():
+        path = root / relative
+        if any(p.is_symlink() for p in [path, *path.parents]) or not path.is_file():
+            raise ValueError("bundle path is not a regular installed file")
+        if hashlib.sha256(path.read_bytes()).hexdigest() != expected:
+            raise ValueError("installed bundle hash mismatch")
+    return {"source_revision": manifest["source_revision"], "verified_files": len(FILES),
+            "manifest_sha256": hashlib.sha256((root / "manifest.json").read_bytes()).hexdigest()}
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--root", type=Path, required=True)
+    parser.add_argument("--expected-revision")
+    args = parser.parse_args()
+    print(json.dumps(verify(args.root, args.expected_revision), sort_keys=True))

src/activity_core/custodian_inputs.py

@@ -0,0 +1,81 @@
+"""Fail-closed admission for trusted-owner fleet snapshots and freshness receipts."""
+import hashlib
+import json
+import stat
+from datetime import UTC, datetime
+from pathlib import Path, PurePosixPath
+
+
+def _read(path):
+    path = Path(path)
+    if any(p.is_symlink() for p in [path, *path.parents]) or not stat.S_ISREG(path.stat().st_mode) or path.stat().st_size > 8 * 1024 * 1024:
+        raise ValueError("unsafe fleet evidence path or size")
+    return path.read_bytes()
+
+
+def admit(fleet, policy_path, receipts_path, repos, now=None):
+    """Files/receipts are protected owner inputs, not signatures or remote verification."""
+    fleet = Path(fleet)
+    policy = json.loads(_read(policy_path))
+    if policy.get("schema") != "activity-core.custodian-input-policy.v1" or set(policy["repositories"]) != set(repos):
+        raise ValueError("fleet policy repository mismatch")
+    raw = _read(fleet / "manifest.json")
+    if _read(fleet / "COMPLETE").decode().strip() != hashlib.sha256(raw).hexdigest():
+        raise ValueError("incomplete fleet snapshot")
+    manifest = json.loads(raw)
+    records = manifest.get("repositories", [])
+    if manifest.get("schema") != "repo-manager.fleet-input.v1" or len(records) != len(repos) or {r["slug"] for r in records} != set(repos):
+        raise ValueError("fleet snapshot repository mismatch")
+    if {p.name for p in fleet.iterdir()} != set(repos) | {"manifest.json", "COMPLETE"}:
+        raise ValueError("snapshot contains unexpected repository or file")
+    receipts = json.loads(_read(receipts_path))
+    now = now or datetime.now(UTC)
+    expected_files = set()
+    total = 0
+    for record in records:
+        valid = False
+        for receipt in receipts:
+            if receipt.get("schema") != "repo-manager.freshness-receipt.v1" or receipt.get("repo_slug") != record["slug"] or receipt.get("forge_commit") != record["commit"]:
+                continue
+            hub = receipt.get("hub", {})
+            if hub.get("verified_primary") is not True or hub.get("instance_role") != "primary" or hub.get("instance_label") != policy["expected_hub_instance"]:
+                continue
+            if receipt.get("outcome") not in {"applied", "noop"}:
+                continue
+            noop = receipt.get("outcome") == "noop" and receipt.get("reason") == "forge_head_matches_last_success"
+            bound = all(receipt.get(key) == record["commit"] for key in ("expected_commit", "derived_commit", "projected_commit"))
+            # Native issuer emits this reason only after HEAD == last successful commit.
+            # Receipts are protected owner evidence; this is not signature verification.
+            bound = bound or noop
+            if not bound:
+                continue
+            observed = datetime.fromisoformat(receipt["finished_at"])
+            if observed.tzinfo is None:
+                continue
+            age = (now - observed).total_seconds()
+            if -policy["max_future_skew_seconds"] <= age <= policy["max_observation_age_seconds"]:
+                valid = True
+        if not valid:
+            raise ValueError("missing accepted fresh Forge receipt")
+        for entry in record["files"]:
+            relative = PurePosixPath(entry["path"])
+            if relative.is_absolute() or ".." in relative.parts or relative.parts[0] not in {"workplans", "decisions"} or relative.suffix != ".md":
+                raise ValueError("unsafe snapshot record")
+            path = fleet / record["slug"] / relative
+            data = _read(path)
+            total += len(data)
+            if total > policy["max_snapshot_bytes"]:
+                raise ValueError("snapshot size policy exceeded")
+            if len(data) != entry["bytes"] or hashlib.sha256(data).hexdigest() != entry["sha256"]:
+                raise ValueError("snapshot record hash mismatch")
+            expected_files.add(path)
+    actual_files = set()
+    for slug in repos:
+        for path in (fleet / slug).rglob("*"):
+            if path.is_symlink():
+                raise ValueError("snapshot symlink refused")
+            if path.is_file():
+                actual_files.add(path)
+    if actual_files != expected_files or total > policy["max_snapshot_bytes"]:
+        raise ValueError("snapshot content outside manifest or size policy")
+    return hashlib.sha256(raw).hexdigest()

src/activity_core/custodian_operations.py

@@ -89,6 +89,10 @@ def build_command(operation: str, params: dict) -> tuple[list[str], Path, Path,
             raise ValueError("explicit canonical security-flow repository list required")
         for repo in repos:
             _path(str(fleet / repo / "workplans"), directory=True)
+        from activity_core.custodian_inputs import admit
+        policy_path = _path(os.environ.get("ACTIVITY_CORE_CUSTODIAN_INPUT_POLICY"))
+        receipts_path = _path(os.environ.get("ACTIVITY_CORE_CUSTODIAN_FRESHNESS_RECEIPTS"))
+        admit(fleet, policy_path, receipts_path, repos)
         hub = os.environ.get("ACTIVITY_CORE_CUSTODIAN_HUB", "")
         parsed = urlsplit(hub)
         if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path not in {"", "/"}:

tests/test_custodian_bundle.py

@@ -0,0 +1,31 @@
+import importlib.util
+import shutil
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+spec = importlib.util.spec_from_file_location("bundle_verifier", ROOT / "scripts/verify_custodian_bundle.py")
+module = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(module)
+
+
+def test_reviewed_bundle_bytes():
+    result = module.verify(ROOT / "k8s/custodian-release/commands", "60d893af1bf77f457504b64ff1c9ffb5c6ae6579")
+    assert result["verified_files"] == 4
+
+
+def test_tampered_installed_file_refused(tmp_path):
+    shutil.copytree(ROOT / "k8s/custodian-release/commands", tmp_path / "bundle")
+    (tmp_path / "bundle/scripts/security_flow_daily.py").write_text("changed")
+    with pytest.raises(ValueError, match="hash mismatch"):
+        module.verify(tmp_path / "bundle")
+
+
+def test_installed_symlink_refused(tmp_path):
+    shutil.copytree(ROOT / "k8s/custodian-release/commands", tmp_path / "bundle")
+    path = tmp_path / "bundle/scripts/security_flow_daily.py"
+    path.unlink()
+    path.symlink_to(ROOT / "k8s/custodian-release/commands/scripts/security_flow_daily.py")
+    with pytest.raises(ValueError, match="regular installed"):
+        module.verify(tmp_path / "bundle")

tests/test_custodian_inputs.py

@@ -0,0 +1,93 @@
+import hashlib
+import json
+from datetime import UTC, datetime
+
+import pytest
+
+from activity_core.custodian_inputs import admit
+
+
+@pytest.fixture
+def evidence(tmp_path):
+    fleet = tmp_path / "fleet"
+    file = fleet / "example/workplans/task.md"
+    file.parent.mkdir(parents=True)
+    file.write_text("task")
+    manifest = {"schema": "repo-manager.fleet-input.v1", "freshness_verified": False,
+                "repositories": [{"slug": "example", "commit": "a" * 40, "files": [
+                    {"path": "workplans/task.md", "bytes": 4, "sha256": hashlib.sha256(b"task").hexdigest()}]}]}
+    raw = json.dumps(manifest).encode()
+    (fleet / "manifest.json").write_bytes(raw)
+    (fleet / "COMPLETE").write_text(hashlib.sha256(raw).hexdigest())
+    policy = tmp_path / "policy.json"
+    policy.write_text(json.dumps({"schema": "activity-core.custodian-input-policy.v1", "repositories": ["example"],
+                                 "max_observation_age_seconds": 7200, "max_future_skew_seconds": 300,
+                                 "max_snapshot_bytes": 1024, "expected_hub_instance": "railiance01"}))
+    receipts = tmp_path / "receipts.json"
+    receipt = {"schema": "repo-manager.freshness-receipt.v1", "repo_slug": "example", "forge_commit": "a" * 40,
+               "outcome": "noop", "reason": "forge_head_matches_last_success", "finished_at": "2026-10-19T10:00:00Z",
+               "hub": {"verified_primary": True, "instance_role": "primary", "instance_label": "railiance01"}}
+    prior = {**receipt, "outcome": "applied", "reason": None,
+             "finished_at": "2026-10-01T00:00:00Z",
+             **{key: "a" * 40 for key in ("expected_commit", "derived_commit", "projected_commit")}}
+    receipts.write_text(json.dumps([receipt, prior]))
+    return fleet, policy, receipts
+
+
+def check(evidence):
+    return admit(*evidence, ["example"], now=datetime(2026, 10, 19, 11, tzinfo=UTC))
+
+
+def test_accepts_independent_recent_receipt(evidence):
+    assert len(check(evidence)) == 64
+
+
+@pytest.mark.parametrize("case", ["missing", "stale", "future", "wrong_commit", "wrong_hub", "failed"])
+def test_rejects_invalid_freshness(evidence, case):
+    path = evidence[2]
+    rows = json.loads(path.read_text())
+    if case == "missing":
+        rows = []
+    elif case == "stale":
+        rows[0]["finished_at"] = "2026-10-18T10:00:00Z"
+    elif case == "future":
+        rows[0]["finished_at"] = "2026-10-20T10:00:00Z"
+    elif case == "wrong_commit":
+        rows[0]["forge_commit"] = "b" * 40
+    elif case == "wrong_hub":
+        rows[0]["hub"]["instance_label"] = "workstation"
+    else:
+        rows[0]["outcome"] = "failed"
+    path.write_text(json.dumps(rows))
+    with pytest.raises(ValueError, match="fresh Forge receipt"):
+        check(evidence)
+
+
+@pytest.mark.parametrize("case", ["tamper", "extra", "marker", "extra_repo"])
+def test_rejects_changed_snapshot(evidence, case):
+    fleet = evidence[0]
+    if case == "tamper":
+        (fleet / "example/workplans/task.md").write_text("changed")
+    elif case == "extra":
+        (fleet / "example/workplans/extra.md").write_text("extra")
+    elif case == "extra_repo":
+        (fleet / "outside").mkdir()
+    else:
+        (fleet / "COMPLETE").write_text("incorrect")
+    with pytest.raises(ValueError):
+        check(evidence)
+
+
+def test_native_noop_uses_protected_issuer_contract(evidence):
+    path = evidence[2]
+    path.write_text(json.dumps(json.loads(path.read_text())[:1]))
+    assert len(check(evidence)) == 64
+
+
+def test_fifo_refused_without_open(evidence):
+    import os
+    path = evidence[2]
+    path.unlink()
+    os.mkfifo(path)
+    with pytest.raises(ValueError, match="unsafe"):
+        check(evidence)

tests/test_custodian_operations.py

@@ -1,4 +1,6 @@
 import json
+import hashlib
+from datetime import datetime, timezone
 from pathlib import Path
 from unittest.mock import patch
 
@@ -21,6 +23,26 @@ def runtime(tmp_path, monkeypatch):
     monkeypatch.setenv("ACTIVITY_CORE_CUSTODIAN_HUB", "http://127.0.0.1:8000")
     monkeypatch.setenv("ACTIVITY_CORE_CUSTODIAN_REPOS", "the-custodian")
     (fleet / "the-custodian/workplans").mkdir(parents=True)
+    (fleet / "the-custodian/workplans/task.md").write_text("task")
+    manifest = {"schema": "repo-manager.fleet-input.v1", "repositories": [{
+        "slug": "the-custodian", "commit": "a" * 40, "files": [{"path": "workplans/task.md",
+        "bytes": 4, "sha256": hashlib.sha256(b"task").hexdigest()}]}]}
+    raw = json.dumps(manifest).encode()
+    (fleet / "manifest.json").write_bytes(raw)
+    (fleet / "COMPLETE").write_text(hashlib.sha256(raw).hexdigest())
+    policy = state / "policy.json"
+    policy.write_text(json.dumps({"schema": "activity-core.custodian-input-policy.v1",
+        "repositories": ["the-custodian"], "max_observation_age_seconds": 7200,
+        "max_future_skew_seconds": 300, "max_snapshot_bytes": 1024,
+        "expected_hub_instance": "railiance01"}))
+    receipts = state / "receipts.json"
+    receipts.write_text(json.dumps([{"schema": "repo-manager.freshness-receipt.v1",
+        "repo_slug": "the-custodian", "outcome": "applied",
+        "finished_at": datetime.now(timezone.utc).isoformat(),
+        "hub": {"verified_primary": True, "instance_role": "primary", "instance_label": "railiance01"},
+        **{key: "a" * 40 for key in ("forge_commit", "expected_commit", "derived_commit", "projected_commit")}}]))
+    monkeypatch.setenv("ACTIVITY_CORE_CUSTODIAN_INPUT_POLICY", str(policy))
+    monkeypatch.setenv("ACTIVITY_CORE_CUSTODIAN_FRESHNESS_RECEIPTS", str(receipts))
     for name in ("security_flow_daily.py", "weekly_allocation_report.py"):
         (root / "scripts" / name).write_text("# owner fixture\n")
     return root, fleet, state
@@ -168,3 +190,23 @@ def test_subprocess_timeout_kills_group():
         with pytest.raises(RuntimeError, match='timed out'):
             runner._execute(['fixed-owner-command'], Path('/tmp'), 1)
     kill.assert_called_once_with(12345, runner.signal.SIGKILL)
+
+
+def test_configured_freshness_admission_refuses_before_subprocess(runtime, monkeypatch):
+    root, fleet, state = runtime
+    policy = state / "policy.json"
+    receipts = state / "receipts.json"
+    policy.write_text("{}")
+    receipts.write_text("[]")
+    monkeypatch.setenv("ACTIVITY_CORE_CUSTODIAN_INPUT_POLICY", str(policy))
+    monkeypatch.setenv("ACTIVITY_CORE_CUSTODIAN_FRESHNESS_RECEIPTS", str(receipts))
+    with patch.object(runner, "_execute") as execute, pytest.raises(ValueError):
+        runner.run_custodian_operation("custodian_security_flow", {"apply": True})
+    execute.assert_not_called()
+
+
+def test_missing_policy_refuses_before_subprocess(runtime, monkeypatch):
+    monkeypatch.delenv("ACTIVITY_CORE_CUSTODIAN_INPUT_POLICY")
+    with patch.object(runner, "_execute") as execute, pytest.raises(ValueError):
+        runner.run_custodian_operation("custodian_security_flow", {"apply": True})
+    execute.assert_not_called()