diff --git a/.github/workflows/vulkan-gpu-gate.yml b/.github/workflows/vulkan-gpu-gate.yml
new file mode 100644
index 000000000..6cd3cc4b9
--- /dev/null
+++ b/.github/workflows/vulkan-gpu-gate.yml
@@ -0,0 +1,392 @@
+name: Vulkan GPU Gate
+
+on:
+  # The controller is loaded from the protected default branch. It may inspect
+  # fork metadata, but it never checks out or executes a fork head.
+  pull_request_target:
+    branches: [main]
+    types: [opened, synchronize, reopened, ready_for_review]
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+concurrency:
+  group: vulkan-gpu-gate-${{ github.event.pull_request.merge_commit_sha || github.event.pull_request.head.sha || github.sha }}
+  queue: max
+
+jobs:
+  trust_policy:
+    name: Trust Policy
+    runs-on: ubuntu-latest
+    timeout-minutes: 5
+    outputs:
+      trusted: ${{ steps.policy.outputs.trusted }}
+      target_sha: ${{ steps.policy.outputs.target_sha }}
+      reason: ${{ steps.policy.outputs.reason }}
+    steps:
+      - name: Classify source without executing it
+        id: policy
+        shell: bash
+        env:
+          EVENT_NAME: ${{ github.event_name }}
+          HEAD_REPOSITORY: ${{ github.event.pull_request.head.repo.full_name }}
+          BASE_REPOSITORY: ${{ github.repository }}
+          WORKFLOW_REF: ${{ github.ref }}
+          PR_DRAFT: ${{ github.event.pull_request.draft }}
+          TARGET_SHA: ${{ github.event.pull_request.merge_commit_sha || github.event.pull_request.head.sha || github.sha }}
+        run: |
+          set -euo pipefail
+          trusted=true
+          reason=trusted
+          if [[ "$EVENT_NAME" == "pull_request_target" && "$HEAD_REPOSITORY" != "$BASE_REPOSITORY" ]]; then
+            trusted=false
+            reason=fork_requires_reviewed_mirror
+          fi
+          if [[ "$EVENT_NAME" == "pull_request_target" && "$PR_DRAFT" == "true" ]]; then
+            trusted=false
+            reason=draft_requires_ready_for_review
+          fi
+          if [[ "$EVENT_NAME" == "workflow_dispatch" && "$WORKFLOW_REF" != "refs/heads/main" ]]; then
+            trusted=false
+            reason=manual_dispatch_requires_main
+          fi
+          echo "trusted=$trusted" >> "$GITHUB_OUTPUT"
+          echo "target_sha=$TARGET_SHA" >> "$GITHUB_OUTPUT"
+          echo "reason=$reason" >> "$GITHUB_OUTPUT"
+          if [[ "$trusted" != "true" ]]; then
+            echo "Persistent GPU execution denied: $reason"
+            echo "Fork changes must be reviewed and mirrored to a branch in $BASE_REPOSITORY."
+          fi
+
+  status_pending:
+    name: Initialize Gate Status
+    runs-on: ubuntu-latest
+    if: always() && needs.trust_policy.result == 'success'
+    needs: trust_policy
+    timeout-minutes: 5
+    permissions:
+      statuses: write
+    steps:
+      - name: Mark the tested commit pending before GPU execution
+        shell: bash
+        env:
+          GH_TOKEN: ${{ github.token }}
+          TARGET_SHA: ${{ needs.trust_policy.outputs.target_sha }}
+        run: |
+          set -euo pipefail
+          run_url="$GITHUB_SERVER_URL/$GITHUB_REPOSITORY/actions/runs/$GITHUB_RUN_ID"
+          gh api --method POST \
+            "repos/$GITHUB_REPOSITORY/statuses/$TARGET_SHA" \
+            -f state=pending \
+            -f context="Vulkan GPU Gate" \
+            -f description="Real Vulkan GPU validation is queued" \
+            -f target_url="$run_url" >/dev/null
+
+  gpu_validation:
+    name: Vulkan GPU Validation
+    needs: [trust_policy, status_pending]
+    if: >-
+      needs.trust_policy.outputs.trusted == 'true' &&
+      needs.status_pending.result == 'success'
+    runs-on: [self-hosted, Windows, X64, label-valerie]
+    environment: vulkan-gpu-persistent
+    timeout-minutes: 120
+    concurrency:
+      group: moerengine-gpu-valerie
+      queue: max
+    env:
+      BUILD_DIR: ${{ runner.temp }}\moerengine-gpu-${{ github.run_id }}-${{ github.run_attempt }}\build
+      OUTPUT_DIR: ${{ runner.temp }}\moerengine-gpu-${{ github.run_id }}-${{ github.run_attempt }}\output
+      EVIDENCE_DIR: ${{ github.workspace }}\target\validation\gpu-ci\${{ github.run_id }}-${{ github.run_attempt }}
+      MOER_GATE_SHA: ${{ needs.trust_policy.outputs.target_sha }}
+      MOER_GATE_PR_NUMBER: ${{ github.event.pull_request.number }}
+      MOER_GATE_RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}
+    steps:
+      - name: Checkout trusted source
+        uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4
+        with:
+          repository: ${{ github.repository }}
+          ref: ${{ needs.trust_policy.outputs.target_sha }}
+          clean: true
+          fetch-depth: 1
+          persist-credentials: false
+          submodules: recursive
+
+      - name: Parser contract tests
+        timeout-minutes: 5
+        shell: powershell
+        run: |
+          $ErrorActionPreference = "Stop"
+          New-Item -ItemType Directory -Force -Path $env:EVIDENCE_DIR | Out-Null
+          $stdoutPath = Join-Path $env:EVIDENCE_DIR "runner-tests.stdout.log"
+          $stderrPath = Join-Path $env:EVIDENCE_DIR "runner-tests.stderr.log"
+          $process = Start-Process -FilePath python `
+            -ArgumentList @("-B", "tools/threading/test_run_parallel_record_vulkan_test.py") `
+            -NoNewWindow -Wait -PassThru `
+            -RedirectStandardOutput $stdoutPath `
+            -RedirectStandardError $stderrPath
+          $stdoutText = [string](Get-Content -Raw $stdoutPath)
+          $stderrText = [string](Get-Content -Raw $stderrPath)
+          Set-Content -NoNewline -Encoding utf8 `
+            "$env:EVIDENCE_DIR\runner-tests.log" `
+            ($stdoutText + $stderrText)
+          if ($stdoutText) { Write-Host $stdoutText }
+          if ($stderrText) { Write-Host $stderrText }
+          if ($process.ExitCode -ne 0) {
+            throw "Vulkan gate parser tests failed with exit code $($process.ExitCode)"
+          }
+
+      - name: Qualify runner, build, and execute GPU matrix
+        timeout-minutes: 105
+        shell: powershell
+        run: |
+          $ErrorActionPreference = "Stop"
+          function Invoke-NativeWithLog {
+            param(
+              [Parameter(Mandatory = $true)][string]$FilePath,
+              [Parameter(Mandatory = $true)][string[]]$ArgumentList,
+              [Parameter(Mandatory = $true)][string]$LogPath
+            )
+            $stdoutPath = "$LogPath.stdout"
+            $stderrPath = "$LogPath.stderr"
+            $process = Start-Process -FilePath $FilePath `
+              -ArgumentList $ArgumentList `
+              -NoNewWindow -Wait -PassThru `
+              -RedirectStandardOutput $stdoutPath `
+              -RedirectStandardError $stderrPath
+            $stdoutText = [string](Get-Content -Raw $stdoutPath)
+            $stderrText = [string](Get-Content -Raw $stderrPath)
+            Set-Content -NoNewline -Encoding utf8 $LogPath ($stdoutText + $stderrText)
+            if ($stdoutText) { Write-Host $stdoutText }
+            if ($stderrText) { Write-Host $stderrText }
+            return $process.ExitCode
+          }
+
+          $preflight = Join-Path $env:EVIDENCE_DIR "preflight"
+          $cmakeEvidence = Join-Path $env:EVIDENCE_DIR "cmake"
+          New-Item -ItemType Directory -Force -Path $preflight, $cmakeEvidence | Out-Null
+
+          $actualSha = (& git rev-parse HEAD).Trim()
+          if ($LASTEXITCODE -ne 0 -or $actualSha -ne $env:MOER_GATE_SHA) {
+            throw "Checkout provenance mismatch: expected $env:MOER_GATE_SHA, got $actualSha"
+          }
+          @(
+            "requested_sha=$env:MOER_GATE_SHA"
+            "actual_sha=$actualSha"
+            "pull_request=$env:MOER_GATE_PR_NUMBER"
+            "run_url=$env:MOER_GATE_RUN_URL"
+          ) | Set-Content -Encoding utf8 (Join-Path $preflight "provenance.txt")
+
+          $vswhere = "${env:ProgramFiles(x86)}\Microsoft Visual Studio\Installer\vswhere.exe"
+          if (-not (Test-Path -LiteralPath $vswhere)) {
+            throw "vswhere.exe is required on the GPU runner"
+          }
+          $vsInstall = & $vswhere -latest -products * `
+            -requires Microsoft.VisualStudio.Component.VC.Tools.x86.x64 `
+            -property installationPath
+          if (-not $vsInstall) {
+            throw "Visual Studio C++ tools are required on the GPU runner"
+          }
+          $vsDevCmd = Join-Path $vsInstall "Common7\Tools\VsDevCmd.bat"
+          $environmentLines = & cmd.exe /d /s /c `
+            "`"$vsDevCmd`" -no_logo -arch=x64 -host_arch=x64 >nul && set"
+          if ($LASTEXITCODE -ne 0) {
+            throw "VsDevCmd failed with exit code $LASTEXITCODE"
+          }
+          foreach ($line in $environmentLines) {
+            if ($line -match '^([^=]+)=(.*)$') {
+              Set-Item -Path "env:$($Matches[1])" -Value $Matches[2]
+            }
+          }
+
+          foreach ($command in @("cmake", "ninja", "clang", "clang++", "python", "rc", "link", "nvidia-smi", "vulkaninfo")) {
+            if (-not (Get-Command $command -ErrorAction SilentlyContinue)) {
+              throw "Required command is unavailable: $command"
+            }
+          }
+
+          $cmakeVersionText = (& cmake --version | Select-Object -First 1)
+          if ($cmakeVersionText -notmatch '(\d+\.\d+\.\d+)') {
+            throw "Unable to parse CMake version: $cmakeVersionText"
+          }
+          $cmakeVersion = [version]$Matches[1]
+          if ($cmakeVersion -lt [version]'3.26.0' -or $cmakeVersion -ge [version]'4.0.0') {
+            throw "CMake >=3.26 and <4.0 is required; found $cmakeVersion"
+          }
+          $clangVersionText = (& clang --version | Select-Object -First 1)
+          if ($clangVersionText -notmatch 'clang version (\d+)\.') {
+            throw "Unable to parse Clang version: $clangVersionText"
+          }
+          $clangMajor = [int]$Matches[1]
+          if ($clangMajor -ne 22) {
+            throw "The qualified GPU runner requires LLVM/Clang 22; found $clangVersionText"
+          }
+          $pythonVersion = & python -c "import platform; print(platform.python_version())"
+          if ([version]$pythonVersion -lt [version]'3.10.0') {
+            throw "Python 3.10 or newer is required; found $pythonVersion"
+          }
+
+          @(
+            $cmakeVersionText
+            (& ninja --version)
+            $clangVersionText
+            "Python $pythonVersion"
+            "Visual Studio: $vsInstall"
+          ) | Set-Content -Encoding utf8 (Join-Path $preflight "toolchain.txt")
+
+          $nvidiaLog = Join-Path $preflight "nvidia-smi.csv"
+          $nvidiaExit = Invoke-NativeWithLog `
+            -FilePath "nvidia-smi" `
+            -ArgumentList @(
+              "--query-gpu=name,pci.bus_id,pci.device_id,driver_version",
+              "--format=csv,noheader"
+            ) `
+            -LogPath $nvidiaLog
+          $nvidiaText = Get-Content -Raw $nvidiaLog
+          if ($nvidiaExit -ne 0 -or $nvidiaText -notmatch 'RTX 5080.*0x2C0210DE') {
+            throw "The qualified runner must expose the registered RTX 5080 (PCI device 0x2C0210DE)"
+          }
+
+          $vulkanLog = Join-Path $preflight "vulkaninfo-summary.txt"
+          $vulkanInfoExit = Invoke-NativeWithLog `
+            -FilePath "vulkaninfo" `
+            -ArgumentList @("--summary") `
+            -LogPath $vulkanLog
+          $vulkanText = Get-Content -Raw $vulkanLog
+          if ($vulkanInfoExit -ne 0) {
+            throw "vulkaninfo --summary failed with exit code $vulkanInfoExit"
+          }
+          if ($vulkanText -notmatch 'VK_LAYER_KHRONOS_validation') {
+            throw "VK_LAYER_KHRONOS_validation is unavailable"
+          }
+          if ($vulkanText -notmatch 'apiVersion\s*=\s*1\.[3-9]\.\d+') {
+            throw "No Vulkan 1.3+ device was reported"
+          }
+
+          Copy-Item template.MoerEngine.toml MoerEngine.toml -Force
+          $configureArguments = @(
+            "-S", ".",
+            "-B", "`"$env:BUILD_DIR`"",
+            "-G", "Ninja",
+            "-DCMAKE_BUILD_TYPE=Debug",
+            "-DCMAKE_C_COMPILER=clang",
+            "-DCMAKE_CXX_COMPILER=clang++",
+            "-DBINARY_ROOT_DIR=`"$env:OUTPUT_DIR`"",
+            "-Dmoer_build_test=ON",
+            "-DMOER_IGNORE_ENABLE_FEATURES=ON",
+            "-DWITH_CUDA=OFF",
+            "-DWITH_NRD=OFF",
+            "-DWITH_RENDERDOC=OFF",
+            "-DWITH_PROFILE=OFF"
+          )
+          $configureExit = Invoke-NativeWithLog `
+            -FilePath "cmake" `
+            -ArgumentList $configureArguments `
+            -LogPath (Join-Path $cmakeEvidence "configure.log")
+          if ($configureExit -ne 0) {
+            throw "CMake configure failed with exit code $configureExit"
+          }
+
+          $buildExit = Invoke-NativeWithLog `
+            -FilePath "cmake" `
+            -ArgumentList @(
+              "--build", "`"$env:BUILD_DIR`"",
+              "--target", "TestRHIParallelRecordVulkan",
+              "--parallel", "30"
+            ) `
+            -LogPath (Join-Path $cmakeEvidence "build.log")
+          if ($buildExit -ne 0) {
+            throw "GPU test build failed with exit code $buildExit"
+          }
+
+          $testExecutable = Join-Path $env:OUTPUT_DIR "bin\Debug\TestRHIParallelRecordVulkan.exe"
+          $modeOutput = Join-Path $env:EVIDENCE_DIR "modes"
+          $matrixExit = Invoke-NativeWithLog `
+            -FilePath "python" `
+            -ArgumentList @(
+              "-B", "tools/threading/run_parallel_record_vulkan_test.py",
+              "--executable", "`"$testExecutable`"",
+              "--outdir", "`"$modeOutput`"",
+              "--timeout", "180",
+              "--strict-gpu-gate",
+              "--require-vendor-id", "0x000010de",
+              "--require-device-id", "0x00002c02",
+              "--require-device-type", "discrete_gpu",
+              "--minimum-device-api", "1.3.0"
+            ) `
+            -LogPath (Join-Path $env:EVIDENCE_DIR "matrix-runner.log")
+          if ($matrixExit -ne 0) {
+            throw "Vulkan GPU matrix failed with exit code $matrixExit"
+          }
+
+      - name: Collect CMake evidence
+        if: always()
+        shell: powershell
+        run: |
+          $ErrorActionPreference = "Stop"
+          $cmakeEvidence = Join-Path $env:EVIDENCE_DIR "cmake"
+          New-Item -ItemType Directory -Force -Path $cmakeEvidence | Out-Null
+          $cache = Join-Path $env:BUILD_DIR "CMakeCache.txt"
+          if (Test-Path -LiteralPath $cache) {
+            Copy-Item $cache $cmakeEvidence -Force
+          }
+          $configureLog = Join-Path $env:BUILD_DIR "CMakeFiles\CMakeConfigureLog.yaml"
+          if (Test-Path -LiteralPath $configureLog) {
+            Copy-Item $configureLog $cmakeEvidence -Force
+          }
+
+      - name: Upload GPU evidence
+        if: always()
+        uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4
+        with:
+          name: moer-vulkan-gpu-${{ github.run_id }}-${{ github.run_attempt }}
+          path: ${{ env.EVIDENCE_DIR }}
+          if-no-files-found: error
+          retention-days: 14
+          compression-level: 6
+
+  gate:
+    name: Publish Gate Status
+    runs-on: ubuntu-latest
+    if: always()
+    needs: [trust_policy, status_pending, gpu_validation]
+    timeout-minutes: 5
+    permissions:
+      statuses: write
+    steps:
+      - name: Publish status on the tested merge commit
+        shell: bash
+        env:
+          GH_TOKEN: ${{ github.token }}
+          TARGET_SHA: ${{ needs.trust_policy.outputs.target_sha }}
+          TRUSTED: ${{ needs.trust_policy.outputs.trusted }}
+          TRUST_REASON: ${{ needs.trust_policy.outputs.reason }}
+          TRUST_RESULT: ${{ needs.trust_policy.result }}
+          PENDING_RESULT: ${{ needs.status_pending.result }}
+          GPU_RESULT: ${{ needs.gpu_validation.result }}
+        run: |
+          set -euo pipefail
+          conclusion=failure
+          description="GPU validation failed or was denied"
+          if [[ "$TRUST_RESULT" == "success" && "$PENDING_RESULT" == "success" && "$TRUSTED" == "true" && "$GPU_RESULT" == "success" ]]; then
+            conclusion=success
+            description="21-mode real Vulkan GPU validation passed"
+          fi
+          run_url="$GITHUB_SERVER_URL/$GITHUB_REPOSITORY/actions/runs/$GITHUB_RUN_ID"
+          latest_url="$(gh api \
+            "repos/$GITHUB_REPOSITORY/commits/$TARGET_SHA/statuses?per_page=100" \
+            --jq '[.[] | select(.context == "Vulkan GPU Gate")][0].target_url // ""')"
+          if [[ "$PENDING_RESULT" == "success" && -n "$latest_url" && "$latest_url" != "$run_url" ]]; then
+            echo "A newer run owns Vulkan GPU Gate on $TARGET_SHA; leaving its status unchanged."
+            exit 0
+          fi
+          gh api --method POST \
+            "repos/$GITHUB_REPOSITORY/statuses/$TARGET_SHA" \
+            -f state="$conclusion" \
+            -f context="Vulkan GPU Gate" \
+            -f description="$description" \
+            -f target_url="$run_url" >/dev/null
+          echo "Published Vulkan GPU Gate=$conclusion on $TARGET_SHA"
+          echo "Trust: $TRUST_RESULT/$TRUSTED ($TRUST_REASON); pending: $PENDING_RESULT; GPU: $GPU_RESULT"
+          [[ "$conclusion" == "success" ]]
diff --git a/source/runtime/render/rhi/vulkan/VulkanDevice.cpp b/source/runtime/render/rhi/vulkan/VulkanDevice.cpp
index 6ea62d48e..dfe9591e3 100644
--- a/source/runtime/render/rhi/vulkan/VulkanDevice.cpp
+++ b/source/runtime/render/rhi/vulkan/VulkanDevice.cpp
@@ -94,6 +94,7 @@ VulkanDevice::VulkanDevice(const VulkanRHIConfig&& _config) : RenderDevice::Impl
     CreateInternalResources();
 
     LoadDefaultExtensions();
+    LogGpuEnvironment(_config.api_version);
 }
 
 void VulkanDevice::PostInit() {
@@ -159,6 +160,13 @@ void VulkanDevice::InitVulkanInstance(uint32 _api_version) {
     TLayerArray instance_layers_required;
     VulkanPlatform::GetInstanceLayers(instance_layers_required);
 
+    validation_layer_requested = std::ranges::any_of(
+        instance_layers_required,
+        [](std::string_view layer) { return layer == "VK_LAYER_KHRONOS_validation"; }
+    );
+    validation_layer_available =
+        instance_layers.contains("VK_LAYER_KHRONOS_validation");
+
     VkDebugUtilsMessengerCreateInfoEXT debug_create_info{};
     Array<const char*>                 instance_layers_loaded;
     bool                               b_validation_layer_enabled = false;
@@ -253,8 +261,11 @@ void VulkanDevice::InitVulkanInstance(uint32 _api_version) {
     VK_CHECK_RESULT(vkCreateInstance(&instance_create_info, nullptr, &m_instance))
     volkLoadInstance(m_instance);
 
-    if (b_validation_layer_enabled)
+    if (b_validation_layer_enabled) {
         SetupDebugUtilsMessengerEXT();
+    }
+    validation_layer_enabled =
+        b_validation_layer_enabled && m_debug_utils_messenger != VK_NULL_HANDLE;
 }
 
 /**
@@ -480,6 +491,63 @@ void VulkanDevice::InitGpu(uint32 _api_version) {
         BuildCooperativeExtensionInfo(m_device_info.optional_extensions, m_device_info.optional_properties);
 }
 
+void VulkanDevice::LogGpuEnvironment(uint32 _requested_api_version) const {
+    const auto& properties = m_device_info.core_properties.core_1_0;
+    const char* device_type = "unknown";
+    switch (properties.deviceType) {
+        case VK_PHYSICAL_DEVICE_TYPE_OTHER:
+            device_type = "other";
+            break;
+        case VK_PHYSICAL_DEVICE_TYPE_INTEGRATED_GPU:
+            device_type = "integrated_gpu";
+            break;
+        case VK_PHYSICAL_DEVICE_TYPE_DISCRETE_GPU:
+            device_type = "discrete_gpu";
+            break;
+        case VK_PHYSICAL_DEVICE_TYPE_VIRTUAL_GPU:
+            device_type = "virtual_gpu";
+            break;
+        case VK_PHYSICAL_DEVICE_TYPE_CPU:
+            device_type = "cpu";
+            break;
+        default:
+            break;
+    }
+
+    constexpr char hex_digits[] = "0123456789abcdef";
+    std::string    device_uuid;
+    device_uuid.reserve(VK_UUID_SIZE * 2);
+    for (const uint8_t byte : m_device_info.core_properties.core_1_1.deviceUUID) {
+        device_uuid.push_back(hex_digits[byte >> 4]);
+        device_uuid.push_back(hex_digits[byte & 0x0f]);
+    }
+
+    LOG_INFO(
+        "[Vulkan][GPU_ENV] schema=1 backend=vulkan "
+        "validation_requested={} validation_layer_available={} validation_enabled={} "
+        "device_type={} vendor_id=0x{:08x} device_id=0x{:08x} device_uuid={} "
+        "requested_api={}.{}.{} device_api={}.{}.{} api_variant={} "
+        "device_api_raw=0x{:08x} driver_id={} driver_version_raw=0x{:08x}",
+        validation_layer_requested,
+        validation_layer_available,
+        validation_layer_enabled,
+        device_type,
+        properties.vendorID,
+        properties.deviceID,
+        device_uuid,
+        VK_API_VERSION_MAJOR(_requested_api_version),
+        VK_API_VERSION_MINOR(_requested_api_version),
+        VK_API_VERSION_PATCH(_requested_api_version),
+        VK_API_VERSION_MAJOR(properties.apiVersion),
+        VK_API_VERSION_MINOR(properties.apiVersion),
+        VK_API_VERSION_PATCH(properties.apiVersion),
+        VK_API_VERSION_VARIANT(properties.apiVersion),
+        properties.apiVersion,
+        static_cast<uint32>(m_device_info.core_properties.core_1_2.driverID),
+        properties.driverVersion
+    );
+}
+
 void VulkanDevice::CreateDevice(uint32 _api_version) {
     const uint32_t graphics_family =
         m_device_info.queue_family_indices.graphics.value();
@@ -760,6 +828,24 @@ void VulkanDevice::Destroy() {
     FlushDeferredReleases();
     vmaDestroyAllocator(m_allocator);
     vkDestroyDevice(m_device, VK_NULL_HANDLE);
+    m_device = VK_NULL_HANDLE;
+
+    // The callback coalesces non-error messages. Flush before removing the
+    // messenger so short-lived GPU gate processes cannot lose a final warning.
+    FlushBufferedDebugMessages();
+    if (m_debug_utils_messenger != VK_NULL_HANDLE) {
+        vkDestroyDebugUtilsMessengerEXT(
+            m_instance, m_debug_utils_messenger, VK_NULL_HANDLE
+        );
+        m_debug_utils_messenger = VK_NULL_HANDLE;
+    }
+    if (m_instance != VK_NULL_HANDLE) {
+        vkDestroyInstance(m_instance, VK_NULL_HANDLE);
+        m_instance = VK_NULL_HANDLE;
+    }
+    // The VkInstanceCreateInfo pNext callback is active during instance
+    // destruction even after the explicit messenger is gone.
+    FlushBufferedDebugMessages();
 
     LOG_INFO("VulkanRHI: Device destroyed.");
 
diff --git a/source/runtime/render/rhi/vulkan/VulkanDevice.h b/source/runtime/render/rhi/vulkan/VulkanDevice.h
index c104590cc..9e19e0464 100644
--- a/source/runtime/render/rhi/vulkan/VulkanDevice.h
+++ b/source/runtime/render/rhi/vulkan/VulkanDevice.h
@@ -356,6 +356,9 @@ class VulkanDevice : public RenderDevice::Impl {
     VulkanDeviceInfo m_device_info{};
 
     VkDebugUtilsMessengerEXT m_debug_utils_messenger = VK_NULL_HANDLE;
+    bool                     validation_layer_requested = false;
+    bool                     validation_layer_available = false;
+    bool                     validation_layer_enabled   = false;
 
     VmaAllocator              m_allocator = VK_NULL_HANDLE;
     VulkanDescriptorHeap      m_global_descriptor_heap{};
@@ -414,6 +417,7 @@ class VulkanDevice : public RenderDevice::Impl {
     void             InitVulkanInstance(uint32 _api_version);
     VkPhysicalDevice SelectGpu(uint32 _api_version);
     void             InitGpu(uint32 _api_version);
+    void             LogGpuEnvironment(uint32 _requested_api_version) const;
     void             CreateDevice(uint32 _api_version);
     void             CreateMemoryAllocator(VkInstance _instance, uint32 _api_version);
     void             CreateDescriptorHeap();
diff --git a/tools/threading/README.md b/tools/threading/README.md
index 300685745..25f76975c 100644
--- a/tools/threading/README.md
+++ b/tools/threading/README.md
@@ -359,21 +359,66 @@ changes, and native copy/clear calls contribute units; GPU byte counts,
 dispatch group counts, and indirect draw counts do not. A wave needs at least
 two qualifying jobs, otherwise it safely uses the serial recorder.
 
-Run the nine-mode Vulkan correctness/fallback gate after building the target:
+Run the 21-mode Vulkan correctness/fallback gate after building the target. The
+native integration target is intentionally excluded from default CTest because
+it requires a real Vulkan device:
 
 ```powershell
 python tools/threading/run_parallel_record_vulkan_test.py `
-  --executable target/bin/Release/TestRHIParallelRecordVulkan.exe `
-  --outdir target/validation/parallel_record/release
+  --executable target/bin/Debug/TestRHIParallelRecordVulkan.exe `
+  --outdir target/validation/parallel_record/debug
 ```
 
-The runner checks serial, forced-parallel, injected worker failure, production
-gate rejection, production-heavy admission, hard cross-queue Translate failure
-retirement, multi-segment prefix-submit/suffix-failure retirement, and bounded
-Submission-pipeline windows 1 and 2. It requires real worker overlap, stable
-`wave -> serial island -> wave` assembly, GPU readback correctness, exact
-failure/fallback counts, no native submit after the injected hard fault, and
-clean Vulkan logs. The ready-native-lane gate accepts a PASS only when the
+The matrix covers serial, forced-parallel, injected worker failure,
+production-gate rejection, production-heavy admission, hard cross-queue
+Translate failure retirement, multi-segment prefix-submit/suffix-failure
+retirement, bounded Submission-pipeline windows 1 and 2, Present ownership and
+hard-fault boundaries, RT export rejection, owning readback futures, occlusion
+queries, complete timestamp-query success/rejection/failure ownership, and the
+GPU-scope query-island/parallel-sibling stream. It requires real worker overlap,
+stable `wave -> serial island -> wave` assembly, GPU readback correctness,
+exact failure/fallback counts, no native submit after the injected hard fault,
+and clean Vulkan logs. Every process also emits one fixed-schema
+`[Vulkan][GPU_ENV]` record after complete Vulkan initialization. The record
+proves which physical device ran the test and distinguishes validation being
+requested, available, and actually enabled.
+
+The pull-request workflow `.github/workflows/vulkan-gpu-gate.yml` runs Debug on
+the dedicated Windows/NVIDIA self-hosted runner and enables the fail-closed
+policy:
+
+```powershell
+python tools/threading/run_parallel_record_vulkan_test.py `
+  --executable target/bin/Debug/TestRHIParallelRecordVulkan.exe `
+  --outdir target/validation/gpu-ci/local `
+  --strict-gpu-gate `
+  --require-vendor-id 0x000010de `
+  --require-device-id 0x00002c02 `
+  --require-device-type discrete_gpu `
+  --minimum-device-api 1.3.0
+```
+
+Strict mode requires the same GPU identity in all 21 processes, complete
+validation-layer proof, Vulkan 1.3 or newer, and zero `TESTCASE` skips. It
+preserves separate stdout/stderr plus a combined log for every mode, including
+partial output on timeout, and writes `summary.json` on success. The workflow
+preflights the toolchain, registered RTX 5080 identity, `vulkaninfo`, and the
+Khronos validation layer; evidence is uploaded even when the gate fails. The
+controller is loaded from the protected default branch via
+`pull_request_target`, but it never checks out or executes a fork head. Only a
+same-repository PR test-merge SHA may reach the persistent self-hosted runner;
+fork changes must be reviewed and mirrored to a repository branch. A separate
+GitHub-hosted job publishes the stable `Vulkan GPU Gate` status on the exact
+tested SHA without giving a write token to the GPU runner. The GPU job is also
+bound to the protected `vulkan-gpu-persistent` environment, whose required
+reviewer and no-self-review policy must remain enabled. Draft PRs are denied.
+Repository branch protection/rulesets must require the exact
+`Vulkan GPU Gate` context after the first main-branch qualification run; the
+workflow cannot make its own status required. The runner
+refuses a non-empty output directory so a failed retry cannot inherit stale
+mode logs or a previous PASS summary.
+
+The ready-native-lane gate accepts a PASS only when the
 `G,G,C,Copy` source order produces first-ready `G,C,Copy` Translate lanes and
 an actually observed serial Submission-owner order of `G,G,C,Copy`. When Copy
 aliases Graphics or Compute, the marker must explicitly report
diff --git a/tools/threading/run_parallel_record_vulkan_test.py b/tools/threading/run_parallel_record_vulkan_test.py
index 6d6c494ba..4fb9a6834 100644
--- a/tools/threading/run_parallel_record_vulkan_test.py
+++ b/tools/threading/run_parallel_record_vulkan_test.py
@@ -1,9 +1,13 @@
 from __future__ import annotations
 
 import argparse
+import json
+import math
+import os
 import re
 import subprocess
 import sys
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Sequence
 
@@ -19,6 +23,112 @@
     r"\[ParallelRecord\]\[Injection\].*?point=worker-throw "
     r"phase=after-first-command.*?batch=(?P<batch>\d+)"
 )
+GPU_ENV_LINE_RE = re.compile(r"\[Vulkan\]\[GPU_ENV\][^\r\n]*")
+GPU_ENV_RE = re.compile(
+    r"\[Vulkan\]\[GPU_ENV\] schema=1 backend=vulkan "
+    r"validation_requested=(?P<validation_requested>true|false) "
+    r"validation_layer_available=(?P<validation_layer_available>true|false) "
+    r"validation_enabled=(?P<validation_enabled>true|false) "
+    r"device_type=(?P<device_type>other|integrated_gpu|discrete_gpu|virtual_gpu|cpu|unknown) "
+    r"vendor_id=(?P<vendor_id>0x[0-9a-f]{8}) "
+    r"device_id=(?P<device_id>0x[0-9a-f]{8}) "
+    r"device_uuid=(?P<device_uuid>[0-9a-f]{32}) "
+    r"requested_api=(?P<requested_api>\d+\.\d+\.\d+) "
+    r"device_api=(?P<device_api>\d+\.\d+\.\d+) "
+    r"api_variant=(?P<api_variant>\d+) "
+    r"device_api_raw=(?P<device_api_raw>0x[0-9a-f]{8}) "
+    r"driver_id=(?P<driver_id>\d+) "
+    r"driver_version_raw=(?P<driver_version_raw>0x[0-9a-f]{8})"
+)
+
+
+@dataclass(frozen=True)
+class VulkanTestCase:
+    mode: str
+    arguments: tuple[str, ...]
+
+
+VULKAN_TEST_CASES = (
+    VulkanTestCase("serial", ()),
+    VulkanTestCase("parallel", ("--parallel",)),
+    VulkanTestCase("fallback", ("--parallel", "--inject-worker-failure")),
+    VulkanTestCase("gated", ("--parallel", "--production-gate")),
+    VulkanTestCase("heavy", ("--parallel", "--production-heavy")),
+    VulkanTestCase("translate-hard", ("--inject-translate-failure",)),
+    VulkanTestCase(
+        "multi-segment-hard", ("--inject-multi-segment-translate-failure",)
+    ),
+    VulkanTestCase("pipeline-window1", ("--pipeline-window1",)),
+    VulkanTestCase("pipeline-window2", ("--pipeline-window2",)),
+    VulkanTestCase("present-boundary", ("--present-boundary",)),
+    VulkanTestCase("present-hard", ("--present-hard",)),
+    VulkanTestCase("present-legacy-owner", ("--present-legacy-owner",)),
+    VulkanTestCase(
+        "present-completion-shutdown", ("--present-completion-shutdown",)
+    ),
+    VulkanTestCase("rt-export-rejection", ("--rt-export-rejection",)),
+    VulkanTestCase("readback-future", ("--readback-future",)),
+    VulkanTestCase("occlusion-query", ("--occlusion-query",)),
+    VulkanTestCase("timestamp-query", ("--timestamp-query",)),
+    VulkanTestCase(
+        "timestamp-query-success-batch", ("--timestamp-query-success-batch",)
+    ),
+    VulkanTestCase(
+        "timestamp-query-mid-failure", ("--timestamp-query-mid-failure",)
+    ),
+    VulkanTestCase(
+        "timestamp-query-record-failure", ("--timestamp-query-record-failure",)
+    ),
+    VulkanTestCase("gpu-scope-stream", ("--gpu-scope-stream",)),
+)
+VULKAN_TEST_CASE_BY_MODE = {case.mode: case for case in VULKAN_TEST_CASES}
+assert len(VULKAN_TEST_CASE_BY_MODE) == len(VULKAN_TEST_CASES)
+
+
+@dataclass(frozen=True)
+class GpuEnvironment:
+    validation_requested: bool
+    validation_layer_available: bool
+    validation_enabled: bool
+    device_type: str
+    vendor_id: int
+    device_id: int
+    device_uuid: str
+    requested_api: tuple[int, int, int]
+    device_api: tuple[int, int, int]
+    api_variant: int
+    device_api_raw: int
+    driver_id: int
+    driver_version_raw: int
+
+
+@dataclass(frozen=True)
+class GpuGateRequirements:
+    require_marker: bool = False
+    vendor_id: int | None = None
+    device_id: int | None = None
+    device_type: str | None = None
+    require_validation: bool = False
+    minimum_device_api: tuple[int, int, int] | None = None
+    reject_testcase_skips: bool = False
+
+    @property
+    def active(self) -> bool:
+        return (
+            self.require_marker
+            or self.vendor_id is not None
+            or self.device_id is not None
+            or self.device_type is not None
+            or self.require_validation
+            or self.minimum_device_api is not None
+            or self.reject_testcase_skips
+        )
+
+
+@dataclass(frozen=True)
+class CaseResult:
+    log_path: Path
+    environment: GpuEnvironment | None
 
 
 class VulkanTestError(RuntimeError):
@@ -34,6 +144,184 @@ def _native_ids_have_alias(*native_ids: str) -> bool:
     return len(set(native_ids)) != len(native_ids)
 
 
+def _parse_api_version(value: str) -> tuple[int, int, int]:
+    parts = value.split(".")
+    if len(parts) not in (2, 3) or any(not part.isdigit() for part in parts):
+        raise ValueError(f"expected MAJOR.MINOR[.PATCH], got {value!r}")
+    numbers = tuple(int(part) for part in parts)
+    if len(numbers) == 2:
+        return numbers[0], numbers[1], 0
+    return numbers
+
+
+def _gpu_environment(text: str, *, required: bool) -> GpuEnvironment | None:
+    lines = GPU_ENV_LINE_RE.findall(text)
+    if not lines:
+        _require(not required, "GPU_ENV: expected exactly one marker, got 0")
+        return None
+    _require(
+        len(lines) == 1,
+        f"GPU_ENV: expected exactly one marker, got {len(lines)}",
+    )
+    match = GPU_ENV_RE.fullmatch(lines[0])
+    _require(match is not None, "GPU_ENV: malformed or unsupported schema")
+    assert match is not None
+
+    environment = GpuEnvironment(
+        validation_requested=match.group("validation_requested") == "true",
+        validation_layer_available=(
+            match.group("validation_layer_available") == "true"
+        ),
+        validation_enabled=match.group("validation_enabled") == "true",
+        device_type=match.group("device_type"),
+        vendor_id=int(match.group("vendor_id"), 16),
+        device_id=int(match.group("device_id"), 16),
+        device_uuid=match.group("device_uuid"),
+        requested_api=_parse_api_version(match.group("requested_api")),
+        device_api=_parse_api_version(match.group("device_api")),
+        api_variant=int(match.group("api_variant")),
+        device_api_raw=int(match.group("device_api_raw"), 16),
+        driver_id=int(match.group("driver_id")),
+        driver_version_raw=int(match.group("driver_version_raw"), 16),
+    )
+    encoded_device_api = (
+        (environment.api_variant << 29)
+        | (environment.device_api[0] << 22)
+        | (environment.device_api[1] << 12)
+        | environment.device_api[2]
+    )
+    _require(
+        encoded_device_api == environment.device_api_raw,
+        "GPU_ENV: device_api fields do not match device_api_raw",
+    )
+    return environment
+
+
+def _validate_gpu_environment(
+    environment: GpuEnvironment | None,
+    requirements: GpuGateRequirements,
+) -> None:
+    if not requirements.active:
+        return
+    _require(environment is not None, "GPU_ENV: required marker is missing")
+    assert environment is not None
+    if requirements.vendor_id is not None:
+        _require(
+            environment.vendor_id == requirements.vendor_id,
+            "GPU_ENV: vendor mismatch: "
+            f"expected 0x{requirements.vendor_id:08x}, "
+            f"got 0x{environment.vendor_id:08x}",
+        )
+    if requirements.device_id is not None:
+        _require(
+            environment.device_id == requirements.device_id,
+            "GPU_ENV: device mismatch: "
+            f"expected 0x{requirements.device_id:08x}, "
+            f"got 0x{environment.device_id:08x}",
+        )
+    if requirements.device_type is not None:
+        _require(
+            environment.device_type == requirements.device_type,
+            "GPU_ENV: device type mismatch: "
+            f"expected {requirements.device_type}, got {environment.device_type}",
+        )
+    if requirements.require_validation:
+        _require(
+            environment.validation_requested
+            and environment.validation_layer_available
+            and environment.validation_enabled,
+            "GPU_ENV: validation requested/available/enabled proof is incomplete",
+        )
+    if requirements.minimum_device_api is not None:
+        _require(
+            environment.api_variant == 0,
+            "GPU_ENV: strict gate requires standard Vulkan API variant 0",
+        )
+        _require(
+            environment.requested_api >= requirements.minimum_device_api,
+            "GPU_ENV: requested API is below required minimum: "
+            f"expected >= {'.'.join(map(str, requirements.minimum_device_api))}, "
+            f"got {'.'.join(map(str, environment.requested_api))}",
+        )
+        _require(
+            environment.device_api >= requirements.minimum_device_api,
+            "GPU_ENV: device API is below required minimum: "
+            f"expected >= {'.'.join(map(str, requirements.minimum_device_api))}, "
+            f"got {'.'.join(map(str, environment.device_api))}",
+        )
+
+
+def _validate_consistent_gpu_environments(results: Sequence[CaseResult]) -> None:
+    environments = [result.environment for result in results]
+    if not any(environment is not None for environment in environments):
+        return
+    _require(
+        all(environment is not None for environment in environments),
+        "GPU_ENV: marker presence changed across test modes",
+    )
+    first = environments[0]
+    _require(
+        all(environment == first for environment in environments[1:]),
+        "GPU_ENV: selected device or validation state changed across test modes",
+    )
+
+
+def _write_summary(outdir: Path, results: Sequence[CaseResult]) -> Path:
+    environment = results[0].environment if results else None
+    gpu_environment = None
+    if environment is not None:
+        gpu_environment = {
+            "validation_requested": environment.validation_requested,
+            "validation_layer_available": environment.validation_layer_available,
+            "validation_enabled": environment.validation_enabled,
+            "device_type": environment.device_type,
+            "vendor_id": f"0x{environment.vendor_id:08x}",
+            "device_id": f"0x{environment.device_id:08x}",
+            "device_uuid": environment.device_uuid,
+            "requested_api": ".".join(map(str, environment.requested_api)),
+            "device_api": ".".join(map(str, environment.device_api)),
+            "api_variant": environment.api_variant,
+            "device_api_raw": f"0x{environment.device_api_raw:08x}",
+            "driver_id": environment.driver_id,
+            "driver_version_raw": f"0x{environment.driver_version_raw:08x}",
+        }
+    payload = {
+        "schema": 1,
+        "result": "PASS",
+        "commit": os.environ.get("MOER_GATE_SHA") or os.environ.get("GITHUB_SHA"),
+        "pull_request": os.environ.get("MOER_GATE_PR_NUMBER"),
+        "workflow_run_url": os.environ.get("MOER_GATE_RUN_URL"),
+        "mode_count": len(results),
+        "gpu_environment": gpu_environment,
+        "modes": [
+            {
+                "mode": result.log_path.stem,
+                "log": result.log_path.name,
+                "result": "PASS",
+            }
+            for result in results
+        ],
+    }
+    outdir.mkdir(parents=True, exist_ok=True)
+    summary_path = outdir / "summary.json"
+    summary_path.write_text(
+        json.dumps(payload, indent=2, sort_keys=True) + "\n",
+        encoding="utf-8",
+    )
+    return summary_path
+
+
+def _prepare_outdir(outdir: Path) -> None:
+    if outdir.exists():
+        _require(outdir.is_dir(), f"output path is not a directory: {outdir}")
+        _require(
+            not any(outdir.iterdir()),
+            f"refusing to reuse non-empty output directory: {outdir}",
+        )
+        return
+    outdir.mkdir(parents=True, exist_ok=False)
+
+
 def _testcase_marker(name: str, text: str) -> tuple[str, dict[str, str]] | None:
     matches = list(
         re.finditer(
@@ -378,6 +666,21 @@ def _require_rt_export_rejection(text: str) -> None:
         and fields.get("replay") == "0",
         "rt-export-rejection: incomplete export transaction contract",
     )
+    materialization = _testcase_marker(
+        "RaytracingAcceptedReadbackMaterializationFailure", text
+    )
+    _require(
+        materialization is not None and materialization[0] == "PASS",
+        "rt-export-rejection: missing accepted-readback failure PASS marker",
+    )
+    _, materialization_fields = materialization
+    _require(
+        materialization_fields.get("native") == "accepted"
+        and materialization_fields.get("payload") == "error"
+        and materialization_fields.get("encoder") == "0"
+        and materialization_fields.get("latch") == "true",
+        "rt-export-rejection: incomplete accepted-readback failure contract",
+    )
 
 
 def _require_bounded_cross_batch_pipeline(mode: str, text: str) -> None:
@@ -738,6 +1041,66 @@ def _require_bounded_cross_batch_pipeline(mode: str, text: str) -> None:
 
 def _require_phase15f_present_boundary(mode: str, text: str) -> None:
     if mode == "present-boundary":
+        source_contract = _testcase_marker(
+            "PresentSourceContractRejection", text
+        )
+        _require(
+            source_contract is not None and source_contract[0] == "PASS",
+            f"{mode}: missing PresentSourceContractRejection PASS marker",
+        )
+        _, source_fields = source_contract
+        rejected = source_fields.get("rejected")
+        copy_commit = source_fields.get("copy_commit")
+        bindless_cross_queue = source_fields.get("bindless_cross_queue")
+        expected_rejected = (
+            15
+            + int(copy_commit == "verified")
+            + int(bindless_cross_queue == "verified")
+        )
+        _require(
+            rejected is not None
+            and rejected.isdigit()
+            and int(rejected) == expected_rejected
+            and all(
+                source_fields.get(field) == "true"
+                for field in (
+                    "null",
+                    "usage",
+                    "transfer_src",
+                    "samples",
+                    "format",
+                    "compressed",
+                    "mip",
+                    "layer",
+                    "offset",
+                    "extent",
+                    "fresh",
+                    "accepted_export",
+                    "stale_clear",
+                    "backend_tracked_same_batch",
+                    "backend_rejected",
+                    "marker_then_clear",
+                    "accepted_mutation_clear",
+                    "rejected_mutation_preserves",
+                    "accepted_copy_mutation",
+                    "marker_then_segmented_state_change",
+                    "accepted_prefix_rejected_suffix",
+                    "wrong_queue_clear",
+                    "rejected_export_clear",
+                    "bindless_source_order",
+                    "bindless_refcount",
+                    "bindless_rejected_update",
+                    "bindless_segments",
+                    "bindless_parallel_record",
+                )
+            )
+            and copy_commit in {"verified", "skipped"}
+            and bindless_cross_queue in {"verified", "skipped"}
+            and source_fields.get("valid_override") == "4"
+            and source_fields.get("owner") == "Submission",
+            f"{mode}: incomplete Present source rejection contract",
+        )
+
         serial_control = _testcase_marker(
             "SerialControlPipelineBoundary", text
         )
@@ -778,6 +1141,23 @@ def _require_phase15f_present_boundary(mode: str, text: str) -> None:
             and shutdown_fields.get("replay") == "0",
             f"{mode}: incomplete queued Present shutdown contract",
         )
+        completion = _testcase_marker(
+            "PresentationCompletionIntegrationBoundary", text
+        )
+        _require(
+            completion is not None and completion[0] == "PASS",
+            f"{mode}: missing PresentationCompletionIntegrationBoundary PASS marker",
+        )
+        _, completion_fields = completion
+        _require(
+            completion_fields.get("present_fence") == "nonblocking_targeted"
+            and completion_fields.get("fence_owner") == "Completion"
+            and completion_fields.get("completion_threads") == "1"
+            and completion_fields.get("queue_idle_fallback") == "targeted"
+            and completion_fields.get("queue_idle_owner") == "Submission"
+            and completion_fields.get("outstanding") == "0",
+            f"{mode}: incomplete Presentation completion integration contract",
+        )
 
     marker_name = (
         "PresentPipelineBoundary"
@@ -1000,6 +1380,252 @@ def _require_timestamp_query_success_batch(text: str) -> None:
     )
 
 
+def _nonnegative_number(value: str | None) -> bool:
+    if value is None:
+        return False
+    try:
+        number = float(value)
+    except ValueError:
+        return False
+    return math.isfinite(number) and number >= 0.0
+
+
+def _number_pair(value: str | None) -> tuple[float, float] | None:
+    if value is None:
+        return None
+    parts = value.split(",")
+    if len(parts) != 2:
+        return None
+    try:
+        pair = float(parts[0]), float(parts[1])
+    except ValueError:
+        return None
+    if not all(math.isfinite(number) for number in pair):
+        return None
+    return pair
+
+
+def _require_present_legacy_owner(text: str) -> None:
+    marker = _testcase_marker("LegacyDirectPresentOwnerBoundary", text)
+    _require(
+        marker is not None and marker[0] == "PASS",
+        "present-legacy-owner: missing legacy owner PASS marker",
+    )
+    _, fields = marker
+    _require(
+        fields.get("rhi_thread") == "false"
+        and fields.get("owner") == "Submission"
+        and fields.get("thread") == "caller"
+        and fields.get("receipt") == "exactly_once"
+        and fields.get("native_present") == "0",
+        "present-legacy-owner: incomplete direct Present ownership contract",
+    )
+
+
+def _require_present_completion_shutdown(text: str) -> None:
+    marker = _testcase_marker(
+        "PresentationCompletionShutdownDrainBoundary", text
+    )
+    _require(
+        marker is not None and marker[0] == "PASS",
+        "present-completion-shutdown: missing shutdown-drain PASS marker",
+    )
+    _, fields = marker
+    _require(
+        fields.get("pending") == "fence+fallback"
+        and fields.get("fence_owner") == "Completion"
+        and fields.get("queue_idle_owner") == "Submission"
+        and fields.get("outstanding") == "0"
+        and fields.get("dispose") == "returned",
+        "present-completion-shutdown: incomplete final owner-drain contract",
+    )
+
+
+def _require_timestamp_query(text: str) -> None:
+    completion = _testcase_marker("TimestampQueryCompletionOwnership", text)
+    _require(
+        completion is not None and completion[0] == "PASS",
+        "timestamp-query: missing Completion ownership PASS marker",
+    )
+    _, completion_fields = completion
+    valid_bits = completion_fields.get("valid_bits")
+    _require(
+        completion_fields.get("status") == "Ready"
+        and completion_fields.get("gpu_completion") == "Ready"
+        and completion_fields.get("owner") == "Completion"
+        and completion_fields.get("order")
+        == "signal->completion->query->ordinary"
+        and completion_fields.get("allocator_slot_reuse") == "verified"
+        and completion_fields.get("large_query_pairs") == "501"
+        and completion_fields.get("post_growth_submit") == "accepted"
+        and valid_bits is not None
+        and valid_bits.isdigit()
+        and 0 < int(valid_bits) <= 64
+        and _nonnegative_number(completion_fields.get("duration_ns"))
+        and _nonnegative_number(completion_fields.get("reused_duration_ns"))
+        and completion_fields.get("readback") == "verified"
+        and completion_fields.get("replay") == "0",
+        "timestamp-query: incomplete Completion ownership contract",
+    )
+
+    _require_timestamp_query_success_batch(text)
+
+    preparation = _testcase_marker("TimestampQueryPreparationRejection", text)
+    _require(
+        preparation is not None and preparation[0] == "PASS",
+        "timestamp-query: missing preparation rejection PASS marker",
+    )
+    _, preparation_fields = preparation
+    async_scope = preparation_fields.get("async_scope")
+    _require(
+        async_scope is not None
+        and async_scope.isdigit()
+        and int(async_scope) > 0
+        and preparation_fields.get("prepare_calls") in {"1", "2"}
+        and preparation_fields.get("status") == "Error"
+        and preparation_fields.get("suffix_query") == "Error"
+        and preparation_fields.get("publish_before_completion") == "true"
+        and preparation_fields.get("signal") == "rejected-not-failed"
+        and preparation_fields.get("native_rejected_batch") == "0"
+        and preparation_fields.get("recovery_submit") == "accepted"
+        and preparation_fields.get("owner") == "Completion"
+        and preparation_fields.get("callbacks") == "exactly_once"
+        and preparation_fields.get("replay") == "0",
+        "timestamp-query: incomplete preparation rejection contract",
+    )
+
+    preflight = _testcase_marker("TimestampQueryPreflightRejection", text)
+    _require(
+        preflight is not None and preflight[0] == "PASS",
+        "timestamp-query: missing preflight rejection PASS marker",
+    )
+    _, preflight_fields = preflight
+    _require(
+        preflight_fields.get("reason") == "multi-segment-query"
+        and preflight_fields.get("sources") == "2"
+        and preflight_fields.get("status") == "Error"
+        and preflight_fields.get("owner") == "Completion"
+        and preflight_fields.get("batch_terminal_before_notify") == "true"
+        and preflight_fields.get("bounded_cross_future_wait") == "true"
+        and preflight_fields.get("ordinary_callback") == "exactly_once"
+        and preflight_fields.get("query_callback") == "exactly_once"
+        and preflight_fields.get("success_callback") == "0"
+        and preflight_fields.get("native_submit") == "0"
+        and preflight_fields.get("replay") == "0",
+        "timestamp-query: incomplete preflight rejection contract",
+    )
+
+
+def _require_timestamp_query_mid_failure(text: str) -> None:
+    marker = _testcase_marker("TimestampQueryMidBatchTranslateFailure", text)
+    _require(
+        marker is not None and marker[0] == "PASS",
+        "timestamp-query-mid-failure: missing mid-batch failure PASS marker",
+    )
+    _, fields = marker
+    _require(
+        fields.get("queues") == "Graphics,Graphics"
+        and fields.get("native_prefix_submit") == "1"
+        and fields.get("suffix_translate") == "main-thread-released"
+        and fields.get("suffix_status") == "Error"
+        and fields.get("prefix_completion") == "Ready"
+        and fields.get("suffix_completion") == "Error"
+        and fields.get("pre_terminal_callback_entry") == "0"
+        and fields.get("batch_terminal_before_notify") == "true"
+        and fields.get("bounded_cross_future_wait") == "true"
+        and fields.get("owner") == "Completion"
+        and fields.get("callbacks") == "exactly_once"
+        and fields.get("replay") == "0",
+        "timestamp-query-mid-failure: incomplete terminal-frontier contract",
+    )
+
+
+def _require_timestamp_query_record_failure(text: str) -> None:
+    marker = _testcase_marker("TimestampQuerySerialRecordFailure", text)
+    _require(
+        marker is not None and marker[0] == "PASS",
+        "timestamp-query-record-failure: missing record failure PASS marker",
+    )
+    _, fields = marker
+    recorded_count = fields.get("recorded_phase_gate_count")
+    failure_count = fields.get("failed_phase_gate_count")
+    _require(
+        fields.get("sources") == "2"
+        and fields.get("failing_source") == "0"
+        and fields.get("serial_query_island") == "true"
+        and fields.get("sibling_query") == "Error"
+        and fields.get("sibling_signal") == "failed"
+        and fields.get("batch_terminal_before_notify") == "true"
+        and recorded_count is not None
+        and recorded_count.isdigit()
+        and failure_count is not None
+        and failure_count.isdigit()
+        and int(recorded_count) + int(failure_count) == 1
+        and fields.get("pre_release_callback") == "0"
+        and fields.get("owner") == "Completion"
+        and fields.get("callbacks") == "exactly_once"
+        and fields.get("native_submit") == "0"
+        and fields.get("replay") == "0",
+        "timestamp-query-record-failure: incomplete record-failure contract",
+    )
+
+
+def _require_gpu_scope_stream(text: str) -> None:
+    marker = _testcase_marker(
+        "GpuScopeStreamCompletionAndParallelIsolation", text
+    )
+    _require(marker is not None, "gpu-scope-stream: missing terminal marker")
+    status, fields = marker
+    if status == "SKIP":
+        _require(
+            fields.get("reason") == "graphics_queue_unavailable",
+            "gpu-scope-stream: invalid SKIP contract",
+        )
+        return
+    _require(status == "PASS", "gpu-scope-stream: missing GPU scope stream PASS marker")
+    frame_id = fields.get("frame_id")
+    durations = _number_pair(fields.get("duration_ns"))
+    exclusives = _number_pair(fields.get("exclusive_ns"))
+    timing_contract = False
+    if durations is not None and exclusives is not None:
+        outer_duration, inner_duration = durations
+        outer_exclusive, inner_exclusive = exclusives
+        approximately_equal = lambda actual, expected: math.isclose(
+            actual,
+            expected,
+            rel_tol=1.0e-9,
+            abs_tol=1.0e-9,
+        )
+        timing_contract = (
+            outer_duration > 0.0
+            and inner_duration > 0.0
+            and outer_exclusive >= 0.0
+            and inner_exclusive >= 0.0
+            and approximately_equal(inner_exclusive, inner_duration)
+            and approximately_equal(
+                outer_exclusive,
+                max(0.0, outer_duration - inner_duration),
+            )
+        )
+    _require(
+        frame_id is not None
+        and frame_id.isdigit()
+        and int(frame_id) > 0
+        and fields.get("queue") == "Graphics"
+        and fields.get("scopes") == "2"
+        and fields.get("hierarchy") == "nested"
+        and fields.get("query_source") == "query-serial-island"
+        and fields.get("query_free_sibling") == "parallel-effective"
+        and fields.get("raw_ticks") == "verified"
+        and timing_contract
+        and fields.get("owner") == "Completion"
+        and fields.get("readback") == "verified"
+        and fields.get("sibling_readbacks") == "8/8"
+        and fields.get("replay") == "0",
+        "gpu-scope-stream: incomplete query-island/parallel-sibling contract",
+    )
+
+
 def _require_owning_readback_future(text: str) -> None:
     marker = _testcase_marker("OwningReadbackFuture", text)
     _require(
@@ -1055,15 +1681,33 @@ def validate_log(mode: str, text: str) -> None:
     _require("[TESTCASE][FAIL]" not in text, f"{mode}: test emitted a FAIL marker")
     _require("VUID-" not in text, f"{mode}: Vulkan validation VUID was emitted")
     _require("Validation Error" not in text, f"{mode}: Vulkan validation error was emitted")
+    if mode == "present-legacy-owner":
+        _require_present_legacy_owner(text)
+        return
+    if mode == "present-completion-shutdown":
+        _require_present_completion_shutdown(text)
+        return
     if mode == "readback-future":
         _require_owning_readback_future(text)
         return
     if mode == "occlusion-query":
         _require_occlusion_query(text)
         return
+    if mode == "timestamp-query":
+        _require_timestamp_query(text)
+        return
     if mode == "timestamp-query-success-batch":
         _require_timestamp_query_success_batch(text)
         return
+    if mode == "timestamp-query-mid-failure":
+        _require_timestamp_query_mid_failure(text)
+        return
+    if mode == "timestamp-query-record-failure":
+        _require_timestamp_query_record_failure(text)
+        return
+    if mode == "gpu-scope-stream":
+        _require_gpu_scope_stream(text)
+        return
     if mode == "rt-export-rejection":
         _require_rt_export_rejection(text)
         return
@@ -1333,54 +1977,93 @@ def validate_log(mode: str, text: str) -> None:
         )
 
 
-def run_case(executable: Path, outdir: Path, mode: str, timeout: float) -> Path:
-    arguments: list[str] = []
-    if mode in ("parallel", "fallback", "gated", "heavy"):
-        arguments.append("--parallel")
-    if mode == "fallback":
-        arguments.append("--inject-worker-failure")
-    if mode == "gated":
-        arguments.append("--production-gate")
-    if mode == "heavy":
-        arguments.append("--production-heavy")
-    if mode == "translate-hard":
-        arguments.append("--inject-translate-failure")
-    if mode == "multi-segment-hard":
-        arguments.append("--inject-multi-segment-translate-failure")
-    if mode == "pipeline-window1":
-        arguments.append("--pipeline-window1")
-    if mode == "pipeline-window2":
-        arguments.append("--pipeline-window2")
-    if mode == "present-boundary":
-        arguments.append("--present-boundary")
-    if mode == "present-hard":
-        arguments.append("--present-hard")
-    if mode == "rt-export-rejection":
-        arguments.append("--rt-export-rejection")
-    if mode == "readback-future":
-        arguments.append("--readback-future")
-    if mode == "occlusion-query":
-        arguments.append("--occlusion-query")
-    if mode == "timestamp-query-success-batch":
-        arguments.append("--timestamp-query-success-batch")
+def _output_text(value: str | bytes | None) -> str:
+    if value is None:
+        return ""
+    if isinstance(value, bytes):
+        return value.decode("utf-8", errors="replace")
+    return value
 
-    completed = subprocess.run(
-        [str(executable), *arguments],
-        cwd=executable.parent,
-        capture_output=True,
-        text=True,
-        encoding="utf-8",
-        errors="replace",
-        timeout=timeout,
-        check=False,
-    )
-    text = completed.stdout + completed.stderr
+
+def _write_case_logs(
+    outdir: Path,
+    mode: str,
+    stdout: str,
+    stderr: str,
+) -> Path:
     outdir.mkdir(parents=True, exist_ok=True)
+    (outdir / f"{mode}.stdout.log").write_text(stdout, encoding="utf-8")
+    (outdir / f"{mode}.stderr.log").write_text(stderr, encoding="utf-8")
     log_path = outdir / f"{mode}.log"
-    log_path.write_text(text, encoding="utf-8")
+    log_path.write_text(stdout + stderr, encoding="utf-8")
+    return log_path
+
+
+def run_case(
+    executable: Path,
+    outdir: Path,
+    mode: str,
+    timeout: float,
+    requirements: GpuGateRequirements | None = None,
+) -> CaseResult:
+    test_case = VULKAN_TEST_CASE_BY_MODE.get(mode)
+    _require(test_case is not None, f"unknown Vulkan test mode: {mode}")
+    assert test_case is not None
+    arguments = list(test_case.arguments)
+
+    try:
+        completed = subprocess.run(
+            [str(executable), *arguments],
+            cwd=executable.parent,
+            capture_output=True,
+            text=True,
+            encoding="utf-8",
+            errors="replace",
+            timeout=timeout,
+            check=False,
+        )
+    except subprocess.TimeoutExpired as error:
+        stdout = _output_text(error.stdout)
+        stderr = _output_text(error.stderr)
+        log_path = _write_case_logs(outdir, mode, stdout, stderr)
+        raise VulkanTestError(
+            f"{mode}: timed out after {timeout:g}s; partial log: {log_path}"
+        ) from error
+
+    stdout = _output_text(completed.stdout)
+    stderr = _output_text(completed.stderr)
+    text = stdout + stderr
+    log_path = _write_case_logs(outdir, mode, stdout, stderr)
     _require(completed.returncode == 0, f"{mode}: executable returned {completed.returncode}")
+    requirements = requirements or GpuGateRequirements()
+    environment = _gpu_environment(text, required=requirements.active)
+    _validate_gpu_environment(environment, requirements)
     validate_log(mode, text)
-    return log_path
+    if requirements.reject_testcase_skips:
+        _require(
+            "[TESTCASE][SKIP]" not in text,
+            f"{mode}: strict GPU gate does not permit TESTCASE skips",
+        )
+    return CaseResult(log_path=log_path, environment=environment)
+
+
+def _uint32_argument(value: str) -> int:
+    try:
+        number = int(value, 0)
+    except ValueError as error:
+        raise argparse.ArgumentTypeError(
+            f"expected a decimal or 0x-prefixed integer, got {value!r}"
+        ) from error
+    if not 0 <= number <= 0xFFFFFFFF:
+        raise argparse.ArgumentTypeError("value must fit in uint32")
+    return number
+
+
+def _api_version_argument(value: str) -> tuple[int, int, int]:
+    try:
+        return _parse_api_version(value)
+    except ValueError as error:
+        raise argparse.ArgumentTypeError(str(error)) from error
 
 
 def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
@@ -1390,9 +2073,53 @@ def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
     parser.add_argument("--executable", required=True, type=Path)
     parser.add_argument("--outdir", required=True, type=Path)
     parser.add_argument("--timeout", type=float, default=180.0)
+    parser.add_argument(
+        "--strict-gpu-gate",
+        action="store_true",
+        help=(
+            "require the GPU_ENV marker, active validation, Vulkan 1.3+, "
+            "and no TESTCASE skips"
+        ),
+    )
+    parser.add_argument(
+        "--require-gpu-env",
+        action="store_true",
+        help="require one exact GPU_ENV schema marker from every process",
+    )
+    parser.add_argument("--require-vendor-id", type=_uint32_argument)
+    parser.add_argument("--require-device-id", type=_uint32_argument)
+    parser.add_argument(
+        "--require-device-type",
+        choices=(
+            "other",
+            "integrated_gpu",
+            "discrete_gpu",
+            "virtual_gpu",
+            "cpu",
+            "unknown",
+        ),
+    )
+    parser.add_argument(
+        "--require-validation",
+        action="store_true",
+        help="require validation to be requested, available, and enabled",
+    )
+    parser.add_argument(
+        "--minimum-device-api",
+        type=_api_version_argument,
+        metavar="MAJOR.MINOR[.PATCH]",
+    )
     args = parser.parse_args(argv)
-    if args.timeout <= 0.0:
-        parser.error("--timeout must be positive")
+    if not math.isfinite(args.timeout) or args.timeout <= 0.0:
+        parser.error("--timeout must be finite and positive")
+    strict_minimum_api = (1, 3, 0)
+    if args.strict_gpu_gate:
+        if args.minimum_device_api is None:
+            args.minimum_device_api = strict_minimum_api
+        elif args.minimum_device_api < strict_minimum_api:
+            parser.error(
+                "--strict-gpu-gate requires --minimum-device-api >= 1.3.0"
+            )
     return args
 
 
@@ -1402,31 +2129,36 @@ def main(argv: Sequence[str] | None = None) -> int:
     if not executable.is_file():
         print(f"parallel Vulkan test: executable not found: {executable}", file=sys.stderr)
         return 2
+    requirements = GpuGateRequirements(
+        require_marker=args.require_gpu_env or args.strict_gpu_gate,
+        vendor_id=args.require_vendor_id,
+        device_id=args.require_device_id,
+        device_type=args.require_device_type,
+        require_validation=args.require_validation or args.strict_gpu_gate,
+        minimum_device_api=args.minimum_device_api,
+        reject_testcase_skips=args.strict_gpu_gate,
+    )
     try:
-        logs = [
-            run_case(executable, args.outdir, mode, args.timeout)
-            for mode in (
-                "serial",
-                "parallel",
-                "fallback",
-                "gated",
-                "heavy",
-                "translate-hard",
-                "multi-segment-hard",
-                "pipeline-window1",
-                "pipeline-window2",
-                "present-boundary",
-                "present-hard",
-                "rt-export-rejection",
-                "readback-future",
-                "occlusion-query",
-                "timestamp-query-success-batch",
+        _prepare_outdir(args.outdir)
+        results = [
+            run_case(
+                executable,
+                args.outdir,
+                test_case.mode,
+                args.timeout,
+                requirements,
             )
+            for test_case in VULKAN_TEST_CASES
         ]
+        _validate_consistent_gpu_environments(results)
+        _write_summary(args.outdir, results)
     except (OSError, subprocess.TimeoutExpired, VulkanTestError) as error:
         print(f"parallel Vulkan test: FAIL: {error}", file=sys.stderr)
         return 1
-    print("parallel Vulkan test: PASS: " + ", ".join(str(path) for path in logs))
+    print(
+        "parallel Vulkan test: PASS: "
+        + ", ".join(str(result.log_path) for result in results)
+    )
     return 0
 
 
diff --git a/tools/threading/test_run_parallel_record_vulkan_test.py b/tools/threading/test_run_parallel_record_vulkan_test.py
index 70efa98e2..20a256ce4 100644
--- a/tools/threading/test_run_parallel_record_vulkan_test.py
+++ b/tools/threading/test_run_parallel_record_vulkan_test.py
@@ -1,5 +1,9 @@
 from __future__ import annotations
 
+import contextlib
+import io
+import json
+import os
 import subprocess
 import tempfile
 import unittest
@@ -147,16 +151,46 @@ def queued_present_shutdown_line() -> str:
     )
 
 
+def present_source_contract_line() -> str:
+    return (
+        "[TESTCASE][PASS] name=PresentSourceContractRejection "
+        "rejected=17 null=true usage=true transfer_src=true samples=true "
+        "format=true compressed=true mip=true layer=true offset=true extent=true "
+        "fresh=true accepted_export=true stale_clear=true "
+        "backend_tracked_same_batch=true backend_rejected=true "
+        "marker_then_clear=true accepted_mutation_clear=true "
+        "rejected_mutation_preserves=true accepted_copy_mutation=true "
+        "marker_then_segmented_state_change=true "
+        "accepted_prefix_rejected_suffix=true copy_commit=verified "
+        "wrong_queue_clear=true rejected_export_clear=true "
+        "bindless_source_order=true bindless_refcount=true "
+        "bindless_rejected_update=true bindless_segments=true "
+        "bindless_cross_queue=verified bindless_parallel_record=true "
+        "valid_override=4 owner=Submission\n"
+    )
+
+
+def presentation_completion_integration_line() -> str:
+    return (
+        "[TESTCASE][PASS] name=PresentationCompletionIntegrationBoundary "
+        "present_fence=nonblocking_targeted fence_owner=Completion "
+        "completion_threads=1 queue_idle_fallback=targeted "
+        "queue_idle_owner=Submission outstanding=0\n"
+    )
+
+
 def present_boundary_line(bridge: str = "required") -> str:
     prefix_native = "2" if bridge == "required" else "0"
     return (
-        serial_control_boundary_line()
+        present_source_contract_line()
+        + serial_control_boundary_line()
         + "[TESTCASE][PASS] name=PresentPipelineBoundary "
         "outcome=Recreate order=Prefix,Bridge?,Present,Later "
         "owner=Submission receipt_attempts=1 submitted=false "
         "recreate=true completion=drained later_batch=success "
         f"present_only=verified bridge={bridge} graphics_native=0 "
         f"prefix_native={prefix_native} readback=verified replay=0\n"
+        + presentation_completion_integration_line()
         + queued_present_shutdown_line()
     )
 
@@ -179,6 +213,8 @@ def rt_export_rejection_line() -> str:
         "readback_retry=frame_accepted recovery_consumed=true encoder=once "
         "decision_table=verified native_rejected=0 callbacks=exactly_once "
         "keepalive=terminal replay=0\n"
+        "[TESTCASE][PASS] name=RaytracingAcceptedReadbackMaterializationFailure "
+        "native=accepted payload=error encoder=0 latch=true\n"
     )
 
 
@@ -255,6 +291,79 @@ def timestamp_query_success_batch_line() -> str:
     )
 
 
+def present_legacy_owner_line() -> str:
+    return (
+        "[TESTCASE][PASS] name=LegacyDirectPresentOwnerBoundary "
+        "rhi_thread=false owner=Submission thread=caller "
+        "receipt=exactly_once native_present=0\n"
+    )
+
+
+def present_completion_shutdown_line() -> str:
+    return (
+        "[TESTCASE][PASS] name=PresentationCompletionShutdownDrainBoundary "
+        "pending=fence+fallback fence_owner=Completion "
+        "queue_idle_owner=Submission outstanding=0 dispose=returned\n"
+    )
+
+
+def timestamp_query_line() -> str:
+    return (
+        "[TESTCASE][PASS] name=TimestampQueryCompletionOwnership "
+        "status=Ready gpu_completion=Ready owner=Completion "
+        "order=signal->completion->query->ordinary "
+        "allocator_slot_reuse=verified large_query_pairs=501 "
+        "post_growth_submit=accepted valid_bits=64 duration_ns=12.5 "
+        "reused_duration_ns=3.25 readback=verified replay=0\n"
+        + timestamp_query_success_batch_line()
+        + "[TESTCASE][PASS] name=TimestampQueryPreparationRejection "
+        "async_scope=5784928276370600517 prepare_calls=2 status=Error "
+        "suffix_query=Error publish_before_completion=true "
+        "signal=rejected-not-failed native_rejected_batch=0 "
+        "recovery_submit=accepted owner=Completion callbacks=exactly_once "
+        "replay=0\n"
+        "[TESTCASE][PASS] name=TimestampQueryPreflightRejection "
+        "reason=multi-segment-query sources=2 status=Error owner=Completion "
+        "batch_terminal_before_notify=true bounded_cross_future_wait=true "
+        "ordinary_callback=exactly_once query_callback=exactly_once "
+        "success_callback=0 native_submit=0 replay=0\n"
+    )
+
+
+def timestamp_query_mid_failure_line() -> str:
+    return (
+        "[TESTCASE][PASS] name=TimestampQueryMidBatchTranslateFailure "
+        "queues=Graphics,Graphics native_prefix_submit=1 "
+        "suffix_translate=main-thread-released suffix_status=Error "
+        "prefix_completion=Ready suffix_completion=Error "
+        "pre_terminal_callback_entry=0 batch_terminal_before_notify=true "
+        "bounded_cross_future_wait=true owner=Completion "
+        "callbacks=exactly_once replay=0\n"
+    )
+
+
+def timestamp_query_record_failure_line() -> str:
+    return (
+        "[TESTCASE][PASS] name=TimestampQuerySerialRecordFailure "
+        "sources=2 failing_source=0 serial_query_island=true "
+        "sibling_query=Error sibling_signal=failed "
+        "batch_terminal_before_notify=true recorded_phase_gate_count=1 "
+        "failed_phase_gate_count=0 pre_release_callback=0 "
+        "owner=Completion callbacks=exactly_once native_submit=0 replay=0\n"
+    )
+
+
+def gpu_scope_stream_line() -> str:
+    return (
+        "[TESTCASE][PASS] name=GpuScopeStreamCompletionAndParallelIsolation "
+        "frame_id=7 queue=Graphics scopes=2 hierarchy=nested "
+        "query_source=query-serial-island "
+        "query_free_sibling=parallel-effective raw_ticks=verified "
+        "duration_ns=12.5,3.25 exclusive_ns=9.25,3.25 "
+        "owner=Completion readback=verified sibling_readbacks=8/8 replay=0\n"
+    )
+
+
 def pass_line(
     mode: str,
     fault: str,
@@ -346,7 +455,308 @@ def summary(batch: int, outcome: str = "parallel") -> str:
     )
 
 
+def gpu_env_line(
+    *,
+    validation_requested: str = "true",
+    validation_layer_available: str = "true",
+    validation_enabled: str = "true",
+    device_type: str = "discrete_gpu",
+    vendor_id: str = "0x000010de",
+    device_id: str = "0x00002c02",
+    device_uuid: str = "3b172c4f63b522546e783cce9b27bd48",
+    requested_api: str = "1.3.0",
+    device_api: str = "1.3.280",
+    device_api_raw: str = "0x00403118",
+) -> str:
+    return (
+        "[Vulkan][GPU_ENV] schema=1 backend=vulkan "
+        f"validation_requested={validation_requested} "
+        f"validation_layer_available={validation_layer_available} "
+        f"validation_enabled={validation_enabled} "
+        f"device_type={device_type} vendor_id={vendor_id} device_id={device_id} "
+        f"device_uuid={device_uuid} requested_api={requested_api} "
+        f"device_api={device_api} api_variant=0 device_api_raw={device_api_raw} "
+        "driver_id=4 driver_version_raw=0x12345678\n"
+    )
+
+
 class VulkanRunnerTests(unittest.TestCase):
+    def test_gpu_environment_schema_is_accepted(self) -> None:
+        environment = runner._gpu_environment(gpu_env_line(), required=True)
+        self.assertIsNotNone(environment)
+        assert environment is not None
+        self.assertEqual(environment.vendor_id, 0x10DE)
+        self.assertEqual(environment.device_id, 0x2C02)
+        self.assertEqual(
+            environment.device_uuid, "3b172c4f63b522546e783cce9b27bd48"
+        )
+        self.assertEqual(environment.device_type, "discrete_gpu")
+        self.assertEqual(environment.device_api, (1, 3, 280))
+
+    def test_gpu_environment_is_required_in_strict_mode(self) -> None:
+        with self.assertRaisesRegex(
+            runner.VulkanTestError, "expected exactly one marker"
+        ):
+            runner._gpu_environment("", required=True)
+
+    def test_gpu_environment_rejects_duplicate_markers(self) -> None:
+        with self.assertRaisesRegex(
+            runner.VulkanTestError, "expected exactly one marker"
+        ):
+            runner._gpu_environment(gpu_env_line() * 2, required=True)
+
+    def test_gpu_environment_rejects_unknown_fields(self) -> None:
+        with self.assertRaisesRegex(
+            runner.VulkanTestError, "malformed or unsupported schema"
+        ):
+            runner._gpu_environment(
+                gpu_env_line().replace(
+                    " driver_id=4", " unexpected=true driver_id=4"
+                ),
+                required=True,
+            )
+
+    def test_gpu_environment_rejects_inconsistent_raw_api(self) -> None:
+        with self.assertRaisesRegex(
+            runner.VulkanTestError, "do not match device_api_raw"
+        ):
+            runner._gpu_environment(
+                gpu_env_line(device_api_raw="0x00403119"),
+                required=True,
+            )
+
+    def test_gpu_gate_requires_full_validation_proof(self) -> None:
+        environment = runner._gpu_environment(
+            gpu_env_line(validation_enabled="false"), required=True
+        )
+        with self.assertRaisesRegex(
+            runner.VulkanTestError, "validation requested/available/enabled"
+        ):
+            runner._validate_gpu_environment(
+                environment,
+                runner.GpuGateRequirements(require_validation=True),
+            )
+
+    def test_gpu_gate_rejects_wrong_vendor(self) -> None:
+        environment = runner._gpu_environment(gpu_env_line(), required=True)
+        with self.assertRaisesRegex(runner.VulkanTestError, "vendor mismatch"):
+            runner._validate_gpu_environment(
+                environment,
+                runner.GpuGateRequirements(vendor_id=0x1002),
+            )
+
+    def test_gpu_gate_rejects_device_api_below_minimum(self) -> None:
+        environment = runner._gpu_environment(
+            gpu_env_line(device_api="1.2.0", device_api_raw="0x00402000"),
+            required=True,
+        )
+        with self.assertRaisesRegex(runner.VulkanTestError, "below required"):
+            runner._validate_gpu_environment(
+                environment,
+                runner.GpuGateRequirements(minimum_device_api=(1, 3, 0)),
+            )
+
+    def test_gpu_gate_rejects_requested_api_below_minimum(self) -> None:
+        environment = runner._gpu_environment(
+            gpu_env_line(requested_api="1.2.0"), required=True
+        )
+        with self.assertRaisesRegex(
+            runner.VulkanTestError, "requested API is below required"
+        ):
+            runner._validate_gpu_environment(
+                environment,
+                runner.GpuGateRequirements(minimum_device_api=(1, 3, 0)),
+            )
+
+    def test_gpu_gate_rejects_environment_changes_across_modes(self) -> None:
+        first = runner._gpu_environment(gpu_env_line(), required=True)
+        second = runner._gpu_environment(
+            gpu_env_line(device_id="0x00002704"), required=True
+        )
+        with self.assertRaisesRegex(runner.VulkanTestError, "changed across"):
+            runner._validate_consistent_gpu_environments(
+                (
+                    runner.CaseResult(Path("first.log"), first),
+                    runner.CaseResult(Path("second.log"), second),
+                )
+            )
+
+    def test_gpu_gate_distinguishes_same_model_devices_by_uuid(self) -> None:
+        first = runner._gpu_environment(gpu_env_line(), required=True)
+        second = runner._gpu_environment(
+            gpu_env_line(device_uuid="00112233445566778899aabbccddeeff"),
+            required=True,
+        )
+        with self.assertRaisesRegex(runner.VulkanTestError, "changed across"):
+            runner._validate_consistent_gpu_environments(
+                (
+                    runner.CaseResult(Path("first.log"), first),
+                    runner.CaseResult(Path("second.log"), second),
+                )
+            )
+
+    def test_parse_args_accepts_strict_gpu_policy(self) -> None:
+        args = runner.parse_args(
+            (
+                "--executable",
+                "test.exe",
+                "--outdir",
+                "logs",
+                "--strict-gpu-gate",
+                "--require-vendor-id",
+                "0x10de",
+                "--require-device-type",
+                "discrete_gpu",
+                "--minimum-device-api",
+                "1.3",
+            )
+        )
+        self.assertTrue(args.strict_gpu_gate)
+        self.assertEqual(args.require_vendor_id, 0x10DE)
+        self.assertEqual(args.minimum_device_api, (1, 3, 0))
+
+    def test_strict_gpu_policy_defaults_to_vulkan_1_3(self) -> None:
+        args = runner.parse_args(
+            (
+                "--executable",
+                "test.exe",
+                "--outdir",
+                "logs",
+                "--strict-gpu-gate",
+            )
+        )
+        self.assertEqual(args.minimum_device_api, (1, 3, 0))
+
+    def test_strict_gpu_policy_rejects_lower_explicit_api(self) -> None:
+        with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(
+            SystemExit
+        ):
+            runner.parse_args(
+                (
+                    "--executable",
+                    "test.exe",
+                    "--outdir",
+                    "logs",
+                    "--strict-gpu-gate",
+                    "--minimum-device-api",
+                    "1.2",
+                )
+            )
+
+    def test_timeout_must_be_finite_and_positive(self) -> None:
+        for value in ("nan", "inf", "-inf", "0", "-1"):
+            with self.subTest(value=value), contextlib.redirect_stderr(
+                io.StringIO()
+            ), self.assertRaises(SystemExit):
+                runner.parse_args(
+                    (
+                        "--executable",
+                        "test.exe",
+                        "--outdir",
+                        "logs",
+                        "--timeout",
+                        value,
+                    )
+                )
+
+    def test_timeout_persists_partial_stdout_and_stderr(self) -> None:
+        timeout = subprocess.TimeoutExpired(
+            cmd=["test.exe"],
+            timeout=1.0,
+            output=b"partial stdout\n",
+            stderr=b"partial stderr\n",
+        )
+        with tempfile.TemporaryDirectory() as temporary_directory:
+            outdir = Path(temporary_directory)
+            with mock.patch.object(runner.subprocess, "run", side_effect=timeout):
+                with self.assertRaisesRegex(runner.VulkanTestError, "timed out"):
+                    runner.run_case(Path("test.exe"), outdir, "serial", 1.0)
+            self.assertEqual(
+                (outdir / "serial.stdout.log").read_text(encoding="utf-8"),
+                "partial stdout\n",
+            )
+            self.assertEqual(
+                (outdir / "serial.stderr.log").read_text(encoding="utf-8"),
+                "partial stderr\n",
+            )
+
+    def test_strict_gpu_gate_rejects_any_testcase_skip(self) -> None:
+        completed = subprocess.CompletedProcess(
+            args=[],
+            returncode=0,
+            stdout=(
+                gpu_env_line()
+                + pipeline_line(1, "false", "blocked")
+                + "[TESTCASE][SKIP] name=Unexpected reason=unsupported\n"
+            ),
+            stderr="",
+        )
+        requirements = runner.GpuGateRequirements(
+            require_marker=True,
+            require_validation=True,
+            reject_testcase_skips=True,
+        )
+        with tempfile.TemporaryDirectory() as temporary_directory:
+            with mock.patch.object(
+                runner.subprocess, "run", return_value=completed
+            ):
+                with self.assertRaisesRegex(
+                    runner.VulkanTestError, "does not permit TESTCASE skips"
+                ):
+                    runner.run_case(
+                        Path("test.exe"),
+                        Path(temporary_directory),
+                        "pipeline-window1",
+                        30.0,
+                        requirements,
+                    )
+
+    def test_summary_records_stable_gpu_identity(self) -> None:
+        environment = runner._gpu_environment(gpu_env_line(), required=True)
+        with tempfile.TemporaryDirectory() as temporary_directory:
+            outdir = Path(temporary_directory)
+            summary_path = runner._write_summary(
+                outdir,
+                (runner.CaseResult(outdir / "serial.log", environment),),
+            )
+            summary_text = summary_path.read_text(encoding="utf-8")
+            self.assertIn('"mode_count": 1', summary_text)
+            self.assertIn('"vendor_id": "0x000010de"', summary_text)
+            self.assertIn(
+                '"device_uuid": "3b172c4f63b522546e783cce9b27bd48"',
+                summary_text,
+            )
+            self.assertIn('"validation_enabled": true', summary_text)
+
+    def test_summary_records_explicit_tested_sha_over_event_sha(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary_directory, mock.patch.dict(
+            os.environ,
+            {
+                "MOER_GATE_SHA": "tested-merge-sha",
+                "GITHUB_SHA": "controller-base-sha",
+                "MOER_GATE_PR_NUMBER": "232",
+                "MOER_GATE_RUN_URL": "https://example.invalid/run/1",
+            },
+            clear=False,
+        ):
+            outdir = Path(temporary_directory)
+            summary_path = runner._write_summary(outdir, ())
+            payload = json.loads(summary_path.read_text(encoding="utf-8"))
+            self.assertEqual(payload["commit"], "tested-merge-sha")
+            self.assertEqual(payload["pull_request"], "232")
+            self.assertEqual(
+                payload["workflow_run_url"], "https://example.invalid/run/1"
+            )
+
+    def test_runner_refuses_nonempty_output_directory(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary_directory:
+            outdir = Path(temporary_directory)
+            (outdir / "stale.log").write_text("old", encoding="utf-8")
+            with self.assertRaisesRegex(
+                runner.VulkanTestError, "refusing to reuse non-empty"
+            ):
+                runner._prepare_outdir(outdir)
+
     def test_pipeline_window1_contract_is_accepted(self) -> None:
         runner.validate_log(
             "pipeline-window1",
@@ -1035,6 +1445,16 @@ def test_pipeline_modes_map_to_focused_executable_arguments(self) -> None:
                 "--present-hard",
                 present_hard_line(),
             ),
+            (
+                "present-legacy-owner",
+                "--present-legacy-owner",
+                present_legacy_owner_line(),
+            ),
+            (
+                "present-completion-shutdown",
+                "--present-completion-shutdown",
+                present_completion_shutdown_line(),
+            ),
             (
                 "rt-export-rejection",
                 "--rt-export-rejection",
@@ -1055,6 +1475,26 @@ def test_pipeline_modes_map_to_focused_executable_arguments(self) -> None:
                 "--timestamp-query-success-batch",
                 timestamp_query_success_batch_line(),
             ),
+            (
+                "timestamp-query",
+                "--timestamp-query",
+                timestamp_query_line(),
+            ),
+            (
+                "timestamp-query-mid-failure",
+                "--timestamp-query-mid-failure",
+                timestamp_query_mid_failure_line(),
+            ),
+            (
+                "timestamp-query-record-failure",
+                "--timestamp-query-record-failure",
+                timestamp_query_record_failure_line(),
+            ),
+            (
+                "gpu-scope-stream",
+                "--gpu-scope-stream",
+                gpu_scope_stream_line(),
+            ),
         )
         with tempfile.TemporaryDirectory() as temporary_directory:
             for mode, expected_argument, output in cases:
@@ -1084,6 +1524,180 @@ def test_pipeline_modes_map_to_focused_executable_arguments(self) -> None:
                         ],
                     )
 
+    def test_declared_gpu_matrix_contains_all_21_unique_modes(self) -> None:
+        modes = [test_case.mode for test_case in runner.VULKAN_TEST_CASES]
+        self.assertEqual(
+            set(modes),
+            {
+                "serial",
+                "parallel",
+                "fallback",
+                "gated",
+                "heavy",
+                "translate-hard",
+                "multi-segment-hard",
+                "pipeline-window1",
+                "pipeline-window2",
+                "present-boundary",
+                "present-hard",
+                "present-legacy-owner",
+                "present-completion-shutdown",
+                "rt-export-rejection",
+                "readback-future",
+                "occlusion-query",
+                "timestamp-query",
+                "timestamp-query-success-batch",
+                "timestamp-query-mid-failure",
+                "timestamp-query-record-failure",
+                "gpu-scope-stream",
+            },
+        )
+        self.assertEqual(len(modes), 21)
+        self.assertEqual(len(set(modes)), 21)
+        self.assertEqual(set(modes), set(runner.VULKAN_TEST_CASE_BY_MODE))
+
+    def test_new_focused_modes_reject_weakened_terminal_contracts(self) -> None:
+        cases = (
+            (
+                "present-legacy-owner",
+                present_legacy_owner_line().replace(
+                    "owner=Submission", "owner=Completion"
+                ),
+            ),
+            (
+                "present-completion-shutdown",
+                present_completion_shutdown_line().replace(
+                    "outstanding=0", "outstanding=1"
+                ),
+            ),
+            (
+                "timestamp-query",
+                timestamp_query_line().replace(
+                    "allocator_slot_reuse=verified",
+                    "allocator_slot_reuse=missing",
+                ),
+            ),
+            (
+                "timestamp-query-mid-failure",
+                timestamp_query_mid_failure_line().replace(
+                    "pre_terminal_callback_entry=0",
+                    "pre_terminal_callback_entry=1",
+                ),
+            ),
+            (
+                "timestamp-query-record-failure",
+                timestamp_query_record_failure_line().replace(
+                    "failed_phase_gate_count=0",
+                    "failed_phase_gate_count=1",
+                ),
+            ),
+            (
+                "gpu-scope-stream",
+                gpu_scope_stream_line().replace(
+                    "query_free_sibling=parallel-effective",
+                    "query_free_sibling=serial",
+                ),
+            ),
+        )
+        for mode, text in cases:
+            with self.subTest(mode=mode), self.assertRaises(
+                runner.VulkanTestError
+            ):
+                runner.validate_log(mode, text)
+
+    def test_gpu_scope_stream_accepts_only_documented_nonstrict_skip(self) -> None:
+        documented = (
+            "[TESTCASE][SKIP] "
+            "name=GpuScopeStreamCompletionAndParallelIsolation "
+            "reason=graphics_queue_unavailable\n"
+        )
+        runner.validate_log("gpu-scope-stream", documented)
+        with self.assertRaisesRegex(runner.VulkanTestError, "invalid SKIP"):
+            runner.validate_log(
+                "gpu-scope-stream",
+                documented.replace(
+                    "graphics_queue_unavailable", "test_disabled"
+                ),
+            )
+
+    def test_gpu_scope_stream_rejects_invalid_timing_evidence(self) -> None:
+        for duration, exclusive in (
+            ("0,3.25", "0,3.25"),
+            ("nan,3.25", "0,3.25"),
+            ("12.5,3.25", "8.0,3.25"),
+        ):
+            with self.subTest(
+                duration=duration, exclusive=exclusive
+            ), self.assertRaisesRegex(
+                runner.VulkanTestError,
+                "query-island/parallel-sibling contract",
+            ):
+                runner.validate_log(
+                    "gpu-scope-stream",
+                    gpu_scope_stream_line()
+                    .replace("duration_ns=12.5,3.25", f"duration_ns={duration}")
+                    .replace("exclusive_ns=9.25,3.25", f"exclusive_ns={exclusive}"),
+                )
+
+    def test_full_timestamp_mode_requires_all_three_unique_boundaries(self) -> None:
+        for marker_name in (
+            "TimestampQueryCompletionOwnership",
+            "TimestampQueryPreparationRejection",
+            "TimestampQueryPreflightRejection",
+        ):
+            with self.subTest(marker=marker_name), self.assertRaises(
+                runner.VulkanTestError
+            ):
+                runner.validate_log(
+                    "timestamp-query",
+                    replace_testcase_marker(
+                        timestamp_query_line(), marker_name, ""
+                    ),
+                )
+
+    def test_timestamp_dynamic_success_variants_are_accepted(self) -> None:
+        runner.validate_log(
+            "timestamp-query",
+            timestamp_query_line().replace("prepare_calls=2", "prepare_calls=1"),
+        )
+        runner.validate_log(
+            "timestamp-query-record-failure",
+            timestamp_query_record_failure_line()
+            .replace("recorded_phase_gate_count=1", "recorded_phase_gate_count=0")
+            .replace("failed_phase_gate_count=0", "failed_phase_gate_count=1"),
+        )
+
+    def test_strict_gate_rejects_documented_gpu_scope_skip(self) -> None:
+        completed = subprocess.CompletedProcess(
+            args=[],
+            returncode=0,
+            stdout=(
+                gpu_env_line()
+                + "[TESTCASE][SKIP] "
+                "name=GpuScopeStreamCompletionAndParallelIsolation "
+                "reason=graphics_queue_unavailable\n"
+            ),
+            stderr="",
+        )
+        requirements = runner.GpuGateRequirements(
+            require_marker=True,
+            require_validation=True,
+            reject_testcase_skips=True,
+        )
+        with tempfile.TemporaryDirectory() as temporary_directory, mock.patch.object(
+            runner.subprocess, "run", return_value=completed
+        ):
+            with self.assertRaisesRegex(
+                runner.VulkanTestError, "does not permit TESTCASE skips"
+            ):
+                runner.run_case(
+                    Path("TestRHIParallelRecordVulkan.exe"),
+                    Path(temporary_directory),
+                    "gpu-scope-stream",
+                    30.0,
+                    requirements,
+                )
+
     def test_readback_future_contract_is_accepted(self) -> None:
         runner.validate_log(
             "readback-future",
@@ -1355,6 +1969,19 @@ def test_rt_export_rejection_requires_readback_recovery(self) -> None:
                 ),
             )
 
+    def test_rt_export_rejection_requires_materialization_failure(self) -> None:
+        with self.assertRaisesRegex(
+            runner.VulkanTestError, "accepted-readback failure"
+        ):
+            runner.validate_log(
+                "rt-export-rejection",
+                replace_testcase_marker(
+                    rt_export_rejection_line(),
+                    "RaytracingAcceptedReadbackMaterializationFailure",
+                    "",
+                ),
+            )
+
     def test_present_boundary_contract_is_accepted(self) -> None:
         runner.validate_log(
             "present-boundary", present_boundary_line()
@@ -1375,6 +2002,30 @@ def test_present_boundary_requires_serial_control_gate(self) -> None:
                 ),
             )
 
+    def test_present_boundary_requires_every_invoked_subtest_marker(self) -> None:
+        for marker_name in (
+            "PresentSourceContractRejection",
+            "PresentationCompletionIntegrationBoundary",
+        ):
+            with self.subTest(marker=marker_name), self.assertRaises(
+                runner.VulkanTestError
+            ):
+                runner.validate_log(
+                    "present-boundary",
+                    replace_testcase_marker(
+                        present_boundary_line(), marker_name, ""
+                    ),
+                )
+
+    def test_present_source_contract_accepts_qualified_queue_skips(self) -> None:
+        runner.validate_log(
+            "present-boundary",
+            present_boundary_line()
+            .replace("rejected=17", "rejected=15")
+            .replace("copy_commit=verified", "copy_commit=skipped")
+            .replace("bindless_cross_queue=verified", "bindless_cross_queue=skipped"),
+        )
+
     def test_serial_control_boundary_requires_blocked_later_translate(
         self,
     ) -> None: