.gitea/scripts/lint_mask_pr_atomicity.py

@@ -0,0 +1,361 @@
+#!/usr/bin/env python3
+"""lint_mask_pr_atomicity — Tier 2d structural enforcement per internal#350.
+
+Rule
+----
+A PR whose diff touches `.gitea/workflows/ci.yml` AND modifies EITHER:
+
+  - any `continue-on-error:` value, OR
+  - the `all-required` sentinel job's `needs:` block
+
+must EITHER:
+
+  - Touch BOTH atomically in the same PR (preferred), OR
+  - Cross-link the paired PR via a literal `Paired: #NNN` reference in
+    the PR body OR in any commit message between BASE_SHA and HEAD_SHA.
+
+The class this prevents
+-----------------------
+PR#665 (interim `continue-on-error: true` on `platform-build`) and
+PR#668 (sentinel-`needs` demotion of the same job) were designed as a
+pair but merged solo — #665 landed at 04:47Z 2026-05-12, #668 was still
+open at 05:07Z when the main-red watchdog (#674) fired. Result: ~20
+minutes of `main` red and a cascade of false-positives on unrelated PRs.
+
+The lint operates on the YAML AST (PyYAML), not grep, per
+`feedback_behavior_based_ast_gates`: a refactor that moves `continue-on-error`
+between job keys, or renames the `all-required` job, would still be
+detected because we walk the parsed structure.
+
+Why this works on Gitea 1.22.6
+------------------------------
+We don't use any 1.22.6-missing endpoints (no `/actions/runs/*`, no
+`branch_protections/*` — Tier 2f/g need those; Tier 2d does not). All
+required inputs come from the workflow `pull_request` event payload
+(BASE_SHA, HEAD_SHA, PR_BODY) and from local git via `git show`/`git log`.
+The auto-injected `GITHUB_TOKEN` is enough; we don't need
+DRIFT_BOT_TOKEN.
+
+Exit codes
+----------
+  0 — ci.yml not in diff, OR diff is no-op for the rule predicates,
+      OR atomicity satisfied (both touched), OR a valid `Paired: #NNN`
+      reference is present.
+  1 — exactly ONE of {coe, sentinel-needs} touched AND no valid
+      `Paired: #NNN` reference. The split-pair regression class.
+  2 — env contract violation (BASE_SHA / HEAD_SHA missing) or YAML
+      parse error on either side.
+
+Env
+---
+  BASE_SHA          — PR base (pull_request.base.sha)
+  HEAD_SHA          — PR head (pull_request.head.sha)
+  PR_BODY           — pull_request.body (may be empty)
+  CI_WORKFLOW_PATH  — defaults to `.gitea/workflows/ci.yml`
+  SENTINEL_JOB_KEY  — defaults to `all-required`
+
+Memory cross-links
+------------------
+  - internal#350 (the RFC that specs this lint)
+  - PR#665 / PR#668 (the empirical split-pair)
+  - mc#664 (the main-red incident)
+  - feedback_strict_root_only_after_class_a
+  - feedback_behavior_based_ast_gates
+"""
+from __future__ import annotations
+
+import os
+import re
+import subprocess
+import sys
+from typing import Any
+
+try:
+    import yaml
+except ImportError:
+    sys.stderr.write(
+        "::error::PyYAML is required. Install with: pip install PyYAML\n"
+    )
+    sys.exit(2)
+
+
+# ---------------------------------------------------------------------------
+# YAML quirk: bare `on:` at the top level becomes Python `True` because
+# `on` is a YAML 1.1 boolean. Not used here but documented for future
+# editors who copy from this module.
+# ---------------------------------------------------------------------------
+
+
+# `Paired: #NNN` reference. `#` is mandatory, NNN must be digits. Any
+# surrounding markdown/whitespace is fine. The match is case-sensitive
+# on `Paired:` because lower-case `paired:` collides with conversational
+# prose ("paired: see comment above") and the convention is the exact
+# capitalisation.
+PAIRED_RE = re.compile(r"\bPaired:\s*#(?P<num>\d+)\b")
+
+
+# ---------------------------------------------------------------------------
+# Env contract
+# ---------------------------------------------------------------------------
+def _env(key: str, default: str | None = None) -> str:
+    v = os.environ.get(key, default)
+    return v if v is not None else ""
+
+
+def _require_env(key: str) -> str:
+    v = os.environ.get(key)
+    if not v:
+        sys.stderr.write(f"::error::missing required env var: {key}\n")
+        sys.exit(2)
+    return v
+
+
+# ---------------------------------------------------------------------------
+# git-show helper. Returns None when the path doesn't exist on that side
+# (new file, deleted file, or rename — git returns exit 128 with "fatal:
+# path not in tree"). We treat None as "no rule predicate triggered on
+# that side".
+# ---------------------------------------------------------------------------
+def git_show(sha: str, path: str) -> str | None:
+    r = subprocess.run(
+        ["git", "show", f"{sha}:{path}"],
+        capture_output=True,
+        text=True,
+    )
+    if r.returncode != 0:
+        return None
+    return r.stdout
+
+
+def git_log_messages(base_sha: str, head_sha: str) -> str:
+    r = subprocess.run(
+        ["git", "log", "--format=%B", f"{base_sha}..{head_sha}"],
+        capture_output=True,
+        text=True,
+    )
+    if r.returncode != 0:
+        return ""
+    return r.stdout
+
+
+def git_diff_paths(base_sha: str, head_sha: str) -> list[str]:
+    r = subprocess.run(
+        ["git", "diff", "--name-only", f"{base_sha}..{head_sha}"],
+        capture_output=True,
+        text=True,
+    )
+    if r.returncode != 0:
+        return []
+    return [p for p in r.stdout.splitlines() if p.strip()]
+
+
+# ---------------------------------------------------------------------------
+# Predicate 1 — any `continue-on-error` value changed between base and head
+# ---------------------------------------------------------------------------
+def _collect_coe(doc: Any) -> dict[str, Any]:
+    """Walk every job in `jobs.*` and collect its continue-on-error value.
+
+    Returns a dict {job_key: coe_value}. Missing keys are absent from
+    the dict (NOT `False` — distinguishes "added the key" from
+    "unchanged absent"). Job-step `continue-on-error` is NOT considered
+    — only job-level, because that's the value that masks job status
+    rollup, which is the class this lint targets.
+    """
+    out: dict[str, Any] = {}
+    if not isinstance(doc, dict):
+        return out
+    jobs = doc.get("jobs")
+    if not isinstance(jobs, dict):
+        return out
+    for k, j in jobs.items():
+        if not isinstance(j, dict):
+            continue
+        if "continue-on-error" in j:
+            out[k] = j["continue-on-error"]
+    return out
+
+
+def coe_changed(base_doc: Any, head_doc: Any) -> tuple[bool, list[str]]:
+    """Return (changed?, [reasons]) describing per-job coe diffs."""
+    base = _collect_coe(base_doc)
+    head = _collect_coe(head_doc)
+    reasons: list[str] = []
+    all_keys = set(base) | set(head)
+    for k in sorted(all_keys):
+        b = base.get(k, "<absent>")
+        h = head.get(k, "<absent>")
+        if b != h:
+            reasons.append(f"job '{k}' continue-on-error: {b!r} → {h!r}")
+    return (bool(reasons), reasons)
+
+
+# ---------------------------------------------------------------------------
+# Predicate 2 — sentinel job's `needs:` changed
+# ---------------------------------------------------------------------------
+def _collect_needs(doc: Any, sentinel_key: str) -> list[str] | None:
+    """Return the sentinel job's needs list (sorted) or None if absent."""
+    if not isinstance(doc, dict):
+        return None
+    jobs = doc.get("jobs")
+    if not isinstance(jobs, dict):
+        return None
+    j = jobs.get(sentinel_key)
+    if not isinstance(j, dict):
+        return None
+    needs = j.get("needs")
+    if needs is None:
+        return []
+    if isinstance(needs, str):
+        return [needs]
+    if isinstance(needs, list):
+        # Sort because `needs:` is order-insensitive at the engine
+        # level; a reorder is not a semantic change and shouldn't
+        # trip the lint.
+        return sorted(str(x) for x in needs)
+    return None
+
+
+def sentinel_needs_changed(
+    base_doc: Any, head_doc: Any, sentinel_key: str
+) -> tuple[bool, str]:
+    """Return (changed?, reason)."""
+    base = _collect_needs(base_doc, sentinel_key)
+    head = _collect_needs(head_doc, sentinel_key)
+    if base == head:
+        return (False, "")
+    return (
+        True,
+        f"sentinel '{sentinel_key}'.needs: {base!r} → {head!r}",
+    )
+
+
+# ---------------------------------------------------------------------------
+# Predicate 3 — `Paired: #NNN` present in body or any commit message
+# ---------------------------------------------------------------------------
+def find_paired_refs(pr_body: str, commit_log: str) -> list[str]:
+    """Return list of `#NNN` strings found (deduped, sorted)."""
+    found: set[str] = set()
+    for src in (pr_body, commit_log):
+        for m in PAIRED_RE.finditer(src or ""):
+            found.add(m.group("num"))
+    return sorted(found)
+
+
+# ---------------------------------------------------------------------------
+# Driver
+# ---------------------------------------------------------------------------
+def _parse(content: str | None, label: str) -> Any:
+    if content is None:
+        return None
+    try:
+        return yaml.safe_load(content)
+    except yaml.YAMLError as e:
+        sys.stderr.write(f"::error::YAML parse error on {label}: {e}\n")
+        sys.exit(2)
+
+
+def run() -> int:
+    base_sha = _require_env("BASE_SHA")
+    head_sha = _require_env("HEAD_SHA")
+    pr_body = _env("PR_BODY", "")
+    ci_path = _env("CI_WORKFLOW_PATH", ".gitea/workflows/ci.yml")
+    sentinel_key = _env("SENTINEL_JOB_KEY", "all-required")
+
+    # Step 0 — is ci.yml even in the diff? If not, the lint doesn't apply.
+    changed_paths = git_diff_paths(base_sha, head_sha)
+    if ci_path not in changed_paths:
+        print(
+            f"::notice::{ci_path} not in PR diff; lint-mask-pr-atomicity "
+            f"skipped (no atomicity risk)."
+        )
+        return 0
+
+    base_yml = git_show(base_sha, ci_path)
+    head_yml = git_show(head_sha, ci_path)
+
+    base_doc = _parse(base_yml, f"{ci_path}@{base_sha}")
+    head_doc = _parse(head_yml, f"{ci_path}@{head_sha}")
+
+    # If the file is newly added (no base), no flip is possible — every
+    # value is "newly introduced", not "changed". Tier 2e covers the
+    # tracking-issue check for new continue-on-error: true. Exit 0.
+    if base_doc is None:
+        print(
+            f"::notice::{ci_path} newly added in this PR; no flip to "
+            f"analyse — lint-mask-pr-atomicity skipped."
+        )
+        return 0
+
+    # If the file is deleted on head, ditto — no atomicity question.
+    if head_doc is None:
+        print(
+            f"::notice::{ci_path} deleted in this PR; "
+            f"lint-mask-pr-atomicity skipped."
+        )
+        return 0
+
+    coe_yes, coe_reasons = coe_changed(base_doc, head_doc)
+    needs_yes, needs_reason = sentinel_needs_changed(
+        base_doc, head_doc, sentinel_key
+    )
+
+    if not coe_yes and not needs_yes:
+        print(
+            f"::notice::{ci_path} touched but neither continue-on-error "
+            f"nor sentinel '{sentinel_key}'.needs changed — no atomicity "
+            f"risk. OK."
+        )
+        return 0
+
+    if coe_yes and needs_yes:
+        print(
+            f"::notice::Atomic change detected: both continue-on-error "
+            f"AND sentinel '{sentinel_key}'.needs touched in same PR. OK."
+        )
+        for r in coe_reasons:
+            print(f"  - {r}")
+        print(f"  - {needs_reason}")
+        return 0
+
+    # Exactly one side touched — require Paired: #NNN reference.
+    commit_log = git_log_messages(base_sha, head_sha)
+    paired = find_paired_refs(pr_body, commit_log)
+
+    one_side = "continue-on-error" if coe_yes else f"sentinel '{sentinel_key}'.needs"
+    other_side = (
+        f"sentinel '{sentinel_key}'.needs" if coe_yes else "continue-on-error"
+    )
+
+    if paired:
+        print(
+            f"::notice::Split-pair detected ({one_side} changed without "
+            f"{other_side}), but Paired reference(s) present: "
+            f"{', '.join('#' + n for n in paired)}. OK."
+        )
+        for r in coe_reasons:
+            print(f"  - {r}")
+        if needs_reason:
+            print(f"  - {needs_reason}")
+        return 0
+
+    # The failure mode this lint exists to prevent.
+    print(
+        f"::error file={ci_path}::lint-mask-pr-atomicity (Tier 2d): "
+        f"PR touches {one_side} in {ci_path} but NOT {other_side}, "
+        f"and no `Paired: #NNN` reference was found in the PR body or "
+        f"in commit messages between {base_sha[:8]}..{head_sha[:8]}. "
+        f"This is the PR#665+#668 split-pair regression class "
+        f"(see internal#350, mc#664). FIX: either (a) include the "
+        f"matching {other_side} change in the same PR (preferred), or "
+        f"(b) add `Paired: #NNN` (literal, capital P, with `#`) to the "
+        f"PR body or a commit message referencing the paired PR."
+    )
+    for r in coe_reasons:
+        print(f"  - {r}")
+    if needs_reason:
+        print(f"  - {needs_reason}")
+    return 1
+
+
+if __name__ == "__main__":
+    sys.exit(run())

.gitea/workflows/lint-mask-pr-atomicity.yml

@@ -0,0 +1,132 @@
+name: lint-mask-pr-atomicity
+
+# Tier 2d hard-gate lint (per internal#350) — blocks PRs that touch
+# `.gitea/workflows/ci.yml` and modify ONLY ONE of {continue-on-error,
+# all-required.sentinel.needs} without a `Paired: #NNN` reference in
+# the PR body or in a commit message.
+#
+# Why this exists
+# ---------------
+# PR#665 (interim `continue-on-error: true` on `platform-build`) and
+# PR#668 (sentinel-`needs` demotion of the same job) were designed as a
+# pair but merged solo — #665 landed at 04:47Z 2026-05-12, #668 was
+# still open at 05:07Z when the main-red watchdog (#674) fired. Result:
+# ~20 minutes of `main` red and a cascade of false-positives on
+# unrelated PRs. This lint structurally prevents that class.
+#
+# How the gate works
+# ------------------
+# 1. The workflow runs on every PR whose diff touches ci.yml (paths
+#    filter). It is NOT a required check on `main` because the rule is
+#    diff-based — running it on PRs that don't touch ci.yml would
+#    produce a `pending` status forever (per
+#    `feedback_path_filtered_workflow_cant_be_required`).
+# 2. The script reads `BASE_SHA:ci.yml` and `HEAD_SHA:ci.yml`, parses
+#    both via PyYAML AST (per `feedback_behavior_based_ast_gates` — no
+#    grep, no regex on the raw text — so a YAML-shape refactor still
+#    detects).
+# 3. Walks `jobs.*.continue-on-error` on each side; flags any value
+#    diff. Reads `jobs.all-required.needs` on each side; flags any
+#    set diff (order-insensitive — `needs:` is engine-unordered).
+# 4. If both predicates fired → atomic, OK. If neither → no risk, OK.
+#    If exactly one fired → require `Paired: #NNN` in PR body OR in
+#    any commit message between base..head; else fail.
+#
+# Phase contract (RFC internal#219 §1 ladder)
+# -------------------------------------------
+# This workflow lands at `continue-on-error: true` (Phase 3 — surface
+# regressions without blocking PRs while the rule beds in).
+# Follow-up PR flips to `false` once we have ≥3 days of clean runs on
+# `main` and no false-positives. Tracking issue: internal#350.
+#
+# Cross-links
+# -----------
+# - internal#350 (the RFC that specs this lint)
+# - PR#665 / PR#668 (the empirical split-pair)
+# - mc#664 (the main-red incident the split caused)
+# - feedback_strict_root_only_after_class_a
+# - feedback_behavior_based_ast_gates
+#
+# Auth: only needs the auto-injected GITHUB_TOKEN (read-only, repo
+# scope). No DRIFT_BOT_TOKEN needed — Tier 2d does NOT call
+# branch_protections (Tier 2g/f do).
+
+on:
+  pull_request:
+    types: [opened, synchronize, reopened, edited]
+    # `edited` is included because the rule depends on PR_BODY: a user
+    # may add `Paired: #NNN` after first push to satisfy the lint. The
+    # rerun on `edited` lets the PR turn green without an empty
+    # commit. Gitea 1.22.6 fires `edited` on body changes — verified
+    # via gitea-source/models/issues/pull_list.go::triggerNewPRWebhook.
+    paths:
+      - '.gitea/workflows/ci.yml'
+      - '.gitea/scripts/lint_mask_pr_atomicity.py'
+      - '.gitea/workflows/lint-mask-pr-atomicity.yml'
+      - 'tests/test_lint_mask_pr_atomicity.py'
+
+env:
+  # Belt-and-suspenders against the runner-default trap
+  # (feedback_act_runner_github_server_url). Runners are configured
+  # with this env via /opt/molecule/runners/config.yaml, but pinning
+  # at the workflow level protects against a runner regenerated
+  # without the config file.
+  GITHUB_SERVER_URL: https://git.moleculesai.app
+
+permissions:
+  contents: read
+  pull-requests: read
+
+# Per-PR concurrency — re-pushes cancel previous runs to keep the
+# queue short. The lint is cheap (one git show + log + a YAML parse).
+concurrency:
+  group: lint-mask-pr-atomicity-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  scan:
+    name: lint-mask-pr-atomicity
+    runs-on: ubuntu-latest
+    timeout-minutes: 5
+    # Phase 3 (RFC #219 §1): surface broken shapes without blocking
+    # PRs. Follow-up PR flips this to `false` once recent runs on main
+    # are confirmed clean (eat-our-own-dogfood discipline mirrors
+    # PR#673's same-shape comment). Tracking: internal#350.
+    continue-on-error: true
+    steps:
+      - name: Check out PR head with full history (need base SHA blobs)
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
+        with:
+          # `git show <base-sha>:<path>` needs the base SHA's blobs.
+          # Shallow=1 would miss it. Same rationale as PR#673 and
+          # check-migration-collisions.yml.
+          fetch-depth: 0
+      - name: Set up Python (PyYAML for AST parsing)
+        uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065  # v5.6.0
+        with:
+          python-version: '3.12'
+      - name: Install PyYAML
+        # Same pin as ci-required-drift.yml + the rest of the Tier 2
+        # lint family — keep runner-cache hits uniform.
+        run: python -m pip install --quiet 'PyYAML==6.0.2'
+      - name: Ensure base ref is reachable locally
+        # fetch-depth=0 usually pulls the base too, but explicit-fetch
+        # is cheap insurance against runner-version drift (matches the
+        # comment in check-migration-collisions.yml and PR#673).
+        run: |
+          git fetch origin "${{ github.event.pull_request.base.ref }}" || true
+      - name: Run lint-mask-pr-atomicity
+        env:
+          BASE_SHA: ${{ github.event.pull_request.base.sha }}
+          HEAD_SHA: ${{ github.event.pull_request.head.sha }}
+          # PR body — the script greps for `Paired: #NNN`.
+          PR_BODY: ${{ github.event.pull_request.body }}
+          CI_WORKFLOW_PATH: .gitea/workflows/ci.yml
+          SENTINEL_JOB_KEY: all-required
+        run: python3 .gitea/scripts/lint_mask_pr_atomicity.py
+      - name: Run lint-mask-pr-atomicity unit tests
+        # Run the test suite in-CI so the lint's own behaviour is
+        # verified on every change. Matches lint-workflow-yaml.yml.
+        run: |
+          python -m pip install --quiet pytest
+          python3 -m pytest tests/test_lint_mask_pr_atomicity.py -v

tests/test_lint_mask_pr_atomicity.py

@@ -0,0 +1,357 @@
+"""Tests for `.gitea/scripts/lint_mask_pr_atomicity.py` — Tier 2d lint.
+
+Structural enforcement of internal#350 Tier 2d: a PR that touches
+`.gitea/workflows/ci.yml` and modifies `continue-on-error` OR the
+`all-required` sentinel's `needs:` block must EITHER:
+
+  - Touch both atomically in the same PR (preferred), OR
+  - Cross-link to the paired PR via `Paired: #NNN` in body OR a commit
+    message.
+
+The class this lint exists to prevent: PR#665 (interim
+continue-on-error: true on platform-build) + PR#668 (sentinel-exempt)
+were designed-as-a-pair but merged solo — #665 landed at 04:47Z, #668
+still open at 05:07Z when the watchdog fired. ~20 min of main red.
+
+Test classes (per `feedback_branch_count_before_approving`, every
+prod branch enumerated):
+
+  - test_diff_touches_neither_passes              — diff is in ci.yml
+    but neither continue-on-error nor all-required.needs is touched.
+    PR is exempt. Exit 0.
+  - test_diff_touches_both_atomically_passes      — both touched in
+    the same PR. Atomic. Exit 0.
+  - test_diff_touches_coe_only_no_pair_fails      — continue-on-error
+    flipped without sentinel-needs change AND no `Paired: #NNN`
+    reference anywhere. Exit 1.
+  - test_diff_touches_needs_only_no_pair_fails    — sentinel `needs:`
+    changed without `continue-on-error` change AND no pair reference.
+    Exit 1.
+  - test_diff_touches_coe_only_pair_in_body       — coe changed, no
+    needs change, body has `Paired: #668`. Exit 0.
+  - test_diff_touches_needs_only_pair_in_commit   — needs changed, no
+    coe change, commit message includes `Paired: #665`. Exit 0.
+  - test_paired_reference_must_be_numeric         — `Paired: #abc` or
+    `Paired: NNNN` (missing `#`) doesn't satisfy the rule. Exit 1.
+  - test_ci_yml_unchanged_skips                   — no ci.yml in the
+    diff at all (defensive — workflow paths-filter already prevents,
+    but the lint should not crash). Exit 0.
+
+The lint receives base SHA + head SHA via env (set by the workflow
+from the pull_request payload) and uses `git show` to read both
+sides without a separate clone. Tests stub `subprocess.run` to drive
+the diff content; the actual git is never invoked.
+
+Run:
+    python3 -m pytest tests/test_lint_mask_pr_atomicity.py -v
+
+Dependencies: stdlib + PyYAML (the script reads ci.yml via PyYAML AST
+per `feedback_behavior_based_ast_gates`). No network. No live git.
+"""
+from __future__ import annotations
+
+import importlib.util
+import os
+import subprocess
+import sys
+import textwrap
+from pathlib import Path
+from unittest import mock
+
+import pytest
+
+
+SCRIPT_PATH = (
+    Path(__file__).resolve().parent.parent
+    / ".gitea"
+    / "scripts"
+    / "lint_mask_pr_atomicity.py"
+)
+
+
+# Minimal ci.yml fixture — only the bits the lint actually parses
+# (a job with continue-on-error + the all-required aggregator).
+CI_YML_BASE = """
+name: CI
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
+jobs:
+  platform-build:
+    runs-on: ubuntu-latest
+    continue-on-error: false
+    steps:
+      - run: echo build
+  canvas-build:
+    runs-on: ubuntu-latest
+    continue-on-error: false
+    steps:
+      - run: echo build
+  all-required:
+    runs-on: ubuntu-latest
+    needs:
+      - platform-build
+      - canvas-build
+    if: always()
+    steps:
+      - run: echo agg
+"""
+
+# Same as base but with continue-on-error flipped on platform-build.
+CI_YML_COE_FLIPPED = CI_YML_BASE.replace(
+    "  platform-build:\n    runs-on: ubuntu-latest\n    continue-on-error: false",
+    "  platform-build:\n    runs-on: ubuntu-latest\n    continue-on-error: true",
+)
+
+# Same as base but with canvas-build dropped from all-required.needs.
+CI_YML_NEEDS_CHANGED = CI_YML_BASE.replace(
+    "    needs:\n      - platform-build\n      - canvas-build",
+    "    needs:\n      - platform-build",
+)
+
+# Both changed at once.
+CI_YML_BOTH = CI_YML_COE_FLIPPED.replace(
+    "    needs:\n      - platform-build\n      - canvas-build",
+    "    needs:\n      - platform-build",
+)
+
+
+def _import_lint(monkeypatch):
+    """Import the lint module under a fresh name per test."""
+    spec = importlib.util.spec_from_file_location(
+        f"lint_mask_pr_atomicity_{os.getpid()}_{id(monkeypatch)}",
+        SCRIPT_PATH,
+    )
+    m = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(m)
+    return m
+
+
+def _stub_git(base_yml: str | None, head_yml: str | None, commits: list[str]):
+    """Build a fake `subprocess.run` that emulates git show + log.
+
+    base_yml / head_yml: contents the lint sees at base/head SHA.
+        Pass `None` to simulate "path didn't exist on that side" (git
+        show returns exit code 128 — file-not-in-tree).
+    commits: list of commit messages on the PR (head's ancestry up to
+        the base merge-base). The lint runs
+        `git log --format=%B base..head` to find Paired: refs.
+    """
+
+    def fake_run(cmd, *args, **kwargs):
+        if not isinstance(cmd, list):
+            raise AssertionError(f"unexpected non-list cmd: {cmd!r}")
+        # `git show <sha>:<path>`
+        if cmd[:2] == ["git", "show"] and len(cmd) >= 3 and ":" in cmd[2]:
+            sha, path = cmd[2].split(":", 1)
+            if "base" in sha or "BASE" in sha:
+                content = base_yml
+            else:
+                content = head_yml
+            if content is None:
+                return subprocess.CompletedProcess(
+                    cmd, returncode=128, stdout="", stderr="fatal: path not in tree"
+                )
+            return subprocess.CompletedProcess(
+                cmd, returncode=0, stdout=content, stderr=""
+            )
+        # `git log --format=%B base..head -- .`
+        if cmd[:2] == ["git", "log"]:
+            body = "\n\n--commit-boundary--\n\n".join(commits)
+            return subprocess.CompletedProcess(
+                cmd, returncode=0, stdout=body, stderr=""
+            )
+        # `git diff --name-only base..head`
+        if cmd[:2] == ["git", "diff"]:
+            # If either side had ci.yml, it's in the diff; else not.
+            paths = []
+            if (base_yml or "") != (head_yml or ""):
+                paths.append(".gitea/workflows/ci.yml")
+            return subprocess.CompletedProcess(
+                cmd, returncode=0, stdout="\n".join(paths) + "\n", stderr=""
+            )
+        raise AssertionError(f"unexpected git invocation: {cmd!r}")
+
+    return fake_run
+
+
+@pytest.fixture()
+def env(monkeypatch):
+    monkeypatch.setenv("BASE_SHA", "base-sha-1")
+    monkeypatch.setenv("HEAD_SHA", "head-sha-1")
+    monkeypatch.setenv("PR_BODY", "")
+    monkeypatch.setenv("CI_WORKFLOW_PATH", ".gitea/workflows/ci.yml")
+    monkeypatch.setenv("SENTINEL_JOB_KEY", "all-required")
+    return monkeypatch
+
+
+# ---------------------------------------------------------------------------
+# Diff in ci.yml but neither rule predicate triggered → pass
+# ---------------------------------------------------------------------------
+def test_diff_touches_neither_passes(env, monkeypatch, capsys):
+    # Add a comment-only change (no coe flip, no needs change).
+    base = CI_YML_BASE
+    head = "# a harmless comment\n" + CI_YML_BASE
+    monkeypatch.setattr(
+        subprocess, "run", _stub_git(base, head, ["chore: comment"])
+    )
+    m = _import_lint(monkeypatch)
+    rc = m.run()
+    assert rc == 0
+    out = capsys.readouterr().out
+    assert "no atomicity risk" in out.lower() or "ok" in out.lower()
+
+
+# ---------------------------------------------------------------------------
+# Diff touches BOTH coe and sentinel.needs in the same PR → atomic, pass
+# ---------------------------------------------------------------------------
+def test_diff_touches_both_atomically_passes(env, monkeypatch, capsys):
+    monkeypatch.setattr(
+        subprocess,
+        "run",
+        _stub_git(CI_YML_BASE, CI_YML_BOTH, ["fix(ci): atomic flip"]),
+    )
+    m = _import_lint(monkeypatch)
+    rc = m.run()
+    assert rc == 0
+    out = capsys.readouterr().out
+    assert "atomic" in out.lower()
+
+
+# ---------------------------------------------------------------------------
+# Diff touches ONLY continue-on-error, no pair reference → fail
+# ---------------------------------------------------------------------------
+def test_diff_touches_coe_only_no_pair_fails(env, monkeypatch, capsys):
+    monkeypatch.setattr(
+        subprocess,
+        "run",
+        _stub_git(
+            CI_YML_BASE,
+            CI_YML_COE_FLIPPED,
+            ["fix(ci): flip coe on platform-build"],
+        ),
+    )
+    m = _import_lint(monkeypatch)
+    rc = m.run()
+    assert rc == 1
+    out = capsys.readouterr().out
+    assert "paired" in out.lower() or "atomicity" in out.lower()
+    # Actionable failure: must name what is missing.
+    assert "continue-on-error" in out.lower()
+
+
+# ---------------------------------------------------------------------------
+# Diff touches ONLY sentinel.needs, no pair reference → fail
+# ---------------------------------------------------------------------------
+def test_diff_touches_needs_only_no_pair_fails(env, monkeypatch, capsys):
+    monkeypatch.setattr(
+        subprocess,
+        "run",
+        _stub_git(
+            CI_YML_BASE,
+            CI_YML_NEEDS_CHANGED,
+            ["fix(ci): drop canvas-build from sentinel"],
+        ),
+    )
+    m = _import_lint(monkeypatch)
+    rc = m.run()
+    assert rc == 1
+    out = capsys.readouterr().out
+    assert "paired" in out.lower() or "atomicity" in out.lower()
+    assert "needs" in out.lower() or "sentinel" in out.lower()
+
+
+# ---------------------------------------------------------------------------
+# COE-only flip with `Paired: #668` in PR body → pass
+# ---------------------------------------------------------------------------
+def test_diff_touches_coe_only_pair_in_body(env, monkeypatch, capsys):
+    monkeypatch.setenv("PR_BODY", "Interim coe flip. Paired: #668")
+    monkeypatch.setattr(
+        subprocess,
+        "run",
+        _stub_git(
+            CI_YML_BASE,
+            CI_YML_COE_FLIPPED,
+            ["fix(ci): flip coe on platform-build"],
+        ),
+    )
+    m = _import_lint(monkeypatch)
+    rc = m.run()
+    assert rc == 0
+    out = capsys.readouterr().out
+    assert "paired" in out.lower()
+    assert "668" in out
+
+
+# ---------------------------------------------------------------------------
+# Needs-only flip with `Paired: #665` in a commit message → pass
+# ---------------------------------------------------------------------------
+def test_diff_touches_needs_only_pair_in_commit(env, monkeypatch, capsys):
+    monkeypatch.setattr(
+        subprocess,
+        "run",
+        _stub_git(
+            CI_YML_BASE,
+            CI_YML_NEEDS_CHANGED,
+            [
+                "fix(ci): drop canvas-build from sentinel\n\nPaired: #665",
+            ],
+        ),
+    )
+    m = _import_lint(monkeypatch)
+    rc = m.run()
+    assert rc == 0
+    out = capsys.readouterr().out
+    assert "paired" in out.lower()
+    assert "665" in out
+
+
+# ---------------------------------------------------------------------------
+# `Paired: #abc` is not a valid issue/PR ref — fail
+# ---------------------------------------------------------------------------
+def test_paired_reference_must_be_numeric(env, monkeypatch, capsys):
+    monkeypatch.setenv("PR_BODY", "Paired: #abc")
+    monkeypatch.setattr(
+        subprocess,
+        "run",
+        _stub_git(
+            CI_YML_BASE,
+            CI_YML_COE_FLIPPED,
+            ["fix(ci): flip coe"],
+        ),
+    )
+    m = _import_lint(monkeypatch)
+    rc = m.run()
+    assert rc == 1
+
+
+# ---------------------------------------------------------------------------
+# Defensive: ci.yml not in diff at all → skip cleanly
+# ---------------------------------------------------------------------------
+def test_ci_yml_unchanged_skips(env, monkeypatch, capsys):
+    monkeypatch.setattr(
+        subprocess, "run", _stub_git(CI_YML_BASE, CI_YML_BASE, ["chore: noop"])
+    )
+    m = _import_lint(monkeypatch)
+    rc = m.run()
+    assert rc == 0
+    out = capsys.readouterr().out
+    assert "ci.yml" in out.lower() or "not in" in out.lower() or "skip" in out.lower()
+
+
+# ---------------------------------------------------------------------------
+# Cross-cutting: file ADDED on head side (no base) — coe inferred as
+# "newly added with coe=true". Should NOT trigger the lint (it's a new
+# file, not a flip — Tier 2e covers tracking-issue for new coe=true).
+# ---------------------------------------------------------------------------
+def test_ci_yml_newly_added_passes(env, monkeypatch, capsys):
+    monkeypatch.setattr(
+        subprocess,
+        "run",
+        _stub_git(None, CI_YML_COE_FLIPPED, ["feat(ci): add ci.yml"]),
+    )
+    m = _import_lint(monkeypatch)
+    rc = m.run()
+    assert rc == 0