diff --git a/.github/workflows/direct-source-render.yml b/.github/workflows/direct-source-render.yml
new file mode 100644
index 00000000..59014a9a
--- /dev/null
+++ b/.github/workflows/direct-source-render.yml
@@ -0,0 +1,158 @@
+name: exact direct-source snapshot Kustomize render
+
+on:
+  pull_request:
+    paths:
+      - .github/workflows/direct-source-render.yml
+      - scripts/install_pinned_kustomize.py
+      - scripts/direct_source_render_canary.py
+      - scripts/check_snapshot_endpoint_policy.py
+      - scripts/test_snapshot_endpoint_policy.py
+      - docs/direct-source-render.md
+      - fixtures/direct-source/fabrication-server.rs/**
+  workflow_dispatch:
+    inputs:
+      cluster_commit:
+        description: Exact ORESoftware/k8s-cluster catalog candidate
+        required: false
+        default: 0c71eed05f8be50ac22ebe73715ded4a871234a8
+        type: string
+      child_commit:
+        description: Exact private source commit represented by the snapshot
+        required: false
+        default: cea4a3c772012e1f2f87050bac24e911b8e2e577
+        type: string
+
+permissions:
+  contents: read
+
+concurrency:
+  group: direct-source-render-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
+defaults:
+  run:
+    shell: bash
+
+env:
+  K8S_CLUSTER_COMMIT: ${{ inputs.cluster_commit || '0c71eed05f8be50ac22ebe73715ded4a871234a8' }}
+  FABRICATION_SERVER_COMMIT: ${{ inputs.child_commit || 'cea4a3c772012e1f2f87050bac24e911b8e2e577' }}
+  SOURCE_SNAPSHOT: fixtures/direct-source/fabrication-server.rs/cea4a3c772012e1f2f87050bac24e911b8e2e577
+
+jobs:
+  render:
+    name: direct source snapshot render / ${{ matrix.os }}
+    runs-on: ${{ matrix.os }}
+    timeout-minutes: 20
+    strategy:
+      fail-fast: false
+      matrix:
+        os: [ubuntu-24.04, macos-15, windows-2025]
+
+    steps:
+      - name: Check out independent test harness and source snapshot
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          fetch-depth: 1
+          persist-credentials: false
+          show-progress: false
+
+      - name: Check out exact cluster catalog without child worktrees
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          repository: ORESoftware/k8s-cluster
+          ref: ${{ env.K8S_CLUSTER_COMMIT }}
+          path: product/k8s-cluster
+          fetch-depth: 1
+          submodules: false
+          persist-credentials: false
+          show-progress: false
+
+      - name: Install Python 3.12
+        uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
+        with:
+          python-version: '3.12'
+
+      - name: Verify immutable cluster checkout and all test syntax
+        run: |
+          set -euo pipefail
+          [[ "$K8S_CLUSTER_COMMIT" =~ ^[0-9a-f]{40}$ ]]
+          [[ "$FABRICATION_SERVER_COMMIT" =~ ^[0-9a-f]{40}$ ]]
+          test "$(git -C product/k8s-cluster rev-parse HEAD)" = "$K8S_CLUSTER_COMMIT"
+          test -f "$SOURCE_SNAPSHOT/source.json"
+          python -m py_compile \
+            scripts/install_pinned_kustomize.py \
+            scripts/direct_source_render_canary.py \
+            scripts/check_snapshot_endpoint_policy.py \
+            scripts/test_snapshot_endpoint_policy.py \
+            "$SOURCE_SNAPSHOT/scripts/ci/check_k8s_network_contract.py" \
+            "$SOURCE_SNAPSHOT/scripts/ci/test_check_k8s_network_contract.py"
+          python scripts/test_snapshot_endpoint_policy.py
+          (
+            cd "$SOURCE_SNAPSHOT"
+            python scripts/ci/test_check_k8s_network_contract.py
+            python scripts/ci/check_k8s_network_contract.py
+          )
+          rm -rf scripts/__pycache__ "$SOURCE_SNAPSHOT/scripts/ci/__pycache__"
+
+      - name: Install checksum-pinned Kustomize v5.8.1
+        run: |
+          set -euo pipefail
+          mkdir -p artifacts
+          kustomize="$(python scripts/install_pinned_kustomize.py \
+            --destination "$RUNNER_TEMP/pinned-kustomize" \
+            --metadata "artifacts/kustomize-install-${RUNNER_OS}-${RUNNER_ARCH}.json")"
+          test -f "$kustomize"
+          echo "KUSTOMIZE_BINARY=$kustomize" >> "$GITHUB_ENV"
+
+      - name: Verify snapshot endpoint and NetworkPolicy parity independently
+        run: |
+          set -euo pipefail
+          python scripts/check_snapshot_endpoint_policy.py \
+            --deployment "$SOURCE_SNAPSHOT/k8s/deployment.yaml" \
+            --network-policy "$SOURCE_SNAPSHOT/k8s/networkpolicy.yaml" \
+            --evidence "artifacts/snapshot-endpoint-policy-${RUNNER_OS}-${RUNNER_ARCH}.json"
+
+      - name: Verify catalog provenance and render exact source snapshot twice
+        run: |
+          set -euo pipefail
+          python scripts/direct_source_render_canary.py \
+            --cluster product/k8s-cluster \
+            --snapshot "$SOURCE_SNAPSHOT" \
+            --source-metadata "$SOURCE_SNAPSHOT/source.json" \
+            --cluster-commit "$K8S_CLUSTER_COMMIT" \
+            --child-commit "$FABRICATION_SERVER_COMMIT" \
+            --kustomize "$KUSTOMIZE_BINARY" \
+            --kustomize-metadata "artifacts/kustomize-install-${RUNNER_OS}-${RUNNER_ARCH}.json" \
+            --rendered "artifacts/fabrication-render-${RUNNER_OS}-${RUNNER_ARCH}.yaml" \
+            --evidence "artifacts/direct-source-render-${RUNNER_OS}-${RUNNER_ARCH}.json"
+          rm -rf scripts/__pycache__
+
+      - name: Require pristine cluster and harness checkouts
+        run: |
+          set -euo pipefail
+          git -C product/k8s-cluster diff --exit-code
+          git -C product/k8s-cluster diff --cached --exit-code
+          cluster_status="$(git -C product/k8s-cluster status --porcelain --untracked-files=all)"
+          if [[ -n "$cluster_status" ]]; then
+            printf 'unexpected cluster drift:\n%s\n' "$cluster_status" >&2
+            exit 1
+          fi
+          harness_status="$(git status --porcelain --untracked-files=all | grep -vE '^\?\? (artifacts|product)/' || true)"
+          if [[ -n "$harness_status" ]]; then
+            printf 'unexpected harness drift:\n%s\n' "$harness_status" >&2
+            exit 1
+          fi
+
+      - name: Upload exact-source snapshot render evidence
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: direct-source-snapshot-render-${{ runner.os }}-${{ runner.arch }}-${{ env.K8S_CLUSTER_COMMIT }}-${{ env.FABRICATION_SERVER_COMMIT }}
+          path: |
+            artifacts/kustomize-install-${{ runner.os }}-${{ runner.arch }}.json
+            artifacts/snapshot-endpoint-policy-${{ runner.os }}-${{ runner.arch }}.json
+            artifacts/direct-source-render-${{ runner.os }}-${{ runner.arch }}.json
+            artifacts/fabrication-render-${{ runner.os }}-${{ runner.arch }}.yaml
+          if-no-files-found: error
+          retention-days: 14
+          include-hidden-files: false
diff --git a/docs/direct-source-render.md b/docs/direct-source-render.md
new file mode 100644
index 00000000..74f5dc18
--- /dev/null
+++ b/docs/direct-source-render.md
@@ -0,0 +1,72 @@
+# Exact direct-source snapshot Kustomize render certification
+
+This `zed-pkg-test` lane proves that the submodule-backed inventory record and the Argo CD direct-source declaration resolve to the same exact private child commit, that the declared endpoint ports are permitted by the app NetworkPolicy, and that checksum-pinned manifest bytes from that commit render deterministically without materializing the submodule in the cluster superproject.
+
+## Private-repository and billing boundaries
+
+A credential-free cross-organization checkout of `daedalus-fab/fabrication-server.rs` failed with `Repository not found`. That is the expected GitHub permission boundary: a test-repository `GITHUB_TOKEN` cannot read a private repository in another organization.
+
+The owner repository's Actions are also currently blocked before job execution by its billing/spending-limit state. Neither limitation is bypassed with a user PAT.
+
+Instead, the five deployment files were exported through the connected repository reader and committed as an immutable test fixture. Every snapshot file retains the exact Git blob SHA from the source commit, and every runner recomputes `git hash-object` before rendering. This proves source-byte provenance and renderability without claiming private-repository reachability.
+
+Private source reachability remains a separate, owner-scoped GitHub App credential contract.
+
+## Immutable inputs
+
+```text
+ORESoftware/k8s-cluster@0c71eed05f8be50ac22ebe73715ded4a871234a8
+daedalus-fab/fabrication-server.rs@cea4a3c772012e1f2f87050bac24e911b8e2e577
+Kustomize v5.8.1
+```
+
+The source commit is owner PR `daedalus-fab/fabrication-server.rs#11`. It changes OTLP egress from gRPC port 4317 to the Deployment's HTTP/protobuf endpoint port 4318 and adds a source-level parity regression.
+
+The cluster commit is stacked PR `ORESoftware/k8s-cluster#1211`. It advances the mode-160000 gitlink and both GitOps catalog pins to the exact source repair while retaining the trigger hardening from `#1209`.
+
+## Three-platform proof
+
+Ubuntu 24.04, macOS 15, and Windows Server 2025 independently:
+
+1. check out the exact cluster commit with `submodules: false`;
+2. load the five-file source snapshot and its revision/repository/blob manifest;
+3. recompute every snapshot file's Git blob SHA and require an exact match with the private source commit;
+4. verify `.gitmodules`, the mode-160000 Git index entry, catalog inventory, catalog direct source, and snapshot metadata identify the same repository and commit;
+5. prove the deployment source is not materialized beneath the superproject checkout;
+6. run unit tests proving an OTLP 4318 endpoint with only 4317 egress fails closed;
+7. extract the real snapshot's NATS and OTLP endpoint ports and require the NetworkPolicy to permit both;
+8. download the official Kustomize v5.8.1 asset selected for the runner platform;
+9. verify the release asset against a hard-coded official SHA-256 digest before safe extraction;
+10. render the checksum-pinned `k8s` snapshot twice and require byte-identical output;
+11. require exactly one `Deployment`, `Service`, `NetworkPolicy`, and `ExternalSecret` in namespace `daedalus`;
+12. reject cluster-scoped resources, a rendered Kubernetes `Secret`, or plaintext `stringData`; and
+13. upload endpoint-policy, source-blob, binary, and render digests plus the rendered YAML.
+
+The pinned release assets and digests come from the official `kubernetes-sigs/kustomize` v5.8.1 GitHub release. Archive extraction rejects path traversal and links.
+
+## Layered evidence
+
+The layers remain intentionally distinct:
+
+- static catalog validation works during a private-repository credential outage;
+- snapshot rendering proves the exact exported deployment bytes still render;
+- endpoint-policy validation proves checked-in egress permits the declared runtime URLs;
+- an authenticated GitHub App checkout must separately prove live private-repository reachability.
+
+A passing static catalog check is not treated as proof of source renderability, and a passing source snapshot does not grant repository or cluster credentials.
+
+## Isolation
+
+All repository checkouts disable persisted credentials. The canary uses no user PAT, GitHub App private key, registry credential, Kubernetes credential, Cloudflare token, DNS permission, R2 key, or database access. It runs `kustomize build` only; it never invokes `kubectl`, `argocd`, Helm installation, a registry push, or an apply operation.
+
+Passing this canary does not activate the inert ApplicationSet and does not authorize production reconciliation.
+
+## Coordination
+
+- Owner source repair: `daedalus-fab/fabrication-server.rs#11`
+- Stacked exact-pin update: `ORESoftware/k8s-cluster#1211`
+- Baseline catalog/trigger repair: `ORESoftware/k8s-cluster#1209`
+- Current-main adversarial matrix: `zed-pkg-test/security-adversarial-e2e#6`
+- Root dispatcher merge: `zed-pkg/zed-cli#242`
+- GitOps roadmap: `ORESoftware/k8s-cluster#1097`
+- Linear: `DEN-2724`, `DEN-2725`, `DEN-630`
diff --git a/fixtures/direct-source/fabrication-server.rs/5e305b9bf40d28139bdabf9342c304f759cbfee5/k8s/deployment.yaml b/fixtures/direct-source/fabrication-server.rs/5e305b9bf40d28139bdabf9342c304f759cbfee5/k8s/deployment.yaml
new file mode 100644
index 00000000..52875e09
--- /dev/null
+++ b/fixtures/direct-source/fabrication-server.rs/5e305b9bf40d28139bdabf9342c304f759cbfee5/k8s/deployment.yaml
@@ -0,0 +1,101 @@
+# dd-fabrication-server — NATS-driven fabrication planner.
+#
+# Layer-2 (app-owned) manifest: namespace-scoped ONLY. The `daedalus` Namespace, its
+# ResourceQuota/LimitRange and the default-deny NetworkPolicy are Layer-1, owned by
+# ORESoftware/k8s-cluster (remote/argocd/projects/daedalus.tenant.yaml). Do NOT add a
+# Namespace/ClusterRole/CRD here — the `daedalus` AppProject enforces
+# clusterResourceWhitelist: [] and the sync will fail.
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: dd-fabrication-server
+  namespace: daedalus
+  labels:
+    app: dd-fabrication-server
+    app.kubernetes.io/name: dd-fabrication-server
+    app.kubernetes.io/part-of: daedalus
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: dd-fabrication-server
+  template:
+    metadata:
+      labels:
+        app: dd-fabrication-server
+        app.kubernetes.io/name: dd-fabrication-server
+        app.kubernetes.io/part-of: daedalus
+      annotations:
+        # Declared for discoverability; this cluster's Prometheus still scrapes
+        # from static_configs in k8s-cluster, so adding a target needs a PR to
+        # remote/argocd/observability/{prometheus,otel-collector}.configmap.yaml.
+        prometheus.io/scrape: "true"
+        prometheus.io/path: /metrics
+        prometheus.io/port: "8113"
+        prometheus.io/scheme: http
+    spec:
+      automountServiceAccountToken: false
+      securityContext:
+        runAsNonRoot: true
+        runAsUser: 10001
+        fsGroup: 10001
+        seccompProfile: { type: RuntimeDefault }
+      containers:
+        - name: fabrication-server
+          image: ghcr.io/daedalus-fab/fabrication-server.rs:main
+          imagePullPolicy: IfNotPresent
+          ports:
+            - name: http
+              containerPort: 8113
+          env:
+            - { name: HOST, value: "0.0.0.0" }
+            - { name: PORT, value: "8113" }
+            # Cluster NATS (Layer-1, messaging namespace). Subjects use the
+            # dd_nats_subject_defs defaults; override only if the contract changes.
+            - { name: NATS_URL, value: "nats://dd-nats.messaging.svc.cluster.local:4222" }
+            # 4318 (HTTP), not 4317 (gRPC): dd-telemetry exports OTLP over
+            # HTTP/protobuf (the gRPC/tonic stack is deliberately excluded) and
+            # appends /v1/traces. Aimed at 4317 every export is an HTTP POST at
+            # a gRPC listener and is dropped silently — which is why this
+            # service has never produced traces despite the NetworkPolicy
+            # already allowing the egress.
+            - { name: OTEL_EXPORTER_OTLP_ENDPOINT, value: "http://dd-otel-collector.observability.svc.cluster.local:4318" }
+            - { name: OTEL_SERVICE_NAME, value: "dd-fabrication-server" }
+            - { name: RUST_LOG, value: "info" }
+            # The DSN provisioned by k8s/externalsecret.yaml. Without this the
+            # ExternalSecret materialized a Secret that nothing read, and
+            # src/persistence.rs — which reads FABRICATION_DATABASE_URL — would
+            # start with persistence silently disabled.
+            - name: FABRICATION_DATABASE_URL
+              valueFrom:
+                secretKeyRef: { name: dd-daedalus-fabrication-secrets, key: FABRICATION_DATABASE_URL }
+            - name: POD_NAME
+              valueFrom: { fieldRef: { fieldPath: metadata.name } }
+            - name: POD_NAMESPACE
+              valueFrom: { fieldRef: { fieldPath: metadata.namespace } }
+            - name: NODE_NAME
+              valueFrom: { fieldRef: { fieldPath: spec.nodeName } }
+          readinessProbe:
+            httpGet: { path: /readyz, port: http }
+            initialDelaySeconds: 5
+            periodSeconds: 10
+          livenessProbe:
+            httpGet: { path: /healthz, port: http }
+            initialDelaySeconds: 15
+            periodSeconds: 20
+          resources:
+            requests: { cpu: 100m, memory: 128Mi }
+            limits: { cpu: 1, memory: 512Mi }
+          securityContext:
+            allowPrivilegeEscalation: false
+            readOnlyRootFilesystem: true
+            capabilities: { drop: ["ALL"] }
+          volumeMounts:
+            - { name: tmp, mountPath: /tmp }
+          lifecycle:
+            preStop:
+              # Drain: let endpoint removal propagate before shutdown begins.
+              exec: { command: ["sleep", "5"] }
+      volumes:
+        - name: tmp
+          emptyDir: {}
diff --git a/fixtures/direct-source/fabrication-server.rs/5e305b9bf40d28139bdabf9342c304f759cbfee5/k8s/externalsecret.yaml b/fixtures/direct-source/fabrication-server.rs/5e305b9bf40d28139bdabf9342c304f759cbfee5/k8s/externalsecret.yaml
new file mode 100644
index 00000000..446a8779
--- /dev/null
+++ b/fixtures/direct-source/fabrication-server.rs/5e305b9bf40d28139bdabf9342c304f759cbfee5/k8s/externalsecret.yaml
@@ -0,0 +1,41 @@
+# Secrets for dd-fabrication-server, materialized by External Secrets Operator.
+#
+# Layer-2 (app-owned), namespace-scoped. The ClusterSecretStore itself is
+# Layer-1 and lives in ORESoftware/k8s-cluster.
+#
+# WHAT BELONGS HERE: only true secrets. The Supabase audience, issuer, and JWKS
+# URL are public values and are set inline in deployment.yaml — routing them
+# through ESO would imply a confidentiality they do not have and make the auth
+# configuration harder to read.
+#
+# Preferring the JWKS URL (RS256/ES256) over the legacy HS256 shared secret is
+# what keeps this Secret down to a single key: with JWKS there is no symmetric
+# secret to distribute at all. DAEDALUS_API_SUPABASE_JWT_SECRET is therefore
+# intentionally absent; add it only if the project is pinned to HS256.
+# NAME IS PER-SERVICE ON PURPOSE. All three daedalus services previously
+# declared this ExternalSecret as `dd-daedalus-secrets` with the same
+# target.name in the same namespace, while each Argo Application syncs with
+# prune+selfHeal. Three controllers claiming one Secret with three different
+# key sets fight forever: whichever syncs last wins and the other two services
+# lose their credentials. One Secret per owning service avoids that entirely.
+apiVersion: external-secrets.io/v1
+kind: ExternalSecret
+metadata:
+  name: dd-daedalus-fabrication-secrets
+  namespace: daedalus
+  labels:
+    app.kubernetes.io/part-of: daedalus
+spec:
+  refreshInterval: 15m
+  secretStoreRef:
+    name: dd-cluster-secrets
+    kind: ClusterSecretStore
+  target:
+    name: dd-daedalus-fabrication-secrets
+    creationPolicy: Owner
+  data:
+    # The daedalus Postgres DSN. Carries a password, so it is never inline.
+    - secretKey: FABRICATION_DATABASE_URL
+      remoteRef:
+        key: dd/remote-dev/daedalus-secrets
+        property: FABRICATION_DATABASE_URL
diff --git a/fixtures/direct-source/fabrication-server.rs/5e305b9bf40d28139bdabf9342c304f759cbfee5/k8s/kustomization.yaml b/fixtures/direct-source/fabrication-server.rs/5e305b9bf40d28139bdabf9342c304f759cbfee5/k8s/kustomization.yaml
new file mode 100644
index 00000000..9d235707
--- /dev/null
+++ b/fixtures/direct-source/fabrication-server.rs/5e305b9bf40d28139bdabf9342c304f759cbfee5/k8s/kustomization.yaml
@@ -0,0 +1,13 @@
+apiVersion: kustomize.config.k8s.io/v1beta1
+kind: Kustomization
+
+# Namespace-scoped resources only — the `daedalus` Namespace itself is Layer-1
+# (owned by ORESoftware/k8s-cluster). See the org AGENTS.md / k8s-cluster
+# docs/app-deploy-contract.md.
+namespace: daedalus
+
+resources:
+  - deployment.yaml
+  - service.yaml
+  - networkpolicy.yaml
+  - externalsecret.yaml
diff --git a/fixtures/direct-source/fabrication-server.rs/5e305b9bf40d28139bdabf9342c304f759cbfee5/k8s/networkpolicy.yaml b/fixtures/direct-source/fabrication-server.rs/5e305b9bf40d28139bdabf9342c304f759cbfee5/k8s/networkpolicy.yaml
new file mode 100644
index 00000000..1be79509
--- /dev/null
+++ b/fixtures/direct-source/fabrication-server.rs/5e305b9bf40d28139bdabf9342c304f759cbfee5/k8s/networkpolicy.yaml
@@ -0,0 +1,51 @@
+# App-owned allow-list. Pairs with the Layer-1 default-deny in the daedalus tenant
+# scaffold: default-deny drops everything, these rules re-open exactly what this app needs.
+apiVersion: networking.k8s.io/v1
+kind: NetworkPolicy
+metadata:
+  name: dd-fabrication-server
+  namespace: daedalus
+spec:
+  podSelector:
+    matchLabels:
+      app: dd-fabrication-server
+  policyTypes: [Ingress, Egress]
+  ingress:
+    # Prometheus/OTel scrape + in-cluster callers from observability.
+    - from:
+        - namespaceSelector:
+            matchLabels:
+              kubernetes.io/metadata.name: observability
+      ports:
+        - { protocol: TCP, port: 8113 }
+    # Sibling daedalus services.
+    - from:
+        - podSelector:
+            matchLabels:
+              app.kubernetes.io/part-of: daedalus
+      ports:
+        - { protocol: TCP, port: 8113 }
+  egress:
+    # DNS
+    - to:
+        - namespaceSelector: {}
+          podSelector:
+            matchLabels:
+              k8s-app: kube-dns
+      ports:
+        - { protocol: UDP, port: 53 }
+        - { protocol: TCP, port: 53 }
+    # NATS (messaging namespace)
+    - to:
+        - namespaceSelector:
+            matchLabels:
+              kubernetes.io/metadata.name: messaging
+      ports:
+        - { protocol: TCP, port: 4222 }
+    # OTLP telemetry
+    - to:
+        - namespaceSelector:
+            matchLabels:
+              kubernetes.io/metadata.name: observability
+      ports:
+        - { protocol: TCP, port: 4317 }
diff --git a/fixtures/direct-source/fabrication-server.rs/5e305b9bf40d28139bdabf9342c304f759cbfee5/k8s/service.yaml b/fixtures/direct-source/fabrication-server.rs/5e305b9bf40d28139bdabf9342c304f759cbfee5/k8s/service.yaml
new file mode 100644
index 00000000..1f4fd233
--- /dev/null
+++ b/fixtures/direct-source/fabrication-server.rs/5e305b9bf40d28139bdabf9342c304f759cbfee5/k8s/service.yaml
@@ -0,0 +1,16 @@
+apiVersion: v1
+kind: Service
+metadata:
+  name: dd-fabrication-server
+  namespace: daedalus
+  labels:
+    app: dd-fabrication-server
+    app.kubernetes.io/part-of: daedalus
+spec:
+  type: ClusterIP
+  selector:
+    app: dd-fabrication-server
+  ports:
+    - name: http
+      port: 8113
+      targetPort: http
diff --git a/fixtures/direct-source/fabrication-server.rs/5e305b9bf40d28139bdabf9342c304f759cbfee5/source.json b/fixtures/direct-source/fabrication-server.rs/5e305b9bf40d28139bdabf9342c304f759cbfee5/source.json
new file mode 100644
index 00000000..7f13707a
--- /dev/null
+++ b/fixtures/direct-source/fabrication-server.rs/5e305b9bf40d28139bdabf9342c304f759cbfee5/source.json
@@ -0,0 +1,29 @@
+{
+  "$schema": "zed-pkg-test/direct-source-snapshot/v1",
+  "exportMode": "connector-read-no-product-credentials",
+  "repository": "github.com/daedalus-fab/fabrication-server.rs",
+  "revision": "5e305b9bf40d28139bdabf9342c304f759cbfee5",
+  "sourcePath": "k8s",
+  "files": {
+    "k8s/deployment.yaml": {
+      "gitBlobSha1": "52875e0935950ba845e1fb04c9675deef795549b"
+    },
+    "k8s/externalsecret.yaml": {
+      "gitBlobSha1": "446a87798b16537de92ed7fbef5fa4de94a87d75"
+    },
+    "k8s/kustomization.yaml": {
+      "gitBlobSha1": "9d23570713e6760f97b41220add5afacf4403753"
+    },
+    "k8s/networkpolicy.yaml": {
+      "gitBlobSha1": "1be79509a0000a9ae33a5d9437fe781a6a44c79b"
+    },
+    "k8s/service.yaml": {
+      "gitBlobSha1": "1f4fd233b99ce2dd3260c9c7d76a96960bbe60d2"
+    }
+  },
+  "limitations": [
+    "This snapshot proves the exported manifest bytes and renderability.",
+    "It does not prove that a test-org GitHub token can read the private source repository.",
+    "Private repository reachability remains a separately provisioned GitHub App credential contract."
+  ]
+}
diff --git a/fixtures/direct-source/fabrication-server.rs/cea4a3c772012e1f2f87050bac24e911b8e2e577/k8s/deployment.yaml b/fixtures/direct-source/fabrication-server.rs/cea4a3c772012e1f2f87050bac24e911b8e2e577/k8s/deployment.yaml
new file mode 100644
index 00000000..52875e09
--- /dev/null
+++ b/fixtures/direct-source/fabrication-server.rs/cea4a3c772012e1f2f87050bac24e911b8e2e577/k8s/deployment.yaml
@@ -0,0 +1,101 @@
+# dd-fabrication-server — NATS-driven fabrication planner.
+#
+# Layer-2 (app-owned) manifest: namespace-scoped ONLY. The `daedalus` Namespace, its
+# ResourceQuota/LimitRange and the default-deny NetworkPolicy are Layer-1, owned by
+# ORESoftware/k8s-cluster (remote/argocd/projects/daedalus.tenant.yaml). Do NOT add a
+# Namespace/ClusterRole/CRD here — the `daedalus` AppProject enforces
+# clusterResourceWhitelist: [] and the sync will fail.
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: dd-fabrication-server
+  namespace: daedalus
+  labels:
+    app: dd-fabrication-server
+    app.kubernetes.io/name: dd-fabrication-server
+    app.kubernetes.io/part-of: daedalus
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: dd-fabrication-server
+  template:
+    metadata:
+      labels:
+        app: dd-fabrication-server
+        app.kubernetes.io/name: dd-fabrication-server
+        app.kubernetes.io/part-of: daedalus
+      annotations:
+        # Declared for discoverability; this cluster's Prometheus still scrapes
+        # from static_configs in k8s-cluster, so adding a target needs a PR to
+        # remote/argocd/observability/{prometheus,otel-collector}.configmap.yaml.
+        prometheus.io/scrape: "true"
+        prometheus.io/path: /metrics
+        prometheus.io/port: "8113"
+        prometheus.io/scheme: http
+    spec:
+      automountServiceAccountToken: false
+      securityContext:
+        runAsNonRoot: true
+        runAsUser: 10001
+        fsGroup: 10001
+        seccompProfile: { type: RuntimeDefault }
+      containers:
+        - name: fabrication-server
+          image: ghcr.io/daedalus-fab/fabrication-server.rs:main
+          imagePullPolicy: IfNotPresent
+          ports:
+            - name: http
+              containerPort: 8113
+          env:
+            - { name: HOST, value: "0.0.0.0" }
+            - { name: PORT, value: "8113" }
+            # Cluster NATS (Layer-1, messaging namespace). Subjects use the
+            # dd_nats_subject_defs defaults; override only if the contract changes.
+            - { name: NATS_URL, value: "nats://dd-nats.messaging.svc.cluster.local:4222" }
+            # 4318 (HTTP), not 4317 (gRPC): dd-telemetry exports OTLP over
+            # HTTP/protobuf (the gRPC/tonic stack is deliberately excluded) and
+            # appends /v1/traces. Aimed at 4317 every export is an HTTP POST at
+            # a gRPC listener and is dropped silently — which is why this
+            # service has never produced traces despite the NetworkPolicy
+            # already allowing the egress.
+            - { name: OTEL_EXPORTER_OTLP_ENDPOINT, value: "http://dd-otel-collector.observability.svc.cluster.local:4318" }
+            - { name: OTEL_SERVICE_NAME, value: "dd-fabrication-server" }
+            - { name: RUST_LOG, value: "info" }
+            # The DSN provisioned by k8s/externalsecret.yaml. Without this the
+            # ExternalSecret materialized a Secret that nothing read, and
+            # src/persistence.rs — which reads FABRICATION_DATABASE_URL — would
+            # start with persistence silently disabled.
+            - name: FABRICATION_DATABASE_URL
+              valueFrom:
+                secretKeyRef: { name: dd-daedalus-fabrication-secrets, key: FABRICATION_DATABASE_URL }
+            - name: POD_NAME
+              valueFrom: { fieldRef: { fieldPath: metadata.name } }
+            - name: POD_NAMESPACE
+              valueFrom: { fieldRef: { fieldPath: metadata.namespace } }
+            - name: NODE_NAME
+              valueFrom: { fieldRef: { fieldPath: spec.nodeName } }
+          readinessProbe:
+            httpGet: { path: /readyz, port: http }
+            initialDelaySeconds: 5
+            periodSeconds: 10
+          livenessProbe:
+            httpGet: { path: /healthz, port: http }
+            initialDelaySeconds: 15
+            periodSeconds: 20
+          resources:
+            requests: { cpu: 100m, memory: 128Mi }
+            limits: { cpu: 1, memory: 512Mi }
+          securityContext:
+            allowPrivilegeEscalation: false
+            readOnlyRootFilesystem: true
+            capabilities: { drop: ["ALL"] }
+          volumeMounts:
+            - { name: tmp, mountPath: /tmp }
+          lifecycle:
+            preStop:
+              # Drain: let endpoint removal propagate before shutdown begins.
+              exec: { command: ["sleep", "5"] }
+      volumes:
+        - name: tmp
+          emptyDir: {}
diff --git a/fixtures/direct-source/fabrication-server.rs/cea4a3c772012e1f2f87050bac24e911b8e2e577/k8s/externalsecret.yaml b/fixtures/direct-source/fabrication-server.rs/cea4a3c772012e1f2f87050bac24e911b8e2e577/k8s/externalsecret.yaml
new file mode 100644
index 00000000..446a8779
--- /dev/null
+++ b/fixtures/direct-source/fabrication-server.rs/cea4a3c772012e1f2f87050bac24e911b8e2e577/k8s/externalsecret.yaml
@@ -0,0 +1,41 @@
+# Secrets for dd-fabrication-server, materialized by External Secrets Operator.
+#
+# Layer-2 (app-owned), namespace-scoped. The ClusterSecretStore itself is
+# Layer-1 and lives in ORESoftware/k8s-cluster.
+#
+# WHAT BELONGS HERE: only true secrets. The Supabase audience, issuer, and JWKS
+# URL are public values and are set inline in deployment.yaml — routing them
+# through ESO would imply a confidentiality they do not have and make the auth
+# configuration harder to read.
+#
+# Preferring the JWKS URL (RS256/ES256) over the legacy HS256 shared secret is
+# what keeps this Secret down to a single key: with JWKS there is no symmetric
+# secret to distribute at all. DAEDALUS_API_SUPABASE_JWT_SECRET is therefore
+# intentionally absent; add it only if the project is pinned to HS256.
+# NAME IS PER-SERVICE ON PURPOSE. All three daedalus services previously
+# declared this ExternalSecret as `dd-daedalus-secrets` with the same
+# target.name in the same namespace, while each Argo Application syncs with
+# prune+selfHeal. Three controllers claiming one Secret with three different
+# key sets fight forever: whichever syncs last wins and the other two services
+# lose their credentials. One Secret per owning service avoids that entirely.
+apiVersion: external-secrets.io/v1
+kind: ExternalSecret
+metadata:
+  name: dd-daedalus-fabrication-secrets
+  namespace: daedalus
+  labels:
+    app.kubernetes.io/part-of: daedalus
+spec:
+  refreshInterval: 15m
+  secretStoreRef:
+    name: dd-cluster-secrets
+    kind: ClusterSecretStore
+  target:
+    name: dd-daedalus-fabrication-secrets
+    creationPolicy: Owner
+  data:
+    # The daedalus Postgres DSN. Carries a password, so it is never inline.
+    - secretKey: FABRICATION_DATABASE_URL
+      remoteRef:
+        key: dd/remote-dev/daedalus-secrets
+        property: FABRICATION_DATABASE_URL
diff --git a/fixtures/direct-source/fabrication-server.rs/cea4a3c772012e1f2f87050bac24e911b8e2e577/k8s/kustomization.yaml b/fixtures/direct-source/fabrication-server.rs/cea4a3c772012e1f2f87050bac24e911b8e2e577/k8s/kustomization.yaml
new file mode 100644
index 00000000..9d235707
--- /dev/null
+++ b/fixtures/direct-source/fabrication-server.rs/cea4a3c772012e1f2f87050bac24e911b8e2e577/k8s/kustomization.yaml
@@ -0,0 +1,13 @@
+apiVersion: kustomize.config.k8s.io/v1beta1
+kind: Kustomization
+
+# Namespace-scoped resources only — the `daedalus` Namespace itself is Layer-1
+# (owned by ORESoftware/k8s-cluster). See the org AGENTS.md / k8s-cluster
+# docs/app-deploy-contract.md.
+namespace: daedalus
+
+resources:
+  - deployment.yaml
+  - service.yaml
+  - networkpolicy.yaml
+  - externalsecret.yaml
diff --git a/fixtures/direct-source/fabrication-server.rs/cea4a3c772012e1f2f87050bac24e911b8e2e577/k8s/networkpolicy.yaml b/fixtures/direct-source/fabrication-server.rs/cea4a3c772012e1f2f87050bac24e911b8e2e577/k8s/networkpolicy.yaml
new file mode 100644
index 00000000..db738e9e
--- /dev/null
+++ b/fixtures/direct-source/fabrication-server.rs/cea4a3c772012e1f2f87050bac24e911b8e2e577/k8s/networkpolicy.yaml
@@ -0,0 +1,51 @@
+# App-owned allow-list. Pairs with the Layer-1 default-deny in the daedalus tenant
+# scaffold: default-deny drops everything, these rules re-open exactly what this app needs.
+apiVersion: networking.k8s.io/v1
+kind: NetworkPolicy
+metadata:
+  name: dd-fabrication-server
+  namespace: daedalus
+spec:
+  podSelector:
+    matchLabels:
+      app: dd-fabrication-server
+  policyTypes: [Ingress, Egress]
+  ingress:
+    # Prometheus/OTel scrape + in-cluster callers from observability.
+    - from:
+        - namespaceSelector:
+            matchLabels:
+              kubernetes.io/metadata.name: observability
+      ports:
+        - { protocol: TCP, port: 8113 }
+    # Sibling daedalus services.
+    - from:
+        - podSelector:
+            matchLabels:
+              app.kubernetes.io/part-of: daedalus
+      ports:
+        - { protocol: TCP, port: 8113 }
+  egress:
+    # DNS
+    - to:
+        - namespaceSelector: {}
+          podSelector:
+            matchLabels:
+              k8s-app: kube-dns
+      ports:
+        - { protocol: UDP, port: 53 }
+        - { protocol: TCP, port: 53 }
+    # NATS (messaging namespace)
+    - to:
+        - namespaceSelector:
+            matchLabels:
+              kubernetes.io/metadata.name: messaging
+      ports:
+        - { protocol: TCP, port: 4222 }
+    # OTLP over HTTP/protobuf, matching OTEL_EXPORTER_OTLP_ENDPOINT in the Deployment.
+    - to:
+        - namespaceSelector:
+            matchLabels:
+              kubernetes.io/metadata.name: observability
+      ports:
+        - { protocol: TCP, port: 4318 }
diff --git a/fixtures/direct-source/fabrication-server.rs/cea4a3c772012e1f2f87050bac24e911b8e2e577/k8s/service.yaml b/fixtures/direct-source/fabrication-server.rs/cea4a3c772012e1f2f87050bac24e911b8e2e577/k8s/service.yaml
new file mode 100644
index 00000000..1f4fd233
--- /dev/null
+++ b/fixtures/direct-source/fabrication-server.rs/cea4a3c772012e1f2f87050bac24e911b8e2e577/k8s/service.yaml
@@ -0,0 +1,16 @@
+apiVersion: v1
+kind: Service
+metadata:
+  name: dd-fabrication-server
+  namespace: daedalus
+  labels:
+    app: dd-fabrication-server
+    app.kubernetes.io/part-of: daedalus
+spec:
+  type: ClusterIP
+  selector:
+    app: dd-fabrication-server
+  ports:
+    - name: http
+      port: 8113
+      targetPort: http
diff --git a/fixtures/direct-source/fabrication-server.rs/cea4a3c772012e1f2f87050bac24e911b8e2e577/scripts/ci/check_k8s_network_contract.py b/fixtures/direct-source/fabrication-server.rs/cea4a3c772012e1f2f87050bac24e911b8e2e577/scripts/ci/check_k8s_network_contract.py
new file mode 100644
index 00000000..ab99bb1e
--- /dev/null
+++ b/fixtures/direct-source/fabrication-server.rs/cea4a3c772012e1f2f87050bac24e911b8e2e577/scripts/ci/check_k8s_network_contract.py
@@ -0,0 +1,128 @@
+#!/usr/bin/env python3
+"""Validate app endpoint ports against the namespace-scoped NetworkPolicy.
+
+The check intentionally uses only the Python standard library so it can run in
+an isolated source checkout without Kubernetes, Helm, or private path
+dependencies.
+"""
+
+from __future__ import annotations
+
+import argparse
+import re
+import sys
+from pathlib import Path
+from urllib.parse import urlparse
+
+
+class ContractError(ValueError):
+    """Raised when the checked-in Kubernetes source violates the app contract."""
+
+
+def _value_for_environment_variable(source: str, name: str) -> str:
+    lines = source.splitlines()
+    name_pattern = re.compile(rf"\bname:\s*{re.escape(name)}\b")
+    value_pattern = re.compile(r"\bvalue:\s*[\"']?([^\"'\s,}}]+)")
+
+    for index, line in enumerate(lines):
+        if not name_pattern.search(line):
+            continue
+        for candidate in lines[index : index + 4]:
+            match = value_pattern.search(candidate)
+            if match:
+                return match.group(1)
+        raise ContractError(f"{name} is declared without a literal value")
+    raise ContractError(f"deployment does not declare {name}")
+
+
+def _url_port(source: str, name: str) -> int:
+    value = _value_for_environment_variable(source, name)
+    parsed = urlparse(value)
+    if not parsed.scheme or not parsed.hostname or parsed.port is None:
+        raise ContractError(f"{name} must be an absolute URL with an explicit port: {value}")
+    return parsed.port
+
+
+def _network_policy_ports(source: str) -> set[int]:
+    ports = {int(value) for value in re.findall(r"(?m)^\s*-?\s*\{?\s*protocol:\s*TCP,\s*port:\s*(\d+)", source)}
+    ports.update(int(value) for value in re.findall(r"(?m)^\s*port:\s*(\d+)\s*$", source))
+    if not ports:
+        raise ContractError("NetworkPolicy does not declare any TCP egress ports")
+    return ports
+
+
+def _metadata_namespace(source: str, resource: str) -> str:
+    metadata = re.search(r"(?ms)^metadata:\s*$\n(?P<body>(?:^[ \t].*\n?)*)", source)
+    if metadata is None:
+        raise ContractError(f"{resource} does not contain metadata")
+    namespace = re.search(r"(?m)^\s+namespace:\s*([^\s#]+)\s*$", metadata.group("body"))
+    if namespace is None:
+        raise ContractError(f"{resource} metadata does not declare a namespace")
+    return namespace.group(1)
+
+
+def validate(deployment: str, network_policy: str) -> dict[str, object]:
+    deployment_namespace = _metadata_namespace(deployment, "Deployment")
+    policy_namespace = _metadata_namespace(network_policy, "NetworkPolicy")
+    if deployment_namespace != policy_namespace:
+        raise ContractError(
+            "Deployment and NetworkPolicy namespaces differ: "
+            f"{deployment_namespace} != {policy_namespace}"
+        )
+
+    endpoint_ports = {
+        "NATS_URL": _url_port(deployment, "NATS_URL"),
+        "OTEL_EXPORTER_OTLP_ENDPOINT": _url_port(
+            deployment, "OTEL_EXPORTER_OTLP_ENDPOINT"
+        ),
+    }
+    allowed_ports = _network_policy_ports(network_policy)
+    missing = {
+        name: port for name, port in endpoint_ports.items() if port not in allowed_ports
+    }
+    if missing:
+        rendered = ", ".join(f"{name}={port}" for name, port in sorted(missing.items()))
+        raise ContractError(
+            f"NetworkPolicy egress does not permit declared endpoint port(s): {rendered}; "
+            f"allowed TCP ports are {sorted(allowed_ports)}"
+        )
+
+    return {
+        "namespace": deployment_namespace,
+        "endpointPorts": endpoint_ports,
+        "allowedTcpPorts": sorted(allowed_ports),
+    }
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--deployment", type=Path, default=Path("k8s/deployment.yaml"))
+    parser.add_argument(
+        "--network-policy", type=Path, default=Path("k8s/networkpolicy.yaml")
+    )
+    return parser.parse_args()
+
+
+def main() -> int:
+    args = parse_args()
+    try:
+        evidence = validate(
+            args.deployment.read_text(encoding="utf-8"),
+            args.network_policy.read_text(encoding="utf-8"),
+        )
+    except (ContractError, OSError) as error:
+        print(f"error: {error}", file=sys.stderr)
+        return 1
+
+    endpoints = ", ".join(
+        f"{name}={port}" for name, port in evidence["endpointPorts"].items()
+    )
+    print(
+        f"Kubernetes endpoint/egress contract valid in {evidence['namespace']}: "
+        f"{endpoints}; allowed TCP ports={evidence['allowedTcpPorts']}"
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/fixtures/direct-source/fabrication-server.rs/cea4a3c772012e1f2f87050bac24e911b8e2e577/scripts/ci/test_check_k8s_network_contract.py b/fixtures/direct-source/fabrication-server.rs/cea4a3c772012e1f2f87050bac24e911b8e2e577/scripts/ci/test_check_k8s_network_contract.py
new file mode 100644
index 00000000..2e6c6056
--- /dev/null
+++ b/fixtures/direct-source/fabrication-server.rs/cea4a3c772012e1f2f87050bac24e911b8e2e577/scripts/ci/test_check_k8s_network_contract.py
@@ -0,0 +1,70 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import unittest
+
+from check_k8s_network_contract import ContractError, validate
+
+
+DEPLOYMENT = """\
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: example
+  namespace: daedalus
+spec:
+  template:
+    spec:
+      containers:
+        - name: app
+          env:
+            - { name: NATS_URL, value: "nats://nats.messaging.svc:4222" }
+            - { name: OTEL_EXPORTER_OTLP_ENDPOINT, value: "http://otel.observability.svc:4318" }
+"""
+
+NETWORK_POLICY = """\
+apiVersion: networking.k8s.io/v1
+kind: NetworkPolicy
+metadata:
+  name: example
+  namespace: daedalus
+spec:
+  egress:
+    - to: []
+      ports:
+        - { protocol: TCP, port: 4222 }
+        - { protocol: TCP, port: 4318 }
+"""
+
+
+class KubernetesNetworkContractTests(unittest.TestCase):
+    def test_declared_endpoint_ports_are_permitted(self) -> None:
+        result = validate(DEPLOYMENT, NETWORK_POLICY)
+        self.assertEqual("daedalus", result["namespace"])
+        self.assertEqual(
+            {"NATS_URL": 4222, "OTEL_EXPORTER_OTLP_ENDPOINT": 4318},
+            result["endpointPorts"],
+        )
+
+    def test_otlp_port_mismatch_fails_closed(self) -> None:
+        with self.assertRaisesRegex(
+            ContractError,
+            r"OTEL_EXPORTER_OTLP_ENDPOINT=4318",
+        ):
+            validate(DEPLOYMENT, NETWORK_POLICY.replace("port: 4318", "port: 4317"))
+
+    def test_namespace_mismatch_fails_closed(self) -> None:
+        with self.assertRaisesRegex(ContractError, r"namespaces differ"):
+            validate(DEPLOYMENT, NETWORK_POLICY.replace("namespace: daedalus", "namespace: other"))
+
+    def test_endpoint_requires_an_explicit_port(self) -> None:
+        deployment = DEPLOYMENT.replace(
+            "http://otel.observability.svc:4318",
+            "http://otel.observability.svc",
+        )
+        with self.assertRaisesRegex(ContractError, r"explicit port"):
+            validate(deployment, NETWORK_POLICY)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/fixtures/direct-source/fabrication-server.rs/cea4a3c772012e1f2f87050bac24e911b8e2e577/source.json b/fixtures/direct-source/fabrication-server.rs/cea4a3c772012e1f2f87050bac24e911b8e2e577/source.json
new file mode 100644
index 00000000..b925f633
--- /dev/null
+++ b/fixtures/direct-source/fabrication-server.rs/cea4a3c772012e1f2f87050bac24e911b8e2e577/source.json
@@ -0,0 +1,36 @@
+{
+  "$schema": "zed-pkg-test/direct-source-snapshot/v1",
+  "exportMode": "connector-read-no-product-credentials",
+  "repository": "github.com/daedalus-fab/fabrication-server.rs",
+  "revision": "cea4a3c772012e1f2f87050bac24e911b8e2e577",
+  "sourcePath": "k8s",
+  "files": {
+    "k8s/deployment.yaml": {
+      "gitBlobSha1": "52875e0935950ba845e1fb04c9675deef795549b"
+    },
+    "k8s/externalsecret.yaml": {
+      "gitBlobSha1": "446a87798b16537de92ed7fbef5fa4de94a87d75"
+    },
+    "k8s/kustomization.yaml": {
+      "gitBlobSha1": "9d23570713e6760f97b41220add5afacf4403753"
+    },
+    "k8s/networkpolicy.yaml": {
+      "gitBlobSha1": "db738e9efd15707c61db573e296e38a0dd62ae6e"
+    },
+    "k8s/service.yaml": {
+      "gitBlobSha1": "1f4fd233b99ce2dd3260c9c7d76a96960bbe60d2"
+    },
+    "scripts/ci/check_k8s_network_contract.py": {
+      "gitBlobSha1": "ab99bb1eeec553fd119a4c68bcc13fc744491122"
+    },
+    "scripts/ci/test_check_k8s_network_contract.py": {
+      "gitBlobSha1": "2e6c605609c51b79125c187bfccc22ce6c9c974d"
+    }
+  },
+  "sourcePullRequest": "daedalus-fab/fabrication-server.rs#11",
+  "limitations": [
+    "This snapshot proves the exported manifest and regression-script bytes plus renderability.",
+    "It does not prove that a test-org GitHub token can read the private source repository.",
+    "Private repository reachability remains a separately provisioned GitHub App credential contract."
+  ]
+}
diff --git a/scripts/check_snapshot_endpoint_policy.py b/scripts/check_snapshot_endpoint_policy.py
new file mode 100644
index 00000000..3fd5fa0c
--- /dev/null
+++ b/scripts/check_snapshot_endpoint_policy.py
@@ -0,0 +1,116 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import sys
+from pathlib import Path
+from urllib.parse import urlparse
+
+
+class ContractError(ValueError):
+    pass
+
+
+def literal_environment_value(source: str, name: str) -> str:
+    lines = source.splitlines()
+    name_pattern = re.compile(rf"\bname:\s*{re.escape(name)}\b")
+    value_pattern = re.compile(r"\bvalue:\s*[\"']?([^\"'\s,}}]+)")
+    for index, line in enumerate(lines):
+        if not name_pattern.search(line):
+            continue
+        for candidate in lines[index : index + 4]:
+            match = value_pattern.search(candidate)
+            if match:
+                return match.group(1)
+        raise ContractError(f"{name} is declared without a literal value")
+    raise ContractError(f"Deployment does not declare {name}")
+
+
+def endpoint_port(source: str, name: str) -> int:
+    value = literal_environment_value(source, name)
+    parsed = urlparse(value)
+    if not parsed.scheme or not parsed.hostname or parsed.port is None:
+        raise ContractError(f"{name} must be an absolute URL with an explicit port: {value}")
+    return parsed.port
+
+
+def network_policy_tcp_ports(source: str) -> set[int]:
+    ports = {
+        int(value)
+        for value in re.findall(
+            r"(?m)^\s*-?\s*\{?\s*protocol:\s*TCP,\s*port:\s*(\d+)",
+            source,
+        )
+    }
+    ports.update(
+        int(value)
+        for value in re.findall(r"(?m)^\s*port:\s*(\d+)\s*$", source)
+    )
+    if not ports:
+        raise ContractError("NetworkPolicy does not declare any TCP ports")
+    return ports
+
+
+def validate(deployment: str, network_policy: str) -> dict[str, object]:
+    endpoints = {
+        "NATS_URL": endpoint_port(deployment, "NATS_URL"),
+        "OTEL_EXPORTER_OTLP_ENDPOINT": endpoint_port(
+            deployment, "OTEL_EXPORTER_OTLP_ENDPOINT"
+        ),
+    }
+    allowed = network_policy_tcp_ports(network_policy)
+    missing = {name: port for name, port in endpoints.items() if port not in allowed}
+    if missing:
+        rendered = ", ".join(f"{name}={port}" for name, port in sorted(missing.items()))
+        raise ContractError(
+            f"NetworkPolicy does not permit declared endpoint port(s): {rendered}; "
+            f"allowed TCP ports are {sorted(allowed)}"
+        )
+    return {"endpointPorts": endpoints, "allowedTcpPorts": sorted(allowed)}
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--deployment", type=Path, required=True)
+    parser.add_argument("--network-policy", type=Path, required=True)
+    parser.add_argument("--evidence", type=Path)
+    return parser.parse_args()
+
+
+def main() -> int:
+    args = parse_args()
+    try:
+        result = validate(
+            args.deployment.read_text(encoding="utf-8"),
+            args.network_policy.read_text(encoding="utf-8"),
+        )
+    except (ContractError, OSError) as error:
+        print(f"error: {error}", file=sys.stderr)
+        return 1
+
+    if args.evidence is not None:
+        args.evidence.parent.mkdir(parents=True, exist_ok=True)
+        args.evidence.write_text(
+            json.dumps(
+                {
+                    "$schema": "zed-pkg-test/snapshot-endpoint-policy/v1",
+                    "result": "passed",
+                    **result,
+                },
+                indent=2,
+                sort_keys=True,
+            )
+            + "\n",
+            encoding="utf-8",
+        )
+    print(
+        "snapshot endpoint/egress contract valid: "
+        f"endpoints={result['endpointPorts']}; allowed={result['allowedTcpPorts']}"
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/scripts/direct_source_render_canary.py b/scripts/direct_source_render_canary.py
new file mode 100644
index 00000000..f730ef41
--- /dev/null
+++ b/scripts/direct_source_render_canary.py
@@ -0,0 +1,423 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import re
+import subprocess
+import sys
+from pathlib import Path
+from typing import Iterable, Sequence
+
+
+SENSITIVE_ENV_FRAGMENTS = (
+    "TOKEN",
+    "SECRET",
+    "PASSWORD",
+    "PRIVATE_KEY",
+    "ACCESS_KEY",
+    "API_KEY",
+    "AUTHORIZATION",
+    "COOKIE",
+)
+EXPECTED_REPOSITORY = "github.com/daedalus-fab/fabrication-server.rs"
+EXPECTED_KINDS = {"Deployment", "ExternalSecret", "NetworkPolicy", "Service"}
+CLUSTER_SCOPED_KINDS = {
+    "ClusterRole",
+    "ClusterRoleBinding",
+    "CustomResourceDefinition",
+    "MutatingWebhookConfiguration",
+    "Namespace",
+    "Node",
+    "PersistentVolume",
+    "PriorityClass",
+    "StorageClass",
+    "ValidatingWebhookConfiguration",
+}
+
+
+class CertificationError(RuntimeError):
+    pass
+
+
+def require(condition: bool, message: str) -> None:
+    if not condition:
+        raise CertificationError(message)
+
+
+def sanitized_environment() -> dict[str, str]:
+    environment = dict(os.environ)
+    for key in list(environment):
+        upper = key.upper()
+        if any(fragment in upper for fragment in SENSITIVE_ENV_FRAGMENTS):
+            environment.pop(key, None)
+        elif upper.startswith(("ZED_PKG_", "ZED_EXTERNAL_", "ZED_PROBE_")):
+            environment.pop(key, None)
+    environment["CI"] = "true"
+    environment["PYTHONDONTWRITEBYTECODE"] = "1"
+    return environment
+
+
+def run(
+    command: Sequence[object],
+    *,
+    environment: dict[str, str],
+    cwd: Path | None = None,
+    expected: int | Iterable[int] = 0,
+    timeout: int = 120,
+) -> subprocess.CompletedProcess[bytes]:
+    expected_codes = {expected} if isinstance(expected, int) else set(expected)
+    result = subprocess.run(
+        [str(item) for item in command],
+        cwd=cwd,
+        env=environment,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        timeout=timeout,
+        check=False,
+    )
+    if result.returncode not in expected_codes:
+        rendered = " ".join(repr(str(item)) for item in command)
+        raise CertificationError(
+            f"command returned {result.returncode}, expected {sorted(expected_codes)}: {rendered}\n"
+            f"stdout:\n{result.stdout.decode(errors='replace')}\n"
+            f"stderr:\n{result.stderr.decode(errors='replace')}"
+        )
+    return result
+
+
+def text(result: subprocess.CompletedProcess[bytes]) -> str:
+    return result.stdout.decode("utf-8", errors="strict").strip()
+
+
+def sha256_bytes(value: bytes) -> str:
+    return hashlib.sha256(value).hexdigest()
+
+
+def canonical_repository(value: str) -> str:
+    normalized = value.strip().replace("\\", "/")
+    if normalized.startswith("git@github.com:"):
+        normalized = "github.com/" + normalized.removeprefix("git@github.com:")
+    elif normalized.startswith("ssh://git@github.com/"):
+        normalized = "github.com/" + normalized.removeprefix("ssh://git@github.com/")
+    elif normalized.startswith("https://github.com/"):
+        normalized = "github.com/" + normalized.removeprefix("https://github.com/")
+    elif normalized.startswith("http://github.com/"):
+        normalized = "github.com/" + normalized.removeprefix("http://github.com/")
+    return normalized.removesuffix("/").removesuffix(".git").lower()
+
+
+def git_head(root: Path, environment: dict[str, str]) -> str:
+    return text(run(["git", "-C", root, "rev-parse", "HEAD"], environment=environment))
+
+
+def gitmodule_for_path(
+    cluster: Path,
+    path: str,
+    environment: dict[str, str],
+) -> tuple[str, str]:
+    result = run(
+        [
+            "git",
+            "-C",
+            cluster,
+            "config",
+            "-f",
+            ".gitmodules",
+            "--get-regexp",
+            r"^submodule\..*\.path$",
+        ],
+        environment=environment,
+    )
+    selected_key: str | None = None
+    for raw_line in text(result).splitlines():
+        key, _, value = raw_line.partition(" ")
+        if value.strip() == path:
+            selected_key = key.removesuffix(".path")
+            break
+    require(selected_key is not None, f".gitmodules has no entry for {path}")
+    url = text(
+        run(
+            [
+                "git",
+                "-C",
+                cluster,
+                "config",
+                "-f",
+                ".gitmodules",
+                "--get",
+                f"{selected_key}.url",
+            ],
+            environment=environment,
+        )
+    )
+    return selected_key, url
+
+
+def indexed_gitlink(
+    cluster: Path,
+    path: str,
+    environment: dict[str, str],
+) -> tuple[str, str, str, str]:
+    output = text(
+        run(
+            ["git", "-C", cluster, "ls-tree", "HEAD", "--", path],
+            environment=environment,
+        )
+    )
+    require(output, f"Git index has no entry for {path}")
+    left, separator, indexed_path = output.partition("\t")
+    require(separator == "\t", f"unexpected ls-tree output: {output}")
+    mode, object_type, revision = left.split()
+    return mode, object_type, revision, indexed_path
+
+
+def verify_snapshot(
+    snapshot: Path,
+    metadata_path: Path,
+    child_commit: str,
+    environment: dict[str, str],
+) -> tuple[dict[str, object], list[dict[str, str]]]:
+    metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
+    require(isinstance(metadata, dict), "source snapshot metadata must be an object")
+    require(
+        metadata.get("$schema") == "zed-pkg-test/direct-source-snapshot/v1",
+        "source snapshot schema drift",
+    )
+    require(
+        canonical_repository(str(metadata.get("repository", ""))) == EXPECTED_REPOSITORY,
+        "source snapshot repository identity drift",
+    )
+    require(metadata.get("revision") == child_commit, "source snapshot revision drift")
+    require(metadata.get("sourcePath") == "k8s", "source snapshot path drift")
+    files = metadata.get("files")
+    require(isinstance(files, dict) and files, "source snapshot file map is empty")
+
+    verified: list[dict[str, str]] = []
+    for relative, record in sorted(files.items()):
+        require(isinstance(relative, str), "snapshot file key must be text")
+        require(isinstance(record, dict), f"snapshot file metadata must be an object: {relative}")
+        expected_blob = record.get("gitBlobSha1")
+        require(
+            isinstance(expected_blob, str) and re.fullmatch(r"[0-9a-f]{40}", expected_blob),
+            f"snapshot file lacks an exact Git blob SHA: {relative}",
+        )
+        candidate = snapshot / relative
+        require(candidate.is_file(), f"snapshot file is missing: {relative}")
+        actual_blob = text(run(["git", "hash-object", candidate], environment=environment))
+        require(
+            actual_blob == expected_blob,
+            f"snapshot bytes differ from source Git blob for {relative}: {actual_blob} != {expected_blob}",
+        )
+        verified.append(
+            {
+                "path": relative,
+                "gitBlobSha1": actual_blob,
+                "sha256": sha256_bytes(candidate.read_bytes()),
+            }
+        )
+    return metadata, verified
+
+
+def parse_rendered_documents(rendered: bytes) -> list[dict[str, str]]:
+    source = rendered.decode("utf-8", errors="strict")
+    require(source.strip(), "Kustomize emitted empty output")
+    require("\r\n" not in source, "Kustomize output contains platform-specific CRLF")
+    documents: list[dict[str, str]] = []
+    for index, document in enumerate(re.split(r"(?m)^---\s*$", source), start=1):
+        if not document.strip():
+            continue
+        api_match = re.search(r"(?m)^apiVersion:\s*([^\s#]+)\s*$", document)
+        kind_match = re.search(r"(?m)^kind:\s*([^\s#]+)\s*$", document)
+        metadata_match = re.search(r"(?m)^metadata:\s*$", document)
+        name_match = re.search(r"(?m)^  name:\s*([^\s#]+)\s*$", document)
+        namespace_match = re.search(r"(?m)^  namespace:\s*([^\s#]+)\s*$", document)
+        require(api_match is not None, f"rendered document {index} lacks apiVersion")
+        require(kind_match is not None, f"rendered document {index} lacks kind")
+        require(metadata_match is not None, f"rendered document {index} lacks metadata")
+        require(name_match is not None, f"rendered document {index} lacks metadata.name")
+        require(
+            namespace_match is not None and namespace_match.group(1) == "daedalus",
+            f"rendered document {index} is not explicitly namespaced to daedalus",
+        )
+        kind = kind_match.group(1)
+        require(kind not in CLUSTER_SCOPED_KINDS, f"cluster-scoped resource rendered: {kind}")
+        documents.append(
+            {
+                "apiVersion": api_match.group(1),
+                "kind": kind,
+                "name": name_match.group(1),
+                "namespace": namespace_match.group(1),
+            }
+        )
+    require(documents, "Kustomize emitted no YAML documents")
+    return documents
+
+
+def certify(args: argparse.Namespace) -> dict[str, object]:
+    environment = sanitized_environment()
+    cluster = args.cluster.resolve(strict=True)
+    snapshot = args.snapshot.resolve(strict=True)
+    kustomize = args.kustomize.resolve(strict=True)
+    install_metadata = json.loads(args.kustomize_metadata.read_text(encoding="utf-8"))
+
+    require(git_head(cluster, environment) == args.cluster_commit, "cluster checkout SHA drift")
+    snapshot_metadata, snapshot_files = verify_snapshot(
+        snapshot,
+        args.source_metadata.resolve(strict=True),
+        args.child_commit,
+        environment,
+    )
+
+    record_path = cluster / "catalog/gitops/apps/dd-fabrication-server.json"
+    record = json.loads(record_path.read_text(encoding="utf-8"))
+    require(isinstance(record, dict), "catalog record must be an object")
+    require(record.get("metadata", {}).get("name") == "dd-fabrication-server", "catalog app name drift")
+    spec = record.get("spec")
+    require(isinstance(spec, dict), "catalog spec must be an object")
+    inventory = spec.get("inventory")
+    source = spec.get("source")
+    migration = spec.get("migration")
+    require(isinstance(inventory, dict), "catalog inventory must be an object")
+    require(isinstance(source, dict), "catalog source must be an object")
+    require(isinstance(migration, dict), "catalog migration must be an object")
+
+    inventory_path = str(inventory.get("path", ""))
+    source_path = str(source.get("path", ""))
+    require(inventory.get("mode") == "git-submodule", "inventory mode drift")
+    require(inventory_path == "remote/deployments/fabrication-server-rs", "inventory path drift")
+    require(inventory.get("revision") == args.child_commit, "inventory revision differs from snapshot")
+    require(source.get("mode") == "direct-repository", "source is not direct-repository")
+    require(source.get("renderer") == "kustomize", "source renderer is not kustomize")
+    require(source_path == "k8s", "source path drift")
+    require(source.get("targetRevision") == args.child_commit, "source targetRevision differs from snapshot")
+    require(migration.get("phase") == "pilot-inert", "migration phase is no longer pilot-inert")
+
+    inventory_repository = canonical_repository(str(inventory.get("repository", "")))
+    source_repository = canonical_repository(str(source.get("repository", "")))
+    require(inventory_repository == EXPECTED_REPOSITORY, "inventory repository identity drift")
+    require(source_repository == EXPECTED_REPOSITORY, "source repository identity drift")
+    require(inventory_repository == source_repository, "inventory/source repository mismatch")
+    require(
+        source_repository == canonical_repository(str(snapshot_metadata.get("repository", ""))),
+        "catalog source differs from snapshot source identity",
+    )
+
+    module_key, module_url = gitmodule_for_path(cluster, inventory_path, environment)
+    require(canonical_repository(module_url) == EXPECTED_REPOSITORY, ".gitmodules repository identity drift")
+    mode, object_type, indexed_revision, indexed_path = indexed_gitlink(
+        cluster, inventory_path, environment
+    )
+    require(mode == "160000", f"inventory entry mode is {mode}, not 160000")
+    require(object_type == "commit", f"inventory entry type is {object_type}, not commit")
+    require(indexed_path == inventory_path, "indexed gitlink path drift")
+    require(indexed_revision == args.child_commit, "indexed gitlink differs from snapshot revision")
+    require(indexed_revision == inventory.get("revision"), "indexed gitlink differs from catalog inventory")
+    require(indexed_revision == source.get("targetRevision"), "indexed gitlink differs from Argo source")
+
+    require(
+        not (cluster / inventory_path / source_path / "kustomization.yaml").exists(),
+        "cluster checkout unexpectedly materialized the child source; snapshot proof is ambiguous",
+    )
+    snapshot_source = snapshot / source_path
+    require((snapshot_source / "kustomization.yaml").is_file(), "snapshot lacks kustomization.yaml")
+
+    version_output = text(run([kustomize, "version"], environment=environment))
+    require("v5.8.1" in version_output, f"unexpected Kustomize version: {version_output}")
+    render_one = run([kustomize, "build", snapshot_source], environment=environment, timeout=180).stdout
+    render_two = run([kustomize, "build", snapshot_source], environment=environment, timeout=180).stdout
+    require(render_one == render_two, "repeated Kustomize renders differ")
+
+    documents = parse_rendered_documents(render_one)
+    kinds = {document["kind"] for document in documents}
+    require(kinds == EXPECTED_KINDS, f"unexpected rendered resource kinds: {sorted(kinds)}")
+    require(len(documents) == len(EXPECTED_KINDS), "rendered duplicate or missing resource kinds")
+    rendered_text = render_one.decode("utf-8")
+    require(not re.search(r"(?m)^kind:\s*Secret\s*$", rendered_text), "plaintext Secret resource rendered")
+    require(not re.search(r"(?m)^stringData:\s*$", rendered_text), "plaintext stringData rendered")
+
+    args.rendered.parent.mkdir(parents=True, exist_ok=True)
+    args.rendered.write_bytes(render_one)
+
+    return {
+        "$schema": "zed-pkg-test/direct-source-snapshot-render/v1",
+        "clusterCommit": args.cluster_commit,
+        "sourceCommit": args.child_commit,
+        "catalog": {
+            "name": record["metadata"]["name"],
+            "inventoryPath": inventory_path,
+            "sourcePath": source_path,
+            "repository": source_repository,
+            "moduleKey": module_key,
+            "gitlinkRevision": indexed_revision,
+            "migrationPhase": migration["phase"],
+        },
+        "snapshot": {
+            "exportMode": snapshot_metadata.get("exportMode"),
+            "fileCount": len(snapshot_files),
+            "files": snapshot_files,
+            "limitation": "Private repository reachability is not claimed by this snapshot render.",
+        },
+        "kustomize": {
+            "versionOutput": version_output,
+            "asset": install_metadata.get("asset"),
+            "assetSha256": install_metadata.get("assetSha256"),
+            "executableSha256": install_metadata.get("executableSha256"),
+        },
+        "render": {
+            "sha256": sha256_bytes(render_one),
+            "byteCount": len(render_one),
+            "documentCount": len(documents),
+            "resources": documents,
+        },
+        "result": "passed",
+        "checks": [
+            "exact-cluster-checkout-identity",
+            "snapshot-git-blob-provenance",
+            "catalog-inventory-source-snapshot-parity",
+            "gitmodules-and-indexed-gitlink-parity",
+            "source-not-materialized-in-superproject",
+            "checksum-pinned-kustomize-version",
+            "deterministic-repeat-render",
+            "namespace-scoped-resource-set",
+            "no-plaintext-secret-resource",
+        ],
+    }
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--cluster", type=Path, required=True)
+    parser.add_argument("--snapshot", type=Path, required=True)
+    parser.add_argument("--source-metadata", type=Path, required=True)
+    parser.add_argument("--cluster-commit", required=True)
+    parser.add_argument("--child-commit", required=True)
+    parser.add_argument("--kustomize", type=Path, required=True)
+    parser.add_argument("--kustomize-metadata", type=Path, required=True)
+    parser.add_argument("--rendered", type=Path, required=True)
+    parser.add_argument("--evidence", type=Path, required=True)
+    return parser.parse_args()
+
+
+def main() -> int:
+    args = parse_args()
+    try:
+        evidence = certify(args)
+    except (CertificationError, OSError, subprocess.SubprocessError, json.JSONDecodeError) as error:
+        print(f"error: {error}", file=sys.stderr)
+        return 1
+    args.evidence.parent.mkdir(parents=True, exist_ok=True)
+    args.evidence.write_text(
+        json.dumps(evidence, indent=2, sort_keys=True) + "\n", encoding="utf-8"
+    )
+    print(
+        f"rendered {evidence['render']['documentCount']} checksum-pinned source resources "
+        f"at {evidence['render']['sha256']}"
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/scripts/install_pinned_kustomize.py b/scripts/install_pinned_kustomize.py
new file mode 100644
index 00000000..0090b9ff
--- /dev/null
+++ b/scripts/install_pinned_kustomize.py
@@ -0,0 +1,184 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import platform
+import shutil
+import stat
+import sys
+import tarfile
+import tempfile
+import urllib.request
+import zipfile
+from pathlib import Path
+
+
+VERSION = "5.8.1"
+BASE_URL = (
+    "https://github.com/kubernetes-sigs/kustomize/releases/download/"
+    f"kustomize/v{VERSION}"
+)
+ASSETS: dict[tuple[str, str], tuple[str, str]] = {
+    ("linux", "x86_64"): (
+        f"kustomize_v{VERSION}_linux_amd64.tar.gz",
+        "029a7f0f4e1932c52a0476cf02a0fd855c0bb85694b82c338fc648dcb53a819d",
+    ),
+    ("linux", "aarch64"): (
+        f"kustomize_v{VERSION}_linux_arm64.tar.gz",
+        "0953ea3e476f66d6ddfcd911d750f5167b9365aa9491b2326398e289fef2c142",
+    ),
+    ("darwin", "x86_64"): (
+        f"kustomize_v{VERSION}_darwin_amd64.tar.gz",
+        "ee7cf0c1e3592aa7bb66ba82b359933a95e7f2e0b36e5f53ed0a4535b017f2f8",
+    ),
+    ("darwin", "arm64"): (
+        f"kustomize_v{VERSION}_darwin_arm64.tar.gz",
+        "8886f8a78474e608cc81234f729fda188a9767da23e28925802f00ece2bab288",
+    ),
+    ("win32", "amd64"): (
+        f"kustomize_v{VERSION}_windows_amd64.zip",
+        "8ec7f5e815e526d4622c06df0a7793d8cfb6eb1c74f816b46166097fef8b26c6",
+    ),
+    ("win32", "x86_64"): (
+        f"kustomize_v{VERSION}_windows_amd64.zip",
+        "8ec7f5e815e526d4622c06df0a7793d8cfb6eb1c74f816b46166097fef8b26c6",
+    ),
+    ("win32", "arm64"): (
+        f"kustomize_v{VERSION}_windows_arm64.zip",
+        "f6f5090c373965760a30a59af84c81853b80d2c7dbb3236424f440e47eed3f5a",
+    ),
+}
+
+
+class InstallError(RuntimeError):
+    pass
+
+
+def sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as stream:
+        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def safe_extract_zip(archive: Path, destination: Path) -> None:
+    root = destination.resolve()
+    with zipfile.ZipFile(archive) as bundle:
+        for member in bundle.infolist():
+            target = (destination / member.filename).resolve()
+            if target != root and root not in target.parents:
+                raise InstallError(f"unsafe zip member: {member.filename}")
+        bundle.extractall(destination)
+
+
+def safe_extract_tar(archive: Path, destination: Path) -> None:
+    root = destination.resolve()
+    with tarfile.open(archive, "r:gz") as bundle:
+        for member in bundle.getmembers():
+            target = (destination / member.name).resolve()
+            if target != root and root not in target.parents:
+                raise InstallError(f"unsafe tar member: {member.name}")
+            if member.issym() or member.islnk():
+                raise InstallError(f"archive links are not allowed: {member.name}")
+        bundle.extractall(destination, filter="data")
+
+
+def normalized_platform() -> tuple[str, str]:
+    machine = platform.machine().lower()
+    aliases = {
+        "amd64": "amd64",
+        "x86_64": "x86_64",
+        "aarch64": "aarch64",
+        "arm64": "arm64",
+    }
+    return sys.platform, aliases.get(machine, machine)
+
+
+def install(destination: Path, metadata_path: Path) -> Path:
+    key = normalized_platform()
+    asset = ASSETS.get(key)
+    if asset is None:
+        raise InstallError(f"unsupported runner platform: {key[0]}/{key[1]}")
+    asset_name, expected_digest = asset
+    url = f"{BASE_URL}/{asset_name}"
+
+    destination.mkdir(parents=True, exist_ok=True)
+    with tempfile.TemporaryDirectory(prefix="kustomize-install-") as temporary:
+        temporary_root = Path(temporary)
+        archive = temporary_root / asset_name
+        request = urllib.request.Request(
+            url,
+            headers={"User-Agent": "zed-pkg-test-direct-source-render/1"},
+        )
+        with urllib.request.urlopen(request, timeout=90) as response, archive.open("wb") as output:
+            shutil.copyfileobj(response, output)
+        actual_digest = sha256(archive)
+        if actual_digest != expected_digest:
+            raise InstallError(
+                f"Kustomize asset digest mismatch: expected {expected_digest}, got {actual_digest}"
+            )
+
+        extracted = temporary_root / "extracted"
+        extracted.mkdir()
+        if asset_name.endswith(".zip"):
+            safe_extract_zip(archive, extracted)
+        else:
+            safe_extract_tar(archive, extracted)
+
+        executable_name = "kustomize.exe" if os.name == "nt" else "kustomize"
+        matches = [path for path in extracted.rglob(executable_name) if path.is_file()]
+        if len(matches) != 1:
+            raise InstallError(
+                f"expected exactly one {executable_name} in {asset_name}, found {len(matches)}"
+            )
+        target = destination / executable_name
+        shutil.copy2(matches[0], target)
+        if os.name != "nt":
+            target.chmod(target.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
+
+    metadata_path.parent.mkdir(parents=True, exist_ok=True)
+    metadata_path.write_text(
+        json.dumps(
+            {
+                "$schema": "zed-pkg-test/pinned-kustomize-install/v1",
+                "version": VERSION,
+                "platform": {"system": key[0], "machine": key[1]},
+                "asset": asset_name,
+                "assetUrl": url,
+                "assetSha256": expected_digest,
+                "executable": str(target),
+                "executableSha256": sha256(target),
+            },
+            indent=2,
+            sort_keys=True,
+        )
+        + "\n",
+        encoding="utf-8",
+    )
+    return target
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--destination", type=Path, required=True)
+    parser.add_argument("--metadata", type=Path, required=True)
+    return parser.parse_args()
+
+
+def main() -> int:
+    args = parse_args()
+    try:
+        executable = install(args.destination.resolve(), args.metadata.resolve())
+    except (InstallError, OSError, urllib.error.URLError, tarfile.TarError, zipfile.BadZipFile) as error:
+        print(f"error: {error}", file=sys.stderr)
+        return 1
+    print(executable)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/scripts/test_snapshot_endpoint_policy.py b/scripts/test_snapshot_endpoint_policy.py
new file mode 100644
index 00000000..412a7963
--- /dev/null
+++ b/scripts/test_snapshot_endpoint_policy.py
@@ -0,0 +1,65 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import unittest
+
+from check_snapshot_endpoint_policy import ContractError, validate
+
+
+DEPLOYMENT = """\
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: app
+  namespace: daedalus
+spec:
+  template:
+    spec:
+      containers:
+        - name: app
+          env:
+            - { name: NATS_URL, value: "nats://nats.messaging.svc:4222" }
+            - { name: OTEL_EXPORTER_OTLP_ENDPOINT, value: "http://otel.observability.svc:4318" }
+"""
+
+POLICY = """\
+apiVersion: networking.k8s.io/v1
+kind: NetworkPolicy
+metadata:
+  name: app
+  namespace: daedalus
+spec:
+  egress:
+    - to: []
+      ports:
+        - { protocol: TCP, port: 4222 }
+        - { protocol: TCP, port: 4318 }
+"""
+
+
+class SnapshotEndpointPolicyTests(unittest.TestCase):
+    def test_declared_endpoints_are_permitted(self) -> None:
+        result = validate(DEPLOYMENT, POLICY)
+        self.assertEqual(
+            {"NATS_URL": 4222, "OTEL_EXPORTER_OTLP_ENDPOINT": 4318},
+            result["endpointPorts"],
+        )
+
+    def test_otlp_grpc_port_does_not_cover_http_export(self) -> None:
+        with self.assertRaisesRegex(
+            ContractError,
+            r"OTEL_EXPORTER_OTLP_ENDPOINT=4318",
+        ):
+            validate(DEPLOYMENT, POLICY.replace("port: 4318", "port: 4317"))
+
+    def test_missing_explicit_endpoint_port_fails_closed(self) -> None:
+        deployment = DEPLOYMENT.replace(
+            "http://otel.observability.svc:4318",
+            "http://otel.observability.svc",
+        )
+        with self.assertRaisesRegex(ContractError, r"explicit port"):
+            validate(deployment, POLICY)
+
+
+if __name__ == "__main__":
+    unittest.main()