fix(merge-control): fail-closed REQUIRED_CHECKS_JSON parsing + BP drift guard (#2401 RC)

CR2 gate-review fixes:

1. REQUIRED_CHECKS_JSON parsing now rejects non-string items:
   - int/bool/null/dict values fail with explicit index+type error
   - whitespace-only strings fail with explicit index error
   - Both scripts (gitea-merge-queue.py, ci-required-drift.py) now
     validate isinstance(item, str) before strip(), exiting non-zero.

2. ci-required-drift.py detects duplicate entries:
   - Set-collapse silently hid duplicates; now tracked via 'seen' set
   - Duplicate triggers ::error:: with index and exit 3.

3. gitea-merge-queue.py adds gate-source drift guard:
   - After fetching BP and parsing JSON, normalizes contexts (strips
     (pull_request)/(push) suffix) and compares sets.
   - Mismatch → queue holds with ::error:: listing JSON-only and BP-only
     contexts. Prevents force-merge bypass of a BP-required context
     omitted from JSON.

4. Regression tests:
   - test_required_checks_from_env_rejects_{non_string,bool,null,whitespace}
     for merge queue (4 tests)
   - Same 4 + duplicate rejection for drift (5 tests)
   - test_process_once_holds_when_json_drifts_from_branch_protection
     proving queue does not merge when JSON omits a BP-required context.

All 87 tests pass.

.gitea/scripts/ci-required-drift.py

@@ -287,7 +287,32 @@ def required_checks_from_env(branch: str) -> set[str]:
             f"::error::REQUIRED_CHECKS_JSON['{branch}'] is {type(branch_checks).__name__}, expected list\n"
         )
         sys.exit(3)
-    result = {str(item).strip() for item in branch_checks if str(item).strip()}
+    seen: set[str] = set()
+    result: set[str] = set()
+    for idx, item in enumerate(branch_checks):
+        if not isinstance(item, str):
+            sys.stderr.write(
+                f"::error::REQUIRED_CHECKS_JSON['{branch}'][{idx}] is "
+                f"{type(item).__name__} (value={item!r}), expected str "
+                "(fail-closed)\n"
+            )
+            sys.exit(3)
+        stripped = item.strip()
+        if not stripped:
+            sys.stderr.write(
+                f"::error::REQUIRED_CHECKS_JSON['{branch}'][{idx}] is "
+                "whitespace-only (fail-closed)\n"
+            )
+            sys.exit(3)
+        if stripped in seen:
+            sys.stderr.write(
+                f"::error::REQUIRED_CHECKS_JSON['{branch}'] contains "
+                f"duplicate entry '{stripped}' at index {idx} "
+                "(fail-closed)\n"
+            )
+            sys.exit(3)
+        seen.add(stripped)
+        result.add(stripped)
     if not result:
         sys.stderr.write(
             f"::error::REQUIRED_CHECKS_JSON['{branch}'] is empty after stripping; "

.gitea/scripts/gitea-merge-queue.py

@@ -333,7 +333,23 @@ def required_checks_from_env(branch: str) -> list[str]:
             f"::error::REQUIRED_CHECKS_JSON['{branch}'] is {type(branch_checks).__name__}, expected list\n"
         )
         sys.exit(2)
-    result = [str(item).strip() for item in branch_checks if str(item).strip()]
+    result: list[str] = []
+    for idx, item in enumerate(branch_checks):
+        if not isinstance(item, str):
+            sys.stderr.write(
+                f"::error::REQUIRED_CHECKS_JSON['{branch}'][{idx}] is "
+                f"{type(item).__name__} (value={item!r}), expected str "
+                "(fail-closed)\n"
+            )
+            sys.exit(2)
+        stripped = item.strip()
+        if not stripped:
+            sys.stderr.write(
+                f"::error::REQUIRED_CHECKS_JSON['{branch}'][{idx}] is "
+                "whitespace-only (fail-closed)\n"
+            )
+            sys.exit(2)
+        result.append(stripped)
     if not result:
         sys.stderr.write(
             f"::error::REQUIRED_CHECKS_JSON['{branch}'] is empty after stripping; "
@@ -1001,6 +1017,29 @@ def process_once(*, dry_run: bool = False) -> int:
         f"pr_required={pr_required_contexts} push_required={push_required}"
     )
 
+    # Gate-source drift guard (mc#1982): REQUIRED_CHECKS_JSON must be
+    # set-equal to branch protection status_check_contexts (modulo event
+    # suffix). A mismatch means a required context could be misclassified
+    # as non-required (or vice versa), weakening the fail-closed contract.
+    def _normalize(ctx: str) -> str:
+        for suffix in (" (pull_request)", " (push)"):
+            if ctx.endswith(suffix):
+                return ctx[: -len(suffix)]
+        return ctx
+
+    bp_contexts = {_normalize(c) for c in bp.required_contexts}
+    json_contexts = {_normalize(c) for c in pr_required_contexts}
+    if bp_contexts != json_contexts:
+        only_in_json = sorted(json_contexts - bp_contexts)
+        only_in_bp = sorted(bp_contexts - json_contexts)
+        sys.stderr.write(
+            f"::error::queue held: REQUIRED_CHECKS_JSON drift against "
+            f"branch protection for {WATCH_BRANCH}. "
+            f"JSON only: {only_in_json}. BP only: {only_in_bp}. "
+            "Fix the drift before merging.\n"
+        )
+        return 0
+
     main_sha = get_branch_head(WATCH_BRANCH)
     main_status = get_combined_status(main_sha)
     # Check push-required contexts explicitly instead of combined state.

.gitea/scripts/tests/test_ci_required_drift.py

@@ -4,6 +4,8 @@ import sys
 from pathlib import Path
 from unittest.mock import patch
 
+import pytest
+
 SCRIPT = Path(__file__).resolve().parents[1] / "ci-required-drift.py"
 spec = importlib.util.spec_from_file_location("ci_required_drift", SCRIPT)
 drift = importlib.util.module_from_spec(spec)
@@ -200,3 +202,48 @@ def test_detect_drift_no_f1_when_needs_empty_even_with_jobs():
                 findings, _ = drift.detect_drift("main")
 
     assert not any("F1 —" in f for f in findings)
+
+
+def test_required_checks_from_env_rejects_non_string_item():
+    os.environ["REQUIRED_CHECKS_JSON"] = json.dumps(
+        {"main": ["ctx-a", 123]}
+    )
+    with pytest.raises(SystemExit) as exc_info:
+        drift.required_checks_from_env("main")
+    assert exc_info.value.code == 3
+
+
+def test_required_checks_from_env_rejects_bool_item():
+    os.environ["REQUIRED_CHECKS_JSON"] = json.dumps(
+        {"main": ["ctx-a", True]}
+    )
+    with pytest.raises(SystemExit) as exc_info:
+        drift.required_checks_from_env("main")
+    assert exc_info.value.code == 3
+
+
+def test_required_checks_from_env_rejects_null_item():
+    os.environ["REQUIRED_CHECKS_JSON"] = json.dumps(
+        {"main": ["ctx-a", None]}
+    )
+    with pytest.raises(SystemExit) as exc_info:
+        drift.required_checks_from_env("main")
+    assert exc_info.value.code == 3
+
+
+def test_required_checks_from_env_rejects_whitespace_only_item():
+    os.environ["REQUIRED_CHECKS_JSON"] = json.dumps(
+        {"main": ["ctx-a", "   "]}
+    )
+    with pytest.raises(SystemExit) as exc_info:
+        drift.required_checks_from_env("main")
+    assert exc_info.value.code == 3
+
+
+def test_required_checks_from_env_rejects_duplicate_item():
+    os.environ["REQUIRED_CHECKS_JSON"] = json.dumps(
+        {"main": ["ctx-a", "ctx-b", "ctx-a"]}
+    )
+    with pytest.raises(SystemExit) as exc_info:
+        drift.required_checks_from_env("main")
+    assert exc_info.value.code == 3

.gitea/scripts/tests/test_gitea_merge_queue.py

@@ -1,4 +1,5 @@
 import importlib.util
+import json
 import sys
 from pathlib import Path
 
@@ -1592,3 +1593,88 @@ def test_process_once_defensive_skip_when_pull_payload_opted_out(monkeypatch):
 
     assert rc == 0
     assert calls["merged"] is None
+
+
+# ---------------------------------------------------------------------------
+# required_checks_from_env — REQUIRED_CHECKS_JSON parsing (fail-closed)
+# ---------------------------------------------------------------------------
+
+def test_required_checks_from_env_parses_json():
+    import os
+    os.environ["REQUIRED_CHECKS_JSON"] = json.dumps(
+        {"main": ["ctx-a"], "staging": ["ctx-b"]}
+    )
+    assert mq.required_checks_from_env("main") == ["ctx-a"]
+    assert mq.required_checks_from_env("staging") == ["ctx-b"]
+
+
+def test_required_checks_from_env_rejects_non_string_item():
+    import os
+    os.environ["REQUIRED_CHECKS_JSON"] = json.dumps(
+        {"main": ["ctx-a", 123]}
+    )
+    with pytest.raises(SystemExit) as exc_info:
+        mq.required_checks_from_env("main")
+    assert exc_info.value.code == 2
+
+
+def test_required_checks_from_env_rejects_bool_item():
+    import os
+    os.environ["REQUIRED_CHECKS_JSON"] = json.dumps(
+        {"main": ["ctx-a", True]}
+    )
+    with pytest.raises(SystemExit) as exc_info:
+        mq.required_checks_from_env("main")
+    assert exc_info.value.code == 2
+
+
+def test_required_checks_from_env_rejects_null_item():
+    import os
+    os.environ["REQUIRED_CHECKS_JSON"] = json.dumps(
+        {"main": ["ctx-a", None]}
+    )
+    with pytest.raises(SystemExit) as exc_info:
+        mq.required_checks_from_env("main")
+    assert exc_info.value.code == 2
+
+
+def test_required_checks_from_env_rejects_whitespace_only_item():
+    import os
+    os.environ["REQUIRED_CHECKS_JSON"] = json.dumps(
+        {"main": ["ctx-a", "   "]}
+    )
+    with pytest.raises(SystemExit) as exc_info:
+        mq.required_checks_from_env("main")
+    assert exc_info.value.code == 2
+
+
+def test_process_once_holds_when_json_drifts_from_branch_protection(monkeypatch):
+    """REQUIRED_CHECKS_JSON missing a BP-required context → HOLD (no merge).
+
+    Regression for CR2 review on #2401: if JSON omits a context that BP
+    requires, the queue must not merge (force_merge could bypass it).
+    """
+    merged = {"called": False}
+    monkeypatch.setattr(mq, "WATCH_BRANCH", "main")
+    monkeypatch.setattr(
+        mq,
+        "get_branch_protection",
+        lambda branch: mq.BranchProtection(
+            required_contexts=[
+                "CI / all-required (pull_request)",
+                "qa-review / approved (pull_request)",
+            ],
+            required_approvals=2,
+            block_on_rejected_reviews=True,
+        ),
+    )
+    monkeypatch.setattr(
+        mq,
+        "required_checks_from_env",
+        lambda branch: ["CI / all-required (pull_request)"],
+    )
+    monkeypatch.setattr(mq, "merge_pull", lambda *a, **k: merged.__setitem__("called", True))
+
+    rc = mq.process_once(dry_run=False)
+    assert rc == 0
+    assert merged["called"] is False