diff --git a/.github/scripts/__tests__/sync-pr-merge-contract.test.js b/.github/scripts/__tests__/sync-pr-merge-contract.test.js
index e9181bd1b..db13b2555 100644
--- a/.github/scripts/__tests__/sync-pr-merge-contract.test.js
+++ b/.github/scripts/__tests__/sync-pr-merge-contract.test.js
@@ -8,6 +8,7 @@ const {
   buildMergeReport,
   classifySyncPrChecks,
   collectDeletableSyncBranches,
+  isTrustedSyncPr,
   normalizeSyncHash,
   parseBooleanInput,
   selectActiveSyncPr,
@@ -60,6 +61,12 @@ test('selectActiveSyncPr falls back to newest sync PR without a target hash', ()
   assert.equal(selection.missingExpected, false);
 });
 
+test('isTrustedSyncPr requires the configured actor and sync branch', () => {
+  const trusted = { ...pr(1, 'sync/workflows-current', '2026-04-25T01:00:00Z'), user: { login: 'stranske' } };
+  assert.equal(isTrustedSyncPr(trusted, ['stranske']), true);
+  assert.equal(isTrustedSyncPr({ ...trusted, user: { login: 'untrusted' } }, ['stranske']), false);
+});
+
 test('selectActiveSyncPr honors target hash instead of newest PR', () => {
   const selection = selectActiveSyncPr(
     [
diff --git a/.github/scripts/sync_pr_merge_contract.js b/.github/scripts/sync_pr_merge_contract.js
index c7472fc15..5cb8f0bec 100644
--- a/.github/scripts/sync_pr_merge_contract.js
+++ b/.github/scripts/sync_pr_merge_contract.js
@@ -27,6 +27,11 @@ function isSyncBranchName(value) {
   return branchNameFromRef(value).startsWith(SYNC_BRANCH_PREFIX);
 }
 
+function isTrustedSyncPr(pr, trustedActors = []) {
+  const actor = String(pr?.user?.login || '').trim();
+  return isSyncBranchName(pr?.head?.ref) && new Set(trustedActors).has(actor);
+}
+
 function parseBooleanInput(value, defaultValue = false) {
   if (value === undefined || value === null || String(value).trim() === '') {
     return Boolean(defaultValue);
@@ -262,6 +267,7 @@ module.exports = {
   classifySyncPrChecks,
   collectDeletableSyncBranches,
   isSyncBranchName,
+  isTrustedSyncPr,
   normalizeSyncHash,
   syncBranchForHash,
   parseBooleanInput,
diff --git a/.github/workflows/maint-68-sync-consumer-repos.yml b/.github/workflows/maint-68-sync-consumer-repos.yml
index 55d7afe6d..b0668a9fd 100644
--- a/.github/workflows/maint-68-sync-consumer-repos.yml
+++ b/.github/workflows/maint-68-sync-consumer-repos.yml
@@ -36,6 +36,15 @@ on:
         description: 'Force sync even if no changes detected'
         type: boolean
         default: false
+      phase:
+        description: 'Run preview (read-only), canary, or promote after green canary evidence'
+        type: choice
+        options: [preview, canary, promote]
+        default: canary
+      canary_evidence_json:
+        description: 'Maint 71 canary evidence JSON for the exact plan ID (required to promote)'
+        type: string
+        required: false
 
 permissions:
   contents: read
@@ -72,6 +81,8 @@ jobs:
     outputs:
       repos: ${{ steps.repos.outputs.matrix }}
       template_hash: ${{ steps.manifest.outputs.template_hash }}
+      plan_id: ${{ steps.manifest.outputs.plan_id }}
+      phase: ${{ steps.repos.outputs.phase }}
     steps:
       - name: Checkout
         uses: actions/checkout@v7
@@ -108,30 +119,46 @@ jobs:
             --output-json manifest.json \
             --github-output "$GITHUB_OUTPUT"
 
-      - name: Build repo matrix
+      - name: Build phase-gated repo matrix
         id: repos
+        env:
+          CANARY_EVIDENCE_JSON: ${{ inputs.canary_evidence_json || '' }}
+          REPOS_INPUT: ${{ inputs.repos }}
         run: |
-          if [ -n "${{ inputs.repos }}" ]; then
-            repos='${{ inputs.repos }}'
+          if [ -n "$REPOS_INPUT" ]; then
+            repos="$REPOS_INPUT"
           else
             repos=$(echo "$REGISTERED_CONSUMER_REPOS" | tr '\n' ',' | sed 's/,$//')
           fi
-          json_array=$(echo "$repos" \
-            | tr ',' '\n' \
-            | grep -v '^$' \
-            | jq -R -s -c '
-              split("\n")
-              | map(select(. != ""))
-              | map(gsub("^\\s+|\\s+$"; ""))
-            ')
-          echo "matrix={\"repo\":$json_array}" >> "$GITHUB_OUTPUT"
-          echo "Repos to sync: $json_array"
-
-      - name: Upload manifest artifact
+
+          # Scheduled and release runs intentionally begin with canaries. Promotion is
+          # a later explicit run with Maint 71's plan-bound evidence; no workflow waits
+          # for cross-repo CI inside this run.
+          phase='${{ inputs.phase || 'canary' }}'
+          selected_repos_args=()
+          if [ -n "$REPOS_INPUT" ]; then
+            selected_repos_args=(--selected-repos "$repos")
+          fi
+          python scripts/select_consumer_sync_phase.py \
+            --plan manifest.json \
+            --phase "$phase" \
+            --registered-repos "$(echo "$REGISTERED_CONSUMER_REPOS" | tr '\n' ',' | sed 's/,$//')" \
+            "${selected_repos_args[@]}" \
+            --canaries config/consumer_sync_canaries.json \
+            --canary-evidence-json "$CANARY_EVIDENCE_JSON" \
+            --output sync-phase-selection.json
+          echo "matrix=$(jq -c '.matrix' sync-phase-selection.json)" >> "$GITHUB_OUTPUT"
+          echo "phase=$phase" >> "$GITHUB_OUTPUT"
+          echo "Selected phase: $phase"
+          jq -r '.selected_repos[]' sync-phase-selection.json
+
+      - name: Upload plan and prospective-diff evidence
         uses: actions/upload-artifact@v7
         with:
-          name: sync-manifest
-          path: manifest.json
+          name: sync-plan-and-prospective-diffs
+          path: |
+            manifest.json
+            sync-phase-selection.json
           retention-days: 1
 
   # ============================================================================
@@ -156,7 +183,7 @@ jobs:
       - name: Download manifest
         uses: actions/download-artifact@v8
         with:
-          name: sync-manifest
+          name: sync-plan-and-prospective-diffs
 
       - name: Validate Python scripts
         run: |
@@ -224,6 +251,7 @@ jobs:
   sync:
     name: Sync ${{ matrix.repo }}
     needs: [prepare, validate]
+    if: needs.prepare.outputs.phase != 'preview'
     runs-on: ubuntu-latest
     strategy:
       fail-fast: false
@@ -231,6 +259,8 @@ jobs:
     env:
       # Job-level alias for repo token (uses OWNER_PR_PAT or SERVICE_BOT_PAT)
       REPO_TOKEN: ${{ secrets.OWNER_PR_PAT || secrets.SERVICE_BOT_PAT }}
+      PLAN_ID: ${{ needs.prepare.outputs.plan_id }}
+      SYNC_PHASE: ${{ needs.prepare.outputs.phase }}
     steps:
       - name: Compute result artifact name
         id: repo_meta
@@ -253,7 +283,7 @@ jobs:
       - name: Download manifest
         uses: actions/download-artifact@v8
         with:
-          name: sync-manifest
+          name: sync-plan-and-prospective-diffs
 
       - name: Setup Python
         uses: actions/setup-python@v7
@@ -812,6 +842,8 @@ jobs:
             --arg source_repository "$GITHUB_REPOSITORY" \
             --arg source_sha "$GITHUB_SHA" \
             --arg template_hash "${{ needs.prepare.outputs.template_hash }}" \
+            --arg plan_id "$PLAN_ID" \
+            --arg sync_phase "$SYNC_PHASE" \
             --arg sync_branch "$branch_name" \
             --arg manifest ".github/sync-manifest.yml" \
             --arg lifecycle_state "opened" \
@@ -824,6 +856,8 @@ jobs:
               source_repository: $source_repository,
               source_sha: $source_sha,
               template_hash: $template_hash,
+              plan_id: $plan_id,
+              sync_phase: $sync_phase,
               sync_branch: $sync_branch,
               manifest: $manifest,
               lifecycle_state: $lifecycle_state,
@@ -843,6 +877,8 @@ jobs:
           **Source:** stranske/Workflows
           **Source SHA:** \`$GITHUB_SHA\`
           **Template hash:** \`${{ needs.prepare.outputs.template_hash }}\`
+          **Consumer-sync plan ID:** \`$PLAN_ID\`
+          **Sync phase:** \`$SYNC_PHASE\`
           **Sync branch:** \`$branch_name\`
           **Consumer repo:** \`${{ matrix.repo }}\`
           **Manifest:** \`.github/sync-manifest.yml\`
@@ -878,6 +914,8 @@ jobs:
         env:
           REPO: ${{ matrix.repo }}
           TEMPLATE_HASH: ${{ needs.prepare.outputs.template_hash }}
+          PLAN_ID: ${{ needs.prepare.outputs.plan_id }}
+          SYNC_PHASE: ${{ needs.prepare.outputs.phase }}
           DRY_RUN: ${{ inputs.dry_run || 'false' }}
           FORCE: ${{ inputs.force || 'false' }}
           SYNC_OUTCOME: ${{ steps.sync.outcome || 'skipped' }}
@@ -923,6 +961,8 @@ jobs:
               "repo": repo,
               "status": status,
               "template_hash": os.environ.get("TEMPLATE_HASH", ""),
+              "plan_id": os.environ.get("PLAN_ID", ""),
+              "sync_phase": os.environ.get("SYNC_PHASE", ""),
               "expected_branch": (
                   f"sync/workflows-{os.environ.get('TEMPLATE_HASH', '')}"
                   if os.environ.get("TEMPLATE_HASH", "")
diff --git a/.github/workflows/maint-71-merge-sync-prs.yml b/.github/workflows/maint-71-merge-sync-prs.yml
index 505a478d8..abb55bc15 100644
--- a/.github/workflows/maint-71-merge-sync-prs.yml
+++ b/.github/workflows/maint-71-merge-sync-prs.yml
@@ -122,6 +122,7 @@ jobs:
           DRY_RUN_INPUT: ${{ inputs.dry_run }}
           SYNC_HASH_INPUT: ${{ inputs.sync_hash || '' }}
           CLEANUP_BRANCHES_INPUT: ${{ inputs.cleanup_branches }}
+          TRUSTED_SYNC_ACTORS: stranske,stranske-automation-bot,github-actions[bot]
           SYNC_PR_MERGE_REPORT_JSON: artifacts/sync-pr-merge-report.json
         with:
           github-token: ${{ secrets.OWNER_PR_PAT || secrets.SERVICE_BOT_PAT || github.token }}
@@ -137,6 +138,7 @@ jobs:
               collectDeletableSyncBranches,
               normalizeSyncHash,
               parseBooleanInput,
+              isTrustedSyncPr,
               selectActiveSyncPr,
               selectSyncPrGatingChecks,
             } = require('./.github/scripts/sync_pr_merge_contract.js');
@@ -250,8 +252,11 @@ jobs:
             const requestedSyncHash = normalizeSyncHash(
               process.env.SYNC_HASH_INPUT ||
                 (context.payload.client_payload && context.payload.client_payload.sync_hash) ||
-                '',
+              '',
             );
+            const trustedSyncActors = process.env.TRUSTED_SYNC_ACTORS.split(',')
+              .map((actor) => actor.trim())
+              .filter(Boolean);
 
             console.log(`Registered consumer repos: ${registeredRepos.join(', ')}`);
             console.log(`Processing repos: ${targetRepos.join(', ')}`);
@@ -262,6 +267,50 @@ jobs:
             }
 
             const results = [];
+            const canaryEvidence = [];
+
+            function syncMetadata(pr) {
+              const match = String(pr.body || '').match(
+                /<!-- workflows-consumer-sync:v1 ([\s\S]*?) -->/,
+              );
+              if (!match) return null;
+              try {
+                return JSON.parse(match[1]);
+              } catch (_) {
+                return null;
+              }
+            }
+
+            async function activeReviewThreadCount(owner, repo, number) {
+              try {
+                const data = await github.graphql(
+                  `query($owner: String!, $repo: String!, $number: Int!) {
+                    repository(owner: $owner, name: $repo) {
+                      pullRequest(number: $number) {
+                        reviewThreads(first: 100) {
+                          pageInfo { hasNextPage }
+                          nodes { isResolved isOutdated }
+                        }
+                      }
+                    }
+                  }`,
+                  { owner, repo, number },
+                );
+                const reviewThreads = data.repository.pullRequest.reviewThreads;
+                if (reviewThreads.pageInfo.hasNextPage) {
+                  core.warning(
+                    `Review-thread pagination exceeded the safe evidence window for ${owner}/${repo}#${number}`,
+                  );
+                  return -1;
+                }
+                return reviewThreads.nodes.filter(
+                  (thread) => !thread.isResolved && !thread.isOutdated,
+                ).length;
+              } catch (error) {
+                core.warning(`Unable to read active review threads for ${owner}/${repo}#${number}: ${error}`);
+                return -1;
+              }
+            }
 
             for (const repoEntry of targetRepos) {
               const [entryOwner, entryRepo] = repoEntry.includes('/')
@@ -281,7 +330,7 @@ jobs:
                   per_page: 20
                 }));
 
-                const syncPRs = prs.filter(pr => pr.head.ref.startsWith('sync/workflows-'));
+                const syncPRs = prs.filter((pr) => isTrustedSyncPr(pr, trustedSyncActors));
 
                 if (cleanupBranches) {
                   try {
@@ -448,6 +497,7 @@ jobs:
 
                 // Process the selected active PR
                 const pr = selection.active;
+                const metadata = syncMetadata(pr);
                 console.log(`\nProcessing active PR #${pr.number}: ${pr.title}`);
                 console.log(`Branch: ${pr.head.ref}`);
                 console.log(`Created: ${pr.created_at}`);
@@ -489,6 +539,17 @@ jobs:
                 const failedChecks = classification.failed;
                 const pendingChecks = classification.pending;
 
+                if (metadata?.sync_phase === 'canary' && metadata?.plan_id) {
+                  canaryEvidence.push({
+                    repo: `${owner}/${repo}`,
+                    plan_id: metadata.plan_id,
+                    pr: pr.number,
+                    required_check_state:
+                      classification.status === 'ready' ? 'success' : classification.status,
+                    active_review_thread_count: await activeReviewThreadCount(owner, repo, pr.number),
+                  });
+                }
+
                 console.log(
                   `Checks (${checkGateMode}): ${gatingChecks.length} gating, ` +
                     `${failedChecks.length} failed, ${pendingChecks.length} pending`,
@@ -686,6 +747,15 @@ jobs:
             const reportPath = process.env.SYNC_PR_MERGE_REPORT_JSON;
             fs.mkdirSync(path.dirname(reportPath), { recursive: true });
             fs.writeFileSync(reportPath, `${JSON.stringify(report, null, 2)}\n`, 'utf8');
+            fs.writeFileSync(
+              'artifacts/sync-canary-evidence.json',
+              `${JSON.stringify({
+                schema: 'workflows.consumer-sync-canary-evidence/v1',
+                version: 1,
+                results: canaryEvidence,
+              }, null, 2)}\n`,
+              'utf8',
+            );
             await core.summary.addRaw(buildMarkdownSummary(report)).write();
 
             const merged = report.summary.merged;
@@ -735,6 +805,8 @@ jobs:
         if: always()
         uses: actions/upload-artifact@v7
         with:
-          name: sync-pr-merge-report
-          path: artifacts/sync-pr-merge-report.json
+          name: sync-pr-merge-and-canary-evidence
+          path: |
+            artifacts/sync-pr-merge-report.json
+            artifacts/sync-canary-evidence.json
           if-no-files-found: warn
diff --git a/config/consumer_sync_canaries.json b/config/consumer_sync_canaries.json
new file mode 100644
index 000000000..f3f6b60f8
--- /dev/null
+++ b/config/consumer_sync_canaries.json
@@ -0,0 +1,18 @@
+{
+  "schema": "workflows.consumer-sync-canaries/v1",
+  "version": 1,
+  "canaries": [
+    {
+      "repo": "stranske/Travel-Plan-Permission",
+      "capabilities": ["standard", "custom-gate"]
+    },
+    {
+      "repo": "stranske/trip-planner",
+      "capabilities": ["lock-heavy", "node-tooling"]
+    },
+    {
+      "repo": "stranske/Manager-Database",
+      "capabilities": ["standard", "python-consumer"]
+    }
+  ]
+}
diff --git a/docs/WORKFLOW_GUIDE.md b/docs/WORKFLOW_GUIDE.md
index cdb667ba4..678842298 100644
--- a/docs/WORKFLOW_GUIDE.md
+++ b/docs/WORKFLOW_GUIDE.md
@@ -58,12 +58,12 @@ _Inline Gate helper_
 - **`maint-62-integration-consumer.yml`** — Nightly + release-triggered integration tests that reuse `reusable-10-ci-python.yml` across multiple matrices and file/resolve the `integration-test` issue via the load-balanced API client (no extra app mint).
 - **`maint-65-sync-label-docs.yml`** — Syncs `docs/LABELS.md` into every registered consumer repo (plus the integration tests repo) when the source doc changes or on demand, using the shared registered-repo helper and PAT gating for cross-repo pushes.
 - **`maint-66-monthly-audit.yml`** — First-of-month workflow that gathers workflow-run stats, runs the API wrapper guard, and files/updates the monthly audit issue; relies on the shared API client so no extra npm installs or App-token mints are needed.
-- **`maint-68-sync-consumer-repos.yml`** — Manifest-driven consumer sync that validates template/scripts, hashes the template set, and opens PAT-backed sync PRs for each registered repo; the jobs now rely solely on the shared API client and repo PATs (no extra App token mints).
+- **`maint-68-sync-consumer-repos.yml`** — Manifest-driven consumer sync that validates template/scripts, hashes the template set, records a prospective per-repo matrix, and opens PAT-backed sync PRs. Normal no-filter runs start with the configured canaries; a later `promote` run requires same-plan, green, review-clear Maint 71 evidence before it targets non-canaries. The jobs rely solely on the shared API client and repo PATs (no extra App token mints).
 - **`maint-69-sync-integration-repo.yml`** — Keeps Workflows-Integration-Tests aligned with `templates/integration-repo/`, regenerates `requirements.lock`, and pushes updates using PATs; no GitHub App token mint is required because the workflow stays inside the two repos.
 - **`maint-69-sync-labels.yml`** — Propagates the canonical `.github/labels-core.yml` set to every registered consumer repo (or a provided subset), reusing the registered-repo helper + load-balanced API client without any additional App-token minting.
 - **`maint-70-fix-integration-formatting.yml`** — Manual formatter for Workflows-Integration-Tests that resolves the repo default branch, applies `black`+`ruff` fixes, and pushes via PAT only when a token is available; runs read-only otherwise.
 - **`maint-71-auto-fix-integration.yml`** — Auto-triggered integration fixer that watches “Integration CI failed” issues/comments, re-runs the formatting routine, and pushes via PAT when available (otherwise posting a skipped note).
-- **`maint-71-merge-sync-prs.yml`** — Scans each registered consumer repo for open `sync/workflows-*` PRs, closes stale duplicates, deletes leftover same-repo sync branches tied to closed/merged sync PRs, and (optionally) auto-merges passing PRs using the shared repo helper + PATs. PRs carrying runtime-AC labels are not merged by this external lane; they must pass through the local Orchestrator runtime AC guard.
+- **`maint-71-merge-sync-prs.yml`** — Scans each registered consumer repo for open `sync/workflows-*` PRs, closes stale duplicates, deletes leftover same-repo sync branches tied to closed/merged sync PRs, and (optionally) auto-merges passing PRs using the shared repo helper + PATs. It emits plan-bound canary evidence with the PR/check/review state for Maint 68 promotion. PRs carrying runtime-AC labels are not merged by this external lane; they must pass through the local Orchestrator runtime AC guard.
 - **`maint-72-fix-pr-body-conflicts.yml`** — Periodically removes stray `pr_body.md` files from consumer repos and ensures `.gitignore` blocks them, reusing the registered-repo helper + PAT discovery so cleanups only run when push access is available.
 - **`maint-74-ledger-base-sync.yml`** — Keeps `.agents` ledger base entries aligned with the repo’s default branch by running `scripts/ledger_migrate_base.py` and opening a helper PR (no extra App token mint needed).
 - **`maint-auto-update-pypi-versions.yml`** — Daily PyPI watcher that updates `autofix-versions.env`, regenerates supporting files, and opens a PR when new tool versions land (runs entirely with the default token + GH CLI).
diff --git a/docs/ops/CONSUMER_REPO_MAINTENANCE.md b/docs/ops/CONSUMER_REPO_MAINTENANCE.md
index 72be1d314..043f6ef2f 100644
--- a/docs/ops/CONSUMER_REPO_MAINTENANCE.md
+++ b/docs/ops/CONSUMER_REPO_MAINTENANCE.md
@@ -233,6 +233,26 @@ handoff is explicitly `shadow`, `write_authority=false`, and
 and promotion blockers remain owned by Orchestrator's
 `consumer_sync_shadow.py` dashboard.
 
+#### Canary-Gated Fan-out
+
+Maint 68 separates a sync plan into `preview`, `canary`, and `promote` phases.
+An ordinary scheduled, release, or manual no-filter run defaults to `canary`:
+it can open sync PRs only for the 2-3 representative repositories declared in
+`config/consumer_sync_canaries.json`. The selection artifact records the exact
+compiled `plan_id`, desired hash, and prospective affected paths for every
+registered repository before any consumer write.
+
+Do not wait for consumer CI in that workflow. Run Maint 71 later to publish
+`sync-canary-evidence.json`, then invoke Maint 68 with `phase=promote` and that
+artifact's JSON as `canary_evidence_json`. Promotion rejects absent, stale or
+mixed-plan evidence, failed required checks, and active non-outdated review
+threads. A successful promotion targets all registered non-canary repositories
+once every configured canary has current, green, review-clear evidence for the
+same plan.
+Use `preview` to produce the plan/evidence artifact without a write matrix.
+Emergency direct promotion remains an explicit audited operator action and is
+limited to a security or production-break fix.
+
 To validate the manifest locally:
 
 ```bash
diff --git a/scripts/select_consumer_sync_phase.py b/scripts/select_consumer_sync_phase.py
new file mode 100644
index 000000000..0df5b9b79
--- /dev/null
+++ b/scripts/select_consumer_sync_phase.py
@@ -0,0 +1,207 @@
+#!/usr/bin/env python3
+"""Select a plan-bound preview, canary, or promotion matrix for Maint 68.
+
+The selector is deliberately read-only.  It turns the already compiled manifest
+plan into prospective per-repository evidence, and only permits the promote
+matrix when every configured canary has current, green, review-clear evidence
+for the exact same plan ID.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+from typing import Any
+
+try:
+    from scripts.sync_manifest_compiler import PLAN_SCHEMA
+except ImportError:
+    from sync_manifest_compiler import PLAN_SCHEMA  # type: ignore[no-redef]
+
+CANARY_SCHEMA = "workflows.consumer-sync-canaries/v1"
+PHASES = {"preview", "canary", "promote"}
+
+
+class PhaseSelectionError(ValueError):
+    """The requested phase cannot safely construct a write matrix."""
+
+
+def _read_json(path: Path) -> Any:
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as exc:
+        raise PhaseSelectionError(f"invalid_json:{path}") from exc
+
+
+def _parse_repos(raw: str) -> list[str]:
+    repos = [repo.strip() for repo in raw.split(",") if repo.strip()]
+    if not repos or len(repos) != len(set(repos)):
+        raise PhaseSelectionError("registered_repos_must_be_nonempty_and_unique")
+    return repos
+
+
+def _load_canaries(path: Path, registered_repos: list[str]) -> list[dict[str, Any]]:
+    config = _read_json(path)
+    if not isinstance(config, dict) or config.get("schema") != CANARY_SCHEMA:
+        raise PhaseSelectionError("unsupported_canary_config_schema")
+    canaries = config.get("canaries")
+    if not isinstance(canaries, list) or not 2 <= len(canaries) <= 3:
+        raise PhaseSelectionError("canary_config_requires_two_or_three_repos")
+    repos: list[str] = []
+    normalized: list[dict[str, Any]] = []
+    for item in canaries:
+        if not isinstance(item, dict):
+            raise PhaseSelectionError("invalid_canary_entry")
+        repo = item.get("repo")
+        capabilities = item.get("capabilities")
+        if (
+            not isinstance(repo, str)
+            or repo not in registered_repos
+            or not isinstance(capabilities, list)
+            or not capabilities
+            or not all(isinstance(capability, str) and capability for capability in capabilities)
+        ):
+            raise PhaseSelectionError("invalid_canary_entry")
+        repos.append(repo)
+        normalized.append({"repo": repo, "capabilities": sorted(capabilities)})
+    if len(repos) != len(set(repos)):
+        raise PhaseSelectionError("duplicate_canary_repo")
+    return normalized
+
+
+def _validate_plan(plan: Any) -> dict[str, Any]:
+    if not isinstance(plan, dict) or plan.get("schema") != PLAN_SCHEMA:
+        raise PhaseSelectionError("unsupported_consumer_sync_plan")
+    if not isinstance(plan.get("plan_id"), str) or not plan["plan_id"].startswith("sha256:"):
+        raise PhaseSelectionError("invalid_consumer_sync_plan_id")
+    entries = plan.get("entries")
+    removals = plan.get("removals")
+    if not isinstance(entries, list) or not isinstance(removals, list):
+        raise PhaseSelectionError("invalid_consumer_sync_plan_collections")
+    return plan
+
+
+def _evidence_rows(raw: str) -> list[dict[str, Any]]:
+    if not raw.strip():
+        return []
+    try:
+        parsed = json.loads(raw)
+    except json.JSONDecodeError as exc:
+        raise PhaseSelectionError("invalid_canary_evidence_json") from exc
+    if isinstance(parsed, dict):
+        parsed = parsed.get("results")
+    if not isinstance(parsed, list) or not all(isinstance(item, dict) for item in parsed):
+        raise PhaseSelectionError("invalid_canary_evidence_json")
+    return parsed
+
+
+def _promotion_rejections(
+    *, plan_id: str, canary_repos: list[str], evidence: list[dict[str, Any]]
+) -> list[str]:
+    by_repo: dict[str, dict[str, Any]] = {}
+    reasons: list[str] = []
+    for item in evidence:
+        repo = str(item.get("repo", ""))
+        if repo in canary_repos and repo in by_repo:
+            reasons.append(f"duplicate_canary_evidence:{repo}")
+            continue
+        by_repo[repo] = item
+    for repo in canary_repos:
+        item = by_repo.get(repo)
+        if item is None:
+            reasons.append(f"missing_canary_evidence:{repo}")
+            continue
+        if item.get("plan_id") != plan_id:
+            reasons.append(f"stale_or_mixed_plan:{repo}")
+        if item.get("required_check_state") != "success":
+            reasons.append(f"required_checks_not_green:{repo}")
+        if item.get("active_review_thread_count") != 0:
+            reasons.append(f"active_review_debt:{repo}")
+    return reasons
+
+
+def select_phase(
+    plan: Any,
+    *,
+    phase: str,
+    registered_repos: list[str],
+    canaries: list[dict[str, Any]],
+    evidence: list[dict[str, Any]] | None = None,
+    selected_repos: list[str] | None = None,
+) -> dict[str, Any]:
+    """Return a deterministic matrix plus the evidence needed to audit it."""
+    if phase not in PHASES:
+        raise PhaseSelectionError("unsupported_sync_phase")
+    plan = _validate_plan(plan)
+    canary_repos = [item["repo"] for item in canaries]
+    if selected_repos is not None and not set(selected_repos) <= set(registered_repos):
+        raise PhaseSelectionError("selected_repos_must_be_registered")
+    target_repos = selected_repos if selected_repos is not None else registered_repos
+    paths = sorted(
+        {str(entry.get("target")) for entry in plan["entries"] if entry.get("target")}
+        | {str(removal.get("target")) for removal in plan["removals"] if removal.get("target")}
+    )
+    prospective = [
+        {
+            "repo": repo,
+            "desired_hash": plan["plan_id"],
+            "affected_paths": paths,
+            "canary": repo in canary_repos,
+        }
+        for repo in target_repos
+    ]
+    if phase == "preview":
+        selected = []
+    elif phase == "canary":
+        selected = target_repos if selected_repos is not None else canary_repos
+    else:
+        reasons = _promotion_rejections(
+            plan_id=plan["plan_id"],
+            canary_repos=canary_repos,
+            evidence=evidence or [],
+        )
+        if reasons:
+            raise PhaseSelectionError("promotion_rejected:" + ",".join(reasons))
+        selected = [repo for repo in target_repos if repo not in canary_repos]
+    return {
+        "schema": "workflows.consumer-sync-phase-selection/v1",
+        "version": 1,
+        "phase": phase,
+        "plan_id": plan["plan_id"],
+        "canaries": canaries,
+        "selected_repos": selected,
+        "matrix": {"repo": selected},
+        "prospective_diffs": prospective,
+        "promotion_allowed": phase == "promote",
+    }
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--plan", type=Path, required=True)
+    parser.add_argument("--phase", choices=sorted(PHASES), required=True)
+    parser.add_argument("--registered-repos", required=True)
+    parser.add_argument("--selected-repos", default="")
+    parser.add_argument("--canaries", type=Path, required=True)
+    parser.add_argument("--canary-evidence-json", default="")
+    parser.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args(argv)
+    try:
+        result = select_phase(
+            _read_json(args.plan),
+            phase=args.phase,
+            registered_repos=_parse_repos(args.registered_repos),
+            canaries=_load_canaries(args.canaries, _parse_repos(args.registered_repos)),
+            evidence=_evidence_rows(args.canary_evidence_json),
+            selected_repos=_parse_repos(args.selected_repos) if args.selected_repos else None,
+        )
+    except PhaseSelectionError as exc:
+        parser.error(str(exc))
+    args.output.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    print(json.dumps(result, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/tests/scripts/test_select_consumer_sync_phase.py b/tests/scripts/test_select_consumer_sync_phase.py
new file mode 100644
index 000000000..63c8ac2cc
--- /dev/null
+++ b/tests/scripts/test_select_consumer_sync_phase.py
@@ -0,0 +1,151 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+from scripts.select_consumer_sync_phase import PhaseSelectionError, select_phase
+from scripts.sync_manifest_compiler import compile_manifest
+
+ROOT = Path(__file__).parents[2]
+REGISTERED = [
+    "stranske/Travel-Plan-Permission",
+    "stranske/trip-planner",
+    "stranske/Manager-Database",
+    "stranske/Ready",
+]
+CANARIES = [
+    {"repo": "stranske/Travel-Plan-Permission", "capabilities": ["custom-gate"]},
+    {"repo": "stranske/trip-planner", "capabilities": ["lock-heavy"]},
+]
+
+
+def plan() -> dict:
+    return compile_manifest(ROOT / ".github" / "sync-manifest.yml", repo_root=ROOT).to_plan()
+
+
+def green_evidence(plan_id: str) -> list[dict]:
+    return [
+        {
+            "repo": canary["repo"],
+            "plan_id": plan_id,
+            "pr": 100 + index,
+            "required_check_state": "success",
+            "active_review_thread_count": 0,
+        }
+        for index, canary in enumerate(CANARIES)
+    ]
+
+
+def test_canary_phase_selects_only_representative_repos() -> None:
+    result = select_phase(plan(), phase="canary", registered_repos=REGISTERED, canaries=CANARIES)
+
+    assert result["selected_repos"] == [item["repo"] for item in CANARIES]
+    assert len(result["prospective_diffs"]) == len(REGISTERED)
+    assert result["matrix"] == {"repo": result["selected_repos"]}
+    assert all(item["desired_hash"] == result["plan_id"] for item in result["prospective_diffs"])
+
+
+def test_preview_never_constructs_a_write_matrix() -> None:
+    result = select_phase(plan(), phase="preview", registered_repos=REGISTERED, canaries=CANARIES)
+
+    assert result["selected_repos"] == []
+    assert len(result["prospective_diffs"]) == len(REGISTERED)
+
+
+def test_promotion_rejects_stale_canary_evidence() -> None:
+    compiled = plan()
+    stale = green_evidence("sha256:" + "0" * 64)
+
+    with pytest.raises(PhaseSelectionError, match="stale_or_mixed_plan"):
+        select_phase(
+            compiled,
+            phase="promote",
+            registered_repos=REGISTERED,
+            canaries=CANARIES,
+            evidence=stale,
+        )
+
+
+def test_promotion_requires_green_review_clear_evidence_for_each_canary() -> None:
+    compiled = plan()
+    evidence = green_evidence(compiled["plan_id"])
+    evidence[0]["required_check_state"] = "failure"
+    evidence[1]["active_review_thread_count"] = 1
+
+    with pytest.raises(PhaseSelectionError, match="required_checks_not_green"):
+        select_phase(
+            compiled,
+            phase="promote",
+            registered_repos=REGISTERED,
+            canaries=CANARIES,
+            evidence=evidence,
+        )
+
+
+def test_promotion_rejects_missing_or_duplicate_canary_evidence() -> None:
+    compiled = plan()
+    evidence = green_evidence(compiled["plan_id"])
+
+    with pytest.raises(PhaseSelectionError, match="missing_canary_evidence"):
+        select_phase(
+            compiled,
+            phase="promote",
+            registered_repos=REGISTERED,
+            canaries=CANARIES,
+            evidence=evidence[:-1],
+        )
+
+    with pytest.raises(PhaseSelectionError, match="duplicate_canary_evidence"):
+        select_phase(
+            compiled,
+            phase="promote",
+            registered_repos=REGISTERED,
+            canaries=CANARIES,
+            evidence=[*evidence, evidence[0]],
+        )
+
+
+def test_promotion_targets_only_non_canary_repos() -> None:
+    compiled = plan()
+    result = select_phase(
+        compiled,
+        phase="promote",
+        registered_repos=REGISTERED,
+        canaries=CANARIES,
+        evidence=green_evidence(compiled["plan_id"]),
+    )
+
+    assert result["selected_repos"] == ["stranske/Manager-Database", "stranske/Ready"]
+
+
+def test_filtered_manual_canary_run_preserves_requested_repositories() -> None:
+    result = select_phase(
+        plan(),
+        phase="canary",
+        registered_repos=REGISTERED,
+        selected_repos=["stranske/Ready"],
+        canaries=CANARIES,
+    )
+
+    assert result["selected_repos"] == ["stranske/Ready"]
+
+
+def test_manual_selection_cannot_target_an_unregistered_repository() -> None:
+    # A manual filtered run must not become a fan-out escape hatch: one unregistered
+    # entry alongside valid ones has to reject the whole selection, not silently drop it.
+    with pytest.raises(PhaseSelectionError, match="selected_repos_must_be_registered"):
+        select_phase(
+            plan(),
+            phase="canary",
+            registered_repos=REGISTERED,
+            selected_repos=["stranske/Ready", "stranske/Not-A-Consumer"],
+            canaries=CANARIES,
+        )
+
+
+def test_checked_in_canary_config_covers_distinct_consumer_shapes() -> None:
+    config = json.loads((ROOT / "config" / "consumer_sync_canaries.json").read_text())
+    covered = {tag for canary in config["canaries"] for tag in canary["capabilities"]}
+
+    assert {"standard", "custom-gate", "lock-heavy"} <= covered
diff --git a/tests/workflows/test_sync_manifest_delivery.py b/tests/workflows/test_sync_manifest_delivery.py
index 48eec109f..86dd33c10 100644
--- a/tests/workflows/test_sync_manifest_delivery.py
+++ b/tests/workflows/test_sync_manifest_delivery.py
@@ -28,6 +28,7 @@
 
 from __future__ import annotations
 
+import json
 from pathlib import Path
 
 import yaml
@@ -206,3 +207,41 @@ def test_prepare_checkout_includes_manifest_owned_github_roots() -> None:
     assert ".gitattributes" in {
         line.strip() for line in sparse_checkout.splitlines() if line.strip()
     }
+
+
+def test_sync_fanout_is_canary_gated_and_promotion_is_plan_bound() -> None:
+    workflow = yaml.safe_load(SYNC_WORKFLOW_PATH.read_text(encoding="utf-8"))
+    dispatch_inputs = workflow.get("on", workflow.get(True))["workflow_dispatch"]["inputs"]
+    prepare = workflow["jobs"]["prepare"]
+    sync = workflow["jobs"]["sync"]
+    source = SYNC_WORKFLOW_PATH.read_text(encoding="utf-8")
+
+    assert dispatch_inputs["phase"]["default"] == "canary"
+    assert set(dispatch_inputs["phase"]["options"]) == {"preview", "canary", "promote"}
+    assert "canary_evidence_json" in dispatch_inputs
+    assert prepare["outputs"]["phase"] == "${{ steps.repos.outputs.phase }}"
+    assert sync["if"] == "needs.prepare.outputs.phase != 'preview'"
+    assert "select_consumer_sync_phase.py" in source
+    upload = next(
+        step for step in prepare["steps"] if step.get("uses") == "actions/upload-artifact@v7"
+    )
+    download = next(
+        step for step in sync["steps"] if step.get("uses") == "actions/download-artifact@v8"
+    )
+    assert upload["with"]["name"] == "sync-plan-and-prospective-diffs"
+    assert download["with"]["name"] == upload["with"]["name"]
+
+    config = json.loads((REPO_ROOT / "config" / "consumer_sync_canaries.json").read_text())
+    assert config["schema"] == "workflows.consumer-sync-canaries/v1"
+    assert 2 <= len(config["canaries"]) <= 3
+
+
+def test_maint_71_emits_canary_evidence_with_review_debt() -> None:
+    source = (REPO_ROOT / ".github" / "workflows" / "maint-71-merge-sync-prs.yml").read_text(
+        encoding="utf-8"
+    )
+
+    assert "sync-canary-evidence.json" in source
+    assert "active_review_thread_count" in source
+    assert "required_check_state" in source
+    assert "plan_id" in source