Separate the general supervisory kernel from software delivery

.github/scripts/build_codex_github_review.py

@@ -0,0 +1,191 @@
+#!/usr/bin/env python3
+"""Build a GitHub review while preserving findings without valid diff anchors."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import posixpath
+import re
+import subprocess
+from pathlib import Path
+
+
+HUNK = re.compile(r"^@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))? @@")
+
+
+def normalize_path(value: str, workspace: str) -> str | None:
+    candidate = value.replace("\\", "/")
+    root = workspace.replace("\\", "/").rstrip("/")
+    if candidate.startswith(root + "/"):
+        candidate = candidate[len(root) + 1 :]
+    while candidate.startswith("./"):
+        candidate = candidate[2:]
+    candidate = posixpath.normpath(candidate)
+    if (
+        not candidate
+        or candidate == "."
+        or candidate.startswith("/")
+        or candidate == ".."
+        or candidate.startswith("../")
+        or "/../" in candidate
+    ):
+        return None
+    return candidate
+
+
+def header_path(line: str) -> str | None:
+    value = line[4:].split("\t", 1)[0]
+    if value == "/dev/null":
+        return None
+    if value.startswith(("a/", "b/")):
+        value = value[2:]
+    return value
+
+
+def changed_lines(diff: str) -> dict[str, set[tuple[str, int]]]:
+    allowed: dict[str, set[tuple[str, int]]] = {"LEFT": set(), "RIGHT": set()}
+    old_path: str | None = None
+    new_path: str | None = None
+    old_line = 0
+    new_line = 0
+    in_hunk = False
+    for line in diff.splitlines():
+        if line.startswith("diff --git "):
+            old_path = new_path = None
+            in_hunk = False
+            continue
+        if not in_hunk and line.startswith("--- "):
+            old_path = header_path(line)
+            continue
+        if not in_hunk and line.startswith("+++ "):
+            new_path = header_path(line)
+            continue
+        match = HUNK.match(line)
+        if match:
+            old_line = int(match.group(1))
+            new_line = int(match.group(3))
+            in_hunk = True
+            continue
+        if not in_hunk or line.startswith("\\"):
+            continue
+        if line.startswith("-"):
+            if old_path is not None:
+                allowed["LEFT"].add((old_path, old_line))
+            old_line += 1
+        elif line.startswith("+"):
+            if new_path is not None:
+                allowed["RIGHT"].add((new_path, new_line))
+            new_line += 1
+        elif line.startswith(" "):
+            old_line += 1
+            new_line += 1
+        else:
+            in_hunk = False
+    return allowed
+
+
+def finding_body(finding: dict[str, object]) -> str:
+    return (
+        f"[P{finding['priority']}] {finding['title']}\n\n"
+        f"{finding['body']}\n\nConfidence: {finding['confidence_score']}"
+    )
+
+
+def build_review(
+    review: dict[str, object],
+    *,
+    commit: str,
+    workspace: str,
+    allowed: dict[str, set[tuple[str, int]]],
+) -> dict[str, object]:
+    body = (
+        "Codex automated review\n\n"
+        f"Verdict: {review['overall_correctness']}\n"
+        f"Confidence: {review['overall_confidence_score']}\n\n"
+        f"{review['overall_explanation']}"
+    )
+    comments: list[dict[str, object]] = []
+    unanchored: list[str] = []
+    for finding in review["findings"]:  # type: ignore[index]
+        location = finding["code_location"]
+        path = normalize_path(location["absolute_file_path"], workspace)
+        side = location["side"]
+        start = location["line_range"]["start"]
+        end = location["line_range"]["end"]
+        valid_anchor = (
+            path is not None
+            and side in allowed
+            and start <= end
+            and all((path, line) in allowed[side] for line in range(start, end + 1))
+        )
+        text = finding_body(finding)
+        if valid_anchor:
+            comment: dict[str, object] = {
+                "path": path,
+                "line": end,
+                "side": side,
+                "body": text,
+            }
+            if start != end:
+                comment["start_line"] = start
+                comment["start_side"] = side
+            comments.append(comment)
+            continue
+        display_path = path or "unresolved-path"
+        unanchored.append(f"{text}\n\nLocation: {display_path}:{start}-{end} ({side})")
+    if unanchored:
+        body += "\n\nFindings without inline diff anchors\n\n" + "\n\n---\n\n".join(unanchored)
+    return {"commit_id": commit, "event": "COMMENT", "body": body, "comments": comments}
+
+
+def pull_request_diff(workspace: str, base: str, head: str) -> str:
+    merge_base = subprocess.run(
+        ["git", "merge-base", base, head],
+        cwd=workspace,
+        check=True,
+        text=True,
+        capture_output=True,
+    ).stdout.strip()
+    return subprocess.run(
+        [
+            "git",
+            "-c",
+            "core.quotePath=false",
+            "diff",
+            "--no-ext-diff",
+            "--no-renames",
+            "--unified=0",
+            merge_base,
+            head,
+        ],
+        cwd=workspace,
+        check=True,
+        text=True,
+        capture_output=True,
+    ).stdout
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--input", required=True)
+    parser.add_argument("--output", required=True)
+    parser.add_argument("--workspace", required=True)
+    parser.add_argument("--base", required=True)
+    parser.add_argument("--head", required=True)
+    args = parser.parse_args()
+    diff = pull_request_diff(args.workspace, args.base, args.head)
+    review = json.loads(Path(args.input).read_text())
+    payload = build_review(
+        review,
+        commit=args.head,
+        workspace=os.path.abspath(args.workspace),
+        allowed=changed_lines(diff),
+    )
+    Path(args.output).write_text(json.dumps(payload, indent=2) + "\n")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

.github/tests/boatstack_test_support.py

@@ -14,12 +14,16 @@ def prescription_cli_arguments(
         correlation,
         "--prescription-id",
         str(prescription["id"]),
+        "--expected-instance-id",
+        str(prescription["expected_instance_id"]),
         "--expected-state-revision",
         str(prescription["expected_state_revision"]),
         "--expected-program-fingerprint",
         str(prescription["expected_program_fingerprint"]),
         "--expected-snapshot-fingerprint",
         str(prescription["expected_snapshot_fingerprint"]),
+        "--expected-objective-binding-fingerprint",
+        str(prescription["expected_objective_binding_fingerprint"]),
         "--authority-fingerprint",
         str(prescription["authority_fingerprint"]),
     ]

.github/tests/test_codex_review_publish.py

@@ -0,0 +1,112 @@
+import importlib.util
+import subprocess
+import tempfile
+import unittest
+from pathlib import Path
+
+
+SCRIPT = Path(__file__).parents[1] / "scripts" / "build_codex_github_review.py"
+SPEC = importlib.util.spec_from_file_location("build_codex_github_review", SCRIPT)
+MODULE = importlib.util.module_from_spec(SPEC)
+assert SPEC.loader is not None
+SPEC.loader.exec_module(MODULE)
+
+
+class CodexReviewPublishTests(unittest.TestCase):
+    def test_invalid_inline_location_is_preserved_in_review_body(self) -> None:
+        diff = """diff --git a/review.go b/review.go
+--- a/review.go
++++ b/review.go
+@@ -2 +2 @@
+-old
++new
+"""
+        review = {
+            "findings": [
+                self.finding("anchored", 2),
+                self.finding("not changed", 9),
+            ],
+            "overall_correctness": "patch is incorrect",
+            "overall_explanation": "One location is outside the pull request diff.",
+            "overall_confidence_score": 0.9,
+        }
+        payload = MODULE.build_review(
+            review,
+            commit="head",
+            workspace="/workspace",
+            allowed=MODULE.changed_lines(diff),
+        )
+        self.assertEqual(len(payload["comments"]), 1)
+        self.assertEqual(payload["comments"][0]["line"], 2)
+        self.assertIn("Findings without inline diff anchors", payload["body"])
+        self.assertIn("not changed", payload["body"])
+        self.assertIn("review.go:9-9 (RIGHT)", payload["body"])
+
+    def test_outside_workspace_path_cannot_become_inline_comment(self) -> None:
+        review = {
+            "findings": [self.finding("outside", 2, path="/tmp/other.go")],
+            "overall_correctness": "patch is incorrect",
+            "overall_explanation": "The path is not repository-relative.",
+            "overall_confidence_score": 0.8,
+        }
+        payload = MODULE.build_review(
+            review,
+            commit="head",
+            workspace="/workspace",
+            allowed={"LEFT": set(), "RIGHT": {("review.go", 2)}},
+        )
+        self.assertEqual(payload["comments"], [])
+        self.assertIn("unresolved-path", payload["body"])
+
+    def test_pull_request_diff_excludes_base_only_changes(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            repository = Path(directory)
+            self.git(repository, "init", "-q")
+            self.git(repository, "config", "user.email", "review@example.invalid")
+            self.git(repository, "config", "user.name", "Review Test")
+            (repository / "base.txt").write_text("original\n")
+            (repository / "head.txt").write_text("original\n")
+            self.git(repository, "add", ".")
+            self.git(repository, "commit", "-q", "-m", "initial")
+            common = self.git(repository, "rev-parse", "HEAD")
+
+            (repository / "base.txt").write_text("base only\n")
+            self.git(repository, "commit", "-q", "-am", "base")
+            base = self.git(repository, "rev-parse", "HEAD")
+
+            self.git(repository, "checkout", "-q", "--detach", common)
+            (repository / "head.txt").write_text("head only\n")
+            self.git(repository, "commit", "-q", "-am", "head")
+            head = self.git(repository, "rev-parse", "HEAD")
+
+            diff = MODULE.pull_request_diff(str(repository), base, head)
+            self.assertIn("head.txt", diff)
+            self.assertNotIn("base.txt", diff)
+
+    @staticmethod
+    def finding(title: str, line: int, path: str = "/workspace/review.go") -> dict:
+        return {
+            "title": title,
+            "body": "Finding detail.",
+            "confidence_score": 0.95,
+            "priority": 1,
+            "code_location": {
+                "absolute_file_path": path,
+                "side": "RIGHT",
+                "line_range": {"start": line, "end": line},
+            },
+        }
+
+    @staticmethod
+    def git(repository: Path, *args: str) -> str:
+        return subprocess.run(
+            ["git", *args],
+            cwd=repository,
+            check=True,
+            text=True,
+            capture_output=True,
+        ).stdout.strip()
+
+
+if __name__ == "__main__":
+    unittest.main()