#!/usr/bin/env bash

# This file is intentionally sourceable. Assertions have no cluster or network side
# effects; executable rendering is guarded at the bottom of the file.

observability_expected_helm_version() { printf '%s\n' 'v3.19.4'; }
observability_expected_kustomize_version() { printf '%s\n' 'v5.8.1'; }
observability_helm_kube_version() { printf '%s\n' '1.36.2'; }

chart_pin_rows() {
  cat <<'ROWS'
prometheus-operator-crds	https://prometheus-community.github.io/helm-charts	prometheus-operator-crds	31.0.0	v0.93.0	5d11cac50321bc7ae56b35d2701e56f21fe079ca3aea872abb28a3e2119d5cfd	infrastructure/controllers/prometheus-operator-crds
kube-prometheus-stack	https://prometheus-community.github.io/helm-charts	kube-prometheus-stack	88.0.1	v0.93.0	0998b3fae8660cd9ea224caa1c7c4b5d2040d5a02c64e3cdfd3f48d184b50117	services/observability/kube-prometheus-stack
loki	https://grafana-community.github.io/helm-charts	loki	18.7.1	3.7.4	de9ab59827f2ed24634e94db120ea2467f5d08b02e8d7c250e6bf88c17bfb48d	services/observability/loki
tempo	https://grafana-community.github.io/helm-charts	tempo	2.2.3	2.10.7	14c52efe5d0cad5456ffa5a8be1e5107be47d184be3e209f748be51a7b0316fe	services/observability/tempo
alloy	https://grafana.github.io/helm-charts	alloy	1.11.0	v1.18.0	11d253b62e47beeacd89eb4283fc056962ecbf143984863c1998be13da0772dd	services/observability/alloy
prometheus-node-exporter	https://prometheus-community.github.io/helm-charts	prometheus-node-exporter	4.56.1	1.12.1	70fab291e00a34c58fd943bb075fecc2e2e73fa95db1bcbbe7cdfe008795354a	services/observability/node-exporter
grafana	https://grafana-community.github.io/helm-charts	grafana	12.10.1	13.1.1	ff865a8b9d1d7563901b8e6aa653b4f345bd4d88344c149065bd23958e2beeb8	services/observability/grafana
prometheus-blackbox-exporter	https://prometheus-community.github.io/helm-charts	prometheus-blackbox-exporter	11.16.0	v0.28.0	932aa65df0538d9dc003c46bb663ebe44a06078af7a6a54577af5339668ac65b	services/observability/blackbox-exporter
ROWS
}

image_pin_rows() {
  cat <<'ROWS'
quay.io/prometheus-operator/prometheus-operator:v0.93.0	sha256:64eb7914e4705dbb64438e3b3193da1226ad2ea4db2924983693999888cda9b2
quay.io/prometheus-operator/prometheus-config-reloader:v0.93.0	sha256:65b90f44d5535b106015ac60bafb57803f65dc928c187874de6cd7a9ec6c8905
quay.io/prometheus/prometheus:v3.13.2-distroless	sha256:ce95cfa77eff5aad28bd7a65aff19868cf78d9e17e4c254da7dfe22ade78318b
quay.io/prometheus/alertmanager:v0.33.1	sha256:a89f8d4520954079275441eecdb71444328bd90633dd4eddfc33b9ed657f349b
registry.k8s.io/kube-state-metrics/kube-state-metrics:v2.19.1	sha256:7661da8c99b733d43117e4cba12bd9865d335e5777191d0af3d789807aded9f4
quay.io/prometheus/node-exporter:v1.12.1	sha256:da83fae85603c4e47e6c68369a7d746e2dda683dc35ea2e234b4f171e0d92798
docker.io/grafana/loki:3.7.4	sha256:d80be589cf74b6d92747a1b155bec6f07bdd65e261f797fe93a78fb4e833032e
docker.io/grafana/tempo:2.10.7	sha256:6616b00287a4d7001951b5de117828ad5c6f93744935c1b7a5e044736373352c
docker.io/grafana/alloy:v1.18.0	sha256:eb21f4c0858edffcdd1b385910ddeef26f692fc2c282f61baa724fc09d274a17
ghcr.io/open-telemetry/opentelemetry-collector-contrib/telemetrygen:v0.153.0	sha256:9708a1d33398c8725549104d1f6009be62569e47e094430543da0f66758cc522
docker.io/library/busybox:1.37.0	sha256:7a3ebe5bfd1a4a19797d20b0c0bb39d44393e9a03fd852c0865b0f540d868df0
quay.io/minio/aistor/mc	sha256:c5ec777c080fd6292b7529309a0f311c58c3a92e268c9c57c0901da86881949b
docker.io/grafana/grafana:13.1.1	sha256:f33c692ba1a5ee15724cf6b22db65e9de39dde14d80f7d73a9546e3fc917270b
quay.io/kiwigrid/k8s-sidecar:2.10.0	sha256:129877c81acf2bc8c3fa000e89a62e020eb89d41ceb94767c657aef5bb0cc0d3
quay.io/prometheus/blackbox-exporter:v0.28.0	sha256:43027b43fb785b7c5adc53bd3b5dbc1a258270a2e8aff24f477b45c4e38dac68
ROWS
}

render_root_rows() {
  cat <<'ROWS'
infrastructure/namespaces/observability
infrastructure/controllers/prometheus-operator-crds
infrastructure/storage/observability-local-pv
services/observability/kube-prometheus-stack
services/observability/loki
services/observability/tempo
services/observability/alloy
services/observability/node-exporter
services/observability/grafana
services/observability/blackbox-exporter
services/observability/network-policies
services/observability/targets
services/observability/rules/platform
services/observability/alerting
services/observability/dashboards/platform
services/observability
ROWS
}

_observability_image_allowlist() {
  image_pin_rows | awk -F '\t' '
    {
      tagged=$1
      base=$1
      sub(/:[^/:]+$/, "", base)
      print base "@" $2
      if (tagged != base) {
        print tagged "@" $2
      }
    }
  '
}

_observability_yaml_assert() {
  local mode="$1"
  shift
  python3 - "$mode" "$@" <<'PY'
import base64
import binascii
import json
import os
import re
import sys

import yaml


class ObservabilitySafeLoader(yaml.SafeLoader):
    pass


# The pinned Prometheus Operator CRDs contain the valid scalar enum value "=".
# PyYAML resolves that token to its YAML value tag but does not provide a safe
# constructor for it, so accept only that scalar tag without enabling unsafe
# object construction.
ObservabilitySafeLoader.add_constructor(
    "tag:yaml.org,2002:value",
    lambda loader, node: loader.construct_scalar(node),
)


def reject(reason):
    print(f"REJECT: observability render assertion failed: {reason}", file=sys.stderr)
    raise SystemExit(23)


def documents(paths):
    result = []
    try:
        for path in paths:
            with open(path, "r", encoding="utf-8") as stream:
                for value in yaml.load_all(stream, Loader=ObservabilitySafeLoader):
                    if value is not None:
                        if not isinstance(value, dict):
                            reject("YAML document is not an object")
                        result.append(value)
    except (OSError, yaml.YAMLError):
        reject("manifest is unreadable or invalid YAML")
    return result


def pod_specs(items):
    def inherited_metadata(parent, template):
        result = dict(template or {})
        for key in ("name", "namespace"):
            if not result.get(key) and (parent or {}).get(key):
                result[key] = parent[key]
        return result

    for item in items:
        kind = str(item.get("kind", ""))
        metadata = item.get("metadata") or {}
        spec = item.get("spec") or {}
        if kind == "Pod":
            yield metadata, spec
        elif kind in {"Deployment", "DaemonSet", "StatefulSet", "Job", "ReplicaSet", "ReplicationController"}:
            template = spec.get("template") or {}
            yield inherited_metadata(metadata, template.get("metadata")), template.get("spec") or {}
        elif kind == "CronJob":
            template = (((spec.get("jobTemplate") or {}).get("spec") or {}).get("template") or {})
            yield inherited_metadata(metadata, template.get("metadata")), template.get("spec") or {}


def walk(value):
    yield value
    if isinstance(value, dict):
        for child in value.values():
            yield from walk(child)
    elif isinstance(value, list):
        for child in value:
            yield from walk(child)


mode = sys.argv[1]
paths = sys.argv[2:]
items = documents(paths)

if mode == "pinned_images":
    allowed = set(filter(None, os.environ.get("OBSERVABILITY_ALLOWED_IMAGES", "").splitlines()))
    if not allowed:
        reject("image pin table is empty")
    images = []
    for _, spec in pod_specs(items):
        for key in ("initContainers", "containers", "ephemeralContainers"):
            for container in spec.get(key) or []:
                image = container.get("image")
                if not isinstance(image, str) or not image:
                    reject("container image is missing")
                images.append(image)
    for image in images:
        if not re.fullmatch(r"[^\s@]+@sha256:[0-9a-f]{64}", image):
            reject("an image is not pinned by amd64 digest")
        if image not in allowed:
            reject("an image is absent from the fixed image table")

elif mode == "no_credentials":
    sensitive_name = re.compile(r"(?:password|passwd|secret|token|access[_-]?key|secret[_-]?key|private[_-]?key)", re.I)
    safe_reference = re.compile(r"^(?:\$\{[^}]+\}|\$\([^)]*\)|<[^>]+>)$")
    for item in items:
        if item.get("kind") == "Secret":
            metadata = item.get("metadata") or {}
            data = item.get("data") or {}
            string_data = item.get("stringData") or {}
            allowed_alertmanager_config = (
                metadata.get("namespace") == "observability"
                and metadata.get("name") == "alertmanager-observability-core-kube-pr-alertmanager"
                and item.get("type") in (None, "Opaque")
                and set(data) == {"alertmanager.yaml"}
                and not string_data
            )
            if not allowed_alertmanager_config:
                reject("Secret resources are forbidden")
            try:
                raw_config = base64.b64decode(data["alertmanager.yaml"], validate=True)
                parsed_config = yaml.load(raw_config, Loader=ObservabilitySafeLoader)
            except (binascii.Error, UnicodeError, yaml.YAMLError):
                reject("allowed Alertmanager config Secret is malformed")
            if not isinstance(parsed_config, dict):
                reject("allowed Alertmanager config Secret is not an object")
            receivers = parsed_config.get("receivers")
            route = parsed_config.get("route") or {}
            if receivers != [{"name": "null"}] or route.get("receiver") != "null" or route.get("routes") != []:
                reject("allowed Alertmanager config Secret is not credential-free")
            if re.search(
                rb"(?i)(?:password|passwd|secret|token|api[_-]?url|webhook|access[_-]?key|private[_-]?key)",
                raw_config,
            ):
                reject("allowed Alertmanager config Secret contains a credential field")
        for _, spec in pod_specs([item]):
            for key in ("initContainers", "containers", "ephemeralContainers"):
                for container in spec.get(key) or []:
                    for env in container.get("env") or []:
                        if sensitive_name.search(str(env.get("name", ""))) and "value" in env:
                            reject("literal credential environment value is forbidden")
        for node in walk(item):
            if not isinstance(node, dict):
                continue
            for key, value in node.items():
                if not sensitive_name.fullmatch(str(key)) or not isinstance(value, str) or not value:
                    continue
                if not safe_reference.fullmatch(value):
                    reject("literal credential payload is forbidden")

elif mode == "namespace_security":
    for metadata, spec in pod_specs(items):
        namespace = str(metadata.get("namespace", ""))
        if namespace not in {"observability", "observability-agent"}:
            reject("Pod namespace must be observability or observability-agent")
        automount = spec.get("automountServiceAccountToken")
        api_reader_allowlist = {
            ("observability", "observability-core-kube-pr-operator"),
            ("observability", "observability-core-kube-state-metrics"),
        }
        identity = (namespace, str(metadata.get("name", "")))
        if identity in api_reader_allowlist:
            if automount is not True:
                reject("allowlisted API reader must explicitly mount its service-account token")
        elif automount is not False:
            reject("Pod service-account automount must be explicit false")
        for field in ("hostNetwork", "hostPID", "hostIPC"):
            if spec.get(field) is True:
                reject(f"{field} is forbidden")
        containers = []
        for key in ("initContainers", "containers", "ephemeralContainers"):
            containers.extend(spec.get(key) or [])
        for container in containers:
            if (container.get("securityContext") or {}).get("privileged") is True:
                reject("privileged containers are forbidden")
            for port in container.get("ports") or []:
                if port.get("hostPort") not in (None, 0):
                    reject("hostPort is forbidden")
        host_volumes = {}
        for volume in spec.get("volumes") or []:
            host_path = volume.get("hostPath")
            if isinstance(host_path, dict):
                host_volumes[str(volume.get("name", ""))] = str(host_path.get("path", ""))
        if namespace == "observability" and host_volumes:
            reject("hostPath is forbidden in observability")
        if namespace == "observability-agent":
            identity_parts = [str(metadata.get("name", ""))]
            identity_parts.extend(str(v) for v in (metadata.get("labels") or {}).values())
            identity = " ".join(identity_parts).lower()
            if "alloy" in identity:
                allowed_paths = {"/var/log/pods"}
            elif "node-exporter" in identity or "node_exporter" in identity:
                allowed_paths = {"/proc", "/sys", "/"}
            else:
                allowed_paths = set()
            for volume_name, host_path in host_volumes.items():
                if host_path not in allowed_paths:
                    reject("agent hostPath is outside the workload allowlist")
                mounts = [
                    mount
                    for container in containers
                    for mount in (container.get("volumeMounts") or [])
                    if mount.get("name") == volume_name
                ]
                if not mounts or any(mount.get("readOnly") is not True for mount in mounts):
                    reject("agent hostPath mounts must be read-only")

elif mode == "no_public_exposure":
    for item in items:
        kind = str(item.get("kind", ""))
        if kind == "Ingress":
            metadata = item.get("metadata") or {}
            spec = item.get("spec") or {}
            expected_spec = {
                "ingressClassName": "traefik",
                "rules": [
                    {
                        "host": "grafana.learn.hyeonworks.com",
                        "http": {
                            "paths": [
                                {
                                    "backend": {
                                        "service": {
                                            "name": "grafana",
                                            "port": {"number": 80},
                                        }
                                    },
                                    "path": "/",
                                    "pathType": "Prefix",
                                }
                            ]
                        },
                    }
                ],
            }
            if (
                metadata.get("namespace") != "observability"
                or metadata.get("name") != "grafana"
                or spec != expected_spec
            ):
                reject("only the exact Host-Nginx-fronted Grafana Ingress is allowed")
        if kind == "Service":
            service_type = str((item.get("spec") or {}).get("type", "ClusterIP"))
            if service_type in {"NodePort", "LoadBalancer", "ExternalName"}:
                reject("non-ClusterIP Service is forbidden")
            spec = item.get("spec") or {}
            if spec.get("externalIPs") or spec.get("externalName"):
                reject("external Service routing is forbidden")

elif mode == "no_loki_auxiliary":
    forbidden = re.compile(r"(?:^|[-_.])(gateway|canary|chunks-cache|results-cache|test|minio)(?:$|[-_.])", re.I)
    for item in items:
        metadata = item.get("metadata") or {}
        identity = " ".join([str(metadata.get("name", ""))] + [str(v) for v in (metadata.get("labels") or {}).values()])
        if forbidden.search(identity) or str(item.get("kind", "")).lower() == "minio":
            reject("forbidden Loki auxiliary resource is present")

elif mode == "tempo_topology":
    forbidden = re.compile(r"(?:metrics-generator|tempo-query|query-frontend)", re.I)
    for item in items:
        metadata = item.get("metadata") or {}
        identity = " ".join([str(metadata.get("name", ""))] + [str(v) for v in (metadata.get("labels") or {}).values()])
        if forbidden.search(identity):
            reject("forbidden Tempo auxiliary resource is present")

    def inspect_config(value):
        if isinstance(value, dict):
            receivers = value.get("receivers")
            if isinstance(receivers, dict):
                for receiver_name, receiver in receivers.items():
                    if not str(receiver_name).lower().startswith("otlp") or not isinstance(receiver, dict):
                        continue
                    protocols = receiver.get("protocols") or {}
                    if "http" in protocols:
                        reject("Tempo OTLP HTTP receiver is forbidden")
                    if set(protocols) - {"grpc"}:
                        reject("Tempo has an unexpected OTLP protocol")
            for child in value.values():
                inspect_config(child)
        elif isinstance(value, list):
            for child in value:
                inspect_config(child)
        elif isinstance(value, str) and ("receivers:" in value or "4318" in value):
            if re.search(r"(?m)^\s*http\s*:", value) or re.search(r"(?<!\d)4318(?!\d)", value):
                reject("Tempo OTLP HTTP receiver is forbidden")
            try:
                nested = yaml.safe_load(value)
            except yaml.YAMLError:
                reject("Tempo generated config is invalid YAML")
            if isinstance(nested, (dict, list)):
                inspect_config(nested)

    for item in items:
        inspect_config(item)

elif mode == "selector_values":
    if len(items) != 1:
        reject("selector values must contain one YAML document")
    spec = (((items[0].get("prometheus") or {}).get("prometheusSpec")) or {})
    expected = (
        "serviceMonitorSelectorNilUsesHelmValues",
        "podMonitorSelectorNilUsesHelmValues",
        "probeSelectorNilUsesHelmValues",
        "ruleSelectorNilUsesHelmValues",
    )
    for key in expected:
        if spec.get(key) is not False:
            reject("Prometheus nilUsesHelmValues selectors must be explicit false")

elif mode == "prometheus_selectors":
    prometheuses = [item for item in items if item.get("kind") == "Prometheus"]
    if not prometheuses:
        reject("Prometheus resource is missing")
    expected = ("serviceMonitorSelector", "podMonitorSelector", "probeSelector", "ruleSelector")
    for prometheus in prometheuses:
        spec = prometheus.get("spec") or {}
        for key in expected:
            if key not in spec or spec.get(key) is None:
                reject("Prometheus discovery selector is missing")

elif mode == "alertmanager_routing":
    alertmanagers = [item for item in items if item.get("kind") == "Alertmanager"]
    if len(alertmanagers) != 1:
        reject("exactly one Alertmanager resource is required")
    alertmanager_spec = alertmanagers[0].get("spec") or {}
    if alertmanager_spec.get("alertmanagerConfiguration") != {"name": "platform-alertmanager"}:
        reject("Alertmanager global configuration reference is not exact")
    for forbidden in ("alertmanagerConfigSelector", "alertmanagerConfigNamespaceSelector"):
        if forbidden in alertmanager_spec:
            reject("Alertmanager discovery selectors are forbidden with the global reference")

    configs = [
        item
        for item in items
        if item.get("kind") == "AlertmanagerConfig"
        and (item.get("metadata") or {}).get("name") == "platform-alertmanager"
    ]
    expected_config = json.loads(r'''{"apiVersion":"monitoring.coreos.com/v1alpha1","kind":"AlertmanagerConfig","metadata":{"name":"platform-alertmanager","namespace":"observability","labels":{"observability.hyeonworks.com/instance":"home"}},"spec":{"route":{"receiver":"platform-slack","groupBy":["cluster","namespace","alertname","severity"],"groupWait":"30s","groupInterval":"5m","repeatInterval":"4h","routes":[{"receiver":"platform-null","matchers":[{"name":"alertname","matchType":"=","value":"InfoInhibitor"}]}]},"inhibitRules":[{"sourceMatch":[{"name":"alertname","matchType":"=","value":"InfoInhibitor"}],"targetMatch":[{"name":"severity","matchType":"=","value":"info"}],"equal":["namespace"]}],"receivers":[{"name":"platform-null"},{"name":"platform-slack","slackConfigs":[{"apiURL":{"name":"alertmanager-slack-webhook","key":"url"},"sendResolved":true,"linkNames":false,"mrkdwnIn":["text","fields"],"fallback":"{{ if eq .Status \"firing\" }}FIRING{{ else }}RESOLVED{{ end }}: {{ if .CommonLabels.severity }}{{ .CommonLabels.severity | toUpper }}{{ else }}UNKNOWN{{ end }} · {{ .CommonLabels.alertname }}","title":"{{ if eq .Status \"firing\" }}[FIRING:{{ .Alerts.Firing | len }}]{{ else }}[RESOLVED]{{ end }} {{ if .CommonLabels.severity }}{{ .CommonLabels.severity | toUpper }}{{ else }}UNKNOWN{{ end }} · {{ .CommonLabels.alertname }}","titleLink":"https://grafana.learn.hyeonworks.com/","color":"{{ if eq .Status \"resolved\" }}good{{ else if or (eq .CommonLabels.severity \"emergency\") (eq .CommonLabels.severity \"critical\") }}danger{{ else if eq .CommonLabels.severity \"warning\" }}warning{{ else }}#439FE0{{ end }}","fields":[{"title":"Status","value":"{{ .Status | toUpper }}","short":true},{"title":"Severity","value":"{{ if .CommonLabels.severity }}{{ .CommonLabels.severity | toUpper }}{{ else }}UNKNOWN{{ end }}","short":true},{"title":"Location","value":"{{ if .CommonLabels.cluster }}{{ .CommonLabels.cluster }}{{ else }}unknown-cluster{{ end }} / {{ if .CommonLabels.namespace }}{{ .CommonLabels.namespace }}{{ else }}cluster-scoped{{ end }}","short":true},{"title":"Alert count","value":"{{ len .Alerts }}","short":true}],"text":"{{ range .Alerts }}\n*Alert status:* {{ .Status | toUpper }}\n*Target:* {{ $target := .Labels.Remove $.GroupLabels.Names }}{{ if $target }}{{ range $target.SortedPairs }}{{ .Name }}={{ .Value }} {{ end }}{{ else }}unknown{{ end }}\n*Summary:* {{ with .Annotations.summary }}{{ . }}{{ else }}No summary provided{{ end }}\n*Details:* {{ with .Annotations.description }}{{ . }}{{ else }}No description provided{{ end }}\n*Started:* {{ .StartsAt.Format \"2006-01-02T15:04:05Z07:00\" }}\n{{ if eq .Status \"resolved\" }}*Ended:* {{ .EndsAt.Format \"2006-01-02T15:04:05Z07:00\" }}{{ end }}\n{{ with .Annotations.runbook_url }}*Runbook:* <{{ . }}|대응 절차 열기>{{ end }}\n\n{{ end }}\n<https://grafana.learn.hyeonworks.com/|Grafana> · <https://grafana.learn.hyeonworks.com/explore|Explore>","footer":"hyeonworks observability · Alertmanager"}]}]}}''')
    if configs != [expected_config]:
        reject("platform AlertmanagerConfig is not the exact global Slack contract")

    policies = [
        item
        for item in items
        if item.get("kind") == "NetworkPolicy"
        and (item.get("metadata") or {}).get("name")
        == "observability-allow-alertmanager-public-https"
    ]
    expected_policy = {
        "apiVersion": "networking.k8s.io/v1",
        "kind": "NetworkPolicy",
        "metadata": {
            "name": "observability-allow-alertmanager-public-https",
            "namespace": "observability",
        },
        "spec": {
            "podSelector": {
                "matchLabels": {
                    "app.kubernetes.io/name": "alertmanager",
                    "app.kubernetes.io/instance": "observability-core-kube-pr-alertmanager",
                }
            },
            "policyTypes": ["Egress"],
            "egress": [
                {
                    "to": [
                        {
                            "ipBlock": {
                                "cidr": "0.0.0.0/0",
                                "except": [
                                    "10.0.0.0/8",
                                    "100.64.0.0/10",
                                    "172.16.0.0/12",
                                    "192.168.0.0/16",
                                ],
                            }
                        }
                    ],
                    "ports": [{"protocol": "TCP", "port": 443}],
                }
            ],
        },
    }
    if policies != [expected_policy]:
        reject("Alertmanager public HTTPS policy is not exact")

elif mode == "no_crds":
    if any(item.get("kind") == "CustomResourceDefinition" for item in items):
        reject("CRD is present in a workload render")

elif mode == "unique_local_pv_storage_classes":
    seen = {}
    for item in items:
        if item.get("kind") != "PersistentVolume":
            continue
        spec = item.get("spec") or {}
        if "local" not in spec:
            continue
        storage_class = str(spec.get("storageClassName", ""))
        if not storage_class:
            reject("Local PV StorageClass is missing")
        seen[storage_class] = seen.get(storage_class, 0) + 1
    if any(count > 1 for count in seen.values()):
        reject("a Local PV StorageClass is shared")

elif mode == "no_unscoped_monitors_rules":
    monitored_kinds = {"ServiceMonitor", "PodMonitor", "Probe", "PrometheusRule"}
    forbidden = re.compile(r"(?:spring|kafka|batch|backup)", re.I)
    for item in items:
        if item.get("kind") not in monitored_kinds:
            continue
        metadata = item.get("metadata") or {}
        if forbidden.search(str(metadata.get("name", ""))):
            reject("out-of-scope monitor or rule is present")
        if item.get("kind") == "PrometheusRule":
            for node in walk((item.get("spec") or {}).get("groups") or []):
                if isinstance(node, dict):
                    for key in ("alert", "record"):
                        if forbidden.search(str(node.get(key, ""))):
                            reject("out-of-scope monitor or rule is present")

else:
    reject("unknown assertion mode")
PY
}

assert_pinned_images() {
  local allowed
  allowed="$(_observability_image_allowlist)" || return
  OBSERVABILITY_ALLOWED_IMAGES="$allowed" _observability_yaml_assert pinned_images "$@"
}

assert_no_credentials() { _observability_yaml_assert no_credentials "$@"; }
assert_namespace_pod_security() { _observability_yaml_assert namespace_security "$@"; }
assert_no_public_exposure() { _observability_yaml_assert no_public_exposure "$@"; }
assert_no_loki_auxiliary() { _observability_yaml_assert no_loki_auxiliary "$@"; }
assert_tempo_topology() { _observability_yaml_assert tempo_topology "$@"; }
assert_selector_values() { _observability_yaml_assert selector_values "$@"; }
assert_prometheus_selectors() { _observability_yaml_assert prometheus_selectors "$@"; }
assert_alertmanager_routing_contract() { _observability_yaml_assert alertmanager_routing "$@"; }
assert_no_crds() { _observability_yaml_assert no_crds "$@"; }
assert_unique_local_pv_storage_classes() { _observability_yaml_assert unique_local_pv_storage_classes "$@"; }
assert_no_unscoped_monitors_rules() { _observability_yaml_assert no_unscoped_monitors_rules "$@"; }

validate_chart_metadata() {
  local chart_yaml="$1"
  local expected_name="$2"
  local expected_version="$3"
  local expected_app_version="$4"
  python3 - "$chart_yaml" "$expected_name" "$expected_version" "$expected_app_version" <<'PY'
import sys
import yaml

try:
    with open(sys.argv[1], "r", encoding="utf-8") as stream:
        chart = yaml.safe_load(stream)
except (OSError, yaml.YAMLError):
    print("REJECT: Chart.yaml is unreadable or invalid", file=sys.stderr)
    raise SystemExit(23)

expected = {"name": sys.argv[2], "version": sys.argv[3], "appVersion": sys.argv[4]}
if not isinstance(chart, dict) or any(str(chart.get(key, "")) != value for key, value in expected.items()):
    print("REJECT: chart name/version/appVersion does not match the fixed table", file=sys.stderr)
    raise SystemExit(23)
PY
}

validate_render_roots() {
  local root="$1"
  shift
  python3 - "$root" "$@" <<'PY'
import os
import stat
import sys


def reject(reason):
    print(f"REJECT: render root preflight failed: {reason}", file=sys.stderr)
    raise SystemExit(23)


def physical_directory(path):
    current = os.sep
    for component in path.split(os.sep)[1:]:
        if not component:
            reject("path is not normalized")
        current = os.path.join(current, component)
        try:
            metadata = os.lstat(current)
        except OSError:
            reject("required directory is missing")
        if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
            reject("directory lineage is not physical")


root = sys.argv[1]
if not os.path.isabs(root) or os.path.normpath(root) != root:
    reject("workspace root must be a normalized absolute path")
physical_directory(root)
if os.path.realpath(root) != root:
    reject("workspace root lineage is not physical")

roots = sys.argv[2:]
if not roots:
    reject("render root list is empty")
for relative in roots:
    parts = relative.split("/")
    if not relative or relative.startswith("/") or any(part in {"", ".", ".."} for part in parts):
        reject("relative render root is unsafe")
    target = os.path.join(root, *parts)
    physical_directory(target)
    if os.path.realpath(target) != target:
        reject("render root lineage is not physical")
PY
}

_observability_handoff_operation() {
  local action="$1"
  local directory="$2"
  local identity="${3:-}"
  local work="${4:-}"
  python3 - "$action" "$directory" "$identity" "$work" <<'PY'
import hashlib
import os
import stat
import sys


EXPECTED_NAMES = (
    "namespaces.yaml", "crds.yaml", "storage.yaml", "kps.yaml", "loki.yaml",
    "tempo.yaml", "alloy.yaml", "node-exporter.yaml", "grafana.yaml", "blackbox.yaml", "core-policies.yaml",
    "targets.yaml", "core-rules.yaml", "alerting.yaml", "dashboards.yaml",
    "aggregate.yaml", "resource-index.tsv",
)
PREFIX = "platform-observability-core-apply."


class Rejected(Exception):
    pass


def reject(reason):
    raise Rejected(reason)


def path_metadata(directory, require_empty):
    if os.path.dirname(directory) != "/tmp":
        reject("handoff must be a direct /tmp child")
    name = os.path.basename(directory)
    if not name.startswith(PREFIX) or len(name) == len(PREFIX):
        reject("handoff suffix is missing")
    try:
        tmp_metadata = os.lstat("/tmp")
        metadata = os.lstat(directory)
    except OSError:
        reject("handoff directory is missing")
    if stat.S_ISLNK(tmp_metadata.st_mode) or not stat.S_ISDIR(tmp_metadata.st_mode):
        reject("/tmp is not a physical directory")
    if os.path.realpath("/tmp") != "/tmp":
        reject("/tmp lineage is not physical")
    if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
        reject("handoff is not a physical directory")
    if metadata.st_uid != os.getuid() or stat.S_IMODE(metadata.st_mode) != 0o700:
        reject("handoff owner or mode differs")
    if require_empty and os.listdir(directory):
        reject("handoff directory is not empty")
    return metadata


def parse_identity(value):
    fields = value.split(":")
    if len(fields) != 2 or any(not field.isdigit() for field in fields):
        reject("handoff identity is malformed")
    return tuple(int(field) for field in fields)


def same_identity(metadata, expected):
    return (metadata.st_dev, metadata.st_ino) == expected


def open_pinned_directory(directory, identity):
    expected = parse_identity(identity)
    metadata = path_metadata(directory, False)
    if not same_identity(metadata, expected):
        reject("handoff directory identity changed")
    try:
        descriptor = os.open(directory, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
    except OSError:
        reject("handoff directory cannot be opened safely")
    opened = os.fstat(descriptor)
    if not same_identity(opened, expected):
        os.close(descriptor)
        reject("opened handoff identity differs")
    return descriptor, expected


def sha256_descriptor(descriptor):
    digest = hashlib.sha256()
    while True:
        chunk = os.read(descriptor, 1024 * 1024)
        if not chunk:
            return digest.hexdigest()
        digest.update(chunk)


def validate_entries(directory, identity):
    descriptor, expected = open_pinned_directory(directory, identity)
    try:
        names = os.listdir(descriptor)
        if set(names) != set(EXPECTED_NAMES) or len(names) != len(EXPECTED_NAMES):
            reject("handoff entry set differs")
        for name in EXPECTED_NAMES:
            metadata = os.stat(name, dir_fd=descriptor, follow_symlinks=False)
            if not stat.S_ISREG(metadata.st_mode) or metadata.st_uid != os.getuid():
                reject("handoff entry type or owner differs")
            if stat.S_IMODE(metadata.st_mode) != 0o600:
                reject("handoff entry mode differs")
        current = os.lstat(directory)
        if not same_identity(current, expected):
            reject("handoff path identity changed")
    finally:
        os.close(descriptor)


def publish(directory, identity, work):
    descriptor, expected = open_pinned_directory(directory, identity)
    created = []
    temporary_created = []
    sources = {}
    try:
        if os.listdir(descriptor):
            reject("handoff directory is not empty immediately before publication")
        try:
            work_metadata = os.lstat(work)
        except OSError:
            reject("verified render directory is missing")
        if stat.S_ISLNK(work_metadata.st_mode) or not stat.S_ISDIR(work_metadata.st_mode):
            reject("verified render directory is not physical")
        for name in EXPECTED_NAMES:
            source = os.path.join(work, name)
            try:
                source_metadata = os.lstat(source)
            except OSError:
                reject("verified output is missing")
            if stat.S_ISLNK(source_metadata.st_mode) or not stat.S_ISREG(source_metadata.st_mode):
                reject("verified output type differs")
            source_descriptor = os.open(source, os.O_RDONLY | os.O_NOFOLLOW)
            try:
                sources[name] = sha256_descriptor(source_descriptor)
            finally:
                os.close(source_descriptor)

        for position, name in enumerate(EXPECTED_NAMES):
            source = os.path.join(work, name)
            temporary_name = f".handoff-{os.getpid()}-{position}"
            try:
                target_descriptor = os.open(
                    temporary_name,
                    os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
                    0o600,
                    dir_fd=descriptor,
                )
            except OSError:
                reject("handoff staging no-clobber creation failed")
            temporary_created.append(temporary_name)
            try:
                source_descriptor = os.open(source, os.O_RDONLY | os.O_NOFOLLOW)
                try:
                    while True:
                        chunk = os.read(source_descriptor, 1024 * 1024)
                        if not chunk:
                            break
                        view = memoryview(chunk)
                        while view:
                            written = os.write(target_descriptor, view)
                            view = view[written:]
                finally:
                    os.close(source_descriptor)
                os.fchmod(target_descriptor, 0o600)
                os.fsync(target_descriptor)
            finally:
                os.close(target_descriptor)
            try:
                os.link(
                    temporary_name,
                    name,
                    src_dir_fd=descriptor,
                    dst_dir_fd=descriptor,
                    follow_symlinks=False,
                )
            except OSError:
                reject("handoff output atomic no-clobber publication failed")
            created.append(name)
            os.unlink(temporary_name, dir_fd=descriptor)
            temporary_created.remove(temporary_name)

        os.fsync(descriptor)
        if set(os.listdir(descriptor)) != set(EXPECTED_NAMES):
            reject("handoff entry set changed during publication")
        for name in EXPECTED_NAMES:
            metadata = os.stat(name, dir_fd=descriptor, follow_symlinks=False)
            if not stat.S_ISREG(metadata.st_mode) or metadata.st_uid != os.getuid():
                reject("published handoff entry is unsafe")
            if stat.S_IMODE(metadata.st_mode) != 0o600:
                reject("published handoff entry mode differs")
            target_descriptor = os.open(name, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=descriptor)
            try:
                if sha256_descriptor(target_descriptor) != sources[name]:
                    reject("published handoff content differs")
            finally:
                os.close(target_descriptor)
        current = os.lstat(directory)
        if not same_identity(current, expected):
            reject("handoff path identity changed during publication")
    except BaseException:
        for name in reversed(created):
            try:
                os.unlink(name, dir_fd=descriptor)
            except OSError:
                pass
        for name in reversed(temporary_created):
            try:
                os.unlink(name, dir_fd=descriptor)
            except OSError:
                pass
        raise
    finally:
        os.close(descriptor)


action, directory, identity, work = sys.argv[1:5]
try:
    if action == "validate":
        path_metadata(directory, True)
    elif action == "pin":
        metadata = path_metadata(directory, True)
        print(f"{metadata.st_dev}:{metadata.st_ino}")
    elif action == "entries":
        validate_entries(directory, identity)
    elif action == "publish":
        publish(directory, identity, work)
    else:
        reject("unknown handoff operation")
except Rejected as error:
    print(f"REJECT: handoff validation failed: {error}", file=sys.stderr)
    raise SystemExit(23)
except OSError:
    print("REJECT: handoff validation failed: filesystem operation failed", file=sys.stderr)
    raise SystemExit(23)
PY
}

validate_handoff_directory() { _observability_handoff_operation validate "$1"; }
pin_handoff_directory() { _observability_handoff_operation pin "$1"; }
validate_handoff_entry_set() { _observability_handoff_operation entries "$1" "$2"; }
publish_verified_handoff() { _observability_handoff_operation publish "$1" "$2" "$3"; }

build_resource_index() {
  local output="$1"
  shift
  python3 - "$output" "$@" <<'PY'
import hashlib
import json
import os
import sys
import tempfile

import yaml


class ResourceIndexSafeLoader(yaml.SafeLoader):
    pass


ResourceIndexSafeLoader.add_constructor(
    "tag:yaml.org,2002:value",
    lambda loader, node: loader.construct_scalar(node),
)


def fail(reason):
    print(f"REJECT: resource index failed: {reason}", file=sys.stderr)
    raise SystemExit(23)


def read_objects(source_specs):
    objects = []
    tuples = set()
    try:
        for source_spec in source_specs:
            if ":" not in source_spec:
                fail("source specification is malformed")
            source, path = source_spec.split(":", 1)
            if not source or not path:
                fail("source specification is malformed")
            with open(path, "r", encoding="utf-8") as stream:
                for document in yaml.load_all(stream, Loader=ResourceIndexSafeLoader):
                    if document is None:
                        continue
                    if not isinstance(document, dict):
                        fail("YAML document is not an object")
                    metadata = document.get("metadata") or {}
                    identity = (
                        str(document.get("apiVersion", "")),
                        str(document.get("kind", "")),
                        str(metadata.get("namespace", "")),
                        str(metadata.get("name", "")),
                    )
                    if not all((identity[0], identity[1], identity[3])):
                        fail("resource identity is incomplete")
                    if identity in tuples:
                        fail("duplicate resource tuple")
                    tuples.add(identity)
                    canonical = json.dumps(document, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
                    digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
                    objects.append((source, identity, digest))
    except (OSError, yaml.YAMLError):
        fail("manifest is unreadable or invalid YAML")
    return objects


output = sys.argv[1]
objects = read_objects(sys.argv[2:])
directory = os.path.dirname(output) or "."
fd, temporary = tempfile.mkstemp(prefix=".resource-index.", dir=directory, text=True)
try:
    with os.fdopen(fd, "w", encoding="utf-8") as stream:
        stream.write("order\tartifact\tapiVersion\tkind\tnamespace\tname\tcanonicalSha256\n")
        for order, (source, identity, digest) in enumerate(objects, 1):
            stream.write("\t".join((str(order), source, *identity, digest)) + "\n")
    os.chmod(temporary, 0o600)
    os.replace(temporary, output)
except BaseException:
    try:
        os.unlink(temporary)
    except OSError:
        pass
    raise
PY
}

assert_child_aggregate_equivalence() {
  local aggregate="$1"
  shift
  python3 - "$aggregate" "$@" <<'PY'
import hashlib
import json
import sys

import yaml


class AggregateSafeLoader(yaml.SafeLoader):
    pass


AggregateSafeLoader.add_constructor(
    "tag:yaml.org,2002:value",
    lambda loader, node: loader.construct_scalar(node),
)


def fail(reason):
    print(f"REJECT: aggregate equivalence failed: {reason}", file=sys.stderr)
    raise SystemExit(23)


def object_map(paths):
    result = {}
    try:
        for path in paths:
            with open(path, "r", encoding="utf-8") as stream:
                for document in yaml.load_all(stream, Loader=AggregateSafeLoader):
                    if document is None:
                        continue
                    if not isinstance(document, dict):
                        fail("YAML document is not an object")
                    metadata = document.get("metadata") or {}
                    identity = (
                        str(document.get("apiVersion", "")),
                        str(document.get("kind", "")),
                        str(metadata.get("namespace", "")),
                        str(metadata.get("name", "")),
                    )
                    if not all((identity[0], identity[1], identity[3])):
                        fail("resource identity is incomplete")
                    if identity in result:
                        fail("duplicate resource tuple")
                    canonical = json.dumps(document, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
                    result[identity] = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
    except (OSError, yaml.YAMLError):
        fail("manifest is unreadable or invalid YAML")
    return result


aggregate_path = sys.argv[1]
child_paths = []
for source_spec in sys.argv[2:]:
    if ":" not in source_spec:
        fail("source specification is malformed")
    source, path = source_spec.split(":", 1)
    if not source or not path:
        fail("source specification is malformed")
    child_paths.append(path)

children = object_map(child_paths)
aggregate = object_map([aggregate_path])
if children.keys() != aggregate.keys():
    fail("child and aggregate resource sets differ")
if children != aggregate:
    fail("child and aggregate canonical object hashes differ")
PY
}

_observability_fail() {
  printf 'ERROR: %s\n' "$*" >&2
  return 1
}

_observability_reject() {
  printf 'REJECT: %s\n' "$*" >&2
  return 23
}

_observability_preflight_chart_cache() {
  local relative_root="$1"
  local chart="$2"
  local version="$3"
  local relative_parent cache
  validate_render_roots "$_OBS_ROOT" "$relative_root" || return
  for relative_parent in "$relative_root/.helm" "$relative_root/.helm/charts"; do
    cache="$_OBS_ROOT/$relative_parent"
    if [[ -e "$cache" || -L "$cache" ]]; then
      validate_render_roots "$_OBS_ROOT" "$relative_parent" || return
    fi
  done
  cache="$_OBS_ROOT/$relative_root/.helm/charts/$chart-$version"
  [[ ! -e "$cache" && ! -L "$cache" ]] || \
    _observability_reject 'exact chart cache already exists'
}

_observability_record_cache_directory() {
  local directory="$1"
  local cleanup_type="$2"
  local identity
  identity="$(stat --format='%d:%i' "$directory")" || return
  _OBS_CREATED_CACHE_DIRS+=("$directory")
  _OBS_CREATED_CACHE_IDENTITIES+=("$identity")
  _OBS_CREATED_CACHE_TYPES+=("$cleanup_type")
}

_observability_create_cache_directory() {
  local relative_root="$1"
  local chart="$2"
  local version="$3"
  local created_rows directory identity cleanup_type
  [[ -v _OBS_CREATED_CACHE_DIRS ]] || _OBS_CREATED_CACHE_DIRS=()
  [[ -v _OBS_CREATED_CACHE_IDENTITIES ]] || _OBS_CREATED_CACHE_IDENTITIES=()
  [[ -v _OBS_CREATED_CACHE_TYPES ]] || _OBS_CREATED_CACHE_TYPES=()

  _observability_preflight_chart_cache "$relative_root" "$chart" "$version" || return
  created_rows="$(python3 - "$_OBS_ROOT" "$relative_root" "$chart-$version" <<'PY'
import errno
import os
import stat
import sys


class Rejected(Exception):
    pass


def reject(reason):
    raise Rejected(reason)


def open_directory(name, parent=None):
    flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW
    try:
        if parent is None:
            return os.open(name, flags)
        return os.open(name, flags, dir_fd=parent)
    except OSError:
        reject("cache directory lineage is not physical")


root, relative_root, exact_name = sys.argv[1:4]
parts = relative_root.split("/")
if not os.path.isabs(root) or os.path.normpath(root) != root:
    reject("workspace root is unsafe")
if not relative_root or any(part in {"", ".", ".."} for part in parts):
    reject("cache root is unsafe")
if not exact_name or "/" in exact_name or exact_name in {".", ".."}:
    reject("exact cache name is unsafe")

descriptors = []
created = []
cache_parts = []
try:
    current = open_directory("/")
    descriptors.append(current)
    for component in root.split("/")[1:]:
        next_descriptor = open_directory(component, current)
        descriptors.append(next_descriptor)
        current = next_descriptor
    for component in parts:
        next_descriptor = open_directory(component, current)
        descriptors.append(next_descriptor)
        current = next_descriptor

    for name, cleanup_type in ((".helm", "empty"), ("charts", "empty")):
        try:
            next_descriptor = os.open(
                name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=current
            )
        except FileNotFoundError:
            try:
                os.mkdir(name, 0o700, dir_fd=current)
            except OSError:
                reject("exclusive cache parent creation failed")
            next_descriptor = open_directory(name, current)
            metadata = os.fstat(next_descriptor)
            full_path = os.path.join(root, relative_root, *cache_parts, name)
            created.append((current, name, full_path, metadata.st_dev, metadata.st_ino, cleanup_type))
        except OSError:
            reject("cache parent is not a physical directory")
        descriptors.append(next_descriptor)
        current = next_descriptor
        cache_parts.append(name)

    try:
        os.mkdir(exact_name, 0o700, dir_fd=current)
    except OSError:
        reject("exclusive exact cache creation failed")
    exact_descriptor = open_directory(exact_name, current)
    descriptors.append(exact_descriptor)
    metadata = os.fstat(exact_descriptor)
    exact_path = os.path.join(root, relative_root, ".helm", "charts", exact_name)
    created.append((current, exact_name, exact_path, metadata.st_dev, metadata.st_ino, "tree"))

    for _, _, path, device, inode, cleanup_type in created:
        print(f"{path}\t{device}:{inode}\t{cleanup_type}")
except Rejected as error:
    for parent, name, _, _, _, _ in reversed(created):
        try:
            os.rmdir(name, dir_fd=parent)
        except OSError:
            pass
    print(f"REJECT: cache creation failed: {error}", file=sys.stderr)
    raise SystemExit(23)
finally:
    for descriptor in reversed(descriptors):
        try:
            os.close(descriptor)
        except OSError:
            pass
PY
  )" || return

  while IFS=$'\t' read -r directory identity cleanup_type; do
    [[ -n "$directory" && -n "$identity" && -n "$cleanup_type" ]] || \
      _observability_reject 'cache creation result is incomplete' || return
    _OBS_CREATED_CACHE_DIRS+=("$directory")
    _OBS_CREATED_CACHE_IDENTITIES+=("$identity")
    _OBS_CREATED_CACHE_TYPES+=("$cleanup_type")
    _OBS_LAST_CACHE_PATH="$directory"
    _OBS_LAST_CACHE_IDENTITY="$identity"
  done <<<"$created_rows"
}

_observability_cache_identity_matches() {
  local directory="$1"
  local identity="$2"
  [[ -d "$directory" && ! -L "$directory" ]] || return 1
  [[ "$(stat --format='%d:%i' "$directory")" == "$identity" ]]
}

_observability_cleanup_owned_cache_directory() {
  local directory="$1"
  local identity="$2"
  local cleanup_type="$3"
  local race_hook="${4:-}"
  python3 - "$_OBS_ROOT" "$directory" "$identity" "$cleanup_type" "$race_hook" <<'PY'
import ctypes
import errno
import os
import secrets
import stat
import subprocess
import sys


RENAME_NOREPLACE = 1


class Rejected(Exception):
    pass


def reject(reason):
    raise Rejected(reason)


def parse_identity(value):
    fields = value.split(":")
    if len(fields) != 2 or any(not field.isdigit() for field in fields):
        reject("tracked identity is malformed")
    return tuple(int(field) for field in fields)


def identity(metadata):
    return metadata.st_dev, metadata.st_ino


libc = ctypes.CDLL(None, use_errno=True)
try:
    renameat2 = libc.renameat2
except AttributeError:
    reject("renameat2 is unavailable")
renameat2.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint]
renameat2.restype = ctypes.c_int


def rename_no_replace(parent_descriptor, source, target):
    result = renameat2(
        parent_descriptor,
        os.fsencode(source),
        parent_descriptor,
        os.fsencode(target),
        RENAME_NOREPLACE,
    )
    if result != 0:
        error = ctypes.get_errno()
        raise OSError(error, os.strerror(error))


def random_name():
    return f".observability-cleanup-{os.getpid()}-{secrets.token_hex(16)}"


def restore_if_possible(parent_descriptor, quarantine, original):
    try:
        rename_no_replace(parent_descriptor, quarantine, original)
        return True
    except OSError:
        return False


def quarantine_entry(parent_descriptor, name):
    try:
        original_metadata = os.stat(name, dir_fd=parent_descriptor, follow_symlinks=False)
    except OSError:
        reject("tracked entry disappeared")
    quarantine = random_name()
    try:
        rename_no_replace(parent_descriptor, name, quarantine)
    except OSError:
        reject("tracked entry quarantine failed")
    try:
        moved_metadata = os.stat(quarantine, dir_fd=parent_descriptor, follow_symlinks=False)
    except OSError:
        reject("quarantined entry disappeared")
    if identity(moved_metadata) != identity(original_metadata):
        restore_if_possible(parent_descriptor, quarantine, name)
        reject("entry identity changed during quarantine")
    return quarantine, moved_metadata


def remove_owned_entry(parent_descriptor, name):
    quarantine, metadata = quarantine_entry(parent_descriptor, name)
    try:
        if stat.S_ISDIR(metadata.st_mode) and not stat.S_ISLNK(metadata.st_mode):
            child_descriptor = os.open(
                quarantine,
                os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                dir_fd=parent_descriptor,
            )
            try:
                if identity(os.fstat(child_descriptor)) != identity(metadata):
                    reject("opened child identity differs")
                clear_owned_directory(child_descriptor)
            finally:
                os.close(child_descriptor)
            current = os.stat(quarantine, dir_fd=parent_descriptor, follow_symlinks=False)
            if identity(current) != identity(metadata):
                reject("child identity changed before removal")
            os.rmdir(quarantine, dir_fd=parent_descriptor)
        else:
            current = os.stat(quarantine, dir_fd=parent_descriptor, follow_symlinks=False)
            if identity(current) != identity(metadata):
                reject("entry identity changed before removal")
            os.unlink(quarantine, dir_fd=parent_descriptor)
    except BaseException:
        restore_if_possible(parent_descriptor, quarantine, name)
        raise


def clear_owned_directory(descriptor):
    for name in os.listdir(descriptor):
        remove_owned_entry(descriptor, name)
    os.fsync(descriptor)


def open_physical_parent(path):
    parent = os.path.dirname(path)
    descriptor = os.open("/", os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
    try:
        for component in parent.split("/")[1:]:
            if not component:
                reject("cleanup path is not normalized")
            next_descriptor = os.open(
                component,
                os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                dir_fd=descriptor,
            )
            os.close(descriptor)
            descriptor = next_descriptor
        return descriptor
    except BaseException:
        os.close(descriptor)
        raise


root, directory, expected_value, cleanup_type, race_hook = sys.argv[1:6]
parent_descriptor = None
owned_descriptor = None
top_quarantine = None
top_name = os.path.basename(directory)
try:
    if not os.path.isabs(root) or os.path.normpath(root) != root:
        reject("workspace root is unsafe")
    if not os.path.isabs(directory) or os.path.normpath(directory) != directory:
        reject("cleanup path is unsafe")
    if os.path.commonpath((root, directory)) != root or directory == root:
        reject("cleanup path escapes workspace root")
    if cleanup_type == "tree":
        if "/.helm/charts/" not in directory:
            reject("recursive cleanup target is not an exact chart cache")
    elif cleanup_type == "empty":
        if not (directory.endswith("/.helm") or directory.endswith("/.helm/charts")):
            reject("empty cleanup target is not a cache parent")
    else:
        reject("cleanup type is invalid")

    expected = parse_identity(expected_value)
    parent_descriptor = open_physical_parent(directory)
    metadata = os.stat(top_name, dir_fd=parent_descriptor, follow_symlinks=False)
    if not stat.S_ISDIR(metadata.st_mode) or stat.S_ISLNK(metadata.st_mode):
        reject("tracked cache is not a physical directory")
    if identity(metadata) != expected:
        reject("tracked cache identity differs")

    top_quarantine, quarantined_metadata = quarantine_entry(parent_descriptor, top_name)
    if identity(quarantined_metadata) != expected:
        restore_if_possible(parent_descriptor, top_quarantine, top_name)
        reject("tracked cache identity changed during quarantine")
    owned_descriptor = os.open(
        top_quarantine,
        os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
        dir_fd=parent_descriptor,
    )
    if identity(os.fstat(owned_descriptor)) != expected:
        reject("opened tracked cache identity differs")

    if race_hook:
        hook_metadata = os.lstat(race_hook)
        if (
            stat.S_ISLNK(hook_metadata.st_mode)
            or not stat.S_ISREG(hook_metadata.st_mode)
            or hook_metadata.st_uid != os.getuid()
            or not os.access(race_hook, os.X_OK)
        ):
            reject("cleanup race hook is unsafe")
        result = subprocess.run(
            [race_hook, directory],
            stdin=subprocess.DEVNULL,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            check=False,
        )
        if result.returncode != 0:
            reject("cleanup race hook failed")

    if cleanup_type == "tree":
        clear_owned_directory(owned_descriptor)
    elif os.listdir(owned_descriptor):
        reject("tracked cache parent is not empty")
    os.close(owned_descriptor)
    owned_descriptor = None

    current = os.stat(top_quarantine, dir_fd=parent_descriptor, follow_symlinks=False)
    if identity(current) != expected:
        reject("quarantined cache identity changed before removal")
    os.rmdir(top_quarantine, dir_fd=parent_descriptor)
    top_quarantine = None
except (Rejected, OSError) as error:
    if owned_descriptor is not None:
        try:
            os.close(owned_descriptor)
        except OSError:
            pass
    if top_quarantine is not None and parent_descriptor is not None:
        restore_if_possible(parent_descriptor, top_quarantine, top_name)
    if isinstance(error, Rejected):
        reason = str(error)
    else:
        reason = "descriptor-relative filesystem operation failed"
    print(f"REJECT: cache cleanup failed: {reason}", file=sys.stderr)
    raise SystemExit(23)
finally:
    if parent_descriptor is not None:
        try:
            os.close(parent_descriptor)
        except OSError:
            pass
PY
}

_observability_cleanup_cache_dirs() {
  local index directory identity cleanup_type cleanup_rc=0
  [[ -v _OBS_CREATED_CACHE_DIRS ]] || return 0
  for ((index=${#_OBS_CREATED_CACHE_DIRS[@]} - 1; index >= 0; index--)); do
    directory="${_OBS_CREATED_CACHE_DIRS[$index]}"
    identity="${_OBS_CREATED_CACHE_IDENTITIES[$index]}"
    cleanup_type="${_OBS_CREATED_CACHE_TYPES[$index]}"
    _observability_cleanup_owned_cache_directory \
      "$directory" "$identity" "$cleanup_type" || cleanup_rc=23
  done
  _OBS_CREATED_CACHE_DIRS=()
  _OBS_CREATED_CACHE_IDENTITIES=()
  _OBS_CREATED_CACHE_TYPES=()
  return "$cleanup_rc"
}

_observability_prepare_chart() {
  local label="$1"
  local repo="$2"
  local chart="$3"
  local version="$4"
  local app_version="$5"
  local expected_sha="$6"
  local relative_root="$7"
  local cache="$_OBS_ROOT/$relative_root/.helm/charts/$chart-$version"
  local archive="$_OBS_WORK/$chart-$version.tgz"
  local actual_sha chart_yaml cache_identity cache_fd cache_fd_identity

  _observability_preflight_chart_cache "$relative_root" "$chart" "$version" || return

  "$_OBS_HELM" pull "$chart" --repo "$repo" --version "$version" --destination "$_OBS_WORK"
  [[ -f "$archive" && ! -L "$archive" ]] || _observability_fail "$label archive is missing" || return
  actual_sha="$(sha256sum "$archive" | awk '{print $1}')"
  [[ "$actual_sha" == "$expected_sha" ]] || _observability_fail "$label archive SHA-256 mismatch" || return
  if ! tar -tzf "$archive" | awk -v prefix="$chart/" '
      $0 == "" || $0 ~ /^\// || $0 ~ /(^|\/)\.\.($|\/)/ || index($0, prefix) != 1 { bad=1 }
      END { exit bad ? 1 : 0 }
    '; then
    _observability_fail "$label archive layout is unsafe"
    return
  fi
  _observability_create_cache_directory "$relative_root" "$chart" "$version" || return
  cache="$_OBS_LAST_CACHE_PATH"
  cache_identity="$_OBS_LAST_CACHE_IDENTITY"
  _observability_cache_identity_matches "$cache" "$cache_identity" || \
    _observability_reject 'exact cache identity changed before extraction' || return
  exec {cache_fd}<"$cache" || _observability_reject 'exact cache cannot be pinned for extraction' || return
  cache_fd_identity="$(stat --dereference --format='%d:%i' "/proc/$$/fd/$cache_fd")"
  if [[ "$cache_fd_identity" != "$cache_identity" ]]; then
    exec {cache_fd}<&-
    _observability_reject 'opened cache identity differs before extraction'
    return
  fi
  if ! tar --extract --gzip --file "$archive" --directory "/proc/$$/fd/$cache_fd" --keep-old-files; then
    exec {cache_fd}<&-
    _observability_reject 'chart extraction failed'
    return
  fi
  cache_fd_identity="$(stat --dereference --format='%d:%i' "/proc/$$/fd/$cache_fd")"
  exec {cache_fd}<&-
  [[ "$cache_fd_identity" == "$cache_identity" ]] || \
    _observability_reject 'opened cache identity changed during extraction' || return
  _observability_cache_identity_matches "$cache" "$cache_identity" || \
    _observability_reject 'exact cache identity changed during extraction' || return
  chart_yaml="$cache/$chart/Chart.yaml"
  [[ -f "$chart_yaml" && ! -L "$chart_yaml" ]] || _observability_fail "$label Chart.yaml is missing" || return
  validate_chart_metadata "$chart_yaml" "$chart" "$version" "$app_version"
  printf 'Verified %-28s SHA-256 %s\n' "$label" "$actual_sha"
}

_observability_render_plain() {
  local output_name="$1"
  local relative_root="$2"
  kubectl kustomize "$_OBS_ROOT/$relative_root" >"$_OBS_WORK/$output_name.yaml"
  [[ -s "$_OBS_WORK/$output_name.yaml" ]] || _observability_fail "$output_name render is empty"
}

_observability_render_with_helm() {
  local output_name="$1"
  local relative_root="$2"
  kubectl kustomize --enable-helm --helm-command "$_OBS_HELM" \
    --helm-kube-version "$(observability_helm_kube_version)" \
    "$_OBS_ROOT/$relative_root" >"$_OBS_WORK/$output_name.yaml"
  [[ -s "$_OBS_WORK/$output_name.yaml" ]] || _observability_fail "$output_name render is empty"
}

_observability_cleanup() {
  local rc=$?
  local cache_cleanup_rc=0
  set +e
  _observability_cleanup_cache_dirs || cache_cleanup_rc=$?
  case "${_OBS_WORK:-}" in
    /tmp/platform-observability-core-render.*) rm -rf -- "$_OBS_WORK" ;;
  esac
  if [[ "$rc" == 0 && "$cache_cleanup_rc" != 0 ]]; then
    rc="$cache_cleanup_rc"
  fi
  exit "$rc"
}

_observability_usage() {
  cat <<'USAGE'
Usage:
  PLATFORM_HELM_BIN=/home/donghyeon/.local/bin/helm \
    bash scripts/validate/render-observability-core.sh

Verified apply handoff (internal interface):
  bash scripts/validate/render-observability-core.sh \
    --verified-output-dir /tmp/platform-observability-core-apply.XXXXXX

This command performs static rendering only. It never applies resources.
USAGE
}

observability_render_main() {
  local command_name kustomize_version label repo chart version app_version expected_sha relative_root
  local output_name bytes digest handoff_identity
  local -a all_manifests workload_manifests child_specs output_names render_roots
  if (( $# == 0 )); then
    _OBS_OUT=""
  elif (( $# == 2 )) && [[ "$1" == "--verified-output-dir" ]]; then
    _OBS_OUT="$2"
  else
    _observability_usage >&2
    return 2
  fi
  _OBS_ROOT="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")/../.." && pwd -P)"

  for command_name in awk cat chmod cmp dirname find install kubectl mktemp python3 rm rmdir sed sha256sum stat tar tr wc; do
    command -v "$command_name" >/dev/null 2>&1 || _observability_fail "$command_name command is required" || return
  done
  python3 -c 'import yaml' >/dev/null 2>&1 || _observability_fail 'Python PyYAML is required' || return

  if [[ -v PLATFORM_HELM_BIN ]]; then
    [[ "$PLATFORM_HELM_BIN" == /* && -f "$PLATFORM_HELM_BIN" && ! -L "$PLATFORM_HELM_BIN" && -x "$PLATFORM_HELM_BIN" ]] || \
      _observability_fail 'PLATFORM_HELM_BIN must be an executable, non-symlink regular file at an absolute path' || return
    _OBS_HELM="$PLATFORM_HELM_BIN"
  else
    command -v helm >/dev/null 2>&1 || _observability_fail 'helm command is required' || return
    _OBS_HELM="$(command -v helm)"
  fi
  [[ "$("$_OBS_HELM" version --template '{{.Version}}')" == "$(observability_expected_helm_version)" ]] || \
    _observability_fail "Helm must be exactly $(observability_expected_helm_version)" || return
  kustomize_version="$(kubectl version --client --output=yaml | sed -n 's/^kustomizeVersion: //p')"
  [[ "$kustomize_version" == "$(observability_expected_kustomize_version)" ]] || \
    _observability_fail "Kustomize must be exactly $(observability_expected_kustomize_version)" || return

  if [[ -n "$_OBS_OUT" ]]; then
    handoff_identity="$(pin_handoff_directory "$_OBS_OUT")"
  fi

  mapfile -t render_roots < <(render_root_rows)
  validate_render_roots "$_OBS_ROOT" "${render_roots[@]}"
  umask 077
  _OBS_WORK="$(mktemp -d /tmp/platform-observability-core-render.XXXXXX)"
  chmod 0700 "$_OBS_WORK"
  _OBS_CREATED_CACHE_DIRS=()
  _OBS_CREATED_CACHE_IDENTITIES=()
  _OBS_CREATED_CACHE_TYPES=()
  trap _observability_cleanup EXIT
  trap 'exit 130' INT
  trap 'exit 143' TERM

  while IFS=$'\t' read -r label repo chart version app_version expected_sha relative_root; do
    _observability_prepare_chart "$label" "$repo" "$chart" "$version" "$app_version" \
      "$expected_sha" "$relative_root"
  done < <(chart_pin_rows)

  _observability_render_plain namespaces infrastructure/namespaces/observability
  _observability_render_with_helm crds infrastructure/controllers/prometheus-operator-crds
  _observability_render_plain storage infrastructure/storage/observability-local-pv
  _observability_render_with_helm kps services/observability/kube-prometheus-stack
  _observability_render_with_helm loki services/observability/loki
  _observability_render_with_helm tempo services/observability/tempo
  _observability_render_with_helm alloy services/observability/alloy
  _observability_render_with_helm node-exporter services/observability/node-exporter
  _observability_render_with_helm grafana services/observability/grafana
  _observability_render_with_helm blackbox services/observability/blackbox-exporter
  _observability_render_plain core-policies services/observability/network-policies
  _observability_render_plain targets services/observability/targets
  _observability_render_plain core-rules services/observability/rules/platform
  _observability_render_plain alerting services/observability/alerting
  _observability_render_plain dashboards services/observability/dashboards/platform
  _observability_render_with_helm aggregate services/observability

  all_manifests=(
    "$_OBS_WORK/namespaces.yaml" "$_OBS_WORK/crds.yaml" "$_OBS_WORK/storage.yaml"
    "$_OBS_WORK/kps.yaml" "$_OBS_WORK/loki.yaml" "$_OBS_WORK/tempo.yaml"
    "$_OBS_WORK/alloy.yaml" "$_OBS_WORK/node-exporter.yaml"
    "$_OBS_WORK/grafana.yaml" "$_OBS_WORK/blackbox.yaml"
    "$_OBS_WORK/core-policies.yaml" "$_OBS_WORK/targets.yaml"
    "$_OBS_WORK/core-rules.yaml" "$_OBS_WORK/alerting.yaml"
    "$_OBS_WORK/dashboards.yaml"
  )
  workload_manifests=(
    "$_OBS_WORK/kps.yaml" "$_OBS_WORK/loki.yaml" "$_OBS_WORK/tempo.yaml"
    "$_OBS_WORK/alloy.yaml" "$_OBS_WORK/node-exporter.yaml"
    "$_OBS_WORK/grafana.yaml" "$_OBS_WORK/blackbox.yaml"
    "$_OBS_WORK/core-policies.yaml" "$_OBS_WORK/targets.yaml"
    "$_OBS_WORK/core-rules.yaml" "$_OBS_WORK/alerting.yaml"
    "$_OBS_WORK/dashboards.yaml"
  )
  assert_pinned_images "${workload_manifests[@]}"
  assert_no_credentials "${all_manifests[@]}" "$_OBS_WORK/aggregate.yaml"
  assert_namespace_pod_security "${workload_manifests[@]}"
  assert_no_public_exposure "${workload_manifests[@]}"
  assert_no_loki_auxiliary "$_OBS_WORK/loki.yaml"
  assert_tempo_topology "$_OBS_WORK/tempo.yaml"
  assert_selector_values "$_OBS_ROOT/services/observability/kube-prometheus-stack/values/home.yaml"
  assert_prometheus_selectors "$_OBS_WORK/kps.yaml"
  assert_alertmanager_routing_contract "$_OBS_WORK/kps.yaml" "$_OBS_WORK/alerting.yaml"
  assert_no_crds "$_OBS_WORK/kps.yaml" "$_OBS_WORK/aggregate.yaml"
  assert_unique_local_pv_storage_classes "$_OBS_WORK/storage.yaml"
  assert_no_unscoped_monitors_rules "${workload_manifests[@]}"

  child_specs=(
    "kps:$_OBS_WORK/kps.yaml"
    "loki:$_OBS_WORK/loki.yaml"
    "tempo:$_OBS_WORK/tempo.yaml"
    "alloy:$_OBS_WORK/alloy.yaml"
    "node-exporter:$_OBS_WORK/node-exporter.yaml"
    "grafana:$_OBS_WORK/grafana.yaml"
    "blackbox:$_OBS_WORK/blackbox.yaml"
    "core-policies:$_OBS_WORK/core-policies.yaml"
    "targets:$_OBS_WORK/targets.yaml"
    "core-rules:$_OBS_WORK/core-rules.yaml"
    "alerting:$_OBS_WORK/alerting.yaml"
    "dashboards:$_OBS_WORK/dashboards.yaml"
  )
  assert_child_aggregate_equivalence "$_OBS_WORK/aggregate.yaml" "${child_specs[@]}"
  build_resource_index "$_OBS_WORK/resource-index.tsv" \
    "namespaces:$_OBS_WORK/namespaces.yaml" \
    "crds:$_OBS_WORK/crds.yaml" \
    "storage:$_OBS_WORK/storage.yaml" \
    "${child_specs[@]}"

  output_names=(namespaces crds storage kps loki tempo alloy node-exporter grafana blackbox core-policies targets core-rules alerting dashboards aggregate)
  for output_name in "${output_names[@]}"; do
    bytes="$(wc -c <"$_OBS_WORK/$output_name.yaml" | tr -d '[:space:]')"
    digest="$(sha256sum "$_OBS_WORK/$output_name.yaml" | awk '{print $1}')"
    printf 'Rendered %-18s %8s bytes  SHA-256 %s\n' "$output_name" "$bytes" "$digest"
  done

  if [[ -n "$_OBS_OUT" ]]; then
    publish_verified_handoff "$_OBS_OUT" "$handoff_identity" "$_OBS_WORK"
    validate_handoff_entry_set "$_OBS_OUT" "$handoff_identity"
  fi

  printf 'OBSERVABILITY CORE STATIC RENDER PASS\n'
}

if [[ "${BASH_SOURCE[0]}" == "$0" ]]; then
  set -Eeuo pipefail
  observability_render_main "$@"
fi