feat: read_head_check_runs completion port method (Nexus-new #2068)

scripts/github_completion_bridge.py

@@ -24,6 +24,41 @@ def _recv() -> dict[str, Any]:
     return value
 
 
+def parse_head_check_runs(result: Any, head_sha: str, observation_cls: Any) -> tuple[Any, ...]:
+    """Validate the host's check-run payload; fail closed on any irregularity.
+
+    Each run must carry its own head_sha equal to the requested one.  App
+    ``app_id`` (int) and ``workflow_path`` (str, "" for non-Actions apps) are
+    passed straight into the observation so the loop can pin the app/workflow.
+    """
+    if not isinstance(head_sha, str) or len(head_sha) != 40 or any(
+        c not in "0123456789abcdef" for c in head_sha
+    ):
+        raise RuntimeError("HEAD_CHECK_RUNS_HEAD_SHA_INVALID")
+    if not isinstance(result, dict) or not isinstance(result.get("check_runs"), list):
+        raise RuntimeError("HEAD_CHECK_RUNS_RESPONSE_INVALID")
+    observations = []
+    for item in result["check_runs"]:
+        if not isinstance(item, dict) or item.get("head_sha") != head_sha:
+            raise RuntimeError("HEAD_CHECK_RUNS_HEAD_SHA_MISMATCH")
+        app_id = item.get("app_id")
+        if not isinstance(app_id, int) or isinstance(app_id, bool):
+            raise RuntimeError("HEAD_CHECK_RUNS_APP_ID_INVALID")
+        if not isinstance(item.get("workflow_path"), str):
+            raise RuntimeError("HEAD_CHECK_RUNS_WORKFLOW_PATH_INVALID")
+        observations.append(
+            observation_cls(
+                name=item.get("name"),
+                status=item.get("status"),
+                conclusion=item.get("conclusion"),
+                head_sha=item.get("head_sha"),
+                app_id=app_id,
+                workflow_path=item["workflow_path"],
+            )
+        )
+    return tuple(observations)
+
+
 def main() -> int:
     start = _recv()
     if start.get("type") != "start":
@@ -44,6 +79,7 @@ def main() -> int:
         CasMergeResult,
         CasMergeStatus,
         DimensionRevalidationReceipt,
+        HeadCheckRunObservation,
         IntegrationMaterializationResult,
         PostMergeReconciliationResult,
         make_dimension_revalidation_receipt,
@@ -149,6 +185,10 @@ def read_required_checks(self, *, head_sha, generation, timeout_seconds=None):
             )
             return tuple(CheckResult.model_validate(item) for item in result["checks"])
 
+        def read_head_check_runs(self, *, head_sha):
+            result = self._call("read_head_check_runs", {"head_sha": head_sha})
+            return parse_head_check_runs(result, head_sha, HeadCheckRunObservation)
+
         def read_reviews(self):
             result = self._call("read_reviews")
             unresolved = int(result.get("unresolved_threads", 0))

scripts/test_github_completion_bridge.py

@@ -0,0 +1,115 @@
+#!/usr/bin/env python3
+"""Unit tests for parse_head_check_runs in github_completion_bridge.py (Nexus-new #2068)."""
+
+import importlib.util
+import os
+import sys
+import unittest
+from pathlib import Path
+
+_spec = importlib.util.spec_from_file_location(
+    "github_completion_bridge", Path(__file__).with_name("github_completion_bridge.py")
+)
+bridge = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(bridge)
+
+SHA = "a" * 40
+
+
+class Obs:
+    def __init__(self, *, name, status, conclusion, head_sha, app_id, workflow_path):
+        self.name, self.status, self.conclusion, self.head_sha = name, status, conclusion, head_sha
+        self.app_id, self.workflow_path = app_id, workflow_path
+
+
+def run(**over):
+    base = {"name": "Exact-base impact gate", "status": "completed", "conclusion": "success",
+            "head_sha": SHA, "app_slug": "github-actions", "app_id": 15368,
+            "workflow_path": ".github/workflows/pytest.yml"}
+    base.update(over)
+    return base
+
+
+class ParseHeadCheckRuns(unittest.TestCase):
+    def parse(self, result, sha=SHA):
+        return bridge.parse_head_check_runs(result, sha, Obs)
+
+    def test_valid_runs(self):
+        out = self.parse({"check_runs": [run(), run(name="x", conclusion=None, status="queued")]})
+        self.assertEqual([o.name for o in out], ["Exact-base impact gate", "x"])
+        self.assertIsNone(out[1].conclusion)
+        self.assertEqual(out[0].app_id, 15368)
+        self.assertEqual(out[0].workflow_path, ".github/workflows/pytest.yml")
+
+    def test_non_actions_app_has_empty_workflow_path(self):
+        out = self.parse({"check_runs": [run(app_id=7, workflow_path="")]})
+        self.assertEqual((out[0].app_id, out[0].workflow_path), (7, ""))
+
+    def test_missing_or_bad_app_fields_fail_closed(self):
+        for bad in [{"app_id": None}, {"app_id": "1"}, {"app_id": True}, {"workflow_path": None}, {"workflow_path": 3}]:
+            with self.assertRaises(RuntimeError):
+                self.parse({"check_runs": [run(**bad)]})
+
+    def test_empty_ok_when_host_says_empty(self):
+        self.assertEqual(self.parse({"check_runs": []}), ())
+
+    def test_bad_requested_sha(self):
+        for bad in ["abc", "A" * 40, None]:
+            with self.assertRaises(RuntimeError):
+                self.parse({"check_runs": []}, bad)
+
+    def test_mismatched_run_sha(self):
+        with self.assertRaises(RuntimeError):
+            self.parse({"check_runs": [run(head_sha="b" * 40)]})
+
+    def test_malformed_payload(self):
+        for bad in [None, [], {}, {"check_runs": None}, {"check_runs": ["x"]}]:
+            with self.assertRaises(RuntimeError):
+                self.parse(bad)
+
+
+def _load_real_observation():
+    """Import the real HeadCheckRunObservation from a Nexus-new checkout.
+
+    Root: $NEXUS_NEW_ROOT or ~/workspace/Nexus-new. Returns (cls, None) or
+    (None, reason) when it cannot be imported (the contract test then skips).
+    """
+    root = Path(os.environ.get("NEXUS_NEW_ROOT", Path.home() / "workspace" / "Nexus-new"))
+    if not (root / "nexus" / "orchestrator" / "github_completion_loop.py").is_file():
+        return None, f"no Nexus-new checkout at {root}"
+    sys.path.insert(0, str(root))
+    try:
+        from nexus.orchestrator.github_completion_loop import HeadCheckRunObservation  # noqa: PLC0415
+    except Exception as exc:  # noqa: BLE001 - any import failure means skip
+        return None, f"cannot import HeadCheckRunObservation from {root}: {type(exc).__name__}: {exc}"
+    return HeadCheckRunObservation, None
+
+
+REAL_OBS, REAL_SKIP_REASON = _load_real_observation()
+
+
+class RealModelContract(unittest.TestCase):
+    """Feeds host-shaped payloads through the REAL Nexus-new model (no fake)."""
+
+    def setUp(self):
+        if REAL_OBS is None:
+            print(f"SKIP RealModelContract: {REAL_SKIP_REASON}", file=sys.stderr)
+            self.skipTest(REAL_SKIP_REASON)
+
+    def test_realistic_mix_validates_with_real_model(self):
+        payload = {"check_runs": [
+            run(name="Exact-base impact gate", workflow_path=".github/workflows/pytest.yml"),
+            run(name="Trusted impact gate (default branch)",
+                workflow_path=".github/workflows/trusted-impact-gate.yml"),
+            run(name="third-party-scan", app_id=7, app_slug="other-app", workflow_path=""),
+        ]}
+        out = bridge.parse_head_check_runs(payload, SHA, REAL_OBS)
+        self.assertEqual([o.app_id for o in out], [15368, 15368, 7])
+        self.assertEqual(
+            [o.workflow_path for o in out],
+            [".github/workflows/pytest.yml", ".github/workflows/trusted-impact-gate.yml", ""],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

src/git-pr-merge.ts

@@ -410,6 +410,39 @@ export interface GitHubCompletionTransport extends GitHubPullRequestTransport {
   compareChangedFiles(repo: string, oldSha: string, newSha: string): Promise<string[]>;
   getPullRequestReviewState(repo: string, prNumber: number): Promise<PullRequestReviewState>;
   isPlatformApprovalRequired(repo: string, branch: string): Promise<boolean>;
+  /**
+   * One page of `GET /repos/{repo}/commits/{sha}/check-runs?filter=latest`.
+   * Rejects on any API error or malformed body; never returns an empty page
+   * to represent a failure.
+   */
+  getCheckRunsPage(
+    repo: string,
+    headSha: string,
+    page: number,
+    perPage: number,
+  ): Promise<CheckRunsPageView>;
+  /**
+   * One page of `GET /repos/{repo}/actions/runs?head_sha={sha}`. Rejects on any
+   * API error or malformed body.
+   */
+  getWorkflowRunsPage(
+    repo: string,
+    headSha: string,
+    page: number,
+    perPage: number,
+  ): Promise<WorkflowRunsPageView>;
+}
+
+/** Raw (unvalidated) Actions workflow-run page. */
+export interface WorkflowRunsPageView {
+  totalCount: number;
+  workflowRuns: unknown[];
+}
+
+/** Raw (unvalidated) check-run page; the completion host validates each run. */
+export interface CheckRunsPageView {
+  totalCount: number;
+  checkRuns: unknown[];
 }
 
 /**
@@ -715,6 +748,52 @@ export class GhCliGitHubTransport implements GitHubCompletionTransport {
     });
   }
 
+  async getCheckRunsPage(
+    repo: string,
+    headSha: string,
+    page: number,
+    perPage: number,
+  ): Promise<CheckRunsPageView> {
+    const body = await this.apiJson([
+      "api",
+      `repos/${repo}/commits/${headSha}/check-runs?filter=latest&per_page=${perPage}&page=${page}`,
+    ]);
+    if (!body || Array.isArray(body)) throw new Error("CHECK_RUNS_RESPONSE_INVALID");
+    const record = body as { total_count?: unknown; check_runs?: unknown };
+    if (
+      typeof record.total_count !== "number"
+      || !Number.isInteger(record.total_count)
+      || record.total_count < 0
+      || !Array.isArray(record.check_runs)
+    ) {
+      throw new Error("CHECK_RUNS_RESPONSE_INVALID");
+    }
+    return { totalCount: record.total_count, checkRuns: record.check_runs };
+  }
+
+  async getWorkflowRunsPage(
+    repo: string,
+    headSha: string,
+    page: number,
+    perPage: number,
+  ): Promise<WorkflowRunsPageView> {
+    const body = await this.apiJson([
+      "api",
+      `repos/${repo}/actions/runs?head_sha=${headSha}&per_page=${perPage}&page=${page}`,
+    ]);
+    if (!body || Array.isArray(body)) throw new Error("WORKFLOW_RUNS_RESPONSE_INVALID");
+    const record = body as { total_count?: unknown; workflow_runs?: unknown };
+    if (
+      typeof record.total_count !== "number"
+      || !Number.isInteger(record.total_count)
+      || record.total_count < 0
+      || !Array.isArray(record.workflow_runs)
+    ) {
+      throw new Error("WORKFLOW_RUNS_RESPONSE_INVALID");
+    }
+    return { totalCount: record.total_count, workflowRuns: record.workflow_runs };
+  }
+
   async getCommitStatus(repo: string, headSha: string): Promise<CommitStatusEvidence[]> {
     const body = await this.apiJson([
       "api",