ci: derive bounded Strix matrix from licence verdict

scripts/ci/release_dependency_gate.py

@@ -141,6 +141,7 @@ def resolve_evidence_binder(script_dir: Path | None = None) -> Path:
 STRIX_BINDING_UNBOUND = "STRIX_BINDING_UNBOUND"
 STRIX_TEXTUAL_PASS_REJECTED = "STRIX_TEXTUAL_PASS_REJECTED"
 STRIX_FINDINGS_OPEN = "STRIX_FINDINGS_OPEN"
+STRIX_MATRIX_LIMIT = 256
 
 SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
 GIT_SHA_RE = re.compile(r"^[0-9a-f]{40}$")
@@ -1844,6 +1845,81 @@ def gate(capture_root: Path, stage: str = FULL_STAGE) -> GateReport:
     return report
 
 
+def strix_fanout_plan(
+    capture_root: Path,
+    license_report: Path,
+    control_sha: str,
+    run_id: int,
+    run_attempt: int,
+) -> dict[str, Any]:
+    """Bind one bounded scan matrix to the passing licence stage's full set."""
+
+    capture = Path(capture_root)
+    report = load_json(license_report, LICENSE_MISSING)
+    release = load_json(capture / "release.json")
+    if not isinstance(report, Mapping) or not isinstance(release, Mapping):
+        raise GateError(CAPTURE_INCOMPLETE, "fanout needs release and licence objects")
+    repository = str(release.get("source_repository", ""))
+    source_sha = str(release.get("source_sha", ""))
+    validate_release_identity(repository, source_sha)
+    if (report.get("result") != "PASS" or report.get("stage") != LICENSE_STAGE
+            or report.get("source_repository") != repository
+            or report.get("source_sha") != source_sha):
+        raise GateError(LICENSE_MISSING, "fanout requires a passing matching licence report")
+    if (not GIT_SHA_RE.fullmatch(control_sha) or type(run_id) is not int or run_id <= 0
+            or type(run_attempt) is not int or run_attempt <= 0):
+        raise GateError(CAPTURE_INCOMPLETE, "fanout execution identity is invalid")
+    rows = report.get("dependencies")
+    if not isinstance(rows, list) or not 1 <= len(rows) <= STRIX_MATRIX_LIMIT:
+        raise GateError(SCOPE_UNVERIFIABLE, "dependency matrix is empty or exceeds 256 jobs")
+    fixtures = capture / "strix" / "fixtures"
+    if fixtures.is_symlink() or not fixtures.is_dir():
+        raise GateError(CAPTURE_INCOMPLETE, "fixture directory is unavailable")
+    planned: list[dict[str, str]] = []
+    members: set[str] = set()
+    keys: set[str] = set()
+    for row in rows:
+        if not isinstance(row, Mapping):
+            raise GateError(CAPTURE_INCOMPLETE, "licence dependency row is malformed")
+        key = row.get("key")
+        expected_digest = row.get("fixture_sha256")
+        if (not isinstance(key, str) or not key or key in keys
+                or not isinstance(expected_digest, str)
+                or not SHA256_RE.fullmatch(expected_digest)):
+            raise GateError(CAPTURE_INCOMPLETE, "licence dependency key or fixture digest is invalid")
+        slug = _slug_for_key(key)
+        if (slug in {"", ".", ".."} or Path(slug).name != slug
+                or "/" in slug or "\\" in slug):
+            raise GateError(CAPTURE_INCOMPLETE, "dependency fixture slug is unsafe")
+        fixture_path = _require_regular_file(fixtures / f"{slug}.json", CAPTURE_INCOMPLETE)
+        digest_path = _require_regular_file(fixtures / f"{slug}.sha256", CAPTURE_INCOMPLETE)
+        fixture = load_json(fixture_path)
+        if (fixture_digest(fixture) != expected_digest
+                or digest_path.read_text(encoding="utf-8").strip() != expected_digest):
+            raise GateError(SOURCE_HASH_MISMATCH, f"{key}: fixture differs from the licence report")
+        artifact_name = f"release-strix-binding-a{run_attempt}-" + hashlib.sha256(
+            key.encode("utf-8")
+        ).hexdigest()
+        planned.append({"key": key, "slug": slug, "fixture_sha256": expected_digest,
+                        "artifact_name": artifact_name})
+        members.update({f"{slug}.json", f"{slug}.sha256"})
+        keys.add(key)
+    if (len({item["slug"] for item in planned}) != len(planned)
+            or {entry.name for entry in fixtures.iterdir()} != members
+            or any(entry.is_symlink() or not entry.is_file() for entry in fixtures.iterdir())):
+        raise GateError(SCOPE_SET_MISMATCH, "fixture directory differs from the exact licence set")
+    return {
+        "schema": "cwl.release-strix-fanout-plan/1",
+        "source_repository": repository,
+        "source_sha": source_sha,
+        "control_sha": control_sha,
+        "run_id": run_id,
+        "run_attempt": run_attempt,
+        "license_report_sha256": _sha256_file(license_report),
+        "dependencies": planned,
+    }
+
+
 # ---------------------------------------------------------------------------
 # Sealed-evidence composition with exact-artifact-sbom-attestation.yml
 # ---------------------------------------------------------------------------
@@ -2259,6 +2335,16 @@ def main(argv: Sequence[str] | None = None) -> int:
     screen.add_argument("--capture", required=True)
     screen.add_argument("--report", required=True)
 
+    fanout = sub.add_parser(
+        "fanout-plan", help="Emit a bounded exact dependency matrix after licence approval"
+    )
+    fanout.add_argument("--capture", required=True)
+    fanout.add_argument("--license-report", required=True)
+    fanout.add_argument("--control-sha", required=True)
+    fanout.add_argument("--run-id", required=True, type=int)
+    fanout.add_argument("--run-attempt", required=True, type=int)
+    fanout.add_argument("--output", required=True)
+
     sub.add_parser(
         "require-strix-credentials", help="Refuse the Strix stage when a credential is absent"
     )
@@ -2350,6 +2436,19 @@ def main(argv: Sequence[str] | None = None) -> int:
             for failure in failures:
                 print(f"ERROR: {failure.code}: {failure.detail}", file=sys.stderr)
             return 2 if failures else 0
+        if args.command == "fanout-plan":
+            plan = strix_fanout_plan(
+                Path(args.capture), Path(args.license_report), args.control_sha,
+                args.run_id, args.run_attempt,
+            )
+            path = Path(args.output)
+            if path.exists() or path.is_symlink():
+                raise GateError(CAPTURE_INCOMPLETE, "fanout plan output already exists")
+            path.write_text(json.dumps(plan, sort_keys=True) + "\n", encoding="utf-8")
+            matrix = {"include": plan["dependencies"]}
+            write_github_output({"matrix_json": json.dumps(matrix, separators=(",", ":"))}, destination)
+            print(json.dumps(matrix, sort_keys=True))
+            return 0
         if args.command in {"gate", "prescreen"}:
             stage = FULL_STAGE if args.command == "gate" else LICENSE_STAGE
             report = gate(Path(args.capture), stage=stage)

tests/test_release_dependency_fanout_plan.py

@@ -0,0 +1,75 @@
+"""The Strix matrix may only come from the passing full licence set."""
+
+from __future__ import annotations
+
+import copy
+import hashlib
+import json
+from pathlib import Path
+
+import pytest
+
+from scripts.ci import release_dependency_gate as gate
+from tests.test_release_dependency_gate import build_capture
+
+
+CONTROL = "d" * 40
+
+
+def _allowed(tmp_path: Path) -> tuple[Path, Path]:
+    capture = build_capture(tmp_path)
+    for fixture in (capture / "strix/fixtures").glob("*.json"):
+        digest = gate.fixture_digest(json.loads(fixture.read_text()))
+        fixture.with_suffix(".sha256").write_text(digest + "\n")
+    report = gate.gate(capture, stage=gate.LICENSE_STAGE)
+    assert report.passed
+    report_path = tmp_path / "license-report.json"
+    report_path.write_text(json.dumps(report.to_json()) + "\n")
+    return capture, report_path
+
+
+def test_fanout_plan_matches_every_prescreened_fixture(tmp_path: Path) -> None:
+    capture, report_path = _allowed(tmp_path)
+    plan = gate.strix_fanout_plan(capture, report_path, CONTROL, 42, 2)
+    report = json.loads(report_path.read_text())
+    assert plan["source_sha"] == report["source_sha"]
+    assert plan["license_report_sha256"] == hashlib.sha256(report_path.read_bytes()).hexdigest()
+    assert (plan["control_sha"], plan["run_id"], plan["run_attempt"]) == (CONTROL, 42, 2)
+    assert {row["key"] for row in plan["dependencies"]} == {
+        row["key"] for row in report["dependencies"]
+    }
+    assert len({row["artifact_name"] for row in plan["dependencies"]}) == len(plan["dependencies"])
+    assert all(row["artifact_name"].startswith("release-strix-binding-a2-") for row in plan["dependencies"])
+
+
+def test_plan_refuses_denied_missing_extra_and_duplicate_scope(tmp_path: Path) -> None:
+    mutators = {
+        "denied": lambda capture, report: report.__setitem__("result", "FAIL"),
+        "duplicate": lambda capture, report: report["dependencies"].append(copy.deepcopy(report["dependencies"][0])),
+        "limit": lambda capture, report: report.__setitem__("dependencies", report["dependencies"] * 257),
+        "missing": lambda capture, report: next((capture / "strix/fixtures").glob("*.json")).unlink(),
+        "extra": lambda capture, report: (capture / "strix/fixtures/unlisted.json").write_text("{}"),
+        "wrong-source": lambda capture, report: report.__setitem__("source_sha", "e" * 40),
+    }
+    for name, mutate in mutators.items():
+        capture, report_path = _allowed(tmp_path / name)
+        report = json.loads(report_path.read_text())
+        mutate(capture, report)
+        report_path.write_text(json.dumps(report) + "\n")
+        with pytest.raises(gate.GateError):
+            gate.strix_fanout_plan(capture, report_path, CONTROL, 42, 2)
+
+
+def test_fanout_cli_emits_one_bounded_matrix_output(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    capture, report_path = _allowed(tmp_path)
+    output = tmp_path / "matrix-output.txt"
+    monkeypatch.setenv("GITHUB_OUTPUT", str(output))
+    plan_path = tmp_path / "plan.json"
+    assert gate.main([
+        "fanout-plan", "--capture", str(capture), "--license-report", str(report_path),
+        "--control-sha", CONTROL, "--run-id", "42", "--run-attempt", "2",
+        "--output", str(plan_path),
+    ]) == 0
+    matrix = json.loads(output.read_text().removeprefix("matrix_json="))
+    assert matrix["include"] == json.loads(plan_path.read_text())["dependencies"]
+    assert len(matrix["include"]) <= gate.STRIX_MATRIX_LIMIT