refactor(workflows): extract shared script helpers

.github/aicodingflow-tests/test_context_helpers.py

@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from script_imports import import_script
+
+
+artifact_contracts = import_script(".github/scripts/artifact_contracts.py", "artifact_contracts")
+context_snapshot = import_script(".github/scripts/context_snapshot.py", "context_snapshot")
+github_event = import_script(".github/scripts/github_event.py", "github_event")
+
+
+class ArtifactContractTests(unittest.TestCase):
+    def test_load_and_write_json_round_trip(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            path = Path(tmpdir) / "artifact.json"
+            artifact_contracts.write_json(path, {"message": "中文"})
+
+            self.assertEqual(artifact_contracts.load_json(path), {"message": "中文"})
+            self.assertTrue(path.read_text(encoding="utf-8").endswith("\n"))
+
+    def test_load_json_returns_default_for_missing_path(self) -> None:
+        self.assertEqual(artifact_contracts.load_json("", default={}), {})
+        self.assertEqual(artifact_contracts.load_json(None, default=[]), [])
+
+    def test_write_github_output_appends_key_values(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            path = Path(tmpdir) / "github_output.txt"
+
+            artifact_contracts.write_github_output(path, {"should_run": "true", "reason": "ok"})
+            artifact_contracts.write_github_output(path, {"issue_number": "243"})
+
+            self.assertEqual(
+                path.read_text(encoding="utf-8"),
+                "should_run=true\nreason=ok\nissue_number=243\n",
+            )
+
+
+class GithubEventTests(unittest.TestCase):
+    def test_extracts_stable_issue_fields(self) -> None:
+        issue = {
+            "labels": [{"name": "ready-to-spec"}, {"name": ""}, {}],
+            "assignees": [{"login": "agent"}, {"login": ""}, {}],
+        }
+
+        self.assertEqual(github_event.label_names(issue), ["ready-to-spec"])
+        self.assertEqual(github_event.assignee_logins(issue), ["agent"])
+
+    def test_triggering_comment_snapshot_can_include_association(self) -> None:
+        event = {
+            "comment": {
+                "id": 123,
+                "user": {"login": "maintainer"},
+                "author_association": "MEMBER",
+                "body": "@agent /triage",
+                "created_at": "2026-06-08T00:00:00Z",
+                "html_url": "https://example.test/comment",
+            }
+        }
+
+        self.assertEqual(
+            github_event.triggering_comment_snapshot(event, include_author_association=True),
+            {
+                "id": 123,
+                "author": "maintainer",
+                "author_association": "MEMBER",
+                "body": "@agent /triage",
+                "created_at": "2026-06-08T00:00:00Z",
+                "url": "https://example.test/comment",
+            },
+        )
+
+    def test_detects_pull_request_issue_event_only_when_present(self) -> None:
+        self.assertTrue(github_event.is_pull_request_issue_event({"issue": {"pull_request": {"url": "pr"}}}))
+        self.assertFalse(github_event.is_pull_request_issue_event({"issue": {"pull_request": None}}))
+        self.assertFalse(github_event.is_pull_request_issue_event({"issue": {}}))
+
+
+class ContextSnapshotTests(unittest.TestCase):
+    def test_flatten_pages_accepts_paginated_and_plain_lists(self) -> None:
+        self.assertEqual(context_snapshot.flatten_pages([[{"id": 1}], [{"id": 2}]]), [{"id": 1}, {"id": 2}])
+        self.assertEqual(context_snapshot.flatten_pages([{"id": 1}]), [{"id": 1}])
+
+    def test_remove_triggering_comment_filters_by_id_or_url(self) -> None:
+        comments = [
+            {"id": 1, "html_url": "https://example.test/1"},
+            {"id": 2, "html_url": "https://example.test/2"},
+            {"id": 3, "html_url": "https://example.test/trigger"},
+        ]
+
+        self.assertEqual(
+            context_snapshot.remove_triggering_comment(
+                comments,
+                {"id": 2, "url": "https://example.test/trigger"},
+            ),
+            [{"id": 1, "html_url": "https://example.test/1"}],
+        )
+
+    def test_format_issue_comments_uses_stable_text_shape(self) -> None:
+        self.assertEqual(
+            context_snapshot.format_issue_comments(
+                [
+                    {
+                        "user": {"login": "octo"},
+                        "created_at": "2026-06-08T00:00:00Z",
+                        "body": "hello",
+                    }
+                ]
+            ),
+            "Author: octo\nCreated: 2026-06-08T00:00:00Z\n\nhello\n\n---\n",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

.github/aicodingflow-tests/test_github_api.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+import json
+import subprocess
+import unittest
+from unittest import mock
+
+from script_imports import import_script
+
+
+github_api = import_script(".github/scripts/github_api.py", "github_api")
+
+
+class GithubApiTests(unittest.TestCase):
+    def test_run_gh_json_invokes_gh_and_parses_stdout(self) -> None:
+        completed = subprocess.CompletedProcess(args=[], returncode=0, stdout=json.dumps({"ok": True}))
+        with mock.patch.object(github_api.subprocess, "run", return_value=completed) as run:
+            self.assertEqual(github_api.run_gh_json(["repo", "view"]), {"ok": True})
+
+        run.assert_called_once_with(["gh", "repo", "view"], check=True, stdout=subprocess.PIPE, text=True)
+
+    def test_run_gh_text_returns_stdout(self) -> None:
+        completed = subprocess.CompletedProcess(args=[], returncode=0, stdout="body\n")
+        with mock.patch.object(github_api.subprocess, "run", return_value=completed):
+            self.assertEqual(github_api.run_gh_text(["pr", "diff"]), "body\n")
+
+    def test_fetch_default_branch_requires_branch_name(self) -> None:
+        with mock.patch.object(github_api, "run_gh_json", return_value={"defaultBranchRef": {"name": "main"}}):
+            self.assertEqual(github_api.fetch_default_branch("owner/repo"), "main")
+
+        with mock.patch.object(github_api, "run_gh_json", return_value={"defaultBranchRef": {}}):
+            with self.assertRaisesRegex(SystemExit, "could not determine default branch"):
+                github_api.fetch_default_branch("owner/repo")
+
+    def test_flatten_gh_pages_delegates_paginated_shape(self) -> None:
+        self.assertEqual(github_api.flatten_gh_pages([[{"number": 1}], [{"number": 2}]]), [{"number": 1}, {"number": 2}])
+
+
+if __name__ == "__main__":
+    unittest.main()

.github/aicodingflow-tests/test_handle_plan_approved.py

@@ -71,6 +71,14 @@ def test_resolves_linked_issue_from_spec_branch_fallback(self) -> None:
             57,
         )
 
+    def test_bare_hash_and_gh_refs_do_not_override_spec_branch_fallback(self) -> None:
+        self.assertEqual(
+            handle_plan_approved.resolve_linked_issue_number(
+                pr(body="See #99 for background", title="GH-88", head_ref="spec/issue-57")
+            ),
+            57,
+        )
+
     def test_does_not_resolve_unqualified_numbers(self) -> None:
         self.assertIsNone(
             handle_plan_approved.resolve_linked_issue_number(

.github/aicodingflow-tests/test_issue_refs.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+import unittest
+
+from script_imports import import_script
+
+
+issue_refs = import_script(".github/scripts/issue_refs.py", "issue_refs")
+
+
+class IssueRefsTest(unittest.TestCase):
+    def test_issue_number_from_text_uses_explicit_tokens_and_branch_suffix(self) -> None:
+        self.assertEqual(issue_refs.issue_number_from_text("Refs #42"), 42)
+        self.assertEqual(issue_refs.issue_number_from_text("GH-42"), 42)
+        self.assertEqual(issue_refs.issue_number_from_text("issue 42"), 42)
+        self.assertEqual(issue_refs.issue_number_from_text("feat/thing-42"), 42)
+        self.assertIsNone(issue_refs.issue_number_from_text("walk through 42 cases"))
+        self.assertIsNone(issue_refs.issue_number_from_text("highlight gh42 as text"))
+
+    def test_issue_number_from_branch_accepts_spec_branch_prefix(self) -> None:
+        self.assertEqual(issue_refs.issue_number_from_branch("spec/issue-57", prefix="spec/issue-"), 57)
+        self.assertIsNone(issue_refs.issue_number_from_branch("spec/issue-not-a-number", prefix="spec/issue-"))
+
+    def test_issue_number_from_strict_text_ignores_bare_hash_and_gh_refs(self) -> None:
+        self.assertEqual(issue_refs.issue_number_from_strict_text("Refs #42"), 42)
+        self.assertEqual(issue_refs.issue_number_from_strict_text("issue 42"), 42)
+        self.assertEqual(issue_refs.issue_number_from_strict_text("issue #42"), 42)
+        self.assertIsNone(issue_refs.issue_number_from_strict_text("See #99 for context"))
+        self.assertIsNone(issue_refs.issue_number_from_strict_text("GH-99"))
+
+    def test_issue_numbers_from_closing_refs_ignores_pr_references(self) -> None:
+        pr = {
+            "title": "Implement flow refs #90",
+            "body": "Refs #88, #89 and PR #123\n\nReferenced pull request #124 should not count.\nFixes #91",
+            "closingIssuesReferences": [{"number": 87}, {"number": "87"}, {"number": 88}],
+        }
+
+        self.assertEqual(issue_refs.issue_numbers_from_closing_refs(pr), [87, 88, 90, 89, 91])
+
+
+if __name__ == "__main__":
+    unittest.main()

.github/aicodingflow-tests/test_ledger_contracts.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from script_imports import import_script
+
+
+ledger_contracts = import_script(".github/scripts/ledger_contracts.py", "ledger_contracts")
+
+
+class LedgerContractsTest(unittest.TestCase):
+    def test_load_ledger_defaults_and_indexes_entries_by_pr(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            missing = Path(directory) / "ledger.json"
+            ledger = ledger_contracts.load_ledger(missing, ledger_name="test")
+
+        self.assertEqual(ledger, {"version": 1, "entries": []})
+        self.assertEqual(
+            ledger_contracts.entries_by_pr({"entries": [{"pr": "1"}, {"pr": None}, {"pr": 2}]}),
+            {1: {"pr": "1"}, 2: {"pr": 2}},
+        )
+
+    def test_load_ledger_rejects_invalid_entries_shape(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "ledger.json"
+            path.write_text('{"entries": {}}', encoding="utf-8")
+
+            with self.assertRaisesRegex(SystemExit, "invalid test ledger entries"):
+                ledger_contracts.load_ledger(path, ledger_name="test")
+
+    def test_set_sorted_entries_orders_by_merge_time_and_pr_number(self) -> None:
+        ledger = ledger_contracts.set_sorted_entries(
+            {},
+            [
+                {"pr": 3, "merged_at": "2026-06-02T00:00:00Z"},
+                {"pr": 1, "merged_at": "2026-06-01T00:00:00Z"},
+                {"pr": 2, "merged_at": "2026-06-01T00:00:00Z"},
+            ],
+        )
+
+        self.assertEqual([entry["pr"] for entry in ledger["entries"]], [1, 2, 3])
+        self.assertEqual(ledger["version"], 1)
+
+
+if __name__ == "__main__":
+    unittest.main()

.github/aicodingflow-tests/test_pr_metadata_contracts.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+import json
+import tempfile
+import unittest
+from pathlib import Path
+
+from script_imports import import_script
+
+
+contracts = import_script(".github/scripts/pr_metadata_contracts.py", "pr_metadata_contracts")
+
+
+class PrMetadataContractTests(unittest.TestCase):
+    def write_json(self, directory: str, name: str, value: object) -> Path:
+        path = Path(directory) / name
+        path.write_text(json.dumps(value), encoding="utf-8")
+        return path
+
+    def test_validate_base_metadata_accepts_common_contract(self) -> None:
+        metadata = {
+            "branch_name": "feature",
+            "pr_title": "feat: add workflow",
+            "pr_summary": "Refs #1\n\n## Summary\nDone.",
+        }
+
+        contracts.validate_base_metadata(metadata)
+
+    def test_validate_base_metadata_rejects_missing_string_and_title(self) -> None:
+        with self.assertRaisesRegex(SystemExit, "missing fields: pr_summary"):
+            contracts.validate_base_metadata({"branch_name": "feature", "pr_title": "feat: ok"})
+
+        with self.assertRaisesRegex(SystemExit, "branch_name must be a non-empty string"):
+            contracts.validate_base_metadata({"branch_name": "", "pr_title": "feat: ok", "pr_summary": "x\nx"})
+
+        with self.assertRaisesRegex(SystemExit, "conventional commit style"):
+            contracts.validate_base_metadata(
+                {"branch_name": "feature", "pr_title": "Add workflow", "pr_summary": "x\nx"}
+            )
+
+    def test_validate_intended_files_normalizes_and_rejects_unsafe_paths(self) -> None:
+        self.assertEqual(
+            contracts.validate_intended_files({"intended_files": [" app.py ", "tests/test_app.py"]}),
+            ["app.py", "tests/test_app.py"],
+        )
+
+        with self.assertRaisesRegex(SystemExit, "repository-relative"):
+            contracts.validate_intended_files({"intended_files": ["../app.py"]})
+
+        with self.assertRaisesRegex(SystemExit, "handoff"):
+            contracts.validate_intended_files({"intended_files": ["pr-metadata.json"]})
+
+        with self.assertRaisesRegex(SystemExit, "generated/cache"):
+            contracts.validate_intended_files({"intended_files": [".codex-runtime/skills/example/SKILL.md"]})
+
+    def test_load_json_object_validates_required_object(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            path = self.write_json(directory, "metadata.json", {"branch_name": "feature"})
+            self.assertEqual(contracts.load_json_object(path), {"branch_name": "feature"})
+
+            missing = Path(directory) / "missing.json"
+            self.assertEqual(contracts.load_json_object(missing, required=False), {})
+
+            invalid = self.write_json(directory, "list.json", [])
+            with self.assertRaisesRegex(SystemExit, "must contain a JSON object"):
+                contracts.load_json_object(invalid)
+
+
+if __name__ == "__main__":
+    unittest.main()

.github/aicodingflow-tests/test_prepare_local_review_inputs.py

@@ -349,7 +349,7 @@ def fake_run(args: list[str], env: dict[str, str] | None = None) -> str:
 
         with (
             mock.patch.object(prepare_local, "current_branch", return_value="fix/github-pr"),
-            mock.patch.object(prepare_local, "run", side_effect=fake_run),
+            mock.patch.object(prepare_local, "run_command", side_effect=fake_run),
         ):
             event = prepare_local.github_pr_event_for_current_branch("owner/repo")
 
@@ -360,7 +360,7 @@ def fake_run(args: list[str], env: dict[str, str] | None = None) -> str:
     def test_github_pr_event_for_current_branch_returns_none_when_fetch_fails(self) -> None:
         with (
             mock.patch.object(prepare_local, "current_branch", return_value="fix/github-pr"),
-            mock.patch.object(prepare_local, "run", side_effect=subprocess.CalledProcessError(1, ["gh"])),
+            mock.patch.object(prepare_local, "run_command", side_effect=subprocess.CalledProcessError(1, ["gh"])),
         ):
             self.assertIsNone(prepare_local.github_pr_event_for_current_branch("owner/repo"))
 

.github/scripts/apply_issue_triage_result.py

@@ -9,24 +9,14 @@
 from pathlib import Path
 from typing import Any
 
+from github_api import flatten_gh_pages as flatten_pages
+from github_api import run_gh_json
+
 
 MARKER = "<!-- aicodingflow:triage-issue -->"
 PROTECTED_LABELS = {"plan-approved", "ready-to-implement", "ready-to-spec"}
 
 
-def run_gh_json(args: list[str]) -> Any:
-    result = subprocess.run(["gh", *args], check=True, stdout=subprocess.PIPE, text=True)
-    return json.loads(result.stdout)
-
-
-def flatten_pages(value: Any) -> list[dict[str, Any]]:
-    if isinstance(value, list) and value and all(isinstance(page, list) for page in value):
-        return [item for page in value for item in page]
-    if isinstance(value, list):
-        return value
-    raise SystemExit("unexpected GitHub API response")
-
-
 def load_json(path: Path) -> dict[str, Any]:
     return json.loads(path.read_text(encoding="utf-8"))