From 0d3c3b483ee7f29adf1e09e473148d5d363ff4af Mon Sep 17 00:00:00 2001
From: Your Name <you@example.com>
Date: Sat, 18 Jul 2026 18:54:04 +0800
Subject: [PATCH] fix(cd): coalesce stale carriers and harden verifiers

---
 .gitea/workflows/cd.yaml                      | 249 ++++++++++++++++--
 .../test_cd_post_deploy_source_link_gate.py   |  67 ++++-
 .../test_executor_network_policy_boundary.py  |   6 +-
 ...t_workload_database_identity_projection.py |   5 +-
 ops/runner/read-public-gitea-actions-queue.py |  44 +++-
 .../select-latest-inclusive-cd-carrier.py     | 191 ++++++++++++++
 .../test_read_public_gitea_actions_queue.py   |  37 +++
 ...test_select_latest_inclusive_cd_carrier.py | 226 ++++++++++++++++
 8 files changed, 797 insertions(+), 28 deletions(-)
 create mode 100644 ops/runner/select-latest-inclusive-cd-carrier.py
 create mode 100644 ops/runner/test_select_latest_inclusive_cd_carrier.py

diff --git a/.gitea/workflows/cd.yaml b/.gitea/workflows/cd.yaml
index a44318155..ac6b210ae 100644
--- a/.gitea/workflows/cd.yaml
+++ b/.gitea/workflows/cd.yaml
@@ -24,11 +24,9 @@ on:
   workflow_dispatch:
     # 手動觸發永遠可用（用於補跑、緊急部署）
 
-# Main deployments run serially so concurrent writers cannot starve production closure.
-# Job timeouts, rollout verification, and deploy markers terminate or expose failed runs.
-concurrency:
-  group: cd-deploy-${{ github.ref }}
-  cancel-in-progress: false
+# Gitea 1.25 ignores the workflow concurrency key. The dedicated runner's verified
+# capacity=1 is the executable queue, while the selectors below revalidate linear
+# main ancestry before tests, immediately before deploy, and before post-deploy readback.
 
 env:
   HARBOR: registry.wooo.work
@@ -60,6 +58,40 @@ env:
   ALERT_CHAIN_API_URL: https://awoooi.wooo.work
 
 jobs:
+  select-latest-carrier:
+    if: ${{ github.event_name != 'push' || (!contains(github.event.head_commit.message, '[skip ci]') && !contains(github.event.head_commit.message, 'cancel-stale-cd') && !contains(github.event.head_commit.message, '[metadata-only]')) }}
+    runs-on: awoooi-non110-host
+    timeout-minutes: 3
+    outputs:
+      selected: ${{ steps.carrier.outputs.selected }}
+      status: ${{ steps.carrier.outputs.status }}
+      source_sha: ${{ steps.carrier.outputs.source_sha }}
+      latest_sha: ${{ steps.carrier.outputs.latest_sha }}
+      main_tip_sha: ${{ steps.carrier.outputs.main_tip_sha }}
+    steps:
+      - name: Checkout exact source from Gitea
+        env:
+          GITEA_SOURCE_URL: http://192.168.0.110:3001/wooo/awoooi.git
+        run: |
+          set -euo pipefail
+          git init .
+          git remote remove origin 2>/dev/null || true
+          git remote add origin "$GITEA_SOURCE_URL"
+          git fetch --no-tags --prune --depth=1 origin "$GITHUB_SHA"
+          git checkout --force --detach FETCH_HEAD
+
+      - name: Select latest inclusive main carrier
+        id: carrier
+        env:
+          GITEA_SOURCE_URL: http://192.168.0.110:3001/wooo/awoooi.git
+        run: |
+          set -euo pipefail
+          python3 ops/runner/select-latest-inclusive-cd-carrier.py \
+            --source-sha "$GITHUB_SHA" \
+            --remote-url "$GITEA_SOURCE_URL" \
+            --debounce-seconds 20 \
+            --github-output "$GITHUB_OUTPUT"
+
   workflow-shape:
     # 2026-06-28 Codex: Gitea 1.25 may mark a workflow invalid when every
     # root job has a job-level `if`. Keep one no-op root job without `if` so
@@ -87,7 +119,8 @@ jobs:
     # on CD-generated deploy commits. Skip jobs explicitly so marker commits
     # do not trigger a self-feeding CD loop; `cancel-stale-cd` is a
     # controlled no-op trigger used only to cancel stale pre-guard runs.
-    if: ${{ github.event_name != 'push' || (!contains(github.event.head_commit.message, '[skip ci]') && !contains(github.event.head_commit.message, 'cancel-stale-cd')) }}
+    if: ${{ needs.select-latest-carrier.outputs.selected == 'true' && (github.event_name != 'push' || (!contains(github.event.head_commit.message, '[skip ci]') && !contains(github.event.head_commit.message, 'cancel-stale-cd'))) }}
+    needs: [select-latest-carrier]
     # 2026-04-30 Codex: run the tests job on the host runner and launch the
     # CI image explicitly. The act-managed job container can disappear mid-test
     # with Docker RWLayer=nil on the shared 110 daemon.
@@ -1370,18 +1403,53 @@ jobs:
             echo "AWOOI API notify failed; direct Telegram fallback disabled to preserve AwoooP receipt chain"
           fi
 
+  revalidate-deploy-carrier:
+    if: ${{ needs.select-latest-carrier.outputs.selected == 'true' }}
+    needs: [select-latest-carrier, tests]
+    runs-on: awoooi-non110-host
+    timeout-minutes: 3
+    outputs:
+      selected: ${{ steps.carrier.outputs.selected }}
+      status: ${{ steps.carrier.outputs.status }}
+      latest_sha: ${{ steps.carrier.outputs.latest_sha }}
+      main_tip_sha: ${{ steps.carrier.outputs.main_tip_sha }}
+    steps:
+      - name: Checkout exact source from Gitea
+        env:
+          GITEA_SOURCE_URL: http://192.168.0.110:3001/wooo/awoooi.git
+        run: |
+          set -euo pipefail
+          git init .
+          git remote remove origin 2>/dev/null || true
+          git remote add origin "$GITEA_SOURCE_URL"
+          git fetch --no-tags --prune --depth=1 origin "$GITHUB_SHA"
+          git checkout --force --detach FETCH_HEAD
+
+      - name: Revalidate latest inclusive deploy carrier
+        id: carrier
+        env:
+          GITEA_SOURCE_URL: http://192.168.0.110:3001/wooo/awoooi.git
+        run: |
+          set -euo pipefail
+          python3 ops/runner/select-latest-inclusive-cd-carrier.py \
+            --source-sha "$GITHUB_SHA" \
+            --remote-url "$GITEA_SOURCE_URL" \
+            --github-output "$GITHUB_OUTPUT"
+
   build-and-deploy:
     # 2026-06-28 Codex: keep CD-generated `[skip ci]` deploy commits and
     # `cancel-stale-cd` queue-cleaning commits from re-entering build/deploy.
     # 2026-07-01 Codex: metadata-only controlled-runtime fixes already run the
     # focused tests above; do not spend the Docker build lock or deploy marker
     # when no app image can change.
-    if: ${{ github.event_name != 'push' || (!contains(github.event.head_commit.message, '[skip ci]') && !contains(github.event.head_commit.message, 'cancel-stale-cd') && !contains(github.event.head_commit.message, '[metadata-only]')) }}
+    if: ${{ needs.revalidate-deploy-carrier.outputs.selected == 'true' && (github.event_name != 'push' || (!contains(github.event.head_commit.message, '[skip ci]') && !contains(github.event.head_commit.message, 'cancel-stale-cd') && !contains(github.event.head_commit.message, '[metadata-only]'))) }}
     # 2026-04-30 Codex: Docker builds run on the host runner. Long docker build
     # steps were killing the transient act job container with RWLayer=nil.
-    needs: [tests]
+    needs: [select-latest-carrier, tests, revalidate-deploy-carrier]
     timeout-minutes: 60
     runs-on: awoooi-non110-host
+    outputs:
+      deploy_applied: ${{ steps.deploy.outputs.applied }}
     steps:
       - name: Bootstrap Host Runner Tools
         # 2026-05-05 Codex: keep the host-mode runner self-healing before
@@ -1977,11 +2045,30 @@ jobs:
             echo "⚡ no Docker build lock to release"
           fi
 
+      - name: Revalidate carrier before production mutations
+        id: preproduction-carrier
+        run: |
+          set -euo pipefail
+          echo "selected=false" >> "$GITHUB_OUTPUT"
+          CARRIER_JSON=$(python3 ops/runner/select-latest-inclusive-cd-carrier.py \
+            --source-sha "$GITHUB_SHA" \
+            --remote-url http://192.168.0.110:3001/wooo/awoooi.git)
+          CARRIER_SELECTED=$(printf '%s' "$CARRIER_JSON" | \
+            python3 -c 'import json,sys; print(str(json.load(sys.stdin)["selected"]).lower())')
+          if [ "$CARRIER_SELECTED" != "true" ]; then
+            CARRIER_LATEST=$(printf '%s' "$CARRIER_JSON" | \
+              python3 -c 'import json,sys; print(json.load(sys.stdin)["latest_sha"])')
+            echo "AWOOOI_BUILD_COALESCED=source=${GITHUB_SHA};latest=${CARRIER_LATEST};production_write_performed=false"
+            exit 0
+          fi
+          echo "selected=true" >> "$GITHUB_OUTPUT"
+
       # Production drift proved the original CREATE TABLE IF NOT EXISTS
       # migration did not update an older CHECK constraint. Reconcile this
       # one additive allowlist before rolling out the scanner and verify it
       # independently without exposing connection details.
       - name: Reconcile Asset Inventory Type Constraint
+        if: ${{ steps.preproduction-carrier.outputs.selected == 'true' }}
         env:
           DATABASE_URL: ${{ secrets.DATABASE_URL }}
           MIGRATION_DATABASE_URL: ${{ secrets.MIGRATION_DATABASE_URL }}
@@ -2122,6 +2209,7 @@ jobs:
       # the index transactionally, and prove the conflict-update path before
       # rolling out the scanner.
       - name: Repair Asset Inventory Canonical Integrity
+        if: ${{ steps.preproduction-carrier.outputs.selected == 'true' }}
         env:
           DATABASE_URL: ${{ secrets.DATABASE_URL }}
           MIGRATION_DATABASE_URL: ${{ secrets.MIGRATION_DATABASE_URL }}
@@ -2319,6 +2407,7 @@ jobs:
       # additive RLS schema, then verify both schema policy and runtime-role
       # read access without printing connection details or row content.
       - name: Apply MCP Version Lifecycle Receipt Schema
+        if: ${{ steps.preproduction-carrier.outputs.selected == 'true' }}
         env:
           DATABASE_URL: ${{ secrets.DATABASE_URL }}
           MIGRATION_DATABASE_URL: ${{ secrets.MIGRATION_DATABASE_URL }}
@@ -2463,6 +2552,7 @@ jobs:
       # independently verified. No source payload or connection detail is
       # printed by this controlled migration receipt.
       - name: Apply MCP Federation Receipt Schema
+        if: ${{ steps.preproduction-carrier.outputs.selected == 'true' }}
         env:
           DATABASE_URL: ${{ secrets.DATABASE_URL }}
           MIGRATION_DATABASE_URL: ${{ secrets.MIGRATION_DATABASE_URL }}
@@ -2612,6 +2702,7 @@ jobs:
       # 2026-03-31 ogt: P0-1 Secrets 自動注入 (ADR-035 強制)
       # 2026-03-31 ogt: 加入 AI API Keys (修復 mock_fallback 問題)
       - name: Inject K8s Secrets
+        if: ${{ steps.preproduction-carrier.outputs.selected == 'true' }}
         env:
           ARGOCD_API_TOKEN: ${{ secrets.ARGOCD_API_TOKEN }}
           AWOOOI_GITEA_API_TOKEN: ${{ secrets.AWOOOI_GITEA_API_TOKEN }}
@@ -3012,9 +3103,12 @@ jobs:
       #   4. 等待 ArgoCD sync + rollout 完成
       #   5. Health Check
       - name: Deploy to K8s (ArgoCD GitOps)
+        if: ${{ steps.preproduction-carrier.outputs.selected == 'true' }}
+        id: deploy
         env:
           CD_PUSH_TOKEN: ${{ secrets.CD_PUSH_TOKEN }}
         run: |
+          echo "applied=false" >> "$GITHUB_OUTPUT"
           prepare_deploy_key() {
             mkdir -p "${HOME}/.ssh"
             umask 077
@@ -3118,6 +3212,45 @@ jobs:
             fi
           }
 
+          # The two directly applied ConfigMaps must be part of the same rollback
+          # boundary as NetworkPolicy. Otherwise a superseded carrier could leave
+          # production configuration mutated without a deploy marker.
+          CONFIGMAP_ROLLBACK_FILE="/tmp/awoooi-deploy-configmap-rollback.json"
+          ssh $SSH_OPTS "wooo@${{ env.K8S_SSH_HOST }}" \
+            "KUBECTL='sudo kubectl --kubeconfig=/etc/rancher/k3s/k3s.yaml --server=${{ env.K8S_API_SERVER }}'; \$KUBECTL get configmap awoooi-config service-registry -n awoooi-prod -o json" \
+            | python3 -c '
+          import json
+          import sys
+
+          source = json.load(sys.stdin)
+          items = []
+          for item in source["items"]:
+              clean = {
+                  "apiVersion": item["apiVersion"],
+                  "kind": item["kind"],
+                  "metadata": {
+                      "name": item["metadata"]["name"],
+                      "namespace": item["metadata"]["namespace"],
+                      "labels": item["metadata"].get("labels", {}),
+                      "annotations": item["metadata"].get("annotations", {}),
+                  },
+                  "data": item.get("data", {}),
+              }
+              if "binaryData" in item:
+                  clean["binaryData"] = item["binaryData"]
+              if "immutable" in item:
+                  clean["immutable"] = item["immutable"]
+              items.append(clean)
+          json.dump({"apiVersion": "v1", "kind": "List", "items": items}, sys.stdout)
+          ' > "$CONFIGMAP_ROLLBACK_FILE"
+          test -s "$CONFIGMAP_ROLLBACK_FILE"
+          restore_deploy_configmaps() {
+            echo "⏪ restoring previous AWOOOI deploy ConfigMaps"
+            cat "$CONFIGMAP_ROLLBACK_FILE" | \
+              ssh $SSH_OPTS "wooo@${{ env.K8S_SSH_HOST }}" \
+              "KUBECTL='sudo kubectl --kubeconfig=/etc/rancher/k3s/k3s.yaml --server=${{ env.K8S_API_SERVER }}'; \$KUBECTL apply -f -"
+          }
+
           # NetworkPolicy is intentionally outside kustomize because commonLabels
           # mutates nested selectors. The checkout is depth=1, so snapshot live
           # policy truth and remove server-only metadata before any mutation.
@@ -3182,12 +3315,15 @@ jobs:
             set +e
             restore_workload_database_projection
             workload_restore_exit=$?
+            restore_deploy_configmaps
+            configmap_restore_exit=$?
             restore_network_policy
             network_restore_exit=$?
             set -e
             if [ "$workload_restore_exit" -ne 0 ] \
+              || [ "$configmap_restore_exit" -ne 0 ] \
               || [ "$network_restore_exit" -ne 0 ]; then
-              echo "❌ deploy rollback incomplete: workload=${workload_restore_exit} network=${network_restore_exit}" >&2
+              echo "❌ deploy rollback incomplete: workload=${workload_restore_exit} configmap=${configmap_restore_exit} network=${network_restore_exit}" >&2
               return 1
             fi
           }
@@ -3207,6 +3343,22 @@ jobs:
             fi
             exit "$exit_code"
           }
+
+          # Last read-only gate before the first production mutation. Builds may
+          # finish out of order on Gitea 1.25; only the latest inclusive source
+          # is allowed to cross this boundary.
+          PREWRITE_CARRIER_JSON=$(python3 ops/runner/select-latest-inclusive-cd-carrier.py \
+            --source-sha "$IMAGE_TAG" \
+            --remote-url http://192.168.0.110:3001/wooo/awoooi.git)
+          PREWRITE_CARRIER_SELECTED=$(printf '%s' "$PREWRITE_CARRIER_JSON" | \
+            python3 -c 'import json,sys; print(str(json.load(sys.stdin)["selected"]).lower())')
+          if [ "$PREWRITE_CARRIER_SELECTED" != "true" ]; then
+            PREWRITE_CARRIER_LATEST=$(printf '%s' "$PREWRITE_CARRIER_JSON" | \
+              python3 -c 'import json,sys; print(json.load(sys.stdin)["latest_sha"])')
+            echo "AWOOOI_DEPLOY_COALESCED_AFTER_PREREQUISITES=source=${IMAGE_TAG};latest=${PREWRITE_CARRIER_LATEST};runtime_deploy_write_performed=false;monotonic_prerequisites_completed=true"
+            exit 0
+          fi
+
           trap rollback_deploy_controls_on_exit EXIT
           cat k8s/awoooi-prod/02-network-policy.yaml | \
             ssh $SSH_OPTS "wooo@${{ env.K8S_SSH_HOST }}" \
@@ -3303,13 +3455,26 @@ jobs:
             k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml
           DEPLOY_REVISION=""
           git diff --cached --quiet && echo "⚡ kustomization.yaml 無變化，跳過 push" || {
-            git commit -m "chore(cd): deploy ${IMAGE_TAG::7} [skip ci]"
             # 用 token 推送（避免 SSH key 需要額外設定 push 權限）
             git remote remove gitea 2>/dev/null || true
             git remote add gitea "http://wooo:${CD_PUSH_TOKEN}@192.168.0.110:3001/wooo/awoooi.git"
-            # 先 rebase 避免 non-fast-forward (其他 commit 在 CI 期間已推入)
-            # 2026-04-17 ogt: -X theirs — kustomization.yaml 衝突時採用當次部署的 image tag
             git fetch gitea main
+            CARRIER_JSON=$(python3 ops/runner/select-latest-inclusive-cd-carrier.py \
+              --source-sha "$IMAGE_TAG" \
+              --remote-url http://192.168.0.110:3001/wooo/awoooi.git)
+            CARRIER_SELECTED=$(printf '%s' "$CARRIER_JSON" | \
+              python3 -c 'import json,sys; print(str(json.load(sys.stdin)["selected"]).lower())')
+            if [ "$CARRIER_SELECTED" != "true" ]; then
+              CARRIER_LATEST=$(printf '%s' "$CARRIER_JSON" | \
+                python3 -c 'import json,sys; print(json.load(sys.stdin)["latest_sha"])')
+              echo "AWOOOI_DEPLOY_SUPERSEDED_AFTER_WRITE=source=${IMAGE_TAG};latest=${CARRIER_LATEST};rollback=required"
+              restore_deploy_controls
+              echo "AWOOOI_DEPLOY_SUPERSEDED_ROLLBACK=source=${IMAGE_TAG};latest=${CARRIER_LATEST};rollback_verified=true"
+              exit 1
+            fi
+            git commit -m "chore(cd): deploy ${IMAGE_TAG::7} [skip ci]"
+            # Rebase only onto the exact main revision validated above. If main
+            # advances after this fetch, the normal push rejects non-fast-forward.
             git rebase -X theirs gitea/main
             DEPLOY_REVISION=$(git rev-parse HEAD)
             git push gitea main
@@ -3723,16 +3888,24 @@ jobs:
             container="$2"
             last_failure="workload_not_ready"
             for attempt in 1 2 3; do
-              ready_pod_ref=$(
+              candidate_pod_refs=$(
                 $KUBECTL get pods -n awoooi-prod -l "app=$workload" \
                   --field-selector=status.phase=Running \
                   --sort-by=.metadata.creationTimestamp \
-                  -o name 2>/dev/null | tail -n 1
-              ) || ready_pod_ref=""
-              if [ -z "$ready_pod_ref" ]; then
+                  -o name 2>/dev/null
+              ) || candidate_pod_refs=""
+              ready_pod_ref=""
+              while IFS= read -r candidate_pod_ref; do
+                [ -n "$candidate_pod_ref" ] || continue
+                if $KUBECTL wait --for=condition=Ready "$candidate_pod_ref" \
+                  -n awoooi-prod --timeout=1s >/dev/null 2>&1; then
+                  ready_pod_ref="$candidate_pod_ref"
+                  break
+                fi
+              done < <(printf '%s\n' "$candidate_pod_refs" | sed '/^$/d' | tac)
+              if [ -z "$candidate_pod_refs" ]; then
                 last_failure="running_pod_unavailable"
-              elif ! $KUBECTL wait --for=condition=Ready "$ready_pod_ref" \
-                -n awoooi-prod --timeout=10s >/dev/null 2>&1; then
+              elif [ -z "$ready_pod_ref" ]; then
                 last_failure="ready_pod_unavailable"
               else
                 container_ready=$(
@@ -4892,9 +5065,11 @@ jobs:
               echo "⚠️ CI/CD rollout risk notification failed (non-fatal)"
             fi
           fi
+          echo "applied=true" >> "$GITHUB_OUTPUT"
           exit 0
 
       - name: Apply Runtime Image Mirror Policy
+        if: ${{ steps.deploy.outputs.applied == 'true' }}
         run: |
           set -euo pipefail
           prepare_deploy_key() {
@@ -4944,6 +5119,7 @@ jobs:
           echo "runtime_image_policy_stage=verified"
 
       - name: Notify Build Deploy Success
+        if: ${{ steps.deploy.outputs.applied == 'true' }}
         run: |
           END_TIME=$(date +%s)
           DURATION=$((END_TIME - ${{ steps.commit.outputs.start_time }}))
@@ -4998,13 +5174,46 @@ jobs:
             echo "AWOOI API notify failed; direct Telegram fallback disabled to preserve AwoooP receipt chain"
           fi
 
+  revalidate-post-deploy-carrier:
+    if: ${{ needs.build-and-deploy.outputs.deploy_applied == 'true' }}
+    needs: [build-and-deploy]
+    runs-on: awoooi-non110-host
+    timeout-minutes: 3
+    outputs:
+      selected: ${{ steps.carrier.outputs.selected }}
+      status: ${{ steps.carrier.outputs.status }}
+      latest_sha: ${{ steps.carrier.outputs.latest_sha }}
+      main_tip_sha: ${{ steps.carrier.outputs.main_tip_sha }}
+    steps:
+      - name: Checkout exact source from Gitea
+        env:
+          GITEA_SOURCE_URL: http://192.168.0.110:3001/wooo/awoooi.git
+        run: |
+          set -euo pipefail
+          git init .
+          git remote remove origin 2>/dev/null || true
+          git remote add origin "$GITEA_SOURCE_URL"
+          git fetch --no-tags --prune --depth=1 origin "$GITHUB_SHA"
+          git checkout --force --detach FETCH_HEAD
+
+      - name: Revalidate latest inclusive post-deploy carrier
+        id: carrier
+        env:
+          GITEA_SOURCE_URL: http://192.168.0.110:3001/wooo/awoooi.git
+        run: |
+          set -euo pipefail
+          python3 ops/runner/select-latest-inclusive-cd-carrier.py \
+            --source-sha "$GITHUB_SHA" \
+            --remote-url "$GITEA_SOURCE_URL" \
+            --github-output "$GITHUB_OUTPUT"
+
   post-deploy-checks:
     # 2026-06-28 Codex: post-deploy checks belong to real deploy runs; skip
     # marker/no-op commits already accounted for by the previous deploy run.
     # 2026-07-01 Codex: `[metadata-only]` commits do not roll a new image, so
     # post-deploy smokes would only retest the previous production artifact.
-    if: ${{ github.event_name != 'push' || (!contains(github.event.head_commit.message, '[skip ci]') && !contains(github.event.head_commit.message, 'cancel-stale-cd') && !contains(github.event.head_commit.message, '[metadata-only]')) }}
-    needs: [build-and-deploy]
+    if: ${{ needs.revalidate-post-deploy-carrier.outputs.selected == 'true' && needs.build-and-deploy.outputs.deploy_applied == 'true' && (github.event_name != 'push' || (!contains(github.event.head_commit.message, '[skip ci]') && !contains(github.event.head_commit.message, 'cancel-stale-cd') && !contains(github.event.head_commit.message, '[metadata-only]'))) }}
+    needs: [build-and-deploy, revalidate-post-deploy-carrier]
     # 2026-07-16 Codex: the fail-hard pressure gate below may legitimately
     # consume 20 minutes while a bounded backup and its metric freshness
     # window drain. Preserve enough time for the independent post-deploy
diff --git a/apps/api/tests/test_cd_post_deploy_source_link_gate.py b/apps/api/tests/test_cd_post_deploy_source_link_gate.py
index a04fe3d8e..45e26a8c1 100644
--- a/apps/api/tests/test_cd_post_deploy_source_link_gate.py
+++ b/apps/api/tests/test_cd_post_deploy_source_link_gate.py
@@ -42,10 +42,67 @@ def test_cd_source_link_gate_uses_current_deploy_canary() -> None:
     assert "codex-sentry-20260513-t15b-v3" not in text
 
 
-def test_cd_main_pushes_queue_without_canceling_active_deploys() -> None:
+def test_cd_main_pushes_use_gitea_125_executable_inclusive_carrier() -> None:
     text = _workflow_text()
-    concurrency = text.split("concurrency:", 1)[1].split("env:", 1)[0]
 
-    assert "group: cd-deploy-${{ github.ref }}" in concurrency
-    assert "cancel-in-progress: false" in concurrency
-    assert "cancel-in-progress: true" not in concurrency
+    assert "\nconcurrency:" not in text
+    assert "Gitea 1.25 ignores the workflow concurrency key" in text
+    assert "select-latest-carrier:" in text
+    assert "needs: [select-latest-carrier]" in text
+    assert "revalidate-deploy-carrier:" in text
+    assert "needs: [select-latest-carrier, tests]" in text
+    assert "needs: [select-latest-carrier, tests, revalidate-deploy-carrier]" in text
+    assert "revalidate-post-deploy-carrier:" in text
+    assert "needs: [build-and-deploy, revalidate-post-deploy-carrier]" in text
+    assert "needs.revalidate-deploy-carrier.outputs.selected == 'true'" in text
+    assert "needs.revalidate-post-deploy-carrier.outputs.selected == 'true'" in text
+    assert "deploy_applied: ${{ steps.deploy.outputs.applied }}" in text
+    assert "AWOOOI_BUILD_COALESCED=" in text
+    assert "AWOOOI_DEPLOY_COALESCED_AFTER_PREREQUISITES=" in text
+    assert "select-latest-inclusive-cd-carrier.py" in text
+    assert "--debounce-seconds 20" in text
+
+
+def test_deploy_revalidates_lineage_after_final_main_fetch() -> None:
+    text = _workflow_text()
+    build_job = text.split("  build-and-deploy:", 1)[1].split(
+        "  revalidate-post-deploy-carrier:", 1
+    )[0]
+    deploy = text.split("- name: Deploy to K8s (ArgoCD GitOps)", 1)[1].split(
+        "- name: Apply Runtime Image Mirror Policy", 1
+    )[0]
+
+    production_gate = build_job.index(
+        "- name: Revalidate carrier before production mutations"
+    )
+    first_database_mutation = build_job.index(
+        "- name: Reconcile Asset Inventory Type Constraint"
+    )
+    secret_mutation = build_job.index("- name: Inject K8s Secrets")
+    deploy_step = build_job.index("- name: Deploy to K8s (ArgoCD GitOps)")
+    prewrite_revalidate = deploy.index("PREWRITE_CARRIER_JSON=")
+    first_production_write = deploy.index(
+        "if ! cat k8s/awoooi-prod/02-network-policy.yaml"
+    )
+    final_fetch = deploy.index("git fetch gitea main", first_production_write)
+    revalidate = deploy.index("CARRIER_JSON=$(python3", final_fetch)
+    marker_commit = deploy.index('git commit -m "chore(cd): deploy', revalidate)
+    normal_push = deploy.index("git push gitea main", marker_commit)
+    assert production_gate < first_database_mutation < secret_mutation < deploy_step
+    assert build_job.count(
+        "if: ${{ steps.preproduction-carrier.outputs.selected == 'true' }}"
+    ) >= 6
+    assert prewrite_revalidate < first_production_write
+    assert final_fetch < revalidate < marker_commit < normal_push
+    assert 'echo "applied=false" >> "$GITHUB_OUTPUT"' in deploy
+    assert "AWOOOI_DEPLOY_COALESCED_AFTER_PREREQUISITES=" in deploy
+    late_supersession = deploy.split(
+        "AWOOOI_DEPLOY_SUPERSEDED_AFTER_WRITE=", 1
+    )[1].split('git commit -m "chore(cd): deploy', 1)[0]
+    assert "restore_deploy_controls" in late_supersession
+    assert "AWOOOI_DEPLOY_SUPERSEDED_ROLLBACK=" in late_supersession
+    assert "exit 1" in late_supersession
+    assert "restore_deploy_configmaps" in deploy
+    assert "awoooi-config service-registry" in deploy
+    assert 'echo "applied=true" >> "$GITHUB_OUTPUT"' in deploy
+    assert "git push --force" not in deploy
diff --git a/apps/api/tests/test_executor_network_policy_boundary.py b/apps/api/tests/test_executor_network_policy_boundary.py
index 1a643eba3..f7f35714e 100644
--- a/apps/api/tests/test_executor_network_policy_boundary.py
+++ b/apps/api/tests/test_executor_network_policy_boundary.py
@@ -208,7 +208,11 @@ def test_cd_ssh_denial_probe_targets_ready_pod_and_classifies_transport() -> Non
     assert "for attempt in 1 2 3; do" in probe
     assert "--field-selector=status.phase=Running" in probe
     assert "--sort-by=.metadata.creationTimestamp" in probe
-    assert '--for=condition=Ready "$ready_pod_ref"' in probe
+    assert "candidate_pod_refs=" in probe
+    assert "while IFS= read -r candidate_pod_ref; do" in probe
+    assert '--for=condition=Ready "$candidate_pod_ref"' in probe
+    assert 'ready_pod_ref="$candidate_pod_ref"' in probe
+    assert "| tac)" in probe
     assert "status.containerStatuses" in probe
     assert '$KUBECTL exec -i "$ready_pod_ref"' in probe
     assert '$KUBECTL exec -i "deployment/$workload"' not in probe
diff --git a/apps/api/tests/test_workload_database_identity_projection.py b/apps/api/tests/test_workload_database_identity_projection.py
index 066fe1ea3..441f1a6d5 100644
--- a/apps/api/tests/test_workload_database_identity_projection.py
+++ b/apps/api/tests/test_workload_database_identity_projection.py
@@ -297,6 +297,9 @@ def test_workload_database_bootstrap_keeps_secret_values_off_output() -> None:
 def test_cd_uses_live_spec_rollback_and_non_mutating_post_verifier() -> None:
     workflow = (REPO_ROOT / ".gitea/workflows/cd.yaml").read_text()
     normalized_workflow = " ".join(workflow.split())
+    workload_rollback = workflow.split("WORKLOAD_DB_ROLLBACK_FILE", 1)[1].split(
+        "# The two directly applied ConfigMaps", 1
+    )[0]
 
     assert "WORKLOAD_DB_ROLLBACK_FILE" in workflow
     assert "WORKLOAD_DB_SECRET_PREEXISTENCE_FILE" in workflow
@@ -305,7 +308,7 @@ def test_cd_uses_live_spec_rollback_and_non_mutating_post_verifier() -> None:
     assert "bash -s -- verify" in workflow
     assert workflow.index("bash -s -- apply") < workflow.index("bash -s -- verify")
     assert '"spec": item["spec"]' in workflow
-    assert '"data": item' not in workflow
+    assert '"data": item' not in workload_rollback
     assert "get secret '${secret_name}' -n awoooi-prod -o jsonpath" not in workflow
     assert "get secret awoooi-api-db -n awoooi-prod -o jsonpath" not in workflow
     assert "HEAD^" not in workflow
diff --git a/ops/runner/read-public-gitea-actions-queue.py b/ops/runner/read-public-gitea-actions-queue.py
index 869de20cc..7d6ec3b5a 100644
--- a/ops/runner/read-public-gitea-actions-queue.py
+++ b/ops/runner/read-public-gitea-actions-queue.py
@@ -136,6 +136,20 @@ _CD_BUILD_DEPLOY_SUCCESS_NOTIFICATION_RE = re.compile(
     r"Z\s+(?:[^\w\"']+\s+)?CI/CD build-deploy success notification mirrored"
 )
 _JOB_SUCCEEDED_RE = re.compile(r"Z\s+(?:[^\w\"']+\s+)?Job succeeded")
+_CD_EXECUTOR_BOUNDARY_FAILURES = (
+    (
+        "signal worker SSH-denial verifier did not prove the boundary",
+        "signal_worker_ssh_denial_verifier_unproven",
+    ),
+    (
+        "API SSH-denial verifier did not prove the boundary",
+        "api_ssh_denial_verifier_unproven",
+    ),
+    (
+        "execution broker has no live route evidence for allowlisted SSH endpoints",
+        "execution_broker_allowlisted_ssh_route_unproven",
+    ),
+)
 
 
 @dataclass(frozen=True)
@@ -694,6 +708,7 @@ def build_readback(
         build_docker_step_timeout_label=build_log_classifier[
             "docker_step_timeout_label"
         ],
+        build_failure_classifier=build_log_classifier["failure_classifier"],
         harbor_110_repair_no_matching_runner_label=(
             harbor_110_repair_no_matching_runner_label
             if not harbor_110_repair_ignored_for_current_cd_context
@@ -1133,6 +1148,8 @@ def build_readback(
             if build_log_classifier["docker_step_timeout"]
             else "blocked_host_web_build_pressure"
             if effective_tests_log_classifier["host_pressure_blocked_or_waiting"]
+            else "blocked_cd_build_verifier"
+            if build_log_classifier["failure_classifier"]
             else "blocked_harbor_110_repair_no_matching_runner"
             if harbor_110_repair_no_matching_runner_label
             else "blocked_no_matching_online_runner"
@@ -1413,6 +1430,7 @@ def _queue_safe_next_action(
     build_harbor_public_route_retrying_unavailable: bool,
     build_docker_step_timeout: bool,
     build_docker_step_timeout_label: str,
+    build_failure_classifier: str,
     harbor_110_repair_no_matching_runner_label: str,
     harbor_110_repair_waiting: bool,
     harbor_110_repair_running: bool,
@@ -1664,6 +1682,20 @@ def _queue_safe_next_action(
             blocker_fields=["latest_visible_cd_host_pressure_classifier"],
         )
 
+    if build_failure_classifier:
+        return action(
+            action_id="fix_exact_cd_build_verifier_then_single_replay",
+            stage="cd_build_log_exact_failure",
+            text=(
+                "Reproduce and fix the exact build/deploy verifier locally, then "
+                "perform one normal integration and terminal production readback."
+            ),
+            reason=build_failure_classifier,
+            command="focused local verifier replay before the next normal push",
+            post_verifier="single exact CD terminal plus production deploy-marker readback",
+            blocker_fields=["latest_visible_cd_failure_classifier"],
+        )
+
     if cd_jobs_stale_or_mismatched:
         return action(
             action_id="ignore_stale_cd_jobs_api_payload_and_poll_visible_cd_or_marker",
@@ -1733,6 +1765,14 @@ def classify_cd_build_log(text: str) -> dict[str, Any]:
     )
     deploy_marker_matches = list(_CD_DEPLOY_MARKER_RE.finditer(text))
     latest_deploy_marker = deploy_marker_matches[-1] if deploy_marker_matches else None
+    executor_boundary_failure = next(
+        (
+            classifier
+            for marker, classifier in _CD_EXECUTOR_BOUNDARY_FAILURES
+            if marker in text
+        ),
+        "",
+    )
     return {
         "job_succeeded": _JOB_SUCCEEDED_RE.search(text) is not None,
         "deploy_marker_pushed": latest_deploy_marker is not None,
@@ -1749,7 +1789,9 @@ def classify_cd_build_log(text: str) -> dict[str, Any]:
             _CD_BUILD_DEPLOY_SUCCESS_NOTIFICATION_RE.search(text) is not None
         ),
         "failure_classifier": (
-            "harbor_registry_public_route_unavailable"
+            executor_boundary_failure
+            if executor_boundary_failure
+            else "harbor_registry_public_route_unavailable"
             if harbor_public_route_blocked
             else "cd_docker_step_timeout"
             if latest_docker_step_timeout
diff --git a/ops/runner/select-latest-inclusive-cd-carrier.py b/ops/runner/select-latest-inclusive-cd-carrier.py
new file mode 100644
index 000000000..16738b271
--- /dev/null
+++ b/ops/runner/select-latest-inclusive-cd-carrier.py
@@ -0,0 +1,191 @@
+#!/usr/bin/env python3
+"""Select one exact main revision as the inclusive AWOOOI CD carrier."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import subprocess
+import time
+from dataclasses import asdict, dataclass
+from pathlib import Path
+from typing import Sequence
+
+
+SHA_RE = re.compile(r"^[0-9a-f]{40}$")
+EXCLUDED_MESSAGE_MARKERS = ("[skip ci]", "cancel-stale-cd", "[metadata-only]")
+
+
+@dataclass(frozen=True)
+class CarrierDecision:
+    selected: bool
+    status: str
+    source_sha: str
+    latest_sha: str
+    main_tip_sha: str
+
+
+def _git(args: Sequence[str], *, cwd: Path, check: bool = True) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        ["git", *args],
+        cwd=cwd,
+        check=check,
+        capture_output=True,
+        text=True,
+    )
+
+
+def _validate_sha(value: str, *, label: str) -> str:
+    normalized = value.strip().lower()
+    if not SHA_RE.fullmatch(normalized):
+        raise RuntimeError(f"{label}_invalid")
+    return normalized
+
+
+def _latest_remote_sha(*, remote_url: str, branch: str, cwd: Path) -> str:
+    result = _git(
+        ["ls-remote", "--heads", remote_url, f"refs/heads/{branch}"],
+        cwd=cwd,
+    )
+    rows = [row.split() for row in result.stdout.splitlines() if row.strip()]
+    if len(rows) != 1 or len(rows[0]) != 2:
+        raise RuntimeError("latest_main_ref_unresolved")
+    return _validate_sha(rows[0][0], label="latest_sha")
+
+
+def _fetch_latest_history(*, remote_url: str, branch: str, cwd: Path, depth: int) -> None:
+    _git(
+        [
+            "fetch",
+            "--no-tags",
+            "--prune",
+            f"--depth={depth}",
+            remote_url,
+            f"+refs/heads/{branch}:refs/remotes/awoooi-carrier/{branch}",
+        ],
+        cwd=cwd,
+    )
+
+
+def _latest_deploy_eligible_sha(*, branch: str, cwd: Path) -> str | None:
+    result = _git(
+        [
+            "log",
+            "--format=%H%x1f%B%x1e",
+            f"refs/remotes/awoooi-carrier/{branch}",
+        ],
+        cwd=cwd,
+    )
+    for record in result.stdout.split("\x1e"):
+        sha, separator, message = record.strip().partition("\x1f")
+        if not separator:
+            continue
+        normalized_message = message.lower()
+        if any(marker in normalized_message for marker in EXCLUDED_MESSAGE_MARKERS):
+            continue
+        return _validate_sha(sha, label="latest_deploy_eligible_sha")
+    return None
+
+
+def _source_is_ancestor(*, source_sha: str, latest_sha: str, cwd: Path) -> bool:
+    result = _git(
+        ["merge-base", "--is-ancestor", source_sha, latest_sha],
+        cwd=cwd,
+        check=False,
+    )
+    if result.returncode not in {0, 1}:
+        raise RuntimeError("carrier_lineage_check_failed")
+    return result.returncode == 0
+
+
+def select_latest_inclusive_carrier(
+    *,
+    source_sha: str,
+    remote_url: str,
+    branch: str = "main",
+    cwd: Path,
+    debounce_seconds: float = 0,
+) -> CarrierDecision:
+    source_sha = _validate_sha(source_sha, label="source_sha")
+    if debounce_seconds < 0 or debounce_seconds > 60:
+        raise RuntimeError("debounce_seconds_out_of_range")
+    if debounce_seconds:
+        time.sleep(debounce_seconds)
+
+    main_tip_sha = _latest_remote_sha(remote_url=remote_url, branch=branch, cwd=cwd)
+    _fetch_latest_history(remote_url=remote_url, branch=branch, cwd=cwd, depth=256)
+    latest_sha = _latest_deploy_eligible_sha(branch=branch, cwd=cwd)
+    if latest_sha is None:
+        _fetch_latest_history(remote_url=remote_url, branch=branch, cwd=cwd, depth=2048)
+        latest_sha = _latest_deploy_eligible_sha(branch=branch, cwd=cwd)
+    if latest_sha is None:
+        raise RuntimeError("latest_deploy_eligible_main_unresolved")
+
+    if source_sha == latest_sha:
+        return CarrierDecision(
+            selected=True,
+            status="selected_latest_main_carrier",
+            source_sha=source_sha,
+            latest_sha=latest_sha,
+            main_tip_sha=main_tip_sha,
+        )
+
+    if not _source_is_ancestor(source_sha=source_sha, latest_sha=latest_sha, cwd=cwd):
+        shallow = _git(
+            ["rev-parse", "--is-shallow-repository"],
+            cwd=cwd,
+        ).stdout.strip()
+        if shallow == "true":
+            _fetch_latest_history(remote_url=remote_url, branch=branch, cwd=cwd, depth=2048)
+
+    if not _source_is_ancestor(source_sha=source_sha, latest_sha=latest_sha, cwd=cwd):
+        raise RuntimeError("latest_main_does_not_include_source")
+
+    return CarrierDecision(
+        selected=False,
+        status="superseded_by_latest_inclusive_carrier",
+        source_sha=source_sha,
+        latest_sha=latest_sha,
+        main_tip_sha=main_tip_sha,
+    )
+
+
+def _write_github_output(path: Path, decision: CarrierDecision) -> None:
+    fields = {
+        "selected": str(decision.selected).lower(),
+        "status": decision.status,
+        "source_sha": decision.source_sha,
+        "latest_sha": decision.latest_sha,
+        "main_tip_sha": decision.main_tip_sha,
+    }
+    with path.open("a", encoding="utf-8") as stream:
+        for key, value in fields.items():
+            stream.write(f"{key}={value}\n")
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--source-sha", required=True)
+    parser.add_argument("--remote-url", required=True)
+    parser.add_argument("--branch", default="main")
+    parser.add_argument("--repo", type=Path, default=Path.cwd())
+    parser.add_argument("--debounce-seconds", type=float, default=0)
+    parser.add_argument("--github-output", type=Path)
+    args = parser.parse_args()
+
+    decision = select_latest_inclusive_carrier(
+        source_sha=args.source_sha,
+        remote_url=args.remote_url,
+        branch=args.branch,
+        cwd=args.repo.resolve(),
+        debounce_seconds=args.debounce_seconds,
+    )
+    if args.github_output:
+        _write_github_output(args.github_output, decision)
+    print(json.dumps(asdict(decision), sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/ops/runner/test_read_public_gitea_actions_queue.py b/ops/runner/test_read_public_gitea_actions_queue.py
index 2c9542deb..269efcad7 100644
--- a/ops/runner/test_read_public_gitea_actions_queue.py
+++ b/ops/runner/test_read_public_gitea_actions_queue.py
@@ -1904,6 +1904,43 @@ def test_failed_cd_with_stale_jobs_payload_gets_failure_classifier() -> None:
     )
 
 
+def test_exact_executor_boundary_failure_supersedes_stale_jobs_classifier() -> None:
+    module = _load_module()
+    payload = module.build_readback(
+        actions_html=_actions_html_failed_cd_run(),
+        actions_list_http_status=401,
+        actions_list_payload={"message": "token is required"},
+        cd_jobs_http_status=200,
+        cd_jobs_payload={
+            "jobs": [
+                {
+                    "run_id": 4043,
+                    "name": "tests",
+                    "head_sha": "9c4e754d3369fa2de7606e7092712cf6d85e18e2",
+                    "conclusion": "success",
+                }
+            ],
+            "total_count": 1,
+        },
+        latest_cd_build_log_http_status=200,
+        latest_cd_build_log_text=(
+            "ssh_denial_probe_transport_unavailable workload=awoooi-worker "
+            "reason=ready_pod_unavailable\n"
+            "signal worker SSH-denial verifier did not prove the boundary\n"
+        ),
+    )
+
+    assert payload["status"] == "blocked_cd_build_verifier"
+    assert payload["readback"]["cd_run_jobs_stale_or_mismatched"] is True
+    assert payload["readback"]["latest_visible_cd_failure_classifier"] == (
+        "signal_worker_ssh_denial_verifier_unproven"
+    )
+    assert (
+        payload["readback"]["safe_next_action_id"]
+        == "fix_exact_cd_build_verifier_then_single_replay"
+    )
+
+
 def test_interrupted_host_pressure_reports_postgres_recovery_cpu() -> None:
     module = _load_module()
     payload = module.build_readback(
diff --git a/ops/runner/test_select_latest_inclusive_cd_carrier.py b/ops/runner/test_select_latest_inclusive_cd_carrier.py
new file mode 100644
index 000000000..653b9347a
--- /dev/null
+++ b/ops/runner/test_select_latest_inclusive_cd_carrier.py
@@ -0,0 +1,226 @@
+from __future__ import annotations
+
+import importlib.util
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+
+ROOT = Path(__file__).resolve().parents[2]
+SCRIPT = ROOT / "ops/runner/select-latest-inclusive-cd-carrier.py"
+
+
+def _load_module():
+    spec = importlib.util.spec_from_file_location("select_latest_inclusive_cd_carrier", SCRIPT)
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+def _git(cwd: Path, *args: str) -> str:
+    return subprocess.run(
+        ["git", *args],
+        cwd=cwd,
+        check=True,
+        capture_output=True,
+        text=True,
+    ).stdout.strip()
+
+
+@pytest.fixture
+def linear_remote(tmp_path: Path) -> tuple[Path, Path, str, str]:
+    source = tmp_path / "source"
+    remote = tmp_path / "remote.git"
+    checkout = tmp_path / "checkout"
+    source.mkdir()
+    _git(source, "init", "-b", "main")
+    _git(source, "config", "user.name", "Carrier Test")
+    _git(source, "config", "user.email", "carrier@example.invalid")
+    (source / "value.txt").write_text("one\n", encoding="utf-8")
+    _git(source, "add", "value.txt")
+    _git(source, "commit", "-m", "first")
+    first = _git(source, "rev-parse", "HEAD")
+    _git(tmp_path, "clone", "--bare", str(source), str(remote))
+    _git(tmp_path, "clone", "--depth=1", remote.as_uri(), str(checkout))
+
+    (source / "value.txt").write_text("two\n", encoding="utf-8")
+    _git(source, "commit", "-am", "second")
+    second = _git(source, "rev-parse", "HEAD")
+    _git(source, "push", str(remote), "main")
+    return checkout, remote, first, second
+
+
+def test_latest_source_is_selected(linear_remote: tuple[Path, Path, str, str]) -> None:
+    module = _load_module()
+    checkout, remote, _, second = linear_remote
+
+    decision = module.select_latest_inclusive_carrier(
+        source_sha=second,
+        remote_url=str(remote),
+        cwd=checkout,
+    )
+
+    assert decision.selected is True
+    assert decision.status == "selected_latest_main_carrier"
+    assert decision.latest_sha == second
+
+
+def test_older_source_is_coalesced_only_when_latest_includes_it(
+    linear_remote: tuple[Path, Path, str, str],
+) -> None:
+    module = _load_module()
+    checkout, remote, first, second = linear_remote
+
+    decision = module.select_latest_inclusive_carrier(
+        source_sha=first,
+        remote_url=str(remote),
+        cwd=checkout,
+    )
+
+    assert decision.selected is False
+    assert decision.status == "superseded_by_latest_inclusive_carrier"
+    assert decision.source_sha == first
+    assert decision.latest_sha == second
+
+
+@pytest.mark.parametrize(
+    "marker_subject",
+    [
+        "chore(cd): deploy abcdef0 [skip ci]",
+        "chore(cd): cancel-stale-cd",
+        "chore(cd): refresh receipt [metadata-only]",
+    ],
+)
+def test_excluded_main_tip_does_not_displace_deployable_carrier(
+    linear_remote: tuple[Path, Path, str, str],
+    tmp_path: Path,
+    marker_subject: str,
+) -> None:
+    module = _load_module()
+    checkout, remote, _, second = linear_remote
+    marker_work = tmp_path / "marker-work"
+    _git(tmp_path, "clone", str(remote), str(marker_work))
+    _git(marker_work, "config", "user.name", "Carrier Test")
+    _git(marker_work, "config", "user.email", "carrier@example.invalid")
+    _git(marker_work, "commit", "--allow-empty", "-m", marker_subject)
+    marker_sha = _git(marker_work, "rev-parse", "HEAD")
+    _git(marker_work, "push", "origin", "main")
+
+    decision = module.select_latest_inclusive_carrier(
+        source_sha=second,
+        remote_url=str(remote),
+        cwd=checkout,
+    )
+
+    assert decision.selected is True
+    assert decision.latest_sha == second
+    assert decision.main_tip_sha == marker_sha
+
+
+@pytest.mark.parametrize(
+    "marker_body",
+    [
+        "release receipt [skip ci]",
+        "queue receipt cancel-stale-cd",
+        "runtime receipt [metadata-only]",
+    ],
+)
+def test_excluded_marker_in_commit_body_does_not_displace_deployable_carrier(
+    linear_remote: tuple[Path, Path, str, str],
+    tmp_path: Path,
+    marker_body: str,
+) -> None:
+    module = _load_module()
+    checkout, remote, _, second = linear_remote
+    marker_work = tmp_path / "marker-body-work"
+    _git(tmp_path, "clone", str(remote), str(marker_work))
+    _git(marker_work, "config", "user.name", "Carrier Test")
+    _git(marker_work, "config", "user.email", "carrier@example.invalid")
+    _git(
+        marker_work,
+        "commit",
+        "--allow-empty",
+        "-m",
+        "receipt marker",
+        "-m",
+        marker_body,
+    )
+    marker_sha = _git(marker_work, "rev-parse", "HEAD")
+    _git(marker_work, "push", "origin", "main")
+
+    decision = module.select_latest_inclusive_carrier(
+        source_sha=second,
+        remote_url=str(remote),
+        cwd=checkout,
+    )
+
+    assert decision.selected is True
+    assert decision.latest_sha == second
+    assert decision.main_tip_sha == marker_sha
+
+
+def test_unrelated_latest_main_fails_closed(tmp_path: Path) -> None:
+    module = _load_module()
+    checkout = tmp_path / "checkout"
+    unrelated = tmp_path / "unrelated"
+    remote = tmp_path / "remote.git"
+    for repo, value in ((checkout, "source"), (unrelated, "unrelated")):
+        repo.mkdir()
+        _git(repo, "init", "-b", "main")
+        _git(repo, "config", "user.name", "Carrier Test")
+        _git(repo, "config", "user.email", "carrier@example.invalid")
+        (repo / "value.txt").write_text(f"{value}\n", encoding="utf-8")
+        _git(repo, "add", "value.txt")
+        _git(repo, "commit", "-m", value)
+    source_sha = _git(checkout, "rev-parse", "HEAD")
+    _git(tmp_path, "clone", "--bare", str(unrelated), str(remote))
+
+    with pytest.raises(RuntimeError, match="latest_main_does_not_include_source"):
+        module.select_latest_inclusive_carrier(
+            source_sha=source_sha,
+            remote_url=str(remote),
+            cwd=checkout,
+        )
+
+
+def test_deep_linear_history_is_proven_after_bounded_deepen(tmp_path: Path) -> None:
+    module = _load_module()
+    source = tmp_path / "source"
+    remote = tmp_path / "remote.git"
+    checkout = tmp_path / "checkout"
+    source.mkdir()
+    _git(source, "init", "-b", "main")
+    _git(source, "config", "user.name", "Carrier Test")
+    _git(source, "config", "user.email", "carrier@example.invalid")
+    _git(source, "commit", "--allow-empty", "-m", "first")
+    first = _git(source, "rev-parse", "HEAD")
+    _git(tmp_path, "clone", "--bare", str(source), str(remote))
+    _git(tmp_path, "clone", "--depth=1", remote.as_uri(), str(checkout))
+    for index in range(270):
+        _git(source, "commit", "--allow-empty", "-m", f"linear-{index}")
+    latest = _git(source, "rev-parse", "HEAD")
+    _git(source, "push", str(remote), "main")
+
+    decision = module.select_latest_inclusive_carrier(
+        source_sha=first,
+        remote_url=str(remote),
+        cwd=checkout,
+    )
+
+    assert decision.selected is False
+    assert decision.latest_sha == latest
+
+
+def test_invalid_source_sha_fails_closed(tmp_path: Path) -> None:
+    module = _load_module()
+
+    with pytest.raises(RuntimeError, match="source_sha_invalid"):
+        module.select_latest_inclusive_carrier(
+            source_sha="not-a-sha",
+            remote_url="unused",
+            cwd=tmp_path,
+        )