diff --git a/Makefile b/Makefile
index 8321f1cd..db138368 100644
--- a/Makefile
+++ b/Makefile
@@ -73,21 +73,6 @@ gen-mirror-helm-secret: ## Generate mirror-registry Helm OCI password file from
 enable-airgap-secrets: ## Uncomment disconnected-mirror secrets (registryCaCert, bootstrap_secrets) in the generated values-secret file
 	@scripts/enable-airgap-secrets.sh
 
-.PHONY: pck-register
-pck-register: ## Register PCK certificates with Intel PCS (requires INTEL_PCS_API_KEY)
-	@if [ -z "$(INTEL_PCS_API_KEY)" ]; then \
-		echo "ERROR: Set INTEL_PCS_API_KEY environment variable"; \
-		echo "  Usage: make pck-register INTEL_PCS_API_KEY=<key>"; \
-		exit 1; \
-	fi
-	@PCS_TOOL="$(HOME)/confidential-computing.tee.dcap/tools/PcsClientTool/pcsclient.py"; \
-	if [ ! -f "$$PCS_TOOL" ]; then \
-		echo "ERROR: PCS Client Tool not found at $$PCS_TOOL"; \
-		echo "  Clone: git clone https://github.com/intel/confidential-computing.tee.dcap ~/confidential-computing.tee.dcap"; \
-		exit 1; \
-	fi; \
-	cd "$$(dirname $$PCS_TOOL)" && python3 pcsclient.py -t register -k "$(INTEL_PCS_API_KEY)"
-
 ##@ Reference Value Collection
 
 .PHONY: collect-firmware-refvals
@@ -99,13 +84,46 @@ collect-azure-refvals: ## Collect PCR reference values (Azure)
 	@scripts/collect_firmware_refvals.py --platform azure
 
 .PHONY: collect-dcap-collateral
-collect-dcap-collateral: ## Collect TDX DCAP collateral from Intel PCS (API key via OS keyring)
+collect-dcap-collateral: dcap-tools ## Collect platform-independent TDX DCAP verification collateral (connected low side)
 	@scripts/collect-dcap-collateral.sh
 
-.PHONY: dcap-offline-provision
-dcap-offline-provision: ## Full DCAP offline provisioning workflow (collect collateral + load secrets)
-	$(MAKE) collect-dcap-collateral
-	$(MAKE) load-secrets
+##@ Intel TDX PCK Lifecycle
+DCAP_PCSCLIENT_REPO ?= $(HOME)/.coco-pattern/intel-dcap
+DCAP_PCSCLIENT_DIR ?= $(DCAP_PCSCLIENT_REPO)/tools/PcsClientTool
+DCAP_PCSCLIENT_REF ?= 64b78f3766e7196d3d2c60e401540f0f853b2deb
+DCAP_PCK_DIR ?= $(HOME)/.coco-pattern/dcap-pck
+DCAP_REQUEST_BUNDLE ?= $(DCAP_PCK_DIR)/platform-request
+DCAP_RESPONSE_BUNDLE ?= $(DCAP_PCK_DIR)/pck-response
+DCAP_NAMESPACE ?= intel-dcap-operator-system
+DCAP_QGS_DAEMONSET ?=
+DCAP_PCK_EXPIRE_HOURS ?= 8760
+DCAP_QGS_TIMEOUT ?= 10m
+export DCAP_PCSCLIENT_REPO DCAP_PCSCLIENT_DIR DCAP_PCSCLIENT_REF DCAP_PCK_DIR DCAP_REQUEST_BUNDLE DCAP_RESPONSE_BUNDLE DCAP_NAMESPACE DCAP_QGS_DAEMONSET DCAP_PCK_EXPIRE_HOURS DCAP_QGS_TIMEOUT
+
+.PHONY: dcap-tools
+dcap-tools: ## Clone the pinned Intel PcsClientTool and install its Python dependencies (connected low side)
+	@python3 scripts/dcap-pck.py tools
+
+.PHONY: dcap-platform-export
+dcap-platform-export: ## Export QGS platform data as a transfer bundle (disconnected high side; requires oc)
+	@python3 scripts/dcap-pck.py export
+
+.PHONY: dcap-pck-generate
+dcap-pck-generate: dcap-tools ## Generate a PCK response bundle (connected low side; requires Intel PCS access)
+	@python3 scripts/dcap-pck.py generate
+
+.PHONY: dcap-pck-import
+dcap-pck-import: ## Import a PCK response bundle and restart QGS (disconnected high side; requires oc)
+	@python3 scripts/dcap-pck.py import
+
+.PHONY: dcap-pck-provision
+dcap-pck-provision: dcap-tools ## Resume export, generation, and import PCK cache (bastion connected to both Intel PCS and cluster)
+	@python3 scripts/dcap-pck.py provision
+
+.PHONY: dcap-status
+dcap-status: ## Check PCK and collateral expiry (requires oc)
+	@$(MAKE) check-pck-expiry
+	@$(MAKE) check-collateral-expiry
 
 .PHONY: check-pck-expiry
 check-pck-expiry: ## Check expiry of PCK cache secrets and platform data (requires oc login)
diff --git a/README.md b/README.md
index 7edb0f6b..c33b815f 100644
--- a/README.md
+++ b/README.md
@@ -39,6 +39,7 @@ Breaking change from v5. Upgrades to OSC 1.13 / Trustee 1.2, adds full airgap (d
 - **Chart architecture:** trustee-chart moved to OCI Helm artifact (`quay.io/validatedpatterns/trustee:0.10.0`). Kyverno chart vendored under `charts/vendor/`. External charts (sandboxed-containers, sandboxed-policies) remain OCI.
 - **TDX MachineConfig:** QGS socket port `socket_port=0` now deployed via MachineConfig drop-in (replaces manual `sed` workaround).
 - **DCAP collateral:** `collect-dcap-collateral.sh` uses `pcsclient.py fetch -p E5` with `jq` fixup for QeIdentity (Red Hat OSC 1.13 disconnected TDX procedure).
+- **TDX PCK lifecycle:** Make targets export platform data, generate PCK cache material on a connected low-side host, and import it into the disconnected high-side cluster. See [the air-gap runbook](airgap/DEPLOY-RUNBOOK.md#phase-e-dcap-and-tdx-attestation).
 - **Experimental:** KubeVirt TDX confidential VMs (`charts/all/kubevirtconfidential/`, `charts/all/kubevirtvm/`). Disabled by default. Requires Intel TDX hardware and KubeVirt post-v1.8.4 for full QGS attestation. See chart readmes for details.
 
 ### Previous versions
@@ -64,7 +65,7 @@ For air-gapped bare metal environments, see [`airgap/DEPLOY-RUNBOOK.md`](airgap/
 
 **Prerequisites for airgap:**
 
-- A jump host with internet access (for `oc-mirror`) and network access to the target cluster
+- A low-side host with Intel PCS access for mirroring, collateral collection, and PCK generation. A connected bastion with access to both Intel PCS and the target cluster can run the combined PCK target.
 - `docker.io/library/registry:2` container running as the mirror registry (setup documented in Phase 0)
 - `scripts/git-http-server.py` serving pattern repos over smart HTTP (required by the patterns-operator's go-git client)
 - All operator images, OCI Helm charts, and workload images mirrored via `airgap/imageset-config-4.22.yaml`
@@ -104,6 +105,16 @@ These scripts generate the cryptographic material and attestation reference valu
    - See [docs/firmware-reference-values.md](docs/firmware-reference-values.md) for detailed workflow and options.
 3. Review and customise `~/values-secret-coco-pattern.yaml` — this file is loaded into Vault and provides secrets to the pattern.
 
+### Intel TDX PCK lifecycle
+
+All lifecycle actions are invoked with Make. `collect-dcap-collateral` prepares the pinned Intel tool automatically, then retrieves public verification collateral on the connected low side; transfer `~/.coco-pattern/dcap-offline/platform_collaterals.json` to the high side, then use the existing `make load-secrets` to load it into Vault.
+
+PCK certificates are platform-specific and use separate request and response bundles. On a disconnected high-side cluster, run `make dcap-platform-export`, transfer `~/.coco-pattern/dcap-pck/platform-request` to the connected low side, then run `make dcap-tools dcap-pck-generate`. Transfer `~/.coco-pattern/dcap-pck/pck-response` back to the high side and run `make dcap-pck-import`.
+
+When one bastion can reach both Intel PCS and the cluster, run `make dcap-pck-provision`. It resumes safely after interruption: matching request and unexpired response bundles are reused, while changed platform data replaces both bundles. The generator accepts `INTEL_PCS_API_KEY` from its environment for automation, or requests it through a hidden terminal prompt. Do not pass the key as a Make variable or command-line argument.
+
+The importer discovers the QGS DaemonSet from the `pck-certs-watcher` pod owner reference. Set `DCAP_QGS_DAEMONSET` only when that discovery cannot identify a single DaemonSet.
+
 > **Note:** `gen-secrets.sh` will not overwrite existing secrets. Delete `~/.coco-pattern/` if you need to regenerate.
 
 ### Single cluster deployment (Azure)
diff --git a/airgap/DEPLOY-RUNBOOK.md b/airgap/DEPLOY-RUNBOOK.md
index e4b242b5..4eb40e87 100644
--- a/airgap/DEPLOY-RUNBOOK.md
+++ b/airgap/DEPLOY-RUNBOOK.md
@@ -16,7 +16,7 @@
 | B | Full Re-Mirror (oc-mirror v2) | ~3-4 hours |
 | C | Cluster Install | ~35 min |
 | D | Bootstrap and Pattern Deploy | ~25 min |
-| E | DCAP and TDX Attestation | ~15 min (+ pcsclient time if QE ID mismatch) |
+| E | DCAP and TDX Attestation | ~15 min (+ connected low-side PCK generation) |
 | F | Verification and Pass/Fail | ~10 min |
 | **Total** | | **~5-6 hours** (first-time add ~30 min) |
 
@@ -1127,74 +1127,36 @@ echo "E-2: QGS socket port verification complete at $(date)" 2>&1 | tee -a "$LOG
 > `oc get mc | grep kata-tdx` shows a MachineConfig with the drop-in content.
 > See Appendix Section A.3 for the temporary sed fallback. Record in DEVIATIONS.md.
 
-### E-3: PCK Registration — pcsclient.py cache (ALWAYS REQUIRED, INTERACTIVE)
+### E-3: PCK Lifecycle
 
-> **This step is ALWAYS required** — even when QE IDs match and cached PCK cert files exist.
-> `pcsclient.py cache` does two things: (1) provisions the PCK certificate secret so QGS can
-> generate quotes, and (2) populates the local cache with **QeIdentity and TcbInfo** that
-> `collect-dcap-collateral.sh` reads in E-4. Skipping this step leaves `qeidentity` empty in
-> `platform_collaterals.json`, causing KBS to fail attestation with
-> `"collateral JSON error: invalid type: string "", expected struct QeIdentity"`.
->
-> **This step must be run interactively** — `pcsclient.py` prompts for the Intel PCS API key
-> via `getpass` and cannot be automated via SSH pipe (see `UPSTREAM-ISSUES-INTEL-DCAP.md` Issue 3).
+PCK certificates are specific to each QGS platform. The Make targets create checksummed request and response bundles, validate the complete QE-ID/PCE-ID set, and restart QGS only after all matching PCK Secrets are applied.
+
+The importer discovers the QGS DaemonSet from the `pck-certs-watcher` pod owner reference, avoiding a version-specific DaemonSet label or name. Set `DCAP_QGS_DAEMONSET` only if the namespace has more than one candidate.
+
+If this bastion can reach **both** the disconnected cluster and Intel PCS, run the combined workflow. The API key is read from `INTEL_PCS_API_KEY` if exported, otherwise Make prompts without echoing it. Never put the key on the Make command line.
 
 ```bash
-echo "=== E-3: PCK Registration (INTERACTIVE — run in tmux) ===" 2>&1 | tee -a "$LOG"
-cd ~/pck-registration
-
-# Step 1: Generate platform_list.json from cluster platform-data secrets
-echo "Generating platform_list.json..." 2>&1 | tee -a "$LOG"
-oc get secrets -o json -n intel-dcap-operator-system -l 'type=platform-data' \
-  | jq '[.items[] | .data | map_values(@base64d)]' > platform_list.json
-echo "platform_list.json: $(wc -l < platform_list.json) lines" 2>&1 | tee -a "$LOG"
-
-# Step 2: Compare QE IDs — cache reuse check (informational only)
-CLUSTER_QE_ID=$(oc get secrets -n intel-dcap-operator-system -l type=platform-data \
-  --no-headers -o custom-columns=NAME:.metadata.name | head -1)
-CACHED_QE_ID=$(ls cache/*_0000 2>/dev/null | head -1 | xargs basename 2>/dev/null | sed 's/_0000//')
-echo "Cluster QE ID: $CLUSTER_QE_ID" 2>&1 | tee -a "$LOG"
-echo "Cached QE ID:  $CACHED_QE_ID" 2>&1 | tee -a "$LOG"
-if [ "$CLUSTER_QE_ID" = "$CACHED_QE_ID" ] && [ -n "$CACHED_QE_ID" ]; then
-  echo "QE ID MATCH — PCK cert cache reusable. Still must run pcsclient.py cache for QeIdentity." \
-    2>&1 | tee -a "$LOG"
-else
-  echo "QE ID MISMATCH — record as DEV-2 in DEVIATIONS.md" 2>&1 | tee -a "$LOG"
-fi
+echo "=== E-3: Combined PCK provisioning ===" 2>&1 | tee -a "$LOG"
+cd ~/coco-pattern
+make dcap-pck-provision 2>&1 | tee -a "$LOG"
+```
 
-# Step 3: Run pcsclient.py cache INTERACTIVELY
-# Will prompt: "Please input ApiKey for Intel PCS:" — paste your key.
-# Populates cache/ with PCK certs AND QeIdentity/TcbInfo for E-4.
-echo "STOP: Run the following command interactively in this tmux window:" 2>&1 | tee -a "$LOG"
-echo ""
-echo "  python3 ~/confidential-computing.tee.dcap/tools/PcsClientTool/pcsclient.py cache \\"
-echo "    -i platform_list.json \\"
-echo "    -e 8760 \\"
-echo "    -t early"
-echo ""
-echo "Enter API key when prompted, then press Enter."
-echo "When complete, continue to Step 4 below."
-
-# Step 4: Apply PCK cert secret to cluster and restart QGS
-# (Run after pcsclient.py cache completes)
-for f in cache/*_0000; do
-  qe_id=$(basename "$f" _0000)
-  echo "Applying PCK cert for QE ID: $qe_id" 2>&1 | tee -a "$LOG"
-  oc create secret generic "${qe_id}-pck" --from-file=certificate="$f" \
-    -n intel-dcap-operator-system --dry-run=client -o yaml | oc apply -f - \
-    2>&1 | tee -a "$LOG"
-done
-oc delete pod -n intel-dcap-operator-system -l app=intel-tdx-qgs \
-  2>&1 | tee -a "$LOG"
+If the high-side cluster cannot reach the connected low-side host, use the split workflow:
+
+```bash
+# High side: create ~/.coco-pattern/dcap-pck/platform-request and transfer it high -> low.
+make dcap-platform-export
+
+# Low side: copy platform-request to ~/.coco-pattern/dcap-pck/, then run these commands.
+make dcap-tools dcap-pck-generate
 
-echo "E-3: PCK registration complete at $(date)" 2>&1 | tee -a "$LOG"
+# Transfer ~/.coco-pattern/dcap-pck/pck-response low -> high, then import it on the high side.
+make dcap-pck-import
 ```
 
 ### E-4: Collect DCAP Collateral and Firmware Reference Values
 
-> **Must run AFTER E-3** — `collect-dcap-collateral.sh` calls `pcsclient.py fetch` which reads
-> from the local cache populated by `pcsclient.py cache` in E-3. Running E-4 before E-3
-> produces `platform_collaterals.json` with `qeidentity: ""` — KBS will fail attestation.
+`collect-dcap-collateral` is independent of PCK provisioning. It collects public Intel TDX verification collateral on the connected low side and does not require an Intel PCS API key. Transfer `~/.coco-pattern/dcap-offline/platform_collaterals.json` to the high side before E-5.
 
 ```bash
 echo "=== E-4: Collect DCAP Collateral + Firmware Refvals ===" 2>&1 | tee -a "$LOG"
@@ -1206,8 +1168,8 @@ make collect-dcap-collateral 2>&1 | tee -a "$LOG"
 python3 -c "
 import json, os
 c = json.load(open(os.path.expanduser('~/.coco-pattern/dcap-offline/platform_collaterals.json')))
-qi = c.get('collaterals', {}).get('qeidentity', '')
-print('qeidentity length:', len(qi), '— PASS' if len(qi) > 100 else '— FAIL: re-run E-3 first')
+qi = c.get('collaterals', {}).get('qeidentity_early') or c.get('collaterals', {}).get('qeidentity')
+print('qeidentity present:', isinstance(qi, dict), '— PASS' if isinstance(qi, dict) else '— FAIL: recollect collateral')
 " 2>&1 | tee -a "$LOG"
 
 make collect-firmware-refvals 2>&1 | tee -a "$LOG"
diff --git a/scripts/check-collateral-expiry.sh b/scripts/check-collateral-expiry.sh
index ca057519..01d66382 100755
--- a/scripts/check-collateral-expiry.sh
+++ b/scripts/check-collateral-expiry.sh
@@ -45,7 +45,9 @@ if ! oc get secret "$SECRET" -n "$NS" &>/dev/null; then
     exit 1
 fi
 
-oc get secret "$SECRET" -n "$NS" \
+while IFS='|' read -r label next_update; do
+    check_date "$label" "$next_update"
+done < <(oc get secret "$SECRET" -n "$NS" \
     -o jsonpath='{.data.platform_collaterals\.json}' | base64 -d | python3 -c "
 import json, sys
 data = json.load(sys.stdin)
@@ -63,22 +65,18 @@ for ti in col.get('tcbinfos', []):
         if nu:
             print(f'{key} FMSPC={fmspc}|{nu}')
 
-for qi in col.get('qeidentities', []):
-    for key in ['qe_identity_early', 'qe_identity']:
-        ei = qi.get(key, {})
-        if isinstance(ei, str):
-            try:
-                ei = json.loads(ei)
-            except json.JSONDecodeError:
-                continue
-        if isinstance(ei, dict):
-            info = ei.get('enclaveIdentity', {})
-            nu = info.get('nextUpdate')
-            if nu:
-                print(f'{key}|{nu}')
-" | while IFS='|' read -r label next_update; do
-    check_date "$label" "$next_update"
-done
+for key in ['qeidentity_early', 'qeidentity', 'tdqeidentity_early', 'tdqeidentity', 'qveidentity_early', 'qveidentity']:
+    identity = col.get(key, {})
+    if isinstance(identity, str):
+        try:
+            identity = json.loads(identity)
+        except json.JSONDecodeError:
+            continue
+    if isinstance(identity, dict):
+        next_update = identity.get('enclaveIdentity', {}).get('nextUpdate')
+        if next_update:
+            print(f'{key}|{next_update}')
+")
 
 echo ""
 if [ "$fail" -ne 0 ]; then
diff --git a/scripts/check-pck-expiry.sh b/scripts/check-pck-expiry.sh
index bb32ef86..a51249fb 100755
--- a/scripts/check-pck-expiry.sh
+++ b/scripts/check-pck-expiry.sh
@@ -52,7 +52,33 @@ else
         echo "  PCK secret: ${secret_name} (QE ID: ${qe_id})"
 
         # The PCK cache blob contains URL-encoded PEM X.509 certs + embedded JSON TCB info
-        oc get secret "$secret_name" -n "$NS" -o jsonpath='{.data.certificate}' | \
+        while IFS='|' read -r rtype val1 val2; do
+            case "$rtype" in
+                NONE)
+                    echo -e "    ${YELLOW}UNKNOWN${NC}  No expiry data found in PCK cache blob"
+                    fail=1
+                    ;;
+                CERT)
+                    iso=$(python3 -c "
+from datetime import datetime
+try:
+    dt = datetime.strptime('$val2', '%b %d %H:%M:%S %Y %Z')
+except ValueError:
+    dt = datetime.strptime('$val2', '%b  %d %H:%M:%S %Y %Z')
+print(dt.strftime('%Y-%m-%dT%H:%M:%SZ'))
+" 2>/dev/null)
+                    if [ -z "$iso" ]; then
+                        echo -e "    ${YELLOW}UNKNOWN${NC}  Cannot parse certificate expiry: $val2"
+                        fail=1
+                    else
+                        check_date "    cert: $val1" "$iso"
+                    fi
+                    ;;
+                TCB)
+                    check_date "    TCB nextUpdate" "$val1"
+                    ;;
+            esac
+        done < <(oc get secret "$secret_name" -n "$NS" -o jsonpath='{.data.certificate}' | \
             base64 -d | python3 -c "
 import sys, re, subprocess, urllib.parse
 data = sys.stdin.buffer.read()
@@ -82,27 +108,7 @@ for m in re.findall(r'\"nextUpdate\":\"([^\"]+)\"', text):
 
 if not certs and not re.search(r'nextUpdate', text):
     print('NONE')
-" | while IFS='|' read -r rtype val1 val2; do
-            case "$rtype" in
-                NONE)
-                    echo -e "    ${YELLOW}UNKNOWN${NC}  No expiry data found in PCK cache blob"
-                    ;;
-                CERT)
-                    iso=$(python3 -c "
-from datetime import datetime
-try:
-    dt = datetime.strptime('$val2', '%b %d %H:%M:%S %Y %Z')
-except ValueError:
-    dt = datetime.strptime('$val2', '%b  %d %H:%M:%S %Y %Z')
-print(dt.strftime('%Y-%m-%dT%H:%M:%SZ'))
-" 2>/dev/null)
-                    [ -n "$iso" ] && check_date "    cert: $val1" "$iso"
-                    ;;
-                TCB)
-                    check_date "    TCB nextUpdate" "$val1"
-                    ;;
-            esac
-        done
+")
     done
 fi
 
diff --git a/scripts/collect-dcap-collateral.sh b/scripts/collect-dcap-collateral.sh
index 0a2852f4..b9e4473a 100755
--- a/scripts/collect-dcap-collateral.sh
+++ b/scripts/collect-dcap-collateral.sh
@@ -10,9 +10,8 @@
 # the Trustee dcap_verifier in file:// mode. This is platform-level data that
 # does NOT change per-cluster — only per CPU family (identified by FMSPC).
 #
-# IMPORTANT: Run AFTER pcsclient.py cache (E-3). The cache step provisions PCK
-# certs for the specific platform. This fetch step gets the verification collateral
-# (QeIdentity, TcbInfo, CRLs) needed by KBS to verify attestation reports.
+# This collection is independent of platform registration and PCK cache generation.
+# It gets public verification collateral (QeIdentity, TcbInfo, CRLs) needed by KBS.
 #
 # Usage:
 #   ./scripts/collect-dcap-collateral.sh [OPTIONS]
@@ -34,7 +33,7 @@
 set -euo pipefail
 
 # Defaults
-PCSCLIENT_DIR="${HOME}/confidential-computing.tee.dcap/tools/PcsClientTool"
+PCSCLIENT_DIR="${PCSCLIENT_DIR:-${DCAP_PCSCLIENT_DIR:-${HOME}/confidential-computing.tee.dcap/tools/PcsClientTool}}"
 OUTPUT_DIR="${HOME}/.coco-pattern/dcap-offline"
 
 # Parse arguments
diff --git a/scripts/dcap-pck.py b/scripts/dcap-pck.py
new file mode 100644
index 00000000..2ab82301
--- /dev/null
+++ b/scripts/dcap-pck.py
@@ -0,0 +1,673 @@
+#!/usr/bin/env python3
+"""Create and import Intel DCAP PCK cache bundles for offline QGS mode."""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import hashlib
+import json
+import os
+import re
+import shutil
+import struct
+import subprocess
+import sys
+import tempfile
+from datetime import UTC, datetime
+from getpass import getpass
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
+
+SCHEMA_VERSION = 1
+REQUEST_TYPE = "dcap-platform-request"
+RESPONSE_TYPE = "dcap-pck-response"
+PLATFORM_FIELDS = (
+    "enc_ppid",
+    "pce_id",
+    "cpu_svn",
+    "pce_svn",
+    "qe_id",
+    "platform_manifest",
+)
+HEX_LENGTHS = {"pce_id": 4, "cpu_svn": 32, "pce_svn": 4, "qe_id": 32}
+MAX_CACHE_BYTES = 750 * 1024
+
+
+def fail(message: str) -> None:
+    raise ValueError(message)
+
+
+def sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as source:
+        for block in iter(lambda: source.read(1024 * 1024), b""):
+            digest.update(block)
+    return digest.hexdigest()
+
+
+def write_json(path: Path, value: Any) -> None:
+    path.write_text(
+        json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8"
+    )
+
+
+def load_json(path: Path) -> Any:
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as error:
+        fail(f"cannot read JSON file {path}: {error}")
+
+
+def validate_platform(platform: dict[str, Any]) -> dict[str, str]:
+    if set(platform) != set(PLATFORM_FIELDS):
+        fail("platform data must contain exactly: " + ", ".join(PLATFORM_FIELDS))
+    normalized: dict[str, str] = {}
+    for field in PLATFORM_FIELDS:
+        value = platform[field]
+        if not isinstance(value, str):
+            fail(f"platform {field} must be a string")
+        normalized[field] = value.lower()
+
+    for field, length in HEX_LENGTHS.items():
+        if not re.fullmatch(rf"[0-9a-f]{{{length}}}", normalized[field]):
+            fail(f"platform {field} must be {length} hexadecimal characters")
+    if not re.fullmatch(r"[0-9a-f]+", normalized["platform_manifest"]):
+        fail("platform_manifest must be non-empty hexadecimal data")
+    if normalized["enc_ppid"] and not re.fullmatch(
+        r"[0-9a-f]+", normalized["enc_ppid"]
+    ):
+        fail("enc_ppid must be empty or hexadecimal data")
+    return normalized
+
+
+def platform_key(platform: dict[str, str]) -> tuple[str, str]:
+    return platform["qe_id"], platform["pce_id"]
+
+
+def validate_platforms(platforms: Any) -> list[dict[str, str]]:
+    if not isinstance(platforms, list) or not platforms:
+        fail("platform list must contain at least one platform")
+    normalized = [
+        validate_platform(platform)
+        for platform in platforms
+        if isinstance(platform, dict)
+    ]
+    if len(normalized) != len(platforms):
+        fail("platform list entries must be objects")
+    keys = [platform_key(platform) for platform in normalized]
+    if len(set(keys)) != len(keys):
+        fail("platform list contains duplicate QE ID/PCE ID pairs")
+    qe_ids = [platform["qe_id"] for platform in normalized]
+    if len(set(qe_ids)) != len(qe_ids):
+        fail("platform list contains duplicate QE IDs")
+    return sorted(normalized, key=platform_key)
+
+
+def create_manifest(
+    bundle_type: str, files: dict[str, Path], **extra: Any
+) -> dict[str, Any]:
+    return {
+        "schemaVersion": SCHEMA_VERSION,
+        "type": bundle_type,
+        "createdAt": datetime.now(UTC).isoformat(),
+        "files": {
+            name: {"sha256": sha256(path), "size": path.stat().st_size}
+            for name, path in files.items()
+        },
+        **extra,
+    }
+
+
+def validate_manifest(bundle: Path, expected_type: str) -> dict[str, Any]:
+    manifest = load_json(bundle / "manifest.json")
+    if not isinstance(manifest, dict):
+        fail("bundle manifest must be an object")
+    if (
+        manifest.get("schemaVersion") != SCHEMA_VERSION
+        or manifest.get("type") != expected_type
+    ):
+        fail(f"unsupported bundle type or schema in {bundle / 'manifest.json'}")
+    files = manifest.get("files")
+    if not isinstance(files, dict) or not files:
+        fail("bundle manifest has no files")
+    for relative, metadata in files.items():
+        path = bundle / relative
+        if (
+            not isinstance(relative, str)
+            or path.is_symlink()
+            or not path.is_file()
+            or not path.resolve().is_relative_to(bundle.resolve())
+        ):
+            fail(f"invalid bundle file: {relative}")
+        if (
+            not isinstance(metadata, dict)
+            or metadata.get("size") != path.stat().st_size
+            or metadata.get("sha256") != sha256(path)
+        ):
+            fail(f"checksum mismatch for {relative}")
+    return manifest
+
+
+def request_platforms(bundle: Path) -> tuple[dict[str, Any], list[dict[str, str]]]:
+    manifest = validate_manifest(bundle, REQUEST_TYPE)
+    if "platform-list.json" not in manifest["files"]:
+        fail("request bundle is missing platform-list.json")
+    return manifest, validate_platforms(load_json(bundle / "platform-list.json"))
+
+
+def run_oc(arguments: list[str], input_text: str | None = None) -> str:
+    result = subprocess.run(
+        ["oc", *arguments],
+        input=input_text,
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+    if result.returncode:
+        detail = result.stderr.strip() or result.stdout.strip()
+        fail(f"oc {' '.join(arguments[:3])} failed: {detail}")
+    return result.stdout
+
+
+def cluster_platforms(namespace: str) -> list[dict[str, str]]:
+    data = json.loads(
+        run_oc(
+            [
+                "get",
+                "secrets",
+                "-n",
+                namespace,
+                "-l",
+                "type=platform-data",
+                "-o",
+                "json",
+            ]
+        )
+    )
+    platforms = []
+    for secret in data.get("items", []):
+        encoded = secret.get("data", {})
+        try:
+            decoded = {
+                field: base64.b64decode(encoded[field]).decode("ascii")
+                for field in PLATFORM_FIELDS
+            }
+        except (KeyError, UnicodeDecodeError, ValueError) as error:
+            name = secret.get("metadata", {}).get("name", "<unknown>")
+            fail(f"invalid platform-data Secret {name}: {error}")
+        platforms.append(decoded)
+    return validate_platforms(platforms)
+
+
+def command_export(arguments: SimpleNamespace) -> None:
+    output = Path(arguments.output).expanduser()
+    if output.exists():
+        fail(f"output bundle already exists: {output}")
+    platforms = cluster_platforms(arguments.namespace)
+    output.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+    output.mkdir(mode=0o700, parents=True)
+    platform_list = output / "platform-list.json"
+    write_json(platform_list, platforms)
+    write_json(
+        output / "manifest.json",
+        create_manifest(
+            REQUEST_TYPE,
+            {"platform-list.json": platform_list},
+            namespace=arguments.namespace,
+            platforms=[
+                {"qe_id": item["qe_id"], "pce_id": item["pce_id"]} for item in platforms
+            ],
+        ),
+    )
+    print(f"Created platform request bundle: {output}")
+
+
+def read_api_key() -> str:
+    api_key = os.environ.pop("INTEL_PCS_API_KEY", "")
+    if not api_key:
+        if not sys.stdin.isatty():
+            fail(
+                "INTEL_PCS_API_KEY is not set and no interactive terminal is available"
+            )
+        api_key = getpass("Intel PCS API key: ")
+    if not api_key.strip():
+        fail("Intel PCS API key is empty")
+    return api_key
+
+
+def generate_cache(
+    tool_dir: Path, platforms_file: Path, output: Path, expire_hours: int, api_key: str
+) -> None:
+    if not (tool_dir / "pcsclient.py").is_file():
+        fail(f"pcsclient.py not found in {tool_dir}; run make dcap-tools first")
+    sys.path.insert(0, str(tool_dir))
+    try:
+        from pcsclient import CacheCreator  # type: ignore[import-not-found]
+    except ImportError as error:
+        fail(f"cannot import Intel PcsClientTool from {tool_dir}: {error}")
+
+    class Credentials:
+        def get_pcs_api_key(self) -> str:
+            return api_key
+
+    cache_args = SimpleNamespace(
+        url=None,
+        input_file=str(platforms_file),
+        output_dir=str(output),
+        expire=expire_hours,
+        tcb_update_type="early",
+        sub_dir=False,
+    )
+    previous = Path.cwd()
+    try:
+        os.chdir(platforms_file.parent)
+        CacheCreator(Credentials(), cache_args).generate_cache()
+    finally:
+        os.chdir(previous)
+
+
+def cache_metadata(path: Path) -> dict[str, int]:
+    if path.stat().st_size > MAX_CACHE_BYTES:
+        fail(f"cache file exceeds {MAX_CACHE_BYTES} bytes: {path.name}")
+    with path.open("rb") as cache:
+        data = cache.read(14)
+    if len(data) != 14:
+        fail(f"cache file is too small: {path.name}")
+    version, flags, expires_at = struct.unpack("<HIQ", data[:14])
+    if version != 1 or flags != 4:
+        fail(f"unsupported PCK cache header in {path.name}")
+    if expires_at <= int(datetime.now(UTC).timestamp()):
+        fail(f"PCK cache is expired: {path.name}")
+    return {"expiresAt": expires_at}
+
+
+def expire_hours(value: str) -> int:
+    try:
+        hours = int(value)
+    except ValueError as error:
+        raise argparse.ArgumentTypeError(
+            "must be an integer between 1 and 8760"
+        ) from error
+    if not 1 <= hours <= 8760:
+        raise argparse.ArgumentTypeError("must be between 1 and 8760")
+    return hours
+
+
+def command_generate(arguments: SimpleNamespace) -> None:
+    request = Path(arguments.input).expanduser()
+    request_manifest, platforms = request_platforms(request)
+    output = Path(arguments.output).expanduser()
+    if output.exists():
+        fail(f"output bundle already exists: {output}")
+    api_key = read_api_key()
+    output.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+    temporary = Path(tempfile.mkdtemp(prefix="dcap-pck-", dir=output.parent))
+    try:
+        cache_dir = temporary / "pck"
+        generate_cache(
+            Path(arguments.pcsclient_dir).expanduser(),
+            request / "platform-list.json",
+            cache_dir,
+            arguments.expire_hours,
+            api_key,
+        )
+        if not cache_dir.is_dir():
+            fail("Intel PcsClientTool did not create a PCK cache directory")
+        expected = {f"{item['qe_id']}_{item['pce_id']}" for item in platforms}
+        generated = {path.name for path in cache_dir.iterdir() if path.is_file()}
+        if generated != expected:
+            message = (
+                f"generated PCK cache does not match request; "
+                f"expected {sorted(expected)}, got {sorted(generated)}"
+            )
+            fail(message)
+        files = {f"pck/{path.name}": path for path in sorted(cache_dir.iterdir())}
+        metadata = {path.name: cache_metadata(path) for path in cache_dir.iterdir()}
+        write_json(
+            temporary / "manifest.json",
+            create_manifest(
+                RESPONSE_TYPE,
+                files,
+                requestSha256=sha256(request / "manifest.json"),
+                platforms=platforms,
+                cache=metadata,
+            ),
+        )
+        temporary.rename(output)
+    finally:
+        api_key = ""
+        if temporary.exists():
+            shutil.rmtree(temporary)
+    print(f"Created PCK response bundle: {output}")
+
+
+def response_cache(
+    bundle: Path,
+) -> tuple[dict[str, Any], list[dict[str, str]], dict[tuple[str, str], Path]]:
+    manifest = validate_manifest(bundle, RESPONSE_TYPE)
+    platforms = validate_platforms(manifest.get("platforms"))
+    cache: dict[tuple[str, str], Path] = {}
+    for platform in platforms:
+        name = f"{platform['qe_id']}_{platform['pce_id']}"
+        path = bundle / "pck" / name
+        if f"pck/{name}" not in manifest["files"] or not path.is_file():
+            fail(f"response bundle is missing cache file {name}")
+        cache[platform_key(platform)] = path
+        cache_metadata(path)
+    if set(manifest["files"]) != {f"pck/{path.name}" for path in cache.values()}:
+        fail("response bundle contains unexpected files")
+    return manifest, platforms, cache
+
+
+def qgs_daemonset(namespace: str, configured_name: str) -> str:
+    if configured_name:
+        run_oc(["get", "daemonset", configured_name, "-n", namespace])
+        return configured_name
+
+    pods = json.loads(run_oc(["get", "pods", "-n", namespace, "-o", "json"])).get(
+        "items", []
+    )
+    names = set()
+    for pod in pods:
+        init_containers = pod.get("spec", {}).get("initContainers", [])
+        if not any(
+            container.get("name") == "pck-certs-watcher"
+            for container in init_containers
+        ):
+            continue
+        for owner in pod.get("metadata", {}).get("ownerReferences", []):
+            if owner.get("kind") == "DaemonSet" and owner.get("name"):
+                names.add(owner["name"])
+    if len(names) != 1:
+        fail(
+            "expected exactly one QGS DaemonSet from pck-certs-watcher pod owners; "
+            "set DCAP_QGS_DAEMONSET to override"
+        )
+    return names.pop()
+
+
+def command_import(arguments: SimpleNamespace) -> None:
+    bundle = Path(arguments.input).expanduser()
+    manifest, expected, cache = response_cache(bundle)
+    current = cluster_platforms(arguments.namespace)
+    if current != expected:
+        fail(
+            "response bundle platform data does not match current platform-data Secrets"
+        )
+    daemonset = qgs_daemonset(arguments.namespace, arguments.qgs_daemonset)
+    resources = []
+    for platform in expected:
+        key = platform_key(platform)
+        secret_name = f"{platform['qe_id']}-pck"
+        source = cache[key]
+        resources.append(
+            run_oc(
+                [
+                    "create",
+                    "secret",
+                    "generic",
+                    secret_name,
+                    "-n",
+                    arguments.namespace,
+                    f"--from-file=certificate={source}",
+                    "--dry-run=client",
+                    "-o",
+                    "yaml",
+                ]
+            )
+        )
+    run_oc(["apply", "-f", "-"], input_text="---\n".join(resources))
+    run_oc(["rollout", "restart", f"daemonset/{daemonset}", "-n", arguments.namespace])
+    run_oc(
+        [
+            "rollout",
+            "status",
+            f"daemonset/{daemonset}",
+            "-n",
+            arguments.namespace,
+            f"--timeout={arguments.timeout}",
+        ]
+    )
+    digest = sha256(bundle / "manifest.json")
+    print(f"Imported {len(cache)} PCK cache Secrets from {bundle} (SHA-256: {digest})")
+
+
+def command_provision(arguments: SimpleNamespace) -> None:
+    request = Path(arguments.request_bundle).expanduser()
+    response = Path(arguments.response_bundle).expanduser()
+    current = cluster_platforms(arguments.namespace)
+
+    if request.exists():
+        _, exported = request_platforms(request)
+        if exported == current:
+            print(f"Reusing matching platform request bundle: {request}")
+        else:
+            shutil.rmtree(request)
+            if response.exists():
+                shutil.rmtree(response)
+            command_export(
+                SimpleNamespace(namespace=arguments.namespace, output=str(request))
+            )
+    else:
+        command_export(
+            SimpleNamespace(namespace=arguments.namespace, output=str(request))
+        )
+
+    request_digest = sha256(request / "manifest.json")
+    if response.exists():
+        response_manifest, response_platforms, _ = response_cache(response)
+        if (
+            response_manifest.get("requestSha256") == request_digest
+            and response_platforms == current
+        ):
+            print(f"Reusing matching PCK response bundle: {response}")
+        else:
+            shutil.rmtree(response)
+            command_generate(
+                SimpleNamespace(
+                    input=str(request),
+                    output=str(response),
+                    pcsclient_dir=arguments.pcsclient_dir,
+                    expire_hours=arguments.expire_hours,
+                )
+            )
+    else:
+        command_generate(
+            SimpleNamespace(
+                input=str(request),
+                output=str(response),
+                pcsclient_dir=arguments.pcsclient_dir,
+                expire_hours=arguments.expire_hours,
+            )
+        )
+
+    command_import(
+        SimpleNamespace(
+            input=str(response),
+            namespace=arguments.namespace,
+            qgs_daemonset=arguments.qgs_daemonset,
+            timeout=arguments.timeout,
+        )
+    )
+
+
+def command_tools(arguments: SimpleNamespace) -> None:
+    repository = Path(arguments.repository).expanduser()
+    tool_dir = Path(arguments.pcsclient_dir).expanduser()
+    repository.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+    if not (repository / ".git").is_dir():
+        result = subprocess.run(
+            [
+                "git",
+                "clone",
+                "https://github.com/intel/confidential-computing.tee.dcap",
+                str(repository),
+            ],
+            check=False,
+        )
+        if result.returncode:
+            fail("could not clone Intel PcsClientTool repository")
+    for command in (
+        [
+            "git",
+            "-C",
+            str(repository),
+            "fetch",
+            "--depth",
+            "1",
+            "origin",
+            arguments.ref,
+        ],
+        ["git", "-C", str(repository), "checkout", "--detach", arguments.ref],
+        [
+            sys.executable,
+            "-m",
+            "pip",
+            "install",
+            "-r",
+            str(tool_dir / "requirements.txt"),
+        ],
+    ):
+        result = subprocess.run(command, check=False)
+        if result.returncode:
+            fail("failed to prepare the pinned Intel PcsClientTool")
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description=__doc__)
+    commands = parser.add_subparsers(dest="command", required=True)
+    tools = commands.add_parser("tools", help="clone the pinned Intel PcsClientTool")
+    tools.add_argument(
+        "--repository",
+        default=os.environ.get("DCAP_PCSCLIENT_REPO", "~/.coco-pattern/intel-dcap"),
+    )
+    tools.add_argument(
+        "--pcsclient-dir",
+        default=os.environ.get(
+            "DCAP_PCSCLIENT_DIR", "~/.coco-pattern/intel-dcap/tools/PcsClientTool"
+        ),
+    )
+    tools.add_argument(
+        "--ref",
+        default=os.environ.get(
+            "DCAP_PCSCLIENT_REF", "64b78f3766e7196d3d2c60e401540f0f853b2deb"
+        ),
+    )
+    tools.set_defaults(func=command_tools)
+    export = commands.add_parser(
+        "export", help="export QGS platform data from the disconnected cluster"
+    )
+    export.add_argument(
+        "--namespace",
+        default=os.environ.get("DCAP_NAMESPACE", "intel-dcap-operator-system"),
+    )
+    export.add_argument(
+        "--output",
+        default=os.environ.get(
+            "DCAP_REQUEST_BUNDLE", "~/.coco-pattern/dcap-pck/platform-request"
+        ),
+    )
+    export.set_defaults(func=command_export)
+    generate = commands.add_parser(
+        "generate", help="generate a PCK response bundle on the connected low side"
+    )
+    generate.add_argument(
+        "--input",
+        default=os.environ.get(
+            "DCAP_REQUEST_BUNDLE", "~/.coco-pattern/dcap-pck/platform-request"
+        ),
+    )
+    generate.add_argument(
+        "--output",
+        default=os.environ.get(
+            "DCAP_RESPONSE_BUNDLE", "~/.coco-pattern/dcap-pck/pck-response"
+        ),
+    )
+    generate.add_argument(
+        "--pcsclient-dir",
+        default=os.environ.get(
+            "DCAP_PCSCLIENT_DIR", "~/.coco-pattern/intel-dcap/tools/PcsClientTool"
+        ),
+    )
+    generate.add_argument(
+        "--expire-hours",
+        type=expire_hours,
+        default=os.environ.get("DCAP_PCK_EXPIRE_HOURS", "8760"),
+        metavar="HOURS",
+    )
+    generate.set_defaults(func=command_generate)
+    importer = commands.add_parser(
+        "import", help="import a PCK response bundle into the disconnected cluster"
+    )
+    importer.add_argument(
+        "--namespace",
+        default=os.environ.get("DCAP_NAMESPACE", "intel-dcap-operator-system"),
+    )
+    importer.add_argument(
+        "--qgs-daemonset", default=os.environ.get("DCAP_QGS_DAEMONSET", "")
+    )
+    importer.add_argument(
+        "--timeout", default=os.environ.get("DCAP_QGS_TIMEOUT", "10m")
+    )
+    importer.add_argument(
+        "--input",
+        default=os.environ.get(
+            "DCAP_RESPONSE_BUNDLE", "~/.coco-pattern/dcap-pck/pck-response"
+        ),
+    )
+    importer.set_defaults(func=command_import)
+    provision = commands.add_parser(
+        "provision", help="resume the connected-bastion PCK lifecycle"
+    )
+    provision.add_argument(
+        "--namespace",
+        default=os.environ.get("DCAP_NAMESPACE", "intel-dcap-operator-system"),
+    )
+    provision.add_argument(
+        "--qgs-daemonset", default=os.environ.get("DCAP_QGS_DAEMONSET", "")
+    )
+    provision.add_argument(
+        "--timeout", default=os.environ.get("DCAP_QGS_TIMEOUT", "10m")
+    )
+    provision.add_argument(
+        "--request-bundle",
+        default=os.environ.get(
+            "DCAP_REQUEST_BUNDLE", "~/.coco-pattern/dcap-pck/platform-request"
+        ),
+    )
+    provision.add_argument(
+        "--response-bundle",
+        default=os.environ.get(
+            "DCAP_RESPONSE_BUNDLE", "~/.coco-pattern/dcap-pck/pck-response"
+        ),
+    )
+    provision.add_argument(
+        "--pcsclient-dir",
+        default=os.environ.get(
+            "DCAP_PCSCLIENT_DIR", "~/.coco-pattern/intel-dcap/tools/PcsClientTool"
+        ),
+    )
+    provision.add_argument(
+        "--expire-hours",
+        type=expire_hours,
+        default=os.environ.get("DCAP_PCK_EXPIRE_HOURS", "8760"),
+        metavar="HOURS",
+    )
+    provision.set_defaults(func=command_provision)
+    return parser
+
+
+def main() -> int:
+    try:
+        arguments = build_parser().parse_args()
+        arguments.func(arguments)
+        return 0
+    except ValueError as error:
+        print(f"ERROR: {error}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/scripts/gen-secrets.sh b/scripts/gen-secrets.sh
index 61edf066..e050bb6c 100755
--- a/scripts/gen-secrets.sh
+++ b/scripts/gen-secrets.sh
@@ -69,8 +69,8 @@ if [ ! -f "${VALUES_FILE}" ]; then
 	echo "  For Bare Metal deployments:"
 	echo "    - Run 'make collect-firmware-refvals' to collect firmware measurements"
 	echo "    - firmwareReferenceValues is already enabled by default; no need to uncomment anything"
-	echo "    - For Intel TDX: run 'make collect-dcap-collateral', then uncomment tdxCollateral"
-	echo "      in the values-secret file for offline attestation"
+	echo "    - For Intel TDX: on the connected low side run 'make collect-dcap-collateral',"
+	echo "      transfer the output to this host, then uncomment tdxCollateral for offline attestation"
 	echo "    - SSH debug is optional (uncomment sshKey if needed)"
 	echo "    - See docs/firmware-reference-values.md for reference value collection"
 	echo
diff --git a/tests/test_dcap_pck.py b/tests/test_dcap_pck.py
new file mode 100644
index 00000000..2694a0a7
--- /dev/null
+++ b/tests/test_dcap_pck.py
@@ -0,0 +1,208 @@
+"""Unit tests for the local DCAP PCK bundle contract."""
+
+import importlib.util
+import json
+import struct
+from pathlib import Path
+
+import pytest
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
+SPEC = importlib.util.spec_from_file_location(
+    "dcap_pck", REPOSITORY_ROOT / "scripts/dcap-pck.py"
+)
+assert SPEC and SPEC.loader
+dcap_pck = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(dcap_pck)
+
+
+def platform(qe_id: str = "a" * 32, pce_id: str = "0000") -> dict[str, str]:
+    return {
+        "enc_ppid": "",
+        "pce_id": pce_id,
+        "cpu_svn": "b" * 32,
+        "pce_svn": "0000",
+        "qe_id": qe_id,
+        "platform_manifest": "c0ffee",
+    }
+
+
+def test_platform_validation_rejects_duplicate_qe_pce_pairs():
+    with pytest.raises(ValueError, match="duplicate"):
+        dcap_pck.validate_platforms([platform(), platform()])
+
+
+def test_platform_validation_rejects_duplicate_qe_ids():
+    with pytest.raises(ValueError, match="duplicate QE IDs"):
+        dcap_pck.validate_platforms([platform(), platform(pce_id="0001")])
+
+
+def test_request_bundle_detects_tampered_platform_list(tmp_path: Path):
+    bundle = tmp_path / "request"
+    bundle.mkdir()
+    platforms = bundle / "platform-list.json"
+    dcap_pck.write_json(platforms, [platform()])
+    dcap_pck.write_json(
+        bundle / "manifest.json",
+        dcap_pck.create_manifest(
+            dcap_pck.REQUEST_TYPE, {"platform-list.json": platforms}
+        ),
+    )
+    platforms.write_text("[]\n", encoding="utf-8")
+
+    with pytest.raises(ValueError, match="checksum mismatch"):
+        dcap_pck.request_platforms(bundle)
+
+
+def test_response_bundle_rejects_expired_cache(tmp_path: Path):
+    bundle = tmp_path / "response"
+    cache_dir = bundle / "pck"
+    cache_dir.mkdir(parents=True)
+    cache = cache_dir / ("a" * 32 + "_0000")
+    cache.write_bytes(struct.pack("<HIQ", 1, 4, 1))
+    dcap_pck.write_json(
+        bundle / "manifest.json",
+        dcap_pck.create_manifest(
+            dcap_pck.RESPONSE_TYPE,
+            {f"pck/{cache.name}": cache},
+            platforms=[platform()],
+        ),
+    )
+
+    with pytest.raises(ValueError, match="expired"):
+        dcap_pck.response_cache(bundle)
+
+
+def test_read_api_key_removes_environment_value(monkeypatch):
+    monkeypatch.setenv("INTEL_PCS_API_KEY", "test-canary")
+
+    assert dcap_pck.read_api_key() == "test-canary"
+    assert "INTEL_PCS_API_KEY" not in dcap_pck.os.environ
+
+
+def test_platform_data_decoding_uses_exact_required_fields(monkeypatch):
+    encoded = {
+        key: dcap_pck.base64.b64encode(value.encode()).decode()
+        for key, value in platform().items()
+    }
+    monkeypatch.setattr(
+        dcap_pck, "run_oc", lambda _: json.dumps({"items": [{"data": encoded}]})
+    )
+
+    assert dcap_pck.cluster_platforms("intel-dcap-operator-system") == [platform()]
+
+
+def test_response_bundle_rejects_cache_larger_than_secret_limit(tmp_path: Path):
+    cache = tmp_path / "oversized"
+    cache.write_bytes(b"x" * (dcap_pck.MAX_CACHE_BYTES + 1))
+
+    with pytest.raises(ValueError, match="exceeds"):
+        dcap_pck.cache_metadata(cache)
+
+
+def test_import_refuses_changed_platform_data(monkeypatch, tmp_path: Path):
+    response = tmp_path / "response"
+    cache_dir = response / "pck"
+    cache_dir.mkdir(parents=True)
+    cache = cache_dir / ("a" * 32 + "_0000")
+    cache.write_bytes(struct.pack("<HIQ", 1, 4, 4_000_000_000))
+    dcap_pck.write_json(
+        response / "manifest.json",
+        dcap_pck.create_manifest(
+            dcap_pck.RESPONSE_TYPE, {f"pck/{cache.name}": cache}, platforms=[platform()]
+        ),
+    )
+    changed = platform()
+    changed["cpu_svn"] = "d" * 32
+    monkeypatch.setattr(dcap_pck, "cluster_platforms", lambda _: [changed])
+
+    with pytest.raises(ValueError, match="platform data does not match"):
+        dcap_pck.command_import(
+            type(
+                "Arguments",
+                (),
+                {
+                    "input": str(response),
+                    "namespace": "test",
+                    "qgs_daemonset": "",
+                    "timeout": "1m",
+                },
+            )()
+        )
+
+
+def test_qgs_daemonset_uses_pod_owner_reference(monkeypatch):
+    monkeypatch.setattr(
+        dcap_pck,
+        "run_oc",
+        lambda _: json.dumps(
+            {
+                "items": [
+                    {
+                        "metadata": {
+                            "ownerReferences": [
+                                {"kind": "DaemonSet", "name": "intel-tdx-dcap-qgs"}
+                            ]
+                        },
+                        "spec": {"initContainers": [{"name": "pck-certs-watcher"}]},
+                    }
+                ]
+            }
+        ),
+    )
+
+    assert (
+        dcap_pck.qgs_daemonset("intel-dcap-operator-system", "") == "intel-tdx-dcap-qgs"
+    )
+
+
+def test_provision_reuses_matching_request_and_response(
+    monkeypatch, tmp_path: Path, capsys
+):
+    request = tmp_path / "request"
+    response = tmp_path / "response"
+    request.mkdir()
+    request_data = request / "platform-list.json"
+    dcap_pck.write_json(request_data, [platform()])
+    dcap_pck.write_json(
+        request / "manifest.json",
+        dcap_pck.create_manifest(
+            dcap_pck.REQUEST_TYPE, {"platform-list.json": request_data}
+        ),
+    )
+    cache_dir = response / "pck"
+    cache_dir.mkdir(parents=True)
+    cache = cache_dir / ("a" * 32 + "_0000")
+    cache.write_bytes(struct.pack("<HIQ", 1, 4, 4_000_000_000))
+    dcap_pck.write_json(
+        response / "manifest.json",
+        dcap_pck.create_manifest(
+            dcap_pck.RESPONSE_TYPE,
+            {f"pck/{cache.name}": cache},
+            requestSha256=dcap_pck.sha256(request / "manifest.json"),
+            platforms=[platform()],
+        ),
+    )
+    monkeypatch.setattr(dcap_pck, "cluster_platforms", lambda _: [platform()])
+    monkeypatch.setattr(dcap_pck, "command_import", lambda _: None)
+    monkeypatch.setattr(
+        dcap_pck, "command_generate", lambda _: pytest.fail("should reuse response")
+    )
+
+    dcap_pck.command_provision(
+        type(
+            "Arguments",
+            (),
+            {
+                "request_bundle": str(request),
+                "response_bundle": str(response),
+                "namespace": "test",
+                "qgs_daemonset": "",
+                "timeout": "1m",
+                "pcsclient_dir": "unused",
+                "expire_hours": 1,
+            },
+        )()
+    )
+
+    assert "Reusing matching platform request bundle" in capsys.readouterr().out
diff --git a/values-secret.yaml.template b/values-secret.yaml.template
index cc537f11..abdb6275 100644
--- a/values-secret.yaml.template
+++ b/values-secret.yaml.template
@@ -257,7 +257,7 @@ secrets:
   # ===================================================================
 
   # TDX DCAP verification collateral (TCB info, QE identity, PCK CRL)
-  # Collected via: make collect-dcap-collateral FMSPC=<hex> INTEL_PCS_API_KEY=<key>
+  # Collected on a connected low-side host via: make collect-dcap-collateral
   # Loaded into Vault, delivered to KBS via ESO + kbsLocalCertCacheSpec
   # Refresh when collateral expires (~30-90 days)
   # Uncomment after running 'make collect-dcap-collateral':
@@ -322,4 +322,3 @@ secrets:
 #        path: ~/.coco-pattern/mirror-registry-password
 #        onMissingValue: error
 # AIRGAP-MIRROR-SECRETS-END
-