home-network-k3s-impl/catalog/render.py

#!/usr/bin/env python3
"""ServiceDefinition renderer (task T-0.3) — the platform's core GitOps contract.

Consumes ONE ServiceDefinition (schemas/servicedefinition.schema.json, catalog/*.yaml)
and emits the concrete manifest set the renderer contract specifies
(schemas/renderer-contract.md): Namespace + default-deny baseline, Deployment/Service or
KubeVirt VirtualMachine + CDI DataVolume, PVCs, per-dependency CiliumNetworkPolicy allows,
egress-class steering label, HTTPRoute(s), ServiceMonitor/alerts, Flux image-automation,
and the auth-mode / backup wiring hints.

Requirements: R-GITOPS-02 (single reviewable object wires every requirement), R-NET-01/02
(default-deny + explicit allows), R-NET-05 (egress class), R-STATE-01/02/04 (storage),
R-TLS/R-DNS (ingressRoutes/exposure), R-OBS-01 (observability), R-UPD-01/02/03 (update),
R-IDP-02/06 (auth), R-PLAT-02/03 (kind/source).

Design (documented per the task's "if the contract is ambiguous, choose the simplest
reproducible option"): a deterministic Python renderer. The contract does not mandate a
mechanism (script / kustomize generator / templates); a render-then-consume script is the
simplest reproducible option the repo can run in CI and that Flux consumes from a rendered
output path. CI invokes it via `render.py <sd>` (or `--all`); the offline L1 render
conformance verifier diffs its deterministic output against committed goldens.

Determinism: fixed document order + fixed key order + PyYAML with sort_keys=False and a
wide line width, so identical input always yields byte-identical output (golden-diffable).

Dependency: PyYAML (input parse + output emit). This is the ONE renderer dependency; the
conformance harness stays stdlib-only and only shells out to this script + diffs text.
"""
from __future__ import annotations

import argparse
import sys
from pathlib import Path

try:
    import yaml
except ImportError:  # pragma: no cover - explicit, actionable failure
    sys.stderr.write(
        "render.py requires PyYAML (the renderer's single dependency). "
        "Install with: pip install pyyaml\n"
    )
    raise SystemExit(2)

# --- Label / naming conventions (match platform/kyverno/policies + platform/*/) --------
LBL = "homelab.keithjackson.dev"
ZONE = "home.keithjackson.dev"
GATEWAY_NAME = "shared"
GATEWAY_NS = "gateway"
KP_RELEASE = "kube-prometheus-stack"  # ServiceMonitor/PrometheusRule discovery label

# egress classes and auth modes fixed by docs/03 / docs/13 + the Kyverno enums.
EGRESS_CLASSES = ("internet", "vpn")
AUTH_MODES = ("oidc", "forward-auth", "native", "none")


# =============================================================================
# Defaults (renderer-contract.md "Default posture") — materialize BEFORE mapping.
# =============================================================================
def apply_defaults(sd: dict) -> dict:
    """Return a copy of the ServiceDefinition with the contract's default posture filled.

    A field absent from a ServiceDefinition renders from its default posture, not "not at
    all" (renderer-contract.md).
    """
    s = dict(sd)
    s.setdefault("kind", "container")
    s.setdefault("exposure", "internal")
    s.setdefault("egress", "internet")
    s.setdefault("storage", [])
    s.setdefault("dependencies", [])
    s.setdefault("ingressRoutes", [])

    # auth.mode default posture: forward-auth for any exposed UI, else none (internal, no
    # UI). An explicit auth block always wins (R-IDP-02).
    auth = dict(s.get("auth") or {})
    if "mode" not in auth:
        auth["mode"] = "forward-auth" if s["ingressRoutes"] else "none"
    auth.setdefault("groups", [])
    s["auth"] = auth

    update = dict(s.get("update") or {})
    update.setdefault("channel", "stable")
    update.setdefault("autoRollbackOnFailedHealth", True)
    s["update"] = update

    obs = dict(s.get("observability") or {})
    obs.setdefault("metrics", True)
    obs.setdefault("slis", [])
    s["observability"] = obs

    # storage entry class default is longhorn (replicated tier).
    s["storage"] = [{**{"class": "longhorn"}, **v} for v in s["storage"]]
    # dependency protocol default is tcp.
    s["dependencies"] = [{**{"protocol": "tcp"}, **d} for d in s["dependencies"]]
    return s


# =============================================================================
# Field -> resource mapping (renderer-contract.md "Field -> resource mapping")
# =============================================================================
def r_namespace(s: dict) -> dict:
    """Namespace = the service trust zone / isolation boundary (R-NET, docs/03).

    Carries app-namespace=true so Kyverno generates the default-deny baseline and the
    egress-class / auth-mode admission policies apply. gateway-access=shared lets its
    HTTPRoutes attach to the shared Gateway (only when it exposes routes).
    """
    labels = {f"{LBL}/app-namespace": "true"}
    if s["ingressRoutes"]:
        labels["gateway-access"] = "shared"
    ann = {}
    backup = (s.get("backup") or {}).get("policy")
    if backup:
        ann[f"{LBL}/backup-policy"] = backup  # R-STATE-03 off-cluster schedule ref (hint)
    meta = {"name": s["name"], "labels": labels}
    if ann:
        meta["annotations"] = ann
    return {"apiVersion": "v1", "kind": "Namespace", "metadata": meta}


def r_default_deny(s: dict) -> dict:
    """The default-deny baseline: deny all ingress+egress, allow ONLY DNS egress (R-NET-01).

    Rendered explicitly (GitOps at namespace creation, renderer-contract §dependencies) —
    it matches the Kyverno-generated baseline shape (platform/kyverno/.../generated-default
    -deny.yaml) so the render carries the guarantee even before admission generates it.
    """
    return {
        "apiVersion": "networking.k8s.io/v1",
        "kind": "NetworkPolicy",
        "metadata": {
            "name": "default-deny",
            "namespace": s["name"],
            "labels": {f"{LBL}/baseline": "default-deny"},
        },
        "spec": {
            "podSelector": {},
            "policyTypes": ["Ingress", "Egress"],
            "ingress": [],
            "egress": [
                {
                    "to": [
                        {
                            "namespaceSelector": {
                                "matchLabels": {
                                    "kubernetes.io/metadata.name": "kube-system"
                                }
                            }
                        }
                    ],
                    "ports": [
                        {"protocol": "UDP", "port": 53},
                        {"protocol": "TCP", "port": 53},
                    ],
                }
            ],
        },
    }


def _pod_labels(s: dict) -> dict:
    """Workload pod labels: identity + the egress-class steering hook + auth-mode.

    egress-class is what the shared Cilium Egress Gateway selects on to steer vpn-class
    pods (enforcement is OUTSIDE the workload — renderer-contract §egress; ADR-0015);
    auth-mode satisfies the Kyverno require-auth-mode-label policy (R-IDP-02).
    """
    return {
        "app.kubernetes.io/name": s["name"],
        f"{LBL}/egress-class": s["egress"],
        f"{LBL}/auth-mode": s["auth"]["mode"],
    }


def _service_ports(s: dict) -> list:
    ports = sorted({r["port"] for r in s["ingressRoutes"]})
    return [{"name": f"port-{p}", "port": p, "targetPort": p} for p in ports]


def r_pvcs(s: dict) -> list:
    """One PVC per storage entry (R-STATE-01/02/04).

    longhorn -> 3-replica replicated tier; bulk -> external media tier (not 3x replicated).
    """
    out = []
    for vol in s["storage"]:
        cls = vol.get("class", "longhorn")
        storage_class = "longhorn" if cls == "longhorn" else "bulk"
        spec = {
            "accessModes": ["ReadWriteOnce"],
            "storageClassName": storage_class,
        }
        if vol.get("size"):
            spec["resources"] = {"requests": {"storage": vol["size"]}}
        out.append(
            {
                "apiVersion": "v1",
                "kind": "PersistentVolumeClaim",
                "metadata": {
                    "name": f"{s['name']}-{vol['name']}",
                    "namespace": s["name"],
                    "labels": {"app.kubernetes.io/name": s["name"]},
                },
                "spec": spec,
            }
        )
    return out


def r_deployment(s: dict) -> dict:
    """Deployment for a container workload (R-PLAT-02)."""
    pod_labels = _pod_labels(s)
    container = {
        "name": s["name"],
        "image": s["source"]["image"],
    }
    ports = _service_ports(s)
    if ports:
        container["ports"] = [{"containerPort": p["port"]} for p in ports]

    # env / envFrom (schema-optional): the k8s EnvVar / EnvFromSource lists are emitted
    # VERBATIM onto the app container. This is what lets a headless-install/DB-backed app
    # (gitea, nextcloud, semaphore, unifi, minecraft) configure itself and pull a fixed
    # secret (e.g. gitea's SECRET_KEY via secretKeyRef) without a supplementary patch.
    # Absent => the key is never set, so apps that declare no env render byte-identical.
    if s.get("env"):
        container["env"] = s["env"]
    if s.get("envFrom"):
        container["envFrom"] = s["envFrom"]

    volume_mounts, volumes = [], []
    for vol in s["storage"]:
        # Default mount is /data/<name>; an explicit `mountPath` (schema-optional) overrides
        # it verbatim for apps that need a fixed on-disk layout (e.g. Gitea keeps all state
        # directly under /data). Omitting it preserves the /data/<name> default (back-compat).
        mount_path = vol.get("mountPath", f"/data/{vol['name']}")
        volume_mounts.append({"name": vol["name"], "mountPath": mount_path})
        volumes.append(
            {
                "name": vol["name"],
                "persistentVolumeClaim": {"claimName": f"{s['name']}-{vol['name']}"},
            }
        )
    if volume_mounts:
        container["volumeMounts"] = volume_mounts

    # lifecycle hooks (schema-optional): the k8s Lifecycle (postStart/preStop exec) is
    # emitted VERBATIM. Used for idempotent break-glass/bootstrap steps (e.g. gitea's
    # postStart exec that seeds a local admin). Absent => byte-identical for other apps.
    if s.get("lifecycle"):
        container["lifecycle"] = s["lifecycle"]

    pod_spec = {"containers": [container]}

    # podSecurityContext (schema-optional): the k8s PodSecurityContext is emitted VERBATIM onto the
    # pod spec. Its primary use is `fsGroup`, which makes a mounted Longhorn PVC group-writable by a
    # non-root image (e.g. semaphore runs as UID 1001 and cannot otherwise write its config PVC).
    # Absent => the key is never set, so apps that declare none render byte-identical.
    if s.get("podSecurityContext"):
        pod_spec["securityContext"] = s["podSecurityContext"]

    if volumes:
        pod_spec["volumes"] = volumes

    ann = _auth_annotations(s)
    tmpl_meta = {"labels": pod_labels}
    if ann:
        tmpl_meta["annotations"] = ann

    spec = {"replicas": 1}
    # An app with persistent storage mounts a ReadWriteOnce Longhorn PVC. The default
    # RollingUpdate strategy tries to start the NEW pod before terminating the old one, but
    # the old pod still holds the RWO volume -> the new pod is stuck ContainerCreating and
    # the rollout deadlocks (single-replica app can never hand the volume over). Recreate
    # terminates the old pod first, freeing the PVC for the new one (R-STATE-01/02).
    if s["storage"]:
        spec["strategy"] = {"type": "Recreate"}
    spec["selector"] = {"matchLabels": {"app.kubernetes.io/name": s["name"]}}
    spec["template"] = {"metadata": tmpl_meta, "spec": pod_spec}

    return {
        "apiVersion": "apps/v1",
        "kind": "Deployment",
        "metadata": {
            "name": s["name"],
            "namespace": s["name"],
            "labels": {"app.kubernetes.io/name": s["name"]},
        },
        "spec": spec,
    }


def r_service(s: dict):
    ports = _service_ports(s)
    if not ports:
        return None
    return {
        "apiVersion": "v1",
        "kind": "Service",
        "metadata": {
            "name": s["name"],
            "namespace": s["name"],
            "labels": {"app.kubernetes.io/name": s["name"]},
        },
        "spec": {
            "selector": {"app.kubernetes.io/name": s["name"]},
            "ports": ports,
        },
    }


def r_dependency_policies(s: dict) -> list:
    """One CiliumNetworkPolicy egress allow per dependency, on top of default-deny (R-NET-02).

    Everything undeclared stays denied; empty dependencies => reaches only DNS.

    Cross-namespace note: a CiliumNetworkPolicy's `toEndpoints` selector is scoped to the
    POLICY's own namespace unless it carries the reserved `k8s:io.kubernetes.pod.namespace`
    label. Every service lives in its own namespace named after it (r_namespace), and a
    dependency always targets a DIFFERENT service/namespace, so the destination's namespace
    (== dep["to"]) MUST be pinned here — otherwise Cilium looks for the destination's app
    label inside the SOURCE namespace, finds nothing, and silently denies the egress (proven
    in-cluster via Hubble: `EGRESS DENIED`). This pins it so the declared dependency's egress
    half actually reaches the destination (verified: EGRESS ALLOWED / FORWARDED).
    """
    out = []
    for dep in s["dependencies"]:
        proto = dep.get("protocol", "tcp").upper()
        out.append(
            {
                "apiVersion": "cilium.io/v2",
                "kind": "CiliumNetworkPolicy",
                "metadata": {
                    "name": f"{s['name']}-allow-{dep['to']}-{dep['port']}",
                    "namespace": s["name"],
                    "labels": {"app.kubernetes.io/name": s["name"]},
                },
                "spec": {
                    "endpointSelector": {
                        "matchLabels": {"app.kubernetes.io/name": s["name"]}
                    },
                    "egress": [
                        {
                            "toEndpoints": [
                                {
                                    "matchLabels": {
                                        "app.kubernetes.io/name": dep["to"],
                                        "k8s:io.kubernetes.pod.namespace": dep["to"],
                                    }
                                }
                            ],
                            "toPorts": [
                                {
                                    "ports": [
                                        {"port": str(dep["port"]), "protocol": proto}
                                    ]
                                }
                            ],
                        }
                    ],
                },
            }
        )
    return out


def build_reverse_map(sds: list) -> dict:
    """Build the cross-service reverse-dependency map (T-2.3 reciprocal ingress).

    A `dependencies` edge is declared ONCE, in the SOURCE's catalog entry, as
    `{to: X, port: P}`. With the default-deny-BOTH-directions baseline (R-NET-01), that one
    declaration must compile into TWO matching halves (docs/03 "Explicit allow rules"; the
    conformance fixtures spell it out: "egress permit on the source, ingress permit on the
    destination"): (a) source egress-allow to X:P — rendered by r_dependency_policies on the
    source — AND (b) X ingress-allow from source:P — the reciprocal half rendered here.

    render.py otherwise processes each ServiceDefinition independently, so the destination X
    never sees the SOURCE's declaration. This pass scans ALL ServiceDefinitions once and
    returns `{destination_name: [{source, port, protocol}, ...]}` so that, when rendering X,
    r_reverse_dependency_policies can emit the ingress half for every source that declares a
    dependency ON X. Deterministic: each destination's inbound list is sorted by
    (source, port, protocol) so the render stays byte-reproducible (golden-diffable).
    """
    rev: dict = {}
    for sd in sds:
        if not isinstance(sd, dict) or "name" not in sd:
            continue  # tolerate non-ServiceDefinition siblings (e.g. a stray yaml)
        s = apply_defaults(sd)
        src = s["name"]
        for dep in s["dependencies"]:
            rev.setdefault(dep["to"], []).append(
                {
                    "source": src,
                    "port": dep["port"],
                    "protocol": dep.get("protocol", "tcp"),
                }
            )
    for dest in rev:
        rev[dest].sort(key=lambda e: (e["source"], int(e["port"]), e["protocol"]))
    return rev


def r_reverse_dependency_policies(s: dict, inbound: list) -> list:
    """Reciprocal ingress-allow CiliumNetworkPolicies on THIS service (R-NET-02).

    `inbound` is this service's slice of the cross-service reverse map (build_reverse_map):
    the list of sources that declared a dependency ON this service. Each becomes an ADDITIVE
    ingress allow on top of default-deny, mirroring the source-side egress allow
    (r_dependency_policies) exactly but inverted: `fromEndpoints` matching the SOURCE's
    identity on the declared port/protocol. As with the egress half, the source lives in a
    DIFFERENT namespace, and a CiliumNetworkPolicy's `fromEndpoints` selector is scoped to the
    POLICY's own namespace unless it carries the reserved `k8s:io.kubernetes.pod.namespace`
    label — so the source's namespace (== edge["source"]) is pinned alongside its app label.
    Without both halves so pinned the destination's default-deny-ingress drops the source's
    traffic and the declared dependency does not actually work (docs/03 R-NET-01/02; proven
    in-cluster via Hubble: pinned => FORWARDED, unpinned => Policy denied DROPPED).
    """
    out = []
    for edge in inbound or []:
        proto = edge.get("protocol", "tcp").upper()
        out.append(
            {
                "apiVersion": "cilium.io/v2",
                "kind": "CiliumNetworkPolicy",
                "metadata": {
                    "name": f"{s['name']}-allow-{edge['source']}-{edge['port']}",
                    "namespace": s["name"],
                    "labels": {"app.kubernetes.io/name": s["name"]},
                },
                "spec": {
                    "endpointSelector": {
                        "matchLabels": {"app.kubernetes.io/name": s["name"]}
                    },
                    "ingress": [
                        {
                            "fromEndpoints": [
                                {
                                    "matchLabels": {
                                        "app.kubernetes.io/name": edge["source"],
                                        "k8s:io.kubernetes.pod.namespace": edge["source"],
                                    }
                                }
                            ],
                            "toPorts": [
                                {
                                    "ports": [
                                        {"port": str(edge["port"]), "protocol": proto}
                                    ]
                                }
                            ],
                        }
                    ],
                },
            }
        )
    return out


def r_gateway_ingress_policy(s: dict):
    """North-south allow: let the shared Cilium Gateway reach an exposed service's port(s).

    Any service that renders an HTTPRoute (non-empty ingressRoutes) is reachable through the
    shared Gateway, but the namespace default-deny has `ingress: []` (deny all east-west and
    north-south). This CiliumNetworkPolicy is the ADDITIVE, least-privilege allow that opens
    ONLY the north-south path (docs/09 "Ingress opens north-south only — it does not relax
    east-west default-deny"): ingress from Cilium's reserved `ingress` identity (the Gateway/
    Ingress Envoy proxy, `reserved:ingress`, numeric identity 8 in Cilium 1.16.5) to ONLY the
    exposed service port(s). Everything else stays denied; undeclared ports stay denied.

    Verified empirically against the cluster: with default-deny in place, Hubble shows the
    Gateway proxy reaching the backend as source identity `(ingress)` and the flow
    `Policy denied DROPPED` on the service port — so `fromEntities: [ingress]` is exactly the
    source the Cilium Gateway uses, and the port is the service targetPort (R-NET-01/02).
    """
    if not s["ingressRoutes"]:
        return None
    ports = _service_ports(s)
    return {
        "apiVersion": "cilium.io/v2",
        "kind": "CiliumNetworkPolicy",
        "metadata": {
            "name": f"{s['name']}-allow-gateway-ingress",
            "namespace": s["name"],
            "labels": {"app.kubernetes.io/name": s["name"]},
        },
        "spec": {
            "endpointSelector": {
                "matchLabels": {"app.kubernetes.io/name": s["name"]}
            },
            "ingress": [
                {
                    "fromEntities": ["ingress"],
                    "toPorts": [
                        {
                            "ports": [
                                {"port": str(p["port"]), "protocol": "TCP"}
                                for p in ports
                            ]
                        }
                    ],
                }
            ],
        },
    }


def _route_meta(s: dict, route: dict) -> dict:
    meta = {
        "name": f"{s['name']}-{route['port']}",
        "namespace": s["name"],
        "labels": {"app.kubernetes.io/name": s["name"]},
    }
    ann = _auth_annotations(s)
    if ann:
        meta["annotations"] = ann
    return meta


def r_internet_egress(s: dict) -> dict:
    """North-south egress allow: let the app reach the INTERNET (Cilium reserved `world`
    entity = everything outside the cluster), which is the whole point of the `egress` class
    (docs/03: internet-class → WAN; vpn-class → steered through the node tun0 PIA tunnel with
    the nftables kill-switch, ADR-0015). The default-deny baseline (R-NET-01) allows only DNS
    so EAST-WEST stays denied; this ADDITIVE policy opens ONLY `world` (non-cluster) egress —
    cluster/east-west traffic stays denied except the declared `dependencies`. Every app
    declares an egress class (EGRESS_CLASSES), so every app gets this. The steering + kill-
    switch (a cluster-wide singleton keyed on the egress-class pod label) decide the PATH; this
    policy decides that north-south egress is ALLOWED at all.
    """
    return {
        "apiVersion": "cilium.io/v2",
        "kind": "CiliumNetworkPolicy",
        "metadata": {
            "name": f"{s['name']}-allow-internet-egress",
            "namespace": s["name"],
            "labels": {"app.kubernetes.io/name": s["name"]},
        },
        "spec": {
            "endpointSelector": {"matchLabels": {"app.kubernetes.io/name": s["name"]}},
            "egress": [{"toEntities": ["world"]}],
        },
    }


def r_httproutes(s: dict) -> list:
    """One HTTPRoute per ingressRoute, attaching host -> service port on the shared Gateway.

    TLS is the Gateway's wildcard *.home.keithjackson.dev cert (no per-service cert step);
    ExternalDNS manages the record from the HTTPRoute (R-TLS/R-DNS, renderer-contract).
    """
    out = []
    for route in s["ingressRoutes"]:
        out.append(
            {
                "apiVersion": "gateway.networking.k8s.io/v1",
                "kind": "HTTPRoute",
                "metadata": _route_meta(s, route),
                "spec": {
                    "parentRefs": [
                        {
                            "name": GATEWAY_NAME,
                            "namespace": GATEWAY_NS,
                            "sectionName": "https",
                        }
                    ],
                    "hostnames": [route["host"]],
                    "rules": [
                        {
                            "matches": [
                                {"path": {"type": "PathPrefix", "value": "/"}}
                            ],
                            "backendRefs": [
                                {"name": s["name"], "port": route["port"]}
                            ],
                        }
                    ],
                },
            }
        )
    return out


def _auth_annotations(s: dict) -> dict:
    """Auth wiring hints (renderer-contract §auth): the platform (oauth2-proxy/Keycloak)
    consumes these to protect ingressRoutes. oidc/forward-auth carry the authorized groups
    (R-IDP-06: access is 'in an authorized group', not merely 'logged in')."""
    mode = s["auth"]["mode"]
    if mode in ("oidc", "forward-auth"):
        groups = ",".join(s["auth"].get("groups", []))
        return {
            f"{LBL}/auth-mode": mode,
            f"{LBL}/auth-groups": groups,
        }
    return {}


def r_servicemonitor(s: dict):
    if not s["observability"]["metrics"]:
        return None
    return {
        "apiVersion": "monitoring.coreos.com/v1",
        "kind": "ServiceMonitor",
        "metadata": {
            "name": s["name"],
            "namespace": s["name"],
            "labels": {"release": KP_RELEASE, "app.kubernetes.io/name": s["name"]},
        },
        "spec": {
            "selector": {"matchLabels": {"app.kubernetes.io/name": s["name"]}},
            "namespaceSelector": {"matchNames": [s["name"]]},
            "endpoints": [{"port": "metrics", "interval": "30s"}],
        },
    }


def r_prometheusrule(s: dict):
    """Each SLI becomes at least one alert rule (R-OBS-01)."""
    slis = s["observability"]["slis"]
    if not slis:
        return None
    rules = []
    for sli in slis:
        if sli == "up":
            rules.append(
                {
                    "alert": f"{s['name']}-down".replace("_", "-"),
                    "expr": f'up{{job="{s["name"]}"}} == 0',
                    "for": "5m",
                    "labels": {"severity": "critical", "service": s["name"]},
                    "annotations": {
                        "summary": f"{s['name']} is down",
                        "description": f"{s['name']} has been unreachable for 5m (SLI: up).",
                    },
                }
            )
        else:
            rules.append(
                {
                    "alert": f"{s['name']}-{sli}".replace("_", "-"),
                    "expr": f'absent({sli}{{service="{s["name"]}"}})',
                    "for": "10m",
                    "labels": {"severity": "warning", "service": s["name"]},
                    "annotations": {"summary": f"{s['name']} SLI {sli} unhealthy"},
                }
            )
    return {
        "apiVersion": "monitoring.coreos.com/v1",
        "kind": "PrometheusRule",
        "metadata": {
            "name": f"{s['name']}-slis",
            "namespace": s["name"],
            "labels": {"release": KP_RELEASE, "app.kubernetes.io/name": s["name"]},
        },
        "spec": {"groups": [{"name": f"{s['name']}.slis", "rules": rules}]},
    }


def r_image_automation(s: dict) -> list:
    """container kind -> Flux ImageRepository + ImagePolicy encoding channel/range (R-UPD).

    Flux image-automation advances the running version on its scan interval; no human/AI
    edits a tag (R-UPD-02). `pin` freezes automation. vm kind uses in-guest updates instead.
    """
    if s["kind"] != "container":
        return []
    image = s["source"]["image"].split(":", 1)[0]
    upd = s["update"]
    interval = "1h0m0s"
    imagerepo = {
        "apiVersion": "image.toolkit.fluxcd.io/v1beta2",
        "kind": "ImageRepository",
        "metadata": {"name": s["name"], "namespace": s["name"]},
        "spec": {"image": image, "interval": interval},
    }
    policy_spec: dict = {"imageRepositoryRef": {"name": s["name"]}}
    if upd.get("pin"):
        # Operator freeze escape hatch: pin to a fixed tag, stopping automation.
        policy_spec["filterTags"] = {"pattern": f"^{upd['pin']}$"}
        policy_spec["policy"] = {"alphabetical": {"order": "asc"}}
    elif upd.get("range"):
        policy_spec["policy"] = {"semver": {"range": upd["range"]}}
    else:
        policy_spec["policy"] = {"semver": {"range": ">=0.0.0"}}
    imagepolicy = {
        "apiVersion": "image.toolkit.fluxcd.io/v1beta2",
        "kind": "ImagePolicy",
        "metadata": {
            "name": s["name"],
            "namespace": s["name"],
            "annotations": {
                f"{LBL}/update-channel": upd.get("channel", "stable"),
                f"{LBL}/auto-rollback": str(
                    upd.get("autoRollbackOnFailedHealth", True)
                ).lower(),
            },
        },
        "spec": policy_spec,
    }
    return [imagerepo, imagepolicy]


# --- VM (kind: vm) path: KubeVirt VirtualMachine + CDI DataVolume ----------------------
def r_datavolume(s: dict) -> dict:
    """Standalone CDI DataVolume (NOT a dataVolumeTemplate) so the disk OUTLIVES the VM
    and survives destroy/rebuild (R-STATE-01) — matches the vm-lifecycle fixture shape."""
    root = s["storage"][0] if s["storage"] else {"name": "rootdisk", "size": "10Gi"}
    return {
        "apiVersion": "cdi.kubevirt.io/v1beta1",
        "kind": "DataVolume",
        "metadata": {
            "name": f"{s['name']}-{root['name']}",
            "namespace": s["name"],
        },
        "spec": {
            "source": {"registry": {"url": f"docker://{s['source']['vmTemplate']}"}},
            "storage": {
                "accessModes": ["ReadWriteOnce"],
                "storageClassName": "longhorn",
                "resources": {"requests": {"storage": root.get("size", "10Gi")}},
            },
        },
    }


VM_CLOUD_INIT = (
    "#cloud-config\n"
    "package_update: true\n"
    "packages:\n"
    "  - dnf-automatic\n"
    "write_files:\n"
    "  - path: /etc/dnf/automatic.conf\n"
    "    content: |\n"
    "      [commands]\n"
    "      upgrade_type = default\n"
    "      random_sleep = 0\n"
    "      download_updates = yes\n"
    "      apply_updates = yes\n"
    "runcmd:\n"
    "  - [ systemctl, enable, --now, dnf-automatic.timer ]\n"
)


def r_virtualmachine(s: dict) -> dict:
    """KubeVirt VirtualMachine; egress-class propagates to the virt-launcher pod so the
    shared Cilium egress steering + Kyverno policy see the class (R-PLAT-03; ADR-0015).
    In-guest dnf-automatic delivers auto-updates with no human step (R-UPD-01, docs/11)."""
    root = s["storage"][0] if s["storage"] else {"name": "rootdisk"}
    pod_labels = {
        "kubevirt.io/vm": s["name"],
        f"{LBL}/egress-class": s["egress"],
        f"{LBL}/auth-mode": s["auth"]["mode"],
    }
    return {
        "apiVersion": "kubevirt.io/v1",
        "kind": "VirtualMachine",
        "metadata": {
            "name": s["name"],
            "namespace": s["name"],
            "labels": {"app.kubernetes.io/name": s["name"]},
        },
        "spec": {
            "running": False,
            "template": {
                "metadata": {"labels": pod_labels},
                "spec": {
                    "domain": {
                        "cpu": {"cores": 1},
                        "memory": {"guest": "1536Mi"},
                        "devices": {
                            "disks": [
                                {"name": "rootdisk", "disk": {"bus": "virtio"}},
                                {"name": "cloudinit", "disk": {"bus": "virtio"}},
                            ],
                            "interfaces": [{"name": "default", "masquerade": {}}],
                        },
                        "resources": {"requests": {"memory": "1536Mi"}},
                    },
                    "networks": [{"name": "default", "pod": {}}],
                    "volumes": [
                        {
                            "name": "rootdisk",
                            "dataVolume": {"name": f"{s['name']}-{root['name']}"},
                        },
                        {
                            "name": "cloudinit",
                            "cloudInitNoCloud": {"userData": VM_CLOUD_INIT},
                        },
                    ],
                },
            },
        },
    }


# =============================================================================
# Top-level render + emit
# =============================================================================
def render(sd: dict, image_automation: bool = True, inbound: list | None = None) -> list:
    """Return an ordered list of (header_comment, resource_dict) for a ServiceDefinition.

    image_automation=False omits the Flux ImageRepository/ImagePolicy CRs — used when
    onboarding apps before the Flux image-reflector/-automation controllers are installed
    (their CRDs would otherwise be missing and the apps Kustomization would fail). The L1
    render goldens use the default (True), so their expected output is unchanged.

    inbound is this service's slice of the cross-service reverse-dependency map
    (build_reverse_map): the sources that declared a dependency ON this service. It renders
    the reciprocal ingress-allow half of each declared dependency (R-NET-02). None/[] (the
    default, and every single-service render with no cross-catalog context) renders no
    reciprocal ingress — so a lone render is unchanged and stays golden-diffable.
    """
    s = apply_defaults(sd)
    if s["kind"] not in ("container", "vm"):
        raise ValueError(f"unknown kind: {s['kind']!r}")
    if s["egress"] not in EGRESS_CLASSES:
        raise ValueError(f"invalid egress class: {s['egress']!r}")
    if s["auth"]["mode"] not in AUTH_MODES:
        raise ValueError(f"invalid auth.mode: {s['auth']['mode']!r}")

    docs: list = []
    docs.append(("Namespace (trust zone; default-deny + policies apply)", r_namespace(s)))
    docs.append(("Default-deny baseline (R-NET-01): deny all, allow only DNS", r_default_deny(s)))

    if s["kind"] == "container":
        # Container PVCs. (For a vm, the CDI DataVolume owns/creates the disk PVC, so we
        # never render a plain PVC for the VM root disk — it would collide with CDI.)
        for pvc in r_pvcs(s):
            docs.append((f"PVC {pvc['metadata']['name']} (R-STATE)", pvc))
        docs.append(("Deployment (R-PLAT-02)", r_deployment(s)))
        svc = r_service(s)
        if svc:
            docs.append(("Service", svc))
    else:
        docs.append(("CDI DataVolume (standalone; survives rebuild, R-STATE-01)", r_datavolume(s)))
        docs.append(("KubeVirt VirtualMachine (R-PLAT-03)", r_virtualmachine(s)))

    for cnp in r_dependency_policies(s):
        docs.append((f"CiliumNetworkPolicy egress allow (R-NET-02): {cnp['metadata']['name']}", cnp))

    for cnp in r_reverse_dependency_policies(s, inbound or []):
        docs.append(
            (
                "CiliumNetworkPolicy reciprocal ingress allow (R-NET-02): "
                f"{cnp['metadata']['name']}",
                cnp,
            )
        )

    gw_allow = r_gateway_ingress_policy(s)
    if gw_allow:
        docs.append(
            (
                "CiliumNetworkPolicy gateway-ingress allow (R-NET-01/02): "
                f"{gw_allow['metadata']['name']}",
                gw_allow,
            )
        )

    ie = r_internet_egress(s)
    docs.append(
        (
            "CiliumNetworkPolicy world egress allow (egress class governs the path; R-NET-05): "
            f"{ie['metadata']['name']}",
            ie,
        )
    )

    for hr in r_httproutes(s):
        docs.append((f"HTTPRoute {hr['metadata']['name']} (R-TLS/R-DNS)", hr))

    sm = r_servicemonitor(s)
    if sm:
        docs.append(("ServiceMonitor (R-OBS-01)", sm))
    pr = r_prometheusrule(s)
    if pr:
        docs.append(("PrometheusRule SLI alerts (R-OBS-01)", pr))

    if image_automation:
        for res in r_image_automation(s):
            docs.append((f"Flux {res['kind']} (R-UPD)", res))

    return docs


def emit(docs: list) -> str:
    """Deterministic multi-document YAML text (golden-diffable)."""
    parts = []
    for comment, resource in docs:
        body = yaml.dump(
            resource,
            sort_keys=False,
            default_flow_style=False,
            width=4096,
        )
        parts.append(f"---\n# {comment}\n{body}".rstrip() + "\n")
    return "".join(parts)


def load_sd(path: Path) -> dict:
    with path.open() as fh:
        return yaml.safe_load(fh)


def _load_catalog(dir_: Path) -> list:
    """Load every *.yaml ServiceDefinition in dir_ (tolerating unparseable siblings).

    Used to build the cross-service reverse-dependency map for a single-file render, so
    `render.py catalog/X.yaml` still emits X's reciprocal ingress for sources elsewhere in
    the same catalog. Deterministic (sorted); render.py itself is not a *.yaml so it is skipped.
    """
    sds = []
    for p in sorted(dir_.glob("*.yaml")):
        try:
            doc = load_sd(p)
        except Exception:  # pragma: no cover - a malformed sibling must not break the render
            continue
        if isinstance(doc, dict):
            sds.append(doc)
    return sds


def render_file(path: Path, image_automation: bool = True, catalog_dir: Path | None = None) -> str:
    """Render one ServiceDefinition file, wiring in its reciprocal ingress allows.

    The reverse-dependency map is built across the whole catalog (catalog_dir, defaulting to
    the file's own directory) so this service receives the ingress half of every dependency
    that OTHER services declare on it (R-NET-02). A directory with only this file (or with no
    inbound edges — e.g. the L1 render fixtures) yields an empty inbound list, leaving the
    render byte-identical to the pre-change output.
    """
    sd = load_sd(path)
    src_dir = catalog_dir if catalog_dir is not None else path.parent
    rev = build_reverse_map(_load_catalog(src_dir))
    inbound = rev.get(sd["name"], [])
    return emit(render(sd, image_automation=image_automation, inbound=inbound))


# =============================================================================
# Wave partitioning — dependency-ordered app-tier delivery (WS-APPS wave delivery)
# =============================================================================
# A single monolithic `apps` Flux Kustomization health-gates (wait:true) the ENTIRE app
# set, so one crashlooping workload blocks ALL apps and every new revision. Instead we split
# the tier into dependency-ordered WAVES: each wave is its own Flux Kustomization
# `apps-wave-<N>` that dependsOn wave N-1 (wave 0 dependsOn platform-config), so a failure in
# a later wave can no longer stall an earlier one, while everything WITHIN a wave still
# reconciles in parallel. This module computes the wave assignment from the declared catalog
# dependency graph and emits the per-wave kustomize + Flux Kustomization CRs. It changes only
# WHERE a rendered manifest is placed — never its bytes (the L1 goldens stay green).

CO_DB_SUFFIX = "-db"  # a co-located CNPG database for `<svc>` is the node `<svc>-db`.

# Non-catalog app-tier units (they carry NO ServiceDefinition, so compute_waves can't see
# them). Each is anchored to a catalog service; `above=True` forces it into the wave STRICTLY
# ABOVE its anchor because it is a hard runtime prerequisite it cannot start without (e.g. the
# Gitea Actions act_runner registers against + clones from Gitea, so Gitea must be healthy
# first); `above=False` would co-locate it in the anchor's own wave.
EXTRA_UNITS: dict = {
    "gitea-actions": {
        "anchor": "gitea",
        "above": True,
        "files": [
            "gitea-actions.yaml",
            "gitea-actions-runner-secret.sops.yaml",
        ],
    },
}


def compute_waves(sds: list) -> dict:
    """Longest-dependency-path wave per node over the catalog dependency DAG.

    Nodes = every catalog service ∪ every dependency target. A dependency target that has NO
    ServiceDefinition (a co-located `<svc>-db` CNPG database) is a leaf. wave(n) = 0 if n
    depends on nothing, else 1 + max(wave(dep)) over its declared dependencies. Because a
    co-located DB is a leaf and its consumer declares a dependency ON it, the consumer always
    lands in a strictly higher wave — the "DB strictly below its consumer" guarantee falls
    straight out of the longest-path rule (no special case). Raises ValueError on a cycle.
    """
    edges: dict = {}   # node -> set(targets it depends on)
    nodes: set = set()
    for sd in sds:
        if not isinstance(sd, dict) or "name" not in sd:
            continue  # tolerate non-ServiceDefinition siblings (e.g. render.py itself)
        s = apply_defaults(sd)
        name = s["name"]
        nodes.add(name)
        edges.setdefault(name, set())
        for dep in s["dependencies"]:
            edges[name].add(dep["to"])
            nodes.add(dep["to"])
            edges.setdefault(dep["to"], set())

    wave: dict = {}
    visiting: set = set()

    def depth(n: str) -> int:
        if n in wave:
            return wave[n]
        if n in visiting:
            raise ValueError(f"dependency cycle through {n!r}")
        visiting.add(n)
        d = 0
        for t in sorted(edges.get(n, ())):
            d = max(d, depth(t) + 1)
        visiting.discard(n)
        wave[n] = d
        return d

    for n in sorted(nodes):
        depth(n)
    return wave


def _catalog_by_name(sds: list) -> dict:
    out = {}
    for sd in sds:
        if isinstance(sd, dict) and "name" in sd:
            s = apply_defaults(sd)
            out[s["name"]] = s
    return out


def service_wave(sds: list, name: str) -> int:
    """The wave a single service's rendered manifest belongs in (for wave-aware `-o`)."""
    return compute_waves(sds).get(name, 0)


def plan_apps_waves(sds: list, present_files: set) -> dict:
    """Map the ONBOARDED app-tier file set into waves -> sorted resource basenames.

    `present_files` is the set of manifest basenames currently committed under apps/ (flat or
    already in wave-<N>/ subdirs), e.g. {"pihole.yaml", "nextcloud-db.yaml",
    "nextcloud-db-app.sops.yaml", ...}. A catalog service is ONBOARDED iff `<svc>.yaml` is
    present (the catalog also holds not-yet-onboarded SDs like keycloak/vm-demo, which must
    NOT be placed). Each onboarded unit's sidecar files inherit the correct wave by naming
    convention:
      - `<svc>.yaml`                        -> wave(svc)
      - `<svc>-bootstrap-secret.sops.yaml`  -> wave(svc)   (consumed by svc's env, same apply)
      - `<svc>-db.yaml`                     -> wave(svc-db) (a strictly-lower wave; CNPG DB)
      - `<svc>-db-app.sops.yaml`            -> wave(svc-db) (CNPG initdb owner secret)
      - `<svc>-db-backup-credentials.sops.yaml` -> wave(svc-db) (CNPG Barman S3 backup secret)
    Non-catalog units (EXTRA_UNITS, e.g. gitea-actions) place by their declared anchor+offset.
    Only files that actually exist in `present_files` are placed, so the plan is robust to a
    service having no bootstrap secret or no DB. Deterministic (sorted).
    """
    waves = compute_waves(sds)
    by_name = _catalog_by_name(sds)
    plan: dict = {}

    def add(w: int, f: str) -> None:
        if f in present_files:
            plan.setdefault(w, set()).add(f)

    onboarded = {n for n in by_name if f"{n}.yaml" in present_files}
    for name in sorted(onboarded):
        w = waves[name]
        add(w, f"{name}.yaml")
        add(w, f"{name}-bootstrap-secret.sops.yaml")
        db = f"{name}{CO_DB_SUFFIX}"
        if any(dep["to"] == db for dep in by_name[name]["dependencies"]):
            wdb = waves[db]
            add(wdb, f"{db}.yaml")
            add(wdb, f"{db}-app.sops.yaml")
            add(wdb, f"{db}-backup-credentials.sops.yaml")   # CNPG Barman S3 creds (R-STATE-03)

    for unit, cfg in EXTRA_UNITS.items():
        anchor = cfg["anchor"]
        if anchor not in onboarded:
            continue
        w = waves[anchor] + (1 if cfg.get("above") else 0)
        for f in cfg["files"]:
            add(w, f)

    return {w: sorted(fs) for w, fs in plan.items()}


def wave_kustomization_text(wave: int, files: list) -> str:
    """The kustomize Kustomization for one wave directory (apps/wave-<N>/kustomization.yaml)."""
    dep = "platform-config" if wave == 0 else f"apps-wave-{wave - 1}"
    header = (
        f"# Wave {wave} of the app tier — GENERATED by `catalog/render.py --emit-waves`.\n"
        f"# Reconciled by the Flux Kustomization `apps-wave-{wave}` "
        f"(clusters/base/apps-wave-{wave}.yaml), which dependsOn `{dep}` and health-gates\n"
        f"# (wait:true) ONLY this wave — a crashloop here cannot stall an earlier wave.\n"
        f"# DO NOT hand-edit: re-run `render.py --emit-waves` after onboarding a service."
    )
    body = {
        "apiVersion": "kustomize.config.k8s.io/v1beta1",
        "kind": "Kustomization",
        "resources": list(files),
    }
    dumped = yaml.dump(body, sort_keys=False, default_flow_style=False, width=4096)
    return f"---\n{header}\n{dumped}"


def flux_wave_kustomization_text(wave: int) -> str:
    """The Flux Kustomization CR for one wave (clusters/base/apps-wave-<N>.yaml).

    dependsOn wave N-1 (wave 0 dependsOn platform-config); prune+wait match the retired
    monolith. SOPS decryption is added by the environment overlay patch, IDENTICALLY to the
    former single `apps` Kustomization (clusters/dev patches each apps-wave-<N> by name) — the
    base stays env-agnostic and references the out-of-band `sops-age` Secret by name only.
    """
    dep = "platform-config" if wave == 0 else f"apps-wave-{wave - 1}"
    header = (
        f"# Flux Kustomization: app-tier WAVE {wave}. Reconciles ./apps/wave-{wave} AFTER "
        f"`{dep}` is healthy.\n"
        f"# GENERATED by `catalog/render.py --emit-waves`. Waves isolate a crashloop to its "
        f"own wave + downstream:\n"
        f"# wait:true health-gates ONLY this wave's resources, so a later wave can no longer "
        f"stall an earlier one.\n"
        f"# SOPS decryption is layered on per-environment (clusters/dev patches this "
        f"Kustomization by name), identical to the retired `apps` Kustomization."
    )
    cr = {
        "apiVersion": "kustomize.toolkit.fluxcd.io/v1",
        "kind": "Kustomization",
        "metadata": {"name": f"apps-wave-{wave}", "namespace": "flux-system"},
        "spec": {
            "interval": "10m",
            "retryInterval": "2m",
            "timeout": "5m",
            "prune": True,
            "wait": True,
            "dependsOn": [{"name": dep}],
            "sourceRef": {"kind": "GitRepository", "name": "flux-system"},
            "path": f"./apps/wave-{wave}",
            "postBuild": {
                "substituteFrom": [{"kind": "ConfigMap", "name": "cluster-vars"}]
            },
        },
    }
    dumped = yaml.dump(cr, sort_keys=False, default_flow_style=False, width=4096)
    return f"---\n{header}\n{dumped}"


def _discover_app_files(apps_dir: Path) -> set:
    """Basenames of every committed app-tier manifest under apps/ (flat OR in wave-<N>/).

    Excludes the generated per-wave kustomization.yaml and any README. Deterministic.
    """
    present = set()
    for p in apps_dir.rglob("*.yaml"):
        if p.name == "kustomization.yaml":
            continue
        present.add(p.name)
    return present


def emit_waves(catalog_dir: Path, apps_dir: Path, clusters_dir: Path) -> dict:
    """Write apps/wave-<N>/kustomization.yaml + clusters/base/apps-wave-<N>.yaml.

    Computes the wave plan from the catalog DAG and the app files present under apps/, then
    (re)generates the per-wave kustomize files and the Flux Kustomization CRs. Idempotent:
    it never moves or rewrites a service manifest (only its placement changes, done in git),
    so the rendered bytes — and the L1 goldens — are untouched. Returns the wave plan.
    """
    sds = _load_catalog(catalog_dir)
    present = _discover_app_files(apps_dir)
    plan = plan_apps_waves(sds, present)
    for wave in sorted(plan):
        wdir = apps_dir / f"wave-{wave}"
        wdir.mkdir(parents=True, exist_ok=True)
        (wdir / "kustomization.yaml").write_text(
            wave_kustomization_text(wave, plan[wave])
        )
        (clusters_dir / f"apps-wave-{wave}.yaml").write_text(
            flux_wave_kustomization_text(wave)
        )
    return plan


# =============================================================================
# Self-test (deterministic, no cluster) — `render.py --selftest`
# =============================================================================
def _selftest() -> int:
    failures = []

    def check(cond, msg):
        if not cond:
            failures.append(msg)

    # 1) A minimal container SD renders the mandatory contract invariants.
    sd = {"name": "demo", "source": {"image": "nginx:1"}}
    docs = render(sd)
    kinds = [d[1]["kind"] for d in docs]
    check("Namespace" in kinds, "container: missing Namespace")
    check("NetworkPolicy" in kinds, "container: missing default-deny NetworkPolicy")
    check("Deployment" in kinds, "container: missing Deployment")

    ns = next(d[1] for d in docs if d[1]["kind"] == "Namespace")
    check(
        ns["metadata"]["labels"].get(f"{LBL}/app-namespace") == "true",
        "namespace missing app-namespace=true (Kyverno hook)",
    )
    dep = next(d[1] for d in docs if d[1]["kind"] == "Deployment")
    plabels = dep["spec"]["template"]["metadata"]["labels"]
    check(
        plabels.get(f"{LBL}/egress-class") == "internet",
        "default egress-class must be internet (R-NET-05)",
    )
    check(
        plabels.get(f"{LBL}/auth-mode") in AUTH_MODES,
        "pod must carry a valid auth-mode label (R-IDP-02)",
    )
    ndp = next(d[1] for d in docs if d[1]["kind"] == "NetworkPolicy")
    check(
        ndp["spec"]["policyTypes"] == ["Ingress", "Egress"],
        "default-deny must cover both directions (R-NET-01)",
    )
    check(ndp["spec"]["ingress"] == [], "default-deny must deny all ingress (R-NET-01)")

    # 2) egress: vpn flips ONLY the egress-class label (R-NET-07).
    internet_docs = render({"name": "x", "source": {"image": "a:1"}, "egress": "internet"})
    vpn_docs = render({"name": "x", "source": {"image": "a:1"}, "egress": "vpn"})
    i_pod = next(d[1] for d in internet_docs if d[1]["kind"] == "Deployment")
    v_pod = next(d[1] for d in vpn_docs if d[1]["kind"] == "Deployment")
    check(
        i_pod["spec"]["template"]["metadata"]["labels"][f"{LBL}/egress-class"] == "internet"
        and v_pod["spec"]["template"]["metadata"]["labels"][f"{LBL}/egress-class"] == "vpn",
        "egress class must render onto the pod label",
    )
    # Everything except the egress-class label must be identical (R-NET-07).
    i_txt = emit(internet_docs).replace("egress-class: internet", "EGRESS")
    v_txt = emit(vpn_docs).replace("egress-class: vpn", "EGRESS")
    check(i_txt == v_txt, "flipping egress class changed more than the egress path (R-NET-07)")

    # 3) Determinism: same input twice -> byte-identical output.
    check(emit(render(sd)) == emit(render(sd)), "render is not deterministic")

    # 4) dependencies -> one CiliumNetworkPolicy allow each (R-NET-02).
    dep_docs = render(
        {
            "name": "y",
            "source": {"image": "a:1"},
            "dependencies": [{"to": "db", "port": 5432}],
        }
    )
    cnps = [d[1] for d in dep_docs if d[1]["kind"] == "CiliumNetworkPolicy"]
    dep_cnps = [c for c in cnps if c["metadata"]["name"] == "y-allow-db-5432"]
    check(len(dep_cnps) == 1, "one dependency must render one CiliumNetworkPolicy allow")
    if dep_cnps:
        to_ep = dep_cnps[0]["spec"]["egress"][0]["toEndpoints"][0]["matchLabels"]
        check(
            to_ep.get("k8s:io.kubernetes.pod.namespace") == "db"
            and to_ep.get("app.kubernetes.io/name") == "db",
            "egress allow must pin the destination namespace (cross-ns Cilium selector)",
        )
    check(
        any(c["metadata"]["name"] == "y-allow-internet-egress" for c in cnps),
        "every app (egress class) must render a world/internet egress allow",
    )

    # 4a) reciprocal ingress: a dependency declared ONCE on the SOURCE must render the
    #     matching ingress-allow on the DESTINATION (R-NET-02). Without it, default-deny-
    #     ingress drops the source and the declared dependency does not actually work.
    src_sd = {
        "name": "src",
        "source": {"image": "a:1"},
        "dependencies": [{"to": "dst", "port": 8080, "protocol": "tcp"}],
    }
    dst_sd = {"name": "dst", "source": {"image": "b:1"}}
    rev = build_reverse_map([src_sd, dst_sd])
    check(
        rev.get("dst") == [{"source": "src", "port": 8080, "protocol": "tcp"}],
        "reverse map must record src->dst:8080 under the destination",
    )
    check("src" not in rev, "a service nothing depends on must have no inbound edges")
    dst_docs = render(dst_sd, inbound=rev.get("dst", []))
    rev_cnps = [
        d[1]
        for d in dst_docs
        if d[1]["kind"] == "CiliumNetworkPolicy"
        and d[1]["metadata"]["name"] == "dst-allow-src-8080"
    ]
    check(len(rev_cnps) == 1, "destination must render one reciprocal ingress allow")
    if rev_cnps:
        ing = rev_cnps[0]["spec"]["ingress"]
        from_ep = ing[0]["fromEndpoints"][0]["matchLabels"]
        check(
            from_ep.get("app.kubernetes.io/name") == "src"
            and from_ep.get("k8s:io.kubernetes.pod.namespace") == "src",
            "reciprocal ingress must be from the SOURCE's identity, namespace-pinned (cross-ns)",
        )
        allowed = {p["port"] for r in ing for tp in r["toPorts"] for p in tp["ports"]}
        check(allowed == {"8080"}, "reciprocal ingress must open only the declared port")
    # additive: default-deny ingress stays [] even with the reciprocal allow present.
    dst_dd = next(d[1] for d in dst_docs if d[1]["kind"] == "NetworkPolicy")
    check(
        dst_dd["spec"]["ingress"] == [],
        "reciprocal ingress must be ADDITIVE — default-deny ingress stays [] (R-NET-01)",
    )
    # a lone render (no inbound) must NOT gain any reciprocal ingress (golden-stable).
    lone_docs = render(dst_sd)
    check(
        not any(
            d[1]["kind"] == "CiliumNetworkPolicy"
            and d[1]["metadata"]["name"].startswith("dst-allow-src")
            for d in lone_docs
        ),
        "a render with no inbound context must be unchanged (no reciprocal ingress)",
    )

    # 4b) exposure (ingressRoutes) -> an ADDITIVE gateway-ingress CiliumNetworkPolicy that
    #     allows ONLY the reserved `ingress` (Gateway proxy) identity to the exposed port,
    #     while default-deny still denies all other ingress (R-NET-01/02).
    exp_docs = render(
        {
            "name": "z",
            "source": {"image": "a:1"},
            "ingressRoutes": [{"host": "z.home.keithjackson.dev", "port": 80}],
        }
    )
    gw = [
        d[1]
        for d in exp_docs
        if d[1]["kind"] == "CiliumNetworkPolicy"
        and d[1]["metadata"]["name"].endswith("-allow-gateway-ingress")
    ]
    check(len(gw) == 1, "an exposed service must render one gateway-ingress allow (R-NET-01)")
    if gw:
        ing = gw[0]["spec"]["ingress"]
        check(
            ing[0]["fromEntities"] == ["ingress"],
            "gateway-ingress allow must be from the reserved `ingress` (Gateway) identity",
        )
        allowed_ports = {p["port"] for r in ing for tp in r["toPorts"] for p in tp["ports"]}
        check(
            allowed_ports == {"80"},
            "gateway-ingress allow must be least-privilege (only the exposed port)",
        )
    # default-deny stays deny-all-ingress even when an exposed service adds the allow.
    exp_dd = next(d[1] for d in exp_docs if d[1]["kind"] == "NetworkPolicy")
    check(
        exp_dd["spec"]["ingress"] == [],
        "gateway-ingress allow must be ADDITIVE — default-deny ingress stays [] (R-NET-01)",
    )
    # an internal (no ingressRoutes) service must NOT get a gateway-ingress allow.
    internal_docs = render({"name": "int", "source": {"image": "a:1"}})
    check(
        not any(
            d[1]["kind"] == "CiliumNetworkPolicy"
            and d[1]["metadata"]["name"].endswith("-allow-gateway-ingress")
            for d in internal_docs
        ),
        "a non-exposed service must not render a gateway-ingress allow (least privilege)",
    )

    # 5) vm kind -> VirtualMachine + standalone DataVolume.
    vm_docs = render(
        {
            "name": "vmx",
            "kind": "vm",
            "source": {"vmTemplate": "quay.io/containerdisks/fedora:41"},
            "storage": [{"name": "rootdisk", "size": "6Gi", "class": "longhorn"}],
        }
    )
    vm_kinds = [d[1]["kind"] for d in vm_docs]
    check("VirtualMachine" in vm_kinds, "vm: missing VirtualMachine")
    check("DataVolume" in vm_kinds, "vm: missing CDI DataVolume")

    # 6) env / envFrom / lifecycle -> emitted VERBATIM onto the app container (T-APPS).
    envd_docs = render(
        {
            "name": "envd",
            "source": {"image": "a:1"},
            "env": [
                {"name": "PLAIN", "value": "literal"},
                {
                    "name": "SECRET",
                    "valueFrom": {"secretKeyRef": {"name": "sec", "key": "k"}},
                },
            ],
            "envFrom": [{"secretRef": {"name": "bulk"}}],
            "lifecycle": {
                "postStart": {"exec": {"command": ["/bin/sh", "-c", "echo hi"]}}
            },
            "podSecurityContext": {"fsGroup": 1001, "runAsUser": 1001},
        }
    )
    envd_pod = next(
        d[1] for d in envd_docs if d[1]["kind"] == "Deployment"
    )["spec"]["template"]["spec"]
    check(
        envd_pod.get("securityContext") == {"fsGroup": 1001, "runAsUser": 1001},
        "podSecurityContext must render verbatim onto the pod spec",
    )
    ec = envd_pod["containers"][0]
    check(
        ec.get("env") == [
            {"name": "PLAIN", "value": "literal"},
            {"name": "SECRET", "valueFrom": {"secretKeyRef": {"name": "sec", "key": "k"}}},
        ],
        "env must render verbatim onto the container (literal + secretKeyRef)",
    )
    check(
        ec.get("envFrom") == [{"secretRef": {"name": "bulk"}}],
        "envFrom must render verbatim onto the container",
    )
    check(
        ec.get("lifecycle")
        == {"postStart": {"exec": {"command": ["/bin/sh", "-c", "echo hi"]}}},
        "lifecycle must render verbatim onto the container",
    )
    # Absence => the keys are never set (existing apps stay byte-identical).
    plain_c = next(
        d[1] for d in render({"name": "plain", "source": {"image": "a:1"}})
        if d[1]["kind"] == "Deployment"
    )["spec"]["template"]["spec"]["containers"][0]
    check(
        "env" not in plain_c
        and "envFrom" not in plain_c
        and "lifecycle" not in plain_c,
        "an SD without env/envFrom/lifecycle must not gain those container keys",
    )
    plain_pod = next(
        d[1] for d in render({"name": "plain2", "source": {"image": "a:1"}})
        if d[1]["kind"] == "Deployment"
    )["spec"]["template"]["spec"]
    check(
        "securityContext" not in plain_pod,
        "an SD without podSecurityContext must not gain a pod securityContext",
    )

    # 7) wave partitioning: longest-path over the dependency DAG, CNPG-DB strictly below.
    wsds = [
        {"name": "a", "source": {"image": "x:1"}},                                    # leaf
        {"name": "b", "source": {"image": "x:1"}},                                    # leaf
        {"name": "c", "source": {"image": "x:1"},
         "dependencies": [{"to": "a", "port": 1}, {"to": "b", "port": 1}]},           # wave 1
        {"name": "d", "source": {"image": "x:1"},
         "dependencies": [{"to": "c", "port": 1}]},                                   # wave 2
        {"name": "svc", "source": {"image": "x:1"},
         "dependencies": [{"to": "svc-db", "port": 5432}]},                           # DB consumer
    ]
    w = compute_waves(wsds)
    check(w["a"] == 0 and w["b"] == 0, "leaf services must be wave 0")
    check(w["c"] == 1, "a service depending on wave-0 nodes must be wave 1")
    check(w["d"] == 2, "longest-path depth must be transitive (d -> c -> {a,b})")
    check(
        w["svc-db"] == 0 and w["svc"] == 1 and w["svc-db"] < w["svc"],
        "a co-located <svc>-db must be a leaf STRICTLY BELOW its consumer (CNPG-DB rule)",
    )
    # cycle detection.
    try:
        compute_waves([
            {"name": "p", "source": {"image": "x:1"}, "dependencies": [{"to": "q", "port": 1}]},
            {"name": "q", "source": {"image": "x:1"}, "dependencies": [{"to": "p", "port": 1}]},
        ])
        check(False, "compute_waves must raise on a dependency cycle")
    except ValueError:
        pass

    # 7a) plan_apps_waves: sidecars inherit their unit's wave; not-onboarded SDs are skipped;
    #     EXTRA_UNITS anchor strictly above; a file absent from the tree is never placed.
    present = {
        "a.yaml", "svc.yaml", "svc-bootstrap-secret.sops.yaml",
        "svc-db.yaml", "svc-db-app.sops.yaml", "svc-db-backup-credentials.sops.yaml",
        "gitea.yaml",
        "gitea-actions.yaml", "gitea-actions-runner-secret.sops.yaml",
    }
    plan_sds = wsds + [{"name": "gitea", "source": {"image": "x:1"}},
                       {"name": "notonboarded", "source": {"image": "x:1"}}]
    plan = plan_apps_waves(plan_sds, present)
    flat = {f: wv for wv, fs in plan.items() for f in fs}
    check(flat.get("a.yaml") == 0, "leaf app must land in wave 0")
    check(flat.get("svc-db.yaml") == 0 and flat.get("svc-db-app.sops.yaml") == 0
          and flat.get("svc-db-backup-credentials.sops.yaml") == 0,
          "CNPG DB manifest + its owner secret + its Barman backup secret must be in the DB's (lower) wave")
    check(flat.get("svc.yaml") == 1 and flat.get("svc-bootstrap-secret.sops.yaml") == 1,
          "a DB-backed app + its bootstrap secret must be in the consumer wave, above the DB")
    check(flat.get("gitea.yaml") == 0, "gitea (leaf) must be wave 0")
    check(flat.get("gitea-actions.yaml") == 1
          and flat.get("gitea-actions-runner-secret.sops.yaml") == 1,
          "gitea-actions (EXTRA_UNITS, above=True) must be one wave above its anchor gitea")
    check("notonboarded.yaml" not in flat,
          "a catalog SD with no committed manifest must NOT be placed (not onboarded)")
    check("b.yaml" not in flat, "a file absent from the tree must never be placed")

    # 7b) generated Flux Kustomization CR wiring (dependsOn chain + prune/wait/path).
    import yaml as _yaml  # already imported at module top; local alias for clarity
    fw0 = _yaml.safe_load(flux_wave_kustomization_text(0))
    fw2 = _yaml.safe_load(flux_wave_kustomization_text(2))
    check(fw0["metadata"]["name"] == "apps-wave-0"
          and fw0["spec"]["dependsOn"] == [{"name": "platform-config"}]
          and fw0["spec"]["path"] == "./apps/wave-0",
          "wave 0 Flux Kustomization must dependsOn platform-config and point at ./apps/wave-0")
    check(fw2["spec"]["dependsOn"] == [{"name": "apps-wave-1"}]
          and fw2["spec"]["prune"] is True and fw2["spec"]["wait"] is True,
          "wave N Flux Kustomization must dependsOn wave N-1 with prune+wait (health-gated)")
    kt = _yaml.safe_load(wave_kustomization_text(1, ["svc.yaml", "sonarr.yaml"]))
    check(kt["kind"] == "Kustomization" and kt["resources"] == ["svc.yaml", "sonarr.yaml"],
          "wave kustomization must list exactly the wave's resources")

    if failures:
        for f in failures:
            sys.stderr.write(f"SELFTEST FAIL: {f}\n")
        return 1
    sys.stderr.write("render.py selftest: PASS\n")
    return 0


def main(argv=None) -> int:
    ap = argparse.ArgumentParser(description="Render a ServiceDefinition to manifests.")
    ap.add_argument("input", nargs="?", help="ServiceDefinition YAML file")
    ap.add_argument("--all", metavar="DIR", help="Render every *.yaml ServiceDefinition in DIR")
    ap.add_argument("-o", "--out-dir", help="With --all: write <name>.yaml files into this dir")
    ap.add_argument("--selftest", action="store_true", help="Run the built-in self-test")
    ap.add_argument("--no-image-automation", action="store_true",
                    help="Omit Flux ImageRepository/ImagePolicy CRs (for onboarding before the "
                         "image-reflector/-automation controllers are installed)")
    ap.add_argument("--wave-layout", action="store_true",
                    help="With -o: place each rendered file under <out>/wave-<N>/<name>.yaml, "
                         "N = its dependency-ordered wave (compute_waves over the catalog)")
    ap.add_argument("--emit-waves", action="store_true",
                    help="Generate the per-wave kustomize + Flux Kustomization CRs from the "
                         "catalog DAG and the committed apps tree (no manifest is rewritten)")
    ap.add_argument("--print-waves", action="store_true",
                    help="Print the computed wave table (node -> wave) for the catalog and exit")
    ap.add_argument("--catalog", default="catalog",
                    help="Catalog dir for --emit-waves/--print-waves/--wave-layout (default: catalog)")
    ap.add_argument("--apps-dir", default="apps",
                    help="App-tier dir for --emit-waves (default: apps)")
    ap.add_argument("--clusters-dir", default="clusters/base",
                    help="Dir to write clusters/base/apps-wave-<N>.yaml (default: clusters/base)")
    args = ap.parse_args(argv)

    if args.selftest:
        return _selftest()

    if args.print_waves:
        waves = compute_waves(_load_catalog(Path(args.catalog)))
        for node in sorted(waves, key=lambda n: (waves[n], n)):
            sys.stdout.write(f"wave {waves[node]}  {node}\n")
        return 0

    if args.emit_waves:
        plan = emit_waves(Path(args.catalog), Path(args.apps_dir), Path(args.clusters_dir))
        for wave in sorted(plan):
            sys.stderr.write(f"[emit-waves] wave {wave}: {', '.join(plan[wave])}\n")
        sys.stderr.write(
            f"[emit-waves] wrote {len(plan)} wave kustomization(s) under {args.apps_dir}/ "
            f"and clusters CRs under {args.clusters_dir}/\n"
        )
        return 0

    image_automation = not args.no_image_automation

    def _out_path(outdir: Path, name: str, sds: list) -> Path:
        if args.wave_layout:
            wdir = outdir / f"wave-{service_wave(sds, name)}"
            wdir.mkdir(parents=True, exist_ok=True)
            return wdir / f"{name}.yaml"
        outdir.mkdir(parents=True, exist_ok=True)
        return outdir / f"{name}.yaml"

    if args.all:
        src = Path(args.all)
        files = sorted(p for p in src.glob("*.yaml") if p.name != "render.py")
        sds = _load_catalog(src) if args.wave_layout else []
        for f in files:
            text = render_file(f, image_automation=image_automation)
            if args.out_dir:
                _out_path(Path(args.out_dir), load_sd(f)["name"], sds).write_text(text)
            else:
                sys.stdout.write(text)
        return 0

    if not args.input:
        ap.error("provide a ServiceDefinition file, --all DIR, --emit-waves, or --selftest")
    inp = Path(args.input)
    text = render_file(inp, image_automation=image_automation)
    if args.out_dir:
        sds = _load_catalog(Path(args.catalog)) if args.wave_layout else []
        _out_path(Path(args.out_dir), load_sd(inp)["name"], sds).write_text(text)
    else:
        sys.stdout.write(text)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())