CI D26: bind COMTRADE canonical native shadow run to exact PR identity

docs/audits/CI_P3J_D26_NATIVE_PR_RUN_IDENTITY.md

@@ -0,0 +1,19 @@
+# D26 CI-P3J: exact PR identity for canonical native shadow comparison
+
+Related: [CI cost #542](https://github.com/masarray/arsas/issues/542). No release or deduplication authority.
+
+## Root cause
+
+The advisory COMTRADE native comparator chose canonical Build ARSAS runs by head SHA and head branch only. Multiple PRs may legitimately share the head branch and SHA while targeting different bases, so the newest unrelated synthetic merge run may shadow the correct run. In ARSAS, RC staging PR #727 and main promotion PR #729 both use the same RC source branch. GitHub's `pull_requests[].head.sha` may be updated for old runs: never trust it as immutable provenance.
+
+## D26 fix (isolated RC draft)
+
+Read the exact triggering pull request from the existing GitHub Actions `GITHUB_EVENT_PATH`, cross-check its positive PR number and head SHA/ref against CLI inputs, then require a matching `pull_requests[].number` on candidate canonical runs, in addition to the exact head SHA/branch/event/build workflow. No branch-only fallback. Bound local event JSON to 2 MiB and fail closed on mismatches or missing PR links.
+
+The comparator still verifies exact synthetic merge SHA, engine lock, ArdIrec lock, archive digest, test inventory and native bridge digest. `independentWindowsCTestRetained=true`, `deduplicationAuthorized=false`, and `releasePromotionAuthority=false` remain unchanged. PR identity is included in advisory result metadata.
+
+Five additional offline unittest methods cover cross-PR SHA collision, missing PR association, mismatched event identity, wrong PR online and explicit output provenance. The existing CI P0 workflow already executes the shadow-test script. This patch modifies only two existing Python files plus this note, with no workflow, native build, WPF or engine edits.
+
+## Acceptance and rollback
+
+Do not equate this change with actual CI savings. CI P0 tests, canonical Windows checks, and COMTRADE native acceptance must pass at exact source before owner-approved integration. No changes to parallel PR #757, #759, safety #753-756, Write #726/#716, SG7 #688, GUI #732 or protected RC. Revert only the two script changes if the new PR association contract proves unavailable from the GitHub API; never weaken independent Windows CMake/CTest to accommodate it.

scripts/compare-ci-native-shadow.py

@@ -75,7 +75,11 @@ def get(self, url: str, *, binary: bool = False):
             raise ShadowError(f"Read-only GitHub shadow evidence request failed: {exc}") from exc
 
 
-def choose_exact_run(payload: dict, *, head_sha: str, head_branch: str) -> dict | None:
+def choose_exact_run(
+    payload: dict, *, head_sha: str, head_branch: str, pr_number: int
+) -> dict | None:
+    if type(pr_number) is not int or pr_number <= 0:
+        raise ShadowError("An exact positive PR number is required for canonical run identity")
     runs = payload.get("workflow_runs")
     if not isinstance(runs, list):
         raise ShadowError("Canonical run list is malformed")
@@ -85,6 +89,11 @@ def choose_exact_run(payload: dict, *, head_sha: str, head_branch: str) -> dict
         and run.get("event") == "pull_request"
         and run.get("head_sha") == head_sha
         and run.get("head_branch") == head_branch
+        # Shared branches/commits can feed multiple PRs with distinct merge trees.
+        # The run's pull_requests[].head.sha is mutable; use its PR number only.
+        and isinstance(run.get("pull_requests"), list)
+        and any(type(link.get("number")) is int and link["number"] == pr_number
+                for link in run["pull_requests"] if isinstance(link, dict))
         and isinstance(run.get("id"), int)
         and isinstance(run.get("run_attempt"), int)
     ]
@@ -174,7 +183,7 @@ def compare_proofs(
 
 
 def compare_online(
-    api, *, head_sha: str, head_branch: str, source_sha: str,
+    api, *, head_sha: str, head_branch: str, pr_number: int, source_sha: str,
     engine_sha: str, ardirec_sha: str, independent_junit: Path,
     independent_bridge: Path, wait_seconds: int = 600, poll_seconds: int = 15,
     require_completed: bool = False,
@@ -185,6 +194,8 @@ def compare_online(
             raise ShadowError(f"Invalid exact {label} SHA")
     if not head_branch or len(head_branch) > 180 or "\n" in head_branch:
         raise ShadowError("Invalid exact PR branch")
+    if type(pr_number) is not int or pr_number <= 0:
+        raise ShadowError("Invalid exact PR number")
     if wait_seconds < 0 or wait_seconds > 1200 or poll_seconds < 1:
         raise ShadowError("Invalid bounded shadow wait")
     base = "https://api.github.com/repos/" + REPO
@@ -196,7 +207,7 @@ def compare_online(
     while True:
         run = choose_exact_run(
             api.get(f"{base}/actions/workflows/build.yml/runs?{query}"),
-            head_sha=head_sha, head_branch=head_branch,
+            head_sha=head_sha, head_branch=head_branch, pr_number=pr_number,
         )
         if run is not None:
             status, conclusion = run.get("status"), run.get("conclusion")
@@ -225,6 +236,7 @@ def compare_online(
                             run_id=run["id"], run_attempt=run["run_attempt"],
                             event="pull_request",
                         )
+                        result["canonicalPrNumber"] = pr_number
                         result["canonicalRunStatus"] = status
                         result["canonicalRunConclusion"] = conclusion
                         result["fullCanonicalBuildPassed"] = (
@@ -238,6 +250,27 @@ def compare_online(
         time.sleep(min(poll_seconds, deadline - time.monotonic()))
 
 
+def read_event_pr_number(path: Path, *, head_sha: str, head_branch: str) -> int:
+    """Bind CI to the exact PR event; never infer identity from shared branches."""
+    try:
+        if path.stat().st_size > 2 * 1024 * 1024:
+            raise ShadowError("GitHub pull-request event payload is oversized")
+        event = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, ValueError, UnicodeError) as exc:
+        raise ShadowError("Unable to read exact GitHub pull-request event") from exc
+    if not isinstance(event, dict):
+        raise ShadowError("GitHub pull-request event is not an object")
+    number, pr = event.get("number"), event.get("pull_request")
+    if (type(number) is not int or number <= 0 or
+            not isinstance(pr, dict) or pr.get("number") != number):
+        raise ShadowError("GitHub event lacks an unambiguous PR number")
+    head = pr.get("head")
+    if (not isinstance(head, dict) or head.get("sha") != head_sha or
+            head.get("ref") != head_branch):
+        raise ShadowError("GitHub PR event head SHA/branch differs from CI arguments")
+    return number
+
+
 def main() -> int:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--head-sha", required=True)
@@ -251,9 +284,15 @@ def main() -> int:
     parser.add_argument("--require-completed-canonical", action="store_true")
     parser.add_argument("--output", type=Path, required=True)
     args = parser.parse_args()
+    event_path = os.environ.get("GITHUB_EVENT_PATH")
+    if not event_path:
+        raise ShadowError("GITHUB_EVENT_PATH is required for exact PR identity")
+    pr_number = read_event_pr_number(
+        Path(event_path), head_sha=args.head_sha, head_branch=args.head_branch,
+    )
     result = compare_online(
         GitHubReadOnly(os.environ.get("GITHUB_TOKEN", "")),
-        head_sha=args.head_sha, head_branch=args.head_branch,
+        head_sha=args.head_sha, head_branch=args.head_branch, pr_number=pr_number,
         source_sha=args.source_sha, engine_sha=args.engine_sha,
         ardirec_sha=args.ardirec_sha,
         independent_junit=args.independent_junit,

scripts/test-ci-native-shadow.py

@@ -27,6 +27,7 @@
 
 HEAD, MERGE, ENGINE, ARDIREC = "d" * 40, "a" * 40, "b" * 40, "c" * 40
 BRANCH = "ci/p3j-c-native-shadow"
+PR = 759
 RUN = 1234
 XML = b'<testsuite tests="2" failures="0" errors="0" skipped="0"><testcase name="ABI"/><testcase name="Locus"/></testsuite>'
 
@@ -49,13 +50,15 @@ def artifact_bytes(*, source=MERGE, bridge=b"canonical-tested-binary", names=XML
 
 class FakeApi:
     def __init__(self, *, blob, digest, run_status="completed", run_conclusion="success",
-                 head=HEAD, source_branch=BRANCH, artifact=True, duplicate=False):
+                 head=HEAD, source_branch=BRANCH, artifact=True, duplicate=False,
+                 pr_number=PR):
         self.blob = blob
         self.digest = digest
         self.run_status = run_status
         self.run_conclusion = run_conclusion
         self.head = head
         self.branch = source_branch
+        self.pr_number = pr_number
         self.artifact = artifact
         self.duplicate = duplicate
 
@@ -66,6 +69,7 @@ def get(self, url: str, *, binary=False):
             return {"workflow_runs": [{
                 "id": RUN, "run_attempt": 1, "name": "Build ARSAS",
                 "head_sha": self.head, "head_branch": self.branch,
+                "pull_requests": [{"number": self.pr_number}],
                 "event": "pull_request", "status": self.run_status,
                 "conclusion": self.run_conclusion,
             }]}
@@ -95,7 +99,7 @@ def setUp(self):
         self.bridge.write_bytes(b"canonical-tested-binary")
         self.blob, self.digest = artifact_bytes()
         self.kwargs = {
-            "head_sha": HEAD, "head_branch": BRANCH,
+            "head_sha": HEAD, "head_branch": BRANCH, "pr_number": PR,
             "source_sha": MERGE, "engine_sha": ENGINE, "ardirec_sha": ARDIREC,
             "independent_junit": self.local, "independent_bridge": self.bridge,
             "wait_seconds": 0,
@@ -251,6 +255,76 @@ def test_extra_file_or_duplicate_artifact_is_rejected(self):
                 **self.kwargs,
             )
 
+    def test_shared_head_sha_branch_uses_correct_pr_association(self):
+        # Newer run from another PR must not shadow our actual canonical run.
+        other = {"id": RUN + 1, "run_attempt": 1, "name": "Build ARSAS",
+                 "head_sha": HEAD, "head_branch": BRANCH, "event": "pull_request",
+                 "pull_requests": [{"number": PR + 1}]}
+        correct = dict(other, id=RUN, pull_requests=[{"number": PR}])
+        chosen = shadow.choose_exact_run(
+            {"workflow_runs": [other, correct]},
+            head_sha=HEAD, head_branch=BRANCH, pr_number=PR,
+        )
+        self.assertEqual(RUN, chosen["id"])
+        self.assertIsNone(shadow.choose_exact_run(
+            {"workflow_runs": [other]},
+            head_sha=HEAD, head_branch=BRANCH, pr_number=PR,
+        ))
+
+    def test_shared_head_run_without_pr_link_fails_closed(self):
+        unlinked = {"id": RUN, "run_attempt": 1, "name": "Build ARSAS",
+                    "head_sha": HEAD, "head_branch": BRANCH, "event": "pull_request"}
+        self.assertIsNone(shadow.choose_exact_run(
+            {"workflow_runs": [unlinked]},
+            head_sha=HEAD, head_branch=BRANCH, pr_number=PR,
+        ))
+        with self.assertRaisesRegex(shadow.ShadowError, "PR number"):
+            shadow.choose_exact_run(
+                {"workflow_runs": [unlinked]},
+                head_sha=HEAD, head_branch=BRANCH, pr_number=True,
+            )
+
+    def test_online_rejects_other_pr_canonical_even_with_same_head(self):
+        with self.assertRaisesRegex(shadow.ShadowError, "timeout"):
+            shadow.compare_online(
+                FakeApi(blob=self.blob, digest=self.digest, pr_number=PR + 1),
+                **self.kwargs,
+            )
+
+    def test_advisory_output_records_exact_pr_number(self):
+        result = shadow.compare_online(
+            FakeApi(blob=self.blob, digest=self.digest), **self.kwargs
+        )
+        self.assertEqual(PR, result["canonicalPrNumber"])
+        self.assertFalse(result["deduplicationAuthorized"])
+
+    def test_event_identity_must_match_pr_number_and_exact_head(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "event.json"
+            valid = {"number": PR, "pull_request": {
+                "number": PR, "head": {"sha": HEAD, "ref": BRANCH}}}
+            path.write_text(json.dumps(valid), encoding="utf-8")
+            self.assertEqual(PR, shadow.read_event_pr_number(
+                path, head_sha=HEAD, head_branch=BRANCH))
+            tampered = [
+                dict(valid, number=PR + 1),
+                dict(valid, pull_request=dict(valid["pull_request"], number=PR + 1)),
+                dict(valid, pull_request=dict(valid["pull_request"],
+                    head={"sha": "f" * 40, "ref": BRANCH})),
+                dict(valid, pull_request=dict(valid["pull_request"],
+                    head={"sha": HEAD, "ref": "other-branch"})),
+            ]
+            for payload in tampered:
+                with self.subTest(payload=payload):
+                    path.write_text(json.dumps(payload), encoding="utf-8")
+                    with self.assertRaises(shadow.ShadowError):
+                        shadow.read_event_pr_number(
+                            path, head_sha=HEAD, head_branch=BRANCH)
+            path.write_text("{malformed", encoding="utf-8")
+            with self.assertRaises(shadow.ShadowError):
+                shadow.read_event_pr_number(
+                    path, head_sha=HEAD, head_branch=BRANCH)
+
     def test_ci_preserves_native_independent_ctest_and_advisory_only(self):
         comtrade = (ROOT.parent / ".github/workflows/comtrade-viewer-integration.yml").read_text()
         canonical = (ROOT.parent / ".github/workflows/build.yml").read_text()